multi-agent-mux/tests/test_issue5_phase2b_autocomplete.py

#!/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"


# ==============================================================================
# 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.delegation 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.delegation 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.delegation 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.delegation 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.delegation 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.delegation 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.delegation 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.delegation 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


# ==============================================================================
# 3. autocomplete — the gate, and everything that must stop it
# ==============================================================================

def _ready_job(tmp_path, mam_sandbox, *, backend="claude-code", opt_in=True,
               artifacts=None, body="report\n"):
    """A job in exactly the state that should auto-complete, plus its evidence."""
    from lib_py.delegation import finalize
    from lib_py.loop import store

    reg = _reg(tmp_path)
    job_id = _make_job(reg, agent=backend, expected_artifacts=artifacts or [])
    from lib_py.delegation import engine
    engine.transition(job_id, "claim", registry_dir=reg)
    if opt_in:
        engine.transition(job_id, "progress", registry_dir=reg,
                          fields={"auto_complete": True})

    scenario = _scenario()
    dispatch = scenario["dispatch_context"]
    dispatch["job_id"] = job_id
    for item in scenario["timeline"]:
        env = item.get("collector_attribution")
        if env:
            env["job_id"] = job_id
        # A non-herdr payload is a legitimate identifier source, so it has to
        # agree too. Leaving the fixture's original id here is a contradiction,
        # and the policy correctly refuses it.
        payload = item.get("payload") or {}
        if item.get("source_kind") != "herdr" and "job_id" in payload:
            payload["job_id"] = job_id
    store.record_dispatch(job_id, dispatch["attempt_id"], dispatch["agent_session_uuid"],
                          dispatch["pane_process_generation"], workspace=str(mam_sandbox))
    store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))

    report = _report(tmp_path, body)
    finalize.finalize_report(report, job_id, dispatch["attempt_id"],
                             side_effects=finalize.parse_side_effects(
                                 ["{}=ref".format(a) for a in (artifacts or [])]))
    return job_id, reg, scenario, report.parent


def test_gate_commits_when_every_condition_holds(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed, result.reason
    record = _load(reg, job_id)
    assert record["status"] == "completed"
    assert record["completed_by"] == "auto_complete", "an inferred terminal must be marked"
    assert record["pending_notification"]["event"] == "completed"


def test_g4_default_off_commits_nothing_and_leaves_the_registry_untouched(
    mam_sandbox, tmp_path, monkeypatch
):
    """G4: with the flag unset, the registry must not change at all."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
    before = _fingerprint(Path(reg))
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False
    assert "disabled" in result.reason
    assert _fingerprint(Path(reg)) == before, "the default must write nothing"


@pytest.mark.parametrize("backend", ["agy", "hermes", "opencode", "unknown-agent"])
def test_g9_a_backend_without_quiescence_evidence_never_auto_completes(
    mam_sandbox, tmp_path, monkeypatch, backend
):
    """Inferring 'finished' from stillness needs stillness to have been observed."""
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, backend=backend)
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False and "quiescent-state evidence" in result.reason
    assert _load(reg, job_id)["status"] == "running"


def test_the_allowlist_can_be_narrowed_but_never_widened(monkeypatch):
    from lib_py.loop import autocomplete

    monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude")
    assert autocomplete.allowed_backends() == ("claude",)
    monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude,hermes,opencode")
    assert autocomplete.allowed_backends() == ("claude",), "unobserved backends are dropped"


def test_a_job_that_did_not_opt_in_is_not_auto_completed(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, opt_in=False)
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False and "opt in" in result.reason


def test_g5_a_rewritten_report_stops_the_gate(mam_sandbox, tmp_path, monkeypatch):
    """G5: the manifest is checked against the file, so a late edit disqualifies."""
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
    (report_dir / "report-final.md").write_text("rewritten after finalization\n", encoding="utf-8")
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False and "changed after finalization" in result.reason
    assert _load(reg, job_id)["status"] == "running"


def test_g5_missing_side_effect_evidence_stops_the_gate(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete
    from lib_py.delegation import finalize

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox,
                                                   artifacts=["git_push"])
    # re-declare without the evidence the job said it would produce
    finalize.finalize_report(report_dir / "report-final.md", job_id, 1, side_effects=[])
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False and "git_push" in result.reason


def test_d7_a_policy_that_cannot_decide_holds(mam_sandbox, tmp_path, monkeypatch):
    """'Could not decide' is never 'done'."""
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.loop import autocomplete, policy

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)

    def explode(_scenario):
        raise RuntimeError("evidence is contradictory")

    monkeypatch.setattr(policy, "decide", explode)
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False and "policy error" in result.reason
    assert _load(reg, job_id)["status"] == "running"


def test_a_late_explicit_terminal_beats_the_gate(mam_sandbox, tmp_path, monkeypatch):
    """If the agent already reported, the boundary refuses and that is correct."""
    monkeypatch.chdir(mam_sandbox)
    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
    from lib_py.delegation import mqtt_common
    from lib_py.loop import autocomplete

    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error")
    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
                                   workspace=str(mam_sandbox), report_dir=report_dir)
    assert result.committed is False
    record = _load(reg, job_id)
    assert record["status"] == "error", "the explicit report wins"
    assert "completed_by" not in record


# ==============================================================================
# 4. notifier — the intent is durable, and recovery re-sends it
# ==============================================================================

def test_the_notification_intent_is_committed_with_the_state(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, notifier

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
    record = _load(reg, job_id)
    assert record["status"] == "completed"
    intent = notifier.pending(job_id, reg)
    assert intent and intent["event"] == "completed" and intent["seq"] == record["transition_seq"]


def test_flush_publishes_then_settles(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, notifier

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    engine.transition(job_id, "completed", registry_dir=reg, notify="completed")

    sent = []
    assert notifier.flush(job_id, reg, publish=lambda j, i, r: sent.append(i) or 0)
    assert len(sent) == 1 and sent[0]["event"] == "completed"
    assert notifier.pending(job_id, reg) is None
    # settling is idempotent: a second flush has nothing to do
    assert notifier.flush(job_id, reg, publish=lambda *a: 1 / 0)


def test_a_failed_send_leaves_the_intent_pending(mam_sandbox, tmp_path, monkeypatch):
    """An unsent notification that looks sent is worse than one that is retried."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, notifier

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    engine.transition(job_id, "completed", registry_dir=reg, notify="completed")

    assert notifier.flush(job_id, reg, publish=lambda *a: 2) is False
    assert notifier.pending(job_id, reg) is not None
    assert notifier.flush(job_id, reg, publish=lambda *a: (_ for _ in ()).throw(OSError())) is False
    assert notifier.pending(job_id, reg) is not None
    assert _load(reg, job_id)["status"] == "completed", "the state stays committed regardless"


def test_g7_restart_recovery_resends_every_owed_notification(mam_sandbox, tmp_path, monkeypatch):
    """G7: the crash window between committing and publishing."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, notifier

    reg = _reg(tmp_path)
    owed = []
    for _ in range(3):
        job_id = _make_job(reg)
        engine.transition(job_id, "claim", registry_dir=reg)
        engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
        owed.append(job_id)
    settled = _make_job(reg)
    engine.transition(settled, "claim", registry_dir=reg)

    assert sorted(notifier.jobs_with_pending(reg)) == sorted(owed)

    sent = []
    results = notifier.recover(reg, publish=lambda j, i, r: sent.append(j) or 0)
    assert all(results.values()) and sorted(sent) == sorted(owed)
    assert notifier.jobs_with_pending(reg) == [], "nothing is owed after recovery"