From 8462a8590514181a031536428a6d33b0ef3cdc87 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Mon, 31 Aug 2026 13:51:29 +0800 Subject: [PATCH 1/3] refactor: extract LLM and loop runtime contracts --- README.md | 17 +- agent_core/__init__.py | 4 + agent_core/llm.py | 93 +++++++ agent_core/loop_types.py | 570 +++++++++++++++++++++++++++++++++++++++ tests/test_llm.py | 60 +++++ tests/test_loop_types.py | 265 ++++++++++++++++++ 6 files changed, 1002 insertions(+), 7 deletions(-) create mode 100644 agent_core/llm.py create mode 100644 agent_core/loop_types.py create mode 100644 tests/test_llm.py create mode 100644 tests/test_loop_types.py diff --git a/README.md b/README.md index 0f4e17c..59b016d 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,9 @@ Version `0.1.x` contains the converged foundation layer: - compaction policy and deterministic compactor; - context-budget estimation and non-blocking tokenizer access; - message trimming. +- provider-neutral LLM response, stream, and client contracts; +- loop configuration, lifecycle contexts, observer protocol, intervention + merging, and observer dispatch helpers. The initial extraction is based on the already-merged integration branches: @@ -26,11 +29,10 @@ The initial extraction is based on the already-merged integration branches: Those revisions are provenance, not runtime dependencies. AgentCore tests and builds without either product checkout. -The LLM call runtime and agent loop are the next migration slice. They remain -in the products until their remaining dependencies (`loop_types`, tool -execution, model profiles, retry classification, and runtime hooks) have a -product-neutral boundary. Moving those files before that boundary exists would -only hide product coupling inside this package. +The LLM call runtime and agent loop remain in the products until tool +execution, model profiles, retry classification, execution-context storage, +and runtime hooks have product-neutral boundaries. Moving those files before +that boundary exists would only hide product coupling inside this package. ## Repository boundary @@ -106,8 +108,9 @@ edit both products' core copies, that is evidence it belongs here. 1. **Foundation** (this version): messages, token estimation, compaction, context budget, trimming. -2. **Runtime contracts:** errors, LLM/tool protocols, loop types, execution - context, retry classification, and explicit product hooks. +2. **Runtime contracts** (in progress): LLM protocols and loop types are now + shared; errors, execution-context storage, retry classification, and + explicit product hooks remain. 3. **LLM runtime:** binding, calls, streaming, response normalization, runaway recovery, and the public `llm_client` facade. 4. **Agent loop:** model/tool parsing, tool execution, and `agent_loop`. diff --git a/agent_core/__init__.py b/agent_core/__init__.py index 707cbc6..f64a392 100644 --- a/agent_core/__init__.py +++ b/agent_core/__init__.py @@ -1,5 +1,6 @@ """Product-neutral building blocks for Apodex agent runtimes.""" +from agent_core.llm import LLMClient, LLMResponse, StreamDelta from agent_core.messages import ( Message, ToolCall, @@ -10,7 +11,10 @@ ) __all__ = [ + "LLMClient", + "LLMResponse", "Message", + "StreamDelta", "ToolCall", "assistant_msg", "system_msg", diff --git a/agent_core/llm.py b/agent_core/llm.py new file mode 100644 index 0000000..6d0437e --- /dev/null +++ b/agent_core/llm.py @@ -0,0 +1,93 @@ +"""LLM client contracts — provider-agnostic chat completion interface.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from typing import Any, Protocol, runtime_checkable + +from agent_core.messages import Message, ToolCall + + +@dataclass +class LLMResponse: + """One non-streaming completion result.""" + + content: Any = "" # str | list[dict] + tool_calls: list[ToolCall] = field(default_factory=list[ToolCall]) + reasoning_content: str = "" + finish_reason: str = "" + model: str = "" + usage: dict[str, int] = field(default_factory=dict[str, int]) + response_metadata: dict[str, Any] = field(default_factory=dict[str, Any]) + + +@dataclass +class StreamDelta: + """Incremental update during a streaming completion.""" + + content: str = "" + reasoning_content: str = "" + tool_call_deltas: list[dict[str, Any]] = field(default_factory=list[dict[str, Any]]) + # Terminal metadata. Providers send these late in the stream — usage on a + # separate ``choices=[]`` chunk (OpenAI ``include_usage``), finish_reason on + # the last content chunk. Carried here so the stream assembler can put them + # on the final ``LLMResponse`` (else streaming usage/billing reads 0 and + # ``finish_reason="length"`` is invisible to truncation/rollback observers). + usage: dict[str, int] = field(default_factory=dict[str, int]) + finish_reason: str = "" + model: str = "" + # Vendor label of the leg serving this stream, stamped by + # ``LLMFallbackChain.stream`` (constant once the chain commits to an + # entry — failover only fires before the first yield). The stream + # assembler folds it into ``LLMResponse.response_metadata`` so per-call + # billing attribution works for streamed calls too — without this the + # streaming path had no channel for the provider and every billing + # consumer read an empty vendor, which split one model's usage across + # a ``provider=""`` bucket and a named bucket. + provider: str = "" + + +@runtime_checkable +class LLMClient(Protocol): + """Minimal async chat completion client.""" + + # Stays a settable attribute: LLMClient is not purely structural — concrete + # clients such as OpenAIClient subclass it and assign ``self.model`` in + # __init__, so a read-only property here would break them at runtime. + model: str + + async def chat( + self, + messages: list[Message], + *, + tools: list[dict[str, Any]] | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + extra_headers: dict[str, str] | None = None, + timeout: float | None = None, + ) -> LLMResponse: + """Send one non-streaming completion request. ``tools`` is a list + of OpenAI function-schema dicts; ``None`` runs without tools.""" + ... + + def stream( + self, + messages: list[Message], + *, + tools: list[dict[str, Any]] | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + extra_headers: dict[str, str] | None = None, + timeout: float | None = None, + ) -> AsyncIterator[StreamDelta]: + """Stream a completion as a sequence of incremental ``StreamDelta``s. + + The terminal ``LLMResponse`` (with assembled content + finalised + tool_calls + usage) is accessible via :meth:`last_response` after the + stream is exhausted. + """ + ... + + +__all__ = ["LLMClient", "LLMResponse", "StreamDelta"] diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py new file mode 100644 index 0000000..e58e143 --- /dev/null +++ b/agent_core/loop_types.py @@ -0,0 +1,570 @@ +"""Loop type contracts for the agent-loop engine.""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass, field +from typing import Any, Literal, Protocol, cast, runtime_checkable + +from agent_core.messages import Message + +logger = logging.getLogger(__name__) + +# Absolute monotonic soft deadline stored in execution-scope metadata. +WALL_DEADLINE_MONOTONIC_KEY = "wall_deadline_monotonic" + + +def deadline_remaining_s(metadata: Mapping[str, Any] | None) -> float | None: + """Return seconds to a structural lease or absolute soft deadline. + + Execution-context storage belongs to each host product. Passing only its + metadata keeps this frozen contract module independent of either host. + """ + deadline = (metadata or {}).get(WALL_DEADLINE_MONOTONIC_KEY) + remaining_s = cast(Callable[[], Any] | None, getattr(deadline, "remaining_s", None)) + if callable(remaining_s): + try: + return float(remaining_s()) + except Exception: + return None + if not isinstance(deadline, (int, float)): + return None + return float(deadline) - time.monotonic() + + +@dataclass(frozen=True) +class LoopPolicy: + """Workflow-specific behavior injected into the generic loop.""" + + phase_id: str = "" + no_tool_behavior: Literal["stop", "nudge"] = "nudge" + no_tool_nudge_message: str = "" + terminal_tool_names: tuple[str, ...] = () + # Tool calls recovered from text on a tool-free landing turn are denied by + # default. Workflows may explicitly allow bounded, end-of-run actions such + # as ``collect_reports`` or ``write_file``. ``None`` preserves the legacy + # terminal-tool allowlist; an empty tuple deliberately allows no tools. + # + # CONTRACT-ONLY IN THIS REPO (loop-v1 freeze, 2026-08-27): the landing-turn + # recovery path that reads this lives in the other consumer. Frozen here so + # both repos type-check against one ``LoopPolicy``; leaving it ``None`` + # keeps this repo's behavior unchanged. + landing_tool_names: tuple[str, ...] | None = None + + +@dataclass +class LoopConfig: + max_turns: int = 50 + max_tool_calls_per_turn: int = 5 + tool_timeout: int = 120 + llm_timeout: int = 180 + # First streamed chunk timeout; None defers to environment configuration. + first_chunk_timeout: float | None = None + # Abort reasoning-only streams after either enabled bound. + reasoning_only_timeout_s: float | None = None + reasoning_only_max_tokens: int | None = None + # Total budget across admission, attempts, backoff, and recovery. + logical_call_timeout_s: float | None = None + context_token_limit: int = 120_000 + compact_after_turns: int = 12 + keep_recent: int = 16 + no_tool_max_retries: int = 2 + # Continuations offered to a reply the output cap cut off mid-sentence. + # Separate from ``no_tool_max_retries`` because the two are opposite signals: + # a tool-less turn is the model choosing to stop, a truncated one is the + # model being stopped, so a truncation must not spend the nudge budget. + truncation_max_continuations: int = 2 + max_llm_retries: int = 5 + # Fixed retry delay; None uses exponential backoff. + retry_wait_fixed: int | None = None + task_id: str = "" + # Optional gateway affinity key; task_id remains the runtime scope. + llm_session_id: str = field(default="", kw_only=True) + role_id: str = "" + loop_policy: LoopPolicy = field(default_factory=LoopPolicy) + # ToolMessage character cap; None preserves full output. + tool_result_max_chars: int | None = None + # Any avoids importing runtime compaction interfaces into this type layer. + compactor: Any = None + compaction_policy: Any = None + tool_result_post_processor: Any = None + + # Stop before tool output makes the next LLM plus summary request overflow. + context_overflow_guard: bool = False + max_context_length: int = 262_144 + max_completion_tokens: int = 32_768 + summary_prompt: str = "" + + # Per-call reminder added to a copy of history, never persisted. + system_addendum_per_call: str = "" + system_addendum_min_turn: int = 0 + + +@dataclass +class TurnContext: + turn: int + max_turns: int + task_id: str + role_id: str + ai_text: str + thinking: str + tool_calls: list[dict[str, Any]] + messages: list[Message] + usage: dict[str, int] | None + metadata: dict[str, Any] + # Reasoning recovered from tags leaked into visible content. + leaked_reasoning: str = "" + # Native content blocks retained for signed/encrypted replay. + thinking_blocks: list[Any] = field(default_factory=list[Any]) + # Calls parsed on a tool-schema-free landing turn but denied by the + # workflow's landing allowlist. Keeping these separate from ``tool_calls`` + # lets observers distinguish "the model answered in plain text" from "the + # runtime blocked an attempted action" instead of inferring from an empty + # list and accidentally finalising or retrying the leaked call. + # + # CONTRACT-ONLY IN THIS REPO (loop-v1 freeze, 2026-08-27): no producer here + # yet, so this stays empty and ``tool_schemas_stripped`` stays ``False``. + # An observer may read them today; it will simply always see the defaults. + blocked_tool_calls: list[dict[str, Any]] = field(default_factory=list[dict[str, Any]]) + # Whether tool schemas were withheld from the request that produced this + # turn — the condition under which a leaked call can appear in + # ``blocked_tool_calls``. + tool_schemas_stripped: bool = False + + +@dataclass +class LLMDeltaContext: + turn: int + max_turns: int + task_id: str + role_id: str + delta: str + accumulated_text: str + delta_index: int + metadata: dict[str, Any] + # Provider-native reasoning, kept separate from visible content. + thinking_delta: str = "" + # Partial JSON args keyed by call id, or index before an id arrives. + tool_call_args_chunks: list[dict[str, Any]] = field(default_factory=list[dict[str, Any]]) + # Identifies deltas from attempts that may later be discarded. + attempt_id: str = "" + attempt_index: int = 1 + call_id: str = "" + + +# Attempt outcome describes delivery; health details live in reason fields. +ATTEMPT_ACCEPTED = "accepted" +ATTEMPT_ACCEPTED_DEGRADED = "accepted_degraded" +ATTEMPT_DISCARDED = "discarded" +ATTEMPT_FAILED = "failed" + +# Both outcomes deliver bytes to the loop and must retain streamed state. +DELIVERED_ATTEMPT_OUTCOMES = frozenset( + { + ATTEMPT_ACCEPTED, + ATTEMPT_ACCEPTED_DEGRADED, + } +) + + +@dataclass +class LLMAttemptContext: + """Summary-only lifecycle snapshot for one provider attempt.""" + + turn: int + max_turns: int + task_id: str + role_id: str + call_id: str + attempt_id: str + attempt_index: int + phase: str + outcome: str = "" + reason: str = "" + recovery_action: str = "" + duration_ms: int = 0 + ttft_ms: int | None = None + usage: dict[str, int] | None = None + finish_reason: str = "" + visible_chars: int = 0 + reasoning_chars: int = 0 + tool_calls_count: int = 0 + max_tokens: int | None = None + # Sampling actually used for this attempt. Distinct from the provider-level + # ``anthropic_thinking_budget`` / ``thinking_budget_tokens`` config in + # ``infra/``: those say what was *requested* for the run, these report what + # this one attempt ran with — which a runaway ladder or a reduced-cap retry + # changes mid-call. + # + # CONTRACT-ONLY IN THIS REPO (loop-v1 freeze, 2026-08-27): the attempt + # emitter here does not populate them yet, so they stay at their defaults. + thinking_mode: str = "profile_default" + thinking_budget: int | None = None + error_type: str = "" + metadata: dict[str, Any] = field(default_factory=dict[str, Any]) + + +@dataclass +class ToolResult: + name: str + args: dict[str, Any] + result: str + duration_ms: int + tool_call_id: str + is_error: bool + # Interrupted results remain in history to preserve tool-call pairing. + interrupted: bool = False + + +@dataclass +class Intervention: + inject_messages: list[str] | None = None + stop_reason: str | None = None + skip_tool_execution: bool = False + # Applied after message injection and before continuing the turn. + pop_last_message: bool = False + continue_to_next_turn: bool = False + + +@dataclass +class ToolCallIntervention: + """Tool-call rewrite, short-circuit result, and metadata updates.""" + + rewrite_args: dict[str, Any] | None = None + skip_with_result: str | None = None + metadata_updates: dict[str, Any] | None = None + + +@dataclass +class AgentLoopResult: + messages: list[Message] + final_content: str = "" + turns_used: int = 0 + tool_calls_count: int = 0 + stopped_by: str = "" + metadata: dict[str, Any] = field(default_factory=dict[str, Any]) + + +@dataclass +class CompactionEvent: + """What one compaction did, for the durable record. + + Compaction is the one history rewrite that leaves no trace: it replaces + messages in place, so a trajectory reading only the post-compaction history + shows the rollup with nothing to compare it against, and the replaced turns + are simply gone. ``selected`` and the token pair say how much was freed; + ``summary`` is the only field that says what survived. + + Three outcomes have to stay distinguishable, because two of them produce an + empty ``summary``: + + * a tier that does not summarise at all (Tier 1 blanking, the + ``tool_compression_*`` fallbacks) — ``summary`` and ``rollback_reason`` + both empty; + * a summariser that ran and produced text — ``summary`` set; + * a summariser that ran and **failed**, whose deterministic slice can still + win — ``summary`` empty but ``rollback_reason`` set. + + Without the third, a failed summariser is indistinguishable from one that + never ran, which is precisely the confusion this record exists to remove. + """ + + turn: int + seq: int + selected: str + tokens_before: int + tokens_after: int + relief_met: bool + spill_refs: int + #: Number of summariser calls made by the selected tier. Zero means the + #: selected compaction path did not run the summariser. + attempts: int = 0 + summary: str = "" + #: Why the summariser rolled back (``llm_error`` / + #: ``llm_error_permanent`` / ``empty_summary``), or empty when it did not + #: run or did not fail. + rollback_reason: str = "" + + +@runtime_checkable +class LoopObserver(Protocol): + """Structural contract implemented by agent-loop observers.""" + + critical: bool + + async def on_loop_start(self, config: LoopConfig) -> None: ... + + async def on_llm_delta(self, ctx: LLMDeltaContext) -> Intervention | None: ... + + async def on_llm_attempt( + self, + ctx: LLMAttemptContext, + ) -> Intervention | None: ... + + async def on_llm_response(self, ctx: TurnContext) -> Intervention | None: ... + + async def on_tool_call( + self, + ctx: TurnContext, + tool_call: dict[str, Any], + ) -> ToolCallIntervention | None: ... + + async def on_tool_result( + self, + ctx: TurnContext, + result: ToolResult, + ) -> ToolResult | None: ... + + async def on_turn_end(self, ctx: TurnContext) -> Intervention | None: ... + + async def on_loop_end(self, result: AgentLoopResult) -> None: ... + + +class BaseObserver: + """No-op observer base; override only required hooks.""" + + critical: bool = False + + async def on_loop_start(self, config: LoopConfig) -> None: + pass + + async def on_llm_delta(self, ctx: LLMDeltaContext) -> Intervention | None: + return None + + async def on_llm_attempt( + self, + ctx: LLMAttemptContext, + ) -> Intervention | None: + return None + + async def on_llm_response(self, ctx: TurnContext) -> Intervention | None: + return None + + async def on_tool_call( + self, + ctx: TurnContext, + tool_call: dict[str, Any], + ) -> ToolCallIntervention | None: + return None + + async def on_tool_result( + self, + ctx: TurnContext, + result: ToolResult, + ) -> ToolResult | None: + return None + + async def on_turn_end(self, ctx: TurnContext) -> Intervention | None: + return None + + 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_loop_end(self, result: AgentLoopResult) -> None: + pass + + async def on_loop_cancelled(self) -> None: + """Release resources when cancellation bypasses ``on_loop_end``.""" + + +# Prevent GC of fire-and-forget observer tasks. +_background_tasks: set[asyncio.Task[None]] = set() + + +# Log each observer-hook failure once at warning level. +_warned_observer_errors: set[tuple[str, str]] = set() + + +def _handle_observer_error( + observer: Any, + method: str, + exc: BaseException, +) -> None: + """Log an observer crash without propagating it into the loop.""" + obs_class = type(observer).__name__ + key = (obs_class, method) + if key in _warned_observer_errors: + logger.debug( + "Observer %s.%s raised (suppressed)", + obs_class, + method, + exc_info=True, + ) + return + _warned_observer_errors.add(key) + logger.warning( + "Observer %s.%s raised: %s — subsequent failures DEBUG only", + obs_class, + method, + exc, + exc_info=True, + ) + + +async def notify_observers( + observers: list[Any], + method: str, + *args: Any, + **kwargs: Any, +) -> list[Intervention]: + """Run hooks, awaiting critical observers and isolating hook errors. + + ``on_loop_end`` drains passive hooks so their side effects are visible on return. + """ + interventions: list[Intervention] = [] + + for obs in observers: + fn = getattr(obs, method, None) + if fn is None: + continue + + if getattr(obs, "critical", False): + try: + rv = await fn(*args, **kwargs) + if isinstance(rv, Intervention): + interventions.append(rv) + except Exception as exc: + _handle_observer_error(obs, method, exc) + else: + + async def _run( + observer: Any = obs, + m: str = method, + f: Callable[..., Awaitable[Any]] = fn, + a: tuple[Any, ...] = args, + kw: dict[str, Any] = kwargs, + ) -> None: + try: + await f(*a, **kw) + except Exception as exc: + _handle_observer_error(observer, m, exc) + + task = asyncio.create_task(_run()) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + + if method == "on_loop_end": + await drain_background_observers() + + return interventions + + +async def drain_background_observers() -> None: + """Drain outstanding passive observer tasks.""" + pending = [task for task in _background_tasks if not task.done()] + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + +def merge_interventions(interventions: list[Intervention]) -> Intervention: + """Merge messages, take the first stop reason, and OR boolean controls.""" + all_messages: list[str] = [] + stop_reason: str | None = None + skip: bool = False + pop_last: bool = False + continue_turn: bool = False + + for iv in interventions: + if iv.inject_messages: + all_messages.extend(iv.inject_messages) + if stop_reason is None and iv.stop_reason is not None: + stop_reason = iv.stop_reason + if iv.skip_tool_execution: + skip = True + if iv.pop_last_message: + pop_last = True + if iv.continue_to_next_turn: + continue_turn = True + + return Intervention( + inject_messages=all_messages if all_messages else None, + stop_reason=stop_reason, + skip_tool_execution=skip, + pop_last_message=pop_last, + continue_to_next_turn=continue_turn, + ) + + +async def notify_tool_call( + observers: list[Any], + ctx: TurnContext, + tool_call: dict[str, Any], +) -> ToolCallIntervention: + """Merge tool-call hooks; last rewrite and first skip win.""" + rewrite: dict[str, Any] | None = None + skip: str | None = None + meta_updates: dict[str, Any] = {} + + for obs in observers: + fn = getattr(obs, "on_tool_call", None) + if fn is None: + continue + try: + rv = await fn(ctx, tool_call) + except Exception as exc: + _handle_observer_error(obs, "on_tool_call", exc) + continue + if rv is None: + continue + if rv.rewrite_args is not None: + rewrite = rv.rewrite_args + if skip is None and rv.skip_with_result is not None: + skip = rv.skip_with_result + if rv.metadata_updates: + meta_updates.update(rv.metadata_updates) + + return ToolCallIntervention( + rewrite_args=rewrite, + skip_with_result=skip, + metadata_updates=meta_updates or None, + ) + + +async def notify_tool_result( + observers: list[Any], + ctx: TurnContext, + result: ToolResult, +) -> ToolResult: + """Chain tool-result hooks with last-mutation-wins semantics.""" + current = result + for obs in observers: + fn = getattr(obs, "on_tool_result", None) + if fn is None: + continue + try: + rv = await fn(ctx, current) + except Exception as exc: + _handle_observer_error(obs, "on_tool_result", exc) + continue + if rv is not None: + current = rv + return current + + +__all__ = [ + "ATTEMPT_ACCEPTED", + "ATTEMPT_ACCEPTED_DEGRADED", + "ATTEMPT_DISCARDED", + "ATTEMPT_FAILED", + "DELIVERED_ATTEMPT_OUTCOMES", + "WALL_DEADLINE_MONOTONIC_KEY", + "AgentLoopResult", + "BaseObserver", + "CompactionEvent", + "Intervention", + "LLMAttemptContext", + "LLMDeltaContext", + "LoopConfig", + "LoopObserver", + "LoopPolicy", + "ToolCallIntervention", + "ToolResult", + "TurnContext", + "deadline_remaining_s", + "merge_interventions", + "notify_observers", +] diff --git a/tests/test_llm.py b/tests/test_llm.py new file mode 100644 index 0000000..299b340 --- /dev/null +++ b/tests/test_llm.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator + +from agent_core.llm import LLMClient, LLMResponse, StreamDelta +from agent_core.messages import Message + + +def test_response_containers_do_not_share_mutable_defaults() -> None: + first = LLMResponse() + second = LLMResponse() + first.tool_calls.append( + { + "id": "call-1", + "type": "function", + "function": {"name": "search", "arguments": "{}"}, + } + ) + first.usage["input_tokens"] = 1 + + assert second.tool_calls == [] + assert second.usage == {} + assert second.response_metadata == {} + + +def test_stream_delta_carries_terminal_metadata() -> None: + delta = StreamDelta( + content="done", + usage={"output_tokens": 3}, + finish_reason="stop", + model="test-model", + provider="test-provider", + ) + assert delta.usage == {"output_tokens": 3} + assert delta.finish_reason == "stop" + assert delta.provider == "test-provider" + + +def test_structural_client_satisfies_runtime_protocol() -> None: + class Client: + model = "test-model" + + async def chat( + self, + messages: list[Message], + **kwargs: object, + ) -> LLMResponse: + return LLMResponse(content="ok") + + async def stream_impl(self) -> AsyncIterator[StreamDelta]: + yield StreamDelta(content="ok") + + def stream( + self, + messages: list[Message], + **kwargs: object, + ) -> AsyncIterator[StreamDelta]: + return self.stream_impl() + + assert isinstance(Client(), LLMClient) diff --git a/tests/test_loop_types.py b/tests/test_loop_types.py new file mode 100644 index 0000000..178c5da --- /dev/null +++ b/tests/test_loop_types.py @@ -0,0 +1,265 @@ +"""loop-v1 §4 — the three merge rule sets. + +Pure functions, no LLM: whatever else drifts between the two repos, these +rules decide what an observer can actually make the loop do, so they are the +cheapest high-value thing to pin. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +import agent_core.loop_types as lt + +Intervention = lt.Intervention +ToolCallIntervention = lt.ToolCallIntervention +ToolResult = lt.ToolResult +merge_interventions = lt.merge_interventions + + +def _tool_result(name: str = "t", result: str = "r"): + return ToolResult( + name=name, + args={}, + result=result, + duration_ms=0, + tool_call_id="c1", + is_error=False, + ) + + +# --- merge_interventions (§4.2) ------------------------------------------- + + +def test_inject_messages_concatenate_in_observer_order(): + merged = merge_interventions( + [ + Intervention(inject_messages=["a", "b"]), + Intervention(inject_messages=None), + Intervention(inject_messages=["c"]), + ] + ) + assert merged.inject_messages == ["a", "b", "c"] + + +def test_no_inject_messages_stays_none_not_empty_list(): + """``None`` and ``[]`` must not be conflated: the loop treats a list as + 'inject this', so an empty list would be a request to inject nothing.""" + assert merge_interventions([Intervention(), Intervention()]).inject_messages is None + + +def test_stop_reason_first_non_none_wins(): + merged = merge_interventions( + [ + Intervention(), + Intervention(stop_reason="first"), + Intervention(stop_reason="second"), + ] + ) + assert merged.stop_reason == "first" + + +@pytest.mark.parametrize( + "flag", + ["skip_tool_execution", "pop_last_message", "continue_to_next_turn"], +) +def test_boolean_flags_are_any_true(flag: str): + merged = merge_interventions( + [ + Intervention(), + Intervention(**{flag: True}), + Intervention(), + ] + ) + assert getattr(merged, flag) is True + assert getattr(merge_interventions([Intervention()] * 3), flag) is False + + +def test_empty_merge_is_the_neutral_intervention(): + merged = merge_interventions([]) + assert merged.inject_messages is None + assert merged.stop_reason is None + assert not merged.skip_tool_execution + assert not merged.pop_last_message + assert not merged.continue_to_next_turn + + +# --- notify_tool_call (§4.3) ---------------------------------------------- + + +class _ToolCallObserver: + critical = True + + def __init__(self, iv): + self._iv = iv + + async def on_tool_call(self, ctx, tool_call): + return self._iv + + +@pytest.mark.asyncio +async def test_rewrite_args_last_writer_wins(): + merged = await lt.notify_tool_call( + [ + _ToolCallObserver(ToolCallIntervention(rewrite_args={"v": 1})), + _ToolCallObserver(ToolCallIntervention(rewrite_args={"v": 2})), + ], + None, + {"name": "t"}, + ) + assert merged.rewrite_args == {"v": 2} + + +@pytest.mark.asyncio +async def test_skip_with_result_first_writer_wins(): + """Opposite of ``rewrite_args`` on purpose: once a call is skipped, a + later rewrite of its arguments would have nothing to apply to.""" + merged = await lt.notify_tool_call( + [ + _ToolCallObserver(ToolCallIntervention(skip_with_result="first")), + _ToolCallObserver(ToolCallIntervention(skip_with_result="second")), + ], + None, + {"name": "t"}, + ) + assert merged.skip_with_result == "first" + + +@pytest.mark.asyncio +async def test_metadata_updates_merge_dict_wise(): + merged = await lt.notify_tool_call( + [ + _ToolCallObserver(ToolCallIntervention(metadata_updates={"a": 1})), + _ToolCallObserver(ToolCallIntervention(metadata_updates={"b": 2})), + ], + None, + {"name": "t"}, + ) + assert merged.metadata_updates == {"a": 1, "b": 2} + + +@pytest.mark.asyncio +async def test_tool_call_observer_crash_cannot_break_dispatch(): + """§3.4 — a buggy observer must never crash a tool dispatch.""" + + class _Boom: + critical = True + + async def on_tool_call(self, ctx, tool_call): + raise RuntimeError("observer bug") + + merged = await lt.notify_tool_call( + [_Boom(), _ToolCallObserver(ToolCallIntervention(rewrite_args={"v": 9}))], + None, + {"name": "t"}, + ) + assert merged.rewrite_args == {"v": 9} + + +# --- notify_tool_result (§4.4) -------------------------------------------- + + +@pytest.mark.asyncio +async def test_tool_result_mutation_chains_through_observers(): + """Each non-None return replaces what the next observer sees.""" + + class _Appender: + critical = True + + def __init__(self, suffix: str): + self.suffix = suffix + self.seen: list[str] = [] + + async def on_tool_result(self, ctx, result): + self.seen.append(result.result) + return _tool_result(result=result.result + self.suffix) + + a, b = _Appender("-a"), _Appender("-b") + out = await lt.notify_tool_result([a, b], None, _tool_result(result="base")) + assert out.result == "base-a-b" + assert a.seen == ["base"], "first observer sees the original" + assert b.seen == ["base-a"], "second observer sees the first's mutation" + + +@pytest.mark.asyncio +async def test_tool_result_none_return_is_read_only(): + class _Reader: + critical = True + + async def on_tool_result(self, ctx, result): + return None + + out = await lt.notify_tool_result([_Reader()], None, _tool_result(result="base")) + assert out.result == "base" + + +@pytest.mark.asyncio +async def test_observer_without_the_hook_is_skipped_not_an_error(): + """Hooks are probed, so an observer implementing only some of them is + a first-class citizen (this is what keeps optional hooks optional).""" + + class _Bare: + critical = True + + out = await lt.notify_tool_result([_Bare()], None, _tool_result(result="base")) + assert out.result == "base" + + +def test_deadline_accepts_structural_lease_and_absolute_time(monkeypatch) -> None: + class Lease: + def remaining_s(self) -> float: + return 12.5 + + metadata = {lt.WALL_DEADLINE_MONOTONIC_KEY: Lease()} + assert lt.deadline_remaining_s(metadata) == 12.5 + + metadata[lt.WALL_DEADLINE_MONOTONIC_KEY] = 125.0 + monkeypatch.setattr(lt.time, "monotonic", lambda: 100.0) + assert lt.deadline_remaining_s(metadata) == 25.0 + + +def test_deadline_rejects_bad_shapes_and_failing_lease() -> None: + class BrokenLease: + def remaining_s(self) -> float: + raise RuntimeError("expired backing store") + + key = lt.WALL_DEADLINE_MONOTONIC_KEY + assert lt.deadline_remaining_s({key: BrokenLease()}) is None + assert lt.deadline_remaining_s({key: object()}) is None + assert lt.deadline_remaining_s(None) is None + + +@pytest.mark.asyncio +async def test_passive_observer_is_non_blocking_and_return_is_ignored() -> None: + entered = asyncio.Event() + release = asyncio.Event() + + class Passive(lt.BaseObserver): + async def on_llm_response(self, ctx): + entered.set() + await release.wait() + return lt.Intervention(stop_reason="must-be-ignored") + + interventions = await lt.notify_observers([Passive()], "on_llm_response", None) + assert interventions == [] + await asyncio.wait_for(entered.wait(), timeout=1) + release.set() + await lt.drain_background_observers() + + +def test_legacy_observer_satisfies_runtime_protocol() -> None: + class Legacy: + critical = True + + async def on_loop_start(self, config): ... + async def on_llm_delta(self, ctx): ... + async def on_llm_attempt(self, ctx): ... + async def on_llm_response(self, ctx): ... + async def on_tool_call(self, ctx, tool_call): ... + async def on_tool_result(self, ctx, result): ... + async def on_turn_end(self, ctx): ... + async def on_loop_end(self, result): ... + + assert isinstance(Legacy(), lt.LoopObserver) From 5b7412f6f3fc71143a9503bac42c8457a3dc63f2 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Mon, 31 Aug 2026 14:32:09 +0800 Subject: [PATCH 2/3] review: restore deadline structural-check rationale and complete __all__ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `deadline_remaining_s` accepts either a renewable-lease view exposing `remaining_s()` or a bare monotonic instant, and the probe is deliberately structural: importing the concrete lease type to run an `isinstance` check would pull a product dependency into this frozen contract module and break the `agent_core` import closure. The port compressed that reasoning away, leaving the next reader with an apparent cleanup opportunity that is actually a constraint. Restore it. Also export the three dispatch helpers a host must call — `notify_tool_call`, `notify_tool_result`, `drain_background_observers` — which `__all__` omitted while listing `notify_observers` and `merge_interventions`, and repair the README scope list punctuation. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 2 +- agent_core/loop_types.py | 21 +++++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 59b016d..271d683 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ Version `0.1.x` contains the converged foundation layer: - token-estimation helpers; - compaction policy and deterministic compactor; - context-budget estimation and non-blocking tokenizer access; -- message trimming. +- message trimming; - provider-neutral LLM response, stream, and client contracts; - loop configuration, lifecycle contexts, observer protocol, intervention merging, and observer dispatch helpers. diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py index e58e143..ac8c7ee 100644 --- a/agent_core/loop_types.py +++ b/agent_core/loop_types.py @@ -22,6 +22,24 @@ def deadline_remaining_s(metadata: Mapping[str, Any] | None) -> float | None: Execution-context storage belongs to each host product. Passing only its metadata keeps this frozen contract module independent of either host. + + ``None`` means no deadline is stamped (no wall observer — plain swarm, + tests, HTTP API): callers keep their configured timeouts unchanged. The + result may be negative once the deadline has passed. + + Two stamp shapes are accepted, and the check is **structural, not + nominal**: + + * any object exposing ``remaining_s() -> float`` — a renewable lease view. + The concrete renewable-lease implementation is a product concern and + must stay out of this module: importing it here to run an ``isinstance`` + check would put a product-side dependency inside the frozen contract + module and break the ``agent_core`` import closure. Do not "clean this + up" into an ``isinstance`` check. + * a bare ``int``/``float`` absolute ``time.monotonic()`` instant. + + Anything else yields ``None``. ``int``/``float`` have no ``remaining_s``, + so the ordering below is unambiguous. """ deadline = (metadata or {}).get(WALL_DEADLINE_MONOTONIC_KEY) remaining_s = cast(Callable[[], Any] | None, getattr(deadline, "remaining_s", None)) @@ -565,6 +583,9 @@ async def notify_tool_result( "ToolResult", "TurnContext", "deadline_remaining_s", + "drain_background_observers", "merge_interventions", "notify_observers", + "notify_tool_call", + "notify_tool_result", ] From 63dba6f0c2a01d7f83b50955c2137936c431db00 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Mon, 31 Aug 2026 15:09:37 +0800 Subject: [PATCH 3/3] fix: harden runtime observer contracts Scope passive observer task draining per agent loop and flush cancellation hooks. Preserve explicit empty injections, ignore empty stop reasons, reject boolean deadlines, formalize optional observer protocols, and correct the streaming contract documentation.\n\nRefs #2, #3, #4, #5, #6. --- agent_core/llm.py | 5 ++- agent_core/loop_types.py | 77 ++++++++++++++++++++++++++++++++-------- tests/test_loop_types.py | 66 ++++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 17 deletions(-) diff --git a/agent_core/llm.py b/agent_core/llm.py index 6d0437e..750df2d 100644 --- a/agent_core/llm.py +++ b/agent_core/llm.py @@ -83,9 +83,8 @@ def stream( ) -> AsyncIterator[StreamDelta]: """Stream a completion as a sequence of incremental ``StreamDelta``s. - The terminal ``LLMResponse`` (with assembled content + finalised - tool_calls + usage) is accessible via :meth:`last_response` after the - stream is exhausted. + Terminal metadata is carried by late deltas; the consuming runtime is + responsible for assembling those deltas into its final response. """ ... diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py index ac8c7ee..e9a5fee 100644 --- a/agent_core/loop_types.py +++ b/agent_core/loop_types.py @@ -7,6 +7,7 @@ import time from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field +from functools import partial from typing import Any, Literal, Protocol, cast, runtime_checkable from agent_core.messages import Message @@ -48,7 +49,10 @@ def deadline_remaining_s(metadata: Mapping[str, Any] | None) -> float | None: return float(remaining_s()) except Exception: return None - if not isinstance(deadline, (int, float)): + # ``bool`` is an ``int`` subclass, but it is never a meaningful monotonic + # instant. Treat it as a bad metadata shape instead of expiring the lease at + # process time 0 or 1. + if isinstance(deadline, bool) or not isinstance(deadline, (int, float)): return None return float(deadline) - time.monotonic() @@ -309,7 +313,12 @@ class CompactionEvent: @runtime_checkable class LoopObserver(Protocol): - """Structural contract implemented by agent-loop observers.""" + """Required structural contract implemented by agent-loop observers. + + Compaction and cancellation notifications are optional so legacy observers + remain valid. Hosts can narrow those hooks with :class:`CompactionObserver` + and :class:`CancellationObserver`; :class:`BaseObserver` implements both. + """ critical: bool @@ -341,6 +350,20 @@ async def on_turn_end(self, ctx: TurnContext) -> Intervention | None: ... async def on_loop_end(self, result: AgentLoopResult) -> None: ... +@runtime_checkable +class CompactionObserver(Protocol): + """Optional observer hook for completed history compactions.""" + + async def on_compaction(self, event: CompactionEvent) -> None: ... + + +@runtime_checkable +class CancellationObserver(Protocol): + """Optional observer hook for cancellation cleanup.""" + + async def on_loop_cancelled(self) -> None: ... + + class BaseObserver: """No-op observer base; override only required hooks.""" @@ -389,8 +412,10 @@ async def on_loop_cancelled(self) -> None: """Release resources when cancellation bypasses ``on_loop_end``.""" -# Prevent GC of fire-and-forget observer tasks. -_background_tasks: set[asyncio.Task[None]] = set() +# Prevent GC of fire-and-forget observer tasks without coupling independent +# loops. Agent loops execute their dispatches from one owning asyncio task, so +# that task is the natural lifecycle boundary for the passive hooks it starts. +_background_tasks_by_owner: dict[asyncio.Task[Any], set[asyncio.Task[None]]] = {} # Log each observer-hook failure once at warning level. @@ -423,6 +448,19 @@ def _handle_observer_error( ) +def _discard_background_task( + owner: asyncio.Task[Any], + completed: asyncio.Task[None], +) -> None: + """Release a completed passive hook and its empty owner bucket.""" + tasks = _background_tasks_by_owner.get(owner) + if tasks is None: + return + tasks.discard(completed) + if not tasks: + _background_tasks_by_owner.pop(owner, None) + + async def notify_observers( observers: list[Any], method: str, @@ -431,7 +469,8 @@ async def notify_observers( ) -> list[Intervention]: """Run hooks, awaiting critical observers and isolating hook errors. - ``on_loop_end`` drains passive hooks so their side effects are visible on return. + Loop end and cancellation drain this loop's passive hooks so their side + effects are visible on return. """ interventions: list[Intervention] = [] @@ -462,34 +501,42 @@ async def _run( _handle_observer_error(observer, m, exc) task = asyncio.create_task(_run()) - _background_tasks.add(task) - task.add_done_callback(_background_tasks.discard) + owner = asyncio.current_task() + if owner is None: # pragma: no cover - create_task also needs a running loop + raise RuntimeError("observer dispatch requires an owning asyncio task") + _background_tasks_by_owner.setdefault(owner, set()).add(task) + task.add_done_callback(partial(_discard_background_task, owner)) - if method == "on_loop_end": + if method in {"on_loop_end", "on_loop_cancelled"}: await drain_background_observers() return interventions async def drain_background_observers() -> None: - """Drain outstanding passive observer tasks.""" - pending = [task for task in _background_tasks if not task.done()] + """Drain passive observer tasks owned by the current agent loop.""" + owner = asyncio.current_task() + if owner is None: + return + pending = [task for task in _background_tasks_by_owner.get(owner, ()) if not task.done()] if pending: await asyncio.gather(*pending, return_exceptions=True) def merge_interventions(interventions: list[Intervention]) -> Intervention: - """Merge messages, take the first stop reason, and OR boolean controls.""" + """Merge messages, take the first non-empty stop, and OR boolean controls.""" all_messages: list[str] = [] + inject_messages_set = False stop_reason: str | None = None skip: bool = False pop_last: bool = False continue_turn: bool = False for iv in interventions: - if iv.inject_messages: + if iv.inject_messages is not None: + inject_messages_set = True all_messages.extend(iv.inject_messages) - if stop_reason is None and iv.stop_reason is not None: + if stop_reason is None and iv.stop_reason: stop_reason = iv.stop_reason if iv.skip_tool_execution: skip = True @@ -499,7 +546,7 @@ def merge_interventions(interventions: list[Intervention]) -> Intervention: continue_turn = True return Intervention( - inject_messages=all_messages if all_messages else None, + inject_messages=all_messages if inject_messages_set else None, stop_reason=stop_reason, skip_tool_execution=skip, pop_last_message=pop_last, @@ -572,7 +619,9 @@ async def notify_tool_result( "WALL_DEADLINE_MONOTONIC_KEY", "AgentLoopResult", "BaseObserver", + "CancellationObserver", "CompactionEvent", + "CompactionObserver", "Intervention", "LLMAttemptContext", "LLMDeltaContext", diff --git a/tests/test_loop_types.py b/tests/test_loop_types.py index 178c5da..4ad1979 100644 --- a/tests/test_loop_types.py +++ b/tests/test_loop_types.py @@ -50,6 +50,11 @@ def test_no_inject_messages_stays_none_not_empty_list(): assert merge_interventions([Intervention(), Intervention()]).inject_messages is None +def test_explicit_empty_inject_messages_stays_empty_not_none(): + merged = merge_interventions([Intervention(), Intervention(inject_messages=[]), Intervention()]) + assert merged.inject_messages == [] + + def test_stop_reason_first_non_none_wins(): merged = merge_interventions( [ @@ -61,6 +66,13 @@ def test_stop_reason_first_non_none_wins(): assert merged.stop_reason == "first" +def test_empty_stop_reason_does_not_shadow_later_reason(): + merged = merge_interventions( + [Intervention(stop_reason=""), Intervention(stop_reason="budget_exhausted")] + ) + assert merged.stop_reason == "budget_exhausted" + + @pytest.mark.parametrize( "flag", ["skip_tool_execution", "pop_last_message", "continue_to_next_turn"], @@ -228,6 +240,8 @@ def remaining_s(self) -> float: key = lt.WALL_DEADLINE_MONOTONIC_KEY assert lt.deadline_remaining_s({key: BrokenLease()}) is None assert lt.deadline_remaining_s({key: object()}) is None + assert lt.deadline_remaining_s({key: True}) is None + assert lt.deadline_remaining_s({key: False}) is None assert lt.deadline_remaining_s(None) is None @@ -249,6 +263,49 @@ async def on_llm_response(self, ctx): await lt.drain_background_observers() +@pytest.mark.asyncio +async def test_loop_end_does_not_drain_another_loops_passive_hooks() -> None: + foreign_entered = asyncio.Event() + foreign_release = asyncio.Event() + end_completed = asyncio.Event() + + class Foreign(lt.BaseObserver): + async def on_turn_end(self, ctx): + foreign_entered.set() + await foreign_release.wait() + + async def foreign_loop() -> None: + await lt.notify_observers([Foreign()], "on_turn_end", None) + await foreign_entered.wait() + await foreign_release.wait() + + async def ending_loop() -> None: + await lt.notify_observers([], "on_loop_end", None) + end_completed.set() + + foreign_task = asyncio.create_task(foreign_loop()) + try: + await foreign_entered.wait() + await asyncio.wait_for(ending_loop(), timeout=1) + assert end_completed.is_set() + finally: + foreign_release.set() + await foreign_task + + +@pytest.mark.asyncio +async def test_loop_cancelled_drains_its_passive_cleanup() -> None: + cleanup_finished = asyncio.Event() + + class Passive(lt.BaseObserver): + async def on_loop_cancelled(self): + await asyncio.sleep(0) + cleanup_finished.set() + + await lt.notify_observers([Passive()], "on_loop_cancelled") + assert cleanup_finished.is_set() + + def test_legacy_observer_satisfies_runtime_protocol() -> None: class Legacy: critical = True @@ -263,3 +320,12 @@ async def on_turn_end(self, ctx): ... async def on_loop_end(self, result): ... assert isinstance(Legacy(), lt.LoopObserver) + assert not isinstance(Legacy(), lt.CompactionObserver) + assert not isinstance(Legacy(), lt.CancellationObserver) + + +def test_base_observer_satisfies_optional_hook_protocols() -> None: + observer = lt.BaseObserver() + assert isinstance(observer, lt.LoopObserver) + assert isinstance(observer, lt.CompactionObserver) + assert isinstance(observer, lt.CancellationObserver)