Four driver failures on CI: a crash in the connector, and three checks that read too early

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)
@@ -1382,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 {
@@ -1392,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)
 	}
 }
 

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()

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

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")
 }
 
@@ -1042,6 +1056,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.

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()
 	}