Jesse_Chen
a88467ffa8
Keep unique-top and width on confirmation only, and stop lagna-frame follow-ups from blocking cards on an already-scored cluster. Co-authored-by: Cursor <cursoragent@cursor.com>
681 lines
27 KiB
Python
681 lines
27 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.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"
|
|
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": ("功能吉凶星", "本轮已叠加本命功能吉凶星。"),
|
|
}
|
|
|
|
|
|
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 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["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 _relative_support(scores: Sequence[Decimal]) -> list[int]:
|
|
if not scores:
|
|
return []
|
|
weights = [max(score, Decimal(0)) for score in scores]
|
|
total = sum(weights, Decimal(0))
|
|
if total == 0:
|
|
base, remainder = divmod(100, len(scores))
|
|
return [base + (1 if index < remainder else 0) for index in range(len(scores))]
|
|
|
|
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(scores)),
|
|
key=lambda index: (-(exact[index] - Decimal(floors[index])), index),
|
|
)
|
|
for index in order[:remaining]:
|
|
floors[index] += 1
|
|
return floors
|
|
|
|
|
|
def build_candidate_decisions(
|
|
rows: Sequence[CandidateScoreRow],
|
|
*,
|
|
result_id: str,
|
|
) -> list[dict[str, Any]]:
|
|
ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"]))
|
|
public_rows = ranked[:3]
|
|
supports = _relative_support([_quantized_score(row) for row in public_rows])
|
|
all_scores = [_quantized_score(row) for row in ranked]
|
|
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)]
|
|
domains = sorted({event["domain"] for event in scoreable_events})
|
|
candidate_presence = _gate(bool(candidate_decisions), candidate_count=len(candidate_decisions))
|
|
event_quality = _gate(
|
|
len(scoreable_events) >= MIN_ACCEPTANCE_EVENTS,
|
|
scoreable_event_count=len(scoreable_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(scoreable_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)
|
|
packet = build_refinement_packet(
|
|
request,
|
|
built,
|
|
representative_time=representative["time"] if representative else None,
|
|
candidate_times=[item["time"] for item in candidate_decisions],
|
|
cluster_width_minutes=width,
|
|
)
|
|
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",
|
|
"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]] = {}
|
|
for decision in candidate_decisions:
|
|
table = compact_house_table_from_contexts(built.get("static_contexts"), decision.get("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"],
|
|
"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
|