diff --git a/README.md b/README.md index 0f4e17c..271d683 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,10 @@ 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. 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..750df2d --- /dev/null +++ b/agent_core/llm.py @@ -0,0 +1,92 @@ +"""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. + + Terminal metadata is carried by late deltas; the consuming runtime is + responsible for assembling those deltas into its final response. + """ + ... + + +__all__ = ["LLMClient", "LLMResponse", "StreamDelta"] diff --git a/agent_core/loop_types.py b/agent_core/loop_types.py new file mode 100644 index 0000000..e9a5fee --- /dev/null +++ b/agent_core/loop_types.py @@ -0,0 +1,640 @@ +"""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 functools import partial +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. + + ``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)) + if callable(remaining_s): + try: + return float(remaining_s()) + except Exception: + return None + # ``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() + + +@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): + """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 + + 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: ... + + +@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.""" + + 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 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. +_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, + ) + + +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, + *args: Any, + **kwargs: Any, +) -> list[Intervention]: + """Run hooks, awaiting critical observers and isolating hook errors. + + Loop end and cancellation drain this loop's 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()) + 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 in {"on_loop_end", "on_loop_cancelled"}: + await drain_background_observers() + + return interventions + + +async def drain_background_observers() -> None: + """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 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 is not None: + inject_messages_set = True + all_messages.extend(iv.inject_messages) + if stop_reason is None and iv.stop_reason: + 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 inject_messages_set 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", + "CancellationObserver", + "CompactionEvent", + "CompactionObserver", + "Intervention", + "LLMAttemptContext", + "LLMDeltaContext", + "LoopConfig", + "LoopObserver", + "LoopPolicy", + "ToolCallIntervention", + "ToolResult", + "TurnContext", + "deadline_remaining_s", + "drain_background_observers", + "merge_interventions", + "notify_observers", + "notify_tool_call", + "notify_tool_result", +] 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..4ad1979 --- /dev/null +++ b/tests/test_loop_types.py @@ -0,0 +1,331 @@ +"""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_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( + [ + Intervention(), + Intervention(stop_reason="first"), + Intervention(stop_reason="second"), + ] + ) + 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"], +) +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({key: True}) is None + assert lt.deadline_remaining_s({key: False}) 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() + + +@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 + + 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) + 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)