multi-agent-mux/tests/test_issue66_stall_watch_sanitized_name.py

"""Issue #66: stall_watch calls herdr with the sanitized agent name.

herdr registers an agent under `sanitize_herdr_agent_name(session)` (lower
case, at most 32 characters). The watcher gets the original session name from
the wrapper, so it must convert it for every herdr call while its logs keep
the original name and the agent kind is still read off the original suffix.

The harness is the one of test_issue30_stall_watch (a real watcher process, a
real job, the conftest mock herdr), with the mock switched to
`strict_agent_names`: like the real herdr it answers an agent only under its
registered (sanitized) name, so the original name gets `agent_not_found`.
"""

from __future__ import annotations

import json
import subprocess
import sys
import time
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent))
import test_issue30_stall_watch as t30  # noqa: E402

from lib_py.agents.sanitize import sanitize_herdr_agent_name  # noqa: E402

LONG = "creator-claude-fix-issue-99-long-branch"  # 39 characters
UPPER = "Creator-Claude-T"
PLAIN = "creator-claude-t"  # needs no conversion
LONG_CODEX = "mam-wt-issue-66-very-long-creator-codex"  # kind only in the original suffix

DIALOG = "\n Dangerous rm operation detected. Do you want to proceed?\n 1. Yes\n"


@pytest.fixture
def strict_box(mam_sandbox, mock_herdr):
    state = json.loads(mock_herdr.read_text())
    state["strict_agent_names"] = True
    mock_herdr.write_text(json.dumps(state, indent=2))
    reg = mam_sandbox / ".mam" / "jobs"
    return mock_herdr, reg, t30._env(mam_sandbox, MAM_STALL_IDLE_S=2)


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


def test_the_names_change_and_the_strict_mock_answers_only_the_sanitized_one(strict_box):
    state, _, env = strict_box
    assert len(LONG) > 32 and sanitize_herdr_agent_name(LONG) != LONG
    assert sanitize_herdr_agent_name(UPPER) == PLAIN
    assert sanitize_herdr_agent_name(PLAIN) == PLAIN
    for name in (LONG, UPPER):
        herdr_name = sanitize_herdr_agent_name(name)
        t30._agent(state, herdr_name, "claude", 4242, "idle", "Ready")
        for cmd in (["get"], ["read"], ["wait", "--until", "idle"], ["prompt"]):
            argv = ["herdr", "agent", cmd[0], name] + cmd[1:] + (["hi"] if cmd == ["prompt"] else [])
            res = subprocess.run(argv, capture_output=True, text=True, env=env)
            assert res.returncode == 1, (argv, res.stdout)
            assert json.loads(res.stderr)["error"]["code"] == "agent_not_found", argv
            argv[3] = herdr_name
            assert subprocess.run(argv, capture_output=True, text=True, env=env).returncode == 0, argv


@pytest.mark.parametrize("target", [LONG, UPPER, PLAIN])
def test_an_idle_agent_is_nudged_through_the_sanitized_name(strict_box, target):
    state, reg, env = strict_box
    herdr_name = sanitize_herdr_agent_name(target)
    jid = t30._job(env, reg)
    agent = t30._proc("exec sleep 300")
    try:
        t30._agent(state, herdr_name, "claude", agent.pid, "idle", t30._pane("claude"))
        w = t30._watch(env, reg, jid, target, "claude")
        assert t30._until(lambda: t30._prompts(state))
        time.sleep(2)
        assert t30._details(reg, jid) == ["stalled:idle"]
        calls = _agent_calls(state)
        assert {c[1] for c in calls} == {"get", "read", "prompt", "wait"}, calls
        assert all(c[2] == herdr_name for c in calls), calls
        assert len(t30._prompts(state)) == 1
        out = t30._finish(env, reg, jid, w)
        assert f"agent {target} idle for" in out
    finally:
        t30._kill(agent)


@pytest.mark.parametrize("target", [LONG, UPPER])
def test_a_blocked_agent_is_reported_with_the_original_name(strict_box, target):
    state, reg, env = strict_box
    herdr_name = sanitize_herdr_agent_name(target)
    jid = t30._job(env, reg)
    agent = t30._proc("exec sleep 300")
    try:
        t30._agent(state, herdr_name, "claude", agent.pid, "blocked", t30._pane("claude"), detection=DIALOG)
        w = t30._watch(env, reg, jid, target, "claude")
        assert t30._until(lambda: t30._details(reg, jid))
        time.sleep(2)
        assert t30._details(reg, jid) == ["stalled:blocked Dangerous rm operation detected. Do you want to proceed?"]
        calls = _agent_calls(state)
        assert calls and all(c[2] == herdr_name for c in calls), calls
        assert not t30._prompts(state)
        out = t30._finish(env, reg, jid, w)
        assert f"agent {target} is blocked" in out
    finally:
        t30._kill(agent)


def test_the_kind_of_a_long_name_comes_from_its_original_suffix(strict_box):
    """No usable --agent: the kind comes from the session name. Read off the
    sanitized name (a hash at the end) it would be unknown, and the nudge would
    be withheld as `input_check_unavailable`."""
    state, reg, env = strict_box
    herdr_name = sanitize_herdr_agent_name(LONG_CODEX)
    assert not herdr_name.endswith("-codex")
    jid = t30._job(env, reg)
    agent = t30._proc("exec sleep 300")
    try:
        t30._agent(state, herdr_name, "codex", agent.pid, "idle", t30._pane("codex"))
        w = t30._watch(env, reg, jid, LONG_CODEX, "")
        assert t30._until(lambda: t30._prompts(state))
        time.sleep(2)
        assert t30._details(reg, jid) == ["stalled:idle"]
        assert all(c[2] == herdr_name for c in _agent_calls(state))
        t30._finish(env, reg, jid, w)
    finally:
        t30._kill(agent)