diff --git a/.github/scripts/e2e-tests-workflow-contract_test.py b/.github/scripts/e2e-tests-workflow-contract_test.py index d988998b3d..d238efb5d4 100644 --- a/.github/scripts/e2e-tests-workflow-contract_test.py +++ b/.github/scripts/e2e-tests-workflow-contract_test.py @@ -102,6 +102,15 @@ class DesktopE2EWorkflowContractTest(unittest.TestCase): self.assertTrue(separator, f"Lint workflow has no {trigger} trigger") self.assertNotIn(" paths:", trigger_block_text.split("\n ", 1)[0]) + def test_timing_lookup_requires_a_profile_artifact(self) -> None: + workflow = E2E_WORKFLOW.read_text(encoding="utf-8") + + self.assertIn("/actions/runs/", workflow) + self.assertIn("candidate.id", workflow) + self.assertIn("/artifacts?per_page=100", workflow) + self.assertIn('artifact.name === "e2e-timing-profile"', workflow) + self.assertIn("!artifact.expired", workflow) + if __name__ == "__main__": unittest.main() diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 14849d805a..198b8e74fe 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -179,7 +179,7 @@ jobs: - name: Build Docker E2E helper binaries run: make build-backend-linux-helpers - - name: Locate latest successful main timing profile + - name: Locate latest successful main timing profile artifact id: main-timing-run env: GH_TOKEN: ${{ github.token }} @@ -190,16 +190,17 @@ jobs: const fs = require("node:fs"); (async () => { - const url = + const runsUrl = process.env.GITHUB_API_URL + "/repos/" + process.env.GITHUB_REPOSITORY + - "/actions/workflows/e2e-tests.yml/runs?branch=main&status=success&per_page=20"; - const response = await fetch(url, { - headers: { - accept: "application/vnd.github+json", - authorization: "Bearer " + process.env.GH_TOKEN, - }, + "/actions/workflows/e2e-tests.yml/runs?branch=main&status=success&per_page=50"; + const headers = { + accept: "application/vnd.github+json", + authorization: "Bearer " + process.env.GH_TOKEN, + }; + const response = await fetch(runsUrl, { + headers, }); if (!response.ok) { console.log( @@ -211,14 +212,45 @@ jobs: return; } const payload = await response.json(); - const run = (payload.workflow_runs ?? []).find( - (candidate) => candidate.head_branch === "main" && candidate.conclusion === "success", - ); + let run; + for (const candidate of payload.workflow_runs ?? []) { + if (candidate.head_branch !== "main" || candidate.conclusion !== "success") continue; + const artifactsResponse = await fetch( + process.env.GITHUB_API_URL + + "/repos/" + + process.env.GITHUB_REPOSITORY + + "/actions/runs/" + + candidate.id + + "/artifacts?per_page=100", + { headers }, + ); + if (!artifactsResponse.ok) { + console.log( + "::warning::Unable to inspect artifacts for main E2E run " + + candidate.id + + " (" + + artifactsResponse.status + + "); trying an older run.", + ); + continue; + } + const artifactsPayload = await artifactsResponse.json(); + const hasTimingProfile = (artifactsPayload.artifacts ?? []).some( + (artifact) => artifact.name === "e2e-timing-profile" && !artifact.expired, + ); + if (hasTimingProfile) { + run = candidate; + break; + } + console.log( + "Skipping main E2E run " + candidate.id + ": no usable timing profile artifact.", + ); + } fs.appendFileSync(process.env.GITHUB_OUTPUT, "run_id=" + (run?.id ?? "") + "\n"); console.log( run - ? "Using successful main E2E run " + run.id + " for timing history." - : "No successful main E2E run found; using count fallback.", + ? "Using successful main E2E run " + run.id + " with a timing profile artifact." + : "No successful main E2E run with a timing profile artifact found; using count fallback.", ); })().catch((error) => { console.error(error); @@ -226,7 +258,7 @@ jobs: }); NODE - - name: Download latest main timing profile + - name: Download selected main timing profile if: steps.main-timing-run.outputs.run_id != '' continue-on-error: true uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 @@ -252,7 +284,7 @@ jobs: set -euo pipefail rm -rf "$GITHUB_WORKSPACE/e2e-manifests" if [[ -f "$GITHUB_WORKSPACE/e2e/previous-profile/timing-profile.json" ]]; then - echo "Planning from the latest successful main timing profile." + echo "Planning from the selected successful main timing profile." pnpm exec tsx e2e/scripts/plan-shards.ts \ --web-root "$GITHUB_WORKSPACE/apps/web" \ --output-dir "$GITHUB_WORKSPACE/e2e-manifests" \ diff --git a/apps/backend/internal/office/dashboard/engine_test_helpers_test.go b/apps/backend/internal/office/dashboard/engine_test_helpers_test.go index fc086d5137..6ad96d4b6b 100644 --- a/apps/backend/internal/office/dashboard/engine_test_helpers_test.go +++ b/apps/backend/internal/office/dashboard/engine_test_helpers_test.go @@ -52,6 +52,12 @@ func (fakeSessionResolver) GetTaskSessionByTaskID( return nil, taskmodels.ErrTaskSessionNotFound } +func (fakeSessionResolver) GetTaskSession( + _ context.Context, _ string, +) (*taskmodels.TaskSession, error) { + return nil, taskmodels.ErrTaskSessionNotFound +} + // noopCallbackRegistry satisfies engine.CallbackRegistry. Never consulted by // RecordParticipantDecision/EvaluateStepQuorum (only HandleTrigger's action // evaluation path uses it), so no test wires a real callback here. @@ -246,6 +252,15 @@ func (r singleTaskSessionResolver) GetTaskSessionByTaskID( return r.GetActiveTaskSessionByTaskID(ctx, taskID) } +func (r singleTaskSessionResolver) GetTaskSession( + _ context.Context, id string, +) (*taskmodels.TaskSession, error) { + if r.session != nil && id == r.session.ID { + return r.session, nil + } + return nil, taskmodels.ErrTaskSessionNotFound +} + // newTestEngineDispatcherWithReevaluation is like newTestEngineDispatcher but // wires a real TransitionStore and a session resolver that reports an // active session for taskID, unlocking the AC-13-17/AC-65 re-evaluation diff --git a/apps/backend/internal/office/engine_dispatcher/dispatcher.go b/apps/backend/internal/office/engine_dispatcher/dispatcher.go index e61f023684..878f919d85 100644 --- a/apps/backend/internal/office/engine_dispatcher/dispatcher.go +++ b/apps/backend/internal/office/engine_dispatcher/dispatcher.go @@ -32,6 +32,13 @@ var ErrNoSession = shared.ErrEngineNoSession type SessionResolver interface { GetActiveTaskSessionByTaskID(ctx context.Context, taskID string) (*taskmodels.TaskSession, error) GetTaskSessionByTaskID(ctx context.Context, taskID string) (*taskmodels.TaskSession, error) + // GetTaskSession resolves a session by its own id, independent of + // which task session is "active" or "latest". Used to target the + // exact session TriggerOnAgentError's payload names (Review F1): an + // office task can carry more than one session per task (one per + // (task, agent) pair), so a task-scoped active/latest lookup can + // return an unrelated sibling instead of the one that failed. + GetTaskSession(ctx context.Context, id string) (*taskmodels.TaskSession, error) } // EngineHandle is the engine surface the dispatcher needs. Defined as a @@ -133,7 +140,7 @@ func (d *Dispatcher) HandleTriggerHandled( if taskID == "" { return false, fmt.Errorf("task_id is required") } - session, err := d.resolveSession(ctx, taskID, trigger) + session, err := d.resolveSession(ctx, taskID, trigger, payload) if err != nil { return false, fmt.Errorf("resolve session: %w", err) } @@ -158,8 +165,22 @@ func (d *Dispatcher) HandleTriggerHandled( } func (d *Dispatcher) resolveSession( - ctx context.Context, taskID string, trigger engine.Trigger, + ctx context.Context, taskID string, trigger engine.Trigger, payload any, ) (*taskmodels.TaskSession, error) { + // Review round-1 F1: an office task carries one session per (task, + // agent) pair (executor_office.go's find-or-create), so + // office-default.yml's multi-agent workflow routinely has more than + // one session on a task at once. A task-scoped active/latest lookup + // can therefore resolve an unrelated sibling session instead of the + // one that actually failed. When the trigger's payload names the + // exact session, resolve it directly and skip the heuristics below + // entirely — this session is correct regardless of its current + // state, since it was already identified as the one that failed. + if trigger == engine.TriggerOnAgentError { + if sessionID := agentErrorFailedSessionID(payload); sessionID != "" { + return d.resolveSessionByID(ctx, taskID, sessionID) + } + } session, err := d.sessions.GetActiveTaskSessionByTaskID(ctx, taskID) if err == nil && session != nil { return session, nil @@ -167,7 +188,7 @@ func (d *Dispatcher) resolveSession( if err != nil && !errors.Is(err, taskmodels.ErrTaskSessionNotFound) { return nil, fmt.Errorf("active session lookup: %w", err) } - if trigger != engine.TriggerOnComment { + if trigger != engine.TriggerOnComment && trigger != engine.TriggerOnAgentError { return nil, nil } // Comment wakes are allowed after an office task's agent session has @@ -175,9 +196,16 @@ func (d *Dispatcher) resolveSession( // keyed by (taskID, sessionID), so a post-completion comment intentionally // resumes the latest reusable session's persisted machine state instead of // starting a fresh state machine here. + // + // Agent-error wakes reach this fallback only for legacy events with no + // FailedSessionID (payload predates session_id — the direct-resolve + // branch above handles every current event). It picks the latest + // session only when that session is itself FAILED, which is at best + // a guess at the failed session's identity, not a guarantee — kept + // only so an old queued event does not silently no-op. session, err = d.sessions.GetTaskSessionByTaskID(ctx, taskID) if err == nil && session != nil { - if !isReusableCommentSession(session.State) { + if !isReusableSessionForTrigger(trigger, session.State) { return nil, nil } return session, nil @@ -188,7 +216,46 @@ func (d *Dispatcher) resolveSession( return nil, nil } -func isReusableCommentSession(state taskmodels.TaskSessionState) bool { +// resolveSessionByID resolves a session by its own id — used for +// TriggerOnAgentError's FailedSessionID (Review F1) rather than a +// task-scoped active/latest lookup. It also verifies that the session belongs +// to the event task before returning it, so the engine never combines state +// from two task records. A not-found or mismatched id resolves to "no +// session" (nil, nil), consistent with every other branch of +// resolveSession, rather than an error: the session row may already be +// gone by the time the trigger dispatches, and that is a normal no-op, +// not a failure. +func (d *Dispatcher) resolveSessionByID( + ctx context.Context, taskID, sessionID string, +) (*taskmodels.TaskSession, error) { + session, err := d.sessions.GetTaskSession(ctx, sessionID) + if err != nil { + if errors.Is(err, taskmodels.ErrTaskSessionNotFound) { + return nil, nil + } + return nil, fmt.Errorf("failed session lookup: %w", err) + } + if session == nil || session.TaskID != taskID { + return nil, nil + } + return session, nil +} + +// agentErrorFailedSessionID extracts OnAgentErrorPayload.FailedSessionID +// when present. Empty for a payload that isn't OnAgentErrorPayload or +// predates session_id, in which case resolveSession falls back to the +// pre-F1 active/latest-FAILED heuristic. +func agentErrorFailedSessionID(payload any) string { + if p, ok := payload.(engine.OnAgentErrorPayload); ok { + return p.FailedSessionID + } + return "" +} + +func isReusableSessionForTrigger(trigger engine.Trigger, state taskmodels.TaskSessionState) bool { + if trigger == engine.TriggerOnAgentError { + return state == taskmodels.TaskSessionStateFailed + } return state == taskmodels.TaskSessionStateCompleted || state == taskmodels.TaskSessionStateIdle } diff --git a/apps/backend/internal/office/engine_dispatcher/dispatcher_test.go b/apps/backend/internal/office/engine_dispatcher/dispatcher_test.go index a9a557ab77..dd0295be74 100644 --- a/apps/backend/internal/office/engine_dispatcher/dispatcher_test.go +++ b/apps/backend/internal/office/engine_dispatcher/dispatcher_test.go @@ -24,6 +24,8 @@ type fakeSessions struct { latestSession *taskmodels.TaskSession activeErr error latestErr error + byID map[string]*taskmodels.TaskSession + byIDErr error } func (f *fakeSessions) GetActiveTaskSessionByTaskID(_ context.Context, _ string) (*taskmodels.TaskSession, error) { @@ -34,6 +36,16 @@ func (f *fakeSessions) GetTaskSessionByTaskID(_ context.Context, _ string) (*tas return f.latestSession, f.latestErr } +func (f *fakeSessions) GetTaskSession(_ context.Context, id string) (*taskmodels.TaskSession, error) { + if f.byIDErr != nil { + return nil, f.byIDErr + } + if session, ok := f.byID[id]; ok { + return session, nil + } + return nil, taskmodels.ErrTaskSessionNotFound +} + type fakeEngine struct { captured engine.HandleInput called bool @@ -462,6 +474,202 @@ func TestDispatcher_CompletedSessionCommentCanApplyTransition(t *testing.T) { } } +// TestDispatcher_UsesLatestFailedSessionForAgentError pins WO-05's E5 +// fix: TriggerOnAgentError must resolve the task's latest session when +// it is FAILED, since GetActiveTaskSessionByTaskID's state filter +// ('CREATED','STARTING','RUNNING','WAITING_FOR_INPUT') never includes +// FAILED. Without this, the on_agent_error dispatch added in +// event_subscribers.go would be nondeterministically inert whenever the +// orchestrator's AgentFailed subscriber wins the race and flips the +// session to FAILED before the office subscriber's engine dispatch +// runs. +func TestDispatcher_UsesLatestFailedSessionForAgentError(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeErr: taskmodels.ErrTaskSessionNotFound, + latestSession: &taskmodels.TaskSession{ + ID: "sess-failed", + State: taskmodels.TaskSessionStateFailed, + }, + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1"}, "agent_error:run-1") + if err != nil { + t.Fatalf("handle: %v", err) + } + if !eng.called { + t.Fatal("engine not invoked") + } + if eng.captured.SessionID != "sess-failed" { + t.Errorf("session id = %q, want sess-failed", eng.captured.SessionID) + } +} + +// TestDispatcher_SkipsLatestSessionForAgentErrorWhenNotFailed pins that +// on_agent_error only reuses a latest session that is actually FAILED — +// an unrelated non-terminal-failure session state must not be treated +// as the failed session's stand-in. +func TestDispatcher_SkipsLatestSessionForAgentErrorWhenNotFailed(t *testing.T) { + for _, state := range []taskmodels.TaskSessionState{ + taskmodels.TaskSessionStateCompleted, + taskmodels.TaskSessionStateIdle, + taskmodels.TaskSessionStateCancelled, + } { + t.Run(string(state), func(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeErr: taskmodels.ErrTaskSessionNotFound, + latestSession: &taskmodels.TaskSession{ + ID: "sess-latest", + State: state, + }, + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1"}, "agent_error:run-1") + if !errors.Is(err, ErrNoSession) { + t.Fatalf("err = %v, want ErrNoSession", err) + } + if eng.called { + t.Fatal("engine should not be invoked for a non-failed latest session") + } + }) + } +} + +// TestDispatcher_AgentErrorUsesFailedSessionIDNotLatestSibling pins the +// Review round-1 F1 fix: an office task has one session per (task, agent) +// (executor_office.go's GetTaskSessionByTaskAndAgent find-or-create), so +// office-default.yml's multi-agent workflow routinely has more than one +// session on a task. The latest-session-if-FAILED heuristic picks whichever +// session started last, not the one that actually failed — here the latest +// session is an unrelated sibling sitting IDLE (a normal post-turn state) +// while the failed session is older. The dispatcher must resolve +// FailedSessionID directly instead of falling through to that heuristic. +func TestDispatcher_AgentErrorUsesFailedSessionIDNotLatestSibling(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeErr: taskmodels.ErrTaskSessionNotFound, + latestSession: &taskmodels.TaskSession{ + ID: "sess-sibling-idle", + State: taskmodels.TaskSessionStateIdle, + }, + byID: map[string]*taskmodels.TaskSession{ + "sess-failed": { + ID: "sess-failed", TaskID: "task-1", State: taskmodels.TaskSessionStateFailed, + }, + }, + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1", FailedSessionID: "sess-failed"}, + "agent_error:run-1") + if err != nil { + t.Fatalf("handle: %v", err) + } + if !eng.called { + t.Fatal("engine not invoked") + } + if eng.captured.SessionID != "sess-failed" { + t.Errorf("session id = %q, want sess-failed (latest-sibling heuristic used instead)", + eng.captured.SessionID) + } +} + +// TestDispatcher_AgentErrorUsesFailedSessionIDOverActiveSibling pins F1's +// second failure mode: a sibling session sitting WAITING_FOR_INPUT is +// returned by the *active*-session lookup before the latest-session +// fallback is ever reached, so the dispatcher would target the wrong +// session's engine state (and, in production, its workflow step) even +// though the trigger never gets ErrNoSession. FailedSessionID must win +// over the active lookup whenever it is present. +func TestDispatcher_AgentErrorUsesFailedSessionIDOverActiveSibling(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeSession: &taskmodels.TaskSession{ + ID: "sess-sibling-waiting", + State: taskmodels.TaskSessionStateWaitingForInput, + }, + byID: map[string]*taskmodels.TaskSession{ + "sess-failed": { + ID: "sess-failed", TaskID: "task-1", State: taskmodels.TaskSessionStateFailed, + }, + }, + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1", FailedSessionID: "sess-failed"}, + "agent_error:run-1") + if err != nil { + t.Fatalf("handle: %v", err) + } + if eng.captured.SessionID != "sess-failed" { + t.Errorf("session id = %q, want sess-failed (active sibling used instead)", + eng.captured.SessionID) + } +} + +// TestDispatcher_AgentErrorFailedSessionIDNotFoundYieldsNoSession covers the +// direct-lookup miss: FailedSessionID names a session that no longer +// resolves (e.g. deleted). This must surface as a normal "no session" +// no-op, not an engine call against some other session and not a hard +// error. +func TestDispatcher_AgentErrorFailedSessionIDNotFoundYieldsNoSession(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeSession: &taskmodels.TaskSession{ + ID: "sess-unrelated-active", + State: taskmodels.TaskSessionStateWaitingForInput, + }, + // byID intentionally empty — "sess-missing" resolves to not-found + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1", FailedSessionID: "sess-missing"}, + "agent_error:run-1") + if !errors.Is(err, ErrNoSession) { + t.Fatalf("err = %v, want ErrNoSession", err) + } + if eng.called { + t.Fatal("engine should not be invoked when the failed session id can't be resolved") + } +} + +// TestDispatcher_AgentErrorRejectsFailedSessionFromAnotherTask pins the +// task/session ownership boundary for the direct FailedSessionID lookup. +func TestDispatcher_AgentErrorRejectsFailedSessionFromAnotherTask(t *testing.T) { + eng := &fakeEngine{} + sessions := &fakeSessions{ + activeSession: &taskmodels.TaskSession{ + ID: "sess-task-1-active", + TaskID: "task-1", + State: taskmodels.TaskSessionStateWaitingForInput, + }, + byID: map[string]*taskmodels.TaskSession{ + "sess-task-2-failed": { + ID: "sess-task-2-failed", TaskID: "task-2", State: taskmodels.TaskSessionStateFailed, + }, + }, + } + d := New(eng, sessions, logger.Default()) + + err := d.HandleTrigger(context.Background(), "task-1", engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{FailedAgentID: "agent-1", FailedSessionID: "sess-task-2-failed"}, + "agent_error:run-1") + if !errors.Is(err, ErrNoSession) { + t.Fatalf("err = %v, want ErrNoSession", err) + } + if eng.called { + t.Fatal("engine should not be invoked for a session owned by another task") + } +} + func TestDispatcher_DoesNotUseLatestSessionForNonCommentTriggers(t *testing.T) { eng := &fakeEngine{} sessions := &fakeSessions{ diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 9accb7d025..73bbd1a995 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -596,7 +596,36 @@ func (s *Service) handleAgentFailed(ctx context.Context, event *bus.Event) error // retry-by-classifier path lives behind HandleRunFailure for // rate-limit-retry callers; we deliberately do NOT call into it // here. See docs/specs/office-agent-error-handling. - return s.HandleAgentFailure(ctx, run, enrichModelFailureMessage(run, data.ErrorMessage)) + errMsg := enrichModelFailureMessage(run, data.ErrorMessage) + if err := s.HandleAgentFailure(ctx, run, errMsg); err != nil { + return err + } + s.dispatchAgentErrorTrigger(ctx, run, data.TaskID, data.SessionID, errMsg) + return nil +} + +// dispatchAgentErrorTrigger wakes the workspace CEO agent (WO-05) after a +// terminal agent-session failure. This is Path A in the office failure +// model; Path B (HandleRunFailure -> escalateFailure, pre-launch retry +// exhaustion) queues its own CEO run directly and never reaches this +// dispatcher, so the two mechanisms cannot double-fire for one failure. +// A dispatch error is logged rather than returned: it must not mask the +// failure bookkeeping HandleAgentFailure already committed. +func (s *Service) dispatchAgentErrorTrigger( + ctx context.Context, run *models.Run, taskID, sessionID, errMsg string, +) { + key := fmt.Sprintf("agent_error:%s", run.ID) + if err := s.dispatchEngineTrigger(ctx, taskID, engine.TriggerOnAgentError, + engine.OnAgentErrorPayload{ + FailedAgentID: run.AgentProfileID, + FailedSessionID: sessionID, + ErrorMessage: errMsg, + }, key); err != nil { + s.logger.Error("engine trigger on_agent_error failed", + zap.String("task_id", taskID), + zap.String("run_id", run.ID), + zap.Error(err)) + } } // enrichModelFailureMessage prepends the actionable "change the model" copy diff --git a/apps/backend/internal/office/service/event_subscribers_engine_test.go b/apps/backend/internal/office/service/event_subscribers_engine_test.go index fab622ccd3..6eed37afa0 100644 --- a/apps/backend/internal/office/service/event_subscribers_engine_test.go +++ b/apps/backend/internal/office/service/event_subscribers_engine_test.go @@ -5,9 +5,11 @@ import ( "sync" "testing" + "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/office/service" "github.com/kandev/kandev/internal/office/shared" "github.com/kandev/kandev/internal/runs/commentkeys" "github.com/kandev/kandev/internal/workflow/engine" @@ -258,3 +260,235 @@ func TestEngineDispatcher_NoSession_DropsTrigger(t *testing.T) { t.Fatalf("no-session: want 0 runs (no legacy fallback), got %d", len(runs)) } } + +// fallbackHandledRoutingDispatcher is a service.RoutingDispatcher fake whose +// HandlePostStartFailure always reports handled=true, letting tests pin the +// "routing already requeued the run" branch of handleAgentFailed. +type fallbackHandledRoutingDispatcher struct{} + +func (fallbackHandledRoutingDispatcher) DispatchWithRouting( + context.Context, *models.Run, *models.AgentInstance, service.LaunchContext, +) (bool, bool, error) { + return false, false, nil +} + +func (fallbackHandledRoutingDispatcher) HandlePostStartFailure( + context.Context, *models.Run, *models.AgentInstance, string, *streams.ProviderError, +) (bool, error) { + return true, nil +} + +func (fallbackHandledRoutingDispatcher) MarkRunSuccessHealth( + context.Context, *models.Run, *models.AgentInstance, +) { +} + +// queueTaskAssignedRunForAgentFailedTests queues + claims a run for taskID +// so an AgentFailed event resolves it via resolveLifecycleRun's +// task+agent fallback (GetClaimedRunByTaskAndAgent). +func queueTaskAssignedRunForAgentFailedTests( + t *testing.T, svc *service.Service, agentID, taskID string, +) *models.Run { + t.Helper() + ctx := context.Background() + if err := svc.QueueRun( + ctx, agentID, service.RunReasonTaskAssigned, `{"task_id":"`+taskID+`"}`, "", + ); err != nil { + t.Fatalf("queue run: %v", err) + } + run, err := svc.ClaimNextRun(ctx) + if err != nil || run == nil { + t.Fatalf("claim run: %v (run=%v)", err, run) + } + return run +} + +// publishAgentFailed publishes an AgentFailed lifecycle event for the +// given task/agent/session, mirroring what the lifecycle manager emits +// on a terminal agent-session failure (see TestRunLifecycle_ErrorEventEmitted). +func publishAgentFailed( + t *testing.T, eb bus.EventBus, taskID, agentID, sessionID, errMsg string, +) { + t.Helper() + evt := bus.NewEvent(events.AgentFailed, "test", map[string]string{ + "task_id": taskID, + "session_id": sessionID, + "error_message": errMsg, + "agent_profile_id": agentID, + }) + if err := eb.Publish(context.Background(), events.AgentFailed, evt); err != nil { + t.Fatalf("publish agent failed: %v", err) + } +} + +// TestEngineDispatcher_AgentFailed_RoutesToDispatcher pins WO-05: a +// terminal agent-session failure dispatches TriggerOnAgentError with a +// typed OnAgentErrorPayload so the workflow engine can queue_run the +// workspace CEO agent. Before this wiring, TriggerOnAgentError had zero +// production dispatchers and the CEO never learned about the failure. +func TestEngineDispatcher_AgentFailed_RoutesToDispatcher(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + disp := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(disp) + + ctx := context.Background() + createTestAgent(t, svc, "ws-1", "worker-1") + insertTestTask(t, svc, "task-1", "ws-1") + setTestTaskAssignee(t, svc, "task-1", "worker-1") + + run := queueTaskAssignedRunForAgentFailedTests(t, svc, "worker-1", "task-1") + + publishAgentFailed(t, eb, "task-1", "worker-1", "sess-err", "boom") + + calls := disp.Calls() + if len(calls) != 1 { + t.Fatalf("dispatcher calls = %d, want 1: %+v", len(calls), calls) + } + got := calls[0] + if got.taskID != "task-1" { + t.Errorf("taskID = %q, want task-1", got.taskID) + } + if got.trigger != engine.TriggerOnAgentError { + t.Errorf("trigger = %q, want %q", got.trigger, engine.TriggerOnAgentError) + } + payload, ok := got.payload.(engine.OnAgentErrorPayload) + if !ok { + t.Fatalf("payload type = %T, want engine.OnAgentErrorPayload", got.payload) + } + if payload.FailedAgentID != "worker-1" { + t.Errorf("payload.FailedAgentID = %q, want worker-1", payload.FailedAgentID) + } + if payload.FailedSessionID != "sess-err" { + t.Errorf("payload.FailedSessionID = %q, want sess-err", payload.FailedSessionID) + } + if payload.ErrorMessage != "boom" { + t.Errorf("payload.ErrorMessage = %q, want boom", payload.ErrorMessage) + } + wantOp := "agent_error:" + run.ID + if got.opID != wantOp { + t.Errorf("operationID = %q, want %q", got.opID, wantOp) + } + + // Failure bookkeeping still ran: the run itself is marked failed. + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + var found bool + for _, r := range runs { + if r.ID == run.ID { + found = true + if r.Status != service.RunStatusFailed { + t.Errorf("run status = %q, want failed", r.Status) + } + } + } + if !found { + t.Fatalf("run %s not found in ws-1 runs", run.ID) + } +} + +// TestEngineDispatcher_AgentFailed_SameRunTwice_OnlyOneCall pins that a +// duplicate AgentFailed event for the same run does not double-dispatch. +// The run is no longer claimed after the first failure, so +// resolveLifecycleRun's task+agent fallback returns sql.ErrNoRows for the +// replay and handleAgentFailed returns early before reaching dispatch. +func TestEngineDispatcher_AgentFailed_SameRunTwice_OnlyOneCall(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + disp := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(disp) + + createTestAgent(t, svc, "ws-1", "worker-1") + insertTestTask(t, svc, "task-1", "ws-1") + setTestTaskAssignee(t, svc, "task-1", "worker-1") + queueTaskAssignedRunForAgentFailedTests(t, svc, "worker-1", "task-1") + + publishAgentFailed(t, eb, "task-1", "worker-1", "sess-err", "boom") + publishAgentFailed(t, eb, "task-1", "worker-1", "sess-err", "boom") + + if calls := disp.Calls(); len(calls) != 1 { + t.Fatalf("dispatcher calls = %d, want 1 (replay guard)", len(calls)) + } +} + +// TestEngineDispatcher_AgentFailed_PostStartFallbackHandled_SkipsDispatch +// pins that when routing already requeued the run (tryPostStartFallback +// returns true), handleAgentFailed returns before reaching either the +// failure bookkeeping or the engine dispatch — the failure was not +// terminal, so there is nothing to escalate. +func TestEngineDispatcher_AgentFailed_PostStartFallbackHandled_SkipsDispatch(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + disp := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(disp) + svc.SetRoutingDispatcher(fallbackHandledRoutingDispatcher{}) + + createTestAgent(t, svc, "ws-1", "worker-1") + insertTestTask(t, svc, "task-1", "ws-1") + setTestTaskAssignee(t, svc, "task-1", "worker-1") + run := queueTaskAssignedRunForAgentFailedTests(t, svc, "worker-1", "task-1") + svc.ExecSQL(t, `UPDATE runs SET resolved_provider_id = 'test-provider' WHERE id = ?`, run.ID) + + publishAgentFailed(t, eb, "task-1", "worker-1", "sess-err", "boom") + + if calls := disp.Calls(); len(calls) != 0 { + t.Fatalf("dispatcher calls = %d, want 0 (post-start fallback handled)", len(calls)) + } +} + +// TestEngineDispatcher_PathBEscalation_DoesNotFireAgentErrorTrigger pins +// that the pre-launch retry-exhaustion escalation path (HandleRunFailure +// -> escalateFailure -> queueCEOAgentError, "Path B" in the WO-05 task +// plan) queues its CEO run directly and never touches the engine +// dispatcher — TriggerOnAgentError is exclusively a Path A (failed agent +// session) concern, so the two mechanisms cannot double-fire for one +// failure. +func TestEngineDispatcher_PathBEscalation_DoesNotFireAgentErrorTrigger(t *testing.T) { + svc, _ := newTestServiceWithBus(t) + disp := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(disp) + + ctx := context.Background() + ceo := &models.AgentInstance{ + WorkspaceID: "ws-1", + Name: "ceo-pathb", + Role: models.AgentRoleCEO, + Status: models.AgentStatusIdle, + } + if err := svc.CreateAgentInstance(ctx, ceo); err != nil { + t.Fatalf("create ceo: %v", err) + } + createTestAgent(t, svc, "ws-1", "worker-pathb") + + if err := svc.QueueRun( + ctx, "worker-pathb", service.RunReasonTaskAssigned, `{"task_id":"t1"}`, "", + ); err != nil { + t.Fatalf("queue: %v", err) + } + run, err := svc.ClaimNextRun(ctx) + if err != nil || run == nil { + t.Fatalf("claim: %v (run=%v)", err, run) + } + run.RetryCount = service.MaxRetryCount + + if err := svc.HandleRunFailure(ctx, run, errForTest("boom")); err != nil { + t.Fatalf("handle run failure: %v", err) + } + + if calls := disp.Calls(); len(calls) != 0 { + t.Fatalf("dispatcher calls = %d, want 0 (Path B does not fire the engine trigger)", len(calls)) + } + + next, err := svc.ClaimNextRun(ctx) + if err != nil { + t.Fatalf("claim ceo run: %v", err) + } + if next == nil { + t.Fatal("expected a queued CEO agent_error run") + } + if next.AgentProfileID != ceo.ID { + t.Errorf("agent = %q, want CEO %q", next.AgentProfileID, ceo.ID) + } + if next.Reason != service.RunReasonAgentError { + t.Errorf("reason = %q, want agent_error", next.Reason) + } +}