feat: add cross-project sync status gate

references/cross_project_contract/sync_policy.v1.json

@@ -0,0 +1,36 @@
+{
+  "schema_version": 1,
+  "sync_model": "research_validates_commercial_receives_mature",
+  "repositories": {
+    "research": "732642856/yinduzhanxing",
+    "commercial": "jesse-ux/Jyotisha"
+  },
+  "directional_gates": {
+    "research_to_commercial": {
+      "source_required": "validated_in_research",
+      "target_required": "commercial_safe",
+      "required_checks": [
+        "privacy_review",
+        "focused_tests",
+        "hash_contract_result"
+      ]
+    },
+    "commercial_to_research": {
+      "source_required": "configuration_free_product_pattern",
+      "target_required": "local_test_double_or_no_secret",
+      "required_checks": [
+        "privacy_review",
+        "focused_tests"
+      ]
+    }
+  },
+  "shared_files": [
+    "references/cross_project_contract/fixture_manifest.v1.json",
+    "references/cross_project_contract/sync_ledger.json",
+    "references/cross_project_contract/sync_policy.v1.json",
+    "scripts/cross_project_contract.py",
+    "scripts/cross_project_sync_status.py",
+    "tests/test_cross_project_contract.py",
+    "tests/test_cross_project_sync_status.py"
+  ]
+}

scripts/cross_project_sync_status.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+"""Compare allow-listed shared contract files between the two Jyotish projects."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_POLICY = ROOT / "references" / "cross_project_contract" / "sync_policy.v1.json"
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def load_policy(path: Path = DEFAULT_POLICY) -> dict[str, Any]:
+    policy = json.loads(path.read_text(encoding="utf-8"))
+    if policy.get("schema_version") != 1:
+        raise ValueError("sync policy schema_version must be 1")
+    if policy.get("sync_model") != "research_validates_commercial_receives_mature":
+        raise ValueError("sync policy must encode research-first commercial-mature flow")
+    if not isinstance(policy.get("shared_files"), list) or not policy["shared_files"]:
+        raise ValueError("sync policy must contain shared_files")
+    return policy
+
+
+def compare_peer(peer_root: Path, *, policy_path: Path = DEFAULT_POLICY, root: Path = ROOT) -> dict[str, Any]:
+    policy = load_policy(policy_path)
+    missing: list[str] = []
+    mismatched: list[str] = []
+    checked: list[dict[str, str]] = []
+
+    for rel_path in policy["shared_files"]:
+        local = root / rel_path
+        peer = peer_root / rel_path
+        if not local.exists() or not peer.exists():
+            missing.append(rel_path)
+            continue
+        local_hash = _sha256(local)
+        peer_hash = _sha256(peer)
+        checked.append({"path": rel_path, "local_sha256": local_hash, "peer_sha256": peer_hash})
+        if local_hash != peer_hash:
+            mismatched.append(rel_path)
+
+    return {
+        "status": "pass" if not missing and not mismatched else "fail",
+        "sync_model": policy["sync_model"],
+        "checked_count": len(checked),
+        "missing": missing,
+        "mismatched": mismatched,
+        "checked": checked,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--peer", type=Path, required=True, help="Path to the other Jyotish repository")
+    parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
+    parser.add_argument("--format", choices=("json",), default="json")
+    args = parser.parse_args()
+
+    report = compare_peer(args.peer, policy_path=args.policy)
+    print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
+    return 0 if report["status"] == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_cross_project_sync_status.py

@@ -0,0 +1,64 @@
+"""Cross-repository sync status checks for the research/commercial pair."""
+
+from __future__ import annotations
+
+import json
+import shutil
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import cross_project_sync_status as sync_status  # noqa: E402
+
+
+POLICY = ROOT / "references" / "cross_project_contract" / "sync_policy.v1.json"
+
+
+def _make_peer_copy(tmp_path: Path) -> Path:
+    peer = tmp_path / "peer"
+    policy = sync_status.load_policy(POLICY)
+    for rel_path in policy["shared_files"]:
+        source = ROOT / rel_path
+        target = peer / rel_path
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(source, target)
+    return peer
+
+
+def test_sync_policy_encodes_research_first_commercial_mature_rule() -> None:
+    policy = sync_status.load_policy(POLICY)
+
+    assert policy["schema_version"] == 1
+    assert policy["sync_model"] == "research_validates_commercial_receives_mature"
+    assert policy["directional_gates"]["research_to_commercial"]["source_required"] == "validated_in_research"
+    assert policy["directional_gates"]["research_to_commercial"]["target_required"] == "commercial_safe"
+    assert "references/cross_project_contract/fixture_manifest.v1.json" in policy["shared_files"]
+
+
+def test_sync_status_passes_when_shared_files_match(tmp_path: Path) -> None:
+    peer = _make_peer_copy(tmp_path)
+
+    report = sync_status.compare_peer(peer, policy_path=POLICY, root=ROOT)
+
+    assert report["status"] == "pass"
+    assert report["missing"] == []
+    assert report["mismatched"] == []
+    assert report["checked_count"] == len(sync_status.load_policy(POLICY)["shared_files"])
+
+
+def test_sync_status_reports_mismatched_shared_file(tmp_path: Path) -> None:
+    peer = _make_peer_copy(tmp_path)
+    changed = peer / "references" / "cross_project_contract" / "fixture_manifest.v1.json"
+    data = json.loads(changed.read_text(encoding="utf-8"))
+    data["fixtures"][0]["compatibility_hash"] = "0" * 64
+    changed.write_text(json.dumps(data, sort_keys=True), encoding="utf-8")
+
+    report = sync_status.compare_peer(peer, policy_path=POLICY, root=ROOT)
+
+    assert report["status"] == "fail"
+    assert report["missing"] == []
+    assert report["mismatched"] == ["references/cross_project_contract/fixture_manifest.v1.json"]