feat: enforce strict structured output primitives

CHANGELOG.md

@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- Core JSON Schema validation via `jsonschema`, including construction-time
+  schema checks and `OutputSchemaError` for malformed definitions.
+- `AgentResult.parsed_output_available` distinguishes a valid JSON `null` from
+  absent or rejected structured output.
+
+### Changed
+
+- Pydantic structured-output parsing now requests strict validation, so values
+  such as `"42"` no longer coerce into integer fields.
+
 ### Fixed
 
 - `AgentKit.run(task=...)` now rejects every explicitly supplied per-field task

docs/api-stability.md

@@ -44,7 +44,12 @@ import the names from the top-level package instead.
   `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).
+  through their own `model_json_schema`/`model_validate(strict=True)` when
+  present). Raw schemas are checked at construction.
+- **Structured-output presence is explicit.** `parsed_output_available` is true
+  for a validated payload, including JSON `null`. Existing non-null custom
+  runtime results opt in automatically; a custom runtime returning valid null
+  must set the flag explicitly.
 
 ## Vendor SDK version policy
 

docs/quickstart.md

@@ -51,9 +51,9 @@ asyncio.run(main())
 
 ## 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.
+Pass a type instead of hand-writing JSON Schema. Dataclasses and `TypedDict`s
+work without Pydantic; Pydantic v2 models are used through their own
+`model_json_schema`/`model_validate` methods and validated with `strict=True`.
 
 ```python
 from dataclasses import dataclass
@@ -75,17 +75,19 @@ async def main() -> None:
             goal="Summarize this repository",
             output_type=RepoSummary,
         )
-        if result.parsed is not None:
+        if result.parsed_output_available:
             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.
+A malformed raw `output_schema` raises `OutputSchemaError` before execution.
+`AgentKit` strictly validates values requested with `output_type`; a mismatch
+yields `finish_reason="failed"` with the details in `result.error`.
+Unsupported Python annotations (sets, plain unions, ...) raise
+`OutputTypeError` at call time. For nullable schemas,
+`parsed_output_available=True` with `parsed_output=None` represents valid JSON
+`null`.
 
 ## Events and custom runtimes
 

pyproject.toml

@@ -38,7 +38,7 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
     "Typing :: Typed",
 ]
-dependencies = []
+dependencies = ["jsonschema>=4.18,<5"]
 
 [project.optional-dependencies]
 # Cautious upper bounds on the pre-1.0 vendor SDKs: they move fast and have shipped
@@ -65,10 +65,12 @@ Issues = "https://github.com/ebarti/agent-runtime-kit/issues"
 dev = [
     "build>=1.2",
     "mypy>=1.8",
+    "pydantic>=2,<3",
     "pytest>=8.0",
     "pytest-asyncio>=0.23",
     "pytest-cov>=5.0",
     "ruff>=0.8",
+    "types-jsonschema>=4.18",
 ]
 
 [tool.uv]

src/agent_runtime_kit/__init__.py

@@ -3,6 +3,7 @@
 from agent_runtime_kit._errors import (
     AgentRuntimeError,
     AgentRuntimeUnavailableError,
+    OutputSchemaError,
     OutputTypeError,
     RuntimeNotRegisteredError,
     UnsupportedTaskInputError,
@@ -59,6 +60,7 @@
     "FinishReason",
     "KIND_ALIASES",
     "McpServerConfig",
+    "OutputSchemaError",
     "OutputTypeError",
     "ParsedResult",
     "PermissionMode",

src/agent_runtime_kit/_errors.py

@@ -38,6 +38,10 @@ class OutputTypeError(AgentRuntimeError, TypeError):
     """
 
 
+class OutputSchemaError(AgentRuntimeError, ValueError):
+    """An output_schema is not a valid JSON Schema."""
+
+
 class RuntimeNotRegisteredError(AgentRuntimeError, LookupError):
     """No runtime factory is registered for the requested runtime kind."""
 

src/agent_runtime_kit/_kit.py

@@ -456,8 +456,9 @@ def _parse_result(output_type: type[_T], result: AgentResult) -> ParsedResult[_T
         # Never expose an adapter's unvalidated raw payload through the typed
         # ``ParsedResult`` surface when the adapter itself reported failure.
         values["parsed_output"] = None
+        values["parsed_output_available"] = False
         return cast("ParsedResult[_T]", ParsedResult(**values))
-    if result.parsed_output is None:
+    if not result.parsed_output_available:
         return cast("ParsedResult[_T]", ParsedResult(**values))
     try:
         instance = parse_as(output_type, result.parsed_output)
@@ -466,6 +467,7 @@ def _parse_result(output_type: type[_T], result: AgentResult) -> ParsedResult[_T
             finish_reason=FinishReason.FAILED.value,
             error=f"structured output does not conform to {output_type.__name__}: {exc}",
             parsed_output=None,
+            parsed_output_available=False,
         )
         return cast("ParsedResult[_T]", ParsedResult(**values))
     values["parsed_output"] = instance

src/agent_runtime_kit/_runtime.py

@@ -1,11 +1,12 @@
-"""Dependency-free runtime implementations used by tests and examples."""
+"""Vendor-independent runtime implementations used by tests and examples."""
 
 from __future__ import annotations
 
 from collections.abc import Mapping
 from typing import Any
 
 from agent_runtime_kit._errors import UnsupportedTaskInputError
+from agent_runtime_kit._schema import resolve_structured_output
 from agent_runtime_kit._types import (
     AgentCapabilities,
     AgentResult,
@@ -64,6 +65,19 @@ async def run(self, task: AgentTask) -> AgentResult:
             _ensure_supported(self.kind, self.capabilities, task)
             output = self._output if self._output is not None else f"Fake result for: {task.goal}"
             parsed = {"output": output} if task.output_schema is not None else None
+            parsed_available = parsed is not None
+            structured_error: str | None = None
+            if task.output_schema is not None:
+                resolution = resolve_structured_output(
+                    task.output_schema,
+                    output,
+                    sdk_label="Fake runtime",
+                    native=parsed,
+                    native_available=True,
+                )
+                parsed = resolution.value
+                parsed_available = resolution.available
+                structured_error = resolution.error
             tool_call = ToolCallAudit(
                 tool_name="fake",
                 arguments={"goal": task.goal},
@@ -94,7 +108,10 @@ async def run(self, task: AgentTask) -> AgentResult:
             )
             result = AgentResult(
                 output=output,
+                finish_reason="failed" if structured_error is not None else "done",
+                error=structured_error,
                 parsed_output=parsed,
+                parsed_output_available=parsed_available,
                 tool_calls=(tool_call,),
                 session_id=task.session_id or task.task_id,
                 rounds=1,
@@ -103,6 +120,9 @@ async def run(self, task: AgentTask) -> AgentResult:
         except Exception as exc:
             await safe_emit(task, task_failed_event(task, self.kind, error=str(exc)))
             raise
+        if result.error is not None:
+            await safe_emit(task, task_failed_event(task, self.kind, error=result.error))
+            return result
         await safe_emit(task, task_completed_event(task, self.kind, result))
         return result
 

src/agent_runtime_kit/_schema.py

@@ -1,4 +1,4 @@
-"""Dependency-free bridge between Python types and JSON schema.
+"""Bridge between Python types, JSON Schema, and structured payloads.
 
 ``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
@@ -17,12 +17,19 @@
 
 import dataclasses
 import enum
+import inspect
+import json
 import types
 import typing
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any, Literal, Union, get_args, get_origin, get_type_hints
 
-from agent_runtime_kit._errors import OutputTypeError
+from jsonschema import Draft202012Validator
+from jsonschema.exceptions import SchemaError, best_match
+from jsonschema.validators import validator_for
+
+from agent_runtime_kit._errors import OutputSchemaError, OutputTypeError
 
 # Deep enough for any sane payload model; recursive dataclasses hit this bound
 # and fail closed instead of overflowing.
@@ -62,15 +69,116 @@ def parse_as(tp: Any, value: Any) -> Any:
     """
 
     if supports_model_protocol(tp):
+        validator = tp.model_validate
+        validate_kwargs: dict[str, Any] = {}
+        try:
+            signature = inspect.signature(validator)
+        except (TypeError, ValueError):
+            signature = None
+        if signature is not None and (
+            "strict" in signature.parameters
+            or any(
+                parameter.kind is inspect.Parameter.VAR_KEYWORD
+                for parameter in signature.parameters.values()
+            )
+        ):
+            validate_kwargs["strict"] = True
         try:
-            return tp.model_validate(value)
+            return validator(value, **validate_kwargs)
         except Exception as exc:
+            strict_note = "(strict=True)" if validate_kwargs else ""
             raise OutputTypeError(
-                f"{_name(tp)}.model_validate() rejected the payload: {exc}"
+                f"{_name(tp)}.model_validate{strict_note} rejected the payload: {exc}"
             ) from exc
     return _parse(tp, value, path="$", depth=0)
 
 
+def validate_output_schema(schema: Mapping[str, Any]) -> None:
+    """Raise OutputSchemaError unless schema is valid and JSON-serializable."""
+
+    materialized = dict(schema)
+    try:
+        json.dumps(materialized, allow_nan=False)
+    except (TypeError, ValueError) as exc:
+        raise OutputSchemaError(f"invalid output_schema: not JSON-serializable: {exc}") from exc
+
+    dialect = materialized.get("$schema")
+    default_validator: Any = Draft202012Validator if dialect is None else None
+    validator_cls = validator_for(
+        materialized,
+        default=default_validator,
+    )
+    if validator_cls is None:
+        raise OutputSchemaError(f"invalid output_schema: unknown $schema dialect {dialect!r}")
+    try:
+        validator_cls.check_schema(materialized)
+    except SchemaError as exc:
+        raise OutputSchemaError(f"invalid output_schema: {exc.message}") from exc
+
+
+def output_value_error(schema: Mapping[str, Any], value: Any) -> str | None:
+    """Return a concise instance-conformance error, or None when value is valid."""
+
+    materialized = dict(schema)
+    validate_output_schema(materialized)
+    validator_cls = validator_for(materialized, default=Draft202012Validator)
+    validator = validator_cls(materialized)
+    error = best_match(validator.iter_errors(value))
+    if error is None:
+        return None
+    return f"{error.json_path}: {error.message}"
+
+
+STRUCTURED_OUTPUT_MISSING = object()
+
+
+@dataclass(frozen=True)
+class StructuredOutputResolution:
+    """A locally validated output with explicit presence semantics."""
+
+    value: Any = None
+    available: bool = False
+    error: str | None = None
+
+
+def resolve_structured_output(
+    schema: Mapping[str, Any],
+    output: str,
+    *,
+    sdk_label: str,
+    native: Any = STRUCTURED_OUTPUT_MISSING,
+    native_available: bool | None = None,
+) -> StructuredOutputResolution:
+    """Resolve native/text JSON, preserve valid null, and enforce its schema."""
+
+    validate_output_schema(schema)
+    if native_available is None:
+        native_available = native is not STRUCTURED_OUTPUT_MISSING and native is not None
+
+    value = native
+    available = native_available
+    if not available:
+        try:
+            value = json.loads(output)
+            available = True
+        except json.JSONDecodeError:
+            value = None
+
+    if not available:
+        return StructuredOutputResolution(
+            error=f"{sdk_label} returned no parseable JSON for the requested output_schema"
+        )
+    mismatch = output_value_error(schema, value)
+    if mismatch is not None:
+        return StructuredOutputResolution(
+            error=(
+                f"{sdk_label} returned structured output that does not conform to "
+                f"output_schema: {mismatch}"
+            )
+        )
+    return StructuredOutputResolution(value=value, available=True)
+
+
 def _schema(tp: Any, *, depth: int) -> dict[str, Any]:
     if depth > _MAX_DEPTH:
         raise OutputTypeError(

src/agent_runtime_kit/_types.py

@@ -377,6 +377,11 @@ class AgentTask:
     def __post_init__(self) -> None:
         object.__setattr__(self, "metadata", _freeze_mapping(self.metadata))
         if self.output_schema is not None:
+            # Local import avoids an import cycle: _schema's typed errors import
+            # AgentRuntimeKind from this module.
+            from agent_runtime_kit._schema import validate_output_schema
+
+            validate_output_schema(self.output_schema)
             object.__setattr__(self, "output_schema", _freeze_mapping(self.output_schema))
 
 
@@ -394,9 +399,15 @@ class AgentResult:
     session_id: str | None = None
     rounds: int = 0
     metadata: Mapping[str, Any] = field(default_factory=dict)
+    # None is both valid JSON and the historical missing-value sentinel. This
+    # trailing bit preserves the distinction without shifting existing positional
+    # constructor arguments.
+    parsed_output_available: bool = False
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "metadata", _freeze_mapping(self.metadata))
+        if self.parsed_output is not None and not self.parsed_output_available:
+            object.__setattr__(self, "parsed_output_available", True)
 
     @property
     def cost_usd(self) -> float:
@@ -420,7 +431,7 @@ class ParsedResult(AgentResult, Generic[_ParsedT]):
 
     @property
     def parsed(self) -> _ParsedT | None:
-        """The validated instance, or ``None`` when the run failed."""
+        """The validated instance; use parsed_output_available to distinguish null."""
 
         return cast("_ParsedT | None", self.parsed_output)