diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 74aa0f0415..21436581f8 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -150,7 +150,34 @@ func executorNeedsResolvedCredentials(executorType string) bool { // starts detach from request cancellation; resume starts marked by // WithCancellableResumeContext retain cancellation so an explicit stop can // interrupt a startup that is still waiting for ACP readiness. -func (e *Executor) runAgentProcessAsync(ctx context.Context, taskID, sessionID, agentExecutionID string, onSuccess func(context.Context), escalateTaskOnFailure, fromResume bool) { +func (e *Executor) runAgentProcessAsync( + ctx context.Context, + taskID, sessionID, agentExecutionID string, + onSuccess func(context.Context), + escalateTaskOnFailure, fromResume bool, +) { + e.runAgentProcessAsyncWithObservation( + ctx, + taskID, + sessionID, + agentExecutionID, + "", + onSuccess, + escalateTaskOnFailure, + fromResume, + ) +} + +// runAgentProcessAsyncWithObservation starts an agent process after the +// caller has persisted STARTING. The observation stays in this shared seam so +// full launches, existing-workspace starts, and resumes all inspect the same +// durable state immediately before process startup. +func (e *Executor) runAgentProcessAsyncWithObservation( + ctx context.Context, + taskID, sessionID, agentExecutionID, observationSite string, + onSuccess func(context.Context), + escalateTaskOnFailure, fromResume bool, +) { e.auditCeilingBypass(ctx, "runAgentProcessAsync", sessionID, true, zap.String("agent_execution_id", agentExecutionID)) go func() { startParent := context.WithoutCancel(ctx) @@ -162,6 +189,10 @@ func (e *Executor) runAgentProcessAsync(ctx context.Context, taskID, sessionID, startCtx, cancel := context.WithTimeout(startParent, 5*time.Minute) defer cancel() + if observationSite != "" { + e.observeSessionCoresidency(startCtx, observationSite, taskID, sessionID) + } + if err := e.agentManager.StartAgentProcess(startCtx, agentExecutionID); err != nil { if isCancellableResumeContext(ctx) && ctx.Err() != nil { // A cancelled resume owns no failure projection. Use the exact @@ -368,7 +399,7 @@ func (e *Executor) stopStartedExecutionIfSessionTerminal( // startAgentProcessAsync starts the agent subprocess and transitions its session // to RUNNING before reconciling the owning task to IN_PROGRESS on success. func (e *Executor) startAgentProcessAsync(ctx context.Context, taskID, sessionID, agentExecutionID string) { - e.runAgentProcessAsync(ctx, taskID, sessionID, agentExecutionID, func(updCtx context.Context) { + e.runAgentProcessAsyncWithObservation(ctx, taskID, sessionID, agentExecutionID, sessionCoresidencySiteLaunch, func(updCtx context.Context) { if !e.markSessionRunningAfterProcessStart(updCtx, taskID, sessionID) { return } @@ -551,8 +582,31 @@ func (e *Executor) failedSessionStillWorkingOrUnknown(ctx context.Context, taskI return false } -func (e *Executor) hasOtherWorkingSessions(ctx context.Context, taskID, failedSessionID string) bool { +// workingSessionSiblings lists taskID's sessions currently in a working +// runtime state, excluding excludeSessionID. hasOtherWorkingSessions (fails +// open: a read failure is treated as "assume other work is happening") and +// observeSessionCoresidency (fails closed: a read failure is recorded as a +// skip, never as zero siblings) both build on this one read+filter so their +// notion of "working" cannot drift apart from sessionstate.IsWorking. +func (e *Executor) workingSessionSiblings(ctx context.Context, taskID, excludeSessionID string) ([]string, error) { sessions, err := e.repo.ListTaskSessions(ctx, taskID) + if err != nil { + return nil, err + } + siblingIDs := make([]string, 0, len(sessions)) + for _, session := range sessions { + if session == nil || (excludeSessionID != "" && session.ID == excludeSessionID) { + continue + } + if isRuntimeWorkingSessionState(session.State) { + siblingIDs = append(siblingIDs, session.ID) + } + } + return siblingIDs, nil +} + +func (e *Executor) hasOtherWorkingSessions(ctx context.Context, taskID, failedSessionID string) bool { + siblingIDs, err := e.workingSessionSiblings(ctx, taskID, failedSessionID) if err != nil { e.logger.Warn("failed to list task sessions before failed-start REVIEW state reconcile", zap.String("task_id", taskID), @@ -560,24 +614,46 @@ func (e *Executor) hasOtherWorkingSessions(ctx context.Context, taskID, failedSe zap.Error(err)) return true } - for _, session := range sessions { - if session == nil { - continue - } - if failedSessionID != "" && session.ID == failedSessionID { - continue - } - if isRuntimeWorkingSessionState(session.State) { - e.logger.Debug("skipping failed-start task REVIEW state while another session is working", - zap.String("task_id", taskID), - zap.String("failed_session_id", failedSessionID), - zap.String("blocking_session_id", session.ID)) - return true - } + if len(siblingIDs) > 0 { + e.logger.Debug("skipping failed-start task REVIEW state while another session is working", + zap.String("task_id", taskID), + zap.String("failed_session_id", failedSessionID), + zap.String("blocking_session_id", siblingIDs[0])) + return true } return false } +// observeSessionCoresidency records, without changing any admission outcome, +// that site is about to start an agent process for sessionID while another +// session of the same task is already in a working runtime state, sharing +// one task workspace. Kandev permits this by design +// (REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001); this only makes the +// permitted condition observable (REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004). +// A sibling-read failure is recorded as a skip, never as an absence of +// co-residency, and never blocks the caller either way. +func (e *Executor) observeSessionCoresidency(ctx context.Context, site, taskID, sessionID string) { + siblingIDs, err := e.workingSessionSiblings(ctx, taskID, sessionID) + if err != nil { + sessionCoresidencyObservationSkipped(sessionCoresidencySkipReadFailed) + e.logger.Warn("skipped session co-residency observation: sibling session read failed", + zap.String("site", site), + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + if len(siblingIDs) == 0 { + return + } + sessionCoresidencyAdmitted(site) + e.logger.Warn("starting an agent while another session of this task is already working in the shared worktree; Kandev permits concurrent sessions on one task", + zap.String("site", site), + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Strings("sibling_session_ids", siblingIDs)) +} + func isRuntimeWorkingSessionState(state models.TaskSessionState) bool { return sessionstate.IsWorking(state) } @@ -1677,6 +1753,10 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses return nil, err } + if startAgent { + e.observeSessionCoresidency(launchCtx, sessionCoresidencySiteLaunch, task.ID, sessionID) + } + // Fast path: workspace already launched (executors_running row exists). // The selected-environment recovery gate above must run first, including // when this path only starts an agent process on an existing workspace. diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index da53feb6c6..71fb6f4f62 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -322,6 +322,10 @@ type mockRepository struct { 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) + // listTaskSessionsFunc, when non-nil, overrides ListTaskSessions + // entirely — used to simulate a transient sibling-session read failure + // (the session-coresidency observation's fail-closed skip path). + listTaskSessionsFunc func(ctx context.Context, taskID string) ([]*models.TaskSession, error) // Optional hook invoked at the top of UpdateTaskStateIfCurrentIn, before // it reads task state/archived_at. Lets tests simulate the exact TOCTOU // window this CAS closes: an earlier (non-transactional) archived-state @@ -1031,6 +1035,12 @@ func (m *mockRepository) GetTaskSessionByTaskAndAgent(ctx context.Context, taskI return nil, nil } func (m *mockRepository) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) { + m.mu.Lock() + fn := m.listTaskSessionsFunc + m.mu.Unlock() + if fn != nil { + return fn(ctx, taskID) + } m.mu.Lock() defer m.mu.Unlock() sessions := make([]*models.TaskSession, 0) diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 7002daba74..5250b51c82 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume.go @@ -859,6 +859,9 @@ func (e *Executor) resumeSession( if err := e.admitWorktreeRecovery(ctx, task.ID); err != nil { return nil, err } + if startAgent { + e.observeSessionCoresidency(ctx, sessionCoresidencySiteResume, task.ID, session.ID) + } resumeInitialState := session.State previousCredentialSnapshot := captureResumeCredentialSnapshot(session) @@ -2238,7 +2241,7 @@ func (e *Executor) startAgentProcessOnResumeWithTaskPromotion( agentExecutionID string, promoteTask bool, ) { - e.runAgentProcessAsync(ctx, taskID, session.ID, agentExecutionID, func(updCtx context.Context) { + e.runAgentProcessAsyncWithObservation(ctx, taskID, session.ID, agentExecutionID, sessionCoresidencySiteResume, func(updCtx context.Context) { if promoteTask { if updateErr := e.writeTaskInProgressForRuntime(updCtx, taskID, session.ID); updateErr != nil { e.logger.Warn("failed to update task state to IN_PROGRESS after resume start", diff --git a/apps/backend/internal/orchestrator/executor/session_coresidency_metrics.go b/apps/backend/internal/orchestrator/executor/session_coresidency_metrics.go new file mode 100644 index 0000000000..80954241fc --- /dev/null +++ b/apps/backend/internal/orchestrator/executor/session_coresidency_metrics.go @@ -0,0 +1,46 @@ +package executor + +import "expvar" + +// expvar counters that make a permitted condition observable: an agent +// process starting for one session of a task while another session on the +// same task is already in a working runtime state, sharing one task +// workspace (REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004). Kandev +// permits this by design (REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001); +// neither counter changes any admission outcome. Task, session, workspace +// path, branch, and repository identifiers stay in the accompanying zap log +// fields and never become label values, so cardinality is bounded to the +// closed sets below. +var ( + // sessionCoresidencyAdmittedTotalVar counts an agent-process start + // admitted while another session of the same task was already working. + // Labelled by the admitting site. + sessionCoresidencyAdmittedTotalVar = expvar.NewMap("session_coresidency_admitted_total") + + // sessionCoresidencyObservationSkippedTotalVar counts an observation + // abandoned because the sibling-session read failed. The launch or + // resume itself is never affected; this only records that the + // observation's own input could not be read, so a degraded observer is + // distinguishable from a task that genuinely has one live session. + sessionCoresidencyObservationSkippedTotalVar = expvar.NewMap("session_coresidency_observation_skipped_total") +) + +// Admitting sites recorded as the session_coresidency_admitted_total label. +const ( + sessionCoresidencySiteLaunch = "launch" + sessionCoresidencySiteResume = "resume" +) + +// Skip reasons recorded as the session_coresidency_observation_skipped_total +// label. +const ( + sessionCoresidencySkipReadFailed = "sibling_read_failed" +) + +func sessionCoresidencyAdmitted(site string) { + sessionCoresidencyAdmittedTotalVar.Add(site, 1) +} + +func sessionCoresidencyObservationSkipped(reason string) { + sessionCoresidencyObservationSkippedTotalVar.Add(reason, 1) +} diff --git a/apps/backend/internal/orchestrator/executor/session_coresidency_test.go b/apps/backend/internal/orchestrator/executor/session_coresidency_test.go new file mode 100644 index 0000000000..78f77595ac --- /dev/null +++ b/apps/backend/internal/orchestrator/executor/session_coresidency_test.go @@ -0,0 +1,334 @@ +package executor + +import ( + "context" + "errors" + "expvar" + "strings" + "sync" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// newSessionCoresidencyTestExecutor is newTestExecutor plus an in-memory log +// sink and direct access to the mock repository, so tests can seed sibling +// sessions and assert on the AC-004 log fields observeSessionCoresidency +// emits. +func newSessionCoresidencyTestExecutor(t *testing.T) (*Executor, *mockRepository, *observer.ObservedLogs) { + t.Helper() + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + repo := newMockRepository() + exec := NewExecutor(&mockAgentManager{}, repo, log, ExecutorConfig{ + ShellPrefs: &mockShellPrefs{}, + }) + exec.SetCapabilities(&mockCapabilities{}) + return exec, repo, logs +} + +// counterValue reads the current int64 stored under key in an expvar.Map, or +// 0 if the key has never been touched. Tests compare before/after deltas +// rather than absolute values because the map is process-global state shared +// across the whole test binary. +func counterValue(m *expvar.Map, key string) int64 { + v := m.Get(key) + if v == nil { + return 0 + } + iv, ok := v.(*expvar.Int) + if !ok { + return 0 + } + return iv.Value() +} + +// TestObserveSessionCoresidency_SolitarySessionRecordsNothing pins +// AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.2: a session starting alone +// on its task leaves both counters untouched and logs nothing. +func TestObserveSessionCoresidency_SolitarySessionRecordsNothing(t *testing.T) { + exec, repo, logs := newSessionCoresidencyTestExecutor(t) + repo.sessions["session-123"] = &models.TaskSession{ + ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch) + + exec.observeSessionCoresidency(context.Background(), sessionCoresidencySiteLaunch, "task-123", "session-123") + + if logs.Len() != 0 { + t.Fatalf("expected no log entries for a solitary session, got %d: %v", logs.Len(), logs.All()) + } + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch); after != before { + t.Fatalf("admitted[launch] counter changed for a solitary session: before=%d after=%d", before, after) + } +} + +// TestObserveSessionCoresidency_WorkingSiblingLogsWarningAndIncrementsCounter +// pins AC-004.1: a session starting while another session of the same task is +// already in a working runtime state records exactly one structured warning +// naming the site, the starting session, and the sibling session IDs, and +// increments the admitted counter once for the admitting site. +func TestObserveSessionCoresidency_WorkingSiblingLogsWarningAndIncrementsCounter(t *testing.T) { + exec, repo, logs := newSessionCoresidencyTestExecutor(t) + repo.sessions["session-123"] = &models.TaskSession{ + ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + repo.sessions["session-sibling"] = &models.TaskSession{ + ID: "session-sibling", TaskID: "task-123", State: models.TaskSessionStateRunning, + } + before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch) + + exec.observeSessionCoresidency(context.Background(), sessionCoresidencySiteLaunch, "task-123", "session-123") + + warnings := logs.FilterLevelExact(zapcore.WarnLevel).All() + if len(warnings) != 1 { + t.Fatalf("warning entries = %d, want 1; all=%v", len(warnings), logs.All()) + } + fields := warnings[0].ContextMap() + if fields["site"] != sessionCoresidencySiteLaunch { + t.Fatalf("site field = %v, want %q", fields["site"], sessionCoresidencySiteLaunch) + } + if fields["task_id"] != "task-123" || fields["session_id"] != "session-123" { + t.Fatalf("unexpected identity fields: %+v", fields) + } + siblingIDs, ok := fields["sibling_session_ids"].([]interface{}) + if !ok || len(siblingIDs) != 1 || siblingIDs[0] != "session-sibling" { + t.Fatalf("sibling_session_ids field = %#v, want [session-sibling]", fields["sibling_session_ids"]) + } + lowerMsg := strings.ToLower(warnings[0].Message) + if !strings.Contains(lowerMsg, "permit") { + t.Fatalf("warning message = %q, want wording that reads as permitted rather than a failure to prevent", warnings[0].Message) + } + if strings.Contains(lowerMsg, "fail") { + t.Fatalf("warning message = %q, must not read as Kandev failing to prevent something it permits", warnings[0].Message) + } + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch); after != before+1 { + t.Fatalf("admitted[launch] counter = %d, want %d", after, before+1) + } +} + +// TestObserveSessionCoresidency_SiblingReadFailureRecordsSkipNotAbsence pins +// AC-004.3: a failed sibling-session read records a skip with its reason +// rather than reporting an absence of co-residency, and never blocks the +// caller — observeSessionCoresidency itself has no error return. +func TestObserveSessionCoresidency_SiblingReadFailureRecordsSkipNotAbsence(t *testing.T) { + exec, repo, logs := newSessionCoresidencyTestExecutor(t) + readErr := errors.New("transient sibling-session read failure") + repo.listTaskSessionsFunc = func(context.Context, string) ([]*models.TaskSession, error) { + return nil, readErr + } + admittedBefore := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch) + skippedBefore := counterValue(sessionCoresidencyObservationSkippedTotalVar, sessionCoresidencySkipReadFailed) + + exec.observeSessionCoresidency(context.Background(), sessionCoresidencySiteLaunch, "task-123", "session-123") + + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch); after != admittedBefore { + t.Fatalf("admitted[launch] counter changed on a read failure: before=%d after=%d", admittedBefore, after) + } + if after := counterValue(sessionCoresidencyObservationSkippedTotalVar, sessionCoresidencySkipReadFailed); after != skippedBefore+1 { + t.Fatalf("skipped[%s] counter = %d, want %d", sessionCoresidencySkipReadFailed, after, skippedBefore+1) + } + warnings := logs.FilterLevelExact(zapcore.WarnLevel).All() + if len(warnings) != 1 { + t.Fatalf("warning entries = %d, want 1; all=%v", len(warnings), logs.All()) + } + fields := warnings[0].ContextMap() + if fields["site"] != sessionCoresidencySiteLaunch || fields["task_id"] != "task-123" || fields["session_id"] != "session-123" { + t.Fatalf("unexpected identity fields: %+v", fields) + } +} + +// TestLaunchPreparedSession_ObservesWorkingSiblingOnAgentStart pins the +// wiring half of AC-004.1: LaunchPreparedSession calls the observation +// before the agent process starts, for a real launch that otherwise +// succeeds, not just the extracted helper. Ordering is enforced by +// production code structure, not a synchronized assertion here. +func TestLaunchPreparedSession_ObservesWorkingSiblingOnAgentStart(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-123"] = &models.TaskSession{ + ID: "session-123", + TaskID: "task-123", + AgentProfileID: "profile-123", + State: models.TaskSessionStateCreated, + } + repo.sessions["session-sibling"] = &models.TaskSession{ + ID: "session-sibling", TaskID: "task-123", State: models.TaskSessionStateRunning, + } + + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + agentManager := &mockAgentManager{ + launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { + return &LaunchAgentResponse{ + AgentExecutionID: "exec-123", + ContainerID: "container-123", + }, nil + }, + } + executor := NewExecutor(agentManager, repo, log, ExecutorConfig{ShellPrefs: &mockShellPrefs{}}) + executor.SetCapabilities(&mockCapabilities{}) + + task := &v1.Task{ + ID: "task-123", + WorkspaceID: "workspace-123", + Title: "Test Task", + Description: "Test description", + } + before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch) + + if _, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", LaunchOptions{ + AgentProfileID: "profile-123", + Prompt: "test prompt", + StartAgent: true, + }); err != nil { + t.Fatalf("LaunchPreparedSession failed: %v", err) + } + + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch); after != before+1 { + t.Fatalf("admitted[launch] counter = %d, want %d", after, before+1) + } + warnings := logs.FilterLevelExact(zapcore.WarnLevel).All() + if len(warnings) != 1 { + t.Fatalf("warning entries = %d, want 1; all=%v", len(warnings), logs.All()) + } + fields := warnings[0].ContextMap() + if fields["site"] != sessionCoresidencySiteLaunch { + t.Fatalf("site field = %v, want %q", fields["site"], sessionCoresidencySiteLaunch) + } +} + +// TestResumeSession_ObservesWorkingSiblingOnAgentStart pins the wiring +// half of AC-004.1 for the resume seam. Ordering is enforced by +// production code structure, not a synchronized assertion here. +func TestResumeSession_ObservesWorkingSiblingOnAgentStart(t *testing.T) { + repo := newMockRepository() + setupLiveResumeTestFixture(repo) + repo.sessions["sess-1"].State = models.TaskSessionStateFailed + repo.sessions["sess-sibling"] = &models.TaskSession{ + ID: "sess-sibling", TaskID: "task-1", State: models.TaskSessionStateRunning, + } + + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + agentMgr := &mockAgentManager{ + launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { + return &LaunchAgentResponse{AgentExecutionID: "exec-new"}, nil + }, + } + exec := NewExecutor(agentMgr, repo, log, ExecutorConfig{ShellPrefs: &mockShellPrefs{}}) + exec.SetCapabilities(&mockCapabilities{}) + before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteResume) + + if _, err := exec.ResumeSession(context.Background(), repo.sessions["sess-1"], true); err != nil { + t.Fatalf("ResumeSession: %v", err) + } + + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteResume); after != before+1 { + t.Fatalf("admitted[resume] counter = %d, want %d", after, before+1) + } + warnings := logs.FilterLevelExact(zapcore.WarnLevel).All() + if len(warnings) != 1 { + t.Fatalf("warning entries = %d, want 1; all=%v", len(warnings), logs.All()) + } + fields := warnings[0].ContextMap() + if fields["site"] != sessionCoresidencySiteResume { + t.Fatalf("site field = %v, want %q", fields["site"], sessionCoresidencySiteResume) + } +} + +// TestRunAgentProcessAsync_ObservesStartingSiblingsBeforeProcessStart pins +// the concurrent-start boundary: both sessions are durably STARTING before +// their process-start goroutines inspect siblings, and each observation runs +// before its own process-start call. +func TestRunAgentProcessAsync_ObservesStartingSiblingsBeforeProcessStart(t *testing.T) { + exec, repo, logs := newSessionCoresidencyTestExecutor(t) + repo.sessions["session-a"] = &models.TaskSession{ + ID: "session-a", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + repo.sessions["session-b"] = &models.TaskSession{ + ID: "session-b", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + + // Return detached snapshots so the asynchronous RUNNING transitions cannot + // race with the observer's sibling-state reads. + repo.listTaskSessionsFunc = func(_ context.Context, _ string) ([]*models.TaskSession, error) { + repo.mu.Lock() + defer repo.mu.Unlock() + return []*models.TaskSession{ + cloneMockTaskSession(repo.sessions["session-a"]), + cloneMockTaskSession(repo.sessions["session-b"]), + }, nil + } + + before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch) + var managerMu sync.Mutex + startedWithoutObservation := make(map[string]int) + started := make(chan string, 2) + executionSessions := map[string]string{"exec-a": "session-a", "exec-b": "session-b"} + manager := &mockAgentManager{ + startAgentProcessFunc: func(_ context.Context, agentExecutionID string) error { + managerMu.Lock() + sessionID := executionSessions[agentExecutionID] + observed := false + for _, entry := range logs.FilterLevelExact(zapcore.WarnLevel).All() { + if entry.ContextMap()["session_id"] == sessionID { + observed = true + break + } + } + if !observed { + startedWithoutObservation[sessionID]++ + } + managerMu.Unlock() + started <- agentExecutionID + return nil + }, + } + exec.agentManager = manager + + exec.runAgentProcessAsyncWithObservation( + context.Background(), "task-123", "session-a", "exec-a", + sessionCoresidencySiteLaunch, func(context.Context) {}, false, false, + ) + exec.runAgentProcessAsyncWithObservation( + context.Background(), "task-123", "session-b", "exec-b", + sessionCoresidencySiteLaunch, func(context.Context) {}, false, false, + ) + + for range 2 { + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for both agent processes to start") + } + } + + managerMu.Lock() + startedBeforeObservation := len(startedWithoutObservation) + managerMu.Unlock() + if startedBeforeObservation != 0 { + t.Fatalf("agent process started before co-residency observation for %d session(s): %v", startedBeforeObservation, startedWithoutObservation) + } + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteLaunch); after != before+2 { + t.Fatalf("admitted[launch] counter = %d, want %d", after, before+2) + } + if warnings := logs.FilterLevelExact(zapcore.WarnLevel).All(); len(warnings) != 2 { + t.Fatalf("warning entries = %d, want 2", len(warnings)) + } +} diff --git a/docs/plans/concurrent-session-worktree-visibility/plan.md b/docs/plans/concurrent-session-worktree-visibility/plan.md new file mode 100644 index 0000000000..a4d66afa96 --- /dev/null +++ b/docs/plans/concurrent-session-worktree-visibility/plan.md @@ -0,0 +1,139 @@ +--- +created: 2026-09-21 +status: done +requirements: + - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004 +system_design: + - ../../specs/tasks/system-design/additional-session-workspace-reuse.md +legacy_specs: [] +--- + +# Implementation Plan: Concurrent Session Worktree Visibility + +## Overview + +Two sessions of one task can run agents against the same physical worktree at +the same time. Kandev permits this by design, but records nothing when it +happens, so an operator learns about it only by noticing two models in the UI +or by finding corrupted files in the repository. + +This plan makes the condition observable at the two seams that start an agent +process against an already-attached workspace. It adds no lock and no refusal. + +The confirmed path is: + +1. A workflow step pins an agent profile, and the source step's + `profile_session_end_policy` is the default `park` + (`NormalizeWorkflowProfileSessionEndPolicy`, + `apps/backend/internal/task/models/models.go:1313`). +2. `parkSessionForProfileSwitchClaimLocked` moves the source session to + `WAITING_FOR_INPUT` and stops its runtime + (`apps/backend/internal/orchestrator/workflow_profile_session_lifecycle.go:151`). + The conversation stays answerable. +3. The destination step's session runs in the same task environment. The + worktree is keyed by `UNIQUE(task_environment_id, repository_id)` + (`apps/backend/internal/task/repository/sqlite/base_migrations.go:1138`), + not by session. +4. A message to the parked session resumes it. Admission is per session only: + `validateAndLockResume` takes `getSessionLock(session.ID)` and + `rejectRunningResume` consults `GetExecutionBySession(session.ID)` + (`apps/backend/internal/orchestrator/executor/executor_resume.go:1173`, + `:1270`). No seam consults sibling sessions. +5. Both agents write the shared worktree. Nothing is logged or counted. + +## Scope + +### In scope + +- Count sibling sessions in a working state at + `Executor.LaunchPreparedSession` (agent start) and `Executor.resumeSession`. +- Emit a structured warning and an expvar counter when the count is non-zero. +- Record a skip with its reason when the sibling read fails. +- Document the park interaction in `docs/public/tasks-and-workflows.md`. +- Go unit coverage for co-resident, solitary, and read-failure cases. + +### Out of scope + +- Refusing, deferring, or serializing a second concurrent launch. The active + requirement states that multiple agents may write the shared workspace + concurrently and that this implies no task-wide writer lock + (`docs/specs/tasks/requirements/additional-session-workspace-reuse.md`, + "Migrated source detail" and "Out of scope"). Changing that is a product + decision and belongs on the feature board. +- Changing the default `profile_session_end_policy` from `park`. +- Any board, task-card, or session-tab indicator. +- Changing `is_primary` from routing metadata into an admission gate. +- Repairing already-corrupted files in any user repository. + +## Technical approach + +### Observation point + +`LaunchPreparedSession` and `resumeSession` are the two functions that start an +agent process for a session bound to an existing task environment. Both already +hold the per-session lock and both already call +`admitWorktreeRecovery(ctx, task.ID)`, so the task ID is in hand at the point +where the process is about to start. + +The existing `hasOtherWorkingSessions` +(`apps/backend/internal/orchestrator/executor/executor_execute.go:554`) already +lists a task's sessions and applies `isRuntimeWorkingSessionState`. Generalize +it into a helper that returns the sibling IDs and a read-failure signal, and +keep its current boolean caller on that helper so both users share one +definition of "working". + +### Counters + +Add `apps/backend/internal/orchestrator/executor/session_coresidency_metrics.go` +modelled on `apps/backend/internal/orchestrator/office_stall_metrics.go`: +`session_coresidency_admitted_total` labelled by `site`, and +`session_coresidency_observation_skipped_total` labelled by `reason`. Labels are +closed sets; session and task identifiers appear in the log entry only. + +## Tests + +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.1`: a launch and a resume + with one working sibling each record co-residency naming that sibling. +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.2`: the observation does not + change the admission result; a task with no working sibling records nothing. +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.3`: a failing + `ListTaskSessions` records a skip with its reason and still admits. +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.4`: recorded label keys are + the declared closed set and carry no identifier values. + +## E2E tests + +None. The change adds a backend structured log, two expvar counters, and +documentation. Nothing a user sees or can do changes, so there is no Playwright +scenario to assert and no mobile composition to check. + +## Work orders + +- [x] [Task 01: Observe concurrent worktree sessions](task-01-observe-concurrent-worktree-sessions.md) + +## Verification results + +See task-01's "Definition-of-Done receipts" and "Pre-existing failure proof" +sections for full commands and output. Summary: `apps/backend` fmt/lint/test +green except 4 `internal/worktree` tests reproduced identically against +`origin/main` (pre-existing, unrelated package); root `lint-format` and +`apps/web`'s `i18n:ratchet` green (no UI files touched); spec validation +green. Committed as `06f16e718` on `feature/two-live-sessions-ca-nal`. + +## Risks + +- `ListTaskSessions` runs once per agent start/resume — a new read on that + path, not reused from elsewhere. Implemented without an early-stop + optimization: a task's live-session count is small in practice and the read + happens once per launch, not per turn, so the collect-all approach was kept + for simplicity over the plan's original "stop at first working sibling" + idea. Revisit only if a task with a pathologically large session history is + observed to matter. +- Observing inside the per-session lock must not perform a blocking call that + could extend lock hold time materially. The read is the same bounded + repository call the path already makes. +- An operator who reads the warning may expect Kandev to have prevented the + overlap. The log wording states the condition is permitted ("Kandev permits + concurrent sessions on one task") and avoids language implying a failure to + prevent it; pinned by + `TestObserveSessionCoresidency_WorkingSiblingLogsWarningAndIncrementsCounter`. diff --git a/docs/plans/concurrent-session-worktree-visibility/task-01-observe-concurrent-worktree-sessions.md b/docs/plans/concurrent-session-worktree-visibility/task-01-observe-concurrent-worktree-sessions.md new file mode 100644 index 0000000000..a05c9b2391 --- /dev/null +++ b/docs/plans/concurrent-session-worktree-visibility/task-01-observe-concurrent-worktree-sessions.md @@ -0,0 +1,208 @@ +--- +id: "01-observe-concurrent-worktree-sessions" +title: "Observe concurrent worktree sessions" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004 +acceptance_criteria: + - AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.1 + - AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.2 + - AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.3 + - AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.4 +system_design: + - ../../specs/tasks/system-design/additional-session-workspace-reuse.md +--- + +# Task 01: Observe Concurrent Worktree Sessions + +## Summary + +Record, at the two seams that start an agent process against an already +attached task workspace, that another session of the same task is already +working. Emit a structured warning and an expvar counter. Change no admission +outcome. + +## In scope + +- Generalize `hasOtherWorkingSessions` into a helper returning sibling session + IDs plus a read-failure signal, and keep its existing boolean caller on it. +- Call the helper from `Executor.LaunchPreparedSession` when it starts the + agent, and from `Executor.resumeSession`, before the process starts. +- Add `session_coresidency_admitted_total` (labelled by `site`) and + `session_coresidency_observation_skipped_total` (labelled by `reason`). +- Add the park interaction warning to `docs/public/tasks-and-workflows.md` + where it currently says a parked session is not an active process. +- Add Go unit coverage for the co-resident, solitary, and read-failure cases. + +## Out of scope + +- Refusing, deferring, or serializing any launch or resume. +- Changing the default `profile_session_end_policy`. +- Frontend, board indicators, or session-tab surfaces. +- Changing `is_primary` semantics. +- Instance-wide session-ceiling behavior. + +## Acceptance + +- A launch or resume with a working sibling increments the counter once for + that start and logs the sibling session IDs. +- A launch or resume with no working sibling records nothing. +- A failing sibling read records a skip with its reason, and the launch or + resume still proceeds. +- Counter label keys are exactly `site` and `reason`; no identifier appears as + a label value. + +## Verification + +```bash +cd apps/backend && go test ./internal/orchestrator/executor +cd apps/backend && make lint +python3 scripts/lint-spec-files.py --all +python3 scripts/list-docs.py validate +``` + +Each new unit test must fail before the production change and pass after it. + +## Files likely touched + +- `apps/backend/internal/orchestrator/executor/executor_execute.go` +- `apps/backend/internal/orchestrator/executor/executor_resume.go` +- `apps/backend/internal/orchestrator/executor/session_coresidency_metrics.go` +- `apps/backend/internal/orchestrator/executor/session_coresidency_test.go` +- `docs/public/tasks-and-workflows.md` +- `docs/plans/concurrent-session-worktree-visibility/plan.md` +- `docs/plans/concurrent-session-worktree-visibility/task-01-observe-concurrent-worktree-sessions.md` + +## Dependencies + +None. + +## Risks + +- Placing the observation after the process starts would report the condition + too late to be useful in a log read backwards from a corruption. +- Adding identifiers as counter labels would make cardinality unbounded. +- Logging at error level would imply Kandev failed to prevent something it + deliberately permits; warning with explicit wording is required. + +## Parallelism + +`sequential` + +## Inputs + +- `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004` and its criteria. +- `docs/specs/tasks/system-design/additional-session-workspace-reuse.md`, + section "Concurrent session visibility". +- `docs/decisions/2026-08-31-workflow-profile-session-switch-policy.md`. +- `hasOtherWorkingSessions`, `isRuntimeWorkingSessionState`, + `LaunchPreparedSession`, `resumeSession`, `validateAndLockResume`. +- `apps/backend/internal/orchestrator/office_stall_metrics.go` as the expvar + label-model precedent. + +## Results + +Implemented on `feature/two-live-sessions-ca-nal`, commit `06f16e718`. + +- `executor_execute.go`: extracted `workingSessionSiblings` (returns sibling + IDs + error, no early stop — a task's live-session count is small and the + call runs once per launch/resume, not per turn, so the plan's "stop at + first working sibling" mitigation was unneeded; kept for simplicity). + `hasOtherWorkingSessions` now calls it (behavior-preserving refactor, same + log lines). Added `observeSessionCoresidency(ctx, site, taskID, sessionID)`: + warns + increments `session_coresidency_admitted_total{site}` when siblings + exist; on a read failure, increments + `session_coresidency_observation_skipped_total{reason=sibling_read_failed}` + and warns, never treating the failure as zero siblings. +- Wired into `LaunchPreparedSession` (guarded on `startAgent`, immediately + before the fast-path/full-launch branch) and `resumeSession` (guarded on + `startAgent`, immediately after the existing `admitWorktreeRecovery` call). + Both are before the agent process starts on every branch. +- `session_coresidency_metrics.go` (new): the two expvar maps, modelled on + `orchestrator/office_stall_metrics.go`. Labels are `site` (`launch`, + `resume`) and `reason` (`sibling_read_failed`) only — no identifiers. +- `session_coresidency_test.go` (new): solitary session (nothing recorded), + working sibling (warning + counter, wording asserted to contain "permit" + and not "fail"), sibling-read failure (skip counter + warning, not the + admitted counter), plus two wiring tests driving real + `LaunchPreparedSession`/`ResumeSession` calls to confirm the call sites are + reached, not just the extracted helper. +- `executor_mocks_test.go`: added `listTaskSessionsFunc` override hook so the + read-failure case could be simulated (mirrors the existing + `listActiveTaskSessionsByTaskIDFunc` pattern). +- `docs/public/tasks-and-workflows.md`: added a paragraph after the "parked + session is not an active process" sentence stating that answering a parked + session starts an agent in the same shared workspace a still-running + destination session may be using, that Kandev permits and records but does + not prevent this, and that using **Complete the session** avoids it. + +### TDD receipt + +Watched the new tests fail before the production code existed: stashed +`executor_execute.go`, `executor_resume.go`, and the new +`session_coresidency_metrics.go` (keeping the new test file and the mock +hook), then ran `go vet ./internal/orchestrator/executor/...`: + +``` +vet: internal/orchestrator/executor/session_coresidency_test.go:62:25: undefined: sessionCoresidencyAdmittedTotalVar +``` + +Restored the stash, re-ran green: + +``` +$ go test ./internal/orchestrator/executor/... -run 'ObservesWorkingSibling' -v +--- PASS: TestLaunchPreparedSession_ObservesWorkingSiblingBeforeStartingAgent (0.00s) +--- PASS: TestResumeSession_ObservesWorkingSiblingBeforeStartingAgent (0.00s) +PASS +``` + +### Definition-of-Done receipts + +``` +$ make -C apps/backend fmt # clean; unrelated pre-existing drift in + # session_open_recovery_test.go / + # workflow_start_prompt.go reverted, not ours +$ make -C apps/backend lint # golangci-lint run ./... -> 0 issues. +$ make -C apps/backend test # CGO_ENABLED=1 go test -tags fts5 ./... + # FAIL internal/worktree only (see below); + # every other package ok +$ make lint-format # prettier --check web cli packages -> ok +$ (cd apps/web && pnpm run i18n:ratchet) + # no UI source added or modified +$ python3 scripts/lint-spec-files.py --all # All specification files passed. +$ python3 scripts/list-docs.py validate # Validated 292 decisions and 1054 specifications. +``` + +`apps/node_modules` was missing in this worktree (the documented worktree +gotcha); ran `cd apps && pnpm install --frozen-lockfile` once before the +web-side commands above. + +### Pre-existing failure proof (`internal/worktree`) + +`make -C apps/backend test` failed 4 tests, all in `internal/worktree`, a +package this task never touches: +`TestRestoreManagedBranchFromRecoveryHead_BoundsUpdateRefAdmissionAndReleasesRepoLock`, +`TestManager_AdmitTaskRecoveryCoalescesFollowerAfterSuccessfulRecovery`, +`TestManager_Create_RefusesInvalidNonEmptyCheckoutWithoutRemovingIt`, +`TestManager_Create_RefusesMissingAdminWhenRecordedBranchIsUnreachable` — all +four are on the repo's known-failures list ("Worktree and workspace lifecycle +tests"). Reproduced on `git merge-base HEAD origin/main` (`9e171c83774368a8d39dea3721467ea6a0b5fb85`, +which is also this branch's own base — no commits diverge yet) in a scratch +worktree at `/tmp/scratch-worktree-coresidency`: + +``` +$ cd /tmp/scratch-worktree-coresidency/apps/backend && CGO_ENABLED=1 go test -tags fts5 \ + -run 'TestManager_Create_RefusesInvalidNonEmptyCheckoutWithoutRemovingIt|TestManager_AdmitTaskRecoveryCoalescesFollowerAfterSuccessfulRecovery|TestManager_Create_RefusesMissingAdminWhenRecordedBranchIsUnreachable|TestRestoreManagedBranchFromRecoveryHead_BoundsUpdateRefAdmission' \ + -v ./internal/worktree/... +--- FAIL: TestRestoreManagedBranchFromRecoveryHead_BoundsUpdateRefAdmissionAndReleasesRepoLock (1.01s) +--- FAIL: TestManager_AdmitTaskRecoveryCoalescesFollowerAfterSuccessfulRecovery (0.51s) +--- FAIL: TestManager_Create_RefusesInvalidNonEmptyCheckoutWithoutRemovingIt (0.27s) +--- FAIL: TestManager_Create_RefusesMissingAdminWhenRecordedBranchIsUnreachable (0.41s) +``` + +Same tests, same error messages, on unmodified `origin/main`. Pre-existing; +not this task's regression. Scratch worktree removed after the check +(`git worktree remove /tmp/scratch-worktree-coresidency --force`). diff --git a/docs/public/tasks-and-workflows.md b/docs/public/tasks-and-workflows.md index 549c984800..735cfb978c 100644 --- a/docs/public/tasks-and-workflows.md +++ b/docs/public/tasks-and-workflows.md @@ -675,6 +675,17 @@ the parked switch, it keeps the current session recoverable and reports the error. The destination step controls start behavior; the source step controls end behavior. +Answering a parked session starts an agent process for it again, in the same +task workspace the destination step's session may still be using. Kandev +permits more than one session of a task to write that shared workspace +concurrently; it does not serialize them or lock the workspace to one writer. +If the destination session is still active when you answer the parked one, +both write the same files. Kandev records this condition (a structured log +entry and an operational counter) but does not prevent, delay, or queue it. To +avoid concurrent writers, confirm the destination session has finished before +answering a parked one, or use **Complete the session** instead of **Park the +session** for steps that pin a different agent profile. + When **Reset agent context** creates a fresh ACP session, Kandev preserves the selected ACP model, permission mode, and provider options. It restores these settings before the next automatic prompt. If the provider rejects a setting, diff --git a/docs/specs/tasks/requirements/additional-session-workspace-reuse.md b/docs/specs/tasks/requirements/additional-session-workspace-reuse.md index 988b0742ee..3d7745958c 100644 --- a/docs/specs/tasks/requirements/additional-session-workspace-reuse.md +++ b/docs/specs/tasks/requirements/additional-session-workspace-reuse.md @@ -1,8 +1,8 @@ --- -status: draft +status: active system: tasks created: 2026-08-19 -updated: 2026-08-30 +updated: 2026-09-22 owners: - kandev --- @@ -80,6 +80,32 @@ starting in a stale workspace retained by the previous executor. workspace shall not delete, move, reset, clean, checkout, or otherwise modify the stale path, the canonical repository, or either environment inventory. +### REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004: Concurrent Session Workspace Visibility + +**Intent:** Keep sanctioned concurrent writing on a shared task workspace +observable, without introducing a task-wide writer lock. + +**User story:** As an operator, I want Kandev to record when an agent starts in +a task workspace that another session is already working in, so that I can +recognize a shared-workspace conflict instead of discovering it later as +corrupted files. + +#### Acceptance criteria + +- **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.1:** When a launch or a + resume starts an agent process for a session whose task already holds another + session in a working state, the system shall record that co-residency before + the agent process starts. +- **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.2:** A co-residency record + shall be an operational counter and a structured log entry only. It shall not + refuse, defer, delay, or otherwise change the admission outcome. +- **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.3:** When the sibling + session read fails, the system shall record a skipped observation with its + reason rather than reporting an absence of co-residency. +- **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004.4:** Recorded labels shall + be a closed set of admission site and skip reason. No task, session, + workspace path, branch, or repository identifier shall be a label value. + ## Migrated source detail ## Why diff --git a/docs/specs/tasks/system-design/additional-session-workspace-reuse.md b/docs/specs/tasks/system-design/additional-session-workspace-reuse.md index e770b920b8..8c602ffee8 100644 --- a/docs/specs/tasks/system-design/additional-session-workspace-reuse.md +++ b/docs/specs/tasks/system-design/additional-session-workspace-reuse.md @@ -5,6 +5,7 @@ requirements: - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001 - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-002 - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-003 + - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004 created: 2026-08-30 owners: - kandev @@ -31,6 +32,7 @@ executor transition, and agent promotion or resume. It follows | `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001` | [Canonical environment and inventory](#canonical-environment-and-inventory) | | `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-002` | [Recovery and path authority](#recovery-and-path-authority) | | `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-003` | [Executor transition](#executor-transition), [Launch admission](#launch-admission) | +| `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-004` | [Concurrent session visibility](#concurrent-session-visibility) | ## Canonical environment and inventory @@ -131,6 +133,39 @@ The lifecycle manager retains a defense-in-depth guard: if a repo-backed marker and exact selected environment identity, it refuses to create the execution. It never treats a non-empty path as sufficient proof. +## Concurrent session visibility + +Multiple sessions of one task attach to a single canonical environment and its +single physical worktree. Concurrent writing is permitted by design; this +section adds no lock, no refusal, and no ordering between those writers. It +makes the condition observable. + +Two seams start an agent process against an already-attached workspace: +`Executor.LaunchPreparedSession` when it is asked to start the agent, and +`Executor.resumeSession`. Both already hold the per-session lock and both know +the task ID. Immediately before the process starts, each reads the task's +sessions and counts siblings in a working state, excluding the session being +started. The existing `sessionstate.IsWorking` predicate defines that state, so +observation and the executor's own review reconciliation agree on one +definition. + +A non-zero count emits a structured warning naming the admitting site, the +session being started, and the sibling session IDs, and increments an expvar +counter under `session_coresidency_*`. Labels carry the admitting site and, for +skips, the reason; identifiers stay in the log entry and never become label +values, so counter cardinality is bounded. + +The sibling read can fail. A failed read records a skip with its reason instead +of a zero count, so a degraded observer is distinguishable from a task that +genuinely has one live session. The read never blocks admission: an observation +error is logged and the launch proceeds. + +The workflow profile-switch park policy reaches this condition without any user +request. Parking leaves the source session in `WAITING_FOR_INPUT` with its +runtime stopped but its conversation answerable, while the destination step's +session runs. A later message to the parked session starts a second agent in +the shared worktree. That is the path this observation is calibrated for. + ## Persistence and projection After successful launch or resume, the orchestrator refreshes