diff --git a/tests/tools/test_pty_query_responder.py b/tests/tools/test_pty_query_responder.py
new file mode 100644
index 0000000000..3c237188a9
--- /dev/null
+++ b/tests/tools/test_pty_query_responder.py
@@ -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
diff --git a/tools/process_registry.py b/tools/process_registry.py
index c6baf8b12e..224e2df371 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -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)
diff --git a/tools/pty_query_responder.py b/tools/pty_query_responder.py
new file mode 100644
index 0000000000..28cccc4ec6
--- /dev/null
+++ b/tools/pty_query_responder.py
@@ -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