fix(rectification): keep clock-stamped events out of window intercept (BUG-631, BUG-632)

Dated life events with clock times were swallowed as birth-window replies, and compare columns still stopped at the first nine candidates after the hour-window cap.

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

scripts/rectification/api_service.py

@@ -218,7 +218,7 @@ def score_candidates(request: RectificationRequest) -> dict[str, Any]:
     fingerprint = sha256({
         key: value
         for key, value in request.items()
-        if key not in {"asked_probe_keys", "dropped_asked_probe_keys"}
+        if key not in {"asked_probe_keys", "dropped_asked_probe_keys", "column_times"}
     })
     result_id = str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}"))
     candidate_decisions = build_candidate_decisions(

scripts/rectification/contracts.py

@@ -52,7 +52,7 @@ _EVENT_PROVENANCE_FIELDS = frozenset({
 })
 _REQUEST_FIELDS = frozenset({
     "birth_date", "start_time", "end_time", "lat", "lon", "tz", "events",
-    "ayanamsa", "node_mode", "asked_probe_keys", "minute_step", "blocks",
+    "ayanamsa", "node_mode", "asked_probe_keys", "column_times", "minute_step", "blocks",
 }) | _REQUEST_PROVENANCE_FIELDS
 ASKED_PROBE_KEY_MAX_LENGTH = 200
 _EVENT_FIELDS = frozenset({"id", "domain", "event_kind", "date_start", "date_end", "precision", "summary"}) | _EVENT_PROVENANCE_FIELDS
@@ -169,6 +169,7 @@ class RectificationRequest(TypedDict):
     local_time_status: NotRequired[str | None]
     asked_probe_keys: NotRequired[list[str]]
     dropped_asked_probe_keys: NotRequired[int]
+    column_times: NotRequired[list[str]]
     minute_step: NotRequired[int]
     blocks: NotRequired[list[dict[str, Any]]]
 
@@ -354,6 +355,20 @@ def normalize_rectification_request(body: Any, *, today: date | None = None) ->
         cleaned_request["asked_probe_keys"] = cleaned_keys
         if dropped:
             cleaned_request["dropped_asked_probe_keys"] = dropped
+    if "column_times" in body:
+        raw_times = body.get("column_times")
+        if not isinstance(raw_times, list) or not 1 <= len(raw_times) <= 64:
+            raise ValueError("column_times must contain between 1 and 64 HH:MM values")
+        cleaned_times: list[str] = []
+        seen_times: set[str] = set()
+        for index, item in enumerate(raw_times):
+            if not isinstance(item, str) or not _CLOCK.fullmatch(item):
+                raise ValueError(f"column_times[{index}] must be HH:MM")
+            if item in seen_times:
+                continue
+            seen_times.add(item)
+            cleaned_times.append(item)
+        cleaned_request["column_times"] = cleaned_times
     if "minute_step" in body:
         minute_step = body.get("minute_step")
         if isinstance(minute_step, bool) or not isinstance(minute_step, int) or not 1 <= minute_step <= 15:

scripts/rectification/decision_policy.py

@@ -51,6 +51,23 @@ def _decimal(value: Any, default: str = "0") -> Decimal:
         return Decimal(default)
 
 
+def _column_times_for_packet(
+    request: RectificationRequest,
+    candidate_decisions: Sequence[dict[str, Any]],
+) -> list[str]:
+    decision_times = [
+        str(item.get("time") or "")[:5]
+        for item in candidate_decisions
+        if str(item.get("time") or "")[:5]
+    ]
+    requested = request.get("column_times")
+    if not isinstance(requested, list) or not requested:
+        return decision_times
+    wanted = {str(item)[:5] for item in requested if isinstance(item, str)}
+    subset = [time for time in decision_times if time in wanted]
+    return subset or decision_times
+
+
 _AUDIT_LABELS = {
     "d1-rashi": ("D1 本命盘", "本轮已按该分钟重算本命宫位。"),
     "d2-hora": ("D2 财帛分盘", "本轮已对照财帛主题。"),
@@ -588,7 +605,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],
+        column_times=_column_times_for_packet(request, candidate_decisions),
     )
     if packet["dasha_agreement"]["status"] == "conflict":
         if overall_confidence == "high":

scripts/rectification/refinement_packet.py

@@ -9,8 +9,11 @@ from __future__ import annotations
 
 from typing import Any, Sequence
 
+from scripts.rectification.candidate_contrast import MAX_PUBLIC_CLUSTERS
 from scripts.rectification.house_table import PLANET_ZH, SIGN_LORDS, SIGNS, SIGNS_CN
 
+COLUMN_COMPARE_BUDGET_MS = 3000
+
 NAKSHATRA_SPAN = 40.0 / 3.0
 NAKSHATRA_BOUNDARY_DEGREES = 2.0
 MATCH_LABELS = {
@@ -263,12 +266,14 @@ def column_times_for_compare(
     candidate_times: Sequence[str],
     representative_time: str | None,
     *,
-    limit: int = 9,
+    limit: int = MAX_PUBLIC_CLUSTERS,
 ) -> list[str]:
     """Unique HH:MM keys for by_time ledgers/windows, capped at `limit`.
 
     Compare cards still project at most three posterior columns; this set is
-    the engine candidate list so those clocks can look up a row.
+    the full public candidate list (≤64) so inference-layer clocks can look up
+    a row. Callers that already know a smaller active set may pass it as
+    `column_times` when a 64-minute compare would exceed COLUMN_COMPARE_BUDGET_MS.
     """
     seen: list[str] = []
     for raw in candidate_times:
@@ -621,14 +626,22 @@ def build_refinement_packet(
     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)
+    minute_step = request.get("minute_step")
+    skip_column_compare = isinstance(minute_step, int) and not isinstance(minute_step, bool) and minute_step > 1
+    if skip_column_compare:
+        columns: list[str] = []
+        ledgers_by_time: dict[str, Any] = {}
+        windows_by_time: dict[str, Any] = {}
+        column_compare_ms = 0.0
+    else:
+        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: