Jesse_Chen
814c924e4a
Keep askable cards after exhaustion, explain each probe, read the adopted credible range in reports and chat, and compare declared periods before the minute grid when the clock is unknown. Co-authored-by: Cursor <cursoragent@cursor.com>
779 lines
31 KiB
Python
779 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from decimal import ROUND_FLOOR, ROUND_HALF_UP, Decimal
|
|
from typing import Any
|
|
from uuid import NAMESPACE_URL, uuid5
|
|
|
|
from scripts.active_rectification_events import CandidateScoreRow
|
|
from scripts.rectification.contracts import (
|
|
EVENT_CONTRACT_VERSION,
|
|
RectificationRequest,
|
|
is_primary_scoreable_event,
|
|
is_scoreable_event,
|
|
)
|
|
from scripts.rectification.house_table import compact_house_table_from_contexts
|
|
from scripts.rectification.horary_observation import build_horary_observation
|
|
from scripts.rectification.refinement_packet import build_refinement_packet
|
|
from scripts.rectification.candidate_contrast import context_time, select_signature_representatives
|
|
from scripts.rectification.case_holdout import holdout_event_ids
|
|
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-v3"
|
|
RECEIPT_VERSION = "candidate-decision-receipt-v2"
|
|
EXECUTION_LEDGER_VERSION = "rectification-execution-ledger-v2"
|
|
SCORE_QUANTUM = Decimal("0.0001")
|
|
TIE_ABSOLUTE_TOLERANCE = Decimal("0.0001")
|
|
MIN_ACCEPTANCE_EVENTS = 3
|
|
MIN_ACCEPTANCE_DOMAINS = 2
|
|
MIN_DATE_QUALITY_MEAN = Decimal("0.65")
|
|
MIN_DIAGNOSTIC_RETENTION = Decimal("0.75")
|
|
MIN_ACCEPTANCE_MARGIN_PERCENT = Decimal("10")
|
|
POLICY_SKIPPED_LAYERS = frozenset({"KP_cusps"})
|
|
|
|
_BAD_DATE_RELIABILITY = frozenset({"low", "uncertain", "unreliable"})
|
|
_CLEAR_DATE_CONFLICT = frozenset({"", "none", "resolved", "no_conflict"})
|
|
|
|
|
|
def _decimal(value: Any, default: str = "0") -> Decimal:
|
|
if isinstance(value, bool):
|
|
return Decimal(default)
|
|
try:
|
|
return Decimal(str(value))
|
|
except Exception:
|
|
return Decimal(default)
|
|
|
|
|
|
_AUDIT_LABELS = {
|
|
"d1-rashi": ("D1 本命盘", "本轮已按该分钟重算本命宫位。"),
|
|
"d2-hora": ("D2 财帛分盘", "本轮已对照财帛主题。"),
|
|
"d4-chaturthamsha": ("D4 迁移分盘", "本轮已对照居所或迁移。"),
|
|
"d5-panchamsha": ("D5 成就分盘", "本轮已对照学业或被委以责任的变化。"),
|
|
"d3-drekkana": ("D3 兄弟分盘", "本轮已对照兄弟姐妹主题。"),
|
|
"d7-saptamsha": ("D7 子女分盘", "本轮已对照子女或伴侣细节。"),
|
|
"d9-navamsa": ("D9 婚姻分盘", "本轮已对照关系主题，可用 D9 上升类型表作校时方法，不是命运承诺。"),
|
|
"d10-dashamsa": ("D10 事业分盘", "本轮已对照事业主题，可用 D10 上升类型表作校时方法，不是命运承诺。"),
|
|
"d11-labhamsha": ("D11 收益分盘", "本轮已对照收益主题。"),
|
|
"d12-dwadashamsha": ("D12 父母分盘", "本轮已对照家人主题。"),
|
|
"d24-chaturvimshamsha": ("D24 教育分盘", "本轮已对照学业主题。"),
|
|
"d30-trimshamsha": ("D30 健康压力分盘", "本轮已对照健康压力主题。"),
|
|
"vimshottari-dasha": ("Vimshottari", "本轮已对照主限。"),
|
|
"narayana-dasha": ("Narayana", "本轮已对照分盘大运。"),
|
|
"gochara": ("Gochara", "本轮已做受控行运辅助对照。"),
|
|
"ashtakavarga": ("Ashtakavarga", "本轮已做 Ashtakavarga 辅助对照。"),
|
|
"shadbala": ("Shadbala", "本轮已做已核验的 Shadbala 分量辅助对照。"),
|
|
"arudha-pada": ("Arudha Pada", "本轮已做 Arudha 辅助对照。"),
|
|
"functional-benefic-malefic": ("功能吉凶星", "本轮已叠加本命功能吉凶星。"),
|
|
"dasha-transition-proximity": ("换运贴近度", "本轮已对照日级事件与候选换运日期的贴近程度。"),
|
|
}
|
|
|
|
|
|
def natal_recast_copy(time: str, lagna: str) -> dict[str, Any]:
|
|
return {
|
|
"time": time[:5],
|
|
"lagna": lagna,
|
|
"user_meaning": (
|
|
f"本命宫位已按 {time[:5]} 重算（上升 {lagna}）。"
|
|
"下面是本轮实际执行的技法，不能当作唯一分钟确认。"
|
|
),
|
|
"unique_minute_claim": False,
|
|
"confirmation_allowed": False,
|
|
}
|
|
|
|
|
|
def _executed_public_methods(built: dict[str, Any]) -> list[str]:
|
|
methods: set[str] = set()
|
|
for contributions in (built.get("matrix") or {}).values():
|
|
if not isinstance(contributions, dict):
|
|
continue
|
|
for cell in contributions.values():
|
|
if not isinstance(cell, dict):
|
|
continue
|
|
for layer in cell.get("technique_layers") or []:
|
|
if layer in _AUDIT_LABELS:
|
|
methods.add(str(layer))
|
|
for rule in cell.get("rule_ids") or []:
|
|
text = str(rule)
|
|
if "transition_proximity" in text:
|
|
methods.add("dasha-transition-proximity")
|
|
elif text.startswith("vim_"):
|
|
methods.add("vimshottari-dasha")
|
|
elif text.startswith("narayana_"):
|
|
methods.add("narayana-dasha")
|
|
elif "functional_benefic" in text or "functional_malefic" in text:
|
|
methods.add("functional-benefic-malefic")
|
|
elif text.startswith("gochara") or "controlled_transit" in text:
|
|
methods.add("gochara")
|
|
elif "ashtakavarga" in text:
|
|
methods.add("ashtakavarga")
|
|
elif "shadbala" in text:
|
|
methods.add("shadbala")
|
|
elif "arudha" in text:
|
|
methods.add("arudha-pada")
|
|
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["unique_minute_path"] = (
|
|
"awaiting_user_consent" if confirmation_allowed else "closed_at_representative"
|
|
)
|
|
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:
|
|
executed.add("d1-rashi")
|
|
rows: list[dict[str, str]] = []
|
|
for method in _AUDIT_LABELS:
|
|
if method not in executed:
|
|
continue
|
|
label, note = _AUDIT_LABELS[method]
|
|
rows.append({"technique": label, "status": "executed", "note": note})
|
|
rows.append(_kp_audit_row(built))
|
|
rows.extend((
|
|
vedastro_audit_row(vedastro_status),
|
|
unique_minute_audit_row(confirmation_allowed),
|
|
))
|
|
return rows
|
|
|
|
|
|
def _kp_audit_row(built: dict[str, Any]) -> dict[str, str]:
|
|
executed = False
|
|
for context in built.get("static_contexts") or []:
|
|
if not isinstance(context, dict):
|
|
continue
|
|
feature = context.get("feature")
|
|
snapshot = feature.get("kp_cusps") if isinstance(feature, dict) else None
|
|
if isinstance(snapshot, dict) and snapshot.get("status") == "executed":
|
|
executed = True
|
|
break
|
|
if executed:
|
|
return {
|
|
"technique": "KP 宫头",
|
|
"status": "executed",
|
|
"note": "已按 Swiss Ephemeris Placidus + Krishnamurti 观察 12 宫头；不计分，不参与提出门或确认门。",
|
|
}
|
|
return {
|
|
"technique": "KP 宫头",
|
|
"status": "blocked",
|
|
"note": "Swiss Ephemeris Placidus 宫头无法计算或尚未执行。KP 观察不计分，不挡提出门。",
|
|
}
|
|
|
|
|
|
def _quantized_score(row: CandidateScoreRow) -> Decimal:
|
|
return _decimal(row.get("score")).quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)
|
|
|
|
|
|
def _distribute_percent(weights: Sequence[Decimal]) -> list[int]:
|
|
if not weights:
|
|
return []
|
|
total = sum(weights, Decimal(0))
|
|
if total == 0:
|
|
base, remainder = divmod(100, len(weights))
|
|
return [base + (1 if index < remainder else 0) for index in range(len(weights))]
|
|
exact = [weight * Decimal(100) / total for weight in weights]
|
|
floors = [int(value.to_integral_value(rounding=ROUND_FLOOR)) for value in exact]
|
|
remaining = 100 - sum(floors)
|
|
order = sorted(
|
|
range(len(weights)),
|
|
key=lambda index: (-(exact[index] - Decimal(floors[index])), index),
|
|
)
|
|
for index in order[:remaining]:
|
|
floors[index] += 1
|
|
return floors
|
|
|
|
|
|
def _relative_support_proportional(scores: Sequence[Decimal]) -> list[int]:
|
|
return _distribute_percent([max(score, Decimal(0)) for score in scores])
|
|
|
|
|
|
def _relative_support_offset(scores: Sequence[Decimal], floor: Decimal) -> list[int]:
|
|
return _distribute_percent([max(score - floor, Decimal(0)) for score in scores])
|
|
|
|
|
|
def _relative_support_softmax(scores: Sequence[Decimal], temperature: Decimal) -> list[int]:
|
|
from math import exp
|
|
if not scores:
|
|
return []
|
|
peak = max(scores)
|
|
temp = temperature if temperature > 0 else Decimal("0.5")
|
|
weights = [Decimal(str(exp(float((score - peak) / temp)))) for score in scores]
|
|
return _distribute_percent(weights)
|
|
|
|
|
|
RELATIVE_SUPPORT_MODE = "proportional"
|
|
RELATIVE_SUPPORT_TEMPERATURE = Decimal("0.5")
|
|
|
|
|
|
def _relative_support(
|
|
scores: Sequence[Decimal],
|
|
*,
|
|
floor: Decimal | None = None,
|
|
mode: str | None = None,
|
|
temperature: Decimal | None = None,
|
|
) -> list[int]:
|
|
if not scores:
|
|
return []
|
|
selected = mode or RELATIVE_SUPPORT_MODE
|
|
if selected == "softmax":
|
|
return _relative_support_softmax(
|
|
scores,
|
|
temperature if temperature is not None else RELATIVE_SUPPORT_TEMPERATURE,
|
|
)
|
|
if selected == "offset" and floor is not None:
|
|
return _relative_support_offset(scores, floor)
|
|
return _relative_support_proportional(scores)
|
|
|
|
|
|
def build_candidate_decisions(
|
|
rows: Sequence[CandidateScoreRow],
|
|
*,
|
|
result_id: str,
|
|
static_contexts: Sequence[dict[str, Any]] | None = None,
|
|
support_mode: str | None = None,
|
|
temperature: Decimal | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"]))
|
|
public_rows = select_signature_representatives(ranked, static_contexts)
|
|
if not public_rows:
|
|
return []
|
|
public_scores = [_quantized_score(row) for row in public_rows]
|
|
all_scores = [_quantized_score(row) for row in ranked]
|
|
floor = min(all_scores) if all_scores else Decimal(0)
|
|
supports = _relative_support(
|
|
public_scores,
|
|
floor=floor,
|
|
mode=support_mode,
|
|
temperature=temperature,
|
|
)
|
|
decisions = []
|
|
for index, row in enumerate(public_rows):
|
|
score = _quantized_score(row)
|
|
tied_minute_count = sum(
|
|
abs(score - other) <= TIE_ABSOLUTE_TOLERANCE
|
|
for other in all_scores
|
|
)
|
|
decisions.append({
|
|
"candidate_id": str(uuid5(NAMESPACE_URL, f"{POLICY_VERSION}:{result_id}:{row['time']}")),
|
|
"rank": index + 1,
|
|
"time": row["time"],
|
|
"relative_support": supports[index],
|
|
"tied_minute_count": tied_minute_count,
|
|
})
|
|
return decisions
|
|
|
|
|
|
def _gate(passed: bool, **details: Any) -> dict[str, Any]:
|
|
return {"passed": passed, **details}
|
|
|
|
|
|
def _actionable_missing_layers(layers: Any) -> list[str]:
|
|
return sorted({
|
|
str(layer)
|
|
for layer in (layers or [])
|
|
if str(layer) not in POLICY_SKIPPED_LAYERS
|
|
})
|
|
|
|
|
|
def _date_quality(events: Sequence[dict[str, Any]]) -> dict[str, Any]:
|
|
weights = [_decimal(precision_weight(str(event["precision"]))) for event in events]
|
|
total = sum(weights, Decimal(0))
|
|
mean = total / Decimal(len(weights)) if weights else Decimal(0)
|
|
low_reliability = sorted(
|
|
event["id"]
|
|
for event in events
|
|
if str(event.get("date_reliability") or "").strip().lower() in _BAD_DATE_RELIABILITY
|
|
)
|
|
unresolved_conflicts = sorted(
|
|
event["id"]
|
|
for event in events
|
|
if str(event.get("date_conflict_status") or "").strip().lower() not in _CLEAR_DATE_CONFLICT
|
|
)
|
|
passed = bool(events) and mean >= MIN_DATE_QUALITY_MEAN and not low_reliability and not unresolved_conflicts
|
|
return _gate(
|
|
passed,
|
|
precision_weight_total=float(total),
|
|
precision_weight_mean=float(mean.quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)),
|
|
minimum_precision_weight_mean=float(MIN_DATE_QUALITY_MEAN),
|
|
low_reliability_event_ids=low_reliability,
|
|
unresolved_conflict_event_ids=unresolved_conflicts,
|
|
)
|
|
|
|
|
|
def _diagnostic_quality(diagnostics: dict[str, Any]) -> dict[str, Any]:
|
|
retention_names = (
|
|
"leave_one_event_out_retention_rate",
|
|
"leave_one_domain_out_retention_rate",
|
|
"date_sensitivity_retention_rate",
|
|
)
|
|
retentions = {name: _decimal(diagnostics.get(name)) for name in retention_names}
|
|
margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
|
|
passed = (
|
|
all(value >= MIN_DIAGNOSTIC_RETENTION for value in retentions.values())
|
|
and margin >= MIN_ACCEPTANCE_MARGIN_PERCENT
|
|
)
|
|
return _gate(
|
|
passed,
|
|
minimum_retention=float(MIN_DIAGNOSTIC_RETENTION),
|
|
minimum_margin_percent=float(MIN_ACCEPTANCE_MARGIN_PERCENT),
|
|
margin_percent=float(margin),
|
|
**{name: float(value) for name, value in retentions.items()},
|
|
)
|
|
|
|
|
|
def build_decision_receipt(
|
|
request: RectificationRequest,
|
|
candidate_decisions: Sequence[dict[str, Any]],
|
|
built: dict[str, Any],
|
|
diagnostics: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
scoreable_events = [event for event in request["events"] if is_primary_scoreable_event(event)]
|
|
holdout = holdout_event_ids(request["events"])
|
|
training_events = [
|
|
event for event in scoreable_events
|
|
if str(event.get("id") or "") not in holdout
|
|
]
|
|
domains = sorted({event["domain"] for event in training_events})
|
|
candidate_presence = _gate(bool(candidate_decisions), candidate_count=len(candidate_decisions))
|
|
event_quality = _gate(
|
|
len(training_events) >= MIN_ACCEPTANCE_EVENTS,
|
|
scoreable_event_count=len(training_events),
|
|
minimum=MIN_ACCEPTANCE_EVENTS,
|
|
)
|
|
domain_diversity = _gate(
|
|
len(domains) >= MIN_ACCEPTANCE_DOMAINS,
|
|
scoreable_domain_count=len(domains),
|
|
minimum=MIN_ACCEPTANCE_DOMAINS,
|
|
domains=domains,
|
|
)
|
|
date_quality = _date_quality(training_events)
|
|
top_tied_count = candidate_decisions[0]["tied_minute_count"] if candidate_decisions else 0
|
|
unique_top = _gate(top_tied_count == 1, tied_minute_count=top_tied_count)
|
|
diagnostic_quality = _diagnostic_quality(diagnostics)
|
|
skipped_layers = sorted(
|
|
str(layer)
|
|
for layer in (built.get("missing_layers") or [])
|
|
if str(layer) in POLICY_SKIPPED_LAYERS
|
|
)
|
|
actionable_missing_layers = _actionable_missing_layers(built.get("missing_layers"))
|
|
required_layers = _gate(
|
|
not actionable_missing_layers,
|
|
missing_layers=actionable_missing_layers,
|
|
skipped_by_policy=skipped_layers,
|
|
)
|
|
|
|
# Adoption is the session result when a representative time exists.
|
|
# Unique-top and diagnostic stability still block confirmation, not accept.
|
|
acceptance_allowed = all((
|
|
candidate_presence["passed"],
|
|
event_quality["passed"],
|
|
domain_diversity["passed"],
|
|
date_quality["passed"],
|
|
))
|
|
margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
|
|
if acceptance_allowed and margin >= Decimal("20"):
|
|
overall_confidence = "high"
|
|
elif acceptance_allowed:
|
|
overall_confidence = "medium"
|
|
else:
|
|
overall_confidence = "low"
|
|
|
|
acceptance_reasons: list[str] = []
|
|
for passed, reason in (
|
|
(candidate_presence["passed"], "no_candidates"),
|
|
(event_quality["passed"], "insufficient_events"),
|
|
(domain_diversity["passed"], "insufficient_domain_diversity"),
|
|
(date_quality["passed"], "low_date_quality"),
|
|
):
|
|
if not passed:
|
|
acceptance_reasons.append(reason)
|
|
confirmation_reasons = []
|
|
if not unique_top["passed"]:
|
|
confirmation_reasons.append("tied_top_score")
|
|
if not diagnostic_quality["passed"]:
|
|
confirmation_reasons.append("insufficient_diagnostic_stability")
|
|
if not required_layers["passed"]:
|
|
confirmation_reasons.append("missing_mandatory_layers")
|
|
reasons = [*acceptance_reasons, *confirmation_reasons]
|
|
|
|
representative = candidate_decisions[0] if candidate_decisions else None
|
|
width = indistinguishable_width_minutes(candidate_decisions)
|
|
grid_times = [str(item)[:5] for item in (built.get("candidate_times") or []) if str(item or "")[:5]]
|
|
if not grid_times:
|
|
for context in built.get("static_contexts") or []:
|
|
if not isinstance(context, dict):
|
|
continue
|
|
time = context_time(context)
|
|
if time and time not in grid_times:
|
|
grid_times.append(time)
|
|
if not grid_times:
|
|
grid_times = [item["time"] for item in candidate_decisions]
|
|
packet = build_refinement_packet(
|
|
request,
|
|
built,
|
|
representative_time=representative["time"] if representative else None,
|
|
candidate_times=grid_times,
|
|
cluster_width_minutes=width,
|
|
include_discriminators=int(request.get("minute_step") or 1) <= 1,
|
|
)
|
|
if packet["dasha_agreement"]["status"] == "conflict":
|
|
if overall_confidence == "high":
|
|
overall_confidence = "medium"
|
|
elif overall_confidence == "medium":
|
|
overall_confidence = "low"
|
|
reasons.append("vimshottari_narayana_conflict")
|
|
confirmation_reasons.append("vimshottari_narayana_conflict")
|
|
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")
|
|
fit_high = packet["event_fit_rate"].get("band") == "high"
|
|
propose_allowed = bool(
|
|
acceptance_allowed
|
|
and required_layers["passed"]
|
|
and (
|
|
fit_high
|
|
or all((
|
|
diagnostic_quality["passed"],
|
|
confirmation_event_quality,
|
|
confirmation_domain_quality,
|
|
))
|
|
)
|
|
)
|
|
engine_granted = all((
|
|
propose_allowed,
|
|
unique_top["passed"],
|
|
adjacent_passed,
|
|
confirmation_margin,
|
|
not dasha_conflict,
|
|
))
|
|
if not engine_granted:
|
|
confirmation_reasons.append("engine_exact_confirmation_not_granted")
|
|
exact_confirmation = {
|
|
"passed": False,
|
|
"fail_closed": True,
|
|
"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",
|
|
"unique_minute_path": "closed_at_representative",
|
|
"required_scoreable_events": MIN_CONFIRMATION_EVENTS,
|
|
"required_scoreable_domains": MIN_CONFIRMATION_DOMAINS,
|
|
"reason": "engine_and_external_validation_must_explicitly_pass",
|
|
}
|
|
receipt = {
|
|
"receipt_version": RECEIPT_VERSION,
|
|
"contract_version": "v2",
|
|
"event_contract_version": EVENT_CONTRACT_VERSION,
|
|
"policy_version": POLICY_VERSION,
|
|
"decision_policy_version": POLICY_VERSION,
|
|
"display_allowed": bool(candidate_decisions),
|
|
"selection_allowed": acceptance_allowed,
|
|
"acceptance_allowed": acceptance_allowed,
|
|
"propose_allowed": propose_allowed,
|
|
"confirmation_allowed": False,
|
|
"accept_allowed": acceptance_allowed,
|
|
"confirm_allowed": False,
|
|
"representative_candidate_id": representative["candidate_id"] if representative else None,
|
|
"representative_time": representative["time"] if representative else None,
|
|
"overall_confidence": overall_confidence,
|
|
"margin_percent": float(margin),
|
|
"reasons": reasons,
|
|
"acceptance_reasons": acceptance_reasons,
|
|
"confirmation_reasons": confirmation_reasons,
|
|
"tie_policy": {
|
|
"score_quantum": float(SCORE_QUANTUM),
|
|
"absolute_tolerance": float(TIE_ABSOLUTE_TOLERANCE),
|
|
"rounding": "ROUND_HALF_UP",
|
|
},
|
|
"gates": {
|
|
"candidate_presence": candidate_presence,
|
|
"event_quality": event_quality,
|
|
"domain_diversity": domain_diversity,
|
|
"date_quality": date_quality,
|
|
"unique_top": unique_top,
|
|
"diagnostic_quality": diagnostic_quality,
|
|
"required_layers": required_layers,
|
|
"exact_confirmation": exact_confirmation,
|
|
},
|
|
}
|
|
house_tables_by_time: dict[str, dict[str, Any]] = {}
|
|
house_times: list[str] = []
|
|
seen_house_times: set[str] = set()
|
|
for context in built.get("static_contexts") or []:
|
|
if not isinstance(context, dict):
|
|
continue
|
|
time = context_time(context)
|
|
if time and time not in seen_house_times:
|
|
seen_house_times.add(time)
|
|
house_times.append(time)
|
|
for decision in candidate_decisions:
|
|
time = str(decision.get("time") or "")[:5]
|
|
if time and time not in seen_house_times:
|
|
seen_house_times.add(time)
|
|
house_times.append(time)
|
|
for time in house_times:
|
|
table = compact_house_table_from_contexts(built.get("static_contexts"), time)
|
|
if table:
|
|
house_tables_by_time[table["time"]] = table
|
|
house_table = house_tables_by_time.get(representative["time"] if representative else "") or compact_house_table_from_contexts(
|
|
built.get("static_contexts"),
|
|
representative["time"] if representative else None,
|
|
)
|
|
if house_table:
|
|
receipt["house_table"] = house_table
|
|
recast = natal_recast_copy(house_table["time"], house_table["lagna"])
|
|
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,
|
|
vedastro_status="not_evaluated",
|
|
confirmation_allowed=False,
|
|
)
|
|
receipt.update({
|
|
"window_scan": packet["window_scan"],
|
|
"event_dasha_ledger": packet["event_dasha_ledger"],
|
|
"event_fit_rate": packet["event_fit_rate"],
|
|
"dasha_agreement": packet["dasha_agreement"],
|
|
"lagna_contrast": packet["lagna_contrast"],
|
|
"nakshatra_boundary": packet["nakshatra_boundary"],
|
|
"precision_stage": packet["precision_stage"],
|
|
"oos_blind_prompts": packet["oos_blind_prompts"],
|
|
"discriminating_event_probes": packet.get("discriminating_event_probes") or [],
|
|
"event_clarification_probes": packet.get("event_clarification_probes") or [],
|
|
"evidence_collection_probes": packet.get("evidence_collection_probes") or [],
|
|
"candidate_contrast_opportunities": packet.get("candidate_contrast_opportunities") or [],
|
|
"holdout_validation_probes": packet.get("holdout_validation_probes") or [],
|
|
"dropped_probes": packet.get("dropped_probes") or [],
|
|
"prospective_probes": packet.get("prospective_probes") or [],
|
|
"horary_observation": build_horary_observation(request),
|
|
"unique_minute_claim": False,
|
|
})
|
|
return apply_confirmation_decision(receipt)
|
|
|
|
|
|
def build_execution_ledger(
|
|
request: RectificationRequest,
|
|
built: dict[str, Any],
|
|
diagnostics: dict[str, Any],
|
|
candidate_decisions: Sequence[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
matrix = built.get("matrix") or {}
|
|
date_sensitivity = {
|
|
item.get("event_id"): item
|
|
for item in built.get("date_sensitivity") or []
|
|
if isinstance(item, dict)
|
|
}
|
|
entries: list[dict[str, Any]] = []
|
|
all_layers: set[str] = set()
|
|
for event in request["events"]:
|
|
candidates = matrix.get(event["id"], {})
|
|
layers = sorted({
|
|
layer
|
|
for contribution in candidates.values()
|
|
for layer in contribution.get("technique_layers", [])
|
|
})
|
|
all_layers.update(layers)
|
|
sensitivity = date_sensitivity.get(event["id"], {})
|
|
scoreable = is_scoreable_event(event)
|
|
entries.append({
|
|
"ledger_version": EXECUTION_LEDGER_VERSION,
|
|
"stage": "event_scoring",
|
|
"status": "executed" if scoreable and candidates else "not_executed" if scoreable else "retained_not_scored",
|
|
"source": "python-engine",
|
|
"event_id": event["id"],
|
|
"domain": event["domain"],
|
|
"event_kind": event["event_kind"],
|
|
"date_precision": event["precision"],
|
|
"precision_weight": precision_weight(event["precision"]),
|
|
"sampled_date_count": len(sensitivity.get("sample_dates") or []),
|
|
"candidate_count": len(candidates),
|
|
"technique_layers": layers,
|
|
})
|
|
for layer in sorted(all_layers):
|
|
entries.append({
|
|
"ledger_version": EXECUTION_LEDGER_VERSION,
|
|
"stage": "technique_layer",
|
|
"method": layer,
|
|
"status": "executed",
|
|
"source": "python-engine",
|
|
})
|
|
entries.extend((
|
|
{
|
|
"ledger_version": EXECUTION_LEDGER_VERSION,
|
|
"stage": "candidate_ranking",
|
|
"status": "executed" if candidate_decisions else "not_executed",
|
|
"source": "python-decision-policy",
|
|
"candidate_count": len(candidate_decisions),
|
|
},
|
|
{
|
|
"ledger_version": EXECUTION_LEDGER_VERSION,
|
|
"stage": "diagnostics",
|
|
"status": "executed" if diagnostics else "not_executed",
|
|
"source": "python-engine",
|
|
"metrics": sorted(diagnostics),
|
|
},
|
|
))
|
|
return entries
|