fix(orchestrator): controller advancement — subtask WAITING guard, plan_mode gate, idle-session reclaim (#2811)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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.

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 {

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) {

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",

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)

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

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

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) {

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

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

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

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
 }

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

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

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,

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 {

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 {

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

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();

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(

apps/web/src/locales/zh-hk/task.json

@@ -1773,7 +1773,7 @@
   "durationUnitMinutes": "分",
   "durationUnitSeconds": "秒",
   "expandPrompt": "展開提示",
-  "promptHistory": "提示紀錄",
+  "promptHistory": "提示歷史",
   "promptHistoryEmpty": "暫無提示。",
   "scrollToPrompt": "滾動到提示"
 }

apps/web/src/locales/zh-tw/task.json

@@ -1773,7 +1773,7 @@
   "durationUnitMinutes": "分",
   "durationUnitSeconds": "秒",
   "expandPrompt": "展開提示",
-  "promptHistory": "提示記錄",
+  "promptHistory": "提示歷史",
   "promptHistoryEmpty": "暫無提示。",
   "scrollToPrompt": "滾動到提示"
 }