rune/tests/test_mimo_tty_contract.py

"""MiMo harness process-ownership contract tests (TTY-repair tranche).

These tests exercise the SAME execution shapes the harness tools use
(node child_process via python subprocess + /bin/sh -c), deterministically,
without needing the MiMo TUI:

  A/B. delayed stdout/stderr after timeout cannot reach any inherited tty
  C.   descendant survival: process-group kill leaves no orphan writer
  D.   explicit detachment: fd 0/1/2 never reference a MiMo pts
  E.   repeated timeouts accumulate no survivors
  F.   ordinary command output is captured (inside the tool result)
  G.   shell quoting is preserved verbatim through the tool layer
  H.   wait-intent compilation fires for genuine intent and NO-OPs otherwise
  I.   await-operation implementation unchanged (healthy path preserved)
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
SCRIPTS = REPO / "scripts"
sys.path.insert(0, str(SCRIPTS))

NODE = shutil.which("node") or "/usr/bin/node"
MARKER = "tty-own-test"


def _run_node(script: str, timeout: float = 60) -> tuple[int, str, str]:
    r = subprocess.run(
        [NODE, "--input-type=module", "-e", script], capture_output=True, text=True, timeout=timeout, cwd="/tmp"
    )
    return r.returncode, r.stdout, r.stderr


def _make_leaky_child(tmp: Path, delay: int = 8) -> Path:
    p = tmp / "leaky.sh"
    p.write_text(
        "#!/bin/sh\n"
        f"echo EARLY-STDOUT-{MARKER}\n"
        f"echo EARLY-STDERR-{MARKER} >&2\n"
        f"( sleep {delay}; echo LATE-OUT-{MARKER}; echo LATE-ERR-{MARKER} >&2 ) &\n"
        "exit 0\n"
    )
    p.chmod(0o755)
    return p


# ── TEST A/B: delayed stdout/stderr after timeout ───────────────────────


def test_ab_delayed_output_cannot_reach_tty(tmp_path):
    """Group-kill wrapper: after the tool finalizes, the surviving group is
    signalled; the late writer's fds are capture sockets, never a pts."""
    child = _make_leaky_child(tmp_path)
    delay = 8
    js = f"""
import {{ spawn }} from 'node:child_process';
const child = spawn('/bin/sh', ['-c', '{child}'], {{
  cwd: '/tmp', stdio: ['ignore','pipe','pipe'], detached: true }});
let out=''; child.stdout.on('data', d => out += d);
child.on('close', (code) => {{
  // tool finalizes here (before the late writer fires)
  setTimeout(() => {{
    const probe = spawn('/bin/sh', ['-c',
      `for p in $(pgrep -f 'sleep {delay}' 2>/dev/null); do
         readlink /proc/$p/fd/1 /proc/$p/fd/2 2>/dev/null
       done | sort -u`], {{ stdio: ['ignore','pipe','pipe'] }});
    let fds=''; probe.stdout.on('data', d => fds += d);
    probe.on('close', () => {{
      console.log(JSON.stringify({{ early: out.includes('EARLY-STDOUT-{MARKER}'), fds }}));
      process.exit(0);
    }});
  }}, 3000);
}});
"""
    rc, out, err = _run_node(js)
    assert rc == 0, err
    res = json.loads(out.strip().splitlines()[-1])
    assert res["early"] is True, "early stdout must still be captured"
    for line in res["fds"].splitlines():
        line = line.strip()
        if not line:
            continue
        assert not line.startswith("/dev/pts/"), f"survivor fd points at a pts: {line} — RAW TUI LEAK"


# ── TEST C: descendant survival under group kill ────────────────────────


def test_c_group_kill_leaves_no_orphan_writer(tmp_path):
    """The v2 contract: process-group SIGTERM+SIGKILL after timeout ⇒ no
    member of the invocation's group survives."""
    js = f"""
import {{ spawn }} from 'node:child_process';
const sleep = (ms) => new Promise(r => setTimeout(r, ms));
const child = spawn('/bin/sh', ['-c',
  '( sleep 30 & sleep 30 ) ; # group test'], {{
  cwd: '/tmp', stdio: ['ignore','pipe','pipe'], detached: true }});
child.on('close', async () => {{
  try {{ process.kill(-child.pid, 'SIGTERM'); }} catch {{}}
  await sleep(300);
  try {{ process.kill(-child.pid, 'SIGKILL'); }} catch {{}}
  await sleep(300);
  const probe = spawn('/bin/sh', ['-c',
    `ps -o pgid= -p $(pgrep -f 'sleep 30' | head -1) 2>/dev/null || echo GONE`],
    {{ stdio: ['ignore','pipe','pipe'] }});
  let res=''; probe.stdout.on('data', d => res += d);
  probe.on('close', () => {{
    console.log(JSON.stringify({{ pgid: res.trim() }}));
    process.exit(0);
  }});
}});
setTimeout(() => {{ try {{ process.kill(-child.pid, 'SIGTERM'); }} catch {{}} }}, 500);
"""
    rc, out, err = _run_node(js)
    assert rc == 0, err
    res = json.loads(out.strip().splitlines()[-1])
    # every remaining 'sleep 30' must NOT belong to the child's process group
    assert "GONE" in res["pgid"] or res["pgid"].strip() != "", f"group survivor still running: {res}"


# ── TEST D: explicit detachment ─────────────────────────────────────────


def test_d_detached_process_has_no_pts_fds(tmp_path):
    log = tmp_path / "detached.log"
    js = f"""
import {{ spawn }} from 'node:child_process';
const out = `exec >>'{log}' 2>&1; exec 0</dev/null; echo detached-marker-{MARKER}; sleep 2`;
const quoted = out.replace(/'/g, `'\\\\''`);
const child = spawn('/bin/sh', ['-c', `setsid /bin/sh -c '${{quoted}}'`],
  {{ cwd: '/tmp', stdio: 'ignore', detached: true }});
child.unref();
setTimeout(() => {{
  const probe = spawn('/bin/sh', ['-c',
    `P=$(pgrep -f 'detached-marker-{MARKER}' | head -1); [ -z "$P" ] && echo NONE;
     [ -n "$P" ] && echo "pts_fds=$(ls -la /proc/$P/fd 2>/dev/null | grep -c pts)"`],
    {{ stdio: ['ignore','pipe','pipe'] }});
  let r=''; probe.stdout.on('data', d => r += d);
  probe.on('close', () => {{ console.log(r.trim()); process.exit(0); }});
}}, 500);
"""
    rc, out, err = _run_node(js)
    assert rc == 0, err
    assert "pts_fds=0" in out, f"detached child holds pts fds: {out}"
    time.sleep(2.5)
    assert f"detached-marker-{MARKER}" in log.read_text(), "detached output must land in the explicit log only"


# ── TEST E: repeated timeouts accumulate no survivors ───────────────────


def test_e_repeated_timeouts_no_accumulation(tmp_path):
    js = f"""
import {{ spawn }} from 'node:child_process';
const sleep = (ms) => new Promise(r => setTimeout(r, ms));
const pgids = [];
for (let i = 0; i < 5; i++) {{
  const child = spawn('/bin/sh', ['-c', 'sleep 30 & sleep 30'], {{
    cwd: '/tmp', stdio: ['ignore','pipe','pipe'], detached: true }});
  pgids.push(child.pid);
  await sleep(200);
  try {{ process.kill(-child.pid, 'SIGTERM'); }} catch {{}}
  await sleep(150);
  try {{ process.kill(-child.pid, 'SIGKILL'); }} catch {{}}
  await sleep(150);
}}
// HERMETIC survivor check: probe OUR OWN process groups with
// kill(-pgid, 0) — a live group answers, a fully-reaped group throws
// ESRCH. The previous measurement (global `pgrep -x -c sleep` delta)
// counted unrelated system sleeps spawned mid-loop on the runner
// (mount helpers, cron helpers) as test survivors — CI run 13255
// false-failed with "1 NEW sleep survivor" from exactly that.
let survivors = 0;
for (const pgid of pgids) {{
  try {{ process.kill(-pgid, 0); survivors++; }} catch {{}}
}}
console.log(JSON.stringify({{ n: String(survivors) }}));
"""
    rc, out, err = _run_node(js)
    assert rc == 0, err
    res = json.loads(out.strip().splitlines()[-1])
    assert res["n"] == "0", f"{res['n']} NEW sleep survivors accumulated after 5 timeouts"


# ── TEST F: ordinary command output stays inside the result ─────────────


def test_f_ordinary_command_captured():
    js = """
import { spawn } from 'node:child_process';
const child = spawn('/bin/sh', ['-c', 'echo hello-owned; echo err-owned >&2'],
  { cwd: '/tmp', stdio: ['ignore','pipe','pipe'] });
let o='', e='';
child.stdout.on('data', d => o += d);
child.stderr.on('data', d => e += d);
child.on('close', (code) => {
  console.log(JSON.stringify({ code, o, e }));
  process.exit(0);
});
"""
    rc, out, err = _run_node(js)
    res = json.loads(out.strip().splitlines()[-1])
    assert res["code"] == 0 and "hello-owned" in res["o"] and "err-owned" in res["e"]


# ── TEST G: shell quoting preserved through the tool layer ──────────────


def test_g_quote_preservation():
    cases = [
        ("printf '%s\\n' 'hello world'", "hello world\n"),
        ("printf '%s\\n' \"hello world\"", "hello world\n"),
        ('sh -c \'printf "%s\\\\n" "hello world"\'', "hello world\n"),
        ("printf '%s\\n' \"a 'b' c\"", "a 'b' c\n"),
        ("printf 'one\\ntwo\\n' | grep 'two'", "two\n"),
        ("printf '%s\\n' 'a;b'", "a;b\n"),
    ]
    for cmd, want in cases:
        r = subprocess.run(["/bin/sh", "-c", cmd], capture_output=True, text=True)
        assert r.stdout == want, f"quoting broken for {cmd!r}: got {r.stdout!r}"
    # pgrep with a quoted multi-word pattern must NOT produce a usage error.
    # (It may match its own sh -c wrapper — rc 0 with a match is fine; the
    # invariant is the ABSENCE of the usage error, not the rc value.)
    r = subprocess.run(
        ["/bin/sh", "-c", "pgrep -f 'definitely-no-such-process-xyz'; echo rc=$?"], capture_output=True, text=True
    )
    assert "only one pattern" not in r.stderr, r.stderr


# ── TEST H: wait-intent compilation boundary ────────────────────────────


def test_h_wait_intent_boundary():
    import await_intent as ai  # noqa: E402

    # genuine wait intent routes (may be AMBIGUOUS if many pends exist —
    # still NOT a normal-exec; the strict gate refuses without a concrete op)
    genuine = ai.normalize_bash("sleep 120 then check CI")
    assert genuine is not None and genuine.get("raw_command_kind") == "sleep_poll"
    # arbitrary shell with quotes/words like sleep must NOT compile
    for c in ["pgrep -f 'mimo run'", "echo 'sleep 999'", "printf '%s\\n' 'check status later'", "sleep 2 && echo done"]:
        assert ai.normalize_bash(c) is None, f"false compile: {c!r}"


# ── TEST I: await-operation implementation preserved ────────────────────


def test_i_await_operation_canonical_unchanged():
    import hashlib

    canon = REPO / "config" / "mimo-tools" / "await-operation.ts"
    installed = Path("/home/nick/.mimocode/tools/await-operation.ts")
    h = lambda p: hashlib.sha256(p.read_bytes()).hexdigest()
    assert canon.is_file() and installed.is_file()
    assert h(canon) == h(installed), "await-operation drifted from canonical"
    # healthy-path invariants present in the accepted implementation
    src = canon.read_text()
    assert "ci_event_wait.py" in src  # canonical continuation
    assert "execSync" in src  # long-lived blocking invocation (accepted design)
    assert "operation_id" in src  # explicit-id path