Adapt selected LLM to governed reasoning

core/engine/core/provider_runtime.py

@@ -9,10 +9,14 @@
 from __future__ import annotations
 
 import contextvars
+import json
+import time
 from contextlib import contextmanager
 from dataclasses import asdict, dataclass
 from typing import Any, Iterator, TypeVar
 
+from core.engine.core.tokens import TokenAccumulator, clear_accumulator, get_accumulator, set_accumulator
+
 _ProviderT = TypeVar("_ProviderT")
 
 
@@ -76,3 +80,89 @@ def provider_resolution(provider: object) -> ProviderResolution | None:
 
     resolution = getattr(provider, "_ace_resolution", None)
     return resolution if isinstance(resolution, ProviderResolution) else None
+
+
+@dataclass(frozen=True)
+class StructuredProviderCallResult:
+    """One structured provider result with only actually observed call facts."""
+
+    structured_json: str
+    provider_id: str | None
+    model_id: str | None
+    configuration_digest: str | None
+    input_units: int | None
+    output_units: int | None
+    duration_ms: int
+
+    @property
+    def unavailable_fields(self) -> tuple[str, ...]:
+        values = {
+            "provider_id": self.provider_id,
+            "model_id": self.model_id,
+            "configuration_digest": self.configuration_digest,
+            "input_units": self.input_units,
+            "output_units": self.output_units,
+        }
+        return tuple(name for name, value in values.items() if value is None)
+
+
+async def complete_structured_provider_call(
+    provider: object,
+    *,
+    prompt: str,
+    model: str | None = None,
+    max_tokens: int = 4096,
+    configuration_digest: str | None = None,
+) -> StructuredProviderCallResult:
+    """Call the selected provider once and retain exact available telemetry.
+
+    The helper never estimates token counts or invents a route. Providers that
+    cannot expose a fact return it as ``None``; governed consumers decide
+    whether their stricter contract can proceed.
+    """
+
+    complete_json = getattr(provider, "complete_json", None)
+    if not callable(complete_json):
+        raise TypeError("selected provider does not support structured JSON completion")
+    inherited = get_accumulator()
+    accumulator = inherited or TokenAccumulator()
+    if inherited is None:
+        set_accumulator(accumulator)
+    token_before = len(accumulator.calls_snapshot())
+    logical_before = len(accumulator.llm_calls_snapshot())
+    started = time.monotonic()
+    try:
+        output = await complete_json(prompt, model=model, max_tokens=max_tokens)
+    finally:
+        duration_ms = max(0, int((time.monotonic() - started) * 1000))
+        if inherited is None:
+            clear_accumulator()
+    if not isinstance(output, dict):
+        raise TypeError("structured provider output must be one JSON object")
+    try:
+        structured_json = json.dumps(
+            output,
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        )
+    except (TypeError, ValueError) as exc:
+        raise TypeError("structured provider output must be finite JSON") from exc
+
+    token_calls = accumulator.calls_snapshot()[token_before:]
+    logical_calls = accumulator.llm_calls_snapshot()[logical_before:]
+    input_units = sum(int(item["input_tokens"]) for item in token_calls) if token_calls else None
+    output_units = sum(int(item["output_tokens"]) for item in token_calls) if token_calls else None
+    logical = logical_calls[-1] if logical_calls else {}
+    provider_id = logical.get("provider")
+    model_id = logical.get("resolved_model") or logical.get("requested_model")
+    return StructuredProviderCallResult(
+        structured_json=structured_json,
+        provider_id=str(provider_id) if provider_id else None,
+        model_id=str(model_id) if model_id else None,
+        configuration_digest=configuration_digest,
+        input_units=input_units,
+        output_units=output_units,
+        duration_ms=duration_ms,
+    )

core/engine/core/structured_reasoning_provider.py

@@ -0,0 +1,124 @@
+"""Adapter from ACE's selected LLM route to the governed reasoning port."""
+
+from __future__ import annotations
+
+import json
+
+from ace.core.contracts import canonical_json
+from ace.core.reasoning import (
+    ProviderExecutionRequestV1Alpha1,
+    ProviderRouteV1Alpha1,
+    ProviderStructuredOutputV1Alpha1,
+    ProviderUsageV1Alpha1,
+)
+from ace.core.runtime_use import CapabilityArtifactIdentityV1Alpha1
+from core.engine.core.provider_runtime import complete_structured_provider_call
+
+
+class SelectedLLMReasoningProviderError(RuntimeError):
+    """The selected LLM could not satisfy the exact governed provider port."""
+
+
+class SelectedLLMReasoningProvider:
+    """Use the already selected LLM once, without introducing another route."""
+
+    def __init__(
+        self,
+        *,
+        provider: object,
+        artifact_identity: CapabilityArtifactIdentityV1Alpha1,
+        configuration_digest: str,
+        model: str | None = None,
+        model_version: str | None = None,
+        max_tokens: int = 4096,
+    ) -> None:
+        exact = CapabilityArtifactIdentityV1Alpha1.model_validate(artifact_identity.model_dump(mode="python"))
+        if exact.capability != "structured_reasoning" or exact.contract != "ace.core.reasoning-provider/v1alpha1":
+            raise SelectedLLMReasoningProviderError("artifact does not implement the governed reasoning port")
+        if not configuration_digest.startswith("sha256:") or len(configuration_digest) != 71:
+            raise SelectedLLMReasoningProviderError("configuration digest must use exact sha256 syntax")
+        if max_tokens < 1:
+            raise SelectedLLMReasoningProviderError("max_tokens must be positive")
+        self.provider = provider
+        self._artifact_identity = exact
+        self.configuration_digest = configuration_digest
+        self.model = model
+        self.model_version = model_version
+        self.max_tokens = max_tokens
+
+    @property
+    def artifact_identity(self) -> CapabilityArtifactIdentityV1Alpha1:
+        return self._artifact_identity
+
+    @staticmethod
+    def _prompt(request: ProviderExecutionRequestV1Alpha1) -> str:
+        return canonical_json(
+            {
+                "context_items": [
+                    {
+                        "content": json.loads(item.content_json),
+                        "context_id": item.context_id,
+                        "material_digest": item.material_digest,
+                    }
+                    for item in request.context_items
+                ],
+                "required_output_envelope": {
+                    "referenced_context_ids": [item.context_id for item in request.context_items],
+                    "structured_result": "object matching trusted_instructions.output_contract",
+                },
+                "trusted_instructions": json.loads(request.instruction_json),
+            }
+        )
+
+    async def execute(self, request: ProviderExecutionRequestV1Alpha1) -> ProviderStructuredOutputV1Alpha1:
+        try:
+            exact = ProviderExecutionRequestV1Alpha1.model_validate(request.model_dump(mode="python"))
+            call = await complete_structured_provider_call(
+                self.provider,
+                prompt=self._prompt(exact),
+                model=self.model,
+                max_tokens=self.max_tokens,
+                configuration_digest=self.configuration_digest,
+            )
+            material = json.loads(call.structured_json)
+        except Exception as exc:
+            raise SelectedLLMReasoningProviderError("selected structured provider call failed") from exc
+        if set(material) != {"referenced_context_ids", "structured_result"}:
+            raise SelectedLLMReasoningProviderError("provider output omitted the exact governed result envelope")
+        expected_context = tuple(sorted(item.context_id for item in exact.context_items))
+        referenced = material["referenced_context_ids"]
+        structured = material["structured_result"]
+        if not isinstance(referenced, list) or not all(isinstance(item, str) for item in referenced):
+            raise SelectedLLMReasoningProviderError("provider output did not reference every exact context item")
+        if (
+            tuple(sorted(referenced)) != expected_context
+            or len(referenced) != len(set(referenced))
+            or not isinstance(structured, dict)
+        ):
+            raise SelectedLLMReasoningProviderError("provider output did not reference every exact context item")
+        if call.unavailable_fields:
+            raise SelectedLLMReasoningProviderError(
+                "selected provider lacks required governed telemetry: " + ", ".join(call.unavailable_fields)
+            )
+        model_version = self.model_version or call.model_id
+        if model_version is None:
+            raise SelectedLLMReasoningProviderError("selected provider lacks a model version")
+        return ProviderStructuredOutputV1Alpha1(
+            route=ProviderRouteV1Alpha1(
+                provider_id=str(call.provider_id),
+                model_id=str(call.model_id),
+                model_version=model_version,
+                configuration_digest=str(call.configuration_digest),
+            ),
+            usage=ProviderUsageV1Alpha1(
+                input_units=int(call.input_units),
+                output_units=int(call.output_units),
+                total_units=int(call.input_units) + int(call.output_units),
+                duration_ms=call.duration_ms,
+            ),
+            structured_json=canonical_json(structured),
+            referenced_context_ids=tuple(referenced),
+        )
+
+
+__all__ = ["SelectedLLMReasoningProvider", "SelectedLLMReasoningProviderError"]

core/engine/core/tokens.py

@@ -118,6 +118,18 @@ def token_call_count(self) -> int:
         with self._lock:
             return len(self._calls)
 
+    def calls_snapshot(self) -> tuple[dict, ...]:
+        """Return copied per-call usage facts for bounded adapter accounting."""
+
+        with self._lock:
+            return tuple(dict(call) for call in self._calls)
+
+    def llm_calls_snapshot(self) -> tuple[dict, ...]:
+        """Return copied logical-call provenance without exposing mutable state."""
+
+        with self._lock:
+            return tuple(dict(call) for call in self._llm_calls)
+
     def total_output(self) -> int:
         with self._lock:
             return sum(c["output_tokens"] for c in self._calls)

tests/test_public_core_boundaries.py

@@ -116,6 +116,7 @@ def test_host_adapters_are_the_only_core_engine_edge_into_public_ace() -> None:
         "core/engine/core/agent_composition_runtime.py",
         "core/engine/core/agent_composition_lifecycle_runtime.py",
         "core/engine/core/external_operations.py",
+        "core/engine/core/structured_reasoning_provider.py",
     }
     offenders = sorted(
         str(path.relative_to(REPO))