diff --git a/.claude/PROVIDER_AUTHORING_VISION.md b/.claude/PROVIDER_AUTHORING_VISION.md index ecb3510..9178e16 100644 --- a/.claude/PROVIDER_AUTHORING_VISION.md +++ b/.claude/PROVIDER_AUTHORING_VISION.md @@ -47,8 +47,10 @@ The intended public surface consists of: - a `Provider` base class; - `Provider.run()` for a normal executable entry point; - `Provider.serve()` for an existing asynchronous event loop; -- overridable startup, shutdown, health, and capability hooks; +- overridable startup, shutdown, and health hooks; - an `@operation()` decorator for namespaced operation identifiers; +- a `@requirement_check()` decorator for read-only runtime prerequisites; +- typed requirement outcomes without exposing Capability Assessment messages; - typed and runtime-validated request and result models; - an `OperationContext` for identity, deadlines, cancellation, workspaces, and progress reporting; @@ -61,13 +63,16 @@ the Provider-author API. ## Intended author experience ```python +from shutil import which + from stateforge_provider import ( - CapabilityAssessment, HealthStatus, - OperationContext, + Platform, Provider, ProviderModel, + RequirementCheckResult, operation, + requirement_check, ) @@ -84,6 +89,7 @@ class GitProvider(Provider): identity = "stateforge.git" display_name = "Git Provider" version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) async def on_startup(self) -> None: """Initialize Provider resources.""" @@ -97,28 +103,30 @@ class GitProvider(Provider): """Report whether the Provider can accept work.""" return HealthStatus.healthy() - async def assess_capability( - self, - operation_identifier: str, - target: dict[str, object], - ) -> CapabilityAssessment: - """Assess an operation against the current target.""" - return CapabilityAssessment.supported() - - @operation("git.resolve_reference") + @requirement_check( + "runtime.git.executable", + description="The Git executable is available to the Provider", + ) + async def check_git_executable(self) -> RequirementCheckResult: + """Inspect one runtime prerequisite without mutating a target.""" + if which("git") is None: + return RequirementCheckResult.incompatible( + "The Git executable is unavailable" + ) + return RequirementCheckResult.satisfied() + + @operation( + "stateforge.git/resolve_reference@1", + requires=("runtime.git.executable",), + ) async def resolve_reference( self, request: ResolveReferenceInput, - context: OperationContext, ) -> ResolveReferenceOutput: commit = await self.git.resolve( repository=request.repository, reference=request.reference, ) - await context.report_progress( - message="Git reference resolved.", - percentage=100, - ) return ResolveReferenceOutput(commit=commit) @@ -136,9 +144,11 @@ async def main() -> None: `run()` owns the event loop. `serve()` runs in the caller's event loop and blocks until the Provider shuts down. -The names above define the intended public shape. Renaming or materially -changing that shape requires an explicit product decision; it must not happen -accidentally while implementing an internal SFP procedure. +The requirement declaration and result names above define the implemented +public capability-assessment shape. `OperationContext` and operation dispatch +remain later procedure work. Renaming or materially changing the author-facing +shape requires an explicit product decision; it must not happen accidentally +while implementing an internal SFP procedure. ## Operations @@ -155,6 +165,28 @@ identifiers, derives their declared schemas, and dispatches SFP requests. The default operation registry is in memory. Provider-controlled persistence is optional when a Provider needs stronger restart guarantees. +## Dynamic runtime requirements + +Capability Assessment is an internal SFP procedure, not an author hook. +Provider authors declare asynchronous, read-only checks with +`@requirement_check(identifier, description, version_constraint="")`, then +reference their stable IDs through `@operation(..., requires=(...))`. + +A check returns exactly one typed result: `satisfied`, `unsatisfied`, +`incompatible`, or `unknown`. The runtime publishes the static requirement +metadata in the manifest, deduplicates checks across a requested operation +set, executes each distinct check once, and privately adapts the outcomes to +P2 findings. Authors never create protobuf messages, findings, sessions, +correlation identifiers, or response envelopes. + +Checks inspect Provider runtime facts such as filesystem access, service +control, or executable availability. They do not dispatch operations, mutate +targets, authorize work, or receive a generic target. Exceptions and invalid +returns become sanitized remote failures while full diagnostics remain local. +The runtime owns deadlines, cancellation, session and manifest fencing, +readiness checks, deterministic result ordering, and final response +validation. + ## Backup and disaster-recovery compliance Backup and disaster recovery are predefined compliance profiles over generic @@ -190,8 +222,11 @@ does not receive a generic restore command. ## Manifest ownership Provider authors declare metadata, operations, trigger types, model schemas, -runtime requirements, and compliance roles through the public Python API. They -do not construct the wire manifest manually. +static or dynamically checked runtime requirements, and compliance roles +through the public Python API. They do not construct the wire manifest +manually. Operation-to-check associations stay private to the Provider +runtime; schema-1 manifests publish only the existing static requirement +metadata. Before publishing an endpoint, the runtime builds and validates the effective manifest. Startup fails when declarations are incomplete, conflicting, or @@ -261,6 +296,7 @@ The Developer RPC Library exclusively owns: - handshake and version negotiation; - manifest construction and exchange; - health-service exposure; +- capability-assessment fencing, check execution, and result adaptation; - operation dispatch, admission, progress, cancellation, and result mapping; - trigger publication and acknowledgement; and - safe startup, replacement, disconnection, and shutdown behavior. diff --git a/.claude/skills/stateforge-python-development/SKILL.md b/.claude/skills/stateforge-python-development/SKILL.md index e766f7c..2a69102 100644 --- a/.claude/skills/stateforge-python-development/SKILL.md +++ b/.claude/skills/stateforge-python-development/SKILL.md @@ -24,6 +24,12 @@ change that leaves the project in a verified state. - Give each module one cohesive responsibility and keep its dependencies explicit. Do not structure a file as though it must be self-contained. +- Before introducing a helper, search the repository for an existing helper + with the same semantics and reuse its established owner when one exists. +- When no equivalent helper exists, keep domain-specific behavior with its + domain. Put behavior that is not domain-specific in the narrowest cohesive + application-wide module that can serve every caller; do not duplicate it + inside an individual feature or service. - Introduce a helper, class, or module only when it improves reuse, clarity, or separation of concerns. Do not add wrappers that merely rename one call. - Keep related mutable state in one typed object. Do not maintain parallel @@ -33,7 +39,9 @@ change that leaves the project in a verified state. - Keep security and protocol validation linear when a single review surface is clearer. Make any complexity suppression narrow and document the reason. - Avoid hidden coupling, catch-all utility modules, circular imports, mutable - global state, and abstractions created only to satisfy a metric. + global state, and abstractions created only to satisfy a metric. A shared + module must have a clear responsibility and dependency direction rather than + becoming a generic `utils` collection. - Use a private third-party interface only when the dependency documents that interface and the reason is visible at the integration boundary. - Apply these rules automatically whenever production code is created or diff --git a/.github/actions/package/verify-installed/verify.py b/.github/actions/package/verify-installed/verify.py index cda3afd..8f26a9d 100644 --- a/.github/actions/package/verify-installed/verify.py +++ b/.github/actions/package/verify-installed/verify.py @@ -54,9 +54,12 @@ def main() -> None: "Provider", "ProviderModel", "ProviderServerError", + "RequirementCheckOutcome", + "RequirementCheckResult", "RuntimeRequirement", "Trigger", "operation", + "requirement_check", } missing = sorted(required_api.difference(stateforge_provider.__all__)) if missing: diff --git a/.github/actions/provider/assert-health/assert_health.py b/.github/actions/provider/assert-health/assert_health.py index 7123968..6c6f330 100644 --- a/.github/actions/provider/assert-health/assert_health.py +++ b/.github/actions/provider/assert-health/assert_health.py @@ -164,6 +164,7 @@ async def _verify(socket_path: str, unit_name: str) -> dict[str, object]: health_contract.PROVIDER_HEALTH_SERVICE, health_contract.BOOTSTRAP_SERVICE, health_contract.MANIFEST_SERVICE, + health_contract.CAPABILITY_ASSESSMENT_SERVICE, ): response = await stub.Check( health_pb2.HealthCheckRequest(service=service), diff --git a/.github/actions/provider/create-fixture/fixture.py b/.github/actions/provider/create-fixture/fixture.py index ce075b9..961a9c8 100644 --- a/.github/actions/provider/create-fixture/fixture.py +++ b/.github/actions/provider/create-fixture/fixture.py @@ -5,6 +5,7 @@ import argparse import asyncio import contextlib +import os import signal from pathlib import Path @@ -15,9 +16,11 @@ Platform, Provider, ProviderModel, + RequirementCheckResult, RuntimeRequirement, Trigger, operation, + requirement_check, ) @@ -119,9 +122,75 @@ async def health(self) -> HealthStatus: raise RuntimeError("fixture health hook failed") return self._health_status - @operation("stateforge.test/echo@1") + @requirement_check( + "runtime.filesystem.writable", + "Provider event directory is writable", + ) + async def writable(self) -> RequirementCheckResult: + """Inspect the fixture directory without mutating it.""" + self._record("requirement:runtime.filesystem.writable") + if os.access(self._event_log.parent, os.W_OK): + return RequirementCheckResult.satisfied() + return RequirementCheckResult.unsatisfied() + + @requirement_check("runtime.pending", "Pending fixture prerequisite") + async def pending(self) -> RequirementCheckResult: + """Expose a deterministic unmet requirement.""" + self._record("requirement:runtime.pending") + return RequirementCheckResult.unsatisfied() + + @requirement_check( + "runtime.incompatible", + "Incompatible fixture runtime", + ) + async def incompatible(self) -> RequirementCheckResult: + """Expose deterministic negative evidence.""" + self._record("requirement:runtime.incompatible") + return RequirementCheckResult.incompatible("Fixture incompatibility") + + @requirement_check("runtime.unknown", "Unobservable fixture prerequisite") + async def unknown(self) -> RequirementCheckResult: + """Expose deterministic unknown evidence.""" + self._record("requirement:runtime.unknown") + return RequirementCheckResult.unknown("Fixture evidence unavailable") + + @operation( + "stateforge.test/echo@1", + requires=("runtime.filesystem.writable",), + ) async def echo(self, request: EchoInput) -> EchoResult: - """Declare one typed operation without invoking it in R11.""" + """Declare one typed operation without invoking it during P2.""" + self._record("operation:echo") + return EchoResult(value=request.value) + + @operation( + "stateforge.test/pending@1", + requires=("runtime.filesystem.writable", "runtime.pending"), + ) + async def pending_operation(self, request: EchoInput) -> EchoResult: + """Declare a supportable operation with an unmet prerequisite.""" + self._record("operation:pending") + return EchoResult(value=request.value) + + @operation( + "stateforge.test/unsupported@1", + requires=("runtime.unknown", "runtime.incompatible"), + ) + async def unsupported_operation(self, request: EchoInput) -> EchoResult: + """Declare an operation with conclusive negative evidence.""" + self._record("operation:unsupported") + return EchoResult(value=request.value) + + @operation( + "stateforge.test/indeterminate@1", + requires=("runtime.unknown",), + ) + async def indeterminate_operation( + self, + request: EchoInput, + ) -> EchoResult: + """Declare an operation whose support cannot be established.""" + self._record("operation:indeterminate") return EchoResult(value=request.value) def _record(self, event: str) -> None: diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 38dc814..6b3f294 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -151,6 +151,10 @@ contract uses Unix-domain sockets and the Agent is packaged as a Linux service. Package identity and import checks remain independent and run on Linux, macOS, and Windows for Python 3.11, 3.12, and 3.13. +Provider-level tests exercise Capability Assessment over a real Unix socket. +There is intentionally no installed runtime P2 scenario until A3 supplies the +real Agent client; CI does not substitute a synthetic Agent-side caller. + ## Evidence and security Build provenance and checksums are retained with the immutable artifact bundle. diff --git a/docs/manifest-authoring.md b/docs/manifest-authoring.md index 46a9bff..148ff2c 100644 --- a/docs/manifest-authoring.md +++ b/docs/manifest-authoring.md @@ -37,14 +37,18 @@ Providers must add the required declarations before they can serve R1. ## Typed operations, triggers, and configuration ```python +from shutil import which + from pydantic import SecretStr from stateforge_provider import ( Platform, Provider, ProviderModel, + RequirementCheckResult, RuntimeRequirement, Trigger, operation, + requirement_check, ) @@ -72,7 +76,22 @@ class ReferenceProvider(Provider): ) triggers = (Trigger("example.reference/changed@1", NormalizeResult),) - @operation("example.reference/normalize@1") + @requirement_check( + "runtime.git.executable", + description="Git is available to normalize repository references", + version_constraint=">=2", + ) + async def git_executable(self) -> RequirementCheckResult: + if which("git") is None: + return RequirementCheckResult.incompatible( + "Git executable is unavailable" + ) + return RequirementCheckResult.satisfied() + + @operation( + "example.reference/normalize@1", + requires=("runtime.git.executable",), + ) async def normalize(self, request: NormalizeInput) -> NormalizeResult: return NormalizeResult(reference=request.reference.strip()) ``` @@ -101,9 +120,55 @@ Operation and trigger identifiers use `namespace/name@major`, with at least two lowercase dotted namespace segments and a positive major version. An incompatible input/result contract requires a new identifier version. Use tuples for platforms, requirements, triggers, and roles. -Requirements are descriptive only; the runtime does not run probes or -interpret their version constraints. An empty constraint means no declared -version restriction. +Plain `runtime_requirements` declarations are descriptive. Requirements +declared by `@requirement_check` add an author-owned, dynamic inspection while +publishing the same static metadata through the schema-1 +`runtime_requirements` map. The runtime does not interpret version-constraint +syntax; the check must verify whatever the named requirement means. An empty +constraint means no declared version restriction. + +## Dynamic requirement checks + +Use `@requirement_check(identifier, description, version_constraint="")` on +an async `(self)` method returning `RequirementCheckResult`. Associate one or +more IDs with an operation through `@operation(..., requires=(...))`. +Operation associations stay private to the Provider runtime and therefore do +not revise manifest schema 1. + +The four results have deliberately narrow meanings: + +| Result | Meaning | P2 support effect | +| --- | --- | --- | +| `satisfied()` | Current evidence proves the prerequisite is met. | Contributes to `SUPPORTED`. | +| `unsatisfied(explanation)` | The prerequisite is currently unmet but can be satisfied. | `SUPPORTED_WITH_REQUIREMENTS` with a `requirement_unmet` reason. | +| `incompatible(explanation)` | Current evidence proves the operation cannot be supported. | `UNSUPPORTED`. | +| `unknown(explanation)` | Available evidence cannot decide. | `INDETERMINATE`, unless another result is incompatible. | + +The `explanation` supplied to `unsatisfied()`, `incompatible()`, or +`unknown()` is copied unchanged into the remote P2 response. It is therefore +an author-sanitized, remote-visible diagnostic: never pass exception text, +credentials, paths, configured values, or vendor responses. The canonical P2 +limit is 1,024 Unicode code points. Exceeding that limit fails the assessment +with a sanitized `INTERNAL` response instead of truncating or leaking the +value. + +Checks inspect Provider-runtime facts such as filesystem access, service +control, or executable availability. They are read-only: do not create a file +as a writability probe, dispatch an operation, mutate a target, or treat an +assessment as authorization. The API intentionally supplies no generic target +or future operation payload. + +For one request, the runtime collects every referenced ID, preserves +first-reference order, runs each distinct check once, and reuses its result +across operations. RPC cancellation or a deadline cancels outstanding checks +where possible. Unhandled exceptions and values other than +`RequirementCheckResult` become sanitized `INTERNAL` failures; they are never +silently converted to `unknown`, and full details remain only in local logs. + +Startup fails before endpoint publication when a check ID is malformed or +duplicated, metadata exceeds Protocol bounds, the method is not async with an +exact `(self)` signature and result annotation, or an operation has a missing, +duplicate, or oversized requirement reference set. ## Compliance claims @@ -202,7 +267,7 @@ 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, target assessment, +This milestone does not implement configuration loading, target-specific assessment, dispatch, OperationContext, trigger publication, Agent caching/routing, or vendor work. Those remain their own milestones; these declarations are their shared input. diff --git a/docs/reference/declarations.md b/docs/reference/declarations.md index a50b637..bc64547 100644 --- a/docs/reference/declarations.md +++ b/docs/reference/declarations.md @@ -2,4 +2,13 @@ ::: stateforge_provider.declarations +::: stateforge_provider.requirements + ::: stateforge_provider.ManifestError + +Requirement-check methods must be asynchronous instance methods with exactly +`(self)` and a `RequirementCheckResult` return annotation. Operation +`requires` values are tuples of stable IDs declared by `@requirement_check`. +Startup rejects malformed or duplicate IDs, missing or repeated references, +invalid signatures, unresolved annotations, and metadata outside canonical +Protocol bounds. diff --git a/docs/reference/provider.md b/docs/reference/provider.md index dc0c341..5217fc8 100644 --- a/docs/reference/provider.md +++ b/docs/reference/provider.md @@ -5,3 +5,18 @@ ::: stateforge_provider.health.HealthStatus ::: stateforge_provider.ProviderServerError + +## Capability Assessment ownership + +`Provider` privately registers the P2 Capability Assessment service on the +same multiplexed endpoint as Bootstrap, Manifest, and Health. Provider authors +do not implement an assessment hook and never construct protocol findings. +They declare read-only requirement checks and operation associations through +the [declaration API](declarations.md). + +The runtime validates the active endpoint, negotiated session, served +manifest revision, requested operations, and current process-wide readiness. +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. diff --git a/docs/runtime-architecture.md b/docs/runtime-architecture.md index 6ba7cac..fc94b47 100644 --- a/docs/runtime-architecture.md +++ b/docs/runtime-architecture.md @@ -9,6 +9,7 @@ stateforge_provider/ _runtime/ # shared endpoint and process infrastructure _services/ bootstrap/ # Handshake and canonical contract enforcement + capability_assessment/ # Private P2 declaration and service adaptation health/ # Standard gRPC Health adaptation and bounded Watch manifest/ # Manifest declaration, validation, and RPC service ``` @@ -46,6 +47,14 @@ Health is the intentional exception to that generated-module path: it uses the official `grpc.health.v1` bindings from `grpcio-health-checking`, registers the actual hosted SFP service names, and adds no StateForge health protobuf. +Capability Assessment combines two private responsibilities. Its declaration +registry discovers author-facing requirement checks and operation +associations, while its service validates P2 through the canonical Protocol +API, enforces active endpoint/session/manifest/readiness fences, executes +deduplicated checks, and adapts typed results to wire findings. The registry +publishes only static requirement metadata through the existing manifest +schema. Provider authors never import this package or any P2 message. + To add a future procedure: 1. create one `_services/` package; diff --git a/examples/reference_provider.py b/examples/reference_provider.py index 1d54a72..113bdf4 100644 --- a/examples/reference_provider.py +++ b/examples/reference_provider.py @@ -1,5 +1,7 @@ """Demonstrate manifest declarations without protobuf or gRPC author code.""" +from shutil import which + from pydantic import SecretStr from stateforge_provider import ( @@ -7,9 +9,11 @@ Platform, Provider, ProviderModel, + RequirementCheckResult, RuntimeRequirement, Trigger, operation, + requirement_check, ) @@ -55,7 +59,22 @@ async def health(self) -> HealthStatus: """Report whether this Provider can accept new SFP work.""" return HealthStatus.healthy() - @operation("example.reference/normalize@1") + @requirement_check( + "runtime.git.executable", + "Git executable is available to this Provider", + ) + async def git_executable(self) -> RequirementCheckResult: + """Inspect executable availability without mutating a target.""" + if which("git") is None: + return RequirementCheckResult.incompatible( + "Git executable is unavailable" + ) + return RequirementCheckResult.satisfied() + + @operation( + "example.reference/normalize@1", + requires=("runtime.git.executable",), + ) async def normalize(self, request: NormalizeInput) -> NormalizeResult: """Normalize a reference; R1 advertises but does not dispatch it.""" return NormalizeResult(reference=request.reference.strip()) diff --git a/src/stateforge_provider/__init__.py b/src/stateforge_provider/__init__.py index 88854b6..64988ac 100644 --- a/src/stateforge_provider/__init__.py +++ b/src/stateforge_provider/__init__.py @@ -8,9 +8,14 @@ RuntimeRequirement, Trigger, operation, + requirement_check, ) from stateforge_provider.health import HealthStatus from stateforge_provider.provider import Provider +from stateforge_provider.requirements import ( + RequirementCheckOutcome, + RequirementCheckResult, +) from stateforge_provider.version import ProtocolVersion __all__ = [ @@ -21,7 +26,10 @@ "Provider", "ProviderModel", "ProviderServerError", + "RequirementCheckOutcome", + "RequirementCheckResult", "RuntimeRequirement", "Trigger", "operation", + "requirement_check", ] diff --git a/src/stateforge_provider/_services/capability_assessment/__init__.py b/src/stateforge_provider/_services/capability_assessment/__init__.py new file mode 100644 index 0000000..ceb2582 --- /dev/null +++ b/src/stateforge_provider/_services/capability_assessment/__init__.py @@ -0,0 +1 @@ +"""Implement private Provider-side Capability Assessment adaptation.""" diff --git a/src/stateforge_provider/_services/capability_assessment/declaration.py b/src/stateforge_provider/_services/capability_assessment/declaration.py new file mode 100644 index 0000000..2ec9057 --- /dev/null +++ b/src/stateforge_provider/_services/capability_assessment/declaration.py @@ -0,0 +1,198 @@ +"""Discover requirement checks and operation associations.""" + +from __future__ import annotations + +import inspect +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import get_type_hints + +from stateforge_protocol.sfp.capability_assessment.v1 import ( + contract as capability_contract, +) +from stateforge_protocol.sfp.manifest.v1.validation import ( + ManifestValidationError, + validate_structure, +) + +from stateforge_provider._services.manifest.errors import ManifestError +from stateforge_provider.declarations import ( + RuntimeRequirement, + _effective_methods, + _mro_namespace, + _Operation, + _RequirementCheck, +) +from stateforge_provider.requirements import RequirementCheckResult + +_BoundCheck = Callable[[], Awaitable[RequirementCheckResult]] + + +@dataclass(frozen=True, slots=True) +class RequirementDeclaration: + """Retain one validated check declaration and its method name.""" + + identifier: str + description: str + version_constraint: str + method_name: str + + def manifest_requirement(self) -> RuntimeRequirement: + """Return the static metadata published through manifest schema 1.""" + return RuntimeRequirement( + self.identifier, + self.description, + self.version_constraint, + ) + + +@dataclass(frozen=True, slots=True) +class RequirementRegistry: + """Freeze checks and Provider-owned operation associations.""" + + declarations: tuple[RequirementDeclaration, ...] + operation_requirements: tuple[tuple[str, tuple[str, ...]], ...] + + @classmethod + def discover(cls, owner: type) -> RequirementRegistry: + """Inspect one Provider class without invoking author code.""" + try: + return cls._discover(owner) + except ManifestValidationError as error: + raise ManifestError(str(error)) from error + + @classmethod + def _discover( # noqa: PLR0912 + cls, owner: type + ) -> RequirementRegistry: + methods, metadata_erased = _effective_methods( + owner, + metadata_attribute="__stateforge_requirement_check__", + metadata_type=_RequirementCheck, + ) + if metadata_erased: + raise ManifestError( + "requirement_checks: overrides must redeclare metadata" + ) + declarations: list[RequirementDeclaration] = [] + identifiers: set[str] = set() + for method_name, (method, declaring_class) in methods.items(): + metadata = getattr( + method, "__stateforge_requirement_check__", None + ) + if not isinstance(metadata, _RequirementCheck): + continue + path = f"requirement_checks[{len(declarations)}]" + if isinstance( + getattr(method, "__stateforge_operation__", None), _Operation + ): + raise ManifestError( + f"{path}: a method cannot be both a requirement check " + "and an operation" + ) + validate_structure(metadata.identifier, "dottedIdentifier", path) + validate_structure( + { + "description": metadata.description, + "versionConstraint": metadata.version_constraint, + }, + "runtimeRequirement", + path, + ) + if metadata.identifier in identifiers: + raise ManifestError( + f"{path}: duplicate requirement identifier" + ) + if not inspect.iscoroutinefunction(method): + raise ManifestError(f"{path}: requirement check must be async") + if tuple(inspect.signature(method).parameters) != ("self",): + raise ManifestError(f"{path}: expected (self) signature") + try: + annotations = get_type_hints( + method, + localns=_mro_namespace(declaring_class), + ) + except (NameError, TypeError): + raise ManifestError( + f"{path}: unresolved result annotation" + ) from None + if annotations.get("return") is not RequirementCheckResult: + raise ManifestError( + f"{path}: return annotation must be RequirementCheckResult" + ) + identifiers.add(metadata.identifier) + declarations.append( + RequirementDeclaration( + identifier=metadata.identifier, + description=metadata.description, + version_constraint=metadata.version_constraint, + method_name=method_name, + ) + ) + + operations: list[tuple[str, tuple[str, ...]]] = [] + operation_identifiers: set[str] = set() + for method, _ in methods.values(): + metadata = getattr(method, "__stateforge_operation__", None) + if not isinstance(metadata, _Operation): + continue + path = f"operations[{len(operations)}].requires" + requirements = metadata.requires + if ( + not isinstance(requirements, tuple) + or len(requirements) + > capability_contract.MAX_REQUIREMENT_FINDINGS_PER_OPERATION + ): + raise ManifestError(f"{path}: must be a bounded tuple") + for identifier in requirements: + validate_structure(identifier, "dottedIdentifier", path) + if len(set(requirements)) != len(requirements): + raise ManifestError(f"{path}: duplicate requirement reference") + missing = tuple( + identifier + for identifier in requirements + if identifier not in identifiers + ) + if missing: + raise ManifestError(f"{path}: unknown requirement reference") + if metadata.identifier in operation_identifiers: + raise ManifestError( + "operations: duplicate operation identifier" + ) + operation_identifiers.add(metadata.identifier) + operations.append((metadata.identifier, requirements)) + + return cls(tuple(declarations), tuple(operations)) + + @property + def manifest_requirements(self) -> tuple[RuntimeRequirement, ...]: + """Return static metadata for P1 manifest publication.""" + return tuple( + declaration.manifest_requirement() + for declaration in self.declarations + ) + + def bind(self, provider: object) -> BoundRequirementRegistry: + """Bind validated method names to one Provider instance.""" + checks: dict[str, _BoundCheck] = {} + for declaration in self.declarations: + method = getattr(provider, declaration.method_name) + if not callable(method): + raise ManifestError( + "requirement_checks: declared method is unavailable" + ) + checks[declaration.identifier] = method + return BoundRequirementRegistry( + declarations={item.identifier: item for item in self.declarations}, + checks=checks, + operation_requirements=dict(self.operation_requirements), + ) + + +@dataclass(frozen=True, slots=True) +class BoundRequirementRegistry: + """Resolve checks and associations for one running Provider instance.""" + + declarations: dict[str, RequirementDeclaration] + checks: dict[str, _BoundCheck] + operation_requirements: dict[str, tuple[str, ...]] diff --git a/src/stateforge_provider/_services/capability_assessment/service.py b/src/stateforge_provider/_services/capability_assessment/service.py new file mode 100644 index 0000000..1b8a211 --- /dev/null +++ b/src/stateforge_provider/_services/capability_assessment/service.py @@ -0,0 +1,435 @@ +"""Serve Provider-owned runtime requirement assessments over P2.""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from uuid import uuid4 + +import grpc +from google.protobuf.timestamp_pb2 import Timestamp +from stateforge_protocol.sfp.capability_assessment.v1 import ( + capability_assessment_pb2, + capability_assessment_pb2_grpc, + contract, + 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, + RequirementDeclaration, +) +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.version import ProtocolVersion + +_LOGGER = logging.getLogger(__name__) + +_SUPPORTED_WITH_REQUIREMENTS = ( + capability_assessment_pb2.SUPPORT_STATE_SUPPORTED_WITH_REQUIREMENTS +) +_UNSATISFIED_EXPLANATION = "A runtime requirement is not satisfied" +_INCOMPATIBLE_EXPLANATION = "A runtime requirement is incompatible" +_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 +): + """Adapt private Provider checks to the canonical P2 wire contract.""" + + def __init__( + self, + *, + endpoint: ProviderEndpointIdentity, + snapshot: ManifestSnapshot, + sessions: SessionRegistry, + health: ProviderHealthService, + requirements: BoundRequirementRegistry, + ) -> None: + self._endpoint = endpoint + self._manifest_revision = snapshot.manifest_revision + self._declared_operations = snapshot.operation_identifiers + self._sessions = sessions + self._health = health + self._requirements = requirements + + # Keep validation and error translation in one auditable RPC boundary. + async def AssessCapabilities( # noqa: N802, PLR0912 + self, + request: capability_assessment_pb2.AssessCapabilitiesRequest, + context: grpc.aio.ServicerContext[ + capability_assessment_pb2.AssessCapabilitiesRequest, + capability_assessment_pb2.AssessCapabilitiesResponse, + ], + ) -> capability_assessment_pb2.AssessCapabilitiesResponse: + """Evaluate each referenced requirement once without dispatching.""" + if request.ByteSize() > contract.MAX_REQUEST_BYTES: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Capability Assessment request limit exceeded", + ) + try: + validation.validate_assess_capabilities_request(request) + except validation.CapabilityAssessmentValidationError: + await context.abort( + grpc.StatusCode.INVALID_ARGUMENT, + "Invalid Capability Assessment request", + ) + + session = await self._validate_fences(request, context) + if not self._health.is_serving(): + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "Provider is not ready for Capability Assessment", + ) + + try: + requirement_ids = self._required_checks(request) + results = await self._execute_checks(requirement_ids) + except _CheckError: + await context.abort( + grpc.StatusCode.INTERNAL, + "Provider requirement check failed", + ) + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.exception( + "Provider failed to evaluate Capability Assessment" + ) + await context.abort( + grpc.StatusCode.INTERNAL, + "Capability Assessment response is unavailable", + ) + + if context.cancelled(): + raise asyncio.CancelledError + if self._sessions.get(request.expected_session_id) is not session: + await context.abort( + grpc.StatusCode.FAILED_PRECONDITION, + "Capability Assessment session is no longer active", + ) + if not self._health.is_serving(): + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "Provider readiness changed during Capability Assessment", + ) + + try: + response = self._response(request, results) + response_size = response.ByteSize() + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.exception( + "Provider failed to construct Capability Assessment response" + ) + await context.abort( + grpc.StatusCode.INTERNAL, + "Capability Assessment response is unavailable", + ) + if response_size > contract.MAX_RESPONSE_BYTES: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Capability Assessment response limit exceeded", + ) + try: + validation.validate_assessment_exchange(request, response) + except validation.CapabilityAssessmentValidationError: + _LOGGER.exception( + "Provider produced an invalid Capability Assessment response" + ) + await context.abort( + grpc.StatusCode.INTERNAL, + "Capability Assessment response is unavailable", + ) + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.exception( + "Provider failed to validate Capability Assessment response" + ) + await context.abort( + grpc.StatusCode.INTERNAL, + "Capability Assessment response is unavailable", + ) + return response + + async def _validate_fences( + self, + request: capability_assessment_pb2.AssessCapabilitiesRequest, + context: grpc.aio.ServicerContext[ + capability_assessment_pb2.AssessCapabilitiesRequest, + capability_assessment_pb2.AssessCapabilitiesResponse, + ], + ) -> Session: + """Validate Provider-owned session, manifest, and operation state.""" + session = self._sessions.get(request.expected_session_id) + if session is None or ( + request.expected_provider_instance_id + != str(self._endpoint.provider_instance_id) + or request.expected_endpoint_generation + != self._endpoint.endpoint_generation + or session.endpoint != self._endpoint + or session.version != ProtocolVersion(1, 0) + ): + await context.abort( + grpc.StatusCode.FAILED_PRECONDITION, + "Capability Assessment session is not active for this " + "endpoint", + ) + if request.expected_manifest_revision != self._manifest_revision: + await context.abort( + grpc.StatusCode.FAILED_PRECONDITION, + "Capability Assessment manifest revision is stale", + ) + if any( + identifier not in self._declared_operations + for identifier in request.operation_identifiers + ): + await context.abort( + grpc.StatusCode.FAILED_PRECONDITION, + "Capability Assessment operation is not declared", + ) + 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], + ) -> capability_assessment_pb2.AssessCapabilitiesResponse: + """Build one deterministic request-correlated response.""" + timestamp = Timestamp() + timestamp.GetCurrentTime() + return capability_assessment_pb2.AssessCapabilitiesResponse( + assessment_request_id=request.assessment_request_id, + provider_instance_id=str(self._endpoint.provider_instance_id), + endpoint_generation=self._endpoint.endpoint_generation, + session_id=request.expected_session_id, + manifest_revision=self._manifest_revision, + assessment_id=str(uuid4()), + assessment_timestamp=timestamp, + findings=( + self._finding(operation, results) + for operation in request.operation_identifiers + ), + ) + + def _finding( + self, + operation: str, + results: dict[str, _EvaluatedRequirement], + ) -> capability_assessment_pb2.CapabilityFinding: + """Aggregate one operation using the locked precedence rules.""" + evaluated = tuple( + results[identifier] + for identifier in self._requirements.operation_requirements[ + operation + ] + ) + outcomes = {item.result.outcome for item in evaluated} + reasons: tuple[capability_assessment_pb2.ReasonDetail, ...] = () + if RequirementCheckOutcome.INCOMPATIBLE in outcomes: + support = capability_assessment_pb2.SUPPORT_STATE_UNSUPPORTED + reasons = ( + _reason( + contract.REASON_CODE_REQUIREMENT_UNMET, + _explanation( + evaluated, RequirementCheckOutcome.INCOMPATIBLE + ), + ), + ) + elif RequirementCheckOutcome.UNKNOWN in outcomes: + support = capability_assessment_pb2.SUPPORT_STATE_INDETERMINATE + reasons = ( + _reason( + contract.REASON_CODE_INSUFFICIENT_EVIDENCE, + _explanation(evaluated, RequirementCheckOutcome.UNKNOWN), + ), + ) + elif RequirementCheckOutcome.UNSATISFIED in outcomes: + support = _SUPPORTED_WITH_REQUIREMENTS + reasons = ( + _reason( + contract.REASON_CODE_REQUIREMENT_UNMET, + _explanation( + evaluated, RequirementCheckOutcome.UNSATISFIED + ), + ), + ) + else: + support = capability_assessment_pb2.SUPPORT_STATE_SUPPORTED + return capability_assessment_pb2.CapabilityFinding( + operation_identifier=operation, + support_state=support, + requirement_findings=( + _requirement_finding(item) for item in evaluated + ), + reasons=reasons, + ) + + +def _requirement_finding( + evaluated: _EvaluatedRequirement, +) -> capability_assessment_pb2.RequirementFinding: + """Adapt one typed author result to a bounded P2 finding.""" + state = { + RequirementCheckOutcome.SATISFIED: ( + capability_assessment_pb2.REQUIREMENT_STATE_SATISFIED + ), + RequirementCheckOutcome.UNSATISFIED: ( + capability_assessment_pb2.REQUIREMENT_STATE_UNSATISFIED + ), + RequirementCheckOutcome.INCOMPATIBLE: ( + capability_assessment_pb2.REQUIREMENT_STATE_UNSATISFIED + ), + RequirementCheckOutcome.UNKNOWN: ( + capability_assessment_pb2.REQUIREMENT_STATE_UNKNOWN + ), + }[evaluated.result.outcome] + declaration = evaluated.declaration + return capability_assessment_pb2.RequirementFinding( + requirement_id=declaration.identifier, + state=state, + description=declaration.description, + version_constraint=declaration.version_constraint, + ) + + +def _explanation( + evaluated: tuple[_EvaluatedRequirement, ...], + outcome: RequirementCheckOutcome, +) -> str: + """Select a bounded nonempty explanation for a normalized reason.""" + item = next(item for item in evaluated if item.result.outcome is outcome) + if item.result.explanation: + return item.result.explanation + return { + RequirementCheckOutcome.UNSATISFIED: _UNSATISFIED_EXPLANATION, + RequirementCheckOutcome.INCOMPATIBLE: _INCOMPATIBLE_EXPLANATION, + RequirementCheckOutcome.UNKNOWN: _UNKNOWN_EXPLANATION, + }[outcome] + + +def _reason( + code: str, + explanation: str, +) -> capability_assessment_pb2.ReasonDetail: + """Build one canonical machine-readable reason.""" + return capability_assessment_pb2.ReasonDetail( + reason_code=code, + explanation=explanation, + ) + + +def registration( + *, + endpoint: ProviderEndpointIdentity, + snapshot: ManifestSnapshot, + sessions: SessionRegistry, + health: ProviderHealthService, + requirements: BoundRequirementRegistry, +) -> ProcedureRegistration: + """Build the bounded P2 registration for one Provider generation.""" + service = CapabilityAssessmentService( + endpoint=endpoint, + snapshot=snapshot, + sessions=sessions, + health=health, + requirements=requirements, + ) + + def install(server: grpc.aio.Server) -> None: + capability_assessment_pb2_grpc.add_CapabilityAssessmentServiceServicer_to_server( + service, + server, + ) + + return ProcedureRegistration( + name=contract.SERVICE_NAME, + max_request_bytes=contract.MAX_REQUEST_BYTES, + max_response_bytes=contract.MAX_RESPONSE_BYTES, + install=install, + ) diff --git a/src/stateforge_provider/_services/health/service.py b/src/stateforge_provider/_services/health/service.py index b19ec1a..fb1db0d 100644 --- a/src/stateforge_provider/_services/health/service.py +++ b/src/stateforge_provider/_services/health/service.py @@ -48,6 +48,14 @@ def __init__( self._observer_count = 0 self._stopping = False + def is_serving(self) -> bool: + """Return current Provider-wide readiness for local SFP services.""" + return ( + not self._stopping + and self._statuses[health_contract.PROVIDER_HEALTH_SERVICE] + == health_pb2.HealthCheckResponse.SERVING + ) + def register_services(self, service_names: Iterable[str]) -> None: """Register hosted SFP services as initially not serving.""" overall_status = self._statuses[ diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index 39e394c..79d4d5c 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -35,6 +35,8 @@ ProviderModel, RuntimeRequirement, Trigger, + _effective_methods, + _mro_namespace, _Operation, ) @@ -70,33 +72,15 @@ def _sequence(owner: type, name: str, limit: int) -> tuple[Any, ...]: return value -def _mro_namespace(owner: type) -> dict[str, Any]: - """Collect the class namespace used where a method was declared.""" - namespace: dict[str, Any] = {} - for base in reversed(owner.__mro__): - namespace.update(vars(base)) - return namespace - - # Keep discovery and its rejection rules adjacent. -def _operations( # noqa: PLR0912 - owner: type, limits: ProviderRuntimeLimits -) -> dict[str, Any]: - methods: dict[str, tuple[Any, type]] = {} - for base in reversed(owner.__mro__): - for name, method in vars(base).items(): - prior = methods.get(name) - if getattr( - prior[0] if prior is not None else None, - "__stateforge_operation__", - None, - ) is not None and not isinstance( - getattr(method, "__stateforge_operation__", None), _Operation - ): - raise ManifestError( - "operations: overrides must redeclare metadata" - ) - methods[name] = (method, base) +def _operations(owner: type, limits: ProviderRuntimeLimits) -> dict[str, Any]: + methods, metadata_erased = _effective_methods( + owner, + metadata_attribute="__stateforge_operation__", + metadata_type=_Operation, + ) + if metadata_erased: + raise ManifestError("operations: overrides must redeclare metadata") declared_methods: list[tuple[Any, type, _Operation]] = [] for method, declaring_class in methods.values(): metadata = getattr(method, "__stateforge_operation__", None) @@ -192,10 +176,17 @@ def discover( owner: type, *, limits: ProviderRuntimeLimits = DEFAULT_RUNTIME_LIMITS, + additional_runtime_requirements: tuple[RuntimeRequirement, ...] = (), ) -> ManifestDeclaration: """Inspect class metadata without invoking business methods.""" try: - return cls._discover(owner, limits=limits) + return cls._discover( + owner, + limits=limits, + additional_runtime_requirements=( + additional_runtime_requirements + ), + ) except ManifestValidationError as error: raise ManifestError(str(error)) from error @@ -205,6 +196,7 @@ def _discover( owner: type, *, limits: ProviderRuntimeLimits, + additional_runtime_requirements: tuple[RuntimeRequirement, ...], ) -> ManifestDeclaration: """Build a declaration while the public boundary translates errors.""" platforms = _sequence( @@ -221,14 +213,25 @@ def _discover( ] for item in platform_documents: validate_structure(item, "platform", "platforms") - requirements = {} - for index, requirement in enumerate( - _sequence( - owner, - "runtime_requirements", - manifest_contract.MAX_RUNTIME_REQUIREMENTS, + declared_requirements = _sequence( + owner, + "runtime_requirements", + manifest_contract.MAX_RUNTIME_REQUIREMENTS, + ) + if not isinstance(additional_runtime_requirements, tuple): + raise ManifestError( + "runtime_requirements: additional declarations must be a tuple" ) + combined_requirements = ( + declared_requirements + additional_runtime_requirements + ) + if ( + len(combined_requirements) + > manifest_contract.MAX_RUNTIME_REQUIREMENTS ): + raise ManifestError("runtime_requirements: entry limit exceeded") + requirements = {} + for index, requirement in enumerate(combined_requirements): path = f"runtime_requirements[{index}]" if not isinstance(requirement, RuntimeRequirement): raise ManifestError(f"{path}: expected RuntimeRequirement") @@ -335,6 +338,19 @@ class ManifestSnapshot: document: bytes + @property + def manifest_revision(self) -> str: + """Return the immutable revision served by this snapshot.""" + return manifest_pb2.GetManifestResponse.FromString( + self.document + ).manifest_revision + + @property + def operation_identifiers(self) -> frozenset[str]: + """Return the operations declared by this immutable snapshot.""" + response = manifest_pb2.GetManifestResponse.FromString(self.document) + return frozenset(response.manifest.operations) + def response(self, session_id: str) -> manifest_pb2.GetManifestResponse: """Return a fresh complete envelope bound to this validated call.""" response = manifest_pb2.GetManifestResponse.FromString(self.document) diff --git a/src/stateforge_provider/declarations.py b/src/stateforge_provider/declarations.py index e42fdbf..96ee497 100644 --- a/src/stateforge_provider/declarations.py +++ b/src/stateforge_provider/declarations.py @@ -53,13 +53,57 @@ class Trigger: class _Operation: identifier: str implements: tuple[str, ...] + requires: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class _RequirementCheck: + identifier: str + description: str + version_constraint: str _Handler = TypeVar("_Handler", bound=Callable[..., Any]) +def _mro_namespace(owner: type) -> dict[str, Any]: + """Collect the class namespace used where a method was declared.""" + namespace: dict[str, Any] = {} + for base in reversed(owner.__mro__): + namespace.update(vars(base)) + return namespace + + +def _effective_methods( + owner: type, + *, + metadata_attribute: str, + metadata_type: type[Any], +) -> tuple[dict[str, tuple[Any, type]], bool]: + """Resolve inherited methods and report erased declaration metadata.""" + methods: dict[str, tuple[Any, type]] = {} + metadata_erased = False + for base in reversed(owner.__mro__): + for name, method in vars(base).items(): + prior = methods.get(name) + if getattr( + prior[0] if prior is not None else None, + metadata_attribute, + None, + ) is not None and not isinstance( + getattr(method, metadata_attribute, None), + metadata_type, + ): + metadata_erased = True + methods[name] = (method, base) + return methods, metadata_erased + + def operation( - identifier: str, *, implements: tuple[str, ...] = () + identifier: str, + *, + implements: tuple[str, ...] = (), + requires: tuple[str, ...] = (), ) -> Callable[[_Handler], _Handler]: """Declare an async method's versioned input/result contract. @@ -69,6 +113,8 @@ def operation( 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. Inherited methods follow Python's MRO; overriding requires redeclaration. """ @@ -79,7 +125,33 @@ def decorate(handler: _Handler) -> _Handler: setattr( # noqa: B010 handler, "__stateforge_operation__", - _Operation(identifier, implements), + _Operation(identifier, implements, requires), + ) + return handler + + return decorate + + +def requirement_check( + identifier: str, + description: str, + version_constraint: str = "", +) -> Callable[[_Handler], _Handler]: + """Declare an asynchronous, read-only runtime prerequisite check. + + Decorate a Provider instance method with a ``(self)`` signature returning + :class:`~stateforge_provider.RequirementCheckResult`. The runtime publishes + the static description in the manifest and privately adapts results to + Capability Assessment findings. + """ + + def decorate(handler: _Handler) -> _Handler: + if hasattr(handler, "__stateforge_requirement_check__"): + raise ValueError("A requirement check cannot be declared twice") + setattr( # noqa: B010 + handler, + "__stateforge_requirement_check__", + _RequirementCheck(identifier, description, version_constraint), ) return handler diff --git a/src/stateforge_provider/provider.py b/src/stateforge_provider/provider.py index c567705..e524ebb 100644 --- a/src/stateforge_provider/provider.py +++ b/src/stateforge_provider/provider.py @@ -18,6 +18,12 @@ from stateforge_provider._runtime.identity import ProviderEndpointIdentity from stateforge_provider._runtime.server import ProviderServer from stateforge_provider._runtime.sessions import SessionRegistry +from stateforge_provider._services.capability_assessment.declaration import ( + RequirementRegistry, +) +from stateforge_provider._services.capability_assessment.service import ( + registration as capability_assessment_registration, +) from stateforge_provider._services.health.service import ProviderHealthService from stateforge_provider._services.manifest.declaration import ( ManifestDeclaration, @@ -150,10 +156,15 @@ async def serve(self) -> None: ManifestError: Declarations violate the P1 manifest contract. ProviderServerError: Endpoint allocation/publication fails. """ + requirement_registry = RequirementRegistry.discover(type(self)) declaration = ManifestDeclaration.discover( type(self), limits=self._configuration.limits, + additional_runtime_requirements=( + requirement_registry.manifest_requirements + ), ) + bound_requirements = requirement_registry.bind(self) health_service = ProviderHealthService( max_watchers=self._configuration.limits.max_health_watchers, watch_queue_capacity=( @@ -177,7 +188,16 @@ def procedures( sessions: SessionRegistry, ) -> tuple[ProcedureRegistration, ...]: snapshot = declaration.snapshot(identity) - return (manifest_registration(snapshot, sessions),) + return ( + manifest_registration(snapshot, sessions), + capability_assessment_registration( + endpoint=identity, + snapshot=snapshot, + sessions=sessions, + health=health_service, + requirements=bound_requirements, + ), + ) def health_registration( registrations: Sequence[ProcedureRegistration], diff --git a/src/stateforge_provider/requirements.py b/src/stateforge_provider/requirements.py new file mode 100644 index 0000000..908d358 --- /dev/null +++ b/src/stateforge_provider/requirements.py @@ -0,0 +1,58 @@ +"""Expose typed author results for dynamic runtime requirement checks.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum + + +class RequirementCheckOutcome(StrEnum): + """Describe one Provider-owned runtime prerequisite outcome.""" + + SATISFIED = "satisfied" + UNSATISFIED = "unsatisfied" + INCOMPATIBLE = "incompatible" + UNKNOWN = "unknown" + + +@dataclass(frozen=True, slots=True) +class RequirementCheckResult: + """Return a requirement outcome without exposing SFP wire models. + + ``explanation`` is optional for satisfied and unsatisfied results. It is + copied unchanged into the remote P2 response for unsatisfied, + incompatible, and unknown results. Authors must supply sanitized, + non-secret text rather than raw exception details. The canonical Protocol + limit is 1,024 Unicode code points; the runtime rejects an assessment that + exceeds it. + """ + + outcome: RequirementCheckOutcome + explanation: str = "" + + def __post_init__(self) -> None: + """Reject values outside the author-facing result vocabulary.""" + if not isinstance(self.outcome, RequirementCheckOutcome): + raise TypeError("outcome must be a RequirementCheckOutcome") + if not isinstance(self.explanation, str): + raise TypeError("explanation must be a string") + + @classmethod + def satisfied(cls) -> RequirementCheckResult: + """Report that the runtime prerequisite is currently met.""" + return cls(RequirementCheckOutcome.SATISFIED) + + @classmethod + def unsatisfied(cls, explanation: str = "") -> RequirementCheckResult: + """Report an unmet prerequisite with optional remote-safe context.""" + return cls(RequirementCheckOutcome.UNSATISFIED, explanation) + + @classmethod + def incompatible(cls, explanation: str = "") -> RequirementCheckResult: + """Report incompatibility with an optional remote-safe explanation.""" + return cls(RequirementCheckOutcome.INCOMPATIBLE, explanation) + + @classmethod + def unknown(cls, explanation: str = "") -> RequirementCheckResult: + """Report uncertainty with an optional remote-safe explanation.""" + return cls(RequirementCheckOutcome.UNKNOWN, explanation) diff --git a/tests/test_capability_assessment_service.py b/tests/test_capability_assessment_service.py new file mode 100644 index 0000000..2484dce --- /dev/null +++ b/tests/test_capability_assessment_service.py @@ -0,0 +1,896 @@ +"""Real-socket evidence for private Provider Capability Assessment.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import shutil +import tempfile +from collections import Counter +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import uuid4 + +import grpc +import pytest +from google.protobuf.message import Message +from google.protobuf.struct_pb2 import Struct +from stateforge_protocol.sfp.bootstrap.v1 import ( + bootstrap_pb2, + bootstrap_pb2_grpc, +) +from stateforge_protocol.sfp.capability_assessment.v1 import ( + capability_assessment_pb2, + capability_assessment_pb2_grpc, +) +from stateforge_protocol.sfp.capability_assessment.v1 import ( + contract as capability_contract, +) +from stateforge_protocol.sfp.capability_assessment.v1 import ( + validation as capability_validation, +) +from stateforge_protocol.sfp.manifest.v1 import ( + manifest_pb2, + manifest_pb2_grpc, +) + +from stateforge_provider import ( + HealthStatus, + Platform, + Provider, + ProviderModel, + RequirementCheckResult, + operation, + requirement_check, +) +from stateforge_provider._services.capability_assessment.service import ( + CapabilityAssessmentService, +) + +_MAX_SINGLE_BYTE_VARINT = 0x7F +_VARINT_CONTINUATION_BIT = 0x80 + + +class _Input(ProviderModel): + value: str + + +class _Output(ProviderModel): + accepted: bool + + +class _CapabilityProvider(Provider): + identity = "example.capability" + display_name = "Capability 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.checks: Counter[str] = Counter() + self.operation_calls = 0 + self.invalid_result = False + self.raise_check = False + self.unsatisfied_explanation = ( + "A prerequisite is currently unavailable" + ) + self.unknown_explanation = "Evidence is unavailable" + self.slow_started = asyncio.Event() + self.slow_release = asyncio.Event() + self.slow_cancelled = asyncio.Event() + + async def health(self) -> HealthStatus: + return self.status + + @requirement_check("runtime.satisfied", "Satisfied prerequisite") + async def satisfied(self) -> RequirementCheckResult: + self.checks["satisfied"] += 1 + return RequirementCheckResult.satisfied() + + @requirement_check("runtime.unsatisfied", "Unmet prerequisite") + async def unsatisfied(self) -> RequirementCheckResult: + self.checks["unsatisfied"] += 1 + return RequirementCheckResult.unsatisfied(self.unsatisfied_explanation) + + @requirement_check("runtime.incompatible", "Incompatible runtime") + async def incompatible(self) -> RequirementCheckResult: + self.checks["incompatible"] += 1 + return RequirementCheckResult.incompatible("Runtime is incompatible") + + @requirement_check("runtime.unknown", "Unobservable prerequisite") + async def unknown(self) -> RequirementCheckResult: + self.checks["unknown"] += 1 + return RequirementCheckResult.unknown(self.unknown_explanation) + + @requirement_check( + "runtime.incompatible_without_explanation", + "x" * 1_025, + ) + async def incompatible_without_explanation( + self, + ) -> RequirementCheckResult: + return RequirementCheckResult.incompatible() + + @requirement_check( + "runtime.unknown_without_explanation", + "x" * 1_025, + ) + async def unknown_without_explanation(self) -> RequirementCheckResult: + return RequirementCheckResult.unknown() + + @requirement_check("runtime.invalid_result", "Typed result required") + async def invalid(self) -> RequirementCheckResult: + self.checks["invalid"] += 1 + if self.invalid_result: + return object() # type: ignore[return-value] + return RequirementCheckResult.satisfied() + + @requirement_check("runtime.failure", "Failure remains local") + async def failure(self) -> RequirementCheckResult: + self.checks["failure"] += 1 + if self.raise_check: + raise RuntimeError("sensitive author failure") + return RequirementCheckResult.satisfied() + + @requirement_check("runtime.slow", "Slow prerequisite") + async def slow(self) -> RequirementCheckResult: + self.checks["slow"] += 1 + self.slow_started.set() + try: + await self.slow_release.wait() + except asyncio.CancelledError: + self.slow_cancelled.set() + raise + return RequirementCheckResult.satisfied() + + async def _operation(self, request: _Input) -> _Output: + self.operation_calls += 1 + return _Output(accepted=bool(request.value)) + + @operation( + "example.capability/supported@1", + requires=("runtime.satisfied",), + ) + async def supported(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/unmet@1", + requires=("runtime.satisfied", "runtime.unsatisfied"), + ) + async def unmet(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/unsupported@1", + requires=("runtime.unknown", "runtime.incompatible"), + ) + async def unsupported(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/indeterminate@1", + requires=("runtime.unknown",), + ) + async def indeterminate(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation("example.capability/empty@1") + async def empty(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/invalid@1", + requires=("runtime.invalid_result",), + ) + async def invalid_operation(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/failure@1", + requires=("runtime.failure",), + ) + async def failure_operation(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/slow@1", + requires=("runtime.slow",), + ) + async def slow_operation(self, request: _Input) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/incompatible_without_explanation@1", + requires=("runtime.incompatible_without_explanation",), + ) + async def incompatible_without_explanation_operation( + self, + request: _Input, + ) -> _Output: + return await self._operation(request) + + @operation( + "example.capability/unknown_without_explanation@1", + requires=("runtime.unknown_without_explanation",), + ) + async def unknown_without_explanation_operation( + self, + request: _Input, + ) -> _Output: + return await self._operation(request) + + +@pytest.fixture +def endpoint_directory() -> Iterator[Path]: + directory = Path(tempfile.mkdtemp(prefix="sfp-r3-", dir="/tmp")) + try: + yield directory + finally: + shutil.rmtree(directory, ignore_errors=True) + + +@asynccontextmanager +async def _running( + provider: _CapabilityProvider, +) -> AsyncIterator[ + tuple[ + capability_assessment_pb2_grpc.CapabilityAssessmentServiceStub, + 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", + capability_contract.MAX_RESPONSE_BYTES, + ), + ( + "grpc.max_send_message_length", + capability_contract.MAX_REQUEST_BYTES + 1, + ), + ), + ) + await asyncio.wait_for(channel.channel_ready(), timeout=2) + yield ( + capability_assessment_pb2_grpc.CapabilityAssessmentServiceStub( + 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: bootstrap_pb2_grpc.BootstrapServiceStub, + manifest: manifest_pb2_grpc.ManifestServiceStub, + descriptor: dict[str, object], +): + 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 _request(handshake, snapshot, operations, **overrides): + return capability_assessment_pb2.AssessCapabilitiesRequest( + **{ + "assessment_request_id": str(uuid4()), + "expected_provider_instance_id": handshake.provider_instance_id, + "expected_endpoint_generation": handshake.endpoint_generation, + "expected_session_id": handshake.session_id, + "expected_manifest_revision": snapshot.manifest_revision, + "execution_context": Struct(), + "operation_identifiers": operations, + **overrides, + } + ) + + +def _varint(value: int) -> bytes: + """Encode one nonnegative protobuf integer for a test-only field.""" + encoded = bytearray() + while value > _MAX_SINGLE_BYTE_VARINT: + encoded.append( + (value & _MAX_SINGLE_BYTE_VARINT) | _VARINT_CONTINUATION_BIT + ) + value >>= 7 + encoded.append(value) + return bytes(encoded) + + +def _message_with_serialized_size( + message: Message, + target_size: int, +) -> Message: + """Pad a message with one ignored field to an exact wire size.""" + content = message.SerializeToString() + tag = _varint((2_047 << 3) | 2) + payload_size = target_size - len(content) - len(tag) - 1 + while True: + encoded_size = _varint(payload_size) + adjusted = target_size - len(content) - len(tag) - len(encoded_size) + if adjusted == payload_size: + break + payload_size = adjusted + if payload_size < 0: + raise ValueError("target size is smaller than the message") + result = type(message).FromString( + content + tag + encoded_size + (b"x" * payload_size) + ) + if result.ByteSize() != target_size: + raise AssertionError("message does not match its target size") + return result + + +@pytest.mark.asyncio +async def test_assesses_over_real_socket_with_deduplication_and_precedence( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + operations = [ + "example.capability/indeterminate@1", + "example.capability/supported@1", + "example.capability/unmet@1", + "example.capability/unsupported@1", + "example.capability/empty@1", + ] + request = _request(handshake, snapshot, operations) + response = await capability.AssessCapabilities(request, timeout=2) + + capability_validation.validate_assessment_exchange(request, response) + assert [ + item.operation_identifier for item in response.findings + ] == operations + assert [item.support_state for item in response.findings] == [ + capability_assessment_pb2.SUPPORT_STATE_INDETERMINATE, + capability_assessment_pb2.SUPPORT_STATE_SUPPORTED, + capability_assessment_pb2.SUPPORT_STATE_SUPPORTED_WITH_REQUIREMENTS, + capability_assessment_pb2.SUPPORT_STATE_UNSUPPORTED, + capability_assessment_pb2.SUPPORT_STATE_SUPPORTED, + ] + assert provider.checks == { + "satisfied": 1, + "unsatisfied": 1, + "incompatible": 1, + "unknown": 1, + } + assert provider.operation_calls == 0 + unmet = response.findings[2] + assert ( + unmet.reasons[0].reason_code + == capability_contract.REASON_CODE_REQUIREMENT_UNMET + ) + assert ( + unmet.reasons[0].explanation + == "A prerequisite is currently unavailable" + ) + unsupported = response.findings[3] + assert ( + unsupported.reasons[0].reason_code + == capability_contract.REASON_CODE_REQUIREMENT_UNMET + ) + assert [ + item.requirement_id for item in unsupported.requirement_findings + ] == [ + "runtime.unknown", + "runtime.incompatible", + ] + + +@pytest.mark.asyncio +async def test_uses_bounded_defaults_for_missing_reason_explanations( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + provider.unsatisfied_explanation = "" + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + [ + "example.capability/unmet@1", + "example.capability/incompatible_without_explanation@1", + "example.capability/unknown_without_explanation@1", + ], + ) + response = await capability.AssessCapabilities(request, timeout=2) + + capability_validation.validate_assessment_exchange(request, response) + assert [ + finding.reasons[0].explanation for finding in response.findings + ] == [ + "A runtime requirement is not satisfied", + "A runtime requirement is incompatible", + "Runtime requirement evidence is unavailable", + ] + + +@pytest.mark.asyncio +async def test_explicit_reason_explanation_is_remote_visible_and_bounded( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + provider.unknown_explanation = "x" * 1_024 + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + ["example.capability/indeterminate@1"], + ) + response = await capability.AssessCapabilities(request, timeout=2) + assert response.findings[0].reasons[0].explanation == "x" * 1_024 + + provider.unknown_explanation = "sensitive:" + ("x" * 1_025) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities(request, timeout=2) + + assert error.value.code() is grpc.StatusCode.INTERNAL + assert "sensitive" not in error.value.details() + + +@pytest.mark.parametrize( + ("change", "code"), + [ + ("malformed", grpc.StatusCode.INVALID_ARGUMENT), + ("session", grpc.StatusCode.FAILED_PRECONDITION), + ("instance", grpc.StatusCode.FAILED_PRECONDITION), + ("generation", grpc.StatusCode.FAILED_PRECONDITION), + ("manifest", grpc.StatusCode.FAILED_PRECONDITION), + ("operation", grpc.StatusCode.FAILED_PRECONDITION), + ], +) +@pytest.mark.asyncio +async def test_maps_validation_and_contextual_fence_failures( + endpoint_directory: Path, + change: str, + code: grpc.StatusCode, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ) + if change == "malformed": + request.assessment_request_id = "not-a-uuid" + elif change == "session": + request.expected_session_id = "not-negotiated" + elif change == "instance": + request.expected_provider_instance_id = str(uuid4()) + elif change == "generation": + request.expected_endpoint_generation += 1 + elif change == "manifest": + request.expected_manifest_revision = "sha256:" + ("0" * 64) + else: + request.operation_identifiers[0] = "example.capability/missing@1" + + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities(request, timeout=2) + assert error.value.code() is code + assert provider.checks == {} + assert provider.operation_calls == 0 + + +@pytest.mark.asyncio +async def test_rejects_not_ready_provider_before_running_checks( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + provider.status = HealthStatus.maintenance() + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ), + timeout=2, + ) + assert error.value.code() is grpc.StatusCode.UNAVAILABLE + assert provider.checks == {} + + +@pytest.mark.parametrize("failure", ["invalid", "exception"]) +@pytest.mark.asyncio +async def test_invalid_author_results_are_sanitized_internal( + endpoint_directory: Path, + failure: str, + caplog: pytest.LogCaptureFixture, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + provider.invalid_result = failure == "invalid" + provider.raise_check = failure == "exception" + suffix = failure if failure == "invalid" else "failure" + operation = f"example.capability/{suffix}@1" + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + with ( + caplog.at_level(logging.ERROR), + pytest.raises(grpc.aio.AioRpcError) as error, + ): + await capability.AssessCapabilities( + _request(handshake, snapshot, [operation]), + timeout=2, + ) + assert error.value.code() is grpc.StatusCode.INTERNAL + assert "sensitive author failure" not in error.value.details() + if failure == "exception": + assert "sensitive author failure" in caplog.text + + +@pytest.mark.asyncio +async def test_revalidates_readiness_after_checks( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + call = asyncio.ensure_future( + capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/slow@1"], + ), + timeout=2, + ) + ) + await asyncio.wait_for(provider.slow_started.wait(), timeout=1) + provider.status = HealthStatus.maintenance() + await provider.refresh_health() + provider.slow_release.set() + with pytest.raises(grpc.aio.AioRpcError) as error: + await call + assert error.value.code() is grpc.StatusCode.UNAVAILABLE + + +@pytest.mark.asyncio +async def test_revalidates_session_after_checks( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + call = asyncio.ensure_future( + capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/slow@1"], + ), + timeout=2, + ) + ) + await asyncio.wait_for(provider.slow_started.wait(), timeout=1) + sessions = provider._server._sessions + assert sessions is not None + sessions.clear() + provider.slow_release.set() + with pytest.raises(grpc.aio.AioRpcError) as error: + await call + assert error.value.code() is grpc.StatusCode.FAILED_PRECONDITION + + +@pytest.mark.asyncio +async def test_rpc_deadline_cancels_in_flight_requirement_checks( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/slow@1"], + ), + timeout=0.05, + ) + assert error.value.code() is grpc.StatusCode.DEADLINE_EXCEEDED + await asyncio.wait_for(provider.slow_cancelled.wait(), timeout=1) + + +@pytest.mark.asyncio +async def test_client_cancellation_stops_in_flight_requirement_checks( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + call = capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/slow@1"], + ), + timeout=2, + ) + await asyncio.wait_for(provider.slow_started.wait(), timeout=1) + assert call.cancel() + with pytest.raises(asyncio.CancelledError): + await call + await asyncio.wait_for(provider.slow_cancelled.wait(), timeout=1) + + +@pytest.mark.asyncio +async def test_enforces_exact_and_oversized_request_boundaries( + endpoint_directory: Path, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + base = _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ) + exact = _message_with_serialized_size( + base, + capability_contract.MAX_REQUEST_BYTES, + ) + response = await capability.AssessCapabilities(exact, timeout=3) + assert response.findings[0].operation_identifier == ( + "example.capability/supported@1" + ) + + oversized = _message_with_serialized_size( + base, + capability_contract.MAX_REQUEST_BYTES + 1, + ) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities(oversized, timeout=3) + assert error.value.code() is grpc.StatusCode.RESOURCE_EXHAUSTED + + +@pytest.mark.parametrize( + ("response_size", "expected_code"), + [ + (capability_contract.MAX_RESPONSE_BYTES, None), + ( + capability_contract.MAX_RESPONSE_BYTES + 1, + grpc.StatusCode.RESOURCE_EXHAUSTED, + ), + ], +) +@pytest.mark.asyncio +async def test_enforces_response_boundaries( + endpoint_directory: Path, + monkeypatch: pytest.MonkeyPatch, + response_size: int, + expected_code: grpc.StatusCode | None, +) -> None: + original_response = CapabilityAssessmentService._response + + def padded_response(self, request, results): + response = original_response(self, request, results) + return _message_with_serialized_size(response, response_size) + + monkeypatch.setattr( + CapabilityAssessmentService, + "_response", + padded_response, + ) + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + call = capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ), + timeout=5, + ) + if expected_code is None: + response = await call + assert response.ByteSize() == response_size + else: + with pytest.raises(grpc.aio.AioRpcError) as error: + await call + assert error.value.code() is expected_code + + +@pytest.mark.asyncio +async def test_sanitizes_unexpected_response_construction_failures( + endpoint_directory: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + def fail_response(*_args, **_kwargs): + raise RuntimeError("sensitive response failure") + + monkeypatch.setattr( + CapabilityAssessmentService, + "_response", + fail_response, + ) + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + with ( + caplog.at_level(logging.ERROR), + pytest.raises(grpc.aio.AioRpcError) as error, + ): + await capability.AssessCapabilities( + _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ), + timeout=2, + ) + + assert error.value.code() is grpc.StatusCode.INTERNAL + assert "sensitive response failure" not in error.value.details() + assert "sensitive response failure" in caplog.text + + +@pytest.mark.asyncio +async def test_delegates_request_and_response_validation_to_protocol( + endpoint_directory: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + provider = _CapabilityProvider(endpoint_directory) + async with _running(provider) as ( + capability, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + request = _request( + handshake, + snapshot, + ["example.capability/supported@1"], + ) + + def reject_request(_request) -> None: + raise capability_validation.CapabilityAssessmentValidationError( + "canonical request" + ) + + monkeypatch.setattr( + capability_validation, + "validate_assess_capabilities_request", + reject_request, + ) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities(request, timeout=2) + assert error.value.code() is grpc.StatusCode.INVALID_ARGUMENT + + monkeypatch.undo() + + def reject_response(_request, _response) -> None: + raise capability_validation.CapabilityAssessmentValidationError( + "canonical response" + ) + + monkeypatch.setattr( + capability_validation, + "validate_assessment_exchange", + reject_response, + ) + with pytest.raises(grpc.aio.AioRpcError) as error: + await capability.AssessCapabilities(request, timeout=2) + assert error.value.code() is grpc.StatusCode.INTERNAL diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 7eb4147..9c72c50 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -189,6 +189,7 @@ def test_fixture_uses_only_the_public_provider_api() -> None: assert "stateforge_provider._" not in fixture assert "Provider(" in fixture assert "@operation(" in fixture + assert "@requirement_check(" in fixture assert "on_startup" in fixture assert "on_shutdown" in fixture @@ -288,13 +289,16 @@ def test_sfp_wire_contracts_are_protocol_owned() -> None: bootstrap = SERVICES / "bootstrap" health = SERVICES / "health" manifest = SERVICES / "manifest" + capability = SERVICES / "capability_assessment" bootstrap_service = read(bootstrap / "service.py") health_service = read(health / "service.py") manifest_service = read(manifest / "service.py") + capability_service = read(capability / "service.py") assert not (bootstrap / "limits.py").exists() assert not (health / "limits.py").exists() assert not (manifest / "limits.py").exists() + assert not (capability / "limits.py").exists() assert "contract.MAX_REQUEST_BYTES" in bootstrap_service assert "contract.MAX_RESPONSE_BYTES" in bootstrap_service assert "contract.SERVICE_NAME" in bootstrap_service @@ -305,8 +309,18 @@ def test_sfp_wire_contracts_are_protocol_owned() -> None: assert "contract.MAX_REQUEST_BYTES" in manifest_service assert "contract.MAX_RESPONSE_BYTES" in manifest_service assert "contract.SERVICE_NAME" in manifest_service + assert "contract.MAX_REQUEST_BYTES" in capability_service + assert "contract.MAX_RESPONSE_BYTES" in capability_service + assert "contract.SERVICE_NAME" in capability_service + assert "contract.REASON_CODE_REQUIREMENT_UNMET" in capability_service + assert "contract.REASON_CODE_INSUFFICIENT_EVIDENCE" in capability_service + assert "stateforge.capability_assessment.v1." not in capability_service + assert "validation.validate_assess_capabilities_request" in ( + capability_service + ) + assert "validation.validate_assessment_exchange" in capability_service - for directory in (bootstrap, health, manifest): + for directory in (bootstrap, health, manifest, capability): combined = "\n".join(read(path) for path in directory.glob("*.py")) assert "MAX_REQUEST_BYTES =" not in combined assert "MAX_RESPONSE_BYTES =" not in combined diff --git a/tests/test_requirement_declarations.py b/tests/test_requirement_declarations.py new file mode 100644 index 0000000..e9afa7e --- /dev/null +++ b/tests/test_requirement_declarations.py @@ -0,0 +1,265 @@ +"""Declaration tests for Provider-owned dynamic runtime requirements.""" + +from __future__ import annotations + +from dataclasses import FrozenInstanceError +from uuid import UUID + +import pytest +from google.protobuf.json_format import MessageToDict +from stateforge_protocol.sfp.capability_assessment.v1 import contract + +from stateforge_provider import ( + ManifestError, + Platform, + Provider, + ProviderModel, + RequirementCheckResult, + RuntimeRequirement, + operation, + requirement_check, +) +from stateforge_provider import declarations as public_declarations +from stateforge_provider._runtime.identity import ProviderEndpointIdentity +from stateforge_provider._services.capability_assessment import ( + declaration as capability_declaration, +) +from stateforge_provider._services.capability_assessment.declaration import ( + RequirementRegistry, +) +from stateforge_provider._services.manifest import ( + declaration as manifest_declaration, +) +from stateforge_provider._services.manifest.declaration import ( + ManifestDeclaration, +) + +_INSTANCE = UUID("7f7db7d8-68b4-4f71-9769-065b5c4f941c") + + +class _Input(ProviderModel): + value: str + + +class _Output(ProviderModel): + accepted: bool + + +class _BaseProvider(Provider): + identity = "example.capability" + display_name = "Capability Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + +class _DeclaredProvider(_BaseProvider): + runtime_requirements = ( + RuntimeRequirement("runtime.python", "Python runtime", ">=3.11"), + ) + + @requirement_check( + "filesystem.var.writable", + "Writable Provider state directory", + ) + async def writable(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + @operation( + "example.capability/write@1", + requires=("filesystem.var.writable",), + ) + async def write(self, request: _Input) -> _Output: + return _Output(accepted=bool(request.value)) + + +def _manifest(owner: type[Provider]): + registry = RequirementRegistry.discover(owner) + declaration = ManifestDeclaration.discover( + owner, + additional_runtime_requirements=registry.manifest_requirements, + ) + response = declaration.snapshot( + ProviderEndpointIdentity(_INSTANCE, 1) + ).response("session") + return registry, MessageToDict( + response.manifest, + always_print_fields_with_no_presence=True, + ) + + +def test_discovers_checks_references_and_manifest_metadata() -> None: + registry, manifest = _manifest(_DeclaredProvider) + + assert registry.operation_requirements == ( + ("example.capability/write@1", ("filesystem.var.writable",)), + ) + assert [item.identifier for item in registry.declarations] == [ + "filesystem.var.writable" + ] + assert manifest["runtimeRequirements"] == { + "filesystem.var.writable": { + "description": "Writable Provider state directory", + "versionConstraint": "", + }, + "runtime.python": { + "description": "Python runtime", + "versionConstraint": ">=3.11", + }, + } + assert ( + "requires" not in manifest["operations"]["example.capability/write@1"] + ) + + +@pytest.mark.parametrize( + ("identifier", "description", "version_constraint"), + [ + ("invalid", "Valid", ""), + ("filesystem.var.writable", "", ""), + ("filesystem.var.writable", "x" * 2_049, ""), + ("filesystem.var.writable", "Valid", "x" * 256), + ], +) +def test_rejects_noncanonical_or_oversized_check_metadata( + identifier: str, + description: str, + version_constraint: str, +) -> None: + class Invalid(_BaseProvider): + @requirement_check(identifier, description, version_constraint) + async def check(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + with pytest.raises(ManifestError): + RequirementRegistry.discover(Invalid) + + +def test_rejects_duplicate_check_identifiers() -> None: + class Duplicate(_BaseProvider): + @requirement_check("runtime.shared", "First") + async def first(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + @requirement_check("runtime.shared", "Second") + async def second(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + with pytest.raises(ManifestError, match="duplicate requirement"): + RequirementRegistry.discover(Duplicate) + + +@pytest.mark.parametrize( + "references", + [ + ("runtime.missing",), + ("runtime.present", "runtime.present"), + ("runtime.present",) + * (contract.MAX_REQUIREMENT_FINDINGS_PER_OPERATION + 1), + ["runtime.present"], + ], +) +def test_rejects_invalid_operation_requirement_references( + references: object, +) -> None: + class Invalid(_BaseProvider): + @requirement_check("runtime.present", "Present") + async def present(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + @operation("example.capability/run@1", requires=references) + async def run(self, request: _Input) -> _Output: + return _Output(accepted=bool(request.value)) + + with pytest.raises(ManifestError): + RequirementRegistry.discover(Invalid) + + +def test_rejects_duplicate_static_and_checked_manifest_requirement() -> None: + class Duplicate(_DeclaredProvider): + runtime_requirements = ( + RuntimeRequirement( + "filesystem.var.writable", + "Duplicate static declaration", + ), + ) + + registry = RequirementRegistry.discover(Duplicate) + with pytest.raises(ManifestError, match="duplicate requirement"): + ManifestDeclaration.discover( + Duplicate, + additional_runtime_requirements=registry.manifest_requirements, + ) + + +def test_rejects_invalid_check_signatures_and_return_annotations() -> None: + def sync(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + async def argument(self, value: str) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + async def wrong_return(self) -> str: + return "satisfied" + + for method in (sync, argument, wrong_return): + owner = type( + "Invalid", + (_BaseProvider,), + { + "check": requirement_check( + "runtime.invalid", + "Invalid check", + )(method) + }, + ) + with pytest.raises(ManifestError): + RequirementRegistry.discover(owner) + + +def test_rejects_erased_or_conflicting_inherited_metadata() -> None: + class Erased(_DeclaredProvider): + async def writable(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + class Conflict(_BaseProvider): + @operation("example.capability/conflict@1") + @requirement_check("runtime.conflict", "Conflict") + async def conflict(self) -> RequirementCheckResult: + return RequirementCheckResult.satisfied() + + for owner in (Erased, Conflict): + with pytest.raises(ManifestError): + RequirementRegistry.discover(owner) + + +def test_author_result_factories_are_typed_and_immutable() -> None: + assert RequirementCheckResult.satisfied().outcome.value == "satisfied" + assert RequirementCheckResult.unsatisfied("later").explanation == "later" + assert RequirementCheckResult.incompatible("never").outcome.value == ( + "incompatible" + ) + assert RequirementCheckResult.unknown("unclear").outcome.value == "unknown" + with pytest.raises(TypeError): + RequirementCheckResult("satisfied") # type: ignore[arg-type] + with pytest.raises(FrozenInstanceError): + RequirementCheckResult.satisfied().explanation = "changed" # type: ignore[misc] + + +def test_declaration_namespace_helper_has_one_owner() -> None: + """Keep class reflection shared across declaration services.""" + assert ( + capability_declaration._mro_namespace + is public_declarations._mro_namespace + ) + assert ( + manifest_declaration._mro_namespace + is public_declarations._mro_namespace + ) + assert ( + capability_declaration._effective_methods + is public_declarations._effective_methods + ) + assert ( + manifest_declaration._effective_methods + is public_declarations._effective_methods + )