fix(expand): stop buffering every SSE response on a marker check that always matched

docs/design.md

@@ -156,6 +156,30 @@ An expired/evicted original resolves to an explicit placeholder rather than bein
 provider requires one `tool_result` per `tool_call_id`). A miss silently turns a lossless offload
 lossy — the known TTL edge.
 
+### The loop on a streaming response
+
+The loop needs a whole assistant message; SSE delivers events. So the host decides per request,
+from the request bytes, whether it can afford to look:
+
+- **no marker in `messages`/`system`** → nothing to expand → stream through untouched;
+- **marker present** → buffer the stream, rebuild the message with `expand.AggregateSSE`
+  (Anthropic dialect only — other dialects return `ok=false` and are replayed raw), inspect, and
+  either continue the loop or replay the buffered bytes verbatim.
+
+Buffering is the one thing that turns a stream into a non-stream, so the marker test must be tight
+in *both* directions: a false negative loses a real expand call, a false positive silently costs
+every request its time-to-first-byte. It scans only model-visible content (`messages`, `system`) —
+scanning the whole body also matched the expand tool description the host injects itself, which made
+it unconditionally true (issue #26). `/stats` exposes `sse_streamed` / `sse_buffered` /
+`sse_buffered_pct` and the two TTFB averages so the fast path is measured, not assumed.
+
+**Markers on the wire are usually HTML-escaped.** `encoding/json` escapes `<` by default — a caller
+can opt out with `Encoder.SetEscapeHTML(false)`, and some non-Go clients never escape it — and `sjson`
+escapes it whenever the value contains a newline, which is how every marker is appended. So `<<cg:H>>`
+in the model's view is normally `<<cg:H>>` in the bytes. Marker matching on *decoded* content
+(`expand.HasPlaceholder`, used by the components) sees the plain form; matching on *raw request
+bytes* (`expand.rawMarkerRe`, used by the host's streaming decision) must accept both, and does.
+
 ## State: the Store
 
 One `Store` interface, in-memory TTL+LRU default (both hosts share it). Defaults: **1800s TTL,
@@ -184,7 +208,13 @@ vocabulary), `Aggregator` (in-process rollups behind `/stats`), `Tee` (fan-out),
 of per-request percentages. It also reports:
 - `wasted_tokens` / `bounces` — content offloaded then re-served via expand (a premature offload);
 - `adjusted_saved` = saved − wasted (bounce-adjusted, may be negative);
-- `top_passthrough` — components that ran but never changed a request: dead weight to drop.
+- `top_passthrough` — components that ran but never changed a request: dead weight to drop;
+- `sse_streamed` / `sse_buffered` / `sse_buffered_pct` and `sse_ttfb_ms_avg` /
+  `sse_ttfb_ms_avg_buffered` — streaming health: how many SSE responses had to be buffered whole to
+  be inspected for an expand call, and what that cost in time-to-first-byte.
+
+Fields are only ever **added** to `/stats`; the harbor harnesses parse it, so no field is renamed
+or removed.
 
 ## Config & registry
 

docs/how-to/recover-context.md

@@ -48,6 +48,51 @@ original resolves to an explicit placeholder rather than being omitted — the p
 `tool_result` per `tool_call_id`. A store miss silently turns a lossless offload lossy: the known
 TTL edge.
 
+### Streaming (SSE): what actually happens
+
+The loop reasons over a complete assistant message, but a streaming response arrives as events, so
+on SSE the proxy makes a per-request choice from the request bytes:
+
+- **No marker in `messages`/`system`** → the model has nothing to expand, so the response is
+  streamed straight through, byte for byte, with no added latency.
+- **A marker is present** → the response is read in full, reconstructed with `expand.AggregateSSE`,
+  and inspected. If it is a lone expand call the loop runs; otherwise the buffered bytes are
+  replayed to the client verbatim.
+
+Buffering costs the client its streaming for that request (time-to-first-byte becomes
+time-to-last-byte), which is why the marker test is narrow. It scans **only** `messages` and
+`system` — the model-visible content. It used to scan the whole request body, which also matched the
+`context_guru_expand` tool description we inject ourselves ("…replaced by a `<<cg:HASH>>` marker"),
+so it was always true and **every** stream was silently buffered (issue #26).
+
+`/stats` reports this directly, counted **once per client request** (not per upstream round, so a
+request that drove several expand rounds is one sample): `sse_streamed`, `sse_buffered`,
+`sse_buffered_pct`, `sse_ttfb_ms_avg` and `sse_ttfb_ms_avg_buffered`. On traffic that never offloads,
+`sse_buffered` stays 0; it starts counting from the first turn that carries a marker. Note that
+`sse_ttfb_ms_avg_buffered` is time-to-*last*-byte by construction — a buffered response is read in
+full before the client is written to — so it is not comparable to `sse_ttfb_ms_avg`.
+
+!!! note "Markers usually arrive HTML-escaped"
+    A marker the model reads as `<<cg:HASH>>` normally travels on the wire as
+    `<<cg:HASH>>`: Go's `encoding/json` escapes `<` by default (callers can opt out
+    with `Encoder.SetEscapeHTML(false)`, and some non-Go clients never escape it), and `sjson`
+    escapes it whenever the value contains a newline — which is how every marker is appended. Any
+    check matching markers against raw request bytes must accept both spellings, case-insensitively
+    (`<` is as valid as `<`); `expand.rawMarkerRe` does, deliberately. A miss there is a
+    false negative — a real expand call streamed past uninspected — which is worse than
+    over-buffering.
+
+    Because that matcher accepts the plain form too, a document or message quoting a literal
+    `<<cg:HASH>>` example (like this page) counts as marker-bearing. That is the intended bias:
+    over-inspect rather than miss a real call.
+
+!!! warning "Streaming restoration is Anthropic-only"
+    `expand.AggregateSSE` reconstructs the Anthropic Messages event stream only. A marker-bearing
+    **OpenAI** streaming response cannot be reconstructed, so it is replayed raw (fail-open) and
+    restoration does not fire on that request. Non-streaming OpenAI restoration works normally, as
+    does streaming Anthropic. Every streaming coding agent in scope speaks the Anthropic dialect;
+    OpenAI SSE aggregation will be added when a real agent needs it.
+
 ### 3. `GET /expand?id=`
 The proxy exposes `GET /expand?id=<hash>` to recover an offloaded original directly, out of band
 from the model loop.

expand/expand.go

@@ -15,6 +15,7 @@ import (
 	"strings"
 
 	"github.com/rossoctl/context-guru/store"
+	"github.com/tidwall/gjson"
 )
 
 // ToolName is the model-callable tool that retrieves offloaded content.
@@ -46,6 +47,44 @@ func HasPlaceholder(s string) bool {
 	return strings.Contains(s, SummaryMarker) || markerRe.MatchString(s)
 }
 
+// rawMarkerRe matches a full <<cg:HASH>> marker in RAW, un-decoded request bytes,
+// where the angle brackets may arrive HTML-escaped as < / >.
+//
+// BOTH spellings are load-bearing, and the escaped one is the COMMON case, not an
+// exotic client quirk: Go's encoding/json HTML-escapes "<" by default (unless a caller
+// opts out via Encoder.SetEscapeHTML(false)), and sjson escapes it whenever the value
+// being set contains a newline — and every Offload marker is appended after a newline.
+// So a marker the MODEL reads as <<cg:HASH>> usually exists in the bytes on the wire
+// only as <<cg:HASH>>. Any check matching markers against a raw body must
+// accept both forms deliberately.
+//
+// The escape alternatives are case-insensitive: \u003C is as valid as \u003c, and a miss
+// here is a FALSE NEGATIVE — a real expand call streamed past uninspected, which is
+// worse than the over-buffering this regexp exists to prevent.
+var rawMarkerRe = regexp.MustCompile(`(?:<|(?i:\\u003c)){2}cg:([A-Za-z0-9_-]{1,64})(?:>|(?i:\\u003e)){2}`)
+
+// HasMarkersInMessages reports whether a request body carries a context-guru
+// placeholder in content the MODEL can see and reference — the messages array and
+// the system prompt.
+//
+// It deliberately does NOT scan the whole body. The `tools` array holds our own
+// injected expand tool, whose description quotes the marker syntax ("…replaced by a
+// <<cg:HASH>> marker"), HTML-escaped by ToolDefRaw's encoding/json. A whole-body check
+// therefore matched the tool we had just injected and was a tautology: every request
+// looked marker-bearing, so every streaming response was fully buffered and the
+// documented zero-added-latency fast path never engaged. Requiring the full marker
+// shape is not sufficient on its own — the tool description contains the full shape
+// too. Scoping to model-visible content is what fixes it.
+func HasMarkersInMessages(body []byte) bool {
+	for _, field := range [...]string{"messages", "system"} {
+		if r := gjson.GetBytes(body, field); r.Exists() &&
+			(rawMarkerRe.MatchString(r.Raw) || strings.Contains(r.Raw, SummaryMarker)) {
+			return true
+		}
+	}
+	return false
+}
+
 // ParseMarkers returns the distinct store keys referenced by any markers in s,
 // in first-seen order.
 func ParseMarkers(s string) []string {

expand/expand_test.go

@@ -1,9 +1,12 @@
 package expand
 
 import (
+	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/rossoctl/context-guru/store"
+	"github.com/tidwall/sjson"
 )
 
 func TestToolDefShape(t *testing.T) {
@@ -34,6 +37,98 @@ func TestParseMarkersDistinctInOrder(t *testing.T) {
 	}
 }
 
+// escLT / escGT are the JSON \uXXXX escapes Go's encoders emit for "<" and ">",
+// obtained by ASKING encoding/json rather than hand-writing them, so a fixture can
+// never claim an escape form the encoder does not actually produce.
+var escLT, escGT = jsonEscape('<'), jsonEscape('>')
+
+// jsonEscape marshals a single rune with HTML escaping on and returns the escape
+// sequence encoding/json chose for it (e.g. "<" for '<').
+func jsonEscape(r rune) string {
+	b, err := json.Marshal(string(r))
+	if err != nil {
+		panic(err)
+	}
+	s := strings.Trim(string(b), `"`)
+	if !strings.HasPrefix(s, `\u`) {
+		panic("encoding/json no longer escapes " + string(r) + " (got " + s + "); " +
+			"the raw-body marker matcher's escaped-form handling needs revisiting")
+	}
+	return s
+}
+
+// upperHex uppercases the hex digits of a \uXXXX escape but not the "u" ("<"
+// -> "<"), matching what a non-Go JSON encoder may legitimately emit.
+func upperHex(esc string) string {
+	return `\u` + strings.ToUpper(strings.TrimPrefix(esc, `\u`))
+}
+
+// TestHasMarkersInMessagesIgnoresOwnInjectedTool is the regression guard for the
+// tautology of issue #26: the injected expand tool's own description quotes the
+// marker syntax, so a whole-body marker check was always true and every streaming
+// response got buffered. The check must look only at model-visible content.
+func TestHasMarkersInMessagesIgnoresOwnInjectedTool(t *testing.T) {
+	for _, provider := range []string{"anthropic", "openai"} {
+		body := []byte(`{"messages":[{"role":"user","content":"hello"}],"tools":[]}`)
+		injected, ok := Inject(provider, InjectAlways, body, true)
+		if !ok {
+			t.Fatalf("%s: Inject should have fired", provider)
+		}
+		// The escaped marker shape IS in the injected bytes — that is exactly what used
+		// to make the old whole-body check unconditionally true.
+		if !strings.Contains(string(injected), escLT+escLT+"cg:") {
+			t.Fatalf("%s: expected the tool description to carry an escaped marker: %s", provider, injected)
+		}
+		if HasMarkersInMessages(injected) {
+			t.Fatalf("%s: our own injected tool must not count as a marker: %s", provider, injected)
+		}
+	}
+}
+
+// TestHasMarkersInMessagesEscapedForm pins the load-bearing accident: markers reach
+// the wire HTML-escaped whenever the value they were appended to contains a newline
+// (sjson/encoding/json escape "<"). Both spellings must be found.
+func TestHasMarkersInMessagesEscapedForm(t *testing.T) {
+	escMarker := escLT + escLT + "cg:ABC" + escGT + escGT
+	cases := map[string]string{
+		"plain":                   `{"messages":[{"role":"user","content":"see <<cg:ABC>>"}]}`,
+		"escaped":                 `{"messages":[{"role":"user","content":"see ` + escMarker + `"}]}`,
+		"escaped after a newline": `{"messages":[{"role":"user","content":"line1\nline2 ` + escMarker + `"}]}`,
+		"summary sentinel":        `{"messages":[{"role":"user","content":"` + SummaryMarker + ` compacted"}]}`,
+		"in the system prompt":    `{"system":"context: ` + escMarker + `","messages":[]}`,
+		"escaped in tool_result":  `{"messages":[{"role":"user","content":[{"type":"tool_result","content":"out\n` + escMarker + `"}]}]}`,
+		// < is as valid an escape as <. Missing it would be a false negative:
+		// a real expand call streamed past uninspected.
+		"uppercase hex escape": `{"messages":[{"role":"user","content":"see ` +
+			upperHex(escLT) + upperHex(escLT) + `cg:ABC` +
+			upperHex(escGT) + upperHex(escGT) + `"}]}`,
+	}
+	for name, body := range cases {
+		if !HasMarkersInMessages([]byte(body)) {
+			t.Errorf("%s: marker not found in %s", name, body)
+		}
+	}
+	// Real sjson output, not a hand-written string: prove the escaping happens exactly
+	// as offload triggers it (marker appended after a newline).
+	sjsonBody, _ := sjson.SetBytes([]byte(`{"messages":[{"role":"user","content":""}]}`),
+		"messages.0.content", "line1\nline2 "+Marker("XYZ"))
+	if !strings.Contains(string(sjsonBody), escLT+escLT+"cg:XYZ") {
+		t.Fatalf("expected sjson to escape the marker after a newline: %s", sjsonBody)
+	}
+	if !HasMarkersInMessages(sjsonBody) {
+		t.Fatalf("sjson-escaped marker must be found: %s", sjsonBody)
+	}
+
+	for name, body := range map[string]string{
+		"no markers":                   `{"messages":[{"role":"user","content":"nothing here"}]}`,
+		"prefix only but not a marker": `{"messages":[{"role":"user","content":"cg: and ` + escLT + `cg:"}]}`,
+	} {
+		if HasMarkersInMessages([]byte(body)) {
+			t.Errorf("%s: false positive on %s", name, body)
+		}
+	}
+}
+
 func TestResolve(t *testing.T) {
 	st := store.NewMemory(store.Options{})
 	if _, ok := Resolve(st, "absent"); ok {

metrics/metrics.go

@@ -84,6 +84,15 @@ type Aggregator struct {
 	upstreamMsByp float64 // upstream latency on bypassed (baseline) requests
 	upstreamN     int64
 	upstreamNByp  int64
+	// SSE time-to-first-byte accounting: how long after the upstream call started the
+	// client got its first response byte, split by whether we had to buffer the whole
+	// stream to inspect it for an expand call. Buffering is the only thing that stops a
+	// stream being a stream, so counting it makes that cost visible instead of inferred
+	// (it used to be unconditional and unmeasured — issue #26).
+	sseTTFBMs    float64
+	sseTTFBMsBuf float64
+	sseStreamed  int64
+	sseBuffered  int64
 }
 
 type compStat struct {
@@ -183,6 +192,21 @@ func (a *Aggregator) RecordUpstreamLatency(ms float64, bypassed bool) {
 	a.mu.Unlock()
 }
 
+// RecordSSE notes one streaming response: ms from issuing the upstream request to
+// the first byte handed to the client, and whether the stream had to be fully
+// buffered first (which turns TTFB into total-response time).
+func (a *Aggregator) RecordSSE(ttfbMs float64, buffered bool) {
+	a.mu.Lock()
+	if buffered {
+		a.sseTTFBMsBuf += ttfbMs
+		a.sseBuffered++
+	} else {
+		a.sseTTFBMs += ttfbMs
+		a.sseStreamed++
+	}
+	a.mu.Unlock()
+}
+
 func (a *Aggregator) Run(r components.RunReport) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
@@ -219,6 +243,21 @@ type Snapshot struct {
 	AddedLatencyMsAvg     float64 `json:"cg_added_ms_avg"`
 	UpstreamMsAvg         float64 `json:"upstream_ms_avg"`
 	UpstreamMsAvgBypassed float64 `json:"upstream_ms_avg_bypassed"`
+	// SSE streaming health (#26). SSEBuffered counts streams context-guru had to read
+	// in full before the client saw a byte (to look for an expand tool call); those
+	// requests lose streaming entirely, so their TTFB is reported separately. A high
+	// buffered share on traffic that never expands is the regression to watch. All four
+	// count once per CLIENT REQUEST, not per upstream round: a request that drove
+	// several expand rounds waited for all of them.
+	SSEStreamed  int64   `json:"sse_streamed"`
+	SSEBuffered  int64   `json:"sse_buffered"`
+	SSETTFBMsAvg float64 `json:"sse_ttfb_ms_avg"` // streamed-through requests: a real TTFB
+	// SSETTFBMsAvgBuf is time-to-LAST-byte by construction, not a comparable TTFB: a
+	// buffered response is read in full before the client is written to, so its first
+	// byte cannot precede the buffer completing. Read it as "what buffering cost these
+	// requests", not as a latency to compare against sse_ttfb_ms_avg.
+	SSETTFBMsAvgBuf float64 `json:"sse_ttfb_ms_avg_buffered"`
+	SSEBufferedPct  float64 `json:"sse_buffered_pct"`
 }
 
 // Snapshot returns a point-in-time copy of the rollups.
@@ -257,11 +296,23 @@ func (a *Aggregator) Snapshot() Snapshot {
 	if a.upstreamNByp > 0 {
 		upAvgByp = a.upstreamMsByp / float64(a.upstreamNByp)
 	}
+	ttfb, ttfbBuf, bufPct := 0.0, 0.0, 0.0
+	if a.sseStreamed > 0 {
+		ttfb = a.sseTTFBMs / float64(a.sseStreamed)
+	}
+	if a.sseBuffered > 0 {
+		ttfbBuf = a.sseTTFBMsBuf / float64(a.sseBuffered)
+	}
+	if n := a.sseStreamed + a.sseBuffered; n > 0 {
+		bufPct = float64(a.sseBuffered) / float64(n) * 100
+	}
 	return Snapshot{
 		Requests: a.requests, TokensBefore: a.before, TokensAfter: a.after,
 		SavedTokens: saved, SavingsPct: pct,
 		WastedTokens: a.wasted, Bounces: a.bounces, AdjustedSaved: saved - a.wasted,
 		Components: comps, TopPassthrough: passthrough,
 		AddedLatencyMsAvg: addedAvg, UpstreamMsAvg: upAvg, UpstreamMsAvgBypassed: upAvgByp,
+		SSEStreamed: a.sseStreamed, SSEBuffered: a.sseBuffered,
+		SSETTFBMsAvg: ttfb, SSETTFBMsAvgBuf: ttfbBuf, SSEBufferedPct: bufPct,
 	}
 }

metrics/metrics_test.go

@@ -73,6 +73,28 @@ func TestLatencyAverages(t *testing.T) {
 	}
 }
 
+// TestSSEBufferingStats: streamed vs buffered SSE responses average separately and
+// the buffered share is reported, so a buffering regression is visible in /stats.
+func TestSSEBufferingStats(t *testing.T) {
+	a := NewAggregator()
+	if s := a.Snapshot(); s.SSEBufferedPct != 0 || s.SSETTFBMsAvg != 0 {
+		t.Fatalf("no SSE traffic should report zeros: %+v", s)
+	}
+	a.RecordSSE(20, false)
+	a.RecordSSE(40, false)
+	a.RecordSSE(900, true)
+	s := a.Snapshot()
+	if s.SSEStreamed != 2 || s.SSEBuffered != 1 {
+		t.Fatalf("counts=%d/%d want 2 streamed / 1 buffered", s.SSEStreamed, s.SSEBuffered)
+	}
+	if s.SSETTFBMsAvg != 30 || s.SSETTFBMsAvgBuf != 900 {
+		t.Fatalf("ttfb=%v buffered=%v want 30/900", s.SSETTFBMsAvg, s.SSETTFBMsAvgBuf)
+	}
+	if got := s.SSEBufferedPct; got < 33.3 || got > 33.4 {
+		t.Fatalf("buffered pct=%v want ~33.33", got)
+	}
+}
+
 // TestMutatedZeroSavingsNotPassthrough locks the fix for cacheinject-style
 // components: they change the request (add cache_control) but save no content
 // tokens, so they must NOT be flagged as dead weight in top_passthrough.