feat(proxy): async mode — cache-safe deferred compaction (HELD: benefit unmeasured since #34)

README.md

@@ -170,6 +170,35 @@ 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. Two other modes trade that off:
+
+| Mode | The request path | Use it when |
+|---|---|---|
+| **`sync`** *(default)* | Compacts inline; the caller waits (~450 ms/req measured on Terminal-Bench). | You want the full saving from turn one. |
+| **`async`** | Forwards immediately, replaying decisions an earlier turn computed; the expensive compaction runs off-path and benefits later turns. | Latency on the request path matters more than saving on turn one. |
+| **`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: async
+async:
+  cache_uncompacted_tail: false   # safe default: protect cache-write economics
+```
+
+Two things worth knowing before reaching for `async`: a cache-write costs **11.5x** a
+cache-read, so by default context-guru refuses to place a cache breakpoint on a tail a
+pending compaction is going to replace — caching it and then replacing it is what
+tripled headroom's cache-write on Terminal-Bench and would make `async` strictly worse
+than `sync`. And 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.
+
+`observe` 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 +211,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 async vs observe: when to use each, async's cache trade-off, how to read observe numbers.
 - [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,46 @@ 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 generation snapshot async mode needs. 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
+	// Async, on the REQUEST path: replay only decisions that are already computed. The
+	// expensive part of a compaction is the LLM call, which is the entire reason async
+	// exists, so the inline pass gets no model clients and every NeedsModel component
+	// degrades to its deterministic path or no-ops (that degradation is already a
+	// documented contract). The off-path job (Deferred) gets the clients.
+	if mode == components.ModeAsync && !o.Deferred {
+		models = components.ModelSpec{}
+	}
 	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
@@ -142,35 +167,88 @@ 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)
+	cacheAware := resolveCacheAware(o.CacheMode, provider, body)
+	// Turn accounting is independent of cache mode: the generation counts TURNS, and a
+	// turn happens whether or not the backend caches. Deriving it inside the cache-aware
+	// branch left every generation at 0 with cache_mode: off, which both disabled the
+	// stale guard and collided with 0's use as "nothing pending".
+	if o.Tracker != nil && !o.Deferred {
+		pl, gen := o.Tracker.Turn(sessionID, len(norm))
+		res.PrevLen, res.Generation = pl, gen
+	}
 	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))
+		//
+		// The boundary comes from the Tracker when the host supplies one: it reads the
+		// previous length and records this turn's in ONE locked call, which is what
+		// removes the concurrent-turn race the old read-then-deferred-write had
+		// (#31/#25). Without a tracker (library callers, /compact) the legacy store path
+		// stands — same numbers, same race, no behavior change for them.
+		switch {
+		case o.PrevLen != nil:
+			maxCachedIdx = *o.PrevLen - 1
+		case o.Tracker != nil:
+			maxCachedIdx = res.PrevLen - 1 // recorded above, in one locked call
+		default:
+			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,
-		CacheAware:   cacheAware,
-		MaxCachedIdx: maxCachedIdx,
+	// Async cache policy: while a compaction for this session is queued but not landed,
+	// the un-compacted tail is about to be REPLACED, so no breakpoint may be committed
+	// at or beyond it (see components.Ctx.NoCacheAtOrAfter). CacheUncompactedTail=true
+	// is the escape hatch for a confirmed non-caching backend, where the protection buys
+	// nothing.
+	//
+	// Three conditions beyond "async", each one a bug found in review:
+	//
+	//   - cacheAware. With cache_mode: off there is no cached prefix to protect and no
+	//     boundary to protect it at, so blocking breakpoints would suppress caching
+	//     forever for nothing (the two knobs interacted backwards).
+	//   - a boundary that exists. On a session's FIRST turn prevLen is 0, so the whole
+	//     request is "tail" and blocking it wrote zero breakpoints — on precisely the
+	//     turn whose job is to write the prefix. There is also nothing to protect yet:
+	//     no compaction is pending, because no earlier turn enqueued one.
+	//   - the tail a pending job will actually replace. The job enqueued by the PREVIOUS
+	//     turn targets that turn's tail, which by now sits at or below the boundary.
+	//     Blocking from the boundary up protected this turn's new messages, which no
+	//     pending job is going to touch — off by one turn, and it protected the wrong
+	//     span. The doomed span starts where the previous turn's own tail started.
+	tailPending, noCacheAt := false, 0
+	if mode == components.ModeAsync && !o.Deferred && !bypass && !o.CacheUncompactedTail &&
+		cacheAware && o.PendingFrom > 0 {
+		tailPending = true
+		noCacheAt = o.PendingFrom
 	}
+	c := &components.Ctx{
+		Ctx:                    ctx,
+		Session:                sessionID,
+		Store:                  st,
+		Model:                  models,
+		Bypass:                 bypass,
+		CtxWindow:              o.Window,
+		CacheAware:             cacheAware,
+		MaxCachedIdx:           maxCachedIdx,
+		Mode:                   mode,
+		Deferred:               o.Deferred,
+		TailCachePending:       tailPending,
+		NoCacheAtOrAfter:       noCacheAt,
+		StripCallerBreakpoints: o.StripCallerBreakpoints,
+	}
+	res.Session = sessionID
 
 	// Canonical form of each normalized message BEFORE the pipeline, so a
 	// count-changing component (summarize) can be mapped back to the body.
@@ -179,7 +257,8 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr
 		normPre[i], _ = json.Marshal(norm[i])
 	}
 
-	pipe.Run(chat, c)
+	res.Run = pipe.Run(chat, c)
+	res.TailUnprotected = c.TailUnprotected()
 
 	// A component changed the message count (summarize restructures the transcript
 	// to [msg0, <summary>, last-K]). Rebuild the messages array preserving each
@@ -188,9 +267,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
@@ -208,14 +289,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)
@@ -227,7 +310,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
@@ -238,7 +322,8 @@ func BodyFull(ctx context.Context, pipe *components.Pipeline, st store.Store, pr
 	if changed && dumpPath != "" {
 		dumpChanges(c.Session, changes)
 	}
-	return out, changed
+	res.Body, res.Changed = out, changed
+	return res
 }
 
 // resolveCacheAware decides whether cache-aware compaction is active for this