From 6c375cd932eb2563ea7112a10029f042a79ceb7e Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:17:47 +0200 Subject: [PATCH 1/4] An update emitted after a Codex session's reader has gone is dropped, not sent on a closed channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reader closes the session's updates when the worker's output ends, but it is not the only goroutine that emits. A prompt's writer finishing a canceled turn, and the policy check ending an unsafe one, both call lastWord from goroutines of their own: each reads the refusals Codex logged on its way out and emits an update for every one. Either can reach that after the reader has gone, and a send on a closed channel is a panic — the connector crashed with one on CI. The close and every send now take the session's lock, and an update emitted after the close is dropped, as an update onto a full buffer already is. The refusal itself is unaffected: it reaches the ledger before the update is offered. A test holds it. The worker's output ends while the worker lives on, so the reader waits out its grace and finishes having read a stderr with no refusal in it; the worker then logs its refusal as it is ended, and the last word is read from a goroutine that is not the reader's. On the previous emit it panics every run. --- internal/connector/driver/codex/codex.go | 37 ++++++++++++++-- internal/connector/driver/codex/codex_test.go | 43 +++++++++++++++++++ 2 files changed, 77 insertions(+), 3 deletions(-) diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index a50d428cd..0b6901504 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -455,6 +455,11 @@ type session struct { verifyDone chan struct{} verifyErr error closed bool + // updatesClosed is the reader's record that the updates channel is + // closed. It is read and written under the same lock every emit takes, + // so a goroutine still finishing a turn cannot send on a channel that + // has just been closed. + updatesClosed bool // writing is a one-slot semaphore around the worker's stdin. A lock // would be worse: a worker that stops reading its input blocks the // write, and everything waiting on the lock — Close among them — waits @@ -620,24 +625,50 @@ func (s *session) finish(t *turn, result driver.PromptResult, err error) { close(t.done) } +// emit offers an update to whoever is reading the session's. It is best +// effort by design: an update nobody is there to take — because the buffer is +// full, or because the session's updates are over — is dropped, never waited +// on. The send is under the lock that closeUpdates takes, so the last word of +// a turn being finished off the reader — the prompt's writer, the policy +// check — is dropped rather than sent on a closed channel. func (s *session) emit(u driver.Update) { u.At = time.Now() u.Tool = s.red.Sanitize(u.Tool) u.ToolCallID = s.red.Sanitize(u.ToolCallID) + s.mu.Lock() + defer s.mu.Unlock() + if s.updatesClosed { + return + } select { case s.updates <- u: default: } } +// closeUpdates ends the session's updates, once. The reader owns the close, +// but it is not the only goroutine that emits: a prompt's writer finishing a +// canceled turn, or the policy check ending an unsafe one, may read a refusal +// from the worker's stderr after the reader has gone. Closing under the lock +// every emit takes is what makes that a dropped update rather than a panic. +func (s *session) closeUpdates() { + s.mu.Lock() + defer s.mu.Unlock() + if s.updatesClosed { + return + } + s.updatesClosed = true + close(s.updates) +} + // read maps the process's JSON lines onto updates and the turn's result until // the process closes its stdout. func (s *session) read() { defer func() { // The updates channel closes last: finishing the turn still emits - // (a refusal read from stderr), and a send on a closed channel is a - // panic, not a dropped update. - defer close(s.updates) + // (a refusal read from stderr), and an update emitted after this is + // dropped rather than sent. + defer s.closeUpdates() s.mu.Lock() s.ended = true t := s.turn diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index 494a506f8..cf6b27791 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -1042,6 +1042,49 @@ func TestEveryRefusalCodexOnlyLogsIsRecorded(t *testing.T) { assert.Len(t, result.Refusals, 3) } +// A refusal read after the session's updates have closed is recorded and its +// update dropped, not a panic. The reader owns the close, but it is not the +// only goroutine that reads the worker's last word: a prompt's writer +// finishing a canceled turn, and the policy check ending an unsafe one, both +// call lastWord from goroutines of their own, and either may reach it after +// the reader has gone. This makes that ordering rather than waiting for it — +// the worker logs its refusal only once the reader has given up on it — and +// then reads the last word from a goroutine that is not the reader's, as +// those two do. +func TestARefusalReadAfterTheUpdatesCloseIsNotAPanic(t *testing.T) { + recorder := &drivertest.Refusals{} + h := newHarness(t, scenario{ + TurnContext: safeTurnContext(), + Events: []string{`{"type":"turn.started"}`}, + // The output ends while the worker lives on, so the reader waits out + // its grace and ends having read a stderr with no refusal in it. + CloseStdout: true, + Hang: true, + // The refusal Codex logs on its way out, after all that. + StderrOnTerm: "patch rejected: writing outside of the project; rejected by user approval settings", + }) + cfg := h.config() + cfg.Refusals = recorder + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + session := s.(*session) + go func() { _, _ = s.Prompt(context.Background(), "Event 1.") }() + // Drains until the session's updates are closed, which is the ordering + // this test is built on: no duration is waited out for it. + for range s.Updates() { //nolint:revive // the drain is the wait + } + require.Empty(t, recorder.Recorded(), "the reader ended having read no refusal") + + // Now the worker is ended, and logs the refusal on its way out. + require.NoError(t, s.Close()) + require.Eventually(t, func() bool { return strings.Contains(session.StderrTail(), "rejected") }, + 10*time.Second, 20*time.Millisecond, "the worker logged its refusal on its way out") + session.stderrRefusals() + + assert.Len(t, recorder.Recorded(), 1, + "the refusal is recorded, and the update it carries is dropped rather than sent on a closed channel") +} + // A refusal Codex logged is recorded even when the turn it belonged to has // already ended: the reader reads the stderr of a worker that is gone, with // no turn left to hang it on. From 9f4f6f499577b267b5a434f503f833993e5bcd32 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:17:57 +0200 Subject: [PATCH 2/4] The fake ACP agent publishes its record in order, so its last word on disk is its latest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A flood of permission requests is answered on a goroutine each, and each writes the whole record once it has its outcome. Two of those interleave: one takes its snapshot, a later one takes a fuller snapshot and renames it into place, and then the first renames its older one over that. The file's last word is then a record the agent has already moved past, and nothing writes again to correct it — so a test waiting for what the agent has done waits for a word that was said and unsaid. The two also shared the temporary file they renamed from. The snapshot and its rename are now one step, so what lands is never older than what landed before it. TestAFloodOfPermissionRequestsIsBounded is the test that spends 60 seconds on this and gives up: it waits for 52 of 60 requests to be answered, and 52 is exactly how many are ever answered, so losing one publication loses the whole wait. With the window between the snapshot and its publication widened to 5ms it fails 5 runs of 5, with the same message and the same 60 seconds CI reports; with that delay still in and the publication ordered, 0 of 10. --- internal/connector/driver/acp/acp_test.go | 20 ++++++++++++++++--- .../connector/driver/acp/fakeagent_test.go | 15 ++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/internal/connector/driver/acp/acp_test.go b/internal/connector/driver/acp/acp_test.go index 2ab1a4d96..f5e1b65c6 100644 --- a/internal/connector/driver/acp/acp_test.go +++ b/internal/connector/driver/acp/acp_test.go @@ -14,6 +14,7 @@ import ( "os" "path/filepath" "slices" + "strconv" "strings" "sync" "sync/atomic" @@ -208,6 +209,13 @@ func gone(pid int) bool { return errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) } +// counter is a count a failed wait can print: testify formats the message +// when the wait gives up, so the value has to be read then and not passed by +// value when the wait is set up. +type counter struct{ atomic.Int32 } + +func (c *counter) String() string { return strconv.Itoa(int(c.Load())) } + func waitGone(t *testing.T, pid int) { t.Helper() require.Eventually(t, func() bool { return gone(pid) }, 10*time.Second, 20*time.Millisecond, "pid %d still exists", pid) @@ -1218,9 +1226,15 @@ func TestAFloodOfPermissionRequestsIsBounded(t *testing.T) { }() require.Eventually(t, func() bool { return deciding.Load() == maxDecisions }, 60*time.Second, 10*time.Millisecond, "the session decides at most %d at once", maxDecisions) - // Every request but the ones stuck in a decision has been answered. - require.Eventually(t, func() bool { return len(h.record().Outcomes) >= flood-maxDecisions }, 60*time.Second, 20*time.Millisecond, - "a flood is answered as it arrives") + // Every request but the ones stuck in a decision has been answered. The + // count is carried into the message so a wait that gives up says how far + // the flood got, rather than only that it did not finish. + var outcomes counter + require.Eventually(t, func() bool { + outcomes.Store(int32(len(h.record().Outcomes))) //nolint:gosec // a count of at most flood + return int(outcomes.Load()) >= flood-maxDecisions + }, 60*time.Second, 20*time.Millisecond, + "a flood is answered as it arrives: %v of the %d not stuck in a decision", &outcomes, flood-maxDecisions) assert.LessOrEqual(t, deciding.Load(), int32(maxDecisions)) answered := h.record().Outcomes close(release) diff --git a/internal/connector/driver/acp/fakeagent_test.go b/internal/connector/driver/acp/fakeagent_test.go index b237c072c..17baae96a 100644 --- a/internal/connector/driver/acp/fakeagent_test.go +++ b/internal/connector/driver/acp/fakeagent_test.go @@ -129,6 +129,16 @@ type fakeAgent struct { sc scenario out *bufio.Writer + // publishing orders the record's publications: a flood of permission + // requests is answered on a goroutine each, and each writes the record + // once it has its outcome. Without it two publications interleave — one + // takes its snapshot, a later one takes a fuller snapshot and renames it + // into place, then the first renames its older one over that — and the + // file's last word is a record the agent has already moved past. A reader + // waiting for what the agent has done then waits for a word that has been + // said and unsaid. + publishing sync.Mutex + mu sync.Mutex rec agentRecord nextID int @@ -239,7 +249,12 @@ func runFakeChild() { time.Sleep(time.Hour) } +// flush publishes the record. The snapshot and its rename are one step, under +// publishing, so what lands is never older than what landed before it — and +// two publications never share the temporary file they rename from. func (a *fakeAgent) flush() { + a.publishing.Lock() + defer a.publishing.Unlock() a.mu.Lock() data, _ := json.Marshal(a.rec) a.mu.Unlock() From 64efec782814887f0c1d1cf348486e3892463fd9 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:18:23 +0200 Subject: [PATCH 3/4] The failed-handshake test asks whether the adapter's group runs, not whether the kernel has reaped it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit kill(pid, 0) succeeds on a zombie, so the test was asking for the whole group to be reaped at the instant NewSession returns. The connector promises something else, and something narrower: ConfirmGroupGone waits until no member of the group is running, and a zombie runs nothing. Reaping what is left of an orphan belongs to whoever adopted it — the agent's child outlives the agent it was started by, so its parent is init or a subreaper, not this process — and that wait is nobody's to promise. Gone is now asked the way the connector asks it, through driver.ProcessGone, and about processes identified while they were still running: a pid on its own is one the kernel may have given away by the time the question is asked. The assertion says which of the two was still there. Run under a subreaper that adopts orphans and never waits for them, the previous check fails all 4 runs, in 12.03 seconds and with the message CI reports; this one passes 3 runs of 3 under the same subreaper. --- internal/connector/driver/acp/acp_test.go | 36 +++++++++++++++++++---- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/internal/connector/driver/acp/acp_test.go b/internal/connector/driver/acp/acp_test.go index f5e1b65c6..1ace24fda 100644 --- a/internal/connector/driver/acp/acp_test.go +++ b/internal/connector/driver/acp/acp_test.go @@ -1396,6 +1396,16 @@ func TestARefusalDecidedAsTheTurnEndsIsOnItsResult(t *testing.T) { // A handshake that fails after the adapter started leaves nothing of its // process group behind by the time NewSession returns: the caller settles the // attempt on that error. +// +// Gone is asked the way the connector asks it, through driver.ProcessGone: a +// process that runs nothing is gone, whether or not the kernel has reaped +// what is left of it. A zombie answers a bare kill(pid, 0) as though it were +// alive, and reaping an orphan is not the connector's to do — it belongs to +// whoever adopted it, which on a machine whose init is slow to wait, or that +// runs its tests under a subreaper that never does, may be much later or +// never. Both processes are identified while they are still running, by pid +// and kernel start time, so the question asked afterwards is about them and +// not about whoever the kernel gave those pids to next. func TestAFailedHandshakeLeavesNoGroupBehind(t *testing.T) { // Several runs: the window this closes is a matter of milliseconds. for run := range 4 { @@ -1406,12 +1416,28 @@ func TestAFailedHandshakeLeavesNoGroupBehind(t *testing.T) { d := h.driver() d.opts.HandshakeTimeout = 3 * time.Second d.opts.CloseGrace = 2 * time.Second - _, err := d.NewSession(context.Background(), h.config()) - require.Error(t, err) + failed := make(chan error, 1) + go func() { + _, err := d.NewSession(context.Background(), h.config()) + failed <- err + }() + + // While the handshake hangs, both are running and can be identified. rec := h.record() - require.NotZero(t, rec.ChildPID) - assert.True(t, gone(rec.ChildPID) && gone(rec.PID), - "run %d: the adapter's group is gone when NewSession returns, not a moment later", run) + require.NotZero(t, rec.ChildPID, "run %d: the agent started no child to leave behind", run) + adapter, err := driver.LookupProcess(rec.PID) + require.NoError(t, err, "run %d: the adapter was not running to be identified", run) + child, err := driver.LookupProcess(rec.ChildPID) + require.NoError(t, err, "run %d: the child was not running to be identified", run) + + require.Error(t, <-failed, "run %d: the handshake was supposed to fail", run) + adapterGone, err := driver.ProcessGone(adapter) + require.NoError(t, err, "run %d: the adapter's identity", run) + childGone, err := driver.ProcessGone(child) + require.NoError(t, err, "run %d: the child's identity", run) + assert.True(t, adapterGone && childGone, + "run %d: the adapter's group is gone when NewSession returns, not a moment later (adapter %d gone=%t, child %d gone=%t)", + run, adapter.PID, adapterGone, child.PID, childGone) } } From 917f2d194655543eacf6ca4561e31616cb9814f1 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:18:33 +0200 Subject: [PATCH 4/4] The unsafe-session test makes its refusal before the verdict rather than racing it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fake Codex writes the policy the driver judges, then names its thread, then says what the turn did. The driver starts the policy check at the thread and, finding the policy unsafe, kills the fake's process group at once — sometimes between the thread and the denial the test reads for. The refusal is then never written, and both the ledger and the result are empty. Nothing is lost by the connector: the session made no refusal to lose. The fake now holds the policy record back until its events are written, so the verdict cannot be reached before the denial is on the stream. The ordering is made rather than waited for, and no timeout is widened. The test also pins the verdict to the policy the fake applied. A check that gave up waiting for a record reads as ErrUnsafeMode too, and only one of the two is what this test is about. Under 128-way CPU load the old ordering fails 30 runs of 100, and 29 of 100 under -race; the new one, 0 of 200 and 0 of 100. --- internal/connector/driver/codex/codex_test.go | 20 ++++++++++++++++--- internal/connector/driver/codex/fake_test.go | 20 ++++++++++++++++++- 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index cf6b27791..7a3b2c8d0 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -1000,21 +1000,35 @@ func TestARefusalLoggedAfterTheOutputEndsIsStillRecorded(t *testing.T) { // A session stopped for running under a policy it was not asked to run under // still reports the refusals it made: they are the ledger's and the result's. +// +// The refusal has to be made before the verdict, and that ordering is made +// rather than waited for: the fake writes the policy the check reads only +// once its events are on the stream, so the kill the verdict brings cannot +// land on a fake that has not yet logged its denial. Left to race, it does +// not: under load the fake is killed between thread.started and the denial +// about a third of the time, and the refusal the test reads for is never +// written at all. func TestAnUnsafeSessionStillReportsItsRefusals(t *testing.T) { recorder := &drivertest.Refusals{} denial := `{"type":"item.completed","item":{"id":"item_9","type":"mcp_tool_call","server":"other","tool":"write","error":{"message":"MCP tool call requires approval, but approval policy is never"},"status":"failed"}}` unsafe := safeTurnContext() unsafe["approval_policy"] = "on-request" h := newHarness(t, scenario{ - TurnContext: unsafe, - Events: []string{`{"type":"turn.started"}`, denial, turnCompleted()}, + TurnContext: unsafe, + TurnContextAfterEvents: true, + Events: []string{`{"type":"turn.started"}`, denial, turnCompleted()}, }) cfg := h.config() cfg.Refusals = recorder s, result, err := h.run(context.Background(), cfg) require.ErrorIs(t, err, driver.ErrUnsafeMode) + // The verdict is the policy the fake applied, not a check that gave up + // waiting for one: both read as ErrUnsafeMode, and only one of them is + // this test's subject. + require.ErrorContains(t, err, `Codex applied "on-request"`, + "the session was stopped for the policy it ran under") waitDone(t, s) - assert.Len(t, recorder.Recorded(), 1) + assert.Len(t, recorder.Recorded(), 1, "the ledger holds the refusal the session made before its verdict") assert.Len(t, result.Refusals, 1, "the result carries what the ledger carries") } diff --git a/internal/connector/driver/codex/fake_test.go b/internal/connector/driver/codex/fake_test.go index c5a92d742..be04c2ca8 100644 --- a/internal/connector/driver/codex/fake_test.go +++ b/internal/connector/driver/codex/fake_test.go @@ -37,6 +37,12 @@ type scenario struct { // OldTurnContext is written before the prompt is read, as an earlier // turn of a resumed thread would be. OldTurnContext map[string]any `json:"old_turn_context"` + // TurnContextAfterEvents holds the turn_context record back until the + // events are written. The policy check starts at thread.started and ends + // an unsafe session the moment it can read the record, so a test about + // what a session said before that verdict orders the two rather than + // racing them. + TurnContextAfterEvents bool `json:"turn_context_after_events"` // Events are written to stdout after thread.started. Events []string `json:"events"` // NoThread skips thread.started. @@ -155,7 +161,10 @@ func fakeCodex() int { } appendRecord(rollout, "session_meta", map[string]any{"id": sc.Thread}) - if sc.TurnContext != nil { + writeTurnContext := func() { + if sc.TurnContext == nil { + return + } tc := map[string]any{} for k, v := range sc.TurnContext { tc[k] = v @@ -167,12 +176,21 @@ func fakeCodex() int { _ = json.Unmarshal([]byte(strings.ReplaceAll(string(raw), "$CWD", obs.Cwd)), &tc) appendRecord(rollout, "turn_context", tc) } + if !sc.TurnContextAfterEvents { + writeTurnContext() + } if !sc.NoThread { fmt.Printf(`{"type":"thread.started","thread_id":%q}`+"\n", sc.Thread) } for _, e := range sc.Events { fmt.Println(e) } + if sc.TurnContextAfterEvents { + // The policy the driver judges is readable only once everything this + // turn had to say is on the stream: the verdict, and the kill it + // brings, cannot land on a fake that has not finished speaking. + writeTurnContext() + } if sc.CloseStdout { _ = os.Stdout.Close() }