// Package task implements the agent runtime's task execution model // (DevDocs/design/04-运行时.md §8.5). A task runs either in the foreground — // where it is promoted to the background automatically if it outlives a short // threshold — or in the background, where the agent can read its real-time // output. When a task ends, its result is returned to the agent over the // in-process bus on the subjects "task.output" and "task.done". // // A foreground task that completes within [autoBackgroundThreshold] is returned // to Start already in a terminal state. One that exceeds the threshold is // promoted: Start returns immediately with the task now KindBackground and // StatusRunning, and the run keeps going on its own goroutine. Background tasks // (KindBackground from the start) behave the same way but Start never waits. // // The task layer is built on top of [internal/runtime/exec].Provider — the // grant-based sandbox entry point — and never runs a command itself. Because the // Provider's Execute returns only the final [exec.ExecResult], the captured // Output is materialized when the task reaches a terminal state; the Output/ // ReadOutput methods are safe to poll at any point and a future streaming exec // API can be layered onto the same Task and Manager surface. package task import ( "context" "errors" "fmt" "sync" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" ) // Kind is how a task runs relative to the caller. type Kind string const ( // KindForeground runs synchronously: Start waits up to the auto-background // threshold, then either returns the finished task or promotes it. KindForeground Kind = "foreground" // KindBackground runs asynchronously: Start returns immediately and the agent // reads output through Output/ReadOutput. KindBackground Kind = "background" ) // Status is the lifecycle state of a task. type Status string const ( // StatusPending is the state before the task begins running. StatusPending Status = "pending" // StatusRunning is the state while the command executes. StatusRunning Status = "running" // StatusDone is the terminal state after a successful run. StatusDone Status = "done" // StatusFailed is the terminal state after a run that returned an error (a // non-zero exit code is NOT an error and is reported via ExitCode). StatusFailed Status = "failed" // StatusAborted is the terminal state after Abort or a parent context cancel. StatusAborted Status = "aborted" ) // isTerminal reports whether a status no longer transitions. func (s Status) isTerminal() bool { return s == StatusDone || s == StatusFailed || s == StatusAborted } // Output is a snapshot of a task's captured standard streams. type Output struct { // Stdout is the captured standard output. Stdout string // Stderr is the captured standard error. Stderr string } // TaskSpec describes a work item to run. It maps directly onto the fields the // [internal/runtime/exec] package needs. type TaskSpec struct { // Command is the argv vector to execute. It must be non-empty. Command []string // Background selects async execution when true (KindBackground) and // foreground/auto-promoting execution when false (KindForeground). Background bool // WorkDir is the process working directory (empty lets the provider pick). WorkDir string // Timeout bounds the run; zero uses exec.DefaultTimeout. Timeout time.Duration // MaxOutputBytes caps combined output; zero uses exec.DefaultMaxOutputBytes. MaxOutputBytes int } // Snapshot is an atomic, read-only view of a task's mutable state, taken under // the task's lock so callers observe a consistent state. type Snapshot struct { ID string Kind Kind Cmd []string Status Status Output Output StartedAt time.Time FinishedAt time.Time ExitCode int Truncated bool Err string } // Task is a single unit of work. ID, Kind, Cmd and StartedAt are fixed before // Start returns (Kind may be flipped foreground→background during promotion, // also before Start returns for a promoted task). The remaining state is mutated // by the task's run goroutine, so it must be read through the accessor methods // (Snapshot, Status, Output, ReadOutput, ...), which take the internal RWMutex. type Task struct { // ID uniquely identifies the task within a Manager. ID string // Kind reports whether the task is running in the background. Kind Kind // Cmd is the argv vector being executed. Cmd []string // StartedAt is when the task began running. StartedAt time.Time mu sync.RWMutex status Status output Output exitCode int truncated bool finErr error finishedAt time.Time ctx context.Context cancel context.CancelFunc done chan struct{} doneOnce sync.Once // m is the owning Manager, used to publish bus events on completion. m *Manager } // Status returns the current lifecycle state. func (t *Task) Status() Status { t.mu.RLock() defer t.mu.RUnlock() return t.status } // Output returns the current captured output. It never blocks and never waits // for the task to finish, so it can be polled while a background task runs. func (t *Task) Output() Output { t.mu.RLock() defer t.mu.RUnlock() return t.output } // FinishedAt returns when the task reached a terminal state, or the zero value // while it is still running. func (t *Task) FinishedAt() time.Time { t.mu.RLock() defer t.mu.RUnlock() return t.finishedAt } // ExitCode returns the process exit code (0 for success; -1 when an error is // reported instead of an exit code). func (t *Task) ExitCode() int { t.mu.RLock() defer t.mu.RUnlock() return t.exitCode } // Truncated reports whether output was dropped because MaxOutputBytes was reached. func (t *Task) Truncated() bool { t.mu.RLock() defer t.mu.RUnlock() return t.truncated } // Error returns the error that caused a failed or aborted task, or nil. func (t *Task) Error() error { t.mu.RLock() defer t.mu.RUnlock() return t.finErr } // Snapshot returns an atomic view of the task's current state. func (t *Task) Snapshot() Snapshot { t.mu.RLock() defer t.mu.RUnlock() s := Snapshot{ ID: t.ID, Kind: t.Kind, Cmd: t.Cmd, Status: t.status, Output: t.output, StartedAt: t.StartedAt, FinishedAt: t.finishedAt, ExitCode: t.exitCode, Truncated: t.truncated, } if t.finErr != nil { s.Err = t.finErr.Error() } return s } // ReadOutput waits until the task reaches a terminal state (or until ctx is // canceled) and returns the final output together with any terminal error. func (t *Task) ReadOutput(ctx context.Context) (Output, error) { select { case <-t.done: return t.Output(), t.Error() case <-ctx.Done(): return Output{}, fmt.Errorf("task: read output %q: %w", t.ID, ctx.Err()) } } // setRunning transitions the task to the running state. It is called by the // Manager before exposing the task, so the status is always non-pending for a // task the caller has been handed. func (t *Task) setRunning() { t.mu.Lock() defer t.mu.Unlock() if !t.status.isTerminal() { t.status = StatusRunning } } // promoteBackground flips a foreground task to KindBackground once it has // outlived the auto-background threshold. It never changes once Start returns. func (t *Task) promoteBackground() { t.mu.Lock() defer t.mu.Unlock() if !t.status.isTerminal() { t.Kind = KindBackground } } // abort cancels the task, marking it aborted, and publishes its terminal state. // It is idempotent for a task that is already terminal. func (t *Task) abort() error { t.mu.Lock() if t.status.isTerminal() { t.mu.Unlock() return fmt.Errorf("task: abort %q: %w", t.ID, ErrAlreadyDone) } t.status = StatusAborted t.finishedAt = time.Now() t.mu.Unlock() t.cancel() t.finalize() return nil } // finish records the outcome of the execution and publishes the terminal state. // A task that was already aborted keeps its aborted status. func (t *Task) finish(res *exec.ExecResult, err error) { t.mu.Lock() if !t.status.isTerminal() { switch { case errors.Is(err, context.Canceled): t.status = StatusAborted case err != nil: t.status = StatusFailed default: t.status = StatusDone } t.finErr = err t.finishedAt = time.Now() if res != nil { t.output = Output{Stdout: res.Stdout, Stderr: res.Stderr} t.exitCode = res.ExitCode t.truncated = res.Truncated } } t.mu.Unlock() t.finalize() } // finalize closes the task's done channel and publishes its terminal output and // done events. It runs at most once per task. func (t *Task) finalize() { t.doneOnce.Do(func() { t.m.publishOutput(t) t.m.publishDone(t) close(t.done) }) }