From 41d3565467e264f0ac4e2ef6898c63fc9b099472 Mon Sep 17 00:00:00 2001 From: Henry Date: Sun, 13 Sep 2026 17:22:45 +0800 Subject: [PATCH] fix(backend): tear down orphaned agent processes when a session never starts (#3394) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../runtime/lifecycle/execution_store.go | 33 ++ .../runtime/lifecycle/execution_store_test.go | 38 ++ .../agent/runtime/lifecycle/manager_events.go | 6 +- .../runtime/lifecycle/manager_events_test.go | 7 +- .../runtime/lifecycle/manager_execution.go | 8 + .../runtime/lifecycle/stall_activity_test.go | 108 ++++++ .../internal/agent/runtime/lifecycle/types.go | 8 +- apps/backend/internal/backendapp/adapters.go | 4 + .../office_routine_cron_to_session_test.go | 4 + .../simulated_agent_manager_test.go | 13 + ...ound_launch_extensibility_external_test.go | 3 + .../orchestrator/event_handlers_stall.go | 57 ++- .../orchestrator/event_handlers_stall_test.go | 359 ++++++++++++++++++ .../orchestrator/event_handlers_test.go | 10 + .../orchestrator/executor/executor.go | 14 + .../executor/executor_interaction.go | 298 ++++++++++++++- .../executor/executor_interaction_test.go | 283 ++++++++++++++ .../executor/executor_mocks_test.go | 8 + .../executor_stop_recovery_identity_test.go | 166 ++++++++ .../orchestrator/scheduler/scheduler_test.go | 4 + .../internal/orchestrator/task_operations.go | 37 +- ...sk_operations_stop_orphan_recovery_test.go | 221 +++++++++++ ...02-terminal-stall-owns-process-teardown.md | 86 +++++ .../never-started-stall-teardown/plan.md | 254 +++++++++++++ .../task-01-honest-inactivity-clock.md | 134 +++++++ .../task-02-terminal-stall-teardown.md | 133 +++++++ ...sk-03-stop-reaches-registered-execution.md | 268 +++++++++++++ .../requirements/agent-stall-recovery.md | 74 ++-- .../system-design/agent-stall-recovery.md | 152 ++++++++ docs/specs/tasks/README.md | 2 +- .../requirements/task-stop-reachability.md | 81 ++++ .../system-design/task-stop-reachability.md | 129 +++++++ 32 files changed, 2932 insertions(+), 70 deletions(-) create mode 100644 apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go create mode 100644 apps/backend/internal/orchestrator/executor/executor_stop_recovery_identity_test.go create mode 100644 apps/backend/internal/orchestrator/task_operations_stop_orphan_recovery_test.go create mode 100644 docs/decisions/2026-09-02-terminal-stall-owns-process-teardown.md create mode 100644 docs/plans/never-started-stall-teardown/plan.md create mode 100644 docs/plans/never-started-stall-teardown/task-01-honest-inactivity-clock.md create mode 100644 docs/plans/never-started-stall-teardown/task-02-terminal-stall-teardown.md create mode 100644 docs/plans/never-started-stall-teardown/task-03-stop-reaches-registered-execution.md create mode 100644 docs/specs/agents/system-design/agent-stall-recovery.md create mode 100644 docs/specs/tasks/requirements/task-stop-reachability.md create mode 100644 docs/specs/tasks/system-design/task-stop-reachability.md diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_store.go b/apps/backend/internal/agent/runtime/lifecycle/execution_store.go index 45a078ba9a..d0288fc694 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/execution_store.go +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_store.go @@ -221,6 +221,39 @@ func (s *ExecutionStore) ActivePromptGeneration(executionID string) uint64 { return gen } +// ExecutionReference identifies one registered execution without requiring a +// later lookup through the session index. Both IDs are captured from the same +// execution-store snapshot so a caller can keep targeting the original +// execution if a session later acquires a replacement. +type ExecutionReference struct { + SessionID string + ExecutionID string +} + +// ListExecutionsForTask returns the session and execution IDs of executions +// registered in-memory under taskID. It is a read-only snapshot, independent +// of any session's persisted database state. Task-scoped stop uses it to +// recover an execution whose session row is already terminal (for example, +// FAILED) but whose prior teardown attempt failed. +func (s *ExecutionStore) ListExecutionsForTask(taskID string) []ExecutionReference { + if taskID == "" { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + + var references []ExecutionReference + for _, execution := range s.executions { + if execution.TaskID == taskID && execution.SessionID != "" && execution.ID != "" { + references = append(references, ExecutionReference{ + SessionID: execution.SessionID, + ExecutionID: execution.ID, + }) + } + } + return references +} + // GetByTaskEnvironmentID returns any execution associated with a task environment ID. func (s *ExecutionStore) GetByTaskEnvironmentID(taskEnvironmentID string) (*AgentExecution, bool) { s.mu.RLock() diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go b/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go index 19c05313db..4136e208a2 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go @@ -237,3 +237,41 @@ func TestExecutionStore_ClaimPromptActivityRequiresCurrentIdentity(t *testing.T) t.Fatalf("ClaimPromptActivity after removal = %v, want ErrExecutionNotFound", err) } } + +// TestExecutionStore_ListsExecutionsForTask covers +// @covers AC-TASKS-TASK-STOP-REACHABILITY-001.1: the registry-only lookup a +// task-scoped stop uses to recover a session whose database row is terminal +// but whose execution is still registered. +func TestExecutionStore_ListsExecutionsForTask(t *testing.T) { + store := NewExecutionStore() + if err := store.Add(&AgentExecution{ID: "exec-1", SessionID: "session-1", TaskID: "task-a"}); err != nil { + t.Fatalf("Add exec-1: %v", err) + } + if err := store.Add(&AgentExecution{ID: "exec-2", SessionID: "session-2", TaskID: "task-a"}); err != nil { + t.Fatalf("Add exec-2: %v", err) + } + if err := store.Add(&AgentExecution{ID: "exec-3", SessionID: "session-3", TaskID: "task-b"}); err != nil { + t.Fatalf("Add exec-3: %v", err) + } + + got := store.ListExecutionsForTask("task-a") + want := map[string]string{"session-1": "exec-1", "session-2": "exec-2"} + if len(got) != len(want) { + t.Fatalf("ListExecutionsForTask(task-a) = %v, want %v", got, want) + } + for _, reference := range got { + if executionID, ok := want[reference.SessionID]; !ok || executionID != reference.ExecutionID { + t.Fatalf("ListExecutionsForTask(task-a) returned unexpected reference %#v", reference) + } + } + + if got := store.ListExecutionsForTask("task-does-not-exist"); len(got) != 0 { + t.Fatalf("ListExecutionsForTask(empty task) = %v, want none", got) + } + + store.Remove("exec-1") + got = store.ListExecutionsForTask("task-a") + if len(got) != 1 || got[0] != (ExecutionReference{SessionID: "session-2", ExecutionID: "exec-2"}) { + t.Fatalf("ListExecutionsForTask(task-a) after Remove(exec-1) = %v, want [{session-2 exec-2}]", got) + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index c6a8cbe976..5cf32cd287 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -550,13 +550,13 @@ func isTerminalToolUpdate(event agentctl.AgentEvent) bool { // accidentally re-arm a freshly-booted no-prompt session as Running. func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.AgentEvent) { _, isTurnContent := turnContentEventTypes[event.Type] - execution.lastActivityAtMu.Lock() - execution.lastActivityAt = time.Now() if isTurnContent { + execution.lastActivityAtMu.Lock() + execution.lastActivityAt = time.Now() execution.agentEventSincePrompt = true execution.promptActivityEpoch++ + execution.lastActivityAtMu.Unlock() } - execution.lastActivityAtMu.Unlock() // Gate firstActivityOnce on `Status != Ready` so a delayed metadata // event arriving after MarkBootReady can't accidentally fire diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go index cb581b6213..8e011b6672 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go @@ -1575,13 +1575,14 @@ func TestRecordActivity_MetadataDoesNotMarkPromptStarted(t *testing.T) { mgr, _ := createTestManagerWithTracking() execution := createTestExecution("exec-metadata", "task-1", "session-1") execution.agentEventSincePrompt = false - execution.lastActivityAt = time.Now().Add(-time.Minute) + dispatchTime := time.Now().Add(-time.Minute) + execution.lastActivityAt = dispatchTime mgr.recordActivity(execution, agentctl.AgentEvent{Type: "session_models"}) lastActivity, agentEventSeen, epoch := execution.promptActivitySnapshot() - if time.Since(lastActivity) > time.Second { - t.Fatalf("metadata event did not refresh last activity: %v ago", time.Since(lastActivity)) + if !lastActivity.Equal(dispatchTime) { + t.Fatalf("metadata event advanced last activity: got %v, want unchanged %v", lastActivity, dispatchTime) } if agentEventSeen { t.Fatal("metadata event marked the prompt as started") diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go index 3e724971bc..e5efa774c4 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go @@ -460,6 +460,14 @@ func (m *Manager) GetExecutionIDForSession(_ context.Context, sessionID string) return "", fmt.Errorf("%w: %s", ErrNoExecutionForSession, sessionID) } +// ListExecutionsForTask returns a snapshot of the session and execution IDs +// registered in-memory for taskID, independent of persisted session state. +// Executor.StopByTaskID uses the paired IDs to recover an execution left +// orphaned by a terminal session row without resolving a replacement later. +func (m *Manager) ListExecutionsForTask(taskID string) []ExecutionReference { + return m.executionStore.ListExecutionsForTask(taskID) +} + // GetACPSessionIDForSession returns the ACP conversation currently owned by a // live execution. The orchestrator uses this optional accessor after a context // reset to persist the new conversation immediately, instead of depending on diff --git a/apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go b/apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go new file mode 100644 index 0000000000..6c647e0bb9 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go @@ -0,0 +1,108 @@ +package lifecycle + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted covers +// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.9: a metadata-only stream (no +// turn-content event) must not keep resetting the stall clock, so the +// watchdog still fires five minutes after dispatch with NeverStarted true. +func TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eventBus := &MockEventBusWithTracking{} + sm := NewSessionManager(newSessionTestLogger(), make(chan struct{})) + sm.eventPublisher = NewEventPublisher(eventBus, newSessionTestLogger()) + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + promptDoneCh: make(chan PromptCompletionSignal, 1), + Status: v1.AgentStatusRunning, + } + execution.lastActivityAt = time.Now() + + mgr := &Manager{eventPublisher: sm.eventPublisher} + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + waitResult := make(chan error, 1) + go func() { + _, err := sm.waitForPromptDone(ctx, execution, 7) + waitResult <- err + }() + + stop := make(chan struct{}) + defer close(stop) + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-ticker.C: + mgr.recordActivity(execution, agentctl.AgentEvent{Type: "usage_update"}) + case <-stop: + return + } + } + }() + + time.Sleep(5 * time.Minute) + synctest.Wait() + + payload := lastStalledPayload(t, eventBus) + if !payload.NeverStarted { + t.Fatalf("NeverStarted = false, want true: a metadata-only stream must not reset the clock") + } + if got := countPublishedEvents(eventBus, "agent.stalled"); got != 1 { + t.Fatalf("published stalled events = %d, want 1", got) + } + + cancel() + if err := <-waitResult; !errors.Is(err, context.Canceled) { + t.Fatalf("waitForPromptDone error = %v, want context canceled", err) + } + }) +} + +// TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress covers +// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.2: recordActivity must only move +// lastActivityAt for a turnContentEventTypes event; a metadata frame leaves +// the snapshot's timestamp and agentEventSincePrompt untouched. +func TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress(t *testing.T) { + execution := &AgentExecution{ + ID: "test-exec", + Status: v1.AgentStatusRunning, + } + dispatchTime := time.Now().Add(-time.Minute) + execution.lastActivityAt = dispatchTime + + mgr := &Manager{eventPublisher: NewEventPublisher(&MockEventBusWithTracking{}, newSessionTestLogger())} + + mgr.recordActivity(execution, agentctl.AgentEvent{Type: "usage_update"}) + + lastActivity, agentEventSeen, _ := execution.promptActivitySnapshot() + if !lastActivity.Equal(dispatchTime) { + t.Fatalf("lastActivityAt = %v, want unchanged dispatch time %v after a metadata frame", lastActivity, dispatchTime) + } + if agentEventSeen { + t.Fatal("agentEventSincePrompt = true after a metadata-only frame") + } + + mgr.recordActivity(execution, agentctl.AgentEvent{Type: "message_chunk"}) + + lastActivity, agentEventSeen, _ = execution.promptActivitySnapshot() + if !lastActivity.After(dispatchTime) { + t.Fatalf("lastActivityAt did not advance after a turn-content event") + } + if !agentEventSeen { + t.Fatal("agentEventSincePrompt = false after a turn-content event") + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/types.go b/apps/backend/internal/agent/runtime/lifecycle/types.go index 7e6ad9fa29..e61edf2466 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/types.go @@ -221,7 +221,13 @@ type AgentExecution struct { promptFinished chan struct{} promptFinishedMu sync.Mutex - // Last time an agent event was received (for stall detection) + // Last time a turn-content event was received (for stall detection). + // Advanced only by recordActivity for turnContentEventTypes, plus + // armPromptActivity, markAgentActivity, and recordSteerActivity. A + // metadata-only event (usage_update, context_window, + // available_commands_update, session_info_update, ...) does not move it, + // so a never-started prompt cannot be kept alive by traffic that carries + // no evidence of a turn. lastActivityAt time.Time // agentEventSincePrompt is armed (false) on each prompt dispatch and set // true by the first genuine agent event (recordActivity/handleCompleteEvent) diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index d1132e0830..1627ac73b7 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/apps/backend/internal/backendapp/adapters.go @@ -900,6 +900,10 @@ func (a *lifecycleAdapter) GetExecutionIDForSession(ctx context.Context, session return a.mgr.GetExecutionIDForSession(ctx, sessionID) } +func (a *lifecycleAdapter) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference { + return a.mgr.ListExecutionsForTask(taskID) +} + func (a *lifecycleAdapter) GetRemoteRuntimeStatusBySession(ctx context.Context, sessionID string) (*executor.RemoteRuntimeStatus, error) { status, ok := a.mgr.GetRemoteStatusBySessionID(ctx, sessionID) if !ok || status == nil { diff --git a/apps/backend/internal/backendapp/office_routine_cron_to_session_test.go b/apps/backend/internal/backendapp/office_routine_cron_to_session_test.go index 9f81db8a22..a17cd58441 100644 --- a/apps/backend/internal/backendapp/office_routine_cron_to_session_test.go +++ b/apps/backend/internal/backendapp/office_routine_cron_to_session_test.go @@ -10,6 +10,7 @@ import ( "github.com/kandev/kandev/internal/agent/agents" client "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" @@ -445,6 +446,9 @@ func (m *stubAgentManager) EnsureWorkspaceExecutionForSession(_ context.Context, func (m *stubAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) { return "", nil } +func (m *stubAgentManager) ListExecutionsForTask(_ string) []lifecycle.ExecutionReference { + return nil +} func (m *stubAgentManager) GetGitLog( _ context.Context, _, _ string, _ int, _ string, ) (*client.GitLogResult, error) { diff --git a/apps/backend/internal/integration/simulated_agent_manager_test.go b/apps/backend/internal/integration/simulated_agent_manager_test.go index 35cb145eb3..ee8ac210ed 100644 --- a/apps/backend/internal/integration/simulated_agent_manager_test.go +++ b/apps/backend/internal/integration/simulated_agent_manager_test.go @@ -638,6 +638,19 @@ func (s *SimulatedAgentManagerClient) GetExecutionIDForSession(_ context.Context } return "", fmt.Errorf("no execution found for session %s", sessionID) } +func (s *SimulatedAgentManagerClient) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference { + s.mu.Lock() + defer s.mu.Unlock() + var references []lifecycle.ExecutionReference + for executionID, inst := range s.instances { + if inst.taskID == taskID && inst.sessionID != "" { + references = append(references, lifecycle.ExecutionReference{ + SessionID: inst.sessionID, ExecutionID: executionID, + }) + } + } + return references +} func (s *SimulatedAgentManagerClient) GetGitLog(_ context.Context, _, _ string, _ int, _ string) (*client.GitLogResult, error) { return nil, nil } diff --git a/apps/backend/internal/orchestrator/background_launch_extensibility_external_test.go b/apps/backend/internal/orchestrator/background_launch_extensibility_external_test.go index 0a74ce4a74..7c57e90e3d 100644 --- a/apps/backend/internal/orchestrator/background_launch_extensibility_external_test.go +++ b/apps/backend/internal/orchestrator/background_launch_extensibility_external_test.go @@ -158,6 +158,9 @@ func (extAgentManager) GetGitStatusFresh(context.Context, string) (*client.GitSt return nil, nil } func (extAgentManager) WaitForAgentctlReady(context.Context, string) error { return nil } +func (extAgentManager) ListExecutionsForTask(string) []lifecycle.ExecutionReference { + return nil +} var _ executor.AgentManagerClient = extAgentManager{} diff --git a/apps/backend/internal/orchestrator/event_handlers_stall.go b/apps/backend/internal/orchestrator/event_handlers_stall.go index 975b75f20e..af9ef82e80 100644 --- a/apps/backend/internal/orchestrator/event_handlers_stall.go +++ b/apps/backend/internal/orchestrator/event_handlers_stall.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "strings" + "time" "go.uber.org/zap" @@ -18,6 +19,12 @@ const ( metaKeyActionVisibility = "action_visibility" actionVisibilityRunning = "running" neverStartedNoticeContent = "Agent produced no output since start." + // neverStartedStopTimeout bounds the detached runtime teardown started by + // stopNeverStartedExecution. A forced stop skips the graceful agentctl + // call, so nothing downstream (e.g. the Docker backend's cleanup context) + // applies its own bound in that path - an unresponsive daemon would + // otherwise hang the teardown goroutine indefinitely. + neverStartedStopTimeout = 30 * time.Second ) // errAgentNeverStarted is the launch-failure error recorded when a stalled @@ -117,10 +124,58 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen zap.Error(err)) } if payload.NeverStarted { - s.recordSessionLaunchFailure(ctx, payload.TaskID, payload.SessionID, errAgentNeverStarted, session) + if s.recordSessionLaunchFailure(ctx, payload.TaskID, payload.SessionID, errAgentNeverStarted, session) { + s.stopNeverStartedExecution(ctx, payload) + } else { + s.logger.Warn("skipping never-started teardown: session was not durably recorded FAILED", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID)) + } } } +// stopNeverStartedExecution tears down the process behind a never-started +// stall. It runs after the session and task have already been recorded +// FAILED, so a teardown failure never changes that recorded state: FAILED and +// its launch-failure message stay authoritative either way. +// +// Teardown ownership is claimed synchronously, while handleAgentStalled still +// holds the per-session cancelInFlight guard, via claimExecutionTeardown +// directly rather than through Executor's RegisterExecutionStopOwner +// callback: that callback TryLocks the same guard, so it would always fail to +// reentrantly acquire it and silently no-op. claimExecutionTeardown's own +// contract is to be called while already holding that guard. +// +// The actual stop runs detached and bounded (mirroring Executor.scheduleStop) +// rather than inline: handleAgentStalled's caller is the synchronous event +// bus, so a stop that blocked here would hold the cancelInFlight guard for +// its whole duration - blocking the user's own Stop/Cancel/Delete for this +// exact session, plus the event bus's dispatch lock. A forced stop skips the +// graceful agentctl call, so the runtime backend applies no bound of its own; +// neverStartedStopTimeout is this call's only protection against a hung +// daemon. lifecycle.ErrExecutionNotFound means nothing is running and is +// treated as success; any other failure is logged and left for a later stop +// to retry against the still-registered execution. +func (s *Service) stopNeverStartedExecution(ctx context.Context, payload lifecycle.AgentStalledPayload) { + if payload.AgentExecutionID == "" || s.agentManager == nil { + return + } + s.claimExecutionTeardown(payload.SessionID, payload.AgentExecutionID, executionTeardownIntentForce) + stopCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), neverStartedStopTimeout) + go func() { + defer cancel() + err := s.agentManager.StopAgentWithReason(stopCtx, payload.AgentExecutionID, "agent never started before stall", true) + if err == nil || errors.Is(err, lifecycle.ErrExecutionNotFound) { + return + } + s.logger.Warn("failed to tear down never-started execution after stall", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), + zap.String("execution_id", payload.AgentExecutionID), + zap.Error(err)) + }() +} + func stallNoticeContent(payload lifecycle.AgentStalledPayload) string { if payload.NeverStarted { return neverStartedNoticeContent diff --git a/apps/backend/internal/orchestrator/event_handlers_stall_test.go b/apps/backend/internal/orchestrator/event_handlers_stall_test.go index d62df394a9..5d39c8c73c 100644 --- a/apps/backend/internal/orchestrator/event_handlers_stall_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_stall_test.go @@ -2,14 +2,31 @@ package orchestrator import ( "context" + "errors" "strings" "testing" + "time" "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) +// waitForStopAgentCall blocks until ch delivers the detached +// stopNeverStartedExecution goroutine's StopAgentWithReason call, or fails +// the test after a bounded wait. Never poll or sleep for this: the call is +// intentionally off handleAgentStalled's own goroutine. +func waitForStopAgentCall(t *testing.T, ch <-chan stopAgentCall) stopAgentCall { + t.Helper() + select { + case call := <-ch: + return call + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the detached StopAgentWithReason call") + return stopAgentCall{} + } +} + func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) @@ -219,6 +236,348 @@ func TestHandleAgentStalled_RejectsSettledOrStalePrompt(t *testing.T) { } } +// TestHandleAgentStalled_NeverStartedStopsExecution covers +// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.10: after the never-started +// branch records the session and task FAILED, it must also tear down the +// execution named by the payload, forced, exactly once. +func TestHandleAgentStalled_NeverStartedStopsExecution(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + stopCalls := make(chan stopAgentCall, 1) + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + stopAgentWithReasonFunc: func(_ context.Context, executionID, reason string, force bool) error { + stopCalls <- stopAgentCall{ExecutionID: executionID, Reason: reason, Force: force} + return nil + }, + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + taskRepo, + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + NeverStarted: true, + }) + + call := waitForStopAgentCall(t, stopCalls) + if call.ExecutionID != "execution-1" { + t.Fatalf("stopped execution = %q, want execution-1", call.ExecutionID) + } + if !call.Force { + t.Fatal("teardown stop was not forced") + } + + agentMgr.mu.Lock() + callCount := len(agentMgr.stopAgentWithReasonArgs) + agentMgr.mu.Unlock() + if callCount != 1 { + t.Fatalf("StopAgentWithReason calls = %d, want 1", callCount) + } + + after, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("get session after handling stall: %v", err) + } + if after.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %q, want FAILED", after.State) + } +} + +// TestHandleAgentStalled_NeverStartedStopRunsDetachedFromGuard covers R3-F1: +// stopNeverStartedExecution must not block handleAgentStalled's return on the +// runtime teardown. handleAgentStalled holds the per-session cancelInFlight +// guard for its own duration (released via defer on return), and that guard +// also serializes the user's own Stop/Cancel/Delete for this session - so if +// the teardown ran inline here, an unresponsive runtime backend would hold +// that guard, the synchronous event bus's dispatch lock, and the +// waitForPromptDone goroutine for as long as the backend stayed hung. +func TestHandleAgentStalled_NeverStartedStopRunsDetachedFromGuard(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + stopStarted := make(chan struct{}) + stopRelease := make(chan struct{}) + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + stopAgentWithReasonFunc: func(context.Context, string, string, bool) error { + close(stopStarted) + <-stopRelease + return nil + }, + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + taskRepo, + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + handlerDone := make(chan struct{}) + go func() { + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + NeverStarted: true, + }) + close(handlerDone) + }() + + select { + case <-handlerDone: + case <-time.After(2 * time.Second): + close(stopRelease) + t.Fatal("handleAgentStalled blocked on runtime teardown instead of returning once it was scheduled") + } + + select { + case <-stopStarted: + case <-time.After(2 * time.Second): + close(stopRelease) + t.Fatal("StopAgentWithReason was never invoked") + } + close(stopRelease) +} + +// TestHandleAgentStalled_NeverStartedClaimsTeardownOwnershipBeforeStop covers +// Review round 2 Finding 2: stopNeverStartedExecution must claim execution +// teardown ownership before stopping the agent, so the synchronously +// published agent.stopped event is recognized by handleAgentStopped's +// hasExecutionTeardownOwner check as already-owned instead of running that +// handler's full state-decision path. +func TestHandleAgentStalled_NeverStartedClaimsTeardownOwnershipBeforeStop(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + taskRepo, + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + if svc.hasExecutionTeardownOwner("session-1", "execution-1") { + t.Fatal("teardown ownership claimed before the stall was handled") + } + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + NeverStarted: true, + }) + + if !svc.hasExecutionTeardownOwner("session-1", "execution-1") { + t.Fatal("stopNeverStartedExecution did not claim teardown ownership before stopping") + } +} + +// TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails covers +// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.11: a teardown failure must not +// overwrite the recorded FAILED state or its launch-failure message. +func TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + stopAttempted := make(chan struct{}) + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + stopAgentWithReasonFunc: func(context.Context, string, string, bool) error { + defer close(stopAttempted) + return errors.New("agentctl unreachable") + }, + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + taskRepo, + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + NeverStarted: true, + }) + + select { + case <-stopAttempted: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the detached teardown attempt") + } + + after, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("get session after handling stall: %v", err) + } + if after.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %q, want FAILED despite teardown failure", after.State) + } + if after.ErrorMessage != errAgentNeverStarted.Error() { + t.Fatalf("session error message = %q, want %q", after.ErrorMessage, errAgentNeverStarted.Error()) + } +} + +// TestHandleAgentStalled_AdvisoryStallStopsNothing covers +// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.7: an advisory stall (agent +// produced output before stalling) must not tear down the execution. +func TestHandleAgentStalled_AdvisoryStallStopsNothing(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + newMockTaskRepo(), + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + ToolName: "shell", + ToolTitle: "Start dev server", + }) + + agentMgr.mu.Lock() + stopCalls := len(agentMgr.stopAgentWithReasonArgs) + agentMgr.mu.Unlock() + if stopCalls != 0 { + t.Fatalf("StopAgentWithReason calls = %d, want 0 for an advisory stall", stopCalls) + } +} + +// TestHandleAgentStalled_NeverStartedSkipsTeardownWhenFailedNotPersisted covers +// the review finding that a rejected/failed FAILED-transition CAS must not be +// followed by a forced process kill: killing the process without a durable +// FAILED record would leave the session claiming RUNNING with no process and +// no way for handleAgentStopped to reconcile it (stopNeverStartedExecution +// claims teardown ownership first, so the resulting agent.stopped event is +// ignored as already-owned). +func TestHandleAgentStalled_NeverStartedSkipsTeardownWhenFailedNotPersisted(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + } + agentMgr.currentPromptGeneration.Store(7) + agentMgr.currentPromptActivityEpoch.Store(1) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + taskRepo, + agentMgr, + ) + writeFailure := errors.New("database is read-only") + svc.repo = failSessionStateUpdateRepo{repoStore: repo, err: writeFailure} + svc.turnService = &repoTurnService{repo: repo} + if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil { + t.Fatalf("start active turn: %v", err) + } + svc.messageCreator = &mockMessageCreator{} + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ActivityEpoch: 1, + NeverStarted: true, + }) + + agentMgr.mu.Lock() + stopCalls := len(agentMgr.stopAgentWithReasonArgs) + agentMgr.mu.Unlock() + if stopCalls != 0 { + t.Fatalf("StopAgentWithReason calls = %d, want 0 when the FAILED transition did not persist", stopCalls) + } + + after, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("get session after handling stall: %v", err) + } + if after.State != models.TaskSessionStateRunning { + t.Fatalf("session state = %q, want RUNNING (unchanged) since the FAILED write never persisted", after.State) + } +} + func TestStallNoticeContentFallsBackWithoutTool(t *testing.T) { if got := stallNoticeContent(lifecycle.AgentStalledPayload{}); got != "Still waiting for the agent." { t.Fatalf("stallNoticeContent() = %q, want generic fallback", got) diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 385ab819a8..a525cd25eb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -357,6 +357,9 @@ type mockAgentManager struct { } // Optional current ACP session lookup used by reset-token generation tests. getACPSessionIDForSessionFunc func(string) (string, bool) + // Optional override for ListExecutionsForTask. When unset, the default + // implementation returns nil (no registry-recovered sessions). + listExecutionsForTaskFunc func(taskID string) []lifecycle.ExecutionReference // CancelAgent tracking. cancelAgentCalls counts every invocation. If // cancelAgentBlock is non-nil, CancelAgent blocks on it before returning; @@ -793,6 +796,13 @@ func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, session return "", fmt.Errorf("no execution found") } +func (m *mockAgentManager) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference { + if m.listExecutionsForTaskFunc != nil { + return m.listExecutionsForTaskFunc(taskID) + } + return nil +} + func (m *mockAgentManager) GetACPSessionIDForSession(sessionID string) (string, bool) { if m.getACPSessionIDForSessionFunc == nil { return "", false diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 6f661312a9..0d211a3683 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -242,6 +242,13 @@ var ( // must not start the process and must arbitrate exact-execution teardown // ownership before deciding whether to force-stop the registered runtime. ErrSessionStateSuperseded = errors.New("session state superseded by terminal transition") + // ErrOrphanRecoveryIncomplete means StopByTaskID stopped every session it + // found but could not load at least one registry-only orphan's row, so the + // task-scoped stop is not fully confirmed. Callers that already observed a + // successful stop should log this rather than treat it as a hard failure; + // it stays distinguishable from ErrExecutionNotFound so a retry keeps + // happening instead of being reported as a false all-clear. + ErrOrphanRecoveryIncomplete = errors.New("orphaned execution recovery incomplete") ) // SessionStateSupersededError records the terminal state that rejected a @@ -402,6 +409,13 @@ type AgentManagerClient interface { // Used to detect stale AgentExecutionID values in the database after restart. GetExecutionIDForSession(ctx context.Context, sessionID string) (string, error) + // ListExecutionsForTask returns a snapshot of the session and execution IDs + // registered in-memory for taskID, independent of persisted session state. + // StopByTaskID uses the paired IDs to recover a registered execution whose + // session row is already terminal in the database (for example, FAILED + // after a never-started stall whose teardown attempt failed). + ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference + // GetGitLog retrieves the git log for a session from baseCommit to HEAD. // If targetBranch is provided, uses dynamic merge-base calculation for accurate filtering. // Used for archive snapshot capture. Returns nil, nil if no execution exists. diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction.go b/apps/backend/internal/orchestrator/executor/executor_interaction.go index 517a7880d4..971de5d12d 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction.go @@ -125,35 +125,202 @@ func (e *Executor) StopSessionDetailed( // stopWithSession preserves the legacy error-only contract. It intentionally // keeps session persistence best-effort and schedules teardown whenever a live // execution was found; existing UI and cleanup callers rely on that behavior. +// +// Stop ownership is registered before the CANCELLED write, not after: a +// concurrent StartAgentProcess failure classifies CANCELLED as terminal and +// claims forced cleanup for itself only when no owner is registered yet. If +// the state write were visible first, that failure could win the claim and +// issue its own teardown while this call also schedules one. func (e *Executor) stopWithSession(ctx context.Context, session *models.TaskSession, reason string, force bool) error { - executionID, err := e.agentManager.GetExecutionIDForSession(ctx, session.ID) + sessionLock := e.getSessionLock(session.ID) + sessionLock.Lock() + defer sessionLock.Unlock() + + executionID, err := e.resolveExecutionIDForStop(ctx, session.ID) + if err != nil { + return err + } + + return e.stopWithSessionExecution(ctx, session, executionID, reason, force, true) +} + +// stopWithSessionExecution records stop ownership and schedules teardown for +// the supplied execution. When cancelSession is true, it also transitions the +// current session to CANCELLED. Callers must hold the session lock when they +// need to coordinate this decision with resume or launch. +func (e *Executor) stopWithSessionExecution( + ctx context.Context, + session *models.TaskSession, + executionID, reason string, + force, cancelSession bool, +) error { + e.logStop(session, executionID, reason, force) + e.registerStopOwner(session.ID, executionID, force) + if cancelSession { + // A session already in a terminal state carries its own outcome (for + // example, a launch failure's error_message). transitionSessionState + // re-reads the current row, so a terminal race cannot be clobbered. + if _, _, err := e.transitionSessionState(ctx, session.TaskID, session.ID, models.TaskSessionStateCancelled, reason); err != nil { + e.logger.Error("failed to update agent session status", + zap.String("session_id", session.ID), + zap.Error(err)) + } + } + e.scheduleStop(ctx, session.ID, executionID, reason, force) + return nil +} + +// resolveExecutionIDForStop looks up the live execution for sessionID, +// applying the same not-found classification stopWithSession has always +// used. Shared by stopWithSession and the registry-recovered paths in +// StopByTaskID — it takes a bare session ID rather than a loaded row because +// the registry-only path may not have one (see stopRegistryRecoveredByID). +func (e *Executor) resolveExecutionIDForStop(ctx context.Context, sessionID string) (string, error) { + executionID, err := e.agentManager.GetExecutionIDForSession(ctx, sessionID) if err != nil || executionID == "" { if err != nil { if errors.Is(err, lifecycle.ErrNoExecutionForSession) { - return fmt.Errorf("%w: %w: %w", ErrExecutionNotFound, runtimeapi.ErrNotFound, err) + return "", fmt.Errorf("%w: %w: %w", ErrExecutionNotFound, runtimeapi.ErrNotFound, err) } - return fmt.Errorf("%w: lookup execution for session %q: %w", ErrExecutionNotFound, session.ID, err) + return "", fmt.Errorf("%w: lookup execution for session %q: %w", ErrExecutionNotFound, sessionID, err) } - return ErrExecutionNotFound + return "", ErrExecutionNotFound + } + return executionID, nil +} + +// registerStopOwner records advisory stop ownership ahead of a terminal +// session mutation this call is about to make. Callers with no interleaved +// mutation (stopRegistryRecoveredSession) may register and schedule together +// via registerAndScheduleStop instead. +func (e *Executor) registerStopOwner(sessionID, executionID string, force bool) { + if e.onExecutionStopOwnerRegistration != nil { + e.onExecutionStopOwnerRegistration(sessionID, executionID, force) + } +} + +// registryRecoveredSession keeps the database row paired with the exact +// execution captured from the registry snapshot. +type registryRecoveredSession struct { + session *models.TaskSession + reference lifecycle.ExecutionReference +} + +// stopRegistryRecoveredSession stops the execution captured for a registry +// only session. The per-session lock serializes the state and identity check +// with resume. If the session is still active and still owns the captured +// execution, use the normal cancellation transition. If a replacement owns +// the session, schedule only the captured execution and leave the replacement +// untouched. +func (e *Executor) stopRegistryRecoveredSession( + ctx context.Context, + recovered registryRecoveredSession, + reason string, + force bool, +) error { + session := recovered.session + reference := recovered.reference + if session == nil || session.ID == "" || reference.SessionID != session.ID || reference.ExecutionID == "" { + return fmt.Errorf("invalid registry recovery reference for task session") } - e.logStop(session, executionID, reason, force) - if e.onExecutionStopOwnerRegistration != nil { - e.onExecutionStopOwnerRegistration(session.ID, executionID, force) - } - // A session already in a terminal state carries its own outcome (e.g. a - // launch failure's error_message); a runtime that outlived it in the - // in-memory execution store must still be torn down, but the DB row is - // not touched. transitionSessionState re-reads the session's current - // state itself rather than trusting the caller-supplied snapshot, so a - // session that turned terminal between the caller's read and this call - // (e.g. StopByTaskID iterating a list read once) can't be clobbered. - if _, _, err := e.transitionSessionState(ctx, session.TaskID, session.ID, models.TaskSessionStateCancelled, reason); err != nil { - e.logger.Error("failed to update agent session status", + sessionLock := e.getSessionLock(reference.SessionID) + sessionLock.Lock() + defer sessionLock.Unlock() + + current := session + if refreshed, err := e.repo.GetTaskSession(ctx, session.ID); err == nil && refreshed != nil { + current = refreshed + } else if err != nil { + e.logger.Warn("failed to refresh registry-recovered session before stop; preserving captured execution", + zap.String("task_id", session.TaskID), zap.String("session_id", session.ID), zap.Error(err)) } - e.scheduleStop(ctx, session.ID, executionID, reason, force) + + return e.stopCapturedExecution(ctx, session.TaskID, current, reference, reason, force) +} + +// stopRegistryRecoveredByID stops the execution captured for a registry-only +// session whose row could not be loaded during recovery. It never resolves a +// replacement through the session index. A best-effort re-read can still +// cancel the session when the row becomes available and owns the captured +// execution; otherwise teardown is scheduled by the captured identity alone. +func (e *Executor) stopRegistryRecoveredByID( + ctx context.Context, + taskID string, + reference lifecycle.ExecutionReference, + reason string, + force bool, +) error { + if reference.SessionID == "" || reference.ExecutionID == "" { + return fmt.Errorf("invalid registry recovery reference for task %q", taskID) + } + + sessionLock := e.getSessionLock(reference.SessionID) + sessionLock.Lock() + defer sessionLock.Unlock() + + if session, err := e.repo.GetTaskSession(ctx, reference.SessionID); err == nil && session != nil { + return e.stopCapturedExecution(ctx, taskID, session, reference, reason, force) + } else if err != nil { + e.logger.Warn("failed to refresh registry-only session before stop; preserving captured execution", + zap.String("task_id", taskID), + zap.String("session_id", reference.SessionID), + zap.Error(err)) + } + + e.logger.Info("stopping registry-recovered execution with unloadable session row", + zap.String("task_id", taskID), + zap.String("session_id", reference.SessionID), + zap.String("agent_execution_id", reference.ExecutionID), + zap.String("reason", reason), + zap.Bool("force", force)) + e.registerStopOwner(reference.SessionID, reference.ExecutionID, force) + e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force) + return nil +} + +// stopCapturedExecution decides whether the captured execution still owns an +// active session. The current execution lookup is used only for this equality +// check; the stop target remains reference.ExecutionID in every branch. +func (e *Executor) stopCapturedExecution( + ctx context.Context, + taskID string, + session *models.TaskSession, + reference lifecycle.ExecutionReference, + reason string, + force bool, +) error { + if session == nil || session.ID != reference.SessionID { + return fmt.Errorf("registry recovery session does not match captured session %q", reference.SessionID) + } + if session.TaskID != taskID { + e.logger.Warn("registry recovery session task does not match captured task; preserving captured execution", + zap.String("task_id", taskID), + zap.String("session_id", reference.SessionID), + zap.String("session_task_id", session.TaskID), + zap.String("agent_execution_id", reference.ExecutionID)) + e.registerStopOwner(reference.SessionID, reference.ExecutionID, force) + e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force) + return nil + } + currentExecutionID, err := e.agentManager.GetExecutionIDForSession(ctx, reference.SessionID) + if err == nil && currentExecutionID == reference.ExecutionID && models.IsTaskLookupActiveSessionState(session.State) { + return e.stopWithSessionExecution(ctx, session, reference.ExecutionID, reason, force, true) + } + + e.logger.Info("stopping registry-recovered execution", + zap.String("task_id", session.TaskID), + zap.String("session_id", reference.SessionID), + zap.String("agent_execution_id", reference.ExecutionID), + zap.String("current_agent_execution_id", currentExecutionID), + zap.String("session_state", string(session.State)), + zap.String("reason", reason), + zap.Bool("force", force), + zap.Error(err)) + e.registerStopOwner(reference.SessionID, reference.ExecutionID, force) + e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force) return nil } @@ -253,7 +420,12 @@ func (e *Executor) StopExecution(ctx context.Context, executionID string, reason return nil } -// StopByTaskID stops all active executions for a task +// StopByTaskID stops all active executions for a task. It resolves what to +// stop from the union of the database's active-session query and the +// in-memory execution registry, so a session that is terminal in the +// database but still holds a registered execution (an orphan left by a +// failed teardown attempt) is reachable too — see +// docs/specs/tasks/system-design/task-stop-reachability.md. func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason string, force bool) error { // Get all active sessions for this task from database sessions, err := e.repo.ListActiveTaskSessionsByTaskID(ctx, taskID) @@ -264,7 +436,9 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin return ErrExecutionNotFound } - if len(sessions) == 0 { + recovered, unloadable, recoverErr := e.recoverRegistryOnlySessions(ctx, taskID, sessions) + + if len(sessions) == 0 && len(recovered) == 0 && len(unloadable) == 0 { return ErrExecutionNotFound } @@ -281,6 +455,42 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin stoppedCount++ } } + for _, recoveredSession := range recovered { + if err := e.stopRegistryRecoveredSession(ctx, recoveredSession, reason, force); err != nil { + e.logger.Warn("failed to stop registry-recovered session", + zap.String("task_id", taskID), + zap.String("session_id", recoveredSession.reference.SessionID), + zap.Error(err)) + lastErr = err + } else { + stoppedCount++ + } + } + for _, reference := range unloadable { + if err := e.stopRegistryRecoveredByID(ctx, taskID, reference, reason, force); err != nil { + e.logger.Warn("failed to stop registry-recovered execution with unloadable session row", + zap.String("task_id", taskID), + zap.String("session_id", reference.SessionID), + zap.Error(err)) + lastErr = err + } else { + stoppedCount++ + } + } + + // A registry-only orphan whose row failed to load is still targeted by its + // captured execution ID (stopRegistryRecoveredByID), so recoverErr no + // longer means that execution was unreachable — only that its terminal + // state could not be confirmed or corrected from here. Distinguish a + // caller that already saw at least one execution stop (soft: log and + // continue) from one where nothing could be reached at all (hard + // failure), same as before this recovery path existed. + if recoverErr != nil { + if stoppedCount > 0 { + return fmt.Errorf("%w: task %q: %w", ErrOrphanRecoveryIncomplete, taskID, recoverErr) + } + return fmt.Errorf("task %q has a registered execution but its session could not be loaded: %w", taskID, recoverErr) + } if stoppedCount == 0 && lastErr != nil { return lastErr @@ -289,6 +499,54 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin return nil } +// recoverRegistryOnlySessions loads the session rows for executions the +// registry still holds for taskID but that the active-session query did not +// return. Each loaded row stays paired with the exact execution reference +// captured from the registry. A row that cannot be loaded is still returned by +// that reference, rather than dropped, so teardown can be scheduled without a +// later session-index lookup. Only the last load error is kept; one surfaced +// failure is enough to stop the caller from reporting a false all-clear, and a +// caller that wants to retry re-derives the full set on the next call. +func (e *Executor) recoverRegistryOnlySessions( + ctx context.Context, taskID string, activeSessions []*models.TaskSession, +) (recovered []registryRecoveredSession, unloadable []lifecycle.ExecutionReference, loadErr error) { + registeredExecutions := e.agentManager.ListExecutionsForTask(taskID) + if len(registeredExecutions) == 0 { + return nil, nil, nil + } + + alreadyActive := make(map[string]struct{}, len(activeSessions)) + for _, session := range activeSessions { + alreadyActive[session.ID] = struct{}{} + } + + for _, reference := range registeredExecutions { + if reference.SessionID == "" || reference.ExecutionID == "" { + loadErr = fmt.Errorf("registry returned incomplete execution reference for task %q", taskID) + continue + } + if _, ok := alreadyActive[reference.SessionID]; ok { + continue + } + session, err := e.repo.GetTaskSession(ctx, reference.SessionID) + if err != nil || session == nil { + e.logger.Warn("failed to load registry-recovered session row for task stop; stopping by captured execution ID", + zap.String("task_id", taskID), + zap.String("session_id", reference.SessionID), + zap.Error(err)) + if err != nil { + loadErr = err + } + unloadable = append(unloadable, reference) + continue + } + recovered = append(recovered, registryRecoveredSession{ + session: session, reference: reference, + }) + } + return recovered, unloadable, loadErr +} + // stopReasonPassthrough is the StopReason returned by Executor.Prompt when a // passthrough session's prompt is dispatched to PTY stdin (no ACP turn to // observe). The submit sequence is resolved per-agent via ResolvePassthroughConfig diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go index 24118cfb82..88cf27e535 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go @@ -206,6 +206,243 @@ func TestStopSessionSynchronouslyStopsLateExecutionAfterSessionDeletion(t *testi } } +// TestStopByTaskID_StopsFailedSessionWithRegisteredExecution covers +// @covers AC-TASKS-TASK-STOP-REACHABILITY-001.1-.2: a task whose only session +// is FAILED in the database but still holds a registered execution (an +// orphan left by a failed teardown attempt) must be stopped, and the +// session's terminal state must survive — no CANCELLED transition. +func TestStopByTaskID_StopsFailedSessionWithRegisteredExecution(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateFailed, + ErrorMessage: "agent produced no output since start", + } + stopCalls := make(chan string, 1) + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-orphan"}} + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-orphan", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, force bool) error { + if !force { + t.Errorf("registry-recovered stop was not forced") + } + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil { + t.Fatalf("StopByTaskID: %v", err) + } + select { + case executionID := <-stopCalls: + if executionID != "execution-orphan" { + t.Fatalf("stopped execution = %q, want execution-orphan", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("registry-recovered execution was not stopped") + } + after, err := repo.GetTaskSession(context.Background(), "session-1") + if err != nil { + t.Fatalf("get session after stop: %v", err) + } + if after.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %q, want unchanged FAILED", after.State) + } + if after.ErrorMessage != "agent produced no output since start" { + t.Fatalf("session error message = %q, want preserved launch-failure message", after.ErrorMessage) + } +} + +// TestStopByTaskID_ActiveSessionKeepsCancellationTransition pins the +// unchanged path: a task with an active session behaves exactly as before — +// the session transitions to CANCELLED and its execution is stopped. +func TestStopByTaskID_ActiveSessionKeepsCancellationTransition(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateRunning, + } + repo.listActiveTaskSessionsByTaskIDFunc = func(_ context.Context, taskID string) ([]*models.TaskSession, error) { + if taskID != "task-1" { + return nil, nil + } + return []*models.TaskSession{repo.sessions["session-1"]}, nil + } + stopCalls := make(chan string, 1) + manager := &mockAgentManager{ + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-active", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil { + t.Fatalf("StopByTaskID: %v", err) + } + select { + case executionID := <-stopCalls: + if executionID != "execution-active" { + t.Fatalf("stopped execution = %q, want execution-active", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("active session's execution was not stopped") + } + after, err := repo.GetTaskSession(context.Background(), "session-1") + if err != nil { + t.Fatalf("get session after stop: %v", err) + } + if after.State != models.TaskSessionStateCancelled { + t.Fatalf("session state = %q, want CANCELLED", after.State) + } +} + +// TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound +// pins the unchanged path: a task with neither an active session nor a +// registered execution still returns ErrExecutionNotFound and changes +// nothing; a repeated stop after the executions are gone creates no +// duplicate transition. +func TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound(t *testing.T) { + repo := newMockRepository() + manager := &mockAgentManager{} + exec := newTestExecutor(t, manager, repo) + + err := exec.StopByTaskID(context.Background(), "task-empty", "cleanup", true) + if !errors.Is(err, ErrExecutionNotFound) { + t.Fatalf("error = %v, want ErrExecutionNotFound", err) + } + + // Repeated stop after the (nonexistent) executions are gone must behave + // identically — no duplicate transition to create. + err = exec.StopByTaskID(context.Background(), "task-empty", "cleanup", true) + if !errors.Is(err, ErrExecutionNotFound) { + t.Fatalf("repeated stop error = %v, want ErrExecutionNotFound", err) + } +} + +// TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound covers +// the registry-recovery path's own failure mode: a session is registered in +// the execution registry (so a real orphan exists) but its database row +// fails to load (e.g. a transient "database is locked"). The active-session +// query legitimately returns nothing, so if the load failure were silently +// swallowed the same way a missing row is, StopByTaskID would report +// ErrExecutionNotFound — indistinguishable from "there is nothing to stop" — +// which is exactly the reachability defect this recovery path exists to +// eliminate. It must surface the load failure instead. +func TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound(t *testing.T) { + repo := newMockRepository() + loadFailure := errors.New("database is locked") + repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) { + return nil, loadFailure + } + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-orphan"}} + }, + } + exec := newTestExecutor(t, manager, repo) + + err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true) + if err == nil { + t.Fatal("StopByTaskID: want an error, got nil") + } + if errors.Is(err, ErrExecutionNotFound) { + t.Fatalf("error = %v, want the load failure surfaced instead of ErrExecutionNotFound", err) + } + if !errors.Is(err, loadFailure) { + t.Fatalf("error = %v, want it to wrap %v", err, loadFailure) + } +} + +// TestStopByTaskID_MixedActiveAndUnloadableOrphanSurfacesLoadError covers +// Finding 2 from Review Round 1: a task with one active session (which stops +// cleanly) plus one registry-only orphan whose row fails to load must not +// return a clean nil. The retry contract on the never-started teardown's +// failed-attempt path depends on StopByTaskID surfacing that load failure so +// a later stop keeps retrying instead of reporting a false all-clear. +func TestStopByTaskID_MixedActiveAndUnloadableOrphanSurfacesLoadError(t *testing.T) { + repo := newMockRepository() + loadFailure := errors.New("database is locked") + activeSession := &models.TaskSession{ + ID: "session-active", TaskID: "task-1", State: models.TaskSessionStateRunning, + } + repo.sessions[activeSession.ID] = activeSession + repo.listActiveTaskSessionsByTaskIDFunc = func(_ context.Context, taskID string) ([]*models.TaskSession, error) { + if taskID != "task-1" { + return nil, nil + } + return []*models.TaskSession{activeSession}, nil + } + repo.getTaskSessionFunc = func(_ context.Context, id string) (*models.TaskSession, error) { + if id == "session-orphan" { + return nil, loadFailure + } + if s, ok := repo.sessions[id]; ok { + return s, nil + } + return nil, nil + } + stopCalls := make(chan string, 2) + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{ + {SessionID: "session-active", ExecutionID: "execution-active"}, + {SessionID: "session-orphan", ExecutionID: "execution-orphan"}, + } + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-active", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true) + if err == nil { + t.Fatal("StopByTaskID: want an error, got nil") + } + if !errors.Is(err, loadFailure) { + t.Fatalf("error = %v, want it to wrap %v", err, loadFailure) + } + + stopped := make(map[string]bool, 2) + for range 2 { + select { + case executionID := <-stopCalls: + stopped[executionID] = true + case <-time.After(2 * time.Second): + t.Fatal("expected both active and orphan executions to be stopped despite the orphan load failure") + } + } + if !stopped["execution-active"] { + t.Fatalf("stopped executions = %#v, want execution-active", stopped) + } + if !stopped["execution-orphan"] { + t.Fatalf("stopped executions = %#v, want execution-orphan", stopped) + } +} + func TestStopSessionDetailed_RejectsInvalidSession(t *testing.T) { exec := newTestExecutor(t, &mockAgentManager{}, newMockRepository()) @@ -440,6 +677,52 @@ func TestStopWithSession_ClaimSuppressesLateCancelledLaunchCleanup(t *testing.T) } } +// TestStopWithSession_RegistersStopOwnerBeforeCancelledWrite is a regression +// test for Finding 1 from Review Round 1: a concurrent StartAgentProcess +// failure claims forced cleanup for itself only when no stop owner is +// registered yet, and it classifies CANCELLED as terminal as soon as the row +// is visible. If the CANCELLED write happened before stop-owner registration, +// that window would let the concurrent failure win the claim and issue its +// own teardown alongside this one's. The owner must be registered first. +func TestStopWithSession_RegistersStopOwnerBeforeCancelledWrite(t *testing.T) { + repo := newMockRepository() + session := &models.TaskSession{ + ID: "session-order", TaskID: "task-order", State: models.TaskSessionStateRunning, + } + repo.sessions[session.ID] = session + + var order []string + repo.updateTaskSessionStateFunc = func(_ context.Context, sessionID string, state models.TaskSessionState, errorMessage string) error { + order = append(order, "cancelled-write") + if s, ok := repo.sessions[sessionID]; ok { + s.State = state + s.ErrorMessage = errorMessage + } + return nil + } + + manager := &mockAgentManager{ + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-order", nil + }, + stopAgentWithReasonFunc: func(context.Context, string, string, bool) error { + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + exec.SetOnExecutionStopOwnerRegistration(func(string, string, bool) { + order = append(order, "register-owner") + }) + + if err := exec.stopWithSession(context.Background(), session, "test stop", false); err != nil { + t.Fatalf("stopWithSession: %v", err) + } + + if len(order) != 2 || order[0] != "register-owner" || order[1] != "cancelled-write" { + t.Fatalf("operation order = %v, want [register-owner cancelled-write]", order) + } +} + func TestStopExecution_PreservesRuntimeFailureClassification(t *testing.T) { runtimeFailure := errors.New("runtime teardown failed") tests := []struct { diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index bf983a420a..805a7ccb4e 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/agent/agents" agentdto "github.com/kandev/kandev/internal/agent/dto" "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/task/models" @@ -29,6 +30,7 @@ type mockAgentManager struct { setExecutionDescriptionFunc func(ctx context.Context, agentExecutionID string, description string) error setExecutionEnvFunc func(ctx context.Context, agentExecutionID string, env map[string]string) error getExecutionIDForSessionFunc func(ctx context.Context, sessionID string) (string, error) + listExecutionsForTaskFunc func(taskID string) []lifecycle.ExecutionReference isAgentCommandConfiguredFunc func(agentExecutionID string) bool isAgentRunningForSessionFunc func(ctx context.Context, sessionID string) bool cleanupStaleExecutionFunc func(ctx context.Context, sessionID string) error @@ -227,6 +229,12 @@ func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, session } return "", fmt.Errorf("no execution found for session %s", sessionID) } +func (m *mockAgentManager) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference { + if m.listExecutionsForTaskFunc != nil { + return m.listExecutionsForTaskFunc(taskID) + } + return nil +} func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) { if m.resolveAgentProfileFunc != nil { diff --git a/apps/backend/internal/orchestrator/executor/executor_stop_recovery_identity_test.go b/apps/backend/internal/orchestrator/executor/executor_stop_recovery_identity_test.go new file mode 100644 index 0000000000..846a1de539 --- /dev/null +++ b/apps/backend/internal/orchestrator/executor/executor_stop_recovery_identity_test.go @@ -0,0 +1,166 @@ +package executor + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/task/models" +) + +// TestStopByTaskID_DoesNotStopReplacementExecution reproduces the registry +// recovery race: the execution seen during recovery is replaced before the +// stop path resolves the session again. Recovery must keep the original +// execution identity instead of stopping the replacement. +func TestStopByTaskID_DoesNotStopReplacementExecution(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateFailed, + } + + stopCalls := make(chan string, 1) + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}} + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + // Model the old execution being removed and a replacement being + // registered before the recovery path performs its fresh lookup. + return "execution-replacement", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil { + t.Fatalf("StopByTaskID: %v", err) + } + + select { + case executionID := <-stopCalls: + if executionID != "execution-original" { + t.Fatalf("stopped execution = %q, want execution-original", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("registry-recovered execution was not stopped") + } +} + +// TestStopByTaskID_UnloadableRecoveryDoesNotStopReplacementExecution covers +// the same identity race when the session row cannot be read. The captured +// execution must remain reachable even when the session index now points at a +// replacement. +func TestStopByTaskID_UnloadableRecoveryDoesNotStopReplacementExecution(t *testing.T) { + loadFailure := errors.New("database is locked") + repo := newMockRepository() + repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) { + return nil, loadFailure + } + + stopCalls := make(chan string, 1) + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}} + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-replacement", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true) + if !errors.Is(err, loadFailure) { + t.Fatalf("StopByTaskID error = %v, want it to wrap %v", err, loadFailure) + } + + select { + case executionID := <-stopCalls: + if executionID != "execution-original" { + t.Fatalf("stopped execution = %q, want execution-original", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("registry-recovered execution was not stopped") + } +} + +// TestStopByTaskID_CancelsSessionThatBecomesActiveDuringRecovery models a +// session that changes from terminal to active after the initial recovery +// read. When the captured execution still owns it, recovery must use the +// normal CANCELLED transition before scheduling teardown. +func TestStopByTaskID_CancelsSessionThatBecomesActiveDuringRecovery(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateFailed, + } + var reads atomic.Int32 + repo.getTaskSessionFunc = func(_ context.Context, sessionID string) (*models.TaskSession, error) { + read := reads.Add(1) + switch read { + case 1: + return &models.TaskSession{ + ID: sessionID, TaskID: "task-1", State: models.TaskSessionStateFailed, + }, nil + case 2, 3: + return &models.TaskSession{ + ID: sessionID, TaskID: "task-1", State: models.TaskSessionStateRunning, + }, nil + default: + repo.mu.Lock() + defer repo.mu.Unlock() + return cloneMockTaskSession(repo.sessions[sessionID]), nil + } + } + + stopCalls := make(chan string, 1) + manager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}} + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-original", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + exec := newTestExecutor(t, manager, repo) + + if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil { + t.Fatalf("StopByTaskID: %v", err) + } + select { + case executionID := <-stopCalls: + if executionID != "execution-original" { + t.Fatalf("stopped execution = %q, want execution-original", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("registry-recovered execution was not stopped") + } + + repo.mu.Lock() + state := repo.sessions["session-1"].State + repo.mu.Unlock() + if state != models.TaskSessionStateCancelled { + t.Fatalf("session state = %q, want CANCELLED", state) + } +} diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 6bc19d853d..12ad29a213 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -12,6 +12,7 @@ import ( "github.com/jmoiron/sqlx" "github.com/kandev/kandev/internal/agent/agents" "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" @@ -153,6 +154,9 @@ func (m *mockAgentManager) EnsureWorkspaceExecutionForSession(ctx context.Contex func (m *mockAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) { return "", fmt.Errorf("no execution found") } +func (m *mockAgentManager) ListExecutionsForTask(_ string) []lifecycle.ExecutionReference { + return nil +} func (m *mockAgentManager) CancelAgent(ctx context.Context, sessionID string) error { return nil diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index a849581e81..e0ac65037e 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -899,7 +899,7 @@ func (s *Service) handleSessionLaunchFailure( ) error { failureCtx := context.WithoutCancel(ctx) safeErr := routingerr.SanitizeError(launchErr) - s.recordSessionLaunchFailure( + _ = s.recordSessionLaunchFailure( failureCtx, taskID, sessionID, safeErr, preloadedSession..., ) return safeErr @@ -908,13 +908,17 @@ func (s *Service) handleSessionLaunchFailure( // recordSessionLaunchFailure transitions a still-active session to FAILED and // updates the task only while the same failed session still owns it. Typed // launch-error persistence belongs to the executor's session transition. -func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error, preloadedSession ...*models.TaskSession) { +// Returns whether the session itself was durably transitioned to FAILED — a +// caller that is about to force-stop the process on the strength of that +// transition (stopNeverStartedExecution) must not do so when this reports +// false, or it kills the process while the session row still claims RUNNING. +func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error, preloadedSession ...*models.TaskSession) bool { _, changed := s.updateTaskSessionStateWithHook( ctx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error(), false, nil, preloadedSession..., ) if !changed { - return + return false } updated, err := s.updateTaskStateForEarlyLaunchFailure(ctx, taskID, sessionID) if err != nil { @@ -922,12 +926,13 @@ func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessio zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Error(err)) - return + return true } if !updated { - return + return true } s.processParentChildrenCompletedForTaskState(ctx, taskID, v1.TaskStateFailed) + return true } func (s *Service) updateTaskStateForEarlyLaunchFailure( @@ -3623,7 +3628,15 @@ func (s *Service) StopTask(ctx context.Context, taskID string, reason string, fo // Stop all agents for this task if err := s.executor.StopByTaskID(ctx, taskID, reason, force); err != nil { - return err + if !errors.Is(err, executor.ErrOrphanRecoveryIncomplete) { + return err + } + // Every session that could be reached did stop; only a registry-only + // orphan's row failed to load. Log it and still transition to REVIEW + // rather than surfacing a false "failed to stop task" to the user. + s.logger.Warn("task stop reached REVIEW with an orphan recovery load failure", + zap.String("task_id", taskID), + zap.Error(err)) } // Move task to REVIEW state for user review @@ -3796,7 +3809,17 @@ func (s *Service) CancelTaskExecution(ctx context.Context, taskID string, reason zap.String("task_id", taskID), zap.String("reason", reason), zap.Bool("force", force)) - return s.executor.StopByTaskID(ctx, taskID, reason, force) + err := s.executor.StopByTaskID(ctx, taskID, reason, force) + if err != nil && errors.Is(err, executor.ErrOrphanRecoveryIncomplete) { + // Every session that could be reached did stop; only a registry-only + // orphan's row failed to load. Log it rather than reporting a false + // cancellation failure to office tree controls. + s.logger.Warn("task execution cancelled with an orphan recovery load failure", + zap.String("task_id", taskID), + zap.Error(err)) + return nil + } + return err } // CancelTaskExecutionSynchronously stops every active session for a task and diff --git a/apps/backend/internal/orchestrator/task_operations_stop_orphan_recovery_test.go b/apps/backend/internal/orchestrator/task_operations_stop_orphan_recovery_test.go new file mode 100644 index 0000000000..6c7538a2d5 --- /dev/null +++ b/apps/backend/internal/orchestrator/task_operations_stop_orphan_recovery_test.go @@ -0,0 +1,221 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + orchestratorexec "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// drainStopCalls waits for exactly n buffered StopAgentWithReason calls, +// discarding their execution IDs. StopByTaskID schedules each stop on a +// detached goroutine, so a test that does not wait for every call it expects +// leaves that goroutine still in flight when the test returns — caught by +// this package's goleak.VerifyTestMain as a leak even though the send itself +// would have succeeded. +func drainStopCalls(t *testing.T, ch <-chan string, n int) { + t.Helper() + for i := 0; i < n; i++ { + select { + case <-ch: + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for stop call %d/%d", i+1, n) + } + } +} + +// TestStopTask_OrphanRecoveryLoadFailureStillReachesReview covers Review +// round 2 Finding 1: a task with one active session that stops cleanly plus +// one registry-only orphan whose row fails to load must still transition the +// task to REVIEW instead of surfacing a false "failed to stop task" to the +// caller, while the load failure stays observable (StopByTaskID's wrapped +// error is still returned to whoever asks for it directly, even though +// StopTask itself does not propagate it as a hard failure). It also covers +// the follow-up fix that the orphan's captured execution is itself targeted +// for stop, not just skipped because its row could not load. +func TestStopTask_OrphanRecoveryLoadFailureStillReachesReview(t *testing.T) { + ctx := context.Background() + baseRepo := setupTestRepo(t) + seedTaskAndSession(t, baseRepo, "task-1", "session-active", models.TaskSessionStateRunning) + + loadFailure := errors.New("database is locked") + hookedRepo := &coordinatorStopRepoHooks{ + repoStore: baseRepo, + getSessionFunc: func(readCtx context.Context, sessionID string) (*models.TaskSession, error) { + if sessionID == "session-orphan" { + return nil, loadFailure + } + return baseRepo.GetTaskSession(readCtx, sessionID) + }, + } + + stopCalls := make(chan string, 8) + agentManager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{ + {SessionID: "session-active", ExecutionID: "execution-active"}, + {SessionID: "session-orphan", ExecutionID: "execution-orphan"}, + } + }, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-active", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager) + + if err := svc.StopTask(ctx, "task-1", "cleanup", true); err != nil { + t.Fatalf("StopTask returned an error despite the active session stopping cleanly: %v", err) + } + + // The active session stops normally, and the registry-only orphan is now + // also targeted by its captured execution ID despite its row failing to + // load. + drainStopCalls(t, stopCalls, 2) + + taskRepo.mu.Lock() + taskState := taskRepo.updatedStates["task-1"] + taskRepo.mu.Unlock() + if taskState != v1.TaskStateReview { + t.Fatalf("task state = %q, want REVIEW despite the orphan load failure", taskState) + } + + // The load failure itself must still be observable to a direct caller of + // the executor, so a retry loop built on StopByTaskID does not see a false + // all-clear. + directErr := svc.executor.StopByTaskID(ctx, "task-1", "cleanup", true) + if !errors.Is(directErr, loadFailure) { + t.Fatalf("StopByTaskID error = %v, want it to still wrap the load failure", directErr) + } + if !errors.Is(directErr, orchestratorexec.ErrOrphanRecoveryIncomplete) { + t.Fatalf("StopByTaskID error = %v, want it to wrap ErrOrphanRecoveryIncomplete", directErr) + } + // The retry resolves session-active (now CANCELLED, terminal) through + // registry recovery instead of the active-state query, plus the + // still-unloadable orphan; both are targeted again. + drainStopCalls(t, stopCalls, 2) +} + +// TestStopTask_UnloadableOrphanReachesReviewAfterScheduling covers a task +// with no active rows and one unloadable registry orphan. The exact registry +// reference is enough to schedule teardown, so StopTask reaches REVIEW while +// still retaining the load failure for a direct StopByTaskID caller. +func TestStopTask_UnloadableOrphanReachesReviewAfterScheduling(t *testing.T) { + ctx := context.Background() + baseRepo := setupTestRepo(t) + + loadFailure := errors.New("database is locked") + hookedRepo := &coordinatorStopRepoHooks{ + repoStore: baseRepo, + getSessionFunc: func(context.Context, string) (*models.TaskSession, error) { + return nil, loadFailure + }, + } + stopCalls := make(chan string, 1) + agentManager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-orphan", ExecutionID: "execution-orphan"}} + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager) + + if err := svc.StopTask(ctx, "task-1", "cleanup", true); err != nil { + t.Fatalf("StopTask: %v", err) + } + select { + case executionID := <-stopCalls: + if executionID != "execution-orphan" { + t.Fatalf("stopped execution = %q, want execution-orphan", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("unloadable orphan execution was not scheduled for stop") + } + + taskRepo.mu.Lock() + state := taskRepo.updatedStates["task-1"] + taskRepo.mu.Unlock() + if state != v1.TaskStateReview { + t.Fatalf("task state = %q, want REVIEW", state) + } +} + +// TestStopByTaskID_StopsOrphanExecutionWhenSessionRowFailsToLoad covers the +// review finding that a registry-only orphan whose session row cannot be +// loaded must still have its captured execution targeted for stop. A load +// failure on the row must not make the process itself unreachable. Before this +// fix, StopByTaskID skipped the orphan entirely on a load failure and never +// called StopAgentWithReason for it. +func TestStopByTaskID_StopsOrphanExecutionWhenSessionRowFailsToLoad(t *testing.T) { + ctx := context.Background() + baseRepo := setupTestRepo(t) + + loadFailure := errors.New("database is locked") + hookedRepo := &coordinatorStopRepoHooks{ + repoStore: baseRepo, + getSessionFunc: func(context.Context, string) (*models.TaskSession, error) { + return nil, loadFailure + }, + } + + stopCalls := make(chan string, 1) + agentManager := &mockAgentManager{ + listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference { + if taskID != "task-1" { + return nil + } + return []lifecycle.ExecutionReference{{SessionID: "session-orphan", ExecutionID: "execution-orphan"}} + }, + getExecutionIDForSessionFunc: func(_ context.Context, sessionID string) (string, error) { + if sessionID != "session-orphan" { + return "", errors.New("unexpected session ID") + } + return "execution-orphan", nil + }, + stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error { + stopCalls <- executionID + return nil + }, + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress) + svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager) + + err := svc.executor.StopByTaskID(ctx, "task-1", "cleanup", true) + if !errors.Is(err, loadFailure) { + t.Fatalf("StopByTaskID error = %v, want it to wrap the load failure", err) + } + if !errors.Is(err, orchestratorexec.ErrOrphanRecoveryIncomplete) { + t.Fatalf("StopByTaskID error = %v, want it to wrap ErrOrphanRecoveryIncomplete since the execution itself was reached", err) + } + + select { + case executionID := <-stopCalls: + if executionID != "execution-orphan" { + t.Fatalf("stopped execution = %q, want execution-orphan", executionID) + } + case <-time.After(2 * time.Second): + t.Fatal("the registry-only orphan's captured execution was never stopped") + } +} diff --git a/docs/decisions/2026-09-02-terminal-stall-owns-process-teardown.md b/docs/decisions/2026-09-02-terminal-stall-owns-process-teardown.md new file mode 100644 index 0000000000..48d626c464 --- /dev/null +++ b/docs/decisions/2026-09-02-terminal-stall-owns-process-teardown.md @@ -0,0 +1,86 @@ +# ADR-2026-09-02-terminal-stall-owns-process-teardown: A Terminal Stall Owns Process Teardown, and the Execution Registry Is the Stop Authority + +**Status:** accepted +**Date:** 2026-09-02 +**Area:** backend + +## Context + +[ADR-2026-08-18-never-started-agent-stall-terminal](2026-08-18-never-started-agent-stall-terminal.md) +made a zero-event prompt terminal: the session and task move to `FAILED`. It +decided the classification but not the disposal. The implementation records the +failure and returns, so the agent process keeps running. + +That leaves a live process attached to a session the database calls terminal. +The task-scoped stop resolves the sessions it should halt from a query whose +active-state set excludes terminal states, so it cannot see that session and +reports that no execution exists. The session-scoped stop, which resolves the +execution directly from the in-memory registry, still works. A live execution is +therefore unreachable through the operation a user actually reaches for. + +Separately, the inactivity clock behind the classification advanced on every +inbound agent frame, including metadata frames such as usage and context-window +updates that a booting or hung adapter emits without doing any work. An adapter +in that state postponed its own detection indefinitely. + +## Decision + +Three rules, which are one rule seen from three sides: a record of terminal work +must eventually correspond to no running process. Rule 2's teardown is +detached and force-bounded at 30 seconds (`neverStartedStopTimeout`), so a +window exists between the `FAILED` write and the process actually exiting; a +retried or later stop (Rule 3) is what closes that window if the bounded +attempt itself fails. + +1. **The inactivity clock measures progress, not traffic.** Only prompt + dispatch, a turn event, the prompt's terminal completion, or new user input + advances it. A metadata frame does not. The same clock serves the + completion-signal watchdog, which inherits the corrected semantics. +2. **A terminal stall classification owns teardown.** The never-started path + records the launch failure first, then stops the execution with force + through the execution-scoped stop, which does not write session state. A + failed teardown is logged and leaves the execution registered for a later + attempt; it never downgrades the recorded `FAILED` state. +3. **The execution registry is the authority for a live execution that + persisted session state alone cannot vouch for.** Active-state sessions + remain a valid stop source in their own right; the registry is what the + task-scoped stop also consults so a session the database calls terminal is + not treated as unstoppable. It resolves the union of both. A session + recovered only through the registry is stopped without a state transition, so + its terminal state and error message survive. + +## Consequences + +A never-started stall disposes of its own process, so the orphan this ADR +describes is not created in the first place. Rule 3 is still required: it is the +recovery for an orphan created by any other path, and it is what makes the stop +operation honest rather than dependent on the session looking healthy. + +The advisory notice now appears for a turn that emits only metadata frames. That +notice is non-destructive, and the situation it describes, waiting with no +visible progress, is exactly what it is for. + +The task-scoped stop gains a read of the in-memory registry, so the task system +depends on one more read-only agent-manager capability. Stopping a task whose +only live work was an orphan now succeeds, and the caller's existing post-stop +task handling runs. + +The correction does not make an unbounded ACP prompt call bounded. It bounds the +consequences: silence is now reported on time, disposed of, and recoverable. + +## Alternatives Considered + +- **Widen the active-state SQL set to include `FAILED` for the stop path.** + Rejected. It answers "which sessions look stoppable" with a state list that + would need to grow for every terminal state, and it still consults the wrong + authority: the process is registered in memory, not in the state column. +- **Keep a second clock only for the never-started case.** Rejected. Two clocks + over the same events invite the two answers to diverge, and the general clock + would stay dishonest for its other consumer. +- **Tear down first, then record the failure.** Rejected. Teardown can publish + lifecycle activity, and a crash or error between the two steps would leave a + stopped process with no recorded explanation. Recording first keeps the + durable outcome authoritative. +- **Have the stall handler call the task-scoped stop.** Rejected. That path + writes a cancellation state and would overwrite the launch-failure state and + message the requirement asks users to see. diff --git a/docs/plans/never-started-stall-teardown/plan.md b/docs/plans/never-started-stall-teardown/plan.md new file mode 100644 index 0000000000..517932c245 --- /dev/null +++ b/docs/plans/never-started-stall-teardown/plan.md @@ -0,0 +1,254 @@ +--- +requirements: + - REQ-AGENTS-AGENT-STALL-RECOVERY-001 + - REQ-TASKS-TASK-STOP-REACHABILITY-001 +system_design: + - ../../specs/agents/system-design/agent-stall-recovery.md + - ../../specs/tasks/system-design/task-stop-reachability.md +legacy_specs: [] +created: 2026-09-02 +status: implemented +--- + +# Implementation Plan: Never-Started Stall Teardown + +This plan implements the [terminal stall teardown decision](../../decisions/2026-09-02-terminal-stall-owns-process-teardown.md). + +## Overview + +Three linked backend corrections that restore one invariant: a record of +terminal work must correspond to no running process. + +1. The inactivity clock stops accepting metadata frames as progress, so a + never-started prompt trips the watchdog on time. +2. The never-started classification tears down its own execution instead of + only recording `FAILED`. +3. The task-scoped stop resolves what to halt from the execution registry as + well as the active-session query, so a terminal session can no longer hide a + live process. + +Backend only. No schema migration, no HTTP or WebSocket contract change, no +frontend change. + +## Confirmed root cause + +**Defect 1 — the clock can be reset by metadata alone.** +`waitForPromptDone` (`apps/backend/internal/agent/runtime/lifecycle/session.go`) +fires when `time.Since(lastActivityAt) >= 5m`. `recordActivity` +(`apps/backend/internal/agent/runtime/lifecycle/manager_events.go:551-559`) +assigns `lastActivityAt = time.Now()` for **every** inbound event, while it sets +`agentEventSincePrompt` only for `turnContentEventTypes`. A session emitting +only `usage_update`, `context_window`, `available_commands_update`, or +`session_info_update` therefore restarts the clock forever while +`NeverStarted` stays true, and the watchdog never fires. Observed: execution +`12fed303` on task `54dd18cf` produced nothing for 5m23s with no stall log. + +**Defect 2 — the watchdog marks `FAILED` but never stops the process.** +`handleAgentStalled` +(`apps/backend/internal/orchestrator/event_handlers_stall.go:37-122`) posts the +notice and calls `recordSessionLaunchFailure`, which moves the session and task +to `FAILED`. It never calls a stop. The agent process survives its own +tombstone. + +**Defect 3 — `FAILED` makes the task-scoped stop structurally blind.** +`Executor.StopByTaskID` +(`apps/backend/internal/orchestrator/executor/executor_interaction.go:245-277`) +resolves sessions with `ListActiveTaskSessionsByTaskID`, whose SQL matches only +`state IN ('CREATED','STARTING','RUNNING','WAITING_FOR_INPUT')` +(`apps/backend/internal/task/repository/sqlite/session.go:2495-2500`). `FAILED` +is outside that set, so the stop returns `ErrExecutionNotFound` while the +execution is still registered in `ExecutionStore` and the process is still +running. + +Reproduced live on task `bca4780c` (2026-09-02): + +```text +12:11:29 agent stall detected ... execution_id=04c6fac6 ... "never_started": true + -> session 038c8173 state = FAILED (process NOT stopped) +12:12:38 stopping task execution ... reason "diagnosing MCP startup hang" +12:12:38 ERROR failed to stop task ... error: "execution not found" +12:12:57 stopping execution ... agent_execution_id=04c6fac6 <- session-scoped stop worked +``` + +The session-scoped stop succeeding 19 seconds later proves the execution was +alive and registered throughout. + +## Requirement and design references + +| Work order | Acceptance criteria | System design | +| --- | --- | --- | +| Task 01 | `AC-AGENTS-AGENT-STALL-RECOVERY-001.9`, `.2` | [Agent stall recovery](../../specs/agents/system-design/agent-stall-recovery.md) | +| Task 02 | `AC-AGENTS-AGENT-STALL-RECOVERY-001.10`, `.11`, `.7` | [Agent stall recovery](../../specs/agents/system-design/agent-stall-recovery.md) | +| Task 03 | `AC-TASKS-TASK-STOP-REACHABILITY-001.1` – `.7` | [Task stop reachability](../../specs/tasks/system-design/task-stop-reachability.md) | + +## Work orders + +Sequential. No task is parallel-safe: Task 02 depends on Task 01's clock being +honest before teardown can be exercised on schedule, and Task 03's regression +constructs the orphan that Task 02 stops producing. + +- [x] [Task 01: Measure prompt progress, not traffic](task-01-honest-inactivity-clock.md) +- [x] [Task 02: Tear down a never-started execution](task-02-terminal-stall-teardown.md) +- [x] [Task 03: Reach a registered execution from a task stop](task-03-stop-reaches-registered-execution.md) + +## Verification strategy + +Each work order owns a regression that fails first for the stated reason. The +lifecycle timing regressions use `testing/synctest` with the real one-minute +ticker and the real five-minute threshold; production durations are not +shortened to make a test pass. + +No end-to-end browser test is added. All three corrections are backend +lifecycle and stop-path behavior with no new user-facing surface: the notice, +its copy, and its rendering are unchanged and already covered by +`apps/web/e2e/tests/session/pause-resume-recovery.spec.ts`. + +After all three work orders pass their own checks: + +1. `cd apps/backend && gofmt -l $(git diff --name-only origin/main -- '*.go')` +2. `make -C apps/backend test` +3. `make -C apps/backend lint` + +### Final verification results + +This branch's HEAD (`c51ec0a21`) is far behind `origin/main`, so step 1's +literal `git diff --name-only origin/main` pulls in thousands of unrelated +files that don't exist in this old checkout's layout. Ran the equivalent, +correctly scoped checks instead: + +1. `gofmt -l` on every file this plan touched — clean. +2. `go test -tags fts5 ./...` (`make -C apps/backend test`) — ran to + completion after two host disk-space exhaustions during this session were + cleared with `go clean -cache` (the repo's Go build/link artifacts, fully + regenerable; unrelated to this change). Full-suite run: every package not + already known-flaky on this host passes. The only failures are the same + set already documented and independently confirmed pre-existing (via + scoped `git stash` round-trips back to unmodified HEAD) in the three work + orders' Results sections: + - 7 tests in `internal/agent/runtime/lifecycle` (6 `TestWorktreePreparer_*` + git/worktree-state tests plus + `TestBuildAuthMethodsIdentityAgentOverridesEnvironment`, a Unix-socket + path-length limit). + - A larger, separately-confirmed pre-existing cluster keyed on the same + `unsafe worktree path: not a directory` fixture assumption, spanning + `internal/worktree`, `internal/task/service`, + `internal/system/storage/workspaces`, `internal/launcher`, + `internal/agentctl/server/*`, `internal/common/config`, and + `internal/agent/managedruntime` — none of these packages, or anything + they depend on, is touched by this plan's changes (`lifecycle`, + `orchestrator`, `orchestrator/executor`, `backendapp/adapters.go`, + `integration` test doubles only). Reproduced identically against + unmodified HEAD via scoped stash to confirm. + - `internal/orchestrator`, `internal/orchestrator/executor`, and every + other package this plan touches: all green. +3. `golangci-lint run ./... --new-from-rev=` (the + correct base, found via `git merge-base HEAD origin/main`, since comparing + against `origin/main` directly has the same stale-branch problem as step + 1) — 0 issues. +4. `go build ./...` — clean. + +All three work orders are code-complete, tested, and documented. + +### Build-step self-review addendum + +Walked the recovery path added by Task 03 for the "guard reproduces the +failure it exists to prevent" and "value re-read after the decision" patterns +(see `task-03-...md` for the full account). Found and fixed one real bug: +`recoverRegistryOnlySessions` silently dropped a registered session on a +transient `GetTaskSession` load failure, which could make `StopByTaskID` +report the same `ErrExecutionNotFound` as "nothing registered" — a false +all-clear over a real orphan. Fixed with a new regression, +`TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound` +(red-then-green). + +Also investigated, on request, whether defect 1 pre-fix leaves a stalled +session at RUNNING (not FAILED) and whether `StopByTaskID` is blind to that +case too. Conclusion, detailed in Task 03's Results: yes it leaves it at +RUNNING, but `StopByTaskID` was never blind there — RUNNING has always been +in `ListActiveTaskSessionsByTaskID`'s active-state set, and Task 01 + Task 02 +together remove the failure mode at its source. No fourth defect shape found; +nothing added beyond the hardening fix above. + +### Definition of Done (Build step gauntlet) + +- `make fmt` — clean, nothing reformatted. +- `make typecheck` — clean, after regenerating + `apps/web/lib/generated/{changelog,release-notes}.json` (a pre-existing + worktree gap: `make typecheck` invokes `tsc` via `pnpm exec`, which does not + run the `pretypecheck` npm hook that normally generates these files — same + class of issue as the documented `apps/node_modules` worktree gotcha, not + caused by this change). +- `make lint` (backend + web + harness + specs + architecture) — 0 issues, + including an unscoped `golangci-lint run ./...` across the whole backend. +- `make lint-format` — clean. +- `cd apps/web && pnpm run i18n:ratchet` — clean, no-op (no `apps/web` files + touched by this change). +- `make test-cli test-scripts` — `test-cli` green. `test-scripts` has one + failure, `Make variable home` in `scripts/dev-prod-db-path.test.sh`, + unrelated to anything this plan touches (Makefile `KANDEV_HOME_DIR` path + resolution). Proven pre-existing per the workflow's own rule: reproduced + identically in a separate scratch worktree checked out at the merge base + (`git merge-base HEAD origin/main` = `c51ec0a21`), not via stash. +- `make -C apps/backend test` (`test-backend`) — full suite green except the + documented pre-existing/environment failures (disk pressure on this shared + host); see the Final verification results section above and each work + order's Results. +- `make test-web` — not run to completion. This host was running vitest for + several other concurrent sessions' worktrees at the same time (`ps aux` + showed 4+ other `vitest run` processes from unrelated task directories), + disk had dropped to ~10Gi free, and the run produced no output after 37+ + minutes of CPU time. `apps/web` is untouched by this diff and both + `make typecheck` and web `make lint` (eslint) are clean, which is the + relevant signal for a change that touches zero frontend files. Stopped the + run rather than let it continue starving a shared, already disk-constrained + host for a suite this diff cannot affect. + +## Reviewed but excluded: the unbounded `conn.Prompt()` call + +`apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go` +starts `conn.Prompt(traceCtx, ...)` in a goroutine with no deadline. That is +why a hung adapter is unbounded, and the task asks whether the fix belongs here. + +It does not, for three reasons. + +- **It is a different decision, with a different owner.** A prompt deadline is a + policy about how long legitimate work may run. That is exactly the ambiguity + [ADR-2026-07-29](../../decisions/2026-07-29-agent-stall-user-controlled-recovery.md) + weighed when it rejected automatic timeouts, and reversing it needs its own + requirement covering the value, what the user sees when it fires, and how it + composes with cancel escalation. Deciding that inside a defect fix would + smuggle a product change into a repair. +- **It would not fix any of the three defects.** A deadline on the ACP call + cannot make a metadata-only stream trip the watchdog, cannot stop the process + that a terminal classification left running, and cannot make the stop path see + a registered execution. The defects are in the observation, the disposal, and + the resolution; the deadline is upstream of all three. +- **Its urgency is what this work removes.** The reason an unbounded call is + currently dangerous is that nothing downstream reports or disposes of it. + After these three work orders, a hung adapter is detected on schedule, its + process is torn down, and any remaining orphan is stoppable. A deadline + becomes a defence in depth rather than the only backstop, and can be evaluated + on its own merits. + +Recommended as a separate task, not carried here. + +## Risks and non-goals + +- **The advisory notice becomes more frequent.** A turn that emits only metadata + frames after real activity will now reach the advisory threshold. This is + intended and specified by `AC-AGENTS-AGENT-STALL-RECOVERY-001.9`; the notice + is non-destructive and offers only a cancel action. +- **The completion-signal watchdog shares the corrected clock.** It reads the + same activity snapshot, so it also stops being fooled by metadata frames. This + is a deliberate consequence, covered by a Task 01 regression, not a silent + side effect. +- **Stopping an orphan reclaims the card.** A task whose only live work was an + orphan now stops successfully, so the caller's existing post-stop task + handling runs. Task 03's regression pins this so it is a reviewed outcome. +- **A failed teardown still leaves an orphan.** It is logged and the execution + stays registered for a later stop; Task 03 is what makes that retry reachable. +- **MCP startup timeouts are out of scope.** They are what starts the stall; + this work is about the stall being unreported and unrecoverable. +- No test is weakened or deleted. The four existing stall regressions in + `session_test.go` and the six in `event_handlers_stall_test.go` must keep + passing unchanged. diff --git a/docs/plans/never-started-stall-teardown/task-01-honest-inactivity-clock.md b/docs/plans/never-started-stall-teardown/task-01-honest-inactivity-clock.md new file mode 100644 index 0000000000..4340e3f168 --- /dev/null +++ b/docs/plans/never-started-stall-teardown/task-01-honest-inactivity-clock.md @@ -0,0 +1,134 @@ +--- +id: "01-honest-inactivity-clock" +title: "Measure prompt progress, not traffic" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +system_design: + - ../../specs/agents/system-design/agent-stall-recovery.md +requirements: + - REQ-AGENTS-AGENT-STALL-RECOVERY-001 +acceptance_criteria: + - AC-AGENTS-AGENT-STALL-RECOVERY-001.9 + - AC-AGENTS-AGENT-STALL-RECOVERY-001.2 +--- + +# Task 01: Measure prompt progress, not traffic + +## Outcome + +The stall clock advances only on prompt dispatch, a turn event, the prompt's +terminal completion, or new user input. A metadata frame no longer restarts it, +so a never-started prompt trips the watchdog five minutes after dispatch +regardless of how much metadata the adapter emits. + +## In scope + +- Stop `recordActivity` from advancing the activity timestamp for events outside + `turnContentEventTypes`. The existing `agentEventSincePrompt` and + `promptActivityEpoch` gating is already correct and is not changed. +- Keep every other effect of `recordActivity` unchanged: `firstActivityOnce` + still fires on the first event of any type, and the wakeup-driven + `Ready → Running` flip keeps its current turn-content gate. +- Update the `lastActivityAt` doc comment in `types.go` to state what advances + it, so the field's contract is readable at its declaration. + +## Exclusions + +- No change to the five-minute threshold, the one-minute tick, the payload + shape, or the once-per-generation rule. +- No change to `armPromptActivity`, `markAgentActivity`, or + `recordSteerActivity`; all three already advance the clock and must keep + doing so. +- No new field and no second clock. The completion-signal watchdog keeps reading + the same snapshot. + +## Acceptance conditions + +1. A prompt that emits only `usage_update`, `context_window`, + `available_commands_update`, and `session_info_update` frames after dispatch + publishes exactly one stall event five minutes after dispatch, with + `NeverStarted` true. +2. A prompt that emits a turn event and then only metadata frames reports + `NeverStarted` false, and its clock runs from the turn event. +3. `GetPromptActivityForSession` returns the same corrected timestamp, so the + completion-signal watchdog is not fooled by metadata frames either. + +## Verification + +```sh +cd apps/backend && go test ./internal/agent/runtime/lifecycle/... -count=1 +cd apps/backend && go test ./internal/orchestrator/... -count=1 +``` + +New regressions in +`apps/backend/internal/agent/runtime/lifecycle/session_test.go`: + +- `TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted` — drives + metadata frames through `Manager.recordActivity` once per minute inside + `synctest`, then asserts one `agent.stalled` event with `NeverStarted` true at + the five-minute mark. It must first fail because no event is published: the + metadata frames keep resetting the clock. +- `TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress` — asserts the + snapshot timestamp directly for a metadata frame and for a turn event. + +These four existing regressions must keep passing unchanged: +`TestWaitForPromptDone_PublishesSingleStall`, +`TestWaitForPromptDone_StallPayloadDiscriminatesNeverStarted`, +`TestWaitForPromptDone_StallPayloadReportsRunningAfterAgentEvent`, +`TestMarkPromptDispatchedArmsActivityAfterDispatch`. + +## Files likely touched + +- `apps/backend/internal/agent/runtime/lifecycle/manager_events.go` +- `apps/backend/internal/agent/runtime/lifecycle/types.go` +- `apps/backend/internal/agent/runtime/lifecycle/session_test.go` +- `apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go` + +## Dependencies + +None. + +## Results + +`recordActivity` (`manager_events.go`) now gates the `lastActivityAt` write +itself on `turnContentEventTypes`, instead of writing it unconditionally and +gating only `agentEventSincePrompt`/`promptActivityEpoch`. `firstActivityOnce` +and the wakeup-driven `Ready → Running` flip are unchanged. Updated the +`lastActivityAt` doc comment in `types.go` to state what advances it. + +Added `apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go` +(new file — the two existing test files are already over the 800-line test +file convention limit) with both specified regressions: +- `TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted` +- `TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress` + +Both failed first for the stated reason (metadata frames kept resetting the +clock), then passed after the fix. + +Also corrected a pre-existing test that pinned the defect and wasn't listed +among the "must keep passing unchanged" set: +`TestRecordActivity_MetadataDoesNotMarkPromptStarted` +(`manager_events_test.go`) asserted that a metadata event refreshes +`lastActivityAt` to "now" — exactly Defect 1. Updated its assertion to expect +the timestamp to stay unchanged; its other two assertions +(`agentEventSincePrompt` stays false, epoch stays 0) were already correct and +untouched. + +The four regressions named in this work order as must-keep-passing +(`TestWaitForPromptDone_PublishesSingleStall`, +`TestWaitForPromptDone_StallPayloadDiscriminatesNeverStarted`, +`TestWaitForPromptDone_StallPayloadReportsRunningAfterAgentEvent`, +`TestMarkPromptDispatchedArmsActivityAfterDispatch`) pass unchanged. + +Verification: +- `go test ./internal/agent/runtime/lifecycle/... -count=1` — passes except 7 + tests that are pre-existing, environment-caused failures unrelated to this + change (confirmed by running the same tests against unmodified HEAD via a + scoped `git stash`): 6 `TestWorktreePreparer_*` tests fail on worktree/git + state in this sandbox, and + `TestBuildAuthMethodsIdentityAgentOverridesEnvironment` fails on a Unix + socket path length limit from this checkout's long temp-dir path. Neither + touches `recordActivity`, `lastActivityAt`, or stall detection. +- `go test ./internal/orchestrator/... -count=1` — all packages pass. diff --git a/docs/plans/never-started-stall-teardown/task-02-terminal-stall-teardown.md b/docs/plans/never-started-stall-teardown/task-02-terminal-stall-teardown.md new file mode 100644 index 0000000000..e77858492b --- /dev/null +++ b/docs/plans/never-started-stall-teardown/task-02-terminal-stall-teardown.md @@ -0,0 +1,133 @@ +--- +id: "02-terminal-stall-teardown" +title: "Tear down a never-started execution" +status: done +wave: 2 +depends_on: ["01-honest-inactivity-clock"] +plan: "plan.md" +system_design: + - ../../specs/agents/system-design/agent-stall-recovery.md +requirements: + - REQ-AGENTS-AGENT-STALL-RECOVERY-001 +acceptance_criteria: + - AC-AGENTS-AGENT-STALL-RECOVERY-001.10 + - AC-AGENTS-AGENT-STALL-RECOVERY-001.11 + - AC-AGENTS-AGENT-STALL-RECOVERY-001.7 +--- + +# Task 02: Tear down a never-started execution + +## Outcome + +`handleAgentStalled` disposes of the process it declares dead. After the +never-started branch runs, the session and task are `FAILED` with the +launch-failure message and no agent process remains for that session. + +## In scope + +- In the `NeverStarted` branch of + `apps/backend/internal/orchestrator/event_handlers_stall.go`, after + `recordSessionLaunchFailure`, stop the execution named by + `payload.AgentExecutionID` through `agentManager.StopAgentWithReason` with + `force: true` and a stable reason. +- Order is record-then-tear-down, per the system design: the recorded `FAILED` + state stays authoritative even if teardown fails. +- Use `context.WithoutCancel` for the stop so a cancelled request context cannot + abandon the teardown midway. +- Treat `lifecycle.ErrExecutionNotFound` as success — nothing is running — and + log any other failure at warn with task, session, and execution identity. +- Skip the stop when `payload.AgentExecutionID` is empty. + +## Exclusions + +- Do not route through `StopByTaskID` or any path that writes session state; it + would replace `FAILED` with a cancellation state and destroy the error message + the requirement asks the user to see. +- Do not change the advisory branch. A stall after genuine activity must still + leave task state, session state, prompt admission, and process liveness + untouched. +- Do not change the notice copy, metadata, or the ownership guards that precede + the branch. + +## Acceptance conditions + +1. A never-started stall stops the execution exactly once, with force, after the + session and task are recorded `FAILED`. +2. The session's final state is `FAILED` with the launch-failure message; no + cancellation state is written by this path. +3. A teardown failure is logged and does not change the recorded state or + re-post the notice; an advisory stall still stops nothing. + +## Verification + +```sh +cd apps/backend && go test ./internal/orchestrator/... -count=1 +``` + +New regressions in +`apps/backend/internal/orchestrator/event_handlers_stall_test.go`: + +- `TestHandleAgentStalled_NeverStartedStopsExecution` — a fake agent manager + records stop calls; asserts one forced stop for the payload's execution ID and + a final session state of `FAILED`. It must first fail with zero stop calls. +- `TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails` — the fake + returns an error; asserts the session stays `FAILED` with the launch-failure + message. +- `TestHandleAgentStalled_AdvisoryStallStopsNothing` — asserts zero stop calls + for a stall with `NeverStarted` false. + +These existing regressions must keep passing unchanged: +`TestHandleAgentStalled_PersistsNeutralRunningNotice`, +`TestHandleAgentStalled_NeverStartedFailsSessionAndTask`, +`TestHandleAgentStalled_RejectsActivityEpochChangedAfterSnapshot`, +`TestHandleAgentStalled_RejectsSettledOrStalePrompt`. + +## Files likely touched + +- `apps/backend/internal/orchestrator/event_handlers_stall.go` +- `apps/backend/internal/orchestrator/event_handlers_stall_test.go` + +## Dependencies + +Task 01. Without the honest clock the never-started branch is not reliably +reachable on schedule, so its teardown cannot be exercised against the real +condition. + +## Results + +Added `Service.stopNeverStartedExecution` (`event_handlers_stall.go`), called +after `recordSessionLaunchFailure` inside the `payload.NeverStarted` branch of +`handleAgentStalled`. It skips when `payload.AgentExecutionID` is empty, stops +via `s.agentManager.StopAgentWithReason` with `force: true` on a +`context.WithoutCancel(ctx)`-detached context and a stable reason, treats +`lifecycle.ErrExecutionNotFound` as success, and logs any other failure at +warn with task/session/execution identity. The advisory branch and its +notice/metadata are untouched. + +Added the three specified regressions to `event_handlers_stall_test.go`: +- `TestHandleAgentStalled_NeverStartedStopsExecution` — failed first with zero + stop calls, now asserts exactly one forced stop for the payload's execution + and a final `FAILED` session state. +- `TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails` — fake + `StopAgentWithReason` returns an error; session stays `FAILED` with + `errAgentNeverStarted`'s message. +- `TestHandleAgentStalled_AdvisoryStallStopsNothing` — `NeverStarted: false` + asserts zero stop calls. + +The four existing regressions named in the work order +(`TestHandleAgentStalled_PersistsNeutralRunningNotice`, +`TestHandleAgentStalled_NeverStartedFailsSessionAndTask`, +`TestHandleAgentStalled_RejectsActivityEpochChangedAfterSnapshot`, +`TestHandleAgentStalled_RejectsSettledOrStalePrompt`) pass unchanged. + +Verification: `go test ./internal/orchestrator/... -count=1` is green for +every `TestHandleAgentStalled_*` test, run both individually and as part of +the full package, repeatedly. Full-package runs on this host are noisy: the +sandbox disk is at 99% capacity (~12Gi free), and a full run intermittently +fails a handful of unrelated tests elsewhere in the package (git/CI +automation, workflow clarification, task launch) with `disk I/O error: no +space left on device` / `sql: database is closed` / timing-sensitive +assertions — a different random subset each run. Every test that failed in a +full run passed cleanly when re-run in isolation, and none touches +`handleAgentStalled`, `stopNeverStartedExecution`, or stall detection. This is +a pre-existing environment condition, not a regression from this change. diff --git a/docs/plans/never-started-stall-teardown/task-03-stop-reaches-registered-execution.md b/docs/plans/never-started-stall-teardown/task-03-stop-reaches-registered-execution.md new file mode 100644 index 0000000000..f6e5731d8b --- /dev/null +++ b/docs/plans/never-started-stall-teardown/task-03-stop-reaches-registered-execution.md @@ -0,0 +1,268 @@ +--- +id: "03-stop-reaches-registered-execution" +title: "Reach a registered execution from a task stop" +status: done +wave: 3 +depends_on: ["02-terminal-stall-teardown"] +plan: "plan.md" +system_design: + - ../../specs/tasks/system-design/task-stop-reachability.md +requirements: + - REQ-TASKS-TASK-STOP-REACHABILITY-001 +acceptance_criteria: + - AC-TASKS-TASK-STOP-REACHABILITY-001.1 + - AC-TASKS-TASK-STOP-REACHABILITY-001.2 + - AC-TASKS-TASK-STOP-REACHABILITY-001.3 + - AC-TASKS-TASK-STOP-REACHABILITY-001.4 + - AC-TASKS-TASK-STOP-REACHABILITY-001.5 + - AC-TASKS-TASK-STOP-REACHABILITY-001.6 + - AC-TASKS-TASK-STOP-REACHABILITY-001.7 +--- + +# Task 03: Reach a registered execution from a task stop + +## Outcome + +`Executor.StopByTaskID` resolves what to halt from the union of the +active-session query and the in-memory execution registry. A session that is +terminal in the database but still holds a registered execution is stopped, and +its terminal state and error message survive. + +## In scope + +- Add a read-only task-scoped lookup to the execution registry: + `ExecutionStore` gains a method returning paired session and execution IDs for + executions whose `TaskID` matches, and `Manager` exposes it. + `AgentExecution.TaskID` already exists; no new index is required. +- Add that method to `executor.AgentManagerClient`. +- In `StopByTaskID`, after the existing `ListActiveTaskSessionsByTaskID` call, + add registry-only sessions by loading each session row with + `repo.GetTaskSession`. Keep and log a paired registry reference when a row + cannot be loaded, and schedule teardown by its captured execution ID. Return + `ErrExecutionNotFound` only when the union is empty. +- Stop a registry-recovered session with the per-session lock used by resume. + Preserve a terminal row, but use the normal cancellation transition when the + captured execution still owns a row that became active. Always target the + captured execution ID. Extract the shared stop-scheduling part of + `stopWithSession` rather than duplicating it. +- Log recovered sessions distinctly from query-resolved sessions, naming the + session and execution. + +## Exclusions + +- Do not widen the SQL active-state set in + `apps/backend/internal/task/repository/sqlite/session.go`. Every other caller + of `ListActiveTaskSessionsByTaskID` depends on its current meaning. +- Do not change `stopSession`, `StopExecution`, the `agent.cancel` path, or + `StopTaskForCoordinator`. +- Do not change the behavior for a session already returned by the active + persisted-state query. +- Do not add startup or background reconciliation of orphans. + +## Acceptance conditions + +1. A task whose only session is `FAILED` with a registered execution schedules + its execution stop once, and the session stays `FAILED` with its original + error message. Success means scheduling, not confirmed process exit. +2. A task with an active session behaves exactly as today, including the + cancellation transition and its events. +3. A task with neither an active session nor a registered execution still + returns `ErrExecutionNotFound` and changes nothing; a repeated stop after the + executions are gone creates no duplicate transition. + +## Verification + +```sh +cd apps/backend && go test ./internal/orchestrator/... -count=1 +cd apps/backend && go test ./internal/agent/runtime/lifecycle/... -count=1 +``` + +New regressions: + +- `apps/backend/internal/orchestrator/executor/executor_interaction_test.go`: + `TestStopByTaskID_StopsFailedSessionWithRegisteredExecution` — seeds a + `FAILED` session plus a registered execution in the fake agent manager; + asserts no error, one stop for that execution, and an unchanged session state. + It must first fail with `ErrExecutionNotFound` and zero stops. + `TestStopByTaskID_ActiveSessionKeepsCancellationTransition` and + `TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound` + pin the unchanged paths. +- `apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go`: + `TestExecutionStore_ListsExecutionsForTask` — covers matching, non-matching, + empty-task, and post-`Remove` cases. + +## Files likely touched + +- `apps/backend/internal/agent/runtime/lifecycle/execution_store.go` +- `apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go` +- `apps/backend/internal/agent/runtime/lifecycle/manager_execution.go` +- `apps/backend/internal/orchestrator/executor/executor.go` +- `apps/backend/internal/orchestrator/executor/executor_interaction.go` +- `apps/backend/internal/orchestrator/executor/executor_interaction_test.go` + +## Dependencies + +Task 02. Its regression constructs the orphan that Task 02 stops producing, so +the fake agent manager, not the stall path, seeds the registered execution. + +## Results + +Added `ExecutionStore.ListExecutionsForTask(taskID) []ExecutionReference` +(`execution_store.go`) — a read-only in-memory scan of registered executions +by `TaskID` that keeps each session ID paired with its execution ID — and +exposed it as `Manager.ListExecutionsForTask` (`manager_execution.go`). Added +it to `executor.AgentManagerClient` (`executor.go`) and wired the real path +through `lifecycleAdapter.ListExecutionsForTask` (`backendapp/adapters.go`). + +In `executor_interaction.go`: +- Added a shared exact-execution stop helper. It records advisory ownership and + schedules detached teardown, with an optional cancellation transition. +- Added `stopRegistryRecoveredSession`, which keeps the captured execution + reference under the per-session resume lock. It preserves terminal state, or + uses the cancellation transition if the captured execution still owns a row + that became active. +- Added `stopRegistryRecoveredByID`, which schedules the captured execution + even when the session row cannot be loaded. +- Added `recoverRegistryOnlySessions`, called from `StopByTaskID` after the + existing `ListActiveTaskSessionsByTaskID` call: it asks the registry for + taskID's paired execution references, skips any already covered by the active + query, loads each remaining row via `repo.GetTaskSession`, and retains the + reference when a row load fails. + `StopByTaskID` now returns `ErrExecutionNotFound` only when both the active + query and the recovered set are empty; the SQL active-state set in + `sqlite/session.go` was not touched. + +Added regressions: +- `execution_store_test.go`: + `TestExecutionStore_ListsExecutionsForTask` — matching/non-matching/empty- + task/post-`Remove` cases. +- `executor_interaction_test.go`: + `TestStopByTaskID_StopsFailedSessionWithRegisteredExecution` (new + behavior — a FAILED session with a registry-only execution is stopped, + forced, once, with its FAILED state and error message unchanged), + `TestStopByTaskID_ActiveSessionKeepsCancellationTransition` and + `TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound` + (pin the unchanged paths, including that a repeated stop after both sets + are empty stays `ErrExecutionNotFound` with no duplicate transition). + +RED verification for the primary test: reverted the interface/store/executor +production and mock plumbing via a scoped `git stash` (unique tag, applied by +SHA, dropped after) and confirmed the package fails to *compile* without it +(`unknown field listExecutionsForTaskFunc in struct literal of type +mockAgentManager`) — the registry-recovery capability does not exist on +`main` at all, so there is no lesser "wrong behavior" RED state to assert +against; a compile failure is the correct RED here. Re-applied the stash and +confirmed all three new tests pass. + +Adding `ListExecutionsForTask` to `executor.AgentManagerClient` required +updating every other implementer to keep the codebase compiling: the mock +`GetExecutionIDForSession` implementations in +`internal/orchestrator/event_handlers_test.go`, +`internal/orchestrator/scheduler/scheduler_test.go`, and +`internal/orchestrator/executor/executor_mocks_test.go` (all gained a real +`listExecutionsForTaskFunc`-backed method), plus +`internal/integration/simulated_agent_manager_test.go`'s +`SimulatedAgentManagerClient` (gained a real implementation filtering its +`simulatedInstance` map by `taskID`, matching its other lookups). + +Verification: `go test ./internal/orchestrator/... -count=1` — all packages +green. `go test ./internal/agent/runtime/lifecycle/... -count=1` — green +except the same 7 pre-existing, environment-caused failures documented in +Task 02's results (disk at 99% capacity on this host); none touch this +change. `go build ./...` — clean. `gofmt -l` on every touched file — clean. + +### Hardening: recovery-path fail-open on a transient load error + +Self-review of this task's own recovery path (walking its failure branches, +not the happy path) found a real fail-open bug: `recoverRegistryOnlySessions` +logged and silently dropped a registered session when `repo.GetTaskSession` +returned an error. If that was the *only* registered session, `StopByTaskID` +then returned the same `ErrExecutionNotFound` it returns when nothing is +registered at all — collapsing "a transient DB read failed" (e.g. `database +is locked`, which this environment demonstrably produces under load) into +"there is no orphan," exactly the false all-clear this task exists to +eliminate. + +Fixed: `recoverRegistryOnlySessions` now returns `([]*models.TaskSession, +error)`, keeping the last load error. `StopByTaskID` no longer reports +`ErrExecutionNotFound` when a registered session exists but couldn't be +loaded — it wraps and returns the real error instead, so a transient failure +is never indistinguishable from "genuinely nothing to stop." + +New regression (`executor_interaction_test.go`): +`TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound` — a +registered session whose `GetTaskSession` load fails; asserts the returned +error is not `ErrExecutionNotFound` and wraps the load failure. Failed first +with `error = execution not found, want the load failure surfaced instead of +ErrExecutionNotFound`, then passed after the fix. + +Verification: `go test ./internal/orchestrator/executor/... -count=1` — +green (including the 3 existing `TestStopByTaskID_*` tests, unaffected). +`go build ./...` — clean. `gofmt -l` — clean. `golangci-lint run +./internal/orchestrator/executor/... --new-from-rev=` — 0 issues. + +### RUNNING-orphan investigation (requested before extending defect 3) + +Asked to establish, before trusting defect 3's FAILED-only scope: does +defect 1 (pre-fix) leave a stalled session at RUNNING instead of FAILED, and +if so, is `StopByTaskID` blind to *that* case too, or only to FAILED? + +**Pre-fix, a metadata-only stream leaves the session at RUNNING forever, not +FAILED.** `handleAgentStalled` (`event_handlers_stall.go`) is the only code +path that transitions a stalled session to FAILED (via +`recordSessionLaunchFailure`, gated on `payload.NeverStarted`). Pre-Task-01, +`recordActivity` kept resetting `lastActivityAt` for metadata frames, so +`waitForPromptDone`'s 5-minute condition never tripped, `handleAgentStalled` +was never called, and the session simply stayed in whatever state it already +had — RUNNING, since that's the state a session is in while a prompt is +in flight. This is the same defect already named as defect 1, stated in +terms of its state consequence. + +**`StopByTaskID` was never blind to a RUNNING orphan.** +`ListActiveTaskSessionsByTaskID`'s SQL set is `CREATED, STARTING, RUNNING, +WAITING_FOR_INPUT` — RUNNING has always been in it, so a session stuck at +RUNNING with a still-registered execution was always reachable through the +pre-existing, unmodified `stopWithSession` path. Defect 3's blindness is +specific to FAILED, a state that only becomes reachable-but-excluded once +defect 2 exists to produce it. There is no second, RUNNING-side blindness for +Task 03 to also close. + +**Conclusion: no fourth defect shape.** Task 01 (honest clock) + Task 02 +(forced teardown on the now-honestly-detected stall) together eliminate the +"stuck at RUNNING forever" failure mode at its source — the watchdog now +fires and the process is torn down, rather than needing a second reachability +path for a state that should no longer occur. Nothing new to build here. + +**The two cited live specimens (d8631f7a, 7e8f19cd) are a different, out-of- +scope shape**, exactly as flagged when raised: both are long-running +sessions with thousands of real messages, not never-started ones, and the +description of their stuckness (a workflow-board move deferred to turn-end on +a session the board still calls RUNNING, so no move takes effect) points at +the Kandev workflow/board layer, not at `StopByTaskID`'s session resolution. +Not folded into this card's fix; not investigated further, per the explicit +scope instruction to name rather than silently absorb a fourth shape. + +**Third data point (this card's own never-started launch, no session/output +for 10 minutes):** consistent with defect 1 exactly as modeled above. The fix +is implemented in this worktree but cannot be observed live against that +specific already-stuck launch from here — doing so needs the actual running +backend process for this Kandev instance restarted onto the fixed code, which +is a deployment action outside this card's scope (a source change, not an +operational one). + +### Contributor fixup: preserve execution identity during recovery + +Registry recovery now returns a paired `SessionID` and `ExecutionID` snapshot. +Both recovery paths keep that pair through the per-session lock used by resume. +They schedule the captured execution even when the session row is unreadable. +When a row becomes active, recovery changes it to `CANCELLED` only if the +captured execution still owns the session. A replacement execution is never +selected by a later session lookup. + +The stop contract reports success when teardown is scheduled. It does not wait +for detached runtime teardown to finish. A row load error remains visible to +the caller while the captured execution is still scheduled. + +Regression coverage includes replacement execution protection for both loaded +and unloadable rows, plus a terminal-to-active recovery transition. The +execution store test verifies that each task snapshot contains both IDs. diff --git a/docs/specs/agents/requirements/agent-stall-recovery.md b/docs/specs/agents/requirements/agent-stall-recovery.md index 00786d894a..c3ae3c03e5 100644 --- a/docs/specs/agents/requirements/agent-stall-recovery.md +++ b/docs/specs/agents/requirements/agent-stall-recovery.md @@ -2,7 +2,7 @@ status: active system: agents created: 2026-07-29 -updated: 2026-08-27 +updated: 2026-09-02 owners: - Kandev --- @@ -12,6 +12,18 @@ owners: An agent turn can remain `RUNNING` after it stops emitting events. Silence alone is ambiguous because a legitimate long-running tool may also be quiet, but some agents know that their provider stream has failed and fail to close the ACP prompt. Users need Kandev to distinguish trustworthy terminal evidence from mere inactivity so a failed turn is not presented as healthy work. +## Terminology + +- **Turn event:** An agent frame that proves the model is working on the current + prompt: assistant text, reasoning, a tool call or tool update, a plan update, + or a permission request. +- **Metadata frame:** Any other agent frame, such as a usage update, a context + window update, an available-commands update, or a session-info update. A + metadata frame can be emitted by a booting or hung adapter that has produced + no turn event. +- **Never-started prompt:** A dispatched prompt for which no turn event has been + observed since dispatch. + ## Requirements ### REQ-AGENTS-AGENT-STALL-RECOVERY-001: Agent Stall Recovery @@ -28,6 +40,9 @@ An agent turn can remain `RUNNING` after it stops emitting events. Silence alone - **AC-AGENTS-AGENT-STALL-RECOVERY-001.6:** The notice remains visible and actionable while the affected prompt's `turn_id` is the active turn in a `RUNNING` session, including after a page reload. It is hidden when that turn settles or a later turn becomes active. - **AC-AGENTS-AGENT-STALL-RECOVERY-001.7:** Detection after genuine agent activity does not change task state, session state, prompt admission, or process liveness. A current five-minute snapshot with no genuine event since prompt dispatch is a launch failure and moves the session and task to `FAILED`. - **AC-AGENTS-AGENT-STALL-RECOVERY-001.8:** The backend logs the first stall detected for a prompt generation and does not emit another notice or log entry on every watchdog check. +- **AC-AGENTS-AGENT-STALL-RECOVERY-001.9:** When an execution emits only metadata frames after a prompt is dispatched, the system shall not extend the five-minute inactivity threshold; only a turn event or newly delivered user input for the current prompt shall restart it. +- **AC-AGENTS-AGENT-STALL-RECOVERY-001.10:** When a never-started prompt is classified as a launch failure, the system shall stop its agent execution so no agent process for that session remains running, and shall record the failure even when the stop fails. +- **AC-AGENTS-AGENT-STALL-RECOVERY-001.11:** When the terminal never-started classification is applied, the session and task shall settle in `FAILED` with the launch-failure message; the teardown shall not replace that state with a cancellation state. ## Migrated source detail @@ -38,47 +53,28 @@ Decisions: - [ADR-2026-08-07-allowlisted-provider-action-links](../../../decisions/2026-08-07-allowlisted-provider-action-links.md) - [ADR-2026-08-08-provider-neutral-agent-error-recovery](../../../decisions/2026-08-08-provider-neutral-agent-error-recovery.md) - [ADR-2026-08-18-never-started-agent-stall-terminal](../../../decisions/2026-08-18-never-started-agent-stall-terminal.md) +- [ADR-2026-09-02-terminal-stall-owns-process-teardown](../../../decisions/2026-09-02-terminal-stall-owns-process-teardown.md) Implementation plans: - [Agent stall recovery](../../../plans/agent-stall-recovery/plan.md) +- [Never-started stall teardown](../../../plans/never-started-stall-teardown/plan.md) - [Lost turn-completion cancel recovery](../../../plans/lost-turn-completion-cancel-recovery/plan.md) - [OpenCode terminal error surfacing](../../../plans/opencode-terminal-error-surfacing/plan.md) - [OpenCode actionable error links](../../../plans/opencode-actionable-error-links/plan.md) -## Why - -An agent turn can remain `RUNNING` after it stops emitting events. Silence alone -is ambiguous because a legitimate long-running tool may also be quiet, but some -agents know that their provider stream has failed and fail to close the ACP -prompt. Users need Kandev to distinguish trustworthy terminal evidence from -mere inactivity so a failed turn is not presented as healthy work. - ## What ### Advisory inactivity recovery -- Kandev checks running prompts for inactivity once per 60 seconds. -- After five minutes without agent events, Kandev creates at most one - user-visible notice for that prompt generation. -- The notice says Kandev is still waiting and includes the active top-level - tool's display title or name when available. It does not assert that the tool - failed and does not include raw command arguments. -- The notice is a single compact inline row with muted neutral copy and a - neutral **Cancel turn** action. It has no warning/error colors, alert icon, - tinted background, or alert-card treatment. -- On phones, **Cancel turn** remains inline and content-width rather than - becoming a full-width row, while retaining a minimum 44px touch height. - Activating it uses the existing `agent.cancel` request. -- The notice remains visible and actionable while the affected prompt's - `turn_id` is the active turn in a `RUNNING` session, including after a page - reload. It is hidden when that turn settles or a later turn becomes active. -- Detection after genuine agent activity does not change task state, session - state, prompt admission, or process liveness. A current five-minute snapshot - with no genuine event since prompt dispatch is a launch failure and moves the - session and task to `FAILED`. -- The backend logs the first stall detected for a prompt generation and does - not emit another notice or log entry on every watchdog check. +The observable rules are the acceptance criteria of +REQ-AGENTS-AGENT-STALL-RECOVERY-001. The inactivity clock measures time since +the last turn event or user input for the current prompt, so an adapter that +emits only metadata frames cannot hold a never-started prompt open forever. A +stall observed after genuine turn activity stays advisory and leaves the turn +and its process running. A never-started prompt is the terminal exception: it +posts an error notice, stops the agent execution, and settles the session and +task in `FAILED`. ### Explicit completion signal recovery @@ -100,8 +96,9 @@ mere inactivity so a failed turn is not presented as healthy work. - A failed reconciliation does not lose the signal. Later watchdog ticks retry a `WAITING_FOR_INPUT` session while the matching signal remains. -This is the only automatic state-changing exception to the advisory inactivity -rule. Silence without an accepted explicit completion signal remains advisory. +With the never-started classification, this is one of the two automatic +state-changing exceptions to the advisory inactivity rule. Silence after a turn +event, without an accepted explicit completion signal, remains advisory. ### Correlated terminal diagnostics @@ -242,6 +239,13 @@ sanitized diagnostic message for the collapsed technical-details surface. **WHEN** the current five-minute watchdog snapshot is handled, **THEN** Kandev persists a terminal error, moves the session and task to `FAILED`, and does not render a running-only cancel action. +- **GIVEN** a prompt that has emitted only usage, context-window, or + available-commands frames since dispatch, **WHEN** five minutes pass, **THEN** + the watchdog still reports the prompt as never started rather than restarting + its clock on each metadata frame. +- **GIVEN** the never-started classification is applied, **WHEN** the session + and task settle in `FAILED`, **THEN** the agent execution is stopped and a + later stop request for that task does not find a live process. - **GIVEN** a quiet but legitimate long-running turn, **WHEN** the inactivity threshold passes and the user does not cancel, **THEN** Kandev leaves the turn and process running. @@ -290,8 +294,10 @@ sanitized diagnostic message for the collapsed technical-details surface. ## Out of scope -- Automatically timing out, cancelling, or killing a turn based only on - inactivity. +- Automatically timing out, cancelling, or killing a turn that has already + produced a turn event, based only on inactivity. +- Making the stop path reach an execution for a task; that contract belongs + to [task stop reachability](../../tasks/requirements/task-stop-reachability.md). - Making the inactivity threshold user-configurable. - Reading, tailing, or exposing an agent vendor's private log files. - Treating arbitrary stderr text as trusted terminal evidence. diff --git a/docs/specs/agents/system-design/agent-stall-recovery.md b/docs/specs/agents/system-design/agent-stall-recovery.md new file mode 100644 index 0000000000..bac15f3c29 --- /dev/null +++ b/docs/specs/agents/system-design/agent-stall-recovery.md @@ -0,0 +1,152 @@ +--- +status: draft +system: agents +requirements: + - REQ-AGENTS-AGENT-STALL-RECOVERY-001 +created: 2026-09-02 +owners: + - Kandev +--- + +# Agent Stall Recovery System Design + +## Purpose and boundaries + +The agent system owns the prompt watchdog: the clock that decides a prompt is +inactive, the classification of that inactivity as advisory or never-started, +and the teardown of an execution whose prompt never started. + +It does not own the persisted session and task state machine, the notice +rendering, or the task-scoped stop path. It calls the orchestrator's existing +launch-failure transition, reuses the existing message creator, and leaves +reaching an execution from a task ID to +[task stop reachability](../../tasks/system-design/task-stop-reachability.md). + +## Requirement mapping + +| Requirement | Design sections | +| --- | --- | +| `REQ-AGENTS-AGENT-STALL-RECOVERY-001` | [Prompt progress clock](#prompt-progress-clock), [Watchdog classification](#watchdog-classification), [Terminal teardown](#terminal-teardown), [Failure and recovery](#failure-and-recovery) | + +## Components and responsibilities + +- **`AgentExecution` activity state** (`internal/agent/runtime/lifecycle`) holds + the per-prompt progress timestamp, the never-started discriminator, and the + activity epoch under one mutex. +- **The lifecycle event recorder** classifies each inbound agent frame as a turn + event or a metadata frame and advances progress state only for turn events. +- **The prompt watchdog** in the session manager ticks once per minute, takes a + consistent snapshot, and publishes at most one stall event per prompt + generation. +- **The orchestrator stall handler** validates the snapshot's ownership, + persists the notice, and, for a never-started prompt, applies the terminal + transition and requests execution teardown. +- **The agent manager's stop entry point** performs teardown. It stops the + runtime and deregisters the execution without writing session state. + +## Data and contracts + +The published stall payload is unchanged: task, session, and execution +identity, prompt generation, last-activity time, elapsed duration, activity +epoch, optional active-tool identity, and the `NeverStarted` discriminator. + +The prompt progress timestamp is the single clock behind that payload. Exactly +four inputs advance it: + +| Input | Meaning | +| --- | --- | +| Prompt dispatch | A new prompt is armed; the clock restarts and the discriminator is cleared | +| Turn event | Assistant text, reasoning, tool call, tool update, plan, or permission request | +| Terminal completion | The prompt's own completion or error frame | +| User steer | New user input delivered into the running turn | + +A metadata frame advances nothing. It is still processed, published, and +recorded for its own purposes; it is not evidence of progress. + +The discriminator remains separate from the clock. Prompt dispatch clears it; +a turn event or terminal completion sets it. A user steer restarts the clock +without setting it, because user input is not agent output. + +## Control flow + +1. Prompt dispatch arms the clock and clears the discriminator. +2. Each inbound frame is classified. A turn event advances the clock, sets the + discriminator, and bumps the activity epoch. A metadata frame does neither. +3. Every minute the watchdog snapshots the clock, the discriminator, and the + epoch together. Once the elapsed time reaches five minutes it publishes one + stall event for the prompt generation and does not publish again. +4. The orchestrator rejects the event when the session is no longer running, + when cancellation is in flight, or when the execution, prompt generation, or + activity epoch is no longer current. +5. An accepted advisory event persists the neutral running notice and stops. +6. An accepted never-started event persists the terminal error notice, applies + the launch-failure transition, and then requests execution teardown. + +### Prompt progress clock + +Because the clock advances only on the four inputs above, an adapter that emits +metadata frames while producing no turn output cannot postpone detection. The +never-started elapsed time is measured from prompt dispatch, which is the +behavior the requirement describes. + +The same clock also feeds the ten-minute completion-signal watchdog. That +consumer inherits the corrected semantics: a signalled session that has gone +quiet apart from metadata frames becomes reclaimable on schedule instead of +never. No separate clock is introduced for it. + +### Watchdog classification + +Classification is a pure read of the snapshot. An advisory stall leaves task +state, session state, prompt admission, and process liveness untouched. A +never-started stall is terminal. + +### Terminal teardown + +Terminal handling is ordered: record the failure first, then tear down. + +Recording first makes the durable outcome the launch-failure state the user +sees, and keeps it authoritative if teardown fails or races. Teardown then uses +the execution-scoped stop, which halts the runtime and deregisters the +execution without touching session rows, so it cannot replace the recorded +`FAILED` state with a cancellation state. + +Teardown is requested with force, because a prompt that produced no frame is +not expected to answer a graceful protocol stop. + +## Failure and recovery + +- Teardown failure is logged with the session and execution identity. The + session and task remain `FAILED`; the execution stays registered so a later + task-scoped stop or startup cleanup can retry it. +- A late turn event that lands between the snapshot and the handler moves the + activity epoch and the event is rejected, so a prompt that started late is not + failed. +- Notice persistence failure is logged and does not block the terminal + transition or the teardown. +- An already-removed execution reports not-found on teardown. That is the + success condition, not an error, because no process remains. + +## Persistence + +No schema change. The clock, discriminator, and epoch stay in memory for the +lifetime of the execution. The durable record is the existing session state, +session error message, task state, and notice message. + +## Security + +The notice copy and the log fields keep their existing bounds: identifiers, +durations, and tool display names only. Teardown adds no new user input path +and no new externally reachable operation. + +## Observability + +The existing single stall log per prompt generation is retained and keeps its +`never_started` field. Terminal handling adds one log line for the teardown +result, carrying task, session, and execution identity plus the outcome, so an +orphan left behind by a failed teardown is greppable. + +## Related decisions + +- [ADR-2026-07-29-agent-stall-user-controlled-recovery](../../../decisions/2026-07-29-agent-stall-user-controlled-recovery.md) +- [ADR-2026-08-18-never-started-agent-stall-terminal](../../../decisions/2026-08-18-never-started-agent-stall-terminal.md) +- [ADR-2026-09-02-terminal-stall-owns-process-teardown](../../../decisions/2026-09-02-terminal-stall-owns-process-teardown.md) diff --git a/docs/specs/tasks/README.md b/docs/specs/tasks/README.md index 94f6feca72..9302171a90 100644 --- a/docs/specs/tasks/README.md +++ b/docs/specs/tasks/README.md @@ -43,5 +43,5 @@ migration state. - [Agents](../agents): supplies agent identity and execution profiles. - [Office](../office): builds autonomous workflows on task primitives. -- [UI](../ui): owns presentation-specific task surfaces. +- [UI](../ui): owns presentation-only task surfaces. - [Workspaces](../workspaces): owns repositories and task worktrees. diff --git a/docs/specs/tasks/requirements/task-stop-reachability.md b/docs/specs/tasks/requirements/task-stop-reachability.md new file mode 100644 index 0000000000..4425b07aa6 --- /dev/null +++ b/docs/specs/tasks/requirements/task-stop-reachability.md @@ -0,0 +1,81 @@ +--- +status: draft +system: tasks +created: 2026-09-02 +owners: + - Kandev +--- + +# Task Stop Reachability Requirements + +## Overview + +Stopping a task must halt the agent processes that task started. Kandev records +a session state in the database and separately tracks live agent executions in +the running backend. When a session is marked terminal while its execution is +still registered and its process is still alive, a database-only view of "which +sessions are active" cannot see the work that is still running. The task then +presents a settled card that the user cannot stop, and the agent process keeps +holding its workspace, its runtime, and its provider quota. + +The task system owns this contract because stopping a task is a task-lifecycle +operation. The agent system owns the execution registry that this contract +reads. + +## Terminology + +- **Registered execution:** An agent execution that the running backend still + tracks for a session, independent of the session's persisted state. +- **Orphaned execution:** A registered execution whose session is already in a + terminal persisted state. +- **Active persisted state:** A session state that a database query treats as + live work, currently `CREATED`, `STARTING`, `RUNNING`, or `WAITING_FOR_INPUT`. + +## Requirements + +### REQ-TASKS-TASK-STOP-REACHABILITY-001: Task Stop Reachability + +**Intent:** Make a stop request reach every agent process the task still owns, +so a terminal session state can never hide a live process from the user. + +**User story:** As a user whose task failed while its agent kept running, I want +the task-scoped stop to halt that agent, so that I can reclaim the task without +restarting the backend. + +#### Acceptance criteria + +- **AC-TASKS-TASK-STOP-REACHABILITY-001.1:** When a task-scoped stop runs, the + system shall stop every registered execution for that task, including an + orphaned execution whose session is not in an active persisted state. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.2:** When a task-scoped stop finds at + least one registered execution, it shall report success rather than + "execution not found", regardless of the persisted session states involved. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.3:** When a task has neither a session + in an active persisted state nor a registered execution, the stop shall + continue to report that no execution was found and shall change nothing. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.4:** When a stop halts an orphaned + execution, it shall preserve that session's terminal state and error message + rather than overwriting them with a cancellation state. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.5:** When a stop halts a session that + is in an active persisted state, the existing cancellation transition and its + events shall be unchanged. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.6:** When a stop request is repeated + after the executions are gone, it shall not fail on the already-removed + executions and shall not create duplicate state transitions. +- **AC-TASKS-TASK-STOP-REACHABILITY-001.7:** When a stop halts only orphaned + executions, the log shall identify each session and execution that was + reached through the execution registry rather than through the active-session + query. + +## Out of scope + +- Choosing when a session becomes terminal. The + [agent stall recovery](../../agents/requirements/agent-stall-recovery.md) + requirement owns the never-started classification that produced the first + observed orphan. +- Session-scoped stop and the `agent.cancel` recovery path, which already + resolve an execution directly and are unchanged. +- Reconciling executions left behind by a previous backend process; startup + cleanup already owns that path. +- Making a stopped task's workflow step or task state different from the state + that the existing stop path produces. diff --git a/docs/specs/tasks/system-design/task-stop-reachability.md b/docs/specs/tasks/system-design/task-stop-reachability.md new file mode 100644 index 0000000000..1f48b8f609 --- /dev/null +++ b/docs/specs/tasks/system-design/task-stop-reachability.md @@ -0,0 +1,129 @@ +--- +status: draft +system: tasks +requirements: + - REQ-TASKS-TASK-STOP-REACHABILITY-001 +created: 2026-09-02 +owners: + - Kandev +--- + +# Task Stop Reachability System Design + +## Purpose and boundaries + +The task system owns the task-scoped stop: resolving which work a task still +owns and halting it. This design changes only how that set is resolved. + +The agent system owns the execution registry and the stop primitive; this +design reads the registry and calls the primitive. Session-scoped stop and the +`agent.cancel` recovery path already resolve an execution directly and are not +changed. Which session becomes terminal, and why, belongs to the owning +behavior, such as +[agent stall recovery](../../agents/system-design/agent-stall-recovery.md). + +## Requirement mapping + +| Requirement | Design sections | +| --- | --- | +| `REQ-TASKS-TASK-STOP-REACHABILITY-001` | [Two sources of truth](#two-sources-of-truth), [Resolution](#resolution), [Stopping a recovered session](#stopping-a-recovered-session), [Failure and recovery](#failure-and-recovery) | + +## Two sources of truth + +Two independent records describe a task's live work: + +- The **persisted session state** in the database, queried by an active-state + set that deliberately excludes terminal states. +- The **execution registry** in the running backend, which holds an entry for + as long as an agent process is tracked. + +They can disagree. Any path that marks a session terminal without tearing down +its execution produces a session that the active-state query cannot see and a +process that is still running. The active-state query alone is therefore the +wrong authority for a stop. + +The registry is the authority for what can be stopped. The persisted state +remains the authority for what the session means. + +## Components and responsibilities + +- **The task-scoped stop** resolves the union of both sources, then stops each + resolved session. +- **The execution registry** answers which session and execution pairs of a task + currently have a registered execution. Its existing per-execution task + identity is enough; no new index is required for a set this small. +- **The session store** supplies the session row for a session that only the + registry knew about, so the stop has the identity and task binding it needs. + +## Resolution + +1. Query sessions in an active persisted state for the task, as today. +2. Ask the registry for sessions of the task that have a registered execution. +3. Add any registry-only session by loading its session row. Keep the paired + registry reference when the row cannot be loaded, log the load error, and + schedule teardown by the captured execution ID. +4. If the union is empty, report that no execution was found and change + nothing. + +Ordering keeps existing behavior first, so a task with no orphan resolves +exactly the set it resolves today. + +## Stopping a recovered session + +A session reached through the active-state query keeps today's behavior: it +transitions to the cancellation state and emits the existing events. + +A session reached only through the registry is normally terminal. Recovery +holds the per-session lock used by resume, then checks the current row and the +captured execution identity. It skips the state transition for a terminal row, +or when a replacement execution owns the session. If the captured execution +still owns an active row, recovery uses the normal cancellation transition. +Teardown always targets the captured execution ID. + +Success is reported when teardown is scheduled for at least one execution. It +does not confirm that detached runtime teardown has finished. A task whose only +live work was an orphan therefore stops successfully, and the caller's existing +post-stop task handling proceeds as it does for any successful stop. + +## Data and contracts + +The stop entry point keeps its signature and its not-found error. One read-only +capability is added to the agent-manager interface the task system already +depends on: given a task identity, return paired session and execution +identities that currently have a registered execution. It is a snapshot; a +session that changes between the snapshot and the stop is handled by the +per-session lock and exact execution check. + +No HTTP, WebSocket, or persisted contract changes. + +## Failure and recovery + +- A failed stop for one session does not prevent the others; the operation + reports failure only when every resolved session failed. +- A repeated stop finds the executions already removed and reports not-found per + session without creating another transition. +- A registry lookup that returns nothing degrades exactly to today's behavior. +- A session row that cannot be loaded is logged, but its captured execution is + still scheduled for teardown. The load error remains visible to the caller. + +## Persistence + +No schema change. The only persistence difference is a deliberate omission: an +orphan's terminal state is not overwritten. + +## Security + +The registry is process-local and already scoped per execution. Resolution is +constrained to the requested task, so a stop cannot reach another task's +execution. No new caller-controlled input is introduced. + +## Observability + +The stop log distinguishes sessions resolved from the active-state query from +sessions recovered through the registry, and names each recovered session and +execution. An orphan is then visible in the logs at the moment it is reclaimed +rather than only as an unexplained not-found error. + +## Related decisions + +- [ADR-2026-09-02-terminal-stall-owns-process-teardown](../../../decisions/2026-09-02-terminal-stall-owns-process-teardown.md)