diff --git a/internal/connector/driver/acp/acp_test.go b/internal/connector/driver/acp/acp_test.go index 2ab1a4d96..1ace24fda 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) @@ -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) } } 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() 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..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") } @@ -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. 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() }