"""Authenticated durable chat routes; all SQLite work is offloaded.""" import asyncio import json from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import StreamingResponse from .app import bounded_json, current_user, internal_authorized router = APIRouter() TERMINAL = {'completed', 'failed', 'cancelled'} async def db(request, method, *args, **kwargs): from .chat_store import ChatConflict try: return await asyncio.to_thread(getattr(request.app.state.chat_store, method), *args, **kwargs) except ChatConflict: raise HTTPException(409, 'Revision or operation conflict') from None except KeyError: raise HTTPException(404, 'Chat resource not found') from None except ValueError: raise HTTPException(400, 'Invalid chat request') from None async def body(request, allowed, required=()): value = await bounded_json(request) if not set(value) <= set(allowed) or not set(required) <= set(value): raise HTTPException(400, 'Invalid chat fields') return value @router.get('/api/capabilities') async def capabilities(request: Request, user=Depends(current_user)): from .chat_features import local_capabilities features = request.app.state.chat_runs.capabilities() features.update(local_capabilities(request.app.state.feature_store)) try: features.update(await request.app.state.feature_provider.feature_capabilities(user['id'])) except HTTPException: features.update({'image_input': False, 'scheduled_tasks': False, 'memory_controls': False, 'connectors': False, 'research': False}) return {'version': 1, 'features': features, 'limits': {'message_bytes': 6000, 'page_size': 100, 'attachments': request.app.state.feature_store.limits.attachments_per_message}} @router.get('/api/conversations') async def conversations(request: Request, q: str = '', workspace_id: str | None = None, archived: bool = False, cursor: str | None = None, limit: int = 50, user=Depends(current_user)): return await db(request, 'list_conversations', user['id'], q=q, workspace_id=workspace_id, archived=archived, cursor=cursor, limit=limit) @router.post('/api/conversations') async def create_conversation(request: Request, user=Depends(current_user)): return {'conversation': await db(request, 'create_conversation', user['id'], **await body(request, ('title','workspace_id','temporary')))} @router.get('/api/conversations/{cid}') async def conversation(cid: str, request: Request, cursor: str | None = None, limit: int = 50, user=Depends(current_user)): return await db(request, 'get_conversation', user['id'], cid, cursor=cursor, limit=limit) @router.patch('/api/conversations/{cid}') async def update_conversation(cid: str, request: Request, user=Depends(current_user)): return {'conversation': await db(request, 'update_conversation', user['id'], cid, **await body(request, ('revision','title','workspace_id','pinned','pin_rank','archived'), ('revision',)))} @router.delete('/api/conversations/{cid}') async def delete_conversation(cid: str, request: Request, user=Depends(current_user)): from .chat_store import ChatConflict try: await request.app.state.chat_runs.cancel_conversation(user['id'], cid) except ChatConflict: raise HTTPException(409, 'Cancellation pending; retry deletion after execution stops') from None except KeyError: raise HTTPException(404, 'Chat resource not found') from None detail = await db(request, 'get_conversation', user['id'], cid) if any(run['status'] not in TERMINAL for run in detail['runs']): raise HTTPException(409, 'Cancellation pending; retry deletion after execution stops') await db(request, 'delete_conversation', user['id'], cid) return {'ok': True} @router.post('/api/conversations/{cid}/branch') async def branch(cid: str, request: Request, user=Depends(current_user)): return {'conversation': await db(request, 'branch', user['id'], cid, **await body(request, ('message_id','content','title'), ('message_id',)))} @router.post('/api/conversations/{cid}/messages', status_code=202) async def send(cid: str, request: Request, user=Depends(current_user)): value = await body(request, ('content','idempotency_key','attachment_ids','mode'), ('content','idempotency_key')) return await request.app.state.chat_runs.send(user['id'], cid, request.state.session_token, **value) @router.get('/api/runs/{rid}') async def run(rid: str, request: Request, user=Depends(current_user)): return await db(request, 'get_run', user['id'], rid) async def event_response(request, owner, rid, valid, after): await db(request, 'get_run', owner, rid) try: after = max(after, int(request.headers.get('last-event-id', '0'))) except ValueError: raise HTTPException(400, 'Invalid event cursor') from None if after < 0: raise HTTPException(400, 'Invalid event cursor') async def stream(): sequence = after while not await request.is_disconnected(): if not await valid(): return try: items = await asyncio.to_thread(request.app.state.chat_store.events, owner, rid, sequence) detail = await asyncio.to_thread(request.app.state.chat_store.get_run, owner, rid) except KeyError: return for item in items: sequence = item['sequence'] yield f"id: {sequence}\nevent: {item['event']}\ndata: {json.dumps(item['data'], ensure_ascii=False)}\n\n" if not items and detail['run']['status'] in TERMINAL: return if not items: yield ': keepalive\n\n' await asyncio.sleep(.25) return StreamingResponse(stream(), media_type='text/event-stream', headers={'Cache-Control':'no-store','X-Accel-Buffering':'no'}) @router.get('/api/runs/{rid}/events') async def events(rid: str, request: Request, after: int = 0, user=Depends(current_user)): async def valid(): account = await asyncio.to_thread(request.app.state.accounts.resolve_session, request.state.session_token) return account is not None and account['id'] == user['id'] return await event_response(request, user['id'], rid, valid, after) @router.post('/api/runs/{rid}/cancel') async def cancel(rid: str, request: Request, user=Depends(current_user)): return {'run': await request.app.state.chat_runs.cancel(user['id'], rid)} @router.post('/api/runs/{rid}/approval') async def approval(rid: str, request: Request, user=Depends(current_user)): return {'run': await request.app.state.chat_runs.approval(user['id'], rid, **await body(request, ('approval_id','decision'), ('approval_id','decision')))} @router.get('/api/workspaces') async def workspaces(request: Request, user=Depends(current_user)): return {'workspaces': await db(request, 'list_workspaces', user['id'])} @router.post('/api/workspaces') async def create_workspace(request: Request, user=Depends(current_user)): return {'workspace': await db(request, 'create_workspace', user['id'], **await body(request, ('name','instructions'), ('name',)))} @router.patch('/api/workspaces/{wid}') async def update_workspace(wid: str, request: Request, user=Depends(current_user)): return {'workspace': await db(request, 'update_workspace', user['id'], wid, **await body(request, ('revision','name','instructions'), ('revision',)))} @router.delete('/api/workspaces/{wid}') async def delete_workspace(wid: str, request: Request, user=Depends(current_user)): await db(request, 'delete_workspace', user['id'], wid) return {'ok': True} @router.get('/api/preferences') async def preferences(request: Request, user=Depends(current_user)): return {'preferences': await db(request, 'get_preferences', user['id'])} @router.put('/api/preferences') async def set_preferences(request: Request, user=Depends(current_user)): return {'preferences': await db(request, 'set_preferences', user['id'], **await body(request, ('revision','palette','appearance'), ('revision','palette','appearance')))} @router.get('/api/sync') async def sync(request: Request, after: int = 0, user=Depends(current_user)): return await db(request, 'sync', user['id'], after=after) async def room_owner(request, room): account = await asyncio.to_thread(request.app.state.accounts.resolve_room, room) if not account or not account.get('conversation_id'): raise HTTPException(404, 'Room chat unavailable') return account @router.post('/internal/chat/rooms/{room}/messages', dependencies=[Depends(internal_authorized)], status_code=202) async def voice_send(room: str, request: Request): account = await room_owner(request, room) value = await body(request, ('content','idempotency_key','modality'), ('content','idempotency_key','modality')) if value.pop('modality') != 'voice': raise HTTPException(400, 'Invalid modality') return await request.app.state.chat_runs.send(account['id'], account['conversation_id'], None, room=room, **value) @router.get('/internal/chat/rooms/{room}/runs/{rid}/events', dependencies=[Depends(internal_authorized)]) async def voice_events(room: str, rid: str, request: Request, after: int = 0): account = await room_owner(request, room) detail = await db(request, 'get_run', account['id'], rid) if detail['run']['conversation_id'] != account['conversation_id']: raise HTTPException(404, 'Run not found') async def valid(): return await asyncio.to_thread(request.app.state.accounts.resolve_room, room) is not None return await event_response(request, account['id'], rid, valid, after) @router.post('/internal/chat/rooms/{room}/runs/{rid}/cancel', dependencies=[Depends(internal_authorized)]) async def voice_cancel(room: str, rid: str, request: Request): account = await room_owner(request, room) detail = await db(request, 'get_run', account['id'], rid) if detail['run']['conversation_id'] != account['conversation_id']: raise HTTPException(404, 'Run not found') return {'run': await request.app.state.chat_runs.cancel(account['id'], rid)} def create_chat_router(): return router