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

apply/apply.go

@@ -1359,7 +1359,41 @@ func rebuildCountChanged(body []byte, orig []gjson.Result, slots []slot, out []b
 	if len(covered) != len(orig) {
 		return body, false
 	}
+	// A RETAINED TOOL MESSAGE WHOSE TEXT A COMPONENT REWROTE, which byte-matching cannot see.
+	//
+	// Anthropic has no tool role: a synthetic role=tool message is this package's internal
+	// representation of a tool_result content block. The loop below matches survivors by BYTES, so
+	// a tool message that some component rewrote (format compacting indented JSON, extract_llm
+	// reducing a retained output) no longer matches its pre-image, falls through to the
+	// "new message" branch, and is marshaled from the bifrost struct — putting `"role":"tool"` on
+	// the wire, which the provider rejects outright:
+	//
+	//	400 messages: Unexpected role "tool". Allowed roles are "user" or "assistant."
+	//
+	// It needs TWO components in one turn, which is why it went unseen: one to change the count so
+	// this rebuild runs at all, and one to rewrite a tool message the first one kept. Observed as a
+	// real 400 on a live session.
+	//
+	// Fixed the same way the equal-count path already handles tool text: write the new text into
+	// the body's tool_result block, so the rebuild only ever decides WHICH messages to keep and
+	// never how to serialize one. Then match those messages by tool_call_id rather than by bytes,
+	// since their text may now legitimately differ from the pre-image. Doing only the second half
+	// would emit the original bytes and silently discard the compaction.
+	if nb, no, ok := writeBackToolText(body, orig, slots, out); ok {
+		body, orig = nb, no
+	}
 	used := make([]bool, len(slots))
+	// toolSlotByID indexes the tool-text slots so a rewritten tool message can still be
+	// recognised as a survivor. Built once; the matching loop is already O(out × slots).
+	toolSlotByID := map[string]int{}
+	for k := range slots {
+		if slots[k].kind != anthropicToolText {
+			continue
+		}
+		if id := toolCallIDAt(orig, slots[k]); id != "" {
+			toolSlotByID[id] = k
+		}
+	}
 	var parts [][]byte
 	// emitted guards against emitting one body message TWICE: several normalized
 	// messages can share a body index (an Anthropic user message with several
@@ -1372,10 +1406,21 @@ func rebuildCountChanged(body []byte, orig []gjson.Result, slots []slot, out []b
 			return body, false
 		}
 		matched := -1
-		for k := range slots {
-			if !used[k] && bytes.Equal(mb, slots[k].pre) {
+		// A synthetic tool message is identified by its tool_call_id, not its bytes: the text
+		// may have been rewritten (and written back into the body just above), and the id is
+		// what pairing actually depends on.
+		if out[i].Role == bschemas.ChatMessageRoleTool && out[i].ChatToolMessage != nil &&
+			out[i].ChatToolMessage.ToolCallID != nil {
+			if k, ok := toolSlotByID[*out[i].ChatToolMessage.ToolCallID]; ok && !used[k] {
 				matched = k
-				break
+			}
+		}
+		if matched < 0 {
+			for k := range slots {
+				if !used[k] && bytes.Equal(mb, slots[k].pre) {
+					matched = k
+					break
+				}
 			}
 		}
 		if matched < 0 {
@@ -1409,6 +1454,67 @@ func rebuildCountChanged(body []byte, orig []gjson.Result, slots []slot, out []b
 	return res, true
 }
 
+// toolCallIDAt reads the tool_use_id of the tool_result block a tool-text slot points at.
+// The slot path is "messages.<i>.content.<b>.content", so the block is its parent.
+func toolCallIDAt(orig []gjson.Result, s slot) string {
+	bi, rel, ok := splitSlotPath(s.path)
+	if !ok || bi < 0 || bi >= len(orig) {
+		return ""
+	}
+	blk := strings.TrimSuffix(rel, ".content")
+	if blk == rel { // not a tool-text path
+		return ""
+	}
+	return orig[bi].Get(blk + ".tool_use_id").String()
+}
+
+// writeBackToolText splices rewritten tool-output text into the body's tool_result blocks before
+// the count-change rebuild reads them, so a retained-but-rewritten tool message can be emitted
+// from body bytes (role intact) instead of marshaled from the bifrost struct (role leaked).
+//
+// This is deliberately the SAME shape of edit the equal-count path makes — only the block's
+// `content` string changes, so the rest of the message stays byte-identical — and it is why the
+// rebuild can keep the rule "decide which messages to keep, never how to serialize one".
+//
+// Returns ok=false when nothing needed writing, so the caller keeps its original slices and no
+// body copy is made on the common path.
+func writeBackToolText(body []byte, orig []gjson.Result, slots []slot,
+	out []bschemas.ChatMessage) ([]byte, []gjson.Result, bool) {
+	byID := map[string]int{}
+	for k := range slots {
+		if slots[k].kind != anthropicToolText {
+			continue
+		}
+		if id := toolCallIDAt(orig, slots[k]); id != "" {
+			byID[id] = k
+		}
+	}
+	var wrote bool
+	for i := range out {
+		if out[i].Role != bschemas.ChatMessageRoleTool || out[i].ChatToolMessage == nil ||
+			out[i].ChatToolMessage.ToolCallID == nil {
+			continue
+		}
+		k, ok := byID[*out[i].ChatToolMessage.ToolCallID]
+		if !ok {
+			continue
+		}
+		txt := schema.MessageText(out[i])
+		if txt == slots[k].preText {
+			continue // unchanged; the original bytes already carry it
+		}
+		nb, err := sjson.SetBytes(body, slots[k].path, txt)
+		if err != nil {
+			return nil, nil, false // fail open: leave the body alone
+		}
+		body, wrote = nb, true
+	}
+	if !wrote {
+		return nil, nil, false
+	}
+	return body, gjson.GetBytes(body, "messages").Array(), true
+}
+
 // splitSlotPath splits a slot path into the body message index and the remainder of
 // the path RELATIVE to that message: "messages.3.content.2.content" -> 3,
 // "content.2.content". A whole-message slot has an empty remainder.

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)
+	}
+}