feat: archive private recovery evidence immutably (#427)

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

Author: Pigbibi

docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md

@@ -32,6 +32,11 @@ AIAuditBridge 的完整 `reconciliation_baseline` 输出。它会同时核验：
 券商、复核双审绑定，并以原子比较并设置方式转换状态；任一失败都保持
 `RECONCILE_ONLY`。
 
+来源发布器还可在显式给出
+`gs://.../reconciliation-recovery/ibkr/source/...` 时，把完整候选与双审回执写入
+私有证据包。该包不发送给 QRS；写入固定使用 GCS `if_generation_match=0`，所以已存在
+对象会失败而不会被覆盖、读取或删除。验证器只能在专用私有存储中读取它。
+
 ## 私有验证器（暂不写状态）
 
 `scripts/verify_reconciliation_recovery.py` 是恢复链路的第二层。它使用一枚不同于

scripts/publish_reconciliation_recovery_source.py

@@ -34,6 +34,7 @@
 
 
 RECONCILIATION_RECOVERY_SYNC_TOKEN_ENV = "RECONCILIATION_RECOVERY_SYNC_TOKEN"
+RECONCILIATION_RECOVERY_PRIVATE_EVIDENCE_SCHEMA_VERSION = "ibkr_reconciliation_recovery_private_evidence.v1"
 _SHA256_LENGTH = 64
 
 
@@ -165,6 +166,87 @@ def build_recovery_source_snapshot(
     ).to_dict()
 
 
+def _parse_private_evidence_uri(value: str) -> tuple[str, str]:
+    """Allow only an explicit, immutable IBKR source-artifact object path."""
+
+    normalized = str(value or "").strip()
+    if not normalized.startswith("gs://"):
+        raise ValueError("private_evidence_uri must use gs://")
+    remainder = normalized.removeprefix("gs://")
+    bucket_name, separator, object_name = remainder.partition("/")
+    required_prefix = "reconciliation-recovery/ibkr/source/"
+    if not bucket_name or not separator or not object_name.startswith(required_prefix) or object_name.endswith("/"):
+        raise ValueError("private_evidence_uri must use the IBKR recovery source prefix")
+    return bucket_name, object_name
+
+
+def build_private_evidence_package(
+    *,
+    snapshot: Mapping[str, object],
+    candidate_payload: Mapping[str, Any],
+    dual_review_payload: Mapping[str, Any],
+) -> dict[str, object]:
+    """Package private inputs for the later verifier, never for QRS ingress."""
+
+    candidate = extract_baseline_candidate(candidate_payload)
+    extract_bound_dual_review(dual_review_payload, candidate=candidate)
+    recoveries = snapshot.get("recoveries")
+    if not isinstance(recoveries, list) or len(recoveries) != 1 or not isinstance(recoveries[0], Mapping):
+        raise ValueError("recovery source snapshot must contain exactly one recovery")
+    recovery_id = str(recoveries[0].get("recovery_id") or "").strip()
+    if not recovery_id:
+        raise ValueError("recovery source snapshot is missing recovery_id")
+    return {
+        "schema_version": RECONCILIATION_RECOVERY_PRIVATE_EVIDENCE_SCHEMA_VERSION,
+        "recovery_id": recovery_id,
+        "candidate_sha256": candidate.candidate_sha256,
+        "source_snapshot": dict(snapshot),
+        "baseline_candidate": dict(candidate_payload),
+        "dual_review": dict(dual_review_payload),
+    }
+
+
+def archive_private_evidence_package(
+    package: Mapping[str, object],
+    *,
+    private_evidence_uri: str,
+    storage_client_factory: Any | None = None,
+) -> dict[str, str]:
+    """Create one immutable private package with a GCS generation precondition.
+
+    The publisher role has create-only access to this prefix. A pre-existing
+    artifact therefore fails instead of being replaced, and this helper never
+    lists, reads, deletes, or rewrites an object.
+    """
+
+    bucket_name, object_name = _parse_private_evidence_uri(private_evidence_uri)
+    if package.get("schema_version") != RECONCILIATION_RECOVERY_PRIVATE_EVIDENCE_SCHEMA_VERSION:
+        raise ValueError("private evidence package has an unsupported schema_version")
+    payload = json.dumps(dict(package), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    if storage_client_factory is None:
+        try:
+            from google.cloud import storage
+        except ImportError as exc:  # pragma: no cover - production dependency is installed in runtime images.
+            raise RuntimeError("google-cloud-storage is required to archive private recovery evidence") from exc
+        client = storage.Client()
+    else:
+        client = storage_client_factory()
+    blob = client.bucket(bucket_name).blob(object_name)
+    try:
+        blob.upload_from_string(
+            payload,
+            content_type="application/json",
+            if_generation_match=0,
+        )
+    except Exception as exc:
+        raise RuntimeError("private recovery evidence archive was not created") from exc
+    return {
+        "uri": f"gs://{bucket_name}/{object_name}",
+        "schema_version": RECONCILIATION_RECOVERY_PRIVATE_EVIDENCE_SCHEMA_VERSION,
+        "candidate_sha256": str(package["candidate_sha256"]),
+    }
+
+
 def publish_recovery_source_snapshot(
     snapshot: Mapping[str, object],
     *,
@@ -252,6 +334,10 @@ def main(argv: list[str] | None = None) -> int:
         "--publish-url",
         help="Explicit QRS /api/internal/sync-reconciliation-recovery-source HTTPS URL; omitted means no network call",
     )
+    parser.add_argument(
+        "--archive-gcs-uri",
+        help="Explicit private gs://.../reconciliation-recovery/ibkr/source/... object; uses create-only generation precondition",
+    )
     args = parser.parse_args(argv)
     try:
         snapshot = build_recovery_source_snapshot(
@@ -261,13 +347,25 @@ def main(argv: list[str] | None = None) -> int:
             source_id=args.source_id,
             now=_parse_time(args.now),
         )
+        output: dict[str, object] = {"snapshot": snapshot}
+        if args.archive_gcs_uri:
+            output["private_evidence_archive"] = archive_private_evidence_package(
+                build_private_evidence_package(
+                    snapshot=snapshot,
+                    candidate_payload=_load_json(args.candidate, label="baseline candidate"),
+                    dual_review_payload=_load_json(args.dual_review, label="dual review"),
+                ),
+                private_evidence_uri=args.archive_gcs_uri,
+            )
         if args.publish_url:
             result = publish_recovery_source_snapshot(
                 snapshot,
                 publish_url=args.publish_url,
                 token=os.environ.get(RECONCILIATION_RECOVERY_SYNC_TOKEN_ENV, ""),
             )
-            print(json.dumps({"snapshot": snapshot, "publish": result}, ensure_ascii=False, sort_keys=True))
+            output["publish"] = result
+        if args.publish_url or args.archive_gcs_uri:
+            print(json.dumps(output, ensure_ascii=False, sort_keys=True))
         else:
             print(json.dumps(snapshot, ensure_ascii=False, sort_keys=True))
     except (ValueError, RuntimeError) as exc:

tests/test_reconciliation_recovery_source_publisher.py

@@ -9,6 +9,8 @@
     evaluate_broker_reconciliation_baseline_enrollment,
 )
 from scripts.publish_reconciliation_recovery_source import (
+    archive_private_evidence_package,
+    build_private_evidence_package,
     build_recovery_source_snapshot,
     publish_recovery_source_snapshot,
 )
@@ -150,3 +152,50 @@ def test_publish_rejects_non_recovery_endpoint() -> None:
             publish_url="https://console.example/api/manual-strategy-switch",
             token="dedicated-token",
         )
+
+
+def test_private_evidence_archive_is_create_only_and_never_uses_console_shape() -> None:
+    start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc)
+    candidate = _candidate_payload(start)
+    candidate_value = candidate["candidate"]
+    assert isinstance(candidate_value, dict)
+    review = _dual_review(str(candidate_value["candidate_sha256"]))
+    snapshot = build_recovery_source_snapshot(
+        candidate_payload=candidate,
+        dual_review_payload=review,
+        recovery_id="ibkr-soxl-live-recovery",
+        now=start + timedelta(minutes=3),
+    )
+    package = build_private_evidence_package(
+        snapshot=snapshot,
+        candidate_payload=candidate,
+        dual_review_payload=review,
+    )
+    observed: dict[str, object] = {}
+
+    class Blob:
+        def upload_from_string(self, value: str, **kwargs: object) -> None:
+            observed["value"] = value
+            observed["kwargs"] = kwargs
+
+    class Bucket:
+        @staticmethod
+        def blob(name: str) -> Blob:
+            observed["object_name"] = name
+            return Blob()
+
+    class Client:
+        @staticmethod
+        def bucket(name: str) -> Bucket:
+            observed["bucket_name"] = name
+            return Bucket()
+
+    result = archive_private_evidence_package(
+        package,
+        private_evidence_uri="gs://private-bucket/reconciliation-recovery/ibkr/source/ibkr-soxl-live-recovery/evidence.json",
+        storage_client_factory=Client,
+    )
+
+    assert result["uri"].startswith("gs://private-bucket/reconciliation-recovery/ibkr/source/")
+    assert observed["kwargs"] == {"content_type": "application/json", "if_generation_match": 0}
+    assert '"baseline_candidate"' in observed["value"]  # type: ignore[operator]