From 3f3392e2c4e1cacd7be1cc45affc6b55d4bc2179 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 13:02:55 +0200 Subject: [PATCH 1/7] The recovery harness reads a worker's word once the worker has said it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The credential check runs the moment the connector exits, and counted there whether every worker that started had taken its token or said why it could not. The word is the worker's, written from its own process at its own pace, and it is due by the worker's exit, not the connector's. At the kill point where the attempt is running before the prompt the two are not one moment: the connector is killed at its "running" line with the token handoff still in flight on the socket's goroutine, and the acp row's worker, whose bind began as the last act of its handshake, learns of the death as a reset on the socket and says so a fraction of a millisecond after the parent has seen the connector go. Read at the exit, the log counted it as a worker that said nothing: 13 of 20 runs on main. The spawn rows passed the same check at the same kill point over nobody, since their worker had not even reached its agent when the connector died a millisecond after starting it. The check now waits for every worker owed a word — the process the ledger recorded for the attempt, and every one that reached its agent — to say it or to be gone, and counts from that reading. A worker that is gone without a word still fails it. 0 of 20 runs after, and the spawn rows now count the worker they start: it dials a dead socket and says so. --- internal/connector/recovery_harness_test.go | 106 +++++++++++++++++--- 1 file changed, 94 insertions(+), 12 deletions(-) diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index 951601983..2a22db848 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -663,21 +663,27 @@ func (h *harness) stopWatchingForTokenFiles() int { // log, the lifecycle messages it posted, the polls it made, the workspace // records), not in any file under a working directory or the state directory // — and no worker saw one appear in those files while it ran. +// +// It runs when the connector has exited, and the workers' side of it is read +// once the workers have said their word (awaitWorkersWord): a worker's word +// is due by its own exit, not by the connector's. func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { t := h.t t.Helper() watched := h.stopWatchingForTokenFiles() tokens := h.taskTokens() - log := h.agentLog() + log := h.awaitWorkersWord(stateDir) // A worker that bound to its task took a token, and the harness kept it. // If it did not, this check has nothing to look for, and says so rather // than passing. - bound, unbound := 0, 0 + started, bound, unbound := 0, 0, 0 var places drivertest.Places for _, e := range log { places.Env = append(places.Env, e.Env...) places.Args = append(places.Args, e.Args...) switch { + case e.Step == "start": + started++ case e.Step == "bound": bound++ case strings.HasPrefix(e.Step, "bind-failed:"): @@ -692,7 +698,7 @@ func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { // whoever its worker was — a fake one, or a real agent through the // bridge, which leaves no agent log at all. require.Len(t, tokens, h.tasksLaunched(stateDir), "every task the connector launched left its token for this check") - require.Equal(t, h.workersStarted(), bound+unbound, + require.Equal(t, started, bound+unbound, "every worker that started either took its task's token or said why it could not") for _, token := range h.takenTokens() { require.Contains(t, tokens, token, "a worker took a token the connector did not mint for its task") @@ -702,7 +708,7 @@ func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { // Nothing to check is a fact about the run, not a pass: a run with // no worker (a kill before the spawn, a start that ran nothing) is // the only way here. - require.Equal(t, h.workersStarted(), unbound, + require.Equal(t, started, unbound, "a worker that started either took a token or said why it could not") return } @@ -727,7 +733,8 @@ func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { require.Positive(t, read, "the scan read files; a scan that read nothing has cleared nothing") files += read } - t.Logf("credential check: %d task tokens, %d files read, %d watched while the run went on", len(tokens), files, watched) + t.Logf("credential check: %d task tokens, %d files read, %d watched while the run went on; %d workers started, %d bound, %d said why not", + len(tokens), files, watched, started, bound, unbound) } // taskTokens is every token the connector minted for a task, as its launch @@ -772,16 +779,91 @@ func (h *harness) tasksLaunched(dir string) int { return n } -// workersStarted counts the worker processes that reached their agent, which -// is every worker that could have been handed a token. -func (h *harness) workersStarted() int { - n := 0 - for _, e := range h.agentLog() { +// awaitWorkersWord waits for every worker of the run to have said what became +// of its token — "bound", or "bind-failed:" and why — and returns the agent +// log once it has. The word is the worker's, written from its own process at +// its own pace, and it is owed by the time the worker exits: not by the time +// the connector does, which is when the credential check runs. +// +// The two are not one moment. A crash row kills the connector at its +// "running" line, written once the socket is armed and the attempt recorded, +// while the handoff itself is still in flight on the socket's own goroutine. +// The acp row's worker began the bind that dials it as the last act of its +// handshake, since the socket cannot be armed before NewSession returns; it +// learns of the death as a reset on the socket and says so a fraction of a +// millisecond after the parent has seen the connector exit. Read at the exit, +// the log counted that worker as one that said nothing (card 10317728733). +// The spawn rows' workers have not even reached their agent by then — the +// connector dies a millisecond after starting them — so read at the exit, the +// check there ran over nobody. +// +// The workers owed a word are the processes the ledger recorded for the run's +// attempts and the ones that reached their agent; a real agent row leaves no +// agent log and is owed none. A worker that is gone without a word is still +// one that started and said nothing, and the caller fails on it: the wait +// ends at its exit, not at a deadline. +func (h *harness) awaitWorkersWord(stateDir string) []agentLogEntry { + h.t.Helper() + var recorded []int + if !h.driver.Real { + recorded = h.recordedWorkerPIDs(stateDir) + } + var log []agentLogEntry + _ = waitFor(context.Background(), func() (bool, error) { + log = h.agentLog() + for _, pid := range workersOwingAWord(log, recorded) { + if !processGone(context.Background(), pid) { + return false, nil + } + } + // Whoever still owes a word is gone, and a process that is gone has + // written all it ever will: this reading is the one to count. + log = h.agentLog() + return true, nil + }) + return log +} + +// workersOwingAWord is every worker, recorded by the ledger or started as far +// as its agent, whose word is not in log. +func workersOwingAWord(log []agentLogEntry, recorded []int) []int { + said := map[int]bool{} + for _, e := range log { + if e.Step == "bound" || strings.HasPrefix(e.Step, "bind-failed:") { + said[e.PID] = true + } + } + var owing []int + seen := map[int]bool{} + owe := func(pid int) { + if pid > 0 && !said[pid] && !seen[pid] { + seen[pid] = true + owing = append(owing, pid) + } + } + for _, pid := range recorded { + owe(pid) + } + for _, e := range log { if e.Step == "start" { - n++ + owe(e.PID) } } - return n + return owing +} + +// recordedWorkerPIDs is every worker process the ledger in dir recorded for +// an attempt, read as tasksLaunched reads the ledger: as it is, or not at all. +func (h *harness) recordedWorkerPIDs(dir string) []int { + h.t.Helper() + ctx := context.Background() + l, err := OpenLedgerReadOnly(ctx, filepath.Join(dir, LedgerFile)) + if errors.Is(err, os.ErrNotExist) { + return nil + } + require.NoError(h.t, err) + defer func() { _ = l.Close() }() + return recordedWorkers(h.t, l) } // scanForSecret reads every file under dirs, in a process of its own, and From 77d00ec739581da4b89eb92ae7510271ca5b5155 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 13:26:52 +0200 Subject: [PATCH 2/7] The credential check holds both halves of the wait, by identity, with the watch still running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wait for the workers' word ran after the parent had stopped watching for a token in a file, though the workers it waits on are exactly the ones still able to write one; it discarded its own deadline, so a worker still alive without a word thirty seconds after the connector died left a partial log to count; a worker the ledger recorded that was gone without ever reaching its agent made the count zero of zero; and it waited on bare pids, which the kernel gives away, so it could have waited on a stranger or taken a stranger's exit for the worker's (Copilot on #752). The watch now stops once the workers have spoken. The wait fails on its deadline, and fails again on a worker gone without a word. Each worker is waited on by its identity — the kernel start time the ledger recorded, or the one the fake writes on its own start entry, which used to carry a wall-clock stamp no identity check would answer about — through the same ProcessGone the connector uses. 0 of 20 runs on the acp row, 0 of 5 each on claude and codex, 0 of 432 subtests over three runs of the whole recovery table. --- internal/connector/recovery_dispatch_test.go | 3 + internal/connector/recovery_harness_test.go | 82 +++++++++++++------- internal/connector/recovery_worker_test.go | 5 +- 3 files changed, 61 insertions(+), 29 deletions(-) diff --git a/internal/connector/recovery_dispatch_test.go b/internal/connector/recovery_dispatch_test.go index a64552a2c..c9218f372 100644 --- a/internal/connector/recovery_dispatch_test.go +++ b/internal/connector/recovery_dispatch_test.go @@ -275,8 +275,11 @@ func recordedAttempts(t *testing.T, l *Ledger) []recordedAttempt { ) require.NoError(t, rows.Scan(&a.id, &a.state, &a.process.PID, &a.process.PGID, &started)) if started.Valid { + // The ledger writes the stamp only when it is the kernel's + // (AttemptProcess.startedStamp), so one read back is exact. a.process.StartedAt, err = parseStamp(started.String) require.NoError(t, err) + a.process.StartedExact = true } out = append(out, a) } diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index 2a22db848..dcca71b09 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -666,13 +666,15 @@ func (h *harness) stopWatchingForTokenFiles() int { // // It runs when the connector has exited, and the workers' side of it is read // once the workers have said their word (awaitWorkersWord): a worker's word -// is due by its own exit, not by the connector's. +// is due by its own exit, not by the connector's. The watch for a token in a +// file runs until then too, since a worker still running is a worker that +// can still write one. func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { t := h.t t.Helper() + log := h.awaitWorkersWord(stateDir) watched := h.stopWatchingForTokenFiles() tokens := h.taskTokens() - log := h.awaitWorkersWord(stateDir) // A worker that bound to its task took a token, and the harness kept it. // If it did not, this check has nothing to look for, and says so rather // than passing. @@ -798,63 +800,85 @@ func (h *harness) tasksLaunched(dir string) int { // check there ran over nobody. // // The workers owed a word are the processes the ledger recorded for the run's -// attempts and the ones that reached their agent; a real agent row leaves no -// agent log and is owed none. A worker that is gone without a word is still -// one that started and said nothing, and the caller fails on it: the wait -// ends at its exit, not at a deadline. +// attempts and the ones that reached their agent, each by its identity — pid +// and kernel start time — so a pid the kernel has since given to a stranger +// is not waited on as if it were the worker. A real agent row leaves no +// agent log and is owed none. Both halves are held: a worker still alive +// without a word when the wait runs out fails, and so does one that is gone +// without ever saying it — a process that could have been handed a token +// and left no account of it is not a pass. func (h *harness) awaitWorkersWord(stateDir string) []agentLogEntry { - h.t.Helper() - var recorded []int + t := h.t + t.Helper() + var recorded []driver.Process if !h.driver.Real { - recorded = h.recordedWorkerPIDs(stateDir) + recorded = h.recordedWorkerProcesses(stateDir) } var log []agentLogEntry - _ = waitFor(context.Background(), func() (bool, error) { + var alive []int + err := waitFor(context.Background(), func() (bool, error) { log = h.agentLog() - for _, pid := range workersOwingAWord(log, recorded) { - if !processGone(context.Background(), pid) { - return false, nil + alive = alive[:0] + for _, p := range workersOwingAWord(log, recorded) { + gone, err := driver.ProcessGone(p) + if err != nil { + return false, fmt.Errorf("worker pid %d: %w", p.PID, err) + } + if !gone { + alive = append(alive, p.PID) } } + if len(alive) > 0 { + return false, nil + } // Whoever still owes a word is gone, and a process that is gone has // written all it ever will: this reading is the one to count. log = h.agentLog() return true, nil }) + require.NoError(t, err, "a worker still owes its word after the connector exited: pids %v", alive) + owing := workersOwingAWord(log, recorded) + silent := make([]int, 0, len(owing)) + for _, p := range owing { + silent = append(silent, p.PID) + } + require.Empty(t, silent, "a worker is gone without saying whether it took its task's token: pids %v", silent) return log } // workersOwingAWord is every worker, recorded by the ledger or started as far -// as its agent, whose word is not in log. -func workersOwingAWord(log []agentLogEntry, recorded []int) []int { +// as its agent, whose word is not in log — by its identity, so it can be +// waited on and told from a later process with its pid. +func workersOwingAWord(log []agentLogEntry, recorded []driver.Process) []driver.Process { said := map[int]bool{} for _, e := range log { if e.Step == "bound" || strings.HasPrefix(e.Step, "bind-failed:") { said[e.PID] = true } } - var owing []int + var owing []driver.Process seen := map[int]bool{} - owe := func(pid int) { - if pid > 0 && !said[pid] && !seen[pid] { - seen[pid] = true - owing = append(owing, pid) + owe := func(p driver.Process) { + if p.PID > 0 && !said[p.PID] && !seen[p.PID] { + seen[p.PID] = true + owing = append(owing, p) } } - for _, pid := range recorded { - owe(pid) + for _, p := range recorded { + owe(p) } for _, e := range log { if e.Step == "start" { - owe(e.PID) + owe(identityOf(e.PID, e.PGID, e.StartedAt)) } } return owing } -// recordedWorkerPIDs is every worker process the ledger in dir recorded for -// an attempt, read as tasksLaunched reads the ledger: as it is, or not at all. -func (h *harness) recordedWorkerPIDs(dir string) []int { +// recordedWorkerProcesses is every worker process the ledger in dir recorded +// for an attempt, with the identity it recorded, read as tasksLaunched reads +// the ledger: as it is, or not at all. +func (h *harness) recordedWorkerProcesses(dir string) []driver.Process { h.t.Helper() ctx := context.Background() l, err := OpenLedgerReadOnly(ctx, filepath.Join(dir, LedgerFile)) @@ -863,7 +887,11 @@ func (h *harness) recordedWorkerPIDs(dir string) []int { } require.NoError(h.t, err) defer func() { _ = l.Close() }() - return recordedWorkers(h.t, l) + var out []driver.Process + for _, a := range recordedAttempts(h.t, l) { + out = append(out, a.process) + } + return out } // scanForSecret reads every file under dirs, in a process of its own, and diff --git a/internal/connector/recovery_worker_test.go b/internal/connector/recovery_worker_test.go index 212a0a011..beef7617b 100644 --- a/internal/connector/recovery_worker_test.go +++ b/internal/connector/recovery_worker_test.go @@ -65,9 +65,10 @@ func newFakeWorker(dir string) (*fakeWorker, error) { w := &fakeWorker{dir: dir, sc: sc, replies: map[int64]int64{}} pgid, _ := syscall.Getpgid(0) // What this agent was started with, for the parent to check no task - // token is in either. + // token is in either. The stamp is the kernel's, as on every entry: the + // parent waits on this process by it, and ends it by it. _ = appendJSONLine(filepath.Join(dir, agentLogFile), agentLogEntry{ - PID: os.Getpid(), PGID: pgid, StartedAt: time.Now(), Step: "start", Args: os.Args[1:], Env: os.Environ(), + PID: os.Getpid(), PGID: pgid, StartedAt: kernelStart(os.Getpid()), Step: "start", Args: os.Args[1:], Env: os.Environ(), }) return w, nil } From e59f064f881c07a58be77c698fdc584175901704 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 14:01:41 +0200 Subject: [PATCH 3/7] The acp fake says its word on every exit, and a word is matched to its worker by identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The acp fake spoke only once its bind had started, and its bind starts at the read-back, the last act of the handshake. A worker whose handshake never got there — the connector dead before it, as on a CI runner where a guard fired mid-launch — exited with a "start" entry and no word, and the check named it as gone without one (run 35339688393, pid 16811). It now says so on that exit: no server started, no token was asked for. The spawn fakes bind before they speak and never had the gap. The check matched a word to a worker by pid alone, though the log and the ledger both accumulate across a harness's connector runs, so a word an earlier worker said could have discharged a later one the kernel named alike. It now matches by pid and kernel start time, the same identity ProcessGone waits on (Copilot on #752). Run with stdin closed before any handshake, the fake of 77d00ec7 leaves "start" alone in the agent log; this one leaves "start" and "bind-failed: the session ended before its MCP servers started, so no token was asked for". --- internal/connector/recovery_acp_test.go | 12 ++++++++- internal/connector/recovery_harness_test.go | 29 ++++++++++++++++----- 2 files changed, 34 insertions(+), 7 deletions(-) diff --git a/internal/connector/recovery_acp_test.go b/internal/connector/recovery_acp_test.go index e8c90bf4d..3a0c068d2 100644 --- a/internal/connector/recovery_acp_test.go +++ b/internal/connector/recovery_acp_test.go @@ -121,10 +121,19 @@ func fakeACPAdapter(w *fakeWorker) int { // A server that was starting when the connector died still says what // became of its token: the parent checks that every worker either took // one or said why it could not, and a process that exited with the bind - // still in flight would answer neither. + // still in flight would answer neither. Nor would one whose handshake + // never reached the read-back — a connector that died first, a session + // it ended on the way — so that exit says so too: no server started, no + // token was asked for. The spawn fakes bind before they speak and so + // always say a word; this one speaks last, and must not leave without. + said := false defer func() { if starting { <-bound + return + } + if !said { + w.log(0, 0, "bind-failed: the session ended before its MCP servers started, so no token was asked for") } }() awaitBind := func() error { @@ -205,6 +214,7 @@ func fakeACPAdapter(w *fakeWorker) int { reported = "bypassPermissions" w.log(0, 0, "bad-mode") w.log(0, 0, "bind-failed: the session ended in its handshake, before its MCP servers started") + said = true } notify("session/update", map[string]any{"sessionId": sessionID, "update": map[string]any{"sessionUpdate": "current_mode_update", "currentModeId": reported}}) diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index dcca71b09..e6ce04848 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -846,21 +846,38 @@ func (h *harness) awaitWorkersWord(stateDir string) []agentLogEntry { return log } +// workerIdentity is a worker as the log and the ledger both name it: its pid +// and the kernel's start time for it, which is what tells it from a later +// process the kernel gave the same pid. Both write the same instant — the +// ledger from the process the driver started, the fake from its own — so +// one worker's entries match each other and nobody else's. +type workerIdentity struct { + pid int + started int64 +} + +func identityKey(p driver.Process) workerIdentity { + return workerIdentity{pid: p.PID, started: p.StartedAt.UnixNano()} +} + // workersOwingAWord is every worker, recorded by the ledger or started as far // as its agent, whose word is not in log — by its identity, so it can be -// waited on and told from a later process with its pid. +// waited on and told from a later process with its pid, and so a word an +// earlier worker said cannot discharge a later one the kernel named alike: +// the log and the ledger both accumulate across a harness's connector runs. func workersOwingAWord(log []agentLogEntry, recorded []driver.Process) []driver.Process { - said := map[int]bool{} + said := map[workerIdentity]bool{} for _, e := range log { if e.Step == "bound" || strings.HasPrefix(e.Step, "bind-failed:") { - said[e.PID] = true + said[identityKey(identityOf(e.PID, e.PGID, e.StartedAt))] = true } } var owing []driver.Process - seen := map[int]bool{} + seen := map[workerIdentity]bool{} owe := func(p driver.Process) { - if p.PID > 0 && !said[p.PID] && !seen[p.PID] { - seen[p.PID] = true + key := identityKey(p) + if p.PID > 0 && !said[key] && !seen[key] { + seen[key] = true owing = append(owing, p) } } From a4542b27346fa8d9e773670da8833c8e0be7ad2e Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 14:01:41 +0200 Subject: [PATCH 4/7] The guard rows give the launch the time the guard's period must cover The guard is due a delay after admission, and the rows kill the connector as it posts the acknowledgement. That kill has to land on an attempt already running, with a worker recorded, or the restart holds an attempt it cannot identify and the surviving run never settles: 120 seconds and a failure that names neither. At 50ms a loaded CI runner lost that race (run 35339688393, both acp rows), and so does this machine under load: 3 of 3 runs, each with the attempt left mid-launch. A second is the guard's own period, sized so a working directory, the ledger, the token socket, the agent's wrapper and, for the acp row, its whole handshake fit inside it; 0 of 10 loaded runs after, 0 of 10 without load. A check after the kill now says in a moment when the launch still lost, rather than at the surviving run's deadline. --- internal/connector/recovery_dispatch_test.go | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/internal/connector/recovery_dispatch_test.go b/internal/connector/recovery_dispatch_test.go index c9218f372..a73f57127 100644 --- a/internal/connector/recovery_dispatch_test.go +++ b/internal/connector/recovery_dispatch_test.go @@ -617,12 +617,28 @@ func TestRecoveryTheGuardAcknowledgementIsPostedAtMostOnce(t *testing.T) { t.Run(row.name, func(t *testing.T) { // A worker that never calls get_dispatch is what the guard is // for: the acknowledgement falls to the connector. + // + // The guard is due a delay after admission, and the rows + // kill the connector as it posts. That kill must land on an + // attempt already running — a worker recorded, so a restart + // can end it and settle — and the delay is what buys the + // launch that time: a working directory, the ledger, the + // token socket, the agent's wrapper and, for the acp row, + // its whole handshake. At 50ms a loaded CI runner lost that + // race, the attempt was left launching with no worker to + // identify, and the restart held it rather than settle, as + // it must. A second is not a sleep for an outcome, it is the + // guard's own period, sized so the launch fits; the check + // after the kill says so when it does not, in a moment + // rather than at the surviving run's deadline. h := newHarness(t, d, harnessScenario{ - GuardDelay: 50 * time.Millisecond, + GuardDelay: time.Second, Plans: map[string][]string{"101#1": {"linger"}}, }) h.publish(feedEntry{Event: todoEvent(101, 5001)}) h.run(harnessRun{Kill: row.kill, Killed: true}) + require.NotEmpty(t, recordedWorkers(t, h.ledger()), + "the guard outran the launch: the attempt was still launching when the guard was due, so no restart can identify its worker or settle it") h.run(harnessRun{}) h.run(harnessRun{}) From 9666371b0c0e3b6b55011aa125e43bd2fa03986f Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 14:22:09 +0200 Subject: [PATCH 5/7] The watch for a token in a file runs for the harness, and a word with no identity discharges nobody MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The watch stopped at each run's credential check, though a worker that has said its word may still be running — one left lingering for a restart to end, a real agent — and can write a token to a file in the gap between one run's exit and the next run's start, or after the last. It now starts with the first run and ends at cleanup, once every worker this harness started has been ended, and reports then (Copilot on #752, the note it held back at 77d00ec7 and again at a4542b27). A word written with no kernel start stamp could have discharged a ledger record with none — no identity matching no identity, by pid alone. Such a word now discharges nobody; the worker it was meant for reaches ProcessGone, or the silent check, and is answered for there. TestAWorkersWordIsMatchedByItsIdentity holds the matching to this. On the pid-only matching of 77d00ec7 three of its four cases fail: an earlier process's word under the same pid discharges the later worker, a word with no identity discharges it, and a start entry is discharged by a word that was never its own. --- internal/connector/recovery_harness_test.go | 90 ++++++++++++++++++--- 1 file changed, 80 insertions(+), 10 deletions(-) diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index e6ce04848..b448de84b 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -364,6 +364,10 @@ func newHarness(t *testing.T, d harnessDriver, sc harnessScenario) *harness { for _, name := range []string{feedFile, storeFile, linesFile, pollsFile, agentLogFile, liveFile, workspaceFile} { require.NoError(t, os.WriteFile(filepath.Join(dir, name), nil, 0o600)) } + // Registered before killAgents so it runs after it: the watch for a token + // in a file ends, and reports, once every worker this harness started has + // been ended. + t.Cleanup(func() { h.stopWatchingForTokenFiles() }) t.Cleanup(h.killAgents) return h } @@ -591,14 +595,24 @@ func (h *harness) killAgents() { } } -// watchForTokenFiles watches, for the rest of this run, every place a task -// token must never be written, for every token a worker takes while it runs. -// The workers watch too, but a worker the connector ends never reports; this -// watcher is the parent's, and always does. +// watchForTokenFiles watches, for the rest of this harness, every place a +// task token must never be written, for every token a worker takes while it +// runs. The workers watch too, but a worker the connector ends never reports; +// this watcher is the parent's, and always does. +// +// For the harness, not for a run: a worker outlives the connector that +// started it — one left lingering for a restart to end, a real agent, one +// still binding when the connector died — and can write a file in the gap +// between one run's exit and the next run's start, or after the last run. +// The watch begins with the first run and ends at cleanup, once every worker +// is ended (newHarness), and reports then. func (h *harness) watchForTokenFiles() { h.t.Helper() h.watchMu.Lock() defer h.watchMu.Unlock() + if h.watchStop != nil { + return + } if h.watching == nil { h.watching = map[string]func() []string{} } @@ -639,7 +653,15 @@ func (h *harness) knownTokens() []string { return out } -// stopWatchingForTokenFiles ends the watchers and reports what they saw. +// watchedTokens is how many tokens the watch has found to watch so far. +func (h *harness) watchedTokens() int { + h.watchMu.Lock() + defer h.watchMu.Unlock() + return len(h.watching) +} + +// stopWatchingForTokenFiles ends the watchers and reports what they saw. It +// is the harness's cleanup: every worker has been ended by then. func (h *harness) stopWatchingForTokenFiles() int { h.watchMu.Lock() defer h.watchMu.Unlock() @@ -667,13 +689,15 @@ func (h *harness) stopWatchingForTokenFiles() int { // It runs when the connector has exited, and the workers' side of it is read // once the workers have said their word (awaitWorkersWord): a worker's word // is due by its own exit, not by the connector's. The watch for a token in a -// file runs until then too, since a worker still running is a worker that -// can still write one. +// file is not stopped here at all: a worker that has spoken may still be +// running — lingering for a restart to end it — and can still write one, so +// the watch runs for the harness and reports at its cleanup +// (watchForTokenFiles). func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { t := h.t t.Helper() log := h.awaitWorkersWord(stateDir) - watched := h.stopWatchingForTokenFiles() + watched := h.watchedTokens() tokens := h.taskTokens() // A worker that bound to its task took a token, and the harness kept it. // If it did not, this check has nothing to look for, and says so rather @@ -868,9 +892,17 @@ func identityKey(p driver.Process) workerIdentity { func workersOwingAWord(log []agentLogEntry, recorded []driver.Process) []driver.Process { said := map[workerIdentity]bool{} for _, e := range log { - if e.Step == "bound" || strings.HasPrefix(e.Step, "bind-failed:") { - said[identityKey(identityOf(e.PID, e.PGID, e.StartedAt))] = true + if e.Step != "bound" && !strings.HasPrefix(e.Step, "bind-failed:") { + continue + } + if e.StartedAt.IsZero() { + // A word with no identity discharges nobody: it could be an + // earlier process's, under a pid the kernel gave again. The + // worker it was meant for reaches ProcessGone, or the silent + // check, and is answered for there. + continue } + said[identityKey(identityOf(e.PID, e.PGID, e.StartedAt))] = true } var owing []driver.Process seen := map[workerIdentity]bool{} @@ -892,6 +924,44 @@ func workersOwingAWord(log []agentLogEntry, recorded []driver.Process) []driver. return owing } +// A worker's word is its own: it is matched by pid and kernel start time, +// so a word an earlier process said under a pid the kernel gave again does +// not discharge the later worker, and a word with no identity discharges no +// one. The log and the ledger both accumulate across a harness's runs, which +// is what makes the pid alone a stranger's key. +func TestAWorkersWordIsMatchedByItsIdentity(t *testing.T) { + earlier := time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC) + later := earlier.Add(time.Second) + recorded := []driver.Process{{PID: 4242, PGID: 4242, StartedAt: later, StartedExact: true}} + pids := func(ps []driver.Process) []int { + out := make([]int, 0, len(ps)) + for _, p := range ps { + out = append(out, p.PID) + } + return out + } + + t.Run("its own word discharges it", func(t *testing.T) { + log := []agentLogEntry{{PID: 4242, PGID: 4242, StartedAt: later, Step: "bound"}} + assert.Empty(t, pids(workersOwingAWord(log, recorded))) + }) + t.Run("an earlier process's word, under the same pid, does not", func(t *testing.T) { + log := []agentLogEntry{{PID: 4242, PGID: 4242, StartedAt: earlier, Step: "bind-failed: the earlier one"}} + assert.Equal(t, []int{4242}, pids(workersOwingAWord(log, recorded))) + }) + t.Run("a word with no identity discharges nobody", func(t *testing.T) { + log := []agentLogEntry{{PID: 4242, PGID: 4242, Step: "bound"}} + assert.Equal(t, []int{4242}, pids(workersOwingAWord(log, recorded))) + }) + t.Run("a start entry is owed by its own identity too", func(t *testing.T) { + log := []agentLogEntry{ + {PID: 4343, PGID: 4343, StartedAt: later, Step: "start"}, + {PID: 4343, PGID: 4343, StartedAt: earlier, Step: "bound"}, + } + assert.Equal(t, []int{4343}, pids(workersOwingAWord(log, nil))) + }) +} + // recordedWorkerProcesses is every worker process the ledger in dir recorded // for an attempt, with the identity it recorded, read as tasksLaunched reads // the ledger: as it is, or not at all. From 2f9b07499f9306dd245a061114537ca3b0047d25 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 14:41:26 +0200 Subject: [PATCH 6/7] The reading of the token watch joins the loop that registers watchers before it reads them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loop that registers a watcher for each token and the cleanup that stopped and read the watchers shared the map with nothing ordering them: the loop could read the tokens on disk, the cleanup could drain the map, and the loop could then register a watcher nobody would stop or read — a leak it went on to see, discarded. The race was per run before this branch and per harness after it; a check inside the loop would have narrowed it (Copilot on #752), and joining the loop removes it. A watcher is now started in one place only, from that loop, and the reading closes the loop and waits for it to return before it touches the map. Then the reading registers, once, whatever token the loop's last interval never got to — a watcher scans before it is stopped — and only then stops and reads every watcher there is. Nothing found is dropped: every watcher ever started is in the map, and every one in the map is read by the one goroutine that can still reach it. TestTheTokenWatchReadsEveryWatcherItCouldHaveStarted holds it: a token taken just before the reading is watched and counted by it, no watcher is registered after, and a later reading watches what it knows and reads it. On the watcher of 9666371b it fails 20 of 20 runs — the token taken just before the reading was not watched at all. --- internal/connector/recovery_harness_test.go | 92 ++++++++++++++++----- 1 file changed, 72 insertions(+), 20 deletions(-) diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index b448de84b..8a94d1a00 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -325,11 +325,13 @@ type harnessScenario struct { type harness struct { t *testing.T dir string - // watching is the parent's watch for each task token, for the run in - // flight; watchStop ends it. + // watching is the parent's watch for each task token, for the life of + // the harness; watchStop ends the loop that registers them, and + // watchDone is closed once that loop has returned. watchMu sync.Mutex watching map[string]func() []string watchStop chan struct{} + watchDone chan struct{} // state is the connector's state directory, under this harness's own // XDG_STATE_HOME and named as the connector names it, so a worker's MCP // server resolves it exactly as `basecamp mcp --connect-state` does. @@ -603,9 +605,15 @@ func (h *harness) killAgents() { // For the harness, not for a run: a worker outlives the connector that // started it — one left lingering for a restart to end, a real agent, one // still binding when the connector died — and can write a file in the gap -// between one run's exit and the next run's start, or after the last run. -// The watch begins with the first run and ends at cleanup, once every worker +// between one run's exit and the next run's start, or after the last. The +// watch begins with the first run and ends at cleanup, once every worker // is ended (newHarness), and reports then. +// +// A watcher is started in one place only, registerTokenWatchers, from the +// loop this starts; and the cleanup that reads the watchers joins that loop +// before it reads them (stopWatchingForTokenFiles). So no watcher can be +// registered after the reading, to be neither stopped nor read: not because +// the loop checks, but because the loop is over. func (h *harness) watchForTokenFiles() { h.t.Helper() h.watchMu.Lock() @@ -616,27 +624,35 @@ func (h *harness) watchForTokenFiles() { if h.watching == nil { h.watching = map[string]func() []string{} } - stop := make(chan struct{}) - h.watchStop = stop + stop, done := make(chan struct{}), make(chan struct{}) + h.watchStop, h.watchDone = stop, done go func() { + defer close(done) for { select { case <-stop: return case <-time.After(5 * time.Millisecond): } - for _, token := range h.knownTokens() { - h.watchMu.Lock() - if _, ok := h.watching[token]; !ok { - h.watching[token] = drivertest.WatchForSecretFiles(token, - h.workDir(), filepath.Join(h.dir, "work-other"), filepath.Join(h.dir, "sessions")) - } - h.watchMu.Unlock() - } + h.registerTokenWatchers() } }() } +// registerTokenWatchers starts a watcher for every token a worker has taken +// that has none yet. It is the one place a watcher is started. +func (h *harness) registerTokenWatchers() { + tokens := h.knownTokens() + h.watchMu.Lock() + defer h.watchMu.Unlock() + for _, token := range tokens { + if _, ok := h.watching[token]; !ok { + h.watching[token] = drivertest.WatchForSecretFiles(token, + h.workDir(), filepath.Join(h.dir, "work-other"), filepath.Join(h.dir, "sessions")) + } + } +} + // knownTokens reads the tokens the workers have taken so far, ignoring a // directory that does not exist yet. func (h *harness) knownTokens() []string { @@ -660,15 +676,28 @@ func (h *harness) watchedTokens() int { return len(h.watching) } -// stopWatchingForTokenFiles ends the watchers and reports what they saw. It -// is the harness's cleanup: every worker has been ended by then. +// stopWatchingForTokenFiles ends the watch and reports what it saw. It is +// the harness's cleanup: every worker has been ended by then. +// +// The loop that registers watchers is joined first, so that once the +// watchers are read nothing can add one; then one last registration pass +// of this goroutine's own, so a token first seen in the loop's final +// interval is watched too — a watcher scans once before it is stopped — +// and only then is every watcher stopped and its finding read. Nothing +// found is dropped: every watcher that was ever started is in the map, and +// every one in the map is read here. func (h *harness) stopWatchingForTokenFiles() int { h.watchMu.Lock() - defer h.watchMu.Unlock() - if h.watchStop != nil { - close(h.watchStop) - h.watchStop = nil + stop, done := h.watchStop, h.watchDone + h.watchStop, h.watchDone = nil, nil + h.watchMu.Unlock() + if stop != nil { + close(stop) + <-done } + h.registerTokenWatchers() + h.watchMu.Lock() + defer h.watchMu.Unlock() watched := len(h.watching) for token, stop := range h.watching { for _, found := range stop() { @@ -679,6 +708,29 @@ func (h *harness) stopWatchingForTokenFiles() int { return watched } +// The watch's reading cannot miss a watcher: a token taken just before the +// reading, in the interval the registering loop never got to, is watched +// and counted by the reading itself; and once the reading has happened, no +// watcher can be registered any more, however the loop's last interval and +// the cleanup happened to interleave. +func TestTheTokenWatchReadsEveryWatcherItCouldHaveStarted(t *testing.T) { + require.NotEmpty(t, harnessDrivers) + h := newHarness(t, harnessDrivers[0], harnessScenario{}) + h.watchForTokenFiles() + tokens := filepath.Join(h.dir, tokensDir) + require.NoError(t, os.MkdirAll(tokens, 0o700)) + // Taken now, and the watch read now: before the loop's next interval. + require.NoError(t, os.WriteFile(filepath.Join(tokens, "taken-1.token"), []byte("tok_one"), 0o600)) + assert.Equal(t, 1, h.stopWatchingForTokenFiles(), "the token taken just before the reading is watched and counted by it") + // And once read, no loop registers any more: a token taken afterwards + // is not watched until something reads again — and a reading watches + // what it knows, once, and reads it, so nothing it could see is dropped. + require.NoError(t, os.WriteFile(filepath.Join(tokens, "taken-2.token"), []byte("tok_two"), 0o600)) + time.Sleep(25 * time.Millisecond) + assert.Zero(t, h.watchedTokens(), "no watcher is registered after the reading") + assert.Equal(t, 2, h.stopWatchingForTokenFiles(), "a later reading watches every token it knows, once, and reads it") +} + // requireNoTaskTokenLeaked holds every run to the credential rule, for every // task token any worker was handed so far: not in an agent's argv or // environment, not in anything the connector wrote (its stdout lines, its From 53857980d8bc6de6f8810dc73abe2f68f2ef8830 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 15:05:59 +0200 Subject: [PATCH 7/7] The guard rows make the order they test rather than wait a second for it The guard is due a delay after admission, and the rows kill the connector as it posts the acknowledgement. That kill has to land on an attempt already running, with a worker recorded, and a4542b27 bought the launch that time with a period of one second: a window, not an order, and a window sized on one machine (Copilot on #752, and the reviewer of 2f9b0749). The fake Basecamp now holds a guard acknowledgement until the connector has written its running line for an attempt ("guard-after-running", a fault of the run's like the two it had). The line is the connector's own, written to the harness's file once MarkRunning has committed, and the post is made outside any transaction of the ledger's, so nothing the launch needs is held while Basecamp waits; the only bound is the harness's 30-second waitFor, which is a deadline on the launch itself. The guard's period goes back to 50ms, and the check after the kill is what that order promises rather than a hope about a machine. Under CPU load, at 50ms without the hold: 3 of 3 runs lost the launch, the attempt left mid-launch. With it: 0 of 10 loaded, 0 of 5 without load across all three drivers. Under -race, where a process takes over a second to start and neither row is in the representative subset, run with that gate lifted for the measurement: 0 of 5 for the guard rows and 0 of 10 for the acp kill-point row, no data race. The whole recovery table: 0 failures over 432 subtests. --- internal/connector/recovery_connector_test.go | 2 +- internal/connector/recovery_dispatch_test.go | 27 ++++++++------- internal/connector/recovery_fakes_test.go | 34 +++++++++++++++++++ internal/connector/recovery_harness_test.go | 4 ++- 4 files changed, 52 insertions(+), 15 deletions(-) diff --git a/internal/connector/recovery_connector_test.go b/internal/connector/recovery_connector_test.go index 55ba2fcb6..05923e51c 100644 --- a/internal/connector/recovery_connector_test.go +++ b/internal/connector/recovery_connector_test.go @@ -322,7 +322,7 @@ func runHarnessConnector(dir string) error { return err } outbox, err := NewOutbox(OutboxOptions{ - Ledger: ledger, Poster: storePoster{dir: dir, kill: kill}, + Ledger: ledger, Poster: storePoster{dir: dir, kill: kill, fault: os.Getenv(harnessFaultEnv)}, Paused: ledger.Held, Lines: lines, Logger: logger, // A sending intent a previous process left is reconciled once it is // this old, so a restart settles it rather than waiting out the diff --git a/internal/connector/recovery_dispatch_test.go b/internal/connector/recovery_dispatch_test.go index a73f57127..1a5a94738 100644 --- a/internal/connector/recovery_dispatch_test.go +++ b/internal/connector/recovery_dispatch_test.go @@ -621,24 +621,25 @@ func TestRecoveryTheGuardAcknowledgementIsPostedAtMostOnce(t *testing.T) { // The guard is due a delay after admission, and the rows // kill the connector as it posts. That kill must land on an // attempt already running — a worker recorded, so a restart - // can end it and settle — and the delay is what buys the - // launch that time: a working directory, the ledger, the - // token socket, the agent's wrapper and, for the acp row, - // its whole handshake. At 50ms a loaded CI runner lost that - // race, the attempt was left launching with no worker to - // identify, and the restart held it rather than settle, as - // it must. A second is not a sleep for an outcome, it is the - // guard's own period, sized so the launch fits; the check - // after the kill says so when it does not, in a moment - // rather than at the surviving run's deadline. + // can end it and settle — and the launch it races (a working + // directory, the ledger, the token socket, the agent's + // wrapper and, for the acp row, its whole handshake) takes + // what the machine gives it: on a loaded CI runner, more than + // the guard's 50ms, and the attempt was left launching with + // no worker to identify, held by the restart rather than + // settled, as it must be. So the fake Basecamp holds the + // acknowledgement until the connector has written its running + // line ("guard-after-running"): the ordering is made, not + // waited for, and the guard's period can stay short. The + // check after the kill is what that ordering promises. h := newHarness(t, d, harnessScenario{ - GuardDelay: time.Second, + GuardDelay: 50 * time.Millisecond, Plans: map[string][]string{"101#1": {"linger"}}, }) h.publish(feedEntry{Event: todoEvent(101, 5001)}) - h.run(harnessRun{Kill: row.kill, Killed: true}) + h.run(harnessRun{Kill: row.kill, Killed: true, Fault: "guard-after-running"}) require.NotEmpty(t, recordedWorkers(t, h.ledger()), - "the guard outran the launch: the attempt was still launching when the guard was due, so no restart can identify its worker or settle it") + "the kill landed on an attempt still launching: the acknowledgement was posted before the running line") h.run(harnessRun{}) h.run(harnessRun{}) diff --git a/internal/connector/recovery_fakes_test.go b/internal/connector/recovery_fakes_test.go index 576cd612a..0a5bc243d 100644 --- a/internal/connector/recovery_fakes_test.go +++ b/internal/connector/recovery_fakes_test.go @@ -508,9 +508,28 @@ func (h *harness) connectorPosts() []storedMessage { type storePoster struct { dir string kill *killSpec + // fault is the run's standing misbehavior (harnessRun.Fault): + // "guard-after-running" holds a guard acknowledgement until the + // connector has written its running line for an attempt. + fault string } func (p storePoster) Post(ctx context.Context, dest Destination, body string) (int64, error) { + if p.fault == "guard-after-running" && body == GuardAckBody { + // The guard is due a delay after admission, whether or not the + // launch it races has finished; a row that kills the connector as + // the acknowledgement is posted needs the kill to land on an + // attempt already running, with a worker recorded, or the restart + // holds an attempt it cannot identify. Basecamp is the one party + // that can wait for that without a seam in the connector: the + // running line is this process's own, written to the harness's + // file once MarkRunning has committed, and the post is made + // outside any transaction of the ledger's, so nothing the launch + // needs is held while it waits. + if err := waitFor(ctx, func() (bool, error) { return runningLineWritten(p.dir) }); err != nil { + return 0, fmt.Errorf("guard-after-running: no attempt was recorded running: %w", err) + } + } if p.kill.at("post-before") { die() } @@ -524,6 +543,21 @@ func (p storePoster) Post(ctx context.Context, dest Destination, body string) (i return id, nil } +// runningLineWritten reports whether the connector in dir has written a +// running line for any attempt: the line follows MarkRunning's commit, so a +// worker is recorded for it by then. +func runningLineWritten(dir string) (bool, error) { + running := false + err := readJSONLines(filepath.Join(dir, linesFile), func(line []byte) error { + var l DispatchLine + if json.Unmarshal(line, &l) == nil && l.Type == "dispatch" && l.State == string(AttemptRunning) { + running = true + } + return nil + }) + return running, err +} + func (p storePoster) List(_ context.Context, dest Destination, since time.Time) ([]PostedMessage, error) { all, err := storedMessages(p.dir) if err != nil { diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go index 8a94d1a00..61ab3580a 100644 --- a/internal/connector/recovery_harness_test.go +++ b/internal/connector/recovery_harness_test.go @@ -444,7 +444,9 @@ type harnessRun struct { StateDir string // Fault is a standing misbehavior of the fake Basecamp for the run: // "stall-catch-up" holds the feed's first poll until the socket has - // served every live event; "repair-stall" never answers a repair poll. + // served every live event; "repair-stall" never answers a repair poll; + // "guard-after-running" holds a guard acknowledgement until the + // connector has written its running line for an attempt. Fault string // Env is added to the connector's environment. Env []string