sync: import fragment migration registries

tests/test_fragment_migration_registries.py

@@ -0,0 +1,46 @@
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+EVENT = ROOT / "references/oracle/event_judgment_fragment_rule_family_registry_2026_07_21.json"
+JOURNEY = ROOT / "references/oracle/research_birth_time_journey_ui_contract_2026_07_21.json"
+INDEX = ROOT / "references/oracle/evidence_packet_index_2026_07_19.json"
+
+
+def test_event_judgment_fragment_registry_ports_rules_not_engine():
+    data = json.loads(EVENT.read_text(encoding="utf-8"))
+    assert data["source_policy"] == "rule_family_inventory_only_no_runtime_copy"
+    assert data["claim_status"] == "ready_contract"
+    routes = {row["route"]: row for row in data["route_families"]}
+    assert {"relationship", "career", "wealth"} <= set(routes)
+    relationship = {row["key"] for row in routes["relationship"]["required_families"]}
+    assert {"d9_navamsa", "upapada_lagna", "vimshottari_current", "narayana_current"} <= relationship
+    career = {row["key"] for row in routes["career"]["required_families"]}
+    assert {"d10_dasamsa", "a10_karma_pada", "shadbala"} <= career
+    wealth = {row["key"] for row in routes["wealth"]["required_families"]}
+    assert {"d2_hora", "ashtakavarga_house_scores", "gains_convergence"} <= wealth
+    assert any("Functional Benefic/Malefic" in item for item in data["global_requirements"])
+
+
+def test_birth_time_journey_contract_ports_behavior_not_commercial_runtime():
+    data = json.loads(JOURNEY.read_text(encoding="utf-8"))
+    assert data["source_policy"] == "behavior_contract_only_no_commercial_code"
+    contracts = {row["contract_id"]: row for row in data["contracts"]}
+    assert {
+        "profile_mode_hides_chat_overlay",
+        "new_chat_restores_chat_surface",
+        "candidate_claim_stays_exploratory",
+        "local_resume_only",
+        "user_error_contract",
+    } <= set(contracts)
+    assert "Supabase runtime" in data["forbidden_imports"]
+    assert "credits/payment/subscription logic" in data["forbidden_imports"]
+
+
+def test_fragment_migration_registries_are_indexed():
+    packets = {
+        row["packet_id"]: row
+        for row in json.loads(INDEX.read_text(encoding="utf-8"))["packets"]
+    }
+    assert packets["event_judgment_fragment_rule_family_registry_2026_07_21"]["claim_status"] == "ready_contract"
+    assert packets["research_birth_time_journey_ui_contract_2026_07_21"]["claim_status"] == "ready_contract"

tests/test_fragment_second_pass_candidate_ledger.py

@@ -0,0 +1,46 @@
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+LEDGER = ROOT / "references/oracle/fragment_second_pass_candidate_ledger_2026_07_21.json"
+
+
+def test_second_pass_fragment_ledger_has_only_allowed_decisions():
+    data = json.loads(LEDGER.read_text(encoding="utf-8"))
+    assert data["scope"] == "fragment_second_pass_candidate_ledger"
+    assert data["claim_status"] == "open_queue"
+    assert data["production_tuning_allowed"] is False
+    assert data["truth_matrix_allowed"] is False
+    allowed = set(data["allowed_decisions"])
+    assert allowed == {
+        "migrate_to_research_test_or_registry",
+        "reference_only",
+        "forbidden_private_or_obsolete",
+    }
+    assert len(data["candidates"]) == 5
+    assert all(row["decision"] in allowed for row in data["candidates"])
+
+
+def test_private_handan_packets_are_forbidden_not_migrated():
+    rows = {
+        row["candidate_id"]: row
+        for row in json.loads(LEDGER.read_text(encoding="utf-8"))["candidates"]
+    }
+    for candidate_id in [
+        "workbuddy_shadbala_handan_operator_card",
+        "workbuddy_first_shadbala_packet_assistant",
+    ]:
+        assert rows[candidate_id]["decision"] == "forbidden_private_or_obsolete"
+        assert "Do not migrate" in rows[candidate_id]["migration_plan"]
+
+
+def test_reusable_fragments_become_tests_or_registries_not_runtime_copy():
+    rows = {
+        row["candidate_id"]: row
+        for row in json.loads(LEDGER.read_text(encoding="utf-8"))["candidates"]
+    }
+    assert rows["workbuddy_event_judgment_engine"]["decision"] == "migrate_to_research_test_or_registry"
+    assert "Do not copy old engine" in rows["workbuddy_event_judgment_engine"]["migration_plan"]
+    assert rows["commercial_birth_time_journey_tests"]["decision"] == "migrate_to_research_test_or_registry"
+    assert "Copy no code" in rows["commercial_birth_time_journey_tests"]["risk"]
+    assert rows["vedicastro_kp_source_table_candidate"]["decision"] == "reference_only"