From c041dbee0eaaa97b1e643662aa2159598bc73ea3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 23 Jul 2026 14:12:58 +0100 Subject: [PATCH] fix(orchestrator): separate office task session ownership (#1893) --- .../backend/internal/mcp/handlers/handlers.go | 2 +- .../mcp/handlers/message_task_test.go | 36 ++++++ .../coordinator_stop_edges_test.go | 1 + .../orchestrator/event_handlers_streaming.go | 8 +- .../orchestrator/executor/executor_execute.go | 4 +- .../executor/executor_interaction_test.go | 1 + .../orchestrator/executor/executor_resume.go | 2 +- .../executor/executor_resume_test.go | 21 +++- .../orchestrator/executor/executor_test.go | 3 +- .../orchestrator/office_turn_complete_test.go | 3 +- .../internal/orchestrator/session_ensure.go | 2 +- .../session_ensure_office_test.go | 49 +++++++- .../internal/orchestrator/task_operations.go | 13 +- .../orchestrator/task_operations_test.go | 111 ++++++++++++++++++ .../task/handlers/message_handlers.go | 2 +- .../task/handlers/message_handlers_test.go | 19 +++ apps/backend/internal/task/models/models.go | 7 ++ .../internal/task/models/models_test.go | 20 ++++ .../0011-transient-provider-error-retry.md | 3 +- 19 files changed, 281 insertions(+), 26 deletions(-) diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 651a63fd31..7f4dcb8be0 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -2467,7 +2467,7 @@ func (h *Handlers) ensureTaskInProgressForTaskMessage(ctx context.Context, taskI if task.State != v1.TaskStateReview { return rollback, nil } - if task.AssigneeAgentProfileID != "" { + if task.IsOfficeOwnedAndAssigned() { return rollback, nil } if _, err := h.taskSvc.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { diff --git a/apps/backend/internal/mcp/handlers/message_task_test.go b/apps/backend/internal/mcp/handlers/message_task_test.go index ad19772737..cdb7848aa6 100644 --- a/apps/backend/internal/mcp/handlers/message_task_test.go +++ b/apps/backend/internal/mcp/handlers/message_task_test.go @@ -963,6 +963,40 @@ func TestHandleMessageTask_WaitingForInput_FiresTurnStart(t *testing.T) { assert.Equal(t, sess.ID, orch.promptCalls[0].sessionID) } +func TestHandleMessageTask_KanbanRunnerTransitionsReviewToInProgress(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + sender, target, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateWaitingForInput) + + task, err := svc.GetTask(ctx, target.ID) + require.NoError(t, err) + task.State = v1.TaskStateReview + task.WorkflowStepID = "step-review" + task.AssigneeAgentProfileID = "kanban-runner" + require.NoError(t, repo.UpdateTask(ctx, task)) + + h, orch := newMessageTaskHandler(t, svc) + orch.onTurnStart = func(ctx context.Context, taskID, sessionID string) error { + assert.Equal(t, target.ID, taskID) + assert.Equal(t, sess.ID, sessionID) + updatedTask, err := svc.GetTask(ctx, taskID) + require.NoError(t, err) + assert.Equal(t, v1.TaskStateInProgress, updatedTask.State) + return nil + } + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayload(target.ID, "review follow-up", sender.ID)) + resp, err := h.handleMessageTask(ctx, msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + updatedTask, err := svc.GetTask(ctx, target.ID) + require.NoError(t, err) + assert.Equal(t, v1.TaskStateInProgress, updatedTask.State) + assert.Equal(t, "step-review", updatedTask.WorkflowStepID) +} + func TestHandleMessageTask_WaitingForInput_UsesSessionSelectedByTurnStart(t *testing.T) { ctx := context.Background() svc, repo := newTestTaskService(t) @@ -1616,6 +1650,7 @@ func TestHandleMessageTask_OfficeReviewDoesNotTransitionTaskState(t *testing.T) require.NoError(t, err) task.State = v1.TaskStateReview task.WorkflowStepID = "step-review" + task.ProjectID = "office-project" task.AssigneeAgentProfileID = "agent-profile-1" require.NoError(t, repo.UpdateTask(ctx, task)) @@ -1645,6 +1680,7 @@ func TestHandleMessageTask_OfficeDispatchErrorRestoresWorkflowStep(t *testing.T) require.NoError(t, err) task.State = v1.TaskStateReview task.WorkflowStepID = "step-review" + task.ProjectID = "office-project" task.AssigneeAgentProfileID = "agent-profile-1" require.NoError(t, repo.UpdateTask(ctx, task)) diff --git a/apps/backend/internal/orchestrator/coordinator_stop_edges_test.go b/apps/backend/internal/orchestrator/coordinator_stop_edges_test.go index 0206ba5186..7cf5561f86 100644 --- a/apps/backend/internal/orchestrator/coordinator_stop_edges_test.go +++ b/apps/backend/internal/orchestrator/coordinator_stop_edges_test.go @@ -60,6 +60,7 @@ func TestStopTaskForCoordinator_ReviewGuardsPreserveTaskState(t *testing.T) { return task, err } copyTask := *task + copyTask.IsFromOffice = true copyTask.AssigneeAgentProfileID = "office-agent" return ©Task, nil }, diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 08eff6dc8c..592e457bb0 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -1089,7 +1089,7 @@ func (s *Service) writeTaskReviewState(ctx context.Context, taskID, completedSes zap.String("task_id", taskID), zap.Error(err)) return - } else if dbTask != nil && dbTask.AssigneeAgentProfileID != "" { + } else if dbTask.IsOfficeOwnedAndAssigned() { s.logger.Debug("skipping REVIEW transition for office task", zap.String("task_id", taskID)) return @@ -1184,7 +1184,7 @@ func (s *Service) writeTaskReviewStateOnCancel(ctx context.Context, taskID, sess } return } - if dbTask.AssigneeAgentProfileID != "" { + if dbTask.IsOfficeOwnedAndAssigned() { return } if taskArchived(dbTask) { @@ -1327,7 +1327,7 @@ func (s *Service) reconcileTaskStateForRuntimeLocked( if taskArchived(task) { return nil } - if state == v1.TaskStateInProgress && task != nil && task.AssigneeAgentProfileID != "" { + if state == v1.TaskStateInProgress && task.IsOfficeOwnedAndAssigned() { return nil } session, err := s.repo.GetTaskSession(ctx, sessionID) @@ -1572,7 +1572,7 @@ func (s *Service) handleOfficeTurnComplete( return false } task, err := s.repo.GetTask(ctx, taskID) - if err != nil || task == nil || task.AssigneeAgentProfileID == "" { + if err != nil || !task.IsOfficeOwnedAndAssigned() { return false } diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index af4773d4ad..81885bbb64 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -44,7 +44,7 @@ func (e *Executor) resolveTaskSessionMCPMode(ctx context.Context, taskID string, if err != nil { return "", fmt.Errorf("load task for MCP mode: %w", err) } - if task != nil && task.AssigneeAgentProfileID != "" { + if task.IsOfficeOwnedAndAssigned() { return McpModeOffice, nil } return "", nil @@ -292,7 +292,7 @@ func (e *Executor) shouldSkipFailedStartReviewForTask(ctx context.Context, taskI zap.Error(err)) return true } - if task != nil && task.AssigneeAgentProfileID != "" { + if task.IsOfficeOwnedAndAssigned() { e.logger.Debug("skipping failed-start task REVIEW state for office task", zap.String("task_id", taskID), zap.String("session_id", failedSessionID)) diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go index e1d7db2c4e..46dced363f 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go @@ -502,6 +502,7 @@ func TestSwitchModelFallback_PreservesRestrictedMCPMode(t *testing.T) { ID: "task-office", WorkspaceID: "workspace-1", Title: "Office task", + IsFromOffice: true, AssigneeAgentProfileID: "office-agent", } repo.sessions["session-office"] = &models.TaskSession{ diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 9729a06bc8..23e82e023d 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume.go @@ -992,7 +992,7 @@ func (e *Executor) writeTaskInProgressForRuntime(ctx context.Context, taskID, se zap.String("task_id", taskID)) return nil } - if task != nil && task.AssigneeAgentProfileID != "" { + if task.IsOfficeOwnedAndAssigned() { e.logger.Debug("skipping IN_PROGRESS transition for office task", zap.String("task_id", taskID)) return nil diff --git a/apps/backend/internal/orchestrator/executor/executor_resume_test.go b/apps/backend/internal/orchestrator/executor/executor_resume_test.go index a3daae8501..97b04849e0 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume_test.go @@ -74,15 +74,18 @@ func TestResumeSession_PassesResolvedTaskSessionMCPModeToAgentManager(t *testing tests := []struct { name string officeAssignee string + isFromOffice bool metadata map[string]interface{} wantMode string }{ {name: "regular task", wantMode: ""}, - {name: "Office task", officeAssignee: "office-agent", wantMode: McpModeOffice}, + {name: "Kanban task with runner", officeAssignee: "office-agent", wantMode: ""}, + {name: "Office task", officeAssignee: "office-agent", isFromOffice: true, wantMode: McpModeOffice}, {name: "Config session", metadata: map[string]interface{}{"config_mode": true}, wantMode: McpModeConfig}, { name: "Config session takes precedence for Office task", officeAssignee: "office-agent", + isFromOffice: true, metadata: map[string]interface{}{"config_mode": true}, wantMode: McpModeConfig, }, @@ -93,6 +96,7 @@ func TestResumeSession_PassesResolvedTaskSessionMCPModeToAgentManager(t *testing repo := newMockRepository() setupLiveResumeTestFixture(repo) repo.tasks["task-1"].AssigneeAgentProfileID = tt.officeAssignee + repo.tasks["task-1"].IsFromOffice = tt.isFromOffice repo.sessions["sess-1"].Metadata = tt.metadata var capturedReq *LaunchAgentRequest @@ -117,6 +121,21 @@ func TestResumeSession_PassesResolvedTaskSessionMCPModeToAgentManager(t *testing } } +func TestWriteTaskInProgressForRuntime_KanbanRunnerUpdatesTaskState(t *testing.T) { + repo := newMockRepository() + setupLiveResumeTestFixture(repo) + repo.tasks["task-1"].AssigneeAgentProfileID = "copilot-runner" + repo.sessions["sess-1"].State = models.TaskSessionStateRunning + + exec := newTestExecutor(t, &mockAgentManager{}, repo) + if err := exec.writeTaskInProgressForRuntime(context.Background(), "task-1", "sess-1"); err != nil { + t.Fatalf("writeTaskInProgressForRuntime: %v", err) + } + if len(repo.updateTaskStateIfNotArchivedCalls) != 1 { + t.Fatalf("runtime state writes = %d, want 1", len(repo.updateTaskStateIfNotArchivedCalls)) + } +} + // TestResumeSession_LiveAgentReturnsAlreadyRunning ensures ResumeSession returns // ErrExecutionAlreadyRunning instead of killing the live agent subprocess when // a live agent is already registered for the session. diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index f3aa5a3b59..5460c9b0cb 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -2074,6 +2074,7 @@ func TestWriteTaskReviewStateIfNoWorkingSessionsSkipsOfficeTask(t *testing.T) { repo := newMockRepository() repo.tasks["task-123"] = &models.Task{ ID: "task-123", + IsFromOffice: true, AssigneeAgentProfileID: "agent-profile-123", } repo.sessions["session-123"] = &models.TaskSession{ @@ -2145,7 +2146,7 @@ func TestStartAgentProcessOnResumePromotesTaskAfterSuccess(t *testing.T) { func TestStartAgentProcessOnResumeSkipsOfficeTaskPromotion(t *testing.T) { repo := newMockRepository() - repo.tasks["task-123"] = &models.Task{ID: "task-123", AssigneeAgentProfileID: "agent-profile-123"} + repo.tasks["task-123"] = &models.Task{ID: "task-123", IsFromOffice: true, AssigneeAgentProfileID: "agent-profile-123"} session := &models.TaskSession{ ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting, } diff --git a/apps/backend/internal/orchestrator/office_turn_complete_test.go b/apps/backend/internal/orchestrator/office_turn_complete_test.go index 0044f40e9c..174f0c0cd6 100644 --- a/apps/backend/internal/orchestrator/office_turn_complete_test.go +++ b/apps/backend/internal/orchestrator/office_turn_complete_test.go @@ -40,7 +40,8 @@ func seedOfficeSession(t *testing.T, repo officeSeedRepo, taskID, sessionID, age WorkflowStepID: stepID, Title: "Office task", State: v1.TaskStateInProgress, - AssigneeAgentProfileID: "agent-1", // marks the task as office-style (per-(task, agent) session) + ProjectID: "project-office", + AssigneeAgentProfileID: "agent-1", CreatedAt: now, UpdatedAt: now, } diff --git a/apps/backend/internal/orchestrator/session_ensure.go b/apps/backend/internal/orchestrator/session_ensure.go index 1ff498e797..316b156998 100644 --- a/apps/backend/internal/orchestrator/session_ensure.go +++ b/apps/backend/internal/orchestrator/session_ensure.go @@ -141,7 +141,7 @@ func (s *Service) findExistingSession(ctx context.Context, taskID string) *Ensur // not here — keeping findExistingSession a pure lookup.) func (s *Service) findOfficeSessionForResume(ctx context.Context, taskID string) *EnsureSessionResponse { task, err := s.repo.GetTask(ctx, taskID) - if err != nil || task == nil || task.AssigneeAgentProfileID == "" { + if err != nil || !task.IsOfficeOwnedAndAssigned() { return nil } agentID := s.agentForViewer(ctx, task) diff --git a/apps/backend/internal/orchestrator/session_ensure_office_test.go b/apps/backend/internal/orchestrator/session_ensure_office_test.go index 50326a711c..68bde00025 100644 --- a/apps/backend/internal/orchestrator/session_ensure_office_test.go +++ b/apps/backend/internal/orchestrator/session_ensure_office_test.go @@ -83,16 +83,55 @@ func TestFindExistingSession_OfficeTaskWithViewerAgent(t *testing.T) { // is_primary lookup intact under the new gating. func TestFindExistingSession_KanbanTask_UsesPrimary(t *testing.T) { repo := setupTestRepo(t) - seedSession(t, repo, "task-k", "sess-k", "step1") + ctx := context.Background() + now := time.Now().UTC() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-k", Name: "Kanban", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatal(err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-k", WorkspaceID: "ws-k", Name: "Kanban", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatal(err) + } + if err := seedWorkflowStep(t, repo, "wfs-k"); err != nil { + t.Fatal(err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-k", + WorkspaceID: "ws-k", + WorkflowID: "wf-k", + WorkflowStepID: "wfs-k", + State: v1.TaskStateInProgress, + AssigneeAgentProfileID: "agent-runner", + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatal(err) + } + for _, session := range []*models.TaskSession{ + {ID: "sess-primary", TaskID: "task-k", AgentProfileID: "agent-primary", IsPrimary: true, State: models.TaskSessionStateIdle, StartedAt: now, UpdatedAt: now}, + {ID: "sess-runner", TaskID: "task-k", AgentProfileID: "agent-runner", State: models.TaskSessionStateIdle, StartedAt: now, UpdatedAt: now}, + } { + if err := repo.CreateTaskSession(ctx, session); err != nil { + t.Fatal(err) + } + } + task, err := repo.GetTask(ctx, "task-k") + if err != nil { + t.Fatal(err) + } + if task.IsFromOffice { + t.Fatal("Kanban task unexpectedly projected as Office-owned") + } svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) - resp := svc.findExistingSession(context.Background(), "task-k") + resp := svc.findExistingSession(ctx, "task-k") if resp == nil { t.Fatal("expected resume target, got nil") } - // Kanban / quick-chat sessions don't go through the office branch. - if resp.Source == "existing_office_agent" { - t.Errorf("kanban session must not surface as office") + if resp.SessionID != "sess-primary" { + t.Errorf("session_id: got %q want sess-primary", resp.SessionID) + } + if resp.Source != "existing_primary" { + t.Errorf("source: got %q want existing_primary", resp.Source) } } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 420d6a8346..c8290d7025 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -764,8 +764,7 @@ func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID s return execution, nil } -// isOfficeTask returns true when the task has an assignee agent profile, which -// identifies it as an office-managed task (as opposed to a kanban / quick-chat task). +// isOfficeTask returns true only for an office-owned task with an assignee. func (s *Service) isOfficeTask(ctx context.Context, taskID string) bool { isOfficeTask, err := s.lookupOfficeTask(ctx, taskID) return err == nil && isOfficeTask @@ -776,7 +775,7 @@ func (s *Service) lookupOfficeTask(ctx context.Context, taskID string) (bool, er if err != nil { return false, err } - return dbTask != nil && dbTask.AssigneeAgentProfileID != "", nil + return dbTask.IsOfficeOwnedAndAssigned(), nil } // prepareSessionForStart creates the session for a launch and propagates any @@ -818,15 +817,15 @@ func (s *Service) resolveIsPassthroughForLaunch(ctx context.Context, sessionID s } // createStartSession picks the right session-creation path for the task: -// office tasks with an assignee use the per-(task, agent) EnsureSessionForAgent -// (so runs reuse one row across turns); kanban / quick-chat fall through to -// the per-launch PrepareSession used since day one. +// office-owned tasks with an assignee use the per-(task, agent) +// EnsureSessionForAgent (so runs reuse one row across turns); kanban / +// quick-chat tasks fall through to the per-launch PrepareSession used since day one. func (s *Service) createStartSession( ctx context.Context, task *v1.Task, agentProfileID, executorID, executorProfileID, workflowStepID string, ) (string, error) { dbTask, err := s.repo.GetTask(ctx, task.ID) - if err == nil && dbTask != nil && dbTask.AssigneeAgentProfileID != "" { + if err == nil && dbTask.IsOfficeOwnedAndAssigned() { session, ensureErr := s.executor.EnsureSessionForAgent( ctx, task, dbTask.AssigneeAgentProfileID, agentProfileID, executorID, executorProfileID, ) diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 2c22a63392..b490e92370 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -65,6 +65,115 @@ func seedTaskAndSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessi } } +func TestCreateStartSession_KanbanRunnerCreatesDistinctSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-kanban", Name: "Kanban", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-kanban", WorkspaceID: "ws-kanban", Name: "Kanban", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + if err := seedWorkflowStep(t, repo, "step-kanban"); err != nil { + t.Fatalf("create workflow step: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-kanban", WorkspaceID: "ws-kanban", WorkflowID: "wf-kanban", WorkflowStepID: "step-kanban", + Title: "Kanban task", State: v1.TaskStateInProgress, AssigneeAgentProfileID: "copilot-runner", + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "existing-session", TaskID: "task-kanban", AgentProfileID: "copilot-runner", + State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create existing session: %v", err) + } + + task, err := repo.GetTask(ctx, "task-kanban") + if err != nil { + t.Fatalf("get task: %v", err) + } + if task.IsFromOffice { + t.Fatal("kanban task unexpectedly projected as office-owned") + } + if task.AssigneeAgentProfileID != "copilot-runner" { + t.Fatalf("runner = %q, want copilot-runner", task.AssigneeAgentProfileID) + } + + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + isOffice, err := svc.lookupOfficeTask(ctx, task.ID) + if err != nil { + t.Fatalf("lookup office task: %v", err) + } + if isOffice { + t.Fatal("kanban task with a runner was classified as office-owned") + } + sessionID, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "", "", "") + if err != nil { + t.Fatalf("create start session: %v", err) + } + if sessionID == "existing-session" { + t.Fatal("kanban launch reused the running runner session instead of creating a distinct session") + } +} + +func TestCreateStartSession_OfficeRunnerReusesPersistentSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-office", Name: "Office", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-office", WorkspaceID: "ws-office", Name: "Office", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + if err := seedWorkflowStep(t, repo, "step-office-start"); err != nil { + t.Fatalf("create workflow step: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-office", WorkspaceID: "ws-office", WorkflowID: "wf-office", WorkflowStepID: "step-office-start", + Title: "Office task", State: v1.TaskStateInProgress, ProjectID: "office-project", AssigneeAgentProfileID: "copilot-runner", + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "existing-office-session", TaskID: "task-office", AgentProfileID: "copilot-runner", + State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create existing session: %v", err) + } + + task, err := repo.GetTask(ctx, "task-office") + if err != nil { + t.Fatalf("get task: %v", err) + } + if !task.IsFromOffice { + t.Fatal("office task was not projected as office-owned") + } + + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + isOffice, err := svc.lookupOfficeTask(ctx, task.ID) + if err != nil { + t.Fatalf("lookup office task: %v", err) + } + if !isOffice { + t.Fatal("office-owned assigned task was not classified as office") + } + sessionID, err := svc.createStartSession(ctx, task.ToAPI(), "copilot-runner", "", "", "") + if err != nil { + t.Fatalf("create start session: %v", err) + } + if sessionID != "existing-office-session" { + t.Fatalf("office launch session = %q, want existing-office-session", sessionID) + } +} + func newCoordinatorStopTestService( repo repoStore, taskRepo scheduler.TaskRepository, @@ -2311,6 +2420,7 @@ func TestStartCreatedSession_OfficeTaskSkipsSchedulingState(t *testing.T) { } dbTask.State = v1.TaskStateReview dbTask.WorkflowStepID = "step-office" + dbTask.ProjectID = "project-office" dbTask.AssigneeAgentProfileID = "office-agent" if err := repo.UpdateTask(ctx, dbTask); err != nil { t.Fatalf("update task: %v", err) @@ -3898,6 +4008,7 @@ func TestEnsureSessionRunning_PreparedOfficeWorkspaceSetsMCPMode(t *testing.T) { t.Fatalf("failed to load task: %v", err) } dbTask.WorkflowStepID = "step-office" + dbTask.ProjectID = "project-office" dbTask.AssigneeAgentProfileID = "office-agent" if err := repo.UpdateTask(ctx, dbTask); err != nil { t.Fatalf("failed to mark task as Office-owned: %v", err) diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 3e7583e2d8..d9bda3cb57 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -336,7 +336,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w requiresSignal := h.orchestrator != nil && h.orchestrator.StepRequiresCompletionSignal(ctx, req.TaskID) storedContent = sysprompt.InjectKandevContextWithOptions(req.TaskID, req.TaskSessionID, storedContent, sysprompt.KandevContextOptions{ RequiresCompletionSignal: requiresSignal, - IncludeCoordinatorTaskControls: task.AssigneeAgentProfileID == "" && !configMode, + IncludeCoordinatorTaskControls: !task.IsOfficeOwnedAndAssigned() && !configMode, }) } req.Content = storedContent diff --git a/apps/backend/internal/task/handlers/message_handlers_test.go b/apps/backend/internal/task/handlers/message_handlers_test.go index b64964cb51..e8b7421b38 100644 --- a/apps/backend/internal/task/handlers/message_handlers_test.go +++ b/apps/backend/internal/task/handlers/message_handlers_test.go @@ -404,6 +404,7 @@ func TestWSAddMessage_CreatedOfficeSessionOmitsCoordinatorTaskControls(t *testin content := runCreatedMessageContextTest(t, &models.Task{ ID: "t1", State: v1.TaskStateInProgress, + IsFromOffice: true, AssigneeAgentProfileID: "office-agent", UpdatedAt: now, }, &models.TaskSession{ @@ -417,6 +418,24 @@ func TestWSAddMessage_CreatedOfficeSessionOmitsCoordinatorTaskControls(t *testin "Office pre-wrap must not persist a task-mode-only tool") } +func TestWSAddMessage_CreatedKanbanRunnerIncludesCoordinatorTaskControls(t *testing.T) { + now := time.Now().UTC() + content := runCreatedMessageContextTest(t, &models.Task{ + ID: "t1", + State: v1.TaskStateInProgress, + AssigneeAgentProfileID: "kanban-runner", + UpdatedAt: now, + }, &models.TaskSession{ + ID: "s1", + TaskID: "t1", + State: models.TaskSessionStateCreated, + AgentProfileID: "profile-1", + UpdatedAt: now, + }) + assert.Contains(t, content, "stop_task_kandev", + "Kanban sessions retain coordinator task controls even with a projected runner") +} + func TestWSAddMessage_CreatedConfigSessionOmitsCoordinatorTaskControls(t *testing.T) { now := time.Now().UTC() content := runCreatedMessageContextTest(t, &models.Task{ diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 9e69d4983d..e31f161057 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -473,6 +473,13 @@ type Task struct { IsFromOffice bool `json:"is_from_office,omitempty"` } +// IsOfficeOwnedAndAssigned reports whether runtime behavior belongs to an +// Office-owned task with a designated runner. Kanban tasks may also project a +// runner from their workflow step, but retain normal per-session semantics. +func (t *Task) IsOfficeOwnedAndAssigned() bool { + return t != nil && t.IsFromOffice && t.AssigneeAgentProfileID != "" +} + // ChildCompletionRow is the compact active-child projection used to decide // whether a parent task's on_children_completed trigger is ready to fire. type ChildCompletionRow struct { diff --git a/apps/backend/internal/task/models/models_test.go b/apps/backend/internal/task/models/models_test.go index 3d04d45e22..9e0bcedf9c 100644 --- a/apps/backend/internal/task/models/models_test.go +++ b/apps/backend/internal/task/models/models_test.go @@ -358,6 +358,26 @@ func TestTaskIsFromOfficeField(t *testing.T) { } } +func TestTaskIsOfficeOwnedAndAssigned(t *testing.T) { + tests := []struct { + name string + task *Task + want bool + }{ + {name: "nil task", task: nil, want: false}, + {name: "kanban runner", task: &Task{AssigneeAgentProfileID: "runner"}, want: false}, + {name: "unassigned office task", task: &Task{IsFromOffice: true}, want: false}, + {name: "assigned office task", task: &Task{IsFromOffice: true, AssigneeAgentProfileID: "runner"}, want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.task.IsOfficeOwnedAndAssigned(); got != tt.want { + t.Errorf("IsOfficeOwnedAndAssigned() = %v, want %v", got, tt.want) + } + }) + } +} + // TestExecutorTypeRuntime pins the ExecutorType → agentruntime.Runtime // mapping table. Every executor variant the codebase ships must // declare its runtime explicitly here — a missing case falls through diff --git a/docs/decisions/0011-transient-provider-error-retry.md b/docs/decisions/0011-transient-provider-error-retry.md index 054e720414..ab4b439a93 100644 --- a/docs/decisions/0011-transient-provider-error-retry.md +++ b/docs/decisions/0011-transient-provider-error-retry.md @@ -42,7 +42,8 @@ calm, paced, visible retry — separate from generic agent failures. Cancel (`session.recover` action `cancel_retry`), it falls through to the existing red Resume / Start-fresh banner (with friendly "stayed overloaded after several retries" copy instead of the raw 529). Gated to non-office tasks - (`isOfficeTask`, i.e. `AssigneeAgentProfileID`) so genuine office tasks keep + (`isOfficeTask`, i.e. authoritative Office ownership plus an assigned runner) + so genuine office tasks keep their structured error UI. The transient branch runs *before* run-mode automation finalization in `handleAgentFailed` — it's the only non-terminal failure path, so finalizing (which reaps the ephemeral worktree) must wait