root/Jyotisha
1
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat: sync mismatch progress evidence closure

Browse Source
This commit is contained in:
732642856
2026-07-20 00:03:56 +08:00
parent 198c0763a6
commit 6d4d22630c
30 changed files with 3766 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/claim_audit_runtime_gate.py
+83
View File
@@ -0,0 +1,83 @@
#!/usr/bin/env python3
"""Runtime claim gate backed by evidence_packet_index."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
HIGH_CLAIMS = {"verified_precise_prediction", "complete_absolute_truth", "production_ready", "external_verified"}
def evaluate_claim(index_path: Path, domain: str, requested_claim: str) -> dict:
index = json.loads(index_path.read_text(encoding="utf-8"))
packets = [row for row in index["packets"] if row["domain"] == domain]
if not packets:
return {
"decision": "block",
"domain": domain,
"requested_claim": requested_claim,
"allowed_claim_status": "blocked_unknown_domain",
"production_tuning_allowed": False,
"blocking_packets": [],
"boundaries": [f"No evidence packet indexed for domain: {domain}"],
}
bad = [row for row in packets if row["claim_status"] in {"blocked", "open_queue"}]
partial = [row for row in packets if row["claim_status"] in {"partial", "observation_only"}]
ready = [row for row in packets if row["claim_status"] == "ready_contract"]
if bad and requested_claim in HIGH_CLAIMS:
return {
"decision": "block",
"domain": domain,
"requested_claim": requested_claim,
"allowed_claim_status": "exploratory_unvalidated",
"production_tuning_allowed": False,
"blocking_packets": [row["packet_id"] for row in bad],
"boundaries": [row["claim_boundary"] for row in bad],
}
if partial and requested_claim in HIGH_CLAIMS:
return {
"decision": "degrade",
"domain": domain,
"requested_claim": requested_claim,
"allowed_claim_status": "partial_method_variant",
"production_tuning_allowed": False,
"blocking_packets": [row["packet_id"] for row in partial],
"boundaries": [row["claim_boundary"] for row in partial],
}
if ready and requested_claim == "ready_contract" and not bad and not partial:
return {
"decision": "allow",
"domain": domain,
"requested_claim": requested_claim,
"allowed_claim_status": "ready_contract",
"production_tuning_allowed": False,
"blocking_packets": [],
"boundaries": [row["claim_boundary"] for row in ready],
}
return {
"decision": "degrade" if partial or bad else "allow",
"domain": domain,
"requested_claim": requested_claim,
"allowed_claim_status": "limited_research_claim",
"production_tuning_allowed": False,
"blocking_packets": [row["packet_id"] for row in bad + partial],
"boundaries": [row["claim_boundary"] for row in bad + partial + ready],
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--index", type=Path, default=Path("references/oracle/evidence_packet_index_2026_07_19.json"))
parser.add_argument("--domain", required=True)
parser.add_argument("--claim", required=True)
args = parser.parse_args()
print(json.dumps(evaluate_claim(args.index, args.domain, args.claim), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/evidence_packet_index_integrity.py
+98
View File
@@ -0,0 +1,98 @@
#!/usr/bin/env python3
"""Validate evidence packet index paths, ids, claim statuses, and boundaries."""
from __future__ import annotations
import argparse
import json
from collections import Counter
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_INDEX = ROOT / "references/oracle/evidence_packet_index_2026_07_19.json"
VALID_CLAIM_STATUSES = {
"blocked",
"blocked_until_human_labels",
"blocked_until_oracle",
"open_queue",
"observation_only",
"partial",
"ready_contract",
"reference_only",
"source_intake_only",
"tooling_observation_only",
}
def repo_path(path: str) -> Path:
candidate = Path(path)
return candidate if candidate.is_absolute() else ROOT / candidate
def build(index_path: Path) -> dict[str, Any]:
index = json.loads(index_path.read_text(encoding="utf-8"))
packets = index.get("packets", [])
ids = [row.get("packet_id") for row in packets]
id_counts = Counter(ids)
duplicate_ids = sorted(packet_id for packet_id, count in id_counts.items() if packet_id and count > 1)
missing_paths = [
{"packet_id": row.get("packet_id"), "path": row.get("path")}
for row in packets
if not row.get("path") or not repo_path(row["path"]).exists()
]
invalid_claim_statuses = [
{"packet_id": row.get("packet_id"), "claim_status": row.get("claim_status")}
for row in packets
if row.get("claim_status") not in VALID_CLAIM_STATUSES
]
missing_required_fields = [
{
"packet_id": row.get("packet_id"),
"missing": [
field
for field in ("packet_id", "path", "domain", "claim_status", "consumer_policy", "claim_boundary")
if not row.get(field)
],
}
for row in packets
]
missing_required_fields = [row for row in missing_required_fields if row["missing"]]
status = "pass"
if missing_paths or duplicate_ids or invalid_claim_statuses or missing_required_fields:
status = "fail"
return {
"scope": "evidence_packet_index_integrity",
"created_at": "2026-07-19",
"status": status,
"source_index": str(index_path.relative_to(ROOT)) if index_path.is_relative_to(ROOT) else str(index_path),
"summary": {
"packet_count": len(packets),
"missing_path_count": len(missing_paths),
"duplicate_packet_id_count": len(duplicate_ids),
"invalid_claim_status_count": len(invalid_claim_statuses),
"missing_required_field_count": len(missing_required_fields),
},
"duplicate_packet_ids": duplicate_ids,
"missing_paths": missing_paths,
"invalid_claim_statuses": invalid_claim_statuses,
"missing_required_fields": missing_required_fields,
"valid_claim_statuses": sorted(VALID_CLAIM_STATUSES),
"boundary": "Integrity pass means indexed packets are present and shaped; it does not upgrade any oracle claim.",
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--index", type=Path, default=DEFAULT_INDEX)
args = parser.parse_args()
print(json.dumps(build(args.index), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/jyotishganit_vs_local_field_comparison.py
+11 -2
View File
@@ -16,7 +16,12 @@ from typing import Any
ROOT = Path(__file__).resolve().parents[1]
TARGETS = {"D2": "D2_Hora", "D4": "D4_Chaturthamsa", "D9": "D9_Navamsa", "D10": "D10_Dasamsa"}
TARGETS = {
"D2": ["D2_Hora"],
"D4": ["D4_Chaturthamsa", "D4_Turyamsa"],
"D9": ["D9_Navamsa"],
"D10": ["D10_Dasamsa"],
}
def stable(data: Any) -> str:
@@ -82,7 +87,11 @@ def jyotishganit_signs(raw: dict[str, Any], code: str) -> dict[str, str]:
def local_signs(raw: dict[str, Any], code: str) -> dict[str, str]:
section = raw.get(TARGETS[code], {})
section = {}
for key in TARGETS[code]:
if key in raw:
section = raw[key]
break
return {
body: value.get("sign")
for body, value in section.items()
scripts/local_oss_observation_inventory.py
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""Inventory local OSS Jyotish candidates as observation-only sources."""
from __future__ import annotations
import hashlib
import json
import subprocess
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
BASE = ROOT / "references" / "open_source_sources"
CANDIDATES = ["VedicAstro", "jyotishganit", "panchanga_api", "rishi-ai-mcp", "jaimini-tropical", "vedic-astro-skills"]
def sha256_file(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def git_commit(path: Path) -> str | None:
try:
return subprocess.check_output(["git", "-C", str(path), "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip()
except Exception:
return None
def first_existing(path: Path, names: list[str]) -> Path | None:
for name in names:
p = path / name
if p.exists():
return p
return None
def license_hint(path: Path) -> dict[str, Any]:
p = first_existing(path, ["LICENSE", "LICENSE.md", "LICENSE.txt", "COPYING"])
if not p:
return {"status": "missing", "file": None, "sha256": None, "hint": "unknown"}
text = p.read_text(encoding="utf-8", errors="ignore")[:3000].lower()
if "mit license" in text or "permission is hereby granted" in text:
hint = "MIT"
elif "apache license" in text:
hint = "Apache-2.0"
elif "bsd" in text:
hint = "BSD-like"
elif "gnu affero" in text:
hint = "AGPL"
elif "gnu general public license" in text:
hint = "GPL"
else:
hint = "unknown"
return {"status": "present", "file": str(p.relative_to(ROOT)), "sha256": sha256_file(p), "hint": hint}
def file_manifest(path: Path) -> list[dict[str, Any]]:
wanted = ["README.md", "README.rst", "pyproject.toml", "package.json", "pubspec.yaml", "requirements.txt"]
rows = []
for name in wanted:
p = path / name
if p.exists():
rows.append({"path": str(p.relative_to(ROOT)), "sha256": sha256_file(p)})
return rows
def api_hints(path: Path) -> list[str]:
hints = []
for pattern in ("*.py", "*.ts", "*.js", "*.dart"):
for p in list(path.rglob(pattern))[:200]:
if any(part in {".git", "node_modules", "__pycache__"} for part in p.parts):
continue
rel = str(p.relative_to(path))
name = p.stem.lower()
if any(k in name or k in rel.lower() for k in ["panch", "muhur", "dasha", "kp", "chart", "asht", "shadbala", "jaimini"]):
hints.append(rel)
if len(hints) >= 30:
return hints
return hints
def main() -> int:
rows = []
for name in CANDIDATES:
path = BASE / name
if not path.exists():
rows.append({"project_id": name, "status": "missing"})
continue
rows.append({
"project_id": name,
"status": "present",
"local_path": str(path.relative_to(ROOT)),
"git_commit": git_commit(path),
"license": license_hint(path),
"manifest_files": file_manifest(path),
"api_surface_hints": api_hints(path),
"claim_status": "observation_only",
"runtime_dependency_allowed": False,
"truth_upgrade_allowed": False,
})
out = {
"scope": "local_oss_observation_inventory",
"created_at": "2026-07-19",
"status": "complete",
"claim_status": "observation_only",
"production_tuning_allowed": False,
"truth_matrix_allowed": False,
"boundary": "Local OSS candidates are pinned for reuse/probe triage only. License and hash metadata do not validate astrological truth.",
"projects": rows,
}
print(json.dumps(out, ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/three_engine_mismatch_progress_ledger.py
+58
View File
@@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""Summarize three-engine mismatch closure progress without mutating queue."""
from __future__ import annotations
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
QUEUE = ROOT / "references/oracle/three_engine_mismatch_closure_queue_2026_07_19.json"
NODE_ATTR = ROOT / "references/oracle/jyotishganit_node_source_attribution_2026_07_19.json"
def build() -> dict:
queue = json.loads(QUEUE.read_text(encoding="utf-8"))
node = json.loads(NODE_ATTR.read_text(encoding="utf-8"))
attributed = [
{
"section": "D10",
"field": body,
"closure_state": "attributed_no_tuning",
"reason": node["attribution"]["primary_delta"],
"effect": node["attribution"]["effect"],
}
for body in ("Rahu", "Ketu")
if f"D10 {body}" in node.get("remaining_mismatches", [])
]
total = queue["summary"]["queue_count"]
return {
"scope": "three_engine_mismatch_progress_ledger",
"created_at": "2026-07-19",
"status": "progress_ledger_ready",
"claim_status": "partial",
"production_tuning_allowed": False,
"truth_matrix_allowed": False,
"source_queue": str(QUEUE.relative_to(ROOT)),
"attribution_sources": [str(NODE_ATTR.relative_to(ROOT))],
"summary": {
"source_queue_count": total,
"attributed_no_tuning_count": len(attributed),
"remaining_open_count": total - len(attributed),
},
"attributed_rows": attributed,
"boundary": (
"Attribution closes explanation work only. The original mismatch queue "
"stays open for formula/endpoint/worked-example evidence, and no "
"engine majority vote is allowed."
),
}
def main() -> int:
print(json.dumps(build(), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/vedicastro_kp_surface_locator.py
+75
View File
@@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""Locate KP star/sub/sub-sub/cusp surface in local VedicAstro source.
Observation-only. Does not execute or vendor VedicAstro.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "references/open_source_sources/VedicAstro"
TARGET = SRC / "vedicastro/VedicAstro.py"
PATTERNS = {
"rl_nl_sl_function": "def get_rl_nl_sl_data",
"planet_sub_lord": "planet_sub_lord",
"planet_sub_sub_lord": "planet_ss_lord",
"house_sub_lord": "house_sub_lord",
"house_sub_sub_lord": "house_ss_lord",
"houses_data_function": "def get_houses_data_from_chart",
}
def sha256(path: Path) -> str | None:
if not path.exists():
return None
return hashlib.sha256(path.read_bytes()).hexdigest()
def locate(text: str, pattern: str) -> list[int]:
return [i for i, line in enumerate(text.splitlines(), 1) if pattern in line]
def main() -> int:
if not TARGET.exists():
payload = {
"scope": "vedicastro_kp_surface_locator",
"status": "source_missing",
"claim_status": "blocked_source_missing",
"production_tuning_allowed": False,
}
print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
return 0
text = TARGET.read_text(encoding="utf-8", errors="ignore")
findings = {
key: {"pattern": pattern, "lines": locate(text, pattern)}
for key, pattern in PATTERNS.items()
}
table_candidates = [
str(p.relative_to(ROOT))
for p in SRC.rglob("*")
if p.is_file() and any(token in p.name.lower() for token in ["kp", "sub", "lord", "division"])
]
payload = {
"scope": "vedicastro_kp_surface_locator",
"created_at": "2026-07-19",
"status": "complete",
"claim_status": "observation_only",
"production_tuning_allowed": False,
"truth_upgrade_allowed": False,
"source_path": str(TARGET.relative_to(ROOT)),
"source_sha256": sha256(TARGET),
"findings": findings,
"external_table_candidates": table_candidates,
"kp_table_status": "fixture_missing" if not table_candidates else "candidate_paths_found",
"boundary": "VedicAstro exposes KP RL/NL/SL/SSL and house cusp fields in source, but no external numeric worked example is validated here.",
}
print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())