Nexum-Expanded/runtime/tests/tooling/test_engineering_proofs.py

"""Engineering workflow proofs: restoration digest, disclosure gate, triage evidence."""

from __future__ import annotations

import hashlib
import subprocess
from pathlib import Path

import pytest

from nexum_runtime.tooling.engineering import (
    DisclosureStore,
    PatchStore,
    ReproductionStore,
    TriageStore,
)


def _run_command(
    command: str, workdir: str, timeout_s: float
) -> tuple[bool, str, str, int | None]:
    completed = subprocess.run(
        command,
        shell=True,
        cwd=workdir,
        capture_output=True,
        text=True,
        timeout=timeout_s,
        check=False,
    )
    return (
        completed.returncode == 0,
        completed.stdout,
        completed.stderr,
        completed.returncode,
    )


def _chain(tmp_path: Path) -> tuple[PatchStore, str, str]:
    workspace = tmp_path
    (workspace / "app.txt").write_text("before\n", encoding="utf-8")
    session = "proof-session"
    reproductions = ReproductionStore(workspace)
    reproduction, _stdout, _stderr = reproductions.run(
        command="grep -q before app.txt",
        working_directory=".",
        snapshot_paths=("app.txt",),
        timeout_s=30.0,
        session_id=session,
        run_command=_run_command,
    )
    assert reproduction.exit_code == 0
    triage = TriageStore(workspace).create(
        reproduction_ids=(reproduction.reproduction_id,),
        observed_facts=("the file contains the reproduced value",),
        hypotheses=("the value requires a verified update",),
        severity="low",
        confidence=0.9,
        next_actions=("apply an exact-state patch and verify",),
        session_id=session,
    )
    patches = PatchStore(workspace)
    patch = patches.begin(
        triage_id=triage.triage_id, paths=("app.txt",), session_id=session
    )
    expected = hashlib.sha256(b"before\n").hexdigest()
    patch, error = patches.apply(
        patch.patch_id,
        (
            {
                "operation": "replace",
                "path": "app.txt",
                "expected_sha256": expected,
                "old_text": "before",
                "new_text": "after",
            },
        ),
        session_id=session,
    )
    assert not error
    patch, verification, _out, _err = patches.verify(
        patch.patch_id,
        command="grep -q after app.txt",
        working_directory=".",
        timeout_s=30.0,
        session_id=session,
        run_command=_run_command,
    )
    assert patch.status == "verified"
    assert verification.exit_code == 0
    patch = patches.commit(patch.patch_id, session_id=session)
    assert patch.status == "committed"
    return patches, patch.patch_id, triage.triage_id


def test_rollback_binds_restoration_proof(tmp_path: Path) -> None:
    patches, patch_id, _triage_id = _chain(tmp_path)

    rolled_back = patches.rollback(patch_id, session_id="proof-session")

    assert rolled_back.status == "rolled_back"
    assert rolled_back.restoration_sha256
    restored_digest = hashlib.sha256(
        (tmp_path / "app.txt").read_bytes()
    ).hexdigest()
    assert restored_digest == hashlib.sha256(b"before\n").hexdigest()
    reloaded = patches.get(patch_id, session_id="proof-session")
    assert reloaded.restoration_sha256 == rolled_back.restoration_sha256
    reproduction, _stdout, _stderr = ReproductionStore(tmp_path).run(
        command="grep -q before app.txt",
        working_directory=".",
        snapshot_paths=("app.txt",),
        timeout_s=30.0,
        session_id="proof-session",
        run_command=_run_command,
    )
    assert reproduction.exit_code == 0


def test_public_disclosure_requires_verified_patch(tmp_path: Path) -> None:
    patches, patch_id, triage_id = _chain(tmp_path)
    disclosures = DisclosureStore(tmp_path)

    record = disclosures.create(
        triage_id=triage_id,
        title="Verified correction",
        summary="A reproduced value was changed through an exact-state transaction.",
        impact="The workspace carries the verified value.",
        remediation="Retain the verification receipt.",
        audience="public",
        patch_id=patch_id,
        session_id="proof-session",
    )
    assert record.artifact_sha256

    patches.rollback(patch_id, session_id="proof-session")
    with pytest.raises(ValueError, match="verified or committed"):
        disclosures.create(
            triage_id=triage_id,
            title="Rolled back state",
            summary="The patch was rolled back.",
            impact="No verified repair remains.",
            remediation="Re-apply and verify before disclosure.",
            audience="public",
            patch_id=patch_id,
            session_id="proof-session",
        )
    maintainer = disclosures.create(
        triage_id=triage_id,
        title="Internal note",
        summary="Rollback was proven against original snapshots.",
        impact="Original bytes restored.",
        remediation="None required.",
        audience="maintainer",
        patch_id="",
        session_id="proof-session",
    )
    assert maintainer.disclosure_id.startswith("dis_")


def test_public_disclosure_rejects_state_drift_after_verification(
    tmp_path: Path,
) -> None:
    _patches, patch_id, triage_id = _chain(tmp_path)
    disclosures = DisclosureStore(tmp_path)
    (tmp_path / "app.txt").write_text("drifted\n", encoding="utf-8")

    with pytest.raises(RuntimeError, match="changed after patch verification"):
        disclosures.create(
            triage_id=triage_id,
            title="Stale repair",
            summary="The covered files changed after verification.",
            impact="The prior verification no longer describes current bytes.",
            remediation="Re-verify the exact current state before disclosure.",
            audience="public",
            patch_id=patch_id,
            session_id="proof-session",
        )


def test_triage_status_exposes_bound_reproduction_evidence(tmp_path: Path) -> None:
    workspace = tmp_path
    (workspace / "state.txt").write_text("one\n", encoding="utf-8")
    reproductions = ReproductionStore(workspace)
    first, _stdout, _stderr = reproductions.run(
        command="printf two > state.txt",
        working_directory=".",
        snapshot_paths=("state.txt",),
        timeout_s=30.0,
        session_id="proof-session",
        run_command=_run_command,
    )
    triage_store = TriageStore(workspace)
    triage = triage_store.create(
        reproduction_ids=(first.reproduction_id,),
        observed_facts=("the command rewrote the file",),
        hypotheses=("state drift is contained",),
        severity="informational",
        confidence=0.8,
        next_actions=("compare the before and after digests",),
        session_id="proof-session",
    )

    status = triage_store.status(triage.triage_id, session_id="proof-session")

    assert status["triage_id"] == triage.triage_id
    assert len(status["evidence"]) == 1
    evidence = status["evidence"][0]
    assert evidence["id"] == first.reproduction_id
    assert evidence["exit_code"] == 0
    assert evidence["state_drift_observed"] is True
    assert status["state_drift_observed"] is True