fix(backend): stop office agents from spawning duplicate concurrent sessions (#3253)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-01 17:57:17 +00:00
committed by GitHub
parent 1bb80b7fc9
commit 63f47a692f
15 changed files with 2837 additions and 191 deletions
@@ -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
}
@@ -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)
@@ -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 {
@@ -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{})
@@ -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.
@@ -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)
}
}
@@ -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
@@ -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
@@ -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
@@ -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)
}
@@ -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)
}
}
@@ -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
@@ -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()
@@ -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).
@@ -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.