diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index e199dd5b33..088539aed8 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -899,7 +899,7 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r } // Stop the agent execution via the runtime that created it - m.stopAgentViaBackend(ctx, executionID, execution, reason, force, agentStopFailed) + _ = m.stopAgentViaBackend(ctx, executionID, execution, reason, force, agentStopFailed) // Update execution status and remove from tracking _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { @@ -1317,27 +1317,29 @@ func (m *Manager) ListExecutions() []*AgentExecution { const agentStartupLivenessGrace = 75 * time.Second -// IsAgentRunningForSession checks if an agent process is running or starting for a session. +// ProbeAgentRunningForSession checks if an agent process is running or +// starting for a session and reports probe failures to the caller. // // For passthrough sessions (direct PTY mode), it checks whether the PTY process is alive // in the InteractiveRunner. For ACP sessions, it probes agentctl's status endpoint. // -// Returns true if: +// A successful probe returns true if: // - Passthrough process is alive in the InteractiveRunner // - Agent status is "running" (actively processing prompts) // - Agent status is "starting" (process launched but not yet ready) // -// Returns false if: +// A successful probe returns false if: // - No execution exists for this session // - Passthrough process ID is set but process is not alive -// - agentctl client is not available -// - Status check fails (network/timeout error) // - Agent is in any other state (stopped, failed, etc.) -func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { +// +// An unavailable agentctl client, interactive runner, or status endpoint is +// returned as an error rather than as a false result. +func (m *Manager) ProbeAgentRunningForSession(ctx context.Context, sessionID string) (bool, error) { // First check if we have an execution tracked for this session execution, exists := m.GetExecutionBySessionID(sessionID) if !exists { - return false + return false, nil } // Launch registers the execution before agentctl is guaranteed to answer @@ -1349,21 +1351,21 @@ func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string if execution.Status == v1.AgentStatusStarting && !execution.StartedAt.IsZero() && time.Since(execution.StartedAt) <= agentStartupLivenessGrace { - return true + return true, nil } // Passthrough sessions run as direct PTY processes via InteractiveRunner, // bypassing agentctl's ACP protocol. Check the process directly. if execution.PassthroughProcessID != "" { if runner := m.GetInteractiveRunner(); runner != nil { - return runner.IsProcessReadyOrPending(execution.PassthroughProcessID) + return runner.IsProcessReadyOrPending(execution.PassthroughProcessID), nil } - return false + return false, fmt.Errorf("interactive runner is unavailable") } // Probe agentctl status to verify the agent process is running if execution.agentctl == nil { - return false + return false, fmt.Errorf("agentctl client is unavailable") } status, err := execution.agentctl.GetStatus(ctx) @@ -1371,10 +1373,17 @@ func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string m.logger.Debug("failed to get agentctl status", zap.String("session_id", sessionID), zap.Error(err)) - return false + return false, err } - return status.IsAgentRunning() + return status.IsAgentRunning(), nil +} + +// IsAgentRunningForSession preserves the legacy boolean probe for callers +// that do not need to distinguish a dead process from an unavailable probe. +func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { + running, _ := m.ProbeAgentRunningForSession(ctx, sessionID) + return running } // IsAgentReadyForPrompt reports whether the session can accept an ACP prompt @@ -1899,9 +1908,9 @@ func (m *Manager) RespondToPermissionBySessionID(sessionID, pendingID, optionID } // stopAgentViaBackend stops the agent execution via the runtime that created it. -func (m *Manager) stopAgentViaBackend(ctx context.Context, executionID string, execution *AgentExecution, reason string, force bool, agentStopFailed bool) { +func (m *Manager) stopAgentViaBackend(ctx context.Context, executionID string, execution *AgentExecution, reason string, force bool, agentStopFailed bool) error { if execution.RuntimeName == "" || m.executorRegistry == nil { - return + return nil } rt, err := m.executorRegistry.GetBackend(execution.RuntimeName) if err != nil { @@ -1909,7 +1918,7 @@ func (m *Manager) stopAgentViaBackend(ctx context.Context, executionID string, e zap.String("execution_id", executionID), zap.Stringer("runtime", execution.RuntimeName), zap.Error(err)) - return + return fmt.Errorf("get runtime %s: %w", execution.RuntimeName, err) } m.stopPassthroughProcess(ctx, executionID, execution, rt) runtimeInstance := &ExecutorInstance{ @@ -1926,15 +1935,17 @@ func (m *Manager) stopAgentViaBackend(ctx context.Context, executionID string, e // During shutdown the runtime instance may already be stopping or // absent. Only surface this at WARN outside shutdown. if m.IsShuttingDown() { - m.logger.Debug("failed to stop runtime instance, continuing with cleanup", + m.logger.Debug("failed to stop runtime instance", zap.String("execution_id", executionID), zap.Error(err)) } else { - m.logger.Warn("failed to stop runtime instance, continuing with cleanup", + m.logger.Warn("failed to stop runtime instance", zap.String("execution_id", executionID), zap.Error(err)) } + return err } + return nil } // stopPassthroughProcess stops the passthrough interactive process if one is running. diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go index 9140e13ce2..76161b15ac 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go @@ -3,6 +3,7 @@ package lifecycle import ( "context" "errors" + "fmt" "sync" "time" @@ -274,14 +275,17 @@ func (m *Manager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionI zap.String("session_id", sessionID), zap.String("execution_id", execution.ID)) - // End session trace span - execution.EndSessionSpan() - // Stop the runtime instance (workspace tracker, shell, etc.) to prevent leaked // goroutines. Without this, the old agentctl instance keeps running when a new // execution is created for the same session, causing git polling on deleted worktrees. // This is idempotent — returns success if the instance is already gone. - m.stopAgentViaBackend(ctx, execution.ID, execution, stopReasonStaleExecutionCleanup, false, false) + if err := m.stopAgentViaBackend(ctx, execution.ID, execution, stopReasonStaleExecutionCleanup, false, false); err != nil { + return fmt.Errorf("stop stale runtime %s: %w", execution.ID, err) + } + + // End session trace span only after the runtime has stopped. A failed stop + // remains retryable and should keep the execution's lifecycle intact. + execution.EndSessionSpan() // Close agentctl connection if it exists if execution.agentctl != nil { diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle_test.go index 85dcd4c4f3..cf7b36b24e 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle_test.go @@ -3,6 +3,7 @@ package lifecycle import ( "context" "encoding/json" + "errors" "net" "net/http" "testing" @@ -212,12 +213,34 @@ func TestManager_CleanupStaleExecution_SkipsStopWhenNoRuntime(t *testing.T) { } } +func TestManager_CleanupStaleExecution_PreservesTrackingOnStopError(t *testing.T) { + log := newTestRegistryLogger() + reg := newTestRegistry() + execRegistry := NewExecutorRegistry(log) + mock := &mockStopTracker{name: "standalone", stopErr: errors.New("runtime unavailable")} + execRegistry.Register(mock) + mgr := NewManager(reg, &MockEventBus{}, execRegistry, &MockCredentialsManager{}, &MockProfileResolver{}, nil, ExecutorFallbackWarn, "", log) + + execution := &AgentExecution{ID: "exec-error", SessionID: "session-error", RuntimeName: "standalone"} + if err := mgr.executionStore.Add(execution); err != nil { + t.Fatalf("failed to seed execution: %v", err) + } + + if err := mgr.CleanupStaleExecutionBySessionID(context.Background(), "session-error"); err == nil { + t.Fatal("cleanup must report a runtime stop failure") + } + if _, found := mgr.GetExecutionBySessionID("session-error"); !found { + t.Fatal("failed cleanup must retain execution tracking for a retry") + } +} + // mockStopTracker is a minimal ExecutorBackend that records StopInstance calls. type mockStopTracker struct { name executor.Name stopCalled bool stoppedInstanceID string stopReason string + stopErr error } func (m *mockStopTracker) Name() executor.Name { return m.name } @@ -231,7 +254,7 @@ func (m *mockStopTracker) StopInstance(ctx context.Context, instance *ExecutorIn m.stopCalled = true m.stoppedInstanceID = instance.InstanceID m.stopReason = instance.StopReason - return nil + return m.stopErr } func TestManagerStopAllAgentsPassesBackendShutdownReason(t *testing.T) { diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence.go b/apps/backend/internal/agent/runtime/lifecycle/persistence.go index 075b9c2e1a..d43d7088e5 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence.go @@ -316,22 +316,18 @@ func (m *Manager) persistExecutorRunningResult(ctx context.Context, execution *A // so the in-memory and persistent state are gone in the same operation. // // Resume-safety invariant (#1597 resume-safety invariant): a row that still holds -// a resume_token is REPAIRED in place (status=stopped, local_pid cleared) rather -// than deleted, so a session stays resumable even if a subsequent relaunch fails. -// On the happy path the relaunch UPSERTs a fresh row over the repaired one, so -// repairing costs nothing; on the failure path it preserves the only handle to a -// resumable conversation. Rows with no resume_token are deleted as before. +// a resume_token, or still claims a running execution, is REPAIRED in place +// (status=stopped, local_pid cleared) rather than deleted. This keeps a +// non-terminal session visible while a stale runtime is being replaced and +// preserves resumable conversations if a subsequent relaunch fails. On the +// happy path the relaunch UPSERTs a fresh row over the repaired one, so +// repairing costs nothing. Rows that are already stopped and have no token are +// deleted as before. // -// Deliberate deviation from models.RowMustBePreserved, which also preserves -// tokenless rows backing a non-terminal session: this path gates on the token -// alone because the token IS the resumable agent state. A row without one means -// the agent never established (or never reported) an ACP session — there is no -// agent-side context a preserved row could resume, and Kandev's own chat -// history lives in the task tables, untouched by this delete. Preserving a -// tokenless row here would keep only incidental metadata that the relaunch -// upsert rebuilds anyway, at the cost of wiring session-state reads into the -// lifecycle tier. Orchestrator-side reconciliation, which already knows session -// state, applies the full invariant via pruneOrRepairExecutorRow. +// This remains a narrower rule than models.RowMustBePreserved because the +// lifecycle tier does not own task-session state. The running status is the +// local signal that a non-terminal session may still need the row; the +// orchestrator applies the full task-state invariant during reconciliation. // // Best-effort: a failure here is logged but doesn't propagate. func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string) { @@ -344,7 +340,8 @@ func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string) { if reader, ok := m.runningWriter.(executorRunningReader); ok { existing, err := reader.GetExecutorRunningBySessionID(ctx, sessionID) switch { - case err == nil && existing != nil && existing.ResumeToken != "": + case err == nil && existing != nil && + (existing.ResumeToken != "" || existing.Status == models.ExecutorRunningStatusRunning): if repairErr := m.runningWriter.RepairExecutorRunningDead(ctx, sessionID); repairErr != nil && !errors.Is(repairErr, models.ErrExecutorRunningNotFound) { m.logger.Warn("failed to repair resumable executors_running row on cleanup; leaving row intact", diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index bc107632aa..d9d0487ce0 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/apps/backend/internal/backendapp/adapters.go @@ -563,6 +563,12 @@ func (a *lifecycleAdapter) IsAgentRunningForSession(ctx context.Context, session return a.mgr.IsAgentRunningForSession(ctx, sessionID) } +// ProbeAgentRunningForSession preserves probe errors so orchestrator cleanup +// can distinguish an unavailable status check from a confirmed dead agent. +func (a *lifecycleAdapter) ProbeAgentRunningForSession(ctx context.Context, sessionID string) (bool, error) { + return a.mgr.ProbeAgentRunningForSession(ctx, sessionID) +} + // IsAgentReadyForPrompt checks if the session can accept a prompt immediately. func (a *lifecycleAdapter) IsAgentReadyForPrompt(ctx context.Context, sessionID string) bool { return a.mgr.IsAgentReadyForPrompt(ctx, sessionID) diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index fb2971c529..9e7d125ad3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -1242,7 +1242,16 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A zap.Bool("workflow_transitioned", transitioned)) if !transitioned && session.State != models.TaskSessionStateWaitingForInput { - s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session) + // Terminal-receipt path. Only flip the session to WAITING_FOR_INPUT + // when the most recent agent-authored message actually asked the + // user for input. Sibling sessions (root task, ParentID empty) + // keep the original affordance so a finishing session on a + // multi-session task still flips to WAITING — only subtasks + // (ParentID non-empty) get the guard. setSessionWaitingForInputIfRequested + // itself inspects the task row to pick the path; for sibling + // sessions the call is a no-op pass-through to the unconditional + // helper, preserving the pre-fix behavior. + s.setSessionWaitingForInputIfRequested(ctx, data.TaskID, data.SessionID, session) } // Capture a git status snapshot before cleanup so it can be served @@ -1255,6 +1264,22 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A // Finalize the automation run: mark status=succeeded so the automation's // concurrency slot is released. The worktree stays. s.finalizeAutomationRun(ctx, data.TaskID, true, "") + + // Settle point: turn has been completed and the session state has been + // reconciled. reclaimIdleSession is the synchronous equivalent of the + // (intentionally-not-built) runtime auto-convergence tick. It only + // proceeds when no live agent process and no active turn are observed, + // so the typical WAITING_FOR_INPUT case (live agent waiting for the + // user) is a no-op pass-through. Subtask terminals that already + // collapsed to COMPLETED inside setSessionWaitingForInputIfRequested + // reclaimed earlier; this call covers sibling/office flows whose + // settled shape has no live runtime. + if err := s.reclaimIdleSession(context.WithoutCancel(ctx), data.SessionID); err != nil { + s.logger.Warn("agent.completed settle: reclaim failed; row preserved", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.Error(err)) + } } // handleAgentFailed handles agent failure events diff --git a/apps/backend/internal/orchestrator/event_handlers_reclaim_wiring_test.go b/apps/backend/internal/orchestrator/event_handlers_reclaim_wiring_test.go new file mode 100644 index 0000000000..24361b6a32 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_reclaim_wiring_test.go @@ -0,0 +1,420 @@ +package orchestrator + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// reclaimTrackingAgentManager is a thin wrapper around mockAgentManager that +// records every CleanupStaleExecutionBySessionID invocation. Tests use the +// recorded slice (and the per-call channel) to assert end-to-end that the +// reclaim primitive fired on the right settle points without resorting to +// time.Sleep — channel-based synchronization, synctest-style. +type reclaimTrackingAgentManager struct { + *mockAgentManager + mu sync.Mutex + calls []string + callCh chan string + closeOnce sync.Once +} + +func newReclaimTrackingAgentManager(inner *mockAgentManager) *reclaimTrackingAgentManager { + return &reclaimTrackingAgentManager{ + mockAgentManager: inner, + callCh: make(chan string, 8), + } +} + +func (m *reclaimTrackingAgentManager) CleanupStaleExecutionBySessionID(_ context.Context, sessionID string) error { + m.mu.Lock() + m.calls = append(m.calls, sessionID) + m.mu.Unlock() + m.closeOnce.Do(func() { close(m.callCh) }) + // signal: another call arrives after the channel is closed — no-op. + return nil +} + +func (m *reclaimTrackingAgentManager) callCount() int { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.calls) +} + +func (m *reclaimTrackingAgentManager) wasCalledFor(sessionID string) bool { + m.mu.Lock() + defer m.mu.Unlock() + for _, sid := range m.calls { + if sid == sessionID { + return true + } + } + return false +} + +func (m *reclaimTrackingAgentManager) callsSnapshot() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.calls...) +} + +// TestSubtaskTerminalCollapse_ReclaimsProviderRuntime is the canonical +// end-to-end wiring test: a child task whose last agent message did not +// request input collapses to COMPLETED inside +// setSessionWaitingForInputIfRequested. With no live agent process and no +// active turn, reclaimIdleSession must fire on this synchronous settle +// point, the executor row must flip to status=stopped with LocalPID=0, +// and the resume_token/worktree_path must remain intact. +func TestSubtaskTerminalCollapse_ReclaimsProviderRuntime(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-task", "s-child", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-task", WorkspaceID: "ws1", Title: "child", + State: v1.TaskStateCompleted, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-task ParentID: %v", err) + } + const token = "rt-keep-child" + const worktree = "/worktrees/child" + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "s-child", + SessionID: "s-child", + TaskID: "child-task", + AgentExecutionID: "exec-child", + ResumeToken: token, + WorktreePath: worktree, + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + LocalPID: 5151, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert child row: %v", err) + } + + taskRepo := newMockTaskRepo() + inner := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: false, // agent process exited before completion event + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessDead + }, + } + agentMgr := newReclaimTrackingAgentManager(inner) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.turnService = &inactiveTurnService{} + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-task", "s-child", "exec-child")) + waitForStopCall(t, inner) + + // 1. Session collapsed to COMPLETED (existing guard behavior). + updated, err := repo.GetTaskSession(ctx, "s-child") + if err != nil { + t.Fatalf("load session: %v", err) + } + if updated.State != models.TaskSessionStateCompleted { + t.Fatalf("subtask terminal must collapse to COMPLETED, got %q", updated.State) + } + + // 2. reclaim fired exactly once for this session. + if !agentMgr.wasCalledFor("s-child") { + t.Fatalf("reclaim must fire on subtask terminal settle; calls=%v", agentMgr.callsSnapshot()) + } + + // 3. Row preserved (resume_token/worktree intact) with status=stopped. + row, err := repo.GetExecutorRunningBySessionID(ctx, "s-child") + if err != nil { + t.Fatalf("row missing after reclaim: %v", err) + } + if row.Status != models.ExecutorRunningStatusStopped { + t.Fatalf("Status = %q, want %q", row.Status, models.ExecutorRunningStatusStopped) + } + if row.LocalPID != 0 { + t.Fatalf("LocalPID = %d, want 0", row.LocalPID) + } + if row.ResumeToken != token { + t.Fatalf("ResumeToken lost: got %q, want %q", row.ResumeToken, token) + } + if row.WorktreePath != worktree { + t.Fatalf("WorktreePath lost: got %q, want %q", row.WorktreePath, worktree) + } +} + +// TestSubtaskTerminalCollapse_LiveAgentBlocksReclaim pins the fail-closed +// guard at the production call site: a subtask whose agent process is +// still alive must NOT be reclaimed even when the session collapses to +// COMPLETED. The provider runtime stays up; the row keeps its status and +// LocalPID for the subsequent cleanup pass. +func TestSubtaskTerminalCollapse_LiveAgentBlocksReclaim(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-live", "s-child-live", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-live", WorkspaceID: "ws1", Title: "child live", + State: v1.TaskStateCompleted, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-live ParentID: %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "s-child-live", SessionID: "s-child-live", TaskID: "child-live", + AgentExecutionID: "exec-live", ResumeToken: "rt-live", + Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, + LocalPID: 8181, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert: %v", err) + } + + taskRepo := newMockTaskRepo() + inner := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, // critical: agent is still alive + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessAlive + }, + } + agentMgr := newReclaimTrackingAgentManager(inner) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.turnService = &inactiveTurnService{} + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-live", "s-child-live", "exec-live")) + waitForStopCall(t, inner) + + if agentMgr.wasCalledFor("s-child-live") { + t.Fatalf("reclaim must be blocked while the agent is alive; calls=%v", agentMgr.callsSnapshot()) + } + row, err := repo.GetExecutorRunningBySessionID(ctx, "s-child-live") + if err != nil { + t.Fatalf("row missing: %v", err) + } + if row.Status != models.ExecutorRunningStatusRunning { + t.Fatalf("live agent row must stay running, got status=%q", row.Status) + } + if row.LocalPID != 8181 { + t.Fatalf("live agent row must keep its LocalPID, got %d", row.LocalPID) + } +} + +// TestSiblingSession_AgentCompletedDoesNotReclaim pins the second half of +// the safety case: a root-task session (ParentID empty) that finishes on +// agent.completed keeps its WAITING_FOR_INPUT affordance and the live +// agent that waits for the user is NOT reclaimed. This is the +// "sibling session / multi-session task" case the parent's guard +// specifically preserves. +func TestSiblingSession_AgentCompletedDoesNotReclaim(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + // Root task (no ParentID) with two sessions: one finishing, one + // still running. The finishing session's agent exit has no + // requests_input and the agent stays up waiting for input from + // the user — the standard WAITING_FOR_INPUT case. + seedSession(t, repo, "t1", "s-finishing", "") + seedExecutorRunning(t, repo, "s-finishing", "t1", "exec-finishing") + requireNoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "s-running", TaskID: "t1", + State: models.TaskSessionStateRunning, + StartedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), + })) + + taskRepo := newMockTaskRepo() + inner := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, // finishing session's agent stays up + } + agentMgr := newReclaimTrackingAgentManager(inner) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.turnService = &inactiveTurnService{} + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("t1", "s-finishing", "exec-finishing")) + waitForStopCall(t, inner) + + // Session still WAITING_FOR_INPUT — root-task affordance preserved. + updated, err := repo.GetTaskSession(ctx, "s-finishing") + if err != nil { + t.Fatalf("load session: %v", err) + } + if updated.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("sibling must stay WAITING_FOR_INPUT, got %q", updated.State) + } + // Reclaim must NOT fire — the live agent is still answering. + if agentMgr.wasCalledFor("s-finishing") { + t.Fatalf("reclaim must not fire while a live agent is waiting for user input; calls=%v", agentMgr.callsSnapshot()) + } +} + +// TestSubtaskTerminalCollapse_ReclaimIsIdempotent pins the wiring +// contract for repeated settle events. A subtask terminal is wired +// twice (subtask collapse inside setSessionWaitingForInputIfRequested + +// the post-handleAgentCompleted settle point), so a single +// agent.completed already produces two reclaim calls — both are +// idempotent at the runtime layer +// ("Safe to call even if the process is already stopped — StopInstance +// is idempotent. Returns nil if no execution exists for the session"). +// The test pins the row invariants that must hold identically across +// repeated calls: the executor row is preserved (resume_token intact, +// status=stopped, LocalPID=0) regardless of how many times the settle +// event fires. +func TestSubtaskTerminalCollapse_ReclaimIsIdempotent(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-idem", "s-idem", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-idem", WorkspaceID: "ws1", Title: "child idem", + State: v1.TaskStateCompleted, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set ParentID: %v", err) + } + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "s-idem", SessionID: "s-idem", TaskID: "child-idem", + AgentExecutionID: "exec-idem", ResumeToken: "rt-idem", + Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, + LocalPID: 2222, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert: %v", err) + } + + taskRepo := newMockTaskRepo() + inner := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: false, + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessDead + }, + } + agentMgr := newReclaimTrackingAgentManager(inner) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.turnService = &inactiveTurnService{} + + // First agent.completed — fires reclaim at least once. The exact + // call count is an implementation detail (the state CAS in + // setSessionWaitingForInputIfRequested short-circuits on the second + // pass, so the post-settle-point wiring is what guarantees the + // invariant on duplicates). + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-idem", "s-idem", "exec-idem")) + waitForStopCall(t, inner) + if agentMgr.callCount() < 1 { + t.Fatalf("first completion must reclaim at least once; calls=%v", agentMgr.callsSnapshot()) + } + assertReclaimedRowInvariants(t, ctx, repo, "s-idem", "rt-idem") + + // A second agent.completed (e.g. a duplicate lifecycle event) must + // preserve the same shape: row invariants unchanged, no PID + // resurrection, no resume_token loss. + prevCalls := agentMgr.callCount() + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-idem", "s-idem", "exec-idem")) + waitForStopCall(t, inner) + if agentMgr.callCount() < prevCalls { + t.Fatalf("second completion dropped reclaim calls: %d -> %d", prevCalls, agentMgr.callCount()) + } + assertReclaimedRowInvariants(t, ctx, repo, "s-idem", "rt-idem") +} + +// assertReclaimedRowInvariants pins the post-reclaim row shape the rest +// of the suite relies on: status=stopped, LocalPID=0, resume_token intact, +// row still present. Channel-based synchronization is unnecessary — +// handleAgentCompleted is synchronous and reclaim runs inline, so by the +// time the call returns the row has settled. +func assertReclaimedRowInvariants(t *testing.T, ctx context.Context, repo executorRowGetter, sessionID, expectedToken string) { + t.Helper() + row, err := repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil { + t.Fatalf("row missing after reclaim: %v", err) + } + if row.Status != models.ExecutorRunningStatusStopped { + t.Fatalf("Status = %q, want stopped", row.Status) + } + if row.LocalPID != 0 { + t.Fatalf("LocalPID = %d, want 0", row.LocalPID) + } + if row.ResumeToken != expectedToken { + t.Fatalf("ResumeToken lost: got %q, want %q", row.ResumeToken, expectedToken) + } +} + +// executorRowGetter is the read-only surface assertReclaimedRowInvariants +// needs. The concrete *sqliterepo.Repository satisfies this; declaring it +// here keeps the helper self-contained without a heavyweight import. +type executorRowGetter interface { + GetExecutorRunningBySessionID(ctx context.Context, sessionID string) (*models.ExecutorRunning, error) +} + +// TestReclaimPreservesRowForResumeTaskSession locks the compatibility +// contract the reviewer asked for: after reclaim runs, the executors_running +// row is still present (status=stopped, resume_token intact), which is the +// pre-condition ResumeTaskSession relies on to re-attach a session. Resume +// must observe the preserved row, not crash on ErrExecutorRunningNotFound. +func TestReclaimPreservesRowForResumeTaskSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "task-resume", "s-resume", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "task-resume", WorkspaceID: "ws1", Title: "resume compat", + State: v1.TaskStateFailed, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set ParentID: %v", err) + } + const token = "rt-resume-keep" + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "s-resume", SessionID: "s-resume", TaskID: "task-resume", + AgentExecutionID: "exec-resume", ResumeToken: token, + Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, + LocalPID: 3333, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert: %v", err) + } + + taskRepo := newMockTaskRepo() + inner := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: false, + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessDead + }, + } + agentMgr := newReclaimTrackingAgentManager(inner) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.turnService = &inactiveTurnService{} + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("task-resume", "s-resume", "exec-resume")) + waitForStopCall(t, inner) + + // ResumeTaskSession requires the executor row to exist; if reclaim + // had pruned it (as a non-preserved terminal row would be) the call + // would return ErrExecutorRunningNotFound. The row is still here + // because resume_token is non-empty (RowMustBePreserved=true), and + // reclaim routes through repairDeadRowLiveness which preserves it. + row, err := repo.GetExecutorRunningBySessionID(ctx, "s-resume") + if err != nil { + t.Fatalf("reclaim must preserve the row for resume: %v", err) + } + if row.Status != models.ExecutorRunningStatusStopped { + t.Fatalf("Status = %q, want %q", row.Status, models.ExecutorRunningStatusStopped) + } + if row.ResumeToken != token { + t.Fatalf("ResumeToken lost during reclaim: got %q, want %q", row.ResumeToken, token) + } + // Confirm the row is the only artifact a resume would see — no + // ErrExecutorRunningNotFound would be returned by the resume path. + if _, err := repo.GetExecutorRunningBySessionID(ctx, "s-resume"); err != nil { + t.Fatalf("ResumeTaskSession precondition (row present) violated: %v", err) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index c3b80d676d..b3bfa3958e 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -1872,6 +1872,80 @@ func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, session s.writeTaskReviewState(ctx, taskID, sessionID) } +// Child terminal receipts use the task's parent relationship and durable +// clarification projection to decide whether a session can collapse to +// COMPLETED. Root-task sibling sessions keep the original WAITING affordance. +// setSessionWaitingForInputIfRequested is the terminal-receipt variant +// of setSessionWaitingForInput. It only applies to subtasks (task.ParentID +// non-empty) because the symptom it guards — "child WAITING_FOR_INPUT +// after the agent's last turn had no active clarification" — only +// arises for child tasks whose task or workflow step is terminal, not to +// surface a UI prompt. Sibling sessions on a root task keep the original +// affordance so a finishing session on a multi-session task still flips to +// WAITING_FOR_INPUT. +// +// When a terminal task has no input request, collapse the session to +// COMPLETED so the child row does not leak in a stuck active state. A clean +// turn on a non-terminal child is not enough evidence for that collapse, so +// it remains WAITING_FOR_INPUT. +func (s *Service) setSessionWaitingForInputIfRequested( + ctx context.Context, + taskID, sessionID string, + preloadedSession ...*models.TaskSession, +) { + task, taskErr := s.repo.GetTask(ctx, taskID) + if taskErr != nil || task == nil { + // A task lookup failure is not evidence that a child reached a terminal + // state. Preserve the promptable session instead of leaving it RUNNING. + s.logger.Warn("failed to load task before terminal receipt; preserving WAITING state", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(taskErr)) + s.setSessionWaitingForInput(ctx, taskID, sessionID, preloadedSession...) + return + } + if task.ParentID == "" { + s.setSessionWaitingForInput(ctx, taskID, sessionID, preloadedSession...) + return + } + activeClarifications, err := s.repo.FindActiveClarificationMessagesBySessionID(ctx, sessionID) + if err != nil { + // A transient reader error is not evidence that the child completed. + // Preserve the promptable state and avoid leaving the session RUNNING. + s.logger.Warn("failed to read active clarifications; preserving WAITING state", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + s.setSessionWaitingForInput(ctx, taskID, sessionID, preloadedSession...) + return + } + taskTerminal := models.IsTerminalTaskState(task.State) || s.workflowStepIsTerminal(ctx, task.WorkflowStepID) + if len(activeClarifications) == 0 && taskTerminal { + s.logger.Debug("subtask terminal: skipping WAITING write; collapsing session to COMPLETED", + zap.String("task_id", taskID), + zap.String("session_id", sessionID)) + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateCompleted, "", false) + // The child has no further use for the provider-runtime reservation. + // reclaimIdleSession is fail-closed (only proceeds when there is no + // live agent and no active turn) and best-effort: a reclaim failure + // logs and returns without affecting the terminal collapse. + if err := s.reclaimIdleSession(ctx, sessionID); err != nil { + s.logger.Warn("subtask terminal: reclaim failed; row preserved", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } + return + } + if len(activeClarifications) == 0 { + s.logger.Debug("subtask turn completed before terminal task state; preserving WAITING state", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("task_state", string(task.State))) + } + s.setSessionWaitingForInput(ctx, taskID, sessionID, preloadedSession...) +} + // taskArchived reports whether a task row has been archived. Runtime-state // writes (IN_PROGRESS on session start, REVIEW on turn completion/cancel/ // startup reconciliation) must never resurrect an archived task's state — @@ -2336,7 +2410,13 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy zap.String("task_id", payload.TaskID), zap.String("session_id", payload.SessionID), zap.String("prev_state", sessionStateString(session))) - s.setSessionWaitingForInput(ctx, payload.TaskID, payload.SessionID, session) + // Terminal-receipt path. Only flip the session to WAITING_FOR_INPUT + // when the most recent agent-authored message actually asked the + // user for input. Sibling sessions (root task, ParentID empty) keep + // the original affordance so a finishing session on a multi-session + // task still flips to WAITING — only subtasks (ParentID non-empty) + // get the guard. + s.setSessionWaitingForInputIfRequested(ctx, payload.TaskID, payload.SessionID, session) } func (s *Service) detachClarificationWaiters(ctx context.Context, sessionID string) { diff --git a/apps/backend/internal/orchestrator/event_handlers_subtask_waiting_test.go b/apps/backend/internal/orchestrator/event_handlers_subtask_waiting_test.go new file mode 100644 index 0000000000..1f2a2c31f1 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_subtask_waiting_test.go @@ -0,0 +1,239 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestHandleAgentCompleted_SubtaskWithoutRequestsInputCollapsesToCompleted +// pins the symptom reported in the v0.88 total-control fix plan: +// "child in a terminal state without requests_input must NOT write +// WAITING_FOR_INPUT". handleAgentCompleted's existing line 1238 writes +// WAITING unconditionally for non-transitioned terminal receipts. +// setSessionWaitingForInputIfRequested (this commit's guard) refuses +// the WAITING write for child tasks and instead collapses the session +// to COMPLETED, so a child agent's clean exit does not leave the +// session in a stuck RUNNING/RUNNING-equivalent state. +// +// The seed task is a subtask (ParentID="parent") in a terminal task state +// and the agent's last message has requests_input=false; the orchestrator +// must therefore finish the session to COMPLETED, not WAITING. +func TestHandleAgentCompleted_SubtaskWithoutRequestsInputCollapsesToCompleted(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-task", "s-child", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-task", WorkspaceID: "ws1", Title: "child", + State: v1.TaskStateCompleted, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-task ParentID: %v", err) + } + seedExecutorRunning(t, repo, "s-child", "child-task", "exec-child") + + // No messages: requests_input must be false, so the guard refuses + // the WAITING write and collapses the session to COMPLETED. + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-task", "s-child", "exec-child")) + waitForStopCall(t, agentMgr) + + updated, err := repo.GetTaskSession(ctx, "s-child") + if err != nil { + t.Fatalf("load session after subtask terminal: %v", err) + } + if updated.State != models.TaskSessionStateCompleted { + t.Fatalf("subtask terminal without requests_input must collapse to COMPLETED, got %q", updated.State) + } +} + +// TestHandleAgentCompleted_NonTerminalSubtaskWithoutRequestsInputWritesWaiting +// protects the lifecycle boundary: a clean agent turn does not prove that a +// child task reached its terminal workflow state. A non-terminal child must +// remain promptable instead of being collapsed to COMPLETED solely because +// its latest message did not request input. +func TestHandleAgentCompleted_NonTerminalSubtaskWithoutRequestsInputWritesWaiting(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-task-open", "s-child-open", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-task-open", WorkspaceID: "ws1", Title: "child open", + State: v1.TaskStateInProgress, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-task-open ParentID: %v", err) + } + seedExecutorRunning(t, repo, "s-child-open", "child-task-open", "exec-child-open") + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-task-open", "s-child-open", "exec-child-open")) + waitForStopCall(t, agentMgr) + + updated, err := repo.GetTaskSession(ctx, "s-child-open") + if err != nil { + t.Fatalf("load session after non-terminal subtask turn: %v", err) + } + if updated.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("non-terminal subtask must remain WAITING_FOR_INPUT, got %q", updated.State) + } +} + +func TestHandleAgentCompleted_SubtaskIgnoresResolvedClarification(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-resolved", "s-child-resolved", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-resolved", WorkspaceID: "ws1", Title: "child resolved", + State: v1.TaskStateCompleted, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-resolved ParentID: %v", err) + } + seedExecutorRunning(t, repo, "s-child-resolved", "child-resolved", "exec-child-resolved") + require.NoError(t, repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-child-resolved", TaskID: "child-resolved", TaskSessionID: "s-child-resolved", + StartedAt: now, UpdatedAt: now, + })) + require.NoError(t, repo.CreateMessage(ctx, &models.Message{ + ID: "m-resolved", TaskSessionID: "s-child-resolved", TaskID: "child-resolved", + TurnID: "turn-child-resolved", AuthorType: models.MessageAuthorAgent, + Content: "The earlier question was answered.", Type: models.MessageTypeClarificationRequest, + RequestsInput: true, Metadata: map[string]interface{}{"status": "answered"}, + CreatedAt: now, UpdatedAt: now, + })) + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-resolved", "s-child-resolved", "exec-child-resolved")) + waitForStopCall(t, agentMgr) + + updated, err := repo.GetTaskSession(ctx, "s-child-resolved") + if err != nil { + t.Fatalf("load session after resolved clarification: %v", err) + } + if updated.State != models.TaskSessionStateCompleted { + t.Fatalf("resolved clarification must not keep a terminal child waiting, got %q", updated.State) + } +} + +// TestHandleAgentCompleted_SubtaskWithRequestsInputStillWritesWaiting +// pins the positive half of the guard: a child task whose agent +// actually asked the user for input (clarification request, +// requests_input=true) MUST still flip the session to WAITING so the +// chat UI surfaces the prompt — the guard is supposed to suppress +// false-positive WAITING writes, not legitimate clarification +// surfaces. +func TestHandleAgentCompleted_SubtaskWithRequestsInputStillWritesWaiting(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + seedSession(t, repo, "child-clarify", "s-child-cl", "") + if err := repo.UpdateTask(ctx, &models.Task{ + ID: "child-clarify", WorkspaceID: "ws1", Title: "child clarify", + State: v1.TaskStateInProgress, ParentID: "parent", + UpdatedAt: now, + }); err != nil { + t.Fatalf("set child-clarify ParentID: %v", err) + } + seedExecutorRunning(t, repo, "s-child-cl", "child-clarify", "exec-child-cl") + + // Agent clarification request — the only path that should still + // surface WAITING for a subtask session. The message row requires a + // turn_id FK; seed a turn first so CreateMessage succeeds. + require.NoError(t, repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-child-cl", TaskID: "child-clarify", TaskSessionID: "s-child-cl", + StartedAt: now, UpdatedAt: now, + })) + require.NoError(t, repo.CreateMessage(ctx, &models.Message{ + ID: "m-clarify", TaskSessionID: "s-child-cl", TaskID: "child-clarify", + TurnID: "turn-child-cl", + AuthorType: models.MessageAuthorAgent, Content: "I need X", + Type: models.MessageTypeClarificationRequest, RequestsInput: true, + CreatedAt: now, UpdatedAt: now, + })) + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("child-clarify", "s-child-cl", "exec-child-cl")) + waitForStopCall(t, agentMgr) + + updated, err := repo.GetTaskSession(ctx, "s-child-cl") + if err != nil { + t.Fatalf("load session after subtask clarification: %v", err) + } + if updated.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("subtask with requests_input=true must still flip to WAITING, got %q", updated.State) + } +} + +// TestHandleAgentCompleted_SiblingSessionWithoutRequestsInputStillWritesWaiting +// pins the negative half of the parent-side guard. A root task with +// multiple sessions has each session finish independently: the finishing +// session must still flip to WAITING (its siblings can keep working), +// even though its last message did not request input. The guard +// therefore matches subtasks only; root tasks keep the original +// affordance. +func TestHandleAgentCompleted_SiblingSessionWithoutRequestsInputStillWritesWaiting(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + // Root task (no ParentID) with two sessions: one finishing, one + // still running. The finishing session's agent exit has no + // requests_input — the guard must NOT collapse it, so the multi- + // session task can carry a per-session WAITING signal. + seedSession(t, repo, "t1", "s-finishing", "") + seedExecutorRunning(t, repo, "s-finishing", "t1", "exec-finishing") + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "s-running", TaskID: "t1", + State: models.TaskSessionStateRunning, + StartedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), + })) + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("t1", "s-finishing", "exec-finishing")) + waitForStopCall(t, agentMgr) + + updated, err := repo.GetTaskSession(ctx, "s-finishing") + if err != nil { + t.Fatalf("load finishing session: %v", err) + } + if updated.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("sibling session on root task must still flip to WAITING, got %q", updated.State) + } +} + +func watcherAgentCompletedData(taskID, sessionID, execID string) watcher.AgentEventData { + return watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: execID, + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_terminal_dispatch_test.go b/apps/backend/internal/orchestrator/event_handlers_terminal_dispatch_test.go new file mode 100644 index 0000000000..1bc9a1d88b --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_terminal_dispatch_test.go @@ -0,0 +1,264 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/queue" + "github.com/kandev/kandev/internal/orchestrator/scheduler" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestHandleTaskStateChanged_FiresBothChildrenCompletedAndDependency pins the +// v0.88 total-control plan's "需求 2: 父总控一次性消费 child terminal +// receipt 并派发既有下一步" guarantee. handleTaskStateChanged is the single +// entry point for terminal task events, and a child terminal receipt must +// not split its work across two phases: the children-completed rollup +// (which transitions the parent workflow) and the dependency resolution +// (which unblocks dependents) must both fire from the same call. +// +// Pre-fix risk: if either path returned early without doing its work, the +// dependent task could be stuck blocked-by-forever, or the parent could +// stay parked on a wait-for-children step. The two paths today are +// sequenced within handleTaskStateChanged (children_completed first, +// dependency dispatch second), each with their own atomicity guard +// (QueueAndInterruptForPeerMessage's cancelInFlight lock, +// deferred_launch_consume's claim+consume gate). The test seeds a parent +// at the children-wait step, a child task whose terminal receipt we fire, +// and a dependent carrying a start-when-unblocked intent whose +// dependency edge points at the child. After the call the parent must +// have moved to its post-children step AND the dependent must have +// launched exactly one session. +func TestHandleTaskStateChanged_FiresBothChildrenCompletedAndDependency(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + // Parent workflow with an OnChildrenCompleted transition on its first + // step so the children-completed rollup has somewhere to go. + const parentID = "parent-1" + const parentSessionID = "parent-session-1" + stepGetter := newMockStepGetter() + stepGetter.steps["step-wait"] = &wfmodels.WorkflowStep{ + ID: "step-wait", + WorkflowID: "wf1", + Name: "Wait for children", + Position: 0, + Events: wfmodels.StepEvents{ + OnChildrenCompleted: []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionMoveToNext}, + }, + }, + } + stepGetter.steps["step-done"] = &wfmodels.WorkflowStep{ + ID: "step-done", + WorkflowID: "wf1", + Name: "Done", + Position: 1, + } + seedSession(t, repo, parentID, parentSessionID, "step-wait") + + // A child that the parent is waiting on, plus a dependent whose + // start-when-unblocked intent is the contract the dependency gate + // must fire on. + const childID = "child-1" + const dependentID = "dependent-1" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: childID, WorkspaceID: "ws1", WorkflowID: "wf1", Title: "Child", + State: v1.TaskStateInProgress, ParentID: parentID, + CreatedAt: now, UpdatedAt: now, + })) + // The orchestrator's allChildrenTerminal predicate reads each child's + // DB state directly, so the child must already be terminal when the + // receipt fires. Without this guard the children_completed path + // silently no-ops because the DB row still says InProgress. + require.NoError(t, repo.UpdateTaskState(ctx, childID, v1.TaskStateCompleted)) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: dependentID, WorkspaceID: "ws1", Title: "Dependent", + State: v1.TaskStateCreated, + CreatedAt: now, UpdatedAt: now, + Metadata: map[string]interface{}{ + models.MetaKeyDeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-1", + "prompt": "child terminal receipt test", + models.DeferredLaunchStartWhenUnblockedKey: true, + }, + }, + })) + + // Spy on the agent launch so the test can assert the dependent's + // start-when-unblocked intent fired exactly once, mirroring + // TestDependencyResolutionLaunchesAnIntactIntent's positive control. + counter := newLaunchCounter() + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + counter.record() + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-dependent-1"}, nil + }, + } + + svc := createEngineService(t, repo, stepGetter, agentMgr) + // createEngineService wires the workflow engine but leaves the + // scheduler nil. The dependency path (handleTaskDependenciesForTerminalState + // → launchDeferredTask → LaunchSession → startTask → scheduler.GetTask) + // dereferences s.scheduler — we must install a scheduler with a + // mockTaskRepo so the dependent resolution does not panic. The + // scheduler is a field the same-package test can write directly. + if mockT, ok := svc.taskRepo.(*mockTaskRepo); ok { + mockT.tasks[dependentID] = &v1.Task{ + ID: dependentID, Title: "Dependent", State: v1.TaskStateCreated, + } + svc.scheduler = scheduler.NewScheduler( + queue.NewTaskQueue(100), + executor.NewExecutor(agentMgr, repo, newTestLogger(), executor.ExecutorConfig{}), + mockT, + newTestLogger(), + scheduler.SchedulerConfig{}, + ) + } + // The child is the only predecessor of the dependent. The gate returns + // unblocked the moment the child reaches a terminal state, so the + // dependency dispatch path in handleTaskStateChanged must fire as + // part of the same call. + svc.SetTaskDependencyReader(&resolvedDependencyReader{dependents: []string{dependentID}}) + + // Fire the child terminal receipt through the same single entry + // point the bus delivers to. + svc.handleTaskStateChanged(ctx, watcher.TaskEventData{ + TaskID: childID, + NewState: ptrToTaskStateCompleted(), + }) + + // children_completed path: the parent must have walked to its + // post-children step in the same call, not the next event. + parentAfter, err := repo.GetTask(ctx, parentID) + require.NoError(t, err) + if parentAfter.WorkflowStepID != "step-done" { + t.Fatalf("children-completed path did not fire: parent.WorkflowStepID = %q, want %q", + parentAfter.WorkflowStepID, "step-done") + } + + // dependency path: the dependent's start-when-unblocked intent must + // have been consumed in the same call, not deferred to the next + // dependency event. A counter record is the precise signal — + // awaiting a deadline would not distinguish "the intent fired during + // this call" from "the intent fired later because a future event + // unblocked it". + if !counter.awaitLaunch(0) { + t.Fatal("dependency dispatch path did not fire in the same call: counter is still at zero") + } + // The intent is single-shot: the same call must not launch a second + // session on a future event because the gate already consumed the + // intent. + depAfter, err := repo.GetTask(ctx, dependentID) + require.NoError(t, err) + if _, still := depAfter.Metadata[models.MetaKeyDeferredLaunch]; still { + t.Fatal("dependency dispatch did not consume the deferred-launch intent") + } + // A second deferred launch (e.g. a redundant dependency event) + // must not spawn a second session replaying the same prompt. + assert.Equal(t, 1, counter.count(), + "the deferred-launch intent must fire exactly once per child terminal receipt") +} + +// TestHandleTaskStateChanged_NonTerminalChildReceiptDoesNotFireParents +// pins the negative half: a child state change that is not terminal +// must be a no-op, even if the parent has a children-completed +// transition waiting. The children-completed path's predicate +// (models.IsTerminalTaskState) and the dependency path's gate +// (unblocked only on a terminal predecessor) both fail closed on +// non-terminal transitions, so this call leaves the parent at its +// current step and the dependent's deferred-launch intent intact. +func TestHandleTaskStateChanged_NonTerminalChildReceiptDoesNotFireParents(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + const parentID = "parent-2" + const parentSessionID = "parent-session-2" + stepGetter := newMockStepGetter() + stepGetter.steps["step-wait"] = &wfmodels.WorkflowStep{ + ID: "step-wait", WorkflowID: "wf-2", Name: "Wait", Position: 0, + Events: wfmodels.StepEvents{ + OnChildrenCompleted: []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionMoveToNext}, + }, + }, + } + stepGetter.steps["step-done"] = &wfmodels.WorkflowStep{ + ID: "step-done", WorkflowID: "wf-2", Name: "Done", Position: 1, + } + seedSession(t, repo, parentID, parentSessionID, "step-wait") + + const childID = "child-2" + const dependentID = "dependent-2" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: childID, WorkspaceID: "ws1", WorkflowID: "wf-2", Title: "Child", + State: v1.TaskStateInProgress, ParentID: parentID, + CreatedAt: now, UpdatedAt: now, + })) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: dependentID, WorkspaceID: "ws1", Title: "Dependent", + State: v1.TaskStateCreated, + CreatedAt: now, UpdatedAt: now, + Metadata: map[string]interface{}{ + models.MetaKeyDeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-1", + "prompt": "child non-terminal test", + models.DeferredLaunchStartWhenUnblockedKey: true, + }, + }, + })) + + counter := newLaunchCounter() + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + counter.record() + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-dependent-2"}, nil + }, + } + + svc := createEngineService(t, repo, stepGetter, agentMgr) + svc.SetTaskDependencyReader(&resolvedDependencyReader{dependents: []string{dependentID}}) + + svc.handleTaskStateChanged(ctx, watcher.TaskEventData{ + TaskID: childID, + NewState: ptrToTaskStateInProgress(), + }) + + parentAfter, err := repo.GetTask(ctx, parentID) + require.NoError(t, err) + if parentAfter.WorkflowStepID != "step-wait" { + t.Fatalf("non-terminal child receipt must not advance parent, got %q", parentAfter.WorkflowStepID) + } + depAfter, err := repo.GetTask(ctx, dependentID) + require.NoError(t, err) + if _, still := depAfter.Metadata[models.MetaKeyDeferredLaunch]; !still { + t.Fatal("non-terminal child receipt must not consume the deferred-launch intent") + } + if counter.count() != 0 { + t.Fatalf("non-terminal child receipt must not launch, count=%d", counter.count()) + } +} + +func ptrToTaskStateCompleted() *v1.TaskState { + s := v1.TaskStateCompleted + return &s +} + +func ptrToTaskStateInProgress() *v1.TaskState { + s := v1.TaskStateInProgress + return &s +} diff --git a/apps/backend/internal/orchestrator/reclaim_lifecycle_test.go b/apps/backend/internal/orchestrator/reclaim_lifecycle_test.go new file mode 100644 index 0000000000..ec445cc4b0 --- /dev/null +++ b/apps/backend/internal/orchestrator/reclaim_lifecycle_test.go @@ -0,0 +1,353 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/task/models" +) + +// inactiveTurnService reports no active turn for any session, satisfying the +// idle-reclaim precondition that no in-flight turn is bound to the session. +// Only GetActiveTurn is exercised by the reclaim primitive; other methods +// panic so a test that grows new behavior fails loudly. +type inactiveTurnService struct{} + +func (*inactiveTurnService) GetActiveTurn(_ context.Context, _ string) (*models.Turn, error) { + return nil, nil +} +func (*inactiveTurnService) StartTurn(context.Context, string) (*models.Turn, error) { + panic("inactiveTurnService: StartTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) ReserveTurn(context.Context, string, *models.PromptDispatchRecovery) (*models.Turn, error) { + panic("inactiveTurnService: ReserveTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) MarkReservedTurnDispatchAttempted(context.Context, *models.Turn) error { + panic("inactiveTurnService: MarkReservedTurnDispatchAttempted should not be called by reclaim tests") +} +func (*inactiveTurnService) PublishReservedTurn(context.Context, *models.Turn) error { + panic("inactiveTurnService: PublishReservedTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) RollbackReservedTurn(context.Context, string, string) (bool, error) { + panic("inactiveTurnService: RollbackReservedTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) ReconcileUnpublishedPromptTurns(context.Context) (int, error) { + return 0, nil +} +func (*inactiveTurnService) CompleteTurn(context.Context, string) error { + panic("inactiveTurnService: CompleteTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) GetTurn(context.Context, string) (*models.Turn, error) { + panic("inactiveTurnService: GetTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) UpdateTurn(context.Context, *models.Turn) error { + panic("inactiveTurnService: UpdateTurn should not be called by reclaim tests") +} +func (*inactiveTurnService) PatchTurnMetadata(context.Context, string, string, map[string]interface{}) error { + panic("inactiveTurnService: PatchTurnMetadata should not be called by reclaim tests") +} +func (*inactiveTurnService) AbandonOpenTurns(context.Context, string) error { + return nil +} + +// TestClassifyIdleReclaimDisposition is the single decision-matrix test for +// the idle-reclaim predicate. Each case names the (state, runtime-live, +// active-turn) tuple the primitive sees and the disposition it must return; +// fail-closed means every uncertain input is a Skipped* disposition. +func TestClassifyIdleReclaimDisposition(t *testing.T) { + tests := []struct { + name string + state models.TaskSessionState + agentRunning bool + hasActiveTurn bool + want idleReclaimDisposition + }{ + { + name: "waiting_for_input, no live runtime, no active turn reclaims", + state: models.TaskSessionStateWaitingForInput, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionReclaimed, + }, + { + name: "idle office state, no live runtime, no active turn reclaims", + state: models.TaskSessionStateIdle, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionReclaimed, + }, + { + name: "running session is never reclaimed", + state: models.TaskSessionStateRunning, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionSkippedState, + }, + { + name: "starting session is never reclaimed", + state: models.TaskSessionStateStarting, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionSkippedState, + }, + { + name: "completed session with no live runtime reclaims", + state: models.TaskSessionStateCompleted, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionReclaimed, + }, + { + name: "failed session is never reclaimed", + state: models.TaskSessionStateFailed, + agentRunning: false, + hasActiveTurn: false, + want: idleReclaimDispositionSkippedState, + }, + { + name: "live runtime blocks reclaim", + state: models.TaskSessionStateWaitingForInput, + agentRunning: true, + hasActiveTurn: false, + want: idleReclaimDispositionSkippedLive, + }, + { + name: "active turn blocks reclaim", + state: models.TaskSessionStateWaitingForInput, + agentRunning: false, + hasActiveTurn: true, + want: idleReclaimDispositionSkippedTurn, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := classifyIdleReclaim(tt.state, tt.agentRunning, tt.hasActiveTurn) + if got != tt.want { + t.Fatalf("disposition = %q, want %q", got, tt.want) + } + }) + } +} + +// TestReclaimIdleSessionReleasesRuntimeAndPreservesRow asserts the happy +// path: an idle-yet-resumable session with no live runtime and no active +// turn gets its provider runtime released and its executors_running row +// repaired (status=stopped, local_pid=0) while the resume_token and +// worktree_path are preserved verbatim. +func TestReclaimIdleSessionReleasesRuntimeAndPreservesRow(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + seedTaskAndSession(t, repo, "taskIdle", "sessionIdle", models.TaskSessionStateWaitingForInput) + const token = "resume-token-keep" + const worktree = "/worktrees/keep" + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "sessionIdle", + SessionID: "sessionIdle", + TaskID: "taskIdle", + AgentExecutionID: "exec-idle", + ResumeToken: token, + WorktreePath: worktree, + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + LocalPID: 4242, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert idle row: %v", err) + } + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{ + isAgentRunning: false, + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessDead + }, + }) + svc.turnService = &inactiveTurnService{} + + if err := svc.reclaimIdleSession(ctx, "sessionIdle"); err != nil { + t.Fatalf("reclaimIdleSession: %v", err) + } + + row, err := repo.GetExecutorRunningBySessionID(ctx, "sessionIdle") + if err != nil { + t.Fatalf("row missing after reclaim: %v", err) + } + if row.Status != models.ExecutorRunningStatusStopped { + t.Fatalf("status = %q, want %q", row.Status, models.ExecutorRunningStatusStopped) + } + if row.LocalPID != 0 { + t.Fatalf("LocalPID = %d, want 0", row.LocalPID) + } + if row.ResumeToken != token { + t.Fatalf("ResumeToken lost during reclaim: got %q, want %q", row.ResumeToken, token) + } + if row.WorktreePath != worktree { + t.Fatalf("WorktreePath lost during reclaim: got %q, want %q", row.WorktreePath, worktree) + } +} + +// TestReclaimIdleSessionRefusesLiveRuntime proves the fail-closed guard: +// a session whose IsAgentRunningForSession returns true is never touched +// even when the session is in an idle state. The row keeps its running +// status and LocalPID; nothing is killed. +func TestReclaimIdleSessionRefusesLiveRuntime(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + seedTaskAndSession(t, repo, "taskLive", "sessionLive", models.TaskSessionStateWaitingForInput) + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "sessionLive", SessionID: "sessionLive", TaskID: "taskLive", + AgentExecutionID: "exec-live", ResumeToken: "rt-keep", + Runtime: agentruntime.RuntimeStandalone, Status: models.ExecutorRunningStatusRunning, + LocalPID: 7777, CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert live row: %v", err) + } + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{ + isAgentRunning: true, // critical: the agent is still alive + rowLivenessFn: func(*models.ExecutorRunning) models.ProcessLiveness { + return models.ProcessLivenessAlive + }, + }) + svc.turnService = &inactiveTurnService{} + + if err := svc.reclaimIdleSession(ctx, "sessionLive"); err != nil { + t.Fatalf("reclaimIdleSession: %v", err) + } + + row, err := repo.GetExecutorRunningBySessionID(ctx, "sessionLive") + if err != nil { + t.Fatalf("row missing after reclaim: %v", err) + } + if row.Status != models.ExecutorRunningStatusRunning { + t.Fatalf("live session must not be stopped; status = %q", row.Status) + } + if row.LocalPID != 7777 { + t.Fatalf("live session must not lose its PID; LocalPID = %d", row.LocalPID) + } +} + +type failingAgentLivenessProbe struct { + *mockAgentManager +} + +func (m *failingAgentLivenessProbe) ProbeAgentRunningForSession(context.Context, string) (bool, error) { + return false, errors.New("agent status unavailable") +} + +func TestReclaimIdleSessionFailsClosedOnLivenessProbeError(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + seedTaskAndSession(t, repo, "task-probe", "session-probe", models.TaskSessionStateWaitingForInput) + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "session-probe", SessionID: "session-probe", TaskID: "task-probe", + AgentExecutionID: "exec-probe", Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, LocalPID: 99, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert probe row: %v", err) + } + + base := &mockAgentManager{isAgentRunning: false} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &failingAgentLivenessProbe{mockAgentManager: base}) + svc.turnService = &inactiveTurnService{} + + if err := svc.reclaimIdleSession(ctx, "session-probe"); err != nil { + t.Fatalf("reclaimIdleSession: %v", err) + } + row, err := repo.GetExecutorRunningBySessionID(ctx, "session-probe") + if err != nil { + t.Fatalf("row missing after uncertain probe: %v", err) + } + if row.Status != models.ExecutorRunningStatusRunning || row.LocalPID != 99 { + t.Fatalf("uncertain liveness must leave row untouched, got status=%q pid=%d", row.Status, row.LocalPID) + } +} + +// TestReclaimIdleSessionRefusesWrongState proves non-idle non-terminal +// states are never reclaimed: a RUNNING or STARTING session that happens +// to have no live runtime must wait for explicit completion, not get +// reaped by idle reclaim. Failed and Cancelled have dedicated cancellation +// cleanup paths (handleTerminalSessionOnStartup, the cancel pipelines) +// and are deliberately excluded from the reclaim predicate. +func TestReclaimIdleSessionRefusesWrongState(t *testing.T) { + tests := []struct { + name string + state models.TaskSessionState + }{ + {name: "running", state: models.TaskSessionStateRunning}, + {name: "starting", state: models.TaskSessionStateStarting}, + {name: "failed", state: models.TaskSessionStateFailed}, + {name: "cancelled", state: models.TaskSessionStateCancelled}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + sessionID := "s-" + tt.name + seedTaskAndSession(t, repo, "task-"+tt.name, sessionID, tt.state) + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: sessionID, SessionID: sessionID, TaskID: "task-" + tt.name, + AgentExecutionID: "exec-" + tt.name, + Runtime: agentruntime.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + LocalPID: 1234, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("upsert: %v", err) + } + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{ + isAgentRunning: false, + }) + svc.turnService = &inactiveTurnService{} + + if err := svc.reclaimIdleSession(ctx, sessionID); err != nil && + !errors.Is(err, context.Canceled) { + t.Fatalf("reclaimIdleSession: %v", err) + } + + row, err := repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil { + t.Fatalf("row missing after reclaim: %v", err) + } + if row.Status != models.ExecutorRunningStatusRunning { + t.Fatalf("%s session must not be reclaimed; status = %q", tt.state, row.Status) + } + }) + } +} + +// TestReclaimIdleSessionMissingRowIsNoOp ensures the primitive is +// idempotent and safe to call when the executor row has already been +// pruned (e.g. by startup reconciliation). +func TestReclaimIdleSessionMissingRowIsNoOp(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + seedTaskAndSession(t, repo, "taskGone", "sessionGone", models.TaskSessionStateWaitingForInput) + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{ + isAgentRunning: false, + }) + svc.turnService = &inactiveTurnService{} + + if err := svc.reclaimIdleSession(ctx, "sessionGone"); err != nil { + t.Fatalf("reclaimIdleSession with missing row: %v", err) + } +} + +// TestReclaimIdleSessionEmptySessionID is the trivial early return: an +// empty session ID never touches the database or the runtime. +func TestReclaimIdleSessionEmptySessionID(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + if err := svc.reclaimIdleSession(context.Background(), ""); err != nil { + t.Fatalf("reclaimIdleSession with empty id: %v", err) + } +} diff --git a/apps/backend/internal/orchestrator/reconcile_liveness.go b/apps/backend/internal/orchestrator/reconcile_liveness.go index c9295a0cec..149bad43b1 100644 --- a/apps/backend/internal/orchestrator/reconcile_liveness.go +++ b/apps/backend/internal/orchestrator/reconcile_liveness.go @@ -24,6 +24,13 @@ type rowLivenessProber interface { RowLiveness(row *models.ExecutorRunning) models.ProcessLiveness } +// agentRunningProber preserves runtime probe errors so reclaim can fail +// closed. The legacy boolean method remains the fallback for test doubles and +// adapters that do not expose the richer probe yet. +type agentRunningProber interface { + ProbeAgentRunningForSession(ctx context.Context, sessionID string) (bool, error) +} + type startupCleanupDisposition string const ( @@ -166,11 +173,153 @@ func (s *Service) pruneOrRepairExecutorRow(ctx context.Context, running *models. // process — status=stopped, local_pid cleared, resume_token/worktree preserved — // satisfying #1597's "never leave a row claiming a dead process" expected // behavior. Best-effort; a missing row is not an error. -func (s *Service) repairDeadRowLiveness(ctx context.Context, running *models.ExecutorRunning) { - if err := s.repo.RepairExecutorRunningDead(ctx, running.SessionID); err != nil && - !errors.Is(err, models.ErrExecutorRunningNotFound) { +func (s *Service) repairDeadRowLiveness(ctx context.Context, running *models.ExecutorRunning) error { + err := s.repo.RepairExecutorRunningDead(ctx, running.SessionID) + if err != nil && !errors.Is(err, models.ErrExecutorRunningNotFound) { s.logger.Warn("failed to repair executor row liveness during reconciliation", zap.String("session_id", running.SessionID), zap.Error(err)) } + if errors.Is(err, models.ErrExecutorRunningNotFound) { + return nil + } + return err +} + +func (s *Service) probeAgentRunning(ctx context.Context, sessionID string) (bool, error) { + if prober, ok := s.agentManager.(agentRunningProber); ok && prober != nil { + return prober.ProbeAgentRunningForSession(ctx, sessionID) + } + return s.agentManager.IsAgentRunningForSession(ctx, sessionID), nil +} + +// idleReclaimDisposition captures why a session is or is not eligible for the +// idle-session reclaim path. Mirrors startupCleanupDisposition's role: keep +// every decision logged and observable without leaking resume tokens. +type idleReclaimDisposition string + +const ( + idleReclaimDispositionReclaimed idleReclaimDisposition = "reclaimed" + idleReclaimDispositionSkippedState idleReclaimDisposition = "skipped_state" + idleReclaimDispositionSkippedLive idleReclaimDisposition = "skipped_live_runtime" + idleReclaimDispositionSkippedTurn idleReclaimDisposition = "skipped_active_turn" +) + +// classifyIdleReclaim is the single decision matrix for the idle-session +// reclaim primitive. Reclaim is fail-closed: only sessions in an idle-yet- +// resumable OR terminal shape AND without a live runtime AND without an +// active turn proceed. Any uncertain signal falls into a skipped +// disposition and the caller leaves the row untouched. +// +// Why include Completed in the allowed set: the subtask terminal collapse +// in setSessionWaitingForInputIfRequested writes session.state=Completed +// when the agent's last turn did not request input, and the parent task +// has no further use for the provider-runtime reservation. Failed and +// Cancelled are deliberately excluded — those have separate cancellation +// cleanup paths (handleTerminalSessionOnStartup and the cancel pipelines) +// that already reconcile executor rows. +func classifyIdleReclaim(sessionState models.TaskSessionState, agentRunning bool, hasActiveTurn bool) idleReclaimDisposition { + switch sessionState { + case models.TaskSessionStateWaitingForInput, + models.TaskSessionStateIdle, + models.TaskSessionStateCompleted: + default: + return idleReclaimDispositionSkippedState + } + if agentRunning { + return idleReclaimDispositionSkippedLive + } + if hasActiveTurn { + return idleReclaimDispositionSkippedTurn + } + return idleReclaimDispositionReclaimed +} + +// reclaimIdleSession releases the provider-runtime reservation backing an +// idle-yet-resumable or terminal session (state in {WaitingForInput, Idle, +// Completed}) when no live agent process and no active turn can be +// observed. Resume token and worktree path are preserved so a later resume +// can re-attach the session without losing context. Never touches a +// running executor; never called from a periodic reaper (a runtime +// auto-convergence tick is a pre-existing design gap handled by an +// independent task, see plan docs). +// +// Synchronous call sites so far: +// - setSessionWaitingForInputIfRequested: subtask terminal collapse to +// Completed when the last agent message did not request input. The +// child has no further use for the provider runtime. +// - handleAgentCompleted: after completeTurnForSession, when the +// session has settled into a non-active terminal/idle state. +// +// Idempotent: a session with no in-memory execution entry returns nil from +// CleanupStaleExecutionBySessionID, and a missing executor row is not an +// error for repairDeadRowLiveness. +func (s *Service) reclaimIdleSession(ctx context.Context, sessionID string) error { + if sessionID == "" { + return nil + } + releaseLifecycleLock := s.acquireSessionLifecycleLock(sessionID) + defer releaseLifecycleLock() + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + if isTaskSessionNotFound(err) { + return nil + } + return fmt.Errorf("load session for idle reclaim: %w", err) + } + running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil { + // A missing executor row means there is nothing to release; skip. + if errors.Is(err, models.ErrExecutorRunningNotFound) { + return nil + } + return fmt.Errorf("load executor row for idle reclaim: %w", err) + } + agentRunning, probeErr := s.probeAgentRunning(ctx, sessionID) + if probeErr != nil { + s.logger.Warn("idle reclaim skipped; agent liveness probe failed", + zap.String("session_id", sessionID), + zap.Error(probeErr)) + return nil + } + hasActiveTurn := s.sessionHasActiveTurn(ctx, sessionID) + decision := classifyIdleReclaim(session.State, agentRunning, hasActiveTurn) + if decision != idleReclaimDispositionReclaimed { + s.logger.Debug("idle reclaim skipped", + zap.String("session_id", sessionID), + zap.String("disposition", string(decision)), + zap.String("session_state", string(session.State)), + zap.Bool("agent_running", agentRunning), + zap.Bool("has_active_turn", hasActiveTurn)) + return nil + } + if err := s.agentManager.CleanupStaleExecutionBySessionID(ctx, sessionID); err != nil { + s.logger.Warn("idle reclaim: cleanup failed; row preserved", + zap.String("session_id", sessionID), + zap.Error(err)) + return nil + } + if err := s.repairDeadRowLiveness(ctx, running); err != nil { + return fmt.Errorf("repair executor row after idle reclaim: %w", err) + } + s.logger.Info("idle reclaim: provider runtime released; row preserved for resume", + zap.String("session_id", sessionID), + zap.String("task_id", running.TaskID), + zap.Bool("has_resume_token", running.ResumeToken != ""), + zap.Bool("has_worktree", running.WorktreePath != "")) + return nil +} + +// sessionHasActiveTurn reports whether the orchestrator has any in-flight +// turn bound to the session. Conservative default (true on lookup error) +// keeps reclaim fail-closed. +func (s *Service) sessionHasActiveTurn(ctx context.Context, sessionID string) bool { + if s.turnService == nil { + return true + } + turn, err := s.turnService.GetActiveTurn(ctx, sessionID) + if err != nil { + return true + } + return turn != nil } diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 2109076651..6f77b940d6 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -507,9 +507,10 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to encode response", nil) } - // Auto-forward message as prompt to running agent if orchestrator is available. - // This runs async so the WS request can respond immediately. - // Use context.WithoutCancel so the prompt continues even if the WebSocket client disconnects. + // Auto-forward every accepted message as a prompt to the running agent when + // an orchestrator is available. This runs async so the WS request can + // respond immediately. Plan mode changes the execution prompt and agent + // behavior; it does not make message.add a record-only operation. if h.orchestrator != nil && !turnStartResult.Queued { h.dispatchPromptAsync(ctx, req, sessionResp.Session.AgentProfileID, startCreatedSession, steer) } diff --git a/apps/backend/internal/task/handlers/message_handlers_plan_mode_test.go b/apps/backend/internal/task/handlers/message_handlers_plan_mode_test.go new file mode 100644 index 0000000000..5a61099540 --- /dev/null +++ b/apps/backend/internal/task/handlers/message_handlers_plan_mode_test.go @@ -0,0 +1,146 @@ +package handlers + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/task/service" + v1 "github.com/kandev/kandev/pkg/api/v1" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// TestWSAddMessage_PlanModePersistsAndDispatches proves that a user message +// sent with plan_mode=true is recorded on the task and still enters the +// normal dispatch path. Plan mode changes the execution prompt; it does not +// turn message.add into a record-only operation. +func TestWSAddMessage_PlanModePersistsAndDispatches(t *testing.T) { + tests := []struct { + name string + planMode bool + wantMessageRows int + }{ + {name: "plan_mode true persists and dispatches", planMode: true, wantMessageRows: 1}, + {name: "plan_mode false dispatches normally", planMode: false, wantMessageRows: 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + now := time.Now().UTC() + repo := &messageAddSwitchRepo{ + tasks: map[string]*models.Task{ + "t1": {ID: "t1", State: v1.TaskStateInProgress, UpdatedAt: now}, + }, + sessions: map[string]*models.TaskSession{ + "s1": {ID: "s1", TaskID: "t1", State: models.TaskSessionStateRunning, AgentProfileID: "profile-1", UpdatedAt: now}, + }, + primaryID: "s1", + } + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + require.NoError(t, err) + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, TaskEnvironments: repo, + Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) + orch := newPlanModeRecordingOrchestrator() + h := NewMessageHandlers(svc, orch, log) + payload := map[string]interface{}{ + "task_id": "t1", "session_id": "s1", "content": "plan-mode instruction", + "plan_mode": tt.planMode, + } + req, err := ws.NewRequest("req-plan-mode", ws.ActionMessageAdd, payload) + require.NoError(t, err) + + resp, err := h.wsAddMessage(t.Context(), req) + require.NoError(t, err) + require.Equal(t, ws.MessageTypeResponse, resp.Type) + + assert.Equal(t, tt.wantMessageRows, repo.messageCount(), + "the user message must be persisted regardless of plan_mode") + + require.Eventually(t, func() bool { return orch.dispatchCalls() > 0 }, + time.Second, 5*time.Millisecond, + "message.add must dispatch via SteerTask or PromptTask") + }) + } +} + +// planModeRecordingOrchestrator records every dispatch-path call so the +// plan_mode gate can be asserted precisely. SteerTask is the path a +// steer-eligible RUNNING session normally takes; PromptTask is the fallback +// and the path for non-steer-eligible sessions. StartCreatedSession covers +// newly created sessions. +type planModeRecordingOrchestrator struct { + mu sync.Mutex + promptCalls int32 + steerCalls int32 + startCalls int32 + queueCalls int32 + turnStartCalls int32 +} + +func newPlanModeRecordingOrchestrator() *planModeRecordingOrchestrator { + return &planModeRecordingOrchestrator{} +} + +func (o *planModeRecordingOrchestrator) dispatchCalls() int32 { + return atomic.LoadInt32(&o.promptCalls) + atomic.LoadInt32(&o.steerCalls) + atomic.LoadInt32(&o.startCalls) +} + +func (o *planModeRecordingOrchestrator) PromptTask(_ context.Context, _, sessionID, _, _ string, _ bool, _ []v1.MessageAttachment, _ bool) (*orchestrator.PromptResult, error) { + o.mu.Lock() + defer o.mu.Unlock() + atomic.AddInt32(&o.promptCalls, 1) + _ = sessionID + return &orchestrator.PromptResult{}, nil +} + +func (o *planModeRecordingOrchestrator) SteerTask(_ context.Context, _, sessionID, _, _ string, _ bool, _ []v1.MessageAttachment) (*orchestrator.PromptResult, error) { + o.mu.Lock() + defer o.mu.Unlock() + atomic.AddInt32(&o.steerCalls, 1) + _ = sessionID + return &orchestrator.PromptResult{}, nil +} + +func (o *planModeRecordingOrchestrator) StartCreatedSession(_ context.Context, _, sessionID, _, _ string, _, _, _ bool, _ []v1.MessageAttachment, _ []v1.EntityReference) error { + o.mu.Lock() + defer o.mu.Unlock() + atomic.AddInt32(&o.startCalls, 1) + _ = sessionID + return nil +} + +func (o *planModeRecordingOrchestrator) QueueUserPrompt(_ context.Context, _, _, _, _ string, _ bool, _ []v1.MessageAttachment, _ map[string]interface{}, _ bool) error { + atomic.AddInt32(&o.queueCalls, 1) + return nil +} + +func (o *planModeRecordingOrchestrator) ProcessOnTurnStart(_ context.Context, _, sessionID string) (orchestrator.ProcessOnTurnStartResult, error) { + atomic.AddInt32(&o.turnStartCalls, 1) + _ = sessionID + return orchestrator.ProcessOnTurnStartResult{}, nil +} + +func (o *planModeRecordingOrchestrator) ResumeTaskSession(context.Context, string, string) error { + return nil +} +func (*planModeRecordingOrchestrator) StepRequiresCompletionSignal(context.Context, string) bool { + return false +} +func (*planModeRecordingOrchestrator) ForegroundActivity(string) v1.ForegroundActivity { + return v1.ForegroundActivityGenerating +} +func (*planModeRecordingOrchestrator) SteerEligible(string, models.TaskSessionState) bool { + return true +} diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index dd2b25d39f..2eda57ebef 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -942,7 +942,8 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { deferredLaunch = map[string]interface{}{ "intent": intent, "agent_profile_id": body.AgentProfileID, "executor_id": body.ExecutorID, "executor_profile_id": body.ExecutorProfileID, "prompt": description, - "plan_mode": body.PlanMode, "attachments": body.Attachments, + "plan_mode": body.PlanMode, + "attachments": body.Attachments, } } @@ -960,7 +961,7 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { Position: body.Position, Metadata: metadata, DeferredLaunch: deferredLaunch, - PlanMode: body.PlanMode && !body.StartAgent, + PlanMode: body.PlanMode, ParentID: body.ParentID, WorkspacePath: body.WorkspacePath, BlockedBy: body.BlockedBy, diff --git a/apps/backend/internal/task/handlers/task_http_handlers_test.go b/apps/backend/internal/task/handlers/task_http_handlers_test.go index 3d1fcf6bec..ba0cc39421 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers_test.go +++ b/apps/backend/internal/task/handlers/task_http_handlers_test.go @@ -997,6 +997,103 @@ func TestHTTPCreateTask_StartAgentKeepsCreatedStateWhenSchedulingUpdateFails(t * requireStartCreatedLaunch(t, startCreatedCalled) } +// TestHTTPCreateTask_PlanModeStartAgentPreservesPlanMode pins the task.create +// contract: plan_mode describes the execution prompt and must survive both +// task persistence and the deferred launch intent, even when start_agent is +// true. +func TestHTTPCreateTask_PlanModeStartAgentPreservesPlanMode(t *testing.T) { + gin.SetMode(gin.TestMode) + log := newTestLogger(t) + + repo := &captureCreateTaskRepo{} + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, TaskEnvironments: repo, + Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) + svc.SetWorkflowStepGetter(repo) + h := &TaskHandlers{service: svc, orchestrator: &captureOrchestrator{}, logger: log} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/tasks", strings.NewReader(`{ + "workspace_id": "ws-1", + "workflow_id": "wf-1", + "workflow_step_id": "step-1", + "title": "Plan then boot", + "description": "Plan a refactor and start an agent", + "priority": "medium", + "agent_profile_id": "profile-1", + "start_agent": true, + "plan_mode": true + }`)) + c.Request.Header.Set("Content-Type", "application/json") + + h.httpCreateTask(c) + + require.Equal(t, http.StatusOK, rec.Code, "body: %s", rec.Body.String()) + require.NotNil(t, repo.captured, "CreateTask must persist the task") + require.NotNil(t, repo.captured.Metadata, "task metadata must hold the deferred intent") + + deferredRaw, ok := repo.captured.Metadata[models.MetaKeyDeferredLaunch] + require.True(t, ok, "start_agent=true must persist a deferred_launch intent") + deferred, ok := deferredRaw.(map[string]interface{}) + require.True(t, ok, "deferred_launch intent must be a map[string]interface{}") + pmFlag, present := deferred["plan_mode"] + require.True(t, present, "the deferred intent must carry the plan_mode key (to assert the !StartAgent guard)") + assert.Equal(t, true, pmFlag, + "plan_mode=true must remain true in the deferred launch intent") +} + +// TestHTTPCreateTask_PlanModePrepareSessionKeepsPlanMode pins the +// non-conflicting half of the matrix: a plan-mode task whose deferred +// intent is prepare (not start_agent) keeps plan_mode=true on the +// intent. Prepare does not launch the agent, so plan mode is allowed +// to ride through to the eventual prompt-bearing start path. +func TestHTTPCreateTask_PlanModePrepareSessionKeepsPlanMode(t *testing.T) { + gin.SetMode(gin.TestMode) + log := newTestLogger(t) + + repo := &captureCreateTaskRepo{} + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, TaskEnvironments: repo, + Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) + svc.SetWorkflowStepGetter(repo) + h := &TaskHandlers{service: svc, orchestrator: &captureOrchestrator{}, logger: log} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/tasks", strings.NewReader(`{ + "workspace_id": "ws-1", + "workflow_id": "wf-1", + "workflow_step_id": "step-1", + "title": "Prepare plan session", + "description": "Prepare a session under plan mode", + "priority": "medium", + "agent_profile_id": "profile-1", + "prepare_session": true, + "plan_mode": true + }`)) + c.Request.Header.Set("Content-Type", "application/json") + + h.httpCreateTask(c) + + require.Equal(t, http.StatusOK, rec.Code, "body: %s", rec.Body.String()) + require.NotNil(t, repo.captured) + deferredRaw, ok := repo.captured.Metadata[models.MetaKeyDeferredLaunch] + require.True(t, ok) + deferred, ok := deferredRaw.(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, true, deferred["plan_mode"], + "plan_mode=true + prepare_session=true must persist plan_mode=true (prepare does not start an agent)") +} + func requireStartCreatedLaunch(t *testing.T, started <-chan struct{}) { t.Helper() select { diff --git a/apps/backend/internal/task/handlers/task_ws_handlers.go b/apps/backend/internal/task/handlers/task_ws_handlers.go index acfc566554..3ca162465c 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers.go @@ -167,7 +167,8 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M deferredLaunch = map[string]interface{}{ "intent": "start", "agent_profile_id": req.AgentProfileID, "executor_id": req.ExecutorID, "executor_profile_id": req.ExecutorProfileID, "prompt": description, - "plan_mode": req.PlanMode, "attachments": req.Attachments, + "plan_mode": req.PlanMode, + "attachments": req.Attachments, } } @@ -184,7 +185,7 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M Position: req.Position, Metadata: req.Metadata, DeferredLaunch: deferredLaunch, - PlanMode: req.PlanMode && !req.StartAgent, + PlanMode: req.PlanMode, ParentID: req.ParentID, }) if err != nil { diff --git a/apps/backend/internal/task/handlers/task_ws_handlers_test.go b/apps/backend/internal/task/handlers/task_ws_handlers_test.go index 676f27cb64..d5950132ae 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers_test.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers_test.go @@ -5,6 +5,7 @@ import ( "fmt" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/kandev/kandev/internal/auth/authn" @@ -423,3 +424,35 @@ func TestWSListTasksSurfacesUnknownWorkflow(t *testing.T) { require.Equal(t, string(ws.ErrorCodeInternalError), payload.Code) require.Equal(t, "Failed to list tasks", payload.Message) } + +// TestWSCreateTask_PlanModeStartAgentPreservesPlanMode pins the task.create +// contract on the WS handler. plan_mode describes the execution prompt and +// must survive task persistence and the deferred launch intent. +func TestWSCreateTask_PlanModeStartAgentPreservesPlanMode(t *testing.T) { + repo := &wsTaskRepo{} + h := newWSTaskHandlers(t, repo) + + resp, err := h.wsCreateTask(context.Background(), wsWorkflowRequest(t, ws.ActionTaskCreate, map[string]any{ + "workspace_id": "ws-b", + "workflow_id": "wf-b", + "title": "Plan then boot", + "agent_profile_id": "profile-1", + "start_agent": true, + "plan_mode": true, + })) + + require.NoError(t, err) + require.Equal(t, ws.MessageTypeResponse, resp.Type, "body: %s", resp.Payload) + require.Len(t, repo.created, 1, "task.create must persist the task") + captured := repo.created[0] + require.NotNil(t, captured.Metadata, "task metadata must hold the deferred intent") + + deferredRaw, ok := captured.Metadata[models.MetaKeyDeferredLaunch] + require.True(t, ok, "start_agent=true must persist a deferred_launch intent") + deferred, ok := deferredRaw.(map[string]interface{}) + require.True(t, ok, "deferred_launch intent must be a map[string]interface{}") + pmFlag, present := deferred["plan_mode"] + require.True(t, present, "the deferred intent must carry the plan_mode key (to assert the !StartAgent guard)") + assert.Equal(t, true, pmFlag, + "plan_mode=true must remain true in the deferred launch intent") +} diff --git a/apps/web/components/task/chat/chat-input-area.test.tsx b/apps/web/components/task/chat/chat-input-area.test.tsx index 3789c1b30b..ee85d90394 100644 --- a/apps/web/components/task/chat/chat-input-area.test.tsx +++ b/apps/web/components/task/chat/chat-input-area.test.tsx @@ -141,7 +141,9 @@ describe("useSubmitHandler", () => { variant: "error", }); }); +}); +describe("useSubmitHandler routing", () => { it("queues through the shared handler instead of preview onSend while clarification is pending", async () => { const onSend = vi.fn(); const { result } = renderHook(() => @@ -167,7 +169,26 @@ describe("useSubmitHandler", () => { expect(onSend).toHaveBeenCalledWith({ message: "direct preview message" }); expect(handleSendMessageMock).not.toHaveBeenCalled(); }); +}); +describe("useSubmitHandler plan mode", () => { + it("uses the normal send path in plan mode", async () => { + const onSend = vi.fn(); + const { result } = renderHook(() => + useSubmitHandler(panelState({ planModeEnabled: true }), onSend), + ); + + await act(async () => { + await result.current.handleSubmit({ message: "plan-mode instruction" }); + }); + + expect(onSend).toHaveBeenCalledWith({ message: "plan-mode instruction" }); + expect(handleSendMessageMock).not.toHaveBeenCalled(); + expect(toastMock).not.toHaveBeenCalled(); + }); +}); + +describe("useSubmitHandler deterministic failures", () => { it.each([ ["connection-unavailable", "Connection unavailable. Reconnect and try again."], ["session-unavailable", "The selected session is not available for input."], @@ -190,7 +211,9 @@ describe("useSubmitHandler", () => { variant: "error", }); }); +}); +describe("useSubmitHandler message comments", () => { it("keeps agent message comments pending when sending fails", async () => { vi.spyOn(console, "error").mockImplementation(() => undefined); const markCommentsSent = vi.fn(); diff --git a/apps/web/components/task/chat/chat-input-area.tsx b/apps/web/components/task/chat/chat-input-area.tsx index 31fbb55a45..694636b5db 100644 --- a/apps/web/components/task/chat/chat-input-area.tsx +++ b/apps/web/components/task/chat/chat-input-area.tsx @@ -164,7 +164,6 @@ function usePanelMessageHandler(panelState: ChatPanelState) { prompts, }); } - /** Builds the composer's submit handler, tracking in-flight sends and * routing errors to a toast. */ export function useSubmitHandler( diff --git a/apps/web/src/locales/zh-hk/task.json b/apps/web/src/locales/zh-hk/task.json index 8a7fae0505..bf43c38ea7 100644 --- a/apps/web/src/locales/zh-hk/task.json +++ b/apps/web/src/locales/zh-hk/task.json @@ -1773,7 +1773,7 @@ "durationUnitMinutes": "分", "durationUnitSeconds": "秒", "expandPrompt": "展開提示", - "promptHistory": "提示紀錄", + "promptHistory": "提示歷史", "promptHistoryEmpty": "暫無提示。", "scrollToPrompt": "滾動到提示" } diff --git a/apps/web/src/locales/zh-tw/task.json b/apps/web/src/locales/zh-tw/task.json index 61d7a020fb..fbd692d7e8 100644 --- a/apps/web/src/locales/zh-tw/task.json +++ b/apps/web/src/locales/zh-tw/task.json @@ -1773,7 +1773,7 @@ "durationUnitMinutes": "分", "durationUnitSeconds": "秒", "expandPrompt": "展開提示", - "promptHistory": "提示記錄", + "promptHistory": "提示歷史", "promptHistoryEmpty": "暫無提示。", "scrollToPrompt": "滾動到提示" }