EscapistsReintroduced/docs/evidence/observer-chat-2026-10-13/archive.py

#!/usr/bin/env python3
"""Archive explicitly selected, stopped observer/chat native runs.

Usage: python3 docs/evidence/observer-chat-2026-10-13/archive.py a01 [a02 ...]
Inputs are /tmp/erri-observer-chat-source-TAG and /tmp/erri-observer-chat-run-TAG.
Existing per-run evidence is immutable. Every selected run must have stopped with
confirmed cleanup and unchanged launch sources before any archive is created.
This script does not launch an engine, submit chat, or contact a server.
"""
import argparse
from collections import Counter
import hashlib
import json
from pathlib import Path
import re
import sys

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[2]
sys.path.insert(0, str(ROOT / "tools"))
from native_provenance import archive_source, encoded, read_launch
from native_report import analyze

RELEVANT = (
    "addons/core/Scripts/Game/ERR/00_Core/ERR_Observer.c",
    "addons/core/Scripts/Game/ERR/00_Core/ERR_ObserverTargets.c",
    "addons/core/Scripts/Game/ERR/00_Core/ERR_ObserverPlayerView.c",
    "addons/core/Scripts/Game/ERR/00_Core/ERR_PlayerMedical.c",
    "addons/core/Scripts/Game/ERR/00_Core/ERR_UnconsciousCameraProbe.c",
    "addons/core/Prefabs/ERR/SpectatorCamera.et",
    "addons/core/Configs/ERR/ObserverInput.conf",
    "addons/core/Configs/ERR/Tests/LegacySites.conf",
)
LOGS = (
    "console.log", "script.log", "error.log", "runner.json",
    "external-addons-inputs.json", "headless-launch.log",
)
SCOPE = (
    "Local Workbench observer/chat fixture; read its assertion labels and source "
    "for the exact exercised paths. Native action injection and scripted widget "
    "operations do not qualify physical keyboard input, native administrator "
    "authentication, a remote client, or a dedicated multiplayer session."
)


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def require(condition, message):
    if not condition:
        raise ValueError(message)


def write_new(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("xb") as stream:
        stream.write(data)
    require(path.read_bytes() == data, f"Archive copy differs: {path}")


def copy(source, destination):
    require(source.is_file() and not source.is_symlink(), f"Missing or symlinked input: {source}")
    write_new(destination, source.read_bytes())


def preflight(tag):
    profile = Path("/tmp/erri-observer-chat-run-" + tag)
    source = Path("/tmp/erri-observer-chat-source-" + tag)
    sessions = list((profile / "logs").glob("logs_*"))
    require(len(sessions) == 1, f"Expected one native session in {profile}")
    session = sessions[0]
    manifest, metadata, runner = read_launch(session)
    require(runner.get("cleanup_confirmed") is True, f"Cleanup is not confirmed: {tag}")
    require(runner.get("cleanup_report", {}).get("idle_verified") is True,
            f"Disposable profile idle state is not confirmed: {tag}")
    require(runner.get("external_addons_unchanged_after_run") is True,
            f"External addon stability is not confirmed: {tag}")
    manifest_bytes, provenance, launched = archive_source(session, source)
    require(launched and metadata["source_root"] == str(source), f"Launch source root differs: {tag}")
    for name in RELEVANT:
        require(digest(source / name) == manifest.get(name), f"Relevant source differs from launch: {name}")
    for name in LOGS:
        path = session / name
        require(path.is_file() and not path.is_symlink(), f"Missing or symlinked native log: {path}")
    require(not (HERE / tag).exists(), f"Refusing to overwrite archived run: {tag}")
    return profile, source, session, manifest, manifest_bytes, provenance, runner


def archive(tag, prepared):
    profile, source, session, manifest, manifest_bytes, provenance, runner = prepared
    destination = HERE / tag
    destination.mkdir(exist_ok=False)
    for name in LOGS:
        copy(session / name, destination / (name + ".txt"))
    for name in ("launch-source-sha256.json", "launch-source.json"):
        copy(session / name, destination / name)
    write_new(destination / "source-sha256.json", manifest_bytes)
    write_new(destination / "source-provenance.json", encoded(provenance))
    matches = {}
    for name in RELEVANT:
        copy(source / name, destination / "source" / name)
        current = ROOT / name
        matches[name] = {
            "launch_sha256": manifest[name],
            "archived_sha256": digest(destination / "source" / name),
            "repository_exists_at_archive": current.is_file(),
            "repository_matches_launch_at_archive": current.is_file() and digest(current) == manifest[name],
        }
    write_new(destination / "relevant-source-matches.json", encoded(matches))
    for capture in sorted([*session.glob("err-*.png"), *session.glob("err-*.bmp")]):
        copy(capture, destination / capture.name)
    for name in ("fixture-inputs.json", "production-sources.json"):
        if (profile / name).exists():
            copy(profile / name, destination / name)
    runner_output = Path("/tmp/erri-observer-chat-" + tag + "-runner.log")
    if runner_output.exists():
        copy(runner_output, destination / "runner-output.log.txt")
    report = analyze((destination / "console.log.txt").read_text(errors="replace"),
                     str((destination / "console.log.txt").relative_to(ROOT)))
    report["scope"] = SCOPE
    write_new(destination / "report.json", encoded(report))
    hashes = {str(path.relative_to(destination)): digest(path)
              for path in sorted(destination.rglob("*")) if path.is_file()}
    write_new(destination / "artifact-sha256.json", encoded(hashes))
    print(f"Archived {tag}: {runner['result']}; {report['assertions_passed']} assertions; cleanup verified")


def normalized(errors):
    return Counter(re.sub(r"\s+", " ", re.sub(r"^\d\d:\d\d:\d\d\.\d+\s+", "", line))
                   for line in errors)


def summarize():
    runs, manifests, errors = {}, {}, {}
    folders = sorted((path for path in HERE.iterdir()
                      if path.is_dir() and re.fullmatch(r"a[0-9]{2,}", path.name)),
                     key=lambda path: int(path.name[1:]))
    for folder in folders:
        manifest, metadata, runner = read_launch(folder, "runner.json.txt", "console.log.txt")
        hashes = json.loads((folder / "artifact-sha256.json").read_text())
        for name, expected in hashes.items():
            require(digest(folder / name) == expected, f"Archived artifact differs: {folder.name}/{name}")
        require((folder / "source-sha256.json").read_bytes() ==
                (folder / "launch-source-sha256.json").read_bytes(),
                f"Archived and launch manifests differ: {folder.name}")
        report = json.loads((folder / "report.json").read_text())
        manifests[folder.name] = manifest
        errors[folder.name] = normalized(report["engine_errors"])
        runs[folder.name] = {
            "launch_id": metadata["launch_id"],
            "runner_result": runner["result"],
            "cleanup_confirmed": runner["cleanup_confirmed"],
            "idle_verified": runner["cleanup_report"]["idle_verified"],
            "external_addons_unchanged_after_run": runner["external_addons_unchanged_after_run"],
            "assertions_passed": report["assertions_passed"],
            "smoke_completions": report["smoke_completions"],
            "script_or_runtime_failure_count": len(report["script_or_runtime_failures"]),
            "engine_error_count": len(report["engine_errors"]),
            "artifact_manifest_sha256": digest(folder / "artifact-sha256.json"),
        }
    changes, comparisons = {}, {}
    tags = list(manifests)
    for previous, current in zip(tags, tags[1:]):
        one, two = manifests[previous], manifests[current]
        key = previous + "_to_" + current
        changes[key] = sorted(name for name in set(one) | set(two) if one.get(name) != two.get(name))
        comparisons[key] = {
            "new_signatures": sorted(set(errors[current]) - set(errors[previous])),
            "removed_signatures": sorted(set(errors[previous]) - set(errors[current])),
            "additional_occurrences": dict(errors[current] - errors[previous]),
        }
    report = {
        "scope": SCOPE,
        "runs": runs,
        "changed_launch_sources": changes,
        "engine_error_comparison": comparisons,
        "provenance": (
            "Original pre-launch manifest, stopped console binding, unchanged frozen addon bytes, "
            "confirmed cleanup/idle state and unchanged external addon inputs are verified. "
            "Per-run artifact hashes bind copied raw logs, relevant sources, reports and captures. "
            "Current repository comparisons describe archive time only."
        ),
    }
    (HERE / "run-summary.json").write_bytes(encoded(report))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("runs", nargs="+", help="Stopped run tags to append, such as a01 a02")
    args = parser.parse_args()
    if len(set(args.runs)) != len(args.runs):
        parser.error("Run tags must be unique")
    for tag in args.runs:
        if not re.fullmatch(r"a[0-9]{2,}", tag):
            parser.error("Expected a run tag such as a01")
    # Validate all selected runs before creating any archive; running sessions fail here.
    prepared = [(tag, preflight(tag)) for tag in args.runs]
    for tag, inputs in prepared:
        archive(tag, inputs)
    summarize()


if __name__ == "__main__":
    main()