feat: add production governance controls

scripts/answer_quality_audit.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+import json, sys
+FORBIDDEN = ["一定发生", "保证结婚", "保证发财", "医疗诊断", "确诊", "签证保证", "exact_day_verified", "full_year_certainty"]
+REQUIRED_WHEN_TIMING = ["候选", "窗口", "边界"]
+def audit_answer(text: str) -> dict:
+    hits = [x for x in FORBIDDEN if x.lower() in text.lower()]
+    timing_missing = ("什么时候" in text or "几月" in text or "哪天" in text) and not any(x in text for x in REQUIRED_WHEN_TIMING)
+    return {"status": "pass" if not hits and not timing_missing else "fail", "forbidden_hits": hits, "timing_boundary_missing": timing_missing}
+def run(path: str | None = None) -> dict:
+    rows = json.load(open(path, encoding="utf-8")) if path else []
+    results = [audit_answer(str(r.get("answer", r))) for r in rows]
+    return {"scope": "answer_quality_audit", "status": "pass" if all(r["status"] == "pass" for r in results) else "fail", "results": results}
+if __name__ == "__main__":
+    print(json.dumps(run(sys.argv[1] if len(sys.argv) > 1 else None), ensure_ascii=False, indent=2))

scripts/long_term_validation_dashboard.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+import json
+from pathlib import Path
+ROOT = Path(__file__).resolve().parents[1]
+def _load(rel: str) -> dict:
+    p = ROOT / rel
+    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {"status": "missing"}
+def run() -> dict:
+    e2e = _load("references/cross_project_contract/commercial_astrology_e2e_runtime_capture_report_2026_07_19.json")
+    return {"scope": "long_term_validation_dashboard", "status": "ready", "items": {
+        "commercial_e2e": e2e.get("status"),
+        "vedastro_hosted_identity": "blocked_until_build_version",
+        "day_month_holdout": "awaiting_independent_labels",
+        "shadbala_av": "partial_worked_example_arbitration",
+        "kp_precision": "probe_only",
+        "muhurta": "gap_registry_probe_only"
+    }}
+if __name__ == "__main__":
+    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))

scripts/production_e2e_monitor.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+import json, os, urllib.request
+from pathlib import Path
+import commercial_astrology_e2e_acceptance_runner as runner
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_CONTEXT = ROOT / "artifacts" / "commercial_astrology_e2e_contexts_2026_07_19_run6"
+
+def _get(url: str, cookie: str = "") -> dict:
+    req = urllib.request.Request(url, headers={"cookie": cookie} if cookie else {})
+    with urllib.request.urlopen(req, timeout=20) as r:
+        return {"status": r.status, "body": r.read(2000).decode("utf-8", "replace")}
+
+def run(base_url: str | None = None, context_dir: Path = DEFAULT_CONTEXT) -> dict:
+    base_url = base_url or os.getenv("JYOTISHA_PRODUCTION_BASE_URL")
+    cookie = os.getenv("JYOTISHA_PRODUCTION_COOKIE", "")
+    health = _get(base_url.rstrip("/") + "/api/health", cookie) if base_url else {"status": "skipped_no_base_url"}
+    acceptance = runner.evaluate(context_dir=context_dir)
+    status = "pass" if acceptance["status"] == "pass" and (not base_url or health["status"] == 200) else "fail"
+    return {"scope": "production_e2e_monitor", "status": status, "health": health, "acceptance": acceptance}
+
+if __name__ == "__main__":
+    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))

scripts/profile_persistence_regression.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+import json, os, urllib.request
+
+def _request(method: str, url: str, payload: dict, cookie: str) -> dict:
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(url, data=data, method=method, headers={"content-type": "application/json", "cookie": cookie})
+    with urllib.request.urlopen(req, timeout=20) as r:
+        return {"status": r.status, "body": r.read(2000).decode("utf-8", "replace")}
+
+def run() -> dict:
+    base = os.getenv("JYOTISHA_PRODUCTION_BASE_URL")
+    cookie = os.getenv("JYOTISHA_PRODUCTION_COOKIE")
+    if not base or not cookie:
+        return {"scope": "profile_persistence_regression", "status": "blocked", "reason": "missing_base_url_or_cookie"}
+    payload = {"displayName": "E2E验证用户", "birth": {"year": 1997, "month": 1, "day": 1, "hour": 12, "minute": 0, "city": "北京", "lat": 39.9042, "lon": 116.4074, "tz": 8}}
+    account = _request("PATCH", base.rstrip()+"/api/account", payload, cookie)
+    return {"scope": "profile_persistence_regression", "status": "pass" if 200 <= account["status"] < 300 else "fail", "account": account}
+
+if __name__ == "__main__":
+    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))

scripts/skill_sync_admission_gate.py

@@ -0,0 +1,16 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+import json
+from pathlib import Path
+ROOT = Path(__file__).resolve().parents[1]
+VIEW = ROOT / "references/oracle/effective_skill_capability_view_2026_07_19.json"
+def run() -> dict:
+    data = json.loads(VIEW.read_text(encoding="utf-8"))
+    blocked = []
+    for item in data.get("items", []):
+        status = item.get("effective_status") or item.get("claim_status")
+        if status in {"blocked", "research_only", "reference_only"} and item.get("commercial_claim_allowed") is True:
+            blocked.append(item.get("technique_id") or item.get("id"))
+    return {"scope": "skill_sync_admission_gate", "status": "pass" if not blocked else "fail", "blocked_promotions": blocked}
+if __name__ == "__main__":
+    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))