diff --git a/internal/runtime/events_subagent.go b/internal/runtime/events_subagent.go index 094564ab..abe258b6 100644 --- a/internal/runtime/events_subagent.go +++ b/internal/runtime/events_subagent.go @@ -24,6 +24,8 @@ const ( EventSubAgentStarted EventType = "subagent_started" // EventSubAgentProgress 在子代理执行每一步后触发。 EventSubAgentProgress EventType = "subagent_progress" + // EventSubAgentRetried 在子代理任务进入重试后触发。 + EventSubAgentRetried EventType = "subagent_retried" // EventSubAgentCompleted 在子代理成功结束后触发。 EventSubAgentCompleted EventType = "subagent_completed" // EventSubAgentFailed 在子代理失败结束后触发。 diff --git a/internal/runtime/subagent_run.go b/internal/runtime/subagent_run.go index 31bbd2b6..b33c3bc5 100644 --- a/internal/runtime/subagent_run.go +++ b/internal/runtime/subagent_run.go @@ -72,7 +72,22 @@ func (s *Service) RunSubAgentTask(ctx context.Context, input SubAgentTaskInput) emitSubAgentProgress(s, input, stepResult, stepErr) if stepErr != nil { - if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) { + if errors.Is(stepErr, context.DeadlineExceeded) { + _ = worker.Stop(subagent.StopReasonTimeout) + result, resultErr := worker.Result() + if resultErr != nil { + result = subagent.Result{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateFailed, + StopReason: subagent.StopReasonTimeout, + Error: errorText(stepErr), + } + } + emitSubAgentTerminal(s, ctx, input, result) + return result, stepErr + } + if errors.Is(stepErr, context.Canceled) { _ = worker.Stop(subagent.StopReasonCanceled) result, resultErr := worker.Result() if resultErr != nil { diff --git a/internal/runtime/subagent_run_test.go b/internal/runtime/subagent_run_test.go index 96f20f81..81770a56 100644 --- a/internal/runtime/subagent_run_test.go +++ b/internal/runtime/subagent_run_test.go @@ -5,7 +5,6 @@ import ( "errors" "strings" "testing" - "time" "neo-code/internal/runtime/controlplane" "neo-code/internal/subagent" @@ -182,6 +181,50 @@ func TestServiceRunSubAgentTaskFailureFlows(t *testing.T) { }) }) + t.Run("context deadline should emit failed timeout", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + service.SetSubAgentFactory(stubSubAgentFactory{ + create: func(role subagent.Role) (subagent.WorkerRuntime, error) { + return &stubSubAgentWorker{ + current: subagent.StateRunning, + stepErr: context.DeadlineExceeded, + resultErr: errors.New("no result"), + result: subagent.Result{ + Role: role, + TaskID: "task-timeout", + }, + }, nil + }, + }) + + result, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-timeout-emit", + Role: subagent.RoleReviewer, + Task: subagent.Task{ + ID: "task-timeout", + Goal: "review", + }, + }) + if err == nil || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("error = %v, want context deadline exceeded", err) + } + if result.State != subagent.StateFailed { + t.Fatalf("result state = %q, want %q", result.State, subagent.StateFailed) + } + if result.StopReason != subagent.StopReasonTimeout { + t.Fatalf("stop reason = %q, want %q", result.StopReason, subagent.StopReasonTimeout) + } + + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{ + EventSubAgentStarted, + EventSubAgentProgress, + EventSubAgentFailed, + }) + }) + t.Run("worker start failed by disallowed capability", func(t *testing.T) { t.Parallel() @@ -340,9 +383,8 @@ func TestServiceRunSubAgentTaskInputValidation(t *testing.T) { t.Fatalf("expected invalid role error") } - ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond) - defer cancel() - time.Sleep(2 * time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + cancel() if _, err := service.RunSubAgentTask(ctx, SubAgentTaskInput{ RunID: "sub-run-timeout", Role: subagent.RoleCoder, diff --git a/internal/subagent/scheduler.go b/internal/subagent/scheduler.go index a53d2c2d..f728a6ab 100644 --- a/internal/subagent/scheduler.go +++ b/internal/subagent/scheduler.go @@ -34,9 +34,12 @@ type schedulerState struct { running map[string]runningTask readySince map[string]time.Time started map[string]int + progress map[string]string outcomes chan taskOutcome } +var errSchedulerFailFast = errors.New("subagent: scheduler fail-fast triggered") + // NewScheduler 创建 Task DAG 调度器，并校验核心依赖是否可用。 func NewScheduler(store TodoStore, factory Factory, cfg SchedulerConfig) (*Scheduler, error) { if store == nil { @@ -58,6 +61,11 @@ func (s *Scheduler) Run(ctx context.Context) (ScheduleResult, error) { return ScheduleResult{}, err } + recovered, recoveredFailed, err := s.recoverInterruptedTodos() + if err != nil { + return ScheduleResult{}, err + } + initial := s.store.ListTodos() graph, err := buildTaskGraph(initial) if err != nil { @@ -69,6 +77,8 @@ func (s *Scheduler) Run(ctx context.Context) (ScheduleResult, error) { StartedAt: now, Total: len(graph.order), Retried: make(map[string]int), + Recovered: append([]string(nil), recovered...), + Failed: append([]string(nil), recoveredFailed...), } state := newSchedulerState(s.cfg.MaxConcurrency) finalize := func(current ScheduleResult) ScheduleResult { @@ -126,6 +136,86 @@ func (s *Scheduler) Run(ctx context.Context) (ScheduleResult, error) { } } +// recoverInterruptedTodos 在调度开始前恢复遗留 in_progress 任务，确保重启后可继续推进。 +func (s *Scheduler) recoverInterruptedTodos() ([]string, []string, error) { + items := s.store.ListTodos() + if len(items) == 0 { + return nil, nil, nil + } + + now := s.cfg.Clock() + recovered := make([]string, 0, len(items)) + failed := make([]string, 0, len(items)) + for _, item := range items { + if item.Status != agentsession.TodoStatusInProgress { + continue + } + + reason := strings.TrimSpace(s.cfg.RecoveryReason) + if reason == "" { + reason = "recovered interrupted subagent execution" + } + retryLimit := item.RetryLimit + if retryLimit <= 0 { + retryLimit = s.cfg.MaxRetries + } + nextRetry := item.RetryCount + 1 + ownerType := "" + ownerID := "" + + var ( + status agentsession.TodoStatus + nextRetryAt time.Time + ) + if s.cfg.recoveryAsFailure() || (retryLimit > 0 && nextRetry > retryLimit) { + status = agentsession.TodoStatusFailed + nextRetryAt = time.Time{} + } else { + status = agentsession.TodoStatusBlocked + nextRetryAt = now + } + + patch := agentsession.TodoPatch{ + Status: &status, + OwnerType: &ownerType, + OwnerID: &ownerID, + FailureReason: &reason, + RetryCount: &nextRetry, + RetryLimit: &retryLimit, + NextRetryAt: &nextRetryAt, + } + if err := s.store.UpdateTodo(item.ID, patch, item.Revision); err != nil { + if isRevisionConflict(err) { + continue + } + return nil, nil, fmt.Errorf("subagent: recover interrupted todo %q: %w", item.ID, err) + } + recovered = append(recovered, item.ID) + if status == agentsession.TodoStatusFailed { + failed = append(failed, item.ID) + } + + if status == agentsession.TodoStatusBlocked { + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentRetried, + TaskID: item.ID, + Attempt: nextRetry, + Reason: "recovered", + At: now, + }) + } else { + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentFailed, + TaskID: item.ID, + Attempt: nextRetry, + Reason: reason, + At: now, + }) + } + } + return recovered, failed, nil +} + // newSchedulerState 初始化调度运行态，避免循环中重复分配映射与通道。 func newSchedulerState(maxConcurrency int) *schedulerState { buffer := maxConcurrency @@ -136,6 +226,7 @@ func newSchedulerState(maxConcurrency int) *schedulerState { running: make(map[string]runningTask, maxConcurrency), readySince: make(map[string]time.Time, 32), started: make(map[string]int, 32), + progress: make(map[string]string, 32), outcomes: make(chan taskOutcome, buffer), } } @@ -160,7 +251,7 @@ func (s *Scheduler) collectReadyTasks( depsSatisfied := dependenciesCompleted(item, snapshot) if !depsSatisfied { - if err := s.ensureBlocked(item, "dependency_unmet"); err != nil { + if err := s.ensureBlocked(item, "dependency_unmet", state); err != nil { return nil, err } continue @@ -183,6 +274,15 @@ func (s *Scheduler) collectReadyTasks( Running: len(state.running), At: now, }) + s.emitProgressIfChanged(state, SchedulerEvent{ + Type: SchedulerEventSubAgentProgress, + TaskID: item.ID, + Attempt: item.RetryCount + 1, + QueueSize: len(ready), + Running: len(state.running), + Reason: "retry_backoff", + At: now, + }) continue } ready = append(ready, item.Clone()) @@ -194,7 +294,7 @@ func (s *Scheduler) collectReadyTasks( } // ensureBlocked 将未满足依赖的任务收敛到 blocked 状态并发出可观测事件。 -func (s *Scheduler) ensureBlocked(item agentsession.TodoItem, reason string) error { +func (s *Scheduler) ensureBlocked(item agentsession.TodoItem, reason string, state *schedulerState) error { if item.Status != agentsession.TodoStatusBlocked { status := agentsession.TodoStatusBlocked patch := agentsession.TodoPatch{Status: &status} @@ -203,12 +303,25 @@ func (s *Scheduler) ensureBlocked(item agentsession.TodoItem, reason string) err } } + running := 0 + if state != nil { + running = len(state.running) + } + now := s.cfg.Clock() s.emit(SchedulerEvent{ Type: SchedulerEventBlocked, TaskID: item.ID, Reason: reason, - Running: 0, - At: s.cfg.Clock(), + Running: running, + At: now, + }) + s.emitProgressIfChanged(state, SchedulerEvent{ + Type: SchedulerEventSubAgentProgress, + TaskID: item.ID, + Attempt: item.RetryCount + 1, + Reason: reason, + Running: running, + At: now, }) return nil } @@ -318,6 +431,21 @@ func (s *Scheduler) startReadyTasks( Running: len(state.running), At: s.cfg.Clock(), }) + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentStarted, + TaskID: item.ID, + Attempt: attempt, + Running: len(state.running), + At: s.cfg.Clock(), + }) + s.emitProgressIfChanged(state, SchedulerEvent{ + Type: SchedulerEventSubAgentProgress, + TaskID: item.ID, + Attempt: attempt, + Running: len(state.running), + Reason: "running", + At: s.cfg.Clock(), + }) go s.executeTaskAsync(ctx, state.outcomes, taskOutcome{ id: item.ID, @@ -361,8 +489,16 @@ func (s *Scheduler) handleOneOutcome(ctx context.Context, state *schedulerState, case <-ctx.Done(): return ctx.Err() case outcome := <-state.outcomes: + running, ok := state.running[outcome.id] + if !ok { + return nil + } + if outcome.attempt != running.attempt { + return nil + } delete(state.running, outcome.id) delete(state.readySince, outcome.id) + delete(state.progress, outcome.id) return s.applyOutcome(outcome, state, summary) case <-time.After(s.cfg.PollInterval): return nil @@ -395,6 +531,13 @@ func (s *Scheduler) applyOutcome(outcome taskOutcome, state *schedulerState, sum Running: len(state.running), At: s.cfg.Clock(), }) + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentCompleted, + TaskID: current.ID, + Attempt: outcome.attempt, + Running: len(state.running), + At: s.cfg.Clock(), + }) return nil } @@ -419,8 +562,12 @@ func (s *Scheduler) handleTaskFailure( backoff := s.cfg.Backoff(nextRetryCount) nextRetryAt := s.cfg.Clock().Add(backoff) status := agentsession.TodoStatusBlocked + ownerType := "" + ownerID := "" patch := agentsession.TodoPatch{ Status: &status, + OwnerType: &ownerType, + OwnerID: &ownerID, FailureReason: &reason, RetryCount: &nextRetryCount, RetryLimit: &retryLimit, @@ -442,13 +589,25 @@ func (s *Scheduler) handleTaskFailure( Running: len(state.running), At: s.cfg.Clock(), }) + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentRetried, + TaskID: current.ID, + Attempt: nextRetryCount, + Reason: reason, + Running: len(state.running), + At: s.cfg.Clock(), + }) return nil } status := agentsession.TodoStatusFailed zeroRetryAt := time.Time{} + ownerType := "" + ownerID := "" patch := agentsession.TodoPatch{ Status: &status, + OwnerType: &ownerType, + OwnerID: &ownerID, FailureReason: &reason, RetryCount: &nextRetryCount, RetryLimit: &retryLimit, @@ -470,6 +629,17 @@ func (s *Scheduler) handleTaskFailure( Running: len(state.running), At: s.cfg.Clock(), }) + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentFailed, + TaskID: current.ID, + Attempt: nextRetryCount, + Reason: reason, + Running: len(state.running), + At: s.cfg.Clock(), + }) + if s.cfg.failFastEnabled() { + return fmt.Errorf("%w: task=%s reason=%s", errSchedulerFailFast, current.ID, reason) + } return nil } @@ -488,8 +658,12 @@ func (s *Scheduler) cancelRunningTodos(state *schedulerState, cause error) { if !ok || item.Status.IsTerminal() { continue } + ownerType := "" + ownerID := "" patch := agentsession.TodoPatch{ Status: &status, + OwnerType: &ownerType, + OwnerID: &ownerID, FailureReason: &reason, } if err := s.store.UpdateTodo(item.ID, patch, item.Revision); err != nil && !isRevisionConflict(err) { @@ -502,6 +676,13 @@ func (s *Scheduler) cancelRunningTodos(state *schedulerState, cause error) { Running: len(state.running), At: s.cfg.Clock(), }) + s.emit(SchedulerEvent{ + Type: SchedulerEventSubAgentCanceled, + TaskID: item.ID, + Reason: reason, + Running: len(state.running), + At: s.cfg.Clock(), + }) } } @@ -573,6 +754,25 @@ func (s *Scheduler) emit(event SchedulerEvent) { s.cfg.Observer(event) } +// emitProgressIfChanged 仅在任务进度状态变化时发射 progress 事件，避免轮询路径的重复噪声。 +func (s *Scheduler) emitProgressIfChanged(state *schedulerState, event SchedulerEvent) { + if event.Type != SchedulerEventSubAgentProgress { + s.emit(event) + return + } + if state == nil { + s.emit(event) + return + } + + key := fmt.Sprintf("%d|%s", event.Attempt, event.Reason) + if state.progress[event.TaskID] == key { + return + } + state.progress[event.TaskID] = key + s.emit(event) +} + // mapTodosByID 将 todo 列表转为 ID 索引映射，便于依赖与状态查询。 func mapTodosByID(items []agentsession.TodoItem) map[string]agentsession.TodoItem { result := make(map[string]agentsession.TodoItem, len(items)) @@ -684,6 +884,12 @@ func appendUniqueString(dst *[]string, value string) { // resolveTaskFailureReason 统一提取失败原因，保证回写文本稳定可读。 func resolveTaskFailureReason(outcome taskOutcome) string { + switch outcome.result.StopReason { + case StopReasonMaxSteps: + return "budget exhausted: max steps reached" + case StopReasonTimeout: + return "budget exhausted: timeout exceeded" + } if outcome.err != nil { if text := strings.TrimSpace(outcome.err.Error()); text != "" { return text diff --git a/internal/subagent/scheduler_executor.go b/internal/subagent/scheduler_executor.go index 807fd6a5..397974d3 100644 --- a/internal/subagent/scheduler_executor.go +++ b/internal/subagent/scheduler_executor.go @@ -27,7 +27,21 @@ func executeTaskWithFactory(ctx context.Context, factory Factory, input schedule for { stepResult, stepErr := worker.Step(ctx) if stepErr != nil { - if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) { + if errors.Is(stepErr, context.DeadlineExceeded) { + _ = worker.Stop(StopReasonTimeout) + result, resultErr := worker.Result() + if resultErr == nil { + return result, stepErr + } + return Result{ + Role: input.Role, + TaskID: input.Task.ID, + State: StateFailed, + StopReason: StopReasonTimeout, + Error: strings.TrimSpace(stepErr.Error()), + }, stepErr + } + if errors.Is(stepErr, context.Canceled) { _ = worker.Stop(StopReasonCanceled) result, resultErr := worker.Result() if resultErr == nil { diff --git a/internal/subagent/scheduler_test.go b/internal/subagent/scheduler_test.go index db1b7b95..6f9fa865 100644 --- a/internal/subagent/scheduler_test.go +++ b/internal/subagent/scheduler_test.go @@ -179,8 +179,17 @@ func TestSchedulerConfigNormalize(t *testing.T) { if role != cfg.DefaultRole { t.Fatalf("RoleSelector fallback = %q, want %q", role, cfg.DefaultRole) } - if got := cfg.Backoff(3); got != 3*time.Second { - t.Fatalf("Backoff(3) = %v, want 3s", got) + if got := cfg.Backoff(3); got != 4*time.Second { + t.Fatalf("Backoff(3) = %v, want 4s", got) + } + if cfg.FailureMode != SchedulerFailureContinueOnError { + t.Fatalf("FailureMode = %q, want %q", cfg.FailureMode, SchedulerFailureContinueOnError) + } + if cfg.RecoveryMode != SchedulerRecoveryRetry { + t.Fatalf("RecoveryMode = %q, want %q", cfg.RecoveryMode, SchedulerRecoveryRetry) + } + if cfg.RecoveryReason == "" { + t.Fatalf("RecoveryReason should not be empty") } if got := cfg.BudgetSelector(agentsession.TodoItem{}, cfg.DefaultBudget); got != cfg.DefaultBudget { t.Fatalf("BudgetSelector() = %+v, want %+v", got, cfg.DefaultBudget) @@ -643,6 +652,443 @@ func TestSchedulerRunRetryAndGiveUp(t *testing.T) { } } +func TestSchedulerRunRecoveryFromInProgress(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + { + ID: "recover-me", + Content: "recover interrupted task", + Status: agentsession.TodoStatusInProgress, + RetryLimit: 2, + }, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = attempt + _ = input + return successStep(taskID), nil + }) + + var ( + mu sync.Mutex + events []SchedulerEvent + ) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 1, + PollInterval: 2 * time.Millisecond, + Backoff: func(attempt int) time.Duration { + _ = attempt + return 0 + }, + Observer: func(event SchedulerEvent) { + mu.Lock() + events = append(events, event) + mu.Unlock() + }, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + result, err := scheduler.Run(ctx) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if !contains(result.Recovered, "recover-me") { + t.Fatalf("Recovered = %v, want recover-me", result.Recovered) + } + + item, ok := store.FindTodo("recover-me") + if !ok { + t.Fatalf("FindTodo(recover-me) not found") + } + if item.Status != agentsession.TodoStatusCompleted { + t.Fatalf("status = %q, want completed", item.Status) + } + + mu.Lock() + defer mu.Unlock() + if !hasEvent(events, SchedulerEventSubAgentRetried, "recover-me") { + t.Fatalf("expected subagent_retried event, events=%+v", events) + } +} + +func TestSchedulerRunRecoveryModeFail(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + { + ID: "recover-fail", + Content: "recover interrupted task", + Status: agentsession.TodoStatusInProgress, + RetryLimit: 2, + }, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = taskID + _ = attempt + _ = input + return successStep("unused"), nil + }) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 1, + PollInterval: 2 * time.Millisecond, + RecoveryMode: SchedulerRecoveryFail, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + result, err := scheduler.Run(ctx) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if !contains(result.Recovered, "recover-fail") { + t.Fatalf("Recovered = %v, want recover-fail", result.Recovered) + } + if !contains(result.Failed, "recover-fail") { + t.Fatalf("Failed = %v, want recover-fail", result.Failed) + } + + item, ok := store.FindTodo("recover-fail") + if !ok { + t.Fatalf("FindTodo(recover-fail) not found") + } + if item.Status != agentsession.TodoStatusFailed { + t.Fatalf("status = %q, want failed", item.Status) + } + if !strings.Contains(item.FailureReason, "recovered interrupted") { + t.Fatalf("FailureReason = %q, want recovered interrupted", item.FailureReason) + } +} + +func TestSchedulerRunRecoveryExceedRetryLimitMarkedFailed(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + { + ID: "recover-over-limit", + Content: "recover interrupted task", + Status: agentsession.TodoStatusInProgress, + RetryLimit: 1, + RetryCount: 1, + }, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = taskID + _ = attempt + _ = input + return successStep("unused"), nil + }) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 1, + PollInterval: 2 * time.Millisecond, + RecoveryMode: SchedulerRecoveryRetry, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + result, err := scheduler.Run(ctx) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if !contains(result.Recovered, "recover-over-limit") { + t.Fatalf("Recovered = %v, want recover-over-limit", result.Recovered) + } + if !contains(result.Failed, "recover-over-limit") { + t.Fatalf("Failed = %v, want recover-over-limit", result.Failed) + } + + item, ok := store.FindTodo("recover-over-limit") + if !ok { + t.Fatalf("FindTodo(recover-over-limit) not found") + } + if item.Status != agentsession.TodoStatusFailed { + t.Fatalf("status = %q, want failed", item.Status) + } +} + +func TestSchedulerRunFailureModes(t *testing.T) { + t.Parallel() + + t.Run("continue on error", func(t *testing.T) { + t.Parallel() + store := newSchedulerStore(t, []agentsession.TodoItem{ + {ID: "a", Content: "a"}, + {ID: "b", Content: "b"}, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = attempt + _ = input + if taskID == "a" { + return StepOutput{}, errors.New("boom-a") + } + return successStep(taskID), nil + }) + + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 2, + PollInterval: 2 * time.Millisecond, + FailureMode: SchedulerFailureContinueOnError, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + result, err := scheduler.Run(ctx) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if !contains(result.Failed, "a") || !contains(result.Succeeded, "b") { + t.Fatalf("result = %+v, want a failed and b succeeded", result) + } + }) + + t.Run("fail fast", func(t *testing.T) { + t.Parallel() + store := newSchedulerStore(t, []agentsession.TodoItem{ + {ID: "a", Content: "a"}, + {ID: "b", Content: "b"}, + }) + started := make(chan string, 2) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = attempt + _ = input + select { + case started <- taskID: + default: + } + if taskID == "a" { + return StepOutput{}, errors.New("boom-a") + } + <-ctx.Done() + return StepOutput{}, ctx.Err() + }) + + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 2, + PollInterval: 2 * time.Millisecond, + FailureMode: SchedulerFailureFailFast, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, err = scheduler.Run(ctx) + if err == nil || !errors.Is(err, errSchedulerFailFast) { + t.Fatalf("Run() error = %v, want fail-fast error", err) + } + + a, _ := store.FindTodo("a") + if a.Status != agentsession.TodoStatusFailed { + t.Fatalf("todo a status = %q, want failed", a.Status) + } + b, _ := store.FindTodo("b") + if b.Status != agentsession.TodoStatusCanceled { + t.Fatalf("todo b status = %q, want canceled", b.Status) + } + }) +} + +func TestSchedulerRunStandardEventSequence(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + {ID: "seq", Content: "sequence task", RetryLimit: 2}, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = taskID + _ = input + if attempt == 1 { + return StepOutput{}, errors.New("first attempt fail") + } + return successStep("seq"), nil + }) + + var ( + mu sync.Mutex + events []SchedulerEventType + ) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 1, + PollInterval: 2 * time.Millisecond, + Backoff: func(attempt int) time.Duration { + _ = attempt + return 0 + }, + Observer: func(event SchedulerEvent) { + mu.Lock() + switch event.Type { + case SchedulerEventSubAgentStarted, SchedulerEventSubAgentRetried, SchedulerEventSubAgentCompleted: + events = append(events, event.Type) + } + mu.Unlock() + }, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if _, err := scheduler.Run(ctx); err != nil { + t.Fatalf("Run() error = %v", err) + } + + mu.Lock() + defer mu.Unlock() + want := []SchedulerEventType{ + SchedulerEventSubAgentStarted, + SchedulerEventSubAgentRetried, + SchedulerEventSubAgentStarted, + SchedulerEventSubAgentCompleted, + } + if len(events) != len(want) { + t.Fatalf("events len = %d, want %d, got=%v", len(events), len(want), events) + } + for idx := range want { + if events[idx] != want[idx] { + t.Fatalf("events[%d] = %q, want %q, all=%v", idx, events[idx], want[idx], events) + } + } +} + +func TestSchedulerRunProgressEventDeduplicatedForRetryBackoff(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + { + ID: "backoff", + Content: "wait retry window", + Status: agentsession.TodoStatusPending, + RetryCount: 1, + RetryLimit: 3, + NextRetryAt: time.Now().Add(200 * time.Millisecond), + }, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = taskID + _ = attempt + _ = input + return successStep("unused"), nil + }) + + var ( + mu sync.Mutex + progressCount int + ) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{ + MaxConcurrency: 1, + PollInterval: 2 * time.Millisecond, + Observer: func(event SchedulerEvent) { + if event.Type != SchedulerEventSubAgentProgress || event.TaskID != "backoff" || event.Reason != "retry_backoff" { + return + } + mu.Lock() + progressCount++ + mu.Unlock() + }, + }) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond) + defer cancel() + if _, err := scheduler.Run(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Run() error = %v, want deadline exceeded", err) + } + + mu.Lock() + defer mu.Unlock() + if progressCount != 1 { + t.Fatalf("progressCount = %d, want 1", progressCount) + } +} + +func TestSchedulerHandleOneOutcomeIgnoresStaleAttempt(t *testing.T) { + t.Parallel() + + store := newSchedulerStore(t, []agentsession.TodoItem{ + {ID: "stale", Content: "task"}, + }) + factory := newScriptedFactory(func(ctx context.Context, taskID string, attempt int, input StepInput) (StepOutput, error) { + _ = ctx + _ = taskID + _ = attempt + _ = input + return successStep("stale"), nil + }) + scheduler, err := NewScheduler(store, factory, SchedulerConfig{}) + if err != nil { + t.Fatalf("NewScheduler() error = %v", err) + } + + state := newSchedulerState(1) + state.running["stale"] = runningTask{id: "stale", attempt: 2} + state.outcomes <- taskOutcome{ + id: "stale", + attempt: 1, + result: Result{ + State: StateSucceeded, + Output: Output{ + Summary: "done", + }, + }, + } + summary := &ScheduleResult{Retried: map[string]int{}} + if err := scheduler.handleOneOutcome(context.Background(), state, summary); err != nil { + t.Fatalf("handleOneOutcome() error = %v", err) + } + running, ok := state.running["stale"] + if !ok { + t.Fatalf("running task removed by stale outcome") + } + if running.attempt != 2 { + t.Fatalf("running attempt = %d, want 2", running.attempt) + } + item, _ := store.FindTodo("stale") + if item.Status != agentsession.TodoStatusPending { + t.Fatalf("todo status = %q, want pending", item.Status) + } + if err := store.ClaimTodo("stale", agentsession.TodoOwnerTypeSubAgent, "subagent-stale", item.Revision); err != nil { + t.Fatalf("ClaimTodo(stale) error = %v", err) + } + + state.outcomes <- taskOutcome{ + id: "stale", + attempt: 2, + result: Result{ + State: StateSucceeded, + Output: Output{ + Summary: "done", + }, + }, + } + if err := scheduler.handleOneOutcome(context.Background(), state, summary); err != nil { + t.Fatalf("handleOneOutcome() for latest attempt error = %v", err) + } + item, _ = store.FindTodo("stale") + if item.Status != agentsession.TodoStatusCompleted { + t.Fatalf("todo status = %q, want completed", item.Status) + } +} + func TestSchedulerRunRevisionConflict(t *testing.T) { t.Parallel() @@ -912,6 +1358,25 @@ func TestExecuteTaskWithFactoryBranches(t *testing.T) { } }) + t.Run("step timeout with result fallback", func(t *testing.T) { + t.Parallel() + result, err := executeTaskWithFactory(context.Background(), fakeFactory{ + create: func(role Role) (WorkerRuntime, error) { + _ = role + return &fakeWorker{ + stepErr: context.DeadlineExceeded, + resultErr: errors.New("no result"), + }, nil + }, + }, scheduleTaskInput{Role: RoleCoder, Task: Task{ID: "t", Goal: "g"}}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("error = %v, want context deadline exceeded", err) + } + if result.State != StateFailed || result.StopReason != StopReasonTimeout { + t.Fatalf("result = %+v, want timeout fallback", result) + } + }) + t.Run("step failed with result", func(t *testing.T) { t.Parallel() result, err := executeTaskWithFactory(context.Background(), fakeFactory{ @@ -1103,6 +1568,19 @@ func TestSchedulerHelpersCoverage(t *testing.T) { if got := defaultRetryBackoff(0); got != 0 { t.Fatalf("defaultRetryBackoff(0) = %v, want 0", got) } + bounded := defaultRetryBackoffWithBounds(time.Second, 8*time.Second) + if got := bounded(1); got != time.Second { + t.Fatalf("bounded(1) = %v, want 1s", got) + } + if got := bounded(2); got != 2*time.Second { + t.Fatalf("bounded(2) = %v, want 2s", got) + } + if got := bounded(4); got != 8*time.Second { + t.Fatalf("bounded(4) = %v, want 8s", got) + } + if got := bounded(8); got != 8*time.Second { + t.Fatalf("bounded(8) = %v, want capped 8s", got) + } cfg := (SchedulerConfig{MaxRetries: -1}).normalize() if cfg.MaxRetries != 0 { t.Fatalf("normalize MaxRetries = %d, want 0", cfg.MaxRetries) diff --git a/internal/subagent/scheduler_types.go b/internal/subagent/scheduler_types.go index 313f08e0..6ab9d2e7 100644 --- a/internal/subagent/scheduler_types.go +++ b/internal/subagent/scheduler_types.go @@ -37,10 +37,43 @@ type ContextFileSelector func(todo agentsession.TodoItem, snapshot map[string]ag // RetryBackoff 计算第 N 次重试的退避时长。 type RetryBackoff func(attempt int) time.Duration +// SchedulerFailureStrategy 描述任务失败后的调度策略。 +type SchedulerFailureStrategy string + +const ( + // SchedulerFailureContinueOnError 表示失败任务仅影响自身，调度继续推进其他可执行任务。 + SchedulerFailureContinueOnError SchedulerFailureStrategy = "continue_on_error" + // SchedulerFailureFailFast 表示任一任务进入不可重试失败后，立即中断本轮调度。 + SchedulerFailureFailFast SchedulerFailureStrategy = "fail_fast" +) + +// SchedulerRecoveryStrategy 描述调度器启动时如何处理遗留 in_progress 任务。 +type SchedulerRecoveryStrategy string + +const ( + // SchedulerRecoveryRetry 表示把遗留 in_progress 任务恢复到可重试状态。 + SchedulerRecoveryRetry SchedulerRecoveryStrategy = "retry" + // SchedulerRecoveryFail 表示把遗留 in_progress 任务直接标记失败。 + SchedulerRecoveryFail SchedulerRecoveryStrategy = "fail" +) + // SchedulerEventType 描述调度器可观测事件类型。 type SchedulerEventType string const ( + // SchedulerEventSubAgentStarted 对齐 issue #278 的标准开始事件。 + SchedulerEventSubAgentStarted SchedulerEventType = "subagent_started" + // SchedulerEventSubAgentProgress 对齐 issue #278 的标准进度事件。 + SchedulerEventSubAgentProgress SchedulerEventType = "subagent_progress" + // SchedulerEventSubAgentCompleted 对齐 issue #278 的标准完成事件。 + SchedulerEventSubAgentCompleted SchedulerEventType = "subagent_completed" + // SchedulerEventSubAgentFailed 对齐 issue #278 的标准失败事件。 + SchedulerEventSubAgentFailed SchedulerEventType = "subagent_failed" + // SchedulerEventSubAgentCanceled 对齐 issue #278 的标准取消事件。 + SchedulerEventSubAgentCanceled SchedulerEventType = "subagent_canceled" + // SchedulerEventSubAgentRetried 对齐 issue #278 的标准重试事件。 + SchedulerEventSubAgentRetried SchedulerEventType = "subagent_retried" + // SchedulerEventQueued 表示任务进入可调度队列。 SchedulerEventQueued SchedulerEventType = "queued" // SchedulerEventClaimed 表示任务已经被领取。 @@ -83,6 +116,7 @@ type ScheduleResult struct { Succeeded []string Failed []string Retried map[string]int + Recovered []string BlockedLeft []string } @@ -92,8 +126,13 @@ type SchedulerConfig struct { DefaultRole Role DefaultBudget Budget MaxRetries int + RetryBaseDelay time.Duration + RetryMaxDelay time.Duration WorkerIDPrefix string PollInterval time.Duration + FailureMode SchedulerFailureStrategy + RecoveryMode SchedulerRecoveryStrategy + RecoveryReason string Clock func() time.Time RoleSelector RoleSelector BudgetSelector BudgetSelector @@ -125,9 +164,27 @@ func (c SchedulerConfig) normalize() SchedulerConfig { if out.MaxRetries < 0 { out.MaxRetries = 0 } + if out.RetryBaseDelay <= 0 { + out.RetryBaseDelay = time.Second + } + if out.RetryMaxDelay <= 0 { + out.RetryMaxDelay = 30 * time.Second + } + if out.RetryMaxDelay < out.RetryBaseDelay { + out.RetryMaxDelay = out.RetryBaseDelay + } if out.PollInterval <= 0 { out.PollInterval = 200 * time.Millisecond } + if out.FailureMode != SchedulerFailureFailFast { + out.FailureMode = SchedulerFailureContinueOnError + } + if out.RecoveryMode != SchedulerRecoveryFail { + out.RecoveryMode = SchedulerRecoveryRetry + } + if out.RecoveryReason == "" { + out.RecoveryReason = "recovered interrupted subagent execution" + } if out.WorkerIDPrefix == "" { out.WorkerIDPrefix = "subagent" } @@ -141,7 +198,7 @@ func (c SchedulerConfig) normalize() SchedulerConfig { out.BudgetSelector = defaultBudgetSelector } if out.Backoff == nil { - out.Backoff = defaultRetryBackoff + out.Backoff = defaultRetryBackoffWithBounds(out.RetryBaseDelay, out.RetryMaxDelay) } if out.Capabilities == nil { out.Capabilities = func(todo agentsession.TodoItem) Capability { @@ -198,10 +255,48 @@ func defaultBudgetSelector(todo agentsession.TodoItem, defaults Budget) Budget { return defaults } -// defaultRetryBackoff 提供线性重试退避。 +// defaultRetryBackoff 提供指数退避默认策略。 func defaultRetryBackoff(attempt int) time.Duration { - if attempt <= 0 { - return 0 + return defaultRetryBackoffWithBounds(time.Second, 30*time.Second)(attempt) +} + +// defaultRetryBackoffWithBounds 提供指数退避并带上限保护。 +func defaultRetryBackoffWithBounds(base, max time.Duration) RetryBackoff { + if base <= 0 { + base = time.Second + } + if max <= 0 { + max = 30 * time.Second + } + if max < base { + max = base + } + + return func(attempt int) time.Duration { + if attempt <= 0 { + return 0 + } + delay := base + for i := 1; i < attempt; i++ { + if delay >= max/2 { + delay = max + break + } + delay *= 2 + } + if delay > max { + return max + } + return delay } - return time.Duration(attempt) * time.Second +} + +// failFastEnabled 判断调度器是否开启 fail-fast 策略。 +func (c SchedulerConfig) failFastEnabled() bool { + return c.FailureMode == SchedulerFailureFailFast +} + +// recoveryAsFailure 判断恢复阶段是否直接失败遗留任务。 +func (c SchedulerConfig) recoveryAsFailure() bool { + return c.RecoveryMode == SchedulerRecoveryFail } diff --git a/internal/subagent/worker.go b/internal/subagent/worker.go index d1efb3cf..6e6b79d2 100644 --- a/internal/subagent/worker.go +++ b/internal/subagent/worker.go @@ -137,7 +137,12 @@ func (w *worker) Step(ctx context.Context) (StepResult, error) { stepOutput, err := w.engine.RunStep(ctx, input) if err != nil { w.mu.Lock() - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + if errors.Is(err, context.DeadlineExceeded) { + result := w.finishLocked(StateFailed, StopReasonTimeout, Output{}, err) + w.mu.Unlock() + return StepResult{State: result.State, Done: true, Step: result.StepCount}, err + } + if errors.Is(err, context.Canceled) { result := w.finishLocked(StateCanceled, StopReasonCanceled, Output{}, nil) w.mu.Unlock() return StepResult{State: result.State, Done: true, Step: result.StepCount}, err diff --git a/internal/subagent/worker_test.go b/internal/subagent/worker_test.go index 895649c2..85bd91b1 100644 --- a/internal/subagent/worker_test.go +++ b/internal/subagent/worker_test.go @@ -596,6 +596,34 @@ func TestWorkerStepCancellationBranches(t *testing.T) { t.Fatalf("state=%q, want %q", wr.State(), StateCanceled) } }) + + t.Run("engine deadline transitions timeout state", func(t *testing.T) { + t.Parallel() + wr, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + _ = ctx + _ = input + return StepOutput{}, context.DeadlineExceeded + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := wr.Start(Task{ID: "t-step-timeout", Goal: "goal"}, Budget{MaxSteps: 3}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + if _, err := wr.Step(context.Background()); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded, got %v", err) + } + if wr.State() != StateFailed { + t.Fatalf("state=%q, want %q", wr.State(), StateFailed) + } + result, err := wr.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.StopReason != StopReasonTimeout { + t.Fatalf("stop reason=%q, want %q", result.StopReason, StopReasonTimeout) + } + }) } func TestWorkerAdditionalUncoveredBranches(t *testing.T) {