From 4ed6136c408721730f24635c7843fd1737327d92 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:26:33 +0200 Subject: [PATCH 1/5] A refusal is read on one goroutine and written on another MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A refusal Codex wrote could be thrown away: recorded nowhere, not merely missing from a turn's result. The read end of the worker's output is released when the session stops waiting for it, and closing it discards whatever the worker wrote that nobody has parsed. Writing a refusal to the ledger is allowed ten seconds and ran on the goroutine doing that parsing, so there was no bound that could be put on the reading. A clock runs out inside a write and takes the pipe away with the worker's next refusal still in it. A bound that waits for the pipe to fall idle never runs out at all against a descendant that keeps writing. A count of bytes is both at once: too permissive on time, because a descendant can drip them out forever, and too strict on volume, because a worker inherits the pipe descriptor and may enlarge it beyond any size assumed for it. I tried the byte count on the way here and it was wrong in both directions. They were never one bound. Waiting and following are different questions, and the reason neither could be answered was that the reader did the slow work itself. So the reader only reads, and a scribe writes. A reader with nothing arriving calls the pipe empty after a window of its own, which is how a stop ordinarily ends. A reader that keeps being handed output follows it only so long, which is the one case that has no other ending. The second is absolute from the asking, and that is honest only because the reader now does nothing slow between reads. What the split costs is named in driver.go rather than left to be found. A refusal is in memory between being read and being written, and a connector that crashes there loses it. It survives everything else: the queue is drained before the session's updates close, which is where the dispatcher settles what the ledger would not take, and the update for a refusal is emitted by the scribe after its write lands, so the contract's own order — recorded before the update — holds exactly rather than being narrowed. The queue blocks the reader and never drops. Before the reader has been asked to stop that is backpressure onto a live worker, which is what a slow ledger already did to this driver when both ran on one goroutine. After the asking it would be fatal, because the clock would run while the reader sat on a full queue, so after the asking the reader never waits and the queue grows instead — a burst at the end of a session, bounded by what one drain can produce. Held past every clock in this path, with a refusal sitting in the pipe the whole time, this fails 10 runs of 10 on main, losing it from both the ledger and the result. Eight more properties are held by tests that each go red when the thing they exist for is deleted. --- internal/connector/driver/codex/codex.go | 57 +++++- internal/connector/driver/codex/codex_test.go | 69 +++++++ internal/connector/driver/codex/scribe.go | 177 +++++++++++++++++ .../connector/driver/codex/scribe_test.go | 174 +++++++++++++++++ internal/connector/driver/driver.go | 21 +- internal/connector/driver/driver_test.go | 99 ++++++++++ internal/connector/driver/worker.go | 179 ++++++++++++++++-- internal/connector/driver/worker_other.go | 2 + 8 files changed, 753 insertions(+), 25 deletions(-) create mode 100644 internal/connector/driver/codex/scribe.go create mode 100644 internal/connector/driver/codex/scribe_test.go diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index 2c62eb86b..9bce2eab7 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -333,6 +333,11 @@ func (d *Driver) start(ctx context.Context, cfg driver.SessionConfig, resumeID s updates: make(chan driver.Update, 256), readerEnd: make(chan struct{}), } + s.scribe = newScribe(s.writeRefusal, s.emit) + // The worker's output is released when this session has read it, not on + // a clock, and the reader promises to do nothing slow between reads — + // which is the scribe's whole reason for existing. + worker.ReadingDone(s.readerEnd) go s.read() //nolint:contextcheck // the reader outlives the start's context: it runs as long as the worker does return s, nil } @@ -443,6 +448,7 @@ type session struct { updates chan driver.Update readerEnd chan struct{} + scribe *scribe mu sync.Mutex id string @@ -591,7 +597,14 @@ func (s *session) Cancel(context.Context) error { if t == nil { return nil } - go s.worker.Terminate(s.grace) + // The reader is asked to stop as well, as Close does. Ending the worker + // is not the end of its output when a descendant outside the group is + // holding the pipe open: the end of file never comes, and a canceled + // turn that only the reader can finish would never be finished at all. + go func() { + s.worker.Terminate(s.grace) + s.readerDone() + }() return nil } @@ -614,6 +627,8 @@ func (s *session) Close() error { } s.worker.Terminate(s.grace) s.readerDone() + // Close says the ledger has every refusal this session read. + s.scribe.close() return nil } @@ -629,7 +644,14 @@ func (s *session) readerDone() { select { case <-s.readerEnd: case <-time.After(s.grace): - s.worker.CloseStdout() + // A reader that never ends means a descendant outside the worker's + // group is holding the output open, so the end of file never comes. + // It is asked to stop, not cut off: it reads what is in the pipe + // first and ends within the drain budget. From here it must never + // be made to wait handing a refusal over, or that budget would run + // while it sat on a full queue. + s.scribe.noWaiting() + s.worker.StopReading() <-s.readerEnd } } @@ -713,10 +735,14 @@ func (s *session) closeUpdates() { // 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 an update emitted after this is + // The updates channel closes last, and the scribe is drained before + // it: the dispatcher settles what the ledger would not take once the + // updates are over, so every refusal the reader read has to have + // been written by then. Finishing the turn still emits — a refusal + // read from stderr — and an update emitted after the close is // dropped rather than sent. defer s.closeUpdates() + defer s.scribe.close() s.mu.Lock() s.ended = true t := s.turn @@ -1035,12 +1061,25 @@ func (s *session) refused(key, id, tool string, kind driver.ToolKind) { if !first { return } - if s.recorder != nil { - // The recorder owns what happens when the ledger refuses the write; - // the refusal happened either way. - _ = s.recorder.RecordRefusal(context.Background(), refusal) + // The write and the update it precedes both belong to the scribe: the + // reader hands them over and goes back to reading, which is what lets + // the bound on its reading be a clock. See scribe.go. + s.scribe.hand(pending{ + refusal: refusal, + update: driver.Update{Kind: driver.UpdatePermission, ToolCallID: id, Tool: tool, ToolKind: kind, Allowed: false}, + }) +} + +// writeRefusal is the ledger write itself, and the one slow thing this +// driver does with a refusal. +func (s *session) writeRefusal(refusal driver.Refusal) { + if s.recorder == nil { + return } - s.emit(driver.Update{Kind: driver.UpdatePermission, ToolCallID: id, Tool: tool, ToolKind: kind, Allowed: false}) + // The recorder owns what happens when the ledger refuses the write; the + // refusal happened either way. + //nolint:contextcheck // a refusal's write is not any caller's to cancel + _ = s.recorder.RecordRefusal(context.Background(), refusal) } func (s *session) turnCompleted(e event) { diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index d5fa4674a..47db16c5b 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -1146,6 +1146,10 @@ func TestAnUnsafeVerdictWaitsForWhatTheWorkerPutOnItsStream(t *testing.T) { got := <-answers require.ErrorIs(t, got.err, driver.ErrUnsafeMode, "an unsafe session is reported as unsafe, whoever gets to the turn first") require.ErrorContains(t, got.err, `Codex applied "on-request"`) + // The ledger is whole when the session is closed: the writes are + // not on the reader any more, and the turn's end is not where + // they are promised — the session's updates are. + require.NoError(t, s.Close()) assert.Len(t, ledger.Recorded(), 2, "both refusals reach the ledger") assert.Len(t, got.result.Refusals, 2, "the result carries what the ledger carries, and carries %d of the %d recorded", @@ -1154,6 +1158,71 @@ func TestAnUnsafeVerdictWaitsForWhatTheWorkerPutOnItsStream(t *testing.T) { } } +// A ledger that takes its time does not cost the worker its last lines. +// +// Writing a refusal is allowed ten seconds. While that ran on the goroutine +// reading the worker's output there was no bound that could be put on the +// reading: a clock runs out inside a write and takes the pipe away with the +// worker's next refusal still in it — recorded nowhere, not merely missing +// from a result — and a bound that waits for the pipe to fall idle never +// runs out at all against a descendant that keeps writing. +// +// The writes are a goroutine of their own now, so the reading is only ever +// reading. The hold here outlasts every clock in this path, with a refusal +// sitting in the pipe the whole time, and it is still read and still +// written. +func TestALedgerThatTakesItsTimeDoesNotCostTheWorkerItsLastLines(t *testing.T) { + ledger := &heldLedger{writing: make(chan struct{}, 1), release: make(chan struct{})} + denial := func(id string) string { + return `{"type":"item.completed","item":{"id":"` + id + `","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" + late := filepath.Join(t.TempDir(), "say-the-second") + h := newHarness(t, scenario{ + TurnContext: unsafe, + TurnContextAfterEvents: true, + Events: []string{`{"type":"turn.started"}`, denial("item_1")}, + // Said once the ledger is holding the first, so the second is in the + // pipe and nowhere else: the reader cannot have taken it into its + // own buffer, because it was not there to take. + LateEvents: []string{denial("item_2")}, + LateAfter: late, + Hang: true, + }) + cfg := h.config() + cfg.Refusals = ledger + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + + answers := make(chan driver.PromptResult, 1) + go func() { + result, _ := s.Prompt(context.Background(), "Task 1. Event 2.") + answers <- result + }() + + <-ledger.writing + require.NoError(t, os.WriteFile(late, nil, 0o600)) + held := make(chan struct{}) + go func() { + defer close(held) + // Longer than the grace and the drain budget together, and well + // inside the ten seconds a refusal's write is allowed. + time.Sleep(6 * time.Second) + close(ledger.release) + }() + + result := <-answers + <-held + require.NoError(t, s.Close()) + assert.Len(t, ledger.Recorded(), 2, + "the refusal that arrived while the ledger was writing is still read, and the ledger holds %d", len(ledger.Recorded())) + assert.Len(t, result.Refusals, 2, + "the result carries what the ledger carries, and carries %d", len(result.Refusals)) +} + // A session closed under a turn still reports the refusals that turn made. // A worker that has stopped reading its input leaves the prompt's write // blocked, and closing the session fails that write: the turn is then diff --git a/internal/connector/driver/codex/scribe.go b/internal/connector/driver/codex/scribe.go new file mode 100644 index 000000000..1195e09dd --- /dev/null +++ b/internal/connector/driver/codex/scribe.go @@ -0,0 +1,177 @@ +package codex + +import ( + "sync" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" +) + +// A refusal is read on one goroutine and written on another +// +// Writing a refusal to the ledger is allowed ten seconds and used to run on +// the goroutine reading the worker's output. That made the reading +// unboundable: the read end of the pipe is released when the session stops +// waiting for it, and any bound on that — a clock, a count of bytes — is +// either spent by a ledger write, taking the pipe away with the worker's +// next refusal still in it, or not spent by a descendant writing without +// end, which then holds the session open for as long as it lives. +// +// So the reader only reads, and the scribe only writes. The reader's promise +// to driver.Worker.ReadingDone — that it does nothing slow between reads — +// is what lets the drain bound be a clock and mean what it says. +// +// # What the scribe owes the contract +// +// A refusal is recorded before the update for it is emitted (driver.go). The +// scribe emits it, after its write lands, so that ordering holds exactly +// rather than being narrowed. What is traded is the other half — that a +// refusal is never held only in a session's memory — for the window between +// the reader handing one over and the scribe writing it. A connector crash +// there loses it; a worker exiting, a turn cut short, and a session closing +// do not, because the queue is drained before the session's updates close, +// which is where the dispatcher settles what the ledger would not take. +// +// # When the queue is full +// +// It blocks the reader, and it never drops. Dropping would be the loss this +// exists to end, wearing a new hat. +// +// Blocking is only safe before the reader has been asked to stop, and there +// it is right: the pipe fills, the worker blocks writing, and an agent +// producing refusals faster than the ledger takes them is slowed to the +// ledger's pace. That is what a slow ledger already did to this driver when +// both ran on one goroutine, so it is not a new way to fail. +// +// After the asking it would be fatal — the drain's clock would run while the +// reader sat on a full queue, and the pipe would be abandoned with the +// worker's output still in it — so after the asking the reader never waits. +// It appends, and the queue grows past its mark. That growth is bounded by +// what one drain can produce, which the drain budget bounds in time, and it +// is a burst at the end of a session rather than a leak. +// +// The mark is not derived from the pipe's size or the scanner's buffer. +// Neither is a bound the code has: a worker inherits the pipe descriptor and +// may enlarge it up to the host's pipe-max-size, and the scanner is allowed +// to grow to 64 MiB. It is simply how many refusals are worth holding before +// a live worker is told to slow down. +const refusalMark = 256 + +// pending is a refusal the reader has read and the scribe has not yet +// written, with the update that is emitted once it has. +type pending struct { + refusal driver.Refusal + update driver.Update +} + +// scribe writes refusals to the ledger, one at a time, off the reader. +type scribe struct { + record func(driver.Refusal) + emit func(driver.Update) + + mu sync.Mutex + room *sync.Cond + work *sync.Cond + queue []pending + nowait bool // the reader has been asked to stop: never make it wait + closing bool + finished bool // the scribe has drained and gone + done chan struct{} + // inflight counts the writes being made on a caller's own goroutine, + // after the scribe has gone. + inflight sync.WaitGroup +} + +func newScribe(record func(driver.Refusal), emit func(driver.Update)) *scribe { + s := &scribe{record: record, emit: emit, done: make(chan struct{})} + s.room = sync.NewCond(&s.mu) + s.work = sync.NewCond(&s.mu) + go s.run() + return s +} + +// hand gives the scribe a refusal to write. It waits for room while the +// reader can afford to, and never once it cannot. +func (s *scribe) hand(p pending) { + s.mu.Lock() + for len(s.queue) >= refusalMark && !s.nowait && !s.closing && !s.finished { + s.room.Wait() + } + if s.finished { + // The tail: a refusal read from the worker's last word by an ending + // that is not the reader's, after the scribe has drained and gone. + // Written here rather than lost. The count is taken under the same + // lock that publishes finished, so a close cannot decide everything + // is written while this one is starting. + s.inflight.Add(1) + s.mu.Unlock() + defer s.inflight.Done() + s.writeNow(p) + return + } + s.queue = append(s.queue, p) + s.work.Signal() + s.mu.Unlock() +} + +// noWaiting says the reader has been asked to stop, so it must not be held +// up handing anything over: from here the queue grows rather than blocks. +func (s *scribe) noWaiting() { + s.mu.Lock() + s.nowait = true + s.room.Broadcast() + s.mu.Unlock() +} + +// close says there is no more, and waits for what there is to be written. +// Whoever calls it promises the reader is through. It is idempotent. +func (s *scribe) close() { + s.mu.Lock() + if !s.closing { + s.closing = true + s.room.Broadcast() + s.work.Broadcast() + } + s.mu.Unlock() + <-s.done + // A refusal handed over after the scribe had gone is written by whoever + // read it, and this waits for those. One discovered after this returns + // is still written, on its reader's goroutine — the same window in which + // an update emitted then is dropped. + s.inflight.Wait() +} + +func (s *scribe) run() { + defer close(s.done) + for { + s.mu.Lock() + for len(s.queue) == 0 && !s.closing { + s.work.Wait() + } + if len(s.queue) == 0 { + // Published under the lock a hand takes, so no refusal can be + // queued to a scribe that has decided it is done. + s.finished = true + s.room.Broadcast() + s.mu.Unlock() + return + } + next := s.queue[0] + s.queue = s.queue[1:] + s.room.Signal() + s.mu.Unlock() + + // Outside the lock: the write is the slow thing, and the reader goes + // on reading while it happens. + s.record(next.refusal) + // The update comes after the write, which is the contract's own + // order: a refusal is recorded before the update for it is emitted. + s.emit(next.update) + } +} + +// writeNow writes a refusal on the caller's own goroutine, for one read +// after the scribe has gone. +func (s *scribe) writeNow(p pending) { + s.record(p.refusal) + s.emit(p.update) +} diff --git a/internal/connector/driver/codex/scribe_test.go b/internal/connector/driver/codex/scribe_test.go new file mode 100644 index 000000000..0505bff90 --- /dev/null +++ b/internal/connector/driver/codex/scribe_test.go @@ -0,0 +1,174 @@ +//go:build unix + +package codex + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" +) + +// A refusal's update is emitted after its write has landed, never before. +// That is the driver's contract — the recorder is called before the update +// is emitted — and it is the half of it the scribe keeps exactly rather than +// narrows. Emitting at hand-off would publish a refusal the ledger has not +// taken. +func TestTheUpdateComesAfterTheWrite(t *testing.T) { + var mu sync.Mutex + var order []string + writing := make(chan struct{}) + release := make(chan struct{}) + s := newScribe( + func(driver.Refusal) { + mu.Lock() + order = append(order, "write") + mu.Unlock() + close(writing) + <-release + }, + func(driver.Update) { + mu.Lock() + order = append(order, "update") + mu.Unlock() + }, + ) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "one"}}) + + // The write has begun and the update must not have been emitted. + <-writing + mu.Lock() + assert.Equal(t, []string{"write"}, order, "the update waits for the write") + mu.Unlock() + + close(release) + s.close() + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []string{"write", "update"}, order) +} + +// Handing a refusal over never loses it, whatever the scribe is doing — +// including after it has drained and gone, when whoever read it writes it. +func TestARefusalHandedOverAfterTheScribeHasGoneIsStillWritten(t *testing.T) { + var mu sync.Mutex + var written []string + s := newScribe( + func(r driver.Refusal) { + mu.Lock() + written = append(written, r.ToolCallID) + mu.Unlock() + }, + func(driver.Update) {}, + ) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "before"}}) + s.close() + s.hand(pending{refusal: driver.Refusal{ToolCallID: "after"}}) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []string{"before", "after"}, written, "nothing is dropped on either side of the close") +} + +// The queue holds a live worker back rather than dropping anything: past the +// mark, handing over waits for room. Dropping would be the loss this exists +// to end. +func TestAFullQueueHoldsTheReaderBack(t *testing.T) { + release := make(chan struct{}) + first := make(chan struct{}) + var once sync.Once + s := newScribe( + func(driver.Refusal) { + once.Do(func() { close(first) }) + <-release + }, + func(driver.Update) {}, + ) + // One is taken and stuck in its write; the rest fill the queue to the + // mark, and the next has to wait. + s.hand(pending{refusal: driver.Refusal{ToolCallID: "stuck"}}) + <-first + for i := range refusalMark { + s.hand(pending{refusal: driver.Refusal{ToolCallID: string(rune('a' + i%26))}}) + } + + waited := make(chan struct{}) + go func() { + defer close(waited) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "over"}}) + }() + select { + case <-waited: + t.Fatal("the queue took more than its mark without holding the reader back") + case <-time.After(200 * time.Millisecond): + } + + close(release) + <-waited + s.close() +} + +// Once the reader has been asked to stop it is never held back, because the +// clock on its drain is running: a reader waiting for room would spend that +// clock and abandon the pipe with the worker's output still in it. +func TestAReaderAskedToStopIsNeverHeldBack(t *testing.T) { + release := make(chan struct{}) + first := make(chan struct{}) + var once sync.Once + s := newScribe( + func(driver.Refusal) { + once.Do(func() { close(first) }) + <-release + }, + func(driver.Update) {}, + ) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "stuck"}}) + <-first + s.noWaiting() + + handed := make(chan struct{}) + go func() { + defer close(handed) + for range refusalMark * 3 { + s.hand(pending{refusal: driver.Refusal{ToolCallID: "more"}}) + } + }() + select { + case <-handed: + case <-time.After(10 * time.Second): + t.Fatal("a reader that has been asked to stop was held back handing refusals over") + } + + close(release) + s.close() +} + +// Everything handed over is written by the time the scribe is closed, which +// is what the session leans on: it closes the scribe before the updates, +// because that is where the dispatcher settles what the ledger would not +// take. +func TestClosingTheScribeWritesEverythingHandedOver(t *testing.T) { + var mu sync.Mutex + count := 0 + s := newScribe( + func(driver.Refusal) { + mu.Lock() + count++ + mu.Unlock() + }, + func(driver.Update) {}, + ) + const handed = 500 + for range handed { + s.hand(pending{refusal: driver.Refusal{ToolCallID: "one"}}) + } + s.close() + + mu.Lock() + defer mu.Unlock() + require.Equal(t, handed, count, "the close is what makes the ledger whole") +} diff --git a/internal/connector/driver/driver.go b/internal/connector/driver/driver.go index 437e52126..a41cafcf0 100644 --- a/internal/connector/driver/driver.go +++ b/internal/connector/driver/driver.go @@ -65,9 +65,19 @@ // recorded in the ledger, once, at the moment the driver answers the request // — or, for an agent that answers its own requests under a mode the driver // froze (claude -p), at the moment the driver first reads that it was -// refused. It is never held only in a session's memory, because a worker that -// exits before its result, a connector that crashes mid-turn, and a turn cut -// short by a deadline all end the session that memory lives in. +// refused. +// +// A driver may hand the write to a goroutine of its own rather than make it +// where it read the refusal, and the codex driver does: a ledger write is +// allowed ten seconds, and a reader that spends them cannot be given a bound +// on its reading that means anything (driver/codex/scribe.go). What that +// costs is named here rather than left to be discovered. Between the reading +// and the write the refusal is in memory, and a connector that crashes there +// loses it. What it does NOT cost is the rest: the queue is drained before +// the session's updates close, so a worker that exits before its result, a +// turn cut short by a deadline, and a session closed under a turn all still +// find it written — and the update for it is emitted by the writer, after +// the write, so the order below holds exactly. // // 1. The driver calls SessionConfig.Refusals.RecordRefusal before it sends // its answer to the agent, or before it emits the update for a refusal @@ -91,7 +101,10 @@ // recorder could not write, and the ended attempt's count is final. The // session's updates are drained before the attempt is released, and the // recorder is called before an update is emitted, so a worker that exits -// between a refusal and its result has already recorded it. +// between a refusal and its result has already recorded it. That holds +// for a driver that writes on a goroutine of its own too: it is the +// writer that emits, once the write has landed, and its queue is drained +// before the updates close. // // Once-ness is the driver's (a set of tool call ids per session), not a key in // the ledger: it holds for as long as a session lives, which is as long as a diff --git a/internal/connector/driver/driver_test.go b/internal/connector/driver/driver_test.go index 71740bede..e1be6d5d0 100644 --- a/internal/connector/driver/driver_test.go +++ b/internal/connector/driver/driver_test.go @@ -5,6 +5,7 @@ package driver import ( "context" "errors" + "io" "os" "os/exec" "path/filepath" @@ -339,3 +340,101 @@ func TestALaterGroupSignalIsNotSentToAGroupTheRecordNoLongerOwns(t *testing.T) { require.NoError(t, signalRecordedGroup(p, syscall.SIGKILL)) assert.Eventually(t, func() bool { return !GroupMembersRemain(p) }, 5*time.Second, 20*time.Millisecond) } + +// A stop does not discard what is in the pipe. Nothing closes the read end +// under a reader, because closing throws away whatever the worker wrote that +// nobody has parsed. Nothing here is concurrent and nothing waits: the lines +// are in the pipe before the stop, so what is asked is only whether the stop +// destroys them. +func TestAStopDoesNotDiscardWhatIsInThePipe(t *testing.T) { + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + t.Cleanup(func() { _ = readEnd.Close(); _ = writeEnd.Close() }) + out := &output{f: readEnd} + + _, err = writeEnd.Write([]byte("one\ntwo\n")) + require.NoError(t, err) + // Asked twice, and by two callers: a close and a cancel can both ask. + out.stop() + out.stop() + + buf := make([]byte, 64) + n, err := out.Read(buf) + require.NoError(t, err) + assert.Equal(t, "one\ntwo\n", string(buf[:n]), "what was in the pipe when the stop came") + + n, err = out.Read(buf) + assert.Zero(t, n) + assert.ErrorIs(t, err, io.EOF, "and then the reader ends, rather than waiting on a pipe nobody will close") +} + +// The drain budget is absolute from the asking, not a window that begins +// again after every read. A descendant that writes a single byte just before +// each window expires makes progress forever without ever letting the pipe +// fall idle, and a per-read window would follow it for as long as it lived — +// hours, for as much output as any byte count would allow. +func TestADescendantDrippingOutputCannotHoldTheReader(t *testing.T) { + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + t.Cleanup(func() { _ = readEnd.Close(); _ = writeEnd.Close() }) + out := &output{f: readEnd} + + dripping := make(chan struct{}) + go func() { + defer close(dripping) + for { + // Just often enough that no window of the reader's own ever + // finds the pipe empty. + time.Sleep(drainWindow / 2) + if _, werr := writeEnd.Write([]byte("x")); werr != nil { + return + } + } + }() + t.Cleanup(func() { _ = readEnd.Close(); <-dripping }) + + out.stop() + started := time.Now() + ended := make(chan error, 1) + go func() { + buf := make([]byte, 4<<10) + for { + if _, rerr := out.Read(buf); rerr != nil { + ended <- rerr + return + } + } + }() + select { + case rerr := <-ended: + assert.ErrorIs(t, rerr, io.EOF, "the reading ends, rather than following a descendant forever") + assert.Less(t, time.Since(started), 4*drainBudget, "and it ends on the budget, not on the descendant") + case <-time.After(30 * time.Second): + t.Fatal("a reader asked to stop never ended while a descendant dripped output into the pipe") + } +} + +// The deadline on the pipe has one owner. Asking a reader to stop sets a +// flag and touches nothing else, so it cannot cut short a window a read has +// opened. That is a property of who writes what, not of a schedule, so it is +// checked as one: a deadline set here survives any number of stops. +func TestAStopDoesNotTouchTheReadDeadline(t *testing.T) { + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + t.Cleanup(func() { _ = readEnd.Close(); _ = writeEnd.Close() }) + out := &output{f: readEnd} + + const mine = 300 * time.Millisecond + require.NoError(t, readEnd.SetReadDeadline(time.Now().Add(mine))) + out.stop() + out.stop() + + started := time.Now() + buf := make([]byte, 64) + n, err := readEnd.Read(buf) // the file itself, not through output + waited := time.Since(started) + + assert.Zero(t, n) + require.ErrorIs(t, err, os.ErrDeadlineExceeded) + assert.GreaterOrEqual(t, waited, mine/2, "the deadline in force is the one set here, not one a stop installed") +} diff --git a/internal/connector/driver/worker.go b/internal/connector/driver/worker.go index 164f92501..25cdff586 100644 --- a/internal/connector/driver/worker.go +++ b/internal/connector/driver/worker.go @@ -11,6 +11,7 @@ import ( "os/exec" "strings" "sync" + "sync/atomic" "syscall" "time" ) @@ -19,6 +20,112 @@ import ( // pipes a stray descendant still holds. const pipeWaitDelay = 2 * time.Second +// drainWindow is how long a reader that has been asked to stop gives the +// pipe to produce what is already on its way, idleWindow is how often a +// reader with nothing looks up to see whether it has been asked, and +// drainBudget is the longest it goes on reading after being asked. The +// windows are opened by the reader for itself: a read that finds nothing in +// one is a pipe with nothing in it. +// +// The two bound different things, and both are needed. +// +// drainWindow bounds WAITING: how long a reader sits with nothing arriving +// before it calls the pipe empty. That is how a stop ordinarily ends, within +// tens of milliseconds, and it needs no clock beyond itself. +// +// drainBudget bounds FOLLOWING: how long a reader goes on taking output that +// keeps arriving. Only one thing produces that after a worker is dead — a +// descendant outside its group writing without stop — and it is the one case +// with no other ending, so a window that begins again after every read is no +// bound at all: a descendant writing a byte before each one expires would +// hold a session open for hours. This one is absolute from the asking. +// +// Absolute is only honest because a reader registered here does nothing slow +// between reads, so the clock cannot run while the reader is working rather +// than waiting. That is a promise, and it is the whole reason this package's +// drivers write to the ledger on a goroutine of their own: a reader that +// wrote to a database here would be inside a write when the clock ran out, +// and would come back to find its own pipe closed with the worker's next +// line still in it. See Worker.ReadingDone. +// +// A count of bytes cannot stand in for either. It is too permissive on time, +// because a descendant can drip them out forever, and too strict on volume, +// because a worker inherits the pipe descriptor and may enlarge it beyond +// any size assumed here. +const ( + drainWindow = 50 * time.Millisecond + idleWindow = 500 * time.Millisecond + drainBudget = 2 * time.Second +) + +// output is the read end of a worker's pipe, and it belongs to whoever reads +// it for as long as they are reading. Nothing closes it underneath them: +// closing discards whatever the worker wrote that nobody has parsed yet, and +// there is no size that could be assumed safe — a worker inherits the +// descriptor and may enlarge the pipe itself, up to the host's +// pipe-max-size. +// +// So a reader is asked to stop rather than cut off, and asking touches +// nothing but a flag. The deadline on the pipe has one owner, the reader, +// which is what makes a window mean what it says: nobody else can shorten +// one, and there is no moment in which a deadline is in force whose reason +// is not yet visible. +// +// A worker is ended before its reader is asked, so everything the worker +// wrote is in the pipe by then. What is given up, past the budget, is only +// what a descendant outside the worker's group goes on writing — which the +// connector had already decided not to wait for, and which is the one case +// that has no other ending. +type output struct { + f *os.File + // stopped is whether the reader has been asked to stop. One field, + // written by one call, so there is no moment in which half the state is + // visible. + stopped atomic.Bool + // stopAt is when this reader first saw that it had been asked, set by + // the reader itself. Only Read touches it, and a pipe has one reader. + stopAt time.Time +} + +func (o *output) Read(p []byte) (int, error) { + for { + stopped := o.stopped.Load() + window := idleWindow + if stopped { + if o.stopAt.IsZero() { + o.stopAt = time.Now() + } + if time.Since(o.stopAt) >= drainBudget { + // Still arriving, and no longer waited for. + return 0, io.EOF + } + window = drainWindow + } + if err := o.f.SetReadDeadline(time.Now().Add(window)); err != nil { + return 0, err + } + n, err := o.f.Read(p) + switch { + case n > 0: + return n, nil + case !errors.Is(err, os.ErrDeadlineExceeded): + return n, err + case stopped: + // A window this read opened, and nothing came. + return 0, io.EOF + } + // Nothing yet, and nobody has asked. Look again. + } +} + +// stop asks the reader to end once it has what is there, and no later than +// the drain budget. It writes one field and nothing else — in particular it +// does not reach for the deadline, so it cannot cut short a window a read +// has opened — and it says the same thing however many times it is called. +func (o *output) stop() { o.stopped.Store(true) } + +func (o *output) close() { _ = o.f.Close() } + // # One owner, one release point // // This is the connector's rule for a task's process tree and its ledger @@ -178,13 +285,16 @@ type Worker struct { cmd *exec.Cmd process Process stdin io.WriteCloser - stdout *os.File + stdout *output stderr *tailBuffer done chan struct{} exit Exit killOnce sync.Once releaseOnce sync.Once + + readingMu sync.Mutex + reading <-chan struct{} } // StartWorker launches cmd through cfg's launcher, in cfg's scope, as a new @@ -244,7 +354,7 @@ func StartWorker(ctx context.Context, cfg SessionConfig, cmd Command) (*Worker, return nil, fmt.Errorf("%w: %w", ErrNotStarted, err) } ec.Stdout = writeEnd - w.stdout = readEnd + w.stdout = &output{f: readEnd} if err := ec.Start(); err != nil { // exec.Cmd.Start returns an error only when no process was created: // a missing binary, a bad directory, a failed fork. @@ -308,11 +418,46 @@ func (w *Worker) Stdin() io.WriteCloser { return w.stdin } // Stdout is the worker's standard output. Read it to end of file. func (w *Worker) Stdout() io.Reader { return w.stdout } -// CloseStdout closes the worker's output: a reader blocked on it returns, and -// the descriptor is released. -// For a worker that is gone while a descendant that left its group still -// holds the pipe. -func (w *Worker) CloseStdout() { _ = w.stdout.Close() } +// ReadingDone hands the Worker the signal that its output has been read to +// the end, so the read end of the pipe is released with the reading rather +// than on a clock. +// +// The descriptor is the Worker's to release; whether the reading is over is +// the reader's to say. Without this the Worker has to guess, and a guess is +// destructive: closing the read end discards whatever the worker wrote that +// nobody has parsed yet. +// +// A Worker with no reader registered keeps the clock, because nothing else +// would release it. +// +// Registering is two promises about the goroutine that reads: that it closes +// the channel, and that it does nothing slow between reads. The second is +// what lets the drain budget be a clock — see drainBudget — and it is why a +// driver's ledger writes belong on a goroutine of their own. +func (w *Worker) ReadingDone(done <-chan struct{}) { + w.readingMu.Lock() + defer w.readingMu.Unlock() + w.reading = done +} + +func (w *Worker) readingSignal() <-chan struct{} { + w.readingMu.Lock() + defer w.readingMu.Unlock() + return w.reading +} + +// StopReading asks the worker's reader to end once it has read what is in +// the pipe, and no later than the drain budget. For a worker that is gone +// while a descendant that left its group still holds the output open, so the +// end of file never comes: the reader is asked rather than having the pipe +// taken from under it, which would discard what the worker wrote and nobody +// has parsed. It says the same thing however many times it is called. +func (w *Worker) StopReading() { w.stdout.stop() } + +// CloseStdout closes the worker's output at once, discarding whatever is in +// the pipe: for releasing the descriptor once nobody is reading it. A reader +// that is still reading is asked to stop instead. +func (w *Worker) CloseStdout() { w.stdout.close() } // Done is closed once the process has exited and been reaped. func (w *Worker) Done() <-chan struct{} { return w.done } @@ -360,12 +505,22 @@ func (w *Worker) Terminate(grace time.Duration) { _ = w.cmd.Process.Kill() }) <-w.done - // The output pipe is the Worker's to release as well. Its reader gets the - // same bound Wait gives a stray descendant to finish draining what the - // worker wrote before it went, and then the descriptor is closed whether - // or not the reader closed it. + // The output pipe is the Worker's to release as well, once its reader is + // through it: a reader that says when it is done is waited for, because + // closing the descriptor under it would discard what the worker wrote + // and nobody has parsed. A worker nobody reads this way gets the same + // bound Wait gives a stray descendant, and then the descriptor is closed + // regardless — nothing else would release it. w.releaseOnce.Do(func() { - time.AfterFunc(pipeWaitDelay, w.CloseStdout) + reading := w.readingSignal() + if reading == nil { + time.AfterFunc(pipeWaitDelay, w.CloseStdout) + return + } + go func() { + <-reading + w.CloseStdout() + }() }) } diff --git a/internal/connector/driver/worker_other.go b/internal/connector/driver/worker_other.go index df6f39846..6eaf5d34c 100644 --- a/internal/connector/driver/worker_other.go +++ b/internal/connector/driver/worker_other.go @@ -23,6 +23,8 @@ func (*Worker) Process() Process { return Process{} } func (*Worker) Stdin() io.WriteCloser { return nil } func (*Worker) Stdout() io.Reader { return nil } func (*Worker) CloseStdout() {} +func (*Worker) StopReading() {} +func (*Worker) ReadingDone(<-chan struct{}) {} func (*Worker) Done() <-chan struct{} { return nil } func (*Worker) Exit() Exit { return Exit{} } func (*Worker) StderrTail(*Redactor) string { return "" } From 36fd490b6cefa4333296b3326c1f506b931a06ae Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:59:53 +0200 Subject: [PATCH 2/5] Admitting a late refusal and deciding the ledger is whole are one decision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four things the scribe had wrong, and the first is this pull request's own defect one layer up. A refusal handed over after the scribe had drained was written by whoever read it, and counted with a WaitGroup beside the lifecycle rather than inside it. So a close could Wait on a zero counter while another goroutine was entering that branch and adding to it — a positive Add racing a zero-count Wait, which the language forbids, and a close deciding the ledger was whole while a write was starting. Admission and completion are one decision under one lock now, and the barrier says what it promises: every refusal handed over so far is written when close returns, including the ones a caller wrote itself. It promises nothing about one handed over after it returns; that is still written, by whoever read it, and it is not waited for. Neither case is dropped. The queue was bounded before the reader is asked to stop and not after it. A drain bounded only by its clock is not bounded in memory: two seconds of reading from a worker refusing as fast as it can write is as much as the machine will take. There is a cap now, and at the cap the reader stops reading — which loses nothing that was read and abandons what was not, exactly as the clock was about to. The drain's clock started when a read next noticed the flag, not when the stop was asked for. A reader sitting in an idle window would not notice for up to that whole window, and the budget would then run from there — longer than the comment beside it promised. The time is taken at the asking now, before the flag that publishes it, and a read's window is never longer than what is left of the budget. And the recorder's own documentation still said RecordRefusal is called on the goroutine reading the agent's stream. It is not, in this driver, which is the whole point. Both places that said so now say which goroutine is the driver's choice and what an implementation may assume. I swept the claim rather than the code this time; the dispatcher's timeout comment said it too. Each is held by a test that goes red when the thing it exists for is deleted, including the clock, which is asked as the question it is — whether the time is taken at the asking — rather than timed, because the gap it guards only opens on a knife-edge and a test on one proves nothing. --- internal/connector/dispatcher.go | 6 +- internal/connector/driver/codex/codex.go | 6 ++ internal/connector/driver/codex/scribe.go | 76 ++++++++++++++---- .../connector/driver/codex/scribe_test.go | 79 +++++++++++++++++++ internal/connector/driver/driver.go | 10 ++- internal/connector/driver/driver_test.go | 33 ++++++++ internal/connector/driver/worker.go | 26 +++--- 7 files changed, 208 insertions(+), 28 deletions(-) diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index 004b9dbd4..222c2186c 100644 --- a/internal/connector/dispatcher.go +++ b/internal/connector/dispatcher.go @@ -1351,8 +1351,10 @@ type refusalRecorder struct { pending int } -// refusalWriteTimeout bounds a refusal's write, which runs on the goroutine -// reading the agent's stream. +// refusalWriteTimeout bounds a refusal's write. Which goroutine it runs on is +// the driver's: on most it is the one reading the agent's stream, and the +// codex driver writes on a goroutine of its own precisely because ten +// seconds is more than a reader can afford to spend. const refusalWriteTimeout = 10 * time.Second // RecordRefusal implements driver.RefusalRecorder. diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index 9bce2eab7..84eeb2efa 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -781,6 +781,12 @@ func (s *session) read() { scanner.Buffer(make([]byte, 64<<10), 64<<20) for scanner.Scan() { s.handle(scanner.Bytes()) + if s.scribe.enough() { + // A drain that has heard more refusals than it can hold. What + // was read is written; what was not is abandoned, which is what + // the drain's own clock was about to do. + break + } } // Drain what a scanner error left, so the process never blocks writing. _, _ = io.Copy(io.Discard, s.worker.Stdout()) diff --git a/internal/connector/driver/codex/scribe.go b/internal/connector/driver/codex/scribe.go index 1195e09dd..b7c953edd 100644 --- a/internal/connector/driver/codex/scribe.go +++ b/internal/connector/driver/codex/scribe.go @@ -54,7 +54,17 @@ import ( // may enlarge it up to the host's pipe-max-size, and the scanner is allowed // to grow to 64 MiB. It is simply how many refusals are worth holding before // a live worker is told to slow down. -const refusalMark = 256 +// +// refusalCap is where the growing stops. A drain bounded only by its clock +// is not bounded in memory: two seconds of reading from a descendant that +// refuses as fast as it can write is as much as the machine will take. At +// the cap the reader stops reading instead, which loses nothing that was +// read and abandons what was not — exactly what the clock would have done a +// moment later. +const ( + refusalMark = 256 + refusalCap = 4096 +) // pending is a refusal the reader has read and the scribe has not yet // written, with the update that is emitted once it has. @@ -75,16 +85,21 @@ type scribe struct { nowait bool // the reader has been asked to stop: never make it wait closing bool finished bool // the scribe has drained and gone - done chan struct{} - // inflight counts the writes being made on a caller's own goroutine, - // after the scribe has gone. - inflight sync.WaitGroup + heard bool // the queue reached its cap during a drain + // late counts the writes being made on a caller's own goroutine because + // the scribe had already gone. It lives under the same lock that + // publishes finished, so admitting one and deciding everything is + // written are a single decision rather than two that can cross. + late int + quiet *sync.Cond + done chan struct{} } func newScribe(record func(driver.Refusal), emit func(driver.Update)) *scribe { s := &scribe{record: record, emit: emit, done: make(chan struct{})} s.room = sync.NewCond(&s.mu) s.work = sync.NewCond(&s.mu) + s.quiet = sync.NewCond(&s.mu) go s.run() return s } @@ -99,20 +114,38 @@ func (s *scribe) hand(p pending) { if s.finished { // The tail: a refusal read from the worker's last word by an ending // that is not the reader's, after the scribe has drained and gone. - // Written here rather than lost. The count is taken under the same - // lock that publishes finished, so a close cannot decide everything - // is written while this one is starting. - s.inflight.Add(1) + // It is written here rather than lost, and counted under the same + // lock that published finished — so a close either waits for this + // one or has not begun, never decides the ledger is whole while it + // is starting. + s.late++ s.mu.Unlock() - defer s.inflight.Done() s.writeNow(p) + s.mu.Lock() + s.late-- + s.quiet.Broadcast() + s.mu.Unlock() return } s.queue = append(s.queue, p) + if s.nowait && len(s.queue) >= refusalCap { + // A drain that has heard as much as it can hold. Nothing read is + // lost; the reader is told to stop reading, which is what its own + // clock was about to do. + s.heard = true + } s.work.Signal() s.mu.Unlock() } +// enough reports that a drain filled the queue to its cap, so the reader +// should stop reading. Nothing that was read is dropped. +func (s *scribe) enough() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.heard +} + // noWaiting says the reader has been asked to stop, so it must not be held // up handing anything over: from here the queue grows rather than blocks. func (s *scribe) noWaiting() { @@ -133,11 +166,24 @@ func (s *scribe) close() { } s.mu.Unlock() <-s.done - // A refusal handed over after the scribe had gone is written by whoever - // read it, and this waits for those. One discovered after this returns - // is still written, on its reader's goroutine — the same window in which - // an update emitted then is dropped. - s.inflight.Wait() + // # What this barrier promises + // + // When it returns, every refusal handed over so far has been written: + // the ones the scribe took, and the ones a caller wrote itself because + // the scribe had already gone. Neither is dropped and neither races + // this, because admitting a late write and counting it down happen + // under the lock this waits on. + // + // It promises nothing about a refusal handed over AFTER it returns. One + // can be: an ending that is not the reader's may read the worker's last + // word later still. That refusal is written too, by whoever read it, on + // that goroutine — it is simply not waited for here, in the same window + // where an update emitted then is dropped rather than sent. + s.mu.Lock() + for s.late > 0 { + s.quiet.Wait() + } + s.mu.Unlock() } func (s *scribe) run() { diff --git a/internal/connector/driver/codex/scribe_test.go b/internal/connector/driver/codex/scribe_test.go index 0505bff90..91ddc6304 100644 --- a/internal/connector/driver/codex/scribe_test.go +++ b/internal/connector/driver/codex/scribe_test.go @@ -172,3 +172,82 @@ func TestClosingTheScribeWritesEverythingHandedOver(t *testing.T) { defer mu.Unlock() require.Equal(t, handed, count, "the close is what makes the ledger whole") } + +// The close waits for a late write that has already begun. A refusal read +// by an ending that is not the reader's can arrive after the scribe has +// drained and gone; whoever read it writes it, and the close either waits +// for that write or has not started — never decides the ledger is whole +// while one is running. +func TestClosingWaitsForALateWriteAlreadyBegun(t *testing.T) { + var mu sync.Mutex + var done []string + writing := make(chan struct{}) + release := make(chan struct{}) + s := newScribe( + func(r driver.Refusal) { + if r.ToolCallID == "late" { + close(writing) + <-release + } + mu.Lock() + done = append(done, r.ToolCallID) + mu.Unlock() + }, + func(driver.Update) {}, + ) + s.close() // the scribe has drained and gone + + handed := make(chan struct{}) + go func() { + defer close(handed) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "late"}}) + }() + <-writing // the late write has begun + + closed := make(chan struct{}) + go func() { + defer close(closed) + s.close() + }() + select { + case <-closed: + t.Fatal("the close decided the ledger was whole while a late write was running") + case <-time.After(200 * time.Millisecond): + } + + close(release) + <-closed + <-handed + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []string{"late"}, done, "and the write it waited for did happen") +} + +// A drain that hears more refusals than it can hold tells the reader to stop +// reading, rather than growing without limit. Nothing that was read is lost; +// what was not read is abandoned, which is what the drain's clock was about +// to do anyway. +func TestADrainThatFillsTheQueueTellsTheReaderToStop(t *testing.T) { + release := make(chan struct{}) + first := make(chan struct{}) + var once sync.Once + s := newScribe( + func(driver.Refusal) { + once.Do(func() { close(first) }) + <-release + }, + func(driver.Update) {}, + ) + s.hand(pending{refusal: driver.Refusal{ToolCallID: "stuck"}}) + <-first + s.noWaiting() + + require.False(t, s.enough(), "nothing has been heard yet") + for range refusalCap { + s.hand(pending{refusal: driver.Refusal{ToolCallID: "more"}}) + } + assert.True(t, s.enough(), "the reader is told to stop once the queue is full") + + close(release) + s.close() +} diff --git a/internal/connector/driver/driver.go b/internal/connector/driver/driver.go index a41cafcf0..c75d2b787 100644 --- a/internal/connector/driver/driver.go +++ b/internal/connector/driver/driver.go @@ -329,8 +329,14 @@ type Refusal struct { } // RefusalRecorder records a refusal at the moment a driver makes or observes -// it (see "Refusals" above). RecordRefusal must not block for long: a driver -// calls it on the goroutine that reads the agent's stream. +// it (see "Refusals" above). +// +// It is called once per refusal, never concurrently with itself, and an +// implementation needs no locking of its own for that. Which goroutine calls +// it is the driver's: most call it on the one reading the agent's stream, so +// it must not block for long — a driver that cannot afford that hands the +// write to a goroutine of its own, as codex does, and then emits the update +// for the refusal only once this has returned. type RefusalRecorder interface { RecordRefusal(ctx context.Context, r Refusal) error } diff --git a/internal/connector/driver/driver_test.go b/internal/connector/driver/driver_test.go index e1be6d5d0..dabbbae85 100644 --- a/internal/connector/driver/driver_test.go +++ b/internal/connector/driver/driver_test.go @@ -438,3 +438,36 @@ func TestAStopDoesNotTouchTheReadDeadline(t *testing.T) { require.ErrorIs(t, err, os.ErrDeadlineExceeded) assert.GreaterOrEqual(t, waited, mine/2, "the deadline in force is the one set here, not one a stop installed") } + +// The drain budget runs from the asking, not from when the reader next +// notices it. A reader with nothing arriving sits in an idle window, and a +// stop landing inside one is not seen until that window is out; a budget +// begun at the noticing would outlast what it promises by that much. +// +// This is a question about when the clock is taken, not about how long +// anything runs, so it is asked that way: after the asking, the time to +// measure from exists. Timing it would be a knife-edge — the reader notices +// as soon as anything arrives, so the gap only opens when output begins just +// as a window expires — and a test on a knife-edge proves nothing either +// way. +func TestTheDrainBudgetRunsFromTheAsking(t *testing.T) { + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + t.Cleanup(func() { _ = readEnd.Close(); _ = writeEnd.Close() }) + out := &output{f: readEnd} + + require.Zero(t, out.stopAt.Load(), "nothing has been asked yet") + before := time.Now() + out.stop() + after := time.Now() + + asked := out.stopAt.Load() + require.NotZero(t, asked, "the budget's clock is taken when the stop is asked for, not when a read next looks") + assert.False(t, time.Unix(0, asked).Before(before), "and it is taken then") + assert.False(t, time.Unix(0, asked).After(after)) + + // And it is the first asking that counts, so a second does not hand the + // reader a fresh budget. + out.stop() + assert.Equal(t, asked, out.stopAt.Load(), "a later stop does not restart the budget") +} diff --git a/internal/connector/driver/worker.go b/internal/connector/driver/worker.go index 25cdff586..bb27c41cf 100644 --- a/internal/connector/driver/worker.go +++ b/internal/connector/driver/worker.go @@ -82,9 +82,12 @@ type output struct { // written by one call, so there is no moment in which half the state is // visible. stopped atomic.Bool - // stopAt is when this reader first saw that it had been asked, set by - // the reader itself. Only Read touches it, and a pipe has one reader. - stopAt time.Time + // stopAt is when the stop was asked for, in Unix nanoseconds, written by + // stop before the flag that publishes it. The reader takes it from here + // rather than from when it next looks, so the budget runs from the + // asking as it says it does — a reader inside an idle window when the + // asking comes would otherwise start the clock up to that window late. + stopAt atomic.Int64 } func (o *output) Read(p []byte) (int, error) { @@ -92,14 +95,14 @@ func (o *output) Read(p []byte) (int, error) { stopped := o.stopped.Load() window := idleWindow if stopped { - if o.stopAt.IsZero() { - o.stopAt = time.Now() - } - if time.Since(o.stopAt) >= drainBudget { + left := drainBudget - time.Since(time.Unix(0, o.stopAt.Load())) + if left <= 0 { // Still arriving, and no longer waited for. return 0, io.EOF } - window = drainWindow + // Never past the budget: the last window is whatever is left of + // it, not a whole one begun at the end. + window = min(drainWindow, left) } if err := o.f.SetReadDeadline(time.Now().Add(window)); err != nil { return 0, err @@ -122,7 +125,12 @@ func (o *output) Read(p []byte) (int, error) { // the drain budget. It writes one field and nothing else — in particular it // does not reach for the deadline, so it cannot cut short a window a read // has opened — and it says the same thing however many times it is called. -func (o *output) stop() { o.stopped.Store(true) } +func (o *output) stop() { + // The time before the flag, so a reader that sees the flag always finds + // a time to measure from, and the budget runs from the asking. + o.stopAt.CompareAndSwap(0, time.Now().UnixNano()) + o.stopped.Store(true) +} func (o *output) close() { _ = o.f.Close() } From 1a2cccabf2f40d9fdf773040d42ae9cdb61b6e9a Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 10:37:39 +0200 Subject: [PATCH 3/5] A turn ends away from the reader, so the reader only reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reader was never the thin pump the bounds assumed. A turn ends by waiting — for the policy check's verdict, and for the worker to be gone so its stderr is whole — and both waits ran on the goroutine reading the worker's output. So the drain's clock could run out entirely inside one, and the reader would come back to a pipe it was no longer allowed to read, with the worker's output still in it. That is this branch's own defect, moved from the ledger write to the turn's ending. Both waits are now taken off the read path. They happen once, at the end of a turn, and for this driver the event that ends a turn is the last one a process sends, so nothing is left unread by moving them. The reader's own ending waits for them before it closes the scribe, because they read the worker's last word too. What is left on the reader is parsing, appending and handing over — so the drain budget is a clock over waiting, and the queue's cap counts what it says. Four more, all real, and the first two are guarantees this branch itself wrote and then broke. A late refusal is written on its reader's goroutine rather than the scribe's, and two endings can make one each — so the recorder could be called twice at once, against the promise in driver.go that it never is. Every write takes the same lock now, wherever it is made from. The queue's cap stopped the reader by breaking out of the scan loop, which skips whatever the scanner had already taken from the pipe: lines nobody would see again. It asks the worker to stop reading instead, and goes on parsing what it holds. I said that cap "loses nothing" when it lost things; it loses nothing now. Terminate waited on a registered reader without asking it to stop, so a descendant holding the output kept the descriptor — and everything waiting on the reading — open for as long as it lived. It asks first. And the stop's clock was kept as Unix nanoseconds and rebuilt with time.Unix, which throws away Go's monotonic reading: a wall clock moving forward ends a drain early with output still buffered, and moving back holds a session past its budget. It is a time.Time now, kept whole. Two bounds that were missing rather than broken, both pre-existing and neither caused by this change. A tool call id is the agent's, and the redactor takes things out of a string without making it shorter — so a refusal could keep a megabyte, on the turn, in the queue and in what the session remembers. And what it remembers had no bound at all, where the acp driver has one for exactly this. Both are bounded, and reaching the second ends the session rather than leave it unable to tell a repeat from a first. --- internal/connector/driver/codex/codex.go | 162 +++++++++++++----- internal/connector/driver/codex/codex_test.go | 71 +++++++- internal/connector/driver/codex/scribe.go | 26 ++- .../connector/driver/codex/scribe_test.go | 43 +++++ internal/connector/driver/driver_test.go | 16 +- internal/connector/driver/worker.go | 43 ++++- 6 files changed, 294 insertions(+), 67 deletions(-) diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index 84eeb2efa..f90616a8a 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -449,6 +449,9 @@ type session struct { updates chan driver.Update readerEnd chan struct{} scribe *scribe + // endings counts the turn endings running off the reader. Only the + // reader adds to it, and only the reader's own ending waits on it. + endings sync.WaitGroup mu sync.Mutex id string @@ -743,6 +746,13 @@ func (s *session) read() { // dropped rather than sent. defer s.closeUpdates() defer s.scribe.close() + // An ending already running off the reader is the one that finishes + // its turn, and it reads the worker's last word on its way. Waiting + // for it here is what keeps this from finishing the same turn as + // ended, and what keeps the scribe open until it has handed over + // whatever it read. Every ending is bounded — the policy check's + // verdict and the worker's exit both are — so this is too. + s.endings.Wait() s.mu.Lock() s.ended = true t := s.turn @@ -782,10 +792,12 @@ func (s *session) read() { for scanner.Scan() { s.handle(scanner.Bytes()) if s.scribe.enough() { - // A drain that has heard more refusals than it can hold. What - // was read is written; what was not is abandoned, which is what - // the drain's own clock was about to do. - break + // A drain that has heard as many refusals as it can hold. The + // pipe stops being read, which is what its own clock was about + // to do; the scanner's bufferful is still parsed, because + // breaking out here would drop lines that are no longer in the + // pipe for anyone else to find. + s.worker.StopReading() } } // Drain what a scanner error left, so the process never blocks writing. @@ -1054,10 +1066,23 @@ func refusedByApproval(message string) bool { // time its tool call id is seen (driver's "Refusals"). A refusal Codex logs // and gives no id gets the key the caller passes. func (s *session) refused(key, id, tool string, kind driver.ToolKind) { - refusal := driver.Refusal{ToolCallID: s.red.Sanitize(id), Tool: s.red.Sanitize(tool)} + // The id is the agent's, and the redactor takes things out of a string + // without making it shorter: an agent that names a tool call a megabyte + // long would otherwise have that megabyte kept on the turn, in the + // queue, and in what this session remembers having refused. + refusal := driver.Refusal{ToolCallID: cut(s.red.Sanitize(id), maxToolCallID), Tool: s.red.Sanitize(tool)} + key = cut(key, maxToolCallID+len("stderr:000:")) s.mu.Lock() first := !s.recorded[key] if first { + if len(s.recorded) >= maxRecorded { + // Past what a session can remember, a repeat cannot be told from + // a first, and recording one twice is worse than this: the + // session ends rather than go on guessing. + s.mu.Unlock() + s.endedOverfull() + return + } s.recorded[key] = true if s.turn != nil { s.turn.refusals = append(s.turn.refusals, refusal) @@ -1100,33 +1125,56 @@ func (s *session) turnCompleted(e event) { s.mu.Unlock() if canceled { // A cancel that won does not wait out the policy check either. - s.finishCanceled(t) + s.endTurn(func() { s.finishCanceled(t) }) return } - if err := s.verified(); err != nil { - s.finishUnsafe(t, err) - return - } - // Codex exits right after the turn it completed, and its stderr is whole - // only once it has: a refusal it logged and did not put on the stream is - // in the tail by then. - select { - case <-s.worker.Done(): - case <-time.After(s.grace): - } - s.stderrRefusals() - s.mu.Lock() - result := driver.PromptResult{Stop: driver.TurnEndTurn, Refusals: slices.Clone(t.refusals)} - if t.canceled { - // Only a cancel the connector asked for reads as canceled. - result.Stop = driver.TurnCanceled - } - s.mu.Unlock() - if e.Usage != nil { - result.Usage = driver.Usage{InputTokens: e.Usage.InputTokens, OutputTokens: e.Usage.OutputTokens} - s.emit(driver.Update{Kind: driver.UpdateUsage, Usage: &result.Usage}) - } - s.finish(t, result, nil) + s.endTurn(func() { + if err := s.verified(); err != nil { + s.finishUnsafe(t, err) + return + } + // Codex exits right after the turn it completed, and its stderr is + // whole only once it has: a refusal it logged and did not put on the + // stream is in the tail by then. + select { + case <-s.worker.Done(): + case <-time.After(s.grace): + } + s.stderrRefusals() + s.mu.Lock() + result := driver.PromptResult{Stop: driver.TurnEndTurn, Refusals: slices.Clone(t.refusals)} + if t.canceled { + // Only a cancel the connector asked for reads as canceled. + result.Stop = driver.TurnCanceled + } + s.mu.Unlock() + if e.Usage != nil { + result.Usage = driver.Usage{InputTokens: e.Usage.InputTokens, OutputTokens: e.Usage.OutputTokens} + s.emit(driver.Update{Kind: driver.UpdateUsage, Usage: &result.Usage}) + } + s.finish(t, result, nil) + }) +} + +// endTurn runs a turn's ending away from the reader, and remembers it so the +// reader's own ending waits for it. +// +// A turn ends by waiting: for the policy check's verdict, and for the worker +// to be gone so its stderr is whole. Neither is reading, and neither belongs +// on the goroutine whose job is to keep a pipe empty — a reader that waits +// is a reader that is not draining, and every bound on its reading is then a +// bound on something else. This is what lets the drain budget be a clock and +// the queue's cap be a count. +// +// The count is taken here, on the reader, before the ending starts, and +// waited for from the reader's own ending: one goroutine adds, the same one +// waits. +func (s *session) endTurn(ending func()) { + s.endings.Add(1) + go func() { + defer s.endings.Done() + ending() + }() } func (s *session) turnFailed() { @@ -1140,21 +1188,51 @@ func (s *session) turnFailed() { canceled := t.canceled s.mu.Unlock() if canceled { - s.finishCanceled(t) + s.endTurn(func() { s.finishCanceled(t) }) return } - // As after a completed turn: the stderr tail is whole once Codex exits. - select { - case <-s.worker.Done(): - case <-time.After(s.grace): - } - s.stderrRefusals() - refusals := s.refusalsOf(t) - if err := s.failedVerification(); err != nil { - s.finishUnsafe(t, err) - return + s.endTurn(func() { + // As after a completed turn: the stderr tail is whole once Codex + // exits. + select { + case <-s.worker.Done(): + case <-time.After(s.grace): + } + s.stderrRefusals() + refusals := s.refusalsOf(t) + if err := s.failedVerification(); err != nil { + s.finishUnsafe(t, err) + return + } + s.finish(t, driver.PromptResult{Refusals: refusals}, errors.New("codex: the turn failed")) + }) +} + +// maxToolCallID is how much of a tool call id a refusal keeps, and +// maxRecorded how many refusals a session remembers having recorded. The +// agent writes both, and neither is a session's to grow without end — the +// acp driver bounds the same two for the same reason. +// +// Reaching maxRecorded is not a bound that can be applied quietly: past it a +// repeat cannot be told from a first, and recording the same refusal twice +// would be worse than stopping. The session ends instead. +const ( + maxToolCallID = 256 + maxRecorded = 4096 +) + +// cut shortens a string the agent wrote to what this driver keeps of it. +func cut(s string, keep int) string { + if len(s) <= keep { + return s } - s.finish(t, driver.PromptResult{Refusals: refusals}, errors.New("codex: the turn failed")) + return s[:keep] +} + +// endedOverfull ends a session that has refused more distinct tool calls +// than it can remember having refused. +func (s *session) endedOverfull() { + s.worker.Terminate(0) } // refusalsOf is a turn's refusals so far. diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index 47db16c5b..fb568649a 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "os" "path/filepath" "slices" @@ -791,10 +792,12 @@ func TestACompletedTurnThatWasCanceledDoesNotWaitForTheCheck(t *testing.T) { s := &session{verifyDone: make(chan struct{}), verifyAfter: time.Hour} turn := &turn{done: make(chan struct{}), canceled: true} s.turn = turn - done := make(chan struct{}) - go func() { s.turnCompleted(event{}); close(done) }() + // The turn's own end is what is waited for. A turn ends away from the + // reader now, so the call returning says nothing about whether the + // ending waited on anything. + s.turnCompleted(event{}) select { - case <-done: + case <-turn.done: case <-time.After(5 * time.Second): t.Fatal("a canceled turn waited for the policy check") } @@ -1445,3 +1448,65 @@ func TestACanceledTurnRecordsItsRefusals(t *testing.T) { } assert.Len(t, recorder.Recorded(), 1, "the refusal Codex logged is recorded, not lost with the cancel") } + +// What a refusal keeps of a tool call id is bounded, and so is how many the +// session remembers having refused. The agent writes both, and the redactor +// takes things out of a string without making it shorter — so an agent that +// names a tool call a megabyte long would otherwise have that megabyte kept +// on the turn, in the queue, and in the session's memory of what it refused. +func TestWhatARefusalKeepsOfWhatTheAgentWroteIsBounded(t *testing.T) { + recorder := &drivertest.Refusals{} + h := newHarness(t, scenario{TurnContext: safeTurnContext()}) + cfg := h.config() + cfg.Refusals = recorder + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + session := s.(*session) + tr := &turn{done: make(chan struct{})} + session.mu.Lock() + session.turn = tr + session.mu.Unlock() + + // Separators, so the redactor reads it as an id rather than as a + // credential and replaces the whole thing — which is what a long run of + // one character gets, and would prove nothing about the cut. + long := strings.Repeat("tool.call-", 8*maxToolCallID/10) + session.refused("item:"+long, long, "mcp__other__write", driver.ToolOther) + require.NoError(t, s.Close()) + + session.mu.Lock() + defer session.mu.Unlock() + require.Len(t, tr.refusals, 1) + assert.LessOrEqual(t, len(tr.refusals[0].ToolCallID), maxToolCallID, + "an id is cut to what this driver keeps of it, and kept %d", len(tr.refusals[0].ToolCallID)) + for key := range session.recorded { + assert.LessOrEqual(t, len(key), maxToolCallID+len("stderr:000:"), + "and what the session remembers is cut too") + } + assert.Len(t, recorder.Recorded(), 1) +} + +// A session that has refused more distinct tool calls than it can remember +// ends, rather than go on unable to tell a repeat from a first — recording +// the same refusal twice would be worse. The acp driver bounds the same +// memory for the same reason. +func TestASessionRemembersSoManyRefusalsAndNoMore(t *testing.T) { + recorder := &drivertest.Refusals{} + h := newHarness(t, scenario{TurnContext: safeTurnContext(), Hang: true}) + cfg := h.config() + cfg.Refusals = recorder + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + session := s.(*session) + + for i := range maxRecorded + 10 { + session.refused(fmt.Sprintf("item:call-%d", i), fmt.Sprintf("call-%d", i), "exec", driver.ToolExecute) + } + session.mu.Lock() + remembered := len(session.recorded) + session.mu.Unlock() + assert.LessOrEqual(t, remembered, maxRecorded, "a session remembers so many and no more, and remembered %d", remembered) + waitDone(t, s) +} diff --git a/internal/connector/driver/codex/scribe.go b/internal/connector/driver/codex/scribe.go index b7c953edd..c6758c067 100644 --- a/internal/connector/driver/codex/scribe.go +++ b/internal/connector/driver/codex/scribe.go @@ -58,9 +58,9 @@ import ( // refusalCap is where the growing stops. A drain bounded only by its clock // is not bounded in memory: two seconds of reading from a descendant that // refuses as fast as it can write is as much as the machine will take. At -// the cap the reader stops reading instead, which loses nothing that was -// read and abandons what was not — exactly what the clock would have done a -// moment later. +// the cap the reader stops taking more FROM THE PIPE, and goes on parsing +// what it has already taken — a scanner holds a bufferful that the pipe no +// longer does, and abandoning that would drop lines nobody had seen. const ( refusalMark = 256 refusalCap = 4096 @@ -92,7 +92,13 @@ type scribe struct { // written are a single decision rather than two that can cross. late int quiet *sync.Cond - done chan struct{} + // writing is held across every call to record, wherever it is made + // from. The recorder is promised it is never called concurrently with + // itself (driver.go), and a late write runs on its reader's goroutine + // rather than the scribe's, so the promise needs a lock rather than a + // single goroutine to keep it. + writing sync.Mutex + done chan struct{} } func newScribe(record func(driver.Refusal), emit func(driver.Update)) *scribe { @@ -206,9 +212,11 @@ func (s *scribe) run() { s.room.Signal() s.mu.Unlock() - // Outside the lock: the write is the slow thing, and the reader goes - // on reading while it happens. + // Outside the queue's lock: the write is the slow thing, and the + // reader goes on reading while it happens. + s.writing.Lock() s.record(next.refusal) + s.writing.Unlock() // The update comes after the write, which is the contract's own // order: a refusal is recorded before the update for it is emitted. s.emit(next.update) @@ -216,8 +224,12 @@ func (s *scribe) run() { } // writeNow writes a refusal on the caller's own goroutine, for one read -// after the scribe has gone. +// after the scribe has gone. Two of these can be in flight at once — two +// endings that are not the reader's — so they take the same lock the scribe +// writes under. func (s *scribe) writeNow(p pending) { + s.writing.Lock() s.record(p.refusal) + s.writing.Unlock() s.emit(p.update) } diff --git a/internal/connector/driver/codex/scribe_test.go b/internal/connector/driver/codex/scribe_test.go index 91ddc6304..c8485b3f1 100644 --- a/internal/connector/driver/codex/scribe_test.go +++ b/internal/connector/driver/codex/scribe_test.go @@ -251,3 +251,46 @@ func TestADrainThatFillsTheQueueTellsTheReaderToStop(t *testing.T) { close(release) s.close() } + +// The recorder is never called concurrently with itself, which driver.go +// promises an implementation it may rely on. A late write runs on its +// reader's goroutine rather than the scribe's, and two endings that are not +// the reader's can make one each, so the promise needs a lock rather than a +// single goroutine to keep it. +func TestTheRecorderIsNeverCalledTwiceAtOnce(t *testing.T) { + var mu sync.Mutex + inside, peak := 0, 0 + held := make(chan struct{}) + var once sync.Once + s := newScribe( + func(driver.Refusal) { + mu.Lock() + inside++ + peak = max(peak, inside) + mu.Unlock() + once.Do(func() { close(held) }) + time.Sleep(20 * time.Millisecond) + mu.Lock() + inside-- + mu.Unlock() + }, + func(driver.Update) {}, + ) + s.close() // every write from here is a late one, on its caller's goroutine + + var wg sync.WaitGroup + for i := range 8 { + wg.Add(1) + go func() { + defer wg.Done() + s.hand(pending{refusal: driver.Refusal{ToolCallID: string(rune('a' + i))}}) + }() + } + <-held + wg.Wait() + s.close() + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, peak, "the recorder saw one call at a time, and %d at once", peak) +} diff --git a/internal/connector/driver/driver_test.go b/internal/connector/driver/driver_test.go index dabbbae85..b422a30bc 100644 --- a/internal/connector/driver/driver_test.go +++ b/internal/connector/driver/driver_test.go @@ -456,18 +456,22 @@ func TestTheDrainBudgetRunsFromTheAsking(t *testing.T) { t.Cleanup(func() { _ = readEnd.Close(); _ = writeEnd.Close() }) out := &output{f: readEnd} - require.Zero(t, out.stopAt.Load(), "nothing has been asked yet") + require.True(t, out.askedAt().IsZero(), "nothing has been asked yet") before := time.Now() out.stop() after := time.Now() - asked := out.stopAt.Load() - require.NotZero(t, asked, "the budget's clock is taken when the stop is asked for, not when a read next looks") - assert.False(t, time.Unix(0, asked).Before(before), "and it is taken then") - assert.False(t, time.Unix(0, asked).After(after)) + asked := out.askedAt() + require.False(t, asked.IsZero(), "the budget's clock is taken when the stop is asked for, not when a read next looks") + assert.False(t, asked.Before(before), "and it is taken then") + assert.False(t, asked.After(after)) + + // It carries a monotonic reading, so the budget is not at the mercy of + // the wall clock moving under it. + assert.NotEqual(t, asked, asked.Round(0), "the time kept is monotonic, not a wall clock rebuilt from a number") // And it is the first asking that counts, so a second does not hand the // reader a fresh budget. out.stop() - assert.Equal(t, asked, out.stopAt.Load(), "a later stop does not restart the budget") + assert.Equal(t, asked, out.askedAt(), "a later stop does not restart the budget") } diff --git a/internal/connector/driver/worker.go b/internal/connector/driver/worker.go index bb27c41cf..dfa3ed950 100644 --- a/internal/connector/driver/worker.go +++ b/internal/connector/driver/worker.go @@ -82,12 +82,19 @@ type output struct { // written by one call, so there is no moment in which half the state is // visible. stopped atomic.Bool - // stopAt is when the stop was asked for, in Unix nanoseconds, written by - // stop before the flag that publishes it. The reader takes it from here - // rather than from when it next looks, so the budget runs from the - // asking as it says it does — a reader inside an idle window when the - // asking comes would otherwise start the clock up to that window late. - stopAt atomic.Int64 + // stopAt is when the stop was asked for, written by stop before the flag + // that publishes it. The reader takes it from here rather than from when + // it next looks, so the budget runs from the asking as it says it does — + // a reader inside an idle window when the asking comes would otherwise + // start the clock up to that window late. + // + // It is kept as a time.Time under a mutex rather than as nanoseconds in + // an atomic, because a time.Time carries a monotonic reading and a + // number does not. Rebuilt from nanoseconds, the wall clock decides the + // budget: a forward jump ends a drain early with the worker's output + // still buffered, and a backward one holds the session open past it. + stopMu sync.Mutex + stopAt time.Time } func (o *output) Read(p []byte) (int, error) { @@ -95,7 +102,7 @@ func (o *output) Read(p []byte) (int, error) { stopped := o.stopped.Load() window := idleWindow if stopped { - left := drainBudget - time.Since(time.Unix(0, o.stopAt.Load())) + left := drainBudget - time.Since(o.askedAt()) if left <= 0 { // Still arriving, and no longer waited for. return 0, io.EOF @@ -127,11 +134,23 @@ func (o *output) Read(p []byte) (int, error) { // has opened — and it says the same thing however many times it is called. func (o *output) stop() { // The time before the flag, so a reader that sees the flag always finds - // a time to measure from, and the budget runs from the asking. - o.stopAt.CompareAndSwap(0, time.Now().UnixNano()) + // a time to measure from, and the budget runs from the asking. The first + // asking is the one that counts. + o.stopMu.Lock() + if o.stopAt.IsZero() { + o.stopAt = time.Now() + } + o.stopMu.Unlock() o.stopped.Store(true) } +// askedAt is when the stop was asked for, monotonic. +func (o *output) askedAt() time.Time { + o.stopMu.Lock() + defer o.stopMu.Unlock() + return o.stopAt +} + func (o *output) close() { _ = o.f.Close() } // # One owner, one release point @@ -525,6 +544,12 @@ func (w *Worker) Terminate(grace time.Duration) { time.AfterFunc(pipeWaitDelay, w.CloseStdout) return } + // Asked before it is waited for: a descendant outside the group can + // hold the write end open, so the end of file never comes and a + // reader left to itself never ends — which would hold this + // worker's descriptor, and everything waiting on the reading, for + // as long as that descendant lived. + w.stdout.stop() go func() { <-reading w.CloseStdout() From 67904ac0b621632dd88fc163eebe05c6eae55395 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 11:18:32 +0200 Subject: [PATCH 4/5] A refusal read for a turn is put on that turn, not on whatever is left MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A canceled turn came back without the refusal Codex logged on its way out — `TestACanceledTurnCarriesALateRefusalInItsResult`, on CI under the race detector. It is not the test being slow: a turn can now be ended by two goroutines, and both read the worker's last word, and only one of them can be the first to take the turn out of the session. Whichever read that stderr second recorded the refusal to the ledger and then looked for a turn to put it on and found none, so the result the caller got had nothing on it. Slowing everything down only widened the window. A refusal read for a turn now goes on that turn. The endings say which one they are settling, and only a refusal read outside one — from the stream, mid-turn — looks the session up. Two more from the same review, both mine and both introduced by this branch. Ending the worker asks its reader to stop, which starts the drain's clock, and the reader could at that moment be waiting for room at the scribe's queue — spending the clock without reading, and abandoning the pipe with the worker's output still in it. Every ending of the worker now frees the reader from waiting before it ends anything. And the scribe was drained only when the reader tore down, so a turn could hand its result back with its refusals still queued in memory — the caller would ask the ledger and find nothing, which is the ordering callers had when the write was on the reader. Every ending drains before it settles. Endings run away from the reader, so that waits for nobody who is reading. The stop's flag was also sampled before the read deadline was installed, so a stop landing between the two was hidden behind a whole idle window. The read looks again rather than installing a window it already knows is wrong. --- internal/connector/driver/codex/codex.go | 77 ++++++++++++++----- internal/connector/driver/codex/codex_test.go | 6 +- internal/connector/driver/codex/scribe.go | 36 ++++++++- .../connector/driver/codex/scribe_test.go | 48 ++++++++++++ internal/connector/driver/worker.go | 7 ++ 5 files changed, 149 insertions(+), 25 deletions(-) diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index f90616a8a..4bc81514c 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -605,7 +605,7 @@ func (s *session) Cancel(context.Context) error { // holding the pipe open: the end of file never comes, and a canceled // turn that only the reader can finish would never be finished at all. go func() { - s.worker.Terminate(s.grace) + s.endWorker(s.grace) s.readerDone() }() return nil @@ -628,7 +628,7 @@ func (s *session) Close() error { case <-s.worker.Done(): case <-time.After(s.grace): } - s.worker.Terminate(s.grace) + s.endWorker(s.grace) s.readerDone() // Close says the ledger has every refusal this session read. s.scribe.close() @@ -653,7 +653,6 @@ func (s *session) readerDone() { // first and ends within the drain budget. From here it must never // be made to wait handing a refusal over, or that budget would run // while it sat on a full queue. - s.scribe.noWaiting() s.worker.StopReading() <-s.readerEnd } @@ -680,6 +679,19 @@ func (s *session) finish(t *turn, result driver.PromptResult, err error) { s.end(t, result, err, false) } +// settled is what every ending does before it hands a turn's result back: +// the refusals that turn made are in the ledger, not still queued. The +// writes are not on the reader any more, so a caller that gets its result +// and asks the ledger would otherwise be asking early — and the dispatcher +// settles an attempt on what the ledger holds. +// +// Endings run away from the reader, so waiting here holds up no reading. +func (s *session) settled() { + if s.scribe != nil { + s.scribe.drain() + } +} + // settle ends a turn this ending claimed, giving up the claim in the same // step so that nothing can come between the two. func (s *session) settle(t *turn, result driver.PromptResult, err error) { @@ -768,11 +780,12 @@ func (s *session) read() { case <-time.After(s.grace): } } - s.stderrRefusals() + s.stderrRefusals(t) if t != nil { s.mu.Lock() canceled := t.canceled s.mu.Unlock() + s.settled() refusals := s.refusalsOf(t) switch { case canceled: @@ -893,12 +906,13 @@ func (s *session) threadStarted(id string) { // for this ending is the one thing it is for. func (s *session) unsafe(err error) { t := s.claim() - s.worker.Terminate(0) + s.endWorker(0) if t == nil { return } s.readerDone() - s.lastWord() + s.lastWord(t) + s.settled() s.settle(t, driver.PromptResult{Refusals: s.refusalsOf(t)}, err) } @@ -913,8 +927,9 @@ func (s *session) unsafe(err error) { // The policy check's own goroutine has unsafe, which waits for the reader // first, for that same reason. func (s *session) finishUnsafe(t *turn, err error) { - s.worker.Terminate(0) - s.lastWord() + s.endWorker(0) + s.lastWord(t) + s.settled() s.finish(t, driver.PromptResult{Refusals: s.refusalsOf(t)}, err) } @@ -942,14 +957,14 @@ func (s *session) failedVerification() error { // and a worker that is gone is not a stream that has been read. An ending // that is not the reader's own waits for the reader too — see unsafe — before // it asks a turn what its refusals are. -func (s *session) lastWord() { +func (s *session) lastWord(t *turn) { if s.worker != nil { select { case <-s.worker.Done(): case <-time.After(s.grace): } } - s.stderrRefusals() + s.stderrRefusals(t) } // finishCanceled ends a turn the connector canceled, after the worker's last @@ -957,7 +972,8 @@ func (s *session) lastWord() { // one still running is not waited for, because the process it would judge is // being ended by the cancel anyway. func (s *session) finishCanceled(t *turn) { - s.lastWord() + s.lastWord(t) + s.settled() // The turn's refusals are read after the worker's last word: the ones it // only logged are read by lastWord just above, and the ones it put on its // stream are already the turn's when the reader is the one ending it, @@ -1036,7 +1052,7 @@ func (s *session) item(kind string, e event) { } s.emit(u) if kind == "item.completed" && it.Type == "mcp_tool_call" && it.Error != nil && refusedByApproval(it.Error.Message) { - s.refused("item:"+it.ID, it.ID, u.Tool, u.ToolKind) + s.refused(nil, "item:"+it.ID, it.ID, u.Tool, u.ToolKind) } } @@ -1065,7 +1081,7 @@ func refusedByApproval(message string) bool { // session's recorder before anything else is done with it, and only the first // time its tool call id is seen (driver's "Refusals"). A refusal Codex logs // and gives no id gets the key the caller passes. -func (s *session) refused(key, id, tool string, kind driver.ToolKind) { +func (s *session) refused(t *turn, key, id, tool string, kind driver.ToolKind) { // The id is the agent's, and the redactor takes things out of a string // without making it shorter: an agent that names a tool call a megabyte // long would otherwise have that megabyte kept on the turn, in the @@ -1073,6 +1089,14 @@ func (s *session) refused(key, id, tool string, kind driver.ToolKind) { refusal := driver.Refusal{ToolCallID: cut(s.red.Sanitize(id), maxToolCallID), Tool: s.red.Sanitize(tool)} key = cut(key, maxToolCallID+len("stderr:000:")) s.mu.Lock() + // The turn the caller is settling, where it named one: an ending reads + // the worker's last word after another ending may already have taken + // the turn out of the session, and a refusal read then belongs to the + // turn it was read for rather than to nobody. + on := t + if on == nil { + on = s.turn + } first := !s.recorded[key] if first { if len(s.recorded) >= maxRecorded { @@ -1084,8 +1108,8 @@ func (s *session) refused(key, id, tool string, kind driver.ToolKind) { return } s.recorded[key] = true - if s.turn != nil { - s.turn.refusals = append(s.turn.refusals, refusal) + if on != nil { + on.refusals = append(on.refusals, refusal) } } s.mu.Unlock() @@ -1140,7 +1164,8 @@ func (s *session) turnCompleted(e event) { case <-s.worker.Done(): case <-time.After(s.grace): } - s.stderrRefusals() + s.stderrRefusals(t) + s.settled() s.mu.Lock() result := driver.PromptResult{Stop: driver.TurnEndTurn, Refusals: slices.Clone(t.refusals)} if t.canceled { @@ -1198,7 +1223,8 @@ func (s *session) turnFailed() { case <-s.worker.Done(): case <-time.After(s.grace): } - s.stderrRefusals() + s.stderrRefusals(t) + s.settled() refusals := s.refusalsOf(t) if err := s.failedVerification(); err != nil { s.finishUnsafe(t, err) @@ -1232,7 +1258,18 @@ func cut(s string, keep int) string { // endedOverfull ends a session that has refused more distinct tool calls // than it can remember having refused. func (s *session) endedOverfull() { - s.worker.Terminate(0) + s.endWorker(0) +} + +// endWorker ends the worker, and frees the reader from waiting for room +// before it does. Ending the worker asks its reader to stop, which starts +// the drain's clock, and a reader held at the scribe's queue is a reader +// spending that clock without reading — the pipe would be abandoned with +// the worker's output still in it. From here the queue grows rather than +// holds anyone up. +func (s *session) endWorker(grace time.Duration) { + s.scribe.noWaiting() + s.worker.Terminate(grace) } // refusalsOf is a turn's refusals so far. @@ -1245,7 +1282,7 @@ func (s *session) refusalsOf(t *turn) []driver.Refusal { // stderrRefusals counts the refusals Codex logs but does not put on its JSON // stream: an edit outside the working directory. Best effort: the stderr // kept is a tail. -func (s *session) stderrRefusals() { +func (s *session) stderrRefusals(t *turn) { if s.worker == nil { return } @@ -1266,7 +1303,7 @@ func (s *session) stderrRefusals() { // way are two, and reading the same output again — every way a turn // can end reads it — records each of them once. seen[line]++ - s.refused("stderr:"+strconv.Itoa(seen[line])+":"+line, "", tool, kind) + s.refused(t, "stderr:"+strconv.Itoa(seen[line])+":"+line, "", tool, kind) } } diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index fb568649a..2e874f295 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -1348,7 +1348,7 @@ func TestARefusalReadAfterTheUpdatesCloseIsNotAPanic(t *testing.T) { 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() + session.stderrRefusals(nil) assert.Len(t, recorder.Recorded(), 1, "the refusal is recorded, and the update it carries is dropped rather than sent on a closed channel") @@ -1472,7 +1472,7 @@ func TestWhatARefusalKeepsOfWhatTheAgentWroteIsBounded(t *testing.T) { // credential and replaces the whole thing — which is what a long run of // one character gets, and would prove nothing about the cut. long := strings.Repeat("tool.call-", 8*maxToolCallID/10) - session.refused("item:"+long, long, "mcp__other__write", driver.ToolOther) + session.refused(nil, "item:"+long, long, "mcp__other__write", driver.ToolOther) require.NoError(t, s.Close()) session.mu.Lock() @@ -1502,7 +1502,7 @@ func TestASessionRemembersSoManyRefusalsAndNoMore(t *testing.T) { session := s.(*session) for i := range maxRecorded + 10 { - session.refused(fmt.Sprintf("item:call-%d", i), fmt.Sprintf("call-%d", i), "exec", driver.ToolExecute) + session.refused(nil, fmt.Sprintf("item:call-%d", i), fmt.Sprintf("call-%d", i), "exec", driver.ToolExecute) } session.mu.Lock() remembered := len(session.recorded) diff --git a/internal/connector/driver/codex/scribe.go b/internal/connector/driver/codex/scribe.go index c6758c067..d529f9a76 100644 --- a/internal/connector/driver/codex/scribe.go +++ b/internal/connector/driver/codex/scribe.go @@ -90,8 +90,12 @@ type scribe struct { // the scribe had already gone. It lives under the same lock that // publishes finished, so admitting one and deciding everything is // written are a single decision rather than two that can cross. - late int - quiet *sync.Cond + late int + // writes counts the writes the scribe itself has taken off the queue + // and not yet finished, so a drain can tell an empty queue from a + // finished one. + writes int + quiet *sync.Cond // writing is held across every call to record, wherever it is made // from. The recorder is promised it is never called concurrently with // itself (driver.go), and a late write runs on its reader's goroutine @@ -161,6 +165,28 @@ func (s *scribe) noWaiting() { s.mu.Unlock() } +// drain waits for everything handed over so far to be written, without +// saying there is no more. A turn's result is settled after this, so a +// refusal read in that turn is in the ledger before the prompt that made it +// comes back — the ordering callers had when the write was on the reader. +// +// It is called from a turn's ending, which runs away from the reader, so +// waiting here holds up no reading. +func (s *scribe) drain() { + s.mu.Lock() + for (len(s.queue) > 0 || s.writes > 0) && !s.finished { + s.quiet.Wait() + } + s.mu.Unlock() + // A late write is on its caller's own goroutine and counted the same + // way; this waits for those too. + s.mu.Lock() + for s.late > 0 { + s.quiet.Wait() + } + s.mu.Unlock() +} + // close says there is no more, and waits for what there is to be written. // Whoever calls it promises the reader is through. It is idempotent. func (s *scribe) close() { @@ -204,11 +230,13 @@ func (s *scribe) run() { // queued to a scribe that has decided it is done. s.finished = true s.room.Broadcast() + s.quiet.Broadcast() s.mu.Unlock() return } next := s.queue[0] s.queue = s.queue[1:] + s.writes++ s.room.Signal() s.mu.Unlock() @@ -220,6 +248,10 @@ func (s *scribe) run() { // The update comes after the write, which is the contract's own // order: a refusal is recorded before the update for it is emitted. s.emit(next.update) + s.mu.Lock() + s.writes-- + s.quiet.Broadcast() + s.mu.Unlock() } } diff --git a/internal/connector/driver/codex/scribe_test.go b/internal/connector/driver/codex/scribe_test.go index c8485b3f1..1882b0629 100644 --- a/internal/connector/driver/codex/scribe_test.go +++ b/internal/connector/driver/codex/scribe_test.go @@ -294,3 +294,51 @@ func TestTheRecorderIsNeverCalledTwiceAtOnce(t *testing.T) { defer mu.Unlock() assert.Equal(t, 1, peak, "the recorder saw one call at a time, and %d at once", peak) } + +// A turn's refusals are in the ledger before its result comes back. The +// writes are not on the reader any more, so an ending that settled a turn +// without waiting would hand a caller a result whose refusals the ledger did +// not yet hold — and the dispatcher settles an attempt on what it holds. +func TestDrainingWaitsForWhatIsQueuedWithoutClosing(t *testing.T) { + var mu sync.Mutex + written := 0 + release := make(chan struct{}) + first := make(chan struct{}) + var once sync.Once + s := newScribe( + func(driver.Refusal) { + once.Do(func() { close(first) }) + <-release + mu.Lock() + written++ + mu.Unlock() + }, + func(driver.Update) {}, + ) + t.Cleanup(s.close) + for range 4 { + s.hand(pending{refusal: driver.Refusal{ToolCallID: "one"}}) + } + <-first + + drained := make(chan struct{}) + go func() { defer close(drained); s.drain() }() + select { + case <-drained: + t.Fatal("the drain returned with refusals still queued") + case <-time.After(200 * time.Millisecond): + } + + close(release) + <-drained + mu.Lock() + assert.Equal(t, 4, written, "everything handed over is written, and %d was", written) + mu.Unlock() + + // And the scribe is still open afterwards: draining is not closing. + s.hand(pending{refusal: driver.Refusal{ToolCallID: "after"}}) + s.drain() + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 5, written, "draining is not closing") +} diff --git a/internal/connector/driver/worker.go b/internal/connector/driver/worker.go index dfa3ed950..43cd0d6ee 100644 --- a/internal/connector/driver/worker.go +++ b/internal/connector/driver/worker.go @@ -114,6 +114,13 @@ func (o *output) Read(p []byte) (int, error) { if err := o.f.SetReadDeadline(time.Now().Add(window)); err != nil { return 0, err } + if !stopped && o.stopped.Load() { + // Asked between the sample and the deadline. The idle window + // just installed would hide that until it expired, and the + // budget would be most of the way gone by then, so the window + // is opened again as a drain's. + continue + } n, err := o.f.Read(p) switch { case n > 0: From 92ebdf4c13379dcfa8d0585f3b4439aa32b37359 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 11:25:09 +0200 Subject: [PATCH 5/5] Two ids that begin alike are two refusals, and the one that fills the memory is recorded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bounding what a session remembers a refusal by, I cut the key — which makes a prefix the identity. Two tool call ids alike for longer than the cut would collapse to one entry, and every later one would be taken for a repeat: not recorded, not on the turn, not on the stream. A bound on memory that loses refusals is the defect this branch exists to close, written by the fix for a different one. The key is a digest past that length now. Bounded, because the agent writes it and a session's memory is not the agent's to grow, and still distinct for ids that only begin alike. And the check for that memory being full ran before the refusal that filled it was taken, so that one was dropped on its way to ending the session. It is recorded like any other now; the session ends after taking it rather than instead of it, and records nothing after, because past the bound a repeat cannot be told from a first and recording one twice would be worse. Both were found in review, and both are held by tests that go red when the thing they exist for is deleted: identity by prefix loses the second of two ids alike, 5 runs of 5, and checking the bound first loses the refusal that reached it, 5 of 5. --- internal/connector/driver/codex/codex.go | 48 +++++++++++++---- internal/connector/driver/codex/codex_test.go | 54 ++++++++++++++++++- 2 files changed, 92 insertions(+), 10 deletions(-) diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index 4bc81514c..e307ba4ce 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -65,6 +65,8 @@ package codex import ( "bufio" "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -465,6 +467,9 @@ type session struct { verifyDone chan struct{} verifyErr error closed bool + // full is a session that has refused more distinct tool calls than it + // can remember having refused. It is ending; it records no more. + full 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 @@ -1087,7 +1092,7 @@ func (s *session) refused(t *turn, key, id, tool string, kind driver.ToolKind) { // long would otherwise have that megabyte kept on the turn, in the // queue, and in what this session remembers having refused. refusal := driver.Refusal{ToolCallID: cut(s.red.Sanitize(id), maxToolCallID), Tool: s.red.Sanitize(tool)} - key = cut(key, maxToolCallID+len("stderr:000:")) + key = identity(key) s.mu.Lock() // The turn the caller is settling, where it named one: an ending reads // the worker's last word after another ending may already have taken @@ -1097,25 +1102,37 @@ func (s *session) refused(t *turn, key, id, tool string, kind driver.ToolKind) { if on == nil { on = s.turn } + if s.full { + // The session is ending for want of memory; it cannot tell a repeat + // from a first any more, and recording one twice would be worse + // than recording neither. + s.mu.Unlock() + return + } first := !s.recorded[key] + overfull := false if first { - if len(s.recorded) >= maxRecorded { - // Past what a session can remember, a repeat cannot be told from - // a first, and recording one twice is worse than this: the - // session ends rather than go on guessing. - s.mu.Unlock() - s.endedOverfull() - return - } s.recorded[key] = true if on != nil { on.refusals = append(on.refusals, refusal) } + // The one that fills the memory is recorded like any other. Past it + // a repeat cannot be told from a first, and recording one twice + // would be worse, so the session ends — after this refusal has been + // taken, not instead of it. + if len(s.recorded) >= maxRecorded { + s.full, overfull = true, true + } } s.mu.Unlock() if !first { return } + defer func() { + if overfull { + s.endedOverfull() + } + }() // The write and the update it precedes both belong to the scribe: the // reader hands them over and goes back to reading, which is what lets // the bound on its reading be a clock. See scribe.go. @@ -1255,6 +1272,19 @@ func cut(s string, keep int) string { return s[:keep] } +// identity is what a session remembers a refusal by. It is bounded, because +// the agent writes it and a session's memory is not the agent's to grow; and +// it is not a prefix, because two tool call ids that begin alike are two +// refusals and cutting them to a common start would record only the first. +// A digest is bounded and keeps them apart. +func identity(key string) string { + if len(key) <= maxToolCallID { + return key + } + sum := sha256.Sum256([]byte(key)) + return hex.EncodeToString(sum[:]) +} + // endedOverfull ends a session that has refused more distinct tool calls // than it can remember having refused. func (s *session) endedOverfull() { diff --git a/internal/connector/driver/codex/codex_test.go b/internal/connector/driver/codex/codex_test.go index 2e874f295..e5bcf3e4b 100644 --- a/internal/connector/driver/codex/codex_test.go +++ b/internal/connector/driver/codex/codex_test.go @@ -1507,6 +1507,58 @@ func TestASessionRemembersSoManyRefusalsAndNoMore(t *testing.T) { session.mu.Lock() remembered := len(session.recorded) session.mu.Unlock() - assert.LessOrEqual(t, remembered, maxRecorded, "a session remembers so many and no more, and remembered %d", remembered) + assert.Equal(t, maxRecorded, remembered, "a session remembers so many and no more, and remembered %d", remembered) + waitDone(t, s) +} + +// Two tool call ids that begin alike are two refusals. What a session +// remembers a refusal by is bounded, because the agent writes it — but +// bounding it by cutting would make a prefix the identity, and every later +// refusal sharing that start would be taken for a repeat and recorded +// nowhere. +func TestTwoIdsThatBeginAlikeAreTwoRefusals(t *testing.T) { + recorder := &drivertest.Refusals{} + h := newHarness(t, scenario{TurnContext: safeTurnContext(), Hang: true}) + cfg := h.config() + cfg.Refusals = recorder + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + session := s.(*session) + tr := &turn{done: make(chan struct{})} + session.mu.Lock() + session.turn = tr + session.mu.Unlock() + + same := strings.Repeat("tool.call-", 4*maxToolCallID/10) + session.refused(tr, "item:"+same+"-one", same+"-one", "exec", driver.ToolExecute) + session.refused(tr, "item:"+same+"-two", same+"-two", "exec", driver.ToolExecute) + require.NoError(t, s.Close()) + + session.mu.Lock() + defer session.mu.Unlock() + assert.Len(t, tr.refusals, 2, "two ids alike for longer than a session keeps are still two refusals") + assert.Len(t, recorder.Recorded(), 2, "and the ledger has both") +} + +// The refusal that fills a session's memory is recorded like any other. The +// session ends on it — past it a repeat cannot be told from a first — but it +// ends after taking that refusal, not instead of it. +func TestTheRefusalThatFillsTheMemoryIsStillRecorded(t *testing.T) { + recorder := &drivertest.Refusals{} + h := newHarness(t, scenario{TurnContext: safeTurnContext(), Hang: true}) + cfg := h.config() + cfg.Refusals = recorder + s, err := h.drv.NewSession(context.Background(), cfg) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + session := s.(*session) + + for i := range maxRecorded { + session.refused(nil, fmt.Sprintf("item:call-%d", i), fmt.Sprintf("call-%d", i), "exec", driver.ToolExecute) + } + require.NoError(t, s.Close()) + assert.Len(t, recorder.Recorded(), maxRecorded, + "every refusal up to and including the one that filled the memory, and the ledger has %d", len(recorder.Recorded())) waitDone(t, s) }