#!/usr/bin/env python3
"""Issue #5 Phase 1 — delegation extraction characterization & contract tests.

Written BEFORE the `lib_py/delegation/` extraction (P1-00) so the refactor has a
real regression oracle: the three shell helpers in `lib.sh` had zero direct test
coverage. Every test here must pass identically against the legacy inline bodies
and against the Python-backed ones, which is why the shell tests are
parametrized over `MAM_DELEGATION_PY` in {"0", "1"}.

Sections
  1. C-01..C-09  shell helper golden behaviour (non-fatal contract, stdout shape,
                 daemon PID/dedup/logging).
  2. CLI contract snapshots — argparse option names/choices/defaults, asserted
     against the parser rather than --help text so cosmetic wording is free.
  3. Auto-completion invariant guards — the three surfaces that decide job status
     must not shift by a single line during Phase 1.
"""
from __future__ import annotations

import json
import os
import re
import signal
import subprocess
import sys
import time
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "phase1"

DELEGATION_MODES = ("0", "1")  # MAM_DELEGATION_PY: legacy inline body / python body


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

def _sandbox_skills(mam_sandbox) -> Path:
    return Path(mam_sandbox) / "skills"


def _run_helper(mam_sandbox, snippet: str, env=None, timeout=30):
    """Source the sandbox lib.sh and run a shell snippet against it."""
    lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
    run_env = dict(os.environ)
    run_env.pop("HERDR_SESSION_NAME", None)
    run_env.pop("HERDR_SERVER_NAME", None)
    if env:
        run_env.update({k: str(v) for k, v in env.items()})
    return subprocess.run(
        ["bash", "-c", 'source "{}" >/dev/null 2>&1; {}'.format(lib_sh, snippet)],
        capture_output=True, text=True, env=run_env, timeout=timeout,
    )


def _stub_env(tmp_path, *, exit_code=0, stdout=""):
    """Point _delegate_py_bin at the recording stub via AGENT_PYTHON_BIN."""
    log = tmp_path / "stub_argv.log"
    return {
        "AGENT_PYTHON_BIN": str(FIXTURES / "fake_python.sh"),
        "STUB_ARGV_LOG": str(log),
        "STUB_EXIT": str(exit_code),
        "STUB_STDOUT": stdout,
    }, log


def _install_fake_monitor(mam_sandbox) -> Path:
    """Replace the sandbox reconcile.sh with a sleeping stub and return its path."""
    target = (Path(mam_sandbox) / ".agents" / "skills"
              / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text((FIXTURES / "fake_reconcile.sh").read_text(encoding="utf-8"),
                      encoding="utf-8")
    target.chmod(0o755)
    return target


def _kill(pids):
    for pid in pids:
        try:
            os.kill(int(pid), signal.SIGKILL)
        except (OSError, ValueError):
            pass


def _alive(pid) -> bool:
    try:
        os.kill(int(pid), 0)
        return True
    except (OSError, ValueError):
        return False


# ==============================================================================
# 1. Shell helper golden behaviour (C-01 .. C-09)
# ==============================================================================

@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c01_publish_event_empty_job_is_noop(mam_sandbox, mode):
    """C-01: an empty job id returns 0 without ever starting an interpreter."""
    env, log = _stub_env(mam_sandbox, exit_code=7)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(mam_sandbox, 'delegate_publish_event "" started', env=env)
    assert res.returncode == 0, res.stderr
    assert not log.exists(), "python must not be invoked for an empty job id"


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c02_publish_event_missing_script_is_non_fatal(mam_sandbox, mode):
    """C-02: a missing publish_event.py must not abort the caller."""
    script = (_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job"
              / "scripts" / "publish_event.py")
    script.unlink()
    env, _ = _stub_env(mam_sandbox)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(mam_sandbox, 'delegate_publish_event j1 started', env=env)
    assert res.returncode == 0, res.stderr


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c03_publish_event_broker_failure_is_non_fatal(mam_sandbox, mode):
    """C-03: a non-zero interpreter exit (broker down) still returns 0."""
    env, log = _stub_env(mam_sandbox, exit_code=2)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_publish_event j1 completed "done"; echo "rc=$?"',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    assert "rc=0" in res.stdout, res.stdout
    assert log.exists(), "the interpreter should have been invoked"


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c04_publish_event_forwards_job_event_detail(mam_sandbox, mode):
    """C-04: the job id, event and detail reach the interpreter argv intact."""
    env, log = _stub_env(mam_sandbox)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_publish_event abc123 progress "half way"',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    argv = log.read_text(encoding="utf-8")
    assert "--job abc123" in argv, argv
    assert "--event progress" in argv, argv
    assert "--detail half way" in argv, argv


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c05_submit_job_stdout_is_exactly_the_job_id(mam_sandbox, mode):
    """C-05: create_session.sh captures this with $( ) — stdout must be the id alone."""
    env, _ = _stub_env(mam_sandbox, stdout="deadbeef")
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_submit_job "a prompt" claude-code herdr:claude',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    assert res.stdout == "deadbeef\n", repr(res.stdout)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c06_watchdog_missing_monitor_script_errors(mam_sandbox, mode):
    """C-06: exact stderr wording and rc=1 when reconcile.sh is absent."""
    workdir = Path(mam_sandbox) / "empty_ws"
    workdir.mkdir()
    expected = (workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
                / "scripts" / "reconcile.sh")
    res = _run_helper(
        mam_sandbox,
        'start_watchdog job1 "{}"'.format(workdir),
        env={"MAM_DELEGATION_PY": mode},
    )
    assert res.returncode == 1, res.stdout
    assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c07_watchdog_spawns_daemon_with_workdir_and_log(mam_sandbox, mode):
    """C-07: prints a live numeric PID, anchors cwd at workdir, appends to the log."""
    _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    spawned = []
    try:
        res = _run_helper(
            mam_sandbox,
            'start_watchdog job1 "{}"'.format(workdir),
            env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"},
        )
        assert res.returncode == 0, res.stderr
        pid = res.stdout.strip()
        assert re.fullmatch(r"\d+", pid), repr(res.stdout)
        spawned.append(pid)
        deadline = time.time() + 5
        log_file = workdir / ".mam" / "multi-agent-mux-monitor.log"
        while time.time() < deadline and not log_file.exists():
            time.sleep(0.1)
        assert log_file.exists(), "monitor log was not created"
        assert _alive(pid), "spawned daemon is not running"
    finally:
        _kill(spawned)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c08_watchdog_second_call_reuses_running_daemon(mam_sandbox, mode):
    """C-08: the pgrep pre-check must return the existing PID, not spawn a second."""
    monitor = _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    spawned = []
    try:
        first = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                            env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"})
        pid1 = first.stdout.strip()
        spawned.append(pid1)
        assert _alive(pid1)
        second = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                             env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"})
        assert second.returncode == 0, second.stderr
        assert second.stdout.strip() == pid1, (first.stdout, second.stdout)
        matches = subprocess.run(
            ["pgrep", "-f", "bash {} --subscribe".format(monitor)],
            capture_output=True, text=True,
        ).stdout.split()
        spawned.extend(matches)
        assert len(matches) == 1, "expected exactly one daemon, got {}".format(matches)
    finally:
        _kill(spawned)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c09_start_watchdog_and_ensure_subscriber_daemon_agree(mam_sandbox, mode):
    """C-09: `start_watchdog` survives as an alias once the python body lands.

    Passes on the legacy tree (only start_watchdog defined) and on the refactored
    tree (both defined and behaviourally identical)."""
    _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    probe = _run_helper(
        mam_sandbox,
        'declare -F start_watchdog >/dev/null && echo HAS_SW; '
        'declare -F ensure_subscriber_daemon >/dev/null && echo HAS_ESD',
        env={"MAM_DELEGATION_PY": mode},
    )
    assert "HAS_SW" in probe.stdout, "start_watchdog must always remain defined"
    if "HAS_ESD" not in probe.stdout:
        return  # legacy tree: alias not introduced yet (P1-02)

    spawned = []
    try:
        a = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                        env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"})
        b = _run_helper(mam_sandbox, 'ensure_subscriber_daemon job1 "{}"'.format(workdir),
                        env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"})
        spawned += [a.stdout.strip(), b.stdout.strip()]
        assert a.returncode == b.returncode == 0, (a.stderr, b.stderr)
        assert a.stdout.strip() == b.stdout.strip(), (a.stdout, b.stdout)
    finally:
        _kill(spawned)


# ==============================================================================
# 2. CLI contract snapshots
# ==============================================================================

def _parser_snapshot(parser):
    """{option string: (nargs-ish kind, choices, default)} for one parser level."""
    snap = {}
    for action in parser._actions:
        if not action.option_strings:
            snap["<positional:{}>".format(action.dest)] = (
                action.type.__name__ if action.type else None,
                tuple(action.choices) if action.choices else None,
                action.default,
            )
            continue
        key = ",".join(sorted(action.option_strings))
        snap[key] = (
            action.type.__name__ if action.type else None,
            tuple(action.choices) if action.choices else None,
            action.default,
        )
    return snap


def _import_delegation(mam_sandbox, name):
    """Import a delegation module the way production callers do (top-level name)."""
    scripts = str(_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job" / "scripts")
    if scripts not in sys.path:
        sys.path.insert(0, scripts)
    return __import__(name)


def test_publish_event_cli_contract_snapshot(mam_sandbox):
    publish_event = _import_delegation(mam_sandbox, "publish_event")
    assert publish_event.VALID_EVENTS == (
        "started", "permission_required", "progress", "completed", "error", "cancelled",
    )
    assert publish_event.TERMINAL_EVENTS == ("completed", "error", "cancelled")
    # main() builds its parser inline, so --help is the only way to exercise it.
    with pytest.raises(SystemExit) as exc:
        publish_event.main(["--help"])
    assert exc.value.code == 0
    # Defaults that shell and wrapper callers depend on.
    src = Path(publish_event.__file__).read_text(encoding="utf-8")
    assert '"--attempts", type=int, default=3' in src
    # Issue #15: resolved per process (MAM_JOBS_DIR, then WORKSPACE_ROOT, then
    # the cwd-relative DEFAULT_REGISTRY_DIR) instead of the bare constant.
    assert '"--registry-dir", default=default_registry_dir()' in src
    assert '"--event", default="progress", choices=VALID_EVENTS' in src


def test_registry_cli_contract_snapshot(mam_sandbox):
    registry = _import_delegation(mam_sandbox, "registry")
    parser = registry._build_parser()
    top = _parser_snapshot(parser)
    assert "--registry-dir" in top
    subparsers = [a for a in parser._actions if hasattr(a, "choices") and isinstance(a.choices, dict)]
    assert subparsers, "registry must expose subcommands"
    commands = set(subparsers[0].choices)
    assert commands == {
        "register", "list", "get", "status", "update", "close", "get-feedback", "pick", "logs",
        "scrub-secrets",  # Issue #12 migration
    }, commands
    reg = _parser_snapshot(subparsers[0].choices["register"])
    assert reg["--prompt"][2] is None
    assert reg["--agent"][2] == "claude-code"
    assert reg["--agent-session"][2] == "herdr:claude"
    assert reg["--role"][2] == "Worker"
    assert reg["--timeout"][2] == 3600
    # Issue #18: the idle window derives from the liveness contract (K x N =
    # 4 x 30 = 120 s, the historical default) unless given explicitly.
    assert reg["--idle-timeout"][2] is None
    assert reg["--heartbeat-sec"][2] == 30
    assert reg["--heartbeat-misses"][2] == 4
    assert reg["--ack-timeout"][2] == 180
    assert reg["--transport"][1] == ("herdr", "remote")
    assert registry.liveness_window(reg["--heartbeat-sec"][2], reg["--heartbeat-misses"][2]) == 120
    assert reg["--bits"][2] == 32
    assert reg["--job-type"][1] == ("direct", "loop", "discuss")
    assert reg["--max-iterations"][2] == 5
    assert registry.VALID_STATUSES == (
        "pending", "running", "completed", "error", "cancelled",
    )


def test_job_subscriber_cli_contract_snapshot(mam_sandbox):
    job_subscriber = _import_delegation(mam_sandbox, "job_subscriber")
    assert job_subscriber.TERMINAL_EVENTS == ("completed", "error")
    src = Path(job_subscriber.__file__).read_text(encoding="utf-8")
    for doc in ("  0  all watched jobs reached", "  1  any watched job reached",
                "  2  no event for the idle window", "  4  the wall-clock budget"):
        assert doc in src, doc
    # Issue #33: codes are named; their values are the frozen contract.
    assert (job_subscriber.RC_COMPLETED, job_subscriber.RC_ERROR, job_subscriber.RC_NO_EVENT,
            job_subscriber.RC_BROKER_UNAVAILABLE, job_subscriber.RC_WALL_CLOCK) == (0, 1, 2, 3, 4)
    assert 'return RC_BROKER_UNAVAILABLE' in src, "fatal-error exit code 3 must survive"
    assert not re.search(r"^\s+return [0-9]+\s*(#.*)?$", src, re.M), "exit codes are returned by name"


# ==============================================================================
# 3. Auto-completion invariant guards (Phase 1 must not move these)
# ==============================================================================

def test_event_to_status_mapping_is_frozen(mam_sandbox):
    publish_event = _import_delegation(mam_sandbox, "publish_event")
    assert publish_event.EVENT_TO_STATUS == {
        "started": "running",
        "completed": "completed",
        "error": "error",
        "cancelled": "cancelled",
    }


def test_wrapper_exit_code_mapping_literals_are_frozen(mam_sandbox):
    wrapper = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
    text = wrapper.read_text(encoding="utf-8")
    for literal in (
        'if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then',
        'elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then',
        'elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then',
        '[ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall"',
        'job_status="completed"',
        'job_status="error"',
    ):
        assert literal in text, literal
    # Issue #33: the wrapper's SUB_RC_* mirror job_subscriber's RC_* exactly,
    # and the wrapper compares subscriber codes only by those names.
    job_subscriber = _import_delegation(mam_sandbox, "job_subscriber")
    mirror = dict(re.findall(r"\bSUB_RC_([A-Z_]+)=([0-9]+)", text))
    assert mirror == {name[3:]: str(getattr(job_subscriber, name))
                      for name in dir(job_subscriber) if name.startswith("RC_")}, mirror
    assert not re.search(r"\$(sub_rc|sub_exit)\"? +(-eq|-ne|=) +\"?[0-9]", text), "subscriber codes compared by value"


def test_reconcile_terminal_mutation_literals_are_frozen():
    reconcile = SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
    text = reconcile.read_text(encoding="utf-8")
    for literal in (
        "s['delegate_job_id'] = None",
        "s['status'] = 'terminated'",
        "kill-session",
        "if _event in ('completed', 'cancelled')",
    ):
        assert literal in text, literal


# ==============================================================================
# 4. P1-01 — package binding, provenance and context matrix (PV-1, PV-4, PV-5)
# ==============================================================================

DELEGATION_MODULES = ("mqtt_common", "registry", "publish_event", "job_subscriber")
PKG_DIR = SKILLS / "lib_py" / "delegation"
SHIM_DIR = SKILLS / "multi-agent-mux-delegate-job" / "scripts"


def test_pv5_all_four_shims_share_one_bootstrap_block():
    """PV-5: the four shims differ only by the module name they bind."""
    normalised = {}
    for mod in DELEGATION_MODULES:
        text = (SHIM_DIR / (mod + ".py")).read_text(encoding="utf-8")
        normalised[mod] = text.replace(mod, "{mod}")
    base = normalised["mqtt_common"]
    for mod, text in normalised.items():
        assert text == base, "shim {} diverged from the shared bootstrap".format(mod)
    assert "sys.path.insert" not in base, "the shim must never take sys.path precedence"
    assert "sys.path.append(_SKILLS)" in base, "standalone fallback must remain"


def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox, tmp_path_factory):
    """PV-1: each alias IS the package module, and all four share one tree."""
    import lib_py.delegation

    paths = set()
    for mod in DELEGATION_MODULES:
        alias = _import_delegation(mam_sandbox, mod)
        pkg_mod = getattr(lib_py.delegation, mod, None)
        if pkg_mod is None:
            import importlib
            pkg_mod = importlib.import_module("lib_py.delegation." + mod)
        assert alias is pkg_mod, "{}: alias is not the package module".format(mod)
        paths.add(str(Path(alias.__file__).resolve().parent))

    assert len(paths) == 1, "delegation modules span multiple trees: {}".format(paths)
    bound = Path(paths.pop())
    real = PKG_DIR.resolve()
    # Provenance, not just identity: the bound tree is either the real package or a
    # copy under pytest's own basetemp. Checked against the actual basetemp rather
    # than sniffing for "pytest-" in the path, so an unrelated directory that merely
    # looks temporary cannot satisfy it.
    basetemp = Path(tmp_path_factory.getbasetemp()).resolve()
    try:
        bound.relative_to(basetemp)
        under_basetemp = True
    except ValueError:
        under_basetemp = False
    assert bound == real or under_basetemp, (
        "delegation bound to an unexpected tree: {} (real={}, basetemp={})".format(
            bound, real, basetemp)
    )
    # the parent package must never be dragged along with it
    assert Path(lib_py.__file__).resolve().parent == (SKILLS / "lib_py").resolve()


def test_delegation_package_imports_no_sibling_lib_py_modules():
    """D1(b): a frozen delegation package must not reach into the parent tree."""
    import ast

    offenders = []
    for path in sorted(PKG_DIR.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 in DELEGATION_MODULES:
                    offenders.append("{}:{} top-level `{}`".format(path.name, node.lineno, name))
                if name == "lib_py" or (
                    name.startswith("lib_py.") and not name.startswith("lib_py.delegation")
                ):
                    offenders.append("{}:{} sibling `{}`".format(path.name, node.lineno, name))
    assert not offenders, "forbidden imports inside lib_py.delegation:\n  " + "\n  ".join(offenders)


def test_importing_delegation_does_not_mutate_environment(mam_sandbox):
    """D6: importing the library must leave os.environ alone; env loads at entry points."""
    import importlib
    mqtt_common = _import_delegation(mam_sandbox, "mqtt_common")
    before = set(os.environ)
    importlib.reload(mqtt_common)
    assert set(os.environ) == before, "library import injected environment variables"
    # idempotent: the second call is a no-op
    mqtt_common.ensure_env_loaded()
    keys = set(os.environ)
    mqtt_common.ensure_env_loaded()
    assert set(os.environ) == keys


def _normalize_help(text: str) -> str:
    text = text.replace("optional arguments:", "options:")
    return re.sub(r"\s+", " ", text).strip()


def test_shim_help_output_matches_pre_move_capture():
    """The shim must be indistinguishable from the implementation at the CLI.

    The captures were taken before the phase 1 move to prove the move changed
    nothing. They are refreshed only when the CLI is *intentionally* extended
    (phase 2b added the finalization options to publish_event); the guard's job
    is to catch drift nobody meant, which it still does.
    """
    for mod in ("publish_event", "registry", "job_subscriber"):
        golden = (FIXTURES / "help-{}.txt".format(mod)).read_text(encoding="utf-8")
        res = subprocess.run(
            [sys.executable, str(SHIM_DIR / (mod + ".py")), "--help"],
            capture_output=True, text=True, timeout=60,
        )
        assert res.returncode == 0, res.stderr
        # argparse re-wraps usage/help text and renamed the "optional arguments"
        # group across Python versions (Issue #28: the captures were taken on
        # 3.9, the Linux host runs 3.14). Compare the content, not the layout:
        # whitespace runs collapsed, the group label unified.
        assert _normalize_help(res.stdout + res.stderr) == _normalize_help(golden), \
            "--help drifted for {}".format(mod)


# --- PV-4: the context matrix, each scenario in a fresh interpreter ------------

@pytest.fixture(scope="module")
def shim_trees(tmp_path_factory):
    """Four copies of the live skills tree: real / sandbox / sandbox2 / freeze."""
    import shutil

    root = tmp_path_factory.mktemp("shim_trees")
    for name in ("real", "sandbox", "sandbox2", "freeze"):
        dest = root / name
        (dest / "multi-agent-mux-delegate-job").mkdir(parents=True)
        shutil.copytree(SKILLS / "lib_py", dest / "lib_py")
        shutil.copytree(SHIM_DIR, dest / "multi-agent-mux-delegate-job" / "scripts")
    return root


def _run_scenario(shim_trees, scenario, pythonpath=None):
    env = dict(os.environ)
    env.pop("PYTHONPATH", None)
    if pythonpath:
        env["PYTHONPATH"] = str(shim_trees / pythonpath)
    res = subprocess.run(
        [sys.executable, str(FIXTURES / "shim_contexts.py"), scenario]
        + [str(shim_trees / t) for t in ("real", "sandbox", "sandbox2", "freeze")],
        capture_output=True, text=True, env=env, timeout=120,
    )
    assert res.returncode == 0, "scenario {} failed:\n{}".format(scenario, res.stderr)
    return json.loads(res.stdout)


def _delegation_trees(report):
    keys = ("alias_mqtt_common", "alias_publish_event", "pkg_delegation",
            "pkg_mqtt_common", "pkg_publish_event")
    return {report[k] for k in keys if report[k] is not None}


@pytest.mark.parametrize("scenario,expected_delegation", [
    ("A", "sandbox"),      # sandbox shim binds first
    ("A_pe", "sandbox"),   # non-leaf module alone: dependencies must follow it
    ("B", "sandbox"),      # real lib_py.agents already imported: delegation still binds to shim tree
    ("B2", "real"),        # real delegation already bound: bind-once keeps it
    ("D", "sandbox"),      # a second sandbox must not rebind
    ("E", "sandbox"),      # direct package import after a shim resolves to the bound tree
])
def test_pv4_pytest_family_binds_one_tree_and_spares_the_parent(
    shim_trees, scenario, expected_delegation
):
    rep = _run_scenario(shim_trees, scenario)
    assert _delegation_trees(rep) == {expected_delegation}, rep
    assert rep["lib_py"] == "real", "parent lib_py must stay on the real tree: {}".format(rep)
    if rep["lib_py_agents"] is not None:
        assert rep["lib_py_agents"] == "real", rep
    for name, same in rep["identity"].items():
        if same is not None:
            assert same, "identity broken for {} in {}".format(name, scenario)
    assert "sandbox" not in rep["lib_py_path_entries"], rep
    assert "sandbox2" not in rep["lib_py_path_entries"], rep


@pytest.mark.parametrize("scenario", ["C1", "C1_pe"])
def test_pv4_standalone_bootstraps_from_its_own_tree(shim_trees, scenario):
    """K4: no ambient lib_py anywhere, so the shim may append its own tree."""
    rep = _run_scenario(shim_trees, scenario)
    assert _delegation_trees(rep) == {"freeze"}, rep
    assert rep["lib_py"] == "freeze", rep


@pytest.mark.parametrize("scenario", ["C2", "C2_pe"])
def test_pv4_frozen_copy_wins_over_a_live_pythonpath(shim_trees, scenario):
    """K2: the shape every delegated job runs -- frozen script, live PYTHONPATH.

    B-13 requires the frozen code to execute; the parent lib_py may stay live
    because D1(b) forbids the delegation package from importing siblings."""
    rep = _run_scenario(shim_trees, scenario, pythonpath="real")
    assert _delegation_trees(rep) == {"freeze"}, rep
    assert rep["lib_py"] == "real", rep
    for name, same in rep["identity"].items():
        if same is not None:
            assert same, name


def test_pv4_k3_live_helpers_resolve_to_the_live_tree(shim_trees):
    rep = _run_scenario(shim_trees, "K3")
    assert _delegation_trees(rep) == {"real"}, rep
    assert rep["lib_py"] == "real", rep


def test_k1_module_execution_follows_pythonpath(shim_trees):
    """K1: `python -m lib_py.delegation.<mod>` is resolved by PYTHONPATH, not by a shim."""
    for tree in ("real", "freeze"):
        env = dict(os.environ)
        env["PYTHONPATH"] = str(shim_trees / tree)
        res = subprocess.run(
            [sys.executable, "-c",
             "import lib_py.delegation as d; print(list(d.__path__)[0])"],
            capture_output=True, text=True, env=env, timeout=60,
        )
        assert res.returncode == 0, res.stderr
        assert res.stdout.strip() == str(shim_trees / tree / "lib_py" / "delegation"), res.stdout


def test_k6_subprocess_executes_the_copy_it_was_given(shim_trees):
    """K6: e2e-style subprocess execution runs the tree the path points at."""
    env = dict(os.environ)
    env.pop("PYTHONPATH", None)
    shim = shim_trees / "sandbox" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
    res = subprocess.run([sys.executable, str(shim), "--help"],
                         capture_output=True, text=True, env=env, timeout=60)
    assert res.returncode == 0, res.stderr
    probe = subprocess.run(
        [sys.executable, "-c",
         "import sys; sys.path.insert(0, {!r}); import publish_event;"
         " print(publish_event.__file__)".format(str(shim.parent))],
        capture_output=True, text=True, env=env, timeout=60,
    )
    assert probe.returncode == 0, probe.stderr
    assert str(shim_trees / "sandbox") in probe.stdout, probe.stdout


# ==============================================================================
# 5. P1-02 — daemon module, cli router, shell-to-python equivalence
# ==============================================================================

def _delegation_env(mam_sandbox):
    """Env for running the package out of the sandbox copy."""
    env = dict(os.environ)
    env["PYTHONPATH"] = str(_sandbox_skills(mam_sandbox))
    env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
    return env


def test_daemon_missing_monitor_raises_with_the_legacy_path(mam_sandbox):
    from lib_py.delegation import daemon

    workdir = Path(mam_sandbox) / "no_monitor"
    workdir.mkdir()
    with pytest.raises(FileNotFoundError) as exc:
        daemon.ensure_subscriber_daemon("job1", str(workdir))
    assert exc.value.args[0] == str(daemon.monitor_script_path(str(workdir)))


def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox, monkeypatch):
    """The daemon must outlive the shell that started it and never double-spawn."""
    from lib_py.delegation import daemon

    monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False)  # issue #19 sandbox default
    monitor = _install_fake_monitor(mam_sandbox)
    workdir = str(mam_sandbox)
    spawned = []
    try:
        first = daemon.ensure_subscriber_daemon("job1", workdir)
        spawned += first
        assert len(first) == 1 and _alive(first[0])
        assert (Path(workdir) / ".mam" / "multi-agent-mux-monitor.log").exists()
        second = daemon.ensure_subscriber_daemon("job1", workdir)
        assert second == first, "a running daemon must be reused, not replaced"
        # detached into its own session, so a parent's process-group signal misses it
        pgid = os.getpgid(int(first[0]))
        assert pgid != os.getpgid(os.getpid())
        assert daemon.find_running(str(monitor)) == first
    finally:
        _kill(spawned)


def test_cli_ensure_daemon_reproduces_the_legacy_failure_line(mam_sandbox):
    workdir = Path(mam_sandbox) / "no_monitor_cli"
    workdir.mkdir()
    res = subprocess.run(
        [sys.executable, "-m", "lib_py.delegation.cli",
         "ensure-daemon", "--job", "j1", "--workdir", str(workdir)],
        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
    )
    assert res.returncode == 1
    expected = str(workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
                   / "scripts" / "reconcile.sh")
    assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)
    assert res.stdout == ""


def test_cli_submit_prints_only_the_job_id(mam_sandbox):
    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)
    res = subprocess.run(
        [sys.executable, "-m", "lib_py.delegation.cli", "submit",
         "--prompt", "hello", "--agent", "claude-code",
         "--agent-session", "herdr:claude", "--registry-dir", str(reg_dir)],
        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
    )
    assert res.returncode == 0, res.stderr
    job_id = res.stdout.strip()
    assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
    assert res.stdout == job_id + "\n", "stdout must carry the id and nothing else"
    assert (reg_dir / (job_id + ".json")).exists()


@pytest.mark.parametrize("entry", ["script", "cli"])
def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry):
    """G-1/G-2/G-4 semantics must be identical through the shim and through the cli.

    Broker points at a non-routable port, so nothing touches the network."""
    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)
    env = _delegation_env(mam_sandbox)
    job_id = subprocess.run(
        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "p",
         "--registry-dir", str(reg_dir)],
        capture_output=True, text=True, env=env, timeout=60,
    ).stdout.strip()

    record = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
    record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
    (reg_dir / (job_id + ".json")).write_text(json.dumps(record), encoding="utf-8")

    if entry == "cli":
        cmd = [sys.executable, "-m", "lib_py.delegation.cli", "publish"]
    else:
        cmd = [sys.executable, str(SHIM_DIR / "publish_event.py")]
    res = subprocess.run(
        cmd + ["--job", job_id, "--event", "completed", "--detail", "done",
               "--attempts", "1", "--registry-dir", str(reg_dir)],
        capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120,
    )
    assert res.returncode == 2, (res.stdout, res.stderr)
    assert res.stdout == "", "publish must stay silent on stdout"
    after = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
    assert after["status"] == "completed", "status syncs even when the publish fails"
    assert after.get("last_seq") == 1, "seq is consumed on failure by design"
    audit = (Path(mam_sandbox) / ".mam" / "delegate_job_logs" / job_id / "events.ndjson")
    events = [json.loads(l) for l in audit.read_text(encoding="utf-8").splitlines() if l.strip()]
    published = [e for e in events if e.get("event") == "published"]
    assert published and published[-1]["published"] is False
    assert published[-1]["publish_error"]


# ==============================================================================
# 6. P1-03 — transports, parity and the benchmark harness (no network at all)
# ==============================================================================

def _prepared(topic="python/mqtt/jobs/j/events", host="broker.example", port=1883):
    from lib_py.delegation import mqtt_common, publish_event

    config = mqtt_common.broker_config_from_env({"host": host, "port": port})
    return publish_event.PreparedEvent(
        job_id="j", event="progress", detail="d", seq=1, topic=topic, retain=False,
        payload={"seq": 1}, body=b'{"seq": 1}', config=config,
    )


def _fake_factory(counters, publishes_ok=True, connect_rc=0):
    from lib_py.delegation.bench import FakeClient, FakeInfo

    def factory(_role, _config):
        client = FakeClient(connect_latency=0.0, counters=counters)
        if not publishes_ok:
            def unacked(*_a, **_k):
                counters["publish"] = counters.get("publish", 0) + 1
                return FakeInfo(published=False)
            client.publish = unacked
        if connect_rc:
            original = client.connect

            def connect(host, port, keepalive=60):
                original(host, port, keepalive)
                client.on_connect(client, None, None, connect_rc, None)
            client.connect = connect
        return client
    return factory


def test_persistent_publisher_reuses_one_connection():
    """The whole point of the prototype: N publishes, 1 connection."""
    from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher

    prepared = _prepared()
    one_counters, persist_counters = {}, {}
    one = OneShotPublisher(client_factory=_fake_factory(one_counters))
    persist = PersistentPublisher(client_factory=_fake_factory(persist_counters))
    try:
        for _ in range(5):
            assert one.publish_with_retry(prepared, attempts=1) == (True, None)
            assert persist.publish_with_retry(prepared, attempts=1) == (True, None)
    finally:
        persist.close()
    assert one_counters["connect"] == 5
    assert persist_counters["connect"] == 1
    assert one_counters["publish"] == persist_counters["publish"] == 5


def test_transport_parity_of_registry_and_audit_records(mam_sandbox, monkeypatch):
    """Same input, two transports: registry and audit records must match field for field."""
    from lib_py.delegation import mqtt_common, publish_event, registry
    from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher

    monkeypatch.chdir(mam_sandbox)
    reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
    os.makedirs(reg_dir, exist_ok=True)

    results = {}
    for label, publisher in (
        ("one_shot", OneShotPublisher(client_factory=_fake_factory({}))),
        ("persistent", PersistentPublisher(client_factory=_fake_factory({}))),
    ):
        job_id = registry.register_job("parity " + label, registry_dir=reg_dir)
        prepared = publish_event.prepare_event(
            job_id=job_id, event="completed", detail="done", registry_dir=reg_dir,
        )
        ok, err = publisher.publish_with_retry(prepared, attempts=1)
        publish_event.record_outcome(prepared, reg_dir, ok, err)
        if hasattr(publisher, "close"):
            publisher.close()
        record = mqtt_common.load_job(job_id, reg_dir)
        events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
        audit = [e for e in events if e.get("event") == "published"][-1]
        results[label] = {
            "ok": ok,
            "status": record["status"],
            "last_seq": record.get("last_seq"),
            "retain": audit["retain"],
            "published": audit["published"],
            "seq": audit["seq"],
            "has_sig": "hmac_sig" in audit["payload"].get("data", {}),
            "source_event": audit["source_event"],
        }

    assert results["one_shot"] == results["persistent"], results
    assert results["one_shot"]["retain"] is True, "terminal events must retain"
    assert results["one_shot"]["status"] == "completed"


def test_persistent_publisher_unacked_publish_retries_then_records_failure(mam_sandbox, monkeypatch):
    """No PUBACK -> exception -> with_retry -> record_outcome(published=False)."""
    from lib_py.delegation import mqtt_common, publish_event, registry
    from lib_py.delegation.publisher import PersistentPublisher

    monkeypatch.chdir(mam_sandbox)
    reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
    os.makedirs(reg_dir, exist_ok=True)
    job_id = registry.register_job("unacked", registry_dir=reg_dir)
    prepared = publish_event.prepare_event(
        job_id=job_id, event="completed", detail="d", registry_dir=reg_dir,
    )
    counters = {}
    publisher = PersistentPublisher(client_factory=_fake_factory(counters, publishes_ok=False))
    try:
        ok, err = publisher.publish_with_retry(prepared, attempts=3)
    finally:
        publisher.close()
    assert ok is False and "not acknowledged" in err
    assert counters["publish"] == 3, "every attempt must reach the transport"
    assert counters["connect"] == 1, "retries reuse the connection"

    publish_event.record_outcome(prepared, reg_dir, ok, err)
    record = mqtt_common.load_job(job_id, reg_dir)
    assert record["status"] == "completed", "status syncs even when the publish failed"
    assert record.get("last_seq") == prepared.seq
    audit = [e for e in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())
             if e.get("event") == "published"][-1]
    assert audit["published"] is False and audit["publish_error"]


def test_persistent_publisher_refuses_a_second_broker():
    from lib_py.delegation.publisher import PersistentPublisher

    publisher = PersistentPublisher(client_factory=_fake_factory({}))
    try:
        assert publisher.publish_with_retry(_prepared(host="a.example"), attempts=1)[0] is True
        ok, err = publisher.publish_with_retry(_prepared(host="b.example"), attempts=1)
        assert ok is False
        assert "bound to a.example" in err, err
    finally:
        publisher.close()


def test_bench_fake_transport_schema_and_counts():
    from lib_py.delegation import bench

    result = bench.run_benchmark(n=7, transport="fake", connect_latency_ms=1.0)
    assert result["transport"] == "fake"
    for arm in ("one_shot", "persistent"):
        assert result[arm]["publishes"] == 7
        assert result[arm]["failures"] == 0
        assert set(result[arm]["latency_ms"]) == {"p50", "p95", "max"}
    assert result["one_shot"]["client_connects"] == 7
    assert result["persistent"]["client_connects"] == 1
    assert result["persistent"]["latency_ms"]["p50"] < result["one_shot"]["latency_ms"]["p50"]


def test_bench_broker_transport_requires_a_job():
    from lib_py.delegation import bench

    with pytest.raises(ValueError):
        bench.run_benchmark(n=1, transport="broker")


# ==============================================================================
# 7. Self-review regressions — the connect handshake is per-connection
#
# Found reviewing phase 1: PersistentPublisher held the CONNACK result on the
# instance, so a refused attempt left its reason code behind for the next one.
# The retry then either rejected a healthy connection or, worse, accepted one
# whose CONNACK had never arrived. Both need an asynchronous callback to
# reproduce, which is how real paho delivers it.
# ==============================================================================

class _AsyncAckClient(object):
    """paho-shaped stub: connect() returns at once, CONNACK lands on a timer."""

    def __init__(self, rc_sequence, calls, ack_delay=0.05):
        self._rc_sequence = rc_sequence
        self._calls = calls
        self._ack_delay = ack_delay
        self.on_connect = None

    def connect(self, host, port, keepalive=60):
        import threading
        idx = self._calls["n"]
        self._calls["n"] += 1
        rc = self._rc_sequence[min(idx, len(self._rc_sequence) - 1)]
        if rc is None:
            return  # models a broker that never acknowledges
        timer = threading.Timer(self._ack_delay,
                                lambda: self.on_connect(self, None, None, rc, None))
        timer.daemon = True
        timer.start()

    def loop_start(self):
        return None

    def loop_stop(self):
        self._calls["loop_stop"] = self._calls.get("loop_stop", 0) + 1

    def reconnect_delay_set(self, **_kw):
        return None

    def publish(self, *_a, **_k):
        from lib_py.delegation.bench import FakeInfo
        self._calls["publish"] = self._calls.get("publish", 0) + 1
        return FakeInfo()

    def disconnect(self):
        self._calls["disconnect"] = self._calls.get("disconnect", 0) + 1


def _async_factory(rc_sequence, calls):
    def factory(_role, _config):
        return _AsyncAckClient(rc_sequence, calls)
    return factory


def test_persistent_publisher_recovers_on_the_attempt_after_a_refusal():
    """A refused connection must not decide the next attempt's outcome."""
    from lib_py.delegation.publisher import PersistentPublisher

    calls = {"n": 0}
    publisher = PersistentPublisher(client_factory=_async_factory([5, 0], calls))
    try:
        ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
    finally:
        publisher.close()
    assert ok is True, err
    assert calls["n"] == 2, (
        "attempt 2 had a healthy connection and should have been the last one; "
        "got {} connect() calls".format(calls["n"])
    )
    assert calls.get("disconnect", 0) >= 1, "the refused client must be torn down, not leaked"


def test_persistent_publisher_never_accepts_a_connection_that_did_not_ack():
    """A silent broker must time out even when an earlier attempt acked."""
    from lib_py.delegation.publisher import PersistentPublisher
    import lib_py.delegation.publisher as publisher_mod

    calls = {"n": 0}
    # first connection is refused (leaves a reason code behind), second never acks
    publisher = PersistentPublisher(client_factory=_async_factory([5, None], calls))
    original = publisher_mod.CONNECT_ACK_TIMEOUT
    publisher_mod.CONNECT_ACK_TIMEOUT = 0.3   # keep the test quick
    try:
        ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
    finally:
        publisher_mod.CONNECT_ACK_TIMEOUT = original
        publisher.close()
    assert ok is False
    assert "no CONNACK" in err, (
        "the silent second connection must time out on its own evidence, not "
        "inherit the first attempt's reason code; got: {}".format(err)
    )


def test_persistent_publisher_reuses_a_healthy_connection_across_events():
    """The successful path still connects exactly once."""
    from lib_py.delegation.publisher import PersistentPublisher

    calls = {"n": 0}
    publisher = PersistentPublisher(client_factory=_async_factory([0], calls))
    try:
        for _ in range(4):
            assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
    finally:
        publisher.close()
    assert calls["n"] == 1, "a healthy connection must be reused"
    assert calls["publish"] == 4


# ==============================================================================
# 8. Cross-review regressions (job 60c980ac)
#
# R1 (codex): `python -m` puts the working directory ahead of PYTHONPATH, so a
#             frozen loop whose cwd contained a lib_py package silently ran live
#             code. The shell helpers now use an explicit bootstrap instead.
# R2 (codex): the persistent connection was reused on host/port alone, ignoring
#             credentials and TLS.
# F-1 (claude): paho queues a publish issued while disconnected and then raises
#             RuntimeError, which sat outside the retry tuple -- so the one state
#             a persistent connection exists to survive failed on first attempt.
# ==============================================================================

def _decoy_tree(root: Path) -> Path:
    """A directory that looks like a skills tree and screams if it is imported."""
    pkg = root / "lib_py" / "delegation"
    pkg.mkdir(parents=True)
    (root / "lib_py" / "__init__.py").write_text("", encoding="utf-8")
    (pkg / "__init__.py").write_text("", encoding="utf-8")
    (pkg / "cli.py").write_text(
        "def main(argv=None):\n"
        "    print('DECOY-TREE-EXECUTED')\n"
        "    return 0\n",
        encoding="utf-8",
    )
    return root


def test_r1_shell_helper_ignores_a_lib_py_in_the_working_directory(mam_sandbox, tmp_path):
    """R1: the helper must run its own tree even when cwd provides lib_py.

    This is the frozen-loop shape: a decoy package sitting in the working
    directory must not win over the tree the sourced lib.sh belongs to.
    """
    decoy = _decoy_tree(tmp_path / "decoy_ws")
    (decoy / ".mam" / "jobs").mkdir(parents=True)

    lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
    env = dict(os.environ)
    env.pop("HERDR_SESSION_NAME", None)
    env.pop("PYTHONPATH", None)
    env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")

    res = subprocess.run(
        ["bash", "-c",
         'cd "{}" && source "{}" >/dev/null 2>&1 && '
         'delegate_submit_job "r1 probe" claude-code herdr:claude'.format(decoy, lib_sh)],
        capture_output=True, text=True, env=env, timeout=120,
    )
    assert "DECOY-TREE-EXECUTED" not in res.stdout, (
        "the working directory's lib_py shadowed the helper's own tree:\n" + res.stdout
    )
    assert res.returncode == 0, res.stderr
    job_id = res.stdout.strip()
    assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
    # Issue #15: the registry follows MAM_JOBS_DIR / WORKSPACE_ROOT (both point
    # at the sandbox here), no longer the cwd — the decoy workspace's own
    # .mam/jobs must stay empty.
    registry = Path(env["MAM_JOBS_DIR"])
    assert (registry / (job_id + ".json")).exists(), (
        "the real cli should have registered the job in the sandbox registry"
    )
    assert not (decoy / ".mam" / "jobs" / (job_id + ".json")).exists(), (
        "the job must not be registered relative to the cwd"
    )


@pytest.mark.parametrize("field,first,second", [
    ("username", "account-a", "account-b"),
    ("password", "secret-a", "secret-b"),
    ("tls", False, True),
    ("ca_certs", None, "/tmp/ca.pem"),
])
def test_r2_persistent_publisher_refuses_a_different_connection_identity(field, first, second):
    """R2: same endpoint is not the same connection when auth or TLS differ."""
    from lib_py.delegation import mqtt_common, publish_event
    from lib_py.delegation.publisher import PersistentPublisher

    built = []

    def factory(_role, config):
        built.append(getattr(config, field))
        from lib_py.delegation.bench import FakeClient
        client = FakeClient(connect_latency=0.0, counters={})
        return client

    def prepared(value):
        config = mqtt_common.broker_config_from_env(
            {"host": "same.example", "port": 1883, field: value})
        return publish_event.PreparedEvent(
            "j", "progress", "d", 1, "t", False, {}, b"{}", config)

    publisher = PersistentPublisher(client_factory=factory)
    try:
        ok, _err = publisher.publish_with_retry(prepared(first), attempts=1)
        assert ok is True
        ok2, err2 = publisher.publish_with_retry(prepared(second), attempts=1)
    finally:
        publisher.close()

    assert ok2 is False, "an event with a different {} must not reuse the connection".format(field)
    assert field in err2, err2
    assert str(second) not in err2, "the error must name the field, never its value"
    assert built == [first], "the second event must not have been published on the first client"


def test_r2_identical_configuration_still_reuses_the_connection():
    """The guard must not become so strict that reuse never happens."""
    from lib_py.delegation.publisher import PersistentPublisher

    counters = {"n": 0}

    def factory(_role, _config):
        counters["n"] += 1
        from lib_py.delegation.bench import FakeClient
        return FakeClient(connect_latency=0.0, counters={})

    publisher = PersistentPublisher(client_factory=factory)
    try:
        for _ in range(3):
            assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
    finally:
        publisher.close()
    assert counters["n"] == 1


class _FlakyConnectionClient(object):
    """paho-shaped: the first N publishes are issued while disconnected."""

    def __init__(self, no_conn_times, calls):
        import paho.mqtt.client as mqtt
        self._mqtt = mqtt
        self._no_conn_times = no_conn_times
        self._calls = calls
        self.on_connect = None

    def connect(self, host, port, keepalive=60):
        self._calls["connect"] = self._calls.get("connect", 0) + 1
        self.on_connect(self, None, None, 0, None)

    def loop_start(self):
        return None

    def loop_stop(self):
        return None

    def reconnect_delay_set(self, **_kw):
        return None

    def publish(self, *_a, **_k):
        from lib_py.delegation.bench import FakeInfo
        n = self._calls.get("publish", 0)
        self._calls["publish"] = n + 1
        if n < self._no_conn_times:
            return FakeInfo(published=False, rc=int(self._mqtt.MQTT_ERR_NO_CONN))
        return FakeInfo(published=True, rc=0)

    def disconnect(self):
        return None


def test_f1_publish_issued_while_disconnected_is_retried_until_reconnect():
    """F-1: the gap between a drop and paho's reconnect must be survivable.

    paho queues such a publish and raises RuntimeError from wait_for_publish,
    which is not in the retry tuple; it used to end the attempt immediately."""
    from lib_py.delegation.publisher import PersistentPublisher

    calls = {}
    publisher = PersistentPublisher(
        client_factory=lambda _r, _c: _FlakyConnectionClient(2, calls))
    try:
        ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
    finally:
        publisher.close()
    assert ok is True, "should have recovered on the third attempt, got: {}".format(err)
    assert calls["publish"] == 3, "every attempt must reach the transport"
    assert calls["connect"] == 1, "recovery must reuse the connection, not rebuild it"


def test_f1_disconnected_publish_still_fails_when_it_never_recovers():
    """Retrying must not mask a genuinely dead connection."""
    from lib_py.delegation.publisher import PersistentPublisher

    calls = {}
    publisher = PersistentPublisher(
        client_factory=lambda _r, _c: _FlakyConnectionClient(99, calls))
    try:
        ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
    finally:
        publisher.close()
    assert ok is False
    assert "not connected" in err, err
    assert calls["publish"] == 2


def test_one_shot_publish_has_a_single_implementation():
    """O-1: the CLI path and the service path must not drift apart again."""
    from lib_py.delegation import publish_event
    from lib_py.delegation.publisher import OneShotPublisher

    source = Path(publish_event.__file__).read_text(encoding="utf-8")
    assert "OneShotPublisher().publish_raw(" in source, (
        "_publish_once must delegate to the transport module"
    )
    for leaked in ("client.loop_start()", "no CONNACK from broker", "wait_for_publish("):
        assert leaked not in source, (
            "publish_event still carries its own copy of the transport logic: {}".format(leaked)
        )
    assert publish_event.CONNECT_ACK_TIMEOUT is not None
    assert hasattr(OneShotPublisher, "publish_raw")