multi-agent-mux/tests/test_b19_headless_reconcile_fixes.py

import os
import sys
import json
import subprocess
import time
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)
    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)
    assert not d_wide.is_overflow
    assert d_wide.direction == "right"

    # 4. Tall pane (split down)
    payload_tall = {"result": {"panes": [{"pane_id": "p1", "rect": {"width": 80, "height": 60}}]}}
    d_tall = compute_2xk_layout(payload_tall)
    assert not d_tall.is_overflow
    assert d_tall.direction == "down"


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"


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