From ebeecc7a3234eba6be161401396d987be781d26f Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 07:11:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(task):=20=E4=BB=BB=E5=8A=A1=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E6=A8=A1=E5=9E=8B=EF=BC=88=E5=90=8E=E5=8F=B0=E5=A3=B0?= =?UTF-8?q?=E6=98=8E/=E5=89=8D=E5=8F=B02s=E8=BD=AC=E5=90=8E=E5=8F=B0/bus?= =?UTF-8?q?=20task.done=20=E5=9B=9E=E8=BE=93=E5=87=BA/=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E8=BE=93=E5=87=BA=E8=AF=BB=E5=8F=96/Abort=20=E9=98=B2=E5=A4=B1?= =?UTF-8?q?=E6=8E=A7=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/runtime/task/manager.go | 295 ++++++++++++++++++++++++++ internal/runtime/task/manager_test.go | 235 ++++++++++++++++++++ internal/runtime/task/run.go | 59 ++++++ internal/runtime/task/task.go | 287 +++++++++++++++++++++++++ internal/runtime/task/task_test.go | 284 +++++++++++++++++++++++++ 5 files changed, 1160 insertions(+) create mode 100644 internal/runtime/task/manager.go create mode 100644 internal/runtime/task/manager_test.go create mode 100644 internal/runtime/task/run.go create mode 100644 internal/runtime/task/task.go create mode 100644 internal/runtime/task/task_test.go diff --git a/internal/runtime/task/manager.go b/internal/runtime/task/manager.go new file mode 100644 index 0000000..81398b4 --- /dev/null +++ b/internal/runtime/task/manager.go @@ -0,0 +1,295 @@ +package task + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sort" + "sync" + "sync/atomic" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" +) + +// autoBackgroundThreshold is how long a foreground task may run before Start +// promotes it to the background (DevDocs/design/04-运行时.md §8.5: "前台运行， +// 超过 2 秒自动移入后台"). It is overridable via WithAutoBackground. +const autoBackgroundThreshold = 2 * time.Second + +// Sentinel errors returned by this package. Use errors.Is to test for them; they +// are returned directly or wrapped with %w so the underlying cause stays visible. +var ( + // ErrNilProvider is returned by New when no exec.Provider is supplied. + ErrNilProvider = errors.New("task: nil exec provider") + // ErrNilLogger is returned by New when no logger is supplied. + ErrNilLogger = errors.New("task: nil logger") + // ErrEmptyCommand is returned by Start when spec.Command is empty. + ErrEmptyCommand = errors.New("task: command is empty") + // ErrNotFound is returned by Get/Abort for an unknown task id. + ErrNotFound = errors.New("task: not found") + // ErrAlreadyDone is returned by Abort for a task that already reached a + // terminal state. + ErrAlreadyDone = errors.New("task: already finished") + // ErrManagerClosed is returned by Start on a Manager that has been closed. + ErrManagerClosed = errors.New("task: manager closed") +) + +// DoneEvent is the payload published to the "task.done" subject when a task +// reaches a terminal state. It carries everything the agent needs to consume the +// result without a second lookup. +type DoneEvent struct { + ID string + Kind Kind + Status Status + Output Output + ExitCode int + Truncated bool + Error string + StartedAt time.Time + FinishedAt time.Time +} + +// OutputEvent is the payload published to the "task.output" subject as output +// becomes available. +type OutputEvent struct { + ID string + Kind Kind + Status Status + Output Output + Truncated bool +} + +// Manager owns the task registry and the execution lifecycle. It owns an +// in-process bus (or one injected via WithBus) and publishes task events there. +type Manager struct { + cfg *config.Config + prov exec.Provider + log *slog.Logger + bus bus.Bus + threshold time.Duration + + ctx context.Context + cancel context.CancelFunc + closed atomic.Bool + + mu sync.RWMutex + tasks map[string]*Task + nextID atomic.Uint64 + active atomic.Int64 +} + +// options holds the resolved construction options. +type options struct { + bus bus.Bus + threshold time.Duration +} + +// Option customizes a Manager built by New. +type Option func(*options) + +// WithBus injects an existing bus instead of letting the Manager create one. +// This is handy for tests that share a bus and for wiring a global bus. +func WithBus(b bus.Bus) Option { + return func(o *options) { o.bus = b } +} + +// WithAutoBackground overrides the foreground→background promotion threshold. +// Zero uses the default of 2 seconds. +func WithAutoBackground(d time.Duration) Option { + return func(o *options) { o.threshold = d } +} + +// New builds a Manager. The primary call form is New(cfg, p, log); the variadic +// options only refine the defaults. The manager takes a child of a fresh +// background context, so tasks are independent of the context passed to Start +// (which is honored only for a foreground wait and is never used to kill a +// background task). Close releases everything. +func New(cfg *config.Config, p exec.Provider, log *slog.Logger, opts ...Option) (*Manager, error) { + if p == nil { + return nil, ErrNilProvider + } + if log == nil { + return nil, ErrNilLogger + } + + o := options{threshold: autoBackgroundThreshold} + for _, opt := range opts { + opt(&o) + } + if o.bus == nil { + o.bus = bus.New() + } + + ctx, cancel := context.WithCancel(context.Background()) + m := &Manager{ + cfg: cfg, + prov: p, + log: logger.WithSource(log, "runtime/task"), + bus: o.bus, + threshold: o.threshold, + ctx: ctx, + cancel: cancel, + tasks: make(map[string]*Task), + } + m.log.Debug("task manager ready", "auto_background_ms", m.threshold.Milliseconds()) + if cfg != nil { + m.log.Debug("exec sandbox", "enabled", cfg.ExecSandbox.Enabled) + } + return m, nil +} + +// Bus returns the bus the Manager publishes task events on. Agents subscribe to +// "task.output" and "task.done" to receive task results. +func (m *Manager) Bus() bus.Bus { return m.bus } + +// Active returns the number of tasks whose run goroutine is currently executing. +func (m *Manager) Active() int { return int(m.active.Load()) } + +// Start schedules a task. For a background task it returns immediately. For a +// foreground task it waits up to the auto-background threshold: if the task +// finishes first it returns the terminal task, otherwise it promotes the task to +// the background and returns immediately. +func (m *Manager) Start(ctx context.Context, spec TaskSpec) (*Task, error) { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("task: start: %w", err) + } + if len(spec.Command) == 0 { + return nil, fmt.Errorf("task: start: %w", ErrEmptyCommand) + } + if m.closed.Load() { + return nil, fmt.Errorf("task: start: %w", ErrManagerClosed) + } + + t := &Task{ + ID: m.genID(), + Kind: kindFrom(spec.Background), + Cmd: append([]string(nil), spec.Command...), + StartedAt: time.Now(), + done: make(chan struct{}), + } + t.ctx, t.cancel = context.WithCancel(m.ctx) + t.m = m + m.add(t) + t.setRunning() + + req := m.buildRequest(spec) + go m.run(t, req) + + if spec.Background { + return t, nil + } + + // Foreground: wait for completion, promotion, or a parent-context cancel. + timer := time.NewTimer(m.threshold) + defer timer.Stop() + select { + case <-t.done: + return t, nil + case <-timer.C: + t.promoteBackground() + return t, nil + case <-ctx.Done(): + _ = t.abort() + return nil, fmt.Errorf("task: start %q: %w", t.ID, ctx.Err()) + } +} + +// List returns a snapshot of all tasks, newest first. The returned pointers are +// live; read their state through the accessor methods. +func (m *Manager) List(ctx context.Context) ([]*Task, error) { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("task: list: %w", err) + } + m.mu.RLock() + out := make([]*Task, 0, len(m.tasks)) + for _, t := range m.tasks { + out = append(out, t) + } + m.mu.RUnlock() + + sort.Slice(out, func(i, j int) bool { + return out[i].StartedAt.After(out[j].StartedAt) + }) + return out, nil +} + +// Get returns the task for id or ErrNotFound. +func (m *Manager) Get(ctx context.Context, id string) (*Task, error) { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("task: get %q: %w", id, err) + } + t := m.get(id) + if t == nil { + return nil, fmt.Errorf("task: get %q: %w", id, ErrNotFound) + } + return t, nil +} + +// Abort cancels the task for id. It returns ErrNotFound if there is no such task +// and ErrAlreadyDone if the task has already finished. +func (m *Manager) Abort(ctx context.Context, id string) error { + if err := ctx.Err(); err != nil { + return fmt.Errorf("task: abort %q: %w", id, err) + } + t := m.get(id) + if t == nil { + return fmt.Errorf("task: abort %q: %w", id, ErrNotFound) + } + return t.abort() +} + +// Close cancels every running task, releases their goroutines, and closes the +// bus. It is idempotent and safe to call concurrently. +func (m *Manager) Close() error { + if m.closed.CompareAndSwap(false, true) { + m.cancel() + return m.bus.Close() + } + return nil +} + +// buildRequest maps a TaskSpec onto the exec request, applying defaults. +func (m *Manager) buildRequest(spec TaskSpec) exec.ExecRequest { + req := exec.ExecRequest{ + Command: spec.Command, + WorkDir: spec.WorkDir, + Timeout: spec.Timeout, + MaxOutputBytes: spec.MaxOutputBytes, + } + if req.Timeout == 0 { + req.Timeout = exec.DefaultTimeout + } + if req.MaxOutputBytes == 0 { + req.MaxOutputBytes = exec.DefaultMaxOutputBytes + } + return req +} + +func (m *Manager) add(t *Task) { + m.mu.Lock() + m.tasks[t.ID] = t + m.mu.Unlock() +} + +func (m *Manager) get(id string) *Task { + m.mu.RLock() + defer m.mu.RUnlock() + return m.tasks[id] +} + +func (m *Manager) genID() string { + n := m.nextID.Add(1) + return fmt.Sprintf("task-%s-%d", time.Now().UTC().Format("20060102150405.000000000"), n) +} + +func kindFrom(background bool) Kind { + if background { + return KindBackground + } + return KindForeground +} diff --git a/internal/runtime/task/manager_test.go b/internal/runtime/task/manager_test.go new file mode 100644 index 0000000..213df87 --- /dev/null +++ b/internal/runtime/task/manager_test.go @@ -0,0 +1,235 @@ +package task + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" +) + +// waitTerminalNoFail polls until terminal without failing the test immediately. +func waitTerminalNoFail(tk *Task, timeout time.Duration) Snapshot { + s, _ := waitTerminal(tk, timeout) + return s +} + +// TestManager_Abort_CancelsTask_NoLeak verifies Abort marks the task aborted, +// releases the run goroutine, and that a second Abort is ErrAlreadyDone. +func TestManager_Abort_CancelsTask_NoLeak(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("never", nil, 10*time.Second)) + + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"sleep"}, Background: true}) + if err != nil { + t.Fatalf("Start: %v", err) + } + waitActive(t, m, 1, 2*time.Second) + + if err := m.Abort(context.Background(), tk.ID); err != nil { + t.Fatalf("Abort: %v", err) + } + if snap := waitTerminalNoFail(tk, 2*time.Second); snap.Status != StatusAborted { + t.Errorf("Status=%s, want aborted", snap.Status) + } + + if err := m.Abort(context.Background(), tk.ID); !errors.Is(err, ErrAlreadyDone) { + t.Errorf("second Abort: err=%v, want ErrAlreadyDone", err) + } + + // The run goroutine must have exited (no leak). + waitActive(t, m, 0, 2*time.Second) +} + +// TestManager_ListGet verifies List and Get behavior, including the not-found path. +func TestManager_ListGet(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("x", nil, 0)) + + a, err := m.Start(context.Background(), TaskSpec{Command: []string{"a"}, Background: true}) + if err != nil { + t.Fatalf("Start a: %v", err) + } + b, err := m.Start(context.Background(), TaskSpec{Command: []string{"b"}, Background: true}) + if err != nil { + t.Fatalf("Start b: %v", err) + } + + list, err := m.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(list) != 2 { + t.Fatalf("List len=%d, want 2", len(list)) + } + found := map[string]bool{} + for _, tk := range list { + found[tk.ID] = true + } + if !found[a.ID] || !found[b.ID] { + t.Errorf("List missing tasks: %v", found) + } + + got, err := m.Get(context.Background(), a.ID) + if err != nil { + t.Fatalf("Get: %v", err) + } + if got.ID != a.ID { + t.Errorf("Get ID=%q, want %q", got.ID, a.ID) + } + + if _, err := m.Get(context.Background(), "nope"); !errors.Is(err, ErrNotFound) { + t.Errorf("Get unknown: err=%v, want ErrNotFound", err) + } + if err := m.Abort(context.Background(), "nope"); !errors.Is(err, ErrNotFound) { + t.Errorf("Abort unknown: err=%v, want ErrNotFound", err) + } +} + +// TestManager_TaskOutputTruncated verifies truncation and exit code propagation. +func TestManager_TaskOutputTruncated(t *testing.T) { + t.Parallel() + p := mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) { + return &exec.ExecResult{Stdout: "truncated", Stderr: "err", ExitCode: 1, Truncated: true}, nil + }} + m := newTestManager(t, p) + + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"big"}, Background: true}) + if err != nil { + t.Fatalf("Start: %v", err) + } + snap, ok := waitTerminal(tk, 2*time.Second) + if !ok { + t.Fatalf("task did not finish, status=%s", tk.Status()) + } + if !snap.Truncated { + t.Errorf("Truncated=false, want true") + } + if snap.ExitCode != 1 { + t.Errorf("ExitCode=%d, want 1", snap.ExitCode) + } + if snap.Output.Stderr != "err" { + t.Errorf("Stderr=%q, want err", snap.Output.Stderr) + } +} + +// TestManager_TaskFailed verifies a provider error surfaces as a failed task and +// is echoed in the done event. +func TestManager_TaskFailed(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("", exec.ErrCommandNotAllowed, 0)) + doneCh, unsub := subscribeDone(t, m) + defer unsub() + + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"bad"}, Background: true}) + if err != nil { + t.Fatalf("Start: %v", err) + } + snap, ok := waitTerminal(tk, 2*time.Second) + if !ok { + t.Fatalf("task did not finish, status=%s", tk.Status()) + } + if snap.Status != StatusFailed { + t.Errorf("Status=%s, want failed", snap.Status) + } + if snap.Err == "" { + t.Errorf("Err is empty, want an error string") + } + + select { + case e := <-doneCh: + if e.Status != StatusFailed { + t.Errorf("done Status=%s, want failed", e.Status) + } + if e.Error == "" { + t.Errorf("done Error is empty") + } + case <-time.After(2 * time.Second): + t.Fatal("no task.done event for failed task") + } +} + +// TestManager_Start_EmptyCommand verifies Start rejects an empty command. +func TestManager_Start_EmptyCommand(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockProvider{}) + if _, err := m.Start(context.Background(), TaskSpec{}); !errors.Is(err, ErrEmptyCommand) { + t.Errorf("err=%v, want ErrEmptyCommand", err) + } +} + +// TestManager_Start_AfterClose verifies Start refuses after Close. +func TestManager_Start_AfterClose(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockProvider{}) + if err := m.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if _, err := m.Start(context.Background(), TaskSpec{Command: []string{"a"}, Background: true}); !errors.Is(err, ErrManagerClosed) { + t.Errorf("err=%v, want ErrManagerClosed", err) + } +} + +// TestManager_WithAutoBackgroundThreshold verifies the promotion threshold is +// configurable via WithAutoBackground. +func TestManager_WithAutoBackgroundThreshold(t *testing.T) { + t.Parallel() + m, err := New(&config.Config{}, mockRun("x", nil, 500*time.Millisecond), discardLogger(), WithAutoBackground(50*time.Millisecond)) + if err != nil { + t.Fatalf("New: %v", err) + } + t.Cleanup(func() { _ = m.Close() }) + + start := time.Now() + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"s"}, Background: false}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if elapsed := time.Since(start); elapsed >= 500*time.Millisecond { + t.Errorf("Start waited %v; want promotion well before 500ms", elapsed) + } + if tk.Kind != KindBackground { + t.Errorf("Kind=%s, want background after promotion", tk.Kind) + } + if s, ok := waitTerminal(tk, 2*time.Second); !ok || s.Status != StatusDone { + t.Errorf("failed to finish after promotion, snap=%+v", s) + } +} + +// TestManager_ConcurrentTasks runs many background tasks concurrently and +// verifies none leak or corrupt shared state. +func TestManager_ConcurrentTasks(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(20 * time.Millisecond): + } + return &exec.ExecResult{Stdout: req.Command[0], ExitCode: 0}, nil + }}) + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"cmd"}, Background: true}) + if err != nil { + t.Errorf("Start: %v", err) + return + } + if s, ok := waitTerminal(tk, 2*time.Second); !ok || s.Status != StatusDone { + t.Errorf("concurrent task not done: %+v", s) + } + }() + } + wg.Wait() + + if n := m.Active(); n != 0 { + t.Errorf("Active()=%d, want 0 (no leaked goroutines)", n) + } +} diff --git a/internal/runtime/task/run.go b/internal/runtime/task/run.go new file mode 100644 index 0000000..97f91d4 --- /dev/null +++ b/internal/runtime/task/run.go @@ -0,0 +1,59 @@ +package task + +import ( + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" +) + +// run executes one task on its own goroutine. It always returns through finish, +// so the done channel is closed and the goroutine never leaks even on a panic. +func (m *Manager) run(t *Task, req exec.ExecRequest) { + m.active.Add(1) + defer m.active.Add(-1) + defer func() { + if r := recover(); r != nil { + m.log.Error("task run panicked", "id", t.ID, "panic", r) + t.finish(nil, fmt.Errorf("task: run panicked: %v", r)) + } + }() + + res, err := m.prov.Execute(t.ctx, req) + t.finish(res, err) +} + +// publishOutput fans the current output out on "task.output". +func (m *Manager) publishOutput(t *Task) { + s := t.Snapshot() + m.publish("task.output", OutputEvent{ + ID: s.ID, + Kind: s.Kind, + Status: s.Status, + Output: s.Output, + Truncated: s.Truncated, + }) +} + +// publishDone fans a task's terminal state out on "task.done". +func (m *Manager) publishDone(t *Task) { + s := t.Snapshot() + m.publish("task.done", DoneEvent{ + ID: s.ID, + Kind: s.Kind, + Status: s.Status, + Output: s.Output, + ExitCode: s.ExitCode, + Truncated: s.Truncated, + Error: s.Err, + StartedAt: s.StartedAt, + FinishedAt: s.FinishedAt, + }) +} + +// publish relays an event to the manager bus, logging a warning (never panicking) +// when the bus rejects it, e.g. after Close. +func (m *Manager) publish(subject string, payload any) { + if err := m.bus.Publish(m.ctx, subject, payload); err != nil { + m.log.Warn("task bus publish failed", "subject", subject, "err", err) + } +} diff --git a/internal/runtime/task/task.go b/internal/runtime/task/task.go new file mode 100644 index 0000000..1bc8be2 --- /dev/null +++ b/internal/runtime/task/task.go @@ -0,0 +1,287 @@ +// 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) + }) +} diff --git a/internal/runtime/task/task_test.go b/internal/runtime/task/task_test.go new file mode 100644 index 0000000..7945786 --- /dev/null +++ b/internal/runtime/task/task_test.go @@ -0,0 +1,284 @@ +package task + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" +) + +// mockProvider is a configurable exec.Provider. It mocks the interface, not a +// concrete type, per the test conventions. +type mockProvider struct { + execute func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) +} + +func (m mockProvider) Execute(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) { + if m.execute == nil { + return &exec.ExecResult{}, nil + } + return m.execute(ctx, req) +} + +// mockRun returns a provider that sleeps (unless sleep==0) and then returns the +// given output or error. It honors ctx cancellation so Abort releases it. +func mockRun(out string, err error, sleep time.Duration) mockProvider { + return mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) { + if sleep > 0 { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(sleep): + } + } + if err != nil { + return nil, err + } + return &exec.ExecResult{Stdout: out, ExitCode: 0}, nil + }} +} + +func discardLogger() *slog.Logger { + return logger.New(logger.Options{Level: "error", Out: io.Discard}) +} + +func newTestManager(t *testing.T, p exec.Provider) *Manager { + t.Helper() + m, err := New(&config.Config{}, p, discardLogger()) + if err != nil { + t.Fatalf("New: %v", err) + } + t.Cleanup(func() { _ = m.Close() }) + return m +} + +func subscribeDone(t *testing.T, m *Manager) (<-chan DoneEvent, func()) { + t.Helper() + ch := make(chan DoneEvent, 16) + unsub, err := m.Bus().Subscribe(context.Background(), "task.done", func(_ context.Context, e bus.Event) error { + if de, ok := e.Payload.(DoneEvent); ok { + ch <- de + } + return nil + }) + if err != nil { + t.Fatalf("subscribe task.done: %v", err) + } + return ch, unsub +} + +// waitTerminal polls a task until it reaches a terminal status or the timeout. +func waitTerminal(tk *Task, timeout time.Duration) (Snapshot, bool) { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if s := tk.Snapshot(); s.Status.isTerminal() { + return s, true + } + time.Sleep(5 * time.Millisecond) + } + return tk.Snapshot(), false +} + +func waitActive(t *testing.T, m *Manager, want int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if m.Active() == want { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("Active()=%d, want %d", m.Active(), want) +} + +// TestManager_New_Validates checks the constructor's nil guards. +func TestManager_New_Validates(t *testing.T) { + t.Parallel() + if _, err := New(&config.Config{}, nil, discardLogger()); !errors.Is(err, ErrNilProvider) { + t.Errorf("nil provider: err=%v, want ErrNilProvider", err) + } + if _, err := New(&config.Config{}, mockProvider{}, nil); !errors.Is(err, ErrNilLogger) { + t.Errorf("nil logger: err=%v, want ErrNilLogger", err) + } +} + +// TestManager_BackgroundTask_RealTimeOutput verifies a background task returns +// immediately, that Output() is readable (empty while running) during the run +// and populated after, and that ReadOutput returns the final output. +func TestManager_BackgroundTask_RealTimeOutput(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("hello world", nil, 400*time.Millisecond)) + + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "hi"}, Background: true}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if tk.Kind != KindBackground { + t.Errorf("Kind=%s, want background", tk.Kind) + } + if got := tk.Output().Stdout; got != "" { + t.Errorf("output during run=%q, want empty (still running)", got) + } + + out, err := tk.ReadOutput(context.Background()) + if err != nil { + t.Fatalf("ReadOutput: %v", err) + } + if out.Stdout != "hello world" { + t.Errorf("Stdout=%q, want hello world", out.Stdout) + } + if snap := tk.Snapshot(); snap.Status != StatusDone { + t.Errorf("Status=%s, want done", snap.Status) + } +} + +// TestManager_TaskOutputAndDoneEvents verifies both bus subjects fire, in order, +// with the expected payloads. +func TestManager_TaskOutputAndDoneEvents(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("hi out", nil, 0)) + + ch := make(chan bus.Event, 16) + subCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + unsub, err := m.Bus().Subscribe(subCtx, "task.>", func(_ context.Context, e bus.Event) error { + ch <- e + return nil + }) + if err != nil { + t.Fatalf("subscribe task.>: %v", err) + } + defer unsub() + + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo"}, Background: true}) + if err != nil { + t.Fatalf("Start: %v", err) + } + + var gotOutput, gotDone bool + timeout := time.After(2 * time.Second) + for !gotDone { + select { + case e := <-ch: + switch e.Subject { + case "task.output": + gotOutput = true + if oe := e.Payload.(OutputEvent); oe.Output.Stdout != "hi out" { + t.Errorf("output event Stdout=%q", oe.Output.Stdout) + } + case "task.done": + gotDone = true + de := e.Payload.(DoneEvent) + if de.ID != tk.ID { + t.Errorf("done ID=%q, want %q", de.ID, tk.ID) + } + if de.Status != StatusDone { + t.Errorf("done Status=%s, want done", de.Status) + } + if de.Output.Stdout != "hi out" { + t.Errorf("done Stdout=%q, want hi out", de.Output.Stdout) + } + } + case <-timeout: + t.Fatal("timeout waiting for bus events") + } + } + if !gotOutput { + t.Errorf("task.output event not received") + } +} + +// TestManager_ForegroundTask_FastCompletes verifies a foreground task that +// finishes inside the threshold is returned already done. +func TestManager_ForegroundTask_FastCompletes(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("fast out", nil, 0)) + + start := time.Now() + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "fast"}, Background: false}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { + t.Errorf("Start took %v, want < 1.5s", elapsed) + } + if tk.Kind != KindForeground { + t.Errorf("Kind=%s, want foreground", tk.Kind) + } + snap := tk.Snapshot() + if snap.Status != StatusDone { + t.Errorf("Status=%s, want done", snap.Status) + } + if snap.Output.Stdout != "fast out" { + t.Errorf("Stdout=%q, want fast out", snap.Output.Stdout) + } + if !snap.FinishedAt.After(snap.StartedAt) { + t.Errorf("FinishedAt=%v should be after StartedAt=%v", snap.FinishedAt, snap.StartedAt) + } +} + +// TestManager_ForegroundTask_PromotedAfterThreshold verifies a foreground task +// that outlives the threshold is promoted: Start returns early with Kind=background +// and Status=running, and the task completes in the background. +func TestManager_ForegroundTask_PromotedAfterThreshold(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("slow out", nil, 3*time.Second)) + + start := time.Now() + tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "slow"}, Background: false}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if elapsed := time.Since(start); elapsed >= 3*time.Second { + t.Errorf("Start waited %v; should promote before completion", elapsed) + } + if tk.Kind != KindBackground { + t.Errorf("Kind=%s, want background after promotion", tk.Kind) + } + if tk.Status() != StatusRunning { + t.Errorf("Status=%s, want running", tk.Status()) + } + + snap, ok := waitTerminal(tk, 4*time.Second) + if !ok { + t.Fatalf("task did not finish after promotion, status=%s", tk.Status()) + } + if snap.Status != StatusDone { + t.Errorf("Status=%s, want done", snap.Status) + } + if snap.Output.Stdout != "slow out" { + t.Errorf("Stdout=%q, want slow out", snap.Output.Stdout) + } +} + +// TestManager_Foreground_ParentContextCancelAborts verifies a foreground Start is +// aborted with the parent context's error when that context is canceled. +func TestManager_Foreground_ParentContextCancelAborts(t *testing.T) { + t.Parallel() + m := newTestManager(t, mockRun("hang", nil, 5*time.Second)) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(100 * time.Millisecond) + cancel() + }() + + start := time.Now() + _, err := m.Start(ctx, TaskSpec{Command: []string{"hang"}, Background: false}) + if err == nil { + t.Fatal("Start: want error on parent context cancel") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("err=%v, want context.Canceled", err) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Errorf("Start took %v, want < 2s", elapsed) + } +}