[TASK][MVP] Implement Provider Capability Assessment

.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.

.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

.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:

.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),

.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:

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.