multi-agent-mux/.agents/skills/lib_py/loop/finalize.py

"""Report finalization: the completion manifest.

Who writes this matters more than what is in it. If the watcher computed the
hash itself, the manifest would attest only that the watcher saw a stable file
-- self-certification, which is exactly the evidence shape a reviewer rejected
earlier in this issue. So the side that knows the work finished declares it, and
the reader independently re-derives the hash rather than trusting the number in
the file.

Consequence, and it is the useful one: an agent that declares nothing leaves no
manifest, a missing manifest already holds, and the explicit-event path is
therefore preserved for unsupported agents without any special case.

    writing            report-final.md.tmp is being written
    stable_candidate   the file stopped changing (a filter, never an authority)
    finalized          atomically renamed, and this manifest exists
"""
from __future__ import annotations

import hashlib
import json
import os
import tempfile
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

MANIFEST_NAME = "report-manifest.json"
SCHEMA_VERSION = "1.0.0"


def manifest_path(report_path) -> Path:
    """The manifest sits beside the report it describes."""
    return Path(report_path).parent / MANIFEST_NAME


def manifest_in(report_dir) -> Path:
    return Path(report_dir) / MANIFEST_NAME


def sha256_of(path) -> str:
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _atomic_write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix="." + path.name + ".", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(text)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp, path)
    except BaseException:
        if os.path.exists(tmp):
            os.unlink(tmp)
        raise


def parse_side_effects(raw: Optional[List[str]]) -> List[Dict[str, str]]:
    """`kind=reference` pairs from the command line into evidence entries."""
    out = []
    for item in raw or []:
        kind, _, reference = str(item).partition("=")
        kind = kind.strip()
        if not kind or not reference.strip():
            raise ValueError("side effect must be '<kind>=<reference>', got {!r}".format(item))
        out.append({"kind": kind, "reference": reference.strip(),
                    "verified_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())})
    return out


def finalize_report(report_path, job_id, attempt_id, side_effects=None,
                    declared_by="agent") -> Dict[str, Any]:
    """Promote a report to finalized and declare the manifest.

    If `<report>.tmp` exists it is renamed into place first, so a reader can
    never observe a half-written report as finalized. The rename is what makes
    the transition atomic; the manifest is what makes it attributable.
    """
    report_path = Path(report_path)
    staged = report_path.with_name(report_path.name + ".tmp")
    if staged.exists():
        os.replace(str(staged), str(report_path))
    if not report_path.exists():
        raise FileNotFoundError("no report to finalize at {}".format(report_path))

    manifest = {
        "schema_version": SCHEMA_VERSION,
        "job_id": job_id,
        "attempt_id": attempt_id,
        "report_path": str(report_path),
        "sha256": sha256_of(report_path),
        "finalized_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "side_effects": list(side_effects or []),
        "declared_by": declared_by,
    }
    _atomic_write(manifest_path(report_path), json.dumps(manifest, indent=2,
                                                         ensure_ascii=False) + "\n")
    return manifest


class ManifestVerdict(object):
    __slots__ = ("ok", "reason", "manifest")

    def __init__(self, ok, reason, manifest=None):
        self.ok = ok
        self.reason = reason
        self.manifest = manifest


def verify_manifest(report_dir, job_id, expected_attempt=None,
                    expected_artifacts=None) -> ManifestVerdict:
    """Read a manifest and check it against the world, not against itself.

    The hash is recomputed from the file on disk. A report rewritten after the
    declaration therefore invalidates it, which is the contract: a new body needs
    a new finalization.
    """
    path = manifest_in(report_dir)
    if not path.exists():
        return ManifestVerdict(False, "no completion manifest")

    try:
        manifest = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        return ManifestVerdict(False, "manifest is unreadable: {}".format(exc))

    if manifest.get("schema_version") != SCHEMA_VERSION:
        return ManifestVerdict(False, "unsupported manifest schema {!r}".format(
            manifest.get("schema_version")))
    if manifest.get("job_id") != job_id:
        return ManifestVerdict(False, "manifest belongs to job {!r}".format(manifest.get("job_id")))
    if expected_attempt is not None and manifest.get("attempt_id") != expected_attempt:
        return ManifestVerdict(False, "manifest belongs to attempt {!r}, not {!r}".format(
            manifest.get("attempt_id"), expected_attempt))

    report = Path(manifest.get("report_path") or "")
    if not report.exists():
        return ManifestVerdict(False, "the finalized report is gone: {}".format(report))
    actual = sha256_of(report)
    if actual != manifest.get("sha256"):
        return ManifestVerdict(
            False, "report changed after finalization (declared {}, found {})".format(
                str(manifest.get("sha256"))[:12], actual[:12]))

    # A job that declared side effects must show evidence for every one of them.
    # Partial evidence is no evidence: the work is not known to be complete.
    declared = {e.get("kind") for e in manifest.get("side_effects") or []}
    missing = [a for a in (expected_artifacts or []) if a not in declared]
    if missing:
        return ManifestVerdict(False, "no side-effect evidence for {}".format(missing))

    return ManifestVerdict(True, "manifest verified", manifest)