[TASK][MVP] Complete Provider Disconnection and Recovery

.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

.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
 
 

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

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",