diff --git a/.github/actions/scenarios/operation/installed-agent/assert.py b/.github/actions/scenarios/operation/installed-agent/assert.py index 5023f71..7fde9ca 100644 --- a/.github/actions/scenarios/operation/installed-agent/assert.py +++ b/.github/actions/scenarios/operation/installed-agent/assert.py @@ -217,12 +217,12 @@ async def _uncertain_admission( (_OPERATION,), execution_context=request.execution_context, ) - lease = await manager._operation_admission_lease(provider_id) + lease = await manager._leases._operation_admission_lease(provider_id) operation_id = await coordinator.prepare( lease, request, assessment, - assert_current=manager._assert_operation_admission_is_current, + assert_current=manager._leases._assert_operation_admission_is_current, ) (state / "drop-next-operation-response").touch() try: @@ -230,7 +230,7 @@ async def _uncertain_admission( lease, operation_id, assessment, - assert_current=manager._assert_operation_admission_is_current, + assert_current=manager._leases._assert_operation_admission_is_current, ) except OperationAdmissionUnresolvedError as error: if error.operation_id != operation_id: @@ -370,7 +370,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR 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) + lease = await manager._leases._operation_admission_lease(provider_id) client = OperationAdmissionClient(timeout_seconds=3) wire = _wire_request( provider_id=provider_id, @@ -461,7 +461,7 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR "Agent admitted while Provider was not serving" ) try: - await manager._operation_admission_lease(provider_id) + await manager._leases._operation_admission_lease(provider_id) except OperationAdmissionPreconditionError: pass else: @@ -566,7 +566,9 @@ async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR ) 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) + new_lease = await manager._leases._operation_admission_lease( + provider_id + ) old_revision = new_lease.manifest_revision stale_revision = old_revision[:-1] + ( "0" if old_revision[-1] != "0" else "1" diff --git a/.github/actions/scenarios/operation/lifecycle/installed-agent/action.yml b/.github/actions/scenarios/operation/lifecycle/installed-agent/action.yml new file mode 100644 index 0000000..df9281d --- /dev/null +++ b/.github/actions/scenarios/operation/lifecycle/installed-agent/action.yml @@ -0,0 +1,166 @@ +name: Verify Operation Lifecycle with the installed Agent +description: Exercise installed Agent lifecycle code against the real Provider endpoint. + +inputs: + agent-artifact-directory: + description: Directory containing the exact Agent RPM build. + required: true + agent-test-image-directory: + description: Bundled Agent Fedora runtime-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 bounded 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 service + uses: ./.github/actions/agent/start + with: + container-name: ${{ steps.system.outputs.container-name }} + + - name: Stage lifecycle Provider fixture + id: fixture + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + RUNTIME_DIRECTORY: ${{ inputs.runtime-directory }} + run: | + set -euo pipefail + root="${RUNTIME_DIRECTORY}/lifecycle-provider" + mkdir -p "${root}" + install -m 0644 "${ACTION_PATH}/fixture.py" "${root}/provider.py" + { + echo "fixture-path=${root}/provider.py" + echo "instance-name=lifecycle-provider" + echo "state-name=lifecycle-provider" + } >> "${GITHUB_OUTPUT}" + + - name: Start real lifecycle 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_lifecycle + 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 installed Agent Operation Lifecycle requester + id: assertion + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/lifecycle + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + JOURNAL_PATH: /var/lib/stateforge-agent/journal/operation-lifecycle-r20.db + 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/lifecycle-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}" \ + --journal-path "${JOURNAL_PATH}" \ + --state-directory "${STATE_DIRECTORY}" \ + --event-log "${EVENT_LOG}" \ + --unit-name "${UNIT_NAME}" \ + > "${EVIDENCE_DIRECTORY}/operation-lifecycle.json" + jq -e '.late_replay and ((.progress_sequences | length) >= 2) + and .cursor_gap_replayed + and .success_result and .expected_failure and .sanitized_failure + and .invalid_result and .cancellation_completed and .cancellation_effects + and .cancellation_race_resolved + and .watch_retry_injected and .watch_retry_recovered + and .requester_restarted and .stale_session_rejected + and .provider_restarted and .endpoint_generation_increased + and .no_redispatch + and .restarted_terminal_state == "OPERATION_STATE_SUCCEEDED" + and .restarted_session_id != .session_id + and .dispatches == 6' \ + "${EVIDENCE_DIRECTORY}/operation-lifecycle.json" >/dev/null + + - name: Capture final Provider endpoint for cleanup + id: final-provider + if: ${{ always() }} + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + DESCRIPTOR_PATH: ${{ steps.provider.outputs.descriptor-path }} + FALLBACK_SOCKET_PATH: ${{ steps.provider.outputs.socket-path }} + run: | + set -euo pipefail + socket_path="${FALLBACK_SOCKET_PATH}" + if [[ -n "${DESCRIPTOR_PATH}" ]] && docker exec \ + "${CONTAINER_NAME}" test -f "${DESCRIPTOR_PATH}"; then + socket_path="$(docker exec "${CONTAINER_NAME}" \ + jq -r .transport.socket_path "${DESCRIPTOR_PATH}")" + fi + echo "socket-path=${socket_path}" >> "${GITHUB_OUTPUT}" + + - name: Stop real lifecycle Provider fixture + if: ${{ always() }} + 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.final-provider.outputs.socket-path }} + event-log: ${{ steps.provider.outputs.event-log }} + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Preserve bounded evidence and clean up + if: ${{ always() }} + uses: ./.github/actions/evidence/collect + with: + stage: operation-lifecycle-agent + container-name: ${{ steps.system.outputs.container-name }} + runtime-directory: ${{ inputs.runtime-directory }} + evidence-directory: ${{ inputs.evidence-directory }} + + - name: Reject sensitive lifecycle evidence + if: ${{ always() }} + shell: bash + env: + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }} + run: | + set -euo pipefail + if grep --recursive --fixed-strings --line-number \ + 'stateforge-r20-secret-never-in-evidence' \ + "${EVIDENCE_DIRECTORY}"; then + echo 'Sensitive operation content reached evidence' >&2 + exit 1 + fi diff --git a/.github/actions/scenarios/operation/lifecycle/installed-agent/assert.py b/.github/actions/scenarios/operation/lifecycle/installed-agent/assert.py new file mode 100644 index 0000000..e248e43 --- /dev/null +++ b/.github/actions/scenarios/operation/lifecycle/installed-agent/assert.py @@ -0,0 +1,703 @@ +"""Exercise installed Agent Operation Lifecycle APIs against a Provider.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import sys +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from importlib.metadata import version +from pathlib import Path +from uuid import UUID, uuid4 + +import grpc +from stateforge_agent._journal import SqliteJournal +from stateforge_agent._sfp import ( + OperationAdmissionRequest, + ProviderAvailabilityState, + ProviderSessionManager, + SfpVersion, +) +from stateforge_agent._sfp._runtime.descriptor import load_provider_descriptor +from stateforge_agent._sfp._services.operation_admission.coordinator import ( + OperationAdmissionCoordinator, +) +from stateforge_agent._sfp._services.operation_lifecycle.client import ( + OperationLifecycleClient, + SfpOperationLifecycleTransportError, +) +from stateforge_agent._sfp._services.operation_lifecycle.coordinator import ( + OperationLifecycleCoordinator, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + operation_lifecycle_pb2, +) + +_DISCOVERY = Path("/run/stateforge-agent/providers") +_DEFAULT_JOURNAL = Path( + "/var/lib/stateforge-agent/journal/operation-lifecycle-r20.db" +) +_TERMINAL = { + "OPERATION_STATE_SUCCEEDED", + "OPERATION_STATE_FAILED", + "OPERATION_STATE_CANCELLED", +} +_MIDDLE_PROGRESS = 50.0 +_COMPLETE_PROGRESS = 100.0 +_EXPECTED_DISPATCHES = 6 +_SCRIPT_PATH = Path(__file__).resolve() +_PROVIDER_ERROR_SENTINEL = "stateforge-r20-secret-never-in-evidence" + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--reopen-operation", type=UUID) + parser.add_argument("--previous-session-id") + parser.add_argument("--previous-generation", type=int) + parser.add_argument("--agent-runtime-id", type=UUID) + parser.add_argument("--provider-instance-id", type=UUID, required=True) + parser.add_argument("--journal-path", type=Path, default=_DEFAULT_JOURNAL) + parser.add_argument("--state-directory", type=Path) + parser.add_argument("--event-log", type=Path) + parser.add_argument("--unit-name") + return parser.parse_args() + + +async def _new_agent( + journal: SqliteJournal, + agent_runtime_id: UUID | None = None, +) -> tuple[ProviderSessionManager, OperationAdmissionCoordinator]: + store = journal.operation_admissions(maximum_unresolved_attempts=32) + await store.initialize() + admission = OperationAdmissionCoordinator( + store, retry_initial_delay_seconds=0.05 + ) + lifecycle = OperationLifecycleCoordinator( + store, retry_initial_delay_seconds=0.05 + ) + manager = ProviderSessionManager( + agent_runtime_id=agent_runtime_id or 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=admission, + operation_lifecycle=lifecycle, + ) + return manager, admission + + +def _events(path: Path) -> tuple[str, ...]: + try: + return tuple(path.read_text(encoding="utf-8").splitlines()) + except FileNotFoundError: + return () + + +async def _wait_until(predicate: Callable[[], bool], message: str) -> None: + try: + async with asyncio.timeout(20): + while not predicate(): # noqa: ASYNC110 + await asyncio.sleep(0.025) + except TimeoutError as error: + raise AssertionError(message) from error + + +def _descriptor_generation(provider_id: UUID) -> int: + try: + descriptor = json.loads( + (_DISCOVERY / f"{provider_id}.sfp.json").read_text( + encoding="utf-8" + ) + ) + except (FileNotFoundError, json.JSONDecodeError): + return 0 + return int(descriptor["endpoint_generation"]) + + +async def _unit_invocation_id(unit_name: str) -> str: + process = await asyncio.create_subprocess_exec( + "/usr/bin/systemctl", + "show", + unit_name, + "--property", + "InvocationID", + "--value", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + output, error = await process.communicate() + if process.returncode != 0: + raise AssertionError( + "Could not inspect Provider systemd invocation: " + + error.decode("utf-8", errors="replace")[-1000:] + ) + return output.decode("utf-8").strip() + + +async def _available( + manager: ProviderSessionManager, provider_id: UUID +) -> None: + await _wait_until( + lambda: ( + (status := manager.availability_status(provider_id)) is not None + and status.state is ProviderAvailabilityState.AVAILABLE + ), + "Agent did not observe Provider availability", + ) + + +async def _admit( + manager: ProviderSessionManager, + provider_id: UUID, + mode: str, +) -> UUID: + result = await manager.admit_operation( + OperationAdmissionRequest( + provider_instance_id=provider_id, + operation_identifier="stateforge.test/execute@1", + payload={"mode": mode}, + execution_deadline=datetime.now(UTC) + timedelta(minutes=2), + execution_context={"scenario": "provider-r20"}, + ) + ) + return result.operation_id + + +async def _status( + manager: ProviderSessionManager, provider_id: UUID, operation_id: UUID +) -> object: + async with asyncio.timeout(20): + while True: + status = await manager.operation_status(provider_id, operation_id) + if status is not None and status.state in _TERMINAL: + return status + await asyncio.sleep(0.025) + + +async def _watch_to_terminal( + manager: ProviderSessionManager, + admission: OperationAdmissionCoordinator, + *, + provider_id: UUID, + operation_id: UUID, + first_response: asyncio.Event, + half_progress: asyncio.Event, +) -> list[operation_lifecycle_pb2.WatchOperationResponse]: + lease = await manager._leases._operation_lifecycle_lease(provider_id) + attempt = await admission.attempt(provider_id, operation_id) + if attempt is None: + raise AssertionError("Agent lost the durable operation identity") + request = operation_lifecycle_pb2.WatchOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=str(provider_id), + expected_endpoint_generation=lease.endpoint_generation, + expected_session_id=lease.session_id, + ), + operation_id=str(operation_id), + expected_request_fingerprint=attempt.request_fingerprint, + after_sequence=1, + ) + responses = [] + async for response in OperationLifecycleClient().watch( + lease.channel, request + ): + responses.append(response) + first_response.set() + if ( + response.snapshot.HasField("progress") + and response.snapshot.progress.HasField("percentage") + and response.snapshot.progress.percentage == _MIDDLE_PROGRESS + ): + half_progress.set() + return responses + + +async def _restarted_requester( + provider_id: UUID, + operation_id: UUID, + *, + previous_session_id: str, + previous_generation: int, + agent_runtime_id: UUID, + journal_path: Path, +) -> dict[str, object]: + """Reopen installed Agent persistence and reject its superseded session.""" + journal = SqliteJournal(journal_path) + await journal.open() + manager, admission = await _new_agent(journal, agent_runtime_id) + try: + descriptor = load_provider_descriptor( + _DISCOVERY / f"{provider_id}.sfp.json", + discovery_directory=_DISCOVERY, + ) + session = await manager.establish(descriptor) + await _available(manager, provider_id) + if session.session_id == previous_session_id: + raise AssertionError( + "Requester restart reused the old SFP session" + ) + status = await manager.operation_status(provider_id, operation_id) + if status is None or status.state != "OPERATION_STATE_SUCCEEDED": + raise AssertionError( + "Restarted requester lost terminal journal state" + ) + attempt = await admission.attempt(provider_id, operation_id) + if attempt is None: + raise AssertionError("Restarted requester lost admission identity") + lease = await manager._leases._operation_lifecycle_lease(provider_id) + stale = operation_lifecycle_pb2.WatchOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=str(provider_id), + expected_endpoint_generation=previous_generation, + expected_session_id=previous_session_id, + ), + operation_id=str(operation_id), + expected_request_fingerprint=attempt.request_fingerprint, + ) + try: + async for _ in OperationLifecycleClient().watch( + lease.channel, stale + ): + pass + except SfpOperationLifecycleTransportError as error: + if error.code is not grpc.StatusCode.FAILED_PRECONDITION: + raise AssertionError( + "Superseded session had the wrong rejection status" + ) from error + else: + raise AssertionError("Superseded Agent session was accepted") + return { + "requester_pid": os.getpid(), + "session_id": session.session_id, + "endpoint_generation": session.endpoint_generation, + "retained_terminal_state": status.state, + "stale_session_rejected": True, + } + finally: + await manager.close() + await journal.close() + + +async def _spawn_restarted_requester( + provider_id: UUID, + operation_id: UUID, + *, + previous_session_id: str, + previous_generation: int, + agent_runtime_id: UUID, + journal_path: Path, +) -> dict[str, object]: + process = await asyncio.create_subprocess_exec( + sys.executable, + "-I", + str(_SCRIPT_PATH), + "--reopen-operation", + str(operation_id), + "--provider-instance-id", + str(provider_id), + "--journal-path", + str(journal_path), + "--previous-session-id", + previous_session_id, + "--previous-generation", + str(previous_generation), + "--agent-runtime-id", + str(agent_runtime_id), + 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( + "Restarted Agent requester timed out" + ) from timeout_error + if process.returncode != 0: + raise AssertionError( + "Restarted Agent requester failed: " + + error.decode("utf-8", errors="replace")[-2000:] + ) + evidence = json.loads(output) + if evidence["requester_pid"] != process.pid or process.pid == os.getpid(): + raise AssertionError( + "Requester restart did not cross a process boundary" + ) + return evidence + + +async def _run(arguments: argparse.Namespace) -> dict[str, object]: # noqa: PLR0912, PLR0915 + if arguments.state_directory is None or arguments.event_log is None: + raise ValueError( + "state and event paths are required for the full scenario" + ) + provider_id: UUID = arguments.provider_instance_id + state: Path = arguments.state_directory + journal = SqliteJournal(arguments.journal_path) + await journal.open() + agent_runtime_id = uuid4() + manager, admission = await _new_agent(journal, agent_runtime_id) + descriptor_path = _DISCOVERY / f"{provider_id}.sfp.json" + descriptor = load_provider_descriptor( + descriptor_path, discovery_directory=_DISCOVERY + ) + initial_session = None + manager_closed = False + journal_closed = False + try: + initial_session = await manager.establish(descriptor) + await _available(manager, provider_id) + + # Arm a single real Provider Watch transport fault. Agent's installed + # requester must reconnect its observer without resubmitting admission. + (state / "fail-next-watch").touch() + success_id = await _admit(manager, provider_id, "progress") + await _wait_until( + lambda: ( + "watch:injected-unavailable" in _events(arguments.event_log) + ), + "Installed Agent did not encounter the injected Watch failure", + ) + release = state / f"release-{success_id}" + await _wait_until( + lambda: any( + event == f"waiting:{success_id}" + for event in _events(arguments.event_log) + ), + "Provider did not hold the progress operation", + ) + first_response = asyncio.Event() + half_progress = asyncio.Event() + live_watch = asyncio.create_task( + _watch_to_terminal( + manager, + admission, + provider_id=provider_id, + operation_id=success_id, + first_response=first_response, + half_progress=half_progress, + ) + ) + first_response_wait = asyncio.create_task(first_response.wait()) + completed, _ = await asyncio.wait( + (first_response_wait, live_watch), + timeout=5, + return_when=asyncio.FIRST_COMPLETED, + ) + if live_watch in completed: + await live_watch + if first_response_wait not in completed: + first_response_wait.cancel() + await asyncio.gather(first_response_wait, return_exceptions=True) + live_watch.cancel() + await asyncio.gather(live_watch, return_exceptions=True) + raise AssertionError( + "Agent lifecycle Watch produced no initial observation" + ) + release.touch() + await _wait_until( + lambda: any( + event == f"half:{success_id}" + for event in _events(arguments.event_log) + ), + "Provider did not hold after publishing intermediate progress", + ) + await asyncio.wait_for(half_progress.wait(), timeout=5) + (state / f"release-half-{success_id}").touch() + progress = await live_watch + success = await _status(manager, provider_id, success_id) + if success.result != {"value": 100.0}: + raise AssertionError( + "Agent did not retain the typed terminal result" + ) + if success.progress_percentage != _COMPLETE_PROGRESS: + raise AssertionError("Agent did not retain terminal progress") + progress_percentages = [ + item.snapshot.progress.percentage + for item in progress + if item.snapshot.HasField("progress") + and item.snapshot.progress.HasField("percentage") + ] + if ( + _MIDDLE_PROGRESS not in progress_percentages + or _COMPLETE_PROGRESS not in progress_percentages + ): + raise AssertionError( + "Agent lifecycle client missed progress updates" + ) + + # A new Agent observer against the retained operation receives the + # terminal snapshot; a nonzero cursor reports coalesced gaps. + lease = await manager._leases._operation_lifecycle_lease(provider_id) + attempt = await admission.attempt(provider_id, success_id) + if attempt is None: + raise AssertionError("Agent lost the durable operation identity") + watch_request = operation_lifecycle_pb2.WatchOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=str(provider_id), + expected_endpoint_generation=lease.endpoint_generation, + expected_session_id=lease.session_id, + ), + operation_id=str(success_id), + expected_request_fingerprint=attempt.request_fingerprint, + after_sequence=0, + ) + replay = [ + item + async for item in OperationLifecycleClient().watch( + lease.channel, watch_request + ) + ] + if not replay or replay[-1].snapshot.state != ( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ): + raise AssertionError("Late Agent Watch did not replay completion") + gap_request = operation_lifecycle_pb2.WatchOperationRequest() + gap_request.CopyFrom(watch_request) + gap_request.after_sequence = 1 + gap = [ + item + async for item in OperationLifecycleClient().watch( + lease.channel, gap_request + ) + ] + if not gap or not gap[-1].updates_skipped: + raise AssertionError("Agent Watch did not report a cursor gap") + + failure_id = await _admit(manager, provider_id, "failure") + failure = await _status(manager, provider_id, failure_id) + if failure.failure_code != "stateforge.test.expected_failure": + raise AssertionError("Agent did not retain the typed failure") + if "fixture" not in (failure.failure_explanation or ""): + raise AssertionError("Agent failure explanation was not retained") + + runtime_error_id = await _admit(manager, provider_id, "runtime-error") + runtime_error = await _status(manager, provider_id, runtime_error_id) + if runtime_error.failure_code != "stateforge.operation.provider_error": + raise AssertionError( + "Unexpected Provider exception was not normalized" + ) + if _PROVIDER_ERROR_SENTINEL in ( + runtime_error.failure_explanation or "" + ): + raise AssertionError( + "Provider exception text leaked to Agent status" + ) + + invalid_id = await _admit(manager, provider_id, "invalid-result") + invalid = await _status(manager, provider_id, invalid_id) + if invalid.failure_code != "stateforge.operation.invalid_result": + raise AssertionError( + "Invalid result was not normalized by Provider" + ) + + cancellation_id = await _admit(manager, provider_id, "cancel") + await _wait_until( + lambda: any( + event == f"waiting:{cancellation_id}" + for event in _events(arguments.event_log) + ), + "Provider did not enter the cancellation fixture", + ) + requested = await manager.cancel_operation( + provider_id, cancellation_id + ) + if requested.state != "OPERATION_STATE_CANCELLATION_REQUESTED": + raise AssertionError( + "Agent treated a request as completed cancellation" + ) + cancelled = await _status(manager, provider_id, cancellation_id) + if cancelled.state != "OPERATION_STATE_CANCELLED": + raise AssertionError( + "Provider did not complete cooperative cancellation" + ) + if ( + cancelled.side_effects != "SIDE_EFFECT_STATE_PRESENT" + or cancelled.cleanup != "CLEANUP_STATE_COMPLETED" + ): + raise AssertionError("Agent lost confirmed cancellation effects") + + race_id = await _admit(manager, provider_id, "race") + await _wait_until( + lambda: any( + event == f"waiting:{race_id}" + for event in _events(arguments.event_log) + ), + "Provider did not enter the cancellation-race fixture", + ) + race_release = asyncio.create_task( + asyncio.to_thread((state / f"release-{race_id}").touch) + ) + try: + await manager.cancel_operation(provider_id, race_id) + except (SfpOperationLifecycleTransportError, RuntimeError): + # The result may win the race; require a durable terminal state. + raced_status = await _status(manager, provider_id, race_id) + if raced_status.state not in _TERMINAL: + raise + await race_release + raced = await _status(manager, provider_id, race_id) + if raced.state not in { + "OPERATION_STATE_SUCCEEDED", + "OPERATION_STATE_CANCELLED", + }: + raise AssertionError( + "Cancellation race had no valid terminal state" + ) + + dispatches = [ + line + for line in _events(arguments.event_log) + if line.startswith("dispatch:") + ] + if len(dispatches) != _EXPECTED_DISPATCHES: + raise AssertionError( + f"Expected six dispatches, found {len(dispatches)}" + ) + if len({line.split(":", 2)[1] for line in dispatches}) != ( + _EXPECTED_DISPATCHES + ): + raise AssertionError( + "Provider received duplicate operation dispatch" + ) + + await manager.close() + manager_closed = True + await journal.close() + journal_closed = True + old_invocation = await _unit_invocation_id(arguments.unit_name) + restart = await asyncio.create_subprocess_exec( + "/usr/bin/systemctl", + "restart", + arguments.unit_name, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + _, restart_error = await restart.communicate() + if restart.returncode != 0: + raise AssertionError( + "Provider systemd restart failed: " + + restart_error.decode("utf-8", errors="replace")[-1000:] + ) + await _wait_until( + lambda: ( + _descriptor_generation(provider_id) + > initial_session.endpoint_generation + ), + "Provider restart did not publish a newer endpoint generation", + ) + new_invocation = await _unit_invocation_id(arguments.unit_name) + if new_invocation == old_invocation: + raise AssertionError( + "Provider restart reused its systemd invocation" + ) + restarted = await _spawn_restarted_requester( + provider_id, + success_id, + previous_session_id=initial_session.session_id, + previous_generation=initial_session.endpoint_generation, + agent_runtime_id=agent_runtime_id, + journal_path=arguments.journal_path, + ) + final_dispatches = [ + line + for line in _events(arguments.event_log) + if line.startswith("dispatch:") + ] + if ( + len(final_dispatches) != _EXPECTED_DISPATCHES + or len({line.split(":", 2)[1] for line in final_dispatches}) + != _EXPECTED_DISPATCHES + ): + raise AssertionError( + "Provider redispatched operation work across restart" + ) + result = { + "agent_version": version("stateforge-agent"), + "provider_version": version("stateforge-provider"), + "protocol_version": version("stateforge-protocol"), + "provider_instance_id": str(provider_id), + "endpoint_generation": initial_session.endpoint_generation, + "restarted_endpoint_generation": restarted["endpoint_generation"], + "session_id": initial_session.session_id, + "watch_retry_injected": "watch:injected-unavailable" + in _events(arguments.event_log), + "watch_retry_recovered": success.state + == "OPERATION_STATE_SUCCEEDED", + "progress_sequences": [ + item.snapshot.sequence for item in progress + ], + "late_replay": replay[-1].snapshot.state + == operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + "cursor_gap_replayed": gap[-1].updates_skipped, + "success_result": success.result, + "expected_failure": failure.failure_code, + "sanitized_failure": runtime_error.failure_code, + "invalid_result": invalid.failure_code, + "cancellation_completed": cancelled.state, + "cancellation_effects": cancelled.side_effects + == "SIDE_EFFECT_STATE_PRESENT" + and cancelled.cleanup == "CLEANUP_STATE_COMPLETED", + "cancellation_race_resolved": raced.state, + "dispatches": len(final_dispatches), + "no_redispatch": len( + {line.split(":", 2)[1] for line in final_dispatches} + ) + == len(final_dispatches), + "requester_restarted": restarted["requester_pid"] != os.getpid(), + "provider_restarted": new_invocation != old_invocation, + "previous_provider_invocation_id": old_invocation, + "provider_invocation_id": new_invocation, + "endpoint_generation_increased": restarted["endpoint_generation"] + > initial_session.endpoint_generation, + "restarted_terminal_state": restarted["retained_terminal_state"], + "stale_session_rejected": restarted["stale_session_rejected"], + "restarted_session_id": restarted["session_id"], + } + return result + finally: + if not manager_closed: + await manager.close() + if not journal_closed: + await journal.close() + + +def main() -> None: + """Run lifecycle assertions with the installed Agent runtime.""" + arguments = _arguments() + if arguments.reopen_operation is not None: + if ( + arguments.previous_session_id is None + or arguments.previous_generation is None + or arguments.agent_runtime_id is None + ): + raise SystemExit( + "--previous-session-id, --previous-generation, and " + "--agent-runtime-id are required" + ) + result = asyncio.run( + _restarted_requester( + arguments.provider_instance_id, + arguments.reopen_operation, + previous_session_id=arguments.previous_session_id, + previous_generation=arguments.previous_generation, + agent_runtime_id=arguments.agent_runtime_id, + journal_path=arguments.journal_path, + ) + ) + else: + result = asyncio.run(_run(arguments)) + print(json.dumps(result, indent=2, sort_keys=True)) # noqa: T201 + + +if __name__ == "__main__": + main() diff --git a/.github/actions/scenarios/operation/lifecycle/installed-agent/fixture.py b/.github/actions/scenarios/operation/lifecycle/installed-agent/fixture.py new file mode 100644 index 0000000..8b8e870 --- /dev/null +++ b/.github/actions/scenarios/operation/lifecycle/installed-agent/fixture.py @@ -0,0 +1,183 @@ +"""Serve deterministic Operation Lifecycle behavior to an installed Agent.""" + +from __future__ import annotations + +import argparse +import asyncio +from pathlib import Path + +import grpc + +from stateforge_provider import ( + OperationCancelledError, + OperationCleanup, + OperationContext, + OperationEffects, + OperationFailedError, + OperationProgressPolicy, + OperationSideEffects, + Platform, + Provider, + ProviderModel, + operation, +) +from stateforge_provider._services.operation_lifecycle.service import ( + OperationLifecycleService, +) + + +class Input(ProviderModel): + """Select one deterministic lifecycle behavior.""" + + mode: str + + +class Result(ProviderModel): + """Return a typed integer operation result.""" + + value: int + + +class LifecycleFixture(Provider): + """Expose successful, failed, and cancellable operation behaviors.""" + + identity = "stateforge.test.operation_lifecycle" + display_name = "StateForge Operation Lifecycle 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 operation controls to scenario-owned runtime paths.""" + super().__init__( + discovery_directory=discovery_directory, + state_directory=state_directory, + ) + self._state_directory = state_directory + self._event_log = event_log + + async def on_startup(self) -> None: + """Record that the independent Provider process is ready.""" + self._record("startup") + + async def on_shutdown(self) -> None: + """Record graceful teardown for the scenario evidence.""" + self._record("shutdown") + + @operation( + "stateforge.test/execute@1", + cancellable=True, + progress=OperationProgressPolicy.determinate(), + ) + async def execute( + self, request: Input, context: OperationContext + ) -> Result: + """Execute one success, failure, cancellation, or race path.""" + operation_id = str(context.operation_id) + self._record(f"dispatch:{operation_id}:{request.mode}") + await context.report_progress("Started", percentage=0) + + if request.mode == "failure": + raise OperationFailedError( + code="stateforge.test.expected_failure", + explanation="The fixture operation failed safely.", + effects=OperationEffects( + side_effects=OperationSideEffects.NONE, + cleanup=OperationCleanup.NOT_REQUIRED, + explanation="No external state was changed.", + ), + ) + + if request.mode == "runtime-error": + raise RuntimeError("stateforge-r20-secret-never-in-evidence") + + if request.mode == "progress": + self._record(f"waiting:{operation_id}") + release = self._state_directory / f"release-{operation_id}" + while not release.exists(): # noqa: ASYNC110 + await asyncio.sleep(0.01) + + if request.mode in {"cancel", "race"}: + self._record(f"waiting:{operation_id}") + release = self._state_directory / f"release-{operation_id}" + while not release.exists(): + if context.cancellation_requested: + break + await asyncio.sleep(0.01) + if context.cancellation_requested: + self._record(f"cancelled:{operation_id}") + raise OperationCancelledError( + OperationEffects( + side_effects=OperationSideEffects.PRESENT, + cleanup=OperationCleanup.COMPLETED, + explanation="Temporary fixture work was removed.", + ) + ) + + await context.report_progress("Half", percentage=50) + if request.mode == "progress": + self._record(f"half:{operation_id}") + release_half = ( + self._state_directory / f"release-half-{operation_id}" + ) + while not release_half.exists(): # noqa: ASYNC110 + await asyncio.sleep(0.01) + await context.report_progress("Finished", percentage=100) + self._record(f"finished:{operation_id}") + if request.mode == "invalid-result": + return Result.model_construct(value="not-an-integer") + return Result(value=100) + + 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 main() -> None: + """Run the fixture as a normal standalone Provider process.""" + 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() + LifecycleFixture.identity = arguments.identity + LifecycleFixture.display_name = arguments.display_name + + original_watch = OperationLifecycleService.WatchOperation + retry_marker = arguments.state_directory / "fail-next-watch" + event_log = arguments.event_log + + async def watch_once_unavailable( + service: OperationLifecycleService, + request: object, + context: grpc.aio.ServicerContext[object, object], + ) -> None: + if retry_marker.exists(): + retry_marker.unlink() + event_log.parent.mkdir(parents=True, exist_ok=True) + with event_log.open("a", encoding="utf-8") as stream: + stream.write("watch:injected-unavailable\n") + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "Injected one-time Watch transport failure", + ) + await original_watch(service, request, context) # type: ignore[arg-type] + + OperationLifecycleService.WatchOperation = watch_once_unavailable + LifecycleFixture( + arguments.discovery_directory, + arguments.state_directory, + 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 cb4d848..d95119e 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -235,6 +235,42 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + operation-lifecycle-agent: + name: Operation Lifecycle / 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 Lifecycle scenario + uses: ./.github/actions/scenarios/operation/lifecycle/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-r20 + evidence-directory: evidence/runtime/operation-lifecycle-agent + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-operation-lifecycle-agent + path: evidence/runtime/operation-lifecycle-agent + if-no-files-found: warn + retention-days: ${{ inputs.retention-days }} + trigger-publication-agent: name: Trigger Publication / installed Agent RPM if: ${{ inputs.scenario-tier == 'full' }} diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 38c9bf2..1c6198f 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -168,7 +168,14 @@ The runtime suite currently covers: immutable terminal replay, requested and repeated cancellation, confirmed cancellation effects, unsupported cancellation, invalid-progress normalization, stale-session rejection, canonical request limits, and clean - endpoint removal; and + endpoint removal; +- end-to-end Operation Lifecycle through the exact installed Agent RPM and + exact Provider and Protocol wheels, including Agent-observed progress, + terminal replay, cursor-gap indication, Watch retry after a deterministic + transport interruption without redispatch, typed failure and invalid-result + normalization, cooperative cancellation and a completion/cancellation + race, durable terminal state across a separate installed-Agent requester + process, and rejection of its superseded session; and - end-to-end Trigger Publication through the exact installed Agent RPM and exact Provider and Protocol wheels, including durable `accepted_new`, a lost acknowledgement, Agent crash and restart, immutable replay with @@ -260,6 +267,19 @@ The retained evidence includes exact artifact checksums, Agent and Provider process identities, endpoint generations, journal excerpts, dispositions, dispatch counts, and final cleanup state. +R20 adds an installed-Agent Operation Lifecycle scenario beside the direct +Protocol-peer scenario. It runs the Agent RPM's session manager, admission and +lifecycle coordinators, journal, and lifecycle client in a standalone +requester process against the real Provider's generation-specific Unix socket. +The systemd Agent service is installed and started, but it is not an operation +ingress path; this verifies installed requester-to-Provider behavior, not +daemon-level submission. A second installed-RPM requester process reopens the +same isolated journal after a systemd-managed Provider restart, establishes a +fresh session on the incremented endpoint generation, confirms retained +terminal state, and proves a Watch using the superseded session and generation +is rejected. Slow-observer and observer-capacity behavior remains covered by +the independent Protocol-peer scenario. + ## Evidence and security Build provenance and checksums are retained with the immutable artifact bundle. diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index d699fa3..cf06a66 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 = 13 +SCENARIO_COUNT = 14 DOWNSTREAM_BUNDLE_CONSUMER_COUNT = 3 DEPENDENCY_CHECKOUT_COUNT = 2 @@ -45,6 +45,7 @@ def test_action_library_has_focused_domain_boundaries() -> None: "scenarios/operation/admission", "scenarios/operation/installed-agent", "scenarios/operation/lifecycle", + "scenarios/operation/lifecycle/installed-agent", "scenarios/resilience/crash-restart", "scenarios/smoke/normal-lifecycle", "scenarios/trigger/publication", @@ -227,14 +228,16 @@ 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 any( - f"scenarios/operation/{name}" in str(scenario) - for name in ("admission", "lifecycle") - ): + relative_path = str(scenario.parent.relative_to(ACTIONS)) + if relative_path in { + "scenarios/trigger/publication", + "scenarios/operation/admission", + "scenarios/operation/lifecycle", + }: assert "./.github/actions/environment/install" in content assert "./.github/actions/agent/" not in content assert '"${ISOLATED_PYTHON}" -I' in content - elif "scenarios/trigger/installed-agent" in str(scenario): + elif relative_path == "scenarios/trigger/installed-agent": assert "./.github/actions/agent/install" in content assert "./.github/actions/agent/start" in content assert ( @@ -244,7 +247,10 @@ 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): + elif relative_path in { + "scenarios/operation/installed-agent", + "scenarios/operation/lifecycle/installed-agent", + }: assert "./.github/actions/agent/install" in content assert "./.github/actions/agent/start" in content assert "./.github/actions/provider/start" in content @@ -276,6 +282,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "operation-admission", "operation-admission-agent", "operation-lifecycle", + "operation-lifecycle-agent", } for scenario in scenarios: assert scenario in workflow @@ -400,6 +407,10 @@ def test_operation_admission_exercises_installed_agent_requester() -> None: assert "Operation Admission / installed Agent RPM" in workflow assert "OperationAdmissionCoordinator" in assertion assert "OperationAdmissionClient" in assertion + assert "manager._leases._operation_admission_lease" in assertion + assert ( + "manager._leases._assert_operation_admission_is_current" in assertion + ) assert "OperationAdmissionService.StartOperation" in fixture assert "drop-next-operation-response" in fixture assert "operation-admission-attempt:" in fixture @@ -462,6 +473,57 @@ def test_operation_lifecycle_keeps_the_focused_protocol_peer() -> None: assert "${{ runner.temp }}/sfp-r6" in workflow +def test_r20_installed_agent_lifecycle_is_wired() -> None: + """Cover R20 requester behavior with exact installed runtime artifacts.""" + scenario = read( + ACTIONS + / "scenarios" + / "operation" + / "lifecycle" + / "installed-agent" + / "action.yml" + ) + fixture = read( + ACTIONS + / "scenarios" + / "operation" + / "lifecycle" + / "installed-agent" + / "fixture.py" + ) + assertion = read( + ACTIONS + / "scenarios" + / "operation" + / "lifecycle" + / "installed-agent" + / "assert.py" + ) + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + + assert "./.github/actions/agent/install" in scenario + assert "./.github/actions/provider/start" in scenario + assert "./.github/actions/provider/stop" in scenario + assert "operation-lifecycle-r20.db" in scenario + assert "operation-lifecycle-agent" in workflow + assert "Operation Lifecycle / installed Agent RPM" in workflow + assert "ProviderSessionManager" in assertion + assert "OperationAdmissionCoordinator" in assertion + assert "OperationLifecycleCoordinator" in assertion + assert "OperationLifecycleClient" in assertion + assert "manager._leases._operation_lifecycle_lease" in assertion + assert '"restart"' in assertion + assert "_descriptor_generation" in assertion + assert "--previous-generation" in assertion + assert "stale_session_rejected" in assertion + assert "endpoint_generation_increased" in assertion + assert "late_replay" in assertion + assert "updates_skipped" in assertion + assert "OperationCancelledError" in fixture + assert "stateforge-r20-secret-never-in-evidence" in fixture + assert "watch:injected-unavailable" in fixture + + def test_trigger_publication_uses_exact_installed_agent_runtime() -> None: """Verify R18 across the real RPM, wheels, socket, and journal.""" scenario = read(