[TASK][MVP] Implement Provider Health and Availability Reporting

.github/actions/package/verify-installed/verify.py

@@ -47,6 +47,7 @@ def main() -> None:
         raise SystemExit("Installed Provider version is incorrect")
 
     required_api = {
+        "HealthStatus",
         "ManifestError",
         "Platform",
         "ProtocolVersion",

.github/actions/provider/assert-health/action.yml

@@ -0,0 +1,40 @@
+name: Assert standard Provider Health behavior
+description: Exercise Check, ordered Watch changes, and shutdown on the installed endpoint.
+
+inputs:
+  container-name:
+    description: Fedora container running the Agent and Provider.
+    required: true
+  unit-name:
+    description: Transient systemd unit running the Provider fixture.
+    required: true
+  socket-path:
+    description: Generation-specific Provider Unix socket.
+    required: true
+  evidence-directory:
+    description: Host directory receiving bounded health evidence.
+    required: true
+
+runs:
+  using: composite
+  steps:
+    - name: Exercise standard Check, Watch, and shutdown behavior
+      shell: bash
+      env:
+        ACTION_PATH: ${{ github.action_path }}
+        CONTAINER_NAME: ${{ inputs.container-name }}
+        EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}
+        SOCKET_PATH: ${{ inputs.socket-path }}
+        UNIT_NAME: ${{ inputs.unit-name }}
+      run: |
+        set -euo pipefail
+        mkdir -p "${EVIDENCE_DIRECTORY}"
+        verifier_directory="/opt/stateforge-provider-tests/health"
+        verifier="${verifier_directory}/assert_health.py"
+        docker exec "${CONTAINER_NAME}" mkdir -p "${verifier_directory}"
+        docker cp "${ACTION_PATH}/assert_health.py" \
+          "${CONTAINER_NAME}:${verifier}"
+        docker exec "${CONTAINER_NAME}" /usr/bin/python3 "${verifier}" \
+          --socket-path "${SOCKET_PATH}" \
+          --unit-name "${UNIT_NAME}" \
+          > "${EVIDENCE_DIRECTORY}/health.json"

.github/actions/provider/assert-health/assert_health.py

@@ -0,0 +1,181 @@
+"""Verify standard Provider Health behavior against an installed runtime."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import re
+import subprocess
+import sys
+
+import grpc
+from grpc_health.v1 import health_pb2, health_pb2_grpc
+
+_BOOTSTRAP_SERVICE = "stateforge.sfp.bootstrap.v1.BootstrapService"
+_MANIFEST_SERVICE = "stateforge.sfp.manifest.v1.ManifestService"
+
+
+async def _read_status(
+    watch: grpc.aio.UnaryStreamCall[
+        health_pb2.HealthCheckRequest,
+        health_pb2.HealthCheckResponse,
+    ],
+) -> str:
+    """Read one bounded Watch response and return its enum name."""
+    response = await asyncio.wait_for(watch.read(), timeout=10)
+    if response is grpc.aio.EOF:
+        raise RuntimeError("Health Watch ended before the expected transition")
+    return health_pb2.HealthCheckResponse.ServingStatus.Name(response.status)
+
+
+def _signal(unit_name: str, signal_name: str) -> None:
+    """Deliver one deterministic fixture transition through systemd."""
+    if not re.fullmatch(
+        r"stateforge-test-[a-z][a-z0-9-]{0,20}\.service",
+        unit_name,
+    ):
+        raise ValueError("unsafe Provider fixture unit name")
+    if signal_name not in {"SIGHUP", "SIGUSR1", "SIGUSR2"}:
+        raise ValueError("unsupported Provider fixture health signal")
+    # Both arguments are allow-listed above and the executable is absolute.
+    subprocess.run(  # noqa: S603
+        ["/usr/bin/systemctl", "kill", "--signal", signal_name, unit_name],
+        check=True,
+        timeout=10,
+    )
+
+
+def _stop(unit_name: str) -> None:
+    """Stop the fixture while its overall Health Watch remains active."""
+    if not re.fullmatch(
+        r"stateforge-test-[a-z][a-z0-9-]{0,20}\.service",
+        unit_name,
+    ):
+        raise ValueError("unsafe Provider fixture unit name")
+    subprocess.run(  # noqa: S603
+        ["/usr/bin/systemctl", "stop", unit_name],
+        check=True,
+        timeout=20,
+    )
+
+
+async def _assert_shutdown_transition(
+    watch: grpc.aio.UnaryStreamCall[
+        health_pb2.HealthCheckRequest,
+        health_pb2.HealthCheckResponse,
+    ],
+    unit_name: str,
+) -> str:
+    """Stop the Provider and require fail-closed Watch delivery."""
+    stop = asyncio.create_task(asyncio.to_thread(_stop, unit_name))
+    try:
+        observed = await _read_status(watch)
+    finally:
+        await stop
+    if observed != "NOT_SERVING":
+        raise RuntimeError(
+            f"Provider shutdown produced {observed}, expected NOT_SERVING"
+        )
+    return observed
+
+
+async def _verify(socket_path: str, unit_name: str) -> dict[str, object]:
+    """Exercise Check, Watch, author transitions, and hook failure."""
+    channel = grpc.aio.insecure_channel(f"unix:{socket_path}")
+    await asyncio.wait_for(channel.channel_ready(), timeout=10)
+    stub = health_pb2_grpc.HealthStub(channel)
+    checks: dict[str, str] = {}
+    transitions: list[str] = []
+    watch = stub.Watch(health_pb2.HealthCheckRequest(service=""))
+    unknown_watch = None
+    try:
+        for service in ("", _BOOTSTRAP_SERVICE, _MANIFEST_SERVICE):
+            response = await stub.Check(
+                health_pb2.HealthCheckRequest(service=service),
+                timeout=10,
+            )
+            name = health_pb2.HealthCheckResponse.ServingStatus.Name(
+                response.status
+            )
+            if name != "SERVING":
+                raise RuntimeError(f"{service!r} reported {name}")
+            checks[service or "<overall>"] = name
+
+        unknown_service = "stateforge.sfp.unknown.v1.UnknownService"
+        try:
+            await stub.Check(
+                health_pb2.HealthCheckRequest(service=unknown_service),
+                timeout=10,
+            )
+        except grpc.aio.AioRpcError as error:
+            if error.code() is not grpc.StatusCode.NOT_FOUND:
+                raise
+            checks[unknown_service] = error.code().name
+        else:
+            raise RuntimeError("unknown Health Check did not return NOT_FOUND")
+
+        unknown_watch = stub.Watch(
+            health_pb2.HealthCheckRequest(service=unknown_service)
+        )
+        unknown_initial = await _read_status(unknown_watch)
+        if unknown_initial != "SERVICE_UNKNOWN":
+            raise RuntimeError(
+                f"unknown Health Watch reported {unknown_initial}"
+            )
+
+        transitions.append(await _read_status(watch))
+        for signal_name, expected in (
+            ("SIGUSR1", "NOT_SERVING"),
+            ("SIGUSR2", "SERVING"),
+            ("SIGHUP", "NOT_SERVING"),
+            ("SIGUSR2", "SERVING"),
+        ):
+            _signal(unit_name, signal_name)
+            observed = await _read_status(watch)
+            if observed != expected:
+                raise RuntimeError(
+                    f"{signal_name} produced {observed}, expected {expected}"
+                )
+            transitions.append(observed)
+
+        transitions.append(await _assert_shutdown_transition(watch, unit_name))
+    finally:
+        watch.cancel()
+        if unknown_watch is not None:
+            unknown_watch.cancel()
+        await channel.close()
+
+    expected_transitions = [
+        "SERVING",
+        "NOT_SERVING",
+        "SERVING",
+        "NOT_SERVING",
+        "SERVING",
+        "NOT_SERVING",
+    ]
+    if transitions != expected_transitions:
+        raise RuntimeError(f"unexpected Health transitions: {transitions}")
+    return {
+        "checks": checks,
+        "unknownWatchInitial": unknown_initial,
+        "watchTransitions": transitions,
+    }
+
+
+def _arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--socket-path", required=True)
+    parser.add_argument("--unit-name", required=True)
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Run the installed-runtime assertion and emit bounded JSON evidence."""
+    arguments = _arguments()
+    evidence = asyncio.run(_verify(arguments.socket_path, arguments.unit_name))
+    sys.stdout.write(json.dumps(evidence, sort_keys=True) + "\n")
+
+
+if __name__ == "__main__":
+    main()

.github/actions/provider/create-fixture/fixture.py

@@ -3,11 +3,15 @@
 from __future__ import annotations
 
 import argparse
+import asyncio
+import contextlib
+import signal
 from pathlib import Path
 
 from pydantic import SecretStr
 
 from stateforge_provider import (
+    HealthStatus,
     Platform,
     Provider,
     ProviderModel,
@@ -72,15 +76,49 @@ def __init__(
             state_directory=state_directory,
         )
         self._event_log = event_log
+        self._health_status = HealthStatus.healthy()
+        self._fail_health = False
+        self._health_tasks: set[asyncio.Task[None]] = set()
 
     async def on_startup(self) -> None:
         """Record that author resources were initialized."""
+        loop = asyncio.get_running_loop()
+        loop.add_signal_handler(
+            signal.SIGUSR1,
+            self._schedule_health,
+            HealthStatus.maintenance(),
+            False,
+        )
+        loop.add_signal_handler(
+            signal.SIGUSR2,
+            self._schedule_health,
+            HealthStatus.healthy(),
+            False,
+        )
+        loop.add_signal_handler(
+            signal.SIGHUP,
+            self._schedule_health,
+            HealthStatus.not_serving(),
+            True,
+        )
         self._record("startup")
 
     async def on_shutdown(self) -> None:
         """Record that author resources were released."""
+        loop = asyncio.get_running_loop()
+        for signal_number in (signal.SIGUSR1, signal.SIGUSR2, signal.SIGHUP):
+            with contextlib.suppress(NotImplementedError, ValueError):
+                loop.remove_signal_handler(signal_number)
+        if self._health_tasks:
+            await asyncio.gather(*self._health_tasks, return_exceptions=True)
         self._record("shutdown")
 
+    async def health(self) -> HealthStatus:
+        """Report deterministic health controlled by scenario signals."""
+        if self._fail_health:
+            raise RuntimeError("fixture health hook failed")
+        return self._health_status
+
     @operation("stateforge.test/echo@1")
     async def echo(self, request: EchoInput) -> EchoResult:
         """Declare one typed operation without invoking it in R11."""
@@ -91,6 +129,29 @@ def _record(self, event: str) -> None:
         with self._event_log.open("a", encoding="utf-8") as stream:
             stream.write(f"{event}\n")
 
+    def _schedule_health(
+        self,
+        status: HealthStatus,
+        fail_health: bool,
+    ) -> None:
+        """Schedule one event-driven public health refresh."""
+        task = asyncio.create_task(
+            self._publish_health(status, fail_health),
+        )
+        self._health_tasks.add(task)
+        task.add_done_callback(self._health_tasks.discard)
+
+    async def _publish_health(
+        self,
+        status: HealthStatus,
+        fail_health: bool,
+    ) -> None:
+        """Update fixture evidence and invoke the public refresh API."""
+        self._health_status = status
+        self._fail_health = fail_health
+        accepted = await self.refresh_health()
+        self._record(f"health:{accepted.value}")
+
 
 def _arguments() -> argparse.Namespace:
     parser = argparse.ArgumentParser()

.github/actions/provider/stop/action.yml

@@ -41,7 +41,11 @@ runs:
       run: |
         set -euo pipefail
         mkdir -p "${EVIDENCE_DIRECTORY}"
-        docker exec "${CONTAINER_NAME}" systemctl stop "${UNIT_NAME}"
+        state="$(docker exec "${CONTAINER_NAME}" \
+          systemctl is-active "${UNIT_NAME}" 2>/dev/null || true)"
+        if [[ "${state}" != "inactive" && "${state}" != "failed" ]]; then
+          docker exec "${CONTAINER_NAME}" systemctl stop "${UNIT_NAME}"
+        fi
 
         deadline=$((SECONDS + TIMEOUT_SECONDS))
         while (( SECONDS < deadline )); do