Sybil-2/ios/Tests/Fixtures/agent_server.py

#!/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 base64
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
assistant_image_capability = False
fail_image_download = False
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
    global assistant_image_capability, fail_image_download
    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
            assistant_image_capability = False; fail_image_download = 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)
            assistant_image_capability = body.get('assistantImages', False)
        elif action == 'assistant-image':
            # Synthetic output supplied by XCTest, never a workspace file or provider.
            data = base64.b64decode(body['png'], validate=True)
            descriptor = dict(id='fixture-output-image', kind='image', filename='fixture-landscape.png',
                              mimeType='image/png', sizeBytes=len(data), sha256=hashlib.sha256(data).hexdigest())
            attachments[descriptor['id']] = dict(descriptor=descriptor, data=data)
            fail_image_download = body.get('failFirstDownload', False)
            row = message(body.get('content', ''), info={'agentAttachments': [descriptor]})
            await broadcast(dict(type='messages', **page([row])))
        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 == 'progress':
            busy = True
            progress = dict(activity(), queued=body['queued'], backgroundWorkers=body['workers'],
                            tools=[dict(id='local-review', name='review_transcript', status='running',
                                        output=f"Checked {body['step']} local entries")])
            await broadcast(dict(type='activity', conversationId='fixture-root', activity=progress))
        elif action == 'edit-last':
            cursor += 1
            row = dict(messages[-1], content=body['content'],
                       metadata=dict(messages[-1]['metadata'], entryId=cursor))
            messages[-1] = row
            await broadcast(dict(type='messages', **page([row])))
        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}
        if assistant_image_capability:
            descriptor.setdefault('capabilities', {})['assistantImages'] = dict(version=1)
        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'):
        if fail_image_download:
            fail_image_download = False
            return attachment_error(503, 'fixture_retry', 'Synthetic image unavailable. Retry the download.')
        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)