feat: add opt-in reconciliation recovery state ledger

.github/workflows/sync-cloud-run-env.yml

@@ -167,6 +167,9 @@ jobs:
       STRATEGY_PLUGIN_ALERT_TELEGRAM_BODY_MAX_CHARS: ${{ vars.STRATEGY_PLUGIN_ALERT_TELEGRAM_BODY_MAX_CHARS }}
       # Strategy-owned defaults continue to come from UsEquityStrategies; this workflow only syncs platform/runtime inputs.
       EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }}
+      # Empty by default. When explicitly set, the workflow downloads one
+      # immutable, private recovery ledger before deriving its sync plan.
+      IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_URI: ${{ vars.IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_URI }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ secrets.GLOBAL_TELEGRAM_CHAT_ID }}
       NOTIFY_LANG: ${{ vars.NOTIFY_LANG }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
@@ -294,6 +297,39 @@ jobs:
         with:
           python-version: "3.12"
 
+      - name: Authenticate to Google Cloud
+        id: auth
+        if: steps.config.outputs.enabled == 'true'
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+
+      - name: Set up gcloud
+        if: steps.config.outputs.enabled == 'true'
+        uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.GCP_PROJECT_ID }}
+          version: ">= 416.0.0"
+
+      - name: Fetch opt-in immutable recovery state ledger
+        if: steps.config.outputs.enabled == 'true' && env.IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_URI != ''
+        env:
+          RECOVERY_STATE_LEDGER_URI: ${{ env.IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_URI }}
+        run: |
+          set -euo pipefail
+          case "${RECOVERY_STATE_LEDGER_URI}" in
+            gs://qsl-reconciliation-recovery-private-ibkr/reconciliation-recovery/ibkr/state/*.json) ;;
+            *)
+              echo "Recovery state ledger URI must use the dedicated private immutable prefix." >&2
+              exit 1
+              ;;
+          esac
+          recovery_state_ledger_path="${RUNNER_TEMP}/ibkr-reconciliation-recovery-state-ledger.json"
+          gcloud storage cp --quiet "${RECOVERY_STATE_LEDGER_URI}" "${recovery_state_ledger_path}" >/dev/null
+          chmod 600 "${recovery_state_ledger_path}"
+          echo "IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_PATH=${recovery_state_ledger_path}" >> "$GITHUB_ENV"
+
       - name: Install strategy status dependencies
         if: steps.config.outputs.enabled == 'true'
         run: |
@@ -406,21 +442,6 @@ jobs:
             exit 1
           fi
 
-      - name: Authenticate to Google Cloud
-        id: auth
-        if: steps.config.outputs.enabled == 'true'
-        uses: google-github-actions/auth@v3
-        with:
-          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
-          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
-
-      - name: Set up gcloud
-        if: steps.config.outputs.enabled == 'true'
-        uses: google-github-actions/setup-gcloud@v3
-        with:
-          project_id: ${{ env.GCP_PROJECT_ID }}
-          version: ">= 416.0.0"
-
       - name: Require configured scheduler migration before HK verify
         if: steps.config.outputs.env_sync_enabled == 'true' && env.WORKFLOW_TARGET == 'hk-verify'
         run: |

docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md

@@ -51,6 +51,28 @@ AIAuditBridge 的完整 `reconciliation_baseline` 输出。它会同时核验：
 券商、执行标记或订单写入代码。下一层单独的最小权限控制器才可消费该计划，并且仍要
 在同一目标上比较五项摘要后执行一次精确 CAS。
 
+## 只读状态账本适配层（默认关闭）
+
+为避免把一段旧的 `RUNTIME_TARGET_JSON` 直接写回 GitHub 变量，部署链路可选择读取一份
+私有、不可覆盖的状态账本。账本的结构固定为
+`ibkr_reconciliation_recovery_state_ledger.v1`，且只含四项：`recovery_id`、
+`service_name` 和上述完整 QPK `transition_plan`（另加版本）。它**不**携带下一个运行
+目标、账户、策略、仓位、订单或执行权限。
+
+消费时系统从当前完整运行目标推导结果，并逐项验证：服务必须唯一匹配、平台必须为
+IBKR、当前状态必须仍是 `RECONCILE_ONLY`、基线 ID 与目标指纹必须等于计划的冻结值，且
+QPK 计划中的五项摘要、`no_order=true`、`execution_authority_granted=false` 和 CAS 标志
+必须完整存在。验证成功后，唯一允许的差异是
+`live_continuity.state: RECONCILE_ONLY -> ACTIVE_LKG`；五项摘要以
+`IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON` 注入运行环境。任何字段缺失、账本重放、目标
+漂移、服务不匹配或多服务误匹配都会失败关闭。
+
+工作流只有在仓库变量 `IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_URI` 显式非空时才会读取
+账本；URI 还必须落在专用私有桶的
+`reconciliation-recovery/ibkr/state/*.json` 前缀。未设置时不下载账本、不改变同步计划，
+既有实盘目标也不受影响。本阶段不写入该变量，也不创建账本对象；因此它只是经过测试的
+兼容入口，而不是一次自动或隐式的实盘恢复。
+
 ## 故障注入回归
 
 恢复链路的回归测试会主动注入：控制台把不可执行策略篡改为可执行、五项摘要之一

scripts/build_cloud_run_env_sync_plan.py

@@ -57,6 +57,9 @@ def _should_add_local_src(candidate: Path) -> bool:
     DEFAULT_MARKET_TIMEZONE,
     resolve_market,
 )
+from scripts.reconciliation_recovery_state_ledger import (  # noqa: E402
+    apply_recovery_state_ledger_from_env,
+)
 
 
 TARGETS_JSON_ENV = "CLOUD_RUN_SERVICE_TARGETS_JSON"
@@ -103,6 +106,7 @@ def _should_add_local_src(candidate: Path) -> bool:
     "IBKR_STRATEGY_CONFIG_PATH",
     "IBKR_STRATEGY_PLUGIN_MOUNTS_JSON",
     "IBKR_RECONCILIATION_OUTPUT_PATH",
+    "IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON",
     "IBKR_DRY_RUN_ONLY",
     "IBKR_EXECUTION_DEDUP_ENABLED",
     "IBKR_PAPER_LIQUIDATE_ONLY",
@@ -329,17 +333,22 @@ def build_sync_plan(env: Mapping[str, str] = os.environ) -> dict[str, object]:
         }
         for row in get_platform_profile_status_matrix()
     }
-    planned_targets = [
-        _build_target_plan(
+    planned_targets = []
+    recovery_ledger_applied_count = 0
+    for target in target_entries:
+        planned_target = _build_target_plan(
             target=target,
             defaults=defaults,
             env=env,
             status_rows=status_rows,
             per_service_mode=per_service_mode,
             platform_config=platform_config,
         )
-        for target in target_entries
-    ]
+        if planned_target.pop("_recovery_state_ledger_applied", False):
+            recovery_ledger_applied_count += 1
+        planned_targets.append(planned_target)
+    if str(env.get("IBKR_RECONCILIATION_RECOVERY_STATE_LEDGER_PATH") or "").strip() and recovery_ledger_applied_count != 1:
+        raise ValueError("reconciliation recovery state ledger must match exactly one Cloud Run target")
     if not planned_targets:
         raise ValueError(
             f"{TARGETS_JSON_ENV}, CLOUD_RUN_SERVICES, or CLOUD_RUN_SERVICE is required"
@@ -394,6 +403,10 @@ def _build_target_plan(
         _target_field(target, defaults, "cloud_run_service"),
     )
     runtime_target = _resolve_runtime_target(target, defaults, env, per_service_mode)
+    runtime_target, recovery_expected_digests = apply_recovery_state_ledger_from_env(
+        runtime_target=runtime_target,
+        env=env,
+    )
     if not service_name:
         service_name = str(runtime_target.get("service_name") or "").strip()
     if not service_name:
@@ -516,6 +529,18 @@ def _build_target_plan(
         else:
             env_values[name] = value
 
+    if recovery_expected_digests is not None:
+        # An immutable ledger is the only source permitted to activate the
+        # frozen baseline, so it deliberately overrides every normal config
+        # layer for the accompanying five reconciliation digests.
+        env_values["IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON"] = json.dumps(
+            recovery_expected_digests,
+            separators=(",", ":"),
+            sort_keys=True,
+        )
+        if "IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON" in remove_env_vars:
+            remove_env_vars.remove("IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON")
+
     if _runtime_target_enabled(env_values):
         _validate_profile_inputs(
             service_name=service_name,
@@ -553,6 +578,7 @@ def _build_target_plan(
         "env": env_values,
         "scheduler": scheduler,
         "remove_env_vars": sorted(set(remove_env_vars) - set(env_values)),
+        "_recovery_state_ledger_applied": recovery_expected_digests is not None,
     }
 
 

scripts/reconciliation_recovery_state_ledger.py

@@ -0,0 +1,113 @@
+"""Fail-closed, local consumption of an immutable recovery state ledger.
+
+This module deliberately has no Google Cloud, GitHub, Cloud Run, broker, or
+order client.  The deployment workflow may opt in by supplying a locally
+downloaded ledger file.  Without that explicit path, callers retain their
+legacy runtime-target behaviour exactly.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+from collections.abc import Mapping
+from pathlib import Path
+
+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"
+
+
+def _ledger_service_name(ledger: Mapping[str, object]) -> str:
+    service_name = ledger.get("service_name")
+    if not isinstance(service_name, str) or service_name != service_name.strip() or not 3 <= len(service_name) <= 127:
+        raise ValueError("reconciliation recovery state ledger service_name is invalid")
+    return service_name
+
+
+def apply_recovery_state_ledger(
+    *,
+    runtime_target: Mapping[str, object],
+    ledger: Mapping[str, object],
+) -> tuple[dict[str, object], dict[str, str]]:
+    """Return the one allowed target transition and its five fixed digests.
+
+    A ledger contains only a QPK transition plan.  The desired target is never
+    supplied by the ledger: it is derived from the current target by changing
+    exactly ``live_continuity.state`` after every frozen-baseline precondition
+    has been revalidated.
+    """
+
+    required = {"schema_version", "recovery_id", "service_name", "transition_plan"}
+    if not isinstance(ledger, Mapping) or set(ledger) != required:
+        raise ValueError("reconciliation recovery state ledger has invalid fields")
+    if ledger.get("schema_version") != RECOVERY_STATE_LEDGER_SCHEMA_VERSION:
+        raise ValueError("unsupported reconciliation recovery state ledger schema")
+    ledger_service_name = _ledger_service_name(ledger)
+
+    raw_plan = ledger.get("transition_plan")
+    if not isinstance(raw_plan, Mapping):
+        raise ValueError("reconciliation recovery state ledger is missing transition_plan")
+    plan = ReconciliationRecoveryTransitionPlan.from_dict(raw_plan)
+    if str(ledger.get("recovery_id") or "").strip() != plan.recovery_id:
+        raise ValueError("reconciliation recovery state ledger recovery_id mismatch")
+
+    target = copy.deepcopy(dict(runtime_target))
+    if str(target.get("platform_id") or "").strip().lower() != "ibkr":
+        raise ValueError("reconciliation recovery state ledger only supports ibkr targets")
+    if str(target.get("service_name") or "").strip() != ledger_service_name:
+        raise ValueError("reconciliation recovery state ledger service_name mismatch")
+    continuity_payload = target.get("live_continuity")
+    continuity = build_live_continuity(continuity_payload)
+    continuity.assert_matches_target(target)
+    if continuity.state != plan.expected_live_continuity_state:
+        raise ValueError("reconciliation recovery state ledger current continuity state mismatch")
+    if continuity.baseline_id != plan.baseline_id:
+        raise ValueError("reconciliation recovery state ledger baseline_id mismatch")
+    if continuity.baseline_target_sha256 != plan.baseline_target_sha256:
+        raise ValueError("reconciliation recovery state ledger baseline digest mismatch")
+
+    next_continuity = continuity.to_dict()
+    next_continuity["state"] = plan.next_live_continuity_state
+    target["live_continuity"] = next_continuity
+    return target, dict(plan.expected_digests)
+
+
+def apply_recovery_state_ledger_from_env(
+    *,
+    runtime_target: Mapping[str, object],
+    env: Mapping[str, str],
+) -> tuple[dict[str, object], dict[str, str] | None]:
+    """Apply an explicitly supplied local ledger, or retain the original target."""
+
+    raw_path = str(env.get(RECOVERY_STATE_LEDGER_PATH_ENV) or "").strip()
+    if not raw_path:
+        return dict(runtime_target), None
+    try:
+        value = json.loads(Path(raw_path).read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise ValueError("reconciliation recovery state ledger cannot be read") from exc
+    except json.JSONDecodeError as exc:
+        raise ValueError("reconciliation recovery state ledger is not valid JSON") from exc
+    if not isinstance(value, Mapping):
+        raise ValueError("reconciliation recovery state ledger must be a JSON object")
+    ledger_service_name = _ledger_service_name(value)
+    target_service_name = str(runtime_target.get("service_name") or "").strip()
+    if target_service_name != ledger_service_name:
+        return dict(runtime_target), None
+    target, expected_digests = apply_recovery_state_ledger(
+        runtime_target=runtime_target,
+        ledger=value,
+    )
+    return target, expected_digests
+
+
+__all__ = [
+    "RECOVERY_STATE_LEDGER_PATH_ENV",
+    "RECOVERY_STATE_LEDGER_SCHEMA_VERSION",
+    "apply_recovery_state_ledger",
+    "apply_recovery_state_ledger_from_env",
+]