From d6f09b80884ee92db35f1b32e475799d815523e4 Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Thu, 13 Aug 2026 17:30:53 -0700 Subject: [PATCH] Adapt selected LLM to governed reasoning --- core/engine/core/provider_runtime.py | 90 ++++++++++ .../core/structured_reasoning_provider.py | 124 ++++++++++++++ core/engine/core/tokens.py | 12 ++ tests/test_public_core_boundaries.py | 1 + tests/test_structured_reasoning_provider.py | 154 ++++++++++++++++++ 5 files changed, 381 insertions(+) create mode 100644 core/engine/core/structured_reasoning_provider.py create mode 100644 tests/test_structured_reasoning_provider.py diff --git a/core/engine/core/provider_runtime.py b/core/engine/core/provider_runtime.py index e4a5422..fd47dd7 100644 --- a/core/engine/core/provider_runtime.py +++ b/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, + ) diff --git a/core/engine/core/structured_reasoning_provider.py b/core/engine/core/structured_reasoning_provider.py new file mode 100644 index 0000000..da7e047 --- /dev/null +++ b/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"] diff --git a/core/engine/core/tokens.py b/core/engine/core/tokens.py index 7a2b658..28b2986 100644 --- a/core/engine/core/tokens.py +++ b/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) diff --git a/tests/test_public_core_boundaries.py b/tests/test_public_core_boundaries.py index 94b089b..3af4e82 100644 --- a/tests/test_public_core_boundaries.py +++ b/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)) diff --git a/tests/test_structured_reasoning_provider.py b/tests/test_structured_reasoning_provider.py new file mode 100644 index 0000000..c2d89f7 --- /dev/null +++ b/tests/test_structured_reasoning_provider.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from ace.core.contracts import canonical_json +from ace.core.reasoning import FrozenContextItemV1Alpha1, ProviderExecutionRequestV1Alpha1 +from ace.core.runtime_use import CapabilityArtifactIdentityV1Alpha1 +from core.engine.core.provider_runtime import complete_structured_provider_call +from core.engine.core.structured_reasoning_provider import ( + SelectedLLMReasoningProvider, + SelectedLLMReasoningProviderError, +) +from core.engine.core.tokens import get_accumulator + +pytestmark = pytest.mark.unit + +NOW = datetime(2026, 8, 14, 12, tzinfo=UTC) +CONFIGURATION_DIGEST = "sha256:" + "c" * 64 +ARTIFACT = CapabilityArtifactIdentityV1Alpha1( + capability="structured_reasoning", + contract="ace.core.reasoning-provider/v1alpha1", + implementation_id="selected_llm_adapter", + implementation_version="1.0.0", + artifact_digest="sha256:" + "a" * 64, +) + + +def _request() -> ProviderExecutionRequestV1Alpha1: + context = FrozenContextItemV1Alpha1( + product_id="product:personal", + record_space="prepared_intelligence", + record_kind="observation", + record_key="observation:one", + storage_id="immutable_record:one", + material_digest="sha256:" + "b" * 64, + payload_contract="ace.intelligence.observation/v1alpha1", + as_of=NOW, + available_at=NOW, + content_json=canonical_json({"status": "changed"}), + ) + return ProviderExecutionRequestV1Alpha1( + product_id="product:personal", + request_id="reasoning_request:one", + request_digest="sha256:" + "d" * 64, + attempt_key="reasoning_attempt:one", + instruction_json=canonical_json({"output_contract": "fixture.brief/v1"}), + context_items=(context,), + cutoff_at=NOW, + started_at=NOW, + ) + + +class _MeasuredProvider: + async def complete_json(self, prompt, *, model, max_tokens): + assert "trusted_instructions" in prompt + assert max_tokens == 2048 + accumulator = get_accumulator() + assert accumulator is not None + accumulator.record( + "complete_json", + input_tokens=17, + output_tokens=9, + provider="fixture_provider", + model=model, + ) + accumulator.record_llm_call( + { + "provider": "fixture_provider", + "requested_model": model, + "resolved_model": "fixture-model-2026-08", + "wall_ms": 3, + "status": "completed", + } + ) + context_id = _request().context_items[0].context_id + return { + "referenced_context_ids": [context_id], + "structured_result": {"summary": "The status changed."}, + } + + +class _UnknownUsageProvider: + async def complete_json(self, _prompt, *, model, max_tokens): + return { + "referenced_context_ids": [_request().context_items[0].context_id], + "structured_result": {"summary": "Unknown usage."}, + } + + +class _FailingProvider: + async def complete_json(self, _prompt, *, model, max_tokens): + raise RuntimeError("secret provider detail") + + +@pytest.mark.asyncio +async def test_selected_provider_adapts_actual_route_usage_and_context() -> None: + provider = SelectedLLMReasoningProvider( + provider=_MeasuredProvider(), + artifact_identity=ARTIFACT, + configuration_digest=CONFIGURATION_DIGEST, + model="fixture-requested-model", + model_version="2026-08", + max_tokens=2048, + ) + result = await provider.execute(_request()) + + assert provider.artifact_identity == ARTIFACT + assert result.route.provider_id == "fixture_provider" + assert result.route.model_id == "fixture-model-2026-08" + assert result.route.model_version == "2026-08" + assert result.route.configuration_digest == CONFIGURATION_DIGEST + assert result.usage.input_units == 17 + assert result.usage.output_units == 9 + assert result.usage.total_units == 26 + assert result.usage.duration_ms >= 0 + assert result.structured_json == canonical_json({"summary": "The status changed."}) + assert result.referenced_context_ids == (_request().context_items[0].context_id,) + + +@pytest.mark.asyncio +async def test_unknown_usage_is_explicit_and_governed_adapter_fails_closed() -> None: + call = await complete_structured_provider_call( + _UnknownUsageProvider(), + prompt="{}", + model="fixture-model", + configuration_digest=CONFIGURATION_DIGEST, + ) + assert set(call.unavailable_fields) == {"provider_id", "model_id", "input_units", "output_units"} + assert call.input_units is None + assert call.output_units is None + + provider = SelectedLLMReasoningProvider( + provider=_UnknownUsageProvider(), + artifact_identity=ARTIFACT, + configuration_digest=CONFIGURATION_DIGEST, + model="fixture-model", + ) + with pytest.raises(SelectedLLMReasoningProviderError, match="lacks required governed telemetry"): + await provider.execute(_request()) + + +@pytest.mark.asyncio +async def test_provider_failure_is_sanitized_and_never_becomes_output() -> None: + provider = SelectedLLMReasoningProvider( + provider=_FailingProvider(), + artifact_identity=ARTIFACT, + configuration_digest=CONFIGURATION_DIGEST, + model="fixture-model", + ) + with pytest.raises(SelectedLLMReasoningProviderError, match="selected structured provider call failed") as exc: + await provider.execute(_request()) + assert "secret provider detail" not in str(exc.value)