From 7f92d01f2ab749fa2f6d6c95bbab25a4e072c9a7 Mon Sep 17 00:00:00 2001 From: Henry Date: Wed, 2 Sep 2026 22:00:11 +0000 Subject: [PATCH] fix(office): stop a failing CEO agent from re-queuing itself forever (#3296) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Claude Sonnet 5 Co-authored-by: Carlos Florencio --- .../internal/office/scheduler/retry.go | 131 ------------------ apps/backend/internal/office/service/retry.go | 14 +- .../service/retry_ceo_self_escalation_test.go | 66 +++++++++ .../orchestrator/workflow_callbacks.go | 1 + .../workflow/engine/phase2_callbacks.go | 34 ++++- .../queue_run_ceo_self_escalation_test.go | 83 +++++++++++ .../office/system-design/scheduler-01.md | 2 +- 7 files changed, 194 insertions(+), 137 deletions(-) delete mode 100644 apps/backend/internal/office/scheduler/retry.go create mode 100644 apps/backend/internal/office/service/retry_ceo_self_escalation_test.go create mode 100644 apps/backend/internal/workflow/engine/queue_run_ceo_self_escalation_test.go diff --git a/apps/backend/internal/office/scheduler/retry.go b/apps/backend/internal/office/scheduler/retry.go deleted file mode 100644 index 98bae7ebb5..0000000000 --- a/apps/backend/internal/office/scheduler/retry.go +++ /dev/null @@ -1,131 +0,0 @@ -package scheduler - -import ( - "context" - "encoding/json" - "fmt" - "math/rand" - "time" - - "go.uber.org/zap" - - "github.com/kandev/kandev/internal/office/models" - "github.com/kandev/kandev/internal/office/service" -) - -// MaxRetryCount is the maximum number of retry attempts before marking failed. -const MaxRetryCount = 4 - -// mustJSON marshals v to JSON string, returning "{}" on error. Used by -// scheduler activity log emitters to attach structured details to a -// run/activity row. -func mustJSON(v interface{}) string { - b, err := json.Marshal(v) - if err != nil { - return "{}" - } - return string(b) -} - -// retryDelays defines the backoff schedule for run retries. -var retryDelays = []time.Duration{ - 2 * time.Minute, - 10 * time.Minute, - 30 * time.Minute, - 2 * time.Hour, -} - -// HandleRunFailure handles a failed run by scheduling a retry or -// marking it as permanently failed and escalating to the CEO agent. -func (ss *SchedulerService) HandleRunFailure( - ctx context.Context, run *models.Run, runErr error, -) error { - if run.RetryCount < MaxRetryCount { - return ss.scheduleRetry(ctx, run) - } - return ss.escalateFailure(ctx, run, runErr) -} - -// scheduleRetry re-queues the run with exponential backoff + jitter. -func (ss *SchedulerService) scheduleRetry(ctx context.Context, run *models.Run) error { - delay := retryDelayWithJitter(run.RetryCount) - retryAt := time.Now().UTC().Add(delay) - newCount := run.RetryCount + 1 - - ss.logger.Info("scheduling run retry", - zap.String("run_id", run.ID), - zap.Int("retry_count", newCount), - zap.Duration("delay", delay), - zap.Time("retry_at", retryAt)) - - return ss.repo.ScheduleRetry(ctx, run.ID, retryAt, newCount) -} - -// escalateFailure marks the run as permanently failed, logs an inbox -// item, and queues an agent_error run for the CEO agent. -func (ss *SchedulerService) escalateFailure( - ctx context.Context, run *models.Run, runErr error, -) error { - if err := ss.FailRun(ctx, run.ID); err != nil { - return fmt.Errorf("fail run: %w", err) - } - - agent, err := ss.svc.GetAgentFromConfig(ctx, run.AgentProfileID) - if err != nil { - ss.logger.Error("failed to get agent for escalation", - zap.String("agent_id", run.AgentProfileID), zap.Error(err)) - return nil - } - - // Log activity as agent error for inbox visibility. - errMsg := "unknown" - if runErr != nil { - errMsg = runErr.Error() - } - ss.svc.LogActivityWithRun(ctx, agent.WorkspaceID, "system", "scheduler", - "agent.error", "agent", run.AgentProfileID, - mustJSON(map[string]string{ - "run_id": run.ID, - "reason": run.Reason, - "error": errMsg, - }), run.ID, "") - - ss.logger.Warn("run permanently failed after max retries", - zap.String("run_id", run.ID), - zap.String("agent", agent.Name), - zap.Int("retry_count", run.RetryCount)) - - // Queue agent_error run for CEO if one exists. - ss.queueCEOAgentError(ctx, agent, run, errMsg) - return nil -} - -// queueCEOAgentError finds the CEO agent in the workspace and queues -// an agent_error run for it. -func (ss *SchedulerService) queueCEOAgentError( - ctx context.Context, agent *models.AgentInstance, - run *models.Run, errMsg string, -) { - ceos, err := ss.svc.ListAgentInstancesFiltered(ctx, agent.WorkspaceID, - service.AgentListFilter{Role: string(models.AgentRoleCEO)}) - if err != nil || len(ceos) == 0 { - return - } - payload := mustJSON(map[string]string{ - "agent_profile_id": run.AgentProfileID, - "run_id": run.ID, - "error": errMsg, - }) - _ = ss.QueueRun(ctx, ceos[0].ID, RunReasonAgentError, payload, "") -} - -// retryDelayWithJitter returns the base delay for a given retry index -// plus up to 25% random jitter. -func retryDelayWithJitter(retryIndex int) time.Duration { - if retryIndex < 0 || retryIndex >= len(retryDelays) { - retryIndex = len(retryDelays) - 1 - } - base := retryDelays[retryIndex] - jitter := time.Duration(rand.Int63n(int64(base / 4))) //nolint:gosec - return base + jitter -} diff --git a/apps/backend/internal/office/service/retry.go b/apps/backend/internal/office/service/retry.go index 8f532b2e22..d3c6716c31 100644 --- a/apps/backend/internal/office/service/retry.go +++ b/apps/backend/internal/office/service/retry.go @@ -146,8 +146,12 @@ func (s *Service) escalateFailure( return nil } -// queueCEOAgentError finds the CEO agent in the workspace and queues -// an agent_error run for it. +// queueCEOAgentError finds the CEO agent in the workspace and queues an +// agent_error run for it. When the failing run's agent IS the CEO, the +// escalation is skipped rather than re-queuing the CEO to handle its own +// failure — the run failure and inbox activity are already recorded by the +// caller, so skipping here loses no visibility, only the self-escalation +// loop. func (s *Service) queueCEOAgentError( ctx context.Context, agent *models.AgentInstance, run *models.Run, errMsg string, @@ -157,6 +161,12 @@ func (s *Service) queueCEOAgentError( if err != nil || len(ceos) == 0 { return } + if run.AgentProfileID == ceos[0].ID { + s.logger.Warn("skipped agent_error self-escalation for CEO's own failure", + zap.String("run_id", run.ID), + zap.String("ceo_agent_id", ceos[0].ID)) + return + } payload := mustJSON(map[string]string{ "agent_profile_id": run.AgentProfileID, "run_id": run.ID, diff --git a/apps/backend/internal/office/service/retry_ceo_self_escalation_test.go b/apps/backend/internal/office/service/retry_ceo_self_escalation_test.go new file mode 100644 index 0000000000..ccbed8bfd4 --- /dev/null +++ b/apps/backend/internal/office/service/retry_ceo_self_escalation_test.go @@ -0,0 +1,66 @@ +package service_test + +// Regression coverage for the Path B self-escalation loop: escalateFailure +// -> queueCEOAgentError (retry.go) previously queued a CEO-targeted +// agent_error run without checking whether the failing run's agent WAS the +// CEO. escalateFailure's FailRun -> transitionRunTerminal path never touches +// the consecutive-failure counter (HandleAgentFailure does; escalateFailure +// does not call it), so a deterministically-failing CEO run retried +// MaxRetryCount times, escalated to itself, and repeated forever — see +// docs/specs/office/system-design/scheduler-01.md's escalation clause. + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/office/service" +) + +// TestHandleRunFailure_CEOOwnFailure_DoesNotSelfEscalate proves the loop is +// broken: a run belonging to the CEO agent itself, exhausted on retries, +// must not produce a new agent_error run for that same CEO. The escalation +// is repeated 3 times to prove the loop stays broken, not just skipped once. +func TestHandleRunFailure_CEOOwnFailure_DoesNotSelfEscalate(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + ctx := context.Background() + + ceo := &models.AgentInstance{ + WorkspaceID: "ws-1", + Name: "ceo-self-fail", + Role: models.AgentRoleCEO, + Status: models.AgentStatusIdle, + } + if err := svc.CreateAgentInstance(ctx, ceo); err != nil { + t.Fatalf("create ceo: %v", err) + } + + for i := 0; i < 3; i++ { + if err := svc.QueueRun( + ctx, ceo.ID, service.RunReasonAgentError, `{}`, "", + ); err != nil { + t.Fatalf("queue ceo run %d: %v", i, err) + } + run, err := svc.ClaimNextRun(ctx) + if err != nil || run == nil { + t.Fatalf("claim ceo run %d: %v (run=%v)", i, err, run) + } + if run.AgentProfileID != ceo.ID { + t.Fatalf("claimed run %d belongs to %q, want ceo %q", i, run.AgentProfileID, ceo.ID) + } + run.RetryCount = service.MaxRetryCount + + if err := svc.HandleRunFailure(ctx, run, errForTest("ceo boom")); err != nil { + t.Fatalf("handle run failure %d: %v", i, err) + } + + next, err := svc.ClaimNextRun(ctx) + if err != nil { + t.Fatalf("claim after failure %d: %v", i, err) + } + if next != nil { + t.Fatalf("iteration %d: expected no self-escalation run queued, got agent=%q reason=%q", + i, next.AgentProfileID, next.Reason) + } + } +} diff --git a/apps/backend/internal/orchestrator/workflow_callbacks.go b/apps/backend/internal/orchestrator/workflow_callbacks.go index 73f4390115..dda65fccaf 100644 --- a/apps/backend/internal/orchestrator/workflow_callbacks.go +++ b/apps/backend/internal/orchestrator/workflow_callbacks.go @@ -45,6 +45,7 @@ func buildWorkflowCallbacks(svc *Service) engine.MapRegistry { CEOResolver: svc.engineCEOResolver, Primary: svc.enginePrimary, TaskSteps: workflowTargetStepResolver(svc.enginePrimary, svc.engineParticipants), + Logger: svc.logger, } if svc.engineParticipants != nil { r[engine.ActionQueueRunForEachParticipant] = engine.QueueRunForEachParticipantCallback{ diff --git a/apps/backend/internal/workflow/engine/phase2_callbacks.go b/apps/backend/internal/workflow/engine/phase2_callbacks.go index 269b4c8d18..0a27280b60 100644 --- a/apps/backend/internal/workflow/engine/phase2_callbacks.go +++ b/apps/backend/internal/workflow/engine/phase2_callbacks.go @@ -65,13 +65,16 @@ type TargetTaskStepResolver interface { } // QueueRunCallback executes the queue_run action by resolving Target/TaskID -// then enqueuing a run via RunQueueAdapter. +// then enqueuing a run via RunQueueAdapter. Logger is nil-safe (AC-24 +// convention, same as EnsureParticipantSeatCallback): a nil Logger skips +// warning emission for the workspace.ceo_agent self-escalation guard. type QueueRunCallback struct { Adapter RunQueueAdapter Participants ParticipantStore CEOResolver CEOAgentResolver Primary PrimaryAgentResolver TaskSteps TargetTaskStepResolver + Logger *logger.Logger } // Execute satisfies ActionCallback. @@ -143,7 +146,7 @@ func (c QueueRunCallback) resolveTarget( if err != nil { return nil, "", err } - agentIDs, err := c.resolveCEO(ctx, taskID) + agentIDs, err := c.resolveCEO(ctx, in, taskID) return agentIDs, stepID, err default: return nil, "", fmt.Errorf("queue_run: unsupported target %q", target) @@ -273,7 +276,13 @@ func roleSeatsForFanOut(ctx context.Context, store ParticipantStore, stepID, tas return seats, nil } -func (c QueueRunCallback) resolveCEO(ctx context.Context, taskID string) ([]string, error) { +// resolveCEO resolves the workspace CEO for a workspace.ceo_agent target. On +// an on_agent_error trigger whose failed agent IS the resolved CEO, it +// returns no agents instead of re-queueing the CEO to escalate its own +// failure to itself — the run failure and inbox activity are already +// recorded by the caller before this trigger fires, so skipping here loses +// no visibility, only the self-escalation loop. +func (c QueueRunCallback) resolveCEO(ctx context.Context, in ActionInput, taskID string) ([]string, error) { if c.CEOResolver == nil { return nil, fmt.Errorf("%w: queue_run target=workspace.ceo_agent requires CEOAgentResolver", ErrActionNotYetWired) } @@ -284,9 +293,28 @@ func (c QueueRunCallback) resolveCEO(ctx context.Context, taskID string) ([]stri if id == "" { return nil, fmt.Errorf("queue_run: workspace has no CEO agent profile for task %s", taskID) } + if in.Trigger == TriggerOnAgentError { + if payload, ok := in.Payload.(OnAgentErrorPayload); ok && payload.FailedAgentID == id { + c.recordCEOSelfEscalationSkipped(taskID, id) + return nil, nil + } + } return []string{id}, nil } +// recordCEOSelfEscalationSkipped logs the skipped self-escalation for +// operator visibility. Nil-safe: see the AC-24 convention note on +// QueueRunCallback. +func (c QueueRunCallback) recordCEOSelfEscalationSkipped(taskID, ceoAgentID string) { + if c.Logger == nil { + return + } + c.Logger.Warn("queue_run: skipped workspace.ceo_agent self-escalation", + zap.String("task_id", taskID), + zap.String("ceo_agent_id", ceoAgentID), + ) +} + // resolveTaskID maps the action's TaskID string into a concrete id, honouring // the "this" sentinel and the default-empty-means-this convention. func resolveTaskID(target, currentTaskID string) string { diff --git a/apps/backend/internal/workflow/engine/queue_run_ceo_self_escalation_test.go b/apps/backend/internal/workflow/engine/queue_run_ceo_self_escalation_test.go new file mode 100644 index 0000000000..a4d39c0706 --- /dev/null +++ b/apps/backend/internal/workflow/engine/queue_run_ceo_self_escalation_test.go @@ -0,0 +1,83 @@ +package engine + +import ( + "context" + "testing" +) + +// newOnAgentErrorInput builds a workspace.ceo_agent queue_run ActionInput +// carrying an on_agent_error trigger and payload, matching how +// office/service/event_subscribers.go's dispatchAgentErrorTrigger dispatches +// Path A (terminal agent-session failure). +func newOnAgentErrorInput(failedAgentID string) ActionInput { + in := newQueueRunInput("workspace.ceo_agent", "this") + in.Trigger = TriggerOnAgentError + in.Payload = OnAgentErrorPayload{FailedAgentID: failedAgentID} + return in +} + +// TestQueueRunCallback_TargetWorkspaceCEO_SkipsSelfEscalation covers the +// guard: an on_agent_error trigger whose FailedAgentID is the CEO's own +// resolved profile id must not re-queue the CEO. Without the guard this +// would queue the CEO to handle its own failure, and a deterministically +// failing CEO would re-trigger the same escalation on the next failure. +func TestQueueRunCallback_TargetWorkspaceCEO_SkipsSelfEscalation(t *testing.T) { + q := &fakeRunQueue{} + log, logs := newObservedSeatLogger(t) + cb := QueueRunCallback{Adapter: q, CEOResolver: fakeCEO{id: "ceo-agent"}, Logger: log} + + if _, err := cb.Execute(context.Background(), newOnAgentErrorInput("ceo-agent")); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(q.calls) != 0 { + t.Fatalf("expected 0 calls (self-escalation skipped), got %d: %+v", len(q.calls), q.calls) + } + + matches := logs.FilterMessage("queue_run: skipped workspace.ceo_agent self-escalation") + if matches.Len() != 1 { + t.Fatalf("expected one warning for self-escalation skip, got %d", matches.Len()) + } + fields := matches.All()[0].ContextMap() + if fields["task_id"] != "task-1" { + t.Fatalf("task_id field = %v, want task-1", fields["task_id"]) + } + if fields["ceo_agent_id"] != "ceo-agent" { + t.Fatalf("ceo_agent_id field = %v, want ceo-agent", fields["ceo_agent_id"]) + } +} + +// TestQueueRunCallback_TargetWorkspaceCEO_OnAgentErrorNonCEOStillQueues is +// the no-regression witness: a sub-agent's failure must still wake the CEO. +func TestQueueRunCallback_TargetWorkspaceCEO_OnAgentErrorNonCEOStillQueues(t *testing.T) { + q := &fakeRunQueue{} + cb := QueueRunCallback{Adapter: q, CEOResolver: fakeCEO{id: "ceo-agent"}} + + if _, err := cb.Execute(context.Background(), newOnAgentErrorInput("worker-agent")); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(q.calls) != 1 { + t.Fatalf("expected 1 call, got %d", len(q.calls)) + } + if q.calls[0].AgentProfileID != "ceo-agent" { + t.Fatalf("agent_profile_id = %q, want ceo-agent", q.calls[0].AgentProfileID) + } +} + +// TestQueueRunCallback_TargetWorkspaceCEO_NonAgentErrorTriggerUnaffected +// pins that the guard only applies to on_agent_error: any other trigger +// targeting workspace.ceo_agent queues normally even if the payload happens +// to be an OnAgentErrorPayload (defense against a too-broad type check). +func TestQueueRunCallback_TargetWorkspaceCEO_NonAgentErrorTriggerUnaffected(t *testing.T) { + q := &fakeRunQueue{} + cb := QueueRunCallback{Adapter: q, CEOResolver: fakeCEO{id: "ceo-agent"}} + + in := newOnAgentErrorInput("ceo-agent") + in.Trigger = TriggerOnComment + + if _, err := cb.Execute(context.Background(), in); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(q.calls) != 1 { + t.Fatalf("expected 1 call, got %d", len(q.calls)) + } +} diff --git a/docs/specs/office/system-design/scheduler-01.md b/docs/specs/office/system-design/scheduler-01.md index a83054ef04..3ed4d8090d 100644 --- a/docs/specs/office/system-design/scheduler-01.md +++ b/docs/specs/office/system-design/scheduler-01.md @@ -253,7 +253,7 @@ A 30-second buffer is added to any parsed time. If no pattern matches or the par Log fields gain `source: "rate_limit_parsed"` vs `source: "backoff"`, plus `parsed_reset_at` (UTC). -**Default backoff for non-rate-limit retries**: 4 attempts at `[2m, 10m, 30m, 2h]` with 25% jitter. After `MaxRetryCount` (4) failures, `escalateFailure` is called - the wakeup is marked `failed`, an `agent.error` inbox item is created, and the coordinator receives an `agent_error` wakeup. +**Default backoff for non-rate-limit retries**: 4 attempts at `[2m, 10m, 30m, 2h]` with 25% jitter. After `MaxRetryCount` (4) failures, `escalateFailure` is called - the wakeup is marked `failed`, an `agent.error` inbox item is created, and the coordinator receives an `agent_error` wakeup. When the failing run's agent IS the coordinator itself, the `agent_error` wakeup is not escalated to it (no self-targeted run is queued) - the `agent.error` inbox item is still created. ### Recovery sweep (unstarted Office tasks)