feat(experiments): DeepSeek stage-1 compaction arms + prefix-preserving summarization

The compaction-experiment middleware stack behind evals.md F35-F38:

- StableToolOutputCapMiddleware: one persistent 40k-byte (nominal 10k-token)
cap when a tool output enters the checkpoint, so every later model call and
the summarizer read identical stable text (prefix-cache friendly), with
capped/original/retained-bytes telemetry and a sha256 audit trail.
- InstrumentedSummarizationMiddleware: trigger-evidence + summary-cost
telemetry (pre/post tokens, summary input/output, retry reasons) on the
stock XML summarization path.
- PrefixPreservingCompactionMiddleware (summarization_strategy=
"structured_prefix"): generates the summary by sending the unchanged
request prefix + one checkpoint instruction with the same model/tools/
settings bound, so the summary call rides the provider cache (94% vs 0%
cache hit on DeepSeek; -14%/trajectory vs the XML arm at identical
compaction dose). Verified analog of Codex's local compaction.
- DeepSeekCacheIsolationMiddleware: per arm/session/trial user_id injection
+ a request-size guard for experiment runs.
- Presets exp_fh_raw / exp_fh_cap10k / exp_c200_raw / exp_c200_cap10k /
exp_c200_cap10k_structured (+ stage-2 exp_c400/c800, built, unrun) in
memory_presets; per-call model_calls usage/cost telemetry with
summarization cost breakdown in telemetry.
- Tests: cap persistence end-to-end, structured retry/tool-call handling,
XML-vs-structured boundary equivalence, multi-compaction, the
no-double-count usage invariant, the langchain-openai cached_tokens
mapping contract, and KB tool-call argument robustness.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

app/chat_service.py

@@ -14,11 +14,13 @@ from threading import Lock
 from typing import Any, AsyncIterator
 from uuid import uuid4
 
+import httpx
 from langchain.agents import create_agent
 from langchain.agents.middleware import (
     AgentMiddleware,
     ClearToolUsesEdit,
     ContextEditingMiddleware,
+    ExtendedModelResponse,
     SummarizationMiddleware,
 )
 from langchain.tools import ToolRuntime, tool
@@ -27,11 +29,16 @@ from langchain_core.messages import (
     AIMessageChunk,
     BaseMessage,
     HumanMessage,
+    RemoveMessage,
     SystemMessage,
+    ToolMessage,
 )
+from langchain_core.messages.utils import count_tokens_approximately, get_buffer_string
 from langchain_openai import ChatOpenAI
 from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
 from langgraph.store.memory import InMemoryStore
+from langgraph.types import Command
 
 from .chat_types import ChatEvent, ChatRequest, ChatTurn, SourceMatch
 from .memory_presets import (
@@ -42,9 +49,13 @@ from .memory_presets import (
 )
 from .telemetry import (
     TurnUsageHandler,
+    aggregate_cost_breakdown,
     context_window_stats,
     estimate_cost_usd,
+    pop_turn_events,
     pop_turn_signals,
+    record_turn_event,
+    record_turn_signal,
     record_turn_signal_max,
     reset_turn_signals,
     usage_totals,
@@ -138,6 +149,9 @@ class AppContext:
     kb_session_id: str = ""
     kb_command_limit: int = DEFAULT_KB_COMMAND_LIMIT
     student_id: str = ""
+    # DeepSeek experiment-only cache namespace. Stable within one trajectory,
+    # distinct across arm/session/trial, and intentionally contains no PII.
+    cache_user_id: str = ""
     # Per-request retrieval token budget (Part C / Axis B sweep); None keeps the
     # retriever's DEFAULT_CONTEXT_TOKEN_BUDGET.
     retrieval_budget: int | None = None
@@ -347,12 +361,24 @@ RETRIEVE_TUTOR_CONTEXT_SCHEMA = {
         },
     },
     "required": ["query"],
+    "additionalProperties": False,
 }
 
 
 @tool(args_schema=RETRIEVE_TUTOR_CONTEXT_SCHEMA)
-def retrieve_tutor_context(query: str, runtime: ToolRuntime[AppContext]) -> str:
+def retrieve_tutor_context(
+    query: str, runtime: ToolRuntime[AppContext], **unsupported: Any
+) -> str:
     """Retrieve relevant course and documentation context for an AI tutor question."""
+    if unsupported:
+        names = ", ".join(sorted(unsupported))
+        logger.warning(
+            "retrieve_tutor_context received unsupported arguments: %s", names
+        )
+        return (
+            "retrieve_tutor_context could not run because it received unsupported "
+            f"argument(s): {names}. Retry the tool with only the query argument."
+        )
     try:
         results = select_retriever(
             getattr(runtime.context, "retriever_kind", "")
@@ -391,14 +417,28 @@ RUN_KB_COMMAND_SCHEMA = {
             "type": "integer",
             "description": "Command timeout in seconds, capped by the runtime.",
             "default": 8,
+            "minimum": 1,
+            "maximum": 30,
+        },
+        "timeout": {
+            "type": "integer",
+            "description": (
+                "Alias for timeout_seconds. The runtime still caps the command "
+                "at 30 seconds."
+            ),
+            "minimum": 1,
+            "maximum": 30,
         },
         "max_output_chars": {
             "type": "integer",
             "description": "Maximum stdout/stderr characters to return, capped by the runtime.",
             "default": 40000,
+            "minimum": 1000,
+            "maximum": 80000,
         },
     },
     "required": ["command"],
+    "additionalProperties": False,
 }
 
 
@@ -406,10 +446,23 @@ RUN_KB_COMMAND_SCHEMA = {
 def run_kb_command(
     command: str,
     runtime: ToolRuntime[AppContext],
-    timeout_seconds: int = 8,
+    timeout_seconds: int | None = None,
     max_output_chars: int = 40000,
+    timeout: int | None = None,
+    **unsupported: Any,
 ) -> str:
     """Run a safe, read-only terminal-style command inside the local KB."""
+    if unsupported:
+        names = ", ".join(sorted(unsupported))
+        logger.warning("run_kb_command received unsupported arguments: %s", names)
+        return (
+            f"$ {command}\n"
+            f"error: unsupported run_kb_command argument(s): {names}. "
+            "Use command, timeout_seconds (or timeout), and max_output_chars."
+        )
+    effective_timeout = timeout_seconds if timeout_seconds is not None else timeout
+    if effective_timeout is None:
+        effective_timeout = 8
     allowed, used = _claim_kb_command_budget(
         runtime.context.kb_session_id,
         runtime.context.kb_command_limit,
@@ -428,11 +481,11 @@ def run_kb_command(
     try:
         result = execute_kb_command(
             command,
-            timeout_seconds=timeout_seconds,
+            timeout_seconds=effective_timeout,
             max_output_chars=max_output_chars,
         )
         return format_command_payload(result)
-    except (KbCommandError, OSError) as exc:
+    except (KbCommandError, OSError, TypeError, ValueError) as exc:
         return f"$ {command}\nerror: {exc}"
 
 
@@ -1200,6 +1253,674 @@ def _turn_id_for(request: Any) -> str:
     return getattr(ctx, "kb_session_id", "") if ctx else ""
 
 
+def _cache_user_id_for_runtime(runtime: Any) -> str:
+    ctx = getattr(runtime, "context", None) if runtime else None
+    return str(getattr(ctx, "cache_user_id", "") or "") if ctx else ""
+
+
+class DeepSeekCacheIsolationMiddleware(AgentMiddleware):
+    """Attach DeepSeek ``user_id`` and guard the experimental request size."""
+
+    def __init__(self, max_request_tokens: int | None = None) -> None:
+        super().__init__()
+        self.max_request_tokens = max_request_tokens
+
+    def _isolate(self, request: Any) -> Any:
+        request_messages = list(getattr(request, "messages", None) or [])
+        system_message = getattr(request, "system_message", None)
+        if system_message is not None:
+            request_messages.insert(0, system_message)
+        request_tokens = int(count_tokens_approximately(request_messages))
+        if (
+            self.max_request_tokens is not None
+            and request_tokens > self.max_request_tokens
+        ):
+            raise RuntimeError(
+                "Agent request exceeds the experiment safety guard: "
+                f"{request_tokens:,} > {self.max_request_tokens:,} "
+                "approximate tokens."
+            )
+        user_id = _cache_user_id_for_runtime(getattr(request, "runtime", None))
+        if not user_id:
+            return request
+        settings = dict(getattr(request, "model_settings", None) or {})
+        extra_body = dict(settings.get("extra_body") or {})
+        extra_body["user_id"] = user_id
+        settings["extra_body"] = extra_body
+        return request.override(model_settings=settings)
+
+    def wrap_model_call(self, request, handler):
+        return handler(self._isolate(request))
+
+    async def awrap_model_call(self, request, handler):
+        return await handler(self._isolate(request))
+
+
+class StableToolOutputCapMiddleware(AgentMiddleware):
+    """Persistently cap tool output once, when it first enters agent history."""
+
+    def __init__(self, max_bytes: int) -> None:
+        super().__init__()
+        self.max_bytes = max(1_024, int(max_bytes))
+
+    @staticmethod
+    def _decode_fragment(fragment: bytes) -> str:
+        return fragment.decode("utf-8", errors="ignore")
+
+    def _cap(self, request: Any, result: Any) -> Any:
+        if not isinstance(result, ToolMessage):
+            return result
+        text = message_content_to_text(result.content)
+        raw = text.encode("utf-8")
+        if len(raw) <= self.max_bytes:
+            return result
+
+        marker = (
+            f"\n\n[... tool output truncated at stable {self.max_bytes}-byte cap; "
+            "middle omitted ...]\n\n"
+        ).encode("utf-8")
+        payload_budget = max(0, self.max_bytes - len(marker))
+        head_bytes = payload_budget // 2
+        tail_bytes = payload_budget - head_bytes
+        capped = (
+            self._decode_fragment(raw[:head_bytes])
+            + marker.decode("utf-8")
+            + self._decode_fragment(raw[-tail_bytes:] if tail_bytes else b"")
+        )
+        capped_bytes = len(capped.encode("utf-8"))
+        metadata = {
+            "original_bytes": len(raw),
+            "original_chars": len(text),
+            "retained_bytes": capped_bytes,
+            "sha256": hashlib.sha256(raw).hexdigest(),
+            "max_bytes": self.max_bytes,
+        }
+        turn = _turn_id_for(request)
+        record_turn_signal(turn, "tool_outputs_capped", 1)
+        record_turn_signal(turn, "tool_output_original_bytes", len(raw))
+        record_turn_signal(turn, "tool_output_retained_bytes", capped_bytes)
+        additional = dict(result.additional_kwargs or {})
+        additional["stable_tool_cap"] = metadata
+        return result.model_copy(
+            update={"content": capped, "additional_kwargs": additional}
+        )
+
+    def wrap_tool_call(self, request, handler):
+        return self._cap(request, handler(request))
+
+    async def awrap_tool_call(self, request, handler):
+        return self._cap(request, await handler(request))
+
+
+class InstrumentedSummarizationMiddleware(SummarizationMiddleware):
+    """SummarizationMiddleware with full event telemetry and loud failures."""
+
+    MAX_SUMMARY_ATTEMPTS = 3
+    RETRY_BASE_DELAY_SECONDS = 1.0
+
+    def __init__(
+        self,
+        *args: Any,
+        summary_input_guard_tokens: int | None = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.summary_input_guard_tokens = summary_input_guard_tokens
+
+    def _configured_token_trigger(self) -> int | None:
+        return next(
+            (
+                int(clause["tokens"])
+                for clause in self._trigger_clauses
+                if "tokens" in clause
+            ),
+            None,
+        )
+
+    @staticmethod
+    def _last_ai_reported_tokens(messages: list[Any]) -> int:
+        last_ai_message = next(
+            (
+                message
+                for message in reversed(messages)
+                if isinstance(message, AIMessage)
+            ),
+            None,
+        )
+        usage = getattr(last_ai_message, "usage_metadata", None) or {}
+        return int(usage.get("total_tokens") or 0)
+
+    def _token_trigger_evidence(
+        self, messages: list[Any], approximate_tokens: int
+    ) -> tuple[str, int]:
+        trigger_tokens = self._configured_token_trigger()
+        if trigger_tokens is None:
+            return "non_token", 0
+        approximate_met = approximate_tokens >= trigger_tokens
+        provider_reported_met = self._should_summarize_based_on_reported_tokens(
+            messages, float(trigger_tokens)
+        )
+        reported_tokens = self._last_ai_reported_tokens(messages)
+        if approximate_met and provider_reported_met:
+            return "approximate_and_provider_reported", reported_tokens
+        if approximate_met:
+            return "approximate", reported_tokens
+        if provider_reported_met:
+            return "provider_reported", reported_tokens
+        return "other", reported_tokens
+
+    def _plan_compaction(self, state: Any) -> dict[str, Any] | None:
+        messages = state["messages"]
+        self._ensure_message_ids(messages)
+        total_tokens = int(self.token_counter(messages))
+        if not self._should_summarize(messages, total_tokens):
+            return None
+        trigger_source, trigger_reported_tokens = self._token_trigger_evidence(
+            messages, total_tokens
+        )
+        cutoff_index = self._determine_cutoff_index(messages)
+        if cutoff_index <= 0:
+            return None
+        selected, preserved = self._partition_messages(messages, cutoff_index)
+        trimmed = self._trim_messages_for_summary(selected)
+        if not trimmed:
+            raise RuntimeError("Summarization selected no usable input messages.")
+        summary_input_tokens = int(self._partial_token_counter(trimmed))
+        if (
+            self.summary_input_guard_tokens is not None
+            and summary_input_tokens > self.summary_input_guard_tokens
+        ):
+            raise RuntimeError(
+                "Summarization input exceeds the experiment safety guard: "
+                f"{summary_input_tokens:,} > "
+                f"{self.summary_input_guard_tokens:,} approximate tokens."
+            )
+        return {
+            "messages": messages,
+            "pre_tokens": total_tokens,
+            "trigger_source": trigger_source,
+            "trigger_reported_tokens": trigger_reported_tokens,
+            "selected": selected,
+            "preserved": preserved,
+            "trimmed": trimmed,
+            "summary_input_tokens": summary_input_tokens,
+        }
+
+    def _summary_prompt_text(self, trimmed: list[Any]) -> str:
+        formatted = get_buffer_string(trimmed, format="xml")
+        return self.summary_prompt.format(messages=formatted).rstrip()
+
+    def _summary_model(self, runtime: Any) -> Any:
+        user_id = _cache_user_id_for_runtime(runtime)
+        if not user_id:
+            return self.model
+        return self.model.bind(extra_body={"user_id": user_id})
+
+    @staticmethod
+    def _is_retryable_exception(exc: BaseException) -> bool:
+        if isinstance(exc, (TimeoutError, ConnectionError, httpx.TransportError)):
+            return True
+        status = getattr(exc, "status_code", None)
+        if status is None:
+            response = getattr(exc, "response", None)
+            status = getattr(response, "status_code", None)
+        if isinstance(status, int):
+            return status in {408, 409, 425, 429} or status >= 500
+        return type(exc).__name__ in {
+            "APIConnectionError",
+            "APITimeoutError",
+            "InternalServerError",
+            "RateLimitError",
+        }
+
+    @staticmethod
+    def _retry_reason(exc: BaseException) -> str:
+        detail = str(exc).strip()
+        return f"{type(exc).__name__}: {detail}" if detail else type(exc).__name__
+
+    @classmethod
+    def _retry_delay(cls, attempt: int) -> float:
+        return cls.RETRY_BASE_DELAY_SECONDS * (2 ** (attempt - 1))
+
+    def _log_summary_retry(
+        self, runtime: Any, attempt: int, reason: str, delay: float
+    ) -> None:
+        ctx = getattr(runtime, "context", None)
+        message_id = str(getattr(ctx, "kb_session_id", "") or "") if ctx else ""
+        logger.warning(
+            "Retrying summarization after attempt %d/%d in %.1fs. "
+            "message_id=%s reason=%s",
+            attempt,
+            self.MAX_SUMMARY_ATTEMPTS,
+            delay,
+            message_id,
+            reason,
+        )
+
+    def _record_compaction(
+        self, runtime: Any, plan: dict[str, Any], summary: str
+    ) -> dict[str, Any]:
+        if not summary.strip():
+            raise RuntimeError("Summarization returned an empty summary.")
+        if summary.startswith("Error generating summary:"):
+            raise RuntimeError(summary)
+        new_messages = self._build_new_messages(summary)
+        preserved = plan["preserved"]
+        turn = _cache_user_id_for_runtime(runtime)
+        ctx = getattr(runtime, "context", None)
+        message_id = str(getattr(ctx, "kb_session_id", "") or "") if ctx else ""
+        event = {
+            "event": "summarization",
+            "summary_strategy": str(plan.get("summary_strategy") or "xml"),
+            "configured_trigger_tokens": self._configured_token_trigger(),
+            "trigger_source": plan["trigger_source"],
+            "trigger_reported_tokens": int(plan["trigger_reported_tokens"]),
+            "pre_compaction_tokens_approx": int(plan["pre_tokens"]),
+            "pre_compaction_messages": len(plan["messages"]),
+            "selected_messages": len(plan["selected"]),
+            "selected_tokens_approx": int(
+                self._partial_token_counter(plan["selected"])
+            ),
+            "summary_input_messages": len(plan["trimmed"]),
+            "summary_input_tokens_approx": int(plan["summary_input_tokens"]),
+            "summary_input_untrimmed": self.trim_tokens_to_summarize is None
+            and len(plan["trimmed"]) == len(plan["selected"]),
+            "summary_attempts": int(plan.get("summary_attempts") or 1),
+            "summary_retry_reasons": list(plan.get("summary_retry_reasons") or []),
+            "summary_output_tokens_approx": int(
+                count_tokens_approximately([HumanMessage(content=summary)])
+            ),
+            "retained_tail_messages": len(preserved),
+            "retained_tail_tokens_approx": int(self._partial_token_counter(preserved)),
+            "post_compaction_tokens_approx": int(
+                self._partial_token_counter([*new_messages, *preserved])
+            ),
+            "cache_user_id_present": bool(turn),
+        }
+        event.update(plan.get("summary_request_telemetry") or {})
+        event.update(plan.get("summary_provider_telemetry") or {})
+        record_turn_signal(message_id, "compactions_this_turn", 1)
+        record_turn_signal_max(
+            message_id,
+            "max_pre_compaction_tokens_approx",
+            int(plan["pre_tokens"]),
+        )
+        record_turn_event(message_id, event)
+        return {
+            "messages": [
+                RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                *new_messages,
+                *preserved,
+            ]
+        }
+
+    def before_model(self, state, runtime):
+        plan = self._plan_compaction(state)
+        if plan is None:
+            return None
+        model = self._summary_model(runtime)
+        prompt = self._summary_prompt_text(plan["trimmed"])
+        retry_reasons: list[str] = []
+        for attempt in range(1, self.MAX_SUMMARY_ATTEMPTS + 1):
+            try:
+                response = model.invoke(
+                    prompt,
+                    config={"metadata": {"lc_source": "summarization"}},
+                )
+                plan["summary_provider_telemetry"] = self._summary_provider_telemetry(
+                    response
+                )
+                summary = message_content_to_text(response.content).strip()
+            except Exception as exc:
+                if (
+                    attempt >= self.MAX_SUMMARY_ATTEMPTS
+                    or not self._is_retryable_exception(exc)
+                ):
+                    raise
+                reason = self._retry_reason(exc)
+            else:
+                if summary:
+                    plan["summary_attempts"] = attempt
+                    plan["summary_retry_reasons"] = retry_reasons
+                    return self._record_compaction(runtime, plan, summary)
+                if attempt >= self.MAX_SUMMARY_ATTEMPTS:
+                    raise RuntimeError(
+                        "Summarization returned an empty summary after "
+                        f"{attempt} attempts."
+                    )
+                reason = "empty response"
+            retry_reasons.append(reason)
+            delay = self._retry_delay(attempt)
+            self._log_summary_retry(runtime, attempt, reason, delay)
+            time.sleep(delay)
+        raise AssertionError("unreachable")
+
+    async def abefore_model(self, state, runtime):
+        plan = self._plan_compaction(state)
+        if plan is None:
+            return None
+        model = self._summary_model(runtime)
+        prompt = self._summary_prompt_text(plan["trimmed"])
+        retry_reasons: list[str] = []
+        for attempt in range(1, self.MAX_SUMMARY_ATTEMPTS + 1):
+            try:
+                response = await model.ainvoke(
+                    prompt,
+                    config={"metadata": {"lc_source": "summarization"}},
+                )
+                plan["summary_provider_telemetry"] = self._summary_provider_telemetry(
+                    response
+                )
+                summary = message_content_to_text(response.content).strip()
+            except Exception as exc:
+                if (
+                    attempt >= self.MAX_SUMMARY_ATTEMPTS
+                    or not self._is_retryable_exception(exc)
+                ):
+                    raise
+                reason = self._retry_reason(exc)
+            else:
+                if summary:
+                    plan["summary_attempts"] = attempt
+                    plan["summary_retry_reasons"] = retry_reasons
+                    return self._record_compaction(runtime, plan, summary)
+                if attempt >= self.MAX_SUMMARY_ATTEMPTS:
+                    raise RuntimeError(
+                        "Summarization returned an empty summary after "
+                        f"{attempt} attempts."
+                    )
+                reason = "empty response"
+            retry_reasons.append(reason)
+            delay = self._retry_delay(attempt)
+            self._log_summary_retry(runtime, attempt, reason, delay)
+            await asyncio.sleep(delay)
+        raise AssertionError("unreachable")
+
+    @staticmethod
+    def _summary_provider_telemetry(response: Any) -> dict[str, Any]:
+        """Expose provider cache accounting on the compaction event itself."""
+        usage = dict(getattr(response, "usage_metadata", None) or {})
+        details = dict(usage.get("input_token_details") or {})
+        input_tokens = int(usage.get("input_tokens") or 0)
+        cache_read = int(details.get("cache_read") or 0)
+        cache_creation = int(details.get("cache_creation") or 0)
+        cache_miss = max(0, input_tokens - cache_read - cache_creation)
+        return {
+            "summary_provider_usage_reported": bool(usage),
+            "summary_provider_cache_details_reported": "cache_read" in details,
+            "summary_provider_input_tokens": input_tokens,
+            "summary_provider_cache_read_tokens": cache_read,
+            "summary_provider_cache_creation_tokens": cache_creation,
+            "summary_provider_cache_miss_tokens": cache_miss,
+            "summary_provider_output_tokens": int(usage.get("output_tokens") or 0),
+            "summary_provider_cache_hit_ratio": (
+                cache_read / input_tokens if input_tokens else None
+            ),
+        }
+
+
+PREFIX_PRESERVING_COMPACTION_PROMPT = """Create a durable checkpoint summary of the older conversation prefix above.
+
+The first {selected_messages} conversation messages will be replaced by your checkpoint. The final {retained_messages} messages (approximately {retained_tokens} tokens) will remain verbatim. Summarize only the older prefix; do not duplicate the retained tail except where a short reference is necessary to explain a dependency between old and recent work.
+
+Preserve concrete facts, current values after corrections, decisions, constraints, unresolved work, prior tool evidence needed later, and enough causal detail to continue without re-running tools. Omit transient chatter and redundant wording. Never call a tool and do not answer the conversation's latest question.
+
+Return only the checkpoint summary."""
+
+
+class PrefixPreservingCompactionMiddleware(InstrumentedSummarizationMiddleware):
+    """Compact through a structured prefix-extension request.
+
+    LangChain's stock summarizer serializes selected messages into a new XML
+    prompt, so even the summary-generation call loses the provider's cached
+    prefix. This middleware instead runs after request-shaping middleware and
+    binds the exact same model, tool schemas, tool choice, model settings, and
+    system message. It then sends the unchanged *entire* current request prefix
+    followed by one checkpoint instruction. The resulting checkpoint summary
+    may overlap the retained tail; that conservative duplication matches the
+    cache-friendly local Codex pattern and reduces the chance of lost evidence.
+
+    Installing the resulting summary necessarily changes the *next* agent
+    prefix; no client-side middleware can avoid that boundary without a
+    provider-native opaque continuation/compaction primitive.
+    """
+
+    def before_model(self, state, runtime):
+        # Planning must happen after all request-shaping middleware has produced
+        # the final ModelRequest, so wrap_model_call owns the operation.
+        return None
+
+    async def abefore_model(self, state, runtime):
+        return None
+
+    def _checkpoint_instruction(self, plan: dict[str, Any]) -> HumanMessage:
+        return HumanMessage(
+            content=PREFIX_PRESERVING_COMPACTION_PROMPT.format(
+                selected_messages=len(plan["selected"]),
+                retained_messages=len(plan["preserved"]),
+                retained_tokens=int(self._partial_token_counter(plan["preserved"])),
+            )
+        )
+
+    def _summary_messages(
+        self, request: Any, plan: dict[str, Any]
+    ) -> list[BaseMessage]:
+        messages: list[BaseMessage] = []
+        if request.system_message is not None:
+            messages.append(request.system_message)
+        messages.extend(request.messages)
+        messages.append(self._checkpoint_instruction(plan))
+        return messages
+
+    def _prepare_summary_request(
+        self, request: Any, plan: dict[str, Any]
+    ) -> tuple[Any, list[BaseMessage]]:
+        if request.response_format is not None:
+            raise RuntimeError(
+                "Structured-prefix compaction does not support a structured "
+                "agent response format."
+            )
+        if list(request.messages) != list(plan["messages"]):
+            raise RuntimeError(
+                "Structured-prefix compaction requires the finalized model view "
+                "to match checkpoint history exactly; an earlier middleware "
+                "changed request.messages without persisting that change."
+            )
+        messages = self._summary_messages(request, plan)
+        request_tokens = int(count_tokens_approximately(messages))
+        if (
+            self.summary_input_guard_tokens is not None
+            and request_tokens > self.summary_input_guard_tokens
+        ):
+            raise RuntimeError(
+                "Structured-prefix summary request exceeds the experiment safety "
+                f"guard: {request_tokens:,} > "
+                f"{self.summary_input_guard_tokens:,} approximate tokens."
+            )
+
+        settings = dict(request.model_settings or {})
+        if request.tools:
+            summary_model = request.model.bind_tools(
+                request.tools,
+                tool_choice=request.tool_choice,
+                **settings,
+            )
+        else:
+            summary_model = request.model.bind(**settings)
+
+        system_tokens = int(
+            count_tokens_approximately([request.system_message])
+            if request.system_message is not None
+            else 0
+        )
+        instruction_tokens = int(count_tokens_approximately([messages[-1]]))
+        plan["summary_strategy"] = "structured_prefix"
+        plan["summary_request_telemetry"] = {
+            "summary_prefix_messages": len(request.messages),
+            "summary_prefix_tokens_approx": int(
+                count_tokens_approximately(request.messages)
+            ),
+            "summary_selected_messages": len(plan["trimmed"]),
+            "summary_selected_tokens_approx": int(plan["summary_input_tokens"]),
+            "summary_request_messages": len(messages),
+            "summary_request_tokens_approx": request_tokens,
+            "summary_request_is_strict_extension": True,
+            "summary_instruction_selected_messages": len(plan["selected"]),
+            "summary_instruction_retained_messages": len(plan["preserved"]),
+            "summary_instruction_retained_tokens_approx": int(
+                self._partial_token_counter(plan["preserved"])
+            ),
+            "summary_system_tokens_approx": system_tokens,
+            "summary_instruction_tokens_approx": instruction_tokens,
+            "summary_system_message_present": request.system_message is not None,
+            "summary_tools_bound": len(request.tools or []),
+            "summary_tool_choice_preserved": True,
+            "summary_tool_choice_value": (
+                str(request.tool_choice) if request.tool_choice is not None else None
+            ),
+            "summary_model_settings_keys": sorted(settings),
+            "summary_cache_user_id_preserved": bool(
+                (settings.get("extra_body") or {}).get("user_id")
+            ),
+        }
+        return summary_model, messages
+
+    def _invoke_prefix_summary(self, request: Any, plan: dict[str, Any]) -> str:
+        model, messages = self._prepare_summary_request(request, plan)
+        retry_reasons: list[str] = []
+        for attempt in range(1, self.MAX_SUMMARY_ATTEMPTS + 1):
+            try:
+                response = model.invoke(
+                    messages,
+                    config={
+                        "metadata": {
+                            "lc_source": "summarization",
+                            "compaction_strategy": "structured_prefix",
+                        }
+                    },
+                )
+                plan["summary_provider_telemetry"] = self._summary_provider_telemetry(
+                    response
+                )
+                summary = message_content_to_text(response.content).strip()
+            except Exception as exc:
+                if (
+                    attempt >= self.MAX_SUMMARY_ATTEMPTS
+                    or not self._is_retryable_exception(exc)
+                ):
+                    raise
+                reason = self._retry_reason(exc)
+            else:
+                if summary:
+                    plan["summary_attempts"] = attempt
+                    plan["summary_retry_reasons"] = retry_reasons
+                    return summary
+                if attempt >= self.MAX_SUMMARY_ATTEMPTS:
+                    raise RuntimeError(
+                        "Structured-prefix summarization returned an empty summary "
+                        f"after {attempt} attempts."
+                    )
+                reason = "empty response"
+            retry_reasons.append(reason)
+            delay = self._retry_delay(attempt)
+            self._log_summary_retry(request.runtime, attempt, reason, delay)
+            time.sleep(delay)
+        raise AssertionError("unreachable")
+
+    async def _ainvoke_prefix_summary(self, request: Any, plan: dict[str, Any]) -> str:
+        model, messages = self._prepare_summary_request(request, plan)
+        retry_reasons: list[str] = []
+        for attempt in range(1, self.MAX_SUMMARY_ATTEMPTS + 1):
+            try:
+                response = await model.ainvoke(
+                    messages,
+                    config={
+                        "metadata": {
+                            "lc_source": "summarization",
+                            "compaction_strategy": "structured_prefix",
+                        }
+                    },
+                )
+                plan["summary_provider_telemetry"] = self._summary_provider_telemetry(
+                    response
+                )
+                summary = message_content_to_text(response.content).strip()
+            except Exception as exc:
+                if (
+                    attempt >= self.MAX_SUMMARY_ATTEMPTS
+                    or not self._is_retryable_exception(exc)
+                ):
+                    raise
+                reason = self._retry_reason(exc)
+            else:
+                if summary:
+                    plan["summary_attempts"] = attempt
+                    plan["summary_retry_reasons"] = retry_reasons
+                    return summary
+                if attempt >= self.MAX_SUMMARY_ATTEMPTS:
+                    raise RuntimeError(
+                        "Structured-prefix summarization returned an empty summary "
+                        f"after {attempt} attempts."
+                    )
+                reason = "empty response"
+            retry_reasons.append(reason)
+            delay = self._retry_delay(attempt)
+            self._log_summary_retry(request.runtime, attempt, reason, delay)
+            await asyncio.sleep(delay)
+        raise AssertionError("unreachable")
+
+    def _compacted_request_and_command(
+        self, request: Any, plan: dict[str, Any], summary: str
+    ) -> tuple[Any, list[BaseMessage]]:
+        update = self._record_compaction(request.runtime, plan, summary)
+        compacted = list(update["messages"])[1:]
+        compacted_state = dict(request.state)
+        compacted_state["messages"] = compacted
+        return request.override(messages=compacted, state=compacted_state), compacted
+
+    @staticmethod
+    def _with_checkpoint_command(
+        response: Any, compacted: list[BaseMessage]
+    ) -> ExtendedModelResponse:
+        # The model response command is applied first by create_agent. The
+        # additional command then replaces the old checkpoint and re-adds that
+        # same response, so it survives REMOVE_ALL_MESSAGES exactly once.
+        return ExtendedModelResponse(
+            model_response=response,
+            command=Command(
+                update={
+                    "messages": [
+                        RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                        *compacted,
+                        *response.result,
+                    ]
+                }
+            ),
+        )
+
+    def wrap_model_call(self, request, handler):
+        plan = self._plan_compaction(request.state)
+        if plan is None:
+            return handler(request)
+        summary = self._invoke_prefix_summary(request, plan)
+        compacted_request, compacted = self._compacted_request_and_command(
+            request, plan, summary
+        )
+        response = handler(compacted_request)
+        return self._with_checkpoint_command(response, compacted)
+
+    async def awrap_model_call(self, request, handler):
+        plan = self._plan_compaction(request.state)
+        if plan is None:
+            return await handler(request)
+        summary = await self._ainvoke_prefix_summary(request, plan)
+        compacted_request, compacted = self._compacted_request_and_command(
+            request, plan, summary
+        )
+        response = await handler(compacted_request)
+        return self._with_checkpoint_command(response, compacted)
+
+
 class SlidingWindowMiddleware(AgentMiddleware):
     """Keep only the last N messages in the model's view; drop older ones.
 
@@ -1572,7 +2293,29 @@ def build_agent_middleware(
     model: Any, memory_config: MemoryConfig
 ) -> list[AgentMiddleware]:
     """Assemble the compaction/memory middleware stack for one preset."""
+    if memory_config.summarization_strategy not in {"xml", "structured_prefix"}:
+        raise ValueError(
+            f"Unknown summarization strategy: {memory_config.summarization_strategy!r}"
+        )
+    if (
+        memory_config.summarization_strategy == "structured_prefix"
+        and not memory_config.summarization
+    ):
+        raise ValueError(
+            "structured_prefix summarization strategy requires summarization=True"
+        )
     middleware: list[AgentMiddleware] = []
+    prefix_compactor: PrefixPreservingCompactionMiddleware | None = None
+    if memory_config.experiment_mode:
+        middleware.append(
+            DeepSeekCacheIsolationMiddleware(
+                memory_config.experiment_request_guard_tokens
+            )
+        )
+    if memory_config.tool_output_cap_bytes is not None:
+        middleware.append(
+            StableToolOutputCapMiddleware(memory_config.tool_output_cap_bytes)
+        )
     if memory_config.context_editing:
         middleware.append(
             ContextEditingMiddleware(
@@ -1596,16 +2339,41 @@ def build_agent_middleware(
             )
         )
     if memory_config.summarization:
+        keep: tuple[str, int]
+        if memory_config.summarization_keep_tokens is not None:
+            keep = ("tokens", memory_config.summarization_keep_tokens)
+        else:
+            keep = ("messages", memory_config.summarization_keep_messages)
         summarization_kwargs: dict[str, Any] = {
             "model": model,
             "trigger": ("tokens", memory_config.summarization_trigger_tokens),
-            "keep": ("messages", memory_config.summarization_keep_messages),
+            "keep": keep,
+            "trim_tokens_to_summarize": memory_config.summarization_trim_tokens,
         }
         # A custom summary prompt (selective_retention / context_reset) overrides
         # the library default; None keeps it.
         if memory_config.summary_prompt:
             summarization_kwargs["summary_prompt"] = memory_config.summary_prompt
-        middleware.append(SummarizationMiddleware(**summarization_kwargs))
+        if memory_config.experiment_mode:
+            summarization_kwargs["summary_input_guard_tokens"] = (
+                memory_config.summarization_input_guard_tokens
+            )
+        if memory_config.summarization_strategy == "structured_prefix":
+            if not memory_config.experiment_mode:
+                raise ValueError(
+                    "structured_prefix summarization is restricted to "
+                    "experiment-mode presets"
+                )
+            prefix_compactor = PrefixPreservingCompactionMiddleware(
+                **summarization_kwargs
+            )
+        else:
+            summary_middleware = (
+                InstrumentedSummarizationMiddleware
+                if memory_config.experiment_mode
+                else SummarizationMiddleware
+            )
+            middleware.append(summary_middleware(**summarization_kwargs))
     # Part C per-call-view mechanisms (each preset enables at most one). They
     # reshape only the request, not the checkpoint, and report via the
     # turn-signal registry.
@@ -1640,6 +2408,11 @@ def build_agent_middleware(
     if memory_config.longterm_memory:
         middleware.append(StudentProfileMiddleware())
     middleware.append(SourcePreferenceMiddleware())
+    if prefix_compactor is not None:
+        # wrap_model_call middleware compose left-to-right (first is outermost).
+        # Keep this last so it observes the finalized system message from every
+        # preceding request-shaping middleware before issuing the prefix call.
+        middleware.append(prefix_compactor)
     return middleware
 
 
@@ -1835,6 +2608,7 @@ def agent_run_config(
                 "include_reasoning": bool(request.include_reasoning),
                 "memory_preset": preset,
                 "student_id": request.student_id,
+                "cache_user_id": request.cache_user_id,
             },
         }
     )
@@ -1984,6 +2758,7 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
                 kb_session_id=message_id,
                 kb_command_limit=DEFAULT_KB_COMMAND_LIMIT,
                 student_id=request.student_id,
+                cache_user_id=request.cache_user_id,
                 retrieval_budget=request.retrieval_budget,
                 retriever_kind=request.retriever,
             ),
@@ -2078,6 +2853,12 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
                 # ToolMessage and must not re-emit it as new tool activity.
                 if step == "tools" and getattr(message, "type", None) == "tool":
                     payload = message_content_to_text(message.content)
+                    cap_metadata = dict(
+                        (getattr(message, "additional_kwargs", None) or {}).get(
+                            "stable_tool_cap"
+                        )
+                        or {}
+                    )
                     tool_call_id = str(
                         getattr(message, "tool_call_id", "") or uuid4().hex
                     )
@@ -2114,6 +2895,11 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
                             "args": tool_call.get("args"),
                             "args_text": format_tool_args(tool_call.get("args")),
                             "output_text": payload,
+                            "output_was_capped": bool(cap_metadata),
+                            "output_original_bytes": cap_metadata.get("original_bytes"),
+                            "output_original_chars": cap_metadata.get("original_chars"),
+                            "output_retained_bytes": cap_metadata.get("retained_bytes"),
+                            "output_sha256": cap_metadata.get("sha256"),
                             "matches": [
                                 source_match_payload(
                                     match,
@@ -2282,6 +3068,15 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
         "messages", []
     )
     totals = usage_totals(usage_handler.usage_metadata)
+    cost_breakdown = aggregate_cost_breakdown(usage_handler.usage_metadata)
+    model_calls = list(usage_handler.model_calls)
+    summarization_cost = sum(
+        float((call.get("cost") or {}).get("total_usd") or 0)
+        for call in model_calls
+        if call.get("source") == "summarization"
+    )
+    turn_events = pop_turn_events(message_id)
+    turn_signals = pop_turn_signals(message_id)
     total_ms = int((time.monotonic() - turn_started) * 1000)
     yield ChatEvent(
         "context_stats",
@@ -2296,8 +3091,25 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
                 model_key: dict(usage)
                 for model_key, usage in usage_handler.usage_metadata.items()
             },
+            "model_calls": model_calls,
             **totals,
+            "cache_miss_tokens": max(
+                0,
+                totals["input_tokens"]
+                - totals["cache_read_tokens"]
+                - totals["cache_creation_tokens"],
+            ),
             "est_cost_usd": estimate_cost_usd(usage_handler.usage_metadata),
+            "cost_breakdown": cost_breakdown,
+            "summarization_cost_usd": summarization_cost,
+            "max_request_context_tokens_approx": max(
+                (
+                    int(call.get("request_context_tokens_approx") or 0)
+                    for call in model_calls
+                ),
+                default=0,
+            ),
+            "compaction_events": turn_events,
             "ttft_ms": (
                 int((first_text_at - turn_started) * 1000)
                 if first_text_at is not None
@@ -2315,7 +3127,7 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
             **context_window_stats(state_messages, CLEARED_TOOL_OUTPUT_PLACEHOLDER),
             # Signals from per-call-view middlewares (this turn only); absent
             # keys simply mean that mechanism did not fire.
-            **pop_turn_signals(message_id),
+            **turn_signals,
         },
     )
     yield ChatEvent(

app/chat_types.py

@@ -39,6 +39,9 @@ class ChatRequest:
     # Long-term memory key: profile-memory presets read and update the stored
     # student profile under this id. Empty disables profile memory I/O.
     student_id: str = ""
+    # Experiment-only DeepSeek KV-cache namespace. The eval runner generates a
+    # stable opaque id per arm/session/trial to prevent cross-arm cache warming.
+    cache_user_id: str = ""
     # Part C / Axis B ablation: drop the run_kb_command tool (and its prompt
     # section) while keeping retrieval, to measure whether KB browsing helps.
     disable_kb: bool = False

app/memory_presets.py

@@ -92,9 +92,26 @@ class MemoryConfig:
     summarization: bool = True
     summarization_trigger_tokens: int = 30_000
     summarization_keep_messages: int = 20
+    # Experiment arms use token-based retention so a single large tool message
+    # cannot make the post-compaction window vary by hundreds of thousands of
+    # tokens. None preserves the historical message-count behavior.
+    summarization_keep_tokens: int | None = None
+    # LangChain defaults this to 4k. None deliberately sends the entire selected
+    # older history to the summarizer (the corrected long-context experiment).
+    summarization_trim_tokens: int | None = 4_000
+    # Fail rather than silently trim if a full-input experimental summary would
+    # approach the provider's context ceiling. None disables the guard.
+    summarization_input_guard_tokens: int | None = None
     # Custom SummarizationMiddleware prompt (None = the library default). Used by
     # the selective_retention / context_reset arms; must template {messages}.
     summary_prompt: str | None = None
+    # ``xml`` is LangChain's historical behavior: serialize selected messages
+    # into one new prompt string. ``structured_prefix`` keeps the original
+    # system message, tool schemas, model settings, and selected message prefix
+    # byte-for-byte at the message boundary, then appends one checkpoint
+    # instruction. The latter is experiment-only because it changes request
+    # shape and checkpoint installation semantics.
+    summarization_strategy: str = "xml"
     context_editing: bool = True
     context_editing_trigger_tokens: int = 5_000
     context_editing_keep: int = 5
@@ -112,6 +129,19 @@ class MemoryConfig:
     truncate_head_chars: int = 2_000
     truncate_tail_chars: int = 500
     truncate_trigger_chars: int = 4_000
+    # Persistent insertion-time cap: unlike truncate_tool_outputs, this changes
+    # the checkpoint itself, so every later model call and the summarizer see the
+    # same stable text. The experiment uses 40k UTF-8 bytes as a nominal 10k-token
+    # cap, matching the reproducible approximation used by the Codex harness.
+    tool_output_cap_bytes: int | None = None
+    # Enables explanatory compaction telemetry and DeepSeek user_id isolation.
+    # Kept off for production and historical presets so this study is additive.
+    experiment_mode: bool = False
+    # Fail before an experimental agent call exceeds this approximate input
+    # size. The approximation over-counted the provider-reported input by about
+    # 29k near 870k, so 990k preserves real headroom inside DeepSeek's 1M window
+    # without prematurely truncating the full-history control.
+    experiment_request_guard_tokens: int = 990_000
     compress_prompt: bool = False  # deterministic per-call text compaction
     # In-context history retrieval (Axis A subsystem): keep the last N turn-blocks
     # and retrieve only the top-k most relevant older blocks. None disables it.
@@ -132,6 +162,78 @@ MEMORY_PRESETS: dict[str, MemoryConfig] = {
     "full_history": MemoryConfig(
         name="full_history", summarization=False, context_editing=False
     ),
+    # --- DeepSeek long-context compaction experiment ----------------------
+    # Four mechanism-isolation arms. All disable age-based context editing;
+    # the capped arms instead perform one stable rewrite when tool output first
+    # enters history. C200 arms summarize the complete selected prefix at 200k
+    # and retain a controlled 50k-token recent tail.
+    "exp_fh_raw": MemoryConfig(
+        name="exp_fh_raw",
+        summarization=False,
+        context_editing=False,
+        experiment_mode=True,
+    ),
+    "exp_fh_cap10k": MemoryConfig(
+        name="exp_fh_cap10k",
+        summarization=False,
+        context_editing=False,
+        tool_output_cap_bytes=40_000,
+        experiment_mode=True,
+    ),
+    "exp_c200_raw": MemoryConfig(
+        name="exp_c200_raw",
+        summarization_trigger_tokens=200_000,
+        summarization_keep_tokens=50_000,
+        summarization_trim_tokens=None,
+        summarization_input_guard_tokens=900_000,
+        context_editing=False,
+        experiment_mode=True,
+    ),
+    "exp_c200_cap10k": MemoryConfig(
+        name="exp_c200_cap10k",
+        summarization_trigger_tokens=200_000,
+        summarization_keep_tokens=50_000,
+        summarization_trim_tokens=None,
+        summarization_input_guard_tokens=900_000,
+        context_editing=False,
+        tool_output_cap_bytes=40_000,
+        experiment_mode=True,
+    ),
+    # Cache-friendly version of exp_c200_cap10k. It deliberately remains a
+    # separate arm so the completed XML run stays reproducible and comparable.
+    "exp_c200_cap10k_structured": MemoryConfig(
+        name="exp_c200_cap10k_structured",
+        summarization_trigger_tokens=200_000,
+        summarization_keep_tokens=50_000,
+        summarization_trim_tokens=None,
+        summarization_input_guard_tokens=900_000,
+        summarization_strategy="structured_prefix",
+        context_editing=False,
+        tool_output_cap_bytes=40_000,
+        experiment_mode=True,
+    ),
+    # Stage-2 threshold sensitivity arms share the exact same cap, summary
+    # input, and post-compaction retention. Only the trigger changes.
+    "exp_c400_cap10k": MemoryConfig(
+        name="exp_c400_cap10k",
+        summarization_trigger_tokens=400_000,
+        summarization_keep_tokens=50_000,
+        summarization_trim_tokens=None,
+        summarization_input_guard_tokens=900_000,
+        context_editing=False,
+        tool_output_cap_bytes=40_000,
+        experiment_mode=True,
+    ),
+    "exp_c800_cap10k": MemoryConfig(
+        name="exp_c800_cap10k",
+        summarization_trigger_tokens=800_000,
+        summarization_keep_tokens=50_000,
+        summarization_trim_tokens=None,
+        summarization_input_guard_tokens=900_000,
+        context_editing=False,
+        tool_output_cap_bytes=40_000,
+        experiment_mode=True,
+    ),
     # What production runs today.
     "prod": MemoryConfig(name="prod"),
     "summarization_only": MemoryConfig(

app/telemetry.py

@@ -16,13 +16,15 @@ number.
 from __future__ import annotations
 
 import threading
+import time
 from dataclasses import dataclass
 from typing import Any
+from uuid import UUID
 
 from langchain_core.callbacks.usage import UsageMetadataCallbackHandler
-from langchain_core.messages import BaseMessage
+from langchain_core.messages import AIMessage, BaseMessage
 from langchain_core.messages.utils import count_tokens_approximately
-from langchain_core.outputs import LLMResult
+from langchain_core.outputs import ChatGeneration, LLMResult
 
 
 # --- Turn-scoped middleware signals ------------------------------------------
@@ -39,6 +41,7 @@ from langchain_core.outputs import LLMResult
 # wrap_model_call in a worker thread, where a ContextVar update lands on a copy
 # and is lost. A plain global + lock is visible from any thread.
 _TURN_SIGNALS: dict[str, dict[str, int]] = {}
+_TURN_EVENTS: dict[str, list[dict[str, Any]]] = {}
 _TURN_SIGNALS_LOCK = threading.Lock()
 _MAX_TRACKED_TURNS = 256
 
@@ -67,8 +70,11 @@ def reset_turn_signals(turn_id: str) -> None:
         # Never clear() the whole dict — that would wipe concurrent in-flight
         # turns' signals and could mis-grade a probe as "compaction never fired".
         while len(_TURN_SIGNALS) >= _MAX_TRACKED_TURNS:
-            _TURN_SIGNALS.pop(next(iter(_TURN_SIGNALS)), None)
+            oldest = next(iter(_TURN_SIGNALS))
+            _TURN_SIGNALS.pop(oldest, None)
+            _TURN_EVENTS.pop(oldest, None)
         _TURN_SIGNALS[turn_id] = {}
+        _TURN_EVENTS[turn_id] = []
 
 
 def record_turn_signal(turn_id: str, name: str, amount: int = 1) -> None:
@@ -103,16 +109,107 @@ def pop_turn_signals(turn_id: str) -> dict[str, int]:
         return _TURN_SIGNALS.pop(turn_id, {})
 
 
+def record_turn_event(turn_id: str, event: dict[str, Any]) -> None:
+    """Append structured middleware telemetry for one turn (thread-safe)."""
+    if not turn_id:
+        return
+    with _TURN_SIGNALS_LOCK:
+        _TURN_EVENTS.setdefault(turn_id, []).append(dict(event))
+
+
+def pop_turn_events(turn_id: str) -> list[dict[str, Any]]:
+    """Take and clear a turn's structured middleware events."""
+    if not turn_id:
+        return []
+    with _TURN_SIGNALS_LOCK:
+        return _TURN_EVENTS.pop(turn_id, [])
+
+
 class TurnUsageHandler(UsageMetadataCallbackHandler):
-    """Aggregate per-model usage for one turn and count chat-model calls."""
+    """Aggregate usage and retain one explanatory record per model call."""
 
     def __init__(self) -> None:
         super().__init__()
         self.llm_calls = 0
+        self.model_calls: list[dict[str, Any]] = []
+        self._call_starts: dict[UUID, dict[str, Any]] = {}
+
+    def on_chat_model_start(
+        self,
+        serialized: dict[str, Any],
+        messages: list[list[BaseMessage]],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        del serialized, parent_run_id, tags, kwargs
+        request_messages = messages[0] if messages else []
+        try:
+            request_tokens = int(count_tokens_approximately(request_messages))
+        except Exception:  # telemetry must never break a model call
+            request_tokens = 0
+        call_metadata = metadata or {}
+        with self._lock:
+            self._call_starts[run_id] = {
+                "started_at": time.monotonic(),
+                "source": str(call_metadata.get("lc_source") or "agent"),
+                "request_context_tokens_approx": request_tokens,
+            }
 
     def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
+        run_id = kwargs.get("run_id")
+        generation = None
+        try:
+            generation = response.generations[0][0]
+        except IndexError:
+            pass
+
+        message = generation.message if isinstance(generation, ChatGeneration) else None
+        usage = (
+            dict(message.usage_metadata or {}) if isinstance(message, AIMessage) else {}
+        )
+        model_name = (
+            str(message.response_metadata.get("model_name") or "")
+            if isinstance(message, AIMessage)
+            else ""
+        )
+        details = dict(usage.get("input_token_details") or {})
+        breakdown = usage_cost_breakdown(model_name, usage) if model_name else None
         with self._lock:
             self.llm_calls += 1
+            start = self._call_starts.pop(run_id, {}) if run_id else {}
+            self.model_calls.append(
+                {
+                    "sequence": self.llm_calls,
+                    "source": start.get("source", "agent"),
+                    "model": model_name,
+                    "input_tokens": int(usage.get("input_tokens", 0) or 0),
+                    "cache_read_tokens": int(details.get("cache_read", 0) or 0),
+                    "cache_miss_tokens": max(
+                        0,
+                        int(usage.get("input_tokens", 0) or 0)
+                        - int(details.get("cache_read", 0) or 0)
+                        - int(details.get("cache_creation", 0) or 0),
+                    ),
+                    "cache_creation_tokens": int(details.get("cache_creation", 0) or 0),
+                    "output_tokens": int(usage.get("output_tokens", 0) or 0),
+                    "total_tokens": int(usage.get("total_tokens", 0) or 0),
+                    "usage_reported": bool(usage),
+                    "cache_details_reported": "cache_read" in details,
+                    "request_context_tokens_approx": int(
+                        start.get("request_context_tokens_approx", 0) or 0
+                    ),
+                    "duration_ms": (
+                        int((time.monotonic() - start["started_at"]) * 1000)
+                        if start.get("started_at") is not None
+                        else None
+                    ),
+                    "cost": breakdown,
+                }
+            )
         super().on_llm_end(response, **kwargs)
 
 
@@ -165,6 +262,52 @@ def pricing_for_model(model_key: str) -> ModelPricing | None:
     return best[1] if best else None
 
 
+def usage_cost_breakdown(
+    model_key: str, usage: dict[str, Any]
+) -> dict[str, float] | None:
+    """Return mutually exclusive USD components for one model's usage."""
+    pricing = pricing_for_model(model_key)
+    if pricing is None:
+        return None
+    input_tokens = int(usage.get("input_tokens", 0) or 0)
+    output_tokens = int(usage.get("output_tokens", 0) or 0)
+    details = usage.get("input_token_details") or {}
+    cache_read = int(details.get("cache_read", 0) or 0)
+    cache_creation = int(details.get("cache_creation", 0) or 0)
+    plain_input = max(0, input_tokens - cache_read - cache_creation)
+    write_rate = (
+        pricing.cache_write if pricing.cache_write is not None else pricing.input
+    )
+    result = {
+        "cache_miss_input_usd": plain_input * pricing.input / 1_000_000,
+        "cache_read_input_usd": cache_read * pricing.cache_read / 1_000_000,
+        "cache_creation_input_usd": cache_creation * write_rate / 1_000_000,
+        "output_usd": output_tokens * pricing.output / 1_000_000,
+    }
+    result["total_usd"] = sum(result.values())
+    return result
+
+
+def aggregate_cost_breakdown(
+    usage_by_model: dict[str, Any],
+) -> dict[str, float] | None:
+    """Sum cost components across every model used in a turn."""
+    totals = {
+        "cache_miss_input_usd": 0.0,
+        "cache_read_input_usd": 0.0,
+        "cache_creation_input_usd": 0.0,
+        "output_usd": 0.0,
+        "total_usd": 0.0,
+    }
+    for model_key, usage in usage_by_model.items():
+        breakdown = usage_cost_breakdown(model_key, usage)
+        if breakdown is None:
+            return None
+        for key in totals:
+            totals[key] += breakdown[key]
+    return totals
+
+
 def usage_totals(usage_by_model: dict[str, Any]) -> dict[str, int]:
     """Sum usage across models into the fields the stats event reports."""
     totals = {
@@ -191,27 +334,8 @@ def estimate_cost_usd(usage_by_model: dict[str, Any]) -> float | None:
     includes them (LangChain's UsageMetadata convention), so they are carved
     out of the plain-input bucket rather than added on top.
     """
-    total = 0.0
-    for model_key, usage in usage_by_model.items():
-        pricing = pricing_for_model(model_key)
-        if pricing is None:
-            return None
-        input_tokens = int(usage.get("input_tokens", 0) or 0)
-        output_tokens = int(usage.get("output_tokens", 0) or 0)
-        details = usage.get("input_token_details") or {}
-        cache_read = int(details.get("cache_read", 0) or 0)
-        cache_creation = int(details.get("cache_creation", 0) or 0)
-        plain_input = max(0, input_tokens - cache_read - cache_creation)
-        write_rate = (
-            pricing.cache_write if pricing.cache_write is not None else pricing.input
-        )
-        total += (
-            plain_input * pricing.input
-            + cache_read * pricing.cache_read
-            + cache_creation * write_rate
-            + output_tokens * pricing.output
-        ) / 1_000_000
-    return total
+    breakdown = aggregate_cost_breakdown(usage_by_model)
+    return breakdown["total_usd"] if breakdown is not None else None
 
 
 def context_window_stats(

tests/test_chat_service.py

@@ -32,6 +32,7 @@ from app.chat_service import (
     extract_shell_source_matches,
     resolve_answer_citations,
     retrieve_tutor_context,
+    run_kb_command,
     supports_gemini_tool_combination,
     sync_thread_with_history,
     stream_chat,
@@ -466,6 +467,59 @@ class ChatServiceTestCase(unittest.TestCase):
         finally:
             _clear_kb_command_budget(session_id)
 
+    def test_kb_command_accepts_timeout_alias_with_runtime_cap(self) -> None:
+        session_id = "test_timeout_alias"
+        _clear_kb_command_budget(session_id)
+        runtime = types.SimpleNamespace(
+            context=types.SimpleNamespace(
+                kb_session_id=session_id,
+                kb_command_limit=3,
+            )
+        )
+        try:
+            with (
+                patch("app.chat_service.ensure_local_vector_db"),
+                patch("app.chat_service.execute_kb_command") as execute,
+                patch("app.chat_service.format_command_payload", return_value="$ ls"),
+            ):
+                result = run_kb_command.func(
+                    command="ls",
+                    runtime=runtime,
+                    timeout=5,
+                )
+        finally:
+            _clear_kb_command_budget(session_id)
+        self.assertEqual(result, "$ ls")
+        execute.assert_called_once_with(
+            "ls",
+            timeout_seconds=5,
+            max_output_chars=40000,
+        )
+
+    def test_kb_command_unknown_arguments_return_soft_tool_error(self) -> None:
+        runtime = types.SimpleNamespace(
+            context=types.SimpleNamespace(
+                kb_session_id="test_unknown_tool_arg",
+                kb_command_limit=3,
+            )
+        )
+        with patch("app.chat_service.execute_kb_command") as execute:
+            result = run_kb_command.func(
+                command="ls",
+                runtime=runtime,
+                working_directory="raw",
+            )
+        self.assertIn("unsupported run_kb_command argument", result)
+        self.assertIn("working_directory", result)
+        execute.assert_not_called()
+
+    def test_kb_command_schema_disallows_unpublished_arguments(self) -> None:
+        schema = run_kb_command.args_schema
+        self.assertIsInstance(schema, dict)
+        self.assertFalse(schema["additionalProperties"])
+        self.assertIn("timeout", schema["properties"])
+        self.assertEqual(schema["properties"]["timeout"]["maximum"], 30)
+
     def test_retrieve_tutor_context_degrades_on_retriever_failure(self) -> None:
         runtime = types.SimpleNamespace(
             context=types.SimpleNamespace(allowed_sources=("transformers",))
@@ -480,6 +534,21 @@ class ChatServiceTestCase(unittest.TestCase):
         # ...and the raw provider error is not exposed in the tool output.
         self.assertNotIn("cohere 500 boom", result)
 
+    def test_retrieve_tutor_context_unknown_arguments_return_soft_error(self) -> None:
+        runtime = types.SimpleNamespace(
+            context=types.SimpleNamespace(allowed_sources=("transformers",))
+        )
+        with patch("app.chat_service.select_retriever") as select:
+            result = retrieve_tutor_context.func(
+                query="What is RAG?",
+                runtime=runtime,
+                top_k=20,
+            )
+        self.assertIn("unsupported argument", result)
+        self.assertIn("top_k", result)
+        select.assert_not_called()
+        self.assertFalse(retrieve_tutor_context.args_schema["additionalProperties"])
+
     def test_resolve_answer_citations_uses_current_turn_evidence(self) -> None:
         retrieval = SourceMatch(
             doc_id="peft:lora",

tests/test_memory_presets.py

@@ -5,7 +5,11 @@ import unittest
 from unittest.mock import patch
 
 from app.chat_service import (
+    DeepSeekCacheIsolationMiddleware,
+    InstrumentedSummarizationMiddleware,
+    PrefixPreservingCompactionMiddleware,
     SourcePreferenceMiddleware,
+    StableToolOutputCapMiddleware,
     StudentProfileMiddleware,
     build_agent,
     build_agent_middleware,
@@ -58,6 +62,26 @@ class MemoryPresetResolutionTests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 resolve_memory_preset("")
 
+    def test_deepseek_experiment_arms_are_single_axis_configs(self) -> None:
+        raw = MEMORY_PRESETS["exp_fh_raw"]
+        capped = MEMORY_PRESETS["exp_fh_cap10k"]
+        compact = MEMORY_PRESETS["exp_c200_cap10k"]
+        self.assertTrue(raw.experiment_mode)
+        self.assertFalse(raw.summarization)
+        self.assertFalse(raw.context_editing)
+        self.assertEqual(capped.tool_output_cap_bytes, 40_000)
+        self.assertEqual(compact.summarization_trigger_tokens, 200_000)
+        self.assertEqual(compact.summarization_keep_tokens, 50_000)
+        self.assertIsNone(compact.summarization_trim_tokens)
+        self.assertEqual(compact.summarization_input_guard_tokens, 900_000)
+        self.assertEqual(compact.experiment_request_guard_tokens, 990_000)
+        self.assertFalse(compact.context_editing)
+        structured = MEMORY_PRESETS["exp_c200_cap10k_structured"]
+        self.assertEqual(structured.summarization_strategy, "structured_prefix")
+        self.assertEqual(structured.summarization_trigger_tokens, 200_000)
+        self.assertEqual(structured.summarization_keep_tokens, 50_000)
+        self.assertEqual(structured.tool_output_cap_bytes, 40_000)
+
 
 class MiddlewareAssemblyTests(unittest.TestCase):
     def test_full_history_disables_compaction(self) -> None:
@@ -102,6 +126,40 @@ class MiddlewareAssemblyTests(unittest.TestCase):
             any(isinstance(m, StudentProfileMiddleware) for m in middleware)
         )
 
+    def test_experiment_stack_has_isolation_cap_and_full_input_summary(self) -> None:
+        middleware = build_agent_middleware(
+            _stub_model(), MEMORY_PRESETS["exp_c200_cap10k"]
+        )
+        self.assertTrue(
+            any(isinstance(m, DeepSeekCacheIsolationMiddleware) for m in middleware)
+        )
+        self.assertTrue(
+            any(isinstance(m, StableToolOutputCapMiddleware) for m in middleware)
+        )
+        summary = next(
+            m for m in middleware if isinstance(m, InstrumentedSummarizationMiddleware)
+        )
+        self.assertEqual(summary.keep, ("tokens", 50_000))
+        self.assertIsNone(summary.trim_tokens_to_summarize)
+        self.assertEqual(summary.summary_input_guard_tokens, 900_000)
+        self.assertFalse(
+            any(isinstance(m, ContextEditingMiddleware) for m in middleware)
+        )
+
+    def test_structured_compactor_runs_after_request_shaping_middleware(self) -> None:
+        middleware = build_agent_middleware(
+            _stub_model(), MEMORY_PRESETS["exp_c200_cap10k_structured"]
+        )
+        self.assertIsInstance(middleware[0], DeepSeekCacheIsolationMiddleware)
+        self.assertTrue(
+            any(isinstance(m, StableToolOutputCapMiddleware) for m in middleware)
+        )
+        self.assertIsInstance(middleware[-2], SourcePreferenceMiddleware)
+        self.assertIsInstance(middleware[-1], PrefixPreservingCompactionMiddleware)
+        self.assertFalse(
+            any(type(m) is InstrumentedSummarizationMiddleware for m in middleware)
+        )
+
     def test_build_agent_cache_keys_include_memory_config(self) -> None:
         build_agent.cache_clear()
         created = []

tests/test_memory_variants.py

@@ -8,22 +8,49 @@ no model client, no API keys, no vector DB.
 
 from __future__ import annotations
 
+import hashlib
 import unittest
 from types import SimpleNamespace
+from unittest import mock
 
 import tiktoken
-from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain.agents import create_agent
+from langchain.agents.middleware import ModelRequest, ModelResponse
+from langchain.tools import tool
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import (
+    AIMessage,
+    HumanMessage,
+    RemoveMessage,
+    SystemMessage,
+    ToolMessage,
+)
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
 
 from app.chat_service import (
+    AppContext,
+    DeepSeekCacheIsolationMiddleware,
     InContextHistoryRetrievalMiddleware,
+    InstrumentedSummarizationMiddleware,
     ObservationTruncationMiddleware,
     PromptCompressionMiddleware,
+    PrefixPreservingCompactionMiddleware,
     SlidingWindowMiddleware,
+    StableToolOutputCapMiddleware,
     build_agent_middleware,
 )
 from app.chroma_rag import LocalChromaRetriever
 from app.memory_presets import resolve_memory_preset
-from app.telemetry import COMPACTION_SIGNAL_NAMES, pop_turn_signals, reset_turn_signals
+from app.telemetry import (
+    COMPACTION_SIGNAL_NAMES,
+    TurnUsageHandler,
+    estimate_cost_usd,
+    pop_turn_events,
+    pop_turn_signals,
+    reset_turn_signals,
+    usage_totals,
+)
 from evals.common import COMPACTION_SIGNAL_KEYS, compaction_active
 
 
@@ -140,6 +167,1171 @@ class ObservationTruncationTests(unittest.TestCase):
         self.assertEqual(pop_turn_signals("t1"), {})
 
 
+OVERSIZED_TOOL_OUTPUT = "HEAD" + "x" * 20_000 + "TAIL"
+
+
+@tool
+def big_lookup(query: str) -> str:
+    """Return deliberately oversized evidence."""
+    del query
+    return OVERSIZED_TOOL_OUTPUT
+
+
+class ToolCallingFakeModel(FakeMessagesListChatModel):
+    """Scripted model that accepts tool binding (the base class raises)."""
+
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+
+class StableToolOutputCapTests(unittest.TestCase):
+    def test_cap_persists_to_checkpoint_and_summarizer_never_sees_raw(self) -> None:
+        model = ToolCallingFakeModel(
+            responses=[
+                AIMessage(
+                    content="",
+                    tool_calls=[
+                        {
+                            "name": "big_lookup",
+                            "args": {"query": "q"},
+                            "id": "call-big",
+                        }
+                    ],
+                ),
+                AIMessage(content="final answer"),
+            ]
+        )
+        agent = create_agent(
+            model=model,
+            tools=[big_lookup],
+            middleware=[StableToolOutputCapMiddleware(2_048)],
+            checkpointer=InMemorySaver(),
+        )
+        config = {"configurable": {"thread_id": "cap-thread"}}
+        reset_turn_signals("cap-e2e")
+        agent.invoke(
+            {"messages": [HumanMessage(content="look this up")]},
+            config=config,
+            context=AppContext(allowed_sources=(), kb_session_id="cap-e2e"),
+        )
+
+        checkpointed = agent.get_state(config).values["messages"]
+        tool_messages = [m for m in checkpointed if isinstance(m, ToolMessage)]
+        self.assertEqual(len(tool_messages), 1)
+        capped = tool_messages[0]
+        self.assertLessEqual(len(capped.content.encode("utf-8")), 2_048)
+        self.assertIn("truncated at stable 2048-byte cap", capped.content)
+        self.assertNotIn(OVERSIZED_TOOL_OUTPUT, capped.content)
+        metadata = capped.additional_kwargs["stable_tool_cap"]
+        self.assertEqual(
+            metadata["sha256"],
+            hashlib.sha256(OVERSIZED_TOOL_OUTPUT.encode("utf-8")).hexdigest(),
+        )
+        self.assertEqual(
+            metadata["original_bytes"], len(OVERSIZED_TOOL_OUTPUT.encode("utf-8"))
+        )
+        self.assertEqual(pop_turn_signals("cap-e2e")["tool_outputs_capped"], 1)
+
+        # XML path: the summarizer's prompt is built from the checkpointed
+        # (capped) ToolMessage, never the raw oversized output.
+        summarizer = InstrumentedSummarizationMiddleware(
+            model=ExperimentCompactionMiddlewareTests.FakeModel(),
+            trigger=("tokens", 100),
+            keep=("tokens", 30),
+            trim_tokens_to_summarize=None,
+        )
+        plan = summarizer._plan_compaction({"messages": checkpointed})
+        self.assertIsNotNone(plan)
+        planned_tool = next(m for m in plan["trimmed"] if isinstance(m, ToolMessage))
+        self.assertEqual(planned_tool.content, capped.content)
+        prompt = summarizer._summary_prompt_text(plan["trimmed"])
+        self.assertIn("truncated at stable 2048-byte cap", prompt)
+        self.assertNotIn(OVERSIZED_TOOL_OUTPUT, prompt)
+
+        # Structured path: the summary request extends the same checkpointed
+        # prefix, so it carries the identical capped ToolMessage.
+        structured = PrefixPreservingCompactionMiddleware(
+            model=ExperimentCompactionMiddlewareTests.FakeModel(),
+            trigger=("tokens", 100),
+            keep=("tokens", 30),
+            trim_tokens_to_summarize=None,
+        )
+        request = ModelRequest(
+            model=ExperimentCompactionMiddlewareTests.FakeModel(),
+            messages=list(checkpointed),
+            system_message=SystemMessage(content="system"),
+            tools=[],
+            tool_choice=None,
+            response_format=None,
+            model_settings={},
+            state={"messages": list(checkpointed)},
+            runtime=SimpleNamespace(
+                context=SimpleNamespace(kb_session_id="cap-e2e", cache_user_id="")
+            ),
+        )
+        structured_plan = structured._plan_compaction(request.state)
+        self.assertIsNotNone(structured_plan)
+        _, summary_request_messages = structured._prepare_summary_request(
+            request, structured_plan
+        )
+        request_tools = [
+            m for m in summary_request_messages if isinstance(m, ToolMessage)
+        ]
+        self.assertEqual([m.content for m in request_tools], [capped.content])
+
+    def test_cap_is_persistent_bounded_and_auditable(self) -> None:
+        raw = "HEAD" + ("é" * 30_000) + "TAIL"
+        request = make_request([], "stable-cap")
+        reset_turn_signals("stable-cap")
+        result = StableToolOutputCapMiddleware(40_000)._cap(
+            request,
+            ToolMessage(content=raw, tool_call_id="call-cap"),
+        )
+        self.assertLessEqual(len(result.content.encode("utf-8")), 40_000)
+        self.assertTrue(result.content.startswith("HEAD"))
+        self.assertTrue(result.content.endswith("TAIL"))
+        metadata = result.additional_kwargs["stable_tool_cap"]
+        self.assertEqual(metadata["original_bytes"], len(raw.encode("utf-8")))
+        self.assertEqual(len(metadata["sha256"]), 64)
+        signals = pop_turn_signals("stable-cap")
+        self.assertEqual(signals["tool_outputs_capped"], 1)
+        self.assertGreater(
+            signals["tool_output_original_bytes"],
+            signals["tool_output_retained_bytes"],
+        )
+
+
+class ExperimentCompactionMiddlewareTests(unittest.TestCase):
+    class FakeModel:
+        _llm_type = "fake-chat-model"
+
+        def __init__(self, responses: list[str] | None = None) -> None:
+            self.bound: list[dict] = []
+            self.prompts: list[str] = []
+            self.responses = list(responses or ["durable full-input summary"])
+
+        def bind(self, **kwargs):
+            self.bound.append(kwargs)
+            return self
+
+        def invoke(self, prompt, config=None):
+            self.prompts.append(prompt)
+            return AIMessage(content=self.responses.pop(0))
+
+        async def ainvoke(self, prompt, config=None):
+            return self.invoke(prompt, config=config)
+
+        def _get_ls_params(self):
+            return {"ls_provider": "deepseek"}
+
+    class StructuredFakeModel(FakeModel):
+        def __init__(self, responses: list[str] | None = None) -> None:
+            super().__init__(responses)
+            self.bound_tools: list[tuple[list, dict]] = []
+            self.invocations: list[tuple[list, dict | None]] = []
+
+        def bind_tools(self, tools, **kwargs):
+            self.bound_tools.append((list(tools), dict(kwargs)))
+            return self
+
+        def invoke(self, prompt, config=None):
+            self.invocations.append((list(prompt), config))
+            return AIMessage(
+                content=self.responses.pop(0),
+                usage_metadata={
+                    "input_tokens": 10_000,
+                    "output_tokens": 100,
+                    "total_tokens": 10_100,
+                    "input_token_details": {"cache_read": 9_000},
+                },
+                response_metadata={"model_name": "deepseek-v4-flash"},
+            )
+
+    class ScriptedMessageModel(FakeModel):
+        """Structured-path fake returning prebuilt AIMessage responses verbatim."""
+
+        def __init__(self, responses: list[AIMessage]) -> None:
+            super().__init__([])
+            self.message_responses = list(responses)
+            self.invocations: list[tuple[list, dict | None]] = []
+
+        def bind_tools(self, tools, **kwargs):
+            return self
+
+        def invoke(self, prompt, config=None):
+            self.invocations.append((list(prompt), config))
+            return self.message_responses.pop(0)
+
+    @staticmethod
+    def _structured_request(model, messages: list, turn_id: str) -> ModelRequest:
+        return ModelRequest(
+            model=model,
+            messages=messages,
+            system_message=SystemMessage(content="system"),
+            tools=[],
+            tool_choice=None,
+            response_format=None,
+            model_settings={},
+            state={"messages": messages},
+            runtime=SimpleNamespace(
+                context=SimpleNamespace(kb_session_id=turn_id, cache_user_id="")
+            ),
+        )
+
+    def test_cache_user_id_is_added_to_agent_model_settings(self) -> None:
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(cache_user_id="eval_abc", kb_session_id="turn")
+        )
+        request = SimpleNamespace(
+            runtime=runtime, model_settings={}, messages=[], system_message=None
+        )
+        request.override = lambda **updates: SimpleNamespace(
+            runtime=runtime,
+            model_settings=updates.get("model_settings", request.model_settings),
+        )
+        isolated = DeepSeekCacheIsolationMiddleware()._isolate(request)
+        self.assertEqual(isolated.model_settings["extra_body"], {"user_id": "eval_abc"})
+
+    def test_agent_request_guard_fails_before_model_handler(self) -> None:
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(cache_user_id="eval_guard", kb_session_id="turn")
+        )
+        request = SimpleNamespace(
+            runtime=runtime,
+            model_settings={},
+            messages=[HumanMessage(content="x" * 4_000)],
+            system_message=None,
+        )
+        with self.assertRaisesRegex(RuntimeError, "Agent request exceeds"):
+            DeepSeekCacheIsolationMiddleware(100)._isolate(request)
+
+    def test_full_selected_history_reaches_summarizer_and_records_event(self) -> None:
+        model = self.FakeModel()
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(content="x" * 1_000)
+            for index in range(28)
+        ]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(
+                kb_session_id="summary-turn", cache_user_id="eval_summary"
+            )
+        )
+        reset_turn_signals("summary-turn")
+        update = middleware.before_model({"messages": messages}, runtime)
+        self.assertIsNotNone(update)
+        events = pop_turn_events("summary-turn")
+        self.assertEqual(len(events), 1)
+        event = events[0]
+        self.assertTrue(event["summary_input_untrimmed"])
+        self.assertEqual(event["configured_trigger_tokens"], 1_000)
+        self.assertGreater(event["summary_input_tokens_approx"], 4_000)
+        self.assertLessEqual(event["retained_tail_tokens_approx"], 500)
+        self.assertEqual(pop_turn_signals("summary-turn")["compactions_this_turn"], 1)
+        self.assertEqual(model.bound[-1]["extra_body"], {"user_id": "eval_summary"})
+        self.assertGreater(len(model.prompts[-1]), 20_000)
+
+    def test_provider_reported_tokens_can_trigger_below_approximation(self) -> None:
+        model = self.FakeModel()
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 200_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000) for _ in range(6)]
+        messages.append(
+            AIMessage(
+                content="previous answer",
+                usage_metadata={
+                    "input_tokens": 205_664,
+                    "output_tokens": 1_672,
+                    "total_tokens": 207_336,
+                },
+                response_metadata={"model_provider": "deepseek"},
+            )
+        )
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(
+                kb_session_id="reported-trigger", cache_user_id="eval_reported"
+            )
+        )
+        reset_turn_signals("reported-trigger")
+        with mock.patch.object(middleware, "token_counter", return_value=199_567):
+            update = middleware.before_model({"messages": messages}, runtime)
+        self.assertIsNotNone(update)
+        event = pop_turn_events("reported-trigger")[0]
+        self.assertEqual(event["pre_compaction_tokens_approx"], 199_567)
+        self.assertEqual(event["trigger_reported_tokens"], 207_336)
+        self.assertEqual(event["trigger_source"], "provider_reported")
+
+    def test_summary_input_guard_fails_before_provider_call(self) -> None:
+        model = self.FakeModel()
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+            summary_input_guard_tokens=200,
+        )
+        messages = [HumanMessage(content="x" * 2_000) for _ in range(4)]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(kb_session_id="guard", cache_user_id="eval_guard")
+        )
+        with self.assertRaisesRegex(RuntimeError, "safety guard"):
+            middleware.before_model({"messages": messages}, runtime)
+        self.assertEqual(model.prompts, [])
+
+    def test_empty_summary_is_retried_and_recorded(self) -> None:
+        model = self.FakeModel(["", "durable retry summary"])
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000) for _ in range(4)]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(kb_session_id="retry", cache_user_id="eval_retry")
+        )
+        reset_turn_signals("retry")
+        with mock.patch("app.chat_service.time.sleep") as sleep:
+            update = middleware.before_model({"messages": messages}, runtime)
+        self.assertIsNotNone(update)
+        self.assertEqual(len(model.prompts), 2)
+        sleep.assert_called_once_with(1.0)
+        event = pop_turn_events("retry")[0]
+        self.assertEqual(event["summary_attempts"], 2)
+        self.assertEqual(event["summary_retry_reasons"], ["empty response"])
+
+    def test_non_retryable_summary_failure_is_not_retried(self) -> None:
+        model = self.FakeModel()
+        model.invoke = mock.Mock(side_effect=ValueError("invalid request"))
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000) for _ in range(4)]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(kb_session_id="no-retry", cache_user_id="eval")
+        )
+        with self.assertRaisesRegex(ValueError, "invalid request"):
+            middleware.before_model({"messages": messages}, runtime)
+        self.assertEqual(model.invoke.call_count, 1)
+
+    def test_real_agent_attributes_summary_and_agent_calls_separately(self) -> None:
+        def response(text: str, input_tokens: int, output_tokens: int) -> AIMessage:
+            return AIMessage(
+                content=text,
+                usage_metadata={
+                    "input_tokens": input_tokens,
+                    "output_tokens": output_tokens,
+                    "total_tokens": input_tokens + output_tokens,
+                    "input_token_details": {"cache_read": 0},
+                },
+                response_metadata={"model_name": "deepseek-v4-flash"},
+            )
+
+        model = FakeMessagesListChatModel(
+            responses=[response("summary", 6_000, 10), response("answer", 700, 20)]
+        )
+        summary = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+            summary_input_guard_tokens=900_000,
+        )
+        agent = create_agent(
+            model=model,
+            tools=[],
+            middleware=[DeepSeekCacheIsolationMiddleware(900_000), summary],
+        )
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(content="x" * 1_000)
+            for index in range(28)
+        ]
+        handler = TurnUsageHandler()
+        reset_turn_signals("integration-turn")
+        result = agent.invoke(
+            {"messages": messages},
+            config={"callbacks": [handler]},
+            context=AppContext(
+                allowed_sources=(),
+                kb_session_id="integration-turn",
+                cache_user_id="eval_integration",
+            ),
+        )
+        self.assertEqual(handler.llm_calls, 2)
+        self.assertEqual(
+            [call["source"] for call in handler.model_calls],
+            ["summarization", "agent"],
+        )
+        self.assertEqual(result["messages"][-1].content, "answer")
+        event = pop_turn_events("integration-turn")[0]
+        self.assertGreater(event["summary_input_tokens_approx"], 4_000)
+        self.assertEqual(
+            pop_turn_signals("integration-turn")["compactions_this_turn"], 1
+        )
+
+    def test_structured_prefix_preserves_request_shape_and_persists_checkpoint(
+        self,
+    ) -> None:
+        model = self.StructuredFakeModel(["durable structured summary"])
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+            summary_input_guard_tokens=900_000,
+        )
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(
+                content=f"m{index}:" + "x" * 1_000
+            )
+            for index in range(28)
+        ]
+        system = SystemMessage(content="stable system prompt")
+        tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "lookup",
+                    "description": "lookup evidence",
+                    "parameters": {"type": "object", "properties": {}},
+                },
+            }
+        ]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(
+                kb_session_id="prefix-turn", cache_user_id="stable-prefix-user"
+            )
+        )
+        request = ModelRequest(
+            model=model,
+            messages=messages,
+            system_message=system,
+            tools=tools,
+            tool_choice=None,
+            response_format=None,
+            model_settings={"extra_body": {"user_id": "stable-prefix-user"}},
+            state={"messages": messages},
+            runtime=runtime,
+        )
+        expected_plan = middleware._plan_compaction(request.state)
+        self.assertIsNotNone(expected_plan)
+        handled: list[ModelRequest] = []
+
+        def handler(compacted_request):
+            handled.append(compacted_request)
+            return ModelResponse(
+                result=[AIMessage(content="final answer", id="answer")]
+            )
+
+        reset_turn_signals("prefix-turn")
+        result = middleware.wrap_model_call(request, handler)
+
+        self.assertEqual(model.bound_tools[0][0], tools)
+        self.assertEqual(
+            model.bound_tools[0][1]["extra_body"],
+            {"user_id": "stable-prefix-user"},
+        )
+        summary_messages, summary_config = model.invocations[0]
+        self.assertIs(summary_messages[0], system)
+        self.assertEqual(summary_messages[1:-1], messages)
+        self.assertNotIn("<messages>", summary_messages[-1].content)
+        self.assertIn(
+            f"final {len(expected_plan['preserved'])} messages",
+            summary_messages[-1].content,
+        )
+        self.assertIn(
+            f"approximately {middleware._partial_token_counter(expected_plan['preserved'])} tokens",
+            summary_messages[-1].content,
+        )
+        self.assertEqual(summary_config["metadata"]["lc_source"], "summarization")
+        self.assertEqual(
+            summary_config["metadata"]["compaction_strategy"],
+            "structured_prefix",
+        )
+
+        compacted = handled[0].messages
+        self.assertEqual(
+            compacted[0].additional_kwargs.get("lc_source"), "summarization"
+        )
+        self.assertEqual(handled[0].state["messages"], compacted)
+        command_messages = result.command.update["messages"]
+        self.assertIsInstance(command_messages[0], RemoveMessage)
+        self.assertEqual(command_messages[0].id, REMOVE_ALL_MESSAGES)
+        self.assertEqual(command_messages[-1].content, "final answer")
+
+        event = pop_turn_events("prefix-turn")[0]
+        self.assertEqual(event["summary_strategy"], "structured_prefix")
+        self.assertTrue(event["summary_request_is_strict_extension"])
+        self.assertEqual(event["summary_prefix_messages"], len(messages))
+        self.assertLess(event["summary_selected_messages"], len(messages))
+        self.assertEqual(
+            event["summary_instruction_retained_messages"],
+            len(expected_plan["preserved"]),
+        )
+        self.assertTrue(event["summary_system_message_present"])
+        self.assertEqual(event["summary_tools_bound"], 1)
+        self.assertTrue(event["summary_cache_user_id_preserved"])
+        self.assertEqual(event["summary_provider_input_tokens"], 10_000)
+        self.assertEqual(event["summary_provider_cache_read_tokens"], 9_000)
+        self.assertEqual(event["summary_provider_cache_miss_tokens"], 1_000)
+        self.assertEqual(event["summary_provider_cache_hit_ratio"], 0.9)
+
+    def test_structured_prefix_safe_tail_never_starts_with_orphaned_tool(self) -> None:
+        model = self.StructuredFakeModel(["summary"])
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 120),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [
+            HumanMessage(content="old " + "x" * 2_000),
+            AIMessage(content="old answer " + "x" * 2_000),
+            HumanMessage(content="tool turn"),
+            AIMessage(
+                content="",
+                tool_calls=[{"name": "lookup", "args": {}, "id": "call-1"}],
+            ),
+            ToolMessage(content="evidence", tool_call_id="call-1"),
+            AIMessage(content="tool answer"),
+            HumanMessage(content="current"),
+        ]
+        plan = middleware._plan_compaction({"messages": messages})
+        self.assertIsNotNone(plan)
+        self.assertTrue(plan["preserved"])
+        self.assertNotIsInstance(plan["preserved"][0], ToolMessage)
+
+    def test_structured_prefix_rejects_an_unpersisted_message_view(self) -> None:
+        model = self.StructuredFakeModel(["summary"])
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        state_messages = [HumanMessage(content="x" * 2_000) for _ in range(4)]
+        request = ModelRequest(
+            model=model,
+            messages=state_messages[1:],
+            system_message=SystemMessage(content="system"),
+            tools=[],
+            response_format=None,
+            state={"messages": state_messages},
+            runtime=SimpleNamespace(
+                context=SimpleNamespace(kb_session_id="mismatched-view")
+            ),
+        )
+        plan = middleware._plan_compaction(request.state)
+        self.assertIsNotNone(plan)
+        with self.assertRaisesRegex(RuntimeError, "match checkpoint history"):
+            middleware._prepare_summary_request(request, plan)
+
+    def test_structured_prefix_empty_summary_is_retried_and_recorded(self) -> None:
+        # Empty text and tool_calls-with-empty-text are both "empty" responses:
+        # each is retried and shows up in summary_retry_reasons.
+        model = self.ScriptedMessageModel(
+            [
+                AIMessage(content=""),
+                AIMessage(
+                    content="",
+                    tool_calls=[{"name": "lookup", "args": {}, "id": "call-empty"}],
+                ),
+                AIMessage(content="structured retry checkpoint"),
+            ]
+        )
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000, id=f"h{i}") for i in range(4)]
+        request = self._structured_request(model, messages, "structured-retry")
+        handled: list[ModelRequest] = []
+
+        def handler(compacted_request):
+            handled.append(compacted_request)
+            return ModelResponse(result=[AIMessage(content="answer", id="a")])
+
+        reset_turn_signals("structured-retry")
+        with mock.patch("app.chat_service.time.sleep") as sleep:
+            middleware.wrap_model_call(request, handler)
+        self.assertEqual(len(model.invocations), 3)
+        self.assertEqual(sleep.call_args_list, [mock.call(1.0), mock.call(2.0)])
+        self.assertEqual(len(handled), 1)
+        self.assertIn("structured retry checkpoint", handled[0].messages[0].content)
+        event = pop_turn_events("structured-retry")[0]
+        self.assertEqual(event["summary_attempts"], 3)
+        self.assertEqual(
+            event["summary_retry_reasons"], ["empty response", "empty response"]
+        )
+        self.assertEqual(
+            pop_turn_signals("structured-retry")["compactions_this_turn"], 1
+        )
+
+    def test_structured_prefix_empty_summary_raises_after_max_attempts(self) -> None:
+        model = self.ScriptedMessageModel([AIMessage(content="")] * 3)
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000, id=f"h{i}") for i in range(4)]
+        request = self._structured_request(model, messages, "structured-exhausted")
+        handled: list[ModelRequest] = []
+        reset_turn_signals("structured-exhausted")
+        with mock.patch("app.chat_service.time.sleep"):
+            with self.assertRaisesRegex(RuntimeError, "empty summary after 3 attempts"):
+                middleware.wrap_model_call(request, handled.append)
+        self.assertEqual(len(model.invocations), 3)
+        # The agent call never runs on a failed checkpoint, and nothing is
+        # recorded as a successful compaction.
+        self.assertEqual(handled, [])
+        self.assertEqual(pop_turn_events("structured-exhausted"), [])
+        self.assertNotIn(
+            "compactions_this_turn", pop_turn_signals("structured-exhausted")
+        )
+
+    def test_structured_prefix_uses_text_and_ignores_summary_tool_calls(self) -> None:
+        model = self.ScriptedMessageModel(
+            [
+                AIMessage(
+                    content="checkpoint despite tool call",
+                    tool_calls=[{"name": "lookup", "args": {}, "id": "call-x"}],
+                )
+            ]
+        )
+        middleware = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000, id=f"h{i}") for i in range(4)]
+        request = self._structured_request(model, messages, "structured-toolcall")
+        handled: list[ModelRequest] = []
+
+        def handler(compacted_request):
+            handled.append(compacted_request)
+            return ModelResponse(result=[AIMessage(content="answer", id="a")])
+
+        reset_turn_signals("structured-toolcall")
+        result = middleware.wrap_model_call(request, handler)
+        self.assertEqual(len(model.invocations), 1)
+        self.assertIn("checkpoint despite tool call", handled[0].messages[0].content)
+        # The summarizer's AIMessage never enters state, so its tool calls can
+        # never be executed: no message anywhere carries call-x.
+        command_messages = result.command.update["messages"]
+        self.assertFalse(
+            any(
+                call["id"] == "call-x"
+                for message in [*handled[0].messages, *command_messages]
+                if isinstance(message, AIMessage)
+                for call in (message.tool_calls or [])
+            )
+        )
+        event = pop_turn_events("structured-toolcall")[0]
+        self.assertEqual(event["summary_attempts"], 1)
+        self.assertEqual(event["summary_retry_reasons"], [])
+        self.assertEqual(
+            pop_turn_signals("structured-toolcall")["compactions_this_turn"], 1
+        )
+
+    def test_real_agent_structured_compaction_rewrites_checkpoint_once(self) -> None:
+        def response(text: str, input_tokens: int, cache_read: int) -> AIMessage:
+            return AIMessage(
+                content=text,
+                usage_metadata={
+                    "input_tokens": input_tokens,
+                    "output_tokens": 10,
+                    "total_tokens": input_tokens + 10,
+                    "input_token_details": {"cache_read": cache_read},
+                },
+                response_metadata={"model_name": "deepseek-v4-flash"},
+            )
+
+        model = FakeMessagesListChatModel(
+            responses=[
+                response("structured checkpoint", 6_000, 5_500),
+                response("answer after checkpoint", 700, 0),
+            ]
+        )
+        compactor = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+            summary_input_guard_tokens=900_000,
+        )
+        agent = create_agent(
+            model=model,
+            tools=[],
+            system_prompt="stable system",
+            middleware=[DeepSeekCacheIsolationMiddleware(900_000), compactor],
+        )
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(
+                content=f"old-{index}:" + "x" * 1_000
+            )
+            for index in range(28)
+        ]
+        handler = TurnUsageHandler()
+        reset_turn_signals("structured-integration")
+        result = agent.invoke(
+            {"messages": messages},
+            config={"callbacks": [handler]},
+            context=AppContext(
+                allowed_sources=(),
+                kb_session_id="structured-integration",
+                cache_user_id="eval-structured-integration",
+            ),
+        )
+
+        self.assertEqual(handler.llm_calls, 2)
+        self.assertEqual(
+            [call["source"] for call in handler.model_calls],
+            ["summarization", "agent"],
+        )
+        summaries = [
+            message
+            for message in result["messages"]
+            if message.additional_kwargs.get("lc_source") == "summarization"
+        ]
+        self.assertEqual(len(summaries), 1)
+        self.assertIn("structured checkpoint", summaries[0].content)
+        answers = [
+            message
+            for message in result["messages"]
+            if isinstance(message, AIMessage)
+            and message.content == "answer after checkpoint"
+        ]
+        self.assertEqual(len(answers), 1)
+        self.assertFalse(
+            any(
+                message.content == messages[0].content for message in result["messages"]
+            )
+        )
+        event = pop_turn_events("structured-integration")[0]
+        self.assertEqual(event["summary_provider_cache_read_tokens"], 5_500)
+        self.assertEqual(event["summary_provider_cache_miss_tokens"], 500)
+        self.assertEqual(
+            pop_turn_signals("structured-integration")["compactions_this_turn"],
+            1,
+        )
+
+
+class ExperimentCompactionMiddlewareAsyncTests(unittest.IsolatedAsyncioTestCase):
+    async def test_async_empty_summary_is_retried(self) -> None:
+        model = ExperimentCompactionMiddlewareTests.FakeModel(["", "async summary"])
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 100),
+            keep=("tokens", 100),
+            trim_tokens_to_summarize=None,
+        )
+        messages = [HumanMessage(content="x" * 2_000) for _ in range(4)]
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(
+                kb_session_id="async-retry", cache_user_id="eval_async_retry"
+            )
+        )
+        reset_turn_signals("async-retry")
+        with mock.patch("app.chat_service.asyncio.sleep") as sleep:
+            update = await middleware.abefore_model({"messages": messages}, runtime)
+        self.assertIsNotNone(update)
+        self.assertEqual(len(model.prompts), 2)
+        sleep.assert_awaited_once_with(1.0)
+        event = pop_turn_events("async-retry")[0]
+        self.assertEqual(event["summary_attempts"], 2)
+
+
+class CompactionPathEquivalenceTests(unittest.TestCase):
+    """Same history + config: both paths must agree on the compaction boundary."""
+
+    TRIGGER = ("tokens", 1_000)
+    KEEP = ("tokens", 500)
+
+    def _history(self) -> list:
+        # Pre-assigned ids let boundary selection be compared across paths.
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(
+                content=f"m{index}:" + "x" * 1_000, id=f"m{index}"
+            )
+            for index in range(24)
+        ]
+        messages += [
+            AIMessage(
+                content="",
+                id="m-toolcall",
+                tool_calls=[{"name": "lookup", "args": {}, "id": "call-1"}],
+            ),
+            ToolMessage(content="evidence", tool_call_id="call-1", id="m-tool"),
+            AIMessage(content="tool answer", id="m-tool-answer"),
+            HumanMessage(content="current question", id="m-current"),
+        ]
+        return messages
+
+    def test_both_paths_select_the_identical_boundary(self) -> None:
+        history = self._history()
+        xml_plan = InstrumentedSummarizationMiddleware(
+            model=ExperimentCompactionMiddlewareTests.FakeModel(),
+            trigger=self.TRIGGER,
+            keep=self.KEEP,
+            trim_tokens_to_summarize=None,
+        )._plan_compaction({"messages": history})
+        structured_plan = PrefixPreservingCompactionMiddleware(
+            model=ExperimentCompactionMiddlewareTests.StructuredFakeModel(),
+            trigger=self.TRIGGER,
+            keep=self.KEEP,
+            trim_tokens_to_summarize=None,
+        )._plan_compaction({"messages": history})
+        self.assertIsNotNone(xml_plan)
+        self.assertIsNotNone(structured_plan)
+        self.assertEqual(
+            [m.id for m in xml_plan["selected"]],
+            [m.id for m in structured_plan["selected"]],
+        )
+        self.assertEqual(
+            [m.id for m in xml_plan["preserved"]],
+            [m.id for m in structured_plan["preserved"]],
+        )
+        # The boundary is a clean partition of the full history.
+        self.assertEqual(
+            [m.id for m in [*xml_plan["selected"], *xml_plan["preserved"]]],
+            [m.id for m in history],
+        )
+        self.assertGreater(len(xml_plan["selected"]), 0)
+        self.assertGreater(len(xml_plan["preserved"]), 0)
+
+    def test_both_paths_install_the_same_post_compaction_structure(self) -> None:
+        history = self._history()
+
+        xml_middleware = InstrumentedSummarizationMiddleware(
+            model=ExperimentCompactionMiddlewareTests.FakeModel(["xml summary"]),
+            trigger=self.TRIGGER,
+            keep=self.KEEP,
+            trim_tokens_to_summarize=None,
+        )
+        reset_turn_signals("eq-xml")
+        xml_update = xml_middleware.before_model(
+            {"messages": list(history)},
+            SimpleNamespace(
+                context=SimpleNamespace(kb_session_id="eq-xml", cache_user_id="")
+            ),
+        )
+        pop_turn_events("eq-xml")
+        pop_turn_signals("eq-xml")
+
+        structured_model = ExperimentCompactionMiddlewareTests.StructuredFakeModel(
+            ["structured summary"]
+        )
+        structured_middleware = PrefixPreservingCompactionMiddleware(
+            model=structured_model,
+            trigger=self.TRIGGER,
+            keep=self.KEEP,
+            trim_tokens_to_summarize=None,
+        )
+        request = ExperimentCompactionMiddlewareTests._structured_request(
+            structured_model, list(history), "eq-structured"
+        )
+        handled: list[ModelRequest] = []
+
+        def handler(compacted_request):
+            handled.append(compacted_request)
+            return ModelResponse(result=[AIMessage(content="answer", id="a")])
+
+        reset_turn_signals("eq-structured")
+        result = structured_middleware.wrap_model_call(request, handler)
+        pop_turn_events("eq-structured")
+        pop_turn_signals("eq-structured")
+
+        self.assertIsInstance(xml_update["messages"][0], RemoveMessage)
+        self.assertEqual(xml_update["messages"][0].id, REMOVE_ALL_MESSAGES)
+        xml_summary, xml_tail = xml_update["messages"][1], xml_update["messages"][2:]
+        compacted = handled[0].messages
+        structured_summary, structured_tail = compacted[0], compacted[1:]
+
+        for summary in (xml_summary, structured_summary):
+            self.assertIsInstance(summary, HumanMessage)
+            self.assertEqual(
+                summary.additional_kwargs.get("lc_source"), "summarization"
+            )
+            self.assertTrue(
+                summary.content.startswith(
+                    "Here is a summary of the conversation to date:"
+                )
+            )
+        self.assertEqual([m.id for m in xml_tail], [m.id for m in structured_tail])
+        self.assertEqual(
+            [m.content for m in xml_tail], [m.content for m in structured_tail]
+        )
+        # Only the summary text differs between the two paths.
+        self.assertIn("xml summary", xml_summary.content)
+        self.assertIn("structured summary", structured_summary.content)
+        # The structured path's checkpoint command installs the same structure.
+        command_messages = result.command.update["messages"]
+        self.assertIsInstance(command_messages[0], RemoveMessage)
+        self.assertIs(command_messages[1], structured_summary)
+        self.assertEqual(
+            [m.id for m in command_messages[2:-1]],
+            [m.id for m in structured_tail],
+        )
+        self.assertEqual(command_messages[-1].content, "answer")
+
+
+class MultiCompactionTests(unittest.TestCase):
+    def test_xml_second_compaction_replaces_prior_summary_without_orphans(
+        self,
+    ) -> None:
+        model = ExperimentCompactionMiddlewareTests.FakeModel(
+            ["first summary", "second summary"]
+        )
+        middleware = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+        )
+        runtime = SimpleNamespace(
+            context=SimpleNamespace(kb_session_id="multi-xml", cache_user_id="")
+        )
+        history = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(
+                content=f"m{index}:" + "x" * 1_000, id=f"m{index}"
+            )
+            for index in range(24)
+        ]
+        reset_turn_signals("multi-xml")
+        first_update = middleware.before_model({"messages": history}, runtime)
+        self.assertIsNotNone(first_update)
+        summarized_state = list(first_update["messages"][1:])
+
+        second_turn = [
+            HumanMessage(content="new question " + "y" * 3_000, id="n0"),
+            AIMessage(
+                content="",
+                id="n1",
+                tool_calls=[{"name": "lookup", "args": {}, "id": "call-2"}],
+            ),
+            ToolMessage(
+                content="evidence " + "y" * 1_000, tool_call_id="call-2", id="n2"
+            ),
+            AIMessage(content="answer two", id="n3"),
+        ]
+        second_update = middleware.before_model(
+            {"messages": [*summarized_state, *second_turn]}, runtime
+        )
+        self.assertIsNotNone(second_update)
+
+        final_state = list(second_update["messages"][1:])
+        summaries = [
+            message
+            for message in final_state
+            if message.additional_kwargs.get("lc_source") == "summarization"
+        ]
+        self.assertEqual(len(summaries), 1)
+        self.assertIn("second summary", summaries[0].content)
+        self.assertNotIn("first summary", summaries[0].content)
+        # The first summary fed the second summarization instead of surviving.
+        self.assertIn("first summary", model.prompts[1])
+        for index, message in enumerate(final_state):
+            if isinstance(message, ToolMessage):
+                self.assertGreater(index, 0)
+                previous = final_state[index - 1]
+                self.assertIsInstance(previous, AIMessage)
+                self.assertIn(
+                    message.tool_call_id,
+                    [call["id"] for call in previous.tool_calls],
+                )
+        self.assertLessEqual(middleware._partial_token_counter(final_state[1:]), 500)
+        self.assertEqual(len(pop_turn_events("multi-xml")), 2)
+        self.assertEqual(pop_turn_signals("multi-xml")["compactions_this_turn"], 2)
+
+    def test_real_agent_structured_second_compaction_on_summarized_thread(
+        self,
+    ) -> None:
+        def response(text: str) -> AIMessage:
+            return AIMessage(
+                content=text,
+                usage_metadata={
+                    "input_tokens": 1_000,
+                    "output_tokens": 10,
+                    "total_tokens": 1_010,
+                    "input_token_details": {"cache_read": 0},
+                },
+                response_metadata={"model_name": "deepseek-v4-flash"},
+            )
+
+        model = FakeMessagesListChatModel(
+            responses=[
+                response("checkpoint one"),
+                response("answer one"),
+                response("checkpoint two"),
+                response("answer two"),
+            ]
+        )
+        compactor = PrefixPreservingCompactionMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+            summary_input_guard_tokens=900_000,
+        )
+        agent = create_agent(
+            model=model,
+            tools=[],
+            system_prompt="stable system",
+            middleware=[compactor],
+            checkpointer=InMemorySaver(),
+        )
+        config = {"configurable": {"thread_id": "structured-multi"}}
+        first_turn = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(
+                content=f"old-{index}:" + "x" * 1_000
+            )
+            for index in range(28)
+        ]
+        reset_turn_signals("structured-multi-t1")
+        agent.invoke(
+            {"messages": first_turn},
+            config=config,
+            context=AppContext(allowed_sources=(), kb_session_id="structured-multi-t1"),
+        )
+        state = agent.get_state(config).values["messages"]
+        first_summaries = [
+            message
+            for message in state
+            if message.additional_kwargs.get("lc_source") == "summarization"
+        ]
+        self.assertEqual(len(first_summaries), 1)
+        self.assertIn("checkpoint one", first_summaries[0].content)
+        self.assertEqual(len(pop_turn_events("structured-multi-t1")), 1)
+        self.assertEqual(
+            pop_turn_signals("structured-multi-t1")["compactions_this_turn"], 1
+        )
+
+        reset_turn_signals("structured-multi-t2")
+        agent.invoke(
+            {"messages": [HumanMessage(content="second wave " + "y" * 6_000)]},
+            config=config,
+            context=AppContext(allowed_sources=(), kb_session_id="structured-multi-t2"),
+        )
+        state = agent.get_state(config).values["messages"]
+        summaries = [
+            message
+            for message in state
+            if message.additional_kwargs.get("lc_source") == "summarization"
+        ]
+        self.assertEqual(len(summaries), 1)
+        self.assertIn("checkpoint two", summaries[0].content)
+        contents = [str(message.content) for message in state]
+        self.assertFalse(any("checkpoint one" in content for content in contents))
+        self.assertFalse(any("old-0:" in content for content in contents))
+        self.assertFalse(any(isinstance(m, ToolMessage) for m in state))
+        # Retained tail survives verbatim, followed by the new answer.
+        self.assertTrue(any(content.startswith("second wave") for content in contents))
+        self.assertEqual(contents.count("answer two"), 1)
+        event = pop_turn_events("structured-multi-t2")[0]
+        self.assertEqual(event["summary_strategy"], "structured_prefix")
+        self.assertEqual(event["summary_instruction_retained_messages"], 1)
+        self.assertEqual(
+            pop_turn_signals("structured-multi-t2")["compactions_this_turn"], 1
+        )
+
+
+class TurnUsageAccountingInvariantTests(unittest.TestCase):
+    def test_model_call_rows_sum_to_the_billed_usage_totals(self) -> None:
+        # est_cost_usd is computed from usage_by_model; the per-call rows are
+        # the explanation. If a call were double-counted (or dropped) on either
+        # side, the two aggregates would disagree.
+        def response(
+            text: str, input_tokens: int, cache_read: int, cache_creation: int
+        ) -> AIMessage:
+            return AIMessage(
+                content=text,
+                usage_metadata={
+                    "input_tokens": input_tokens,
+                    "output_tokens": 40,
+                    "total_tokens": input_tokens + 40,
+                    "input_token_details": {
+                        "cache_read": cache_read,
+                        "cache_creation": cache_creation,
+                    },
+                },
+                response_metadata={"model_name": "deepseek-v4-flash"},
+            )
+
+        model = FakeMessagesListChatModel(
+            responses=[
+                response("summary", 6_000, 5_000, 500),
+                response("answer", 700, 100, 50),
+            ]
+        )
+        summary = InstrumentedSummarizationMiddleware(
+            model=model,
+            trigger=("tokens", 1_000),
+            keep=("tokens", 500),
+            trim_tokens_to_summarize=None,
+        )
+        agent = create_agent(model=model, tools=[], middleware=[summary])
+        messages = [
+            (HumanMessage if index % 2 == 0 else AIMessage)(content="x" * 1_000)
+            for index in range(28)
+        ]
+        handler = TurnUsageHandler()
+        reset_turn_signals("usage-invariant")
+        agent.invoke(
+            {"messages": messages},
+            config={"callbacks": [handler]},
+            context=AppContext(allowed_sources=(), kb_session_id="usage-invariant"),
+        )
+        pop_turn_events("usage-invariant")
+        pop_turn_signals("usage-invariant")
+
+        self.assertEqual(handler.llm_calls, 2)
+        self.assertEqual(len(handler.model_calls), 2)
+        self.assertEqual(
+            sorted(call["source"] for call in handler.model_calls),
+            ["agent", "summarization"],
+        )
+        totals = usage_totals(handler.usage_metadata)
+        summed = {
+            field: sum(call[field] for call in handler.model_calls)
+            for field in (
+                "input_tokens",
+                "output_tokens",
+                "total_tokens",
+                "cache_read_tokens",
+                "cache_creation_tokens",
+            )
+        }
+        self.assertEqual(summed, totals)
+        # Known scripted usage pins the absolute numbers, not just consistency.
+        self.assertEqual(totals["input_tokens"], 6_700)
+        self.assertEqual(totals["output_tokens"], 80)
+        self.assertEqual(totals["cache_read_tokens"], 5_100)
+        self.assertEqual(totals["cache_creation_tokens"], 550)
+        estimated = estimate_cost_usd(handler.usage_metadata)
+        self.assertIsNotNone(estimated)
+        self.assertAlmostEqual(
+            estimated,
+            sum(call["cost"]["total_usd"] for call in handler.model_calls),
+        )
+
+
 class PromptCompressionTests(unittest.TestCase):
     def test_collapses_whitespace_and_reports(self) -> None:
         reset_turn_signals("t1")

tests/test_telemetry.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import unittest
+from uuid import uuid4
 
 from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
 from langchain_core.outputs import ChatGeneration, LLMResult
@@ -11,9 +12,12 @@ from app.telemetry import (
     TurnUsageHandler,
     context_window_stats,
     estimate_cost_usd,
+    aggregate_cost_breakdown,
+    pop_turn_events,
     pop_turn_signals,
     pricing_for_model,
     record_turn_signal,
+    record_turn_event,
     record_turn_signal_max,
     reset_turn_signals,
     usage_totals,
@@ -93,6 +97,27 @@ class CostEstimateTests(unittest.TestCase):
         }
         self.assertIsNone(estimate_cost_usd(mixed))
 
+    def test_deepseek_cost_breakdown_is_mutually_exclusive(self) -> None:
+        usage = {
+            "deepseek-v4-flash": {
+                "input_tokens": 1_000_000,
+                "output_tokens": 100_000,
+                "input_token_details": {"cache_read": 900_000},
+            }
+        }
+        breakdown = aggregate_cost_breakdown(usage)
+        self.assertIsNotNone(breakdown)
+        self.assertAlmostEqual(breakdown["cache_miss_input_usd"], 0.014)
+        self.assertAlmostEqual(breakdown["cache_read_input_usd"], 0.00252)
+        self.assertAlmostEqual(breakdown["output_usd"], 0.028)
+        self.assertAlmostEqual(
+            breakdown["total_usd"],
+            breakdown["cache_miss_input_usd"]
+            + breakdown["cache_read_input_usd"]
+            + breakdown["cache_creation_input_usd"]
+            + breakdown["output_usd"],
+        )
+
 
 class ContextWindowStatsTests(unittest.TestCase):
     def test_counts_summaries_and_cleared_tool_outputs(self) -> None:
@@ -145,6 +170,108 @@ class TurnUsageHandlerTests(unittest.TestCase):
         self.assertEqual(usage["input_tokens"], 150)
         self.assertEqual(usage["output_tokens"], 30)
 
+    def test_records_one_explanatory_row_per_call(self) -> None:
+        handler = TurnUsageHandler()
+        run_id = uuid4()
+        handler.on_chat_model_start(
+            {},
+            [[HumanMessage(content="hello")]],
+            run_id=run_id,
+            metadata={"lc_source": "summarization"},
+        )
+        message = AIMessage(
+            content="summary",
+            usage_metadata={
+                "input_tokens": 100,
+                "output_tokens": 20,
+                "total_tokens": 120,
+                "input_token_details": {"cache_read": 80},
+            },
+            response_metadata={"model_name": "deepseek-v4-flash"},
+        )
+        handler.on_llm_end(
+            LLMResult(generations=[[ChatGeneration(message=message)]]),
+            run_id=run_id,
+        )
+        call = handler.model_calls[0]
+        self.assertEqual(call["source"], "summarization")
+        self.assertEqual(call["cache_read_tokens"], 80)
+        self.assertEqual(call["cache_miss_tokens"], 20)
+        self.assertTrue(call["cache_details_reported"])
+        self.assertGreater(call["request_context_tokens_approx"], 0)
+        self.assertAlmostEqual(
+            call["cost"]["total_usd"],
+            estimate_cost_usd({"deepseek-v4-flash": message.usage_metadata}),
+        )
+
+
+class LangchainOpenAICacheFieldContractTests(unittest.TestCase):
+    """Pin the langchain-openai usage conversion our cache accounting rides on.
+
+    DeepSeek (and OpenAI) report cached prompt tokens as
+    ``prompt_tokens_details.cached_tokens``; langchain-openai must surface that
+    as ``usage_metadata.input_token_details.cache_read`` or TurnUsageHandler's
+    cache buckets (and the ~50x DeepSeek cache-read discount) silently read 0.
+    """
+
+    def _convert(self, payload: dict) -> dict:
+        try:
+            from langchain_openai.chat_models.base import _create_usage_metadata
+        except ImportError as exc:
+            self.fail(
+                "langchain_openai.chat_models.base._create_usage_metadata is no "
+                f"longer importable ({exc}). A langchain-openai upgrade moved "
+                "the usage conversion; re-pin the prompt_tokens_details."
+                "cached_tokens -> input_token_details.cache_read mapping "
+                "against its new location."
+            )
+        return _create_usage_metadata(payload)
+
+    def test_cached_tokens_map_to_cache_read_details(self) -> None:
+        usage = self._convert(
+            {
+                "prompt_tokens": 1_000,
+                "completion_tokens": 100,
+                "total_tokens": 1_100,
+                "prompt_tokens_details": {"cached_tokens": 900},
+            }
+        )
+        # LangChain convention: input_tokens INCLUDES the cached bucket; the
+        # cost code carves cache_read out instead of adding it on top.
+        self.assertEqual(usage["input_tokens"], 1_000)
+        self.assertEqual(usage["output_tokens"], 100)
+        self.assertEqual(usage["input_token_details"]["cache_read"], 900)
+
+    def test_converted_usage_flows_through_turn_usage_handler(self) -> None:
+        usage = self._convert(
+            {
+                "prompt_tokens": 1_000,
+                "completion_tokens": 100,
+                "total_tokens": 1_100,
+                "prompt_tokens_details": {"cached_tokens": 900},
+            }
+        )
+        handler = TurnUsageHandler()
+        run_id = uuid4()
+        handler.on_chat_model_start({}, [[HumanMessage(content="q")]], run_id=run_id)
+        message = AIMessage(
+            content="answer",
+            usage_metadata=usage,
+            response_metadata={"model_name": "deepseek-v4-flash"},
+        )
+        handler.on_llm_end(
+            LLMResult(generations=[[ChatGeneration(message=message)]]),
+            run_id=run_id,
+        )
+        call = handler.model_calls[0]
+        self.assertEqual(call["cache_read_tokens"], 900)
+        self.assertEqual(call["cache_miss_tokens"], 100)
+        self.assertTrue(call["cache_details_reported"])
+        # deepseek-v4-flash: $0.14 miss / $0.0028 cache-read / $0.28 output
+        # per MTok, so the cache discount must show up in the billed cost.
+        expected = (100 * 0.14 + 900 * 0.0028 + 100 * 0.28) / 1_000_000
+        self.assertAlmostEqual(call["cost"]["total_usd"], expected)
+
 
 class TurnSignalRegistryTests(unittest.TestCase):
     def test_accumulates_and_pops_per_turn(self) -> None:
@@ -158,6 +285,15 @@ class TurnSignalRegistryTests(unittest.TestCase):
         # Popping clears the entry: a second pop is empty.
         self.assertEqual(pop_turn_signals("turn-a"), {})
 
+    def test_structured_events_are_isolated_and_popped(self) -> None:
+        reset_turn_signals("turn-events")
+        record_turn_event("turn-events", {"event": "summarization", "tokens": 9})
+        self.assertEqual(
+            pop_turn_events("turn-events"),
+            [{"event": "summarization", "tokens": 9}],
+        )
+        self.assertEqual(pop_turn_events("turn-events"), [])
+
     def test_turns_are_isolated_and_noops_are_ignored(self) -> None:
         reset_turn_signals("turn-x")
         reset_turn_signals("turn-y")