scripts/finalize_observability_supervision.py
"""Archive completed geometry attribution without inference or publishing weights."""
import argparse
import gzip
import json
import shutil
import tarfile
from pathlib import Path
from mcss.mechanism_pilot.small_training import sha, write_json
EXPERIMENT = "EXP-3D-CONTEXT-OBSERVABILITY-SUPERVISION-V1"
PHASES = ("baseline", "formal")
def read(path):
return json.loads(path.read_text())
def require(condition, message):
if not condition:
raise RuntimeError(message)
def write_gzip(source, destination):
destination.parent.mkdir(parents=True, exist_ok=True)
with destination.open("wb") as output:
with gzip.GzipFile(filename="", fileobj=output, mode="wb", mtime=0) as compressed:
with source.open("rb") as raw:
shutil.copyfileobj(raw, compressed)
def source_snapshot(root, repository):
paths = set(repository.glob("src/mcss/**/*.py"))
paths.update(repository.glob("scripts/*observability*.py"))
paths.update(repository.glob("scripts/*optimization_bounds*.py"))
paths.update(repository.glob("scripts/*direct_capacity*.py"))
paths.update(repository.glob("tests/test_*observability*.py"))
paths.update(repository.glob("tests/test_*optimization*.py"))
paths.update(repository.glob("tests/test_direct_capacity*.py"))
for name in ("pyproject.toml", "uv.lock"):
if (repository / name).is_file():
paths.add(repository / name)
paths = sorted(p for p in paths if p.is_file())
sources = {str(p.relative_to(repository)): sha(p) for p in paths}
write_json(root / "source_manifest.json", sources)
# Filesystem enumeration deliberately includes new, not-yet-git-tracked source files.
with (root / "source_snapshot.tar.gz").open("wb") as output:
with gzip.GzipFile(filename="", fileobj=output, mode="wb", mtime=0) as compressed:
with tarfile.open(fileobj=compressed, mode="w|") as archive:
for path in paths:
info = archive.gettarinfo(str(path), arcname=str(path.relative_to(repository)))
info.uid = info.gid = 0
info.uname = info.gname = ""
info.mtime = 0
with path.open("rb") as handle:
archive.addfile(info, handle)
return sources
def validate(root, docs):
require(root.name == docs.name == EXPERIMENT, "Wrong experiment directory")
require(
root != docs and root not in docs.parents and docs not in root.parents,
"Output and docs must be separate",
)
require(
not list(docs.rglob("*.pt")) and not list(docs.rglob("*.npz")),
"Destination already contains excluded weights or predictions",
)
required = [
"README.md",
"STATUS.md",
"commands.sh",
"tests.json",
"statistics_reproduction.json",
"observability_analysis.json",
"supervision_analysis.json",
"gap_closure_analysis.json",
"wrong_scene_analysis.json",
"bootstrap_results.json",
"cost_analysis.json",
"audit/statistics_reproduction_audit.json",
]
for name in required:
require((root / name).is_file(), f"Missing completed artifact: {name}")
passes = [
"baseline_reproduction.json",
"audit/postrun_validation.json",
"audit/statistics_reproduction_audit.json",
"audit/observability_integrity.json",
]
for phase in PHASES:
passes += [f"{phase}_optimization_complete.json", f"raw/{phase}_evaluation_integrity.json"]
for name in passes:
record = read(root / name)
require(record.get("status") == "PASS", f"Not PASS: {name}")
for field in ("final_holdout_touched", "dynamic_ttt_run", "new_carrier_trained"):
require(
not record.get(field, False) and not record.get(field.upper(), False),
f"Forbidden action: {field}",
)
post = read(root / "audit/postrun_validation.json")
require(post["n_states"] == 136, "Incomplete audit")
tests = read(root / "tests.json")
full = tests.get("full_suite", tests)
require(full.get("passed", 0) > 0, "No full-suite pass evidence")
for record in (tests, full):
require(str(record.get("status", "PASS")).upper() in ("PASS", "PASSED"), "Tests failed")
require(
all(
record.get(k, 0) in (0, None, False)
for k in ("failed", "failures", "errors", "error")
),
"Test errors remain",
)
prereg = read(root / "preregistration.json")
for path, digest in prereg["locked_file_sha256"].items():
require(sha(root / path) == digest, f"Preregistration hash mismatch: {path}")
for record in ("config.json",):
for path, digest in read(root / record)["source_sha256"].items():
require(sha(Path(path)) == digest, f"Source changed: {path}")
for name in ("statistics_reproduction.json",):
record = read(root / name)
require(
record["raw_only"] and not record["media_or_state_checkpoint_loaded"], "Not raw-only"
)
for mapping in ("input_sha256", "source_sha256"):
for path, digest in record[mapping].items():
require(sha(Path(path)) == digest, f"Statistics input changed: {path}")
seal = read(root / "audit/previous_experiment_seal.json")
require(len(seal["files"]) == 11285, "Expected 11285 old sealed artifacts")
for name, record in seal["files"].items():
path = Path(name)
require(
path.stat().st_size == record["bytes"] and sha(path) == record["sha256"],
f"Previous artifact changed: {name}",
)
require(len(list((root / "figures").glob("*.png"))) >= 7, "Seven figures required")
def checkpoint_index(root):
specs = read(root / "state_plan.json")["states"]
require(len(specs) == 136 and len({s["key"] for s in specs}) == 136, "Incomplete state plan")
rows, summaries, trajectories = [], {}, {}
for spec in specs:
folder = root / spec["optimization_path"]
state = root / spec["state_path"]
summary, completion = read(folder / "summary.json"), read(folder / "completion.json")
require(
completion["status"] == "PASS" and completion["key"] == spec["key"],
"State completion identity mismatch",
)
require(completion["plan_sha256"] == sha(root / "state_plan.json"), "Plan changed")
require(completion["config_sha256"] == sha(root / "config.json"), "Config changed")
require(
completion["state_file_sha256"] == summary["state_file_sha256"] == sha(state),
"Checkpoint changed",
)
require(completion["summary_sha256"] == sha(folder / "summary.json"), "Summary changed")
require(completion["state_hash"] == summary["state_hash"], "State tensor hash mismatch")
require(
summary["variant"] == spec["variant"] and summary["selected_step"] == 10000,
"Frozen variant/final budget changed",
)
checks = read(folder / "full_objective_checkpoints.json")
trace = read(folder / "trace.json")
require([c["step"] for c in checks] == list(range(0, 10001, 100)), "Incomplete checkpoints")
require([r["step"] for r in trace] == [1] + list(range(10, 10001, 10)), "Incomplete trace")
summaries[spec["key"]] = {"spec": spec, "summary": summary}
trajectories[spec["key"]] = {
"spec": spec,
"summary": summary,
"full_objective_checkpoints": checks,
"trace": trace,
"lock": read(folder / "lock.json"),
"completion": completion,
"optimizer_GT_access": read(folder / "access.json"),
"source_file_sha256": {
name: sha(folder / name)
for name in (
"trace.json",
"full_objective_checkpoints.json",
"optimization_trace.csv",
"summary.json",
"lock.json",
"completion.json",
"access.json",
)
},
}
rows.append(
{
**spec,
"bytes": state.stat().st_size,
"state_file_sha256": sha(state),
"state_tensor_hash": summary["state_hash"],
"selected_step": 10000,
"summary_sha256": sha(folder / "summary.json"),
"completion_sha256": sha(folder / "completion.json"),
}
)
# Preserve byte identity of a pre-analysis aggregate on repeated finalization.
aggregate = root / "raw/optimization_summaries.json"
if aggregate.exists():
require(read(aggregate) == summaries, "Existing statistics aggregate differs from states")
else:
write_json(aggregate, summaries)
trajectory_path = root / "raw/optimization_trajectories.json"
if trajectory_path.exists():
require(read(trajectory_path) == trajectories, "Existing full trajectory archive changed")
else:
write_json(trajectory_path, trajectories)
write_json(
root / "checkpoints/index.json",
{
"n_states": 136,
"states": rows,
"distribution": "Metadata only; GPU checkpoint weights remain in local outputs",
},
)
def finalize(root, docs):
root, docs = Path(root).resolve(), Path(docs).resolve()
validate(root, docs)
checkpoint_index(root)
sources = source_snapshot(root, Path(__file__).resolve().parents[1])
write_json(
root / "audit/finalizer_preservation_recheck.json",
{
"status": "PASS",
"old_files_unchanged": 11285,
"previous_seal_sha256": sha(root / "audit/previous_experiment_seal.json"),
"protected_media_bytes_reopened": False,
},
)
docs.mkdir(parents=True, exist_ok=True)
for path in sorted(root.iterdir()):
if path.is_file() and (
path.suffix in (".json", ".md", ".sh", ".txt", ".patch")
or path.name == "source_snapshot.tar.gz"
):
if path.name not in ("artifact_manifest.json", "integrity.json"):
shutil.copy2(path, docs / path.name)
for sub in ("figures",):
shutil.copytree(root / sub, docs / sub, dirs_exist_ok=True)
(docs / "checkpoints").mkdir(exist_ok=True)
shutil.copy2(root / "checkpoints/index.json", docs / "checkpoints/index.json")
for sub in ("raw", "audit"):
for path in sorted((root / sub).rglob("*")):
if not path.is_file() or path.suffix not in (
".json",
".jsonl",
".csv",
".txt",
".log",
".patch",
".xml",
):
continue
target = docs / path.relative_to(root)
if sub == "audit" and path.suffix == ".json" and path.stat().st_size <= 131072:
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(path, target)
else:
write_gzip(path, Path(str(target) + ".gz"))
(docs / "ARCHIVE_CONTENTS.md").write_text(
"# Portable numeric archive\n\n"
"The archive contains all numeric query/context rows, complete 136-state trajectories "
"(trace, full-objective checkpoints and fixed-final summaries), protocol, "
"source snapshot, "
"audits and the 136-state hash index. GPU weights, predictions and datasets "
"stay in outputs.\n\n"
"From the repository root, reproduce primary statistics without media or weights:\n\n"
f"`.venv/bin/python scripts/analyze_observability_supervision.py --root {docs} "
"--output /tmp/context-geometry-statistics-reproduction`\n\n"
"The statistics loader reads compressed raw directly. Historical chronology is evidence "
"of the original run; copying files does not recreate historical timestamps.\n"
)
files = {
str(p.relative_to(root)): sha(p)
for p in sorted(root.rglob("*"))
if p.is_file() and p.name not in ("artifact_manifest.json", "integrity.json")
}
write_json(root / "artifact_manifest.json", files)
integrity = {
"status": "PASS",
"experiment": EXPERIMENT,
"checkpoint_states_indexed": 136,
"old_files_unchanged": 11285,
"source_files_archived": len(sources),
"checkpoint_weights_copied_to_docs": False,
"final_holdout_touched": False,
"new_carrier_trained": False,
"dynamic_ttt_run": False,
"artifact_manifest_sha256": sha(root / "artifact_manifest.json"),
}
write_json(root / "integrity.json", integrity)
for name in ("artifact_manifest.json", "integrity.json"):
shutil.copy2(root / name, docs / name)
require(
not list(docs.rglob("*.pt")) and not list(docs.rglob("*.npz")),
"Weights/prediction arrays must not enter docs",
)
write_json(
docs / "report_manifest.json",
{
str(p.relative_to(docs)): sha(p)
for p in sorted(docs.rglob("*"))
if p.is_file() and p.name != "report_manifest.json"
},
)
return integrity
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", required=True)
parser.add_argument("--docs", required=True)
args = parser.parse_args()
print(json.dumps(finalize(args.root, args.docs), indent=2))