fix(rectification): three-column candidate compare card (BUG-597, BUG-598)

Replace the minute-row delivery card with up to three compare columns so users can pick the time that fits, and apply the adult-year floor on inspect fallbacks.

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

scripts/rectification/decision_policy.py

@@ -564,6 +564,7 @@ def build_decision_receipt(
         candidate_times=grid_times,
         cluster_width_minutes=width,
         include_discriminators=int(request.get("minute_step") or 1) <= 1,
+        column_times=[item["time"] for item in candidate_decisions],
     )
     if packet["dasha_agreement"]["status"] == "conflict":
         if overall_confidence == "high":
@@ -694,6 +695,9 @@ def build_decision_receipt(
     receipt.update({
         "window_scan": packet["window_scan"],
         "event_dasha_ledger": packet["event_dasha_ledger"],
+        "event_dasha_ledger_by_time": packet.get("event_dasha_ledger_by_time") or {},
+        "prospective_windows_by_time": packet.get("prospective_windows_by_time") or {},
+        "column_compare_ms": packet.get("column_compare_ms"),
         "event_fit_rate": packet["event_fit_rate"],
         "dasha_agreement": packet["dasha_agreement"],
         "lagna_contrast": packet["lagna_contrast"],

scripts/rectification/event_probes.py

@@ -1709,6 +1709,74 @@ def discriminating_event_probes(
     return probes
 
 
+def prospective_windows_for_time(
+    request: dict[str, Any],
+    built: dict[str, Any],
+    time: str,
+    *,
+    today: date | None = None,
+) -> list[dict[str, Any]]:
+    """1–2 event windows in the next 12 months for one candidate minute."""
+    birth_date = str(request.get("birth_date") or "").strip()
+    clock = str(time or "")[:5]
+    if not birth_date or len(clock) < 5:
+        return []
+    now = today or date.today()
+    horizon = now + timedelta(days=366)
+    by_time = {_context_time(item): item for item in _static_contexts(built)}
+    context = by_time.get(clock)
+    if not isinstance(context, dict):
+        return []
+    moon = (context.get("planet_longitudes") or {}).get("Moon")
+    if not isinstance(moon, (int, float)):
+        return []
+    try:
+        starts = [
+            item
+            for item in _vim_start_dates(
+                birth_date,
+                float(moon),
+                now.year,
+                horizon.year,
+                include_pratyantar=True,
+            )
+            if now <= item <= horizon
+        ]
+    except (KeyError, TypeError, ValueError):
+        return []
+    rows: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for start in sorted(set(starts)):
+        key = f"{start.year:04d}-{start.month:02d}"
+        if key in seen:
+            continue
+        seen.add(key)
+        rows.append({
+            "domain": "career",
+            "from": key,
+            "to": key,
+        })
+        if len(rows) >= 2:
+            break
+    return rows
+
+
+def prospective_windows_by_time(
+    request: dict[str, Any],
+    built: dict[str, Any],
+    times: Sequence[str],
+    *,
+    today: date | None = None,
+) -> dict[str, list[dict[str, Any]]]:
+    out: dict[str, list[dict[str, Any]]] = {}
+    for raw in times:
+        clock = str(raw or "")[:5]
+        if len(clock) < 5 or clock in out:
+            continue
+        out[clock] = prospective_windows_for_time(request, built, clock, today=today)
+    return out
+
+
 def prospective_event_windows(
     request: dict[str, Any],
     built: dict[str, Any],

scripts/rectification/refinement_packet.py

@@ -240,6 +240,49 @@ def window_scan(
     return payload
 
 
+def _clock_key(value: Any) -> str | None:
+    raw = str(value or "")[:5]
+    return raw if len(raw) == 5 and raw[2] == ":" else None
+
+
+def _event_year_month(event: dict[str, Any]) -> str | None:
+    for key in ("date_start", "date", "occurred_from"):
+        raw = str(event.get(key) or "")
+        if len(raw) >= 7 and raw[4] == "-":
+            return raw[:7]
+    year = event.get("year")
+    month = event.get("month")
+    if isinstance(year, int) and 1900 <= year <= 2100:
+        if isinstance(month, int) and 1 <= month <= 12:
+            return f"{year:04d}-{month:02d}"
+        return f"{year:04d}"
+    return None
+
+
+def column_times_for_compare(
+    candidate_times: Sequence[str],
+    representative_time: str | None,
+    *,
+    limit: int = 3,
+) -> list[str]:
+    """At most `limit` unique HH:MM keys, ranked as given (engine score order)."""
+    seen: list[str] = []
+    for raw in candidate_times:
+        clock = _clock_key(raw)
+        if not clock or clock in seen:
+            continue
+        seen.append(clock)
+        if len(seen) >= limit:
+            break
+    representative = _clock_key(representative_time)
+    if representative and representative not in seen:
+        if len(seen) < limit:
+            seen.insert(0, representative)
+        elif seen:
+            seen[-1] = representative
+    return seen
+
+
 def event_dasha_ledger(
     request: dict[str, Any],
     built: dict[str, Any],
@@ -264,6 +307,8 @@ def event_dasha_ledger(
             for track in tracks
         ) or "现有大运层"
         gochara_hit = _has_gochara(rule_ids)
+        domain = str(event.get("domain") or "").strip()
+        year_month = _event_year_month(event)
         rows.append({
             "summary": summary[:80],
             "match": level,
@@ -274,10 +319,24 @@ def event_dasha_ledger(
                 f"{summary[:40]}：{MATCH_LABELS[level]}（{track_text}"
                 f"{'；Gochara 激活相关宫' if gochara_hit else '；Gochara 未见对应'}）"
             ),
+            **({"domain": domain} if domain else {}),
+            **({"year_month": year_month} if year_month else {}),
         })
     return rows
 
 
+def event_dasha_ledgers_by_time(
+    request: dict[str, Any],
+    built: dict[str, Any],
+    times: Sequence[str],
+) -> dict[str, list[dict[str, Any]]]:
+    return {
+        clock: event_dasha_ledger(request, built, clock)
+        for raw in times
+        if (clock := _clock_key(raw))
+    }
+
+
 def event_fit_rate(rows: Sequence[dict[str, Any]]) -> dict[str, Any]:
     total = len(rows)
     matched = sum(1 for row in rows if row.get("match") in {"strong", "medium"})
@@ -549,16 +608,32 @@ def build_refinement_packet(
     candidate_times: Sequence[str],
     cluster_width_minutes: int | None = None,
     include_discriminators: bool = True,
+    column_times: Sequence[str] | None = None,
 ) -> dict[str, Any]:
+    from time import perf_counter
+
+    from scripts.rectification.event_probes import prospective_windows_by_time
+
     scan = window_scan(built)
     cluster = cluster_scan(built, candidate_times, representative_time, cluster_width_minutes)
     ledger = event_dasha_ledger(request, built, representative_time)
+    columns = column_times_for_compare(
+        column_times if column_times is not None else candidate_times,
+        representative_time,
+    )
+    compare_started = perf_counter()
+    ledgers_by_time = event_dasha_ledgers_by_time(request, built, columns)
+    windows_by_time = prospective_windows_by_time(request, built, columns)
+    column_compare_ms = round((perf_counter() - compare_started) * 1000, 1)
     agreement = dasha_agreement(built, candidate_times)
     stage = precision_stage(cluster, len(request.get("events") or []))
     if not include_discriminators:
         return {
             "window_scan": scan,
             "event_dasha_ledger": ledger,
+            "event_dasha_ledger_by_time": ledgers_by_time,
+            "prospective_windows_by_time": windows_by_time,
+            "column_compare_ms": column_compare_ms,
             "event_fit_rate": event_fit_rate(ledger),
             "dasha_agreement": agreement,
             "lagna_contrast": lagna_contrast(built),
@@ -634,6 +709,9 @@ def build_refinement_packet(
     return {
         "window_scan": scan,
         "event_dasha_ledger": ledger,
+        "event_dasha_ledger_by_time": ledgers_by_time,
+        "prospective_windows_by_time": windows_by_time,
+        "column_compare_ms": column_compare_ms,
         "event_fit_rate": event_fit_rate(ledger),
         "dasha_agreement": agreement,
         "lagna_contrast": lagna_contrast(built),