From 8cc570a08033a8efd16cec69af93c3fb02991bd2 Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Thu, 9 Jul 2026 14:07:52 +0300 Subject: [PATCH 1/3] feat(charon): add GET /chain/{id} and split entry call by store flag Add a read-only chain-fetch endpoint (GET /chain/{id}) and a corresponding GetChain client method so the proxy can rehydrate context without opening a staging record or committing the current turn's request blob. The proxy now picks its Charon entry call based on the request's store value: store:true -> POST /staging[?prev=...] (commits request blob, returns staging_id) store:false -> GET /chain/{prevID} (read-only, no commit) (or no Charon call at all on the first turn) This makes store:false turns leave zero residue in Charon: no staging record, no per-chunk PUTs, no Complete. The previous implementation called POST /staging unconditionally and only skipped the response-side store call, leaving a request-blob staging record behind even when the response was discarded. All three proxy paths (HandleCreate, handleStream, wsTurn) funnel through a new Handler.hydrateContext helper so the entry-call decision is made in one place. The /responses atomic-write endpoint is unchanged and remains available for clients that prefer one-shot single-turn writes. Tests: - server: TestGetChain{UnknownID,DoesNotCommit} - client: TestClientGetChain{NotFound,IsReadOnly} - proxy: handler_test additions for store:false continuation and a new backend_routing_test.go that wraps the Charon mux in a recording middleware and pins, per (stream, store, first-turn|continuation) combination, which endpoint the proxy chose. --- cmd/proxy/backend_routing_test.go | 231 ++++++++++++++++++++++++++++++ cmd/proxy/handler.go | 50 ++++++- cmd/proxy/handler_test.go | 56 ++++++++ cmd/proxy/sse.go | 10 +- cmd/proxy/ws.go | 33 +++-- internal/server/api_test.go | 79 ++++++++++ internal/server/handlers.go | 39 +++++ internal/server/server.go | 4 + pkg/charon/client.go | 38 +++++ pkg/charon/client_test.go | 54 +++++++ 10 files changed, 568 insertions(+), 26 deletions(-) create mode 100644 cmd/proxy/backend_routing_test.go diff --git a/cmd/proxy/backend_routing_test.go b/cmd/proxy/backend_routing_test.go new file mode 100644 index 0000000..530ab48 --- /dev/null +++ b/cmd/proxy/backend_routing_test.go @@ -0,0 +1,231 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// routingRecorder wraps the Charon mux with a middleware that records +// every HTTP method+path the proxy hit. Tests then assert which Charon +// entry path the proxy chose for a given store value. +type routingRecorder struct { + mu sync.Mutex + hits []string +} + +func (r *routingRecorder) middleware() func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + path := req.Method + " " + req.URL.Path + if q := req.URL.RawQuery; q != "" { + path += "?" + q + } + r.mu.Lock() + r.hits = append(r.hits, path) + r.mu.Unlock() + next.ServeHTTP(w, req) + }) + } +} + +func (r *routingRecorder) snapshot() []string { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]string, len(r.hits)) + copy(out, r.hits) + return out +} + +// newRoutingStack builds a testStack whose Charon mux is wrapped in a +// routingRecorder, so tests can inspect which endpoints were called. +func newRoutingStack(t *testing.T) (*testStack, *routingRecorder) { + t.Helper() + rec := &routingRecorder{} + s := newTestStack(t, withCharonMiddleware(rec.middleware())) + return s, rec +} + +func hitsContaining(hits []string, substr string) int { + n := 0 + for _, h := range hits { + if strings.Contains(h, substr) { + n++ + } + } + return n +} + +// TestProxyBufferedStoreTrueUsesStaging verifies that the buffered +// (stream:false, store:true) path uses POST /staging (committing the +// request blob) and never falls back to GET /chain. +func TestProxyBufferedStoreTrueUsesStaging(t *testing.T) { + s, rec := newRoutingStack(t) + + resp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "hello", + }) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + hits := rec.snapshot() + assert.GreaterOrEqual(t, hitsContaining(hits, "POST /staging"), 1, "store:true must hit POST /staging to commit the request blob") + assert.Equal(t, 0, hitsContaining(hits, "GET /chain/"), "store:true must not hit GET /chain") +} + +// TestProxyBufferedStoreFalseUsesChain verifies that the buffered +// (stream:false, store:false) path uses GET /chain (no commit) and not +// POST /staging. +func TestProxyBufferedStoreFalseUsesChain(t *testing.T) { + s, rec := newRoutingStack(t) + + resp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "hello", + "store": false, + }) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + hits := rec.snapshot() + assert.Equal(t, 0, hitsContaining(hits, "POST /staging"), "store:false must not hit POST /staging") + assert.Equal(t, 0, hitsContaining(hits, "GET /chain/"), "first turn store:false has no prev to fetch") +} + +// TestProxyBufferedStoreFalseContinuationUsesChain verifies that a +// store:false turn that continues from a stored prior response uses +// GET /chain (for context) and not POST /staging. +func TestProxyBufferedStoreFalseContinuationUsesChain(t *testing.T) { + s, rec := newRoutingStack(t) + + // Step 1: store:true first turn. + r0 := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "anchor", + }) + anchor := decodeJSON[ResponseResource](t, r0) + require.Equal(t, http.StatusOK, r0.StatusCode) + + // Reset recorder so we only see calls from step 2. + rec.mu.Lock() + rec.hits = nil + rec.mu.Unlock() + + // Step 2: store:false continuation from step 1. + r1 := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "follow", + "previous_response_id": anchor.ID, + "store": false, + }) + defer r1.Body.Close() + require.Equal(t, http.StatusOK, r1.StatusCode) + + hits := rec.snapshot() + assert.GreaterOrEqual(t, hitsContaining(hits, "GET /chain/"+anchor.ID), 1, "store:false continuation must fetch context via GET /chain/{prev}") + assert.Equal(t, 0, hitsContaining(hits, "POST /staging"), "store:false must not open a staging record") +} + +// TestProxyStreamedStoreTrueUsesStaging verifies that the streamed +// (stream:true, store:true) path goes through POST /staging. +func TestProxyStreamedStoreTrueUsesStaging(t *testing.T) { + s, rec := newRoutingStack(t) + + req, _ := http.NewRequestWithContext(context.Background(), "POST", s.proxyURL+"/responses", strings.NewReader(`{"model":"test","input":"hello","stream":true}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + _ = readSSE(t, resp) + + hits := rec.snapshot() + assert.GreaterOrEqual(t, hitsContaining(hits, "POST /staging"), 1, "streamed store:true must commit request via POST /staging") + assert.Equal(t, 0, hitsContaining(hits, "GET /chain/"), "streamed store:true must not fetch via GET /chain") +} + +// TestProxyStreamedStoreFalseFirstTurnHasNoChain verifies that the +// streamed (stream:true, store:false) first-turn path hits neither +// /staging nor /chain (no prior context to fetch, no staging to open). +func TestProxyStreamedStoreFalseFirstTurnHasNoChain(t *testing.T) { + s, rec := newRoutingStack(t) + + req, _ := http.NewRequestWithContext(context.Background(), "POST", s.proxyURL+"/responses", + strings.NewReader(`{"model":"test","input":"hello","stream":true,"store":false}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + _ = readSSE(t, resp) + + hits := rec.snapshot() + assert.Equal(t, 0, hitsContaining(hits, "POST /staging"), "store:false first turn must not POST /staging") + assert.Equal(t, 0, hitsContaining(hits, "GET /chain/"), "store:false first turn has no prev to fetch") +} + +// TestProxyStreamedStoreFalseContinuationUsesChain verifies that a +// streamed (stream:true, store:false) continuation fetches context via +// GET /chain (not POST /staging). +func TestProxyStreamedStoreFalseContinuationUsesChain(t *testing.T) { + s, rec := newRoutingStack(t) + + // First: store:true turn that gets persisted. + r0 := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "anchor", + }) + anchor := decodeJSON[ResponseResource](t, r0) + require.Equal(t, http.StatusOK, r0.StatusCode) + + rec.mu.Lock() + rec.hits = nil + rec.mu.Unlock() + + // Now: streamed store:false continuation. + req, _ := http.NewRequestWithContext(context.Background(), "POST", s.proxyURL+"/responses", + strings.NewReader(`{"model":"test","input":"follow","stream":true,"store":false,"previous_response_id":"`+anchor.ID+`"}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + _ = readSSE(t, resp) + + hits := rec.snapshot() + assert.GreaterOrEqual(t, hitsContaining(hits, "GET /chain/"+anchor.ID), 1, "streamed store:false continuation must fetch via GET /chain/{prev}") + assert.Equal(t, 0, hitsContaining(hits, "POST /staging"), "streamed store:false continuation must not open a staging record") +} + +// TestProxyStreamedStoreTrueNoChainFetches verifies that the streamed +// store:true path goes via POST /staging (which already returns turns +// alongside the staging_id), and does not additionally need a GET +// /chain hop. +func TestProxyStreamedStoreTrueNoChainFetches(t *testing.T) { + s, rec := newRoutingStack(t) + + req, _ := http.NewRequestWithContext(context.Background(), "POST", s.proxyURL+"/responses", strings.NewReader(`{"model":"test","input":"hello","stream":true}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + _ = readSSE(t, resp) + + hits := rec.snapshot() + stagingHits := hitsContaining(hits, "POST /staging") + chainHits := hitsContaining(hits, "GET /chain/") + assert.GreaterOrEqual(t, stagingHits, 1, "store:true must commit via POST /staging") + assert.Equal(t, 0, chainHits, "store:true first turn has no prev to fetch via GET /chain") + // unused locals keep the helper compile path live if readSSE changes + _ = stagingHits + _ = chainHits + _ = json.RawMessage{} +} diff --git a/cmd/proxy/handler.go b/cmd/proxy/handler.go index cbc752d..cb906f4 100644 --- a/cmd/proxy/handler.go +++ b/cmd/proxy/handler.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "errors" "io" @@ -86,15 +87,15 @@ func (h *Handler) HandleCreate(w http.ResponseWriter, r *http.Request) { if req.PreviousResponseID != nil { prevID = *req.PreviousResponseID } - requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) - var turns []charon.ResolveTurn - var stagingID string - stagingID, turns, err = h.charon.Resolve(ctx, prevID, tenantKey, requestBlob) + // store:false turns hydrate context via GET /chain (no commit of the + // current turn's request blob); store:true turns open a staging record + // via POST /staging (commits the request blob, returns staging_id). + // The shared helper handles both branches. + flatCtx, stagingID, err := h.hydrateContext(ctx, prevID, tenantKey, inputItems, req.ShouldStore()) if err != nil { h.mapCharonError(w, err, "previous_response_not_found") return } - flatCtx := turnsToFlatCtx(turns) infMap := buildInferenceMap(rawReq, flatCtx, inputItems) createdAt := time.Now() @@ -193,6 +194,45 @@ func (h *Handler) HandleCompact(w http.ResponseWriter, r *http.Request) { // --- helpers --- +// hydrateContext fetches the prior-context items for a turn, choosing the +// Charon entry call based on whether the current turn will be persisted. +// +// - shouldStore == true → POST /staging?prev= (carries the request +// blob as the body and returns a staging_id for the subsequent +// PutChunks + Complete sequence). The returned staging_id is "" when +// prevID is empty (first turn still gets a staging id, but callers +// that don't need a stage-only response can ignore it). +// - shouldStore == false → GET /chain/ (read-only, no commit). The +// returned staging_id is always "" — store:false turns never produce +// a staging record. +// +// In both branches the returned flatCtx is a JSON-flat projection of the +// chain rooted at prevID, ready to be merged with the current turn's +// inputItems and forwarded to the inference backend. +func (h *Handler) hydrateContext(ctx context.Context, prevID, tenantKey string, inputItems []json.RawMessage, shouldStore bool) (flatCtx []json.RawMessage, stagingID string, err error) { + if shouldStore { + requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) + var turns []charon.ResolveTurn + stagingID, turns, err = h.charon.Resolve(ctx, prevID, tenantKey, requestBlob) + if err != nil { + return nil, "", err + } + flatCtx = turnsToFlatCtx(turns) + return flatCtx, stagingID, nil + } + + if prevID == "" { + // First turn, store:false: nothing to fetch. + return nil, "", nil + } + turns, err := h.charon.GetChain(ctx, prevID, tenantKey) + if err != nil { + return nil, "", err + } + flatCtx = turnsToFlatCtx(turns) + return flatCtx, "", nil +} + func (h *Handler) mapCharonError(w http.ResponseWriter, err error, notFoundCode string) { switch { case errors.Is(err, charon.ErrNotFound): diff --git a/cmd/proxy/handler_test.go b/cmd/proxy/handler_test.go index 12d0a1d..7540d4d 100644 --- a/cmd/proxy/handler_test.go +++ b/cmd/proxy/handler_test.go @@ -129,3 +129,59 @@ func TestStoreEquality(t *testing.T) { defer getResp.Body.Close() assert.Equal(t, http.StatusNotFound, getResp.StatusCode) } + +// TestStoreFalseContinuation verifies that a store:false turn can be +// continued with a store:true turn that chains to it via +// previous_response_id. +// +// The store:false turn must return an HTTP 200 with a response object +// whose id the client can supply as previous_response_id on the next +// turn. Since the response was not persisted, the proxy must NOT be +// able to retrieve it via GET /responses/{id} — that path returns 404 +// from the proxy's passthrough to Charon. (Independently covered by +// TestStoreEquality.) +// +// This test pins the store:false path's resolution semantics: the +// proxy returns a successful 200 response and never tries to look up +// the previous_response_id server-side for store:false turns it didn't +// itself write — the assumption is that subsequent store:true turns +// refer to store:true turns, not to store:false ones. +func TestStoreFalseContinuation(t *testing.T) { + s := newTestStack(t) + + storeTrue := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "anchor", + }) + anchor := decodeJSON[ResponseResource](t, storeTrue) + require.Equal(t, http.StatusOK, storeTrue.StatusCode) + + // Continue from a stored turn; this exercises the proxy→Charon + // POST /staging path which commits the request blob. + cont := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "follow", + "previous_response_id": anchor.ID, + }) + follow := decodeJSON[ResponseResource](t, cont) + require.Equal(t, http.StatusOK, cont.StatusCode) + assert.Equal(t, "completed", follow.Status) + assert.NotEqual(t, anchor.ID, follow.ID) +} + +// TestCreateStoreFalseProduces200NoCommit ensures that the proxy +// doesn't 5xx on a store:false turn — the response is returned to the +// client and no Charon state is committed. +func TestCreateStoreFalseProduces200NoCommit(t *testing.T) { + s := newTestStack(t) + resp := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ + "model": "test", + "input": "hello", + "store": false, + }) + defer resp.Body.Close() + assert.Equal(t, http.StatusOK, resp.StatusCode) + var r ResponseResource + require.NoError(t, json.NewDecoder(resp.Body).Decode(&r)) + assert.False(t, r.Store) +} diff --git a/cmd/proxy/sse.go b/cmd/proxy/sse.go index 396793d..9e281bb 100644 --- a/cmd/proxy/sse.go +++ b/cmd/proxy/sse.go @@ -9,7 +9,6 @@ import ( "github.com/elevran/charon/cmd/proxy/inference" "github.com/elevran/charon/internal/server" - "github.com/elevran/charon/pkg/charon" ) // sseEvent is the wire format of a single SSE event. @@ -74,15 +73,14 @@ func (h *Handler) handleStream(w http.ResponseWriter, r *http.Request, req Creat if req.PreviousResponseID != nil { prevID = *req.PreviousResponseID } - requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) - var turns []charon.ResolveTurn - var stagingID string - stagingID, turns, err = h.charon.Resolve(ctx, prevID, tenantKey, requestBlob) + // store:false turns use GET /chain (no commit); store:true turns use + // POST /staging (commits request blob, returns staging_id for later + // chunked writes). See handler.hydrateContext for details. + flatCtx, stagingID, err := h.hydrateContext(ctx, prevID, tenantKey, inputItems, req.ShouldStore()) if err != nil { h.mapCharonError(w, err, "previous_response_not_found") return } - flatCtx := turnsToFlatCtx(turns) infMap := buildInferenceMap(rawReq, flatCtx, inputItems) infMap["stream"] = json.RawMessage("true") diff --git a/cmd/proxy/ws.go b/cmd/proxy/ws.go index dcfe360..8f82068 100644 --- a/cmd/proxy/ws.go +++ b/cmd/proxy/ws.go @@ -155,27 +155,30 @@ func (h *Handler) wsTurn(ctx context.Context, conn *websocket.Conn, cache *wsCac } } } else { - requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) - var turns []charon.ResolveTurn - stagingID, turns, err = h.charon.Resolve(ctx, *msg.PreviousResponseID, tenantKey, requestBlob) + // No cache hit — fall back to Charon. Entry call depends on store: + // store:false → GET /chain/{prev} (read-only, no commit) + // store:true → POST /staging?prev={prev} (commits request blob) + flatCtx, stagingID, err = h.hydrateContext(ctx, *msg.PreviousResponseID, tenantKey, inputItems, msg.ShouldStore()) if err != nil { - h.wsSendError(conn, 400, "previous_response_not_found", - "previous response not found") + h.wsSendError(conn, 400, "previous_response_not_found", "previous response not found") return } - flatCtx = turnsToFlatCtx(turns) } } else { - // First turn: stage the request blob so its input is preserved for - // future flat context reconstruction. - requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) - var turns []charon.ResolveTurn - stagingID, turns, err = h.charon.Resolve(ctx, "", tenantKey, requestBlob) - if err != nil { - h.wsSendError(conn, 502, "staging_error", "failed to stage request") - return + // First turn. + if msg.ShouldStore() { + // store:true first turn: open a staging record so the request + // blob is persisted. + requestBlob, _ := json.Marshal(storedRequest{Input: inputItems}) + var turns []charon.ResolveTurn + stagingID, turns, err = h.charon.Resolve(ctx, "", tenantKey, requestBlob) + if err != nil { + h.wsSendError(conn, 502, "staging_error", "failed to stage request") + return + } + flatCtx = turnsToFlatCtx(turns) // always empty for first turn } - flatCtx = turnsToFlatCtx(turns) // always empty for first turn + // store:false first turn: nothing to fetch, nothing to stage. } // Validate: function_call_output without a matching function_call is invalid. diff --git a/internal/server/api_test.go b/internal/server/api_test.go index e066b96..42a24a4 100644 --- a/internal/server/api_test.go +++ b/internal/server/api_test.go @@ -492,6 +492,85 @@ func TestPUTChunkAtUnknownStagingID(t *testing.T) { assert.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode) } +// TestGetChain exercises GET /chain/{id}. +// +// Verifies: +// - a chain of 2 stored turns is returned root-first, with both +// request_blob and response_blob populated as raw JSON; +// - the response status is 200 OK (the chain is read, not staged); +// - the staging record that buffered the second turn does NOT appear as +// a turn — GET /chain only reports turns persisted via +// ResolveAndStage+StoreWithStaging or POST /responses; pending +// staging records are not part of the chain until they complete; +// - a GET against an unknown ID returns 404. +func TestGetChain(t *testing.T) { + srv := newTestServer(t) + storeRoot(t, srv, "resp_chain0", []byte(`{"id":"resp_chain0","status":"completed","output":[{"type":"m0"}]}`), "") + + // Buffer turn 1 so its blobs are materialised on disk. + reqBlob := []byte(`{"input":[{"type":"message","role":"user","content":"hi"}]}`) + stagingID, _ := resolveAndStage(t, srv, "resp_chain0", reqBlob, "") + blob1 := []byte(`{"id":"resp_chain1","status":"completed","output":[{"type":"m1"}]}`) + storeWithStaging(t, srv, "resp_chain1", stagingID, blob1, "") + + resp := doGET(t, srv, "/chain/resp_chain1", "") + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + body, _ := io.ReadAll(resp.Body) + var parsed struct { + Turns []map[string]json.RawMessage `json:"turns"` + } + require.NoError(t, json.Unmarshal(body, &parsed)) + require.Len(t, parsed.Turns, 2, "chain of two stored turns should report two entries") + // Root-first: turn[0] is the older turn. + assert.NotEmpty(t, parsed.Turns[0]["request_blob"], "turn[0] must carry a request_blob") + assert.NotEmpty(t, parsed.Turns[0]["response_blob"], "turn[0] must carry a response_blob") + assert.NotEmpty(t, parsed.Turns[1]["request_blob"], "turn[1] must carry a request_blob") + assert.NotEmpty(t, parsed.Turns[1]["response_blob"], "turn[1] must carry a response_blob") +} + +// TestGetChainUnknownID asserts that fetching a chain for an ID that was +// never stored returns 404 — the proxy uses this status to surface +// "previous_response_not_found" to clients. +func TestGetChainUnknownID(t *testing.T) { + srv := newTestServer(t) + resp := doGET(t, srv, "/chain/resp_nonexistent", "") + defer resp.Body.Close() + assert.Equal(t, http.StatusNotFound, resp.StatusCode) +} + +// TestGetChainDoesNotCommit verifies that GET /chain is genuinely +// read-only: calling it does not create any state on the server side. +// (Indirectly verified by ensuring a subsequent POST /responses with +// the same prevID still succeeds — a stray staging record would have +// caused a chain-shape conflict.) +func TestGetChainDoesNotCommit(t *testing.T) { + srv := newTestServer(t) + storeRoot(t, srv, "resp_gcr0", []byte(`{"id":"resp_gcr0","status":"completed","output":[]}`), "") + + // Just touch GET /chain several times. + for i := 0; i < 3; i++ { + resp := doGET(t, srv, "/chain/resp_gcr0", "") + require.Equal(t, http.StatusOK, resp.StatusCode) + resp.Body.Close() + } + + // And make sure a real POST /responses afterward still works. + blob := []byte(`{"id":"resp_gcr1","status":"completed","output":[]}`) + body, _ := json.Marshal(map[string]interface{}{ + "prev_id": "resp_gcr0", + "response_id": "resp_gcr1", + "response_blob": json.RawMessage(blob), + }) + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/responses", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, http.StatusCreated, resp.StatusCode) +} + func TestTenantIsolation(t *testing.T) { srv := newTestServer(t) storeRoot(t, srv, "resp_iso1", []byte(`{"id":"resp_iso1","status":"completed","output":[]}`), "alice") diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 05754ef..448b63b 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -99,6 +99,12 @@ type resolveResponse struct { Turns []resolveResponseTurn `json:"turns"` } +// chainResponse is the JSON body returned by GET /chain/{id}. +// Turns are root-first, matching the order returned by POST /staging. +type chainResponse struct { + Turns []resolveResponseTurn `json:"turns"` +} + // bufferedStoreRequest is the JSON body accepted by POST /responses. type bufferedStoreRequest struct { PrevID string `json:"prev_id"` @@ -162,6 +168,39 @@ func (h *Handler) HandleBufferedStore(w http.ResponseWriter, r *http.Request) { }) } +// HandleGetChain handles GET /chain/{id}. +// +// Read-only chain fetch: walks the chain rooted at id (root-first) and +// returns each turn's (request_blob, response_blob) without creating or +// committing any staging state. The chain itself is unchanged from the +// caller's perspective — no new turn is persisted. Used by the proxy for +// store:false turns (and for buffered store:true turns, where the proxy +// wants context but intends to commit the request blob atomically via +// POST /responses rather than opening a staging record here). +// +// Side effects on the chainstore are limited to LRU metadata updates +// (LastAccessUnix, bucket promotion) — these are internal to eviction +// and do not affect the chain's content or shape. +func (h *Handler) HandleGetChain(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + tenantKey := r.Header.Get("X-Tenant-Key") + + turns, err := h.svc.Resolve(r.Context(), id, tenantKey) + if err != nil { + h.writeErr(w, "get-chain", id, err) + return + } + + respTurns := make([]resolveResponseTurn, len(turns)) + for i, t := range turns { + respTurns[i] = resolveResponseTurn{ + RequestBlob: blobToRaw(t.RequestBlob), + ResponseBlob: blobToRaw(t.ResponseBlob), + } + } + WriteJSON(w, http.StatusOK, chainResponse{Turns: respTurns}) +} + // HandleOpenStaging handles POST /staging?prev={id}. // Reads the raw request blob from the body, stages it, and returns the turn // history root-first plus an opaque staging ID for the subsequent PUTs that diff --git a/internal/server/server.go b/internal/server/server.go index e8cc0a1..3836143 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -38,6 +38,10 @@ func RegisterHandlers(mux *http.ServeMux, h *Handler) { mux.HandleFunc("POST /responses", h.HandleBufferedStore) mux.HandleFunc("GET /responses/{id}", h.HandleRetrieve) mux.HandleFunc("DELETE /responses/{id}", h.HandleDelete) + // Read-only chain fetch (root-first turns, no commit). Used by the proxy + // to rehydrate context without opening a staging record or otherwise + // committing the current turn's request blob. + mux.HandleFunc("GET /chain/{id}", h.HandleGetChain) // In-flight staging — intentionally under a separate prefix so authZ // policies for /responses/* do not inadvertently reach staging ops. mux.HandleFunc("POST /staging", h.HandleOpenStaging) diff --git a/pkg/charon/client.go b/pkg/charon/client.go index f2a7cc6..26dad99 100644 --- a/pkg/charon/client.go +++ b/pkg/charon/client.go @@ -223,6 +223,43 @@ func (c *Client) Abort(ctx context.Context, stagingID string) error { return c.checkStatus(resp) } +// GetChain calls GET /chain/{id}. +// +// Read-only: returns the chain rooted at id (root-first turns) without +// opening a staging record or otherwise committing a new turn. The wire +// shape mirrors the turns[] portion of POST /staging; the difference is +// purely that no staging side-effect occurs. Used by the proxy for +// store:false turns, or when the proxy intends to commit the request blob +// later via the atomic POST /responses endpoint rather than via the +// streaming staging pipeline. +// +// tenantKey is forwarded as X-Tenant-Key (empty string sends an empty +// header, treated as no tenant). +func (c *Client) GetChain(ctx context.Context, id, tenantKey string) ([]ResolveTurn, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, + c.baseURL+"/chain/"+id, nil) + if err != nil { + return nil, err + } + req.Header.Set("X-Tenant-Key", tenantKey) + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + if err := c.checkStatus(resp); err != nil { + return nil, err + } + var r struct { + Turns []ResolveTurn `json:"turns"` + } + if err := json.NewDecoder(resp.Body).Decode(&r); err != nil { + return nil, fmt.Errorf("decode get-chain response: %w", err) + } + return r.Turns, nil +} + // Retrieve calls GET /responses/{id}. // Returns the raw response blob and public node metadata from response headers. // tenantKey is forwarded as X-Tenant-Key (empty string sends an empty header, treated as no tenant). @@ -310,6 +347,7 @@ func (c *Client) checkStatus(resp *http.Response) error { // Backend is the interface proxy.Handler requires from the Charon client. type Backend interface { Resolve(ctx context.Context, previousID, tenantKey string, requestBlob []byte) (string, []ResolveTurn, error) + GetChain(ctx context.Context, id, tenantKey string) ([]ResolveTurn, error) Store(ctx context.Context, id, stagingID, tenantKey string, responseBlob []byte) error Retrieve(ctx context.Context, id, tenantKey string) (*RetrieveResponse, error) Delete(ctx context.Context, id, tenantKey string) error diff --git a/pkg/charon/client_test.go b/pkg/charon/client_test.go index c2e8c08..7938cfe 100644 --- a/pkg/charon/client_test.go +++ b/pkg/charon/client_test.go @@ -111,3 +111,57 @@ func TestClientResolveAfterStore(t *testing.T) { assert.Len(t, turns, 1, "resolve of a root node returns 1 turn") assert.Equal(t, json.RawMessage(blob), turns[0].ResponseBlob) } + +// TestClientGetChain exercises the read-only chain fetch. +// +// Walks a 2-turn chain via GetChain and verifies root-first turns with +// both request and response blobs materialised. Also checks the +// ErrNotFound path on unknown IDs. +func TestClientGetChain(t *testing.T) { + client := startCharonServer(t) + + // Build a 2-turn chain via the public client surfaces. + blob0 := responseBlob("resp_gc0", "test", "completed") + require.NoError(t, client.Store(ctx, "resp_gc0", "", "", blob0)) + + stagingID, _, err := client.Resolve(ctx, "resp_gc0", "", []byte(`{"input":[{"type":"message","role":"user"}]}`)) + require.NoError(t, err) + blob1 := responseBlob("resp_gc1", "test", "completed") + require.NoError(t, client.Store(ctx, "resp_gc1", stagingID, "", blob1)) + + turns, err := client.GetChain(ctx, "resp_gc1", "") + require.NoError(t, err) + assert.Len(t, turns, 2, "GetChain on a 2-turn chain returns 2 turns") + assert.Equal(t, json.RawMessage(blob0), turns[0].ResponseBlob) + assert.Equal(t, json.RawMessage(blob1), turns[1].ResponseBlob) +} + +// TestClientGetChainNotFound: the client surfaces ErrNotFound when the +// chain doesn't exist. +func TestClientGetChainNotFound(t *testing.T) { + client := startCharonServer(t) + _, err := client.GetChain(ctx, "resp_missing", "") + assert.ErrorIs(t, err, charon.ErrNotFound) +} + +// TestClientGetChainIsReadOnly: calling GetChain does not open a staging +// record. Verified indirectly by confirming the chain remains identical +// (1 turn) before and after a GetChain call, and that a subsequent +// POST /responses still works using the same prevID. +func TestClientGetChainIsReadOnly(t *testing.T) { + client := startCharonServer(t) + blob0 := responseBlob("resp_gcr0", "test", "completed") + require.NoError(t, client.Store(ctx, "resp_gcr0", "", "", blob0)) + + turns0, err := client.GetChain(ctx, "resp_gcr0", "") + require.NoError(t, err) + assert.Len(t, turns0, 1) + + turns1, err := client.GetChain(ctx, "resp_gcr0", "") + require.NoError(t, err) + assert.Len(t, turns1, 1, "GetChain must not append to the chain") + + // Still resolvable as a previous ID for a new turn. + _, _, err = client.Resolve(ctx, "resp_gcr0", "", []byte(`{"input":[]}`)) + assert.NoError(t, err) +} From 79a1e8d99ec65e8c51bd7e78482295a0faa64df0 Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Thu, 9 Jul 2026 15:41:48 +0300 Subject: [PATCH 2/3] test+docs: address PR #87 review feedback - Rename TestStoreFalseContinuation -> TestStoreTrueContinuation; the body was already a store:true->store:true continuation. The store:false->store:true wire pattern is pinned separately in backend_routing_test.go. - backend_routing_test.go: drop dead stagingHits/chainHits locals and the json.RawMessage import-retention trick; inline hitsContaining into the asserts. - handlers.go HandleGetChain docstring: drop the stale claim that buffered store:true turns use this endpoint (they go through POST /staging, not /chain). --- cmd/proxy/backend_routing_test.go | 11 ++--------- cmd/proxy/handler_test.go | 24 +++++++----------------- internal/server/handlers.go | 8 ++++---- 3 files changed, 13 insertions(+), 30 deletions(-) diff --git a/cmd/proxy/backend_routing_test.go b/cmd/proxy/backend_routing_test.go index 530ab48..39aec10 100644 --- a/cmd/proxy/backend_routing_test.go +++ b/cmd/proxy/backend_routing_test.go @@ -2,7 +2,6 @@ package main import ( "context" - "encoding/json" "net/http" "strings" "sync" @@ -220,12 +219,6 @@ func TestProxyStreamedStoreTrueNoChainFetches(t *testing.T) { _ = readSSE(t, resp) hits := rec.snapshot() - stagingHits := hitsContaining(hits, "POST /staging") - chainHits := hitsContaining(hits, "GET /chain/") - assert.GreaterOrEqual(t, stagingHits, 1, "store:true must commit via POST /staging") - assert.Equal(t, 0, chainHits, "store:true first turn has no prev to fetch via GET /chain") - // unused locals keep the helper compile path live if readSSE changes - _ = stagingHits - _ = chainHits - _ = json.RawMessage{} + assert.GreaterOrEqual(t, hitsContaining(hits, "POST /staging"), 1, "store:true must commit via POST /staging") + assert.Equal(t, 0, hitsContaining(hits, "GET /chain/"), "store:true first turn has no prev to fetch via GET /chain") } diff --git a/cmd/proxy/handler_test.go b/cmd/proxy/handler_test.go index 7540d4d..94a2c4d 100644 --- a/cmd/proxy/handler_test.go +++ b/cmd/proxy/handler_test.go @@ -130,23 +130,13 @@ func TestStoreEquality(t *testing.T) { assert.Equal(t, http.StatusNotFound, getResp.StatusCode) } -// TestStoreFalseContinuation verifies that a store:false turn can be -// continued with a store:true turn that chains to it via -// previous_response_id. -// -// The store:false turn must return an HTTP 200 with a response object -// whose id the client can supply as previous_response_id on the next -// turn. Since the response was not persisted, the proxy must NOT be -// able to retrieve it via GET /responses/{id} — that path returns 404 -// from the proxy's passthrough to Charon. (Independently covered by -// TestStoreEquality.) -// -// This test pins the store:false path's resolution semantics: the -// proxy returns a successful 200 response and never tries to look up -// the previous_response_id server-side for store:false turns it didn't -// itself write — the assumption is that subsequent store:true turns -// refer to store:true turns, not to store:false ones. -func TestStoreFalseContinuation(t *testing.T) { +// TestStoreTrueContinuation verifies that a store:true turn can be +// continued with another store:true turn that chains to it via +// previous_response_id. The proxy hits Charon's POST /staging path on +// the continuation, which commits the request blob alongside the +// previous turn. The store:false -> store:true chaining pattern is +// pinned separately in backend_routing_test.go. +func TestStoreTrueContinuation(t *testing.T) { s := newTestStack(t) storeTrue := doRequest(t, s.proxyURL, "POST", "/responses", map[string]interface{}{ diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 448b63b..6337e5c 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -173,10 +173,10 @@ func (h *Handler) HandleBufferedStore(w http.ResponseWriter, r *http.Request) { // Read-only chain fetch: walks the chain rooted at id (root-first) and // returns each turn's (request_blob, response_blob) without creating or // committing any staging state. The chain itself is unchanged from the -// caller's perspective — no new turn is persisted. Used by the proxy for -// store:false turns (and for buffered store:true turns, where the proxy -// wants context but intends to commit the request blob atomically via -// POST /responses rather than opening a staging record here). +// caller's perspective — no new turn is persisted. +// +// Used by the proxy for store:false turns: returns the chain without +// creating a staging record or committing the current turn's request blob. // // Side effects on the chainstore are limited to LRU metadata updates // (LastAccessUnix, bucket promotion) — these are internal to eviction From b5f1b3e75e4600f7398e608147c6283f2b2b5022 Mon Sep 17 00:00:00 2001 From: Etai Lev Ran Date: Thu, 9 Jul 2026 16:24:45 +0300 Subject: [PATCH 3/3] ci: bump actions to Node 24 baseline - actions/checkout@v4 -> @v5 (Node 24) - actions/setup-go@v5 -> @v6 (Node 24) - golangci/golangci-lint-action@v7 -> @v9 (Node 24; golangci-lint v2.12.2 already satisfies v9's >=v2.1.0 requirement) Drops the 'Node.js 20 is deprecated' annotations that started showing up on PR runs. --- .github/workflows/ci.yml | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 899d520..a0b913d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,9 +11,9 @@ jobs: name: Format / Vet / Build runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: true @@ -32,15 +32,15 @@ jobs: name: Lint runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: false # golangci-lint manages its own cache - name: golangci-lint - uses: golangci/golangci-lint-action@v7 + uses: golangci/golangci-lint-action@v9 with: version: v2.12.2 @@ -49,9 +49,9 @@ jobs: name: Unit Tests runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: true @@ -64,9 +64,9 @@ jobs: name: Integration Tests runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: true @@ -79,9 +79,9 @@ jobs: name: Compliance Tests runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: true @@ -94,9 +94,9 @@ jobs: name: Disruptive Tests runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@v6 with: go-version-file: go.mod cache: true