feat(rectification): expand dated dasha_boundary probe supply

Distinguish-stage reverse-verify questions were exhausting after 1–3
dated probes. Union boundary windows across representative pairs, keep
multiple years per domain, raise the public cap, and allow activation
fallback without relaxing MIN_BOUNDARY_DAYS or scoring.

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

tests/test_rectification_event_probes.py

@@ -5,6 +5,10 @@ from datetime import date, datetime
 
 from scripts.rectification.candidate_contrast import distinguish_contract_errors
 from scripts.rectification.event_probes import (
+    MAX_BOUNDARY_CANDIDATES_PER_DOMAIN,
+    MAX_COLLECTION_PROBES,
+    MAX_PROBES,
+    MAX_PROBES_PER_DOMAIN,
     _agent_brief,
     discriminating_event_probes,
     event_clarification_probes,
@@ -134,6 +138,25 @@ def _probes(request: dict, built: dict, times: list[str], representative: str, *
     )
 
 
+def _multi_layer_window_built() -> dict:
+    """Two signature clusters, moons far enough apart that Vimshottari windows exist.
+
+    Before supply expansion this fixture published exactly three probes
+    (career.2024.02 / relationship.2024.02 / relocation.2015.03).
+    """
+    early = dict(d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=1, d10_asc=1, d12_asc=1, d24_asc=1)
+    late = dict(d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0, d9_asc=2, d10_asc=2, d12_asc=2, d24_asc=2)
+    return {
+        "static_contexts": [
+            _context("04:47", **early),
+            _context("04:48", **late),
+            _context("05:00", **early),
+            _context("05:06", **late),
+            _context("05:07", **late),
+        ]
+    }
+
+
 class EventProbesTest(unittest.TestCase):
     def test_agent_brief_locks_meaning_without_forcing_template_copy(self) -> None:
         brief = _agent_brief(
@@ -247,6 +270,7 @@ class EventProbesTest(unittest.TestCase):
         self.assertEqual(probes, [])
         collection = evidence_collection_probes(request, today=date(2026, 8, 22))
         self.assertTrue(collection)
+        self.assertEqual(len(collection), MAX_COLLECTION_PROBES)
         self.assertTrue(all(item["phase"] == "evidence_collection" for item in collection))
         self.assertTrue(all(item["role"] == "collect" for item in collection))
         self.assertTrue(all(item["source"] == "age_band" for item in collection))
@@ -404,7 +428,7 @@ class EventProbesTest(unittest.TestCase):
         probes = _probes(request, built, ["05:13", "05:40"], "05:13", precision_current="d5_refine")
         self.assertFalse(any(item["source"] == "known_event_quality" for item in probes))
         self.assertTrue(any(item["source"] in {"dasha_activation", "dasha_boundary"} for item in probes))
-        self.assertLessEqual(len(probes), 3)
+        self.assertLessEqual(len(probes), MAX_PROBES)
         for probe in probes:
             self.assertEqual(distinguish_contract_errors(probe), [])
 
@@ -639,6 +663,239 @@ class EventProbesTest(unittest.TestCase):
         relationship = [item for item in probes if item["domain"] == "relationship"]
         self.assertFalse(any(item["year"] == 2023 for item in relationship))
 
+    def test_discriminating_probes_expand_beyond_previous_cap(self) -> None:
+        probes = _probes(_request(), _multi_layer_window_built(), ["05:00", "05:06", "05:07"], "05:00")
+        keys = [item["semantic_key"] for item in probes]
+        self.assertGreater(len(probes), 3)
+        self.assertLessEqual(len(probes), MAX_PROBES)
+        self.assertTrue(keys)
+        for probe in probes:
+            self.assertIsInstance(probe.get("year"), int)
+            self.assertGreaterEqual(int(probe["year"]), 1900)
+            self.assertNotEqual(int(probe["year"]), 0)
+            self.assertEqual(distinguish_contract_errors(probe), [])
+            self.assertIn(probe["source"], {"dasha_boundary", "dasha_activation"})
+            self.assertGreaterEqual(len(probe["style_options"]), 4)
+            self.assertGreaterEqual(len(probe["candidate_ids"]), 2)
+            self.assertGreaterEqual(len(probe["expected_outcomes"]), 2)
+
+    def test_per_domain_boundary_evaluation_respects_k_budget(self) -> None:
+        from unittest.mock import patch
+
+        from scripts.rectification import event_probes as probes_mod
+
+        real = probes_mod._evaluate_contexts
+        calls_by_domain: dict[str, int] = {}
+
+        def wrapped(
+            contexts,
+            *,
+            birth_date: str,
+            domain: str,
+            year: int,
+            source: str,
+            month: int | None = None,
+            clusters=None,
+            set_version=None,
+        ):
+            if source == "dasha_boundary":
+                calls_by_domain[domain] = calls_by_domain.get(domain, 0) + 1
+            return real(
+                contexts,
+                birth_date=birth_date,
+                domain=domain,
+                year=year,
+                source=source,
+                month=month,
+                clusters=clusters,
+                set_version=set_version,
+            )
+
+        with patch.object(probes_mod, "_evaluate_contexts", side_effect=wrapped):
+            probes = _probes(_request(), _multi_layer_window_built(), ["05:00", "05:06", "05:07"], "05:00")
+        self.assertTrue(probes)
+        self.assertTrue(calls_by_domain)
+        filled: dict[str, set[int]] = {}
+        for probe in probes:
+            if probe["source"] != "dasha_boundary":
+                continue
+            filled.setdefault(str(probe["domain"]), set()).add(int(probe["year"]))
+        for domain, count in calls_by_domain.items():
+            if len(filled.get(domain, set())) >= 2:
+                self.assertLessEqual(
+                    count,
+                    MAX_BOUNDARY_CANDIDATES_PER_DOMAIN,
+                    {domain: count, "filled": filled.get(domain)},
+                )
+
+    def test_adjacent_pair_boundary_is_used_when_first_last_has_none(self) -> None:
+        from unittest.mock import patch
+
+        from scripts.rectification import event_probes as probes_mod
+
+        built = {
+            "static_contexts": [
+                _context("04:50", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=1),
+                _context("05:10", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0, d9_asc=1),
+                _context("05:40", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0, d9_asc=2),
+            ]
+        }
+
+        def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]:
+            return [date(2018, 3, 15)] if moon <= 100.0 else [date(2018, 9, 20)]
+
+        def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]:
+            moon = float(planets.get("Moon") or 0)
+            return fake_vim(_birth_date, moon, _lo, _hi)
+
+        def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict:
+            del birth_date, domain
+            if year == 2018 and month in {3, 9}:
+                return {
+                    "rule_ids": ["vim_md_domain_house"]
+                    if probes_mod._context_time(context) == "04:50"
+                    else ["no_domain_activation"]
+                }
+            return {"rule_ids": ["no_domain_activation"]}
+
+        with (
+            patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim),
+            patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana),
+            patch.object(probes_mod, "_score_year", side_effect=fake_score),
+        ):
+            probes = _probes(_request(), built, ["04:50", "05:10", "05:40"], "04:50")
+        boundary = [item for item in probes if item["source"] == "dasha_boundary"]
+        self.assertTrue(boundary)
+        self.assertTrue(all(item["year"] == 2018 for item in boundary))
+        self.assertTrue(all(item.get("month") in {3, 9} for item in boundary))
+        self.assertTrue(all(item.get("year") for item in probes))
+
+    def test_same_domain_keeps_multiple_boundary_years(self) -> None:
+        from unittest.mock import patch
+
+        from scripts.rectification import event_probes as probes_mod
+
+        built = {
+            "static_contexts": [
+                _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0),
+                _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0),
+            ]
+        }
+
+        def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]:
+            if moon <= 100.0:
+                return [date(2016, 3, 15), date(2022, 3, 15)]
+            return [date(2016, 9, 20), date(2022, 9, 20)]
+
+        def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]:
+            moon = float(planets.get("Moon") or 0)
+            return fake_vim(_birth_date, moon, _lo, _hi)
+
+        def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict:
+            del birth_date, domain, month
+            early = probes_mod._context_time(context) == "05:13"
+            if year in {2016, 2022}:
+                return {"rule_ids": ["vim_md_domain_house"] if early else ["no_domain_activation"]}
+            return {"rule_ids": ["no_domain_activation"]}
+
+        with (
+            patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim),
+            patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana),
+            patch.object(probes_mod, "_score_year", side_effect=fake_score),
+        ):
+            probes = _probes(_request(), built, ["05:13", "05:40"], "05:13")
+        relocation = [item for item in probes if item["domain"] == "relocation" and item["source"] == "dasha_boundary"]
+        years = {int(item["year"]) for item in relocation}
+        keys = {
+            (item["domain"], int(item["year"]), int(item.get("month") or 0), item["source"])
+            for item in relocation
+        }
+        self.assertGreaterEqual(len(years), 2)
+        self.assertEqual(years, {2016, 2022})
+        self.assertEqual(len(keys), len(relocation))
+
+    def test_activation_supplements_boundary_without_exceeding_domain_cap(self) -> None:
+        from unittest.mock import patch
+
+        from scripts.rectification import event_probes as probes_mod
+
+        built = {
+            "static_contexts": [
+                _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0),
+                _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0),
+            ]
+        }
+
+        def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]:
+            return [date(2020, 3, 15)] if moon <= 100.0 else [date(2020, 9, 20)]
+
+        def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]:
+            moon = float(planets.get("Moon") or 0)
+            return fake_vim(_birth_date, moon, _lo, _hi)
+
+        def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict:
+            del birth_date, domain
+            early = probes_mod._context_time(context) == "05:13"
+            if (year == 2020 and month in {3, 9}) or (year == 2018 and month is None):
+                return {"rule_ids": ["vim_md_domain_house"] if early else ["no_domain_activation"]}
+            return {"rule_ids": ["no_domain_activation"]}
+
+        with (
+            patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim),
+            patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana),
+            patch.object(probes_mod, "_score_year", side_effect=fake_score),
+        ):
+            probes = _probes(_request(), built, ["05:13", "05:40"], "05:13")
+        relocation = [item for item in probes if item["domain"] == "relocation"]
+        sources = {item["source"] for item in relocation}
+        self.assertIn("dasha_boundary", sources)
+        self.assertIn("dasha_activation", sources)
+        self.assertLessEqual(len(relocation), MAX_PROBES_PER_DOMAIN)
+        self.assertLessEqual(len(probes), MAX_PROBES)
+        self.assertTrue(all(isinstance(item.get("year"), int) and item["year"] for item in relocation))
+        activation = next(item for item in relocation if item["source"] == "dasha_activation")
+        self.assertEqual(activation["year"], 2018)
+        self.assertNotIn("month", activation)
+
+    def test_domain_at_n_does_not_add_activation_past_cap(self) -> None:
+        from unittest.mock import patch
+
+        from scripts.rectification import event_probes as probes_mod
+
+        built = {
+            "static_contexts": [
+                _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=100.0),
+                _context("05:40", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=101.0),
+            ]
+        }
+
+        def fake_vim(_birth_date: str, moon: float, _lo: int, _hi: int) -> list[date]:
+            if moon <= 100.0:
+                return [date(2016, 3, 15), date(2019, 3, 15), date(2022, 3, 15)]
+            return [date(2016, 9, 20), date(2019, 9, 20), date(2022, 9, 20)]
+
+        def fake_narayana(_asc: int, planets: dict, _birth_date: str, _lo: int, _hi: int) -> list[date]:
+            moon = float(planets.get("Moon") or 0)
+            return fake_vim(_birth_date, moon, _lo, _hi)
+
+        def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict:
+            del birth_date, domain, month
+            early = probes_mod._context_time(context) == "05:13"
+            if year in {2016, 2019, 2022, 2018}:
+                return {"rule_ids": ["vim_md_domain_house"] if early else ["no_domain_activation"]}
+            return {"rule_ids": ["no_domain_activation"]}
+
+        with (
+            patch.object(probes_mod, "_vim_start_dates", side_effect=fake_vim),
+            patch.object(probes_mod, "_narayana_start_dates", side_effect=fake_narayana),
+            patch.object(probes_mod, "_score_year", side_effect=fake_score),
+        ):
+            probes = _probes(_request(), built, ["05:13", "05:40"], "05:13")
+        relocation = [item for item in probes if item["domain"] == "relocation"]
+        self.assertLessEqual(len(relocation), MAX_PROBES_PER_DOMAIN)
+        self.assertFalse(any(item["source"] == "dasha_activation" for item in relocation))
+        self.assertEqual({int(item["year"]) for item in relocation}, {2016, 2019, 2022})
+
 
 if __name__ == "__main__":
     unittest.main()