"""Local I/O probe for completion batching, not a hosted-model or native UI test. Run with the repo's Python: evals/completion_backlog_probe.py REPO OUTPUT.json. Real shell children feed ProcessRegistry; production CLI/poller/post-turn routing feeds a loopback HTTP turn sink. The sink replaces chat/_run_prompt_submit, NOT ownership, consumption, queue draining, batching, or formatting. """ import argparse import contextlib import json import os from pathlib import Path import queue import shlex import sys import tempfile import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace from unittest.mock import patch from urllib.request import Request, urlopen def probe(surface, scenario, directory): from cli import HermesCLI from tools import process_registry as pr from tools.process_registry_notifications import format_process_notification from tui_gateway import server directory = Path(directory) directory.mkdir(parents=True, exist_ok=True) registry = pr.ProcessRegistry() received, statuses = [], [] class Sink(BaseHTTPRequestHandler): def do_POST(self): received.append(json.loads(self.rfile.read(int(self.headers['Content-Length'])))) self.send_response(200) self.end_headers() self.wfile.write(b'{"accepted":true}') def log_message(self, *_args): return wire = ThreadingHTTPServer(('127.0.0.1', 0), Sink) thread = threading.Thread(target=wire.serve_forever, daemon=True) thread.start() session = {'session_key': 'backlog-owner', 'history_lock': threading.RLock()} cli = HermesCLI.__new__(HermesCLI) cli.session_id = session['session_key'] cli._session_db = None cli._pending_input = queue.Queue() cli._pending_resume_sessions = [] cli._typed_voice_stop = lambda _text: False cli.handle_bang_shell = lambda _text: False cli._print_user_message_preview = lambda _text: None cli._turn_summary_begin = lambda: None cli._app = SimpleNamespace(invalidate=lambda: None) cli._tui_after_turn = lambda: None def submit(text): request = Request(f'http://127.0.0.1:{wire.server_port}/turn', data=json.dumps({'text': str(text)}).encode(), headers={'Content-Type': 'application/json'}) with urlopen(request, timeout=10) as response: assert response.status == 200 session['running'] = False return True def tui_submit(_rid, _sid, _session, text, **_kwargs): return submit(text) cli.chat = lambda text, **_kwargs: submit(text) processes = [] try: with contextlib.ExitStack() as stack: stack.enter_context(patch.object(pr, 'process_registry', registry)) stack.enter_context(patch.object(pr, 'CHECKPOINT_PATH', directory / 'processes.json')) stack.enter_context(patch.object(server, '_sessions', {'ui-owner': session})) stack.enter_context(patch.object(server, '_get_db', lambda: None)) stack.enter_context(patch.object(server, '_emit', lambda *args: statuses.append(args))) stack.enter_context(patch.object(server, '_run_prompt_submit', tui_submit)) stack.enter_context(patch.object(server, '_drain_queued_prompt', lambda *_a: False)) count = 1 if scenario == 'single' else 12 gate = directory / 'release' for index in range(count): code = ('import pathlib,time,sys; p=pathlib.Path(sys.argv[1]); ' '\nwhile not p.exists(): time.sleep(.01)\n' f'print("BACKLOG_{index}"); sys.exit({7 if index == count - 1 else 0})') process = registry.spawn_local( f'{shlex.quote(sys.executable)} -c {shlex.quote(code)} {shlex.quote(str(gate))}', cwd=str(directory), session_key='backlog-owner') process.notify_on_complete = True processes.append(process) gate.touch() deadline = time.monotonic() + 30 while registry.completion_queue.qsize() < count and time.monotonic() < deadline: time.sleep(.01) assert registry.completion_queue.qsize() == count raw = list(registry.completion_queue.queue) delegation = None if scenario == 'mixed': from tools import async_delegation as ad delegation = {'type': 'async_delegation', 'delegation_id': f'deleg-{surface}', 'session_key': 'backlog-owner', 'goal': 'Fixture delegation', 'status': 'completed', 'summary': 'DELEGATION_RESULT', 'dispatched_at': time.time(), 'completed_at': time.time()} ad._persist_dispatch(delegation) ad._persist_completion(delegation, {'status': 'completed'}) watch = {'type': 'watch_match', 'session_key': 'backlog-owner', 'session_id': processes[0].id, 'pattern': 'READY', 'output': 'WATCH_READY'} while not registry.completion_queue.empty(): registry.completion_queue.get_nowait() raw = raw[:4] + [watch] + raw[4:8] + [delegation] + raw[8:] for event in raw: registry.completion_queue.put(event) expected = [format_process_notification(event) for event in raw] if scenario == 'foreign': session['session_key'] = cli.session_id = 'another-owner' server._sessions['actual-owner'] = {'session_key': 'backlog-owner'} if surface == 'cli': cli._drain_process_notifications('cli-idle') if scenario == 'consumed': for process in processes: assert registry.wait(process.id, timeout=1)["status"] == "exited" while not cli._pending_input.empty(): cli._tui_process_one_input(cli._pending_input.get_nowait()) elif surface == 'post-turn': if scenario == 'consumed': for process in processes: assert registry.wait(process.id, timeout=1)["status"] == "exited" server._run_post_turn_followups('probe', 'ui-owner', session, {}, None) else: if scenario == 'consumed': for process in processes: assert registry.wait(process.id, timeout=1)["status"] == "exited" stop = threading.Event() # Skip unrelated scheduled jobs; exercise the production poller loop. stack.enter_context(patch.object(server, '_maybe_fire_tui_loop_tick', lambda *_a: None)) stack.enter_context(patch.object(server, '_maybe_fire_tui_heartbeat_tick', lambda *_a: None)) stack.enter_context(patch.object(server, '_notif_poll_kanban', lambda *_a: None)) poller = threading.Thread(target=server._notification_poller_loop, args=(stop, 'ui-owner', session)) poller.start() deadline = time.monotonic() + (1 if scenario == 'foreign' else 10) while time.monotonic() < deadline: if scenario != 'foreign' and registry.completion_queue.empty(): break time.sleep(.01) stop.set() poller.join(10) assert not poller.is_alive() texts = [item['text'] for item in received] delegation_delivered = None if delegation: with ad._transaction() as conn: row = conn.execute("SELECT delivery_state, delivery_attempts FROM async_delegations WHERE delegation_id=?", (delegation['delegation_id'],)).fetchone() delegation_delivered = tuple(row) == ('delivered', 1) return {'surface': surface, 'scenario': scenario, 'children': count, 'wire_turns': len(texts), 'texts': texts, 'delegation_delivered_once': delegation_delivered, 'payload_order': [next((i for i, turn in enumerate(texts) if payload in turn), -1) for payload in expected], 'all_payloads_preserved': all(any(text in turn for turn in texts) for text in expected), 'single_exact': texts == expected if count == 1 else None, 'queue_remaining': registry.completion_queue.qsize(), 'status_events': len(statuses)} finally: wire.shutdown() wire.server_close() thread.join() for process in processes: if not process.exited: registry.kill_process(process.id) def main(): parser = argparse.ArgumentParser() parser.add_argument('repo') parser.add_argument('output') args = parser.parse_args() sys.path.insert(0, args.repo) with tempfile.TemporaryDirectory(prefix='completion-probe-') as home: os.environ['HERMES_HOME'] = home os.environ['HOME'] = home results = [probe(surface, scenario, Path(home) / surface / scenario) for surface in ('cli', 'poller', 'post-turn') for scenario in ('backlog', 'single', 'consumed', 'foreign', 'mixed')] Path(args.output).write_text(json.dumps(results, indent=2), encoding="utf-8") print(json.dumps([{k: v for k, v in row.items() if k != 'texts'} for row in results], indent=2)) if __name__ == '__main__': main()