diff --git a/DONE-NOTE.md b/DONE-NOTE.md index 412d612..0bd7671 100644 --- a/DONE-NOTE.md +++ b/DONE-NOTE.md @@ -12,6 +12,7 @@ the other. Each note keeps its own `# DONE-NOTE - ` heading verbatim. | `model_performance-x7p` | `protected_tool_results=0` protected ALL tool results (negative-slice bug) | | `model_performance-x1r` | tool-result budget (token-denominated, head+tail, per-tool) + spill-to-disk | | `model_performance-2o9` | `clear_at_least` — a worth-the-rebuild predicate in front of compaction (+ summary shrink guard) | +| `model_performance-7k2` | `summary_call_mode` — cache-safe forking of the summarization call | --- @@ -912,3 +913,219 @@ property of the module, not new exposure, but now with one more field. No PII, no team-internal data, no individual attribution in any output. No merges to main; no files touched outside this module. + +--- + +# DONE-NOTE — W3-2 / `model_performance-7k2` + +`context-simple: cache-safe forking of the summarization call (summary_call_mode: "fork")` + +**Status: draft PR, mechanism shipped behind an off-by-default flag, nothing +measured.** Default is byte-identical to today. Lane spend: **$0.00** (the +item's authority was $0; no API calls, no DTU, no infrastructure created, +nothing to tear down). + +## 1. Step 0 — the free measurement that could have cancelled this item + +The item said: read the summarizer call; if it already prepends the parent's +assembled request, §2h conflict #8 resolves in favour of [P6]'s 2.4% table +figure and **there is nothing to build**. + +It does not. `_run_summary_compaction_task` built, verbatim: + +```python +request = ChatRequest( + messages=[ + Message(role="system", content=prompt), # the ~955-char prompt + Message(role="user", content=formatted), # the span, re-rendered as plain text + ], + model=self.summarization_model, +) +``` + +No parent history. No tools. Its own `role: "system"` prompt. This is the +standalone shape, so **the branch that cancels the item does not apply** and +the fix is real work. §2d's indirect evidence was right: G3 had to split out +the summarizer's own `instructions` population to get a clean hash precisely +*because* the summarizer sends its own separate system prompt. + +**What this does NOT resolve:** the 8.3–10.9% (prose) vs 2.4% (same source's +own table) conflict about the summarizer's cost *share*. Being standalone +makes the cost avoidable; it does not say how much cost there is. That number +must come from a run's own per-call table, and the follow-up item requires it. + +## 2. What shipped + +`summary_call_mode: "standalone" | "fork"`, default `"standalone"`, consulted +only when `compaction_strategy == "summary"`. + +- **`"standalone"`** — the pre-existing two-message call, unchanged. + `"inline"` is accepted as an alias (the lane brief and the work item named + the same default differently; both mean "today's behaviour"). Unknown values + warn and fall back, matching every other enum in this module. +- **`"fork"`** — the same ask re-issued as a **pure append**: + `[...the exact messages of the last request...] + [one user message: prompt + scope]`. + Pure append is the one mutation measured as a HIT under grow-only + (P4: identical-repeat 9,789 HIT, pure-append 9,789 HIT, truncation 0, + middle-drop 0). + +Three consequences, each load-bearing rather than incidental: + +1. **The span is not re-sent.** It is already inside the prefix. Re-sending it + would cost what standalone costs today *plus* the prefix — a regression + wearing an optimization's clothes. This is the single decision that + determines whether the feature is worth anything. +2. **The prompt moves into the appended `role: "user"` message.** A fork must + not add a `role: "system"` message: providers hoist every system-role + message into one top-level system block, so a per-summarization one would + rewrite the cached system prefix — the exact failure already measured for + the summary tier (cache_read 46,307 → 21,523) and the compaction notice. +3. **Scope must be stated.** Standalone scopes by construction (it can only + see the span). A fork can see everything, so the appended message names the + span: its message count plus a bounded (300-char) verbatim excerpt of its + final message as the "summarize up to here" marker. **This is a real + behavioural difference between the two modes**, not formatting — flagged as + a residual risk in the follow-up item, not hand-waved. + +New optional public seam: **`note_request_sent(messages=None, *, tools=None, +model=None)`** — see §3, which is the finding this lane most wants read. + +Observability: `last_summary_call_stats` property (`mode_requested`, +`mode_used`, `reason`, `prefix_messages`, `fork_fallbacks`) and a `call_mode` +field on `context:pre_summarize` / `context:post_summarize`, so an eval arm +can count *real* forks without patching the module. + +Size: **170 executable lines** across the new helpers (item estimated 50–70). +The overrun is entirely the refusal ladder (§3) and the staleness guard (§4); +the happy path is ~30 lines. + +## 3. The finding: this cannot be done from inside the context module alone + +**Tool specs are part of the cached prefix and this module is never handed +them.** Anthropic serializes `tools` *ahead of* the system block; OpenAI's +implicit cache matches forward from a cached entry. A summarizer request with +`tools=None` diverges from the parent at byte zero, so it hits nothing — and +because it now carries the whole conversation, it costs *more than the +standalone call it replaced*. That is not a missed win, it is a regression. + +I looked for a way to obtain them without changing any caller and found none +that is contract-backed: + +- `amplifier_core.interfaces` defines no tool-registry protocol; the + coordinator exposes `mount_points`, not tool specs in send order. +- The `llm:request` event carries `model`/`message_count`/thinking flags, and + the actual tool payload only under provider `raw: true` — where it is wire + dicts passed through `redact_secrets()`, i.e. not reconstructable into + `ToolSpec` byte-faithfully. +- The default orchestrator lives in the compiled Rust engine, so there is no + seam to read it from. + +So fork mode requires the caller to say what it sent. **Decision taken without +waiting** (per SCOPE-OUTS): add one optional public method rather than a +config knob that lets an operator *assert* alignment they cannot verify. +Passing `tools` **at all** — even `None`/`[]` for a genuinely tool-free +session — is what arms the fork; "not supplied" and "supplied as empty" are +deliberately distinguishable, because guessing between them is exactly how a +fork silently misaligns. + +**Every misalignment refuses and falls back to standalone, loudly** — WARNING +naming the precondition (once per distinct reason, so a session that can never +fork costs a handful of log lines, not one per summarization), a session +counter, and the mode actually used reported on both hooks and the stats +property. The refusals: seam never called · `summarization_model` set (a +summarizer routed elsewhere reads none of the main line's cache — deepseek +states this explicitly) · no request recorded yet · prefix ends on an assistant +turn with unanswered `tool_calls` (appending there would interleave between +`tool_use` and `tool_result`) · span absent from the recorded prefix · the +forked request fails to build. + +Falling back to *standalone* rather than skipping is deliberate: standalone is +correct and costs what it always did. Losing a summary over a cache +optimization would be strictly worse than not optimizing. + +**Consequence for the eval, stated plainly:** the S5-CRAC harness must wire +`note_request_sent()` or the treatment arm silently degrades into the control +arm and every gate is vacuous. That is the first thing the follow-up item +(`model_performance-6da`) tells the runner to verify — on a 1-run smoke, +before any spend. + +## 4. Provider asymmetry, and the staleness trap + +**Anthropic vs OpenAI need different fork sources, and the seam covers both.** +Anthropic places its cache breakpoint on the last *stable* message, walking +back past ephemeral/injected content — which is exactly where this module's +own returned view ends. So `tools` alone is enough there. OpenAI's implicit +cache measured **MISS on strict truncation** (P4), i.e. it needs a strict +superset of a cached request; an orchestrator-injected tail this module never +sees would break that. So on OpenAI the caller must also pass `messages`. +`_capture_fork_prefix` prefers the caller's record and falls back to the +module's own view. + +**The trap I built a guard for.** A caller that wires `note_request_sent()` +once (a startup helper, first turn only) would have turn 40's fork append to +turn 1's request — a guaranteed miss *and* a wasted cache write, wearing a +correct-looking API call. A `_view_serial`/`_sent_serial` pair means a +`messages` record is only used while it still describes the most recent +request served; a stale one is ignored (module view used instead), never +trusted. Tested. + +## 5. Evidence — 282 tests green (250 before, 32 new) + +`tests/test_summary_call_mode_fork.py`, written against the ways this goes +silently wrong rather than the way it is supposed to work: + +| Group | What it pins | +|---|---| +| **A — the default must not move** | The standalone request is asserted against **independently rebuilt** expected content (2 messages, system prompt then formatted span, `model` from `summarization_model`), not against itself. Plus: the fork bookkeeping is never even *written* unless armed (`_last_request_view is None` after 5 requests in both default configs) — an always-on capture would be a silent per-request list allocation on the hot path. Plus: `note_request_sent()` is fully inert when fork is not configured. | +| **B — pure append** | `sha256(fork[:-1]) == sha256(parent_view)` (G-FORK-PREFIX in unit form); exactly one appended message; **no new system message**; tools and model pinned from the seam; the span is **not** re-sent (`formatted not in appended`, and the appended message is shorter than the formatted span); the caller's `messages` record wins verbatim when supplied, including an injected tail. | +| **C — the main line is not the summarizer's scratchpad** | No `_seq` consumed; history byte-identical before/after; `_removed_seqs` unmoved; **`_last_sent_estimate` unmoved** (building the fork through `_finalize_view` would silently rewrite the hybrid meter's conservatism comparand with the summarizer's own request — this is why `_build_fork_request` calls `_strip_internal_metadata` directly); and a forked session serves views **bit-for-bit identical** to an unforked control, same `_removed_seqs`. | +| **D — refusals** | All six refusal paths fall back to the 2-message standalone request with the reason recorded; warning fires; warn-once-per-reason while the counter still counts every one; a failed fork build still produces the summary; hooks report the mode actually used; the stale-record guard prefers the fresh view. | +| **E — tool pairs** | `_select_summary_absorb_seqs` returns **identical** spans in both modes on a tool-pair-heavy history; end-to-end, no tool result is ever served without its call. The call mode changes how the summarizer is *called*, never what is *selected*. | +| **F — reset/lifecycle** | `clear()` and `set_messages()` drop all fork alignment state; the fork prefix is snapshotted at **trigger** time, not at task-scheduling time (proved by mutating `_last_request_view` while the task is in flight); the single-summarization-in-flight guard still holds under `asyncio.gather` of three concurrent requests. | + +`ruff check` clean. `ruff format` was **not** run: the repo is not +format-clean today (7 pre-existing files would be reformatted), so running it +would bury this change in unrelated noise. + +## 6. What is NOT claimed + +- **No cache win. No cost win. Nothing was measured.** G-FORK-PREFIX / + G-FORK-CACHED / G-FORK-COST / G-FORK-NOBOUNDARY are the follow-up item + (`model_performance-6da`, filed `discovered-from` this one, with the arm + design, the blocking prerequisite, and the residual risks). $0 spent here by + mandate. +- **Honest ceiling, unchanged from the item's own statement:** this reduces + the separate summarizer charge only. It does **not** touch the boundary + rebuild, which §2d measured as the dominant cost (+84% boundaries → +83% run + cost). If the summarizer's real share is the table's 2.4% rather than the + prose's 8.3–10.9%, a *perfect* fork is worth ~2% of run cost and the honest + recommendation may be "mechanism proven, not worth enabling". The follow-up + item is required to report that share from the run's own table and to say so + if it lands there. +- **Retention parity is not established.** Fork mode scopes by instruction + where standalone scopes by construction. The follow-up item reports S5 score + and `b_constraints`/`c_post_compaction` per arm so a cost win bought with a + retention loss cannot be reported as a win. +- **`reasoning_effort` / thinking configuration is not reproduced** by the + fork — it is not observable from inside a context module. If a provider keys + its cache on it, the fork misses despite byte-aligned messages. + G-FORK-CACHED is the detector; this is disclosed, not designed around. +- **The compaction-buffer reserve (condition (a) in the item) is not + implemented.** The appended instruction is ~1.2k chars, not the ~8,000-token + reserve the item contemplated, and the fork never grows the *main* line's + request. If the follow-up eval shows the fork's own request crowding the + window, that reserve is the fix, and G-FORK-NOBOUNDARY is the gate that + would catch it. + +## 7. Deliverable ledger + +| Deliverable | Status | +|---|---| +| DRAFT PR on origin, branch `lane/7k2-summary-call-fork`, default byte-identical, tests green | **DONE** | +| Prefix-stability test proving the fork does not touch the main line | **DONE** — Group C (no `_seq`, history byte-identical, `_last_sent_estimate` unmoved, forked-vs-control views identical) | +| Follow-up eval item filed via `work_file` with its arm design | **DONE** — `model_performance-6da` | +| DONE-NOTE.md in the PR body | **DONE** — this section | + +No PII, no team-internal data, no individual attribution. No merges to main. +No files touched outside this module. No infrastructure created; nothing to +tear down. diff --git a/README.md b/README.md index 8af2339..632c947 100644 --- a/README.md +++ b/README.md @@ -364,6 +364,7 @@ for the full write-up. Neither defect is inherited here: module = "context-simple" config = { compaction_strategy = "summary", # default: "progressive" + summary_call_mode = "standalone", # default; "fork" appends onto the live prefix summary_trigger = 0.60, # usage fraction that starts the async summarizer summarization_model = "...", # optional; None uses the provider default summarization_prompt_path = "...", # optional file override for the 5-section prompt @@ -439,6 +440,127 @@ absorptions cannot cycle; **hysteresis** on `summary_trigger` (arm at should be validated against the same cost metrics before this flag is enabled anywhere by default. +## Cache-safe fork of the summarization call (`summary_call_mode`) + +**Default is `"standalone"` — byte-identical to the summarizer call this +module has always made.** `"inline"` is accepted as an alias for it. Only +consulted when `compaction_strategy == "summary"`. + +### The defect + +The summarizer's request is, today, a **standalone** two-message call: its +own ~955-char `role: "system"` prompt, plus a freshly formatted plain-text +rendering of the span being absorbed. It shares **not one byte of prefix** +with the main conversation. So every token of that span is billed as +*fresh input* — while the provider is already holding that exact span warm +in the main line's cache. + +This resolves the internal conflict recorded in `00-what-we-know.md` §2d +(prose said the summarizer was "8.3–10.9% of run cost", the same source's +own table said 2.4%): whatever the share, the call is genuinely standalone, +so the cost is genuinely avoidable. It does **not** touch the boundary +rebuild, which §2d shows is the dominant cost — see "honest ceiling" below. + +### The shape + +`summary_call_mode: "fork"` re-issues the same ask as a **pure append**: + +``` +[ ...the exact messages of the last request... ] + [ one user message: prompt + scope ] +``` + +Pure append is the *one* mutation measured as a cache HIT under the +grow-only rule (probe P4: identical-repeat 9,789 HIT, **pure-append 9,789 +HIT**, strict truncation 0 MISS, middle-drop 0 MISS). + +Three things follow, and each is load-bearing: + +1. **The span is not re-sent.** It is already inside the prefix. Re-sending + it would cost exactly what standalone costs today *plus* the prefix — a + regression wearing an optimization's clothes. +2. **The prompt moves into the appended `role: "user"` message.** A + per-summarization `role: "system"` message is hoisted into the + provider's single top-level system block and rewrites the cached system + prefix — the same failure already measured for the summary tier and the + compaction notice. +3. **Scope has to be stated.** Standalone scopes by construction (it can + only see the span). A fork can see everything, so the appended message + names the span explicitly: message count, plus a bounded verbatim + excerpt of the span's final message as the "summarize up to here" + marker. This is a real behavioral difference between the two modes, not + a formatting detail. + +### The precondition you must wire: `note_request_sent()` + +Tool specs are serialized **ahead of** the system block, and this module is +handed messages, never tools. It also never sees a tail an orchestrator +injects *after* `get_messages_for_request()` returns. A fork missing either +is not an append onto the cached prefix at all. + +So fork mode requires the caller to say what it actually sent: + +```python +messages = await context.get_messages_for_request(provider=provider) +request = ChatRequest(messages=messages, tools=tools, model=model) +context.note_request_sent(messages, tools=tools, model=model) # <- every request +response = await provider.complete(request) +``` + +- Passing `tools` **at all** — even `None`/`[]` for a genuinely tool-free + session — is what arms the fork. "Not supplied" and "supplied as empty" + are deliberately distinguishable; guessing between them is how a fork + silently misaligns. +- Passing `messages` gives byte-parity with the wire, which an **implicit, + match-forward-only cache (OpenAI)** requires — its measured behavior + misses on anything that is not a strict superset of a cached request + (that is the same finding as "strict truncation → 0"). Omit it and the + fork appends to this module's own last returned view, which is what an + **explicit-breakpoint cache (Anthropic)** needs, since the breakpoint is + placed on the last *stable* message — exactly where this module's view + ends, before any ephemeral injection. +- Call it **every request**. A record that no longer describes the latest + request is ignored, not trusted: a one-time wiring would otherwise have + turn 40's fork append to turn 1's request. + +### A silently unforked fork is the failure mode that matters + +A fork that does not reproduce the parent's prefix wins nothing **and pays +for the whole conversation** — strictly worse than the standalone call it +replaces. Every misalignment therefore refuses, falls back to standalone, +and says so: a `WARNING` naming the precondition (once per distinct +reason), a session counter, and the mode actually used reported on +`context:pre_summarize` / `context:post_summarize` and via +`last_summary_call_stats`. + +The refusals: `note_request_sent()` never called · `summarization_model` +set (a summarizer routed elsewhere reads none of the main line's cache) · +no request recorded yet (first request of a session) · the prefix ends on +an assistant turn with unanswered `tool_calls` (appending there would +interleave between `tool_use` and `tool_result`) · the span is absent from +the recorded prefix · the forked request fails to build. + +### What is proven here, and what is not + +**Proven, structurally, by `tests/test_summary_call_mode_fork.py`:** the +default request is byte-identical and the fork bookkeeping is never even +written unless armed; the forked request is the parent prefix plus exactly +one `role: "user"` message (`sha256(fork[:-1]) == sha256(parent)`); no +`_seq` is consumed, history is untouched, `_last_sent_estimate` (the hybrid +meter's comparand) does not move, and a forked session serves views +bit-for-bit identical to an unforked control; tool-pair snapping is +unchanged; every refusal falls back loudly. + +**Not proven — no cache or cost win is claimed.** G-FORK-PREFIX / +G-FORK-CACHED / G-FORK-COST / G-FORK-NOBOUNDARY are a separately funded +evaluation; none of it was run here. + +**Honest ceiling, stated in advance:** this reduces the separate summarizer +charge only. It does **not** reduce the boundary rebuild, which §2d +measured as the dominant cost (+84% boundaries drives the +83% run cost). +If the summarizer really is 2.4% of run cost rather than 8.3–10.9%, then +even a perfect fork is worth ~2% — the arm design must therefore report the +summarizer's cost share from the run's own per-call table, not from prose. + ## Tool-result budget, shape, and spill **Every flag in this section defaults to a no-op.** With no configuration the diff --git a/amplifier_module_context_simple/__init__.py b/amplifier_module_context_simple/__init__.py index fdc71ce..9fed112 100644 --- a/amplifier_module_context_simple/__init__.py +++ b/amplifier_module_context_simple/__init__.py @@ -155,6 +155,66 @@ / trigger hysteresis). OPT-IN, EXPERIMENTAL -- do not enable by default. +Cache-safe forking of the summarization call (opt-in, default off -- see +config `summary_call_mode`; only consulted when compaction_strategy == +"summary"): + • THE PROBLEM. The summarizer's request is, today, a STANDALONE two + message call: its own ~955-char system prompt plus a freshly + formatted plain-text rendering of the span being absorbed. It shares + not one byte of prefix with the main line, so every token of the span + is billed as FRESH input -- even though the provider is already + holding that exact span in a warm cache for the main conversation. + • THE SHAPE THAT FIXES IT. `summary_call_mode: "fork"` re-issues the + same ask as a PURE APPEND onto the prefix the main line already sent: + [ ...the exact messages of the last request... ] + [ one user message + carrying the summarization instruction ]. Pure append is the one + mutation measured as a cache HIT under the grow-only rule (probe P4: + identical-repeat 9,789 HIT, pure-append 9,789 HIT, strict truncation + 0 MISS, middle-drop 0 MISS). The span itself is NOT re-sent -- it is + already in the appended-to prefix; re-sending it would cost exactly + what standalone costs today PLUS the prefix, i.e. a regression. + • THE PROMPT MOVES INTO THE USER MESSAGE. The standalone call puts the + summarization prompt in a `role: "system"` message. A fork MUST NOT: + providers hoist every system-role message into a single top-level + system block, so a per-summarization system message would rewrite the + cached system prefix -- the exact failure mode already documented for + the summary tier and the compaction notice. In fork mode the prompt + rides the appended `role: "user"` message instead. + • THE MAIN LINE IS NEVER TOUCHED. The forked request is built from a + read-only snapshot. No `_seq` is consumed, nothing is appended to + `self.messages`, no sticky/removal state moves, and `_last_sent_estimate` + (the hybrid meter's conservatism comparand) is NOT rewritten -- which + is why the fork calls `_strip_internal_metadata` directly rather than + `_finalize_view`. The summary that eventually lands is produced by the + unchanged `_swap_in_pending_summary` path. + • A SILENTLY UNFORKED FORK IS THE FAILURE MODE THAT MATTERS. If the + forked request does not reproduce the parent's prefix byte-for-byte + it wins nothing AND pays for the whole conversation -- strictly worse + than standalone. So the fork is attempted ONLY when every alignment + precondition holds, and any miss falls back to the standalone call + LOUDLY (a warning naming the precondition, a counter, and the mode + actually used reported on `context:post_summarize` and in + `last_summary_call_stats`). It never silently half-forks. + • THE PRECONDITION YOU MUST WIRE. Tool specs are part of the cached + prefix (they are serialized ahead of the system block), and this + module is handed messages, never tools. So a caller that wants fork + mode MUST hand over the request it actually sent, via the optional + public `note_request_sent(messages=..., tools=..., model=...)` seam. + Without it the fork refuses (warning, once) and standalone is used. + Supplying `messages` too gives exact parity with what went on the + wire -- including any hook-injected tail the orchestrator appended + after this module returned its view, which is what an implicit, + match-forward-only cache (OpenAI) requires. With only `tools`, the + fork appends to this module's own last returned view, which is what + an explicit-breakpoint cache (Anthropic) needs, since the provider + places its breakpoint at the last STABLE message -- i.e. exactly + where this module's view ends. + • UNMEASURED. The gates (G-FORK-PREFIX / G-FORK-CACHED / G-FORK-COST / + G-FORK-NOBOUNDARY) are a separately funded eval; nothing in this + module claims a measured cache or cost win yet. What IS proven here + is structural: default byte-identity, main-line non-mutation, and + pure-append shape. + Tool-result budget and spill (opt-in, default off -- see config `tool_result_budget_tokens` / `tool_result_shape` / `tool_result_budget_by_tool` / `tool_result_exempt_tools` / @@ -288,6 +348,29 @@ COMPACTION_STRATEGY_SUMMARY, ) +# summary_call_mode config values -- HOW the summarizer's own LLM call is +# shaped when compaction_strategy == "summary". "standalone" (default) is +# byte-identical to the pre-existing behavior: its own tiny two-message +# request (system prompt + formatted span), sharing nothing with the main +# line. "fork" issues the SAME summarization ask as a PURE APPEND onto the +# prefix the main line already sent, so the provider's prompt cache can be +# read instead of paying fresh input tokens for the span. See module +# docstring "Cache-safe forking of the summarization call". +# +# "inline" is accepted as an alias for "standalone": the lane brief that +# commissioned this work named the default mode "inline" while the work +# item named it "standalone". Both mean "today's behavior, unchanged". +SUMMARY_CALL_MODE_STANDALONE = "standalone" +SUMMARY_CALL_MODE_FORK = "fork" +_VALID_SUMMARY_CALL_MODES = (SUMMARY_CALL_MODE_STANDALONE, SUMMARY_CALL_MODE_FORK) +_SUMMARY_CALL_MODE_ALIASES = {"inline": SUMMARY_CALL_MODE_STANDALONE} + +# How much of the span's final message is quoted back in the fork +# instruction as the "summarize up to HERE" boundary marker. Long enough to +# be unambiguous in a real transcript, short enough that it is never a +# material share of the appended (uncached) tail. +_FORK_BOUNDARY_EXCERPT_CHARS = 300 + # tool_result_shape config values. "head" (default) keeps the leading slice # only -- the pre-existing behavior. "head_tail" splits the budget in half # and keeps both ends with an explicit omission marker between them. See @@ -419,6 +502,28 @@ def _is_real_user_message(entry: dict[str, Any]) -> bool: """ +def _normalize_summary_call_mode(value: Any) -> str: + """Canonicalize a `summary_call_mode` config value. + + Accepts the two real modes plus the documented "inline" alias for + "standalone" (the commissioning lane brief and the work item named the + same default differently -- see _SUMMARY_CALL_MODE_ALIASES). An + unusable value logs a warning and falls back to "standalone", matching + how every other enum in this module is validated: a context manager + that refuses to start is worse than one that runs at the old default + and says so. + """ + mode = _SUMMARY_CALL_MODE_ALIASES.get(value, value) + if mode not in _VALID_SUMMARY_CALL_MODES: + logger.warning( + f"context-simple: unknown summary_call_mode {value!r} (expected " + f"one of {_VALID_SUMMARY_CALL_MODES!r}); falling back to " + f"{SUMMARY_CALL_MODE_STANDALONE!r}" + ) + return SUMMARY_CALL_MODE_STANDALONE + return mode + + async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = None): """ Mount the simple context manager. @@ -476,8 +581,21 @@ async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = boundaries / +83% run cost in the T0/T1 eval. Raising this is the cheapest lever; see README "Known issue: boundary refire". + - summary_call_mode: "standalone" (default, byte-identical to the + pre-existing summarizer call) or "fork" -- issue the + summarization ask as a pure append onto the prefix the main + line already sent, so it can be cache-read instead of paying + fresh input tokens for the span. "inline" is accepted as an + alias for "standalone". Only consulted when + compaction_strategy == "summary". Fork mode additionally + requires the caller to have called note_request_sent(); see + module docstring "Cache-safe forking of the summarization + call". - summarization_model: Model identifier passed to the summarizer's - ChatRequest (default: None, i.e. provider default). + ChatRequest (default: None, i.e. provider default). Setting it + CONFLICTS with summary_call_mode "fork" (a summarizer routed to + a different model cannot read the main line's cache) and makes + the fork fall back to standalone, loudly. - summarization_prompt_path: Path to a file overriding DEFAULT_SUMMARIZATION_PROMPT (default: None). - summarization_timeout_s: Seconds to wait for the summarizer's @@ -538,6 +656,10 @@ async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = ) compaction_strategy = COMPACTION_STRATEGY_PROGRESSIVE + summary_call_mode = _normalize_summary_call_mode( + config.get("summary_call_mode", SUMMARY_CALL_MODE_STANDALONE) + ) + context = SimpleContextManager( max_tokens=config.get("max_tokens", 200_000), compact_threshold=config.get("compact_threshold", 0.92), @@ -561,6 +683,7 @@ async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = ), compaction_strategy=compaction_strategy, summary_trigger=config.get("summary_trigger", 0.60), + summary_call_mode=summary_call_mode, summarization_model=config.get("summarization_model"), summarization_prompt_path=config.get("summarization_prompt_path"), summarization_timeout_s=config.get("summarization_timeout_s", 30.0), @@ -652,6 +775,7 @@ def __init__( compact_max_consecutive_skips: int = DEFAULT_MAX_CONSECUTIVE_SKIPS, compaction_strategy: str = COMPACTION_STRATEGY_PROGRESSIVE, summary_trigger: float = 0.60, + summary_call_mode: str = SUMMARY_CALL_MODE_STANDALONE, summarization_model: str | None = None, summarization_prompt_path: str | None = None, summarization_timeout_s: float = 30.0, @@ -717,8 +841,21 @@ def __init__( compaction boundaries (+84%) and run cost (+83%) -- see module docstring and README "Known issue: boundary refire". + summary_call_mode: "standalone" (default; the pre-existing + two-message summarizer call, byte-identical) or "fork" (the + same ask appended onto the main line's already-sent prefix + so the provider can cache-read it). "inline" is an accepted + alias for "standalone". Only consulted when + compaction_strategy == "summary". See module docstring + "Cache-safe forking of the summarization call" -- in + particular, fork mode needs note_request_sent() to have been + called, and falls back to standalone (loudly) when it has + not. summarization_model: Model identifier for the summarizer's own ChatRequest. None uses the provider's default model. + Incompatible with summary_call_mode "fork" -- a summarizer + pointed at another model reads no cache the main line wrote, + so the fork falls back to standalone rather than pretending. summarization_prompt_path: Path to a file overriding DEFAULT_SUMMARIZATION_PROMPT. None uses the built-in prompt. summarization_timeout_s: Seconds to wait for the summarizer's @@ -778,6 +915,7 @@ def __init__( ) compaction_strategy = COMPACTION_STRATEGY_PROGRESSIVE self.compaction_strategy = compaction_strategy + self.summary_call_mode = _normalize_summary_call_mode(summary_call_mode) self.summary_trigger = summary_trigger self.summarization_model = summarization_model self.summarization_prompt_path = summarization_prompt_path @@ -852,6 +990,43 @@ def __init__( self._summarization_failures: int = 0 self._summarization_task: "asyncio.Task[None] | None" = None self._summary_absorbed_count: int = 0 + # --- Cache-safe summarizer fork state (summary_call_mode == "fork") --- + # ALL of these stay None/0/False unless BOTH compaction_strategy == + # "summary" AND summary_call_mode == "fork"; nothing below is even + # written in any other configuration (see _finalize_view), which is + # what makes the default path byte-identical by construction rather + # than by inspection. + # + # `_last_request_view` is this module's own last returned view, kept + # PRE-strip (it still carries `_seq`) so the fork can verify the span + # it was asked to summarize is actually present in the prefix it is + # about to append to. `_sent_*` is the richer, optional truth handed + # over by the caller through note_request_sent(). + # + # `_view_serial` / `_sent_serial` exist for one reason: a caller + # that wires note_request_sent() ONCE (at startup, in a helper that + # only runs on the first turn) would otherwise hand the fork a + # request from turn 1 to append to on turn 40. That prefix is not + # the parent's prefix any more, so appending to it is a guaranteed + # miss AND a wasted cache write -- a silent misalignment wearing a + # correct-looking API call. Matching serials is how "the caller told + # us about THIS request" is distinguished from "the caller told us + # about SOME request, once". + self._last_request_view: list[dict[str, Any]] | None = None + self._view_serial: int = 0 + self._sent_messages: list[dict[str, Any]] | None = None + self._sent_serial: int | None = None + self._sent_tools: Any = None + self._sent_tools_supplied: bool = False + self._sent_model: str | None = None + # Observability: what the LAST summarizer call actually did, and how + # many times a requested fork had to fall back. An eval arm reads + # these to tell a real fork from a silently unforked one. + self._last_summary_call: dict[str, Any] | None = None + self._summary_fork_fallbacks: int = 0 + # Precondition names already warned about, so a session that can + # never fork logs once per reason instead of once per summarization. + self._fork_warned: set[str] = set() # Real-usage token meter state (see _on_llm_response / # _measure_working_tokens). `_last_measured_prompt_tokens` holds the # most recent real usage observed via `llm:response` @@ -979,6 +1154,78 @@ async def set_system_prompt_factory( self._system_prompt_factory = factory logger.info("System prompt factory registered - will refresh on each request") + def note_request_sent( + self, + messages: list[dict[str, Any]] | None = None, + *, + tools: Any = None, + model: str | None = None, + ) -> None: + """OPTIONAL. Tell this module what request the caller actually sent. + + This exists for exactly one reason: `summary_call_mode: "fork"` + re-issues the summarization ask as a PURE APPEND onto the prefix the + main line already sent, and this module cannot see that prefix in + full. It is handed messages; it is never handed the tool specs, and + it never sees the hook-injected tail an orchestrator may append + AFTER `get_messages_for_request()` returns. Both are part of what a + provider caches -- tool specs are serialized ahead of the system + block -- so a fork built without them is not an append onto the + cached prefix at all, and would pay full price for the whole + conversation. That is strictly worse than the standalone call it + replaces, so fork mode refuses to run without this. + + Completely inert unless BOTH compaction_strategy == "summary" AND + summary_call_mode == "fork". Never mutates history: nothing here + enters `self.messages`, consumes a `_seq`, or moves any compaction + state. Callers that do not know about it lose nothing. + + Call it EVERY request, not once. A `messages` record is only used + while it still describes the most recent request this module served + -- a one-time wiring would otherwise have turn 40's fork append to + turn 1's request, which is a guaranteed cache miss dressed up as a + correct API call. A stale record is ignored (this module's own last + view is used instead), never trusted. + + Args: + messages: The exact message array sent, if known. Gives the fork + byte-parity with the wire -- required for an implicit, + match-forward-only cache (OpenAI, whose measured behavior + misses on anything that is not a strict superset of a cached + request). Omit it and the fork appends to this module's own + last returned view instead, which is what an + explicit-breakpoint cache (Anthropic) needs, since the + breakpoint lands on the last STABLE message -- exactly where + this module's view ends, before any ephemeral injection. + tools: The tool specs sent, in the order sent. Passing this at + all -- even as None or [] for a genuinely tool-free session + -- is what arms fork mode; "not supplied" and "supplied as + empty" are deliberately distinguishable, because guessing + between them is how a fork silently misaligns. + model: The resolved model, if the caller knows it. Pins the + fork to the same model the parent used; a summarizer routed + elsewhere reads no cache the main line wrote. + """ + if messages is not None: + self._sent_messages = list(messages) + self._sent_serial = self._view_serial + self._sent_tools = list(tools) if isinstance(tools, list) else tools + self._sent_tools_supplied = True + if model is not None: + self._sent_model = model + + @property + def last_summary_call_stats(self) -> dict[str, Any] | None: + """What the most recent summarizer call actually did. + + None before the first one. Otherwise a dict with `mode_requested`, + `mode_used`, `reason` (None when the requested mode was honored), + `prefix_messages`, and `fork_fallbacks` (session-cumulative). This + is how an eval arm distinguishes a real fork from a silently + unforked one WITHOUT patching the module. + """ + return dict(self._last_summary_call) if self._last_summary_call else None + async def get_messages_for_request( self, token_budget: int | None = None, @@ -1370,6 +1617,19 @@ def _reset_summary_strategy_state(self) -> None: self._summarization_failures = 0 self._summarization_task = None self._summary_absorbed_count = 0 + # Fork state is history-derived: a reset session's old prefix is not + # a prefix of anything any more, and the caller's note_request_sent() + # facts describe a request that no longer relates to this history. + # Keeping either would be exactly the stale-alignment bug fork mode + # exists to refuse. `_summary_fork_fallbacks` is a session-cumulative + # observability counter and deliberately survives. + self._last_request_view = None + self._view_serial = 0 + self._sent_messages = None + self._sent_serial = None + self._sent_tools = None + self._sent_tools_supplied = False + self._sent_model = None async def should_compact(self) -> bool: """Check if context should be compacted. @@ -1509,6 +1769,14 @@ def _finalize_view(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]] """ view = self._strip_internal_metadata(messages) self._last_sent_estimate = self._estimate_tokens(view) + # Remember the PRE-strip list (it still carries `_seq`) so a forked + # summarizer call can both reproduce this exact view byte-for-byte + # AND verify the span it was asked to summarize is present in it. + # Guarded so the default path allocates nothing new and stays + # byte-identical by construction, not by inspection. + if self._fork_armed(): + self._last_request_view = list(messages) + self._view_serial += 1 return view def _cache_aggregates(self) -> dict[str, int] | None: @@ -3762,9 +4030,18 @@ async def _maybe_trigger_summary_compaction( ) return + # Snapshot the fork prefix HERE, synchronously, rather than letting + # the background task read it whenever it happens to be scheduled. + # `_last_request_view` is rewritten on every request; a task that + # read it later would append to a prefix chosen by scheduling order. + # Capturing at trigger time makes the forked request a pure function + # of this moment -- deterministic, and therefore testable. None in + # every configuration but fork. + fork_prefix = self._capture_fork_prefix() + self._is_summarizing = True self._summarization_task = asyncio.create_task( - self._run_summary_compaction_task(seqs) + self._run_summary_compaction_task(seqs, fork_prefix=fork_prefix) ) def _select_summary_absorb_seqs(self, excess_tokens: int) -> list[int] | None: @@ -3881,7 +4158,284 @@ def result_indices(assistant_msg: dict[str, Any]) -> list[int]: return 0 return 0 # defensive; unreachable given the termination argument above - async def _run_summary_compaction_task(self, seqs: list[int]) -> None: + # --- Cache-safe fork of the summarizer call (summary_call_mode) -------- + # + # Everything from here to _run_summary_compaction_task is inert unless + # BOTH compaction_strategy == "summary" AND summary_call_mode == "fork". + # The default path never calls any of it (see _build_summary_request's + # first branch), which is what makes "default is byte-identical" a + # structural property rather than a claim. + + def _fork_armed(self) -> bool: + """True only when a forked summarizer call is actually configured.""" + return ( + self.compaction_strategy == COMPACTION_STRATEGY_SUMMARY + and self.summary_call_mode == SUMMARY_CALL_MODE_FORK + ) + + def _capture_fork_prefix(self) -> list[dict[str, Any]] | None: + """Snapshot the message array a forked call would append to. + + Prefers the caller's own `note_request_sent(messages=...)` record + (byte-parity with the wire, including any tail the orchestrator + injected after this module returned). Falls back to this module's + last returned view, which still ends exactly where an + explicit-breakpoint provider places its cache breakpoint. + """ + if not self._fork_armed(): + return None + source = self._last_request_view + if self._sent_messages is not None: + if self._sent_serial == self._view_serial: + source = self._sent_messages + else: + logger.debug( + "context-simple: ignoring a stale note_request_sent() " + f"message record (recorded at view {self._sent_serial}, " + f"now at view {self._view_serial}); appending to this " + "module's own last returned view instead" + ) + return list(source) if source is not None else None + + @staticmethod + def _message_identity(msg: dict[str, Any]) -> tuple[str, str, str]: + """A content-level identity for a message, used only to check span + presence when the prefix came from a caller and therefore had its + internal `_seq` already stripped.""" + return ( + str(msg.get("role", "")), + str(msg.get("content", "")), + str(msg.get("tool_call_id", "")), + ) + + def _prefix_contains_span( + self, prefix: list[dict[str, Any]], span: list[dict[str, Any]] + ) -> bool: + """Is every message of the span actually present in the prefix? + + A fork does NOT re-send the span -- the whole point is that the span + is already inside the prefix being appended to. If it is not (a + compaction removed it between the last request and this trigger), + the forked call would be asking the model to summarize text it + cannot see. Checked by `_seq` when the prefix carries them, and by + content identity when it came from a caller (post-strip). + """ + prefix_seqs = {self._extract_seq(m) for m in prefix} - {None} + span_seqs = {self._extract_seq(m) for m in span} - {None} + if span_seqs and span_seqs <= prefix_seqs: + return True + present = {self._message_identity(m) for m in prefix} + return all(self._message_identity(m) in present for m in span) + + def _fork_refusal_reason( + self, + messages_to_summarize: list[dict[str, Any]], + fork_prefix: list[dict[str, Any]] | None, + ) -> str | None: + """Why this summarization CANNOT be forked, or None if it can. + + Every branch here is a case where the forked request would not be a + true append onto the parent's cached prefix. A fork that misses pays + for the entire conversation as fresh input -- strictly worse than + the standalone call it replaces -- so each of these falls back + rather than half-forking. + """ + if not self._sent_tools_supplied: + return ( + "note_request_sent() has never been called, so the tool specs " + "the parent sent are unknown; tool specs are serialized ahead " + "of the system block, so a fork without them is not an append " + "onto the cached prefix at all" + ) + if self.summarization_model: + return ( + f"summarization_model={self.summarization_model!r} points the " + "summarizer at a different model than the main line, which " + "reads none of the cache the main line wrote" + ) + if not fork_prefix: + return ( + "no request has been recorded yet, so there is no prefix to " + "append to" + ) + if fork_prefix[-1].get("tool_calls"): + return ( + "the prefix ends on an assistant turn with unanswered " + "tool_calls; appending a user message there would interleave " + "between tool_use and tool_result" + ) + if not self._prefix_contains_span(fork_prefix, messages_to_summarize): + return ( + "the span selected for absorption is not present in the " + "recorded prefix, so a forked call could not see the text it " + "was asked to summarize" + ) + return None + + def _note_fork_fallback(self, reason: str) -> None: + """Record and (once per distinct reason) announce a refused fork. + + WARNING level and counted, never silent: a fork that quietly did not + happen is the exact failure mode that makes this treatment look like + it does not work. Once per reason, not once per call, so a session + that can never fork logs a handful of lines instead of hundreds. + """ + self._summary_fork_fallbacks += 1 + if reason in self._fork_warned: + logger.debug(f"context-simple: summarizer fork refused again ({reason})") + return + self._fork_warned.add(reason) + logger.warning( + f"context-simple: summary_call_mode='fork' requested but this " + f"summarization ran STANDALONE instead -- {reason}. The standalone " + "call is correct and costs what it always did; no cache reuse was " + "attempted. This is logged once per distinct reason." + ) + + def _span_boundary_excerpt(self, msg: dict[str, Any]) -> str: + """A short verbatim excerpt of the span's final message, used as the + 'summarize up to HERE' marker in the fork instruction.""" + content = msg.get("content", "") + if isinstance(content, list): + parts: list[str] = [] + for block in content: + if isinstance(block, dict): + text = block.get("text", "") + if text: + parts.append(str(text)) + elif hasattr(block, "text"): + parts.append(str(block.text)) + content = "\n".join(parts) + text = str(content).strip() + if not text: + # A tool_calls-only assistant turn has no text of its own; name + # the tools instead of emitting an empty, useless marker. + names = [ + str(tc.get("name") or tc.get("tool") or "") + for tc in (msg.get("tool_calls") or []) + if isinstance(tc, dict) + ] + names = [n for n in names if n] + text = ( + f"[{msg.get('role', 'unknown')} turn calling: {', '.join(names)}]" + if names + else f"[{msg.get('role', 'unknown')} turn with no text content]" + ) + if len(text) > _FORK_BOUNDARY_EXCERPT_CHARS: + text = text[:_FORK_BOUNDARY_EXCERPT_CHARS] + "..." + return text + + def _format_fork_instruction( + self, messages_to_summarize: list[dict[str, Any]], prompt: str + ) -> str: + """The single user message a forked call appends. + + Carries the summarization prompt (which in standalone mode is a + `role: "system"` message -- see module docstring for why a fork must + NOT add one) plus explicit scoping, because a fork does not re-send + the span: the model reads it from the prefix it is already holding, + so the instruction has to say which part of that prefix to + summarize. This is a REAL difference from standalone, which scopes + by construction: standalone can only see the span, a fork can see + everything and is asked to attend to the span. + """ + n = len(messages_to_summarize) + excerpt = self._span_boundary_excerpt(messages_to_summarize[-1]) + return ( + f"{prompt}\n\n" + "SCOPE OF THIS SUMMARY. Summarize ONLY the OLDEST part of the " + f"conversation above: the first {n} message(s) following the " + "system prompt -- the span about to be retired from context to " + "make room. That span ENDS with the message excerpted below. Do " + "not summarize anything after it, and do not describe this " + "instruction.\n\n" + "--- final message of the span (verbatim excerpt) ---\n" + f"{excerpt}\n" + "--- end excerpt ---" + ) + + def _build_summary_request( + self, + messages_to_summarize: list[dict[str, Any]], + fork_prefix: list[dict[str, Any]] | None, + ) -> tuple[Any, str, str | None]: + """Build the summarizer's ChatRequest. + + Returns (request, mode_actually_used, fallback_reason_or_None). The + standalone branch below is verbatim the pre-existing call and is the + ONLY branch reachable unless fork mode is both configured and + satisfiable. + """ + from amplifier_core import ChatRequest, Message + + prompt = self._get_summarization_prompt() + reason: str | None = None + + if self._fork_armed(): + reason = self._fork_refusal_reason(messages_to_summarize, fork_prefix) + if reason is None: + try: + assert fork_prefix is not None # guaranteed by the check above + return ( + self._build_fork_request( + messages_to_summarize, fork_prefix, prompt + ), + SUMMARY_CALL_MODE_FORK, + None, + ) + except Exception as e: + # A prefix message this module never created (unexpected + # content shape from a caller, say) can fail Message + # validation. Falling back keeps the summary happening at + # today's cost instead of turning a cache optimization + # into a lost summary. + reason = f"the forked request could not be built ({e!r})" + self._note_fork_fallback(reason) + + formatted = self._format_messages_for_summarization(messages_to_summarize) + request = ChatRequest( + messages=[ + Message(role="system", content=prompt), + Message(role="user", content=formatted), + ], + model=self.summarization_model, + ) + return request, SUMMARY_CALL_MODE_STANDALONE, reason + + def _build_fork_request( + self, + messages_to_summarize: list[dict[str, Any]], + fork_prefix: list[dict[str, Any]], + prompt: str, + ) -> Any: + """The forked request: the parent's prefix, then ONE appended user + message. Nothing else -- no extra system message, no re-sent span, + no reordering. + + Deliberately calls `_strip_internal_metadata` and NOT + `_finalize_view`: the latter also rewrites `_last_sent_estimate`, + the hybrid meter's conservatism comparand, which describes the view + the MAIN line sent. A summarizer call must not move it. + """ + from amplifier_core import ChatRequest, Message + + prefix_view = self._strip_internal_metadata(fork_prefix) + messages = [Message(**msg) for msg in prefix_view] + messages.append( + Message( + role="user", + content=self._format_fork_instruction(messages_to_summarize, prompt), + ) + ) + return ChatRequest( + messages=messages, + tools=self._sent_tools, + model=self._sent_model, + ) + + async def _run_summary_compaction_task( + self, seqs: list[int], fork_prefix: list[dict[str, Any]] | None = None + ) -> None: """Background task: call the summarizer over the message span identified by `seqs` and stash the result in `_pending_summary` for the next get_messages_for_request()/_compact_ephemeral() call to @@ -3908,24 +4462,29 @@ async def _run_summary_compaction_task(self, seqs: list[int]) -> None: if provider is None: raise RuntimeError("no cached provider available for summary compaction") - prompt = self._get_summarization_prompt() - formatted = self._format_messages_for_summarization(messages_to_summarize) - - from amplifier_core import ChatRequest, Message - - request = ChatRequest( - messages=[ - Message(role="system", content=prompt), - Message(role="user", content=formatted), - ], - model=self.summarization_model, + request, call_mode, fallback_reason = self._build_summary_request( + messages_to_summarize, fork_prefix ) + self._last_summary_call = { + "mode_requested": self.summary_call_mode, + "mode_used": call_mode, + "reason": fallback_reason, + "prefix_messages": ( + len(request.messages) - 1 + if call_mode == SUMMARY_CALL_MODE_FORK + else 0 + ), + "fork_fallbacks": self._summary_fork_fallbacks, + } if self._hooks is not None: try: await self._hooks.emit( "context:pre_summarize", - {"message_count": len(messages_to_summarize)}, + { + "message_count": len(messages_to_summarize), + "call_mode": call_mode, + }, ) except Exception as e: logger.warning(f"Could not emit context:pre_summarize: {e}") @@ -3944,7 +4503,10 @@ async def _run_summary_compaction_task(self, seqs: list[int]) -> None: try: await self._hooks.emit( "context:post_summarize", - {"summary_length": len(summary_text)}, + { + "summary_length": len(summary_text), + "call_mode": call_mode, + }, ) except Exception as e: logger.warning(f"Could not emit context:post_summarize: {e}") diff --git a/tests/test_summary_call_mode_fork.py b/tests/test_summary_call_mode_fork.py new file mode 100644 index 0000000..e039f93 --- /dev/null +++ b/tests/test_summary_call_mode_fork.py @@ -0,0 +1,875 @@ +"""Adversarial tests for `summary_call_mode: "fork"`. + +STEP 0 finding this file exists to lock in: before this change the +summarizer sent a STANDALONE two-message request -- its own ~955-char +`role: "system"` prompt plus a freshly formatted plain-text rendering of +the span -- sharing not one byte of prefix with the main conversation. +Every token of the span was billed as fresh input while the provider was +already holding that exact span warm for the main line. + +`summary_call_mode: "fork"` re-issues the same ask as a PURE APPEND onto +the prefix the main line already sent. Pure append is the one mutation +measured as a cache HIT under the grow-only rule (probe P4: identical +repeat 9,789 HIT, pure append 9,789 HIT, strict truncation 0 MISS, +middle-drop 0 MISS). + +The tests below are written against the ways this can silently go wrong, +not against the way it is supposed to work: + + 1. THE DEFAULT MUST NOT MOVE. A new branch that is "usually" inert is + not inert. Group A pins the standalone request byte-for-byte and + proves the fork bookkeeping is never even written unless armed. + 2. A SILENTLY UNFORKED FORK IS THE REAL FAILURE. A fork that does not + reproduce the parent's prefix wins nothing AND pays for the whole + conversation -- strictly worse than what it replaces. Group D proves + every misalignment refuses, falls back to the standalone call, and + SAYS SO (warning + counter + reported mode). + 3. THE MAIN LINE IS NOT THE SUMMARIZER'S SCRATCHPAD. Group C proves the + fork consumes no `_seq`, appends nothing to history, does not move + the hybrid meter's `_last_sent_estimate` comparand, and leaves the + next served view byte-identical to an unforked control. + 4. TOOL PAIRS STAY WHOLE. Group E proves the absorb-boundary snapping + (the donor's exact production failure) is not perturbed by the call + mode -- the fork changes how the summarizer is CALLED, never what is + selected. +""" + +import asyncio +import hashlib +import json +import logging + +import pytest +from amplifier_core import ChatResponse, Message, TextBlock +from amplifier_module_context_simple import SimpleContextManager, mount + + +class _FakeProvider: + """Minimal stand-in for a Provider -- records every request it is + handed so tests can assert on the exact shape that would go on the + wire. Deliberately has neither `get_model_info` nor `get_info`, so + `_calculate_budget` falls back to `self.max_tokens`.""" + + def __init__(self, response_text: str = "SUMMARY TEXT"): + self.response_text = response_text + self.calls: list = [] + + async def complete(self, request): + self.calls.append(request) + return ChatResponse(content=[TextBlock(type="text", text=self.response_text)]) + + +class _Coordinator: + def __init__(self): + self.hooks = None + self.mounted = {} + + async def mount(self, kind, instance): + self.mounted[kind] = instance + + +def _tools(): + """A tool spec list shaped like what an orchestrator actually sends.""" + return [ + { + "name": "bash", + "description": "run a command", + "parameters": {"type": "object", "properties": {}}, + } + ] + + +def _digest(messages) -> str: + """Canonical hash of a message array, whatever form it arrives in.""" + dumps = [] + for m in messages: + if isinstance(m, Message): + dumps.append(m.model_dump()) + else: + dumps.append(Message(**m).model_dump()) + return hashlib.sha256( + json.dumps(dumps, sort_keys=True, default=str).encode() + ).hexdigest() + + +def _strip_timestamps(messages: list[dict]) -> list[dict]: + result = [] + for msg in messages: + meta = dict(msg.get("metadata") or {}) + meta.pop("timestamp", None) + result.append({**msg, "metadata": meta}) + return result + + +async def _await_pending_task(context: SimpleContextManager) -> None: + task = context._summarization_task + assert task is not None, "expected a background summarization task in flight" + await task + + +def _summary_manager(**overrides) -> SimpleContextManager: + """A manager whose summary trigger is reachable without needing a + 200k-token fixture. `max_tokens` is corrected per test once real + estimator usage is known (same technique the existing summary tests + use).""" + kwargs = dict( + compaction_strategy="summary", + compaction_notice_enabled=False, + protected_recent=0.5, + summary_trigger=0.3, + target_usage=0.2, + compact_threshold=0.99, # keep the outer progressive gate CLOSED + max_tokens=1_000_000, + ) + kwargs.update(overrides) + return SimpleContextManager(**kwargs) + + +async def _fill(context: SimpleContextManager, turns: int = 30) -> None: + for i in range(turns): + await context.add_message( + {"role": "user", "content": f"user turn {i} " + "x" * 40} + ) + await context.add_message( + {"role": "assistant", "content": f"assistant reply {i} " + "y" * 40} + ) + + +async def _arm_below_trigger(context: SimpleContextManager) -> list[dict]: + """Serve one request with usage BELOW summary_trigger. + + This is what records the prefix a later fork appends to -- and it is + also the honest ordering: on the very first request of a session there + is no sent prefix yet, so there is nothing to fork onto. + """ + raw = context._estimate_tokens(context.messages) + context.max_tokens = int(raw / 0.1) # usage ~0.10, below trigger 0.3 + view = await context.get_messages_for_request(provider=_FakeProvider()) + assert context._is_summarizing is False, "must not trigger below summary_trigger" + return view + + +def _cross_trigger(context: SimpleContextManager) -> None: + raw = context._estimate_tokens(context.messages) + context.max_tokens = int(raw / 0.5) # usage ~0.50: above 0.3, below 0.99 + + +# --------------------------------------------------------------------------- +# Group A -- the default must not move +# --------------------------------------------------------------------------- + + +def test_default_summary_call_mode_is_standalone(): + assert SimpleContextManager().summary_call_mode == "standalone" + assert ( + SimpleContextManager(compaction_strategy="summary").summary_call_mode + == "standalone" + ) + + +def test_inline_is_an_accepted_alias_for_standalone(caplog): + """The commissioning lane brief named the default mode "inline"; the + work item named it "standalone". Both must mean today's behavior, and + neither may produce a warning.""" + with caplog.at_level(logging.WARNING): + context = SimpleContextManager(summary_call_mode="inline") + assert context.summary_call_mode == "standalone" + assert not [r for r in caplog.records if "summary_call_mode" in r.message] + + +def test_unknown_summary_call_mode_falls_back_with_warning(caplog): + with caplog.at_level(logging.WARNING): + context = SimpleContextManager(summary_call_mode="bogus") + assert context.summary_call_mode == "standalone" + assert any("unknown summary_call_mode" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_mount_threads_summary_call_mode_through(): + coordinator = _Coordinator() + await mount(coordinator, {"compaction_strategy": "summary", "summary_call_mode": "fork"}) + assert coordinator.mounted["context"].summary_call_mode == "fork" + + coordinator = _Coordinator() + await mount(coordinator, {"summary_call_mode": "inline"}) + assert coordinator.mounted["context"].summary_call_mode == "standalone" + + coordinator = _Coordinator() + await mount(coordinator, {"summary_call_mode": "nonsense"}) + assert coordinator.mounted["context"].summary_call_mode == "standalone" + + +@pytest.mark.asyncio +async def test_default_mode_summarizer_request_is_byte_identical(): + """The standalone request must remain EXACTLY what it was before this + feature existed: two messages, system prompt then formatted span, + model from `summarization_model`. Asserted against independently + rebuilt expected content, not against itself.""" + context = _summary_manager(summarization_model="gpt-test") + await _fill(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls) == 1 + request = provider.calls[0] + assert len(request.messages) == 2 + assert request.model == "gpt-test" + assert request.tools is None + + seqs = sorted(context._pending_summary["seqs"]) + span = [m for m in context.messages if context._extract_seq(m) in set(seqs)] + assert request.messages[0].role == "system" + assert request.messages[0].content == context._get_summarization_prompt() + assert request.messages[1].role == "user" + assert request.messages[1].content == context._format_messages_for_summarization(span) + assert context.last_summary_call_stats["mode_used"] == "standalone" + assert context.last_summary_call_stats["reason"] is None + + +@pytest.mark.asyncio +async def test_default_mode_never_records_a_fork_prefix(): + """The fork bookkeeping must not merely be unused in the default mode + -- it must never be WRITTEN. An always-on capture would be a silent + per-request list allocation on the hot path.""" + for kwargs in ({}, {"compaction_strategy": "summary"}): + context = SimpleContextManager(**kwargs) + for i in range(5): + await context.add_message({"role": "user", "content": f"m{i}"}) + await context.get_messages_for_request(provider=_FakeProvider()) + assert context._last_request_view is None + assert context._sent_tools_supplied is False + assert context._summary_fork_fallbacks == 0 + assert context.last_summary_call_stats is None + + +@pytest.mark.asyncio +async def test_note_request_sent_is_inert_when_fork_is_not_configured(): + """A caller that always calls the seam must not change behavior for + every session that has not opted into forking.""" + context = _summary_manager() # standalone + await _fill(context) + context.note_request_sent(tools=_tools(), model="pinned-model") + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + assert len(request.messages) == 2 + assert request.tools is None + assert request.model is None + assert context._last_request_view is None + assert context._summary_fork_fallbacks == 0 + + +@pytest.mark.asyncio +async def test_note_request_sent_never_touches_history(): + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=3) + before = json.dumps(_strip_timestamps(context.messages), default=str) + seq_before = context._next_seq + + context.note_request_sent( + [{"role": "user", "content": "an injected tail the orchestrator added"}], + tools=_tools(), + model="m", + ) + + assert json.dumps(_strip_timestamps(context.messages), default=str) == before + assert context._next_seq == seq_before + + +# --------------------------------------------------------------------------- +# Group B -- the fork is a pure append onto what the main line actually sent +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fork_request_is_the_parent_prefix_plus_exactly_one_user_message(): + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools(), model="pinned-model") + parent_view = await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + assert context.last_summary_call_stats["mode_used"] == "fork" + assert len(request.messages) == len(parent_view) + 1, ( + "a fork appends exactly one message -- no extra system message, no " + "re-sent span" + ) + # G-FORK-PREFIX, in unit form: the fork minus its trailing message is + # byte-identical to the parent request. + assert _digest(request.messages[:-1]) == _digest(parent_view) + assert request.messages[-1].role == "user" + + +@pytest.mark.asyncio +async def test_fork_adds_no_system_message(): + """A per-summarization `role: "system"` message would be hoisted into + the provider's single top-level system block and rewrite the cached + system prefix -- the exact failure already measured for the summary + tier and the compaction notice. The prompt must ride the appended user + message instead.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + parent_view = await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + parent_systems = sum(1 for m in parent_view if m.get("role") == "system") + fork_systems = sum(1 for m in request.messages if m.role == "system") + assert fork_systems == parent_systems + assert context._get_summarization_prompt() in request.messages[-1].content + + +@pytest.mark.asyncio +async def test_fork_pins_tools_and_model_from_note_request_sent(): + """Tool specs are serialized ahead of the system block, and a + summarizer routed to another model reads none of the cache the main + line wrote. Both must come from what the caller says it sent.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + tools = _tools() + context.note_request_sent(tools=tools, model="pinned-model") + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + assert request.model == "pinned-model" + assert request.tools is not None and len(request.tools) == len(tools) + assert request.tools[0].name == "bash" + + +@pytest.mark.asyncio +async def test_fork_does_not_resend_the_span(): + """The entire point: the span is already inside the prefix. Re-sending + it would cost exactly what standalone costs today PLUS the prefix -- + a regression dressed as an optimization.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + appended = provider.calls[0].messages[-1].content + seqs = set(context._pending_summary["seqs"]) + span = [m for m in context.messages if context._extract_seq(m) in seqs] + assert len(span) > 3, "fixture must produce a span worth not re-sending" + formatted = context._format_messages_for_summarization(span) + assert formatted not in appended + # Only the boundary marker is quoted back, and it is bounded. + assert len(appended) < len(formatted) + + +@pytest.mark.asyncio +async def test_fork_instruction_carries_the_prompt_and_explicit_scope(): + """A fork can SEE the whole conversation, unlike standalone which can + only see the span. Scoping therefore has to be stated, and the span's + end has to be identifiable.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + appended = provider.calls[0].messages[-1].content + seqs = set(context._pending_summary["seqs"]) + span = [m for m in context.messages if context._extract_seq(m) in seqs] + + assert context._get_summarization_prompt() in appended + assert f"the first {len(span)} message(s)" in appended + assert "Do not summarize anything after it" in appended + last_text = str(span[-1]["content"])[:60] + assert last_text in appended, "the span's final message must be identifiable" + + +@pytest.mark.asyncio +async def test_fork_uses_caller_supplied_messages_verbatim_when_given(): + """An orchestrator may append hook-injected content AFTER this module + returns its view. An implicit, match-forward-only cache misses on + anything that is not a strict superset of what it cached, so when the + caller tells us what actually went on the wire, THAT is what gets + appended to.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + module_view = await _arm_below_trigger(context) + wire = [*module_view, {"role": "user", "content": "injected"}] + context.note_request_sent(wire, tools=_tools()) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + assert len(request.messages) == len(wire) + 1 + assert _digest(request.messages[:-1]) == _digest(wire) + assert "injected" in request.messages[-2].content + + +# --------------------------------------------------------------------------- +# Group C -- the main line is not the summarizer's scratchpad +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fork_consumes_no_seq_and_appends_nothing_to_history(): + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + seq_before = context._next_seq + history_before = json.dumps(_strip_timestamps(context.messages), default=str) + removed_before = set(context._removed_seqs) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert context.last_summary_call_stats["mode_used"] == "fork" + assert context._next_seq == seq_before, "the fork must not consume a _seq" + assert ( + json.dumps(_strip_timestamps(context.messages), default=str) == history_before + ), "the fork must not append to, reorder, or edit history" + assert context._removed_seqs == removed_before + + +@pytest.mark.asyncio +async def test_fork_does_not_move_the_hybrid_meter_comparand(): + """`_last_sent_estimate` describes the view the MAIN line sent; it is + what the hybrid meter's conservatism guard compares the next + `llm:response` against. Building the fork through `_finalize_view` + would silently rewrite it with the summarizer's own request.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + estimate_after_main_request = context._last_sent_estimate + await _await_pending_task(context) + + assert context.last_summary_call_stats["mode_used"] == "fork" + assert context._last_sent_estimate == estimate_after_main_request + + +@pytest.mark.asyncio +async def test_forked_and_unforked_sessions_serve_identical_views(): + """The call mode changes how the summarizer is CALLED. What the main + line is served must be bit-for-bit the same either way.""" + forked = _summary_manager(summary_call_mode="fork") + control = _summary_manager() + for ctx in (forked, control): + await _fill(ctx) + forked.note_request_sent(tools=_tools()) + await _arm_below_trigger(forked) + await _arm_below_trigger(control) + for ctx in (forked, control): + _cross_trigger(ctx) + + for ctx in (forked, control): + await ctx.get_messages_for_request(provider=_FakeProvider("SAME SUMMARY")) + await _await_pending_task(ctx) + + assert forked.last_summary_call_stats["mode_used"] == "fork" + assert control.last_summary_call_stats["mode_used"] == "standalone" + + for ctx in (forked, control): + ctx.compact_threshold = 0.3 + forked_view = await forked.get_messages_for_request(provider=_FakeProvider()) + control_view = await control.get_messages_for_request(provider=_FakeProvider()) + + assert _strip_timestamps(forked_view) == _strip_timestamps(control_view) + assert forked._removed_seqs == control._removed_seqs + + +# --------------------------------------------------------------------------- +# Group D -- every misalignment refuses, falls back, and says so +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fork_without_note_request_sent_falls_back_loudly(caplog): + """Tool specs are part of the cached prefix and this module is never + handed them. Guessing "probably no tools" is precisely how a fork + silently misaligns and pays for the whole conversation.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + with caplog.at_level(logging.WARNING): + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2, "must be the standalone request" + stats = context.last_summary_call_stats + assert stats["mode_requested"] == "fork" + assert stats["mode_used"] == "standalone" + assert "note_request_sent()" in stats["reason"] + assert context._summary_fork_fallbacks == 1 + assert any("ran STANDALONE instead" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_fork_with_summarization_model_falls_back_loudly(caplog): + context = _summary_manager(summary_call_mode="fork", summarization_model="other") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + with caplog.at_level(logging.WARNING): + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2 + assert provider.calls[0].model == "other", "the explicit model is still honored" + assert "different model" in context.last_summary_call_stats["reason"] + assert any("ran STANDALONE instead" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_fork_on_the_first_request_of_a_session_falls_back(): + """Nothing has been sent yet, so there is no prefix to append to. This + is normal, not an error -- but it must not pretend to fork.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2 + assert "no request has been recorded yet" in context.last_summary_call_stats["reason"] + + +@pytest.mark.asyncio +async def test_fork_refuses_when_the_prefix_ends_on_unanswered_tool_calls(): + """Appending a user message after an assistant turn whose tool results + have not arrived interleaves between tool_use and tool_result -- the + same atomicity the compaction notice already guards.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=10) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + # The recorded prefix now ends on an assistant turn awaiting results. + context._last_request_view = [ + *context._last_request_view, + { + "role": "assistant", + "content": "", + "tool_calls": [{"id": "call-1", "tool": "bash", "arguments": {}}], + }, + ] + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2 + assert "unanswered" in context.last_summary_call_stats["reason"] + + +@pytest.mark.asyncio +async def test_a_stale_caller_message_record_is_ignored_not_trusted(): + """A caller that wires note_request_sent() ONCE (startup helper, first + turn only) would otherwise have turn N's fork append to turn 1's + request -- a guaranteed miss AND a wasted cache write, wearing a + correct-looking API call. The fresh module view must win.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=10) + # Turn 1: the caller records what it sent. + await _arm_below_trigger(context) + context.note_request_sent( + [{"role": "user", "content": "turn one, long ago"}], tools=_tools() + ) + # Several more turns go by without the caller telling us anything. + await _fill(context, turns=10) + fresh_view = await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + request = provider.calls[0] + assert context.last_summary_call_stats["mode_used"] == "fork" + assert _digest(request.messages[:-1]) == _digest(fresh_view) + assert not any("long ago" in str(m.content) for m in request.messages) + + +@pytest.mark.asyncio +async def test_fork_refuses_when_the_span_is_absent_from_the_prefix(): + """A fork does not re-send the span. If the recorded prefix no longer + contains it, the model would be asked to summarize text it cannot + see.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + # A prefix that is real but unrelated to the span being absorbed. + context.note_request_sent( + [{"role": "user", "content": "an unrelated conversation"}], tools=_tools() + ) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2 + assert "not present in the recorded prefix" in ( + context.last_summary_call_stats["reason"] + ) + + +@pytest.mark.asyncio +async def test_repeated_fallbacks_warn_once_per_reason_but_count_every_time(caplog): + """A session that can never fork should cost a handful of log lines, + not one per summarization -- while the counter still tells the truth.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=6) + + with caplog.at_level(logging.WARNING): + for _ in range(4): + context._note_fork_fallback("reason one") + context._note_fork_fallback("reason two") + + assert context._summary_fork_fallbacks == 5 + warnings = [r for r in caplog.records if "ran STANDALONE instead" in r.message] + assert len(warnings) == 2 + + +@pytest.mark.asyncio +async def test_a_failed_fork_build_falls_back_instead_of_losing_the_summary(): + """A prefix message this module never created can fail request + validation. Losing the summary over a cache optimization would be a + strictly worse outcome than paying today's price.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + await _arm_below_trigger(context) + context.note_request_sent( + [{"role": "not-a-real-role", "content": "x"}], tools=_tools() + ) + # Make the span check pass so the failure comes from request building. + context._prefix_contains_span = lambda prefix, span: True + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + await _await_pending_task(context) + + assert len(provider.calls[0].messages) == 2 + assert "could not be built" in context.last_summary_call_stats["reason"] + assert context._pending_summary is not None, "the summary still happened" + + +@pytest.mark.asyncio +async def test_hooks_report_the_mode_actually_used(): + """An eval arm has to be able to count real forks without patching the + module.""" + events: list[tuple[str, dict]] = [] + + class _Hooks: + async def emit(self, event, data): + events.append((event, data)) + + context = _summary_manager(summary_call_mode="fork") + context._hooks = _Hooks() + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + await context.get_messages_for_request(provider=_FakeProvider()) + await _await_pending_task(context) + + modes = { + name: data.get("call_mode") + for name, data in events + if name in ("context:pre_summarize", "context:post_summarize") + } + assert modes == { + "context:pre_summarize": "fork", + "context:post_summarize": "fork", + } + + +# --------------------------------------------------------------------------- +# Group E -- tool-pair integrity is not perturbed by the call mode +# --------------------------------------------------------------------------- + + +async def _fill_with_tool_pairs(context: SimpleContextManager) -> None: + for i in range(12): + await context.add_message( + {"role": "user", "content": f"do thing {i} " + "x" * 30} + ) + await context.add_message( + { + "role": "assistant", + "content": "working", + "tool_calls": [{"id": f"call-{i}", "tool": "bash", "arguments": {}}], + } + ) + await context.add_message( + {"role": "tool", "tool_call_id": f"call-{i}", "content": "out " + "y" * 30} + ) + + +@pytest.mark.asyncio +async def test_call_mode_does_not_change_which_span_is_selected(): + forked = _summary_manager(summary_call_mode="fork") + control = _summary_manager() + for ctx in (forked, control): + await _fill_with_tool_pairs(ctx) + + assert forked._select_summary_absorb_seqs(500) == control._select_summary_absorb_seqs( + 500 + ) + + +@pytest.mark.asyncio +async def test_fork_mode_never_serves_an_orphaned_tool_result(): + """The donor's exact production failure (a dropped `function_call` + whose `function_call_output` survived) must stay impossible in fork + mode too.""" + context = _summary_manager(summary_call_mode="fork", protected_recent=0.3) + await _fill_with_tool_pairs(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + await context.get_messages_for_request(provider=_FakeProvider()) + await _await_pending_task(context) + context.compact_threshold = 0.3 + view = await context.get_messages_for_request(provider=_FakeProvider()) + + call_ids = { + tc.get("id") + for m in view + for tc in (m.get("tool_calls") or []) + if isinstance(tc, dict) + } + result_ids = {m.get("tool_call_id") for m in view if m.get("role") == "tool"} + assert result_ids <= call_ids, "a tool result was served without its call" + + +# --------------------------------------------------------------------------- +# Group F -- reset +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_reset_clears_fork_alignment_state(): + """After set_messages()/clear() the recorded prefix is a prefix of + nothing, and the caller's facts describe a request unrelated to this + history. Keeping either is the stale-alignment bug fork mode exists to + refuse.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=4) + context.note_request_sent(tools=_tools(), model="m") + await _arm_below_trigger(context) + assert context._last_request_view is not None + + await context.clear() + + assert context._last_request_view is None + assert context._sent_messages is None + assert context._sent_tools is None + assert context._sent_tools_supplied is False + assert context._sent_model is None + + +@pytest.mark.asyncio +async def test_fork_state_survives_nothing_across_set_messages(): + context = _summary_manager(summary_call_mode="fork") + await _fill(context, turns=4) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + + await context.set_messages([{"role": "user", "content": "resumed session"}]) + + assert context._last_request_view is None + assert context._sent_tools_supplied is False + + +@pytest.mark.asyncio +async def test_fork_snapshot_is_taken_at_trigger_time_not_task_time(): + """The background task must append to the prefix chosen when the + trigger fired, not to whatever `_last_request_view` happens to hold + when the event loop gets around to running it.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + parent_view = await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await context.get_messages_for_request(provider=provider) + # Simulate another request landing while the summarizer is in flight. + context._last_request_view = [{"role": "user", "content": "a later, different view"}] + await _await_pending_task(context) + + assert _digest(provider.calls[0].messages[:-1]) == _digest(parent_view) + + +@pytest.mark.asyncio +async def test_concurrent_forks_are_still_serialized_by_the_in_flight_guard(): + """Nothing about forking may weaken the single-summarization-in-flight + invariant.""" + context = _summary_manager(summary_call_mode="fork") + await _fill(context) + context.note_request_sent(tools=_tools()) + await _arm_below_trigger(context) + _cross_trigger(context) + + provider = _FakeProvider() + await asyncio.gather( + context.get_messages_for_request(provider=provider), + context.get_messages_for_request(provider=provider), + context.get_messages_for_request(provider=provider), + ) + task = context._summarization_task + if task is not None: + await task + + assert len(provider.calls) == 1 + assert context._pending_summary is not None + assert context.last_summary_call_stats["mode_used"] == "fork"