feat(engine): unify compaction under one Compactor abstraction; add summarizer + truncation; comprehensive stats

CLAUDE.md

@@ -14,7 +14,7 @@ is the behavioral reference — port its *logic*, re-implement its transport in
 - **No AuthBridge / kagenti-extensions code lives here.** That plugin is built in
   `kagenti-extensions` and depends on this repo. Keep the public API (`engine`,
   `surfaces`, `config`) clean and importable; never reach into another repo.
-- **Fail open, always.** Any error in any stage forwards the original request untouched.
+- **Fail open, always.** Any error in any compactor forwards the original request untouched.
   Reductions must be reversible (markers + rewind store). Never drop content that is only
   *predicted* unused — `provable_only` is on by default.
 
@@ -29,7 +29,7 @@ is the behavioral reference — port its *logic*, re-implement its transport in
 
 ## Layout
 
-`cmd/proxy` (binary) · `engine` (Transform/Expand + stages) · `surfaces` (wire⇄canonical)
+`cmd/proxy` (binary) · `engine` (Transform/Expand + Compactor pipeline) · `surfaces` (wire⇄canonical)
 · `internal/*` (types, extract, relevance, signals, taxonomy, actions, cache, markers,
 rewind, zones, session, compaction, tokens) · `config` · `observability` · `deploy` ·
 `docs`.

README.md

@@ -51,8 +51,9 @@ cheap-model extractor enabled, forwarding all traffic to an upstream gateway:
 `proxy` flags: `--addr` (default `:8080`), `--preset`
 (`safe|balanced|aggressive|cache|coding|mcp`), `--config` (a YAML file that names which
 components run, overrides `--preset`), `--upstream` (forward ALL requests here; default
-routes by provider), `--extract-model/-provider/-auth/-base` (enable + configure the
-cheap-model extractor; these override the config's transport block), `--max-body-bytes`
+routes by provider), `--extract-model/-provider/-auth/-base` and
+`--summarize-model/-provider/-auth/-base` (enable + configure the cheap-model extractor /
+summarizer; these override the config's transport block), `--max-body-bytes`
 (default 32 MiB; `0` = no cap), `--upstream-timeout` (default `0s` = none; a non-zero
 value caps the whole request and can truncate long SSE streams).
 
@@ -67,14 +68,14 @@ Endpoints the proxy serves:
 
 - `GET /stats` — process-wide reduction snapshot as JSON (tokens before/after/saved,
   cache injected, extracted, stage errors, added-latency p50/p95).
-- `GET /winnow/expand?id=<marker-id>` — returns the original bytes behind a reversibility
+- `GET /labcx/expand?id=<marker-id>` — returns the original bytes behind a reversibility
   marker.
 - `GET /health`, `GET /ready` — liveness/readiness.
 
 Bypass / disable:
 
-- `WINNOW_DISABLE=1` env var — run the proxy as a transparent passthrough.
-- `x-winnow-bypass: true` request header — skip reduction for that single request.
+- `LABCX_DISABLE=1` env var — run the proxy as a transparent passthrough.
+- `x-labcx-bypass: true` request header — skip reduction for that single request.
 
 ## Components
 
@@ -89,23 +90,33 @@ Bypass / disable:
   tool/MCP output; accepted only if **structurally contained** in the original. Strategies
   `code` (model writes a filter run in a **Starlark sandbox** over the full body), `single`
   (one-shot JSON-return filter), `rlm` (chunked), and a `deterministic` fallback.
+- **Summarize** (cheap model, opt-in) — replaces older turns with one factual
+  `<summary>` (ReSum-style prompt), keeping the last `keep_last` messages verbatim; the
+  dropped span is stored and recoverable.
+- **Truncate** (no model, opt-in) — the naive baseline: keep the last `keep_last`
+  messages, drop the rest behind one recoverable note. The control to measure smarter
+  compactors against.
 - **Tokenizer** — real BPE token counts via tiktoken `o200k_base` (`internal/tokens`); the
   same counter gates every reduction (a component never inflates an output).
 - **Reversibility** — namespaced markers + a content-addressed rewind store; recover via
-  `engine.FindMarkers` + `engine.Expand`, or the `/winnow/expand` endpoint.
+  `engine.FindMarkers` + `engine.Expand`, or the `/labcx/expand` endpoint.
 - **Observability** — OpenTelemetry GenAI semantic conventions (`gen_ai.*`) plus the live
   `/stats` aggregate.
 
 ## Config
 
+Every compaction approach — `reduce`, `extract`, `summarize`, `truncate`, `cache` —
+implements one interface, `engine.Compactor` (given the conversation, return the
+transformed conversation). The `compactors:` list selects which run and in what order.
+
 The example config is [`configs/lab-cx.yaml`](configs/lab-cx.yaml). It folds onto a base
-`preset`, then selects which stages/reducers/encoders/extract-strategies run, **purely by
-name**. Extension path — no core edit beyond one registration:
+`preset`, then selects which compactors/reducers/encoders/extract-strategies run, **purely
+by name**. Extension path — no core edit beyond one registration:
 
-1. **Add** a new encoder/reducer/extract-strategy/stage, and **register by name** in its
-   registry (encoders in `internal/reduce/actions.go`, reducers via
-   `reduce.RegisterReducer`, strategies in `internal/extract` `RunExtraction`, stages in
-   `engine` `builtinStages`).
+1. **Add** a new encoder/reducer/extract-strategy/compactor, and **register by name** in
+   its registry (encoders in `internal/reduce/actions.go`, reducers via
+   `reduce.RegisterReducer`, strategies in `internal/extract` `RunExtraction`, compactors
+   via `engine.(*Engine).Register`).
 2. **List it by name** in the matching list in `configs/lab-cx.yaml` (an empty/omitted list
    means "all built-in defaults"; list order sets priority/run order).
 
@@ -149,7 +160,7 @@ end-to-end **SWE-bench** run is a **documented runbook, not yet executed**
 ```
 cmd/proxy/        the lab-cx binary (proxy | stats | version)
 cmd/labcx-bench/  offline + --extract measurement harness
-engine/           Engine: Transform / Expand, stage orchestration (builtinStages)
+engine/           Engine: Transform / Expand, Compactor pipeline (builtinCompactors)
 surfaces/         anthropic | openai | gemini  wire <-> canonical request
 config/           Settings + presets, YAML config loader
 canon/            canonical request type

canon/request.go

@@ -8,7 +8,7 @@
 // (max_tokens, temperature, stream, metadata, ...) and per-block extensions
 // (cache_control). Keeping the whole object as decoded JSON preserves every field
 // across a decode→encode round-trip and lets stages add breakpoints freely — exactly
-// what the winnow prototype relied on. Typed helpers give ergonomic access to the
+// what the reference prototype relied on. Typed helpers give ergonomic access to the
 // parts stages actually touch (messages and content blocks).
 package canon
 
@@ -80,6 +80,17 @@ func (r Request) Messages() []map[string]any {
 	return out
 }
 
+// SetMessages replaces the message list. Compactors that rebuild the conversation
+// (summarize, truncate) write the new list back through this; block-level compactors
+// mutate the maps returned by Messages in place and need not call it.
+func (r Request) SetMessages(msgs []map[string]any) {
+	raw := make([]any, len(msgs))
+	for i, m := range msgs {
+		raw[i] = m
+	}
+	r.Root["messages"] = raw
+}
+
 // Blocks returns the content blocks of a message as a slice of maps. Anthropic
 // content may be a plain string (returned as a single synthesized text block) or a
 // list of block objects.

cmd/labcx-bench/main.go

@@ -92,7 +92,7 @@ func isolate(reducers, encoders []string, cmdFilter bool) config.Settings {
 		CmdFilter:       cmdFilter,
 		Reducers:        reducers,
 		Encoders:        encoders,
-		Stages:          []string{"reduce"},
+		Compactors:      []string{"reduce"},
 		ExtractEnabled:  false,
 	}
 }
@@ -245,7 +245,7 @@ func measure(fx fixture, fixtureText string, c component) row {
 	return r
 }
 
-// verifyReversible confirms the reduction is recoverable: if the component left a winnow
+// verifyReversible confirms the reduction is recoverable: if the component left a lab-cx
 // marker, the rewind store must return the exact original; if it left the text untouched
 // (a no-op for this fixture/component), that is trivially reversible. A reduction that
 // changed the text but exposes no recoverable original would be a (disallowed) lossy drop.
@@ -467,27 +467,29 @@ func measureExtract(fx extractFixture, fixtureText string, model engine.Model) e
 		ProvableOnly:    false,
 		CollapseOutputs: true,
 		Reducers:        []string{"__none__"}, // no deterministic reducer touches the body first
-		Stages:          []string{"reduce", "extract"},
-		ExtractEnabled:  true,
+		Compactors:      []string{"reduce"},
 		LLMCompactFloor: extractFloor,
 	}
 	eng := engine.New(settings, nil, nil)
 
 	cfg := engine.DefaultExtractConfig()
 	cfg.Floor = extractFloor
-	// EnableExtract performs the real wiring/splice; we then override its func with a
-	// behaviorally identical one that also records the chosen strategy.
-	eng.EnableExtract(model, cfg)
-
-	var chosen string
-	eng.SetExtract(capturingExtractFunc(eng, model, cfg.Floor, &chosen))
 
 	req := buildExtractRequest(fx, fixtureText)
 	before := req.Clone()
 	beforeText := toolResultTextAny(before)
 
 	start := time.Now()
+	// reduce runs first (no deterministic reducer touches the body); then we run the
+	// extraction ourselves over the same candidate set the engine's Extractor would, so
+	// we can record which strategy RunExtraction selected for the row.
 	out, _ := eng.Transform(context.Background(), req)
+	opts := reduce.DefaultOpts()
+	opts.ProtectRecent = 1
+	opts.ProvableOnly = false
+	opts.LLMCompactFloor = extractFloor
+	cands := reduce.SelectLLMCandidates(out, opts)
+	chosen := runCapturingExtract(eng, out, cands, model, cfg.Floor)
 	elapsed := time.Since(start)
 
 	afterText := toolResultTextAny(out)
@@ -513,37 +515,37 @@ func measureExtract(fx extractFixture, fixtureText string, model engine.Model) e
 	return r
 }
 
-// capturingExtractFunc returns an extract func equivalent to engine.EnableExtract's, but
-// it records the strategy RunExtraction selected for the (single) candidate into *chosen
-// and performs the reversible splice through the engine's public store + recovery marker.
-func capturingExtractFunc(eng *engine.Engine, model engine.Model, floor int, chosen *string) engine.ExtractFunc {
+// runCapturingExtract runs the same extraction the engine's Extractor would over cands,
+// but records which strategy RunExtraction selected and performs the reversible splice
+// through the engine's public store + recovery marker. Returns the chosen strategy.
+func runCapturingExtract(eng *engine.Engine, req canon.Request, cands []reduce.Candidate, model engine.Model, floor int) string {
 	icfg := extract.Cfg{Mode: "auto", Floor: floor, AllowDeterministic: true, MaxChars: 4000}
 	cache := extract.NewCache()
-	return func(ctx context.Context, req canon.Request, cands []reduce.Candidate) error {
-		goal := recentGoalText(req, 6)
-		keep := extract.HarvestIdentifiers(goal, 60)
-		for _, c := range cands {
-			func() {
-				defer func() { _ = recover() }() // fail-open per candidate
-				key := goalKey(extract.ContentKey(c.Text), goal, keep)
-				result, ok := cache.Get(key)
-				var strat string
-				if ok {
-					strat = "cache"
-				} else {
-					result, strat = extract.RunExtraction(ctx, c.Text, goal, keep, c.TokenEst, icfg, model)
-					if strat == "none" || result == "" {
-						*chosen = "none"
-						return
-					}
-					cache.Put(key, result)
+	goal := recentGoalText(req, 6)
+	keep := extract.HarvestIdentifiers(goal, 60)
+	ctx := context.Background()
+	chosen := "none"
+	for _, c := range cands {
+		func() {
+			defer func() { _ = recover() }() // fail-open per candidate
+			key := goalKey(extract.ContentKey(c.Text), goal, keep)
+			result, ok := cache.Get(key)
+			var strat string
+			if ok {
+				strat = "cache"
+			} else {
+				result, strat = extract.RunExtraction(ctx, c.Text, goal, keep, c.TokenEst, icfg, model)
+				if strat == "none" || result == "" {
+					chosen = "none"
+					return
 				}
-				*chosen = strat
-				spliceResult(eng, req, c, result)
-			}()
-		}
-		return nil
+				cache.Put(key, result)
+			}
+			chosen = strat
+			spliceResult(eng, req, c, result)
+		}()
 	}
+	return chosen
 }
 
 // spliceResult mirrors engine.(*Engine).splice using the public API: store the original