fix(rectification): refresh remaining probes and targeted collect before delivering range (BUG-653/654)

Dated-choice exhaustion is not convergence. Refresh probes from remaining
active candidates, then ask a targeted collect, then deliver. Skill 10.0.24.

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", "column_times"}
+        if key not in {"asked_probe_keys", "dropped_asked_probe_keys", "column_times", "refresh_probes"}
     })
     result_id = str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}"))
     candidate_decisions = build_candidate_decisions(

scripts/rectification/contracts.py

@@ -53,6 +53,7 @@ _EVENT_PROVENANCE_FIELDS = frozenset({
 _REQUEST_FIELDS = frozenset({
     "birth_date", "start_time", "end_time", "lat", "lon", "tz", "events",
     "ayanamsa", "node_mode", "asked_probe_keys", "column_times", "minute_step", "blocks",
+    "refresh_probes",
 }) | _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
@@ -170,6 +171,7 @@ class RectificationRequest(TypedDict):
     asked_probe_keys: NotRequired[list[str]]
     dropped_asked_probe_keys: NotRequired[int]
     column_times: NotRequired[list[str]]
+    refresh_probes: NotRequired[bool]
     minute_step: NotRequired[int]
     blocks: NotRequired[list[dict[str, Any]]]
 
@@ -369,6 +371,11 @@ def normalize_rectification_request(body: Any, *, today: date | None = None) ->
             seen_times.add(item)
             cleaned_times.append(item)
         cleaned_request["column_times"] = cleaned_times
+    if "refresh_probes" in body:
+        if body.get("refresh_probes") is not True and body.get("refresh_probes") is not False:
+            raise ValueError("refresh_probes must be a boolean")
+        if body.get("refresh_probes") is True:
+            cleaned_request["refresh_probes"] = True
     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/event_probes.py

@@ -49,6 +49,10 @@ from scripts.rectification.refinement_packet import match_level
 # three domains keep two years plus a couple of activation fallbacks without
 # flooding the ask layer, which still ranks globally by information_gain.
 MAX_PROBES = 8
+# Refresh against a small remaining candidate set may keep a few more years.
+REFRESH_MAX_PROBES = 12
+REFRESH_MAX_PROBES_PER_DOMAIN = 4
+REFRESH_REMAINING_CAP = 5
 # Collection asks an age-band cue for every missing catalog domain.
 MAX_COLLECTION_PROBES = 7
 # N: keep the top scored probes per domain (boundary years, plus at most one
@@ -331,6 +335,21 @@ def _differing_layers(contexts: Sequence[dict[str, Any]]) -> set[str]:
     return {layer for layer, bucket in values.items() if len(bucket) > 1}
 
 
+def _probe_caps(*, refresh: bool, remaining_count: int) -> tuple[int, int]:
+    if refresh and remaining_count <= REFRESH_REMAINING_CAP:
+        return REFRESH_MAX_PROBES, REFRESH_MAX_PROBES_PER_DOMAIN
+    return MAX_PROBES, MAX_PROBES_PER_DOMAIN
+
+
+def _monthly_family_dasha_boundary(probe: dict[str, Any]) -> bool:
+    if str(probe.get("source") or "") != "dasha_boundary":
+        return False
+    if str(probe.get("domain") or "") != "family":
+        return False
+    month = probe.get("month")
+    return isinstance(month, int) and 1 <= month <= 12
+
+
 def _probe_domains(
     remaining_layers: set[str],
     _events: Sequence[dict[str, Any]],
@@ -855,6 +874,8 @@ def _answer_priors_for(probe: dict[str, Any]) -> dict[str, float]:
     if kind not in {"existence", "event_quality"}:
         kind = "existence"
     domain = str(probe.get("domain") or "")
+    if _monthly_family_dasha_boundary(probe):
+        return dict(ANSWER_PRIORS[("family", "existence")])
     if domain == "family" and kind == "existence" and not _broad_existence_window(probe):
         return dict(_DEFAULT_EXISTENCE_PRIORS)
     priors = ANSWER_PRIORS.get((domain, kind))
@@ -886,6 +907,8 @@ def _dominant_existence_prior(probe: dict[str, Any], priors: dict[str, float]) -
         return False
     if str(probe.get("source") or "") == "known_event_quality":
         return False
+    if _monthly_family_dasha_boundary(probe):
+        return False
     return max(priors.values()) > DOMINANT_ANSWER_PRIOR
 
 
@@ -1509,19 +1532,30 @@ def _discriminating_event_probe_lists(
     full = _static_contexts(built)
     if len(full) < 2:
         return empty
-    clusters = cluster_contexts_by_signature(full)
-    if len(clusters) < 2:
-        remaining = _remaining_contexts(built, candidate_times) or full
-        clusters = cluster_contexts_by_signature(remaining)
+    refresh = request.get("refresh_probes") is True
+    remaining = _remaining_contexts(built, candidate_times) if candidate_times else []
+    if refresh and remaining:
+        work = remaining
+        clusters = cluster_contexts_by_signature(work)
+    else:
+        work = full
+        clusters = cluster_contexts_by_signature(full)
+        if len(clusters) < 2:
+            work = remaining or full
+            clusters = cluster_contexts_by_signature(work)
     if len(clusters) < 2:
         return empty
     reps = [cluster["representative"] for cluster in clusters if _scoreable(cluster["representative"])]
     if len(reps) < 2:
-        reps = [item for item in full if _scoreable(item)]
+        reps = [item for item in work if _scoreable(item)]
     if len(reps) < 2:
         return empty
+    max_probes, max_per_domain = _probe_caps(
+        refresh=refresh,
+        remaining_count=len(remaining) if refresh else len(full),
+    )
     set_version = candidate_set_version([cluster["times"] for cluster in clusters])
-    remaining_layers = _differing_layers(full)
+    remaining_layers = _differing_layers(work if refresh else full)
     if not remaining_layers:
         remaining_layers = {
             layer for layer in SCORING_LAYERS
@@ -1572,8 +1606,8 @@ def _discriminating_event_probe_lists(
             found.append(row)
             if evaluated > sample_size:
                 break
-        kept = _best_probe_per_year(found)[:MAX_PROBES_PER_DOMAIN]
-        if len(kept) < MAX_PROBES_PER_DOMAIN:
+        kept = _best_probe_per_year(found)[:max_per_domain]
+        if len(kept) < max_per_domain:
             activation = _try_activation_probe(
                 reps=reps,
                 birth_date=birth_date,
@@ -1608,13 +1642,15 @@ def _discriminating_event_probe_lists(
     ))
     _annotate_nearby_ledger(probes, events)
     probes.sort(key=_probe_sort_key)
-    public, dropped = _partition_ranked_probes(probes)
+    public, dropped = _partition_ranked_probes(probes, max_probes=max_probes)
     assert_distinguish_contract(public)
     return public, dropped
 
 
 def _partition_ranked_probes(
     probes: Sequence[dict[str, Any]],
+    *,
+    max_probes: int = MAX_PROBES,
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
     public: list[dict[str, Any]] = []
     dropped: list[dict[str, Any]] = []
@@ -1651,7 +1687,7 @@ def _partition_ranked_probes(
             continue
         seen.add(key)
         public.append(ranked)
-        if len(public) >= MAX_PROBES:
+        if len(public) >= max_probes:
             break
     public.sort(key=_probe_sort_key)
     return public, dropped