diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py new file mode 100644 index 0000000..45cbd64 --- /dev/null +++ b/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"