feat: sync worked example closure bridge

2026-07-20 02:48:15 +08:00
parent d048e6a652
commit 956a8a28ca
7 changed files with 605 additions and 2 deletions
@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Link three-engine mismatch owner tracks to worked-example intake domains."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+PLAN = ROOT / "references/oracle/three_engine_owner_track_batch_plan_2026_07_20.json"
+INTAKE = ROOT / "references/oracle/worked_example_packet_intake_plan_2026_07_20.json"
+
+TRACK_TO_INTAKE = {
+    "formula_source": ["shadbala_component_closure"],
+    "derived_total": ["shadbala_component_closure"],
+}
+
+
+def build(date: str) -> dict[str, Any]:
+    plan = json.loads(PLAN.read_text(encoding="utf-8"))
+    intake = json.loads(INTAKE.read_text(encoding="utf-8"))
+    intake_by_domain = {row["domain"]: row for row in intake["domain_queues"]}
+    links = []
+    for batch in plan["batches"]:
+        domains = TRACK_TO_INTAKE.get(batch["owner_track"], [])
+        linked = [intake_by_domain[d] for d in domains if d in intake_by_domain]
+        links.append(
+            {
+                "owner_track": batch["owner_track"],
+                "ticket_count": batch["ticket_count"],
+                "categories": batch["categories"],
+                "linked_intake_domains": domains,
+                "linked_candidate_count": sum(row["candidate_count"] for row in linked),
+                "linked_blocking_fields": sorted({field for row in linked for field in row["blocking_fields"]}),
+                "next_non_numeric_evidence": batch["next_evidence"] if not domains else [],
+                "closure_condition": "Close only by field-level replay against a numeric packet or by explicit method-variant attribution.",
+                "claim_boundary": "Bridge only; does not close mismatches, tune production, or majority-vote truth.",
+            }
+        )
+    return {
+        "scope": "three_engine_worked_example_bridge",
+        "created_at": date,
+        "status": "bridge_ready",
+        "claim_status": "open_queue",
+        "production_tuning_allowed": False,
+        "truth_matrix_allowed": False,
+        "sources": {
+            "owner_track_batch_plan": str(PLAN.relative_to(ROOT)),
+            "worked_example_packet_intake_plan": str(INTAKE.relative_to(ROOT)),
+        },
+        "summary": {
+            "owner_track_count": len(links),
+            "linked_owner_track_count": sum(1 for row in links if row["linked_intake_domains"]),
+            "linked_candidate_count": sum(row["linked_candidate_count"] for row in links),
+            "closed_mismatch_count": 0,
+        },
+        "owner_track_links": links,
+        "boundary": "Bridge converts owner tracks into evidence asks; all mismatch rows remain open until replay/attribution artifacts close them.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--date", default="2026-07-20")
+    args = parser.parse_args()
+    print(json.dumps(build(args.date), ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""Build a domain intake plan from numeric worked-example eligibility rows."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ELIGIBILITY = ROOT / "references/oracle/worked_example_numeric_packet_eligibility_2026_07_20.json"
+OUT = ROOT / "references/oracle/worked_example_packet_intake_plan_2026_07_20.json"
+
+DOMAIN_BY_TOPIC = {
+    "KP cusp": "kp_precision_timing",
+    "KP sub-lord table": "kp_precision_timing",
+    "Tarabala/Chandrabala": "muhurta_factor_scoring",
+    "Rahu Kalam": "muhurta_factor_scoring",
+    "Shadbala Virupa": "shadbala_component_closure",
+}
+
+STATUS_RANK = {
+    "runtime_only_public_oracle_missing": 4,
+    "reference_table_hash_needed": 3,
+    "raw_capture_needed": 2,
+    "fixture_and_raw_capture_needed": 2,
+    "formula_reference_only": 1,
+}
+
+CANONICAL_BLOCKERS = {
+    "fixed public input case": "public_numeric_expected_values",
+    "per-component Virupa expected values": "public_numeric_expected_values",
+    "expected factor table": "public_numeric_expected_values",
+    "expected interval": "public_numeric_expected_values",
+    "worked example link": "public_numeric_expected_values",
+    "raw output hash": "raw_capture_hash",
+    "captured raw hash": "raw_capture_hash",
+    "raw capture hash": "raw_capture_hash",
+    "table hash": "source_table_hash",
+    "captured table raw": "source_table_raw",
+    "license/copyright boundary": "license_boundary",
+    "formula variant": "method_variant",
+    "method settings": "method_settings",
+    "ayanamsa": "method_settings",
+    "house_system": "method_settings",
+}
+
+
+def canonicalize(fields: list[str]) -> list[str]:
+    out = []
+    for field in fields:
+        key = CANONICAL_BLOCKERS.get(field, field)
+        if key not in out:
+            out.append(key)
+    return out
+
+
+def topic_domain(topic: str) -> str:
+    return DOMAIN_BY_TOPIC.get(topic, "worked_example_collection")
+
+
+def build(date: str) -> dict[str, Any]:
+    source = json.loads(ELIGIBILITY.read_text(encoding="utf-8"))
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in source["rows"]:
+        item = {
+            "topic": row["topic"],
+            "url": row["url"],
+            "candidate_type": row["candidate_type"],
+            "eligibility_status": row["eligibility_status"],
+            "runtime_observation_available": row["runtime_observation_available"],
+            "blocking_fields": canonicalize(row["missing_for_oracle"]),
+            "upgrade_policy": "observation_only_until_numeric_packet",
+            "next_capture_artifact": f"references/oracle/artifacts/{row['topic'].lower().replace('/', '_').replace(' ', '_')}_worked_example_packet.json",
+        }
+        grouped[topic_domain(row["topic"])].append(item)
+
+    domain_queues = []
+    for domain in sorted(grouped):
+        items = grouped[domain]
+        statuses = [item["eligibility_status"] for item in items]
+        blockers = []
+        for item in items:
+            for field in item["blocking_fields"]:
+                if field not in blockers:
+                    blockers.append(field)
+        domain_queues.append(
+            {
+                "domain": domain,
+                "candidate_count": len(items),
+                "highest_status": max(statuses, key=lambda x: STATUS_RANK.get(x, 0)),
+                "blocking_fields": blockers,
+                "next_action_owner": "oracle_intake",
+                "closure_condition": "Promote only after exact public input, method settings, expected numeric values, raw capture, raw hash, and replay comparison are archived.",
+                "items": items,
+            }
+        )
+
+    return {
+        "scope": "worked_example_packet_intake_plan",
+        "created_at": date,
+        "status": "intake_plan_ready",
+        "claim_status": "open_queue",
+        "production_tuning_allowed": False,
+        "truth_matrix_allowed": False,
+        "sources": {"eligibility": str(ELIGIBILITY.relative_to(ROOT))},
+        "summary": {
+            "domain_count": len(domain_queues),
+            "candidate_count": sum(row["candidate_count"] for row in domain_queues),
+            "oracle_ready_count": 0,
+            "domains_with_runtime_observation": sum(
+                1 for row in domain_queues if any(item["runtime_observation_available"] for item in row["items"])
+            ),
+        },
+        "domain_queues": domain_queues,
+        "boundary": "Intake plan only; public examples remain observation-only until numeric packets are captured and replayed.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--date", default="2026-07-20")
+    args = parser.parse_args()
+    print(json.dumps(build(args.date), ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())