fix: keep tasks active while sibling sessions run (#1551)

* fix: keep tasks active while sibling sessions run

* fix: avoid duplicate task review writes

* fix: clear completed sessions before review guard

* fix: serialize session launch state

* test: cover completed sibling session states

* fix: bound completed execution markers

* fix: amortize completed execution marker cleanup

* fix: fail closed on stale session starts

* fix: ignore stale completed execution events

* test: cover stale completed tool calls

* fix: harden stale session state guards

* fix: handle terminal execution races

* fix: avoid duplicate terminal complete side effects

* fix: harden terminal execution race guards

* fix: guard task review reconciliation

* fix: handle late terminal complete side effects

* fix: align review-state test guards

* fix: fail closed on stale review bridges

* fix: reduce review reconcile complexity

---------

Co-authored-by: root <root@kandev.ts.net>

apps/backend/internal/backendapp/turn_adapters.go

@@ -24,6 +24,10 @@ func (a *turnServiceAdapter) CompleteTurn(ctx context.Context, turnID string) er
 	return a.svc.CompleteTurn(ctx, turnID)
 }
 
+func (a *turnServiceAdapter) GetTurn(ctx context.Context, turnID string) (*models.Turn, error) {
+	return a.svc.GetTurn(ctx, turnID)
+}
+
 func (a *turnServiceAdapter) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) {
 	return a.svc.GetActiveTurn(ctx, sessionID)
 }

apps/backend/internal/integration/test_server_test.go

@@ -219,6 +219,10 @@ func (a *testTurnServiceAdapter) CompleteTurn(ctx context.Context, turnID string
 	return a.svc.CompleteTurn(ctx, turnID)
 }
 
+func (a *testTurnServiceAdapter) GetTurn(ctx context.Context, turnID string) (*models.Turn, error) {
+	return a.svc.GetTurn(ctx, turnID)
+}
+
 func (a *testTurnServiceAdapter) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) {
 	return a.svc.GetActiveTurn(ctx, sessionID)
 }

apps/backend/internal/office/service/event_subscribers.go

@@ -190,6 +190,7 @@ func (s *Service) RegisterEventSubscribers(eb bus.EventBus) error {
 type AgentTurnMessageData struct {
 	TaskID    string `json:"task_id"`
 	SessionID string `json:"session_id"`
+	TurnID    string `json:"turn_id"`
 	AgentText string `json:"agent_text"`
 	AgentID   string `json:"agent_id"`
 }
@@ -214,7 +215,12 @@ func (s *Service) handleAgentTurnMessageSaved(ctx context.Context, event *bus.Ev
 	// For streaming agents, Data.Text is empty because chunks were already
 	// drained. Fall back to the last agent message saved in session messages.
 	agentText := data.AgentText
-	if agentText == "" && data.SessionID != "" && s.taskWorkspace != nil {
+	if agentText == "" && data.TurnID != "" && s.taskWorkspace != nil {
+		if lastMsg, msgErr := s.taskWorkspace.GetLastAgentMessageForTurn(ctx, data.TurnID); msgErr == nil && lastMsg != "" {
+			agentText = lastMsg
+		}
+	}
+	if agentText == "" && data.TurnID == "" && data.SessionID != "" && s.taskWorkspace != nil {
 		if lastMsg, msgErr := s.taskWorkspace.GetLastAgentMessage(ctx, data.SessionID); msgErr == nil && lastMsg != "" {
 			agentText = lastMsg
 		}

apps/backend/internal/office/service/event_subscribers_fallback_test.go

@@ -15,7 +15,8 @@ import (
 // stubTaskWorkspace is a minimal implementation of service.TaskWorkspaceService
 // that returns a fixed message from GetLastAgentMessage.
 type stubTaskWorkspace struct {
-	lastMsg string
+	lastMsg       string
+	lastMsgByTurn map[string]string
 }
 
 func (s *stubTaskWorkspace) GetWorkspace(_ context.Context, _ string) (*taskmodels.Workspace, error) {
@@ -36,6 +37,9 @@ func (s *stubTaskWorkspace) DeleteTask(_ context.Context, _ string) error { retu
 func (s *stubTaskWorkspace) GetLastAgentMessage(_ context.Context, _ string) (string, error) {
 	return s.lastMsg, nil
 }
+func (s *stubTaskWorkspace) GetLastAgentMessageForTurn(_ context.Context, turnID string) (string, error) {
+	return s.lastMsgByTurn[turnID], nil
+}
 
 // newTestServiceWithTaskWorkspace creates a service+bus like newTestServiceWithBus
 // but injects the given TaskWorkspace stub.
@@ -86,3 +90,70 @@ func TestAutoPostAgentComment_FallbackPath(t *testing.T) {
 	}
 	t.Fatalf("expected session comment with body from GetLastAgentMessage, got %+v", comments)
 }
+
+func TestAutoPostAgentComment_TurnScopedFallbackPath(t *testing.T) {
+	stub := &stubTaskWorkspace{
+		lastMsg: "newer active turn reply",
+		lastMsgByTurn: map[string]string{
+			"turn-terminal": "terminal turn reply",
+		},
+	}
+	svc, eb := newTestServiceWithTaskWorkspace(t, stub)
+	ctx := context.Background()
+
+	createTestAgent(t, svc, "ws-1", "agent-turn-fallback")
+	taskID := createOfficeTask(t, svc, "ws-1", "agent-turn-fallback")
+
+	event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{
+		"task_id":    taskID,
+		"session_id": "sess-fallback",
+		"turn_id":    "turn-terminal",
+		"agent_text": "",
+		"agent_id":   "agent-turn-fallback",
+	})
+	if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil {
+		t.Fatalf("publish event: %v", err)
+	}
+
+	comments, err := svc.ListComments(ctx, taskID)
+	if err != nil {
+		t.Fatalf("list comments: %v", err)
+	}
+
+	for _, c := range comments {
+		if c.Source == "session" && c.AuthorType == "agent" && c.Body == "terminal turn reply" {
+			return
+		}
+	}
+	t.Fatalf("expected session comment with body from turn-scoped fallback, got %+v", comments)
+}
+
+func TestAutoPostAgentComment_TurnScopedFallbackMissDoesNotUseSessionMessage(t *testing.T) {
+	stub := &stubTaskWorkspace{lastMsg: "newer active turn reply"}
+	svc, eb := newTestServiceWithTaskWorkspace(t, stub)
+	ctx := context.Background()
+
+	createTestAgent(t, svc, "ws-1", "agent-turn-miss")
+	taskID := createOfficeTask(t, svc, "ws-1", "agent-turn-miss")
+
+	event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", map[string]string{
+		"task_id":    taskID,
+		"session_id": "sess-fallback",
+		"turn_id":    "turn-terminal",
+		"agent_text": "",
+		"agent_id":   "agent-turn-miss",
+	})
+	if err := eb.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil {
+		t.Fatalf("publish event: %v", err)
+	}
+
+	comments, err := svc.ListComments(ctx, taskID)
+	if err != nil {
+		t.Fatalf("list comments: %v", err)
+	}
+	for _, c := range comments {
+		if c.Source == "session" {
+			t.Fatalf("expected no session comment from session fallback when turn lookup misses, got %+v", comments)
+		}
+	}
+}

apps/backend/internal/office/service/service.go

@@ -106,6 +106,7 @@ type TaskWorkspaceService interface {
 	ListTasksByWorkspace(ctx context.Context, workspaceID, workflowID, repositoryID, query string, page, pageSize int, includeArchived, includeEphemeral, onlyEphemeral, excludeConfig bool) ([]*taskmodels.Task, int, error)
 	DeleteTask(ctx context.Context, id string) error
 	GetLastAgentMessage(ctx context.Context, sessionID string) (string, error)
+	GetLastAgentMessageForTurn(ctx context.Context, turnID string) (string, error)
 }
 
 // WorkspaceGroupCleaner removes Kandev-owned materialized task workspaces

apps/backend/internal/office/service/workspace_deletion_test.go

@@ -148,6 +148,10 @@ func (f *fakeWorkspaceTaskService) GetLastAgentMessage(context.Context, string)
 	return "", nil
 }
 
+func (f *fakeWorkspaceTaskService) GetLastAgentMessageForTurn(context.Context, string) (string, error) {
+	return "", nil
+}
+
 type fakeWorkspaceGroupCleaner struct {
 	svc                       *service.Service
 	groupID                   string

apps/backend/internal/orchestrator/event_handlers_agent.go

@@ -490,6 +490,7 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv
 	// Update scheduler and remove from queue
 	s.scheduler.HandleTaskCompleted(data.TaskID, true)
 	s.scheduler.RemoveTask(data.TaskID)
+	s.markExecutionCompleted(data.SessionID, data.AgentExecutionID)
 
 	// Check for workflow transition based on session's current step.
 	session, err := s.repo.GetTaskSession(ctx, data.SessionID)
@@ -533,8 +534,9 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv
 		return
 	}
 
+	s.completeTurnForSession(context.WithoutCancel(ctx), data.SessionID)
+
 	if s.sessionHasPendingClarification(ctx, data.SessionID) {
-		s.completeTurnForSession(context.WithoutCancel(ctx), data.SessionID)
 		s.logger.Info("deferring on_turn_complete on agent.completed while clarification is pending",
 			zap.String("task_id", data.TaskID),
 			zap.String("session_id", data.SessionID))
@@ -548,29 +550,18 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv
 
 	transitioned := s.processOnTurnCompleteViaEngine(ctx, data.TaskID, session)
 
-	// Agent-exit path: unlike the streaming complete event, this path does NOT
-	// itself complete the open turn or clear the session's RUNNING state — it
-	// relies entirely on the workflow engine's on_turn_complete transition.
-	// workflow_transitioned=false means no transition fired, so the session may
-	// still be RUNNING (only the task row moves to REVIEW below) and the chat UI
-	// keeps showing the agent as working. Pairs with the frontend [session:state]
-	// / [session:turns] trace — filter both by the same task_id.
-	s.logger.Debug("agent.completed turn-complete decision (session state not cleared by this path)",
+	// Agent-exit path: processOnTurnCompleteViaEngine handles normal
+	// on_turn_complete transitions. If it did not transition, ensure the
+	// completed session leaves RUNNING and let setSessionWaitingForInput perform
+	// the guarded task REVIEW reconciliation when needed.
+	s.logger.Debug("agent.completed turn-complete decision",
 		zap.String("task_id", data.TaskID),
 		zap.String("session_id", data.SessionID),
 		zap.String("session_state", string(session.State)),
 		zap.Bool("workflow_transitioned", transitioned))
 
-	// If no workflow transition occurred, move task to REVIEW state for user review
-	if !transitioned {
-		if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil {
-			s.logger.Error("failed to update task state to REVIEW",
-				zap.String("task_id", data.TaskID),
-				zap.Error(err))
-		} else {
-			s.logger.Info("task moved to REVIEW state after agent completion",
-				zap.String("task_id", data.TaskID))
-		}
+	if !transitioned && session.State != models.TaskSessionStateWaitingForInput {
+		s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session)
 	}
 
 	// Capture a git status snapshot before cleanup so it can be served
@@ -593,6 +584,8 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent
 		zap.String("agent_execution_id", data.AgentExecutionID),
 		zap.String("error_message", data.ErrorMessage))
 
+	s.markExecutionFailed(data.SessionID, data.AgentExecutionID)
+
 	// Transient provider errors (529 Overloaded) get a paced, visible
 	// retry-with-backoff before any red banner. This is the ONLY non-terminal
 	// failure path, so it runs before automation finalization below — otherwise
@@ -631,15 +624,11 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent
 		return
 	}
 
-	// No session — fall back to scheduler retry + task to REVIEW.
+	// No session — fall back to scheduler retry + task to REVIEW unless another
+	// session is still working.
 	s.scheduler.HandleTaskCompleted(data.TaskID, false)
 	s.scheduler.RetryTask(data.TaskID)
-
-	if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil {
-		s.logger.Error("failed to update task state to REVIEW after failure",
-			zap.String("task_id", data.TaskID),
-			zap.Error(err))
-	}
+	s.writeTaskReviewState(ctx, data.TaskID, data.SessionID)
 
 	go s.cleanupAgentExecution(data.AgentExecutionID, data.TaskID, data.SessionID)
 }
@@ -712,12 +701,8 @@ func (s *Service) handleResumeFailure(ctx context.Context, data watcher.AgentEve
 	// 3. Set session to WAITING_FOR_INPUT (not FAILED) so the user can interact.
 	s.updateTaskSessionState(ctx, data.TaskID, data.SessionID, models.TaskSessionStateWaitingForInput, "", false)
 
-	// 4. Ensure task is in REVIEW state.
-	if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil {
-		s.logger.Warn("failed to set task to REVIEW after resume failure",
-			zap.String("task_id", data.TaskID),
-			zap.Error(err))
-	}
+	// 4. Ensure task is in REVIEW state unless another session is still working.
+	s.writeTaskReviewState(ctx, data.TaskID, data.SessionID)
 
 	return true
 }
@@ -757,12 +742,8 @@ func (s *Service) handleRecoverableFailure(ctx context.Context, data watcher.Age
 	}
 	s.updateTaskSessionState(ctx, data.TaskID, data.SessionID, nextState, data.ErrorMessage, false)
 
-	// Ensure task is in REVIEW state.
-	if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil {
-		s.logger.Warn("failed to set task to REVIEW after recoverable failure",
-			zap.String("task_id", data.TaskID),
-			zap.Error(err))
-	}
+	// Ensure task is in REVIEW state unless another session is still working.
+	s.writeTaskReviewState(ctx, data.TaskID, data.SessionID)
 
 	// Clean up the agent execution.
 	go s.cleanupAgentExecution(data.AgentExecutionID, data.TaskID, data.SessionID)

apps/backend/internal/orchestrator/event_handlers_agent_clarification_test.go

@@ -19,6 +19,7 @@ type repoBackedTurnService struct {
 type testRepo interface {
 	CreateTurn(ctx context.Context, turn *models.Turn) error
 	CompleteTurn(ctx context.Context, id string) error
+	GetTurn(ctx context.Context, id string) (*models.Turn, error)
 	GetActiveTurnBySessionID(ctx context.Context, sessionID string) (*models.Turn, error)
 	UpdateTurn(ctx context.Context, turn *models.Turn) error
 }
@@ -39,6 +40,10 @@ func (s *repoBackedTurnService) CompleteTurn(ctx context.Context, turnID string)
 	return s.repo.CompleteTurn(ctx, turnID)
 }
 
+func (s *repoBackedTurnService) GetTurn(ctx context.Context, turnID string) (*models.Turn, error) {
+	return s.repo.GetTurn(ctx, turnID)
+}
+
 func (s *repoBackedTurnService) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) {
 	return s.repo.GetActiveTurnBySessionID(ctx, sessionID)
 }

apps/backend/internal/orchestrator/event_handlers_clarification.go

@@ -14,7 +14,6 @@ import (
 	"github.com/kandev/kandev/internal/events"
 	"github.com/kandev/kandev/internal/events/bus"
 	"github.com/kandev/kandev/internal/task/models"
-	v1 "github.com/kandev/kandev/pkg/api/v1"
 )
 
 // subscribeClarificationEvents subscribes to clarification-related events.
@@ -80,11 +79,7 @@ func (s *Service) handleClarificationStaleDismissed(ctx context.Context, event *
 	s.finalizeAutomationRunIfEphemeral(writeCtx, data.TaskID, data.SessionID, true, "")
 	transitioned := s.processOnTurnCompleteViaEngine(writeCtx, data.TaskID, session)
 	if !transitioned {
-		if err := s.taskRepo.UpdateTaskState(writeCtx, data.TaskID, v1.TaskStateReview); err != nil {
-			s.logger.Error("failed to update task state to REVIEW after stale-dismiss",
-				zap.String("task_id", data.TaskID),
-				zap.Error(err))
-		}
+		s.writeTaskReviewState(writeCtx, data.TaskID, data.SessionID)
 	}
 	return nil
 }

apps/backend/internal/orchestrator/event_handlers_clarification_test.go

@@ -244,6 +244,7 @@ func TestHandleClarificationStaleDismissed(t *testing.T) {
 			ID: "step1", WorkflowID: "wf1", Name: "Plan", Position: 0,
 		}
 		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 		svc := createEngineService(t, repo, stepGetter, &mockAgentManager{})
 		svc.taskRepo = taskRepo
 

apps/backend/internal/orchestrator/event_handlers_streaming.go

@@ -2,6 +2,7 @@ package orchestrator
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"go.uber.org/zap"
@@ -10,6 +11,7 @@ import (
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	"github.com/kandev/kandev/internal/events"
 	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/internal/orchestrator/sessionstate"
 	"github.com/kandev/kandev/internal/task/models"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 )
@@ -23,10 +25,29 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
 	taskID := payload.TaskID
 	sessionID := payload.SessionID
 	eventType := payload.Data.Type
+	terminalCompleteStream := false
 
-	// Any agent stream activity means the agent resumed after clarification.
-	// Cancel primary-path clarification watchdogs for this session.
-	s.cancelClarificationWatchdogsForSession(sessionID, eventType)
+	if eventType == agentEventComplete {
+		if marker, ok := s.terminalExecutionMarker(sessionID, payload.ExecutionID); ok {
+			if !marker.allowCompleteStream {
+				s.logger.Debug("ignoring complete stream event from terminal failed execution",
+					zap.String("task_id", taskID),
+					zap.String("session_id", sessionID),
+					zap.String("agent_execution_id", payload.ExecutionID))
+				return
+			}
+			terminalCompleteStream = true
+		}
+	} else if s.shouldDropCompletedExecutionStreamEvent(payload) {
+		return
+	}
+
+	if !terminalCompleteStream {
+		// Any live agent stream activity means the agent resumed after clarification.
+		// Cancel primary-path clarification watchdogs for this session. Late terminal
+		// completes are excluded because they belong to an already-finished execution.
+		s.cancelClarificationWatchdogsForSession(sessionID, eventType)
+	}
 
 	s.logger.Debug("handling agent stream event",
 		zap.String("task_id", taskID),
@@ -201,6 +222,9 @@ func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.Ag
 			zap.String("tool_call_id", payload.Data.ToolCallID))
 		return
 	}
+	if s.shouldDropCompletedExecutionStreamEvent(payload) {
+		return
+	}
 
 	if s.messageCreator != nil {
 		if err := s.messageCreator.CreateToolCallMessage(
@@ -229,7 +253,7 @@ func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.Ag
 		// flipped to IN_PROGRESS in lockstep — otherwise an out-of-turn tool
 		// event (e.g. a Monitor watcher firing after on_turn_complete moved
 		// the task to REVIEW) leaves session=RUNNING with task=REVIEW.
-		s.setSessionRunning(ctx, payload.TaskID, payload.SessionID)
+		s.setSessionRunningForExecution(ctx, payload.TaskID, payload.SessionID, payload.ExecutionID)
 	}
 }
 
@@ -240,9 +264,23 @@ func (s *Service) saveAgentTextIfPresent(ctx context.Context, payload *lifecycle
 	if payload.Data.Text == "" || payload.SessionID == "" {
 		return
 	}
+	s.saveAgentTextForTurn(ctx, payload, s.getActiveTurnID(payload.SessionID))
+}
+
+func (s *Service) saveAgentTextForTurn(ctx context.Context, payload *lifecycle.AgentStreamEventPayload, turnID string) {
+	if payload.Data.Text == "" || payload.SessionID == "" {
+		return
+	}
+	if turnID == "" {
+		s.logger.Debug("skipping agent text without a target turn",
+			zap.String("task_id", payload.TaskID),
+			zap.String("session_id", payload.SessionID),
+			zap.String("agent_execution_id", payload.ExecutionID))
+		return
+	}
 
 	if s.messageCreator != nil {
-		if err := s.messageCreator.CreateAgentMessage(ctx, payload.TaskID, payload.Data.Text, payload.SessionID, s.getActiveTurnID(payload.SessionID)); err != nil {
+		if err := s.messageCreator.CreateAgentMessage(ctx, payload.TaskID, payload.Data.Text, payload.SessionID, turnID); err != nil {
 			s.logger.Error("failed to create agent message",
 				zap.String("task_id", payload.TaskID),
 				zap.Error(err))
@@ -259,6 +297,10 @@ func (s *Service) saveAgentTextIfPresent(ctx context.Context, payload *lifecycle
 // The subscriber (office comment bridge) uses the task/session IDs to look up
 // the agent's last message and auto-post it as a task comment.
 func (s *Service) publishAgentTurnComplete(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) {
+	s.publishAgentTurnCompleteForTurn(ctx, payload, "")
+}
+
+func (s *Service) publishAgentTurnCompleteForTurn(ctx context.Context, payload *lifecycle.AgentStreamEventPayload, turnID string) {
 	if s.eventBus == nil || payload.TaskID == "" || payload.SessionID == "" {
 		return
 	}
@@ -271,6 +313,7 @@ func (s *Service) publishAgentTurnComplete(ctx context.Context, payload *lifecyc
 		"session_id": payload.SessionID,
 		"agent_text": payload.Data.Text,
 		"agent_id":   payload.AgentID,
+		"turn_id":    turnID,
 	}
 	event := bus.NewEvent(events.AgentTurnMessageSaved, "orchestrator", data)
 	if err := s.eventBus.Publish(ctx, events.AgentTurnMessageSaved, event); err != nil {
@@ -334,6 +377,9 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.
 			zap.String("tool_call_id", payload.Data.ToolCallID))
 		return
 	}
+	if s.shouldDropCompletedExecutionStreamEvent(payload) {
+		return
+	}
 
 	if s.messageCreator == nil {
 		return
@@ -369,11 +415,25 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.
 		// see comment in handleToolCallEvent for the REVIEW/RUNNING split bug.
 		if payload.Data.ToolStatus == agentEventComplete || payload.Data.ToolStatus == agentEventCompleted ||
 			payload.Data.ToolStatus == "success" || payload.Data.ToolStatus == agentEventError || payload.Data.ToolStatus == agentEventFailed {
-			s.setSessionRunning(ctx, payload.TaskID, payload.SessionID)
+			s.setSessionRunningForExecution(ctx, payload.TaskID, payload.SessionID, payload.ExecutionID)
 		}
 	}
 }
 
+func (s *Service) shouldDropCompletedExecutionStreamEvent(payload *lifecycle.AgentStreamEventPayload) bool {
+	if payload == nil || payload.ExecutionID == "" || payload.SessionID == "" {
+		return false
+	}
+	if !s.isExecutionCompleted(payload.SessionID, payload.ExecutionID) {
+		return false
+	}
+	s.logger.Debug("ignoring stream event from completed execution",
+		zap.String("task_id", payload.TaskID),
+		zap.String("session_id", payload.SessionID),
+		zap.String("agent_execution_id", payload.ExecutionID))
+	return true
+}
+
 // updateTaskSessionState transitions a session to nextState with guard checks.
 // When a preloadedSession is provided, its State is used for guard conditions (terminal-state
 // check, same-state check). This is an optimistic fast-path: between load and check another
@@ -540,6 +600,151 @@ func isTerminalSessionState(s models.TaskSessionState) bool {
 		s == models.TaskSessionStateCancelled
 }
 
+const completedExecutionRetention = 10 * time.Minute
+
+type terminalExecutionMarker struct {
+	expiresAt           time.Time
+	allowCompleteStream bool
+	turnID              string
+}
+
+func terminalExecutionKey(sessionID, executionID string) string {
+	return sessionID + "\x00" + executionID
+}
+
+func (s *Service) markExecutionCompleted(sessionID, executionID string) {
+	s.markTerminalExecution(sessionID, executionID, true)
+}
+
+func (s *Service) markExecutionFailed(sessionID, executionID string) {
+	s.markTerminalExecution(sessionID, executionID, false)
+}
+
+func (s *Service) markTerminalExecution(sessionID, executionID string, allowCompleteStream bool) {
+	if sessionID == "" || executionID == "" {
+		return
+	}
+	key := terminalExecutionKey(sessionID, executionID)
+	expiresAt := time.Now().Add(completedExecutionRetention)
+	s.completedExecutions.Store(key, terminalExecutionMarker{
+		expiresAt:           expiresAt,
+		allowCompleteStream: allowCompleteStream,
+		turnID:              s.currentTurnIDForSession(context.Background(), sessionID),
+	})
+	time.AfterFunc(completedExecutionRetention, func() {
+		s.deleteCompletedExecutionIfExpired(key, expiresAt)
+	})
+}
+
+func (s *Service) isExecutionCompleted(sessionID, executionID string) bool {
+	_, ok := s.terminalExecutionMarker(sessionID, executionID)
+	return ok
+}
+
+func (s *Service) terminalCompleteStreamMarker(sessionID, executionID string) (terminalExecutionMarker, bool) {
+	marker, ok := s.terminalExecutionMarker(sessionID, executionID)
+	return marker, ok && marker.allowCompleteStream
+}
+
+func (s *Service) terminalExecutionMarker(sessionID, executionID string) (terminalExecutionMarker, bool) {
+	if sessionID == "" || executionID == "" {
+		return terminalExecutionMarker{}, false
+	}
+	key := terminalExecutionKey(sessionID, executionID)
+	value, ok := s.completedExecutions.Load(key)
+	if !ok {
+		return terminalExecutionMarker{}, false
+	}
+	marker, ok := value.(terminalExecutionMarker)
+	if !ok || time.Now().After(marker.expiresAt) {
+		s.deleteCompletedExecutionIfExpired(key, marker.expiresAt)
+		return terminalExecutionMarker{}, false
+	}
+	return marker, true
+}
+
+func (s *Service) currentTurnIDForSession(ctx context.Context, sessionID string) string {
+	if sessionID == "" {
+		return ""
+	}
+	if turnIDVal, ok := s.activeTurns.Load(sessionID); ok {
+		if turnID, ok := turnIDVal.(string); ok && turnID != "" {
+			return turnID
+		}
+	}
+	if s.turnService == nil {
+		return ""
+	}
+	turn, err := s.turnService.GetActiveTurn(ctx, sessionID)
+	if err != nil || turn == nil {
+		return ""
+	}
+	return turn.ID
+}
+
+func (s *Service) deleteCompletedExecutionIfExpired(key string, expiresAt time.Time) {
+	value, ok := s.completedExecutions.Load(key)
+	if !ok {
+		return
+	}
+	current, ok := value.(terminalExecutionMarker)
+	if !ok || !current.expiresAt.After(expiresAt) {
+		s.completedExecutions.Delete(key)
+	}
+}
+
+func (s *Service) setSessionStarting(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error {
+	if session == nil {
+		return nil
+	}
+
+	var publishSession *models.TaskSession
+	var stateUpdatedAt *time.Time
+	var oldState models.TaskSessionState
+	if err := func() error {
+		s.taskRuntimeStateMu.Lock()
+		defer s.taskRuntimeStateMu.Unlock()
+
+		current, err := s.repo.GetTaskSession(ctx, session.ID)
+		if err != nil {
+			return err
+		}
+		allowedTerminalRecovery := !promoteTask &&
+			session.State == models.TaskSessionStateStarting &&
+			current.State == models.TaskSessionStateFailed
+		if isTerminalSessionState(current.State) && !allowedTerminalRecovery {
+			return fmt.Errorf("session %s is %s; cannot mark STARTING", session.ID, current.State)
+		}
+
+		oldState = current.State
+		if err := s.repo.UpdateTaskSession(ctx, session); err != nil {
+			return err
+		}
+
+		if oldState != session.State {
+			if refreshed, err := s.repo.GetTaskSession(ctx, session.ID); err == nil && refreshed != nil {
+				if !refreshed.UpdatedAt.IsZero() {
+					t := refreshed.UpdatedAt.UTC()
+					stateUpdatedAt = &t
+				}
+				publishSession = refreshed
+			}
+		}
+
+		if promoteTask {
+			s.writeTaskInProgressForRuntime(ctx, taskID, session.ID)
+		}
+		return nil
+	}(); err != nil {
+		return err
+	}
+
+	if publishSession != nil {
+		s.publishTaskSessionStateChanged(ctx, taskID, session.ID, oldState, session.State, session.ErrorMessage, stateUpdatedAt, publishSession)
+	}
+	return nil
+}
+
 func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, sessionID string, preloadedSession ...*models.TaskSession) {
 	// Resolve session up front so we can skip the redundant task-state write
 	// when the session was already WAITING_FOR_INPUT. Without this guard, every
@@ -558,7 +763,7 @@ func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, session
 			// write so a transient lookup failure doesn't drop a needed
 			// REVIEW transition.
 			s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false)
-			s.writeTaskReviewState(ctx, taskID)
+			s.writeTaskReviewState(ctx, taskID, sessionID)
 			return
 		}
 	}
@@ -574,20 +779,91 @@ func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, session
 		return
 	}
 
-	s.writeTaskReviewState(ctx, taskID)
+	s.writeTaskReviewState(ctx, taskID, sessionID)
 }
 
-func (s *Service) writeTaskReviewState(ctx context.Context, taskID string) {
-	if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview); err != nil {
+func (s *Service) writeTaskReviewState(ctx context.Context, taskID, completedSessionID string) {
+	// Task lookup errors fail closed so office/archived guards cannot be bypassed
+	// by a transient repository failure.
+	if dbTask, err := s.repo.GetTask(ctx, taskID); err != nil {
+		s.logger.Warn("failed to load task before REVIEW state reconcile",
+			zap.String("task_id", taskID),
+			zap.Error(err))
+		return
+	} else if dbTask != nil && dbTask.AssigneeAgentProfileID != "" {
+		s.logger.Debug("skipping REVIEW transition for office task",
+			zap.String("task_id", taskID))
+		return
+	}
+
+	s.taskRuntimeStateMu.Lock()
+	defer s.taskRuntimeStateMu.Unlock()
+
+	if completedSessionID != "" {
+		if session, err := s.repo.GetTaskSession(ctx, completedSessionID); err == nil && session != nil && isWorkingSessionState(session.State) {
+			s.logger.Debug("skipping task REVIEW state because completed session is active again",
+				zap.String("task_id", taskID),
+				zap.String("session_id", completedSessionID),
+				zap.String("session_state", string(session.State)))
+			return
+		}
+	}
+
+	if blockingSessionID, ok := s.otherWorkingSessionID(ctx, taskID, completedSessionID); !ok {
+		return
+	} else if blockingSessionID != "" {
+		s.logger.Debug("skipping task REVIEW state while another session is working",
+			zap.String("task_id", taskID),
+			zap.String("completed_session_id", completedSessionID),
+			zap.String("blocking_session_id", blockingSessionID))
+		return
+	}
+	updated, err := s.taskRepo.UpdateTaskStateIfCurrentIn(
+		ctx,
+		taskID,
+		v1.TaskStateReview,
+		[]v1.TaskState{v1.TaskStateInProgress, v1.TaskStateScheduling},
+	)
+	if err != nil {
 		s.logger.Error("failed to update task state to REVIEW",
 			zap.String("task_id", taskID),
 			zap.Error(err))
 		return
 	}
+	if !updated {
+		return
+	}
 	s.logger.Info("task moved to REVIEW state",
 		zap.String("task_id", taskID))
 }
 
+func isWorkingSessionState(state models.TaskSessionState) bool {
+	return sessionstate.IsWorking(state)
+}
+
+func (s *Service) otherWorkingSessionID(ctx context.Context, taskID, currentSessionID string) (string, bool) {
+	sessions, err := s.repo.ListTaskSessions(ctx, taskID)
+	if err != nil {
+		s.logger.Warn("failed to list task sessions before REVIEW state reconcile",
+			zap.String("task_id", taskID),
+			zap.String("session_id", currentSessionID),
+			zap.Error(err))
+		return "", false
+	}
+	for _, session := range sessions {
+		if session == nil {
+			continue
+		}
+		if currentSessionID != "" && session.ID == currentSessionID {
+			continue
+		}
+		if isWorkingSessionState(session.State) {
+			return session.ID, true
+		}
+	}
+	return "", true
+}
+
 // writeTaskReviewStateOnCancel clears the kanban "actively working" task
 // states when the user cancels a turn mid-flight by landing the task in
 // REVIEW — the same bucket a normal turn completion uses, so the sidebar
@@ -595,7 +871,7 @@ func (s *Service) writeTaskReviewState(ctx context.Context, taskID string) {
 // Office task status reflects workflow position, not runtime cancel, so those
 // tasks are left alone. Only actively-working tasks are reconciled; tasks
 // already past IN_PROGRESS / SCHEDULING keep their state.
-func (s *Service) writeTaskReviewStateOnCancel(ctx context.Context, taskID string) {
+func (s *Service) writeTaskReviewStateOnCancel(ctx context.Context, taskID, sessionID string) {
 	dbTask, err := s.repo.GetTask(ctx, taskID)
 	if err != nil || dbTask == nil {
 		if err != nil {
@@ -609,6 +885,29 @@ func (s *Service) writeTaskReviewStateOnCancel(ctx context.Context, taskID strin
 		return
 	}
 
+	s.taskRuntimeStateMu.Lock()
+	defer s.taskRuntimeStateMu.Unlock()
+
+	if sessionID != "" {
+		if session, err := s.repo.GetTaskSession(ctx, sessionID); err == nil && session != nil && isWorkingSessionState(session.State) {
+			s.logger.Debug("skipping task REVIEW state after cancel because session is active again",
+				zap.String("task_id", taskID),
+				zap.String("session_id", sessionID),
+				zap.String("session_state", string(session.State)))
+			return
+		}
+	}
+
+	if blockingSessionID, ok := s.otherWorkingSessionID(ctx, taskID, sessionID); !ok {
+		return
+	} else if blockingSessionID != "" {
+		s.logger.Debug("skipping task REVIEW state after cancel while another session is working",
+			zap.String("task_id", taskID),
+			zap.String("session_id", sessionID),
+			zap.String("blocking_session_id", blockingSessionID))
+		return
+	}
+
 	updated, err := s.taskRepo.UpdateTaskStateIfCurrentIn(
 		ctx,
 		taskID,
@@ -629,6 +928,13 @@ func (s *Service) writeTaskReviewStateOnCancel(ctx context.Context, taskID strin
 }
 
 func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID string, preloadedSession ...*models.TaskSession) {
+	s.setSessionRunningForExecution(ctx, taskID, sessionID, "", preloadedSession...)
+}
+
+func (s *Service) setSessionRunningForExecution(ctx context.Context, taskID, sessionID, executionID string, preloadedSession ...*models.TaskSession) {
+	s.taskRuntimeStateMu.Lock()
+	defer s.taskRuntimeStateMu.Unlock()
+
 	// Resolve session up front so we can guard the task write against terminal
 	// states. updateTaskSessionState silently no-ops for terminal sessions, so
 	// without this guard a buffered tool event arriving after a CANCELLED /
@@ -646,6 +952,13 @@ func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID strin
 	if isTerminalSessionState(session.State) {
 		return
 	}
+	if session.State == models.TaskSessionStateWaitingForInput && s.isExecutionCompleted(sessionID, executionID) {
+		s.logger.Debug("ignoring stream event for completed execution",
+			zap.String("task_id", taskID),
+			zap.String("session_id", sessionID),
+			zap.String("agent_execution_id", executionID))
+		return
+	}
 
 	// Skip the redundant task-state write when the session is already RUNNING.
 	// Tool calls fire many events per turn and each was triggering an
@@ -663,6 +976,24 @@ func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID strin
 		return
 	}
 
+	s.writeTaskInProgressForRuntime(ctx, taskID, sessionID)
+}
+
+func (s *Service) writeTaskInProgressForRuntime(ctx context.Context, taskID, sessionID string) {
+	// Task lookup errors fail closed so office/archived guards cannot be bypassed
+	// by a transient repository failure.
+	task, err := s.repo.GetTask(ctx, taskID)
+	if err != nil {
+		s.logger.Warn("skipping IN_PROGRESS transition because task lookup failed",
+			zap.String("task_id", taskID),
+			zap.Error(err))
+		return
+	}
+	if task != nil && task.ArchivedAt != nil {
+		s.logger.Debug("skipping IN_PROGRESS transition for archived task",
+			zap.String("task_id", taskID))
+		return
+	}
 	// Office tasks do NOT transition to IN_PROGRESS when their agent's
 	// session enters RUNNING. The user-facing task status (todo /
 	// in_review / done / blocked) reflects workflow position, not the
@@ -670,11 +1001,32 @@ func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID strin
 	// spinner and the inline session timeline entry instead. Skipping
 	// the transition prevents the REVIEW → IN_PROGRESS → REVIEW flicker
 	// across follow-up comment turns.
-	if dbTask, err := s.repo.GetTask(ctx, taskID); err == nil && dbTask != nil && dbTask.AssigneeAgentProfileID != "" {
+	if task != nil && task.AssigneeAgentProfileID != "" {
 		s.logger.Debug("skipping IN_PROGRESS transition for office task",
 			zap.String("task_id", taskID))
 		return
 	}
+	if sessionID != "" {
+		session, sessionErr := s.repo.GetTaskSession(ctx, sessionID)
+		if sessionErr != nil {
+			s.logger.Warn("skipping IN_PROGRESS transition because session lookup failed",
+				zap.String("task_id", taskID),
+				zap.String("session_id", sessionID),
+				zap.Error(sessionErr))
+			return
+		}
+		if session == nil || !sessionstate.IsWorking(session.State) {
+			state := ""
+			if session != nil {
+				state = string(session.State)
+			}
+			s.logger.Debug("skipping IN_PROGRESS transition because session is no longer active",
+				zap.String("task_id", taskID),
+				zap.String("session_id", sessionID),
+				zap.String("session_state", state))
+			return
+		}
+	}
 
 	if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil {
 		s.logger.Error("failed to update task state to IN_PROGRESS",
@@ -691,6 +1043,7 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy
 	s.logger.Debug("handling complete stream event",
 		zap.String("task_id", payload.TaskID),
 		zap.String("session_id", payload.SessionID))
+	terminalMarker, terminalCompleteStream := s.terminalCompleteStreamMarker(payload.SessionID, payload.ExecutionID)
 
 	// Load session once up front — used by storeResumeToken, state check, and setSessionWaitingForInput.
 	var session *models.TaskSession
@@ -717,10 +1070,27 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy
 		s.storeResumeToken(ctx, payload.TaskID, payload.SessionID, payload.ExecutionID, payload.Data.ACPSessionID, lastMsgUUID)
 	}
 
+	s.publishPromptUsage(ctx, payload, session)
+
+	if terminalCompleteStream {
+		s.saveAgentTextForTurn(ctx, payload, terminalMarker.turnID)
+		s.publishAgentPlanForTurn(ctx, payload, terminalMarker.turnID, false)
+		s.persistTurnPromptMetadataForTurn(ctx, payload, session, terminalMarker.turnID)
+		if terminalMarker.turnID != "" {
+			s.publishAgentTurnCompleteForTurn(ctx, payload, terminalMarker.turnID)
+		}
+		s.detachClarificationWaiters(ctx, payload.SessionID)
+		s.logger.Debug("complete stream from terminal execution flushed final data; skipping active turn and runtime reconciliation",
+			zap.String("task_id", payload.TaskID),
+			zap.String("session_id", payload.SessionID),
+			zap.String("agent_execution_id", payload.ExecutionID),
+			zap.String("turn_id", terminalMarker.turnID))
+		return
+	}
+
 	s.saveAgentTextIfPresent(ctx, payload)
 	s.publishAgentPlanIfPresent(ctx, payload)
 	s.persistTurnPromptMetadata(ctx, payload, session)
-	s.publishPromptUsage(ctx, payload, session)
 	s.completeTurnForSession(ctx, payload.SessionID)
 
 	// Publish agent turn message event so the office comment bridge can
@@ -732,13 +1102,7 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy
 
 	// Detach any pending clarifications so WaitForResponse unblocks while the
 	// overlay stays interactive for a deferred answer via the event fallback path.
-	if s.clarificationCanceller != nil && payload.SessionID != "" {
-		if n := s.clarificationCanceller.DetachSessionAndNotify(ctx, payload.SessionID); n > 0 {
-			s.logger.Info("detached pending clarifications on turn complete",
-				zap.String("session_id", payload.SessionID),
-				zap.Int("count", n))
-		}
-	}
+	s.detachClarificationWaiters(ctx, payload.SessionID)
 
 	// Capture a fresh git status snapshot on every turn completion so the sidebar
 	// diff badge stays current even when the agent remains running (the
@@ -803,6 +1167,17 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy
 	s.setSessionWaitingForInput(ctx, payload.TaskID, payload.SessionID, session)
 }
 
+func (s *Service) detachClarificationWaiters(ctx context.Context, sessionID string) {
+	if s.clarificationCanceller == nil || sessionID == "" {
+		return
+	}
+	if n := s.clarificationCanceller.DetachSessionAndNotify(ctx, sessionID); n > 0 {
+		s.logger.Info("detached pending clarifications on turn complete",
+			zap.String("session_id", sessionID),
+			zap.Int("count", n))
+	}
+}
+
 // sessionStateString renders a session's state for logging, returning "" when
 // the session is nil (e.g. a complete event with no session ID). Kept tiny so
 // state-transition trace logs don't add branching to their hot-path callers.
@@ -927,6 +1302,10 @@ func (s *Service) handleAgentPlanEvent(ctx context.Context, payload *lifecycle.A
 // publishAgentPlanIfPresent extracts plan_content from a complete event and creates
 // a dedicated agent_plan message in the session.
 func (s *Service) publishAgentPlanIfPresent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) {
+	s.publishAgentPlanForTurn(ctx, payload, "", true)
+}
+
+func (s *Service) publishAgentPlanForTurn(ctx context.Context, payload *lifecycle.AgentStreamEventPayload, turnID string, allowLazyTurn bool) {
 	if payload.SessionID == "" || payload.Data.Data == nil || s.messageCreator == nil {
 		return
 	}
@@ -940,9 +1319,15 @@ func (s *Service) publishAgentPlanIfPresent(ctx context.Context, payload *lifecy
 	}
 
 	sessionID := payload.SessionID
+	if turnID == "" && allowLazyTurn {
+		turnID = s.getActiveTurnID(sessionID)
+	}
+	if turnID == "" {
+		return
+	}
 	if err := s.messageCreator.CreateSessionMessage(
 		ctx, payload.TaskID, planContent, sessionID,
-		string(models.MessageTypeAgentPlan), s.getActiveTurnID(sessionID), nil, false,
+		string(models.MessageTypeAgentPlan), turnID, nil, false,
 	); err != nil {
 		s.logger.Error("failed to create agent plan message",
 			zap.String("task_id", payload.TaskID),
@@ -1021,6 +1406,38 @@ func (s *Service) persistTurnPromptMetadata(
 	if turn == nil {
 		return
 	}
+	s.persistPromptMetadataOnTurn(ctx, payload, session, turn)
+}
+
+func (s *Service) persistTurnPromptMetadataForTurn(
+	ctx context.Context,
+	payload *lifecycle.AgentStreamEventPayload,
+	session *models.TaskSession,
+	turnID string,
+) {
+	if payload == nil || payload.Data == nil || payload.SessionID == "" || payload.Data.Usage == nil || s.turnService == nil || turnID == "" {
+		return
+	}
+	turn, err := s.turnService.GetTurn(ctx, turnID)
+	if err != nil {
+		s.logger.Warn("failed to get terminal turn for prompt usage metadata",
+			zap.String("turn_id", turnID),
+			zap.String("session_id", payload.SessionID),
+			zap.Error(err))
+		return
+	}
+	if turn == nil {
+		return
+	}
+	s.persistPromptMetadataOnTurn(ctx, payload, session, turn)
+}
+
+func (s *Service) persistPromptMetadataOnTurn(
+	ctx context.Context,
+	payload *lifecycle.AgentStreamEventPayload,
+	session *models.TaskSession,
+	turn *models.Turn,
+) {
 	model, agentType := resolvePromptUsageLabels(payload, session)
 	metadata := turn.Metadata
 	if metadata == nil {

apps/backend/internal/orchestrator/event_handlers_streaming_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kandev/kandev/internal/automation"
 	"github.com/kandev/kandev/internal/events"
 	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/internal/orchestrator/watcher"
 	"github.com/kandev/kandev/internal/task/models"
 	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
@@ -28,6 +29,23 @@ type recordedEvent struct {
 	event   *bus.Event
 }
 
+type recordingClarificationCanceller struct {
+	sessions []string
+}
+
+func (c *recordingClarificationCanceller) DetachSessionAndNotify(_ context.Context, sessionID string) int {
+	c.sessions = append(c.sessions, sessionID)
+	return 1
+}
+
+type listTaskSessionsErrorRepo struct {
+	sessionExecutorStore
+}
+
+func (r listTaskSessionsErrorRepo) ListTaskSessions(context.Context, string) ([]*models.TaskSession, error) {
+	return nil, errors.New("list task sessions failed")
+}
+
 type failSetSessionMetadataRepo struct {
 	repoStore
 }
@@ -499,6 +517,603 @@ func TestToolEventsWakeSessionAndTaskTogether(t *testing.T) {
 	}
 }
 
+func TestToolUpdateFromCompletedExecutionDoesNotWakeWaitingSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	seedExecutorRunning(t, repo, "s1", "t1", "exec-1")
+
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+	svc.messageCreator = &mockMessageCreator{}
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "exec-1",
+	})
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateWaitingForInput, updated.State)
+	require.Equal(t, v1.TaskStateReview, taskRepo.updatedStates["t1"])
+
+	svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			ToolCallID: "tc1",
+			ToolStatus: agentEventComplete,
+		},
+	})
+
+	updated, err = repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateWaitingForInput, updated.State,
+		"late tool events from the completed execution must not revive the session")
+	require.Equal(t, 1, taskRepo.stateWrites["t1"],
+		"late completed-execution tool event must not move the task back to IN_PROGRESS")
+}
+
+func TestToolUpdateFromCompletedExecutionDoesNotCreateMessage(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			ToolCallID: "tc1",
+			ToolStatus: agentEventComplete,
+		},
+	})
+
+	require.Zero(t, messages.toolUpdateWrites,
+		"stale completed-execution tool updates must be dropped before message side effects")
+}
+
+func TestToolCallFromCompletedExecutionDoesNotCreateMessage(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleToolCallEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			ToolCallID: "tc1",
+			ToolStatus: "running",
+		},
+	})
+
+	require.Zero(t, messages.toolCallWrites,
+		"stale completed-execution tool calls must be dropped before message side effects")
+}
+
+func TestToolCallStreamFromCompletedExecutionDoesNotSaveAgentText(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks["t1"] = &v1.Task{ID: "t1", State: v1.TaskStateInProgress}
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:       agentEventToolCall,
+			Text:       "stale text from completed execution",
+			ToolCallID: "tc1",
+			ToolStatus: "running",
+		},
+	})
+
+	require.Zero(t, messages.agentMessageWrites,
+		"top-level tool_call guard must run before saveAgentTextIfPresent")
+	require.Zero(t, messages.toolCallWrites,
+		"top-level tool_call guard must not fall through to handleToolCallEvent")
+}
+
+func TestCompleteStreamFromCompletedExecutionFlushesAgentText(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	svc.turnService = &repoTurnService{repo: repo}
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+	firstTurn, err := svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+	svc.markExecutionCompleted("s1", "exec-1")
+	svc.completeTurnForSession(ctx, "s1")
+	nextTurn, err := svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+			Text: "final flushed text",
+		},
+	})
+
+	require.Equal(t, 1, messages.agentMessageWrites,
+		"final complete streams must flush text even if agent.completed arrived first")
+	require.Equal(t, firstTurn.ID, messages.agentMessages[0].turnID,
+		"late terminal complete must write to the turn that belonged to the terminal execution")
+	activeTurn, err := svc.turnService.GetActiveTurn(ctx, "s1")
+	require.NoError(t, err)
+	require.NotNil(t, activeTurn)
+	require.Equal(t, nextTurn.ID, activeTurn.ID,
+		"late terminal complete must not close a newer active turn")
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"final complete streams from terminal executions must not re-run task state reconciliation")
+}
+
+func TestCompleteStreamFromCompletedExecutionPersistsTerminalTurnMetadata(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	svc.turnService = &repoTurnService{repo: repo}
+	firstTurn, err := svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+	svc.markExecutionCompleted("s1", "exec-1")
+	svc.completeTurnForSession(ctx, "s1")
+	nextTurn, err := svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		AgentID:     "agent-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:           agentEventComplete,
+			CurrentModelID: "gpt-5.5",
+			Usage: &streams.PromptUsage{
+				InputTokens:                  10,
+				OutputTokens:                 20,
+				TotalTokens:                  42,
+				ProviderReportedCostSubcents: 123,
+			},
+		},
+	})
+
+	terminalTurn, err := repo.GetTurn(ctx, firstTurn.ID)
+	require.NoError(t, err)
+	require.Equal(t, "gpt-5.5", terminalTurn.Metadata["model"])
+	require.Equal(t, "agent-1", terminalTurn.Metadata["agent_id"])
+	usage, ok := terminalTurn.Metadata["prompt_usage"].(map[string]interface{})
+	require.True(t, ok)
+	require.Equal(t, float64(42), usage["total_tokens"])
+	require.Equal(t, float64(123), usage["provider_reported_cost_subcents"])
+
+	activeTurn, err := repo.GetTurn(ctx, nextTurn.ID)
+	require.NoError(t, err)
+	require.NotContains(t, activeTurn.Metadata, "prompt_usage",
+		"late terminal complete metadata must not be written to a newer active turn")
+}
+
+func TestCompleteStreamFromCompletedExecutionPublishesTerminalTurn(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	eb := &recordingEventBus{}
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+	svc.eventBus = eb
+	svc.turnService = &repoTurnService{repo: repo}
+	firstTurn, err := svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+	svc.markExecutionCompleted("s1", "exec-1")
+	svc.completeTurnForSession(ctx, "s1")
+	_, err = svc.turnService.StartTurn(ctx, "s1")
+	require.NoError(t, err)
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+		},
+	})
+
+	require.Len(t, eb.events, 1)
+	require.Equal(t, events.AgentTurnMessageSaved, eb.events[0].subject)
+	data, ok := eb.events[0].event.Data.(map[string]string)
+	require.True(t, ok)
+	require.Equal(t, firstTurn.ID, data["turn_id"],
+		"late terminal complete publish must identify the completed turn")
+}
+
+func TestCompleteStreamFromCompletedExecutionSkipsDuplicateOfficeTeardown(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedOfficeSession(t, repo, "t-office-terminal", "s-office-terminal", "exec-office-terminal")
+
+	taskRepo := newMockTaskRepo()
+	mgr := &mockAgentManager{}
+	mgr.stopAgentArgs = append(mgr.stopAgentArgs, stopAgentCall{ExecutionID: "exec-office-terminal"})
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, mgr)
+	svc.markExecutionCompleted("s-office-terminal", "exec-office-terminal")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t-office-terminal",
+		SessionID:   "s-office-terminal",
+		ExecutionID: "exec-office-terminal",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+			Data: map[string]interface{}{
+				"stop_reason": "end_turn",
+			},
+		},
+	})
+
+	mgr.mu.Lock()
+	stopCalls := append([]stopAgentCall(nil), mgr.stopAgentArgs...)
+	mgr.mu.Unlock()
+	require.Equal(t, []stopAgentCall{{ExecutionID: "exec-office-terminal"}}, stopCalls,
+		"terminal complete streams must not re-run office StopAgent teardown")
+	require.Zero(t, taskRepo.stateWrites["t-office-terminal"],
+		"terminal complete streams must not re-run task state reconciliation")
+}
+
+func TestCompleteStreamFromCompletedExecutionSkipsDuplicateAutomationFinalize(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedRunModeAutomationSession(t, repo, "t-auto-terminal", "s-auto-terminal", "exec-auto-terminal")
+
+	taskRepo := newMockTaskRepo()
+	mgr := &mockAgentManager{}
+	automationSvc := &mockAutomationRunService{succeededTaskIDs: []string{"t-auto-terminal"}}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, mgr)
+	svc.SetAutomationService(automationSvc)
+	svc.markExecutionCompleted("s-auto-terminal", "exec-auto-terminal")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t-auto-terminal",
+		SessionID:   "s-auto-terminal",
+		ExecutionID: "exec-auto-terminal",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+			Data: map[string]interface{}{
+				"stop_reason": "end_turn",
+			},
+		},
+	})
+
+	require.Equal(t, []string{"t-auto-terminal"}, automationSvc.succeededTaskIDs,
+		"terminal complete streams must not re-run automation success finalization")
+	require.Empty(t, automationSvc.failedTaskIDs)
+	mgr.mu.Lock()
+	stopCalls := append([]stopAgentCall(nil), mgr.stopAgentArgs...)
+	mgr.mu.Unlock()
+	require.Empty(t, stopCalls,
+		"terminal complete streams must not re-run automation StopAgent teardown")
+}
+
+func TestMessageStreamFromCompletedExecutionDoesNotCreateTurn(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	svc.turnService = &repoTurnService{repo: repo}
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:      "message_streaming",
+			MessageID: "msg-1",
+			Text:      "stale text from completed execution",
+		},
+	})
+
+	require.Zero(t, messages.agentStreamWrites,
+		"stale completed-execution message streams must be dropped before message side effects")
+	turn, err := svc.turnService.GetActiveTurn(ctx, "s1")
+	require.NoError(t, err)
+	require.Nil(t, turn, "stale stream events must not lazily create a new turn")
+}
+
+func TestCompletedExecutionStreamDoesNotCancelClarificationWatchdog(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+	watchCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	svc.clarificationWatchdogs.Store(
+		svc.clarificationWatchdogKey("s1", "pending-1"),
+		&clarificationWatchdogEntry{cancel: cancel},
+	)
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:       "tool_update",
+			ToolCallID: "tc1",
+			ToolStatus: agentEventComplete,
+		},
+	})
+
+	require.Equal(t, 1, countClarificationWatchdogs(svc),
+		"stale completed-execution streams must be dropped before watchdog cancellation")
+	select {
+	case <-watchCtx.Done():
+		t.Fatal("stale completed-execution stream cancelled clarification watchdog")
+	default:
+	}
+}
+
+func TestTerminalCompleteStreamDoesNotCancelClarificationWatchdog(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+	watchCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	svc.clarificationWatchdogs.Store(
+		svc.clarificationWatchdogKey("s1", "pending-1"),
+		&clarificationWatchdogEntry{cancel: cancel},
+	)
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+			Text: "late complete",
+		},
+	})
+
+	require.Equal(t, 1, countClarificationWatchdogs(svc),
+		"late terminal complete streams must not cancel clarification watchdogs")
+	select {
+	case <-watchCtx.Done():
+		t.Fatal("late terminal complete cancelled clarification watchdog")
+	default:
+	}
+}
+
+func TestTerminalCompleteStreamDetachesClarificationWaitersWithoutCancellingWatchdog(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+	canceller := &recordingClarificationCanceller{}
+	svc.clarificationCanceller = canceller
+	watchCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	svc.clarificationWatchdogs.Store(
+		svc.clarificationWatchdogKey("s1", "pending-1"),
+		&clarificationWatchdogEntry{cancel: cancel},
+	)
+	svc.markExecutionCompleted("s1", "exec-1")
+
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+		},
+	})
+
+	require.Equal(t, []string{"s1"}, canceller.sessions)
+	require.Equal(t, 1, countClarificationWatchdogs(svc),
+		"late terminal complete should detach waiters without cancelling watchdog fallback entries")
+	select {
+	case <-watchCtx.Done():
+		t.Fatal("late terminal complete cancelled clarification watchdog")
+	default:
+	}
+}
+
+func TestWriteTaskReviewStateSkipsWhenSessionListFails(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateWaitingForInput
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	svc.repo = listTaskSessionsErrorRepo{sessionExecutorStore: svc.repo}
+
+	svc.writeTaskReviewState(ctx, "t1", "s1")
+
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"REVIEW writes must fail closed when sibling session reconciliation fails")
+}
+
+func TestWriteTaskReviewStateSkipsTerminalTaskState(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateWaitingForInput
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+	task, err := repo.GetTask(ctx, "t1")
+	require.NoError(t, err)
+	task.State = v1.TaskStateCompleted
+	require.NoError(t, repo.UpdateTask(ctx, task))
+
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks["t1"] = &v1.Task{ID: "t1", State: v1.TaskStateCompleted}
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	svc.writeTaskReviewState(ctx, "t1", "s1")
+
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"REVIEW reconcile must not rewind terminal task states")
+}
+
+func TestWriteTaskInProgressForRuntimeSkipsArchivedTask(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	require.NoError(t, repo.ArchiveTask(ctx, "t1"))
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	svc.writeTaskInProgressForRuntime(ctx, "t1", "s1")
+
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"runtime reconciliation must not promote archived tasks to IN_PROGRESS")
+}
+
+func TestWriteTaskReviewStateOnCancelSkipsWhenSessionActiveAgain(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateRunning
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks["t1"] = &v1.Task{ID: "t1", State: v1.TaskStateInProgress}
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	svc.writeTaskReviewStateOnCancel(ctx, "t1", "s1")
+
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"cancel reconcile must not move a restarted same session back to REVIEW")
+}
+
+func TestWriteTaskReviewStateOnCancelSkipsWhenSessionListFails(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateWaitingForInput
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks["t1"] = &v1.Task{ID: "t1", State: v1.TaskStateInProgress}
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+	svc.repo = listTaskSessionsErrorRepo{sessionExecutorStore: svc.repo}
+
+	svc.writeTaskReviewStateOnCancel(ctx, "t1", "s1")
+
+	require.Zero(t, taskRepo.stateWrites["t1"],
+		"cancel REVIEW writes must fail closed when sibling session reconciliation fails")
+}
+
+func TestToolUpdateFromFailedExecutionDoesNotCreateMessage(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+
+	svc.handleAgentFailed(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "exec-1",
+		ErrorMessage:     "agent failed",
+	})
+
+	svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			ToolCallID: "tc1",
+			ToolStatus: agentEventComplete,
+		},
+	})
+
+	require.Zero(t, messages.toolUpdateWrites,
+		"late tool events from a failed execution must be dropped before message side effects")
+}
+
+func TestCompletedExecutionMarkerExpiresAndDeletes(t *testing.T) {
+	svc := &Service{}
+
+	svc.markExecutionCompleted("s1", "exec-1")
+	require.True(t, svc.isExecutionCompleted("s1", "exec-1"))
+
+	currentKey := terminalExecutionKey("s1", "exec-1")
+	expiredAt := time.Now().Add(-time.Minute)
+	svc.completedExecutions.Store(currentKey, terminalExecutionMarker{
+		expiresAt:           expiredAt,
+		allowCompleteStream: true,
+	})
+	require.False(t, svc.isExecutionCompleted("s1", "exec-1"))
+	_, ok := svc.completedExecutions.Load(currentKey)
+	require.False(t, ok, "expired marker should be deleted on lookup")
+
+	laterExpiry := time.Now().Add(time.Hour)
+	svc.completedExecutions.Store(currentKey, terminalExecutionMarker{
+		expiresAt:           laterExpiry,
+		allowCompleteStream: true,
+	})
+	svc.deleteCompletedExecutionIfExpired(currentKey, time.Now())
+	_, ok = svc.completedExecutions.Load(currentKey)
+	require.True(t, ok, "old expiry callback must not delete a refreshed marker")
+
+	svc.deleteCompletedExecutionIfExpired(currentKey, laterExpiry)
+	_, ok = svc.completedExecutions.Load(currentKey)
+	require.False(t, ok, "matching expiry callback should delete the marker")
+}
+
 // TestSetSessionRunning_NoRedundantTaskWrites locks in the dedup: when the
 // session is already RUNNING, setSessionRunning must not re-write tasks.state.
 // Without the guard, every tool_call / tool_update fired UpdateTaskState
@@ -572,6 +1187,134 @@ func TestSetSessionRunning_WritesOnTransition(t *testing.T) {
 	require.Equal(t, v1.TaskStateInProgress, taskRepo.updatedStates["t1"])
 }
 
+func TestSetSessionStartingWritesTaskInProgress(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateStarting
+	session.ErrorMessage = ""
+	session.UpdatedAt = time.Now().UTC()
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	require.NoError(t, svc.setSessionStarting(ctx, "t1", session, true))
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateStarting, updated.State)
+	require.Equal(t, 1, taskRepo.stateWrites["t1"])
+	require.Equal(t, v1.TaskStateInProgress, taskRepo.updatedStates["t1"])
+}
+
+func TestSetSessionStartingCanDeferTaskInProgress(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateStarting
+	session.ErrorMessage = ""
+	session.UpdatedAt = time.Now().UTC()
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	require.NoError(t, svc.setSessionStarting(ctx, "t1", session, false))
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateStarting, updated.State)
+	require.Empty(t, taskRepo.stateWrites)
+}
+
+func TestSetSessionStartingRejectsTerminalSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	current, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	current.State = models.TaskSessionStateCancelled
+	require.NoError(t, repo.UpdateTaskSession(ctx, current))
+
+	next := *current
+	next.State = models.TaskSessionStateStarting
+	next.ErrorMessage = ""
+	next.UpdatedAt = time.Now().UTC()
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	require.Error(t, svc.setSessionStarting(ctx, "t1", &next, true))
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateCancelled, updated.State)
+	require.Empty(t, taskRepo.stateWrites)
+}
+
+func TestSetSessionStartingRejectsCancelledTerminalResumeWithoutPromotion(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	current, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	current.State = models.TaskSessionStateCancelled
+	require.NoError(t, repo.UpdateTaskSession(ctx, current))
+
+	next := *current
+	next.State = models.TaskSessionStateStarting
+	next.ErrorMessage = ""
+	next.UpdatedAt = time.Now().UTC()
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	require.Error(t, svc.setSessionStarting(ctx, "t1", &next, false))
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateCancelled, updated.State)
+	require.Empty(t, taskRepo.stateWrites)
+}
+
+func TestSetSessionStartingAllowsTerminalResumeWithoutTaskPromotion(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	current, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	current.State = models.TaskSessionStateFailed
+	current.ErrorMessage = "previous failure"
+	completedAt := time.Now().UTC()
+	current.CompletedAt = &completedAt
+	require.NoError(t, repo.UpdateTaskSession(ctx, current))
+
+	next := *current
+	next.State = models.TaskSessionStateStarting
+	next.ErrorMessage = ""
+	next.CompletedAt = nil
+	next.UpdatedAt = time.Now().UTC()
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	require.NoError(t, svc.setSessionStarting(ctx, "t1", &next, false))
+
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateStarting, updated.State)
+	require.Nil(t, updated.CompletedAt)
+	require.Empty(t, taskRepo.stateWrites)
+}
+
 // Pins the call-site wiring: cancelled office turn must NOT leave the session at IDLE.
 func TestHandleCompleteStreamEvent_CancelledOfficeSessionLandsWaitingForInput(t *testing.T) {
 	ctx := context.Background()
@@ -848,6 +1591,7 @@ func TestSetSessionWaitingForInput_WritesOnTransition(t *testing.T) {
 	require.NoError(t, repo.UpdateTaskSession(ctx, session))
 
 	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 	svc := createTestService(repo, newMockStepGetter(), taskRepo)
 
 	svc.setSessionWaitingForInput(ctx, "t1", "s1")
@@ -857,6 +1601,51 @@ func TestSetSessionWaitingForInput_WritesOnTransition(t *testing.T) {
 	require.Equal(t, v1.TaskStateReview, taskRepo.updatedStates["t1"])
 }
 
+func TestSetSessionWaitingForInput_DoesNotMoveTaskToReviewWhileSiblingRuns(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s-finishing", "step1")
+
+	now := time.Now().UTC()
+	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()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	svc.setSessionWaitingForInput(ctx, "t1", "s-finishing")
+
+	updatedFinishing, err := repo.GetTaskSession(ctx, "s-finishing")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateWaitingForInput, updatedFinishing.State)
+	require.Equal(t, 0, taskRepo.stateWrites["t1"],
+		"finishing one session must not move the task to REVIEW while another session is running")
+}
+
+func TestWriteTaskReviewState_DoesNotMoveTaskToReviewWhenSameSessionRestarted(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	session, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	session.State = models.TaskSessionStateRunning
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	taskRepo := newMockTaskRepo()
+	svc := createTestService(repo, newMockStepGetter(), taskRepo)
+
+	svc.writeTaskReviewState(ctx, "t1", "s1")
+
+	require.Empty(t, taskRepo.updatedStates,
+		"the just-completed session may have restarted before REVIEW reconciliation")
+}
+
 func TestSessionStateString(t *testing.T) {
 	require.Equal(t, "", sessionStateString(nil),
 		"nil session must render as empty so trace logs stay clean")

apps/backend/internal/orchestrator/event_handlers_test.go

@@ -98,6 +98,10 @@ func newMockTaskRepo() *mockTaskRepo {
 	}
 }
 
+func seedMockTaskState(repo *mockTaskRepo, taskID string, state v1.TaskState) {
+	repo.tasks[taskID] = &v1.Task{ID: taskID, State: state}
+}
+
 func (m *mockTaskRepo) GetTask(_ context.Context, taskID string) (*v1.Task, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -722,6 +726,7 @@ func TestHandleAgentReadyGuards(t *testing.T) {
 		repo := setupTestRepo(t)
 		seedSession(t, repo, "t1", "s1", "step1")
 		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 
 		// Register the workflow step so processOnTurnComplete can resolve it.
 		stepGetter := newMockStepGetter()
@@ -1046,6 +1051,250 @@ func TestHandleAgentCompleted_CleansUpExecution(t *testing.T) {
 	}
 }
 
+func TestHandleAgentCompleted_MarksRotatedExecutionCompleted(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "")
+	seedExecutorRunning(t, repo, "s1", "t1", "exec-live")
+
+	taskRepo := newMockTaskRepo()
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "exec-old",
+	})
+	waitForStopCall(t, agentMgr)
+
+	if !svc.isExecutionCompleted("s1", "exec-old") {
+		t.Fatal("rotated agent.completed events must still block late tool events from the old execution")
+	}
+}
+
+func TestHandleAgentCompleted_DoesNotMoveTaskToReviewWhileSiblingRuns(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s-finishing", "")
+	seedExecutorRunning(t, repo, "s-finishing", "t1", "exec-finishing")
+
+	now := time.Now().UTC()
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID:        "s-running",
+		TaskID:    "t1",
+		State:     models.TaskSessionStateRunning,
+		StartedAt: now.Add(time.Second),
+		UpdatedAt: now.Add(time.Second),
+	}); err != nil {
+		t.Fatalf("failed to create running sibling session: %v", err)
+	}
+
+	taskRepo := newMockTaskRepo()
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s-finishing",
+		AgentExecutionID: "exec-finishing",
+	})
+	waitForStopCall(t, agentMgr)
+
+	if got := taskRepo.stateWrites["t1"]; got != 0 {
+		t.Fatalf("agent completion must not move the task to REVIEW while another session is running, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+
+	updatedFinishing, err := repo.GetTaskSession(ctx, "s-finishing")
+	if err != nil {
+		t.Fatalf("failed to load finishing session: %v", err)
+	}
+	if updatedFinishing.State != models.TaskSessionStateWaitingForInput {
+		t.Fatalf("expected finishing session to leave RUNNING, got %q", updatedFinishing.State)
+	}
+}
+
+func TestHandleAgentCompleted_MovesTaskToReviewWhenLastSiblingFinishes(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s-first", "step1")
+	seedExecutorRunning(t, repo, "s-first", "t1", "exec-first")
+
+	now := time.Now().UTC()
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID:        "s-last",
+		TaskID:    "t1",
+		State:     models.TaskSessionStateRunning,
+		StartedAt: now.Add(time.Second),
+		UpdatedAt: now.Add(time.Second),
+	}); err != nil {
+		t.Fatalf("failed to create running sibling session: %v", err)
+	}
+	seedExecutorRunning(t, repo, "s-last", "t1", "exec-last")
+
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
+		ID:         "step1",
+		WorkflowID: "wf1",
+		Name:       "Step 1",
+		Position:   0,
+	}
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s-first",
+		AgentExecutionID: "exec-first",
+	})
+
+	if got := taskRepo.stateWrites["t1"]; got != 0 {
+		t.Fatalf("first completion must not move task to REVIEW while sibling runs, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s-last",
+		AgentExecutionID: "exec-last",
+	})
+
+	if got := taskRepo.stateWrites["t1"]; got != 1 {
+		t.Fatalf("last sibling completion must move task to REVIEW once, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+	if state := taskRepo.updatedStates["t1"]; state != v1.TaskStateReview {
+		t.Fatalf("expected task state %q, got %q", v1.TaskStateReview, state)
+	}
+}
+
+func TestHandleAgentCompleted_NoWorkflowStepLastSiblingMovesTaskToReview(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s-first", "")
+	seedExecutorRunning(t, repo, "s-first", "t1", "exec-first")
+
+	now := time.Now().UTC()
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID:        "s-last",
+		TaskID:    "t1",
+		State:     models.TaskSessionStateRunning,
+		StartedAt: now.Add(time.Second),
+		UpdatedAt: now.Add(time.Second),
+	}); err != nil {
+		t.Fatalf("failed to create running sibling session: %v", err)
+	}
+	seedExecutorRunning(t, repo, "s-last", "t1", "exec-last")
+
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s-first",
+		AgentExecutionID: "exec-first",
+	})
+
+	if got := taskRepo.stateWrites["t1"]; got != 0 {
+		t.Fatalf("first no-step completion must not move task to REVIEW while sibling runs, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+	updatedFirst, err := repo.GetTaskSession(ctx, "s-first")
+	if err != nil {
+		t.Fatalf("failed to load first session: %v", err)
+	}
+	if updatedFirst.State != models.TaskSessionStateWaitingForInput {
+		t.Fatalf("expected first session state %q, got %q", models.TaskSessionStateWaitingForInput, updatedFirst.State)
+	}
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s-last",
+		AgentExecutionID: "exec-last",
+	})
+
+	if got := taskRepo.stateWrites["t1"]; got != 1 {
+		t.Fatalf("last no-step sibling completion must move task to REVIEW once, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+	updatedFirst, err = repo.GetTaskSession(ctx, "s-first")
+	if err != nil {
+		t.Fatalf("failed to reload first session: %v", err)
+	}
+	if updatedFirst.State != models.TaskSessionStateWaitingForInput {
+		t.Fatalf("expected first session to remain %q, got %q", models.TaskSessionStateWaitingForInput, updatedFirst.State)
+	}
+	updatedLast, err := repo.GetTaskSession(ctx, "s-last")
+	if err != nil {
+		t.Fatalf("failed to load last session: %v", err)
+	}
+	if updatedLast.State != models.TaskSessionStateWaitingForInput {
+		t.Fatalf("expected last session state %q, got %q", models.TaskSessionStateWaitingForInput, updatedLast.State)
+	}
+}
+
+func TestHandleAgentCompleted_ExitOnlyCompletesActiveTurn(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "task1", "session1", "")
+	seedExecutorRunning(t, repo, "session1", "task1", "exec-1")
+
+	taskRepo := newMockTaskRepo()
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+	svc.turnService = &repoTurnService{repo: repo}
+
+	if _, err := svc.turnService.StartTurn(ctx, "session1"); err != nil {
+		t.Fatalf("failed to seed active turn: %v", err)
+	}
+	if open := openTurnCount(t, repo, "session1"); open != 1 {
+		t.Fatalf("expected 1 open turn before completion, got %d", open)
+	}
+
+	svc.handleAgentCompleted(ctx, watcher.AgentEventData{
+		TaskID:           "task1",
+		SessionID:        "session1",
+		AgentExecutionID: "exec-1",
+	})
+	waitForStopCall(t, agentMgr)
+
+	if open := openTurnCount(t, repo, "session1"); open != 0 {
+		t.Fatalf("expected agent.completed to close the active turn, got %d open turns", open)
+	}
+	updated, err := repo.GetTaskSession(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to load session: %v", err)
+	}
+	if updated.State != models.TaskSessionStateWaitingForInput {
+		t.Fatalf("expected session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State)
+	}
+}
+
+func TestHandleAgentFailedWithoutSession_DoesNotMoveTaskToReviewWhileSessionRuns(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s-running", "")
+
+	taskRepo := newMockTaskRepo()
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
+
+	svc.handleAgentFailed(ctx, watcher.AgentEventData{
+		TaskID:       "t1",
+		ErrorMessage: "agent failed before session binding",
+	})
+
+	if got := taskRepo.stateWrites["t1"]; got != 0 {
+		t.Fatalf("sessionless failure must not move the task to REVIEW while a session is running, writes=%d state=%q",
+			got, taskRepo.updatedStates["t1"])
+	}
+}
+
 func TestHandleAgentFailed_CleansUpExecution(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)
@@ -1496,6 +1745,7 @@ func TestHandleRecoverableFailure(t *testing.T) {
 		seedSession(t, repo, "t1", "s1", "step1")
 
 		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 		agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
 		svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr)
 
@@ -1603,6 +1853,7 @@ func TestHandleResumeFailure(t *testing.T) {
 		})
 
 		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 		svc := createTestService(repo, newMockStepGetter(), taskRepo)
 
 		result := svc.handleResumeFailure(ctx, watcher.AgentEventData{

apps/backend/internal/orchestrator/event_handlers_transient_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
 	"github.com/kandev/kandev/internal/orchestrator/watcher"
 	"github.com/kandev/kandev/internal/task/models"
 )
@@ -78,6 +79,33 @@ func TestHandleTransientFailure_SchedulesRetryAndEmitsWarning(t *testing.T) {
 	}
 }
 
+func TestTransientFailedExecutionToolUpdateDoesNotCreateMessage(t *testing.T) {
+	svc, mc := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+
+	svc.handleAgentFailed(context.Background(), watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "exec-1",
+		ErrorMessage:     overloaded529,
+	})
+
+	svc.handleAgentStreamEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "exec-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:       "tool_update",
+			ToolCallID: "tc1",
+			ToolStatus: agentEventComplete,
+		},
+	})
+
+	if mc.toolUpdateWrites != 0 {
+		t.Fatalf("expected stale transient-failure tool update to be dropped, got %d writes", mc.toolUpdateWrites)
+	}
+}
+
 func TestHandleTransientFailure_NonTransientReturnsFalse(t *testing.T) {
 	svc, mc := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -2159,12 +2159,13 @@ func (s *Service) applyEngineTransition(
 	// IN_PROGRESS, leaving the spinner spinning in the new column even though
 	// the agent has paused. If the target step's on_enter starts another agent,
 	// setSessionRunning will flip tasks.state back to IN_PROGRESS — the
-	// REVIEW write is a safe intermediate that any active-running follow-up
-	// will overwrite.
+	// REVIEW write is a safe intermediate when no sibling session is already
+	// working; otherwise the task should remain IN_PROGRESS until all active
+	// agent work has paused.
 	if session.State == models.TaskSessionStateRunning || session.State == models.TaskSessionStateStarting {
 		s.updateTaskSessionState(ctx, taskID, session.ID, models.TaskSessionStateWaitingForInput, "", false, session)
 		session.State = models.TaskSessionStateWaitingForInput
-		s.writeTaskReviewState(ctx, taskID)
+		s.writeTaskReviewState(ctx, taskID, session.ID)
 	}
 
 	// Launch processOnEnter asynchronously to avoid blocking the stream reader goroutine.

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

@@ -494,6 +494,16 @@ type TaskStateChangeFunc func(ctx context.Context, taskID string, state v1.TaskS
 // publish events (e.g. WebSocket notifications) alongside the DB update.
 type SessionStateChangeFunc func(ctx context.Context, taskID, sessionID string, state models.TaskSessionState, errorMessage string) error
 
+// SessionStartingFunc is called when the executor has prepared/resumed an
+// execution and needs to mark the session STARTING while preserving other
+// session-row updates such as metadata. promoteTask controls whether the
+// callback should also move the parent task to IN_PROGRESS immediately.
+type SessionStartingFunc func(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error
+
+// TaskReviewStateReconcileFunc is called when runtime work stopped and the
+// parent task should move to REVIEW only if no session is still STARTING/RUNNING.
+type TaskReviewStateReconcileFunc func(ctx context.Context, taskID, completedSessionID string)
+
 // AgentStartFailedFunc is called when the agent process fails to start.
 // It receives the task/session/execution IDs and the error. fromResume is true
 // when the failure occurred during a background session resume (rather than a
@@ -541,6 +551,16 @@ type Executor struct {
 	// updates the DB and publishes WebSocket events.
 	onSessionStateChange SessionStateChangeFunc
 
+	// Callback for STARTING writes that carry full session-row changes. Set by
+	// the orchestrator so launch/resume/model-switch transitions serialize with
+	// runtime task-state reconciliation.
+	onSessionStarting SessionStartingFunc
+
+	// Callback for REVIEW reconciliation after runtime start failures. Set by the
+	// orchestrator so failed-start writes share the same serialized guard as
+	// normal turn completion.
+	onTaskReviewStateReconcile TaskReviewStateReconcileFunc
+
 	// Callback for agent process start failures. When set, the executor
 	// delegates failure handling to this callback, allowing the orchestrator
 	// to detect auth errors and treat them as recoverable.
@@ -633,6 +653,17 @@ func (e *Executor) SetOnSessionStateChange(fn SessionStateChangeFunc) {
 	e.onSessionStateChange = fn
 }
 
+// SetOnSessionStarting sets a callback for full session-row STARTING updates.
+func (e *Executor) SetOnSessionStarting(fn SessionStartingFunc) {
+	e.onSessionStarting = fn
+}
+
+// SetOnTaskReviewStateReconcile sets the guarded task REVIEW reconciliation
+// callback used after resume/start failures.
+func (e *Executor) SetOnTaskReviewStateReconcile(fn TaskReviewStateReconcileFunc) {
+	e.onTaskReviewStateReconcile = fn
+}
+
 // SetRepoCloner sets the cloner used to clone provider-backed repositories on launch.
 func (e *Executor) SetRepoCloner(cloner RepoCloner, updater RepoUpdater) {
 	e.repoCloner = cloner

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

@@ -13,6 +13,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/orchestrator/sessionstate"
 	"github.com/kandev/kandev/internal/sysprompt"
 	"github.com/kandev/kandev/internal/task/models"
 	"github.com/kandev/kandev/internal/worktree"
@@ -80,6 +81,8 @@ func (e *Executor) runAgentProcessAsync(ctx context.Context, taskID, sessionID,
 						zap.String("task_id", taskID),
 						zap.Error(updateErr))
 				}
+			} else {
+				e.writeTaskReviewStateIfNoWorkingSessions(updateCtx, taskID, sessionID)
 			}
 			// Clean up the execution environment (e.g., destroy remote Sprites instance).
 			// Use force=true since the agent process never fully started.
@@ -106,6 +109,113 @@ func (e *Executor) startAgentProcessAsync(ctx context.Context, taskID, sessionID
 	}, true, false)
 }
 
+func (e *Executor) stopUnstartedExecution(ctx context.Context, sessionID, agentExecutionID string) {
+	if agentExecutionID == "" {
+		return
+	}
+	stopCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+	defer cancel()
+	if stopErr := e.agentManager.StopAgent(stopCtx, agentExecutionID, true); stopErr != nil {
+		e.logger.Warn("failed to stop unstarted agent execution",
+			zap.String("session_id", sessionID),
+			zap.String("agent_execution_id", agentExecutionID),
+			zap.Error(stopErr))
+	}
+}
+
+func (e *Executor) writeTaskReviewStateIfNoWorkingSessions(ctx context.Context, taskID, failedSessionID string) {
+	if e.onTaskReviewStateReconcile != nil {
+		e.onTaskReviewStateReconcile(ctx, taskID, failedSessionID)
+		return
+	}
+
+	if e.shouldSkipFailedStartReviewForTask(ctx, taskID, failedSessionID) {
+		return
+	}
+	if e.failedSessionStillWorkingOrUnknown(ctx, taskID, failedSessionID) {
+		return
+	}
+	if e.hasOtherWorkingSessions(ctx, taskID, failedSessionID) {
+		return
+	}
+	if updateErr := e.updateTaskState(ctx, taskID, v1.TaskStateReview); updateErr != nil {
+		e.logger.Warn("failed to update task state to REVIEW after start error",
+			zap.String("task_id", taskID),
+			zap.Error(updateErr))
+	}
+}
+
+func (e *Executor) shouldSkipFailedStartReviewForTask(ctx context.Context, taskID, failedSessionID string) bool {
+	task, err := e.repo.GetTask(ctx, taskID)
+	if err != nil {
+		e.logger.Warn("failed to load task before failed-start REVIEW state reconcile",
+			zap.String("task_id", taskID),
+			zap.String("session_id", failedSessionID),
+			zap.Error(err))
+		return true
+	}
+	if task != nil && task.AssigneeAgentProfileID != "" {
+		e.logger.Debug("skipping failed-start task REVIEW state for office task",
+			zap.String("task_id", taskID),
+			zap.String("session_id", failedSessionID))
+		return true
+	}
+	return false
+}
+
+func (e *Executor) failedSessionStillWorkingOrUnknown(ctx context.Context, taskID, failedSessionID string) bool {
+	if failedSessionID == "" {
+		return false
+	}
+	session, err := e.repo.GetTaskSession(ctx, failedSessionID)
+	if err != nil {
+		e.logger.Warn("failed to load failed session before failed-start REVIEW state reconcile",
+			zap.String("task_id", taskID),
+			zap.String("session_id", failedSessionID),
+			zap.Error(err))
+		return true
+	}
+	if session != nil && isRuntimeWorkingSessionState(session.State) {
+		e.logger.Debug("skipping failed-start task REVIEW state because failed session is active again",
+			zap.String("task_id", taskID),
+			zap.String("session_id", failedSessionID),
+			zap.String("session_state", string(session.State)))
+		return true
+	}
+	return false
+}
+
+func (e *Executor) hasOtherWorkingSessions(ctx context.Context, taskID, failedSessionID string) bool {
+	sessions, err := e.repo.ListTaskSessions(ctx, taskID)
+	if err != nil {
+		e.logger.Warn("failed to list task sessions before failed-start REVIEW state reconcile",
+			zap.String("task_id", taskID),
+			zap.String("session_id", failedSessionID),
+			zap.Error(err))
+		return true
+	}
+	for _, session := range sessions {
+		if session == nil {
+			continue
+		}
+		if failedSessionID != "" && session.ID == failedSessionID {
+			continue
+		}
+		if isRuntimeWorkingSessionState(session.State) {
+			e.logger.Debug("skipping failed-start task REVIEW state while another session is working",
+				zap.String("task_id", taskID),
+				zap.String("failed_session_id", failedSessionID),
+				zap.String("blocking_session_id", session.ID))
+			return true
+		}
+	}
+	return false
+}
+
+func isRuntimeWorkingSessionState(state models.TaskSessionState) bool {
+	return sessionstate.IsWorking(state)
+}
+
 // updateTaskState updates a task's state, using the callback if set for event publishing,
 // or falling back to the raw repository.
 func (e *Executor) updateTaskState(ctx context.Context, taskID string, state v1.TaskState) error {
@@ -124,6 +234,16 @@ func (e *Executor) updateSessionState(ctx context.Context, taskID, sessionID str
 	return e.repo.UpdateTaskSessionState(ctx, sessionID, state, errorMessage)
 }
 
+// updateSessionStarting persists a full session-row STARTING transition, using
+// the orchestrator callback when present so task/session runtime state stays
+// serialized with guarded REVIEW reconciliation.
+func (e *Executor) updateSessionStarting(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error {
+	if e.onSessionStarting != nil {
+		return e.onSessionStarting(ctx, taskID, session, promoteTask)
+	}
+	return e.repo.UpdateTaskSession(ctx, session)
+}
+
 // shouldUseWorktree returns true if the given executor type should use Git worktrees.
 func shouldUseWorktree(executorType string) bool {
 	return models.ExecutorType(executorType) == models.ExecutorTypeWorktree
@@ -550,7 +670,10 @@ func (e *Executor) handleLaunchFailure(ctx context.Context, taskID, sessionID st
 // finalizeLaunch persists launch state and returns the resulting TaskExecution.
 func (e *Executor) finalizeLaunch(ctx context.Context, task *v1.Task, session *models.TaskSession, agentProfileID, sessionID string, repoInfo *repoInfo, resp *LaunchAgentResponse, startAgent bool, execCfg executorConfig) (*TaskExecution, error) {
 	now := time.Now().UTC()
-	e.persistLaunchState(ctx, task.ID, sessionID, session, resp, startAgent, now)
+	if err := e.persistLaunchState(ctx, task.ID, sessionID, session, resp, startAgent, now); err != nil {
+		e.stopUnstartedExecution(ctx, sessionID, resp.AgentExecutionID)
+		return nil, err
+	}
 	e.persistWorktreeAssociation(ctx, task.ID, session, repoInfo.RepositoryID, resp)
 
 	sessionState := v1.TaskSessionStateCreated
@@ -991,10 +1114,11 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T
 	session.State = models.TaskSessionStateStarting
 	session.ErrorMessage = ""
 	session.UpdatedAt = now
-	if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
+	if err := e.updateSessionStarting(ctx, task.ID, session, true); err != nil {
 		e.logger.Error("failed to update session state for agent start",
 			zap.String("session_id", session.ID),
 			zap.Error(err))
+		return nil, err
 	}
 
 	execution := &TaskExecution{

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

@@ -409,7 +409,10 @@ func (e *Executor) launchModelSwitchAgent(ctx context.Context, taskID, sessionID
 		return fmt.Errorf("failed to launch agent with new model: %w", err)
 	}
 
-	e.persistModelSwitchState(ctx, taskID, sessionID, session, newModel)
+	if err := e.persistModelSwitchState(ctx, taskID, sessionID, session, newModel); err != nil {
+		e.stopUnstartedExecution(ctx, sessionID, resp.AgentExecutionID)
+		return err
+	}
 
 	if err := e.agentManager.StartAgentProcess(ctx, resp.AgentExecutionID); err != nil {
 		e.logger.Error("failed to start agent process after model switch",
@@ -516,7 +519,7 @@ func (e *Executor) applyWorktreeToSwitchRequest(req *LaunchAgentRequest, session
 // after a model switch launch. The executors_running row's agent_execution_id /
 // container_id / status are written by the lifecycle manager during the launch
 // itself (lifecycle.persistExecutorRunning) and not touched here.
-func (e *Executor) persistModelSwitchState(ctx context.Context, taskID, sessionID string, session *models.TaskSession, newModel string) {
+func (e *Executor) persistModelSwitchState(ctx context.Context, taskID, sessionID string, session *models.TaskSession, newModel string) error {
 	session.State = models.TaskSessionStateStarting
 	session.UpdatedAt = time.Now().UTC()
 
@@ -525,14 +528,15 @@ func (e *Executor) persistModelSwitchState(ctx context.Context, taskID, sessionI
 	}
 	session.AgentProfileSnapshot["model"] = newModel
 
-	if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
+	if err := e.updateSessionStarting(ctx, taskID, session, true); err != nil {
 		e.logger.Error("failed to update session after model switch",
 			zap.String("task_id", taskID),
 			zap.String("session_id", sessionID),
 			zap.Error(err))
-		return
+		return err
 	}
 	e.persistRuntimeModelMetadata(ctx, sessionID, session, newModel)
+	return nil
 }
 
 // persistInPlaceModelSwitch updates the session snapshot model after a successful

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

@@ -230,7 +230,7 @@ func (e *Executor) backfillRepoDefaultBranch(ctx context.Context, repo *models.R
 // What remains here: state transitions (e.g., STARTING) and prepare-result
 // metadata merge, both of which are session-row concerns the lifecycle manager
 // doesn't know about.
-func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID string, session *models.TaskSession, resp *LaunchAgentResponse, startAgent bool, now time.Time) {
+func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID string, session *models.TaskSession, resp *LaunchAgentResponse, startAgent bool, now time.Time) error {
 	if startAgent {
 		session.State = models.TaskSessionStateStarting
 	}
@@ -247,12 +247,20 @@ func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID str
 		session.Metadata["prepare_result"] = buildPrepareResultMetadata(resp.PrepareResult)
 	}
 
-	if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
+	var updateErr error
+	if startAgent {
+		updateErr = e.updateSessionStarting(ctx, taskID, session, true)
+	} else {
+		updateErr = e.repo.UpdateTaskSession(ctx, session)
+	}
+	if updateErr != nil {
 		e.logger.Error("failed to update agent session after launch",
 			zap.String("task_id", taskID),
 			zap.String("session_id", sessionID),
-			zap.Error(err))
+			zap.Error(updateErr))
+		return updateErr
 	}
+	return nil
 }
 
 // buildPrepareResultMetadata serializes a prepare result for storage in session metadata.
@@ -389,7 +397,10 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio
 		return nil, err
 	}
 
-	e.persistResumeState(ctx, task.ID, session, startAgent)
+	if err := e.persistResumeState(ctx, task.ID, session, startAgent); err != nil {
+		e.stopUnstartedExecution(ctx, session.ID, resp.AgentExecutionID)
+		return nil, err
+	}
 	e.persistWorktreeAssociation(ctx, task.ID, session, repositoryID, resp)
 	// Refresh task_environments after a successful resume so the row reflects
 	// the new worktree, container, and ready status. Without this, sessions
@@ -827,19 +838,27 @@ func resolveResumeTaskDirName(existingEnv *models.TaskEnvironment, task *v1.Task
 // and not touched here — see lifecycle.persistExecutorRunning. The
 // orchestrator's only remaining responsibility is the session-row state
 // machine (STARTING / CompletedAt-clear).
-func (e *Executor) persistResumeState(ctx context.Context, taskID string, session *models.TaskSession, startAgent bool) {
+func (e *Executor) persistResumeState(ctx context.Context, taskID string, session *models.TaskSession, startAgent bool) error {
 	session.ErrorMessage = ""
 	if startAgent {
 		session.State = models.TaskSessionStateStarting
 		session.CompletedAt = nil
 	}
 
-	if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
+	var updateErr error
+	if startAgent {
+		updateErr = e.updateSessionStarting(ctx, taskID, session, false)
+	} else {
+		updateErr = e.repo.UpdateTaskSession(ctx, session)
+	}
+	if updateErr != nil {
 		e.logger.Error("failed to update task session for resume",
 			zap.String("task_id", taskID),
 			zap.String("session_id", session.ID),
-			zap.Error(err))
+			zap.Error(updateErr))
+		return updateErr
 	}
+	return nil
 }
 
 // prNumberFromMetadata extracts a GitHub PR number from a task_repository's
@@ -876,9 +895,50 @@ func prNumberFromMetadata(metadata map[string]interface{}) int {
 // just logs successful process start.
 func (e *Executor) startAgentProcessOnResume(ctx context.Context, taskID string, session *models.TaskSession, agentExecutionID string) {
 	e.runAgentProcessAsync(ctx, taskID, session.ID, agentExecutionID, func(updCtx context.Context) {
+		if updateErr := e.writeTaskInProgressForRuntime(updCtx, taskID, session.ID); updateErr != nil {
+			e.logger.Warn("failed to update task state to IN_PROGRESS after resume start",
+				zap.String("task_id", taskID),
+				zap.String("session_id", session.ID),
+				zap.Error(updateErr))
+		}
 		e.logger.Debug("agent resumed successfully",
 			zap.String("task_id", taskID),
 			zap.String("session_id", session.ID),
 			zap.String("session_state", string(session.State)))
 	}, false, true)
 }
+
+func (e *Executor) writeTaskInProgressForRuntime(ctx context.Context, taskID, sessionID string) error {
+	task, err := e.repo.GetTask(ctx, taskID)
+	if err != nil {
+		return err
+	}
+	if task != nil && task.ArchivedAt != nil {
+		e.logger.Debug("skipping IN_PROGRESS transition for archived task",
+			zap.String("task_id", taskID))
+		return nil
+	}
+	if task != nil && task.AssigneeAgentProfileID != "" {
+		e.logger.Debug("skipping IN_PROGRESS transition for office task",
+			zap.String("task_id", taskID))
+		return nil
+	}
+	if sessionID != "" {
+		session, err := e.repo.GetTaskSession(ctx, sessionID)
+		if err != nil {
+			return err
+		}
+		if session == nil || !isRuntimeWorkingSessionState(session.State) {
+			state := ""
+			if session != nil {
+				state = string(session.State)
+			}
+			e.logger.Debug("skipping IN_PROGRESS transition because resumed session is no longer active",
+				zap.String("task_id", taskID),
+				zap.String("session_id", sessionID),
+				zap.String("session_state", state))
+			return nil
+		}
+	}
+	return e.updateTaskState(ctx, taskID, v1.TaskStateInProgress)
+}

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

@@ -2,6 +2,7 @@ package executor
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os/exec"
 	"sync"
@@ -262,6 +263,71 @@ func TestLaunchPreparedSession_Success(t *testing.T) {
 	}
 }
 
+func TestLaunchPreparedSession_AbortsWhenStartingPersistenceFails(t *testing.T) {
+	repo := newMockRepository()
+	session := &models.TaskSession{
+		ID:             "session-abort",
+		TaskID:         "task-abort",
+		AgentProfileID: "profile-123",
+		State:          models.TaskSessionStateCreated,
+		StartedAt:      time.Now(),
+		UpdatedAt:      time.Now(),
+	}
+	repo.sessions[session.ID] = session
+
+	startCalled := make(chan struct{}, 1)
+	stopCalled := make(chan string, 1)
+	agentManager := &mockAgentManager{
+		launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) {
+			return &LaunchAgentResponse{
+				AgentExecutionID: "exec-abort",
+				Status:           v1.AgentStatusStarting,
+			}, nil
+		},
+		startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error {
+			startCalled <- struct{}{}
+			return nil
+		},
+		stopAgentFunc: func(ctx context.Context, agentExecutionID string, force bool) error {
+			if !force {
+				t.Error("expected cleanup stop to be forced")
+			}
+			stopCalled <- agentExecutionID
+			return nil
+		},
+	}
+
+	persistErr := errors.New("session is terminal")
+	executor := newTestExecutor(t, agentManager, repo)
+	executor.SetOnSessionStarting(func(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error {
+		return persistErr
+	})
+
+	task := &v1.Task{ID: "task-abort", WorkspaceID: "workspace-123", Title: "Test Task"}
+
+	execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-abort", LaunchOptions{
+		AgentProfileID: "profile-123",
+		StartAgent:     true,
+	})
+	if !errors.Is(err, persistErr) {
+		t.Fatalf("expected persistence error, got execution=%v err=%v", execution, err)
+	}
+
+	select {
+	case got := <-stopCalled:
+		if got != "exec-abort" {
+			t.Fatalf("stopped execution %q, want exec-abort", got)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected unstarted execution to be stopped")
+	}
+	select {
+	case <-startCalled:
+		t.Fatal("agent process must not start after STARTING persistence fails")
+	default:
+	}
+}
+
 // TestLaunchPreparedSession_PropagatesIsPassthrough mirrors
 // TestResumeSession_PropagatesIsPassthrough for the initial-launch path: the
 // session's IsPassthrough snapshot must reach the LaunchAgentRequest so the
@@ -1050,6 +1116,7 @@ verified:
 // other recorded fields, removing the need for atomicity or polling.
 type runAgentProcessAsyncFailureFixture struct {
 	exec              *Executor
+	repo              *mockRepository
 	taskStateUpdates  []string
 	sessionFailedSeen bool
 	startFailedCalls  int
@@ -1063,7 +1130,7 @@ func newRunAgentProcessAsyncFailureFixture(t *testing.T) *runAgentProcessAsyncFa
 	repo.sessions["session-123"] = &models.TaskSession{
 		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting,
 	}
-	f := &runAgentProcessAsyncFailureFixture{stopCh: make(chan struct{})}
+	f := &runAgentProcessAsyncFailureFixture{repo: repo, stopCh: make(chan struct{})}
 	agentManager := &mockAgentManager{
 		startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error {
 			return fmt.Errorf("ACP initialize handshake failed: context deadline exceeded")
@@ -1113,8 +1180,8 @@ func TestRunAgentProcessAsync_ResumeDoesNotEscalateTaskState(t *testing.T) {
 	if !f.sessionFailedSeen {
 		t.Error("expected session state FAILED")
 	}
-	if len(f.taskStateUpdates) != 0 {
-		t.Errorf("expected no task state updates on resume failure, got %v", f.taskStateUpdates)
+	if len(f.taskStateUpdates) != 1 || f.taskStateUpdates[0] != string(v1.TaskStateReview) {
+		t.Errorf("expected resume failure to reconcile task state to REVIEW, got %v", f.taskStateUpdates)
 	}
 	if f.startFailedCalls != 1 {
 		t.Errorf("expected onAgentStartFailed called once, got %d", f.startFailedCalls)
@@ -1124,6 +1191,244 @@ func TestRunAgentProcessAsync_ResumeDoesNotEscalateTaskState(t *testing.T) {
 	}
 }
 
+func TestRunAgentProcessAsync_ResumeFailureDoesNotReviewWhileSiblingWorks(t *testing.T) {
+	f := newRunAgentProcessAsyncFailureFixture(t)
+	f.repo.sessions["session-456"] = &models.TaskSession{
+		ID: "session-456", TaskID: "task-123", State: models.TaskSessionStateRunning,
+	}
+
+	f.exec.runAgentProcessAsync(context.Background(), "task-123", "session-123", "exec-456",
+		func(ctx context.Context) { t.Error("onSuccess should not run on failure") },
+		false, true)
+	f.awaitStop(t)
+
+	if !f.sessionFailedSeen {
+		t.Error("expected session state FAILED")
+	}
+	if len(f.taskStateUpdates) != 0 {
+		t.Errorf("expected working sibling to block REVIEW reconcile, got %v", f.taskStateUpdates)
+	}
+}
+
+func TestRunAgentProcessAsync_ResumeFailureUsesReviewReconcileCallback(t *testing.T) {
+	f := newRunAgentProcessAsyncFailureFixture(t)
+	var reviewCalls []string
+	f.exec.SetOnTaskReviewStateReconcile(func(ctx context.Context, taskID, completedSessionID string) {
+		reviewCalls = append(reviewCalls, taskID+"/"+completedSessionID)
+	})
+
+	f.exec.runAgentProcessAsync(context.Background(), "task-123", "session-123", "exec-456",
+		func(ctx context.Context) { t.Error("onSuccess should not run on failure") },
+		false, true)
+	f.awaitStop(t)
+
+	if len(reviewCalls) != 1 || reviewCalls[0] != "task-123/session-123" {
+		t.Fatalf("expected guarded review reconcile callback once, got %v", reviewCalls)
+	}
+	if len(f.taskStateUpdates) != 0 {
+		t.Fatalf("expected callback to replace direct task REVIEW writes, got %v", f.taskStateUpdates)
+	}
+}
+
+func TestWriteTaskReviewStateIfNoWorkingSessionsSkipsSameSessionActiveAgain(t *testing.T) {
+	repo := newMockRepository()
+	repo.sessions["session-123"] = &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateRunning,
+	}
+	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	var taskStateUpdates []v1.TaskState
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateUpdates = append(taskStateUpdates, state)
+		return nil
+	})
+
+	exec.writeTaskReviewStateIfNoWorkingSessions(context.Background(), "task-123", "session-123")
+
+	if len(taskStateUpdates) != 0 {
+		t.Errorf("expected same active session to block REVIEW reconcile, got %v", taskStateUpdates)
+	}
+}
+
+func TestWriteTaskReviewStateIfNoWorkingSessionsSkipsOnFailedSessionReadError(t *testing.T) {
+	repo := newMockRepository()
+	repo.tasks["task-123"] = &models.Task{ID: "task-123"}
+	repo.getTaskSessionFunc = func(context.Context, string) (*models.TaskSession, error) {
+		return nil, errors.New("temporary session read failure")
+	}
+	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	var taskStateUpdates []v1.TaskState
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateUpdates = append(taskStateUpdates, state)
+		return nil
+	})
+
+	exec.writeTaskReviewStateIfNoWorkingSessions(context.Background(), "task-123", "session-123")
+
+	if len(taskStateUpdates) != 0 {
+		t.Errorf("expected failed session read to block REVIEW reconcile, got %v", taskStateUpdates)
+	}
+}
+
+func TestWriteTaskReviewStateIfNoWorkingSessionsSkipsOfficeTask(t *testing.T) {
+	repo := newMockRepository()
+	repo.tasks["task-123"] = &models.Task{
+		ID:                     "task-123",
+		AssigneeAgentProfileID: "agent-profile-123",
+	}
+	repo.sessions["session-123"] = &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateFailed,
+	}
+	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	var taskStateUpdates []v1.TaskState
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateUpdates = append(taskStateUpdates, state)
+		return nil
+	})
+
+	exec.writeTaskReviewStateIfNoWorkingSessions(context.Background(), "task-123", "session-123")
+
+	if len(taskStateUpdates) != 0 {
+		t.Errorf("expected office task to keep workflow state, got %v", taskStateUpdates)
+	}
+}
+
+func TestStartAgentProcessOnResumePromotesTaskAfterSuccess(t *testing.T) {
+	repo := newMockRepository()
+	session := &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting,
+	}
+	repo.sessions[session.ID] = session
+
+	taskStateCh := make(chan v1.TaskState, 1)
+	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateCh <- state
+		return nil
+	})
+
+	exec.startAgentProcessOnResume(context.Background(), "task-123", session, "exec-456")
+
+	select {
+	case state := <-taskStateCh:
+		if state != v1.TaskStateInProgress {
+			t.Fatalf("task state = %q, want %q", state, v1.TaskStateInProgress)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected resume process success to promote task to IN_PROGRESS")
+	}
+}
+
+func TestStartAgentProcessOnResumeSkipsOfficeTaskPromotion(t *testing.T) {
+	repo := newMockRepository()
+	repo.tasks["task-123"] = &models.Task{ID: "task-123", AssigneeAgentProfileID: "agent-profile-123"}
+	session := &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting,
+	}
+	repo.sessions[session.ID] = session
+
+	startedCh := make(chan struct{})
+	agentManager := &mockAgentManager{
+		startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error {
+			close(startedCh)
+			return nil
+		},
+	}
+	taskStateCh := make(chan v1.TaskState, 1)
+	exec := newTestExecutor(t, agentManager, repo)
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateCh <- state
+		return nil
+	})
+
+	exec.startAgentProcessOnResume(context.Background(), "task-123", session, "exec-456")
+
+	select {
+	case <-startedCh:
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected resume process start")
+	}
+	select {
+	case state := <-taskStateCh:
+		t.Fatalf("office resume promoted task state to %q", state)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestStartAgentProcessOnResumeSkipsCancelledSessionPromotion(t *testing.T) {
+	repo := newMockRepository()
+	session := &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting,
+	}
+	repo.sessions[session.ID] = session
+
+	startedCh := make(chan struct{})
+	agentManager := &mockAgentManager{
+		startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error {
+			if err := repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateCancelled, "stopped"); err != nil {
+				return err
+			}
+			close(startedCh)
+			return nil
+		},
+	}
+	taskStateCh := make(chan v1.TaskState, 1)
+	exec := newTestExecutor(t, agentManager, repo)
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateCh <- state
+		return nil
+	})
+
+	exec.startAgentProcessOnResume(context.Background(), "task-123", session, "exec-456")
+
+	select {
+	case <-startedCh:
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected resume process start")
+	}
+	select {
+	case state := <-taskStateCh:
+		t.Fatalf("cancelled resume promoted task state to %q", state)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestStartAgentProcessOnResumeSkipsArchivedTaskPromotion(t *testing.T) {
+	repo := newMockRepository()
+	archivedAt := time.Now()
+	repo.tasks["task-123"] = &models.Task{ID: "task-123", ArchivedAt: &archivedAt}
+	session := &models.TaskSession{
+		ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting,
+	}
+	repo.sessions[session.ID] = session
+
+	startedCh := make(chan struct{})
+	agentManager := &mockAgentManager{
+		startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error {
+			close(startedCh)
+			return nil
+		},
+	}
+	taskStateCh := make(chan v1.TaskState, 1)
+	exec := newTestExecutor(t, agentManager, repo)
+	exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error {
+		taskStateCh <- state
+		return nil
+	})
+
+	exec.startAgentProcessOnResume(context.Background(), "task-123", session, "exec-456")
+
+	select {
+	case <-startedCh:
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected resume process start")
+	}
+	select {
+	case state := <-taskStateCh:
+		t.Fatalf("archived resume promoted task state to %q", state)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
 func TestRunAgentProcessAsync_FreshStartEscalatesTaskState(t *testing.T) {
 	f := newRunAgentProcessAsyncFailureFixture(t)
 	f.exec.runAgentProcessAsync(context.Background(), "task-123", "session-123", "exec-456",
@@ -1268,7 +1573,9 @@ func TestPersistResumeState_SetsStartingState(t *testing.T) {
 		}
 		repo.sessions[session.ID] = session
 
-		executor.persistResumeState(context.Background(), "task-1", session, true)
+		if err := executor.persistResumeState(context.Background(), "task-1", session, true); err != nil {
+			t.Fatalf("persistResumeState: %v", err)
+		}
 
 		if session.State != models.TaskSessionStateStarting {
 			t.Errorf("expected state STARTING, got %s", session.State)
@@ -1278,6 +1585,35 @@ func TestPersistResumeState_SetsStartingState(t *testing.T) {
 		}
 	})
 
+	t.Run("defers task promotion when startAgent is true", func(t *testing.T) {
+		session := &models.TaskSession{
+			ID:        "session-promote",
+			TaskID:    "task-1",
+			State:     models.TaskSessionStateWaitingForInput,
+			UpdatedAt: now,
+		}
+		repo.sessions[session.ID] = session
+		var gotPromoteTask *bool
+		executor.SetOnSessionStarting(func(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error {
+			gotPromoteTask = &promoteTask
+			return repo.UpdateTaskSession(ctx, session)
+		})
+		t.Cleanup(func() {
+			executor.SetOnSessionStarting(nil)
+		})
+
+		if err := executor.persistResumeState(context.Background(), "task-1", session, true); err != nil {
+			t.Fatalf("persistResumeState: %v", err)
+		}
+
+		if gotPromoteTask == nil {
+			t.Fatal("expected onSessionStarting callback")
+		}
+		if *gotPromoteTask {
+			t.Fatal("resume STARTING persistence must defer task promotion until process start succeeds")
+		}
+	})
+
 	t.Run("does not change state when startAgent is false", func(t *testing.T) {
 		session := &models.TaskSession{
 			ID:        "session-2",
@@ -1287,7 +1623,9 @@ func TestPersistResumeState_SetsStartingState(t *testing.T) {
 		}
 		repo.sessions[session.ID] = session
 
-		executor.persistResumeState(context.Background(), "task-1", session, false)
+		if err := executor.persistResumeState(context.Background(), "task-1", session, false); err != nil {
+			t.Fatalf("persistResumeState: %v", err)
+		}
 
 		if session.State != models.TaskSessionStateWaitingForInput {
 			t.Errorf("expected state WAITING_FOR_INPUT, got %s", session.State)

apps/backend/internal/orchestrator/service.go

@@ -92,6 +92,7 @@ type MessageCreator interface {
 type TurnService interface {
 	StartTurn(ctx context.Context, sessionID string) (*models.Turn, error)
 	CompleteTurn(ctx context.Context, turnID string) error
+	GetTurn(ctx context.Context, turnID string) (*models.Turn, error)
 	GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error)
 	UpdateTurn(ctx context.Context, turn *models.Turn) error
 	// AbandonOpenTurns buries any open turns for a session by setting
@@ -364,6 +365,19 @@ type Service struct {
 	// Active turns map: sessionID -> turnID
 	activeTurns sync.Map
 
+	// taskRuntimeStateMu serializes task-state flips derived from session
+	// runtime state. Without it, a completion/cancel path can check for active
+	// sibling sessions just before another handler marks one RUNNING, then
+	// clobber the task back to REVIEW while work is active.
+	taskRuntimeStateMu sync.Mutex
+
+	// completedExecutions records execution IDs that have reached a terminal
+	// agent lifecycle event. Buffered stream/tool events for these executions
+	// must not wake their session back to RUNNING after the terminal path makes
+	// it promptable again. Entries expire after a short grace window so the
+	// guard does not grow without bound in long-running backend processes.
+	completedExecutions sync.Map
+
 	// Session reset flags: sessionID -> true while resetAgentContext is restarting process.
 	// Used to suppress stale ready events and avoid draining queued prompts mid-reset.
 	resetInProgressSessions sync.Map
@@ -479,6 +493,12 @@ func NewService(
 		s.updateTaskSessionState(ctx, taskID, sessionID, state, errorMessage, true)
 		return nil
 	})
+	exec.SetOnSessionStarting(func(ctx context.Context, taskID string, session *models.TaskSession, promoteTask bool) error {
+		return s.setSessionStarting(ctx, taskID, session, promoteTask)
+	})
+	exec.SetOnTaskReviewStateReconcile(func(ctx context.Context, taskID, completedSessionID string) {
+		s.writeTaskReviewState(ctx, taskID, completedSessionID)
+	})
 	exec.SetOnLaunchFailed(s.handleSessionLaunchFailed)
 	exec.SetOnAgentStartFailed(s.handleAgentStartFailed)
 	if caps, ok := agentManager.(executor.ExecutorTypeCapabilities); ok {

apps/backend/internal/orchestrator/sessionstate/sessionstate.go

@@ -0,0 +1,7 @@
+package sessionstate
+
+import "github.com/kandev/kandev/internal/task/models"
+
+func IsWorking(state models.TaskSessionState) bool {
+	return state == models.TaskSessionStateRunning || state == models.TaskSessionStateStarting
+}

apps/backend/internal/orchestrator/task_operations.go

@@ -2252,7 +2252,7 @@ func (s *Service) handlePromptError(ctx context.Context, taskID, sessionID strin
 	// in progress while it retries — so don't flap it to REVIEW here.
 	if !isTransientPromptError(err) && !errors.Is(err, lifecycle.ErrCancelEscalated) &&
 		!routingerr.IsTransientProviderError(err.Error()) {
-		_ = s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview)
+		s.writeTaskReviewState(ctx, taskID, sessionID)
 	}
 	s.completeTurnForSession(ctx, sessionID)
 	return err
@@ -2450,7 +2450,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
 	// cancelled turn is treated as finished work the user may want to review.
 	if session != nil {
 		s.updateTaskSessionState(ctx, session.TaskID, sessionID, models.TaskSessionStateWaitingForInput, "", true, session)
-		s.writeTaskReviewStateOnCancel(ctx, session.TaskID)
+		s.writeTaskReviewStateOnCancel(ctx, session.TaskID, sessionID)
 	}
 
 	// Record cancellation in the message history

apps/backend/internal/orchestrator/task_operations_test.go

@@ -813,9 +813,15 @@ func TestStartCreatedSession_EmptyProfileFallsBackToWorkflowDefault(t *testing.T
 // mockMessageCreator implements MessageCreator for testing.
 // Only CreateUserMessage is tracked; all other methods are no-op stubs.
 type mockMessageCreator struct {
-	userMessages    []mockUserMessage
-	sessionMessages []mockSessionMessage
-	userMessageErr  error
+	userMessages       []mockUserMessage
+	sessionMessages    []mockSessionMessage
+	agentMessages      []mockAgentMessage
+	agentMessageWrites int
+	agentStreamWrites  int
+	thinkingWrites     int
+	toolCallWrites     int
+	toolUpdateWrites   int
+	userMessageErr     error
 }
 
 type mockUserMessage struct {
@@ -829,6 +835,10 @@ type mockSessionMessage struct {
 	requestsInput                                   bool
 }
 
+type mockAgentMessage struct {
+	taskID, content, sessionID, turnID string
+}
+
 func (m *mockMessageCreator) CreateUserMessage(_ context.Context, taskID, content, sessionID, turnID string, metadata map[string]interface{}) error {
 	if m.userMessageErr != nil {
 		return m.userMessageErr
@@ -837,15 +847,19 @@ func (m *mockMessageCreator) CreateUserMessage(_ context.Context, taskID, conten
 	return nil
 }
 
-func (m *mockMessageCreator) CreateAgentMessage(context.Context, string, string, string, string) error {
+func (m *mockMessageCreator) CreateAgentMessage(_ context.Context, taskID, content, sessionID, turnID string) error {
+	m.agentMessages = append(m.agentMessages, mockAgentMessage{taskID, content, sessionID, turnID})
+	m.agentMessageWrites++
 	return nil
 }
 
 func (m *mockMessageCreator) CreateToolCallMessage(context.Context, string, string, string, string, string, string, string, *streams.NormalizedPayload) error {
+	m.toolCallWrites++
 	return nil
 }
 
 func (m *mockMessageCreator) UpdateToolCallMessage(context.Context, string, string, string, string, string, string, string, string, string, *streams.NormalizedPayload) error {
+	m.toolUpdateWrites++
 	return nil
 }
 
@@ -871,18 +885,22 @@ func (m *mockMessageCreator) UpdatePermissionMessage(context.Context, string, st
 }
 
 func (m *mockMessageCreator) CreateAgentMessageStreaming(context.Context, string, string, string, string, string) error {
+	m.agentStreamWrites++
 	return nil
 }
 
 func (m *mockMessageCreator) AppendAgentMessage(context.Context, string, string) error {
+	m.agentStreamWrites++
 	return nil
 }
 
 func (m *mockMessageCreator) CreateThinkingMessageStreaming(context.Context, string, string, string, string, string) error {
+	m.thinkingWrites++
 	return nil
 }
 
 func (m *mockMessageCreator) AppendThinkingMessage(context.Context, string, string) error {
+	m.thinkingWrites++
 	return nil
 }
 func (m *mockMessageCreator) InvalidateModelCache(string) {}

apps/backend/internal/orchestrator/turn_lifecycle_test.go

@@ -41,6 +41,10 @@ func (a *repoTurnService) CompleteTurn(ctx context.Context, turnID string) error
 	return a.repo.CompleteTurn(ctx, turnID)
 }
 
+func (a *repoTurnService) GetTurn(ctx context.Context, turnID string) (*models.Turn, error) {
+	return a.repo.GetTurn(ctx, turnID)
+}
+
 func (a *repoTurnService) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) {
 	turn, err := a.repo.GetActiveTurnBySessionID(ctx, sessionID)
 	if errors.Is(err, sql.ErrNoRows) {

apps/backend/internal/orchestrator/workflow_e2e_test.go

@@ -434,6 +434,7 @@ func TestProcessOnTurnCompleteViaEngine_WritesTaskStateReview(t *testing.T) {
 
 	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
 	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
 	svc := createEngineService(t, repo, sg, agentMgr)
 	svc.taskRepo = taskRepo
 

apps/backend/internal/task/service/service_office.go

@@ -42,6 +42,27 @@ func (s *Service) GetLastAgentMessage(ctx context.Context, sessionID string) (st
 	return s.sessions.GetLastAgentMessage(ctx, sessionID)
 }
 
+// GetLastAgentMessageForTurn returns the most recent text agent message for a
+// single turn. Used by the office comment bridge when a late terminal complete
+// refers to a historical turn while a newer turn may already be active.
+func (s *Service) GetLastAgentMessageForTurn(ctx context.Context, turnID string) (string, error) {
+	messages, err := s.messages.ListMessagesByTurnID(ctx, turnID)
+	if err != nil {
+		return "", err
+	}
+	for i := len(messages) - 1; i >= 0; i-- {
+		message := messages[i]
+		if message == nil ||
+			message.AuthorType != models.MessageAuthorAgent ||
+			message.Type != models.MessageTypeMessage ||
+			message.Content == "" {
+			continue
+		}
+		return message.Content, nil
+	}
+	return "", nil
+}
+
 // ListTaskTree returns a flat list of non-archived tasks for tree building.
 func (s *Service) ListTaskTree(ctx context.Context, workspaceID string, filters models.TaskTreeFilters) ([]*models.Task, error) {
 	tasks, err := s.tasks.ListTaskTree(ctx, workspaceID, filters)

apps/backend/internal/task/service/service_turns.go

@@ -54,6 +54,11 @@ func (s *Service) StartTurn(ctx context.Context, sessionID string) (*models.Turn
 	return turn, nil
 }
 
+// GetTurn returns a turn by ID.
+func (s *Service) GetTurn(ctx context.Context, turnID string) (*models.Turn, error) {
+	return s.turns.GetTurn(ctx, turnID)
+}
+
 // CompleteTurn marks a turn as completed and publishes the turn.completed event.
 func (s *Service) CompleteTurn(ctx context.Context, turnID string) error {
 	if turnID == "" {