diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index acdbb79fdf..3a198dd4a5 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -156,6 +156,23 @@ func (r failSessionStateUpdateRepo) UpdateTaskSessionState( return r.err } +// UpdateTaskSessionStateIfCurrent must also fail: repoStore now declares this +// method, so the embedded repoStore field promotes it and +// persistStrictTaskSessionState's conditionalTaskSessionStateUpdater +// assertion succeeds, routing state transitions through this narrow CAS +// instead of the plain UpdateTaskSessionState this double otherwise +// overrides. Without this override, the transition would silently succeed +// against the embedded real repo instead of surfacing the injected failure. +func (r failSessionStateUpdateRepo) UpdateTaskSessionStateIfCurrent( + context.Context, + string, + models.TaskSessionState, + models.TaskSessionState, + string, +) (bool, time.Time, error) { + return false, time.Time{}, r.err +} + type failSetBaselineRepo struct { repoStore } diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index d8d09721d0..c3148d6e8b 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -56,6 +56,15 @@ type executorStore interface { SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error UpdateTaskSession(ctx context.Context, session *models.TaskSession) error UpdateTaskSessionIfCurrentState(ctx context.Context, session *models.TaskSession, expected models.TaskSessionState) (bool, error) + // UpdateTaskSessionStateIfCurrent transitions only the state-related + // columns (state, error_message, completed_at, updated_at) guarded by a + // CAS on the row's current state. Prefer this over + // UpdateTaskSessionIfCurrentState for a pure state transition: the latter + // writes the full row from the caller's (possibly stale) in-memory copy, + // which can silently revert a concurrent update to unrelated fields + // (profile snapshot, routing, execution identifiers) that landed between + // the caller's read and its write. + UpdateTaskSessionStateIfCurrent(ctx context.Context, id string, expected, status models.TaskSessionState, errorMessage string) (bool, time.Time, error) UpdateTaskSessionState(ctx context.Context, id string, state models.TaskSessionState, errorMessage string) error UpdateTaskSessionBaseCommit(ctx context.Context, id string, baseCommitSHA string) error GetTaskSessionByTaskAndAgent(ctx context.Context, taskID, agentInstanceID string) (*models.TaskSession, error) diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index b3b6355252..a7885ba0f3 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -276,13 +276,19 @@ type mockRepository struct { // Optional hook to inject behavior into GetTaskSession (e.g. simulate a // transient DB error); if nil, the default map lookup is used. - getTaskSessionFunc func(ctx context.Context, id string) (*models.TaskSession, error) - getTaskFunc func(ctx context.Context, id string) (*models.Task, error) - getTaskEnvironmentFunc func(ctx context.Context, id string) (*models.TaskEnvironment, error) - getTaskEnvironmentByTaskIDFunc func(ctx context.Context, taskID string) (*models.TaskEnvironment, error) - createTaskEnvironmentRepoErr error - finalizeTaskEnvironmentErr error - createTaskSessionFunc func(ctx context.Context, session *models.TaskSession) error + getTaskSessionFunc func(ctx context.Context, id string) (*models.TaskSession, error) + getTaskFunc func(ctx context.Context, id string) (*models.Task, error) + getTaskEnvironmentFunc func(ctx context.Context, id string) (*models.TaskEnvironment, error) + getTaskEnvironmentByTaskIDFunc func(ctx context.Context, taskID string) (*models.TaskEnvironment, error) + createTaskEnvironmentRepoErr error + finalizeTaskEnvironmentErr error + createTaskSessionFunc func(ctx context.Context, session *models.TaskSession) error + // getTaskSessionByTaskAndAgentFunc, when non-nil, overrides + // GetTaskSessionByTaskAndAgent entirely — used to simulate a transient + // lookup failure (e.g. the AC-003.7 re-read-after-conflict arm in + // createOfficeSessionWithBoundedRecovery), which the default map lookup + // can never produce on its own. + getTaskSessionByTaskAndAgentFunc func(ctx context.Context, taskID, agentInstanceID string) (*models.TaskSession, error) updateTaskSessionStateFunc func(ctx context.Context, sessionID string, state models.TaskSessionState, errorMessage string) error listActiveTaskSessionsByTaskIDFunc func(ctx context.Context, taskID string) ([]*models.TaskSession, error) // Optional hook invoked at the top of UpdateTaskStateIfCurrentIn, before @@ -300,20 +306,23 @@ type mockRepository struct { updateTaskStateIfNotArchivedCh chan struct{} // Track calls for verification - createTaskSessionCalls []*models.TaskSession - updateTaskSessionCalls []*models.TaskSession - updateTaskSessionSnapshots []*models.TaskSession - updateTaskSessionIfCurrentCalls int - updateTaskSessionIfCurrentFailOn int - updateTaskSessionIfCurrentFailErr error - setSessionMetadataKeyCalls []setSessionMetadataKeyCall - setSessionPrimaryCalls []string - createTaskEnvironmentCalls []*models.TaskEnvironment - sharedWorkspaceBindingCalls []sharedWorkspaceBindingCall - updateTaskEnvironmentCalls []*models.TaskEnvironment - finalizeTaskEnvironmentCalls []*models.TaskEnvironment - updateTaskStateIfCurrentInCalls []updateTaskStateIfCurrentInCall - updateTaskStateIfNotArchivedCalls []updateTaskStateIfNotArchivedCall + createTaskSessionCalls []*models.TaskSession + updateTaskSessionCalls []*models.TaskSession + updateTaskSessionSnapshots []*models.TaskSession + updateTaskSessionIfCurrentCalls int + updateTaskSessionIfCurrentFailOn int + updateTaskSessionIfCurrentFailErr error + updateTaskSessionStateIfCurrentCalls int + updateTaskSessionStateIfCurrentFailOn int + updateTaskSessionStateIfCurrentFailErr error + setSessionMetadataKeyCalls []setSessionMetadataKeyCall + setSessionPrimaryCalls []string + createTaskEnvironmentCalls []*models.TaskEnvironment + sharedWorkspaceBindingCalls []sharedWorkspaceBindingCall + updateTaskEnvironmentCalls []*models.TaskEnvironment + finalizeTaskEnvironmentCalls []*models.TaskEnvironment + updateTaskStateIfCurrentInCalls []updateTaskStateIfCurrentInCall + updateTaskStateIfNotArchivedCalls []updateTaskStateIfNotArchivedCall // writeCallLog records the relative order of environment-row and // environment-status writes (e.g. "create_repo", "update_env") so tests @@ -461,6 +470,43 @@ func (m *mockRepository) UpdateTaskSessionIfCurrentState( return true, nil } +// UpdateTaskSessionStateIfCurrent mirrors the production narrow-CAS +// semantics: only state/error_message/completed_at/updated_at change, so a +// concurrent update to any other field on the stored session survives. +func (m *mockRepository) UpdateTaskSessionStateIfCurrent( + _ context.Context, + id string, + expected, status models.TaskSessionState, + errorMessage string, +) (bool, time.Time, error) { + m.mu.Lock() + defer m.mu.Unlock() + current, ok := m.sessions[id] + if !ok { + return false, time.Time{}, models.ErrTaskSessionNotFound + } + if current.State != expected { + return false, time.Time{}, nil + } + m.updateTaskSessionStateIfCurrentCalls++ + if m.updateTaskSessionStateIfCurrentFailOn > 0 && + m.updateTaskSessionStateIfCurrentCalls == m.updateTaskSessionStateIfCurrentFailOn { + return false, time.Time{}, m.updateTaskSessionStateIfCurrentFailErr + } + now := time.Now().UTC() + current.State = status + current.ErrorMessage = errorMessage + current.UpdatedAt = now + if status == models.TaskSessionStateCompleted || + status == models.TaskSessionStateFailed || + status == models.TaskSessionStateCancelled { + current.CompletedAt = &now + } else { + current.CompletedAt = nil + } + return true, now, nil +} + func (m *mockRepository) UpdateTaskSessionIfCurrentStateRemovingMetadataKeys( ctx context.Context, session *models.TaskSession, @@ -874,6 +920,12 @@ func (m *mockRepository) GetActiveTaskSessionByTaskID(ctx context.Context, taskI return nil, nil } func (m *mockRepository) GetTaskSessionByTaskAndAgent(ctx context.Context, taskID, agentInstanceID string) (*models.TaskSession, error) { + m.mu.Lock() + fn := m.getTaskSessionByTaskAndAgentFunc + m.mu.Unlock() + if fn != nil { + return fn(ctx, taskID, agentInstanceID) + } m.mu.Lock() defer m.mu.Unlock() for _, s := range m.sessions { @@ -1287,7 +1339,7 @@ func (m *mockCapabilities) ShouldApplyPreferredShell(executorType string) bool { } // Helper to create a test executor -func newTestExecutor(t *testing.T, agentManager AgentManagerClient, repo *mockRepository) *Executor { +func newTestExecutor(t *testing.T, agentManager AgentManagerClient, repo executorStore) *Executor { t.Helper() log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) if err != nil { diff --git a/apps/backend/internal/orchestrator/executor/executor_office.go b/apps/backend/internal/orchestrator/executor/executor_office.go index 2aa915ff02..8bd0b3fc86 100644 --- a/apps/backend/internal/orchestrator/executor/executor_office.go +++ b/apps/backend/internal/orchestrator/executor/executor_office.go @@ -80,21 +80,55 @@ func (e *Executor) EnsureSessionForAgentWithCreation( } } - created, err := e.createOfficeSession(ctx, task, agentInstanceID, agentProfileID, executorID, executorProfileID) - if err != nil && errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { - // Lost the race against a concurrent caller. Re-read and reuse. - raced, lookupErr := e.repo.GetTaskSessionByTaskAndAgent(ctx, task.ID, agentInstanceID) - if lookupErr == nil && raced != nil { - if rebindErr := e.rebindOfficeSessionExecutionProfile(ctx, raced, agentProfileID); rebindErr != nil { - return nil, false, rebindErr - } - reused, _ := e.tryReuseExistingSession(ctx, raced) - if reused != nil { - return reused, false, nil - } + return e.createOfficeSessionWithBoundedRecovery(ctx, task, agentInstanceID, agentProfileID, executorID, executorProfileID) +} + +// maxOfficeSessionCreateAttempts bounds EnsureSessionForAgentWithCreation's +// create-then-recover loop (AC-003.3): one initial attempt plus one retry +// after losing a create race and finding no live winner to reuse yet. +// Unbounded retry would spin forever if the guard and the recovery lookup +// disagree in a way that never resolves. +const maxOfficeSessionCreateAttempts = 2 + +// createOfficeSessionWithBoundedRecovery attempts to create a fresh office +// session, and on losing the race to a concurrent creator ( +// taskrepo.ErrOfficeSessionRaceConflict) re-reads the pair and reuses the +// winner instead of surfacing the conflict. A non-conflict create failure, +// or a failure while re-reading after a conflict, is returned as itself — +// never laundered into the sentinel and never as a nil session with a nil +// error. +func (e *Executor) createOfficeSessionWithBoundedRecovery( + ctx context.Context, task *v1.Task, agentInstanceID, agentProfileID, executorID, executorProfileID string, +) (*models.TaskSession, bool, error) { + var lastErr error + for attempt := 0; attempt < maxOfficeSessionCreateAttempts; attempt++ { + created, err := e.createOfficeSession(ctx, task, agentInstanceID, agentProfileID, executorID, executorProfileID) + if err == nil { + return created, true, nil } + lastErr = err + if !errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + return nil, false, err + } + + raced, lookupErr := e.repo.GetTaskSessionByTaskAndAgent(ctx, task.ID, agentInstanceID) + if lookupErr != nil { + return nil, false, fmt.Errorf("lookup (task,agent) session after create race: %w", lookupErr) + } + if raced == nil { + continue + } + if rebindErr := e.rebindOfficeSessionExecutionProfile(ctx, raced, agentProfileID); rebindErr != nil { + return nil, false, rebindErr + } + reused, decision := e.tryReuseExistingSession(ctx, raced) + if decision != reuseDecisionTerminal { + return reused, false, nil + } + // The winner turned out terminal by the time we read it; retry the + // create rather than reusing a row that can't be reused. } - return created, err == nil, err + return nil, false, lastErr } func (e *Executor) rebindOfficeSessionExecutionProfile( @@ -182,13 +216,7 @@ func (e *Executor) tryReuseExistingSession( ) (*models.TaskSession, reuseDecision) { switch session.State { case models.TaskSessionStateIdle: - if err := e.repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateRunning, ""); err != nil { - e.logger.Warn("failed to flip office session IDLE→RUNNING; returning row anyway", - zap.String("session_id", session.ID), zap.Error(err)) - } else { - session.State = models.TaskSessionStateRunning - } - return session, reuseDecisionReused + return e.tryFlipIdleSessionToRunning(ctx, session) case models.TaskSessionStateCreated, models.TaskSessionStateStarting, models.TaskSessionStateRunning, models.TaskSessionStateWaitingForInput: return session, reuseDecisionReused @@ -199,12 +227,56 @@ func (e *Executor) tryReuseExistingSession( } } +// tryFlipIdleSessionToRunning flips an IDLE row to RUNNING guarded by a +// CAS on the observed state, rather than an unconditional write. A blind +// write here could resurrect a row to RUNNING after a concurrent terminal +// transition (COMPLETED/FAILED/CANCELLED) landed between the read that +// classified this session as IDLE and this write — the same two-pool gap +// documented for creates, reachable via both the initial lookup and the +// bounded-recovery re-read. A CAS mismatch means exactly that race +// happened, so the fresh state — not the write's absence of an error — is +// what decides the outcome. +// +// The CAS itself uses the narrow, state-only UpdateTaskSessionStateIfCurrent +// rather than a full-row write: this session's in-memory copy was read +// before the flip, so a concurrent update to any other field (profile +// snapshot, routing, execution identifiers) between that read and this +// write must not be reverted. The row is always reloaded after the CAS, +// win or lose, so the caller observes the current stored values rather than +// a stale local copy. +func (e *Executor) tryFlipIdleSessionToRunning( + ctx context.Context, session *models.TaskSession, +) (*models.TaskSession, reuseDecision) { + _, _, err := e.repo.UpdateTaskSessionStateIfCurrent( + ctx, session.ID, models.TaskSessionStateIdle, models.TaskSessionStateRunning, "", + ) + if err != nil { + e.logger.Warn("failed to flip office session IDLE→RUNNING; treating outcome as unknown", + zap.String("session_id", session.ID), zap.Error(err)) + return nil, reuseDecisionTerminal + } + + fresh, err := e.repo.GetTaskSession(ctx, session.ID) + if err != nil || fresh == nil { + e.logger.Warn("failed to reload office session after IDLE→RUNNING attempt; treating outcome as unknown", + zap.String("session_id", session.ID), zap.Error(err)) + return nil, reuseDecisionTerminal + } + if isStopTerminalSessionState(fresh.State) { + return nil, reuseDecisionTerminal + } + *session = *fresh + return session, reuseDecisionReused +} + // createOfficeSession inserts a fresh task_sessions row for the given // (task, agent) pair with state CREATED. Mirrors PrepareSession's repo lookups // (primary repo, executor config, agent profile snapshot) but stores // agent_profile_id so the row can participate in office-session uniqueness -// enforcement once it exists (no index currently constrains this pair — see -// taskrepo.ErrOfficeSessionRaceConflict's doc comment). +// enforcement, which today is an in-transaction guard inside +// CreateOfficeTaskSession (not a database constraint) that returns +// taskrepo.ErrOfficeSessionRaceConflict on conflict — see that sentinel's +// doc comment. // // is_primary is left false: office sessions don't use the primary mechanism; // it stays for kanban / quick-chat advanced-mode resume. @@ -279,11 +351,28 @@ func (e *Executor) persistOfficeSession(ctx context.Context, taskID string, sess return e.persistOfficeSessionFallback(ctx, taskID, session) } +// persistOfficeSessionFallback is the in-process equivalent of +// CreateOfficeTaskSession for repositories that don't implement +// officeTaskSessionCreator. Callers hold e.officeSessionLock(taskID) for the +// duration, which is this fallback's only mutual-exclusion primitive — so +// the live-pair guard below must run inside that same critical section, +// restricted to the (task, agent) pair before classifying terminality, +// exactly like the repository-native guard. func (e *Executor) persistOfficeSessionFallback(ctx context.Context, taskID string, session *models.TaskSession) error { existingSessions, err := e.repo.ListTaskSessions(ctx, taskID) if err != nil { return fmt.Errorf("list task sessions before creating office session: %w", err) } + if session.AgentProfileID != "" { + for _, existing := range existingSessions { + if existing.AgentProfileID != session.AgentProfileID { + continue + } + if !isStopTerminalSessionState(existing.State) { + return taskrepo.ErrOfficeSessionRaceConflict + } + } + } if len(existingSessions) == 0 { if session.Metadata == nil { session.Metadata = make(map[string]interface{}) diff --git a/apps/backend/internal/orchestrator/executor/executor_office_test.go b/apps/backend/internal/orchestrator/executor/executor_office_test.go index 0ba8b38c40..b14a3cae88 100644 --- a/apps/backend/internal/orchestrator/executor/executor_office_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_office_test.go @@ -287,6 +287,77 @@ func TestEnsureSessionForAgent_RebindsExecutionProfileAfterCreateRace(t *testing } } +// TestEnsureSessionForAgent_RefusalThenRecoveryFlipsIdleWinnerAndClearsMetadata +// covers the full refusal-then-recovery shape end to end (AC-001.9): this +// caller's own create attempt loses the race and is refused with +// ErrOfficeSessionRaceConflict, the bounded-recovery re-read then observes +// the winning row — which by the time it's read has already gone IDLE, not +// RUNNING — on a different execution profile with stale provider metadata +// still attached. Unlike TestEnsureSessionForAgent_RebindsExecutionProfileAfterCreateRace +// (which only asserts the rebound ExecutionProfileID) and +// TestEnsureSessionForAgent_RebindsExecutionProfileOnReuse (which asserts the +// metadata clear but reaches rebind via a direct initial lookup, never +// through a forced refusal), this test proves rebindOfficeSessionExecutionProfile's +// metadata clear AND tryFlipIdleSessionToRunning's CAS flip to RUNNING both +// happen together on the specific row recovered after a refusal. +func TestEnsureSessionForAgent_RefusalThenRecoveryFlipsIdleWinnerAndClearsMetadata(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + winner := &models.TaskSession{ + ID: "sess-refusal-recovery-winner", + TaskID: "task-office", + AgentProfileID: "agent-1", + ExecutionProfileID: "codex-profile", + State: models.TaskSessionStateIdle, + StartedAt: time.Now().UTC(), + Metadata: map[string]interface{}{ + "acp_session_id": "codex-session", + models.SessionMetaKeySessionMode: "default", + }, + } + repo.createTaskSessionFunc = func(_ context.Context, _ *models.TaskSession) error { + // Simulate a concurrent creator: this caller's own create attempt is + // refused, but the row it lost the race to is written here, as the + // AC-003.7 tests in office_session_race_guard_test.go already do. + repo.mu.Lock() + repo.sessions[winner.ID] = winner + repo.mu.Unlock() + return fmt.Errorf("%w: concurrent insert", taskrepo.ErrOfficeSessionRaceConflict) + } + + got, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), officeTestTask(), "agent-1", "claude-profile", "exec-1", "", + ) + if err != nil { + t.Fatalf("EnsureSessionForAgentWithCreation: %v", err) + } + if wasCreated { + t.Fatal("wasCreated = true, want false (this caller reused the recovered winner, it did not create)") + } + if got == nil || got.ID != winner.ID { + t.Fatalf("session = %#v, want the recovered winner row %q", got, winner.ID) + } + if got.State != models.TaskSessionStateRunning { + t.Fatalf("state = %q, want RUNNING (the recovery-reused row must still flip IDLE->RUNNING)", got.State) + } + if got.ExecutionProfileID != "claude-profile" { + t.Fatalf("execution profile = %q, want claude-profile", got.ExecutionProfileID) + } + if _, exists := got.Metadata["acp_session_id"]; exists { + t.Error("acp_session_id metadata survived the execution profile rebind on the recovered row") + } + if _, exists := got.Metadata[models.SessionMetaKeySessionMode]; exists { + t.Error("session mode metadata survived the execution profile rebind on the recovered row") + } + stored := repo.sessions[winner.ID] + if stored.State != models.TaskSessionStateRunning { + t.Fatalf("stored session state = %q, want RUNNING", stored.State) + } + if _, exists := stored.Metadata["acp_session_id"]; exists { + t.Error("stored session metadata still carries acp_session_id after the rebind+flip") + } +} + // TestEnsureSessionForAgent_ReusesIdleAndFlipsRunning covers the canonical // reuse path: the second run for the same (task, agent) reuses the existing // row and flips IDLE → RUNNING. No new row is inserted. diff --git a/apps/backend/internal/orchestrator/executor/office_session_race_guard_test.go b/apps/backend/internal/orchestrator/executor/office_session_race_guard_test.go new file mode 100644 index 0000000000..eb69933108 --- /dev/null +++ b/apps/backend/internal/orchestrator/executor/office_session_race_guard_test.go @@ -0,0 +1,921 @@ +package executor + +// Tests for the office-session live-pair guard in the in-process fallback +// creator (Change 1b) and bounded recovery in EnsureSessionForAgentWithCreation +// (Change 3). See +// docs/specs/office/{requirements,system-design}/task-session-identity*.md. +// mockRepository does not implement officeTaskSessionCreator, so every +// EnsureSessionForAgent* test in this package that uses it directly exercises +// persistOfficeSessionFallback rather than a repository-native creator. +// officeTaskSessionCreatingRepository (below) is the one exception: it wraps +// mockRepository to add officeTaskSessionCreator, for tests that must prove +// persistOfficeSession's other branch — the one taken in production, where +// the repository's own transaction is the only serialization and the +// executor's per-task fallback lock is never acquired at all. + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// TestPersistOfficeSessionFallbackRefusesSecondLivePair drives the fallback +// creator directly (bypassing the lookup-then-create dance in +// EnsureSessionForAgentWithCreation) to prove it — not just the SQLite +// repository — enforces the live-pair guard: the first call for a pair +// succeeds, a second call for the SAME pair while the first is still live is +// refused with ErrOfficeSessionRaceConflict, and refusal writes nothing. +func TestPersistOfficeSessionFallbackRefusesSecondLivePair(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + const taskID = "task-office-fallback-live-pair" + agent := "agent-fallback-live-pair" + + first := &models.TaskSession{ID: "fallback-live-pair-1", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated} + if err := exec.persistOfficeSessionFallback(ctx, taskID, first); err != nil { + t.Fatalf("first persistOfficeSessionFallback: %v", err) + } + + second := &models.TaskSession{ID: "fallback-live-pair-2", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated} + err := exec.persistOfficeSessionFallback(ctx, taskID, second) + if !errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + t.Fatalf("second persistOfficeSessionFallback error = %v, want ErrOfficeSessionRaceConflict", err) + } + + sessions, err := repo.ListTaskSessions(ctx, taskID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 1 { + t.Fatalf("sessions for task = %d, want 1 (refused call must not have written a row)", len(sessions)) + } +} + +// TestPersistOfficeSessionFallbackAllowsCreateWhenExistingPairRowIsTerminal +// mirrors the repository-layer guard's terminal-retry allowance: a pair +// whose only existing row is terminal must not block a fresh create. +func TestPersistOfficeSessionFallbackAllowsCreateWhenExistingPairRowIsTerminal(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + const taskID = "task-office-fallback-terminal-retry" + agent := "agent-fallback-terminal-retry" + + repo.sessions["fallback-terminal-retry-old"] = &models.TaskSession{ + ID: "fallback-terminal-retry-old", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCompleted, StartedAt: time.Now().UTC(), + } + + fresh := &models.TaskSession{ID: "fallback-terminal-retry-new", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated} + if err := exec.persistOfficeSessionFallback(ctx, taskID, fresh); err != nil { + t.Fatalf("persistOfficeSessionFallback: %v", err) + } + + sessions, err := repo.ListTaskSessions(ctx, taskID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 2 { + t.Fatalf("sessions for task = %d, want 2", len(sessions)) + } +} + +// TestPersistOfficeSessionFallbackSkipsGuardForEmptyAgentProfileID proves the +// fallback guard is bypassed for kanban-shaped rows with no agent profile, +// exactly like the repository-layer guard. +func TestPersistOfficeSessionFallbackSkipsGuardForEmptyAgentProfileID(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + const taskID = "task-office-fallback-empty-agent" + + first := &models.TaskSession{ID: "fallback-empty-agent-1", TaskID: taskID, State: models.TaskSessionStateCreated} + if err := exec.persistOfficeSessionFallback(ctx, taskID, first); err != nil { + t.Fatalf("first persistOfficeSessionFallback: %v", err) + } + second := &models.TaskSession{ID: "fallback-empty-agent-2", TaskID: taskID, State: models.TaskSessionStateCreated} + if err := exec.persistOfficeSessionFallback(ctx, taskID, second); err != nil { + t.Fatalf("second persistOfficeSessionFallback: %v", err) + } + + sessions, err := repo.ListTaskSessions(ctx, taskID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 2 { + t.Fatalf("sessions for task = %d, want 2 (guard must not apply to empty agent_profile_id)", len(sessions)) + } +} + +// TestPersistOfficeSessionFallbackIgnoresLiveSessionForDifferentAgentOnSameTask +// proves the fallback guard restricts to the (task, agent_profile_id) PAIR +// before it ever asks whether an existing row is terminal: a different +// agent's live (non-terminal) session on the same task must not block +// creating a session for a new agent. Without the pair restriction running +// first, a task-wide "any non-terminal session exists" check would also +// reject this call — and every other fallback test in this file seeds either +// no other agent's session or only a terminal one, so none of them would +// catch that regression. +func TestPersistOfficeSessionFallbackIgnoresLiveSessionForDifferentAgentOnSameTask(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + const taskID = "task-office-fallback-other-agent-live" + + repo.sessions["fallback-other-agent-live"] = &models.TaskSession{ + ID: "fallback-other-agent-live", TaskID: taskID, AgentProfileID: "agent-a", + State: models.TaskSessionStateRunning, StartedAt: time.Now().UTC(), + } + + fresh := &models.TaskSession{ + ID: "fallback-other-agent-new", TaskID: taskID, AgentProfileID: "agent-b", + State: models.TaskSessionStateCreated, + } + if err := exec.persistOfficeSessionFallback(ctx, taskID, fresh); err != nil { + t.Fatalf("persistOfficeSessionFallback: %v, want nil (a different agent's live session must not block creation)", err) + } + + sessions, err := repo.ListTaskSessions(ctx, taskID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 2 { + t.Fatalf("sessions for task = %d, want 2", len(sessions)) + } +} + +// TestPersistOfficeSessionFallbackRefusesLiveForNonCanonicalLiveStates closes +// Review entry 19's MUST-FIX 3 (Go fallback half): the fallback guard's +// `!isStopTerminalSessionState(existing.State)` check must be driven by "not +// terminal", not by an enumerated allow-list of whichever live states the +// other fallback tests in this file happen to use (CREATED/RUNNING). STARTING +// and WAITING_FOR_INPUT are live states none of them touch. If the check were +// rewritten as an allow-list of just those two states, every other fallback +// test here would stay green while a pair whose only row sits in STARTING or +// WAITING_FOR_INPUT would wrongly allow a duplicate create. +// +// IDLE and a fabricated out-of-enum state close Review round 5's Finding B at +// the executor layer: IDLE is an office session's dominant resting state +// between turns and no fallback test previously seeded it as the blocking +// row. +func TestPersistOfficeSessionFallbackRefusesLiveForNonCanonicalLiveStates(t *testing.T) { + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateStarting, + models.TaskSessionStateWaitingForInput, + models.TaskSessionStateIdle, + models.TaskSessionState("PAUSED_FUTURE_STATE"), + } { + t.Run(string(state), func(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + taskID := "task-office-fallback-noncanonical-live-" + string(state) + agent := "agent-fallback-noncanonical-live" + + repo.sessions["fallback-noncanonical-live-existing"] = &models.TaskSession{ + ID: "fallback-noncanonical-live-existing", TaskID: taskID, AgentProfileID: agent, + State: state, StartedAt: time.Now().UTC(), + } + + fresh := &models.TaskSession{ + ID: "fallback-noncanonical-live-blocked", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + } + err := exec.persistOfficeSessionFallback(ctx, taskID, fresh) + if !errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + t.Fatalf("persistOfficeSessionFallback error = %v, want ErrOfficeSessionRaceConflict (a pair whose only row is %s must be treated as live)", err, state) + } + + sessions, err := repo.ListTaskSessions(ctx, taskID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 1 { + t.Fatalf("sessions for task = %d, want 1 (the refused create must not have written a row)", len(sessions)) + } + }) + } +} + +// TestEnsureSessionForAgentWithCreation_ConcurrentCallersConvergeOnOneSession +// is the executor-level convergence test: N concurrent callers for the same +// (task, agent) pair, none of which observe an existing row at lookup time +// (simulated by starting them all from an empty repo and letting the +// per-task mutex in persistOfficeSession serialize the real race), must +// converge on exactly one row — the losers recover via re-read-and-reuse +// rather than surfacing the conflict to their caller (AC-003.1, AC-003.2). +// AC-003.2 is specifically a one-creator/N-1-reusers contract, not just a +// same-ID contract, so this also captures and asserts the "created by this +// call" boolean each caller receives: exactly one true among the n results. +func TestEnsureSessionForAgentWithCreation_ConcurrentCallersConvergeOnOneSession(t *testing.T) { + const n = 4 + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + results := make([]*models.TaskSession, n) + created := make([]bool, n) + errs := make([]error, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + session, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), task, "agent-convergence", "profile-1", "exec-1", "", + ) + results[i] = session + created[i] = wasCreated + errs[i] = err + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("caller %d: %v", i, err) + } + if results[i] == nil { + t.Fatalf("caller %d: session = nil, want non-nil", i) + } + } + firstID := results[0].ID + for i, session := range results { + if session.ID != firstID { + t.Fatalf("caller %d converged on %q, want %q (all callers must share one row)", i, session.ID, firstID) + } + } + + creatorCount := 0 + for i, wasCreated := range created { + if wasCreated { + creatorCount++ + } + t.Logf("caller %d created=%v", i, wasCreated) + } + if creatorCount != 1 { + t.Fatalf("callers reporting created=true = %d, want exactly 1 (one creator, %d reusers)", creatorCount, n-1) + } + + sessions, err := repo.ListTaskSessions(context.Background(), task.ID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 1 { + t.Fatalf("sessions for task = %d, want exactly 1", len(sessions)) + } + if len(repo.createTaskSessionCalls) != 1 { + t.Fatalf("CreateTaskSession calls = %d, want exactly 1 (no duplicate CREATED event)", len(repo.createTaskSessionCalls)) + } +} + +// TestEnsureSessionForAgentWithCreation_RecoveryStopsAfterTwoAttempts proves +// the bounded-retry contract (AC-003.3): if every create attempt loses the +// race AND the recovery re-read also fails to observe a winner (a +// pathological case that should not happen in practice, but the retry must +// still be bounded rather than looping), EnsureSessionForAgentWithCreation +// makes at most 2 create attempts total and returns the conflict rather than +// retrying forever. +func TestEnsureSessionForAgentWithCreation_RecoveryStopsAfterTwoAttempts(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + attempts := 0 + repo.createTaskSessionFunc = func(_ context.Context, _ *models.TaskSession) error { + attempts++ + return fmt.Errorf("%w: perpetual loser", taskrepo.ErrOfficeSessionRaceConflict) + } + + session, err := exec.EnsureSessionForAgent( + context.Background(), task, "agent-bounded-retry", "profile-1", "exec-1", "", + ) + if err == nil { + t.Fatal("expected error when every create attempt and every recovery lookup fails") + } + if session != nil { + t.Fatalf("session = %#v, want nil on unrecoverable conflict", session) + } + if !errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + t.Fatalf("error = %v, want errors.Is match against ErrOfficeSessionRaceConflict", err) + } + if attempts != 2 { + t.Fatalf("create attempts = %d, want exactly 2 (bounded retry)", attempts) + } +} + +// TestEnsureSessionForAgentWithCreation_NonConflictCreateFailureNotLaundered +// proves a genuine (non-conflict) create failure is returned as itself, not +// masked as a race conflict or silently swallowed by the recovery branch +// (AC-003.7). +func TestEnsureSessionForAgentWithCreation_NonConflictCreateFailureNotLaundered(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + wantErr := errors.New("disk full") + repo.createTaskSessionFunc = func(_ context.Context, _ *models.TaskSession) error { + return wantErr + } + + session, err := exec.EnsureSessionForAgent( + context.Background(), task, "agent-non-conflict-failure", "profile-1", "exec-1", "", + ) + if session != nil { + t.Fatalf("session = %#v, want nil", session) + } + if !errors.Is(err, wantErr) { + t.Fatalf("error = %v, want errors.Is match against the underlying failure", err) + } + if errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + t.Fatalf("non-conflict failure %v must not classify as ErrOfficeSessionRaceConflict", err) + } + if len(repo.createTaskSessionCalls) != 1 { + t.Fatalf("CreateTaskSession calls = %d, want exactly 1 (a non-conflict failure must not spend a bounded-retry attempt, AC-003.7)", len(repo.createTaskSessionCalls)) + } +} + +// TestTryReuseExistingSession_IdleCASMismatchDoesNotResurrectTerminalSession +// proves the IDLE→RUNNING flip in tryReuseExistingSession is guarded by a CAS +// on the observed state (SEC-001): a caller that read the row as IDLE before +// a concurrent terminal transition (COMPLETED/FAILED/CANCELLED) landed must +// not have its stale write resurrect the now-terminal row to RUNNING. Drives +// the CAS mismatch directly — the mock's underlying row is already COMPLETED +// while the argument passed to tryReuseExistingSession still reflects the +// stale IDLE read — rather than relying on goroutine timing. +func TestTryReuseExistingSession_IdleCASMismatchDoesNotResurrectTerminalSession(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-cas-mismatch" + current := &models.TaskSession{ID: "idle-cas-mismatch", TaskID: taskID, State: models.TaskSessionStateCompleted} + repo.sessions[current.ID] = current + + staleView := &models.TaskSession{ID: current.ID, TaskID: taskID, State: models.TaskSessionStateIdle} + result, decision := exec.tryReuseExistingSession(ctx, staleView) + + if decision != reuseDecisionTerminal { + t.Fatalf("decision = %v, want reuseDecisionTerminal", decision) + } + if result != nil { + t.Fatalf("result = %#v, want nil", result) + } + stored := repo.sessions[current.ID] + if stored.State != models.TaskSessionStateCompleted { + t.Fatalf("stored session state = %v, want unchanged COMPLETED (must not be resurrected to RUNNING)", stored.State) + } +} + +// TestTryReuseExistingSession_IdleCASSuccessThenTerminalReloadRequiresFreshSession +// covers the second half of the IDLE->RUNNING race: another actor can terminate +// the row after the CAS succeeds but before the follow-up reload returns. The +// reloaded terminal state must still force a fresh-session decision. +func TestTryReuseExistingSession_IdleCASSuccessThenTerminalReloadRequiresFreshSession(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-cas-success-terminal" + const sessionID = "idle-cas-success-terminal" + repo.sessions[sessionID] = &models.TaskSession{ + ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle, + } + repo.getTaskSessionFunc = func(_ context.Context, id string) (*models.TaskSession, error) { + if id != sessionID { + t.Fatalf("GetTaskSession id = %q, want %q", id, sessionID) + } + repo.mu.Lock() + defer repo.mu.Unlock() + repo.sessions[id].State = models.TaskSessionStateCompleted + return cloneMockTaskSession(repo.sessions[id]), nil + } + + view := &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle} + result, decision := exec.tryReuseExistingSession(ctx, view) + + if decision != reuseDecisionTerminal { + t.Fatalf("decision = %v, want reuseDecisionTerminal", decision) + } + if result != nil { + t.Fatalf("result = %#v, want nil when reload observes terminal state", result) + } + if stored := repo.sessions[sessionID].State; stored != models.TaskSessionStateCompleted { + t.Fatalf("stored session state = %v, want unchanged COMPLETED", stored) + } +} + +// TestTryReuseExistingSession_IdleFlipsToRunningWhenStateStillMatches proves +// the happy path still works with the CAS write: an IDLE row whose state has +// not changed since the read flips to RUNNING and is returned as reused. +func TestTryReuseExistingSession_IdleFlipsToRunningWhenStateStillMatches(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-cas-match" + current := &models.TaskSession{ID: "idle-cas-match", TaskID: taskID, State: models.TaskSessionStateIdle} + repo.sessions[current.ID] = current + + view := &models.TaskSession{ID: current.ID, TaskID: taskID, State: models.TaskSessionStateIdle} + result, decision := exec.tryReuseExistingSession(ctx, view) + + if decision != reuseDecisionReused { + t.Fatalf("decision = %v, want reuseDecisionReused", decision) + } + if result == nil || result.State != models.TaskSessionStateRunning { + t.Fatalf("result = %#v, want non-nil session with State RUNNING", result) + } + if repo.updateTaskSessionStateIfCurrentCalls != 1 { + t.Fatalf("UpdateTaskSessionStateIfCurrent calls = %d, want 1", repo.updateTaskSessionStateIfCurrentCalls) + } + stored := repo.sessions[current.ID] + if stored.State != models.TaskSessionStateRunning { + t.Fatalf("stored session state = %v, want RUNNING", stored.State) + } +} + +// TestTryReuseExistingSession_IdleFlipPreservesConcurrentFieldUpdate proves +// the IDLE→RUNNING flip uses a narrow, state-only CAS rather than writing +// the caller's full (possibly stale) in-memory copy: a concurrent update to +// an unrelated field (AgentProfileSnapshot here) that lands on the stored +// row after the caller's read but before the flip must survive the flip, +// not be silently reverted to the caller's pre-update snapshot. +func TestTryReuseExistingSession_IdleFlipPreservesConcurrentFieldUpdate(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-flip-preserves-field" + const sessionID = "idle-flip-preserves-field" + stale := map[string]interface{}{"model": "stale-model"} + current := &models.TaskSession{ + ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle, + AgentProfileSnapshot: stale, + } + repo.sessions[sessionID] = current + + // The caller's view was read before a concurrent writer updated the + // snapshot on the stored row. + view := &models.TaskSession{ + ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle, + AgentProfileSnapshot: stale, + } + fresh := map[string]interface{}{"model": "fresh-model"} + current.AgentProfileSnapshot = fresh + + result, decision := exec.tryReuseExistingSession(ctx, view) + + if decision != reuseDecisionReused { + t.Fatalf("decision = %v, want reuseDecisionReused", decision) + } + if result == nil || result.State != models.TaskSessionStateRunning { + t.Fatalf("result = %#v, want non-nil session with State RUNNING", result) + } + if result.AgentProfileSnapshot["model"] != "fresh-model" { + t.Fatalf("result.AgentProfileSnapshot = %#v, want the concurrently-written fresh-model snapshot to survive the flip", result.AgentProfileSnapshot) + } + stored := repo.sessions[sessionID] + if stored.AgentProfileSnapshot["model"] != "fresh-model" { + t.Fatalf("stored session AgentProfileSnapshot = %#v, want the concurrent update to survive the flip", stored.AgentProfileSnapshot) + } +} + +// TestTryReuseExistingSession_UnrecognizedStateIsTreatedAsReusable closes the +// reuse-side half of Review round 5's Finding B: AC-001.4 requires "a state +// added later" to be treated identically by the create-side guard (complement +// of terminal, proven above at the SQLite and fallback layers) and by this +// reuse rule. tryReuseExistingSession's default arm returns +// reuseDecisionReused for anything outside the three explicitly-handled +// groups (IDLE, the four other live states, and the three terminal states) — +// this proves a fabricated out-of-enum state hits that default arm rather +// than being silently treated as terminal. +func TestTryReuseExistingSession_UnrecognizedStateIsTreatedAsReusable(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-unrecognized-state-reusable" + unrecognized := models.TaskSessionState("PAUSED_FUTURE_STATE") + current := &models.TaskSession{ID: "unrecognized-state-reusable", TaskID: taskID, State: unrecognized} + repo.sessions[current.ID] = current + + result, decision := exec.tryReuseExistingSession(ctx, current) + + if decision != reuseDecisionReused { + t.Fatalf("decision = %v, want reuseDecisionReused for unrecognized state %q", decision, unrecognized) + } + if result == nil || result.State != unrecognized { + t.Fatalf("result = %#v, want non-nil session with unchanged state %q", result, unrecognized) + } +} + +// TestTryReuseExistingSession_IdleFlipCASUpdateErrorTreatsOutcomeAsUnknown +// closes Review entry 19's MUST-FIX 1 (CAS-update-error branch, +// executor_office.go:244-249): a genuine backend error from +// UpdateTaskSessionIfCurrentState — distinct from a CAS mismatch, which +// already goes through the reload-and-recheck-terminal path — must not be +// treated as "reused". The caller has no way to know whether the write +// actually applied, so the outcome must come back unknown +// (reuseDecisionTerminal, nil session) rather than handing back the +// pre-transition row as if it were safe to use. +func TestTryReuseExistingSession_IdleFlipCASUpdateErrorTreatsOutcomeAsUnknown(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-cas-update-error" + const sessionID = "idle-flip-cas-update-error" + repo.sessions[sessionID] = &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle} + + wantErr := errors.New("disk full") + repo.updateTaskSessionStateIfCurrentFailOn = 1 + repo.updateTaskSessionStateIfCurrentFailErr = wantErr + + view := &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle} + result, decision := exec.tryReuseExistingSession(ctx, view) + + if decision != reuseDecisionTerminal { + t.Fatalf("decision = %v, want reuseDecisionTerminal", decision) + } + if result != nil { + t.Fatalf("result = %#v, want nil (unknown outcome must not be handed back as reusable)", result) + } + if repo.updateTaskSessionStateIfCurrentCalls != 1 { + t.Fatalf("UpdateTaskSessionStateIfCurrent calls = %d, want 1", repo.updateTaskSessionStateIfCurrentCalls) + } + stored := repo.sessions[sessionID] + if stored.State != models.TaskSessionStateIdle { + t.Fatalf("stored session state = %v, want unchanged IDLE (a failed write must not silently apply)", stored.State) + } +} + +// TestTryReuseExistingSession_IdleFlipReloadAfterMismatchErrorTreatsOutcomeAsUnknown +// closes Review entry 19's MUST-FIX 1 (reload-error branch, +// executor_office.go:255-260): the CAS write correctly detects a mismatch +// (the row already moved off IDLE — here, to a genuine terminal state, +// exactly the shape that produces this mismatch in production) and the +// follow-up reload to see what it became then fails for an unrelated +// reason. That reload failure must not be treated as "reused" either — the +// caller never learned the row went terminal, so the outcome must come back +// unknown rather than handing back the stale pre-transition view. +func TestTryReuseExistingSession_IdleFlipReloadAfterMismatchErrorTreatsOutcomeAsUnknown(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + ctx := context.Background() + + const taskID = "task-idle-reload-error" + const sessionID = "idle-flip-reload-error" + repo.sessions[sessionID] = &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateCompleted} + + wantErr := errors.New("db unavailable") + reloadCalls := 0 + repo.getTaskSessionFunc = func(_ context.Context, id string) (*models.TaskSession, error) { + reloadCalls++ + if id != sessionID { + t.Fatalf("GetTaskSession id = %q, want %q", id, sessionID) + } + return nil, wantErr + } + + // The caller's view is stale: it read the row as IDLE just before the + // concurrent terminal transition (above) landed. + view := &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateIdle} + result, decision := exec.tryReuseExistingSession(ctx, view) + + if decision != reuseDecisionTerminal { + t.Fatalf("decision = %v, want reuseDecisionTerminal", decision) + } + if result != nil { + t.Fatalf("result = %#v, want nil (unknown outcome must not be handed back as reusable)", result) + } + if reloadCalls != 1 { + t.Fatalf("GetTaskSession calls = %d, want 1 (the mismatch must trigger exactly one reload attempt)", reloadCalls) + } +} + +// officeTaskSessionCreatingRepository wraps mockRepository to additionally +// implement officeTaskSessionCreator, mirroring the live-pair-then-create +// transaction in the real Repository.CreateOfficeTaskSession (session.go): +// count live (non-terminal) sessions for the (task, agent_profile) pair, +// refuse with ErrOfficeSessionRaceConflict if any exist, otherwise mark +// task-initial origin on the first session for the task and persist. The +// whole check-then-write runs under creatingMu, standing in for the real +// implementation's DB transaction — persistOfficeSession takes this branch +// precisely when it will NOT also acquire the executor's own per-task lock, +// so this double's serialization must come from the same place production's +// does: the "transaction," not caller-side locking. +type officeTaskSessionCreatingRepository struct { + *mockRepository + creatingMu sync.Mutex + createOfficeTaskSessionN int +} + +func newOfficeTaskSessionCreatingRepository() *officeTaskSessionCreatingRepository { + return &officeTaskSessionCreatingRepository{mockRepository: newMockRepository()} +} + +func (r *officeTaskSessionCreatingRepository) CreateOfficeTaskSession( + ctx context.Context, session *models.TaskSession, +) error { + r.creatingMu.Lock() + defer r.creatingMu.Unlock() + r.createOfficeTaskSessionN++ + + r.mu.Lock() + var sessionCount, liveCount int + for _, existing := range r.sessions { + if existing.TaskID != session.TaskID { + continue + } + sessionCount++ + if session.AgentProfileID != "" && existing.AgentProfileID == session.AgentProfileID && + !isStopTerminalSessionState(existing.State) { + liveCount++ + } + } + r.mu.Unlock() + + if liveCount > 0 { + return taskrepo.ErrOfficeSessionRaceConflict + } + if sessionCount == 0 { + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + session.Metadata[models.SessionMetaKeyOrigin] = models.SessionOriginTaskInitial + } + return r.CreateTaskSession(ctx, session) +} + +// TestEnsureSessionForAgentWithCreation_ConvergesThroughRepositoryNativeCreator +// is TestEnsureSessionForAgentWithCreation_ConcurrentCallersConvergeOnOneSession's +// counterpart for the officeTaskSessionCreator branch of persistOfficeSession: +// N concurrent callers for the same (task, agent) pair, racing against a +// repository double that implements officeTaskSessionCreator, must still +// converge on exactly one row via that branch specifically — not the +// in-process fallback lock, which is never acquired on this path. Proves the +// interface-detection routing in persistOfficeSession is exercised end to +// end, not just the mock-fallback convergence covered elsewhere in this file. +func TestEnsureSessionForAgentWithCreation_ConvergesThroughRepositoryNativeCreator(t *testing.T) { + const n = 4 + repo := newOfficeTaskSessionCreatingRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + results := make([]*models.TaskSession, n) + created := make([]bool, n) + errs := make([]error, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + session, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), task, "agent-native-creator-convergence", "profile-1", "exec-1", "", + ) + results[i] = session + created[i] = wasCreated + errs[i] = err + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("caller %d: %v", i, err) + } + if results[i] == nil { + t.Fatalf("caller %d: session = nil, want non-nil", i) + } + } + firstID := results[0].ID + for i, session := range results { + if session.ID != firstID { + t.Fatalf("caller %d converged on %q, want %q (all callers must share one row)", i, session.ID, firstID) + } + } + + creatorCount := 0 + for _, wasCreated := range created { + if wasCreated { + creatorCount++ + } + } + if creatorCount != 1 { + t.Fatalf("callers reporting created=true = %d, want exactly 1 (one creator, %d reusers)", creatorCount, n-1) + } + + // A caller whose initial GetTaskSessionByTaskAndAgent lookup lands after + // the winner's row is already committed reuses it directly and never + // reaches CreateOfficeTaskSession at all, so the call count is not + // pinned to n — only bounded by it. What must hold regardless of + // interleaving: at least one call happened (the interface branch was + // actually exercised) and no caller needed the bounded-recovery loop's + // second create attempt (which would push the count past n). + if repo.createOfficeTaskSessionN < 1 || repo.createOfficeTaskSessionN > n { + t.Fatalf("CreateOfficeTaskSession calls = %d, want between 1 and %d (repository-native branch exercised, no runaway recovery retries)", + repo.createOfficeTaskSessionN, n) + } + if len(repo.createTaskSessionCalls) != 1 { + t.Fatalf("underlying CreateTaskSession calls = %d, want exactly 1 (no duplicate CREATED event)", len(repo.createTaskSessionCalls)) + } + sessions, err := repo.ListTaskSessions(context.Background(), task.ID) + if err != nil { + t.Fatalf("ListTaskSessions: %v", err) + } + if len(sessions) != 1 { + t.Fatalf("sessions for task = %d, want exactly 1", len(sessions)) + } +} + +// TestEnsureSessionForAgentWithCreation_RecoveryLookupFailurePropagatesAsItself +// exercises the AC-003.7 re-read-error arm in +// createOfficeSessionWithBoundedRecovery: after a create attempt loses the +// race (ErrOfficeSessionRaceConflict), the recovery re-read via +// GetTaskSessionByTaskAndAgent can itself fail for an unrelated reason (a +// transient DB error, say). That failure must surface as itself — wrapped, +// not laundered into ErrOfficeSessionRaceConflict and not swallowed into a +// nil-session/nil-error result — exactly like the sibling non-conflict +// create-failure case already covered above. +func TestEnsureSessionForAgentWithCreation_RecoveryLookupFailurePropagatesAsItself(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + repo.createTaskSessionFunc = func(_ context.Context, _ *models.TaskSession) error { + return taskrepo.ErrOfficeSessionRaceConflict + } + wantErr := errors.New("db unavailable") + var lookups int + repo.getTaskSessionByTaskAndAgentFunc = func(_ context.Context, _, _ string) (*models.TaskSession, error) { + lookups++ + if lookups == 1 { + // EnsureSessionForAgentWithCreation's initial lookup: no existing + // row, so it proceeds to create (which will lose the race above). + return nil, nil + } + // The bounded-recovery re-read after losing the create race: this is + // the AC-003.7 arm under test. + return nil, wantErr + } + + session, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), task, "agent-recovery-lookup-failure", "profile-1", "exec-1", "", + ) + if session != nil { + t.Fatalf("session = %#v, want nil", session) + } + if wasCreated { + t.Fatal("wasCreated = true, want false") + } + if !errors.Is(err, wantErr) { + t.Fatalf("error = %v, want errors.Is match against the underlying lookup failure", err) + } + if errors.Is(err, taskrepo.ErrOfficeSessionRaceConflict) { + t.Fatalf("recovery lookup failure %v must not classify as ErrOfficeSessionRaceConflict", err) + } +} + +// TestEnsureSessionForAgentWithCreation_RecoveryReReadFindsTerminalWinnerRetriesCreate +// covers createOfficeSessionWithBoundedRecovery's terminal-retry arm +// (executor_office.go), which had zero coverage: when the post-conflict +// recovery re-read finds a row that has itself gone terminal by the time +// tryReuseExistingSession inspects it, the loop must retry the create rather +// than returning the terminal row. This is distinct from the already-covered +// raced == nil shape (RecoveryStopsAfterTwoAttempts) and the lookup-error +// shape (RecoveryLookupFailurePropagatesAsItself) above - here the re-read +// itself succeeds and returns a real, terminal row. +// +// Left as-is, a regression collapsing `if decision != reuseDecisionTerminal { +// return reused, false, nil }` to an unconditional return would leave this +// package green while reproducing a reachable production nil-pointer panic: +// task_operations.go:1601 dereferences the returned session without a nil +// check on the (nil, false, nil) shape this arm would otherwise produce. +func TestEnsureSessionForAgentWithCreation_RecoveryReReadFindsTerminalWinnerRetriesCreate(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + var createAttempts int + repo.createTaskSessionFunc = func(_ context.Context, session *models.TaskSession) error { + createAttempts++ + if createAttempts == 1 { + return taskrepo.ErrOfficeSessionRaceConflict + } + repo.mu.Lock() + repo.sessions[session.ID] = session + repo.mu.Unlock() + return nil + } + + terminalWinner := &models.TaskSession{ + ID: "recovery-terminal-winner", TaskID: task.ID, AgentProfileID: "agent-recovery-terminal", + State: models.TaskSessionStateCompleted, StartedAt: time.Now().UTC(), + } + var lookups int + repo.getTaskSessionByTaskAndAgentFunc = func(_ context.Context, _, _ string) (*models.TaskSession, error) { + lookups++ + if lookups == 1 { + // Initial lookup: no existing row, so it proceeds to create + // (which will lose the race below). + return nil, nil + } + // The bounded-recovery re-read after losing the create race: the + // winner has already gone terminal by the time we observe it. + return terminalWinner, nil + } + + session, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), task, "agent-recovery-terminal", "profile-1", "exec-1", "", + ) + if err != nil { + t.Fatalf("err = %v, want nil", err) + } + if !wasCreated { + t.Fatal("wasCreated = false, want true (the loop must retry the create rather than reuse the terminal row)") + } + if session == nil || session.ID == terminalWinner.ID { + t.Fatalf("session = %#v, want a freshly created row distinct from the terminal winner %q", session, terminalWinner.ID) + } + if createAttempts != 2 { + t.Fatalf("create attempts = %d, want exactly 2 (first loses the race, second retries after the terminal re-read)", createAttempts) + } + if lookups != 2 { + t.Fatalf("recovery lookups = %d, want exactly 2 (initial lookup + post-conflict re-read)", lookups) + } +} + +// TestEnsureSessionForAgentWithCreation_RecoveryReReadFindsLiveWinnerReusesWithoutRetry +// is the mirror of the terminal-retry test above and closes two related +// AC-003.7 gaps flagged by Review entry 13: +// +// 1. "No retry spent" is otherwise unasserted: the concurrent convergence +// test proves one creator and N-1 reusers but never asserts that a +// caller which reuses via the recovery re-read stops after exactly the +// one (failed) create attempt that lost the race, rather than looping to +// spend the bounded loop's second attempt as well. +// 2. The concurrent convergence test doesn't reliably prove the +// bounded-recovery re-read path was traversed at all - sampling found +// roughly 1-in-5 runs where every goroutine happened to win or lose +// without any of them observing a live winner through the recovery +// re-read. This test drives that exact interleaving deterministically, +// independent of goroutine scheduling. +func TestEnsureSessionForAgentWithCreation_RecoveryReReadFindsLiveWinnerReusesWithoutRetry(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + task := officeTestTask() + + var createAttempts int + repo.createTaskSessionFunc = func(_ context.Context, _ *models.TaskSession) error { + createAttempts++ + return taskrepo.ErrOfficeSessionRaceConflict + } + + liveWinner := &models.TaskSession{ + ID: "recovery-live-winner", TaskID: task.ID, AgentProfileID: "agent-recovery-live", + State: models.TaskSessionStateRunning, StartedAt: time.Now().UTC(), + // Matches the execution profile passed to EnsureSessionForAgentWithCreation + // below so rebindOfficeSessionExecutionProfile's no-op fast path (the + // row is already on the right profile) applies - this test targets + // tryReuseExistingSession's outcome, not the rebind CAS update, and the + // winner row was never written to the mock's session store. + ExecutionProfileID: "profile-1", + } + var lookups int + repo.getTaskSessionByTaskAndAgentFunc = func(_ context.Context, _, _ string) (*models.TaskSession, error) { + lookups++ + if lookups == 1 { + return nil, nil + } + // The recovery re-read: the winner has already committed and is + // still live (not terminal), so this caller must reuse it. + return liveWinner, nil + } + + session, wasCreated, err := exec.EnsureSessionForAgentWithCreation( + context.Background(), task, "agent-recovery-live", "profile-1", "exec-1", "", + ) + if err != nil { + t.Fatalf("err = %v, want nil", err) + } + if wasCreated { + t.Fatal("wasCreated = true, want false (this caller reused the live winner, it did not create)") + } + if session == nil || session.ID != liveWinner.ID { + t.Fatalf("session = %#v, want the live winner row %q", session, liveWinner.ID) + } + if createAttempts != 1 { + t.Fatalf("create attempts = %d, want exactly 1 (reusing a live winner must not spend the bounded loop's second attempt)", createAttempts) + } + if lookups != 2 { + t.Fatalf("recovery lookups = %d, want exactly 2 (initial lookup + post-conflict re-read), proving the recovery branch was actually traversed", lookups) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index c20e000c31..dfe1d86d6b 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -309,6 +309,7 @@ type sessionExecutorStore interface { RenameTaskSession(ctx context.Context, id, name string) error UpdateTaskSession(ctx context.Context, session *models.TaskSession) error UpdateTaskSessionIfCurrentState(ctx context.Context, session *models.TaskSession, expected models.TaskSessionState) (bool, error) + UpdateTaskSessionStateIfCurrent(ctx context.Context, id string, expected, status models.TaskSessionState, errorMessage string) (bool, time.Time, error) UpdateTaskSessionState(ctx context.Context, id string, state models.TaskSessionState, errorMessage string) error ClaimPromptableTaskSessionIfActive(ctx context.Context, id string) (models.PromptableTaskSessionClaim, error) UpdateTaskSessionBaseCommit(ctx context.Context, id string, baseCommitSHA string) error diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 1f742f75b8..67951eceb0 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -5247,6 +5247,78 @@ func TestStartTaskPublishesCreatedSessionBeforeLaunch(t *testing.T) { assert.True(t, publishedBeforeLaunch, "created session event must arrive before the runtime starts") } +// TestStartTaskWithEnv_OfficeCreateThenReusePublishesOneCreatedEvent drives +// two sequential office wakeups for the same (task, agent) pair through the +// full StartTaskWithEnv path — not just the repository call-count proxy used +// by office_session_race_guard_test.go's convergence tests — and asserts on +// the actual event bus. The first wakeup has no live session and must create +// one, publishing exactly one TaskSessionStateChanged/CREATED event for it. +// The second wakeup reuses that same live session (EnsureSessionForAgentWithCreation) +// and must not publish a second CREATED event for it: office wakeups often +// share a session across many turns, and a duplicate CREATED event would make +// the frontend re-adopt an already-running session as if it were new. +func TestStartTaskWithEnv_OfficeCreateThenReusePublishesOneCreatedEvent(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "existing-session", models.TaskSessionStateCompleted) + dbTask, err := repo.GetTask(ctx, "task1") + require.NoError(t, err) + dbTask.ProjectID = "office-project" + require.NoError(t, repo.UpdateTask(ctx, dbTask)) + + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", + Title: "Office task", + Description: "Do the work", + State: v1.TaskStateInProgress, + } + eventBus := &mockEventBus{} + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-" + req.SessionID}, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.eventBus = eventBus + env := validOfficeRuntimeEnv() + + firstExec, err := svc.StartTaskWithEnv( + ctx, "task1", "office-runner", "", "", "", "Do the work", + "", false, false, nil, env, + ) + require.NoError(t, err) + require.NotNil(t, firstExec) + firstSessionID := firstExec.SessionID + require.NotEqual(t, "existing-session", firstSessionID, + "a completed session must not be reused as the live office session") + + secondExec, err := svc.StartTaskWithEnv( + ctx, "task1", "office-runner", "", "", "", "Do the work", + "", false, false, nil, env, + ) + require.NoError(t, err) + require.NotNil(t, secondExec) + require.Equal(t, firstSessionID, secondExec.SessionID, + "second office wakeup for the same (task, agent) pair must reuse the live session") + + createdEventsForSession := 0 + for _, published := range eventBus.published() { + if published.Subject != events.TaskSessionStateChanged { + continue + } + data, ok := published.Event.Data.(map[string]any) + if !ok { + continue + } + if data[metaKeySessionID] == firstSessionID && data[metaKeyNewState] == string(models.TaskSessionStateCreated) { + createdEventsForSession++ + } + } + require.Equal(t, 1, createdEventsForSession, + "create+reuse across two office wakeups must publish exactly one CREATED event, not zero and not two") +} + func TestStartTask_PreservesOnlyResolvedWorkflowPromptExpansion(t *testing.T) { for _, tc := range []struct { name string diff --git a/apps/backend/internal/task/repository/sqlite/errors.go b/apps/backend/internal/task/repository/sqlite/errors.go index aeb1a4bfe9..e52ee5e9dc 100644 --- a/apps/backend/internal/task/repository/sqlite/errors.go +++ b/apps/backend/internal/task/repository/sqlite/errors.go @@ -40,33 +40,26 @@ var ErrNoPrimarySession = errors.New("no primary session") // return the winner rather than treating this as a hard failure. var ErrExternalIDConflict = repoerrors.ErrExternalIDConflict -// ErrOfficeSessionRaceConflict is classified by isOfficeTaskSessionUniqueViolation -// (office_task_session_uniqueness.go) and returned by every session-create -// method that routes through the unexported createTaskSession — -// CreateTaskSession, CreateTaskSessionWithInitialRuntimeSeed, -// CreateTaskSessionWithWorkspaceBinding, -// CreateTaskSessionWithSharedGroupWorkspaceBinding, and -// CreateOfficeTaskSession — and by every full-row session update that routes -// through the unexported updateTaskSessionWithStateGuard — UpdateTaskSession, -// UpdateTaskSessionWithMetadata, UpdateTaskSessionIfCurrentState, and -// UpdateTaskSessionIfCurrentStateRemovingMetadataKeys — when a write violates -// uniq_office_task_session — i.e. two callers raced past their -// SELECT-then-INSERT (or a resume raced a fresh create) for the same -// (task_id, agent_profile_id) pair while both rows -// are "live" (CREATED, STARTING, RUNNING, IDLE, or WAITING_FOR_INPUT). -// Terminal rows for the same pair are unaffected — the index only -// constrains the live set. Callers should re-read and reuse the winning row -// rather than treating this as a hard failure (see -// executor_office.go's EnsureSessionForAgentWithCreation). -// -// uniq_office_task_session does NOT exist in the schema yet: a table-wide -// version of it broke live kanban-relaunch and workflow-replacement flows, -// which intentionally create a second live session for the same pair (see -// TestCreateStartSession_KanbanRunnerCreatesDistinctSession). Scoping the -// index to office sessions only is tracked as Kandev task -// 05864a73-2dd9-4b15-98ab-35643d9d55e4; until it lands, this sentinel cannot -// fire and the guarded call sites are dead code kept for that follow-up to -// activate. +// ErrOfficeSessionRaceConflict is returned by CreateOfficeTaskSession (and, +// for repositories that don't implement the office creator interface, by the +// equivalent in-process fallback in executor_office.go's +// persistOfficeSessionFallback) when an in-transaction guard finds a row +// already live for the same (task_id, agent_profile_id) pair — i.e. two +// callers raced past their lookup-then-create for the same pair while both +// rows are "live" (any state other than COMPLETED, FAILED, or CANCELLED). +// The guard is a SELECT ... WHERE ... AND state NOT IN (...) check inside +// CreateOfficeTaskSession's own transaction (PostgreSQL's row lock and +// SQLite's single writer connection make that check-then-insert atomic +// against concurrent callers), not a database constraint — deliberately +// scoped to the office creation path only, never table-wide: a table-wide +// unique index on this pair broke live kanban-relaunch and +// workflow-replacement flows, which intentionally create a second live +// session for the same pair (see +// TestCreateStartSession_KanbanRunnerCreatesDistinctSession). Terminal rows +// for the same pair are unaffected — the guard only refuses when a live row +// exists. Callers should re-read and reuse the winning row rather than +// treating this as a hard failure (see executor_office.go's +// EnsureSessionForAgentWithCreation). var ErrOfficeSessionRaceConflict = errors.New("office task session race conflict") // ErrWorkflowResolutionConflict is returned by UpdateTaskIfWorkflowMatches and diff --git a/apps/backend/internal/task/repository/sqlite/office_session_race_guard_test.go b/apps/backend/internal/task/repository/sqlite/office_session_race_guard_test.go new file mode 100644 index 0000000000..fb4efe5e2a --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/office_session_race_guard_test.go @@ -0,0 +1,569 @@ +package sqlite + +// Tests for the in-transaction office-session live-pair guard (Change 1) and +// the live-preferring lookup ordering (Change 2). See +// docs/specs/office/{requirements,system-design}/task-session-identity*.md. + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/testutil" +) + +// insertOfficeSessionWithStartedAt is insertOfficeSession +// (office_task_session_uniqueness_test.go) with an explicit started_at, so +// ordering tests can control which row is "newer" independent of insertion +// order. +func insertOfficeSessionWithStartedAt(t *testing.T, db *sqlx.DB, s officeSessionSeed, startedAt time.Time) { + t.Helper() + now := time.Now().UTC() + _, err := db.Exec(db.Rebind(` + INSERT INTO task_sessions ( + id, task_id, agent_profile_id, executor_id, executor_profile_id, environment_id, + repository_id, base_branch, base_commit_sha, + agent_profile_snapshot, executor_snapshot, environment_snapshot, repository_snapshot, + state, error_message, metadata, started_at, completed_at, updated_at, + is_primary, review_status, is_passthrough, task_environment_id + ) VALUES (?, ?, ?, '', '', '', '', '', '', + '{}', '{}', '{}', '{}', + ?, '', '{}', ?, NULL, ?, + 0, '', 0, '') + `), s.id, s.taskID, s.agentProfileID, string(s.state), startedAt, now) + if err != nil { + t.Fatalf("insert office session %s: %v", s.id, err) + } +} + +// TestCreateOfficeTaskSessionRefusesConcurrentLivePair races two +// CreateOfficeTaskSession calls for the SAME (task_id, agent_profile_id) +// pair. Unlike TestCreateOfficeTaskSessionMarksOnlyTheFirstConcurrentSessionAsOrigin +// (which races two DIFFERENT pairs and asserts both succeed), this must +// converge to exactly one live row: one call succeeds, the other is refused +// with ErrOfficeSessionRaceConflict, and the refused call's row is never +// written (AC-001.1, AC-001.2, AC-001.6). +func TestCreateOfficeTaskSessionRefusesConcurrentLivePair(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-live-pair-race" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office live pair race"})) + + agent := "agent-live-pair" + sessions := []*models.TaskSession{ + {ID: "office-live-pair-1", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated}, + {ID: "office-live-pair-2", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated}, + } + errs := make([]error, len(sessions)) + var wg sync.WaitGroup + for i, session := range sessions { + wg.Add(1) + go func(i int, session *models.TaskSession) { + defer wg.Done() + errs[i] = repo.CreateOfficeTaskSession(ctx, session) + }(i, session) + } + wg.Wait() + + successCount, conflictCount := 0, 0 + for _, err := range errs { + switch { + case err == nil: + successCount++ + case errors.Is(err, ErrOfficeSessionRaceConflict): + conflictCount++ + default: + t.Fatalf("unexpected error: %v", err) + } + } + require.Equal(t, 1, successCount, "exactly one concurrent create should win the pair") + require.Equal(t, 1, conflictCount, "the loser should be refused with ErrOfficeSessionRaceConflict") + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 1, "the refused call must not have written a row") +} + +func TestPostgresCreateOfficeTaskSessionRefusesConcurrentLivePair(t *testing.T) { + db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2) + repo, err := NewWithDB(db, db, nil) + require.NoError(t, err) + ctx := context.Background() + const taskID = "task-office-live-pair-race-pg" + now := time.Now().UTC() + _, err = db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), taskID, "", "Office live pair race (Postgres)", now, now) + require.NoError(t, err) + + agent := "agent-live-pair-pg" + sessions := []*models.TaskSession{ + {ID: "office-live-pair-pg-1", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated}, + {ID: "office-live-pair-pg-2", TaskID: taskID, AgentProfileID: agent, State: models.TaskSessionStateCreated}, + } + errs := make([]error, len(sessions)) + var wg sync.WaitGroup + for i, session := range sessions { + wg.Add(1) + go func(i int, session *models.TaskSession) { + defer wg.Done() + errs[i] = repo.CreateOfficeTaskSession(ctx, session) + }(i, session) + } + wg.Wait() + + successCount, conflictCount := 0, 0 + for _, err := range errs { + switch { + case err == nil: + successCount++ + case errors.Is(err, ErrOfficeSessionRaceConflict): + conflictCount++ + default: + t.Fatalf("unexpected error: %v", err) + } + } + require.Equal(t, 1, successCount) + require.Equal(t, 1, conflictCount) + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 1) +} + +// TestCreateOfficeTaskSessionAllowsCreateWhenExistingPairRowIsTerminal proves +// the guard's predicate is complement-of-terminal, not "any row for this +// pair": a pair whose only existing row is terminal must still allow a fresh +// create (AC-001.3, AC-001.4). +func TestCreateOfficeTaskSessionAllowsCreateWhenExistingPairRowIsTerminal(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-terminal-pair-retry" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office terminal pair retry"})) + + agent := "agent-terminal-retry" + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-terminal-retry-1", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCompleted, + })) + + err := repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-terminal-retry-2", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + }) + require.NoError(t, err, "a terminal-only pair must not block a fresh create") + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 2) +} + +// TestCreateOfficeTaskSessionRefusesLiveWhenTerminalRowsAlsoExist proves the +// guard fires even when older terminal rows coexist with the live one — the +// predicate is "any live row exists", not "all rows are live". +func TestCreateOfficeTaskSessionRefusesLiveWhenTerminalRowsAlsoExist(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-live-amid-terminal" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office live amid terminal"})) + + agent := "agent-live-amid-terminal" + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-live-amid-terminal-old", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCompleted, + })) + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-live-amid-terminal-live", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateRunning, + })) + + err := repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-live-amid-terminal-blocked", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrOfficeSessionRaceConflict)) + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 2, "the refused create must not have written a row") +} + +// TestCreateOfficeTaskSessionRefusalLeavesExistingRowsUnmodified closes +// Review round 5's Finding C: AC-001.6 requires a refused create to leave +// every existing row for the pair unmodified — no row updated, cancelled, +// merged, or deleted, and no metadata moved between rows. Every other +// refusal test in this file asserts row COUNT only, which a future +// row-healing regression (e.g. cancelling the stale duplicate inside the +// guard) could satisfy while still mutating a row. This snapshots the full +// pre-existing rows before the refused call and asserts they are unchanged +// afterward. +func TestCreateOfficeTaskSessionRefusalLeavesExistingRowsUnmodified(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-refusal-leaves-rows-unmodified" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office refusal leaves rows unmodified"})) + + agent := "agent-refusal-unmodified" + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "refusal-unmodified-old", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCompleted, + })) + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "refusal-unmodified-live", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateRunning, + ExecutionProfileID: "profile-before-refusal", + DownstreamACPSessionID: "acp-session-before-refusal", + })) + + before, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, before, 2) + + err = repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "refusal-unmodified-blocked", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrOfficeSessionRaceConflict)) + + after, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, after, 2, "the refused create must not have written a row") + require.Equal(t, before, after, + "a refused create must leave every existing row for the pair byte-identical — no update, cancel, merge, or metadata move") +} + +// TestCreateOfficeTaskSessionSkipsGuardForEmptyAgentProfileID proves the +// guard is bypassed entirely when agent_profile_id is empty (AC-001.5): +// two rows for the same task with no agent profile are unrestricted, exactly +// like today's kanban behavior. +func TestCreateOfficeTaskSessionSkipsGuardForEmptyAgentProfileID(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-empty-agent-bypass" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office empty agent bypass"})) + + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-empty-agent-1", TaskID: taskID, State: models.TaskSessionStateCreated, + })) + err := repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-empty-agent-2", TaskID: taskID, State: models.TaskSessionStateCreated, + }) + require.NoError(t, err, "empty agent_profile_id must bypass the live-pair guard") + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 2) +} + +// TestCreateOfficeTaskSessionRefusesLiveRowCreatedByNonOfficePath is the +// repository-layer half of AC-OFFICE-SESSION-IDENTITY-001.8: the guard +// applies to any live row for the pair, including one created by a +// non-office path such as a session spawned on an office task that inherited +// the spawner's profile. Every other guard test in this file seeds its +// pre-existing row through CreateOfficeTaskSession itself, so none of them +// proves the guard also fires against a row that was never inserted through +// the office path. `task_sessions` has no column distinguishing which path +// created a row, so the plain CreateTaskSession call here stands in for that +// non-office path: it produces a live row identical in shape to one the +// office guard would itself have written. +func TestCreateOfficeTaskSessionRefusesLiveRowCreatedByNonOfficePath(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-nonoffice-live-pair" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office guard vs non-office row"})) + + agent := "agent-nonoffice-spawned" + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "nonoffice-spawned-1", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateRunning, + })) + + err := repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "office-wakeup-1", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrOfficeSessionRaceConflict), + "the office guard must refuse a live row for the pair even when that row was not created via CreateOfficeTaskSession") + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 1, "the refused office create must not have written a second row") +} + +// TestCreateOfficeTaskSessionRefusesLiveForNonCanonicalLiveStates closes +// Review entry 19's MUST-FIX 3 (SQLite half): AC-OFFICE-SESSION-IDENTITY-001.4 +// requires the guard to be the complement of the terminal set +// (COMPLETED/FAILED/CANCELLED), not an enumerated allow-list of whichever +// live states happen to be exercised elsewhere. Every other guard test in +// this file seeds CREATED or RUNNING; STARTING and WAITING_FOR_INPUT are +// live states none of them touch. If the guard's SQL predicate were rewritten +// as `state IN ('CREATED','RUNNING')` instead of `state NOT IN (terminal +// set)`, every other test here would stay green while a pair whose only row +// sits in one of these two states would wrongly allow a duplicate create. +// +// IDLE and a fabricated out-of-enum state close Review round 5's Finding B: +// IDLE is an office session's dominant resting state between turns, and no +// guard test at any layer previously seeded it as the blocking row — an +// allow-list regression omitting IDLE would have shipped green. The +// fabricated state proves the complement form extends to states that don't +// exist yet, per AC-001.4's "a state added later is treated identically" +// requirement. +func TestCreateOfficeTaskSessionRefusesLiveForNonCanonicalLiveStates(t *testing.T) { + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateStarting, + models.TaskSessionStateWaitingForInput, + models.TaskSessionStateIdle, + models.TaskSessionState("PAUSED_FUTURE_STATE"), + } { + t.Run(string(state), func(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + taskID := "task-office-noncanonical-live-" + string(state) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office non-canonical live state"})) + + agent := "agent-noncanonical-live" + require.NoError(t, repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "noncanonical-live-existing", TaskID: taskID, AgentProfileID: agent, + State: state, + })) + + err := repo.CreateOfficeTaskSession(ctx, &models.TaskSession{ + ID: "noncanonical-live-blocked", TaskID: taskID, AgentProfileID: agent, + State: models.TaskSessionStateCreated, + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrOfficeSessionRaceConflict), + "a pair whose only row is %s must be treated as live, not terminal", state) + + created, err := repo.ListTaskSessions(ctx, taskID) + require.NoError(t, err) + require.Len(t, created, 1, "the refused create must not have written a row") + }) + } +} + +// TestGetTaskSessionByTaskAndAgentPrefersLiveOverNewerTerminal is the +// 62201cdb-shaped livelock regression test: a terminal row started AFTER a +// live row must not shadow it. Before Change 2, ORDER BY started_at DESC +// alone would return the newer terminal row, causing the office scheduler to +// treat a live pair as if it had no session and create a duplicate. +func TestGetTaskSessionByTaskAndAgentPrefersLiveOverNewerTerminal(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-lookup-livelock" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office lookup livelock"})) + agent := "agent-lookup-livelock" + + older := time.Now().UTC().Add(-time.Hour) + newer := time.Now().UTC() + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-livelock-live", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateRunning, + }, older) + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-livelock-terminal", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCompleted, + }, newer) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-livelock-live", got.ID, "the live row must win over a newer terminal row") +} + +// TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc covers +// the total tiebreak: multiple live rows for the same pair (legacy +// duplicate-pair data, AC-003.6) resolve deterministically by started_at +// DESC, then id DESC when started_at ties exactly. +func TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-lookup-tiebreak" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office lookup tiebreak"})) + agent := "agent-lookup-tiebreak" + + same := time.Now().UTC() + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-tiebreak-a", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, same) + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-tiebreak-b", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, same) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-tiebreak-b", got.ID, "equal started_at must tiebreak on id DESC") +} + +// TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescBeforeID exercises +// the started_at DESC clause itself, not just its id DESC tiebreak: two live +// rows with DISTINCT started_at values, where the row with the LATER +// started_at has the lexicographically SMALLER id. Ordering by id DESC alone +// (or with started_at DESC dropped or reordered after id) would pick the +// wrong row here, unlike TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc +// above, whose two rows share one started_at and so cannot tell started_at +// DESC apart from an id-only ordering. +func TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescBeforeID(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-lookup-started-at-precedence" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office lookup started_at precedence"})) + agent := "agent-lookup-started-at-precedence" + + older := time.Now().UTC().Add(-time.Hour) + newer := time.Now().UTC() + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-started-at-precedence-z-older", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, older) + insertOfficeSessionWithStartedAt(t, repo.db, officeSessionSeed{ + id: "lookup-started-at-precedence-a-newer", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, newer) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-started-at-precedence-a-newer", got.ID, + "the later started_at must win even though its id sorts lower than the older row's") +} + +func TestPostgresGetTaskSessionByTaskAndAgentPrefersLiveOverNewerTerminal(t *testing.T) { + db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2) + repo, err := NewWithDB(db, db, nil) + require.NoError(t, err) + ctx := context.Background() + const taskID = "task-office-lookup-livelock-pg" + now := time.Now().UTC() + _, err = db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), taskID, "", "Office lookup livelock (Postgres)", now, now) + require.NoError(t, err) + agent := "agent-lookup-livelock-pg" + + older := now.Add(-time.Hour) + newer := now + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-livelock-pg-live", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateRunning, + }, older) + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-livelock-pg-terminal", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCompleted, + }, newer) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-livelock-pg-live", got.ID) +} + +// TestPostgresGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc +// is the PostgreSQL dialect counterpart of +// TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc: two +// live rows sharing one started_at value tiebreak deterministically on id +// DESC, proving the ordering clause carries the same total order on this +// dialect, not just SQLite's. +func TestPostgresGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescThenIDDesc(t *testing.T) { + db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2) + repo, err := NewWithDB(db, db, nil) + require.NoError(t, err) + ctx := context.Background() + const taskID = "task-office-lookup-tiebreak-pg" + now := time.Now().UTC() + _, err = db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), taskID, "", "Office lookup tiebreak (Postgres)", now, now) + require.NoError(t, err) + agent := "agent-lookup-tiebreak-pg" + + same := now + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-tiebreak-pg-a", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, same) + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-tiebreak-pg-b", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, same) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-tiebreak-pg-b", got.ID, "equal started_at must tiebreak on id DESC") +} + +// TestPostgresGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescBeforeID +// is the PostgreSQL dialect counterpart of +// TestGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescBeforeID: two live +// rows with distinct started_at values where the later row has the +// lexicographically smaller id, proving the query orders by started_at DESC +// ahead of id DESC rather than the reverse. +func TestPostgresGetTaskSessionByTaskAndAgentOrdersLiveByStartedAtDescBeforeID(t *testing.T) { + db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2) + repo, err := NewWithDB(db, db, nil) + require.NoError(t, err) + ctx := context.Background() + const taskID = "task-office-lookup-started-at-precedence-pg" + now := time.Now().UTC() + _, err = db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), taskID, "", "Office lookup started_at precedence (Postgres)", now, now) + require.NoError(t, err) + agent := "agent-lookup-started-at-precedence-pg" + + older := now.Add(-time.Hour) + newer := now + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-started-at-precedence-pg-z-older", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, older) + insertOfficeSessionWithStartedAt(t, db, officeSessionSeed{ + id: "lookup-started-at-precedence-pg-a-newer", taskID: taskID, agentProfileID: agent, + state: models.TaskSessionStateCreated, + }, newer) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, agent) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, "lookup-started-at-precedence-pg-a-newer", got.ID, + "the later started_at must win even though its id sorts lower than the older row's") +} + +// TestGetTaskSessionByTaskAndAgentReturnsNilForAbsentPairOrEmptyIdentifiers +// covers AC-002's nil,nil contract: no matching row, and both +// empty-identifier short-circuits, must not be treated as an error. +func TestGetTaskSessionByTaskAndAgentReturnsNilForAbsentPairOrEmptyIdentifiers(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const taskID = "task-office-lookup-absent" + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: taskID, Title: "Office lookup absent"})) + + got, err := repo.GetTaskSessionByTaskAndAgent(ctx, taskID, "agent-never-seen") + require.NoError(t, err) + require.Nil(t, got) + + got, err = repo.GetTaskSessionByTaskAndAgent(ctx, "", "agent-never-seen") + require.NoError(t, err) + require.Nil(t, got) + + got, err = repo.GetTaskSessionByTaskAndAgent(ctx, taskID, "") + require.NoError(t, err) + require.Nil(t, got) +} diff --git a/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_postgres_test.go b/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_postgres_test.go index 4d6ef158c6..79b82d8e95 100644 --- a/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_postgres_test.go +++ b/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_postgres_test.go @@ -1,27 +1,36 @@ package sqlite import ( - "context" - "errors" "testing" + "time" - "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/testutil" ) -// TestPostgresOfficeTaskSessionUniqueViolation covers the typed PostgreSQL -// error branch and both repository write paths with a temporary scoped index. -// The production migration remains deferred because the shared task_sessions -// table also serves Kanban sessions. -func TestPostgresOfficeTaskSessionUniqueViolation(t *testing.T) { +// TestIsOfficeTaskSessionUniqueViolation_ClassifiesPostgresConstraint +// exercises isOfficeTaskSessionUniqueViolation directly against a real +// PostgreSQL unique-constraint violation. The classifier's two production +// call sites (createTaskSession's INSERT and updateTaskSessionWithStateGuard's +// UPDATE) were deleted per the system design's "Baseline" instruction — +// office session-uniqueness enforcement now lives only in +// CreateOfficeTaskSession's in-transaction guard (session.go), which never +// inspects driver error types — so this function is intentionally +// referenced only by this test and its SQLite sibling +// (TestIsOfficeTaskSessionUniqueViolation_ClassifiesSQLiteMessage). +// uniq_office_task_session is not wired into base_schema.go, so this test +// builds and drops its own scoped index rather than relying on production +// schema, and drives the raw insert directly instead of through +// CreateTaskSession/UpdateTaskSessionIfCurrentState, since neither +// production call path translates this error anymore. +func TestIsOfficeTaskSessionUniqueViolation_ClassifiesPostgresConstraint(t *testing.T) { db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) repo, err := NewWithDB(db, db, nil) if err != nil { t.Fatalf("init postgres schema: %v", err) } - ctx := context.Background() - const taskID = "task-office-unique-pg" + const taskID = "task-classify-pg" seedPostgresTask(t, repo, taskID) + agent := "agent-classify-pg" if _, err := repo.db.Exec(` CREATE UNIQUE INDEX uniq_office_task_session @@ -36,43 +45,48 @@ func TestPostgresOfficeTaskSessionUniqueViolation(t *testing.T) { _, _ = repo.db.Exec(`DROP INDEX IF EXISTS uniq_office_task_session`) }) - first := &models.TaskSession{ - ID: "sess-office-unique-pg-1", - TaskID: taskID, - AgentProfileID: "agent-office-pg", - State: models.TaskSessionStateCreated, - } - if err := repo.CreateTaskSession(ctx, first); err != nil { - t.Fatalf("create first office session: %v", err) + now := time.Now().UTC() + if _, err := repo.db.Exec(repo.db.Rebind(` + INSERT INTO task_sessions (id, task_id, agent_profile_id, state, started_at, updated_at) + VALUES (?, ?, ?, 'CREATED', ?, ?) + `), "sess-classify-pg-1", taskID, agent, now, now); err != nil { + t.Fatalf("seed first office session: %v", err) } - duplicate := &models.TaskSession{ - ID: "sess-office-unique-pg-duplicate", - TaskID: taskID, - AgentProfileID: "agent-office-pg", - State: models.TaskSessionStateCreated, + _, err = repo.db.Exec(repo.db.Rebind(` + INSERT INTO task_sessions (id, task_id, agent_profile_id, state, started_at, updated_at) + VALUES (?, ?, ?, 'CREATED', ?, ?) + `), "sess-classify-pg-2", taskID, agent, now, now) + if err == nil { + t.Fatal("expected inserting a second live row for the same pair to violate the scoped index") } - err = repo.CreateTaskSession(ctx, duplicate) - if err == nil || !errors.Is(err, ErrOfficeSessionRaceConflict) { - t.Fatalf("duplicate insert error = %v, want ErrOfficeSessionRaceConflict", err) - } - - updateCandidate := &models.TaskSession{ - ID: "sess-office-unique-pg-update", - TaskID: taskID, - AgentProfileID: "agent-other-pg", - State: models.TaskSessionStateCreated, - } - if err := repo.CreateTaskSession(ctx, updateCandidate); err != nil { - t.Fatalf("create update candidate: %v", err) - } - updateCandidate.AgentProfileID = "agent-office-pg" - changed, err := repo.UpdateTaskSessionIfCurrentState( - ctx, - updateCandidate, - models.TaskSessionStateCreated, - ) - if changed || err == nil || !errors.Is(err, ErrOfficeSessionRaceConflict) { - t.Fatalf("conflicting update = changed=%t err=%v, want typed race conflict", changed, err) + if !isOfficeTaskSessionUniqueViolation(err) { + t.Fatalf("expected err to classify as an office task session unique violation, got: %v", err) + } +} + +// TestIsOfficeTaskSessionUniqueViolation_IgnoresUnrelatedPostgresViolation +// asserts the classifier does not fire on an unrelated PostgreSQL +// unique-constraint violation (the tasks primary key here), mirroring the +// SQLite sibling's negative case. +func TestIsOfficeTaskSessionUniqueViolation_IgnoresUnrelatedPostgresViolation(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + const taskID = "task-classify-pg-2" + seedPostgresTask(t, repo, taskID) + + now := time.Now().UTC() + _, err = repo.db.Exec(repo.db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), taskID, "ws-"+taskID, taskID, now, now) + if err == nil { + t.Fatal("expected duplicate task id to violate the tasks primary key") + } + if isOfficeTaskSessionUniqueViolation(err) { + t.Fatalf("expected primary-key violation not to be misclassified as an office task session unique violation: %v", err) } } diff --git a/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_test.go b/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_test.go index da6faac12a..a548318998 100644 --- a/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_test.go +++ b/apps/backend/internal/task/repository/sqlite/office_task_session_uniqueness_test.go @@ -1,8 +1,6 @@ package sqlite import ( - "context" - "errors" "testing" "time" @@ -42,16 +40,18 @@ func insertOfficeSession(t *testing.T, db *sqlx.DB, s officeSessionSeed) { } // TestIsOfficeTaskSessionUniqueViolation_ClassifiesSQLiteMessage exercises -// isOfficeTaskSessionUniqueViolation against a real SQLite UNIQUE-constraint -// error, then drives both production call sites that wrap it in -// ErrOfficeSessionRaceConflict (createTaskSession's INSERT and -// updateTaskSessionWithStateGuard's UPDATE) to prove the wrapping itself, -// not just the classifier. uniq_office_task_session is not wired into -// base_schema.go — a table-wide partial index on this pair broke live -// kanban-relaunch and workflow-replacement flows (see the follow-up card -// linked from ErrOfficeSessionRaceConflict's doc comment) — so this test -// builds and drops its own scoped index rather than relying on production -// schema. +// isOfficeTaskSessionUniqueViolation directly against a real SQLite +// UNIQUE-constraint error. The classifier's two production call sites +// (createTaskSession's INSERT and updateTaskSessionWithStateGuard's UPDATE) +// were deleted per the system design's "Baseline" instruction — office +// session-uniqueness enforcement lives only in CreateOfficeTaskSession's +// in-transaction guard now (session.go), which never parses driver error +// text — so this function is intentionally referenced only by this test +// file. uniq_office_task_session is not wired into base_schema.go — a +// table-wide partial index on this pair broke live kanban-relaunch and +// workflow-replacement flows (see the follow-up card linked from +// ErrOfficeSessionRaceConflict's doc comment) — so this test builds and +// drops its own scoped index rather than relying on production schema. func TestIsOfficeTaskSessionUniqueViolation_ClassifiesSQLiteMessage(t *testing.T) { repo := newRepoForHealTests(t) insertTask(t, repo.db, "task-classify-1") @@ -93,42 +93,6 @@ func TestIsOfficeTaskSessionUniqueViolation_ClassifiesSQLiteMessage(t *testing.T if !isOfficeTaskSessionUniqueViolation(err) { t.Fatalf("expected err to classify as an office task session unique violation, got: %v", err) } - - // The raw INSERT above only proves SQLite's message format. Also drive - // the two real production call sites that route through the - // classifier, so a change that breaks either wrapping fails here rather - // than only after the follow-up index lands. - ctx := context.Background() - - // CreateTaskSession -> createTaskSession's INSERT path. - createErr := repo.CreateTaskSession(ctx, &models.TaskSession{ - ID: "sess-classify-3", TaskID: "task-classify-1", AgentProfileID: agent, - State: models.TaskSessionStateCreated, - }) - if createErr == nil { - t.Fatal("expected CreateTaskSession into an occupied live (task, agent) slot to fail") - } - if !errors.Is(createErr, ErrOfficeSessionRaceConflict) { - t.Fatalf("expected errors.Is(err, ErrOfficeSessionRaceConflict) from CreateTaskSession, got: %v", createErr) - } - - // UpdateTaskSession -> updateTaskSessionWithStateGuard's UPDATE path - // (e.g. a terminal-session resume racing another live row into the - // same slot). - insertOfficeSession(t, repo.db, officeSessionSeed{ - id: "sess-classify-4", taskID: "task-classify-1", agentProfileID: "agent-classify-other", - state: models.TaskSessionStateCreated, - }) - updateErr := repo.UpdateTaskSession(ctx, &models.TaskSession{ - ID: "sess-classify-4", TaskID: "task-classify-1", AgentProfileID: agent, - State: models.TaskSessionStateCreated, - }) - if updateErr == nil { - t.Fatal("expected UpdateTaskSession moving a session into an occupied live (task, agent) slot to fail") - } - if !errors.Is(updateErr, ErrOfficeSessionRaceConflict) { - t.Fatalf("expected errors.Is(err, ErrOfficeSessionRaceConflict) from UpdateTaskSession, got: %v", updateErr) - } } // TestIsOfficeTaskSessionUniqueViolation_IgnoresUnrelatedViolation asserts diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index bbea598a2a..a7d2d7b468 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -841,7 +841,10 @@ func (r *Repository) loadInitialSessionRuntimeSeedTx( // CreateOfficeTaskSession creates an Office session and atomically marks it as // the task's initial session when no earlier session exists. The task row lock // serializes callers across PostgreSQL connections; SQLite's single writer -// connection serializes the transaction. +// connection serializes the transaction. Within that same transaction it also +// enforces office-session uniqueness for the (task_id, agent_profile_id) pair: +// if a live row already exists for the pair, the insert is refused with +// ErrOfficeSessionRaceConflict rather than creating a second live session. func (r *Repository) CreateOfficeTaskSession(ctx context.Context, session *models.TaskSession) error { tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -871,6 +874,20 @@ func (r *Repository) CreateOfficeTaskSession(ctx context.Context, session *model session.Metadata[models.SessionMetaKeyOrigin] = models.SessionOriginTaskInitial } + if session.AgentProfileID != "" { + var liveCount int + if err := tx.QueryRowContext(ctx, r.db.Rebind(` + SELECT COUNT(*) FROM task_sessions + WHERE task_id = ? AND agent_profile_id = ? + AND state NOT IN ('COMPLETED', 'FAILED', 'CANCELLED') + `), session.TaskID, session.AgentProfileID).Scan(&liveCount); err != nil { + return fmt.Errorf("check live office session for pair: %w", err) + } + if liveCount > 0 { + return ErrOfficeSessionRaceConflict + } + } + if err := r.createTaskSession(ctx, tx, session); err != nil { return err } @@ -949,14 +966,6 @@ func (r *Repository) createTaskSession(ctx context.Context, exec taskSessionExec dialect.BoolToInt(session.IsPrimary), session.ReviewStatus, dialect.BoolToInt(session.IsPassthrough), session.TaskEnvironmentID, session.Name) - if err != nil && isOfficeTaskSessionUniqueViolation(err) { - // Two callers raced past their SELECT-then-INSERT for the same - // (task_id, agent_profile_id) — surface a typed sentinel so callers - // can classify with errors.Is rather than driver-message matching. - // Currently unreachable: no index enforces this pair yet (see - // ErrOfficeSessionRaceConflict's doc comment). - return fmt.Errorf("%w: %w", ErrOfficeSessionRaceConflict, err) - } return err } @@ -1214,10 +1223,15 @@ func (r *Repository) GetActiveTaskSessionByTaskID(ctx context.Context, taskID st // GetTaskSessionByTaskAndAgent retrieves the office task session for the given // (task_id, agent_profile_id) pair. This pair is NOT unique at the schema -// level — no index enforces it (see ErrOfficeSessionRaceConflict's doc -// comment for why) — so both live and terminal rows can accumulate for the -// same pair, and ORDER BY started_at DESC LIMIT 1 is load-bearing to pick the -// most recent one. Returns nil, nil when no session exists for the pair. +// level — the office creation guard (CreateOfficeTaskSession) constrains it +// only at the moment of insert, so legacy duplicate-pair data and rows +// created outside that guard can still leave multiple rows for the same +// pair. The ORDER BY therefore prefers a live row over a terminal one +// regardless of which is newer — a terminal row created after a live row +// (e.g. a stale duplicate resolving while the real session is still running) +// must not shadow the live one — then falls back to started_at DESC, then id +// DESC as a total tiebreak. Returns nil, nil when no session exists for the +// pair. func (r *Repository) GetTaskSessionByTaskAndAgent(ctx context.Context, taskID, agentInstanceID string) (*models.TaskSession, error) { if taskID == "" || agentInstanceID == "" { return nil, nil @@ -1225,7 +1239,11 @@ func (r *Repository) GetTaskSessionByTaskAndAgent(ctx context.Context, taskID, a row := r.ro.QueryRowContext(ctx, r.ro.Rebind( `SELECT `+taskSessionSelectCols+` `+taskSessionFromClause+` WHERE ts.task_id = ? AND ts.agent_profile_id = ? - ORDER BY ts.started_at DESC LIMIT 1`, + ORDER BY + CASE WHEN ts.state IN ('COMPLETED', 'FAILED', 'CANCELLED') THEN 1 ELSE 0 END, + ts.started_at DESC, + ts.id DESC + LIMIT 1`, ), taskID, agentInstanceID) session, err := r.scanTaskSession(ctx, row, "task_sessions: no matching row") if errors.Is(err, models.ErrTaskSessionNotFound) { @@ -1422,9 +1440,9 @@ func (r *Repository) updateTaskSessionWithStateGuard( // caller's in-memory copy may be stale. // agent_profile_id is stored as NULL when empty. No unique index - // currently constrains (task_id, agent_profile_id) — see - // ErrOfficeSessionRaceConflict's doc comment (errors.go) for why, and - // for what still guards this pair despite that. + // constrains (task_id, agent_profile_id) at this UPDATE path — office + // session-uniqueness enforcement lives only in CreateOfficeTaskSession's + // in-transaction guard (see ErrOfficeSessionRaceConflict's doc comment). var agentProfileID interface{} if session.AgentProfileID != "" { agentProfileID = session.AgentProfileID @@ -1452,16 +1470,6 @@ func (r *Repository) updateTaskSessionWithStateGuard( } result, err := exec.ExecContext(ctx, r.db.Rebind(query), args...) if err != nil { - if isOfficeTaskSessionUniqueViolation(err) { - // A terminal-session resume (updateSessionStarting) or another - // full-row update raced another live row into the same - // (task_id, agent_profile_id) slot — same classification as the - // createTaskSession INSERT path, so callers can errors.Is this - // regardless of which write hit the constraint. Currently - // unreachable: no index enforces this pair yet (see - // ErrOfficeSessionRaceConflict's doc comment). - return false, fmt.Errorf("%w: %w", ErrOfficeSessionRaceConflict, err) - } return false, err } rows, err := result.RowsAffected() diff --git a/docs/specs/office/requirements/task-session-identity.md b/docs/specs/office/requirements/task-session-identity.md new file mode 100644 index 0000000000..d531d1f23e --- /dev/null +++ b/docs/specs/office/requirements/task-session-identity.md @@ -0,0 +1,340 @@ +--- +status: draft +system: office +created: 2026-08-31 +owners: + - kandev +--- + +# Office Task Session Identity Requirements + +## Overview + +`AC-OFFICE-TASKS-001.2` already states the intended invariant: an office task +has "one [session] per `(task_id, agent_instance_id)` pair". Nothing enforces +it. Concurrent office wakeups for one `(task, agent)` pair create three or four +live `task_sessions` rows within a second, and the guard written to catch that +(`ErrOfficeSessionRaceConflict`) has never fired: the condition it tests for +cannot occur. + +This capability makes the invariant real and scopes it correctly. It is true of +the **office session-reuse path only**, not of the `task_sessions` table: a +previous attempt to enforce it table-wide with a unique index broke three +shipped flows that deliberately hold two live rows for one pair (see +REQ-OFFICE-SESSION-IDENTITY-004). + +Two defects produce the observed behaviour, and both must be fixed together or +the fix is worse than the bug: + +1. **A read-then-write window across two connection pools.** The office + find-or-create reads through the read-only pool (four connections) and + inserts through the single writer. Concurrent callers all read "no session", + then all insert. The reader pool's width and the observed duplicate count + match. +2. **A non-deterministic lookup that can prefer a dead row.** The lookup + returns the newest row by `started_at` regardless of state. When the newest + row for a pair is terminal and an older row is still live, the caller reads + terminal, decides to create, and any guard on creation then refuses. The + caller re-reads, gets the same terminal row, and fails permanently. This is + not hypothetical: one pair in the shipped database is already in that shape, + so enforcement without a lookup fix converts a duplicate-row bug into a task + that can never start again. + +Office owns this contract because office owns the notion that a `(task, agent)` +pair names one durable conversation. The task repository owns the storage +primitive and the non-office creation paths, which this contract leaves +untouched. + +## Terminology + +- **Office session** - a `task_sessions` row created through the office + find-or-create path (`EnsureSessionForAgentWithCreation` -> + `createOfficeSession` -> the repository's office creator). Office sessions are + the only rows this contract constrains. +- **Terminal state** - `COMPLETED`, `FAILED`, or `CANCELLED`. +- **Live** - any state that is not terminal. Defined as the complement, not an + enumerated list, so a state added later is treated as live by both the reuse + rule and the creation guard rather than by only one. +- **Pair** - a `(task_id, agent_profile_id)` tuple with a non-empty + `agent_profile_id`. + +## Prior art + +**Our own prior reasoning (wiki):** unavailable - no `OBSIDIAN_VAULT_PATH` and +no vault on this machine, so the leg could not run. Recorded, not faked. + +**In-repo prior decisions**, which did apply: + +- `AC-OFFICE-TASKS-001.2` and `AC-OFFICE-TASKS-001.3` already assert one + session per pair and define the terminal set. This capability enforces an + existing contract; it does not invent one. Where the two could drift, this + document defers to `tasks.md`. +- `system-design/tasks-01.md`'s DDL block for a unique index on + `task_sessions(task_id, agent_instance_id)` names a column ADR 0005 renamed, + and the index was never created. Stale, not a live decision, superseded here. +- **ADR 0008** (idempotent migrations, pre-migration snapshot on upgrade boots) + and **ADR 0027** (replayable schema migrations) both favour a design needing + no schema change and no backfill, which this one does not. +- **ADR 0005 (agent model unification)** made `agent_profile_id` shared between + kanban and office. That is precisely why the pair cannot be treated as + office-only at the schema level, and why enforcement belongs on the office + code path instead. + +**What others shipped (saas-kb, `category: ai_sdlc`):** nothing useful. Two +queries returned unrelated vendor docs at relevance 0.0094-0.0137; no vendor in +that corpus documents per-agent session identity, so there is nothing to copy +or deliberately diverge from. + +**What we are doing differently:** the obvious move, which the stale DDL and the +previous attempt both reached for, is a database unique index. We reject it. The +invariant is a property of one code path, not of the table, and the column a +partial index predicate would need is shared with the paths that must stay +allowed. Enforcement therefore lives in the office creation path's existing +transaction, which already establishes the mutual exclusion required. This +trades a cross-process guarantee for correctness on the paths that exist; see +"Out of scope". + +## Requirements + +### REQ-OFFICE-SESSION-IDENTITY-001: At most one live office session per pair + +**Intent:** Office session creation must refuse to add a second live row for a +pair that already has one, so that `AC-OFFICE-TASKS-001.2` holds at runtime and +`ErrOfficeSessionRaceConflict` becomes reachable. + +#### Acceptance criteria + +- **AC-OFFICE-SESSION-IDENTITY-001.1:** When office session creation is asked + to insert a row for a pair, the system shall determine whether a live row + already exists for that pair and shall perform that determination and the + insert within a single database transaction. +- **AC-OFFICE-SESSION-IDENTITY-001.2:** If a live row already exists for the + pair, then the system shall not insert a row and shall return an error + satisfying `errors.Is(err, ErrOfficeSessionRaceConflict)`. +- **AC-OFFICE-SESSION-IDENTITY-001.3:** While every existing row for the pair + is terminal, the system shall insert the new row and shall succeed. Terminal + rows accumulate as retry history and are never an obstacle to creation. +- **AC-OFFICE-SESSION-IDENTITY-001.4:** The system shall classify a row as live + by testing that its state is not one of `COMPLETED`, `FAILED`, `CANCELLED`. + It shall not test membership of an enumerated live list, so that a state + added later is treated identically by this guard and by the reuse rule in + `AC-OFFICE-TASKS-001.3`. +- **AC-OFFICE-SESSION-IDENTITY-001.5:** If the row being inserted has an empty + `agent_profile_id`, then the system shall skip the guard entirely and insert. + Such a row is stored with SQL `NULL` and names no pair. +- **AC-OFFICE-SESSION-IDENTITY-001.6:** When the guard refuses an insert, the + transaction that refuses shall write nothing. It shall not insert the new row, + and it shall leave every existing row for the pair unmodified: no row is + updated, cancelled, merged, or deleted, and no `metadata.acp_session_id` is + moved between rows. This obligation is on the refusing transaction, which + rolls back. It does not constrain what the caller does after the refusal has + returned; that is governed by `AC-OFFICE-SESSION-IDENTITY-001.9` and + REQ-OFFICE-SESSION-IDENTITY-003. +- **AC-OFFICE-SESSION-IDENTITY-001.7:** Where the executor's repository does not + implement the office creator interface and the in-process fallback creation + path is used instead, the system shall apply the same rule as + `AC-OFFICE-SESSION-IDENTITY-001.2` under that path's existing per-task lock, + so the fallback is not a silently weaker guarantee - an equivalence that holds + because exactly one `Executor` exists per process, a second one sharing the + same database being an unshipped topology this document does not defend. That + path holds a + task-scoped list of sessions rather than a SQL result set, so the rule is + applied in application code and in this order: the system shall first restrict + the list to rows whose `agent_profile_id` equals the one being inserted, and + shall then classify those rows using the predicate required by + `AC-OFFICE-SESSION-IDENTITY-001.10`. Applying the state test before the pair + restriction would refuse a legitimate insert whenever any other agent on the + same task holds a live session, so the order is part of the contract. + `AC-OFFICE-SESSION-IDENTITY-001.5` binds this path identically: where the row + being inserted has an empty `agent_profile_id`, the fallback shall skip the + rule and insert, and shall not instead match that empty value against other + rows that also carry none, since no pair is named. +- **AC-OFFICE-SESSION-IDENTITY-001.8:** The system shall apply the guard to any + live row for the pair, including one created by a non-office path such as a + session spawned on an office task that inherited the spawner's profile. + `task_sessions` carries no column distinguishing which path created a row, so + the office side treats the pair as naming one conversation regardless of + origin, and an office wakeup reuses that row rather than adding a second. This + is a deliberate consequence of having no discriminator, not an oversight; it + is the same assumption the pair lookup has always made. +- **AC-OFFICE-SESSION-IDENTITY-001.9:** After a refusal has returned, the + recovery required by `AC-OFFICE-SESSION-IDENTITY-003.2` shall be permitted to + write to the one row it reuses, and this capability shall not suppress those + writes. Two are pre-existing behaviour of the reuse path and shall be + preserved: transitioning a reused row from `IDLE` to `RUNNING`, and rebinding + that row's execution profile, which includes clearing its + `metadata.acp_session_id` so that a fresh agent connection is established. The + permission extends no further: the writes are confined to the single row the + recovery returns, no other row for the pair is touched, no + `metadata.acp_session_id` is copied or moved from one row to another, and no + row is deleted or merged on any path in this document. Their relative order is + deliberately unconstrained: the two writes touch disjoint fields and leave the + same observable end state either way. Their convergence under concurrency is + scoped. Two recoveries reaching the same row with the *same* execution profile + may each perform both writes without detecting the other, and the row settles + in the same state. Two carrying *different* execution profiles do not + converge; that case is reachable, and for it the rule is last-writer-wins on + the stored row, with each caller's returned in-memory session reflecting its + own write and not guaranteed to match what is stored. That is pre-existing + reuse behaviour: this capability shall preserve it unchanged and shall add no + third write, no new ordering, and no new coordination point. Making the + divergent case converge is out of scope; the system design records why it is + reachable and what closing it would cost. +- **AC-OFFICE-SESSION-IDENTITY-001.10:** The terminal classification required by + `AC-OFFICE-SESSION-IDENTITY-001.4` shall have exactly one implementation in + application code, reachable from every enforcement site that needs it, and the + system shall not introduce a second one. It shall not reuse + `models.IsResumableSessionState` for this purpose: that function answers a + different question and classifies `CREATED` as non-resumable, whereas this + contract classifies `CREATED` as live. The guard's SQL predicate expresses the + same set in SQL, and the two shall be kept in correspondence. + +### REQ-OFFICE-SESSION-IDENTITY-002: Deterministic, live-preferring lookup + +**Intent:** The office lookup for a pair must return the row a caller can +actually use, and must return the same row for the same data every time. +Without this, REQ-OFFICE-SESSION-IDENTITY-001 converts an existing duplicate +into a permanently unstartable task. + +#### Acceptance criteria + +- **AC-OFFICE-SESSION-IDENTITY-002.1:** When more than one row exists for a + pair, the system shall return a live row in preference to a terminal row, + regardless of `started_at`. +- **AC-OFFICE-SESSION-IDENTITY-002.2:** Among rows of equal liveness, the + system shall return the row with the greatest `started_at`. +- **AC-OFFICE-SESSION-IDENTITY-002.3:** If two or more candidate rows tie on + both liveness and `started_at`, then the system shall break the tie by + `task_sessions.id` descending. This column is named to make selection total + and reproducible; it carries no meaning and no recency. No current data + exhibits such a tie (measured: zero tied groups). +- **AC-OFFICE-SESSION-IDENTITY-002.4:** While no row exists for the pair, the + lookup shall return no session and no error. If either identifier is empty, + then the lookup shall likewise return no session and no error rather than + querying, since no pair is named. +- **AC-OFFICE-SESSION-IDENTITY-002.5:** The ordering in + `AC-OFFICE-SESSION-IDENTITY-002.1` through `.3` shall be expressed so that it + evaluates identically on SQLite and PostgreSQL. + +### REQ-OFFICE-SESSION-IDENTITY-003: Convergent find-or-create under concurrency + +**Intent:** Concurrent office wakeups for one pair must converge on one live +session and every caller must receive it. A caller that loses the race must not +receive an error. + +#### Acceptance criteria + +- **AC-OFFICE-SESSION-IDENTITY-003.1:** When N callers concurrently request an + office session for the same pair and no live row exists, the system shall + create exactly one row and shall return that same row to all N callers. +- **AC-OFFICE-SESSION-IDENTITY-003.2:** When a caller's insert is refused under + `AC-OFFICE-SESSION-IDENTITY-001.2`, the system shall re-read the pair, apply + the existing reuse rule, and return the resulting session with its + "was created by this call" indicator false. +- **AC-OFFICE-SESSION-IDENTITY-003.3:** If the re-read in + `AC-OFFICE-SESSION-IDENTITY-003.2` yields no usable session because the + blocking row reached a terminal state between the guard and the re-read, then + the system shall retry the create-and-recover sequence at most once more. If + that second attempt is itself refused under + `AC-OFFICE-SESSION-IDENTITY-001.2` and its own re-read again yields no usable + session, then the system shall return `ErrOfficeSessionRaceConflict`. The + system shall not retry unboundedly: at most two create attempts are made per + call, whatever the outcome. +- **AC-OFFICE-SESSION-IDENTITY-003.4:** The system shall never return a nil + session together with a nil error from the office find-or-create. +- **AC-OFFICE-SESSION-IDENTITY-003.5:** When a caller reuses an existing row + rather than creating one, the system shall not publish a second `CREATED` + lifecycle event for that row. +- **AC-OFFICE-SESSION-IDENTITY-003.6:** Given a pair that already holds more + than one live row from data written before this capability, when an office + session is requested for that pair, the system shall return one of those rows + and shall not fail. Convergence on legacy duplicates is achieved by selection + under REQ-OFFICE-SESSION-IDENTITY-002, not by modifying data. +- **AC-OFFICE-SESSION-IDENTITY-003.7:** Only a refusal under + `AC-OFFICE-SESSION-IDENTITY-001.2` shall be treated as a conflict. If a create + attempt fails for any other reason, or if the re-read in + `AC-OFFICE-SESSION-IDENTITY-003.2` fails, then the system shall return that + failure to the caller as itself, wrapped so that the originating error remains + inspectable with `errors.Is`. Such a failure shall not drive the retry in + `AC-OFFICE-SESSION-IDENTITY-003.3` and shall not be reported as + `ErrOfficeSessionRaceConflict`. The re-read's error shall not be discarded. + The sentinel's meaning to a caller is "the pair is already served, re-read and + reuse"; a database, transport, or transaction fault is not that, and reporting + one as the other converts an operational failure into an outcome the contract + tells callers to treat as benign. + +### REQ-OFFICE-SESSION-IDENTITY-004: Non-office session creation is unconstrained + +**Intent:** Three shipped flows deliberately hold two live rows for one pair. +This requirement exists so that a future implementation cannot satisfy +REQ-OFFICE-SESSION-IDENTITY-001 by a mechanism that also reaches them. It is +the guard against repeating the regression that produced this card. + +#### Acceptance criteria + +- **AC-OFFICE-SESSION-IDENTITY-004.1:** The system shall apply the constraint in + REQ-OFFICE-SESSION-IDENTITY-001 only to rows created through the office + creation path, and shall not apply it through any table-level database + constraint on `task_sessions`. +- **AC-OFFICE-SESSION-IDENTITY-004.2:** When a kanban task is relaunched for an + agent profile whose existing session for that task is still live, the system + shall create a second, distinct live session and shall succeed. +- **AC-OFFICE-SESSION-IDENTITY-004.3:** When a workflow replacement session is + prepared for an existing environment, the system shall create the replacement + while the current session for the same pair is still live, and shall succeed. +- **AC-OFFICE-SESSION-IDENTITY-004.4:** When a session is spawned without an + explicit agent profile and inherits the spawner's profile on the same task, + the system shall deliver a session and shall succeed, this being the + documented default of that tool. Where the task is not office-owned the + delivered session shall be a newly created one, as today. Where it is + office-owned and carries an assignee, the spawn converges on that task's one + persistent office session by pre-existing routing, and the criterion is met by + that session being delivered without error; there this capability shall neither + refuse the convergence nor add a second live row to make the outcome a + creation. +- **AC-OFFICE-SESSION-IDENTITY-004.5:** The system shall introduce no new column + on `task_sessions`, no new index on `task_sessions`, and no backfill or + repair of existing `task_sessions` rows. + +## Out of scope + +- **Cross-process concurrency on one database file.** The guarantee in + REQ-OFFICE-SESSION-IDENTITY-001 rests on serialization within a single + process: the single-connection SQLite writer pool, and the task row lock + already taken on PostgreSQL. Two kandev processes writing one SQLite file + concurrently is not a supported topology and is not defended against here. A + table-level constraint would cover it; that is rejected for the reasons in + REQ-OFFICE-SESSION-IDENTITY-004, and the upgrade path, should the topology + ever be supported, is a dedicated office-slot discriminator column plus a + partial unique index over it. Named so the trade is on record, not an + omission. +- **Repairing the existing duplicate pairs.** Measured on the shipped database + on 2026-08-31: 119 rows across 48 pairs, of which 101 are terminal retry + history that is legitimate and must be preserved. Exactly one pair holds more + than one live row; `AC-OFFICE-SESSION-IDENTITY-003.6` makes it usable without + touching it. The census drifts upward while the system runs, so no acceptance + criterion depends on the count, and `AC-OFFICE-SESSION-IDENTITY-004.5` forbids + a backfill at any count. No migration is authorised by this document. +- **Making concurrent recoveries with different execution profiles converge on a + defined winner.** `AC-OFFICE-SESSION-IDENTITY-001.9` records the semantics that + hold - last-writer-wins, each caller returning its own write - instead of + asserting a convergence that does not, so nothing is left to invent. Both ways + of closing the window change shipped reuse behaviour and no criterion here + depends on either; the system design records them, and records that the + bounded retry widens this window in frequency and not in kind. Named so the + trade is on record and not an omission. +- **The stale DDL in `system-design/tasks-01.md`.** It names a column that does + not exist and is superseded by this contract. Correcting or deleting that + block is editorial cleanup for a separate change, not part of this capability. +- **`ErrOfficeSessionRaceConflict`'s driver-error classification.** With + enforcement expressed as an explicit in-transaction guard, the sentinel is + returned directly and no `UNIQUE constraint` message need be parsed. Whether + the parent card's dual-dialect classifier is retained, and whether the + `strings.Contains` match that could never have matched SQLite's error text is + removed, is that card's disposition. This contract requires only that the + sentinel is reachable and testable with `errors.Is`. +- **Any user-visible surface.** No new copy, no i18n work, no UI change. + +## System design + +The technical design is [part 1](../system-design/task-session-identity-01.md). diff --git a/docs/specs/office/system-design/task-session-identity-01.md b/docs/specs/office/system-design/task-session-identity-01.md new file mode 100644 index 0000000000..6aa9adec00 --- /dev/null +++ b/docs/specs/office/system-design/task-session-identity-01.md @@ -0,0 +1,526 @@ +--- +status: draft +system: office +requirements: + - REQ-OFFICE-SESSION-IDENTITY-001 + - REQ-OFFICE-SESSION-IDENTITY-002 + - REQ-OFFICE-SESSION-IDENTITY-003 + - REQ-OFFICE-SESSION-IDENTITY-004 +--- + +# Office Task Session Identity System Design + +## Purpose and boundaries + +This design places enforcement of "one live office session per pair" on the +office creation path and makes the office lookup deterministic. It changes two +functions in the task repository and one recovery branch in the executor, and +adds no column, no index, and no migration. + +The boundary that matters: `task_sessions` has six insert entry points, all +funnelling into one private `createTaskSession`. Exactly one, +`CreateOfficeTaskSession`, is the office path, and its sole production caller is +the executor's `persistOfficeSession`; the other five carry kanban relaunch, +workflow replacement, and spawned sessions. Enforcement attaches to the office +entry point, above the shared private insert, which makes +`AC-OFFICE-SESSION-IDENTITY-004.1` through `.4` true by construction rather than +by predicate tuning. + +## Baseline + +This design is stated against commit `a96f0d51b`, which is this branch's `HEAD` +and also its merge base with `origin/main`. Every "currently", "already", and +"no change to" below describes that tree. `origin/main` has since moved ahead of +it; the merge base, not the remote tip, is the reference. This pin is exact, and +it is the only part of this section that does not age. + +The working tree is **not** that tree. It carries uncommitted work from card +`6def9e4c`'s abandoned attempt, and `6def9e4c` kept acting on it after this +design was written, so no snapshot here describes it. **Build shall re-derive +the tree state at the start of implementation** with `git status --porcelain` +and decide with "The go/no-go test" below rather than by matching a list. Expect +files on this design's own "Files expected to change" list to already carry +another card's edits; expect `base_schema.go` not to, and stop if it does. + +### The go/no-go test + +The question is not which files are dirty - dirty files here are expected, and +this design modifies three of them itself. It is whether the tree contains +anything violating `AC-OFFICE-SESSION-IDENTITY-004.1`. The two stop clauses are +independent and **either one alone stops the card**, so a partial match is a +decision, not an ambiguity; proceeding requires that neither has matched. Where +Build cannot determine whether an artifact is reachable from `initSchema`, it +counts as present: the default on ambiguity is to stop. + +- **Neither stop clause matches: proceed**, however many files are dirty and + even if none are. This is the expected state. Build implements on top of the + existing edits, reconciling them to this design wherever they conflict. +- **Any schema artifact present: stop and raise a blocker against `6def9e4c`.** + Concretely: a modification to `base_schema.go`, an index-creating or + row-healing function reachable from `initSchema`, or a migration file. Any one + violates `004.1` on arrival and would create the index on the next boot; one is + enough, and the absence of the others is not mitigation. +- **A tripwire test present in non-baseline form: stop and raise a blocker.** + Specifically + `TestCreateTaskSessionWithInitialRuntimeSeedConsumesOnceAcrossConcurrentAndReplacementSessions` + asserting one conflict and one row rather than two successful creates and two + rows. This design requires it in baseline form (see "What must be proven"); a + rewritten one is restored by the card that rewrote it. + +No clause here licenses reverting another card's uncommitted work unasked. +Raising the blocker is the action; editing their tree is not. + +### The residual `office_task_session_uniqueness*` files + +`office_task_session_uniqueness.go` and its test are staged in the index and +carry the previous attempt's dual-dialect UNIQUE-violation classifier. They are +**not** a schema artifact, so on their own they neither trip the go/no-go test +nor block this card. This design does not need them: with an in-transaction +guard the sentinel is returned directly and no driver error string is parsed. + +**The classifier is not inert, though.** `isOfficeTaskSessionUniqueViolation` +has two live call sites in `session.go`: the INSERT branch in +`createTaskSession` (`:949`), and an UPDATE branch in +`updateTaskSessionWithStateGuard` (`:1448`) that `6def9e4c` added and the +baseline does not have, reachable from Change 3's own recovery through +`rebindOfficeSessionExecutionProfile`. Both are dead only because no index +exists, and both carry a comment asserting "no index enforces this pair yet" +pointing at the doc comment this design requires Build to rewrite to say the +opposite. **Build shall delete both branches**, leaving plain `return err` and +`return false, err`; `session.go` is already on the change list. Keeping them +would give `ErrOfficeSessionRaceConflict` three producers, one an UPDATE inside +the recovery path, so `AC-OFFICE-SESSION-IDENTITY-003.7` would hold only by the +accident that no index exists - and would leave an enforcement branch in the +shared private insert `## Change 1` requires to stay enforcement-free. + +The two **files** stay untouched: whether they are kept, reworked, or removed is +`6def9e4c`'s disposition, as `## Out of scope` in the requirements records. +Deleting the call sites leaves the function referenced only by its own test, +which is intended; if a linter objects to that, or if the files are still +present at PR time, Build says so in the PR description rather than restoring a +call site. + +## Root cause + +The office find-or-create reads and writes through different connection pools. +`GetTaskSessionByTaskAndAgent` executes against `r.ro`, the read-only pool, +opened with four connections. `CreateOfficeTaskSession` executes against `r.db`, +the writer, opened with `SetMaxOpenConns(1)`. `EnsureSessionForAgentWithCreation` +therefore reads on a reader connection, decides, then inserts on the writer, and +nothing holds across the two: four concurrent callers each read "no session for +this pair", each decide to create, and each insert succeeds, because the writer +transaction checks only `SELECT COUNT(*) FROM task_sessions WHERE task_id = ?` +for the initial-origin marker and never checks the pair. The pool is four wide +and the observed incident produced four rows in five seconds. + +The fix is to move the pair check inside the writer transaction, where mutual +exclusion already exists. + +## Why the transaction already suffices + +`CreateOfficeTaskSession` opens a transaction and, on PostgreSQL, takes +`SELECT id FROM tasks WHERE id = ? FOR UPDATE` first, to keep the initial-origin +marker correct. That lock gives exactly the property the guard needs. + +- **PostgreSQL.** Two concurrent calls for one task serialize on the task row + lock. The second blocks until the first commits, then, under the default + READ COMMITTED isolation, its guard query observes the committed row. +- **SQLite.** The writer pool is one connection, held by a transaction for its + whole lifetime, so a second office insert cannot begin until the first commits + or rolls back. The driver's deferred-transaction subtlety does not apply: + exclusion comes from the pool, not the transaction's lock mode. + +`AC-OFFICE-SESSION-IDENTITY-001.1` is therefore satisfied by one query inside an +existing transaction, with no new locking primitive. The limit of the argument is +in the requirements' "Out of scope": it holds within one process. + +## Change 1 - the creation guard + +Inside `CreateOfficeTaskSession`'s transaction, after the task lock and before +the insert, query whether the pair already holds a live row, and refuse if so. + +The predicate is written as the complement of the terminal set, satisfying +`AC-OFFICE-SESSION-IDENTITY-001.4`: + +```sql +SELECT COUNT(*) FROM task_sessions + WHERE task_id = ? + AND agent_profile_id = ? + AND state NOT IN ('COMPLETED', 'FAILED', 'CANCELLED') +``` + +A non-zero count returns `ErrOfficeSessionRaceConflict` and rolls the +transaction back, so nothing is written and no existing row is touched +(`AC-OFFICE-SESSION-IDENTITY-001.2`, `.6`). A zero count proceeds to the insert, +keeping the terminal-retry path working (`.3`). An empty `agent_profile_id` is +stored as SQL `NULL` by `createTaskSession`; `NULL = ?` never matches and such a +row names no pair, so the guard is skipped for it outright rather than relied +upon to match nothing (`AC-OFFICE-SESSION-IDENTITY-001.5`). + +The guard belongs in `CreateOfficeTaskSession`, not `createTaskSession`: the +private helper is shared with the five non-office entry points, and putting the +guard there would reproduce exactly the regression this card exists to avoid. + +The executor's `persistOfficeSessionFallback`, used when the repository does not +implement the office creator interface, already serializes on a per-task mutex +and already lists the task's sessions via `ListTaskSessions(ctx, taskID)`. It +applies the same live-pair rule under that lock to satisfy +`AC-OFFICE-SESSION-IDENTITY-001.7`. That list is task-scoped, not pair-scoped, +so the rule is applied in Go, and the order matters: restrict to rows whose +`agent_profile_id` equals the one being inserted, then test those rows for +terminality. Testing state first would refuse a legitimate insert whenever any +other agent on the same task happened to be live. + +The predicate is the executor package's existing `isStopTerminalSessionState` +(`executor_execute.go`), which already tests exactly `COMPLETED`, `FAILED` and +`CANCELLED`, and which lives in the same package as the fallback. It is the +single Go implementation `AC-OFFICE-SESSION-IDENTITY-001.10` requires; no second +one is written for this path. `models.IsResumableSessionState` is not usable +here - it excludes `CREATED`, which this contract classifies as live, and its own +doc comment says it answers a different question. The abandoned attempt's +enumerated live list is not usable either: `AC-OFFICE-SESSION-IDENTITY-001.4` +forbids an allow-list. + +## Change 2 - the lookup ordering + +`GetTaskSessionByTaskAndAgent` currently orders `started_at DESC LIMIT 1` with +no state preference. Change 1 alone would therefore permanently fail any pair +whose newest row is terminal while an older row is live: the caller reads +terminal, decides to create, the guard refuses, the caller re-reads, gets the +same terminal row, and fails again on every wakeup forever. That state exists in +the shipped database today - task `62201cdb`, agent profile `3628b4c9-3`, has a +`COMPLETED` newest row above three live rows - so Change 1 without Change 2 would +make that task permanently unstartable. + +The ordering becomes live-first, then recency, then a total tiebreak: + +```sql +ORDER BY CASE WHEN ts.state IN ('COMPLETED', 'FAILED', 'CANCELLED') + THEN 1 ELSE 0 END, + ts.started_at DESC, + ts.id DESC +LIMIT 1 +``` + +The `CASE` expression, `DESC` and `LIMIT` are standard on both dialects, +satisfying `AC-OFFICE-SESSION-IDENTITY-002.5`. The `id` tiebreak makes selection +total and reproducible and carries no recency (`002.3`). No tied group exists in +current data. + +The three other callers - the office session terminator, the orchestrator's +session-ensure path, and the office test harness - all want "the current session +for this pair", and preferring a live row over a dead one is strictly more +correct for each, so no per-caller variant is needed. A pair carrying legacy +duplicate live rows now resolves to one of them and is usable, which is how +`AC-OFFICE-SESSION-IDENTITY-003.6` is met without a backfill: the extra rows stay +in place and simply stop being selected. + +## Change 3 - bounded recovery + +The executor's existing conflict branch re-reads the pair and reuses what it +finds. Three gaps remain against REQ-OFFICE-SESSION-IDENTITY-003. + +If the re-read yields nothing usable - possible only in the narrow window where +the blocking row goes terminal between the guard and the re-read - the branch +currently falls through and returns a nil session with the conflict error. That +becomes a bounded retry of the create-and-recover sequence, at most one +additional attempt, then the conflict error (`AC-OFFICE-SESSION-IDENTITY-003.3`). +It must also never return a nil session with a nil error (`003.4`), and reuse +must continue not to publish a second `CREATED` lifecycle event, which the +existing "was created by this call" return value already carries (`003.5`). + +The third gap is that the branch discards an error. It reads +`raced, lookupErr := e.repo.GetTaskSessionByTaskAndAgent(...)` then tests +`if lookupErr == nil && raced != nil`, so a genuine lookup failure is +indistinguishable from "no row for the pair" and the call returns the conflict +sentinel for what may be a disk, permission, or transport fault. +`AC-OFFICE-SESSION-IDENTITY-003.7` closes this: only a refusal under `001.2` is +a conflict and only a conflict drives the bounded retry. Any other create +failure, and any re-read failure, is returned as itself, wrapped so `errors.Is` +still reaches the original. The bound is therefore two create attempts per call +at most, and a non-conflict failure ends the call at the attempt that produced it +rather than consuming the retry. + +With Change 2 in place the common recovery is deterministic: the guard fires only +when a live row exists and the lookup now prefers live rows, so the re-read finds +the row that caused the conflict unless it has since gone terminal. + +### Divergent execution profiles during recovery + +`AC-OFFICE-SESSION-IDENTITY-001.9` scopes its convergence claim to recoveries +carrying the same execution profile and names last-writer-wins for the divergent +case. The mechanism, recorded here so the criterion need not carry it: the pair +is keyed on the stable agent identity (`agentInstanceID`, from +`dbTask.AssigneeAgentProfileID`) while the execution profile is a *separate* +per-call parameter `createStartSession` takes from the launch request, so two +wakeups can legitimately resolve different profiles for one pair. +`rebindOfficeSessionExecutionProfile` (`executor_office.go:100-165`) then guards +its update on the row's **state**, via +`UpdateTaskSessionIfCurrentStateRemovingMetadataKeys`, not on its execution +profile - so two concurrent rebinds sharing an expected state both succeed, the +last wins, and each caller returns the in-memory session it built rather than a +re-read. + +This is shipped behaviour and Change 3 must not alter it. Closing the window +would mean guarding the rebind on the observed execution profile as well as the +state, or serializing recovery per pair; both change shipped reuse semantics and +no criterion here depends on either, which is why the requirements name it out of +scope rather than leaving it unstated. Change 3 changes only frequency: a +conflict-driven second attempt re-enters the same recovery, so the two writes +`AC-OFFICE-SESSION-IDENTITY-001.9` permits can occur twice. It adds no new class +of write, and the two-attempt cap bounds the repetition. + +## Files expected to change + +| File | Why | +|---|---| +| `internal/task/repository/sqlite/session.go` | the guard in `CreateOfficeTaskSession`; the ordering in `GetTaskSessionByTaskAndAgent`, plus that function's comment, which describes a `uniq_office_task_session` index that does not exist; deleting the two residual classifier branches named under "Baseline" | +| `internal/orchestrator/executor/executor_office.go` | bounded recovery in `EnsureSessionForAgentWithCreation`; the live-pair rule in `persistOfficeSessionFallback` | +| `internal/task/repository/sqlite/errors.go` | the sentinel's doc comment, which must describe an in-transaction guard rather than an index | + +No change to `base_schema.go`, `base_migrations.go`, or any rebuild block; +`AC-OFFICE-SESSION-IDENTITY-004.5` forbids it. That describes what this card +adds to the baseline above, not the working tree as found. + +## What must be proven + +**Test doubles: what is permitted.** Several bullets below need an interleaving +or a failure that data setup alone cannot produce, and there is nothing to +inherit: no backend test references `EnsureSessionForAgentWithCreation`, and no +double in `internal/orchestrator/` implements `CreateOfficeTaskSession`. +Build **may** author a double that wraps the real repository and controls only +*timing and failure* - blocking a call until another goroutine has passed a +chosen point, counting calls, or returning a chosen error from one named method +on a chosen attempt. It **may not** replace the behaviour under test: guard +decision, ordering, terminal classification and recovery all run as production +runs them, and the double implements `officeTaskSessionCreator` wherever a +bullet says the transactional path is proven. Holding a loser until the winner +has committed is permitted; returning the sentinel without consulting the guard +is not, and neither is stubbing `GetTaskSessionByTaskAndAgent` to return a row +it did not read. "Must not be satisfied by an injected conflict" below bans +faking the *refusal*, not this seam. + +- A real concurrent office create for one pair yields exactly one row, with the + sentinel raised unfaked (`AC-OFFICE-SESSION-IDENTITY-003.1`, `.2`). **This test + must be authored; no existing test has its shape.** N goroutines, N >= 2, call + `CreateOfficeTaskSession` on one task with an *identical* `agent_profile_id`; + exactly one returns nil, every other returns an error satisfying + `errors.Is(err, ErrOfficeSessionRaceConflict)`, and a subsequent list shows + exactly one row for the pair. It must not be satisfied by an injected conflict. + + Do **not** model it on + `TestCreateOfficeTaskSessionMarksOnlyTheFirstConcurrentSessionAsOrigin`, the + only existing office concurrency test: it races `agent-1` against `agent-2` - + two *different* pairs on one task - and asserts **both** creates succeed, so + copying its shape produces a test that never reaches the guard. That is also + why the guard must key on the pair and never on the task alone. It must pass + unmodified, and so must its PostgreSQL twin + `TestPostgresCreateOfficeTaskSessionMarksOnlyTheFirstConcurrentSessionAsOrigin`. +- The *convergence* half of `AC-OFFICE-SESSION-IDENTITY-003.1` is proven at the + layer that delivers it. The bullet above races `CreateOfficeTaskSession`, + where every loser receives an error; that proves "exactly one row" but cannot + observe "and shall return that same row to all N callers", which lives in the + executor's recovery (`executor_office.go:84-97`). **This test must also be + authored.** N goroutines, N >= 4 so the race is at least as wide as the reader + pool that produced the incident, call `EnsureSessionForAgentWithCreation` for + one pair on a task with no session for it; every call returns a nil error, + every returned session carries the same `ID`, exactly one reports + `created == true`, and a subsequent list shows exactly one row for the pair. + + Those four assertions are **not sufficient alone and must not ship alone.** + They all hold on an interleaving in which every goroutine after the first finds + the row on its *initial* lookup and returns through ordinary reuse, so the + recovery branch never runs and the test is green whether or not Change 3 works + - the one thing it exists to distinguish. The test therefore uses the seam + above to force at least one caller past its initial lookup into + `CreateOfficeTaskSession`, and asserts through the seam's call counting that at + least one such call was refused under `001.2` and that its caller still + received a session. A run in which none reaches the recovery must **fail**. + + The nil-error assertion is the load-bearing one of the four: + REQ-OFFICE-SESSION-IDENTITY-003's Intent is that a caller which loses the race + must not receive an error, while the repository-level bullet above + deliberately asserts the opposite at its own layer. It must not be satisfied by + an injected conflict, and it must run against a repository implementing + `officeTaskSessionCreator` - not incidental, since `persistOfficeSession` + branches on that, so a double that does not implement it exercises the + in-process fallback instead of the transactional guard and would pass while + proving nothing about the path production takes. +- A terminal-only pair still accepts a new row (`AC-OFFICE-SESSION-IDENTITY-001.3`). +- A pair shaped like `62201cdb` - terminal newest, live older - resolves to the + live row and does not fail (`AC-OFFICE-SESSION-IDENTITY-002.1`, `003.6`). This + is the regression test for the livelock Change 2 exists to prevent. +- A refusal writes nothing: the pair's rows are byte-identical before and after + a refused create, and no row is added (`AC-OFFICE-SESSION-IDENTITY-001.6`). + Separately, the recovery that follows a refusal may still write to the row it + reuses - an `IDLE` row is observed to reach `RUNNING`, and a rebind clears that + row's `metadata.acp_session_id` - while every other row for the pair is untouched + (`AC-OFFICE-SESSION-IDENTITY-001.9`). Proving only the first would lock in a + suppression of shipped behaviour. The second half is not reachable by seeding - + a pre-seeded `IDLE` row is found by the first lookup and reused - so force the + refusal with the seam above. +- A non-conflict failure is not laundered into the sentinel: when the create + fails for a reason other than the guard, or when the recovery re-read itself + fails, the caller receives that error and `errors.Is(err, + ErrOfficeSessionRaceConflict)` is false, and no retry is spent + (`AC-OFFICE-SESSION-IDENTITY-003.7`). Neither failure is reachable by data + setup; both come from the seam above. +- The three non-office flows still create a second live row for a shared pair. + Each needs its own named proof; the previous attempt was caught only because + the first of the three happened to have one. All three ids below are + `AC-OFFICE-SESSION-IDENTITY-*`. + - **Kanban relaunch (`004.2`).** + `TestCreateStartSession_KanbanRunnerCreatesDistinctSession` + (`internal/orchestrator/task_operations_test.go`) exists, must pass + unmodified, and is the tripwire that caught the previous attempt. + - **Workflow replacement (`004.3`).** + `TestCreateTaskSessionWithInitialRuntimeSeedConsumesOnceAcrossConcurrentAndReplacementSessions` + (`session_test.go`) is the existing repository-level proof and must pass in + its **baseline** form. Replacement reaches the shared insert through + `prepareSession` -> `createPreparedSession` -> + `CreateTaskSessionWithInitialRuntimeSeed`, and that test races two sessions + carrying one *identical* `agent_profile_id` on one task through exactly that + entry point, asserting both succeed and two rows result - a direct tripwire + for enforcement leaking off the office path. The abandoned attempt rewrote it + (see "Baseline"): a version asserting one conflict is the regression, not the + target. `PrepareSessionForExistingEnvironment`'s own two tests are negative - + they reject an unusable workspace before any session is created - and prove + nothing here. + - **Spawn profile inheritance (`004.4`).** + `TestHandleSpawnSession_SameTask_DefaultsToSenderProfile` + (`internal/mcp/handlers/spawn_session_test.go`) locks the inheritance + default against a live `RUNNING` spawner session on the same task and must + pass unmodified, but it proves only profile *resolution*: it asserts against + a mock orchestrator's recorded launch calls and writes no `task_sessions` + row, so it would still pass if the resulting insert were refused. The missing + half - that the insert is not refused - is **already covered; author nothing + new.** The entry point was traced, not assumed: `handleSpawnSession` calls + `LaunchSession` with `IntentStart` + (`internal/mcp/handlers/spawn_session.go:46-51`), reaching `launchStart` -> + `startTask` -> `prepareSessionForStart` -> `createStartSession` + (`internal/orchestrator/task_operations.go:1399`, `:1591`) - the entry point + kanban relaunch uses. + `TestCreateStartSession_KanbanRunnerCreatesDistinctSession`, named above for + `.2`, seeds a live `RUNNING` session for the pair and asserts + `createStartSession` still returns a *distinct* session id, so it proves for + spawn exactly what it proves for relaunch. Build cites it for both criteria; + what must not happen is recording it as proven by the resolution test alone. + + Do **not** author an office-task version. Where `IsFromOffice` is true and + the task carries an assignee, `createStartSession` routes to + `EnsureSessionForAgentWithCreation` keyed on the **task assignee** rather + than the profile the spawn call resolved, so it converges on the one + persistent office row and reports `created == false`. + `TestCreateStartSession_OfficeRunnerReusesPersistentSession` + (`task_operations_test.go:185`) asserts that reuse and must also pass + unmodified, so a test asserting a second distinct row on an office task would + contradict one this design requires to pass. That convergence is pre-existing + routing this capability does not change, and `.4` is satisfied on that shape + by the session being delivered without error, not by a row being added. +- The guard's predicate is the complement of the terminal set, demonstrated by a + row in a state outside both sets being treated as live + (`AC-OFFICE-SESSION-IDENTITY-001.4`). The fallback also restricts to the pair + before testing state: a task carrying a live session for a *different* agent + profile still accepts an office create for its own pair (`001.7`). +- **The empty-`agent_profile_id` bypass inserts, on both paths** + (`AC-OFFICE-SESSION-IDENTITY-001.5`, and the final sentence of + `AC-OFFICE-SESSION-IDENTITY-001.7`). Two rows carrying an empty + `agent_profile_id` are created in sequence on one task and **both** succeed; + the guard is not consulted for either. Prove it separately at the repository + and at the fallback: the two are safe for different reasons and only one is + safe by accident. In SQL the branch cannot misfire - `createTaskSession` stores + an empty string as `NULL`, and `agent_profile_id = ?` never matches `NULL`. The + fallback works on Go structs, where that row reads back as + `AgentProfileID == ""`, so the natural expression of + `AC-OFFICE-SESSION-IDENTITY-001.7`'s pair restriction - + `s.AgentProfileID == session.AgentProfileID` - makes `""` match `""`, + classifies a second empty-profile row as a live row for the pair, and + **refuses a legitimate insert**, exactly what that criterion's final sentence + forbids. The fallback case is load-bearing and must run against a repository + that does *not* implement `officeTaskSessionCreator`; the natural wrong + implementation passes the repository case and fails only here. +- **A live row for the pair created by a non-office path is reused, not + duplicated** (`AC-OFFICE-SESSION-IDENTITY-001.8`). Two assertions at two layers; + do not fold them into one test, because neither layer can observe both halves. + **Repository:** seed a live row for the pair through a non-office insert entry + point - `CreateTaskSession` is the simplest - then call + `CreateOfficeTaskSession` for that pair and assert it is refused under `001.2` + and adds no row. **Executor:** seed the same shape, then call + `EnsureSessionForAgentWithCreation` and assert a nil error, `created == false`, + a returned session carrying the seeded row's `ID`, and still one row for the + pair. Assert the `ID`, not the whole row: `001.9` permits the recovery to write + to the row it reuses. Do **not** assert a refusal at the executor layer - the + lookup runs first, Change 2 orders the live seeded row to the top of it, and + reuse returns before `createOfficeSession` is called, so no create happens + there to refuse. This makes the guard deliberately origin-blind, the behaviour + most likely to be read as a bug and "fixed" later. +- **The ordering is total, and evaluates identically on both dialects** + (`AC-OFFICE-SESSION-IDENTITY-002.2`, `.3`, `.5`). One pair is seeded so that + each comparison is forced in turn: two live rows differing only in + `started_at` select the greater `started_at`; two live rows sharing one + `started_at` select the greater `id`; and the assertions run against both + SQLite and PostgreSQL, since `.5` is a claim about the expression evaluating + the same way on each and a SQLite-only test cannot observe it. Follow the + existing dual-dialect convention: this bullet's test takes the + `TestPostgres`-prefixed twin shape named above. The + `62201cdb`-shaped bullet proves liveness outranks recency; this one proves the + two tiebreaks beneath it. +- **An absent pair and an empty identifier each return no session and no error** + (`AC-OFFICE-SESSION-IDENTITY-002.4`). Three cases, each asserting a nil + session *and* a nil error: a task with no row for the pair, an empty `task_id`, + and an empty `agent_profile_id`. This locks shipped behaviour Change 2 must not + disturb - the ordering rewrite edits this same function, and turning "no rows" + into an error would break every caller that branches on `session != nil`. +- **The retry is bounded at two create attempts** + (`AC-OFFICE-SESSION-IDENTITY-003.3`). The count is asserted directly: in the + worst case, where every create is refused under + `AC-OFFICE-SESSION-IDENTITY-001.2` and every re-read yields no usable session, + count the create attempts and assert **exactly two**, never three, and that the + call then returns `ErrOfficeSessionRaceConflict` rather than looping. That + worst case - the blocking row live at each guard and terminal before each + re-read - is doubly nested and has no data-setup shape; build it with the seam + above, whose call-counting permission is what makes "exactly two" observable at + all. A bound no test observes is one a later edit removes silently, and the + failure it prevents - unbounded retry against a pair that stays terminal - is + this card's own livelock relocated one layer up. +- **Reuse publishes no second `CREATED` event** + (`AC-OFFICE-SESSION-IDENTITY-003.5`). An office wakeup that reuses an existing + row emits no `CREATED` lifecycle event, while the wakeup that created it did. + Assert on the emitted event, not only on `EnsureSessionForAgentWithCreation`'s + `created` boolean, which is the input to that decision, not an observation. + +**Coverage of the remaining criteria.** Every criterion is named in a bullet +above or accounted for here, so one carrying no proof is a recorded decision +rather than an oversight. `001.1` and `.2` are asserted inside the two +concurrency bullets rather than separately - a guard that failed to refuse would +fail those first. `001.3`, `.4`, `.6` and `.9` each have their own bullet. `002.1` +is proven by the `62201cdb`-shaped bullet. `003.1`, `.2`, `.6` and `.7` have +their own bullets; `003.4` (never a nil session with a nil error) already holds +in the shipped recovery and is a standing invariant of both concurrency tests +rather than a case of its own. `004.2`, `.3` and `.4` are the three tripwires. + +`004.1`, `004.5` and `001.10` are architectural negatives - no column, no index, +no migration, no schema change, and exactly one terminal-classification predicate +rather than a second copy - proven by the diff rather than by a test. `001.10` +sits here deliberately: no behavioural assertion can distinguish +`persistOfficeSessionFallback` calling `isStopTerminalSessionState` from it +carrying an identical inline copy, so like the other two it is a review check +against the `## Baseline` go/no-go test, not a Build-authored test. + +## Risks + +- **The guard is process-local.** Stated and accepted in the requirements' "Out + of scope" - a real reduction against a table constraint, taken because a table + constraint provably breaks shipped flows. +- **Change 2 alters a shared lookup.** Three other callers are affected. The + argument that live-first is strictly more correct for all of them is made + above and should be confirmed per call site during implementation, not assumed. +- **A non-office row can now absorb an office wakeup.** Under + `AC-OFFICE-SESSION-IDENTITY-001.8`, if a spawned session on an office task + inherited the profile and is live, the next office wakeup reuses it instead of + creating an office row; before this change it created a second row. No column + tells the two apart, so this follows from the design rather than being chosen, + and is recorded because behaviour changes. +- **The guard adds a query inside the writer transaction.** An indexed lookup on + `task_id` under a lock the transaction already holds, on a path that runs once + per office session creation, not per turn.