Skip to content

Commit 7f74748

Browse files
committed
feat(api): /v1/replay endpoint + trace_token on every retrieval response
handleQuery and handleAnswer now stamp a deterministic trace_token into the response body. The exact bytes sent on the wire are also stored in retrieval.ReplayStore under that token. POST /v1/replay with {trace_token, query, document_id} returns those bytes verbatim — same wire bytes, same Content-Type, same trailing newline. The byte-exactness chain: 1. The response map is marshalled once with json.Marshal. Go's encoding/json sorts map[string]any keys lexicographically, so the same Go value always produces the same []byte. 2. marshalJSONForReplay appends the same trailing newline that json.Encoder.Encode would, so the wire format is unchanged from the pre-3.1 behaviour and existing clients see no diff. 3. writeJSONWithReplay writes those bytes to the response AND hands them to the replay store in lock-step — a single []byte, two writes. 4. The replay handler returns store.Get(token).ResponseJSON verbatim. No re-marshalling, no normalisation. Replay validation: missing/empty fields → 400, unknown token → 404, mismatched document_id → 409 with details=document_id differs, mismatched query → 409 with details=query differs. The order matters: document_id is checked first because it's the highest-cardinality identifier and surfaces the most useful "you're pointing at the wrong document" signal. cmd/engine/main.go wires LRUReplayStore when retrieval.replay.enabled is true (the default). When disabled, Deps.Replay is nil — handlers skip the per-response Put and /v1/replay returns 501. OpenAPI adds: - trace_token field on QueryResponse + AnswerResponse - ReplayRequest schema (all three fields required) - /v1/replay path with 200/400/404/409/501 documented - 200's body is oneOf [QueryResponse, AnswerResponse] so the spec encodes the actual replayed shape config.example.yaml gets the retrieval.replay block with inline guidance (opt-out semantics, in-memory v1 caveats, forward pointer to Phase 3.2). The internal/api/replay_test.go suite covers byte-exact replay, unknown token, both mismatch flavours, disabled-store 501, required-fields 400, malformed JSON 400, unicode/whitespace preservation, and an end-to-end test that re-marshals the same map twice and asserts encoding/json is deterministic over the shape the engine actually emits.
1 parent 0c69f8e commit 7f74748

5 files changed

Lines changed: 657 additions & 3 deletions

File tree

cmd/engine/main.go

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,22 @@ func run() error {
151151
}
152152
}
153153

154+
// Replay store: Phase 3.1. On by default; operators opt out via
155+
// retrieval.replay.enabled=false (or VLE_RETRIEVAL_REPLAY_ENABLED=false).
156+
// In-memory only — Phase 3.2 will swap this for a durable store
157+
// behind the same retrieval.ReplayStore interface.
158+
var replayStore retrieval.ReplayStore
159+
if cfg.Retrieval.Replay.Enabled {
160+
replayStore = retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{
161+
MaxEntries: cfg.Retrieval.Replay.MaxEntries,
162+
TTL: time.Duration(cfg.Retrieval.Replay.TTLSeconds) * time.Second,
163+
})
164+
logger.Info("retrieval: replay store enabled",
165+
"max_entries", cfg.Retrieval.Replay.MaxEntries,
166+
"ttl_seconds", cfg.Retrieval.Replay.TTLSeconds,
167+
)
168+
}
169+
154170
pipeline := ingest.NewPipeline(ingest.Pipeline{
155171
DB: pool,
156172
Storage: store,
@@ -181,6 +197,7 @@ func run() error {
181197
Planning: cfg.Retrieval.Planning,
182198
ReRanker: reRanker,
183199
ReRank: cfg.Retrieval.ReRank,
200+
Replay: replayStore,
184201
}
185202

186203
srv := &http.Server{

config.example.yaml

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,28 @@ retrieval:
182182
# re-rank pass to do the final selection.
183183
top_k: 0
184184

185+
# replay: Phase 3.1 reproducibility store. Every /v1/query and
186+
# /v1/answer response carries a deterministic `trace_token`; the
187+
# response body is stored in an in-memory LRU under that token so
188+
# POST /v1/replay can return the byte-identical response on demand.
189+
#
190+
# OPT-OUT. Default enabled — replay is a moat versus stateless
191+
# vector RAG and should ship on by default. Disable to free the
192+
# memory budget when audit/replay isn't part of the operator's
193+
# flow. When disabled the response `trace_token` field is empty
194+
# and /v1/replay returns 501.
195+
#
196+
# The store is in-memory and not durable across process restarts.
197+
# Phase 3.2 will swap this for a persistent store + per-document
198+
# versioning behind the same interface.
199+
replay:
200+
enabled: true
201+
# LRU capacity. Older entries are evicted under memory pressure.
202+
max_entries: 1024
203+
# How long an entry remains valid. 86400 = 24 hours. Long
204+
# audit flows may bump this; tight memory budgets shrink it.
205+
ttl_seconds: 86400
206+
185207
ingest:
186208
# The summarize and HyDE stages run concurrently. This caps the total
187209
# number of LLM calls in flight across both stages combined, so the

internal/api/replay_test.go

Lines changed: 336 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,336 @@
1+
package api
2+
3+
import (
4+
"bytes"
5+
"encoding/json"
6+
"io"
7+
"net/http"
8+
"net/http/httptest"
9+
"strings"
10+
"testing"
11+
12+
"github.com/go-chi/chi/v5"
13+
14+
"github.com/hallelx2/vectorless-engine/pkg/retrieval"
15+
"github.com/hallelx2/vectorless-engine/pkg/tree"
16+
)
17+
18+
// newReplayRouter wires only the routes /v1/replay actually touches.
19+
// This avoids spinning up DB / Storage / Queue / Strategy just to
20+
// exercise the replay endpoint contract.
21+
func newReplayRouter(d Deps) http.Handler {
22+
r := chi.NewRouter()
23+
r.Route("/v1", func(r chi.Router) {
24+
r.Post("/replay", d.handleReplay)
25+
})
26+
return r
27+
}
28+
29+
// TestReplayByteExact: the central invariant of Phase 3.1.
30+
// Put a response into the store, replay it, assert the bytes
31+
// returned by the handler match what was stored — character for
32+
// character.
33+
func TestReplayByteExact(t *testing.T) {
34+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
35+
want := []byte(`{"answer":"hello","strategy":"chunked-tree","trace_token":"abc123"}` + "\n")
36+
store.Put("token-1", retrieval.ReplayEntry{
37+
DocumentID: "doc_x",
38+
Query: "what is x?",
39+
ResponseJSON: want,
40+
})
41+
42+
d := Deps{Replay: store}
43+
srv := httptest.NewServer(newReplayRouter(d))
44+
defer srv.Close()
45+
46+
body, _ := json.Marshal(map[string]any{
47+
"trace_token": "token-1",
48+
"query": "what is x?",
49+
"document_id": "doc_x",
50+
})
51+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
52+
if err != nil {
53+
t.Fatal(err)
54+
}
55+
defer resp.Body.Close()
56+
if resp.StatusCode != http.StatusOK {
57+
t.Fatalf("status = %d, want 200", resp.StatusCode)
58+
}
59+
got, _ := io.ReadAll(resp.Body)
60+
if !bytes.Equal(got, want) {
61+
t.Errorf("replay bytes differ:\n got %q\n want %q", got, want)
62+
}
63+
}
64+
65+
// TestReplayUnknownToken: 404 with a clear error message.
66+
func TestReplayUnknownToken(t *testing.T) {
67+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
68+
d := Deps{Replay: store}
69+
srv := httptest.NewServer(newReplayRouter(d))
70+
defer srv.Close()
71+
72+
body, _ := json.Marshal(map[string]any{
73+
"trace_token": "never-stored",
74+
"query": "q",
75+
"document_id": "doc_x",
76+
})
77+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
78+
if err != nil {
79+
t.Fatal(err)
80+
}
81+
defer resp.Body.Close()
82+
if resp.StatusCode != http.StatusNotFound {
83+
t.Errorf("status = %d, want 404", resp.StatusCode)
84+
}
85+
}
86+
87+
// TestReplayDocumentIDMismatch: 409 with details=document_id differs.
88+
func TestReplayDocumentIDMismatch(t *testing.T) {
89+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
90+
store.Put("t", retrieval.ReplayEntry{
91+
DocumentID: "doc_real",
92+
Query: "q",
93+
ResponseJSON: []byte(`{"x":1}` + "\n"),
94+
})
95+
d := Deps{Replay: store}
96+
srv := httptest.NewServer(newReplayRouter(d))
97+
defer srv.Close()
98+
99+
body, _ := json.Marshal(map[string]any{
100+
"trace_token": "t",
101+
"query": "q",
102+
"document_id": "doc_fake",
103+
})
104+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
105+
if err != nil {
106+
t.Fatal(err)
107+
}
108+
defer resp.Body.Close()
109+
if resp.StatusCode != http.StatusConflict {
110+
t.Fatalf("status = %d, want 409", resp.StatusCode)
111+
}
112+
var errBody map[string]string
113+
_ = json.NewDecoder(resp.Body).Decode(&errBody)
114+
if errBody["error"] != "input mismatch" {
115+
t.Errorf("error = %q, want input mismatch", errBody["error"])
116+
}
117+
if !strings.Contains(errBody["details"], "document_id") {
118+
t.Errorf("details should mention document_id, got %q", errBody["details"])
119+
}
120+
}
121+
122+
// TestReplayQueryMismatch: 409 with details=query differs.
123+
func TestReplayQueryMismatch(t *testing.T) {
124+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
125+
store.Put("t", retrieval.ReplayEntry{
126+
DocumentID: "doc_x",
127+
Query: "real query",
128+
ResponseJSON: []byte(`{"x":1}` + "\n"),
129+
})
130+
d := Deps{Replay: store}
131+
srv := httptest.NewServer(newReplayRouter(d))
132+
defer srv.Close()
133+
134+
body, _ := json.Marshal(map[string]any{
135+
"trace_token": "t",
136+
"query": "tampered query",
137+
"document_id": "doc_x",
138+
})
139+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
140+
if err != nil {
141+
t.Fatal(err)
142+
}
143+
defer resp.Body.Close()
144+
if resp.StatusCode != http.StatusConflict {
145+
t.Fatalf("status = %d, want 409", resp.StatusCode)
146+
}
147+
var errBody map[string]string
148+
_ = json.NewDecoder(resp.Body).Decode(&errBody)
149+
if !strings.Contains(errBody["details"], "query") {
150+
t.Errorf("details should mention query, got %q", errBody["details"])
151+
}
152+
}
153+
154+
// TestReplayDisabled: when Deps.Replay is nil the endpoint returns
155+
// 501 Not Implemented. This is the opt-out path documented in the
156+
// config block.
157+
func TestReplayDisabled(t *testing.T) {
158+
d := Deps{Replay: nil}
159+
srv := httptest.NewServer(newReplayRouter(d))
160+
defer srv.Close()
161+
162+
body, _ := json.Marshal(map[string]any{
163+
"trace_token": "anything",
164+
"query": "q",
165+
"document_id": "doc_x",
166+
})
167+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
168+
if err != nil {
169+
t.Fatal(err)
170+
}
171+
defer resp.Body.Close()
172+
if resp.StatusCode != http.StatusNotImplemented {
173+
t.Errorf("status = %d, want 501", resp.StatusCode)
174+
}
175+
}
176+
177+
// TestReplayRequiresFields: every field in the body is required.
178+
// A missing field is a client error, not a 404 (which would
179+
// otherwise be confusing — "the token isn't found" when really
180+
// "you didn't send a token").
181+
func TestReplayRequiresFields(t *testing.T) {
182+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
183+
d := Deps{Replay: store}
184+
srv := httptest.NewServer(newReplayRouter(d))
185+
defer srv.Close()
186+
187+
cases := []map[string]any{
188+
{"query": "q", "document_id": "doc_x"}, // missing trace_token
189+
{"trace_token": "t", "document_id": "doc_x"}, // missing query
190+
{"trace_token": "t", "query": "q"}, // missing document_id
191+
{"trace_token": "", "query": "q", "document_id": "doc"}, // empty trace_token
192+
}
193+
for i, body := range cases {
194+
raw, _ := json.Marshal(body)
195+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(raw))
196+
if err != nil {
197+
t.Fatalf("case %d: %v", i, err)
198+
}
199+
resp.Body.Close()
200+
if resp.StatusCode != http.StatusBadRequest {
201+
t.Errorf("case %d: status = %d, want 400", i, resp.StatusCode)
202+
}
203+
}
204+
}
205+
206+
// TestReplayBadJSON: malformed JSON request body → 400.
207+
func TestReplayBadJSON(t *testing.T) {
208+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
209+
d := Deps{Replay: store}
210+
srv := httptest.NewServer(newReplayRouter(d))
211+
defer srv.Close()
212+
213+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", strings.NewReader("{not json"))
214+
if err != nil {
215+
t.Fatal(err)
216+
}
217+
defer resp.Body.Close()
218+
if resp.StatusCode != http.StatusBadRequest {
219+
t.Errorf("status = %d, want 400", resp.StatusCode)
220+
}
221+
}
222+
223+
// TestReplayEndToEndByteExact simulates the production flow: the
224+
// server marshals a /v1/query response via marshalJSONForReplay,
225+
// stores it under the same trace token it surfaced to the client,
226+
// and the replay endpoint hands the bytes back verbatim. This is
227+
// the end-to-end byte-exactness invariant the Phase 3.1 spec
228+
// demands.
229+
//
230+
// The test uses Go's encoding/json directly (the same package the
231+
// handler uses) so any drift between "serialised on write" and
232+
// "served on replay" surfaces here.
233+
func TestReplayEndToEndByteExact(t *testing.T) {
234+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
235+
d := Deps{Replay: store}
236+
237+
// Build a representative response that exercises Go map JSON
238+
// emission (lexicographic key sort) and a varied payload shape.
239+
traceToken := retrieval.ComputeTraceToken("doc_x", "1", "claude-sonnet-4-5",
240+
[]tree.SectionID{"sec_a", "sec_b"})
241+
resp := map[string]any{
242+
"document_id": "doc_x",
243+
"query": "what does the report say?",
244+
"strategy": "chunked-tree",
245+
"model": "claude-sonnet-4-5",
246+
"sections": []map[string]any{
247+
{"id": "sec_a", "title": "Setup"},
248+
{"id": "sec_b", "title": "Usage"},
249+
},
250+
"elapsed_ms": 42,
251+
"trace_token": traceToken,
252+
}
253+
254+
raw, err := marshalJSONForReplay(resp)
255+
if err != nil {
256+
t.Fatalf("marshal: %v", err)
257+
}
258+
259+
// Simulate the handler writing the response to wire AND storing.
260+
d.Replay.Put(traceToken, retrieval.ReplayEntry{
261+
DocumentID: "doc_x",
262+
Query: "what does the report say?",
263+
Model: "claude-sonnet-4-5",
264+
ResponseJSON: raw,
265+
})
266+
267+
// Re-marshal the same Go value: encoding/json sorts map keys
268+
// lexicographically, so the bytes must be identical. This is
269+
// the property that makes byte-exact replay viable even when
270+
// the response is built from a map[string]any rather than a
271+
// struct with a fixed field order.
272+
raw2, err := marshalJSONForReplay(resp)
273+
if err != nil {
274+
t.Fatalf("remarshal: %v", err)
275+
}
276+
if !bytes.Equal(raw, raw2) {
277+
t.Errorf("encoding/json is non-deterministic on map[string]any:\n first %q\n second %q", raw, raw2)
278+
}
279+
280+
// Replay over HTTP. Bytes must equal what we stored.
281+
srv := httptest.NewServer(newReplayRouter(d))
282+
defer srv.Close()
283+
body, _ := json.Marshal(map[string]any{
284+
"trace_token": traceToken,
285+
"query": "what does the report say?",
286+
"document_id": "doc_x",
287+
})
288+
got, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
289+
if err != nil {
290+
t.Fatal(err)
291+
}
292+
defer got.Body.Close()
293+
if got.StatusCode != http.StatusOK {
294+
t.Fatalf("status = %d", got.StatusCode)
295+
}
296+
gotBytes, _ := io.ReadAll(got.Body)
297+
if !bytes.Equal(gotBytes, raw) {
298+
t.Errorf("end-to-end byte drift:\n stored %q\n got %q", raw, gotBytes)
299+
}
300+
}
301+
302+
// TestReplayPreservesUnicodeAndWhitespace replays a payload chosen
303+
// to expose any normalisation in the storage path: unicode, mixed
304+
// whitespace, embedded newlines. The byte sequence must come back
305+
// identical.
306+
func TestReplayPreservesUnicodeAndWhitespace(t *testing.T) {
307+
store := retrieval.NewLRUReplayStore(retrieval.LRUReplayConfig{MaxEntries: 10})
308+
// Hand-crafted bytes — deliberately not pretty-printed JSON, and
309+
// includes content that round-tripping through encoding/json
310+
// would re-escape.
311+
want := []byte("{\"text\":\"héllo\\nworld 🌍\",\"k\": 42}\n")
312+
store.Put("u", retrieval.ReplayEntry{
313+
DocumentID: tree.DocumentID("doc_u"),
314+
Query: "q",
315+
ResponseJSON: want,
316+
})
317+
318+
d := Deps{Replay: store}
319+
srv := httptest.NewServer(newReplayRouter(d))
320+
defer srv.Close()
321+
322+
body, _ := json.Marshal(map[string]any{
323+
"trace_token": "u",
324+
"query": "q",
325+
"document_id": "doc_u",
326+
})
327+
resp, err := http.Post(srv.URL+"/v1/replay", "application/json", bytes.NewReader(body))
328+
if err != nil {
329+
t.Fatal(err)
330+
}
331+
defer resp.Body.Close()
332+
got, _ := io.ReadAll(resp.Body)
333+
if !bytes.Equal(got, want) {
334+
t.Errorf("byte drift:\n got %q\n want %q", got, want)
335+
}
336+
}

0 commit comments

Comments
 (0)