#!/usr/bin/env bash
# Proves the tmux host is invisible to the program inside it: the exact bytes a terminal
# sends arrive unchanged, and a window resize is delivered. Runs a probe program twice --
# bare on a pty, then through the session host -- and requires identical output.
#
# Ctrl-B is the load-bearing case: Claude Code binds it, and a multiplexer with a prefix
# key would swallow it.
#
# Hosting requires both tmux and the user manager. Remote test builders without either skip
# this workstation-only integration rather than reporting a passthrough regression.
set -uo pipefail

if ! command -v tmux >/dev/null 2>&1 || ! systemctl --user show-environment >/dev/null 2>&1; then
  printf 'skip tmux or user systemd unavailable on this host\n'
  exit 0
fi

if [[ -n "${TMPJAIL_ACTIVE:-}" ]]; then
  printf 'refuse: inside the /tmp jail the user manager'"'"'s tmux server is unreachable.\n' >&2
  printf 'run it outside:\n  systemd-run --user --pipe --wait --collect --same-dir --setenv=TERM=%s -- bash %s\n' \
    "${TERM:-xterm-256color}" "$0" >&2
  exit 1
fi

# tmux refuses to attach a client whose terminfo entry lacks `clear`, so a shell with no TERM
# (a CI job, a systemd unit) would report the tmux host as a passthrough regression.
if ! tput -T "${TERM:-dumb}" clear >/dev/null 2>&1; then
  for _t in xterm-256color xterm vt100; do
    if tput -T "$_t" clear >/dev/null 2>&1; then
      export TERM="$_t"
      break
    fi
  done
fi
if ! tput -T "${TERM:-dumb}" clear >/dev/null 2>&1; then
  printf 'skip no terminfo entry tmux can drive on this host\n'
  exit 0
fi

TESTROOT=$(mktemp -d -t agent-passthrough.XXXXXX)
LEDGER_ID="passthrough-$$-${TESTROOT##*.}"
SOCKET_KEY="$(printf '%s' "$LEDGER_ID" | sha256sum | cut -c1-32)"
SOCK="$TESTROOT/state/sock/$SOCKET_KEY/tmux.sock"
cleanup() {
  [[ -S "$SOCK" ]] && tmux -S "$SOCK" kill-server 2>/dev/null
  rm -rf "$TESTROOT"
}
trap cleanup EXIT

MODULE="$(cd "$(dirname "$0")/.." && pwd)"
mkdir -p "$TESTROOT/bin" "$TESTROOT/lib" "$TESTROOT/state/sessions"
printf '{"schemaVersion":1,"ledgerId":"%s"}\n' "$LEDGER_ID" \
  >"$TESTROOT/state/sessions/$LEDGER_ID.json"
cp "$MODULE/bin/_agent-session-tmux" "$MODULE/bin/_tmpjail-shim.sh" "$MODULE/bin/tmpjail" "$TESTROOT/bin/"
python3 - "$TESTROOT/bin/_agent-session-tmux" "$TESTROOT/bin/seat-authority" <<'PY'
import sys
path, authority = sys.argv[1:]
text = open(path, encoding="utf-8").read()
text = text.replace("/usr/local/bin/overdeck-seat-scope-entry", authority)
open(path, "w", encoding="utf-8").write(text)
PY
ln -s _tmpjail-shim.sh "$TESTROOT/bin/probeagent"
cp "$MODULE/lib/human-session.tmux.conf" "$MODULE/lib/agent-session-reader.mjs" "$TESTROOT/lib/"
printf '#!/usr/bin/env bash\nshift\nexec "$@"\n' >"$TESTROOT/bin/_agent-session-admission"
cat >"$TESTROOT/bin/seat-authority" <<'SH'
#!/usr/bin/env bash
set -uo pipefail
attach=0
socket=""
reap_identity=""
while (( $# )); do
  case "$1" in
    --local-session) shift ;;
    --seat-id) shift 2 ;;
    --socket) socket="$2"; shift 2 ;;
    --cwd) cwd="$2"; shift 2 ;;
    --reap-identity) reap_identity="$2"; shift 2 ;;
    --attach) attach=1; shift ;;
    --) shift; break ;;
    *) exit 64 ;;
  esac
done
[[ -n "$socket" && -n "${cwd:-}" && -n "$reap_identity" && $# -gt 0 ]] || exit 64
mkdir -p "$(dirname "$socket")"
tmux -S "$socket" -f "$(dirname "$0")/../lib/human-session.tmux.conf" new-session -d -s main -c "$cwd" -- "$@" \; \
  set-option -g remain-on-exit on \; set-option -g exit-empty off \; \
  set-option -t main @agent_reap_identity "$reap_identity" || exit 1
if (( attach )); then tmux -S "$socket" attach-session -t main; fi
while :; do
  pane=$(tmux -S "$socket" list-panes -t main -F '#{pane_dead} #{pane_dead_status}' 2>/dev/null) || exit 1
  [[ "$pane" == "1 "* ]] && exit "${pane#1 }"
  sleep 0.05
done
SH
cat >"$TESTROOT/bin/sudo" <<'SH'
#!/usr/bin/env bash
[[ "$1" == -n ]] || exit 64
shift
exec "$@"
SH
chmod +x "$TESTROOT/bin/_agent-session-tmux" "$TESTROOT/bin/_agent-session-admission" \
  "$TESTROOT/bin/tmpjail" "$TESTROOT/bin/seat-authority" "$TESTROOT/bin/sudo"

cat >"$TESTROOT/probe.py" <<'PY'
import os, signal, sys, termios, tty
fd = sys.stdin.fileno()
tty.setraw(fd)
log_path = os.environ.get("PROBE_LOG")
def emit(event):
    if log_path:
        with open(log_path, "a", encoding="utf8") as sink:
            sink.write(event + "\n")
    sys.stdout.write(event + "\r\n")
    sys.stdout.flush()
signal.signal(signal.SIGWINCH, lambda *_: emit("WINCH:%s" % os.get_terminal_size().columns))
sys.stdout.write("READY\r\n"); sys.stdout.flush()
seen = b""
while b"\x11" not in seen:
    chunk = os.read(fd, 1)
    if not chunk:
        break
    seen += chunk
    emit("".join("%02x" % b for b in chunk))
termios.tcsetattr(fd, termios.TCSADRAIN, termios.tcgetattr(fd))
PY

cat >"$TESTROOT/drive.py" <<'PY'
import os, pty, select, signal, sys, time, fcntl, struct
argv = sys.argv[1:]
pid, fd = pty.fork()
if pid == 0:
    os.execvp(argv[0], argv)
fcntl.ioctl(fd, termios_ws := 0x5414, struct.pack("HHHH", 24, 80, 0, 0))
out = b""
deadline = time.time() + 12
def pump(until_ready=False):
    global out
    while time.time() < deadline:
        r, _, _ = select.select([fd], [], [], 0.3)
        if not r:
            if until_ready:
                continue
            return
        try:
            data = os.read(fd, 4096)
        except OSError:
            return
        if not data:
            return
        out += data
        if until_ready and b"READY" in out:
            return
pump(until_ready=True)
# Ctrl-B, Ctrl-C, Ctrl-D, Ctrl-Z, Ctrl-R, Ctrl-Space, Up arrow, then Ctrl-Q to stop.
for payload in (b"\x02", b"\x03", b"\x04", b"\x1a", b"\x12", b"\x00", b"\x1b[A"):
    os.write(fd, payload)
    time.sleep(0.15)
    pump()
fcntl.ioctl(fd, 0x5414, struct.pack("HHHH", 30, 100, 0, 0))
time.sleep(1.2)
pump()
os.write(fd, b"\x11")
time.sleep(0.3)
pump()
try:
    os.kill(pid, signal.SIGKILL)
except ProcessLookupError:
    pass
os.waitpid(pid, 0)
sys.stdout.write(out.decode("utf8", "replace"))
PY

python3 "$TESTROOT/drive.py" env PROBE_LOG="$TESTROOT/bare.events" \
  python3 "$TESTROOT/probe.py" >/dev/null
screen=$(AGENT_SESSIONS_DIR="$TESTROOT/state" AGENT_LEDGER_ID="$LEDGER_ID" \
  AGENT_LEDGER_MUX_SOCKET="$SOCK" AGENT_LEDGER_MUX_TARGET=main AGENT_LEDGER_MUX_ATTACH=1 \
  PATH="$TESTROOT/bin:$PATH" python3 "$TESTROOT/drive.py" "$TESTROOT/bin/_agent-session-tmux" \
    env PROBE_LOG="$TESTROOT/wrapped.events" python3 "$TESTROOT/probe.py")

bare=$(tr '\n' ' ' <"$TESTROOT/bare.events")
wrapped=$(tr '\n' ' ' <"$TESTROOT/wrapped.events")

fail=0
report() { if [[ $1 == 1 ]]; then printf 'ok   %s\n' "$2"; else printf 'FAIL %s\n' "$2"; fail=1; fi; }

[[ $bare == *"02 "* ]] && report 1 "probe harness receives Ctrl-B when unwrapped" ||
  report 0 "probe harness receives Ctrl-B when unwrapped (got: $bare)"
[[ $wrapped == *"02 "* ]] && report 1 "Ctrl-B reaches the program inside the container" ||
  report 0 "Ctrl-B reaches the program inside the container (got: $wrapped)"
[[ $wrapped == *"00 "* ]] && report 1 "Ctrl-Space reaches the program inside the container" ||
  report 0 "Ctrl-Space reaches the program inside the container (got: $wrapped)"
[[ $wrapped == *"1b 5b 41 "* ]] && report 1 "arrow keys reach the program inside the container" ||
  report 0 "arrow keys reach the program inside the container (got: $wrapped)"

# A tmux client renders a screen rather than replaying the pane's raw output stream, so ANSI
# cursor movement can share a physical line with the probe text. Compare the bytes recorded
# by the program inside the pane; that is the passthrough contract being tested.
bare_keys=$(tr ' ' '\n' <<<"$bare" | grep -E '^[0-9a-f]{2,6}$' | tr '\n' ' ')
wrapped_keys=$(tr ' ' '\n' <<<"$wrapped" | grep -E '^[0-9a-f]{2,6}$' | tr '\n' ' ')
[[ $bare_keys == "$wrapped_keys" ]] && report 1 "keystroke byte stream identical wrapped and unwrapped" ||
  report 0 "keystroke byte stream identical wrapped and unwrapped
  bare:    $bare_keys
  wrapped: $wrapped_keys"
[[ $wrapped == *"WINCH:100"* ]] && report 1 "terminal resize reaches the program" ||
  report 0 "terminal resize reaches the program (got: $wrapped)"

# The other direction: what the program writes must reach the outer terminal, not just the
# side-channel log. tmux renders a screen rather than replaying bytes, so this asserts the
# text is on it, not that the two byte streams match.
[[ $screen == *READY* ]] && report 1 "program output reaches the outer terminal" ||
  report 0 "program output reaches the outer terminal"

exit $fail