diff --git a/apps/backend/internal/backendapp/turn_adapters.go b/apps/backend/internal/backendapp/turn_adapters.go index eb554ec4a3..1d3c786cd4 100644 --- a/apps/backend/internal/backendapp/turn_adapters.go +++ b/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) } diff --git a/apps/backend/internal/integration/test_server_test.go b/apps/backend/internal/integration/test_server_test.go index 6dc4b6c134..c099e28795 100644 --- a/apps/backend/internal/integration/test_server_test.go +++ b/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) } diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index ce80db62b0..03ee0d8650 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/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 } diff --git a/apps/backend/internal/office/service/event_subscribers_fallback_test.go b/apps/backend/internal/office/service/event_subscribers_fallback_test.go index 15c756b217..2e523d7fbc 100644 --- a/apps/backend/internal/office/service/event_subscribers_fallback_test.go +++ b/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) + } + } +} diff --git a/apps/backend/internal/office/service/service.go b/apps/backend/internal/office/service/service.go index 372c877332..dafc559408 100644 --- a/apps/backend/internal/office/service/service.go +++ b/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 diff --git a/apps/backend/internal/office/service/workspace_deletion_test.go b/apps/backend/internal/office/service/workspace_deletion_test.go index 0c3eb08ad7..25a54b634f 100644 --- a/apps/backend/internal/office/service/workspace_deletion_test.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 6cbc8d5823..8e878d81ec 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/event_handlers_agent_clarification_test.go b/apps/backend/internal/orchestrator/event_handlers_agent_clarification_test.go index a962b37f1d..5c8579fa58 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent_clarification_test.go +++ b/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) } diff --git a/apps/backend/internal/orchestrator/event_handlers_clarification.go b/apps/backend/internal/orchestrator/event_handlers_clarification.go index 1677945638..d696a16756 100644 --- a/apps/backend/internal/orchestrator/event_handlers_clarification.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/event_handlers_clarification_test.go b/apps/backend/internal/orchestrator/event_handlers_clarification_test.go index 0cf2771c04..af881dbfd9 100644 --- a/apps/backend/internal/orchestrator/event_handlers_clarification_test.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 323627d749..fde3683cc4 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index afcfd0cf00..b5e72a6835 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/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") diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index a287dfbee6..a3da52ac14 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/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{ diff --git a/apps/backend/internal/orchestrator/event_handlers_transient_test.go b/apps/backend/internal/orchestrator/event_handlers_transient_test.go index f93768c63d..868f7bbc45 100644 --- a/apps/backend/internal/orchestrator/event_handlers_transient_test.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index daa8a722eb..38f8a2e1b6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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. diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 27a69bff83..3ca2eb0b7d 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index fee99c2d7e..3311c0a7de 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/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{ diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction.go b/apps/backend/internal/orchestrator/executor/executor_interaction.go index c5a74c7fe3..7900194dbc 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 1e37040b6f..a7d85add64 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/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) +} diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index abf4f5363c..5b7ae37798 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index e106053fda..9575260c82 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/sessionstate/sessionstate.go b/apps/backend/internal/orchestrator/sessionstate/sessionstate.go new file mode 100644 index 0000000000..a87fd8184f --- /dev/null +++ b/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 +} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 3e9a55c50e..54ad6260a6 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index ce906bc9b6..bb66d413c2 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/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) {} diff --git a/apps/backend/internal/orchestrator/turn_lifecycle_test.go b/apps/backend/internal/orchestrator/turn_lifecycle_test.go index 5e7ade0e6d..bd207dde96 100644 --- a/apps/backend/internal/orchestrator/turn_lifecycle_test.go +++ b/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) { diff --git a/apps/backend/internal/orchestrator/workflow_e2e_test.go b/apps/backend/internal/orchestrator/workflow_e2e_test.go index 2aee826cfe..bab74639bd 100644 --- a/apps/backend/internal/orchestrator/workflow_e2e_test.go +++ b/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 diff --git a/apps/backend/internal/task/service/service_office.go b/apps/backend/internal/task/service/service_office.go index 8e899b508a..955ac39d64 100644 --- a/apps/backend/internal/task/service/service_office.go +++ b/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) diff --git a/apps/backend/internal/task/service/service_turns.go b/apps/backend/internal/task/service/service_turns.go index 8323849c30..7442ee80b2 100644 --- a/apps/backend/internal/task/service/service_turns.go +++ b/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 == "" {