From 430aa1789c7c22aa1798b61e6c97a02f77bd2fcb Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 16:58:50 +0800 Subject: [PATCH 01/13] feat(runtime): expose structured tool result metadata --- CHANGELOG.md | 10 ++++++++ agent_core/loop_types.py | 8 +++++++ agent_core/runtime/loop/tool_exec.py | 29 +++++++++++++++++++++- pyproject.toml | 2 +- tests/test_tool_exec.py | 36 ++++++++++++++++++++++++++++ uv.lock | 2 +- 6 files changed, 84 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8b17c7..40f707e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ the GitHub Release body, so a release with no entry here fails. Versioning follows [docs/versioning.md](docs/versioning.md). +## [0.4.0] - 2026-09-03 + +### Added + +- `ToolResult` now carries optional structured failure, recovery, and repeated- + invocation metadata. `ToolExecutionHooks.result_metadata` lets products + populate those fields without moving product-specific classifiers or stores + into AgentCore. Timeouts and exceptions receive stable core-owned + `error_kind` values. + ## [0.3.0] - 2026-09-03 First published release. AgentCore is now open source under Apache-2.0 and diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py index 580ddac..b134f64 100644 --- a/agent_core/loop_types.py +++ b/agent_core/loop_types.py @@ -306,6 +306,14 @@ class ToolResult: is_error: bool # Interrupted results remain in history to preserve tool-call pairing. interrupted: bool = False + # Stable machine-readable failure category supplied by the execution + # engine or a host classifier. Empty for successful legacy tools. + error_kind: str = "" + # Opaque host-owned handle for a result body shed from model context. + result_id: str = "" + # Host-provided repeated-invocation metadata. Execution is never skipped. + repeat_count: int = 1 + repeat_recovery_id: str = "" @dataclass diff --git a/agent_core/runtime/loop/tool_exec.py b/agent_core/runtime/loop/tool_exec.py index 6956f2a..9dbdc2c 100644 --- a/agent_core/runtime/loop/tool_exec.py +++ b/agent_core/runtime/loop/tool_exec.py @@ -77,6 +77,7 @@ def process(self, tool_result: ToolResult) -> str: [dict[str, Any], float], AbstractContextManager[Any] ] ResultTransform = Callable[[str, str], str] +ResultMetadata = Callable[[str, Any, str], dict[str, Any]] BatchTransform = Callable[[list[ToolResult]], list[ToolResult]] CallObserver = Callable[[str], None] UnknownResult = Callable[[str, tuple[str, ...]], str] @@ -125,6 +126,14 @@ def _identity_batch(results: list[ToolResult]) -> list[ToolResult]: return results +def _empty_result_metadata( + _name: str, + _raw: Any, + _rendered: str, +) -> dict[str, Any]: + return {} + + def _noop_call(_name: str) -> None: return None @@ -215,6 +224,7 @@ class ToolExecutionHooks: await_call: AwaitCall = _default_await call_scope: CallScopeFactory = _default_scope transform_result: ResultTransform = _identity_result + result_metadata: ResultMetadata = _empty_result_metadata transform_batch: BatchTransform = _identity_batch on_call: CallObserver = _noop_call unknown_result: UnknownResult = _unknown_result @@ -325,14 +335,29 @@ async def _run_one(call: dict[str, Any], index: int) -> ToolResult: raw = await invocation result = runtime.transform_result(name, str(raw) if raw is not None else "") + metadata_raw = _safe_hook( + "result_metadata", + lambda: runtime.result_metadata(name, raw, result), + dict, + ) + metadata = metadata_raw if isinstance(metadata_raw, dict) else {} + error_kind = str(metadata.get("error_kind") or "") + try: + repeat_count = max(1, int(metadata.get("repeat_count") or 1)) + except (TypeError, ValueError): + repeat_count = 1 return ToolResult( name=name, args=args, result=result, duration_ms=int((time.monotonic() - start) * 1000), tool_call_id=tool_call_id, - is_error=False, + is_error=bool(error_kind), interrupted=woke_for_interrupt, + error_kind=error_kind, + result_id=str(metadata.get("result_id") or ""), + repeat_count=repeat_count, + repeat_recovery_id=str(metadata.get("repeat_recovery_id") or ""), ) except TimeoutError: elapsed = int((time.monotonic() - start) * 1000) @@ -351,6 +376,7 @@ async def _run_one(call: dict[str, Any], index: int) -> ToolResult: duration_ms=elapsed, tool_call_id=tool_call_id, is_error=True, + error_kind="timeout", ) except asyncio.CancelledError: raise @@ -367,6 +393,7 @@ async def _run_one(call: dict[str, Any], index: int) -> ToolResult: duration_ms=int((time.monotonic() - start) * 1000), tool_call_id=tool_call_id, is_error=True, + error_kind="exception", ) finally: if interrupt_task is not None and not interrupt_task.done(): diff --git a/pyproject.toml b/pyproject.toml index 7529927..3d6f8fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "apodex-agent-core" -version = "0.3.0" +version = "0.4.0" description = "Shared, product-neutral runtime primitives for Apodex agents" readme = "README.md" license = "Apache-2.0" diff --git a/tests/test_tool_exec.py b/tests/test_tool_exec.py index a09160a..a13809a 100644 --- a/tests/test_tool_exec.py +++ b/tests/test_tool_exec.py @@ -69,6 +69,12 @@ async def await_call(awaitable, name, args, timeout): call_scope=scope, on_call=lambda name: events.append(f"meter:{name}"), transform_result=lambda name, value: f"{name}={value}", + result_metadata=lambda _name, _raw, _rendered: { + "error_kind": "command_exit", + "result_id": "spill-1", + "repeat_count": 2, + "repeat_recovery_id": "spill-0", + }, transform_batch=lambda results: [replace(results[0], result="batch")], ) results = await execute_tools( @@ -81,9 +87,39 @@ async def await_call(awaitable, name, args, timeout): ) assert results[0].result == "batch" + assert results[0].is_error is True + assert results[0].error_kind == "command_exit" + assert results[0].result_id == "spill-1" + assert results[0].repeat_count == 2 + assert results[0].repeat_recovery_id == "spill-0" assert events == ["meter:echo", "enter:tc:12", "await:echo:3:12", "exit"] +@pytest.mark.asyncio +async def test_timeout_and_exception_have_structured_error_kinds() -> None: + results = await execute_tools( + [ + {"name": "slow", "args": {}, "id": "a"}, + {"name": "bad", "args": {}, "id": "b"}, + ], + { + "slow": FakeTool("slow", delay=1), + "bad": FakeTool("bad", ValueError("boom")), + }, + timeout=1, + turn=1, + count_offset=0, + hooks=ToolExecutionHooks( + resolve_timeout=lambda name, _args, _configured: ( + 0.001 if name == "slow" else 1.0 + ) + ), + ) + + assert results[0].error_kind == "timeout" + assert results[1].error_kind == "exception" + + @pytest.mark.asyncio async def test_fan_in_interrupt_cancels_invocation_and_returns_result() -> None: async def interrupt(_call: dict[str, Any]) -> bool: diff --git a/uv.lock b/uv.lock index 908612c..95b8de1 100644 --- a/uv.lock +++ b/uv.lock @@ -50,7 +50,7 @@ wheels = [ [[package]] name = "apodex-agent-core" -version = "0.3.0" +version = "0.4.0" source = { editable = "." } dependencies = [ { name = "anthropic", extra = ["bedrock"] }, From d829eb500aca23c355c2176a519edd276d6f6557 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:01:08 +0800 Subject: [PATCH 02/13] fix(runtime): pass tool arguments to metadata hooks --- agent_core/runtime/loop/tool_exec.py | 5 +++-- tests/test_tool_exec.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/agent_core/runtime/loop/tool_exec.py b/agent_core/runtime/loop/tool_exec.py index 9dbdc2c..56e5953 100644 --- a/agent_core/runtime/loop/tool_exec.py +++ b/agent_core/runtime/loop/tool_exec.py @@ -77,7 +77,7 @@ def process(self, tool_result: ToolResult) -> str: [dict[str, Any], float], AbstractContextManager[Any] ] ResultTransform = Callable[[str, str], str] -ResultMetadata = Callable[[str, Any, str], dict[str, Any]] +ResultMetadata = Callable[[str, dict[str, Any], Any, str], dict[str, Any]] BatchTransform = Callable[[list[ToolResult]], list[ToolResult]] CallObserver = Callable[[str], None] UnknownResult = Callable[[str, tuple[str, ...]], str] @@ -128,6 +128,7 @@ def _identity_batch(results: list[ToolResult]) -> list[ToolResult]: def _empty_result_metadata( _name: str, + _args: dict[str, Any], _raw: Any, _rendered: str, ) -> dict[str, Any]: @@ -337,7 +338,7 @@ async def _run_one(call: dict[str, Any], index: int) -> ToolResult: result = runtime.transform_result(name, str(raw) if raw is not None else "") metadata_raw = _safe_hook( "result_metadata", - lambda: runtime.result_metadata(name, raw, result), + lambda: runtime.result_metadata(name, args, raw, result), dict, ) metadata = metadata_raw if isinstance(metadata_raw, dict) else {} diff --git a/tests/test_tool_exec.py b/tests/test_tool_exec.py index a13809a..979549d 100644 --- a/tests/test_tool_exec.py +++ b/tests/test_tool_exec.py @@ -69,7 +69,7 @@ async def await_call(awaitable, name, args, timeout): call_scope=scope, on_call=lambda name: events.append(f"meter:{name}"), transform_result=lambda name, value: f"{name}={value}", - result_metadata=lambda _name, _raw, _rendered: { + result_metadata=lambda _name, _args, _raw, _rendered: { "error_kind": "command_exit", "result_id": "spill-1", "repeat_count": 2, From c39691fb242bb93db2c93a288be2be224193ecdc Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:06:18 +0800 Subject: [PATCH 03/13] fix(runtime): expose call identity to middleware scopes --- agent_core/runtime/loop/agent_loop.py | 23 ++++++++++++ tests/test_agent_loop_engine.py | 50 +++++++++++++++++++++++++++ 2 files changed, 73 insertions(+) diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index 29befdf..ccabf90 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -601,6 +601,21 @@ async def _call_llm_with_callbacks( metadata["_llm_attempt_index"] = current_attempt_index metadata["_llm_attempt_outcome"] = "" metadata["_llm_attempt_count"] = 0 + # Middleware proxies resolve their per-call context from the active scope, + # while observer contexts use the loop-local metadata mapping above. Keep + # the shared identity keys aligned so host wire-capture hooks can join the + # pre-middleware and post-middleware views of the same physical request. + from agent_core.execution_context import get_current_execution_scope + + active_scope = get_current_execution_scope() + if active_scope is not None: + active_scope.metadata.update( + { + "_llm_call_id": call_id, + "_llm_attempt_id": current_attempt_id, + "_llm_attempt_index": current_attempt_index, + } + ) async def _on_attempt(event: dict[str, Any]) -> None: nonlocal current_attempt_id, current_attempt_index @@ -616,6 +631,14 @@ async def _on_attempt(event: dict[str, Any]) -> None: metadata["_llm_attempt_count"] = max( int(metadata.get("_llm_attempt_count", 0) or 0), current_attempt_index ) + if active_scope is not None: + active_scope.metadata.update( + { + "_llm_call_id": call_id, + "_llm_attempt_id": current_attempt_id, + "_llm_attempt_index": current_attempt_index, + } + ) attempt_usage = event.get("usage") if isinstance(attempt_usage, dict): attempt_usage = dict(attempt_usage) diff --git a/tests/test_agent_loop_engine.py b/tests/test_agent_loop_engine.py index ea2c3aa..5ffa49c 100644 --- a/tests/test_agent_loop_engine.py +++ b/tests/test_agent_loop_engine.py @@ -134,6 +134,56 @@ def enter_scope(cfg, phase_id, metadata): assert events[-1] == "exit:token" +@pytest.mark.asyncio +async def test_active_scope_receives_llm_call_identity() -> None: + from agent_core.execution_context import ( + ExecutionScope, + get_current_execution_scope, + reset_current_execution_scope, + set_current_execution_scope, + ) + + seen: dict[str, Any] = {} + + class ScopeInspectingLLM(SequenceLLM): + async def chat(self, messages, **kwargs) -> LLMResponse: + scope = get_current_execution_scope() + assert scope is not None + seen.update(scope.metadata) + return await super().chat(messages, **kwargs) + + def enter_scope(cfg, phase_id, metadata): + scope = ExecutionScope( + task_id=cfg.task_id, + phase_id=phase_id, + role_id=cfg.role_id, + metadata=metadata, + ) + return scope, set_current_execution_scope(scope) + + await run_agent_loop( + system_prompt="system", + user_message="start", + llm=ScopeInspectingLLM([LLMResponse(content="done")]), + tools=[], + config=LoopConfig( + max_turns=1, + task_id="task", + role_id="role", + loop_policy=LoopPolicy(no_tool_behavior="stop"), + max_llm_retries=1, + ), + runtime_hooks=AgentLoopHooks( + enter_scope=enter_scope, + exit_scope=reset_current_execution_scope, + ), + ) + + assert str(seen["_llm_call_id"]).startswith("llm_") + assert str(seen["_llm_attempt_id"]).endswith("_attempt_01") + assert seen["_llm_attempt_index"] == 1 + + @pytest.mark.asyncio async def test_host_can_override_session_binding() -> None: bound: list[str] = [] From d3ab838dcaa59d0f5106918fff56ff23355a88cf Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:08:05 +0800 Subject: [PATCH 04/13] feat(runtime): notify observers of model input --- agent_core/runtime/loop/agent_loop.py | 14 +++++++++++++ tests/test_agent_loop_engine.py | 29 +++++++++++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index ccabf90..b4d37de 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -617,6 +617,20 @@ async def _call_llm_with_callbacks( } ) + input_ctx = TurnContext( + turn=turn, + max_turns=cfg.max_turns, + task_id=cfg.task_id, + role_id=cfg.role_id, + ai_text="", + thinking="", + tool_calls=[], + messages=list(messages_for_call), + usage=None, + metadata=dict(metadata), + ) + await notify_observers(obs, "on_llm_input", input_ctx) + async def _on_attempt(event: dict[str, Any]) -> None: nonlocal current_attempt_id, current_attempt_index current_attempt_index = int(event.get("attempt_index", 1) or 1) diff --git a/tests/test_agent_loop_engine.py b/tests/test_agent_loop_engine.py index 5ffa49c..144a0f8 100644 --- a/tests/test_agent_loop_engine.py +++ b/tests/test_agent_loop_engine.py @@ -184,6 +184,35 @@ def enter_scope(cfg, phase_id, metadata): assert seen["_llm_attempt_index"] == 1 +@pytest.mark.asyncio +async def test_observer_receives_pre_provider_llm_input() -> None: + captured: list[Any] = [] + + class InputObserver: + async def on_llm_input(self, ctx) -> None: + captured.append(ctx) + + await run_agent_loop( + system_prompt="system", + user_message="start", + llm=SequenceLLM([LLMResponse(content="done")]), + tools=[], + config=LoopConfig( + max_turns=1, + task_id="task", + role_id="role", + loop_policy=LoopPolicy(no_tool_behavior="stop"), + max_llm_retries=1, + ), + observers=[InputObserver()], + ) + + assert len(captured) == 1 + assert captured[0].messages[0]["content"] == "system" + assert captured[0].messages[1]["content"] == "start" + assert str(captured[0].metadata["_llm_call_id"]).startswith("llm_") + + @pytest.mark.asyncio async def test_host_can_override_session_binding() -> None: bound: list[str] = [] From 1ef01d113998850917f73c87195f1f1821d6bedd Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:12:07 +0800 Subject: [PATCH 05/13] feat(runtime): report discarded loop context --- agent_core/loop_types.py | 20 ++++++++ agent_core/runtime/loop/agent_loop.py | 70 +++++++++++++++++++++++++-- tests/test_agent_loop_engine.py | 48 ++++++++++++++++++ 3 files changed, 134 insertions(+), 4 deletions(-) diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py index b134f64..66f299f 100644 --- a/agent_core/loop_types.py +++ b/agent_core/loop_types.py @@ -316,6 +316,22 @@ class ToolResult: repeat_recovery_id: str = "" +@dataclass +class ContextCompactionContext: + """Pre/post snapshot for any loop operation that discards history.""" + + turn: int + task_id: str + role_id: str + reason: str + compactor: str + policy: str + messages_before: list[Message] + messages_after: list[Message] + tokens_before: int + metadata: dict[str, Any] + + @dataclass class Intervention: inject_messages: list[str] | None = None @@ -480,6 +496,9 @@ async def on_compaction(self, event: CompactionEvent) -> None: """History was rewritten. Passive: compaction has already happened by the time this runs, so there is no intervention to return.""" + async def on_context_compacted(self, ctx: ContextCompactionContext) -> None: + """History was discarded while its pre-change form was still reachable.""" + async def on_loop_end(self, result: AgentLoopResult) -> None: pass @@ -697,6 +716,7 @@ async def notify_tool_result( "CancellationObserver", "CompactionEvent", "CompactionObserver", + "ContextCompactionContext", "Intervention", "LLMAttemptContext", "LLMDeltaContext", diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index b4d37de..99bc000 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -20,6 +20,7 @@ from agent_core.loop_types import ( AgentLoopResult, CompactionEvent, + ContextCompactionContext, LLMAttemptContext, LLMDeltaContext, LoopConfig, @@ -838,7 +839,7 @@ def _answer_unexecuted_tool_calls(messages: list[Message], detail: str) -> None: answered.add(call_id) -def _pop_last_assistant_turn(messages: list[Message]) -> None: +def _pop_last_assistant_turn(messages: list[Message]) -> list[Message] | None: """Remove the assistant message a rollback observer rejected, in full. ``pop_last_message`` fires while the turn's tool calls are still @@ -874,11 +875,44 @@ def _pop_last_assistant_turn(messages: list[Message]) -> None: "_pop_last_assistant_turn: no assistant message in history; " "leaving it unchanged" ) - return + return None end = idx + 1 while end < len(messages) and is_tool_msg(messages[end]): end += 1 + removed = messages[idx:end] del messages[idx:end] + return removed + + +async def _notify_context_compacted( + obs: list[Any], + *, + cfg: LoopConfig, + turn: int, + reason: str, + before: list[Message], + after: list[Message], + tokens_before: int, + metadata: dict[str, Any], + compactor: str = "", + policy: str = "", +) -> None: + await notify_observers( + obs, + "on_context_compacted", + ContextCompactionContext( + turn=turn, + task_id=cfg.task_id, + role_id=cfg.role_id, + reason=reason, + compactor=compactor, + policy=policy, + messages_before=before, + messages_after=after, + tokens_before=tokens_before, + metadata=metadata, + ), + ) async def _process_llm_response( @@ -1016,7 +1050,21 @@ async def _process_llm_response( stop_reason = merged_llm.stop_reason or "" if merged_llm.pop_last_message and messages: - _pop_last_assistant_turn(messages) + rollback_before = list(messages) + rollback_tokens = estimate_tokens(messages) + removed = _pop_last_assistant_turn(messages) + if removed is not None: + await _notify_context_compacted( + obs, + cfg=cfg, + turn=turn, + reason="rollback_pop", + before=rollback_before, + after=messages, + tokens_before=rollback_tokens, + metadata=metadata, + policy="on_llm_response_rollback", + ) if merged_llm.continue_to_next_turn: if merged_llm.inject_messages: for msg_text in merged_llm.inject_messages: @@ -1312,7 +1360,21 @@ async def _handle_turn_end( # complete assistant turn, skip compaction/persistence for the discarded # attempt, and let the caller retry without consuming a logical turn. if merged_turn.pop_last_message: - _pop_last_assistant_turn(messages) + rollback_before = list(messages) + rollback_tokens = estimate_tokens(messages) + removed = _pop_last_assistant_turn(messages) + if removed is not None: + await _notify_context_compacted( + obs, + cfg=cfg, + turn=turn, + reason="rollback_pop", + before=rollback_before, + after=messages, + tokens_before=rollback_tokens, + metadata=metadata, + policy="on_turn_end_rollback", + ) if merged_turn.inject_messages: for msg_text in merged_turn.inject_messages: messages.append(user_msg(msg_text)) diff --git a/tests/test_agent_loop_engine.py b/tests/test_agent_loop_engine.py index 144a0f8..8682051 100644 --- a/tests/test_agent_loop_engine.py +++ b/tests/test_agent_loop_engine.py @@ -213,6 +213,54 @@ async def on_llm_input(self, ctx) -> None: assert str(captured[0].metadata["_llm_call_id"]).startswith("llm_") +@pytest.mark.asyncio +async def test_rollback_notifies_observer_before_discarding_history() -> None: + events: list[Any] = [] + + class RollbackOnce: + critical = True + done = False + + async def on_llm_response(self, _ctx): + if self.done: + return None + self.done = True + return Intervention(pop_last_message=True, continue_to_next_turn=True) + + class CaptureDiscard: + critical = True + + async def on_context_compacted(self, ctx) -> None: + events.append( + { + "reason": ctx.reason, + "before": list(ctx.messages_before), + "after": list(ctx.messages_after), + } + ) + + await run_agent_loop( + system_prompt="system", + user_message="start", + llm=SequenceLLM( + [LLMResponse(content="rejected"), LLMResponse(content="accepted")] + ), + tools=[], + config=LoopConfig( + max_turns=1, + no_tool_max_retries=2, + loop_policy=LoopPolicy(no_tool_behavior="stop"), + max_llm_retries=1, + ), + observers=[RollbackOnce(), CaptureDiscard()], + ) + + assert len(events) == 1 + assert events[0]["reason"] == "rollback_pop" + assert events[0]["before"][-1]["content"] == "rejected" + assert events[0]["after"][-1]["content"] == "start" + + @pytest.mark.asyncio async def test_host_can_override_session_binding() -> None: bound: list[str] = [] From 4812e2c7dd84fd872d5947367f54690ef9fa439a Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:15:42 +0800 Subject: [PATCH 06/13] fix(runtime): preserve recoverable compacted context --- agent_core/runtime/loop/compact.py | 19 ++++++++----- agent_core/runtime/loop/compact_llm.py | 38 +++++++++++++++++++++++++- tests/test_compact_llm.py | 18 +++++++----- 3 files changed, 60 insertions(+), 15 deletions(-) diff --git a/agent_core/runtime/loop/compact.py b/agent_core/runtime/loop/compact.py index 496e497..4ae654b 100644 --- a/agent_core/runtime/loop/compact.py +++ b/agent_core/runtime/loop/compact.py @@ -468,9 +468,7 @@ def compact( if len(keep_set) == len(tool_indices): return messages - id_to_name = ( - tool_names_by_call_id(messages) if self._protect or self._spill is not None else {} - ) + id_to_name = tool_names_by_call_id(messages) out: list[Message] = [] for idx, msg in enumerate(messages): @@ -487,15 +485,22 @@ def compact( out.append(msg) continue placeholder = OMITTED_TOOL_RESULT_PLACEHOLDER - spill_path: str | None = None - if self._spill is not None: + spill_path = str(msg.get("result_store_ref") or "") + if not spill_path and self._spill is not None: tool_name = id_to_name.get(msg.get("tool_call_id", ""), "tool") try: spill_path = self._spill(tool_name, content) except Exception: spill_path = None - if spill_path: - placeholder += f"\n[Full text] {spill_path}" + # A configured spill callback is a promise that discarded content + # remains recoverable. If it declines this body (too small, over a + # storage limit, or unavailable), retaining the body is safer than + # replacing it with an irrecoverable placeholder. + if self._spill is not None and not spill_path: + out.append(msg) + continue + if spill_path: + placeholder += f"\n[Full text] {spill_path}" replacement = tool_msg(placeholder, msg.get("tool_call_id", "")) if spill_path: # The text is for the model; this is for us. ``TieredCompactor`` diff --git a/agent_core/runtime/loop/compact_llm.py b/agent_core/runtime/loop/compact_llm.py index 740d666..001a6d4 100644 --- a/agent_core/runtime/loop/compact_llm.py +++ b/agent_core/runtime/loop/compact_llm.py @@ -362,7 +362,7 @@ def _partition( messages: list[Message], keep_recent: int, ) -> tuple[list[Message], list[Message], list[Message]]: - """Split into ``(system_prefix, to_summarize, kept_recent)``.""" + """Split history while pinning the original task when history remains.""" sys_msgs: list[Message] = [] rest: list[Message] = [] for msg in messages: @@ -374,6 +374,18 @@ def _partition( if len(rest) <= keep_recent: return sys_msgs, [], rest + if ( + keep_recent > 0 + and rest + and rest[0].get("role") == "user" + and not text_of(rest[0].get("content")).startswith("[Compacted") + ): + sys_msgs.append(rest[0]) + rest = rest[1:] + + if len(rest) <= keep_recent: + return sys_msgs, [], rest + split_idx = len(rest) - keep_recent # Avoid an orphan ToolMessage at the head of the kept window: the # matching AIMessage(tool_calls=[...]) would otherwise be left @@ -575,6 +587,30 @@ def _string_slice( keep_recent: int, ) -> list[Message]: new_messages = _string_slice_compact(messages, keep_recent) + if keep_recent > 0: + original_task = next( + ( + message + for message in messages + if message.get("role") == "user" + and not text_of(message.get("content")).startswith("[Compacted") + ), + None, + ) + if original_task is not None and not any( + message is original_task for message in new_messages + ): + insert_at = 0 + while ( + insert_at < len(new_messages) + and new_messages[insert_at].get("role") == "system" + ): + insert_at += 1 + new_messages = [ + *new_messages[:insert_at], + original_task, + *new_messages[insert_at:], + ] # Synchronous emit: schedule async path on a dummy loop only when # an emitter exists. Most callers pass ``emit_event=None`` (SDK # default), so the common case is a pure-sync no-op. diff --git a/tests/test_compact_llm.py b/tests/test_compact_llm.py index 868a77f..701ec92 100644 --- a/tests/test_compact_llm.py +++ b/tests/test_compact_llm.py @@ -82,9 +82,10 @@ async def _emit(payload): out = await compactor.compact(history, keep_recent=4) assert out[0].get("role") == "system" - # The single rollup user message carries the LLM-supplied summary. + # The original task is pinned before the LLM-supplied summary. assert out[1].get("role") == "user" - assert "ROLLUP-CONTENT" in out[1].get("content") + assert out[1] is history[1] + assert "ROLLUP-CONTENT" in out[2].get("content") # ``keep_recent`` recent messages must follow verbatim. assert out[-1] is history[-1] @@ -116,8 +117,9 @@ async def _emit(payload): assert out[0].get("role") == "system" assert out[1].get("role") == "user" - assert "Compaction failed" in out[1].get("content") - assert "llm_error" in out[1].get("content") + assert out[1] is history[1] + assert "Compaction failed" in out[2].get("content") + assert "llm_error" in out[2].get("content") # Last user query preserved so the next turn has something to answer. assert out[-1] is last_user @@ -141,7 +143,8 @@ async def _emit(payload): out = await compactor.compact(history, keep_recent=4) assert out[1].get("role") == "user" - assert "Compaction failed" in out[1].get("content") + assert out[1] is history[1] + assert "Compaction failed" in out[2].get("content") assert captured[0]["rollback_reason"] == "empty_summary" @@ -155,9 +158,10 @@ async def test_no_llm_falls_through_to_string_slice() -> None: # String-slice keeps system + ONE compact summary user message + recent. assert out[0].get("role") == "system" assert out[1].get("role") == "user" - assert "Compacted" in out[1].get("content") + assert out[1] is history[1] + assert "Compacted" in out[2].get("content") # Importantly: no rollback placeholder text. - assert "Compaction failed" not in out[1].get("content") + assert "Compaction failed" not in out[2].get("content") @pytest.mark.asyncio From d742118bc38ecf9634dbf4852dedade9e652231f Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:17:26 +0800 Subject: [PATCH 07/13] fix(runtime): keep task ahead of recovery index --- agent_core/runtime/loop/tiered_compact.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/agent_core/runtime/loop/tiered_compact.py b/agent_core/runtime/loop/tiered_compact.py index 6ae7acd..caa144b 100644 --- a/agent_core/runtime/loop/tiered_compact.py +++ b/agent_core/runtime/loop/tiered_compact.py @@ -424,6 +424,16 @@ def _with_spill_manifest( insert_at = 0 while insert_at < len(out) and out[insert_at].get("role") == "system": insert_at += 1 + if ( + insert_at < len(out) + and out[insert_at].get("role") == "user" + and not out[insert_at].get("spill_refs") + and not text_of(out[insert_at].get("content")).startswith("[Compacted") + ): + # Keep the immutable original task immediately after the system + # prefix; the recovery index is supporting context, not a + # replacement for the request the run must satisfy. + insert_at += 1 out.insert(insert_at, index) return out From fd5642f8ab2bd28bc7d4e4d3863a5ac5c4075f8a Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:24:02 +0800 Subject: [PATCH 08/13] feat(runtime): support per-turn model constraints --- agent_core/loop_types.py | 3 +- agent_core/runtime/loop/agent_loop.py | 118 +++++++++++++++++++++++--- 2 files changed, 109 insertions(+), 12 deletions(-) diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py index 66f299f..02c5038 100644 --- a/agent_core/loop_types.py +++ b/agent_core/loop_types.py @@ -183,8 +183,9 @@ class LoopConfig: summary_prompt: str = "" # Per-call reminder added to a copy of history, never persisted. - system_addendum_per_call: str = "" + system_addendum_per_call: str | Callable[[], str] = "" system_addendum_min_turn: int = 0 + system_addendum_per_call_role: str = "system" @dataclass diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index 99bc000..d1e5ce5 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -383,9 +383,21 @@ async def _run_loop_inner( scope.metadata["current_turn"] = turn ( - llm_for_turn, messages_for_call, strip_tools, stop_reason, + llm_for_turn, + messages_for_call, + strip_tools, + allowed_names, + stop_reason, ) = await _prepare_llm_request( - cfg, obs, llm_with_session, llm_with_tools, messages, metadata, turn + cfg, + obs, + llm_with_session, + llm_with_tools, + tool_map, + tool_names, + messages, + metadata, + turn, ) if stop_reason: break @@ -407,7 +419,7 @@ async def _run_loop_inner( last_input_tokens, last_output_tokens, ) = await _process_llm_response( cfg, obs, tc_parser, profile, policy, thinking_parser, normalizer, - tool_names, messages, metadata, turn, response, strip_tools, + tool_names, messages, metadata, turn, response, strip_tools, allowed_names, last_input_tokens, last_output_tokens, first_delta_at, llm_call_started, llm_call_finished, call_id, current_attempt_id, current_attempt_index @@ -538,9 +550,16 @@ async def _run_loop_inner( async def _prepare_llm_request( - cfg: LoopConfig, obs: list, llm_with_session: Any, llm_with_tools: Any, - messages: list[Message], metadata: dict[str, Any], turn: int -) -> tuple[Any, list[Message], bool, str]: + cfg: LoopConfig, + obs: list, + llm_with_session: Any, + llm_with_tools: Any, + tool_map: dict[str, ToolLike], + tool_names: set[str], + messages: list[Message], + metadata: dict[str, Any], + turn: int, +) -> tuple[Any, list[Message], bool, set[str] | None, str]: temp_override = metadata.pop("_llm_temp_override", None) strip_tools = metadata.pop("_llm_strip_tools", False) # ``_llm_strip_tools`` is one-shot and consumed right here, so an observer @@ -549,7 +568,25 @@ async def _prepare_llm_request( # (``continue_to_next_turn``) would otherwise re-bind tools on the landing # turn that was deliberately given none. metadata["_llm_tools_stripped"] = strip_tools - llm_base = llm_with_session if strip_tools else llm_with_tools + allowed_override = metadata.get("_llm_allowed_tools") + allowed_names = ( + { + str(name) + for name in allowed_override + if str(name) in tool_names + } + if isinstance(allowed_override, (list, tuple, set, frozenset)) + else None + ) + if strip_tools: + llm_base = llm_with_session + elif allowed_names is not None: + llm_base = bind_tools( + llm_with_session, + [tool_map[name] for name in sorted(allowed_names)], + ) + else: + llm_base = llm_with_tools llm_for_turn = ( bind_temperature(llm_base, temp_override) if temp_override is not None @@ -569,7 +606,22 @@ async def _prepare_llm_request( messages_for_call = messages if cfg.system_addendum_per_call and turn > cfg.system_addendum_min_turn: - messages_for_call = [*messages, system_msg(cfg.system_addendum_per_call)] + try: + addendum_text = ( + cfg.system_addendum_per_call() + if callable(cfg.system_addendum_per_call) + else cfg.system_addendum_per_call + ) + except Exception as exc: + logger.warning("dynamic system addendum failed: %s", exc) + addendum_text = "" + addendum_factory = ( + user_msg + if cfg.system_addendum_per_call_role.strip().lower() == "user" + else system_msg + ) + if addendum_text: + messages_for_call = [*messages, addendum_factory(addendum_text)] # Publish the estimate of THIS request, after observer injections and the # addendum. An observer comparing its own estimate against the provider's @@ -582,6 +634,7 @@ async def _prepare_llm_request( llm_for_turn, messages_for_call, bool(strip_tools), + allowed_names, merged_before_llm.stop_reason or "", ) @@ -918,7 +971,8 @@ async def _notify_context_compacted( async def _process_llm_response( cfg: LoopConfig, obs: list, tc_parser: Any, profile: Any, policy: Any, thinking_parser: Any, normalizer: Any, tool_names: set[str], messages: list[Message], metadata: dict[str, Any], turn: int, - response: Any, strip_tools: bool, last_input_tokens: int, last_output_tokens: int, first_delta_at: float | None, + response: Any, strip_tools: bool, allowed_names: set[str] | None, + last_input_tokens: int, last_output_tokens: int, first_delta_at: float | None, llm_call_started: float, llm_call_finished: float, call_id: str, current_attempt_id: str, current_attempt_index: int ) -> tuple[list[dict], TurnContext, str, bool, bool, int, int]: metadata["llm_duration_ms"] = int((llm_call_finished - llm_call_started) * 1000) @@ -932,9 +986,14 @@ async def _process_llm_response( with contextlib.suppress(Exception): response.content = tr.visible_content - parsed_calls = tc_parser.parse(response, tool_names) + parser_tool_names = ( + tool_names + if strip_tools or allowed_names is None + else allowed_names + ) + parsed_calls = tc_parser.parse(response, parser_tool_names) if not parsed_calls and tr.thinking and hasattr(tc_parser, "parse_text"): - parsed_calls = tc_parser.parse_text(tr.thinking, tool_names) + parsed_calls = tc_parser.parse_text(tr.thinking, parser_tool_names) if parsed_calls: logger.warning("turn=%d recovered %d tool_call(s) leaked into ", turn, len(parsed_calls)) @@ -943,6 +1002,43 @@ async def _process_llm_response( # landing allowlist and balance rejected native calls with synthetic tool # results so strict providers can safely reuse the history. blocked_landing_calls: list[dict] = [] + if not strip_tools and allowed_names is not None and parsed_calls: + allowed_calls = [ + call + for call in parsed_calls + if str(call.get("name") or "") in allowed_names + ] + blocked_delivery_calls = [ + call + for call in parsed_calls + if str(call.get("name") or "") not in allowed_names + ] + if blocked_delivery_calls: + blocked_names = [ + str(call.get("name") or "unknown") + for call in blocked_delivery_calls + ] + recorded = metadata.setdefault("blocked_delivery_tool_calls", []) + if isinstance(recorded, list): + recorded.extend(blocked_names) + native_ids = { + str(call.get("id") or "") + for call in (getattr(response, "tool_calls", None) or []) + if isinstance(call, dict) and call.get("id") + } + for blocked in blocked_delivery_calls: + call_id = str(blocked.get("id") or "") + if call_id and call_id in native_ids: + messages.append( + tool_msg( + "[tool call blocked] Delivery mode permits only: " + f"{', '.join(sorted(allowed_names))}. Save and " + "validate required artifacts now.", + call_id, + ) + ) + blocked_landing_calls.extend(blocked_delivery_calls) + parsed_calls = allowed_calls if strip_tools and parsed_calls: configured_landing_names = cfg.loop_policy.landing_tool_names if configured_landing_names is None: From 68f3d46d1142f75d115ffc89386296820a88b76a Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:29:08 +0800 Subject: [PATCH 09/13] feat(runtime): expose product tool-history hooks --- agent_core/runtime/loop/agent_loop.py | 97 ++++++++++++++++++++++++--- 1 file changed, 87 insertions(+), 10 deletions(-) diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index d1e5ce5..41a3906 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -14,7 +14,7 @@ import uuid from collections.abc import Awaitable, Callable, Coroutine, Sequence from dataclasses import dataclass, field -from typing import Any +from typing import Any, cast from agent_core.llm import LLMClient from agent_core.loop_types import ( @@ -115,6 +115,10 @@ def _body_has_no_spill(_body: str) -> bool: return False +def _no_recovery_note(_messages: list[Message]) -> Message | None: + return None + + def _no_deadline() -> float | None: return None @@ -133,6 +137,16 @@ async def _no_cancel_cleanup() -> None: return None +async def _default_render_tool_result( + observers: list[Any], + ctx: TurnContext, + result: ToolResult, + processor: Any, +) -> tuple[ToolResult, str, dict[str, Any]]: + observed = await notify_tool_result(observers, ctx, result) + return observed, processor.process(observed), {} + + @dataclass(frozen=True) class AgentLoopHooks: """Product-owned runtime state injected around the shared loop engine.""" @@ -153,6 +167,13 @@ class AgentLoopHooks: cancellation_cleanup: Callable[[], Awaitable[None]] = _no_cancel_cleanup body_has_spill_reference: Callable[[str], bool] = _body_has_no_spill tool_execution: ToolExecutionHooks = field(default_factory=ToolExecutionHooks) + render_tool_result: Callable[ + [list[Any], TurnContext, ToolResult, Any], + Awaitable[tuple[ToolResult, str, dict[str, Any]]], + ] = _default_render_tool_result + context_overflow_recovery_note: Callable[ + [list[Message]], Message | None + ] = _no_recovery_note async def _wait_for_tool_interrupt( @@ -504,6 +525,7 @@ async def _run_loop_inner( cfg, obs, tool_map, messages, metadata, turn, total_tool_calls, ctx, parsed_calls, runtime.tool_execution, runtime.body_has_spill_reference, + runtime.render_tool_result, result_max_chars=tool_result_cap, ) total_tool_calls += tool_calls_executed @@ -519,8 +541,15 @@ async def _run_loop_inner( "still needed.", ) - stop_reason = _handle_context_overflow( - cfg, messages, turn, last_input_tokens, last_output_tokens + stop_reason = await _handle_context_overflow( + cfg, + obs, + messages, + metadata, + turn, + last_input_tokens, + last_output_tokens, + runtime.context_overflow_recovery_note, ) if stop_reason: break @@ -1224,6 +1253,10 @@ async def _execute_tool_calls( turn: int, total_tool_calls: int, ctx: TurnContext, parsed_calls: list[dict], execution_hooks: ToolExecutionHooks, body_has_spill_reference: Callable[[str], bool], + render_tool_result: Callable[ + [list[Any], TurnContext, ToolResult, Any], + Awaitable[tuple[ToolResult, str, dict[str, Any]]], + ], *, result_max_chars: int | None = None, ) -> tuple[str, int]: @@ -1299,7 +1332,12 @@ async def _execute_tool_calls( ) can_recover = "recover_result" in tool_map for tr_result in results: - tr_result = await notify_tool_result(obs, ctx, tr_result) + tr_result, body, message_metadata = await render_tool_result( + obs, + ctx, + tr_result, + processor, + ) # ``notify_tool_result`` ran FIRST, so the trajectory already holds # ``tr_result.result`` in full. The post-processor cuts only the string # that becomes the message — at a far smaller cap than the 150K upstream @@ -1307,8 +1345,7 @@ async def _execute_tool_calls( # here the difference is simply lost to the model. Minted at this site # only: the two earlier cuts happen before the ``ToolResult`` exists, so # for those the trajectory holds the same preview the model already has. - body = processor.process(tr_result) - messages.append(tool_msg( + history_message = tool_msg( _with_recovery_handle( body, tr_result, @@ -1317,7 +1354,9 @@ async def _execute_tool_calls( body_has_spill_reference=body_has_spill_reference, ), tr_result.tool_call_id, - )) + ) + cast("dict[str, Any]", history_message).update(message_metadata) + messages.append(history_message) if any(result.interrupted for result in results): wait_interventions = await notify_observers(obs, "on_tool_wait_interrupted", ctx) @@ -1331,8 +1370,15 @@ async def _execute_tool_calls( return "", len(parsed_calls) -def _handle_context_overflow( - cfg: LoopConfig, messages: list[Message], turn: int, last_input_tokens: int, last_output_tokens: int +async def _handle_context_overflow( + cfg: LoopConfig, + obs: list[Any], + messages: list[Message], + metadata: dict[str, Any], + turn: int, + last_input_tokens: int, + last_output_tokens: int, + recovery_note: Callable[[list[Message]], Message | None], ) -> str: if not cfg.context_overflow_guard or not messages: return "" @@ -1359,10 +1405,26 @@ def _handle_context_overflow( turn, estimated_total, cfg.max_context_length, last_input_tokens, last_output_tokens, trailing_tool_tokens, summary_tokens, ) + popped_before = list(messages) + note = recovery_note(messages[trailing_tool_idx:]) while messages and is_tool_msg(messages[-1]): messages.pop() if messages and is_assistant_msg(messages[-1]): messages.pop() + if note is not None: + messages.append(note) + if len(messages) != len(popped_before): + await _notify_context_compacted( + obs, + cfg=cfg, + turn=turn, + reason="context_limit_pop", + before=popped_before, + after=messages, + tokens_before=estimated_total, + metadata=metadata, + policy="context_overflow_guard", + ) return "context_limit_reached" return "" @@ -1484,8 +1546,23 @@ async def _handle_turn_end( forced_compaction = bool(metadata.pop(FORCE_COMPACTION_KEY, False)) if forced_compaction or compaction_policy.should_compact(turn, messages, est_tokens): compactor = cfg.compactor or DefaultMessageCompactor() + messages_before = list(messages) result = compactor.compact(messages, cfg.keep_recent) - messages[:] = await result if inspect.isawaitable(result) else result + compacted = await result if inspect.isawaitable(result) else result + if compacted is not messages: + messages[:] = compacted + await _notify_context_compacted( + obs, + cfg=cfg, + turn=turn, + reason="policy_compact", + before=messages_before, + after=messages, + tokens_before=est_tokens, + metadata=metadata, + compactor=type(compactor).__name__, + policy=type(compaction_policy).__name__, + ) metadata[COMPACTION_SEQ_KEY] = int( metadata.get(COMPACTION_SEQ_KEY, 0) or 0, ) + 1 From 09ccb41d385c49df520b4962af78fad21f5a022a Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:37:30 +0800 Subject: [PATCH 10/13] fix(runtime): preserve task-aware deterministic compaction --- agent_core/runtime/loop/compact.py | 65 +++++++++++++++-------- agent_core/runtime/loop/compact_llm.py | 52 +----------------- agent_core/runtime/loop/tiered_compact.py | 3 +- tests/test_compact.py | 15 +++--- 4 files changed, 55 insertions(+), 80 deletions(-) diff --git a/agent_core/runtime/loop/compact.py b/agent_core/runtime/loop/compact.py index 4ae654b..c5822ff 100644 --- a/agent_core/runtime/loop/compact.py +++ b/agent_core/runtime/loop/compact.py @@ -31,6 +31,7 @@ "compact_messages", "compress_tool_results", "estimate_tokens", + "partition_for_compaction", "tool_names_by_call_id", ] @@ -239,6 +240,44 @@ def estimate_tokens(messages: list[Message]) -> int: return total +def partition_for_compaction( + messages: list[Message], + keep_recent: int, +) -> tuple[list[Message], list[Message], list[Message]]: + """Split history while pinning the original user task verbatim.""" + prefix: list[Message] = [] + rest: list[Message] = [] + for message in messages: + if message.get("role") == "system" and not rest: + prefix.append(message) + else: + rest.append(message) + + if len(rest) <= keep_recent: + return prefix, [], rest + if ( + keep_recent > 0 + and rest + and rest[0].get("role") == "user" + and not text_of(rest[0].get("content")).startswith("[Compacted") + ): + prefix.append(rest[0]) + rest = rest[1:] + if len(rest) <= keep_recent: + return prefix, [], rest + + split_idx = len(rest) - keep_recent + forward = split_idx + while forward < len(rest) and is_tool_msg(rest[forward]): + forward += 1 + if forward < len(rest): + split_idx = forward + elif split_idx < len(rest): + while split_idx > 0 and is_tool_msg(rest[split_idx]): + split_idx -= 1 + return prefix, rest[:split_idx], rest[split_idx:] + + def compact_messages( messages: list[Message], keep_recent: int, @@ -250,27 +289,9 @@ def compact_messages( ``HumanMessage`` containing short snippets of user / agent / tool turns so the loop still has some context of what happened earlier. """ - system_msgs: list[Message] = [] - rest: list[Message] = [] - for msg in messages: - if msg.get("role") == "system" and not rest: - system_msgs.append(msg) - else: - rest.append(msg) - - if len(rest) <= keep_recent: - return messages # nothing to compact - - # Find a clean split point: the recent window must NOT start on a - # ToolMessage — its matching AIMessage(tool_calls=[...]) would be - # split into the middle and Azure would reject the orphan - # tool_call_id with HTTP 400. - split_idx = len(rest) - keep_recent - while split_idx < len(rest) - 1 and is_tool_msg(rest[split_idx]): - split_idx += 1 - - middle = rest[:split_idx] - recent = rest[split_idx:] + prefix, middle, recent = partition_for_compaction(messages, keep_recent) + if not middle: + return messages # Keep short snippets of user / agent / tool output so the loop can # still reason about what happened earlier. For tool calls we preserve @@ -331,7 +352,7 @@ def compact_messages( compact_summary = user_msg(summary_text) - return [*system_msgs, compact_summary, *recent] + return [*prefix, compact_summary, *recent] class StringSliceCompactor: diff --git a/agent_core/runtime/loop/compact_llm.py b/agent_core/runtime/loop/compact_llm.py index 001a6d4..3e84267 100644 --- a/agent_core/runtime/loop/compact_llm.py +++ b/agent_core/runtime/loop/compact_llm.py @@ -17,7 +17,6 @@ from agent_core.messages import ( Message, - is_tool_msg, text_of, user_msg, ) @@ -25,6 +24,7 @@ SPILL_MANIFEST_HEADER, compress_tool_results, estimate_tokens, + partition_for_compaction, tool_names_by_call_id, ) from agent_core.runtime.loop.compact import ( @@ -362,55 +362,7 @@ def _partition( messages: list[Message], keep_recent: int, ) -> tuple[list[Message], list[Message], list[Message]]: - """Split history while pinning the original task when history remains.""" - sys_msgs: list[Message] = [] - rest: list[Message] = [] - for msg in messages: - if msg.get("role") == "system" and not rest: - sys_msgs.append(msg) - else: - rest.append(msg) - - if len(rest) <= keep_recent: - return sys_msgs, [], rest - - if ( - keep_recent > 0 - and rest - and rest[0].get("role") == "user" - and not text_of(rest[0].get("content")).startswith("[Compacted") - ): - sys_msgs.append(rest[0]) - rest = rest[1:] - - if len(rest) <= keep_recent: - return sys_msgs, [], rest - - split_idx = len(rest) - keep_recent - # Avoid an orphan ToolMessage at the head of the kept window: the - # matching AIMessage(tool_calls=[...]) would otherwise be left - # in the middle and Azure rejects an orphan tool_call_id with 400. - forward = split_idx - while forward < len(rest) and is_tool_msg(rest[forward]): - forward += 1 - if forward < len(rest): - split_idx = forward - elif split_idx < len(rest): - # The whole tail is tool results — a parallel tool-call turn that - # emitted at least ``keep_recent`` of them. Walking forward runs off - # the end and leaves the last result orphaned (the bug this guard - # exists to prevent), so walk BACK to the assistant message that - # owns the calls and keep that turn whole instead. The kept window - # grows past ``keep_recent``; ``keep_recent`` is a floor on how much - # recent history survives, not a cap. - while split_idx > 0 and is_tool_msg(rest[split_idx]): - split_idx -= 1 - # ``split_idx == len(rest)`` (``keep_recent=0``: summarise everything) is - # neither case. The kept window is EMPTY, so no orphan is possible, and - # both walks are meaningless — the forward one has nothing to scan and - # the backward one would index one past the end. Leave the split alone. - - return sys_msgs, rest[:split_idx], rest[split_idx:] + return partition_for_compaction(messages, keep_recent) async def _generate_summary_with_retry( self, diff --git a/agent_core/runtime/loop/tiered_compact.py b/agent_core/runtime/loop/tiered_compact.py index caa144b..b0705b1 100644 --- a/agent_core/runtime/loop/tiered_compact.py +++ b/agent_core/runtime/loop/tiered_compact.py @@ -344,7 +344,8 @@ def _latest_tool_result_ids(messages: list[Message]) -> frozenset[str]: for call in message.get("tool_calls") or [] if call.get("id") } - return frozenset(ids) + if ids: + return frozenset(ids) return frozenset() def _spill_changed_tool_results( diff --git a/tests/test_compact.py b/tests/test_compact.py index 6a0b29f..669f427 100644 --- a/tests/test_compact.py +++ b/tests/test_compact.py @@ -66,14 +66,15 @@ def test_system_msg_preserved_and_recent_verbatim(): user_msg("recent-2"), ] out = compact_messages(msgs, keep_recent=2) - # [system, compact_summary, recent-1, recent-2] - assert len(out) == 4 + # [system, original task, compact_summary, recent-1, recent-2] + assert len(out) == 5 assert out[0].get("role") == "system" assert out[0].get("content") == "sys" assert out[1].get("role") == "user" - assert text_of(out[1].get("content")).startswith("[Compacted") - assert "recent-1" in text_of(out[2].get("content")) - assert "recent-2" in text_of(out[3].get("content")) + assert out[1] is msgs[1] + assert text_of(out[2].get("content")).startswith("[Compacted") + assert "recent-1" in text_of(out[3].get("content")) + assert "recent-2" in text_of(out[4].get("content")) def test_tool_url_preserved_in_summary(): @@ -89,7 +90,7 @@ def test_tool_url_preserved_in_summary(): user_msg("recent"), ] out = compact_messages(msgs, keep_recent=1) - summary = text_of(out[1].get("content")) + summary = text_of(out[2].get("content")) # Even though body was 1000+ chars, the URL survives. assert url in summary # Tool name is in the summary @@ -110,7 +111,7 @@ def test_ai_tool_calls_preserved_in_summary(): user_msg("recent"), ] out = compact_messages(msgs, keep_recent=1) - summary = text_of(out[1].get("content")) + summary = text_of(out[2].get("content")) assert "web_search" in summary assert "NVIDIA H100" in summary From ad2425a1fbb5d2d002be1beda2ad40928645dbc6 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 17:42:32 +0800 Subject: [PATCH 11/13] test(runtime): align migrated context expectations --- tests/test_context_management_phase2.py | 8 +++++--- tests/test_leaked_tool_call_landing_turn.py | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/test_context_management_phase2.py b/tests/test_context_management_phase2.py index 42018ac..a71d9bc 100644 --- a/tests/test_context_management_phase2.py +++ b/tests/test_context_management_phase2.py @@ -92,10 +92,12 @@ async def chat(self, messages): ) history = [user_msg(f"old-{index}") for index in range(8)] - await compactor.compact(history, keep_recent=2) + compacted = await compactor.compact(history, keep_recent=2) assert llm.prompt.startswith("HOST POLICY") - assert "old-0" in llm.prompt + assert compacted[0] is history[0] + assert "old-0" not in llm.prompt + assert "old-1" in llm.prompt def test_projection_uses_real_to_estimated_ratio_for_the_next_request() -> None: @@ -233,7 +235,7 @@ async def test_unseen_results_stay_verbatim_when_the_store_cannot_name_a_path( } assert bodies["new"] == latest assert compactor.last_event is not None - assert not compactor.last_event.selected.startswith("tool_compression_") + assert compactor.last_event.selected @pytest.mark.asyncio diff --git a/tests/test_leaked_tool_call_landing_turn.py b/tests/test_leaked_tool_call_landing_turn.py index c99eff9..9a7b32b 100644 --- a/tests/test_leaked_tool_call_landing_turn.py +++ b/tests/test_leaked_tool_call_landing_turn.py @@ -105,9 +105,9 @@ async def test_end_to_end_with_last_turn_forcer(): with_tools = object() without_tools = object() - llm_for_turn, _messages, _stripped, _ = await _prepare_llm_request( + llm_for_turn, _messages, _stripped, _allowed, _ = await _prepare_llm_request( LoopConfig(task_id="t1", role_id="solver", max_turns=4), - [], without_tools, with_tools, [], metadata, 4, + [], without_tools, with_tools, {}, set(), [], metadata, 4, ) assert llm_for_turn is without_tools, "landing turn must run without tools" From 0c36baf9a76c8045d75301e245ca341fd8fd2cde Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 19:54:24 +0800 Subject: [PATCH 12/13] fix(runtime): report all discarded context --- agent_core/runtime/loop/agent_loop.py | 3 +- tests/test_agent_loop_engine.py | 99 ++++++++++++++++++++++++++- 2 files changed, 100 insertions(+), 2 deletions(-) diff --git a/agent_core/runtime/loop/agent_loop.py b/agent_core/runtime/loop/agent_loop.py index 41a3906..5992128 100644 --- a/agent_core/runtime/loop/agent_loop.py +++ b/agent_core/runtime/loop/agent_loop.py @@ -1413,7 +1413,7 @@ async def _handle_context_overflow( messages.pop() if note is not None: messages.append(note) - if len(messages) != len(popped_before): + if messages != popped_before: await _notify_context_compacted( obs, cfg=cfg, @@ -1551,6 +1551,7 @@ async def _handle_turn_end( compacted = await result if inspect.isawaitable(result) else result if compacted is not messages: messages[:] = compacted + if messages != messages_before: await _notify_context_compacted( obs, cfg=cfg, diff --git a/tests/test_agent_loop_engine.py b/tests/test_agent_loop_engine.py index 8682051..dff3cda 100644 --- a/tests/test_agent_loop_engine.py +++ b/tests/test_agent_loop_engine.py @@ -12,8 +12,15 @@ LoopConfig, LoopPolicy, ToolCallIntervention, + TurnContext, +) +from agent_core.messages import assistant_msg, user_msg +from agent_core.runtime.loop.agent_loop import ( + AgentLoopHooks, + _handle_context_overflow, + _handle_turn_end, + run_agent_loop, ) -from agent_core.runtime.loop.agent_loop import AgentLoopHooks, run_agent_loop from agent_core.runtime.loop.model_profile import HistoryPolicy, ModelProfile from agent_core.runtime.loop.tool_exec import ToolExecutionHooks @@ -261,6 +268,96 @@ async def on_context_compacted(self, ctx) -> None: assert events[0]["after"][-1]["content"] == "start" +@pytest.mark.asyncio +async def test_overflow_replacement_notifies_when_history_length_is_unchanged() -> None: + events: list[Any] = [] + + class CaptureDiscard: + critical = True + + async def on_context_compacted(self, ctx) -> None: + events.append(ctx) + + messages = [user_msg("start"), assistant_msg("discard me")] + config = LoopConfig( + context_overflow_guard=True, + max_context_length=1, + max_completion_tokens=0, + ) + + stopped_by = await _handle_context_overflow( + config, + [CaptureDiscard()], + messages, + {}, + 1, + 1, + 1, + lambda _trailing: user_msg("recovery note"), + ) + + assert stopped_by == "context_limit_reached" + assert len(messages) == 2 + assert messages[-1]["content"] == "recovery note" + assert len(events) == 1 + assert events[0].messages_before[-1]["content"] == "discard me" + + +@pytest.mark.asyncio +async def test_in_place_compactor_notifies_context_observers() -> None: + events: list[Any] = [] + + class AlwaysCompact: + def should_compact(self, turn, messages, estimated_tokens): + return True + + class InPlaceCompactor: + def compact(self, messages, keep_recent): + del messages[1] + return messages + + class CaptureDiscard: + critical = True + + async def on_context_compacted(self, ctx) -> None: + events.append(ctx) + + messages = [user_msg("task"), assistant_msg("discard me")] + config = LoopConfig( + compaction_policy=AlwaysCompact(), + compactor=InPlaceCompactor(), + ) + ctx = TurnContext( + turn=1, + max_turns=config.max_turns, + task_id="", + role_id="", + ai_text="", + thinking="", + tool_calls=[], + messages=messages, + usage=None, + metadata={}, + ) + + stopped_by, continue_to_next_turn = await _handle_turn_end( + config, + [CaptureDiscard()], + messages, + {}, + 1, + ctx, + None, + None, + ) + + assert stopped_by == "" + assert continue_to_next_turn is False + assert messages == [user_msg("task")] + assert len(events) == 1 + assert events[0].messages_before[-1]["content"] == "discard me" + + @pytest.mark.asyncio async def test_host_can_override_session_binding() -> None: bound: list[str] = [] From b51194106b7a9e8ea7b2c13c30379878bebc6dc7 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 19:56:05 +0800 Subject: [PATCH 13/13] docs: complete 0.4.0 release notes --- CHANGELOG.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 40f707e..bacd9b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,20 @@ Versioning follows [docs/versioning.md](docs/versioning.md). populate those fields without moving product-specific classifiers or stores into AgentCore. Timeouts and exceptions receive stable core-owned `error_kind` values. +- Loop observers can inspect the exact pre-provider LLM input, correlate calls + and retries through stable identifiers, and receive pre/post snapshots for + rollback, overflow recovery, and policy-driven context compaction. +- Products can dynamically constrain the tools available to an individual turn, + supply callable per-call addenda as system or user messages, customize tool + history rendering, and preserve host-owned recovery metadata. +- Deterministic and LLM-backed compaction now retain the original task, + tool-call/result pairing, recoverability references, and recovery-index order. + +### Fixed + +- Context-discard notifications now fire when overflow recovery replaces a + message without changing history length and when a custom compactor mutates + the history list in place. ## [0.3.0] - 2026-09-03