diff --git a/opc/core/config.py b/opc/core/config.py index e661e5e..001c4b4 100644 --- a/opc/core/config.py +++ b/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 diff --git a/opc/layer1_perception/context_loader.py b/opc/layer1_perception/context_loader.py index 7cb23df..4a395f4 100644 --- a/opc/layer1_perception/context_loader.py +++ b/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, diff --git a/opc/layer2_organization/secretary.py b/opc/layer2_organization/secretary.py index 0e95edc..3d397bf 100644 --- a/opc/layer2_organization/secretary.py +++ b/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, diff --git a/opc/layer3_agent/runtime_v2/runtime.py b/opc/layer3_agent/runtime_v2/runtime.py index 0f8ea23..3c2c479 100644 --- a/opc/layer3_agent/runtime_v2/runtime.py +++ b/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, diff --git a/opc/layer5_memory/history_compactor.py b/opc/layer5_memory/history_compactor.py index 0d9a3ac..c94cb6b 100644 --- a/opc/layer5_memory/history_compactor.py +++ b/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) diff --git a/opc/layer5_memory/memory_manager.py b/opc/layer5_memory/memory_manager.py index 89ed3ee..c6589b0 100644 --- a/opc/layer5_memory/memory_manager.py +++ b/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") diff --git a/opc/plugins/office_ui/dispatcher.py b/opc/plugins/office_ui/dispatcher.py index 17163f3..0d4559e 100644 --- a/opc/plugins/office_ui/dispatcher.py +++ b/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, diff --git a/tests/test_runtime_context_compaction.py b/tests/test_runtime_context_compaction.py new file mode 100644 index 0000000..1f6858d --- /dev/null +++ b/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()