claude: Claude Code's output is read to the end before its run ends, not cut by Wait closing the pipe (CI Test run 37415124103 on 963a8d8f: TestClientGoneRecordedAsCanceled got event: error … "read |0: file already closed" in place of message_stop on ubuntu, race shard) — the run read Claude Code's stdout from cmd.StdoutPipe, and launch's goroutine called Wait as soon as Claude Code exited, which closes that pipe. A run stopped as its client went away (letGo → abort → Kill) could have Wait close the pipe under readOutput before abort's finish closed the segment: the scanner failed with os.ErrClosed, readOutput emitted it as a KError, and relay wrote it to the client as the stream's error. The same cut lost what Claude Code wrote last whenever it exited: whatever was still in the pipe never reached the client. Its stdout and stderr now come through os.Pipe pipes of the run's own, which Wait doesn't touch; once Claude Code has exited and its group been ended, the run waits for both readers (up to outputDrain, 2s, for a process that left the group and still holds them), closes the pipes, then finishes. A read cut by that close is taken as the output's end, not an error. TestClaudeLastWordsAreRead (a fake Claude Code writes ~300KB and exits) fails on the old code on every run (the reply's last delta lost) and passes; TestClientGoneRecordedAsCanceled|Keepalive|TestClaudeLastWordsAreRead pass -race -count=50 under GOMAXPROCS=2 with 4 busy cores; the gateway's race suite through build/race-shards.sh (4 shards, as CI) and go vet pass. The CI failure itself didn't reproduce on macOS. Not changed: the "http: panic serving …: invalid concurrent Body.Read call" lines in the same log are in shards that pass and in green runs too (5 in 37416877275), near the Copilot auto / count_tokens tests; their handler wasn't found.

internal/gateway/claude_output_drain_test.go

@@ -0,0 +1,50 @@
+package gateway
+
+import (
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"testing"
+
+	"github.com/yetone/magpie/internal/provider"
+)
+
+// What Claude Code writes as it exits reaches the client: its output is
+// read to the end before the run ends, not cut by Wait closing the pipe
+// under the reader ("read |0: file already closed", CI on 963a8d8f).
+func TestClaudeLastWordsAreRead(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("a shell script stands in for Claude Code")
+	}
+	dir := t.TempDir()
+	// a long reply, more than a pipe holds, written all at once on exit
+	script := `#!/bin/sh
+read -r line
+d='{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"` + strings.Repeat("x", 1000) + `"}}}'
+{
+echo '{"type":"stream_event","event":{"type":"message_start","message":{"id":"m","model":"claude-sonnet-5","usage":{"input_tokens":1}}}}'
+echo '{"type":"stream_event","event":{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}}'
+i=0; while [ $i -lt 300 ]; do echo "$d"; i=$((i+1)); done
+echo '{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"the very end"}}}'
+echo '{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":3}}}'
+echo '{"type":"stream_event","event":{"type":"message_stop"}}'
+}
+`
+	os.WriteFile(filepath.Join(dir, "claude"), []byte(script), 0o755)
+	t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	for i := 0; i < 5; i++ {
+		s := New()
+		p := provider.Provider{ID: "claude", Account: &provider.Account{Agent: "claude", User: "u"}}
+		body := `{"model":"claude-sonnet-5","max_tokens":100,"stream":true,"messages":[{"role":"user","content":"say a lot"}]}`
+		rec := httptest.NewRecorder()
+		var u Usage
+		s.serveClaudeSubscription(rec, httptest.NewRequest("POST", "/v1/messages", strings.NewReader(body)), provider.Anthropic, p, "claude-sonnet-5", []byte(body), &u)
+		s.subscription.abortAll()
+		out := rec.Body.String()
+		if !strings.Contains(out, "the very end") || strings.Contains(out, "event: error") {
+			t.Fatalf("run %d: the reply's end: %q", i, out[max(0, len(out)-400):])
+		}
+	}
+}

internal/gateway/claude_subscription.go

@@ -122,7 +122,10 @@ type subscriptionRun struct {
 	model  string
 	cmd    *exec.Cmd
 	tree   *proc.Tree // cmd once started, with all it starts
-	tmp    string
+	// the read ends of Claude Code's output, and their readers
+	outputs []*os.File
+	reading sync.WaitGroup
+	tmp     string
 	// schema says the client asked for an answer fitting a JSON schema,
 	// which Claude Code gives as its StructuredOutput call
 	schema bool
@@ -551,16 +554,22 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 		cleanup()
 		return nil, nil, err
 	}
-	stdout, err := cmd.StdoutPipe()
+	// Claude Code's output comes through pipes of the run's own, not
+	// cmd.StdoutPipe's, which Wait closes as Claude Code exits: what it
+	// wrote last, or the reading under way as it was stopped, failed
+	// "file already closed", and the client heard that for an error.
+	stdout, stdoutW, err := os.Pipe()
 	if err != nil {
 		cleanup()
 		return nil, nil, err
 	}
-	stderr, err := cmd.StderrPipe()
+	stderr, stderrW, err := os.Pipe()
 	if err != nil {
+		_, _ = stdout.Close(), stdoutW.Close()
 		cleanup()
 		return nil, nil, err
 	}
+	cmd.Stdout, cmd.Stderr = stdoutW, stderrW
 
 	run := &subscriptionRun{bridge: b, token: token, model: model, cmd: cmd, tmp: tmp, schema: len(req.Schema) > 0, pending: map[string]chan mcpToolResult{}, stdin: stdin, owner: owner, effort: req.Effort, sessions: sessions}
 	run.safeguardBeta = req.SafeguardBeta
@@ -575,14 +584,24 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 	b.runs[token] = run
 	b.mu.Unlock()
 
-	if err := run.launch(); err != nil {
+	run.outputs = []*os.File{stdout, stderr}
+	run.reading.Add(2)
+	err = run.launch()
+	_, _ = stdoutW.Close(), stderrW.Close() // Claude Code has its own
+	if err != nil {
+		_, _ = stdout.Close(), stderr.Close()
 		b.removeRun(run)
 		return nil, nil, err
 	}
 	go func() {
+		defer run.reading.Done()
 		_, _ = io.Copy(&lockedWriter{run: run}, io.LimitReader(stderr, 1<<20))
+		_, _ = io.Copy(io.Discard, stderr)
+	}()
+	go func() {
+		defer run.reading.Done()
+		run.readOutput(stdout)
 	}()
-	go run.readOutput(stdout)
 
 	if len(req.Safeguards) > 0 {
 		// Initialize the SDK before applying settings. Send the potentially
@@ -2117,7 +2136,7 @@ func (r *subscriptionRun) readOutput(rd io.Reader) {
 	pmu.Lock()
 	defer pmu.Unlock()
 	settle()
-	if err := s.Err(); err != nil {
+	if err := s.Err(); err != nil && !errors.Is(err, os.ErrClosed) {
 		r.emit(Event{Kind: KError, Text: err.Error()})
 	} else if cut {
 		failed("and Claude Code stopped")
@@ -2795,11 +2814,34 @@ func (r *subscriptionRun) launch() error {
 	r.mu.Unlock()
 	go func() {
 		_ = t.Wait()
+		r.drain()
 		r.finish()
 	}()
 	return nil
 }
 
+// outputDrain is how long the run's output is read once Claude Code has
+// exited and its group been ended: a process that left the group may
+// still hold the pipes, and is not waited for longer.
+var outputDrain = 2 * time.Second
+
+// drain waits for what Claude Code wrote to be read, so its last lines
+// reach the client before the run ends, then closes its pipes.
+func (r *subscriptionRun) drain() {
+	read := make(chan struct{})
+	go func() {
+		r.reading.Wait()
+		close(read)
+	}()
+	select {
+	case <-read:
+	case <-time.After(outputDrain):
+	}
+	for _, f := range r.outputs {
+		_ = f.Close()
+	}
+}
+
 func (r *subscriptionRun) abort() {
 	r.mu.Lock()
 	t := r.tree