diff --git a/components/component.go b/components/component.go index 954c4d1..519603c 100644 --- a/components/component.go +++ b/components/component.go @@ -126,6 +126,14 @@ type PrefixUsage struct { CacheWrite int Fresh int Output int + // ViaTool records that the answer arrived as a tool_use for the proxy's own structured-answer + // tool rather than as reply TEXT. Reported for the same reason CacheRead is: a caller that + // declares a verdict tool in order to get a schema-shaped answer cannot otherwise tell whether it + // got one. A sweep whose tool is never used and a sweep that works look IDENTICAL in every other + // counter -- both return parsed verdicts and real savings -- because the prose parser accepts + // both. Measured: a review of PR #137 found 0 of 5 live asks using the declared tool while every + // verdict arrived through the text path, and no counter in this repo could show that. + ViaTool bool } // PrefixAsker completes `ask` as a trailing user message appended to the EXACT body this session diff --git a/components/offload/extract_sweep.go b/components/offload/extract_sweep.go index 00b991d..97b62e9 100644 --- a/components/offload/extract_sweep.go +++ b/components/offload/extract_sweep.go @@ -694,6 +694,23 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component } else if !fellBack { r.event("sweep_prefix_cache_read_ok") } + // HOW THE ANSWER ARRIVED: the proxy's structured-answer tool, or reply prose. Split because + // nothing else in this component can tell the two apart. ParseVerdicts reads a tool_use `input` + // and a JSON array in text identically -- by design, so that declaring the tool is additive -- + // which means a run where the declared tool is never touched produces the same verdicts, the same + // savings and the same gate counts as one where it is used every time. A review of PR #137 + // measured exactly that divergence live (0 of 5 asks used the tool, all 5 answered in prose) + // against that PR's claim of 6 of 6, and no published counter could adjudicate between them. + // + // Only for the PREFIX ask. The fallback calls Complete(), which has no tool to declare, so + // counting it as prose would report a shape that was never on offer. + if !fellBack { + if usage.ViaTool { + r.event("sweep_answered_via_tool") + } else { + r.event("sweep_answered_via_prose") + } + } for range items { r.event("sweep_adjudicated") } diff --git a/components/offload/extract_sweep_test.go b/components/offload/extract_sweep_test.go index fdc3826..e0db0a7 100644 --- a/components/offload/extract_sweep_test.go +++ b/components/offload/extract_sweep_test.go @@ -24,9 +24,13 @@ type fakeAsker struct { reply string cacheRead int err error - calls int64 - lastAsk atomic.Value - lastSess atomic.Value + // viaTool reports the reply as having arrived through the proxy's structured-answer tool rather + // than as text, which is the one thing about a reply the parser CANNOT infer: it reads a + // tool_use input and a JSON array in prose identically. + viaTool bool + calls int64 + lastAsk atomic.Value + lastSess atomic.Value } func (f *fakeAsker) Ask(_ context.Context, session, ask string) (string, components.PrefixUsage, error) { @@ -36,7 +40,8 @@ func (f *fakeAsker) Ask(_ context.Context, session, ask string) (string, compone if f.err != nil { return "", components.PrefixUsage{}, f.err } - return f.reply, components.PrefixUsage{CacheRead: f.cacheRead, Fresh: 40, Output: 90}, nil + return f.reply, components.PrefixUsage{CacheRead: f.cacheRead, Fresh: 40, Output: 90, + ViaTool: f.viaTool}, nil } func (f *fakeAsker) ask() string { s, _ := f.lastAsk.Load().(string); return s } @@ -832,3 +837,71 @@ func TestSweepSendsTheInventoryAndNotTheOutputs(t *testing.T) { } } } + +// The verdict tool's whole purpose is that the model ANSWERS with it, and until this counter existed +// nothing could say whether it did. extract.ParseVerdicts reads a tool_use `input` and a JSON array in +// reply text identically -- deliberately, so that declaring the tool is additive -- with the +// consequence that a run where the declared tool is never touched produces the same verdicts, the same +// savings and the same gates as one where it is used every time. That is not hypothetical: a review of +// PR #137 measured 0 of 5 live asks using the tool while the PR claimed 6 of 6, and no published +// figure in this repo could adjudicate between the two readings. +func TestSweepCountsWhetherTheAnswerCameViaTheToolOrProse(t *testing.T) { + verdicts := `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]` + for _, tc := range []struct { + name string + viaTool bool + want string + notWant string + }{ + {"tool_use", true, "sweep_answered_via_tool", "sweep_answered_via_prose"}, + {"prose", false, "sweep_answered_via_prose", "sweep_answered_via_tool"}, + } { + t.Run(tc.name, func(t *testing.T) { + asker := &fakeAsker{reply: verdicts, cacheRead: 19595, viaTool: tc.viaTool} + e := newSweepSmall(t, "") + rep := &components.Report{} + if _, err := e.Offload(sweepReq(), rep, + preExpiryCtx("s-"+tc.name, asker, store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + // PRECONDITION. Without it a turn that never reached the ask at all would pass both + // assertions below by producing neither event, which is exactly the vacuous shape this + // repo's convention exists to catch. + if rep.Events["sweep_prefix_cache_read_ok"] != 1 { + t.Fatalf("the prefix ask never happened, so no reply shape was observed "+ + "(gates: %v events: %v)", rep.Gates, rep.Events) + } + if rep.Events[tc.want] != 1 { + t.Errorf("%s: %s = %d, want 1 (events: %v)", tc.name, tc.want, + rep.Events[tc.want], rep.Events) + } + if rep.Events[tc.notWant] != 0 { + t.Errorf("%s: %s = %d, want 0 -- the two shapes must not both be counted "+ + "(events: %v)", tc.name, tc.notWant, rep.Events[tc.notWant], rep.Events) + } + }) + } +} + +// The FALLBACK is neither shape. fallbackAsk calls Model.Complete(), which carries no tool to declare, +// so counting its reply as prose would report a shape that was never on offer -- and would silently +// inflate the prose count with calls that could not have used the tool even in principle. +func TestSweepDoesNotAttributeAReplyShapeToTheFallback(t *testing.T) { + // cacheRead 0 sends this down the fallback fork, which is the only way in without a real + // non-Anthropic route. + asker := &fakeAsker{reply: `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]`, cacheRead: 0} + e := newSweepSmall(t, "") + rep := &components.Report{} + if _, err := e.Offload(sweepReq(), rep, + preExpiryCtx("s-fb", asker, store.NewMemory(store.Options{}))); err != nil { + t.Fatal(err) + } + if rep.Gates["sweep_prefix_cache_read_ZERO"] != 1 { + t.Fatalf("the fallback fork was never taken, so this asserts nothing "+ + "(gates: %v events: %v)", rep.Gates, rep.Events) + } + if rep.Events["sweep_answered_via_tool"] != 0 || rep.Events["sweep_answered_via_prose"] != 0 { + t.Errorf("the fallback was attributed a reply shape it could not have had (events: %v)", + rep.Events) + } +} diff --git a/docs/components/extract_llm_sweep.md b/docs/components/extract_llm_sweep.md index d9f524f..d65aaf7 100644 --- a/docs/components/extract_llm_sweep.md +++ b/docs/components/extract_llm_sweep.md @@ -92,10 +92,73 @@ transcript as of the previous turn, so the newest tool output is invisible to it content has had no turns in which to be superseded — and it keeps a large model call off the agent's critical path. -Three construction facts, each measured and each a test: `tool_choice: none` is **not** in the cache key -so forcing it is free, and **necessary**, or the prefix's tools make the model answer with a `tool_use`; -`tools` **are** in the key, so stripping them reads a different, smaller entry; and the route rejects -assistant prefill, which an appended user message satisfies by construction. +Two construction facts, each measured and each a test: `tools` **are** in the cache key, so stripping +them reads a different, smaller entry; and the route rejects assistant prefill, which an appended user +message satisfies by construction. + +### `tool_choice`, and why the answer is a declared tool rather than a suppressed one + +This page previously stated the inverse of what is now measured — that forcing `tool_choice: none` was +"free, and **necessary**, or the prefix's tools make the model answer with a `tool_use`". The second +half of that sentence was **right about the mechanism and wrong about the remedy**, and the fix is not +to suppress the tool call but to declare a tool worth calling. + +The ask carries the agent's own tools, because `tools` are in the cache key and stripping them costs +the read. Three arms, same transcript, same ask model (`aws/claude-sonnet-5` over LOCA S2L, three tasks, +three passes each, run sequentially): + +| arm | `tool_choice` | verdict tool declared | asks replied | unusable | answered via `tool_use` | verdict coverage | +|---|---|---|---|---|---|---| +| A | `{"type":"none"}` | no | 20 | **6 (30.0%)** | 0 | 71.5% | +| B | omitted | **no** | 24 | **14 (58.3%)** | 0 | 41.2% | +| C | omitted | **yes** | 77 | **7 (9.1%)** | **43 (55.8%)** | 90.9% | + +Fisher two-tailed: A vs C `p = 0.0245`, B vs C `p = 0.0000`, A vs B `p = 0.0755`. + +Arm **B is the one that had never been run**, and it is the reason the tool is declared rather than the +`tool_choice` simply being dropped. Removing the suppression without offering an answer tool is +**worse than either other arm**: the model, now free to call something and offered only the agent's +tools and `context_guru_expand`, calls one of those. Directly observed on the wire by logging every +reply's content blocks — `thinking,tool_use:context_guru_expand` with **no text block at all**, which +the text-only extraction reads as the empty string and files as unusable. Arm B also lost 5 asks to the +90 s `llmCallTimeout` against 0 in arm A and 1 in arm C. + +So the three shapes are: suppress the call and the model argues its verdicts in prose (arm A's failures); +allow the call with nothing worth calling and the answer is lost into somebody else's tool (arm B); +allow it and declare the right tool, and 55.8% of replies come back schema-shaped, at cache-read price, +with the rest still read by the unchanged prose parser (arm C). + +`tool_choice` is not free in every form either. Naming a tool wrote a **second** cache entry (8,378 +against the 8,268 already cached), so `tool_choice` does participate in the key when it names a tool, +even though `none` does not. Omitting it entirely is what reads the prefix for free. + +The residual 9.1% in arm C is dominated by a reply with a `thinking` block and no answer at all. That +is a separate defect, present in every arm, and it is tracked apart from this component's contract. + +### The verdict tool is advertised only where it can be used + +`context_guru_adjudicate` is appended to the `tools` array of every request on the **Anthropic** route +whose pipeline **contains `extract_llm_sweep`** — and nowhere else. It costs a measured 946 bytes at the +head of the cacheable prefix, so both conditions matter: + +- **Pipeline membership.** A preset with no sweep can never adjudicate. `off` is the control arm of every + published comparison in this repo, and injecting there perturbed the baseline of all of them. +- **Provider.** `prefixAskerFor` returns nil for anything but Anthropic and `cheapmodel/openai.go` has no + `CompletePrefixed` at all, so on the OpenAI route the definition is unreachable by construction. + +Neither condition varies per turn — pipeline membership is fixed per **config document**, the provider +by the route — so the prefix stays byte-stable for every request under a given config and never flaps. +That is the distinction the cache argument actually requires: it forbids gating on something that +changes turn to turn, not on something fixed for the config. (Not "fixed at config load": `tenancy.go` +rebuilds a tenant's `*Pipeline` when the config document changes, mid-session. A config change that adds +or drops this component has already invalidated the prefix for much bigger reasons than 946 bytes.) + +Because the model is told not to call it and sometimes still would, a call the **agent** makes is +answered on the response path before the client is written to — **except** when the model calls it +alongside a *client* tool in the same assistant turn. The response loop cannot continue a turn whose +other `tool_use` only the client can execute, so it hands that round over whole and +`adjudicate.AnswerStrayCalls` repairs it on the next request: the agent pays one turn, the session is +fine. `/stats` publishes `adjudicate_stray` either way — measured 0 across all three passes of arm C. ## The trigger: pre-expiry, not cold diff --git a/docs/reference/routes.md b/docs/reference/routes.md index 89daa99..0ea5485 100644 --- a/docs/reference/routes.md +++ b/docs/reference/routes.md @@ -73,6 +73,7 @@ consumer that reads by key keeps working. The tables below group the `Snapshot` | `adjusted_saved` | `saved − wasted`. May be negative. | | `components` | Per-component rollup (see below). | | `top_passthrough` | Components that ran but never *changed* a request — dead weight. A component that mutated without saving content tokens (`cachesplit`, `cacheinject`) is **not** listed here. | +| `adjudicate_stray` | Times the **agent** called `context_guru_adjudicate`, the verdict tool the proxy injects for [`extract_llm_sweep`](../components/extract_llm_sweep.md) and tells the model not to call. Counted whether the call was answered on the response path or repaired on the next request. Most cost the agent nothing, because the response path answers them in band before the client is written to; one co-called with a *client* tool in the same assistant turn costs one turn, because the loop hands that round over whole and the next request's repair fixes it. Either way this is a *description-health* number rather than an error: non-zero means the "do not call this yourself" wording has stopped working. Measured 0 over three benchmark passes. Also exported as `cg_adjudicate_stray_total`. | | `top_discarded` | Components whose changes the **writeback layer threw away** at least once. Any entry needs investigating: the component ran, mutated, and had no effect on the wire. | Per-component (`components.` and `potential_components.`): diff --git a/expand/response.go b/expand/response.go index 4924700..63ca9a9 100644 --- a/expand/response.go +++ b/expand/response.go @@ -16,26 +16,43 @@ type Call struct { // true if the model also called some OTHER tool — in that case the host cannot // safely auto-continue (the client must resolve the other tools), so the loop // bails and returns the response as-is. -func ResponseCalls(provider string, resp []byte) (calls []Call, otherTools bool) { +// otherProxyTools are additional names the caller owns. otherTools means "a tool the CLIENT +// implements", which is what makes the response loop hand the turn over untouched — so a second +// proxy-injected tool counted as "other" caused the loop to bail and stream that proxy tool_use to +// the client. Variadic so the existing callers, which advertise only expand, are unchanged. +func ResponseCalls(provider string, resp []byte, otherProxyTools ...string) (calls []Call, otherTools bool) { + ours := func(name string) bool { + if name == ToolName { + return true + } + for _, t := range otherProxyTools { + if t != "" && name == t { + return true + } + } + return false + } switch provider { case "anthropic": gjson.GetBytes(resp, "content").ForEach(func(_, blk gjson.Result) bool { if blk.Get("type").String() != "tool_use" { return true } - if blk.Get("name").String() == ToolName { + switch name := blk.Get("name").String(); { + case name == ToolName: calls = append(calls, Call{CallID: blk.Get("id").String(), HashID: blk.Get("input.id").String()}) - } else { + case !ours(name): otherTools = true } return true }) default: // openai and compatibles gjson.GetBytes(resp, "choices.0.message.tool_calls").ForEach(func(_, tc gjson.Result) bool { - if tc.Get("function.name").String() == ToolName { + switch name := tc.Get("function.name").String(); { + case name == ToolName: hash := gjson.Get(tc.Get("function.arguments").String(), "id").String() calls = append(calls, Call{CallID: tc.Get("id").String(), HashID: hash}) - } else { + case !ours(name): otherTools = true } return true diff --git a/internal/adjudicate/tool.go b/internal/adjudicate/tool.go new file mode 100644 index 0000000..c0630fe --- /dev/null +++ b/internal/adjudicate/tool.go @@ -0,0 +1,299 @@ +// Package adjudicate declares the context-maintenance tool the cold-sweep adjudication asks with, and +// the wire helpers that keep it byte-stable in the prompt-cache prefix. +// +// WHY A TOOL AT ALL, rather than only asking for a JSON array in the reply text. The ask carries the +// AGENT's tools, because `tools` are in the cache key and stripping them costs the read — so the only +// question is what the model does with that freedom. Three arms, same transcript and ask model, three +// benchmark passes each, run sequentially: +// +// tool_choice verdict tool asks replied unusable answered via tool_use +// {"type":"none"} no 20 6 (30.0%) 0 +// (omitted) NO 24 14 (58.3%) 0 +// (omitted) yes 77 7 (9.1%) 43 (55.8%) +// +// Fisher two-tailed: row 1 vs row 3 p = 0.0245, row 2 vs row 3 p = 0.0000. +// +// The MIDDLE row is why this package exists, and it is the arm nobody had run: removing +// tool_choice:none on its own is WORSE than leaving it in. Freed to call a tool and offered only the +// agent's own plus context_guru_expand, the model calls one of those; logging every reply's content +// blocks caught it as `thinking,tool_use:context_guru_expand` with no text block at all, which the +// text-only extraction reads as "" and files as unusable. That arm also lost 5 asks to the 90 s +// llmCallTimeout against 0 and 1 in the others. +// +// So the two halves are one change. Suppressing the call trades a lost answer for prose (row 1's +// failures); allowing it with nothing worth calling loses the answer outright (row 2); allowing it and +// declaring the right tool gets 55.8% of replies back schema-shaped at cache-read price, with the rest +// still read by the unchanged prose parser (row 3). +// +// FORCING the tool by name is separately not free: it produced a second cache entry (8,378 written +// against the 8,268 already cached), so tool_choice DOES participate in the key when it names a tool, +// even though `none` does not. +// +// WHERE IT IS INJECTED. On every request of an Anthropic route whose pipeline contains +// extract_llm_sweep, and nowhere else. Every-request matters because `tools` hashes before system and +// messages, so a tool that appears when the sweep fires and vanishes on the next turn invalidates the +// prefix from position zero -- the flap expand's `always` mode exists to prevent. But that argument +// only forbids gating on something that varies PER TURN: pipeline membership is fixed per config +// DOCUMENT and the provider by the route, so both conditions are byte-stable for every request under a +// given config. Not "fixed at config load", which is what this used to claim: proxy/tenancy.go rebuilds +// a tenant's *Pipeline when the config document changes, mid-session. That rebuild is a +// prefix-invalidating event in its own right, so the byte-stability argument holds where it matters. +// Injecting without +// them cost a measured 946 bytes at the head of the cacheable prefix of every preset including `off`, +// the control arm of every published comparison here, and including presets with no sweep at all. +// +// This does not replace the text path. extract.ParseVerdicts and extract.BuildFallbackAsk are still +// the fallback, and a model that answers in prose anyway is still read exactly as before; the tool +// only changes which reply shape is PREFERRED. +package adjudicate + +import ( + "strconv" + "sync/atomic" + + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +// strayAnswered counts tool_results rewritten because the AGENT called this tool. Surfaced in /stats +// as adjudicate_stray: the rate is the only signal for whether the description is doing its job. +// MEASURED at 0 across ~4,900 requests with the description below, which is why the answer text is a +// cheap insurance policy rather than a hot path. +var strayAnswered atomic.Int64 + +// StrayAnswered returns how many stray calls have been answered, on either path. +func StrayAnswered() int64 { return strayAnswered.Load() } + +// NoteAnsweredInBand counts stray calls answered on the RESPONSE path, before the client saw them. +// The same counter as the request-path repair on purpose: /stats publishes "the agent called a tool it +// was told not to call, n times", and which of the two defences caught it does not change that number +// — it is the rate that says whether the description is still working. Which path caught it is in the +// cg.adjudicate_stray log line, where it belongs. +func NoteAnsweredInBand(n int) { + if n > 0 { + strayAnswered.Add(int64(n)) + } +} + +// ToolName is the wire name. Prefixed like the expand tool so an operator reading a transcript can +// tell at a glance which tools the proxy injected and which the client owns. +const ToolName = "context_guru_adjudicate" + +// StrayAnswer is what a stray call from the AGENT gets. The model does call an advertised tool it was +// told to leave alone -- directly observed with context_guru_expand, which a run called at step 2 -- +// and the client cannot execute a tool the proxy injected, so it answers something like +// "Tool 'context_guru_adjudicate' not found" and the agent loses a turn to a dead end. This gives it a +// definite, uninteresting answer instead. +const StrayAnswer = "Context maintenance runs automatically in the background. No action is required " + + "from you, and you do not need to call this tool. Continue with the task." + +// toolDesc tells the model who the tool is for. "Do not call it yourself" does not GUARANTEE it will +// not (see StrayAnswer), but it measured 0 strays in ~4,900 requests and costs nothing. +const toolDesc = "Internal to the context manager. Reports which earlier tool outputs are spent and " + + "safe to remove from the transcript. This is invoked by the context manager, not by you - do not " + + "call it yourself." + +// schemaJSON constrains the answer, and its field names are exactly extract.Verdict's JSON tags so the +// existing parser reads a tool input unchanged. +// +// The label is a small INTEGER, never the tool_use id: asked for opaque ids the model REGULARISED them +// (answering toolu_01..07 for toolu_probe_00..07), because reproducing a random identifier from +// thousands of tokens back is a copying task rather than a judgement. With integer labels it was 0 bad +// labels across 40+ trials. +const schemaJSON = `{"type":"object","properties":{"verdicts":{"type":"array","description":` + + `"One entry per label you were shown. Answer for EVERY label.","items":{"type":"object","properties":{` + + `"i":{"type":"integer","description":"The label you were shown."},` + + `"needed_by":{"type":"string","enum":["a","b","c","none"],"description":` + + `"Which outstanding obligation still needs this output, or none if it is spent."},` + + `"quote":{"type":"string","description":"Verbatim transcript text creating that obligation; empty when needed_by is none."},` + + `"verdict":{"type":"string","enum":["keep","drop"],"description":"drop requires needed_by to be none."}},` + + `"required":["i","needed_by","verdict"]}}},"required":["verdicts"]}` + +// anthropicDef and openAIDef are the tool definitions, kept as raw JSON so injection is a byte splice +// and the cached prefix stays stable to the byte -- a re-marshalled map would be free to reorder keys. +const anthropicDef = `{"name":"` + ToolName + `","description":"` + toolDesc + `","input_schema":` + schemaJSON + `}` + +const openAIDef = `{"type":"function","function":{"name":"` + ToolName + `","description":"` + toolDesc + + `","parameters":` + schemaJSON + `}}` + +// ToolDefRaw returns the provider-shaped tool definition. +func ToolDefRaw(provider string) []byte { + if provider == "anthropic" { + return []byte(anthropicDef) + } + return []byte(openAIDef) +} + +// HasTool reports whether body already declares the tool. This is the ADVERTISE test, and a host must +// answer stray calls exactly when it is true. +func HasTool(provider string, body []byte) bool { + field := "function.name" + if provider == "anthropic" { + field = "name" + } + for _, t := range gjson.GetBytes(body, "tools").Array() { + if t.Get(field).String() == ToolName { + return true + } + } + return false +} + +// Inject appends the tool to body's tools array, byte-stably and idempotently. +// +// Appended LAST so the client's own tools keep their exact order, and skipped when a forcing +// tool_choice is present so tool selection is never perturbed. Also skipped when the request declares +// no tools at all: handing the model its first tool changes what it believes it can do, and that is +// the riskiest case for a model that penalizes an unexpected tool. Fail-open — any trouble returns the +// original body. +func Inject(provider string, body []byte) (out []byte, injected bool) { + if tc := gjson.GetBytes(body, "tool_choice"); tc.Exists() && !toolChoiceIsAuto(tc) { + return body, false + } + tools := gjson.GetBytes(body, "tools") + if !tools.Exists() || !tools.IsArray() || len(tools.Array()) == 0 { + return body, false + } + if HasTool(provider, body) { + return body, false + } + nb, err := sjson.SetRawBytes(body, "tools.-1", ToolDefRaw(provider)) + if err != nil { + return body, false // fail open + } + return nb, true +} + +// toolChoiceIsAuto reports whether a tool_choice leaves the model free to choose. OpenAI: the string +// "auto". Anthropic: {"type":"auto"}. Anything else (none/required/any/a named tool) is forcing, and +// injection is skipped. +func toolChoiceIsAuto(tc gjson.Result) bool { + if tc.Type == gjson.String { + return tc.String() == "auto" + } + if tc.IsObject() { + return tc.Get("type").String() == "auto" + } + return false +} + +// ResponseCallIDs returns the ids of calls the AGENT made to this tool in an upstream RESPONSE. +// +// The request-path repair (AnswerStrayCalls) is a backstop and cannot be the primary defence: by the +// time it runs, the client has already SEEN the call, already failed to execute a tool it never +// declared, and already spent a turn answering "not found". Answering the call in-band on the +// response path -- before the client is written to -- is what makes the repair a backstop, and these +// ids are what the response loop needs to build that answer. +// +// "Backstop" is not "unreachable", and this comment used to imply it was. The response loop declines +// the in-band answer, deliberately, when the assistant turn ALSO calls a client tool: it cannot +// continue a turn whose other tool_use only the client can execute, so it hands the round over and +// leaves the next request's repair to fix it. On that path the repair is the ONLY defence and the agent +// does pay one turn. +func ResponseCallIDs(provider string, resp []byte) (ids []string) { + if provider == "anthropic" { + gjson.GetBytes(resp, "content").ForEach(func(_, blk gjson.Result) bool { + if blk.Get("type").String() == "tool_use" && blk.Get("name").String() == ToolName { + if id := blk.Get("id").String(); id != "" { + ids = append(ids, id) + } + } + return true + }) + return ids + } + gjson.GetBytes(resp, "choices.0.message.tool_calls").ForEach(func(_, tc gjson.Result) bool { + if tc.Get("function.name").String() == ToolName { + if id := tc.Get("id").String(); id != "" { + ids = append(ids, id) + } + } + return true + }) + return ids +} + +// AnswerStrayCalls replaces the client's tool_result for any call the AGENT made to this tool with a +// definite answer, and reports how many it replaced. +// +// Same request-path shape as expand.RepairToolResults, and for the same reason: the client executes +// the real tools itself, finds no tool by this name because the PROXY injected it, and answers +// something like "Tool 'context_guru_adjudicate' not found". Left alone, the model reads a failure it +// cannot act on and may retry. Rewriting it on the next request works WITH the client's loop instead +// of against it, needs nothing implemented client-side, and is deterministic — the same substitution +// every turn, so the prefix does not flap. +// +// Reading the tool_use out of the transcript rather than trusting the tool_result is what makes this +// safe: a client's tool_result is rewritten only when the assistant turn it answers called OUR tool. +// A body with no such call comes back byte-identical, because this runs on every request and a +// gratuitous rewrite would change the prefix and cost a cache write for nothing. +// +// The response loop answers most strays before this ever runs, but NOT all of them: an assistant turn +// that calls this tool alongside a CLIENT tool is handed to the client whole, so for that turn this +// repair is the only defence rather than a second one. See the co-call note in the response loop. +func AnswerStrayCalls(provider string, body []byte) (out []byte, answered int) { + msgs := gjson.GetBytes(body, "messages") + if !msgs.IsArray() { + return body, 0 + } + // An id-less tool_use would make "" a live key, and then a tool_result carrying no tool_use_id -- + // someone else's block -- would match it and be overwritten. Both halves of the pair must name an id. + ours := map[string]bool{} + for _, m := range msgs.Array() { + if provider == "anthropic" { + for _, blk := range m.Get("content").Array() { + if blk.Get("type").String() == "tool_use" && blk.Get("name").String() == ToolName { + if id := blk.Get("id").String(); id != "" { + ours[id] = true + } + } + } + continue + } + for _, tc := range m.Get("tool_calls").Array() { + if tc.Get("function.name").String() == ToolName { + if id := tc.Get("id").String(); id != "" { + ours[id] = true + } + } + } + } + if len(ours) == 0 { + return body, 0 + } + out = body + for mi, m := range msgs.Array() { + base := "messages." + strconv.Itoa(mi) + if provider != "anthropic" { + if m.Get("role").String() != "tool" || !ours[m.Get("tool_call_id").String()] { + continue + } + if nb, err := sjson.SetBytes(out, base+".content", StrayAnswer); err == nil { + out, answered = nb, answered+1 + } + continue + } + for bi, blk := range m.Get("content").Array() { + if blk.Get("type").String() != "tool_result" || !ours[blk.Get("tool_use_id").String()] { + continue + } + path := base + ".content." + strconv.Itoa(bi) + nb, err := sjson.SetBytes(out, path+".content", StrayAnswer) + if err != nil { + continue + } + out, answered = nb, answered+1 + // The block is no longer an error: leaving is_error set tells the model its own call + // failed while handing it the answer to that call. + if blk.Get("is_error").Exists() { + if nb, err := sjson.SetBytes(out, path+".is_error", false); err == nil { + out = nb + } + } + } + } + if answered > 0 { + strayAnswered.Add(int64(answered)) + } + return out, answered +} diff --git a/internal/adjudicate/tool_test.go b/internal/adjudicate/tool_test.go new file mode 100644 index 0000000..88a186f --- /dev/null +++ b/internal/adjudicate/tool_test.go @@ -0,0 +1,154 @@ +package adjudicate + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/tidwall/gjson" +) + +// The tool must land LAST and be idempotent, because `tools` hashes before system and messages: a tool +// inserted anywhere else, or twice, invalidates the prompt-cache prefix from position zero. +func TestInjectIsByteStableAndIdempotent(t *testing.T) { + body := []byte(`{"model":"m","tools":[{"name":"Read","description":"d","input_schema":{"type":"object"}}],` + + `"messages":[{"role":"user","content":"go"}]}`) + out, ok := Inject("anthropic", body) + if !ok { + t.Fatal("did not inject into a request that declares tools") + } + tools := gjson.GetBytes(out, "tools").Array() + if len(tools) != 2 { + t.Fatalf("expected 2 tools, got %d", len(tools)) + } + if tools[0].Get("name").String() != "Read" { + t.Error("the client's own tool moved; its order must be preserved exactly") + } + if tools[1].Get("name").String() != ToolName { + t.Errorf("our tool is not last: %q", tools[1].Get("name").String()) + } + // The schema must actually parse, or the provider rejects every request carrying it — on every + // request, since this is injected unconditionally. + var schema map[string]any + if err := json.Unmarshal([]byte(tools[1].Get("input_schema").Raw), &schema); err != nil { + t.Fatalf("input_schema is not valid JSON: %v", err) + } + // The verdict fields are extract.Verdict's JSON tags; that is what lets the existing parser read a + // tool input unchanged, so a rename here silently reverts the fix to the text path. + props := gjson.GetBytes(out, `tools.1.input_schema.properties.verdicts.items.properties`) + for _, f := range []string{"i", "needed_by", "quote", "verdict"} { + if !props.Get(f).Exists() { + t.Errorf("schema is missing verdict field %q; extract.ParseVerdicts reads that name", f) + } + } + again, ok2 := Inject("anthropic", out) + if ok2 { + t.Error("injected twice; a duplicated tool changes the prefix on every turn") + } + if string(again) != string(out) { + t.Error("a second injection altered the body") + } + // Byte-stability across calls: the definition is spliced raw precisely so two injections of the + // same body produce the same bytes rather than a re-marshalled map's key order. + out2, _ := Inject("anthropic", body) + if string(out2) != string(out) { + t.Error("two injections of the same body differ byte-wise; the prefix would flap") + } +} + +// Two cases where injecting would change what the model believes it can do, or which tool it is +// compelled to call. Both must be refused. +func TestInjectRefusesWhenItWouldPerturbSelection(t *testing.T) { + noTools := []byte(`{"model":"m","messages":[{"role":"user","content":"go"}]}`) + if _, ok := Inject("anthropic", noTools); ok { + t.Error("injected into a request with NO tools; that hands the model its first tool and " + + "changes what it believes it can do") + } + forced := []byte(`{"model":"m","tool_choice":{"type":"tool","name":"Read"},` + + `"tools":[{"name":"Read","description":"d","input_schema":{"type":"object"}}],` + + `"messages":[{"role":"user","content":"go"}]}`) + if _, ok := Inject("anthropic", forced); ok { + t.Error("injected under a forcing tool_choice; tool selection must never be perturbed") + } + // An explicit auto is not forcing, so it must still inject. + auto := []byte(`{"model":"m","tool_choice":{"type":"auto"},` + + `"tools":[{"name":"Read","description":"d","input_schema":{"type":"object"}}],` + + `"messages":[{"role":"user","content":"go"}]}`) + if _, ok := Inject("anthropic", auto); !ok { + t.Error("refused an auto tool_choice, which leaves the model free to choose") + } +} + +// A stray call from the AGENT must get a definite answer, and the real tools' results must survive +// untouched. The model does call advertised tools it was told to leave alone -- directly observed with +// the expand tool, which a run called at step 2 -- so this path is load-bearing, not defensive. +func TestAnswerStrayCallsLeavesRealResultsAlone(t *testing.T) { + before := StrayAnswered() + body := []byte(`{"model":"m","messages":[ + {"role":"assistant","content":[ + {"type":"tool_use","id":"u1","name":"` + ToolName + `","input":{"verdicts":[]}}, + {"type":"tool_use","id":"u2","name":"Read","input":{"path":"a.py"}}]}, + {"role":"user","content":[ + {"type":"tool_result","tool_use_id":"u1","is_error":true,"content":"Tool '` + ToolName + `' not found"}, + {"type":"tool_result","tool_use_id":"u2","content":"real file contents"}]} + ]}`) + out, n := AnswerStrayCalls("anthropic", body) + if n != 1 { + t.Fatalf("answered %d stray calls, want 1", n) + } + s := string(out) + if strings.Contains(s, "not found") { + t.Error("the client's dead-end refusal reached the model; the point is to replace it") + } + if !strings.Contains(s, "runs automatically") { + t.Error("no substitute answer was written") + } + if !strings.Contains(s, "real file contents") { + t.Error("a REAL tool's result was overwritten; only our own calls may be touched") + } + if gjson.GetBytes(out, "messages.1.content.0.is_error").Bool() { + t.Error("is_error survived, so the model reads a failure alongside the answer to that call") + } + if got := StrayAnswered() - before; got != 1 { + t.Errorf("counted %d stray calls, want 1; the rate is the signal for whether the tool's "+ + "description is working", got) + } + // OpenAI dialect: a role=tool message answering one call. + oa := []byte(`{"model":"m","messages":[ + {"role":"assistant","tool_calls":[{"id":"c1","function":{"name":"` + ToolName + `","arguments":"{}"}}]}, + {"role":"tool","tool_call_id":"c1","content":"Tool not found"}]}`) + out2, n2 := AnswerStrayCalls("openai", oa) + if n2 != 1 || !strings.Contains(string(out2), "runs automatically") { + t.Errorf("OpenAI dialect not handled: answered=%d body=%.180s", n2, out2) + } +} + +// A body with no calls to our tool must come back byte-identical: this runs on every request, and a +// gratuitous rewrite would change the prefix and cost a cache write for nothing. +func TestAnswerStrayCallsIsANoOpWhenUninvolved(t *testing.T) { + body := []byte(`{"model":"m","messages":[ + {"role":"assistant","content":[{"type":"tool_use","id":"u9","name":"Read","input":{}}]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"u9","content":"data"}]}]}`) + out, n := AnswerStrayCalls("anthropic", body) + if n != 0 { + t.Errorf("answered %d calls on a body that never called our tool", n) + } + if string(out) != string(body) { + t.Error("body was rewritten with nothing to do; that costs a cache write for nothing") + } +} + +// HasTool is the ADVERTISE test: a host must answer stray calls exactly when it is true, so it has to +// distinguish our tool from a client tool by the same shape in both dialects. +func TestHasTool(t *testing.T) { + oa, ok := Inject("openai", []byte(`{"tools":[{"type":"function","function":{"name":"Read"}}]}`)) + if !ok { + t.Fatal("openai injection refused on a request that declares tools") + } + if !HasTool("openai", oa) { + t.Error("HasTool did not see the tool we just injected in the openai dialect") + } + if HasTool("anthropic", []byte(`{"tools":[{"name":"Read"}]}`)) { + t.Error("HasTool matched a request that declares only the client's own tools") + } +} diff --git a/internal/cheapmodel/anthropic.go b/internal/cheapmodel/anthropic.go index f37f1c4..7077bdb 100644 --- a/internal/cheapmodel/anthropic.go +++ b/internal/cheapmodel/anthropic.go @@ -18,6 +18,7 @@ import ( "github.com/tidwall/sjson" "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/internal/adjudicate" ) // Anthropic calls a small Anthropic model with a single user prompt and returns the @@ -193,6 +194,10 @@ type PrefixUsage struct { CacheWrite int Fresh int Output int + // ViaTool: the answer came back as a tool_use for adjudicate.ToolName, not as reply text. See + // components.PrefixUsage.ViaTool -- the field exists so the caller can COUNT which shape it got, + // because the text parser accepts both and therefore hides the difference. + ViaTool bool } // CompletePrefixed sends `ask` as a trailing user message appended to prefixBody — an ENTIRE @@ -211,11 +216,36 @@ type PrefixUsage struct { // // - appending a trailing user message to a byte-identical prefix reads the whole prefix from // cache and writes nothing: 19,595 read / 0 created. -// - `tool_choice` is NOT part of the cache key, so forcing it to "none" is free — and necessary, -// because the prefix carries the agent's tools and the model will otherwise answer with a -// tool_use instead of the verdicts. +// +// - NO `tool_choice` IS SET, and a structured-answer tool IS declared. Those two halves are ONE +// change: the earlier comment here claimed forcing "none" was free and necessary "because the +// model will otherwise answer with a tool_use instead of the verdicts", and it was RIGHT about +// the mechanism and WRONG about the remedy. Three arms over the same transcript and ask model, +// three benchmark passes each, run sequentially: +// +// tool_choice tool declared asks replied unusable answered via tool_use +// {"type":"none"} no 20 6 (30.0%) 0 +// (omitted) NO 24 14 (58.3%) 0 +// (omitted) yes 77 7 (9.1%) 43 (55.8%) +// +// Fisher two-tailed: row 1 vs row 3 p = 0.0245, row 2 vs row 3 p = 0.0000. +// +// THE MIDDLE ROW IS THE POINT, and it is the arm that had never been run before: dropping the +// suppression WITHOUT declaring an answer tool is worse than leaving it in. Freed to call +// something and offered only the agent's tools plus context_guru_expand, the model calls one of +// those — observed directly by logging every reply's content blocks, as +// `thinking,tool_use:context_guru_expand` with NO text block, which this function's text +// extraction reads as "" and the caller files as unusable. That arm also lost 5 asks to the 90 s +// llmCallTimeout, against 0 and 1 in the others. So `none` was suppressing a real failure mode; +// declaring a tool worth calling is what removes the mode instead of trading it for prose. +// +// Forcing a NAMED tool is separately not free: it wrote a second cache entry (8,378 against the +// 8,268 already cached), so tool_choice DOES participate in the key when it names a tool even +// though "none" does not. Omitting it entirely reads the prefix for free. +// // - `tools` ARE part of the key. They are therefore left exactly as the prefix had them; dropping // them read a different, smaller entry (19,129) i.e. a separate cache line and a fresh write. +// // - this route REJECTS assistant prefill ("the conversation must end with a user message"), which // the appended user message satisfies by construction — but it means prefixBody must not be // extended any other way. @@ -234,10 +264,6 @@ func (a Anthropic) CompletePrefixed(ctx context.Context, prefixBody []byte, ask if err != nil { return "", u, err } - // tool_choice: free (not in the cache key) and required, or the model answers with a tool_use. - if body, err = sjson.SetBytes(body, "tool_choice", map[string]any{"type": "none"}); err != nil { - return "", u, err - } maxTok := a.MaxTokens if maxTok == 0 { maxTok = PrefixAskMaxTokens @@ -279,7 +305,10 @@ func (a Anthropic) CompletePrefixed(ctx context.Context, prefixBody []byte, ask } var out struct { Content []struct { - Text string `json:"text"` + Type string `json:"type"` + Text string `json:"text"` + Name string `json:"name"` + Input json.RawMessage `json:"input"` } `json:"content"` Usage struct { InputTokens int `json:"input_tokens"` @@ -295,6 +324,21 @@ func (a Anthropic) CompletePrefixed(ctx context.Context, prefixBody []byte, ask Fresh: out.Usage.InputTokens, Output: out.Usage.OutputTokens} recordUsageCache(ctx, a.Model, out.Usage.InputTokens, out.Usage.OutputTokens, out.Usage.CacheCreationTok, out.Usage.CacheReadTok) + // OUR TOOL'S INPUT BEATS TEXT. When the prefix advertises the structured-answer tool the model uses + // it of its own accord, and the input arrives already schema-shaped — which removes three failure + // modes the text path had: prose instead of JSON, verdicts for only part of the batch, and an array + // cut off by the output budget mid-flight. The raw input is returned as-is, because its field names + // are the caller's own JSON tags and its `verdicts` array is what the existing parser scans for; the + // text path below is unchanged and still handles a model that answered in prose anyway. + // + // Only OUR tool, by name: the prefix carries the AGENT's tools too, and returning a `Read` call's + // input would replace a usable prose answer with an argument list that cannot parse. + for _, c := range out.Content { + if c.Type == "tool_use" && c.Name == adjudicate.ToolName && len(c.Input) > 0 { + u.ViaTool = true + return string(c.Input), u, nil + } + } for _, c := range out.Content { if c.Text != "" { return c.Text, u, nil diff --git a/metrics/metrics.go b/metrics/metrics.go index deebb5d..6a1eaf2 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -612,9 +612,14 @@ type Snapshot struct { // have minted resolved to nothing, i.e. a cut advertised as reversible was not — a defect, and // the ALERTABLE one. Neither can be inferred from WastedTokens, which counts successful // re-serves and therefore reads identically whether a stash broke or expand was never called. - ExpandUnresolvedMalformed int64 `json:"expand_unresolved_malformed"` - ExpandUnresolvedMissing int64 `json:"expand_unresolved_missing"` - Components map[string]compStat `json:"components"` + ExpandUnresolvedMalformed int64 `json:"expand_unresolved_malformed"` + ExpandUnresolvedMissing int64 `json:"expand_unresolved_missing"` + // AdjudicateStray counts tool_results rewritten because the AGENT called the proxy-injected + // adjudication tool. The proxy advertises that tool on every request and tells the model not to + // call it; this is the number that says whether the telling works (measured 0 across ~4,900 + // requests). Non-zero is a lost agent turn per count, not a correctness failure. + AdjudicateStray int64 `json:"adjudicate_stray"` + Components map[string]compStat `json:"components"` // TopPassthrough names components that ran but never saved a token — dead // weight in the pipeline, candidates to drop from the config. TopPassthrough []string `json:"top_passthrough"` diff --git a/proxy/adjudicatetool_test.go b/proxy/adjudicatetool_test.go new file mode 100644 index 0000000..af49ed1 --- /dev/null +++ b/proxy/adjudicatetool_test.go @@ -0,0 +1,428 @@ +package proxy_test + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/tidwall/gjson" + + "github.com/rossoctl/context-guru/internal/adjudicate" +) + +// sweepPipeline is a pipeline that CONTAINS extract_llm_sweep, i.e. one that can actually adjudicate. +// The gate on the injection is (Anthropic route AND this component present), so a fixture built with +// `pipeline: []` cannot exercise the injection at all — and a test that asserted the tool appeared +// under `pipeline: []` was asserting the defect: the tool was reaching the `off` control arm of every +// published comparison. See proxy.chat. +const sweepPipeline = "pipeline: [extract_llm_sweep]\n" + +// forwardedBody posts one request through the proxy on the ANTHROPIC route with a sweep-bearing +// pipeline, and returns exactly what reached upstream. +func forwardedBody(t *testing.T, body []byte) []byte { + t.Helper() + return forwardedOn(t, "/anthropic/v1/messages", sweepPipeline, body) +} + +// forwardedOn is forwardedBody with the route and pipeline spelled out, for the cases whose whole +// point is that one of those two differs. +func forwardedOn(t *testing.T, route, yaml string, body []byte) []byte { + t.Helper() + var got []byte + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got, _ = io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + if strings.Contains(route, "anthropic") { + _, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` + + `"content":[{"type":"text","text":"ok"}],"stop_reason":"end_turn",` + + `"usage":{"input_tokens":5,"output_tokens":1}}`)) + return + } + _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) + })) + defer upstream.Close() + h, _ := buildHandler(t, yaml, upstream.URL) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + resp, err := http.Post(srv.URL+route, "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + return got +} + +// toolsRequest is the ANTHROPIC dialect, because that is the only dialect the injection targets: +// prefixAskerFor returns nil for every other provider and cheapmodel/openai.go has no CompletePrefixed +// at all, so the definition could never be read there. +func toolsRequest(t *testing.T, msgs ...map[string]any) []byte { + t.Helper() + b, err := json.Marshal(map[string]any{ + "model": "claude-sonnet-5", + "max_tokens": 64, + "tools": []any{map[string]any{"name": "read_file", "description": "read a file", + "input_schema": map[string]any{"type": "object"}}}, + "messages": msgs, + }) + if err != nil { + t.Fatal(err) + } + return b +} + +// streamingToolsRequest is toolsRequest with stream:true, for the SSE splice path. +func streamingToolsRequest(t *testing.T, msgs ...map[string]any) []byte { + t.Helper() + b, err := json.Marshal(map[string]any{ + "model": "claude-sonnet-5", + "max_tokens": 64, + "stream": true, + "tools": []any{map[string]any{"name": "read_file", "description": "read a file", + "input_schema": map[string]any{"type": "object"}}}, + "messages": msgs, + }) + if err != nil { + t.Fatal(err) + } + return b +} + +// toolsRequestOpenAI is the same request in the OpenAI dialect, for the provider half of the gate. +func toolsRequestOpenAI(t *testing.T, msgs ...map[string]any) []byte { + t.Helper() + b, err := json.Marshal(map[string]any{ + "model": "gpt-x", + "tools": []any{map[string]any{"type": "function", "function": map[string]any{"name": "read_file"}}}, + "messages": msgs, + }) + if err != nil { + t.Fatal(err) + } + return b +} + +// ADVERTISED ON EVERY TURN, not only on the turn a sweep is about to ask. `tools` hashes ahead of +// system and messages, so a tool that appears when the sweep fires and disappears on the next turn +// invalidates the cached prefix from position zero — the flap expand's InjectAuto exists to prevent. +// And the declaration is the whole mechanism, measured over three passes per arm on the same transcript: +// with the tool declared and no tool_choice, unparseable replies ran 9.1% (7 of 77 replied asks) against +// 30.0% (6 of 20) on main's tool_choice:none — Fisher two-tailed p = 0.0245 — and 55.8% of replies came +// back carrying a tool_use where main produced none at all. Dropping tool_choice WITHOUT declaring the +// tool is worse than main (58.3% unparseable), which is why the DECLARATION and not the tool_choice +// removal is what earns its place. The "6 of 6 against 0 of 6" this comment used to cite was a six-item +// hand pass and is retracted: main returns verdicts on 71.5% of the items it asks about, so the defect +// is that ~30% of its asks come back unusable, not that all of them do. +func TestAdjudicateToolAdvertisedOnEveryTurn(t *testing.T) { + got := forwardedBody(t, toolsRequest(t, map[string]any{"role": "user", "content": "go"})) + if len(got) == 0 { + t.Fatal("nothing reached upstream") + } + if !strings.Contains(string(got), adjudicate.ToolName) { + t.Fatalf("the adjudication tool was not advertised on an ordinary turn: %s", got) + } + // Appended last, after the client's own tool, so the client's order is untouched. + tools := gjson.GetBytes(got, "tools").Array() + if n := len(tools); n < 2 || tools[n-1].Get("name").String() != adjudicate.ToolName { + t.Errorf("our tool is not last in the tools array: %s", gjson.GetBytes(got, "tools").Raw) + } + if tools[0].Get("name").String() != "read_file" { + t.Error("the client's own tool moved; its order must be preserved exactly") + } + // A turn with nothing to adjudicate must carry it too — that is the point of injecting always. + got2 := forwardedBody(t, toolsRequest(t, map[string]any{"role": "user", "content": "and again"})) + if !strings.Contains(string(got2), adjudicate.ToolName) { + t.Errorf("the tool came and went between turns, which discards the whole cached prefix: %s", got2) + } +} + +// A bypassed compaction request must not get it, for the same reason expand skips one: bypass promises +// a byte-identical forward. +func TestAdjudicateToolNotAdvertisedOnAnAgentCompaction(t *testing.T) { + got := forwardedBody(t, toolsRequest(t, map[string]any{"role": "user", "content": ccCompactPrompt})) + if strings.Contains(string(got), adjudicate.ToolName) { + t.Errorf("injected into a bypassed compaction, which must forward byte-identically: %s", got) + } +} + +// A stray call the AGENT made must be answered on the request path. The client cannot execute a tool +// the proxy injected, so it answers "not found" and the agent loses a turn to a dead end. +func TestAdjudicateStrayCallIsAnsweredOnTheRequestPath(t *testing.T) { + before := adjudicate.StrayAnswered() + got := forwardedBody(t, toolsRequest(t, + map[string]any{"role": "user", "content": "go"}, + map[string]any{"role": "assistant", "content": []any{map[string]any{ + "type": "tool_use", "id": "c1", "name": adjudicate.ToolName, "input": map[string]any{}, + }}}, + map[string]any{"role": "user", "content": []any{map[string]any{ + "type": "tool_result", "tool_use_id": "c1", "is_error": true, + "content": "Error: No such tool available: " + adjudicate.ToolName, + }}}, + )) + if strings.Contains(string(got), "No such tool available") { + t.Errorf("the client's dead-end refusal was forwarded to the model unchanged: %s", got) + } + if !strings.Contains(string(got), "runs automatically") { + t.Errorf("no substitute answer was written: %s", got) + } + if adjudicate.StrayAnswered() == before { + t.Error("the stray was not counted; adjudicate_stray is the only signal that the tool's " + + "description stopped working") + } +} + +// THE GATE, half one: a pipeline with no extract_llm_sweep can never adjudicate, so advertising the +// tool there buys nothing and costs the cacheable prefix. `off` is the A/B CONTROL ARM of every +// published comparison in this repo, and codesmart is a shipped preset with no sweep in it; injecting +// unconditionally perturbed both by a measured 946 bytes at the head of the prefix on every request. +func TestAdjudicateToolNotAdvertisedWhenThePipelineCannotAdjudicate(t *testing.T) { + for _, yaml := range []string{"pipeline: []\n", "pipeline: [format]\n"} { + got := forwardedOn(t, "/anthropic/v1/messages", yaml, + toolsRequest(t, map[string]any{"role": "user", "content": "go"})) + if len(got) == 0 { + t.Fatalf("%s: nothing reached upstream", yaml) + } + if strings.Contains(string(got), adjudicate.ToolName) { + t.Errorf("%s: advertised on a pipeline that cannot adjudicate: %s", yaml, got) + } + } +} + +// THE GATE, half two: the provider. prefixAskerFor returns nil for anything but Anthropic and +// cheapmodel/openai.go has no CompletePrefixed at all, so on the OpenAI route the definition is +// unreachable by construction — it was pure prefix cost, ~217 tokens per request. +func TestAdjudicateToolNotAdvertisedOnANonAnthropicRoute(t *testing.T) { + got := forwardedOn(t, "/openai/v1/chat/completions", sweepPipeline, + toolsRequestOpenAI(t, map[string]any{"role": "user", "content": "go"})) + if len(got) == 0 { + t.Fatal("nothing reached upstream") + } + if strings.Contains(string(got), adjudicate.ToolName) { + t.Errorf("advertised on a route that can never read it: %s", got) + } +} + +// THE LEAK, non-streaming path. A tool_use for a proxy-injected tool must never reach the client: the +// client never declared it, cannot execute it, and answers "not found", losing the agent a turn. It +// must instead be answered IN BAND, before the client is written to, which leaves the request-path +// repair as a backstop rather than the primary defence. +func TestAdjudicateStrayCallDoesNotReachTheClientOnTheJSONPath(t *testing.T) { + round := 0 + var second []byte + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + round++ + w.Header().Set("Content-Type", "application/json") + if round == 1 { + // The model calls OUR tool, which is always a defect by construction. + _, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` + + `"content":[{"type":"tool_use","id":"stray1","name":"` + adjudicate.ToolName + + `","input":{"verdicts":[]}}],"stop_reason":"tool_use",` + + `"usage":{"input_tokens":5,"output_tokens":1}}`)) + return + } + second = body + _, _ = w.Write([]byte(`{"id":"m2","type":"message","role":"assistant","model":"claude",` + + `"content":[{"type":"text","text":"done"}],"stop_reason":"end_turn",` + + `"usage":{"input_tokens":6,"output_tokens":2}}`)) + })) + defer upstream.Close() + h, _ := buildHandler(t, sweepPipeline, upstream.URL) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + resp, err := http.Post(srv.URL+"/anthropic/v1/messages", "application/json", + bytes.NewReader(toolsRequest(t, map[string]any{"role": "user", "content": "go"}))) + if err != nil { + t.Fatal(err) + } + got, _ := io.ReadAll(resp.Body) + resp.Body.Close() + + // PRECONDITION: the loop must actually have run a second round, or this asserts nothing. + if round < 2 { + t.Fatalf("the stray was never intercepted -- only %d upstream round(s), client got: %s", + round, got) + } + if strings.Contains(string(got), adjudicate.ToolName) { + t.Errorf("the proxy-injected tool_use reached the CLIENT: %s", got) + } + // Answered in band, so the model could finish its turn rather than wait for a tool nobody runs. + if !strings.Contains(string(second), "runs automatically") { + t.Errorf("the stray was withheld but never answered upstream: %s", second) + } + if !strings.Contains(string(got), "done") { + t.Errorf("the client did not receive the finished turn: %s", got) + } +} + +// THE HOLE IN "answered in band on both paths", pinned rather than papered over. When the model calls +// our tool AND a client tool in the SAME assistant turn, expand.ResponseCalls reports otherTools, the +// response loop bail()s, and our tool_use reaches the client raw — the loop DOES see this path and +// defers it deliberately, because it cannot continue a turn whose other tool_use only the client can +// execute without either inventing a result for the client's tool or dropping the client's call. +// +// This test asserts what actually happens today, not what would be nicer: the leak this turn, the +// repair on the next request, the client's own tool_result untouched, and the stray counted exactly +// once. It exists so that the behaviour is a decision on the record rather than something a later +// change "fixes" by guessing. If the deferral is ever replaced by a real in-band answer for co-called +// turns, this test SHOULD fail and be rewritten — that is the point of pinning it. +func TestAdjudicateStrayCoCalledWithClientToolLeaks(t *testing.T) { + // The assistant turn both requests share: one call to the CLIENT's tool, one to ours. + assistantCoCall := map[string]any{"role": "assistant", "content": []any{ + map[string]any{"type": "tool_use", "id": "cli1", "name": "read_file", + "input": map[string]any{"path": "a.go"}}, + map[string]any{"type": "tool_use", "id": "stray1", "name": adjudicate.ToolName, + "input": map[string]any{"verdicts": []any{}}}, + }} + + // --- Turn 1: the leak. ------------------------------------------------------------------- + rounds := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.ReadAll(r.Body) + rounds++ + w.Header().Set("Content-Type", "application/json") + blocks, _ := json.Marshal(assistantCoCall["content"]) + _, _ = w.Write([]byte(`{"id":"m1","type":"message","role":"assistant","model":"claude",` + + `"content":` + string(blocks) + `,"stop_reason":"tool_use",` + + `"usage":{"input_tokens":5,"output_tokens":2}}`)) + })) + defer upstream.Close() + h, _ := buildHandler(t, sweepPipeline, upstream.URL) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + resp, err := http.Post(srv.URL+"/anthropic/v1/messages", "application/json", + bytes.NewReader(toolsRequest(t, map[string]any{"role": "user", "content": "go"}))) + if err != nil { + t.Fatal(err) + } + leaked, _ := io.ReadAll(resp.Body) + resp.Body.Close() + + // PRECONDITION: exactly one upstream round. A continuation would mean the loop answered in band + // after all, and then the rest of this test is asserting nothing about the co-call path. + if rounds != 1 { + t.Fatalf("expected the loop to bail on otherTools after ONE round, got %d — the co-call path "+ + "no longer defers, so this test's premise is gone: %s", rounds, leaked) + } + // The leak itself. Asserted, not lamented: this is the documented cost of the deferral. + if !strings.Contains(string(leaked), adjudicate.ToolName) { + t.Errorf("expected the co-called proxy tool_use to reach the client this turn (the known, "+ + "deliberate deferral); it did not, so the response loop changed: %s", leaked) + } + if !strings.Contains(string(leaked), "cli1") { + t.Errorf("the CLIENT's own tool_use did not survive the round: %s", leaked) + } + + // --- Turn 2: the repair. ----------------------------------------------------------------- + // The client did what a client does: ran its own tool, and answered ours "not found" because the + // PROXY injected it and the client never declared it. + before := adjudicate.StrayAnswered() + got := forwardedBody(t, toolsRequest(t, + map[string]any{"role": "user", "content": "go"}, + assistantCoCall, + map[string]any{"role": "user", "content": []any{ + map[string]any{"type": "tool_result", "tool_use_id": "cli1", + "content": "package main"}, + map[string]any{"type": "tool_result", "tool_use_id": "stray1", "is_error": true, + "content": "Error: No such tool available: " + adjudicate.ToolName}, + }}, + )) + if strings.Contains(string(got), "No such tool available") { + t.Errorf("the client's dead-end refusal reached the model unchanged: %s", got) + } + if !strings.Contains(string(got), "runs automatically") { + t.Errorf("the stray was leaked AND never repaired on the next request: %s", got) + } + // The client's own tool_result must come through untouched — the repair keys off the tool_use it + // answers, so a bug here would rewrite somebody else's result. + blocks := gjson.GetBytes(got, `messages.2.content`).Array() + if len(blocks) != 2 { + t.Fatalf("the repaired turn does not have both tool_results: %s", got) + } + if c := blocks[0].Get("content").String(); c != "package main" { + t.Errorf("the CLIENT's tool_result was rewritten to %q; the repair must only touch ours", c) + } + if blocks[0].Get("is_error").Exists() { + t.Errorf("is_error was invented on the client's own tool_result: %s", blocks[0].Raw) + } + // Ours: answered, and no longer an error — leaving is_error set tells the model its call failed + // while handing it that call's answer. + if blocks[1].Get("is_error").Bool() { + t.Errorf("is_error stayed set on the repaired block: %s", blocks[1].Raw) + } + if n := adjudicate.StrayAnswered() - before; n != 1 { + t.Errorf("the leaked stray was counted %d times, want exactly 1 — adjudicate_stray is the "+ + "only signal that the tool's description stopped working", n) + } +} + +// THE LEAK, streaming path. The splicer withheld only the expand tool by name, so an adjudication call +// streamed through event by event and the client saw it live. +func TestAdjudicateStrayCallDoesNotReachTheClientOnTheSSEPath(t *testing.T) { + round := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.ReadAll(r.Body) + round++ + if round == 1 { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + for _, ev := range []string{ + `event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"m1","type":"message","role":"assistant","model":"claude","content":[],"usage":{"input_tokens":5,"output_tokens":0}}}`, + `event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"stray1","name":"` + adjudicate.ToolName + `","input":{}}}`, + `event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`, + `event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":1}}`, + `event: message_stop` + "\n" + `data: {"type":"message_stop"}`, + } { + _, _ = w.Write([]byte(ev + "\n\n")) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + } + return + } + // The continuation MUST also stream. The request said stream:true, and a JSON body on a + // later round is a documented upstream anomaly that cannot be spliced into an event + // stream -- the loop then hands the withheld events back, which is the very leak this + // test is trying to observe. A fixture that answers in JSON therefore fails for a reason + // that has nothing to do with the withhold set. + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + for _, ev := range []string{ + `event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"m2","type":"message","role":"assistant","model":"claude","content":[],"usage":{"input_tokens":6,"output_tokens":0}}}`, + `event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"done"}}`, + `event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`, + `event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}`, + `event: message_stop` + "\n" + `data: {"type":"message_stop"}`, + } { + _, _ = w.Write([]byte(ev + "\n\n")) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + } + })) + defer upstream.Close() + h, _ := buildHandler(t, sweepPipeline, upstream.URL) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + body := streamingToolsRequest(t, map[string]any{"role": "user", "content": "go"}) + resp, err := http.Post(srv.URL+"/anthropic/v1/messages", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + got, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if round < 2 { + t.Fatalf("the streamed stray was never intercepted -- only %d round(s), client got: %s", + round, got) + } + if strings.Contains(string(got), adjudicate.ToolName) { + t.Errorf("the proxy-injected tool_use was STREAMED to the client: %s", got) + } + if !strings.Contains(string(got), "done") { + t.Errorf("the client did not receive the finished turn: %s", got) + } +} diff --git a/proxy/prefixask_test.go b/proxy/prefixask_test.go index 5df342d..b00a6ba 100644 --- a/proxy/prefixask_test.go +++ b/proxy/prefixask_test.go @@ -11,7 +11,9 @@ import ( bschemas "github.com/maximhq/bifrost/core/schemas" "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/internal/adjudicate" "github.com/rossoctl/context-guru/internal/cheapmodel" + "github.com/rossoctl/context-guru/internal/extract" ) // The prefix ask exists for ONE reason: to read the provider's prompt cache instead of paying fresh @@ -121,11 +123,18 @@ func TestCompletePrefixedAppendsWithoutDisturbingThePrefix(t *testing.T) { if len(sent.System) != 1 || sent.System[0]["cache_control"] == nil { t.Errorf("the prefix's cache_control was lost, so there is no entry to read: %v", sent.System) } - // TOOL_CHOICE: NONE is required and free. It is not part of the cache key, and without it the - // prefix's tools make the model answer with a tool_use instead of a verdict. - if sent.ToolChoice == nil || sent.ToolChoice["type"] != "none" { - t.Errorf("tool_choice is not forced to none, so the model will answer with a tool_use: %v", - sent.ToolChoice) + // NO TOOL_CHOICE AT ALL, and this reverses what this test used to assert. Two separate findings on + // the same prefix with only tool_choice varying. On the CACHE: {"type":"tool",name} MISSED the entry + // and wrote 8,378 tokens against the 8,268 already there, so naming a tool DOES participate in the + // key, while omitting tool_choice read that entry for free. On the ANSWER: with the tool declared + // and no tool_choice, unparseable replies ran 9.1% (7 of 77 replied asks) against 30.0% (6 of 20) + // under main's {"type":"none"}, Fisher two-tailed p = 0.0245 — setting `none` to PREVENT a tool_use + // is what drove the model into prose, which the caller then scored as an unparseable failure. + // The "0 of 6 / 6 of 6" verdict counts this comment used to cite came from a six-item hand pass and + // are retracted; main returns verdicts on 71.5% of the items it asks about, not none of them. + if sent.ToolChoice != nil { + t.Errorf("tool_choice was set (%v); `none` drives the model into prose and a named tool costs "+ + "a fresh cache write", sent.ToolChoice) } // stream is the one deliberate removal: the caller wants a single JSON answer. if sent.Stream != nil { @@ -133,6 +142,74 @@ func TestCompletePrefixedAppendsWithoutDisturbingThePrefix(t *testing.T) { } } +// THE TOOL INPUT IS PREFERRED OVER TEXT. With the tool declared in the prefix the model answers by +// calling it, and that input is already schema-shaped — which is the whole point: it removes prose, +// partial batches, and an array truncated by the output budget. The raw input is returned so the +// caller's existing parser reads it unchanged, and a `Read` call sitting alongside must NOT be +// mistaken for the answer. +func TestCompletePrefixedPrefersOurToolInputOverText(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("content-type", "application/json") + _, _ = w.Write([]byte(`{"content":[` + + `{"type":"thinking","thinking":"weighing the outputs"},` + + `{"type":"text","text":"I think output 1 is still needed."},` + + `{"type":"tool_use","id":"t0","name":"Read","input":{"path":"a.py"}},` + + `{"type":"tool_use","id":"t1","name":"` + adjudicate.ToolName + `",` + + `"input":{"verdicts":[{"i":1,"needed_by":"none","verdict":"drop"}]}}],` + + `"usage":{"input_tokens":1,"output_tokens":2,"cache_creation_input_tokens":0,` + + `"cache_read_input_tokens":8268}}`)) + })) + t.Cleanup(srv.Close) + cli := cheapmodel.Anthropic{BaseURL: srv.URL, Model: "claude-sonnet-5", APIKey: "k"} + reply, u, err := cli.CompletePrefixed(context.Background(), []byte(prefixBodyFixture), "ASK") + if err != nil { + t.Fatalf("CompletePrefixed: %v", err) + } + // REPORTED, not just used. The caller counts which reply shape it got, and it cannot infer this + // one: extract.ParseVerdicts reads a tool_use input and a JSON array in text identically, so a + // run whose declared tool is never touched looks exactly like one where it always is. That + // ambiguity is what left the reviewer's "0 of 5 asks used the tool" and this PR's "6 of 6" + // un-adjudicable against each other. See components.PrefixUsage.ViaTool. + if !u.ViaTool { + t.Error("the answer arrived as a tool_use but was not reported as one") + } + if strings.Contains(reply, "I think output 1") { + t.Errorf("the prose text was returned in preference to the tool input: %q", reply) + } + if strings.Contains(reply, "a.py") { + t.Errorf("the AGENT's own Read call was mistaken for the answer: %q", reply) + } + if !strings.Contains(reply, `"verdicts"`) || !strings.Contains(reply, `"drop"`) { + t.Errorf("the tool input did not reach the caller verbatim: %q", reply) + } + // The existing text parser must read that input unchanged — it scans for an array that decodes, + // and the tool's `verdicts` array is one. This is what keeps the fix additive. + vs, ok := extract.ParseVerdicts(reply) + if !ok || len(vs) != 1 || vs[0].Label != 1 || vs[0].Verdict != "drop" { + t.Errorf("extract.ParseVerdicts could not read the tool input: ok=%v verdicts=%+v", ok, vs) + } +} + +// With no tool_use in the reply the TEXT path must still work: main's parse path and its fallback are +// untouched by this change, and a model that answers in prose anyway is read exactly as before. +func TestCompletePrefixedStillReadsTextWhenNoToolWasCalled(t *testing.T) { + srv := newCapturePrefixed(t, 8268) + cli := cheapmodel.Anthropic{BaseURL: srv.srv.URL, Model: "m", APIKey: "k"} + reply, u, err := cli.CompletePrefixed(context.Background(), []byte(prefixBodyFixture), "ASK") + // And the shape is reported as PROSE, so the two counters cannot both be inflated by the same + // reply. A ViaTool that defaulted to true would make every prose answer look like a tool answer, + // which is the exact confusion the field exists to remove. + if u.ViaTool { + t.Error("a text-only reply was reported as having arrived via the tool") + } + if err != nil { + t.Fatalf("CompletePrefixed: %v", err) + } + if reply != "[]" { + t.Errorf("reply = %q; the text path must survive unchanged", reply) + } +} + // A prefix body with no messages array cannot be appended to, and must fail loudly rather than // sending something that reads no cache. func TestCompletePrefixedRefusesABodyWithNoMessages(t *testing.T) { diff --git a/proxy/promexport.go b/proxy/promexport.go index e1f0b34..4215a80 100644 --- a/proxy/promexport.go +++ b/proxy/promexport.go @@ -13,6 +13,7 @@ import ( "github.com/rossoctl/context-guru/components/offload" "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/internal/adjudicate" "github.com/rossoctl/context-guru/internal/cheapmodel" "github.com/rossoctl/context-guru/metrics" "github.com/rossoctl/context-guru/store" @@ -429,6 +430,14 @@ func (h *Handler) renderMetrics() string { promLine(&b, "cg_frozen_decisions_total", `outcome="repaired"`, float64(repaired)) } + // Same rule as the two families above, and this counter would have had the same bug: + // Snapshot.AdjudicateStray is filled only by the /stats handler (proxy.go), so a promLine + // off `s` here would export a hard-wired 0 on every scrape however often the agent called + // the tool. Read from adjudicate's own counter instead. + promHeaderProc(&b, "cg_adjudicate_stray_total", + "Times the AGENT called the proxy-injected context_guru_adjudicate tool, which it is told not to. Each one is a lost agent turn, not a correctness failure; rising means the tool's description has stopped working.", "counter") + promLine(&b, "cg_adjudicate_stray_total", "", float64(adjudicate.StrayAnswered())) + promHeaderProc(&b, "cg_sse_streams_total", "Responses by streaming path.", "counter") promLine(&b, "cg_sse_streams_total", `path="streamed"`, float64(s.SSEStreamed)) promLine(&b, "cg_sse_streams_total", `path="buffered"`, float64(s.SSEBuffered)) diff --git a/proxy/promexport_coverage_test.go b/proxy/promexport_coverage_test.go index 8b71e4f..7e07772 100644 --- a/proxy/promexport_coverage_test.go +++ b/proxy/promexport_coverage_test.go @@ -1,6 +1,7 @@ package proxy import ( + "fmt" "os" "reflect" "regexp" @@ -8,6 +9,7 @@ import ( "testing" "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/internal/adjudicate" "github.com/rossoctl/context-guru/metrics" ) @@ -24,9 +26,12 @@ import ( // notExportedWhy with a reason. Adding a field to Snapshot without doing one of the two // FAILS THE BUILD. // -// EXPECTED TO FAIL ON PR #137, and that is the test working as intended: that PR adds an -// `adjudicate_stray` field to Snapshot and exports it nowhere. The fix is to export it, or -// to list it below with an honest reason — not to delete this test. +// This CAUGHT PR #137, which added an `adjudicate_stray` field to Snapshot and exported it +// nowhere — the test working exactly as intended. Resolved the way the second group below +// prescribes: the series IS exported, as cg_adjudicate_stray_total, but sourced from +// adjudicate.StrayAnswered() rather than off `s`, because the /stats handler fills that field +// AFTER renderMetrics takes its snapshot, so a promLine off `s` would have exported a +// permanent 0 while passing this test. // // What it checks is that the exporter READS the field, not that a particular series // renders, and NOT that the value is real: `s` in renderMetrics is the bare aggregator @@ -76,6 +81,7 @@ var notExportedWhy = map[string]string{ // renderMetrics holds the bare aggregator snapshot, where those fields are zero. "ExpandUnresolvedMalformed": `cg_expand_unresolved_total{reason="malformed"}, from expand.Unresolved()`, "ExpandUnresolvedMissing": `cg_expand_unresolved_total{reason="missing"}, from expand.Unresolved()`, + "AdjudicateStray": "cg_adjudicate_stray_total, from adjudicate.StrayAnswered()", "LLMCalls": "cg_llm_calls_total, from cheapmodel.Usage()", "LLMInputTokens": `cg_llm_tokens_total{direction="input"}, from cheapmodel.Usage()`, "LLMOutputTokens": `cg_llm_tokens_total{direction="output"}, from cheapmodel.Usage()`, @@ -173,6 +179,58 @@ func TestExpandUnresolvedSeriesRender(t *testing.T) { } } +// TestAdjudicateStraySeriesRender is the guard half of the pattern TestExpandUnresolvedSeriesRender +// establishes, applied to cg_adjudicate_stray_total: the line RENDERS even at zero, and its value MOVES +// when the counter behind it does. +// +// The second half is the whole point, and this PR shipped without it. `s` in renderMetrics is the bare +// aggregator snapshot; Snapshot.AdjudicateStray is filled by the /stats handler AFTER that snapshot is +// taken, so sourcing this series from `float64(s.AdjudicateStray)` exports a hard-wired 0 on every +// scrape however often the agent calls the tool. TestEverySnapshotFieldIsExportedOrExempt cannot catch +// that, by its own documented design: it checks that the exporter READS the field, not that the value is +// real. Nothing else in ./proxy caught it either — the suite stayed green with that revert applied, +// which is what a reviewer demonstrated. This test is the thing that fails on it. +// +// Baseline-relative rather than absolute because strayAnswered is a process-wide counter and the +// adjudicate tests in package proxy_test share this test binary with it. +func TestAdjudicateStraySeriesRender(t *testing.T) { + h := New(nil, nil, metrics.NewAggregator(), Options{}) + before := adjudicate.StrayAnswered() + want := fmt.Sprintf("cg_adjudicate_stray_total %d", before) + if !containsLine(h.renderMetrics(), want) { + t.Fatalf("/metrics is missing the line %q — a family that appears only once something breaks "+ + "renders \"No data\" in Grafana, which reads as healthy", want) + } + if help := helpLine(h.renderMetrics(), "cg_adjudicate_stray_total"); help == "" { + t.Error("cg_adjudicate_stray_total renders with no HELP, so nothing on the panel says what a " + + "non-zero value means") + } + + // And the value moves. Two strays answered in band; the series must read two higher. + adjudicate.NoteAnsweredInBand(2) + body := h.renderMetrics() + if containsLine(body, want) { + t.Errorf("still reads %q after two answered strays — the series is not reading "+ + "adjudicate.StrayAnswered() (Snapshot.AdjudicateStray is filled only by /stats, so a "+ + "promLine off `s` exports a permanent 0)", want) + } + if now := fmt.Sprintf("cg_adjudicate_stray_total %d", before+2); !containsLine(body, now) { + t.Errorf("expected the line %q, got:\n%s", now, + strings.Join(linesWithPrefix(body, "cg_adjudicate_stray_total"), "\n")) + } +} + +// linesWithPrefix is for failure messages: showing the series that DID render is what tells you whether +// the value was stale or the line vanished altogether. +func linesWithPrefix(body, prefix string) (out []string) { + for _, ln := range strings.Split(body, "\n") { + if strings.HasPrefix(ln, prefix) { + out = append(out, ln) + } + } + return out +} + func containsLine(body, line string) bool { for _, ln := range strings.Split(body, "\n") { if ln == line { diff --git a/proxy/proxy.go b/proxy/proxy.go index 5d65a96..531f525 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -30,6 +30,7 @@ import ( "github.com/rossoctl/context-guru/components/offload" "github.com/rossoctl/context-guru/dash" "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/internal/adjudicate" "github.com/rossoctl/context-guru/internal/cheapmodel" "github.com/rossoctl/context-guru/internal/logging" "github.com/rossoctl/context-guru/internal/modelinfo" @@ -1097,6 +1098,24 @@ func (h *Handler) chat(provider bschemas.ModelProvider, static upstream, pick fu // and the request side caught it. lg.Debug("cg.expand_repair", "restored", repaired) } + // The same problem for the adjudication tool, and the same remedy. The client + // cannot execute a tool the proxy injected, so a call the AGENT made to + // context_guru_adjudicate comes back as "not found" and the agent loses a turn + // to a dead end. Not defensive: a model was directly observed calling + // context_guru_expand at step 2 of a run. Measured at 0 strays across ~4,900 + // requests with the "do not call this yourself" description, and 0 again across + // three further benchmark passes. This is now the BACKSTOP, for two distinct paths. + // The response loop answers a stray in band when the turn called proxy tools only. + // It DEFERS to this repair when the model co-called a client tool in the same turn + // (otherTools -> bail, see the response loop), because it cannot continue a turn + // whose other tool_use only the client can execute — on that path our tool_use does + // reach the client and this repair is the only thing that answers it, at a cost of + // one agent turn. It also catches a round the loop could not reconstruct at all. + // The counter (adjudicate_stray) is what says whether the description still works. + var strays int + if body, strays = adjudicate.AnswerStrayCalls(string(provider), body); strays > 0 { + lg.Debug("cg.adjudicate_stray", "answered", strays) + } } var added time.Duration body, added, tr = h.applyMode(&reqInfo{ @@ -1178,6 +1197,36 @@ func (h *Handler) chat(provider bschemas.ModelProvider, static upstream, pick fu if im != expand.InjectAuto || tn.Pipe.HasOffload() { body, _ = expand.Inject(string(provider), im, body, tn.Store.Persists()) } + // The adjudication tool, on every request of a pipeline that can ACTUALLY + // adjudicate, and on no other. + // + // The cache argument for injecting it unconditionally is real but narrower than it + // was first applied. `tools` hashes before system and messages, so a tool that + // appears on the turn a sweep fires and vanishes on the next invalidates the prefix + // from position zero — the flap expand's `always` mode exists to prevent. That + // forbids gating on anything that varies PER TURN. It does not forbid these two + // conditions: pipeline membership is fixed per config DOCUMENT and the provider is + // fixed by the route, so both are byte-stable for every request under a given config + // and the prefix never flaps. "Fixed at config load" is what this used to say and it + // is false: tenancy.go rebuilds a tenant's *Pipeline when the config document + // changes, mid-session, by its own design. The conclusion survives, because a config + // change that adds or drops extract_llm_sweep has already invalidated the prefix for + // much bigger reasons than this tool's 946 bytes — but the premise had to be stated + // as what the code actually guarantees. + // + // Injecting it unconditionally instead cost 946 bytes (measured on the wire) at the + // head of the cacheable prefix of EVERY preset, including `off` — which is the + // control arm of every published comparison in this repo — and including presets + // like codesmart that contain no extract_llm_sweep and so can never adjudicate. + // The provider half is not cosmetic either: prefixAskerFor returns nil for anything + // but Anthropic and cheapmodel/openai.go has no CompletePrefixed at all, so on the + // OpenAI route the ~217-token definition is unreachable by construction. + // + // Kept rather than dropped because a three-way A/B showed the DECLARATION is what + // makes removing tool_choice:none safe; see cheapmodel.CompletePrefixed. + if provider == bschemas.Anthropic && tn.Pipe.Has("extract_llm_sweep") { + body, _ = adjudicate.Inject(string(provider), body) + } } }() // Load the request's one INFO line with everything the pipeline decided. serve @@ -1293,7 +1342,10 @@ func (h *Handler) serve(w http.ResponseWriter, r *http.Request, provider bschema // and does the same job: sseSplicer withholds only the block that calls the expand tool // and streams everything else as it arrives (see ssepeek.go). Non-Anthropic dialects are // not inspected at all. - advertised := expand.HasTool(string(provider), body) + // Advertised covers BOTH proxy-injected tools. The response loop is what keeps a + // proxy-injected tool_use away from the client, and gating it on expand alone let an + // adjudication call stream straight through on a request that advertised only that one. + advertised := expand.HasTool(string(provider), body) || adjudicate.HasTool(string(provider), body) // SSE accounting is PER CLIENT REQUEST, not per upstream round: one client request // that drives several expand rounds waited for all of them, so timing a single // round would report a healthy TTFB for a client that waited three round-trips. @@ -1456,7 +1508,7 @@ func (h *Handler) serve(w http.ResponseWriter, r *http.Request, provider bschema // withheld and round 1's message_stop may be the only terminator this client // will ever get (see the terminator check below). prevWithheld = withheld - respBody, withheld, found = sp.pass(resp.Body, expand.ToolName) + respBody, withheld, found = sp.pass(resp.Body, expand.ToolName, adjudicate.ToolName) resp.Body.Close() switch { case found && sp.blocks == 0: @@ -1523,6 +1575,16 @@ func (h *Handler) serve(w http.ResponseWriter, r *http.Request, provider bschema writeRaw(w, resp, respBody) return } + // KNOWN CONFLATION, recorded rather than fixed. This counts "the peek withheld a + // proxy tool_use and handed it back", but the metric is named for expand, so an + // adjudicate leak (the co-call deferral below) lands in the expand bucket. Not split + // here for three reasons: the field is /stats-only and reaches no dashboard or alert + // (notExportedWhy calls it "NOT EXPORTED YET"); two of the three bail sites — a round + // that failed SSE aggregation, and a spent maxExpandRounds — cannot attribute the leak + // at all, because nothing has parsed the turn yet; and a leak can be BOTH at once, so + // the honest split is two counters and a new exported family with its own render and + // vacuity guard, which is a metrics change rather than this PR's subject. Filed + // separately; until then, read this counter as "proxy tool_use reached the client". if withheld != nil && h.agg != nil { h.agg.RecordSSEExpandAfterStream() } @@ -1551,15 +1613,41 @@ func (h *Handler) serve(w http.ResponseWriter, r *http.Request, provider bschema msg = agg } - calls, otherTools := expand.ResponseCalls(string(provider), msg) - if len(calls) == 0 || otherTools { - bail() // normal answer (or other tools) — hand it over unchanged + calls, otherTools := expand.ResponseCalls(string(provider), msg, adjudicate.ToolName) + // Calls the AGENT made to the adjudication tool. Answered HERE, in band, whenever this turn + // called PROXY tools only: by the time the request-path repair runs, the client has already + // received a tool_use for a tool it does not implement, already answered "not found", and + // already lost the turn. + // + // NOT on every path, and the exception is worth more than the claim it replaces. When the model + // calls this tool AND a client tool in the same assistant turn, otherTools is true, the bail() + // below hands the round to the client whole, and our tool_use reaches it raw — exactly as it did + // before this change. That is a deliberate deferral, not an oversight: answering in band means + // continuing the turn upstream, and this loop cannot continue a turn whose other tool_use only + // the CLIENT can execute. It would have to invent a result for the client's tool or drop the + // client's call, and both are worse than one lost turn. adjudicate.AnswerStrayCalls repairs it + // on the NEXT request instead: the substitute answer lands, is_error clears, the client's own + // tool_result is untouched, and the stray is counted — so fail-open holds and the price is one + // agent turn, not a broken session. Pinned by TestAdjudicateStrayCoCalledWithClientToolLeaks. + // + // So AnswerStrayCalls backstops TWO different things: this co-call path, which the loop DOES + // see and declines, and a round the loop genuinely cannot reconstruct (SSE aggregation failed, + // or maxExpandRounds is spent). An earlier version of this comment said the backstop was only + // for "a path this loop does not see", which was false for the first of those. + strays := adjudicate.ResponseCallIDs(string(provider), msg) + if (len(calls) == 0 && len(strays) == 0) || otherTools { + bail() // normal answer (or a CLIENT tool) — hand it over unchanged return } // Build a tool_result for EVERY expand call — the provider requires one per // tool_call_id or the continuation is malformed. Expired/unknown ids get an // explicit placeholder rather than being omitted. resolved := map[string]string{} + // Every stray adjudication call gets the same definite, uninteresting answer, so the model can + // finish its turn instead of waiting on a tool nobody will run. + for _, id := range strays { + resolved[id] = adjudicate.StrayAnswer + } got := 0 for _, c := range calls { if orig, ok := expand.Resolve(tn.Store, c.HashID); ok { @@ -1594,6 +1682,14 @@ func (h *Handler) serve(w http.ResponseWriter, r *http.Request, provider bschema bail() // malformed shapes — fail open, hand the response over unchanged return } + // Counted only once the continuation exists, NOT when the answer was composed. Incrementing + // before this point double-counted: a Continuation failure bails, the tool_use reaches the + // client after all, and the request-path repair then counts the very same stray a second time + // on the next turn. The counter is meant to be one-per-stray-call however it was answered. + if len(strays) > 0 { + adjudicate.NoteAnsweredInBand(len(strays)) + lg.Debug("cg.adjudicate_stray", "answered_in_band", len(strays), "round", round) + } // got == 0 CONTINUES, and that is the change that let the expand tool be advertised // on every request in a session (expand.InjectAuto). `resolved` already carries a // placeholder for every unresolved id, so the continuation is well formed whether @@ -1801,6 +1897,11 @@ func (h *Handler) stats(w http.ResponseWriter, r *http.Request) { // tokens counts successful re-serves, so a broken stash was indistinguishable from a session // that simply never called expand. See expand/unresolved.go. snap.ExpandUnresolvedMalformed, snap.ExpandUnresolvedMissing = expand.Unresolved() + // Stray calls the AGENT made to the injected adjudication tool, answered on the request path. + // Published because the whole justification for advertising a tool the model is told not to call + // is that it does not call it: measured 0 across ~4,900 requests, and a non-zero figure here says + // the description stopped working — which nothing else in this snapshot can reveal. + snap.AdjudicateStray = adjudicate.StrayAnswered() snap.FrozenHits, snap.FrozenMisses = offload.FrozenStats() if fl, ok := h.store.(*store.Memory); ok { // process store; hosted per-tenant stores report via the dashboard snap.FrozenDropped, snap.FrozenRepaired = fl.FrozenLossStats() diff --git a/proxy/ssepeek.go b/proxy/ssepeek.go index aea640c..56bca77 100644 --- a/proxy/ssepeek.go +++ b/proxy/ssepeek.go @@ -176,7 +176,15 @@ const sseRetainMaxBytes = 16 << 20 // can be intercepted, and everything is forwarded as it arrives. found still reports whether // the response called expand, because the client then receives that call and the caller has // to count it. -func (sp *sseSplicer) pass(body io.Reader, expandTool string) (whole, withheld []byte, found bool) { +// +// ONE REQUIRED NAME plus a variadic tail, rather than a bare `proxyTools ...string`. A bare variadic +// compiles with NO names at all, and an empty withhold set silently disables this entire defence: +// startsProxyToolCall can never match, found stays false, cut stays -1, and every proxy-injected +// tool_use streams straight to the client while the call site reads as though it were protected. The +// two-argument signature this grew out of made that unrepresentable; this keeps that property while +// still taking a set. +func (sp *sseSplicer) pass(body io.Reader, proxyTool string, moreProxyTools ...string) (whole, withheld []byte, found bool) { + proxyTools := append([]string{proxyTool}, moreProxyTools...) br := bufio.NewReader(body) // One event buffer for the round, not one per event: a 2.25 MB round is ~19,000 events. // Worth doing and not where the money was — recorder-free, per-event churn is 9.60x the @@ -190,7 +198,7 @@ func (sp *sseSplicer) pass(body io.Reader, expandTool string) (whole, withheld [ e, err := readSSEEvent(br, &ev) if len(e) > 0 { sent := false - if startsExpandCall(e, expandTool) { + if startsProxyToolCall(e, proxyTools) { found = true if !over && cut < 0 { cut = buf.Len() @@ -333,9 +341,14 @@ func sseEventPayload(ev []byte) string { return "" } -// startsExpandCall reports whether this event OPENS a call to the expand tool — the one -// block a client must never receive, because only this proxy implements the tool. -func startsExpandCall(ev []byte, expandTool string) bool { +// startsProxyToolCall reports whether this event OPENS a tool_use block for ANY tool the proxy +// injected — the one class of block a client must never receive, because only this proxy implements +// those tools. A set rather than one name because the proxy now advertises two — expand and the +// adjudication tool — and withholding only the first let the second stream straight to the client, +// which is the #103 class of defect: a client cannot execute a tool it never declared, so it answers +// "not found" and the agent loses a turn. Unlike expand, an agent call to the adjudication tool is +// never useful, so there is nothing on the other side of the trade. +func startsProxyToolCall(ev []byte, proxyTools []string) bool { if !bytes.Contains(ev, []byte("content_block_start")) { return false // cheap reject: the parse below is the expensive half } @@ -344,7 +357,16 @@ func startsExpandCall(ev []byte, expandTool string) bool { return false } cb := p.Get("content_block") - return cb.Get("type").String() == "tool_use" && cb.Get("name").String() == expandTool + if cb.Get("type").String() != "tool_use" { + return false + } + name := cb.Get("name").String() + for _, t := range proxyTools { + if t != "" && name == t { + return true + } + } + return false } // terminate closes the client's turn if nothing has. A continuation round is not obliged to diff --git a/proxy/ssepeek_test.go b/proxy/ssepeek_test.go index 1e45f4f..9f71107 100644 --- a/proxy/ssepeek_test.go +++ b/proxy/ssepeek_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/internal/adjudicate" ) const ( @@ -50,10 +51,30 @@ func TestStartsExpandCall(t *testing.T) { {"[DONE]", "data: [DONE]\n\n", false}, {"no data line at all", "event: content_block_start\n\n", false}, } { - if got := startsExpandCall([]byte(tc.ev), expand.ToolName); got != tc.want { + if got := startsProxyToolCall([]byte(tc.ev), []string{expand.ToolName}); got != tc.want { t.Errorf("%s: got %v, want %v", tc.name, got, tc.want) } } + + // THE SET, not one name. The proxy injects two tools now, and this predicate deciding on expand + // alone is what streamed an adjudication tool_use to the client event by event: the client never + // declared that tool, cannot run it, and answers "not found". + adj := "event: content_block_start\n" + + `data: {"type":"content_block_start","index":0,"content_block":` + + `{"type":"tool_use","id":"a1","name":"` + adjudicate.ToolName + `","input":{}}}` + + "\n\n" + if startsProxyToolCall([]byte(adj), []string{expand.ToolName}) { + t.Error("fixture is wrong: the adjudication call must not match an expand-only set") + } + if !startsProxyToolCall([]byte(adj), []string{expand.ToolName, adjudicate.ToolName}) { + t.Error("a second proxy-injected tool was not withheld, so its tool_use reaches the client") + } + if !startsProxyToolCall([]byte(pkExpand()), []string{expand.ToolName, adjudicate.ToolName}) { + t.Error("adding a name to the set stopped expand itself being withheld") + } + if startsProxyToolCall([]byte(pkOtherT), []string{expand.ToolName, adjudicate.ToolName}) { + t.Error("a CLIENT tool was withheld; only proxy-injected tools may be") + } } // pass must never eat or duplicate input: what it forwarded plus what it withheld has to diff --git a/proxy/stats_golden_test.go b/proxy/stats_golden_test.go index 9f92a89..4062b66 100644 --- a/proxy/stats_golden_test.go +++ b/proxy/stats_golden_test.go @@ -19,6 +19,10 @@ import ( // Adding a field here alongside the new key is the intended way to change it. var statsGoldenTopLevel = []string{ "actual_baseline_tokens", + // Stray calls the agent made to the proxy-injected adjudication tool. Added to the reviewed + // contract rather than loosening the assertion, per the rule above: it is the only figure that + // can show a "do not call this yourself" description having stopped working. + "adjudicate_stray", "adjusted_saved", // agentdiet_* are the same three fail-open figures for the `agentdiet` baseline, // which owns its own per-call budget (a window of steps sits between extract_llm's