From 5f7ba091bbab02c4e43feae2f6ed193da69ecb39 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Mon, 2 Nov 2026 14:08:27 +0200 Subject: [PATCH 1/3] test: verify operation admission against installed Agent --- .../scenarios/operation/admission/verify.py | 380 ++++++++++- .../operation/installed-agent/action.yml | 167 +++++ .../operation/installed-agent/assert.py | 594 ++++++++++++++++++ .../operation/installed-agent/fixture.py | 171 +++++ .../workflows/provider-runtime-scenarios.yml | 36 ++ docs/ci-automation.md | 32 +- pyproject.toml | 2 +- .../_services/operation_admission/errors.py | 9 +- .../_services/operation_admission/service.py | 2 - tests/test_ci_automation.py | 42 +- tests/test_operation_admission.py | 4 +- uv.lock | 4 +- 12 files changed, 1416 insertions(+), 27 deletions(-) create mode 100644 .github/actions/scenarios/operation/installed-agent/action.yml create mode 100644 .github/actions/scenarios/operation/installed-agent/assert.py create mode 100644 .github/actions/scenarios/operation/installed-agent/fixture.py diff --git a/.github/actions/scenarios/operation/admission/verify.py b/.github/actions/scenarios/operation/admission/verify.py index dd50757..0cd6e44 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,310 @@ 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 + 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") + 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 +577,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 +663,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 +701,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 +727,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..cbf5cdd --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/action.yml @@ -0,0 +1,167 @@ +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 the installed Agent package + 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 .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_fence_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..459d921 --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/assert.py @@ -0,0 +1,594 @@ +"""Verify real RPM-installed Agent admission against a Provider process.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import sys +import time +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) + 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 _wait_for_dispatches_sync(path: Path, expected: int) -> None: + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if _events(path).count("operation:inspect") == 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 _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_count = _events(event_log).count("operation:inspect") + 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 _events(event_log).count("operation:inspect") != 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 = _events(event_log).count("operation:inspect") + 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 _events(event_log).count("operation:inspect") != 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() + manager, coordinator = await _manager(journal, attempts=1) + await manager.establish(initial) + await _wait_for_availability( + manager, provider_id, ProviderAvailabilityState.AVAILABLE + ) + rebuilt_id = await _uncertain_admission( + manager, coordinator, provider_id, state, "requester-rebuild" + ) + await _wait_for_dispatches(event_log, initial_count + 4) + await manager.close() + 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 _events(event_log).count("operation:inspect") != 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 _events(event_log).count("operation:inspect") != initial_count + 5: + raise AssertionError("Indeterminate resolution re-dispatched work") + old_fence = admission_pb2.StartOperationRequest() + old_fence.CopyFrom(wire) + old_fence.invocation.operation_id = str(uuid4()) + try: + new_lease = await manager._operation_admission_lease(provider_id) + await client.start(new_lease.channel, old_fence) + except SfpOperationAdmissionRejectedError as error: + if "OPERATION_PRECONDITION_FAILED" not in error.rejection.reason: + raise AssertionError( + "Stale fence had the wrong reason" + ) from error + else: + raise AssertionError("Superseded generation was accepted") + 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(old_fence) + expired.session_binding.expected_endpoint_generation = ( + replacement.endpoint_generation + ) + expired.session_binding.expected_session_id = new_lease.session_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 _events(event_log).count("operation:inspect") != initial_count + 6: + raise AssertionError("Expired identity re-dispatched work") + 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, + "provider_restart_indeterminate": True, + "expired_record_rejected": True, + "schema_rejected": True, + "deadline_rejected": True, + "readiness_rejected_and_recovered": True, + "agent_service_restarted": True, + "stale_fence_rejected": True, + "changed_content_rejected": True, + "dispatches": _events(event_log).count("operation:inspect") + - initial_count, + "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.""" + sys.stdout.write( + json.dumps(asyncio.run(_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..c7aac39 --- /dev/null +++ b/.github/actions/scenarios/operation/installed-agent/fixture.py @@ -0,0 +1,171 @@ +"""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, + 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) -> Result: + """Perform one read-only operation and record its dispatch.""" + self._record("operation:inspect") + 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..c938079 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,16 @@ 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, uses the real Provider process on its Unix +socket, and correlates Agent dispositions with Provider dispatch events and +serialized-request digests. 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..ceb1722 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,30 @@ 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 "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] From 33b8d205e5fac0cf8adacaa304d0965843f8f9f3 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Mon, 2 Nov 2026 14:44:14 +0200 Subject: [PATCH 2/3] test: strengthen installed Agent admission verification --- .../operation/installed-agent/action.yml | 9 +- .../operation/installed-agent/assert.py | 223 +++++++++++++++--- .../operation/installed-agent/fixture.py | 7 +- docs/ci-automation.md | 14 +- tests/test_ci_automation.py | 6 + 5 files changed, 213 insertions(+), 46 deletions(-) diff --git a/.github/actions/scenarios/operation/installed-agent/action.yml b/.github/actions/scenarios/operation/installed-agent/action.yml index cbf5cdd..8b5870e 100644 --- a/.github/actions/scenarios/operation/installed-agent/action.yml +++ b/.github/actions/scenarios/operation/installed-agent/action.yml @@ -75,7 +75,7 @@ runs: endpoint-generation: ${{ steps.provider.outputs.endpoint-generation }} evidence-directory: ${{ inputs.evidence-directory }}/agent - - name: Exercise admission through the installed Agent package + - name: Exercise admission through installed Agent requester code id: assertion shell: bash env: @@ -95,14 +95,17 @@ runs: --state-directory "${STATE_DIRECTORY}" \ --unit-name "${UNIT_NAME}" \ > "${EVIDENCE_DIRECTORY}/operation-admission.json" - jq -e '.dispatches == 6 and .lost_response_attempts == 2 + 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_fence_rejected and .changed_content_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' \ diff --git a/.github/actions/scenarios/operation/installed-agent/assert.py b/.github/actions/scenarios/operation/installed-agent/assert.py index 459d921..5023f71 100644 --- a/.github/actions/scenarios/operation/installed-agent/assert.py +++ b/.github/actions/scenarios/operation/installed-agent/assert.py @@ -6,8 +6,10 @@ 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 @@ -57,6 +59,7 @@ def _arguments() -> argparse.Namespace: 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() @@ -74,10 +77,18 @@ def _events(path: Path) -> tuple[str, ...]: 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 _events(path).count("operation:inspect") == expected: + if len(_dispatch_ids(path)) == expected: return time.sleep(0.05) raise AssertionError(f"Expected exactly {expected} Provider dispatches") @@ -238,6 +249,77 @@ async def _uncertain_admission( 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: @@ -270,7 +352,8 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR await _wait_for_availability( manager, provider_id, ProviderAvailabilityState.AVAILABLE ) - initial_count = _events(event_log).count("operation:inspect") + initial_dispatches = _dispatch_ids(event_log) + initial_count = len(initial_dispatches) accepted = await manager.admit_operation( _request(provider_id, "first") ) @@ -308,7 +391,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR ): raise AssertionError("Concurrent admission had multiple winners") await _wait_for_dispatches(event_log, initial_count + 2) - if _events(event_log).count("operation:inspect") != 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 ( @@ -331,7 +414,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR else: raise AssertionError("Changed content was accepted") - before_rejections = _events(event_log).count("operation:inspect") + before_rejections = len(_dispatch_ids(event_log)) malformed = admission_pb2.StartOperationRequest() malformed.CopyFrom(wire) malformed.invocation.operation_id = str(uuid4()) @@ -389,7 +472,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR await _wait_for_availability( manager, provider_id, ProviderAvailabilityState.AVAILABLE ) - if _events(event_log).count("operation:inspect") != before_rejections: + if len(_dispatch_ids(event_log)) != before_rejections: raise AssertionError("Rejected work reached the author operation") before_retry = tuple( @@ -427,16 +510,13 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR raise AssertionError("Fault hook did not discard a response") await manager.close() - manager, coordinator = await _manager(journal, attempts=1) - await manager.establish(initial) - await _wait_for_availability( - manager, provider_id, ProviderAvailabilityState.AVAILABLE - ) - rebuilt_id = await _uncertain_admission( - manager, coordinator, provider_id, state, "requester-rebuild" + await journal.close() + rebuilt_id, uncertain_requester_pid = await _spawn_uncertain_requester( + arguments ) await _wait_for_dispatches(event_log, initial_count + 4) - await manager.close() + journal = SqliteJournal(_JOURNAL) + await journal.open() manager, coordinator = await _manager(journal) await manager.establish(initial) await _wait_for_availability( @@ -450,7 +530,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR raise AssertionError( "Rebuilt Agent did not resolve Provider record" ) - if _events(event_log).count("operation:inspect") != initial_count + 4: + if len(_dispatch_ids(event_log)) != initial_count + 4: raise AssertionError("Requester rebuild re-dispatched work") await manager.close() @@ -484,21 +564,56 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR raise AssertionError( "Provider restart blindly re-admitted old work" ) - if _events(event_log).count("operation:inspect") != initial_count + 5: + if len(_dispatch_ids(event_log)) != initial_count + 5: raise AssertionError("Indeterminate resolution re-dispatched work") - old_fence = admission_pb2.StartOperationRequest() - old_fence.CopyFrom(wire) - old_fence.invocation.operation_id = str(uuid4()) - try: - new_lease = await manager._operation_admission_lease(provider_id) - await client.start(new_lease.channel, old_fence) - except SfpOperationAdmissionRejectedError as error: - if "OPERATION_PRECONDITION_FAILED" not in error.rejection.reason: - raise AssertionError( - "Stale fence had the wrong reason" - ) from error - else: - raise AssertionError("Superseded generation was accepted") + 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") ) @@ -514,11 +629,15 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR ) await asyncio.sleep(1) expired = admission_pb2.StartOperationRequest() - expired.CopyFrom(old_fence) - expired.session_binding.expected_endpoint_generation = ( - replacement.endpoint_generation + 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.session_binding.expected_session_id = new_lease.session_id expired.invocation.CopyFrom(fresh_attempt.invocation) expired.admission_mode = admission_pb2.ADMISSION_MODE_RESOLVE_ONLY try: @@ -530,8 +649,23 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR ) from error else: raise AssertionError("Expired Provider record was resolved") - if _events(event_log).count("operation:inspect") != initial_count + 6: + 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", @@ -562,16 +696,22 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR "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_fence_rejected": True, + "stale_generation_rejected": True, + "stale_session_rejected": True, + "stale_manifest_rejected": True, "changed_content_rejected": True, - "dispatches": _events(event_log).count("operation:inspect") - - initial_count, + "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): @@ -584,8 +724,17 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR def main() -> None: """Emit only bounded, secret-safe JSON evidence to stdout.""" + arguments = _arguments() sys.stdout.write( - json.dumps(asyncio.run(_run(_arguments())), indent=2, sort_keys=True) + json.dumps( + asyncio.run( + _prepare_uncertain_process(arguments) + if arguments.prepare_uncertain + else _run(arguments) + ), + indent=2, + sort_keys=True, + ) + "\n" ) diff --git a/.github/actions/scenarios/operation/installed-agent/fixture.py b/.github/actions/scenarios/operation/installed-agent/fixture.py index c7aac39..7d17870 100644 --- a/.github/actions/scenarios/operation/installed-agent/fixture.py +++ b/.github/actions/scenarios/operation/installed-agent/fixture.py @@ -17,6 +17,7 @@ from stateforge_provider import ( HealthStatus, + OperationContext, Platform, Provider, ProviderModel, @@ -91,9 +92,11 @@ async def health(self) -> HealthStatus: return self._health @operation("stateforge.test/inspect@1") - async def inspect(self, request: Input) -> Result: + async def inspect( + self, request: Input, context: OperationContext + ) -> Result: """Perform one read-only operation and record its dispatch.""" - self._record("operation:inspect") + self._record(f"operation:inspect:{context.operation_id}") return Result(value=request.value) def _record(self, event: str) -> None: diff --git a/docs/ci-automation.md b/docs/ci-automation.md index c938079..38c9bf2 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -246,10 +246,16 @@ 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, uses the real Provider process on its Unix -socket, and correlates Agent dispositions with Provider dispatch events and -serialized-request digests. A response-loss marker is confined to the CI -fixture; no production author API or wire contract changes for this test. +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. diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index ceb1722..d699fa3 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -403,6 +403,12 @@ def test_operation_admission_exercises_installed_agent_requester() -> None: 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 From 5ba661cbba62528a8c8f6b465ef940feb77e5c9e Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Mon, 2 Nov 2026 15:03:05 +0200 Subject: [PATCH 3/3] test: wait for reused admission dispatch --- .github/actions/scenarios/operation/admission/verify.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/actions/scenarios/operation/admission/verify.py b/.github/actions/scenarios/operation/admission/verify.py index 0cd6e44..42637a0 100644 --- a/.github/actions/scenarios/operation/admission/verify.py +++ b/.github/actions/scenarios/operation/admission/verify.py @@ -404,6 +404,7 @@ async def _verify_expiry_and_restart( # noqa: PLR0912, PLR0915 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, @@ -423,6 +424,7 @@ async def _verify_expiry_and_restart( # noqa: PLR0912, PLR0915 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()