From af57def2a5e0decbdddd03954ed82043989fd418 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 12 Oct 2026 18:14:07 +0000 Subject: [PATCH] feat(offload): cacheanchor component and mask deep_batch, a breakpoint before the first edited message Signed-off-by: Osher-Elhadad --- apply/cacheanchor_wire_test.go | 113 +++++++++++++ components/offload/cacheanchor.go | 183 ++++++++++++++++++++++ components/offload/cacheanchor_test.go | 124 +++++++++++++++ components/offload/commitgate_test.go | 2 + components/offload/mask.go | 36 ++++- components/offload/mask_deepbatch_test.go | 39 +++++ docs/components/advanced-offload.md | 13 ++ 7 files changed, 508 insertions(+), 2 deletions(-) create mode 100644 apply/cacheanchor_wire_test.go create mode 100644 components/offload/cacheanchor.go create mode 100644 components/offload/cacheanchor_test.go create mode 100644 components/offload/mask_deepbatch_test.go diff --git a/apply/cacheanchor_wire_test.go b/apply/cacheanchor_wire_test.go new file mode 100644 index 000000000..33c7a4cdb --- /dev/null +++ b/apply/cacheanchor_wire_test.go @@ -0,0 +1,113 @@ +package apply + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/tidwall/gjson" + + "github.com/rossoctl/context-guru/components" + _ "github.com/rossoctl/context-guru/components/all" + "github.com/rossoctl/context-guru/store" +) + +// anchorBody is a Claude Code shaped request: two marked system blocks, a tool, and (assistant tool_use, +// user tool_result) pairs with a mark on the last message only. +func anchorBody(turns int) []byte { + msgs := []any{map[string]any{"role": "user", "content": []any{map[string]any{"type": "text", "text": "investigate"}}}} + for t := 1; t <= turns; t++ { + var sb strings.Builder + lines := 120 + if t <= 4 { + lines = 3 // small early outputs are never masked, so the early prefix stays unchanged + } + for j := 0; j < lines; j++ { + fmt.Fprintf(&sb, "2026-10-11 worker-%d reconciled batch %d for catalog item %d ok\n", j%7, t*1000+j, j) + } + id := fmt.Sprintf("toolu_%03d", t) + msgs = append(msgs, + map[string]any{"role": "assistant", "content": []any{map[string]any{"type": "text", "text": "step"}, + map[string]any{"type": "tool_use", "id": id, "name": "Bash", "input": map[string]any{"command": "s"}}}}, + map[string]any{"role": "user", "content": []any{map[string]any{"type": "tool_result", "tool_use_id": id, "content": sb.String()}}}) + } + last := msgs[len(msgs)-1].(map[string]any)["content"].([]any) + last[len(last)-1].(map[string]any)["cache_control"] = map[string]any{"type": "ephemeral"} + b, _ := json.Marshal(map[string]any{ + "model": "claude-opus-5", "max_tokens": 64, + "system": []any{ + map[string]any{"type": "text", "text": "billing header", "cache_control": map[string]any{"type": "ephemeral"}}, + map[string]any{"type": "text", "text": "You are an agent.", "cache_control": map[string]any{"type": "ephemeral"}}}, + "tools": []any{map[string]any{"name": "Bash", "description": "run", "input_schema": map[string]any{"type": "object"}}}, + "messages": msgs}) + return b +} + +// assistantMarks lists assistant messages other than the last message that carry cache_control. +func assistantMarks(body []byte) []int { + var at []int + for i, m := range gjson.GetBytes(body, "messages").Array() { + if m.Get("role").String() != "assistant" { + continue + } + for _, b := range m.Get("content").Array() { + if b.Get("cache_control").Exists() { + at = append(at, i) + } + } + } + return at +} + +func TestCacheAnchorReachesTheWireBehindABatchedDeepMask(t *testing.T) { + run := func(spec string) (anchored int, maxBP int, edited bool) { + var comps []components.Component + for _, name := range strings.Split(spec, ",") { + cfg := "" + if name == "mask" { + cfg = "keep_recent: 2\nmin_tokens: 100\ndeep_batch: 3\n" + } + c, err := components.New(name, []byte(cfg)) + if err != nil { + t.Fatal(err) + } + comps = append(comps, c) + } + pipe := components.NewPipeline(comps, nil) + st := store.NewMemory(store.Options{MaxEntries: 4000}) + for turns := 1; turns <= 16; turns++ { + raw := anchorBody(turns) + res := BodyOpts(context.Background(), pipe, st, Opts{Provider: bschemas.Anthropic, Body: raw, Session: "anchor-wire"}) + out := res.Body + if out == nil { + out = raw + } + if n := CountBreakpoints(out).Total(); n > maxBP { + maxBP = n + } + if len(assistantMarks(out)) > 0 { + anchored++ + } + if res.Changed && strings.Contains(string(out), "masked") { + edited = true + } + } + return + } + a, bp, edited := run("mask,cacheanchor") + if !edited { + t.Fatal("the batched mask never edited history, so the scenario did not exercise the anchor") + } + if a == 0 { + t.Error("an edit inside the cached history was sent without an anchor breakpoint") + } + if bp > 4 { + t.Errorf("%d breakpoints on the wire, a fifth is a 400", bp) + } + if c, _, _ := run("mask"); c != 0 { + t.Errorf("control: without cacheanchor no assistant turn should carry a mark, got %d requests", c) + } +} diff --git a/components/offload/cacheanchor.go b/components/offload/cacheanchor.go new file mode 100644 index 000000000..b2e0bdbcb --- /dev/null +++ b/components/offload/cacheanchor.go @@ -0,0 +1,183 @@ +package offload + +import ( + "crypto/sha1" + "encoding/hex" + "encoding/json" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/internal/thread" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +func init() { components.Register("cacheanchor", newCacheAnchor) } + +// CacheAnchor keeps an edit inside the cached history from costing a rewrite of everything after the +// system prompt. When a request differs from the thread's previous forwarded request at message p, the +// provider can read only up to the last cache entry that still describes this request's prefix. With the +// client's single breakpoint on its last message there is none before p inside the 20-block look-back, so +// the whole conversation is written again. CacheAnchor puts one cache_control mark on the assistant turn +// that follows the last earlier request end e whose forwarded prefix is unchanged (e <= p-2): that +// breakpoint looks back one block, finds the entry the request ending at e wrote, and reads the prefix up +// to it; only what follows is rewritten. +// +// It must run AFTER every component that edits history, because it compares what is about to be sent. +// It adds at most one breakpoint, never beyond the provider's cap of four counting the client's own +// (Ctx.ExistingBreakpoints), skips a cold request (the whole prefix is rewritten anyway), and adds no +// mark when p is the previous end or later (an append: the client's own mark already reads the prefix). +type CacheAnchor struct{} + +type anchorEnd struct { + H string `json:"h"` // prefix hash written at this request end + T int64 `json:"t"` // unix ms when the entry was last written or read +} + +type anchorState struct { + Prefix []string `json:"p"` // prefix hash after each message of the previous forwarded request + Ends map[int]anchorEnd `json:"e"` // request end index -> entry +} + +// anchorNow is the clock; a test replaces it. +var anchorNow = time.Now + +// anchorEntryTTL is how long an entry is trusted to be alive: the provider keeps one five minutes after +// it was last written or read, and a margin is kept for clock and network skew. +const anchorEntryTTL = 270 * time.Second + +const anchorMaxEnds = 64 + +func newCacheAnchor(raw []byte) (components.Component, error) { + if err := components.Decode(raw, &struct{}{}); err != nil { + return nil, err + } + return CacheAnchor{}, nil +} + +func (CacheAnchor) Name() string { return "cacheanchor" } +func (CacheAnchor) Enabled(*components.Ctx) bool { return true } + +func anchorKey(c *components.Ctx) string { + return store.SweepKeepPrefix + "anchor:" + thread.Key(c.Session, c.Thread) +} + +func prefixHashes(msgs []bschemas.ChatMessage) []string { + h := sha1.New() + out := make([]string, len(msgs)) + for i := range msgs { + h.Write([]byte(msgs[i].Role)) + h.Write([]byte{0}) + h.Write([]byte(schema.MessageText(msgs[i]))) + h.Write([]byte{0}) + out[i] = hex.EncodeToString(h.Sum(nil))[:12] + } + return out +} + +func (CacheAnchor) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, c *components.Ctx) ([]string, error) { + rep.Skipped = true + if c == nil || c.Store == nil || len(req.Input) == 0 { + return nil, nil + } + ph := prefixHashes(req.Input) + var st anchorState + if b, ok := c.Store.Get(anchorKey(c)); ok { + _ = json.Unmarshal(b, &st) + } + if st.Ends == nil { + st.Ends = map[int]anchorEnd{} + } + // first message whose prefix differs from the previous forwarded request + p := -1 + for k := 0; k < len(st.Prefix) && k < len(ph); k++ { + if st.Prefix[k] != ph[k] { + p = k + break + } + } + if p < 0 && len(ph) < len(st.Prefix) { + p = len(ph) // the transcript got shorter: everything after is new + } + now := anchorNow().UnixMilli() + if p >= 2 && !c.ColdCache { + // the latest earlier request end, strictly before p-1, whose entry still describes our prefix + best := -1 + for e := p - 2; e >= 0; e-- { + if en, ok := st.Ends[e]; ok && e < len(ph) && en.H == ph[e] && + now-en.T < anchorEntryTTL.Milliseconds() { + best = e + break + } + } + if best < 0 { + rep.Gate("anchor_no_valid_entry") + } else if idx := best + 1; !anchorMarkable(&req.Input[idx]) { + rep.Gate("anchor_not_markable") + } else if anchorBreakpoints(req, c) >= 4 { + rep.Gate("anchor_no_free_slot") + } else { + last := &req.Input[idx].Content.ContentBlocks[len(req.Input[idx].Content.ContentBlocks)-1] + last.CacheControl = &bschemas.CacheControl{Type: bschemas.CacheControlTypeEphemeral} + st.Ends[best] = anchorEnd{H: st.Ends[best].H, T: now} // our read refreshes it + rep.Event("anchor_placed") + rep.Skipped = false + } + } + // remember what we send: entries whose prefix no longer matches are stale, and this request's end is new + for e, en := range st.Ends { + if e >= len(ph) || en.H != ph[e] || now-en.T >= anchorEntryTTL.Milliseconds() { + delete(st.Ends, e) + } + } + st.Ends[len(ph)-1] = anchorEnd{H: ph[len(ph)-1], T: now} + if len(st.Ends) > anchorMaxEnds { + lo := len(ph) + for e := range st.Ends { + if e < lo { + lo = e + } + } + delete(st.Ends, lo) + } + st.Prefix = ph + if b, err := json.Marshal(st); err == nil { + c.Store.Put(anchorKey(c), b) + } + return nil, nil +} + +// anchorMarkable: an assistant turn with a block array whose last block carries no mark yet. (A tool +// result message normalises to a string and cannot carry a block-level mark through the pipeline.) +func anchorMarkable(m *bschemas.ChatMessage) bool { + if m.Role != bschemas.ChatMessageRoleAssistant || m.Content == nil || len(m.Content.ContentBlocks) == 0 { + return false + } + return m.Content.ContentBlocks[len(m.Content.ContentBlocks)-1].CacheControl == nil +} + +// anchorBreakpoints counts the breakpoints on the request: the host's raw count (system, tools and marks +// no component sees) or the visible ones, whichever is larger. +func anchorBreakpoints(req *bschemas.BifrostChatRequest, c *components.Ctx) int { + visible := 0 + for i := range req.Input { + if req.Input[i].Content == nil { + continue + } + for _, b := range req.Input[i].Content.ContentBlocks { + if b.CacheControl != nil { + visible++ + } + } + } + if c.ExistingBreakpoints > visible { + return c.ExistingBreakpoints + } + return visible +} + +func init() { + components.RegisterFields("cacheanchor", struct{}{}, nil) +} diff --git a/components/offload/cacheanchor_test.go b/components/offload/cacheanchor_test.go new file mode 100644 index 000000000..7f9661e55 --- /dev/null +++ b/components/offload/cacheanchor_test.go @@ -0,0 +1,124 @@ +package offload + +import ( + "strconv" + "strings" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +func anchorBlockAssistant(text string) bschemas.ChatMessage { + return bschemas.ChatMessage{Role: bschemas.ChatMessageRoleAssistant, + Content: &bschemas.ChatMessageContent{ContentBlocks: []bschemas.ChatContentBlock{{Type: bschemas.ChatContentBlockTypeText, Text: &text}}}} +} + +// anchorConv: user, then n (assistant tool turn, tool result) pairs; a request ends on its last tool result. +func anchorConv(n int) *bschemas.BifrostChatRequest { + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{userMsg("go")}} + for i := 0; i < n; i++ { + req.Input = append(req.Input, anchorBlockAssistant("step "+strconv.Itoa(i)), + toolResultMsgWithID("toolu_a"+strconv.Itoa(i), "output "+strconv.Itoa(i)+"\n"+strings.Repeat("line\n", 200))) + } + return req +} + +func marks(req *bschemas.BifrostChatRequest) []int { + var at []int + for i := range req.Input { + if req.Input[i].Content == nil { + continue + } + for _, b := range req.Input[i].Content.ContentBlocks { + if b.CacheControl != nil { + at = append(at, i) + } + } + } + return at +} + +func runAnchor(t *testing.T, c *components.Ctx, req *bschemas.BifrostChatRequest) *components.Report { + t.Helper() + comp, _ := newCacheAnchor(nil) + rep := &components.Report{Component: "cacheanchor"} + if _, err := comp.(components.Offload).Offload(req, rep, c); err != nil { + t.Fatal(err) + } + return rep +} + +func anchorCtx(st store.Store) *components.Ctx { + return &components.Ctx{Session: "anc", Store: st, ExistingBreakpoints: 3} +} + +func TestCacheAnchorMarksTheTurnAfterTheLastValidEntryBeforeAnEdit(t *testing.T) { + now := time.Unix(1_000_000, 0) + anchorNow = func() time.Time { return now } + defer func() { anchorNow = time.Now }() + st := store.NewMemory(store.Options{MaxEntries: 100}) + // three appends: each request ends one pair later + for n := 3; n <= 6; n++ { + if rep := runAnchor(t, anchorCtx(st), anchorConv(n)); rep.Events["anchor_placed"] != 0 { + t.Fatalf("an append must not be anchored (n=%d)", n) + } + now = now.Add(20 * time.Second) + } + // the next request edits the tool result of pair 5 (message 10); requests ended at 6, 8, 10 and 12 before + req := anchorConv(7) + schema.SetMessageText(&req.Input[10], "[masked]") + rep := runAnchor(t, anchorCtx(st), req) + got := marks(req) + // first change p=10; latest earlier end e<=8 is 8 -> mark message 9 (the assistant turn after it) + if rep.Events["anchor_placed"] != 1 || len(got) != 1 || got[0] != 9 { + t.Fatalf("want one mark on message 9, got %v (events %v gates %v)", got, rep.Events, rep.Gates) + } +} + +func TestCacheAnchorRespectsTheCapTheClockAndColdRequests(t *testing.T) { + now := time.Unix(2_000_000, 0) + anchorNow = func() time.Time { return now } + defer func() { anchorNow = time.Now }() + prime := func(st store.Store) { + for n := 3; n <= 6; n++ { + runAnchor(t, anchorCtx(st), anchorConv(n)) + now = now.Add(20 * time.Second) + } + } + edited := func() *bschemas.BifrostChatRequest { + r := anchorConv(7) + schema.SetMessageText(&r.Input[10], "[masked]") + return r + } + // no free slot + st := store.NewMemory(store.Options{MaxEntries: 100}) + prime(st) + c := anchorCtx(st) + c.ExistingBreakpoints = 4 + r := edited() + if rep := runAnchor(t, c, r); len(marks(r)) != 0 || rep.Gates["anchor_no_free_slot"] != 1 { + t.Errorf("a fifth breakpoint is a 400: %v %v", marks(r), rep.Gates) + } + // the entries have expired + st = store.NewMemory(store.Options{MaxEntries: 100}) + prime(st) + now = now.Add(10 * time.Minute) + r = edited() + if runAnchor(t, anchorCtx(st), r); len(marks(r)) != 0 { + t.Errorf("entries older than the TTL must not be trusted: %v", marks(r)) + } + // cold request: the whole prefix is rewritten anyway + st = store.NewMemory(store.Options{MaxEntries: 100}) + prime(st) + c = anchorCtx(st) + c.ColdCache = true + r = edited() + if runAnchor(t, c, r); len(marks(r)) != 0 { + t.Errorf("no anchor on a cold request: %v", marks(r)) + } +} diff --git a/components/offload/commitgate_test.go b/components/offload/commitgate_test.go index e035059da..83bb3c1da 100644 --- a/components/offload/commitgate_test.go +++ b/components/offload/commitgate_test.go @@ -112,6 +112,8 @@ type gateCase struct { // TestEveryOffloadComponentIsHeldToTheCommitGate — which is the point: the invariant's whole // weakness is the caller nobody thought about. var gateExempt = map[string]string{ + "cacheanchor": "removes nothing and stashes nothing: it only adds one cache_control mark to an assistant " + + "turn, so there is no marker, stash or frozen decision for a commit gate to guard; covered by cacheanchor_test.go", "failed_run": "needs a superseded failed RUN (a later successful invocation of the same " + "command) to have anything to collapse; the shape is covered by mask, which shares " + "its commit path verbatim — tryMark, commitMark, SetMessageText, freeze", diff --git a/components/offload/mask.go b/components/offload/mask.go index 957b59353..9e85e033d 100644 --- a/components/offload/mask.go +++ b/components/offload/mask.go @@ -5,6 +5,7 @@ import ( "github.com/rossoctl/context-guru/components" "github.com/rossoctl/context-guru/expand" "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" ) func init() { components.Register("mask", newMask) } @@ -19,6 +20,7 @@ type Mask struct { keepHeadChars int mode markerMode coldCache bool + deepBatch int } type maskConfig struct { @@ -33,6 +35,10 @@ type maskConfig struct { // replacement is a pure function of (content, config), which is what makes the // sweep safe; see coldCacheDefault. ColdCache *bool `yaml:"cold_cache"` + // DeepBatch > 0 lifts the cached-prefix gate on a turn when at least this many eligible older outputs + // are waiting to be masked, so they are masked together (one cache rewrite, not one per turn). Meant + // to run with the cacheanchor component, which keeps that rewrite to the tail. 0 = off (tail only). + DeepBatch int `yaml:"deep_batch"` } func newMask(raw []byte) (components.Component, error) { @@ -45,7 +51,7 @@ func newMask(raw []byte) (components.Component, error) { keepHead = *cfg.KeepHeadChars } return &Mask{keepRecent: cfg.KeepRecent, minTokens: cfg.MinTokens, keepHeadChars: keepHead, - mode: parseMarkerMode(cfg.MarkerMode), coldCache: coldCacheDefault(cfg.ColdCache)}, nil + mode: parseMarkerMode(cfg.MarkerMode), coldCache: coldCacheDefault(cfg.ColdCache), deepBatch: cfg.DeepBatch}, nil } func (Mask) Name() string { return "mask" } @@ -62,6 +68,7 @@ func (m *Mask) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, } var keys []string changed := 0 + deep := m.deepBatch > 0 && m.pendingMasks(req, tools[:len(tools)-m.keepRecent], c) >= m.deepBatch // Mask every tool output except the most recent keepRecent. for _, i := range tools[:len(tools)-m.keepRecent] { msg := &req.Input[i] @@ -102,7 +109,7 @@ func (m *Mask) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, // provably expired (cold_cache, ON by default): there is no cached prefix left to // flip, so the freeze below simply establishes the decision a turn earlier than the // tail would have. - if !c.TailOnlyCold(i, m.coldCache) && !repairLostFreeze(c, m.Name(), content) { + if !deep && !c.TailOnlyCold(i, m.coldCache) && !repairLostFreeze(c, m.Name(), content) { rep.Gate("cached_prefix") continue } @@ -134,6 +141,8 @@ func (m *Mask) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, func init() { components.RegisterFields("mask", maskConfig{}, []components.Field{ + {Key: "deep_batch", Type: components.FieldInt, Default: 0, Min: 0, + Hint: "0 = mask only in the uncached tail. N > 0 = when N or more older outputs are waiting, mask them all at once even inside the cached prefix (one cache rewrite); use with the cacheanchor component."}, {Key: "keep_recent", Type: components.FieldInt, Default: 3, Hint: "How many of the newest tool results stay verbatim. Everything older is masked; 0 masks them all."}, {Key: "min_tokens", Type: components.FieldInt, Default: 100, Min: 1, @@ -196,3 +205,26 @@ func coldCacheDefault(v *bool) bool { } return *v } + +// pendingMasks counts the older outputs that would be masked now but are not yet (no frozen decision). +func (m *Mask) pendingMasks(req *bschemas.BifrostChatRequest, older []int, c *components.Ctx) int { + n := 0 + for _, i := range older { + msg := req.Input[i] + if !schema.Rewritable(msg) { + continue + } + content := schema.MessageText(msg) + if content == "" || schema.TextTokens(content) < m.minTokens { + continue + } + if _, skip := skipReduce(c, content); skip { + continue + } + if store.Peek(c.Store, frozenKey(c.Session, m.Name(), contentKey(content))) { + continue + } + n++ + } + return n +} diff --git a/components/offload/mask_deepbatch_test.go b/components/offload/mask_deepbatch_test.go new file mode 100644 index 000000000..e2659a4e3 --- /dev/null +++ b/components/offload/mask_deepbatch_test.go @@ -0,0 +1,39 @@ +package offload + +import ( + "strings" + "testing" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +func TestMaskDeepBatchLiftsTheCachedPrefixGateOnlyWhenEnoughOutputsWait(t *testing.T) { + big := func(i string) string { return strings.Repeat("log line "+i+" with some words in it\n", 60) } + for _, tc := range []struct { + batch string + masked int + }{{"", 0}, {"deep_batch: 5\n", 0}, {"deep_batch: 3\n", 3}} { + c, err := newMask([]byte("keep_recent: 1\nmin_tokens: 20\n" + tc.batch)) + if err != nil { + t.Fatal(err) + } + req := toolMsgs(big("a"), big("b"), big("c"), big("d")) + ctx := ctxFor(store.NewMemory(store.Options{MaxEntries: 100})) + ctx.CacheAware, ctx.MaxCachedIdx = true, len(req.Input)-1 // everything is inside the cached prefix + rep := &components.Report{Component: "mask"} + if _, err := c.(components.Offload).Offload(req, rep, ctx); err != nil { + t.Fatal(err) + } + n := 0 + for i := range req.Input { + if strings.Contains(schema.MessageText(req.Input[i]), "masked") { + n++ + } + } + if n != tc.masked { + t.Errorf("deep_batch %q: masked %d outputs, want %d (gates %v)", tc.batch, n, tc.masked, rep.Gates) + } + } +} diff --git a/docs/components/advanced-offload.md b/docs/components/advanced-offload.md index bab44389a..7a1488160 100644 --- a/docs/components/advanced-offload.md +++ b/docs/components/advanced-offload.md @@ -9,6 +9,19 @@ requirements) that the deterministic components in --- +## cacheanchor + +An edit inside the cached history normally costs a rewrite of everything after the system prompt, because +the client marks only its last message and the provider's look-back is 20 blocks. `cacheanchor` is a +pipeline-level answer for any component that edits earlier messages. It runs after the editors, compares +the request with the thread's previous forwarded request, and when they first differ at message p it puts +one cache_control mark on the assistant turn that follows the latest earlier request end whose forwarded +prefix is unchanged (and whose entry is younger than the cache lifetime). That mark reads the prefix up to +the entry and only the tail is rewritten. It adds at most one breakpoint, never beyond four counting the +client's own, skips cold requests, and does nothing for appends. It is not in any preset: add it as the +last component of a pipeline that edits history, for example `mask` with `deep_batch: N`, which masks N or +more older outputs at once inside the cached prefix. + ## extract_llm_sweep **Kind:** Offload. **Reversible:** yes (`marker_mode: full`, the default). **In presets:** `housellm`.