feat: publish immutable recovery state ledgers (#430)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md

@@ -73,6 +73,13 @@ QPK 计划中的五项摘要、`no_order=true`、`execution_authority_granted=fa
 既有实盘目标也不受影响。本阶段不写入该变量，也不创建账本对象；因此它只是经过测试的
 兼容入口，而不是一次自动或隐式的实盘恢复。
 
+`scripts/publish_reconciliation_recovery_state_ledger.py` 则是与消费端分离的发布适配器。
+它只接受刚生成的 `ibkr_reconciliation_recovery_verification.v1`：验证结果必须无阻断、
+完整携带 QPK 计划，并且仍声明 `verify_only`、无订单、无执行授权和未尝试状态写入。默认
+只打印最小账本 JSON；只有显式提供状态前缀的 GCS URI 时才调用
+`if_generation_match=0` 创建对象。该写入不读取、列举、覆盖或删除对象，也不会设置工作流
+URI、部署 Cloud Run、连接券商或提交订单。实际启用仍需要单独的高风险控制面动作。
+
 ## 故障注入回归
 
 恢复链路的回归测试会主动注入：控制台把不可执行策略篡改为可执行、五项摘要之一

scripts/publish_reconciliation_recovery_state_ledger.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Build or explicitly create an immutable IBKR recovery state ledger.
+
+By default this command only prints a locally derived ledger.  The optional
+GCS write uses create-only semantics and does not deploy, modify a runtime
+target, connect to IBKR, or submit an order.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+from scripts.reconciliation_recovery_state_ledger import (
+    RECOVERY_STATE_LEDGER_SCHEMA_VERSION,
+    build_recovery_state_ledger,
+)
+
+
+_STATE_LEDGER_PREFIX = "reconciliation-recovery/ibkr/state/"
+
+
+def _load_json(path: Path) -> dict[str, object]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise ValueError("unable to read reconciliation recovery verification") from exc
+    except json.JSONDecodeError as exc:
+        raise ValueError("reconciliation recovery verification is not valid JSON") from exc
+    if not isinstance(value, dict):
+        raise ValueError("reconciliation recovery verification must be a JSON object")
+    return value
+
+
+def _parse_state_ledger_uri(value: str) -> tuple[str, str]:
+    normalized = str(value or "").strip()
+    if not normalized.startswith("gs://"):
+        raise ValueError("state_ledger_uri must use gs://")
+    bucket_name, separator, object_name = normalized.removeprefix("gs://").partition("/")
+    if not bucket_name or not separator or not object_name.startswith(_STATE_LEDGER_PREFIX) or not object_name.endswith(".json"):
+        raise ValueError("state_ledger_uri must use the IBKR recovery state prefix and .json suffix")
+    return bucket_name, object_name
+
+
+def archive_recovery_state_ledger(
+    ledger: Mapping[str, object],
+    *,
+    state_ledger_uri: str,
+    storage_client_factory: Any | None = None,
+) -> dict[str, str]:
+    """Create an immutable ledger object without reading, listing, or replacing it."""
+
+    if ledger.get("schema_version") != RECOVERY_STATE_LEDGER_SCHEMA_VERSION:
+        raise ValueError("reconciliation recovery state ledger has an unsupported schema")
+    bucket_name, object_name = _parse_state_ledger_uri(state_ledger_uri)
+    if storage_client_factory is None:
+        try:
+            from google.cloud import storage
+        except ImportError as exc:  # pragma: no cover - runtime image provides this dependency.
+            raise RuntimeError("google-cloud-storage is required to archive a recovery state ledger") from exc
+        client = storage.Client()
+    else:
+        client = storage_client_factory()
+    try:
+        client.bucket(bucket_name).blob(object_name).upload_from_string(
+            json.dumps(dict(ledger), ensure_ascii=False, sort_keys=True, separators=(",", ":")),
+            content_type="application/json",
+            if_generation_match=0,
+        )
+    except Exception as exc:
+        raise RuntimeError("reconciliation recovery state ledger was not created") from exc
+    return {
+        "uri": f"gs://{bucket_name}/{object_name}",
+        "schema_version": RECOVERY_STATE_LEDGER_SCHEMA_VERSION,
+        "recovery_id": str(ledger["recovery_id"]),
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Build or explicitly archive an immutable IBKR recovery state ledger."
+    )
+    parser.add_argument("--verification", type=Path, required=True, help="Fresh verify-only recovery result")
+    parser.add_argument("--service-name", required=True, help="One exact Cloud Run service name")
+    parser.add_argument(
+        "--archive-gcs-uri",
+        help="Explicit private gs://.../reconciliation-recovery/ibkr/state/*.json object; uses create-only generation precondition",
+    )
+    args = parser.parse_args(argv)
+    try:
+        ledger = build_recovery_state_ledger(
+            verification=_load_json(args.verification),
+            service_name=args.service_name,
+        )
+        if args.archive_gcs_uri:
+            print(json.dumps({"ledger": ledger, "archive": archive_recovery_state_ledger(
+                ledger,
+                state_ledger_uri=args.archive_gcs_uri,
+            )}, ensure_ascii=False, sort_keys=True))
+        else:
+            print(json.dumps(ledger, ensure_ascii=False, sort_keys=True))
+    except (RuntimeError, ValueError) as exc:
+        parser.error(str(exc))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/reconciliation_recovery_state_ledger.py

@@ -12,13 +12,15 @@
 import json
 from collections.abc import Mapping
 from pathlib import Path
+from typing import Any
 
 from quant_platform_kit.common.live_continuity import build_live_continuity
 from quant_platform_kit.common.reconciliation_recovery import ReconciliationRecoveryTransitionPlan
 
 
 RECOVERY_STATE_LEDGER_SCHEMA_VERSION = "ibkr_reconciliation_recovery_state_ledger.v1"
 RECOVERY_STATE_LEDGER_PATH_ENV = "IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_PATH"
+RECOVERY_VERIFICATION_SCHEMA_VERSION = "ibkr_reconciliation_recovery_verification.v1"
 
 
 def _ledger_service_name(ledger: Mapping[str, object]) -> str:
@@ -28,6 +30,64 @@ def _ledger_service_name(ledger: Mapping[str, object]) -> str:
     return service_name
 
 
+def build_recovery_state_ledger(
+    *,
+    verification: Mapping[str, object],
+    service_name: str,
+) -> dict[str, object]:
+    """Build one minimal state ledger from an already fresh verify-only result.
+
+    This is pure serialization.  It never creates a cloud object or changes a
+    runtime target.  Requiring the complete verify-only receipt prevents a
+    caller from turning a standalone plan into an activation intent.
+    """
+
+    required = {
+        "schema_version",
+        "recovery_id",
+        "candidate_sha256",
+        "confirmation_sha256",
+        "ready_for_atomic_state_transition",
+        "findings",
+        "transition_plan",
+        "policy",
+    }
+    if not isinstance(verification, Mapping) or set(verification) != required:
+        raise ValueError("reconciliation recovery verification has invalid fields")
+    if verification.get("schema_version") != RECOVERY_VERIFICATION_SCHEMA_VERSION:
+        raise ValueError("unsupported reconciliation recovery verification schema")
+    if verification.get("ready_for_atomic_state_transition") is not True or verification.get("findings") != []:
+        raise ValueError("reconciliation recovery verification is not ready for a state ledger")
+    expected_policy = {
+        "controller_mode": "verify_only",
+        "no_order": True,
+        "execution_authority_granted": False,
+        "state_write_attempted": False,
+    }
+    if verification.get("policy") != expected_policy:
+        raise ValueError("reconciliation recovery verification has an invalid non-execution policy")
+    raw_plan = verification.get("transition_plan")
+    if not isinstance(raw_plan, Mapping):
+        raise ValueError("reconciliation recovery verification is missing transition_plan")
+    plan = ReconciliationRecoveryTransitionPlan.from_dict(raw_plan)
+    recovery_id = str(verification.get("recovery_id") or "").strip()
+    candidate_sha256 = str(verification.get("candidate_sha256") or "").strip().lower().removeprefix("sha256:")
+    confirmation_sha256 = str(verification.get("confirmation_sha256") or "").strip().lower().removeprefix("sha256:")
+    if recovery_id != plan.recovery_id:
+        raise ValueError("reconciliation recovery verification recovery_id mismatch")
+    if candidate_sha256 != plan.candidate_sha256:
+        raise ValueError("reconciliation recovery verification candidate digest mismatch")
+    if confirmation_sha256 != plan.confirmation_sha256:
+        raise ValueError("reconciliation recovery verification confirmation digest mismatch")
+    _ledger_service_name({"service_name": service_name})
+    return {
+        "schema_version": RECOVERY_STATE_LEDGER_SCHEMA_VERSION,
+        "recovery_id": plan.recovery_id,
+        "service_name": service_name,
+        "transition_plan": plan.to_dict(),
+    }
+
+
 def apply_recovery_state_ledger(
     *,
     runtime_target: Mapping[str, object],
@@ -108,6 +168,8 @@ def apply_recovery_state_ledger_from_env(
 __all__ = [
     "RECOVERY_STATE_LEDGER_PATH_ENV",
     "RECOVERY_STATE_LEDGER_SCHEMA_VERSION",
+    "RECOVERY_VERIFICATION_SCHEMA_VERSION",
     "apply_recovery_state_ledger",
     "apply_recovery_state_ledger_from_env",
+    "build_recovery_state_ledger",
 ]

tests/test_reconciliation_recovery_state_ledger_publisher.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from quant_platform_kit.common.reconciliation_recovery import ReconciliationRecoveryTransitionPlan
+
+from scripts.publish_reconciliation_recovery_state_ledger import archive_recovery_state_ledger
+from scripts.reconciliation_recovery_state_ledger import (
+    RECOVERY_STATE_LEDGER_SCHEMA_VERSION,
+    build_recovery_state_ledger,
+)
+
+
+def _digest(character: str) -> str:
+    return character * 64
+
+
+def _verification() -> dict[str, object]:
+    plan = ReconciliationRecoveryTransitionPlan(
+        recovery_id="ibkr-soxl-live-recovery",
+        candidate_sha256=_digest("a"),
+        confirmation_sha256=_digest("b"),
+        baseline_id="soxl-ibkr-lkg-20260830",
+        baseline_target_sha256=_digest("c"),
+        expected_digests={
+            "positions_sha256": _digest("d"),
+            "cash_sha256": _digest("e"),
+            "open_orders_sha256": _digest("f"),
+            "recent_executions_sha256": _digest("0"),
+            "local_execution_ledger_sha256": _digest("1"),
+        },
+        verified_at=datetime(2026, 8, 31, 1, 5, tzinfo=timezone.utc),
+    )
+    return {
+        "schema_version": "ibkr_reconciliation_recovery_verification.v1",
+        "recovery_id": plan.recovery_id,
+        "candidate_sha256": plan.candidate_sha256,
+        "confirmation_sha256": plan.confirmation_sha256,
+        "ready_for_atomic_state_transition": True,
+        "findings": [],
+        "transition_plan": plan.to_dict(),
+        "policy": {
+            "controller_mode": "verify_only",
+            "no_order": True,
+            "execution_authority_granted": False,
+            "state_write_attempted": False,
+        },
+    }
+
+
+def test_build_state_ledger_requires_a_complete_non_executable_verification() -> None:
+    ledger = build_recovery_state_ledger(
+        verification=_verification(),
+        service_name="interactive-brokers-live-service",
+    )
+
+    assert ledger["schema_version"] == RECOVERY_STATE_LEDGER_SCHEMA_VERSION
+    assert ledger["service_name"] == "interactive-brokers-live-service"
+    assert set(ledger) == {"schema_version", "recovery_id", "service_name", "transition_plan"}
+
+    invalid = _verification()
+    invalid["ready_for_atomic_state_transition"] = False
+    with pytest.raises(ValueError, match="not ready"):
+        build_recovery_state_ledger(
+            verification=invalid,
+            service_name="interactive-brokers-live-service",
+        )
+
+
+def test_archive_state_ledger_uses_create_only_private_state_prefix() -> None:
+    ledger = build_recovery_state_ledger(
+        verification=_verification(),
+        service_name="interactive-brokers-live-service",
+    )
+    received: dict[str, object] = {}
+
+    class Blob:
+        def upload_from_string(self, payload: str, **kwargs: object) -> None:
+            received["payload"] = payload
+            received["kwargs"] = kwargs
+
+    class Bucket:
+        @staticmethod
+        def blob(name: str) -> Blob:
+            received["object_name"] = name
+            return Blob()
+
+    class Client:
+        @staticmethod
+        def bucket(name: str) -> Bucket:
+            received["bucket_name"] = name
+            return Bucket()
+
+    archive = archive_recovery_state_ledger(
+        ledger,
+        state_ledger_uri="gs://private-bucket/reconciliation-recovery/ibkr/state/recovery-1.json",
+        storage_client_factory=Client,
+    )
+
+    assert archive["uri"] == "gs://private-bucket/reconciliation-recovery/ibkr/state/recovery-1.json"
+    assert received["bucket_name"] == "private-bucket"
+    assert received["object_name"] == "reconciliation-recovery/ibkr/state/recovery-1.json"
+    assert received["kwargs"] == {
+        "content_type": "application/json",
+        "if_generation_match": 0,
+    }
+
+    with pytest.raises(ValueError, match="state prefix"):
+        archive_recovery_state_ledger(
+            ledger,
+            state_ledger_uri="gs://private-bucket/reconciliation-recovery/ibkr/source/recovery-1.json",
+            storage_client_factory=Client,
+        )