Port from openai/codex#41436: answer blocking terminal queries in background PTY sessions
Programs run via terminal(background=true, pty=true) can block forever when they probe their terminal — device-status (ESC[5n), window-size (ESC[18t), cursor-position (ESC[6n), or DEC private-mode (ESC[?N$p) queries — because nothing on the PTY master side answers, and the raw query bytes leak into captured output. - tools/pty_query_responder.py: incremental byte scanner that strips the handled queries from PTY output (chunk splits included) and produces bounded replies; everything else passes through untouched. - tools/process_registry.py: wire the responder into _pty_reader_loop (POSIX only — ConPTY answers its own queries); flush partial escape tails at end-of-stream. - tests mirror the codex fixtures plus a live-PTY E2E where a subprocess blocks on ESC[6n until answered.
This commit is contained in:
173
tests/tools/test_pty_query_responder.py
Normal file
173
tests/tools/test_pty_query_responder.py
Normal file
@@ -0,0 +1,173 @@
|
||||
"""Tests for tools.pty_query_responder (port of openai/codex#41436).
|
||||
|
||||
Covers the byte-level scanner (exact queries, chunk splits, DEC private-mode
|
||||
queries, passthrough of unhandled sequences) and a live PTY E2E where a
|
||||
subprocess blocks on a cursor-position report until answered.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.pty_query_responder import PtyQueryResponder
|
||||
|
||||
|
||||
def test_plain_output_passes_through():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"hello world\n")
|
||||
assert out == b"hello world\n"
|
||||
assert resp == b""
|
||||
|
||||
|
||||
def test_device_status_report_answered_and_stripped():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"before\x1b[5nafter")
|
||||
assert out == b"beforeafter"
|
||||
assert resp == b"\x1b[0n"
|
||||
|
||||
|
||||
def test_window_size_query_reports_spawn_dimensions():
|
||||
r = PtyQueryResponder(rows=30, cols=120)
|
||||
out, resp = r.process(b"\x1b[18t")
|
||||
assert out == b""
|
||||
assert resp == b"\x1b[8;30;120t"
|
||||
|
||||
|
||||
def test_cursor_position_report():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"\x1b[6n")
|
||||
assert out == b""
|
||||
assert resp == b"\x1b[1;1R"
|
||||
|
||||
|
||||
def test_dec_private_mode_query_reported_unrecognized():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"\x1b[?1049$p")
|
||||
assert out == b""
|
||||
assert resp == b"\x1b[?1049;0$y"
|
||||
|
||||
|
||||
def test_combined_stream_matches_codex_fixture():
|
||||
# Mirrors the driver-backed test in openai/codex#41436: queries split
|
||||
# across chunks, mixed with a color escape and plain text.
|
||||
r = PtyQueryResponder()
|
||||
out1, resp1 = r.process(b"before\x1b[")
|
||||
out2, resp2 = r.process(b"5n\x1b[18t\x1b[6n\x1b[?1049$p\x1b[31mafter")
|
||||
assert resp1 + resp2 == b"\x1b[0n\x1b[8;24;80t\x1b[1;1R\x1b[?1049;0$y"
|
||||
assert out1 + out2 + r.flush() == b"before\x1b[31mafter"
|
||||
|
||||
|
||||
def test_query_split_across_many_chunks():
|
||||
r = PtyQueryResponder()
|
||||
total_out = b""
|
||||
total_resp = b""
|
||||
for b in (b"\x1b", b"[", b"6", b"n"):
|
||||
out, resp = r.process(b)
|
||||
total_out += out
|
||||
total_resp += resp
|
||||
assert total_out == b""
|
||||
assert total_resp == b"\x1b[1;1R"
|
||||
|
||||
|
||||
def test_unhandled_csi_sequence_passes_through():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"\x1b[31mred\x1b[0m")
|
||||
assert out == b"\x1b[31mred\x1b[0m"
|
||||
assert resp == b""
|
||||
|
||||
|
||||
def test_non_csi_escape_passes_through():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"\x1bMreverse")
|
||||
assert out == b"\x1bMreverse"
|
||||
assert resp == b""
|
||||
|
||||
|
||||
def test_fresh_esc_aborts_partial_sequence():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"\x1b[6\x1b[5n")
|
||||
# The aborted partial "\x1b[6" is flushed through; the complete
|
||||
# device-status query is answered and stripped.
|
||||
assert out == b"\x1b[6"
|
||||
assert resp == b"\x1b[0n"
|
||||
|
||||
|
||||
def test_oversized_mode_query_passes_through():
|
||||
r = PtyQueryResponder()
|
||||
seq = b"\x1b[?12345678901$p" # 11 digits > MAX_MODE_DIGITS
|
||||
out, resp = r.process(seq)
|
||||
assert resp == b""
|
||||
assert out + r.flush() == seq
|
||||
|
||||
|
||||
def test_flush_returns_incomplete_tail():
|
||||
r = PtyQueryResponder()
|
||||
out, resp = r.process(b"text\x1b[1")
|
||||
assert out == b"text"
|
||||
assert resp == b""
|
||||
assert r.flush() == b"\x1b[1"
|
||||
# flush is destructive
|
||||
assert r.flush() == b""
|
||||
|
||||
|
||||
def test_dec_mode_non_digit_passes_through():
|
||||
r = PtyQueryResponder()
|
||||
seq = b"\x1b[?10a9$p"
|
||||
out, resp = r.process(seq)
|
||||
assert resp == b""
|
||||
assert out + r.flush() == seq
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ptyprocess only")
|
||||
def test_live_pty_subprocess_unblocked_by_cursor_report(tmp_path):
|
||||
"""E2E: a PTY subprocess blocking on ESC[6n exits once answered.
|
||||
|
||||
Mirrors direct_terminal_queries_are_answered from openai/codex#41436.
|
||||
"""
|
||||
ptyprocess = pytest.importorskip("ptyprocess")
|
||||
from tools.pty_query_responder import PtyQueryResponder
|
||||
|
||||
script = (
|
||||
"stty -echo -icanon; printf 'alpha\\033[6n'; "
|
||||
"dd bs=1 count=6 2>/dev/null; printf '\\nok'"
|
||||
)
|
||||
proc = ptyprocess.PtyProcess.spawn(
|
||||
["/bin/sh", "-c", script], dimensions=(24, 80)
|
||||
)
|
||||
responder = PtyQueryResponder()
|
||||
output = b""
|
||||
deadline = time.time() + 10
|
||||
try:
|
||||
while proc.isalive() and time.time() < deadline:
|
||||
try:
|
||||
chunk = proc.read(4096)
|
||||
except EOFError:
|
||||
break
|
||||
if not chunk:
|
||||
continue
|
||||
out, replies = responder.process(chunk)
|
||||
output += out
|
||||
if replies:
|
||||
proc.write(replies)
|
||||
# Drain anything left after exit.
|
||||
while True:
|
||||
try:
|
||||
chunk = proc.read(4096)
|
||||
except EOFError:
|
||||
break
|
||||
if not chunk:
|
||||
break
|
||||
out, replies = responder.process(chunk)
|
||||
output += out
|
||||
finally:
|
||||
if proc.isalive():
|
||||
proc.terminate(force=True)
|
||||
pytest.fail(f"subprocess still blocked; output={output!r}")
|
||||
output += responder.flush()
|
||||
assert b"alpha" in output
|
||||
assert b"ok" in output
|
||||
# The query itself must have been stripped from the captured output,
|
||||
# and the echoed reply is what dd consumed (not visible w/ -echo).
|
||||
assert b"\x1b[6n" not in output
|
||||
@@ -1272,12 +1272,30 @@ class ProcessRegistry(ProcessCheckpointMixin):
|
||||
# PTY reads can split a multibyte UTF-8 character across chunks just like pipe reads — hold partial
|
||||
# sequences until the rest arrives. (Ported from openclaw/openclaw#112325.)
|
||||
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
|
||||
# Programs in a PTY can block waiting for replies to device-status / window-size /
|
||||
# cursor-position / DEC private-mode queries. Answer the bounded set and strip the
|
||||
# queries from captured output. POSIX only: Windows ConPTY is a real console host that
|
||||
# answers itself (and pywinpty yields str chunks, not bytes).
|
||||
responder = None
|
||||
if not _IS_WINDOWS:
|
||||
from tools.pty_query_responder import PtyQueryResponder
|
||||
responder = PtyQueryResponder(rows=30, cols=120)
|
||||
try:
|
||||
while pty.isalive():
|
||||
try:
|
||||
chunk = pty.read(4096)
|
||||
if chunk:
|
||||
# ptyprocess returns bytes; pywinpty returns str
|
||||
if responder is not None and isinstance(chunk, bytes):
|
||||
chunk, replies = responder.process(chunk)
|
||||
if replies:
|
||||
try:
|
||||
pty.write(replies)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"PTY query response write failed",
|
||||
exc_info=True,
|
||||
)
|
||||
text = chunk if isinstance(chunk, str) else decoder.decode(chunk)
|
||||
if text:
|
||||
self._ingest_output(session, text)
|
||||
@@ -1285,6 +1303,11 @@ class ProcessRegistry(ProcessCheckpointMixin):
|
||||
break
|
||||
except Exception as e:
|
||||
logger.debug("PTY stdout reader ended: %s", e)
|
||||
if responder is not None:
|
||||
# A query prefix split across the final reads is plain output after all.
|
||||
tail = decoder.decode(responder.flush())
|
||||
if tail:
|
||||
self._ingest_output(session, tail)
|
||||
self._finish_reader(
|
||||
session, decoder, lambda t: self._ingest_output(session, t), "PTY",
|
||||
pty.wait, lambda: pty.exitstatus if hasattr(pty, 'exitstatus') else -1)
|
||||
|
||||
118
tools/pty_query_responder.py
Normal file
118
tools/pty_query_responder.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""Answer a bounded set of blocking terminal queries from PTY subprocesses.
|
||||
|
||||
Programs running inside a background PTY session (``terminal(background=true,
|
||||
pty=true)``) sometimes probe their "terminal" with ANSI queries — device
|
||||
status reports (``ESC[5n``), window-size queries (``ESC[18t``), cursor
|
||||
position reports (``ESC[6n``), or DEC private-mode queries
|
||||
(``ESC[?<mode>$p``). A real terminal emulator answers these on stdin; Hermes'
|
||||
PTY has no emulator on the master side, so the subprocess either blocks
|
||||
forever waiting for a reply or the raw query bytes leak into the captured
|
||||
output as garbage.
|
||||
|
||||
This module ports openai/codex#41436 (``terminal_queries.rs``): a tiny
|
||||
byte-level state machine that scans PTY output for the handled queries,
|
||||
strips them from the output stream (queries split across read chunks
|
||||
included), and produces the bounded responses to write back to the
|
||||
subprocess. Everything else — colors, other escape sequences, partial
|
||||
UTF-8 — passes through untouched.
|
||||
|
||||
Only the POSIX ``ptyprocess`` path uses this. On Windows, ConPTY is a real
|
||||
console host that answers queries itself.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
_ESC = 0x1B
|
||||
|
||||
# DEC private-mode queries carry a numeric mode of bounded length; anything
|
||||
# longer is not a query we answer (and is passed through untouched).
|
||||
_MAX_MODE_DIGITS = 10
|
||||
# Longest handled sequence: ESC [ ? <digits> $ p
|
||||
_MAX_QUERY_BYTES = _MAX_MODE_DIGITS + 5
|
||||
|
||||
|
||||
class PtyQueryResponder:
|
||||
"""Incremental scanner for terminal queries in a PTY output stream.
|
||||
|
||||
Feed raw output chunks through :meth:`process`; it returns the chunk with
|
||||
any handled queries removed, plus the response bytes to write to the
|
||||
subprocess's stdin. Call :meth:`flush` at end-of-stream to recover any
|
||||
trailing partial escape sequence that never completed.
|
||||
"""
|
||||
|
||||
def __init__(self, rows: int = 24, cols: int = 80):
|
||||
# Exact-match queries and their responses. Window size reports the
|
||||
# PTY's actual spawn dimensions; cursor position reports home (1;1) —
|
||||
# we don't emulate a screen, a bounded answer just unblocks the
|
||||
# subprocess (same policy as openai/codex#41436).
|
||||
self._query_responses: tuple[tuple[bytes, bytes], ...] = (
|
||||
# Device status report: terminal operating normally.
|
||||
(b"\x1b[5n", b"\x1b[0n"),
|
||||
# Window-size query: report the PTY's row/col text area.
|
||||
(b"\x1b[18t", b"\x1b[8;%d;%dt" % (rows, cols)),
|
||||
# Cursor-position report: row 1, column 1.
|
||||
(b"\x1b[6n", b"\x1b[1;1R"),
|
||||
)
|
||||
self._pending = bytearray()
|
||||
|
||||
def process(self, data: bytes) -> tuple[bytes, bytes]:
|
||||
"""Scan ``data``; return ``(output_bytes, response_bytes)``."""
|
||||
if not self._pending and _ESC not in data:
|
||||
return data, b""
|
||||
|
||||
output = bytearray()
|
||||
responses = bytearray()
|
||||
pending = self._pending
|
||||
|
||||
for byte in data:
|
||||
if not pending and byte != _ESC:
|
||||
output.append(byte)
|
||||
continue
|
||||
|
||||
if byte == _ESC:
|
||||
# A fresh ESC aborts any partial sequence — flush it through.
|
||||
output += pending
|
||||
pending.clear()
|
||||
pending.append(byte)
|
||||
|
||||
if (
|
||||
len(pending) == 1
|
||||
or bytes(pending) == b"\x1b["
|
||||
or (
|
||||
pending[1] == ord("[")
|
||||
and not (0x40 <= byte <= 0x7E)
|
||||
and len(pending) < _MAX_QUERY_BYTES
|
||||
)
|
||||
):
|
||||
# Still accumulating a possible query.
|
||||
continue
|
||||
|
||||
seq = bytes(pending)
|
||||
matched = False
|
||||
for query, response in self._query_responses:
|
||||
if seq == query:
|
||||
responses += response
|
||||
matched = True
|
||||
break
|
||||
if not matched:
|
||||
mode = seq[3:-2]
|
||||
if (
|
||||
seq.startswith(b"\x1b[?")
|
||||
and seq.endswith(b"$p")
|
||||
and 0 < len(mode) <= _MAX_MODE_DIGITS
|
||||
and mode.isdigit()
|
||||
):
|
||||
# DEC private-mode query: report mode as unrecognized.
|
||||
responses += b"\x1b[?" + mode + b";0$y"
|
||||
else:
|
||||
# Not a handled query — pass the sequence through.
|
||||
output += pending
|
||||
pending.clear()
|
||||
|
||||
return bytes(output), bytes(responses)
|
||||
|
||||
def flush(self) -> bytes:
|
||||
"""Return any incomplete trailing sequence held back by the scanner."""
|
||||
tail = bytes(self._pending)
|
||||
self._pending.clear()
|
||||
return tail
|
||||
Reference in New Issue
Block a user