"""Issue #30: a stall watcher that only notifies.

The watcher is a real `python -m lib_py.stall_watch` process against
a job registered by the real wrapper (`submit --dry-run`) and moved by the real
producer (`publish_event.py`); the broker is unreachable, so publish_event
exits 2 after the record and events.log were written. herdr is the conftest
mock, whose `pane process-info` answer has the shape of a real herdr 0.9.1
answer (S1), and whose `agent read` text is a real idle-pane capture
(tests/fixtures/panes, S14). The agent process and its children are real
processes, so the descendant scan reads the real /proc.
"""

from __future__ import annotations

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

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
STALL_WATCH = SKILLS / "lib_py" / "stall_watch.py"
PANES = REPO_ROOT / "tests" / "fixtures" / "panes"
if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

from lib_py.agents.input_region import extract_input_region  # noqa: E402
from lib_py import stall_watch  # noqa: E402
from lib_py.loop import watcher  # noqa: E402

pytestmark = pytest.mark.skipif(not sys.platform.startswith("linux"), reason="real process trees via /proc")

UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}

# `herdr pane process-info --pane w1:p1` on herdr 0.9.1, 2026-09-25 (plan F11).
REAL_PROCESS_INFO = {"result": {"process_info": {
    "foreground_process_group_id": 11407,
    "foreground_processes": [{"argv": ["claude"], "cmdline": "claude --dangerously-skip-permissions",
                              "cwd": "/home/u/ws", "name": "claude", "pid": 11407}],
    "pane_id": "w1:p1", "shell_pid": 8187}, "type": "pane_process_info"}}

# The input line of each idle capture, and a synthetic "someone is typing"
# variant of it (no real typing capture was taken: typing into another agent's
# pane was out of scope for the capture job).
TYPING = {"claude": (r"^❯\s*$", "❯ please continue"), "agy": (r"^>\s*$", "> please continue"),
          "grok": (r"│ ❯ +", "│ ❯ please continue "), "opencode": (r"^  ┃\s*$", "  ┃  please continue"),
          "codex": (r"^› Ask Codex to do anything", "› please continue")}


def _pane(agent: str, typing: bool = False) -> str:
    text = "".join(l for l in (PANES / f"{agent}_idle.txt").read_text().splitlines(True) if not l.startswith("# "))
    if typing:
        pat, rep = TYPING[agent]
        typed = re.sub(pat, rep, text, count=1, flags=re.M)
        assert typed != text, agent
        return typed
    return text


# ---------------------------------------------------------------------------
# S14 - the input-box facts are pinned to real captures
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("agent", sorted(TYPING))
def test_s14_a_real_idle_capture_has_an_empty_input_box_and_typing_does_not(agent):
    assert extract_input_region(_pane(agent), agent) == ""
    assert extract_input_region(_pane(agent, typing=True), agent) == "please continue"


def test_s14_rule_agents_keep_their_path_and_an_agent_without_facts_raises():
    ruled = "old\n" + "─" * 20 + "\n❯ \n" + "─" * 20 + "\nfooter"
    assert extract_input_region(ruled, "hermes") == ""
    with pytest.raises(ValueError):
        extract_input_region(ruled, "no-such-agent")
    # no whole-pane fallback: a pane without any input line is not "empty"
    assert extract_input_region("just a transcript\n", "codex") != ""
    assert extract_input_region("just a transcript\n", "opencode") != ""


# ---------------------------------------------------------------------------
# judge - the pure decision, as a table
# ---------------------------------------------------------------------------

def _sig(**kw):
    base = dict(agent_status="idle", idle_for=10.0, agent_pid=123, background_children=[],
                input_empty=True, job_status="running")
    base.update(kw)
    return stall_watch.Signals(**base)


@pytest.mark.parametrize("sig,st,want", [
    (_sig(), {}, stall_watch.NOTIFY_IDLE_AND_NUDGE),
    (_sig(agent_status="done"), {}, stall_watch.NOTIFY_IDLE_AND_NUDGE),
    (_sig(idle_for=1.0), {}, None),
    (_sig(background_children=["9:sleep"]), {}, None),
    (_sig(input_empty=False), {}, None),
    (_sig(input_empty=None), {}, None),
    (_sig(agent_pid=None), {}, None),
    (_sig(job_status="completed"), {}, None),
    (_sig(), {"nudged": True}, None),
    (_sig(), {"reliable": False}, None),
    (_sig(agent_status="working"), {}, None),
    (_sig(agent_status="blocked"), {}, stall_watch.NOTIFY_BLOCKED),
    (_sig(agent_status="blocked"), {"reliable": False}, stall_watch.NOTIFY_BLOCKED),
    (_sig(agent_status="blocked"), {"blocked_notified": True}, None),
])
def test_judge_table(sig, st, want):
    assert stall_watch.judge(sig, stall_watch.WatchState(idle_s=5.0, **st)) == want


# ---------------------------------------------------------------------------
# harness
# ---------------------------------------------------------------------------

def _env(sandbox: Path, **extra) -> dict:
    env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}
    env.update(UNREACHABLE, DELEGATE_JOB_PYTHON=sys.executable, PYTHONPATH=str(SKILLS), MAM_JOB_ID="",
               DELEGATE_JOB_LOGS_DIR=str(sandbox / ".mam" / "logs"))
    env.update({k: str(v) for k, v in extra.items()})
    return env


def _job(env: dict, reg: Path) -> str:
    res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:img",
                          "--dry-run", "--registry-dir", str(reg), "--prompt", "work"],
                         capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=120)
    assert res.returncode == 0, res.stdout + res.stderr
    jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
    _publish(env, reg, jid, "started")
    assert _record(reg, jid)["status"] == "running"
    return jid


def _record(reg: Path, jid: str) -> dict:
    return json.loads((reg / f"{jid}.json").read_text())


def _publish(env: dict, reg: Path, jid: str, event: str, *extra: str) -> None:
    res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid, "--event", event,
                          "--expect-dispatch-token", _record(reg, jid)["dispatch_token"], "--attempts", "1", *extra],
                         capture_output=True, text=True, env=env, timeout=120)
    assert res.returncode in (0, 2), res.stderr


def _details(reg: Path, jid: str) -> list:
    log = reg / f"{jid}.events.log"
    if not log.exists():
        return []
    out = []
    for line in log.read_text().splitlines():
        try:
            ev = json.loads(line)
        except ValueError:
            continue
        if ev.get("event") == "progress":
            out.append(ev.get("detail", ""))
    return out


def _prompts(state_file: Path) -> list:
    return [c for c in json.loads(state_file.read_text()).get("calls", []) if c[:2] == ["agent", "prompt"]]


def _agent(state_file: Path, name: str, agent: str, pid: int, status: str, buffer: str, detection: str = ""):
    state = json.loads(state_file.read_text())
    state["wait_blocks_s"] = 0.5
    state.setdefault("agents", {})[name] = {
        "name": name, "agent": agent, "status": "running", "agent_status": status, "pane_id": "w1:p1",
        "workspace_id": "w1", "pid": pid, "buffer": buffer, "detection": detection}
    state_file.write_text(json.dumps(state, indent=2))


def _proc(script: str) -> subprocess.Popen:
    p = subprocess.Popen(["bash", "-c", script], start_new_session=True)
    time.sleep(0.3)  # let `exec` and the background child settle
    return p


def _kill(p: subprocess.Popen) -> None:
    try:
        os.killpg(p.pid, 9)
    except ProcessLookupError:
        pass
    p.wait()


def _watch(env: dict, reg: Path, jid: str, target: str, agent: str) -> subprocess.Popen:
    tok = _record(reg, jid)["dispatch_token"]
    return subprocess.Popen([sys.executable, "-m", "lib_py.stall_watch", "--job", jid,
                             "--registry-dir", str(reg), "--target", target, "--agent", agent,
                             "--expect-dispatch-token", tok],
                            env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)


def _until(pred, limit=20.0):
    end = time.monotonic() + limit
    while time.monotonic() < end:
        if pred():
            return True
        time.sleep(0.2)
    return pred()


def _finish(env, reg, jid, w: subprocess.Popen) -> str:
    """Complete the job for real: the watcher sees the terminal record and exits 0."""
    _publish(env, reg, jid, "completed")
    out, _ = w.communicate(timeout=30)
    assert w.returncode == 0, out
    return out


@pytest.fixture
def box(mam_sandbox, mock_herdr):
    reg = mam_sandbox / ".mam" / "jobs"
    env = _env(mam_sandbox, MAM_STALL_IDLE_S=2)
    return mam_sandbox, mock_herdr, reg, env


# ---------------------------------------------------------------------------
# S1 - the mock answers in the shape of the real herdr
# ---------------------------------------------------------------------------

def test_s1_mock_process_info_has_the_real_herdr_shape(box):
    _, state, _, env = box
    _agent(state, "creator-claude-t", "claude", 4242, "idle", "Ready")
    out = subprocess.run(["herdr", "pane", "process-info", "--pane", "w1:p1"], capture_output=True, text=True,
                         env=env).stdout
    mock = json.loads(out)
    real, got = REAL_PROCESS_INFO["result"], mock["result"]
    assert set(real) <= set(got)
    assert set(real["process_info"]) <= set(got["process_info"])
    assert set(real["process_info"]["foreground_processes"][0]) <= set(got["process_info"]["foreground_processes"][0])
    assert got["process_info"]["foreground_processes"][0]["pid"] == 4242


# ---------------------------------------------------------------------------
# S2 - blocked: reported once with the dialog text, never answered
# ---------------------------------------------------------------------------

def test_s2_blocked_is_reported_once_with_the_dialog_and_never_answered(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("exec sleep 300")
    try:
        _agent(state, "creator-claude-t", "claude", agent.pid, "blocked", _pane("claude"),
               detection="\n Dangerous rm operation detected. Do you want to proceed?\n 1. Yes\n")
        w = _watch(env, reg, jid, "creator-claude-t", "claude")
        assert _until(lambda: _details(reg, jid))
        time.sleep(2)
        assert _details(reg, jid) == ["stalled:blocked Dangerous rm operation detected. Do you want to proceed?"]
        calls = json.loads(state.read_text())["calls"]
        assert not [c for c in calls if c[:2] == ["agent", "prompt"] or "send-keys" in c]
        assert _record(reg, jid)["status"] == "running"
        out = _finish(env, reg, jid, w)
        assert "WARNING" in out and "blocked" in out
    finally:
        _kill(agent)


# ---------------------------------------------------------------------------
# S3/S4/S5 - idle: nudged once only without background work
# ---------------------------------------------------------------------------

def test_s3_idle_without_children_is_nudged_exactly_once(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("exec sleep 300")
    try:
        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"))
        w = _watch(env, reg, jid, "creator-claude-t", "claude")
        assert _until(lambda: _prompts(state))
        time.sleep(4)
        assert _details(reg, jid) == ["stalled:idle"]
        prompts = _prompts(state)
        assert len(prompts) == 1
        assert prompts[0][2] == "creator-claude-t" and prompts[0][3].startswith(f"Job {jid}: you look idle")
        assert _record(reg, jid)["status"] == "running"
        _finish(env, reg, jid, w)
    finally:
        _kill(agent)


def test_s4_idle_with_a_background_child_is_left_alone(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("sleep 300 & exec sleep 300")
    try:
        assert stall_watch.background_children(agent.pid)
        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"))
        w = _watch(env, reg, jid, "creator-claude-t", "claude")
        time.sleep(6)
        assert not [d for d in _details(reg, jid) if d.startswith("stalled:")]
        assert not _prompts(state)
        _finish(env, reg, jid, w)
    finally:
        _kill(agent)


@pytest.mark.parametrize("agent_name,nudged", [("codex", True), ("claude", False)])
def test_s5_the_adapters_resident_children_change_the_judgement(box, agent_name, nudged):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("(exec -a codex-code-mode-host sleep 300) & exec sleep 300")
    try:
        _agent(state, f"reviewer-{agent_name}-t", agent_name, agent.pid, "idle", _pane(agent_name))
        w = _watch(env, reg, jid, f"reviewer-{agent_name}-t", agent_name)
        if nudged:
            assert _until(lambda: _prompts(state))
        else:
            time.sleep(6)
        assert len(_prompts(state)) == (1 if nudged else 0)
        assert (_details(reg, jid) == ["stalled:idle"]) is nudged
        _finish(env, reg, jid, w)
    finally:
        _kill(agent)


# ---------------------------------------------------------------------------
# S6 - the heartbeat timer is never background work
# ---------------------------------------------------------------------------

def test_s6_the_heartbeat_timer_is_not_a_background_child(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    tok = _record(reg, jid)["dispatch_token"]
    publish = (f"{sys.executable} {PUBLISH} --registry-dir {reg} --job {jid} --event progress --detail x "
               f"--expect-dispatch-token {tok} --attempts 1 --heartbeat 1; exec sleep 300")
    agent = subprocess.Popen(["bash", "-c", publish], env=env, start_new_session=True)
    timer = []
    try:
        def _timer():
            out = subprocess.run(["pgrep", "-f", f"lib_py.delegation.heartbeat --job {jid}"],
                                 capture_output=True, text=True).stdout.split()
            timer[:] = [int(p) for p in out]
            return timer
        assert _until(_timer, 30)
        assert _until(lambda: open(f"/proc/{agent.pid}/cmdline", "rb").read().startswith(b"sleep"), 30)
        kids = stall_watch.background_children(agent.pid)
        assert not [k for k in kids if int(k.split(":")[0]) in timer], kids
    finally:
        for pid in timer:
            try:
                os.kill(pid, 9)
            except ProcessLookupError:
                pass
        _kill(agent)


# ---------------------------------------------------------------------------
# S7 - someone typing, or no input facts: no nudge, and the latter says so
# ---------------------------------------------------------------------------

def test_s7_a_typed_input_box_withholds_the_nudge(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("exec sleep 300")
    try:
        _agent(state, "reviewer-grok-t", "grok", agent.pid, "idle", _pane("grok", typing=True))
        w = _watch(env, reg, jid, "reviewer-grok-t", "grok")
        time.sleep(6)
        assert _details(reg, jid) == []
        assert not _prompts(state)
        out = _finish(env, reg, jid, w)
        assert "not empty" in out
    finally:
        _kill(agent)


def test_s7b_an_agent_without_input_facts_is_withheld_and_reported_once(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    agent = _proc("exec sleep 300")
    try:
        _agent(state, "worker-x", "noinput", agent.pid, "idle", "whatever")
        w = _watch(env, reg, jid, "worker-x", "noinput")
        assert _until(lambda: _details(reg, jid))
        time.sleep(5)
        assert _details(reg, jid) == ["stall_watch:input_check_unavailable:noinput"]
        assert not _prompts(state)
        out = _finish(env, reg, jid, w)
        assert out.count("input check unavailable") == 1
    finally:
        _kill(agent)


# ---------------------------------------------------------------------------
# S8 - a failing herdr is never "no event"
# ---------------------------------------------------------------------------

def test_s8_herdr_failure_is_reported_and_the_watcher_stops_with_rc_2(box):
    _, state, reg, env = box
    jid = _job(env, reg)
    st = json.loads(state.read_text())
    st["force_error_code"] = "protocol_mismatch"
    state.write_text(json.dumps(st))
    env["MAM_STALL_HERDR_FAIL_S"] = "3"
    w = _watch(env, reg, jid, "creator-claude-t", "claude")
    out, _ = w.communicate(timeout=60)
    assert w.returncode == 2, out
    assert _details(reg, jid) == ["stall_watch:herdr_unavailable:protocol_mismatch"]
    assert "herdr unavailable" in out and "WARNING" in out
    assert _record(reg, jid)["status"] == "running"


# ---------------------------------------------------------------------------
# S12 - the watcher has no path that closes a job
# ---------------------------------------------------------------------------

def test_s12_the_watcher_can_only_publish_progress():
    tree = ast.parse(STALL_WATCH.read_text())
    names = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}
    attrs = {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)}
    imported = {a.name for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom)) for a in n.names}
    froms = {n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)}
    for forbidden in ("close", "update_status", "_operator_write"):
        assert forbidden not in names | attrs | imported, forbidden
    assert "lib_py.delegation.registry" not in froms and "registry" not in imported
    # every `--event` in the file is followed by the literal "progress"
    events = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.List, ast.Tuple)):
            elts = node.elts
            for i, e in enumerate(elts[:-1]):
                if isinstance(e, ast.Constant) and e.value == "--event":
                    events.append(getattr(elts[i + 1], "value", None))
    assert events == ["progress"]
    publishers = [n for n in ast.walk(tree) if isinstance(n, ast.Constant) and n.value == "lib_py.delegation.publish_event"]
    assert len(publishers) == 1


# ---------------------------------------------------------------------------
# S13 - an adapter whose agent_status is unreliable gets no idle judgement
# ---------------------------------------------------------------------------

def test_s13_unreliable_agent_status_turns_idle_off_but_keeps_blocked(box):
    _, state, reg, env = box
    env["MAM_TEST_UNRELIABLE_AGENT"] = "claude"
    jid = _job(env, reg)
    agent = _proc("exec sleep 300")
    try:
        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"),
               detection="Do you want to proceed?")
        w = _watch(env, reg, jid, "creator-claude-t", "claude")
        time.sleep(5)
        assert _details(reg, jid) == [] and not _prompts(state)
        st = json.loads(state.read_text())
        st["agents"]["creator-claude-t"]["agent_status"] = "blocked"
        state.write_text(json.dumps(st))
        assert _until(lambda: _details(reg, jid))
        assert _details(reg, jid) == ["stalled:blocked Do you want to proceed?"]
        out = _finish(env, reg, jid, w)
        assert "idle judgement off for claude" in out
    finally:
        _kill(agent)


# ---------------------------------------------------------------------------
# scope - the #34 rule reaches the python herdr calls
# ---------------------------------------------------------------------------

def test_scoped_env_drops_the_pane_inherited_workspace_for_name_lookups_only():
    pane = {"HERDR_SESSION": "multi-agent-mux", "HERDR_WORKSPACE_ID": "w6", "HERDR_PANE_ID": "w6:p1"}
    by_name = watcher.scoped_env("multi-agent-mux", environ=pane, argv=["herdr", "agent", "wait", "rev"])
    assert "HERDR_WORKSPACE_ID" not in by_name and by_name["HERDR_PANE_ID"] == "w6:p1"
    by_pane = watcher.scoped_env("multi-agent-mux", environ=pane, argv=["herdr", "pane", "read", "w2:p1"])
    assert by_pane["HERDR_WORKSPACE_ID"] == "w6"
    chosen = dict(pane, HERDR_WORKSPACE_ID="w5")
    assert watcher.scoped_env("multi-agent-mux", environ=chosen, argv=["herdr", "agent", "get", "rev"])[
        "HERDR_WORKSPACE_ID"] == "w5"
    assert watcher.scoped_env(None, environ={"HERDR_WORKSPACE_ID": "w6"},
                              argv=["herdr", "agent", "get", "rev"])["HERDR_WORKSPACE_ID"] == "w6"


# ---------------------------------------------------------------------------
# S9 / S10 - the real wrapper starts and stops the watcher (the #18 harness)
# ---------------------------------------------------------------------------

import test_issue18_liveness_contract as t18  # noqa: E402


def _watchers(jid: str) -> list:
    out = subprocess.run(["pgrep", "-f", f"lib_py.stall_watch --job {jid}"], capture_output=True, text=True).stdout
    return [int(p) for p in out.split()]


@pytest.mark.parametrize("switch", ["1", "0"])
def test_s9_the_wrapper_runs_the_watcher_for_the_jobs_life(tmp_path, mock_herdr, switch):
    reg = tmp_path / "jobs"
    reg.mkdir()
    t18._headless_agent(Path(mock_herdr), "headless-codex")
    proc, env = t18._submit(tmp_path, reg, "herdr:headless-codex", "codex", ["--ack-timeout", "40"],
                            env_extra={"MAM_STALL_WATCH": switch}, timeout_s=60)
    jid = t18._wait_job_id(reg)
    try:
        time.sleep(1)
        assert t18._publish(reg, jid, "started", env).returncode == 0
        if switch == "1":
            assert _until(lambda: (reg / f"{jid}.stall.out").exists() and _watchers(jid), 20), "no watcher"
        else:
            time.sleep(2)
        assert t18._publish(reg, jid, "completed", env, "--detail", "done").returncode == 0
        stdout, stderr = proc.communicate(timeout=60)
    finally:
        if proc.poll() is None:
            proc.kill()
    assert proc.returncode == 0, stdout + stderr
    assert _watchers(jid) == [], "the watcher outlived its job"
    assert (reg / f"{jid}.stall.out").exists() is (switch == "1")
    assert "stall watch log" not in stderr  # a completed job carries no watcher log


@pytest.mark.parametrize("agent,session,why", [("human", "herdr:human", "human job"),
                                               ("claude", "remote:far-01", "remote job")])
def test_s10_remote_and_human_jobs_are_not_watched(tmp_path, agent, session, why):
    reg = tmp_path / "jobs"
    reg.mkdir()
    proc, env = t18._submit(tmp_path, reg, session, agent, ["--ack-timeout", "40"], timeout_s=60,
                            env_extra={"MAM_DISPATCH_SECRET": "s10-secret"})
    jid = t18._wait_job_id(reg)
    try:
        time.sleep(1)
        assert t18._publish(reg, jid, "started", env).returncode == 0
        time.sleep(1)
        assert t18._publish(reg, jid, "completed", env, "--detail", "done").returncode == 0
        stdout, stderr = proc.communicate(timeout=60)
    finally:
        if proc.poll() is None:
            proc.kill()
    assert proc.returncode == 0, stdout + stderr
    assert f"stall watch: {why}, no local pane - stall detection off" in stderr
    assert not (reg / f"{jid}.stall.out").exists()


# ---------------------------------------------------------------------------
# S11 - the loop runner relays the watcher's notices once, and keeps waiting
# ---------------------------------------------------------------------------

def test_s11_the_runner_warns_once_per_stall_notice(mam_sandbox, monkeypatch, capsys):
    reg = mam_sandbox / ".mam" / "jobs"
    env = _env(mam_sandbox)
    jid = _job(env, reg)
    for d in ("stalled:idle", "stalled:idle", "stall_watch:herdr_unavailable:protocol_mismatch", "ordinary"):
        _publish(env, reg, jid, "progress", "--detail", d)
    monkeypatch.setenv("WORKSPACE_ROOT", str(mam_sandbox))
    monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
    from lib_py.loop import runner as lr
    r = lr.LoopRunner(["--creator", "c", "--task", "t"])
    r.parse_args()
    import threading
    threading.Timer(2.5, lambda: _publish(env, reg, jid, "completed")).start()
    assert r.wait_for_job(jid, max_wait=30, check_interval=0.3) is True
    out = capsys.readouterr()
    text = out.out + out.err
    assert text.count(f"job {jid}: stalled:idle") == 1, text
    assert text.count(f"job {jid}: stall_watch:herdr_unavailable:protocol_mismatch") == 1, text
    assert "ordinary" not in text