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 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/office/scheduler/retry.go

@@ -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
-}

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,

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)
+		}
+	}
+}

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{

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 {

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))
+	}
+}

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)