refactor(runtime): centralize capability result projection

AGENTS.md

@@ -208,6 +208,13 @@ checker authorization out of plugins and search code.
   validation. Exercise incompatible-but-individually-valid values through the
   serialized installed-operation boundary and assert an invalid-request result
   with no execution or publication.
+- Parse agent-supplied JSON strictly into the owning Pydantic request model;
+  advertised integers must not accept numeric strings or other coercions. An
+  adapter prepares that typed request before provider readiness and executes
+  only the prepared value. Strict transport encoding is lossless: do not reduce
+  rationals, normalize Unicode, or otherwise repair semantic input before the
+  owning model validates it. At the final projection, require the published
+  Pydantic model to match the installed output contract before serializing it.
 - Mathematical inputs are not presumed confidential. Public diagnostics should
   expose a stable domain reason, path, limit, and recovery direction—not
   arbitrary rejected values, which may be unbounded or user-controlled. This

docs/explanation/architecture.md

@@ -165,10 +165,15 @@ external payload
 Pydantic owns complete request validation, including relationships among
 otherwise valid fields. Required agreement of parents, characteristics,
 presentations, axes, bases, labels, and bound identities is checked during the
-single request parse, before preflight or any provider call. JSON Schema is
-generated for discovery and is not executed as an additional validation pass
-for built-ins. Provider and subprocess output is a separate untrusted boundary
-and is parsed independently.
+single strict, lossless request parse, before provider readiness, preflight, or
+execution. Domain canonicalization happens only after the owning request model
+has accepted the original values; transport parsing does not reduce rationals
+or normalize Unicode.
+JSON Schema is generated for discovery and is not executed as an additional
+validation pass for built-ins. Before the single final serialization, dispatch
+checks that the published Pydantic model still generates the installed output
+contract. Provider and subprocess output is a separate untrusted boundary and
+is parsed independently.
 
 Preflight distinguishes supported, unsupported, provider unavailable, and
 resource-limit-exceeded outcomes. Where practical it estimates work, output
@@ -273,8 +278,8 @@ Using a request-local carrier does not change a value or grant assurance.
 ## Structural JSON
 
 The runtime retains direct bounded JSON functions rather than a codec
-framework: a strict loader, deterministic serializer, Pydantic adapters, and
-one explicit schema-only adapter. Generic JSON rejects duplicate keys and
+framework: a strict loader, deterministic serializer, and Pydantic adapters.
+Generic JSON rejects duplicate keys and
 unsupported numbers, enforces depth/member/byte limits, preserves keys and
 strings exactly, and performs no Unicode or mathematical normalization.
 

docs/reference/provider-runtime.md

@@ -86,26 +86,28 @@ SMT-LIB parser and proof APIs are checked lazily at first use. Missing or
 mismatched package identity is a broken installation, while a first-use
 readiness failure fails the selected invocation closed.
 
-Source-backed adapters can construct metadata without importing their
-implementation:
+Source-backed checker implementations can construct metadata without importing
+their implementation:
 
 ```python
 from jacobian.contracts.capabilities import CapabilityInstallTier
 from jacobian.provider_runtime import source_provider_runtime
 
 runtime = source_provider_runtime(
-    "example.provider",
+    "example.checker",
     version="1",
-    entrypoint="example_adapter:create_adapter",
+    entrypoint="example_checker:check",
     install_tier=CapabilityInstallTier.T1,
     license_id="MIT",
     license_files=("LICENSE",),
-    features=("exact-example-operation",),
+    features=("exact-example-check",),
 )
 ```
 
-The adapter places `runtime` in its descriptor. Registration remains
-fail-closed if the source identity cannot be resolved.
+The checker registration binds `runtime` to its independently authorized
+entrypoint. Registration remains fail-closed if the source identity cannot be
+resolved. External operation packages and adapter entrypoint discovery are not
+supported.
 
 ### External checker runtimes
 

src/jacobian/adapters/mcp/tools.py

@@ -309,11 +309,7 @@ def _bounded_run_result(
     if not (result.diagnostics and result.diagnostics[0].code == "UNKNOWN_CAPABILITY"):
         return result
     payload = result.model_dump(mode="json")
-    output = {
-        key: value
-        for key, value in payload["output"].items()
-        if key != "available_capability_ids"
-    }
+    output = payload["output"]
     output.update(_unknown_capability_context(runtime, result.capability_id))
     payload["output"] = output
     return CapabilityResult.model_validate(payload)

src/jacobian/builtin_capabilities.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from jacobian.capability_adapters import parse_capability_input
 from jacobian.contracts.capabilities import (
     CapabilityDescriptor,
     CapabilityDiagnostic,
@@ -11,7 +12,7 @@
 )
 from jacobian.contracts.lean import (
     LeanCheckOutput,
-    LeanEnvironment,
+    LeanCheckRequest,
 )
 from jacobian.contracts.results import ExecutionStatus
 from jacobian.lean_frontend.service import LeanService
@@ -40,16 +41,7 @@ def __init__(
             ),
             provider="jacobian.lean4",
             provider_runtime=provider_runtime,
-            input_schema={
-                "type": "object",
-                "properties": {
-                    "statement": {"type": "string", "minLength": 1, "maxLength": 2000},
-                    "proof": {"type": "string", "minLength": 1, "maxLength": 20000},
-                    "environment": {"enum": ["CORE", "MATHLIB"]},
-                },
-                "required": ["statement", "proof"],
-                "additionalProperties": False,
-            },
+            input_schema=LeanCheckRequest.model_json_schema(),
             output_schema=LeanCheckOutput.model_json_schema(),
             tags=(
                 "lean",
@@ -84,12 +76,14 @@ def __init__(
     def descriptor(self) -> CapabilityDescriptor:
         return self._descriptor
 
-    def invoke(self, request: CapabilityRequest) -> OperationProjection:
-        payload = request.input
+    def prepare(self, request: CapabilityRequest) -> LeanCheckRequest:
+        return parse_capability_input(LeanCheckRequest, request.input)
+
+    def invoke(self, payload: LeanCheckRequest) -> OperationProjection:
         checked = self.lean.verify(
-            statement=str(payload["statement"]),
-            proof=str(payload["proof"]),
-            environment=LeanEnvironment(str(payload.get("environment", "CORE"))),
+            statement=payload.statement,
+            proof=payload.proof,
+            environment=payload.environment,
         )
         verified = checked.result.verification_record_uri is not None
         evidence = (checked.certificate_uri,)

src/jacobian/canonical.py

@@ -169,6 +169,68 @@ def _normalize(value: Any, *, limits: CanonicalLimits, depth: int) -> Any:
     raise CanonicalizationError("unsupported JSON value type")
 
 
+def _validate_json_value(value: Any, *, limits: CanonicalLimits, depth: int) -> None:
+    """Validate bounded interoperable JSON without changing semantic values."""
+
+    if depth > limits.max_depth:
+        raise CanonicalizationError("JSON nesting exceeds the configured depth limit")
+    if isinstance(value, float):
+        raise CanonicalizationError("JSON floating-point numbers are not allowed")
+    if isinstance(value, int):
+        _validate_json_integer(value)
+        return
+    if isinstance(value, list):
+        _validate_json_items(value, limits=limits, depth=depth)
+        return
+    if isinstance(value, dict):
+        _validate_json_object(value, limits=limits, depth=depth)
+        return
+    if value is None or isinstance(value, str):
+        return
+    raise CanonicalizationError("unsupported JSON value type")
+
+
+def _validate_json_integer(value: int) -> None:
+    if abs(value) > _MAX_SAFE_JSON_INTEGER:
+        raise CanonicalizationError(
+            "JSON integers outside the interoperable range must be encoded as strings"
+        )
+
+
+def _validate_json_items(
+    values: list[Any], *, limits: CanonicalLimits, depth: int
+) -> None:
+    for value in values:
+        _validate_json_value(value, limits=limits, depth=depth + 1)
+
+
+def _validate_json_object(
+    value: dict[Any, Any], *, limits: CanonicalLimits, depth: int
+) -> None:
+    for key, nested in value.items():
+        if not isinstance(key, str):
+            raise CanonicalizationError("JSON object keys must be strings")
+        _validate_json_value(nested, limits=limits, depth=depth + 1)
+
+
+def encode_strict_json(
+    value: Any,
+    *,
+    limits: CanonicalLimits | None = None,
+) -> bytes:
+    """Encode bounded JSON deterministically without semantic normalization."""
+
+    active_limits = limits or CanonicalLimits()
+    _validate_json_value(value, limits=active_limits, depth=0)
+    try:
+        encoded = rfc8785.dumps(value)
+    except (rfc8785.CanonicalizationError, RecursionError) as exc:
+        raise CanonicalizationError("value cannot be encoded as strict JSON") from exc
+    if len(encoded) > active_limits.max_output_bytes:
+        raise CanonicalizationError("JSON exceeds the configured size limit")
+    return encoded
+
+
 def _normalize_object(
     value: dict[str, Any], *, limits: CanonicalLimits, depth: int
 ) -> dict[str, Any]:

src/jacobian/capability_adapters.py

@@ -2,29 +2,72 @@
 
 from __future__ import annotations
 
-from typing import Literal, Protocol, runtime_checkable
+from typing import Any, Protocol, TypeVar
 
+from pydantic import BaseModel
+
+from jacobian.canonical import CanonicalizationError, encode_strict_json
+from jacobian.capability_errors import CapabilityInvocationError
+from jacobian.contracts.base import ContractModel
 from jacobian.contracts.capabilities import (
     CapabilityDescriptor,
+    CapabilityDiagnostic,
     CapabilityRequest,
 )
 from jacobian.operation_projection import OperationProjection
 
+PreparedT = TypeVar("PreparedT")
 
-class CapabilityAdapter(Protocol):
-    """Operator-installed adapter; registration requires no MCP changes."""
 
-    @property
-    def descriptor(self) -> CapabilityDescriptor: ...
+def _contains_typed_value(value: Any) -> bool:
+    if isinstance(value, ContractModel):
+        return True
+    if isinstance(value, list):
+        return any(_contains_typed_value(item) for item in value)
+    if isinstance(value, dict):
+        return any(_contains_typed_value(item) for item in value.values())
+    return False
+
+
+def parse_capability_input[ModelT: BaseModel](
+    model: type[ModelT], payload: dict[str, Any]
+) -> ModelT:
+    """Parse one bounded request into its owning model.
 
-    def invoke(self, request: CapabilityRequest) -> OperationProjection: ...
+    Ordinary caller input crosses the strict JSON parser exactly once. Requests
+    containing values resolved from typed input ports bind those already-
+    validated objects directly in strict Python mode; the operation adapter has
+    separately accounted for their canonical JSON projection before this call.
+    """
 
+    if _contains_typed_value(payload):
+        return model.model_validate(payload, strict=True)
+    try:
+        encoded = encode_strict_json(payload)
+    except CanonicalizationError as exc:
+        raise CapabilityInvocationError(
+            CapabilityDiagnostic(
+                code="INVALID_REQUEST",
+                stage="capability_input_validation",
+                message="The capability request is not valid bounded JSON.",
+                hint=(
+                    "Use only JSON objects, arrays, strings, booleans, null, "
+                    "and supported finite numbers within the configured limits."
+                ),
+            )
+        ) from exc
+    return model.model_validate_json(encoded, strict=True)
+
+
+class CapabilityAdapter(Protocol[PreparedT]):
+    """Installed typed adapter; registration requires no MCP changes."""
+
+    @property
+    def descriptor(self) -> CapabilityDescriptor: ...
 
-@runtime_checkable
-class TypedInputAdapter(Protocol):
-    """Adapter that owns one typed input parse and needs no schema execution."""
+    def prepare(self, request: CapabilityRequest) -> PreparedT: ...
 
-    typed_input: Literal[True]
+    def invoke(self, prepared: PreparedT) -> OperationProjection: ...
 
 
-__all__ = ["CapabilityAdapter", "TypedInputAdapter"]
+__all__ = ["CapabilityAdapter", "parse_capability_input"]