From 7ad0ae302682e911ff81a82d3ddcedf25db3c5da Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 15:45:38 +0200 Subject: [PATCH 1/6] feat: implement operation lifecycle --- .../scenarios/operation/lifecycle/action.yml | 45 + .../scenarios/operation/lifecycle/verify.py | 545 ++++++++ .../workflows/provider-runtime-scenarios.yml | 34 + docs/ci-automation.md | 7 +- docs/operation-admission.md | 125 +- docs/reference/declarations.md | 5 +- docs/reference/operations.md | 35 +- docs/runtime-architecture.md | 9 +- mkdocs.yml | 2 +- src/stateforge_provider/__init__.py | 20 +- .../_runtime/configuration.py | 25 + .../_services/manifest/declaration.py | 16 +- .../_services/operation_admission/runtime.py | 100 +- .../_services/operation_admission/service.py | 36 +- .../_services/operation_lifecycle/__init__.py | 1 + .../_services/operation_lifecycle/errors.py | 55 + .../_services/operation_lifecycle/runtime.py | 509 ++++++++ .../_services/operation_lifecycle/service.py | 329 +++++ src/stateforge_provider/declarations.py | 23 +- src/stateforge_provider/operations.py | 188 ++- src/stateforge_provider/provider.py | 14 + tests/test_ci_automation.py | 56 +- tests/test_operation_admission.py | 64 +- tests/test_operation_lifecycle.py | 1143 +++++++++++++++++ 24 files changed, 3296 insertions(+), 90 deletions(-) create mode 100644 .github/actions/scenarios/operation/lifecycle/action.yml create mode 100644 .github/actions/scenarios/operation/lifecycle/verify.py create mode 100644 src/stateforge_provider/_services/operation_lifecycle/__init__.py create mode 100644 src/stateforge_provider/_services/operation_lifecycle/errors.py create mode 100644 src/stateforge_provider/_services/operation_lifecycle/runtime.py create mode 100644 src/stateforge_provider/_services/operation_lifecycle/service.py create mode 100644 tests/test_operation_lifecycle.py diff --git a/.github/actions/scenarios/operation/lifecycle/action.yml b/.github/actions/scenarios/operation/lifecycle/action.yml new file mode 100644 index 0000000..c26398d --- /dev/null +++ b/.github/actions/scenarios/operation/lifecycle/action.yml @@ -0,0 +1,45 @@ +name: Test Operation Lifecycle with a direct Protocol peer +description: Verify installed progress, results, and cancellation over the real Provider endpoint. + +inputs: + provider-artifact-directory: + description: Directory containing the exact Provider build. + required: true + protocol-artifact-directory: + description: Directory containing the exact Protocol build. + 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 Provider and Protocol wheels + id: environment + uses: ./.github/actions/environment/install + with: + provider-artifact-directory: ${{ inputs.provider-artifact-directory }} + protocol-artifact-directory: ${{ inputs.protocol-artifact-directory }} + environment-directory: ${{ inputs.runtime-directory }}/venv + evidence-directory: ${{ inputs.evidence-directory }}/installation + + - name: Exercise Operation Lifecycle over the installed endpoint + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }} + ISOLATED_PYTHON: ${{ steps.environment.outputs.python }} + RUNTIME_DIRECTORY: ${{ inputs.runtime-directory }}/p + run: | + set -euo pipefail + mkdir -p "${EVIDENCE_DIRECTORY}" + evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)" + cd "${RUNNER_TEMP}" + unset PYTHONPATH PYTHONHOME + "${ISOLATED_PYTHON}" -I "${ACTION_PATH}/verify.py" \ + --runtime-directory "${RUNTIME_DIRECTORY}" \ + --output "${evidence_directory}/operation-lifecycle.json" diff --git a/.github/actions/scenarios/operation/lifecycle/verify.py b/.github/actions/scenarios/operation/lifecycle/verify.py new file mode 100644 index 0000000..ff3ad45 --- /dev/null +++ b/.github/actions/scenarios/operation/lifecycle/verify.py @@ -0,0 +1,545 @@ +"""Verify installed R6 behavior through an independent Protocol peer.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +from datetime import UTC, datetime, timedelta +from importlib.metadata import version +from pathlib import Path +from uuid import uuid4 + +import grpc +from google.protobuf.json_format import MessageToDict, ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from grpc_status import rpc_status +from stateforge_protocol.sfp.bootstrap.v1 import ( + bootstrap_pb2, + bootstrap_pb2_grpc, +) +from stateforge_protocol.sfp.manifest.v1 import ( + manifest_pb2, + manifest_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, + operation_admission_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + validation as admission_validation, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + contract as lifecycle_contract, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + operation_lifecycle_pb2, + operation_lifecycle_pb2_grpc, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + validation as lifecycle_validation, +) + +from stateforge_provider import ( + OperationCancelledError, + OperationCleanup, + OperationContext, + OperationEffects, + OperationProgressPolicy, + OperationSideEffects, + Platform, + Provider, + ProviderModel, + operation, +) + +_TERMINAL_STATES = { + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + operation_lifecycle_pb2.OPERATION_STATE_FAILED, + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED, +} +_COMPLETE_PERCENTAGE = 100.0 + + +class Input(ProviderModel): + """Select one deterministic lifecycle path.""" + + mode: str + + +class Result(ProviderModel): + """Return one typed terminal result.""" + + value: int + + +class LifecycleProvider(Provider): + """Expose lifecycle behavior through only the public author API.""" + + identity = "stateforge.test.operation_lifecycle" + display_name = "StateForge Operation Lifecycle CI Provider" + version = "0.1.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__(self, runtime_directory: Path) -> None: + """Bind deterministic author controls to scenario-owned paths.""" + super().__init__( + discovery_directory=runtime_directory / "discovery", + state_directory=runtime_directory / "state", + ) + self.dispatches = 0 + self.success_started = asyncio.Event() + self.success_release = asyncio.Event() + self.cancel_started = asyncio.Event() + self.cancellation_seen = asyncio.Event() + self.cancellation_release = asyncio.Event() + self.unsupported_started = asyncio.Event() + self.unsupported_release = asyncio.Event() + + @operation( + "stateforge.test/execute@1", + cancellable=True, + progress=OperationProgressPolicy.determinate(), + ) + async def execute( + self, + request: Input, + context: OperationContext, + ) -> Result: + """Exercise success, cancellation, and invalid progress.""" + self.dispatches += 1 + if request.mode == "invalid-progress": + await context.report_progress("Half", percentage=50) + await context.report_progress("Backwards", percentage=40) + await context.report_progress("Started", percentage=0) + if request.mode == "cancel": + self.cancel_started.set() + await context.wait_for_cancellation() + self.cancellation_seen.set() + await self.cancellation_release.wait() + raise OperationCancelledError( + OperationEffects( + side_effects=OperationSideEffects.PRESENT, + cleanup=OperationCleanup.COMPLETED, + explanation="Temporary work was removed.", + ) + ) + self.success_started.set() + await self.success_release.wait() + await context.report_progress("Half", percentage=50) + await context.report_progress("Finished", percentage=100) + return Result(value=100) + + @operation("stateforge.test/unsupported@1") + async def unsupported( + self, + request: Input, + context: OperationContext, + ) -> Result: + """Keep one non-cancellable operation active for its decision.""" + del request, context + self.dispatches += 1 + self.unsupported_started.set() + await self.unsupported_release.wait() + return Result(value=1) + + +def _timestamp(value: datetime) -> Timestamp: + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +async def _descriptor( + directory: Path, + provider_task: asyncio.Task[None], +) -> dict[str, object]: + async with asyncio.timeout(5): + while True: + paths = await asyncio.to_thread( + lambda: tuple(directory.glob("*.sfp.json")) + ) + if paths: + return json.loads(await asyncio.to_thread(paths[0].read_text)) + if provider_task.done(): + await provider_task + await asyncio.sleep(0.01) + + +async def _session( + channel: grpc.aio.Channel, + descriptor: dict[str, object], +) -> tuple[ + bootstrap_pb2.HandshakeResponse, + manifest_pb2.GetManifestResponse, +]: + handshake = await bootstrap_pb2_grpc.BootstrapServiceStub( + channel + ).Handshake( + bootstrap_pb2.HandshakeRequest( + agent_runtime=bootstrap_pb2.AgentRuntime( + runtime_id=str(uuid4()), + software_version="0.1.0", + ), + expected_provider_instance_id=descriptor["provider_instance_id"], + expected_endpoint_generation=descriptor["endpoint_generation"], + supported_sfp_versions=[ + bootstrap_pb2.ProtocolVersion(major=1, minor=0) + ], + ), + timeout=3, + ) + manifest = await manifest_pb2_grpc.ManifestServiceStub( + channel + ).GetManifest( + manifest_pb2.GetManifestRequest( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + supported_manifest_schema_versions=[1], + ), + timeout=3, + ) + return handshake, manifest + + +def _start_request( + handshake: bootstrap_pb2.HandshakeResponse, + manifest: manifest_pb2.GetManifestResponse, + *, + mode: str, + identifier: str = "stateforge.test/execute@1", +) -> operation_admission_pb2.StartOperationRequest: + payload = Struct() + ParseDict({"mode": mode}, payload) + return operation_admission_pb2.StartOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + ), + invocation=operation_admission_pb2.OperationInvocation( + operation_id=str(uuid4()), + operation_identifier=identifier, + payload=payload, + execution_deadline=_timestamp( + datetime.now(UTC) + timedelta(minutes=5) + ), + contract=operation_admission_pb2.OperationContractContext( + manifest_revision=manifest.manifest_revision, + ), + execution_context=Struct(), + ), + admission_mode=operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE, + ) + + +def _watch_request( + start: operation_admission_pb2.StartOperationRequest, + *, + after_sequence: int = 0, +) -> operation_lifecycle_pb2.WatchOperationRequest: + return operation_lifecycle_pb2.WatchOperationRequest( + session_binding=start.session_binding, + operation_id=start.invocation.operation_id, + expected_request_fingerprint=( + admission_validation.compute_request_fingerprint(start.invocation) + ), + after_sequence=after_sequence, + ) + + +def _cancel_request( + start: operation_admission_pb2.StartOperationRequest, +) -> operation_lifecycle_pb2.CancelOperationRequest: + return operation_lifecycle_pb2.CancelOperationRequest( + session_binding=start.session_binding, + operation_id=start.invocation.operation_id, + expected_request_fingerprint=( + admission_validation.compute_request_fingerprint(start.invocation) + ), + ) + + +async def _terminal( + lifecycle: operation_lifecycle_pb2_grpc.OperationLifecycleServiceStub, + request: operation_lifecycle_pb2.WatchOperationRequest, +) -> tuple[ + list[operation_lifecycle_pb2.WatchOperationResponse], + operation_lifecycle_pb2.OperationSnapshot, +]: + call = lifecycle.WatchOperation(request, timeout=5) + responses: list[operation_lifecycle_pb2.WatchOperationResponse] = [] + previous_sequence: int | None = None + while True: + response = await call.read() + if response is grpc.aio.EOF: + break + lifecycle_validation.validate_watch_exchange( + request, + response, + previous_sequence=previous_sequence, + ) + responses.append(response) + previous_sequence = response.snapshot.sequence + if response.snapshot.state in _TERMINAL_STATES: + if await call.read() is not grpc.aio.EOF: + raise RuntimeError("terminal Watch remained open") + break + if not responses: + raise RuntimeError("Watch returned no lifecycle snapshot") + return responses, responses[-1].snapshot + + +async def _expect_watch_error( + lifecycle: operation_lifecycle_pb2_grpc.OperationLifecycleServiceStub, + request: operation_lifecycle_pb2.WatchOperationRequest, + code: grpc.StatusCode, + reason: int, +) -> str: + call = lifecycle.WatchOperation(request, timeout=3) + try: + await call.read() + except grpc.aio.AioRpcError as error: + if error.code() is not code: + raise RuntimeError( + f"expected {code.name}, got {error.code().name}" + ) from error + status = rpc_status.from_call(error) + if status is None or not status.details: + raise RuntimeError( + "Lifecycle error omitted typed details" + ) from error + detail = operation_lifecycle_pb2.OperationLifecycleErrorDetail() + if not status.details[0].Unpack(detail) or detail.reason != reason: + raise RuntimeError("Lifecycle error detail mismatch") from error + return error.code().name + raise RuntimeError("Lifecycle Watch unexpectedly succeeded") + + +async def _run( # noqa: PLR0912, PLR0915 + runtime_directory: Path, +) -> dict[str, object]: + discovery = runtime_directory / "discovery" + discovery.mkdir(parents=True, exist_ok=True) + provider = LifecycleProvider(runtime_directory) + provider_task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + try: + 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}", + options=( + ( + "grpc.max_receive_message_length", + lifecycle_contract.MAX_RESPONSE_BYTES, + ), + ("grpc.max_send_message_length", 9 * 1024 * 1024), + ), + ) + await asyncio.wait_for(channel.channel_ready(), timeout=3) + handshake, manifest = await _session(channel, descriptor) + admission = operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ) + lifecycle = operation_lifecycle_pb2_grpc.OperationLifecycleServiceStub( + channel + ) + + success = _start_request(handshake, manifest, mode="success") + await admission.StartOperation(success, timeout=3) + await asyncio.wait_for(provider.success_started.wait(), timeout=3) + success_watch = _watch_request(success) + success_call = lifecycle.WatchOperation(success_watch, timeout=5) + initial = await success_call.read() + lifecycle_validation.validate_watch_exchange(success_watch, initial) + provider.success_release.set() + success_responses = [initial] + previous = initial.snapshot.sequence + while initial.snapshot.state not in _TERMINAL_STATES: + initial = await success_call.read() + lifecycle_validation.validate_watch_exchange( + success_watch, + initial, + previous_sequence=previous, + ) + success_responses.append(initial) + previous = initial.snapshot.sequence + if await success_call.read() is not grpc.aio.EOF: + raise RuntimeError("successful terminal Watch remained open") + success_terminal = success_responses[-1].snapshot + if MessageToDict(success_terminal.success.result) != {"value": 100.0}: + raise RuntimeError("typed terminal result changed") + if success_terminal.progress.percentage != _COMPLETE_PERCENTAGE: + raise RuntimeError("determinate progress did not end at 100") + + replay_request = _watch_request( + success, + after_sequence=success_terminal.sequence, + ) + replay_responses, replay_terminal = await _terminal( + lifecycle, replay_request + ) + lifecycle_validation.validate_terminal_replay( + success_terminal, replay_terminal + ) + + cancellation = _start_request(handshake, manifest, mode="cancel") + await admission.StartOperation(cancellation, timeout=3) + await asyncio.wait_for(provider.cancel_started.wait(), timeout=3) + requested = await lifecycle.CancelOperation( + _cancel_request(cancellation), timeout=3 + ) + lifecycle_validation.validate_cancel_exchange( + _cancel_request(cancellation), requested + ) + await asyncio.wait_for(provider.cancellation_seen.wait(), timeout=3) + repeated = await lifecycle.CancelOperation( + _cancel_request(cancellation), timeout=3 + ) + if repeated.disposition != ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_ALREADY_REQUESTED + ): + raise RuntimeError("repeated cancellation changed its decision") + provider.cancellation_release.set() + _, cancellation_terminal = await _terminal( + lifecycle, _watch_request(cancellation) + ) + if cancellation_terminal.state != ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED + ): + raise RuntimeError("cooperative cancellation was not confirmed") + + unsupported = _start_request( + handshake, + manifest, + mode="unsupported", + identifier="stateforge.test/unsupported@1", + ) + await admission.StartOperation(unsupported, timeout=3) + await asyncio.wait_for(provider.unsupported_started.wait(), timeout=3) + unsupported_response = await lifecycle.CancelOperation( + _cancel_request(unsupported), timeout=3 + ) + if unsupported_response.disposition != ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_UNSUPPORTED + ): + raise RuntimeError("unsupported cancellation was not rejected") + provider.unsupported_release.set() + await _terminal(lifecycle, _watch_request(unsupported)) + + invalid = _start_request(handshake, manifest, mode="invalid-progress") + await admission.StartOperation(invalid, timeout=3) + _, invalid_terminal = await _terminal( + lifecycle, _watch_request(invalid) + ) + if invalid_terminal.failure.code != ( + "stateforge.operation.invalid_progress" + ): + raise RuntimeError("invalid progress was not normalized") + + stale = _watch_request(success) + stale.session_binding.expected_session_id = "stale-session" + stale_status = await _expect_watch_error( + lifecycle, + stale, + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_SESSION_PRECONDITION_FAILED, + ) + oversized = _watch_request(success) + oversized.expected_request_fingerprint = "x" * ( + lifecycle_contract.MAX_REQUEST_BYTES + 1 + ) + oversized_status = await _expect_watch_error( + lifecycle, + oversized, + grpc.StatusCode.INVALID_ARGUMENT, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_INVALID_REQUEST, + ) + + provider.request_shutdown() + await asyncio.wait_for(provider_task, timeout=5) + if await asyncio.to_thread(Path(socket_path).exists): + raise RuntimeError("Provider socket survived shutdown") + if await asyncio.to_thread( + lambda: tuple(discovery.glob("*.sfp.json")) + ): + raise RuntimeError("Provider descriptor survived shutdown") + + return { + "installed_artifacts": { + "stateforge-provider": version("stateforge-provider"), + "stateforge-protocol": version("stateforge-protocol"), + }, + "provider_instance_id": handshake.provider_instance_id, + "endpoint_generation": handshake.endpoint_generation, + "session_id": handshake.session_id, + "success_sequences": [ + response.snapshot.sequence for response in success_responses + ], + "success_result": MessageToDict(success_terminal.success.result), + "terminal_progress": success_terminal.progress.percentage, + "terminal_replay_count": len(replay_responses), + "cancellation_requested": ( + operation_lifecycle_pb2.CancellationDisposition.Name( + requested.disposition + ) + ), + "cancellation_repeated": ( + operation_lifecycle_pb2.CancellationDisposition.Name( + repeated.disposition + ) + ), + "cancellation_terminal": ( + operation_lifecycle_pb2.OperationState.Name( + cancellation_terminal.state + ) + ), + "unsupported_cancellation": ( + operation_lifecycle_pb2.CancellationDisposition.Name( + unsupported_response.disposition + ) + ), + "invalid_progress_code": invalid_terminal.failure.code, + "stale_fence_status": stale_status, + "oversized_request_status": oversized_status, + "author_dispatches": provider.dispatches, + "final_process_state": "stopped", + "descriptor_removed": True, + "socket_removed": True, + } + finally: + provider.request_shutdown() + if not provider_task.done(): + await asyncio.wait_for(provider_task, timeout=5) + if channel is not None: + await channel.close() + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--runtime-directory", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> None: + """Run the installed-wheel scenario and write bounded evidence.""" + arguments = _arguments() + arguments.output.write_text( + json.dumps( + asyncio.run(_run(arguments.runtime_directory)), + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/provider-runtime-scenarios.yml b/.github/workflows/provider-runtime-scenarios.yml index 4a4ff8b..5c5f977 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -165,6 +165,40 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + operation-lifecycle: + name: Operation Lifecycle / direct Protocol peer + if: ${{ inputs.scenario-tier == 'full' }} + runs-on: ubuntu-24.04 + timeout-minutes: 20 + 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 Operation Lifecycle scenario + uses: ./.github/actions/scenarios/operation/lifecycle + with: + provider-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/provider + protocol-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/protocol + runtime-directory: ${{ runner.temp }}/sfp-r6 + evidence-directory: evidence/runtime/operation-lifecycle + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-operation-lifecycle + path: evidence/runtime/operation-lifecycle + 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 d7f4b20..6028b34 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -156,7 +156,12 @@ The runtime suite currently covers: 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; - and +- 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 + cancellation effects, unsupported cancellation, invalid-progress + normalization, stale-session rejection, canonical request limits, and clean + endpoint removal; 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 diff --git a/docs/operation-admission.md b/docs/operation-admission.md index 24462d5..302c51d 100644 --- a/docs/operation-admission.md +++ b/docs/operation-admission.md @@ -1,4 +1,4 @@ -# Operation admission +# Operations, admission, and lifecycle An operation declaration becomes executable when the Provider starts. The runtime exposes SFP Operation Admission on the same local endpoint as @@ -10,7 +10,13 @@ gRPC, session fences, fingerprints, or retry decisions. ```python from stateforge_provider import ( + OperationCancelledError, + OperationCleanup, OperationContext, + OperationEffects, + OperationFailedError, + OperationProgressPolicy, + OperationSideEffects, Platform, Provider, ProviderModel, @@ -45,12 +51,25 @@ class ServiceProvider(Provider): "example.systemd/restart@1", requires=("runtime.systemd.control",), implements=("example.service.v1.restart",), + cancellable=True, + progress=OperationProgressPolicy.determinate(), ) async def restart( self, request: RestartInput, context: OperationContext, ) -> RestartResult: + await context.report_progress("Stopping service", percentage=0) + # Perform Provider-owned work and check cancellation between phases. + if context.cancellation_requested: + raise OperationCancelledError( + OperationEffects( + side_effects=OperationSideEffects.NONE, + cleanup=OperationCleanup.NOT_REQUIRED, + explanation="No service change was started.", + ) + ) + await context.report_progress("Service restarted", percentage=100) return RestartResult(restarted=True) ``` @@ -65,6 +84,8 @@ request. When present, it must be annotated as `OperationContext`. | `execution_deadline` | Absolute deadline carried by the admitted request | | `execution_context` | Recursively read-only JSON metadata approved by the Agent | +It also exposes `report_progress()`, `cancellation_requested`, and +`wait_for_cancellation()`. These methods affect only this admitted operation. The context contains no raw session, transport, or authorization object. A handler cannot mutate it to change the admitted request. @@ -100,17 +121,92 @@ The record is inserted before the handler task is created. Concurrent identical requests therefore converge on one accepted record and exactly one author dispatch. -Admission is not operation completion. R5 owns request acceptance and safe -dispatch; progress, terminal results, errors, and cancellation are reported by -the later operation-lifecycle procedure. +Admission is not operation completion. The lifecycle service retains the +complete current snapshot for the same operation identity. A late observer +receives the latest snapshot immediately; an observer that falls behind sees +`updates_skipped` rather than an unbounded event history. A terminal snapshot +is immutable and replayable without dispatching author code again. + +## Report progress + +Progress is explicit. The runtime sends no synthetic `0%` update and does not +infer completion from `100%`. A handler reports only meaningful changes: + +```python +await context.report_progress("Reading configuration") +await context.report_progress("Applying changes", percentage=50) +``` + +The `progress=` argument on `@operation` is optional: + +| Declaration | Contract | +| --- | --- | +| omitted | Messages and percentages are optional; percentages may move in either direction and success below `100` is valid | +| `OperationProgressPolicy.determinate()` | Every update has a percentage, percentages do not decrease, and success requires a final `100` | +| `OperationProgressPolicy.indeterminate()` | Every update has a message and percentages are forbidden | +| custom `OperationProgressPolicy(...)` | Select percentage mode, monotonicity, zero start, final `100`, progress-on-success, and message requirements independently | + +All percentages must be finite and between `0` and `100`. An invalid update is +not committed. If it escapes the handler, the operation ends with the +sanitized failure code `stateforge.operation.invalid_progress`. + +## Return results or typed failures + +A normal return must be an instance of the result model declared by the +handler annotation. The runtime validates it against the manifest schema +pinned at admission and commits that exact typed value as `SUCCEEDED`. Invalid +results become `stateforge.operation.invalid_result`; their unsafe value is +not published. + +Use `OperationFailedError` for a stable business failure and include explicit +side-effect and cleanup evidence: + +```python +raise OperationFailedError( + code="example.systemd.restart_failed", + explanation="The service did not become active.", + effects=OperationEffects( + side_effects=OperationSideEffects.PRESENT, + cleanup=OperationCleanup.INCOMPLETE, + explanation="The service remains stopped.", + ), +) +``` + +Unexpected exceptions are logged locally and become the sanitized +`stateforge.operation.provider_error`. Secret-bearing payload, result, and +exception values are never copied into remote failures. + +## Cooperate with cancellation + +Set `cancellable=True` only when the handler can stop safely and describe what +remains. Cancellation is a request, not an asynchronous task kill: + +```python +await context.wait_for_cancellation() +await remove_temporary_state() +raise OperationCancelledError( + OperationEffects( + side_effects=OperationSideEffects.NONE, + cleanup=OperationCleanup.COMPLETED, + explanation="Temporary state was removed.", + ) +) +``` + +The runtime returns one atomic disposition: `REQUESTED`, `ALREADY_REQUESTED`, +`ALREADY_TERMINAL`, or `UNSUPPORTED`. It commits `CANCELLED` only after author +code confirms execution has stopped. If useful work finishes after a request, +`SUCCEEDED` or `FAILED` may still win. Cancelling a Watch detaches that observer +and never cancels the business operation. ## Shutdown and failures -Shutdown stops new admissions, keeps retained records resolvable during the -transition, cancels active author tasks, and waits for them before endpoint -cleanup. An exception or invalid return value from author code stays local and -does not revise the immutable admission record. Payloads, execution context, -and secret-bearing fields are never copied into remote errors or CI evidence. +Shutdown stops new admissions, detaches observers, cancels active author tasks, +and waits for them before endpoint cleanup. Runtime shutdown does not fabricate +a business terminal outcome. Retained terminal records remain replayable until +their advertised expiry; active work is not discarded merely because that +time passes. Capacity and retention are bounded by Provider configuration: @@ -118,13 +214,18 @@ Capacity and retention are bounded by Provider configuration: | --- | ---: | | `max_operation_admission_records` | 1,024 | | `max_operation_dispatches` | 128 | +| `max_operation_observers_per_operation` | 16 | | `operation_record_retention_seconds` | 86,400 | | `operation_tombstone_retention_seconds` | 86,400 | +| `operation_observer_delivery_timeout_seconds` | 5 | The admission-record limit covers live records and retained expiry tombstones together. Tombstones are never discarded early to make room because doing so would turn a known expired identity into an ambiguous absent identity. -Callers must treat `RESOURCE_EXHAUSTED` and shutdown-related `UNAVAILABLE` as -retryable only under their own bounded policy. Retrying the same immutable -request is safe because admission resolves by ID and fingerprint. +One current-state slot is retained per observer. A stalled observer is detached +after the bounded delivery timeout without affecting execution or other +observers. Callers must treat `RESOURCE_EXHAUSTED` and shutdown-related +`UNAVAILABLE` as retryable only under their own bounded policy. Retrying the +same immutable request is safe because admission resolves by ID and +fingerprint. diff --git a/docs/reference/declarations.md b/docs/reference/declarations.md index 83c9681..89a41a6 100644 --- a/docs/reference/declarations.md +++ b/docs/reference/declarations.md @@ -17,4 +17,7 @@ Operation methods are asynchronous instance methods with either `(self, request)` or `(self, request, context)`. The request and return annotations must be `ProviderModel` subclasses. The optional third parameter must be annotated as the public `OperationContext`; it exposes only the -operation ID, execution deadline, and recursively immutable execution context. +operation ID, execution deadline, recursively immutable execution context, +explicit progress reporting, and cooperative cancellation state. Setting +`cancellable=True` or `progress=OperationProgressPolicy(...)` requires that +context parameter. Both decorator options are optional. diff --git a/docs/reference/operations.md b/docs/reference/operations.md index 5e12beb..84f9e08 100644 --- a/docs/reference/operations.md +++ b/docs/reference/operations.md @@ -1,13 +1,36 @@ # Operation API +Operation handlers are declared with [`@operation`](declarations.md). The +runtime supplies the typed request and, when declared, an `OperationContext`. +Authors return their annotated result model or raise a typed terminal outcome; +they never construct SFP messages. + +## Execution context + ::: stateforge_provider.operations.OperationContext ::: stateforge_provider.operations.FrozenJsonValue -Operation handlers are declared with -[`@operation`](declarations.md) and receive the Pydantic request model declared -by their signature. The optional `OperationContext` parameter is constructed -only after canonical admission succeeds and is recursively immutable. +## Progress + +::: stateforge_provider.operations.OperationProgressPolicy + +::: stateforge_provider.operations.ProgressPercentageMode + +::: stateforge_provider.operations.OperationProgressError + +## Failures and cancellation + +::: stateforge_provider.operations.OperationFailedError + +::: stateforge_provider.operations.OperationCancelledError + +::: stateforge_provider.operations.OperationEffects + +::: stateforge_provider.operations.OperationSideEffects + +::: stateforge_provider.operations.OperationCleanup -See [Operation admission](../operation-admission.md) for the complete handler, -validation, idempotency, capacity, and shutdown contract. +See [Operations, admission, and lifecycle](../operation-admission.md) for the +complete handler, progress, result, cancellation, idempotency, capacity, and +shutdown contract. diff --git a/docs/runtime-architecture.md b/docs/runtime-architecture.md index 74dc2b6..085ddae 100644 --- a/docs/runtime-architecture.md +++ b/docs/runtime-architecture.md @@ -13,6 +13,7 @@ stateforge_provider/ health/ # Standard gRPC Health adaptation and bounded Watch manifest/ # Manifest declaration, validation, and RPC service operation_admission/ # Private P4 admission and dispatch runtime + operation_lifecycle/ # Private P5 progress, results, and cancellation trigger_publication/ # Private P3 stream, queue, retry, and replay runtime ``` @@ -70,7 +71,13 @@ manifest to their author methods. Its private P4 service validates canonical requests and active runtime fences, re-evaluates referenced requirements, owns the bounded idempotency registry, records admission before dispatch, and passes typed values plus an immutable public `OperationContext` to author code. -It does not own later operation progress or terminal results. + +Operation Lifecycle shares that same admitted-operation registry rather than +creating a second identity store. Its private P5 service owns current-state +snapshots, bounded observers, progress-policy enforcement, typed terminal +results, normalized failures, execution deadlines, and cooperative +cancellation. It retains only the latest complete snapshot, so observer +pressure is bounded and cannot delay Provider-owned business execution. To add a future procedure: diff --git a/mkdocs.yml b/mkdocs.yml index 7330214..26ef914 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -22,7 +22,7 @@ nav: - Manifest declarations: manifest-authoring.md - Health and readiness: health-reporting.md - Trigger publication: trigger-publication.md - - Operation admission: operation-admission.md + - Operations, admission, and lifecycle: operation-admission.md - Package and runtime CI/CD: ci-automation.md - Python API: - Provider: reference/provider.md diff --git a/src/stateforge_provider/__init__.py b/src/stateforge_provider/__init__.py index 591681c..963bd34 100644 --- a/src/stateforge_provider/__init__.py +++ b/src/stateforge_provider/__init__.py @@ -11,7 +11,17 @@ requirement_check, ) from stateforge_provider.health import HealthStatus -from stateforge_provider.operations import OperationContext +from stateforge_provider.operations import ( + OperationCancelledError, + OperationCleanup, + OperationContext, + OperationEffects, + OperationFailedError, + OperationProgressError, + OperationProgressPolicy, + OperationSideEffects, + ProgressPercentageMode, +) from stateforge_provider.provider import Provider from stateforge_provider.requirements import ( RequirementCheckOutcome, @@ -34,8 +44,16 @@ __all__ = [ "HealthStatus", "ManifestError", + "OperationCancelledError", + "OperationCleanup", "OperationContext", + "OperationEffects", + "OperationFailedError", + "OperationProgressError", + "OperationProgressPolicy", + "OperationSideEffects", "Platform", + "ProgressPercentageMode", "ProtocolVersion", "Provider", "ProviderModel", diff --git a/src/stateforge_provider/_runtime/configuration.py b/src/stateforge_provider/_runtime/configuration.py index 6e1fd8d..3d089a5 100644 --- a/src/stateforge_provider/_runtime/configuration.py +++ b/src/stateforge_provider/_runtime/configuration.py @@ -7,6 +7,9 @@ from pathlib import Path from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + contract as lifecycle_contract, +) from stateforge_protocol.sfp.trigger_publication.v1 import ( contract as trigger_contract, ) @@ -25,6 +28,9 @@ class ProviderRuntimeLimits: max_trigger_in_flight_publications: int = 128 max_operation_admission_records: int = 1_024 max_operation_dispatches: int = 128 + max_operation_observers_per_operation: int = ( + lifecycle_contract.MAX_OBSERVERS_PER_OPERATION + ) def __post_init__(self) -> None: """Reject unusable operational limits.""" @@ -38,6 +44,7 @@ def __post_init__(self) -> None: "max_trigger_in_flight_publications", "max_operation_admission_records", "max_operation_dispatches", + "max_operation_observers_per_operation", ): value = getattr(self, name) if ( @@ -62,6 +69,14 @@ def __post_init__(self) -> None: "max_trigger_in_flight_publications must not exceed the " "Trigger Publication contract" ) + if ( + self.max_operation_observers_per_operation + > lifecycle_contract.MAX_OBSERVERS_PER_OPERATION + ): + raise ValueError( + "max_operation_observers_per_operation must not exceed the " + "Operation Lifecycle contract" + ) DEFAULT_RUNTIME_LIMITS = ProviderRuntimeLimits() @@ -79,6 +94,7 @@ class ProviderRuntimeConfiguration: trigger_retry_max_seconds: float = 30.0 operation_record_retention_seconds: float = 86_400.0 operation_tombstone_retention_seconds: float = 86_400.0 + operation_observer_delivery_timeout_seconds: float = 5.0 limits: ProviderRuntimeLimits = field( default_factory=ProviderRuntimeLimits ) @@ -118,6 +134,7 @@ def __post_init__(self) -> None: "trigger_retry_max_seconds", "operation_record_retention_seconds", "operation_tombstone_retention_seconds", + "operation_observer_delivery_timeout_seconds", ): timeout = getattr(self, name) if ( @@ -132,5 +149,13 @@ def __post_init__(self) -> None: "trigger_retry_base_seconds must not exceed " "trigger_retry_max_seconds" ) + if ( + self.operation_observer_delivery_timeout_seconds + > lifecycle_contract.MAX_STALLED_DELIVERY_SECONDS + ): + raise ValueError( + "operation_observer_delivery_timeout_seconds must not exceed " + "the Operation Lifecycle contract" + ) if not isinstance(self.limits, ProviderRuntimeLimits): raise ValueError("limits must be ProviderRuntimeLimits") diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index c2e9d56..896f222 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -40,7 +40,10 @@ _mro_namespace, _Operation, ) -from stateforge_provider.operations import OperationContext +from stateforge_provider.operations import ( + OperationContext, + OperationProgressPolicy, +) from stateforge_provider.triggers import _TriggerSource @@ -55,6 +58,8 @@ class OperationDeclaration: implements: tuple[str, ...] requires: tuple[str, ...] accepts_context: bool + cancellable: bool + progress: OperationProgressPolicy | None def _provider_model(model: object, path: str) -> type[ProviderModel]: @@ -159,6 +164,13 @@ def _operations( # noqa: PLR0912 raise ManifestError( f"{path}: context must be annotated as OperationContext" ) + if (metadata.cancellable or metadata.progress is not None) and not ( + accepts_context + ): + raise ManifestError( + f"{path}: cancellation and progress policies require " + "OperationContext" + ) roles = metadata.implements if ( not isinstance(roles, tuple) @@ -195,6 +207,8 @@ def _operations( # noqa: PLR0912 implements=tuple(sorted(roles)), requires=metadata.requires, accepts_context=accepts_context, + cancellable=metadata.cancellable, + progress=metadata.progress, ) ) return declarations, tuple( diff --git a/src/stateforge_provider/_services/operation_admission/runtime.py b/src/stateforge_provider/_services/operation_admission/runtime.py index 252fff3..73a76dc 100644 --- a/src/stateforge_provider/_services/operation_admission/runtime.py +++ b/src/stateforge_provider/_services/operation_admission/runtime.py @@ -1,4 +1,4 @@ -"""Own atomic in-memory Operation Admission records and dispatch tasks.""" +"""Own atomic admitted-operation records and their lifecycle state.""" from __future__ import annotations @@ -10,12 +10,26 @@ from enum import Enum, auto from google.protobuf.timestamp_pb2 import Timestamp +from jsonschema import Draft202012Validator +from pydantic import JsonValue from stateforge_protocol.sfp.operation_admission.v1 import ( operation_admission_pb2, ) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + contract as lifecycle_contract, +) + +from stateforge_provider._services.operation_lifecycle.runtime import ( + OperationLifecycleRecord, +) +from stateforge_provider.declarations import ProviderModel +from stateforge_provider.operations import ( + OperationContext, + OperationProgressPolicy, +) _LOGGER = logging.getLogger(__name__) -_Dispatch = Callable[[], Awaitable[None]] +_Dispatch = Callable[[OperationContext], Awaitable[ProviderModel]] class AdmissionRegistryFailure(Enum): @@ -45,11 +59,12 @@ class AdmissionDecision: @dataclass(slots=True) class _Entry: - """Retain one immutable record until its advertised expiry.""" + """Retain admission and lifecycle evidence under one identity.""" fingerprint: str record: operation_admission_pb2.OperationAdmissionRecord expires_at: datetime + lifecycle: OperationLifecycleRecord class OperationAdmissionRuntime: @@ -67,9 +82,13 @@ def __init__( maximum_dispatches: int, record_retention_seconds: float, tombstone_retention_seconds: float, + maximum_observers_per_operation: int = ( + lifecycle_contract.MAX_OBSERVERS_PER_OPERATION + ), ) -> None: self._maximum_records = maximum_records self._maximum_dispatches = maximum_dispatches + self._maximum_observers_per_operation = maximum_observers_per_operation self._record_retention = timedelta(seconds=record_retention_seconds) self._tombstone_retention = timedelta( seconds=tombstone_retention_seconds @@ -102,6 +121,14 @@ def admit( operation_id: str, fingerprint: str, endpoint_generation: int, + operation_identifier: str, + manifest_revision: str, + result_model: type[ProviderModel], + result_validator: Draft202012Validator, + cancellable: bool, + progress_policy: OperationProgressPolicy | None, + execution_deadline: datetime, + execution_context: dict[str, JsonValue], dispatch: _Dispatch, ) -> AdmissionDecision: """Atomically insert a record and schedule exactly one dispatch.""" @@ -126,13 +153,29 @@ def admit( record_expires_at=_timestamp(expires_at), admitted_endpoint_generation=endpoint_generation, ) + lifecycle = OperationLifecycleRecord( + admission_record=record, + operation_identifier=operation_identifier, + manifest_revision=manifest_revision, + result_model=result_model, + result_validator=result_validator, + cancellable=cancellable, + progress_policy=progress_policy, + maximum_observers=self._maximum_observers_per_operation, + ) self._records[operation_id] = _Entry( fingerprint=fingerprint, record=_copy_record(record), expires_at=expires_at, + lifecycle=lifecycle, ) task = asyncio.create_task( - self._run_dispatch(operation_id, dispatch), + lifecycle.execute( + operation_id=operation_id, + execution_deadline=execution_deadline, + execution_context=execution_context, + dispatch=dispatch, + ), name=f"stateforge-operation-{operation_id}", ) self._tasks[operation_id] = task @@ -143,6 +186,26 @@ def task_done(completed: asyncio.Task[None]) -> None: task.add_done_callback(task_done) return AdmissionDecision(record, True) + def resolve_lifecycle( + self, + operation_id: str, + fingerprint: str, + ) -> OperationLifecycleRecord: + """Resolve lifecycle state for one exact admitted invocation.""" + now = datetime.now(UTC) + self._prune(now) + entry = self._records.get(operation_id) + if entry is None: + failure = ( + AdmissionRegistryFailure.EXPIRED + if operation_id in self._tombstones + else AdmissionRegistryFailure.UNAVAILABLE + ) + raise AdmissionRegistryError(failure) + if entry.fingerprint != fingerprint: + raise AdmissionRegistryError(AdmissionRegistryFailure.CONFLICT) + return entry.lifecycle + def begin_shutdown(self) -> None: """Reject new admissions while allowing retained resolution.""" self._closed = True @@ -151,41 +214,34 @@ async def close(self) -> None: """Stop admission and cancel every active author dispatch.""" self.begin_shutdown() tasks = tuple(self._tasks.values()) + for entry in self._records.values(): + entry.lifecycle.detach_all() for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) - async def _run_dispatch( - self, - operation_id: str, - dispatch: _Dispatch, - ) -> None: - """Run author behavior without changing immutable admission state.""" - try: - await dispatch() - except asyncio.CancelledError: - raise - except Exception: - _LOGGER.error( - "Provider operation dispatch failed", - extra={"operation_id": operation_id}, - ) - def _task_done( self, operation_id: str, task: asyncio.Task[None], ) -> None: - """Release dispatch capacity without consuming task exceptions.""" + """Release capacity and consume unexpected internal exceptions.""" if self._tasks.get(operation_id) is task: del self._tasks[operation_id] + if task.cancelled(): + return + if task.exception() is not None: + _LOGGER.error( + "Provider operation runtime task failed", + extra={"operation_id": operation_id}, + ) def _prune(self, now: datetime) -> None: """Replace expired records with bounded expiry evidence.""" expired = tuple( identifier for identifier, entry in self._records.items() - if entry.expires_at <= now + if entry.expires_at <= now and entry.lifecycle.terminal ) for identifier in expired: del self._records[identifier] diff --git a/src/stateforge_provider/_services/operation_admission/service.py b/src/stateforge_provider/_services/operation_admission/service.py index e48015a..3b7053c 100644 --- a/src/stateforge_provider/_services/operation_admission/service.py +++ b/src/stateforge_provider/_services/operation_admission/service.py @@ -6,7 +6,6 @@ from collections.abc import Awaitable, Callable from datetime import UTC, datetime from typing import NoReturn, cast -from uuid import UUID import grpc from google.protobuf.json_format import MessageToDict @@ -43,6 +42,7 @@ AdmissionRegistryFailure, OperationAdmissionRuntime, ) +from stateforge_provider.declarations import ProviderModel from stateforge_provider.operations import OperationContext from stateforge_provider.requirements import RequirementCheckOutcome from stateforge_provider.version import ProtocolVersion @@ -221,11 +221,23 @@ async def StartOperation( # noqa: N802, PLR0912 ) dispatch = self._dispatch(operation, invocation, payload) + execution_context = cast( + dict[str, JsonValue], + MessageToDict(invocation.execution_context), + ) try: decision = self._runtime.admit( operation_id=operation_id, fingerprint=fingerprint, endpoint_generation=self._endpoint.endpoint_generation, + operation_identifier=invocation.operation_identifier, + manifest_revision=invocation.contract.manifest_revision, + result_model=operation.declaration.result_model, + result_validator=operation.result_validator, + cancellable=operation.declaration.cancellable, + progress_policy=operation.declaration.progress, + execution_deadline=execution_deadline, + execution_context=execution_context, dispatch=dispatch, ) except AdmissionRegistryError as error: @@ -311,29 +323,17 @@ def _dispatch( operation: BoundOperation, invocation: operation_admission_pb2.OperationInvocation, payload: object, - ) -> Callable[[], Awaitable[None]]: + ) -> Callable[[OperationContext], Awaitable[ProviderModel]]: """Capture typed immutable inputs for post-record dispatch.""" - operation_context = OperationContext._create( - operation_id=UUID(invocation.operation_id), - execution_deadline=invocation.execution_deadline.ToDatetime( - tzinfo=UTC - ), - execution_context=cast( - dict[str, JsonValue], - MessageToDict(invocation.execution_context), - ), - ) - async def dispatch() -> None: + async def dispatch( + operation_context: OperationContext, + ) -> ProviderModel: if operation.declaration.accepts_context: result = await operation.handler(payload, operation_context) else: result = await operation.handler(payload) - if not isinstance(result, operation.declaration.result_model): - raise TypeError("operation returned an invalid result type") - serialized = result.model_dump(mode="json") - if not operation.result_validator.is_valid(serialized): - raise ValueError("operation result violates its schema") + return result return dispatch diff --git a/src/stateforge_provider/_services/operation_lifecycle/__init__.py b/src/stateforge_provider/_services/operation_lifecycle/__init__.py new file mode 100644 index 0000000..07b3b37 --- /dev/null +++ b/src/stateforge_provider/_services/operation_lifecycle/__init__.py @@ -0,0 +1 @@ +"""Implement Provider-owned SFP Operation Lifecycle behavior.""" diff --git a/src/stateforge_provider/_services/operation_lifecycle/errors.py b/src/stateforge_provider/_services/operation_lifecycle/errors.py new file mode 100644 index 0000000..4a031c2 --- /dev/null +++ b/src/stateforge_provider/_services/operation_lifecycle/errors.py @@ -0,0 +1,55 @@ +"""Translate lifecycle lookup decisions into canonical rich gRPC status.""" + +from __future__ import annotations + +from dataclasses import dataclass +from uuid import UUID + +import grpc +from google.protobuf.any_pb2 import Any +from google.rpc import status_pb2 # type: ignore[import-untyped] +from grpc_status import rpc_status +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + operation_lifecycle_pb2, + validation, +) + +_UUID_VERSION = 4 + + +@dataclass(frozen=True, slots=True) +class LifecycleFailureError(Exception): + """Carry one sanitized typed Operation Lifecycle rejection.""" + + code: grpc.StatusCode + reason: operation_lifecycle_pb2.LifecycleErrorReason.ValueType + explanation: str + operation_id: str = "" + + def status(self) -> grpc.Status: + """Build a validated rich status with one versioned detail.""" + detail = operation_lifecycle_pb2.OperationLifecycleErrorDetail( + operation_id=_valid_operation_id(self.operation_id), + reason=self.reason, + explanation=self.explanation, + ) + validation.validate_lifecycle_error_detail(detail) + packed = Any() + packed.Pack(detail) + status = status_pb2.Status( + code=self.code.value[0], + message=self.explanation, + details=(packed,), + ) + return rpc_status.to_status(status) + + +def _valid_operation_id(value: str) -> str: + """Return only a canonical UUIDv4 suitable for an error detail.""" + try: + identifier = UUID(value) + except (ValueError, AttributeError): + return "" + if identifier.version != _UUID_VERSION or str(identifier) != value: + return "" + return value diff --git a/src/stateforge_provider/_services/operation_lifecycle/runtime.py b/src/stateforge_provider/_services/operation_lifecycle/runtime.py new file mode 100644 index 0000000..8a16e28 --- /dev/null +++ b/src/stateforge_provider/_services/operation_lifecycle/runtime.py @@ -0,0 +1,509 @@ +"""Own one admitted operation's state, observers, and author execution.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import TypeAlias +from uuid import UUID + +from google.protobuf.json_format import ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from jsonschema import Draft202012Validator +from pydantic import JsonValue +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + operation_lifecycle_pb2, + validation, +) + +from stateforge_provider.declarations import ProviderModel +from stateforge_provider.operations import ( + OperationCancelledError, + OperationCleanup, + OperationContext, + OperationEffects, + OperationFailedError, + OperationProgressError, + OperationProgressPolicy, + OperationSideEffects, + ProgressPercentageMode, +) + +_LOGGER = logging.getLogger(__name__) +_Dispatch: TypeAlias = Callable[[OperationContext], Awaitable[ProviderModel]] +_TERMINAL_STATES = frozenset( + { + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + operation_lifecycle_pb2.OPERATION_STATE_FAILED, + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED, + } +) +_CANCELLATION_REQUESTED = ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLATION_REQUESTED +) +_COMPLETE_PERCENTAGE = 100.0 + + +class LifecycleRecordError(RuntimeError): + """Report a local operation-record decision.""" + + +class LifecycleObserverCapacityError(LifecycleRecordError): + """Reject an observer when the operation is at capacity.""" + + +class LifecycleCursorAheadError(LifecycleRecordError): + """Reject a cursor newer than retained operation state.""" + + +@dataclass(eq=False, slots=True) +class LifecycleObserver: + """Retain one bounded current-state notification slot.""" + + changed: asyncio.Event = field(default_factory=asyncio.Event) + closed: bool = False + + +class OperationLifecycleRecord: + """Coordinate one admitted operation without retaining event history.""" + + def __init__( + self, + *, + admission_record: operation_admission_pb2.OperationAdmissionRecord, + operation_identifier: str, + manifest_revision: str, + result_model: type[ProviderModel], + result_validator: Draft202012Validator, + cancellable: bool, + progress_policy: OperationProgressPolicy | None, + maximum_observers: int, + ) -> None: + snapshot = operation_lifecycle_pb2.OperationSnapshot( + admission_record=admission_record, + operation_identifier=operation_identifier, + manifest_revision=manifest_revision, + sequence=1, + state=operation_lifecycle_pb2.OPERATION_STATE_ACCEPTED, + updated_at=admission_record.admitted_at, + ) + validation.validate_operation_snapshot(snapshot) + self._snapshot = snapshot.SerializeToString() + self._result_model = result_model + self._result_validator = result_validator + self._cancellable = cancellable + self._progress_policy = progress_policy + self._maximum_observers = maximum_observers + self._cancellation = asyncio.Event() + self._observers: set[LifecycleObserver] = set() + self._last_percentage: float | None = None + + @property + def terminal(self) -> bool: + """Return whether one immutable terminal outcome is retained.""" + return self.snapshot().state in _TERMINAL_STATES + + def snapshot(self) -> operation_lifecycle_pb2.OperationSnapshot: + """Return an independent copy of the complete current state.""" + return operation_lifecycle_pb2.OperationSnapshot.FromString( + self._snapshot + ) + + def attach( + self, + after_sequence: int, + ) -> tuple[LifecycleObserver, operation_lifecycle_pb2.OperationSnapshot]: + """Atomically subscribe and capture current state.""" + current = self.snapshot() + if after_sequence > current.sequence: + raise LifecycleCursorAheadError + if len(self._observers) >= self._maximum_observers: + raise LifecycleObserverCapacityError + observer = LifecycleObserver() + self._observers.add(observer) + return observer, current + + async def next_snapshot( + self, + observer: LifecycleObserver, + after_sequence: int, + ) -> operation_lifecycle_pb2.OperationSnapshot: + """Wait for current state to advance beyond one delivered sequence.""" + while not observer.closed: + current = self.snapshot() + if current.sequence > after_sequence: + return current + observer.changed.clear() + current = self.snapshot() + if current.sequence > after_sequence: + return current + await observer.changed.wait() + raise asyncio.CancelledError + + def detach(self, observer: LifecycleObserver) -> None: + """Detach one observer without changing business execution.""" + if observer in self._observers: + self._observers.remove(observer) + observer.closed = True + observer.changed.set() + + def detach_all(self) -> None: + """Detach every observer during bounded Provider shutdown.""" + for observer in tuple(self._observers): + self.detach(observer) + + def request_cancellation( + self, + ) -> tuple[ + operation_lifecycle_pb2.CancellationDisposition.ValueType, + operation_lifecycle_pb2.OperationSnapshot, + ]: + """Make one atomic cooperative-cancellation decision.""" + current = self.snapshot() + if current.state in _TERMINAL_STATES: + return ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_ALREADY_TERMINAL, + current, + ) + if ( + current.state + == operation_lifecycle_pb2.OPERATION_STATE_CANCELLATION_REQUESTED + ): + return ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_ALREADY_REQUESTED, + current, + ) + if not self._cancellable: + return ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_UNSUPPORTED, + current, + ) + requested = self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLATION_REQUESTED + ) + self._commit(requested) + self._cancellation.set() + return ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED, + self.snapshot(), + ) + + async def execute( + self, + *, + operation_id: str, + execution_deadline: datetime, + execution_context: dict[str, JsonValue], + dispatch: _Dispatch, + ) -> None: + """Run author behavior once and retain one normalized outcome.""" + if self._cancellation.is_set(): + self._commit_cancelled( + OperationEffects( + side_effects=OperationSideEffects.NONE, + cleanup=OperationCleanup.NOT_REQUIRED, + explanation="Execution was prevented from starting.", + ) + ) + return + self._commit( + self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_RUNNING + ) + ) + context = OperationContext._create( + operation_id=UUID(operation_id), + execution_deadline=execution_deadline, + execution_context=execution_context, + reporter=self._report_progress, + cancellation=self._cancellation, + ) + try: + remaining = ( + execution_deadline - datetime.now(UTC) + ).total_seconds() + if remaining <= 0: + raise TimeoutError + async with asyncio.timeout(remaining): + result = await dispatch(context) + if datetime.now(UTC) >= execution_deadline: + raise TimeoutError + self._validate_completion_policy() + self._commit_success(result) + except OperationCancelledError as cancelled: + if self.snapshot().state != _CANCELLATION_REQUESTED: + self._commit_runtime_failure( + "Operation reported cancellation without a request." + ) + else: + self._commit_cancelled(cancelled.effects) + except OperationFailedError as failed: + self._commit_failure( + failed.code, + failed.explanation, + failed.effects, + ) + except OperationProgressError: + self._commit_failure( + "stateforge.operation.invalid_progress", + "Operation progress violated its declared policy.", + _unknown_effects(), + ) + except TimeoutError: + self._commit_failure( + "stateforge.operation.execution_deadline_exceeded", + "Operation execution exceeded its deadline.", + _unknown_effects(), + ) + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.error( + "Provider operation execution failed", + extra={"operation_id": operation_id}, + ) + self._commit_runtime_failure( + "Provider operation execution failed." + ) + + async def _report_progress( + self, + message: str, + percentage: float | None, + ) -> None: + """Validate policy and commit one current-progress replacement.""" + current = self.snapshot() + if current.state not in { + operation_lifecycle_pb2.OPERATION_STATE_RUNNING, + _CANCELLATION_REQUESTED, + }: + raise OperationProgressError( + "progress can only be reported while execution is active" + ) + policy = self._progress_policy + if policy is not None: + if policy.require_message and not message.strip(): + raise OperationProgressError("progress message is required") + if ( + policy.percentage_mode is ProgressPercentageMode.REQUIRED + and percentage is None + ): + raise OperationProgressError("progress percentage is required") + if ( + policy.percentage_mode is ProgressPercentageMode.FORBIDDEN + and percentage is not None + ): + raise OperationProgressError( + "progress percentage is forbidden" + ) + if percentage is not None: + if ( + policy.require_zero_start + and self._last_percentage is None + and percentage != 0 + ): + raise OperationProgressError( + "first progress percentage must be zero" + ) + if ( + policy.monotonic + and self._last_percentage is not None + and percentage < self._last_percentage + ): + raise OperationProgressError( + "progress percentage cannot decrease" + ) + candidate = self._state_snapshot(current.state) + candidate.progress.sequence = candidate.sequence + candidate.progress.updated_at.CopyFrom(candidate.updated_at) + candidate.progress.message = message + if percentage is not None: + candidate.progress.percentage = percentage + self._commit(candidate) + if percentage is not None: + self._last_percentage = percentage + + def _validate_completion_policy(self) -> None: + """Require optional progress evidence before successful completion.""" + policy = self._progress_policy + if policy is None: + return + current = self.snapshot() + if policy.require_progress_on_success and not current.HasField( + "progress" + ): + raise OperationProgressError( + "successful operation requires a progress update" + ) + if ( + policy.require_100_on_success + and self._last_percentage != _COMPLETE_PERCENTAGE + ): + raise OperationProgressError( + "successful operation requires progress to reach 100" + ) + + def _commit_success(self, result: ProviderModel) -> None: + """Validate a typed result against the originally bound schema.""" + if not isinstance(result, self._result_model): + self._commit_invalid_result() + return + serialized = result.model_dump(mode="json") + if not self._result_validator.is_valid(serialized): + self._commit_invalid_result() + return + try: + result_message = ParseDict(serialized, Struct()) + candidate = self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ) + candidate.success.result.CopyFrom(result_message) + self._commit(candidate) + except ( + TypeError, + ValueError, + validation.OperationLifecycleValidationError, + ): + self._commit_invalid_result() + + def _commit_invalid_result(self) -> None: + """Replace unsafe author output with a sanitized terminal failure.""" + self._commit_failure( + "stateforge.operation.invalid_result", + "Operation returned an invalid result.", + _unknown_effects(), + ) + + def _commit_runtime_failure(self, explanation: str) -> None: + """Normalize an unexpected runtime or author failure.""" + self._commit_failure( + "stateforge.operation.provider_error", + explanation, + _unknown_effects(), + ) + + def _commit_failure( + self, + code: str, + explanation: str, + effects: OperationEffects, + ) -> None: + """Commit one sanitized failure or fall back to provider_error.""" + candidate = self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_FAILED + ) + candidate.failure.code = code + candidate.failure.explanation = explanation + _copy_effects(effects, candidate.failure.effects) + try: + self._commit(candidate) + except validation.OperationLifecycleValidationError: + fallback = self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_FAILED + ) + fallback.failure.code = "stateforge.operation.provider_error" + fallback.failure.explanation = ( + "Provider operation execution failed." + ) + _copy_effects(_unknown_effects(), fallback.failure.effects) + self._commit(fallback) + + def _commit_cancelled(self, effects: OperationEffects) -> None: + """Commit cancellation only after execution has stopped.""" + candidate = self._state_snapshot( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED + ) + _copy_effects(effects, candidate.cancellation.effects) + try: + self._commit(candidate) + except validation.OperationLifecycleValidationError: + self._commit_runtime_failure( + "Provider returned invalid cancellation evidence." + ) + + def _state_snapshot( + self, + state: operation_lifecycle_pb2.OperationState.ValueType, + ) -> operation_lifecycle_pb2.OperationSnapshot: + """Build the next complete state while retaining latest progress.""" + current = self.snapshot() + candidate = operation_lifecycle_pb2.OperationSnapshot( + admission_record=current.admission_record, + operation_identifier=current.operation_identifier, + manifest_revision=current.manifest_revision, + sequence=current.sequence + 1, + state=state, + updated_at=_timestamp(datetime.now(UTC)), + ) + if current.HasField("progress"): + candidate.progress.CopyFrom(current.progress) + return candidate + + def _commit( + self, + candidate: operation_lifecycle_pb2.OperationSnapshot, + ) -> None: + """Validate and atomically retain one adjacent state transition.""" + current = self.snapshot() + validation.validate_snapshot_transition(current, candidate) + self._snapshot = candidate.SerializeToString() + for observer in tuple(self._observers): + observer.changed.set() + + +def _timestamp(value: datetime) -> Timestamp: + """Create an independent protobuf timestamp.""" + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +def _unknown_effects() -> OperationEffects: + """Return conservative effects when execution evidence is unavailable.""" + return OperationEffects( + side_effects=OperationSideEffects.UNKNOWN, + cleanup=OperationCleanup.UNKNOWN, + explanation="Side effects and cleanup are unknown.", + ) + + +def _copy_effects( + source: OperationEffects, + target: operation_lifecycle_pb2.OperationEffects, +) -> None: + """Adapt author effects into the private Protocol representation.""" + side_effects = { + OperationSideEffects.NONE: ( + operation_lifecycle_pb2.SIDE_EFFECT_STATE_NONE + ), + OperationSideEffects.PRESENT: ( + operation_lifecycle_pb2.SIDE_EFFECT_STATE_PRESENT + ), + OperationSideEffects.UNKNOWN: ( + operation_lifecycle_pb2.SIDE_EFFECT_STATE_UNKNOWN + ), + } + cleanup = { + OperationCleanup.NOT_REQUIRED: ( + operation_lifecycle_pb2.CLEANUP_STATE_NOT_REQUIRED + ), + OperationCleanup.COMPLETED: ( + operation_lifecycle_pb2.CLEANUP_STATE_COMPLETED + ), + OperationCleanup.INCOMPLETE: ( + operation_lifecycle_pb2.CLEANUP_STATE_INCOMPLETE + ), + OperationCleanup.UNKNOWN: ( + operation_lifecycle_pb2.CLEANUP_STATE_UNKNOWN + ), + } + target.side_effects = side_effects[source.side_effects] + target.cleanup = cleanup[source.cleanup] + target.explanation = source.explanation diff --git a/src/stateforge_provider/_services/operation_lifecycle/service.py b/src/stateforge_provider/_services/operation_lifecycle/service.py new file mode 100644 index 0000000..397632f --- /dev/null +++ b/src/stateforge_provider/_services/operation_lifecycle/service.py @@ -0,0 +1,329 @@ +"""Serve retained operation state and cooperative cancellation over P5.""" + +from __future__ import annotations + +import asyncio +from typing import NoReturn, TypeVar + +import grpc +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + contract, + operation_lifecycle_pb2, + operation_lifecycle_pb2_grpc, + validation, +) + +from stateforge_provider._runtime.grpc import ProcedureRegistration +from stateforge_provider._runtime.identity import ProviderEndpointIdentity +from stateforge_provider._runtime.sessions import SessionRegistry +from stateforge_provider._services.operation_admission.runtime import ( + AdmissionRegistryError, + AdmissionRegistryFailure, + OperationAdmissionRuntime, +) +from stateforge_provider._services.operation_lifecycle.errors import ( + LifecycleFailureError, +) +from stateforge_provider._services.operation_lifecycle.runtime import ( + LifecycleCursorAheadError, + LifecycleObserver, + LifecycleObserverCapacityError, + OperationLifecycleRecord, +) +from stateforge_provider.version import ProtocolVersion + +_TERMINAL_STATES = frozenset( + { + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + operation_lifecycle_pb2.OPERATION_STATE_FAILED, + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED, + } +) +_Request = TypeVar("_Request") +_Response = TypeVar("_Response") + + +class OperationLifecycleService( + operation_lifecycle_pb2_grpc.OperationLifecycleServiceServicer +): + """Observe or cancel work already admitted through P4.""" + + def __init__( + self, + *, + endpoint: ProviderEndpointIdentity, + sessions: SessionRegistry, + runtime: OperationAdmissionRuntime, + delivery_timeout_seconds: float, + ) -> None: + self._endpoint = endpoint + self._sessions = sessions + self._runtime = runtime + self._delivery_timeout_seconds = delivery_timeout_seconds + + async def WatchOperation( # type: ignore[override] # noqa: N802 + self, + request: operation_lifecycle_pb2.WatchOperationRequest, + context: grpc.aio.ServicerContext[ + operation_lifecycle_pb2.WatchOperationRequest, + operation_lifecycle_pb2.WatchOperationResponse, + ], + ) -> None: + """Replay current state and write newer complete snapshots.""" + operation_id = request.operation_id + try: + validation.validate_watch_operation_request(request) + except validation.OperationLifecycleValidationError: + await self._reject( + context, + _failure( + grpc.StatusCode.INVALID_ARGUMENT, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_INVALID_REQUEST, + "Invalid Operation Lifecycle Watch request", + operation_id, + ), + ) + try: + self._validate_fences(request.session_binding, operation_id) + record = self._resolve( + operation_id, + request.expected_request_fingerprint, + ) + except LifecycleFailureError as failure: + await self._reject(context, failure) + observer: LifecycleObserver | None = None + try: + try: + observer, snapshot = record.attach(request.after_sequence) + except LifecycleCursorAheadError: + await self._reject( + context, + _failure( + grpc.StatusCode.OUT_OF_RANGE, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_CURSOR_AHEAD, + "Operation cursor is newer than retained state", + operation_id, + ), + ) + except LifecycleObserverCapacityError: + await self._reject( + context, + _failure( + grpc.StatusCode.RESOURCE_EXHAUSTED, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_OBSERVER_CAPACITY_EXHAUSTED, + "Operation observer capacity is exhausted", + operation_id, + ), + ) + previous_sequence = request.after_sequence + while True: + try: + self._validate_fences( + request.session_binding, operation_id + ) + except LifecycleFailureError as failure: + await self._reject(context, failure) + response = operation_lifecycle_pb2.WatchOperationResponse( + provider_instance_id=str( + self._endpoint.provider_instance_id + ), + endpoint_generation=self._endpoint.endpoint_generation, + session_id=request.session_binding.expected_session_id, + snapshot=snapshot, + updates_skipped=( + previous_sequence > 0 + and snapshot.sequence > previous_sequence + 1 + ), + ) + try: + validation.validate_watch_exchange( + request, + response, + previous_sequence=( + None + if previous_sequence == request.after_sequence + else previous_sequence + ), + ) + async with asyncio.timeout(self._delivery_timeout_seconds): + await context.write(response) + except TimeoutError: + await self._reject( + context, + _failure( + grpc.StatusCode.RESOURCE_EXHAUSTED, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_SLOW_OBSERVER, + "Operation observer delivery stalled", + operation_id, + ), + ) + if snapshot.state in _TERMINAL_STATES: + return + previous_sequence = snapshot.sequence + snapshot = await record.next_snapshot( + observer, + previous_sequence, + ) + finally: + if observer is not None: + record.detach(observer) + + async def CancelOperation( # noqa: N802 + self, + request: operation_lifecycle_pb2.CancelOperationRequest, + context: grpc.aio.ServicerContext[ + operation_lifecycle_pb2.CancelOperationRequest, + operation_lifecycle_pb2.CancelOperationResponse, + ], + ) -> operation_lifecycle_pb2.CancelOperationResponse: + """Return one atomic cooperative-cancellation decision.""" + operation_id = request.operation_id + try: + validation.validate_cancel_operation_request(request) + except validation.OperationLifecycleValidationError: + await self._reject( + context, + _failure( + grpc.StatusCode.INVALID_ARGUMENT, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_INVALID_REQUEST, + "Invalid Operation Lifecycle Cancel request", + operation_id, + ), + ) + try: + self._validate_fences(request.session_binding, operation_id) + record = self._resolve( + operation_id, + request.expected_request_fingerprint, + ) + except LifecycleFailureError as failure: + await self._reject(context, failure) + disposition, snapshot = record.request_cancellation() + response = operation_lifecycle_pb2.CancelOperationResponse( + provider_instance_id=str(self._endpoint.provider_instance_id), + endpoint_generation=self._endpoint.endpoint_generation, + session_id=request.session_binding.expected_session_id, + disposition=disposition, + snapshot=snapshot, + ) + try: + validation.validate_cancel_exchange(request, response) + except validation.OperationLifecycleValidationError: + await context.abort( + grpc.StatusCode.INTERNAL, + "Provider produced an invalid Operation Lifecycle response", + ) + return response + + def _resolve( + self, + operation_id: str, + fingerprint: str, + ) -> OperationLifecycleRecord: + """Resolve one exact retained lifecycle record.""" + try: + return self._runtime.resolve_lifecycle(operation_id, fingerprint) + except AdmissionRegistryError as error: + failures = { + AdmissionRegistryFailure.CONFLICT: _failure( + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_FINGERPRINT_MISMATCH, + "Operation fingerprint does not match retained state", + operation_id, + ), + AdmissionRegistryFailure.EXPIRED: _failure( + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_OPERATION_RECORD_EXPIRED, + "Operation record has expired", + operation_id, + ), + AdmissionRegistryFailure.UNAVAILABLE: _failure( + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + "Operation record is unavailable", + operation_id, + ), + AdmissionRegistryFailure.CAPACITY: _failure( + grpc.StatusCode.RESOURCE_EXHAUSTED, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_OBSERVER_CAPACITY_EXHAUSTED, + "Operation lifecycle capacity is exhausted", + operation_id, + ), + AdmissionRegistryFailure.STOPPING: _failure( + grpc.StatusCode.UNAVAILABLE, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_PROVIDER_UNAVAILABLE, + "Provider operation runtime is stopping", + operation_id, + ), + } + raise failures[error.failure] from error + + def _validate_fences( + self, + binding: operation_admission_pb2.OperationSessionBinding, + operation_id: str, + ) -> None: + """Require the exact active Provider endpoint and SFP session.""" + session = self._sessions.get(binding.expected_session_id) + if session is None or ( + binding.expected_provider_instance_id + != str(self._endpoint.provider_instance_id) + or binding.expected_endpoint_generation + != self._endpoint.endpoint_generation + or session.endpoint != self._endpoint + or session.version != ProtocolVersion(1, 0) + ): + raise _failure( + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_SESSION_PRECONDITION_FAILED, + "Operation Lifecycle session is stale or inactive", + operation_id, + ) + + async def _reject( + self, + context: grpc.aio.ServicerContext[_Request, _Response], + failure: LifecycleFailureError, + ) -> NoReturn: + """Abort one RPC with a bounded versioned error detail.""" + await context.abort_with_status(failure.status()) + raise AssertionError("gRPC abort returned") + + +def _failure( + code: grpc.StatusCode, + reason: operation_lifecycle_pb2.LifecycleErrorReason.ValueType, + explanation: str, + operation_id: str, +) -> LifecycleFailureError: + """Build one sanitized lifecycle failure.""" + return LifecycleFailureError(code, reason, explanation, operation_id) + + +def registration( + *, + endpoint: ProviderEndpointIdentity, + sessions: SessionRegistry, + runtime: OperationAdmissionRuntime, + delivery_timeout_seconds: float, +) -> ProcedureRegistration: + """Build the bounded P5 registration for one Provider generation.""" + service = OperationLifecycleService( + endpoint=endpoint, + sessions=sessions, + runtime=runtime, + delivery_timeout_seconds=delivery_timeout_seconds, + ) + return ProcedureRegistration( + name=contract.SERVICE_NAME, + max_request_bytes=contract.MAX_REQUEST_BYTES, + max_response_bytes=contract.MAX_RESPONSE_BYTES, + install=lambda server: ( + operation_lifecycle_pb2_grpc.add_OperationLifecycleServiceServicer_to_server( + service, server + ) + ), + ) diff --git a/src/stateforge_provider/declarations.py b/src/stateforge_provider/declarations.py index 7924f03..8a1439a 100644 --- a/src/stateforge_provider/declarations.py +++ b/src/stateforge_provider/declarations.py @@ -8,6 +8,8 @@ from pydantic import BaseModel, ConfigDict +from stateforge_provider.operations import OperationProgressPolicy + class ProviderModel(BaseModel): """Describe an object input, result, configuration, or trigger payload. @@ -54,6 +56,8 @@ class _Operation: identifier: str implements: tuple[str, ...] requires: tuple[str, ...] + cancellable: bool + progress: OperationProgressPolicy | None @dataclass(frozen=True, slots=True) @@ -104,6 +108,8 @@ def operation( *, implements: tuple[str, ...] = (), requires: tuple[str, ...] = (), + cancellable: bool = False, + progress: OperationProgressPolicy | None = None, ) -> Callable[[_Handler], _Handler]: """Declare an async method's versioned input/result contract. @@ -117,17 +123,32 @@ def operation( ``requires`` contains stable IDs declared by :func:`requirement_check`. The runtime evaluates each distinct check once per assessment or admission attempt. + ``cancellable`` opts into cooperative operation-scoped cancellation. + ``progress`` optionally adds author-selected consistency requirements; + omitting it preserves the flexible SFP progress contract. Inherited methods follow Python's MRO; overriding requires redeclaration. """ def decorate(handler: _Handler) -> _Handler: if hasattr(handler, "__stateforge_operation__"): raise ValueError("An operation method cannot be declared twice") + if not isinstance(cancellable, bool): + raise TypeError("cancellable must be a boolean") + if progress is not None and not isinstance( + progress, OperationProgressPolicy + ): + raise TypeError("progress must be OperationProgressPolicy") # Attach metadata while retaining the original callable's static type. setattr( # noqa: B010 handler, "__stateforge_operation__", - _Operation(identifier, implements, requires), + _Operation( + identifier, + implements, + requires, + cancellable, + progress, + ), ) return handler diff --git a/src/stateforge_provider/operations.py b/src/stateforge_provider/operations.py index 1fe1316..78571f2 100644 --- a/src/stateforge_provider/operations.py +++ b/src/stateforge_provider/operations.py @@ -2,9 +2,12 @@ from __future__ import annotations -from collections.abc import Mapping +import asyncio +import math +from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass from datetime import datetime +from enum import Enum from types import MappingProxyType from typing import TypeAlias from uuid import UUID @@ -21,6 +24,129 @@ | None ) +_ProgressReporter: TypeAlias = Callable[[str, float | None], Awaitable[None]] +_MAX_PERCENTAGE = 100.0 + + +class ProgressPercentageMode(Enum): + """Control whether an operation reports determinate progress.""" + + OPTIONAL = "optional" + REQUIRED = "required" + FORBIDDEN = "forbidden" + + +@dataclass(frozen=True, slots=True) +class OperationProgressPolicy: + """Apply optional author-selected progress consistency checks. + + Omitting ``progress=`` from :func:`stateforge_provider.operation` keeps + the protocol's flexible behavior. A policy can make percentages + determinate, monotonic, or required to finish at 100 before success. + """ + + percentage_mode: ProgressPercentageMode = ProgressPercentageMode.OPTIONAL + monotonic: bool = False + require_zero_start: bool = False + require_100_on_success: bool = False + require_progress_on_success: bool = False + require_message: bool = False + + def __post_init__(self) -> None: + """Reject contradictory progress requirements.""" + if not isinstance(self.percentage_mode, ProgressPercentageMode): + raise TypeError("percentage_mode must be ProgressPercentageMode") + for name in ( + "monotonic", + "require_zero_start", + "require_100_on_success", + "require_progress_on_success", + "require_message", + ): + if not isinstance(getattr(self, name), bool): + raise TypeError(f"{name} must be a boolean") + if self.percentage_mode is ProgressPercentageMode.FORBIDDEN and ( + self.monotonic + or self.require_zero_start + or self.require_100_on_success + ): + raise ValueError( + "percentage checks cannot be used when percentages are " + "forbidden" + ) + + @classmethod + def determinate(cls) -> OperationProgressPolicy: + """Require monotonic percentages ending at 100 on success.""" + return cls( + percentage_mode=ProgressPercentageMode.REQUIRED, + monotonic=True, + require_100_on_success=True, + ) + + @classmethod + def indeterminate(cls) -> OperationProgressPolicy: + """Require message-only progress without percentages.""" + return cls( + percentage_mode=ProgressPercentageMode.FORBIDDEN, + require_message=True, + ) + + +class OperationProgressError(ValueError): + """Report a rejected author progress update.""" + + +class OperationSideEffects(Enum): + """Describe residual effects after an operation stops.""" + + NONE = "none" + PRESENT = "present" + UNKNOWN = "unknown" + + +class OperationCleanup(Enum): + """Describe cleanup completed before a terminal outcome.""" + + NOT_REQUIRED = "not_required" + COMPLETED = "completed" + INCOMPLETE = "incomplete" + UNKNOWN = "unknown" + + +@dataclass(frozen=True, slots=True) +class OperationEffects: + """Describe side effects and cleanup without implying rollback.""" + + side_effects: OperationSideEffects + cleanup: OperationCleanup + explanation: str + + def __post_init__(self) -> None: + """Reject malformed author-facing effects.""" + if not isinstance(self.side_effects, OperationSideEffects): + raise TypeError("side_effects must be OperationSideEffects") + if not isinstance(self.cleanup, OperationCleanup): + raise TypeError("cleanup must be OperationCleanup") + if not isinstance(self.explanation, str) or not self.explanation: + raise ValueError("explanation must be a non-empty string") + + +@dataclass(frozen=True, slots=True) +class OperationFailedError(Exception): + """End an operation with a stable sanitized business failure.""" + + code: str + explanation: str + effects: OperationEffects + + +@dataclass(frozen=True, slots=True) +class OperationCancelledError(Exception): + """Confirm that cooperatively cancelled execution has stopped.""" + + effects: OperationEffects + @dataclass(frozen=True, slots=True) class OperationContext: @@ -35,6 +161,49 @@ class OperationContext: operation_id: UUID execution_deadline: datetime execution_context: Mapping[str, FrozenJsonValue] + _reporter: _ProgressReporter + _cancellation: asyncio.Event + + @property + def cancellation_requested(self) -> bool: + """Return whether cooperative cancellation has been requested.""" + return self._cancellation.is_set() + + async def wait_for_cancellation(self) -> None: + """Wait until cooperative cancellation is requested.""" + await self._cancellation.wait() + + async def report_progress( + self, + message: str = "", + *, + percentage: float | None = None, + ) -> None: + """Commit one sanitized current-progress update. + + Raises: + OperationProgressError: The update violates the operation's + protocol bounds or optional author-selected progress policy. + """ + if not isinstance(message, str): + raise OperationProgressError("progress message must be a string") + normalized_percentage: float | None = None + if percentage is not None: + if ( + isinstance(percentage, bool) + or not isinstance(percentage, (int, float)) + or not math.isfinite(percentage) + or not 0 <= percentage <= _MAX_PERCENTAGE + ): + raise OperationProgressError( + "progress percentage must be finite and between 0 and 100" + ) + normalized_percentage = float(percentage) + if not message and normalized_percentage is None: + raise OperationProgressError( + "progress requires a message or percentage" + ) + await self._reporter(message, normalized_percentage) @classmethod def _create( @@ -42,6 +211,8 @@ def _create( operation_id: UUID, execution_deadline: datetime, execution_context: dict[str, JsonValue], + reporter: _ProgressReporter, + cancellation: asyncio.Event, ) -> OperationContext: """Create a context with recursively immutable JSON values.""" frozen = { @@ -52,6 +223,8 @@ def _create( operation_id, execution_deadline, MappingProxyType(frozen), + reporter, + cancellation, ) @@ -66,4 +239,15 @@ def _freeze_json(value: JsonValue) -> FrozenJsonValue: return value -__all__ = ["FrozenJsonValue", "OperationContext"] +__all__ = [ + "FrozenJsonValue", + "OperationCancelledError", + "OperationCleanup", + "OperationContext", + "OperationEffects", + "OperationFailedError", + "OperationProgressError", + "OperationProgressPolicy", + "OperationSideEffects", + "ProgressPercentageMode", +] diff --git a/src/stateforge_provider/provider.py b/src/stateforge_provider/provider.py index 1251df4..7142533 100644 --- a/src/stateforge_provider/provider.py +++ b/src/stateforge_provider/provider.py @@ -41,6 +41,9 @@ from stateforge_provider._services.operation_admission.service import ( registration as operation_admission_registration, ) +from stateforge_provider._services.operation_lifecycle.service import ( + registration as operation_lifecycle_registration, +) from stateforge_provider._services.trigger_publication.runtime import ( TriggerPublicationRuntime, ) @@ -233,6 +236,9 @@ async def serve(self) -> None: # noqa: PLR0912, PLR0915 maximum_dispatches=( self._configuration.limits.max_operation_dispatches ), + maximum_observers_per_operation=( + self._configuration.limits.max_operation_observers_per_operation + ), record_retention_seconds=( self._configuration.operation_record_retention_seconds ), @@ -296,6 +302,14 @@ def procedures( operations=bound_operations, runtime=operation_runtime, ), + operation_lifecycle_registration( + endpoint=identity, + sessions=sessions, + runtime=operation_runtime, + delivery_timeout_seconds=( + self._configuration.operation_observer_delivery_timeout_seconds + ), + ), trigger_publication_registration( endpoint=identity, snapshot=snapshot, diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 8225da8..ddf6e66 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 = 11 +SCENARIO_COUNT = 12 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/lifecycle", "scenarios/resilience/crash-restart", "scenarios/smoke/normal-lifecycle", "scenarios/trigger/publication", @@ -227,7 +228,7 @@ def test_each_runtime_scenario_is_an_independent_action() -> None: content = read(scenario) if "scenarios/trigger/publication" in str( scenario - ) or "scenarios/operation/admission" in str(scenario): + ) or "scenarios/operation/" in str(scenario): assert "./.github/actions/environment/install" in content assert "./.github/actions/agent/" not in content assert '"${ISOLATED_PYTHON}" -I' in content @@ -264,6 +265,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "trigger-publication-agent", "capability-assessment", "operation-admission", + "operation-lifecycle", } for scenario in scenarios: assert scenario in workflow @@ -370,6 +372,56 @@ def test_operation_admission_keeps_the_focused_protocol_peer() -> None: assert "${{ runner.temp }}/sfp-r5" in workflow +def test_operation_lifecycle_keeps_the_focused_protocol_peer() -> None: + """Prove R6 installed behavior without simulating a future Agent.""" + scenario = read( + ACTIONS / "scenarios" / "operation" / "lifecycle" / "action.yml" + ) + verifier = read( + ACTIONS / "scenarios" / "operation" / "lifecycle" / "verify.py" + ) + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + + assert "./.github/actions/environment/install" in scenario + assert "./.github/actions/agent/" not in scenario + assert '"${ISOLATED_PYTHON}" -I' in scenario + assert ( + 'evidence_directory="$(cd "${EVIDENCE_DIRECTORY}" && pwd -P)"' + in scenario + ) + assert ( + '--output "${evidence_directory}/operation-lifecycle.json"' in scenario + ) + assert "stateforge_protocol" in verifier + assert "stateforge_agent" not in verifier + assert "@operation(" in verifier + assert "OperationProgressPolicy.determinate()" in verifier + assert "report_progress(" in verifier + assert "CancelOperation" in verifier + assert "WatchOperation" in verifier + for evidence_field in ( + "installed_artifacts", + "success_sequences", + "success_result", + "terminal_progress", + "terminal_replay_count", + "cancellation_requested", + "cancellation_repeated", + "cancellation_terminal", + "unsupported_cancellation", + "invalid_progress_code", + "stale_fence_status", + "oversized_request_status", + "author_dispatches", + "final_process_state", + "descriptor_removed", + "socket_removed", + ): + assert evidence_field in verifier + assert "Operation Lifecycle / direct Protocol peer" in workflow + assert "${{ runner.temp }}/sfp-r6" in workflow + + def test_trigger_publication_uses_exact_installed_agent_runtime() -> None: """Verify R18 across the real RPM, wheels, socket, and journal.""" scenario = read( diff --git a/tests/test_operation_admission.py b/tests/test_operation_admission.py index 8515c16..c456273 100644 --- a/tests/test_operation_admission.py +++ b/tests/test_operation_admission.py @@ -20,6 +20,7 @@ from google.protobuf.struct_pb2 import Struct from google.protobuf.timestamp_pb2 import Timestamp from grpc_status import rpc_status +from jsonschema import Draft202012Validator from pydantic import Field from stateforge_protocol.sfp.bootstrap.v1 import ( bootstrap_pb2, @@ -624,8 +625,33 @@ async def test_dispatch_failure_is_local_and_record_still_resolves( async def test_runtime_retains_expiry_tombstone_and_bounds_capacity() -> None: first_release = asyncio.Event() - async def blocked() -> None: + async def blocked(_context: OperationContext) -> ProviderModel: await first_release.wait() + return _Output(accepted=True) + + result_validator = Draft202012Validator( + _Output.model_json_schema(mode="serialization") + ) + + def admit( + runtime: OperationAdmissionRuntime, + operation_id: str, + fingerprint: str, + ): + return runtime.admit( + operation_id=operation_id, + fingerprint=fingerprint, + endpoint_generation=1, + operation_identifier="example.admission/run@1", + manifest_revision="sha256:" + ("a" * 64), + result_model=_Output, + result_validator=result_validator, + cancellable=False, + progress_policy=None, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=blocked, + ) runtime = OperationAdmissionRuntime( maximum_records=1, @@ -633,18 +659,16 @@ async def blocked() -> None: record_retention_seconds=0.01, tombstone_retention_seconds=0.01, ) - first = runtime.admit( - operation_id=str(uuid4()), - fingerprint="sha256:" + ("1" * 64), - endpoint_generation=1, - dispatch=blocked, + first = admit( + runtime, + str(uuid4()), + "sha256:" + ("1" * 64), ) with pytest.raises(AdmissionRegistryError) as capacity: - runtime.admit( - operation_id=str(uuid4()), - fingerprint="sha256:" + ("2" * 64), - endpoint_generation=1, - dispatch=blocked, + admit( + runtime, + str(uuid4()), + "sha256:" + ("2" * 64), ) assert capacity.value.failure is AdmissionRegistryFailure.CAPACITY @@ -662,22 +686,20 @@ async def blocked() -> None: second_operation_id = str(uuid4()) second_fingerprint = "sha256:" + ("2" * 64) with pytest.raises(AdmissionRegistryError) as tombstone_capacity: - runtime.admit( - operation_id=second_operation_id, - fingerprint=second_fingerprint, - endpoint_generation=1, - dispatch=blocked, + admit( + runtime, + second_operation_id, + second_fingerprint, ) assert ( tombstone_capacity.value.failure is AdmissionRegistryFailure.CAPACITY ) await asyncio.sleep(0.02) - admitted = runtime.admit( - operation_id=second_operation_id, - fingerprint=second_fingerprint, - endpoint_generation=1, - dispatch=blocked, + admitted = admit( + runtime, + second_operation_id, + second_fingerprint, ) assert admitted.accepted_new await runtime.close() diff --git a/tests/test_operation_lifecycle.py b/tests/test_operation_lifecycle.py new file mode 100644 index 0000000..89be992 --- /dev/null +++ b/tests/test_operation_lifecycle.py @@ -0,0 +1,1143 @@ +"""Real-socket evidence for Provider Operation Lifecycle behavior.""" + +from __future__ import annotations + +import asyncio +import json +import shutil +import tempfile +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import cast +from uuid import uuid4 + +import grpc +import pytest +from google.protobuf.json_format import MessageToDict, ParseDict +from google.protobuf.struct_pb2 import Struct +from google.protobuf.timestamp_pb2 import Timestamp +from grpc_status import rpc_status +from jsonschema import Draft202012Validator +from stateforge_protocol.sfp.bootstrap.v1 import ( + bootstrap_pb2, + bootstrap_pb2_grpc, +) +from stateforge_protocol.sfp.manifest.v1 import ( + manifest_pb2, + manifest_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + contract as admission_contract, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + operation_admission_pb2, + operation_admission_pb2_grpc, +) +from stateforge_protocol.sfp.operation_admission.v1 import ( + validation as admission_validation, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + contract as lifecycle_contract, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + operation_lifecycle_pb2, + operation_lifecycle_pb2_grpc, +) +from stateforge_protocol.sfp.operation_lifecycle.v1 import ( + validation as lifecycle_validation, +) + +from stateforge_provider import ( + OperationCancelledError, + OperationCleanup, + OperationContext, + OperationEffects, + OperationFailedError, + OperationProgressPolicy, + OperationSideEffects, + Platform, + ProgressPercentageMode, + ProtocolVersion, + Provider, + ProviderModel, + operation, +) +from stateforge_provider._runtime.identity import ProviderEndpointIdentity +from stateforge_provider._runtime.sessions import SessionRegistry +from stateforge_provider._services.manifest.declaration import ( + ManifestDeclaration, +) +from stateforge_provider._services.manifest.errors import ManifestError +from stateforge_provider._services.operation_admission.runtime import ( + AdmissionRegistryError, + AdmissionRegistryFailure, + OperationAdmissionRuntime, +) +from stateforge_provider._services.operation_lifecycle.runtime import ( + LifecycleObserverCapacityError, + OperationLifecycleRecord, +) +from stateforge_provider._services.operation_lifecycle.service import ( + OperationLifecycleService, +) + +_COMPLETE_PERCENTAGE = 100.0 +_CANCELLING_PERCENTAGE = 50.0 +_CANCELLATION_REQUEST_COUNT = 8 +_REPEATED_CANCELLATION_COUNT = _CANCELLATION_REQUEST_COUNT - 1 +_CURSOR_AHEAD = operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_CURSOR_AHEAD +_FINGERPRINT_MISMATCH = ( + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_FINGERPRINT_MISMATCH +) +_SESSION_PRECONDITION_FAILED = ( + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_SESSION_PRECONDITION_FAILED +) +_CANCELLATION_REQUEST_SEQUENCE = 2 +_PRE_START_TERMINAL_SEQUENCE = 3 +_COALESCED_TERMINAL_SEQUENCE = 5 +_ALREADY_REQUESTED = ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_ALREADY_REQUESTED +) + + +class _Input(ProviderModel): + mode: str + + +class _Output(ProviderModel): + value: int + + +class _LargeOutput(ProviderModel): + value: str + + +class _LifecycleProvider(Provider): + identity = "example.lifecycle" + display_name = "Lifecycle test Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__(self, directory: Path) -> None: + super().__init__( + discovery_directory=directory, + state_directory=directory / "state", + ) + self.calls = 0 + self.started = asyncio.Event() + self.release = asyncio.Event() + self.cancellation_seen = asyncio.Event() + self.cancellation_release = asyncio.Event() + self.context: OperationContext | None = None + + @operation( + "example.lifecycle/run@1", + cancellable=True, + progress=OperationProgressPolicy.determinate(), + ) + async def run_operation( + self, + request: _Input, + context: OperationContext, + ) -> _Output: + self.calls += 1 + self.context = context + if request.mode == "invalid-progress": + await context.report_progress("Half", percentage=50) + await context.report_progress("Backwards", percentage=40) + if request.mode == "no-100": + await context.report_progress("Nearly done", percentage=90) + return _Output(value=90) + await context.report_progress("Started", percentage=0) + self.started.set() + if request.mode in {"cancel", "cancel-success"}: + await context.wait_for_cancellation() + self.cancellation_seen.set() + await context.report_progress("Stopping safely", percentage=50) + await self.cancellation_release.wait() + if request.mode == "cancel-success": + await context.report_progress("Finished", percentage=100) + return _Output(value=100) + raise OperationCancelledError( + OperationEffects( + side_effects=OperationSideEffects.PRESENT, + cleanup=OperationCleanup.COMPLETED, + explanation="Temporary work was removed.", + ) + ) + if request.mode == "failure": + raise OperationFailedError( + code="example.lifecycle.failed", + explanation="The requested lifecycle test failed.", + effects=OperationEffects( + side_effects=OperationSideEffects.NONE, + cleanup=OperationCleanup.NOT_REQUIRED, + explanation="No changes were made.", + ), + ) + if request.mode == "invalid-result": + await context.report_progress("Finished", percentage=100) + return cast(_Output, ProviderModel()) + await self.release.wait() + await context.report_progress("Finished", percentage=100) + return _Output(value=100) + + @operation("example.lifecycle/unsupported@1") + async def unsupported( + self, + request: _Input, + context: OperationContext, + ) -> _Output: + self.calls += 1 + self.context = context + self.started.set() + await self.release.wait() + return _Output(value=1) + + @operation("example.lifecycle/flexible@1") + async def flexible( + self, + request: _Input, + context: OperationContext, + ) -> _Output: + del request + self.calls += 1 + await context.report_progress("Phase one", percentage=100) + await context.report_progress("Phase two", percentage=0) + return _Output(value=2) + + @operation( + "example.lifecycle/indeterminate@1", + progress=OperationProgressPolicy.indeterminate(), + ) + async def indeterminate( + self, + request: _Input, + context: OperationContext, + ) -> _Output: + del request + self.calls += 1 + await context.report_progress("Discovering resources") + return _Output(value=3) + + @operation("example.lifecycle/oversized_result@1") + async def oversized_result( + self, + request: _Input, + context: OperationContext, + ) -> _LargeOutput: + del request, context + self.calls += 1 + return _LargeOutput( + value="x" * (lifecycle_contract.MAX_RESULT_STRING_CODE_POINTS + 1) + ) + + +@pytest.fixture +def endpoint_directory() -> Iterator[Path]: + directory = Path(tempfile.mkdtemp(prefix="sfp-r6-", dir="/tmp")) + try: + yield directory + finally: + shutil.rmtree(directory, ignore_errors=True) + + +@asynccontextmanager +async def _running( + provider: _LifecycleProvider, +) -> AsyncIterator[tuple[object, ...]]: + task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + try: + async with asyncio.timeout(3): + while not list( + provider._configuration.discovery_directory.glob("*.sfp.json") + ): + if task.done(): + await task + await asyncio.sleep(0.001) + descriptor = json.loads(provider._server.descriptor_path.read_text()) + channel = grpc.aio.insecure_channel( + f"unix:{descriptor['transport']['socket_path']}", + options=( + ( + "grpc.max_receive_message_length", + lifecycle_contract.MAX_RESPONSE_BYTES, + ), + ( + "grpc.max_send_message_length", + max( + lifecycle_contract.MAX_REQUEST_BYTES, + admission_contract.MAX_REQUEST_BYTES, + ), + ), + ), + ) + await asyncio.wait_for(channel.channel_ready(), timeout=2) + yield ( + operation_admission_pb2_grpc.OperationAdmissionServiceStub( + channel + ), + operation_lifecycle_pb2_grpc.OperationLifecycleServiceStub( + channel + ), + bootstrap_pb2_grpc.BootstrapServiceStub(channel), + manifest_pb2_grpc.ManifestServiceStub(channel), + descriptor, + ) + finally: + provider.request_shutdown() + await asyncio.wait_for(task, timeout=3) + if channel is not None: + await channel.close() + + +async def _session(bootstrap, manifest, descriptor): + handshake = await bootstrap.Handshake( + bootstrap_pb2.HandshakeRequest( + agent_runtime=bootstrap_pb2.AgentRuntime( + runtime_id=str(uuid4()), + software_version="1.0.0", + ), + expected_provider_instance_id=descriptor["provider_instance_id"], + expected_endpoint_generation=descriptor["endpoint_generation"], + supported_sfp_versions=[ + bootstrap_pb2.ProtocolVersion(major=1, minor=0) + ], + ), + timeout=2, + ) + snapshot = await manifest.GetManifest( + manifest_pb2.GetManifestRequest( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + supported_manifest_schema_versions=[1], + ), + timeout=2, + ) + return handshake, snapshot + + +def _timestamp(value: datetime) -> Timestamp: + timestamp = Timestamp() + timestamp.FromDatetime(value) + return timestamp + + +def _start_request( + handshake, + snapshot, + *, + mode: str = "success", + identifier: str = "example.lifecycle/run@1", +): + payload = Struct() + ParseDict({"mode": mode}, payload) + return operation_admission_pb2.StartOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=handshake.provider_instance_id, + expected_endpoint_generation=handshake.endpoint_generation, + expected_session_id=handshake.session_id, + ), + invocation=operation_admission_pb2.OperationInvocation( + operation_id=str(uuid4()), + operation_identifier=identifier, + payload=payload, + execution_deadline=_timestamp( + datetime.now(UTC) + timedelta(minutes=5) + ), + contract=operation_admission_pb2.OperationContractContext( + manifest_revision=snapshot.manifest_revision, + ), + execution_context=Struct(), + ), + admission_mode=( + operation_admission_pb2.ADMISSION_MODE_ADMIT_OR_RESOLVE + ), + ) + + +def _watch_request(start, handshake, *, after_sequence: int = 0): + return operation_lifecycle_pb2.WatchOperationRequest( + session_binding=start.session_binding, + operation_id=start.invocation.operation_id, + expected_request_fingerprint=( + admission_validation.compute_request_fingerprint(start.invocation) + ), + after_sequence=after_sequence, + ) + + +def _cancel_request(start): + return operation_lifecycle_pb2.CancelOperationRequest( + session_binding=start.session_binding, + operation_id=start.invocation.operation_id, + expected_request_fingerprint=( + admission_validation.compute_request_fingerprint(start.invocation) + ), + ) + + +async def _terminal(call, request, *, previous_sequence=None): + responses = [] + previous = previous_sequence + while True: + response = await call.read() + if response is grpc.aio.EOF: + break + lifecycle_validation.validate_watch_exchange( + request, + response, + previous_sequence=previous, + ) + responses.append(response) + previous = response.snapshot.sequence + if response.snapshot.state in { + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + operation_lifecycle_pb2.OPERATION_STATE_FAILED, + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED, + }: + assert await call.read() is grpc.aio.EOF + break + return responses + + +def _detail(error: grpc.aio.AioRpcError): + status = rpc_status.from_call(error) + assert status is not None + packed = status.details[0] + detail = operation_lifecycle_pb2.OperationLifecycleErrorDetail() + assert packed.Unpack(detail) + lifecycle_validation.validate_lifecycle_error_detail(detail) + return detail + + +@pytest.mark.asyncio +async def test_progress_success_late_replay_and_no_redispatch( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot) + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + + request = _watch_request(start, handshake) + call = lifecycle.WatchOperation(request, timeout=2) + first = await call.read() + assert first.snapshot.state == ( + operation_lifecycle_pb2.OPERATION_STATE_RUNNING + ) + assert first.snapshot.progress.percentage == 0 + + provider.release.set() + responses = [ + first, + *( + await _terminal( + call, + request, + previous_sequence=first.snapshot.sequence, + ) + ), + ] + terminal = responses[-1].snapshot + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ) + assert MessageToDict(terminal.success.result) == {"value": 100.0} + assert terminal.progress.percentage == _COMPLETE_PERCENTAGE + + replay_request = _watch_request( + start, + handshake, + after_sequence=terminal.sequence, + ) + replay_call = lifecycle.WatchOperation(replay_request, timeout=2) + replay = await replay_call.read() + assert replay.snapshot == terminal + assert await replay_call.read() is grpc.aio.EOF + assert provider.calls == 1 + + +@pytest.mark.asyncio +async def test_cancellation_dispositions_and_confirmed_terminal( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot, mode="cancel") + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + + request = _cancel_request(start) + decisions = await asyncio.gather( + *( + lifecycle.CancelOperation(request, timeout=2) + for _ in range(_CANCELLATION_REQUEST_COUNT) + ) + ) + assert ( + sum( + item.disposition + == operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED + for item in decisions + ) + == 1 + ) + assert ( + sum(item.disposition == _ALREADY_REQUESTED for item in decisions) + == _REPEATED_CANCELLATION_COUNT + ) + requested = next( + item + for item in decisions + if item.disposition + == operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED + ) + assert requested.disposition == ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED + ) + assert requested.snapshot.state == ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLATION_REQUESTED + ) + await asyncio.wait_for(provider.cancellation_seen.wait(), timeout=1) + repeated = next( + item + for item in decisions + if item.disposition == _ALREADY_REQUESTED + ) + assert repeated.snapshot.sequence >= requested.snapshot.sequence + assert repeated.snapshot.state == ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLATION_REQUESTED + ) + + provider.cancellation_release.set() + watch = _watch_request(start, handshake) + terminal = ( + await _terminal(lifecycle.WatchOperation(watch, timeout=2), watch) + )[-1].snapshot + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED + ) + assert terminal.cancellation.effects.side_effects == ( + operation_lifecycle_pb2.SIDE_EFFECT_STATE_PRESENT + ) + assert terminal.progress.percentage == _CANCELLING_PERCENTAGE + completed = await lifecycle.CancelOperation(request, timeout=2) + assert completed.disposition == ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_ALREADY_TERMINAL + ) + + +@pytest.mark.asyncio +async def test_success_can_win_after_cancellation_is_requested( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot, mode="cancel-success") + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + requested = await lifecycle.CancelOperation( + _cancel_request(start), timeout=2 + ) + assert requested.disposition == ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED + ) + await asyncio.wait_for(provider.cancellation_seen.wait(), timeout=1) + provider.cancellation_release.set() + watch = _watch_request(start, handshake) + terminal = ( + await _terminal(lifecycle.WatchOperation(watch, timeout=2), watch) + )[-1].snapshot + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ) + + +@pytest.mark.asyncio +async def test_unsupported_cancellation_does_not_signal_execution( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request( + handshake, + snapshot, + identifier="example.lifecycle/unsupported@1", + ) + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + response = await lifecycle.CancelOperation( + _cancel_request(start), timeout=2 + ) + assert response.disposition == ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_UNSUPPORTED + ) + assert provider.context is not None + assert not provider.context.cancellation_requested + provider.release.set() + + +@pytest.mark.parametrize( + ("mode", "code"), + [ + ("invalid-progress", "stateforge.operation.invalid_progress"), + ("no-100", "stateforge.operation.invalid_progress"), + ("invalid-result", "stateforge.operation.invalid_result"), + ("failure", "example.lifecycle.failed"), + ], +) +@pytest.mark.asyncio +async def test_normalizes_terminal_failures( + endpoint_directory: Path, + mode: str, + code: str, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot, mode=mode) + await admission.StartOperation(start, timeout=2) + request = _watch_request(start, handshake) + terminal = ( + await _terminal( + lifecycle.WatchOperation(request, timeout=2), request + ) + )[-1].snapshot + assert terminal.state == operation_lifecycle_pb2.OPERATION_STATE_FAILED + assert terminal.failure.code == code + assert terminal.failure.explanation + + +@pytest.mark.parametrize( + ("identifier", "expected_progress"), + [ + ("example.lifecycle/flexible@1", 0), + ("example.lifecycle/indeterminate@1", None), + ], +) +@pytest.mark.asyncio +async def test_optional_and_indeterminate_progress_policies( + endpoint_directory: Path, + identifier: str, + expected_progress: int | None, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request( + handshake, + snapshot, + identifier=identifier, + ) + await admission.StartOperation(start, timeout=2) + request = _watch_request(start, handshake) + terminal = ( + await _terminal( + lifecycle.WatchOperation(request, timeout=2), request + ) + )[-1].snapshot + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ) + if expected_progress is None: + assert not terminal.progress.HasField("percentage") + else: + assert terminal.progress.percentage == expected_progress + + +@pytest.mark.asyncio +async def test_watch_cancellation_does_not_cancel_business_execution( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot) + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + request = _watch_request(start, handshake) + call = lifecycle.WatchOperation(request) + await call.read() + call.cancel() + await asyncio.sleep(0) + assert provider.context is not None + assert not provider.context.cancellation_requested + + provider.release.set() + reconnect, _ = await _session(bootstrap, manifest, descriptor) + request.session_binding.expected_session_id = reconnect.session_id + terminal = ( + await _terminal( + lifecycle.WatchOperation(request, timeout=2), request + ) + )[-1].snapshot + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + ) + assert provider.calls == 1 + + +@pytest.mark.parametrize("failure", ["cursor", "fingerprint", "session"]) +@pytest.mark.asyncio +async def test_structured_watch_rejections( + endpoint_directory: Path, + failure: str, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot) + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + request = _watch_request(start, handshake) + if failure == "cursor": + request.after_sequence = 999 + elif failure == "fingerprint": + request.expected_request_fingerprint = "sha256:" + ("f" * 64) + else: + request.session_binding.expected_session_id = "stale-session" + call = lifecycle.WatchOperation(request, timeout=2) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await call.read() + detail = _detail(captured.value) + expected = { + "cursor": _CURSOR_AHEAD, + "fingerprint": _FINGERPRINT_MISMATCH, + "session": _SESSION_PRECONDITION_FAILED, + }[failure] + assert detail.reason == expected + provider.release.set() + + +@pytest.mark.parametrize( + ("failure", "code", "reason"), + [ + ( + "invalid", + grpc.StatusCode.INVALID_ARGUMENT, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_INVALID_REQUEST, + ), + ( + "fingerprint", + grpc.StatusCode.FAILED_PRECONDITION, + _FINGERPRINT_MISMATCH, + ), + ( + "session", + grpc.StatusCode.FAILED_PRECONDITION, + _SESSION_PRECONDITION_FAILED, + ), + ( + "unavailable", + grpc.StatusCode.FAILED_PRECONDITION, + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_OPERATION_RECORD_UNAVAILABLE, + ), + ], +) +@pytest.mark.asyncio +async def test_structured_cancel_rejections( + endpoint_directory: Path, + failure: str, + code: grpc.StatusCode, + reason: int, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request(handshake, snapshot) + await admission.StartOperation(start, timeout=2) + await asyncio.wait_for(provider.started.wait(), timeout=1) + request = _cancel_request(start) + if failure == "invalid": + request.operation_id = "" + elif failure == "fingerprint": + request.expected_request_fingerprint = "sha256:" + ("f" * 64) + elif failure == "session": + request.session_binding.expected_session_id = "stale-session" + else: + request.operation_id = str(uuid4()) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await lifecycle.CancelOperation(request, timeout=2) + assert captured.value.code() is code + assert _detail(captured.value).reason == reason + provider.release.set() + + +@pytest.mark.asyncio +async def test_oversized_watch_request_and_result_are_rejected_safely( + endpoint_directory: Path, +) -> None: + provider = _LifecycleProvider(endpoint_directory) + async with _running(provider) as ( + admission, + lifecycle, + bootstrap, + manifest, + descriptor, + ): + handshake, snapshot = await _session(bootstrap, manifest, descriptor) + start = _start_request( + handshake, + snapshot, + identifier="example.lifecycle/oversized_result@1", + ) + await admission.StartOperation(start, timeout=2) + watch = _watch_request(start, handshake) + terminal = ( + await _terminal(lifecycle.WatchOperation(watch, timeout=2), watch) + )[-1].snapshot + assert terminal.state == operation_lifecycle_pb2.OPERATION_STATE_FAILED + assert terminal.failure.code == "stateforge.operation.invalid_result" + + oversized = _watch_request(start, handshake) + oversized.expected_request_fingerprint = "x" * ( + lifecycle_contract.MAX_REQUEST_BYTES + 1 + ) + call = lifecycle.WatchOperation(oversized, timeout=2) + with pytest.raises(grpc.aio.AioRpcError) as captured: + await call.read() + assert captured.value.code() is grpc.StatusCode.INVALID_ARGUMENT + assert _detail(captured.value).reason == ( + operation_lifecycle_pb2.LIFECYCLE_ERROR_REASON_INVALID_REQUEST + ) + + +def test_progress_policy_presets_and_declaration_require_context() -> None: + determinate = OperationProgressPolicy.determinate() + assert determinate.percentage_mode is ProgressPercentageMode.REQUIRED + assert determinate.monotonic + assert determinate.require_100_on_success + indeterminate = OperationProgressPolicy.indeterminate() + assert indeterminate.percentage_mode is ProgressPercentageMode.FORBIDDEN + assert indeterminate.require_message + + class _InvalidProvider(_LifecycleProvider): + @operation( + "example.lifecycle/invalid@1", + cancellable=True, + ) + async def invalid(self, request: _Input) -> _Output: + return _Output(value=1) + + with pytest.raises(ManifestError, match="require OperationContext"): + ManifestDeclaration.discover(_InvalidProvider) + + +def test_rejects_contradictory_progress_policy() -> None: + with pytest.raises(ValueError, match="percentages are forbidden"): + OperationProgressPolicy( + percentage_mode=ProgressPercentageMode.FORBIDDEN, + monotonic=True, + ) + + +def _runtime_record( + *, + cancellable: bool = True, + maximum_observers: int = 1, +) -> OperationLifecycleRecord: + admitted = datetime.now(UTC) + return OperationLifecycleRecord( + admission_record=operation_admission_pb2.OperationAdmissionRecord( + operation_id=str(uuid4()), + request_fingerprint="sha256:" + ("a" * 64), + admitted_at=_timestamp(admitted), + record_expires_at=_timestamp(admitted + timedelta(minutes=5)), + admitted_endpoint_generation=1, + ), + operation_identifier="example.lifecycle/run@1", + manifest_revision="sha256:" + ("b" * 64), + result_model=_Output, + result_validator=Draft202012Validator( + _Output.model_json_schema(mode="serialization") + ), + cancellable=cancellable, + progress_policy=None, + maximum_observers=maximum_observers, + ) + + +@pytest.mark.asyncio +async def test_pre_start_cancellation_prevents_dispatch() -> None: + record = _runtime_record() + calls = 0 + + async def dispatch(_context: OperationContext) -> ProviderModel: + nonlocal calls + calls += 1 + return _Output(value=1) + + disposition, requested = record.request_cancellation() + assert disposition == ( + operation_lifecycle_pb2.CANCELLATION_DISPOSITION_REQUESTED + ) + assert requested.sequence == _CANCELLATION_REQUEST_SEQUENCE + await record.execute( + operation_id=requested.admission_record.operation_id, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=dispatch, + ) + terminal = record.snapshot() + assert calls == 0 + assert terminal.state == ( + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED + ) + assert terminal.sequence == _PRE_START_TERMINAL_SEQUENCE + assert terminal.cancellation.effects.side_effects == ( + operation_lifecycle_pb2.SIDE_EFFECT_STATE_NONE + ) + + +@pytest.mark.asyncio +async def test_observer_capacity_and_current_state_coalescing() -> None: + record = _runtime_record(maximum_observers=1) + observer, accepted = record.attach(0) + with pytest.raises(LifecycleObserverCapacityError): + record.attach(0) + + async def dispatch(context: OperationContext) -> ProviderModel: + await context.report_progress("First", percentage=10) + await context.report_progress("Second", percentage=20) + return _Output(value=2) + + await record.execute( + operation_id=accepted.admission_record.operation_id, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=dispatch, + ) + current = await record.next_snapshot(observer, accepted.sequence) + assert current.state == operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED + assert current.sequence == _COALESCED_TERMINAL_SEQUENCE + assert current.progress.message == "Second" + record.detach(observer) + + +@pytest.mark.asyncio +async def test_observers_are_isolated_and_shutdown_detaches_waiters() -> None: + record = _runtime_record(maximum_observers=2) + first, accepted = record.attach(0) + second, _ = record.attach(0) + record.detach(first) + + waiting = asyncio.create_task( + record.next_snapshot(second, accepted.sequence) + ) + await asyncio.sleep(0) + record.detach_all() + with pytest.raises(asyncio.CancelledError): + await waiting + assert first.closed + assert second.closed + + +@pytest.mark.asyncio +async def test_active_record_survives_expiry_then_maps_to_expired() -> None: + release = asyncio.Event() + + async def dispatch(_context: OperationContext) -> ProviderModel: + await release.wait() + return _Output(value=1) + + runtime = OperationAdmissionRuntime( + maximum_records=1, + maximum_dispatches=1, + record_retention_seconds=0.01, + tombstone_retention_seconds=1, + ) + operation_id = str(uuid4()) + fingerprint = "sha256:" + ("d" * 64) + decision = runtime.admit( + operation_id=operation_id, + fingerprint=fingerprint, + endpoint_generation=1, + operation_identifier="example.lifecycle/run@1", + manifest_revision="sha256:" + ("e" * 64), + result_model=_Output, + result_validator=Draft202012Validator( + _Output.model_json_schema(mode="serialization") + ), + cancellable=False, + progress_policy=None, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=dispatch, + ) + await asyncio.sleep(0.02) + assert runtime.resolve_lifecycle(operation_id, fingerprint).snapshot() + assert runtime.lookup(operation_id, fingerprint) is not None + + release.set() + await asyncio.sleep(0) + await asyncio.sleep(0) + with pytest.raises(AdmissionRegistryError) as captured: + runtime.resolve_lifecycle(operation_id, fingerprint) + assert captured.value.failure is AdmissionRegistryFailure.EXPIRED + assert decision.accepted_new + await runtime.close() + + +@pytest.mark.asyncio +async def test_slow_observer_is_detached_without_stopping_execution() -> None: + release = asyncio.Event() + + async def dispatch(_context: OperationContext) -> ProviderModel: + await release.wait() + return _Output(value=1) + + endpoint = ProviderEndpointIdentity(uuid4(), 1) + sessions = SessionRegistry(endpoint) + session_id = sessions.negotiate(ProtocolVersion(1, 0)) + runtime = OperationAdmissionRuntime( + maximum_records=1, + maximum_dispatches=1, + record_retention_seconds=60, + tombstone_retention_seconds=60, + ) + operation_id = str(uuid4()) + fingerprint = "sha256:" + ("f" * 64) + runtime.admit( + operation_id=operation_id, + fingerprint=fingerprint, + endpoint_generation=1, + operation_identifier="example.lifecycle/run@1", + manifest_revision="sha256:" + ("a" * 64), + result_model=_Output, + result_validator=Draft202012Validator( + _Output.model_json_schema(mode="serialization") + ), + cancellable=False, + progress_policy=None, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=dispatch, + ) + service = OperationLifecycleService( + endpoint=endpoint, + sessions=sessions, + runtime=runtime, + delivery_timeout_seconds=0.01, + ) + request = operation_lifecycle_pb2.WatchOperationRequest( + session_binding=operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=str(endpoint.provider_instance_id), + expected_endpoint_generation=endpoint.endpoint_generation, + expected_session_id=session_id, + ), + operation_id=operation_id, + expected_request_fingerprint=fingerprint, + ) + + class _AbortedError(Exception): + def __init__(self, status) -> None: + self.status = status + + class _StalledContext: + async def write(self, _response) -> None: + await asyncio.Event().wait() + + async def abort_with_status(self, status) -> None: + raise _AbortedError(status) + + with pytest.raises(_AbortedError) as captured: + await service.WatchOperation(request, _StalledContext()) # type: ignore[arg-type] + assert captured.value.status.code is grpc.StatusCode.RESOURCE_EXHAUSTED + assert not runtime.resolve_lifecycle(operation_id, fingerprint).terminal + release.set() + await asyncio.sleep(0) + await asyncio.sleep(0) + assert runtime.resolve_lifecycle(operation_id, fingerprint).terminal + await runtime.close() + + +@pytest.mark.asyncio +async def test_execution_deadline_produces_sanitized_failure() -> None: + record = _runtime_record(cancellable=False) + cancelled = asyncio.Event() + + async def dispatch(_context: OperationContext) -> ProviderModel: + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + return _Output(value=1) + + operation_id = record.snapshot().admission_record.operation_id + await record.execute( + operation_id=operation_id, + execution_deadline=datetime.now(UTC) + timedelta(milliseconds=10), + execution_context={}, + dispatch=dispatch, + ) + assert cancelled.is_set() + terminal = record.snapshot() + assert terminal.state == operation_lifecycle_pb2.OPERATION_STATE_FAILED + assert terminal.failure.code == ( + "stateforge.operation.execution_deadline_exceeded" + ) From 2f8991042151bebdea0166b56e25db53d7852de1 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 15:58:10 +0200 Subject: [PATCH 2/6] refactor: use protocol UUID validation --- .../_services/operation_admission/errors.py | 12 ++---------- .../_services/operation_lifecycle/errors.py | 12 ++---------- 2 files changed, 4 insertions(+), 20 deletions(-) diff --git a/src/stateforge_provider/_services/operation_admission/errors.py b/src/stateforge_provider/_services/operation_admission/errors.py index f56f1fc..3a49234 100644 --- a/src/stateforge_provider/_services/operation_admission/errors.py +++ b/src/stateforge_provider/_services/operation_admission/errors.py @@ -3,19 +3,17 @@ from __future__ import annotations from dataclasses import dataclass -from uuid import UUID import grpc from google.protobuf.any_pb2 import Any from google.rpc import status_pb2 # type: ignore[import-untyped] from grpc_status import rpc_status +from stateforge_protocol.sfp import primitives from stateforge_protocol.sfp.operation_admission.v1 import ( operation_admission_pb2, validation, ) -_UUID_VERSION = 4 - @dataclass(frozen=True, slots=True) class AdmissionFailureError(Exception): @@ -48,10 +46,4 @@ def status(self) -> grpc.Status: def _valid_operation_id(value: str) -> str: """Return only a canonical UUIDv4 suitable for an error detail.""" - try: - identifier = UUID(value) - except (ValueError, AttributeError): - return "" - if identifier.version != _UUID_VERSION or str(identifier) != value: - return "" - return value + return value if primitives.is_canonical_uuid4(value) else "" diff --git a/src/stateforge_provider/_services/operation_lifecycle/errors.py b/src/stateforge_provider/_services/operation_lifecycle/errors.py index 4a031c2..5d31a7e 100644 --- a/src/stateforge_provider/_services/operation_lifecycle/errors.py +++ b/src/stateforge_provider/_services/operation_lifecycle/errors.py @@ -3,19 +3,17 @@ from __future__ import annotations from dataclasses import dataclass -from uuid import UUID import grpc from google.protobuf.any_pb2 import Any from google.rpc import status_pb2 # type: ignore[import-untyped] from grpc_status import rpc_status +from stateforge_protocol.sfp import primitives from stateforge_protocol.sfp.operation_lifecycle.v1 import ( operation_lifecycle_pb2, validation, ) -_UUID_VERSION = 4 - @dataclass(frozen=True, slots=True) class LifecycleFailureError(Exception): @@ -46,10 +44,4 @@ def status(self) -> grpc.Status: def _valid_operation_id(value: str) -> str: """Return only a canonical UUIDv4 suitable for an error detail.""" - try: - identifier = UUID(value) - except (ValueError, AttributeError): - return "" - if identifier.version != _UUID_VERSION or str(identifier) != value: - return "" - return value + return value if primitives.is_canonical_uuid4(value) else "" From 3f8f5147b888d6d1f90f6ea5f33ac6cef49e43f4 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 16:05:59 +0200 Subject: [PATCH 3/6] ci: migrate macOS runner to macos-15 --- .github/workflows/provider-installed-test.yml | 2 +- tests/test_ci_automation.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/provider-installed-test.yml b/.github/workflows/provider-installed-test.yml index 37c7937..f43b650 100644 --- a/.github/workflows/provider-installed-test.yml +++ b/.github/workflows/provider-installed-test.yml @@ -32,7 +32,7 @@ jobs: matrix: os: - ubuntu-24.04 - - macos-14 + - macos-15 - windows-2022 python: - "3.11" diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index ddf6e66..483979c 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -625,7 +625,7 @@ def test_package_matrix_is_separate_from_unix_runtime() -> None: runtime = read(WORKFLOWS / "provider-runtime-scenarios.yml") assert "windows-2022" in installed - assert "macos-14" in installed + assert "macos-15" in installed assert installed.count(' - "3.1') == SUPPORTED_PYTHON_COUNT assert "windows-" not in runtime assert "macos-" not in runtime From 0f3b66234e1f6a4661dd4c18532aea379aeb656c Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 16:20:55 +0200 Subject: [PATCH 4/6] fix: classify invalid progress correctly --- .../_services/operation_lifecycle/runtime.py | 7 ++++++- tests/test_operation_lifecycle.py | 6 ++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/stateforge_provider/_services/operation_lifecycle/runtime.py b/src/stateforge_provider/_services/operation_lifecycle/runtime.py index 8a16e28..8cdabe5 100644 --- a/src/stateforge_provider/_services/operation_lifecycle/runtime.py +++ b/src/stateforge_provider/_services/operation_lifecycle/runtime.py @@ -326,7 +326,12 @@ async def _report_progress( candidate.progress.message = message if percentage is not None: candidate.progress.percentage = percentage - self._commit(candidate) + try: + self._commit(candidate) + except validation.OperationLifecycleValidationError as error: + raise OperationProgressError( + "progress violates the Operation Lifecycle contract" + ) from error if percentage is not None: self._last_percentage = percentage diff --git a/tests/test_operation_lifecycle.py b/tests/test_operation_lifecycle.py index 89be992..f6a1998 100644 --- a/tests/test_operation_lifecycle.py +++ b/tests/test_operation_lifecycle.py @@ -147,6 +147,11 @@ async def run_operation( if request.mode == "invalid-progress": await context.report_progress("Half", percentage=50) await context.report_progress("Backwards", percentage=40) + if request.mode == "oversized-progress": + await context.report_progress( + "x" * (lifecycle_contract.MAX_EXPLANATION_CODE_POINTS + 1), + percentage=0, + ) if request.mode == "no-100": await context.report_progress("Nearly done", percentage=90) return _Output(value=90) @@ -615,6 +620,7 @@ async def test_unsupported_cancellation_does_not_signal_execution( ("mode", "code"), [ ("invalid-progress", "stateforge.operation.invalid_progress"), + ("oversized-progress", "stateforge.operation.invalid_progress"), ("no-100", "stateforge.operation.invalid_progress"), ("invalid-result", "stateforge.operation.invalid_result"), ("failure", "example.lifecycle.failed"), From 825e514e7cad19cef2985b598ff109c6b5c6a7a8 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 16:22:38 +0200 Subject: [PATCH 5/6] fix: distinguish operation timeouts from deadlines --- .../_services/operation_lifecycle/runtime.py | 43 ++++++++++++++----- tests/test_operation_lifecycle.py | 19 ++++++++ 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/src/stateforge_provider/_services/operation_lifecycle/runtime.py b/src/stateforge_provider/_services/operation_lifecycle/runtime.py index 8cdabe5..7d091cc 100644 --- a/src/stateforge_provider/_services/operation_lifecycle/runtime.py +++ b/src/stateforge_provider/_services/operation_lifecycle/runtime.py @@ -63,6 +63,10 @@ class LifecycleCursorAheadError(LifecycleRecordError): """Reject a cursor newer than retained operation state.""" +class _ExecutionDeadlineExceededError(RuntimeError): + """Identify expiry of the runtime-owned operation deadline.""" + + @dataclass(eq=False, slots=True) class LifecycleObserver: """Retain one bounded current-state notification slot.""" @@ -226,15 +230,11 @@ async def execute( cancellation=self._cancellation, ) try: - remaining = ( - execution_deadline - datetime.now(UTC) - ).total_seconds() - if remaining <= 0: - raise TimeoutError - async with asyncio.timeout(remaining): - result = await dispatch(context) - if datetime.now(UTC) >= execution_deadline: - raise TimeoutError + result = await _dispatch_before_deadline( + execution_deadline=execution_deadline, + context=context, + dispatch=dispatch, + ) self._validate_completion_policy() self._commit_success(result) except OperationCancelledError as cancelled: @@ -256,7 +256,7 @@ async def execute( "Operation progress violated its declared policy.", _unknown_effects(), ) - except TimeoutError: + except _ExecutionDeadlineExceededError: self._commit_failure( "stateforge.operation.execution_deadline_exceeded", "Operation execution exceeded its deadline.", @@ -470,6 +470,29 @@ def _timestamp(value: datetime) -> Timestamp: return timestamp +async def _dispatch_before_deadline( + *, + execution_deadline: datetime, + context: OperationContext, + dispatch: _Dispatch, +) -> ProviderModel: + """Run author behavior under only the runtime-owned deadline.""" + remaining = (execution_deadline - datetime.now(UTC)).total_seconds() + if remaining <= 0: + raise _ExecutionDeadlineExceededError + deadline = asyncio.timeout(remaining) + try: + async with deadline: + result = await dispatch(context) + except TimeoutError: + if deadline.expired(): + raise _ExecutionDeadlineExceededError from None + raise + if datetime.now(UTC) >= execution_deadline: + raise _ExecutionDeadlineExceededError + return result + + def _unknown_effects() -> OperationEffects: """Return conservative effects when execution evidence is unavailable.""" return OperationEffects( diff --git a/tests/test_operation_lifecycle.py b/tests/test_operation_lifecycle.py index f6a1998..488af49 100644 --- a/tests/test_operation_lifecycle.py +++ b/tests/test_operation_lifecycle.py @@ -1147,3 +1147,22 @@ async def dispatch(_context: OperationContext) -> ProviderModel: assert terminal.failure.code == ( "stateforge.operation.execution_deadline_exceeded" ) + + +@pytest.mark.asyncio +async def test_author_timeout_is_not_reported_as_execution_deadline() -> None: + record = _runtime_record(cancellable=False) + + async def dispatch(_context: OperationContext) -> ProviderModel: + raise TimeoutError("an operation dependency timed out") + + operation_id = record.snapshot().admission_record.operation_id + await record.execute( + operation_id=operation_id, + execution_deadline=datetime.now(UTC) + timedelta(minutes=1), + execution_context={}, + dispatch=dispatch, + ) + terminal = record.snapshot() + assert terminal.state == operation_lifecycle_pb2.OPERATION_STATE_FAILED + assert terminal.failure.code == "stateforge.operation.provider_error" From 7a693179cccfdbc1f37ab7e3d16378a1fda5e224 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 16:23:10 +0200 Subject: [PATCH 6/6] fix: normalize result serialization failures --- .../_services/operation_lifecycle/runtime.py | 10 ++++++---- tests/test_operation_lifecycle.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/stateforge_provider/_services/operation_lifecycle/runtime.py b/src/stateforge_provider/_services/operation_lifecycle/runtime.py index 7d091cc..4f162e0 100644 --- a/src/stateforge_provider/_services/operation_lifecycle/runtime.py +++ b/src/stateforge_provider/_services/operation_lifecycle/runtime.py @@ -15,6 +15,7 @@ from google.protobuf.timestamp_pb2 import Timestamp from jsonschema import Draft202012Validator from pydantic import JsonValue +from pydantic_core import PydanticSerializationError from stateforge_protocol.sfp.operation_admission.v1 import ( operation_admission_pb2, ) @@ -360,11 +361,11 @@ def _commit_success(self, result: ProviderModel) -> None: if not isinstance(result, self._result_model): self._commit_invalid_result() return - serialized = result.model_dump(mode="json") - if not self._result_validator.is_valid(serialized): - self._commit_invalid_result() - return try: + serialized = result.model_dump(mode="json") + if not self._result_validator.is_valid(serialized): + self._commit_invalid_result() + return result_message = ParseDict(serialized, Struct()) candidate = self._state_snapshot( operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED @@ -374,6 +375,7 @@ def _commit_success(self, result: ProviderModel) -> None: except ( TypeError, ValueError, + PydanticSerializationError, validation.OperationLifecycleValidationError, ): self._commit_invalid_result() diff --git a/tests/test_operation_lifecycle.py b/tests/test_operation_lifecycle.py index 488af49..8542ee3 100644 --- a/tests/test_operation_lifecycle.py +++ b/tests/test_operation_lifecycle.py @@ -20,6 +20,7 @@ from google.protobuf.timestamp_pb2 import Timestamp from grpc_status import rpc_status from jsonschema import Draft202012Validator +from pydantic import field_serializer from stateforge_protocol.sfp.bootstrap.v1 import ( bootstrap_pb2, bootstrap_pb2_grpc, @@ -110,6 +111,12 @@ class _Output(ProviderModel): value: int +class _SerializationFailureOutput(_Output): + @field_serializer("value") + def serialize_value(self, _value: int) -> int: + raise ValueError("result serialization failed") + + class _LargeOutput(ProviderModel): value: str @@ -185,6 +192,9 @@ async def run_operation( if request.mode == "invalid-result": await context.report_progress("Finished", percentage=100) return cast(_Output, ProviderModel()) + if request.mode == "serialization-failure": + await context.report_progress("Finished", percentage=100) + return _SerializationFailureOutput(value=100) await self.release.wait() await context.report_progress("Finished", percentage=100) return _Output(value=100) @@ -623,6 +633,7 @@ async def test_unsupported_cancellation_does_not_signal_execution( ("oversized-progress", "stateforge.operation.invalid_progress"), ("no-100", "stateforge.operation.invalid_progress"), ("invalid-result", "stateforge.operation.invalid_result"), + ("serialization-failure", "stateforge.operation.invalid_result"), ("failure", "example.lifecycle.failed"), ], )