feat(proxy): add observe mode — measure what compaction would save, without enforcing it

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.

apply/apply.go

@@ -112,21 +112,38 @@ 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
 	// the original body unchanged. This makes CLAUDE.md's fail-open invariant hold for
 	// 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,38 +166,50 @@ 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 {
 		dumpToolOutputs(norm)
 	}
 	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,
 		Session:      sessionID,
 		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, <summary>, 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

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
+}

cmd/context-guru-proxy/main.go

@@ -36,13 +36,21 @@ 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()
 
 	cfg, err := loadConfig(*cfgPath, *preset)
 	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)
 	}

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.