diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 512b635..0a65d16 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -205,6 +205,15 @@ def main(argv=None) -> int: help="force retain=True (auto for completed/error)") parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR) parser.add_argument("--attempts", type=int, default=3) + parser.add_argument("--finalize-report", default=None, metavar="PATH", + help="atomically finalize this report and declare its " + "completion manifest before publishing") + parser.add_argument("--side-effect", action="append", default=[], dest="side_effects", + metavar="KIND=REFERENCE", + help="evidence for an external side effect this job performed") + parser.add_argument("--attempt-id", type=int, default=None, + help="attempt this finalization belongs to (default: the " + "job's current iteration)") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) @@ -227,6 +236,33 @@ def main(argv=None) -> int: logger.error("no job to publish for (unknown --job or no pending job)") return 1 + # Finalize before publishing: a `completed` event whose report is still being + # written would invite exactly the partial-report adoption the contract + # forbids. Failing here fails the publish, on purpose. + if args.finalize_report: + from lib_py.loop import finalize # local import: optional dependency of this CLI + + try: + side_effects = finalize.parse_side_effects(args.side_effects) + except ValueError as exc: + logger.error("%s", exc) + return 1 + attempt_id = args.attempt_id + if attempt_id is None: + try: + attempt_id = load_job(job_id, args.registry_dir).get("iteration", 1) + except FileNotFoundError: + attempt_id = 1 + try: + manifest = finalize.finalize_report( + args.finalize_report, job_id, attempt_id, + side_effects=side_effects, declared_by="agent") + except (OSError, FileNotFoundError) as exc: + logger.error("could not finalize report: %s", exc) + return 1 + logger.info("finalized %s (sha256 %s...)", manifest["report_path"], + manifest["sha256"][:12]) + try: prepared = prepare_event( job_id=job_id, diff --git a/.agents/skills/lib_py/loop/finalize.py b/.agents/skills/lib_py/loop/finalize.py new file mode 100644 index 0000000..3445502 --- /dev/null +++ b/.agents/skills/lib_py/loop/finalize.py @@ -0,0 +1,158 @@ +"""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 '=', 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 `.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) diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py index 45cbd64..818a5d3 100644 --- a/tests/test_issue5_phase2b_autocomplete.py +++ b/tests/test_issue5_phase2b_autocomplete.py @@ -138,3 +138,136 @@ def test_x05_the_explicit_event_path_is_untouched_by_default(mam_sandbox, tmp_pa record = _load(reg, job_id) assert record["status"] == "completed" assert "completed_by" not in record, "an explicit completion carries no auto marker" + + +# ============================================================================== +# 2. finalize — declared by the agent, re-derived by the reader +# ============================================================================== + +def _report(tmp_path, body="done\n", staged=False): + d = tmp_path / "claude-reports" + d.mkdir(parents=True, exist_ok=True) + path = d / "report-final.md" + (path.with_name(path.name + ".tmp") if staged else path).write_text(body, encoding="utf-8") + return path + + +def test_finalize_promotes_a_staged_report_atomically(tmp_path): + """A reader must never see a half-written report as finalized.""" + from lib_py.loop import finalize + + path = _report(tmp_path, "final body\n", staged=True) + assert not path.exists() and path.with_name(path.name + ".tmp").exists() + manifest = finalize.finalize_report(path, "job1", 1) + assert path.exists() and not path.with_name(path.name + ".tmp").exists() + assert manifest["sha256"] == finalize.sha256_of(path) + + +def test_finalize_verifies_against_the_file_not_the_declaration(tmp_path): + """The hash is recomputed, so a forged manifest cannot certify itself.""" + from lib_py.loop import finalize + + path = _report(tmp_path) + finalize.finalize_report(path, "job1", 1) + assert finalize.verify_manifest(path.parent, "job1").ok + + manifest_file = finalize.manifest_in(path.parent) + forged = json.loads(manifest_file.read_text(encoding="utf-8")) + forged["sha256"] = "0" * 64 + manifest_file.write_text(json.dumps(forged), encoding="utf-8") + verdict = finalize.verify_manifest(path.parent, "job1") + assert not verdict.ok and "changed after finalization" in verdict.reason + + +def test_finalize_is_invalidated_by_a_rewritten_report(tmp_path): + """A new body needs a new finalization; the old hash is void immediately.""" + from lib_py.loop import finalize + + path = _report(tmp_path, "first\n") + finalize.finalize_report(path, "job1", 1) + assert finalize.verify_manifest(path.parent, "job1").ok + path.write_text("rewritten\n", encoding="utf-8") + verdict = finalize.verify_manifest(path.parent, "job1") + assert not verdict.ok and "changed after finalization" in verdict.reason + + +@pytest.mark.parametrize("kwargs,fragment", [ + ({"job_id": "other"}, "belongs to job"), + ({"expected_attempt": 9}, "belongs to attempt"), +]) +def test_finalize_rejects_a_manifest_for_a_different_job_or_attempt(tmp_path, kwargs, fragment): + from lib_py.loop import finalize + path = _report(tmp_path) + finalize.finalize_report(path, "job1", 1) + args = {"job_id": "job1"} + args.update(kwargs) + verdict = finalize.verify_manifest(path.parent, **args) + assert not verdict.ok and fragment in verdict.reason + + +def test_finalize_requires_evidence_for_every_declared_side_effect(tmp_path): + """Partial evidence is no evidence: the work is not known to be complete.""" + from lib_py.loop import finalize + + path = _report(tmp_path) + finalize.finalize_report(path, "job1", 1, + side_effects=finalize.parse_side_effects(["git_push=abc123"])) + ok = finalize.verify_manifest(path.parent, "job1", expected_artifacts=["git_push"]) + assert ok.ok + missing = finalize.verify_manifest(path.parent, "job1", + expected_artifacts=["git_push", "api_call"]) + assert not missing.ok and "api_call" in missing.reason + + +def test_finalize_rejects_a_malformed_side_effect_pair(): + from lib_py.loop import finalize + with pytest.raises(ValueError): + finalize.parse_side_effects(["nokey"]) + with pytest.raises(ValueError): + finalize.parse_side_effects(["kind="]) + + +def test_finalize_missing_manifest_is_a_refusal_not_a_crash(tmp_path): + from lib_py.loop import finalize + d = tmp_path / "empty" + d.mkdir() + verdict = finalize.verify_manifest(d, "job1") + assert not verdict.ok and "no completion manifest" in verdict.reason + + +def test_publish_event_finalizes_before_it_publishes(mam_sandbox, tmp_path, monkeypatch): + """The CLI surface an agent actually calls.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import publish_event + from lib_py.loop import finalize + + reg = _reg(tmp_path) + job_id = _make_job(reg) + record = _load(reg, job_id) + record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False} + Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8") + + path = _report(tmp_path, "the report\n", staged=True) + rc = publish_event.main([ + "--registry-dir", reg, "--job", job_id, "--event", "completed", + "--attempts", "1", "--finalize-report", str(path), + "--side-effect", "git_push=deadbeef", + ]) + assert rc == 2, "the broker is unreachable, but finalization still happened" + assert path.exists(), "the staged report was promoted" + verdict = finalize.verify_manifest(path.parent, job_id) + assert verdict.ok + assert verdict.manifest["side_effects"][0]["kind"] == "git_push" + + +def test_publish_event_reports_a_bad_side_effect_instead_of_publishing(mam_sandbox, tmp_path, monkeypatch): + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import publish_event + reg = _reg(tmp_path) + job_id = _make_job(reg) + path = _report(tmp_path) + rc = publish_event.main([ + "--registry-dir", reg, "--job", job_id, "--event", "completed", + "--finalize-report", str(path), "--side-effect", "malformed", + ]) + assert rc == 1