tegwick
4d915a2617
Some checks failed
tamq-ci / test (push) Failing after 36s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02f34-ead4-7a60-85cd-d7f08e22fa0e
55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import pty
|
|
import select
|
|
import signal
|
|
import sys
|
|
from collections.abc import Callable, Sequence
|
|
|
|
from .broker import InputBroker
|
|
|
|
|
|
class PtyTap:
|
|
"""Full-duplex PTY proxy. Input is observed, never rewritten."""
|
|
|
|
def __init__(self, command: Sequence[str], broker: InputBroker, *, on_line: Callable[[str], None] | None = None):
|
|
self.command = list(command)
|
|
self.broker = broker
|
|
self.on_line = on_line
|
|
|
|
def run(self) -> int:
|
|
pid, master = pty.fork()
|
|
if pid == 0:
|
|
os.execvp(self.command[0], self.command)
|
|
input_buffer = bytearray()
|
|
try:
|
|
while True:
|
|
readable, _, _ = select.select([master, sys.stdin], [], [])
|
|
if master in readable:
|
|
try:
|
|
data = os.read(master, 65536)
|
|
except OSError:
|
|
break
|
|
if not data:
|
|
break
|
|
os.write(sys.stdout.fileno(), data)
|
|
if sys.stdin in readable:
|
|
data = os.read(sys.stdin.fileno(), 65536)
|
|
if not data:
|
|
break
|
|
os.write(master, data)
|
|
input_buffer.extend(data)
|
|
while b"\n" in input_buffer:
|
|
raw, _, rest = input_buffer.partition(b"\n")
|
|
input_buffer = bytearray(rest)
|
|
line = raw.decode("utf-8", errors="replace")
|
|
routed = self.broker.inspect_line(line)
|
|
if routed and self.on_line:
|
|
self.on_line(line)
|
|
finally:
|
|
try:
|
|
_, status = os.waitpid(pid, 0)
|
|
return os.waitstatus_to_exitcode(status)
|
|
except ChildProcessError:
|
|
return 0
|