wait_for_text: entry cursor row's content stays suppressed for the whole wait

"""Reproduction: entry-cursor-row content is suppressed for the whole wait.

wait_for_text snapshots the entry cursor row's CONTENT and filters every
later row equal to it, for the entire wait. On a shell prompt the entry
row is the prompt, so any later row that renders byte-identically to it
is unmatchable -- including the prompt returning, which is the single
most common thing a caller waits for after sending a command.

Run against a throwaway socket; the script creates and destroys it.
"""

from __future__ import annotations

import asyncio
import json
import os
import subprocess
import sys

SOCK = "repro-entryrow"


def sh(*args: str) -> str:
    return subprocess.run(
        ["tmux", "-L", SOCK, *args], capture_output=True, text=True, check=False
    ).stdout.rstrip("\n")


async def case(name: str, patterns: list[str] | None, payload_row: str = "$") -> dict:
    from libtmux_mcp.tools.pane_tools.wait import wait_for_text

    sh("kill-server")
    # A pane parked on a prompt-shaped row that never moves on its own.
    subprocess.run(
        ["tmux", "-L", SOCK, "new-session", "-d", "-s", "s", "-x", "80", "-y", "24",
         "sh", "-c", "printf '$ '; sleep 120"],
        check=True,
    )
    pane = sh("list-panes", "-t", "s", "-F", "#{pane_id}")
    tty = sh("display-message", "-p", "-t", pane, "#{pane_tty}")
    await asyncio.sleep(1.0)
    entry_row = sh("capture-pane", "-p", "-t", pane).splitlines()[0]

    async def emit() -> None:
        # Land four rows whose rendered content is byte-identical to the
        # entry row. The leading newline is what puts them on FRESH rows
        # instead of appending to the entry row.
        await asyncio.sleep(1.0)
        fd = os.open(tty, os.O_WRONLY)
        try:
            os.write(fd, ("\n" + "\n".join([payload_row] * 4) + "\n").encode())
        finally:
            os.close(fd)

    task = asyncio.create_task(emit())
    result = await wait_for_text(
        pane_id=pane, patterns=patterns, timeout=5.0, socket_name=SOCK
    )
    await task
    after = sh("capture-pane", "-p", "-t", pane)
    sh("kill-server")

    return {
        "case": name,
        "entry_row_content": repr(entry_row),
        "rows_added": len([ln for ln in after.splitlines() if ln.strip()]) - 1,
        "found": result.found,
        "outcome": result.outcome,
        "elapsed_seconds": result.elapsed_seconds,
        "saw_new_output": result.saw_new_output,
        "matched_at_entry": result.matched_at_entry,
        "tail": result.tail[-2:],
    }


async def main() -> int:
    rows = [
        # BUG: the pattern matches text that IS on screen and IS new.
        await case("patterns=['\\\\$'] matching the entry row's content", [r"\$"]),
        # CONTROL: identical setup, but the arriving rows differ from the
        # entry row -- this MUST match, proving the harness can succeed and
        # that the miss above is about the content equality, nothing else.
        await case("control: rows differ from the entry row", ["READY"], "READY"),
        # CONTROL: patterns=null still sees the rows.
        await case("control: patterns=null", None),
    ]
    print(json.dumps(rows, indent=2))
    bug, ctrl_match, ctrl_null = rows
    print()
    ok = (not bug["found"]) and ctrl_match["found"] and ctrl_null["found"]
    print("REPRODUCED (and both controls behaved)" if ok else "NOT reproduced / control failed")
    return 0


if __name__ == "__main__":
    sys.path.insert(0, "/home/d/work/python/libtmux-mcp-trunk/src")
    sys.exit(asyncio.run(main()))