Merge pull request #140 from QuantStrategyLab/codex/reconciliation-recovery-authority

fix: require human approval for broker recovery

Author: Pigbibi

docs/reconciliation_baseline_dual_review.md

@@ -28,12 +28,16 @@
 
 ## 私有控制面职责
 
-审计通过仍不等于恢复实盘。只有私有、受权限保护的控制面可以：
+审计通过仍不等于恢复实盘。恢复现有券商下单能力在自动化权限策略中属于高风险
+`broker_or_order_execution` 操作；审计输出会明确标记
+`requires_human_recovery_approval=true`。统一管理站点必须让操作者确认一次，且只有
+私有、受权限保护的控制面才可以：
 
 1. 验证候选和每份来源收据来自受信任的运行时，且内容地址未变；
 2. 验证双审结果为通过，并绑定同一个候选摘要；
 3. 在同一 runtime target 上原子写入五个预期状态摘要并转换到 `ACTIVE_LKG`；
 4. 保存不可变审计记录、操作人/服务身份、时间和目标版本，以便回滚和复核。
 
 任一条件不满足，控制面必须保持 `RECONCILE_ONLY`。AIAuditBridge 不持有券商密钥，
-也不拥有下单或直接修改 Cloud Run runtime target 的权限。
+也不拥有下单或直接修改 Cloud Run runtime target 的权限；双 AI 审计不能绕过这条
+人工确认边界。

scripts/run_dual_review_pipeline.py

@@ -9,6 +9,11 @@
 from pathlib import Path
 from typing import Any
 
+from service.automation_authority import (
+    CLASS_BROKER_OR_ORDER_EXECUTION,
+    evaluate_automation_authority,
+)
+from service.autonomy import ACTION_ORDER
 from service.dual_review import VERDICT_DISAGREEMENT, VERDICT_FAIL, VERDICT_UNAVAILABLE
 from service.dual_review_dispatch import dispatch_dual_review_result
 from service.dual_review_orchestrator import orchestrate_from_payload
@@ -135,6 +140,16 @@ def run_pipeline(
         return {"ok": False, "error": "orchestration_failed", "payload": payload}
 
     result = outcome.to_dict()
+    if trigger == "reconciliation_baseline":
+        # A matching, dual-reviewed candidate proves only that a legacy state
+        # is ready for a human recovery decision.  The authority policy treats
+        # restoring broker execution as high risk and must never auto-apply it.
+        result["recovery_authority"] = evaluate_automation_authority(
+            ["broker_reconciliation_baseline"],
+            trusted_metadata={"change_class": CLASS_BROKER_OR_ORDER_EXECUTION},
+            proposed_action=ACTION_ORDER,
+        )
+        result["requires_human_recovery_approval"] = True
     if outcome.outcome == VERDICT_UNAVAILABLE:
         result["skipped"] = ["reviewers_unavailable"]
         result["degraded"] = True

tests/test_dual_review_pipeline.py

@@ -85,6 +85,30 @@ def test_all_reviewers_unavailable_degrades_with_durable_alert(self, mock_orches
     def test_disagreement_is_a_hard_block(self) -> None:
         self.assertEqual(_exit_code({"ok": True, "outcome": "disagreement"}), 2)
 
+    @patch("scripts.run_dual_review_pipeline.orchestrate_from_payload")
+    def test_reconciliation_baseline_requires_human_recovery_approval(self, mock_orchestrate) -> None:
+        from service.dual_review import DualReviewTrigger
+        from service.dual_review_orchestrator import DualReviewResult
+
+        candidate_sha256 = "a" * 64
+        mock_orchestrate.return_value = DualReviewResult(
+            trigger=DualReviewTrigger.RECONCILIATION_BASELINE,
+            strategy_profile="soxl_soxx_trend_income",
+            primary_review={"verdict": "approve", "confidence": 0.95},
+            outcome="pass",
+            evidence_binding_sha256=candidate_sha256,
+        )
+        result = run_pipeline(
+            trigger="reconciliation_baseline",
+            strategy_profile="soxl_soxx_trend_income",
+            context={"reconciliation_candidate_sha256": candidate_sha256},
+            primary_review={"verdict": "approve", "confidence": 0.95},
+        )
+
+        self.assertTrue(result["requires_human_recovery_approval"])
+        self.assertTrue(result["recovery_authority"]["human_review_required"])
+        self.assertEqual(result["recovery_authority"]["final_action"], "escalate")
+
     @patch.dict("os.environ", {"DUAL_REVIEW_GATE_SKIP": "1"}, clear=False)
     def test_from_evidence_cli_without_trigger_flags(self) -> None:
         with tempfile.TemporaryDirectory() as tmp: