#!/usr/bin/env python3 """Local-only deterministic Agent contract fixture. Never contacts external services. Run with Python + aiohttp, then xcodebuild test -scheme SybilAgentUI. """ import asyncio import datetime import json from aiohttp import web sockets = set() messages = [] requests = {} read_cursor = 0 cursor = 0 busy = False attempts = {} log = [] reply_tasks = set() def stamp(): return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z') def activity(): return dict(state='thinking' if busy else 'idle', isThinking=busy, queued=0, backgroundWorkers=0, tools=[], queuedMessages=[]) def message(content, role='assistant', id=None, info=None): global cursor cursor += 1 row = dict(id=id or f'outbound:{cursor}', role=role, content=content, createdAt=stamp(), metadata=dict(kind='message', entryId=cursor, status='complete', **(info or {}))) messages.append(row) return row def page(rows, position=None, more=False): return dict(conversationId='fixture-root', messages=rows, cursor=str(cursor if position is None else position), beforeCursor='1', hasMore=more) async def broadcast(frame): for socket in list(sockets): try: await socket.send_json(frame) except ConnectionError: sockets.discard(socket) async def emit(content): row = message(content) await broadcast(dict(type='messages', **page([row]))) return row async def reply(content): global busy, cursor await asyncio.sleep(1.2) tool = message('Inspecting the fixture', 'tool', id=f'tool:{cursor + 1}') tool['metadata'].update(kind='tool_call', toolCallId=tool['id'], toolName='fixture_search', status='initiated', args={'query': content}, startedAt=stamp()) await broadcast(dict(type='messages', **page([tool]))) await asyncio.sleep(0.8) cursor += 1 tool = dict(tool, metadata=dict(tool['metadata'], entryId=cursor, status='completed', resultPreview='Fixture result: 3 matches')) messages[messages.index(next(m for m in messages if m['id'] == tool['id']))] = tool await broadcast(dict(type='messages', **page([tool]))) failed = message('Fixture unavailable', 'tool', id=f'tool:{cursor + 1}') failed['metadata'].update(kind='tool_call', toolName='fixture_error', status='failed', args={'source': 'local only'}, error='Simulated tool failure') await broadcast(dict(type='messages', **page([failed]))) await emit(f'**Finished:** {content}\n\nThe local fixture returned three matches.') # Deliberately stale busy frame after the public reply. await broadcast(dict(type='activity', conversationId='fixture-root', activity=activity())) await asyncio.sleep(0.5) busy = False await broadcast(dict(type='activity', conversationId='fixture-root', activity=activity())) async def route(request): global read_cursor, cursor, busy path = request.path log.append(dict(method=request.method, path=path, query=dict(request.query))) if path == '/control': body = await request.json() action = body.get('action') if action == 'reset': for task in reply_tasks: task.cancel() reply_tasks.clear() messages.clear(); requests.clear(); attempts.clear(); cursor = 0; read_cursor = 0; busy = False for i in range(body.get('count', 3)): message(f'History message {i + 1}: local simulator fixture.') elif action == 'reply': await emit(body.get('content', 'Late asynchronous fixture reply')) elif action == 'long-tool': row = message('Local tool output', 'tool') row['metadata'].update(kind='tool_call', toolName='fixture_long', status='completed', args={'source': 'local only'}, result='\n'.join(f'Result line {i}' for i in range(80))) await broadcast(dict(type='messages', **page([row]))) elif action == 'busy': busy = body.get('value', True) await broadcast(dict(type='activity', conversationId='fixture-root', activity=activity())) elif action == 'read': read_cursor = cursor elif action == 'disconnect': for socket in list(sockets): await socket.close(code=1012) elif action == 'malformed': await broadcast(dict(type='messages', conversationId='fixture-root', messages=[{'id': 'broken'}], cursor='999999', beforeCursor=None, hasMore=False)) return web.json_response(dict(cursor=cursor, readCursor=read_cursor, messages=messages, requests=requests, log=log)) if path == '/api/v1/auth/session': return web.json_response(dict(authenticated=True, mode='token')) if path == '/api/v1/workspace-items': return web.json_response(dict(items=[], hasMore=False)) if path == '/api/v1/chats': return web.json_response(dict(chats=[])) if path == '/api/v1/searches': return web.json_response(dict(searches=[])) if path == '/api/v1/models': return web.json_response(dict(providers={})) if path == '/api/v1/active-runs': return web.json_response(dict(chats=[], searches=[])) if path.startswith('/api/v1/agent') and path != '/api/v1/agent/ws': if request.headers.get('Authorization') != 'Bearer fixture-token': return web.json_response(dict(error='Fixture token missing'), status=401) if path == '/api/v1/agent': return web.json_response(dict(id='fixture-root', title='Sybil Agent', type='persistent-agent', model='fixture', protocol='pi-room.v1', websocket='/v1/agent/ws', activity=activity())) if path == '/api/v1/agent/read-state': if request.method == 'PUT': read_cursor = max(read_cursor, int((await request.json())['cursor'])) return web.json_response(dict(cursor=str(read_cursor), updatedAt=stamp())) if path == '/api/v1/agent/messages': if request.method == 'POST': body = await request.json() key, content = body['requestId'], body['content'] attempts[key] = attempts.get(key, 0) + 1 if content == 'retry fixture' and attempts[key] == 1: return web.json_response(dict(error='Simulated uncertain admission; retry'), status=502) receipt = dict(id=len(requests) + 1, status='queued', messageId=f'inbound:sybil:{key}') if key not in requests: requests[key] = receipt row = message(content, 'user', id=receipt['messageId']) busy = True await broadcast(dict(type='messages', **page([row]))) await broadcast(dict(type='activity', conversationId='fixture-root', activity=activity())) task = asyncio.create_task(reply(content)) reply_tasks.add(task) task.add_done_callback(reply_tasks.discard) if content == 'echo wins fixture': await asyncio.sleep(0.3) return web.json_response(dict(error='Lost HTTP acknowledgement'), status=502) return web.json_response(requests[key], status=202) after = int(request.query.get('after', 0)) rows = [m for m in messages if m['metadata']['entryId'] > after] limit = min(int(request.query.get('limit', 200)), 200) selected = rows[:limit] return web.json_response(page(selected, selected[-1]['metadata']['entryId'] if len(rows) > limit else cursor, len(rows) > limit)) if path == '/api/v1/agent/ws': if request.query: return web.Response(status=400) socket = web.WebSocketResponse(); await socket.prepare(request) auth = await socket.receive_json(timeout=5) if auth.get('token') != 'fixture-token' or auth.get('type') != 'auth': await socket.close(code=4401); return socket log.append(dict(authAfter=auth.get('after'))) sockets.add(socket) await socket.send_json(dict(type='ready', protocol='pi-room.v1', conversationId='fixture-root')) await socket.send_json(dict(type='messages', **page([m for m in messages if m['metadata']['entryId'] > int(auth.get('after', 0))]))) await socket.send_json(dict(type='activity', conversationId='fixture-root', activity=activity())) try: async for frame in socket: if frame.type == web.WSMsgType.TEXT: data = json.loads(frame.data) if data.get('type') == 'ping': await socket.send_json(dict(type='pong')) finally: sockets.discard(socket) return socket return web.json_response(dict(error='No fixture for this endpoint'), status=404) for i in range(3): message(f'History message {i + 1}: local simulator fixture.') app = web.Application() app.router.add_route('*', '/{path:.*}', route) web.run_app(app, host='127.0.0.1', port=8799)