Jyotisha/tests/test_owner_case_purge_contract.py

"""Reference cleanup must not turn removed evidence into verified capability."""

from __future__ import annotations

import hashlib
import json
from collections import Counter
from pathlib import Path

from scripts.skill_truth_overlay_view import build_effective_registry

ROOT = Path(__file__).resolve().parents[1]
ORACLE = ROOT / "references/oracle"


def load(name: str) -> dict:
    return json.loads((ORACLE / name).read_text(encoding="utf-8"))


def test_evidence_index_counts_and_paths_remain_consistent() -> None:
    index = load("evidence_packet_index_2026_07_19.json")
    packets = index["packets"]
    assert index["production_tuning_allowed"] is False
    assert index["summary"] == {
        "packet_count": len(packets),
        "blocked_or_partial_count": sum(row["claim_status"] != "ready_contract" for row in packets),
        "human_review_required_count": sum(row["consumer_policy"] == "human_review_required" for row in packets),
    }
    assert len({row["packet_id"] for row in packets}) == len(packets)
    assert all((ROOT / row["path"]).is_file() for row in packets)


def test_removed_overlay_evidence_preserves_restrictive_statuses() -> None:
    overlay = load("skill_truth_overlay_2026_07_19.json")
    rows = {row["technique_id"]: row for row in overlay["overrides"]}
    assert overlay["production_tuning_allowed"] is False
    assert {key: row["corrected_status"] for key, row in rows.items()} == {
        "kp_system": "reference_only",
        "muhurta": "reference_only",
        "gochara_event_timing": "reference_only",
        "sahams": "blocked",
        "sphuta_trisphuta_family": "blocked",
        "tajika_yogas": "partial",
        "conception_chart": "research_only_blocked",
        "relationship_combinations": "partial_registry_only",
    }
    for key in ("kp_system", "muhurta", "gochara_event_timing"):
        assert rows[key]["evidence"] is None
        assert rows[key]["evidence_status"] == "removed"
        assert "restriction remains in force" in rows[key]["reason"]
    assert all(
        row["evidence"] is None or (ROOT / row["evidence"]).is_file()
        for row in rows.values()
    )


def test_effective_view_matches_overlay_without_replacement_oracle() -> None:
    stored = load("effective_skill_capability_view_2026_07_19.json")
    rebuilt = build_effective_registry(
        ROOT / "references/technique_registry.json",
        ORACLE / "skill_truth_overlay_2026_07_19.json",
    )
    # The stored view predates unrelated registry additions. This cleanup only
    # changes evidence references; it must not promote additional capabilities.
    rebuilt_rows = {row["technique_id"]: row for row in rebuilt["techniques"]}
    assert stored["overlay_count"] == rebuilt["overlay_count"]
    assert all(row == rebuilt_rows[row["technique_id"]] for row in stored["techniques"])
    assert stored["technique_count"] == len(stored["techniques"])


def test_missing_layer_plan_keeps_holdout_and_observation_gates() -> None:
    plan = load("rectification_missing_layer_integration_plan_2026_07_19.json")
    assert plan["production_tuning_allowed"] is False
    layer = next(row for row in plan["layers"] if row["layer_id"] == "gochara_transit_trigger_score")
    assert layer["audit_evidence_status"] == "removed"
    assert layer["implementation_status"] == "partial_observation_holdout_blocked"
    assert layer["output_status"] == "exploratory_observation_only"
    assert "negative holdout" in layer["entry_gate"]
    assert "until holdout passes" in layer["claim_boundary"]


def test_source_queue_marks_removed_scan_without_promoting_claims() -> None:
    queue = load("source_runtime_closure_queue_2026_07_21.json")
    assert queue["source_scan"] is None
    assert queue["source_scan_status"] == "removed"
    assert queue["claim_status"] == "open_queue"
    assert queue["truth_matrix_allowed"] is False
    assert queue["production_tuning_allowed"] is False
    assert queue["boundary"] == "queue_only_no_adapter_or_truth_upgrade"


def test_formalization_retains_boundaries_without_deleted_outputs() -> None:
    registry = load("workbuddy_round4_formalization_registry_2026_07_21.json")
    assert registry["source_ledger"] is None
    assert registry["source_ledger_status"] == "removed"
    assert registry["truth_matrix_allowed"] is False
    assert registry["production_tuning_allowed"] is False
    assert registry["excluded_round4_candidates"] == []
    assert "remain forbidden" in registry["excluded_candidate_policy"]
    for row in registry["formalized_domains"]:
        assert row["runtime_copy_allowed"] is False
        assert row["claim_upgrade"] == "none"
        assert row["next_required_closure"]
        assert all((ROOT / path).is_file() for path in row["formal_outputs"])
        if not row["formal_outputs"]:
            assert row["formalization_status"] == "evidence_removed"
            assert "not usable oracle evidence" in row["evidence_boundary"]


def test_usage_queue_counts_removed_evidence_without_formalization_claim() -> None:
    queue = load("workbuddy_round4_usage_closure_queue_2026_07_21.json")
    rows = queue["candidate_status"]
    counts = Counter(row["usage_status"] for row in rows)
    assert queue["summary"] == {
        "round4_migrate_candidate_count": len(rows),
        "formalized_count": counts["formalized"],
        "deferred_pending_oracle_count": counts["deferred_pending_oracle"],
        "deferred_pending_invocation_audit_count": counts["deferred_pending_invocation_audit"],
        "evidence_removed_count": counts["evidence_removed"],
    }
    assert queue["source_ledger"] is None
    assert queue["source_ledger_status"] == "removed"
    assert queue["truth_matrix_allowed"] is False
    assert queue["production_tuning_allowed"] is False
    for row in rows:
        assert row["claim_upgrade"] == "none"
        if row["usage_status"] == "evidence_removed":
            assert row["formal_output"] is None
            assert "No claim upgrade" in row["next_action"]
        else:
            assert (ROOT / row["formal_output"]).is_file()


def test_invocation_matrix_paths_counts_hash_and_boundaries() -> None:
    matrix = load("full_technique_invocation_matrix_2026_07_22.json")
    digest = matrix["content_hash"]
    matrix["content_hash"] = ""
    assert digest == hashlib.sha256(
        json.dumps(matrix, ensure_ascii=False, sort_keys=True).encode("utf-8")
    ).hexdigest()
    rows = matrix["rows"]
    by_id = {row["technique_id"]: row for row in rows}
    assert matrix["production_tuning_allowed"] is False
    assert matrix["truth_matrix_allowed"] is False
    assert matrix["summary"] == {
        "technique_count": len(rows),
        "first_batch_count": sum(row["first_batch_requested"] for row in rows),
        "material_but_not_fully_invoked_count": sum(row["has_material_but_not_fully_invoked"] for row in rows),
        "top50_count": len(matrix["top50_material_not_invoked"]),
        **{f"{batch.lower()}_count": sum(row["priority_batch"] == batch for row in rows) for batch in ("P0", "P1", "P2")},
    }
    for key in ("first_batch_execution_queue", "top50_material_not_invoked"):
        assert all(row == by_id[row["technique_id"]] for row in matrix[key])
    for bucket in matrix["migration_batches"].values():
        assert all(row == by_id[row["technique_id"]] for row in bucket)
    for row in rows:
        for paths in row["sample_files"].values():
            assert all((ROOT / path).is_file() for path in paths)
    for key in ("kp_exact_cusp", "kp_star_sub_sub", "timing_holdout", "saham", "tajika", "gulika", "sphuta", "prashna"):
        assert by_id[key]["claim_status"] == "blocked_or_observation_only"
        assert by_id[key]["commercial_sync_policy"] == "sync_observation_or_boundary_contract_only"


def test_manifest_private_paths_are_placeholders_not_runtime_evidence() -> None:
    manifest = json.loads((ROOT / "references/rangacharya_source_manifest.json").read_text(encoding="utf-8"))
    paths = [
        path
        for source in manifest["sources"]
        for path in ([source["path"]] if "path" in source else source.get("paths", []))
    ]
    assert paths
    assert all(path.startswith("<home>/") for path in paths)
    assert all(rule["adjudication_enabled"] is False for rule in manifest["rules"])