diff --git a/.github/actions/scenarios/operation/admission/verify.py b/.github/actions/scenarios/operation/admission/verify.py index dd50757..42637a0 100644 --- a/.github/actions/scenarios/operation/admission/verify.py +++ b/.github/actions/scenarios/operation/admission/verify.py @@ -1,10 +1,11 @@ -"""Verify installed R5 behavior through an independent Protocol peer.""" +"""Verify R5 and R19 admission through an independent Protocol peer.""" from __future__ import annotations import argparse import asyncio import json +from dataclasses import replace from datetime import UTC, datetime, timedelta from importlib.metadata import version from pathlib import Path @@ -12,7 +13,7 @@ from uuid import uuid4 import grpc -from google.protobuf.json_format import ParseDict +from google.protobuf.json_format import ParseDict, ParseError from google.protobuf.struct_pb2 import Struct from google.protobuf.timestamp_pb2 import Timestamp from grpc_status import rpc_status @@ -40,6 +41,18 @@ operation, requirement_check, ) +from stateforge_provider._services.operation_admission.service import ( + OperationAdmissionService, +) + +_ACCEPTED_NEW = operation_admission_pb2.ADMISSION_DISPOSITION_ACCEPTED_NEW +_RESOLVED_EXISTING = ( + operation_admission_pb2.ADMISSION_DISPOSITION_RESOLVED_EXISTING +) +_RECORD_RETENTION_SECONDS = 0.2 +_TOMBSTONE_RETENTION_SECONDS = 0.2 +_PRUNE_GRACE_SECONDS = 0.05 +_EXPECTED_LOST_RESPONSE_ATTEMPTS = 2 class Input(ProviderModel): @@ -63,14 +76,28 @@ class AdmissionProvider(Provider): version = "0.1.0" platforms = (Platform("linux", "x86_64"),) - def __init__(self, runtime_directory: Path) -> None: + def __init__( + self, + runtime_directory: Path, + *, + short_retention: bool = False, + ) -> None: """Bind the fixture to scenario-owned runtime directories.""" super().__init__( discovery_directory=runtime_directory / "discovery", state_directory=runtime_directory / "state", ) + if short_retention: + self._configuration = replace( + self._configuration, + operation_record_retention_seconds=_RECORD_RETENTION_SECONDS, + operation_tombstone_retention_seconds=( + _TOMBSTONE_RETENTION_SECONDS + ), + ) self.requirement_satisfied = True self.dispatches = 0 + self.completed = asyncio.Event() self.started = asyncio.Event() self.release = asyncio.Event() self.context: OperationContext | None = None @@ -97,6 +124,7 @@ async def execute( self.context = context self.started.set() await self.release.wait() + self.completed.set() return Result(accepted=request.value > 0) @@ -115,10 +143,17 @@ def _request( secret: str | None = None, generation: int | None = None, roles: tuple[str, ...] = ("stateforge.test.v1.audited",), + payload_fields: dict[str, str | int | bool | None] | None = None, + mode: operation_admission_pb2.AdmissionMode.ValueType = ( + operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE + ), + deadline: datetime | None = None, ) -> operation_admission_pb2.StartOperationRequest: payload = Struct() ParseDict( - {"value": value, "secret": secret or "not-evidence"}, + payload_fields + if payload_fields is not None + else {"value": value, "secret": secret or "not-evidence"}, payload, ) execution_context = Struct() @@ -141,7 +176,9 @@ def _request( operation_identifier="stateforge.test/execute@1", payload=payload, execution_deadline=_timestamp( - datetime.now(UTC) + timedelta(minutes=5) + deadline + if deadline is not None + else datetime.now(UTC) + timedelta(minutes=5) ), contract=operation_admission_pb2.OperationContractContext( manifest_revision=manifest.manifest_revision, @@ -149,8 +186,312 @@ def _request( ), execution_context=execution_context, ), - admission_mode=operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE, + admission_mode=mode, + ) + + +async def _verify_canonical_rejections( + provider: AdmissionProvider, + admission: grpc.aio.UnaryUnaryMultiCallable[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + handshake: bootstrap_pb2.HandshakeResponse, + manifest: manifest_pb2.GetManifestResponse, +) -> dict[str, str | bool]: + """Replay canonical malformed, schema, and deadline rejections.""" + statuses: dict[str, str | bool] = {} + statuses["malformed_request_status"] = await _expect_error( + admission, + operation_admission_pb2.StartOperationRequest(), + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, ) + before = provider.dispatches + statuses["schema_payload_status"] = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=str(uuid4()), + payload_fields={"value": "one", "secret": "not-evidence"}, + ), + grpc.StatusCode.INVALID_ARGUMENT, + operation_admission_pb2.ADMISSION_ERROR_REASON_INVALID_OPERATION_REQUEST, + ) + array_rejected = False + try: + ParseDict(["not", "an", "object"], Struct()) + except (ParseError, ValueError): + array_rejected = True + if not array_rejected: + raise RuntimeError("array payload parsed into an envelope field") + statuses["array_payload_rejected_at_envelope"] = True + identity = str(uuid4()) + statuses["deadline_status"] = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=identity, + deadline=datetime.now(UTC) - timedelta(seconds=1), + ), + grpc.StatusCode.DEADLINE_EXCEEDED, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_DEADLINE_EXPIRED, + ) + statuses["deadline_record_absent_status"] = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=identity, + mode=operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY, + ), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + ) + if provider.dispatches != before: + raise RuntimeError("canonical rejection dispatched the operation") + return statuses + + +async def _verify_lost_response_replay( # noqa: PLR0915 + base_directory: Path, +) -> dict[str, object]: + """Lose one admitted response and prove identical replay resolves.""" + original = OperationAdmissionService.StartOperation + attempts: list[bytes] = [] + drop_once = True + + async def start_operation( + service: OperationAdmissionService, + request: operation_admission_pb2.StartOperationRequest, + context: grpc.aio.ServicerContext[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + ) -> operation_admission_pb2.StartOperationResponse: + nonlocal drop_once + attempts.append(request.SerializeToString(deterministic=True)) + response = await original(service, request, context) + if drop_once: + drop_once = False + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "Injected response loss after durable admission", + ) + return response + + OperationAdmissionService.StartOperation = start_operation + provider_discovery = base_directory / "discovery" + provider_discovery.mkdir(parents=True) + provider = AdmissionProvider(base_directory) + provider_task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + try: + discovery = base_directory / "discovery" + descriptor = await _descriptor(discovery, provider_task) + transport = descriptor["transport"] + if not isinstance(transport, dict): + raise RuntimeError("Provider descriptor transport is invalid") + socket_path = str(transport["socket_path"]) + channel = grpc.aio.insecure_channel(f"unix:{socket_path}") + await asyncio.wait_for(channel.channel_ready(), timeout=3) + handshake, manifest = await _session(channel, descriptor) + admission = operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ).StartOperation + request = _request(handshake, manifest, operation_id=str(uuid4())) + try: + await admission(request, timeout=3) + except grpc.aio.AioRpcError as error: + if error.code() is not grpc.StatusCode.UNAVAILABLE: + raise RuntimeError("wrong response-loss status") from error + else: + raise RuntimeError("first response was not lost") + await asyncio.wait_for(provider.started.wait(), timeout=2) + before = provider.dispatches + replayed = await admission(request, timeout=3) + validation.validate_admission_exchange(request, replayed) + if replayed.disposition != _RESOLVED_EXISTING: + raise RuntimeError("lost-response replay did not resolve existing") + if ( + len(attempts) != _EXPECTED_LOST_RESPONSE_ATTEMPTS + or attempts[0] != attempts[1] + ): + raise RuntimeError("lost-response replay changed request bytes") + if provider.dispatches != before: + raise RuntimeError("lost-response replay dispatched again") + return { + "request_fingerprint": replayed.record.request_fingerprint, + "lost_response_attempts": len(attempts), + "lost_response_replay_status": ( + operation_admission_pb2.AdmissionDisposition.Name( + replayed.disposition + ) + ), + "lost_response_replay_dispatches": provider.dispatches, + } + finally: + OperationAdmissionService.StartOperation = original + provider.request_shutdown() + if not provider_task.done(): + await asyncio.wait_for(provider_task, timeout=5) + if channel is not None: + await channel.close() + + +async def _connect_provider( + discovery_directory: Path, + provider_task: asyncio.Task[None], +) -> tuple[ + grpc.aio.Channel, + grpc.aio.UnaryUnaryMultiCallable[ + operation_admission_pb2.StartOperationRequest, + operation_admission_pb2.StartOperationResponse, + ], + bootstrap_pb2.HandshakeResponse, + manifest_pb2.GetManifestResponse, + dict[str, object], +]: + """Open one admission session to a serving fixture instance.""" + descriptor = await _descriptor(discovery_directory, provider_task) + transport = descriptor["transport"] + if not isinstance(transport, dict): + raise RuntimeError("Provider descriptor transport is invalid") + socket_path = str(transport["socket_path"]) + channel = grpc.aio.insecure_channel(f"unix:{socket_path}") + await asyncio.wait_for(channel.channel_ready(), timeout=3) + handshake, manifest = await _session(channel, descriptor) + admission = operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ).StartOperation + return channel, admission, handshake, manifest, descriptor + + +async def _verify_expiry_and_restart( # noqa: PLR0912, PLR0915 + base_directory: Path, +) -> dict[str, object]: + """Expire a retained record, reuse the identity, and restart.""" + provider_discovery = base_directory / "discovery" + provider_discovery.mkdir(parents=True) + provider: AdmissionProvider | None = None + provider_task: asyncio.Task[None] | None = None + restart_provider: AdmissionProvider | None = None + restart_task: asyncio.Task[None] | None = None + channel: grpc.aio.Channel | None = None + restart_channel: grpc.aio.Channel | None = None + try: + provider = AdmissionProvider(base_directory, short_retention=True) + provider_task = asyncio.create_task(provider.serve()) + ( + channel, + admission, + handshake, + manifest, + descriptor, + ) = await _connect_provider( + base_directory / "discovery", + provider_task, + ) + identity = str(uuid4()) + request = _request(handshake, manifest, operation_id=identity) + admitted = await admission(request, timeout=3) + validation.validate_admission_exchange(request, admitted) + if admitted.disposition != _ACCEPTED_NEW: + raise RuntimeError("expiry identity was not admitted") + await asyncio.wait_for(provider.started.wait(), timeout=2) + provider.release.set() + await asyncio.wait_for(provider.completed.wait(), timeout=2) + reuse_dispatches = provider.dispatches + provider.started.clear() + await asyncio.sleep(_RECORD_RETENTION_SECONDS + _PRUNE_GRACE_SECONDS) + expired_status = await _expect_error( + admission, + _request( + handshake, + manifest, + operation_id=identity, + mode=operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY, + ), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_EXPIRED, + ) + await asyncio.sleep( + _TOMBSTONE_RETENTION_SECONDS + _PRUNE_GRACE_SECONDS + ) + reused = await admission(request, timeout=3) + validation.validate_admission_exchange(request, reused) + if reused.disposition != _ACCEPTED_NEW: + raise RuntimeError("expiry identity reuse did not re-admit") + await asyncio.wait_for(provider.started.wait(), timeout=2) + if provider.dispatches != reuse_dispatches + 1: + raise RuntimeError("expiry identity reuse dispatched unexpectedly") + provider.request_shutdown() + await asyncio.wait_for(provider_task, timeout=5) + restart_provider = AdmissionProvider( + base_directory, short_retention=True + ) + restart_task = asyncio.create_task(restart_provider.serve()) + ( + restart_channel, + restart_admission, + restart_handshake, + restart_manifest, + restart_descriptor, + ) = await _connect_provider( + base_directory / "discovery", + restart_task, + ) + if int(restart_descriptor["endpoint_generation"]) <= int( + descriptor["endpoint_generation"] + ): + raise RuntimeError("restart did not allocate a newer generation") + unavailable_status = await _expect_error( + restart_admission, + _request( + restart_handshake, + restart_manifest, + operation_id=identity, + mode=operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY, + ), + grpc.StatusCode.FAILED_PRECONDITION, + operation_admission_pb2.ADMISSION_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + ) + if restart_provider.dispatches != 0: + raise RuntimeError("restart resolve dispatched the operation") + transport = restart_descriptor["transport"] + if not isinstance(transport, dict): + raise RuntimeError("restart descriptor transport is invalid") + restart_socket = str(transport["socket_path"]) + restart_provider.request_shutdown() + await asyncio.wait_for(restart_task, timeout=5) + if await asyncio.to_thread(Path(restart_socket).exists): + raise RuntimeError("restart socket survived shutdown") + return { + "record_expiry_status": expired_status, + "expiry_identity_reuse_status": ( + operation_admission_pb2.AdmissionDisposition.Name( + reused.disposition + ) + ), + "expiry_identity_reuse_dispatches": provider.dispatches, + "restart_unavailable_status": unavailable_status, + "restart_resolve_dispatches": restart_provider.dispatches, + } + finally: + if provider is not None: + provider.request_shutdown() + if provider_task is not None and not provider_task.done(): + await asyncio.wait_for(provider_task, timeout=5) + if restart_provider is not None: + restart_provider.request_shutdown() + if restart_task is not None and not restart_task.done(): + await asyncio.wait_for(restart_task, timeout=5) + if channel is not None: + await channel.close() + if restart_channel is not None: + await restart_channel.close() async def _descriptor( @@ -238,7 +579,7 @@ async def _expect_error( ) -async def _run( # noqa: PLR0915 +async def _run( # noqa: PLR0912, PLR0915 runtime_directory: Path, ) -> dict[str, object]: discovery = runtime_directory / "discovery" @@ -324,6 +665,27 @@ async def _run( # noqa: PLR0915 operation_admission_pb2.ADMISSION_ERROR_REASON_IDEMPOTENCY_CONFLICT, ) provider.release.set() + await asyncio.wait_for(provider.completed.wait(), timeout=2) + terminal_request = operation_admission_pb2.StartOperationRequest() + terminal_request.CopyFrom(request) + terminal_request.admission_mode = ( + operation_admission_pb2.ADMISSION_MODE_RESOLVE_ONLY + ) + terminal_response = await admission(terminal_request, timeout=3) + validation.validate_admission_exchange( + terminal_request, + terminal_response, + ) + if terminal_response.disposition != _RESOLVED_EXISTING: + raise RuntimeError("terminal record replay did not resolve") + if provider.dispatches != 1: + raise RuntimeError("terminal record replay dispatched again") + statuses = await _verify_canonical_rejections( + provider, + admission, + handshake, + manifest, + ) context = provider.context if context is None or not isinstance( @@ -341,6 +703,9 @@ async def _run( # noqa: PLR0915 if remaining_descriptors: raise RuntimeError("Provider descriptor survived shutdown") + lost = await _verify_lost_response_replay(runtime_directory / "lost") + expiry = await _verify_expiry_and_restart(runtime_directory / "expiry") + evidence = { "installed_artifacts": { "stateforge-provider": version("stateforge-provider"), @@ -364,6 +729,9 @@ async def _run( # noqa: PLR0915 "final_process_state": "stopped", "descriptor_removed": True, "socket_removed": True, + **statuses, + **lost, + **expiry, } serialized = json.dumps(evidence, sort_keys=True) if payload_marker in serialized: diff --git a/.github/actions/scenarios/operation/installed-agent/action.yml b/.github/actions/scenarios/operation/installed-agent/action.yml new file mode 100644 index 0000000..8b5870e --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/action.yml @@ -0,0 +1,170 @@ +name: Verify Operation Admission with the installed Agent +description: Exercise the exact Provider and Protocol wheels through the Agent RPM. + +inputs: + agent-artifact-directory: + description: Directory containing the exact Agent RPM build. + required: true + agent-test-image-directory: + description: Bundled exact Agent Fedora test-image definition. + required: true + provider-artifact-directory: + description: Directory containing the exact Provider wheel. + required: true + protocol-artifact-directory: + description: Directory containing the exact Protocol wheel. + required: true + runtime-directory: + description: Scenario-owned directory under the runner temp directory. + required: true + evidence-directory: + description: Directory receiving scenario evidence. + required: true + +runs: + using: composite + steps: + - name: Install exact runtime artifacts + id: system + uses: ./.github/actions/agent/install + with: + agent-artifact-directory: ${{ inputs.agent-artifact-directory }} + agent-test-image-directory: ${{ inputs.agent-test-image-directory }} + provider-artifact-directory: ${{ inputs.provider-artifact-directory }} + protocol-artifact-directory: ${{ inputs.protocol-artifact-directory }} + evidence-directory: ${{ inputs.evidence-directory }}/installation + + - name: Start installed Agent + uses: ./.github/actions/agent/start + with: + container-name: ${{ steps.system.outputs.container-name }} + + - name: Stage operation Provider fixture + id: fixture + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + RUNTIME_DIRECTORY: ${{ inputs.runtime-directory }} + run: | + set -euo pipefail + root="${RUNTIME_DIRECTORY}/admission-provider" + mkdir -p "${root}" + install -m 0644 "${ACTION_PATH}/fixture.py" "${root}/provider.py" + { + echo "fixture-path=${root}/provider.py" + echo "instance-name=admission-provider" + echo "state-name=admission-provider" + } >> "${GITHUB_OUTPUT}" + + - name: Start real Provider fixture + id: provider + uses: ./.github/actions/provider/start + with: + container-name: ${{ steps.system.outputs.container-name }} + fixture-path: ${{ steps.fixture.outputs.fixture-path }} + instance-name: ${{ steps.fixture.outputs.instance-name }} + state-name: ${{ steps.fixture.outputs.state-name }} + identity: stateforge.test.operation_admission + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Assert Agent-accepted session + uses: ./.github/actions/agent/assert-session + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + endpoint-generation: ${{ steps.provider.outputs.endpoint-generation }} + evidence-directory: ${{ inputs.evidence-directory }}/agent + + - name: Exercise admission through installed Agent requester code + id: assertion + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/admission + PROVIDER_INSTANCE_ID: ${{ steps.provider.outputs.provider-instance-id }} + STATE_DIRECTORY: ${{ steps.provider.outputs.state-directory }} + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + mkdir -p "${EVIDENCE_DIRECTORY}" + verifier=/opt/stateforge-provider-tests/admission-assert.py + docker cp "${ACTION_PATH}/assert.py" "${CONTAINER_NAME}:${verifier}" + docker exec "${CONTAINER_NAME}" /usr/bin/python3 "${verifier}" \ + --provider-instance-id "${PROVIDER_INSTANCE_ID}" \ + --state-directory "${STATE_DIRECTORY}" \ + --unit-name "${UNIT_NAME}" \ + > "${EVIDENCE_DIRECTORY}/operation-admission.json" + jq -e '.dispatches == 6 and (.dispatch_operation_ids | length) == 6 + and .uncertain_requester_pid != .resolving_requester_pid + and .lost_response_attempts == 2 + and .replacement_endpoint_generation > .initial_endpoint_generation + and .concurrent_requests == 20 and .concurrent_accepted_new == 1 + and .schema_rejected and .deadline_rejected + and .readiness_rejected_and_recovered + and .requester_rebuild_resolved == "resolved_existing" + and .provider_restart_indeterminate and .expired_record_rejected + and .stale_generation_rejected and .stale_session_rejected + and .stale_manifest_rejected and .changed_content_rejected + and .agent_service_restarted and .target_unchanged' \ + "${EVIDENCE_DIRECTORY}/operation-admission.json" >/dev/null + socket="$(jq -r '.final_socket_path' \ + "${EVIDENCE_DIRECTORY}/operation-admission.json")" + echo "final-socket-path=${socket}" >> "${GITHUB_OUTPUT}" + + - name: Verify Provider shutdown cleanup + uses: ./.github/actions/provider/stop + with: + container-name: ${{ steps.system.outputs.container-name }} + unit-name: ${{ steps.provider.outputs.unit-name }} + descriptor-path: ${{ steps.provider.outputs.descriptor-path }} + socket-path: ${{ steps.assertion.outputs.final-socket-path }} + event-log: ${{ steps.provider.outputs.event-log }} + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Preserve admission correlation evidence + if: ${{ always() }} + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/admission + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + STATE_DIRECTORY: ${{ steps.provider.outputs.state-directory }} + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + mkdir -p "${EVIDENCE_DIRECTORY}" + if [[ -z "${CONTAINER_NAME}" ]]; then + exit 0 + fi + docker exec "${CONTAINER_NAME}" cat "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/operation-events.txt" 2>&1 || true + docker exec "${CONTAINER_NAME}" cat \ + "${STATE_DIRECTORY}/identity.json" \ + > "${EVIDENCE_DIRECTORY}/provider-identity.json" 2>&1 || true + docker exec "${CONTAINER_NAME}" systemctl show "${UNIT_NAME}" \ + --property InvocationID,MainPID,ActiveState \ + > "${EVIDENCE_DIRECTORY}/provider-process.txt" 2>&1 || true + + - name: Preserve bounded evidence and clean up + if: ${{ always() }} + uses: ./.github/actions/evidence/collect + with: + stage: operation-admission-agent + container-name: ${{ steps.system.outputs.container-name }} + runtime-directory: ${{ inputs.runtime-directory }} + evidence-directory: ${{ inputs.evidence-directory }} + + - name: Reject sensitive operation evidence + if: ${{ always() }} + shell: bash + env: + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }} + run: | + set -euo pipefail + if grep --recursive --fixed-strings --line-number \ + 'stateforge-r19-secret-never-in-evidence' \ + "${EVIDENCE_DIRECTORY}"; then + echo 'Sensitive operation content reached evidence' >&2 + exit 1 + fi diff --git a/.github/actions/scenarios/operation/installed-agent/assert.py b/.github/actions/scenarios/operation/installed-agent/assert.py new file mode 100644 index 0000000..5023f71 --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/assert.py @@ -0,0 +1,743 @@ +"""Verify real RPM-installed Agent admission against a Provider process.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import os +import sys +import time +from collections import Counter +from datetime import UTC, datetime, timedelta +from importlib.metadata import version +from pathlib import Path +from uuid import UUID, uuid4 + +from google.protobuf.json_format import ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from stateforge_agent._journal import SqliteJournal +from stateforge_agent._sfp import ( + OperationAdmissionDisposition, + OperationAdmissionRequest, + ProviderAvailabilityState, + ProviderSessionManager, + SfpVersion, +) +from stateforge_agent._sfp._runtime.descriptor import ( + ProviderEndpointDescriptor, + load_provider_descriptor, +) +from stateforge_agent._sfp._services.capability_assessment.client import ( + SfpCapabilityAssessmentUnavailableError, +) +from stateforge_agent._sfp._services.operation_admission.client import ( + OperationAdmissionClient, + SfpOperationAdmissionRejectedError, +) +from stateforge_agent._sfp._services.operation_admission.coordinator import ( + OperationAdmissionCoordinator, + OperationAdmissionPreconditionError, + OperationAdmissionUnresolvedError, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2 as admission_pb2, +) + +_DISCOVERY = Path("/run/stateforge-agent/providers") +_JOURNAL = Path("/var/lib/stateforge-agent/journal/agent.db") +_OPERATION = "stateforge.test/inspect@1" +_ATTEMPT_PREFIX = "operation-admission-attempt:" +_SECRET = "stateforge-r19-secret-never-in-evidence" # noqa: S105 +_EXPECTED_RETRY_ATTEMPTS = 2 + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--provider-instance-id", type=UUID, required=True) + parser.add_argument("--state-directory", type=Path, required=True) + parser.add_argument("--unit-name", required=True) + parser.add_argument("--prepare-uncertain", action="store_true") + return parser.parse_args() + + +def _descriptor(provider_id: UUID) -> ProviderEndpointDescriptor: + return load_provider_descriptor( + _DISCOVERY / f"{provider_id}.sfp.json", + discovery_directory=_DISCOVERY, + ) + + +def _events(path: Path) -> tuple[str, ...]: + try: + return tuple(path.read_text(encoding="utf-8").splitlines()) + except FileNotFoundError: + return () + + +def _dispatch_ids(path: Path) -> tuple[UUID, ...]: + return tuple( + UUID(event.removeprefix("operation:inspect:")) + for event in _events(path) + if event.startswith("operation:inspect:") + ) + + +def _wait_for_dispatches_sync(path: Path, expected: int) -> None: + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if len(_dispatch_ids(path)) == expected: + return + time.sleep(0.05) + raise AssertionError(f"Expected exactly {expected} Provider dispatches") + + +async def _wait_for_dispatches(path: Path, expected: int) -> None: + await asyncio.to_thread(_wait_for_dispatches_sync, path, expected) + + +async def _wait_for_availability( + manager: ProviderSessionManager, + provider_id: UUID, + expected: ProviderAvailabilityState, +) -> None: + async with asyncio.timeout(15): + while True: + status = manager.availability_status(provider_id) + if status is not None and status.state is expected: + return + await asyncio.sleep(0.05) + + +async def _manager( + journal: SqliteJournal, + *, + attempts: int = 2, +) -> tuple[ProviderSessionManager, OperationAdmissionCoordinator]: + store = journal.operation_admissions(maximum_unresolved_attempts=32) + await store.initialize() + coordinator = OperationAdmissionCoordinator( + store, + attempts=attempts, + retry_initial_delay_seconds=0.05, + ) + manager = ProviderSessionManager( + agent_runtime_id=uuid4(), + agent_software_version=version("stateforge-agent"), + supported_sfp_versions=(SfpVersion(major=1, minor=0),), + capability_assessment_retry_initial_delay_seconds=0.05, + operation_admission=coordinator, + ) + return manager, coordinator + + +def _request( + provider_id: UUID, + label: str, + *, + deadline: datetime | None = None, +) -> OperationAdmissionRequest: + return OperationAdmissionRequest( + provider_instance_id=provider_id, + operation_identifier=_OPERATION, + payload={"value": label}, + execution_deadline=deadline + or datetime.now(UTC) + timedelta(minutes=5), + execution_context={"scenario": "provider-r19"}, + ) + + +def _wire_request( + *, + provider_id: UUID, + generation: int, + session_id: str, + manifest_revision: str, + operation_id: UUID, +) -> admission_pb2.StartOperationRequest: + payload = Struct() + ParseDict({"value": _SECRET}, payload) + context = Struct() + ParseDict({"scenario": "provider-r19"}, context) + deadline = Timestamp() + deadline.FromDatetime(datetime.now(UTC) + timedelta(minutes=5)) + return admission_pb2.StartOperationRequest( + session_binding=admission_pb2.OperationSessionBinding( + expected_provider_instance_id=str(provider_id), + expected_endpoint_generation=generation, + expected_session_id=session_id, + ), + invocation=admission_pb2.OperationInvocation( + operation_id=str(operation_id), + operation_identifier=_OPERATION, + payload=payload, + execution_deadline=deadline, + contract=admission_pb2.OperationContractContext( + manifest_revision=manifest_revision, + ), + execution_context=context, + ), + admission_mode=admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE, + ) + + +async def _restart(unit_name: str) -> None: + process = await asyncio.create_subprocess_exec( + "/usr/bin/systemctl", "restart", unit_name + ) + if await process.wait() != 0: + raise AssertionError(f"Failed to restart {unit_name}") + + +async def _signal(unit_name: str, name: str) -> None: + process = await asyncio.create_subprocess_exec( + "/usr/bin/systemctl", + "kill", + "--kill-whom=main", + f"--signal={name}", + unit_name, + ) + if await process.wait() != 0: + raise AssertionError(f"Failed to send {name} to {unit_name}") + + +async def _uncertain_admission( + manager: ProviderSessionManager, + coordinator: OperationAdmissionCoordinator, + provider_id: UUID, + state: Path, + label: str, +) -> UUID: + """Lose one response and retain an uncertain Agent intent.""" + request = _request(provider_id, label) + assessment = await manager.assess_capabilities( + provider_id, + (_OPERATION,), + execution_context=request.execution_context, + ) + lease = await manager._operation_admission_lease(provider_id) + operation_id = await coordinator.prepare( + lease, + request, + assessment, + assert_current=manager._assert_operation_admission_is_current, + ) + (state / "drop-next-operation-response").touch() + try: + await coordinator.submit( + lease, + operation_id, + assessment, + assert_current=manager._assert_operation_admission_is_current, + ) + except OperationAdmissionUnresolvedError as error: + if error.operation_id != operation_id: + raise AssertionError( + "Uncertainty changed operation identity" + ) from error + else: + raise AssertionError("One-attempt response loss was not unresolved") + attempt = await coordinator.attempt(provider_id, operation_id) + if ( + attempt is None + or attempt.attempt_count != 1 + or attempt.result is not None + ): + raise AssertionError("Uncertain identity was not durably retained") + return operation_id + + +async def _prepare_uncertain_process( + arguments: argparse.Namespace, +) -> dict[str, str | int]: + """Leave one unresolved intent in the installed Agent journal.""" + journal = SqliteJournal(_JOURNAL) + await journal.open() + manager, coordinator = await _manager(journal, attempts=1) + try: + provider_id = arguments.provider_instance_id + await manager.establish(_descriptor(provider_id)) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + operation_id = await _uncertain_admission( + manager, + coordinator, + provider_id, + arguments.state_directory, + "requester-rebuild", + ) + return { + "operation_id": str(operation_id), + "requester_pid": os.getpid(), + } + finally: + await manager.close() + await journal.close() + + +async def _spawn_uncertain_requester( + arguments: argparse.Namespace, +) -> tuple[UUID, int]: + """Run the unresolved admission in a distinct installed-RPM process.""" + process = await asyncio.create_subprocess_exec( + sys.executable, + "-I", + str(Path(__file__)), + "--provider-instance-id", + str(arguments.provider_instance_id), + "--state-directory", + str(arguments.state_directory), + "--unit-name", + arguments.unit_name, + "--prepare-uncertain", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + output, error = await asyncio.wait_for( + process.communicate(), timeout=30 + ) + except TimeoutError as timeout_error: + process.kill() + await process.wait() + raise AssertionError( + "Uncertain requester did not exit" + ) from timeout_error + if process.returncode != 0: + raise AssertionError( + "Uncertain requester failed: " + + error.decode("utf-8", errors="replace")[-2000:] + ) + checkpoint = json.loads(output) + requester_pid = checkpoint["requester_pid"] + if requester_pid != process.pid or requester_pid == os.getpid(): + raise AssertionError( + "Uncertain request did not cross a process boundary" + ) + return UUID(checkpoint["operation_id"]), requester_pid + + +async def _replacement( + provider_id: UUID, previous_generation: int +) -> ProviderEndpointDescriptor: + async with asyncio.timeout(15): + while True: + try: + descriptor = _descriptor(provider_id) + except (FileNotFoundError, ValueError): + await asyncio.sleep(0.05) + continue + if descriptor.endpoint_generation > previous_generation: + return descriptor + await asyncio.sleep(0.05) + + +async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR0912, PLR0915 + provider_id: UUID = arguments.provider_instance_id + state: Path = arguments.state_directory + event_log = ( + Path("/var/log/stateforge-provider-tests") / state.name / "events.log" + ) + initial = _descriptor(provider_id) + target_path = state / "target-state.txt" + target_digest = hashlib.sha256(target_path.read_bytes()).hexdigest() + journal = SqliteJournal(_JOURNAL) + await journal.open() + manager, coordinator = await _manager(journal) + try: + await manager.establish(initial) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + initial_dispatches = _dispatch_ids(event_log) + initial_count = len(initial_dispatches) + accepted = await manager.admit_operation( + _request(provider_id, "first") + ) + if ( + accepted.disposition + is not OperationAdmissionDisposition.ACCEPTED_NEW + ): + raise AssertionError("Agent did not accept new operation") + await _wait_for_dispatches(event_log, initial_count + 1) + cached = await manager.resolve_operation( + provider_id, accepted.operation_id + ) + if cached != accepted: + raise AssertionError("Agent did not retain its admission result") + + # The installed Agent client, not a substitute peer, sends these calls. + lease = await manager._operation_admission_lease(provider_id) + client = OperationAdmissionClient(timeout_seconds=3) + wire = _wire_request( + provider_id=provider_id, + generation=lease.endpoint_generation, + session_id=lease.session_id, + manifest_revision=lease.manifest_revision, + operation_id=uuid4(), + ) + replies = await asyncio.gather( + *(client.start(lease.channel, wire) for _ in range(20)) + ) + if ( + sum( + r.disposition is OperationAdmissionDisposition.ACCEPTED_NEW + for r in replies + ) + != 1 + ): + raise AssertionError("Concurrent admission had multiple winners") + await _wait_for_dispatches(event_log, initial_count + 2) + if len(_dispatch_ids(event_log)) != initial_count + 2: + raise AssertionError("Concurrent admission dispatched twice") + replay = await client.start(lease.channel, wire) + if ( + replay.disposition + is not OperationAdmissionDisposition.RESOLVED_EXISTING + ): + raise AssertionError("Provider did not resolve retained admission") + if replay.request_fingerprint != replies[0].request_fingerprint: + raise AssertionError("Resolved record changed fingerprint") + changed = admission_pb2.StartOperationRequest() + changed.CopyFrom(wire) + changed.invocation.payload.fields["value"].string_value = "changed" + try: + await client.start(lease.channel, changed) + except SfpOperationAdmissionRejectedError as error: + if "IDEMPOTENCY_CONFLICT" not in error.rejection.reason: + raise AssertionError( + "Changed content was not a conflict" + ) from error + else: + raise AssertionError("Changed content was accepted") + + before_rejections = len(_dispatch_ids(event_log)) + malformed = admission_pb2.StartOperationRequest() + malformed.CopyFrom(wire) + malformed.invocation.operation_id = str(uuid4()) + malformed.invocation.payload.Clear() + ParseDict( + {"unknown": "not-a-declared-input"}, malformed.invocation.payload + ) + try: + await client.start(lease.channel, malformed) + except SfpOperationAdmissionRejectedError as error: + if "INVALID_OPERATION_REQUEST" not in error.rejection.reason: + raise AssertionError( + "Schema rejection had the wrong reason" + ) from error + else: + raise AssertionError( + "Provider accepted an invalid operation payload" + ) + try: + await manager.admit_operation( + _request( + provider_id, + "past", + deadline=datetime.now(UTC) - timedelta(seconds=1), + ) + ) + except SfpOperationAdmissionRejectedError as error: + if "OPERATION_DEADLINE_EXPIRED" not in error.rejection.reason: + raise AssertionError( + "Deadline rejection had the wrong reason" + ) from error + else: + raise AssertionError("Agent admitted an expired operation") + await _signal(arguments.unit_name, "SIGUSR1") + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.NOT_SERVING + ) + try: + await manager.admit_operation(_request(provider_id, "not-ready")) + except SfpCapabilityAssessmentUnavailableError: + pass + else: + raise AssertionError( + "Agent admitted while Provider was not serving" + ) + try: + await manager._operation_admission_lease(provider_id) + except OperationAdmissionPreconditionError: + pass + else: + raise AssertionError( + "Agent exposed an admission lease while not serving" + ) + await _signal(arguments.unit_name, "SIGUSR2") + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + if len(_dispatch_ids(event_log)) != before_rejections: + raise AssertionError("Rejected work reached the author operation") + + before_retry = tuple( + line + for line in _events(event_log) + if line.startswith(_ATTEMPT_PREFIX) + ) + (state / "drop-next-operation-response").touch() + recovered = await manager.admit_operation( + _request(provider_id, "retry") + ) + if ( + recovered.disposition + is not OperationAdmissionDisposition.RESOLVED_EXISTING + ): + raise AssertionError("Lost response did not resolve existing") + await _wait_for_dispatches(event_log, initial_count + 3) + retry_attempt = await coordinator.attempt( + provider_id, recovered.operation_id + ) + if ( + retry_attempt is None + or retry_attempt.attempt_count != _EXPECTED_RETRY_ATTEMPTS + ): + raise AssertionError("Agent did not retry the same identity") + after_retry = tuple( + line + for line in _events(event_log) + if line.startswith(_ATTEMPT_PREFIX) + ) + delta = after_retry[len(before_retry) :] + if len(delta) != _EXPECTED_RETRY_ATTEMPTS or delta[0] != delta[1]: + raise AssertionError("Agent retry changed immutable request bytes") + if "operation-admission-response:dropped" not in _events(event_log): + raise AssertionError("Fault hook did not discard a response") + + await manager.close() + await journal.close() + rebuilt_id, uncertain_requester_pid = await _spawn_uncertain_requester( + arguments + ) + await _wait_for_dispatches(event_log, initial_count + 4) + journal = SqliteJournal(_JOURNAL) + await journal.open() + manager, coordinator = await _manager(journal) + await manager.establish(initial) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + rebuilt = await manager.resolve_operation(provider_id, rebuilt_id) + if ( + rebuilt.disposition + is not OperationAdmissionDisposition.RESOLVED_EXISTING + ): + raise AssertionError( + "Rebuilt Agent did not resolve Provider record" + ) + if len(_dispatch_ids(event_log)) != initial_count + 4: + raise AssertionError("Requester rebuild re-dispatched work") + + await manager.close() + manager, coordinator = await _manager(journal, attempts=1) + await manager.establish(initial) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + lost_id = await _uncertain_admission( + manager, coordinator, provider_id, state, "provider-restart" + ) + await _wait_for_dispatches(event_log, initial_count + 5) + + (state / "short-operation-retention").touch() + await _restart(arguments.unit_name) + replacement = await _replacement( + provider_id, initial.endpoint_generation + ) + await manager.establish(replacement) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + try: + await manager.resolve_operation(provider_id, lost_id) + except OperationAdmissionUnresolvedError as error: + if "OPERATION_RECORD_UNAVAILABLE" not in error.reason: + raise AssertionError( + "Restart had the wrong indeterminate reason" + ) from error + else: + raise AssertionError( + "Provider restart blindly re-admitted old work" + ) + if len(_dispatch_ids(event_log)) != initial_count + 5: + raise AssertionError("Indeterminate resolution re-dispatched work") + new_lease = await manager._operation_admission_lease(provider_id) + old_revision = new_lease.manifest_revision + stale_revision = old_revision[:-1] + ( + "0" if old_revision[-1] != "0" else "1" + ) + for name, generation, session_id, revision in ( + ( + "generation", + initial.endpoint_generation, + new_lease.session_id, + new_lease.manifest_revision, + ), + ( + "session", + replacement.endpoint_generation, + wire.session_binding.expected_session_id, + new_lease.manifest_revision, + ), + ( + "manifest", + replacement.endpoint_generation, + new_lease.session_id, + stale_revision, + ), + ): + stale_request = _wire_request( + provider_id=provider_id, + generation=generation, + session_id=session_id, + manifest_revision=revision, + operation_id=uuid4(), + ) + try: + await client.start(new_lease.channel, stale_request) + except SfpOperationAdmissionRejectedError as error: + if ( + "OPERATION_PRECONDITION_FAILED" + not in error.rejection.reason + ): + raise AssertionError( + f"Stale {name} had the wrong reason" + ) from error + else: + raise AssertionError(f"Stale {name} was accepted") + if len(_dispatch_ids(event_log)) != initial_count + 5: + raise AssertionError( + "Stale fence request reached author operation" + ) + fresh = await manager.admit_operation( + _request(provider_id, "replacement") + ) + if fresh.disposition is not OperationAdmissionDisposition.ACCEPTED_NEW: + raise AssertionError("Replacement generation did not admit work") + await _wait_for_dispatches(event_log, initial_count + 6) + fresh_attempt = await coordinator.attempt( + provider_id, fresh.operation_id + ) + if fresh_attempt is None: + raise AssertionError( + "Fresh identity was not retained by the Agent" + ) + await asyncio.sleep(1) + expired = admission_pb2.StartOperationRequest() + expired.CopyFrom( + _wire_request( + provider_id=provider_id, + generation=replacement.endpoint_generation, + session_id=new_lease.session_id, + manifest_revision=new_lease.manifest_revision, + operation_id=fresh.operation_id, + ) + ) + expired.invocation.CopyFrom(fresh_attempt.invocation) + expired.admission_mode = admission_pb2.ADMISSION_MODE_RESOLVE_ONLY + try: + await client.start(new_lease.channel, expired) + except SfpOperationAdmissionRejectedError as error: + if "OPERATION_RECORD_EXPIRED" not in error.rejection.reason: + raise AssertionError( + "Expired record had the wrong reason" + ) from error + else: + raise AssertionError("Expired Provider record was resolved") + if len(_dispatch_ids(event_log)) != initial_count + 6: + raise AssertionError("Expired identity re-dispatched work") + expected_dispatches = ( + accepted.operation_id, + UUID(wire.invocation.operation_id), + recovered.operation_id, + rebuilt_id, + lost_id, + fresh.operation_id, + ) + observed_dispatches = _dispatch_ids(event_log)[initial_count:] + if len(set(expected_dispatches)) != len(expected_dispatches): + raise AssertionError("Scenario reused an operation identity") + if Counter(observed_dispatches) != Counter(expected_dispatches): + raise AssertionError( + "Dispatches did not match admitted identities" + ) + await _restart("stateforge-agent.service") + agent_state = await asyncio.create_subprocess_exec( + "/usr/bin/systemctl", + "is-active", + "--quiet", + "stateforge-agent.service", + ) + if await agent_state.wait() != 0: + raise AssertionError("Installed Agent did not restart cleanly") + if ( + hashlib.sha256(target_path.read_bytes()).hexdigest() + != target_digest + ): + raise AssertionError("Provider operation changed fixture target") + evidence = { + "agent_version": version("stateforge-agent"), + "protocol_version": version("stateforge-protocol"), + "provider_version": version("stateforge-provider"), + "provider_instance_id": str(provider_id), + "initial_endpoint_generation": initial.endpoint_generation, + "replacement_endpoint_generation": replacement.endpoint_generation, + "final_socket_path": str(replacement.socket_path), + "accepted_new_operation_id": str(accepted.operation_id), + "accepted_new_fingerprint": accepted.request_fingerprint, + "resolved_existing": replay.disposition.value, + "concurrent_requests": len(replies), + "concurrent_accepted_new": 1, + "lost_response_recovered": recovered.disposition.value, + "lost_response_attempts": retry_attempt.attempt_count, + "requester_rebuild_resolved": rebuilt.disposition.value, + "uncertain_requester_pid": uncertain_requester_pid, + "resolving_requester_pid": os.getpid(), + "provider_restart_indeterminate": True, + "expired_record_rejected": True, + "schema_rejected": True, + "deadline_rejected": True, + "readiness_rejected_and_recovered": True, + "agent_service_restarted": True, + "stale_generation_rejected": True, + "stale_session_rejected": True, + "stale_manifest_rejected": True, + "changed_content_rejected": True, + "dispatches": len(observed_dispatches), + "dispatch_operation_ids": [ + str(operation_id) for operation_id in observed_dispatches + ], + "target_unchanged": True, + } + if _SECRET in json.dumps(evidence): + raise AssertionError("Sensitive operation content leaked") + return evidence + finally: + await manager.close() + await journal.close() + + +def main() -> None: + """Emit only bounded, secret-safe JSON evidence to stdout.""" + arguments = _arguments() + sys.stdout.write( + json.dumps( + asyncio.run( + _prepare_uncertain_process(arguments) + if arguments.prepare_uncertain + else _run(arguments) + ), + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +if __name__ == "__main__": + main() diff --git a/.github/actions/scenarios/operation/installed-agent/fixture.py b/.github/actions/scenarios/operation/installed-agent/fixture.py new file mode 100644 index 0000000..7d17870 --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/fixture.py @@ -0,0 +1,174 @@ +"""Operation-capable Provider fixture with a narrow response-loss hook.""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import hashlib +import signal +from dataclasses import replace +from pathlib import Path + +import grpc +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2 as admission_pb2, +) + +from stateforge_provider import ( + HealthStatus, + OperationContext, + Platform, + Provider, + ProviderModel, + operation, +) +from stateforge_provider._services.operation_admission.service import ( + OperationAdmissionService, +) + + +class Input(ProviderModel): + """One read-only fixture input.""" + + value: str + + +class Result(ProviderModel): + """One read-only fixture result.""" + + value: str + + +class AdmissionFixture(Provider): + """Expose the real Provider author API and Health service to the Agent.""" + + identity = "stateforge.test.operation_admission" + display_name = "StateForge Admission CI Provider" + version = "0.1.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__( + self, + *, + discovery_directory: Path, + state_directory: Path, + event_log: Path, + ) -> None: + """Bind only scenario-owned paths to the fixture.""" + super().__init__( + discovery_directory=discovery_directory, + state_directory=state_directory, + ) + if (state_directory / "short-operation-retention").exists(): + self._configuration = replace( + self._configuration, + operation_record_retention_seconds=0.5, + operation_tombstone_retention_seconds=30.0, + ) + self._event_log = event_log + self._health = HealthStatus.healthy() + self._health_tasks: set[asyncio.Task[None]] = set() + + async def on_startup(self) -> None: + """Register deterministic Health controls.""" + loop = asyncio.get_running_loop() + loop.add_signal_handler(signal.SIGUSR1, self._schedule_health, False) + loop.add_signal_handler(signal.SIGUSR2, self._schedule_health, True) + self._record("startup") + + async def on_shutdown(self) -> None: + """Release Health controls and complete pending updates.""" + loop = asyncio.get_running_loop() + for number in (signal.SIGUSR1, signal.SIGUSR2): + with contextlib.suppress(NotImplementedError, ValueError): + loop.remove_signal_handler(number) + if self._health_tasks: + await asyncio.gather(*self._health_tasks, return_exceptions=True) + self._record("shutdown") + + async def health(self) -> HealthStatus: + """Report the fixture's current process-wide Health.""" + return self._health + + @operation("stateforge.test/inspect@1") + async def inspect( + self, request: Input, context: OperationContext + ) -> Result: + """Perform one read-only operation and record its dispatch.""" + self._record(f"operation:inspect:{context.operation_id}") + return Result(value=request.value) + + def _record(self, event: str) -> None: + self._event_log.parent.mkdir(parents=True, exist_ok=True) + with self._event_log.open("a", encoding="utf-8") as stream: + stream.write(f"{event}\n") + + def _schedule_health(self, ready: bool) -> None: + task = asyncio.create_task(self._set_health(ready)) + self._health_tasks.add(task) + task.add_done_callback(self._health_tasks.discard) + + async def _set_health(self, ready: bool) -> None: + self._health = ( + HealthStatus.healthy() if ready else HealthStatus.maintenance() + ) + await self.refresh_health() + self._record("health:ready" if ready else "health:not-serving") + + +def _observe_admission(event_log: Path, state_directory: Path) -> None: + """Discard one post-admission response only when the marker is armed.""" + original = OperationAdmissionService.StartOperation + marker = state_directory / "drop-next-operation-response" + + async def start_operation( + service: OperationAdmissionService, + request: admission_pb2.StartOperationRequest, + context: grpc.aio.ServicerContext[ + admission_pb2.StartOperationRequest, + admission_pb2.StartOperationResponse, + ], + ) -> admission_pb2.StartOperationResponse: + digest = hashlib.sha256( + request.SerializeToString(deterministic=True) + ).hexdigest() + event_log.parent.mkdir(parents=True, exist_ok=True) + with event_log.open("a", encoding="utf-8") as stream: + stream.write(f"operation-admission-attempt:{digest}\n") + response = await original(service, request, context) + if marker.exists(): + marker.unlink() + with event_log.open("a", encoding="utf-8") as stream: + stream.write("operation-admission-response:dropped\n") + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "Injected response loss after admission", + ) + return response + + OperationAdmissionService.StartOperation = start_operation + + +def main() -> None: + """Run the real Provider service until systemd stops it.""" + parser = argparse.ArgumentParser() + parser.add_argument("--identity", required=True) + parser.add_argument("--display-name", required=True) + parser.add_argument("--discovery-directory", type=Path, required=True) + parser.add_argument("--state-directory", type=Path, required=True) + parser.add_argument("--event-log", type=Path, required=True) + parser.add_argument("--target-path", type=Path, required=True) + arguments = parser.parse_args() + AdmissionFixture.identity = arguments.identity + AdmissionFixture.display_name = arguments.display_name + _observe_admission(arguments.event_log, arguments.state_directory) + AdmissionFixture( + discovery_directory=arguments.discovery_directory, + state_directory=arguments.state_directory, + event_log=arguments.event_log, + ).run() + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/provider-runtime-scenarios.yml b/.github/workflows/provider-runtime-scenarios.yml index 5c5f977..cb4d848 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -165,6 +165,42 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + operation-admission-agent: + name: Operation Admission / installed Agent RPM + if: ${{ inputs.scenario-tier == 'full' }} + runs-on: ubuntu-24.04 + timeout-minutes: 40 + steps: + - name: Check out scenario automation + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Download exact runtime bundle + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + name: ${{ inputs.artifact-name }} + path: ${{ runner.temp }}/stateforge-provider-bundle + + - name: Run installed-Agent Operation Admission scenario + uses: ./.github/actions/scenarios/operation/installed-agent + with: + agent-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent + agent-test-image-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent/test-image + provider-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/provider + protocol-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/protocol + runtime-directory: ${{ runner.temp }}/sfp-operation-agent + evidence-directory: evidence/runtime/operation-admission-agent + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-operation-admission-agent + path: evidence/runtime/operation-admission-agent + if-no-files-found: warn + retention-days: ${{ inputs.retention-days }} + operation-lifecycle: name: Operation Lifecycle / direct Protocol peer if: ${{ inputs.scenario-tier == 'full' }} diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 6028b34..38c9bf2 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -155,7 +155,14 @@ The runtime suite currently covers: - installed-wheel Operation Admission through an independent Protocol peer, including exact fences, role and requirement rejection, 20 concurrent identical requests converging on one dispatch, typed idempotency conflict, - immutable author context, secret-safe evidence, and clean endpoint removal; + immutable author context, lost-response replay, deadline and schema + rejection, record expiry, restart, secret-safe evidence, and clean endpoint + removal; +- end-to-end Operation Admission through the exact installed Agent RPM and + exact Provider and Protocol wheels, including Agent-normalized admission, + identical concurrent requests, a discarded post-admission response and + immutable retry, changed-content conflict, replacement-generation fencing, + exactly-once dispatch, and bounded evidence; - installed-wheel Operation Lifecycle through an independent Protocol peer, including determinate progress, typed terminal results, coalesced state, immutable terminal replay, requested and repeated cancellation, confirmed @@ -167,17 +174,18 @@ The runtime suite currently covers: acknowledgement, Agent crash and restart, immutable replay with `accepted_duplicate`, callback publication, retry across journal write failure with an explicitly correlated retryable rejection, -conflicting-content rejection, a newer Provider endpoint generation, -rejection of publication from the superseded session, and clean shutdown. + conflicting-content rejection, a newer Provider endpoint generation, + rejection of publication from the superseded session, and clean shutdown. Runtime fixtures exercise Provider behavior through the public `stateforge_provider` API. The R18 fixture uses narrowly scoped private hooks only to discard one acknowledgement and assign one deterministic publication ID for fault injection; its publications still originate from the public -decorator and callback APIs. Each runtime scenario installs the exact Agent -RPM into an isolated Fedora system, starts -the packaged `stateforge-agent.service`, installs the exact Provider and -Protocol wheels beside it, and observes acceptance or rejection through the +decorator and callback APIs. R19 similarly discards one Operation Admission +response only after the real Provider has admitted it. Installed-Agent +scenarios install the exact Agent RPM into an isolated Fedora system, start +the packaged `stateforge-agent.service`, install the exact Provider and +Protocol wheels beside it, and observe acceptance or rejection through the real Agent journal. Runtime scenarios run on Linux because the current endpoint 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, @@ -236,6 +244,22 @@ while accepting a newer endpoint generation, then proves a publication queued by the superseded Provider is never acknowledged or added to durable Agent state. +R19 keeps the independent Protocol peer for deterministic Provider registry +checks and adds a separate installed-Agent scenario. The latter imports the +requester from the installed RPM in a standalone verifier process, uses the real +systemd-managed Provider on its Unix socket, and correlates Agent dispositions +with operation-ID-specific Provider dispatch events and serialized-request +digests. It also checks generation, session, and manifest fences independently +and resolves an uncertain admission from a new requester process using the same +durable Agent journal. The Agent daemon runs and monitors the Provider, but it +does not originate these admission calls: it has no operation-submission ingress +yet. This is installed Agent requester-code integration, not daemon-level +admission. A response-loss marker is confined to the CI fixture; no production +author API or wire contract changes for this test. +The retained evidence includes exact artifact checksums, Agent and Provider +process identities, endpoint generations, journal excerpts, dispositions, +dispatch counts, and final cleanup state. + ## Evidence and security Build provenance and checksums are retained with the immutable artifact bundle. diff --git a/pyproject.toml b/pyproject.toml index 9b9514d..a8ee531 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,7 @@ environments = ["python_version == '3.13'"] [tool.uv.sources] -stateforge-protocol = { git = "https://github.com/stateforge-org/stateforge-protocol.git", rev = "cb4e9af352fc89204585d1f0b16d2870a990c560" } +stateforge-protocol = { git = "https://github.com/stateforge-org/stateforge-protocol.git", rev = "64514b35dabfe88d49aabfb63d756e98a97fe5dc" } [tool.hatch.build.targets.wheel] diff --git a/src/stateforge_provider/_services/operation_admission/errors.py b/src/stateforge_provider/_services/operation_admission/errors.py index 3a49234..c902dd4 100644 --- a/src/stateforge_provider/_services/operation_admission/errors.py +++ b/src/stateforge_provider/_services/operation_admission/errors.py @@ -10,6 +10,7 @@ from grpc_status import rpc_status from stateforge_protocol.sfp import primitives from stateforge_protocol.sfp.operation_admission.v1 import ( + contract, operation_admission_pb2, validation, ) @@ -23,17 +24,19 @@ class AdmissionFailureError(Exception): reason: operation_admission_pb2.AdmissionErrorReason.ValueType explanation: str operation_id: str = "" - retryable: bool = False def status(self) -> grpc.Status: """Build a validated rich status with one versioned detail.""" + semantics = contract.ADMISSION_ERROR_CONTRACTS[self.reason] detail = operation_admission_pb2.OperationAdmissionErrorDetail( operation_id=_valid_operation_id(self.operation_id), reason=self.reason, - retryable=self.retryable, + retryable=semantics.retryable, explanation=self.explanation, ) - validation.validate_admission_error_detail(detail) + validation.validate_admission_rejection( + detail, grpc_status=self.code.name + ) packed = Any() packed.Pack(detail) status = status_pb2.Status( diff --git a/src/stateforge_provider/_services/operation_admission/service.py b/src/stateforge_provider/_services/operation_admission/service.py index 3b7053c..e77579c 100644 --- a/src/stateforge_provider/_services/operation_admission/service.py +++ b/src/stateforge_provider/_services/operation_admission/service.py @@ -422,14 +422,12 @@ async def _reject_registry( operation_admission_pb2.ADMISSION_ERROR_REASON_PROVIDER_CAPACITY_EXHAUSTED, "Provider admission capacity is exhausted", operation_id, - True, ), AdmissionRegistryFailure.STOPPING: AdmissionFailureError( grpc.StatusCode.UNAVAILABLE, operation_admission_pb2.ADMISSION_ERROR_REASON_PROVIDER_UNAVAILABLE, "Provider operation runtime is stopping", operation_id, - True, ), }[error.failure] await self._reject(context, failure) diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 483979c..d699fa3 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -8,7 +8,7 @@ SERVICES = ROOT / "src" / "stateforge_provider" / "_services" SHA_LENGTH = 40 SUPPORTED_PYTHON_COUNT = 3 -SCENARIO_COUNT = 12 +SCENARIO_COUNT = 13 DOWNSTREAM_BUNDLE_CONSUMER_COUNT = 3 DEPENDENCY_CHECKOUT_COUNT = 2 @@ -43,6 +43,7 @@ def test_action_library_has_focused_domain_boundaries() -> None: "scenarios/lifecycle/concurrent-providers", "scenarios/lifecycle/same-instance-replacement", "scenarios/operation/admission", + "scenarios/operation/installed-agent", "scenarios/operation/lifecycle", "scenarios/resilience/crash-restart", "scenarios/smoke/normal-lifecycle", @@ -226,9 +227,10 @@ def test_each_runtime_scenario_is_an_independent_action() -> None: assert len(scenarios) == SCENARIO_COUNT for scenario in scenarios: content = read(scenario) - if "scenarios/trigger/publication" in str( - scenario - ) or "scenarios/operation/" in str(scenario): + if "scenarios/trigger/publication" in str(scenario) or any( + f"scenarios/operation/{name}" in str(scenario) + for name in ("admission", "lifecycle") + ): assert "./.github/actions/environment/install" in content assert "./.github/actions/agent/" not in content assert '"${ISOLATED_PYTHON}" -I' in content @@ -242,6 +244,13 @@ def test_each_runtime_scenario_is_an_independent_action() -> None: assert "./.github/actions/provider/stop" in content assert "./.github/actions/evidence/collect" in content assert "if: ${{ always() }}" in content + elif "scenarios/operation/installed-agent" in str(scenario): + assert "./.github/actions/agent/install" in content + assert "./.github/actions/agent/start" in content + assert "./.github/actions/provider/start" in content + assert "./.github/actions/provider/stop" in content + assert "./.github/actions/evidence/collect" in content + assert "if: ${{ always() }}" in content else: assert "./.github/actions/agent/install" in content assert "./.github/actions/agent/start" in content @@ -265,6 +274,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "trigger-publication-agent", "capability-assessment", "operation-admission", + "operation-admission-agent", "operation-lifecycle", } for scenario in scenarios: @@ -372,6 +382,36 @@ def test_operation_admission_keeps_the_focused_protocol_peer() -> None: assert "${{ runner.temp }}/sfp-r5" in workflow +def test_operation_admission_exercises_installed_agent_requester() -> None: + """R19 must run the exact RPM and Provider wheel in separate processes.""" + scenario = read( + ACTIONS / "scenarios" / "operation" / "installed-agent" / "action.yml" + ) + fixture = read( + ACTIONS / "scenarios" / "operation" / "installed-agent" / "fixture.py" + ) + assertion = read( + ACTIONS / "scenarios" / "operation" / "installed-agent" / "assert.py" + ) + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + assert "./.github/actions/agent/install" in scenario + assert "./.github/actions/agent/assert-session" in scenario + assert "operation-admission-agent" in workflow + assert "Operation Admission / installed Agent RPM" in workflow + assert "OperationAdmissionCoordinator" in assertion + assert "OperationAdmissionClient" in assertion + assert "OperationAdmissionService.StartOperation" in fixture + assert "drop-next-operation-response" in fixture + assert "operation-admission-attempt:" in fixture + assert "operation:inspect:{context.operation_id}" in fixture + assert "--prepare-uncertain" in assertion + assert "stale_generation_rejected" in assertion + assert "stale_session_rejected" in assertion + assert "stale_manifest_rejected" in assertion + assert "dispatch_operation_ids" in assertion + assert "stateforge-r19-secret-never-in-evidence" in scenario + + def test_operation_lifecycle_keeps_the_focused_protocol_peer() -> None: """Prove R6 installed behavior without simulating a future Agent.""" scenario = read( diff --git a/tests/test_operation_admission.py b/tests/test_operation_admission.py index c456273..5ac0927 100644 --- a/tests/test_operation_admission.py +++ b/tests/test_operation_admission.py @@ -276,7 +276,9 @@ def _detail(error: grpc.aio.AioRpcError): packed = status.details[0] detail = operation_admission_pb2.OperationAdmissionErrorDetail() assert packed.Unpack(detail) - admission_validation.validate_admission_error_detail(detail) + admission_validation.validate_admission_rejection( + detail, grpc_status=error.code().name + ) return detail diff --git a/uv.lock b/uv.lock index 2d4b070..024d9fc 100644 --- a/uv.lock +++ b/uv.lock @@ -1613,7 +1613,7 @@ wheels = [ [[package]] name = "stateforge-protocol" version = "0.1.0" -source = { git = "https://github.com/stateforge-org/stateforge-protocol.git?rev=cb4e9af352fc89204585d1f0b16d2870a990c560#cb4e9af352fc89204585d1f0b16d2870a990c560" } +source = { git = "https://github.com/stateforge-org/stateforge-protocol.git?rev=64514b35dabfe88d49aabfb63d756e98a97fe5dc#64514b35dabfe88d49aabfb63d756e98a97fe5dc" } dependencies = [ { name = "grpcio" }, { name = "grpcio-health-checking" }, @@ -1664,7 +1664,7 @@ requires-dist = [ { name = "grpcio-status", specifier = ">=1.83" }, { name = "jsonschema", specifier = ">=4.25,<5" }, { name = "pydantic", specifier = ">=2.12,<3" }, - { name = "stateforge-protocol", git = "https://github.com/stateforge-org/stateforge-protocol.git?rev=cb4e9af352fc89204585d1f0b16d2870a990c560" }, + { name = "stateforge-protocol", git = "https://github.com/stateforge-org/stateforge-protocol.git?rev=64514b35dabfe88d49aabfb63d756e98a97fe5dc" }, ] [package.metadata.requires-dev]