fix: validate structured output across providers

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   schema checks and `OutputSchemaError` for malformed definitions.
 - `AgentResult.parsed_output_available` distinguishes a valid JSON `null` from
   absent or rejected structured output.
+- All built-in adapters and public fake runtimes now validate returned values
+  against `output_schema` locally, including textual JSON fallbacks.
 
 ### Changed
 

README.md

@@ -141,10 +141,11 @@ a field (for example only Claude maps `budget_usd`; Codex and Antigravity reject
 it with a typed `UnsupportedTaskInputError`) the adapter raises rather than
 silently dropping it.
 
-`AgentResult` returns output, finish reason (see `FinishReason`), parsed
-structured output, usage, cost, session id, tool-call audits, and provider
-metadata. `artifacts` is a reserved field: no built-in runtime populates it yet,
-so it is always an empty tuple today.
+`AgentResult` returns output, finish reason (see `FinishReason`), locally
+validated structured output, usage, cost, session id, tool-call audits, and
+provider metadata. `parsed_output_available` distinguishes an absent payload
+from valid JSON `null` (both use `parsed_output=None`). `artifacts` is a reserved
+field: no built-in runtime populates it yet, so it is always an empty tuple today.
 
 ## Docs
 

docs/api-stability.md

@@ -45,7 +45,9 @@ import the names from the top-level package instead.
   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(strict=True)` when
-  present). Raw schemas are checked at construction.
+  present). Raw schemas are checked at construction and again at dispatch
+  because their nested values are only shallow-frozen. Every built-in adapter
+  validates returned structured values locally.
 - **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

docs/capability-matrix.md

@@ -20,6 +20,10 @@ The matrix is intentionally not a lowest-common-denominator contract. Adapters
 reject unsupported inputs (see below) when silently dropping them would be
 misleading.
 
+For every provider, malformed schemas are rejected locally before dispatch and
+returned structured values are validated locally after the SDK completes.
+`parsed_output_available` distinguishes valid JSON `null` from no parsed value.
+
 ## Permission mapping
 
 A single portable `PermissionMode` maps to each vendor's native controls. The

docs/providers.md

@@ -12,7 +12,8 @@
 
 ## Install Model
 
-The plain `agent-runtime-kit` package is the dependency-free core. Use
+The plain `agent-runtime-kit` package is the vendor-SDK-free core and includes
+lightweight JSON Schema validation. Use
 `agent-runtime-kit[all]` when you want Claude, Codex, and Antigravity adapters
 available in one install, or use a single extra such as
 `agent-runtime-kit[codex]` when your application only needs one runtime.

docs/quickstart.md

@@ -84,6 +84,8 @@ async def main() -> None:
 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`.
+All built-in adapters also validate results requested with a raw
+`output_schema` and include the failing JSON path 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

src/agent_runtime_kit/adapters/_common.py

@@ -3,13 +3,19 @@
 from __future__ import annotations
 
 import inspect
-import json
 import os
 from collections.abc import Iterable, Mapping
 from importlib import metadata, util
 from typing import Any
 
 from agent_runtime_kit._errors import UnsupportedTaskInputError
+from agent_runtime_kit._schema import (
+    STRUCTURED_OUTPUT_MISSING as STRUCTURED_OUTPUT_MISSING,
+)
+from agent_runtime_kit._schema import (
+    resolve_structured_output as resolve_structured_output,
+)
+from agent_runtime_kit._schema import validate_output_schema
 from agent_runtime_kit._types import (
     AgentRuntimeKind,
     AgentTask,
@@ -88,35 +94,18 @@ def output_schema_from(
     """Resolve output schema from first-class task field or metadata aliases."""
 
     if task_output_schema is not None:
+        # Revalidate at dispatch: AgentTask's mapping freeze is deliberately
+        # shallow, so nested schema values may have changed since construction.
+        validate_output_schema(task_output_schema)
         return task_output_schema
     for key in ("output_schema", "json_schema"):
         raw = metadata_values.get(key)
         if isinstance(raw, Mapping):
+            validate_output_schema(raw)
             return raw
     return None
 
 
-def parse_json_output(output: str) -> Any | None:
-    """Best-effort JSON parsing for structured-output fallbacks."""
-
-    try:
-        return json.loads(output)
-    except json.JSONDecodeError:
-        return None
-
-
-def structured_output_unsatisfied_error(sdk_label: str) -> str:
-    """Uniform error text when a requested ``output_schema`` produced no JSON.
-
-    The adapters verify JSON-parseability, not schema conformance (validation
-    stays with the vendor SDK and the caller), so the message claims exactly
-    that. Kept in one place so all three adapters fail identically instead of
-    each inventing its own message (and, previously, its own verdict).
-    """
-
-    return f"{sdk_label} returned no parseable JSON for the requested output_schema"
-
-
 def empty_completion_error(sdk_label: str) -> str:
     """Uniform error text when a runtime completed with nothing usable.
 

src/agent_runtime_kit/adapters/antigravity.py

@@ -36,9 +36,8 @@
     optional_str,
     output_schema_from,
     package_availability,
-    parse_json_output,
     reject_unsupported_inputs,
-    structured_output_unsatisfied_error,
+    resolve_structured_output,
 )
 from agent_runtime_kit.events import (
     output_delta_event,
@@ -375,8 +374,6 @@ async def _invoke(
             metadata["dropped_options"] = list(dropped_options)
 
         output = "".join(text_parts).strip()
-        if structured_output is None and schema is not None:
-            structured_output = parse_json_output(output)
 
         # A non-natural stop (token limit, safety block) is a failure, not a
         # successful completion of whatever partial text arrived first.
@@ -386,25 +383,35 @@ async def _invoke(
                 output=output,
                 finish_reason=terminal_reason,
                 error=terminal_error,
-                parsed_output=structured_output,
                 usage=_usage_from(usage_metadata),
                 tool_calls=tuple(tool_calls),
                 session_id=session_id,
                 rounds=1,
                 metadata=metadata,
             )
-        if schema is not None and structured_output is None:
-            return AgentResult(
-                output=output,
-                finish_reason="failed",
-                error=structured_output_unsatisfied_error("Antigravity SDK"),
-                usage=_usage_from(usage_metadata),
-                tool_calls=tuple(tool_calls),
-                session_id=session_id,
-                rounds=1,
-                metadata=metadata,
+        parsed_output: Any = None
+        parsed_output_available = False
+        if schema is not None:
+            resolution = resolve_structured_output(
+                schema,
+                output,
+                sdk_label="Antigravity SDK",
+                native=structured_output,
             )
-        if not output and not tool_calls and structured_output is None:
+            if resolution.error is not None:
+                return AgentResult(
+                    output=output,
+                    finish_reason="failed",
+                    error=resolution.error,
+                    usage=_usage_from(usage_metadata),
+                    tool_calls=tuple(tool_calls),
+                    session_id=session_id,
+                    rounds=1,
+                    metadata=metadata,
+                )
+            parsed_output = resolution.value
+            parsed_output_available = resolution.available
+        if not output and not tool_calls and not parsed_output_available:
             return AgentResult(
                 output="",
                 finish_reason="failed",
@@ -417,7 +424,8 @@ async def _invoke(
             )
         return AgentResult(
             output=output,
-            parsed_output=structured_output,
+            parsed_output=parsed_output,
+            parsed_output_available=parsed_output_available,
             usage=_usage_from(usage_metadata),
             tool_calls=tuple(tool_calls),
             session_id=session_id,

src/agent_runtime_kit/adapters/claude.py

@@ -23,6 +23,7 @@
     Usage,
 )
 from agent_runtime_kit.adapters._common import (
+    STRUCTURED_OUTPUT_MISSING,
     close_vendor_resource,
     empty_completion_error,
     ensure_supported_model,
@@ -33,9 +34,8 @@
     optional_str,
     output_schema_from,
     package_availability,
-    parse_json_output,
     reject_unsupported_inputs,
-    structured_output_unsatisfied_error,
+    resolve_structured_output,
 )
 from agent_runtime_kit.events import (
     output_delta_event,
@@ -475,7 +475,7 @@ def _translate_messages(
     rounds = 0
     error: str | None = None
     finish_reason = "done"
-    structured_output: Any | None = None
+    structured_output: Any = STRUCTURED_OUTPUT_MISSING
 
     for message in messages:
         message_type = _message_type(message)
@@ -503,15 +503,23 @@ def _translate_messages(
                 finish_reason, error = _result_failure(message, result_text)
 
     output = "\n".join(part for part in content_parts if part).strip()
-    schema_requested = output_schema_from(task.output_schema, task.metadata) is not None
-    if structured_output is None and schema_requested:
-        structured_output = parse_json_output(output)
-    if error is None and schema_requested and structured_output is None:
-        # Match Codex/Antigravity: a requested schema that cannot be satisfied is a
-        # failure, not a silently-successful task with parsed_output=None.
-        finish_reason = "failed"
-        error = structured_output_unsatisfied_error("Claude Agent SDK")
-    elif error is None and not output and not tool_calls and structured_output is None:
+    schema = output_schema_from(task.output_schema, task.metadata)
+    parsed_output: Any = None
+    parsed_output_available = False
+    if error is None and schema is not None:
+        resolution = resolve_structured_output(
+            schema,
+            output,
+            sdk_label="Claude Agent SDK",
+            native=structured_output,
+        )
+        if resolution.error is not None:
+            finish_reason = "failed"
+            error = resolution.error
+        else:
+            parsed_output = resolution.value
+            parsed_output_available = resolution.available
+    if error is None and not output and not tool_calls and not parsed_output_available:
         finish_reason = "failed"
         error = empty_completion_error("Claude Agent SDK")
     usage = Usage(
@@ -535,7 +543,8 @@ def _translate_messages(
         output=output,
         finish_reason=finish_reason,
         error=error,
-        parsed_output=structured_output,
+        parsed_output=parsed_output,
+        parsed_output_available=parsed_output_available,
         usage=usage,
         tool_calls=tuple(tool_calls),
         session_id=session_id,

src/agent_runtime_kit/adapters/codex.py

@@ -30,9 +30,8 @@
     optional_int,
     output_schema_from,
     package_availability,
-    parse_json_output,
     reject_unsupported_inputs,
-    structured_output_unsatisfied_error,
+    resolve_structured_output,
 )
 from agent_runtime_kit.events import (
     output_delta_event,
@@ -499,19 +498,25 @@ def _translate_run_result(
     # status is "completed" or missing (dict-based fakes): treat as success and keep
     # the empty-output and schema-parse-failure branches.
     schema = output_schema_from(task.output_schema, task.metadata)
-    parsed = parse_json_output(output) if schema is not None else None
-    if schema is not None and parsed is None:
-        return AgentResult(
-            output=output,
-            finish_reason="failed",
-            error=structured_output_unsatisfied_error("Codex SDK"),
-            usage=usage,
-            tool_calls=tool_calls,
-            session_id=session_id,
-            rounds=1,
-            metadata=metadata,
-        )
-    if not output and not tool_calls and parsed is None:
+    parsed: Any = None
+    parsed_available = False
+    if schema is not None:
+        resolution = resolve_structured_output(schema, output, sdk_label="Codex SDK")
+        if resolution.error is None:
+            parsed = resolution.value
+            parsed_available = resolution.available
+        else:
+            return AgentResult(
+                output=output,
+                finish_reason="failed",
+                error=resolution.error,
+                usage=usage,
+                tool_calls=tool_calls,
+                session_id=session_id,
+                rounds=1,
+                metadata=metadata,
+            )
+    if not output and not tool_calls and not parsed_available:
         # Empty output alone is not a failure when the turn did real tool work;
         # only a completion with nothing usable is (matches Claude/Antigravity).
         return AgentResult(
@@ -527,6 +532,7 @@ def _translate_run_result(
     return AgentResult(
         output=output,
         parsed_output=parsed,
+        parsed_output_available=parsed_available,
         usage=usage,
         tool_calls=tool_calls,
         session_id=session_id,
@@ -737,4 +743,3 @@ def _uses_bedrock_provider(config_overrides: tuple[str, ...]) -> bool:
         if normalized_key == "model_provider" and normalized_value == "amazon-bedrock":
             return True
     return False
-

src/agent_runtime_kit/testing/fakes.py

@@ -7,6 +7,7 @@
 from typing import Any
 
 from agent_runtime_kit._errors import AgentRuntimeUnavailableError, UnsupportedTaskInputError
+from agent_runtime_kit._schema import resolve_structured_output
 from agent_runtime_kit._types import (
     AgentCapabilities,
     AgentResult,
@@ -39,9 +40,14 @@ class FakeSDKScenario:
     timeout: bool = False
     session_id: str | None = "fake-session"
     structured_output: Any | None = None
+    structured_output_available: bool = False
     tool_events: tuple[ToolCallAudit, ...] = ()
     metadata: Mapping[str, Any] = field(default_factory=dict)
 
+    def __post_init__(self) -> None:
+        if self.structured_output is not None and not self.structured_output_available:
+            object.__setattr__(self, "structured_output_available", True)
+
 
 class FakeSDKHarness:
     """Credential-free fake SDK surface with scripted outcomes."""
@@ -76,9 +82,30 @@ async def invoke(self, task: AgentTask) -> AgentResult:
                 session_id=scenario.session_id,
                 metadata=dict(scenario.metadata),
             )
+        parsed_output = scenario.structured_output
+        parsed_output_available = scenario.structured_output_available
+        if task.output_schema is not None:
+            resolution = resolve_structured_output(
+                task.output_schema,
+                scenario.output,
+                sdk_label="Fake SDK",
+                native=scenario.structured_output,
+                native_available=scenario.structured_output_available,
+            )
+            if resolution.error is not None:
+                return AgentResult(
+                    output=scenario.output,
+                    finish_reason="failed",
+                    error=resolution.error,
+                    session_id=scenario.session_id,
+                    metadata=dict(scenario.metadata),
+                )
+            parsed_output = resolution.value
+            parsed_output_available = resolution.available
         return AgentResult(
             output=scenario.output,
-            parsed_output=scenario.structured_output,
+            parsed_output=parsed_output,
+            parsed_output_available=parsed_output_available,
             tool_calls=scenario.tool_events,
             session_id=scenario.session_id,
             rounds=1,