test(issue-5): phase 2b fail-closed preconditions before auto-completion exists

Pins what the system refuses to do today, so nothing added later can weaken it
quietly. X-01 records the property the whole phase leans on: a job with no
completion manifest is already held, which is why an agent that declares nothing
stays on the explicit path without a single new branch.

X-03 and X-04 record the layering decision and its evidence. The engine compares
an expected attempt against a record field that does not exist, falling back to
the loop-iteration counter, which means something else. So auto-completion does
not use that guard; the attempt check stays where attempt identity actually
lives, comparing observed evidence against the dispatch record. X-03 will fail
if a record ever starts carrying an attempt id, which is the moment to revisit
the decision.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

tests/test_issue5_phase2b_autocomplete.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""Issue #5 Phase 2b — flag-gated auto-completion.
+
+Characterize first, as in 2a: the interesting property here is what the system
+*refuses* to do, so the refusals are pinned before anything is added that could
+weaken them.
+
+Sections
+  1. X-01..X-05  the fail-closed preconditions as they stand today.
+  2. finalize    the completion manifest: written by the agent, re-derived here.
+  3. autocomplete the gate, and everything that must stop it.
+  4. notifier    durable pending notification and restart recovery.
+  5. flags       default off, evidence-backed allowlist, per-job opt-in.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WATCHER_FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "watcher"
+
+
+def _reg(tmp_path) -> str:
+    d = tmp_path / ".mam" / "jobs"
+    d.mkdir(parents=True, exist_ok=True)
+    return str(d)
+
+
+def _make_job(registry_dir, **over):
+    from lib_py.delegation import registry
+    return registry.register_job(over.pop("prompt", "p"), registry_dir=registry_dir, **over)
+
+
+def _load(registry_dir, job_id):
+    from lib_py.delegation import mqtt_common
+    return mqtt_common.load_job(job_id, registry_dir)
+
+
+def _fingerprint(path: Path) -> dict:
+    out = {}
+    for p in sorted(Path(path).rglob("*")):
+        if p.is_file():
+            out[str(p.relative_to(path))] = hashlib.sha256(p.read_bytes()).hexdigest()
+    return out
+
+
+def _scenario(name="finalized_candidate"):
+    return json.loads((WATCHER_FIXTURES / (name + ".json")).read_text(encoding="utf-8"))
+
+
+@pytest.fixture(autouse=True)
+def _default_flags(monkeypatch):
+    """Auto-completion is off unless a test turns it on. That is the product
+    default too, and every test that does not opt in is therefore also a check
+    that the default really is off."""
+    monkeypatch.delenv("MAM_AUTO_COMPLETE", raising=False)
+    monkeypatch.delenv("MAM_AUTO_COMPLETE_BACKENDS", raising=False)
+    monkeypatch.setenv("MAM_ENGINE_CAS", "1")
+
+
+# ==============================================================================
+# 1. X-01..X-05 — the fail-closed preconditions, before anything is added
+# ==============================================================================
+
+def test_x01_a_job_without_a_manifest_is_never_approved():
+    """The property that makes unsupported agents safe without any new branch."""
+    import copy
+    from lib_py.loop import policy
+
+    scenario = _scenario()
+    assert policy.decide(scenario).decision == "candidate_approved"
+    stripped = copy.deepcopy(scenario)
+    for item in stripped["timeline"]:
+        (item.get("payload") or {}).pop("report_manifest", None)
+    stripped["filesystem"]["manifest_committed"] = False
+    decision = policy.decide(stripped)
+    assert decision.decision == "hold" and "manifest" in decision.reason
+
+
+def test_x02_no_production_code_writes_a_completion_manifest_yet():
+    """Pins the starting point: the protocol is entirely new in phase 2b."""
+    hits = []
+    for path in sorted((SKILLS / "lib_py").rglob("*.py")):
+        if path.name in ("finalize.py", "autocomplete.py"):
+            continue
+        text = path.read_text(encoding="utf-8")
+        if "report-manifest" in text:
+            hits.append(path.name)
+    assert not hits, "manifest writers already exist: {}".format(hits)
+
+
+def test_x03_the_record_carries_iteration_not_an_attempt_id(mam_sandbox, tmp_path, monkeypatch):
+    """Why auto-completion does not use the engine's expected_attempt.
+
+    The engine compares expected_attempt against `attempt_id` falling back to
+    `iteration`. A record has only the latter, and the two mean different things,
+    so the attempt guard is applied where attempt identity actually lives: the
+    dispatch record in the shadow store, compared against observed evidence."""
+    monkeypatch.chdir(mam_sandbox)
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    record = _load(reg, job_id)
+    assert "iteration" in record
+    assert "attempt_id" not in record, (
+        "if this ever changes, revisit the layering decision in the phase 2b report"
+    )
+
+
+def test_x04_policy_rejects_evidence_from_a_superseded_attempt():
+    """The attempt guard that auto-completion actually relies on."""
+    import copy
+    from lib_py.loop import policy
+
+    scenario = _scenario()
+    superseded = copy.deepcopy(scenario)
+    superseded["dispatch_context"]["attempt_id"] = 2   # evidence still says 1
+    decision = policy.decide(superseded)
+    assert decision.decision == "hold" and "attempt_id" in decision.reason
+
+
+def test_x05_the_explicit_event_path_is_untouched_by_default(mam_sandbox, tmp_path, monkeypatch):
+    """An explicitly completed job looks exactly as it always did."""
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="running")
+    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
+    record = _load(reg, job_id)
+    assert record["status"] == "completed"
+    assert "completed_by" not in record, "an explicit completion carries no auto marker"