fix(rectification): wire unique-minute confirmation to VedAstro and sealed holdout

Keep confirmation fail-closed until the public AA set is ready, and rewrite Technique Audit from the attached VedAstro status instead of a hardcoded blocked row.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/rectification/decision_policy.py

@@ -15,6 +15,13 @@ from scripts.rectification.contracts import (
 from scripts.rectification.house_table import compact_house_table_from_contexts
 from scripts.rectification.refinement_packet import build_refinement_packet
 from scripts.rectification.scoring_service import precision_weight
+from scripts.rectification.sealed_holdout import holdout_passed, load_sealed_minute_holdout
+from scripts.rectification_policy import (
+    MAX_CONFIRMATION_WIDTH_MINUTES,
+    MIN_CONFIRMATION_DOMAINS,
+    MIN_CONFIRMATION_EVENTS,
+    MIN_CONFIRMATION_MARGIN_PERCENT,
+)
 
 POLICY_VERSION = "rectification-candidate-policy-v2"
 RECEIPT_VERSION = "candidate-decision-receipt-v2"
@@ -105,10 +112,151 @@ def _executed_public_methods(built: dict[str, Any]) -> list[str]:
     return [key for key in _AUDIT_LABELS if key in methods]
 
 
+def _clock_minutes(value: Any) -> int | None:
+    text = str(value or "")[:5]
+    if len(text) != 5 or text[2] != ":":
+        return None
+    try:
+        hour = int(text[:2])
+        minute = int(text[3:])
+    except ValueError:
+        return None
+    if hour > 23 or minute > 59:
+        return None
+    return hour * 60 + minute
+
+
+def indistinguishable_width_minutes(candidates: Sequence[dict[str, Any]]) -> int:
+    if not candidates:
+        return 0
+    ranked = sorted(candidates, key=lambda row: int(row.get("rank") or 0))
+    top = ranked[0]
+    minutes = [value for value in (_clock_minutes(row.get("time")) for row in ranked) if value is not None]
+    span = (max(minutes) - min(minutes) + 1) if minutes else 0
+    tied = int(top.get("tied_minute_count") or 1)
+    return max(tied, span, 1)
+
+
+def vedastro_audit_row(status: str) -> dict[str, str]:
+    if status == "passed":
+        return {
+            "technique": "VedAstro 分钟级校验",
+            "status": "executed",
+            "note": "官方分钟敏感校验已通过。仍不能单独确认唯一分钟。",
+        }
+    if status == "failed":
+        return {
+            "technique": "VedAstro 分钟级校验",
+            "status": "blocked",
+            "note": "官方分钟敏感校验未能区分相邻分钟，不能写确认。",
+        }
+    return {
+        "technique": "VedAstro 分钟级校验",
+        "status": "blocked",
+        "note": "官方分钟敏感校验尚未跑通。未调用不等于失败，但缺这一层不能写确认。",
+    }
+
+
+def unique_minute_audit_row(allowed: bool) -> dict[str, str]:
+    if allowed:
+        return {
+            "technique": "唯一分钟确认",
+            "status": "executed",
+            "note": "确认门已允许。只有用户明确同意才能写已确认校正时间。",
+        }
+    return {
+        "technique": "唯一分钟确认",
+        "status": "blocked",
+        "note": "采用不等于确认唯一分钟。",
+    }
+
+
+def rewrite_confirmation_audit_rows(receipt: dict[str, Any]) -> None:
+    rows = receipt.get("technique_audit_table")
+    if not isinstance(rows, list):
+        return
+    exact = (receipt.get("gates") or {}).get("exact_confirmation") or {}
+    vedastro_status = str(exact.get("external_validation_status") or "not_evaluated")
+    allowed = receipt.get("confirmation_allowed") is True
+    replacements = {
+        "VedAstro 分钟级校验": vedastro_audit_row(vedastro_status),
+        "唯一分钟确认": unique_minute_audit_row(allowed),
+    }
+    receipt["technique_audit_table"] = [
+        replacements.get(str(row.get("technique")), row) if isinstance(row, dict) else row
+        for row in rows
+    ]
+
+
+def apply_confirmation_decision(receipt: dict[str, Any]) -> dict[str, Any]:
+    gates = receipt.setdefault("gates", {})
+    exact = gates.setdefault("exact_confirmation", {})
+    vedastro_status = str(exact.get("external_validation_status") or "not_evaluated")
+    if vedastro_status not in {"passed", "failed", "not_evaluated"}:
+        vedastro_status = "not_evaluated"
+        exact["external_validation_status"] = vedastro_status
+    holdout = load_sealed_minute_holdout()
+    engine_granted = exact.get("engine_granted") is True
+    adjacent_passed = exact.get("adjacent_passed") is True
+    vedastro_ok = vedastro_status == "passed"
+    holdout_ok = holdout_passed(holdout)
+    confirmation_allowed = bool(engine_granted and adjacent_passed and vedastro_ok and holdout_ok)
+    reasons = [str(item) for item in (receipt.get("confirmation_reasons") or []) if str(item)]
+    for flag, reason in (
+        (engine_granted, "engine_exact_confirmation_not_granted"),
+        (vedastro_ok, "external_validation_not_passed"),
+        (adjacent_passed, "adjacent_minutes_indistinguishable"),
+        (holdout_ok, "public_aa_holdout_not_ready"),
+    ):
+        if flag:
+            reasons = [item for item in reasons if item != reason]
+        elif reason not in reasons:
+            reasons.append(reason)
+    receipt["confirmation_allowed"] = confirmation_allowed
+    receipt["confirm_allowed"] = confirmation_allowed
+    receipt["unique_minute_claim"] = False
+    exact["passed"] = confirmation_allowed
+    exact["fail_closed"] = True
+    exact["external_validation_status"] = vedastro_status
+    exact["holdout"] = holdout
+    exact["reason"] = (
+        "confirmation_allowed"
+        if confirmation_allowed
+        else "engine_and_external_validation_must_explicitly_pass"
+    )
+    receipt["confirmation_reasons"] = reasons
+    receipt["reasons"] = [
+        *list(receipt.get("acceptance_reasons") or []),
+        *reasons,
+    ]
+    natal = receipt.get("natal_recast")
+    if isinstance(natal, dict):
+        natal["confirmation_allowed"] = False
+        natal["unique_minute_claim"] = False
+    rewrite_confirmation_audit_rows(receipt)
+    return receipt
+
+
+def apply_vedastro_minute_sensitive_to_receipt(
+    receipt: dict[str, Any],
+    status: str,
+    *,
+    summary: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    exact = receipt.setdefault("gates", {}).setdefault("exact_confirmation", {})
+    normalized = status if status in {"passed", "failed", "not_evaluated"} else "not_evaluated"
+    exact["external_validation_status"] = normalized
+    if summary is not None:
+        exact["vedastro_minute_sensitive"] = summary
+    return apply_confirmation_decision(receipt)
+
+
 def build_technique_audit(
     built: dict[str, Any],
     *,
     house_table: dict[str, Any] | None,
+    vedastro_status: str = "not_evaluated",
+    confirmation_allowed: bool = False,
 ) -> list[dict[str, str]]:
     executed = set(_executed_public_methods(built))
     if house_table:
@@ -125,16 +273,8 @@ def build_technique_audit(
         "note": "KP 宫头本轮未计算。",
     })
     rows.extend((
-        {
-            "technique": "VedAstro 分钟级校验",
-            "status": "blocked",
-            "note": "官方分钟级校验尚未评估。",
-        },
-        {
-            "technique": "唯一分钟确认",
-            "status": "blocked",
-            "note": "采用不等于确认唯一分钟。",
-        },
+        vedastro_audit_row(vedastro_status),
+        unique_minute_audit_row(confirmation_allowed),
     ))
     return rows
 
@@ -301,7 +441,6 @@ def build_decision_receipt(
         confirmation_reasons.append("insufficient_diagnostic_stability")
     if not required_layers["passed"]:
         confirmation_reasons.append("missing_mandatory_layers")
-    confirmation_reasons.extend(("engine_exact_confirmation_not_granted", "external_validation_not_passed"))
     reasons = [*acceptance_reasons, *confirmation_reasons]
 
     representative = candidate_decisions[0] if candidate_decisions else None
@@ -318,13 +457,43 @@ def build_decision_receipt(
             overall_confidence = "low"
         reasons.append("vimshottari_narayana_conflict")
         confirmation_reasons.append("vimshottari_narayana_conflict")
+    width = indistinguishable_width_minutes(candidate_decisions)
+    adjacent_passed = unique_top["passed"] and width <= MAX_CONFIRMATION_WIDTH_MINUTES
+    confirmation_event_quality = len(scoreable_events) >= MIN_CONFIRMATION_EVENTS
+    confirmation_domain_quality = len(domains) >= MIN_CONFIRMATION_DOMAINS
+    confirmation_margin = margin >= Decimal(MIN_CONFIRMATION_MARGIN_PERCENT)
+    dasha_conflict = packet["dasha_agreement"]["status"] == "conflict"
+    if not confirmation_event_quality:
+        confirmation_reasons.append("insufficient_confirmation_events")
+    if not confirmation_domain_quality:
+        confirmation_reasons.append("insufficient_confirmation_domains")
+    if not confirmation_margin:
+        confirmation_reasons.append("insufficient_confirmation_margin")
+    if not adjacent_passed:
+        confirmation_reasons.append("adjacent_minutes_indistinguishable")
+    engine_granted = all((
+        acceptance_allowed,
+        unique_top["passed"],
+        diagnostic_quality["passed"],
+        required_layers["passed"],
+        confirmation_event_quality,
+        confirmation_domain_quality,
+        confirmation_margin,
+        adjacent_passed,
+        not dasha_conflict,
+    ))
+    if not engine_granted:
+        confirmation_reasons.append("engine_exact_confirmation_not_granted")
     exact_confirmation = {
         "passed": False,
         "fail_closed": True,
-        "engine_granted": False,
+        "engine_granted": engine_granted,
+        "adjacent_passed": adjacent_passed,
+        "indistinguishable_width_minutes": width,
+        "max_confirmation_width_minutes": MAX_CONFIRMATION_WIDTH_MINUTES,
         "external_validation_status": "not_evaluated",
-        "required_scoreable_events": 4,
-        "required_scoreable_domains": 3,
+        "required_scoreable_events": MIN_CONFIRMATION_EVENTS,
+        "required_scoreable_domains": MIN_CONFIRMATION_DOMAINS,
         "reason": "engine_and_external_validation_must_explicitly_pass",
     }
     receipt = {
@@ -377,7 +546,12 @@ def build_decision_receipt(
         receipt["natal_recast"] = recast
     if house_tables_by_time:
         receipt["house_tables_by_time"] = house_tables_by_time
-    receipt["technique_audit_table"] = build_technique_audit(built, house_table=house_table)
+    receipt["technique_audit_table"] = build_technique_audit(
+        built,
+        house_table=house_table,
+        vedastro_status="not_evaluated",
+        confirmation_allowed=False,
+    )
     receipt.update({
         "window_scan": packet["window_scan"],
         "event_dasha_ledger": packet["event_dasha_ledger"],
@@ -388,7 +562,7 @@ def build_decision_receipt(
         "oos_blind_prompts": packet["oos_blind_prompts"],
         "unique_minute_claim": False,
     })
-    return receipt
+    return apply_confirmation_decision(receipt)
 
 
 def build_execution_ledger(

scripts/rectification/sealed_holdout.py

@@ -0,0 +1,42 @@
+"""Product-facing sealed public AA holdout contract.
+
+This module does not invent cases. A human updates
+`references/rectification_sealed_holdout.v1.json` after a valid passing
+evaluation of a frozen public AA set. LOEO/LODO are not a holdout.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Final
+
+ROOT = Path(__file__).resolve().parents[2]
+HOLDOUT_PATH: Final = ROOT / "references" / "rectification_sealed_holdout.v1.json"
+PILOT_REPORT_PATH: Final = (
+    ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v2_pilot_report.json"
+)
+
+
+def load_sealed_minute_holdout() -> dict[str, Any]:
+    data = json.loads(HOLDOUT_PATH.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("sealed holdout contract must be an object")
+    status = str(data.get("status") or "not_ready").strip() or "not_ready"
+    return {
+        "sealed_benchmark_id": str(data.get("sealed_benchmark_id") or "").strip(),
+        "status": status if status == "ready" else "not_ready",
+        "valid_public_aa_cases": int(data.get("valid_public_aa_cases") or 0),
+        "required_cases": int(data.get("required_cases") or 20),
+        "top_1_rate": float(data.get("top_1_rate") or 0),
+        "confirmation_coverage_rate": float(data.get("confirmation_coverage_rate") or 0),
+    }
+
+
+def holdout_passed(holdout: dict[str, Any] | None = None) -> bool:
+    row = holdout or load_sealed_minute_holdout()
+    return (
+        row["status"] == "ready"
+        and int(row["valid_public_aa_cases"]) >= int(row["required_cases"])
+        and float(row["confirmation_coverage_rate"]) > 0
+    )