fix(orchestrator): separate office task session ownership (#1893)
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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))
|
||||
|
||||
|
||||
@@ -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
|
||||
},
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user