feat(orchestrator): observe concurrent sessions reusing one worktree (#3851)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-22 17:32:33 +08:00
committed by GitHub
parent f5a505c468
commit 8690df2f79
10 changed files with 912 additions and 20 deletions
@@ -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.
@@ -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)
@@ -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",
@@ -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)
}
@@ -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))
}
}
@@ -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`.
@@ -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`).
+11
View File
@@ -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,
@@ -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
@@ -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