Refactor panes and filesystem; replace FUSE with 9P

Consolidate pane, layout, memory and host code. Serve 9P by default over Unix sockets, with runtime mounts and optional TCP/QUIC transports. Remove FUSE and obsolete proof-of-concept examples.

Fix highlighting and terminal-history performance, expand differential and stress-test infrastructure, sort navigation results while preserving the next occurrence, add syntax-colored Braille minimaps, remove SPC-k, and document 9P interaction as a repository skill.

1 files changed, 171 insertions, 0 deletions

diff --git a/test/agent_session.py b/test/agent_session.py
new file mode 100644
index 00000000..b920b006
--- /dev/null
+++ b/test/agent_session.py
@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import shlex
+import signal
+import sys
+import tempfile
+import time
+import uuid
+
+from fs import session
+
+
+def screen_text(client):
+    screen = client.screen()
+    cols = screen['cols']
+    return '\n'.join(''.join(cell[0] for cell in screen['cells'][at:at + cols])
+                     for at in range(0, len(screen['cells']), cols))
+
+
+def expired(signum, frame):
+    raise TimeoutError('session deadline exceeded')
+
+
+def wait_ready(client, ready):
+    previous = None
+    while True:
+        text = screen_text(client)
+        if ready in text and text == previous:
+            return text
+        previous = text
+        time.sleep(.05)
+
+
+def run(args, command, status):
+    temporary = tempfile.TemporaryDirectory(prefix='pardes-agent-session-')
+    try:
+        root = Path(temporary.name)
+        config = root / 'config' / 'pardes'
+        config.mkdir(parents=True)
+        (config / 'init').write_text('Shell /bin/sh\n')
+        script = root / 'command.sh'
+        script.write_text('exec ' + shlex.join(command) + '\n')
+        old_handler = signal.signal(signal.SIGALRM, expired)
+        signal.setitimer(signal.ITIMER_REAL, args.timeout)
+        try:
+            context = session(str(args.binary.resolve()), root, 'agent-session', tty=True,
+                              socket_name='agent-session', launch=['--tty', '--9p=agent-session'],
+                              inherited={'PARDES_TEST_GRID': '1'} if args.gui_grid else {})
+            client, _ = context.__enter__()
+            try:
+                try:
+                    while not client.read('/self/pane/1/body').strip():
+                        time.sleep(.01)
+                    started = time.monotonic()
+                    client.write('/self/pane/1/pty/data', shlex.join(['/bin/sh', str(script)]).encode() + b'\r')
+                    status['phase'] = 'readiness'
+                    wait_ready(client, args.ready)
+                    if args.prepare_key is not None:
+                        status['phase'] = 'preparation'
+                        client.write('/self/pane/1/pty/data', args.prepare_key)
+                        wait_ready(client, args.ready)
+                    ready_ms = (time.monotonic() - started) * 1000
+                    status.update(phase='history', required_body_rows=args.min_rows)
+                    while True:
+                        started = time.monotonic()
+                        body = client.read('/self/pane/1/body')
+                        body_read_ms = (time.monotonic() - started) * 1000
+                        rows = body.count(b'\n') + int(bool(body) and not body.endswith(b'\n'))
+                        status.update(body_bytes=len(body), body_rows=rows)
+                        if rows >= args.min_rows:
+                            break
+                        time.sleep(.05)
+                    status['phase'] = 'settling'
+                    before = wait_ready(client, args.ready)
+                    status['expected_visible_text_sha256'] = hashlib.sha256(before.encode()).hexdigest()
+                    probe = 'p' + uuid.uuid4().hex[:12]
+                    if probe in before:
+                        raise AssertionError('probe is already visible')
+                    status['phase'] = 'probe-visible'
+                    started = time.monotonic()
+                    client.write('/self/pane/1/pty/data', probe.encode())
+                    while probe not in screen_text(client).replace('\n', ''):
+                        time.sleep(.01)
+                    write_ms = (time.monotonic() - started) * 1000
+                    status['phase'] = 'probe-cleared'
+                    started = time.monotonic()
+                    client.write('/self/pane/1/pty/data', b'\x15')
+                    while True:
+                        observed = screen_text(client)
+                        status['observed_visible_text_sha256'] = hashlib.sha256(observed.encode()).hexdigest()
+                        if observed == before:
+                            break
+                        time.sleep(.01)
+                    status['metrics'] = {
+                        'ready_ms': ready_ms, 'body_read_ms': body_read_ms,
+                        'ninep_write_to_observation_ms': write_ms,
+                        'ninep_clear_to_observation_ms': (time.monotonic() - started) * 1000,
+                        'visible_text_sha256': hashlib.sha256(before.encode()).hexdigest(),
+                        'body_bytes': len(body), 'body_rows': rows,
+                        'body_sha256': hashlib.sha256(body).hexdigest(),
+                    }
+                finally:
+                    previous_phase = status['phase']
+                    status['phase'] = 'interrupt-cleanup'
+                    signal.setitimer(signal.ITIMER_REAL, 0)
+                    client.socket.settimeout(.25)
+                    for _ in range(2):
+                        try:
+                            client.write('/self/pane/1/pty/data', b'\x03')
+                        except (OSError, EOFError):
+                            pass
+                        time.sleep(.05)
+                    status['phase'] = previous_phase
+            finally:
+                previous_phase = status['phase']
+                status['phase'] = 'session-cleanup'
+                context.__exit__(*sys.exc_info())
+                status['phase'] = previous_phase
+        finally:
+            signal.setitimer(signal.ITIMER_REAL, 0)
+            signal.signal(signal.SIGALRM, old_handler)
+    finally:
+        previous_phase = status['phase']
+        status['phase'] = 'files-cleanup'
+        temporary.cleanup()
+        status['phase'] = previous_phase
+    return status.pop('metrics')
+
+
+def main(argv):
+    parser = argparse.ArgumentParser(description='Check an unsubmitted input probe through an owned Pardes 9P session.')
+    parser.add_argument('binary', type=Path)
+    parser.add_argument('--ready', required=True, help='visible application readiness text')
+    parser.add_argument('--timeout', type=float, default=60, help='interaction deadline in seconds, at most 600')
+    parser.add_argument('--min-rows', type=int, default=0)
+    parser.add_argument('--prepare-key', help='one hex control byte before history/probe checks; never CR or LF')
+    parser.add_argument('--gui-grid', action='store_true', help='use the GUI build headless grid host')
+    if '--' not in argv:
+        parser.error('supply the interactive command after --')
+    split = argv.index('--')
+    args = parser.parse_args(argv[:split])
+    command = argv[split + 1:]
+    if not command or not command[0] or not args.ready.strip():
+        parser.error('a command and nonempty --ready text are required')
+    if not 0 < args.timeout <= 600 or args.min_rows < 0:
+        parser.error('--timeout must be in (0, 600] and --min-rows must be nonnegative')
+    if args.prepare_key is not None:
+        try:
+            args.prepare_key = bytes.fromhex(args.prepare_key)
+        except ValueError:
+            parser.error('--prepare-key must be one hex control byte')
+        if len(args.prepare_key) != 1 or not 0 < args.prepare_key[0] < 32 or args.prepare_key[0] in (10, 13):
+            parser.error('--prepare-key must be one control byte other than CR or LF')
+    status = {'phase': 'startup'}
+    try:
+        metrics = run(args, command, status)
+    except (OSError, EOFError, AssertionError, ValueError) as error:
+        status['error'] = type(error).__name__
+        if isinstance(error, OSError):
+            status['errno'] = error.errno
+        print(json.dumps(status), file=sys.stderr)
+        return 1
+    print(json.dumps(metrics))
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main(sys.argv[1:]))