From 7371ff85988acc4dded2313bfee1d5028508d5c2 Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Fri, 10 Jul 2026 11:42:41 +0300 Subject: [PATCH 1/3] =?UTF-8?q?test:=20end-to-end=20proxy=20test=20for=20c?= =?UTF-8?q?hunked=20save=20=E2=86=92=20chain=20serve?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes a real test gap surfaced after PR #88 merged: nothing proved that a response saved via the chunked staging protocol (POST /staging → PUT /staging/{sid}/chunks/{k} → PUT /staging/{sid}/complete) was correctly reassembled into a chain entry that the proxy could serve back as part of a subsequent turn's flat context. Existing coverage: - cmd/proxy/chunk_test.go: unit-level chunk→chunked byte round-trip via recordingBackend. - internal/chainstore/streaming_test.go: server-side chunk reassembly via AppendChunk/Complete/Retrieve. - cmd/proxy/backend_routing_test.go: wire-pattern pinning via routingRecorder (counts paths, doesn't inspect bytes). - cmd/proxy/handler_test.go::TestStoreTrueContinuation: continuation succeeds but doesn't verify the inference call received the prior turn's output. What this adds: - cmd/proxy/inference/mock.go: MockServer gains a RequestBodies() accessor that snapshots every POST /responses body in arrival order. Tests use this to inspect what the proxy actually forwarded to the inference backend. - cmd/proxy/chunked_roundtrip_test.go::TestChunkedRoundtripReassemblesIntoChain: saves an anchor turn with maxChunkBytes=64 (forces ≥2 AppendChunk), asserts the wire pattern (≥2 chunks, 1 complete, 0 abort), issues a continuation via previous_response_id, and verifies the continuation's inference call received the anchor's response output (msg_ok, role=assistant) in its input array — the strongest end-to-end proof that chunks → chain → served-context is intact. --- cmd/proxy/chunked_roundtrip_test.go | 106 ++++++++++++++++++++++++++++ cmd/proxy/inference/mock.go | 36 +++++++++- 2 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 cmd/proxy/chunked_roundtrip_test.go diff --git a/cmd/proxy/chunked_roundtrip_test.go b/cmd/proxy/chunked_roundtrip_test.go new file mode 100644 index 0000000..8e9e48e --- /dev/null +++ b/cmd/proxy/chunked_roundtrip_test.go @@ -0,0 +1,106 @@ +package main + +import ( + "encoding/json" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestChunkedRoundtripReassemblesIntoChain closes the test gap between the +// chunked-staging wire pattern (asserted by backend_routing_test.go) and the +// downstream chain-serve path: a response saved via PUT /staging/{sid}/chunks/{k} +// → PUT /staging/{sid}/complete must be reassembled by Charon into a single +// entry, and that entry must be served back as part of the chained context +// for any subsequent turn that references the anchor via previous_response_id. +// +// The proxy builds the chained context in turnsToFlatCtx and prepends it to +// the current turn's input before forwarding to the inference backend (see +// buildInferenceMap in assemble.go). So the strongest end-to-end assertion is +// that the inference call for the continuation sees the anchor's response +// output (msg_ok) somewhere in its input array — proving the chunks were +// reassembled into a chain entry that the proxy can serve. +// +// Pairs with: +// - cmd/proxy/chunk_test.go: unit-level chunk → chunked byte round-trip. +// - internal/chainstore/streaming_test.go: server-side chunk reassembly. +// - cmd/proxy/backend_routing_test.go: wire-pattern pinning (no byte check). +func TestChunkedRoundtripReassemblesIntoChain(t *testing.T) { + s, rec := newRoutingStack(t, withMaxChunkBytes(64)) + + // Step 1: anchor turn. With maxChunkBytes=64 the stored response blob + // (mock returns ~140-byte output) splits across multiple AppendChunk calls. + anchorResp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "anchor turn input", + }) + anchor := decodeJSON[ResponseResource](t, anchorResp) + require.Equal(t, http.StatusOK, anchorResp.StatusCode) + + // Step 2: wire-pattern check — confirm the anchor really was chunked, + // not just a single PUT that happened to fit. + hits := rec.snapshot() + assert.GreaterOrEqual(t, hitsContaining(hits, "/chunks/"), 2, + "tiny cap must force ≥2 AppendChunk calls for a non-trivial stored blob") + assert.Equal(t, 1, hitsContaining(hits, "/complete"), + "exactly 1 Complete call on success") + assert.Equal(t, 0, hitsContaining(hits, "/abort"), + "no abort on success") + + // Step 3: continuation. The proxy's HandleCreate calls hydrateContext, + // which fetches the chain rooted at anchor.ID and produces a flatCtx. + // buildInferenceMap prepends flatCtx to the new input items before the + // inference call. + contResp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "follow up input", + "previous_response_id": anchor.ID, + }) + follow := decodeJSON[ResponseResource](t, contResp) + require.Equal(t, http.StatusOK, contResp.StatusCode) + require.Equal(t, "completed", follow.Status) + + // Step 4: capture the inference backend's request bodies via the mock. + // Bodies arrive in order: anchor turn (call 1), continuation turn (call 2). + bodies := s.mockInf.RequestBodies() + require.GreaterOrEqual(t, len(bodies), 2, + "both anchor and continuation should have hit the inference backend") + + // Step 5: assert the continuation's inference call carried the anchor's + // output item in its input array. The MockServer returns a single item + // with id="msg_ok" and role="assistant" — distinguishing it from the + // user-role input items the proxy assembles around it. + var infReq struct { + Input json.RawMessage `json:"input"` + } + require.NoError(t, json.Unmarshal(bodies[1], &infReq)) + + var inputItems []map[string]interface{} + require.NoError(t, json.Unmarshal(infReq.Input, &inputItems), + "continuation input must be a JSON array (stringified for stateless inference)") + + // Locate the anchor's response output in the assembled input array. + found := false + for _, item := range inputItems { + if id, _ := item["id"].(string); id == "msg_ok" { + role, _ := item["role"].(string) + assert.Equal(t, "assistant", role, + "anchor's msg_ok must appear as an assistant item in the chained context") + found = true + break + } + } + assert.True(t, found, + "continuation's inference call must include the anchor's response output "+ + "(msg_ok) in its input — proves the chunked save was reassembled into "+ + "a chain entry the proxy can serve back") + + // Sanity check: both turn inputs should also be present, in the right order. + // Flat context order (per turnsToFlatCtx): turn input, turn output, turn input, turn output, ... + // Combined order (per buildInferenceMap): flatCtx, then new input items. + // So expected: [anchor_input, msg_ok(anchor), follow_input]. + assert.GreaterOrEqual(t, len(inputItems), 3, + "chained input must contain anchor input + anchor output + continuation input") +} diff --git a/cmd/proxy/inference/mock.go b/cmd/proxy/inference/mock.go index c1ee930..7d368d1 100644 --- a/cmd/proxy/inference/mock.go +++ b/cmd/proxy/inference/mock.go @@ -3,8 +3,10 @@ package inference import ( "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" + "sync" "sync/atomic" ) @@ -19,6 +21,9 @@ import ( type MockServer struct { *httptest.Server counter atomic.Int64 + + mu sync.Mutex + bodies [][]byte // captured request bodies, in arrival order } // NewMockServer starts a mock inference server. The caller must call Close(). @@ -33,6 +38,17 @@ func NewMockServer() *MockServer { // Calls returns the total number of inference requests handled. func (m *MockServer) Calls() int64 { return m.counter.Load() } +// RequestBodies returns a snapshot of every request body the mock has +// received, in arrival order. Tests use this to assert what the proxy +// actually forwarded to the inference backend. +func (m *MockServer) RequestBodies() [][]byte { + m.mu.Lock() + defer m.mu.Unlock() + out := make([][]byte, len(m.bodies)) + copy(out, m.bodies) + return out +} + // BaseURL returns the mock server's base URL (no trailing slash), satisfying // the inference.Backend interface. func (m *MockServer) BaseURL() string { return m.URL } @@ -43,11 +59,12 @@ func (m *MockServer) nextID() string { } func (m *MockServer) handle(w http.ResponseWriter, r *http.Request) { + body, _ := readAndRecord(r, m) + var req struct { Stream bool `json:"stream"` } - _ = json.NewDecoder(r.Body).Decode(&req) - _ = r.Body.Close() + _ = json.Unmarshal(body, &req) id := m.nextID() outputItem := json.RawMessage(`{"type":"message","id":"msg_ok","role":"assistant","status":"completed","content":[{"type":"output_text","text":"OK."}]}`) @@ -60,6 +77,21 @@ func (m *MockServer) handle(w http.ResponseWriter, r *http.Request) { m.writeComplete(w, id, outputItem, usage) } +// readAndRecord reads the request body, appends a copy to m.bodies under +// the mutex, and returns the bytes for further decoding. +func readAndRecord(r *http.Request, m *MockServer) ([]byte, error) { + body, err := io.ReadAll(r.Body) + if err != nil { + return nil, err + } + cp := make([]byte, len(body)) + copy(cp, body) + m.mu.Lock() + m.bodies = append(m.bodies, cp) + m.mu.Unlock() + return body, nil +} + func (m *MockServer) writeComplete(w http.ResponseWriter, id string, item json.RawMessage, usage *UsageInfo) { w.Header().Set("Content-Type", "application/json") resp := Response{ From ec7dc2de56fe4853d4370e1368e25c96adda4a9f Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Fri, 10 Jul 2026 15:07:08 +0300 Subject: [PATCH 2/3] test(inference): harden RequestBodies + recordBody against silent errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the 3 Important findings from the PR #90 review: 1. recordBody call-order dependency — added a comment in handle() above recordAndReplaceBody explaining that it MUST run before any JSON decode of r.Body, and what would break if a future change decodes first. 2. Silent io.ReadAll error → r.Body left in unknown state. Reworked readAndRecord into recordAndReplaceBody that always closes the original r.Body and re-installs io.NopCloser(bytes.NewReader(body)) with what was read. On a read error the body may be partial or empty — but the downstream decoder now sees a deterministic payload rather than re-draining an unknown state. 3. RequestBodies() shallow-copied slice headers. Now deep-copies each body via append([]byte(nil), b...) so callers may mutate the returned slices without affecting each other or future snapshots. Also renamed readAndRecord → recordAndReplaceBody to match the new contract (it now always replaces r.Body, not just records). --- cmd/proxy/inference/mock.go | 38 ++++++++++++++++++++++++------------- 1 file changed, 25 insertions(+), 13 deletions(-) diff --git a/cmd/proxy/inference/mock.go b/cmd/proxy/inference/mock.go index 7d368d1..4b72dbd 100644 --- a/cmd/proxy/inference/mock.go +++ b/cmd/proxy/inference/mock.go @@ -1,6 +1,7 @@ package inference import ( + "bytes" "encoding/json" "fmt" "io" @@ -39,13 +40,17 @@ func NewMockServer() *MockServer { func (m *MockServer) Calls() int64 { return m.counter.Load() } // RequestBodies returns a snapshot of every request body the mock has -// received, in arrival order. Tests use this to assert what the proxy -// actually forwarded to the inference backend. +// received, in arrival order. Each returned byte slice is a deep copy of +// the stored body, so callers may mutate the slices without affecting +// future snapshots or each other. Tests use this to assert what the +// proxy actually forwarded to the inference backend. func (m *MockServer) RequestBodies() [][]byte { m.mu.Lock() defer m.mu.Unlock() out := make([][]byte, len(m.bodies)) - copy(out, m.bodies) + for i, b := range m.bodies { + out[i] = append([]byte(nil), b...) + } return out } @@ -59,12 +64,17 @@ func (m *MockServer) nextID() string { } func (m *MockServer) handle(w http.ResponseWriter, r *http.Request) { - body, _ := readAndRecord(r, m) + // recordAndReplaceBody MUST run before any JSON decode of r.Body — it + // drains r.Body and re-installs a fresh reader so subsequent decoding + // observes the same bytes we captured. If a future change decodes + // r.Body directly before this call, stream=false would be observed + // for every request, regardless of what the client sent. + recordAndReplaceBody(r, m) var req struct { Stream bool `json:"stream"` } - _ = json.Unmarshal(body, &req) + _ = json.NewDecoder(r.Body).Decode(&req) id := m.nextID() outputItem := json.RawMessage(`{"type":"message","id":"msg_ok","role":"assistant","status":"completed","content":[{"type":"output_text","text":"OK."}]}`) @@ -77,19 +87,21 @@ func (m *MockServer) handle(w http.ResponseWriter, r *http.Request) { m.writeComplete(w, id, outputItem, usage) } -// readAndRecord reads the request body, appends a copy to m.bodies under -// the mutex, and returns the bytes for further decoding. -func readAndRecord(r *http.Request, m *MockServer) ([]byte, error) { - body, err := io.ReadAll(r.Body) - if err != nil { - return nil, err - } +// recordAndReplaceBody reads r.Body fully, stores a copy in m.bodies, and +// re-installs a fresh reader on r.Body so subsequent decoders in the handler +// see the same bytes. On read error the body may be partial or empty — +// either way we still replace r.Body so the handler decodes a deterministic +// (possibly empty) payload rather than re-draining an unknown state. +func recordAndReplaceBody(r *http.Request, m *MockServer) { + body, _ := io.ReadAll(r.Body) + _ = r.Body.Close() + r.Body = io.NopCloser(bytes.NewReader(body)) + cp := make([]byte, len(body)) copy(cp, body) m.mu.Lock() m.bodies = append(m.bodies, cp) m.mu.Unlock() - return body, nil } func (m *MockServer) writeComplete(w http.ResponseWriter, id string, item json.RawMessage, usage *UsageInfo) { From da6d138e9436684eb4e0e333fb8b1b412a0a1c60 Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Sun, 12 Jul 2026 14:10:10 +0300 Subject: [PATCH 3/3] test: simplify chunked-roundtrip test (drop wire-pattern duplication) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the 4 Minor findings from the PR #90 review: 4. Wire-pattern assertions (steps 1+2) duplicated TestBufferedProxyMultipleChunks. Dropped. The wire pattern is pinned by that existing test — this test focuses on the unique chain-reassembly property (anchor's response output appears in the continuation's inference input). 5. With the wire-pattern assertions gone, no need for routingRecorder. Switched from newRoutingStack to newTestStack — drops the recorder import and aligns with how backend_routing_test.go and other proxy tests set up the inference-side stack. 6. decodeJSON closes the body, so calling it before require.Equal on StatusCode swallows the status assertion on non-200 responses. Swapped the order in both anchor and continuation turns — status check now runs first, with a self-diagnosing message. 7. Loose []map[string]interface{} replaced with a local inputItem struct that decodes only the fields the assertions check. No equivalent type exists in inference/types.go (Response / SSEEvent / UsageInfo are server-side, not input items), so the local definition is justified; typed access also documents the shape under test. --- cmd/proxy/chunked_roundtrip_test.go | 94 ++++++++++++++--------------- 1 file changed, 45 insertions(+), 49 deletions(-) diff --git a/cmd/proxy/chunked_roundtrip_test.go b/cmd/proxy/chunked_roundtrip_test.go index 8e9e48e..e496ebe 100644 --- a/cmd/proxy/chunked_roundtrip_test.go +++ b/cmd/proxy/chunked_roundtrip_test.go @@ -9,10 +9,21 @@ import ( "github.com/stretchr/testify/require" ) +// inputItem is the subset of fields we assert on for an item in the chained +// context. The full shape is open-ended (user messages, assistant messages, +// tool calls, ...) so we only decode what we check. +type inputItem struct { + ID string `json:"id,omitempty"` + Role string `json:"role,omitempty"` + Type string `json:"type,omitempty"` + Text json.RawMessage `json:"content,omitempty"` // string for user, []item for assistant +} + // TestChunkedRoundtripReassemblesIntoChain closes the test gap between the -// chunked-staging wire pattern (asserted by backend_routing_test.go) and the -// downstream chain-serve path: a response saved via PUT /staging/{sid}/chunks/{k} -// → PUT /staging/{sid}/complete must be reassembled by Charon into a single +// chunked-staging wire pattern (asserted by backend_routing_test.go, in +// particular TestBufferedProxyMultipleChunks) and the downstream chain-serve +// path: a response saved via PUT /staging/{sid}/chunks/{k} → +// PUT /staging/{sid}/complete must be reassembled by Charon into a single // entry, and that entry must be served back as part of the chained context // for any subsequent turn that references the anchor via previous_response_id. // @@ -28,79 +39,64 @@ import ( // - internal/chainstore/streaming_test.go: server-side chunk reassembly. // - cmd/proxy/backend_routing_test.go: wire-pattern pinning (no byte check). func TestChunkedRoundtripReassemblesIntoChain(t *testing.T) { - s, rec := newRoutingStack(t, withMaxChunkBytes(64)) + // maxChunkBytes=64 forces ≥2 AppendChunk calls for a non-trivial stored + // blob. The wire pattern itself is pinned by TestBufferedProxyMultipleChunks + // — this test focuses on the chain-serve property. + s := newTestStack(t, withMaxChunkBytes(64)) - // Step 1: anchor turn. With maxChunkBytes=64 the stored response blob - // (mock returns ~140-byte output) splits across multiple AppendChunk calls. - anchorResp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + // Anchor turn — store:true, multi-chunk save under the hood. + anchorHTTP := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ "model": "test", "input": "anchor turn input", }) - anchor := decodeJSON[ResponseResource](t, anchorResp) - require.Equal(t, http.StatusOK, anchorResp.StatusCode) - - // Step 2: wire-pattern check — confirm the anchor really was chunked, - // not just a single PUT that happened to fit. - hits := rec.snapshot() - assert.GreaterOrEqual(t, hitsContaining(hits, "/chunks/"), 2, - "tiny cap must force ≥2 AppendChunk calls for a non-trivial stored blob") - assert.Equal(t, 1, hitsContaining(hits, "/complete"), - "exactly 1 Complete call on success") - assert.Equal(t, 0, hitsContaining(hits, "/abort"), - "no abort on success") + require.Equal(t, http.StatusOK, anchorHTTP.StatusCode, + "anchor turn must succeed before we can chain off it") + anchor := decodeJSON[ResponseResource](t, anchorHTTP) - // Step 3: continuation. The proxy's HandleCreate calls hydrateContext, - // which fetches the chain rooted at anchor.ID and produces a flatCtx. - // buildInferenceMap prepends flatCtx to the new input items before the - // inference call. - contResp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + // Continuation — references the anchor via previous_response_id. + contHTTP := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ "model": "test", "input": "follow up input", "previous_response_id": anchor.ID, }) - follow := decodeJSON[ResponseResource](t, contResp) - require.Equal(t, http.StatusOK, contResp.StatusCode) - require.Equal(t, "completed", follow.Status) + require.Equal(t, http.StatusOK, contHTTP.StatusCode, + "continuation must succeed") + follow := decodeJSON[ResponseResource](t, contHTTP) + assert.Equal(t, "completed", follow.Status) - // Step 4: capture the inference backend's request bodies via the mock. - // Bodies arrive in order: anchor turn (call 1), continuation turn (call 2). + // Capture the inference backend's request bodies in arrival order: + // call 1 = anchor, call 2 = continuation. bodies := s.mockInf.RequestBodies() require.GreaterOrEqual(t, len(bodies), 2, "both anchor and continuation should have hit the inference backend") - // Step 5: assert the continuation's inference call carried the anchor's - // output item in its input array. The MockServer returns a single item - // with id="msg_ok" and role="assistant" — distinguishing it from the - // user-role input items the proxy assembles around it. + // Decode the continuation's inference request and walk its input array. var infReq struct { Input json.RawMessage `json:"input"` } require.NoError(t, json.Unmarshal(bodies[1], &infReq)) - var inputItems []map[string]interface{} + var inputItems []inputItem require.NoError(t, json.Unmarshal(infReq.Input, &inputItems), "continuation input must be a JSON array (stringified for stateless inference)") - // Locate the anchor's response output in the assembled input array. - found := false - for _, item := range inputItems { - if id, _ := item["id"].(string); id == "msg_ok" { - role, _ := item["role"].(string) - assert.Equal(t, "assistant", role, - "anchor's msg_ok must appear as an assistant item in the chained context") - found = true + // Strongest assertion: the anchor's response output (msg_ok, role=assistant) + // appears in the chained context. Proves chunks → chain → served-context. + var foundAssistant *inputItem + for i := range inputItems { + if inputItems[i].ID == "msg_ok" { + foundAssistant = &inputItems[i] break } } - assert.True(t, found, + require.NotNil(t, foundAssistant, "continuation's inference call must include the anchor's response output "+ - "(msg_ok) in its input — proves the chunked save was reassembled into "+ - "a chain entry the proxy can serve back") + "(msg_ok) in its input — proves the chunked save was reassembled "+ + "into a chain entry the proxy can serve back") + assert.Equal(t, "assistant", foundAssistant.Role, + "anchor's msg_ok must appear as an assistant item in the chained context") - // Sanity check: both turn inputs should also be present, in the right order. - // Flat context order (per turnsToFlatCtx): turn input, turn output, turn input, turn output, ... - // Combined order (per buildInferenceMap): flatCtx, then new input items. - // So expected: [anchor_input, msg_ok(anchor), follow_input]. + // Sanity: at least three items — anchor input, anchor output, continuation input. assert.GreaterOrEqual(t, len(inputItems), 3, "chained input must contain anchor input + anchor output + continuation input") }