feat: sync astrology provenance gates

2026-07-19 08:59:33 +08:00
parent 5298b49a89
commit 82e7b0a95a
12 changed files with 623 additions and 1 deletions
@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Append independently sourced day-level timing holdout annotations."""
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+from scripts.day_level_holdout_validator import REQUIRED, validate
+
+
+def _row_errors(row: dict, prohibited: set[str]) -> list[dict]:
+    errors = []
+    for key in sorted(REQUIRED - set(row)):
+        errors.append({"field": key, "error": "missing"})
+    if row.get("label") not in {"target_event", "no_target_event"}:
+        errors.append({"field": "label", "error": "invalid"})
+    if not str(row.get("source_url") or "").startswith(("https://", "http://")):
+        errors.append({"field": "source_url", "error": "not_public_url"})
+    if row.get("independent_human_reviewed") is not True:
+        errors.append({"field": "independent_human_reviewed", "error": "not_independently_human_reviewed"})
+    if row.get("source_path") in prohibited:
+        errors.append({"field": "source_path", "error": "prohibited_tuning_source"})
+    return errors
+
+
+def append_annotation(path: Path, row: dict) -> dict:
+    data = json.loads(path.read_text(encoding="utf-8"))
+    prohibited = set(data.get("prohibited_tuning_data") or [])
+    errors = _row_errors(row, prohibited)
+    if errors:
+        return {"appended": False, "errors": errors, "validation": validate(path)}
+    next_row = {
+        **row,
+        "frozen_before_scoring": True,
+        "ingested_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+    }
+    data.setdefault("annotations", []).append(next_row)
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return {"appended": True, "errors": [], "validation": validate(path)}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("manifest", type=Path)
+    parser.add_argument("--row-json", required=True, help="One annotation JSON object.")
+    args = parser.parse_args()
+    print(json.dumps(append_annotation(args.manifest, json.loads(args.row_json)), ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Build Shadbala/Ashtakavarga component provenance registry from mismatch arbitration."""
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any
+
+
+CATEGORY_POLICY = {
+    "shadbala_formula_variant": {
+        "component_family": "shadbala_components",
+        "unit_contract": "Virupa/Rupa component unit must be explicit before parity claims.",
+        "allowed_claim": "component_method_variant",
+        "next_evidence_required": "worked example or source text for each six-force component formula and unit.",
+    },
+    "derived_total_from_component_variants": {
+        "component_family": "shadbala_total",
+        "unit_contract": "Total Rupa/Virupa cannot be arbitrated before component units close.",
+        "allowed_claim": "derived_total_blocked_until_components_close",
+        "next_evidence_required": "close sthana/dig/kala/chesta/naisargika/drik first, then recompute totals.",
+    },
+    "ashtakavarga_table_or_contributor_variant": {
+        "component_family": "ashtakavarga",
+        "unit_contract": "BAV/SAV tables must name contributor set, shodhana state, and Lagna inclusion.",
+        "allowed_claim": "table_variant",
+        "next_evidence_required": "public worked table with same contributor semantics and row/column schema.",
+    },
+    "endpoint_or_varga_semantics": {
+        "component_family": "varga_endpoint",
+        "unit_contract": "Sign values only; endpoint must prove requested varga/method semantics.",
+        "allowed_claim": "current_target_observation_only",
+        "next_evidence_required": "identified endpoint contract for D2/D4/D9/D10 ayanamsa/node/method.",
+    },
+}
+
+
+def build_registry(arbitration_path: str | Path) -> dict[str, Any]:
+    path = Path(arbitration_path)
+    arbitration = json.loads(path.read_text(encoding="utf-8"))
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in arbitration.get("rows") or []:
+        grouped[row["category"]].append(row)
+    registry = []
+    for category, rows in sorted(grouped.items()):
+        policy = CATEGORY_POLICY.get(category, {
+            "component_family": "unknown",
+            "unit_contract": "unknown",
+            "allowed_claim": "current_target_observation_only",
+            "next_evidence_required": "manual provenance review required.",
+        })
+        registry.append({
+            "category": category,
+            "component_family": policy["component_family"],
+            "row_count": len(rows),
+            "sections": sorted({str(row.get("section")) for row in rows}),
+            "sample_fields": [str(row.get("field")) for row in rows[:8]],
+            "unit_contract": policy["unit_contract"],
+            "allowed_claim": policy["allowed_claim"],
+            "next_evidence_required": policy["next_evidence_required"],
+            "truth_status": "classified_unresolved",
+        })
+    return {
+        "scope": "shadbala_av_component_provenance_registry",
+        "source_arbitration": str(path),
+        "status": "classified_unresolved",
+        "truth_policy": "method_variant_not_majority_vote",
+        "production_tuning_allowed": False,
+        "summary": {
+            "source_mismatch_count": arbitration.get("mismatch_count", 0),
+            "registry_count": len(registry),
+            "category_counts": dict(Counter({row["category"]: row["row_count"] for row in registry})),
+        },
+        "registry": registry,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("arbitration", nargs="?", default="references/oracle/three_engine_mismatch_arbitration_2026_07_19.json")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    registry = build_registry(args.arbitration)
+    text = json.dumps(registry, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(text, encoding="utf-8")
+    print(text, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Create a secret-free identity contract for a pinned VedAstro container build."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+def _sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _git_commit(root: Path) -> str:
+    completed = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=root, text=True, capture_output=True, check=False
+    )
+    return completed.stdout.strip() if completed.returncode == 0 else ""
+
+
+def _inspect_image(tag: str) -> dict[str, Any] | None:
+    completed = subprocess.run(
+        ["docker", "image", "inspect", tag], text=True, capture_output=True, check=False
+    )
+    if completed.returncode != 0:
+        return None
+    payload = json.loads(completed.stdout)
+    return payload[0] if isinstance(payload, list) and payload else None
+
+
+def build_identity(
+    source_root: Path,
+    *,
+    source_commit: str | None = None,
+    image_inspect: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    source_root = source_root.resolve()
+    dockerfile = source_root / "API/Dockerfile"
+    if not dockerfile.is_file():
+        raise FileNotFoundError(dockerfile)
+    docker_text = dockerfile.read_text(encoding="utf-8")
+    base_images: list[str] = []
+    stage_names: set[str] = set()
+    for match in re.finditer(
+        r"^FROM\s+([^\s]+)(?:\s+AS\s+([^\s]+))?",
+        docker_text,
+        flags=re.MULTILINE | re.IGNORECASE,
+    ):
+        image, stage = match.group(1), match.group(2)
+        if image not in stage_names:
+            base_images.append(image)
+        if stage:
+            stage_names.add(stage)
+    project_files = sorted(source_root.glob("**/*.csproj"))
+    project_hashes = {
+        str(path.relative_to(source_root)): _sha256(path)
+        for path in project_files
+        if "/bin/" not in path.as_posix() and "/obj/" not in path.as_posix()
+    }
+    image_id = (image_inspect or {}).get("Id", "")
+    repo_digests = (image_inspect or {}).get("RepoDigests") or []
+    return {
+        "scope": "vedastro_reproducible_build_identity",
+        "source_root": str(source_root),
+        "source_commit": source_commit or _git_commit(source_root),
+        "dockerfile_path": "API/Dockerfile",
+        "dockerfile_sha256": _sha256(dockerfile),
+        "base_images": base_images,
+        "project_file_hashes": project_hashes,
+        "image_id": image_id,
+        "repo_digests": repo_digests,
+        "status": "reproducible_candidate_built" if image_id else "source_pinned_image_not_built",
+        "boundary": "Identifies the pinned local candidate only; it does not identify api.vedastro.org.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("source_root", type=Path)
+    parser.add_argument("--image-tag", default="")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    report = build_identity(
+        args.source_root,
+        image_inspect=_inspect_image(args.image_tag) if args.image_tag else None,
+    )
+    text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(text, encoding="utf-8")
+    print(text, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())