feat: add astrology e2e acceptance runner

2026-07-19 12:05:10 +08:00
parent e7c11b2aee
commit 6e45b1ba1a
2 changed files with 227 additions and 0 deletions
@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Validate the commercial eight-question astrology acceptance contract.
+
+Default mode is deliberately blocked: the contract exists, but real E2E proof
+requires captured answer-context JSON files for each question id.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_CONTRACT = ROOT / "references" / "cross_project_contract" / "commercial_astrology_e2e_acceptance_questions_2026_07_19.json"
+
+LAYER_PATTERNS: dict[str, list[str]] = {
+    "Vimshottari Dasha": ["vimshottari", "dasha"],
+    "Narayana Dasha": ["narayana"],
+    "D9": ["d9"],
+    "D10": ["d10"],
+    "D2": ["d2"],
+    "D11": ["d11"],
+    "D4": ["d4"],
+    "UL": ["ul", "upapada"],
+    "A10": ["a10"],
+    "Darakaraka": ["darakaraka", "dk"],
+    "Shadbala": ["shadbala"],
+    "Ashtakavarga": ["ashtakavarga"],
+    "Shadbala/AV provenance registry": ["shadbala", "ashtakavarga", "provenance"],
+    "VedAstro gateway boundary": ["vedastro"],
+    "timing precision contract": ["verified_window", "candidate_windows", "exact_triggers"],
+    "gender interpretation boundary": ["gender", "spouse", "supplement"],
+    "functional benefic/malefic": ["functional_benefic_malefic"],
+    "12th house": ["12th", "twelfth", "house_12"],
+    "Rahu/Ketu context": ["rahu", "ketu"],
+    "Ashtakoot boundary": ["ashtakoot"],
+    "birth-time uncertainty boundary": ["birth_time", "uncertain"],
+    "rectification boundary": ["rectification"],
+    "candidate windows": ["candidate_windows"],
+    "not_auto_rectified": ["not_auto_rectified"],
+    "no_majority_vote": ["no_majority_vote"],
+    "verified_window": ["verified_window"],
+    "candidate_windows": ["candidate_windows"],
+    "exact_triggers": ["exact_triggers"],
+    "day-level holdout boundary": ["holdout", "exploratory_unvalidated"],
+}
+
+FORBIDDEN_PATTERNS: dict[str, list[str]] = {
+    "exact day/month verified": ["exact_day_verified", "exact_month_verified"],
+    "single-factor Venus/Jupiter truth": ["single_factor_venus_jupiter_truth"],
+    "gender changes career math": ["gender_changes_career_math"],
+    "D10 missing when present": ["d10_missing_when_present"],
+    "guaranteed financial outcome": ["guaranteed_financial_outcome"],
+    "D2/D11 missing when present": ["d2_d11_missing_when_present"],
+    "visa/legal guarantee": ["visa_guarantee", "legal_guarantee"],
+    "relationship guarantee": ["relationship_guarantee"],
+    "gender binary forced when not provided": ["forced_binary_gender"],
+    "auto-rectified exact birth time": ["auto_rectified_exact_birth_time"],
+    "majority vote truth": ["majority_vote_truth"],
+    "method variant as absolute error": ["method_variant_absolute_error"],
+    "day/month prediction verified before holdout": ["day_month_verified_before_holdout"],
+    "exact_triggers as guaranteed events": ["exact_triggers_guaranteed_events"],
+}
+
+
+def _flatten(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True).lower()
+
+
+def _has_layer(text: str, layer: str) -> bool:
+    patterns = LAYER_PATTERNS.get(layer, [layer.lower()])
+    return all(pattern.lower() in text for pattern in patterns)
+
+
+def _has_forbidden(text: str, claim: str) -> bool:
+    return any(pattern.lower() in text for pattern in FORBIDDEN_PATTERNS.get(claim, [claim.lower()]))
+
+
+def _load_json(path: Path) -> Any:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def evaluate(contract_path: Path = DEFAULT_CONTRACT, context_dir: Path | None = None) -> dict[str, Any]:
+    contract = _load_json(contract_path)
+    rows: list[dict[str, Any]] = []
+    status = "pass"
+
+    for question in contract["questions"]:
+        qid = question["id"]
+        context_path = context_dir / f"{qid}.json" if context_dir else None
+        if not context_path or not context_path.exists():
+            rows.append(
+                {
+                    "id": qid,
+                    "status": "blocked",
+                    "reason": "missing_runtime_context_json",
+                    "required_layers": question["required_layers"],
+                }
+            )
+            status = "blocked"
+            continue
+
+        text = _flatten(_load_json(context_path))
+        missing = [layer for layer in question["required_layers"] if not _has_layer(text, layer)]
+        forbidden = [claim for claim in question["must_not_claim"] if _has_forbidden(text, claim)]
+        row_status = "pass" if not missing and not forbidden else "fail"
+        if row_status == "fail":
+            status = "fail"
+        rows.append({"id": qid, "status": row_status, "missing_layers": missing, "forbidden_claim_hits": forbidden})
+
+    return {
+        "scope": "commercial_astrology_e2e_acceptance_runner",
+        "contract": str(contract_path.relative_to(ROOT) if contract_path.is_relative_to(ROOT) else contract_path),
+        "status": status,
+        "runtime_context_required": True,
+        "question_count": len(contract["questions"]),
+        "rows": rows,
+    }
+
+
+def write_question_manifest(contract_path: Path, output_path: Path) -> dict[str, Any]:
+    contract = _load_json(contract_path)
+    manifest = {
+        "scope": "commercial_astrology_e2e_question_manifest",
+        "contract_id": contract["contract_id"],
+        "capture_instruction": "Run each user_question through the commercial consultation flow and save the answer context as <id>.json for this runner.",
+        "questions": [
+            {
+                "id": question["id"],
+                "user_question": question["user_question"],
+                "required_layers": question["required_layers"],
+                "context_filename": f"{question['id']}.json",
+            }
+            for question in contract["questions"]
+        ],
+    }
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return manifest
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--contract", type=Path, default=DEFAULT_CONTRACT)
+    parser.add_argument("--context-dir", type=Path)
+    parser.add_argument("--write-question-manifest", type=Path)
+    args = parser.parse_args()
+
+    if args.write_question_manifest:
+        manifest = write_question_manifest(args.contract, args.write_question_manifest)
+        print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True))
+        return 0
+
+    report = evaluate(args.contract, args.context_dir)
+    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())
@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import json
+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 commercial_astrology_e2e_acceptance_runner as runner  # noqa: E402
+
+
+def test_runner_blocks_without_runtime_contexts() -> None:
+    report = runner.evaluate()
+
+    assert report["status"] == "blocked"
+    assert report["question_count"] == 8
+    assert {row["status"] for row in report["rows"]} == {"blocked"}
+    assert report["runtime_context_required"] is True
+
+
+def test_runner_passes_when_all_required_layers_are_present(tmp_path: Path) -> None:
+    contract = json.loads(runner.DEFAULT_CONTRACT.read_text(encoding="utf-8"))
+    all_terms = sorted({term for patterns in runner.LAYER_PATTERNS.values() for term in patterns})
+    payload = {"context": " ".join(all_terms), "claim_status": "exploratory_unvalidated"}
+
+    for question in contract["questions"]:
+        (tmp_path / f"{question['id']}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+    report = runner.evaluate(context_dir=tmp_path)
+
+    assert report["status"] == "pass"
+    assert {row["status"] for row in report["rows"]} == {"pass"}
+
+
+def test_runner_fails_on_forbidden_claim_hits(tmp_path: Path) -> None:
+    contract = json.loads(runner.DEFAULT_CONTRACT.read_text(encoding="utf-8"))
+    all_terms = sorted({term for patterns in runner.LAYER_PATTERNS.values() for term in patterns})
+    payload = {"context": " ".join(all_terms), "claim": "majority_vote_truth"}
+
+    for question in contract["questions"]:
+        (tmp_path / f"{question['id']}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+    report = runner.evaluate(context_dir=tmp_path)
+    engine = next(row for row in report["rows"] if row["id"] == "engine_disagreement")
+
+    assert report["status"] == "fail"
+    assert engine["status"] == "fail"
+    assert engine["forbidden_claim_hits"] == ["majority vote truth"]
+
+
+def test_runner_writes_question_manifest_for_context_capture(tmp_path: Path) -> None:
+    output = tmp_path / "questions.json"
+
+    manifest = runner.write_question_manifest(runner.DEFAULT_CONTRACT, output)
+    saved = json.loads(output.read_text(encoding="utf-8"))
+
+    assert manifest == saved
+    assert saved["scope"] == "commercial_astrology_e2e_question_manifest"
+    assert len(saved["questions"]) == 8
+    assert saved["questions"][0]["context_filename"] == "marriage_timing.json"