diff --git a/.github/actions/scenarios/operation/admission/verify.py b/.github/actions/scenarios/operation/admission/verify.py index 42637a0..856aff6 100644 --- a/.github/actions/scenarios/operation/admission/verify.py +++ b/.github/actions/scenarios/operation/admission/verify.py @@ -31,6 +31,13 @@ operation_admission_pb2_grpc, validation, ) +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 ( OperationContext, @@ -255,10 +262,10 @@ async def _verify_canonical_rejections( return statuses -async def _verify_lost_response_replay( # noqa: PLR0915 +async def _verify_lost_response_replay( # noqa: PLR0912, PLR0915 base_directory: Path, ) -> dict[str, object]: - """Lose one admitted response and prove identical replay resolves.""" + """Reconnect after a lost admission response and resume retained work.""" original = OperationAdmissionService.StartOperation attempts: list[bytes] = [] drop_once = True @@ -301,7 +308,8 @@ async def start_operation( admission = operation_admission_pb2_grpc.OperationAdmissionServiceStub( channel ).StartOperation - request = _request(handshake, manifest, operation_id=str(uuid4())) + operation_id = str(uuid4()) + request = _request(handshake, manifest, operation_id=operation_id) try: await admission(request, timeout=3) except grpc.aio.AioRpcError as error: @@ -310,18 +318,94 @@ async def start_operation( else: raise RuntimeError("first response was not lost") await asyncio.wait_for(provider.started.wait(), timeout=2) + await channel.close() + ( + channel, + admission, + reconnect, + reconnect_manifest, + reconnect_descriptor, + ) = await _connect_provider( + base_directory / "discovery", + provider_task, + ) + if ( + reconnect.provider_instance_id != handshake.provider_instance_id + or reconnect.endpoint_generation != handshake.endpoint_generation + or reconnect.session_id == handshake.session_id + or reconnect_manifest.manifest_revision + != manifest.manifest_revision + or reconnect_descriptor["endpoint_generation"] + != descriptor["endpoint_generation"] + ): + raise RuntimeError("same-process reconnect identity changed") + replay_request = operation_admission_pb2.StartOperationRequest() + replay_request.CopyFrom(request) + replay_request.session_binding.CopyFrom( + operation_admission_pb2.OperationSessionBinding( + expected_provider_instance_id=reconnect.provider_instance_id, + expected_endpoint_generation=reconnect.endpoint_generation, + expected_session_id=reconnect.session_id, + ) + ) before = provider.dispatches - replayed = await admission(request, timeout=3) - validation.validate_admission_exchange(request, replayed) + replayed = await admission(replay_request, timeout=3) + validation.validate_admission_exchange(replay_request, replayed) if replayed.disposition != _RESOLVED_EXISTING: raise RuntimeError("lost-response replay did not resolve existing") if ( len(attempts) != _EXPECTED_LOST_RESPONSE_ATTEMPTS - or attempts[0] != attempts[1] + or operation_admission_pb2.StartOperationRequest.FromString( + attempts[0] + ).invocation + != operation_admission_pb2.StartOperationRequest.FromString( + attempts[1] + ).invocation + or operation_admission_pb2.StartOperationRequest.FromString( + attempts[1] + ) + != replay_request + or request.invocation != replay_request.invocation ): - raise RuntimeError("lost-response replay changed request bytes") + raise RuntimeError("reconnect replay changed invocation content") if provider.dispatches != before: raise RuntimeError("lost-response replay dispatched again") + lifecycle = operation_lifecycle_pb2_grpc.OperationLifecycleServiceStub( + channel + ) + fingerprint = validation.compute_request_fingerprint( + replay_request.invocation + ) + watch_request = operation_lifecycle_pb2.WatchOperationRequest( + session_binding=replay_request.session_binding, + operation_id=operation_id, + expected_request_fingerprint=fingerprint, + after_sequence=0, + ) + watch = lifecycle.WatchOperation(watch_request, timeout=3) + first_snapshot = await watch.read() + lifecycle_validation.validate_watch_exchange( + watch_request, + first_snapshot, + ) + provider.release.set() + terminal = first_snapshot.snapshot + while terminal.state not in { + operation_lifecycle_pb2.OPERATION_STATE_SUCCEEDED, + operation_lifecycle_pb2.OPERATION_STATE_FAILED, + operation_lifecycle_pb2.OPERATION_STATE_CANCELLED, + }: + update = await watch.read() + lifecycle_validation.validate_watch_exchange( + watch_request, + update, + previous_sequence=terminal.sequence, + ) + terminal = update.snapshot + if provider.dispatches != 1: + raise RuntimeError( + "reconnected operation dispatched more than once" + ) return { "request_fingerprint": replayed.record.request_fingerprint, "lost_response_attempts": len(attempts), @@ -331,6 +415,14 @@ async def start_operation( ) ), "lost_response_replay_dispatches": provider.dispatches, + "reconnect_session_changed": reconnect.session_id + != handshake.session_id, + "reconnect_generation_unchanged": reconnect.endpoint_generation + == handshake.endpoint_generation, + "reconnected_lifecycle_terminal_state": ( + operation_lifecycle_pb2.OperationState.Name(terminal.state) + ), + "reconnected_lifecycle_sequence": terminal.sequence, } finally: OperationAdmissionService.StartOperation = original diff --git a/.github/actions/scenarios/trigger/publication/verify.py b/.github/actions/scenarios/trigger/publication/verify.py index c7bdbdc..bf31c9f 100644 --- a/.github/actions/scenarios/trigger/publication/verify.py +++ b/.github/actions/scenarios/trigger/publication/verify.py @@ -35,6 +35,7 @@ ProviderModel, Trigger, TriggerPublicationDisposition, + TriggerPublicationUnavailableError, trigger, ) @@ -264,6 +265,164 @@ async def _prepare_runtime(runtime_directory: Path) -> None: ) +async def _verify_unprovable_trigger_state( + runtime_directory: Path, +) -> dict[str, object]: + """Do not recreate an unacknowledged publication after runtime loss.""" + await _prepare_runtime(runtime_directory) + descriptor, publication_id = await _publish_then_stop(runtime_directory) + return await _assert_runtime_replacement_does_not_replay( + runtime_directory, + descriptor, + publication_id, + ) + + +async def _publish_then_stop( + runtime_directory: Path, +) -> tuple[dict[str, object], str]: + """Stop one runtime after delivery but before the trigger ACK.""" + first_provider = TriggerProvider(runtime_directory) + first_task = asyncio.create_task(first_provider.serve()) + first_channel: grpc.aio.Channel | None = None + first_call: ( + grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ] + | None + ) = None + first_receipt = None + try: + descriptor = await _descriptor( + runtime_directory / "discovery", + first_task, + ) + first_channel = grpc.aio.insecure_channel( + f"unix:{descriptor['transport']['socket_path']}" + ) + await asyncio.wait_for(first_channel.channel_ready(), timeout=3) + handshake, snapshot = await _session( + bootstrap_pb2_grpc.BootstrapServiceStub(first_channel), + manifest_pb2_grpc.ManifestServiceStub(first_channel), + descriptor, + ) + first_call = await _stream( + trigger_publication_pb2_grpc.TriggerPublicationServiceStub( + first_channel + ), + handshake, + snapshot, + ) + first_receipt = await first_provider.publish_trigger( + DEPLOYMENT_COMPLETED, + DeploymentCompleted( + environment="recovery", + revision="unacknowledged", + ), + ) + publication = (await first_call.read()).publication + _require( + publication.publication_id == str(first_receipt.publication_id), + "unacknowledged publication identity", + ) + publication_id = publication.publication_id + first_call.cancel() + await first_channel.close() + first_provider.request_shutdown() + await asyncio.wait_for(first_task, timeout=5) + try: + await first_receipt.outcome() + except TriggerPublicationUnavailableError: + receipt_closed = True + else: + receipt_closed = False + _require(receipt_closed, "lost runtime reports unresolved receipt") + return descriptor, publication_id + finally: + if first_call is not None: + first_call.cancel() + if first_channel is not None: + await first_channel.close() + if not first_task.done(): + first_provider.request_shutdown() + await asyncio.wait_for(first_task, timeout=5) + + +async def _assert_runtime_replacement_does_not_replay( + runtime_directory: Path, + previous_descriptor: dict[str, object], + publication_id: str, +) -> dict[str, object]: + """Require a new runtime to leave unprovable trigger state absent.""" + provider = TriggerProvider(runtime_directory) + task = asyncio.create_task(provider.serve()) + channel: grpc.aio.Channel | None = None + call: ( + grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ] + | None + ) = None + try: + descriptor = await _descriptor( + runtime_directory / "discovery", + task, + ) + _require( + descriptor["provider_instance_id"] + == previous_descriptor["provider_instance_id"], + "same installation identity after runtime replacement", + ) + _require( + int(descriptor["endpoint_generation"]) + > int(previous_descriptor["endpoint_generation"]), + "new endpoint generation after runtime replacement", + ) + channel = grpc.aio.insecure_channel( + f"unix:{descriptor['transport']['socket_path']}" + ) + await asyncio.wait_for(channel.channel_ready(), timeout=3) + handshake, snapshot = await _session( + bootstrap_pb2_grpc.BootstrapServiceStub(channel), + manifest_pb2_grpc.ManifestServiceStub(channel), + descriptor, + ) + call = await _stream( + trigger_publication_pb2_grpc.TriggerPublicationServiceStub( + channel + ), + handshake, + snapshot, + ) + try: + async with asyncio.timeout(0.2): + await call.read() + except TimeoutError: + publication_not_fabricated = True + else: + publication_not_fabricated = False + _require( + publication_not_fabricated, + "unprovable publication is not reconstructed", + ) + return { + "unacknowledged_publication_id": publication_id, + "provider_instance_preserved": True, + "endpoint_generation_increased": True, + "lost_publication_recreated": False, + } + finally: + if call is not None: + call.cancel() + provider.request_shutdown() + if not task.done(): + await asyncio.wait_for(task, timeout=5) + if channel is not None: + await channel.close() + + async def _prove_source_acknowledgement( provider: TriggerProvider, call: grpc.aio.StreamStreamCall[ @@ -592,6 +751,9 @@ async def _verify(runtime_directory: Path) -> dict[str, object]: evidence["socket_removed"] = ( socket_path is not None and not socket_path.exists() ) + evidence.update( + await _verify_unprovable_trigger_state(runtime_directory / "lost") + ) return evidence diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 1c6198f..e501585 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -151,13 +151,17 @@ The runtime suite currently covers: including callback and decorated sources, a bounded receive window, retryable rejection with immutable replay, accepted duplicate disposition, terminal rejection, stale-fence rejection, stream loss, and replay under a - fresh fenced session; + fresh fenced session; runtime replacement with an unacknowledged event also + proves that lost in-memory publication state is reported unresolved and is + not fabricated or assigned a new publication identity; - installed-wheel Operation Admission through an independent Protocol peer, including exact fences, role and requirement rejection, 20 concurrent identical requests converging on one dispatch, typed idempotency conflict, - immutable author context, lost-response replay, deadline and schema - rejection, record expiry, restart, secret-safe evidence, and clean endpoint - removal; + immutable author context, a lost reply resolved after a fresh session on the + same Provider generation, lifecycle observation resumed from retained + state, deadline and schema rejection, record expiry, restart with explicit + unavailable state and no redispatch, secret-safe evidence, and clean + endpoint removal; - end-to-end Operation Admission through the exact installed Agent RPM and exact Provider and Protocol wheels, including Agent-normalized admission, identical concurrent requests, a discarded post-admission response and diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index cf06a66..244c599 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -325,6 +325,7 @@ def test_trigger_publication_keeps_the_focused_protocol_peer() -> None: assert "publish_trigger(" in verifier assert "compute_publication_content_fingerprint" in verifier assert "source_replayed_after_reconnect" in verifier + assert "_verify_unprovable_trigger_state" in verifier for evidence_field in ( "installed_artifacts", "initial_session_id", @@ -336,6 +337,9 @@ def test_trigger_publication_keeps_the_focused_protocol_peer() -> None: "source_acknowledgement_observed", "sentinel_publication_id", "sentinel_outcome", + "unacknowledged_publication_id", + "endpoint_generation_increased", + "lost_publication_recreated", "final_process_state", "descriptor_removed", "socket_removed", @@ -369,6 +373,8 @@ def test_operation_admission_keeps_the_focused_protocol_peer() -> None: assert "@operation(" in verifier assert "OperationContext" in verifier assert "StartOperation" in verifier + assert '"reconnect_session_changed"' in verifier + assert '"reconnected_lifecycle_terminal_state"' in verifier for evidence_field in ( "installed_artifacts", "concurrent_requests",