fix: three general defects in summarize, extract_llm triggers and prompt sizing

apply/parallel_wire_test.go

@@ -0,0 +1,139 @@
+package apply_test
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	bschemas "github.com/maximhq/bifrost/core/schemas"
+	"github.com/tidwall/gjson"
+
+	"github.com/rossoctl/context-guru/apply"
+	"github.com/rossoctl/context-guru/components"
+	"github.com/rossoctl/context-guru/store"
+)
+
+// THE SHAPE LIVE TRAFFIC CARRIES. In Anthropic's wire format a PARALLEL tool call is ONE assistant
+// message carrying several tool_use blocks, answered by ONE user message carrying several
+// tool_result blocks. apply normalizes that user message into several synthetic role=tool messages,
+// so N normalized messages share ONE body index.
+//
+// This asserts tool pairing in BOTH directions on the emitted wire, because the two fail
+// independently and each hid the other:
+//
+//   - FORWARD (every tool_use is answered). This was PR #80's iter011 defect, where the body
+//     message holding both tool_results was dropped and the parallel call went unanswered —
+//     28 of 75 live runs rejected. Fixed upstream; asserted here as a regression guard.
+//   - BACKWARD (every tool_result answers a call that PRECEDES it). keep_last counts messages and
+//     a tool result is a message, so the tail boundary could begin mid-exchange: the assistant's
+//     calls were summarized away while their results survived. At keep_last 2 and 3 this emitted
+//     [user, summary, user(tool_result pa_h, tool_result pb_h), user] — two results, no call.
+//
+// A forward-only check passes on that wire, which is why the direction matters. Both are provider
+// rejections of the entire request, not degraded output.
+func TestSummarizeNeverSplitsAToolExchange(t *testing.T) {
+	big := strings.Repeat("verbose parallel tool output\n", 60)
+	msgs := []map[string]any{
+		{"role": "user", "content": "start the task"},
+	}
+	for i := 0; i < 8; i++ {
+		a, b := "pa_"+string(rune('a'+i)), "pb_"+string(rune('a'+i))
+		msgs = append(msgs,
+			map[string]any{"role": "assistant", "content": []map[string]any{
+				{"type": "text", "text": "calling two"},
+				{"type": "tool_use", "id": a, "name": "Read", "input": map[string]any{}},
+				{"type": "tool_use", "id": b, "name": "Read", "input": map[string]any{}},
+			}},
+			// BOTH results in ONE user message -- Anthropic's requirement for a parallel call.
+			map[string]any{"role": "user", "content": []map[string]any{
+				{"type": "tool_result", "tool_use_id": a, "content": big},
+				{"type": "tool_result", "tool_use_id": b, "content": big},
+			}},
+		)
+	}
+	msgs = append(msgs, map[string]any{"role": "user", "content": "final question"})
+	body, _ := json.Marshal(map[string]any{"model": "claude-x", "messages": msgs})
+
+	// Preconditions, so a wire that carries no tool content at all cannot pass silently.
+	var sawResult, sawParallelCall, acted bool
+
+	for _, keep := range []int{1, 2, 3, 4, 5} {
+		cfg := pipe(t, "pipeline: [summarize]\ncomponents:\n  summarize: {keep_last: "+
+			string(rune('0'+keep))+", start_from_message: 0, min_tokens: 1}\n")
+		p, _ := cfg.Build(nil)
+		out, changed := apply.BodyWithModel(context.Background(), p,
+			store.NewMemory(store.Options{}), bschemas.Anthropic, body, "", false,
+			components.ModelSpec{Incoming: stubModel{resp: "essential facts"}})
+		if !changed {
+			continue
+		}
+		acted = true
+		arr := gjson.GetBytes(out, "messages").Array()
+
+		// Ids are collected AS THE TRANSCRIPT IS WALKED, so a result can only pair with a call
+		// that precedes it -- the same rule schema.ToolCalls documents and the provider enforces.
+		declared := map[string]bool{}
+		for i, m := range arr {
+			var uses []string
+			m.Get("content").ForEach(func(_, blk gjson.Result) bool {
+				switch blk.Get("type").String() {
+				case "tool_use":
+					id := blk.Get("id").String()
+					uses = append(uses, id)
+					declared[id] = true
+				case "tool_result":
+					sawResult = true
+					if id := blk.Get("tool_use_id").String(); !declared[id] {
+						t.Errorf("keep_last=%d: wire message %d carries tool_result %q with no "+
+							"preceding tool_use -- the provider rejects this", keep, i, id)
+						dumpWire(t, arr)
+					}
+				}
+				return true
+			})
+			if len(uses) >= 2 {
+				sawParallelCall = true
+			}
+			if len(uses) == 0 {
+				continue
+			}
+			// FORWARD: the results must be in the message immediately after the call.
+			answered := map[string]bool{}
+			if i+1 < len(arr) {
+				arr[i+1].Get("content").ForEach(func(_, blk gjson.Result) bool {
+					if blk.Get("type").String() == "tool_result" {
+						answered[blk.Get("tool_use_id").String()] = true
+					}
+					return true
+				})
+			}
+			for _, u := range uses {
+				if !answered[u] {
+					t.Errorf("keep_last=%d: wire message %d declares tool_use %q with no "+
+						"tool_result immediately after -- the provider rejects this", keep, i, u)
+					dumpWire(t, arr)
+				}
+			}
+		}
+	}
+
+	if !acted {
+		t.Fatal("summarize never acted, so no wire was checked -- the assertions are vacuous")
+	}
+	if !sawResult {
+		t.Fatal("no tool_result ever reached the wire, so the BACKWARD assertion never ran")
+	}
+	if !sawParallelCall {
+		t.Fatal("no parallel tool_use pair ever reached the wire, so the FORWARD assertion " +
+			"never exercised the shape this test exists for")
+	}
+}
+
+func dumpWire(t *testing.T, arr []gjson.Result) {
+	t.Helper()
+	for k, mm := range arr {
+		t.Errorf("    [%d] role=%s content_head=%.90s", k,
+			mm.Get("role").String(), mm.Get("content").Raw)
+	}
+}

components/offload/extract_llm.go

@@ -232,7 +232,19 @@ var staticWindows = modelinfo.DefaultStatic()
 
 // inputLimit resolves the extraction model's input-token budget. Config pin first, then the
 // model's own window as DATA (modelinfo's table), then a conservative default.
-func (e *ExtractLLM) inputLimit(c *components.Ctx) int {
+//
+// effSource is the source the extraction model was ACTUALLY resolved from, which is not always
+// the configured one: ModelSpec falls back from `incoming` to the static client whenever no
+// incoming client could be built (the proxy returns nil when no usable credential is on the
+// request). Sizing the prompt by e.modelSource instead would then hand the REQUEST model's window
+// to a call that is really going to the small static model — over-estimating, on a coding agent by
+// as much as 1M against 200k, which is the direction fitsModelContext calls the costly one: the
+// request goes out, the upstream rejects it, and the round-trip buys nothing. Pass "" when the
+// effective source is not known and the configured one is used as before.
+func (e *ExtractLLM) inputLimit(c *components.Ctx, effSource string) int {
+	if effSource == "" {
+		effSource = e.modelSource
+	}
 	if e.modelMaxInput > 0 {
 		return e.modelMaxInput
 	}
@@ -245,7 +257,7 @@ func (e *ExtractLLM) inputLimit(c *components.Ctx) int {
 	// No pinned model. `source: config` means the host's separate cheap client, whose id we
 	// never see — stay conservative. Otherwise the extraction model IS the proxied model,
 	// and the host already resolved its window onto the Ctx.
-	if e.modelSource != "config" && c.CtxWindow > 0 {
+	if effSource != "config" && c.CtxWindow > 0 {
 		return c.CtxWindow
 	}
 	return unknownModelInputLimit
@@ -701,6 +713,9 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R
 		return nil, nil
 	}
 	model := e.modelClient
+	// The source the model is ACTUALLY resolved from, which the prompt budget below must be
+	// sized against rather than the configured one — see inputLimit.
+	effSource := e.modelSource
 	if model == nil {
 		// ForModel, not For: `model.model` names the model to COMPACT with even when the
 		// source is the incoming request. Without that, compaction on a coding agent runs on
@@ -709,6 +724,9 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R
 		// it. Same endpoint, same credential, cheap model.
 		var usedSource string
 		model, usedSource = c.Model.ForModelSource(e.modelSource, e.modelName)
+		if usedSource != "" {
+			effSource = usedSource
+		}
 		// The fallback from `incoming` to the static model is a DIFFERENT credential on a
 		// DIFFERENT endpoint, so it cannot be silent: an operator whose config says
 		// `source: incoming` would otherwise have no way to learn that none of their calls
@@ -817,7 +835,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R
 	keepIDs := extract.HarvestIdentifiers(conversationContext(req, ctxRecent, e.ctxMessages), 40)
 	// Per-call context budget (constant across this request's candidates): the extraction
 	// model's input limit, and the prompt's fixed cost around the tool output itself.
-	inputLimit := e.inputLimit(c)
+	inputLimit := e.inputLimit(c, effSource)
 	promptOverhead := extractPromptOverheadTokens + schema.TextTokens(goal)
 	// The same prompt, for the COST model rather than the window check: callCost adds the
 	// static preamble itself, so it must be given only the variable part.
@@ -1046,7 +1064,29 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R
 		// of the few parts of this component that unambiguously pays. A metric that argues for
 		// optimizing something already working is worse than no metric.
 		metrics.RecordExtractionCacheLookup(false)
-		if huge := e.trigger.IsHuge(sz, c.CtxWindow); !c.CacheAware && !fires && !huge {
+		// The operator's REQUEST-level trigger, now honored on WARM caching turns.
+		//
+		// This condition used to carry `!c.CacheAware`, and that spelling was too broad. What
+		// legitimately bypasses a request-size threshold is a COLD SWEEP: on a cold turn the
+		// whole transcript re-bills at the cache-write rate whatever the request's size, so the
+		// request-level threshold answers the wrong question and the sweep brings its own floor
+		// (e.cold.MinTokens, set above). That is why `sweeping` already overrides the cadence
+		// gate and the pressure gate — this check simply had not been given the same treatment.
+		//
+		// `c.CacheAware` is true on warm caching turns as well as cold ones, so the old spelling
+		// also discarded the threshold on every warm turn, where it means exactly what the
+		// operator wrote. And it could ONLY discard operator configuration: Trigger's zero value
+		// fires always (see components/trigger.go — "a zero field is no constraint"), so `!fires`
+		// is reachable only when min_request_tokens / min_request_frac / min_messages was set and
+		// not met. There is no derived value in `fires` for a cache carve-out to protect; the
+		// derived pressure trigger is separate and gates the model earlier via shouldFire.
+		//
+		// Found by the housellm cold-sweep preset test, which fails if `sweeping` is dropped
+		// here — the sweep is the part of the old carve-out that was carrying real weight.
+		//
+		// IsHuge still overrides, unchanged: a single output that large is worth a call whatever
+		// the request-level threshold says.
+		if huge := e.trigger.IsHuge(sz, c.CtxWindow); !fires && !huge && !sweeping {
 			rep.Gate("request_trigger_not_fired")
 			continue
 		}

components/offload/extract_llm_ctxguard_test.go

@@ -117,20 +117,29 @@ func TestExtractLLMInputLimit(t *testing.T) {
 		name      string
 		yaml      string
 		ctxWindow int
+		effSource string
 		want      int
 	}{
-		{"config pin wins", "model_max_input_tokens: 4096\nmodel:\n  model: claude-haiku-4-5\n", 200_000, 4096},
-		{"pinned model resolved from the table", "model:\n  model: claude-haiku-4-5\n", 0, 200_000},
-		{"unnameable pinned model falls back", "model:\n  model: qwen3-coder-30b-local\n", 999_999, unknownModelInputLimit},
-		{"incoming model uses the host-resolved window", "", 128_000, 128_000},
-		{"incoming model, window unknown", "", 0, unknownModelInputLimit},
-		{"source config hides the model id", "model:\n  source: config\n", 128_000, unknownModelInputLimit},
+		{"config pin wins", "model_max_input_tokens: 4096\nmodel:\n  model: claude-haiku-4-5\n", 200_000, "", 4096},
+		{"pinned model resolved from the table", "model:\n  model: claude-haiku-4-5\n", 0, "", 200_000},
+		{"unnameable pinned model falls back", "model:\n  model: qwen3-coder-30b-local\n", 999_999, "", unknownModelInputLimit},
+		{"incoming model uses the host-resolved window", "", 128_000, "", 128_000},
+		{"incoming model, window unknown", "", 0, "", unknownModelInputLimit},
+		{"source config hides the model id", "model:\n  source: config\n", 128_000, "", unknownModelInputLimit},
+		// `source: incoming` that FELL BACK to the static client. The call is going to the
+		// small config model, so the proxied model's window is the wrong budget — sizing the
+		// prompt by it over-estimates and the upstream rejects the request.
+		{"incoming fell back to config: window is not the request model's",
+			"model:\n  source: incoming\n", 1_000_000, "config", unknownModelInputLimit},
+		// The same config that did NOT fall back keeps using the host-resolved window.
+		{"incoming that stayed incoming keeps the window",
+			"model:\n  source: incoming\n", 1_000_000, "incoming", 1_000_000},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			e := newCtxGuardComponent(t, &silentModel{}, tc.yaml)
 			c := &components.Ctx{Ctx: context.Background(), CtxWindow: tc.ctxWindow}
-			if got := e.inputLimit(c); got != tc.want {
+			if got := e.inputLimit(c, tc.effSource); got != tc.want {
 				t.Fatalf("inputLimit = %d, want %d", got, tc.want)
 			}
 		})

components/offload/extract_trigger_cacheaware_test.go

@@ -0,0 +1,86 @@
+package offload
+
+import (
+	"context"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	bschemas "github.com/maximhq/bifrost/core/schemas"
+	"github.com/rossoctl/context-guru/components"
+	"github.com/rossoctl/context-guru/store"
+)
+
+// An operator's REQUEST-level trigger must be honored on a prompt-caching backend.
+//
+// The guard on this check used to read `!c.CacheAware && !fires && !huge`, so with CACHE_MODE=on
+// (or any caching provider) a configured trigger.min_request_tokens had no effect at all. That is
+// not a cache-awareness carve-out protecting a heuristic: Trigger's zero value fires always (see
+// components/trigger.go), so `!fires` is reachable ONLY when the operator configured a request
+// threshold that was not met. The condition could therefore only ever void explicit
+// configuration, silently.
+//
+// Both arms run the SAME config against the SAME request and differ ONLY in CacheAware, which is
+// what makes this an assertion about the carve-out rather than about triggers in general.
+//
+// The COLD SWEEP is the part of the old carve-out that carried real weight and is deliberately
+// still exempt: on a cold turn the whole transcript re-bills whatever the request's size, so a
+// request-size threshold answers the wrong question there and the sweep brings its own floor.
+// Both arms here run warm (ColdCache unset), and the exemption is pinned separately by
+// TestHousellmColdSweepActuallyFires in components/all — which is what caught the first version
+// of this fix removing it.
+//
+// economic_gate: false is required setup, not a convenience: it also sets
+// allow_on_caching_backend, without which the component disables itself entirely on the
+// CacheAware arm and the arm would pass for a completely unrelated reason.
+func TestExplicitRequestTriggerIsHonoredOnCachingBackends(t *testing.T) {
+	// One output far above any floor, in a request far below the configured request threshold.
+	big := strings.Repeat("2024-01-01 GET /users/42 200 12ms src/api/users.py\n", 400)
+	newReq := func() *bschemas.BifrostChatRequest {
+		return &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{
+			userMsg("Find the auth timeout in src/api/users.py and fix it."),
+			assistantMsg("Reading the log."),
+			toolResultMsg(big),
+			userMsg("keep going"),
+		}}
+	}
+
+	for _, cacheAware := range []bool{false, true} {
+		name := "cache_aware=false"
+		if cacheAware {
+			name = "cache_aware=true"
+		}
+		t.Run(name, func(t *testing.T) {
+			// A threshold no request in this test can meet, so `fires` is false in BOTH arms.
+			comp, err := newExtractLLM([]byte(
+				"economic_gate: false\ntrigger:\n  min_request_tokens: 10000000\n"))
+			if err != nil {
+				t.Fatalf("config: %v", err)
+			}
+			e := comp.(*ExtractLLM)
+			model := &silentModel{}
+			req, rep := newReq(), components.Report{}
+			ctx := &components.Ctx{
+				Session: "s", Ctx: context.Background(),
+				Store: store.NewMemory(store.Options{}), CtxWindow: 1_000_000,
+				CacheAware: cacheAware, MaxCachedIdx: -1,
+				Model: components.ModelSpec{Static: model, Incoming: model},
+			}
+			if _, err := e.Offload(req, &rep, ctx); err != nil {
+				t.Fatalf("offload: %v", err)
+			}
+			// Precondition: the candidate must have REACHED this gate. Filtered earlier by the
+			// floor or the cached-prefix tail gate, the assertion below would be vacuous.
+			if rep.Gates["below_output_floor"] > 0 || rep.Gates["cached_prefix"] > 0 {
+				t.Fatalf("candidate never reached the request trigger; gates=%v", rep.Gates)
+			}
+			if rep.Gates["request_trigger_not_fired"] == 0 {
+				t.Errorf("configured min_request_tokens was ignored: expected the "+
+					"request_trigger_not_fired gate, got gates=%v", rep.Gates)
+			}
+			if n := atomic.LoadInt64(&model.calls); n != 0 {
+				t.Errorf("model called %d time(s) despite an unmet configured request trigger", n)
+			}
+		})
+	}
+}