fix(backend): tear down orphaned agent processes when a session never starts (#3394)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/agent/runtime/lifecycle/execution_store.go

@@ -221,6 +221,39 @@ func (s *ExecutionStore) ActivePromptGeneration(executionID string) uint64 {
 	return gen
 }
 
+// ExecutionReference identifies one registered execution without requiring a
+// later lookup through the session index. Both IDs are captured from the same
+// execution-store snapshot so a caller can keep targeting the original
+// execution if a session later acquires a replacement.
+type ExecutionReference struct {
+	SessionID   string
+	ExecutionID string
+}
+
+// ListExecutionsForTask returns the session and execution IDs of executions
+// registered in-memory under taskID. It is a read-only snapshot, independent
+// of any session's persisted database state. Task-scoped stop uses it to
+// recover an execution whose session row is already terminal (for example,
+// FAILED) but whose prior teardown attempt failed.
+func (s *ExecutionStore) ListExecutionsForTask(taskID string) []ExecutionReference {
+	if taskID == "" {
+		return nil
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var references []ExecutionReference
+	for _, execution := range s.executions {
+		if execution.TaskID == taskID && execution.SessionID != "" && execution.ID != "" {
+			references = append(references, ExecutionReference{
+				SessionID:   execution.SessionID,
+				ExecutionID: execution.ID,
+			})
+		}
+	}
+	return references
+}
+
 // GetByTaskEnvironmentID returns any execution associated with a task environment ID.
 func (s *ExecutionStore) GetByTaskEnvironmentID(taskEnvironmentID string) (*AgentExecution, bool) {
 	s.mu.RLock()

apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go

@@ -237,3 +237,41 @@ func TestExecutionStore_ClaimPromptActivityRequiresCurrentIdentity(t *testing.T)
 		t.Fatalf("ClaimPromptActivity after removal = %v, want ErrExecutionNotFound", err)
 	}
 }
+
+// TestExecutionStore_ListsExecutionsForTask covers
+// @covers AC-TASKS-TASK-STOP-REACHABILITY-001.1: the registry-only lookup a
+// task-scoped stop uses to recover a session whose database row is terminal
+// but whose execution is still registered.
+func TestExecutionStore_ListsExecutionsForTask(t *testing.T) {
+	store := NewExecutionStore()
+	if err := store.Add(&AgentExecution{ID: "exec-1", SessionID: "session-1", TaskID: "task-a"}); err != nil {
+		t.Fatalf("Add exec-1: %v", err)
+	}
+	if err := store.Add(&AgentExecution{ID: "exec-2", SessionID: "session-2", TaskID: "task-a"}); err != nil {
+		t.Fatalf("Add exec-2: %v", err)
+	}
+	if err := store.Add(&AgentExecution{ID: "exec-3", SessionID: "session-3", TaskID: "task-b"}); err != nil {
+		t.Fatalf("Add exec-3: %v", err)
+	}
+
+	got := store.ListExecutionsForTask("task-a")
+	want := map[string]string{"session-1": "exec-1", "session-2": "exec-2"}
+	if len(got) != len(want) {
+		t.Fatalf("ListExecutionsForTask(task-a) = %v, want %v", got, want)
+	}
+	for _, reference := range got {
+		if executionID, ok := want[reference.SessionID]; !ok || executionID != reference.ExecutionID {
+			t.Fatalf("ListExecutionsForTask(task-a) returned unexpected reference %#v", reference)
+		}
+	}
+
+	if got := store.ListExecutionsForTask("task-does-not-exist"); len(got) != 0 {
+		t.Fatalf("ListExecutionsForTask(empty task) = %v, want none", got)
+	}
+
+	store.Remove("exec-1")
+	got = store.ListExecutionsForTask("task-a")
+	if len(got) != 1 || got[0] != (ExecutionReference{SessionID: "session-2", ExecutionID: "exec-2"}) {
+		t.Fatalf("ListExecutionsForTask(task-a) after Remove(exec-1) = %v, want [{session-2 exec-2}]", got)
+	}
+}

apps/backend/internal/agent/runtime/lifecycle/manager_events.go

@@ -550,13 +550,13 @@ func isTerminalToolUpdate(event agentctl.AgentEvent) bool {
 // accidentally re-arm a freshly-booted no-prompt session as Running.
 func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.AgentEvent) {
 	_, isTurnContent := turnContentEventTypes[event.Type]
-	execution.lastActivityAtMu.Lock()
-	execution.lastActivityAt = time.Now()
 	if isTurnContent {
+		execution.lastActivityAtMu.Lock()
+		execution.lastActivityAt = time.Now()
 		execution.agentEventSincePrompt = true
 		execution.promptActivityEpoch++
+		execution.lastActivityAtMu.Unlock()
 	}
-	execution.lastActivityAtMu.Unlock()
 
 	// Gate firstActivityOnce on `Status != Ready` so a delayed metadata
 	// event arriving after MarkBootReady can't accidentally fire

apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go

@@ -1575,13 +1575,14 @@ func TestRecordActivity_MetadataDoesNotMarkPromptStarted(t *testing.T) {
 	mgr, _ := createTestManagerWithTracking()
 	execution := createTestExecution("exec-metadata", "task-1", "session-1")
 	execution.agentEventSincePrompt = false
-	execution.lastActivityAt = time.Now().Add(-time.Minute)
+	dispatchTime := time.Now().Add(-time.Minute)
+	execution.lastActivityAt = dispatchTime
 
 	mgr.recordActivity(execution, agentctl.AgentEvent{Type: "session_models"})
 
 	lastActivity, agentEventSeen, epoch := execution.promptActivitySnapshot()
-	if time.Since(lastActivity) > time.Second {
-		t.Fatalf("metadata event did not refresh last activity: %v ago", time.Since(lastActivity))
+	if !lastActivity.Equal(dispatchTime) {
+		t.Fatalf("metadata event advanced last activity: got %v, want unchanged %v", lastActivity, dispatchTime)
 	}
 	if agentEventSeen {
 		t.Fatal("metadata event marked the prompt as started")

apps/backend/internal/agent/runtime/lifecycle/manager_execution.go

@@ -460,6 +460,14 @@ func (m *Manager) GetExecutionIDForSession(_ context.Context, sessionID string)
 	return "", fmt.Errorf("%w: %s", ErrNoExecutionForSession, sessionID)
 }
 
+// ListExecutionsForTask returns a snapshot of the session and execution IDs
+// registered in-memory for taskID, independent of persisted session state.
+// Executor.StopByTaskID uses the paired IDs to recover an execution left
+// orphaned by a terminal session row without resolving a replacement later.
+func (m *Manager) ListExecutionsForTask(taskID string) []ExecutionReference {
+	return m.executionStore.ListExecutionsForTask(taskID)
+}
+
 // GetACPSessionIDForSession returns the ACP conversation currently owned by a
 // live execution. The orchestrator uses this optional accessor after a context
 // reset to persist the new conversation immediately, instead of depending on

apps/backend/internal/agent/runtime/lifecycle/stall_activity_test.go

@@ -0,0 +1,108 @@
+package lifecycle
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+// TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted covers
+// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.9: a metadata-only stream (no
+// turn-content event) must not keep resetting the stall clock, so the
+// watchdog still fires five minutes after dispatch with NeverStarted true.
+func TestWaitForPromptDone_MetadataOnlyStreamStillReportsNeverStarted(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		eventBus := &MockEventBusWithTracking{}
+		sm := NewSessionManager(newSessionTestLogger(), make(chan struct{}))
+		sm.eventPublisher = NewEventPublisher(eventBus, newSessionTestLogger())
+		execution := &AgentExecution{
+			ID:           "test-exec",
+			TaskID:       "test-task",
+			SessionID:    "test-session",
+			promptDoneCh: make(chan PromptCompletionSignal, 1),
+			Status:       v1.AgentStatusRunning,
+		}
+		execution.lastActivityAt = time.Now()
+
+		mgr := &Manager{eventPublisher: sm.eventPublisher}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		t.Cleanup(cancel)
+		waitResult := make(chan error, 1)
+		go func() {
+			_, err := sm.waitForPromptDone(ctx, execution, 7)
+			waitResult <- err
+		}()
+
+		stop := make(chan struct{})
+		defer close(stop)
+		go func() {
+			ticker := time.NewTicker(time.Minute)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ticker.C:
+					mgr.recordActivity(execution, agentctl.AgentEvent{Type: "usage_update"})
+				case <-stop:
+					return
+				}
+			}
+		}()
+
+		time.Sleep(5 * time.Minute)
+		synctest.Wait()
+
+		payload := lastStalledPayload(t, eventBus)
+		if !payload.NeverStarted {
+			t.Fatalf("NeverStarted = false, want true: a metadata-only stream must not reset the clock")
+		}
+		if got := countPublishedEvents(eventBus, "agent.stalled"); got != 1 {
+			t.Fatalf("published stalled events = %d, want 1", got)
+		}
+
+		cancel()
+		if err := <-waitResult; !errors.Is(err, context.Canceled) {
+			t.Fatalf("waitForPromptDone error = %v, want context canceled", err)
+		}
+	})
+}
+
+// TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress covers
+// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.2: recordActivity must only move
+// lastActivityAt for a turnContentEventTypes event; a metadata frame leaves
+// the snapshot's timestamp and agentEventSincePrompt untouched.
+func TestRecordActivity_MetadataFrameDoesNotAdvancePromptProgress(t *testing.T) {
+	execution := &AgentExecution{
+		ID:     "test-exec",
+		Status: v1.AgentStatusRunning,
+	}
+	dispatchTime := time.Now().Add(-time.Minute)
+	execution.lastActivityAt = dispatchTime
+
+	mgr := &Manager{eventPublisher: NewEventPublisher(&MockEventBusWithTracking{}, newSessionTestLogger())}
+
+	mgr.recordActivity(execution, agentctl.AgentEvent{Type: "usage_update"})
+
+	lastActivity, agentEventSeen, _ := execution.promptActivitySnapshot()
+	if !lastActivity.Equal(dispatchTime) {
+		t.Fatalf("lastActivityAt = %v, want unchanged dispatch time %v after a metadata frame", lastActivity, dispatchTime)
+	}
+	if agentEventSeen {
+		t.Fatal("agentEventSincePrompt = true after a metadata-only frame")
+	}
+
+	mgr.recordActivity(execution, agentctl.AgentEvent{Type: "message_chunk"})
+
+	lastActivity, agentEventSeen, _ = execution.promptActivitySnapshot()
+	if !lastActivity.After(dispatchTime) {
+		t.Fatalf("lastActivityAt did not advance after a turn-content event")
+	}
+	if !agentEventSeen {
+		t.Fatal("agentEventSincePrompt = false after a turn-content event")
+	}
+}

apps/backend/internal/agent/runtime/lifecycle/types.go

@@ -221,7 +221,13 @@ type AgentExecution struct {
 	promptFinished   chan struct{}
 	promptFinishedMu sync.Mutex
 
-	// Last time an agent event was received (for stall detection)
+	// Last time a turn-content event was received (for stall detection).
+	// Advanced only by recordActivity for turnContentEventTypes, plus
+	// armPromptActivity, markAgentActivity, and recordSteerActivity. A
+	// metadata-only event (usage_update, context_window,
+	// available_commands_update, session_info_update, ...) does not move it,
+	// so a never-started prompt cannot be kept alive by traffic that carries
+	// no evidence of a turn.
 	lastActivityAt time.Time
 	// agentEventSincePrompt is armed (false) on each prompt dispatch and set
 	// true by the first genuine agent event (recordActivity/handleCompleteEvent)

apps/backend/internal/backendapp/adapters.go

@@ -900,6 +900,10 @@ func (a *lifecycleAdapter) GetExecutionIDForSession(ctx context.Context, session
 	return a.mgr.GetExecutionIDForSession(ctx, sessionID)
 }
 
+func (a *lifecycleAdapter) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference {
+	return a.mgr.ListExecutionsForTask(taskID)
+}
+
 func (a *lifecycleAdapter) GetRemoteRuntimeStatusBySession(ctx context.Context, sessionID string) (*executor.RemoteRuntimeStatus, error) {
 	status, ok := a.mgr.GetRemoteStatusBySessionID(ctx, sessionID)
 	if !ok || status == nil {

apps/backend/internal/backendapp/office_routine_cron_to_session_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/kandev/kandev/internal/agent/agents"
 	client "github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/db"
@@ -445,6 +446,9 @@ func (m *stubAgentManager) EnsureWorkspaceExecutionForSession(_ context.Context,
 func (m *stubAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) {
 	return "", nil
 }
+func (m *stubAgentManager) ListExecutionsForTask(_ string) []lifecycle.ExecutionReference {
+	return nil
+}
 func (m *stubAgentManager) GetGitLog(
 	_ context.Context, _, _ string, _ int, _ string,
 ) (*client.GitLogResult, error) {

apps/backend/internal/integration/simulated_agent_manager_test.go

@@ -638,6 +638,19 @@ func (s *SimulatedAgentManagerClient) GetExecutionIDForSession(_ context.Context
 	}
 	return "", fmt.Errorf("no execution found for session %s", sessionID)
 }
+func (s *SimulatedAgentManagerClient) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var references []lifecycle.ExecutionReference
+	for executionID, inst := range s.instances {
+		if inst.taskID == taskID && inst.sessionID != "" {
+			references = append(references, lifecycle.ExecutionReference{
+				SessionID: inst.sessionID, ExecutionID: executionID,
+			})
+		}
+	}
+	return references
+}
 func (s *SimulatedAgentManagerClient) GetGitLog(_ context.Context, _, _ string, _ int, _ string) (*client.GitLogResult, error) {
 	return nil, nil
 }

apps/backend/internal/orchestrator/background_launch_extensibility_external_test.go

@@ -158,6 +158,9 @@ func (extAgentManager) GetGitStatusFresh(context.Context, string) (*client.GitSt
 	return nil, nil
 }
 func (extAgentManager) WaitForAgentctlReady(context.Context, string) error { return nil }
+func (extAgentManager) ListExecutionsForTask(string) []lifecycle.ExecutionReference {
+	return nil
+}
 
 var _ executor.AgentManagerClient = extAgentManager{}
 

apps/backend/internal/orchestrator/event_handlers_stall.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -18,6 +19,12 @@ const (
 	metaKeyActionVisibility     = "action_visibility"
 	actionVisibilityRunning     = "running"
 	neverStartedNoticeContent   = "Agent produced no output since start."
+	// neverStartedStopTimeout bounds the detached runtime teardown started by
+	// stopNeverStartedExecution. A forced stop skips the graceful agentctl
+	// call, so nothing downstream (e.g. the Docker backend's cleanup context)
+	// applies its own bound in that path - an unresponsive daemon would
+	// otherwise hang the teardown goroutine indefinitely.
+	neverStartedStopTimeout = 30 * time.Second
 )
 
 // errAgentNeverStarted is the launch-failure error recorded when a stalled
@@ -117,10 +124,58 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen
 			zap.Error(err))
 	}
 	if payload.NeverStarted {
-		s.recordSessionLaunchFailure(ctx, payload.TaskID, payload.SessionID, errAgentNeverStarted, session)
+		if s.recordSessionLaunchFailure(ctx, payload.TaskID, payload.SessionID, errAgentNeverStarted, session) {
+			s.stopNeverStartedExecution(ctx, payload)
+		} else {
+			s.logger.Warn("skipping never-started teardown: session was not durably recorded FAILED",
+				zap.String("task_id", payload.TaskID),
+				zap.String("session_id", payload.SessionID))
+		}
 	}
 }
 
+// stopNeverStartedExecution tears down the process behind a never-started
+// stall. It runs after the session and task have already been recorded
+// FAILED, so a teardown failure never changes that recorded state: FAILED and
+// its launch-failure message stay authoritative either way.
+//
+// Teardown ownership is claimed synchronously, while handleAgentStalled still
+// holds the per-session cancelInFlight guard, via claimExecutionTeardown
+// directly rather than through Executor's RegisterExecutionStopOwner
+// callback: that callback TryLocks the same guard, so it would always fail to
+// reentrantly acquire it and silently no-op. claimExecutionTeardown's own
+// contract is to be called while already holding that guard.
+//
+// The actual stop runs detached and bounded (mirroring Executor.scheduleStop)
+// rather than inline: handleAgentStalled's caller is the synchronous event
+// bus, so a stop that blocked here would hold the cancelInFlight guard for
+// its whole duration - blocking the user's own Stop/Cancel/Delete for this
+// exact session, plus the event bus's dispatch lock. A forced stop skips the
+// graceful agentctl call, so the runtime backend applies no bound of its own;
+// neverStartedStopTimeout is this call's only protection against a hung
+// daemon. lifecycle.ErrExecutionNotFound means nothing is running and is
+// treated as success; any other failure is logged and left for a later stop
+// to retry against the still-registered execution.
+func (s *Service) stopNeverStartedExecution(ctx context.Context, payload lifecycle.AgentStalledPayload) {
+	if payload.AgentExecutionID == "" || s.agentManager == nil {
+		return
+	}
+	s.claimExecutionTeardown(payload.SessionID, payload.AgentExecutionID, executionTeardownIntentForce)
+	stopCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), neverStartedStopTimeout)
+	go func() {
+		defer cancel()
+		err := s.agentManager.StopAgentWithReason(stopCtx, payload.AgentExecutionID, "agent never started before stall", true)
+		if err == nil || errors.Is(err, lifecycle.ErrExecutionNotFound) {
+			return
+		}
+		s.logger.Warn("failed to tear down never-started execution after stall",
+			zap.String("task_id", payload.TaskID),
+			zap.String("session_id", payload.SessionID),
+			zap.String("execution_id", payload.AgentExecutionID),
+			zap.Error(err))
+	}()
+}
+
 func stallNoticeContent(payload lifecycle.AgentStalledPayload) string {
 	if payload.NeverStarted {
 		return neverStartedNoticeContent

apps/backend/internal/orchestrator/event_handlers_stall_test.go

@@ -2,14 +2,31 @@ package orchestrator
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
 	"github.com/kandev/kandev/internal/task/models"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 )
 
+// waitForStopAgentCall blocks until ch delivers the detached
+// stopNeverStartedExecution goroutine's StopAgentWithReason call, or fails
+// the test after a bounded wait. Never poll or sleep for this: the call is
+// intentionally off handleAgentStalled's own goroutine.
+func waitForStopAgentCall(t *testing.T, ch <-chan stopAgentCall) stopAgentCall {
+	t.Helper()
+	select {
+	case call := <-ch:
+		return call
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for the detached StopAgentWithReason call")
+		return stopAgentCall{}
+	}
+}
+
 func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)
@@ -219,6 +236,348 @@ func TestHandleAgentStalled_RejectsSettledOrStalePrompt(t *testing.T) {
 	}
 }
 
+// TestHandleAgentStalled_NeverStartedStopsExecution covers
+// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.10: after the never-started
+// branch records the session and task FAILED, it must also tear down the
+// execution named by the payload, forced, exactly once.
+func TestHandleAgentStalled_NeverStartedStopsExecution(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	stopCalls := make(chan stopAgentCall, 1)
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, reason string, force bool) error {
+			stopCalls <- stopAgentCall{ExecutionID: executionID, Reason: reason, Force: force}
+			return nil
+		},
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		taskRepo,
+		agentMgr,
+	)
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+		AgentExecutionID: "execution-1",
+		TaskID:           "task-1",
+		SessionID:        "session-1",
+		PromptGeneration: 7,
+		ActivityEpoch:    1,
+		NeverStarted:     true,
+	})
+
+	call := waitForStopAgentCall(t, stopCalls)
+	if call.ExecutionID != "execution-1" {
+		t.Fatalf("stopped execution = %q, want execution-1", call.ExecutionID)
+	}
+	if !call.Force {
+		t.Fatal("teardown stop was not forced")
+	}
+
+	agentMgr.mu.Lock()
+	callCount := len(agentMgr.stopAgentWithReasonArgs)
+	agentMgr.mu.Unlock()
+	if callCount != 1 {
+		t.Fatalf("StopAgentWithReason calls = %d, want 1", callCount)
+	}
+
+	after, err := repo.GetTaskSession(ctx, "session-1")
+	if err != nil {
+		t.Fatalf("get session after handling stall: %v", err)
+	}
+	if after.State != models.TaskSessionStateFailed {
+		t.Fatalf("session state = %q, want FAILED", after.State)
+	}
+}
+
+// TestHandleAgentStalled_NeverStartedStopRunsDetachedFromGuard covers R3-F1:
+// stopNeverStartedExecution must not block handleAgentStalled's return on the
+// runtime teardown. handleAgentStalled holds the per-session cancelInFlight
+// guard for its own duration (released via defer on return), and that guard
+// also serializes the user's own Stop/Cancel/Delete for this session - so if
+// the teardown ran inline here, an unresponsive runtime backend would hold
+// that guard, the synchronous event bus's dispatch lock, and the
+// waitForPromptDone goroutine for as long as the backend stayed hung.
+func TestHandleAgentStalled_NeverStartedStopRunsDetachedFromGuard(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	stopStarted := make(chan struct{})
+	stopRelease := make(chan struct{})
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+		stopAgentWithReasonFunc: func(context.Context, string, string, bool) error {
+			close(stopStarted)
+			<-stopRelease
+			return nil
+		},
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		taskRepo,
+		agentMgr,
+	)
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	handlerDone := make(chan struct{})
+	go func() {
+		svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+			AgentExecutionID: "execution-1",
+			TaskID:           "task-1",
+			SessionID:        "session-1",
+			PromptGeneration: 7,
+			ActivityEpoch:    1,
+			NeverStarted:     true,
+		})
+		close(handlerDone)
+	}()
+
+	select {
+	case <-handlerDone:
+	case <-time.After(2 * time.Second):
+		close(stopRelease)
+		t.Fatal("handleAgentStalled blocked on runtime teardown instead of returning once it was scheduled")
+	}
+
+	select {
+	case <-stopStarted:
+	case <-time.After(2 * time.Second):
+		close(stopRelease)
+		t.Fatal("StopAgentWithReason was never invoked")
+	}
+	close(stopRelease)
+}
+
+// TestHandleAgentStalled_NeverStartedClaimsTeardownOwnershipBeforeStop covers
+// Review round 2 Finding 2: stopNeverStartedExecution must claim execution
+// teardown ownership before stopping the agent, so the synchronously
+// published agent.stopped event is recognized by handleAgentStopped's
+// hasExecutionTeardownOwner check as already-owned instead of running that
+// handler's full state-decision path.
+func TestHandleAgentStalled_NeverStartedClaimsTeardownOwnershipBeforeStop(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		taskRepo,
+		agentMgr,
+	)
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	if svc.hasExecutionTeardownOwner("session-1", "execution-1") {
+		t.Fatal("teardown ownership claimed before the stall was handled")
+	}
+
+	svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+		AgentExecutionID: "execution-1",
+		TaskID:           "task-1",
+		SessionID:        "session-1",
+		PromptGeneration: 7,
+		ActivityEpoch:    1,
+		NeverStarted:     true,
+	})
+
+	if !svc.hasExecutionTeardownOwner("session-1", "execution-1") {
+		t.Fatal("stopNeverStartedExecution did not claim teardown ownership before stopping")
+	}
+}
+
+// TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails covers
+// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.11: a teardown failure must not
+// overwrite the recorded FAILED state or its launch-failure message.
+func TestHandleAgentStalled_NeverStartedKeepsFailedStateWhenStopFails(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	stopAttempted := make(chan struct{})
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+		stopAgentWithReasonFunc: func(context.Context, string, string, bool) error {
+			defer close(stopAttempted)
+			return errors.New("agentctl unreachable")
+		},
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		taskRepo,
+		agentMgr,
+	)
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+		AgentExecutionID: "execution-1",
+		TaskID:           "task-1",
+		SessionID:        "session-1",
+		PromptGeneration: 7,
+		ActivityEpoch:    1,
+		NeverStarted:     true,
+	})
+
+	select {
+	case <-stopAttempted:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for the detached teardown attempt")
+	}
+
+	after, err := repo.GetTaskSession(ctx, "session-1")
+	if err != nil {
+		t.Fatalf("get session after handling stall: %v", err)
+	}
+	if after.State != models.TaskSessionStateFailed {
+		t.Fatalf("session state = %q, want FAILED despite teardown failure", after.State)
+	}
+	if after.ErrorMessage != errAgentNeverStarted.Error() {
+		t.Fatalf("session error message = %q, want %q", after.ErrorMessage, errAgentNeverStarted.Error())
+	}
+}
+
+// TestHandleAgentStalled_AdvisoryStallStopsNothing covers
+// @covers AC-AGENTS-AGENT-STALL-RECOVERY-001.7: an advisory stall (agent
+// produced output before stalling) must not tear down the execution.
+func TestHandleAgentStalled_AdvisoryStallStopsNothing(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		newMockTaskRepo(),
+		agentMgr,
+	)
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+		AgentExecutionID: "execution-1",
+		TaskID:           "task-1",
+		SessionID:        "session-1",
+		PromptGeneration: 7,
+		ActivityEpoch:    1,
+		ToolName:         "shell",
+		ToolTitle:        "Start dev server",
+	})
+
+	agentMgr.mu.Lock()
+	stopCalls := len(agentMgr.stopAgentWithReasonArgs)
+	agentMgr.mu.Unlock()
+	if stopCalls != 0 {
+		t.Fatalf("StopAgentWithReason calls = %d, want 0 for an advisory stall", stopCalls)
+	}
+}
+
+// TestHandleAgentStalled_NeverStartedSkipsTeardownWhenFailedNotPersisted covers
+// the review finding that a rejected/failed FAILED-transition CAS must not be
+// followed by a forced process kill: killing the process without a durable
+// FAILED record would leave the session claiming RUNNING with no process and
+// no way for handleAgentStopped to reconcile it (stopNeverStartedExecution
+// claims teardown ownership first, so the resulting agent.stopped event is
+// ignored as already-owned).
+func TestHandleAgentStalled_NeverStartedSkipsTeardownWhenFailedNotPersisted(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task-1", "session-1", "step-1")
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup:   repo,
+		currentPromptExecutionID: "execution-1",
+	}
+	agentMgr.currentPromptGeneration.Store(7)
+	agentMgr.currentPromptActivityEpoch.Store(1)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(
+		repo,
+		newMockStepGetter(),
+		taskRepo,
+		agentMgr,
+	)
+	writeFailure := errors.New("database is read-only")
+	svc.repo = failSessionStateUpdateRepo{repoStore: repo, err: writeFailure}
+	svc.turnService = &repoTurnService{repo: repo}
+	if _, err := svc.turnService.StartTurn(ctx, "session-1"); err != nil {
+		t.Fatalf("start active turn: %v", err)
+	}
+	svc.messageCreator = &mockMessageCreator{}
+
+	svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
+		AgentExecutionID: "execution-1",
+		TaskID:           "task-1",
+		SessionID:        "session-1",
+		PromptGeneration: 7,
+		ActivityEpoch:    1,
+		NeverStarted:     true,
+	})
+
+	agentMgr.mu.Lock()
+	stopCalls := len(agentMgr.stopAgentWithReasonArgs)
+	agentMgr.mu.Unlock()
+	if stopCalls != 0 {
+		t.Fatalf("StopAgentWithReason calls = %d, want 0 when the FAILED transition did not persist", stopCalls)
+	}
+
+	after, err := repo.GetTaskSession(ctx, "session-1")
+	if err != nil {
+		t.Fatalf("get session after handling stall: %v", err)
+	}
+	if after.State != models.TaskSessionStateRunning {
+		t.Fatalf("session state = %q, want RUNNING (unchanged) since the FAILED write never persisted", after.State)
+	}
+}
+
 func TestStallNoticeContentFallsBackWithoutTool(t *testing.T) {
 	if got := stallNoticeContent(lifecycle.AgentStalledPayload{}); got != "Still waiting for the agent." {
 		t.Fatalf("stallNoticeContent() = %q, want generic fallback", got)

apps/backend/internal/orchestrator/event_handlers_test.go

@@ -357,6 +357,9 @@ type mockAgentManager struct {
 	}
 	// Optional current ACP session lookup used by reset-token generation tests.
 	getACPSessionIDForSessionFunc func(string) (string, bool)
+	// Optional override for ListExecutionsForTask. When unset, the default
+	// implementation returns nil (no registry-recovered sessions).
+	listExecutionsForTaskFunc func(taskID string) []lifecycle.ExecutionReference
 
 	// CancelAgent tracking. cancelAgentCalls counts every invocation. If
 	// cancelAgentBlock is non-nil, CancelAgent blocks on it before returning;
@@ -793,6 +796,13 @@ func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, session
 	return "", fmt.Errorf("no execution found")
 }
 
+func (m *mockAgentManager) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference {
+	if m.listExecutionsForTaskFunc != nil {
+		return m.listExecutionsForTaskFunc(taskID)
+	}
+	return nil
+}
+
 func (m *mockAgentManager) GetACPSessionIDForSession(sessionID string) (string, bool) {
 	if m.getACPSessionIDForSessionFunc == nil {
 		return "", false

apps/backend/internal/orchestrator/executor/executor.go

@@ -242,6 +242,13 @@ var (
 	// must not start the process and must arbitrate exact-execution teardown
 	// ownership before deciding whether to force-stop the registered runtime.
 	ErrSessionStateSuperseded = errors.New("session state superseded by terminal transition")
+	// ErrOrphanRecoveryIncomplete means StopByTaskID stopped every session it
+	// found but could not load at least one registry-only orphan's row, so the
+	// task-scoped stop is not fully confirmed. Callers that already observed a
+	// successful stop should log this rather than treat it as a hard failure;
+	// it stays distinguishable from ErrExecutionNotFound so a retry keeps
+	// happening instead of being reported as a false all-clear.
+	ErrOrphanRecoveryIncomplete = errors.New("orphaned execution recovery incomplete")
 )
 
 // SessionStateSupersededError records the terminal state that rejected a
@@ -402,6 +409,13 @@ type AgentManagerClient interface {
 	// Used to detect stale AgentExecutionID values in the database after restart.
 	GetExecutionIDForSession(ctx context.Context, sessionID string) (string, error)
 
+	// ListExecutionsForTask returns a snapshot of the session and execution IDs
+	// registered in-memory for taskID, independent of persisted session state.
+	// StopByTaskID uses the paired IDs to recover a registered execution whose
+	// session row is already terminal in the database (for example, FAILED
+	// after a never-started stall whose teardown attempt failed).
+	ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference
+
 	// GetGitLog retrieves the git log for a session from baseCommit to HEAD.
 	// If targetBranch is provided, uses dynamic merge-base calculation for accurate filtering.
 	// Used for archive snapshot capture. Returns nil, nil if no execution exists.

apps/backend/internal/orchestrator/executor/executor_interaction.go

@@ -125,35 +125,202 @@ func (e *Executor) StopSessionDetailed(
 // stopWithSession preserves the legacy error-only contract. It intentionally
 // keeps session persistence best-effort and schedules teardown whenever a live
 // execution was found; existing UI and cleanup callers rely on that behavior.
+//
+// Stop ownership is registered before the CANCELLED write, not after: a
+// concurrent StartAgentProcess failure classifies CANCELLED as terminal and
+// claims forced cleanup for itself only when no owner is registered yet. If
+// the state write were visible first, that failure could win the claim and
+// issue its own teardown while this call also schedules one.
 func (e *Executor) stopWithSession(ctx context.Context, session *models.TaskSession, reason string, force bool) error {
-	executionID, err := e.agentManager.GetExecutionIDForSession(ctx, session.ID)
+	sessionLock := e.getSessionLock(session.ID)
+	sessionLock.Lock()
+	defer sessionLock.Unlock()
+
+	executionID, err := e.resolveExecutionIDForStop(ctx, session.ID)
+	if err != nil {
+		return err
+	}
+
+	return e.stopWithSessionExecution(ctx, session, executionID, reason, force, true)
+}
+
+// stopWithSessionExecution records stop ownership and schedules teardown for
+// the supplied execution. When cancelSession is true, it also transitions the
+// current session to CANCELLED. Callers must hold the session lock when they
+// need to coordinate this decision with resume or launch.
+func (e *Executor) stopWithSessionExecution(
+	ctx context.Context,
+	session *models.TaskSession,
+	executionID, reason string,
+	force, cancelSession bool,
+) error {
+	e.logStop(session, executionID, reason, force)
+	e.registerStopOwner(session.ID, executionID, force)
+	if cancelSession {
+		// A session already in a terminal state carries its own outcome (for
+		// example, a launch failure's error_message). transitionSessionState
+		// re-reads the current row, so a terminal race cannot be clobbered.
+		if _, _, err := e.transitionSessionState(ctx, session.TaskID, session.ID, models.TaskSessionStateCancelled, reason); err != nil {
+			e.logger.Error("failed to update agent session status",
+				zap.String("session_id", session.ID),
+				zap.Error(err))
+		}
+	}
+	e.scheduleStop(ctx, session.ID, executionID, reason, force)
+	return nil
+}
+
+// resolveExecutionIDForStop looks up the live execution for sessionID,
+// applying the same not-found classification stopWithSession has always
+// used. Shared by stopWithSession and the registry-recovered paths in
+// StopByTaskID — it takes a bare session ID rather than a loaded row because
+// the registry-only path may not have one (see stopRegistryRecoveredByID).
+func (e *Executor) resolveExecutionIDForStop(ctx context.Context, sessionID string) (string, error) {
+	executionID, err := e.agentManager.GetExecutionIDForSession(ctx, sessionID)
 	if err != nil || executionID == "" {
 		if err != nil {
 			if errors.Is(err, lifecycle.ErrNoExecutionForSession) {
-				return fmt.Errorf("%w: %w: %w", ErrExecutionNotFound, runtimeapi.ErrNotFound, err)
+				return "", fmt.Errorf("%w: %w: %w", ErrExecutionNotFound, runtimeapi.ErrNotFound, err)
 			}
-			return fmt.Errorf("%w: lookup execution for session %q: %w", ErrExecutionNotFound, session.ID, err)
+			return "", fmt.Errorf("%w: lookup execution for session %q: %w", ErrExecutionNotFound, sessionID, err)
 		}
-		return ErrExecutionNotFound
+		return "", ErrExecutionNotFound
+	}
+	return executionID, nil
+}
+
+// registerStopOwner records advisory stop ownership ahead of a terminal
+// session mutation this call is about to make. Callers with no interleaved
+// mutation (stopRegistryRecoveredSession) may register and schedule together
+// via registerAndScheduleStop instead.
+func (e *Executor) registerStopOwner(sessionID, executionID string, force bool) {
+	if e.onExecutionStopOwnerRegistration != nil {
+		e.onExecutionStopOwnerRegistration(sessionID, executionID, force)
+	}
+}
+
+// registryRecoveredSession keeps the database row paired with the exact
+// execution captured from the registry snapshot.
+type registryRecoveredSession struct {
+	session   *models.TaskSession
+	reference lifecycle.ExecutionReference
+}
+
+// stopRegistryRecoveredSession stops the execution captured for a registry
+// only session. The per-session lock serializes the state and identity check
+// with resume. If the session is still active and still owns the captured
+// execution, use the normal cancellation transition. If a replacement owns
+// the session, schedule only the captured execution and leave the replacement
+// untouched.
+func (e *Executor) stopRegistryRecoveredSession(
+	ctx context.Context,
+	recovered registryRecoveredSession,
+	reason string,
+	force bool,
+) error {
+	session := recovered.session
+	reference := recovered.reference
+	if session == nil || session.ID == "" || reference.SessionID != session.ID || reference.ExecutionID == "" {
+		return fmt.Errorf("invalid registry recovery reference for task session")
 	}
 
-	e.logStop(session, executionID, reason, force)
-	if e.onExecutionStopOwnerRegistration != nil {
-		e.onExecutionStopOwnerRegistration(session.ID, executionID, force)
-	}
-	// A session already in a terminal state carries its own outcome (e.g. a
-	// launch failure's error_message); a runtime that outlived it in the
-	// in-memory execution store must still be torn down, but the DB row is
-	// not touched. transitionSessionState re-reads the session's current
-	// state itself rather than trusting the caller-supplied snapshot, so a
-	// session that turned terminal between the caller's read and this call
-	// (e.g. StopByTaskID iterating a list read once) can't be clobbered.
-	if _, _, err := e.transitionSessionState(ctx, session.TaskID, session.ID, models.TaskSessionStateCancelled, reason); err != nil {
-		e.logger.Error("failed to update agent session status",
+	sessionLock := e.getSessionLock(reference.SessionID)
+	sessionLock.Lock()
+	defer sessionLock.Unlock()
+
+	current := session
+	if refreshed, err := e.repo.GetTaskSession(ctx, session.ID); err == nil && refreshed != nil {
+		current = refreshed
+	} else if err != nil {
+		e.logger.Warn("failed to refresh registry-recovered session before stop; preserving captured execution",
+			zap.String("task_id", session.TaskID),
 			zap.String("session_id", session.ID),
 			zap.Error(err))
 	}
-	e.scheduleStop(ctx, session.ID, executionID, reason, force)
+
+	return e.stopCapturedExecution(ctx, session.TaskID, current, reference, reason, force)
+}
+
+// stopRegistryRecoveredByID stops the execution captured for a registry-only
+// session whose row could not be loaded during recovery. It never resolves a
+// replacement through the session index. A best-effort re-read can still
+// cancel the session when the row becomes available and owns the captured
+// execution; otherwise teardown is scheduled by the captured identity alone.
+func (e *Executor) stopRegistryRecoveredByID(
+	ctx context.Context,
+	taskID string,
+	reference lifecycle.ExecutionReference,
+	reason string,
+	force bool,
+) error {
+	if reference.SessionID == "" || reference.ExecutionID == "" {
+		return fmt.Errorf("invalid registry recovery reference for task %q", taskID)
+	}
+
+	sessionLock := e.getSessionLock(reference.SessionID)
+	sessionLock.Lock()
+	defer sessionLock.Unlock()
+
+	if session, err := e.repo.GetTaskSession(ctx, reference.SessionID); err == nil && session != nil {
+		return e.stopCapturedExecution(ctx, taskID, session, reference, reason, force)
+	} else if err != nil {
+		e.logger.Warn("failed to refresh registry-only session before stop; preserving captured execution",
+			zap.String("task_id", taskID),
+			zap.String("session_id", reference.SessionID),
+			zap.Error(err))
+	}
+
+	e.logger.Info("stopping registry-recovered execution with unloadable session row",
+		zap.String("task_id", taskID),
+		zap.String("session_id", reference.SessionID),
+		zap.String("agent_execution_id", reference.ExecutionID),
+		zap.String("reason", reason),
+		zap.Bool("force", force))
+	e.registerStopOwner(reference.SessionID, reference.ExecutionID, force)
+	e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force)
+	return nil
+}
+
+// stopCapturedExecution decides whether the captured execution still owns an
+// active session. The current execution lookup is used only for this equality
+// check; the stop target remains reference.ExecutionID in every branch.
+func (e *Executor) stopCapturedExecution(
+	ctx context.Context,
+	taskID string,
+	session *models.TaskSession,
+	reference lifecycle.ExecutionReference,
+	reason string,
+	force bool,
+) error {
+	if session == nil || session.ID != reference.SessionID {
+		return fmt.Errorf("registry recovery session does not match captured session %q", reference.SessionID)
+	}
+	if session.TaskID != taskID {
+		e.logger.Warn("registry recovery session task does not match captured task; preserving captured execution",
+			zap.String("task_id", taskID),
+			zap.String("session_id", reference.SessionID),
+			zap.String("session_task_id", session.TaskID),
+			zap.String("agent_execution_id", reference.ExecutionID))
+		e.registerStopOwner(reference.SessionID, reference.ExecutionID, force)
+		e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force)
+		return nil
+	}
+	currentExecutionID, err := e.agentManager.GetExecutionIDForSession(ctx, reference.SessionID)
+	if err == nil && currentExecutionID == reference.ExecutionID && models.IsTaskLookupActiveSessionState(session.State) {
+		return e.stopWithSessionExecution(ctx, session, reference.ExecutionID, reason, force, true)
+	}
+
+	e.logger.Info("stopping registry-recovered execution",
+		zap.String("task_id", session.TaskID),
+		zap.String("session_id", reference.SessionID),
+		zap.String("agent_execution_id", reference.ExecutionID),
+		zap.String("current_agent_execution_id", currentExecutionID),
+		zap.String("session_state", string(session.State)),
+		zap.String("reason", reason),
+		zap.Bool("force", force),
+		zap.Error(err))
+	e.registerStopOwner(reference.SessionID, reference.ExecutionID, force)
+	e.scheduleStop(ctx, reference.SessionID, reference.ExecutionID, reason, force)
 	return nil
 }
 
@@ -253,7 +420,12 @@ func (e *Executor) StopExecution(ctx context.Context, executionID string, reason
 	return nil
 }
 
-// StopByTaskID stops all active executions for a task
+// StopByTaskID stops all active executions for a task. It resolves what to
+// stop from the union of the database's active-session query and the
+// in-memory execution registry, so a session that is terminal in the
+// database but still holds a registered execution (an orphan left by a
+// failed teardown attempt) is reachable too — see
+// docs/specs/tasks/system-design/task-stop-reachability.md.
 func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason string, force bool) error {
 	// Get all active sessions for this task from database
 	sessions, err := e.repo.ListActiveTaskSessionsByTaskID(ctx, taskID)
@@ -264,7 +436,9 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin
 		return ErrExecutionNotFound
 	}
 
-	if len(sessions) == 0 {
+	recovered, unloadable, recoverErr := e.recoverRegistryOnlySessions(ctx, taskID, sessions)
+
+	if len(sessions) == 0 && len(recovered) == 0 && len(unloadable) == 0 {
 		return ErrExecutionNotFound
 	}
 
@@ -281,6 +455,42 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin
 			stoppedCount++
 		}
 	}
+	for _, recoveredSession := range recovered {
+		if err := e.stopRegistryRecoveredSession(ctx, recoveredSession, reason, force); err != nil {
+			e.logger.Warn("failed to stop registry-recovered session",
+				zap.String("task_id", taskID),
+				zap.String("session_id", recoveredSession.reference.SessionID),
+				zap.Error(err))
+			lastErr = err
+		} else {
+			stoppedCount++
+		}
+	}
+	for _, reference := range unloadable {
+		if err := e.stopRegistryRecoveredByID(ctx, taskID, reference, reason, force); err != nil {
+			e.logger.Warn("failed to stop registry-recovered execution with unloadable session row",
+				zap.String("task_id", taskID),
+				zap.String("session_id", reference.SessionID),
+				zap.Error(err))
+			lastErr = err
+		} else {
+			stoppedCount++
+		}
+	}
+
+	// A registry-only orphan whose row failed to load is still targeted by its
+	// captured execution ID (stopRegistryRecoveredByID), so recoverErr no
+	// longer means that execution was unreachable — only that its terminal
+	// state could not be confirmed or corrected from here. Distinguish a
+	// caller that already saw at least one execution stop (soft: log and
+	// continue) from one where nothing could be reached at all (hard
+	// failure), same as before this recovery path existed.
+	if recoverErr != nil {
+		if stoppedCount > 0 {
+			return fmt.Errorf("%w: task %q: %w", ErrOrphanRecoveryIncomplete, taskID, recoverErr)
+		}
+		return fmt.Errorf("task %q has a registered execution but its session could not be loaded: %w", taskID, recoverErr)
+	}
 
 	if stoppedCount == 0 && lastErr != nil {
 		return lastErr
@@ -289,6 +499,54 @@ func (e *Executor) StopByTaskID(ctx context.Context, taskID string, reason strin
 	return nil
 }
 
+// recoverRegistryOnlySessions loads the session rows for executions the
+// registry still holds for taskID but that the active-session query did not
+// return. Each loaded row stays paired with the exact execution reference
+// captured from the registry. A row that cannot be loaded is still returned by
+// that reference, rather than dropped, so teardown can be scheduled without a
+// later session-index lookup. Only the last load error is kept; one surfaced
+// failure is enough to stop the caller from reporting a false all-clear, and a
+// caller that wants to retry re-derives the full set on the next call.
+func (e *Executor) recoverRegistryOnlySessions(
+	ctx context.Context, taskID string, activeSessions []*models.TaskSession,
+) (recovered []registryRecoveredSession, unloadable []lifecycle.ExecutionReference, loadErr error) {
+	registeredExecutions := e.agentManager.ListExecutionsForTask(taskID)
+	if len(registeredExecutions) == 0 {
+		return nil, nil, nil
+	}
+
+	alreadyActive := make(map[string]struct{}, len(activeSessions))
+	for _, session := range activeSessions {
+		alreadyActive[session.ID] = struct{}{}
+	}
+
+	for _, reference := range registeredExecutions {
+		if reference.SessionID == "" || reference.ExecutionID == "" {
+			loadErr = fmt.Errorf("registry returned incomplete execution reference for task %q", taskID)
+			continue
+		}
+		if _, ok := alreadyActive[reference.SessionID]; ok {
+			continue
+		}
+		session, err := e.repo.GetTaskSession(ctx, reference.SessionID)
+		if err != nil || session == nil {
+			e.logger.Warn("failed to load registry-recovered session row for task stop; stopping by captured execution ID",
+				zap.String("task_id", taskID),
+				zap.String("session_id", reference.SessionID),
+				zap.Error(err))
+			if err != nil {
+				loadErr = err
+			}
+			unloadable = append(unloadable, reference)
+			continue
+		}
+		recovered = append(recovered, registryRecoveredSession{
+			session: session, reference: reference,
+		})
+	}
+	return recovered, unloadable, loadErr
+}
+
 // stopReasonPassthrough is the StopReason returned by Executor.Prompt when a
 // passthrough session's prompt is dispatched to PTY stdin (no ACP turn to
 // observe). The submit sequence is resolved per-agent via ResolvePassthroughConfig

apps/backend/internal/orchestrator/executor/executor_interaction_test.go

@@ -206,6 +206,243 @@ func TestStopSessionSynchronouslyStopsLateExecutionAfterSessionDeletion(t *testi
 	}
 }
 
+// TestStopByTaskID_StopsFailedSessionWithRegisteredExecution covers
+// @covers AC-TASKS-TASK-STOP-REACHABILITY-001.1-.2: a task whose only session
+// is FAILED in the database but still holds a registered execution (an
+// orphan left by a failed teardown attempt) must be stopped, and the
+// session's terminal state must survive — no CANCELLED transition.
+func TestStopByTaskID_StopsFailedSessionWithRegisteredExecution(t *testing.T) {
+	repo := newMockRepository()
+	repo.sessions["session-1"] = &models.TaskSession{
+		ID:           "session-1",
+		TaskID:       "task-1",
+		State:        models.TaskSessionStateFailed,
+		ErrorMessage: "agent produced no output since start",
+	}
+	stopCalls := make(chan string, 1)
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-orphan"}}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-orphan", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, force bool) error {
+			if !force {
+				t.Errorf("registry-recovered stop was not forced")
+			}
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopByTaskID: %v", err)
+	}
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-orphan" {
+			t.Fatalf("stopped execution = %q, want execution-orphan", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("registry-recovered execution was not stopped")
+	}
+	after, err := repo.GetTaskSession(context.Background(), "session-1")
+	if err != nil {
+		t.Fatalf("get session after stop: %v", err)
+	}
+	if after.State != models.TaskSessionStateFailed {
+		t.Fatalf("session state = %q, want unchanged FAILED", after.State)
+	}
+	if after.ErrorMessage != "agent produced no output since start" {
+		t.Fatalf("session error message = %q, want preserved launch-failure message", after.ErrorMessage)
+	}
+}
+
+// TestStopByTaskID_ActiveSessionKeepsCancellationTransition pins the
+// unchanged path: a task with an active session behaves exactly as before —
+// the session transitions to CANCELLED and its execution is stopped.
+func TestStopByTaskID_ActiveSessionKeepsCancellationTransition(t *testing.T) {
+	repo := newMockRepository()
+	repo.sessions["session-1"] = &models.TaskSession{
+		ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateRunning,
+	}
+	repo.listActiveTaskSessionsByTaskIDFunc = func(_ context.Context, taskID string) ([]*models.TaskSession, error) {
+		if taskID != "task-1" {
+			return nil, nil
+		}
+		return []*models.TaskSession{repo.sessions["session-1"]}, nil
+	}
+	stopCalls := make(chan string, 1)
+	manager := &mockAgentManager{
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-active", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopByTaskID: %v", err)
+	}
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-active" {
+			t.Fatalf("stopped execution = %q, want execution-active", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("active session's execution was not stopped")
+	}
+	after, err := repo.GetTaskSession(context.Background(), "session-1")
+	if err != nil {
+		t.Fatalf("get session after stop: %v", err)
+	}
+	if after.State != models.TaskSessionStateCancelled {
+		t.Fatalf("session state = %q, want CANCELLED", after.State)
+	}
+}
+
+// TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound
+// pins the unchanged path: a task with neither an active session nor a
+// registered execution still returns ErrExecutionNotFound and changes
+// nothing; a repeated stop after the executions are gone creates no
+// duplicate transition.
+func TestStopByTaskID_NoActiveSessionAndNoRegisteredExecutionReportsNotFound(t *testing.T) {
+	repo := newMockRepository()
+	manager := &mockAgentManager{}
+	exec := newTestExecutor(t, manager, repo)
+
+	err := exec.StopByTaskID(context.Background(), "task-empty", "cleanup", true)
+	if !errors.Is(err, ErrExecutionNotFound) {
+		t.Fatalf("error = %v, want ErrExecutionNotFound", err)
+	}
+
+	// Repeated stop after the (nonexistent) executions are gone must behave
+	// identically — no duplicate transition to create.
+	err = exec.StopByTaskID(context.Background(), "task-empty", "cleanup", true)
+	if !errors.Is(err, ErrExecutionNotFound) {
+		t.Fatalf("repeated stop error = %v, want ErrExecutionNotFound", err)
+	}
+}
+
+// TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound covers
+// the registry-recovery path's own failure mode: a session is registered in
+// the execution registry (so a real orphan exists) but its database row
+// fails to load (e.g. a transient "database is locked"). The active-session
+// query legitimately returns nothing, so if the load failure were silently
+// swallowed the same way a missing row is, StopByTaskID would report
+// ErrExecutionNotFound — indistinguishable from "there is nothing to stop" —
+// which is exactly the reachability defect this recovery path exists to
+// eliminate. It must surface the load failure instead.
+func TestStopByTaskID_RegistryRecoveryLoadFailureIsNotReportedAsNotFound(t *testing.T) {
+	repo := newMockRepository()
+	loadFailure := errors.New("database is locked")
+	repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) {
+		return nil, loadFailure
+	}
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-orphan"}}
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true)
+	if err == nil {
+		t.Fatal("StopByTaskID: want an error, got nil")
+	}
+	if errors.Is(err, ErrExecutionNotFound) {
+		t.Fatalf("error = %v, want the load failure surfaced instead of ErrExecutionNotFound", err)
+	}
+	if !errors.Is(err, loadFailure) {
+		t.Fatalf("error = %v, want it to wrap %v", err, loadFailure)
+	}
+}
+
+// TestStopByTaskID_MixedActiveAndUnloadableOrphanSurfacesLoadError covers
+// Finding 2 from Review Round 1: a task with one active session (which stops
+// cleanly) plus one registry-only orphan whose row fails to load must not
+// return a clean nil. The retry contract on the never-started teardown's
+// failed-attempt path depends on StopByTaskID surfacing that load failure so
+// a later stop keeps retrying instead of reporting a false all-clear.
+func TestStopByTaskID_MixedActiveAndUnloadableOrphanSurfacesLoadError(t *testing.T) {
+	repo := newMockRepository()
+	loadFailure := errors.New("database is locked")
+	activeSession := &models.TaskSession{
+		ID: "session-active", TaskID: "task-1", State: models.TaskSessionStateRunning,
+	}
+	repo.sessions[activeSession.ID] = activeSession
+	repo.listActiveTaskSessionsByTaskIDFunc = func(_ context.Context, taskID string) ([]*models.TaskSession, error) {
+		if taskID != "task-1" {
+			return nil, nil
+		}
+		return []*models.TaskSession{activeSession}, nil
+	}
+	repo.getTaskSessionFunc = func(_ context.Context, id string) (*models.TaskSession, error) {
+		if id == "session-orphan" {
+			return nil, loadFailure
+		}
+		if s, ok := repo.sessions[id]; ok {
+			return s, nil
+		}
+		return nil, nil
+	}
+	stopCalls := make(chan string, 2)
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{
+				{SessionID: "session-active", ExecutionID: "execution-active"},
+				{SessionID: "session-orphan", ExecutionID: "execution-orphan"},
+			}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-active", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true)
+	if err == nil {
+		t.Fatal("StopByTaskID: want an error, got nil")
+	}
+	if !errors.Is(err, loadFailure) {
+		t.Fatalf("error = %v, want it to wrap %v", err, loadFailure)
+	}
+
+	stopped := make(map[string]bool, 2)
+	for range 2 {
+		select {
+		case executionID := <-stopCalls:
+			stopped[executionID] = true
+		case <-time.After(2 * time.Second):
+			t.Fatal("expected both active and orphan executions to be stopped despite the orphan load failure")
+		}
+	}
+	if !stopped["execution-active"] {
+		t.Fatalf("stopped executions = %#v, want execution-active", stopped)
+	}
+	if !stopped["execution-orphan"] {
+		t.Fatalf("stopped executions = %#v, want execution-orphan", stopped)
+	}
+}
+
 func TestStopSessionDetailed_RejectsInvalidSession(t *testing.T) {
 	exec := newTestExecutor(t, &mockAgentManager{}, newMockRepository())
 
@@ -440,6 +677,52 @@ func TestStopWithSession_ClaimSuppressesLateCancelledLaunchCleanup(t *testing.T)
 	}
 }
 
+// TestStopWithSession_RegistersStopOwnerBeforeCancelledWrite is a regression
+// test for Finding 1 from Review Round 1: a concurrent StartAgentProcess
+// failure claims forced cleanup for itself only when no stop owner is
+// registered yet, and it classifies CANCELLED as terminal as soon as the row
+// is visible. If the CANCELLED write happened before stop-owner registration,
+// that window would let the concurrent failure win the claim and issue its
+// own teardown alongside this one's. The owner must be registered first.
+func TestStopWithSession_RegistersStopOwnerBeforeCancelledWrite(t *testing.T) {
+	repo := newMockRepository()
+	session := &models.TaskSession{
+		ID: "session-order", TaskID: "task-order", State: models.TaskSessionStateRunning,
+	}
+	repo.sessions[session.ID] = session
+
+	var order []string
+	repo.updateTaskSessionStateFunc = func(_ context.Context, sessionID string, state models.TaskSessionState, errorMessage string) error {
+		order = append(order, "cancelled-write")
+		if s, ok := repo.sessions[sessionID]; ok {
+			s.State = state
+			s.ErrorMessage = errorMessage
+		}
+		return nil
+	}
+
+	manager := &mockAgentManager{
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-order", nil
+		},
+		stopAgentWithReasonFunc: func(context.Context, string, string, bool) error {
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+	exec.SetOnExecutionStopOwnerRegistration(func(string, string, bool) {
+		order = append(order, "register-owner")
+	})
+
+	if err := exec.stopWithSession(context.Background(), session, "test stop", false); err != nil {
+		t.Fatalf("stopWithSession: %v", err)
+	}
+
+	if len(order) != 2 || order[0] != "register-owner" || order[1] != "cancelled-write" {
+		t.Fatalf("operation order = %v, want [register-owner cancelled-write]", order)
+	}
+}
+
 func TestStopExecution_PreservesRuntimeFailureClassification(t *testing.T) {
 	runtimeFailure := errors.New("runtime teardown failed")
 	tests := []struct {

apps/backend/internal/orchestrator/executor/executor_mocks_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kandev/kandev/internal/agent/agents"
 	agentdto "github.com/kandev/kandev/internal/agent/dto"
 	"github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/task/models"
@@ -29,6 +30,7 @@ type mockAgentManager struct {
 	setExecutionDescriptionFunc      func(ctx context.Context, agentExecutionID string, description string) error
 	setExecutionEnvFunc              func(ctx context.Context, agentExecutionID string, env map[string]string) error
 	getExecutionIDForSessionFunc     func(ctx context.Context, sessionID string) (string, error)
+	listExecutionsForTaskFunc        func(taskID string) []lifecycle.ExecutionReference
 	isAgentCommandConfiguredFunc     func(agentExecutionID string) bool
 	isAgentRunningForSessionFunc     func(ctx context.Context, sessionID string) bool
 	cleanupStaleExecutionFunc        func(ctx context.Context, sessionID string) error
@@ -227,6 +229,12 @@ func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, session
 	}
 	return "", fmt.Errorf("no execution found for session %s", sessionID)
 }
+func (m *mockAgentManager) ListExecutionsForTask(taskID string) []lifecycle.ExecutionReference {
+	if m.listExecutionsForTaskFunc != nil {
+		return m.listExecutionsForTaskFunc(taskID)
+	}
+	return nil
+}
 
 func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) {
 	if m.resolveAgentProfileFunc != nil {

apps/backend/internal/orchestrator/executor/executor_stop_recovery_identity_test.go

@@ -0,0 +1,166 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// TestStopByTaskID_DoesNotStopReplacementExecution reproduces the registry
+// recovery race: the execution seen during recovery is replaced before the
+// stop path resolves the session again. Recovery must keep the original
+// execution identity instead of stopping the replacement.
+func TestStopByTaskID_DoesNotStopReplacementExecution(t *testing.T) {
+	repo := newMockRepository()
+	repo.sessions["session-1"] = &models.TaskSession{
+		ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateFailed,
+	}
+
+	stopCalls := make(chan string, 1)
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			// Model the old execution being removed and a replacement being
+			// registered before the recovery path performs its fresh lookup.
+			return "execution-replacement", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopByTaskID: %v", err)
+	}
+
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-original" {
+			t.Fatalf("stopped execution = %q, want execution-original", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("registry-recovered execution was not stopped")
+	}
+}
+
+// TestStopByTaskID_UnloadableRecoveryDoesNotStopReplacementExecution covers
+// the same identity race when the session row cannot be read. The captured
+// execution must remain reachable even when the session index now points at a
+// replacement.
+func TestStopByTaskID_UnloadableRecoveryDoesNotStopReplacementExecution(t *testing.T) {
+	loadFailure := errors.New("database is locked")
+	repo := newMockRepository()
+	repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) {
+		return nil, loadFailure
+	}
+
+	stopCalls := make(chan string, 1)
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-replacement", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true)
+	if !errors.Is(err, loadFailure) {
+		t.Fatalf("StopByTaskID error = %v, want it to wrap %v", err, loadFailure)
+	}
+
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-original" {
+			t.Fatalf("stopped execution = %q, want execution-original", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("registry-recovered execution was not stopped")
+	}
+}
+
+// TestStopByTaskID_CancelsSessionThatBecomesActiveDuringRecovery models a
+// session that changes from terminal to active after the initial recovery
+// read. When the captured execution still owns it, recovery must use the
+// normal CANCELLED transition before scheduling teardown.
+func TestStopByTaskID_CancelsSessionThatBecomesActiveDuringRecovery(t *testing.T) {
+	repo := newMockRepository()
+	repo.sessions["session-1"] = &models.TaskSession{
+		ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateFailed,
+	}
+	var reads atomic.Int32
+	repo.getTaskSessionFunc = func(_ context.Context, sessionID string) (*models.TaskSession, error) {
+		read := reads.Add(1)
+		switch read {
+		case 1:
+			return &models.TaskSession{
+				ID: sessionID, TaskID: "task-1", State: models.TaskSessionStateFailed,
+			}, nil
+		case 2, 3:
+			return &models.TaskSession{
+				ID: sessionID, TaskID: "task-1", State: models.TaskSessionStateRunning,
+			}, nil
+		default:
+			repo.mu.Lock()
+			defer repo.mu.Unlock()
+			return cloneMockTaskSession(repo.sessions[sessionID]), nil
+		}
+	}
+
+	stopCalls := make(chan string, 1)
+	manager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-1", ExecutionID: "execution-original"}}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-original", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+
+	if err := exec.StopByTaskID(context.Background(), "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopByTaskID: %v", err)
+	}
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-original" {
+			t.Fatalf("stopped execution = %q, want execution-original", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("registry-recovered execution was not stopped")
+	}
+
+	repo.mu.Lock()
+	state := repo.sessions["session-1"].State
+	repo.mu.Unlock()
+	if state != models.TaskSessionStateCancelled {
+		t.Fatalf("session state = %q, want CANCELLED", state)
+	}
+}

apps/backend/internal/orchestrator/scheduler/scheduler_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/jmoiron/sqlx"
 	"github.com/kandev/kandev/internal/agent/agents"
 	"github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/db"
@@ -153,6 +154,9 @@ func (m *mockAgentManager) EnsureWorkspaceExecutionForSession(ctx context.Contex
 func (m *mockAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) {
 	return "", fmt.Errorf("no execution found")
 }
+func (m *mockAgentManager) ListExecutionsForTask(_ string) []lifecycle.ExecutionReference {
+	return nil
+}
 
 func (m *mockAgentManager) CancelAgent(ctx context.Context, sessionID string) error {
 	return nil

apps/backend/internal/orchestrator/task_operations.go

@@ -899,7 +899,7 @@ func (s *Service) handleSessionLaunchFailure(
 ) error {
 	failureCtx := context.WithoutCancel(ctx)
 	safeErr := routingerr.SanitizeError(launchErr)
-	s.recordSessionLaunchFailure(
+	_ = s.recordSessionLaunchFailure(
 		failureCtx, taskID, sessionID, safeErr, preloadedSession...,
 	)
 	return safeErr
@@ -908,13 +908,17 @@ func (s *Service) handleSessionLaunchFailure(
 // recordSessionLaunchFailure transitions a still-active session to FAILED and
 // updates the task only while the same failed session still owns it. Typed
 // launch-error persistence belongs to the executor's session transition.
-func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error, preloadedSession ...*models.TaskSession) {
+// Returns whether the session itself was durably transitioned to FAILED — a
+// caller that is about to force-stop the process on the strength of that
+// transition (stopNeverStartedExecution) must not do so when this reports
+// false, or it kills the process while the session row still claims RUNNING.
+func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error, preloadedSession ...*models.TaskSession) bool {
 	_, changed := s.updateTaskSessionStateWithHook(
 		ctx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error(), false,
 		nil, preloadedSession...,
 	)
 	if !changed {
-		return
+		return false
 	}
 	updated, err := s.updateTaskStateForEarlyLaunchFailure(ctx, taskID, sessionID)
 	if err != nil {
@@ -922,12 +926,13 @@ func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessio
 			zap.String("task_id", taskID),
 			zap.String("session_id", sessionID),
 			zap.Error(err))
-		return
+		return true
 	}
 	if !updated {
-		return
+		return true
 	}
 	s.processParentChildrenCompletedForTaskState(ctx, taskID, v1.TaskStateFailed)
+	return true
 }
 
 func (s *Service) updateTaskStateForEarlyLaunchFailure(
@@ -3623,7 +3628,15 @@ func (s *Service) StopTask(ctx context.Context, taskID string, reason string, fo
 
 	// Stop all agents for this task
 	if err := s.executor.StopByTaskID(ctx, taskID, reason, force); err != nil {
-		return err
+		if !errors.Is(err, executor.ErrOrphanRecoveryIncomplete) {
+			return err
+		}
+		// Every session that could be reached did stop; only a registry-only
+		// orphan's row failed to load. Log it and still transition to REVIEW
+		// rather than surfacing a false "failed to stop task" to the user.
+		s.logger.Warn("task stop reached REVIEW with an orphan recovery load failure",
+			zap.String("task_id", taskID),
+			zap.Error(err))
 	}
 
 	// Move task to REVIEW state for user review
@@ -3796,7 +3809,17 @@ func (s *Service) CancelTaskExecution(ctx context.Context, taskID string, reason
 		zap.String("task_id", taskID),
 		zap.String("reason", reason),
 		zap.Bool("force", force))
-	return s.executor.StopByTaskID(ctx, taskID, reason, force)
+	err := s.executor.StopByTaskID(ctx, taskID, reason, force)
+	if err != nil && errors.Is(err, executor.ErrOrphanRecoveryIncomplete) {
+		// Every session that could be reached did stop; only a registry-only
+		// orphan's row failed to load. Log it rather than reporting a false
+		// cancellation failure to office tree controls.
+		s.logger.Warn("task execution cancelled with an orphan recovery load failure",
+			zap.String("task_id", taskID),
+			zap.Error(err))
+		return nil
+	}
+	return err
 }
 
 // CancelTaskExecutionSynchronously stops every active session for a task and

apps/backend/internal/orchestrator/task_operations_stop_orphan_recovery_test.go

@@ -0,0 +1,221 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	orchestratorexec "github.com/kandev/kandev/internal/orchestrator/executor"
+	"github.com/kandev/kandev/internal/task/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+// drainStopCalls waits for exactly n buffered StopAgentWithReason calls,
+// discarding their execution IDs. StopByTaskID schedules each stop on a
+// detached goroutine, so a test that does not wait for every call it expects
+// leaves that goroutine still in flight when the test returns — caught by
+// this package's goleak.VerifyTestMain as a leak even though the send itself
+// would have succeeded.
+func drainStopCalls(t *testing.T, ch <-chan string, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		select {
+		case <-ch:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("timed out waiting for stop call %d/%d", i+1, n)
+		}
+	}
+}
+
+// TestStopTask_OrphanRecoveryLoadFailureStillReachesReview covers Review
+// round 2 Finding 1: a task with one active session that stops cleanly plus
+// one registry-only orphan whose row fails to load must still transition the
+// task to REVIEW instead of surfacing a false "failed to stop task" to the
+// caller, while the load failure stays observable (StopByTaskID's wrapped
+// error is still returned to whoever asks for it directly, even though
+// StopTask itself does not propagate it as a hard failure). It also covers
+// the follow-up fix that the orphan's captured execution is itself targeted
+// for stop, not just skipped because its row could not load.
+func TestStopTask_OrphanRecoveryLoadFailureStillReachesReview(t *testing.T) {
+	ctx := context.Background()
+	baseRepo := setupTestRepo(t)
+	seedTaskAndSession(t, baseRepo, "task-1", "session-active", models.TaskSessionStateRunning)
+
+	loadFailure := errors.New("database is locked")
+	hookedRepo := &coordinatorStopRepoHooks{
+		repoStore: baseRepo,
+		getSessionFunc: func(readCtx context.Context, sessionID string) (*models.TaskSession, error) {
+			if sessionID == "session-orphan" {
+				return nil, loadFailure
+			}
+			return baseRepo.GetTaskSession(readCtx, sessionID)
+		},
+	}
+
+	stopCalls := make(chan string, 8)
+	agentManager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{
+				{SessionID: "session-active", ExecutionID: "execution-active"},
+				{SessionID: "session-orphan", ExecutionID: "execution-orphan"},
+			}
+		},
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return "execution-active", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager)
+
+	if err := svc.StopTask(ctx, "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopTask returned an error despite the active session stopping cleanly: %v", err)
+	}
+
+	// The active session stops normally, and the registry-only orphan is now
+	// also targeted by its captured execution ID despite its row failing to
+	// load.
+	drainStopCalls(t, stopCalls, 2)
+
+	taskRepo.mu.Lock()
+	taskState := taskRepo.updatedStates["task-1"]
+	taskRepo.mu.Unlock()
+	if taskState != v1.TaskStateReview {
+		t.Fatalf("task state = %q, want REVIEW despite the orphan load failure", taskState)
+	}
+
+	// The load failure itself must still be observable to a direct caller of
+	// the executor, so a retry loop built on StopByTaskID does not see a false
+	// all-clear.
+	directErr := svc.executor.StopByTaskID(ctx, "task-1", "cleanup", true)
+	if !errors.Is(directErr, loadFailure) {
+		t.Fatalf("StopByTaskID error = %v, want it to still wrap the load failure", directErr)
+	}
+	if !errors.Is(directErr, orchestratorexec.ErrOrphanRecoveryIncomplete) {
+		t.Fatalf("StopByTaskID error = %v, want it to wrap ErrOrphanRecoveryIncomplete", directErr)
+	}
+	// The retry resolves session-active (now CANCELLED, terminal) through
+	// registry recovery instead of the active-state query, plus the
+	// still-unloadable orphan; both are targeted again.
+	drainStopCalls(t, stopCalls, 2)
+}
+
+// TestStopTask_UnloadableOrphanReachesReviewAfterScheduling covers a task
+// with no active rows and one unloadable registry orphan. The exact registry
+// reference is enough to schedule teardown, so StopTask reaches REVIEW while
+// still retaining the load failure for a direct StopByTaskID caller.
+func TestStopTask_UnloadableOrphanReachesReviewAfterScheduling(t *testing.T) {
+	ctx := context.Background()
+	baseRepo := setupTestRepo(t)
+
+	loadFailure := errors.New("database is locked")
+	hookedRepo := &coordinatorStopRepoHooks{
+		repoStore: baseRepo,
+		getSessionFunc: func(context.Context, string) (*models.TaskSession, error) {
+			return nil, loadFailure
+		},
+	}
+	stopCalls := make(chan string, 1)
+	agentManager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-orphan", ExecutionID: "execution-orphan"}}
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager)
+
+	if err := svc.StopTask(ctx, "task-1", "cleanup", true); err != nil {
+		t.Fatalf("StopTask: %v", err)
+	}
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-orphan" {
+			t.Fatalf("stopped execution = %q, want execution-orphan", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("unloadable orphan execution was not scheduled for stop")
+	}
+
+	taskRepo.mu.Lock()
+	state := taskRepo.updatedStates["task-1"]
+	taskRepo.mu.Unlock()
+	if state != v1.TaskStateReview {
+		t.Fatalf("task state = %q, want REVIEW", state)
+	}
+}
+
+// TestStopByTaskID_StopsOrphanExecutionWhenSessionRowFailsToLoad covers the
+// review finding that a registry-only orphan whose session row cannot be
+// loaded must still have its captured execution targeted for stop. A load
+// failure on the row must not make the process itself unreachable. Before this
+// fix, StopByTaskID skipped the orphan entirely on a load failure and never
+// called StopAgentWithReason for it.
+func TestStopByTaskID_StopsOrphanExecutionWhenSessionRowFailsToLoad(t *testing.T) {
+	ctx := context.Background()
+	baseRepo := setupTestRepo(t)
+
+	loadFailure := errors.New("database is locked")
+	hookedRepo := &coordinatorStopRepoHooks{
+		repoStore: baseRepo,
+		getSessionFunc: func(context.Context, string) (*models.TaskSession, error) {
+			return nil, loadFailure
+		},
+	}
+
+	stopCalls := make(chan string, 1)
+	agentManager := &mockAgentManager{
+		listExecutionsForTaskFunc: func(taskID string) []lifecycle.ExecutionReference {
+			if taskID != "task-1" {
+				return nil
+			}
+			return []lifecycle.ExecutionReference{{SessionID: "session-orphan", ExecutionID: "execution-orphan"}}
+		},
+		getExecutionIDForSessionFunc: func(_ context.Context, sessionID string) (string, error) {
+			if sessionID != "session-orphan" {
+				return "", errors.New("unexpected session ID")
+			}
+			return "execution-orphan", nil
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, executionID, _ string, _ bool) error {
+			stopCalls <- executionID
+			return nil
+		},
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
+	svc := newCoordinatorStopTestService(hookedRepo, taskRepo, agentManager)
+
+	err := svc.executor.StopByTaskID(ctx, "task-1", "cleanup", true)
+	if !errors.Is(err, loadFailure) {
+		t.Fatalf("StopByTaskID error = %v, want it to wrap the load failure", err)
+	}
+	if !errors.Is(err, orchestratorexec.ErrOrphanRecoveryIncomplete) {
+		t.Fatalf("StopByTaskID error = %v, want it to wrap ErrOrphanRecoveryIncomplete since the execution itself was reached", err)
+	}
+
+	select {
+	case executionID := <-stopCalls:
+		if executionID != "execution-orphan" {
+			t.Fatalf("stopped execution = %q, want execution-orphan", executionID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("the registry-only orphan's captured execution was never stopped")
+	}
+}