diff --git a/README.md b/README.md index be84eb9..6f2d4e3 100644 --- a/README.md +++ b/README.md @@ -170,6 +170,34 @@ are in **[docs/components.md](docs/components.md)** and **[docs/results/componen | `smartcrush` | Offload | keeps anchor items of a long JSON array, drops the middle | | `summarize` | Offload (LLM) | compresses the middle of the trajectory into one summary (run alone) | +## Operating modes + +`sync` (the default) compacts inline and the caller waits. One other mode changes that: + +| Mode | The request path | Use it when | +|---|---|---| +| **`sync`** *(default)* | Compacts inline; the caller waits. | You want the savings. | +| **`observe`** | Forwards the request **untouched, byte for byte**, and reports what compaction *would* have saved. | You want to evaluate context-guru on your own traffic without enforcing it. | + +```yaml +mode: observe +``` + +Byte-identity is **structural**: in observe mode the request path never runs the pipeline +at all (and never injects the expand tool), so no code path could alter a forwarded body. +Measured cost to the enforced path: **0.062 ms/req**, against 1,599 ms for `sync` on the +same benchmark. + +Observe-mode numbers are reported under their own `potential_*` / `projected_*` keys that +share no name with an enforced metric, so a hypothetical can never be read as a realized +saving. On identical traffic its projection matches what `sync` actually achieved exactly +(23.06% both sides), and on traffic with nothing to save it correctly projects 0%. + +This is a genuine differentiator, not a port: headroom has no observe/shadow/dry-run mode +at all — its `token` and `cache` modes are both enforcing. + +Details in [docs/how-to/operating-modes.md](docs/how-to/operating-modes.md). + ## Integrate | Option | What | Where | @@ -182,7 +210,8 @@ Details in [docs/integrations.md](docs/integrations.md). ## Docs -- [docs/design.md](docs/design.md) — architecture: component model, fail-open pipeline, store, session, expand loop, metrics. +- [docs/design.md](docs/design.md) — architecture: component model, fail-open pipeline, store, session, expand loop, metrics, operating modes. +- [docs/how-to/operating-modes.md](docs/how-to/operating-modes.md) — sync vs observe: when to use each, and how to read observe's projections. - [docs/components.md](docs/components.md) — every registered component: how it works, live before→after, lossiness, config, best use. - [docs/integrations.md](docs/integrations.md) — proxy gateway vs AuthBridge plugin, with request paths. - [docs/setup.md](docs/setup.md) — setup + a concrete SWE-bench run through the eval-containers gateway. diff --git a/apply/apply.go b/apply/apply.go index 14ce934..194d7b5 100644 --- a/apply/apply.go +++ b/apply/apply.go @@ -112,7 +112,19 @@ func BodyWithModelWindow(ctx context.Context, pipe *components.Pipeline, st stor // cache-awareness when the backend is a prompt-caching provider or the request // already carries cache_control breakpoints; "on" forces it; "off" restores the // legacy compact-everything behavior (correct for confirmed non-caching backends). -func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, provider bschemas.ModelProvider, body []byte, explicitSession string, bypass bool, models components.ModelSpec, window int, cacheMode string) (result []byte, changedBody bool) { +func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, provider bschemas.ModelProvider, body []byte, explicitSession string, bypass bool, models components.ModelSpec, window int, cacheMode string) ([]byte, bool) { + r := BodyOpts(ctx, pipe, st, Opts{ + Provider: provider, Body: body, Session: explicitSession, Bypass: bypass, + Models: models, Window: window, CacheMode: cacheMode, + }) + return r.Body, r.Changed +} + +// BodyOpts is the full entry point: everything BodyFull takes plus the operating mode +// (#31) and the per-session boundary tracker. Hosts that support modes call this; +// BodyFull is the positional shim every other caller keeps using. +func BodyOpts(ctx context.Context, pipe *components.Pipeline, st store.Store, o Opts) (res Result) { + body, provider, bypass := o.Body, o.Provider, o.Bypass // Top-level fail-open backstop: the per-component recover in pipeline.runOne only // covers component code. A panic anywhere else on the rewrite path (normalize, the // sjson splice, rebuildCountChanged, a marshal) must NOT 500 the client — forward @@ -120,13 +132,18 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr // the whole entry point, not just inside components. defer func() { if r := recover(); r != nil { - slog.Error("context-guru: recovered from panic in BodyFull; forwarding original request", "panic", r) - result, changedBody = body, false + slog.Error("context-guru: recovered from panic in BodyOpts; forwarding original request", "panic", r) + res = Result{Body: body} } }() + mode := o.Mode + if mode == "" { + mode = components.ModeSync + } + models := o.Models msgsRaw := gjson.GetBytes(body, "messages") if !msgsRaw.Exists() || !msgsRaw.IsArray() { - return body, false + return Result{Body: body} } // Volatile-tail split, before anything else touches the body. This is a @@ -149,7 +166,7 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr norm, slots := normalize(provider, msgsRaw.Array()) if len(norm) == 0 { - return body, systemSplit // keep the split even with nothing to compact + return Result{Body: body, Changed: systemSplit} // keep the split even with nothing to compact } if debugTraffic { @@ -157,16 +174,27 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr } chat := &bschemas.BifrostChatRequest{Provider: provider, Input: norm} sys, firstUser := systemAndFirstUser(norm) - sessionID := session.Resolve(explicitSession, sys, firstUser) - cacheAware := resolveCacheAware(cacheMode, provider, body) + sessionID := session.Resolve(o.Session, sys, firstUser) + res.Session = sessionID + cacheAware := resolveCacheAware(o.CacheMode, provider, body) maxCachedIdx := -1 if cacheAware && !bypass { // Messages present on the previous turn of this session are already committed // to the provider cache; only the new tail is being cache-written this turn. // Restrict supersession/age offloaders to that tail so they never mutate the // cached prefix. Growth-based (dialect-agnostic; needs no cache_control mapping). - maxCachedIdx = prevLen(st, sessionID) - 1 - defer putLen(st, sessionID, len(norm)) + // + // With a Tracker the boundary is read and this turn's recorded in ONE locked call. + // The legacy path below read it from the store and wrote it back in a `defer`, so + // two concurrent turns of one session raced on it — see the modes package comment. + // Callers without a tracker (library users, /compact) keep the legacy path: same + // numbers, same race, no behavior change for them. + if o.Tracker != nil { + maxCachedIdx = o.Tracker.Turn(sessionID, len(norm)) - 1 + } else { + maxCachedIdx = prevLen(st, sessionID) - 1 + defer putLen(st, sessionID, len(norm)) + } } c := &components.Ctx{ Ctx: ctx, @@ -174,13 +202,14 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr Store: st, Model: models, Bypass: bypass, - CtxWindow: window, + CtxWindow: o.Window, CacheAware: cacheAware, MaxCachedIdx: maxCachedIdx, // Every breakpoint already on the wire — including the ones no component can // see (`system`, `tools`, and the marks our own normalize drops). The // provider's cap of four counts them all (issue #32, defect 2). ExistingBreakpoints: wireBreakpoints(body), + Mode: mode, } // Canonical form of each normalized message BEFORE the pipeline, so a @@ -191,6 +220,7 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr } rr := pipe.Run(chat, c) + res.Run = rr // A component changed the message count (summarize restructures the transcript // to [msg0, , last-K]). Rebuild the messages array preserving each @@ -199,9 +229,11 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr if len(chat.Input) != len(norm) { nb, ok := rebuildCountChanged(body, msgsRaw.Array(), normPre, slots, chat.Input) if !ok && systemSplit { - return body, true // keep the split even when the rebuild declined + res.Body, res.Changed = body, true // keep the split even when the rebuild declined + return res } - return nb, ok || systemSplit + res.Body, res.Changed = nb, ok || systemSplit + return res } out := body @@ -222,14 +254,16 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr } var err error if out, err = sjson.SetBytes(out, s.path, newText); err != nil { - return body, false + res.Body = body + return res } changed = true changes = append(changes, mkChange(s.path, s.preText, newText)) default: // wholeMessage post, err := json.Marshal(chat.Input[i]) if err != nil { - return body, false + res.Body = body + return res } if bytes.Equal(post, s.pre) { continue // unmodified — keep the original bytes verbatim (I1) @@ -253,7 +287,8 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr continue } if out, err = sjson.SetRawBytes(out, s.path, post); err != nil { - return body, false + res.Body = body + return res } changed = true var pm bschemas.ChatMessage @@ -274,7 +309,8 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr "breakpoints", n, "inbound", c.ExistingBreakpoints, "cap", maxWireBreakpoints, "session", c.Session) } - return out, changed + res.Body, res.Changed = out, changed + return res } // resolveCacheAware decides whether cache-aware compaction is active for this diff --git a/apply/opts.go b/apply/opts.go new file mode 100644 index 0000000..ea66396 --- /dev/null +++ b/apply/opts.go @@ -0,0 +1,47 @@ +package apply + +import ( + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/modes" +) + +// Opts is BodyOpts' input: everything the positional BodyFull takes, plus the operating +// mode (#31) and the per-session boundary tracker. +// +// A struct rather than more positional arguments: the parameter list was already at the +// limit of readability, and these fields are set by one host only. +type Opts struct { + Provider bschemas.ModelProvider + Body []byte + // Session is the host-supplied session id ("" => content hash). + Session string + Bypass bool + Models components.ModelSpec + // Window is the model's resolved context window (max input tokens; 0 = unknown). + Window int + // CacheMode is "auto" (default) | "on" | "off" — see resolveCacheAware. + CacheMode string + + // Mode is the operating mode. Empty means components.ModeSync, so a caller that does + // not know about modes gets exactly today's behavior. + Mode components.Mode + // Tracker, when set, owns the per-session cached-prefix boundary. Supplying it also + // removes the concurrent-turn race in the legacy read-then-deferred-write of prevLen. + // nil => the legacy store-backed path, unchanged for library callers and /compact. + Tracker *modes.Tracker +} + +// Result is BodyOpts' output. +type Result struct { + // Body is the body to forward. Always valid: on any trouble it is the input. + Body []byte + // Changed is false when Body is the untouched input. + Changed bool + // Session is the resolved session id (the caller usually cannot compute it: it falls + // back to a content hash of system + first user message). + Session string + // Run is the pipeline's report for this request, nil when the pipeline did not run. + // Observe mode needs it: the run is the ONLY output, since the body is thrown away. + Run *components.RunReport +} diff --git a/cmd/context-guru-proxy/main.go b/cmd/context-guru-proxy/main.go index 0a0d08d..cfc1d23 100644 --- a/cmd/context-guru-proxy/main.go +++ b/cmd/context-guru-proxy/main.go @@ -36,6 +36,7 @@ func main() { anthropic = flag.String("anthropic-upstream", envOr("ANTHROPIC_UPSTREAM", "https://api.anthropic.com"), "Anthropic upstream base URL") bob = flag.String("bob-upstream", envOr("BOB_UPSTREAM", ""), "Bob (BobShell) backend base URL; enables the Bob gateway routes when set (e.g. https://api.us-east.bob.ibm.com)") storeFlag = flag.String("store", envOr("STORE", ""), "override state store: true|false (default: config store.enabled, else on)") + modeFlag = flag.String("mode", envOr("MODE", ""), "operating mode: sync (default) | observe (overrides the config's mode:)") ) flag.Parse() @@ -43,6 +44,13 @@ func main() { if err != nil { log.Fatalf("config: %v", err) } + if *modeFlag != "" { + cfg.Mode = *modeFlag // flag/env wins over the config file when set + } + mode, err := cfg.OperatingMode() + if err != nil { + log.Fatalf("config: %v", err) + } if v, ok := parseBool(*storeFlag); ok { cfg.Store.Enabled = &v // flag/env wins over the config file when set } @@ -67,6 +75,11 @@ func main() { InjectExpand: os.Getenv("INJECT_EXPAND"), // auto (default) | always | never CacheMode: os.Getenv("CACHE_MODE"), // auto (default) | on | off — cache-aware compaction Windows: modelWindows(), // dynamic context-window resolver (fraction triggers) + Mode: mode, // sync (default) | observe — explicit, never inferred + Observe: proxy.ObserveOptions{ + MaxQueue: cfg.Observe.MaxQueue, + Workers: cfg.Observe.Workers, + }, // Per-request /compact override: swap the pipeline (?preset / header) while // keeping this config's component blocks. nil-safe in the handler. @@ -86,7 +99,12 @@ func main() { }, }) - slog.Info("context-guru-proxy listening", "addr", addr, "pipeline", cfg.Pipeline) + defer h.Close() // stop the off-path worker pool cleanly (no-op in sync mode) + if mode == components.ModeObserve { + slog.Warn("context-guru: OBSERVE MODE — requests are forwarded UNMODIFIED; " + + "/stats reports what compaction WOULD have saved under potential_*/projected_* keys") + } + slog.Info("context-guru-proxy listening", "addr", addr, "pipeline", cfg.Pipeline, "mode", mode) if err := http.ListenAndServe(addr, h.Mux()); err != nil { log.Fatal(err) } diff --git a/components/component.go b/components/component.go index 4986d25..d30771c 100644 --- a/components/component.go +++ b/components/component.go @@ -19,6 +19,7 @@ package components import ( "context" + "fmt" "time" "github.com/maximhq/bifrost/core/schemas" @@ -97,6 +98,38 @@ func (m ModelSpec) For(source string) Model { return m.Static } +// Mode is context-guru's operating mode for one request. The host sets it explicitly +// (proxy Options / config `mode:`); it is NEVER inferred. +// +// ModeSync — compact inline; the caller waits and the compacted request is sent. +// The default, byte-identical to pre-mode behavior. +// ModeObserve — the pipeline runs on a copy whose output is discarded. The agent +// receives the untouched original; results land in a strictly separate +// (hypothetical) metric namespace. +// +// An async mode — deferring compaction off the request path — is designed and +// implemented on a separate branch (#31/#35), held back because its measured benefit +// collapsed once the expensive component stopped running on caching backends. Mode is a +// closed set here so an unknown value fails loudly rather than silently meaning sync. +type Mode string + +// The operating modes. See Mode. +const ( + ModeSync Mode = "sync" + ModeObserve Mode = "observe" +) + +// ParseMode validates a configured mode string; empty means sync. +func ParseMode(s string) (Mode, error) { + switch Mode(s) { + case "", ModeSync: + return ModeSync, nil + case ModeObserve: + return ModeObserve, nil + } + return ModeSync, fmt.Errorf("mode must be sync|observe, got %q", s) +} + // Ctx is the per-request runtime handed to every component. type Ctx struct { Ctx context.Context @@ -137,6 +170,18 @@ type Ctx struct { // enough to put 6 on the wire and take a 400 (issue #32). The host fills it from // the raw body; 0 means "unknown, fall back to what you can see". ExistingBreakpoints int + // Mode is the operating mode this request runs under. Components that behave + // differently off the request path read this rather than inferring anything. + Mode Mode +} + +// effMode is Ctx.Mode with the zero value normalized to sync, so a Ctx built by older +// code (or a test) reports the default rather than an empty mode string. +func (c *Ctx) effMode() Mode { + if c == nil || c.Mode == "" { + return ModeSync + } + return c.Mode } // FilterStatsSink records cmdfilter's per-filter/per-family ledger. metrics.Aggregator @@ -188,6 +233,10 @@ type Report struct { // then silently discarded looked identical to one that works, which is how issue // #32 survived two benchmark studies. Discarded int + // Mode is the operating mode the run happened under, stamped by the pipeline from + // Ctx.Mode. Emitters MUST branch on it: an observe-mode report is a HYPOTHETICAL and + // may never be summed into enforced savings. + Mode Mode } // Saved returns non-negative tokens saved by this component. @@ -205,6 +254,8 @@ type RunReport struct { TokensAfter int DurationMs float64 Components []Report + // Mode is the operating mode this run happened under (see Report.Mode). + Mode Mode } // Saved returns the net tokens saved across the run. diff --git a/components/pipeline.go b/components/pipeline.go index 0fc2d22..02d56ac 100644 --- a/components/pipeline.go +++ b/components/pipeline.go @@ -30,7 +30,7 @@ func NewPipeline(comps []Component, e Emitter) *Pipeline { // report. req is mutated; on any per-component failure that component's changes // are rolled back, so the returned request is never worse than the input. func (p *Pipeline) Run(req *schemas.BifrostChatRequest, c *Ctx) *RunReport { - rr := &RunReport{Session: c.Session, TokensBefore: schema.MessagesTokens(req)} + rr := &RunReport{Session: c.Session, TokensBefore: schema.MessagesTokens(req), Mode: c.effMode()} // Hand cmdfilter its per-family ledger sink when the emitter implements one, so no // host has to thread a second field through every Ctx construction site. if c.FilterStats == nil { @@ -68,7 +68,7 @@ func safeEmit(fn func()) { // never-worse guard. It never returns an error — failures are recorded on the // Report and the request is reverted. func (p *Pipeline) runOne(comp Component, req *schemas.BifrostChatRequest, c *Ctx) (rep Report) { - rep = Report{Component: comp.Name()} + rep = Report{Component: comp.Name(), Mode: c.effMode()} before := schema.CloneMessages(req.Input) rep.TokensBefore = tokensOf(before) start := clock() diff --git a/config/config.go b/config/config.go index adf6a78..c49bb33 100644 --- a/config/config.go +++ b/config/config.go @@ -26,6 +26,26 @@ type Config struct { Pipeline []string `yaml:"pipeline"` Components map[string]yaml.Node `yaml:"components"` Store store.Options `yaml:"store"` + // Mode is the operating mode: sync (default) | observe. See #31 and + // docs/how-to/operating-modes.md. Empty = sync, which is byte-identical to the + // behavior before modes existed. + Mode string `yaml:"mode"` + // Observe tunes observe mode's off-path measurement; ignored in sync mode. + Observe ObserveConfig `yaml:"observe"` +} + +// ObserveConfig is the `observe:` block. +type ObserveConfig struct { + // MaxQueue bounds the off-path measurement queue (0 = 256). A full queue drops, + // counted, and never blocks the request path. + MaxQueue int `yaml:"max_queue"` + // Workers is the number of drain goroutines (0 = 1). + Workers int `yaml:"workers"` +} + +// OperatingMode validates and returns the configured mode. +func (c *Config) OperatingMode() (components.Mode, error) { + return components.ParseMode(c.Mode) } // Load reads and parses a YAML config file (strict: unknown keys are rejected). @@ -49,6 +69,9 @@ func LoadBytes(b []byte) (*Config, error) { if err := c.applyPreset(); err != nil { return nil, err } + if _, err := c.OperatingMode(); err != nil { + return nil, fmt.Errorf("config: %w", err) + } return &c, nil } diff --git a/docs/design.md b/docs/design.md index 8f1645d..ee4ca0a 100644 --- a/docs/design.md +++ b/docs/design.md @@ -20,6 +20,7 @@ infrastructure the components sit on. | `expand/` | reversibility: `<>` marker, the `context_guru_expand` tool def, response parsing + continuation | | `store/` | `Store` interface + in-memory TTL+LRU backend (rewind + sticky ids) | | `session/` | resolve the session key (explicit id, else content hash) | +| `modes/` | per-session cached-prefix boundary (`Tracker`) + the bounded off-path worker pool (`Pool`) | | `metrics/` | `Emitter` implementations: `Slog`, `Aggregator` (for `/stats`), `Tee` | | `config/` | strict YAML loader, presets, pipeline builder | | `proxy/` | the standalone/gateway HTTP proxy | @@ -308,6 +309,71 @@ of per-request percentages. It also reports: Fields are only ever **added** to `/stats`; the harbor harnesses parse it, so no field is renamed or removed. +## Operating modes + +Two modes, set explicitly by `mode:` (or `--mode` / `MODE`) and threaded onto +`components.Ctx` as `Ctx.Mode`. Never inferred. `sync` is the default and reproduces +pre-mode behavior byte for byte; a golden test compares the two entry points' output. + +See [Operating modes](how-to/operating-modes.md) for the operator's view. What follows is +the mechanism. + +### Observe + +The request path does **not** run the pipeline, and skips `expand.Inject` too — a tool +declaration is a modification. Byte-identity is therefore structural, not a property of +careful copying: no code path in observe mode can alter a forwarded body. + +The off-path copy runs against `Handler.shadow`, observe's OWN store: as persistent as the +live one and completely disjoint from it. Both halves are load-bearing, and both were found +by comparing observe's projection against sync's actuals rather than by reading the code: + +- **Persistent**, because offloaders freeze a decision and replay it on every later turn — + that replay is where most of the sustained saving lives. Running observe against a + discarded buffer makes it see only the current tail and under-project by ~3x. +- **Disjoint**, because a decision observe made must never be replayable by a real request. + That would be a request modification arriving by the back door. + +Observe also shares the `Tracker`, so the projection is gated by the same cached-prefix +boundary an enforcing mode would use. Without it `MaxCachedIdx` is -1, the tail gate never +fires, and the projection overstates savings by the amount cache-awareness costs (9.5% +projected vs 0.8% enforced, measured). Sharing it is safe off-path: the boundary only ever +grows, so a late job cannot move it backwards. + +Measurements run on `modes.Pool` — one bounded queue plus a fixed set of drain goroutines +owned by the proxy, not a goroutine per request. The shape is headroom's +`BackgroundCompressor`: dedup by key with the pending slot claimed **before** the job is +observable in the queue (atomic against a concurrent enqueue), a full queue that **drops** +and counts rather than blocking, jobs under the pool's context rather than the request's, +and fail-open on every path including a panicking job. `Stop` bounds its wait: cancelling +cannot interrupt an in-flight HTTP call to the cheap model, and nothing waits on a +measurement's result. + +### The cached-prefix boundary + +`modes.Tracker` holds, per session under one lock, how many normalized messages the +previous turn carried — the boundary above which offloaders may mutate +(`Ctx.MaxCachedIdx`). `Turn(session, n)` reads it and records the new value in ONE locked +call. + +That single call is the fix for a real race: the previous implementation read the value +from the TTL store and wrote it back in a `defer`, so two concurrent turns of one session +both read the same length and the second's write-back could land first, leaving a boundary +describing neither turn. A boundary that is too high lets an offloader mutate content the +provider has already cached, which costs a full cache-write of the suffix. Callers without +a tracker (library users, `/compact`) keep the legacy path unchanged. + +The boundary only ever grows: an agent re-sending a shorter transcript must not shrink it, +or cached content falls back into the mutable tail. + +### Fail-open per mode + +- `sync`: `apply` has a top-level recover, the pipeline has a per-component one, and the + proxy backstops the whole pre-forward block. The pristine inbound body is always a valid + fallback. +- `observe`: the forwarded body *is* the input, so there is nothing for a failure to + damage; a panicking observation is contained by the pool and counted. + ## Config & registry One strict YAML struct serves both hosts. `pipeline:` is an ordered name-list (order + diff --git a/docs/how-to/measure-savings.md b/docs/how-to/measure-savings.md index 454267f..c3cb70a 100644 --- a/docs/how-to/measure-savings.md +++ b/docs/how-to/measure-savings.md @@ -16,12 +16,24 @@ The proxy exposes `GET /stats` with in-process savings rollups. Savings are **to | `bounces` | how many offloads were re-served (the count behind `wasted_tokens`) | | `adjusted_saved` | `saved − wasted` — bounce-adjusted, may be negative | | `top_passthrough` | components that ran but never changed a request: dead weight to drop | +| `mode` | the operating mode these numbers came from: `sync` \| `observe` | +| `sync_enforced` | requests whose forwarded body context-guru actually shaped. **0 in observe mode by construction.** | !!! tip "Reading top_passthrough" A component in `top_passthrough` isn't necessarily broken. `cacheinject` always lands there — its savings are provider-side KV-cache hits, invisible to content-token counts. But a content-offloader that never fires is a candidate to drop from your pipeline. +!!! warning "Enforced vs hypothetical" + Everything above is what context-guru **actually did**. In + [observe mode](operating-modes.md#observe-measure-without-enforcing) nothing is applied, + so every savings field above reads zero and the numbers appear instead under + `potential_*` / `projected_*`, alongside an `observe_notice` banner. The two + vocabularies never share a key: a hypothetical cannot be summed into a real saving even + by accident. Two enforced keys stay deliberately real there — `cg_added_ms_avg` (the + actual enforced-path latency, ~0, which is the point) and context-guru's own model + spend, labelled by `observe_llm_notice` as the cost of measuring rather than enforcing. + ## The Emitter interface The pipeline depends only on the `Emitter` interface (`Component(Report)` + `Run(RunReport)`), so it diff --git a/docs/how-to/operating-modes.md b/docs/how-to/operating-modes.md new file mode 100644 index 0000000..2ddbbda --- /dev/null +++ b/docs/how-to/operating-modes.md @@ -0,0 +1,159 @@ +# Operating modes: sync and observe + +context-guru runs in one of two modes. `sync` is the default and reproduces the behavior +that existed before modes did, byte for byte. + +```yaml +mode: sync # sync | observe +observe: + max_queue: 256 + workers: 1 +``` + +Or `--mode` / `MODE=` on the proxy binary, which wins over the config file. + +The mode is always explicit. Nothing infers it from the rest of your configuration, +because the two modes make materially different promises about your requests and a guess +about which one you wanted is not a thing you should have to debug. + +## Which one do I want + +| You want | Mode | +|---|---| +| Savings, and can absorb compaction latency on the request path | `sync` | +| To find out what context-guru *would* save, without it touching anything | `observe` | + +## sync — compact inline + +The request path runs the pipeline and forwards its output. The caller waits. + +That wait is real: measured on Terminal-Bench, **~450 ms per request** in the +configuration where the LLM-based trimmer runs, almost all of it that model call. + +## observe — measure without enforcing + +The agent receives its request **untouched, byte for byte**. The request path does not run +the pipeline at all, and does not inject the expand tool either — injecting a tool +declaration would modify the request, which is the one thing this mode promises never to +do. A copy of the request runs off-path against observe's own state store, disjoint from +the live one, purely to record what compaction *would* have achieved. + +Byte-identity is therefore **structural**, not a property of careful copying: there is no +code path in observe mode that could alter a forwarded body, because the pipeline never +sees it. A test asserts it anyway. + +This is the answer to "will context-guru help *my* traffic" that does not require +enforcing it in production and comparing against history. Neither reference implementation +offers it — headroom has no observe/shadow/dry-run mode at all; its `token` and `cache` +modes are both enforcing, and its only control arm is a 10% output-shaper holdout. + +### How to read observe numbers + +Observe-mode numbers live under their own keys and **never share a key with an enforced +metric**: + +| Key | Means | +|---|---| +| `observe_notice` | The banner. Present whenever hypotheticals are reported. | +| `observe_hypothetical_requests` | Requests observed. | +| `actual_baseline_tokens` | What the agent really sent. Actual, not hypothetical. | +| `projected_optimized_tokens` | What it would have sent under this pipeline. | +| `potential_saved_tokens` | The difference. | +| `potential_savings_pct` | The difference as a percentage. | +| `potential_components` | Per-component hypothetical contributions. | +| `potential_overhead_ms_avg` | What compaction *would* have added per request — measured off-path, so it is what `sync` would cost you, not what `observe` costs you. | + +In observe mode every enforced **savings** aggregate is zero by construction: +`requests`, `tokens_before`, `tokens_after`, `saved_tokens`, `sync_enforced` and the +`components` map. That zero is the machine-readable form of "context-guru did not modify +any request". + +Two enforced-namespace fields are deliberately **not** zero, because they are real +measurements rather than hypotheticals: + +- `cg_added_ms_avg` — the actual latency added to the enforced path, which in observe mode + is ~0 precisely because that path does no pipeline work. Zeroing it would hide the + mode's headline result. +- `llm_calls` / `llm_input_tokens` / `llm_output_tokens` — context-guru's own model spend. + Observe measures off-path, and that measuring costs real money. The spend is not + hypothetical, so it stays where cost tooling already reads it, labelled by + `observe_llm_notice` as the cost of measuring rather than of enforcing. + +A mislabelled hypothetical is worse than no number at all, because it silently inflates a +savings claim. The separation is therefore structural — two physically separate +accumulators with disjoint serialized names — and a test asserts that no enforced savings +aggregate can reach an observe result. + +### Why observe's numbers should match sync's + +The projection is measured under the **same** conditions an enforcing mode would run +under, because that agreement is what validates the mode. Two things are required and +neither is obvious: + +- **The same cached-prefix boundary.** Observe shares the per-session boundary the + enforced path uses. Without it, cache-awareness never gates anything, every message in + the transcript looks compactable, and the projection overstates savings by exactly the + amount cache-awareness costs — measured at 9.5% projected against 0.8% actually + achieved on the same SWE-bench tasks. +- **State that accumulates across turns.** Offloaders *freeze* a decision and replay it on + every later turn, which is where most of the sustained saving comes from. So observe + keeps a store of its own — as persistent as the live one, and completely disjoint from + it. Discarding its state each turn instead makes it see only the current tail and + **under**-project by ~3x. + +The live store stays pristine either way: observe never writes a byte into it, or a later +real request could replay a decision that was never enforced — a request modification +arriving by the back door. Both properties are asserted by tests. + +### What observe cannot tell you + +- **Cache effects are projected, not measured.** The forwarded request is the agent's own, + so the provider's real cache behavior is the *baseline's*, not the compacted one's. + `potential_saved_tokens` is a content-token figure; the cache consequence of actually + enforcing is not measured here. +- **Reversibility is not exercised.** Nothing was offloaded, so no expand bounce can + happen and `wasted_tokens` stays at zero. Under `sync`, some savings do come back as + bounces. Treat observe's projection as an **upper bound** on content savings. +- **Measuring is not free.** If your pipeline includes `extract_llm`, observe mode spends + cheap-model tokens (see `observe_llm_notice`). It costs money and CPU — just not request + latency. + +### Reading the off-path queue counters + +Observations run on a bounded, owned worker pool rather than a goroutine per request: + +- a full queue **drops** and counts it, never blocks the request path — the request has + already been forwarded, so a drop costs a measurement, not correctness; +- enqueue dedups by key, with the pending slot claimed before the job is observable in the + queue, so dedup is atomic against a concurrent enqueue; +- jobs run under the pool's own context, not the request's (which is cancelled the moment + the response is written); +- a panicking observation is contained and counted in `errors`; the worker survives. + +`dropped` is the counter that says *we silently gave up a measurement*, and it is surfaced +deliberately — headroom's dashboard shows only `queued`, which hides exactly that. + +## Switching modes + +Mode is per-process, not per-request: it decides what happens to every request the proxy +handles, and the mode is reported in `/stats` so a consumer never has to guess which +regime produced a number. + +Session state (frozen decisions, stashes) carries across a restart only as far as the +store does — in-memory by default, so a restart starts cold in either mode. + +## An async mode is designed but not shipped + +A third mode — deferring compaction off the request path so subsequent turns use the +result — is implemented and reviewed on a separate branch ([#35][pr35]), and deliberately +held back. Its measured benefit came almost entirely from deferring the LLM-based trimmer, +which no longer runs on prompt-caching backends, so on the primary workload there is +nothing expensive left to defer. It also has to decline on agents that set their own cache +breakpoints, which claude-code does. + +It is held rather than discarded because the hard parts — per-session compaction +generations, a bounded worker pool, and a cache policy that refuses to write a breakpoint +onto a span it is about to replace — survived a hostile review intact. What it needs is a +paired benchmark arm establishing a benefit, not more code. + +[pr35]: https://github.com/rossoctl/context-guru/pull/35 diff --git a/docs/reference/config.md b/docs/reference/config.md index 2372387..e99c9cb 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -15,6 +15,24 @@ The document has four top-level fields (from the `Config` struct in | `pipeline` | `[]string` | Ordered component names — controls **order + enablement**. Overrides the preset's pipeline when present. | | `components:` | map | Each component's typed config block, handed to its constructor verbatim. | | `store` | object | State store options (`enabled`, `ttl_seconds` (default **10000**, sliding), `max_entries`, …). | +| `mode` | string | Operating mode: `sync` (default) \| `observe`. See [Operating modes](../how-to/operating-modes.md). | +| `observe` | object | Observe-mode tuning; ignored in sync mode. | + +### `mode` + +| Value | Behavior | +|---|---| +| `sync` (default) | Compact inline; the caller waits. Byte-identical to the behavior before modes existed. | +| `observe` | Forward the request untouched and report what compaction *would* have saved, under `potential_*` / `projected_*` keys. | + +Always explicit — nothing infers it from the rest of the configuration. + +### `observe` + +| Field | Default | Purpose | +|---|---|---| +| `max_queue` | `256` | Bound on the off-path measurement queue. A full queue **drops** (counted as `dropped`) and never blocks the request path. | +| `workers` | `1` | Drain goroutines. One keeps a single measurement's cheap-model call in flight per process, which keeps that spend and gateway rate limits predictable. | !!! warning "Strict: unknown keys are rejected" The YAML loader runs with `KnownFields(true)`, so a typo'd key fails loudly @@ -29,6 +47,7 @@ components: collapse: { max_tokens: 2000, head_lines: 20, tail_lines: 20 } smartcrush: { min_items: 5, keep_first: 3, keep_last: 2 } store: { ttl_seconds: 10000, max_entries: 1000 } +mode: sync # sync | observe ``` A component registers its constructor + config type via `init()`, so adding one @@ -47,6 +66,7 @@ for every component's config block. | `OPENAI_API_KEY` / `ANTHROPIC_API_KEY` | — | Real key injected on forward (gateway mode); empty = pass client auth through. | | `FORCE_MODEL` | — | Overwrite the request `model` (eval-containers uses `EVAL_MODEL`). | | `--store` / `STORE` | on | Enable/disable the state store; `--store=false` disables offload reversibility. Wins over the file's `store:` block. | +| `--mode` / `MODE` | `sync` | Operating mode: `sync` \| `observe`. Wins over the file's `mode:`. | ## Diagnostics diff --git a/docs/results/observe-mode.md b/docs/results/observe-mode.md new file mode 100644 index 0000000..17c29f8 --- /dev/null +++ b/docs/results/observe-mode.md @@ -0,0 +1,130 @@ +# Results — observe mode + +Live through the harness, `claude-code` agent on `aws/claude-sonnet-5`, `codesmart` +pipeline, cache-aware billed cost (fresh $2/M · cache-read $0.20/M · cache-write $2.50/M · +output $10/M) recomputed from each trial's token tiers. See [REPRODUCE.md](REPRODUCE.md). + +**Scale caveat, stated up front.** 2 SWE-bench tasks and 2 Terminal-Bench tasks per mode at +n=1, plus one real Claude Code session per mode. Enough to answer the two questions observe +mode has to answer — does it add latency, and do its projections mean anything — and +nowhere near enough for a cost or solve-rate claim. The 50-task arms in the other results +pages are the ones to cite for savings. + +## Does observe add latency to the enforced path? + +**No.** + +| | SWE-bench | Terminal-Bench | Live Claude Code session | +|---|---|---|---| +| `sync` added latency / req | 1,599.4 ms | 26.9 ms | 28.964 ms | +| **`observe` added latency / req** | **0.062 ms** | **0.076 ms** | **0.209 ms** | + +Four orders of magnitude on SWE-bench, and it is structural rather than tuned: the request +path never runs the pipeline, so the only cost is copying the body and an enqueue. + +Observe is not *free* in other respects — it moved 75.0 s of compaction off-path on +SWE-bench and spent $0.0779 of cheap-model tokens doing the measuring. It costs money and +CPU, just not request latency, and `observe_llm_notice` labels that spend for what it is. + +## Do observe's projections match what sync actually achieved? + +This is the question that validates the mode. Three independent lines of evidence, in +descending order of strength: + +### 1. Controlled same-traffic comparison — exact agreement + +The same five turns driven through the real handler under each mode: + +``` +sync: before=43445 saved=10020 (23.06%) +observe: baseline=43445 potential=10020 (23.06%) +``` + +Identical. A test pins this, and it fails at ratio 0.33 if observe's own store is removed. + +### 2. Terminal-Bench — correct agreement near zero (negative control) + +| | `sync` | `observe` | +|---|---|---| +| content savings (enforced) | 1.02% | — (0 by construction) | +| projected savings | — | **0%** | +| added latency / req | 26.9 ms | 0.076 ms | +| enforced requests | 60 | **0** | + +On traffic where sync achieves almost nothing, observe correctly projects almost nothing +rather than inventing a headline. This is the more convincing shape of the evidence: a mode +that only ever agreed on high-savings traffic would be much weaker proof that its +projections mean anything. It also correctly reported the overhead sync *would* have added +as 9.1 ms/req — small here because the pipeline made no model calls on these tasks. + +### 3. SWE-bench arms — consistent, but too noisy to confirm independently + +6.40% projected against 0.82% enforced. The gap is **not** explained away: + +- the arms are different agent trajectories (22.5 vs 15.5 mean steps) — observe saw 46 + requests and 492,652 baseline tokens, sync saw 35 and 244,319. These are not the same + conversations. +- observe's projection never pays a bounce. Nothing is offloaded, so no `expand` round trip + can claw savings back and `wasted_tokens` is structurally 0. Under `sync` some savings do + come back. Observe's projection is an **upper bound** on content savings, documented as + one. +- 2 tasks at n=1 cannot separate a real bias from trajectory noise. + +A 50-task paired run is the honest next step for this line specifically. + +### Two bugs this question found + +Answering it honestly was the most valuable thing the benchmark did, because the first +comparison was wrong twice, in opposite directions: + +1. **11x overstatement.** The observe job ran without the session tracker, so its + cached-prefix boundary was unknown, the tail gate never fired, and 50 `extract_llm` + candidates passed where sync allowed 5 — 9.53% projected against 0.82% enforced. A + projection that ignores cache-awareness projects what a *cache-blind* proxy would do and + overstates by exactly what cache-awareness costs. +2. **3x understatement.** Fixing that exposed the opposite error: observe ran against a + discarded buffer and so lost the frozen decisions offloaders replay on every later turn + — where most of the sustained saving lives. + +Both are fixed, both have a test that fails without its fix, and the exact agreement in §1 +is the result. + +## Namespace separation, verified in production + +From the SWE-bench observe arm's live `/stats`: + +- **enforced:** `requests: 0`, `saved_tokens: 0`, `sync_enforced: 0`, `components: {}` — + all zero, all empty; +- **hypothetical:** `observe_hypothetical_requests: 46`, + `actual_baseline_tokens: 492652`, `projected_optimized_tokens: 461112`, + `potential_saved_tokens: 31540`, `potential_components: {…}` — fully populated. + +No aggregate over the enforced savings rollups can reach a hypothetical, because they are +different accumulators with disjoint serialized names. + +## Live Claude Code sessions + +Same prompt and workspace through each mode against the live gateway: + +| | `sync` | `observe` | +|---|---|---| +| requests (enforced) | 4 | **0** | +| `sync_enforced` | 4 | 0 | +| added latency / req | 28.964 ms | **0.209 ms** | +| baseline tokens | 6,025 | 6,025 *(as `actual_baseline_tokens`)* | +| task answered correctly | yes | yes | + +Observe's `actual_baseline_tokens` = 6,025 is *exactly* sync's `tokens_before` = 6,025 on +the same prompt — the hypothetical namespace accounts for identical traffic identically, +measured independently. And `requests: 0` with every enforced savings aggregate at zero is +the machine-readable form of "context-guru did not modify anything". + +## What is not established here + +- Any cost or solve-rate claim per mode. 2 tasks at n=1; the billed-cost figures track + trajectory length far more than they track mode. +- Whether observe's projection matches sync on *large-savings* traffic at scale. §1 shows + exact agreement on controlled traffic and §2 correct agreement near zero; the SWE-bench + arms are too small and too differently-shaped to confirm the middle of that range. +- The off-path queue under pressure: `dropped` was 0 on every arm, so that path is + exercised only by tests, never yet by production load. diff --git a/metrics/metrics.go b/metrics/metrics.go index 355b446..a86d104 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -112,6 +112,16 @@ type Aggregator struct { sseTTFBMsBuf float64 sseStreamed int64 sseBuffered int64 + // Mode dimension (#31). Enforced requests are counted per mode; observe-mode results + // are kept in PHYSICALLY separate fields with their own serialized names, so no query + // over the enforced rollups can accidentally include a hypothetical. + mode components.Mode // the configured mode, for the /stats banner + syncRequests int64 + potentialRuns int64 + potentialBefore int64 + potentialAfter int64 + potentialMs float64 + potentialComp map[string]*compStat // cmdfilter's per-family / per-filter ledger, plus the selector-miss ledger that // makes the next filter to write data instead of guesswork. filterFam map[string]*filterStat @@ -174,6 +184,15 @@ func (a *Aggregator) FilterMiss(selector string) { a.filterMiss[selector]++ } +// SetMode records the configured operating mode so /stats can label itself — the +// observe-mode banner has to be unmistakable, and a consumer needs to know from the +// payload alone whether the numbers were enforced. +func (a *Aggregator) SetMode(m components.Mode) { + a.mu.Lock() + a.mode = m + a.mu.Unlock() +} + type compStat struct { Runs int64 `json:"runs"` Acted int64 `json:"acted"` // runs that actually saved tokens @@ -201,6 +220,15 @@ func NewAggregator() *Aggregator { return &Aggregator{perComp: map[string]*compS func (a *Aggregator) Component(r components.Report) { a.mu.Lock() defer a.mu.Unlock() + // Observe mode is a HYPOTHETICAL: nothing was applied to any request. Its numbers live + // in a physically separate map with its own vocabulary, so no aggregate over the + // enforced rollups can reach them. Mixing them would silently inflate the product's + // headline savings claim, which is why this is a correctness boundary rather than a + // presentation choice (#31). + if r.Mode == components.ModeObserve { + a.observeComp(r) + return + } cs := a.perComp[r.Component] if cs == nil { cs = &compStat{} @@ -250,6 +278,50 @@ func (a *Aggregator) Component(r components.Report) { } } +// observeComp accumulates one observe-mode component report into the hypothetical +// namespace. Caller holds the lock. +func (a *Aggregator) observeComp(r components.Report) { + if a.potentialComp == nil { + a.potentialComp = map[string]*compStat{} + } + cs := a.potentialComp[r.Component] + if cs == nil { + cs = &compStat{} + a.potentialComp[r.Component] = cs + } + cs.Runs++ + cs.Saved += int64(r.Saved()) + cs.DurationMs += r.DurationMs + if saved := int64(r.Saved()); saved > 0 && !r.Reverted && !r.Skipped { + if cs.seenKeys == nil { + cs.seenKeys = map[string]struct{}{} + } + if len(r.CacheKeys) == 0 { + cs.SavedUnique += saved + } else { + newKeys := 0 + for _, k := range r.CacheKeys { + if _, seen := cs.seenKeys[k]; !seen { + cs.seenKeys[k] = struct{}{} + newKeys++ + } + } + if newKeys > 0 { + cs.SavedUnique += saved * int64(newKeys) / int64(len(r.CacheKeys)) + } + } + } + if r.Reverted { + cs.Reverted++ + } + if !r.Reverted && !r.Skipped { + cs.Mutated++ + } + if r.Saved() > 0 && !r.Reverted && !r.Skipped { + cs.Acted++ + } +} + // RecordExpand notes that `tokens` of previously-offloaded content had to be // re-served (the model called expand). This is the bounce signal: it means an // offload was premature, so the honest savings figure subtracts it (lean-ctx's @@ -302,9 +374,19 @@ func (a *Aggregator) RecordSSE(ttfbMs float64, buffered bool) { func (a *Aggregator) Run(r components.RunReport) { a.mu.Lock() defer a.mu.Unlock() + // Observe: hypothetical. Separate counters, separate JSON keys (potential_* / + // projected_*), never added to requests/before/after. + if r.Mode == components.ModeObserve { + a.potentialRuns++ + a.potentialBefore += int64(r.TokensBefore) + a.potentialAfter += int64(r.TokensAfter) + a.potentialMs += r.DurationMs + return + } a.requests++ a.before += int64(r.TokensBefore) a.after += int64(r.TokensAfter) + a.syncRequests++ } // Snapshot is the JSON served at /stats. It reports both gross savings and the @@ -375,6 +457,40 @@ type Snapshot struct { CmdfilterFamilies map[string]filterStat `json:"cmdfilter_families,omitempty"` CmdfilterFilters map[string]filterStat `json:"cmdfilter_filters,omitempty"` CmdfilterMisses []SelectorMiss `json:"cmdfilter_selector_misses,omitempty"` + + // --- Operating mode (#31). Everything below is ADDITIVE: no existing key was renamed + // or removed, because deploy/harbor/*.py parses this payload. --- + + // Mode is the configured operating mode ("sync" | "observe"). + Mode string `json:"mode"` + // SyncEnforced counts requests whose forwarded body context-guru actually shaped. In + // observe mode it is 0 BY CONSTRUCTION — that is the machine-readable form of + // "context-guru did not modify requests". + SyncEnforced int64 `json:"sync_enforced"` + + // Observe mode: HYPOTHETICALS. Distinct keys (potential_* / projected_*) that never + // share a name with an enforced metric, so a consumer cannot sum a hypothetical into a + // real saving even by accident. All zero outside observe mode. + ObserveNotice string `json:"observe_notice,omitempty"` + ObserveRequests int64 `json:"observe_hypothetical_requests"` + ActualBaselineTokens int64 `json:"actual_baseline_tokens"` // what the agent really sent + ProjectedOptimizedTokens int64 `json:"projected_optimized_tokens"` // what it would have sent + PotentialSavedTokens int64 `json:"potential_saved_tokens"` + PotentialSavingsPct float64 `json:"potential_savings_pct"` + PotentialComponents map[string]compStat `json:"potential_components,omitempty"` + // PotentialOverheadMsAvg is the mean wall time a compaction WOULD have added to each + // request had this mode been enforcing — measured off-path, so it is what sync would + // cost, not what observe costs. + PotentialOverheadMsAvg float64 `json:"potential_overhead_ms_avg"` + // ObserveLLMNotice warns that context-guru's own model spend (llm_calls / + // llm_input_tokens / llm_output_tokens, which feed cg_llm_cost in the harnesses) is + // OFF-PATH measurement cost in observe mode, not the cost of an enforced compaction. + // The tokens were really spent — the number is not hypothetical and must not be moved + // into the potential_* namespace — but attributing it to enforcement would be wrong. + // cg_added_ms_avg is likewise a real measurement of the enforced path, and in observe + // mode it correctly reads ~0 because that path does no pipeline work. Zeroing either + // would hide a true number rather than protect anyone. + ObserveLLMNotice string `json:"observe_llm_notice,omitempty"` } // SelectorMiss is one output shape that matched no filter, with how often it appeared. @@ -383,6 +499,18 @@ type SelectorMiss struct { Count int64 `json:"count"` } +// observeNotice is the machine- and human-readable banner: in observe mode nothing was +// applied, and every number prefixed potential_/projected_ is a hypothetical. +const observeNotice = "OBSERVE MODE: context-guru did not modify any request. " + + "Every potential_*/projected_* field is a hypothetical, not a realized saving." + +// observeLLMNotice covers the one place observe legitimately writes an enforced-namespace +// key: its own model spend is real money, so it stays where cost tooling already reads it, +// labelled for what it is. +const observeLLMNotice = "In observe mode llm_calls/llm_input_tokens/llm_output_tokens " + + "are the cost of MEASURING off-path, not of enforcing a compaction. The spend is real " + + "(not hypothetical); it simply bought a projection rather than a saving." + // Snapshot returns a point-in-time copy of the rollups. func (a *Aggregator) Snapshot() Snapshot { a.mu.Lock() @@ -433,7 +561,11 @@ func (a *Aggregator) Snapshot() Snapshot { if n := a.sseStreamed + a.sseBuffered; n > 0 { bufPct = float64(a.sseBuffered) / float64(n) * 100 } - return Snapshot{ + mode := a.mode + if mode == "" { + mode = components.ModeSync + } + snap := Snapshot{ Requests: a.requests, TokensBefore: a.before, TokensAfter: a.after, SavedTokens: saved, SavingsPct: pct, WastedTokens: a.wasted, Bounces: a.bounces, AdjustedSaved: saved - a.wasted, @@ -444,7 +576,36 @@ func (a *Aggregator) Snapshot() Snapshot { CmdfilterFamilies: copyFilterStats(a.filterFam), CmdfilterFilters: copyFilterStats(a.filterName), CmdfilterMisses: topMisses(a.filterMiss, 20), + Mode: string(mode), + SyncEnforced: a.syncRequests, + } + if a.potentialRuns > 0 || mode == components.ModeObserve { + snap.ObserveNotice = observeNotice + snap.ObserveLLMNotice = observeLLMNotice + snap.ObserveRequests = a.potentialRuns + snap.ActualBaselineTokens = a.potentialBefore + snap.ProjectedOptimizedTokens = a.potentialAfter + snap.PotentialSavedTokens = a.potentialBefore - a.potentialAfter + if a.potentialBefore > 0 { + snap.PotentialSavingsPct = float64(a.potentialBefore-a.potentialAfter) / float64(a.potentialBefore) * 100 + } + if a.potentialRuns > 0 { + snap.PotentialOverheadMsAvg = a.potentialMs / float64(a.potentialRuns) + } + if len(a.potentialComp) > 0 { + pc := make(map[string]compStat, len(a.potentialComp)) + for k, v := range a.potentialComp { + cs := *v + if cs.SavedUnique > 0 { + cs.OvercountRatio = float64(cs.Saved) / float64(cs.SavedUnique) + } + cs.seenKeys = nil + pc[k] = cs + } + snap.PotentialComponents = pc + } } + return snap } func copyFilterStats(src map[string]*filterStat) map[string]filterStat { diff --git a/mkdocs.yml b/mkdocs.yml index dd77d02..ec85a19 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -133,6 +133,7 @@ nav: - How-to Guides: - Use with Claude Code: how-to/use-with-claude-code.md - Choose a preset: how-to/choose-a-preset.md + - Operating modes (sync/observe): how-to/operating-modes.md - Run behind a proxy or gateway: integrations.md - Integrate as a bifrost plugin: how-to/bifrost-plugin.md - Write a custom DSL filter: how-to/custom-dsl-filter.md @@ -151,6 +152,7 @@ nav: - "Results: context-guru": results/context-guru.md - "Results: headroom": results/headroom.md - "Results: rtk": results/rtk.md + - "Results: observe mode": results/observe-mode.md - Reproduce the results: results/REPRODUCE.md - Reference: - Routes & headers: reference/routes.md diff --git a/modes/modes_test.go b/modes/modes_test.go new file mode 100644 index 0000000..a60db0e --- /dev/null +++ b/modes/modes_test.go @@ -0,0 +1,260 @@ +package modes + +import ( + "context" + "runtime" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" +) + +// --- Tracker ---------------------------------------------------------------- + +func TestTurnReturnsPreviousLength(t *testing.T) { + tr := NewTracker(0) + if pl := tr.Turn("s", 5); pl != 0 { + t.Fatalf("first turn: got %d, want 0", pl) + } + if pl := tr.Turn("s", 9); pl != 5 { + t.Fatalf("second turn: got %d, want 5", pl) + } + // A shorter turn must not shrink the boundary: content the provider already cached + // would otherwise fall back into the mutable tail. + if pl := tr.Turn("s", 3); pl != 9 { + t.Fatalf("shorter turn moved the boundary: got %d, want 9", pl) + } + if pl := tr.Turn("s", 12); pl != 9 { + t.Fatalf("boundary not preserved: got %d, want 9", pl) + } +} + +func TestSessionsAreIsolated(t *testing.T) { + tr := NewTracker(0) + tr.Turn("a", 7) + if pl := tr.Turn("b", 2); pl != 0 { + t.Fatalf("session b saw session a's length: %d", pl) + } +} + +// TestConcurrentTurnsDoNotCorruptState is the race this type exists to remove: the +// previous implementation read prevLen from the store and wrote it back in a `defer`, so +// two concurrent turns of one session could both read the same value and the second's +// write-back could land first, leaving a boundary describing neither turn. Every observed +// value must be a length some turn really carried, and the final boundary the largest. +// Run under -race. +func TestConcurrentTurnsDoNotCorruptState(t *testing.T) { + tr := NewTracker(0) + const n = 64 + + seen := make([]int, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + seen[i] = tr.Turn("s", i+1) + }(i) + } + wg.Wait() + + for i, pl := range seen { + if pl < 0 || pl > n { + t.Fatalf("turn %d observed an impossible prevLen %d", i, pl) + } + } + if pl := tr.Turn("s", 0); pl != n { + t.Fatalf("final boundary is %d, want %d (a concurrent write was lost)", pl, n) + } +} + +// The tracker's session cap is its eviction policy (there is no session-end signal on +// this wire — an agent simply stops sending), so the cap must hold under an unbounded +// stream of distinct sessions. +func TestTrackerStaysBounded(t *testing.T) { + small := NewTracker(2) + for i := 0; i < 20; i++ { + small.Turn("s"+strconv.Itoa(i), 1) + } + if n := small.Sessions(); n > 2 { + t.Fatalf("tracker exceeded its bound: %d sessions", n) + } +} + +// --- Pool ------------------------------------------------------------------- + +func TestPoolRunsJobs(t *testing.T) { + p := NewPool(0, 0) + defer p.Stop() + done := make(chan struct{}) + if !p.Enqueue("k", func(context.Context) { close(done) }) { + t.Fatal("enqueue refused") + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("job never ran") + } + waitFor(t, func() bool { return p.Stats().Processed == 1 }) +} + +// TestEnqueueDedupIsAtomic hammers one key from many goroutines while the worker is +// blocked. Exactly one may be accepted: the pending slot is claimed before the job is +// observable in the queue, so a concurrent enqueue cannot slip past the check. +func TestEnqueueDedupIsAtomic(t *testing.T) { + p := NewPool(0, 1) + defer p.Stop() + + release := make(chan struct{}) + var ran atomic.Int64 + block := func(context.Context) { ran.Add(1); <-release } + + var accepted atomic.Int64 + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if p.Enqueue("same-key", block) { + accepted.Add(1) + } + }() + } + wg.Wait() + if n := accepted.Load(); n != 1 { + t.Fatalf("dedup admitted %d jobs for one key, want 1", n) + } + close(release) + waitFor(t, func() bool { return p.Stats().Pending == 0 }) + if n := ran.Load(); n != 1 { + t.Fatalf("job body ran %d times, want 1", n) + } +} + +// TestFullQueueDropsAndNeverBlocks: the request has already been forwarded, so a drop +// costs a measurement only — but it must be counted, and Enqueue must not block. +func TestFullQueueDropsAndNeverBlocks(t *testing.T) { + p := NewPool(2, 1) + defer p.Stop() + + release := make(chan struct{}) + defer close(release) + p.Enqueue("busy", func(context.Context) { <-release }) // occupy the single worker + waitFor(t, func() bool { return p.Stats().Pending == 1 }) + + noop := func(context.Context) {} + accepted, dropped := 0, 0 + deadline := time.After(5 * time.Second) + for i := 0; i < 50; i++ { + ok := make(chan bool, 1) + go func(i int) { ok <- p.Enqueue(strconv.Itoa(i), noop) }(i) + select { + case v := <-ok: + if v { + accepted++ + } else { + dropped++ + } + case <-deadline: + t.Fatal("Enqueue blocked on a full queue") + } + } + if dropped == 0 { + t.Fatal("a full queue accepted everything") + } + if got := p.Stats().Dropped; got != int64(dropped) { + t.Fatalf("dropped counter is %d, want %d", got, dropped) + } + if accepted > 2 { + t.Fatalf("queue of 2 accepted %d jobs", accepted) + } +} + +func TestStopLeaksNoGoroutines(t *testing.T) { + settle() + before := runtime.NumGoroutine() + + p := NewPool(16, 4) + p.Enqueue("a", func(context.Context) {}) + waitFor(t, func() bool { return p.Stats().Processed >= 1 }) + if !p.Stop() { + t.Fatal("Stop reported an unclean exit with no job running") + } + p.Stop() // idempotent + + settle() + if after := runtime.NumGoroutine(); after > before { + t.Fatalf("goroutine leak: %d before, %d after Stop", before, after) + } + if p.Enqueue("b", func(context.Context) {}) { + t.Fatal("a stopped pool accepted a job") + } +} + +// TestStopDoesNotWaitForASlowJob: cancelling asks a job to stop, but one blocked in an +// HTTP call to the cheap model only notices when that call returns — and its client +// timeout is minutes. Shutdown must not inherit that: the result is a measurement nobody +// waits for, and main defers Close(). +func TestStopDoesNotWaitForASlowJob(t *testing.T) { + p := NewPool(0, 1) + release := make(chan struct{}) + defer close(release) + + started := make(chan struct{}) + p.Enqueue("slow", func(context.Context) { + close(started) + <-release // ignores cancellation, like an in-flight HTTP call + }) + <-started + + done := make(chan bool, 1) + go func() { done <- p.Stop() }() + select { + case clean := <-done: + if clean { + t.Fatal("Stop reported a clean exit while a job was still running") + } + case <-time.After(stopGrace + 3*time.Second): + t.Fatal("Stop blocked past its grace period on an uncancellable job") + } +} + +// TestPanickingJobIsContained: fail-open. Nothing was riding on the job. +func TestPanickingJobIsContained(t *testing.T) { + p := NewPool(0, 1) + defer p.Stop() + p.Enqueue("boom", func(context.Context) { panic("nope") }) + waitFor(t, func() bool { return p.Stats().Errors == 1 }) + + done := make(chan struct{}) + p.Enqueue("after", func(context.Context) { close(done) }) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("the worker died with the panicking job") + } +} + +// --- helpers ---------------------------------------------------------------- + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatal("condition never became true") +} + +// settle gives already-finishing goroutines a chance to exit so a leak check compares +// like with like. +func settle() { + for i := 0; i < 20; i++ { + runtime.Gosched() + time.Sleep(5 * time.Millisecond) + } +} diff --git a/modes/pool.go b/modes/pool.go new file mode 100644 index 0000000..31e2b5c --- /dev/null +++ b/modes/pool.go @@ -0,0 +1,200 @@ +package modes + +import ( + "context" + "log/slog" + "sync" + "time" +) + +// Pool is the bounded off-path worker pool observe mode measures on: one queue, a fixed +// number of drain goroutines, owned by the host rather than spawned per request. +// +// The shape is headroom's BackgroundCompressor, ported: +// +// - a bounded queue that DROPS rather than blocks — the request has already been +// forwarded, so a drop costs a measurement, never correctness, and the request path +// must never wait on this; +// - dedup by key, with the pending slot claimed BEFORE the job becomes observable in the +// queue, so dedup is atomic against a concurrent enqueue of the same key; +// - no request-coupled deadline: jobs run under the pool's own context, not the inbound +// request's, which is cancelled the moment the response is written; +// - fail-open on every path, including a panicking job; +// - the FULL counter tuple exposed, `dropped` included. headroom's dashboard shows only +// `queued`, which hides exactly the counter that says "we silently gave up a +// measurement". +type Pool struct { + q chan job + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + started bool + + mu sync.Mutex + pending map[string]struct{} + processed int64 + dropped int64 + errors int64 +} + +type job struct { + key string + run func(context.Context) +} + +// Stats is the queue's counter tuple, surfaced whole in /stats. +type Stats struct { + Queued int64 `json:"queued"` + Pending int64 `json:"pending"` + Processed int64 `json:"processed"` + Dropped int64 `json:"dropped"` + Errors int64 `json:"errors"` +} + +// Defaults for the pool's two knobs. One worker is deliberate: an observation may make a +// cheap-model call, and one in flight per process keeps that spend and the gateway's rate +// limit predictable while still keeping the work off the request path. +const ( + DefaultMaxQueue = 256 + DefaultWorkers = 1 +) + +// NewPool builds and starts a pool. maxQueue/workers <= 0 take the defaults. Call Stop to +// shut it down; a stopped pool drops every later enqueue. +func NewPool(maxQueue, workers int) *Pool { + if maxQueue <= 0 { + maxQueue = DefaultMaxQueue + } + if workers <= 0 { + workers = DefaultWorkers + } + ctx, cancel := context.WithCancel(context.Background()) + p := &Pool{ + q: make(chan job, maxQueue), + ctx: ctx, + cancel: cancel, + started: true, + pending: map[string]struct{}{}, + } + for i := 0; i < workers; i++ { + p.wg.Add(1) + go p.drain() + } + return p +} + +// Enqueue queues run under key, returning false if it was dropped — because the key is +// already queued or in flight, the queue is full, or the pool is stopped. Never blocks. +func (p *Pool) Enqueue(key string, run func(context.Context)) bool { + if p == nil || run == nil { + return false + } + p.mu.Lock() + if !p.started { + p.mu.Unlock() + return false + } + if _, dup := p.pending[key]; dup { + p.mu.Unlock() + return false + } + // Claim the slot BEFORE the job is observable in the queue, so a concurrent Enqueue of + // the same key cannot slip past the dedup check. + p.pending[key] = struct{}{} + p.mu.Unlock() + + select { + case p.q <- job{key: key, run: run}: + return true + default: + p.mu.Lock() + delete(p.pending, key) + p.dropped++ + p.mu.Unlock() + slog.Warn("context-guru: observe queue full, dropping a measurement (request already forwarded)", + "key", key, "max_queue", cap(p.q)) + return false + } +} + +func (p *Pool) drain() { + defer p.wg.Done() + for { + select { + case <-p.ctx.Done(): + return + case j, ok := <-p.q: + if !ok { + return + } + p.runOne(j) + } + } +} + +func (p *Pool) runOne(j job) { + defer func() { + p.mu.Lock() + delete(p.pending, j.key) + if r := recover(); r != nil { + p.errors++ + p.mu.Unlock() + slog.Error("context-guru: recovered from panic in an off-path observation", "key", j.key, "panic", r) + return + } + p.processed++ + p.mu.Unlock() + }() + j.run(p.ctx) +} + +// Stats returns the counter tuple. +func (p *Pool) Stats() Stats { + if p == nil { + return Stats{} + } + p.mu.Lock() + defer p.mu.Unlock() + return Stats{ + Queued: int64(len(p.q)), + Pending: int64(len(p.pending)), + Processed: p.processed, + Dropped: p.dropped, + Errors: p.errors, + } +} + +// stopGrace bounds how long Stop waits for an in-flight job. Cancelling the context asks a +// job to stop, but one sitting in an HTTP call to the cheap model only notices when that +// call returns, and its client timeout is minutes. Since the job's result is a measurement +// nobody is waiting for, shutdown must not inherit that timeout — main defers Close(). +const stopGrace = 2 * time.Second + +// Stop cancels the pool's context and waits briefly for its workers to exit. Queued jobs +// are abandoned — they were measurements, and the requests they belonged to went out long +// ago. Returns false if a worker was still running at the grace deadline (its goroutine is +// left to exit on its own; nothing depends on its result). Idempotent. +func (p *Pool) Stop() bool { + if p == nil { + return true + } + p.mu.Lock() + if !p.started { + p.mu.Unlock() + return true + } + p.started = false + p.mu.Unlock() + p.cancel() + + done := make(chan struct{}) + go func() { p.wg.Wait(); close(done) }() + select { + case <-done: + return true + case <-time.After(stopGrace): + slog.Warn("context-guru: an observation was still running at shutdown; abandoning it", + "grace", stopGrace) + return false + } +} diff --git a/modes/tracker.go b/modes/tracker.go new file mode 100644 index 0000000..cf5fa85 --- /dev/null +++ b/modes/tracker.go @@ -0,0 +1,73 @@ +// Package modes holds the per-session state context-guru's operating modes need. +// +// Today that is one thing: the cached-prefix boundary, i.e. how many normalized +// messages the previous turn of a session carried. Everything at or below it is already +// committed to the provider's cache, so supersession/age-based offloaders must confine +// their mutations to the tail above it (components.Ctx.MaxCachedIdx). +// +// It lives here rather than in the TTL store because it is turn accounting, not cached +// payload, and because reading it and recording the new value must be ONE atomic step. +// The previous implementation read it from the store and wrote it back in a `defer`, so +// two concurrent turns of one session raced: both read the same length, and the second's +// write-back could land before the first's, leaving the boundary describing neither turn. +// A boundary that is too high lets an offloader mutate content the provider has cached, +// which costs a full cache-write of the suffix. +package modes + +import "sync" + +// Tracker holds the per-session cached-prefix boundary, each session's state guarded by +// one lock so concurrent turns cannot interleave a read and a write. +type Tracker struct { + mu sync.Mutex + m map[string]int + max int // bound on tracked sessions; 0 => default +} + +// defaultMaxSessions bounds the tracker so an unbounded stream of distinct sessions +// cannot grow it without limit. Matches the store's sticky-set bound. +const defaultMaxSessions = 1000 + +// NewTracker returns an empty tracker. maxSessions <= 0 uses the default bound. +func NewTracker(maxSessions int) *Tracker { + if maxSessions <= 0 { + maxSessions = defaultMaxSessions + } + return &Tracker{m: map[string]int{}, max: maxSessions} +} + +// Turn records that this session's current turn carries n normalized messages and +// returns the PREVIOUS turn's count — the cached-prefix boundary the request must be +// built against. Read and write happen under one lock, which is what removes the race +// described in the package comment. +// +// The boundary only ever grows: an agent that re-sends a shorter transcript (a rewind, +// or a smaller second request under the same session id) must not shrink it, or content +// the provider already cached would fall back into the mutable tail. +func (t *Tracker) Turn(session string, n int) (prevLen int) { + t.mu.Lock() + defer t.mu.Unlock() + prevLen, ok := t.m[session] + if !ok && len(t.m) >= t.max { + // ponytail: arbitrary eviction, same policy as the store's sticky sets. A dropped + // session restarts at 0, which means "treat everything as tail" — correct, just + // less saving. Add an LRU only if session churn is shown to cost real savings. + for k := range t.m { + delete(t.m, k) + break + } + } + if n > prevLen { + t.m[session] = n + } else { + t.m[session] = prevLen + } + return prevLen +} + +// Sessions reports how many sessions are tracked (test/telemetry aid). +func (t *Tracker) Sessions() int { + t.mu.Lock() + defer t.mu.Unlock() + return len(t.m) +} diff --git a/proxy/modes.go b/proxy/modes.go new file mode 100644 index 0000000..1fd63db --- /dev/null +++ b/proxy/modes.go @@ -0,0 +1,114 @@ +package proxy + +import ( + "context" + "strconv" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/apply" + "github.com/rossoctl/context-guru/components" +) + +// Operating modes on the request path (#31). +// +// sync — run the pipeline inline and forward its output. Unchanged from before modes +// existed, down to the bytes; this is the default. +// observe — forward the ORIGINAL body, byte for byte, and run the pipeline off-path on a +// copy purely to record what it WOULD have saved. +// +// Byte-identity in observe mode is structural, not a property of careful copying: the +// request path never touches the pipeline at all. Fail-open follows from that too — there +// is nothing for a failure to damage, because the forwarded body is the input. + +// applyMode rewrites body for forwarding according to the handler's mode, and returns the +// body to forward plus the wall time to charge to the request path. Never returns a nil +// body. +func (h *Handler) applyMode(r *reqInfo) ([]byte, time.Duration) { + mode := h.mode() + start := time.Now() + + // Observe: the enforced path does nothing at all. Not "runs and discards" — it never + // runs, which is what makes the byte-identity guarantee structural. The measurement + // happens off-path, on a copy, and the request pays only the enqueue. + if mode == components.ModeObserve && !r.bypassed { + h.observe(r) + return r.body, time.Since(start) + } + + res := apply.BodyOpts(r.ctx, h.pipe, h.store, apply.Opts{ + Provider: r.provider, Body: r.body, Session: r.session, Bypass: r.bypassed, + Models: r.models, Window: r.window, CacheMode: h.opts.CacheMode, + Mode: mode, Tracker: h.tracker, + }) + added := time.Since(start) + if res.Body == nil { + return r.body, added + } + return res.Body, added +} + +// reqInfo is the per-request input both the inline pass and an off-path observation need. +// Bundled because an off-path run outlives the *http.Request it came from and must +// therefore hold a copy of everything, never a pointer into request-scoped state. +type reqInfo struct { + ctx context.Context + provider bschemas.ModelProvider + body []byte + session string + bypassed bool + models components.ModelSpec + window int +} + +func (h *Handler) mode() components.Mode { + if h.opts.Mode == "" { + return components.ModeSync + } + return h.opts.Mode +} + +// observe runs the pipeline off-path on a COPY of the request, against observe's own +// disjoint store, and records the result into the hypothetical metric namespace. Two +// independent reasons the enforced request cannot be affected: it was already forwarded +// from the untouched original, and this run touches no state the live path reads. +func (h *Handler) observe(r *reqInfo) { + if h.pool == nil { + return + } + // The job runs after the response is written, so nothing may alias request-scoped + // memory — and the context must be the pool's, not the request's, which is cancelled + // the moment the handler returns. + info := *r + info.body = append([]byte(nil), r.body...) + // A plain counter as the dedup key: one observation per call, coalescing nothing. Also + // why observe needs no session resolve on the request path — one more thing the + // enforced path does not pay for. + key := "observe:" + strconv.FormatUint(h.observeSeq.Add(1), 10) + + h.pool.Enqueue(key, func(ctx context.Context) { + apply.BodyOpts(ctx, h.pipe, h.shadow, apply.Opts{ + Provider: info.provider, Body: info.body, Session: info.session, + Models: info.models, Window: info.window, CacheMode: h.opts.CacheMode, + Mode: components.ModeObserve, + // The Tracker, so the projection is measured under the SAME cached-prefix + // boundary an enforcing mode would use. Without it the boundary is unknown, + // MaxCachedIdx is -1, the tail gate never fires, and every message in the + // transcript looks compactable — which inflates the projection against what + // sync actually achieves. Measured on SWE-bench: 9.5% projected against 0.8% + // enforced, because 50 extract_llm candidates passed the gate instead of 5. + // + // Safe off-path despite jobs finishing out of order: the boundary only ever + // grows, so a late job for a shorter turn cannot move it backwards. + Tracker: h.tracker, + // h.shadow, not the live store: see Handler.shadow. The live store must stay + // clean (a real request must never replay a decision that was never enforced), + // but the frozen decisions still have to accumulate across turns or the + // projection under-reports what enforcing would achieve. + }) + // The pipeline already emitted mode-stamped reports through the emitter; the + // Aggregator routes anything stamped observe into the potential_* namespace. No + // separate recording call here, which is what keeps the two namespaces from + // drifting apart. + }) +} diff --git a/proxy/modes_test.go b/proxy/modes_test.go new file mode 100644 index 0000000..dc720a6 --- /dev/null +++ b/proxy/modes_test.go @@ -0,0 +1,543 @@ +package proxy_test + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "runtime" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/rossoctl/context-guru/components" + _ "github.com/rossoctl/context-guru/components/all" + "github.com/rossoctl/context-guru/config" + "github.com/rossoctl/context-guru/metrics" + "github.com/rossoctl/context-guru/proxy" + "github.com/rossoctl/context-guru/store" +) + +// modeHandler is buildHandler plus an explicit operating mode, handing back the aggregator +// so a test can read the mode-partitioned rollups. +func modeHandler(t *testing.T, yaml, upstream string, mode components.Mode) (*proxy.Handler, *metrics.Aggregator) { + t.Helper() + return newModeHandler(t, yaml, upstream, mode, "") +} + +// newModeHandler is modeHandler with an explicit cache mode. "on" forces cache-awareness +// even on the OpenAI route, which is what makes the tail gate (and so MaxCachedIdx) +// actually participate — several behaviors are only observable then. +func newModeHandler(t *testing.T, yaml, upstream string, mode components.Mode, cacheMode string) (*proxy.Handler, *metrics.Aggregator) { + t.Helper() + cfg, err := config.LoadBytes([]byte(yaml)) + if err != nil { + t.Fatal(err) + } + agg := metrics.NewAggregator() + pipe, err := cfg.Build(agg) + if err != nil { + t.Fatal(err) + } + h := proxy.New(pipe, store.NewMemory(store.Options{}), agg, proxy.Options{ + OpenAIUpstream: upstream, AnthropicUpstream: upstream, Mode: mode, CacheMode: cacheMode, + }) + t.Cleanup(h.Close) + return h, agg +} + +// captureUpstream records every body the upstream receives. +func captureUpstream(t *testing.T) (*httptest.Server, func() [][]byte) { + t.Helper() + var mu sync.Mutex + var got [][]byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + mu.Lock() + got = append(got, b) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"ok":true}`)) + })) + t.Cleanup(srv.Close) + return srv, func() [][]byte { + mu.Lock() + defer mu.Unlock() + return append([][]byte(nil), got...) + } +} + +const modePipeline = "pipeline: [dedup, cacheinject]\n" + +func dupBody() []byte { + dump := strings.Repeat("a verbose repeated tool output line\n", 60) + return openAIBody( + map[string]any{"role": "user", "content": "do the thing"}, + map[string]any{"role": "tool", "tool_call_id": "a", "content": dump}, + map[string]any{"role": "tool", "tool_call_id": "b", "content": dump}, + ) +} + +func post(t *testing.T, srv *httptest.Server, body []byte) { + t.Helper() + resp, err := http.Post(srv.URL+"/openai/v1/chat/completions", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() +} + +// awaitSnapshot polls until cond holds, so an off-path result can land. +func awaitSnapshot(t *testing.T, agg *metrics.Aggregator, cond func(metrics.Snapshot) bool) metrics.Snapshot { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + var snap metrics.Snapshot + for time.Now().Before(deadline) { + snap = agg.Snapshot() + if cond(snap) { + return snap + } + time.Sleep(5 * time.Millisecond) + } + return snap +} + +// TestSyncIsTheDefaultAndUnchanged: an unset mode must behave exactly like the explicit +// sync mode, which is the pre-change behavior. The forwarded bodies are compared byte for +// byte — the golden test, expressed against the code's own output rather than a checked-in +// fixture that would drift on every unrelated component change. +func TestSyncIsTheDefaultAndUnchanged(t *testing.T) { + body := dupBody() + + upA, gotA := captureUpstream(t) + hA, _ := modeHandler(t, modePipeline, upA.URL, "") // unset + srvA := httptest.NewServer(hA.Mux()) + defer srvA.Close() + post(t, srvA, body) + + upB, gotB := captureUpstream(t) + hB, _ := modeHandler(t, modePipeline, upB.URL, components.ModeSync) + srvB := httptest.NewServer(hB.Mux()) + defer srvB.Close() + post(t, srvB, body) + + a, b := gotA(), gotB() + if len(a) != 1 || len(b) != 1 { + t.Fatalf("expected one forward each, got %d and %d", len(a), len(b)) + } + if !bytes.Equal(a[0], b[0]) { + t.Fatalf("default mode differs from explicit sync\n default: %s\n sync: %s", a[0], b[0]) + } + // And sync really did compact: otherwise the comparison above is vacuous. + if bytes.Equal(a[0], body) { + t.Fatal("sync forwarded the original unchanged — the golden comparison proves nothing") + } +} + +// TestObserveForwardsByteIdenticalBody is the mode's core promise: the agent receives +// exactly what it sent, while the hypothetical savings are still recorded. +func TestObserveForwardsByteIdenticalBody(t *testing.T) { + up, got := captureUpstream(t) + h, agg := modeHandler(t, modePipeline, up.URL, components.ModeObserve) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + + body := dupBody() + post(t, srv, body) + + fwd := got() + if len(fwd) != 1 { + t.Fatalf("expected one forward, got %d", len(fwd)) + } + if !bytes.Equal(fwd[0], body) { + t.Fatalf("observe mode MODIFIED the forwarded body\n sent: %s\n fwd: %s", body, fwd[0]) + } + + snap := awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.ObserveRequests > 0 }) + if snap.ObserveRequests == 0 { + t.Fatal("observe mode recorded nothing") + } + if snap.PotentialSavedTokens <= 0 { + t.Fatalf("no potential savings recorded: %+v", snap) + } + if snap.ActualBaselineTokens <= snap.ProjectedOptimizedTokens { + t.Fatalf("projected usage is not below the actual baseline: %d vs %d", + snap.ProjectedOptimizedTokens, snap.ActualBaselineTokens) + } + if snap.ObserveNotice == "" { + t.Fatal("observe mode did not emit its banner") + } + if snap.Mode != string(components.ModeObserve) { + t.Fatalf("mode not reported: %q", snap.Mode) + } +} + +// TestObserveMetricsCannotBeSummedIntoEnforcedTotals is the correctness requirement: a +// hypothetical must be unreachable from every enforced savings aggregate, or the product's +// headline claim is silently inflated. +func TestObserveMetricsCannotBeSummedIntoEnforcedTotals(t *testing.T) { + up, _ := captureUpstream(t) + h, agg := modeHandler(t, modePipeline, up.URL, components.ModeObserve) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + + for i := 0; i < 3; i++ { + post(t, srv, dupBody()) + } + snap := awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.ObserveRequests > 0 }) + if snap.ObserveRequests == 0 { + t.Fatal("nothing was observed; the test proves nothing") + } + if snap.Requests != 0 || snap.TokensBefore != 0 || snap.TokensAfter != 0 || snap.SavedTokens != 0 { + t.Fatalf("observe results leaked into the enforced totals: %+v", snap) + } + if snap.SyncEnforced != 0 { + t.Fatalf("observe counted as enforced: %d", snap.SyncEnforced) + } + if len(snap.Components) != 0 { + t.Fatalf("observe results leaked into the enforced per-component map: %v", snap.Components) + } + if len(snap.PotentialComponents) == 0 { + t.Fatal("per-component hypotheticals were not recorded at all") + } + // Two enforced-namespace fields are deliberately NOT zeroed, because they are real + // measurements rather than hypotheticals — cg_added_ms_avg (the actual enforced-path + // latency, ~0 here, which IS the headline) and context-guru's own model spend (observe + // measures off-path, and that costs real money). The notice labels the latter so it is + // not read as the cost of enforcing. + if snap.ObserveLLMNotice == "" { + t.Fatal("observe did not label its own off-path model spend") + } + // The serialized payload must keep the two vocabularies disjoint. + m := marshalMap(t, snap) + for _, enforced := range []string{"saved_tokens", "savings_pct", "tokens_before", "tokens_after", "requests", "components"} { + if _, ok := m[enforced]; !ok { + t.Fatalf("%q disappeared from /stats — backward compatibility broken", enforced) + } + } + for _, hypothetical := range []string{ + "potential_saved_tokens", "projected_optimized_tokens", "actual_baseline_tokens", + "potential_components", "observe_notice", "observe_hypothetical_requests", + } { + if _, ok := m[hypothetical]; !ok { + t.Fatalf("hypothetical key %q missing from the payload", hypothetical) + } + } +} + +// TestStatsStaysBackwardCompatible: deploy/harbor/*.py parses this payload, so fields may +// be added but never renamed or removed. +func TestStatsStaysBackwardCompatible(t *testing.T) { + up, _ := captureUpstream(t) + h, _ := modeHandler(t, modePipeline, up.URL, components.ModeSync) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + post(t, srv, dupBody()) + + resp, err := http.Get(srv.URL + "/stats") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var m map[string]any + if err := json.NewDecoder(resp.Body).Decode(&m); err != nil { + t.Fatal(err) + } + for _, k := range []string{ + "requests", "tokens_before", "tokens_after", "saved_tokens", "savings_pct", + "wasted_tokens", "bounces", "adjusted_saved", "components", "top_passthrough", + "llm_calls", "llm_input_tokens", "llm_output_tokens", + "cg_added_ms_avg", "upstream_ms_avg", "upstream_ms_avg_bypassed", + } { + if _, ok := m[k]; !ok { + t.Fatalf("/stats lost the pre-existing field %q", k) + } + } + for _, k := range []string{"mode", "sync_enforced"} { + if _, ok := m[k]; !ok { + t.Fatalf("/stats is missing the new field %q", k) + } + } + if m["mode"] != string(components.ModeSync) { + t.Fatalf("mode is %v, want sync", m["mode"]) + } + if m["sync_enforced"].(float64) < 1 { + t.Fatalf("sync request not counted as enforced: %v", m["sync_enforced"]) + } +} + +// TestObserveDoesNotInjectTheExpandTool: nothing was offloaded, so there is nothing to +// recover — and injecting a tool declaration would MODIFY the request, which is the one +// thing this mode promises never to do. +func TestObserveDoesNotInjectTheExpandTool(t *testing.T) { + up, got := captureUpstream(t) + h, _ := modeHandler(t, modePipeline, up.URL, components.ModeObserve) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + + // A body that already declares a tool: expand.Inject's "auto" mode would append to it. + body, err := json.Marshal(map[string]any{ + "model": "gpt-x", + "tools": []map[string]any{{"type": "function", "function": map[string]any{"name": "ls"}}}, + "messages": []map[string]any{ + {"role": "user", "content": "go"}, + {"role": "tool", "tool_call_id": "a", "content": strings.Repeat("noise\n", 200)}, + }, + }) + if err != nil { + t.Fatal(err) + } + post(t, srv, body) + + fwd := got() + if len(fwd) != 1 || !bytes.Equal(fwd[0], body) { + t.Fatalf("observe mode altered a tool-carrying request:\n sent: %s\n fwd: %s", body, fwd[0]) + } + if bytes.Contains(fwd[0], []byte("context_guru_expand")) { + t.Fatal("observe mode injected the expand tool") + } +} + +// TestCloseLeavesNoGoroutines: the pool the handler owns must be reclaimed. +func TestCloseLeavesNoGoroutines(t *testing.T) { + // Everything unrelated (the mock upstream's own goroutines) is created BEFORE the + // baseline, so the only difference this measures is the pool's. + up, _ := captureUpstream(t) + settleGoroutines() + before := runtime.NumGoroutine() + + cfg, err := config.LoadBytes([]byte(modePipeline)) + if err != nil { + t.Fatal(err) + } + agg := metrics.NewAggregator() + pipe, err := cfg.Build(agg) + if err != nil { + t.Fatal(err) + } + h := proxy.New(pipe, store.NewMemory(store.Options{}), agg, proxy.Options{ + OpenAIUpstream: up.URL, Mode: components.ModeObserve, + }) + h.Close() + h.Close() // idempotent + + settleGoroutines() + if after := runtime.NumGoroutine(); after > before { + t.Fatalf("goroutine leak after Close: %d before, %d after", before, after) + } +} + +// TestSyncModeStartsNoPool: sync adds no off-path machinery at all. +func TestSyncModeStartsNoPool(t *testing.T) { + up, _ := captureUpstream(t) + _, agg := modeHandler(t, modePipeline, up.URL, components.ModeSync) + raw, err := json.Marshal(agg.Snapshot()) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(raw, []byte("observe_notice")) { + t.Fatalf("sync mode emitted an observe banner: %s", raw) + } +} + +func TestUnknownModeIsRejected(t *testing.T) { + for _, bad := range []string{"turbo", "async"} { + if _, err := config.LoadBytes([]byte("pipeline: [dedup]\nmode: " + bad + "\n")); err == nil { + t.Fatalf("mode %q was accepted", bad) + } + } + for _, ok := range []string{"", "sync", "observe"} { + if _, err := config.LoadBytes([]byte("pipeline: [dedup]\nmode: " + ok + "\n")); err != nil { + t.Fatalf("mode %q rejected: %v", ok, err) + } + } +} + +// TestCompactEndpointIgnoresMode: /compact hands the compacted body back in the response, +// so it is synchronous by contract regardless of how forwarded traffic is handled. Worth +// pinning because observe mode turning /compact into a no-op would silently break offline +// replay and the llm-d-router integration, and nothing else would notice. +func TestCompactEndpointIgnoresMode(t *testing.T) { + body := dupBody() + var outs [][]byte + for _, mode := range []components.Mode{components.ModeSync, components.ModeObserve} { + up, _ := captureUpstream(t) + h, _ := modeHandler(t, "pipeline: [dedup]\n", up.URL, mode) + srv := httptest.NewServer(h.Mux()) + resp, err := http.Post(srv.URL+"/compact", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + out, _ := io.ReadAll(resp.Body) + resp.Body.Close() + srv.Close() + if bytes.Equal(out, body) { + t.Fatalf("/compact returned the original unchanged under mode %s", mode) + } + outs = append(outs, out) + } + if !bytes.Equal(outs[0], outs[1]) { + t.Fatalf("/compact output depends on the operating mode:\n %s\n %s", outs[0], outs[1]) + } +} + +// TestObserveProjectionAgreesWithSyncActuals is the check that validates the whole mode: +// run the SAME turns under sync and under observe, and observe's projected saving must +// match what sync actually achieved. It caught two real errors — without the shared cache +// boundary observe over-projected 11x, and without its own persistent store it +// under-projected 3x. +func TestObserveProjectionAgreesWithSyncActuals(t *testing.T) { + dump := strings.Repeat("a long stale tool output worth offloading\n", 80) + // Each turn appends SEVERAL tool outputs, so more than one lands beyond the previous + // turn's boundary. With only one new output per turn it is always the one mask keeps, + // nothing is eligible in the tail, and both arms trivially save zero — which would hide + // the very disagreement this test exists for. + turns := func() [][]byte { + var out [][]byte + for n := 1; n <= 5; n++ { + msgs := []map[string]any{{"role": "user", "content": "go"}} + for i := 0; i < n*4; i++ { + msgs = append(msgs, + map[string]any{"role": "assistant", "content": "step " + strconv.Itoa(i)}, + map[string]any{"role": "tool", "tool_call_id": "t" + strconv.Itoa(i), "content": dump + strconv.Itoa(i)}) + } + out = append(out, openAIBody(msgs...)) + } + return out + }() + + // keep_last: 1 so each turn's growth pushes the previous tool output into mask's range. + yaml := "pipeline: [mask]\ncomponents:\n mask:\n keep_last: 1\n" + drive := func(mode components.Mode) metrics.Snapshot { + up, _ := captureUpstream(t) + // cache=on so cache-awareness (and therefore the tail gate) is live: that gate is + // exactly what observe used to ignore, and without it the two arms agree trivially. + h, agg := newModeHandler(t, yaml, up.URL, mode, "on") + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + for _, b := range turns { + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/openai/v1/chat/completions", bytes.NewReader(b)) + req.Header.Set("x-context-guru-session", "agree") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + if mode == components.ModeObserve { + return awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.ObserveRequests >= int64(len(turns)) }) + } + return agg.Snapshot() + } + + sync := drive(components.ModeSync) + obs := drive(components.ModeObserve) + + t.Logf("sync: before=%d saved=%d (%.2f%%)", sync.TokensBefore, sync.SavedTokens, sync.SavingsPct) + t.Logf("observe: baseline=%d potential=%d (%.2f%%) reqs=%d", + obs.ActualBaselineTokens, obs.PotentialSavedTokens, obs.PotentialSavingsPct, obs.ObserveRequests) + + if sync.SavedTokens == 0 || obs.PotentialSavedTokens == 0 { + t.Fatalf("one side saved nothing; the agreement check is vacuous (sync=%d observe=%d)", + sync.SavedTokens, obs.PotentialSavedTokens) + } + // Both saw the same traffic under the same boundary, so the projection must track the + // actual closely. A generous band still catches the class of bug this found (observe was + // 3x low before the shadow store, 11x high before the shared tracker). + ratio := float64(obs.PotentialSavedTokens) / float64(sync.SavedTokens) + if ratio < 0.75 || ratio > 1.33 { + t.Fatalf("observe's projection disagrees with sync's actual: %d vs %d (ratio %.2f)", + obs.PotentialSavedTokens, sync.SavedTokens, ratio) + } +} + +// TestObserveNeverWritesTheLiveStore: observe gets a store of its own so its frozen +// decisions accumulate across turns (without that it under-projects by ~3x), but the live +// store must stay pristine — otherwise a later real request would replay a decision that +// was never enforced, which is a request modification arriving by the back door. +func TestObserveNeverWritesTheLiveStore(t *testing.T) { + up, _ := captureUpstream(t) + cfg, err := config.LoadBytes([]byte("pipeline: [mask]\ncomponents:\n mask:\n keep_last: 1\n")) + if err != nil { + t.Fatal(err) + } + agg := metrics.NewAggregator() + pipe, err := cfg.Build(agg) + if err != nil { + t.Fatal(err) + } + live := &countingStore{Store: store.NewMemory(store.Options{})} + h := proxy.New(pipe, live, agg, proxy.Options{ + OpenAIUpstream: up.URL, Mode: components.ModeObserve, CacheMode: "on", + }) + t.Cleanup(h.Close) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + + dump := strings.Repeat("a long stale tool output worth offloading\n", 80) + for n := 1; n <= 4; n++ { + msgs := []map[string]any{{"role": "user", "content": "go"}} + for i := 0; i < n*4; i++ { + msgs = append(msgs, + map[string]any{"role": "assistant", "content": "step " + strconv.Itoa(i)}, + map[string]any{"role": "tool", "tool_call_id": "t" + strconv.Itoa(i), "content": dump + strconv.Itoa(i)}) + } + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/openai/v1/chat/completions", bytes.NewReader(openAIBody(msgs...))) + req.Header.Set("x-context-guru-session", "isolated") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + snap := awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.PotentialSavedTokens > 0 }) + if snap.PotentialSavedTokens == 0 { + t.Fatal("observe recorded no savings; the isolation check is vacuous") + } + if n := live.puts.Load(); n != 0 { + t.Fatalf("observe mode wrote %d entries into the LIVE store", n) + } +} + +// countingStore counts writes so a test can assert none happened. +type countingStore struct { + store.Store + puts atomic.Int64 +} + +func (c *countingStore) Put(key string, payload []byte) { + c.puts.Add(1) + c.Store.Put(key, payload) +} + +func (c *countingStore) MarkSticky(session, id string) { + c.puts.Add(1) + c.Store.MarkSticky(session, id) +} + +func marshalMap(t *testing.T, v any) map[string]json.RawMessage { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + return m +} + +func settleGoroutines() { + for i := 0; i < 20; i++ { + runtime.Gosched() + time.Sleep(5 * time.Millisecond) + } +} diff --git a/proxy/proxy.go b/proxy/proxy.go index 77d5a79..8c454d9 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -19,6 +19,7 @@ import ( "net/http" "os" "strings" + "sync/atomic" "time" bschemas "github.com/maximhq/bifrost/core/schemas" @@ -28,6 +29,7 @@ import ( "github.com/rossoctl/context-guru/expand" "github.com/rossoctl/context-guru/internal/cheapmodel" "github.com/rossoctl/context-guru/metrics" + "github.com/rossoctl/context-guru/modes" "github.com/rossoctl/context-guru/schema" "github.com/rossoctl/context-guru/store" "github.com/tidwall/gjson" @@ -79,6 +81,22 @@ type Options struct { // handler always uses the configured pipeline. Supplied by main (which holds // the config + emitter) so proxy stays decoupled from the config package. PipelineFor func(preset string, names []string) (*components.Pipeline, error) + // Mode is the operating mode (#31): components.ModeSync (default, and byte-identical + // to pre-mode behavior) or ModeObserve. Empty = sync. Explicit by design — never + // inferred from the rest of the configuration. + Mode components.Mode + // Observe tunes observe mode's off-path measurement. Ignored in sync mode. + Observe ObserveOptions +} + +// ObserveOptions tunes observe mode: one option per real decision. +type ObserveOptions struct { + // MaxQueue bounds the off-path measurement queue; a full queue DROPS (counted) rather + // than blocking the request path. 0 = modes.DefaultMaxQueue. + MaxQueue int `yaml:"max_queue"` + // Workers is the number of drain goroutines. 0 = modes.DefaultWorkers (1), which keeps + // one measurement's cheap-model call in flight per process. + Workers int `yaml:"workers"` } // upstream binds a provider to its base URL, the canonical provider path to POST @@ -97,6 +115,24 @@ type Handler struct { agg *metrics.Aggregator opts Options client *http.Client + // tracker owns the per-session cached-prefix boundary. Always present: every mode + // benefits from reading and recording it in one locked step (the previous + // read-then-deferred-write raced between concurrent turns of a session). + tracker *modes.Tracker + // pool runs off-path measurements. nil in sync mode — there are none. + pool *modes.Pool + // observeSeq numbers observations so each request enqueues one. + observeSeq atomic.Uint64 + // shadow is observe mode's own state store, separate from the live one. Observe must + // not write into the live store — a real request would then replay a decision that was + // never enforced — but it also cannot simply discard its writes: offloaders FREEZE a + // decision and replay it on every later turn, which is where most of the sustained + // saving comes from. Throwing that away each turn makes observe see only the current + // tail and UNDER-project by ~3x against what sync achieves. + // + // So observe gets a store of its own: as persistent as the live one, and completely + // disjoint from it. + shadow store.Store } // New builds the proxy handler. agg may be nil (no /stats rollups). @@ -105,9 +141,22 @@ func New(pipe *components.Pipeline, st store.Store, agg *metrics.Aggregator, opt if c == nil { c = &http.Client{Timeout: 5 * time.Minute} } - return &Handler{pipe: pipe, store: st, agg: agg, opts: opts, client: c} + h := &Handler{pipe: pipe, store: st, agg: agg, opts: opts, client: c, tracker: modes.NewTracker(0)} + if h.mode() == components.ModeObserve { + h.pool = modes.NewPool(opts.Observe.MaxQueue, opts.Observe.Workers) + h.shadow = store.NewMemory(store.Options{}) + } + if agg != nil { + agg.SetMode(h.mode()) + } + return h } +// Close shuts down the off-path worker pool and waits briefly for its goroutines to exit, +// so a host that builds and discards handlers (tests, a reload) leaks none. Safe on a +// sync-mode handler and safe to call twice. +func (h *Handler) Close() { h.pool.Stop() } + // Mux wires the routes: chat proxying + health/stats/expand management. func (h *Handler) Mux() *http.ServeMux { m := http.NewServeMux() @@ -361,24 +410,32 @@ func (h *Handler) chat(provider bschemas.ModelProvider, up upstream) http.Handle body = orig } }() - applyStart := time.Now() - body, _ = apply.BodyFull( - r.Context(), h.pipe, h.store, provider, body, - r.Header.Get("x-context-guru-session"), - bypassed, - models, window, h.opts.CacheMode, - ) + var added time.Duration + body, added = h.applyMode(&reqInfo{ + ctx: r.Context(), + provider: provider, + body: body, + session: r.Header.Get("x-context-guru-session"), + bypassed: bypassed, + models: models, + window: window, + }) if h.agg != nil && !bypassed { - h.agg.RecordAddedLatency(float64(time.Since(applyStart).Microseconds()) / 1000.0) + h.agg.RecordAddedLatency(float64(added.Microseconds()) / 1000.0) } // Advertise the expand tool so the model can recover any offloaded content // (closes the reversibility loop h.serve drives). Sticky/idempotent + appended // last to keep the provider prefix cache warm; gated by InjectExpand + store. - mode := h.opts.InjectExpand - if mode == "" { - mode = expand.InjectAuto + // Skipped in observe mode: nothing was offloaded, so there is nothing to + // recover, and injecting a tool declaration would MODIFY the request — which is + // precisely the one thing observe mode promises never to do. + if h.mode() != components.ModeObserve { + im := h.opts.InjectExpand + if im == "" { + im = expand.InjectAuto + } + body, _ = expand.Inject(string(provider), im, body, h.store.Persists()) } - body, _ = expand.Inject(string(provider), mode, body, h.store.Persists()) }() h.serve(w, r, provider, up, body, bypassed) }