multi-agent-mux/tests/test_issue5_phase2a_loop.py

#!/usr/bin/env python3
"""Issue #5 Phase 2a — lib_py/loop core and shadow watcher.

Written in the order the plan fixes: characterize the current writers BEFORE
routing them through a CAS boundary, so the refactor has an oracle rather than a
promise. Phase 1 proved the value of that ordering; it also proved that a fake
which cannot express a failure hides it, so every double here models the real
contract (herdr's exit/stream split, the registry lock, the record on disk).

Sections
  1. X-01..X-04  current writer behaviour (must hold before and after P2A-01).
  2. engine      the single CAS transition boundary and its lock discipline.
  3. watcher     herdr agent wait wrapping, stream split, reconnect backoff.
  4. policy      decisions checked against the phase-0 scenario contracts.
  5. shadow      the runner that cannot write, proven by hashing the registry.
"""
from __future__ import annotations

import hashlib
import json
import os
import subprocess
import sys
import time
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"

CAS_MODES = ("0", "1")  # MAM_ENGINE_CAS: legacy unconditional merge / CAS engine


# ==============================================================================
# Helpers
# ==============================================================================

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
    job_id = registry.register_job(over.pop("prompt", "p"), registry_dir=registry_dir, **over)
    return job_id


def _load(registry_dir, job_id):
    from lib_py.delegation import mqtt_common
    return mqtt_common.load_job(job_id, registry_dir)


def _dir_fingerprint(path: Path) -> dict:
    """File name -> sha256 for every file under path. Detects any byte change."""
    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


@pytest.fixture(autouse=True)
def _cas_on(monkeypatch):
    """Default every test to the CAS engine unless it says otherwise."""
    monkeypatch.setenv("MAM_ENGINE_CAS", "1")


# ==============================================================================
# 1. X-01..X-04 — the writers as they behave today
# ==============================================================================

@pytest.mark.parametrize("mode", CAS_MODES)
def test_x01_update_job_status_merges_arbitrary_fields(mam_sandbox, tmp_path, monkeypatch, mode):
    """X-01: update_job_status merges whatever it is given, in both modes.

    Its signature, return value and side effects are load-bearing: phase 1's
    G-1..G-4 assert through it, so routing it through the engine must not change
    what a caller observes."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    out = mqtt_common.update_job_status(job_id, reg, status="running", iteration=4)
    assert out["status"] == "running"
    assert out["iteration"] == 4
    assert out["updated_at"], "updated_at must always be refreshed"
    assert _load(reg, job_id)["status"] == "running"


@pytest.mark.parametrize("mode", CAS_MODES)
def test_x02_pick_pending_claims_only_pending_and_only_once(mam_sandbox, tmp_path, monkeypatch, mode):
    """X-02: the claim path's real contract.

    pick_pending is not broken today: it filters on pending under the lock, so
    pending->running is already a correct swap. It is a *duplicate* of the
    chokepoint, not a hole in it. Routing it through the engine must preserve
    exactly this behaviour."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import registry

    reg = _reg(tmp_path)
    job_id = _make_job(reg, agent_session="herdr:s1")
    assert registry.pick_pending("herdr:s1", reg) == job_id
    assert _load(reg, job_id)["status"] == "running"
    # already claimed -> no second winner
    assert registry.pick_pending("herdr:s1", reg) is None
    # a different session never sees another session's work
    other = _make_job(reg, agent_session="herdr:s2")
    assert registry.pick_pending("herdr:s1", reg) is None
    assert registry.pick_pending("herdr:s2", reg) == other


@pytest.mark.parametrize("mode", CAS_MODES)
def test_x03_status_write_mirrors_into_the_audit_log(mam_sandbox, tmp_path, monkeypatch, mode):
    """X-03: a status change is mirrored to the audit log, in both modes.

    P2A-01 moves that mirror out of the lock; it must not lose it."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    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, status="completed")

    logged = mqtt_common.read_logged_status(job_id, mqtt_common.get_logs_dir())
    assert logged and logged.get("status") == "completed"
    events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
    changes = [e for e in events if e.get("event") == "status_changed"]
    assert changes, "a status change must appear in the audit log"
    assert changes[-1]["to"] == "completed"


@pytest.mark.parametrize("mode", CAS_MODES)
def test_x04_next_seq_and_register_are_unaffected(mam_sandbox, tmp_path, monkeypatch, mode):
    """X-04: the non-status writers keep their behaviour under either mode."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    assert _load(reg, job_id)["status"] == "pending"
    assert mqtt_common.next_seq(job_id, reg) == 1
    assert mqtt_common.next_seq(job_id, reg) == 2
    assert _load(reg, job_id)["last_seq"] == 2
    assert _load(reg, job_id)["status"] == "pending", "next_seq must not touch status"


# ==============================================================================
# 2. engine — the CAS table (state-transitions.md §3) and lock discipline
# ==============================================================================

def test_engine_pending_to_running_on_claim(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending")
    assert res.committed and res.from_status == "pending" and res.to_status == "running"


@pytest.mark.parametrize("request_", ["progress", "permission_required"])
def test_engine_non_terminal_requests_keep_running(mam_sandbox, tmp_path, monkeypatch, request_):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    res = engine.transition(job_id, request_, registry_dir=reg,
                            fields={"blocked_reason": "tool"})
    assert res.committed
    assert _load(reg, job_id)["status"] == "running", "must stay non-terminal"
    assert _load(reg, job_id)["blocked_reason"] == "tool"


@pytest.mark.parametrize("terminal", ["completed", "error", "cancelled"])
def test_engine_explicit_terminal_commits(mam_sandbox, tmp_path, monkeypatch, terminal):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    res = engine.transition(job_id, terminal, registry_dir=reg)
    assert res.committed and res.to_status == terminal


def test_engine_repeated_terminal_is_an_idempotent_noop(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    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)
    before = _load(reg, job_id)
    res = engine.transition(job_id, "completed", registry_dir=reg)
    assert res.committed is False
    assert "idempotent" in res.reason
    assert _load(reg, job_id)["transition_seq"] == before["transition_seq"], "no second write"


@pytest.mark.parametrize("late", ["error", "cancelled", "progress"])
def test_engine_refuses_to_reverse_a_terminal_and_audits_the_conflict(
    mam_sandbox, tmp_path, monkeypatch, late
):
    """A late or conflicting signal must never walk a terminal record backwards."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common
    from lib_py.delegation import engine
    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)

    res = engine.transition(job_id, late, registry_dir=reg, source="late-arrival")
    assert res.committed is False
    assert _load(reg, job_id)["status"] == "completed", "terminal must be preserved"
    events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
    refusals = [e for e in events if e.get("event") == "conflict_refused"]
    assert refusals, "a refused conflict must be auditable"
    assert refusals[-1]["current"] == "completed"


def test_engine_rejects_a_stale_attempt(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg, fields={"attempt_id": 2})
    res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=1)
    assert res.committed is False
    assert "stale attempt" in res.reason
    assert _load(reg, job_id)["status"] == "running"


def test_engine_deadline_timeout_becomes_error(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    res = engine.transition(job_id, "deadline_timeout", registry_dir=reg)
    assert res.committed and res.to_status == "error"


def test_engine_precondition_failure_does_not_write(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)          # now running
    res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending")
    assert res.committed is False and "precondition failed" in res.reason


def test_engine_transition_seq_is_monotonic(mam_sandbox, tmp_path, monkeypatch):
    """The audit mirror runs outside the lock, so ordering rides on this counter."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    seqs = []
    for req in ("claim", "progress", "progress", "completed"):
        engine.transition(job_id, req, registry_dir=reg)
        seqs.append(_load(reg, job_id)["transition_seq"])
    assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs), seqs


def test_g5_no_blocking_io_while_the_registry_lock_is_held(mam_sandbox, tmp_path, monkeypatch):
    """G5: the lock may cover a decision and one record write, nothing else.

    Watches the calls that can block on another process or the network and
    asserts none of them happen between acquiring and releasing the lock."""
    monkeypatch.chdir(mam_sandbox)
    import socket
    from lib_py.delegation import mqtt_common
    from lib_py.delegation import engine

    reg = _reg(tmp_path)
    job_id = _make_job(reg)

    held = {"depth": 0}
    violations = []
    real_lock = mqtt_common.registry_lock

    import contextlib

    @contextlib.contextmanager
    def watched_lock(registry_dir):
        held["depth"] += 1
        try:
            with real_lock(registry_dir):
                yield
        finally:
            held["depth"] -= 1

    def guard(name):
        def spy(*args, **kwargs):
            if held["depth"] > 0:
                violations.append(name)
            raise AssertionError("{} must not run under the registry lock".format(name))
        return spy

    monkeypatch.setattr(mqtt_common, "registry_lock", watched_lock)
    monkeypatch.setattr(engine.mqtt_common, "registry_lock", watched_lock, raising=False)
    monkeypatch.setattr(subprocess, "run", guard("subprocess.run"))
    monkeypatch.setattr(subprocess, "Popen", guard("subprocess.Popen"))
    monkeypatch.setattr(socket, "socket", guard("socket.socket"))
    real_sleep = time.sleep

    def sleep_spy(seconds):
        if held["depth"] > 0:
            violations.append("time.sleep")
        return real_sleep(0)

    monkeypatch.setattr(time, "sleep", sleep_spy)

    for req in ("claim", "progress", "completed", "completed", "error"):
        engine.transition(job_id, req, registry_dir=reg)

    assert not violations, "blocking calls under the lock: {}".format(violations)


def test_g6_every_record_writer_is_the_engine_or_a_declared_exception():
    """G6: no new bypass of the transition boundary can be added unnoticed."""
    import ast

    allowed = {
        # creation, and the sequence counter, which never touches status
        ("registry.py", "register_job"),
        ("mqtt_common.py", "next_seq"),
        ("mqtt_common.py", "_update_job_status_legacy"),
        ("registry.py", "_pick_pending_legacy"),
        ("engine.py", "transition"),
        # Issue #12 migration: drops broker.password only, never status
        ("registry.py", "_cmd_scrub_secrets"),
    }
    found = set()
    for path in list((SKILLS / "lib_py" / "delegation").glob("*.py")) + \
                list((SKILLS / "lib_py" / "loop").glob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.FunctionDef):
                continue
            for inner in ast.walk(node):
                if isinstance(inner, ast.Call):
                    fn = inner.func
                    name = getattr(fn, "attr", getattr(fn, "id", None))
                    if name == "_atomic_write_record":
                        found.add((path.name, node.name))
    assert found <= allowed, (
        "job records are written outside the transition boundary: {}".format(found - allowed)
    )


# ==============================================================================
# 3. watcher — herdr agent wait wrapping. No test ever runs a real herdr.
# ==============================================================================

class _Proc(object):
    def __init__(self, returncode, stdout="", stderr=""):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


def _runner(proc, seen=None):
    def run(argv, **_kw):
        if seen is not None:
            seen.append(argv)
        if isinstance(proc, Exception):
            raise proc
        return proc
    return run


AGENT_SNAPSHOT = {
    "id": "cli:agent:get",
    "result": {"agent": {"agent_status": "idle", "interactive_ready": True,
                         "pane_id": "w2X:p1", "state_change_seq": 18},
               "type": "agent_info"},
}


def test_watcher_passes_timeout_in_milliseconds_and_always_names_until():
    """The unit is the contract: --timeout is MS, and omitting --until drops `working`."""
    from lib_py.loop import watcher
    seen = []
    watcher.wait_once("sess", until=("idle", "working"), timeout_ms=15000,
                      run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT)), seen))
    argv = seen[0]
    assert argv[:4] == ["herdr", "agent", "wait", "sess"]
    assert argv[argv.index("--timeout") + 1] == "15000", argv
    assert argv.count("--until") == 2 and "working" in argv, argv


def test_watcher_rejects_a_nonpositive_timeout():
    from lib_py.loop import watcher
    with pytest.raises(ValueError):
        watcher.build_argv("sess", timeout_ms=0)


def test_watcher_matched_snapshot_comes_from_stdout():
    from lib_py.loop import watcher
    out = watcher.wait_once("sess", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
    assert out.kind == watcher.MATCHED
    assert out.agent["agent_status"] == "idle"
    assert out.healthy


@pytest.mark.parametrize("code,expected", [
    ("timeout", "timeout"),
    ("agent_not_found", "agent_not_found"),
])
def test_watcher_reads_errors_from_stderr_not_stdout(code, expected):
    """exit 1 puts the payload on stderr and leaves stdout empty."""
    from lib_py.loop import watcher
    payload = json.dumps({"error": {"code": code, "message": "x"}, "id": "cli:agent:wait"})
    out = watcher.wait_once("sess", run=_runner(_Proc(1, "", payload)))
    assert out.kind == expected
    assert out.error_code == code
    assert out.healthy, "a socket that answered is healthy even when it says no"


def test_watcher_unparseable_or_unexpected_output_is_a_transport_error():
    from lib_py.loop import watcher
    assert watcher.wait_once("s", run=_runner(_Proc(1, "", "not json"))).kind == watcher.TRANSPORT_ERROR
    assert watcher.wait_once("s", run=_runner(_Proc(0, "{}"))).kind == watcher.TRANSPORT_ERROR
    assert watcher.wait_once("s", run=_runner(_Proc(7, "", "{}"))).kind == watcher.TRANSPORT_ERROR


def test_watcher_reports_a_failed_launch_rather_than_raising():
    from lib_py.loop import watcher
    out = watcher.wait_once("s", run=_runner(OSError("herdr missing")))
    assert out.kind == watcher.TRANSPORT_ERROR and not out.healthy


def test_watcher_only_matched_outcomes_become_evidence():
    """A timeout proves nothing about agent state, so it must not become an observation."""
    from lib_py.loop import watcher
    matched = watcher.wait_once("s", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
    assert watcher.observation_from(matched).payload["agent_status"] == "idle"
    timed_out = watcher.wait_once(
        "s", run=_runner(_Proc(1, "", json.dumps({"error": {"code": "timeout"}}))))
    assert watcher.observation_from(timed_out) is None


def test_watcher_backoff_doubles_to_a_ceiling_and_resets():
    from lib_py.loop import watcher
    b = watcher.ReconnectBackoff()
    assert b.sequence(7) == [1.0, 2.0, 4.0, 8.0, 16.0, 16.0, 16.0]
    b.record_failure(); b.record_failure()
    assert b.consecutive_failures == 2
    b.record_success({"agent_status": "idle"})
    assert b.consecutive_failures == 0 and b.current == 0.0


def test_watcher_drops_its_cached_snapshot_when_the_socket_fails():
    """After a gap the watcher must re-observe; stale state is what makes a
    disconnect look like a completion."""
    from lib_py.loop import watcher
    b = watcher.ReconnectBackoff()
    b.record_success({"agent_status": "idle"})
    assert b.cached_snapshot is not None
    b.record_failure()
    assert b.cached_snapshot is None


# ==============================================================================
# 4. store — attribution lives outside the registry
# ==============================================================================

def test_store_keeps_attribution_out_of_the_job_record(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.loop import store
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    before = _dir_fingerprint(Path(reg))

    store.record_dispatch(job_id, 1, "uuid-1", "w2X:p1:1234:2026-09-06T00:00:00Z",
                          workspace=str(mam_sandbox))
    ctx = store.load_dispatch(job_id, workspace=str(mam_sandbox))
    assert ctx["attempt_id"] == 1
    assert ctx["execution_anchor_observed"] is False, "dispatch ack is not execution"
    assert _dir_fingerprint(Path(reg)) == before, "the registry must be untouched"


def test_store_execution_anchor_is_recorded_separately(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.loop import store
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    store.record_dispatch(job_id, 1, "u", "w2X:p1:1:t", workspace=str(mam_sandbox))
    ctx = store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))
    assert ctx["execution_anchor_observed"] is True


# ==============================================================================
# 5. attribution — the four-fold predicate
# ==============================================================================

def _ctx(**over):
    base = {"job_id": "j", "attempt_id": 1, "agent_session_uuid": "u",
            "pane_process_generation": "w2X:p1:1234:t",
            "dispatch_ack": True, "execution_anchor_observed": True}
    base.update(over)
    return base


def test_attribution_requires_all_four_plus_the_anchor():
    from lib_py.loop import attribution
    assert attribution.check(_ctx()).ok
    assert not attribution.check(_ctx(execution_anchor_observed=False)).ok
    assert not attribution.check(_ctx(dispatch_ack=False)).ok
    for field in attribution.REQUIRED_FIELDS:
        broken = _ctx()
        del broken[field]
        assert not attribution.check(broken).ok, field
        empty = _ctx(**{field: None})
        assert not attribution.check(empty).ok, field


def test_attribution_rejects_a_mismatch_on_any_identifier():
    from lib_py.loop import attribution
    assert not attribution.check(_ctx(), {"attempt_id": 2}).ok
    assert not attribution.check(_ctx(), {"agent_session_uuid": "other"}).ok
    assert attribution.check(_ctx(), {"attempt_id": 1, "job_id": "j"}).ok


def test_attribution_pane_generation_prefix_is_not_a_substring_match():
    from lib_py.loop import attribution
    assert attribution.pane_matches_generation("w2X:p1", "w2X:p1:1234:t")
    assert not attribution.pane_matches_generation("w2X:p1", "w2X:p10:1234:t")
    assert not attribution.pane_matches_generation("", "w2X:p1:1:t")


# ==============================================================================
# 6. policy — checked against the phase-0 scenario contracts, not fresh rules
# ==============================================================================

def _scenarios():
    out = {}
    for path in sorted(WATCHER_FIXTURES.glob("*.json")):
        if path.name == "manifest.json":
            continue
        data = json.loads(path.read_text(encoding="utf-8"))
        out[data["scenario_id"]] = data
    return out


@pytest.mark.parametrize("scenario_id", [
    # the phase-0 seven ...
    "initial_idle", "tool_running", "report_pause_15s", "finalized_candidate",
    "blocked_permission", "transport_disconnected", "restarted_generation",
    # ... and the five P2A-04 requires, held to the same contract
    "race_double_dispatch", "attempt_reuse", "partial_report",
    "unknown_status", "restart_midflight",
    # ... and the three phase 2b adds, held to the same contract
    "side_effect_unproven", "no_verdict_fail_closed", "notification_crash_window",
])
def test_g7_policy_matches_the_phase0_expected_decision(scenario_id):
    """G7: the seven scenarios that survived three-reviewer scrutiny are the
    contract. The policy must reproduce their decisions exactly."""
    from lib_py.loop import policy
    scenario = _scenarios()[scenario_id]
    expected = scenario["expected_policy"]["decision"]
    got = policy.decide(scenario)
    assert got.decision == expected, (
        "{}: expected {!r}, got {!r} ({})".format(scenario_id, expected, got.decision, got.reason)
    )


def test_policy_is_pure_no_io_no_clock(monkeypatch):
    """Purity is what makes the fixtures usable as vectors; guard it."""
    import socket
    from lib_py.loop import policy

    scenarios = list(_scenarios().values())   # load before the guards go up

    def boom(*_a, **_k):
        raise AssertionError("policy.decide must not perform I/O")

    monkeypatch.setattr(subprocess, "run", boom)
    monkeypatch.setattr(socket, "socket", boom)
    monkeypatch.setattr(Path, "read_text", boom)
    monkeypatch.setattr(time, "time", boom)
    for scenario in scenarios:
        policy.decide(scenario)


def test_policy_debounce_alone_never_authorises_completion():
    """A stable mtime is a candidate filter, not an authority to complete."""
    from lib_py.loop import policy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    scenario["filesystem"]["manifest_committed"] = False
    assert policy.decide(scenario).decision == "hold"
    scenario["filesystem"]["manifest_committed"] = True
    scenario["filesystem"]["hash_verified"] = False
    assert policy.decide(scenario).decision == "hold"


def test_policy_holds_when_the_execution_anchor_is_missing():
    """Idle is not evidence of finishing; a reused session looks identical."""
    from lib_py.loop import policy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    scenario["dispatch_context"]["execution_anchor_observed"] = False
    decision = policy.decide(scenario)
    assert decision.decision == "hold" and "anchor" in decision.reason


# ==============================================================================
# 7. shadow — cannot write, proven structurally and by hashing the registry
# ==============================================================================

def test_g4_shadow_run_leaves_the_registry_byte_identical(mam_sandbox, tmp_path, monkeypatch):
    """G4: stronger than 'status was not changed' -- nothing under the registry
    directory changes at all."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.loop import shadow

    reg = Path(_reg(tmp_path))
    job_ids = [_make_job(str(reg)) for _ in range(3)]
    from lib_py.delegation import engine
    engine.transition(job_ids[0], "claim", registry_dir=str(reg))
    before = _dir_fingerprint(reg)
    assert before, "fixture precondition: the registry has content to protect"

    runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
    results = runner.evaluate_all(list(_scenarios().values()))

    assert _dir_fingerprint(reg) == before, "shadow mode wrote to the registry"
    assert any(r["would_complete"] for r in results), "the approved scenario must be reported"
    assert all(r["shadow"] is True for r in results)


def test_g4_shadow_runner_holds_no_writer_at_all():
    """Structural, not promised: the runner has no route to a write."""
    import ast

    src = (SKILLS / "lib_py" / "loop" / "shadow.py").read_text(encoding="utf-8")
    tree = ast.parse(src)
    imported = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and node.module:
            for alias in node.names:
                imported.add("{}.{}".format(node.module, alias.name))
        elif isinstance(node, ast.Import):
            for alias in node.names:
                imported.add(alias.name)
    assert not any("engine" in name for name in imported), imported
    for forbidden in ("update_job_status", "_atomic_write_record", "transition("):
        assert forbidden not in src, "shadow must not reference {}".format(forbidden)


def test_shadow_records_its_would_complete_decisions(mam_sandbox, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.loop import shadow, store

    scenario = _scenarios()["finalized_candidate"]
    runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
    entry = runner.evaluate(scenario)
    assert entry["would_complete"] is True
    logged = store.read_decisions(entry["job_id"], workspace=str(mam_sandbox))
    assert logged and logged[-1]["decision"] == "candidate_approved"


def test_g9_loop_package_imports_only_delegation_among_siblings():
    """G9: the rule that caught the engine's original placement."""
    import ast

    offenders = []
    for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            names = []
            if isinstance(node, ast.Import):
                names = [a.name for a in node.names]
            elif isinstance(node, ast.ImportFrom) and node.module:
                names = [node.module]
            for name in names:
                if name.startswith("lib_py.") and \
                        not name.startswith("lib_py.loop") and \
                        not name.startswith("lib_py.delegation"):
                    offenders.append("{}:{} {}".format(path.name, node.lineno, name))
    assert not offenders, offenders


# The G9 guard reads CODE, not prose. Its claim is that no frozen lib.sh
# helper is *invoked* from lib_py/loop; a comment or docstring that names
# lib.sh (say, to point at the shell twin of a function so the two stay in
# sync) invokes nothing. Every way Python could actually reach a lib.sh helper
# - a path or `source` inside a subprocess argv, an identifier - lives in code
# text, which is still scanned in full, so the guard loses none of its reach.
# History: it first fired on `hashlib.sha256` (a word boundary fixed that),
# then on a docstring cross-reference in watcher.py; both were mentions.
_G9_PATTERNS = {
    "ensure_subscriber_daemon": r"ensure_subscriber_daemon",
    "preexec_fn": r"preexec_fn",
    "lib.sh": r"\blib\.sh\b",     # word boundary: `hashlib.sha256` is not lib.sh
}


def _code_only(src):
    """`src` with comments and docstrings blanked, line structure kept."""
    import ast
    import io
    import tokenize

    lines = src.splitlines(True)
    blank = set()
    tree = ast.parse(src)
    for node in ast.walk(tree):
        if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
            body = getattr(node, "body", None) or []
            first = body[0] if body else None
            if (isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant)
                    and isinstance(first.value.value, str)):
                blank.update(range(first.lineno, first.end_lineno + 1))
    out = []
    for tok in tokenize.generate_tokens(io.StringIO(src).readline):
        if tok.type == tokenize.COMMENT:
            (r, c), (_, c2) = tok.start, tok.end
            lines[r - 1] = lines[r - 1][:c] + " " * (c2 - c) + lines[r - 1][c2:]
    for i, line in enumerate(lines, 1):
        out.append("\n" if i in blank else line)
    return "".join(out)


def _g9_offenders(src):
    import re
    code = _code_only(src)
    return [name for name, pat in _G9_PATTERNS.items() if re.search(pat, code)]


def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper():
    """The two phase 1 handoffs stay dormant: preexec_fn is unreachable because
    nothing here spawns, and no frozen lib.sh helper is invoked."""
    for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
        src = path.read_text(encoding="utf-8")
        hits = _g9_offenders(src)
        assert not hits, "{} references {}".format(path.name, ", ".join(hits))


def test_g9b_the_guard_still_catches_an_invocation_and_ignores_prose():
    """The narrowing must not cost the guard its purpose: an invocation-shaped
    reference in code text still trips it; the same words in a docstring or a
    comment do not."""
    invoked = (
        'import subprocess\n'
        'def go():\n'
        '    return subprocess.run(["bash", "-c", "source .agents/skills/lib.sh; ensure_subscriber_daemon x"])\n'
    )
    assert set(_g9_offenders(invoked)) == {"lib.sh", "ensure_subscriber_daemon"}
    path_only = 'CMD = ["bash", "/x/.agents/skills/lib.sh"]\n'
    assert _g9_offenders(path_only) == ["lib.sh"]
    prose = (
        '"""Module doc: same convention as lib.sh\'s _herdr_caller_server."""\n'
        'import hashlib\n'
        'def f():\n'
        '    """See lib.sh; never call ensure_subscriber_daemon from here."""\n'
        '    # preexec_fn is documented in lib.sh\n'
        '    return hashlib.sha256(b"x").hexdigest()\n'
    )
    assert _g9_offenders(prose) == []


# ==============================================================================
# 8. Cross-review regressions (job 9d1e0077)
#
# F-A (claude): the CAS table omitted two legitimate wrapper transitions, so
#               `submit --type loop|discuss` silently stopped working by default
#               and the CLI still exited 0.
# F-B (claude): the audit status mirror moved outside the lock and became
#               last-writer-wins, so a slow flush could leave it behind.
# R1  (codex):  the policy trusted the dispatch record's own flags and never
#               confronted the identifiers the evidence carried.
# R2  (codex):  a supplied expected_attempt was ignored when the record had none.
# R3  (codex):  a raw status field rode in on a metadata merge.
# ==============================================================================

@pytest.mark.parametrize("mode", CAS_MODES)
def test_fa_wrapper_loop_redispatch_and_finalize_work(mam_sandbox, tmp_path, monkeypatch, mode):
    """F-A: the exact CLI calls the wrapper makes for --type loop|discuss.

    Iteration reset hands a finished job to the next session; exhausting the
    budget finalises it as an error. Both must work in either mode."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common, registry

    reg = _reg(tmp_path)
    job_id = _make_job(reg, agent_session="herdr:claude", job_type="loop")
    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")

    rc = registry.main(["--registry-dir", reg, "update", "--job", job_id,
                        "--agent-session", "herdr:rev", "--prompt", "review",
                        "--iteration", "2", "--role", "Reviewer", "--status", "pending"])
    record = _load(reg, job_id)
    assert rc == 0, "the wrapper's iteration reset must succeed"
    assert record["status"] == "pending"
    assert record["iteration"] == 2, "metadata must not be discarded"
    assert record["agent_session"] == "herdr:rev"

    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
    rc2 = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "error"])
    assert rc2 == 0
    assert _load(reg, job_id)["status"] == "error"


def test_fa_redispatch_requires_operator_authority_and_a_forward_attempt(
    mam_sandbox, tmp_path, monkeypatch
):
    """The new authority must not become a way for an agent to undo a terminal."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    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)

    agent = engine.transition(job_id, "redispatch", registry_dir=reg,
                              fields={"iteration": 2}, source="publish_event")
    assert agent.committed is False and "operator authority" in agent.reason

    stale = engine.transition(job_id, "redispatch", registry_dir=reg,
                              fields={"iteration": 1}, source="registry_cli")
    assert stale.committed is False and "does not advance" in stale.reason

    ok = engine.transition(job_id, "redispatch", registry_dir=reg,
                           fields={"iteration": 2}, source="registry_cli")
    assert ok.committed and ok.to_status == "pending"


def test_fa_agent_still_cannot_move_a_terminal(mam_sandbox, tmp_path, monkeypatch):
    """The protection the CAS table exists for must survive the new authority."""
    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="completed")
    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error")
    assert _load(reg, job_id)["status"] == "completed"


def test_fa_a_refused_operator_write_exits_non_zero(mam_sandbox, tmp_path, monkeypatch):
    """The silence is what hid F-A: a refused write used to exit 0."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, registry
    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)
    # cancelled after completed is a genuine conflict, and must be reported
    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"])
    assert rc != 0, "a refused operator write must not look like success"
    assert _load(reg, job_id)["status"] == "completed"


def test_fb_a_late_status_mirror_cannot_overwrite_a_newer_one(mam_sandbox, tmp_path, monkeypatch):
    """F-B: the mirror runs outside the lock, so ordering rides on transition_seq."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    logs = mqtt_common.get_logs_dir()

    mqtt_common.update_logged_status(job_id, "completed", logs, transition_seq=2)
    mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1)  # late flush
    assert mqtt_common.read_logged_status(job_id, logs)["status"] == "completed"
    # a genuinely newer transition still lands
    mqtt_common.update_logged_status(job_id, "error", logs, transition_seq=3)
    assert mqtt_common.read_logged_status(job_id, logs)["status"] == "error"


@pytest.mark.parametrize("field,value", [
    ("agent_session_uuid", "someone-else"),
    ("attempt_id", 999),
    ("job_id", "another-job"),
])
def test_r1_policy_rejects_evidence_that_contradicts_the_dispatch(field, value):
    """R1: attribution must confront what was observed, not just the record."""
    from lib_py.loop import policy
    import copy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    assert policy.decide(scenario).decision == "candidate_approved"
    mutated = copy.deepcopy(scenario)
    # The collector envelope is where an identifier legitimately lives; a native
    # payload is not an identifier source at all.
    mutated["timeline"][2]["collector_attribution"][field] = value
    decision = policy.decide(mutated)
    assert decision.decision == "hold", "{}={!r} must not be approved".format(field, value)
    assert field in decision.reason


def test_r1_policy_rejects_a_manifest_from_another_attempt():
    """A finalized report proves *an* attempt finished, not necessarily this one."""
    from lib_py.loop import policy
    import copy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    mutated = copy.deepcopy(scenario)
    mutated["timeline"][1]["payload"]["report_manifest"]["attempt_id"] = 999
    decision = policy.decide(mutated)
    assert decision.decision == "hold" and "manifest" in decision.reason


def test_r2_expected_attempt_must_be_proved_not_merely_uncontradicted(
    mam_sandbox, tmp_path, monkeypatch
):
    """R2: a precondition the record cannot answer must fail closed."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine, mqtt_common
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)

    record = _load(reg, job_id)
    record.pop("iteration", None)
    record.pop("attempt_id", None)
    mqtt_common._atomic_write_record(job_id, reg, record)
    before = _dir_fingerprint(Path(reg))

    res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=999)
    assert res.committed is False and "carries none" in res.reason
    assert _dir_fingerprint(Path(reg)) == before, "a rejection must not write"

    # with an attempt present the precondition behaves normally
    engine.transition(job_id, "progress", registry_dir=reg, fields={"attempt_id": 7})
    assert engine.transition(job_id, "completed", registry_dir=reg,
                             expected_attempt=8).committed is False
    assert engine.transition(job_id, "completed", registry_dir=reg,
                             expected_attempt=7).committed is True


@pytest.mark.parametrize("request_", ["progress", "permission_required", "merge"])
def test_r3_a_status_cannot_ride_in_as_a_metadata_field(
    mam_sandbox, tmp_path, monkeypatch, request_
):
    """R3: the boundary itself must refuse it, not just one wrapper."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    res = engine.transition(job_id, request_, registry_dir=reg,
                            fields={"status": "completed"})
    assert res.committed is False and "must be requested" in res.reason
    assert _load(reg, job_id)["status"] == "running"


def test_r3_an_unknown_request_is_refused(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    engine.transition(job_id, "claim", registry_dir=reg)
    res = engine.transition(job_id, "not_a_request", registry_dir=reg, fields={"x": 1})
    assert res.committed is False and "unknown request" in res.reason
    assert "x" not in _load(reg, job_id)


# ==============================================================================
# 9. Cross-review round 2 regressions (job 57166ae2)
#
# F-C (claude): an idempotent ACK was reported as a refusal, so the wrapper's
#               approved path aborted under set -e before its validation hook.
# F-D (claude): routing around update_status lost the status vocabulary check.
# R1  (codex):  absence of evidence still approved; only present identifiers
#               were compared.
# R4  (codex):  the mirror's ordering guard was a check followed by a write with
#               no lock between them.
# ==============================================================================

@pytest.mark.parametrize("mode", CAS_MODES)
def test_fc_idempotent_finalize_is_success_not_refusal(mam_sandbox, tmp_path, monkeypatch, mode):
    """F-C: the wrapper finalises an approved job that is already completed.

    It runs under `set -e`, so reporting failure there aborts the success path
    before the validation hook and the final stdout line."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common, registry

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")

    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "completed"])
    assert rc == 0, "re-affirming a state that already holds is an ack, not a failure"
    assert _load(reg, job_id)["status"] == "completed"


def test_fc_a_genuine_conflict_is_still_reported(mam_sandbox, tmp_path, monkeypatch):
    """The F-C fix must not soften a real refusal back into silence."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common, registry
    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"])
    assert rc == 4
    assert _load(reg, job_id)["status"] == "completed"


def test_fc_engine_marks_an_idempotent_result(mam_sandbox, tmp_path, monkeypatch):
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import engine
    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)
    again = engine.transition(job_id, "completed", registry_dir=reg)
    assert again.committed is False and again.idempotent is True
    conflict = engine.transition(job_id, "cancelled", registry_dir=reg)
    assert conflict.committed is False and conflict.idempotent is False


@pytest.mark.parametrize("mode", CAS_MODES)
def test_fd_invalid_status_is_rejected_in_both_modes(mam_sandbox, tmp_path, monkeypatch, mode):
    """F-D: the vocabulary check must survive the routing, especially on the
    rollback path where the engine is not there to catch it."""
    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import registry

    reg = _reg(tmp_path)
    job_id = _make_job(reg)
    before = _dir_fingerprint(Path(reg))
    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "bogus"])
    assert rc == 1, "an invalid status is a usage error, not a CAS refusal"
    assert _dir_fingerprint(Path(reg)) == before, "an invalid status must not be written"


@pytest.mark.parametrize("field", [
    "job_id", "attempt_id", "agent_session_uuid", "pane_process_generation",
])
def test_r1_missing_attribution_evidence_holds(field):
    """R1: absence of proof is not proof. Every identifier must be supplied."""
    import copy
    from lib_py.loop import policy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    assert policy.decide(scenario).decision == "candidate_approved"

    mutated = copy.deepcopy(scenario)
    for item in mutated["timeline"]:
        (item.get("collector_attribution") or {}).pop(field, None)
        (item.get("payload") or {}).pop(field, None)
    decision = policy.decide(mutated)
    assert decision.decision == "hold", "missing {} must not be approved".format(field)
    assert field in decision.reason


def test_r1_missing_report_manifest_holds():
    import copy
    from lib_py.loop import policy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    mutated = copy.deepcopy(scenario)
    for item in mutated["timeline"]:
        (item.get("payload") or {}).pop("report_manifest", None)
    decision = policy.decide(mutated)
    assert decision.decision == "hold" and "manifest" in decision.reason


def test_r1_identifiers_inside_a_native_payload_are_not_evidence():
    """A MAM field cannot appear in a native Herdr payload, so one found there is
    either fabricated or a contract violation. Either way it proves nothing."""
    import copy
    from lib_py.loop import policy
    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
    mutated = copy.deepcopy(scenario)
    dispatch = mutated["dispatch_context"]
    herdr_item = [i for i in mutated["timeline"] if i["source_kind"] == "herdr"][-1]
    herdr_item.pop("collector_attribution", None)
    for field in ("job_id", "attempt_id", "agent_session_uuid"):
        herdr_item["payload"][field] = dispatch[field]
    assert policy.decide(mutated).decision != "candidate_approved"


def test_r4_overlapping_mirror_flushes_cannot_lose_the_newer_state(mam_sandbox, monkeypatch):
    """R4: a sequence number orders nothing unless compare and write are atomic.

    Interrupts a flush after it has read, lets a newer one complete, and requires
    the newer state to survive."""
    import threading
    from lib_py.delegation import mqtt_common

    monkeypatch.chdir(mam_sandbox)
    logs = mqtt_common.get_logs_dir()
    job_id = "r4probe"
    mqtt_common.update_logged_status(job_id, "pending", logs, transition_seq=0)
    path = mqtt_common.job_log_path(job_id, mqtt_common.STATUS_FILENAME, logs)

    real_read = Path.read_text
    state = {}

    def intercept(self, *a, **k):
        text = real_read(self, *a, **k)
        if str(self) == str(path) and "fired" not in state:
            state["fired"] = True
            thread = threading.Thread(
                target=mqtt_common.update_logged_status,
                args=(job_id, "completed", logs), kwargs={"transition_seq": 2})
            thread.start()
            state["thread"] = thread
            thread.join(timeout=0.5)
            state["blocked"] = thread.is_alive()
        return text

    monkeypatch.setattr(Path, "read_text", intercept)
    mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1)
    monkeypatch.undo()
    state["thread"].join(timeout=5)

    assert state.get("blocked") is True, "the competing flush must serialise, not race"
    final = json.loads(path.read_text(encoding="utf-8"))
    assert final["status"] == "completed" and final["transition_seq"] == 2


def test_r4_each_mirror_write_uses_its_own_temp_file(mam_sandbox, monkeypatch):
    """A shared temp name lets concurrent writers clobber each other pre-rename."""
    monkeypatch.chdir(mam_sandbox)
    from lib_py.delegation import mqtt_common
    source = Path(SKILLS / "lib_py" / "delegation" / "mqtt_common.py").read_text(encoding="utf-8")
    body = source[source.index("def update_logged_status"):source.index("def init_job_log")]
    assert "mkstemp" in body, "the mirror must not reuse one fixed .tmp path"
    logs = mqtt_common.get_logs_dir()
    mqtt_common.update_logged_status("tmpprobe", "running", logs, transition_seq=1)
    leftovers = list((Path(logs) / "tmpprobe").glob("*.tmp"))
    assert not leftovers, "temp files must be renamed away, not left behind: {}".format(leftovers)