multi-agent-mux/tests/test_b19_headless_reconcile_fixes.py

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

from lib_py.layout import compute_2xk_layout


def test_bug2_headless_layout_does_not_overflow():
    """Verify Bug 2: w=0, h=0 in headless mode does not trigger overflow."""
    # 1. Headless 0x0
    payload_0x0 = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 0, "height": 0}}]}}
    d_0x0 = compute_2xk_layout(payload_0x0)
    assert not d_0x0.is_overflow, f"Headless 0x0 should not overflow, got {d_0x0}"
    assert d_0x0.direction in ("right", "down"), f"Headless 0x0 direction must be right or down, got {d_0x0.direction}"

    # 2. Genuine small pane (overflow)
    payload_small = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 50, "height": 30}}]}}
    d_small = compute_2xk_layout(payload_small, min_cols=60, min_rows=20)
    assert d_small.is_overflow, f"Small pane should be overflow, got {d_small}"
    assert d_small.direction == "overflow"

    # 3. Wide pane (split right)
    payload_wide = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 160, "height": 30}}]}}
    d_wide = compute_2xk_layout(payload_wide, min_cols=60, min_rows=20)
    assert not d_wide.is_overflow
    assert d_wide.direction == "right"

    # 4. Tall-but-narrow pane: N=1 is always right; 80//2 < min_cols 60 → overflow
    payload_tall = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 80, "height": 60}}]}}
    d_tall = compute_2xk_layout(payload_tall, min_cols=60, min_rows=20)
    assert d_tall.is_overflow
    assert d_tall.direction == "overflow"

    payload_tall_wide = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 160, "height": 60}}]}}
    d_tall_wide = compute_2xk_layout(payload_tall_wide, min_cols=60, min_rows=20)
    assert not d_tall_wide.is_overflow
    assert d_tall_wide.direction == "right"


def test_bug3_reconcile_skills_dir_passed_and_fallback():
    """Verify Bug 3: reconcile.sh evaluates valid SKILLS_DIR and passes it to Python subshells."""
    recon_path = os.path.abspath(".agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh")
    with open(recon_path, "r", encoding="utf-8") as f:
        content = f.read()

    # Assert SKILLS_DIR is explicitly passed to env_python / atomic_dump_yaml
    assert 'SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" env_python' in content
    assert 'SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" atomic_dump_yaml' in content

    # Read the actual line from reconcile.sh and verify it uses && pwd instead of || pwd
    line19 = next(l for l in content.splitlines() if l.startswith("SKILLS_DIR="))
    assert "&& pwd" in line19 and "|| pwd" not in line19, f"Invalid SKILLS_DIR evaluation: {line19}"

    # Functionally evaluate that exact line from reconcile.sh in bash
    script = f"""#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(dirname "{recon_path}")"
{line19}
echo "RESOLVED_SKILLS_DIR=$SKILLS_DIR"
if [ -z "$SKILLS_DIR" ]; then
  echo "ERROR: SKILLS_DIR is empty" >&2
  exit 1
fi
if [ ! -d "$SKILLS_DIR" ]; then
  echo "ERROR: directory does not exist" >&2
  exit 1
fi
if [ ! -f "$SKILLS_DIR/lib.sh" ]; then
  echo "ERROR: lib.sh missing" >&2
  exit 1
fi
"""
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
    assert res.returncode == 0, f"Script failed: {res.stderr}"
    assert "RESOLVED_SKILLS_DIR=" in res.stdout


def test_bug4_send_keys_safe_gating_order():
    """Verify Bug 4: send_keys_safe checks quiescence and dialogs before agent prompt fast-path."""
    lib_path = os.path.abspath(".agents/skills/lib.sh")
    with open(lib_path, "r", encoding="utf-8") as f:
        content = f.read()

    sks_idx = content.find("send_keys_safe() {")
    assert sks_idx != -1

    sks_body = content[sks_idx:sks_idx + 2500]
    quiescent_idx = sks_body.find("_pane_quiescent")
    dialog_idx = sks_body.find("_pane_dialog_open")
    prompt_idx = sks_body.find("agent prompt")

    assert quiescent_idx != -1, "_pane_quiescent not found in send_keys_safe"
    assert dialog_idx != -1, "_pane_dialog_open not found in send_keys_safe"
    assert prompt_idx != -1, "agent prompt not found in send_keys_safe"

    # Ordering check: quiescence and dialog checks MUST precede agent prompt
    assert quiescent_idx < prompt_idx, "_pane_quiescent must execute before agent prompt fast-path"
    assert dialog_idx < prompt_idx, "_pane_dialog_open must execute before agent prompt fast-path"


_LIB_SH = str(Path(__file__).resolve().parent.parent / ".agents" / "skills" / "lib.sh")

_FULLSCREEN_TIP = """Claude Code
Try the new fullscreen renderer — flicker-free output, mouse support, auto-copy on select · /tui fullscreen
❯
"""

_FULLSCREEN_MODAL = """Try the new fullscreen renderer?
Flicker-free output
Selected text auto-copies to your clipboard
Yes, try it
"""


def _run_lib_helpers(script_body, env_extra=None):
    env = dict(os.environ)
    if env_extra:
        env.update(env_extra)
    wrapper = f"""
set -euo pipefail
source "{_LIB_SH}"
_init_herdr_isolation() {{ :; }}
{script_body}
"""
    return subprocess.run(["bash", "-c", wrapper], capture_output=True, text=True, env=env)


def test_agent_start_success_tokens_exclude_startup_timeout():
    """D-1: dead-process timeout must not be promoted to success; agent_not_ready may."""
    content = open(_LIB_SH, encoding="utf-8").read()
    success_line = next(l for l in content.splitlines()
                        if 'grep -qE "agent_started' in l)
    assert "agent_not_ready" in success_line
    assert "timed out waiting for agent startup" not in success_line
    err_line = next(i for i, l in enumerate(content.splitlines())
                    if 'grep -qiE "^usage:' in l)
    ok_line = next(i for i, l in enumerate(content.splitlines())
                   if 'grep -qE "agent_started' in l)
    assert err_line < ok_line


def test_fullscreen_tip_is_not_a_blocking_dialog():
    """D-2: idle /tui fullscreen tip must not trip _pane_dialog_open or send keys."""
    script = f"""
_pane_capture() {{ printf '%s' '{_FULLSCREEN_TIP}'; }}
KEYS=/tmp/mam-fs-tip-keys.$$
: > "$KEYS"
_sks_herdr() {{
  if [ "${{1:-}}" = "send-keys" ]; then echo "$*" >> "$KEYS"; fi
  return 0
}}
if _pane_dialog_open dummy; then echo "DIALOG_OPEN"; else echo "DIALOG_CLOSED"; fi
handle_startup_dialogs dummy 2
echo "KEYS_CONTENT=$(tr '\\n' '|' < "$KEYS")"
rm -f "$KEYS"
"""
    res = _run_lib_helpers(script)
    assert res.returncode == 0, res.stderr + res.stdout
    assert "DIALOG_CLOSED" in res.stdout
    assert "Escape" not in res.stdout
    assert "Enter" not in res.stdout.split("KEYS_CONTENT=")[-1]


def _lib_helper_src():
    content = open(_LIB_SH, encoding="utf-8").read()
    start = content.find("_herdr_ws_id_file() {")
    end = content.find('cmd="${1:-}"', start)
    assert start != -1 and end != -1
    return content[start:end]


def test_resolve_pane_id_fails_cleanly_under_set_e():
    """D-4: _resolve_herdr_pane_id must return 1 with empty stdout under set -e."""
    helper = _lib_helper_src()
    script = """
set -euo pipefail
unset HERDR_WORKSPACE_ID
source "LIB_SH_PLACEHOLDER"
_init_herdr_isolation() { :; }
_real_herdr() { return 1; }
HELPER_PLACEHOLDER
_resolve_herdr_pane_id nonexistent >/dev/null 2>&1 || true
echo SURVIVED
rc=0
out=$(_resolve_herdr_pane_id nonexistent 2>/dev/null) || rc=$?
printf 'OUT=%s\\n' "$out"
echo "RC=$rc"
""".replace("LIB_SH_PLACEHOLDER", _LIB_SH).replace("HELPER_PLACEHOLDER", helper)
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True)
    assert res.returncode == 0, res.stderr + res.stdout
    assert "SURVIVED" in res.stdout
    assert "RC=1" in res.stdout
    out_line = [l for l in res.stdout.splitlines() if l.startswith("OUT=")][0]
    assert out_line == "OUT="


def test_send_keys_safe_returns_3_when_paste_buffer_fails():
    """D-5 (ISSUE-1): paste-buffer failure returns 3 and still deletes the buffer."""
    script = """
_pane_quiescent() { return 0; }
_pane_dialog_open() { return 1; }
LOG=/tmp/mam-d5-sks.$$
: > "$LOG"
_sks_herdr() {
  echo "$*" >> "$LOG"
  if [ "${1:-}" = "agent" ] && [ "${2:-}" = "prompt" ]; then return 1; fi
  if [ "${1:-}" = "paste-buffer" ]; then return 1; fi
  return 0
}
rc=0
send_keys_safe "d5-sess" "hello" "job-d5" || rc=$?
echo "RC=$rc"
echo "LOG_CONTENT=$(tr '\\n' '|' < "$LOG")"
rm -f "$LOG"
"""
    res = _run_lib_helpers(script)
    assert res.returncode == 0, res.stderr + res.stdout
    assert "RC=3" in res.stdout
    log = res.stdout.split("LOG_CONTENT=")[-1]
    assert "C-m" not in log
    assert "delete-buffer" in log


def test_no_substring_matching_remains_in_lib_sh():
    """D-6 (ISSUE-2): no `in tn` substring matching remains in lib.sh."""
    import re
    content = open(_LIB_SH, encoding="utf-8").read()
    assert re.search(r"\bin tn\b", content) is None
    assert re.search(r'agent"\) in tn', content) is None
    start = content.find("_resolve_herdr_pane_id()")
    assert start != -1
    body = content[start:content.find('cmd="${1:-}"', start)]
    assert r"^w[A-Za-z0-9]+:p[A-Za-z0-9]+$" in body


def test_paste_buffer_branch_never_submits():
    """D-7 (ISSUE-1): paste-buffer arm must not submit or invoke prompt/run."""
    content = open(_LIB_SH, encoding="utf-8").read()
    start = content.find("  paste-buffer)")
    assert start != -1
    end = content.find("\n  delete-buffer)", start)
    assert end != -1
    body = content[start:end]
    for token in ("Enter", "C-m", "pane run", "agent prompt"):
        assert token not in body, token


def test_fullscreen_modal_is_rejected_not_accepted():
    """D-3: fullscreen modal is dismissed with Escape, never Enter."""
    script = f"""
_pane_capture() {{ printf '%s' '{_FULLSCREEN_MODAL}'; }}
KEYS=/tmp/mam-fs-modal-keys.$$
: > "$KEYS"
_sks_herdr() {{
  if [ "${{1:-}}" = "send-keys" ]; then echo "$*" >> "$KEYS"; fi
  return 0
}}
handle_startup_dialogs dummy 1
echo "KEYS_CONTENT=$(tr '\\n' '|' < "$KEYS")"
rm -f "$KEYS"
"""
    res = _run_lib_helpers(script)
    assert res.returncode == 0, res.stderr + res.stdout
    keys = res.stdout.split("KEYS_CONTENT=")[-1]
    assert "Escape" in keys
    assert "Enter" not in keys


def test_prose_yes_try_it_is_not_a_dialog():
    """N-1 / F-1: conversational 'Yes, try it' must not trip _pane_dialog_open."""
    script = r"""
_pane_capture() { printf '%s' 'Sure — if the build fails again, Yes, try it with the --clean flag.'; }
if _pane_dialog_open dummy; then echo "DIALOG_OPEN"; else echo "DIALOG_CLOSED"; fi
"""
    res = _run_lib_helpers(script)
    assert res.returncode == 0, res.stderr + res.stdout
    assert "DIALOG_CLOSED" in res.stdout


def test_mam_dialog_tokens_exclude_yes_try_it():
    """N-1 / F-1: token list must not contain Yes, try it; Escape branch stays."""
    content = open(_LIB_SH, encoding="utf-8").read()
    token_line = next(l for l in content.splitlines() if l.startswith("_MAM_DIALOG_TOKENS="))
    assert "Yes, try it" not in token_line
    assert "grep -q 'Yes, try it'" in content


def test_wait_for_tui_ready_succeeds_on_fullscreen_tip():
    """D-2c: tip + ready banner must not deadlock wait_for_tui_ready."""
    script = f"""
_pane_capture() {{ printf '%s' '{_FULLSCREEN_TIP}'; }}
_sks_herdr() {{
  if [ "${{1:-}}" = "capture-pane" ]; then printf '%s' '{_FULLSCREEN_TIP}'; return 0; fi
  if [ "${{1:-}}" = "send-keys" ]; then echo "KEY:$*" >&2; return 0; fi
  return 0
}}
sleep() {{ :; }}
export MAM_READY_TOKENS='Claude Code|Welcome'
wait_for_tui_ready dummy-sess claude
"""
    res = _run_lib_helpers(script)
    assert res.returncode == 0, res.stderr + res.stdout
    assert "TUI detected ready" in res.stdout
    assert "KEY:" not in res.stderr


def test_bug4_no_duplicate_input_on_rpc_success(tmp_path):
    """Verify Bug 4: when herdr agent prompt succeeds, send_keys_safe returns 0 without calling paste-buffer."""
    test_script = f"""#!/usr/bin/env bash
set -euo pipefail
SKILL_DIR="{os.path.abspath('.agents/skills')}"
source "$SKILL_DIR/lib.sh"

_pane_quiescent() {{ return 0; }}
_pane_dialog_open() {{ return 1; }}

PASTE_CALLED=0
_sks_herdr() {{
  if [ "${{1:-}}" = "agent" ] && [ "${{2:-}}" = "prompt" ]; then
    return 0
  fi
  if [ "${{1:-}}" = "paste-buffer" ]; then
    PASTE_CALLED=1
  fi
  return 0
}}

send_keys_safe "test-sess" "my prompt" "job-1"
if [ "$PASTE_CALLED" = "1" ]; then
  echo "ERROR: paste-buffer was called after agent prompt success (duplicate input)" >&2
  exit 1
fi
echo "SUCCESS"
"""
    res = subprocess.run(["bash", "-c", test_script], capture_output=True, text=True)
    assert res.returncode == 0, f"Expected clean exit 0 without duplicate paste-buffer call, got {res.returncode}. Stderr: {res.stderr}"
    assert "SUCCESS" in res.stdout


def test_bug4_headless_unobservable_fast_path(tmp_path):
    """Verify Bug 4 / R-1 + I-2: in headless mode where capture-pane is empty,
    send_keys_safe bypasses dialogs and succeeds immediately via the RPC fast-path.

    The elapsed-time bound is a contract, not a nicety: removing the
    SKS_EMPTY_GIVEUP early exit leaves every functional assertion green and only
    changes the wall clock (measured 1.22s -> 10.21s), so this is the sole
    assertion that can detect that regression.
    """
    test_script = f"""#!/usr/bin/env bash
set -euo pipefail
SKILL_DIR="{os.path.abspath('.agents/skills')}"
source "$SKILL_DIR/lib.sh"

PROMPT_CALLED=0
PASTE_CALLED=0
_sks_herdr() {{
  if [ "${{1:-}}" = "capture-pane" ]; then
    # Headless / unobservable pane returns empty output
    echo ""
    return 0
  fi
  if [ "${{1:-}}" = "agent" ] && [ "${{2:-}}" = "prompt" ]; then
    PROMPT_CALLED=1
    return 0
  fi
  if [ "${{1:-}}" = "paste-buffer" ]; then
    PASTE_CALLED=1
  fi
  return 0
}}

# Run send_keys_safe without stubbing _pane_quiescent
send_keys_safe "headless-sess" "my prompt" "job-headless"
if [ "$PROMPT_CALLED" != "1" ]; then
  echo "ERROR: agent prompt was not called in headless mode" >&2
  exit 1
fi
if [ "$PASTE_CALLED" = "1" ]; then
  echo "ERROR: paste-buffer was called unexpectedly" >&2
  exit 1
fi
echo "HEADLESS_OK"
"""
    # Remove SKS_* from env so lib.sh defaults apply cleanly
    env = {k: v for k, v in os.environ.items()
           if k not in ("SKS_QUIESCENT_TRIES", "SKS_QUIESCENT_INTERVAL", "SKS_EMPTY_GIVEUP")}

    t0 = time.perf_counter()
    res = subprocess.run(["bash", "-c", test_script], capture_output=True, text=True, env=env)
    elapsed = time.perf_counter() - t0

    assert res.returncode == 0, f"Headless send_keys_safe failed: {res.stderr}"
    assert "HEADLESS_OK" in res.stdout
    assert elapsed < 5.0, (
        f"headless fast-path took {elapsed:.2f}s (limit 5.0s) — the "
        f"SKS_EMPTY_GIVEUP early exit in _pane_quiescent is likely gone; "
        f"the full 10s quiescence window was consumed instead"
    )


def test_bug4_slow_settling_pane_success(tmp_path):
    """Verify N-1 / G-2: a pane that takes 3 seconds of changing output to settle stabilizes cleanly and executes RPC prompt."""
    count_file = str(tmp_path / "capture_count.txt")
    with open(count_file, "w") as f:
        f.write("0")

    prompt_flag = str(tmp_path / "prompt_called.txt")
    paste_flag = str(tmp_path / "paste_called.txt")

    test_script = f"""#!/usr/bin/env bash
set -euo pipefail
SKILL_DIR="{os.path.abspath('.agents/skills')}"
source "$SKILL_DIR/lib.sh"

COUNT_FILE="{count_file}"
PROMPT_FLAG="{prompt_flag}"
PASTE_FLAG="{paste_flag}"

_sks_herdr() {{
  if [ "${{1:-}}" = "capture-pane" ]; then
    local c
    c=$(cat "$COUNT_FILE" 2>/dev/null || echo "0")
    c=$((c + 1))
    echo "$c" > "$COUNT_FILE"
    # Change for first 5 captures (2.5s), then stabilize
    if [ "$c" -le 5 ]; then
      echo "Rendering frame $c..."
    else
      echo "Stable Idle Screen"
    fi
    return 0
  fi
  if [ "${{1:-}}" = "agent" ] && [ "${{2:-}}" = "prompt" ]; then
    touch "$PROMPT_FLAG"
    return 0
  fi
  if [ "${{1:-}}" = "paste-buffer" ]; then
    touch "$PASTE_FLAG"
    return 0
  fi
  return 0
}}

# Run send_keys_safe on slow-settling pane with default 20x0.5 window
send_keys_safe "slow-sess" "my prompt" "job-slow"
if [ ! -f "$PROMPT_FLAG" ]; then
  echo "ERROR: agent prompt was not called on slow-settling pane" >&2
  exit 1
fi
if [ -f "$PASTE_FLAG" ]; then
  echo "ERROR: paste-buffer was called unexpectedly" >&2
  exit 1
fi
echo "SLOW_SETTLE_OK"
"""
    res = subprocess.run(["bash", "-c", test_script], capture_output=True, text=True)
    assert res.returncode == 0, f"Slow settling pane failed: {res.stderr}"
    assert "SLOW_SETTLE_OK" in res.stdout