diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bc8f00..51c4432 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,32 @@ All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 0.4.0 - 2026-07-02 + +### Added + +- `AgentKit`, a FastAPI-flavored hub over the registry: keyword-native + `await kit.run("claude", goal=..., permissions="strict", ...)` assembling + the frozen `AgentTask` internally (a prebuilt `task=` still passes through), + short aliases for the built-in kinds via the documented `KIND_ALIASES` + mapping, per-kind runtime caching closed by `aclose()`/`async with`, + `@kit.on(...)` sync/async event handlers that tee alongside a task's own + sink and can never break a run, and `@kit.runtime(...)` decorator + registration for third-party kinds. +- Typed structured output: `kit.run(..., output_type=SomeType)` derives the + wire schema from a dataclass/`TypedDict` (dependency-free, bounded subset, + fail-closed via the new `OutputTypeError`) or from a Pydantic-style model's + own `model_json_schema`, and validates `parsed_output` back into the type. + The result is `ParsedResult[T]` — a runtime-identical `AgentResult` + subclass with a typed `parsed` accessor; `AgentResult` itself stays + non-generic. Non-conforming payloads yield `finish_reason="failed"`, the + adapters' own structured-output convention. +- `PermissionProfile` accepts string literals for `mode`/`filesystem` + ("strict", "read-only", ...) and coerces them to real enum members at + construction; unknown values raise `ValueError` listing the vocabulary. + Previously a bare string silently matched none of the adapters' identity + checks and ran at the default posture. + ## 0.3.0 - 2026-07-02 ### Added diff --git a/README.md b/README.md index e173ff4..0ef3dd4 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,41 @@ are added through optional extras. ## Real Providers +`AgentKit` is the keyword-native hub: it registers the built-in runtimes, +caches them per kind, and turns Python types into structured output. + +```python +import asyncio +from dataclasses import dataclass + +from agent_runtime_kit import AgentKit + + +@dataclass +class RepoSummary: + name: str + languages: list[str] + + +async def main() -> None: + async with AgentKit() as kit: + diagnostic = kit.availability_for("claude") + if not diagnostic.available: + raise RuntimeError(diagnostic.message) + result = await kit.run( + "claude", + goal="Summarize this repository", + permissions="strict", + output_type=RepoSummary, + ) + print(result.parsed.languages if result.parsed else result.error) + + +asyncio.run(main()) +``` + +Adapters also work standalone when you need vendor-specific configuration: + ```python import asyncio diff --git a/docs/api-stability.md b/docs/api-stability.md index 8bc85aa..7a7d604 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -37,6 +37,14 @@ import the names from the top-level package instead. a task field raises `UnsupportedTaskInputError`; the one exception is vendor-option drift, which is recorded in `AgentResult.metadata["dropped_options"]` instead. +- **`AgentKit` is sugar, not a second API.** The hub assembles the same frozen + `AgentTask` and returns the same `AgentResult` the runtimes produce + (`ParsedResult` is a runtime-identical subclass adding only the typed + `parsed` accessor). Kind aliases are limited to the documented + `KIND_ALIASES` mapping; exact kind strings always work. `output_type=` + supports a bounded, documented subset of the typing system and raises + `OutputTypeError` on anything outside it (Pydantic-style models are used + through their own `model_json_schema`/`model_validate` when present). ## Vendor SDK version policy diff --git a/docs/quickstart.md b/docs/quickstart.md index f389e4b..5ca8f24 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -25,23 +25,98 @@ Claude-only application from installing Codex or Antigravity packages, keep the core usable when one vendor package is unavailable on a platform, and make missing-provider errors actionable. -Run a task: +## Run a task + +`AgentKit` is the hub: it registers the built-in runtimes, resolves them +lazily, and gives `run()` a keyword-native surface. ```python import asyncio -from agent_runtime_kit import AgentTask -from agent_runtime_kit.adapters import ClaudeAgentRuntime +from agent_runtime_kit import AgentKit async def main() -> None: - runtime = ClaudeAgentRuntime(default_model="claude-sonnet-4-6") - result = await runtime.run(AgentTask(goal="Summarize this repository")) - print(result.output) + async with AgentKit() as kit: + result = await kit.run( + "claude", # short alias; "claude-agent-sdk" works too + goal="Summarize this repository", + permissions="strict", + ) + print(result.output) asyncio.run(main()) ``` -Use `runtime.availability()` before dispatching work in applications that need -clear setup diagnostics. +## Typed structured output + +Pass a type instead of hand-writing JSON schema. Dataclasses and `TypedDict`s +work out of the box (no dependency); Pydantic v2 models are used through their +own `model_json_schema`/`model_validate` when you have Pydantic installed. + +```python +from dataclasses import dataclass + +from agent_runtime_kit import AgentKit + + +@dataclass +class RepoSummary: + name: str + languages: list[str] + risky_areas: list[str] + + +async def main() -> None: + async with AgentKit() as kit: + result = await kit.run( + "claude", + goal="Summarize this repository", + output_type=RepoSummary, + ) + if result.parsed is not None: + print(result.parsed.languages) # typed: list[str] + else: + print(result.error) +``` + +A payload that does not conform yields `finish_reason="failed"` with the +mismatch in `result.error` — the same convention the adapters use for +unsatisfied structured output. Unsupported annotations (sets, plain unions, +...) raise `OutputTypeError` at call time rather than sending a half-true +schema. + +## Events and custom runtimes + +```python +kit = AgentKit() + + +@kit.on("agent.tool.completed") +def log_tool(event) -> None: # sync or async; exceptions never break a run + print(event["summary"]) + + +@kit.runtime("x-myorg-agent") +def my_runtime(**kwargs): # must stay zero-arg constructible + return MyRuntime(**kwargs) +``` + +## Lower-level use + +The hub is sugar over the same objects you can use directly — construct an +adapter yourself when you need vendor-specific configuration, and pass either +the instance or a prebuilt `AgentTask` to `kit.run` (or call `runtime.run(task)` +without the hub at all): + +```python +from agent_runtime_kit import AgentTask +from agent_runtime_kit.adapters import ClaudeAgentRuntime + +runtime = ClaudeAgentRuntime(default_model="claude-sonnet-4-6", reuse_process=True) +result = await runtime.run(AgentTask(goal="Summarize this repository")) +``` + +Use `kit.availability()` (or `runtime.availability()`) before dispatching work +in applications that need clear setup diagnostics. diff --git a/pyproject.toml b/pyproject.toml index 15ecc58..9d36756 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "agent-runtime-kit" -version = "0.3.0" +version = "0.4.0" description = "One typed runtime API for Claude, Codex, and Antigravity agent SDKs." readme = "README.md" requires-python = ">=3.10" diff --git a/src/agent_runtime_kit/__init__.py b/src/agent_runtime_kit/__init__.py index fe1b848..3a98d79 100644 --- a/src/agent_runtime_kit/__init__.py +++ b/src/agent_runtime_kit/__init__.py @@ -3,9 +3,11 @@ from agent_runtime_kit._errors import ( AgentRuntimeError, AgentRuntimeUnavailableError, + OutputTypeError, RuntimeNotRegisteredError, UnsupportedTaskInputError, ) +from agent_runtime_kit._kit import KIND_ALIASES, AgentKit from agent_runtime_kit._runtime import FakeAgentRuntime from agent_runtime_kit._types import ( AgentCapabilities, @@ -19,6 +21,7 @@ FilesystemAccess, FinishReason, McpServerConfig, + ParsedResult, PermissionMode, PermissionProfile, RuntimeAvailability, @@ -41,6 +44,7 @@ __all__ = [ "AgentCapabilities", + "AgentKit", "AgentResult", "AgentRuntime", "AgentRuntimeError", @@ -53,7 +57,10 @@ "FakeAgentRuntime", "FilesystemAccess", "FinishReason", + "KIND_ALIASES", "McpServerConfig", + "OutputTypeError", + "ParsedResult", "PermissionMode", "PermissionProfile", "RuntimeAvailability", diff --git a/src/agent_runtime_kit/_errors.py b/src/agent_runtime_kit/_errors.py index 51466ac..cd2b1a3 100644 --- a/src/agent_runtime_kit/_errors.py +++ b/src/agent_runtime_kit/_errors.py @@ -26,6 +26,18 @@ def __init__(self, kind: AgentRuntimeKind | str, field: str, message: str) -> No super().__init__(f"{runtime_kind_value(self.kind)} cannot honor {field}: {message}") +class OutputTypeError(AgentRuntimeError, TypeError): + """A structured ``output_type`` cannot be honored. + + Raised when a Python type cannot be converted to a JSON schema (unsupported + annotation — the stdlib bridge deliberately supports a bounded subset and + fails closed on everything else), or when a returned payload does not + conform to the requested type. Schema-generation failures surface at call + time; payload mismatches are converted by ``AgentKit`` into a failed + ``AgentResult`` instead of raising. + """ + + class RuntimeNotRegisteredError(AgentRuntimeError, LookupError): """No runtime factory is registered for the requested runtime kind.""" diff --git a/src/agent_runtime_kit/_kit.py b/src/agent_runtime_kit/_kit.py new file mode 100644 index 0000000..8be1da8 --- /dev/null +++ b/src/agent_runtime_kit/_kit.py @@ -0,0 +1,426 @@ +"""FastAPI-style hub over the runtime registry.""" + +from __future__ import annotations + +import asyncio +import inspect +from collections.abc import Callable, Mapping, Sequence +from dataclasses import fields as dataclass_fields +from pathlib import Path +from typing import Any, TypeVar, cast, overload + +from agent_runtime_kit._errors import OutputTypeError +from agent_runtime_kit._schema import json_schema_for, parse_as +from agent_runtime_kit._types import ( + AgentCapabilities, + AgentResult, + AgentRuntime, + AgentRuntimeKind, + AgentTask, + EventSink, + FilesystemAccess, + FinishReason, + McpServerConfig, + ParsedResult, + PermissionMode, + PermissionProfile, + RuntimeAvailability, + SessionResumeState, +) +from agent_runtime_kit.registry import RuntimeFactory, RuntimeRegistry, create_default_registry + +_T = TypeVar("_T") +# An event handler receives one normalized event dict; sync or async. +_EventHandler = Callable[[Mapping[str, Any]], Any] +_HandlerT = TypeVar("_HandlerT", bound=_EventHandler) +_FactoryT = TypeVar("_FactoryT", bound=RuntimeFactory) + +# Short spellings for the built-in kinds, resolved only by AgentKit (the +# registry itself stays alias-free; the full kind strings always work). +KIND_ALIASES: dict[str, AgentRuntimeKind] = { + "fake": AgentRuntimeKind.FAKE, + "claude": AgentRuntimeKind.CLAUDE_AGENT_SDK, + "codex": AgentRuntimeKind.CODEX_AGENT_SDK, + "antigravity": AgentRuntimeKind.ANTIGRAVITY_AGENT_SDK, +} + + +class AgentKit: + """One object to hold: registry, runtimes, and a kwargs-native ``run``. + + ``AgentKit()`` builds a registry with the fake runtime and the vendor + adapters registered (adapters resolve their SDKs lazily, so this works + without any extra installed — ``availability_for`` reports what is + missing). Pass ``registry=`` to bring your own; the other flags then do + not apply. + + Runtimes resolved through the hub are constructed zero-arg, cached per + kind, and closed by ``aclose()`` / ``async with``. A custom-configured + adapter instance can be passed directly to ``run`` instead of a kind. + """ + + def __init__( + self, + *, + registry: RuntimeRegistry | None = None, + include_fake: bool = True, + register_default_adapters: bool = True, + ) -> None: + if registry is None: + registry = create_default_registry(include_fake=include_fake) + if register_default_adapters: + # Imported here, not at module scope: the adapters import the + # top-level package, which imports this module. + from agent_runtime_kit.adapters import register_adapters + + register_adapters(registry) + self._registry = registry + self._runtimes: dict[AgentRuntimeKind | str, AgentRuntime] = {} + self._handlers: list[tuple[str, _EventHandler]] = [] + self._cache_lock = asyncio.Lock() + + @property + def registry(self) -> RuntimeRegistry: + return self._registry + + def kinds(self) -> tuple[AgentRuntimeKind | str, ...]: + return self._registry.kinds() + + def availability(self) -> tuple[RuntimeAvailability, ...]: + """Availability diagnostics for every registered kind.""" + + return tuple(self._registry.availability_for(kind) for kind in self._registry.kinds()) + + def availability_for(self, kind: AgentRuntimeKind | str) -> RuntimeAvailability: + return self._registry.availability_for(self._normalize_kind(kind)) + + def capabilities_for(self, kind: AgentRuntimeKind | str) -> AgentCapabilities: + return self._registry.capabilities_for(self._normalize_kind(kind)) + + def on(self, event: str = "*") -> Callable[[_HandlerT], _HandlerT]: + """Register an event handler for tasks assembled by this hub. + + ``event`` is an exact normalized event name (``"agent.tool.completed"``, + ...) or ``"*"`` for everything. Handlers may be sync or async and their + exceptions are swallowed — the ``safe_emit`` contract: observability + must never break a run. Registration applies to runs started after it. + """ + + def register(handler: _HandlerT) -> _HandlerT: + self._handlers.append((event, handler)) + return handler + + return register + + def runtime( + self, kind: AgentRuntimeKind | str, *, replace: bool = False + ) -> Callable[[_FactoryT], _FactoryT]: + """Register a runtime factory under ``kind`` (decorator form). + + The factory must remain constructible with zero arguments: + ``capabilities_for``/``availability_for`` build it that way. + """ + + def register(factory: _FactoryT) -> _FactoryT: + self._registry.register(kind, factory, replace=replace) + return factory + + return register + + @overload + async def run( + self, + runtime: AgentRuntimeKind | str | AgentRuntime, + *, + output_type: type[_T], + goal: str | None = ..., + task: AgentTask | None = ..., + system: str | None = ..., + model: str | None = ..., + reasoning_effort: str | None = ..., + working_directory: Path | str | None = ..., + permissions: PermissionProfile | PermissionMode | str | None = ..., + filesystem: FilesystemAccess | str | None = ..., + allowed_tools: Sequence[str] = ..., + disallowed_tools: Sequence[str] = ..., + output_schema: Mapping[str, Any] | None = ..., + event_sink: EventSink | None = ..., + mcp_servers: Sequence[McpServerConfig] = ..., + session_id: str | None = ..., + resume_from: SessionResumeState | None = ..., + budget_usd: float | None = ..., + sdk_executions: int = ..., + task_id: str | None = ..., + metadata: Mapping[str, Any] | None = ..., + ) -> ParsedResult[_T]: ... + + @overload + async def run( + self, + runtime: AgentRuntimeKind | str | AgentRuntime, + *, + output_type: None = None, + goal: str | None = ..., + task: AgentTask | None = ..., + system: str | None = ..., + model: str | None = ..., + reasoning_effort: str | None = ..., + working_directory: Path | str | None = ..., + permissions: PermissionProfile | PermissionMode | str | None = ..., + filesystem: FilesystemAccess | str | None = ..., + allowed_tools: Sequence[str] = ..., + disallowed_tools: Sequence[str] = ..., + output_schema: Mapping[str, Any] | None = ..., + event_sink: EventSink | None = ..., + mcp_servers: Sequence[McpServerConfig] = ..., + session_id: str | None = ..., + resume_from: SessionResumeState | None = ..., + budget_usd: float | None = ..., + sdk_executions: int = ..., + task_id: str | None = ..., + metadata: Mapping[str, Any] | None = ..., + ) -> AgentResult: ... + + async def run( + self, + runtime: AgentRuntimeKind | str | AgentRuntime, + *, + output_type: type[Any] | None = None, + goal: str | None = None, + task: AgentTask | None = None, + system: str | None = None, + model: str | None = None, + reasoning_effort: str | None = None, + working_directory: Path | str | None = None, + permissions: PermissionProfile | PermissionMode | str | None = None, + filesystem: FilesystemAccess | str | None = None, + allowed_tools: Sequence[str] = (), + disallowed_tools: Sequence[str] = (), + output_schema: Mapping[str, Any] | None = None, + event_sink: EventSink | None = None, + mcp_servers: Sequence[McpServerConfig] = (), + session_id: str | None = None, + resume_from: SessionResumeState | None = None, + budget_usd: float | None = None, + sdk_executions: int = 1, + task_id: str | None = None, + metadata: Mapping[str, Any] | None = None, + ) -> AgentResult: + """Run one task, assembling the ``AgentTask`` from keyword arguments. + + Exactly one of ``goal`` (plus the other field kwargs) or ``task`` must + be provided; ``output_type``, ``output_schema``, and ``event_sink`` + may accompany a prebuilt ``task`` and override its fields. + ``output_type`` derives ``output_schema`` from a Python type and + validates the result's ``parsed_output`` into it; a payload that does + not conform yields a ``finish_reason="failed"`` result (the same + convention the adapters use for unsatisfied structured output), never + an exception. + """ + + if output_type is not None and output_schema is not None: + raise ValueError("output_type and output_schema are mutually exclusive") + schema = json_schema_for(output_type) if output_type is not None else output_schema + + if task is not None: + built = self._merge_into_task( + task, + goal=goal, + system=system, + model=model, + reasoning_effort=reasoning_effort, + working_directory=working_directory, + permissions=permissions, + filesystem=filesystem, + allowed_tools=allowed_tools, + disallowed_tools=disallowed_tools, + event_sink=event_sink, + mcp_servers=mcp_servers, + session_id=session_id, + resume_from=resume_from, + budget_usd=budget_usd, + task_id=task_id, + metadata=metadata, + schema=schema, + ) + else: + if goal is None: + raise ValueError("run() needs either goal=... or task=...") + task_kwargs: dict[str, Any] = { + "goal": goal, + "system": system, + "model": model, + "reasoning_effort": reasoning_effort, + "working_directory": _as_path(working_directory), + "mcp_servers": tuple(mcp_servers), + "permissions": _normalize_permissions( + permissions, filesystem, allowed_tools, disallowed_tools + ), + "event_sink": self._compose_sink(event_sink), + "sdk_executions": sdk_executions, + "budget_usd": budget_usd, + "session_id": session_id, + "resume_from": resume_from, + "output_schema": schema, + "metadata": dict(metadata) if metadata is not None else {}, + } + if task_id is not None: + task_kwargs["task_id"] = task_id + built = AgentTask(**task_kwargs) + + agent = runtime if not isinstance(runtime, str) else await self._runtime_for(runtime) + result = await agent.run(built) + if output_type is None: + return result + return _parse_result(output_type, result) + + async def aclose(self) -> None: + """Close every runtime this hub constructed and cached.""" + + async with self._cache_lock: + runtimes = list(self._runtimes.values()) + self._runtimes.clear() + for agent in runtimes: + await agent.aclose() + + async def __aenter__(self) -> AgentKit: + return self + + async def __aexit__(self, _exc_type: object, _exc: object, _tb: object) -> None: + await self.aclose() + + def _normalize_kind(self, kind: AgentRuntimeKind | str) -> AgentRuntimeKind | str: + if isinstance(kind, str) and not isinstance(kind, AgentRuntimeKind): + alias = KIND_ALIASES.get(kind) + if alias is not None: + return alias + return AgentRuntimeKind.coerce(kind) + + async def _runtime_for(self, kind: AgentRuntimeKind | str) -> AgentRuntime: + normalized = self._normalize_kind(kind) + async with self._cache_lock: + agent = self._runtimes.get(normalized) + if agent is None: + agent = self._registry.resolve(normalized) + self._runtimes[normalized] = agent + return agent + + def _merge_into_task( + self, + task: AgentTask, + *, + schema: Mapping[str, Any] | None, + event_sink: EventSink | None, + **field_kwargs: Any, + ) -> AgentTask: + conflicting = sorted( + name + for name, value in field_kwargs.items() + if value not in (None, (), {}, []) # defaults mean "not provided" + ) + if conflicting: + raise ValueError( + "task= is mutually exclusive with per-field kwargs; got both task and " + + ", ".join(conflicting) + ) + replacements: dict[str, Any] = {} + if schema is not None: + replacements["output_schema"] = schema + effective_sink = event_sink if event_sink is not None else task.event_sink + composed_sink = self._compose_sink(effective_sink) + if composed_sink is not task.event_sink: + replacements["event_sink"] = composed_sink + if not replacements: + return task + values = {f.name: getattr(task, f.name) for f in dataclass_fields(task)} + values.update(replacements) + return AgentTask(**values) + + def _compose_sink(self, downstream: EventSink | None) -> EventSink | None: + if not self._handlers: + return downstream + return _TeeSink(tuple(self._handlers), downstream) + + +class _TeeSink: + """Fan events out to hub handlers, then to the task's own sink. + + Handler and downstream failures are swallowed independently (the + ``safe_emit`` contract): observability must never break a run, and one bad + handler must not starve the others or the downstream sink. + """ + + def __init__( + self, + handlers: tuple[tuple[str, _EventHandler], ...], + downstream: EventSink | None, + ) -> None: + self._handlers = handlers + self._downstream = downstream + + async def emit(self, event: Mapping[str, Any]) -> None: + name = str(event.get("name", "")) + for pattern, handler in self._handlers: + if pattern != "*" and pattern != name: + continue + try: + outcome = handler(event) + if inspect.isawaitable(outcome): + await outcome + except Exception: + continue + if self._downstream is not None: + try: + await self._downstream.emit(event) + except Exception: + return + + +def _as_path(value: Path | str | None) -> Path | None: + if value is None or isinstance(value, Path): + return value + return Path(value) + + +def _normalize_permissions( + permissions: PermissionProfile | PermissionMode | str | None, + filesystem: FilesystemAccess | str | None, + allowed_tools: Sequence[str], + disallowed_tools: Sequence[str], +) -> PermissionProfile: + if isinstance(permissions, PermissionProfile): + if filesystem is not None or allowed_tools or disallowed_tools: + raise ValueError( + "pass filesystem/allowed_tools/disallowed_tools inside the " + "PermissionProfile, not alongside one" + ) + return permissions + # Strings coerce to enum members in PermissionProfile.__post_init__. + profile_kwargs: dict[str, Any] = { + "allowed_tools": tuple(allowed_tools), + "disallowed_tools": tuple(disallowed_tools), + } + if permissions is not None: + profile_kwargs["mode"] = permissions + if filesystem is not None: + profile_kwargs["filesystem"] = filesystem + return PermissionProfile(**profile_kwargs) + + +def _parse_result(output_type: type[_T], result: AgentResult) -> ParsedResult[_T]: + values = {f.name: getattr(result, f.name) for f in dataclass_fields(result)} + if result.error is not None or result.parsed_output is None: + # Adapter-reported failures (including unsatisfied structured output) + # pass through untyped; .parsed stays None. + return cast("ParsedResult[_T]", ParsedResult(**values)) + try: + instance = parse_as(output_type, result.parsed_output) + except OutputTypeError as exc: + values.update( + finish_reason=FinishReason.FAILED.value, + error=f"structured output does not conform to {output_type.__name__}: {exc}", + parsed_output=None, + ) + return cast("ParsedResult[_T]", ParsedResult(**values)) + values["parsed_output"] = instance + return cast("ParsedResult[_T]", ParsedResult(**values)) diff --git a/src/agent_runtime_kit/_schema.py b/src/agent_runtime_kit/_schema.py new file mode 100644 index 0000000..7a43c4f --- /dev/null +++ b/src/agent_runtime_kit/_schema.py @@ -0,0 +1,272 @@ +"""Dependency-free bridge between Python types and JSON schema. + +``json_schema_for`` turns an ``output_type`` into the JSON schema sent to the +vendor SDK; ``parse_as`` turns the returned payload back into an instance of +that type. Both support a deliberately bounded subset of the typing system — +scalars, ``X | None``, ``list[X]``, ``dict[str, X]``, ``Literal``, ``Enum``, +dataclasses, and ``TypedDict`` — and fail closed with ``OutputTypeError`` on +anything else rather than emitting a half-true schema. + +Types that expose ``model_json_schema()`` and ``model_validate()`` (Pydantic +v2 models, or anything shaped like them) are delegated to those methods via a +structural check, so users who already have Pydantic get its full type +coverage without this package importing or depending on it. +""" + +from __future__ import annotations + +import dataclasses +import enum +import types +import typing +from collections.abc import Mapping +from typing import Any, Literal, Union, get_args, get_origin, get_type_hints + +from agent_runtime_kit._errors import OutputTypeError + +# Deep enough for any sane payload model; recursive dataclasses hit this bound +# and fail closed instead of overflowing. +_MAX_DEPTH = 16 + +_SCALARS: dict[type, str] = {bool: "boolean", int: "integer", float: "number", str: "string"} + + +def supports_model_protocol(tp: Any) -> bool: + """True when ``tp`` is Pydantic-shaped (model_json_schema + model_validate).""" + + return callable(getattr(tp, "model_json_schema", None)) and callable( + getattr(tp, "model_validate", None) + ) + + +def json_schema_for(tp: Any) -> dict[str, Any]: + """Return the JSON schema for ``tp``, or raise ``OutputTypeError``.""" + + if supports_model_protocol(tp): + schema = tp.model_json_schema() + if not isinstance(schema, Mapping): + raise OutputTypeError( + f"{_name(tp)}.model_json_schema() returned {type(schema).__name__}, " + "expected a mapping" + ) + return dict(schema) + return _schema(tp, depth=0) + + +def parse_as(tp: Any, value: Any) -> Any: + """Validate ``value`` against ``tp`` and return the typed instance. + + Strict by design: extra object keys, missing required fields, and + cross-type coercions (``"42"`` for ``int``, ``True`` for ``int``) all raise + ``OutputTypeError``. ``AgentKit`` narrows the return to the requested type. + """ + + if supports_model_protocol(tp): + try: + return tp.model_validate(value) + except Exception as exc: + raise OutputTypeError( + f"{_name(tp)}.model_validate() rejected the payload: {exc}" + ) from exc + return _parse(tp, value, path="$", depth=0) + + +def _schema(tp: Any, *, depth: int) -> dict[str, Any]: + if depth > _MAX_DEPTH: + raise OutputTypeError( + f"output_type nesting exceeds {_MAX_DEPTH} levels (recursive types are unsupported)" + ) + if tp is None or tp is type(None): + return {"type": "null"} + optional = _optional_inner(tp) + if optional is not None: + return {"anyOf": [_schema(optional, depth=depth + 1), {"type": "null"}]} + origin = get_origin(tp) + if origin is list: + (item,) = get_args(tp) or (None,) + if item is None: + raise OutputTypeError("bare list is unsupported; use list[X]") + return {"type": "array", "items": _schema(item, depth=depth + 1)} + if origin is dict: + args = get_args(tp) + if len(args) != 2 or args[0] is not str: + raise OutputTypeError("only dict[str, X] mappings are supported") + return {"type": "object", "additionalProperties": _schema(args[1], depth=depth + 1)} + if origin is Literal: + values = get_args(tp) + if not all(isinstance(item, (str, int, bool)) for item in values): + raise OutputTypeError("Literal values must be str, int, or bool") + return {"enum": list(values)} + if isinstance(tp, type): + if tp in _SCALARS: + return {"type": _SCALARS[tp]} + if issubclass(tp, enum.Enum): + values = tuple(member.value for member in tp) + if all(isinstance(item, str) for item in values): + return {"type": "string", "enum": list(values)} + raise OutputTypeError(f"enum {_name(tp)} must have all-string values") + if dataclasses.is_dataclass(tp): + return _object_schema( + _dataclass_hints(tp), + required=[ + schema_field.name + for schema_field in dataclasses.fields(tp) + if schema_field.default is dataclasses.MISSING + and schema_field.default_factory is dataclasses.MISSING + ], + depth=depth, + ) + if typing.is_typeddict(tp): + # getattr: mypy narrows tp to `type` here, which has no + # __required_keys__; every TypedDict class carries it at runtime. + required_keys: frozenset[str] = getattr(tp, "__required_keys__", frozenset()) + return _object_schema( + get_type_hints(tp), + required=sorted(str(key) for key in required_keys), + depth=depth, + ) + raise OutputTypeError( + f"unsupported annotation for output_type: {tp!r}; supported: str/int/float/bool, " + "X | None, list[X], dict[str, X], Literal, str-valued Enum, dataclasses, TypedDict, " + "or a Pydantic-style model" + ) + + +def _object_schema( + hints: Mapping[str, Any], *, required: list[str], depth: int +) -> dict[str, Any]: + schema: dict[str, Any] = { + "type": "object", + "properties": { + name: _schema(annotation, depth=depth + 1) for name, annotation in hints.items() + }, + "additionalProperties": False, + } + if required: + schema["required"] = required + return schema + + +def _parse(tp: Any, value: Any, *, path: str, depth: int) -> Any: + if depth > _MAX_DEPTH: + raise OutputTypeError(f"{path}: payload nesting exceeds {_MAX_DEPTH} levels") + if tp is None or tp is type(None): + if value is not None: + raise OutputTypeError(f"{path}: expected null, got {type(value).__name__}") + return None + optional = _optional_inner(tp) + if optional is not None: + if value is None: + return None + return _parse(optional, value, path=path, depth=depth + 1) + origin = get_origin(tp) + if origin is list: + (item,) = get_args(tp) + if not isinstance(value, list): + raise OutputTypeError(f"{path}: expected array, got {type(value).__name__}") + return [ + _parse(item, entry, path=f"{path}[{index}]", depth=depth + 1) + for index, entry in enumerate(value) + ] + if origin is dict: + _, item = get_args(tp) + if not isinstance(value, Mapping): + raise OutputTypeError(f"{path}: expected object, got {type(value).__name__}") + return { + str(key): _parse(item, entry, path=f"{path}.{key}", depth=depth + 1) + for key, entry in value.items() + } + if origin is Literal: + if value not in get_args(tp): + raise OutputTypeError(f"{path}: {value!r} is not one of {list(get_args(tp))}") + return value + if isinstance(tp, type): + if tp in _SCALARS: + return _parse_scalar(tp, value, path=path) + if issubclass(tp, enum.Enum): + try: + return tp(value) + except ValueError: + valid = ", ".join(repr(member.value) for member in tp) + raise OutputTypeError(f"{path}: {value!r} is not one of {valid}") from None + if dataclasses.is_dataclass(tp): + return _parse_dataclass(tp, value, path=path, depth=depth) + if typing.is_typeddict(tp): + return _parse_typeddict(tp, value, path=path, depth=depth) + raise OutputTypeError(f"{path}: unsupported annotation {tp!r}") + + +def _parse_scalar(tp: type, value: Any, *, path: str) -> Any: + # bool is an int subclass: check it first and never let it satisfy int/float. + if tp is bool: + if isinstance(value, bool): + return value + elif tp is int: + if isinstance(value, int) and not isinstance(value, bool): + return value + elif tp is float: + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + elif tp is str: + if isinstance(value, str): + return value + raise OutputTypeError(f"{path}: expected {tp.__name__}, got {type(value).__name__}") + + +def _parse_dataclass(tp: type, value: Any, *, path: str, depth: int) -> Any: + if not isinstance(value, Mapping): + raise OutputTypeError(f"{path}: expected object, got {type(value).__name__}") + hints = _dataclass_hints(tp) + extra = sorted(set(value) - set(hints)) + if extra: + raise OutputTypeError(f"{path}: unexpected keys {extra} for {_name(tp)}") + kwargs: dict[str, Any] = {} + for schema_field in dataclasses.fields(tp): + name = schema_field.name + if name in value: + kwargs[name] = _parse(hints[name], value[name], path=f"{path}.{name}", depth=depth + 1) + elif ( + schema_field.default is dataclasses.MISSING + and schema_field.default_factory is dataclasses.MISSING + ): + raise OutputTypeError(f"{path}: missing required key {name!r} for {_name(tp)}") + return tp(**kwargs) + + +def _parse_typeddict(tp: Any, value: Any, *, path: str, depth: int) -> Any: + if not isinstance(value, Mapping): + raise OutputTypeError(f"{path}: expected object, got {type(value).__name__}") + hints = get_type_hints(tp) + extra = sorted(set(value) - set(hints)) + if extra: + raise OutputTypeError(f"{path}: unexpected keys {extra} for {_name(tp)}") + missing = sorted(key for key in tp.__required_keys__ if key not in value) + if missing: + raise OutputTypeError(f"{path}: missing required keys {missing} for {_name(tp)}") + return { + key: _parse(hints[key], entry, path=f"{path}.{key}", depth=depth + 1) + for key, entry in value.items() + } + + +def _optional_inner(tp: Any) -> Any | None: + """For ``X | None`` return ``X``; otherwise ``None``. Wider unions fail closed.""" + + origin = get_origin(tp) + if origin is not Union and origin is not types.UnionType: + return None + args = [arg for arg in get_args(tp) if arg is not type(None)] + if len(args) == len(get_args(tp)): + raise OutputTypeError("plain unions are unsupported; only X | None is") + if len(args) != 1: + raise OutputTypeError("only two-member optionals (X | None) are supported") + return args[0] + + +def _dataclass_hints(tp: type) -> dict[str, Any]: + # Resolves string annotations (`from __future__ import annotations`) to types. + return get_type_hints(tp) + + +def _name(tp: Any) -> str: + return str(getattr(tp, "__name__", tp)) diff --git a/src/agent_runtime_kit/_types.py b/src/agent_runtime_kit/_types.py index 1eea65b..b2ed181 100644 --- a/src/agent_runtime_kit/_types.py +++ b/src/agent_runtime_kit/_types.py @@ -6,9 +6,11 @@ from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, NoReturn, Protocol, runtime_checkable +from typing import Any, Generic, NoReturn, Protocol, TypeVar, cast, runtime_checkable from uuid import uuid4 +_EnumT = TypeVar("_EnumT", bound=Enum) + class _FrozenMapping(dict[str, Any]): """A ``dict`` that rejects in-place mutation. @@ -63,6 +65,22 @@ def _freeze_mapping(value: Mapping[str, Any]) -> Mapping[str, Any]: return _FrozenMapping(value) +def _coerce_enum(enum_cls: type[_EnumT], value: Any, field_name: str) -> _EnumT: + """Coerce a raw value (typically a string literal) into an enum member. + + Boundary coercion must yield the actual member, never an equal bare string: + the adapters compare these fields with identity (``mode is + PermissionMode.STRICT``), so an uncoerced string would silently match + nothing and run at the default posture. + """ + + try: + return enum_cls(value) + except ValueError: + valid = ", ".join(sorted(str(member.value) for member in enum_cls)) + raise ValueError(f"invalid {field_name} {value!r}; valid values: {valid}") from None + + class AgentRuntimeKind(str, Enum): """Supported runtime families.""" @@ -244,7 +262,13 @@ def __post_init__(self) -> None: @dataclass(frozen=True) class PermissionProfile: - """Portable permission request mapped by each adapter.""" + """Portable permission request mapped by each adapter. + + ``mode`` and ``filesystem`` also accept their string values ("strict", + "read-only", ...) and are coerced to enum members at construction, so a + literal that slips past type checking can never silently bypass the + adapters' identity comparisons. Unknown values raise ``ValueError``. + """ mode: PermissionMode = PermissionMode.DEFAULT filesystem: FilesystemAccess = FilesystemAccess.WORKSPACE_WRITE @@ -252,6 +276,16 @@ class PermissionProfile: disallowed_tools: tuple[str, ...] = () network: bool | None = None + def __post_init__(self) -> None: + if not isinstance(self.mode, PermissionMode): + object.__setattr__(self, "mode", _coerce_enum(PermissionMode, self.mode, "mode")) + if not isinstance(self.filesystem, FilesystemAccess): + object.__setattr__( + self, + "filesystem", + _coerce_enum(FilesystemAccess, self.filesystem, "filesystem"), + ) + @dataclass(frozen=True) class ToolCallAudit: @@ -371,6 +405,26 @@ def cost_usd(self) -> float: return self.usage.cost_usd +_ParsedT = TypeVar("_ParsedT") + + +@dataclass(frozen=True) +class ParsedResult(AgentResult, Generic[_ParsedT]): + """An ``AgentResult`` whose ``parsed_output`` was validated as ``output_type``. + + Produced by ``AgentKit.run(..., output_type=T)``. Runtime-identical to + ``AgentResult`` — it only adds the typed accessor. ``AgentResult`` itself + stays non-generic so existing bare ``AgentResult`` annotations remain valid + under downstream ``disallow_any_generics`` strictness. + """ + + @property + def parsed(self) -> _ParsedT | None: + """The validated instance, or ``None`` when the run failed.""" + + return cast("_ParsedT | None", self.parsed_output) + + @runtime_checkable class AgentRuntime(Protocol): """Async runtime that drives an ``AgentTask`` to completion.""" diff --git a/tests/test_core.py b/tests/test_core.py index b9d03cd..c60668d 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -15,8 +15,11 @@ AgentTask, ArtifactRef, FakeAgentRuntime, + FilesystemAccess, FinishReason, McpServerConfig, + PermissionMode, + PermissionProfile, RuntimeAvailability, RuntimeNotRegisteredError, ToolCallAudit, @@ -66,6 +69,30 @@ def test_registry_accepts_namespaced_third_party_kind() -> None: assert isinstance(runtime, FakeAgentRuntime) +def test_permission_profile_coerces_string_literals_to_enum_members() -> None: + profile = PermissionProfile(mode="strict", filesystem="read-only") # type: ignore[arg-type] + + # Identity, not just equality: adapters gate on `mode is PermissionMode.STRICT`, + # so coercion must produce the actual members or the posture silently loosens. + assert profile.mode is PermissionMode.STRICT + assert profile.filesystem is FilesystemAccess.READ_ONLY + # Enum members pass through untouched. + assert PermissionProfile(mode=PermissionMode.CAUTIOUS).mode is PermissionMode.CAUTIOUS + + +def test_permission_profile_rejects_unknown_literals() -> None: + with pytest.raises(ValueError) as exc_info: + PermissionProfile(mode="paranoid") # type: ignore[arg-type] + + message = str(exc_info.value) + assert "paranoid" in message + # The error teaches the valid vocabulary. + assert "strict" in message and "permissive" in message + + with pytest.raises(ValueError): + PermissionProfile(filesystem="ro") # type: ignore[arg-type] + + def test_agent_task_model_fields_are_keyword_only() -> None: from pathlib import Path diff --git a/tests/test_kit.py b/tests/test_kit.py new file mode 100644 index 0000000..b9a302e --- /dev/null +++ b/tests/test_kit.py @@ -0,0 +1,340 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pytest + +from agent_runtime_kit import ( + AgentKit, + AgentResult, + AgentRuntimeKind, + AgentTask, + FakeAgentRuntime, + FilesystemAccess, + ParsedResult, + PermissionMode, + PermissionProfile, + RuntimeAvailability, +) +from agent_runtime_kit._schema import json_schema_for +from agent_runtime_kit.testing import RecordingEventSink + + +@dataclass +class Point: + x: int + y: int + + +class RecordingRuntime: + """Protocol-complete runtime that records the task and returns a canned result.""" + + kind = AgentRuntimeKind.FAKE + capabilities = FakeAgentRuntime().capabilities + + def __init__(self, result: AgentResult | None = None) -> None: + self.task: AgentTask | None = None + self.closed = False + self._result = result or AgentResult(output="recorded") + + def availability(self) -> RuntimeAvailability: + return RuntimeAvailability.ok(self.kind) + + async def run(self, task: AgentTask) -> AgentResult: + self.task = task + return self._result + + async def cancel(self, task_id: str) -> None: + del task_id + + async def aclose(self) -> None: + self.closed = True + + async def __aenter__(self) -> RecordingRuntime: + return self + + async def __aexit__(self, _exc_type: object, _exc: object, _tb: object) -> None: + await self.aclose() + + +@pytest.mark.asyncio +async def test_kit_runs_registered_fake_by_kind() -> None: + kit = AgentKit(register_default_adapters=False) + + result = await kit.run("fake", goal="hello") + + assert result.finish_reason == "done" + assert result.output + + +@pytest.mark.asyncio +async def test_kit_kwargs_map_onto_task_fields() -> None: + runtime = RecordingRuntime() + kit = AgentKit(register_default_adapters=False) + + await kit.run( + runtime, + goal="g", + system="be careful", + model="model-x", + reasoning_effort="high", + working_directory="/tmp/ws", + session_id="sess-1", + budget_usd=1.5, + task_id="task-1", + metadata={"stage": "demo"}, + ) + + task = runtime.task + assert task is not None + assert task.goal == "g" + assert task.system == "be careful" + assert task.model == "model-x" + assert task.reasoning_effort == "high" + assert task.working_directory == Path("/tmp/ws") + assert task.session_id == "sess-1" + assert task.budget_usd == 1.5 + assert task.task_id == "task-1" + assert task.metadata["stage"] == "demo" + + +@pytest.mark.asyncio +async def test_kit_permission_literals_reach_adapters_as_enum_members() -> None: + runtime = RecordingRuntime() + kit = AgentKit(register_default_adapters=False) + + await kit.run( + runtime, + goal="g", + permissions="strict", + filesystem="read-only", + allowed_tools=("view_file",), + ) + + task = runtime.task + assert task is not None + # Identity: the adapters' `is` checks must see real members. + assert task.permissions.mode is PermissionMode.STRICT + assert task.permissions.filesystem is FilesystemAccess.READ_ONLY + assert task.permissions.allowed_tools == ("view_file",) + + +@pytest.mark.asyncio +async def test_kit_rejects_profile_alongside_tool_kwargs() -> None: + kit = AgentKit(register_default_adapters=False) + + with pytest.raises(ValueError, match="inside the"): + await kit.run( + RecordingRuntime(), + goal="g", + permissions=PermissionProfile(), + allowed_tools=("view_file",), + ) + + +@pytest.mark.asyncio +async def test_kit_requires_exactly_one_of_goal_and_task() -> None: + kit = AgentKit(register_default_adapters=False) + + with pytest.raises(ValueError, match="either goal"): + await kit.run(RecordingRuntime()) + + with pytest.raises(ValueError, match="mutually exclusive"): + await kit.run(RecordingRuntime(), task=AgentTask(goal="a"), goal="b") + + +@pytest.mark.asyncio +async def test_kit_task_passthrough_runs_unchanged() -> None: + runtime = RecordingRuntime() + kit = AgentKit(register_default_adapters=False) + task = AgentTask(goal="prebuilt", session_id="keep-me") + + await kit.run(runtime, task=task) + + assert runtime.task is task + + +def test_kit_aliases_resolve_builtin_kinds() -> None: + kit = AgentKit() + + # Adapters register lazily and are constructible without vendor SDKs. + assert kit.capabilities_for("claude").streaming is True + assert kit.capabilities_for("codex").streaming is False + # Exact kind strings keep working; unknown aliases are not invented. + assert kit.capabilities_for("claude-agent-sdk").streaming is True + assert set(kit.kinds()) >= { + AgentRuntimeKind.FAKE, + AgentRuntimeKind.CLAUDE_AGENT_SDK, + AgentRuntimeKind.CODEX_AGENT_SDK, + AgentRuntimeKind.ANTIGRAVITY_AGENT_SDK, + } + assert len(kit.availability()) == len(kit.kinds()) + + +@pytest.mark.asyncio +async def test_kit_output_type_generates_schema_and_parses_result() -> None: + runtime = RecordingRuntime( + AgentResult(output='{"x": 1, "y": 2}', parsed_output={"x": 1, "y": 2}) + ) + kit = AgentKit(register_default_adapters=False) + + result = await kit.run(runtime, goal="point please", output_type=Point) + + assert isinstance(result, ParsedResult) + assert result.parsed == Point(x=1, y=2) + assert result.parsed_output == Point(x=1, y=2) + # The wire schema came from the type, not hand-written JSON schema. + task = runtime.task + assert task is not None + assert dict(task.output_schema or {}) == json_schema_for(Point) + + +@pytest.mark.asyncio +async def test_kit_output_type_mismatch_fails_the_result_not_the_call() -> None: + runtime = RecordingRuntime( + AgentResult(output='{"x": "no"}', parsed_output={"x": "no", "y": 2}) + ) + kit = AgentKit(register_default_adapters=False) + + result = await kit.run(runtime, goal="point please", output_type=Point) + + assert result.finish_reason == "failed" + assert result.error is not None and "Point" in result.error + assert result.parsed is None + + +@pytest.mark.asyncio +async def test_kit_output_type_passes_adapter_failures_through() -> None: + failed = AgentResult(output="", finish_reason="failed", error="vendor exploded") + runtime = RecordingRuntime(failed) + kit = AgentKit(register_default_adapters=False) + + result = await kit.run(runtime, goal="g", output_type=Point) + + assert result.finish_reason == "failed" + assert result.error == "vendor exploded" + assert result.parsed is None + + +@pytest.mark.asyncio +async def test_kit_output_type_and_output_schema_are_exclusive() -> None: + kit = AgentKit(register_default_adapters=False) + + with pytest.raises(ValueError, match="mutually exclusive"): + await kit.run( + RecordingRuntime(), + goal="g", + output_type=Point, + output_schema={"type": "object"}, + ) + + +@pytest.mark.asyncio +async def test_kit_output_type_composes_with_task_passthrough() -> None: + runtime = RecordingRuntime(AgentResult(output="{}", parsed_output={"x": 3, "y": 4})) + kit = AgentKit(register_default_adapters=False) + + result = await kit.run(runtime, task=AgentTask(goal="prebuilt"), output_type=Point) + + assert result.parsed == Point(x=3, y=4) + task = runtime.task + assert task is not None + assert task.output_schema is not None # schema injected into the prebuilt task + + +@pytest.mark.asyncio +async def test_kit_on_handlers_filter_compose_and_never_break_runs() -> None: + kit = AgentKit(register_default_adapters=False) + completed: list[str] = [] + everything: list[str] = [] + awaited: list[str] = [] + user_sink = RecordingEventSink() + + @kit.on("agent.task.completed") + def only_completed(event: Mapping[str, Any]) -> None: + completed.append(str(event["name"])) + + @kit.on() + def wildcard(event: Mapping[str, Any]) -> None: + everything.append(str(event["name"])) + + @kit.on("agent.task.started") + async def async_handler(event: Mapping[str, Any]) -> None: + awaited.append(str(event["name"])) + + @kit.on() + def exploding(event: Mapping[str, Any]) -> None: + raise RuntimeError("handler bug") + + result = await kit.run("fake", goal="observe me", event_sink=user_sink) + + assert result.finish_reason == "done" + assert completed == ["agent.task.completed"] + assert awaited == ["agent.task.started"] + # The fake runtime emits the full normalized sequence; wildcard saw it all. + assert everything[0] == "agent.task.started" + assert everything[-1] == "agent.task.completed" + # The raising handler broke neither the run, the other handlers, nor the + # user's own sink. + assert [event["name"] for event in user_sink.events] == everything + + +@pytest.mark.asyncio +async def test_kit_on_handlers_apply_to_prebuilt_tasks() -> None: + runtime = RecordingRuntime() + kit = AgentKit(register_default_adapters=False) + seen: list[str] = [] + + @kit.on() + def watch(event: Mapping[str, Any]) -> None: + seen.append(str(event["name"])) + + await kit.run(runtime, task=AgentTask(goal="prebuilt")) + + task = runtime.task + assert task is not None + assert task.event_sink is not None # tee injected into the passthrough task + await task.event_sink.emit({"name": "agent.task.started"}) + assert seen == ["agent.task.started"] + + +@pytest.mark.asyncio +async def test_kit_runtime_decorator_registers_factory() -> None: + kit = AgentKit(register_default_adapters=False) + + @kit.runtime("x-third-party") + def factory(**_: Any) -> FakeAgentRuntime: + return FakeAgentRuntime(output="third-party output") + + result = await kit.run("x-third-party", goal="g") + + assert result.output == "third-party output" + assert "x-third-party" in kit.kinds() + # Zero-arg constructibility keeps diagnostics working. + assert kit.availability_for("x-third-party").available is True + + +@pytest.mark.asyncio +async def test_kit_caches_runtimes_per_kind_and_closes_them() -> None: + constructed: list[RecordingRuntime] = [] + + def factory(**_: Any) -> RecordingRuntime: + runtime = RecordingRuntime() + constructed.append(runtime) + return runtime + + async with AgentKit(register_default_adapters=False) as kit: + kit.registry.register("x-counting", factory) + await kit.run("x-counting", goal="one") + await kit.run("x-counting", goal="two") + assert len(constructed) == 1 + + assert constructed[0].closed is True + # A directly passed instance is the caller's to manage: never closed by the hub. + outside = RecordingRuntime() + async with AgentKit(register_default_adapters=False) as kit: + await kit.run(outside, goal="g") + assert outside.closed is False diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..d717d3d --- /dev/null +++ b/tests/test_schema.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from typing import Any, Literal, TypedDict + +import pytest + +from agent_runtime_kit import OutputTypeError +from agent_runtime_kit._schema import json_schema_for, parse_as + + +class Color(str, enum.Enum): + RED = "red" + BLUE = "blue" + + +@dataclass +class Point: + x: int + y: int + label: str = "origin" + + +@dataclass +class Shape: + name: Literal["circle", "square"] + points: list[Point] + color: Color | None = None + tags: dict[str, str] = field(default_factory=dict) + + +class Movie(TypedDict): + title: str + year: int + + +def test_schema_scalars_and_containers() -> None: + assert json_schema_for(str) == {"type": "string"} + assert json_schema_for(bool) == {"type": "boolean"} + assert json_schema_for(int) == {"type": "integer"} + assert json_schema_for(float) == {"type": "number"} + assert json_schema_for(list[int]) == {"type": "array", "items": {"type": "integer"}} + assert json_schema_for(dict[str, bool]) == { + "type": "object", + "additionalProperties": {"type": "boolean"}, + } + assert json_schema_for(str | None) == {"anyOf": [{"type": "string"}, {"type": "null"}]} + assert json_schema_for(Literal["a", "b"]) == {"enum": ["a", "b"]} + assert json_schema_for(Color) == {"type": "string", "enum": ["red", "blue"]} + + +def test_schema_dataclass_shape() -> None: + schema = json_schema_for(Shape) + + assert schema["type"] == "object" + assert schema["additionalProperties"] is False + # Only fields without defaults are required. + assert schema["required"] == ["name", "points"] + assert schema["properties"]["name"] == {"enum": ["circle", "square"]} + assert schema["properties"]["points"] == { + "type": "array", + "items": json_schema_for(Point), + } + assert schema["properties"]["color"] == { + "anyOf": [{"type": "string", "enum": ["red", "blue"]}, {"type": "null"}] + } + assert json_schema_for(Point)["required"] == ["x", "y"] + + +def test_schema_typeddict_shape() -> None: + schema = json_schema_for(Movie) + + assert schema["type"] == "object" + assert schema["additionalProperties"] is False + assert schema["required"] == ["title", "year"] + assert schema["properties"]["year"] == {"type": "integer"} + + +@pytest.mark.parametrize( + "annotation", + [ + set[str], + tuple[str, ...], + str | int, + list, + dict[int, str], + bytes, + Any, + ], + ids=["set", "tuple", "union", "bare-list", "int-keyed-dict", "bytes", "any"], +) +def test_schema_unsupported_annotations_fail_closed(annotation: Any) -> None: + with pytest.raises(OutputTypeError): + json_schema_for(annotation) + + +def test_schema_recursive_dataclass_fails_closed() -> None: + @dataclass + class Node: + children: list[Node] + + # get_type_hints resolves the local forward reference via localns lookup + # failing — either way the bounded walk must raise, never overflow. + with pytest.raises((OutputTypeError, NameError)): + json_schema_for(Node) + + +def test_parse_as_nested_dataclass_happy_path() -> None: + shape = parse_as( + Shape, + { + "name": "circle", + "points": [{"x": 1, "y": 2}, {"x": 3, "y": 4, "label": "corner"}], + "color": "red", + "tags": {"kind": "demo"}, + }, + ) + + assert isinstance(shape, Shape) + assert shape.points[0] == Point(x=1, y=2) # default label filled + assert shape.points[1].label == "corner" + assert shape.color is Color.RED + assert shape.tags == {"kind": "demo"} + # Optional accepts absent and null alike. + assert parse_as(Shape, {"name": "square", "points": [], "color": None}).color is None + + +def test_parse_as_rejects_extra_and_missing_keys() -> None: + with pytest.raises(OutputTypeError, match="unexpected keys"): + parse_as(Point, {"x": 1, "y": 2, "z": 3}) + with pytest.raises(OutputTypeError, match="missing required key"): + parse_as(Point, {"x": 1}) + with pytest.raises(OutputTypeError, match="missing required keys"): + parse_as(Movie, {"title": "Alien"}) + + +def test_parse_as_is_strict_about_scalars() -> None: + # No cross-type coercion: strings stay strings, bool never satisfies int. + with pytest.raises(OutputTypeError, match=r"\$\.x: expected int"): + parse_as(Point, {"x": "1", "y": 2}) + with pytest.raises(OutputTypeError, match="expected int"): + parse_as(int, True) + # float accepts an integral JSON number (JSON has one number type). + assert parse_as(float, 3) == 3.0 + assert parse_as(bool, True) is True + + +def test_parse_as_error_paths_point_into_the_payload() -> None: + with pytest.raises(OutputTypeError, match=r"\$\.points\[1\]\.y"): + parse_as(Shape, {"name": "circle", "points": [{"x": 1, "y": 2}, {"x": 1, "y": "no"}]}) + with pytest.raises(OutputTypeError, match="not one of"): + parse_as(Shape, {"name": "triangle", "points": []}) + with pytest.raises(OutputTypeError, match="not one of"): + parse_as(Color, "green") + + +class DuckModel: + """Pydantic-shaped without Pydantic: structural protocol must suffice.""" + + @classmethod + def model_json_schema(cls) -> dict[str, Any]: + return {"type": "object", "properties": {"n": {"type": "integer"}}} + + @classmethod + def model_validate(cls, value: Any) -> DuckModel: + if not isinstance(value, dict) or "n" not in value: + raise ValueError("n is required") + instance = cls() + instance.n = value["n"] # type: ignore[attr-defined] + return instance + + +def test_model_protocol_duck_typing_covers_both_directions() -> None: + assert json_schema_for(DuckModel) == { + "type": "object", + "properties": {"n": {"type": "integer"}}, + } + + parsed = parse_as(DuckModel, {"n": 7}) + assert isinstance(parsed, DuckModel) + assert parsed.n == 7 # type: ignore[attr-defined] + + # Validation failures wrap into the typed error, whatever the model raised. + with pytest.raises(OutputTypeError, match="rejected the payload"): + parse_as(DuckModel, {"wrong": 1}) + + +def test_model_protocol_bad_schema_shape_fails_closed() -> None: + class BadDuck: + @classmethod + def model_json_schema(cls) -> list[int]: + return [1] + + @classmethod + def model_validate(cls, value: Any) -> BadDuck: + return cls() + + with pytest.raises(OutputTypeError, match="expected a mapping"): + json_schema_for(BadDuck) diff --git a/uv.lock b/uv.lock index c8cb881..df10712 100644 --- a/uv.lock +++ b/uv.lock @@ -12,7 +12,7 @@ required-markers = [ ] [options] -exclude-newer = "2026-06-24T15:20:56.671469Z" +exclude-newer = "2026-06-24T17:18:14.378186Z" exclude-newer-span = "P8D" [manifest] @@ -29,7 +29,7 @@ wheels = [ [[package]] name = "agent-runtime-kit" -version = "0.3.0" +version = "0.4.0" source = { editable = "." } [package.optional-dependencies]