Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234#!/usr/bin/env python3"""Actual ordinary TS/Rust PTY queue browsing and mutation comparison.
Operator-requested delta: Rust visits only user rows and swaps with the adjacentuser row; TS visits every projected item. Full decoded styled frames and rawANSI are retained, without claiming layout equality. A protocol fixture controlsonly the queue snapshot; both clients parse real input and construct mutations."""import argparseimport copyimport jsonimport osimport selectfrom pathlib import Pathimport subprocessimport sysimport tempfileimport time
from review_agent_bounce_parity import Clientfrom review_dock_screen import DockScreenfrom review_draft_disconnect_parity import Supervisor, Terminal, digest, env_for, TS_ARCHIVE_SHA256from review_goal_fixture_lifecycle import OwnedProcesses, TAG, finish
USER_A, USER_B = 'QUEUE_BROWSE_USER_A', 'QUEUE_BROWSE_USER_B'INTERNAL = 'Agent message received: QUEUE_BROWSE_INTERNAL'DRAFT = 'QUEUE_BROWSE_DRAFT'OLDER, NEWER, LATER = '\x1b[1;3A', '\x1b[1;3B', '\x1b[1;7B'
def check_mutation(kind, command): expected = {'type': 'mutate_queued_message', 'activeSessionId': 's1', 'lane': 'steering', 'index': 0, 'expectedText': USER_A, 'mutation': {'type': 'move', 'direction': 1 if kind == 'ts' else 2}} # TS places its request ID inside the command as well as its envelope. # Retain that raw transport ID, compare only the addressed mutation here. if 'id' in command: assert isinstance(command['id'], str) and command['id'], command assert {key: value for key, value in command.items() if key != 'id'} == expected, command
class QueueSupervisor(Supervisor): def __init__(self, path, hello, root, internal_only): self.queue = {'steering': [INTERNAL] if internal_only else [USER_A, INTERNAL, USER_B], 'followUp': []} self.mutations = [] super().__init__(path, hello, 'queue-browse', root)
def actions(self): return {'queuedCount': len(self.queue['steering']), 'steering': self.queue['steering'][:], 'followUps': []}
def send(self, sock, data): data = copy.deepcopy(data) name = data.get('command') if data.get('type') == 'daemon_hello': # Advertise the actual queue-mutation protocol this subclass owns. # The shared draft-only server deliberately omits this capability. data['serverCapabilities'] = ['session_input_admission', 'queue_message_mutation'] if data.get('type') == 'response': if name == 'attach': data['data']['snapshot']['state']['sessionActions'] = self.actions() elif name in ('get_state', 'get_connection_state'): data['data']['sessionActions'] = self.actions() super().send(sock, data)
def extra_command(self, command, response): name = command['type'] if name == 'get_queue': response['data'] = copy.deepcopy(self.queue) return True if name != 'mutate_queued_message': return False assert command['lane'] == 'steering', command assert command['mutation']['type'] == 'move', command index = command['index'] assert self.queue['steering'][index] == command['expectedText'], command target = index + command['mutation']['direction'] assert 0 <= target < len(self.queue['steering']), command self.queue['steering'][index], self.queue['steering'][target] = self.queue['steering'][target], self.queue['steering'][index] self.mutations.append(copy.deepcopy(command)) response['data'] = {'status': 'applied'} return True
def discover_hello(binary, root, env, observed): path = root / 'hello.sock' scope = OwnedProcesses(binary, root, path) daemon = client = None with open(root / 'hello.log', 'wb') as log: try: command = [str(binary), '--mode', 'daemon', '--daemon-socket', str(path)] daemon = subprocess.Popen(command, cwd=root, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) scope.register(daemon, command) client = Client(path, daemon) hello = client.hello assert hello.get('schemaId'), hello assert 'queue_message_mutation' in hello.get('serverCapabilities', []), hello observed['hello'] = hello finally: finish(scope, client, daemon, observed) assert observed['cleanup_verified'], observed return hello
def frame(terminal): return {'text': terminal.screen.text(), 'styles': terminal.screen.styles, 'unsupported_sgr': terminal.screen.unsupported_sgr[:], 'unsupported_terminal': terminal.screen.unsupported_terminal[:]}
def observe(binary, kind, root, internal_only, observed): env = env_for(root, kind) env[TAG] = str(root) hello_cleanup = observed['hello_discovery'] = {} hello = discover_hello(binary, root, env, hello_cleanup) server = QueueSupervisor(root / 'ui.sock', hello, root, internal_only) scope = OwnedProcesses(binary, root, server.path) terminal = None observed['frames'] = {} try: command = [str(binary), '--daemon-socket', str(server.path)] terminal = Terminal(command, root, env) terminal.screen = DockScreen() scope.register(terminal.process, command) def snap(label, predicate): terminal.wait(label, lambda: predicate(terminal.screen.text()), server) observed['frames'][label] = copy.deepcopy(frame(terminal)) snap('initial', lambda text: server.attaches > 0 and (INTERNAL in text if kind == 'ts' else '1 agent message' in text)) terminal.send(DRAFT) snap('draft', lambda text: DRAFT in text) if internal_only: if kind == 'rust': assert 'to browse and edit queued messages' not in terminal.screen.text() terminal.send(OLDER) if kind == 'rust': # The next typed character is an input-order barrier: Alt+Up was # handled before this character, so no absence-by-sleep assertion. terminal.send('_BARRIER') snap('internal_only_input', lambda text: DRAFT + '_BARRIER' in text) assert 'enter steers' not in terminal.screen.text() else: snap('internal_only_input', lambda text: 'steering 1' in text and 'enter steers' in text) else: terminal.send(OLDER) snap('newest', lambda text: ('steering 3' if kind == 'ts' else 'steering 2') in text and 'enter steers' in text) terminal.send(OLDER) snap('older', lambda text: ('steering 2' if kind == 'ts' else 'steering 1') in text and 'enter steers' in text) terminal.send(OLDER) if kind == 'ts': snap('oldest', lambda text: 'steering 1' in text and 'enter steers' in text) else: # At the oldest boundary, an input barrier proves the no-op was # processed; restore text so the reorder expectedText stays original. terminal.send('_BARRIER') snap('oldest', lambda text: USER_A + '_BARRIER' in text) terminal.send('\x7f' * len('_BARRIER')) snap('oldest_restored', lambda text: USER_A + '_BARRIER' not in text and 'steering 1' in text) terminal.send(LATER) terminal.wait('reorder command', lambda: len(server.mutations) == 1, server) check_mutation(kind, server.mutations[0]) snap('reordered', lambda text: ('steering 2' if kind == 'ts' else 'steering 2') in text and 'enter steers' in text) expected = [INTERNAL, USER_A, USER_B] if kind == 'ts' else [USER_B, INTERNAL, USER_A] assert server.queue == {'steering': expected, 'followUp': []}, server.queue assert not server.prompts, 'browse submitted a prompt' assert not terminal.screen.unsupported_sgr and not terminal.screen.unsupported_terminal, 'unsupported terminal sequences' observed['behavior_verified'] = True finally: try: if terminal: if terminal.process.poll() is None: terminal.send('\x03\x03') deadline = time.monotonic() + 8 while terminal.process.poll() is None and time.monotonic() < deadline: terminal.pump() # Drain already-readable final output after process exit too. while select.select([terminal.master], [], [], 0)[0]: before = len(terminal.raw) terminal.pump() if len(terminal.raw) == before: break # PTY EOF remains readable. observed['raw_ansi_complete_hex'] = bytes(terminal.raw).hex() observed['ui_exit_code'] = terminal.process.poll() # Do not call the older Terminal.close() force-stop helper. # Strict cleanup observes graceful exit or leaves this case RED. os.close(terminal.master) finally: try: server.close() finally: observed['commands'] = server.commands observed['mutations'] = server.mutations observed['queue_after'] = server.queue observed['fixture_errors'] = server.errors finish(scope, None, None, observed) assert observed['cleanup_verified'], observed assert not server.errors, server.errors
def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--receipt', required=True, type=Path) args = parser.parse_args() receipt = {'verified': False, 'scope': __doc__, 'runs': {}, 'not_verified': ['real daemon queue admission (separate queue_priority fixture)', 'pixel/font equality', 'delete/replace/attachments', 'wire-typed child and continuation riders (Rust units)']} try: assert sys.platform == 'linux', 'hosted Linux only' assert digest(Path(os.environ['PA_TS_ARCHIVE'])) == TS_ARCHIVE_SHA256, 'reference archive mismatch' receipt['ts_archive_sha256'] = TS_ARCHIVE_SHA256 with tempfile.TemporaryDirectory(prefix='pa-qbrowse-', dir='/tmp') as folder: for kind in ('ts', 'rust'): binary = Path(os.environ[f'PA_{kind.upper()}_BINARY']).resolve() for internal_only in (False, True): label = kind + ('_internal_only' if internal_only else '_mixed') observed = receipt['runs'][label] = {'binary_sha256': digest(binary)} try: observe(binary, kind, Path(folder) / label, internal_only, observed) except Exception as error: observed['error'] = str(error) assert all(run.get('behavior_verified') and run.get('cleanup_verified') and 'error' not in run for run in receipt['runs'].values()), 'actual PTY comparison failed' receipt['verified'] = True except Exception as error: receipt['error'] = str(error) args.receipt.parent.mkdir(parents=True, exist_ok=True) args.receipt.write_text(json.dumps(receipt, ensure_ascii=False, indent=2)) print(json.dumps({'verified': receipt['verified'], 'error': receipt.get('error')})) return 0 if receipt['verified'] else 1
if __name__ == '__main__': sys.exit(main())