scripts/finalize_3d_mechanism_pilot.py

"""Assemble honest blocked scientific deliverables; never invent missing results."""

import hashlib
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
NAME = "EXP-3D-20260927-state-history-pilot-v1"
DOC = ROOT / "docs/experiments" / NAME
OUT = ROOT / "outputs" / NAME


def write(path, value):
    path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n")


def main():
    audit = json.loads((DOC / "carrier_audit.json").read_text())
    blocked = {
        "status": "NOT_EVALUATED_CARRIER_BLOCKED",
        "scientific_scene_count": 0,
        "reason": "No qualified dynamic checkpoint and no calibrated 3D cohort.",
        "metrics": None,
        "synthetic_results": "smoke/ only; not scientific evidence",
    }
    for name in [
        "static_state_results",
        "controlled_history_results",
        "natural_history_results",
        "policy_branch_rewards",
        "policy_decisions",
    ]:
        write(OUT / "raw" / (name + ".json"), dict(blocked, rows=[]))
    for name in [
        "static_state_analysis",
        "history_interaction_analysis",
        "action_oracle_analysis",
        "bootstrap_results",
        "robustness_analysis",
        "cost_analysis",
    ]:
        write(OUT / "analysis" / (name + ".json"), blocked)
    write(OUT / "analysis/carrier_analysis.json", audit)
    write(
        OUT / "analysis/policy_analysis.json",
        dict(
            blocked,
            POLICY_STAGE_SKIPPED=True,
            FEEDBACK_IDENTIFIABILITY_STATUS="SKIPPED",
            training_runs=0,
        ),
    )
    for name in ["information_contract.json"]:
        (OUT / "audit" / name).write_bytes((DOC / name).read_bytes())
    for name in ["label_access_log", "sealed_query_access_log"]:
        write(
            OUT / "audit" / (name + ".json"),
            {
                "scope": "FORMAL_SCIENTIFIC",
                "events": [],
                "reads": 0,
                "synthetic_access_log": "../smoke/access_log.json",
            },
        )
    write(OUT / "locks/carrier_lock.json", audit)
    write(OUT / "locks/metric_lock.json", json.loads((DOC / "preregistration.json").read_text()))
    write(
        OUT / "locks/policy_lock.json",
        {
            "POLICY_STAGE_SKIPPED": True,
            "reason": "Scientific Phase B unavailable",
            "trained_models": [],
        },
    )
    inputs = ["src/mcss/mechanism_pilot", "scripts", "tests"]
    files = [
        p
        for folder in inputs
        for p in (ROOT / folder).glob("*.py")
        if folder == "src/mcss/mechanism_pilot" or "mechanism" in p.name
    ]
    write(
        OUT / "locks/implementation_lock.json",
        {
            "hashes": {
                str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest()
                for p in sorted(files)
            }
        },
    )
    status = {
        "CARRIER_STATUS": "BLOCKED",
        "STATIC_STATE_STATUS": "BLOCKED_CARRIER_NOT_READY",
        "WRITE_OPPORTUNITY_STATUS": "NOT_ESTABLISHED",
        "HISTORY_ACTION_STATUS": "NOT_ESTABLISHED",
        "EVALUATION_STATUS": "NOT_EVALUATED_CARRIER_BLOCKED",
        "status_interpretation": "Not evaluated, not a negative experimental finding.",
        "FEEDBACK_IDENTIFIABILITY_STATUS": "SKIPPED",
        "POLICY_STAGE_SKIPPED": True,
        "N_TRAIN_SCENES": 0,
        "N_DEV_SCENES": 0,
        "N_EVAL_SCENES": 0,
        "2D_BRANCH_STATUS": "SEALED_PENDING_INDEPENDENT_CONFIRMATION",
    }
    for key in [
        "STATIC_R_FIXED",
        "STATIC_R_RESIDUAL",
        "STATIC_ANCHOR",
        "CROSS_SCENE_REPLACEMENT_DELTA",
        "BEST_FIXED_WRITE",
        "WRITE_ORACLE_GAIN",
        "BENEFICIAL_ACTION_FLIPS",
        "HISTORY_INTERACTION_MEAN",
        "HISTORY_INTERACTION_CI",
        "GLOBAL_ACTION_GAP",
        "STATE_DEPENDENT_ACTION_GAP",
        "POLICY_P0_GAIN",
        "POLICY_P1_GAIN",
        "POLICY_P2_GAIN",
        "POLICY_REGRET",
    ]:
        status[key] = None
    status["CONTROLLED_HISTORY_PAIRS"] = 0
    status.update(SEALED_QUERY_LEAKAGE=False, TEST_DEPTH_IN_POLICY=False)
    # Replay existing discovery-only comparator without modifying its old output directory.
    import importlib.util

    spec = importlib.util.spec_from_file_location(
        "fair2d", ROOT / "scripts/explore_v2_discovery.py"
    )
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    source = ROOT / "docs/experiments/EXP-2D-opportunity-selector-v2-20260926T015728+0800"
    rows = [json.loads(x) for x in (source / "discovery_rows.jsonl").read_text().splitlines()]
    decisions = mod.crossfit_global(rows)
    old = json.loads(
        (ROOT / "docs/experiments/EXP-2D-followup-20260927/discovery_diagnostics.json").read_text()
    )
    assert decisions == old["LOSO_global_decisions"]
    result = {
        "POSTHOC_FAIR_BASELINE": True,
        "scope": "discovery only; not preregistered",
        "2D_BRANCH_STATUS": status["2D_BRANCH_STATUS"],
        "decisions": decisions,
        "methods": old["methods"],
        "comparisons": old["comparisons"],
        "replayed_decisions_identical": True,
        "media_reads": 0,
        "source_hashes": old["source_hashes"],
    }
    write(OUT / "analysis/2d_outer_fold_fixed16.json", result)
    for key, method in [
        ("2D_OUTER_FOLD_FIXED16", "LOSO_global16"),
        ("2D_GATEONLY_OOF", "GateOnly"),
        ("2D_CYCLEGATE_OOF", "CycleGate"),
    ]:
        summary = old["methods"][method]
        status[key] = sum(x["J"] for x in summary["per_sequence"]) / len(summary["per_sequence"])
    tests_file = OUT / "audit/tests.json"
    tests = json.loads(tests_file.read_text()) if tests_file.exists() else {"status": "PENDING"}
    status["TESTS"] = tests
    initial = json.loads((DOC / "implementation_audit.json").read_text())["initial_hashes"]
    changed = [
        name
        for name, digest in initial.items()
        if not (ROOT / name).is_file()
        or hashlib.sha256((ROOT / name).read_bytes()).hexdigest() != digest
    ]
    smoke = OUT / "smoke/summary.json"
    status["CANDIDATE_STATE_ISOLATION"] = (
        json.loads(smoke.read_text()).get("candidate_isolation_verified", False)
        if smoke.exists()
        else False
    )
    integrity = {
        "status": "PASS"
        if not changed and tests.get("status") == "PASS" and smoke.exists()
        else "INCOMPLETE",
        "initial_file_checks": len(initial),
        "changed_preexisting_files": changed,
        "scientific_data_reads": 0,
        "formal_results_fabricated": False,
        "synthetic_separate": True,
        "tests": tests.get("status"),
        "interpretation": "Artifact/engineering integrity only; carrier remains blocked.",
    }
    write(OUT / "audit/integrity.json", integrity)
    status["FINAL_INTEGRITY"] = integrity["status"]
    status["REPORT_DIR"] = str(DOC)
    write(DOC / "terminal_summary.json", status)
    text = (
        "3D REVERSE-JEPA + DYNAMIC TTT MECHANISM PILOT\n"
        + "\n".join(f"{k}=" + ("NA" if v is None else json.dumps(v)) for k, v in status.items())
        + "\n"
    )
    (DOC / "terminal_summary.txt").write_text(text)
    (DOC / "STATUS.md").write_text("# Status\n\n```text\n" + text + "```\n")
    print(text)


if __name__ == "__main__":
    main()