test: synchronise every fixture-upstream capture a test goroutine reads

internal/cheapmodel/cheapmodel_test.go

@@ -19,19 +19,26 @@ func TestAnthropicSkipsNonTextLeadingBlock(t *testing.T) {
 	}
 }
 
+// authHeaders is what a fixture upstream saw on the wire. Handlers run on the test server's
+// own goroutine, so the values travel back over a buffered channel rather than through a
+// captured variable: the HTTP round trip is not a happens-before edge, and a plain capture
+// read by the test goroutine is a data race whether or not -race happens to observe it.
+type authHeaders struct{ auth, key, version string }
+
 func TestAnthropicBearerAuth(t *testing.T) {
-	var gotAuth, gotKey, gotVersion string
+	seen := make(chan authHeaders, 1)
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		gotAuth = r.Header.Get("Authorization")
-		gotKey = r.Header.Get("x-api-key")
-		gotVersion = r.Header.Get("anthropic-version")
+		seen <- authHeaders{r.Header.Get("Authorization"), r.Header.Get("x-api-key"),
+			r.Header.Get("anthropic-version")}
 		_, _ = io.WriteString(w, `{"content":[{"type":"text","text":"OK"}]}`)
 	}))
 	defer srv.Close()
 	_, err := Anthropic{BaseURL: srv.URL, APIKey: "tok", Model: "m", AuthScheme: "bearer"}.Complete(context.Background(), "p")
 	if err != nil {
 		t.Fatalf("err %v", err)
 	}
+	got := <-seen
+	gotAuth, gotKey, gotVersion := got.auth, got.key, got.version
 	if gotAuth != "Bearer tok" {
 		t.Fatalf("Authorization = %q, want %q", gotAuth, "Bearer tok")
 	}
@@ -44,17 +51,18 @@ func TestAnthropicBearerAuth(t *testing.T) {
 }
 
 func TestAnthropicDefaultAuthUsesAPIKey(t *testing.T) {
-	var gotAuth, gotKey string
+	seen := make(chan authHeaders, 1)
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		gotAuth = r.Header.Get("Authorization")
-		gotKey = r.Header.Get("x-api-key")
+		seen <- authHeaders{auth: r.Header.Get("Authorization"), key: r.Header.Get("x-api-key")}
 		_, _ = io.WriteString(w, `{"content":[{"type":"text","text":"OK"}]}`)
 	}))
 	defer srv.Close()
 	_, err := Anthropic{BaseURL: srv.URL, APIKey: "tok", Model: "m"}.Complete(context.Background(), "p")
 	if err != nil {
 		t.Fatalf("err %v", err)
 	}
+	got := <-seen
+	gotAuth, gotKey := got.auth, got.key
 	if gotKey != "tok" {
 		t.Fatalf("x-api-key = %q, want %q", gotKey, "tok")
 	}

proxy/adjudicatetool_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/tidwall/gjson"
@@ -32,9 +33,9 @@ func forwardedBody(t *testing.T, body []byte) []byte {
 // point is that one of those two differs.
 func forwardedOn(t *testing.T, route, yaml string, body []byte) []byte {
 	t.Helper()
-	var got []byte
+	var up upstreamCapture
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		got, _ = io.ReadAll(r.Body)
+		up.record(r)
 		w.Header().Set("Content-Type", "application/json")
 		if strings.Contains(route, "anthropic") {
 			_, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` +
@@ -53,7 +54,7 @@ func forwardedOn(t *testing.T, route, yaml string, body []byte) []byte {
 		t.Fatal(err)
 	}
 	resp.Body.Close()
-	return got
+	return up.last().body
 }
 
 // toolsRequest is the ANTHROPIC dialect, because that is the only dialect the injection targets:
@@ -210,11 +211,9 @@ func TestAdjudicateToolNotAdvertisedOnANonAnthropicRoute(t *testing.T) {
 // must instead be answered IN BAND, before the client is written to, which leaves the request-path
 // repair as a backstop rather than the primary defence.
 func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) {
-	round := 0
-	var second []byte
+	var up upstreamCapture
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, _ := io.ReadAll(r.Body)
-		round++
+		round := up.record(r)
 		w.Header().Set("Content-Type", "application/json")
 		if round == 1 {
 			// The model calls OUR tool, which is always a defect by construction.
@@ -224,7 +223,6 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) {
 				`"usage":{"input_tokens":5,"output_tokens":1}}`))
 			return
 		}
-		second = body
 		_, _ = w.Write([]byte(`{"id":"m2","type":"message","role":"assistant","model":"claude",` +
 			`"content":[{"type":"text","text":"done"}],"stop_reason":"end_turn",` +
 			`"usage":{"input_tokens":6,"output_tokens":2}}`))
@@ -242,10 +240,11 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) {
 	resp.Body.Close()
 
 	// PRECONDITION: the loop must actually have run a second round, or this asserts nothing.
-	if round < 2 {
+	if round := up.hits(); round < 2 {
 		t.Fatalf("the stray was never intercepted -- only %d upstream round(s), client got: %s",
 			round, got)
 	}
+	second := up.body(2)
 	if strings.Contains(string(got), adjudicate.ToolName) {
 		t.Errorf("the proxy-injected tool_use reached the CLIENT: %s", got)
 	}
@@ -279,10 +278,10 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) {
 	}}
 
 	// --- Turn 1: the leak. -------------------------------------------------------------------
-	rounds := 0
+	var rounds atomic.Int64 // an atomic carries the happens-before edge the round trip does not
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		_, _ = io.ReadAll(r.Body)
-		rounds++
+		rounds.Add(1)
 		w.Header().Set("Content-Type", "application/json")
 		blocks, _ := json.Marshal(assistantCoCall["content"])
 		_, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` +
@@ -303,7 +302,7 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) {
 
 	// PRECONDITION: exactly one upstream round. A continuation would mean the loop answered in band
 	// after all, and then the rest of this test is asserting nothing about the co-call path.
-	if rounds != 1 {
+	if rounds := rounds.Load(); rounds != 1 {
 		t.Fatalf("expected the loop to bail on otherTools after ONE round, got %d — the co-call path "+
 			"no longer defers, so this test's premise is gone: %s", rounds, leaked)
 	}
@@ -362,11 +361,11 @@ func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) {
 // THE LEAK, streaming path. The splicer withheld only the expand tool by name, so an adjudication call
 // streamed through event by event and the client saw it live.
 func TestAdjudicateStrayCallDoesNotReachTheClientOnTheSSEPath(t *testing.T) {
-	round := 0
+	var round atomic.Int64 // an atomic carries the happens-before edge the round trip does not
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		_, _ = io.ReadAll(r.Body)
-		round++
-		if round == 1 {
+		n := round.Add(1)
+		if n == 1 {
 			w.Header().Set("Content-Type", "text/event-stream")
 			w.WriteHeader(http.StatusOK)
 			for _, ev := range []string{
@@ -415,7 +414,7 @@ func TestAdjudicateStrayCallDoesNotReachTheClientOnTheSSEPath(t *testing.T) {
 	}
 	got, _ := io.ReadAll(resp.Body)
 	resp.Body.Close()
-	if round < 2 {
+	if round := round.Load(); round < 2 {
 		t.Fatalf("the streamed stray was never intercepted -- only %d round(s), client got: %s",
 			round, got)
 	}

proxy/agentcompaction_test.go

@@ -98,9 +98,9 @@ func TestAgentCompactionIsBypassed(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			var got []byte
+			var up upstreamCapture
 			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				got, _ = io.ReadAll(r.Body)
+				up.record(r)
 				w.Header().Set("Content-Type", "application/json")
 				w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`))
 			}))
@@ -126,6 +126,7 @@ func TestAgentCompactionIsBypassed(t *testing.T) {
 			}
 			resp.Body.Close()
 
+			got := up.last().body
 			if tc.wantBypass {
 				if !bytes.Equal(got, body) {
 					t.Fatalf("a compaction request must be forwarded byte-identical.\n sent: %s\n  got: %s", body, got)
@@ -221,9 +222,9 @@ func TestExpandToolAdvertisedOnEveryTurnButNeverOnACompaction(t *testing.T) {
 			[]map[string]any{marked, {"role": "user", "content": ccCompactPrompt}}, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			var got []byte
+			var up upstreamCapture
 			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				got, _ = io.ReadAll(r.Body)
+				up.record(r)
 				w.Header().Set("Content-Type", "application/json")
 				w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`))
 			}))
@@ -253,9 +254,9 @@ func TestExpandToolAdvertisedOnEveryTurnButNeverOnACompaction(t *testing.T) {
 			}
 			resp.Body.Close()
 
-			advertised := strings.Contains(string(got), "context_guru_expand")
+			advertised := strings.Contains(string(up.last().body), "context_guru_expand")
 			if advertised != tc.want {
-				t.Fatalf("expand tool advertised=%v, want %v: %s", advertised, tc.want, got)
+				t.Fatalf("expand tool advertised=%v, want %v: %s", advertised, tc.want, up.last().body)
 			}
 		})
 	}

proxy/dashexpand_test.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/rossoctl/context-guru/dash"
@@ -15,12 +16,11 @@ import (
 // tile read zero while /stats showed the true count, and Overview.SavedAdjusted
 // (SavedUnique − ExpandTokens) OVER-REPORTED net savings by the whole bounce.
 func TestDashboardRecordsExpands(t *testing.T) {
-	var calls int
+	var calls atomic.Int64 // an atomic carries the happens-before edge the round trip does not
 	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		io.Copy(io.Discard, r.Body)
-		calls++
 		w.Header().Set("Content-Type", "application/json")
-		if calls == 1 {
+		if calls.Add(1) == 1 {
 			// The model asks for the offloaded original back.
 			w.Write([]byte(`{"choices":[{"message":{"role":"assistant","tool_calls":[` +
 				`{"id":"call_1","type":"function","function":{"name":"context_guru_expand","arguments":"{\"id\":\"HASH\"}"}}` +
@@ -46,8 +46,8 @@ func TestDashboardRecordsExpands(t *testing.T) {
 		t.Fatal(err)
 	}
 	resp.Body.Close()
-	if calls != 2 {
-		t.Fatalf("expected the expand continuation (2 upstream calls), got %d", calls)
+	if n := calls.Load(); n != 2 {
+		t.Fatalf("expected the expand continuation (2 upstream calls), got %d", n)
 	}
 	waitForRows(t, rec, 1)
 

proxy/expandgate_test.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/tidwall/gjson"
-	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -59,9 +58,9 @@ func TestExpandToolIsAdvertisedOnlyWhereMarkersCanExist(t *testing.T) {
 		{"a pipeline with an offloader", "pipeline: [linecap, cachesplit]\n", true},
 	} {
 		t.Run(c.name, func(t *testing.T) {
-			var forwarded []byte
+			var up upstreamCapture
 			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				forwarded, _ = io.ReadAll(r.Body)
+				up.record(r)
 				w.Header().Set("Content-Type", "application/json")
 				fmt.Fprint(w, `{"type":"message","usage":{"input_tokens":1}}`)
 			}))
@@ -89,6 +88,7 @@ func TestExpandToolIsAdvertisedOnlyWhereMarkersCanExist(t *testing.T) {
 				t.Fatal(err)
 			}
 			resp.Body.Close()
+			forwarded := up.last().body
 			if len(forwarded) == 0 {
 				t.Fatal("upstream received nothing")
 			}

proxy/expandsplice_test.go

@@ -60,18 +60,15 @@ const round2Answer = "event: message_start\ndata: {\"type\":\"message_start\",\"
 func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) {
 	for _, leadType := range []string{"text", "thinking"} {
 		t.Run(leadType, func(t *testing.T) {
-			var calls int
-			var secondBody []byte
+			var up upstreamCapture
 			head, tail := leadThenExpand(leadType)
 			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				b, _ := io.ReadAll(r.Body)
-				calls++
+				round := up.record(r)
 				w.Header().Set("Content-Type", "text/event-stream")
-				if calls == 1 {
+				if round == 1 {
 					w.Write([]byte(head + tail))
 					return
 				}
-				secondBody = b
 				w.Write([]byte(round2Answer))
 			}))
 			defer upstream.Close()
@@ -93,9 +90,10 @@ func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) {
 				t.Fatalf("the client received a raw tool_use for OUR tool — this is the "+
 					"reported bug (`No such tool available: context_guru_expand`):\n%s", out)
 			}
-			if calls != 2 {
+			if calls := up.hits(); calls != 2 {
 				t.Fatalf("the expand call must drive a continuation round, got %d upstream calls", calls)
 			}
+			secondBody := up.body(2)
 			if !strings.Contains(string(secondBody), "THE ORIGINAL CONTENT") {
 				t.Fatalf("continuation must carry the resolved original: %s", secondBody)
 			}
@@ -135,13 +133,16 @@ func TestExpandCalledAfterALeadingBlockIsNeverGivenToTheClient(t *testing.T) {
 // cannot pass this, which is the whole difference between splicing and re-buffering.
 func TestTheStreamedPrefixReachesTheClientBeforeTheExpandCall(t *testing.T) {
 	release := make(chan struct{})
-	var calls int
+	// Two rounds, each served from a goroutine of upstream's own: the counter is read on a
+	// different goroutine from the one that last wrote it, never mind that the proxy issues the
+	// rounds one after the other. Which conn goroutine serves round 2 is net/http's business,
+	// not something the test may assume, so the counter carries its own edge.
+	var calls atomic.Int64
 	head, tail := leadThenExpand("text")
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		io.Copy(io.Discard, r.Body)
-		calls++
 		w.Header().Set("Content-Type", "text/event-stream")
-		if calls > 1 {
+		if calls.Add(1) > 1 {
 			w.Write([]byte(round2Answer))
 			return
 		}
@@ -202,9 +203,9 @@ func TestTheStreamedPrefixReachesTheClientBeforeTheExpandCall(t *testing.T) {
 // This is the whole sequence a user lives through, in two requests, because that is the only
 // place it is visible: the leak on the first and the repair on the second.
 func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) {
-	var lastUpstream []byte
+	var up upstreamCapture
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		lastUpstream, _ = io.ReadAll(r.Body)
+		up.record(r)
 		w.Header().Set("Content-Type", "text/event-stream")
 		w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"role\":\"assistant\"}}\n\n" +
 			"event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" +
@@ -260,6 +261,7 @@ func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) {
 	io.Copy(io.Discard, resp2.Body)
 	resp2.Body.Close()
 
+	lastUpstream := up.last().body
 	if strings.Contains(string(lastUpstream), "No such tool available") {
 		t.Fatalf("the model received the client's error for OUR tool:\n%s", lastUpstream)
 	}
@@ -277,9 +279,9 @@ func TestTheClientsNoSuchToolErrorNeverReachesTheModel(t *testing.T) {
 // while the response was buffered and nothing had been written, and is garbage appended to
 // the model's turn now that the prefix has already streamed.
 func TestAFailedContinuationRoundStillEndsTheClientsTurn(t *testing.T) {
-	// atomic, unlike the counters in the tests above: the second round's connection is closed
-	// without a response, so there is no happens-before edge between the handler's write and
-	// the assertion's read.
+	// The second round's connection is closed without a response, so not even a completed round
+	// trip stands between the handler's write and the assertion's read; the atomic carries the
+	// edge itself.
 	var calls atomic.Int64
 	head, tail := leadThenExpand("text")
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -342,12 +344,11 @@ func TestAFailedContinuationRoundStillEndsTheClientsTurn(t *testing.T) {
 // client's turn must still terminate: it gets the events the splice withheld, which is a
 // complete message, rather than a stream cut off mid-block.
 func TestAJSONContinuationRoundStillEndsTheClientsTurn(t *testing.T) {
-	var calls int
+	var calls atomic.Int64
 	head, tail := leadThenExpand("text")
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		io.Copy(io.Discard, r.Body)
-		calls++
-		if calls > 1 {
+		if calls.Add(1) > 1 {
 			w.Header().Set("Content-Type", "application/json")
 			w.Write([]byte(`{"type":"message","role":"assistant","content":[{"type":"text","text":"ANSWERED"}]}`))
 			return
@@ -370,8 +371,8 @@ func TestAJSONContinuationRoundStillEndsTheClientsTurn(t *testing.T) {
 	out, _ := io.ReadAll(resp.Body)
 	resp.Body.Close()
 
-	if calls != 2 {
-		t.Fatalf("expected the continuation round, got %d upstream calls", calls)
+	if n := calls.Load(); n != 2 {
+		t.Fatalf("expected the continuation round, got %d upstream calls", n)
 	}
 	if strings.Contains(string(out), `{"type":"message","role":"assistant"`) {
 		t.Fatalf("a JSON body was appended to an open event stream:\n%s", out)