test: add reconciliation recovery fault injection (#426)

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

Author: Pigbibi

docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md

@@ -45,3 +45,11 @@ AIAuditBridge 的完整 `reconciliation_baseline` 输出。它会同时核验：
 `execution_authority_granted=false`、`state_write_attempted=false`；它没有 Cloud Run、
 券商、执行标记或订单写入代码。下一层单独的最小权限控制器才可消费该计划，并且仍要
 在同一目标上比较五项摘要后执行一次精确 CAS。
+
+## 故障注入回归
+
+恢复链路的回归测试会主动注入：控制台把不可执行策略篡改为可执行、五项摘要之一
+漂移、人工确认过期、确认与新回执同秒、运行目标不再是 `RECONCILE_ONLY`、以及控制器
+令牌被试图发送到普通管理站路径。每一种情况都必须抛出拒绝或返回没有
+`transition_plan` 的结果；测试同时断言 `state_write_attempted=false`。这让后续接入
+最小权限 CAS 时能持续证明“异常只能保持冻结，不能意外恢复实盘”。

tests/test_reconciliation_recovery_fault_injection.py

@@ -0,0 +1,108 @@
+"""Fault injection for the non-executable IBKR legacy recovery path.
+
+Each case simulates an untrusted or stale input at a trust boundary.  A test
+passes only when the verifier rejects it or returns no transition plan; none of
+these paths has a state-write or broker-order implementation.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from quant_platform_kit.common.broker_reconciliation import (
+    calculate_broker_reconciliation_evidence_sha256,
+)
+
+from tests.test_reconciliation_recovery_verify_only import (
+    _candidate_payload,
+    _console_response,
+    _dual_review,
+    _evidence,
+    _runtime_target,
+)
+from scripts.verify_reconciliation_recovery import (
+    read_console_confirmation,
+    verify_reconciliation_recovery,
+)
+
+
+def _inputs(start: datetime) -> tuple[dict[str, object], str]:
+    candidate = _candidate_payload(start)
+    candidate_value = candidate["candidate"]
+    assert isinstance(candidate_value, dict)
+    return candidate, str(candidate_value["candidate_sha256"])
+
+
+def test_fault_injection_tampered_console_policy_is_rejected_before_evidence_evaluation() -> None:
+    start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc)
+    candidate, candidate_sha256 = _inputs(start)
+    confirmation = _console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3))
+    confirmation["policy"] = {
+        "no_order": True,
+        "execution_authority_granted": True,
+        "controller_must_reverify": True,
+    }
+
+    with pytest.raises(ValueError, match="non-execution policy"):
+        verify_reconciliation_recovery(
+            candidate_payload=candidate,
+            dual_review_payload=_dual_review(candidate_sha256),
+            confirmation_payload=confirmation,
+            current_receipt_payload={"evidence": _evidence(observed_at=start + timedelta(minutes=4)).to_dict()},
+            runtime_target_payload=_runtime_target(),
+            recovery_id="ibkr-soxl-live-recovery",
+            now=start + timedelta(minutes=5),
+        )
+
+
+def test_fault_injection_current_cash_digest_drift_closes_transition_plan() -> None:
+    start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc)
+    candidate, candidate_sha256 = _inputs(start)
+    current = _evidence(observed_at=start + timedelta(minutes=4)).to_dict()
+    current["cash_sha256"] = "9" * 64
+    current["evidence_sha256"] = calculate_broker_reconciliation_evidence_sha256(current)
+
+    result = verify_reconciliation_recovery(
+        candidate_payload=candidate,
+        dual_review_payload=_dual_review(candidate_sha256),
+        confirmation_payload=_console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3)),
+        current_receipt_payload={"evidence": current},
+        runtime_target_payload=_runtime_target(),
+        recovery_id="ibkr-soxl-live-recovery",
+        now=start + timedelta(minutes=5),
+    )
+
+    assert result["ready_for_atomic_state_transition"] is False
+    assert result["transition_plan"] is None
+    assert "broker_reconciliation_cash_mismatch" in result["findings"]
+    assert result["policy"]["state_write_attempted"] is False  # type: ignore[index]
+
+
+def test_fault_injection_stale_human_confirmation_closes_transition_plan() -> None:
+    start = datetime(2026, 8, 31, 1, 0, tzinfo=timezone.utc)
+    candidate, candidate_sha256 = _inputs(start)
+
+    result = verify_reconciliation_recovery(
+        candidate_payload=candidate,
+        dual_review_payload=_dual_review(candidate_sha256),
+        confirmation_payload=_console_response(candidate_sha256, confirmed_at=start + timedelta(minutes=3)),
+        current_receipt_payload={"evidence": _evidence(observed_at=start + timedelta(minutes=40)).to_dict()},
+        runtime_target_payload=_runtime_target(),
+        recovery_id="ibkr-soxl-live-recovery",
+        now=start + timedelta(minutes=40),
+    )
+
+    assert result["ready_for_atomic_state_transition"] is False
+    assert result["transition_plan"] is None
+    assert "reconciliation_recovery_confirmation_stale" in result["findings"]
+
+
+def test_fault_injection_controller_read_url_cannot_be_a_general_console_endpoint() -> None:
+    with pytest.raises(ValueError, match="controller read endpoint"):
+        read_console_confirmation(
+            confirmation_url="https://console.example/api/reconciliation-recovery",
+            recovery_id="ibkr-soxl-live-recovery",
+            token="controller-token",
+        )