feat: require dual review for reconciliation baselines

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/reconciliation_baseline_dual_review.md

@@ -0,0 +1,39 @@
+# 旧实盘基线：强制双 AI 审计
+
+本流程处理已经获授权运行、但由于迁移、配置漂移或运行异常而进入
+`RECONCILE_ONLY` 的旧实盘实例。它不是策略研究晋级流程，也不会修改订单、仓位或
+资金。
+
+## 输入边界
+
+平台适配器只能在只读 broker 会话中生成 QPK 的
+`broker_reconciliation_evidence.v1`。原始账户、现金、持仓、订单和成交明细必须留在
+受控环境；传入审计的只可以是 QPK 生成的
+`broker_reconciliation_baseline_candidate.v1` 中的摘要和时间窗。
+
+候选至少需要两次间隔采样且所有状态摘要一致。候选的内容地址
+`candidate_sha256` 是后续审计和控制面操作的唯一绑定值。
+
+## 审计门槛
+
+调用 `run_dual_review_pipeline.py --trigger reconciliation_baseline` 时，必须提供：
+
+- `strategy_profile`；
+- `reconciliation_candidate_sha256`；
+- 不包含敏感资料的候选摘要，例如来源收据数量和观察窗口。
+
+该触发器不会采用“主审置信度足够即可放行”的普通优化。它始终运行 Codex 主审与
+独立 GPT/Claude 复审；缺少候选绑定、任一审计不可用、结论分歧或拒绝都不能通过。
+输出中的 `evidence_binding_sha256` 必须与候选的 `candidate_sha256` 完全一致。
+
+## 私有控制面职责
+
+审计通过仍不等于恢复实盘。只有私有、受权限保护的控制面可以：
+
+1. 验证候选和每份来源收据来自受信任的运行时，且内容地址未变；
+2. 验证双审结果为通过，并绑定同一个候选摘要；
+3. 在同一 runtime target 上原子写入五个预期状态摘要并转换到 `ACTIVE_LKG`；
+4. 保存不可变审计记录、操作人/服务身份、时间和目标版本，以便回滚和复核。
+
+任一条件不满足，控制面必须保持 `RECONCILE_ONLY`。AIAuditBridge 不持有券商密钥，
+也不拥有下单或直接修改 Cloud Run runtime target 的权限。

scripts/run_dual_review_pipeline.py

@@ -163,7 +163,10 @@ def _exit_code(result: dict[str, Any]) -> int:
 
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description="Run Codex primary + dual API secondary review pipeline.")
-    parser.add_argument("--trigger", choices=("promotion", "hit_rate", "drift"))
+    parser.add_argument(
+        "--trigger",
+        choices=("promotion", "hit_rate", "drift", "reconciliation_baseline"),
+    )
     parser.add_argument("--strategy-profile")
     parser.add_argument("--context-json", default="{}", help="Inline JSON or file path for trigger context")
     parser.add_argument("--evidence-file", help="Evidence package path (promotion)")

service/dual_review.py

@@ -18,6 +18,7 @@
 VERDICT_INVALID = "invalid_review"
 
 DEFAULT_ESCALATION_THRESHOLD = 0.8
+MANDATORY_MULTI_REVIEW_TRIGGERS = frozenset({"reconciliation_baseline"})
 
 _PASS_VALUES = frozenset({"pass", "approve", "approved", "accept", "accepted"})
 _FAIL_VALUES = frozenset({"fail", "reject", "rejected", "deny", "denied", "block", "blocked"})
@@ -31,6 +32,14 @@ class DualReviewTrigger(str, Enum):
     PROMOTION = "promotion"
     HIT_RATE = "hit_rate"
     DRIFT = "drift"
+    RECONCILIATION_BASELINE = "reconciliation_baseline"
+
+
+def requires_mandatory_multi_review(trigger: DualReviewTrigger | str) -> bool:
+    """Return whether a decision may never be approved by one reviewer alone."""
+
+    value = trigger.value if isinstance(trigger, DualReviewTrigger) else str(trigger).strip().lower()
+    return value in MANDATORY_MULTI_REVIEW_TRIGGERS
 
 
 def _normalize_verdict(value: Any) -> str | None:

service/dual_review_orchestrator.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import re
 from dataclasses import dataclass, field
 from typing import Any, Callable
 
@@ -16,6 +17,7 @@
     compare_three_reviews,
     extract_confidence,
     extract_verdict,
+    requires_mandatory_multi_review,
     should_escalate,
 )
 from service.dual_review_secondary import dual_api_secondary_reviewer, is_dual_api_secondary
@@ -24,6 +26,7 @@
 from service.model_router import route_model
 
 SecondaryReviewer = Callable[["DualReviewRequest"], dict[str, Any]]
+_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
 
 
 @dataclass(frozen=True)
@@ -33,6 +36,7 @@ class DualReviewRequest:
     primary_review: dict[str, Any]
     context: dict[str, Any] = field(default_factory=dict)
     escalation_threshold: float = DEFAULT_ESCALATION_THRESHOLD
+    evidence_binding_sha256: str | None = None
 
 
 @dataclass
@@ -46,6 +50,7 @@ class DualReviewResult:
     outcome: str = ""
     reason: str = ""
     model_route: dict[str, str] = field(default_factory=dict)
+    evidence_binding_sha256: str | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -58,6 +63,7 @@ def to_dict(self) -> dict[str, Any]:
             "outcome": self.outcome,
             "reason": self.reason,
             "model_route": self.model_route,
+            "evidence_binding_sha256": self.evidence_binding_sha256,
         }
 
 
@@ -77,12 +83,19 @@ def build_request_from_payload(payload: dict[str, Any]) -> DualReviewRequest | N
     except (TypeError, ValueError):
         cutoff = DEFAULT_ESCALATION_THRESHOLD
     context = {k: v for k, v in payload.items() if k not in {"primary_review", "secondary_review"}}
+    evidence_binding_sha256 = None
+    if trigger is DualReviewTrigger.RECONCILIATION_BASELINE:
+        candidate = str(payload.get("reconciliation_candidate_sha256") or "").strip().lower()
+        if not _SHA256_PATTERN.fullmatch(candidate):
+            return None
+        evidence_binding_sha256 = candidate
     return DualReviewRequest(
         trigger=trigger,
         strategy_profile=strategy_profile,
         primary_review=primary,
         context=context,
         escalation_threshold=cutoff,
+        evidence_binding_sha256=evidence_binding_sha256,
     )
 
 
@@ -127,27 +140,32 @@ def orchestrate_dual_review(
         strategy_profile=request.strategy_profile,
         primary_review=request.primary_review,
         model_route=dict(route),
+        evidence_binding_sha256=request.evidence_binding_sha256,
     )
 
-    primary_confidence = extract_confidence(request.primary_review)
-    if primary_confidence is None:
+    if requires_mandatory_multi_review(request.trigger):
         result.escalated = True
-        result.reason = "primary confidence missing; escalating"
-    elif should_escalate(primary_confidence, threshold=request.escalation_threshold):
-        result.escalated = True
-        result.reason = f"primary confidence {primary_confidence:.2f} below {request.escalation_threshold:.2f}"
+        result.reason = "mandatory multi-review for reconciliation baseline enrollment"
     else:
-        primary_verdict = extract_verdict(request.primary_review)
-        if primary_verdict == VERDICT_PASS:
-            result.outcome = VERDICT_PASS
-            result.reason = "primary confidence sufficient; approved without secondary review"
-            return result
-        if primary_verdict == VERDICT_FAIL:
-            result.outcome = VERDICT_FAIL
-            result.reason = "primary confidence sufficient; rejected without secondary review"
-            return result
-        result.escalated = True
-        result.reason = "primary verdict unclear; escalating"
+        primary_confidence = extract_confidence(request.primary_review)
+        if primary_confidence is None:
+            result.escalated = True
+            result.reason = "primary confidence missing; escalating"
+        elif should_escalate(primary_confidence, threshold=request.escalation_threshold):
+            result.escalated = True
+            result.reason = f"primary confidence {primary_confidence:.2f} below {request.escalation_threshold:.2f}"
+        else:
+            primary_verdict = extract_verdict(request.primary_review)
+            if primary_verdict == VERDICT_PASS:
+                result.outcome = VERDICT_PASS
+                result.reason = "primary confidence sufficient; approved without secondary review"
+                return result
+            if primary_verdict == VERDICT_FAIL:
+                result.outcome = VERDICT_FAIL
+                result.reason = "primary confidence sufficient; rejected without secondary review"
+                return result
+            result.escalated = True
+            result.reason = "primary verdict unclear; escalating"
 
     secondary = reviewer(request)
     result.secondary_review = secondary

service/dual_review_primary.py

@@ -13,7 +13,7 @@
 from service.dual_review_secondary import parse_llm_review_output
 
 _PRIMARY_SYSTEM = (
-    "You are the primary Codex reviewer for quantitative strategy promotion and risk decisions. "
+    "You are the primary Codex reviewer for quantitative strategy promotion, risk, and recovery decisions. "
     "Respond with JSON only: "
     '{"verdict":"approve"|"reject","confidence":0.0-1.0,"summary":"..."}'
 )
@@ -37,6 +37,9 @@ def build_primary_prompt(
         "drift_sigma",
         "drift_score",
         "repository",
+        "reconciliation_candidate_sha256",
+        "source_evidence_count",
+        "observation_window_seconds",
     ):
         value = context.get(key)
         if value not in (None, ""):
@@ -65,7 +68,9 @@ def build_primary_prompt(
         except (OSError, json.JSONDecodeError):
             lines.append(f"evidence_file: {evidence_path}")
     lines.append(
-        "Provide an independent primary review for whether this strategy change should proceed."
+        "Provide an independent primary review. For reconciliation_baseline, approve only when the "
+        "candidate is a fresh, matching, read-only observation set; this decision still requires "
+        "independent secondary approval and never authorises orders."
     )
     return "\n".join(lines)
 

service/dual_review_secondary.py

@@ -80,7 +80,16 @@ def build_secondary_prompt(request: DualReviewRequest) -> str:
         f"Strategy profile: {request.strategy_profile}",
         f"Trigger: {request.trigger.value}",
     ]
-    for key in ("domain", "old_status", "new_status", "drift_sigma", "drift_score"):
+    for key in (
+        "domain",
+        "old_status",
+        "new_status",
+        "drift_sigma",
+        "drift_score",
+        "reconciliation_candidate_sha256",
+        "source_evidence_count",
+        "observation_window_seconds",
+    ):
         if key in context and context[key] not in (None, ""):
             lines.append(f"{key}: {context[key]}")
     monthly = context.get("monthly_hit_rates")
@@ -90,8 +99,9 @@ def build_secondary_prompt(request: DualReviewRequest) -> str:
     if evidence:
         lines.append(f"context: {evidence}")
     lines.append(
-        "Decide whether this strategy change should proceed. "
-        "Use approve only when evidence supports promotion or continued operation."
+        "Decide whether this strategy change should proceed. For reconciliation_baseline, approve "
+        "only when the candidate is fresh, read-only, internally consistent, and safely scoped; "
+        "approval never authorises orders or a state change by itself."
     )
     return "\n".join(lines)
 

tests/test_dual_review.py

@@ -14,6 +14,7 @@
     DualReviewTrigger,
     compare_reviews,
     compare_three_reviews,
+    requires_mandatory_multi_review,
     should_escalate,
 )
 
@@ -140,11 +141,16 @@ def test_trigger_values(self) -> None:
         self.assertEqual(DualReviewTrigger.PROMOTION.value, "promotion")
         self.assertEqual(DualReviewTrigger.HIT_RATE.value, "hit_rate")
         self.assertEqual(DualReviewTrigger.DRIFT.value, "drift")
+        self.assertEqual(DualReviewTrigger.RECONCILIATION_BASELINE.value, "reconciliation_baseline")
 
     def test_trigger_is_string_enum(self) -> None:
         self.assertIsInstance(DualReviewTrigger.PROMOTION, str)
         self.assertEqual(DualReviewTrigger("drift"), DualReviewTrigger.DRIFT)
 
+    def test_reconciliation_baseline_requires_multiple_reviewers(self) -> None:
+        self.assertTrue(requires_mandatory_multi_review(DualReviewTrigger.RECONCILIATION_BASELINE))
+        self.assertFalse(requires_mandatory_multi_review(DualReviewTrigger.PROMOTION))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_dual_review_orchestrator.py

@@ -107,6 +107,37 @@ def test_primary_high_confidence_fail(self) -> None:
         self.assertEqual(result.outcome, VERDICT_FAIL)
         self.assertFalse(result.escalated)
 
+    def test_reconciliation_baseline_forces_independent_review_and_binds_candidate(self) -> None:
+        candidate_sha256 = "a" * 64
+        result = orchestrate_from_payload(
+            {
+                "trigger": "reconciliation_baseline",
+                "strategy_profile": "soxl_soxx_trend_income",
+                "reconciliation_candidate_sha256": candidate_sha256,
+                "primary_review": {"verdict": "approve", "confidence": 0.99},
+            },
+            secondary_review={
+                "gpt": {"verdict": "approve", "confidence": 0.95},
+                "claude": {"verdict": "approve", "confidence": 0.94},
+            },
+        )
+
+        assert result is not None
+        self.assertTrue(result.escalated)
+        self.assertEqual(result.outcome, VERDICT_PASS)
+        self.assertEqual(result.evidence_binding_sha256, candidate_sha256)
+
+    def test_reconciliation_baseline_requires_candidate_binding(self) -> None:
+        request = build_request_from_payload(
+            {
+                "trigger": "reconciliation_baseline",
+                "strategy_profile": "soxl_soxx_trend_income",
+                "primary_review": {"verdict": "approve", "confidence": 0.99},
+            }
+        )
+
+        self.assertIsNone(request)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_dual_review_primary.py

@@ -125,7 +125,7 @@ def test_budget_error_is_unavailable(self, review) -> None:
         result = run_codex_primary_review(prompt="review")
         self.assertEqual(result["verdict"], VERDICT_UNAVAILABLE)
         expected_prompt = (
-            "You are the primary Codex reviewer for quantitative strategy promotion and risk decisions. "
+            "You are the primary Codex reviewer for quantitative strategy promotion, risk, and recovery decisions. "
             "Respond with JSON only: "
             '{"verdict":"approve"|"reject","confidence":0.0-1.0,"summary":"..."}'
             "\n\nreview"