From 0dc92a9e9f7816a474b3112e1d4267781071ddca Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Mon, 20 Jul 2026 20:46:23 +0200 Subject: [PATCH 1/8] fix(acp): interrupt idle reads on cancellation --- internal/acp/jsonrpc.go | 52 +++++++++++++++--- internal/acp/jsonrpc_test.go | 104 +++++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 9 deletions(-) diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go index 51df655fa..1c416ec94 100644 --- a/internal/acp/jsonrpc.go +++ b/internal/acp/jsonrpc.go @@ -80,8 +80,9 @@ type NotifyFunc func(ctx context.Context, params json.RawMessage) // requests/notifications — needed because ACP is bidirectional (the agent calls // the client for session/request_permission, fs/*, terminal/*). type Conn struct { - reader *bufio.Reader - w io.Writer + reader *bufio.Reader + readerCloser io.Closer + w io.Writer writeMu sync.Mutex // serializes all writes to w @@ -96,14 +97,18 @@ type Conn struct { wg sync.WaitGroup // tracks in-flight inbound handlers } -// NewConn builds a peer reading ndjson from r and writing ndjson to w. +// NewConn builds a peer reading ndjson from r and writing ndjson to w. If r is +// closable, Serve closes it to interrupt an idle read when its context is +// cancelled. func NewConn(r io.Reader, w io.Writer) *Conn { + readerCloser, _ := r.(io.Closer) return &Conn{ - reader: bufio.NewReader(r), - w: w, - handlers: make(map[string]HandlerFunc), - notifiers: make(map[string]NotifyFunc), - pending: make(map[int64]chan rpcMessage), + reader: bufio.NewReader(r), + readerCloser: readerCloser, + w: w, + handlers: make(map[string]HandlerFunc), + notifiers: make(map[string]NotifyFunc), + pending: make(map[int64]chan rpcMessage), } } @@ -119,12 +124,36 @@ func (c *Conn) HandleNotify(method string, fn NotifyFunc) { c.notifiers[method] // blocks the loop from delivering session/cancel or a permission response. func (c *Conn) Serve(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) + readLoopDone := make(chan struct{}) + var endReadOnce sync.Once + readInterrupted := false + endRead := func(interrupt bool) { + endReadOnce.Do(func() { + readInterrupted = interrupt + if interrupt && c.readerCloser != nil { + _ = c.readerCloser.Close() + } + close(readLoopDone) + }) + } + var readerWatcher sync.WaitGroup + readerWatcher.Add(1) + go func() { + defer readerWatcher.Done() + select { + case <-ctx.Done(): + endRead(true) + case <-readLoopDone: + } + }() // On exit, cancel in-flight handlers (so a blocked outbound Call unblocks via // ctx) and then wait for them to finish writing their responses. Without this, // a finite input stream (e.g. piped ndjson that EOFs right after a request) // would race the dispatch goroutine and drop the response. defer func() { + endRead(false) cancel() + readerWatcher.Wait() c.wg.Wait() }() @@ -133,12 +162,17 @@ func (c *Conn) Serve(ctx context.Context) error { // decoder) keeps a single malformed line from making the whole connection // unrecoverable — we report -32700 and continue. line, err := c.reader.ReadBytes('\n') + if err != nil { + // Claim terminal completion before handling a partial final frame so + // concurrent cancellation cannot close the reader or mask this error. + endRead(false) + } if len(bytes.TrimSpace(line)) > 0 { c.handleLine(ctx, line) } if err != nil { c.failAllPending(err) - if errors.Is(err, io.EOF) || ctx.Err() != nil { + if errors.Is(err, io.EOF) || readInterrupted { return nil } return err diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index d9e5d3df3..9e24a7597 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -3,11 +3,28 @@ package acp import ( "context" "encoding/json" + "errors" "io" "testing" "time" ) +type testReadCloser struct { + read func([]byte) (int, error) + close func() error +} + +func (r testReadCloser) Read(p []byte) (int, error) { return r.read(p) } +func (r testReadCloser) Close() error { return r.close() } + +type testReader func([]byte) (int, error) + +func (r testReader) Read(p []byte) (int, error) { return r(p) } + +type testWriter func([]byte) (int, error) + +func (w testWriter) Write(p []byte) (int, error) { return w(p) } + // connPair wires two Conns together over in-memory pipes and serves both. func connPair(t *testing.T) (a, b *Conn, stop func()) { t.Helper() @@ -25,6 +42,93 @@ func connPair(t *testing.T) (a, b *Conn, stop func()) { } } +func TestConnServeCancellationInterruptsIdleRead(t *testing.T) { + pipeReader, writer := io.Pipe() + t.Cleanup(func() { _ = writer.Close() }) + readStarted := make(chan struct{}) + closeCalls := make(chan struct{}, 2) + reader := testReadCloser{ + read: func(p []byte) (int, error) { + close(readStarted) + return pipeReader.Read(p) + }, + close: func() error { + closeCalls <- struct{}{} + return pipeReader.Close() + }, + } + conn := NewConn(reader, io.Discard) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- conn.Serve(ctx) }() + + <-readStarted + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("Serve returned %v after cancellation, want nil", err) + } + case <-time.After(time.Second): + t.Fatal("Serve did not return after cancelling an idle connection") + } + if got := len(closeCalls); got != 1 { + t.Fatalf("reader Close calls = %d, want 1", got) + } +} + +func TestConnServePreservesTerminalReadErrorDuringCancellation(t *testing.T) { + for _, closable := range []bool{false, true} { + name := "non-closable" + if closable { + name = "closable" + } + t.Run(name, func(t *testing.T) { + wantErr := errors.New("read failed") + read := testReader(func(p []byte) (int, error) { + return copy(p, "not json"), wantErr + }) + closeCalls := make(chan struct{}, 1) + var reader io.Reader = read + if closable { + reader = testReadCloser{ + read: read, + close: func() error { + closeCalls <- struct{}{} + return nil + }, + } + } + writeStarted := make(chan struct{}) + releaseWrite := make(chan struct{}) + writer := testWriter(func(p []byte) (int, error) { + close(writeStarted) + <-releaseWrite + return len(p), nil + }) + conn := NewConn(reader, writer) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- conn.Serve(ctx) }() + + <-writeStarted + cancel() + close(releaseWrite) + select { + case err := <-done: + if !errors.Is(err, wantErr) { + t.Fatalf("Serve returned %v, want terminal read error %v", err, wantErr) + } + case <-time.After(time.Second): + t.Fatal("Serve did not return after terminal read error") + } + if got := len(closeCalls); got != 0 { + t.Fatalf("reader Close calls = %d, want 0 after terminal read", got) + } + }) + } +} + func TestConnRequestResponse(t *testing.T) { a, b, stop := connPair(t) defer stop() From f5d772816086e364793bcf75c2e4752816216e26 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sat, 25 Jul 2026 17:12:32 +0200 Subject: [PATCH 2/8] fix(acp): make idle-read interruption per-Read, not per-Serve-call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the P2 on #782. The previous design raced a single, whole-connection Once (endRead) between the read loop's own "I have a terminal result" claim and a watcher goroutine's "ctx is done, treat this as interrupted" claim. This has a real, deterministic failure mode that has nothing to do with goroutine-scheduling luck: bufio.Reader can obtain data AND a terminal error from ONE underlying Read call (io.Reader explicitly permits returning (n>0, err) together). If a line delimiter is found within that data, ReadBytes returns the line with a nil error for that call, but bufio caches the error internally and returns it — without ever touching the underlying reader again — on the very next ReadBytes call. If cancellation happens in the gap between that first, successful read and the next one (e.g. while a synchronously-dispatched handler is still writing its response), the watcher's ctx.Done() case fires and claims the Once BEFORE the read loop ever gets back to a ReadBytes call — so when the cached error then surfaces (with zero contact with the underlying reader, let alone anything closing it), it is misattributed as a cancellation-driven interruption and silently swallowed as a clean shutdown, hiding a genuine broken transport. The fix moves interruption-detection down to the Read call itself instead of bracketing the whole Serve loop. interruptibleReader wraps the connection's reader: each Read runs the underlying call on its own goroutine and races it against ctx.Done(). If the underlying call finishes first, its result is returned untouched — cancellation a moment later has no bearing on a call that already had its answer. If ctx.Done() fires first, the reader is closed (when closable) to nudge a genuinely blocked read, a generation counter is bumped, and the call still waits for and returns the underlying call's own result rather than a synthesized one. Serve then brackets each ReadBytes call (which may invoke the wrapped Read zero or more times — zero when bufio answers from its own buffer, including a previously-cached error) with a snapshot of that counter before and after. A read whose bracket shows no change was never raced against cancellation — either answered from the buffer or genuinely completed before ctx.Done() could matter — so its outcome is reported truthfully regardless of the ctx state by the time it surfaces. Only a bracket that changed proves this specific call actually took the cancellation branch. Verified the fix has teeth: reverted to the original implementation and confirmed the new regression test (which forces the exact bufio-caching sequence deterministically, with a panic guard proving the underlying reader is never called a second time) fails 100% of the time against it, then restored the fix and confirmed it passes. Also applied the two open review threads on the existing tests: readStarted and writeStarted are now buffered channels with non-blocking sends rather than close(), so a future change causing an extra read or write can't panic on a double-close. Validated with go build/vet/test on Windows, plus go vet/build and `go test -race ./internal/acp/... -count=50` (and count=100 focused on the cancellation/error tests) under WSL/Linux — this session's Windows host has no C toolchain for -race directly. --- internal/acp/jsonrpc.go | 124 +++++++++++++++++++++++++---------- internal/acp/jsonrpc_test.go | 90 +++++++++++++++++++++++-- 2 files changed, 177 insertions(+), 37 deletions(-) diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go index 1c416ec94..1e7172390 100644 --- a/internal/acp/jsonrpc.go +++ b/internal/acp/jsonrpc.go @@ -80,7 +80,7 @@ type NotifyFunc func(ctx context.Context, params json.RawMessage) // requests/notifications — needed because ACP is bidirectional (the agent calls // the client for session/request_permission, fs/*, terminal/*). type Conn struct { - reader *bufio.Reader + rawReader io.Reader // wrapped lazily in Serve, once ctx is known — see interruptibleReader readerCloser io.Closer w io.Writer @@ -103,7 +103,7 @@ type Conn struct { func NewConn(r io.Reader, w io.Writer) *Conn { readerCloser, _ := r.(io.Closer) return &Conn{ - reader: bufio.NewReader(r), + rawReader: r, readerCloser: readerCloser, w: w, handlers: make(map[string]HandlerFunc), @@ -124,55 +124,45 @@ func (c *Conn) HandleNotify(method string, fn NotifyFunc) { c.notifiers[method] // blocks the loop from delivering session/cancel or a permission response. func (c *Conn) Serve(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) - readLoopDone := make(chan struct{}) - var endReadOnce sync.Once - readInterrupted := false - endRead := func(interrupt bool) { - endReadOnce.Do(func() { - readInterrupted = interrupt - if interrupt && c.readerCloser != nil { - _ = c.readerCloser.Close() - } - close(readLoopDone) - }) - } - var readerWatcher sync.WaitGroup - readerWatcher.Add(1) - go func() { - defer readerWatcher.Done() - select { - case <-ctx.Done(): - endRead(true) - case <-readLoopDone: - } - }() // On exit, cancel in-flight handlers (so a blocked outbound Call unblocks via // ctx) and then wait for them to finish writing their responses. Without this, // a finite input stream (e.g. piped ndjson that EOFs right after a request) // would race the dispatch goroutine and drop the response. defer func() { - endRead(false) cancel() - readerWatcher.Wait() c.wg.Wait() }() + interruptible := newInterruptibleReader(c.rawReader, c.readerCloser, ctx) + reader := bufio.NewReader(interruptible) + for { // One ndjson value per line. Framing per line (rather than streaming the // decoder) keeps a single malformed line from making the whole connection // unrecoverable — we report -32700 and continue. - line, err := c.reader.ReadBytes('\n') - if err != nil { - // Claim terminal completion before handling a partial final frame so - // concurrent cancellation cannot close the reader or mask this error. - endRead(false) - } + // + // generation brackets this SPECIFIC ReadBytes call: bufio may invoke + // interruptible.Read zero or more times underneath it (zero when the + // answer is already sitting in bufio's own buffer — including a + // previously-buffered, not-yet-surfaced error: bufio can return a + // complete line with a nil error while quietly caching an error it + // received ALONGSIDE that line's bytes, which then surfaces on the VERY + // NEXT call with no further contact with interruptible.Read at all). The + // generation only advances inside interruptible.Read's own cancellation + // branch, so "unchanged across this call" is proof this call's outcome + // didn't pass through that branch — whether because it was answered from + // the buffer or because it raced ctx.Done() and won. Only that proof + // makes it safe to call the outcome genuine. + before := interruptible.generation() + line, err := reader.ReadBytes('\n') + interrupted := interruptible.generation() != before + if len(bytes.TrimSpace(line)) > 0 { c.handleLine(ctx, line) } if err != nil { c.failAllPending(err) - if errors.Is(err, io.EOF) || readInterrupted { + if errors.Is(err, io.EOF) || interrupted { return nil } return err @@ -180,6 +170,74 @@ func (c *Conn) Serve(ctx context.Context) error { } } +// interruptibleReader wraps a reader so a context cancellation can unblock a +// currently in-flight Read without ever discarding — or misattributing — the +// real outcome of a call that wasn't actually blocked. +// +// Each Read runs the underlying call on its own goroutine and races it against +// ctx.Done(). If the underlying call finishes first, its result is returned +// exactly as received; ctx being cancelled a moment later has no bearing on a +// call that already had its answer. If ctx.Done() fires first, closer (when +// non-nil) is closed to nudge a genuinely blocked read, the generation counter +// is bumped, and the call STILL waits for and returns the underlying call's own +// result — never a synthesized one — since closing a reader's outcome is +// whatever the reader itself reports for it, not ours to invent. +// +// The counter is what lets Serve tell a call that was actually raced against +// cancellation apart from one merely answered while cancellation happened to be +// pending — the distinction the read loop's own doc comment explains. +type interruptibleReader struct { + inner io.Reader + closer io.Closer + ctx context.Context + + closeOnce sync.Once + + mu sync.Mutex + gen int64 +} + +func newInterruptibleReader(inner io.Reader, closer io.Closer, ctx context.Context) *interruptibleReader { + return &interruptibleReader{inner: inner, closer: closer, ctx: ctx} +} + +func (r *interruptibleReader) generation() int64 { + r.mu.Lock() + defer r.mu.Unlock() + return r.gen +} + +type interruptibleReadResult struct { + n int + err error +} + +func (r *interruptibleReader) Read(p []byte) (int, error) { + resultCh := make(chan interruptibleReadResult, 1) + go func() { + n, err := r.inner.Read(p) + resultCh <- interruptibleReadResult{n, err} + }() + select { + case res := <-resultCh: + return res.n, res.err + case <-r.ctx.Done(): + r.mu.Lock() + r.gen++ + r.mu.Unlock() + if r.closer != nil { + r.closeOnce.Do(func() { _ = r.closer.Close() }) + } + // Wait for the real result rather than returning synthetically: p is + // still being written by the goroutine above until it does, so returning + // early would race that write, and the actual error (whatever closing + // the reader produced, or whatever the reader was already about to + // report) is what the caller should see. + res := <-resultCh + return res.n, res.err + } +} + // handleLine decodes and dispatches one ndjson frame. A parse failure replies // with -32700 (id null) and keeps the connection alive. func (c *Conn) handleLine(ctx context.Context, line []byte) { diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index 9e24a7597..3de35659c 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "io" + "sync/atomic" "testing" "time" ) @@ -45,11 +46,17 @@ func connPair(t *testing.T) (a, b *Conn, stop func()) { func TestConnServeCancellationInterruptsIdleRead(t *testing.T) { pipeReader, writer := io.Pipe() t.Cleanup(func() { _ = writer.Close() }) - readStarted := make(chan struct{}) + // Buffered + non-blocking send, not close: interruptibleReader retries each + // underlying Read on its own goroutine, so if bufio ever calls this more than + // once, a close() here would panic on the second call. + readStarted := make(chan struct{}, 1) closeCalls := make(chan struct{}, 2) reader := testReadCloser{ read: func(p []byte) (int, error) { - close(readStarted) + select { + case readStarted <- struct{}{}: + default: + } return pipeReader.Read(p) }, close: func() error { @@ -99,10 +106,16 @@ func TestConnServePreservesTerminalReadErrorDuringCancellation(t *testing.T) { }, } } - writeStarted := make(chan struct{}) + // Buffered + non-blocking send, not close: a later change adding a + // second write (e.g. an additional error frame) must not panic on a + // repeated close of an already-closed channel. + writeStarted := make(chan struct{}, 1) releaseWrite := make(chan struct{}) writer := testWriter(func(p []byte) (int, error) { - close(writeStarted) + select { + case writeStarted <- struct{}{}: + default: + } <-releaseWrite return len(p), nil }) @@ -129,6 +142,75 @@ func TestConnServePreservesTerminalReadErrorDuringCancellation(t *testing.T) { } } +// TestConnServePreservesReadErrorBufferedAlongsideAPriorLine is the deterministic +// regression test for jatmn's #782 finding: a real, non-EOF ReadBytes failure can +// be DECIDED before cancellation ever happens, yet only SURFACE afterward, and +// must still be reported rather than swallowed as a clean shutdown. +// +// bufio.Reader can obtain data AND a terminal error from ONE underlying Read call +// (io.Reader explicitly permits returning (n>0, err) together). If a delimiter is +// found within that data, ReadBytes returns the line with a nil error for THIS +// call, but bufio caches the error internally and returns it — WITHOUT calling the +// underlying reader again — on the very next ReadBytes call. This test forces that +// exact sequence: the underlying reader hands back one complete, validly-framed +// line together with wantErr in a SINGLE call (asserted below to be its ONLY +// call), the loop dispatches that line through a synchronous path that blocks the +// read loop, cancellation is asserted to have happened before the loop attempts +// its next read, and only then is the block released. By the time the read loop +// asks for more input, cancellation is already in effect and the underlying +// reader is never touched again — so if Serve attributed the resulting error to +// cancellation merely because ctx was already done, it would incorrectly return +// nil. It must instead recognize that this read was answered entirely from +// bufio's own cache, never raced against ctx.Done(), and report wantErr. +func TestConnServePreservesReadErrorBufferedAlongsideAPriorLine(t *testing.T) { + wantErr := errors.New("read failed") + var readCalls atomic.Int32 + read := testReader(func(p []byte) (int, error) { + if n := readCalls.Add(1); n > 1 { + panic("underlying Read called more than once; bufio should have served the cached error") + } + // An unsupported jsonrpc version with an id takes the SYNCHRONOUS + // writeError path in handleLine (not a dispatched goroutine), giving this + // test a reliable blocking point between processing this line and the + // loop's next ReadBytes call. + return copy(p, `{"jsonrpc":"1.0","id":1}`+"\n"), wantErr + }) + writeStarted := make(chan struct{}, 1) + releaseWrite := make(chan struct{}) + writer := testWriter(func(p []byte) (int, error) { + select { + case writeStarted <- struct{}{}: + default: + } + <-releaseWrite + return len(p), nil + }) + conn := NewConn(read, writer) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- conn.Serve(ctx) }() + + // The synchronous write confirms the first (and, per the panic guard above, + // only) read has already returned and is being processed — cancelling now + // lands squarely in the gap between that read and the loop's next one, never + // racing an in-flight interruptibleReader.Read call. + <-writeStarted + cancel() + close(releaseWrite) + + select { + case err := <-done: + if !errors.Is(err, wantErr) { + t.Fatalf("Serve returned %v, want the buffered terminal read error %v", err, wantErr) + } + case <-time.After(time.Second): + t.Fatal("Serve did not return after the buffered terminal read error") + } + if got := readCalls.Load(); got != 1 { + t.Fatalf("underlying Read calls = %d, want exactly 1 (bufio must have served the second ReadBytes from its cache)", got) + } +} + func TestConnRequestResponse(t *testing.T) { a, b, stop := connPair(t) defer stop() From 6dbe1d131d2c82244112e07928e19720d6684553 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sun, 26 Jul 2026 12:58:48 +0200 Subject: [PATCH 3/8] fix(acp): preserve decided read results and require explicit reader ownership Addresses jatmn's two #782 review findings: - interruptibleReader.Read's select between resultCh and ctx.Done() chooses pseudo-randomly when both are ready, so a real terminal read error decided just before cancellation could be misattributed to shutdown and silently swallowed. Re-check resultCh non-blockingly before claiming a call for cancellation. - NewConn inferred reader ownership from an io.Closer assertion and closed any reader that happened to implement it, including one a caller merely borrowed. Split into NewConn (never closes r) and NewOwnedConn (explicit ownership, closes r on cancellation), and switch the sole production call site (process stdin) to the latter. Also applies CodeRabbit's atomic.Int64 and leading-ctx-parameter nits. Co-Authored-By: Claude Sonnet 5 --- internal/acp/jsonrpc.go | 63 +++++++++++++++++++++++------------- internal/acp/jsonrpc_test.go | 52 +++++++++++++++++++++++++++-- internal/cli/acp.go | 4 ++- 3 files changed, 93 insertions(+), 26 deletions(-) diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go index 1e7172390..c79e17c27 100644 --- a/internal/acp/jsonrpc.go +++ b/internal/acp/jsonrpc.go @@ -17,6 +17,7 @@ import ( "fmt" "io" "sync" + "sync/atomic" ) // JSON-RPC 2.0 standard error codes. @@ -97,21 +98,34 @@ type Conn struct { wg sync.WaitGroup // tracks in-flight inbound handlers } -// NewConn builds a peer reading ndjson from r and writing ndjson to w. If r is -// closable, Serve closes it to interrupt an idle read when its context is -// cancelled. +// NewConn builds a peer reading ndjson from r and writing ndjson to w. This +// Conn does not own r: cancelling Serve's context cannot forcibly close it, so +// a Read on r that's genuinely idle (never returns on its own) leaves Serve +// blocked until r itself produces something — e.g. a test that closes r's own +// write side to unblock it. Use NewOwnedConn when this Conn should own r's +// lifetime. func NewConn(r io.Reader, w io.Writer) *Conn { - readerCloser, _ := r.(io.Closer) return &Conn{ - rawReader: r, - readerCloser: readerCloser, - w: w, - handlers: make(map[string]HandlerFunc), - notifiers: make(map[string]NotifyFunc), - pending: make(map[int64]chan rpcMessage), + rawReader: r, + w: w, + handlers: make(map[string]HandlerFunc), + notifiers: make(map[string]NotifyFunc), + pending: make(map[int64]chan rpcMessage), } } +// NewOwnedConn is like NewConn but declares that this Conn exclusively owns r: +// if r implements io.Closer, Serve closes it on context cancellation to +// interrupt an otherwise-idle blocking Read. Only pass a reader whose lifetime +// this Conn should control — e.g. the process's own stdin — never a reader a +// caller retains or shares elsewhere, since cancellation tears it down +// permanently for every consumer, not just this Conn. +func NewOwnedConn(r io.Reader, w io.Writer) *Conn { + c := NewConn(r, w) + c.readerCloser, _ = r.(io.Closer) + return c +} + // Handle registers a request handler for method. func (c *Conn) Handle(method string, fn HandlerFunc) { c.handlers[method] = fn } @@ -133,7 +147,7 @@ func (c *Conn) Serve(ctx context.Context) error { c.wg.Wait() }() - interruptible := newInterruptibleReader(c.rawReader, c.readerCloser, ctx) + interruptible := newInterruptibleReader(ctx, c.rawReader, c.readerCloser) reader := bufio.NewReader(interruptible) for { @@ -192,20 +206,14 @@ type interruptibleReader struct { ctx context.Context closeOnce sync.Once - - mu sync.Mutex - gen int64 + gen atomic.Int64 } -func newInterruptibleReader(inner io.Reader, closer io.Closer, ctx context.Context) *interruptibleReader { +func newInterruptibleReader(ctx context.Context, inner io.Reader, closer io.Closer) *interruptibleReader { return &interruptibleReader{inner: inner, closer: closer, ctx: ctx} } -func (r *interruptibleReader) generation() int64 { - r.mu.Lock() - defer r.mu.Unlock() - return r.gen -} +func (r *interruptibleReader) generation() int64 { return r.gen.Load() } type interruptibleReadResult struct { n int @@ -222,9 +230,18 @@ func (r *interruptibleReader) Read(p []byte) (int, error) { case res := <-resultCh: return res.n, res.err case <-r.ctx.Done(): - r.mu.Lock() - r.gen++ - r.mu.Unlock() + // select chooses pseudo-randomly among ready cases: resultCh may already + // hold the real, decided outcome even though ctx.Done() became ready at + // essentially the same moment (e.g. a terminal transport error racing + // cancellation). Re-check it non-blockingly before claiming this call for + // cancellation — only a call whose result truly wasn't ready yet may be + // attributed to the ctx.Done() branch below. + select { + case res := <-resultCh: + return res.n, res.err + default: + } + r.gen.Add(1) if r.closer != nil { r.closeOnce.Do(func() { _ = r.closer.Close() }) } diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index 3de35659c..80465e213 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -64,7 +64,7 @@ func TestConnServeCancellationInterruptsIdleRead(t *testing.T) { return pipeReader.Close() }, } - conn := NewConn(reader, io.Discard) + conn := NewOwnedConn(reader, io.Discard) ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- conn.Serve(ctx) }() @@ -119,7 +119,14 @@ func TestConnServePreservesTerminalReadErrorDuringCancellation(t *testing.T) { <-releaseWrite return len(p), nil }) - conn := NewConn(reader, writer) + newConn := NewConn + if closable { + // Ownership matters here: this case exists to prove that even a + // Conn entitled to close its reader on cancellation still + // preserves a terminal error that was already decided. + newConn = NewOwnedConn + } + conn := newConn(reader, writer) ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- conn.Serve(ctx) }() @@ -211,6 +218,47 @@ func TestConnServePreservesReadErrorBufferedAlongsideAPriorLine(t *testing.T) { } } +// TestInterruptibleReaderPrefersDecidedResultOverCancellation is the regression +// test for jatmn's second #782 finding: interruptibleReader.Read's select +// between resultCh and ctx.Done() chooses pseudo-randomly when both are ready, +// so a real, already-decided outcome (e.g. a terminal transport error) could be +// misattributed to cancellation purely because ctx also happened to be done by +// the time the select ran. Each iteration lets the underlying Read return (so +// resultCh already holds the real outcome) before cancelling, maximizing the +// odds of landing exactly in that ambiguous window; the generation counter must +// never advance when the real, already-decided result is what gets returned. +func TestInterruptibleReaderPrefersDecidedResultOverCancellation(t *testing.T) { + wantErr := errors.New("read failed") + for i := 0; i < 100; i++ { + resultReady := make(chan struct{}) + read := testReader(func(p []byte) (int, error) { + defer close(resultReady) + return copy(p, "x"), wantErr + }) + ctx, cancel := context.WithCancel(context.Background()) + r := newInterruptibleReader(ctx, read, nil) + + readDone := make(chan interruptibleReadResult, 1) + go func() { + n, err := r.Read(make([]byte, 8)) + readDone <- interruptibleReadResult{n, err} + }() + + <-resultReady + time.Sleep(time.Millisecond) // bias resultCh's send ahead of cancel + cancel() + + res := <-readDone + if !errors.Is(res.err, wantErr) { + t.Fatalf("iteration %d: Read returned %v, want %v", i, res.err, wantErr) + } + if got := r.generation(); got != 0 { + t.Fatalf("iteration %d: generation = %d, want 0 (a decided result must never be attributed to cancellation)", i, got) + } + cancel() + } +} + func TestConnRequestResponse(t *testing.T) { a, b, stop := connPair(t) defer stop() diff --git a/internal/cli/acp.go b/internal/cli/acp.go index dfade3b32..3eb2e0c22 100644 --- a/internal/cli/acp.go +++ b/internal/cli/acp.go @@ -42,7 +42,9 @@ func runACP(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) int } } - conn := acp.NewConn(deps.stdin, stdout) + // The process's own stdin belongs exclusively to this Conn for the life of + // the command, so cancellation may close it to interrupt an idle read. + conn := acp.NewOwnedConn(deps.stdin, stdout) acp.NewAgent(conn, acp.Deps{ ResolveConfig: deps.resolveConfig, DiscoverModels: func(ctx context.Context, profile config.ProviderProfile) ([]providermodeldiscovery.Model, error) { From c981ed57398d93f8826c88d81059a7fee8f73187 Mon Sep 17 00:00:00 2001 From: Amp Date: Sun, 26 Jul 2026 20:49:12 +0000 Subject: [PATCH 4/8] fix(acp): synchronize read completion with cancellation Amp-Thread-ID: https://ampcode.com/threads/T-019fa018-9e4e-7506-85ab-e151888bfd71 Co-authored-by: Pierre Bruno --- internal/acp/jsonrpc.go | 39 +++++++++++++++++++++++------------- internal/acp/jsonrpc_test.go | 24 +++++++++++++++++++--- 2 files changed, 46 insertions(+), 17 deletions(-) diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go index c79e17c27..ac66a8315 100644 --- a/internal/acp/jsonrpc.go +++ b/internal/acp/jsonrpc.go @@ -220,30 +220,41 @@ type interruptibleReadResult struct { err error } +// interruptibleReadDecision arbitrates whether an underlying Read completed or +// cancellation claimed it first. Completion is recorded before the result is +// published so cancellation cannot misclassify a finished read merely because +// its goroutine has not sent the result yet. +type interruptibleReadDecision struct { + once sync.Once +} + +func (d *interruptibleReadDecision) complete() { d.once.Do(func() {}) } + +func (d *interruptibleReadDecision) interrupt() (claimed bool) { + d.once.Do(func() { claimed = true }) + return claimed +} + func (r *interruptibleReader) Read(p []byte) (int, error) { resultCh := make(chan interruptibleReadResult, 1) + var decision interruptibleReadDecision go func() { n, err := r.inner.Read(p) + decision.complete() resultCh <- interruptibleReadResult{n, err} }() select { case res := <-resultCh: return res.n, res.err case <-r.ctx.Done(): - // select chooses pseudo-randomly among ready cases: resultCh may already - // hold the real, decided outcome even though ctx.Done() became ready at - // essentially the same moment (e.g. a terminal transport error racing - // cancellation). Re-check it non-blockingly before claiming this call for - // cancellation — only a call whose result truly wasn't ready yet may be - // attributed to the ctx.Done() branch below. - select { - case res := <-resultCh: - return res.n, res.err - default: - } - r.gen.Add(1) - if r.closer != nil { - r.closeOnce.Do(func() { _ = r.closer.Close() }) + // resultCh may not be readable yet even when inner.Read has returned: its + // goroutine can be descheduled between recording completion and publishing + // the result. Only interrupt a call that cancellation actually claims. + if decision.interrupt() { + r.gen.Add(1) + if r.closer != nil { + r.closeOnce.Do(func() { _ = r.closer.Close() }) + } } // Wait for the real result rather than returning synthetically: p is // still being written by the goroutine above until it does, so returning diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index 80465e213..b3fbf80a1 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -173,9 +173,7 @@ func TestConnServePreservesReadErrorBufferedAlongsideAPriorLine(t *testing.T) { wantErr := errors.New("read failed") var readCalls atomic.Int32 read := testReader(func(p []byte) (int, error) { - if n := readCalls.Add(1); n > 1 { - panic("underlying Read called more than once; bufio should have served the cached error") - } + readCalls.Add(1) // An unsupported jsonrpc version with an id takes the SYNCHRONOUS // writeError path in handleLine (not a dispatched goroutine), giving this // test a reliable blocking point between processing this line and the @@ -259,6 +257,26 @@ func TestInterruptibleReaderPrefersDecidedResultOverCancellation(t *testing.T) { } } +// TestInterruptibleReadCompletionWinsBeforeResultPublication covers the gap +// after inner.Read returns and claims completion but before its goroutine sends +// the result. Cancellation must not reclassify that finished read as interrupted. +func TestInterruptibleReadCompletionWinsBeforeResultPublication(t *testing.T) { + var decision interruptibleReadDecision + decision.complete() + + // An unbuffered send cannot publish until the receive below, leaving the + // result deliberately unpublished while cancellation tries to claim the call. + resultCh := make(chan interruptibleReadResult) + go func() { + resultCh <- interruptibleReadResult{err: errors.New("read failed")} + }() + claimed := decision.interrupt() + <-resultCh + if claimed { + t.Fatal("cancellation claimed a read that completed before result publication") + } +} + func TestConnRequestResponse(t *testing.T) { a, b, stop := connPair(t) defer stop() From 30d54982e6e1c10114b5dc610501d932c780ea19 Mon Sep 17 00:00:00 2001 From: Amp Date: Mon, 27 Jul 2026 21:20:59 +0000 Subject: [PATCH 5/8] fix(acp): preserve read completion across cancellation Amp-Thread-ID: https://ampcode.com/threads/T-019fa55a-483a-7119-92aa-4953a72f2a76 Co-authored-by: Pierre Bruno --- internal/acp/jsonrpc.go | 18 ++++++++++--- internal/acp/jsonrpc_test.go | 50 ++++++++++++++++++++++++++++-------- 2 files changed, 54 insertions(+), 14 deletions(-) diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go index ac66a8315..7d9171692 100644 --- a/internal/acp/jsonrpc.go +++ b/internal/acp/jsonrpc.go @@ -235,13 +235,25 @@ func (d *interruptibleReadDecision) interrupt() (claimed bool) { return claimed } +// readInner records completion in the Read call's deferred epilogue. It also +// converts a reader panic into a terminal error so the helper goroutine always +// publishes a result for Read to join. +func (r *interruptibleReader) readInner(p []byte, decision *interruptibleReadDecision) (res interruptibleReadResult) { + defer func() { + if recovered := recover(); recovered != nil { + res = interruptibleReadResult{err: fmt.Errorf("acp: reader panicked: %v", recovered)} + } + decision.complete() + }() + res.n, res.err = r.inner.Read(p) + return res +} + func (r *interruptibleReader) Read(p []byte) (int, error) { resultCh := make(chan interruptibleReadResult, 1) var decision interruptibleReadDecision go func() { - n, err := r.inner.Read(p) - decision.complete() - resultCh <- interruptibleReadResult{n, err} + resultCh <- r.readInner(p, &decision) }() select { case res := <-resultCh: diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index b3fbf80a1..ce8e2f009 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -257,24 +257,52 @@ func TestInterruptibleReaderPrefersDecidedResultOverCancellation(t *testing.T) { } } -// TestInterruptibleReadCompletionWinsBeforeResultPublication covers the gap -// after inner.Read returns and claims completion but before its goroutine sends -// the result. Cancellation must not reclassify that finished read as interrupted. +// TestInterruptibleReadCompletionWinsBeforeResultPublication exercises the +// actual inner-read wrapper at the handoff between the underlying Read and the +// helper goroutine's result publication. Cancellation must not reclassify a +// result returned by this wrapper as interrupted. func TestInterruptibleReadCompletionWinsBeforeResultPublication(t *testing.T) { + wantErr := errors.New("read failed") + read := testReader(func(p []byte) (int, error) { + return copy(p, "x"), wantErr + }) + r := newInterruptibleReader(context.Background(), read, nil) var decision interruptibleReadDecision - decision.complete() - - // An unbuffered send cannot publish until the receive below, leaving the - // result deliberately unpublished while cancellation tries to claim the call. + readComplete := make(chan struct{}) resultCh := make(chan interruptibleReadResult) go func() { - resultCh <- interruptibleReadResult{err: errors.New("read failed")} + res := r.readInner(make([]byte, 8), &decision) + close(readComplete) + resultCh <- res }() - claimed := decision.interrupt() - <-resultCh - if claimed { + + <-readComplete + if decision.interrupt() { t.Fatal("cancellation claimed a read that completed before result publication") } + res := <-resultCh + if res.n != 1 || !errors.Is(res.err, wantErr) { + t.Fatalf("inner read result = (%d, %v), want (1, %v)", res.n, res.err, wantErr) + } +} + +func TestConnServeReportsReaderPanicAsTerminalError(t *testing.T) { + read := testReader(func([]byte) (int, error) { + panic("boom") + }) + conn := NewConn(read, io.Discard) + done := make(chan error, 1) + go func() { done <- conn.Serve(context.Background()) }() + + select { + case err := <-done: + const want = "acp: reader panicked: boom" + if err == nil || err.Error() != want { + t.Fatalf("Serve returned %v, want %q", err, want) + } + case <-time.After(time.Second): + t.Fatal("Serve did not return after reader panic") + } } func TestConnRequestResponse(t *testing.T) { From f7cdb741565dbb302f8ae363062a38e42177c151 Mon Sep 17 00:00:00 2001 From: Amp Date: Mon, 27 Jul 2026 21:30:39 +0000 Subject: [PATCH 6/8] test(acp): bound buffered error synchronization Amp-Thread-ID: https://ampcode.com/threads/T-019fa55a-483a-7119-92aa-4953a72f2a76 Co-authored-by: Pierre Bruno --- internal/acp/jsonrpc_test.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go index ce8e2f009..7d3b0c9da 100644 --- a/internal/acp/jsonrpc_test.go +++ b/internal/acp/jsonrpc_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "io" + "sync" "sync/atomic" "testing" "time" @@ -182,6 +183,9 @@ func TestConnServePreservesReadErrorBufferedAlongsideAPriorLine(t *testing.T) { }) writeStarted := make(chan struct{}, 1) releaseWrite := make(chan struct{}) + var releaseWriteOnce sync.Once + release := func() { releaseWriteOnce.Do(func() { close(releaseWrite) }) } + t.Cleanup(release) writer := testWriter(func(p []byte) (int, error) { select { case writeStarted <- struct{}{}: @@ -199,9 +203,13 @@ func TestConnServePreservesReadErrorBufferedAlongsideAPriorLine(t *testing.T) { // only) read has already returned and is being processed — cancelling now // lands squarely in the gap between that read and the loop's next one, never // racing an in-flight interruptibleReader.Read call. - <-writeStarted + select { + case <-writeStarted: + case <-time.After(time.Second): + t.Fatal("Serve did not reach the synchronous write") + } cancel() - close(releaseWrite) + release() select { case err := <-done: From 7791cf34e354ce48786745186dcdfe6b4c8b6082 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 10:40:37 +0000 Subject: [PATCH 7/8] fix(acp): report shutdown transport errors Amp-Thread-ID: https://ampcode.com/threads/T-019fa7a0-1223-701d-9529-48ba5d7cf8c8 Co-authored-by: Pierre Bruno --- internal/cli/acp.go | 2 +- internal/cli/acp_test.go | 119 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 1 deletion(-) create mode 100644 internal/cli/acp_test.go diff --git a/internal/cli/acp.go b/internal/cli/acp.go index 3eb2e0c22..2c04be029 100644 --- a/internal/cli/acp.go +++ b/internal/cli/acp.go @@ -77,7 +77,7 @@ func runACP(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) int ctx, stop := signalContext() defer stop() - if err := conn.Serve(ctx); err != nil && ctx.Err() == nil { + if err := conn.Serve(ctx); err != nil { return writeAppError(stderr, "acp: "+err.Error(), exitCrash) } return exitSuccess diff --git a/internal/cli/acp_test.go b/internal/cli/acp_test.go new file mode 100644 index 000000000..3ce213980 --- /dev/null +++ b/internal/cli/acp_test.go @@ -0,0 +1,119 @@ +package cli + +import ( + "bytes" + "errors" + "io" + "os" + "strings" + "sync" + "testing" + "time" +) + +type acpTestReader func([]byte) (int, error) + +func (r acpTestReader) Read(p []byte) (int, error) { return r(p) } + +type acpTestWriter func([]byte) (int, error) + +func (w acpTestWriter) Write(p []byte) (int, error) { return w(p) } + +func TestRunACPCancellationPreservesTerminalReadError(t *testing.T) { + wantErr := errors.New("transport read failed") + reader := acpTestReader(func(p []byte) (int, error) { + return copy(p, `{"jsonrpc":"1.0","id":1}`+"\n"), wantErr + }) + writeStarted := make(chan struct{}, 1) + releaseWrite := make(chan struct{}) + var stdout bytes.Buffer + writer := acpTestWriter(func(p []byte) (int, error) { + writeStarted <- struct{}{} + <-releaseWrite + return stdout.Write(p) + }) + var stderr bytes.Buffer + done := make(chan int, 1) + go func() { + done <- runACP(nil, writer, &stderr, fillAppDeps(appDeps{stdin: reader})) + }() + + select { + case <-writeStarted: + case <-time.After(time.Second): + t.Fatal("ACP did not reach the response write") + } + if err := signalInterrupt(); err != nil { + close(releaseWrite) + t.Fatalf("send interrupt: %v", err) + } + // Keep Serve in the synchronous write until signal.NotifyContext has had a + // chance to cancel its context; the regression requires the error to surface + // while cancellation is already observable by runACP. + time.Sleep(50 * time.Millisecond) + close(releaseWrite) + + select { + case code := <-done: + if code != exitCrash { + t.Fatalf("exit code = %d, want crash %d", code, exitCrash) + } + case <-time.After(time.Second): + t.Fatal("ACP did not exit after terminal read error") + } + if got := stderr.String(); !strings.Contains(got, "acp: "+wantErr.Error()) { + t.Fatalf("stderr = %q, want terminal read error", got) + } +} + +func TestRunACPSIGINTStillExitsCleanly(t *testing.T) { + pipeReader, pipeWriter := io.Pipe() + t.Cleanup(func() { _ = pipeWriter.Close() }) + readStarted := make(chan struct{}, 1) + reader := &acpNotifyingReadCloser{ReadCloser: pipeReader, readStarted: readStarted} + var stdout, stderr bytes.Buffer + done := make(chan int, 1) + go func() { + done <- runACP(nil, &stdout, &stderr, fillAppDeps(appDeps{stdin: reader})) + }() + + select { + case <-readStarted: + case <-time.After(time.Second): + t.Fatal("ACP did not begin reading") + } + if err := signalInterrupt(); err != nil { + t.Fatalf("send interrupt: %v", err) + } + + select { + case code := <-done: + if code != exitSuccess { + t.Fatalf("exit code = %d, want success %d; stderr: %s", code, exitSuccess, stderr.String()) + } + case <-time.After(time.Second): + t.Fatal("ACP did not exit after SIGINT") + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } +} + +type acpNotifyingReadCloser struct { + io.ReadCloser + readStarted chan<- struct{} + once sync.Once +} + +func (r *acpNotifyingReadCloser) Read(p []byte) (int, error) { + r.once.Do(func() { r.readStarted <- struct{}{} }) + return r.ReadCloser.Read(p) +} + +func signalInterrupt() error { + process, err := os.FindProcess(os.Getpid()) + if err != nil { + return err + } + return process.Signal(os.Interrupt) +} From a6e25308fd0ef67e6e50ab5ff6167e095ad27e06 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 10:52:41 +0000 Subject: [PATCH 8/8] test(acp): inject cancellation context Amp-Thread-ID: https://ampcode.com/threads/T-019fa7a0-1223-701d-9529-48ba5d7cf8c8 Co-authored-by: Pierre Bruno --- internal/cli/acp.go | 4 +++- internal/cli/acp_test.go | 45 ++++++++++++++++++++-------------------- 2 files changed, 26 insertions(+), 23 deletions(-) diff --git a/internal/cli/acp.go b/internal/cli/acp.go index 2c04be029..c261c19ef 100644 --- a/internal/cli/acp.go +++ b/internal/cli/acp.go @@ -27,6 +27,8 @@ Usage: Not meant to be run interactively — point your editor's ACP / external-agent setting at "zero acp".` +var acpSignalContext = signalContext + // runACP serves ACP over stdio so an editor can drive ZERO's agent core. It // speaks JSON-RPC 2.0 (newline-delimited JSON) on stdin/stdout; stderr stays free // for human-readable diagnostics. The session lifecycle maps onto ZERO's own @@ -75,7 +77,7 @@ func runACP(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) int AgentInfo: acp.Implementation{Name: "zero", Version: version}, }) - ctx, stop := signalContext() + ctx, stop := acpSignalContext() defer stop() if err := conn.Serve(ctx); err != nil { return writeAppError(stderr, "acp: "+err.Error(), exitCrash) diff --git a/internal/cli/acp_test.go b/internal/cli/acp_test.go index 3ce213980..c957432b8 100644 --- a/internal/cli/acp_test.go +++ b/internal/cli/acp_test.go @@ -2,9 +2,9 @@ package cli import ( "bytes" + "context" "errors" "io" - "os" "strings" "sync" "testing" @@ -19,7 +19,22 @@ type acpTestWriter func([]byte) (int, error) func (w acpTestWriter) Write(p []byte) (int, error) { return w(p) } +func installACPTestContext(t *testing.T) context.CancelFunc { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + previous := acpSignalContext + acpSignalContext = func() (context.Context, context.CancelFunc) { + return ctx, cancel + } + t.Cleanup(func() { + cancel() + acpSignalContext = previous + }) + return cancel +} + func TestRunACPCancellationPreservesTerminalReadError(t *testing.T) { + cancel := installACPTestContext(t) wantErr := errors.New("transport read failed") reader := acpTestReader(func(p []byte) (int, error) { return copy(p, `{"jsonrpc":"1.0","id":1}`+"\n"), wantErr @@ -43,14 +58,9 @@ func TestRunACPCancellationPreservesTerminalReadError(t *testing.T) { case <-time.After(time.Second): t.Fatal("ACP did not reach the response write") } - if err := signalInterrupt(); err != nil { - close(releaseWrite) - t.Fatalf("send interrupt: %v", err) - } - // Keep Serve in the synchronous write until signal.NotifyContext has had a - // chance to cancel its context; the regression requires the error to surface - // while cancellation is already observable by runACP. - time.Sleep(50 * time.Millisecond) + // Keep Serve in the synchronous write until cancellation is observable. The + // regression requires the genuine read error to win over that cancellation. + cancel() close(releaseWrite) select { @@ -66,7 +76,8 @@ func TestRunACPCancellationPreservesTerminalReadError(t *testing.T) { } } -func TestRunACPSIGINTStillExitsCleanly(t *testing.T) { +func TestRunACPIdleCancellationExitsCleanly(t *testing.T) { + cancel := installACPTestContext(t) pipeReader, pipeWriter := io.Pipe() t.Cleanup(func() { _ = pipeWriter.Close() }) readStarted := make(chan struct{}, 1) @@ -82,9 +93,7 @@ func TestRunACPSIGINTStillExitsCleanly(t *testing.T) { case <-time.After(time.Second): t.Fatal("ACP did not begin reading") } - if err := signalInterrupt(); err != nil { - t.Fatalf("send interrupt: %v", err) - } + cancel() select { case code := <-done: @@ -92,7 +101,7 @@ func TestRunACPSIGINTStillExitsCleanly(t *testing.T) { t.Fatalf("exit code = %d, want success %d; stderr: %s", code, exitSuccess, stderr.String()) } case <-time.After(time.Second): - t.Fatal("ACP did not exit after SIGINT") + t.Fatal("ACP did not exit after cancellation") } if stderr.Len() != 0 { t.Fatalf("stderr = %q, want empty", stderr.String()) @@ -109,11 +118,3 @@ func (r *acpNotifyingReadCloser) Read(p []byte) (int, error) { r.once.Do(func() { r.readStarted <- struct{}{} }) return r.ReadCloser.Read(p) } - -func signalInterrupt() error { - process, err := os.FindProcess(os.Getpid()) - if err != nil { - return err - } - return process.Signal(os.Interrupt) -}