Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
22 kB · 502 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503from __future__ import annotations
import osimport subprocessimport sysimport tempfileimport termiosimport unittestfrom pathlib import Pathfrom types import SimpleNamespacefrom unittest.mock import Mock, patch
import pexpect
from terminal import MAX_PENDING_INPUT, PROBE_INTERVAL, Display, Terminalfrom ui import input_ready
class Clock: def __init__(self): self.now = 4.0
def __call__(self): return self.now
class Editor: """A PTY fixture with independently delayed terminal mode, input, and rendering."""
def __init__( self, clock, *, label="", raw_after=0, accept_after=0, process_after=0, render_delay=0, cleanup_delay=0, placeholder="", partial_cleanup=False, fragmented=False, exit_after=None, echo=False, ): self.clock = clock self.start = clock.now self.raw_at = self.start + raw_after self.accept_at = self.start + accept_after self.process_at = self.start + process_after self.render_delay = render_delay self.cleanup_delay = cleanup_delay self.placeholder = placeholder self.partial_cleanup = partial_cleanup self.fragmented = fragmented self.exit_at = None if exit_after is None else self.start + exit_after self.echo = echo self.child_fd = 123 self.editor = "" self.sent = [] self.events = [] self.schedule(self.start, label + "\r\n")
def schedule(self, when, event): self.events.append((when, event)) self.events.sort(key=lambda item: item[0])
def attrs(self, fd): assert fd == self.child_fd canonical = termios.ICANON if self.clock.now < self.raw_at else 0 return [0, 0, 0, canonical | (termios.ECHO if self.echo else 0)]
def send(self, text): self.sent.append((self.clock.now, text)) assert "\r" not in text and "\n" not in text, "probe must never submit" if self.clock.now < self.accept_at: return len(text) self.schedule(max(self.clock.now, self.process_at), lambda: self.input(text)) return len(text)
def input(self, text): if text.startswith("\x7f"): count = 1 if self.partial_cleanup else len(text) self.editor = self.editor[: max(0, len(self.editor) - count)] delay = self.cleanup_delay else: self.editor += text delay = self.render_delay # Full redraw, hidden hardware cursor, and an optional empty-editor placeholder. visible = self.editor or self.placeholder frame = f"\x1b[?25l\x1b[3;1H\x1b[2K> {visible}\x1b[3;{3 + len(self.editor)}H" if self.fragmented: boundary = frame.index("> ") + 2 self.schedule(self.clock.now + delay, "\x1b[?2026h" + frame[:boundary]) self.schedule(self.clock.now + delay + 0.01, frame[boundary:] + "\x1b[?2026l") else: self.schedule(self.clock.now + delay, frame)
def read_nonblocking(self, size, timeout): self.clock.now += 0.005 if self.exit_at is not None and self.clock.now >= self.exit_at: raise pexpect.EOF("fixture exited") while self.events and self.events[0][0] <= self.clock.now: _, event = self.events.pop(0) if callable(event): event() else: return event raise pexpect.TIMEOUT("no output")
class ReadinessTests(unittest.TestCase): def launch(self, **options): clock = Clock() child = Editor(clock, **options) terminal = Terminal.__new__(Terminal) terminal.started = clock.now - 0.2 # Include spawn overhead, not just ready() time. terminal.child = child terminal._pending_input = b"" terminal.display = Display(terminal._send) terminal.raw = [] terminal.bytes = 0 self.addCleanup(patch.stopall) patch("terminal.time.perf_counter", clock).start() patch("terminal.termios.tcgetattr", child.attrs).start() patch("terminal.os.write", side_effect=lambda _fd, data: child.send(data.decode())).start() # Different deterministic generations let the tests detect stale/buffered probes. patch("terminal.secrets.token_hex", side_effect=(f"{i:08x}" for i in range(1000))).start() return terminal, child, clock
def test_until_pumps_until_the_display_matches(self): terminal, editor, clock = self.launch() editor.schedule(clock.now + 0.03, "target") terminal.until(lambda display: "target" in display.text(), 0.1) self.assertIn("target", terminal.display.text())
def test_until_times_out_when_the_display_never_matches(self): terminal, _, clock = self.launch() with self.assertRaisesRegex(TimeoutError, "terminal display"): terminal.until(lambda display: "missing" in display.text(), 0.1) self.assertLess(clock.now - 4.0, 0.11)
def test_until_output_ignores_matching_text_already_on_screen(self): terminal, editor, clock = self.launch() terminal.display.feed("agents view") editor.schedule(clock.now + 0.03, "fresh agents view") terminal.until_output(lambda output: "fresh agents view" in output, 0.1) self.assertIn("fresh agents view", terminal.display.text())
def test_transcript_readiness_uses_real_terminal_and_excludes_cleanup(self): terminal, editor, clock = self.launch(label="target-tail", accept_after=0.1, cleanup_delay=0.2) ready_at = input_ready(terminal, "target-tail", timeout=1) self.assertLess(ready_at - editor.start, 0.15) self.assertGreater(clock.now - ready_at, 0.19) self.assertEqual(editor.editor, "")
def test_transcript_tail_is_required_before_editor_probe(self): terminal, editor, _ = self.launch(label="other-tail") with self.assertRaisesRegex(TimeoutError, "expected terminal state"): input_ready(terminal, "target-tail", timeout=0.1) self.assertEqual(editor.sent, [])
def test_until_does_not_accept_a_partial_synchronized_frame(self): terminal, editor, clock = self.launch(label="\x1b[?2026htarget-tail") editor.schedule(clock.now + 0.1, "\x1b[?2026l") terminal.until(lambda display: "target-tail" in display.text(), 1) self.assertGreaterEqual(clock.now - editor.start, 0.1)
def test_changed_absent_and_stale_labels_do_not_gate_input(self): for label in ("agents/resume", "manage", "sessions / continue", "", "benchready"): with self.subTest(label=label): terminal, editor, _ = self.launch(label=label) elapsed = terminal.ready() self.assertGreaterEqual(elapsed, 0.2) self.assertLess(elapsed, 0.25) self.assertEqual(editor.editor, "") self.assertEqual([data for _, data in editor.sent], ["b00000000r", "\x7f" * 10]) patch.stopall()
def test_waits_for_raw_mode_even_without_new_output(self): terminal, editor, _ = self.launch(label="agents/resume", raw_after=0.4) elapsed = terminal.ready() self.assertGreaterEqual(editor.sent[0][0], editor.raw_at) self.assertGreaterEqual(elapsed, 0.6) self.assertEqual(editor.editor, "")
def test_retries_input_dropped_after_raw_mode(self): terminal, editor, _ = self.launch(label="agents/resume", accept_after=0.35) elapsed = terminal.ready() self.assertGreater(elapsed, 0.55) self.assertGreater(len(editor.sent), 4) self.assertEqual(editor.editor, "")
def test_delayed_input_detection_has_a_short_retry_bound(self): terminal, editor, _ = self.launch(accept_after=0.35) elapsed = terminal.ready() visible_at = terminal.started + elapsed self.assertLess(visible_at - editor.accept_at, PROBE_INTERVAL + 0.015)
def test_buffered_retries_cannot_outlive_cleanup(self): terminal, editor, _ = self.launch(process_after=0.35) terminal.ready() self.assertGreater(len(editor.sent), 4) self.assertEqual(editor.editor, "") self.assertFalse(editor.events) self.assertEqual((terminal.display.screen.cursor.y, terminal.display.screen.cursor.x), (2, 2))
def test_timing_stops_at_first_probe_not_latest_probe_or_cleanup(self): terminal, editor, clock = self.launch(process_after=0.35, cleanup_delay=0.2) elapsed = terminal.ready() self.assertGreater(elapsed, 0.55) self.assertLess(elapsed, 0.60) self.assertGreater(clock.now - terminal.started - elapsed, 0.19) self.assertEqual(editor.editor, "")
def test_cleanup_allows_placeholder_with_same_leading_character(self): terminal, editor, _ = self.launch(placeholder="begin typing here") terminal.ready() self.assertEqual(editor.editor, "") self.assertIn("begin typing here", terminal.display.text())
def test_partial_cleanup_is_a_failure_even_when_full_marker_disappears(self): terminal, editor, clock = self.launch(partial_cleanup=True) with self.assertRaisesRegex(TimeoutError, "editor probe cleanup"): terminal.ready(seconds=0.3) self.assertEqual(editor.editor, "b00000000") self.assertNotIn("b00000000r", terminal.display.text()) self.assertLess(clock.now - editor.start, 0.31)
def test_fragmented_redraw_cannot_report_partial_cleanup_as_empty(self): terminal, editor, _ = self.launch(partial_cleanup=True, fragmented=True) with self.assertRaisesRegex(TimeoutError, "editor probe cleanup"): terminal.ready(seconds=0.3) self.assertEqual(editor.editor, "b00000000")
def test_echo_mode_never_counts_as_input_readiness(self): terminal, editor, _ = self.launch(label="agents/resume", echo=True) with self.assertRaisesRegex(TimeoutError, "noncanonical, no-echo terminal input"): terminal.ready(seconds=0.2) self.assertEqual(editor.sent, [])
def test_failed_startup_preserves_eof_failure(self): terminal, editor, _ = self.launch(exit_after=0.1, accept_after=1) with self.assertRaisesRegex(RuntimeError, "exited before the measurement completed"): terminal.ready() self.assertEqual(editor.sent[-1][1], "\x7f" * 10)
def test_never_ready_has_one_deadline_and_clears_last_probe(self): terminal, editor, clock = self.launch(raw_after=0.2, accept_after=5) with self.assertRaisesRegex(TimeoutError, "editor input rendering"): terminal.ready(seconds=0.4) self.assertLess(clock.now - editor.start, 0.41) self.assertEqual(editor.sent[-1][1], "\x7f" * 10)
def test_cleanup_uses_remaining_deadline_not_an_extra_timeout(self): terminal, editor, clock = self.launch(accept_after=0.25, cleanup_delay=0.3) with self.assertRaisesRegex(TimeoutError, "editor probe cleanup"): terminal.ready(seconds=0.4) self.assertLess(clock.now - editor.start, 0.41)
def test_saturated_writes_bound_retries_and_final_cleanup(self): terminal, editor, clock = self.launch() with patch("terminal.os.write", side_effect=BlockingIOError) as write: with self.assertRaisesRegex(TimeoutError, "editor input rendering"): terminal.ready(seconds=0.2) self.assertLess(clock.now - editor.start, 0.21) self.assertEqual(editor.sent, []) self.assertEqual(terminal._pending_input, b"\x7f" * 10) # One queued probe is retried, not an ever-growing list of probe generations. self.assertTrue(all(call.args[1] == b"b00000000r" for call in write.call_args_list[:-1])) self.assertEqual(write.call_args_list[-1].args[1], b"\x7f" * 10) self.assertLess(write.call_count, 50)
def test_query_reply_backpressure_cannot_block_pump_or_readiness(self): terminal, editor, clock = self.launch(label="\x1b[6n") with patch("terminal.os.write", side_effect=BlockingIOError) as write: with self.assertRaises(TimeoutError): terminal.ready(seconds=0.2) self.assertLess(clock.now - editor.start, 0.21) self.assertEqual(editor.sent, []) self.assertEqual(terminal._pending_input, b"\x1b[1;1R") self.assertTrue(all(call.args[1] == b"\x1b[1;1R" for call in write.call_args_list))
def test_pending_cleanup_cannot_pass_on_an_empty_redraw(self): terminal, editor, clock = self.launch()
def write(_fd, data): if data.startswith(b"\x7f"): # An unrelated redraw looks empty, but none of the erase reached stdin. editor.schedule(clock.now, "\x1b[3;1H\x1b[2K> ") raise BlockingIOError return editor.send(data.decode())
with patch("terminal.os.write", side_effect=write): with self.assertRaisesRegex(TimeoutError, "editor probe cleanup"): terminal.ready(seconds=0.2) self.assertEqual(editor.editor, "b00000000r") self.assertEqual(terminal._pending_input, b"\x7f" * 10)
def test_failed_cleanup_write_does_not_mask_original_failure(self): terminal, editor, _ = self.launch(exit_after=0.015, accept_after=1)
def write(_fd, data): if data.startswith(b"\x7f"): raise OSError("PTY closed") return editor.send(data.decode())
with patch("terminal.os.write", side_effect=write): with self.assertRaisesRegex(RuntimeError, "exited before the measurement completed"): terminal.ready()
class InputQueueTests(unittest.TestCase): def setUp(self): self.terminal = Terminal.__new__(Terminal) self.terminal.child = SimpleNamespace(child_fd=123) self.terminal._pending_input = b""
def test_short_writes_eagain_and_eintr_preserve_ordered_suffixes(self): with patch("terminal.os.write", side_effect=[3, BlockingIOError, InterruptedError, 2, 3]) as write: self.terminal._send("abcdef") self.assertEqual(self.terminal._pending_input, b"def") self.terminal._send("gh") self.assertEqual(self.terminal._pending_input, b"defgh") self.terminal._flush_input() self.assertEqual(self.terminal._pending_input, b"defgh") self.terminal._flush_input() self.assertEqual(self.terminal._pending_input, b"fgh") self.terminal._flush_input() self.assertEqual(self.terminal._pending_input, b"") self.terminal._flush_input() self.assertEqual( [call.args[1] for call in write.call_args_list], [b"abcdef", b"defgh", b"defgh", b"defgh", b"fgh"], )
def test_zero_progress_does_not_spin_or_drop_bytes(self): with patch("terminal.os.write", return_value=0) as write: self.terminal._send("probe") self.assertEqual(write.call_count, 1) self.assertEqual(self.terminal._pending_input, b"probe")
def test_query_flood_cannot_grow_the_pending_queue_without_bound(self): display = Display(self.terminal._send) with patch("terminal.os.write", side_effect=BlockingIOError): self.terminal._send("x" * MAX_PENDING_INPUT) with self.assertRaisesRegex(RuntimeError, "Terminal input queue exceeds 64 KiB"): display.feed("\x1b[6n") self.assertEqual(self.terminal._pending_input, b"x" * MAX_PENDING_INPUT)
def test_pump_flushes_writes_even_when_child_has_no_output(self): self.terminal.child.read_nonblocking = Mock(side_effect=pexpect.TIMEOUT("no output")) with patch("terminal.os.write", side_effect=[BlockingIOError, 3]): self.terminal._send("abc") self.assertFalse(self.terminal.pump()) self.assertEqual(self.terminal._pending_input, b"")
PTY_FIXTURE = r"""import os, select, sys, termios, time, ttymode, raw_delay, input_delay = sys.argv[1], float(sys.argv[2]), float(sys.argv[3])attrs = termios.tcgetattr(0)attrs[3] |= termios.ECHO | termios.ICANONtermios.tcsetattr(0, termios.TCSANOW, attrs)os.write(1, b'changed startup header\r\n')if mode == 'exit': os.write(1, b'fixture startup failed\r\n') raise SystemExit(3)# If the harness sends during canonical startup, record it instead of hiding the bug.if select.select([0], [], [], raw_delay)[0]: raise RuntimeError('input sent before raw mode')if mode == 'echo': time.sleep(5) raise SystemExit(0)tty.setraw(0)accept_at = time.monotonic() + input_delaytext = ''while True: byte = os.read(0, 1) if byte in (b'\r', b'\n'): raise RuntimeError('benchmark submitted a prompt') if byte == b'\x03': break if time.monotonic() < accept_at: continue if byte == b'\x7f': text = text[:-1] else: text += byte.decode() # Batched TUI-style full redraw with hidden cursor and arbitrary placeholder. visible = text or 'begin typing here' os.write(1, f'\x1b[?25l\x1b[3;1H\x1b[2K> {visible}\x1b[3;{3 + len(text)}H'.encode())"""
SATURATED_PTY = r"""import os, pathlib, signal, sys, timefrom terminal import Terminalroot = pathlib.Path(sys.argv[1])def guard(_signum, _frame): raise RuntimeError('saturated PTY exceeded outer guard')signal.signal(signal.SIGALRM, guard)signal.setitimer(signal.ITIMER_REAL, 2, 2)terminal = Terminal( [sys.executable, '-c', "import os,time,tty; tty.setraw(0); os.write(1,b'raw-ready'); time.sleep(60)"], root, os.environ.copy(), root / 'saturated',)try: deadline = time.perf_counter() + 1 while 'raw-ready' not in terminal.display.text(): assert time.perf_counter() < deadline, 'fixture did not enter raw mode' terminal.pump() fd = terminal.child.child_fd was_blocking = os.get_blocking(fd) os.set_blocking(fd, False) try: while True: os.write(fd, b'x' * 4096) except BlockingIOError: pass finally: os.set_blocking(fd, was_blocking) started = time.perf_counter() try: terminal.ready(seconds=0.15) raise AssertionError('unresponsive editor was marked ready') except TimeoutError as error: assert 'editor input rendering' in str(error), str(error) elapsed = time.perf_counter() - started assert elapsed < 0.5, elapsed started = time.perf_counter() terminal.close() assert time.perf_counter() - started < 1.5, 'close blocked on saturated PTY' assert not terminal.child.isalive() print(f'saturated ready timed out in {elapsed:.3f}s; close stopped child')finally: signal.setitimer(signal.ITIMER_REAL, 0) terminal.child.close(force=True)"""
class RealPTYReadinessTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name)
def launch(self, mode="editor", raw_delay=0.15, input_delay=0.2): root = self.root script = root / "editor.py" script.write_text(PTY_FIXTURE) terminal = Terminal( [sys.executable, str(script), mode, str(raw_delay), str(input_delay)], root, os.environ.copy(), root / "transcript", ) self.addCleanup(terminal.close) return terminal
def test_saturated_real_pty_respects_ready_and_close_bounds(self): result = subprocess.run( [sys.executable, "-c", SATURATED_PTY, str(self.root)], cwd=Path(__file__).resolve().parents[1], capture_output=True, text=True, timeout=6, ) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertIn("close stopped child", result.stdout)
def test_real_pty_transcript_and_editor_readiness(self): terminal = self.launch() ready_at = input_ready(terminal, "changed startup header", timeout=3) self.assertGreater(ready_at - terminal.started, 0.35) self.assertIn("begin typing here", terminal.display.text()) self.assertNotIn("RuntimeError", "".join(terminal.raw))
def test_real_pty_delayed_raw_mode_and_input_handler(self): terminal = self.launch() elapsed = terminal.ready(seconds=3) self.assertGreater(elapsed, 0.35) self.assertIn("begin typing here", terminal.display.text()) self.assertNotIn("RuntimeError", "".join(terminal.raw))
def test_real_failed_startup_keeps_raw_and_screen_diagnostics(self): terminal = self.launch(mode="exit") with self.assertRaisesRegex(RuntimeError, "exited before the measurement completed"): terminal.ready(seconds=1) terminal.close() self.assertIn("fixture startup failed", (self.root / "transcript.raw").read_text()) self.assertIn("fixture startup failed", (self.root / "transcript.txt").read_text())
def test_real_canonical_echo_does_not_accept_a_probe(self): terminal = self.launch(mode="echo", raw_delay=0.01, input_delay=0) with self.assertRaisesRegex(TimeoutError, "noncanonical, no-echo terminal input"): terminal.ready(seconds=0.25) self.assertIn("changed startup header", terminal.display.text())
if __name__ == "__main__": unittest.main()