#!/usr/bin/env python3 """Exercise editable global, column and pane tags through real host input. python3 -B test/column_tags.py zig-out/column-review/bin/pardes-gui \\ /tmp/pardes-column-tags --project /path/to/agave 9P observes state and prepares owned scratch panes. SGR mouse events and keyboard bytes enter through the session's real PTY input parser. SDL mode captures GPU PNGs; --tty records the core grid and styles. This is not an independent terminal-emulator screenshot or an SDL pixel-coordinate test. """ import argparse import codecs import contextlib import json import os from pathlib import Path import re from fs import session from ninep import Client from ui_review import capture, command, panes, require, wait_for class Input: def __init__(self, fd): self.fd = fd def send(self, data): os.write(self.fd, data.encode() if isinstance(data, str) else data) def mouse(self, button, col, row, release=False, drag=False): code = button + (32 if drag else 0) self.send(f'\x1b[<{code};{col + 1};{row + 1}{"m" if release else "M"}') def click(self, col, row, button=0): self.mouse(button, col, row) self.mouse(button, col, row, release=True) def drag(self, first, last, row): self.mouse(0, first, row) self.mouse(0, last, row, drag=True) self.mouse(0, last, row, release=True) def escape(self): # SDL's test feed reserves a partial ESC]777;finger prefix. CSI-u # identifies Escape unambiguously without waiting for another byte. self.send(b'\x1b[27u') def row_text(client, row): frame = client.screen() return ''.join(cell[0] for cell in frame['cells'][row * frame['cols']:(row + 1) * frame['cols']]).rstrip() def locate(client, word, row=None, minimum_col=0): """Find an ASCII needle using cell positions, even after wide glyphs.""" frame = client.screen() rows = range(frame['rows']) if row is None else [row] for y in rows: cells = frame['cells'][y * frame['cols']:(y + 1) * frame['cols']] for x in range(minimum_col, frame['cols']): if ''.join(cell[0] for cell in cells[x:x + len(word)]) == word: return x, y return None def await_text(client, row, text): require(wait_for(lambda: text in row_text(client, row)), f'row {row} never showed {text!r}') def tag(client, serial): return client.read(f'/pane/{serial}/tag').decode().rstrip('\n') def new_serial(client, previous): return next((serial for serial in panes(client) if serial not in previous), None) def exec_tag_without_cleanup(client, serial, text): """Restore can replace the whole core, so never touch the old pane after it.""" client.write(f'/pane/{serial}/tag', (' ' + text).encode()) whole = client.read(f'/pane/{serial}/tag') start = whole.rfind(text.encode()) client.write(f'/pane/{serial}/event', f'Mx{start} {start + len(text.encode())}\n'.encode()) def reconnect(address): try: return Client(address) except (OSError, EOFError): return None def run(args): root = args.output.resolve() root.mkdir(parents=True, mode=0o700, exist_ok=True) root.chmod(0o700) config = root / 'config/pardes' config.mkdir(parents=True, exist_ok=True) (config / 'init').write_text('Shell /bin/sh\nTheme orchard\n') source = args.project.resolve() / 'gossip/src/crds_value.rs' original = source.read_bytes() name = 'column-tags-' + str(os.getpid()) launch = ['-c', 'stty cols 150 rows 48; exec "$@"', 'column-tags', str(args.binary.resolve()), '--9p=' + name, str(source)] env = {} if args.tty else { 'PARDES_TEST': '1', 'PARDES_TEST_CAPTURE_DIR': str(root), 'PARDES_TEST_COLS': '150', 'PARDES_TEST_ROWS': '48', 'SDL_VIDEODRIVER': 'x11', } handles = {} checks = [] with session('/bin/sh', root, name, socket_name=name, tty=True, launch=launch, inherited=env, terminal=handles) as (client, address), contextlib.ExitStack() as connections: print('socket', address, flush=True) keys = Input(handles['input_fd']) await_text(client, 1, 'New Tty Find Grep Joincol') require(locate(client, 'New Tty', row=1) == (0, 1), 'left column tag does not start flush-left') require(tag(client, 1).startswith(str(source) + ' Save Tty Del'), 'generated gap still separates the filename from pane commands') checks.append('pane commands follow the filename with one space') # Global header: actual left drag replaces Joincol, End appends Unicode, # and backspace removes a wide character as one grapheme. keys.drag(7, 14, 0) keys.send('Help') require(wait_for(lambda: row_text(client, 0).startswith('Newcol Help ')), 'global drag replacement failed') keys.send(b'\x1b[F') keys.send(' Ω界') await_text(client, 0, ' Ω界') keys.send(b'\x7f') require(wait_for(lambda: row_text(client, 0).endswith(' Ω')), 'wide-character backspace split a grapheme') keys.send('界') await_text(client, 0, ' Ω界') global_text = row_text(client, 0) keys.escape() checks.append('global drag replacement, typing, Unicode and grapheme backspace') before = panes(client) keys.click(*require(locate(client, 'Newcol', row=0), 'Newcol missing'), button=1) # Newcol makes an empty column and gives its tag the keyboard; the # pane made next fills it. require(wait_for(lambda: client.read('/focus').strip() == b''), 'Newcol did not focus its empty column') right = int(client.read('/pane/new')) right_name = root / 'right.rs' client.write(f'/pane/{right}/name', f'{right_name}\n'.encode()) right_body = b'// Column-local scratch\nfn main() {\n let ready = true;\n}\n' client.write(f'/pane/{right}/body', right_body) client.write(f'/pane/{right}/dirty', b'0\n') require(wait_for(lambda: locate(client, 'New Tty', row=1, minimum_col=75)), 'right column tag missing') require(locate(client, 'New Tty', row=1, minimum_col=75) == (75, 1), 'right column tag retains an inset') checks.append('column tags start at the first cell of each column') # Column header uses half-open drag ranges; replacement must not leak # into its neighbour, the global header, or the active pane's body. col_start, _ = locate(client, 'New Tty', row=1, minimum_col=75) keys.click(col_start, 1) keys.send(b'\x1b[F') keys.send(' Scope λ界') await_text(client, 1, 'Scope λ界') scope_x, _ = require(locate(client, 'Scope', row=1, minimum_col=75), 'column suffix missing') keys.drag(scope_x, scope_x + len('Scope'), 1) keys.send('Work') await_text(client, 1, 'Work λ界') require('Scope' not in row_text(client, 1), 'column replacement appended to selection') keys.escape() require(row_text(client, 0) == global_text, 'column edit changed global tag') checks.append('column drag replacement and Unicode are scoped to that column') # Click the left source body, then execute New on the inactive RIGHT # column. A distinct marker must render on the right half afterwards. keys.click(12, 4) before = panes(client) keys.click(col_start, 1, button=1) extra = require(wait_for(lambda: new_serial(client, before)), 'right-column New did not create a pane') client.write(f'/pane/{extra}/body', b'RIGHT-COLUMN-NEW\n') require(wait_for(lambda: locate(client, 'RIGHT-COLUMN-NEW', minimum_col=75)), 'right-column command targeted another column') client.write(f'/pane/{extra}/exec', b'Del\n') checks.append('middle-click New targets the inactive right column') # An owned short-path file makes its complete tag visible. Pane-tag # selections retain the existing inclusive endpoint convention. save_x, pane_y = require(wait_for(lambda: locate(client, 'Save Tty', minimum_col=75)), 'right pane actions missing') keys.click(save_x, pane_y) keys.send(b'\x1b[F') keys.send(' TailMark') require(wait_for(lambda: tag(client, right).endswith(' TailMark')), 'pane tail typing failed') mark_x, mark_y = require(locate(client, 'TailMark', minimum_col=75), 'pane tail marker not visible') keys.drag(mark_x, mark_x + len('TailMark') - 1, mark_y) keys.send('Local') require(wait_for(lambda: tag(client, right).endswith(' Local')), 'pane selection replacement failed') keys.escape() checks.append('pane command tail supports typing and selection replacement') filename_x, filename_y = require(locate(client, 'right.rs', minimum_col=75), 'short filename not visible') keys.click(filename_x, filename_y) keys.send('X') require(wait_for(lambda: 'Xright.rs' in tag(client, right)), 'filename draft did not become editable') keys.escape() require(wait_for(lambda: tag(client, right).startswith(str(right_name) + ' ')), 'Escape did not cancel filename draft') require(not (root / 'Xright.rs').exists(), 'editing a filename wrote a file') require(client.read(f'/pane/{right}/body') == right_body, 'tag editing changed file body') checks.append('filename draft cancels without disk writes or body changes') # Keep a real embedded shell in the right column for useful captures. keys.click(12, 4) before = panes(client) keys.click(*require(locate(client, 'Tty', row=1, minimum_col=75), 'Tty missing'), button=1) terminal = require(wait_for(lambda: new_serial(client, before)), 'column Tty did not create a pane') require('pty' in client.list(f'/pane/{terminal}'), 'Tty created a non-terminal pane') client.write(f'/pane/{terminal}/pty/data', b"printf 'Column-local terminal ready\\n'; pwd\r") require(wait_for(lambda: b'\nColumn-local terminal ready\n' in client.read(f'/pane/{terminal}/body').replace(b'\r', b'')), 'terminal command output missing') checks.append('right-column embedded terminal executes harmless commands') # Persistence is checked through real Dump/Restore dispatch and the # resulting live header rows, not just serializer string matching. command(client, right, 'Dump') dump_path = root / (name + '.dump.zon') require(wait_for(dump_path.exists), 'Dump did not create an isolated session file') dumped = dump_path.read_bytes() saved_global = re.search(rb'\.topbar_custom\s*=\s*"((?:[^"\\]|\\.)*)"', dumped) require(saved_global and codecs.escape_decode(saved_global[1])[0].decode() == global_text, 'global custom tag absent from dump') require(b'New Tty Find Grep Joincol Work' in dumped, 'column custom tag absent from dump') keys.click(1, 0) keys.send(b'\x1b[F') keys.send(' TEMP') await_text(client, 0, ' TEMP') keys.escape() try: exec_tag_without_cleanup(client, right, 'Restore ' + str(dump_path)) except (ConnectionResetError, BrokenPipeError, EOFError): # Restore replaces the core and listener; even the write's final # clunk may lose its connection after successful dispatch. pass client.socket.close() client = connections.enter_context(require(wait_for(lambda: reconnect(address)), '9P did not return after Restore')) require(wait_for(lambda: row_text(client, 0) == global_text), 'Restore lost the custom global tag') await_text(client, 1, 'Work λ界') checks.append('Dump/Restore preserves custom global and column text') source_serial = require(next((serial for serial, text in panes(client).items() if str(source) in text), None), 'source pane lost after restore') right = require(next((serial for serial, text in panes(client).items() if str(right_name) in text), None), 'scratch pane lost after restore') (root / 'verified-headers.json').write_text(json.dumps({ 'global': global_text, 'columns_row': row_text(client, 1), 'pane_tag': tag(client, right), 'dump': str(dump_path), }, indent=2, ensure_ascii=False)) # Keep the tested Unicode and persistence evidence above. The published # workspace uses useful command text, edited through the same real input # route rather than modifying screenshots or injecting rendered cells. clean_global = 'Newcol Help Changelog Tutor Dump NextColor Debug Kill' keys.click(1, 0) keys.send(b'\x1b[H\x1b[1;2F') keys.send(clean_global) require(wait_for(lambda: row_text(client, 0) == clean_global), 'clean global tag replacement failed') keys.escape() col_start, _ = require(locate(client, 'New Tty', row=1, minimum_col=75), 'right column tag missing after restore') keys.click(col_start, 1) keys.send(b'\x1b[H\x1b[1;2F') keys.send('New Tty Find Grep Joincol Work') require(wait_for(lambda: row_text(client, 1).endswith('New Tty Find Grep Joincol Work')), 'clean column tag replacement failed') keys.escape() clean_pane = str(right_name) + ' Save Tty Del Collapse' current_pane = tag(client, right) require(current_pane.startswith(clean_pane), 'unexpected scratch tag before presentation cleanup') suffix = current_pane[len(clean_pane):] require(suffix.isascii(), 'presentation cleanup expects its own ASCII test suffix') keys.click(*require(locate(client, 'Save Tty', minimum_col=75), 'scratch pane commands missing')) keys.send(b'\x1b[F') keys.send(b'\x7f' * len(suffix)) require(wait_for(lambda: tag(client, right) == clean_pane), 'clean pane command tail replacement failed') keys.escape() checks.append('presentation tags cleaned through actual input after preserving Unicode evidence') for theme in ['orchard', 'atelier']: command(client, source_serial, 'Theme ' + theme) capture(client, root, 'columns-' + theme, not args.tty) # The source path exceeds the left column's width. Moving its tag # caret to End must reveal the actions without changing the filename. source_tag = tag(client, source_serial) keys.click(12, 2) keys.send(b'\x1b[F') require(wait_for(lambda: (point := locate(client, 'Save Tty Del', row=2)) and point[0] < 75), 'long-path tag did not reveal its command tail at End') require(tag(client, source_serial) == source_tag, 'caret movement changed the source tag') capture(client, root, 'columns-tag-reveal', not args.tty) keys.escape() checks.append('long-path pane tag reveals its command tail when the caret moves to End') require(client.read(f'/pane/{source_serial}/body') == original == source.read_bytes(), 'Agave source changed') checks.append('Agave source bytes unchanged') for serial in panes(client): if 'pty' in client.list(f'/pane/{serial}'): client.write(f'/pane/{serial}/pty/data', b'exit\r') report = {'checks': checks, 'cols': 150, 'rows': 48, 'host': 'tty' if args.tty else 'sdl', 'input': 'keyboard and SGR mouse through the host PTY parser', 'source': str(source), 'artifacts': str(root)} (root / 'report.json').write_text(json.dumps(report, indent=2)) print(json.dumps(report, indent=2)) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('binary', type=Path) parser.add_argument('output', type=Path) parser.add_argument('--project', required=True, type=Path) parser.add_argument('--tty', action='store_true') run(parser.parse_args())