#!/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 import hashlib import unicodedata import uuid from urllib.parse import quote from aiohttp import web sockets = set() messages = [] requests = {} read_cursor = 0 cursor = 0 busy = False attempts = {} log = [] reply_tasks = set() attachments = {} uploads = {} attachment_capability = True next_upload_error = None blocked_proxy = False transcript_comparison = False PUSH_SOURCE = '11111111-1111-4111-8111-111111111111' push_installations = {} push_retired = set() push_requests = [] push_available = True auth_required = False CAPABILITY = dict(version=1, maxCount=8, maxTotalBytes=33554432, maxImageBytes=6291456, maxFileBytes=8388608, imageMimeTypes=['image/png', 'image/jpeg', 'image/gif', 'image/webp'], maxImagePixels=50000000, maxImageDimension=16384, quotaBytes=1073741824, usedBytes=0, remainingBytes=1073741824, maxConcurrentUploads=2, maxConcurrentDecodes=1) def attachment_error(status, code, message, **fields): return web.json_response(dict(error=True, message=message, statusCode=status, code=code, **fields), status=status, headers={"Retry-After": "2"} if status == 429 else None) 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, attachment_capability, next_upload_error, blocked_proxy, transcript_comparison, push_available, auth_required 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 attachment_capability = True; next_upload_error = None; blocked_proxy = False; transcript_comparison = False push_installations.clear(); push_retired.clear(); push_requests.clear(); log.clear() push_available = True; auth_required = False for i in range(body.get('count', 3)): message(f'History message {i + 1}: local simulator fixture.') elif action == 'presentation-samples': # Same source messages in normal chat and Agent; loopback-only UI comparisons. messages.clear(); cursor = 0; read_cursor = 0 transcript_comparison = True for sample in body['messages']: message(sample['content'], sample['role']) elif action == 'transcript-comparison': transcript_comparison = True elif action == 'capability': attachment_capability = body.get('enabled', True) elif action == 'upload-error': next_upload_error = body elif action == 'blocked-proxy': blocked_proxy = body.get('enabled', True) elif action == 'attachment-history': selected = [attachments[key]['descriptor'] for key in body['ids']] row = message(body.get('content') or 'Attached: ' + ', '.join(a['filename'] for a in selected) + '.', 'user', info={'agentAttachments': selected}) await broadcast(dict(type='messages', **page([row]))) 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 await broadcast(dict(type='read_state', sourceId=PUSH_SOURCE, cursor=str(read_cursor), updatedAt=stamp())) elif action == 'push-available': push_available = body.get('enabled', True) elif action == 'auth-required': auth_required = body.get('enabled', True) 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, pushRequests=push_requests, pushActive=len(push_installations), pushRetired=len(push_retired))) if path == '/api/v1/auth/session': if auth_required and request.headers.get('Authorization') != 'Bearer fixture-token': return web.json_response(dict(message='Fixture authentication required'), status=401) return web.json_response(dict(authenticated=True, mode='token')) comparison = dict(id='fade-comparison', title='Transcript fade comparison', createdAt=stamp(), updatedAt=stamp()) if path == '/api/v1/workspace-items': return web.json_response(dict(items=[dict(comparison, type='chat')] if transcript_comparison else [], hasMore=False)) if path == '/api/v1/chats': return web.json_response(dict(chats=[comparison] if transcript_comparison else [])) if path == '/api/v1/chats/fade-comparison' and transcript_comparison: return web.json_response(dict(chat=dict(comparison, messages=messages))) 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 blocked_proxy: return attachment_error(503, 'attachments_unavailable', 'Synthetic proxy requires safely authenticated Agent access.') if path == '/api/v1/agent': descriptor = dict(id='fixture-root', title='Sybil Agent', type='persistent-agent', model='fixture', protocol='pi-room.v1', websocket='/v1/agent/ws', activity=activity()) if attachment_capability: descriptor['capabilities'] = {'attachments': CAPABILITY} return web.json_response(descriptor) if path == '/api/v1/agent/push/config': if not push_available: return web.json_response(dict(message='Old server fixture'), status=404) return web.json_response(dict(version=1, enabled=True, apnsAvailable=True, webPushAvailable=False, sourceId=PUSH_SOURCE, quarantined=False, policy={})) if path.startswith('/api/v1/agent/push/installations/'): installation = str(uuid.UUID(path.rsplit('/', 1)[1])) if request.method == 'DELETE': push_installations.pop(installation, None); push_retired.add(installation) push_requests.append(dict(method='DELETE', installationId=installation)) return web.Response(status=204) if request.method == 'PUT': body = await request.json() assert set(body) == {'transport', 'enabled', 'token', 'environment', 'topic'} assert body['transport'] == 'apns' and body['enabled'] is True assert body['environment'] == 'sandbox' and body['topic'] == 'net.buzzert.sybil2' assert body['token'] == '00010aff' + '42' * 28 # explicitly injected synthetic token if installation in push_retired: return web.json_response(dict(message='Retired installation'), status=409) version = push_installations.get(installation, 0) + 1 push_installations[installation] = version push_requests.append(dict(method='PUT', installationId=installation, environment='sandbox')) return web.json_response(dict(installationId=installation, transport='apns', enabled=True, version=version)) if path == '/api/v1/agent/attachments' and request.method == 'POST': if not attachment_capability: return attachment_error(503, 'attachments_unavailable', 'Attachments are unavailable.') if next_upload_error: failure = next_upload_error; next_upload_error = None return attachment_error(failure.get('status', 507), failure.get('code', 'attachment_quota_exceeded'), 'Synthetic attachment storage is full; contact the operator or retry later.', **({'quotaBytes': 1073741824, 'remainingBytes': 0, 'requiredBytes': 20000} if failure.get('status', 507) == 507 else {})) reader = await request.multipart() parts = {} while part := await reader.next(): if part.name not in ('file', 'uploadId') or part.name in parts: return attachment_error(400, 'invalid_upload', 'Exactly one file and uploadId are required.') parts[part.name] = (part.filename, part.headers.get('Content-Type'), bytes(await part.read(decode=False))) if set(parts) != {'file', 'uploadId'}: return attachment_error(400, 'invalid_upload', 'Missing file or uploadId.') try: upload_id = str(uuid.UUID(parts['uploadId'][2].decode())) filename = unicodedata.normalize('NFC', parts['file'][0]) assert filename and len(filename.encode()) <= 255 and filename not in ('.', '..') assert not any(c in '/\\' or unicodedata.category(c) == 'Cc' for c in filename) except (ValueError, AssertionError, TypeError): return attachment_error(400, 'invalid_upload', 'Invalid upload ID or filename.') _, mime, data = parts['file'] mime = (mime or 'application/octet-stream').lower() if len(data) > CAPABILITY['maxFileBytes']: return attachment_error(413, 'attachment_limit_exceeded', 'File exceeds the upload limit.') digest = hashlib.sha256(data).hexdigest() identity = (filename, mime, digest) if upload_id in uploads: old_identity, attachment_id = uploads[upload_id] if old_identity != identity: return attachment_error(409, 'idempotency_conflict', 'Upload ID already belongs to another original.') return web.json_response({'attachment': attachments[attachment_id]['descriptor']}) kind = 'file' if mime in CAPABILITY['imageMimeTypes']: kind = 'image' # Synthetic fixture images are generated by the test; runtime validates decoding. elif mime.startswith('text/'): try: if '\0' not in data.decode('utf-8-sig'): kind = 'text' except UnicodeDecodeError: pass attachment_id = 'fixture-' + str(uuid.uuid4()) descriptor = dict(id=attachment_id, kind=kind, filename=filename, mimeType=mime, sizeBytes=len(data), sha256=digest, workspacePath='/data/workspace/uploads/' + attachment_id + '/' + ('_' + filename[1:] if filename.startswith('.') else filename)) attachments[attachment_id] = dict(descriptor=descriptor, data=data) uploads[upload_id] = (identity, attachment_id) return web.json_response({'attachment': descriptor}, status=201) if path.startswith('/api/v1/agent/attachments/') and path.endswith('/content'): attachment_id = path.removeprefix('/api/v1/agent/attachments/').removesuffix('/content') stored = attachments.get(attachment_id) if stored is None: return attachment_error(404, 'attachment_not_found', 'The original is unavailable.') descriptor = stored['descriptor'] mime = descriptor['mimeType'] if descriptor['kind'] == 'image' else 'text/plain' if descriptor['kind'] == 'text' else 'application/octet-stream' return web.Response(body=stored['data'], content_type=mime, headers={'Cache-Control': 'private, no-store', 'Content-Disposition': 'attachment; filename="original"; filename*=UTF-8\'\'' + quote(descriptor['filename'], safe=''), 'X-Content-Type-Options': 'nosniff'}) 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(), sourceId=PUSH_SOURCE)) if path == '/api/v1/agent/messages': if request.method == 'POST': body = await request.json() key, content = body['requestId'], body['content'] refs = body.get('attachments', []) ids = [ref.get('id') for ref in refs] if refs and not attachment_capability: return attachment_error(503, 'attachments_unavailable', 'Attachments are unavailable.') if any(set(ref) != {'id'} for ref in refs) or len(ids) != len(set(ids)) or len(ids) > 8: return attachment_error(400, 'invalid_attachment', 'Invalid ordered references.') if any(key not in attachments for key in ids): return attachment_error(404, 'attachment_not_found', 'The attachment does not exist.') if not content.strip() and not ids: return attachment_error(400, 'invalid_attachment', 'Enter a caption or attach files.') if key in requests and (ids or requests[key].get('attachmentIds')) and requests[key].get('_payload') != [content, ids]: return attachment_error(409, 'idempotency_conflict', 'Message identity is already bound to another payload.') 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 ids: receipt['attachmentIds'] = ids if key not in requests: requests[key] = dict(receipt, _payload=[content, ids]) descriptors = [attachments[key]['descriptor'] for key in ids] display = content if content.strip() else 'Attached: ' + ', '.join(a['filename'] for a in descriptors) + '.' row = message(display, 'user', id=receipt['messageId'], info={'agentAttachments': descriptors} if ids else None) 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({k: v for k, v in requests[key].items() if k != '_payload'}, 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) if blocked_proxy: await socket.close(code=4503, message=b'attachments_unavailable') return socket 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(client_max_size=8 * 1024 * 1024 + 64 * 1024) app.router.add_route('*', '/{path:.*}', route) web.run_app(app, host='127.0.0.1', port=8799)