feat(runtime): replace per-round history trimming with threshold-triggered LLM compaction

Align native context management with the Claude Code / Codex model:
entry-capped tool results, history frozen below the threshold, one
high-quality summary at the wall — instead of the old pipeline that
microcompacted old messages from 60% usage and hid everything past 40
messages behind a snip marker with no summary.

- context pipeline: history below the hard threshold is never rewritten
  (model quality and prompt-cache prefixes depend on byte-identical old
  messages); the 60% tool-aware microcompact and the 40-message history
  snip move to an emergency-only fallback used under overflow pressure
  when the summarizer is unavailable or circuit-broken.
- durable compaction (was a stub): at usage >= context_guard.hard_threshold
  (now 0.90, soft_threshold removed) the old span is folded into a
  9-section summary via the new HistoryCompactor.summarize_runtime_history,
  keeping the system head, the seed user request verbatim on every round
  (injected session-memory/artifact messages shift the stale
  base_prefix_len, so the fold start is structure-aware), and a
  pairing-safe recent tail. A previous summary stays foldable, so exactly
  one summary exists at a time and rounds chain.
- token accounting anchors on the provider-reported prompt size of the
  latest request (max with the local estimate).
- reactive_compaction.circuit_breaker_failures (previously unread) now
  stops repeated summarizer failures; provider overflow errors retry
  through the same pipeline, summary-first.
- tool-result budget clip keeps head and tail instead of tail-chopping.
- chat-side transcripts get the same treatment: new
  MemoryManager.maybe_compact_session_history wires the threshold-gated
  maybe_compact_session into secretary, office_ui dispatcher, and
  context_loader before prompt building, closing the unbounded-growth
  path; dead no-op compactor entries (maybe_compact_after_message,
  should_compact_prompt) removed.

Verified by 13 new tests (history sanctity below threshold, multi-round
single-summary/seed-verbatim/chain invariants, breaker, emergency
fallback, provider-overflow end-to-end recovery) plus a live-provider
probe: multi-round compaction with the model completing correctly from
summarized context. Full suite: 1859 passed.

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

opc/core/config.py

@@ -585,8 +585,9 @@ class StreamRenderingConfig(BaseModel):
 
 class ContextGuardConfig(BaseModel):
     enabled: bool = True
-    soft_threshold: float = 0.60
-    hard_threshold: float = 0.80
+    # Below this usage ratio the history is never rewritten; at or above it
+    # the runtime folds old messages into one LLM summary (durable compaction).
+    hard_threshold: float = 0.90
     warn_remaining_pct: int = 15
     tool_output_char_budget: int = 12_000
     shell_stdout_char_budget: int = 12_000

opc/layer1_perception/context_loader.py

@@ -91,6 +91,10 @@ class ContextLoader:
             session_id=None,
             include_project_knowledge=include_project_knowledge,
         )
+        if session_id:
+            maybe_compact = getattr(self.memory, "maybe_compact_session_history", None)
+            if callable(maybe_compact):
+                await maybe_compact(session_id, project_id=project_id)
         ctx.session_memory = (
             await self.memory.build_session_prompt_context(
                 session_id,

opc/layer2_organization/secretary.py

@@ -103,6 +103,9 @@ class SecretaryService:
     async def _build_prompt(self, content: str, project_id: str | None, session_id: str) -> str:
         policy_summary = self.policies.summarize_policies(project_id=project_id)
         project_knowledge = await self.memory.build_project_knowledge_context(project_id=project_id)
+        maybe_compact = getattr(self.memory, "maybe_compact_session_history", None)
+        if callable(maybe_compact):
+            await maybe_compact(session_id, project_id=project_id)
         session_history = await self.memory.build_session_prompt_context(
             session_id,
             include_latest_user_turn=False,

opc/layer3_agent/runtime_v2/runtime.py

@@ -264,6 +264,7 @@ class NativeRuntimeV2:
 
         total_cost = 0.0
         total_usage: dict[str, int] = {"prompt_tokens": 0, "completion_tokens": 0}
+        last_observed_prompt_tokens = 0
         aggregated_artifacts: dict[str, Any] = {}
         overflow_retries = 0
         max_overflow_retries = max(
@@ -372,6 +373,7 @@ class NativeRuntimeV2:
                 todo_state=todo_state,
                 runtime_notes=runtime_notes,
                 active_subagents=subagents.list_agents().get("agents", []),
+                observed_tokens=last_observed_prompt_tokens,
             )
             context_usage = await self._emit_context_usage(
                 runtime_session_id=runtime_session_id,
@@ -476,6 +478,8 @@ class NativeRuntimeV2:
                         prompt_tokens = int(event.payload.get("prompt_tokens", 0) or 0)
                         completion_tokens = int(event.payload.get("completion_tokens", 0) or 0)
                         estimated_cost_delta = float(event.payload.get("estimated_cost_delta", 0.0) or 0.0)
+                        if prompt_tokens:
+                            last_observed_prompt_tokens = prompt_tokens
                         total_usage["prompt_tokens"] += prompt_tokens
                         total_usage["completion_tokens"] += completion_tokens
                         total_cost += estimated_cost_delta
@@ -546,6 +550,7 @@ class NativeRuntimeV2:
                         todo_state=todo_state,
                         runtime_notes=runtime_notes,
                         active_subagents=subagents.list_agents().get("agents", []),
+                        observed_tokens=last_observed_prompt_tokens,
                     )
                     continue
                 recovered_turn = await self._recover_tool_protocol_stream_error(
@@ -1803,21 +1808,47 @@ class NativeRuntimeV2:
         runtime_notes: dict[str, Any],
         active_subagents: list[dict[str, Any]],
         force_compact: bool = False,
+        observed_tokens: int = 0,
     ) -> list[dict[str, Any]]:
+        # History below the hard threshold is never rewritten: model quality
+        # and prompt-cache prefixes both depend on old messages staying
+        # byte-identical. The only routine mutation is the idempotent
+        # per-message tool-result budget (same clip an entry already got).
         bounded = self._apply_tool_result_budget(messages)
-        apply_soft_compaction = force_compact or self._should_apply_soft_compaction(bounded, tool_schemas)
-        microcompacted = self._apply_tool_aware_microcompact(bounded, base_prefix_len) if apply_soft_compaction else bounded
-        compacted = await self._apply_durable_compaction(
-            microcompacted,
-            tool_schemas=tool_schemas,
-            task=task,
-            force_compact=force_compact or self._should_apply_hard_compaction(microcompacted, tool_schemas),
+        pipeline_steps = ["tool_result_budgeting"]
+        compacted = bounded
+        durable_applied = False
+        wants_compaction = force_compact or self._should_apply_hard_compaction(
+            bounded, tool_schemas, observed_tokens=observed_tokens
         )
+        if wants_compaction:
+            breaker_limit = max(
+                1,
+                int(self.config.system.native_runtime.reactive_compaction.circuit_breaker_failures or 2),
+            )
+            failures = int(runtime_notes.get("durable_compaction_failures", 0) or 0)
+            if failures < breaker_limit:
+                compacted, durable_applied = await self._apply_durable_compaction(
+                    bounded,
+                    task=task,
+                    base_prefix_len=base_prefix_len,
+                    runtime_session_id=runtime_session_id,
+                )
+                if durable_applied:
+                    pipeline_steps.append("durable_compaction")
+                    runtime_notes["durable_compaction_failures"] = 0
+                else:
+                    runtime_notes["durable_compaction_failures"] = failures + 1
+            if not durable_applied and force_compact:
+                # Emergency-only mechanical fallback: overflow pressure with
+                # the summarizing compactor unavailable or circuit-broken.
+                compacted = self._apply_tool_aware_microcompact(compacted, base_prefix_len)
+                pipeline_steps.append("emergency_microcompact")
         if compacted != bounded:
             boundary_record = {
                 "summary": "Runtime V2 context pipeline compacted persisted history.",
                 "message_count": len(compacted),
-                "pipeline": ["tool_result_budgeting", "tool_aware_microcompact", "durable_compaction", "session_memory_reinjection"],
+                "pipeline": [*pipeline_steps, "session_memory_reinjection"],
             }
             compaction_boundaries.append(boundary_record)
             store = getattr(self.memory_manager, "store", None)
@@ -1856,9 +1887,18 @@ class NativeRuntimeV2:
             if message.get("role") == "tool":
                 content = str(message.get("content", "") or "")
                 if len(content) > budget:
+                    # Keep head and tail: openings carry the command/context,
+                    # endings carry the verdict (exit codes, tracebacks).
+                    head = max(1, budget // 2)
+                    tail = max(0, budget - head)
+                    omitted = len(content) - head - tail
                     compacted.append({
                         **message,
-                        "content": content[:budget] + "\n[tool result truncated by runtime_v2]",
+                        "content": (
+                            content[:head]
+                            + f"\n[tool result truncated by runtime_v2: {omitted} chars omitted]\n"
+                            + (content[-tail:] if tail else "")
+                        ),
                     })
                     continue
             compacted.append(message)
@@ -2007,18 +2047,106 @@ class NativeRuntimeV2:
             })
         return compacted
 
+    _DURABLE_COMPACTION_MARKER = "[runtime_v2 durable compaction]"
+
     async def _apply_durable_compaction(
         self,
         messages: list[dict[str, Any]],
         *,
-        tool_schemas: list[dict[str, Any]] | None,
         task: Task | None,
-        force_compact: bool = False,
-    ) -> list[dict[str, Any]]:
-        _ = tool_schemas
-        _ = task
-        _ = force_compact
-        return messages
+        base_prefix_len: int,
+        runtime_session_id: str,
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """Fold old messages into one LLM summary, keeping prefix and tail.
+
+        Returns (messages, applied). On any summarizer failure the original
+        list is returned unchanged so the caller can count failures and the
+        model keeps seeing the full history for this round.
+        """
+        compactor = self.history_compactor
+        summarize = getattr(compactor, "summarize_runtime_history", None) if compactor else None
+        if not callable(summarize):
+            return messages, False
+        preserve_recent = max(
+            4,
+            int(self.config.system.native_runtime.tool_aware_microcompact.preserve_recent_messages or 8),
+        )
+        start = max(base_prefix_len, len(messages) - preserve_recent)
+        # Never split an assistant tool_calls message from its tool results.
+        while start > base_prefix_len and str(messages[start].get("role", "") or "") == "tool":
+            start -= 1
+        # base_prefix_len goes stale once session-memory/artifact messages are
+        # injected into the prefix region, shifting real prefix messages past
+        # the boundary. Never fold the system head, and keep the seed user
+        # request verbatim on every round (Codex-style): a previous summary is
+        # a user message too, but carries the marker and must stay foldable so
+        # exactly one summary exists at a time.
+        fold_start = base_prefix_len
+        while fold_start < start and str(messages[fold_start].get("role", "") or "") == "system":
+            fold_start += 1
+        if (
+            fold_start < start
+            and str(messages[fold_start].get("role", "") or "") == "user"
+            and self._DURABLE_COMPACTION_MARKER not in str(messages[fold_start].get("content", "") or "")
+            and not any(str(item.get("role", "") or "") == "user" for item in messages[:fold_start])
+        ):
+            fold_start += 1
+        folded = messages[fold_start:start]
+        if len(folded) < 4:
+            return messages, False
+        try:
+            summary = await summarize(
+                project_id=str(getattr(task, "project_id", "") or ""),
+                session_id=runtime_session_id,
+                messages=self._render_messages_for_compaction(folded),
+            )
+        except Exception as exc:
+            logger.warning(f"Durable compaction failed; keeping full history this round: {exc}")
+            return messages, False
+        summary_text = str(summary or "").strip()
+        if not summary_text:
+            return messages, False
+        summary_message = {
+            "role": "user",
+            "content": (
+                f"{self._DURABLE_COMPACTION_MARKER} Earlier conversation was compacted to stay "
+                "within the context window. Continue seamlessly from this summary; the full "
+                "transcript remains persisted and queryable.\n\n" + summary_text
+            ),
+        }
+        return [*messages[:fold_start], summary_message, *messages[start:]], True
+
+    def _render_messages_for_compaction(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        per_message_budget = 4_000
+        rendered: list[dict[str, Any]] = []
+        for message in messages:
+            role = str(message.get("role", "") or "assistant")
+            content = str(message.get("content", "") or "")
+            tool_calls = message.get("tool_calls") or []
+            if tool_calls:
+                names: list[str] = []
+                for call in tool_calls:
+                    if not isinstance(call, dict):
+                        continue
+                    function = call.get("function", "")
+                    name = function.get("name", "") if isinstance(function, dict) else str(function or "")
+                    if name:
+                        names.append(str(name))
+                if names:
+                    content = (content + "\n[called tools: " + ", ".join(names) + "]").strip()
+            if role == "tool":
+                content = f"[tool result {str(message.get('tool_call_id', '') or '')}] {content}".strip()
+            if len(content) > per_message_budget:
+                head = per_message_budget // 2
+                tail = per_message_budget - head - 100
+                content = (
+                    content[:head]
+                    + f"\n[{len(content) - head - tail} chars omitted]\n"
+                    + content[-tail:]
+                )
+            if content:
+                rendered.append({"role": role, "content": content})
+        return rendered
 
     async def _reinject_session_memory(
         self,
@@ -3624,8 +3752,16 @@ class NativeRuntimeV2:
         self,
         messages: list[dict[str, Any]],
         tool_schemas: list[dict[str, Any]] | None,
+        *,
+        observed_tokens: int = 0,
     ) -> dict[str, Any]:
-        token_count = self._safe_count_input_tokens(messages, tool_schemas)
+        # Anchor on the provider-reported prompt size of the latest request
+        # when it exceeds the local estimate: the context only grows within a
+        # turn, so max() protects against estimator undercounting.
+        token_count = max(
+            self._safe_count_input_tokens(messages, tool_schemas),
+            int(observed_tokens or 0),
+        )
         context_window = self._context_window_limit()
         remaining_tokens = max(0, context_window - token_count) if context_window > 0 else 0
         remaining_pct = int((remaining_tokens / context_window) * 100) if context_window > 0 else 0
@@ -3637,35 +3773,23 @@ class NativeRuntimeV2:
             "context_remaining_tokens": remaining_tokens,
             "context_remaining_pct": remaining_pct,
             "usage_ratio": round(usage_ratio, 4),
-            "soft_threshold": float(self.config.system.native_runtime.context_guard.soft_threshold or 0.60),
-            "hard_threshold": float(self.config.system.native_runtime.context_guard.hard_threshold or 0.80),
+            "hard_threshold": float(self.config.system.native_runtime.context_guard.hard_threshold or 0.90),
         }
 
-    def _should_apply_soft_compaction(
-        self,
-        messages: list[dict[str, Any]],
-        tool_schemas: list[dict[str, Any]] | None,
-    ) -> bool:
-        config = self.config.system.native_runtime.context_guard
-        if not config.enabled:
-            return True
-        payload = self._context_usage_payload(messages, tool_schemas)
-        if payload["context_window"] <= 0:
-            return len(messages) > self.config.system.native_runtime.history_snip_trigger_messages
-        return float(payload["usage_ratio"]) >= float(config.soft_threshold or 0.60)
-
     def _should_apply_hard_compaction(
         self,
         messages: list[dict[str, Any]],
         tool_schemas: list[dict[str, Any]] | None,
+        *,
+        observed_tokens: int = 0,
     ) -> bool:
         config = self.config.system.native_runtime.context_guard
         if not config.enabled:
             return False
-        payload = self._context_usage_payload(messages, tool_schemas)
+        payload = self._context_usage_payload(messages, tool_schemas, observed_tokens=observed_tokens)
         if payload["context_window"] <= 0:
             return False
-        return float(payload["usage_ratio"]) >= float(config.hard_threshold or 0.80)
+        return float(payload["usage_ratio"]) >= float(config.hard_threshold or 0.90)
 
     def _clip_tool_result_for_history(
         self,

opc/layer5_memory/history_compactor.py

@@ -13,7 +13,6 @@ from opc.core.models import (
     AgentMemorySnapshotRecord,
     SessionCompactionRecord,
     SessionMemorySnapshotRecord,
-    SessionMessageRecord,
 )
 
 
@@ -39,9 +38,55 @@ class HistoryCompactor:
         self.task_type = task_type
         self.compression_threshold = compression_threshold
 
-    async def maybe_compact_after_message(self, message: SessionMessageRecord) -> None:
-        _ = message
-        return
+    async def summarize_runtime_history(
+        self,
+        *,
+        project_id: str,
+        session_id: str,
+        messages: list[dict[str, Any]],
+        existing_summary: str = "",
+    ) -> str:
+        """Summarize in-memory runtime messages for durable context compaction.
+
+        Used by NativeRuntimeV2 when live context reaches the hard threshold:
+        the returned summary replaces the folded span of the message list.
+        Raises on non-recoverable LLM errors so the caller can count failures.
+        """
+        if not messages:
+            return ""
+        if not self.llm:
+            return self._fallback_session_summary(messages, existing_summary)["history_summary"]
+        payload = {
+            "project_id": project_id,
+            "session_id": session_id,
+            "existing_summary": existing_summary,
+            "messages": messages,
+        }
+        raw = await self._simple_chat_with_retry(
+            payload=payload,
+            system=(
+                "You are compacting the live working context of an agent that must "
+                "continue its task seamlessly from your output.\n"
+                "Return strict JSON with a single key `history_summary`.\n"
+                "`history_summary` must be detailed markdown with sections:\n"
+                "1. Primary Request and Intent\n"
+                "2. Key Technical Concepts\n"
+                "3. Files and Code Sections\n"
+                "4. Errors and Fixes (especially user corrections)\n"
+                "5. Problem Solving\n"
+                "6. All User Messages\n"
+                "7. Pending Tasks\n"
+                "8. Current Work\n"
+                "9. Next Step\n"
+                "Quote exact identifiers, paths, commands, and values the agent will "
+                "need to continue; do not invent details."
+            ),
+        )
+        parsed = self._parse_json_response(raw)
+        summary = str((parsed or {}).get("history_summary", "")).strip()
+        if summary:
+            return summary
+        return self._fallback_session_summary(messages, existing_summary)["history_summary"]
 
     async def maybe_compact_session(
         self,
@@ -276,19 +321,6 @@ class HistoryCompactor:
             threshold = max(0, threshold - reserve_tokens)
         return threshold
 
-    def should_compact_prompt(
-        self,
-        messages: list[dict[str, Any]],
-        *,
-        tools: list[dict[str, Any]] | None = None,
-        force: bool = False,
-        reserve_tokens: int = 0,
-    ) -> bool:
-        _ = messages
-        _ = tools
-        _ = force
-        _ = reserve_tokens
-        return False
 
     def _is_context_overflow_error(self, error: Exception) -> bool:
         detector = getattr(self.llm, "is_context_overflow_error", None)

opc/layer5_memory/memory_manager.py

@@ -65,6 +65,33 @@ class MemoryManager:
     def set_history_compactor(self, compactor: Any | None) -> None:
         self.history_compactor = compactor
 
+    async def maybe_compact_session_history(
+        self,
+        session_id: str,
+        project_id: str | None = None,
+    ) -> bool:
+        """Threshold-gated session-transcript compaction.
+
+        Chat-style callers invoke this before building prompt context so a
+        long transcript is folded into a summary snapshot instead of growing
+        without bound. Best-effort: failures never block prompt building.
+        """
+        compactor = self.history_compactor
+        maybe_compact = getattr(compactor, "maybe_compact_session", None) if compactor else None
+        if not callable(maybe_compact) or not session_id:
+            return False
+        try:
+            return bool(
+                await maybe_compact(
+                    project_id=self._resolve_project_id(project_id),
+                    session_id=session_id,
+                    force=False,
+                )
+            )
+        except Exception as exc:
+            logger.debug(f"Session history compaction skipped: {exc}")
+            return False
+
     def _resolve_project_id(self, project_id: str | None = None) -> str:
         return str(project_id or self.project_id or "default")
 

opc/plugins/office_ui/dispatcher.py

@@ -315,6 +315,9 @@ class Dispatcher:
         history_context = ""
         if self.engine.memory and session_id:
             try:
+                maybe_compact = getattr(self.engine.memory, "maybe_compact_session_history", None)
+                if callable(maybe_compact):
+                    await maybe_compact(session_id)
                 history_context = await self.engine.memory.build_session_prompt_context(
                     session_id,
                     include_latest_user_turn=False,

tests/test_runtime_context_compaction.py

@@ -0,0 +1,601 @@
+"""Runtime V2 context compaction behavior.
+
+Contract (aligned with the claude-code / codex reference implementations):
+history below the hard threshold is never rewritten; at the threshold the
+old span is folded into one durable LLM summary with the recent tail kept
+verbatim; mechanical trimming is an emergency-only fallback under overflow
+pressure when the summarizer is unavailable.
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+from opc.core.config import OPCConfig
+from opc.core.models import Task, TaskStatus
+from opc.layer3_agent.runtime_v2.runtime import NativeRuntimeV2
+from opc.layer4_tools.registry import ToolDefinition, ToolRegistry
+from opc.layer5_memory.history_compactor import HistoryCompactor
+from opc.layer5_memory.memory_manager import MemoryManager
+
+
+class _CountingLLM:
+    """Minimal LLM stub with controllable token accounting."""
+
+    def __init__(self, *, token_count: int = 0, context_window: int = 100_000) -> None:
+        self.token_count = token_count
+        self.context_window = context_window
+        self.config = type("Cfg", (), {"max_tokens": 2048})()
+
+    def count_input_tokens(self, messages, tools=None):
+        _ = (messages, tools)
+        return self.token_count
+
+    def get_context_window(self):
+        return self.context_window
+
+    def is_context_overflow_error(self, error: Exception) -> bool:
+        _ = error
+        return False
+
+
+def _runtime(llm, *, compactor=None) -> NativeRuntimeV2:
+    return NativeRuntimeV2(
+        llm=llm,
+        tool_registry=ToolRegistry(),
+        config=OPCConfig(),
+        history_compactor=compactor,
+    )
+
+
+def _paired_tool_round(index: int) -> list[dict[str, object]]:
+    call_id = f"call-{index}"
+    return [
+        {
+            "role": "assistant",
+            "content": f"step {index}",
+            "tool_calls": [
+                {"id": call_id, "type": "function", "function": {"name": "demo", "arguments": "{}"}}
+            ],
+        },
+        {"role": "tool", "tool_call_id": call_id, "content": f"tool output {index} " + ("x" * 200)},
+    ]
+
+
+def _long_history(rounds: int) -> list[dict[str, object]]:
+    messages: list[dict[str, object]] = [{"role": "system", "content": "system prompt"}]
+    messages.append({"role": "user", "content": "original request"})
+    for index in range(rounds):
+        messages.extend(_paired_tool_round(index))
+    return messages
+
+
+class ContextPipelineTests(unittest.IsolatedAsyncioTestCase):
+    async def _pipeline(
+        self,
+        runtime: NativeRuntimeV2,
+        messages: list[dict[str, object]],
+        *,
+        boundaries: list[dict[str, object]] | None = None,
+        runtime_notes: dict[str, object] | None = None,
+        force: bool = False,
+        observed: int = 0,
+    ) -> list[dict[str, object]]:
+        return await runtime._apply_context_pipeline(
+            messages,
+            tool_schemas=None,
+            task=None,
+            base_prefix_len=1,
+            runtime_session_id="rt-test",
+            compaction_boundaries=boundaries if boundaries is not None else [],
+            todo_state=[],
+            runtime_notes=runtime_notes if runtime_notes is not None else {},
+            active_subagents=[],
+            force_compact=force,
+            observed_tokens=observed,
+        )
+
+    @staticmethod
+    def _originals_in(result: list[dict[str, object]], originals: list[dict[str, object]]) -> list[dict[str, object]]:
+        """The original messages that survived, in result order.
+
+        The pipeline may prepend injected context (session memory, runtime
+        artifacts); the compaction contract is about the original messages
+        staying verbatim and in order.
+        """
+        return [item for item in result if item in originals]
+
+    async def test_history_below_threshold_is_never_rewritten(self) -> None:
+        # 50% usage, 62 messages: the legacy pipeline would microcompact and
+        # snip this history; the new contract keeps every message verbatim.
+        llm = _CountingLLM(token_count=50_000)
+        runtime = _runtime(llm)
+        messages = _long_history(rounds=30)
+
+        result = await self._pipeline(runtime, [dict(item) for item in messages])
+
+        self.assertEqual(self._originals_in(result, messages), messages)
+        flattened = json.dumps(result, ensure_ascii=False)
+        self.assertNotIn("[runtime_v2 snip]", flattened)
+        self.assertNotIn("microcompacted", flattened)
+        self.assertNotIn("truncated by runtime_v2", flattened)
+        self.assertNotIn("durable compaction", flattened)
+
+    async def test_durable_compaction_folds_old_history_into_summary(self) -> None:
+        llm = _CountingLLM(token_count=95_000)
+        compactor = SimpleNamespace(
+            summarize_runtime_history=AsyncMock(return_value="SUMMARY-OF-EARLIER-WORK"),
+        )
+        runtime = _runtime(llm, compactor=compactor)
+        messages = _long_history(rounds=20)
+        # One trailing assistant message so the naive tail cut would land on a
+        # tool result and must walk back to its assistant tool_calls message.
+        messages.append({"role": "assistant", "content": "wrap up"})
+        boundaries: list[dict[str, object]] = []
+        notes: dict[str, object] = {}
+
+        result = await self._pipeline(
+            runtime, list(messages), boundaries=boundaries, runtime_notes=notes
+        )
+
+        self.assertEqual(result[0], messages[0])
+        summary_indexes = [
+            index
+            for index, item in enumerate(result)
+            if "[runtime_v2 durable compaction]" in str(item.get("content", ""))
+        ]
+        self.assertEqual(len(summary_indexes), 1)
+        summary_message = result[summary_indexes[0]]
+        self.assertEqual(summary_message["role"], "user")
+        self.assertIn("SUMMARY-OF-EARLIER-WORK", summary_message["content"])
+        # Survivors are the prefix, the seed user request (kept verbatim on
+        # every round), and the recent tail, which starts at the assistant
+        # message owning the tool results (a naive cut would split the pair).
+        survivors = self._originals_in(result, messages)
+        tail = survivors[2:]
+        self.assertEqual([messages[0], messages[1], *tail], survivors)
+        self.assertEqual(messages[1]["content"], "original request")
+        self.assertEqual(tail, messages[len(messages) - len(tail):])
+        self.assertEqual(tail[0]["role"], "assistant")
+        self.assertTrue(tail[0].get("tool_calls"))
+        compactor.summarize_runtime_history.assert_awaited_once()
+        rendered = compactor.summarize_runtime_history.await_args.kwargs["messages"]
+        self.assertTrue(all(set(item) == {"role", "content"} for item in rendered))
+        self.assertEqual(notes.get("durable_compaction_failures"), 0)
+        self.assertEqual(len(boundaries), 1)
+        self.assertIn("durable_compaction", boundaries[0]["pipeline"])
+        self.assertNotIn("emergency_microcompact", boundaries[0]["pipeline"])
+
+    async def test_repeated_compaction_keeps_one_summary_and_seed_request(self) -> None:
+        llm = _CountingLLM(token_count=95_000)
+        compactor = SimpleNamespace(
+            summarize_runtime_history=AsyncMock(
+                side_effect=[f"SUMMARY-ROUND-{n}" for n in range(1, 10)]
+            ),
+        )
+        runtime = _runtime(llm, compactor=compactor)
+        # Simulate injected context having shifted the prefix boundary: an
+        # artifact-style system message sits between the system prompt and the
+        # seed request, so the seed request lives beyond base_prefix_len.
+        current: list[dict[str, object]] = [
+            {"role": "system", "content": "system prompt"},
+            {"role": "system", "content": "## Runtime Artifact: injected context"},
+            {"role": "user", "content": "SEED-REQUEST keep me verbatim"},
+        ]
+        next_round = 0
+        for _ in range(12):
+            current.extend(_paired_tool_round(next_round))
+            next_round += 1
+        notes: dict[str, object] = {}
+
+        for round_no in range(1, 4):
+            current = await self._pipeline(runtime, current, runtime_notes=notes)
+            markers = [
+                item
+                for item in current
+                if "[runtime_v2 durable compaction]" in str(item.get("content", ""))
+            ]
+            self.assertEqual(len(markers), 1, f"round {round_no}: exactly one summary must exist")
+            self.assertIn(f"SUMMARY-ROUND-{round_no}", str(markers[0]["content"]))
+            seeds = [item for item in current if "SEED-REQUEST" in str(item.get("content", ""))]
+            self.assertEqual(len(seeds), 1, f"round {round_no}: seed request must survive")
+            self.assertEqual(seeds[0]["content"], "SEED-REQUEST keep me verbatim")
+            self.assertEqual(seeds[0]["role"], "user")
+            for _ in range(4):
+                current.extend(_paired_tool_round(next_round))
+                next_round += 1
+
+        self.assertEqual(compactor.summarize_runtime_history.await_count, 3)
+        # Chain continuity: each round re-summarizes the previous summary.
+        second_input = json.dumps(
+            compactor.summarize_runtime_history.await_args_list[1].kwargs["messages"],
+            ensure_ascii=False,
+        )
+        self.assertIn("SUMMARY-ROUND-1", second_input)
+        third_input = json.dumps(
+            compactor.summarize_runtime_history.await_args_list[2].kwargs["messages"],
+            ensure_ascii=False,
+        )
+        self.assertIn("SUMMARY-ROUND-2", third_input)
+
+    async def test_summarizer_failure_keeps_history_and_trips_breaker(self) -> None:
+        llm = _CountingLLM(token_count=95_000)
+        compactor = SimpleNamespace(
+            summarize_runtime_history=AsyncMock(side_effect=RuntimeError("summarizer down")),
+        )
+        runtime = _runtime(llm, compactor=compactor)
+        messages = _long_history(rounds=10)
+        notes: dict[str, object] = {}
+
+        first = await self._pipeline(runtime, list(messages), runtime_notes=notes)
+        second = await self._pipeline(runtime, list(messages), runtime_notes=notes)
+        third = await self._pipeline(runtime, list(messages), runtime_notes=notes)
+
+        self.assertEqual(self._originals_in(first, messages), messages)
+        self.assertEqual(self._originals_in(second, messages), messages)
+        self.assertEqual(self._originals_in(third, messages), messages)
+        self.assertEqual(notes.get("durable_compaction_failures"), 2)
+        # circuit_breaker_failures defaults to 2: the third round must not
+        # have attempted another summary.
+        self.assertEqual(compactor.summarize_runtime_history.await_count, 2)
+
+    async def test_overflow_force_uses_emergency_fallback_without_compactor(self) -> None:
+        llm = _CountingLLM(token_count=95_000)
+        runtime = _runtime(llm)
+        messages = _long_history(rounds=30)
+        boundaries: list[dict[str, object]] = []
+
+        result = await self._pipeline(runtime, list(messages), boundaries=boundaries, force=True)
+
+        self.assertNotEqual(result, messages)
+        flattened = json.dumps(result, ensure_ascii=False)
+        self.assertIn("[runtime_v2 snip]", flattened)
+        self.assertEqual(len(boundaries), 1)
+        self.assertIn("emergency_microcompact", boundaries[0]["pipeline"])
+        self.assertNotIn("durable_compaction", boundaries[0]["pipeline"])
+
+    async def test_tool_result_budget_keeps_head_and_tail(self) -> None:
+        runtime = _runtime(_CountingLLM())
+        content = "HEAD" + ("x" * 30_000) + "TAIL-MARKER"
+
+        [message] = runtime._apply_tool_result_budget(
+            [{"role": "tool", "tool_call_id": "c", "content": content}]
+        )
+
+        self.assertTrue(message["content"].startswith("HEAD"))
+        self.assertTrue(message["content"].endswith("TAIL-MARKER"))
+        self.assertIn("chars omitted", message["content"])
+        self.assertLess(len(message["content"]), 13_000)
+
+    async def test_threshold_uses_observed_prompt_tokens_anchor(self) -> None:
+        # The local estimator undercounts badly; the provider-reported prompt
+        # size of the latest request must still trigger compaction.
+        llm = _CountingLLM(token_count=1_000)
+        runtime = _runtime(llm)
+        messages = _long_history(rounds=5)
+
+        self.assertFalse(runtime._should_apply_hard_compaction(messages, None))
+        self.assertTrue(
+            runtime._should_apply_hard_compaction(messages, None, observed_tokens=95_000)
+        )
+
+
+class SummarizeRuntimeHistoryTests(unittest.IsolatedAsyncioTestCase):
+    async def test_summary_prompt_contract_and_parsing(self) -> None:
+        captured: dict[str, str] = {}
+
+        class _LLM:
+            async def simple_chat(self, *, prompt: str, system: str, task_type: str) -> str:
+                captured["system"] = system
+                captured["prompt"] = prompt
+                return json.dumps({"history_summary": "NINE-SECTION-SUMMARY"})
+
+        compactor = HistoryCompactor(llm=_LLM(), store=None, memory_manager=None)
+
+        summary = await compactor.summarize_runtime_history(
+            project_id="proj1",
+            session_id="rt-1",
+            messages=[{"role": "user", "content": "build the feature"}],
+        )
+
+        self.assertEqual(summary, "NINE-SECTION-SUMMARY")
+        self.assertIn("Primary Request and Intent", captured["system"])
+        self.assertIn("All User Messages", captured["system"])
+        self.assertIn("build the feature", captured["prompt"])
+
+    async def test_without_llm_falls_back_to_mechanical_summary(self) -> None:
+        compactor = HistoryCompactor(llm=None, store=None, memory_manager=None)
+
+        summary = await compactor.summarize_runtime_history(
+            project_id="proj1",
+            session_id="rt-1",
+            messages=[{"role": "user", "content": "important detail"}],
+        )
+
+        self.assertTrue(summary.strip())
+        self.assertIn("important detail", summary)
+
+
+class MaybeCompactSessionHistoryTests(unittest.IsolatedAsyncioTestCase):
+    async def test_calls_compactor_with_resolved_project(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            memory = MemoryManager(Path(tmpdir), "proj1", store=None)
+            compactor = SimpleNamespace(maybe_compact_session=AsyncMock(return_value=True))
+            memory.set_history_compactor(compactor)
+
+            self.assertTrue(await memory.maybe_compact_session_history("sess-1"))
+
+            compactor.maybe_compact_session.assert_awaited_once_with(
+                project_id="proj1", session_id="sess-1", force=False
+            )
+
+    async def test_absent_or_failing_compactor_is_safe(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            memory = MemoryManager(Path(tmpdir), "proj1", store=None)
+
+            self.assertFalse(await memory.maybe_compact_session_history("sess-1"))
+
+            memory.set_history_compactor(
+                SimpleNamespace(maybe_compact_session=AsyncMock(side_effect=RuntimeError("boom")))
+            )
+            self.assertFalse(await memory.maybe_compact_session_history("sess-1"))
+
+
+class OverflowRecoveryEndToEndTests(unittest.IsolatedAsyncioTestCase):
+    async def test_provider_overflow_error_recovers_via_forced_compaction(self) -> None:
+        """A real provider overflow mid-run must force-compact and retry.
+
+        Token accounting is deliberately kept far below the threshold so the
+        ONLY thing that can rescue the run is the reactive overflow path.
+        """
+
+        def _event(event_type: str, payload: dict[str, object]):
+            return type("Evt", (), {"event_type": event_type, "payload": payload, "model": "stub"})()
+
+        class _OverflowThenRecoverLLM:
+            def __init__(self) -> None:
+                self.calls = 0
+                self.overflow_thrown = False
+                self.prompts: list[list[dict[str, object]]] = []
+                self.config = type("Cfg", (), {"max_tokens": 2048})()
+
+            def prepare_user_message_content(self, content: str, attachment_refs=None):
+                _ = attachment_refs
+                return content
+
+            def get_tool_definitions(self, tools):
+                return tools
+
+            def is_context_overflow_error(self, error: Exception) -> bool:
+                return "maximum context length" in str(error)
+
+            def count_input_tokens(self, messages, tools=None):
+                _ = (messages, tools)
+                return 100
+
+            def get_context_window(self):
+                return 10_000
+
+            async def chat_stream(self, messages, tools=None):
+                _ = tools
+                self.calls += 1
+                self.prompts.append([dict(item) for item in messages])
+                if self.calls > 6 and not self.overflow_thrown:
+                    self.overflow_thrown = True
+                    raise RuntimeError("provider rejected: maximum context length exceeded")
+                yield _event("message_start", {})
+                if self.calls <= 6:
+                    yield _event("assistant_delta", {"text": f"working {self.calls}"})
+                    yield _event(
+                        "tool_call_delta",
+                        {
+                            "index": 0,
+                            "id": f"tool-{self.calls}",
+                            "name": "demo_tool",
+                            "arguments": "{\"value\": \"go\"}",
+                        },
+                    )
+                else:
+                    yield _event("assistant_delta", {"text": "final answer"})
+                yield _event("usage", {"prompt_tokens": 1_000, "completion_tokens": 10})
+                yield _event("message_stop", {"finish_reason": "stop"})
+
+        async def demo_tool(value: str) -> dict[str, str]:
+            return {"echo": value + ("-detail" * 50)}
+
+        registry = ToolRegistry()
+        registry.register(
+            ToolDefinition(
+                name="demo_tool",
+                description="Demo runtime tool",
+                parameters={
+                    "type": "object",
+                    "properties": {"value": {"type": "string"}},
+                    "required": ["value"],
+                },
+                func=demo_tool,
+                concurrency_safe=True,
+                read_only=True,
+            )
+        )
+
+        llm = _OverflowThenRecoverLLM()
+        compactor = SimpleNamespace(
+            summarize_runtime_history=AsyncMock(return_value="OVERFLOW-RECOVERY-SUMMARY"),
+        )
+        runtime = NativeRuntimeV2(
+            llm=llm,
+            tool_registry=registry,
+            config=OPCConfig(),
+            history_compactor=compactor,
+            max_iterations=12,
+        )
+
+        result = await runtime.run(
+            system_prompt="You are a runtime.",
+            user_message="run a long task",
+            task=Task(
+                id="overflow-task",
+                title="overflow task",
+                session_id="sess-overflow",
+                project_id="proj1",
+                metadata={"mode": "task", "execution_mode": "task_mode"},
+            ),
+        )
+
+        self.assertEqual(result.status, TaskStatus.DONE)
+        self.assertIn("final answer", str(result.content))
+        # The failing request carried no summary; the retry after the forced
+        # compaction did, and the summarizer ran exactly once.
+        compactor.summarize_runtime_history.assert_awaited_once()
+        failing_prompt = json.dumps(llm.prompts[-2], ensure_ascii=False)
+        retry_prompt = json.dumps(llm.prompts[-1], ensure_ascii=False)
+        self.assertNotIn("[runtime_v2 durable compaction]", failing_prompt)
+        self.assertIn("[runtime_v2 durable compaction]", retry_prompt)
+        self.assertIn("OVERFLOW-RECOVERY-SUMMARY", retry_prompt)
+
+
+class DurableCompactionEndToEndTests(unittest.IsolatedAsyncioTestCase):
+    async def test_run_compacts_midway_on_observed_usage_and_completes(self) -> None:
+        def _event(event_type: str, payload: dict[str, object]):
+            return type("Evt", (), {"event_type": event_type, "payload": payload, "model": "stub"})()
+
+        class _LongRunLLM:
+            """Tool-looping stub whose provider usage crosses the threshold mid-run.
+
+            count_input_tokens deliberately returns 0 so the trigger can only
+            come from the observed usage anchor.
+            """
+
+            def __init__(self) -> None:
+                self.calls = 0
+                self.prompts: list[list[dict[str, object]]] = []
+                self.config = type("Cfg", (), {"max_tokens": 2048})()
+
+            def prepare_user_message_content(self, content: str, attachment_refs=None):
+                _ = attachment_refs
+                return content
+
+            def get_tool_definitions(self, tools):
+                return tools
+
+            def is_context_overflow_error(self, error: Exception) -> bool:
+                _ = error
+                return False
+
+            def count_input_tokens(self, messages, tools=None):
+                _ = (messages, tools)
+                return 0
+
+            def get_context_window(self):
+                return 10_000
+
+            async def chat_stream(self, messages, tools=None):
+                _ = tools
+                self.calls += 1
+                self.prompts.append([dict(item) for item in messages])
+                yield _event("message_start", {})
+                if self.calls <= 12:
+                    yield _event("assistant_delta", {"text": f"working {self.calls}"})
+                    yield _event(
+                        "tool_call_delta",
+                        {
+                            "index": 0,
+                            "id": f"tool-{self.calls}",
+                            "name": "demo_tool",
+                            "arguments": "{\"value\": \"go\"}",
+                        },
+                    )
+                else:
+                    yield _event("assistant_delta", {"text": "final answer"})
+                yield _event(
+                    "usage",
+                    {
+                        "prompt_tokens": 9_500 if self.calls >= 5 else 1_000,
+                        "completion_tokens": 10,
+                    },
+                )
+                yield _event("message_stop", {"finish_reason": "stop"})
+
+        async def demo_tool(value: str) -> dict[str, str]:
+            return {"echo": value + ("-detail" * 50)}
+
+        registry = ToolRegistry()
+        registry.register(
+            ToolDefinition(
+                name="demo_tool",
+                description="Demo runtime tool",
+                parameters={
+                    "type": "object",
+                    "properties": {"value": {"type": "string"}},
+                    "required": ["value"],
+                },
+                func=demo_tool,
+                concurrency_safe=True,
+                read_only=True,
+            )
+        )
+
+        llm = _LongRunLLM()
+        compactor = SimpleNamespace(
+            summarize_runtime_history=AsyncMock(
+                side_effect=[f"MIDRUN-SUMMARY-{n}" for n in range(1, 10)]
+            ),
+        )
+        runtime = NativeRuntimeV2(
+            llm=llm,
+            tool_registry=registry,
+            config=OPCConfig(),
+            history_compactor=compactor,
+            max_iterations=16,
+        )
+
+        result = await runtime.run(
+            system_prompt="You are a runtime.",
+            user_message="run a long task",
+            task=Task(
+                id="long-task",
+                title="long task",
+                session_id="sess-long",
+                project_id="proj1",
+                metadata={"mode": "task", "execution_mode": "task_mode"},
+            ),
+        )
+
+        self.assertEqual(result.status, TaskStatus.DONE)
+        self.assertIn("final answer", str(result.content))
+        # The run must compact more than once, and every request must hold at
+        # most ONE summary message with the seed request still verbatim.
+        rounds = compactor.summarize_runtime_history.await_count
+        self.assertGreaterEqual(rounds, 2)
+        for index, prompt in enumerate(llm.prompts):
+            marker_messages = [
+                item
+                for item in prompt
+                if "[runtime_v2 durable compaction]" in str(item.get("content", ""))
+            ]
+            self.assertLessEqual(len(marker_messages), 1, f"request {index}")
+            self.assertTrue(
+                any(
+                    item.get("role") == "user" and "run a long task" in str(item.get("content", ""))
+                    for item in prompt
+                ),
+                f"request {index}: seed request must stay verbatim",
+            )
+        final_prompt = json.dumps(llm.prompts[-1], ensure_ascii=False)
+        self.assertIn(f"MIDRUN-SUMMARY-{rounds}", final_prompt)
+        # Chain continuity: the second summary round saw the first summary.
+        second_input = json.dumps(
+            compactor.summarize_runtime_history.await_args_list[1].kwargs["messages"],
+            ensure_ascii=False,
+        )
+        self.assertIn("MIDRUN-SUMMARY-1", second_input)
+
+
+if __name__ == "__main__":
+    unittest.main()