From b224704fb94a1461106298aa0da45ead32cc3067 Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 1 Sep 2026 21:26:31 +0300 Subject: [PATCH 01/29] feat(coref): port the co-reference index and component onto main The branch this replaces carried 97 commits against a merge base main has since moved 295 files past. Three of its general fixes landed separately (#136 schema shape validation, #137 the adjudicate verdict tool, #138 collapse's byte gate), and main absorbed the prefix-ask infrastructure independently as extract_llm_sweep. What is left is the part that was only ever about co-reference, re-cut onto main rather than rebased: internal/coref (the index), the coref component, and prefix_econ (the shared cache-write price). The pre-recut history is preserved on origin at archive/coref-compaction-pre-recut (f31b451) -- the experiment record and the refuted arms are evidence, not clutter, and a force-push must not be the only copy of them. Two things main required that the branch predated: RegisterFields for coref. main added a test asserting every registered component declares exactly its configurable YAML keys, and its motivating story is this branch's own: an account whose extract_llm ran 251 times and acted zero times because the two keys that decided it were not on the settings page. coref had a constructor and no descriptors, so it was that bug waiting to happen. Each hint carries the measurement behind the default rather than restating the field name -- including the two thresholds (closed_dist, open_reps) that are explicitly NOT measured yet, which is why cut_closed ships off. MarkKeptVerbatim lost its session argument. The branch had made keep-verbatim marks session-scoped; main's are global, so a mark written by one session exempts the same bytes in every other session sharing the store -- preferentially on content that recurs across sessions, which is exactly the content most worth cutting. That is a real lost-savings leak, but fixing it is a store key-format change in state.go affecting every offloader plus a read-both-shapes migration for marks already on disk, so it is filed separately rather than smuggled in here. The two tests that prove the leak are SKIPPED with their bodies preserved commented, not adapted: an adapted body would compile, read as a real test, and assert nothing, because both sessions would share one global mark. Signed-off-by: David Amid Assisted-By: Claude Opus 5 (1M context) Signed-off-by: DAVID AMID --- components/offload/coref.go | 467 +++++++++++++++++++++ components/offload/coref_test.go | 663 ++++++++++++++++++++++++++++++ components/offload/corefstub.go | 144 +++++++ components/offload/prefix_econ.go | 113 +++++ internal/coref/coref.go | 392 ++++++++++++++++++ internal/coref/coref_test.go | 403 ++++++++++++++++++ 6 files changed, 2182 insertions(+) create mode 100644 components/offload/coref.go create mode 100644 components/offload/coref_test.go create mode 100644 components/offload/corefstub.go create mode 100644 components/offload/prefix_econ.go create mode 100644 internal/coref/coref.go create mode 100644 internal/coref/coref_test.go diff --git a/components/offload/coref.go b/components/offload/coref.go new file mode 100644 index 00000000..a93aa3e0 --- /dev/null +++ b/components/offload/coref.go @@ -0,0 +1,467 @@ +package offload + +import ( + "math" + "strconv" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/internal/coref" + "github.com/rossoctl/context-guru/schema" + "gopkg.in/yaml.v3" +) + +func init() { components.Register("coref", newCoref) } + +// Coref is co-reference-aware compaction: at a threshold crossing it cuts the tool +// outputs that later turns never carried anything forward from, in ONE batched pass. +// Design and derivation: docs/proposals/coref-compaction.md. +// +// It differs from every other offloader here in one deliberate way, and the whole shape +// of the component follows from it: it MUTATES THE CACHED PREFIX on purpose. Age-based +// offloaders refuse to (Ctx.TailOnly) because breaking the prefix hash at index i +// cache-WRITES the suffix, at 11.5x a cache-read. So: +// +// - The cut is BATCHED. One rewrite has to serve every cut in the pass, because a +// single early cut can never repay its own rewrite on tokens: at 5k cut from 20% +// depth of a 150k transcript the break-even is T > 276 turns. Batched, 60k of the +// same transcript needs T > 23, which a long session actually reaches. Hence +// min_batch_frac, and hence "rare and certain" rather than "per output per turn". +// - The spend is BUDGETED and reported, not incidental (rewrite_budget). A component +// that spends cache-writes on purpose has to be answerable for how many. +// - The decision is LATCHED and replayed, never re-derived. See the freeze note in +// Offload; this is the constraint that rules out repairLostFreeze. +// +// The default cut set is `unreferenced` only — outputs no later turn exactly reuses +// anything from. That is the honest ceiling of a zero-LLM implementation and needs no +// calibrated threshold. `closed` (the case-A large cut: referenced once or twice, long +// ago) is off by default because its two thresholds are the OUTPUT of the measurement +// pass in deploy/harbor/coref.py, which has not yet run on real traffic; shipping a +// guessed closed_dist would be shipping the one number the proposal says must be +// measured. Turn it on with cut_closed once there are numbers. +type Coref struct { + trigger components.Trigger + minTokens int + closedDist int + openReps int + minLaterTurns int + cutUnreferenced bool + cutClosed bool + rewriteBudget int + minBatchFrac float64 + breakEven bool + keepHeadChars int + mode markerMode +} + +type corefConfig struct { + Trigger components.Trigger `yaml:"trigger"` + // MinTokens is the per-output floor; matches coref.py's min_output default so the + // component and the measurement consider the same population. + MinTokens int `yaml:"min_tokens"` + // ClosedDist / OpenReps are the open-vs-closed thresholds. Defaults mirror + // coref.py's, and are placeholders until it runs on captured traffic. + ClosedDist int `yaml:"closed_dist"` + OpenReps int `yaml:"open_reps"` + // MinLaterTurns is the opportunity floor: an output with fewer model turns after it + // than this is never cut, because it has not yet had a chance to be referenced. + MinLaterTurns *int `yaml:"min_later_turns"` + // CutUnreferenced / CutClosed select the cut set (see Coref). + CutUnreferenced *bool `yaml:"cut_unreferenced"` + CutClosed *bool `yaml:"cut_closed"` + // RewriteBudget caps prefix-rewrite passes per session. 0 disables the component's + // only cache-spending path entirely (replay of already-latched decisions continues). + RewriteBudget *int `yaml:"rewrite_budget"` + // MinBatchFrac is the batching constraint: the pass must cut at least this fraction + // of the request before it is worth a rewrite. + // + // The default was 0.15, derived from the illustrative arithmetic in the proposal's §4 + // and never checked against how much mass is actually available. Measured, Tier-1 + // matching finds a mean 4.4% of the request (`unreferenced`) or 9.6% (`+closed`) on real + // long sessions — so 0.15 admitted ONE of nineteen sessions past the agent's compaction + // threshold, and zero at the shipped cut set. A gate no traffic can clear is not a + // conservative default, it is an off switch that looks like a threshold. 0.05 admits + // 16/19; the honest position is that the right value is an experimental result and this + // is a starting point, not a claim. + MinBatchFrac *float64 `yaml:"min_batch_frac"` + // BreakEven applies the S*T > 11.5*W inequality with an estimated T. Ignored when + // the context window is unknown, like every other fraction-based threshold here. + BreakEven *bool `yaml:"break_even"` + // KeepHeadChars leaves a one-line peek inside the marker so the model knows what was + // cut without a blind expand round-trip; 0 disables. + KeepHeadChars *int `yaml:"keep_head_chars"` + MarkerMode string `yaml:"marker_mode"` // full (default) | summary | off +} + +func newCoref(raw []byte) (components.Component, error) { + cfg := corefConfig{MinTokens: 300, ClosedDist: corefClosedDistDefault, OpenReps: corefOpenRepsDefault} + if len(raw) > 0 { + if err := yaml.Unmarshal(raw, &cfg); err != nil { + return nil, err + } + } + cf := &Coref{ + trigger: cfg.Trigger, + minTokens: cfg.MinTokens, + closedDist: cfg.ClosedDist, + openReps: cfg.OpenReps, + minLaterTurns: corefMinLaterDefault, + cutUnreferenced: true, + cutClosed: false, + rewriteBudget: 3, + minBatchFrac: 0.05, + breakEven: true, + keepHeadChars: 96, + mode: parseMarkerMode(cfg.MarkerMode), + } + if cfg.MinLaterTurns != nil { + cf.minLaterTurns = *cfg.MinLaterTurns + } + if cfg.CutUnreferenced != nil { + cf.cutUnreferenced = *cfg.CutUnreferenced + } + if cfg.CutClosed != nil { + cf.cutClosed = *cfg.CutClosed + } + if cfg.RewriteBudget != nil { + cf.rewriteBudget = *cfg.RewriteBudget + } + if cfg.MinBatchFrac != nil { + cf.minBatchFrac = *cfg.MinBatchFrac + } + if cfg.BreakEven != nil { + cf.breakEven = *cfg.BreakEven + } + if cfg.KeepHeadChars != nil { + cf.keepHeadChars = *cfg.KeepHeadChars + } + return cf, nil +} + +func (Coref) Name() string { return "coref" } +func (Coref) Enabled(*components.Ctx) bool { return true } + +// plannedCut is one accepted candidate, held until the whole batch clears its gates — +// nothing is stashed or rewritten before then, so a batch that fails a gate leaves the +// request byte-identical. +type plannedCut struct { + idx int + original string + newText string + key string + eff markerMode + saved int +} + +func (cf *Coref) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, c *components.Ctx) ([]string, error) { + // The trigger is pure request shape, so ask before paying for the index. Replay + // (below) is NOT gated on it: a latched cut must be re-applied on every turn whether + // or not this turn would have decided to cut anything, or the output flips + // cut→full→cut and churns the very cache this component is budgeting. + fires := cf.trigger.Fires(req, c.CtxWindow) + + // Index the PRISTINE request, before any replay rewrites a message. Two reasons, and + // the second is the load-bearing one: the agent re-sends originals every turn, so the + // pristine transcript is the stable input the offline measurement also reads — and if + // the index were built after replay, an earlier cut would remove identifiers from the + // exclusion sets and silently reclassify unrelated outputs. That is history dependence + // on our OWN past output, which is how a "keep" turns into a "cut" turns into a + // different set of bytes at the same index. + classes := map[int]coref.Class{} + if fires { + for _, r := range coref.Index(flattenForCoref(req), cf.trigger.OutputFloor(c.CtxWindow, cf.minTokens), schema.TextTokens) { + classes[r.Idx] = coref.Classify(r, cf.closedDist, cf.openReps, cf.minLaterTurns) + } + } + + pristineTokens := schema.MessagesTokens(req) + var keys []string + changed := 0 + + // Replay latched decisions, on every tool output, every turn, at any depth. + replayed := map[int]bool{} + for _, i := range toolIndices(req) { + m := &req.Input[i] + if !schema.Rewritable(*m) || schema.MessageText(*m) == "" { + continue + } + if fk, _, ok := reapplyFrozen(c, cf.Name(), m); ok { + replayed[i] = true + changed++ + keys = append(keys, fk...) + } + } + + if !fires { + rep.Gate("trigger") + if changed == 0 { + rep.Skipped = true + } + return keys, nil + } + + // A new pass spends a cache-write. Ask before planning one. + spent := corefRewrites(c) + if cf.rewriteBudget <= 0 || spent >= cf.rewriteBudget { + rep.Gate("rewrite_budget") + if changed == 0 { + rep.Skipped = true + } + return keys, nil + } + + plan, err := cf.planCuts(req, rep, c, classes, replayed) + if err != nil { + return keys, err + } + if len(plan) == 0 { + if changed == 0 { + rep.Skipped = true + } + return keys, nil + } + + saved := 0 + for _, p := range plan { + saved += p.saved + } + // Batching: one rewrite must serve the whole pass, so the pass has to be big enough + // to be worth one. A batch below the floor is not a small win, it is a loss. + if cf.minBatchFrac > 0 && pristineTokens > 0 && + float64(saved) < cf.minBatchFrac*float64(pristineTokens) { + rep.Gate("batch_too_small") + if changed == 0 { + rep.Skipped = true + } + return keys, nil + } + if cf.breakEven { + if _, _, ok := cf.breakEvenTurns(req, plan, c); !ok { + rep.Gate("break_even") + if changed == 0 { + rep.Skipped = true + } + return keys, nil + } + } + + // Commit the whole batch, then charge the session exactly one rewrite for it. + for _, p := range plan { + commitMark(c, rep, p.eff, p.key, p.original) + schema.SetMessageText(&req.Input[p.idx], p.newText) + // Latch. From here the bytes for this content are fixed for the session: the next + // turn replays them rather than re-deciding, because the decision is a function of + // the transcript that existed when it was taken, not of the content alone. + // + // This is also why coref must NOT consult repairLostFreeze, which mask and + // failed_run legitimately do. That repair re-derives a lost decision at depth on the + // grounds that the replacement is a pure function of (content, config) and so + // reproduces the bytes the provider already cached. A co-reference decision is + // history-dependent by construction — re-deriving it against a longer transcript can + // yield a DIFFERENT class and different bytes, which is precisely the prefix flip the + // repair exists to avoid. A lost coref freeze therefore just declines. + freeze(c, cf.Name(), p.original, p.newText) + changed++ + if p.key != "" { + keys = append(keys, p.key) + } + } + setCorefRewrites(c, spent+1) + + if changed == 0 { + rep.Skipped = true + } + return keys, nil +} + +// planCuts builds the batch without side effects: every candidate is size-checked with +// its marker included (tryMark stashes nothing), so a batch that later fails a gate +// leaves the request untouched. +func (cf *Coref) planCuts(req *bschemas.BifrostChatRequest, rep *components.Report, c *components.Ctx, + classes map[int]coref.Class, replayed map[int]bool) ([]plannedCut, error) { + + floor := cf.trigger.OutputFloor(c.CtxWindow, cf.minTokens) + var plan []plannedCut + for _, i := range toolIndices(req) { + if replayed[i] { + continue // already latched; counted above, never re-decided + } + m := req.Input[i] + if !schema.Rewritable(m) { + rep.Gate("non_text_blocks") + continue + } + content := schema.MessageText(m) + if content == "" || schema.TextTokens(content) < floor { + rep.Gate("below_min_tokens") + continue + } + if skipReduce(c, content) { + rep.Gate("marker_or_kept_verbatim") + continue + } + class, ok := classes[i] + if !ok { + rep.Gate("not_indexed") // below the index floor, or not a recorded output + continue + } + if !(class == coref.Unreferenced && cf.cutUnreferenced) && + !(class == coref.Closed && cf.cutClosed) { + rep.Gate("class_" + string(class)) + continue + } + // The marker states WHAT was removed and never why it was safe to remove. The + // earlier wording ("no later turn referred back to it") asserted the one claim that + // is false whenever the reference was transformed or semantic — tiers 2 and 3, which + // this index cannot see — so it read as reassurance and discouraged the expand call + // that would have repaired the mistake. Since only the model can initiate recovery, a + // marker that talks it out of recovering is worse than an opaque one. + note := "[tool output compacted" + if stub := corefStub(content); stub != "" { + // Structured content: describe the shape, which is addressable. "200 records, + // fields: address, id, name" tells a model hunting for an address that this is + // where addresses live; a head peek of one arbitrary row does not. + note += "; " + stub + } else if peek := headPeek(content, cf.keepHeadChars); peek != "" { + note += "; starts: " + peek + } + note += "] " + newText, key, eff, ok := tryMark(c, cf.mode, content, " [full output: call "+expand.ToolName+"]", + func(tok string) string { return note + tok }) + if !ok { + rep.Gate("marker_no_win") + continue + } + plan = append(plan, plannedCut{ + idx: i, original: content, newText: newText, key: key, eff: eff, + saved: schema.TextTokens(content) - schema.TextTokens(newText), + }) + } + return plan, nil +} + +// breakEvenTurns applies the inequality from the proposal's §4: +// +// cost = W x (2.50 - 0.20) = 11.5 x W (in cache-read-equivalents) +// benefit = S x T x 0.20 = S x T +// worth it when S x T > 11.5 x W +// +// S is what the batch cuts; W is the suffix this cut forces the provider to re-write — +// counted from the shallowest cut index to the CACHED boundary, because content past it +// was never cached and would be written on this turn regardless; T is how many more +// turns the session has to collect the saving on, which is the quantity nobody has, so +// it is estimated from how fast the transcript has been growing. +// +// The consequence is counter-intuitive and worth stating: firing at 90% of the window +// means T is nearly zero, i.e. paying a rewrite for a saving that will be collected +// once. The profitable moment to compact is EARLIER than the moment of maximum pressure. +// +// Returns (needed T, estimated T, whether it clears). Always clears when the context +// window is unknown — same convention as every fraction-based threshold here: an +// unresolvable threshold imposes no constraint rather than silently disabling the pass. +func (cf *Coref) breakEvenTurns(req *bschemas.BifrostChatRequest, plan []plannedCut, c *components.Ctx) (need, have int, ok bool) { + saved := 0 + shallowest := len(req.Input) + for _, p := range plan { + saved += p.saved + if p.idx < shallowest { + shallowest = p.idx + } + } + // The arithmetic itself lives in prefix_econ.go, shared with extract_llm's + // allow_cached_prefix path so the two components cannot price the same cache-write + // differently. + return prefixRewritePays(req, saved, shallowest, c) +} + +// flattenForCoref projects a request onto the neutral message list internal/coref +// indexes, 1:1 with req.Input indices so a Record points back at its message. +// +// The split is what defines a reference: a tool message is MASS (Results), everything +// else is a reference-bearing SURFACE (Texts) — prose plus the tool-call name and +// arguments, which is where a model names the path/symbol/id it took from an earlier +// output. A later tool result echoing a token is the environment repeating itself, not +// the model using the value, so it never counts as a reference. +func flattenForCoref(req *bschemas.BifrostChatRequest) []coref.Message { + out := make([]coref.Message, len(req.Input)) + for i := range req.Input { + m := req.Input[i] + if m.Role == bschemas.ChatMessageRoleTool { + id := "" + if m.ChatToolMessage != nil && m.ChatToolMessage.ToolCallID != nil { + id = *m.ChatToolMessage.ToolCallID + } + out[i] = coref.Message{Results: []coref.Result{{ID: id, Text: schema.MessageText(m)}}} + continue + } + texts := []string{} + if t := schema.MessageText(m); t != "" { + texts = append(texts, t) + } + if m.ChatAssistantMessage != nil { + for _, tc := range m.ChatAssistantMessage.ToolCalls { + name := "" + if tc.Function.Name != nil { + name = *tc.Function.Name + } + texts = append(texts, name+" "+tc.Function.Arguments) + } + } + out[i] = coref.Message{Texts: texts} + } + return out +} + +// --- per-session rewrite budget --------------------------------------------- +// +// The one number this component is answerable for. Every other offloader's cache +// discipline is "never touch the prefix"; coref's is "touch it at most N times", so N +// has to be counted somewhere durable rather than inferred from the savings. + +func corefRewritesKey(session string) string { return "cg:coref:rw:" + session } + +func corefRewrites(c *components.Ctx) int { + b, ok := c.Store.Get(corefRewritesKey(c.Session)) + if !ok { + return 0 + } + n, err := strconv.Atoi(string(b)) + if err != nil || n < 0 { + // Unreadable counter reads as EXHAUSTED, not as zero. Fail-open here means fail + // open on the request (which is unaffected — no cut is taken), not fail open on an + // unbounded cache spend. + return math.MaxInt32 + } + return n +} + +func setCorefRewrites(c *components.Ctx, n int) { + c.Store.Put(corefRewritesKey(c.Session), []byte(strconv.Itoa(n))) +} + +func init() { + components.RegisterFields("coref", corefConfig{}, append([]components.Field{ + {Key: "min_tokens", Type: components.FieldInt, Default: 300, Min: 1, + Hint: "Per-output floor for entering the co-reference index. Matches deploy/harbor/coref.py's min_output so the component and the offline measurement consider the SAME population — change one and the calibration no longer describes the code."}, + {Key: "closed_dist", Type: components.FieldInt, Default: corefClosedDistDefault, Min: 1, + Hint: "How many messages ago the last literal reference must be before an output counts as CLOSED. A placeholder, not a result: this is one of the two numbers deploy/harbor/coref.py exists to measure on captured traffic, and it has not run on real traffic yet. Only consulted when cut_closed is on."}, + {Key: "open_reps", Type: components.FieldInt, Default: corefOpenRepsDefault, Min: 1, + Hint: "How many literal references keep an output OPEN regardless of age. The second of the two uncalibrated thresholds; see closed_dist. Only consulted when cut_closed is on."}, + {Key: "min_later_turns", Type: components.FieldInt, Default: corefMinLaterDefault, Min: 0, + Hint: "OPPORTUNITY FLOOR: an output with fewer of the model's turns after it than this is never cut, because it has not yet had a chance to be referenced — so 'unreferenced' says nothing about it. This is the guard that separates 'nothing reused it' from 'nothing has happened yet', and removing it makes every fresh output look spent."}, + {Key: "cut_unreferenced", Type: components.FieldBool, Default: true, + Hint: "Cut outputs no later turn literally reuses anything from. ON by default: it is the honest ceiling of a zero-LLM implementation and needs no calibrated threshold. Measured on held-out future as ground truth it removed 11.8% at 95% live-kept — the best discriminator of the ten arms tried, model arms included."}, + {Key: "cut_closed", Type: components.FieldBool, Default: false, + Hint: "Also cut outputs referenced once or twice, long ago (the 'closed' case). OFF by default because its two thresholds (closed_dist, open_reps) are the OUTPUT of a measurement pass that has not run on real traffic — shipping a guessed closed_dist would ship the one number the design says must be measured. Offline it removed 26.5% but dropped live content 21% of the time against 11%. Turn it on when there are numbers."}, + {Key: "rewrite_budget", Type: components.FieldInt, Default: 3, Min: 0, + Hint: "Cap on prefix-rewrite passes per session — this component's only cache-spending path. 0 disables new cuts entirely while REPLAY of already-latched decisions continues, which is what keeps a session's earlier saving from being undone. A component that spends cache-writes on purpose has to be answerable for how many."}, + {Key: "min_batch_frac", Type: components.FieldFloat, Default: 0.05, Min: 0, + Hint: "The pass must cut at least this fraction of the request before a rewrite is worth taking. One rewrite serves every cut in the pass, because a single early cut cannot repay its own rewrite: 5k cut at 20% depth of a 150k transcript needs T > 276 turns, while 60k of the same transcript needs T > 23. The old 0.15 default came from illustrative arithmetic and admitted ONE of nineteen real long sessions (zero at the shipped cut set) — a gate no traffic can clear is an off switch that looks like a threshold. 0.05 admits 16/19, and is a starting point rather than a claim."}, + {Key: "break_even", Type: components.FieldBool, Default: true, + Hint: "Apply S*T > 11.5*W with an estimated T before spending the rewrite. Shared with extract_llm_sweep's econ trigger rather than restated, so the two components cannot price the same cache-write differently. Ignored when the context window is unknown, like every other fraction-based threshold here: an unresolvable threshold imposes no constraint rather than silently disabling the pass."}, + {Key: "keep_head_chars", Type: components.FieldInt, Default: 96, Min: 0, + Hint: "Leave a one-line peek inside the marker so the model can tell WHICH marker holds what it wants without a blind expand round-trip. 0 disables. This does not change whether a wrong cut is recoverable — the stash always holds the bytes — it changes whether the model can find the right marker on the first try instead of expanding several or giving up."}, + markerModeField(), + }, components.TriggerFields("trigger")...)) +} diff --git a/components/offload/coref_test.go b/components/offload/coref_test.go new file mode 100644 index 00000000..eb31888c --- /dev/null +++ b/components/offload/coref_test.go @@ -0,0 +1,663 @@ +package offload + +import ( + "fmt" + "strings" + "testing" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +// --- fixture helpers --------------------------------------------------------- + +func corefUser(text string) bschemas.ChatMessage { + t := text + return bschemas.ChatMessage{Role: bschemas.ChatMessageRoleUser, + Content: &bschemas.ChatMessageContent{ContentStr: &t}} +} + +// corefAsst builds a model turn: prose plus a tool call, which together are the +// reference-bearing surface the index reads. +func corefAsst(text, name, args string) bschemas.ChatMessage { + t, n := text, name + return bschemas.ChatMessage{ + Role: bschemas.ChatMessageRoleAssistant, + Content: &bschemas.ChatMessageContent{ContentStr: &t}, + ChatAssistantMessage: &bschemas.ChatAssistantMessage{ + ToolCalls: []bschemas.ChatAssistantMessageToolCall{ + {Function: bschemas.ChatAssistantMessageToolCallFunction{Name: &n, Arguments: args}}, + }, + }, + } +} + +func corefTool(id, text string) bschemas.ChatMessage { + t, i := text, id + return bschemas.ChatMessage{Role: bschemas.ChatMessageRoleTool, + Content: &bschemas.ChatMessageContent{ContentStr: &t}, + ChatToolMessage: &bschemas.ChatToolMessage{ToolCallID: &i}} +} + +// corefBody is a distinct multi-line output; distinct because shared filler would be +// discarded as session boilerplate by the index. +func corefBody(tag string) string { + var b strings.Builder + for i := 0; i < 60; i++ { + fmt.Fprintf(&b, "%4d\t%s_line_%d = compute_%s_%d(arg_%d)\n", i, tag, i, tag, i, i) + } + return b.String() +} + +// Sentinels that say whether an output is still verbatim. They sit at the END of their +// output on purpose: the marker carries a head peek of what it replaced, so a sentinel at +// the head survives the cut and "is it still there?" stops meaning "was it kept?". +const ( + corefNovelUsed = "TOKEN_GRACE_SECONDS_41ab" // introduced by the read, then carried forward + corefNovelUnused = "TREE_SCAN_MARKER_9d7c" // introduced by the listing, never used again + corefNovelFresh = "FRESH_SCAN_MARKER_5e1f" // introduced by a later listing, never used +) + +// corefReq is a transcript with exactly two large tool outputs: index 2 is REFERENCED by +// the following model turn (must survive) and index 5 is referenced by nothing (the cut). +func corefReq() *bschemas.BifrostChatRequest { + return &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + corefUser("Fix the failing test test_auth_expiry in src/auth.py"), + corefAsst("Reading the auth module.", "Read", `{"path":"src/auth.py"}`), + corefTool("t1", "src/auth.py\n"+corefBody("auth")+"\n"+corefNovelUsed+" = 0\n"), + corefAsst("The bug is "+corefNovelUsed+"; it must be 300.", "Edit", + `{"path":"src/auth.py","old":"`+corefNovelUsed+` = 0"}`), + corefAsst("Surveying the tree.", "Bash", `{"cmd":"ls -R"}`), + corefTool("t2", corefBody("tree")+"\n"+corefNovelUnused+"\n"), + }} +} + +// corefWithSecondListing appends a fresh model turn plus a second unreferenced output — +// a new candidate for a pass that must be turned away. +func corefWithSecondListing() *bschemas.BifrostChatRequest { + req := corefReq() + req.Input = append(req.Input, + corefAsst("Listing again.", "Bash", `{"cmd":"ls /srv"}`), + corefTool("t3", corefBody("srv")+"\n"+corefNovelFresh+"\n"), + ) + return req +} + +const ( + corefRefIdx = 2 // referenced output — must be left verbatim + corefCutIdx = 5 // unreferenced output — the cut + corefFreshIdx = 7 // the second listing's output (see corefWithSecondListing) +) + +func corefFor(t *testing.T, extraYAML string) *Coref { + t.Helper() + // The batch floor is the gate under test in exactly one case, so default it out of the + // way here rather than repeating it — and never define it twice (yaml rejects that). + // Defaults that would otherwise dominate every case: the batch floor, and the + // opportunity floor (the fixture is a short transcript whose cut candidate sits at the + // tail). Each is the gate under test in exactly one place, so a default is only added + // when the case does not set it — yaml rejects a duplicated key. + base := "min_tokens: 20\n" + for k, v := range map[string]string{"min_batch_frac": "0", "min_later_turns": "0"} { + if !strings.Contains(extraYAML, k) { + base += k + ": " + v + "\n" + } + } + comp, err := newCoref([]byte(base + extraYAML)) + if err != nil { + t.Fatal(err) + } + return comp.(*Coref) +} + +func corefCtx(st store.Store) *components.Ctx { + return &components.Ctx{Session: "s", Store: st} +} + +// --- behaviour --------------------------------------------------------------- + +// The deterministic ceiling: cut what nothing referred back to, keep what was referred +// to. Nothing else in the request may move. +func TestCorefCutsOnlyUnreferencedOutputs(t *testing.T) { + cf := corefFor(t, "") + req, orig := corefReq(), corefReq() + c := corefCtx(store.NewMemory(store.Options{})) + var rep components.Report + + keys, err := cf.Offload(req, &rep, c) + if err != nil { + t.Fatal(err) + } + if rep.Skipped { + t.Fatal("component skipped; expected the unreferenced output to be cut") + } + if got := schema.MessageText(req.Input[corefRefIdx]); got != schema.MessageText(orig.Input[corefRefIdx]) { + t.Errorf("the REFERENCED output was modified; it is load-bearing:\n%q", got) + } + cut := schema.MessageText(req.Input[corefCutIdx]) + if cut == schema.MessageText(orig.Input[corefCutIdx]) { + t.Fatal("the unreferenced output was not cut") + } + if !strings.Contains(cut, "tool output compacted") { + t.Errorf("marker does not say what happened: %q", cut) + } + assertMarkerMakesNoSafetyClaim(t, cut) + for i := range req.Input { + if i == corefCutIdx { + continue + } + if schema.MessageText(req.Input[i]) != schema.MessageText(orig.Input[i]) { + t.Errorf("message %d changed; only the cut output may move", i) + } + } + // Reversible: the original must be retrievable under the returned key. + if len(keys) != 1 { + t.Fatalf("cache keys = %v, want exactly one stashed original", keys) + } + got, ok := c.Store.Get(keys[0]) + if !ok || string(got) != schema.MessageText(orig.Input[corefCutIdx]) { + t.Error("the stashed original does not round-trip; the cut is not reversible") + } + if ms := expand.ParseMarkers(cut); len(ms) != 1 || ms[0] != keys[0] { + t.Errorf("marker in the cut text = %v, want the stash key %q", ms, keys[0]) + } +} + +// Latching, and the monotonicity that pays for it. Once a cut is taken, later turns +// replay the SAME BYTES even when fresh evidence would now classify the output as open. +// Re-deciding is what rewrites the prefix a second time, so new evidence may never +// resurrect a span — keep→cut only, in one direction. +func TestCorefLatchesAndNeverResurrects(t *testing.T) { + cf := corefFor(t, "") + st := store.NewMemory(store.Options{}) + + req := corefReq() + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(st)); err != nil { + t.Fatal(err) + } + latched := schema.MessageText(req.Input[corefCutIdx]) + if latched == schema.MessageText(corefReq().Input[corefCutIdx]) { + t.Fatal("nothing was cut on the first turn") + } + + // A later turn now references the previously-unreferenced output, repeatedly and + // recently — i.e. it would classify as OPEN if the decision were re-derived. + for turn := 1; turn <= 3; turn++ { + next := corefReq() // the agent re-sends the originals every turn + for k := 0; k < turn; k++ { + next.Input = append(next.Input, + corefAsst(corefNovelUnused+" again, attempt "+fmt.Sprint(k), "Bash", `{"cmd":"ls -R"}`)) + } + var r components.Report + if _, err := cf.Offload(next, &r, corefCtx(st)); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(next.Input[corefCutIdx]); got != latched { + t.Fatalf("turn %d re-derived the decision:\n got %q\nwant %q", turn, got, latched) + } + } +} + +// The budget is the component's answer for the cache-writes it spends on purpose. Once +// spent, further passes decline — while already-latched decisions keep being replayed, +// because NOT replaying them is itself the cache-destructive move. +func TestCorefRewriteBudgetIsSpentOnceAndEnforced(t *testing.T) { + cf := corefFor(t, "rewrite_budget: 1\n") + st := store.NewMemory(store.Options{}) + + req := corefReq() + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(st)); err != nil { + t.Fatal(err) + } + latched := schema.MessageText(req.Input[corefCutIdx]) + if latched == schema.MessageText(corefReq().Input[corefCutIdx]) { + t.Fatal("nothing was cut on the first turn") + } + if n := corefRewrites(corefCtx(st)); n != 1 { + t.Errorf("rewrites charged = %d, want exactly 1 for the whole batch", n) + } + + // A second pass with a brand-new unreferenced output must decline: the budget is gone. + next := corefWithSecondListing() + var r2 components.Report + if _, err := cf.Offload(next, &r2, corefCtx(st)); err != nil { + t.Fatal(err) + } + if r2.Gates["rewrite_budget"] == 0 { + t.Error("expected the rewrite_budget gate to turn the second pass away") + } + if got := schema.MessageText(next.Input[corefFreshIdx]); !strings.Contains(got, corefNovelFresh) { + t.Error("the new output was cut despite an exhausted budget") + } + if got := schema.MessageText(next.Input[corefCutIdx]); got != latched { + t.Error("an exhausted budget stopped the replay of an already-latched decision, " + + "which is the flip the budget exists to prevent") + } + if n := corefRewrites(corefCtx(st)); n != 1 { + t.Errorf("rewrites charged = %d after a declined pass, want 1", n) + } +} + +// Batching is the reason this component exists in this shape: one rewrite has to serve +// the whole pass. A batch below the floor leaves the request byte-identical rather than +// taking a small, losing cut. +func TestCorefDeclinesABatchTooSmallToPayForItsRewrite(t *testing.T) { + cf := corefFor(t, "min_batch_frac: 0.9\n") + req, orig := corefReq(), corefReq() + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + if rep.Gates["batch_too_small"] == 0 { + t.Error("expected the batch_too_small gate") + } + if !rep.Skipped { + t.Error("a declined pass must report Skipped") + } + for i := range req.Input { + if schema.MessageText(req.Input[i]) != schema.MessageText(orig.Input[i]) { + t.Fatalf("message %d was modified by a pass that declined; planning must be side-effect free", i) + } + } +} + +// Firing at maximum pressure is firing when there is almost nothing left to collect the +// saving on. The break-even inequality has to say no there, otherwise the component pays +// a cache-write for a single turn of savings. +func TestCorefBreakEvenDeclinesAtTheWindowEdge(t *testing.T) { + cf := corefFor(t, "") + req := corefReq() + // A window barely above the current request: the estimated turns remaining collapses + // to ~0, so no cut can repay the rewrite. + c := corefCtx(store.NewMemory(store.Options{})) + c.CtxWindow = schema.MessagesTokens(req) + 1 + var rep components.Report + if _, err := cf.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + if rep.Gates["break_even"] == 0 { + t.Fatalf("expected the break_even gate at the window edge; gates=%v", rep.Gates) + } + if got := schema.MessageText(req.Input[corefCutIdx]); !strings.Contains(got, corefNovelUnused) { + t.Error("a cut was taken that cannot repay its own cache-write") + } + + // With room to run, the same request and the same cut must clear. + roomy := corefReq() + c2 := corefCtx(store.NewMemory(store.Options{})) + c2.CtxWindow = schema.MessagesTokens(roomy) * 50 + var rep2 components.Report + if _, err := cf.Offload(roomy, &rep2, c2); err != nil { + t.Fatal(err) + } + if rep2.Gates["break_even"] != 0 { + t.Errorf("break_even declined with 50x the window headroom; gates=%v", rep2.Gates) + } +} + +// coref is the one offloader that mutates the already-cached prefix on purpose — that is +// its entire function, and it is why the spend is budgeted instead of forbidden. A tail +// restriction here would make the component a no-op on exactly the transcripts it exists +// for, since by the time a session crosses the threshold the mass is all in the prefix. +func TestCorefDeliberatelyCutsInsideTheCachedPrefix(t *testing.T) { + cf := corefFor(t, "") + req := corefReq() + c := corefCtx(store.NewMemory(store.Options{})) + c.CacheAware, c.MaxCachedIdx = true, len(req.Input)-1 // everything already cached + var rep components.Report + if _, err := cf.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(req.Input[corefCutIdx]); strings.Contains(got, corefNovelUnused) { + t.Fatal("coref respected the cache tail; it is supposed to spend the rewrite, " + + "under budget, or it can never act on a long session") + } +} + +// The trigger gates only NEW decisions. Replay is unconditional, because a latched cut +// that stops being replayed flips cut→full inside the cached prefix. +func TestCorefTriggerGatesNewCutsButNotReplay(t *testing.T) { + st := store.NewMemory(store.Options{}) + open := corefFor(t, "") + req := corefReq() + var rep components.Report + if _, err := open.Offload(req, &rep, corefCtx(st)); err != nil { + t.Fatal(err) + } + latched := schema.MessageText(req.Input[corefCutIdx]) + if latched == schema.MessageText(corefReq().Input[corefCutIdx]) { + t.Fatal("nothing was cut on the first turn") + } + + // Same store, but a trigger that cannot fire on this request shape. + shut := corefFor(t, "trigger:\n min_messages: 9999\n") + next := corefWithSecondListing() + var r2 components.Report + if _, err := shut.Offload(next, &r2, corefCtx(st)); err != nil { + t.Fatal(err) + } + if r2.Gates["trigger"] == 0 { + t.Error("expected the trigger gate") + } + if got := schema.MessageText(next.Input[corefFreshIdx]); !strings.Contains(got, corefNovelFresh) { + t.Error("a new cut was taken while the trigger was shut") + } + if got := schema.MessageText(next.Input[corefCutIdx]); got != latched { + t.Error("a shut trigger suppressed the replay of a latched decision") + } + if r2.Skipped { + t.Error("a turn that replayed a latched decision did act; Skipped is wrong") + } +} + +// The kept-verbatim exemption must NOT leak across sessions. One agent expanding a config +// dump, a manifest or a schema used to exempt that byte-identical content in every session +// thereafter — permanently, silently, and preferentially on the content most worth cutting, +// because recurring-across-sessions is exactly what makes content valuable to compact. +// +// This is the negative half of TestCorefLeavesExpandedContentAlone, and it is the half that +// fails on the old key layout: the guard is real within its session and absent outside it. +func TestKeptVerbatimDoesNotLeakAcrossSessions(t *testing.T) { + // SKIPPED, NOT DELETED, and the distinction matters: this test fails on `main` because the leak is + // real there, not because the test is wrong. `main`'s keptKey(ck) carries no session, so a mark + // written by one session exempts the same bytes in every other session sharing the store — + // preferentially on content that recurs across sessions, which is exactly the content most worth + // cutting. The fix is a store key-format change in state.go affecting every offloader, plus a + // read-both-shapes migration for marks already on disk, so it does not belong to coref. Filed + // separately; un-skip with the three-argument MarkKeptVerbatim when that lands. + t.Skip("cross-session kept-verbatim scoping is a main-wide state.go fix, tracked separately") + // The body is preserved COMMENTED rather than adapted to the two-argument signature, because an + // adapted body would compile, read as a real test, and assert nothing — both sessions would share + // one global mark. Restore it verbatim when the scoped key lands. + + // cf := corefFor(t, "") + // st := store.NewMemory(store.Options{}) + // req := corefReq() + // MarkKeptVerbatim(st, "session-a", schema.MessageText(req.Input[corefCutIdx])) + + // // A DIFFERENT session sends the same bytes. It has never expanded anything, so it cannot + // // be in an expand loop and must not inherit session-a's exemption. + // var rep components.Report + // if _, err := cf.Offload(req, &rep, &components.Ctx{Session: "session-b", Store: st}); err != nil { + // t.Fatal(err) + // } + // if got := schema.MessageText(req.Input[corefCutIdx]); strings.Contains(got, corefNovelUnused) { + // t.Fatal("session-b inherited session-a's kept-verbatim exemption; the guard is leaking") + // } + + // // And the guard still holds for the session that earned it. + // req2 := corefReq() + // var rep2 components.Report + // if _, err := cf.Offload(req2, &rep2, &components.Ctx{Session: "session-a", Store: st}); err != nil { + // t.Fatal(err) + // } + // if got := schema.MessageText(req2.Input[corefCutIdx]); !strings.Contains(got, corefNovelUnused) { + // t.Fatal("session-a lost its own exemption; scoping broke the guard it was meant to keep") + // } +} + +// An empty session must not fall back to a global mark — that is the leak, reinstated. +func TestMarkKeptVerbatimIgnoresAnEmptySession(t *testing.T) { + t.Skip("same main-wide state.go fix as TestKeptVerbatimDoesNotLeakAcrossSessions") + + // st := store.NewMemory(store.Options{}) + // MarkKeptVerbatim(st, "", "content expanded by nobody in particular") + // if _, ok := st.Get(keptKey("", contentKey("content expanded by nobody in particular"))); ok { + // t.Fatal("an empty session wrote a mark; it must be a no-op") + // } +} + +// An output the agent expanded must never be re-cut: doing so just makes it expand again, +// once per turn, paying a round-trip and a cache-write each time. +func TestCorefLeavesExpandedContentAlone(t *testing.T) { + cf := corefFor(t, "") + st := store.NewMemory(store.Options{}) + req := corefReq() + MarkKeptVerbatim(st, schema.MessageText(req.Input[corefCutIdx])) + + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(st)); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(req.Input[corefCutIdx]); !strings.Contains(got, corefNovelUnused) { + t.Fatal("re-cut content the agent had expanded; that is the expand bounce loop") + } + if rep.Gates["marker_or_kept_verbatim"] == 0 { + t.Error("expected the kept-verbatim gate to record the declined candidate") + } +} + +// cut_closed is off by default: its thresholds are the OUTPUT of the measurement pass, +// so until that has run the component must not take the large case-A cut. Enabling it +// must then actually take it. +func TestCorefClosedCutIsOptIn(t *testing.T) { + req := corefReq() + // Push the reference far enough into the past that the referenced output is `closed` + // rather than `open` (recency is measured from the head). + for k := 0; k < 20; k++ { + req.Input = append(req.Input, corefAsst("thinking "+fmt.Sprint(k), "Bash", `{"cmd":"true"}`)) + } + + off := corefFor(t, "") + var rep components.Report + if _, err := off.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(req.Input[corefRefIdx]); !strings.Contains(got, corefNovelUsed) { + t.Fatal("the closed output was cut with cut_closed off (the default)") + } + if rep.Gates["class_closed"] == 0 { + t.Fatalf("expected a declined closed candidate; gates=%v", rep.Gates) + } + + on := corefFor(t, "cut_closed: true\n") + req2 := corefReq() + for k := 0; k < 20; k++ { + req2.Input = append(req2.Input, corefAsst("thinking "+fmt.Sprint(k), "Bash", `{"cmd":"true"}`)) + } + var rep2 components.Report + if _, err := on.Offload(req2, &rep2, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + got := schema.MessageText(req2.Input[corefRefIdx]) + if strings.Contains(got, corefNovelUsed) { + t.Fatalf("cut_closed: true did not take the closed cut; gates=%v", rep2.Gates) + } + assertMarkerMakesNoSafetyClaim(t, got) +} + +// An unreadable budget counter must read as EXHAUSTED. Failing open on the request (no +// cut taken) is correct; failing open on an unbounded cache spend is not. +func TestCorefUnreadableBudgetCounterDeclines(t *testing.T) { + st := store.NewMemory(store.Options{}) + st.Put(corefRewritesKey("s"), []byte("not-a-number")) + cf := corefFor(t, "") + req := corefReq() + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(st)); err != nil { + t.Fatal(err) + } + if rep.Gates["rewrite_budget"] == 0 { + t.Error("a corrupt counter must read as exhausted, not as zero spent") + } + if got := schema.MessageText(req.Input[corefCutIdx]); !strings.Contains(got, corefNovelUnused) { + t.Error("a cut was taken against an unreadable budget") + } +} + +func TestCorefEstimateTurnsRemaining(t *testing.T) { + for _, tc := range []struct { + name string + reqTokens, turns, window int + want int + }{ + {"unknown window imposes nothing", 1000, 10, 0, 0}, + {"already at the window", 1000, 10, 1000, 0}, + {"half full, 100/turn", 1000, 10, 2000, 10}, + {"early in a long session", 1000, 10, 11000, 100}, + {"no turns yet", 1000, 0, 5000, 0}, + } { + if got := estimateTurnsRemaining(tc.reqTokens, tc.turns, tc.window); got != tc.want { + t.Errorf("%s: got %d, want %d", tc.name, got, tc.want) + } + } +} + +func TestCorefEmptyRequestIsANoOp(t *testing.T) { + cf := corefFor(t, "") + req := &bschemas.BifrostChatRequest{} + var rep components.Report + keys, err := cf.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))) + if err != nil || len(keys) != 0 || !rep.Skipped { + t.Errorf("empty request: keys=%v skipped=%v err=%v", keys, rep.Skipped, err) + } +} + +// The opportunity floor, at the component level: an output too new to have been referenced +// must survive the pass. Without it a batched cut would preferentially remove the most +// RECENT context, since recency and "no references yet" are the same thing at the tail. +func TestCorefOpportunityFloorProtectsTheTail(t *testing.T) { + cf := corefFor(t, "min_later_turns: 8\n") + req := corefReq() + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(req.Input[corefCutIdx]); !strings.Contains(got, corefNovelUnused) { + t.Fatal("cut a tail output that had had no chance to be referenced") + } + if rep.Gates["class_open"] == 0 { + t.Errorf("expected the tail output to be declined as open; gates=%v", rep.Gates) + } +} + +// An output whose values the index cannot see (records of plain names/ids) must never be +// cut by the DEFAULT config. Raised in review on PR #80. +func TestCorefNeverCutsOpaqueOutputs(t *testing.T) { + people := strings.Repeat( + `[{"name":"david","id":123,"address":"foobarbaz"},{"name":"osher","id":235,"address":"banana"}]`, 60) + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + corefUser("look up the people directory"), + corefAsst("Querying.", "query_people", `{}`), + corefTool("p1", people), + corefAsst("I need to remember david 123 address.", "Bash", `{"cmd":"true"}`), + }} + for k := 0; k < 20; k++ { + req.Input = append(req.Input, corefAsst(fmt.Sprintf("step %d", k), "Bash", `{"cmd":"true"}`)) + } + cf := corefFor(t, "") + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + if got := schema.MessageText(req.Input[2]); !strings.Contains(got, "foobarbaz") { + t.Fatal("cut an output the index has no evidence about; the agent had just said it " + + "needs david's address, and the payload was never copied into a model turn") + } + if rep.Gates["class_opaque"] == 0 { + t.Errorf("expected the candidate to be declined as opaque; gates=%v", rep.Gates) + } +} + +// A marker may say WHAT was removed. It may never claim the removal was safe. +// +// The wording it replaced ("no later turn referred back to it") asserted exactly the claim +// that is false whenever the reference was transformed or semantic — tiers 2 and 3, which +// this index cannot see. Only the model can initiate recovery, so a marker that reads as +// reassurance suppresses the expand call that would have repaired the mistake. That failure +// is silent: no counter this component keeps can distinguish "never needed" from "needed and +// never asked for". +func assertMarkerMakesNoSafetyClaim(t *testing.T, marker string) { + t.Helper() + for _, claim := range []string{ + "no later turn referred back", + "survives in a later turn", + "nothing referred", + "safe to", + "not needed", + "no longer needed", + } { + if strings.Contains(strings.ToLower(marker), claim) { + t.Errorf("marker asserts its own safety (%q), which discourages recovery: %q", claim, marker) + } + } +} + +// For structured output the residue must be ADDRESSABLE — the shape, not one arbitrary row. +// An agent looking for someone's address has to be able to tell from the marker alone that +// this is the output where addresses live. +func TestCorefMarkerDescribesStructuredShape(t *testing.T) { + people := strings.Repeat( + `{"name":"david","id":123456,"address":"foobarbaz","city":"haifa"},`, 200) + body := "[" + strings.TrimSuffix(people, ",") + "]" + + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + corefUser("load the directory"), + corefAsst("Loading.", "query_people", `{"q":"all"}`), + corefTool("p1", body), + corefAsst("Loaded; moving on to unrelated work.", "Bash", `{"cmd":"true"}`), + }} + for k := 0; k < 20; k++ { + req.Input = append(req.Input, corefAsst(fmt.Sprintf("step %d", k), "Bash", `{"cmd":"true"}`)) + } + cf := corefFor(t, "") + var rep components.Report + if _, err := cf.Offload(req, &rep, corefCtx(store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + got := schema.MessageText(req.Input[2]) + if strings.Contains(got, "foobarbaz") { + t.Fatalf("not cut, so there is no marker to check; gates=%v", rep.Gates) + } + for _, want := range []string{"200 records", "address", "name"} { + if !strings.Contains(got, want) { + t.Errorf("marker omits %q, so the model cannot tell what is in here: %q", want, got) + } + } + assertMarkerMakesNoSafetyClaim(t, got) +} + +func TestCorefStub(t *testing.T) { + for _, tc := range []struct{ name, in, want string }{ + {"array of records", `[{"b":1,"a":2},{"a":3,"b":4}]`, "2 records, fields: a, b"}, + {"array of scalars", `[1,2,3]`, "3 records"}, + {"wrapped collection", `{"total":2,"rows":[{"x":1},{"x":2}]}`, "object, fields: rows, total (rows: 2 items)"}, + {"plain text has no shape", "Traceback (most recent call last):", ""}, + {"empty", "", ""}, + {"malformed json", `[{"a":`, ""}, + } { + if got := corefStub(tc.in); got != tc.want { + t.Errorf("%s: corefStub(%q) = %q, want %q", tc.name, tc.in, got, tc.want) + } + } + // Field order must be stable: the marker text is replayed byte-for-byte every later + // turn, so a map-iteration-ordered descriptor would flip the prefix and cache-write it. + first := corefStub(`[{"z":1,"a":2,"m":3}]`) + for i := 0; i < 50; i++ { + if got := corefStub(`[{"z":1,"a":2,"m":3}]`); got != first { + t.Fatalf("descriptor is not deterministic: %q vs %q", got, first) + } + } + // A wide record must not turn the marker into a schema dump. + var wide strings.Builder + wide.WriteString("[{") + for i := 0; i < 40; i++ { + if i > 0 { + wide.WriteString(",") + } + fmt.Fprintf(&wide, `"field%02d":%d`, i, i) + } + wide.WriteString("}]") + got := corefStub(wide.String()) + if !strings.Contains(got, "…+28") { + t.Errorf("wide record not truncated: %q", got) + } + if len([]rune(got)) > stubCap { + t.Errorf("descriptor exceeds stubCap: %d runes", len([]rune(got))) + } +} diff --git a/components/offload/corefstub.go b/components/offload/corefstub.go new file mode 100644 index 00000000..a5536373 --- /dev/null +++ b/components/offload/corefstub.go @@ -0,0 +1,144 @@ +package offload + +import ( + "encoding/json" + "sort" + "strconv" + "strings" +) + +// The residue a cut leaves behind, and why it is more than a cosmetic choice. +// +// Reversibility is a CAPABILITY, not a guarantee. The stash guarantees the bytes can be +// recovered; only the model can decide to recover them, by calling the expand tool. So a +// wrong cut has three outcomes, not one: +// +// 1. the model notices and expands the right marker — one round-trip plus a cache-write; +// 2. it notices something is missing but cannot tell WHICH marker holds it — several +// expands, or it gives up; +// 3. it never notices, and answers from less information than it had. +// +// Only (1) is the cost the design originally claimed. (3) is silent, and no counter this +// component keeps can see it — expand-rate measures noticed errors only, which is why +// reward is the sole instrument that detects it. +// +// What the residue can actually influence is the gap between (1) and (2): whether the +// model can tell, without expanding, that THIS marker is where the thing it wants lives. +// A head peek — the first ~96 characters — does that well for a file read or a traceback, +// where the head identifies the whole. It does it badly for a record set, where the head +// is one arbitrary row: an agent hunting for someone's address cannot tell from +// `[{"name":"david","id":123,...` whether addresses are in here at all, let alone whose. +// +// So for structured content the residue describes the SHAPE instead: how many records, and +// what fields they carry. That is addressable — "records with keys name/id/address, 200 of +// them" tells the model where to look — where a peek is merely evocative. + +// stubCap bounds the descriptor so the marker can never dominate the message it replaces +// (tryMark's never-worse check would drop the rewrite anyway, but a cut that fails to +// shrink is a wasted candidate rather than a bug). +const stubCap = 200 + +// maxStubKeys bounds how many field names the descriptor lists. Enough to identify what +// the records hold; not a schema dump. +const maxStubKeys = 12 + +// corefStub describes what was cut, in the terms most likely to let the model decide +// whether it needs it back. Returns "" when it can say nothing useful, in which case the +// caller falls back to a head peek. +// +// Deliberately structural and never evaluative: it says what the content IS, never what it +// was worth. An earlier version of this component wrote "no later turn referred back to +// it" into the marker, which is precisely the claim that is FALSE whenever the reference +// was transformed or semantic (tiers 2 and 3) — so it read as reassurance and discouraged +// the expand call that would have repaired the mistake. A marker must not talk the model +// out of recovering content. +func corefStub(content string) string { + t := strings.TrimSpace(content) + if len(t) == 0 { + return "" + } + switch t[0] { + case '[': + return stubArray(t) + case '{': + return stubObject(t) + } + return "" +} + +// stubArray describes a JSON array: its length, and the union of keys across the records +// it holds (sampled — a 10k-element array does not need a full scan to be described). +func stubArray(t string) string { + var items []json.RawMessage + if json.Unmarshal([]byte(t), &items) != nil { + return "" + } + if len(items) == 0 { + return "" + } + keys := map[string]struct{}{} + sampled := 0 + for _, it := range items { + if sampled >= 32 { + break + } + var obj map[string]json.RawMessage + if json.Unmarshal(it, &obj) != nil { + continue // scalar or nested array: no field names to report + } + sampled++ + for k := range obj { + keys[k] = struct{}{} + } + } + out := strconv.Itoa(len(items)) + " records" + if ks := sortedKeys(keys); len(ks) > 0 { + out += ", fields: " + joinKeys(ks) + } + return clipRunes(out, stubCap) +} + +// stubObject describes a JSON object by its top-level keys, and — the common shape for a +// tool that wraps its payload — the length of the one array it contains. +func stubObject(t string) string { + var obj map[string]json.RawMessage + if json.Unmarshal([]byte(t), &obj) != nil { + return "" + } + if len(obj) == 0 { + return "" + } + keys := map[string]struct{}{} + for k := range obj { + keys[k] = struct{}{} + } + out := "object, fields: " + joinKeys(sortedKeys(keys)) + // A single wrapped collection is worth counting: "rows: 400" is the fact that decides + // whether this is the output holding what the model is looking for. + for _, k := range sortedKeys(keys) { + var arr []json.RawMessage + if json.Unmarshal(obj[k], &arr) == nil && len(arr) > 0 { + out += " (" + k + ": " + strconv.Itoa(len(arr)) + " items)" + break + } + } + return clipRunes(out, stubCap) +} + +func sortedKeys(m map[string]struct{}) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) // stable order: the marker text must be byte-identical on replay + return out +} + +// joinKeys lists field names, truncating past maxStubKeys so a wide record does not turn +// the marker into a schema dump. +func joinKeys(ks []string) string { + if len(ks) > maxStubKeys { + return strings.Join(ks[:maxStubKeys], ", ") + ", …+" + strconv.Itoa(len(ks)-maxStubKeys) + } + return strings.Join(ks, ", ") +} diff --git a/components/offload/prefix_econ.go b/components/offload/prefix_econ.go new file mode 100644 index 00000000..9eda34c8 --- /dev/null +++ b/components/offload/prefix_econ.go @@ -0,0 +1,113 @@ +package offload + +import ( + "math" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" +) + +// Economics of deliberately mutating the provider's CACHED PREFIX. +// +// Every other offloader refuses to touch the prefix (Ctx.TailOnly) because breaking the +// prefix hash at index i forces the provider to cache-WRITE everything from i onward. Two +// components choose to pay that on purpose — coref, and extract_llm when +// allow_cached_prefix is set — so the price lives here rather than being reimplemented, +// slightly differently, in each of them. +// +// cost = W x (2.50 - 0.20) = 11.5 x W (in cache-read-equivalents) +// benefit = S x T x 0.20 = S x T +// worth it when S x T > 11.5 x W +// +// S is the mass removed; W is the suffix the mutation forces the provider to re-write — +// counted from the shallowest mutated index to the CACHED boundary, because content past +// that boundary was never cached and would be written this turn regardless; T is how many +// turns remain to collect the saving on, which nobody has, so it is estimated from how +// fast the transcript has been growing. +// +// The consequence is counter-intuitive and worth restating wherever this is used: firing +// at 90% of the window means T is nearly zero — paying a rewrite for a saving collected +// once. The profitable moment to compact is EARLIER than the moment of maximum pressure. + +// cacheWriteX is one cache-write in cache-read-equivalents: ($2.50 - $0.20) / $0.20 on +// Anthropic's published per-MTok prices. Shared with deploy/harbor/coref.py. +const cacheWriteX = 11.5 + +// Co-reference classifier defaults, shared by coref and by extract_llm's prefix +// pre-filter. One definition on purpose: if the two components classified the same output +// differently, the "free deterministic pre-filter" would be answering a different question +// from the component whose measurements calibrated it. Mirrors deploy/harbor/coref.py. +const ( + corefClosedDistDefault = 12 + corefOpenRepsDefault = 3 + // corefMinLaterDefault is the opportunity floor: an output with fewer model turns after + // it has not yet HAD a chance to be referenced, so "unreferenced" says nothing about it. + corefMinLaterDefault = 8 +) + +// prefixRewriteWindow reports the last message index the provider is believed to already +// hold. An unknown boundary assumes the whole transcript is cached, which over-states the +// rewrite cost rather than under-stating it. +func prefixRewriteWindow(req *bschemas.BifrostChatRequest, c *components.Ctx) int { + end := len(req.Input) - 1 + if c != nil && c.CacheAware && c.MaxCachedIdx >= 0 && c.MaxCachedIdx < end { + end = c.MaxCachedIdx + } + return end +} + +// prefixRewritePays applies S*T > 11.5*W for a mutation of `saved` tokens whose shallowest +// touched index is `shallowest`. Returns (needed T, estimated T, whether it clears). +// +// Always clears when the context window is unknown — the same convention as every +// fraction-based threshold in this package: an unresolvable threshold imposes no +// constraint rather than silently disabling the pass. +func prefixRewritePays(req *bschemas.BifrostChatRequest, saved, shallowest int, c *components.Ctx) (need, have int, ok bool) { + if c == nil || c.CtxWindow <= 0 { + return 0, 0, true + } + if saved <= 0 { + return 0, 0, false + } + end := prefixRewriteWindow(req, c) + rewritten := 0 + for j := shallowest; j <= end && j < len(req.Input); j++ { + rewritten += schema.TextTokens(schema.MessageText(req.Input[j])) + } + rewritten -= saved // the removed mass is not part of what gets written back + if rewritten <= 0 { + return 0, estimateTurnsRemaining(schema.MessagesTokens(req), modelTurns(req), c.CtxWindow), true + } + need = int(math.Ceil(cacheWriteX * float64(rewritten) / float64(saved))) + have = estimateTurnsRemaining(schema.MessagesTokens(req), modelTurns(req), c.CtxWindow) + return need, have, need <= have +} + +// estimateTurnsRemaining projects how many more turns fit before the request reaches the +// model's window, assuming the transcript keeps growing at the average rate it has so +// far. Crude on purpose: T only has to be right to an order of magnitude to separate +// "this rewrite pays for itself" from "this rewrite is charity", and every cheaper proxy +// (elapsed turns, observed step rate) is the same shape of guess. +func estimateTurnsRemaining(reqTokens, turns, window int) int { + if window <= 0 || turns <= 0 || reqTokens <= 0 || reqTokens >= window { + return 0 + } + perTurn := reqTokens / turns + if perTurn <= 0 { + return 0 + } + return (window - reqTokens) / perTurn +} + +// modelTurns counts assistant messages — the closest thing in a request to "steps taken", +// which is the unit the growth rate is per. +func modelTurns(req *bschemas.BifrostChatRequest) int { + n := 0 + for i := range req.Input { + if req.Input[i].Role == bschemas.ChatMessageRoleAssistant { + n++ + } + } + return n +} diff --git a/internal/coref/coref.go b/internal/coref/coref.go new file mode 100644 index 00000000..aebfbdf4 --- /dev/null +++ b/internal/coref/coref.go @@ -0,0 +1,392 @@ +// Package coref builds the Tier-1 co-reference index that co-reference-aware +// compaction decides from: for each tool output in a request, which identifiers that +// output INTRODUCED, and whether any later turn carried them forward. +// +// It is deliberately free of the bifrost schema, of the components package and of the +// tokenizer. The index is a pure function of a flattened message list, which is what +// lets the Go component and the offline measurement pass +// (deploy/harbor/coref.py) share ONE definition of "reference" and be checked against +// the same known-answer fixture. A component whose notion of a reference had drifted +// from the script's would be calibrated against thresholds measured for a different +// algorithm — the thresholds are the whole output of the measurement, so that drift +// would be silent and total. +// +// See docs/proposals/coref-compaction.md for what the index is FOR: §2 (the three +// tiers and the echo confound), §3 (open vs closed, and why recency is measured from +// the head of the transcript rather than from the output). +package coref + +import ( + "regexp" + "strings" +) + +// identRe matches an identifier-ish token: the things a model actually carries forward +// out of a tool output — paths, symbols, ids, hashes, error codes, line numbers. Prose +// is filtered out by distinctive below rather than by a stopword list, which does not +// survive a change of domain (or of natural language). +var identRe = regexp.MustCompile(`[A-Za-z_][A-Za-z0-9_./:\-]{2,63}|\b\d{3,}\b|\b[0-9a-f]{7,40}\b`) + +// numericRe matches a token that is purely a number (with thousands/decimal separators). +var numericRe = regexp.MustCompile(`^\d[\d,._]*$`) + +// punctEdge is the punctuation trimmed from a token's ends before it is judged. Interior +// punctuation is structure; trailing punctuation is usually a sentence mark. +const punctEdge = "._:-/" + +// distinctive keeps tokens that look like an identifier rather than an English word: after +// trimming surrounding punctuation they carry INTERIOR structure (_ . / : -), a digit, or +// CamelCase. +// +// The point is precision, not recall. A token that also occurs in ordinary prose produces a +// spurious reference, and a spurious reference makes an output look load-bearing when it is +// not — which suppresses compaction silently. Missing a real reference is the safe +// direction: it can only make the index report LESS cuttable mass than exists. +// +// All three rules below were measured rather than guessed. An earlier version also accepted +// any token of 10+ characters, any token containing punctuation anywhere, and any number of +// 3+ digits; run over real agent traffic, the top "references" it produced were +// `description`, `transparency`, `efficiency`, `conditions`, `e.g.`, `try:`, `None:` and +// `2026`, which inflated referenced mass from 51% to 71% of all tool-output tokens: +// +// - No bare length rule. Long English words are still English words, and a real +// identifier almost always carries structure, a digit, or camelCase (`src/auth.py`, +// `session_id`, `GraphStore`). One carrying none of those cannot be told from prose. +// - Punctuation must be INTERIOR. `e.g.` / `try:` / `memory.` are prose plus a sentence +// mark; trimmed they become `e.g` / `try` / `memory` and fail on their own merits, while +// `src/auth.py` and `v1/messages` are untouched. +// - A bare number needs 5+ digits or a separator. `2026` is a year and `447` is a line +// number or a count; both recur everywhere. Hashes, ids and versions survive. +func distinctive(t string) bool { + t = strings.Trim(t, punctEdge) + if len(t) < 4 { + return false + } + if numericRe.MatchString(t) { + digits := 0 + for i := 0; i < len(t); i++ { + if t[i] >= '0' && t[i] <= '9' { + digits++ + } + } + return digits >= 5 || strings.ContainsAny(t, ",._") + } + if strings.ContainsAny(t, "_./:-") { + return true + } + for i := 0; i < len(t); i++ { + if t[i] >= '0' && t[i] <= '9' { + return true + } + } + return hasCamel(t) +} + +// hasCamel reports whether t contains a lower→upper ASCII transition (camelCase). +// identRe only ever yields ASCII, so a byte scan is equivalent to the `[a-z][A-Z]` +// pattern the measurement script uses. +func hasCamel(t string) bool { + for i := 1; i < len(t); i++ { + if t[i-1] >= 'a' && t[i-1] <= 'z' && t[i] >= 'A' && t[i] <= 'Z' { + return true + } + } + return false +} + +// Idents returns the distinctive identifier tokens in s, as a set. Tokens are trimmed of +// surrounding punctuation so `memory.` and `memory` are ONE token rather than two that can +// never match each other. +func Idents(s string) map[string]struct{} { + out := map[string]struct{}{} + for _, t := range identRe.FindAllString(s, -1) { + if distinctive(t) { + out[strings.Trim(t, punctEdge)] = struct{}{} + } + } + return out +} + +// Result is one tool output under evaluation, identified by the tool-call id that +// produced it (a single turn can carry several). +type Result struct { + ID string + Text string +} + +// Message is one flattened transcript entry. Texts is the reference-BEARING surface +// (prose, plus tool-call names and arguments); Results is the mass under evaluation. +// +// Both surfaces feed the echo-exclusion set, and only Texts can constitute a +// reference — see Index. Callers build these 1:1 with their own message indices so a +// Record's Idx points back at the message it came from. +type Message struct { + Texts []string + Results []Result +} + +// Class is the verdict for one tool output. +type Class string + +// The four verdicts. Cutting Unreferenced needs no reference model beyond "nobody +// ever used this"; cutting Closed is the large, early cut that needs the thresholds. +const ( + // Opaque — this output introduced NO identifier the index can track, so there is no + // evidence either way and the only honest verdict is no opinion. Never cut. + // + // This is not a corner case, and collapsing it into Unreferenced was a real defect: + // the two look identical in the arithmetic (refs == 0) and mean opposite things. + // "Introduced 200 identifiers, nobody touched one" is evidence of deadness; + // "introduced nothing I can see" is absence of evidence. A tool returning records of + // human-readable values — `[{"name":"david","id":123,"address":"foobarbaz"}]` — yields + // no distinctive tokens at all, because short lowercase words and 3-digit numbers are + // exactly what the precision rules in distinctive exclude. Classified as Unreferenced, + // that output is cut by the DEFAULT config while the agent still needs the address. + // + // So the blind spot is now a class rather than a silent vote for deletion, and the + // asymmetry is deliberate: an opaque output costs tokens, a wrongly cut one costs an + // expand round-trip plus a cache-write and can cost the task. + Opaque Class = "opaque" + // Unreferenced — the output introduced identifiers and no later turn reused any of + // them. The safest cut on Tier 1, and blind to Tier 2 (a value that was transformed + // before being restated leaves no exact match). Never read this as "unused". + Unreferenced Class = "unreferenced" + // Closed — referenced a small number of times, and not for a long time. Whatever + // the model took survives in the turn that took it, so the original is redundant + // with content still in the request. The case-A large-cut candidate. + Closed Class = "closed" + // Open — referenced recently, or repeatedly. Still load-bearing; keep. + Open Class = "open" +) + +// Record is the per-output measurement the classifier decides from. +type Record struct { + // Idx is the caller's message index; ID the tool-call id within it. + Idx int + ID string + // SizeTokens is the output's own size — the mass a cut would recover. + SizeTokens int + // Novel counts the identifiers this output INTRODUCED (see Index). + Novel int + // Refs counts the later turns that reused at least one novel identifier. + Refs int + // RefAge is how many messages AGO the last reference was, counted from the head of + // the transcript; -1 when there was none. This is the A/B axis. The tempting + // quantities — the output's own depth, or the gap from the output to its reference — + // are both something else: "recent messages vs early messages" is a statement about + // now, so it has to be measured from now. + RefAge int + // ConsumeLag is how many messages after the output its LAST reference was; -1 when + // there was none. Reported separately from RefAge because it answers a different + // question — how long the output stayed live — and conflating the two is what makes + // a hot old span look like a cold one. + ConsumeLag int + // UsedFrac is the share of the novel identifiers the model actually carried + // forward. A low value on a referenced output is the "took one value, does not need + // the rest" pattern, measured rather than assumed. + // + // It is NOT sufficient on its own to justify a cut, and the reason is worth stating + // where it will be read. A low UsedFrac is ambiguous: it can mean the model took the + // value it needed and the remainder is chaff, or it can mean the model took an ANCHOR + // (a name, an id) precisely in order to point at a payload it never copied. Given + // `[{"name":"david","id":123,"address":"foobarbaz"}, ...]` and a model that says "I + // need to remember david 123 address", the reference is real, the payload is not in + // the model's turn, and cutting the output loses the address. + UsedFrac float64 + // LaterTurns is how many model turns follow this output — its OPPORTUNITY to be + // referenced. Near the tail this approaches zero, and an output that has not had a + // chance to be used must not be scored as unused. See Classify's minLater. + LaterTurns int +} + +// Classify applies the open/closed predicate. closedDist is the recency floor (a last +// reference NEWER than this many messages ago keeps the output open); openReps is the +// repetition ceiling (referenced at least this many times keeps it open regardless of +// age, because a span referenced repeatedly is a hot span that happens to be old). +// +// minLater is the opportunity floor: an output with fewer than this many model turns +// after it is reported Open regardless of everything else, because it has not yet HAD a +// chance to be referenced. Without it the newest outputs classify as Unreferenced purely +// for being new, and a batched pass would preferentially cut the most recent context — +// the worst possible choice, and the reason mask carries keep_recent. 0 disables. +func Classify(r Record, closedDist, openReps, minLater int) Class { + if r.Novel == 0 { + return Opaque // no evidence either way; see Opaque + } + if minLater > 0 && r.LaterTurns < minLater { + return Open // too new to have been referenced yet — absence of opportunity + } + if r.Refs == 0 { + return Unreferenced + } + if (openReps > 0 && r.Refs >= openReps) || r.RefAge < closedDist { + return Open + } + return Closed +} + +// Index computes one Record per tool output at least minOutputTokens in size, using +// tok to measure size (nil falls back to the ~4-chars/token proxy the measurement +// script uses). +// +// Two exclusions do the real work, and neither is optional: +// +// ECHO. Only identifiers the output INTRODUCED are eligible. If the agent calls +// Read(src/auth.py), the path arrives in the tool-call ARGUMENT, is echoed by the +// result, and appears again in a later Edit(src/auth.py) — an exact matcher sees a +// reference from the output to a later turn, but nothing was ever taken FROM the +// output. So a token is novel only if it appears nowhere at or before this message +// (in any surface), nor in a sibling result of the same turn. On the fixture, dropping +// this guard flips a plainly-unreferenced file read to open and halves the measured +// cuttable mass: it is the difference between a usable measurement and one that +// reports everything as load-bearing. +// +// BOILERPLATE. A token echoed by many outputs (more than max(5, outputs/4)) is +// session furniture — a banner, a prompt, a repeated header — not a carried value. +// +// Outputs below minOutputTokens get no Record but still contribute to both exclusion +// sets, because they are part of the context the model saw. +func Index(msgs []Message, minOutputTokens int, tok func(string) int) []Record { + return index(msgs, minOutputTokens, tok, true) +} + +// index is Index with the echo guard made switchable, so the test can run the negative +// control that proves the guard is what produces the result. The knob is unexported on +// purpose: priorGuard=false is a KNOWN-WRONG index (it counts the tool-call argument +// echoed by its own result as a reference), and no caller should be able to select it. +func index(msgs []Message, minOutputTokens int, tok func(string) int, priorGuard bool) []Record { + if tok == nil { + tok = approxTokens + } + n := len(msgs) + + // Per-surface token sets. refTokens is the reference-bearing surface of each + // message; resTokens the outputs, keyed by tool-call id. + refTokens := make([]map[string]struct{}, n) + resTokens := make([]map[string]map[string]struct{}, n) + nOut := 0 + for i := range msgs { + refTokens[i] = Idents(strings.Join(msgs[i].Texts, " ")) + resTokens[i] = make(map[string]map[string]struct{}, len(msgs[i].Results)) + for _, r := range msgs[i].Results { + resTokens[i][r.ID] = Idents(r.Text) + nOut++ + } + } + + // firstSeen[t] is the lowest message index at which t occurs in ANY surface, so + // "t was already in context before message i" is firstSeen[t] < i. This replaces a + // per-message snapshot of the running union — same answer, but O(distinct tokens) + // memory instead of O(messages × tokens), which matters at the transcript sizes + // this component fires on. + firstSeen := make(map[string]int) + note := func(i int, set map[string]struct{}) { + for t := range set { + if _, ok := firstSeen[t]; !ok { + firstSeen[t] = i + } + } + } + for i := range msgs { + note(i, refTokens[i]) + for _, toks := range resTokens[i] { + note(i, toks) + } + } + + // spread[t] is how many distinct outputs carry t; past a threshold it is furniture. + spread := make(map[string]int) + for i := range msgs { + for _, toks := range resTokens[i] { + for t := range toks { + spread[t]++ + } + } + } + commonAt := nOut / 4 + if commonAt < 5 { + commonAt = 5 + } + + var recs []Record + for i := range msgs { + for _, r := range msgs[i].Results { + size := tok(r.Text) + if size < minOutputTokens { + continue + } + // Sibling results of this same turn: the producing tool call normally lands in + // the previous message (and so in firstSeen), but a batched turn carries several + // results at once and they must not credit each other. + siblings := map[string]struct{}{} + for id, toks := range resTokens[i] { + if id == r.ID { + continue + } + for t := range toks { + siblings[t] = struct{}{} + } + } + novel := map[string]struct{}{} + for t := range resTokens[i][r.ID] { + if fs, ok := firstSeen[t]; priorGuard && ok && fs < i { + continue // already in context before this output existed + } + if _, ok := siblings[t]; ok { + continue + } + if spread[t] > commonAt { + continue // session furniture + } + if _, ok := refTokens[i][t]; ok { + continue // the same turn's own prose/arguments + } + novel[t] = struct{}{} + } + + rec := Record{Idx: i, ID: r.ID, SizeTokens: size, Novel: len(novel), RefAge: -1, ConsumeLag: -1} + used := map[string]struct{}{} + last := -1 + for j := i + 1; j < n; j++ { + // A later MODEL turn, judged by whether it has a model-authored surface at all + // — not by whether that surface happens to contain trackable identifiers. The + // distinction matters twice: it is the definition coref.py uses (so the two + // must agree), and a turn with no identifiers is still an opportunity that was + // declined rather than an opportunity that never existed. + if len(msgs[j].Texts) == 0 { + continue + } + rec.LaterTurns++ + hit := false + for t := range novel { + if _, ok := refTokens[j][t]; ok { + used[t] = struct{}{} + hit = true + } + } + if hit { + rec.Refs++ + last = j + } + } + if last >= 0 { + rec.RefAge = n - last + rec.ConsumeLag = last - i + } + if len(novel) > 0 { + rec.UsedFrac = float64(len(used)) / float64(len(novel)) + } + recs = append(recs, rec) + } + } + return recs +} + +// approxTokens is the ~4-chars/token proxy the offline pass uses, so an Index built +// without a real tokenizer sizes outputs the same way the measurement did. +func approxTokens(s string) int { + if n := len(s) / 4; n > 1 { + return n + } + return 1 +} diff --git a/internal/coref/coref_test.go b/internal/coref/coref_test.go new file mode 100644 index 00000000..be839cd5 --- /dev/null +++ b/internal/coref/coref_test.go @@ -0,0 +1,403 @@ +package coref + +import ( + "encoding/json" + "fmt" + "strings" + "testing" +) + +// The fixture below is the Go twin of deploy/harbor/coref_fixture.py: four tool outputs +// whose correct classification is fixed by construction. It is duplicated rather than +// loaded so this package's test has no Python dependency, and so a change to the index +// that silently disagrees with the offline measurement fails HERE — the two must share +// one definition of a reference or the thresholds the measurement produces are +// calibrated for a different algorithm. +// +// #1 src/auth.py read -> closed (novel TOKEN_GRACE_SECONDS lifted out once, early) +// #2 src/config.py read -> unreferenced (only overlap is the echoed path == the argument) +// #3 ls -R listing -> unreferenced (nothing ever comes back to it) +// #4 pytest failure -> open (novel error id reused 3x, most recently 1 turn ago) + +// filler builds per-output distinct lines. Shared filler would be dropped as session +// boilerplate, which would mask whether the novel-token logic works at all. +func filler(tag string, n int) string { + var b strings.Builder + for i := 0; i < n; i++ { + fmt.Fprintf(&b, "%4d\t%s_line_%d = compute_%s_%d(arg_%d)\n", i, tag, i, tag, i, i) + } + return b.String() +} + +// fixture returns the flattened transcript, ending on the last MODEL turn. Ending there +// (rather than on the trailing tool result) is what the offline fixture captures, and it +// is the state in which the last turn's references are visible — recency is measured from +// the head, so the two must agree on where the head is. +func fixture() []Message { + toolUse := func(name string, args map[string]any) string { + b, _ := json.Marshal(args) + return name + " " + string(b) + } + text := func(t string) Message { return Message{Texts: []string{t}} } + asst := func(t, name string, args map[string]any) Message { + return Message{Texts: []string{t, toolUse(name, args)}} + } + res := func(id, t string) Message { return Message{Results: []Result{{ID: id, Text: t}}} } + + msgs := []Message{ + text("Fix the failing test test_auth_expiry in src/auth.py"), + + // #1 closed + asst("Reading the auth module.", "Read", map[string]any{"path": "src/auth.py"}), + res("t1", "src/auth.py\n"+filler("auth", 240)+"\nTOKEN_GRACE_SECONDS = 0 # novel\n"), + asst("The bug is TOKEN_GRACE_SECONDS is 0; it must be 300.", "Edit", + map[string]any{"path": "src/auth.py", "old": "TOKEN_GRACE_SECONDS = 0", "new": "TOKEN_GRACE_SECONDS = 300"}), + res("t2", "ok"), + + // #2 unreferenced — the echo confound + asst("Checking config.", "Read", map[string]any{"path": "src/config.py"}), + res("t3", "src/config.py\n"+filler("config", 240)), + asst("Config is fine, adjusting anyway.", "Edit", map[string]any{"path": "src/config.py", "old": "a", "new": "b"}), + res("t4", "ok"), + + // #3 unreferenced + asst("Surveying the tree.", "Bash", map[string]any{"cmd": "ls -R"}), + res("t5", filler("tree", 240)), + + // #4 open + asst("Running the suite.", "Bash", map[string]any{"cmd": "pytest -q"}), + res("t6", "1 failed, 42 passed\n"+filler("pytest", 240)+"\nE AssertionError: XPIRE_DRIFT_7f3a\n"), + } + for k := 0; k < 3; k++ { + msgs = append(msgs, + asst(fmt.Sprintf("XPIRE_DRIFT_7f3a again; attempt %d.", k), "Bash", map[string]any{"cmd": "pytest -q"}), + ) + if k < 2 { // the transcript ends on the model turn, so the last result is not sent + msgs = append(msgs, res(fmt.Sprintf("r%d", k), "1 failed\nE AssertionError: XPIRE_DRIFT_7f3a\n")) + } + } + return msgs +} + +// The offline pass's defaults, so the Go index is checked at the same operating point. +const ( + testClosedDist = 12 + testOpenReps = 3 + testMinOutput = 300 + // The fixture's outputs sit near the tail of a short transcript, so the opportunity + // floor is disabled for the ground-truth cases; it has its own test below. + testMinLater = 0 +) + +func classifyFixture(t *testing.T, guard bool) map[string]Class { + t.Helper() + recs := index(fixture(), testMinOutput, nil, guard) + got := map[string]Class{} + for _, r := range recs { + got[r.ID] = Classify(r, testClosedDist, testOpenReps, testMinLater) + } + return got +} + +func TestFixtureClassification(t *testing.T) { + got := classifyFixture(t, true) + want := map[string]Class{"t1": Closed, "t3": Unreferenced, "t5": Unreferenced, "t6": Open} + if len(got) != len(want) { + t.Fatalf("recorded outputs = %v, want exactly the four above min_output", got) + } + for id, w := range want { + if got[id] != w { + t.Errorf("output %s classified %q, want %q", id, got[id], w) + } + } +} + +// TestEchoGuardIsLoadBearing is the negative control. Without the prior-vocabulary +// exclusion, the src/config.py read is scored as REFERENCED — its only later overlap is +// the path that arrived as the tool-call argument, so the output introduced nothing that +// was carried forward. An index that gets this wrong reports nearly all mass as +// load-bearing, which reads as "there is nothing to cut" rather than as a bug. +func TestEchoGuardIsLoadBearing(t *testing.T) { + if got := classifyFixture(t, true)["t3"]; got != Unreferenced { + t.Fatalf("with the guard, t3 = %q, want %q", got, Unreferenced) + } + if got := classifyFixture(t, false)["t3"]; got == Unreferenced { + t.Fatal("without the guard, t3 stayed unreferenced: the control proves nothing, " + + "so the guard is no longer what produces the result") + } +} + +// Cuttable mass (unreferenced + closed) must be materially larger with the guard on. +// This is the measurement-level statement of the control: the guard's effect is not a +// reclassified edge case, it is most of the answer. +func TestEchoGuardChangesCuttableMass(t *testing.T) { + mass := func(guard bool) (cuttable, total int) { + for _, r := range index(fixture(), testMinOutput, nil, guard) { + total += r.SizeTokens + if c := Classify(r, testClosedDist, testOpenReps, testMinLater); c != Open { + cuttable += r.SizeTokens + } + } + return cuttable, total + } + onCut, onTotal := mass(true) + offCut, offTotal := mass(false) + if onTotal != offTotal || onTotal == 0 { + t.Fatalf("total mass differs between arms (%d vs %d): the arms are not comparable", onTotal, offTotal) + } + if onCut <= offCut { + t.Errorf("cuttable mass with guard = %d/%d, without = %d/%d; the guard must INCREASE it", + onCut, onTotal, offCut, offTotal) + } +} + +func TestRecencyIsMeasuredFromTheHead(t *testing.T) { + recs := index(fixture(), testMinOutput, nil, true) + n := len(fixture()) + byID := map[string]Record{} + for _, r := range recs { + byID[r.ID] = r + } + // #1 is referenced once, by the turn immediately after it. Its RefAge must be the + // distance from the HEAD (large — the reference is ancient), while its ConsumeLag is + // small (the value was taken immediately). Swapping the two is the modelling error + // this assertion exists to catch: it makes every early output look freshly used. + r := byID["t1"] + if r.Refs != 1 { + t.Fatalf("t1 refs = %d, want 1", r.Refs) + } + if r.RefAge != n-3 { + t.Errorf("t1 RefAge = %d, want %d (messages ago, from the head)", r.RefAge, n-3) + } + if r.ConsumeLag != 1 { + t.Errorf("t1 ConsumeLag = %d, want 1 (the very next turn took the value)", r.ConsumeLag) + } + if r.RefAge <= r.ConsumeLag { + t.Error("t1 RefAge must exceed ConsumeLag here; the two axes have been conflated") + } + // An unreferenced output reports both as absent rather than as zero — zero would read + // as "referenced by the current turn", the opposite of the truth. + if u := byID["t5"]; u.RefAge != -1 || u.ConsumeLag != -1 { + t.Errorf("t5 (unreferenced) RefAge/ConsumeLag = %d/%d, want -1/-1", u.RefAge, u.ConsumeLag) + } +} + +func TestUsedFracShowsPartialConsumption(t *testing.T) { + // #1 introduced ~240 lines' worth of identifiers and the model carried exactly one + // value forward. "Took a value, does not need the rest" should therefore be visible + // as a small UsedFrac rather than assumed. + for _, r := range index(fixture(), testMinOutput, nil, true) { + if r.ID != "t1" { + continue + } + if r.Novel < 100 { + t.Fatalf("t1 novel tokens = %d, want the filler identifiers to count", r.Novel) + } + if r.UsedFrac <= 0 || r.UsedFrac > 0.05 { + t.Errorf("t1 UsedFrac = %.4f, want a small positive fraction", r.UsedFrac) + } + } +} + +func TestClassifyBoundaries(t *testing.T) { + for _, tc := range []struct { + name string + rec Record + want Class + }{ + {"no trackable identifiers is opaque, not unreferenced", Record{Novel: 0, Refs: 0, RefAge: -1}, Opaque}, + {"never referenced", Record{Novel: 20, Refs: 0, RefAge: -1, LaterTurns: 99}, Unreferenced}, + {"referenced exactly at the recency floor is closed", Record{Novel: 20, Refs: 1, RefAge: 12, LaterTurns: 99}, Closed}, + {"one message newer than the floor is open", Record{Novel: 20, Refs: 1, RefAge: 11, LaterTurns: 99}, Open}, + {"repetition keeps it open however old", Record{Novel: 20, Refs: 3, RefAge: 9999, LaterTurns: 99}, Open}, + {"just under the repetition ceiling, and old", Record{Novel: 20, Refs: 2, RefAge: 9999, LaterTurns: 99}, Closed}, + } { + if got := Classify(tc.rec, testClosedDist, testOpenReps, testMinLater); got != tc.want { + t.Errorf("%s: got %q, want %q", tc.name, got, tc.want) + } + } +} + +func TestDistinctiveRejectsProse(t *testing.T) { + // Precision matters more than recall: a prose word scored as an identifier mints a + // spurious reference, and a spurious reference silently suppresses a cut. + // + // Every entry in the second group below is a REGRESSION CASE — each was among the top + // reference-producing "identifiers" on real agent traffic before the rules in + // distinctive were tightened, and together they inflated referenced mass from 51% to 71%. + for _, w := range []string{"the", "have", "should", "config", "failing", "module"} { + if distinctive(w) { + t.Errorf("distinctive(%q) = true, want false (prose)", w) + } + } + for _, w := range []string{ + "description", "transparency", "integration", "efficiency", "conditions", + "persistent", "orientation", "effectiveness", "environment", "conversation", + "e.g.", "try:", "None:", "memory.", "2026", "447", + } { + if distinctive(w) { + t.Errorf("distinctive(%q) = true, want false (measured false positive)", w) + } + } + for _, w := range []string{ + "src/auth.py", "TOKEN_GRACE_SECONDS", "XPIRE_DRIFT_7f3a", "camelCaseName", "12345", + "v1/messages", "session_id", "GraphStore", "config.py", "claude-sonnet-5", "1.2.3", + } { + if !distinctive(w) { + t.Errorf("distinctive(%q) = false, want true (identifier)", w) + } + } +} + +// A token's surrounding punctuation must not split it in two: `memory.` at the end of a +// sentence and `memory` in a tool argument have to match, or a real reference is missed. +func TestIdentsTrimEdgePunctuation(t *testing.T) { + got := Idents("wrote src/auth.py. then read src/auth.py") + if _, ok := got["src/auth.py"]; !ok { + t.Errorf("Idents lost the trimmed form: %v", got) + } + if _, ok := got["src/auth.py."]; ok { + t.Errorf("Idents kept an untrimmed duplicate: %v", got) + } +} + +func TestSiblingResultsDoNotCreditEachOther(t *testing.T) { + // A batched turn carries several results at once. If one sibling's identifiers count + // as a reference to another's, a parallel tool call makes both look load-bearing. + body := filler("batch", 240) + msgs := []Message{ + {Texts: []string{"go"}}, + {Results: []Result{{ID: "a", Text: body}, {ID: "b", Text: body}}}, + {Texts: []string{"done"}}, + } + for _, r := range index(msgs, testMinOutput, nil, true) { + if r.Refs != 0 { + t.Errorf("output %s refs = %d, want 0 (its only overlap is its sibling)", r.ID, r.Refs) + } + } +} + +func TestBoilerplateIsNotAReference(t *testing.T) { + // A banner repeated by many outputs is furniture. Counting it makes the FIRST output + // that emitted it look referenced by every later turn that echoes it. + const banner = "=== build_harness_v2.1 /opt/ci/run.sh ===" + var msgs []Message + msgs = append(msgs, Message{Texts: []string{"start"}}) + for i := 0; i < 12; i++ { + msgs = append(msgs, + Message{Results: []Result{{ID: fmt.Sprintf("o%d", i), Text: banner + "\n" + filler(fmt.Sprintf("run%d", i), 240)}}}, + Message{Texts: []string{banner + " again"}}, + ) + } + for _, r := range index(msgs, testMinOutput, nil, true) { + if r.Refs != 0 { + t.Errorf("output %s refs = %d, want 0 (its only later overlap is the banner)", r.ID, r.Refs) + } + } +} + +func TestBelowFloorOutputsStillExclude(t *testing.T) { + // A small output gets no Record, but the identifiers it introduced must still be in + // the exclusion set: otherwise a large output re-emitting them looks like it + // introduced them, and a later mention of them looks like a reference to it. + const novel = "GRACE_WINDOW_88fa" + msgs := []Message{ + {Texts: []string{"start"}}, + {Results: []Result{{ID: "small", Text: novel}}}, + {Results: []Result{{ID: "big", Text: novel + "\n" + filler("big", 240)}}}, + {Texts: []string{"using " + novel}}, + } + for _, r := range index(msgs, testMinOutput, nil, true) { + if r.ID != "big" { + t.Fatalf("unexpected record for %q: the small output is below the floor", r.ID) + } + if r.Refs != 0 { + t.Errorf("big refs = %d, want 0: %s was introduced by the earlier small output", r.Refs, novel) + } + } +} + +func TestToolResultsAreNotReferenceBearing(t *testing.T) { + // The environment repeating a token is not the model using it. Otherwise a flaky + // command that prints the same error every turn keeps its own first output alive. + msgs := []Message{ + {Texts: []string{"start"}}, + {Results: []Result{{ID: "first", Text: "E AssertionError: DRIFT_9c2b\n" + filler("first", 240)}}}, + {Results: []Result{{ID: "second", Text: "E AssertionError: DRIFT_9c2b\n" + filler("second", 240)}}}, + } + for _, r := range index(msgs, testMinOutput, nil, true) { + if r.ID == "first" && r.Refs != 0 { + t.Errorf("first refs = %d, want 0 (only a later tool_result echoes it)", r.Refs) + } + } +} + +func TestIndexHandlesEmptyAndNil(t *testing.T) { + if recs := Index(nil, testMinOutput, nil); len(recs) != 0 { + t.Errorf("Index(nil) = %v, want none", recs) + } + if recs := Index([]Message{{}, {Results: []Result{{ID: "x", Text: ""}}}}, 0, nil); len(recs) != 1 { + t.Errorf("an empty output should still record (at the zero floor), got %v", recs) + } +} + +// An output whose values the tokenizer cannot see must come back `opaque`, never +// `unreferenced`. Both have refs == 0 and they mean opposite things: "introduced 200 +// identifiers, nobody touched one" is evidence of deadness, "introduced nothing I can see" +// is absence of evidence. Collapsing them made the DEFAULT config cut a record set the +// agent had explicitly said it still needed. +// +// Raised in review on PR #80 with exactly this shape: the agent references an ANCHOR +// (`david`, `123`) in order to point at a payload (`foobarbaz`) it never copied. +func TestRecordsOfPlainValuesAreOpaqueNotUnreferenced(t *testing.T) { + people := strings.Repeat( + `[{"name":"david","id":123,"address":"foobarbaz"},{"name":"osher","id":235,"address":"banana"}]`, 60) + msgs := []Message{ + {Texts: []string{"look up the people directory"}}, + {Texts: []string{"Querying.", "query_people {}"}}, + {Results: []Result{{ID: "t1", Text: people}}}, + {Texts: []string{"I need to remember david 123 address."}}, + } + for i := 0; i < 20; i++ { + msgs = append(msgs, Message{Texts: []string{fmt.Sprintf("unrelated step %d", i)}}) + } + recs := index(msgs, testMinOutput, nil, true) + if len(recs) != 1 { + t.Fatalf("expected one record, got %d", len(recs)) + } + if recs[0].Novel != 0 { + t.Fatalf("fixture assumption broken: Novel = %d, expected the tokenizer to see nothing "+ + "in short lowercase words and 3-digit numbers", recs[0].Novel) + } + if got := Classify(recs[0], testClosedDist, testOpenReps, testMinLater); got != Opaque { + t.Errorf("classified %q, want %q — an output the index cannot see into must not be "+ + "a silent vote to delete", got, Opaque) + } +} + +// An output near the tail has had no chance to be referenced, so scoring it as unused would +// make a batched pass preferentially cut the most RECENT context. This is why mask carries +// keep_recent, and coref needs the same idea expressed in turns. +func TestOpportunityFloorProtectsRecentOutputs(t *testing.T) { + body := filler("recent", 240) + msgs := []Message{ + {Texts: []string{"start"}}, + {Texts: []string{"Reading.", "Read {\"path\":\"x\"}"}}, + {Results: []Result{{ID: "fresh", Text: body}}}, + {Texts: []string{"ok"}}, // exactly one later model turn + } + recs := index(msgs, testMinOutput, nil, true) + if len(recs) != 1 || recs[0].Novel == 0 { + t.Fatalf("fixture assumption broken: %+v", recs) + } + if recs[0].LaterTurns != 1 { + t.Errorf("LaterTurns = %d, want 1", recs[0].LaterTurns) + } + if got := Classify(recs[0], testClosedDist, testOpenReps, 0); got != Unreferenced { + t.Errorf("with the floor disabled: got %q, want %q", got, Unreferenced) + } + if got := Classify(recs[0], testClosedDist, testOpenReps, 8); got != Open { + t.Errorf("with the floor at 8: got %q, want %q — one later turn is not an "+ + "opportunity to be referenced", got, Open) + } +} From 24a8a9ed54fac65d81aef23cc490cd065f19e6fc Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 1 Sep 2026 21:27:01 +0300 Subject: [PATCH 02/29] feat(extract_llm_sweep): fill the evidence seam and add the economic trigger main left Evidence as an explicitly empty seam ("A SEAM, deliberately empty on `main`. There is no co-reference index here") and left a note at the candidate append site saying that when PR #80 brings index-driven selection, "the index's verdict belongs in the prompt as EVIDENCE for the model to weigh (extract.AdjudicationItem.Evidence), never as a gate that pre-decides the answer." This fills that seam on exactly those terms, and adds the second trigger. THE EVIDENCE IS NOT A FILTER, which is the whole lesson of the prefix_still_referenced thinner: it removed 149,681 candidates and left about one per request, silently turning a bulk adjudication arm into the per-output shape refuted at 6% live-kept -- while the arm reported itself as bulk throughout. It was self-defeating twice: it starved the comparison, and it meant the model only ever saw what the index had already judged spent, destroying the veto on the index's exact-match blind spot that the mechanism exists to provide. So the record goes into the inventory line and the candidate set is untouched. The contract gains a paragraph, conditional on an item actually carrying evidence. main's seam comment names one hazard -- a prompt teaching the model to read counters the prompt never carries. The converse is equally real: counters carried with no explanation invite the model to invent a reading of them. Tying the paragraph to the data avoids both. It frames the index as a fallible witness in its own words, because the index matches text EXACTLY: reuse in transformed form is invisible to it, that blind spot is precisely what the model is here to cover, and a model told the index has decided has nothing left to contribute. THE SECOND TRIGGER IS THE CLAIM OF THIS BRANCH. main sweeps only in the pre-expiry window. Neither trigger contains the other: pre-expiry fires on the clock and cannot fire at all on a session whose cache keeps being refreshed -- the long agent run with the most to save -- while econ fires on mass and cannot know how much time is left, so it must clear S*T > 11.5*W first. The price lives in prefix_econ.go, shared with coref rather than restated, because two components pricing the same cache-write differently would be two answers to one question. S is an upper bound and this says so: it is the inventory's whole mass, but the model drops only some of it, and how much is unknown until after the call the test is deciding whether to make. W leans the other way (an unknown cache boundary is assumed to be the whole transcript). Neither is calibrated, so prefix_rewrite_repaid / prefix_rewrite_not_repaid is what makes the split observable rather than assumed. Both knobs default OFF, and a test asserts a disabled trigger reports no decision at all -- a gate that fires when the feature is off reads, on a dashboard, as a feature that is on and failing. VACUITY: all five mutations were verified to fail the right test -- reverting the trigger, reverting the evidence fill, removing only the contract paragraph, flipping the default on, and INTRODUCING the pre-filter. That last one initially passed, which is the finding: sweepReqStocked produces `novel=0 refs=0 later_turns=0` for every record, so a filter keyed on references removes nothing there and the guard could not fail. Hence sweepReqCoref, which contains an output later turns reuse character-for-character (refs=2) and one never mentioned again (refs=0). The guard now catches the thinner at 11 candidates against 12. Signed-off-by: David Amid Assisted-By: Claude Opus 5 (1M context) Signed-off-by: DAVID AMID --- components/offload/extract_sweep.go | 160 ++++++++++- components/offload/extract_sweep_econ_test.go | 250 ++++++++++++++++++ internal/extract/adjudicate.go | 54 ++++ 3 files changed, 457 insertions(+), 7 deletions(-) create mode 100644 components/offload/extract_sweep_econ_test.go diff --git a/components/offload/extract_sweep.go b/components/offload/extract_sweep.go index 97b62e98..110259cc 100644 --- a/components/offload/extract_sweep.go +++ b/components/offload/extract_sweep.go @@ -15,6 +15,7 @@ import ( "github.com/rossoctl/context-guru/components" "github.com/rossoctl/context-guru/expand" "github.com/rossoctl/context-guru/internal/cheapmodel" + "github.com/rossoctl/context-guru/internal/coref" "github.com/rossoctl/context-guru/internal/extract" "github.com/rossoctl/context-guru/internal/logging" "github.com/rossoctl/context-guru/metrics" @@ -62,6 +63,11 @@ type ExtractSweep struct { // sweeping() for why the window is where it is, and why its WIDTH is the one number here that // no measurement settles. preExpiry time.Duration + // evidence adds the co-reference index's record to each inventory line. See renderEvidence. + evidence bool + // econTrigger enables the SECOND trigger: fire on economics even while the cache is live. See + // econPays() for the break-even, and why one trigger is not a superset of the other. + econTrigger bool mode markerMode } @@ -90,6 +96,15 @@ type extractSweepConfig struct { MinInventory int `yaml:"min_inventory"` // PreExpirySeconds is the width of the pre-expiry window (0 = defaultPreExpiry). PreExpirySeconds int `yaml:"pre_expiry_seconds"` + // Evidence adds the co-reference index's record to each candidate's inventory line, as input the + // model weighs rather than a filter that pre-decides. OFF by default: it changes the adjudication + // CONTRACT (the prompt gains a paragraph teaching how to read the counters), and the contract is + // the part with measurements attached to it. + Evidence bool `yaml:"evidence"` + // EconTrigger adds the economic trigger alongside the pre-expiry window. OFF by default: it + // deliberately invalidates a LIVE cached prefix, which is a cost the pre-expiry trigger exists to + // avoid, and it is only worth paying when the saving is collected over enough remaining turns. + EconTrigger bool `yaml:"econ_trigger"` // BlockFallback refuses the fallback path: when the prefix ask cannot read the cache, decline // instead of asking again with the output content copied into the prompt. // @@ -205,7 +220,8 @@ func newExtractSweep(raw []byte) (components.Component, error) { return &ExtractSweep{ minTokens: cfg.MinTokens, minInventory: cfg.MinInventory, preExpiry: pre, mode: parseMarkerMode(cfg.MarkerMode), - blockFallback: cfg.BlockFallback, + blockFallback: cfg.BlockFallback, econTrigger: cfg.EconTrigger, + evidence: cfg.Evidence, }, nil } @@ -245,6 +261,80 @@ func (e *ExtractSweep) sweeping(c *components.Ctx) bool { return remaining > 0 && remaining <= e.preExpiry } +// econPays is the SECOND trigger, and the claim this component's econ_trigger mode exists to test: +// that deferring an output's removal to a deep sweep pays for itself even when the cached prefix is +// still LIVE, because the removal is collected on every remaining turn while the cache-write it forces +// is paid once. +// +// NEITHER TRIGGER IS A SUPERSET OF THE OTHER, which is why both are kept: +// +// pre-expiry fires on TIME and knows nothing about mass. It is nearly free — what it invalidates +// is about to expire anyway — but it cannot fire at all on a session whose cache keeps +// being refreshed, which is exactly the long agent run with the most to save. +// econ fires on MASS and knows nothing about the clock. It reaches those sessions, and it +// pays a real cache-write to do it, so it must clear S*T > 11.5*W first. +// +// See prefix_econ.go for the break-even itself; it is shared with coref rather than restated, because +// two components pricing the same cache-write differently would be two answers to one question. +// +// S IS AN UPPER BOUND, and this is the trigger's known optimism. S is the inventory's whole token +// mass, but the model drops only some of it, and how much is not known until after the call this test +// is deciding whether to make. The counter-bias is in W: prefixRewriteWindow assumes the WHOLE +// transcript is cached whenever the boundary is unknown, over-stating what the mutation rewrites. The +// two lean opposite ways and neither is calibrated, so read a fired econ trigger as "this batch was +// worth asking about", not as a realised saving. `prefix_rewrite_not_repaid` vs +// `prefix_rewrite_repaid` is what makes the split observable. +func (e *ExtractSweep) econPays(req *bschemas.BifrostChatRequest, c *components.Ctx, cands []sweepCand) (need, have int, ok bool) { + if !e.econTrigger || len(cands) == 0 { + return 0, 0, false + } + saved, shallowest := 0, cands[0].i + for _, cd := range cands { + saved += schema.TextTokens(cd.content) + if cd.i < shallowest { + shallowest = cd.i + } + } + return prefixRewritePays(req, saved, shallowest, c) +} + +// renderEvidence formats one output's co-reference record for the inventory line. Counts only — no +// identifier lists — because the measured win came from comparative RANKING, not from more detail, and +// every token here is paid on every candidate on every sweeping turn. +// +// The classifier's own verdict is included deliberately. It is the index stating its conclusion, which +// the model is free to overrule; that disagreement is the signal the design wants, and it is +// unavailable if the index only ships raw counters and keeps its judgement to itself. +func renderEvidence(r *coref.Record, laterTurns int) string { + if r == nil { + // No record: the output was below the index's size floor, so the index has no opinion. Say so + // plainly rather than emitting zeros, which would read as "nothing referenced it" — the one + // misreading that could turn a silent index into a drop. + return fmt.Sprintf("no index record (below size floor); later_turns=%d", laterTurns) + } + age := "never" + if r.RefAge >= 0 { + age = fmt.Sprintf("%d messages ago", r.RefAge) + } + return fmt.Sprintf("novel=%d refs=%d ref_age=%s used_frac=%.2f later_turns=%d verdict_of_index=%s", + r.Novel, r.Refs, age, r.UsedFrac, r.LaterTurns, + coref.Classify(*r, corefClosedDistDefault, corefOpenRepsDefault, corefMinLaterDefault)) +} + +// laterModelTurns counts assistant messages after index i — the opportunity an output has HAD to be +// referenced. Only used when the index has no record, where it is the one honest thing still sayable: +// an output with few later turns has not yet had a chance to be referenced, so "unreferenced" says +// nothing about it. +func laterModelTurns(req *bschemas.BifrostChatRequest, i int) int { + n := 0 + for j := i + 1; j < len(req.Input); j++ { + if req.Input[j].Role == bschemas.ChatMessageRoleAssistant { + n++ + } + } + return n +} + // sweepUnusableSamples bounds how many unparseable replies get logged in full. Process-wide, because // the question it answers — what is the model actually emitting? — is answered by the first few. // @@ -258,8 +348,12 @@ var sweepUnusableSamples atomic.Int64 const maxSweepUnusableSamples = 5 func (e *ExtractSweep) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, c *components.Ctx) ([]string, error) { - sweeping := e.sweeping(c) - if !sweeping { + // preExpiry is trigger one. Trigger two (econ) cannot be evaluated yet: it prices the candidate + // mass, and the mass is not known until the collection loop below has run. So collection runs + // whenever EITHER trigger could fire, and the econ decision is taken at the ask. + preExpiry := e.sweeping(c) + collecting := preExpiry || e.econTrigger + if !preExpiry { // NOT a return. The frozen replays below still run, and they are the reason a sweep's saving // survives past the turn that earned it: without them a later turn would re-send every // removed output verbatim, undoing the removal AND breaking the byte-stability of the prefix @@ -318,7 +412,7 @@ func (e *ExtractSweep) Offload(req *bschemas.BifrostChatRequest, rep *components } continue } - if !sweeping { + if !collecting { // Outside the window no NEW decision is taken; the replays above already ran, which is // all such a turn has to do. continue @@ -462,7 +556,24 @@ func (e *ExtractSweep) Offload(req *bschemas.BifrostChatRequest, rep *components // Phase 2: ONE ASK for every candidate. Not a batch and not a call per output — nothing is // copied per candidate, so there is nothing to divide. - if len(cands) > 0 && sweeping { + // Trigger two is decided HERE, where the mass exists. Only consulted when the pre-expiry window + // did not already fire: the two are OR'd, and pricing a rewrite the first trigger has already + // justified would let an unrepaid verdict veto a nearly-free removal. + asking := preExpiry + if !asking { + need, have, ok := e.econPays(req, c, cands) + if ok { + asking = true + rep.Event("prefix_rewrite_repaid") + } else if e.econTrigger { + rep.Gate("prefix_rewrite_not_repaid") + } + if e.econTrigger { + slog.Debug("cg.sweep.econ", "decision", ok, "needTurns", need, "haveTurns", have, + "candidates", len(cands)) + } + } + if len(cands) > 0 && asking { drop, call := e.adjudicate(req, c, rep, cands) for _, g := range call.gates { rep.Gate(g) @@ -558,14 +669,45 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component r.event("sweep_inventory_of_one") } + // FILL THE EVIDENCE SEAM. The co-reference index's record for each candidate goes into the + // inventory line as EVIDENCE for the model to weigh — never as a filter over `cands`. That + // distinction is the whole lesson of the `prefix_still_referenced` thinner documented above: a + // pre-filter left about one candidate per request, which silently turned a bulk arm into the + // per-output shape refuted at 6% live-kept, AND meant the model only ever saw what the index had + // already judged spent, destroying the veto on the index's blind spot that the mechanism exists to + // provide. Evidence preserves the veto: the index states what it saw, the model may disagree. + // + // Keyed by message index, which is what both sides already agree on — Record.Idx and sweepCand.i + // are the same coordinate. A candidate with no record is normal, not an error: the index applies + // its own size floor, and saying so beats emitting zeros that read as "nothing referenced it". + byIdx := map[int]*coref.Record{} + if e.evidence { + recs := coref.Index(flattenForCoref(req), e.minTokens, schema.TextTokens) + for i := range recs { + byIdx[recs[i].Idx] = &recs[i] + } + } items := make([]extract.AdjudicationItem, 0, len(cands)) for k := range cands { - items = append(items, extract.AdjudicationItem{ + it := extract.AdjudicationItem{ Label: k, ID: cands[k].toolID, // the wire's id, not our content key — see #123 SizeTokens: schema.TextTokens(cands[k].content), Head: extract.HeadLine(cands[k].content, extract.AdjudicationHeadChars), - }) + } + if e.evidence { + it.Evidence = renderEvidence(byIdx[cands[k].i], laterModelTurns(req, cands[k].i)) + } + items = append(items, it) + } + if e.evidence { + // Counted, because "the index had an opinion" and "the index was silent" produce the same + // inventory line length and would otherwise be indistinguishable in a run's counters. + for k := range cands { + if byIdx[cands[k].i] == nil { + r.event("evidence_no_index_record") + } + } } // The transcript, flattened, so a claimed obligation quote is VERIFIED against what the agent was // actually told rather than trusted. This is the only remaining signal that the model is @@ -978,6 +1120,10 @@ func init() { Hint: "How long before the prompt cache's believed expiry the sweep may fire. The window is where BOTH halves are cheap: the ask still reads a live cache, and the prefix it invalidates has little life left. The TTL itself is read from the request, never assumed. This WIDTH is the component's one unmeasured number — wider fires more often and invalidates more remaining TTL, narrower fires rarely, and nothing measures either side."}, {Key: "block_fallback", Type: components.FieldBool, Hint: "Decline instead of falling back when the prefix ask could not read the cache. Unset = FALSE: the fallback asks again with a bounded sample of each output copied into the prompt, which keeps the component working on a session's first turn and whenever a cache entry has gone — but pays fresh for content the cached path reads for a tenth of the price. Set true where the bill matters more than the removal. The miss is counted either way."}, + {Key: "evidence", Type: components.FieldBool, + Hint: "Add the co-reference index's record (novel/refs/ref_age/used_frac/later_turns and the index's own verdict) to each candidate's inventory line. Unset = FALSE. It is EVIDENCE the model weighs, never a filter over the candidates: a co-reference PRE-FILTER left about one candidate per request, which silently turned a bulk arm into the per-output shape refuted at 6% live-kept and meant the model only ever saw what the index had already judged spent — destroying the veto on the index's blind spot that the mechanism exists to provide. Enabling this also adds a paragraph to the adjudication contract teaching how to read the counters; a prompt carrying counters it never explains is worse than one carrying neither."}, + {Key: "econ_trigger", Type: components.FieldBool, + Hint: "Add the ECONOMIC trigger alongside the pre-expiry window: sweep a LIVE cached prefix when the removal's saving, collected over the turns estimated to remain, exceeds the cache-write it forces (S*T > 11.5*W). Unset = FALSE, because it deliberately invalidates a prefix the provider still holds. The two triggers are OR'd and neither contains the other — pre-expiry fires on the clock and cannot reach a session whose cache keeps being refreshed, which is the long run with the most to save; econ fires on mass and cannot know how much time is left. S is the inventory's whole mass and so an upper bound on the batch's real saving; read prefix_rewrite_repaid / prefix_rewrite_not_repaid rather than assuming a fired trigger banked anything."}, markerModeField(), }) } diff --git a/components/offload/extract_sweep_econ_test.go b/components/offload/extract_sweep_econ_test.go new file mode 100644 index 00000000..fc2e43be --- /dev/null +++ b/components/offload/extract_sweep_econ_test.go @@ -0,0 +1,250 @@ +package offload + +import ( + "fmt" + "strconv" + "strings" + "sync/atomic" + "testing" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/store" +) + +// THE SECOND TRIGGER IS THE CLAIM OF PR #80, so these are the tests that have to be able to fail. +// +// `main` sweeps only inside the pre-expiry window, which cannot fire at all on a session whose cache +// keeps being refreshed — the long agent run with the most to save. The econ trigger reaches those +// sessions by paying a real cache-write, so every test here asserts BOTH halves: that it fires when +// the arithmetic clears, and that it declines and says so when it does not. + +// Outside the window, with the trigger on and a batch whose mass dwarfs the suffix it rewrites, the +// sweep asks anyway. This is the behaviour `main` does not have. +func TestSweepEconTriggerFiresOutsideTheWindow(t *testing.T) { + asker := &labelAsker{verdict: "keep", needed: "a", quote: "Find the auth timeout"} + asker.cacheRead = 19595 + e := newSweep(t, "econ_trigger: true\n") + req := sweepReqStocked() + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + c.IdleMs = 30 * 1000 // plenty of TTL left: trigger one is OFF + + rep := &components.Report{} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatal(err) + } + if e.sweeping(c) { + t.Fatal("fixture is wrong: the pre-expiry window fired, so this proves nothing about econ") + } + if rep.Gates["not_in_pre_expiry_window"] == 0 { + t.Errorf("trigger one must still record that it did not fire (gates: %v)", rep.Gates) + } + if n := atomic.LoadInt64(&asker.calls); n == 0 { + t.Fatalf("the econ trigger did not ask outside the window (gates: %v events: %v)", + rep.Gates, rep.Events) + } + if rep.Events["prefix_rewrite_repaid"] == 0 { + t.Errorf("an ask happened but nothing recorded WHY it was worth it (events: %v)", rep.Events) + } + if rep.Gates["prefix_rewrite_not_repaid"] != 0 { + t.Errorf("the trigger both fired and declined: %v", rep.Gates) + } +} + +// The counter-intuitive half of the design, asserted rather than only written down: firing near the +// window's ceiling means T is nearly zero, so the rewrite is paid once and collected once. The +// profitable moment to compact is EARLIER than the moment of maximum pressure. +func TestSweepEconTriggerDeclinesWithNoTurnsLeftToCollectOn(t *testing.T) { + asker := &labelAsker{verdict: "drop", needed: "none"} + asker.cacheRead = 19595 + e := newSweep(t, "econ_trigger: true\n") + req := sweepReqStocked() + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + c.IdleMs = 30 * 1000 + // The request already fills the window, so there is no turn left to collect a saving on. + c.CtxWindow = 1000 + + rep := &components.Report{} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatal(err) + } + if n := atomic.LoadInt64(&asker.calls); n != 0 { + t.Fatalf("paid for %d ask(s) for a rewrite with no turns left to repay it", n) + } + if rep.Gates["prefix_rewrite_not_repaid"] == 0 { + t.Errorf("declined without saying why (gates: %v)", rep.Gates) + } + if rep.Events["prefix_rewrite_repaid"] != 0 { + t.Errorf("recorded a repayment it did not get: %v", rep.Events) + } +} + +// OFF BY DEFAULT, and this is the test that says the second trigger cannot change a deployment that +// did not ask for it. Without it, "we added a trigger" and "we widened everyone's sweep" are the same +// commit. +func TestSweepWithoutEconTriggerIsUnchangedOutsideTheWindow(t *testing.T) { + asker := &labelAsker{verdict: "drop", needed: "none"} + asker.cacheRead = 19595 + e := newSweep(t, "") // default: econ trigger off + req := sweepReqStocked() + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + c.IdleMs = 30 * 1000 + + rep := &components.Report{} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatal(err) + } + if n := atomic.LoadInt64(&asker.calls); n != 0 { + t.Fatalf("the default configuration asked %d time(s) outside the window", n) + } + // And it must not report on a trigger it is not running: a gate that fires when the feature is + // off reads, on a dashboard, as a feature that is on and failing. + if rep.Gates["prefix_rewrite_not_repaid"] != 0 || rep.Events["prefix_rewrite_repaid"] != 0 { + t.Errorf("a disabled trigger reported a decision (gates: %v events: %v)", rep.Gates, rep.Events) + } +} + +// The evidence seam, filled: the index's record reaches the prompt AND the prompt explains how to read +// it. `main` shipped the field empty with the note that teaching a model to read counters the prompt +// never carries is teaching it to read a field that does not exist; the converse — counters with no +// explanation — invites it to invent a reading. Both halves are asserted here. +func TestSweepEvidenceReachesTheAskAndExplainsItself(t *testing.T) { + asker := &labelAsker{verdict: "keep", needed: "a", quote: "Find the auth timeout"} + asker.cacheRead = 19595 + e := newSweep(t, "evidence: true\n") + req := sweepReqCoref() + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + + rep := &components.Report{} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatal(err) + } + ask := asker.ask() + if ask == "" { + t.Fatal("no ask was made, so there is nothing to inspect") + } + if !strings.Contains(ask, "evidence: ") { + t.Errorf("the inventory carried no evidence field:\n%s", firstLines(ask, 40)) + } + // The INDEX'S OWN NUMBERS, not just the label. On this fixture the referenced output is reused twice + // by later turns, so a run that reached the prompt with the field present but the counters empty -- + // a rendering that dropped them, or a record silently missing -- still fails here. + if !strings.Contains(ask, "refs=2") { + t.Errorf("the evidence field carried no reference count from the index:\n%s", firstLines(ask, 40)) + } + // And the orphan output's zero, which is the OTHER direction: an index that reported references for + // everything would be as useless as one that reported none. + if !strings.Contains(ask, "refs=0") { + t.Errorf("no output was reported unreferenced, so the index is not discriminating:\n%s", + firstLines(ask, 40)) + } + // A record below the index's size floor must say so rather than emit zeros, which would read as + // "nothing referenced it" -- the one misreading that could turn a silent index into a drop. + if strings.Contains(ask, "novel=0 refs=0 ref_age=never used_frac=0.00 later_turns=0") && + !strings.Contains(ask, "no index record") { + t.Error("an output with no index record was rendered as all-zeros instead of saying so") + } + if !strings.Contains(ask, `HOW TO READ THE "evidence" FIELD`) { + t.Error("the ask carries evidence counters but never explains them") + } + // The index must be presented as fallible, not authoritative. Without this the mechanism collapses + // into the pre-filter that starved three iterations: a model told the index has decided has nothing + // left to contribute, and the veto on the index's exact-match blind spot is what it is here for. + if !strings.Contains(ask, "WITNESS, NOT A JUDGE") { + t.Error("the evidence paragraph does not tell the model it may overrule the index") + } +} + +// With evidence off, NEITHER the field nor its explanation appears. Guards the pairing in both +// directions — a contract that explains a field it does not carry is the seam's original hazard. +func TestSweepWithoutEvidenceCarriesNeitherFieldNorExplanation(t *testing.T) { + asker := &labelAsker{verdict: "keep", needed: "a", quote: "Find the auth timeout"} + asker.cacheRead = 19595 + e := newSweep(t, "") + req := sweepReqStocked() + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + + rep := &components.Report{} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatal(err) + } + ask := asker.ask() + if strings.Contains(ask, "evidence: ") { + t.Error("evidence is off but the inventory carried one") + } + if strings.Contains(ask, `HOW TO READ THE "evidence" FIELD`) { + t.Error("evidence is off but the contract still explains the field") + } +} + +// EVIDENCE IS NOT A FILTER. The single most expensive mistake in this component's history was a +// co-reference pre-filter that left about one candidate per request, silently turning a bulk +// adjudication arm into the per-output shape refuted at 6% live-kept — while the arm reported itself as +// bulk throughout. This asserts the inventory the model is SHOWN is identical with the index on and +// off, so the index can only ever inform the comparison, never thin it. +func TestEvidenceDoesNotThinTheInventory(t *testing.T) { + count := func(yaml string) int { + asker := &labelAsker{verdict: "keep", needed: "a", quote: "Find the auth timeout"} + asker.cacheRead = 19595 + e := newSweep(t, yaml) + c := preExpiryCtx("s", asker, store.NewMemory(store.Options{})) + rep := &components.Report{} + if _, err := e.Offload(sweepReqCoref(), rep, c); err != nil { + t.Fatal(err) + } + return strings.Count(asker.ask(), "\n [") + } + withOut, with := count(""), count("evidence: true\n") + if withOut == 0 { + t.Fatal("fixture offered no candidates, so this proves nothing") + } + if with != withOut { + t.Fatalf("the index thinned the inventory: %d candidates with evidence, %d without. "+ + "Evidence must inform the comparison, never filter it", with, withOut) + } +} + +// sweepReqCoref is the fixture for tests whose subject is the co-reference INDEX, and it exists because +// sweepReqStocked cannot serve them: every record the index forms on that fixture is +// `novel=0 refs=0 later_turns=0`, so a filter keyed on references removes nothing there and a test +// asserting "nothing was filtered" passes without being able to fail. That vacuity was found by +// introducing the filter and watching the test stay green. +// +// So this fixture contains outputs the index can form an opinion about IN BOTH DIRECTIONS: one whose +// identifiers later turns literally reuse, and one whose identifiers are never mentioned again. +func sweepReqCoref() *bschemas.BifrostChatRequest { + var referenced, orphan strings.Builder + for i := 0; i < 400; i++ { + fmt.Fprintf(&referenced, "ORDER-%05d sku_%06d shipped from depot NORTH-%02d\n", 10000+i, 880000+i, i%12) + } + for i := 0; i < 400; i++ { + fmt.Fprintf(&orphan, "TRACE-%05d span_%06d took %dms in handler QUIET-%02d\n", 70000+i, 550000+i, i%97, i%12) + } + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + userMsg("Reconcile the shipped orders against the depot manifest."), + toolResultMsgWithID("toolu_referenced", referenced.String()), + // Later turns reuse this output's identifiers CHARACTER FOR CHARACTER, which is the only kind of + // reuse the index can see. That is the point of the fixture, and also the point of the blind spot + // the model is asked to cover. + assistantMsg("ORDER-10004 and ORDER-10007 both left depot NORTH-04; sku_880004 is the mismatch."), + toolResultMsgWithID("toolu_orphan", orphan.String()), + assistantMsg("The manifest agrees for ORDER-10004. Next I will check sku_880007."), + userMsg("keep going"), + assistantMsg("Still reconciling the remaining orders."), + }} + for i := 0; i < defaultMinInventory; i++ { + req.Input = append(req.Input, toolResultMsgWithID("toolu_filler_"+strconv.Itoa(i), + strings.Repeat("record "+strconv.Itoa(i)+" of the audit log\n", 900))) + } + req.Input = append(req.Input, assistantMsg("Summarising now.")) + return req +} + +func firstLines(s string, n int) string { + ls := strings.Split(s, "\n") + if len(ls) > n { + ls = ls[:n] + } + return strings.Join(ls, "\n") +} diff --git a/internal/extract/adjudicate.go b/internal/extract/adjudicate.go index 663cc505..f702ec5f 100644 --- a/internal/extract/adjudicate.go +++ b/internal/extract/adjudicate.go @@ -146,6 +146,41 @@ above; do not put it in your reply. Reply with ONLY a JSON array, one object per output, no prose: [{"i":