"""Verify Profile routing against real Hermes with a local fake model endpoint. Run with the Hermes virtualenv Python; no production config or source is changed. Use --baseline to reproduce the unpatched metadata/environment override defects. """ import argparse import asyncio import importlib.util import json import os import shutil import socket import subprocess import tempfile import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import requests import websockets REPO = Path(__file__).resolve().parents[3] PLUGIN = REPO / 'integrations/hermes-bots-bridge' def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--source', type=Path, default=Path.home() / '.hermes/hermes-agent') parser.add_argument('--baseline', action='store_true') parser.add_argument('--no-env-seed', action='store_true', help='Isolate the default-Profile metadata leak') args = parser.parse_args() source = args.source.resolve() root = Path(tempfile.mkdtemp(prefix='yaoyao-profile-routing-')) overlay = root / 'source'; overlay.mkdir() # Copy Python packages: native bootstrap modules resolve their own path and # prepend that root. Symlinking them would silently load the installed code. for item in source.iterdir(): if item.is_dir() and (item / '__init__.py').is_file(): shutil.copytree(item, overlay / item.name, ignore=shutil.ignore_patterns('__pycache__')) elif item.is_file() and item.suffix == '.py': shutil.copy2(item, overlay / item.name) else: (overlay / item.name).symlink_to(item, target_is_directory=item.is_dir()) home = root / 'home' seen_models = [] class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): self.send_response(200); self.end_headers() self.wfile.write(json.dumps({'tools': [{'id': 'fixture_ping', 'name': 'fixture_ping', 'description': 'Return fixture proof', 'inputSchema': {'type': 'object', 'properties': {}}}]} if self.path == '/tools/list' else {'data': [{'id': 'fixture-default'}]}).encode()) def do_POST(self): body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0)))) if self.path == '/tools/list': return self.do_GET() seen_models.append(body.get('model')) message = {'role': 'assistant', 'content': 'fixture reply'} self.send_response(200) if body.get('stream'): self.send_header('Content-Type', 'text/event-stream'); self.end_headers() chunk = {'id': 'fixture', 'object': 'chat.completion.chunk', 'choices': [{'index': 0, 'delta': message, 'finish_reason': 'stop'}]} self.wfile.write(('data: ' + json.dumps(chunk) + '\n\ndata: [DONE]\n\n').encode()) else: self.send_header('Content-Type', 'application/json'); self.end_headers() self.wfile.write(json.dumps({'id': 'fixture', 'choices': [{'index': 0, 'message': message, 'finish_reason': 'stop'}], 'usage': {'prompt_tokens': 100, 'completion_tokens': 5, 'total_tokens': 105}}).encode()) http = ThreadingHTTPServer(('127.0.0.1', 0), Handler) threading.Thread(target=http.serve_forever, daemon=True).start() model_url = f'http://127.0.0.1:{http.server_port}' spec = importlib.util.spec_from_file_location('installer', REPO / 'scripts/install-hermes-bridge.py') installer = importlib.util.module_from_spec(spec); spec.loader.exec_module(installer) for profile, model in [('default', 'fixture-default'), ('yaoer', 'grok-4.6')]: selected = home if profile == 'default' else home / 'profiles' / profile (selected / 'workspace').mkdir(parents=True) config = {'model': {'default': model, 'provider': 'custom', 'base_url': model_url + '/v1', 'api_key': 'fixture', 'context_length': 65536}, 'terminal': {'cwd': str(selected / 'workspace')}, 'plugins': {'enabled': ['yaoyao-bot-bridge']}, 'toolsets': ['yaoyao_bot_bridge'], 'tools': {'tool_search': {'enabled': 'off'}}, 'dashboard': {'turn_isolation': False}, 'compression': {'enabled': False}, 'memory': {'memory_enabled': False, 'user_profile_enabled': False}} (selected / 'config.yaml').write_text(json.dumps(config)) (selected / '.env').write_text('') installer.install(home, profile, PLUGIN, repair_source=None if args.baseline else overlay) with socket.socket() as sock: sock.bind(('127.0.0.1', 0)); port = sock.getsockname()[1] env = {'PATH': os.environ['PATH'], 'HOME': str(root / 'os-home'), 'LANG': 'en_US.UTF-8', 'HERMES_HOME': str(home), 'PYTHONPATH': str(overlay), 'HERMES_MODEL': 'gpt-5.6-terra', 'HERMES_DASHBOARD_SESSION_TOKEN': 'fixture-profile-token-123456789', 'NO_PROXY': '127.0.0.1,localhost', 'PYTHONDONTWRITEBYTECODE': '1'} if args.no_env_seed: env.pop('HERMES_MODEL') log = (root / 'hermes.log').open('w') child = subprocess.Popen([str(source / 'venv/bin/python'), '-m', 'hermes_cli.main', 'serve', '--host', '127.0.0.1', '--port', str(port), '--skip-build', '--isolated'], env=env, cwd=root, stdout=log, stderr=log) base = f'http://127.0.0.1:{port}' client = requests.Session(); client.trust_env = False client.headers['X-Hermes-Session-Token'] = env['HERMES_DASHBOARD_SESSION_TOKEN'] def request(path, body=None): response = client.get(base + path, timeout=30) if body is None else client.post(base + path, json=body, timeout=30) response.raise_for_status() return response.json() async def rpc(ws, method, params): key = str(uuid.uuid4()) await ws.send(json.dumps({'jsonrpc': '2.0', 'id': key, 'method': method, 'params': params})) while True: frame = json.loads(await asyncio.wait_for(ws.recv(), 30)) if frame.get('id') == key: assert 'error' not in frame, frame return frame['result'] async def verify(): results = [] for profile, expected in [('default', 'fixture-default'), ('yaoer', 'grok-4.6')]: await asyncio.to_thread(request, '/api/config?profile=' + profile) caps = await asyncio.to_thread(request, '/api/plugins/yaoyao-bot-bridge/capabilities?profile=' + profile) assert caps['ready'], caps async with websockets.connect(f'ws://127.0.0.1:{port}/api/ws?token=' + env['HERMES_DASHBOARD_SESSION_TOKEN'], origin=base, proxy=None) as ws: while 'gateway.ready' not in await asyncio.wait_for(ws.recv(), 30): pass created = await rpc(ws, 'session.create', {'profile': profile, 'source': 'yaoyao_workspace', 'close_on_disconnect': False}) stored = created.get('stored_session_id') or created['session_key'] resumed = await rpc(ws, 'session.resume', {'profile': profile, 'session_id': stored, 'omit_messages': True, 'close_on_disconnect': False}) sid = resumed['session_id']; generation = str(uuid.uuid4()) start = time.monotonic() bound = await asyncio.to_thread(request, '/api/plugins/yaoyao-bot-bridge/bind', { 'session_id': sid, 'stored_session_id': stored, 'profile': profile, 'generation': generation, 'bridge_url': model_url, 'token': 'a' * 40, 'expires_at': time.time() * 1000 + 60000, 'native_tools': True, 'workspace_memory': True}) assert bound['ok'], bound elapsed = round(time.monotonic() - start, 3) offset = len(seen_models) await ws.send(json.dumps({'jsonrpc': '2.0', 'id': 'prompt', 'method': 'prompt.submit', 'params': {'session_id': sid, 'text': 'Hello'}})) while True: frame = json.loads(await asyncio.wait_for(ws.recv(), 30)) event = frame.get('params', {}).get('type') assert event != 'error' and 'error' not in frame, frame if event == 'message.complete': break models = seen_models[offset:] expected_model = 'gpt-5.6-terra' if args.baseline and not args.no_env_seed else expected expected_info = ('fixture-default' if args.no_env_seed else 'gpt-5.6-terra') if args.baseline else expected assert models and all(model == expected_model for model in models), models assert created['info']['model'] == expected_info, created assert resumed['info']['model'] == expected_info, resumed results.append({'profile': profile, 'configured': expected, 'create': created['info']['model'], 'resume': resumed['info']['model'], 'requests': models, 'bindSeconds': elapsed}) await asyncio.to_thread(request, '/api/plugins/yaoyao-bot-bridge/unbind', {'session_id': sid, 'generation': generation}) print(json.dumps({'baseline': args.baseline, 'environmentSeed': not args.no_env_seed, 'results': results}, ensure_ascii=False, indent=2), flush=True) try: deadline = time.monotonic() + 30 while time.monotonic() < deadline: if child.poll() is not None: raise RuntimeError('Hermes exited; inspect ' + str(root / 'hermes.log')) try: if client.get(base + '/api/status', timeout=1).status_code == 200: break except requests.RequestException: pass time.sleep(.2) else: raise TimeoutError('Hermes startup exceeded 30 seconds') asyncio.run(verify()) finally: child.terminate() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill(); child.wait() http.shutdown(); http.server_close(); log.close(); client.close() print('Fixture logs: ' + str(root), flush=True) if __name__ == '__main__': main()