From 1f597fc0a25c77ad9500f682aa530a0f44f14141 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 13:38:17 +0200 Subject: [PATCH 1/5] Implement Provider operation admission --- .../scenarios/operation/admission/action.yml | 44 ++ .../scenarios/operation/admission/verify.py | 401 +++++++++++ .../workflows/provider-runtime-scenarios.yml | 34 + README.md | 4 + docs/ci-automation.md | 9 +- docs/manifest-authoring.md | 11 +- docs/operation-admission.md | 126 ++++ docs/reference/declarations.md | 6 + docs/reference/operations.md | 13 + docs/reference/provider.md | 13 + docs/runtime-architecture.md | 8 + mkdocs.yml | 2 + pyproject.toml | 2 + src/stateforge_provider/__init__.py | 2 + .../_runtime/configuration.py | 8 + .../capability_assessment/evaluation.py | 92 +++ .../capability_assessment/service.py | 106 +-- .../_services/manifest/declaration.py | 159 ++++- .../_services/operation_admission/__init__.py | 3 + .../operation_admission/declaration.py | 62 ++ .../_services/operation_admission/errors.py | 57 ++ .../_services/operation_admission/runtime.py | 210 ++++++ .../_services/operation_admission/service.py | 478 +++++++++++++ src/stateforge_provider/operations.py | 69 ++ src/stateforge_provider/provider.py | 71 +- tests/test_ci_automation.py | 46 +- tests/test_operation_admission.py | 640 ++++++++++++++++++ uv.lock | 42 ++ 28 files changed, 2594 insertions(+), 124 deletions(-) create mode 100644 .github/actions/scenarios/operation/admission/action.yml create mode 100644 .github/actions/scenarios/operation/admission/verify.py create mode 100644 docs/operation-admission.md create mode 100644 docs/reference/operations.md create mode 100644 src/stateforge_provider/_services/capability_assessment/evaluation.py create mode 100644 src/stateforge_provider/_services/operation_admission/__init__.py create mode 100644 src/stateforge_provider/_services/operation_admission/declaration.py create mode 100644 src/stateforge_provider/_services/operation_admission/errors.py create mode 100644 src/stateforge_provider/_services/operation_admission/runtime.py create mode 100644 src/stateforge_provider/_services/operation_admission/service.py create mode 100644 src/stateforge_provider/operations.py create mode 100644 tests/test_operation_admission.py diff --git a/.github/actions/scenarios/operation/admission/action.yml b/.github/actions/scenarios/operation/admission/action.yml new file mode 100644 index 0000000..76dd41e --- /dev/null +++ b/.github/actions/scenarios/operation/admission/action.yml @@ -0,0 +1,44 @@ +name: Test Operation Admission with a direct Protocol peer +description: Verify the installed Provider wheel over its real Unix socket. + +inputs: + provider-artifact-directory: + description: Directory containing the exact Provider build. + required: true + protocol-artifact-directory: + description: Directory containing the exact Protocol build. + required: true + runtime-directory: + description: Scenario-owned directory under the runner temp directory. + required: true + evidence-directory: + description: Directory receiving scenario evidence. + required: true + +runs: + using: composite + steps: + - name: Install exact Provider and Protocol wheels + id: environment + uses: ./.github/actions/environment/install + with: + provider-artifact-directory: ${{ inputs.provider-artifact-directory }} + protocol-artifact-directory: ${{ inputs.protocol-artifact-directory }} + environment-directory: ${{ inputs.runtime-directory }}/venv + evidence-directory: ${{ inputs.evidence-directory }}/installation + + - name: Exercise Operation Admission over the installed endpoint + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }} + ISOLATED_PYTHON: ${{ steps.environment.outputs.python }} + RUNTIME_DIRECTORY: ${{ inputs.runtime-directory }}/p + run: | + set -euo pipefail + mkdir -p "${EVIDENCE_DIRECTORY}" + cd "${RUNNER_TEMP}" + unset PYTHONPATH PYTHONHOME + "${ISOLATED_PYTHON}" -I "${ACTION_PATH}/verify.py" \ + --runtime-directory "${RUNTIME_DIRECTORY}" \ + --output "${EVIDENCE_DIRECTORY}/operation-admission.json" diff --git a/.github/actions/scenarios/operation/admission/verify.py b/.github/actions/scenarios/operation/admission/verify.py new file mode 100644 index 0000000..dd50757 --- /dev/null +++ b/.github/actions/scenarios/operation/admission/verify.py @@ -0,0 +1,401 @@ +"""Verify installed R5 behavior through an independent Protocol peer.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +from datetime import UTC, datetime, timedelta +from importlib.metadata import version +from pathlib import Path +from types import MappingProxyType +from uuid import uuid4 + +import grpc +from google.protobuf.json_format import ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from grpc_status import rpc_status +from pydantic import Field +from stateforge_protocol.sfp.bootstrap.v1 import ( + bootstrap_pb2, + bootstrap_pb2_grpc, +) +from stateforge_protocol.sfp.manifest.v1 import ( + manifest_pb2, + manifest_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, + operation_admission_pb2_grpc, + validation, +) + +from stateforge_provider import ( + OperationContext, + Platform, + Provider, + ProviderModel, + RequirementCheckResult, + operation, + requirement_check, +) + + +class Input(ProviderModel): + """One typed request with a secret-bearing field.""" + + value: int + secret: str = Field(json_schema_extra={"writeOnly": True}) + + +class Result(ProviderModel): + """One typed operation result.""" + + accepted: bool + + +class AdmissionProvider(Provider): + """Expose one fenced operation through only the public author API.""" + + identity = "stateforge.test.operation_admission" + display_name = "StateForge Operation Admission CI Provider" + version = "0.1.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__(self, runtime_directory: Path) -> None: + """Bind the fixture to scenario-owned runtime directories.""" + super().__init__( + discovery_directory=runtime_directory / "discovery", + state_directory=runtime_directory / "state", + ) + self.requirement_satisfied = True + self.dispatches = 0 + self.started = asyncio.Event() + self.release = asyncio.Event() + self.context: OperationContext | None = None + + @requirement_check("runtime.ready", "Runtime is ready") + async def ready(self) -> RequirementCheckResult: + """Report the deterministic runtime prerequisite.""" + if self.requirement_satisfied: + return RequirementCheckResult.satisfied() + return RequirementCheckResult.unsatisfied("Runtime is not ready") + + @operation( + "stateforge.test/execute@1", + requires=("runtime.ready",), + implements=("stateforge.test.v1.audited",), + ) + async def execute( + self, + request: Input, + context: OperationContext, + ) -> Result: + """Retain dispatch so concurrent retries converge visibly.""" + self.dispatches += 1 + self.context = context + self.started.set() + await self.release.wait() + return Result(accepted=request.value > 0) + + +def _timestamp(value: datetime) -> Timestamp: + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +def _request( + handshake: bootstrap_pb2.HandshakeResponse, + manifest: manifest_pb2.GetManifestResponse, + *, + operation_id: str, + value: int = 1, + secret: str | None = None, + generation: int | None = None, + roles: tuple[str, ...] = ("stateforge.test.v1.audited",), +) -> operation_admission_pb2.StartOperationRequest: + payload = Struct() + ParseDict( + {"value": value, "secret": secret or "not-evidence"}, + payload, + ) + execution_context = Struct() + ParseDict( + {"authority": "ci-policy", "nested": {"items": [1, 2]}}, + execution_context, + ) + return operation_admission_pb2.StartOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=( + generation + if generation is not None + else handshake.endpoint_generation + ), + expected_session_id=handshake.session_id, + ), + invocation=operation_admission_pb2.OperationInvocation( + operation_id=operation_id, + operation_identifier="stateforge.test/execute@1", + payload=payload, + execution_deadline=_timestamp( + datetime.now(UTC) + timedelta(minutes=5) + ), + contract=operation_admission_pb2.OperationContractContext( + manifest_revision=manifest.manifest_revision, + required_implementation_roles=roles, + ), + execution_context=execution_context, + ), + admission_mode=operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE, + ) + + +async def _descriptor( + directory: Path, + provider_task: asyncio.Task[None], +) -> dict[str, object]: + async with asyncio.timeout(5): + while True: + paths = await asyncio.to_thread( + lambda: tuple(directory.glob("*.sfp.json")) + ) + if paths: + return json.loads(await asyncio.to_thread(paths[0].read_text)) + if provider_task.done(): + await provider_task + await asyncio.sleep(0.01) + + +async def _session( + channel: grpc.aio.Channel, + descriptor: dict[str, object], +) -> tuple[ + bootstrap_pb2.HandshakeResponse, + manifest_pb2.GetManifestResponse, +]: + handshake = await bootstrap_pb2_grpc.BootstrapServiceStub( + channel + ).Handshake( + bootstrap_pb2.HandshakeRequest( + agent_runtime=bootstrap_pb2.AgentRuntime( + runtime_id=str(uuid4()), + software_version="0.1.0", + ), + expected_provider_instance_id=descriptor["provider_instance_id"], + expected_endpoint_generation=descriptor["endpoint_generation"], + supported_sfp_versions=[ + bootstrap_pb2.ProtocolVersion(major=1, minor=0) + ], + ), + timeout=3, + ) + manifest = await manifest_pb2_grpc.ManifestServiceStub( + channel + ).GetManifest( + manifest_pb2.GetManifestRequest( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + supported_manifest_schema_versions=[1], + ), + timeout=3, + ) + return handshake, manifest + + +async def _expect_error( + call: grpc.aio.UnaryUnaryMultiCallable[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + request: operation_admission_pb2.StartOperationRequest, + code: grpc.StatusCode, + reason: int, +) -> str: + try: + await call(request, timeout=3) + except grpc.aio.AioRpcError as error: + if error.code() is not code: + raise RuntimeError( + f"expected {code.name}, got {error.code().name}" + ) from error + status = rpc_status.from_call(error) + if status is None or not status.details: + raise RuntimeError( + "Operation Admission error omitted typed details" + ) from error + detail = operation_admission_pb2.OperationAdmissionErrorDetail() + if not status.details[0].Unpack(detail) or detail.reason != reason: + raise RuntimeError( + "Operation Admission error detail mismatch" + ) from error + return error.code().name + raise RuntimeError( + f"Operation Admission unexpectedly accepted {code.name}" + ) + + +async def _run( # noqa: PLR0915 + runtime_directory: Path, +) -> dict[str, object]: + discovery = runtime_directory / "discovery" + discovery.mkdir(parents=True) + provider = AdmissionProvider(runtime_directory) + provider_task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + try: + descriptor = await _descriptor(discovery, provider_task) + transport = descriptor["transport"] + if not isinstance(transport, dict): + raise RuntimeError("Provider descriptor transport is invalid") + socket_path = str(transport["socket_path"]) + channel = grpc.aio.insecure_channel(f"unix:{socket_path}") + await asyncio.wait_for(channel.channel_ready(), timeout=3) + handshake, manifest = await _session(channel, descriptor) + admission = operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ).StartOperation + + stale_status = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=str(uuid4()), + generation=handshake.endpoint_generation + 1, + ), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_PRECONDITION_FAILED, + ) + role_status = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=str(uuid4()), + roles=("stateforge.test.v1.missing",), + ), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_INCOMPATIBLE, + ) + provider.requirement_satisfied = False + requirement_status = await _expect_error( + admission, + _request(handshake, manifest, operation_id=str(uuid4())), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_PRECONDITION_FAILED, + ) + provider.requirement_satisfied = True + + operation_id = str(uuid4()) + request = _request(handshake, manifest, operation_id=operation_id) + responses = await asyncio.gather( + *(admission(request, timeout=3) for _ in range(20)) + ) + await asyncio.wait_for(provider.started.wait(), timeout=2) + if provider.dispatches != 1: + raise RuntimeError( + "identical concurrent requests dispatched twice" + ) + accepted_new = sum( + response.disposition + == operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW + for response in responses + ) + if accepted_new != 1: + raise RuntimeError("atomic admission did not select one winner") + for response in responses: + validation.validate_admission_exchange(request, response) + + payload_marker = "must-not-appear-in-evidence" + conflict_status = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=operation_id, + value=2, + secret=payload_marker, + ), + grpc.StatusCode.ALREADY_EXISTS, + operation_admission_pb2.ADMISSION_ERROR_REASON_IDEMPOTENCY_CONFLICT, + ) + provider.release.set() + + context = provider.context + if context is None or not isinstance( + context.execution_context, MappingProxyType + ): + raise RuntimeError("Provider author context is not immutable") + + provider.request_shutdown() + await asyncio.wait_for(provider_task, timeout=5) + if await asyncio.to_thread(Path(socket_path).exists): + raise RuntimeError("Provider socket survived shutdown") + remaining_descriptors = await asyncio.to_thread( + lambda: tuple(discovery.glob("*.sfp.json")) + ) + if remaining_descriptors: + raise RuntimeError("Provider descriptor survived shutdown") + + evidence = { + "installed_artifacts": { + "stateforge-provider": version("stateforge-provider"), + "stateforge-protocol": version("stateforge-protocol"), + }, + "provider_instance_id": handshake.provider_instance_id, + "endpoint_generation": handshake.endpoint_generation, + "session_id": handshake.session_id, + "manifest_revision": manifest.manifest_revision, + "operation_id": operation_id, + "request_fingerprint": responses[0].record.request_fingerprint, + "concurrent_requests": len(responses), + "accepted_new": accepted_new, + "resolved_existing": len(responses) - accepted_new, + "author_dispatches": provider.dispatches, + "stale_fence_status": stale_status, + "missing_role_status": role_status, + "requirement_status": requirement_status, + "idempotency_conflict_status": conflict_status, + "immutable_context": True, + "final_process_state": "stopped", + "descriptor_removed": True, + "socket_removed": True, + } + serialized = json.dumps(evidence, sort_keys=True) + if payload_marker in serialized: + raise RuntimeError("secret leaked into scenario evidence") + return evidence + finally: + provider.request_shutdown() + if not provider_task.done(): + await asyncio.wait_for(provider_task, timeout=5) + if channel is not None: + await channel.close() + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--runtime-directory", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> None: + """Run the focused installed-wheel scenario and write its evidence.""" + arguments = _arguments() + arguments.output.write_text( + json.dumps( + asyncio.run(_run(arguments.runtime_directory)), + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/provider-runtime-scenarios.yml b/.github/workflows/provider-runtime-scenarios.yml index 55cac3e..4a4ff8b 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -131,6 +131,40 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + operation-admission: + name: Operation Admission / direct Protocol peer + if: ${{ inputs.scenario-tier == 'full' }} + runs-on: ubuntu-24.04 + timeout-minutes: 20 + steps: + - name: Check out scenario automation + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Download exact runtime bundle + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + name: ${{ inputs.artifact-name }} + path: ${{ runner.temp }}/stateforge-provider-bundle + + - name: Run Operation Admission scenario + uses: ./.github/actions/scenarios/operation/admission + with: + provider-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/provider + protocol-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/protocol + runtime-directory: ${{ runner.temp }}/sfp-r5 + evidence-directory: evidence/runtime/operation-admission + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-operation-admission + path: evidence/runtime/operation-admission + if-no-files-found: warn + retention-days: ${{ inputs.retention-days }} + trigger-publication-agent: name: Trigger Publication / installed Agent RPM if: ${{ inputs.scenario-tier == 'full' }} diff --git a/README.md b/README.md index 88564cc..1f287bf 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,10 @@ Typed triggers can be published from a managed `@trigger` async source or an existing callback through `publish_trigger()`. The runtime owns P3 streaming, bounded flow control, acknowledgement receipts, retry, and reconnect replay. See the [trigger publication guide](docs/trigger-publication.md). +Operation handlers receive typed requests and an optional immutable +`OperationContext`; the runtime owns fences, requirement revalidation, +idempotency, record-before-dispatch, and bounded shutdown. See the +[operation admission guide](docs/operation-admission.md). Install the Provider framework from PyPI: diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 1f97dab..d7f4b20 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -151,12 +151,17 @@ The runtime suite currently covers: including callback and decorated sources, a bounded receive window, retryable rejection with immutable replay, accepted duplicate disposition, terminal rejection, stale-fence rejection, stream loss, and replay under a - fresh fenced session; and + fresh fenced session; +- installed-wheel Operation Admission through an independent Protocol peer, + including exact fences, role and requirement rejection, 20 concurrent + identical requests converging on one dispatch, typed idempotency conflict, + immutable author context, secret-safe evidence, and clean endpoint removal; + and - end-to-end Trigger Publication through the exact installed Agent RPM and exact Provider and Protocol wheels, including durable `accepted_new`, a lost acknowledgement, Agent crash and restart, immutable replay with `accepted_duplicate`, callback publication, retry across journal write -failure with an explicitly correlated retryable rejection, + failure with an explicitly correlated retryable rejection, conflicting-content rejection, a newer Provider endpoint generation, rejection of publication from the superseded session, and clean shutdown. diff --git a/docs/manifest-authoring.md b/docs/manifest-authoring.md index 17bfe4c..dcddb03 100644 --- a/docs/manifest-authoring.md +++ b/docs/manifest-authoring.md @@ -109,8 +109,9 @@ projection; result/trigger schemas use its serialization projection. The runtime adds the explicit Draft 2020-12 dialect and validates the object schema. It never serializes a configured model instance into the manifest. -Decorated methods are async and currently have `(self, request)` plus typed -input and return annotations. Metadata discovery does not call them. Methods +Decorated methods are async and have `(self, request)` or +`(self, request, context: OperationContext)` plus typed input and return +annotations. Metadata discovery does not call them. Methods are inherited using Python's MRO; overriding a declared method requires a new `@operation` declaration. Duplicate operation IDs or stacked decorators are errors. Unresolvable annotations fail startup; keep model classes importable @@ -274,6 +275,6 @@ Install the package and its development dependencies from `pyproject.toml`: python -m pip install -e . --group dev --group docs ``` -This milestone does not implement configuration loading, operation dispatch, -OperationContext, Agent routing, or vendor work. Those remain their own -milestones; these declarations are their shared input. +The runtime uses accepted operation declarations as the input to +[Operation Admission](operation-admission.md). Agent routing and vendor work +remain outside manifest construction. diff --git a/docs/operation-admission.md b/docs/operation-admission.md new file mode 100644 index 0000000..8a615dd --- /dev/null +++ b/docs/operation-admission.md @@ -0,0 +1,126 @@ +# Operation admission + +An operation declaration becomes executable when the Provider starts. The +runtime exposes SFP Operation Admission on the same local endpoint as +Bootstrap, Manifest, Health, Capability Assessment, and Trigger Publication. +Provider authors receive typed Python values; they do not handle protobuf, +gRPC, session fences, fingerprints, or retry decisions. + +## Author an operation + +```python +from stateforge_provider import ( + OperationContext, + Platform, + Provider, + ProviderModel, + RequirementCheckResult, + operation, + requirement_check, +) + + +class RestartInput(ProviderModel): + unit: str + + +class RestartResult(ProviderModel): + restarted: bool + + +class ServiceProvider(Provider): + identity = "example.systemd" + display_name = "Example systemd Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + @requirement_check( + "runtime.systemd.control", + "The Provider can control systemd services", + ) + async def can_control_systemd(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + @operation( + "example.systemd/restart@1", + requires=("runtime.systemd.control",), + implements=("example.service.v1.restart",), + ) + async def restart( + self, + request: RestartInput, + context: OperationContext, + ) -> RestartResult: + return RestartResult(restarted=True) +``` + +The context parameter may be omitted for handlers that need only the typed +request. When present, it must be annotated as `OperationContext`. + +`OperationContext` contains: + +| Field | Meaning | +| --- | --- | +| `operation_id` | Immutable UUID assigned to this logical operation | +| `execution_deadline` | Absolute deadline carried by the admitted request | +| `execution_context` | Recursively read-only JSON metadata approved by the Agent | + +The context contains no raw session, transport, or authorization object. A +handler cannot mutate it to change the admitted request. + +## What happens before dispatch + +The runtime validates each new request in this order: + +1. canonical Protocol structure and message limits; +2. exact endpoint generation, session, and manifest revision; +3. a declared operation identifier; +4. required implementation-role membership; +5. JSON Schema and typed Pydantic payload validation; +6. an unexpired execution deadline; +7. current process-wide Health; and +8. fresh results from every requirement check attached to the operation. + +Only then does it atomically record admission and schedule the author handler. +Rejected requests create no admission record and never invoke author code. + +## Idempotency + +`operation_id` names one immutable logical request. The runtime calculates the +canonical Protocol fingerprint over its invocation content. + +| Later request | Result | +| --- | --- | +| Same ID and same content while the record is retained | Resolve the existing record; do not dispatch again | +| Same ID with changed immutable content | Reject with `ALREADY_EXISTS` and a typed idempotency-conflict detail | +| Unknown ID in resolve-only mode | Reject without creating a record | +| ID whose retained record expired | Return the canonical expired-record outcome while its tombstone remains | + +The record is inserted before the handler task is created. Concurrent +identical requests therefore converge on one accepted record and exactly one +author dispatch. + +Admission is not operation completion. R5 owns request acceptance and safe +dispatch; progress, terminal results, errors, and cancellation are reported by +the later operation-lifecycle procedure. + +## Shutdown and failures + +Shutdown stops new admissions, keeps retained records resolvable during the +transition, cancels active author tasks, and waits for them before endpoint +cleanup. An exception or invalid return value from author code stays local and +does not revise the immutable admission record. Payloads, execution context, +and secret-bearing fields are never copied into remote errors or CI evidence. + +Capacity and retention are bounded by Provider configuration: + +| Setting | Default | +| --- | ---: | +| `max_operation_admission_records` | 1,024 | +| `max_operation_dispatches` | 128 | +| `operation_record_retention_seconds` | 86,400 | +| `operation_tombstone_retention_seconds` | 86,400 | + +Callers must treat `RESOURCE_EXHAUSTED` and shutdown-related `UNAVAILABLE` as +retryable only under their own bounded policy. Retrying the same immutable +request is safe because admission resolves by ID and fingerprint. diff --git a/docs/reference/declarations.md b/docs/reference/declarations.md index bc64547..83c9681 100644 --- a/docs/reference/declarations.md +++ b/docs/reference/declarations.md @@ -12,3 +12,9 @@ Requirement-check methods must be asynchronous instance methods with exactly Startup rejects malformed or duplicate IDs, missing or repeated references, invalid signatures, unresolved annotations, and metadata outside canonical Protocol bounds. + +Operation methods are asynchronous instance methods with either +`(self, request)` or `(self, request, context)`. The request and return +annotations must be `ProviderModel` subclasses. The optional third parameter +must be annotated as the public `OperationContext`; it exposes only the +operation ID, execution deadline, and recursively immutable execution context. diff --git a/docs/reference/operations.md b/docs/reference/operations.md new file mode 100644 index 0000000..5e12beb --- /dev/null +++ b/docs/reference/operations.md @@ -0,0 +1,13 @@ +# Operation API + +::: stateforge_provider.operations.OperationContext + +::: stateforge_provider.operations.FrozenJsonValue + +Operation handlers are declared with +[`@operation`](declarations.md) and receive the Pydantic request model declared +by their signature. The optional `OperationContext` parameter is constructed +only after canonical admission succeeds and is recursively immutable. + +See [Operation admission](../operation-admission.md) for the complete handler, +validation, idempotency, capacity, and shutdown contract. diff --git a/docs/reference/provider.md b/docs/reference/provider.md index 5217fc8..7aa595b 100644 --- a/docs/reference/provider.md +++ b/docs/reference/provider.md @@ -20,3 +20,16 @@ It executes each distinct referenced check at most once per request, honors RPC cancellation and deadlines, then revalidates the session and readiness before returning a canonically validated response. Assessment is point-in-time evidence only: it neither executes nor authorizes an operation. + +## Operation Admission ownership + +`Provider` also registers the private P4 Operation Admission service. The +runtime validates the exact endpoint, session, accepted manifest, operation +declaration, payload schema, requested implementation roles, current Health, +execution deadline, and fresh requirement-check results. It atomically records +the canonical request fingerprint before scheduling author code, so concurrent +identical retries dispatch exactly once and changed content is rejected. + +Authors implement only the typed operation method. See +[Operation admission](../operation-admission.md) for the public handler and +idempotency contract. diff --git a/docs/runtime-architecture.md b/docs/runtime-architecture.md index 34f3a9a..74dc2b6 100644 --- a/docs/runtime-architecture.md +++ b/docs/runtime-architecture.md @@ -12,6 +12,7 @@ stateforge_provider/ capability_assessment/ # Private P2 declaration and service adaptation health/ # Standard gRPC Health adaptation and bounded Watch manifest/ # Manifest declaration, validation, and RPC service + operation_admission/ # Private P4 admission and dispatch runtime trigger_publication/ # Private P3 stream, queue, retry, and replay runtime ``` @@ -64,6 +65,13 @@ stream receive window, acknowledgement correlation, retry backoff, and replay after a fresh session. Publications remain process-local in this version; durability across Provider process restarts is not implied. +Operation Admission binds the operation declarations already accepted for the +manifest to their author methods. Its private P4 service validates canonical +requests and active runtime fences, re-evaluates referenced requirements, +owns the bounded idempotency registry, records admission before dispatch, and +passes typed values plus an immutable public `OperationContext` to author code. +It does not own later operation progress or terminal results. + To add a future procedure: 1. create one `_services/` package; diff --git a/mkdocs.yml b/mkdocs.yml index ee00cda..7330214 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -22,11 +22,13 @@ nav: - Manifest declarations: manifest-authoring.md - Health and readiness: health-reporting.md - Trigger publication: trigger-publication.md + - Operation admission: operation-admission.md - Package and runtime CI/CD: ci-automation.md - Python API: - Provider: reference/provider.md - Declarations: reference/declarations.md - Trigger publication: reference/triggers.md + - Operations: reference/operations.md - Protocol version: reference/version.md plugins: diff --git a/pyproject.toml b/pyproject.toml index 73a3c06..9b9514d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ classifiers = [ dependencies = [ "grpcio>=1.83", "grpcio-health-checking>=1.83", + "grpcio-status>=1.83", "jsonschema>=4.25,<5", "pydantic>=2.12,<3", "stateforge-protocol>=0.1.0,<0.2.0", @@ -52,6 +53,7 @@ dev = [ "ruff>=0.12", "types-grpcio", "types-grpcio-health-checking", + "types-grpcio-status", "types-jsonschema", "types-protobuf", ] diff --git a/src/stateforge_provider/__init__.py b/src/stateforge_provider/__init__.py index 6e47310..591681c 100644 --- a/src/stateforge_provider/__init__.py +++ b/src/stateforge_provider/__init__.py @@ -11,6 +11,7 @@ requirement_check, ) from stateforge_provider.health import HealthStatus +from stateforge_provider.operations import OperationContext from stateforge_provider.provider import Provider from stateforge_provider.requirements import ( RequirementCheckOutcome, @@ -33,6 +34,7 @@ __all__ = [ "HealthStatus", "ManifestError", + "OperationContext", "Platform", "ProtocolVersion", "Provider", diff --git a/src/stateforge_provider/_runtime/configuration.py b/src/stateforge_provider/_runtime/configuration.py index 31db358..6e1fd8d 100644 --- a/src/stateforge_provider/_runtime/configuration.py +++ b/src/stateforge_provider/_runtime/configuration.py @@ -23,6 +23,8 @@ class ProviderRuntimeLimits: health_watch_queue_capacity: int = 16 max_pending_trigger_publications: int = 1_024 max_trigger_in_flight_publications: int = 128 + max_operation_admission_records: int = 1_024 + max_operation_dispatches: int = 128 def __post_init__(self) -> None: """Reject unusable operational limits.""" @@ -34,6 +36,8 @@ def __post_init__(self) -> None: "health_watch_queue_capacity", "max_pending_trigger_publications", "max_trigger_in_flight_publications", + "max_operation_admission_records", + "max_operation_dispatches", ): value = getattr(self, name) if ( @@ -73,6 +77,8 @@ class ProviderRuntimeConfiguration: health_watch_delivery_timeout_seconds: float = 1.0 trigger_retry_base_seconds: float = 0.25 trigger_retry_max_seconds: float = 30.0 + operation_record_retention_seconds: float = 86_400.0 + operation_tombstone_retention_seconds: float = 86_400.0 limits: ProviderRuntimeLimits = field( default_factory=ProviderRuntimeLimits ) @@ -110,6 +116,8 @@ def __post_init__(self) -> None: "health_watch_delivery_timeout_seconds", "trigger_retry_base_seconds", "trigger_retry_max_seconds", + "operation_record_retention_seconds", + "operation_tombstone_retention_seconds", ): timeout = getattr(self, name) if ( diff --git a/src/stateforge_provider/_services/capability_assessment/evaluation.py b/src/stateforge_provider/_services/capability_assessment/evaluation.py new file mode 100644 index 0000000..38c2ed9 --- /dev/null +++ b/src/stateforge_provider/_services/capability_assessment/evaluation.py @@ -0,0 +1,92 @@ +"""Evaluate Provider-owned runtime requirements through one shared path.""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass + +from stateforge_provider._services.capability_assessment.declaration import ( + BoundRequirementRegistry, + RequirementDeclaration, +) +from stateforge_provider.requirements import RequirementCheckResult + +_LOGGER = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class EvaluatedRequirement: + """Pair validated declaration metadata with one author result.""" + + declaration: RequirementDeclaration + result: RequirementCheckResult + + +class RequirementEvaluationError(RuntimeError): + """Hide author check failures from remote protocol boundaries.""" + + +def required_checks( + requirements: BoundRequirementRegistry, + operation_identifiers: tuple[str, ...], +) -> tuple[str, ...]: + """Collect unique checks in stable first-reference order.""" + identifiers: list[str] = [] + seen: set[str] = set() + for operation in operation_identifiers: + for identifier in requirements.operation_requirements[operation]: + if identifier not in seen: + seen.add(identifier) + identifiers.append(identifier) + return tuple(identifiers) + + +async def evaluate_requirements( + requirements: BoundRequirementRegistry, + identifiers: tuple[str, ...], +) -> dict[str, EvaluatedRequirement]: + """Run each distinct author check once and cancel peers on failure.""" + tasks = { + identifier: asyncio.create_task( + _invoke_check(requirements, identifier), + name=f"stateforge-requirement-{identifier}", + ) + for identifier in identifiers + } + try: + values = await asyncio.gather(*tasks.values()) + except BaseException: + for task in tasks.values(): + task.cancel() + await asyncio.gather(*tasks.values(), return_exceptions=True) + raise + return dict(zip(tasks, values, strict=True)) + + +async def _invoke_check( + requirements: BoundRequirementRegistry, + identifier: str, +) -> EvaluatedRequirement: + """Normalize one author check without leaking its failure remotely.""" + check = requirements.checks[identifier] + try: + result: object = await check() + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.exception( + "Provider requirement check failed", + extra={"requirement_id": identifier}, + ) + raise RequirementEvaluationError from None + if not isinstance(result, RequirementCheckResult): + _LOGGER.error( + "Provider requirement check returned an invalid result", + extra={"requirement_id": identifier}, + ) + raise RequirementEvaluationError + return EvaluatedRequirement( + requirements.declarations[identifier], + result, + ) diff --git a/src/stateforge_provider/_services/capability_assessment/service.py b/src/stateforge_provider/_services/capability_assessment/service.py index 1b8a211..3371e4b 100644 --- a/src/stateforge_provider/_services/capability_assessment/service.py +++ b/src/stateforge_provider/_services/capability_assessment/service.py @@ -4,7 +4,6 @@ import asyncio import logging -from dataclasses import dataclass from uuid import uuid4 import grpc @@ -21,14 +20,16 @@ from stateforge_provider._runtime.sessions import Session, SessionRegistry from stateforge_provider._services.capability_assessment.declaration import ( BoundRequirementRegistry, - RequirementDeclaration, +) +from stateforge_provider._services.capability_assessment.evaluation import ( + EvaluatedRequirement, + RequirementEvaluationError, + evaluate_requirements, + required_checks, ) from stateforge_provider._services.health.service import ProviderHealthService from stateforge_provider._services.manifest.declaration import ManifestSnapshot -from stateforge_provider.requirements import ( - RequirementCheckOutcome, - RequirementCheckResult, -) +from stateforge_provider.requirements import RequirementCheckOutcome from stateforge_provider.version import ProtocolVersion _LOGGER = logging.getLogger(__name__) @@ -41,18 +42,6 @@ _UNKNOWN_EXPLANATION = "Runtime requirement evidence is unavailable" -@dataclass(frozen=True, slots=True) -class _EvaluatedRequirement: - """Pair validated declaration metadata with one author result.""" - - declaration: RequirementDeclaration - result: RequirementCheckResult - - -class _CheckError(Exception): - """Hide author failure details from the remote RPC boundary.""" - - class CapabilityAssessmentService( capability_assessment_pb2_grpc.CapabilityAssessmentServiceServicer ): @@ -105,9 +94,15 @@ async def AssessCapabilities( # noqa: N802, PLR0912 ) try: - requirement_ids = self._required_checks(request) - results = await self._execute_checks(requirement_ids) - except _CheckError: + requirement_ids = required_checks( + self._requirements, + tuple(request.operation_identifiers), + ) + results = await evaluate_requirements( + self._requirements, + requirement_ids, + ) + except RequirementEvaluationError: await context.abort( grpc.StatusCode.INTERNAL, "Provider requirement check failed", @@ -214,71 +209,10 @@ async def _validate_fences( ) return session - def _required_checks( - self, - request: capability_assessment_pb2.AssessCapabilitiesRequest, - ) -> tuple[str, ...]: - """Collect unique checks in stable first-reference order.""" - identifiers: list[str] = [] - seen: set[str] = set() - for operation in request.operation_identifiers: - for identifier in self._requirements.operation_requirements[ - operation - ]: - if identifier not in seen: - seen.add(identifier) - identifiers.append(identifier) - return tuple(identifiers) - - async def _execute_checks( - self, - identifiers: tuple[str, ...], - ) -> dict[str, _EvaluatedRequirement]: - """Run each distinct author check once and cancel peers on failure.""" - tasks = { - identifier: asyncio.create_task(self._invoke_check(identifier)) - for identifier in identifiers - } - try: - values = await asyncio.gather(*tasks.values()) - except BaseException: - for task in tasks.values(): - task.cancel() - await asyncio.gather(*tasks.values(), return_exceptions=True) - raise - return dict(zip(tasks, values, strict=True)) - - async def _invoke_check( - self, - identifier: str, - ) -> _EvaluatedRequirement: - """Normalize one author check without leaking its failure remotely.""" - check = self._requirements.checks[identifier] - try: - result: object = await check() - except asyncio.CancelledError: - raise - except Exception: - _LOGGER.exception( - "Provider requirement check failed", - extra={"requirement_id": identifier}, - ) - raise _CheckError from None - if not isinstance(result, RequirementCheckResult): - _LOGGER.error( - "Provider requirement check returned an invalid result", - extra={"requirement_id": identifier}, - ) - raise _CheckError - return _EvaluatedRequirement( - self._requirements.declarations[identifier], - result, - ) - def _response( self, request: capability_assessment_pb2.AssessCapabilitiesRequest, - results: dict[str, _EvaluatedRequirement], + results: dict[str, EvaluatedRequirement], ) -> capability_assessment_pb2.AssessCapabilitiesResponse: """Build one deterministic request-correlated response.""" timestamp = Timestamp() @@ -300,7 +234,7 @@ def _response( def _finding( self, operation: str, - results: dict[str, _EvaluatedRequirement], + results: dict[str, EvaluatedRequirement], ) -> capability_assessment_pb2.CapabilityFinding: """Aggregate one operation using the locked precedence rules.""" evaluated = tuple( @@ -352,7 +286,7 @@ def _finding( def _requirement_finding( - evaluated: _EvaluatedRequirement, + evaluated: EvaluatedRequirement, ) -> capability_assessment_pb2.RequirementFinding: """Adapt one typed author result to a bounded P2 finding.""" state = { @@ -379,7 +313,7 @@ def _requirement_finding( def _explanation( - evaluated: tuple[_EvaluatedRequirement, ...], + evaluated: tuple[EvaluatedRequirement, ...], outcome: RequirementCheckOutcome, ) -> str: """Select a bounded nonempty explanation for a normalized reason.""" diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index 0d6e8d4..c2e9d56 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -40,19 +40,39 @@ _mro_namespace, _Operation, ) +from stateforge_provider.operations import OperationContext from stateforge_provider.triggers import _TriggerSource +@dataclass(frozen=True, slots=True) +class OperationDeclaration: + """Retain one operation's accepted author and manifest contract.""" + + identifier: str + method_name: str + request_model: type[ProviderModel] + result_model: type[ProviderModel] + implements: tuple[str, ...] + requires: tuple[str, ...] + accepts_context: bool + + +def _provider_model(model: object, path: str) -> type[ProviderModel]: + """Return one author model with a type understood by static analysis.""" + if not isinstance(model, type) or not issubclass(model, ProviderModel): + raise ManifestError(f"{path}: must be a ProviderModel subclass") + return model + + def _model_schema( model: object, path: str, mode: JsonSchemaMode, limits: ProviderRuntimeLimits, ) -> dict[str, Any]: - if not isinstance(model, type) or not issubclass(model, ProviderModel): - raise ManifestError(f"{path}: must be a ProviderModel subclass") + resolved_model = _provider_model(model, path) try: - schema = model.model_json_schema(mode=mode) + schema = resolved_model.model_json_schema(mode=mode) except ( PydanticUndefinedAnnotation, PydanticUserError, @@ -75,7 +95,10 @@ def _sequence(owner: type, name: str, limit: int) -> tuple[Any, ...]: # Keep discovery and its rejection rules adjacent. -def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: +def _operations( # noqa: PLR0912 + owner: type, + limits: ProviderRuntimeLimits, +) -> tuple[dict[str, Any], tuple[OperationDeclaration, ...]]: methods, metadata_erased = _effective_methods( owner, metadata_attribute="__stateforge_operation__", @@ -83,16 +106,19 @@ def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: ) if metadata_erased: raise ManifestError("operations: overrides must redeclare metadata") - declared_methods: list[tuple[Any, type, _Operation]] = [] - for method, declaring_class in methods.values(): + declared_methods: list[tuple[str, Any, type, _Operation]] = [] + for method_name, (method, declaring_class) in methods.items(): metadata = getattr(method, "__stateforge_operation__", None) if isinstance(metadata, _Operation): - declared_methods.append((method, declaring_class, metadata)) + declared_methods.append( + (method_name, method, declaring_class, metadata) + ) if len(declared_methods) > manifest_contract.MAX_OPERATIONS: raise ManifestError("operations: entry limit exceeded") declarations: dict[str, Any] = {} - for method, declaring_class, metadata in declared_methods: + runtime_declarations: list[OperationDeclaration] = [] + for method_name, method, declaring_class, metadata in declared_methods: path = f"operations[{len(declarations)}]" validate_structure(metadata.identifier, "identifier", path) if metadata.identifier in declarations: @@ -100,8 +126,14 @@ def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: if not inspect.iscoroutinefunction(method): raise ManifestError(f"{path}: operation method must be async") parameters = tuple(inspect.signature(method).parameters) - if parameters != ("self", "request"): - raise ManifestError(f"{path}: expected (self, request) signature") + if parameters not in ( + ("self", "request"), + ("self", "request", "context"), + ): + raise ManifestError( + f"{path}: expected (self, request) or " + "(self, request, context) signature" + ) try: annotations = get_type_hints( method, @@ -111,6 +143,22 @@ def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: raise ManifestError( f"{path}: unresolved model annotations" ) from None + request_model = _provider_model( + annotations.get("request"), f"{path}.input" + ) + result_model = _provider_model( + annotations.get("return"), f"{path}.result" + ) + accepts_context = len(parameters) == len( + ("self", "request", "context") + ) + if ( + accepts_context + and annotations.get("context") is not OperationContext + ): + raise ManifestError( + f"{path}: context must be annotated as OperationContext" + ) roles = metadata.implements if ( not isinstance(roles, tuple) @@ -121,22 +169,37 @@ def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: validate_structure(role, "dottedIdentifier", f"{path}.implements") if len(set(roles)) != len(roles): raise ManifestError(f"{path}.implements: duplicate role") + input_schema = _model_schema( + request_model, + f"{path}.input", + "validation", + limits, + ) + result_schema = _model_schema( + result_model, + f"{path}.result", + "serialization", + limits, + ) declarations[metadata.identifier] = { - "inputSchema": _model_schema( - annotations.get("request"), - f"{path}.input", - "validation", - limits, - ), - "resultSchema": _model_schema( - annotations.get("return"), - f"{path}.result", - "serialization", - limits, - ), + "inputSchema": input_schema, + "resultSchema": result_schema, "implements": sorted(roles), } - return declarations + runtime_declarations.append( + OperationDeclaration( + identifier=metadata.identifier, + method_name=method_name, + request_model=request_model, + result_model=result_model, + implements=tuple(sorted(roles)), + requires=metadata.requires, + accepts_context=accepts_context, + ) + ) + return declarations, tuple( + sorted(runtime_declarations, key=lambda item: item.identifier) + ) def _triggers( # noqa: PLR0912 @@ -278,6 +341,21 @@ class ManifestDeclaration: document: bytes trigger_declarations: tuple[Trigger, ...] = () trigger_source_methods: tuple[tuple[str, str], ...] = () + operation_declarations: tuple[OperationDeclaration, ...] = () + + @property + def operation_schemas( + self, + ) -> dict[str, tuple[dict[str, Any], dict[str, Any]]]: + """Return fresh accepted input/result schemas keyed by operation.""" + document = json.loads(self.document) + return { + identifier: ( + declaration["inputSchema"], + declaration["resultSchema"], + ) + for identifier, declaration in document["operations"].items() + } @property def trigger_schemas(self) -> dict[str, dict[str, Any]]: @@ -368,7 +446,7 @@ def _discover( owner, limits, ) - operations = _operations(owner, limits) + operations, operation_declarations = _operations(owner, limits) configuration = _model_schema( getattr(owner, "configuration_model", ProviderModel), "configuration_schema", @@ -397,7 +475,12 @@ def _discover( document, limits=contract_limits, ) - return cls(content, trigger_declarations, trigger_sources) + return cls( + document=content, + trigger_declarations=trigger_declarations, + trigger_source_methods=trigger_sources, + operation_declarations=operation_declarations, + ) def snapshot(self, identity: ProviderEndpointIdentity) -> ManifestSnapshot: """Construct the exchange after identity allocation and startup.""" @@ -437,9 +520,10 @@ def _snapshot( validate_manifest_response(response) response.ClearField("session_id") return ManifestSnapshot( - response.SerializeToString(), - self.trigger_declarations, - self.trigger_source_methods, + document=response.SerializeToString(), + trigger_declarations=self.trigger_declarations, + trigger_source_methods=self.trigger_source_methods, + operation_declarations=self.operation_declarations, ) @@ -450,6 +534,7 @@ class ManifestSnapshot: document: bytes trigger_declarations: tuple[Trigger, ...] = () trigger_source_methods: tuple[tuple[str, str], ...] = () + operation_declarations: tuple[OperationDeclaration, ...] = () @property def manifest_revision(self) -> str: @@ -464,6 +549,24 @@ def operation_identifiers(self) -> frozenset[str]: response = manifest_pb2.GetManifestResponse.FromString(self.document) return frozenset(response.manifest.operations) + @property + def operation_schemas( + self, + ) -> dict[str, tuple[dict[str, Any], dict[str, Any]]]: + """Return fresh input/result schemas from the accepted manifest.""" + response = manifest_pb2.GetManifestResponse.FromString(self.document) + projected = MessageToDict( + response.manifest, + always_print_fields_with_no_presence=True, + ) + return { + identifier: ( + declaration["inputSchema"], + declaration["resultSchema"], + ) + for identifier, declaration in projected["operations"].items() + } + @property def trigger_identifiers(self) -> frozenset[str]: """Return the triggers declared by this immutable snapshot.""" diff --git a/src/stateforge_provider/_services/operation_admission/__init__.py b/src/stateforge_provider/_services/operation_admission/__init__.py new file mode 100644 index 0000000..c81eb10 --- /dev/null +++ b/src/stateforge_provider/_services/operation_admission/__init__.py @@ -0,0 +1,3 @@ +"""Implement Provider-side SFP Operation Admission.""" + +"""Implement Provider-side SFP Operation Admission.""" diff --git a/src/stateforge_provider/_services/operation_admission/declaration.py b/src/stateforge_provider/_services/operation_admission/declaration.py new file mode 100644 index 0000000..2d22901 --- /dev/null +++ b/src/stateforge_provider/_services/operation_admission/declaration.py @@ -0,0 +1,62 @@ +"""Bind accepted operation declarations to one Provider instance.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any + +from jsonschema import Draft202012Validator + +from stateforge_provider._services.manifest.declaration import ( + OperationDeclaration, +) +from stateforge_provider.declarations import ProviderModel + +_BoundHandler = Callable[..., Awaitable[ProviderModel]] + + +@dataclass(frozen=True, slots=True) +class BoundOperation: + """Pair one accepted declaration with its bound author handler.""" + + declaration: OperationDeclaration + handler: _BoundHandler + input_validator: Draft202012Validator + result_validator: Draft202012Validator + + +@dataclass(frozen=True, slots=True) +class BoundOperationRegistry: + """Resolve exact manifest-declared operations for one Provider instance.""" + + operations: dict[str, BoundOperation] + + @classmethod + def bind( + cls, + provider: object, + declarations: tuple[OperationDeclaration, ...], + schemas: dict[str, tuple[dict[str, Any], dict[str, Any]]], + ) -> BoundOperationRegistry: + """Bind declarations to the methods used to create the manifest.""" + operations: dict[str, BoundOperation] = {} + for declaration in declarations: + handler = getattr(provider, declaration.method_name, None) + if not callable(handler): + raise TypeError( + "Accepted operation handler is unavailable: " + f"{declaration.identifier}" + ) + input_schema, result_schema = schemas[declaration.identifier] + operations[declaration.identifier] = BoundOperation( + declaration=declaration, + handler=handler, + input_validator=Draft202012Validator(input_schema), + result_validator=Draft202012Validator(result_schema), + ) + return cls(operations) + + def get(self, identifier: str) -> BoundOperation | None: + """Return one exact declared operation without normalization.""" + return self.operations.get(identifier) diff --git a/src/stateforge_provider/_services/operation_admission/errors.py b/src/stateforge_provider/_services/operation_admission/errors.py new file mode 100644 index 0000000..f56f1fc --- /dev/null +++ b/src/stateforge_provider/_services/operation_admission/errors.py @@ -0,0 +1,57 @@ +"""Translate Operation Admission decisions into canonical gRPC status.""" + +from __future__ import annotations + +from dataclasses import dataclass +from uuid import UUID + +import grpc +from google.protobuf.any_pb2 import Any +from google.rpc import status_pb2 # type: ignore[import-untyped] +from grpc_status import rpc_status +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, + validation, +) + +_UUID_VERSION = 4 + + +@dataclass(frozen=True, slots=True) +class AdmissionFailureError(Exception): + """Carry one sanitized typed Operation Admission rejection.""" + + code: grpc.StatusCode + reason: operation_admission_pb2.AdmissionErrorReason.ValueType + explanation: str + operation_id: str = "" + retryable: bool = False + + def status(self) -> grpc.Status: + """Build a validated rich status with one versioned detail.""" + detail = operation_admission_pb2.OperationAdmissionErrorDetail( + operation_id=_valid_operation_id(self.operation_id), + reason=self.reason, + retryable=self.retryable, + explanation=self.explanation, + ) + validation.validate_admission_error_detail(detail) + packed = Any() + packed.Pack(detail) + status = status_pb2.Status( + code=self.code.value[0], + message=self.explanation, + details=(packed,), + ) + return rpc_status.to_status(status) + + +def _valid_operation_id(value: str) -> str: + """Return only a canonical UUIDv4 suitable for an error detail.""" + try: + identifier = UUID(value) + except (ValueError, AttributeError): + return "" + if identifier.version != _UUID_VERSION or str(identifier) != value: + return "" + return value diff --git a/src/stateforge_provider/_services/operation_admission/runtime.py b/src/stateforge_provider/_services/operation_admission/runtime.py new file mode 100644 index 0000000..2e36d50 --- /dev/null +++ b/src/stateforge_provider/_services/operation_admission/runtime.py @@ -0,0 +1,210 @@ +"""Own atomic in-memory Operation Admission records and dispatch tasks.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from enum import Enum, auto + +from google.protobuf.timestamp_pb2 import Timestamp +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, +) + +_LOGGER = logging.getLogger(__name__) +_Dispatch = Callable[[], Awaitable[None]] + + +class AdmissionRegistryFailure(Enum): + """Classify an exact atomic-registry rejection.""" + + CONFLICT = auto() + EXPIRED = auto() + UNAVAILABLE = auto() + CAPACITY = auto() + STOPPING = auto() + + +@dataclass(frozen=True, slots=True) +class AdmissionRegistryError(Exception): + """Report a registry decision without carrying business payloads.""" + + failure: AdmissionRegistryFailure + + +@dataclass(frozen=True, slots=True) +class AdmissionDecision: + """Return one immutable record and whether this call inserted it.""" + + record: operation_admission_pb2.OperationAdmissionRecord + accepted_new: bool + + +@dataclass(slots=True) +class _Entry: + """Retain one immutable record until its advertised expiry.""" + + fingerprint: str + record: operation_admission_pb2.OperationAdmissionRecord + expires_at: datetime + + +class OperationAdmissionRuntime: + """Converge concurrent retries before invoking Provider author code. + + Every mutation is synchronous and runs on the Provider event loop. No + await occurs between the final lookup, record insertion, and dispatch-task + creation. + """ + + def __init__( + self, + *, + maximum_records: int, + maximum_dispatches: int, + record_retention_seconds: float, + tombstone_retention_seconds: float, + ) -> None: + self._maximum_records = maximum_records + self._maximum_dispatches = maximum_dispatches + self._record_retention = timedelta(seconds=record_retention_seconds) + self._tombstone_retention = timedelta( + seconds=tombstone_retention_seconds + ) + self._records: dict[str, _Entry] = {} + self._tombstones: dict[str, datetime] = {} + self._tasks: dict[str, asyncio.Task[None]] = {} + self._closed = False + + def lookup( + self, + operation_id: str, + fingerprint: str, + ) -> AdmissionDecision | None: + """Resolve retained state before new-admission preconditions.""" + now = datetime.now(UTC) + self._prune(now) + entry = self._records.get(operation_id) + if entry is not None: + if entry.fingerprint != fingerprint: + raise AdmissionRegistryError(AdmissionRegistryFailure.CONFLICT) + return AdmissionDecision(_copy_record(entry.record), False) + if operation_id in self._tombstones: + raise AdmissionRegistryError(AdmissionRegistryFailure.EXPIRED) + return None + + def admit( + self, + *, + operation_id: str, + fingerprint: str, + endpoint_generation: int, + dispatch: _Dispatch, + ) -> AdmissionDecision: + """Atomically insert a record and schedule exactly one dispatch.""" + existing = self.lookup(operation_id, fingerprint) + if existing is not None: + return existing + if self._closed: + raise AdmissionRegistryError(AdmissionRegistryFailure.STOPPING) + if ( + len(self._records) >= self._maximum_records + or len(self._tasks) >= self._maximum_dispatches + ): + raise AdmissionRegistryError(AdmissionRegistryFailure.CAPACITY) + + admitted_at = datetime.now(UTC) + expires_at = admitted_at + self._record_retention + record = operation_admission_pb2.OperationAdmissionRecord( + operation_id=operation_id, + request_fingerprint=fingerprint, + admitted_at=_timestamp(admitted_at), + record_expires_at=_timestamp(expires_at), + admitted_endpoint_generation=endpoint_generation, + ) + self._records[operation_id] = _Entry( + fingerprint=fingerprint, + record=_copy_record(record), + expires_at=expires_at, + ) + task = asyncio.create_task( + self._run_dispatch(operation_id, dispatch), + name=f"stateforge-operation-{operation_id}", + ) + self._tasks[operation_id] = task + + def task_done(completed: asyncio.Task[None]) -> None: + self._task_done(operation_id, completed) + + task.add_done_callback(task_done) + return AdmissionDecision(record, True) + + def begin_shutdown(self) -> None: + """Reject new admissions while allowing retained resolution.""" + self._closed = True + + async def close(self) -> None: + """Stop admission and cancel every active author dispatch.""" + self.begin_shutdown() + tasks = tuple(self._tasks.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + async def _run_dispatch( + self, + operation_id: str, + dispatch: _Dispatch, + ) -> None: + """Run author behavior without changing immutable admission state.""" + try: + await dispatch() + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.error( + "Provider operation dispatch failed", + extra={"operation_id": operation_id}, + ) + + def _task_done( + self, + operation_id: str, + task: asyncio.Task[None], + ) -> None: + """Release dispatch capacity without consuming task exceptions.""" + if self._tasks.get(operation_id) is task: + del self._tasks[operation_id] + + def _prune(self, now: datetime) -> None: + """Replace expired records with bounded expiry evidence.""" + expired = tuple( + identifier + for identifier, entry in self._records.items() + if entry.expires_at <= now + ) + for identifier in expired: + del self._records[identifier] + self._tombstones[identifier] = now + self._tombstone_retention + for identifier, expires_at in tuple(self._tombstones.items()): + if expires_at <= now: + del self._tombstones[identifier] + + +def _timestamp(value: datetime) -> Timestamp: + """Create an independent protobuf timestamp.""" + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +def _copy_record( + record: operation_admission_pb2.OperationAdmissionRecord, +) -> operation_admission_pb2.OperationAdmissionRecord: + """Return a message callers cannot use to mutate retained state.""" + return operation_admission_pb2.OperationAdmissionRecord.FromString( + record.SerializeToString() + ) diff --git a/src/stateforge_provider/_services/operation_admission/service.py b/src/stateforge_provider/_services/operation_admission/service.py new file mode 100644 index 0000000..4f8308a --- /dev/null +++ b/src/stateforge_provider/_services/operation_admission/service.py @@ -0,0 +1,478 @@ +"""Serve typed Provider operation admission over canonical SFP P4.""" + +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime +from typing import NoReturn, cast +from uuid import UUID + +import grpc +from google.protobuf.json_format import MessageToDict +from pydantic import JsonValue, ValidationError +from stateforge_protocol.sfp.operation_admission.v1 import ( + contract, + operation_admission_pb2, + operation_admission_pb2_grpc, + validation, +) + +from stateforge_provider._runtime.grpc import ProcedureRegistration +from stateforge_provider._runtime.identity import ProviderEndpointIdentity +from stateforge_provider._runtime.sessions import Session, SessionRegistry +from stateforge_provider._services.capability_assessment.declaration import ( + BoundRequirementRegistry, +) +from stateforge_provider._services.capability_assessment.evaluation import ( + RequirementEvaluationError, + evaluate_requirements, + required_checks, +) +from stateforge_provider._services.health.service import ProviderHealthService +from stateforge_provider._services.manifest.declaration import ManifestSnapshot +from stateforge_provider._services.operation_admission.declaration import ( + BoundOperation, + BoundOperationRegistry, +) +from stateforge_provider._services.operation_admission.errors import ( + AdmissionFailureError, +) +from stateforge_provider._services.operation_admission.runtime import ( + AdmissionRegistryError, + AdmissionRegistryFailure, + OperationAdmissionRuntime, +) +from stateforge_provider.operations import OperationContext +from stateforge_provider.requirements import RequirementCheckOutcome +from stateforge_provider.version import ProtocolVersion + +_LOGGER = logging.getLogger(__name__) +_RESOLVED_EXISTING = ( + operation_admission_pb2.ADMISSION_DISPOSITION_RESOLVED_EXISTING +) + + +class OperationAdmissionService( + operation_admission_pb2_grpc.OperationAdmissionServiceServicer +): + """Validate immutable invocations before exactly-once dispatch.""" + + def __init__( + self, + *, + endpoint: ProviderEndpointIdentity, + snapshot: ManifestSnapshot, + sessions: SessionRegistry, + health: ProviderHealthService, + requirements: BoundRequirementRegistry, + operations: BoundOperationRegistry, + runtime: OperationAdmissionRuntime, + ) -> None: + self._endpoint = endpoint + self._manifest_revision = snapshot.manifest_revision + self._sessions = sessions + self._health = health + self._requirements = requirements + self._operations = operations + self._runtime = runtime + + async def StartOperation( # noqa: N802, PLR0912 + self, + request: operation_admission_pb2.StartOperationRequest, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + ) -> operation_admission_pb2.StartOperationResponse: + """Resolve or atomically admit one immutable operation invocation.""" + operation_id = request.invocation.operation_id + try: + validation.validate_start_operation_request(request) + except validation.OperationAdmissionValidationError: + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, + "Invalid Operation Admission request", + operation_id, + ), + ) + + invocation = request.invocation + operation = self._operations.get(invocation.operation_identifier) + if operation is None: + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.UNIMPLEMENTED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_NOT_DECLARED, + "Operation is not declared by this Provider", + operation_id, + ), + ) + session = await self._validate_fences(request, context) + try: + self._validate_contract(invocation, operation) + payload = self._validate_payload(invocation, operation) + except AdmissionFailureError as failure: + await self._reject(context, failure) + fingerprint = validation.compute_request_fingerprint(invocation) + + try: + existing = self._runtime.lookup(operation_id, fingerprint) + except AdmissionRegistryError as error: + await self._reject_registry(context, operation_id, error) + if existing is not None: + return await self._response( + request, + existing.record, + False, + context, + ) + if ( + request.admission_mode + == operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY + ): + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + "Operation admission record is unavailable", + operation_id, + ), + ) + + execution_deadline = invocation.execution_deadline.ToDatetime( + tzinfo=UTC + ) + if execution_deadline <= datetime.now(UTC): + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.DEADLINE_EXCEEDED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_DEADLINE_EXPIRED, + "Operation execution deadline has elapsed", + operation_id, + ), + ) + if not self._health.is_serving(): + await self._reject_precondition( + context, operation_id, "Provider is not currently serving" + ) + + try: + identifiers = required_checks( + self._requirements, + (invocation.operation_identifier,), + ) + evaluated = await evaluate_requirements( + self._requirements, identifiers + ) + except RequirementEvaluationError: + await context.abort( + grpc.StatusCode.INTERNAL, + "Provider requirement evaluation failed", + ) + if execution_deadline <= datetime.now(UTC): + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.DEADLINE_EXCEEDED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_DEADLINE_EXPIRED, + "Operation execution deadline elapsed during admission", + operation_id, + ), + ) + incompatible = any( + item.result.outcome is RequirementCheckOutcome.INCOMPATIBLE + for item in evaluated.values() + ) + if incompatible: + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_INCOMPATIBLE, + "Operation is incompatible with the current runtime", + operation_id, + ), + ) + if any( + item.result.outcome is not RequirementCheckOutcome.SATISFIED + for item in evaluated.values() + ): + await self._reject_precondition( + context, + operation_id, + "Operation runtime requirements are not satisfied", + ) + if ( + self._sessions.get(request.session_binding.expected_session_id) + is not session + or not self._health.is_serving() + ): + await self._reject_precondition( + context, + operation_id, + "Operation Admission prerequisites changed", + ) + + dispatch = self._dispatch(operation, invocation, payload) + try: + decision = self._runtime.admit( + operation_id=operation_id, + fingerprint=fingerprint, + endpoint_generation=self._endpoint.endpoint_generation, + dispatch=dispatch, + ) + except AdmissionRegistryError as error: + await self._reject_registry(context, operation_id, error) + return await self._response( + request, + decision.record, + decision.accepted_new, + context, + ) + + async def _validate_fences( + self, + request: operation_admission_pb2.StartOperationRequest, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + ) -> Session: + """Require the exact current endpoint, session, and manifest.""" + binding = request.session_binding + session = self._sessions.get(binding.expected_session_id) + if session is None or ( + binding.expected_provider_instance_id + != str(self._endpoint.provider_instance_id) + or binding.expected_endpoint_generation + != self._endpoint.endpoint_generation + or session.endpoint != self._endpoint + or session.version != ProtocolVersion(1, 0) + or request.invocation.contract.manifest_revision + != self._manifest_revision + ): + await self._reject_precondition( + context, + request.invocation.operation_id, + "Operation Admission session or manifest is stale", + ) + return session + + def _validate_contract( + self, + invocation: operation_admission_pb2.OperationInvocation, + operation: BoundOperation, + ) -> None: + """Validate requested profile claims against the accepted manifest.""" + if any( + role not in operation.declaration.implements + for role in invocation.contract.required_implementation_roles + ): + raise AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_INCOMPATIBLE, + "Operation requires an undeclared implementation role", + invocation.operation_id, + ) + + def _validate_payload( + self, + invocation: operation_admission_pb2.OperationInvocation, + operation: BoundOperation, + ) -> object: + """Validate and construct one typed author request without defaults.""" + payload = cast(dict[str, JsonValue], MessageToDict(invocation.payload)) + if not operation.input_validator.is_valid(payload): + raise AdmissionFailureError( + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, + "Operation payload does not match its declared schema", + invocation.operation_id, + ) + try: + return operation.declaration.request_model.model_validate(payload) + except ValidationError: + raise AdmissionFailureError( + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, + "Operation payload is invalid", + invocation.operation_id, + ) from None + + def _dispatch( + self, + operation: BoundOperation, + invocation: operation_admission_pb2.OperationInvocation, + payload: object, + ) -> Callable[[], Awaitable[None]]: + """Capture typed immutable inputs for post-record dispatch.""" + operation_context = OperationContext._create( + operation_id=UUID(invocation.operation_id), + execution_deadline=invocation.execution_deadline.ToDatetime( + tzinfo=UTC + ), + execution_context=cast( + dict[str, JsonValue], + MessageToDict(invocation.execution_context), + ), + ) + + async def dispatch() -> None: + if operation.declaration.accepts_context: + result = await operation.handler(payload, operation_context) + else: + result = await operation.handler(payload) + if not isinstance(result, operation.declaration.result_model): + raise TypeError("operation returned an invalid result type") + serialized = result.model_dump(mode="json") + if not operation.result_validator.is_valid(serialized): + raise ValueError("operation result violates its schema") + + return dispatch + + async def _response( + self, + request: operation_admission_pb2.StartOperationRequest, + record: operation_admission_pb2.OperationAdmissionRecord, + accepted_new: bool, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + ) -> operation_admission_pb2.StartOperationResponse: + """Build and validate one request-correlated admission response.""" + response = operation_admission_pb2.StartOperationResponse( + provider_instance_id=str(self._endpoint.provider_instance_id), + endpoint_generation=self._endpoint.endpoint_generation, + session_id=request.session_binding.expected_session_id, + manifest_revision=self._manifest_revision, + disposition=( + operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW + if accepted_new + else _RESOLVED_EXISTING + ), + record=record, + ) + try: + validation.validate_admission_exchange(request, response) + except validation.OperationAdmissionValidationError: + _LOGGER.error("Provider produced an invalid admission response") + await context.abort( + grpc.StatusCode.INTERNAL, + "Provider produced an invalid Operation Admission response", + ) + return response + + async def _reject_precondition( + self, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + operation_id: str, + explanation: str, + ) -> NoReturn: + await self._reject( + context, + AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_PRECONDITION_FAILED, + explanation, + operation_id, + ), + ) + + async def _reject_registry( + self, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + operation_id: str, + error: AdmissionRegistryError, + ) -> NoReturn: + failure = { + AdmissionRegistryFailure.CONFLICT: AdmissionFailureError( + grpc.StatusCode.ALREADY_EXISTS, + operation_admission_pb2.ADMISSION_ERROR_REASON_IDEMPOTENCY_CONFLICT, + "Operation ID is bound to different immutable content", + operation_id, + ), + AdmissionRegistryFailure.EXPIRED: AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_EXPIRED, + "Operation admission record has expired", + operation_id, + ), + AdmissionRegistryFailure.UNAVAILABLE: AdmissionFailureError( + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + "Operation admission record is unavailable", + operation_id, + ), + AdmissionRegistryFailure.CAPACITY: AdmissionFailureError( + grpc.StatusCode.RESOURCE_EXHAUSTED, + operation_admission_pb2.ADMISSION_ERROR_REASON_PROVIDER_CAPACITY_EXHAUSTED, + "Provider admission capacity is exhausted", + operation_id, + True, + ), + AdmissionRegistryFailure.STOPPING: AdmissionFailureError( + grpc.StatusCode.UNAVAILABLE, + operation_admission_pb2.ADMISSION_ERROR_REASON_PROVIDER_UNAVAILABLE, + "Provider operation runtime is stopping", + operation_id, + True, + ), + }[error.failure] + await self._reject(context, failure) + + async def _reject( + self, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + failure: AdmissionFailureError, + ) -> NoReturn: + await context.abort_with_status(failure.status()) + raise AssertionError("gRPC abort returned") + + +def registration( + *, + endpoint: ProviderEndpointIdentity, + snapshot: ManifestSnapshot, + sessions: SessionRegistry, + health: ProviderHealthService, + requirements: BoundRequirementRegistry, + operations: BoundOperationRegistry, + runtime: OperationAdmissionRuntime, +) -> ProcedureRegistration: + """Build the bounded P4 registration for one Provider generation.""" + service = OperationAdmissionService( + endpoint=endpoint, + snapshot=snapshot, + sessions=sessions, + health=health, + requirements=requirements, + operations=operations, + runtime=runtime, + ) + return ProcedureRegistration( + name=contract.SERVICE_NAME, + max_request_bytes=contract.MAX_REQUEST_BYTES, + max_response_bytes=contract.MAX_RESPONSE_BYTES, + install=lambda server: ( + operation_admission_pb2_grpc.add_OperationAdmissionServiceServicer_to_server( + service, server + ) + ), + ) diff --git a/src/stateforge_provider/operations.py b/src/stateforge_provider/operations.py new file mode 100644 index 0000000..1fe1316 --- /dev/null +++ b/src/stateforge_provider/operations.py @@ -0,0 +1,69 @@ +"""Expose typed author-facing operation execution context.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime +from types import MappingProxyType +from typing import TypeAlias +from uuid import UUID + +from pydantic import JsonValue + +FrozenJsonValue: TypeAlias = ( + str + | int + | float + | bool + | tuple["FrozenJsonValue", ...] + | Mapping[str, "FrozenJsonValue"] + | None +) + + +@dataclass(frozen=True, slots=True) +class OperationContext: + """Describe one admitted operation without exposing SFP internals. + + Attributes: + operation_id: Agent-assigned immutable operation identity. + execution_deadline: Absolute deadline for the business attempt. + execution_context: Read-only approved execution metadata. + """ + + operation_id: UUID + execution_deadline: datetime + execution_context: Mapping[str, FrozenJsonValue] + + @classmethod + def _create( + cls, + operation_id: UUID, + execution_deadline: datetime, + execution_context: dict[str, JsonValue], + ) -> OperationContext: + """Create a context with recursively immutable JSON values.""" + frozen = { + key: _freeze_json(value) + for key, value in execution_context.items() + } + return cls( + operation_id, + execution_deadline, + MappingProxyType(frozen), + ) + + +def _freeze_json(value: JsonValue) -> FrozenJsonValue: + """Freeze nested JSON containers before exposing them to author code.""" + if isinstance(value, dict): + return MappingProxyType( + {key: _freeze_json(item) for key, item in value.items()} + ) + if isinstance(value, list): + return tuple(_freeze_json(item) for item in value) + return value + + +__all__ = ["FrozenJsonValue", "OperationContext"] diff --git a/src/stateforge_provider/provider.py b/src/stateforge_provider/provider.py index 2b6654e..1251df4 100644 --- a/src/stateforge_provider/provider.py +++ b/src/stateforge_provider/provider.py @@ -32,6 +32,15 @@ from stateforge_provider._services.manifest.service import ( registration as manifest_registration, ) +from stateforge_provider._services.operation_admission.declaration import ( + BoundOperationRegistry, +) +from stateforge_provider._services.operation_admission.runtime import ( + OperationAdmissionRuntime, +) +from stateforge_provider._services.operation_admission.service import ( + registration as operation_admission_registration, +) from stateforge_provider._services.trigger_publication.runtime import ( TriggerPublicationRuntime, ) @@ -107,6 +116,7 @@ def __init__( self._health_service: ProviderHealthService | None = None self._health_refresh_lock = asyncio.Lock() self._trigger_runtime: TriggerPublicationRuntime | None = None + self._operation_runtime: OperationAdmissionRuntime | None = None async def on_startup(self) -> None: """Override to initialize Provider resources before publication.""" @@ -191,7 +201,7 @@ async def run_with_signals() -> None: asyncio.run(run_with_signals()) @final - async def serve(self) -> None: # noqa: PLR0915 + async def serve(self) -> None: # noqa: PLR0912, PLR0915 """Validate declarations and serve on the caller's event loop. The caller retains signal-handler ownership and may initiate shutdown @@ -211,6 +221,26 @@ async def serve(self) -> None: # noqa: PLR0915 ), ) bound_requirements = requirement_registry.bind(self) + bound_operations = BoundOperationRegistry.bind( + self, + declaration.operation_declarations, + declaration.operation_schemas, + ) + operation_runtime = OperationAdmissionRuntime( + maximum_records=( + self._configuration.limits.max_operation_admission_records + ), + maximum_dispatches=( + self._configuration.limits.max_operation_dispatches + ), + record_retention_seconds=( + self._configuration.operation_record_retention_seconds + ), + tombstone_retention_seconds=( + self._configuration.operation_tombstone_retention_seconds + ), + ) + self._operation_runtime = operation_runtime trigger_runtime = TriggerPublicationRuntime( declaration, queue_capacity=( @@ -257,6 +287,15 @@ def procedures( health=health_service, requirements=bound_requirements, ), + operation_admission_registration( + endpoint=identity, + snapshot=snapshot, + sessions=sessions, + health=health_service, + requirements=bound_requirements, + operations=bound_operations, + runtime=operation_runtime, + ), trigger_publication_registration( endpoint=identity, snapshot=snapshot, @@ -285,6 +324,7 @@ def health_registration( await self._shutdown_requested.wait() finally: health_service.begin_shutdown() + operation_runtime.begin_shutdown() trigger_failure: BaseException | None = None try: await trigger_runtime.close() @@ -294,6 +334,14 @@ def health_registration( await trigger_runtime.stop_sources() except BaseException as error: trigger_failure = error + operation_failure: BaseException | None = None + try: + async with asyncio.timeout( + self._configuration.shutdown_timeout_seconds + ): + await operation_runtime.close() + except BaseException as error: + operation_failure = error shutdown_called = False async def shutdown_author_resources() -> None: @@ -323,6 +371,7 @@ async def shutdown_author_resources() -> None: self._health_service = None self._trigger_runtime = None + self._operation_runtime = None if endpoint_failure is not None: if trigger_failure is not None: endpoint_failure.add_note( @@ -334,6 +383,11 @@ async def shutdown_author_resources() -> None: "Provider author-resource shutdown also failed: " f"{shutdown_failure}" ) + if operation_failure is not None: + endpoint_failure.add_note( + "Provider operation-runtime shutdown also failed: " + f"{operation_failure}" + ) raise endpoint_failure if trigger_failure is not None: if shutdown_failure is not None: @@ -341,7 +395,19 @@ async def shutdown_author_resources() -> None: "Provider author-resource shutdown also failed: " f"{shutdown_failure}" ) + if operation_failure is not None: + trigger_failure.add_note( + "Provider operation-runtime shutdown also failed: " + f"{operation_failure}" + ) raise trigger_failure + if operation_failure is not None: + if shutdown_failure is not None: + operation_failure.add_note( + "Provider author-resource shutdown also failed: " + f"{shutdown_failure}" + ) + raise operation_failure if shutdown_failure is not None: raise shutdown_failure @@ -351,6 +417,9 @@ def request_shutdown(self) -> None: health_service = self._health_service if health_service is not None: health_service.begin_shutdown() + operation_runtime = self._operation_runtime + if operation_runtime is not None: + operation_runtime.begin_shutdown() self._shutdown_requested.set() async def _refresh_health( diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 2baf7ae..9c898e8 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -8,7 +8,7 @@ SERVICES = ROOT / "src" / "stateforge_provider" / "_services" SHA_LENGTH = 40 SUPPORTED_PYTHON_COUNT = 3 -SCENARIO_COUNT = 10 +SCENARIO_COUNT = 11 DOWNSTREAM_BUNDLE_CONSUMER_COUNT = 3 DEPENDENCY_CHECKOUT_COUNT = 2 @@ -42,6 +42,7 @@ def test_action_library_has_focused_domain_boundaries() -> None: "scenarios/health/standard-health", "scenarios/lifecycle/concurrent-providers", "scenarios/lifecycle/same-instance-replacement", + "scenarios/operation/admission", "scenarios/resilience/crash-restart", "scenarios/smoke/normal-lifecycle", "scenarios/trigger/publication", @@ -224,7 +225,9 @@ def test_each_runtime_scenario_is_an_independent_action() -> None: assert len(scenarios) == SCENARIO_COUNT for scenario in scenarios: content = read(scenario) - if "scenarios/trigger/publication" in str(scenario): + if "scenarios/trigger/publication" in str( + scenario + ) or "scenarios/operation/admission" in str(scenario): assert "./.github/actions/environment/install" in content assert "./.github/actions/agent/" not in content assert '"${ISOLATED_PYTHON}" -I' in content @@ -260,6 +263,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "trigger-publication", "trigger-publication-agent", "capability-assessment", + "operation-admission", } for scenario in scenarios: assert scenario in workflow @@ -314,6 +318,44 @@ def test_trigger_publication_keeps_the_focused_protocol_peer() -> None: assert "Trigger Publication / direct Protocol peer" in workflow +def test_operation_admission_keeps_the_focused_protocol_peer() -> None: + """Prove R5 installed behavior without simulating a future Agent.""" + scenario = read( + ACTIONS / "scenarios" / "operation" / "admission" / "action.yml" + ) + verifier = read( + ACTIONS / "scenarios" / "operation" / "admission" / "verify.py" + ) + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + + assert "./.github/actions/environment/install" in scenario + assert "./.github/actions/agent/" not in scenario + assert '"${ISOLATED_PYTHON}" -I' in scenario + assert "stateforge_protocol" in verifier + assert "stateforge_agent" not in verifier + assert "@operation(" in verifier + assert "OperationContext" in verifier + assert "StartOperation" in verifier + for evidence_field in ( + "installed_artifacts", + "concurrent_requests", + "accepted_new", + "resolved_existing", + "author_dispatches", + "stale_fence_status", + "missing_role_status", + "requirement_status", + "idempotency_conflict_status", + "immutable_context", + "final_process_state", + "descriptor_removed", + "socket_removed", + ): + assert evidence_field in verifier + assert "Operation Admission / direct Protocol peer" in workflow + assert "${{ runner.temp }}/sfp-r5" in workflow + + def test_trigger_publication_uses_exact_installed_agent_runtime() -> None: """Verify R18 across the real RPM, wheels, socket, and journal.""" scenario = read( diff --git a/tests/test_operation_admission.py b/tests/test_operation_admission.py new file mode 100644 index 0000000..d6f61b2 --- /dev/null +++ b/tests/test_operation_admission.py @@ -0,0 +1,640 @@ +"""Real-socket evidence for Provider Operation Admission and idempotency.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import shutil +import tempfile +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import MappingProxyType +from uuid import uuid4 + +import grpc +import pytest +from google.protobuf.json_format import ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from grpc_status import rpc_status +from pydantic import Field +from stateforge_protocol.sfp.bootstrap.v1 import ( + bootstrap_pb2, + bootstrap_pb2_grpc, +) +from stateforge_protocol.sfp.manifest.v1 import ( + manifest_pb2, + manifest_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + contract as admission_contract, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, + operation_admission_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + validation as admission_validation, +) + +from stateforge_provider import ( + HealthStatus, + OperationContext, + Platform, + Provider, + ProviderModel, + RequirementCheckResult, + operation, + requirement_check, +) +from stateforge_provider._services.manifest.declaration import ( + ManifestDeclaration, +) +from stateforge_provider._services.operation_admission.runtime import ( + AdmissionRegistryError, + AdmissionRegistryFailure, + OperationAdmissionRuntime, +) + +_PRECONDITION_FAILED = operation_admission_pb2.AdmissionErrorReason.Value( + "ADMISSION_ERROR_REASON_OPERATION_PRECONDITION_FAILED" +) +_DEADLINE_EXPIRED = operation_admission_pb2.AdmissionErrorReason.Value( + "ADMISSION_ERROR_REASON_OPERATION_DEADLINE_EXPIRED" +) + + +class _Input(ProviderModel): + """Describe an operation request with one secret-bearing field.""" + + value: int + secret: str = Field(json_schema_extra={"writeOnly": True}) + + +class _Output(ProviderModel): + accepted: bool + + +class _AdmissionProvider(Provider): + identity = "example.admission" + display_name = "Admission test Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__(self, directory: Path) -> None: + super().__init__( + discovery_directory=directory, + state_directory=directory / "state", + ) + self.status = HealthStatus.healthy() + self.requirement = RequirementCheckResult.satisfied() + self.pause_requirement = False + self.requirement_started = asyncio.Event() + self.requirement_release = asyncio.Event() + self.calls = 0 + self.contexts: list[OperationContext] = [] + self.started = asyncio.Event() + self.release = asyncio.Event() + self.cancelled = asyncio.Event() + self.fail_dispatch = False + + async def health(self) -> HealthStatus: + return self.status + + @requirement_check("runtime.ready", "Runtime is ready") + async def ready(self) -> RequirementCheckResult: + if self.pause_requirement: + self.requirement_started.set() + await self.requirement_release.wait() + return self.requirement + + @operation( + "example.admission/run@1", + requires=("runtime.ready",), + implements=("example.operation.v1.audited",), + ) + async def run_operation( + self, + request: _Input, + context: OperationContext, + ) -> _Output: + self.calls += 1 + self.contexts.append(context) + self.started.set() + if self.fail_dispatch: + raise RuntimeError(f"secret author failure: {request.secret}") + try: + await self.release.wait() + except asyncio.CancelledError: + self.cancelled.set() + raise + return _Output(accepted=request.value > 0) + + @operation("example.admission/legacy@1") + async def legacy(self, request: _Input) -> _Output: + return _Output(accepted=request.value > 0) + + +@pytest.fixture +def endpoint_directory() -> Iterator[Path]: + directory = Path(tempfile.mkdtemp(prefix="sfp-r5-", dir="/tmp")) + try: + yield directory + finally: + shutil.rmtree(directory, ignore_errors=True) + + +@asynccontextmanager +async def _running( + provider: _AdmissionProvider, +) -> AsyncIterator[ + tuple[ + operation_admission_pb2_grpc.OperationAdmissionServiceStub, + bootstrap_pb2_grpc.BootstrapServiceStub, + manifest_pb2_grpc.ManifestServiceStub, + dict[str, object], + ] +]: + task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + try: + async with asyncio.timeout(3): + while not list( + provider._configuration.discovery_directory.glob("*.sfp.json") + ): + if task.done(): + await task + await asyncio.sleep(0.001) + descriptor = json.loads(provider._server.descriptor_path.read_text()) + channel = grpc.aio.insecure_channel( + f"unix:{descriptor['transport']['socket_path']}", + options=( + ( + "grpc.max_receive_message_length", + admission_contract.MAX_RESPONSE_BYTES, + ), + ( + "grpc.max_send_message_length", + admission_contract.MAX_REQUEST_BYTES + 1, + ), + ), + ) + await asyncio.wait_for(channel.channel_ready(), timeout=2) + yield ( + operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ), + bootstrap_pb2_grpc.BootstrapServiceStub(channel), + manifest_pb2_grpc.ManifestServiceStub(channel), + descriptor, + ) + finally: + provider.request_shutdown() + await asyncio.wait_for(task, timeout=3) + if channel is not None: + await channel.close() + + +async def _session(bootstrap, manifest, descriptor): + handshake = await bootstrap.Handshake( + bootstrap_pb2.HandshakeRequest( + agent_runtime=bootstrap_pb2.AgentRuntime( + runtime_id=str(uuid4()), + software_version="1.0.0", + ), + expected_provider_instance_id=descriptor["provider_instance_id"], + expected_endpoint_generation=descriptor["endpoint_generation"], + supported_sfp_versions=[ + bootstrap_pb2.ProtocolVersion(major=1, minor=0) + ], + ), + timeout=2, + ) + snapshot = await manifest.GetManifest( + manifest_pb2.GetManifestRequest( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + supported_manifest_schema_versions=[1], + ), + timeout=2, + ) + return handshake, snapshot + + +def _timestamp(value: datetime) -> Timestamp: + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +def _request( + handshake, + snapshot, + *, + operation_id: str | None = None, + operation_identifier: str = "example.admission/run@1", + payload: dict[str, object] | None = None, + deadline: datetime | None = None, + roles: tuple[str, ...] = ("example.operation.v1.audited",), + mode: int = operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE, +): + payload_message = Struct() + ParseDict(payload or {"value": 1, "secret": "hidden"}, payload_message) + context = Struct() + ParseDict({"authority": "policy-1", "nested": {"items": [1, 2]}}, context) + return operation_admission_pb2.StartOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + ), + invocation=operation_admission_pb2.OperationInvocation( + operation_id=operation_id or str(uuid4()), + operation_identifier=operation_identifier, + payload=payload_message, + execution_deadline=_timestamp( + deadline or datetime.now(UTC) + timedelta(minutes=5) + ), + contract=operation_admission_pb2.OperationContractContext( + manifest_revision=snapshot.manifest_revision, + required_implementation_roles=roles, + ), + execution_context=context, + ), + admission_mode=mode, + ) + + +def _detail(error: grpc.aio.AioRpcError): + status = rpc_status.from_call(error) + assert status is not None + packed = status.details[0] + detail = operation_admission_pb2.OperationAdmissionErrorDetail() + assert packed.Unpack(detail) + admission_validation.validate_admission_error_detail(detail) + return detail + + +@pytest.mark.asyncio +async def test_concurrent_identical_requests_dispatch_once_over_real_socket( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request(handshake, snapshot) + responses = await asyncio.gather( + *(admission.StartOperation(request, timeout=2) for _ in range(20)) + ) + await asyncio.wait_for(provider.started.wait(), timeout=1) + + assert provider.calls == 1 + assert ( + sum( + item.disposition + == operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW + for item in responses + ) + == 1 + ) + assert {item.record.request_fingerprint for item in responses} == { + admission_validation.compute_request_fingerprint( + request.invocation + ) + } + for response in responses: + admission_validation.validate_admission_exchange(request, response) + + context = provider.contexts[0] + assert str(context.operation_id) == request.invocation.operation_id + assert isinstance(context.execution_context, MappingProxyType) + with pytest.raises(TypeError): + context.execution_context["authority"] = "changed" # type: ignore[index] + nested = context.execution_context["nested"] + assert isinstance(nested, MappingProxyType) + + provider.release.set() + + +@pytest.mark.asyncio +async def test_existing_record_resolves_before_deadline_and_readiness_checks( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + deadline=datetime.now(UTC) + timedelta(milliseconds=50), + ) + accepted = await admission.StartOperation(request, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + await asyncio.sleep(0.06) + provider.status = HealthStatus.maintenance() + await provider.refresh_health() + resolved = await admission.StartOperation(request, timeout=2) + + assert accepted.disposition == ( + operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW + ) + assert resolved.disposition == ( + operation_admission_pb2.ADMISSION_DISPOSITION_RESOLVED_EXISTING + ) + assert provider.calls == 1 + provider.release.set() + + +@pytest.mark.asyncio +async def test_deadline_is_rechecked_after_requirement_evaluation( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + provider.pause_requirement = True + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + deadline=datetime.now(UTC) + timedelta(milliseconds=50), + ) + pending = asyncio.ensure_future(admission.StartOperation(request)) + await asyncio.wait_for(provider.requirement_started.wait(), timeout=1) + await asyncio.sleep(0.06) + provider.requirement_release.set() + + with pytest.raises(grpc.aio.AioRpcError) as captured: + await pending + assert captured.value.code() is grpc.StatusCode.DEADLINE_EXCEEDED + assert _detail(captured.value).reason == _DEADLINE_EXPIRED + assert provider.calls == 0 + + +@pytest.mark.asyncio +async def test_changed_content_conflicts_with_structured_safe_error( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + operation_id = str(uuid4()) + original = _request(handshake, snapshot, operation_id=operation_id) + await admission.StartOperation(original, timeout=2) + changed = _request( + handshake, + snapshot, + operation_id=operation_id, + payload={"value": 2, "secret": "must-not-leak"}, + ) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await admission.StartOperation(changed, timeout=2) + + assert captured.value.code() is grpc.StatusCode.ALREADY_EXISTS + detail = _detail(captured.value) + assert detail.reason == ( + operation_admission_pb2.ADMISSION_ERROR_REASON_IDEMPOTENCY_CONFLICT + ) + assert detail.operation_id == operation_id + assert "must-not-leak" not in captured.value.details() + provider.release.set() + + +@pytest.mark.parametrize( + ("change", "expected_code", "expected_reason"), + [ + ( + "missing", + grpc.StatusCode.UNIMPLEMENTED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_NOT_DECLARED, + ), + ( + "payload", + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, + ), + ( + "role", + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_INCOMPATIBLE, + ), + ( + "session", + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_PRECONDITION_FAILED, + ), + ( + "deadline", + grpc.StatusCode.DEADLINE_EXCEEDED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_DEADLINE_EXPIRED, + ), + ( + "resolve", + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + ), + ], +) +@pytest.mark.asyncio +async def test_rejects_before_creating_a_record( + endpoint_directory: Path, + change: str, + expected_code: grpc.StatusCode, + expected_reason, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request(handshake, snapshot) + if change == "missing": + request.invocation.operation_identifier = ( + "example.admission/missing@1" + ) + elif change == "payload": + request.invocation.payload.Clear() + request.invocation.payload.update( + {"value": "wrong", "secret": "hidden"} + ) + elif change == "role": + request.invocation.contract.required_implementation_roles[:] = [ + "example.operation.v1.missing" + ] + elif change == "session": + request.session_binding.expected_session_id = "stale-session" + elif change == "deadline": + request.invocation.execution_deadline.CopyFrom( + _timestamp(datetime.now(UTC) - timedelta(seconds=1)) + ) + else: + request.admission_mode = ( + operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY + ) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await admission.StartOperation(request, timeout=2) + + assert captured.value.code() is expected_code + assert _detail(captured.value).reason == expected_reason + assert provider.calls == 0 + + +@pytest.mark.parametrize("outcome", ["unmet", "incompatible"]) +@pytest.mark.asyncio +async def test_revalidates_current_requirements_before_admission( + endpoint_directory: Path, + outcome: str, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + provider.requirement = ( + RequirementCheckResult.unsatisfied("not ready") + if outcome == "unmet" + else RequirementCheckResult.incompatible("not supported") + ) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await admission.StartOperation( + _request(handshake, snapshot), timeout=2 + ) + + assert captured.value.code() is grpc.StatusCode.FAILED_PRECONDITION + assert _detail(captured.value).reason == ( + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_INCOMPATIBLE + if outcome == "incompatible" + else _PRECONDITION_FAILED + ) + assert provider.calls == 0 + + +@pytest.mark.asyncio +async def test_shutdown_cancels_admitted_author_dispatch( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + await admission.StartOperation( + _request(handshake, snapshot), timeout=2 + ) + await asyncio.wait_for(provider.started.wait(), timeout=1) + provider.request_shutdown() + await asyncio.wait_for(provider.cancelled.wait(), timeout=1) + + +@pytest.mark.asyncio +async def test_dispatch_failure_is_local_and_record_still_resolves( + endpoint_directory: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + provider.fail_dispatch = True + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + payload={"value": 1, "secret": "private-value"}, + ) + with caplog.at_level(logging.ERROR): + accepted = await admission.StartOperation(request, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + await asyncio.sleep(0) + resolved = await admission.StartOperation(request, timeout=2) + + assert accepted.disposition == ( + operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW + ) + assert resolved.disposition == ( + operation_admission_pb2.ADMISSION_DISPOSITION_RESOLVED_EXISTING + ) + assert provider.calls == 1 + assert "private-value" not in caplog.text + + +@pytest.mark.asyncio +async def test_runtime_retains_expiry_tombstone_and_bounds_capacity() -> None: + first_release = asyncio.Event() + + async def blocked() -> None: + await first_release.wait() + + runtime = OperationAdmissionRuntime( + maximum_records=1, + maximum_dispatches=1, + record_retention_seconds=0.01, + tombstone_retention_seconds=1, + ) + first = runtime.admit( + operation_id=str(uuid4()), + fingerprint="sha256:" + ("1" * 64), + endpoint_generation=1, + dispatch=blocked, + ) + with pytest.raises(AdmissionRegistryError) as capacity: + runtime.admit( + operation_id=str(uuid4()), + fingerprint="sha256:" + ("2" * 64), + endpoint_generation=1, + dispatch=blocked, + ) + assert capacity.value.failure is AdmissionRegistryFailure.CAPACITY + + await asyncio.sleep(0.02) + with pytest.raises(AdmissionRegistryError) as expired: + runtime.lookup( + first.record.operation_id, + first.record.request_fingerprint, + ) + assert expired.value.failure is AdmissionRegistryFailure.EXPIRED + await runtime.close() + + +def test_public_context_and_legacy_handler_are_both_discoverable() -> None: + declaration = ManifestDeclaration.discover(_AdmissionProvider) + operations = { + item.identifier: item for item in declaration.operation_declarations + } + assert operations["example.admission/run@1"].accepts_context + assert not operations["example.admission/legacy@1"].accepts_context diff --git a/uv.lock b/uv.lock index 640528e..2d4b070 100644 --- a/uv.lock +++ b/uv.lock @@ -427,6 +427,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f7/ec/67fbef5d497f86283db54c22eec6f6140243aae73265799baaaa19cd17fb/ghp_import-2.1.0-py3-none-any.whl", hash = "sha256:8337dd7b50877f163d4c0289bc1f1c7f127550241988d568c1db512c4324a619", size = 11034, upload-time = "2022-05-02T15:47:14.552Z" }, ] +[[package]] +name = "googleapis-common-protos" +version = "1.76.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/62/b6/bc095b23a3e0d978e75c8c18cabe4c11f791837410670e4486f0d539bfba/googleapis_common_protos-1.76.0.tar.gz", hash = "sha256:71db0e44a1d4c2dd024ebad44ca67129cc1e2892e4861df466764749d5ef8642", size = 156250, upload-time = "2026-10-22T23:43:36.263Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/05/6d/78dc67a9e779e594b49bf62604034d3a6aa428a55afc729cafda4b7333e1/googleapis_common_protos-1.76.0-py3-none-any.whl", hash = "sha256:59e13344a9f3651aa201a03e2bee815a3bf43a4baf1f9ff8a9b62a840ce25fac", size = 307707, upload-time = "2026-10-22T23:43:10.445Z" }, +] + [[package]] name = "griffelib" version = "2.4.0" @@ -510,6 +522,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/7c/4eb98f1e644b95704ac97693995a8158bd43c3c4030b63c9e980dab7bacb/grpcio_health_checking-1.84.0-py3-none-any.whl", hash = "sha256:3eab9b688e5e09bd1d584c296d3440358f11817388db87b807edd61f45b464d1", size = 19088, upload-time = "2026-09-14T07:09:00.21Z" }, ] +[[package]] +name = "grpcio-status" +version = "1.84.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "googleapis-common-protos" }, + { name = "grpcio" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/52/45/f80309cdb6a7dbf8f65e2082dd2ddc9797ba7180516a73c54d966ba632c4/grpcio_status-1.84.0.tar.gz", hash = "sha256:5caf28ba7184b81f618b5f7f094859fd2541bf429d2189bbbcd715c9c2cdcee2", size = 14015, upload-time = "2026-09-14T07:10:29.402Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/c4/3a77e4273e866b1b0c412afd80882e95941a37170032b5109d847c501124/grpcio_status-1.84.0-py3-none-any.whl", hash = "sha256:0c182ca0d6e60acbfd0e14499cf39a155e4827a1c3fd9f7638e49af15a74c30a", size = 14639, upload-time = "2026-09-14T07:10:15.175Z" }, +] + [[package]] name = "idna" version = "3.20" @@ -1604,6 +1630,7 @@ source = { editable = "." } dependencies = [ { name = "grpcio" }, { name = "grpcio-health-checking" }, + { name = "grpcio-status" }, { name = "jsonschema" }, { name = "pydantic" }, { name = "stateforge-protocol" }, @@ -1619,6 +1646,7 @@ dev = [ { name = "ruff" }, { name = "types-grpcio" }, { name = "types-grpcio-health-checking" }, + { name = "types-grpcio-status" }, { name = "types-jsonschema" }, { name = "types-protobuf" }, ] @@ -1633,6 +1661,7 @@ docs = [ requires-dist = [ { name = "grpcio", specifier = ">=1.83" }, { name = "grpcio-health-checking", specifier = ">=1.83" }, + { name = "grpcio-status", specifier = ">=1.83" }, { name = "jsonschema", specifier = ">=4.25,<5" }, { name = "pydantic", specifier = ">=2.12,<3" }, { name = "stateforge-protocol", git = "https://github.com/stateforge-org/stateforge-protocol.git?rev=cb4e9af352fc89204585d1f0b16d2870a990c560" }, @@ -1648,6 +1677,7 @@ dev = [ { name = "ruff", specifier = ">=0.12" }, { name = "types-grpcio" }, { name = "types-grpcio-health-checking" }, + { name = "types-grpcio-status" }, { name = "types-jsonschema" }, { name = "types-protobuf" }, ] @@ -1680,6 +1710,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/19/de/eca76e380f7e3b93bd7032847e4e9d43904e604b5782c83be9cf733be962/types_grpcio_health_checking-1.0.0.20261021-py3-none-any.whl", hash = "sha256:bac5319aa52e2012f745dba9c125f77326e60db14d60e15cff268dc8759387f5", size = 9828, upload-time = "2026-10-21T06:42:10.574Z" }, ] +[[package]] +name = "types-grpcio-status" +version = "1.0.0.20260408" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "types-grpcio" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/fa/ff/a0270f7083f76167030dacc1eaa81f9424a450b1828733c7b7f1607e7d40/types_grpcio_status-1.0.0.20260408.tar.gz", hash = "sha256:8c8b5a5fd5e661488bb3dfb924fd3562892bae01b9a24e607a26bb0029f3fb71", size = 7092, upload-time = "2026-04-08T04:35:18.169Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c4/b1/dcdd89efd40eb4a4c2458681a53faf057096db8e2b52a4f94f2bb935aef0/types_grpcio_status-1.0.0.20260408-py3-none-any.whl", hash = "sha256:5d421512b6a9ae5d0f84cde9c251394da5bd77ddc96e122221ac23c8e1ce3acf", size = 7859, upload-time = "2026-04-08T04:35:17.434Z" }, +] + [[package]] name = "types-jsonschema" version = "4.26.0.20261006" From 616dac0b16378b4b887d91793537f80398fc8f29 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 13:57:58 +0200 Subject: [PATCH 2/5] Fix direct-peer scenario evidence paths --- .../scenarios/operation/admission/action.yml | 3 ++- .../scenarios/trigger/publication/action.yml | 3 ++- tests/test_ci_automation.py | 14 ++++++++++++++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/.github/actions/scenarios/operation/admission/action.yml b/.github/actions/scenarios/operation/admission/action.yml index 76dd41e..d608983 100644 --- a/.github/actions/scenarios/operation/admission/action.yml +++ b/.github/actions/scenarios/operation/admission/action.yml @@ -37,8 +37,9 @@ runs: run: | set -euo pipefail mkdir -p "${EVIDENCE_DIRECTORY}" + evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)" cd "${RUNNER_TEMP}" unset PYTHONPATH PYTHONHOME "${ISOLATED_PYTHON}" -I "${ACTION_PATH}/verify.py" \ --runtime-directory "${RUNTIME_DIRECTORY}" \ - --output "${EVIDENCE_DIRECTORY}/operation-admission.json" + --output "${evidence_directory}/operation-admission.json" diff --git a/.github/actions/scenarios/trigger/publication/action.yml b/.github/actions/scenarios/trigger/publication/action.yml index 94209d8..56f66c0 100644 --- a/.github/actions/scenarios/trigger/publication/action.yml +++ b/.github/actions/scenarios/trigger/publication/action.yml @@ -37,8 +37,9 @@ runs: run: | set -euo pipefail mkdir -p "${EVIDENCE_DIRECTORY}" + evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)" cd "${RUNNER_TEMP}" unset PYTHONPATH PYTHONHOME "${ISOLATED_PYTHON}" -I "${ACTION_PATH}/verify.py" \ --runtime-directory "${RUNTIME_DIRECTORY}" \ - --output "${EVIDENCE_DIRECTORY}/trigger-publication.json" + --output "${evidence_directory}/trigger-publication.json" diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 9c898e8..8225da8 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -292,6 +292,13 @@ def test_trigger_publication_keeps_the_focused_protocol_peer() -> None: assert '"${ISOLATED_PYTHON}" -I' in scenario assert "${{ inputs.runtime-directory }}/venv" in scenario assert "${{ inputs.runtime-directory }}/p" in scenario + assert ( + 'evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)"' + in scenario + ) + assert ( + '--output "${evidence_directory}/trigger-publication.json"' in scenario + ) assert "${{ runner.temp }}/sfp-r4" in workflow assert "stateforge_protocol" in verifier assert "stateforge_agent" not in verifier @@ -331,6 +338,13 @@ def test_operation_admission_keeps_the_focused_protocol_peer() -> None: assert "./.github/actions/environment/install" in scenario assert "./.github/actions/agent/" not in scenario assert '"${ISOLATED_PYTHON}" -I' in scenario + assert ( + 'evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)"' + in scenario + ) + assert ( + '--output "${evidence_directory}/operation-admission.json"' in scenario + ) assert "stateforge_protocol" in verifier assert "stateforge_agent" not in verifier assert "@operation(" in verifier From 134cca1de08b9475e1ec1aa0a52968d2a3a35432 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 14:12:40 +0200 Subject: [PATCH 3/5] Validate admission fences before declarations --- .../_services/operation_admission/service.py | 2 +- tests/test_operation_admission.py | 27 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/stateforge_provider/_services/operation_admission/service.py b/src/stateforge_provider/_services/operation_admission/service.py index 4f8308a..e48015a 100644 --- a/src/stateforge_provider/_services/operation_admission/service.py +++ b/src/stateforge_provider/_services/operation_admission/service.py @@ -101,6 +101,7 @@ async def StartOperation( # noqa: N802, PLR0912 ) invocation = request.invocation + session = await self._validate_fences(request, context) operation = self._operations.get(invocation.operation_identifier) if operation is None: await self._reject( @@ -112,7 +113,6 @@ async def StartOperation( # noqa: N802, PLR0912 operation_id, ), ) - session = await self._validate_fences(request, context) try: self._validate_contract(invocation, operation) payload = self._validate_payload(invocation, operation) diff --git a/tests/test_operation_admission.py b/tests/test_operation_admission.py index d6f61b2..b4d9cc6 100644 --- a/tests/test_operation_admission.py +++ b/tests/test_operation_admission.py @@ -423,6 +423,33 @@ async def test_changed_content_conflicts_with_structured_safe_error( provider.release.set() +@pytest.mark.asyncio +async def test_stale_session_is_rejected_before_declaration_lookup( + endpoint_directory: Path, +) -> None: + provider = _AdmissionProvider(endpoint_directory) + async with _running(provider) as ( + admission, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + operation_identifier="example.admission/missing@1", + ) + request.session_binding.expected_session_id = "stale-session" + + with pytest.raises(grpc.aio.AioRpcError) as captured: + await admission.StartOperation(request, timeout=2) + + assert captured.value.code() is grpc.StatusCode.FAILED_PRECONDITION + assert _detail(captured.value).reason == _PRECONDITION_FAILED + assert provider.calls == 0 + + @pytest.mark.parametrize( ("change", "expected_code", "expected_reason"), [ From ebec33ee26a93ee65e3160edd995d9146b2c4257 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 14:13:18 +0200 Subject: [PATCH 4/5] Bound operation admission tombstones --- docs/operation-admission.md | 4 +++ .../_services/operation_admission/runtime.py | 3 ++- tests/test_operation_admission.py | 27 ++++++++++++++++++- 3 files changed, 32 insertions(+), 2 deletions(-) diff --git a/docs/operation-admission.md b/docs/operation-admission.md index 8a615dd..24462d5 100644 --- a/docs/operation-admission.md +++ b/docs/operation-admission.md @@ -121,6 +121,10 @@ Capacity and retention are bounded by Provider configuration: | `operation_record_retention_seconds` | 86,400 | | `operation_tombstone_retention_seconds` | 86,400 | +The admission-record limit covers live records and retained expiry tombstones +together. Tombstones are never discarded early to make room because doing so +would turn a known expired identity into an ambiguous absent identity. + Callers must treat `RESOURCE_EXHAUSTED` and shutdown-related `UNAVAILABLE` as retryable only under their own bounded policy. Retrying the same immutable request is safe because admission resolves by ID and fingerprint. diff --git a/src/stateforge_provider/_services/operation_admission/runtime.py b/src/stateforge_provider/_services/operation_admission/runtime.py index 2e36d50..252fff3 100644 --- a/src/stateforge_provider/_services/operation_admission/runtime.py +++ b/src/stateforge_provider/_services/operation_admission/runtime.py @@ -110,8 +110,9 @@ def admit( return existing if self._closed: raise AdmissionRegistryError(AdmissionRegistryFailure.STOPPING) + retained_identities = len(self._records) + len(self._tombstones) if ( - len(self._records) >= self._maximum_records + retained_identities >= self._maximum_records or len(self._tasks) >= self._maximum_dispatches ): raise AdmissionRegistryError(AdmissionRegistryFailure.CAPACITY) diff --git a/tests/test_operation_admission.py b/tests/test_operation_admission.py index b4d9cc6..8515c16 100644 --- a/tests/test_operation_admission.py +++ b/tests/test_operation_admission.py @@ -631,7 +631,7 @@ async def blocked() -> None: maximum_records=1, maximum_dispatches=1, record_retention_seconds=0.01, - tombstone_retention_seconds=1, + tombstone_retention_seconds=0.01, ) first = runtime.admit( operation_id=str(uuid4()), @@ -648,6 +648,9 @@ async def blocked() -> None: ) assert capacity.value.failure is AdmissionRegistryFailure.CAPACITY + first_release.set() + await asyncio.sleep(0) + await asyncio.sleep(0) await asyncio.sleep(0.02) with pytest.raises(AdmissionRegistryError) as expired: runtime.lookup( @@ -655,6 +658,28 @@ async def blocked() -> None: first.record.request_fingerprint, ) assert expired.value.failure is AdmissionRegistryFailure.EXPIRED + + second_operation_id = str(uuid4()) + second_fingerprint = "sha256:" + ("2" * 64) + with pytest.raises(AdmissionRegistryError) as tombstone_capacity: + runtime.admit( + operation_id=second_operation_id, + fingerprint=second_fingerprint, + endpoint_generation=1, + dispatch=blocked, + ) + assert ( + tombstone_capacity.value.failure is AdmissionRegistryFailure.CAPACITY + ) + + await asyncio.sleep(0.02) + admitted = runtime.admit( + operation_id=second_operation_id, + fingerprint=second_fingerprint, + endpoint_generation=1, + dispatch=blocked, + ) + assert admitted.accepted_new await runtime.close() From 9cc7b4080f0c7caeeec69d6f7540f7b90ebc2e7f Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 14:14:35 +0200 Subject: [PATCH 5/5] Document operation runtime dispatch --- src/stateforge_provider/declarations.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/stateforge_provider/declarations.py b/src/stateforge_provider/declarations.py index 96ee497..7924f03 100644 --- a/src/stateforge_provider/declarations.py +++ b/src/stateforge_provider/declarations.py @@ -108,13 +108,15 @@ def operation( """Declare an async method's versioned input/result contract. Annotate ``request`` and the return value with ProviderModel subclasses. - This milestone discovers metadata only; it does not dispatch the method. + The runtime derives the manifest schemas from those annotations and invokes + the method only after canonical Operation Admission succeeds. ``implements`` contains versioned semantic/compliance profile claims. Each claim says the operation conforms to that profile's required schemas and behavior. Claims are not permissions or Control Plane contract identifiers and are never inferred from a method or model name. ``requires`` contains stable IDs declared by :func:`requirement_check`. - The runtime evaluates each distinct check at most once per assessment. + The runtime evaluates each distinct check once per assessment or admission + attempt. Inherited methods follow Python's MRO; overriding requires redeclaration. """