fix(rc06): unify publication transaction status

yuqei-infra/scripts/apply-rc06-publication-batch.py

@@ -59,12 +59,11 @@ SCENE_LEGAL_REVIEW_BUILDER_PATH = SCRIPT_DIR / "build-scene-source-legal-review.
 
 APPLY_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-apply-request.v3"
 APPLY_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-apply-response.v2"
-APPLY_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-apply-status.v2"
+TRANSACTION_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-transaction-status.v3"
 ROLLBACK_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-rollback-request.v2"
 ROLLBACK_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-rollback-response.v2"
 FINALIZE_REQUEST_SCHEMA_VERSION = "yuqei.rc06-online-finalize-request.v1"
 FINALIZE_RESPONSE_SCHEMA_VERSION = "yuqei.rc06-online-finalize-response.v1"
-FINALIZE_STATUS_SCHEMA_VERSION = "yuqei.rc06-online-finalize-status.v1"
 EXECUTION_EVIDENCE_SCHEMA_VERSION = "yuqei.rc06-online-apply-evidence.v2"
 FINAL_EVIDENCE_SCHEMA_VERSION = "yuqei.rc06-finalized-apply-evidence.v1"
 LEDGER_SCHEMA_VERSION = "yuqei.rc06-local-apply-ledger.v2"
@@ -334,7 +333,11 @@ class ApplyLedger:
                 receipt = json.loads(row[2])
                 self.connection.execute("COMMIT")
                 return "replay", receipt
-            if row is not None and row[1] in {"dispatching", "verifying"}:
+            if row is not None and row[1] in {
+                "dispatching",
+                "verifying",
+                "irreversible_failure_no_go",
+            }:
                 timestamp = now.astimezone(UTC).isoformat(timespec="seconds")
                 self._event(
                     key,
@@ -346,11 +349,17 @@ class ApplyLedger:
                 return "recover", None
             if row is not None:
                 raise ApplyPublicationError("an apply with this idempotency key needs reconciliation")
-            if not initial_dispatch_authorized:
-                raise ApplyPublicationError(
-                    "apply challenge is not currently valid for an initial dispatch"
-                )
             timestamp = now.astimezone(UTC).isoformat(timespec="seconds")
+            if not initial_dispatch_authorized:
+                self.connection.execute(
+                    "INSERT INTO rc06_apply_operations "
+                    "(idempotency_key, request_sha256, state, updated_at) "
+                    "VALUES (?, ?, 'probing_remote', ?)",
+                    (key, request_sha256, timestamp),
+                )
+                self._event(key, "probing_remote", request_sha256, timestamp)
+                self.connection.execute("COMMIT")
+                return "probe", None
             self.connection.execute(
                 "INSERT INTO rc06_apply_operations "
                 "(idempotency_key, request_sha256, state, updated_at) VALUES (?, ?, 'dispatching', ?) "
@@ -503,27 +512,14 @@ def _validate_apply_response(
     return row
 
 
-def _validate_online_status(
+_TRANSACTION_STATUS_FIELDS = (
-    status: Any,
-    *,
-    key: str,
-    transaction_id: str,
-    legal_review_binding_sha256: str,
-    environment_id: str,
-    tenant_id: str,
-    online_transaction_receipt: Mapping[str, Any],
-    runtime_snapshot_receipt: Mapping[str, Any],
-    expected_publication_results: list[dict[str, Any]],
-    expected_runtime_results: list[dict[str, Any]],
-) -> dict[str, Any]:
-    row = _require_exact(
-        status,
-        "online apply status",
-        (
     "schema_version",
     "idempotency_key",
+    "request_sha256",
     "transaction_id",
     "state",
+    "state_version",
+    "no_go",
     "legal_source_review_binding_sha256",
     "environment_id",
     "tenant_id",
@@ -533,51 +529,251 @@ def _validate_online_status(
     "publication_item_count",
     "runtime_snapshot_dependencies",
     "runtime_snapshot_dependency_count",
-            "online_transaction_receipt_sha256",
-            "runtime_snapshot_receipt_sha256",
     "participant_evidence",
-        ),
+    "online_transaction_receipt",
-    )
+    "online_transaction_receipt_sha256",
+    "runtime_snapshot_receipt",
+    "runtime_snapshot_receipt_sha256",
+    "apply_receipt",
+    "apply_receipt_sha256",
+    "finalized_at",
+    "rollback_response",
+    "rollback_response_sha256",
+    "failure_code",
+    "failure_at",
+    "failure_evidence_sha256",
+    "status_updated_at",
+)
+
+_TRANSACTION_STATE_VERSION = {
+    "prepared": 1,
+    "rolled_back": 2,
+    # This is the post-Contract-commit, no-rollback boundary. Replaying the
+    # exact request may still recover read-back/KMS evidence and advance to 3.
+    "irreversible_failure": 2,
+    "committed_and_verified": 3,
+    "finalized": 4,
+}
+
+
+def _validate_transaction_status(
+    status: Any,
+    *,
+    manifest: Mapping[str, Any],
+    manifest_raw_sha256: str,
+    approval: Mapping[str, Any],
+    approval_raw_sha256: str,
+    approval_keyring: Mapping[str, Any],
+    approval_root_public_key: Ed25519PublicKey,
+    apply_service_public_key: Ed25519PublicKey,
+    online_service_public_key: Any,
+    request_sha256: str,
+    key: str,
+    transaction_id: str,
+    legal_review_binding_sha256: str,
+    legal_review_expires_at: str,
+    environment_id: str,
+    tenant_id: str,
+    minimum_state_version: int = 1,
+    expected_apply_receipt: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Validate the one exact forward GET status contract for every client."""
+
+    row = _require_exact(status, "online transaction status", _TRANSACTION_STATUS_FIELDS)
+    state = row["state"]
+    expected_state_version = _TRANSACTION_STATE_VERSION.get(state)
     if (
-        row["schema_version"] != APPLY_STATUS_SCHEMA_VERSION
+        row["schema_version"] != TRANSACTION_STATUS_SCHEMA_VERSION
         or row["idempotency_key"] != key
+        or row["request_sha256"] != request_sha256
         or row["transaction_id"] != transaction_id
-        or row["state"] != "committed_and_verified"
+        or expected_state_version is None
+        or not isinstance(row["state_version"], int)
+        or isinstance(row["state_version"], bool)
+        or row["state_version"] != expected_state_version
+        or row["state_version"] < minimum_state_version
+        or row["no_go"] is not (state != "finalized")
         or row["legal_source_review_binding_sha256"] != legal_review_binding_sha256
         or row["environment_id"] != environment_id
         or row["tenant_id"] != tenant_id
         or row["transaction_mode"] != batch.TRANSACTION_MODE
         or row["cross_database_atomicity"] is not False
-        or row["publication_item_results"] != expected_publication_results
+    ):
+        raise ApplyPublicationError("online transaction status identity or state drift")
+    batch._require_sha(row["request_sha256"], "online status request SHA-256")
+    status_updated_at = batch._parse_time(
+        row["status_updated_at"], "online status_updated_at"
+    )
+
+    commit_empty = (
+        row["publication_item_results"] == []
+        and row["publication_item_count"] == 0
+        and row["runtime_snapshot_dependencies"] == []
+        and row["runtime_snapshot_dependency_count"] == 0
+        and row["participant_evidence"] == []
+        and row["online_transaction_receipt"] is None
+        and row["online_transaction_receipt_sha256"] is None
+        and row["runtime_snapshot_receipt"] is None
+        and row["runtime_snapshot_receipt_sha256"] is None
+    )
+    final_empty = (
+        row["apply_receipt"] is None
+        and row["apply_receipt_sha256"] is None
+        and row["finalized_at"] is None
+    )
+    rollback_empty = (
+        row["rollback_response"] is None
+        and row["rollback_response_sha256"] is None
+    )
+    failure_empty = (
+        row["failure_code"] is None
+        and row["failure_at"] is None
+        and row["failure_evidence_sha256"] is None
+    )
+
+    if state in {"prepared", "irreversible_failure", "rolled_back"} and not commit_empty:
+        raise ApplyPublicationError("non-committed online status contains commit evidence")
+    if state != "finalized" and not final_empty:
+        raise ApplyPublicationError("non-finalized online status contains finalization evidence")
+    if state != "rolled_back" and not rollback_empty:
+        raise ApplyPublicationError("non-rollback online status contains rollback evidence")
+    if state != "irreversible_failure" and not failure_empty:
+        raise ApplyPublicationError("non-failure online status contains failure evidence")
+
+    if state == "prepared":
+        if not rollback_empty or not failure_empty:
+            raise ApplyPublicationError("prepared online status evidence is invalid")
+        return row
+
+    if state == "rolled_back":
+        rollback = _validate_rollback_response(
+            row["rollback_response"],
+            manifest=manifest,
+            key=key,
+            transaction_id=transaction_id,
+        )
+        if row["rollback_response_sha256"] != batch._sha256_json(rollback):
+            raise ApplyPublicationError("online rollback status hash drift")
+        return row
+
+    if state == "irreversible_failure":
+        _require_text(row["failure_code"], "online failure_code", maximum=128)
+        failure_at = batch._parse_time(row["failure_at"], "online failure_at")
+        batch._require_sha(
+            row["failure_evidence_sha256"], "online failure evidence SHA-256"
+        )
+        if failure_at > status_updated_at:
+            raise ApplyPublicationError("online failure postdates its status update")
+        return row
+
+    expected_publication_results = batch._expected_apply_item_results(manifest)
+    expected_runtime_results = expected_runtime_snapshot_results(manifest)
+    online_transaction_receipt = row["online_transaction_receipt"]
+    runtime_snapshot_receipt = row["runtime_snapshot_receipt"]
+    if not isinstance(online_transaction_receipt, dict) or not isinstance(
+        runtime_snapshot_receipt, dict
+    ):
+        raise ApplyPublicationError("committed online status omits signed receipt evidence")
+    online_receipt = batch.validate_online_transaction_receipt(
+        manifest,
+        manifest_raw_sha256,
+        approval_raw_sha256,
+        approval,
+        apply_service_public_key,
+        online_service_public_key,
+        online_transaction_receipt,
+        legal_review_binding_sha256=legal_review_binding_sha256,
+        legal_review_expires_at=legal_review_expires_at,
+        environment_id=environment_id,
+        tenant_id=tenant_id,
+    )
+    batch.validate_runtime_snapshot_receipt(
+        manifest,
+        manifest_raw_sha256,
+        approval_raw_sha256,
+        approval,
+        apply_service_public_key,
+        online_service_public_key,
+        online_receipt,
+        runtime_snapshot_receipt,
+        legal_review_binding_sha256=legal_review_binding_sha256,
+        legal_review_expires_at=legal_review_expires_at,
+        environment_id=environment_id,
+        tenant_id=tenant_id,
+    )
+    if (
+        row["publication_item_results"] != expected_publication_results
+        or row["publication_item_results"]
+        != online_transaction_receipt["payload"]["item_results"]
         or row["publication_item_count"] != 74
         or row["runtime_snapshot_dependencies"] != expected_runtime_results
+        or row["runtime_snapshot_dependencies"]
+        != runtime_snapshot_receipt["payload"]["runtime_snapshot_dependencies"]
         or row["runtime_snapshot_dependency_count"] != 14
+        or row["participant_evidence"]
+        != online_transaction_receipt["payload"]["participant_evidence"]
         or row["online_transaction_receipt_sha256"]
         != batch._sha256_json(online_transaction_receipt)
         or row["runtime_snapshot_receipt_sha256"]
         != batch._sha256_json(runtime_snapshot_receipt)
-        or row["participant_evidence"]
+        or not rollback_empty
-        != online_transaction_receipt["payload"]["participant_evidence"]
+        or not failure_empty
     ):
-        raise ApplyPublicationError("online read-after-write hash verification failed")
+        raise ApplyPublicationError("online read-after-write receipt verification failed")
+    latest_commit_evidence_at = max(
+        batch._parse_time(
+            online_transaction_receipt["payload"]["committed_at"],
+            "online receipt committed_at",
+        ),
+        batch._parse_time(
+            runtime_snapshot_receipt["payload"]["captured_at"],
+            "runtime snapshot captured_at",
+        ),
+    )
+    if status_updated_at < latest_commit_evidence_at:
+        raise ApplyPublicationError("online status predates its signed commit evidence")
+    if state == "committed_and_verified":
+        if not final_empty:
+            raise ApplyPublicationError("committed status contains finalization evidence")
         return row
 
-
+    apply_receipt = row["apply_receipt"]
-def _remote_status_state(
+    if not isinstance(apply_receipt, dict):
-    status: Any,
+        raise ApplyPublicationError("finalized status omits the signed apply receipt")
-    *,
+    batch.validate_apply_receipt(
-    key: str,
+        manifest,
-    transaction_id: str,
+        manifest_raw_sha256,
-) -> str | None:
+        approval,
+        approval_keyring,
+        approval_root_public_key,
+        apply_service_public_key,
+        online_service_public_key,
+        approval_raw_sha256,
+        apply_receipt,
+        legal_review_binding_sha256=legal_review_binding_sha256,
+        legal_review_expires_at=legal_review_expires_at,
+        environment_id=environment_id,
+        tenant_id=tenant_id,
+    )
     if (
-        not isinstance(status, dict)
+        apply_receipt["payload"]["online_transaction_receipt"]
-        or status.get("schema_version") != APPLY_STATUS_SCHEMA_VERSION
+        != online_transaction_receipt
-        or status.get("idempotency_key") != key
+        or apply_receipt["payload"]["runtime_snapshot_receipt"]
-        or status.get("transaction_id") != transaction_id
+        != runtime_snapshot_receipt
-        or not isinstance(status.get("state"), str)
+        or row["apply_receipt_sha256"] != batch._sha256_json(apply_receipt)
+        or (
+            expected_apply_receipt is not None
+            and apply_receipt != expected_apply_receipt
+        )
     ):
-        return None
+        raise ApplyPublicationError("finalized status receipt binding drift")
-    return status["state"]
+    finalized_at = batch._parse_time(row["finalized_at"], "online finalized_at")
+    countersigned_at = batch._parse_time(
+        apply_receipt["payload"]["countersigned_at"], "apply countersigned_at"
+    )
+    if finalized_at < countersigned_at or status_updated_at < finalized_at:
+        raise ApplyPublicationError("online finalized status timestamp drift")
+    return row
 
 
 def _validate_compensation_hashes(
@@ -619,37 +815,13 @@ def _validate_compensation_hashes(
     raise ApplyPublicationError(f"{label} action is unsupported")
 
 
-def _compensate(
+def _validate_rollback_response(
-    client: JsonHttpClient,
+    response: Any,
     *,
     manifest: Mapping[str, Any],
     key: str,
     transaction_id: str,
-    headers: Mapping[str, str],
-    manifest_raw_sha256: str,
-    approval_raw_sha256: str,
-    legal_review_binding_sha256: str,
-    environment_id: str,
-    tenant_id: str,
 ) -> dict[str, Any]:
-    request = {
-        "schema_version": ROLLBACK_REQUEST_SCHEMA_VERSION,
-        "idempotency_key": key,
-        "transaction_id": transaction_id,
-        "batch_manifest_raw_sha256": manifest_raw_sha256,
-        "approval_receipt_sha256": approval_raw_sha256,
-        "legal_source_review_binding_sha256": legal_review_binding_sha256,
-        "environment_id": environment_id,
-        "tenant_id": tenant_id,
-        "transaction_mode": batch.TRANSACTION_MODE,
-        "cross_database_atomicity": False,
-    }
-    response = client.request(
-        "POST",
-        f"/internal/rc06/publication-transactions/{transaction_id}/rollback",
-        payload=request,
-        headers=headers,
-    )
     row = _require_exact(
         response,
         "online rollback response",
@@ -886,6 +1058,45 @@ def _compensate(
     return row
 
 
+def _compensate(
+    client: JsonHttpClient,
+    *,
+    manifest: Mapping[str, Any],
+    key: str,
+    transaction_id: str,
+    headers: Mapping[str, str],
+    manifest_raw_sha256: str,
+    approval_raw_sha256: str,
+    legal_review_binding_sha256: str,
+    environment_id: str,
+    tenant_id: str,
+) -> dict[str, Any]:
+    request = {
+        "schema_version": ROLLBACK_REQUEST_SCHEMA_VERSION,
+        "idempotency_key": key,
+        "transaction_id": transaction_id,
+        "batch_manifest_raw_sha256": manifest_raw_sha256,
+        "approval_receipt_sha256": approval_raw_sha256,
+        "legal_source_review_binding_sha256": legal_review_binding_sha256,
+        "environment_id": environment_id,
+        "tenant_id": tenant_id,
+        "transaction_mode": batch.TRANSACTION_MODE,
+        "cross_database_atomicity": False,
+    }
+    response = client.request(
+        "POST",
+        f"/internal/rc06/publication-transactions/{transaction_id}/rollback",
+        payload=request,
+        headers=headers,
+    )
+    return _validate_rollback_response(
+        response,
+        manifest=manifest,
+        key=key,
+        transaction_id=transaction_id,
+    )
+
+
 def _validate_execution_evidence(
     evidence: Any,
     *,
@@ -968,6 +1179,69 @@ def _validate_execution_evidence(
     return row
 
 
+def _execution_from_transaction_status(
+    status: Mapping[str, Any],
+    *,
+    request_sha256: str,
+    legal_review_expires_at: str,
+) -> dict[str, Any]:
+    if status["state"] not in {"committed_and_verified", "finalized"}:
+        raise ApplyPublicationError("online status does not contain committed execution evidence")
+    return {
+        "schema_version": EXECUTION_EVIDENCE_SCHEMA_VERSION,
+        "idempotency_key": status["idempotency_key"],
+        "transaction_id": status["transaction_id"],
+        "request_sha256": request_sha256,
+        "legal_source_review_binding_sha256": status[
+            "legal_source_review_binding_sha256"
+        ],
+        "legal_source_review_expires_at": legal_review_expires_at,
+        "environment_id": status["environment_id"],
+        "tenant_id": status["tenant_id"],
+        "transaction_mode": status["transaction_mode"],
+        "cross_database_atomicity": status["cross_database_atomicity"],
+        "publication_item_count": status["publication_item_count"],
+        "runtime_snapshot_dependency_count": status[
+            "runtime_snapshot_dependency_count"
+        ],
+        "online_transaction_receipt": status["online_transaction_receipt"],
+        "runtime_snapshot_receipt": status["runtime_snapshot_receipt"],
+    }
+
+
+def _finalize_response_from_transaction_status(
+    status: Mapping[str, Any],
+) -> dict[str, Any]:
+    if status["state"] != "finalized":
+        raise ApplyPublicationError("online status is not finalized")
+    return {
+        "schema_version": FINALIZE_RESPONSE_SCHEMA_VERSION,
+        "idempotency_key": status["idempotency_key"],
+        "transaction_id": status["transaction_id"],
+        "state": "finalized",
+        "no_go": False,
+        "apply_receipt_sha256": status["apply_receipt_sha256"],
+        "finalized_at": status["finalized_at"],
+    }
+
+
+def _final_evidence_from_transaction_status(
+    execution: Mapping[str, Any],
+    status: Mapping[str, Any],
+) -> dict[str, Any]:
+    return {
+        "schema_version": FINAL_EVIDENCE_SCHEMA_VERSION,
+        "transaction_mode": batch.TRANSACTION_MODE,
+        "cross_database_atomicity": False,
+        "publication_item_count": 74,
+        "runtime_snapshot_dependency_count": 14,
+        "online_execution": execution,
+        "apply_receipt": status["apply_receipt"],
+        "online_finalize_response": _finalize_response_from_transaction_status(status),
+        "online_finalize_status": status,
+    }
+
+
 def execute_online_publication_batch(
     *,
     manifest: Mapping[str, Any],
@@ -1129,7 +1403,69 @@ def execute_online_publication_batch(
         legal_review_binding_sha256,
         tenant_id,
     )
-    if reservation == "recover" and not challenge_is_current:
+
+    def validate_remote_status(
+        value: Any,
+        *,
+        minimum_state_version: int = 1,
+        expected_apply_receipt: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        return _validate_transaction_status(
+            value,
+            manifest=manifest,
+            manifest_raw_sha256=manifest_raw_sha256,
+            approval=approval,
+            approval_raw_sha256=approval_raw_sha256,
+            approval_keyring=approval_keyring,
+            approval_root_public_key=approval_root_public_key,
+            apply_service_public_key=apply_service_public_key,
+            online_service_public_key=online_service_public_key,
+            request_sha256=request_sha256,
+            key=key,
+            transaction_id=transaction_id,
+            legal_review_binding_sha256=legal_review_binding_sha256,
+            legal_review_expires_at=legal_review_expires_at,
+            environment_id=environment_id,
+            tenant_id=tenant_id,
+            minimum_state_version=minimum_state_version,
+            expected_apply_receipt=expected_apply_receipt,
+        )
+
+    def adopt_committed_status(status_row: Mapping[str, Any]) -> dict[str, Any]:
+        execution = _execution_from_transaction_status(
+            status_row,
+            request_sha256=request_sha256,
+            legal_review_expires_at=legal_review_expires_at,
+        )
+        execution = _validate_execution_evidence(
+            execution,
+            manifest=manifest,
+            manifest_raw_sha256=manifest_raw_sha256,
+            approval=approval,
+            approval_raw_sha256=approval_raw_sha256,
+            apply_service_public_key=apply_service_public_key,
+            online_service_public_key=online_service_public_key,
+            legal_review_binding_sha256=legal_review_binding_sha256,
+            legal_review_expires_at=legal_review_expires_at,
+            environment_id=environment_id,
+            tenant_id=tenant_id,
+            idempotency_key=key,
+            transaction_id=transaction_id,
+        )
+        if status_row["state"] == "finalized":
+            final_evidence = _final_evidence_from_transaction_status(execution, status_row)
+            ledger.state(key, "succeeded", status_row, now, receipt=final_evidence)
+        else:
+            ledger.state(
+                key,
+                "online_committed_pending_countersign",
+                status_row,
+                now,
+                receipt=execution,
+            )
+        return execution
+
+    if reservation in {"probe", "recover"} and not challenge_is_current:
         try:
             reconciliation = client.request(
                 "GET",
@@ -1137,11 +1473,7 @@ def execute_online_publication_batch(
                 payload=None,
                 headers=headers,
             )
-            remote_state = _remote_status_state(
+            reconciliation = validate_remote_status(reconciliation)
-                reconciliation,
-                key=key,
-                transaction_id=transaction_id,
-            )
         except Exception as exc:
             ledger.state(
                 key,
@@ -1151,17 +1483,15 @@ def execute_online_publication_batch(
                 error="expired challenge recovery could not establish remote commit status",
             )
             raise ApplyPublicationError(
-                "expired challenge cannot be retransmitted until an existing terminal commit is proven"
+                "apply challenge is not currently valid for an initial dispatch and no existing transaction was proven"
             ) from exc
+        remote_state = reconciliation["state"]
+        if remote_state in {"committed_and_verified", "finalized"}:
+            return adopt_committed_status(reconciliation)
         if remote_state == "rolled_back":
             ledger.state(key, "rolled_back", reconciliation, now)
             raise ApplyPublicationError("online apply was already rolled back")
-        if remote_state not in {
+        if remote_state == "prepared":
-            "contract_committed",
-            "evidence_pending",
-            "committed_and_verified",
-            "online_committed_pending_countersign",
-        }:
             ledger.state(
                 key,
                 "expired_challenge_recovery_no_go",
@@ -1172,6 +1502,10 @@ def execute_online_publication_batch(
             raise ApplyPublicationError(
                 "expired challenge cannot authorize a new or non-terminal retransmission"
             )
+        # irreversible_failure proves that this exact request already crossed
+        # the no-rollback boundary. Replaying it may only recover its evidence.
+        if remote_state != "irreversible_failure":
+            raise ApplyPublicationError("online transaction state is unsupported")
     dispatch_started = False
     response: Any = None
     try:
@@ -1231,42 +1565,15 @@ def execute_online_publication_batch(
             payload=None,
             headers=headers,
         )
-        status = _validate_online_status(
+        status = validate_remote_status(status, minimum_state_version=3)
-            status,
+        if status["state"] not in {"committed_and_verified", "finalized"}:
-            key=key,
+            raise ApplyPublicationError("online apply did not reach a committed status")
-            transaction_id=transaction_id,
+        if (
-            legal_review_binding_sha256=legal_review_binding_sha256,
+            status["online_transaction_receipt"] != online_receipt
-            environment_id=environment_id,
+            or status["runtime_snapshot_receipt"] != runtime_receipt
-            tenant_id=tenant_id,
+        ):
-            online_transaction_receipt=online_receipt,
+            raise ApplyPublicationError("online apply response and durable status receipt drift")
-            runtime_snapshot_receipt=runtime_receipt,
+        return adopt_committed_status(status)
-            expected_publication_results=batch._expected_apply_item_results(manifest),
-            expected_runtime_results=expected_runtime_snapshot_results(manifest),
-        )
-        evidence = {
-            "schema_version": EXECUTION_EVIDENCE_SCHEMA_VERSION,
-            "idempotency_key": key,
-            "transaction_id": transaction_id,
-            "request_sha256": request_sha256,
-            "legal_source_review_binding_sha256": legal_review_binding_sha256,
-            "legal_source_review_expires_at": legal_review_expires_at,
-            "environment_id": environment_id,
-            "tenant_id": tenant_id,
-            "transaction_mode": batch.TRANSACTION_MODE,
-            "cross_database_atomicity": False,
-            "publication_item_count": 74,
-            "runtime_snapshot_dependency_count": 14,
-            "online_transaction_receipt": online_receipt,
-            "runtime_snapshot_receipt": runtime_receipt,
-        }
-        ledger.state(
-            key,
-            "online_committed_pending_countersign",
-            status,
-            now,
-            receipt=evidence,
-        )
-        return evidence
     except Exception as exc:
         if not isinstance(exc, ApplyPublicationError):
             failure = ApplyPublicationError("unexpected online apply failure")
@@ -1281,11 +1588,7 @@ def execute_online_publication_batch(
                     payload=None,
                     headers=headers,
                 )
-                remote_state = _remote_status_state(
+                reconciliation = validate_remote_status(reconciliation)
-                    reconciliation,
-                    key=key,
-                    transaction_id=transaction_id,
-                )
             except Exception as reconciliation_exc:
                 ledger.state(
                     key,
@@ -1300,26 +1603,24 @@ def execute_online_publication_batch(
                 raise ApplyPublicationError(
                     "online apply evidence failed and commit status is unknown; no rollback was attempted"
                 ) from reconciliation_exc
-            if remote_state in {
+            remote_state = reconciliation["state"]
-                "contract_committed",
+            if remote_state in {"committed_and_verified", "finalized"}:
-                "evidence_pending",
+                return adopt_committed_status(reconciliation)
-                "committed_and_verified",
+            if remote_state == "irreversible_failure":
-                "online_committed_pending_countersign",
-            }:
                 ledger.state(
                     key,
-                    "committed_evidence_invalid_no_go",
+                    "irreversible_failure_no_go",
                     reconciliation,
                     now,
                     error=str(failure),
                 )
                 raise ApplyPublicationError(
-                    "online saga is terminally committed but local evidence verification failed; rollback is forbidden"
+                    "online saga crossed the irreversible boundary but evidence recovery is incomplete; rollback is forbidden"
                 ) from failure
             if remote_state == "rolled_back":
                 ledger.state(key, "rolled_back", reconciliation, now, error=str(failure))
                 raise failure
-            if remote_state not in {"prepared", "ai_committed", "failed_before_commit"}:
+            if remote_state != "prepared":
                 ledger.state(
                     key,
                     "commit_status_unknown",
@@ -1405,17 +1706,6 @@ def finalize_publication_batch(
         idempotency_key=key,
         transaction_id=transaction_id,
     )
-    state, stored = ledger.load(key, execution["request_sha256"])
-    if state not in {
-        "online_committed_pending_countersign",
-        "finalize_failed_no_go",
-        "succeeded",
-    }:
-        raise ApplyPublicationError("online saga is not eligible for offline finalization")
-    if state == "succeeded" and stored.get("schema_version") == FINAL_EVIDENCE_SCHEMA_VERSION:
-        return stored
-    if stored != execution:
-        raise ApplyPublicationError("supplied online evidence differs from the durable ledger")
     batch.validate_apply_receipt(
         manifest,
         manifest_raw_sha256,
@@ -1439,6 +1729,67 @@ def finalize_publication_batch(
     ):
         raise ApplyPublicationError("offline countersign is not bound to this receipt pair")
     apply_receipt_sha256 = batch._sha256_json(apply_receipt)
+    state, stored = ledger.load(key, execution["request_sha256"])
+    if state not in {
+        "online_committed_pending_countersign",
+        "finalize_failed_no_go",
+        "succeeded",
+    }:
+        raise ApplyPublicationError("online saga is not eligible for offline finalization")
+
+    def validate_remote_status(value: Any, *, minimum_state_version: int) -> dict[str, Any]:
+        return _validate_transaction_status(
+            value,
+            manifest=manifest,
+            manifest_raw_sha256=manifest_raw_sha256,
+            approval=approval,
+            approval_raw_sha256=approval_raw_sha256,
+            approval_keyring=approval_keyring,
+            approval_root_public_key=approval_root_public_key,
+            apply_service_public_key=apply_service_public_key,
+            online_service_public_key=online_service_public_key,
+            request_sha256=execution["request_sha256"],
+            key=key,
+            transaction_id=transaction_id,
+            legal_review_binding_sha256=legal_review_binding_sha256,
+            legal_review_expires_at=legal_review_expires_at,
+            environment_id=environment_id,
+            tenant_id=tenant_id,
+            minimum_state_version=minimum_state_version,
+            expected_apply_receipt=apply_receipt,
+        )
+
+    if state == "succeeded":
+        stored_final = _require_exact(
+            stored,
+            "stored finalized evidence",
+            (
+                "schema_version",
+                "transaction_mode",
+                "cross_database_atomicity",
+                "publication_item_count",
+                "runtime_snapshot_dependency_count",
+                "online_execution",
+                "apply_receipt",
+                "online_finalize_response",
+                "online_finalize_status",
+            ),
+        )
+        if (
+            stored_final["schema_version"] != FINAL_EVIDENCE_SCHEMA_VERSION
+            or stored_final["online_execution"] != execution
+            or stored_final["apply_receipt"] != apply_receipt
+        ):
+            raise ApplyPublicationError("stored finalized evidence differs from the exact replay")
+        stored_status = validate_remote_status(
+            stored_final["online_finalize_status"], minimum_state_version=4
+        )
+        if stored_final != _final_evidence_from_transaction_status(execution, stored_status):
+            raise ApplyPublicationError("stored finalized evidence is internally inconsistent")
+        return stored_final
+    if stored != execution:
+        raise ApplyPublicationError("supplied online evidence differs from the durable ledger")
+
     finalize_request = {
         "schema_version": FINALIZE_REQUEST_SCHEMA_VERSION,
         "idempotency_key": key,
@@ -1464,6 +1815,7 @@ def finalize_publication_batch(
         legal_review_binding_sha256,
         tenant_id,
     )
+    response_row: dict[str, Any] | None = None
     try:
         response = client.request(
             "POST",
@@ -1509,69 +1861,62 @@ def finalize_publication_batch(
             payload=None,
             headers=headers,
         )
-        status_row = _require_exact(
+        status_row = validate_remote_status(status, minimum_state_version=4)
-            status,
+        if status_row["state"] != "finalized":
-            "online finalize status",
+            raise ApplyPublicationError("online finalize did not reach finalized status")
-            (
+        if response_row != _finalize_response_from_transaction_status(status_row):
-                "schema_version",
+            raise ApplyPublicationError("online finalize response and durable status drift")
-                "idempotency_key",
-                "transaction_id",
-                "state",
-                "no_go",
-                "transaction_mode",
-                "cross_database_atomicity",
-                "publication_item_count",
-                "runtime_snapshot_dependency_count",
-                "online_transaction_receipt_sha256",
-                "runtime_snapshot_receipt_sha256",
-                "apply_receipt_sha256",
-                "finalized_at",
-            ),
-        )
-        if (
-            status_row["schema_version"] != FINALIZE_STATUS_SCHEMA_VERSION
-            or status_row["idempotency_key"] != key
-            or status_row["transaction_id"] != transaction_id
-            or status_row["state"] != "finalized"
-            or status_row["no_go"] is not False
-            or status_row["transaction_mode"] != batch.TRANSACTION_MODE
-            or status_row["cross_database_atomicity"] is not False
-            or status_row["publication_item_count"] != 74
-            or status_row["runtime_snapshot_dependency_count"] != 14
-            or status_row["online_transaction_receipt_sha256"]
-            != finalize_request["online_transaction_receipt_sha256"]
-            or status_row["runtime_snapshot_receipt_sha256"]
-            != finalize_request["runtime_snapshot_receipt_sha256"]
-            or status_row["apply_receipt_sha256"] != apply_receipt_sha256
-            or status_row["finalized_at"] != response_row["finalized_at"]
-        ):
-            raise ApplyPublicationError("online finalized status verification failed")
     except Exception as exc:
         failure = (
             exc
             if isinstance(exc, ApplyPublicationError)
             else ApplyPublicationError("online finalize request failed")
         )
+        try:
+            reconciliation = client.request(
+                "GET",
+                f"/internal/rc06/publication-transactions/{transaction_id}/status",
+                payload=None,
+                headers=headers,
+            )
+            reconciliation = validate_remote_status(
+                reconciliation, minimum_state_version=3
+            )
+            if reconciliation["state"] == "finalized":
+                final_evidence = _final_evidence_from_transaction_status(
+                    execution, reconciliation
+                )
+                ledger.state(
+                    key, "succeeded", reconciliation, now, receipt=final_evidence
+                )
+                return final_evidence
+            if reconciliation["state"] != "committed_and_verified":
+                raise ApplyPublicationError(
+                    "online finalization reconciliation regressed from committed state"
+                )
+        except Exception as reconciliation_exc:
             ledger.state(
                 key,
                 "finalize_failed_no_go",
-            {"error": str(failure)},
+                {
+                    "finalize_error": str(failure),
+                    "reconciliation_error": str(reconciliation_exc),
+                },
+                now,
+                receipt=execution,
+                error=f"{failure}; finalization status is untrusted",
+            )
+            raise failure from reconciliation_exc
+        ledger.state(
+            key,
+            "finalize_failed_no_go",
+            {"finalize_error": str(failure), "status": reconciliation},
             now,
             receipt=execution,
             error=str(failure),
         )
         raise failure
-    final_evidence = {
+    final_evidence = _final_evidence_from_transaction_status(execution, status_row)
-        "schema_version": FINAL_EVIDENCE_SCHEMA_VERSION,
-        "transaction_mode": batch.TRANSACTION_MODE,
-        "cross_database_atomicity": False,
-        "publication_item_count": 74,
-        "runtime_snapshot_dependency_count": 14,
-        "online_execution": execution,
-        "apply_receipt": apply_receipt,
-        "online_finalize_response": response_row,
-        "online_finalize_status": status_row,
-    }
     ledger.state(key, "succeeded", final_evidence, now, receipt=final_evidence)
     return final_evidence
 

yuqei-infra/tests/test_rc06_publication_apply.py

@@ -495,9 +495,12 @@ class FakePublicationService:
         self.rollback_calls = 0
         self.finalize_calls = 0
         self.request: dict | None = None
+        self.request_sha256: str | None = None
         self.response: dict | None = None
         self.state: str | None = None
         self.finalize_request: dict | None = None
+        self.finalized_at: str | None = None
+        self.rollback_response: dict | None = None
         # The fake online service is deliberately incapable of minting the
         # offline authority receipt.
         assert not hasattr(self, "apply_private")
@@ -632,8 +635,11 @@ class FakePublicationService:
     def dispatch(self, request: dict, request_sha256: str) -> tuple[int, dict]:
         self.apply_calls += 1
         if self.response is not None:
+            if self.state == "irreversible_failure":
+                self.state = "committed_and_verified"
             return 200, self.response
         self.request = request
+        self.request_sha256 = request_sha256
         if self.fail_before_commit:
             self.state = "prepared"
             return 409, {"error": "injected failure before any durable commit"}
@@ -703,10 +709,11 @@ class FakePublicationService:
             "unchanged_item_count": 0,
         }
         self.state = "rolled_back"
-        return 200, {
+        self.rollback_response = {
             **projection,
             "rollback_evidence_sha256": apply.batch._sha256_json(projection),
         }
+        return 200, self.rollback_response
 
     def finalize(self, transaction_id: str, request: dict) -> tuple[int, dict]:
         self.finalize_calls += 1
@@ -724,6 +731,7 @@ class FakePublicationService:
             )
             + timedelta(seconds=1)
         ).isoformat(timespec="seconds")
+        self.finalized_at = finalized_at
         return 200, {
             "schema_version": apply.FINALIZE_RESPONSE_SCHEMA_VERSION,
             "idempotency_key": request["idempotency_key"],
@@ -735,51 +743,19 @@ class FakePublicationService:
         }
 
     def status(self, transaction_id: str) -> tuple[int, dict]:
-        assert self.request is not None
+        if self.request is None:
-        if self.state in {"prepared", "rolled_back"}:
+            return 404, {"error": "transaction not found"}
-            return 200, {
+        assert self.request_sha256 is not None
-                "schema_version": apply.APPLY_STATUS_SCHEMA_VERSION,
+        state = self.state
+        assert state in apply._TRANSACTION_STATE_VERSION
+        result = {
+            "schema_version": apply.TRANSACTION_STATUS_SCHEMA_VERSION,
             "idempotency_key": self.request["idempotency_key"],
+            "request_sha256": self.request_sha256,
             "transaction_id": transaction_id,
-                "state": self.state,
+            "state": state,
-            }
+            "state_version": apply._TRANSACTION_STATE_VERSION[state],
-        assert self.response is not None
+            "no_go": state != "finalized",
-        online = self.response["online_transaction_receipt"]
-        runtime = self.response["runtime_snapshot_receipt"]
-        if self.state == "finalized":
-            assert self.finalize_request is not None
-            finalized_at = (
-                datetime.fromisoformat(
-                    self.finalize_request["apply_receipt"]["payload"]["countersigned_at"]
-                )
-                + timedelta(seconds=1)
-            ).isoformat(timespec="seconds")
-            return 200, {
-                "schema_version": apply.FINALIZE_STATUS_SCHEMA_VERSION,
-                "idempotency_key": self.request["idempotency_key"],
-                "transaction_id": transaction_id,
-                "state": "finalized",
-                "no_go": False,
-                "transaction_mode": apply.batch.TRANSACTION_MODE,
-                "cross_database_atomicity": False,
-                "publication_item_count": 74,
-                "runtime_snapshot_dependency_count": 14,
-                "online_transaction_receipt_sha256": apply.batch._sha256_json(online),
-                "runtime_snapshot_receipt_sha256": apply.batch._sha256_json(runtime),
-                "apply_receipt_sha256": self.finalize_request["apply_receipt_sha256"],
-                "finalized_at": finalized_at,
-            }
-        publication = copy.deepcopy(online["payload"]["item_results"])
-        snapshots = copy.deepcopy(runtime["payload"]["runtime_snapshot_dependencies"])
-        if self.tamper_status == "publication":
-            publication[0]["online_content_sha256"] = "f" * 64
-        if self.tamper_status == "runtime":
-            snapshots[0]["online_content_sha256"] = "f" * 64
-        return 200, {
-            "schema_version": apply.APPLY_STATUS_SCHEMA_VERSION,
-            "idempotency_key": self.request["idempotency_key"],
-            "transaction_id": transaction_id,
-            "state": "committed_and_verified",
             "legal_source_review_binding_sha256": self.request[
                 "legal_source_review_binding_sha256"
             ],
@@ -787,14 +763,110 @@ class FakePublicationService:
             "tenant_id": self.request["tenant_id"],
             "transaction_mode": apply.batch.TRANSACTION_MODE,
             "cross_database_atomicity": False,
-            "publication_item_results": publication,
+            "publication_item_results": [],
-            "publication_item_count": 74,
+            "publication_item_count": 0,
-            "runtime_snapshot_dependencies": snapshots,
+            "runtime_snapshot_dependencies": [],
-            "runtime_snapshot_dependency_count": 14,
+            "runtime_snapshot_dependency_count": 0,
-            "online_transaction_receipt_sha256": apply.batch._sha256_json(online),
+            "participant_evidence": [],
-            "runtime_snapshot_receipt_sha256": apply.batch._sha256_json(runtime),
+            "online_transaction_receipt": None,
-            "participant_evidence": online["payload"]["participant_evidence"],
+            "online_transaction_receipt_sha256": None,
+            "runtime_snapshot_receipt": None,
+            "runtime_snapshot_receipt_sha256": None,
+            "apply_receipt": None,
+            "apply_receipt_sha256": None,
+            "finalized_at": None,
+            "rollback_response": None,
+            "rollback_response_sha256": None,
+            "failure_code": None,
+            "failure_at": None,
+            "failure_evidence_sha256": None,
+            "status_updated_at": self.now.isoformat(timespec="seconds"),
         }
+        if state in {"committed_and_verified", "finalized"}:
+            assert self.response is not None
+            online = self.response["online_transaction_receipt"]
+            runtime = self.response["runtime_snapshot_receipt"]
+            result.update(
+                {
+                    "publication_item_results": copy.deepcopy(
+                        online["payload"]["item_results"]
+                    ),
+                    "publication_item_count": 74,
+                    "runtime_snapshot_dependencies": copy.deepcopy(
+                        runtime["payload"]["runtime_snapshot_dependencies"]
+                    ),
+                    "runtime_snapshot_dependency_count": 14,
+                    "participant_evidence": copy.deepcopy(
+                        online["payload"]["participant_evidence"]
+                    ),
+                    "online_transaction_receipt": copy.deepcopy(online),
+                    "online_transaction_receipt_sha256": apply.batch._sha256_json(
+                        online
+                    ),
+                    "runtime_snapshot_receipt": copy.deepcopy(runtime),
+                    "runtime_snapshot_receipt_sha256": apply.batch._sha256_json(
+                        runtime
+                    ),
+                }
+            )
+            if self.tamper_status == "publication":
+                result["publication_item_results"][0]["online_content_sha256"] = "f" * 64
+            if self.tamper_status == "runtime":
+                result["runtime_snapshot_dependencies"][0]["online_content_sha256"] = (
+                    "f" * 64
+                )
+        if state == "finalized":
+            assert self.finalize_request is not None and self.finalized_at is not None
+            result.update(
+                {
+                    "apply_receipt": copy.deepcopy(
+                        self.finalize_request["apply_receipt"]
+                    ),
+                    "apply_receipt_sha256": self.finalize_request[
+                        "apply_receipt_sha256"
+                    ],
+                    "finalized_at": self.finalized_at,
+                    "status_updated_at": self.finalized_at,
+                }
+            )
+        elif state == "rolled_back":
+            assert self.rollback_response is not None
+            result["rollback_response"] = copy.deepcopy(self.rollback_response)
+            result["rollback_response_sha256"] = apply.batch._sha256_json(
+                self.rollback_response
+            )
+        elif state == "irreversible_failure":
+            failure = {
+                "code": "contract_committed_evidence_pending",
+                "failed_at": self.now.isoformat(timespec="seconds"),
+                "transaction_id": transaction_id,
+            }
+            result["failure_code"] = failure["code"]
+            result["failure_at"] = failure["failed_at"]
+            result["failure_evidence_sha256"] = apply.batch._sha256_json(failure)
+        if self.tamper_status == "extra_field":
+            result["unexpected"] = True
+        if self.tamper_status == "request_sha256":
+            result["request_sha256"] = "f" * 64
+        if self.tamper_status == "state_version":
+            result["state_version"] -= 1
+        if self.tamper_status == "finalized_apply_sha" and state == "finalized":
+            result["apply_receipt_sha256"] = "f" * 64
+        if self.tamper_status == "state_downgrade" and state == "finalized":
+            result.update(
+                {
+                    "state": "committed_and_verified",
+                    "state_version": apply._TRANSACTION_STATE_VERSION[
+                        "committed_and_verified"
+                    ],
+                    "no_go": True,
+                    "apply_receipt": None,
+                    "apply_receipt_sha256": None,
+                    "finalized_at": None,
+                    "status_updated_at": self.now.isoformat(timespec="seconds"),
+                }
+            )
+        return 200, result
 
 
 @contextmanager
@@ -1218,7 +1290,7 @@ def test_expired_challenge_is_rejected_before_initial_dispatch(
             ).fetchone()[0]
         finally:
             ledger.close()
-    assert operation_count == 0
+    assert operation_count == 1
     assert service.apply_calls == 0
 
 
@@ -1332,7 +1404,84 @@ def test_restart_recovers_committed_dispatch_after_short_challenge_expires(
 
     assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION
     assert state == "online_committed_pending_countersign"
+    # Status v3 contains the full signed receipt pair, so restart recovery does
+    # not need to retransmit an already committed apply.
+    assert service.apply_calls == 1
+    assert service.rollback_calls == 0
+
+
+def test_lost_apply_response_recovers_from_full_status_v3(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material)
+
+    class LoseApplyResponse:
+        def __init__(self, delegate) -> None:
+            self.delegate = delegate
+            self.lost = False
+
+        def request(self, method, path, *, payload, headers):
+            response = self.delegate.request(method, path, payload=payload, headers=headers)
+            if method == "POST" and path.endswith("/apply") and not self.lost:
+                self.lost = True
+                raise RuntimeError("simulated lost apply response")
+            return response
+
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
+        try:
+            execution = _execute(material, LoseApplyResponse(client), ledger)
+            state = ledger.connection.execute(
+                "SELECT state FROM rc06_apply_operations"
+            ).fetchone()[0]
+        finally:
+            ledger.close()
+    assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION
+    assert state == "online_committed_pending_countersign"
+    assert service.apply_calls == 1
+    assert service.rollback_calls == 0
+
+
+def test_irreversible_failure_only_replays_exact_apply_to_recover_evidence(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material)
+
+    class LoseEvidenceAfterContractCommit:
+        def __init__(self, delegate) -> None:
+            self.delegate = delegate
+            self.failed = False
+
+        def request(self, method, path, *, payload, headers):
+            response = self.delegate.request(method, path, payload=payload, headers=headers)
+            if method == "POST" and path.endswith("/apply") and not self.failed:
+                self.failed = True
+                service.state = "irreversible_failure"
+                raise RuntimeError("simulated post-commit KMS failure")
+            return response
+
+    ledger_path = tmp_path / "ledger.sqlite3"
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        ledger = apply.ApplyLedger(ledger_path)
+        try:
+            with pytest.raises(apply.ApplyPublicationError, match="rollback is forbidden"):
+                _execute(material, LoseEvidenceAfterContractCommit(client), ledger)
+        finally:
+            ledger.close()
+        assert service.rollback_calls == 0
+        material["now"] += timedelta(minutes=10)
+        ledger = apply.ApplyLedger(ledger_path)
+        try:
+            execution = _execute(material, client, ledger)
+        finally:
+            ledger.close()
+    assert execution["schema_version"] == apply.EXECUTION_EVIDENCE_SCHEMA_VERSION
     assert service.apply_calls == 2
+    assert service.state == "committed_and_verified"
     assert service.rollback_calls == 0
 
 
@@ -1348,14 +1497,17 @@ def test_terminal_commit_verification_failure_never_rolls_back(
         client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
         ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
         try:
-            with pytest.raises(apply.ApplyPublicationError, match="rollback is forbidden"):
+            with pytest.raises(
+                apply.ApplyPublicationError,
+                match="commit status is unknown; no rollback was attempted",
+            ):
                 _execute(material, client, ledger)
             state = ledger.connection.execute(
                 "SELECT state FROM rc06_apply_operations"
             ).fetchone()[0]
         finally:
             ledger.close()
-    assert state == "committed_evidence_invalid_no_go"
+    assert state == "commit_status_unknown"
     assert service.state == "committed_and_verified"
     assert service.rollback_calls == 0
 
@@ -1488,6 +1640,138 @@ def test_finalize_failure_remains_no_go_and_never_rolls_back(
     assert service.rollback_calls == 0
 
 
+def test_lost_finalize_response_recovers_exact_finalized_status(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material)
+
+    class LoseFinalizeResponse:
+        def __init__(self, delegate) -> None:
+            self.delegate = delegate
+            self.lost = False
+
+        def request(self, method, path, *, payload, headers):
+            response = self.delegate.request(method, path, payload=payload, headers=headers)
+            if method == "POST" and path.endswith("/finalize") and not self.lost:
+                self.lost = True
+                raise RuntimeError("simulated lost finalize response")
+            return response
+
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
+        try:
+            execution = _execute(material, client, ledger)
+            at = material["now"] + timedelta(minutes=10)
+            receipt = _countersign(material, execution, at)
+            final = _finalize(
+                material,
+                LoseFinalizeResponse(client),
+                ledger,
+                execution,
+                receipt,
+                at,
+            )
+        finally:
+            ledger.close()
+    assert final["online_finalize_status"]["state"] == "finalized"
+    assert final["apply_receipt"] == receipt
+    assert service.finalize_calls == 1
+    assert service.rollback_calls == 0
+
+
+def test_finalized_remote_recovers_after_local_ledger_loss_without_redispatch(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material)
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        first = apply.ApplyLedger(tmp_path / "first.sqlite3")
+        try:
+            execution = _execute(material, client, first)
+            at = material["now"] + timedelta(minutes=10)
+            receipt = _countersign(material, execution, at)
+            original_final = _finalize(material, client, first, execution, receipt, at)
+        finally:
+            first.close()
+
+        material["now"] = at
+        recovered = apply.ApplyLedger(tmp_path / "replacement.sqlite3")
+        try:
+            recovered_execution = _execute(material, client, recovered)
+            recovered_final = _finalize(
+                material,
+                client,
+                recovered,
+                recovered_execution,
+                receipt,
+                at,
+            )
+        finally:
+            recovered.close()
+    assert recovered_execution == execution
+    assert recovered_final == original_final
+    assert service.apply_calls == 1
+    assert service.finalize_calls == 1
+    assert service.rollback_calls == 0
+
+
+@pytest.mark.parametrize("tamper_status", ("state_downgrade", "finalized_apply_sha"))
+def test_finalized_status_downgrade_or_receipt_tamper_preserves_apply_evidence(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    tamper_status: str,
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material)
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
+        try:
+            execution = _execute(material, client, ledger)
+            service.tamper_status = tamper_status
+            at = material["now"] + timedelta(minutes=10)
+            receipt = _countersign(material, execution, at)
+            with pytest.raises(apply.ApplyPublicationError):
+                _finalize(material, client, ledger, execution, receipt, at)
+            state, stored = ledger.load(
+                material["idempotency_key"], execution["request_sha256"]
+            )
+        finally:
+            ledger.close()
+    assert state == "finalize_failed_no_go"
+    assert stored == execution
+    assert service.rollback_calls == 0
+
+
+@pytest.mark.parametrize("tamper_status", ("extra_field", "request_sha256", "state_version"))
+def test_status_v3_rejects_exact_field_identity_and_version_tamper(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    tamper_status: str,
+) -> None:
+    material = _material(monkeypatch, datetime(2026, 10, 6, 8, 0, tzinfo=UTC))
+    service = _service(material, tamper_status=tamper_status)
+    with _serve(service) as url:
+        client = apply.JsonHttpClient(url, "test-token", allow_http_loopback=True)
+        ledger = apply.ApplyLedger(tmp_path / "ledger.sqlite3")
+        try:
+            with pytest.raises(
+                apply.ApplyPublicationError,
+                match="commit status is unknown; no rollback was attempted",
+            ):
+                _execute(material, client, ledger)
+            state = ledger.connection.execute(
+                "SELECT state FROM rc06_apply_operations"
+            ).fetchone()[0]
+        finally:
+            ledger.close()
+    assert state == "commit_status_unknown"
+    assert service.rollback_calls == 0
+
+
 @pytest.mark.parametrize("mode", ["missing", "expired", "untrusted_keyring"])
 def test_legal_source_review_is_fail_closed_before_http(
     tmp_path: Path,