fix(rectification): stop unwritten-evidence claims and same-cluster dasha false conflicts (BUG-635–640)

Host only says 记下了 after a real write; Mastra schema rejections fail closed. Ledger year keys no longer drop quality probes, dual-dasha agreement is per cluster, width uses cluster span, and public house tables follow the inference minute.

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

scripts/rectification/decision_policy.py

@@ -156,8 +156,16 @@ def indistinguishable_width_minutes(candidates: Sequence[dict[str, Any]]) -> int
         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
+    starts: list[int] = []
+    ends: list[int] = []
+    for row in ranked:
+        start = _clock_minutes(row.get("cluster_start") or row.get("time"))
+        end = _clock_minutes(row.get("cluster_end") or row.get("time"))
+        if start is not None:
+            starts.append(start)
+        if end is not None:
+            ends.append(end)
+    span = (max(ends) - min(starts) + 1) if starts and ends else 0
     tied = int(top.get("tied_minute_count") or 1)
     return max(tied, span, 1)
 
@@ -606,6 +614,7 @@ def build_decision_receipt(
         cluster_width_minutes=width,
         include_discriminators=int(request.get("minute_step") or 1) <= 1,
         column_times=_column_times_for_packet(request, candidate_decisions),
+        clusters=candidate_decisions,
     )
     if packet["dasha_agreement"]["status"] == "conflict":
         if overall_confidence == "high":

scripts/rectification/refinement_packet.py

@@ -58,6 +58,29 @@ def _clock(value: str) -> int:
     return int(value[:2]) * 60 + int(value[3:5])
 
 
+def _cluster_span_key(time: str, clusters: Sequence[dict[str, Any]] | None) -> str:
+    clock = str(time)[:5]
+    if not clusters or len(clock) != 5 or clock[2] != ":":
+        return clock
+    clock_minutes = _clock(clock)
+    for row in clusters:
+        if not isinstance(row, dict):
+            continue
+        times = [str(item)[:5] for item in (row.get("cluster_times") or []) if str(item or "")[:5]]
+        start = str(row.get("cluster_start") or "")[:5]
+        end = str(row.get("cluster_end") or "")[:5]
+        representative = str(row.get("time") or "")[:5]
+        in_times = clock in times or clock == representative
+        in_span = False
+        if len(start) == 5 and start[2] == ":" and len(end) == 5 and end[2] == ":":
+            in_span = _clock(start) <= clock_minutes <= _clock(end)
+        if in_times or in_span:
+            if len(start) == 5 and len(end) == 5:
+                return f"{start}–{end}"
+            return clock
+    return clock
+
+
 def _feature_time(feature: dict[str, Any]) -> str | None:
     raw = feature.get("time")
     if isinstance(raw, str) and len(raw) >= 5:
@@ -380,7 +403,11 @@ def event_fit_rate(rows: Sequence[dict[str, Any]]) -> dict[str, Any]:
     }
 
 
-def dasha_agreement(built: dict[str, Any], candidate_times: Sequence[str]) -> dict[str, Any]:
+def dasha_agreement(
+    built: dict[str, Any],
+    candidate_times: Sequence[str],
+    clusters: Sequence[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
     times = [str(item)[:5] for item in candidate_times if isinstance(item, str) and len(str(item)) >= 5]
     if not times:
         return {
@@ -413,7 +440,7 @@ def dasha_agreement(built: dict[str, Any], candidate_times: Sequence[str]) -> di
         }
     vim_top = max(times, key=lambda time: (vim_scores[time], -_clock(time)))
     narayana_top = max(times, key=lambda time: (narayana_scores[time], -_clock(time)))
-    if vim_top == narayana_top:
+    if vim_top == narayana_top or _cluster_span_key(vim_top, clusters) == _cluster_span_key(narayana_top, clusters):
         return {
             "status": "agree",
             "vimshottari_top": vim_top,
@@ -618,6 +645,7 @@ def build_refinement_packet(
     cluster_width_minutes: int | None = None,
     include_discriminators: bool = True,
     column_times: Sequence[str] | None = None,
+    clusters: Sequence[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     from time import perf_counter
 
@@ -642,7 +670,7 @@ def build_refinement_packet(
         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)
+    agreement = dasha_agreement(built, candidate_times, clusters=clusters)
     stage = precision_stage(cluster, len(request.get("events") or []))
     if not include_discriminators:
         return {