feat(rectification): expose candidate result reports
Independent Staging Quality Gate / validate (push) Successful in 16m46s
Independent Staging Quality Gate / publish (push) Successful in 19m37s

This commit is contained in:
Jesse_Chen
2026-08-31 05:51:30 +08:00
parent 6cbf1f22e2
commit e8bd3a5208
6 changed files with 366 additions and 5 deletions
+54 -1
View File
@@ -108,6 +108,57 @@ class CandidateResult(TypedDict):
stability_diagnostics: dict[str, Any]
missing_layers: list[str]
candidate_ranking_summary: NotRequired[list[dict[str, Any]]]
candidate_summary: NotRequired[dict[str, Any]]
def build_candidate_result_summary(result: dict[str, Any]) -> dict[str, Any]:
"""Project a candidate result into stable, non-confirmatory next steps."""
supported: dict[str, dict[str, Any]] = {}
unconfirmed: dict[str, dict[str, Any]] = {}
contradictory: dict[str, dict[str, Any]] = {}
for item in result.get("evidence", []):
if not isinstance(item, dict):
continue
domain = str(item.get("domain") or "unknown")
points = float(item.get("points") or 0)
bucket = supported if points > 0 else contradictory if points < 0 else unconfirmed
row = bucket.setdefault(domain, {"domain": domain, "event_count": 0, "total_points": 0.0, "rule_ids": set()})
row["event_count"] += 1
row["total_points"] += points
row["rule_ids"].update(str(rule_id) for rule_id in item.get("rule_ids", []))
def rows(bucket: dict[str, dict[str, Any]], *, include_points: bool) -> list[dict[str, Any]]:
output = []
for row in bucket.values():
item: dict[str, Any] = {
"domain": row["domain"],
"event_count": row["event_count"],
"rule_ids": sorted(row["rule_ids"]),
}
if include_points:
item["total_points"] = round(row["total_points"], 3)
output.append(item)
return sorted(output, key=lambda item: (-item.get("total_points", 0), item["domain"]))
segment = result.get("winning_segment")
next_steps: list[str] = []
if int(result.get("event_count") or 0) < 5:
next_steps.append("collect_at_least_five_events")
if float(result.get("margin_percent") or 0) <= 0 or "tied_leader" in (result.get("reasons") or []):
next_steps.append("resolve_candidate_tie_or_narrow_window")
if isinstance(segment, dict) and int(segment.get("width_minutes") or 0) > 5:
next_steps.append("narrow_window_before_minute_claim")
next_steps.append("do_not_apply_as_birth_time_truth")
return {
"claim_status": "candidate_range_not_birth_time_truth",
"candidate_range": segment,
"supporting_evidence": rows(supported, include_points=True),
"unconfirmed_evidence": rows(unconfirmed, include_points=False),
"contradictory_evidence": rows(contradictory, include_points=True),
"reasons": list(result.get("reasons") or []),
"next_step_codes": next_steps,
}
def precision_weight(precision: EventPrecision) -> float:
@@ -347,4 +398,6 @@ def score_life_events(request: RectificationEventRequest) -> CandidateResult:
"""Compute actual candidate rows, then apply the versioned confidence gates."""
from scripts.active_rectification_event_engine import compute_event_candidate_result
return compute_event_candidate_result(request)
result = compute_event_candidate_result(request)
result["candidate_summary"] = build_candidate_result_summary(result)
return result