diff --git a/CLAUDE.md b/CLAUDE.md index 5549ad0..2c04931 100644 --- a/CLAUDE.md +++ b/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`. diff --git a/README.md b/README.md index 5f14954..988e51f 100644 --- a/README.md +++ b/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=` — returns the original bytes behind a reversibility +- `GET /labcx/expand?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 + `` (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 diff --git a/canon/request.go b/canon/request.go index 325b035..1c784a7 100644 --- a/canon/request.go +++ b/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. diff --git a/cmd/labcx-bench/main.go b/cmd/labcx-bench/main.go index 0b4d17d..79cda7e 100644 --- a/cmd/labcx-bench/main.go +++ b/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 diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index cc507ad..1338ec0 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -12,6 +12,7 @@ import ( "io" "net/http" "os" + "strings" "time" "github.com/kagenti/lab-context-engineering/config" @@ -80,6 +81,11 @@ func runProxy(args []string) { extractProvider := fs.String("extract-provider", "anthropic", "extraction model provider: anthropic|openai") extractBase := fs.String("extract-base", "", "base URL for the extraction model (default per provider)") extractAuth := fs.String("extract-auth", "x-api-key", "anthropic auth scheme: x-api-key|bearer (bearer for gateway endpoints)") + summarizeModel := fs.String("summarize-model", "", "enable LLM summarization with this model (e.g. claude-haiku-4-5)") + summarizeProvider := fs.String("summarize-provider", "anthropic", "summarizer model provider: anthropic|openai") + summarizeBase := fs.String("summarize-base", "", "base URL for the summarizer model (default per provider)") + summarizeAuth := fs.String("summarize-auth", "x-api-key", "anthropic auth scheme: x-api-key|bearer") + reduceCachedPrefix := fs.Bool("reduce-cached-prefix", false, "also run the deterministic reducers and the LLM extractor on the client's self-cached prefix (e.g. Claude Code) instead of deferring to its cache") maxBodyBytes := fs.Int64("max-body-bytes", 33554432, "cap request body size in bytes (32 MiB default); 0 means no cap") upstreamTimeout := fs.String("upstream-timeout", "0s", "max total time per upstream request (e.g. 30s); 0 means no timeout. WARNING: a non-zero value caps the WHOLE request including streamed responses and can truncate long LLM SSE streams") _ = fs.Parse(args) @@ -90,19 +96,20 @@ func runProxy(args []string) { os.Exit(2) } - if os.Getenv("WINNOW_DISABLE") == "1" { - fmt.Fprintln(os.Stderr, "lab-cx: WINNOW_DISABLE=1 — running as a transparent passthrough") + if os.Getenv("LABCX_DISABLE") == "1" { + fmt.Fprintln(os.Stderr, "lab-cx: LABCX_DISABLE=1 — running as a transparent passthrough") } // Resolve settings: a --config file (with named component selection) takes - // precedence over --preset. The file may also name the extraction transport - // (provider/model/auth/base); the matching --extract-* flags still override it. + // precedence over --preset. The file may also name each LLM compactor's transport + // (provider/model/auth/base/credentials); the matching --extract-*/--summarize-* + // flags still override it. var settings config.Settings - var tr config.Transport + var tr config.Transports settingsSrc := "preset=" + *preset if *configPath != "" { var loadErr error - settings, tr, loadErr = config.LoadWithTransport(*configPath) + settings, tr, loadErr = config.LoadFull(*configPath) if loadErr != nil { fmt.Fprintf(os.Stderr, "lab-cx: %v\n", loadErr) os.Exit(2) @@ -111,63 +118,91 @@ func runProxy(args []string) { } else { settings = config.Preset(*preset) } - // --extract-* flags override the config's transport block (flag set explicitly wins; - // otherwise fall back to the config value, then the flag default). - if v := flagOrConfig(fs, "extract-model", *extractModel, tr.Model); v != "" { + // --extract-*/--summarize-* flags override the config transport block (explicit flag + // wins; else the config value; else the flag default). + if v := flagOrConfig(fs, "extract-model", *extractModel, tr.Extract.Model); v != "" { *extractModel = v } - if v := flagOrConfig(fs, "extract-provider", *extractProvider, tr.Provider); v != "" { + if v := flagOrConfig(fs, "extract-provider", *extractProvider, tr.Extract.Provider); v != "" { *extractProvider = v } - if v := flagOrConfig(fs, "extract-auth", *extractAuth, tr.Auth); v != "" { + if v := flagOrConfig(fs, "extract-auth", *extractAuth, tr.Extract.Auth); v != "" { *extractAuth = v } - if v := flagOrConfig(fs, "extract-base", *extractBase, tr.Base); v != "" { + if v := flagOrConfig(fs, "extract-base", *extractBase, tr.Extract.Base); v != "" { *extractBase = v } + if v := flagOrConfig(fs, "summarize-model", *summarizeModel, tr.Summarize.Model); v != "" { + *summarizeModel = v + } + if v := flagOrConfig(fs, "summarize-provider", *summarizeProvider, tr.Summarize.Provider); v != "" { + *summarizeProvider = v + } + if v := flagOrConfig(fs, "summarize-auth", *summarizeAuth, tr.Summarize.Auth); v != "" { + *summarizeAuth = v + } + if v := flagOrConfig(fs, "summarize-base", *summarizeBase, tr.Summarize.Base); v != "" { + *summarizeBase = v + } - if os.Getenv("WINNOW_DISABLE") == "1" { + if os.Getenv("LABCX_DISABLE") == "1" { settings.Disabled = true } + if *reduceCachedPrefix { + settings.ReduceCachedPrefix = true + } eng := engine.New(settings, nil, nil) - if *extractModel != "" { - key := os.Getenv("WINNOW_EXTRACT_KEY") - var model engine.Model - switch *extractProvider { - case "openai": - if key == "" { - key = os.Getenv("OPENAI_API_KEY") - } - model = cheapmodel.OpenAI{ - BaseURL: *extractBase, APIKey: key, Model: *extractModel, - } - case "anthropic", "": - if key == "" { - key = os.Getenv("ANTHROPIC_API_KEY") - } - if key == "" { - key = os.Getenv("ANTHROPIC_AUTH_TOKEN") - } - model = cheapmodel.Anthropic{ - BaseURL: *extractBase, APIKey: key, Model: *extractModel, - AuthScheme: *extractAuth, - } - default: - fmt.Fprintf(os.Stderr, "lab-cx: unknown --extract-provider %q (want anthropic|openai)\n", *extractProvider) + + // Wire the extract compactor. Source "incoming" reuses the proxied request's own + // model + credentials (no static client); otherwise build a static client. + useIncoming := false + if tr.Extract.Source == "incoming" { + eng.EnableExtractSpec(engine.ModelSpec{UseIncoming: true}, engine.DefaultExtractConfig()) + useIncoming = true + fmt.Fprintln(os.Stderr, "lab-cx: extraction enabled (source=incoming)") + } else if *extractModel != "" { + key := resolveKey(tr.Extract, *extractProvider, "LABCX_EXTRACT_KEY") + model, err := buildModel(*extractProvider, *extractModel, *extractBase, *extractAuth, key) + if err != nil { + fmt.Fprintf(os.Stderr, "lab-cx: %v\n", err) os.Exit(2) } eng.EnableExtract(model, engine.DefaultExtractConfig()) fmt.Fprintf(os.Stderr, "lab-cx: extraction enabled (provider=%s model=%s)\n", *extractProvider, *extractModel) } + + // Wire the summarize compactor (same credential mechanism as extract). + if tr.Summarize.Source == "incoming" { + eng.EnableSummarizeSpec(engine.ModelSpec{UseIncoming: true}, engine.DefaultSummarizeConfig()) + useIncoming = true + fmt.Fprintln(os.Stderr, "lab-cx: summarization enabled (source=incoming)") + } else if *summarizeModel != "" { + key := resolveKey(tr.Summarize, *summarizeProvider, "LABCX_SUMMARIZE_KEY") + model, err := buildModel(*summarizeProvider, *summarizeModel, *summarizeBase, *summarizeAuth, key) + if err != nil { + fmt.Fprintf(os.Stderr, "lab-cx: %v\n", err) + os.Exit(2) + } + eng.EnableSummarize(model, engine.DefaultSummarizeConfig()) + fmt.Fprintf(os.Stderr, "lab-cx: summarization enabled (provider=%s model=%s)\n", *summarizeProvider, *summarizeModel) + } + agg := observability.NewAggregator(defaultCostRates()) - handler := proxyhttp.New(proxyhttp.Config{ + cfg := proxyhttp.Config{ Engine: eng, Upstream: *upstream, // Stream each event via slog AND fold it into the Aggregator served at /stats. Emitter: observability.Tee{observability.SlogEmitter{}, agg}, Aggregator: agg, MaxBodyBytes: *maxBodyBytes, UpstreamTimeout: upstreamTimeoutDur, - }) + } + if useIncoming { + // A compactor reuses the proxied request's own model + credentials: build a + // per-request model client from the incoming auth header, request model id, and + // resolved upstream base. + cfg.BuildRequestModel = incomingModel + } + handler := proxyhttp.New(cfg) fmt.Fprintf(os.Stderr, "lab-cx proxy listening on %s (%s)\n", *addr, settingsSrc) if err := http.ListenAndServe(*addr, handler); err != nil { @@ -233,3 +268,69 @@ func flagOrConfig(fs *flag.FlagSet, name, flagVal, cfgVal string) string { } return flagVal } + +// resolveKey resolves an LLM compactor's API key: an inline key in the config wins, +// then the config-named env var (key_env), then the compactor's fallback env var, then +// the provider default env. Secrets are never required in YAML, but allowed. +func resolveKey(tr config.ModelTransport, provider, fallbackEnv string) string { + if tr.APIKey != "" { + return tr.APIKey + } + if tr.KeyEnv != "" { + if v := os.Getenv(tr.KeyEnv); v != "" { + return v + } + } + if fallbackEnv != "" { + if v := os.Getenv(fallbackEnv); v != "" { + return v + } + } + if provider == "openai" { + return os.Getenv("OPENAI_API_KEY") + } + if v := os.Getenv("ANTHROPIC_API_KEY"); v != "" { + return v + } + return os.Getenv("ANTHROPIC_AUTH_TOKEN") +} + +// buildModel constructs a cheap-model client for a provider/model/base/auth/key. +func buildModel(provider, model, base, auth, key string) (engine.Model, error) { + switch provider { + case "openai": + return cheapmodel.OpenAI{BaseURL: base, APIKey: key, Model: model}, nil + case "anthropic", "": + return cheapmodel.Anthropic{BaseURL: base, APIKey: key, Model: model, AuthScheme: auth}, nil + default: + return nil, fmt.Errorf("unknown provider %q (want anthropic|openai)", provider) + } +} + +// incomingModel builds a per-request model client that reuses the proxied request's own +// model id + credentials (from the incoming auth header) and the resolved upstream base. +// Returns nil when no usable credential is present (the compactor then no-ops). +func incomingModel(surfaceName, model, base string, h http.Header) engine.Model { + switch surfaceName { + case "openai": + if key := bearerToken(h.Get("Authorization")); key != "" { + return cheapmodel.OpenAI{BaseURL: base, APIKey: key, Model: model} + } + case "anthropic": + if key := h.Get("x-api-key"); key != "" { + return cheapmodel.Anthropic{BaseURL: base, APIKey: key, Model: model, AuthScheme: "x-api-key"} + } + if key := bearerToken(h.Get("Authorization")); key != "" { + return cheapmodel.Anthropic{BaseURL: base, APIKey: key, Model: model, AuthScheme: "bearer"} + } + } + return nil +} + +func bearerToken(h string) string { + const p = "Bearer " + if strings.HasPrefix(h, p) { + return strings.TrimSpace(h[len(p):]) + } + return "" +} diff --git a/config/config.go b/config/config.go index 13e217d..94ecc84 100644 --- a/config/config.go +++ b/config/config.go @@ -1,6 +1,7 @@ // Package config holds the engine's settings and named presets. Settings are plain -// data; the proxy resolves them from WINNOW_*-style env / preset and hands them to the -// engine. Mirrors the knobs winnow exposes, trimmed to what the Go core uses. +// data; the proxy resolves them from LABCX_*-style env / preset and hands them to the +// engine. Mirrors the knobs the reference prototype exposes, trimmed to what the Go core +// uses. package config // Settings configures a reduction pass end to end. @@ -24,26 +25,38 @@ type Settings struct { CmdFilter bool RehydrateOnCompaction bool - // Extract stage (cheap-model projection of large structured outputs). + // Extract compactor (cheap-model projection of large structured outputs). ExtractEnabled bool ExtractMode string // "" | auto | single | rlm | deterministic LLMCompactFloor int LLMCompactStructuredOnly bool + // Summarize compactor (LLM trajectory summarization; off by default). + SummarizeEnabled bool + SummarizeLevel string // concise | regular | highly_detailed + SummarizeKeepLast int // recent messages kept verbatim after the summary + SummarizeTriggerTokens int // skip below this whole-conversation token count + + // Truncate compactor (naive: keep last N messages, drop older; off by default). + TruncateEnabled bool + TruncateKeepLast int + TruncateTriggerTokens int + // Named component selection. Each list is referenced BY NAME so that a component - // added tomorrow (a new reducer/encoder/extract strategy/stage) is controlled purely - // from config — register it by name in its package, then list it here. An empty list - // means "use all built-in defaults" (no filtering), preserving prior behavior. + // added tomorrow (a new reducer/encoder/extract strategy/compactor) is controlled + // purely from config — register it by name in its package, then list it here. An + // empty list means "use all built-in defaults" (no filtering). // // Reducers filters internal/reduce reducers by Reducer.Name (e.g. collapse, skeleton, // format). Encoders filters AND orders the format re-encoders by name (e.g. // json_compact, toon, jsonl, markdown_kv, tsv, csv). ExtractStrategies restricts the - // internal/extract strategy order (e.g. code, single, rlm, deterministic). Stages - // selects which engine stages run and in what order (reduce, extract, cache). + // internal/extract strategy order (e.g. code, single, rlm, deterministic). Compactors + // selects which compactors run and in what order (reduce, extract, summarize, + // truncate, cache). Reducers []string Encoders []string ExtractStrategies []string - Stages []string + Compactors []string } // Default is the "balanced" preset: lossless caching + accuracy-safe reduction on, diff --git a/config/file.go b/config/file.go index fcae72b..57e3de5 100644 --- a/config/file.go +++ b/config/file.go @@ -9,18 +9,18 @@ import ( ) // fileConfig is the on-disk YAML shape. It is intentionally separate from Settings: -// the file groups knobs under stage sections (reduce/extract/cache) and carries the -// named component-selection lists, which we then fold onto a Settings derived from the -// chosen preset. Pointers distinguish "key absent" (keep preset/default) from "key set -// to the zero value" (an explicit override). +// the file groups knobs under compactor sections (reduce/extract/summarize/truncate/ +// cache) and carries the named component-selection lists, which we then fold onto a +// Settings derived from the chosen preset. Pointers distinguish "key absent" (keep +// preset/default) from "key set to the zero value" (an explicit override). // // Extending the file is a one-line change: a new reducer/encoder/extract-strategy is // referenced purely by NAME in the relevant list (reduce.reducers, reduce.encoders, -// extract.strategies) — see configs/lab-cx.yaml. No new struct field is needed to wire -// a newly-registered component on or off. +// extract.strategies, compactors) — see configs/lab-cx.yaml. No new struct field is +// needed to wire a newly-registered component on or off. type fileConfig struct { - Preset string `yaml:"preset"` - Stages []string `yaml:"stages"` + Preset string `yaml:"preset"` + Compactors []string `yaml:"compactors"` Reduce *struct { Enabled *bool `yaml:"enabled"` @@ -37,20 +37,28 @@ type fileConfig struct { } `yaml:"reduce"` Extract *struct { - Enabled *bool `yaml:"enabled"` - Mode string `yaml:"mode"` - Strategies []string `yaml:"strategies"` - Floor *int `yaml:"floor"` - StructuredOnly *bool `yaml:"structured_only"` - // Provider/model/auth/base are transport concerns the proxy reads off the loaded - // config; the engine core ignores them. They live here so a single file fully - // describes a run. - Provider string `yaml:"provider"` - Model string `yaml:"model"` - Auth string `yaml:"auth"` - Base string `yaml:"base"` + Enabled *bool `yaml:"enabled"` + Mode string `yaml:"mode"` + Strategies []string `yaml:"strategies"` + Floor *int `yaml:"floor"` + StructuredOnly *bool `yaml:"structured_only"` + modelTransportYAML `yaml:",inline"` } `yaml:"extract"` + Summarize *struct { + Enabled *bool `yaml:"enabled"` + Level string `yaml:"level"` + KeepLast *int `yaml:"keep_last"` + TriggerTokens *int `yaml:"trigger_tokens"` + modelTransportYAML `yaml:",inline"` + } `yaml:"summarize"` + + Truncate *struct { + Enabled *bool `yaml:"enabled"` + KeepLast *int `yaml:"keep_last"` + TriggerTokens *int `yaml:"trigger_tokens"` + } `yaml:"truncate"` + Cache *struct { Enabled *bool `yaml:"enabled"` Breakpoints *int `yaml:"breakpoints"` @@ -59,42 +67,71 @@ type fileConfig struct { } `yaml:"cache"` } -// Transport holds the extraction model's transport knobs parsed from a config file. -// The engine ignores these; the proxy uses them to construct the cheap-model client. -// Empty fields mean "fall back to flag/default". -type Transport struct { +// modelTransportYAML is the LLM-model transport block shared by every LLM compactor +// (extract, summarize). The engine core ignores these; the proxy reads them to build +// the cheap-model client. Inlined into each compactor's YAML section. +type modelTransportYAML struct { + Source string `yaml:"source"` // "config" (default) | "incoming" + Provider string `yaml:"provider"` + Model string `yaml:"model"` + Auth string `yaml:"auth"` + Base string `yaml:"base"` + APIKey string `yaml:"api_key"` // inline secret (allowed); prefer key_env for hygiene + KeyEnv string `yaml:"key_env"` // env var name to read the key from +} + +// ModelTransport holds one LLM compactor's transport + credentials parsed from a config +// file. The engine ignores these; the proxy uses them to construct the model client. +// Empty fields fall back to flag/default. Source "incoming" reuses the proxied request's +// own model + credentials instead of a configured static client. +type ModelTransport struct { + Source string Provider string Model string Auth string Base string + APIKey string + KeyEnv string +} + +func (m modelTransportYAML) transport() ModelTransport { + return ModelTransport{ + Source: m.Source, Provider: m.Provider, Model: m.Model, + Auth: m.Auth, Base: m.Base, APIKey: m.APIKey, KeyEnv: m.KeyEnv, + } +} + +// Transports carries the per-compactor model transports the proxy needs to construct +// model clients. +type Transports struct { + Extract ModelTransport + Summarize ModelTransport } // Load reads a YAML config file and folds it onto the preset-derived Settings. Unknown // top-level keys are an error (strict). Missing keys fall back to the preset/defaults. -// It returns the resolved Settings; use LoadWithTransport when the extraction model's -// provider/model/auth/base are also needed. func Load(path string) (Settings, error) { - s, _, err := LoadWithTransport(path) + s, _, err := LoadFull(path) return s, err } -// LoadWithTransport is Load plus the extraction transport block (provider/model/auth/ -// base) the proxy needs to construct the cheap-model client. -func LoadWithTransport(path string) (Settings, Transport, error) { +// LoadFull is Load plus the per-compactor model transports (provider/model/auth/base/ +// credentials/source) the proxy needs to construct the model clients. +func LoadFull(path string) (Settings, Transports, error) { raw, err := os.ReadFile(path) if err != nil { - return Settings{}, Transport{}, fmt.Errorf("config: read %s: %w", path, err) + return Settings{}, Transports{}, fmt.Errorf("config: read %s: %w", path, err) } var fc fileConfig dec := yaml.NewDecoder(bytes.NewReader(raw)) dec.KnownFields(true) // strict: unknown keys are an error if err := dec.Decode(&fc); err != nil { - return Settings{}, Transport{}, fmt.Errorf("config: parse %s: %w", path, err) + return Settings{}, Transports{}, fmt.Errorf("config: parse %s: %w", path, err) } s := Preset(fc.Preset) - if fc.Stages != nil { - s.Stages = fc.Stages + if fc.Compactors != nil { + s.Compactors = fc.Compactors } if r := fc.Reduce; r != nil { @@ -127,6 +164,21 @@ func LoadWithTransport(path string) (Settings, Transport, error) { } } + if su := fc.Summarize; su != nil { + setBool(&s.SummarizeEnabled, su.Enabled) + if su.Level != "" { + s.SummarizeLevel = su.Level + } + setInt(&s.SummarizeKeepLast, su.KeepLast) + setInt(&s.SummarizeTriggerTokens, su.TriggerTokens) + } + + if tr := fc.Truncate; tr != nil { + setBool(&s.TruncateEnabled, tr.Enabled) + setInt(&s.TruncateKeepLast, tr.KeepLast) + setInt(&s.TruncateTriggerTokens, tr.TriggerTokens) + } + if c := fc.Cache; c != nil { setBool(&s.CacheEnabled, c.Enabled) setInt(&s.CacheBreakpoints, c.Breakpoints) @@ -134,11 +186,14 @@ func LoadWithTransport(path string) (Settings, Transport, error) { setBool(&s.CacheToolsBreakpoint, c.ToolsBreakpoint) } - var tr Transport + var t Transports if e := fc.Extract; e != nil { - tr = Transport{Provider: e.Provider, Model: e.Model, Auth: e.Auth, Base: e.Base} + t.Extract = e.transport() + } + if su := fc.Summarize; su != nil { + t.Summarize = su.transport() } - return s, tr, nil + return s, t, nil } func setBool(dst *bool, v *bool) { diff --git a/config/file_test.go b/config/file_test.go index 6735461..f25eaba 100644 --- a/config/file_test.go +++ b/config/file_test.go @@ -49,7 +49,7 @@ extract: func TestLoadFullShape(t *testing.T) { p := writeTmp(t, ` preset: balanced -stages: [reduce, extract, cache] +compactors: [reduce, extract, cache] reduce: protect_recent: 5 provable_only: false @@ -74,8 +74,8 @@ cache: if s.ProvableOnly { t.Errorf("ProvableOnly = true, want false (override)") } - if !reflect.DeepEqual(s.Stages, []string{"reduce", "extract", "cache"}) { - t.Errorf("Stages = %v", s.Stages) + if !reflect.DeepEqual(s.Compactors, []string{"reduce", "extract", "cache"}) { + t.Errorf("Compactors = %v", s.Compactors) } if !reflect.DeepEqual(s.ExtractStrategies, []string{"code", "single", "deterministic"}) { t.Errorf("ExtractStrategies = %v", s.ExtractStrategies) @@ -94,6 +94,53 @@ cache: } } +// TestLoadCompactorsAndTransports exercises the summarize/truncate blocks and the +// per-compactor LLM transport + credential fields, returned via LoadFull. +func TestLoadCompactorsAndTransports(t *testing.T) { + p := writeTmp(t, ` +preset: balanced +compactors: [extract, summarize, truncate, cache] +extract: + enabled: true + source: config + provider: anthropic + model: claude-haiku-4-5 + api_key: sk-inline +summarize: + enabled: true + level: highly_detailed + keep_last: 4 + trigger_tokens: 5000 + source: incoming + provider: openai + model: gpt-4o-mini + key_env: MY_SUM_KEY +truncate: + enabled: true + keep_last: 8 + trigger_tokens: 12000 +`) + s, tr, err := LoadFull(p) + if err != nil { + t.Fatalf("LoadFull: %v", err) + } + if !reflect.DeepEqual(s.Compactors, []string{"extract", "summarize", "truncate", "cache"}) { + t.Errorf("Compactors = %v", s.Compactors) + } + if !s.SummarizeEnabled || s.SummarizeLevel != "highly_detailed" || s.SummarizeKeepLast != 4 || s.SummarizeTriggerTokens != 5000 { + t.Errorf("summarize settings not folded: %+v", s) + } + if !s.TruncateEnabled || s.TruncateKeepLast != 8 || s.TruncateTriggerTokens != 12000 { + t.Errorf("truncate settings not folded: %+v", s) + } + if tr.Extract.Source != "config" || tr.Extract.Model != "claude-haiku-4-5" || tr.Extract.APIKey != "sk-inline" { + t.Errorf("extract transport = %+v", tr.Extract) + } + if tr.Summarize.Source != "incoming" || tr.Summarize.Provider != "openai" || tr.Summarize.KeyEnv != "MY_SUM_KEY" { + t.Errorf("summarize transport = %+v", tr.Summarize) + } +} + // TestLoadDefaultsFromPreset: missing keys fall back to the preset/defaults. func TestLoadDefaultsFromPreset(t *testing.T) { p := writeTmp(t, `preset: safe`) diff --git a/configs/lab-cx.yaml b/configs/lab-cx.yaml index ba38b18..fc6d7cb 100644 --- a/configs/lab-cx.yaml +++ b/configs/lab-cx.yaml @@ -11,38 +11,61 @@ # list that name under `reduce.reducers`. # • new EXTRACT STRATEGY : add it to internal/extract `RunExtraction`'s switch + # `rawStrategyOrder`, then list its name under `extract.strategies`. -# • new STAGE : register it in engine `builtinStages`, then list it under `stages`. +# • new COMPACTOR : implement engine.Compactor, register it by name (Engine.Register), +# then list its name under `compactors`. # Each list is referenced by NAME; enabling/disabling/ordering is pure config. # An empty or omitted list means "use all built-in defaults". # ───────────────────────────────────────────────────────────────────────────────────── preset: balanced # base defaults: safe|balanced|aggressive|cache|coding|mcp -stages: [reduce, extract, cache] # which engine stages run, in order +# Which compactors run, in order. Built-ins: reduce, extract, summarize, truncate, cache. +# extract runs before reduce so it sees large structured tool outputs intact. +compactors: [extract, reduce, cache] reduce: protect_recent: 2 # keep the N most recent turns at full fidelity provable_only: true # never drop merely-predicted-unused content # Enabled reducers, by Reducer.Name. Built-ins: collapse, skeleton, format. - # (cmdfilter and dedup are separate pipeline passes toggled by their own keys below.) reducers: [collapse, skeleton, format] cmd_filter: true # dedup/trim noisy shell command output # Format re-encoders, by name — this list also sets PRIORITY (first listed wins ties). - # Built-ins: json_compact, toon, jsonl, markdown_kv, tsv, csv. encoders: [json_compact, toon, jsonl, markdown_kv, tsv, csv] extract: enabled: true # cheap-model projection of large structured outputs mode: auto # auto | single | rlm | deterministic - # Allowed strategies, by name; intersected with the computed order. Built-ins: - # code, single, rlm, deterministic. strategies: [code, single, rlm, deterministic] floor: 3000 # token floor before extraction is considered - # Extraction model transport (read by the proxy; --extract-* flags override these): + # Model transport + credentials (read by the proxy; --extract-* flags override these). + # source: config (default) uses the model/credentials below; source: incoming reuses + # the proxied request's OWN model + credentials (no static client needed). + source: config # config | incoming provider: anthropic # anthropic | openai model: claude-haiku-4-5 auth: x-api-key # x-api-key | bearer (bearer for gateway endpoints) # base: https://gateway.example/v1 # optional: override the model base URL + # api_key: sk-... # optional inline key (prefer key_env for hygiene) + key_env: LABCX_EXTRACT_KEY # env var holding the key (default fallbacks apply) + +# LLM trajectory summarization (off unless listed in `compactors` AND enabled here). +# Replaces older messages with one factual summary, keeping the last keep_last verbatim. +summarize: + enabled: false + level: regular # concise | regular | highly_detailed + keep_last: 3 # recent messages kept verbatim after the summary + trigger_tokens: 8000 # skip summarizing below this whole-conversation size + source: config # same transport mechanism as extract (config|incoming) + provider: anthropic + model: claude-haiku-4-5 + auth: x-api-key + key_env: LABCX_SUMMARIZE_KEY + +# Naive truncation baseline (no model): keep the last keep_last messages, drop the rest. +truncate: + enabled: false + keep_last: 6 + trigger_tokens: 8000 cache: enabled: true diff --git a/deploy/eval-containers/README.md b/deploy/eval-containers/README.md index b593bda..395168f 100644 --- a/deploy/eval-containers/README.md +++ b/deploy/eval-containers/README.md @@ -26,7 +26,7 @@ docker compose -f compose.yaml -f .../deploy/eval-containers/compose.override.ya -y --abort-on-container-exit ``` -It adds a `winnow` service on the `internal` network, repoints the runner's +It adds a `lab-cx` service on the `internal` network, repoints the runner's `ANTHROPIC_BASE_URL` / `OPENAI_BASE_URL` / `GOOGLE_GEMINI_BASE_URL` at it, and sets `--upstream http://gateway:4000` so lab-cx forwards the full (prefixed) path on to the gateway. The agent still holds only `sk-proxy`; the real key stays in the gateway. @@ -38,7 +38,7 @@ runner (agent) ──▶ gateway ──▶ lab-cx ──▶ provider ``` lab-cx sees normalized (OpenAI-shaped) traffic. Point the gateway's `OPENAI_API_BASE` -at lab-cx (`http://winnow:8080`) and put `winnow` on the `upstream` network, with +at lab-cx (`http://lab-cx:8080`) and put `lab-cx` on the `upstream` network, with `--upstream` set to the real provider base. ## Notes diff --git a/deploy/eval-containers/compose.override.yaml b/deploy/eval-containers/compose.override.yaml index b8073e8..7271096 100644 --- a/deploy/eval-containers/compose.override.yaml +++ b/deploy/eval-containers/compose.override.yaml @@ -12,9 +12,28 @@ # placement instead, see README.md in this directory. services: - winnow: + lab-cx: image: ghcr.io/kagenti/lab-context-engineering:latest - command: ["proxy", "--addr", ":8080", "--preset", "balanced", "--upstream", "http://gateway:4000"] + # Run our components: reduce + cache (preset balanced) AND the cheap-model extractor. + # The extractor calls claude-haiku-4-5 back THROUGH the gateway's /anthropic route + # using the same sk-proxy credential the agent holds (x-api-key) — no real key here, + # no new egress. Summarize/truncate are intentionally NOT enabled for these runs. + command: + - "proxy" + - "--addr" + - ":8080" + - "--preset" + - "balanced" + - "--upstream" + - "http://gateway:4000" + - "--extract-model" + - "claude-haiku-4-5" + - "--extract-base" + - "http://gateway:4000/anthropic" + - "--extract-auth" + - "x-api-key" + environment: + LABCX_EXTRACT_KEY: sk-proxy networks: - internal @@ -22,8 +41,8 @@ services: environment: # Route the agent through lab-cx; it forwards the (full, prefixed) path to the # gateway. Suffix routing means /anthropic/v1/messages etc. are still reduced. - ANTHROPIC_BASE_URL: http://winnow:8080/anthropic - OPENAI_BASE_URL: http://winnow:8080/openai/v1 - GOOGLE_GEMINI_BASE_URL: http://winnow:8080/genai + ANTHROPIC_BASE_URL: http://lab-cx:8080/anthropic + OPENAI_BASE_URL: http://lab-cx:8080/openai/v1 + GOOGLE_GEMINI_BASE_URL: http://lab-cx:8080/genai depends_on: - - winnow + - lab-cx diff --git a/docs/RESULTS-offline.md b/docs/RESULTS-offline.md index 1bb6aba..acfcaff 100644 --- a/docs/RESULTS-offline.md +++ b/docs/RESULTS-offline.md @@ -9,7 +9,7 @@ same counter the engine uses to gate reductions. Counts are real BPE token counts, not a chars/4 estimate. - **Fixtures:** 10 real tool outputs committed under `testdata/fixtures/` (rtk command - outputs + winnow structured tool outputs). Provenance for every file is in + outputs + reference-prototype structured tool outputs). Provenance for every file is in `testdata/fixtures/README.md`. ## Reproduce diff --git a/docs/RESULTS.md b/docs/RESULTS.md index 65fc558..ca112c3 100644 --- a/docs/RESULTS.md +++ b/docs/RESULTS.md @@ -50,12 +50,24 @@ verbatim harness output. No numbers are invented; this page only indexes them. Bob backend; `/stats` after a 3-task suite. - **Detail:** [integration/bob.md](integration/bob.md). -## SWE-bench / eval-containers integration - -- **Headline:** wiring **validated** — the merged compose renders `winnow` before the - gateway and the lab-cx image builds (CGO, distroless/base). The **full SWE-bench run was - NOT executed** this session (multi-GB pulls + real model spend); the doc is the runbook - to execute, with no fabricated benchmark numbers. -- **Date:** 2026-06-24. **How produced:** structural `docker compose ... config` merge - validation (no image pull). -- **Detail + runbook:** [integration/swe-bench.md](integration/swe-bench.md). +## SWE-bench / eval-containers integration (real runs) + +- **Headline:** **10 real SWE-bench Verified tasks** (Claude Code agent → `lab-cx` → + gateway, `claude-sonnet-4-6` agent / `claude-haiku-4-5` extractor) reduced the agent's + input traffic by **−35.5% aggregate** (4,405,821 → 2,840,296 tokens; **1,565,525 saved** + over 263 requests), ranging **−10.4%…−51.8%** per task — entirely from the deterministic + reducers (collapse/skeleton/format/cmdfilter, **1,598 blocks reduced**) running on Claude + Code's self-cached prefix via `--reduce-cached-prefix`. 0 stage errors; p95 added latency + ~130–250 ms. +- **Caveats (honest):** the LLM extractor did not fire on these astropy tasks + (`extract_candidates=0` — no large *structured* tool outputs; its value is shown by the + extractor results above). Resolve status is reported per task, but Epoch AI marks the + **arm64** SWE-bench bases (used here for native Apple-Silicon runs) best-effort/untested + for grading, so `not resolved` rows may understate accuracy; **token reduction is + architecture-independent and valid**. `--reduce-cached-prefix` trades the client's + prompt-cache hits for these reductions (the deliberate choice for self-caching agents). +- **Date:** 2026-06-25. **How produced:** per-task eval images built locally + (`combination.Dockerfile` over the public Epoch base + the claude-code agent), run via + the eval-containers compose with [`deploy/eval-containers/compose.override.yaml`](../deploy/eval-containers/compose.override.yaml); + per-task reduction read from `lab-cx`'s `/stats`, resolve status from `result.json`. +- **Detail + per-task table + runbook:** [integration/swe-bench.md](integration/swe-bench.md). diff --git a/docs/RUNNING.md b/docs/RUNNING.md index 9d7f52a..578d174 100644 --- a/docs/RUNNING.md +++ b/docs/RUNNING.md @@ -32,17 +32,28 @@ OPENAI_BASE_URL=http://localhost:8080/v1 # OpenAI-wire agents ``` Flags: `--addr`, `--preset` (`safe|balanced|aggressive|cache|coding|mcp`), `--config`, -`--upstream`, `--extract-model/-provider/-auth/-base`, `--max-body-bytes` (default -33554432 = 32 MiB; `0` = no cap), `--upstream-timeout` (default `0s`; non-zero caps the -whole request incl. streamed responses). - -Extractor API key comes from the env: `WINNOW_EXTRACT_KEY`, else `ANTHROPIC_API_KEY` / -`ANTHROPIC_AUTH_TOKEN` (anthropic provider) or `OPENAI_API_KEY` (openai provider). +`--upstream`, `--extract-model/-provider/-auth/-base`, +`--summarize-model/-provider/-auth/-base`, `--reduce-cached-prefix` (run the reducers + +extractor on a self-caching client's cached prefix instead of deferring to its cache — +see below), `--max-body-bytes` (default 33554432 = 32 MiB; `0` = no cap), +`--upstream-timeout` (default `0s`; non-zero caps the whole request incl. streamed +responses). + +Self-caching agents (Claude Code): by default lab-cx defers to the client's own +`cache_control` (the cache stage stands down) and reduces little. Pass +`--reduce-cached-prefix` (or `reduce.reduce_cached_prefix: true`) to run the deterministic +reducers and the LLM extractor on the cached prefix too — real token reduction at the cost +of invalidating the client's cached prefix. + +Model credentials: an inline `api_key` (or `key_env`) in the config wins; else the +compactor's env var (`LABCX_EXTRACT_KEY` / `LABCX_SUMMARIZE_KEY`); else the provider +default (`ANTHROPIC_API_KEY` / `ANTHROPIC_AUTH_TOKEN`, or `OPENAI_API_KEY`). Set +`source: incoming` to reuse the proxied request's own model + credentials instead. Disable / bypass: -- `WINNOW_DISABLE=1` — transparent passthrough for the whole process. -- `x-winnow-bypass: true` header — skip reduction for one request. +- `LABCX_DISABLE=1` — transparent passthrough for the whole process. +- `x-labcx-bypass: true` header — skip reduction for one request. ## Read /stats @@ -52,9 +63,22 @@ curl -s http://localhost:8080/stats # raw JSON ./bin/lab-cx stats --addr http://localhost:8080 ``` -`/stats` fields: `requests`, `tokens_before`, `tokens_after`, `tokens_saved`, -`cache_injected`, `extracted`, `stage_errors`, `added_latency_p50_ms`, -`added_latency_p95_ms`. Recover an omitted block: `GET /winnow/expand?id=`. +`/stats` is comprehensive — global totals, a per-session breakdown, and a recent-per-call +ring: + +- **Global:** `requests`, `tokens_before/after/saved`, `reduction_ratio`, `cache_injected`, + `extracted`, `reduced_blocks`, `extract_candidates`, `stage_errors`, `cost_saved_usd`, + `sessions`, and a `latency` object (`p50_ms`, `p95_ms`, `p99_ms`, `max_ms`, `mean_ms`; + the flat `added_latency_p50_ms`/`p95_ms` are kept for compatibility). +- **`session_stats[]`** — per conversation: the same token/cache/extract/reduced metrics, + `cost_saved_usd`, `tools_total`, `tool_def_tokens`, `model`, `surface`, and its own + `latency` distribution. +- **`recent_calls[]`** (last 256) — every metric for each call: tokens + ratio, + `cache_injected`, `extracted`, `reduced_blocks`, `extract_candidates`, `frozen_messages`, + `rehydrated`, `at_compaction`, `session_id`, `request_model`, and `added_latency_ms`. + +The same per-call fields are emitted as a structured slog event per request +(`gen_ai.*` + `context_engineering.*`). Recover an omitted block: `GET /labcx/expand?id=`. ## Measure: offline deterministic harness @@ -106,18 +130,25 @@ name in the config — no other core edit needed. | Component | Register it in | Then list it under | |---|---|---| -| **Stage** | `engine` `builtinStages` map (`engine/engine.go`) | `stages:` | +| **Compactor** | `engine.(*Engine).Register` / `builtinCompactors` map (`engine/engine.go`) | `compactors:` | | **Reducer** | `reduce.RegisterReducer(...)` (its `Reducer.Name`) | `reduce.reducers:` | | **Encoder** (format re-encoder) | `allEncoders` table in `internal/reduce/actions.go` (unique name + rank) | `reduce.encoders:` (order = tie-break priority) | | **Extract strategy** | `RunExtraction`'s switch + `rawStrategyOrder` in `internal/extract/extract.go` | `extract.strategies:` | -Built-ins: stages `reduce, extract, cache`; reducers `collapse, skeleton, format` -(`cmdfilter` and `dedup` are separate passes toggled by their own keys); encoders -`json_compact, toon, jsonl, markdown_kv, tsv, csv`; strategies +Every compaction approach (`reduce`, `extract`, `summarize`, `truncate`, `cache`) +implements one `engine.Compactor` interface — given the conversation, return the +transformed conversation. Built-ins: compactors `reduce, extract, summarize, truncate, +cache` (default order `extract, reduce, cache`; summarize/truncate are opt-in); reducers +`collapse, skeleton, format` (`cmdfilter` and `dedup` are separate passes toggled by their +own keys); encoders `json_compact, toon, jsonl, markdown_kv, tsv, csv`; strategies `code, single, rlm, deterministic`. Other `reduce` keys: `protect_recent` (keep N most recent turns at full fidelity), `provable_only` (never drop merely-predicted-unused content), `cmd_filter` (bool). -`extract` keys: `enabled`, `mode` (`auto|single|rlm|deterministic`), `floor` (token floor -before extraction is considered), and the transport block (`provider`, `model`, `auth`, -`base`) which the `--extract-*` flags override. `cache` keys: `enabled`, `breakpoints`. +`extract` and `summarize` share one LLM transport block: `provider`, `model`, `auth`, +`base`, plus credentials (`api_key` inline or `key_env` naming an env var) and `source` +(`config` = use the configured model; `incoming` = reuse the proxied request's own model + +credentials). `--extract-*` / `--summarize-*` flags override it. `extract` keys: `enabled`, +`mode` (`auto|single|rlm|deterministic`), `floor`. `summarize` keys: `enabled`, `level` +(`concise|regular|highly_detailed`), `keep_last`, `trigger_tokens`. `truncate` keys (no +model): `enabled`, `keep_last`, `trigger_tokens`. `cache` keys: `enabled`, `breakpoints`. diff --git a/docs/integration/bob.md b/docs/integration/bob.md index 513490a..0d1d31f 100644 --- a/docs/integration/bob.md +++ b/docs/integration/bob.md @@ -8,7 +8,7 @@ unlike Claude Code lab-cx's cache-injection lever fires on every request. ## Setup ```sh -# Bob's API key (here from winnow/.env: BOBSHELL_API_KEY=bob_...) +# Bob's API key (here from ../winnow/.env: BOBSHELL_API_KEY=bob_...) export BOBSHELL_API_KEY=... # Start lab-cx: agent upstream = Bob's backend; cheap-model extractor = your gateway. diff --git a/docs/integration/claude-code.md b/docs/integration/claude-code.md index 65250c4..d831c84 100644 --- a/docs/integration/claude-code.md +++ b/docs/integration/claude-code.md @@ -49,7 +49,7 @@ The task completed correctly through the proxy. `/stats` after the run: integration works. - **`cache_injected: 0`** — Claude Code is a *self-caching* client (it sends its own `cache_control` breakpoints), so lab-cx's cache stage correctly **stands down** rather - than fight the client's cache. This matches the winnow finding that on Claude Code the + than fight the client's cache. This matches the reference prototype's finding that on Claude Code the proxy is cost-neutral by design. - **`tokens_saved: 0`** — a tiny 2-file, 2-turn task has no large/stale/duplicate tool outputs to reduce and nothing over the extraction floor. lab-cx safely did nothing. diff --git a/docs/integration/swe-bench.md b/docs/integration/swe-bench.md index a1ee450..f5ae646 100644 --- a/docs/integration/swe-bench.md +++ b/docs/integration/swe-bench.md @@ -17,12 +17,12 @@ EVAL_TASK_ID=astropy__astropy-12907 EVAL_AGENT=claude-code EVAL_MODEL=anthropic/ OPENAI_API_BASE=... OPENAI_API_KEY=... \ docker compose -f compose.yaml -f /path/to/lab-context-engineering/deploy/eval-containers/compose.override.yaml \ config --services -# -> otelcol, gateway, winnow, runner (winnow inserted before the gateway) +# -> otelcol, gateway, lab-cx, runner (lab-cx inserted before the gateway) ``` -The override adds a `winnow` service on the `internal` network, repoints the runner's -`ANTHROPIC_BASE_URL` (`http://gateway:4000/anthropic`) to `http://winnow:8080/anthropic`, -and sets `winnow --upstream http://gateway:4000`. The agent still holds only `sk-proxy`; +The override adds a `lab-cx` service on the `internal` network, repoints the runner's +`ANTHROPIC_BASE_URL` (`http://gateway:4000/anthropic`) to `http://lab-cx:8080/anthropic`, +and sets `lab-cx --upstream http://gateway:4000`. The agent still holds only `sk-proxy`; the real key stays in the gateway. lab-cx's suffix routing reduces the `/anthropic/...` prefixed path and forwards the full path on. @@ -36,7 +36,7 @@ prefixed path and forwards the full path on. `OPENAI_API_BASE`/`OPENAI_API_KEY` + `EVAL_MODEL` per the eval-containers docs. For an **Anthropic-compatible gateway** (e.g. the IBM LiteLLM endpoint used in this repo's other tests), either configure the eval gateway's provider to that endpoint, or set - `winnow --upstream` directly to it (bypassing the eval gateway) — winnow forwards the + `lab-cx --upstream` directly to it (bypassing the eval gateway) — lab-cx forwards the agent's `Authorization` header through. Use `--extract-auth bearer` for bearer-token gateways. 3. Run a task (or a slice from `tasks.txt`) with and without the override and compare: @@ -48,30 +48,79 @@ prefixed path and forwards the full path on. 4. Read the metrics: - **eval-containers**: `result.json` (resolve status) + `trajectory.jsonl` (per-call `total_tokens`, `cost_usd`) in the shared `output` volume. - - **lab-cx**: `curl http://winnow:8080/stats` (tokens_before/after/saved, cache, + - **lab-cx**: `curl http://lab-cx:8080/stats` (tokens_before/after/saved, cache, extracted, added latency) — run from inside the compose network, or expose the port. - Compare with-vs-without lab-cx on the same `EVAL_TASK_ID` for the token/cost delta; `result.json` resolve status confirms accuracy is preserved. -## Status (honest) - -- **Wiring: validated** — the merged compose renders correctly (`winnow` before - `gateway`) and the lab-cx image builds (CGO, distroless/base). -- **Full SWE-bench run: NOT executed in this session.** It requires multi-GB image - pulls, configuring the eval gateway against the model provider, and real model spend + - long agent-loop runtime — out of scope for a single session, and we do not fabricate - benchmark numbers. The runbook above is what to execute. -- The measured evidence we DO have is real and reproducible: deterministic components - −93% aggregate on real fixtures ([../RESULTS-offline.md](../RESULTS-offline.md)) and - the haiku extractor −56%…−80% on structured outputs - ([../RESULTS-extract.md](../RESULTS-extract.md)), plus a real Claude Code run through - the proxy ([claude-code.md](claude-code.md)). SWE-bench is the end-to-end validation to - run next with a provisioned gateway. +## Results (10 real tasks, 2026-06-25) + +Ten SWE-bench Verified tasks run Claude Code (`claude-sonnet-4-6`) → `lab-cx` → gateway, +with the **deterministic reducers + extractor on the cached prefix** (`--reduce-cached-prefix`) +and the extractor pinned to `claude-haiku-4-5`. Per-task input-token reduction from +`lab-cx`'s `/stats`, resolve from `result.json`: + +| task | reqs | tokens before | tokens after | saved | % | blocks | resolve | +|------|-----:|--------------:|-------------:|------:|-----:|-------:|---------| +| astropy-12907 | 16 | 210,638 | 101,472 | 109,166 | 51.8 | — | passed | +| astropy-13033 | 29 | 659,220 | 573,936 | 85,284 | 12.9 | 133 | — | +| astropy-13236 | 45 | 416,300 | 237,675 | 178,625 | 42.9 | 350 | not resolved | +| astropy-13398 | 28 | 792,753 | 401,839 | 390,914 | 49.3 | 176 | not resolved | +| astropy-13453 | 41 | 1,129,296 | 567,732 | 561,564 | 49.7 | 425 | passed | +| astropy-13579 | 9 | 52,283 | 26,409 | 25,874 | 49.5 | 7 | not resolved | +| astropy-13977 | 27 | 354,322 | 282,680 | 71,642 | 20.2 | 154 | not resolved | +| astropy-14096 | 19 | 259,925 | 223,951 | 35,974 | 13.8 | 58 | passed | +| astropy-14182 | 36 | 452,919 | 354,583 | 98,336 | 21.7 | 286 | passed | +| astropy-14309 | 10 | 78,165 | 70,019 | 8,146 | 10.4 | 9 | passed | +| **TOTAL** | **263** | **4,405,821** | **2,840,296** | **1,565,525** | **35.5** | **1,598** | | + +- **Reduction is real and substantial: −35.5% aggregate** (−10.4%…−51.8% per task), all + from the deterministic reducers (1,598 blocks). 0 stage errors; p95 added latency + ~130–250 ms. +- `extract_candidates=0` on every task: these astropy tasks emit no large *structured* + tool outputs, so the LLM extractor had nothing to claim — its value is the separate + −56%…−80% on structured fixtures ([../RESULTS-extract.md](../RESULTS-extract.md)). +- **Honest caveats:** runs used the **arm64** Epoch bases for native Apple-Silicon + execution; Epoch marks arm64 grading best-effort/untested, so `not resolved` rows may + understate accuracy (token reduction is architecture-independent and valid). 12907/13033 + predate per-task `result.json` capture. `--reduce-cached-prefix` trades the client's + prompt-cache hits for these reductions. + +### How these were produced (reproduce) + +Most of the 500 `tasks.txt` instances have **no prebuilt** `evals/--claude-code` +image; only a demo (12907) is published. Build per-task images locally from the **public** +Epoch base, then stitch with the agent: + +```sh +# 1. benchmark image (public Epoch base; arm64 for native Apple Silicon, x86_64 otherwise) +docker build --platform linux/arm64 -f containers/benchmarks/swe-bench/Dockerfile \ + --build-arg EVAL_TASK_ID= --build-arg EVAL_BASE_ARCH=arm64 \ + --build-context "ghcr.io/exgentic/core/entrypoint=docker-image://ghcr.io/exgentic/core/entrypoint:latest" \ + -t localhost:5001/swe-bench-:latest containers/benchmarks/swe-bench +docker push localhost:5001/swe-bench-:latest # local registry: BuildKit resolves FROM via a registry +# 2. stitch benchmark + agent + gosu into the runner image +docker build --platform linux/arm64 -f containers/core/combination.Dockerfile \ + --build-arg BENCHMARK_IMAGE=localhost:5001/swe-bench-:latest \ + --build-arg AGENT_IMAGE=ghcr.io/exgentic/agents/claude-code:latest \ + --build-arg GOSU_IMAGE=ghcr.io/exgentic/core/gosu:latest \ + -t ghcr.io/exgentic/evals/swe-bench---claude-code:latest containers/core +``` + +Gotchas learned: (a) build for the host arch — an amd64-only image won't run `node` +without emulation on arm64; (b) Claude Code 2.1.x sends adaptive **thinking** that some +gateways reject for haiku — run the agent on a thinking-capable model +(`ANTHROPIC_MODEL=claude-sonnet-4-6`) or cap `ANTHROPIC_MODEL_SUPPORTED_CAPABILITIES=effort`; +(c) the eval gateway force-maps to `EVAL_MODEL`, so to keep the extractor on haiku point +`--extract-base` at the model endpoint directly (the override does this). ## Note for self-caching agents -Claude Code self-caches, so on SWE-bench lab-cx's biggest lever is the **Reduce** -pre-passes (cmdfilter/dedup/skeleton/format) and the **extractor** on large tool -outputs, not cache injection (it defers — see [claude-code.md](claude-code.md)). Enable -`reduce_cached_prefix` in the config to also re-cache a smaller prefix on self-caching -clients. +Claude Code self-caches, so by default lab-cx defers to its `cache_control` (cache stage +stands down) and reduces little. To run the **Reduce** pre-passes +(cmdfilter/dedup/skeleton/format) and the **extractor** on the cached prefix anyway — the +configuration that produced the results above — pass `--reduce-cached-prefix` (or +`reduce.reduce_cached_prefix: true`). This re-reduces (and so invalidates) the client's +cached prefix: you trade its cheap cached-token reads for the reductions, which is the +right call when measuring the components or when reduction matters more than the client +cache. diff --git a/engine/compactor_test.go b/engine/compactor_test.go new file mode 100644 index 0000000..19622cb --- /dev/null +++ b/engine/compactor_test.go @@ -0,0 +1,137 @@ +package engine + +import ( + "context" + "strings" + "testing" + + "github.com/kagenti/lab-context-engineering/canon" + "github.com/kagenti/lab-context-engineering/config" +) + +// fakeModel returns a fixed completion — enough to drive summarize without a network. +type fakeModel struct{ out string } + +func (f fakeModel) Complete(_ context.Context, _ string) (string, error) { return f.out, nil } + +// sixMessages builds a request with one system field and six alternating turns. +func sixMessages(t *testing.T) canon.Request { + t.Helper() + body := []byte(`{"system":"be helpful","messages":[ + {"role":"user","content":[{"type":"text","text":"task: fix the bug"}]}, + {"role":"assistant","content":[{"type":"text","text":"looking"}]}, + {"role":"user","content":[{"type":"text","text":"more context"}]}, + {"role":"assistant","content":[{"type":"text","text":"investigating"}]}, + {"role":"user","content":[{"type":"text","text":"any progress"}]}, + {"role":"assistant","content":[{"type":"text","text":"almost done"}]} + ]}`) + req, err := canon.Decode(body) + if err != nil { + t.Fatalf("decode: %v", err) + } + return req +} + +// TestTruncateKeepsLastAndRecovers: naive truncate keeps the last N messages behind one +// recoverable note. +func TestTruncateKeepsLastAndRecovers(t *testing.T) { + s := config.Default() + s.Compactors = []string{truncateName} + s.TruncateEnabled = true + s.TruncateKeepLast = 2 + e := New(s, nil, nil) + + out, _ := e.Transform(context.Background(), sixMessages(t)) + msgs := out.Messages() + if len(msgs) != 3 { // 1 note + last 2 + t.Fatalf("want 3 messages (note + last 2), got %d", len(msgs)) + } + note, _ := msgs[0]["content"].(string) + if !strings.Contains(note, "Truncated history") { + t.Fatalf("first message is not the truncation note: %q", note) + } + body, _ := out.Encode() + ids := FindMarkers(string(body)) + if len(ids) == 0 { + t.Fatalf("truncation left no recoverable marker") + } + if _, ok := e.Expand(ids[0]); !ok { + t.Fatalf("Expand could not recover the truncated span %s", ids[0]) + } +} + +// TestSummarizeReplacesHistory: summarize replaces older turns with one and +// keeps the last KeepLast verbatim; the dropped span is recoverable. +func TestSummarizeReplacesHistory(t *testing.T) { + s := config.Default() + s.Compactors = []string{summarizeName} + s.SummarizeKeepLast = 1 + e := New(s, nil, nil) + e.EnableSummarize(fakeModel{out: "KEYFACTS about the bug"}, DefaultSummarizeConfig()) + + out, _ := e.Transform(context.Background(), sixMessages(t)) + msgs := out.Messages() + if len(msgs) != 2 { // 1 summary note + last 1 + t.Fatalf("want 2 messages (summary + last 1), got %d", len(msgs)) + } + note, _ := msgs[0]["content"].(string) + if !strings.Contains(note, "") || !strings.Contains(note, "KEYFACTS") { + t.Fatalf("summary note missing summary/content: %q", note) + } + body, _ := out.Encode() + if ids := FindMarkers(string(body)); len(ids) == 0 { + t.Fatalf("summarize left no recoverable marker") + } +} + +// taggerCompactor is a trivial custom Compactor used to prove a new approach plugs in +// purely by name, with no engine edits. +type taggerCompactor struct{} + +func (taggerCompactor) Name() string { return "tagger" } +func (taggerCompactor) Enabled(*Context) bool { return true } +func (taggerCompactor) Compact(req canon.Request, _ *Report, _ *Context) (canon.Request, error) { + msgs := req.Messages() + msgs = append(msgs, map[string]any{"role": "user", "content": "TAGGED"}) + req.SetMessages(msgs) + return req, nil +} + +// TestCustomCompactorByName: registering a Compactor by name and listing it in +// Settings.Compactors runs it — the generic-abstraction extension path. +func TestCustomCompactorByName(t *testing.T) { + s := config.Default() + s.Compactors = []string{"tagger"} + e := New(s, nil, nil) + e.Register("tagger", taggerCompactor{}) + + out, _ := e.Transform(context.Background(), sixMessages(t)) + msgs := out.Messages() + last, _ := msgs[len(msgs)-1]["content"].(string) + if last != "TAGGED" { + t.Fatalf("custom compactor did not run; last message = %q", last) + } +} + +// TestIncomingModelSpec: a compactor with ModelSpec{UseIncoming:true} no-ops without a +// request model and runs when the host supplies one via WithRequestModel. +func TestIncomingModelSpec(t *testing.T) { + s := config.Default() + s.Compactors = []string{summarizeName} + s.SummarizeKeepLast = 1 + e := New(s, nil, nil) + e.EnableSummarizeSpec(ModelSpec{UseIncoming: true}, DefaultSummarizeConfig()) + + // No request model → fail-open no-op (all six messages survive). + out, _ := e.Transform(context.Background(), sixMessages(t)) + if len(out.Messages()) != 6 { + t.Fatalf("without a request model, summarize should no-op; got %d messages", len(out.Messages())) + } + + // With a request model → summarized down to note + last 1. + ctx := WithRequestModel(context.Background(), fakeModel{out: "incoming"}) + out2, _ := e.Transform(ctx, sixMessages(t)) + if len(out2.Messages()) != 2 { + t.Fatalf("with a request model, summarize should run; got %d messages", len(out2.Messages())) + } +} diff --git a/engine/engine.go b/engine/engine.go index 5ee1f72..1fc7990 100644 --- a/engine/engine.go +++ b/engine/engine.go @@ -1,4 +1,4 @@ -// Package engine is the public entry point: it runs the stage pipeline over a +// Package engine is the public entry point: it runs a pipeline of Compactors over a // canonical request and exposes Transform (reduce a request, fail-open) and Expand // (recover a collapsed block). A host — the proxy binary, an eval-containers adapter, // or a Kagenti AuthBridge plugin — wraps an Engine; the engine has no transport code. @@ -20,32 +20,73 @@ import ( // (e.g. ahead of a summarization turn). Pair with Engine.Expand to recover originals. func FindMarkers(text string) []string { return markers.FindIDs(text) } -// ExtractFunc applies cheap-model extraction to large candidate outputs in place. -// It is injected by the host (the engine core has no model client). nil disables the -// Extract stage. It must fail open: on any error, leave candidates untouched. -type ExtractFunc func(ctx context.Context, req canon.Request, cands []reduce.Candidate) error +// Model is the LLM client an LLM-using Compactor calls. The host (the proxy, or a +// Kagenti AuthBridge plugin) injects a concrete implementation; the engine core has no +// model client of its own. +type Model interface { + Complete(ctx context.Context, prompt string) (string, error) +} + +// ModelSpec resolves to the Model a Compactor should use for a request. This is the +// ONE credential mechanism shared by every LLM compactor (extract, summarize, and any +// future approach): Source "config" uses Static — a client built once from the +// configured transport + credentials; Source "incoming" (UseIncoming) uses the +// per-request model the host built from the proxied request's own model + credentials +// (Context.RequestModel). +type ModelSpec struct { + Static Model + UseIncoming bool +} + +// Resolve returns the Model for this request, or nil if none is available (fail-open: +// the compactor then leaves the conversation untouched). +func (s ModelSpec) Resolve(c *Context) Model { + if s.UseIncoming { + return c.RequestModel + } + return s.Static +} + +type reqModelKey struct{} + +// WithRequestModel returns a context carrying the model built from the incoming +// request's own model + credentials. The proxy sets this per request; compactors with +// a ModelSpec{UseIncoming:true} resolve to it. +func WithRequestModel(ctx context.Context, m Model) context.Context { + return context.WithValue(ctx, reqModelKey{}, m) +} + +func requestModel(ctx context.Context) Model { + m, _ := ctx.Value(reqModelKey{}).(Model) + return m +} -// Context is the per-request state a stage may read. Plain services only — no -// transport types — so the same stages run in any host. +// Context is the per-request state a compactor may read. type Context struct { Settings config.Settings Store store.Rewind Evictions *store.Eviction ClientCacheFloor int // cache.FloorIndex of the incoming request StickyIDs map[string]struct{} // ids reduced on prior turns (cache stability) - Extract ExtractFunc + RequestModel Model // model built from the incoming request (nil if none) GoCtx context.Context } -// Stage is one transformation over the canonical request. -type Stage interface { +// Compactor is the single abstraction every compaction approach implements: given the +// canonical request (messages + tools) it returns the transformed request. reduce, +// extract, summarize, truncate and cache all implement it. Add a new approach by +// implementing Compactor and registering it by name (Engine.Register), then listing +// that name in Settings.Compactors. +type Compactor interface { Name() string Enabled(*Context) bool - Run(req canon.Request, agg *Report, ctx *Context) error + // Compact transforms the conversation and returns the new request. Fail-open: the + // engine snapshots and restores the request if Compact returns an error or panics. + Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) } // Report aggregates what the pipeline did. Embeds the reduce report and adds -// cross-stage info. +// cross-compactor info. type Report struct { Skipped bool StageErrors int @@ -54,45 +95,31 @@ type Report struct { Reduce reduce.Report } -// Engine runs stages over requests. Construct with New. +// Engine runs a name-resolved pipeline of Compactors over requests. Construct with New. type Engine struct { settings config.Settings store store.Rewind evict *store.Eviction - stages []Stage - extract ExtractFunc + registry map[string]Compactor } -// builtinStages maps a stage NAME to its built-in implementation. Stage selection in a -// config file ("stages: [reduce, extract, cache]") is resolved through this table, so a -// stage added tomorrow is wired on/off purely by name — register it here, then list it. -var builtinStages = map[string]func() Stage{ - "reduce": func() Stage { return ReduceStage{} }, - "extract": func() Stage { return ExtractStage{} }, - "cache": func() Stage { return CacheStage{} }, -} +// Compactor names. defaultPipeline runs extract BEFORE reduce so the extractor sees +// large structured tool outputs intact; reduce then applies lossless actions to +// whatever remains, and cache injects breakpoints last. +const ( + reduceName = "reduce" + extractName = "extract" + summarizeName = "summarize" + truncateName = "truncate" + cacheName = "cache" +) -// defaultStageOrder is the pipeline used when Settings.Stages is empty. -var defaultStageOrder = []string{"reduce", "extract", "cache"} +var defaultPipeline = []string{extractName, reduceName, cacheName} -// stagesFor resolves the stage list from names, falling back to the default order when -// the list is empty. Unknown names are skipped (fail-open: a typo can't crash startup). -func stagesFor(names []string) []Stage { - order := names - if len(order) == 0 { - order = defaultStageOrder - } - out := make([]Stage, 0, len(order)) - for _, n := range order { - if mk, ok := builtinStages[n]; ok { - out = append(out, mk()) - } - } - return out -} - -// New builds an engine with the configured stage pipeline (default: Reduce → Extract → -// Cache, or Settings.Stages by name when provided). +// New builds an engine with the built-in deterministic compactors registered +// (reduce, truncate, cache). LLM compactors (extract, summarize) are added by the host +// via EnableExtract / EnableSummarize. The pipeline order is Settings.Compactors (by +// name), or defaultPipeline when empty; names with no registered compactor are skipped. func New(settings config.Settings, st store.Rewind, ev *store.Eviction) *Engine { if st == nil { st = store.NewMemory(0) @@ -100,22 +127,22 @@ func New(settings config.Settings, st store.Rewind, ev *store.Eviction) *Engine if ev == nil { ev = store.NewEviction() } - return &Engine{ - settings: settings, store: st, evict: ev, - stages: stagesFor(settings.Stages), - } + e := &Engine{settings: settings, store: st, evict: ev, registry: map[string]Compactor{}} + e.Register(reduceName, Reducer{}) + e.Register(truncateName, Truncator{}) + e.Register(cacheName, Cacher{}) + return e } -// SetExtract injects the cheap-model extraction function (enables the Extract stage). -func (e *Engine) SetExtract(fn ExtractFunc) { e.extract = fn } +// Register adds (or overrides) a compactor by name. It participates in the pipeline +// only when its name appears in the resolved order (Settings.Compactors or default). +func (e *Engine) Register(name string, c Compactor) { e.registry[name] = c } -// RegisterStage inserts a custom stage at index (appended if index < 0 or too large). -func (e *Engine) RegisterStage(s Stage, index int) { - if index < 0 || index > len(e.stages) { - e.stages = append(e.stages, s) - return +func (e *Engine) order() []string { + if len(e.settings.Compactors) > 0 { + return e.settings.Compactors } - e.stages = append(e.stages[:index], append([]Stage{s}, e.stages[index:]...)...) + return defaultPipeline } // Store exposes the rewind store so the host can serve expand requests. @@ -124,10 +151,10 @@ func (e *Engine) Store() store.Rewind { return e.store } // Expand recovers the original content for a rewind id. func (e *Engine) Expand(id string) (string, bool) { return e.store.Get(id) } -// Transform reduces req in place and returns a Report. It never returns an error: -// any stage failure is isolated (the request is restored to its pre-stage state) and -// counted, so the worst case is a no-op forward. The returned request is the one the -// host should forward. +// Transform runs the compactor pipeline over req and returns a Report. It never returns +// an error: any compactor failure is isolated (the request is restored to its pre-step +// state) and counted, so the worst case is a no-op forward. The returned request is the +// one the host should forward. func (e *Engine) Transform(goctx context.Context, req canon.Request) (canon.Request, Report) { if e.settings.Disabled { return req, Report{Skipped: true} @@ -135,33 +162,42 @@ func (e *Engine) Transform(goctx context.Context, req canon.Request) (canon.Requ ctx := &Context{ Settings: e.settings, Store: e.store, Evictions: e.evict, ClientCacheFloor: cache.FloorIndex(req.Root), - Extract: e.extract, GoCtx: goctx, + RequestModel: requestModel(goctx), GoCtx: goctx, } var agg Report - for _, st := range e.stages { - if !st.Enabled(ctx) { + for _, name := range e.order() { + c, ok := e.registry[name] + if !ok { continue } - if err := safeRun(st, req, &agg, ctx); err != nil { + if !c.Enabled(ctx) { + continue + } + out, err := safeRun(c, req, &agg, ctx) + if err != nil { agg.StageErrors++ + continue } + req = out } return req, agg } -// safeRun isolates a stage: it snapshots the request and restores it if the stage -// returns an error or panics, so a faulty stage cannot corrupt the forwarded request. -func safeRun(st Stage, req canon.Request, agg *Report, ctx *Context) (err error) { +// safeRun isolates a compactor: it snapshots the request and restores it if Compact +// returns an error or panics, so a faulty compactor cannot corrupt the forwarded +// request. +func safeRun(c Compactor, req canon.Request, agg *Report, ctx *Context) (out canon.Request, err error) { snapshot := req.Clone() + out = req defer func() { if r := recover(); r != nil { - req.Root = snapshot.Root + out = canon.Request{Root: snapshot.Root} err = errFromPanic(r) } }() - if e := st.Run(req, agg, ctx); e != nil { - req.Root = snapshot.Root - return e + res, e := c.Compact(req, agg, ctx) + if e != nil { + return canon.Request{Root: snapshot.Root}, e } - return nil + return res, nil } diff --git a/engine/engine_test.go b/engine/engine_test.go index 5b29783..8b7591c 100644 --- a/engine/engine_test.go +++ b/engine/engine_test.go @@ -45,7 +45,7 @@ func TestTransformReducesAndCaches(t *testing.T) { out2, _ := out.Encode() ids := markers.FindIDs(string(out2)) if len(ids) == 0 { - t.Fatalf("expected a winnow marker in the reduced request") + t.Fatalf("expected a lab-cx marker in the reduced request") } if _, ok := e.Expand(ids[0]); !ok { t.Fatalf("Expand could not recover marker %s", ids[0]) diff --git a/engine/extractfunc.go b/engine/extractfunc.go index c1ff36e..7211d8f 100644 --- a/engine/extractfunc.go +++ b/engine/extractfunc.go @@ -1,7 +1,6 @@ package engine import ( - "context" "crypto/sha256" "encoding/hex" "strings" @@ -10,17 +9,12 @@ import ( "github.com/kagenti/lab-context-engineering/internal/extract" "github.com/kagenti/lab-context-engineering/internal/markers" "github.com/kagenti/lab-context-engineering/internal/reduce" + "github.com/kagenti/lab-context-engineering/internal/store" "github.com/kagenti/lab-context-engineering/internal/tokens" ) const goalTurns = 6 // recent turns whose text conditions extraction + keep-set -// Model is the cheap-model client extraction calls. A host (the proxy, or a Kagenti -// AuthBridge plugin) implements it; the engine has no model client of its own. -type Model interface { - Complete(ctx context.Context, prompt string) (string, error) -} - // ExtractConfig configures cheap-model extraction (public mirror of the internal cfg). type ExtractConfig struct { Mode string // "auto" | "single" | "rlm" | "deterministic" @@ -45,49 +39,85 @@ func (c ExtractConfig) internal() extract.Cfg { AllowedStrategies: c.Strategies} } -// EnableExtract turns on cheap-model extraction with the given model and config. It -// builds the candidate→extraction→reversible-splice adapter and registers it; the -// Extract stage then runs after Reduce. Fail-open per candidate. Settings that name -// component selection (ExtractMode, ExtractStrategies) override cfg when set, so a -// config file fully drives the run. +// Extractor is the cheap-model extraction compactor. It does its OWN detection of large +// structured tool outputs in the conversation (independent of the reduce pass), projects +// each to a smaller faithful subset via the model, and splices it back with a reversible +// recovery marker. Fail-open per candidate. Implements Compactor. +type Extractor struct { + spec ModelSpec + icfg extract.Cfg + cache *extract.Cache +} + +// NewExtractor builds the extract compactor with the given model spec + config. +func NewExtractor(spec ModelSpec, cfg ExtractConfig) *Extractor { + return &Extractor{spec: spec, icfg: cfg.internal(), cache: extract.NewCache()} +} + +func (*Extractor) Name() string { return extractName } +func (*Extractor) Enabled(c *Context) bool { return c.Settings.ExtractEnabled } + +func (x *Extractor) Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) { + model := x.spec.Resolve(c) + if model == nil { + return req, nil // no model available (e.g. incoming creds missing) — fail-open + } + opts := reduce.DefaultOpts() + opts.ProtectRecent = c.Settings.ProtectRecent + opts.ProtectRecentToolUses = c.Settings.ProtectRecentToolUses + opts.ProvableOnly = c.Settings.ProvableOnly + opts.ContextLimit = c.Settings.ContextLimit + opts.ReduceCachedPrefix = c.Settings.ReduceCachedPrefix + opts.CacheFloor = c.ClientCacheFloor + opts.LLMCompactFloor = c.Settings.LLMCompactFloor + opts.LLMCompactStructuredOnly = c.Settings.LLMCompactStructuredOnly + + cands := reduce.SelectLLMCandidates(req, opts) + agg.Candidates = cands + goal := recentGoalText(req, goalTurns) + keep := extract.HarvestIdentifiers(goal, 60) + for _, cand := range cands { + func() { + defer func() { _ = recover() }() // fail-open per candidate + key := goalKey(extract.ContentKey(cand.Text), goal, keep) + result, ok := x.cache.Get(key) + if !ok { + var strat string + result, strat = extract.RunExtraction(c.GoCtx, cand.Text, goal, keep, cand.TokenEst, x.icfg, model) + if strat == "none" || result == "" { + return + } + x.cache.Put(key, result) + } + splice(req, cand, result, c.Store) + }() + } + return req, nil +} + +// EnableExtract registers the extract compactor backed by a static model (config +// source). Settings.ExtractMode / ExtractStrategies override cfg when set, so a config +// file fully drives the run. func (e *Engine) EnableExtract(model Model, cfg ExtractConfig) { + e.EnableExtractSpec(ModelSpec{Static: model}, cfg) +} + +// EnableExtractSpec registers the extract compactor with an explicit ModelSpec — use +// ModelSpec{UseIncoming:true} to reuse the proxied request's own model + credentials. +func (e *Engine) EnableExtractSpec(spec ModelSpec, cfg ExtractConfig) { if e.settings.ExtractMode != "" { cfg.Mode = e.settings.ExtractMode } if len(e.settings.ExtractStrategies) > 0 { cfg.Strategies = e.settings.ExtractStrategies } - icfg := cfg.internal() - cache := extract.NewCache() e.settings.ExtractEnabled = true - e.extract = func(ctx context.Context, req canon.Request, cands []reduce.Candidate) error { - goal := recentGoalText(req, goalTurns) - 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) - if !ok { - var strat string - result, strat = extract.RunExtraction(ctx, c.Text, goal, keep, c.TokenEst, icfg, model) - if strat == "none" || result == "" { - return - } - cache.Put(key, result) - } - e.splice(req, c, result) - }() - } - return nil - } + e.Register(extractName, NewExtractor(spec, cfg)) } // goalKey makes the extraction cache goal-aware: the cache value is the FILTERED -// result, which depends on the goal and keep-set, not just the body. Keying on body -// alone would let the same tool output re-read under a different goal reuse the first -// goal's filtered result. The key composites the body's content key with the goal and -// the keep-set so a different goal is a cache miss. +// result, which depends on the goal and keep-set, not just the body. The key composites +// the body's content key with the goal and the keep-set so a different goal is a miss. func goalKey(contentKey, goal string, keep []string) string { h := sha256.New() h.Write([]byte(contentKey)) @@ -102,12 +132,12 @@ func goalKey(contentKey, goal string, keep []string) string { // splice replaces the candidate block with the extracted result plus a reversible // recovery marker (original stored), only if that is strictly smaller. -func (e *Engine) splice(req canon.Request, c reduce.Candidate, result string) { +func splice(req canon.Request, c reduce.Candidate, result string, st store.Rewind) { block := blockAt(req, c.MsgIndex, c.BlockIndex) if block == nil { return } - rid := e.store.Put(c.Text) + rid := st.Put(c.Text) label := c.FilePath if label == "" { label = c.ToolName diff --git a/engine/live_test.go b/engine/live_test.go new file mode 100644 index 0000000..66a7449 --- /dev/null +++ b/engine/live_test.go @@ -0,0 +1,92 @@ +package engine + +import ( + "context" + "os" + "strings" + "testing" + + "github.com/kagenti/lab-context-engineering/canon" + "github.com/kagenti/lab-context-engineering/config" + "github.com/kagenti/lab-context-engineering/internal/cheapmodel" +) + +// These are opt-in live checks of the summarize/truncate compactors against a real +// claude-haiku-4-5. They skip unless LABCX_LIVE=1 (+ gateway env), so CI never calls out. +// +// LABCX_LIVE=1 CGO_ENABLED=1 go test ./engine -run TestLive -v + +func liveModel(t *testing.T) Model { + t.Helper() + base, key := os.Getenv("ANTHROPIC_BASE_URL"), os.Getenv("ANTHROPIC_AUTH_TOKEN") + if os.Getenv("LABCX_LIVE") != "1" || base == "" || key == "" { + t.Skip("set LABCX_LIVE=1 + ANTHROPIC_BASE_URL + ANTHROPIC_AUTH_TOKEN to run the live check") + } + return cheapmodel.Anthropic{BaseURL: base, APIKey: key, Model: "claude-haiku-4-5", AuthScheme: "bearer"} +} + +func multiTurn(t *testing.T) canon.Request { + t.Helper() + body := []byte(`{"system":"You are a coding agent.","messages":[ + {"role":"user","content":[{"type":"text","text":"Bug: parse_config crashes on empty files. Repo myapp, file src/config.py."}]}, + {"role":"assistant","content":[{"type":"text","text":"I will read src/config.py."}]}, + {"role":"user","content":[{"type":"text","text":"src/config.py defines parse_config(path): it opens the file and calls json.load without handling empty content."}]}, + {"role":"assistant","content":[{"type":"text","text":"I will add an empty-file guard returning {}."}]}, + {"role":"user","content":[{"type":"text","text":"Tests live in tests/test_config.py; add a case for the empty file."}]}, + {"role":"assistant","content":[{"type":"text","text":"Patch drafted; running tests."}]}, + {"role":"user","content":[{"type":"text","text":"All tests pass now. Summarize what changed."}]} + ]}`) + req, err := canon.Decode(body) + if err != nil { + t.Fatalf("decode: %v", err) + } + return req +} + +func TestLiveSummarize(t *testing.T) { + model := liveModel(t) + s := config.Default() + s.Compactors = []string{summarizeName} + s.SummarizeKeepLast = 2 + e := New(s, nil, nil) + e.EnableSummarize(model, DefaultSummarizeConfig()) + + out, _ := e.Transform(context.Background(), multiTurn(t)) + msgs := out.Messages() + if len(msgs) != 3 { // summary note + last 2 + t.Fatalf("want 3 messages (summary + last 2), got %d", len(msgs)) + } + note, _ := msgs[0]["content"].(string) + if !strings.Contains(note, "") { + t.Fatalf("summary note missing : %q", note) + } + body, _ := out.Encode() + ids := FindMarkers(string(body)) + if len(ids) == 0 { + t.Fatal("no recovery marker on summarized request") + } + if _, ok := e.Expand(ids[0]); !ok { + t.Fatal("Expand failed for summarized span") + } + t.Logf("7 messages -> %d (1 summary + last 2); dropped span recoverable via id %s", len(msgs), ids[0]) + t.Logf("LIVE SUMMARY NOTE:\n%s", note) +} + +func TestLiveTruncate(t *testing.T) { + if os.Getenv("LABCX_LIVE") != "1" { + t.Skip("set LABCX_LIVE=1 to run") + } + s := config.Default() + s.Compactors = []string{truncateName} + s.TruncateEnabled = true + s.TruncateKeepLast = 2 + e := New(s, nil, nil) + + out, _ := e.Transform(context.Background(), multiTurn(t)) + msgs := out.Messages() + if len(msgs) != 3 { + t.Fatalf("want 3 messages (note + last 2), got %d", len(msgs)) + } + t.Logf("7 messages -> %d (1 note + last 2)", len(msgs)) + t.Logf("TRUNCATE NOTE:\n%s", msgs[0]["content"].(string)) +} diff --git a/engine/stages.go b/engine/stages.go index f1aeafd..9ab6ef4 100644 --- a/engine/stages.go +++ b/engine/stages.go @@ -1,22 +1,28 @@ package engine import ( + "encoding/json" "fmt" "github.com/kagenti/lab-context-engineering/canon" "github.com/kagenti/lab-context-engineering/internal/cache" + "github.com/kagenti/lab-context-engineering/internal/markers" "github.com/kagenti/lab-context-engineering/internal/reduce" + "github.com/kagenti/lab-context-engineering/internal/tokens" ) -func errFromPanic(r any) error { return fmt.Errorf("stage panic: %v", r) } +func errFromPanic(r any) error { return fmt.Errorf("compactor panic: %v", r) } -// ReduceStage runs the deterministic, lossless-first reduction pass. -type ReduceStage struct{} +// Reducer runs the deterministic, lossless-first reduction pass (collapse / skeleton / +// format, cmdfilter, dedup). No model. Large structured tool outputs are left for the +// extract compactor (which runs first in the default pipeline); whatever it does not +// claim, Reducer reduces losslessly here. +type Reducer struct{} -func (ReduceStage) Name() string { return "reduce" } -func (ReduceStage) Enabled(c *Context) bool { return c.Settings.ReduceEnabled } +func (Reducer) Name() string { return reduceName } +func (Reducer) Enabled(c *Context) bool { return c.Settings.ReduceEnabled } -func (ReduceStage) Run(req canon.Request, agg *Report, c *Context) error { +func (Reducer) Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) { s := c.Settings opts := reduce.DefaultOpts() opts.ProtectRecent = s.ProtectRecent @@ -31,45 +37,23 @@ func (ReduceStage) Run(req canon.Request, agg *Report, c *Context) error { opts.EnabledEncoders = s.Encoders opts.CacheFloor = c.ClientCacheFloor opts.StickyIDs = c.StickyIDs - // Only mark candidates when an extractor is wired; otherwise leave large outputs - // for the deterministic actions to handle. - opts.LLMCompact = c.Extract != nil && s.ExtractEnabled - opts.LLMCompactFloor = s.LLMCompactFloor - opts.LLMCompactStructuredOnly = s.LLMCompactStructuredOnly rep := reduce.ReduceRequest(req, c.Store, c.Evictions, opts) agg.Reduce = rep - agg.Candidates = rep.LLMCandidates - return nil + return req, nil } -// ExtractStage hands the large candidate outputs to the injected cheap-model -// extractor. No-op when no extractor is wired or there are no candidates. -type ExtractStage struct{} +// Cacher injects ephemeral cache_control breakpoints on the stable prefix. Only active +// when the client did not already self-cache (ClientCacheFloor < 0): a self-caching +// client (e.g. Claude Code) keeps its own breakpoints. +type Cacher struct{} -func (ExtractStage) Name() string { return "extract" } -func (ExtractStage) Enabled(c *Context) bool { - return c.Settings.ExtractEnabled && c.Extract != nil -} - -func (ExtractStage) Run(req canon.Request, agg *Report, c *Context) error { - if len(agg.Candidates) == 0 { - return nil - } - return c.Extract(c.GoCtx, req, agg.Candidates) -} - -// CacheStage injects ephemeral cache_control breakpoints on the stable prefix. Only -// active when the client did not already self-cache (ClientCacheFloor < 0), matching -// winnow: a self-caching client (e.g. Claude Code) keeps its own breakpoints. -type CacheStage struct{} - -func (CacheStage) Name() string { return "cache" } -func (CacheStage) Enabled(c *Context) bool { +func (Cacher) Name() string { return cacheName } +func (Cacher) Enabled(c *Context) bool { return c.Settings.CacheEnabled && c.ClientCacheFloor < 0 } -func (CacheStage) Run(req canon.Request, agg *Report, c *Context) error { +func (Cacher) Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) { s := c.Settings anchor := -1 if msgs, ok := req.Root["messages"].([]any); ok { @@ -77,5 +61,42 @@ func (CacheStage) Run(req canon.Request, agg *Report, c *Context) error { } cache.Inject(req.Root, s.CacheBreakpoints, s.CacheStableGap, anchor, s.CacheToolsBreakpoint) agg.CacheInjected = true - return nil + return req, nil +} + +// Truncator is the naive baseline: it keeps the last KeepLast messages and drops the +// older ones, replacing them with one recoverable note. No relevance scoring, no model. +// +// ponytail: naive by design — it does not enforce user/assistant role alternation after +// the drop. It is the no-LLM control to measure smarter compactors against. +type Truncator struct{} + +func (Truncator) Name() string { return truncateName } +func (Truncator) Enabled(c *Context) bool { return c.Settings.TruncateEnabled } + +func (Truncator) Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) { + s := c.Settings + keep := s.TruncateKeepLast + if keep <= 0 { + keep = 3 + } + msgs := req.Messages() + if len(msgs) <= keep { + return req, nil + } + if s.TruncateTriggerTokens > 0 { + if b, err := json.Marshal(msgs); err == nil && tokens.Count(string(b)) < s.TruncateTriggerTokens { + return req, nil + } + } + dropped := msgs[:len(msgs)-keep] + b, _ := json.Marshal(dropped) + rid := c.Store.Put(string(b)) + note := map[string]any{ + "role": "user", + "content": "=== Truncated history ===\n" + + markers.RecoveryNote(fmt.Sprintf("%d earlier message(s)", len(dropped)), "truncated", rid), + } + req.SetMessages(append([]map[string]any{note}, msgs[len(msgs)-keep:]...)) + return req, nil } diff --git a/engine/summarizefunc.go b/engine/summarizefunc.go new file mode 100644 index 0000000..7f046d1 --- /dev/null +++ b/engine/summarizefunc.go @@ -0,0 +1,133 @@ +package engine + +import ( + "encoding/json" + "fmt" + "strings" + + "github.com/kagenti/lab-context-engineering/canon" + "github.com/kagenti/lab-context-engineering/internal/markers" + "github.com/kagenti/lab-context-engineering/internal/summarize" + "github.com/kagenti/lab-context-engineering/internal/tokens" +) + +// SummarizeConfig configures the summarize compactor. +type SummarizeConfig struct { + Level string // concise | regular | highly_detailed + KeepLast int // recent messages kept verbatim after the summary + TriggerTokens int // skip summarizing below this whole-conversation token count +} + +// DefaultSummarizeConfig returns sensible summarizer defaults. +func DefaultSummarizeConfig() SummarizeConfig { + return SummarizeConfig{Level: summarize.LevelRegular, KeepLast: 3} +} + +// Summarizer is the trajectory-summarization compactor (ported from CE-Manager): it +// replaces the older messages with one LLM-produced summary and keeps the last KeepLast +// messages verbatim, storing the dropped span for recovery. Implements Compactor. +type Summarizer struct { + spec ModelSpec + cfg SummarizeConfig +} + +// NewSummarizer builds the summarize compactor with the given model spec + config. +func NewSummarizer(spec ModelSpec, cfg SummarizeConfig) *Summarizer { + if cfg.KeepLast <= 0 { + cfg.KeepLast = 3 + } + if cfg.Level == "" { + cfg.Level = summarize.LevelRegular + } + return &Summarizer{spec: spec, cfg: cfg} +} + +func (*Summarizer) Name() string { return summarizeName } +func (*Summarizer) Enabled(c *Context) bool { return c.Settings.SummarizeEnabled } + +func (x *Summarizer) Compact(req canon.Request, agg *Report, c *Context) (canon.Request, error) { + model := x.spec.Resolve(c) + if model == nil { + return req, nil // no model available — fail-open + } + msgs := req.Messages() + if len(msgs) <= x.cfg.KeepLast { + return req, nil + } + if x.cfg.TriggerTokens > 0 { + if b, err := json.Marshal(msgs); err == nil && tokens.Count(string(b)) < x.cfg.TriggerTokens { + return req, nil + } + } + dropped := msgs[:len(msgs)-x.cfg.KeepLast] + summary, err := summarize.Summarize(c.GoCtx, trajectoryString(dropped), x.cfg.Level, model) + if err != nil { + return req, err + } + b, _ := json.Marshal(dropped) + rid := c.Store.Put(string(b)) + note := map[string]any{ + "role": "user", + "content": "=== History Summary ===\nThe earlier trajectory is summarized below.\n\n" + + summary + "\n\n" + + markers.RecoveryNote(fmt.Sprintf("%d earlier message(s)", len(dropped)), "summarized", rid), + } + req.SetMessages(append([]map[string]any{note}, msgs[len(msgs)-x.cfg.KeepLast:]...)) + return req, nil +} + +// EnableSummarize registers the summarize compactor backed by a static model (config +// source). Settings summarize knobs override cfg when set. +func (e *Engine) EnableSummarize(model Model, cfg SummarizeConfig) { + e.EnableSummarizeSpec(ModelSpec{Static: model}, cfg) +} + +// EnableSummarizeSpec registers the summarize compactor with an explicit ModelSpec — +// use ModelSpec{UseIncoming:true} to reuse the proxied request's own model + creds. +func (e *Engine) EnableSummarizeSpec(spec ModelSpec, cfg SummarizeConfig) { + if e.settings.SummarizeLevel != "" { + cfg.Level = e.settings.SummarizeLevel + } + if e.settings.SummarizeKeepLast > 0 { + cfg.KeepLast = e.settings.SummarizeKeepLast + } + if e.settings.SummarizeTriggerTokens > 0 { + cfg.TriggerTokens = e.settings.SummarizeTriggerTokens + } + e.settings.SummarizeEnabled = true + e.Register(summarizeName, NewSummarizer(spec, cfg)) +} + +// trajectoryString flattens a message list into a "[role]\ncontent" block transcript +// for the summarizer prompt. +func trajectoryString(msgs []map[string]any) string { + var b strings.Builder + for _, m := range msgs { + role, _ := m["role"].(string) + b.WriteString("[" + role + "]\n") + for _, blk := range canon.Blocks(m) { + switch canon.BlockType(blk) { + case "text": + if t, ok := blk["text"].(string); ok { + b.WriteString(t) + } + case "tool_use": + if name, ok := blk["name"].(string); ok { + b.WriteString("(tool_use: " + name + ")") + } + case "tool_result": + switch cc := blk["content"].(type) { + case string: + b.WriteString(cc) + default: + if bb, err := json.Marshal(cc); err == nil { + b.Write(bb) + } + } + } + b.WriteString("\n") + } + b.WriteString("\n") + } + return b.String() +} diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 3066782..9c4127b 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -1,6 +1,6 @@ // Package cache injects Anthropic ephemeral cache_control breakpoints on the stable // prefix of a request. Lossless: nothing is dropped or rewritten, only annotated -// where the provider may cache. Ported from winnow's cache.py. +// where the provider may cache. Ported from the reference prototype's cache.py. // // All functions operate on the canonical Root map (Anthropic shape) in place. package cache diff --git a/internal/extract/contain.go b/internal/extract/contain.go index a9c564b..08d6f9f 100644 --- a/internal/extract/contain.go +++ b/internal/extract/contain.go @@ -3,10 +3,10 @@ // containment validator here PROVES the result is a lossless projection of the // original, so a chatty model can never corrupt a value the agent relies on. On any // failure it falls back to a deterministic projection, then to the original -// (fail-open). Ported from winnow's actions/llm_compact.py. +// (fail-open). Ported from the reference prototype's actions/llm_compact.py. // // ponytail: the model returns the selected JSON directly and containment verifies it — -// no model-generated code (winnow's Python sandbox) and no custom spec language. The +// no model-generated code (the reference prototype's Python sandbox) and no custom spec language. The // containment proof is what makes the mechanism safe, whatever produced the subset. package extract diff --git a/internal/extract/deterministic.go b/internal/extract/deterministic.go index a47dcd2..21b569f 100644 --- a/internal/extract/deterministic.go +++ b/internal/extract/deterministic.go @@ -8,7 +8,7 @@ import ( // Deterministic, model-free projection: filter a parsed value to the parts that match // the keep-set (plus the important-key "spine"). Every leaf it emits is an unchanged // value, a string prefix, or a contiguous window, so its output always passes -// IsContained by construction and is never empty. Ported from winnow's +// IsContained by construction and is never empty. Ported from the reference prototype's // deterministic_project. var importantKeyTokens = []string{"id", "status", "state", "name", "error", "reason", "date", "time"} diff --git a/internal/extract/extract.go b/internal/extract/extract.go index 7549728..660e5d5 100644 --- a/internal/extract/extract.go +++ b/internal/extract/extract.go @@ -35,7 +35,7 @@ type Cfg struct { AllowedStrategies []string } -// DefaultCfg mirrors winnow's ExtractCfg defaults. +// DefaultCfg mirrors the reference prototype's ExtractCfg defaults. func DefaultCfg() Cfg { return Cfg{Mode: "auto", Floor: 3000, AllowDeterministic: true, MaxChars: sampleChars} } diff --git a/internal/extract/prompt.go b/internal/extract/prompt.go index 5c40157..18b33e8 100644 --- a/internal/extract/prompt.go +++ b/internal/extract/prompt.go @@ -8,7 +8,7 @@ import ( // Prompt building. Because the model returns the filtered VALUE (which containment // then verifies), it must SEE the values — so the prompt shows the actual JSON/text // (truncated). For very large lists the RLM strategy chunks the body so each chunk is -// shown in full. The rule set is winnow's "select, never summarize, recall-first" +// shown in full. The rule set is the reference prototype's "select, never summarize, recall-first" // contract, retargeted from "write a function" to "return the JSON". // sampleMarker precedes the body in the prompt; tests and the (future) model both diff --git a/internal/markers/markers.go b/internal/markers/markers.go index 083ed21..6e8169e 100644 --- a/internal/markers/markers.go +++ b/internal/markers/markers.go @@ -1,6 +1,7 @@ -// Package markers handles reversible, namespaced content markers. winnow's marker -// is «winnow:HEXID»; foreign markers from other reducers (headroom, claw) are left -// alone so winnow stacks cleanly on top of them. Ported from winnow's markers.py. +// Package markers handles reversible, namespaced content markers. lab-cx's marker +// is «labcx:HEXID»; foreign markers from other reducers (headroom, claw) are left +// alone so lab-cx stacks cleanly on top of them. Ported from the reference prototype's +// markers.py. package markers import ( @@ -9,23 +10,23 @@ import ( ) var ( - winnowRe = regexp.MustCompile(`«winnow:([0-9a-f]{4,64})»`) + labcxRe = regexp.MustCompile(`«labcx:([0-9a-f]{4,64})»`) foreignRe = regexp.MustCompile(`<>|\[rewind:[0-9a-zA-Z]{4,64}\]`) ) // Make returns the marker for a rewind id. -func Make(rewindID string) string { return fmt.Sprintf("«winnow:%s»", rewindID) } +func Make(rewindID string) string { return fmt.Sprintf("«labcx:%s»", rewindID) } // RecoveryNote is a self-advertising, model-readable note appended to a reduced // block so the omission is a known unknown the model can recover, not a silent drop. func RecoveryNote(label, what, rewindID string) string { - return fmt.Sprintf("[winnow: %s %s; call winnow_expand(%q) to restore] %s", + return fmt.Sprintf("[labcx: %s %s; call labcx_expand(%q) to restore] %s", label, what, rewindID, Make(rewindID)) } -// FindIDs returns all winnow rewind ids referenced in text. +// FindIDs returns all lab-cx rewind ids referenced in text. func FindIDs(text string) []string { - m := winnowRe.FindAllStringSubmatch(text, -1) + m := labcxRe.FindAllStringSubmatch(text, -1) out := make([]string, 0, len(m)) for _, g := range m { out = append(out, g[1]) @@ -33,11 +34,16 @@ func FindIDs(text string) []string { return out } +// Has reports whether text already carries a lab-cx marker (i.e. the block was +// already reduced/extracted on this or a prior turn). Compactors use it to avoid +// re-processing an already-reduced block. +func Has(text string) bool { return labcxRe.MatchString(text) } + // HasForeign reports whether text carries another reducer's marker. func HasForeign(text string) bool { return foreignRe.MatchString(text) } -// Strip removes winnow and foreign markers from text (used for a marker-insensitive +// Strip removes lab-cx and foreign markers from text (used for a marker-insensitive // content key). func Strip(text string) string { - return foreignRe.ReplaceAllString(winnowRe.ReplaceAllString(text, ""), "") + return foreignRe.ReplaceAllString(labcxRe.ReplaceAllString(text, ""), "") } diff --git a/internal/proxyhttp/proxy.go b/internal/proxyhttp/proxy.go index fb85159..6d18404 100644 --- a/internal/proxyhttp/proxy.go +++ b/internal/proxyhttp/proxy.go @@ -41,6 +41,12 @@ type Config struct { // UpstreamTimeout bounds each upstream request when Client is not set; 0 means // no timeout (http.DefaultClient). UpstreamTimeout time.Duration + // BuildRequestModel, when set, builds an engine.Model from the incoming request's + // own model + credentials (surface name, model id, resolved upstream base, request + // headers) for LLM compactors configured with source "incoming". Returning nil means + // "no per-request model" (those compactors then no-op). The host (proxy binary) + // supplies this so cheapmodel construction stays out of this package. + BuildRequestModel func(surfaceName, model, base string, h http.Header) engine.Model } type proxy struct { @@ -76,7 +82,7 @@ func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { case path == "/health" || path == "/ready": p.ok(w, r) - case path == "/winnow/expand": + case path == "/labcx/expand": p.expand(w, r) case path == "/stats": p.stats(w, r) @@ -133,18 +139,32 @@ func (p *proxy) model(surface surfaces.Surface, providerDefault string) http.Han outBody := body if !bypassed(r) { start := time.Now() - reduced, rep, ok := p.reduce(r.Context(), surface, body) + ctx := r.Context() + if p.cfg.BuildRequestModel != nil { + mb := modelBase(base, r.URL.Path, surface.Name()) + if m := p.cfg.BuildRequestModel(surface.Name(), modelOf(body), mb, r.Header); m != nil { + ctx = engine.WithRequestModel(ctx, m) + } + } + reduced, rep, ok := p.reduce(ctx, surface, body) latencyMs := int(time.Since(start).Milliseconds()) if ok { outBody = reduced p.cfg.Emitter.Emit(r.Context(), observability.Event{ System: surface.Name(), Surface: surface.Name(), - RequestModel: modelOf(body), + RequestModel: modelOf(body), SessionID: rep.Reduce.SessionID, TokensBefore: rep.Reduce.TokensBefore, TokensAfter: rep.Reduce.TokensAfter, TokensSaved: rep.Reduce.TokensSaved, Ratio: rep.Reduce.Ratio, CacheInject: rep.CacheInjected, Extracted: len(rep.Candidates) > 0, - StageErrors: rep.StageErrors, - LatencyMillis: latencyMs, + StageErrors: rep.StageErrors, + ToolsTotal: rep.Reduce.ToolsTotal, + ToolDefTokens: rep.Reduce.ToolDefTokens, + ReducedCount: len(rep.Reduce.ReducedIDs), + CandidatesCount: len(rep.Candidates), + FrozenCount: rep.Reduce.FrozenCount, + Rehydrated: rep.Reduce.Rehydrated, + AtCompaction: rep.Reduce.AtCompaction, + LatencyMillis: latencyMs, }) } } @@ -172,6 +192,22 @@ func (p *proxy) reduce(ctx context.Context, surface surfaces.Surface, body []byt return rendered, report, true } +// modelBase returns the base URL a per-request (source: incoming) model client should +// use so that appending the surface's fixed path (/v1/messages or /v1/chat/completions) +// reproduces the SAME upstream URL the main request forwards to — preserving any route +// prefix (e.g. /anthropic when the agent points at http://gateway:4000/anthropic). +func modelBase(upstream, reqPath, surfaceName string) string { + suffix := "/v1/messages" + if surfaceName == "openai" { + suffix = "/v1/chat/completions" + } + prefix := strings.TrimSuffix(reqPath, suffix) + if prefix == reqPath { // suffix not present; fall back to the bare upstream + prefix = "" + } + return strings.TrimRight(upstream, "/") + prefix +} + // modelOf best-effort reads the "model" field from a request body for telemetry. func modelOf(body []byte) string { var m struct { @@ -274,5 +310,5 @@ func (p *proxy) forward(w http.ResponseWriter, r *http.Request, base string, bod } func bypassed(r *http.Request) bool { - return strings.EqualFold(r.Header.Get("x-winnow-bypass"), "true") + return strings.EqualFold(r.Header.Get("x-labcx-bypass"), "true") } diff --git a/internal/proxyhttp/proxy_test.go b/internal/proxyhttp/proxy_test.go index 910243c..03304b6 100644 --- a/internal/proxyhttp/proxy_test.go +++ b/internal/proxyhttp/proxy_test.go @@ -59,7 +59,7 @@ func TestProxyReducesBeforeForwarding(t *testing.T) { t.Fatalf("upstream body not reduced: got %d, original %d", len(upstreamGot), len(body)) } if len(markers.FindIDs(upstreamGot)) == 0 { - t.Fatalf("expected a winnow marker in the forwarded body") + t.Fatalf("expected a lab-cx marker in the forwarded body") } } @@ -71,7 +71,7 @@ func TestBypassForwardsOriginal(t *testing.T) { body := `{"messages":[{"role":"user","content":[{"type":"text","text":"hi"}]}]}` req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body)) - req.Header.Set("x-winnow-bypass", "true") + req.Header.Set("x-labcx-bypass", "true") rec := httptest.NewRecorder() h.ServeHTTP(rec, req) @@ -230,3 +230,19 @@ func TestHealth(t *testing.T) { t.Fatalf("health = %d", rec.Code) } } + +// TestModelBasePreservesPrefix: the per-request (source: incoming) model base must +// reproduce the main request's upstream URL, including any route prefix. +func TestModelBasePreservesPrefix(t *testing.T) { + cases := []struct{ upstream, path, surface, want string }{ + {"http://gateway:4000", "/anthropic/v1/messages", "anthropic", "http://gateway:4000/anthropic"}, + {"http://gateway:4000/", "/v1/messages", "anthropic", "http://gateway:4000"}, + {"https://api.anthropic.com", "/v1/messages", "anthropic", "https://api.anthropic.com"}, + {"http://gateway:4000", "/openai/v1/chat/completions", "openai", "http://gateway:4000/openai"}, + } + for _, c := range cases { + if got := modelBase(c.upstream, c.path, c.surface); got != c.want { + t.Errorf("modelBase(%q,%q,%q) = %q, want %q", c.upstream, c.path, c.surface, got, c.want) + } + } +} diff --git a/internal/reduce/actions.go b/internal/reduce/actions.go index d759adf..0ad547f 100644 --- a/internal/reduce/actions.go +++ b/internal/reduce/actions.go @@ -28,9 +28,9 @@ func collapse(text, filePath, reason string, st store.Rewind, terse bool) (strin label = "tool output" } if terse { - return fmt.Sprintf("[winnow %s: %s] %s", reason, label, marker), rid + return fmt.Sprintf("[labcx %s: %s] %s", reason, label, marker), rid } - return fmt.Sprintf("[winnow: %s omitted (%s); call winnow_expand(%q) to restore] %s", + return fmt.Sprintf("[labcx: %s omitted (%s); call labcx_expand(%q) to restore] %s", label, reason, rid, marker), rid } diff --git a/internal/reduce/candidates.go b/internal/reduce/candidates.go new file mode 100644 index 0000000..fd22ed4 --- /dev/null +++ b/internal/reduce/candidates.go @@ -0,0 +1,63 @@ +package reduce + +import ( + "github.com/kagenti/lab-context-engineering/canon" + "github.com/kagenti/lab-context-engineering/internal/markers" +) + +// SelectLLMCandidates returns the large structured tool outputs eligible for +// cheap-model extraction, WITHOUT mutating req. The extract compactor owns this +// detection so it is fully independent of the reduce pass (no candidate hand-off). +// It mirrors the candidate predicate the reduce loop previously applied: structured, +// not frozen, not protected, above the token floor, and not already marked. +func SelectLLMCandidates(req canon.Request, opts Opts) []Candidate { + msgs := req.Messages() + items := ExtractItems(req) + inputTokens := 0 + for _, it := range items { + inputTokens += it.TokenEst + } + zones := ComputeZones(len(msgs), inputTokens, opts.ContextLimit, 0) + frozen := zones.FrozenCount + if !opts.ReduceCachedPrefix && opts.CacheFloor+1 > frozen { + frozen = opts.CacheFloor + 1 + } + + scoreOpts := DefaultScoreOpts(opts.ProtectRecent) + scoreOpts.CollapseOutputs = opts.CollapseOutputs + scoreOpts.ProtectRecentToolUses = opts.ProtectRecentToolUses + scoreOpts.ProvableOnly = opts.ProvableOnly + verdicts := map[string]Verdict{} + for _, v := range ScoreRelevance(items, scoreOpts) { + verdicts[v.ItemID] = v + } + + var out []Candidate + for _, item := range items { + if item.MsgIndex < frozen || markers.HasForeign(item.Text) || markers.Has(item.Text) { + continue + } + v, ok := verdicts[item.ID] + if !ok { + continue + } + structured := IsStructured(item.Text) + reasonOK := false + switch { + case opts.LLMCompactStructuredOnly && !structured: + reasonOK = false + case structured: + reasonOK = v.Reason == "unused" || v.Reason == "lossy_candidate" || v.Reason == "kept_default" + default: + reasonOK = v.Reason == "unused" || v.Reason == "lossy_candidate" + } + if reasonOK && item.TokenEst >= opts.LLMCompactFloor { + out = append(out, Candidate{ + ID: item.ID, MsgIndex: item.MsgIndex, BlockIndex: item.BlockIndex, + Text: item.Text, FilePath: item.FilePath, ToolName: item.ToolName, + TokenEst: item.TokenEst, + }) + } + } + return out +} diff --git a/internal/reduce/cmdfilter.go b/internal/reduce/cmdfilter.go index e698dae..d36847f 100644 --- a/internal/reduce/cmdfilter.go +++ b/internal/reduce/cmdfilter.go @@ -9,7 +9,7 @@ import ( // Deterministic, LLM-free compaction of KNOWN command outputs (rtk-style): keep the // signal (failures, errors, summary) and drop routine noise. Lossy but reversible — // the caller stores the original and applies only when strictly smaller. Ported from -// winnow's cmdfilter.py. +// the reference prototype's cmdfilter.py. type commandRule struct { name string @@ -125,6 +125,6 @@ func compactCommandOutput(command, output string) (string, bool) { if len(cmd) > 60 { cmd = cmd[:60] } - result += fmt.Sprintf("\n[winnow: %d routine line(s) filtered from `%s`]", dropped, cmd) + result += fmt.Sprintf("\n[labcx: %d routine line(s) filtered from `%s`]", dropped, cmd) return result, true } diff --git a/internal/reduce/compaction.go b/internal/reduce/compaction.go index 42c5a11..046b18b 100644 --- a/internal/reduce/compaction.go +++ b/internal/reduce/compaction.go @@ -10,7 +10,7 @@ import ( // Phrases characterizing a known agent compaction prompt (Claude Code / Codex / // Gemini CLI). A false positive only forgoes one turn's savings. Ported from -// winnow's compaction.py. +// the reference prototype's compaction.py. var compactionPhrases = []string{ "this session is being continued from a previous conversation", "create a detailed summary of the conversation", @@ -61,7 +61,7 @@ func IsCompactionRequest(req canon.Request) bool { return false } -// RehydrateMarkers replaces any winnow-collapsed block with its stored original, in +// RehydrateMarkers replaces any lab-cx-collapsed block with its stored original, in // place. Returns the number of blocks restored. func RehydrateMarkers(req canon.Request, st store.Rewind) int { restored := 0 diff --git a/internal/reduce/pipeline.go b/internal/reduce/pipeline.go index 949c776..540f524 100644 --- a/internal/reduce/pipeline.go +++ b/internal/reduce/pipeline.go @@ -38,7 +38,6 @@ type Report struct { ToolDefTokens int ToolsTotal int ReducedIDs []string - LLMCandidates []Candidate CompactionPassthrough bool Rehydrated int } @@ -50,7 +49,6 @@ type Opts struct { StickyIDs map[string]struct{} CollapseOutputs bool CacheFloor int - LLMCompact bool LLMCompactFloor int LLMCompactStructuredOnly bool RehydrateOnCompaction bool @@ -67,7 +65,7 @@ type Opts struct { EnabledEncoders []string } -// DefaultOpts mirrors winnow's reduce_request defaults. +// DefaultOpts mirrors the reference prototype's reduce_request defaults. func DefaultOpts() Opts { return Opts{ CollapseOutputs: true, CacheFloor: -1, LLMCompactFloor: 3000, @@ -133,7 +131,7 @@ func blockAt(msgs []map[string]any, mi, bi int) map[string]any { } // ReduceRequest reduces req in place and returns a Report. Fail-open: a per-item -// reducer error is counted and leaves that item verbatim. Ported from winnow's +// reducer error is counted and leaves that item verbatim. Ported from the reference prototype's // reduce_request. func ReduceRequest(req canon.Request, st store.Rewind, ev *store.Eviction, opts Opts) Report { msgs := req.Messages() @@ -292,13 +290,14 @@ func ReduceRequest(req canon.Request, st store.Rewind, ev *store.Eviction, opts } } - var llmCandidates []Candidate - for _, item := range items { if _, done := handled[item.ID]; done { continue } - if item.MsgIndex < frozen || markers.HasForeign(item.Text) { + // Skip frozen, foreign-marked, and already-lab-cx-marked blocks. The last guard + // keeps reduce off blocks the extract compactor already rewrote (extract runs + // first in the default pipeline), so they are not double-processed. + if item.MsgIndex < frozen || markers.HasForeign(item.Text) || markers.Has(item.Text) { continue } block := blockAt(msgs, item.MsgIndex, item.BlockIndex) @@ -321,26 +320,6 @@ func ReduceRequest(req canon.Request, st store.Rewind, ev *store.Eviction, opts continue } - // LLM-extraction candidate selection. - structured := IsStructured(item.Text) - reasonOK := false - switch { - case opts.LLMCompactStructuredOnly && !structured: - reasonOK = false - case structured: - reasonOK = v.Reason == "unused" || v.Reason == "lossy_candidate" || v.Reason == "kept_default" - default: - reasonOK = v.Reason == "unused" || v.Reason == "lossy_candidate" - } - if opts.LLMCompact && reasonOK && item.TokenEst >= opts.LLMCompactFloor && len(markers.FindIDs(item.Text)) == 0 { - llmCandidates = append(llmCandidates, Candidate{ - ID: item.ID, MsgIndex: item.MsgIndex, BlockIndex: item.BlockIndex, - Text: item.Text, FilePath: item.FilePath, ToolName: item.ToolName, - TokenEst: item.TokenEst, - }) - continue - } - reduced := route(byID[item.ID], v, st, opts.EnabledReducers, opts.EnabledEncoders) if reduced.NewText != nil { setBlockText(block, *reduced.NewText) @@ -366,6 +345,6 @@ func ReduceRequest(req canon.Request, st store.Rewind, ev *store.Eviction, opts SessionID: sid, AtCompaction: zones.AtCompaction, FrozenCount: frozen, TokensBefore: b, TokensAfter: a, TokensSaved: saved, Ratio: ratio, ReducerErrors: reducerErrors, ToolDefTokens: toolDefTokens, ToolsTotal: len(toolsList), - ReducedIDs: ids, LLMCandidates: llmCandidates, + ReducedIDs: ids, } } diff --git a/internal/reduce/pipeline_test.go b/internal/reduce/pipeline_test.go index a98b646..11fe9f2 100644 --- a/internal/reduce/pipeline_test.go +++ b/internal/reduce/pipeline_test.go @@ -48,7 +48,7 @@ func reduceFixture(t *testing.T, body string, opts Opts) (canon.Request, Report, return req, rep, st } -// markerOriginal returns the stored original for the first winnow marker found in text. +// markerOriginal returns the stored original for the first lab-cx marker found in text. func markerOriginal(t *testing.T, st store.Rewind, text string) (string, bool) { ids := markers.FindIDs(text) if len(ids) == 0 { diff --git a/internal/reduce/relevance.go b/internal/reduce/relevance.go index 8ac7206..208df89 100644 --- a/internal/reduce/relevance.go +++ b/internal/reduce/relevance.go @@ -8,7 +8,7 @@ import ( // Relevance scoring — the core contribution. For each reducible read/tool_result // item, decide whether it is still relevant using deterministic transcript signals. -// Ported from winnow's relevance.py. +// Ported from the reference prototype's relevance.py. var ( salientRe = regexp.MustCompile(`[A-Za-z_][A-Za-z0-9_./-]{7,}`) @@ -26,7 +26,7 @@ type ScoreOpts struct { LiteralSignal bool } -// DefaultScoreOpts mirrors winnow's score_relevance defaults. +// DefaultScoreOpts mirrors the reference prototype's score_relevance defaults. func DefaultScoreOpts(protectRecent int) ScoreOpts { return ScoreOpts{ ProtectRecent: protectRecent, CollapseOutputs: true, diff --git a/internal/reduce/signals.go b/internal/reduce/signals.go index 227234e..792fcfc 100644 --- a/internal/reduce/signals.go +++ b/internal/reduce/signals.go @@ -13,7 +13,7 @@ import ( // Code-reference signals. The question is not "was the basename restated" but: was // the read file's PATH referenced later, or a SYMBOL it DEFINES used later, or a // distinctive LITERAL it contains reused later? Every signal biases toward KEEP (the -// safe direction; reductions are reversible). Ported from winnow's signals/refs.py. +// safe direction; reductions are reversible). Ported from the reference prototype's signals/refs.py. // // definedSymbols uses tree-sitter (go-tree-sitter + per-grammar forest), which catches // methods the old regex missed and ignores commented-out definitions. It still biases diff --git a/internal/reduce/taxonomy.go b/internal/reduce/taxonomy.go index 6f9eaec..ecdce32 100644 --- a/internal/reduce/taxonomy.go +++ b/internal/reduce/taxonomy.go @@ -6,9 +6,9 @@ import ( ) // Tool-name taxonomy covering common coding agents (Claude Code, Cursor, Aider, -// Cline, Continue, Codex). Ported from winnow's taxonomy.py. +// Cline, Continue, Codex). Ported from the reference prototype's taxonomy.py. // -// ponytail: defaults only; WINNOW_TOOLS-style env overrides land with the config +// ponytail: defaults only; LABCX_TOOLS-style env overrides land with the config // package rather than being re-parsed here. var ( readTools = set( diff --git a/internal/reduce/types.go b/internal/reduce/types.go index be958fe..dfe5bb6 100644 --- a/internal/reduce/types.go +++ b/internal/reduce/types.go @@ -1,5 +1,5 @@ // Package reduce is the deterministic, lossless-first reduction core ported from -// winnow: parse a canonical request into items, score each item's relevance from +// the reference prototype: parse a canonical request into items, score each item's relevance from // transcript signals, and apply the cheapest faithful, reversible action. No I/O, // clock, or randomness beyond the injected rewind store. package reduce diff --git a/internal/store/store.go b/internal/store/store.go index 921bcce..cdf43b9 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -5,7 +5,7 @@ // The default Rewind implementation is in-memory — correct for a single proxy or // sidecar process. A Redis/SQLite-backed implementation can satisfy the same // interface later for multi-replica recovery (the plan's deferred option). -// Ported from winnow's rewind.py / session.py. +// Ported from the reference prototype's rewind.py / session.py. package store import ( diff --git a/internal/summarize/summarize.go b/internal/summarize/summarize.go new file mode 100644 index 0000000..d479d64 --- /dev/null +++ b/internal/summarize/summarize.go @@ -0,0 +1,113 @@ +// Package summarize produces a compact, factual summary of an agent trajectory. The +// prompts are ported verbatim from the CE-Manager / ACE "ReSum" summarizer. The +// engine's summarize compactor flattens the conversation to a trajectory string, calls +// Summarize, and rebuilds the message list around the result. +package summarize + +import ( + "context" + "errors" + "strings" +) + +// Model is the LLM client the summarizer calls. The host injects a concrete +// implementation; this core stays transport-free. +type Model interface { + Complete(ctx context.Context, prompt string) (string, error) +} + +// Summary verbosity levels. +const ( + LevelConcise = "concise" + LevelRegular = "regular" + LevelHighlyDetailed = "highly_detailed" +) + +// systemPrompt and userPrompt are verbatim from the ReSum summarizer (prompts/summarizer.py). +const systemPrompt = "You analyze long agent trajectories with tool calls and produce compact, factual summaries. Do not guess or invent information." + +const userPrompt = `You are an expert at analyzing conversation history and extracting relevant information. Your task is +to thoroughly evaluate the conversation history and current question to provide a comprehensive +summary that will help answer the question. + +Task Guidelines: +1. Information Analysis +• Carefully analyze the conversation history to identify truly useful information. +• Focus on information that directly contributes to answering the question. +• Do NOT make assumptions, guesses, or inferences beyond what is explicitly stated in +the conversation. +• If information is missing or unclear, do NOT include it in your summary. + +2. Summary Requirements +• Extract only the most relevant information that is explicitly present in the conversation. +• Synthesize information from multiple exchanges when relevant. Only include information that is certain and clearly stated in the conversation. +• Do NOT output or mention any information that is uncertain, insufficient, or cannot be +confirmed from the conversation. + +3. Output Format Your response should be structured as follows: + +• Essential Information: [Organize the relevant and certain information from the conversation history that helps address the question.] + + +Strictly avoid fabricating, inferring, or exaggerating any information not present in the conversation. +Only output information that is certain and explicitly stated. + +Trajectory: {trajectory} +` + +const ( + suffixConcise = "Please generate a concise summary" + suffixRegular = "Please generate a comprehensive and useful summary" + suffixDetailed = "Please generate a highly detailed, fully comprehensive, explicitly grounded summary " + + "that includes every relevant and certain piece of information from the conversation." +) + +func suffixFor(level string) string { + switch level { + case LevelConcise: + return suffixConcise + case LevelHighlyDetailed: + return suffixDetailed + default: // regular (and unknown) + return suffixRegular + } +} + +// Prompt builds the (system, user) prompt pair for a trajectory at a verbosity level. +func Prompt(trajectory, level string) (system, user string) { + user = strings.ReplaceAll(userPrompt, "{trajectory}", trajectory) + if sfx := suffixFor(level); sfx != "" { + user = user + "\n" + sfx + } + return systemPrompt, user +} + +// Summarize asks the model to summarize the trajectory and returns the result wrapped +// in .... The Model interface takes a single prompt, so the system +// instruction is prepended to the user prompt. +func Summarize(ctx context.Context, trajectory, level string, model Model) (string, error) { + if model == nil { + return "", errors.New("summarize: no model") + } + system, user := Prompt(trajectory, level) + out, err := model.Complete(ctx, system+"\n\n"+user) + if err != nil { + return "", err + } + return EnsureFormat(out), nil +} + +// EnsureFormat guarantees the summary is wrapped in ... tags. +func EnsureFormat(s string) string { + s = strings.TrimSpace(s) + if s == "" { + return "Empty trajectory" + } + if !strings.Contains(s, "") { + s = "\n" + s + } + if !strings.Contains(s, "") { + s = s + "\n" + } + return s +} diff --git a/observability/aggregator.go b/observability/aggregator.go index f119ee9..5ed95db 100644 --- a/observability/aggregator.go +++ b/observability/aggregator.go @@ -21,18 +21,98 @@ type CostRate struct { OutputPerMTok float64 `json:"output_per_mtok"` } -// maxLatencySamples bounds the retained latency samples so memory stays flat over a -// long-running process. The most recent maxLatencySamples are kept (ring buffer). -const maxLatencySamples = 1024 - -// Aggregator is an Emitter that accumulates Events into process-wide reduction stats, -// safe for concurrent use. A host installs it as the proxy Emitter and serves its -// Snapshot/WriteJSON on a /stats endpoint. It does not replace a streaming Emitter — -// a host can fan out to both. +const ( + // maxLatencySamples bounds retained latency samples (per scope) so memory stays + // flat over a long-running process — most recent samples kept (ring buffer). + maxLatencySamples = 1024 + // maxRecentCalls bounds the retained per-call records served at /stats. + maxRecentCalls = 256 + // maxSessions bounds the per-session breakdown; the least-recently-seen session is + // evicted past this. + maxSessions = 512 +) + +// LatencyStats summarizes a set of added-latency samples (milliseconds). +type LatencyStats struct { + P50Millis int `json:"p50_ms"` + P95Millis int `json:"p95_ms"` + P99Millis int `json:"p99_ms"` + MaxMillis int `json:"max_ms"` + MeanMillis int `json:"mean_ms"` +} + +// latRing is a bounded ring of latency samples with quantile/aggregate helpers. +type latRing struct { + buf []int + next int + filled int +} + +func newLatRing(n int) latRing { return latRing{buf: make([]int, n)} } + +func (r *latRing) add(v int) { + r.buf[r.next] = v + r.next = (r.next + 1) % len(r.buf) + if r.filled < len(r.buf) { + r.filled++ + } +} + +func (r *latRing) stats() LatencyStats { + if r.filled == 0 { + return LatencyStats{} + } + xs := make([]int, r.filled) + copy(xs, r.buf[:r.filled]) + sort.Ints(xs) + sum := 0 + for _, x := range xs { + sum += x + } + q := func(p float64) int { + idx := int(p*float64(r.filled)+0.999999) - 1 + if idx < 0 { + idx = 0 + } + if idx >= r.filled { + idx = r.filled - 1 + } + return xs[idx] + } + return LatencyStats{ + P50Millis: q(0.50), P95Millis: q(0.95), P99Millis: q(0.99), + MaxMillis: xs[r.filled-1], MeanMillis: sum / r.filled, + } +} + +// sessionAgg accumulates one conversation's metrics. +type sessionAgg struct { + requests int64 + tokensBefore int64 + tokensAfter int64 + tokensSaved int64 + cacheInjected int64 + extracted int64 + stageErrors int64 + reducedTotal int64 + candidates int64 + costSavedUSD float64 + toolsTotal int + toolDefTokens int + lastModel string + lastSurface string + lastSeq int64 + lat latRing +} + +// Aggregator is an Emitter that accumulates Events into process-wide reduction stats, a +// per-session breakdown, and a recent-per-call ring — safe for concurrent use. A host +// installs it as the proxy Emitter and serves its Snapshot/WriteJSON on /stats. type Aggregator struct { rates map[string]CostRate mu sync.Mutex + seq int64 requests int64 tokensBefore int64 tokensAfter int64 @@ -40,26 +120,60 @@ type Aggregator struct { cacheInjected int64 extracted int64 stageErrors int64 + reducedTotal int64 + candidates int64 costSavedUSD float64 + lat latRing - // latency holds a bounded ring of recent added-latency samples (ms). next is the - // next write index; filled tracks how many slots are valid. - latency [maxLatencySamples]int - next int - filled int + sessions map[string]*sessionAgg + recent []CallRecord // ring of recent per-call records + rNext int + rFilled int } // NewAggregator returns an Aggregator pricing savings with rates (keyed by model, plus // an optional DefaultCostKey fallback). A nil rates map disables cost estimation. func NewAggregator(rates map[string]CostRate) *Aggregator { - return &Aggregator{rates: rates} + return &Aggregator{ + rates: rates, + lat: newLatRing(maxLatencySamples), + sessions: make(map[string]*sessionAgg), + recent: make([]CallRecord, maxRecentCalls), + } +} + +// CallRecord is the full per-call metric set retained for /stats (recent_calls). +type CallRecord struct { + Seq int64 `json:"seq"` + System string `json:"system"` + RequestModel string `json:"request_model"` + Surface string `json:"surface"` + SessionID string `json:"session_id"` + TokensBefore int `json:"tokens_before"` + TokensAfter int `json:"tokens_after"` + TokensSaved int `json:"tokens_saved"` + Ratio float64 `json:"reduction_ratio"` + CacheInjected bool `json:"cache_injected"` + Extracted bool `json:"extracted"` + StageErrors int `json:"stage_errors"` + ToolsTotal int `json:"tools_total"` + ToolDefTokens int `json:"tool_def_tokens"` + ReducedCount int `json:"reduced_blocks"` + CandidatesCount int `json:"extract_candidates"` + FrozenCount int `json:"frozen_messages"` + Rehydrated int `json:"rehydrated"` + AtCompaction bool `json:"at_compaction"` + AddedLatencyMs int `json:"added_latency_ms"` } -// Emit folds one Event into the running totals. +// Emit folds one Event into the global totals, its session, and the recent-calls ring. func (a *Aggregator) Emit(_ context.Context, e Event) { a.mu.Lock() defer a.mu.Unlock() + a.seq++ + cost := float64(e.TokensSaved) / 1e6 * a.inputRate(e.RequestModel) + a.requests++ a.tokensBefore += int64(e.TokensBefore) a.tokensAfter += int64(e.TokensAfter) @@ -71,13 +185,70 @@ func (a *Aggregator) Emit(_ context.Context, e Event) { a.extracted++ } a.stageErrors += int64(e.StageErrors) - a.costSavedUSD += float64(e.TokensSaved) / 1e6 * a.inputRate(e.RequestModel) + a.reducedTotal += int64(e.ReducedCount) + a.candidates += int64(e.CandidatesCount) + a.costSavedUSD += cost + a.lat.add(e.LatencyMillis) + + // Per-session. + if e.SessionID != "" { + s := a.sessions[e.SessionID] + if s == nil { + a.evictSessionsLocked() + s = &sessionAgg{lat: newLatRing(maxLatencySamples)} + a.sessions[e.SessionID] = s + } + s.requests++ + s.tokensBefore += int64(e.TokensBefore) + s.tokensAfter += int64(e.TokensAfter) + s.tokensSaved += int64(e.TokensSaved) + if e.CacheInject { + s.cacheInjected++ + } + if e.Extracted { + s.extracted++ + } + s.stageErrors += int64(e.StageErrors) + s.reducedTotal += int64(e.ReducedCount) + s.candidates += int64(e.CandidatesCount) + s.costSavedUSD += cost + s.toolsTotal = e.ToolsTotal + s.toolDefTokens = e.ToolDefTokens + s.lastModel = e.RequestModel + s.lastSurface = e.Surface + s.lastSeq = a.seq + s.lat.add(e.LatencyMillis) + } - a.latency[a.next] = e.LatencyMillis - a.next = (a.next + 1) % maxLatencySamples - if a.filled < maxLatencySamples { - a.filled++ + // Recent per-call ring. + a.recent[a.rNext] = CallRecord{ + Seq: a.seq, System: e.System, RequestModel: e.RequestModel, Surface: e.Surface, + SessionID: e.SessionID, TokensBefore: e.TokensBefore, TokensAfter: e.TokensAfter, + TokensSaved: e.TokensSaved, Ratio: e.Ratio, CacheInjected: e.CacheInject, + Extracted: e.Extracted, StageErrors: e.StageErrors, ToolsTotal: e.ToolsTotal, + ToolDefTokens: e.ToolDefTokens, ReducedCount: e.ReducedCount, + CandidatesCount: e.CandidatesCount, FrozenCount: e.FrozenCount, + Rehydrated: e.Rehydrated, AtCompaction: e.AtCompaction, AddedLatencyMs: e.LatencyMillis, } + a.rNext = (a.rNext + 1) % maxRecentCalls + if a.rFilled < maxRecentCalls { + a.rFilled++ + } +} + +// evictSessionsLocked drops the least-recently-seen session when at capacity. +func (a *Aggregator) evictSessionsLocked() { + if len(a.sessions) < maxSessions { + return + } + var oldestID string + var oldestSeq int64 = 1<<63 - 1 + for id, s := range a.sessions { + if s.lastSeq < oldestSeq { + oldestSeq, oldestID = s.lastSeq, id + } + } + delete(a.sessions, oldestID) } // inputRate resolves the input price for a model, falling back to DefaultCostKey, then 0. @@ -94,63 +265,95 @@ func (a *Aggregator) inputRate(model string) float64 { return 0 } -// Snapshot is a point-in-time, JSON-serializable view of the aggregated stats. +// SessionSnapshot is one conversation's aggregated metrics. +type SessionSnapshot struct { + SessionID string `json:"session_id"` + Requests int64 `json:"requests"` + TokensBefore int64 `json:"tokens_before"` + TokensAfter int64 `json:"tokens_after"` + TokensSaved int64 `json:"tokens_saved"` + Ratio float64 `json:"reduction_ratio"` + CacheInjected int64 `json:"cache_injected"` + Extracted int64 `json:"extracted"` + StageErrors int64 `json:"stage_errors"` + ReducedBlocks int64 `json:"reduced_blocks"` + Candidates int64 `json:"extract_candidates"` + CostSavedUSD float64 `json:"cost_saved_usd"` + ToolsTotal int `json:"tools_total"` + ToolDefTokens int `json:"tool_def_tokens"` + Model string `json:"model"` + Surface string `json:"surface"` + Latency LatencyStats `json:"latency"` +} + +// Snapshot is a point-in-time, JSON-serializable view of the aggregated stats: global +// totals, a per-session breakdown, and the most recent per-call records. type Snapshot struct { - Requests int64 `json:"requests"` - TokensBefore int64 `json:"tokens_before"` - TokensAfter int64 `json:"tokens_after"` - TokensSaved int64 `json:"tokens_saved"` - Ratio float64 `json:"reduction_ratio"` // tokens_saved / tokens_before (higher is more reduction; 0 = no savings) - CacheInjected int64 `json:"cache_injected"` - Extracted int64 `json:"extracted"` - StageErrors int64 `json:"stage_errors"` - CostSavedUSD float64 `json:"cost_saved_usd"` - AddedLatencyP50Millis int `json:"added_latency_p50_ms"` - AddedLatencyP95Millis int `json:"added_latency_p95_ms"` -} - -// Snapshot returns the current totals. + Requests int64 `json:"requests"` + TokensBefore int64 `json:"tokens_before"` + TokensAfter int64 `json:"tokens_after"` + TokensSaved int64 `json:"tokens_saved"` + Ratio float64 `json:"reduction_ratio"` // tokens_saved / tokens_before (0 = no savings) + CacheInjected int64 `json:"cache_injected"` + Extracted int64 `json:"extracted"` + StageErrors int64 `json:"stage_errors"` + ReducedBlocks int64 `json:"reduced_blocks"` + Candidates int64 `json:"extract_candidates"` + CostSavedUSD float64 `json:"cost_saved_usd"` + Sessions int `json:"sessions"` + + // Flat latency fields (kept for compatibility) + full distribution. + AddedLatencyP50Millis int `json:"added_latency_p50_ms"` + AddedLatencyP95Millis int `json:"added_latency_p95_ms"` + Latency LatencyStats `json:"latency"` + + SessionStats []SessionSnapshot `json:"session_stats,omitempty"` + RecentCalls []CallRecord `json:"recent_calls,omitempty"` +} + +// Snapshot returns the current totals plus per-session and recent-call detail. func (a *Aggregator) Snapshot() Snapshot { a.mu.Lock() defer a.mu.Unlock() + lat := a.lat.stats() s := Snapshot{ - Requests: a.requests, - TokensBefore: a.tokensBefore, - TokensAfter: a.tokensAfter, - TokensSaved: a.tokensSaved, - CacheInjected: a.cacheInjected, - Extracted: a.extracted, - StageErrors: a.stageErrors, - CostSavedUSD: a.costSavedUSD, + Requests: a.requests, TokensBefore: a.tokensBefore, TokensAfter: a.tokensAfter, + TokensSaved: a.tokensSaved, CacheInjected: a.cacheInjected, Extracted: a.extracted, + StageErrors: a.stageErrors, ReducedBlocks: a.reducedTotal, Candidates: a.candidates, + CostSavedUSD: a.costSavedUSD, Sessions: len(a.sessions), + AddedLatencyP50Millis: lat.P50Millis, AddedLatencyP95Millis: lat.P95Millis, + Latency: lat, } if a.tokensBefore > 0 { - // Fraction of input tokens removed: 0 = no savings, 0.9 = 90% reduced. s.Ratio = float64(a.tokensSaved) / float64(a.tokensBefore) } - s.AddedLatencyP50Millis = a.percentileLocked(0.50) - s.AddedLatencyP95Millis = a.percentileLocked(0.95) - return s -} -// percentileLocked computes the p-quantile of the retained latency samples. Caller -// holds a.mu. Returns 0 when no samples have been recorded. -func (a *Aggregator) percentileLocked(p float64) int { - if a.filled == 0 { - return 0 - } - xs := make([]int, a.filled) - copy(xs, a.latency[:a.filled]) - sort.Ints(xs) - // Nearest-rank: index = ceil(p*N)-1, clamped. - idx := int(p*float64(a.filled)+0.999999) - 1 - if idx < 0 { - idx = 0 + for id, sa := range a.sessions { + ss := SessionSnapshot{ + SessionID: id, Requests: sa.requests, TokensBefore: sa.tokensBefore, + TokensAfter: sa.tokensAfter, TokensSaved: sa.tokensSaved, + CacheInjected: sa.cacheInjected, Extracted: sa.extracted, StageErrors: sa.stageErrors, + ReducedBlocks: sa.reducedTotal, Candidates: sa.candidates, CostSavedUSD: sa.costSavedUSD, + ToolsTotal: sa.toolsTotal, ToolDefTokens: sa.toolDefTokens, + Model: sa.lastModel, Surface: sa.lastSurface, Latency: sa.lat.stats(), + } + if sa.tokensBefore > 0 { + ss.Ratio = float64(sa.tokensSaved) / float64(sa.tokensBefore) + } + s.SessionStats = append(s.SessionStats, ss) } - if idx >= a.filled { - idx = a.filled - 1 + // Stable order: most recently active session first. + sort.Slice(s.SessionStats, func(i, j int) bool { + return a.sessions[s.SessionStats[i].SessionID].lastSeq > a.sessions[s.SessionStats[j].SessionID].lastSeq + }) + + // Recent calls in chronological order (oldest retained → newest). + for i := 0; i < a.rFilled; i++ { + idx := (a.rNext - a.rFilled + i + maxRecentCalls*2) % maxRecentCalls + s.RecentCalls = append(s.RecentCalls, a.recent[idx]) } - return xs[idx] + return s } // WriteJSON writes the current Snapshot as indented JSON. @@ -169,9 +372,9 @@ func (a *Aggregator) Summary() string { // client (e.g. `lab-cx stats`) that holds only a decoded Snapshot. func SummaryOf(s Snapshot) string { return fmt.Sprintf( - "requests=%d tokens %d→%d saved=%d (reduction=%.1f%%) cost_saved=$%.4f cache=%d extract=%d errors=%d latency p50=%dms p95=%dms", - s.Requests, s.TokensBefore, s.TokensAfter, s.TokensSaved, s.Ratio*100, - s.CostSavedUSD, s.CacheInjected, s.Extracted, s.StageErrors, - s.AddedLatencyP50Millis, s.AddedLatencyP95Millis, + "requests=%d sessions=%d tokens %d→%d saved=%d (reduction=%.1f%%) cost_saved=$%.4f cache=%d extract=%d reduced=%d errors=%d latency p50=%dms p95=%dms p99=%dms max=%dms", + s.Requests, s.Sessions, s.TokensBefore, s.TokensAfter, s.TokensSaved, s.Ratio*100, + s.CostSavedUSD, s.CacheInjected, s.Extracted, s.ReducedBlocks, s.StageErrors, + s.AddedLatencyP50Millis, s.AddedLatencyP95Millis, s.Latency.P99Millis, s.Latency.MaxMillis, ) } diff --git a/observability/aggregator_test.go b/observability/aggregator_test.go index 57c19bc..7a6ce82 100644 --- a/observability/aggregator_test.go +++ b/observability/aggregator_test.go @@ -106,3 +106,43 @@ func TestAggregatorDefaultCostRate(t *testing.T) { t.Fatalf("CostSavedUSD = %v, want ~2.0", s.CostSavedUSD) } } + +// TestAggregatorPerSessionAndRecentCalls verifies the per-session breakdown, the +// recent-call ring, and the richer latency distribution. +func TestAggregatorPerSessionAndRecentCalls(t *testing.T) { + a := NewAggregator(nil) + ctx := context.Background() + // Two sessions, one with two calls. + a.Emit(ctx, Event{SessionID: "s1", RequestModel: "m", Surface: "anthropic", + TokensBefore: 1000, TokensAfter: 600, TokensSaved: 400, ReducedCount: 2, + CandidatesCount: 1, ToolsTotal: 3, ToolDefTokens: 50, LatencyMillis: 10}) + a.Emit(ctx, Event{SessionID: "s1", RequestModel: "m", Surface: "anthropic", + TokensBefore: 500, TokensAfter: 400, TokensSaved: 100, ReducedCount: 1, LatencyMillis: 30}) + a.Emit(ctx, Event{SessionID: "s2", RequestModel: "m", Surface: "openai", + TokensBefore: 200, TokensAfter: 200, TokensSaved: 0, LatencyMillis: 50}) + + s := a.Snapshot() + if s.Requests != 3 || s.Sessions != 2 { + t.Fatalf("requests=%d sessions=%d, want 3/2", s.Requests, s.Sessions) + } + if len(s.RecentCalls) != 3 { + t.Fatalf("recent_calls=%d, want 3", len(s.RecentCalls)) + } + if len(s.SessionStats) != 2 { + t.Fatalf("session_stats=%d, want 2", len(s.SessionStats)) + } + // Find s1 and check its aggregate (2 calls, 500 saved, reduced 3). + var s1 *SessionSnapshot + for i := range s.SessionStats { + if s.SessionStats[i].SessionID == "s1" { + s1 = &s.SessionStats[i] + } + } + if s1 == nil || s1.Requests != 2 || s1.TokensSaved != 500 || s1.ReducedBlocks != 3 { + t.Fatalf("s1 aggregate wrong: %+v", s1) + } + // Latency distribution populated. + if s.Latency.MaxMillis != 50 || s.Latency.P50Millis == 0 { + t.Fatalf("global latency wrong: %+v", s.Latency) + } +} diff --git a/observability/observability.go b/observability/observability.go index 8c14bb1..735dd4c 100644 --- a/observability/observability.go +++ b/observability/observability.go @@ -14,20 +14,35 @@ import ( "log/slog" ) -// Event is one reduction's telemetry. Field comments give the OTel attribute key the -// default emitter uses. +// Event is one reduction's telemetry — every field a single call exposes. Field +// comments give the OTel attribute key the default emitter uses. The Aggregator folds +// these into process-wide totals, a per-session breakdown, and a recent-calls ring, +// all served at /stats. type Event struct { - System string // gen_ai.system (e.g. "anthropic", "openai") - RequestModel string // gen_ai.request.model - Surface string // context_engineering.surface - TokensBefore int // context_engineering.tokens.before - TokensAfter int // context_engineering.tokens.after - TokensSaved int // context_engineering.tokens.saved - Ratio float64 // context_engineering.tokens.ratio - CacheInject bool // context_engineering.cache_injected - Extracted bool // context_engineering.extracted - StageErrors int // context_engineering.stage_errors - LatencyMillis int // context_engineering.added_latency_ms (time the reduce path added) + System string // gen_ai.system (e.g. "anthropic", "openai") + RequestModel string // gen_ai.request.model + Surface string // context_engineering.surface + SessionID string // context_engineering.session_id (stable per conversation) + + TokensBefore int // context_engineering.tokens.before + TokensAfter int // context_engineering.tokens.after + TokensSaved int // context_engineering.tokens.saved + Ratio float64 // context_engineering.tokens.ratio + + CacheInject bool // context_engineering.cache_injected + Extracted bool // context_engineering.extracted + StageErrors int // context_engineering.stage_errors + + // Richer per-call reduction detail (from the engine Report). + ToolsTotal int // context_engineering.tools.total (tool definitions in the request) + ToolDefTokens int // context_engineering.tools.def_tokens (their token cost) + ReducedCount int // context_engineering.reduced_blocks (blocks the deterministic pass reduced) + CandidatesCount int // context_engineering.extract_candidates (large outputs handed to the extractor) + FrozenCount int // context_engineering.frozen_messages (prefix left untouched) + Rehydrated int // context_engineering.rehydrated (markers restored on a compaction turn) + AtCompaction bool // context_engineering.at_compaction + + LatencyMillis int // context_engineering.added_latency_ms (time the reduce path added) } // Emitter records reduction events. Implementations must be safe for concurrent use. @@ -64,6 +79,7 @@ func (s SlogEmitter) Emit(ctx context.Context, e Event) { slog.String("gen_ai.system", e.System), slog.String("gen_ai.request.model", e.RequestModel), slog.String("context_engineering.surface", e.Surface), + slog.String("context_engineering.session_id", e.SessionID), slog.Int("context_engineering.tokens.before", e.TokensBefore), slog.Int("context_engineering.tokens.after", e.TokensAfter), slog.Int("context_engineering.tokens.saved", e.TokensSaved), @@ -71,6 +87,13 @@ func (s SlogEmitter) Emit(ctx context.Context, e Event) { slog.Bool("context_engineering.cache_injected", e.CacheInject), slog.Bool("context_engineering.extracted", e.Extracted), slog.Int("context_engineering.stage_errors", e.StageErrors), + slog.Int("context_engineering.tools.total", e.ToolsTotal), + slog.Int("context_engineering.tools.def_tokens", e.ToolDefTokens), + slog.Int("context_engineering.reduced_blocks", e.ReducedCount), + slog.Int("context_engineering.extract_candidates", e.CandidatesCount), + slog.Int("context_engineering.frozen_messages", e.FrozenCount), + slog.Int("context_engineering.rehydrated", e.Rehydrated), + slog.Bool("context_engineering.at_compaction", e.AtCompaction), slog.Int("context_engineering.added_latency_ms", e.LatencyMillis), ) } diff --git a/surfaces/openai.go b/surfaces/openai.go index f17ecbd..e370424 100644 --- a/surfaces/openai.go +++ b/surfaces/openai.go @@ -11,7 +11,7 @@ import ( // Tool-result reads are the main reduction target, so ToInternal normalizes the // whole request but Render writes back only the reduced tool_result content into the // original OpenAI message list — assistant tool_calls and plain text are left -// structurally intact. Ported from winnow's openai_adapter. +// structurally intact. Ported from the reference prototype's openai_adapter. type OpenAI struct{} func (OpenAI) Name() string { return "openai" }