diff --git a/control/app.py b/control/app.py index 539eb59..5336fa3 100644 --- a/control/app.py +++ b/control/app.py @@ -206,13 +206,25 @@ class Lifecycle: await asyncio.gather(*tasks, return_exceptions=True) -def create_app(settings=None, provisioner=None, model_catalog=None): +def create_app(settings=None, provisioner=None, model_catalog=None, chat_adapter=None, feature_provider=None): settings = settings or Settings() @asynccontextmanager async def lifespan(app): store = Accounts(settings.db_path) app.state.accounts = store + from .chat_store import ChatStore + from .chat_runs import ChatRuns + app.state.chat_store = await asyncio.to_thread(ChatStore, store) + from .chat_features import FeatureStore, FeatureProvider + from pathlib import Path + feature_root = str(Path(settings.db_path).absolute().parent / 'chat-features') + app.state.feature_store = await asyncio.to_thread(FeatureStore, store, feature_root) + app.state.feature_provider = feature_provider or FeatureProvider(store) + from .chat_features import create_feature_router + app.include_router(create_feature_router(app.state.feature_store, app.state.feature_provider)) + app.state.chat_runs = await asyncio.to_thread(ChatRuns, store, app.state.chat_store, chat_adapter, app.state.feature_store) + app.state.chat_runs.feature_provider = app.state.feature_provider try: if not store.users(): if not settings.admin_password: @@ -231,9 +243,13 @@ def create_app(settings=None, provisioner=None, model_catalog=None): app.state.lifecycle.schedule(user['id']) app.state.dummy_hash = await asyncio.to_thread(password_hash, 'timing-only-password') loop = asyncio.create_task(app.state.lifecycle.loop(settings.reconcile_interval)) + await app.state.chat_runs.start() try: yield finally: + await app.state.chat_runs.stop() + if hasattr(app.state.feature_provider, 'aclose'): + await app.state.feature_provider.aclose() loop.cancel() await asyncio.gather(loop, return_exceptions=True) await app.state.lifecycle.stop() @@ -407,6 +423,7 @@ def create_app(settings=None, provisioner=None, model_catalog=None): if user['status'] != 'ready' or not user['hermes_url'] or not user['hermes_api_key']: raise HTTPException(503, 'User Hermes is not ready') return {'user': public_user(user), 'hermes_url': user['hermes_url'], 'hermes_api_key': user['hermes_api_key'], + 'conversation_id': user.get('conversation_id'), 'model_settings': app.state.accounts.get_model_settings(user['id'])} @app.get('/internal/users/{user_id}', dependencies=[Depends(internal_authorized)]) @@ -430,13 +447,21 @@ def create_app(settings=None, provisioner=None, model_catalog=None): raise HTTPException(400, 'Invalid room grant') if type(ttl) is not int or not 1 <= ttl <= 86400: raise HTTPException(400, 'Invalid room expiry') + conversation_id = body.get('conversation_id') try: - await asyncio.to_thread(app.state.accounts.grant_room, user_id, room, ttl, body.get('session_token')) + if conversation_id is None: + conversation = await asyncio.to_thread(app.state.chat_store.create_conversation, user_id) + conversation_id = conversation['id'] + else: + await asyncio.to_thread(app.state.chat_store.get_conversation, user_id, conversation_id) + await asyncio.to_thread(app.state.accounts.grant_room, user_id, room, ttl, body.get('session_token'), conversation_id) + except KeyError: + raise HTTPException(404, 'Conversation not found') from None except ValueError: raise HTTPException(409, 'Account not ready') from None except sqlite3.IntegrityError: raise HTTPException(409, 'Room already granted') from None - return {'room_name': room, 'user_id': user_id} + return {'room_name': room, 'user_id': user_id, 'conversation_id': conversation_id} @app.get('/internal/rooms/{room_name}', dependencies=[Depends(internal_authorized)]) async def resolve_room(room_name: str): @@ -446,4 +471,8 @@ def create_app(settings=None, provisioner=None, model_catalog=None): async def health(): return {'status': 'ok'} + from .chat_api import create_chat_router + app.include_router(create_chat_router()) + from .chat_exports import create_export_router + app.include_router(create_export_router()) return app diff --git a/control/chat_api.py b/control/chat_api.py new file mode 100644 index 0000000..11ff4c3 --- /dev/null +++ b/control/chat_api.py @@ -0,0 +1,189 @@ +"""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 diff --git a/control/chat_runs.py b/control/chat_runs.py new file mode 100644 index 0000000..e2a17ef --- /dev/null +++ b/control/chat_runs.py @@ -0,0 +1,499 @@ +"""Single account execution coordinator and trusted Hermes runs transport. + +Restart recovery reattaches by upstream run ID; admission retries use the exact +persisted payload and stable idempotency key. No transcript/session replay occurs. +Revocation cancels the initiating session's execution (including room grants). +""" +from __future__ import annotations + +import asyncio +import json +import time +from collections import defaultdict +from urllib.parse import urlsplit + +import httpx +from fastapi import HTTPException + +from .accounts import token_hash + +TERMINAL = {'completed', 'failed', 'cancelled'} + + +class HermesRuns: + """Credentials obtained exclusively from trusted Accounts rows.""" + available = True + approvals = True + + def __init__(self, accounts, client=None): + self.accounts = accounts + self.client = client or httpx.AsyncClient(timeout=httpx.Timeout(30, read=60), follow_redirects=False) + self.owns_client = client is None + + async def connection(self, owner): + user = await asyncio.to_thread(self.accounts.get, owner) + if not user or user['desired'] == 'deleted' or user['status'] != 'ready': + raise RuntimeError('Account unavailable') + root, key = user.get('hermes_url'), user.get('hermes_api_key') + parsed = urlsplit(root or '') + if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or not key or '\n' in key or '\r' in key: + raise RuntimeError('Invalid trusted connection') + root = root.rstrip('/') + if not root.endswith('/v1'): + root += '/v1' + return root, {'Authorization': f'Bearer {key}', 'X-Hermes-Session-Key': f'hope:{owner}'} + + async def request(self, owner, method, path, **kwargs): + root, headers = await self.connection(owner) + headers.update(kwargs.pop('headers', {})) + response = await self.client.request(method, root + path, headers=headers, **kwargs) + response.raise_for_status() + if len(response.content) > 2 * 1024 * 1024: + raise RuntimeError('Response too large') + return response.json() + + async def start(self, owner, rid, payload): + return await self.request(owner, 'POST', '/runs', json=payload, headers={'Idempotency-Key': rid}) + + async def status(self, owner, upstream): + return await self.request(owner, 'GET', f'/runs/{upstream}') + + async def stop(self, owner, upstream): + return await self.request(owner, 'POST', f'/runs/{upstream}/stop') + + async def approve(self, owner, upstream, approval_id, decision): + return await self.request(owner, 'POST', f'/runs/{upstream}/approval', json={'request_id': approval_id, 'choice': 'once' if decision == 'approve' else 'deny'}) + + async def events(self, owner, upstream): + root, headers = await self.connection(owner) + headers['Accept'] = 'text/event-stream' + async with self.client.stream('GET', root + f'/runs/{upstream}/events', headers=headers) as response: + response.raise_for_status() + lines, size = [], 0 + async for line in response.aiter_lines(): + if not line: + if lines: + value = json.loads('\n'.join(lines)) + if isinstance(value, dict): + yield value + lines, size = [], 0 + elif line.startswith('data:'): + size += len(line) + if size > 1024 * 1024: + raise RuntimeError('Event too large') + lines.append(line[5:].lstrip()) + if lines: + value = json.loads('\n'.join(lines)) + if isinstance(value, dict): + yield value + + async def aclose(self): + if self.owns_client: + await self.client.aclose() + + +class ChatRuns: + def __init__(self, accounts, store, adapter=None, feature_store=None): + self.accounts, self.store = accounts, store + self.feature_store = feature_store + self.feature_provider = None + self.adapter = adapter or HermesRuns(accounts) + self.locks = defaultdict(asyncio.Lock) + self.admission_locks = defaultdict(asyncio.Lock) + self.tasks = {} + self.confirmed_stops = set() + self.task_owners = {} + self.closed = False + with accounts.lock, accounts.db: + accounts.db.execute('''CREATE TABLE IF NOT EXISTS chat_transport ( + run_id TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + conversation_id TEXT NOT NULL, metadata TEXT NOT NULL)''') + + def capabilities(self): + enabled = bool(getattr(self.adapter, 'available', True)) + return {'conversations': True, 'workspaces': True, 'preferences': True, + 'search': True, 'branching': True, 'sync': True, 'text_chat': enabled, + 'durable_runs': enabled, 'cancellation': enabled, + 'approvals': enabled and bool(getattr(self.adapter, 'approvals', False)), + 'attachments': self.feature_store is not None, 'artifacts': False, 'tasks': False, + 'memory': False, 'research': False, 'voice_chat': enabled, 'temporary_chat': False} + + def metadata(self, owner, rid, value=None): + with self.accounts.lock, self.accounts.db: + if value is not None: + existing = self.accounts.db.execute('SELECT metadata FROM chat_transport WHERE owner=? AND run_id=?', (owner, rid)).fetchone() + if existing: + prior = json.loads(existing[0]) + for flag in ('stop_requested', 'storage_failed', 'remote_terminal'): + if prior.get(flag): + value[flag] = True + self.accounts.db.execute('INSERT INTO chat_transport VALUES(?,?,?,?) ON CONFLICT(run_id) DO UPDATE SET metadata=excluded.metadata', + (rid, owner, value['conversation_id'], json.dumps(value, ensure_ascii=False))) + row = self.accounts.db.execute('SELECT metadata FROM chat_transport WHERE owner=? AND run_id=?', (owner, rid)).fetchone() + return json.loads(row[0]) if row else None + + def authorized(self, owner, metadata): + with self.accounts.lock: + if metadata.get('room'): + user = self.accounts.resolve_room(metadata['room']) + return bool(user and user['id'] == owner and user.get('conversation_id') == metadata['conversation_id']) + row = self.accounts.db.execute("SELECT 1 FROM sessions s JOIN users u ON u.id=s.user_id WHERE s.hash=? AND s.user_id=? AND s.expires_at>? AND u.desired!='deleted'", (metadata.get('session_hash'), owner, time.time())).fetchone() + return row is not None + + async def call(self, method, *args, **kwargs): + return await asyncio.to_thread(getattr(self.store, method), *args, **kwargs) + + async def send(self, owner, cid, session_token, *, content, idempotency_key, attachment_ids=None, room=None, mode='normal'): + from .chat_store import ChatConflict + if mode not in ('normal', 'research'): + raise HTTPException(400, 'Invalid chat mode') + if mode == 'research': + if self.feature_provider is None or not (await self.feature_provider.feature_capabilities(owner)).get('research'): + raise HTTPException(409, 'Research is unavailable') + parts = None + if attachment_ids: + if self.feature_store is None: + raise HTTPException(400, 'Attachments are not enabled') + vision = False + if self.feature_provider is not None: + qualified = await self.feature_provider.feature_capabilities(owner) + vision = qualified.get('image_input') is True + parts = await self.feature_store.provider_content(owner, attachment_ids, vision=vision) + if not self.capabilities()['text_chat']: + raise HTTPException(503, 'Chat unavailable') + try: + # Coordinate payload admission with execution and recovery metadata. + async with self.admission_locks[owner]: + if sum(not task.done() for task in self.tasks.values()) >= 256 or sum(not task.done() and self.task_owners.get(rid) == owner for rid, task in self.tasks.items()) >= 8: + raise HTTPException(429, 'Chat queue full') + operation = await self.call('send', owner, cid, content, idempotency_key, attachment_ids, mode=mode) + rid = operation['run']['id'] + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if metadata is None: + settings = await asyncio.to_thread(self.accounts.get_model_settings, owner) + session_state = await self.call('get_session', owner, cid) + session = session_state['hermes_session_id'] or 'hope-' + cid + seed_history = None + if not session_state['hermes_seeded']: + def copied_history(): + with self.accounts.lock: + return [{'role': row['role'], 'content': row['content']} for row in self.accounts.db.execute("SELECT role,content FROM chat_messages WHERE owner=? AND conversation_id=? AND run_id IS NULL AND status='completed' ORDER BY position", (owner, cid))] + seed_history = await asyncio.to_thread(copied_history) + if len(seed_history) > 200 or len(json.dumps(seed_history).encode()) > 256000: + raise HTTPException(409, 'Branch context exceeds the supported history limit') + await self.call('set_session', owner, cid, session, False) + user_input = content if not parts else [{'role': 'user', 'content': [{'type': 'text', 'text': content}, *parts]}] + payload = {'input': user_input, 'session_id': session, 'model': settings['model'], + 'provider': 'custom:hope-upstream', + 'model_options': {'reasoning': {'enabled': False} if settings['thinking_level'] == 'none' else {'enabled': True, 'effort': settings['thinking_level']}}, + 'instructions': 'Respond concisely for spoken conversation. Avoid markdown.' if room else 'Respond helpfully to a text chat. Use readable markdown when useful.'} + if mode == 'research': + payload['instructions'] += '\n' + await self.feature_provider.research_instruction(owner) + conversation = (await self.call('get_conversation', owner, cid))['conversation'] + if conversation['workspace_id']: + workspace = await self.call('get_workspace', owner, conversation['workspace_id']) + payload['instructions'] += '\nWorkspace instructions:\n' + workspace['instructions'] + if seed_history: + payload['conversation_history'] = seed_history + metadata = {'conversation_id': cid, 'payload': payload, + 'session_hash': token_hash(session_token) if session_token else None, + 'room': room, 'upstream_id': None, 'assistant_id': None} + await asyncio.to_thread(self.metadata, owner, rid, metadata) + if operation['run']['status'] not in TERMINAL: + self.schedule(owner, rid) + return operation + except ChatConflict: + raise HTTPException(409, 'Idempotency 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 + + def schedule(self, owner, rid): + if not self.closed and (rid not in self.tasks or self.tasks[rid].done()): + self.task_owners[rid] = owner + self.tasks[rid] = asyncio.create_task(self.execute(owner, rid)) + + async def start(self): + def rows(): + with self.accounts.lock: + return [(r['owner'], r['run_id']) for r in self.accounts.db.execute('SELECT owner,run_id FROM chat_transport')] + known = await asyncio.to_thread(rows) + known_ids = {rid for _, rid in known} + for run in await self.call('active_runs'): + if run['id'] not in known_ids: + await self.set_status(run['owner'], run['id'], 'failed', 'Chat admission interrupted. Start a new request.') + await self.emit(run['owner'], run['id'], 'done', {'run': (await self.call('get_run', run['owner'], run['id']))['run']}) + for owner, rid in known: + try: + detail = await self.call('get_run', owner, rid) + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if detail['run']['status'] not in TERMINAL or (metadata.get('stop_requested') and metadata.get('upstream_id') and not metadata.get('remote_terminal')): + self.schedule(owner, rid) + except KeyError: + continue + + async def emit(self, owner, rid, event, data): + try: + await self.call('append_event', owner, rid, event, data) + except ValueError: + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if metadata: + metadata['storage_failed'] = True + metadata['stop_requested'] = True + await asyncio.to_thread(self.metadata, owner, rid, metadata) + raise + + async def set_status(self, owner, rid, status, error=None): + current = (await self.call('get_run', owner, rid))['run'] + if current['status'] in TERMINAL: + return current + run = await self.call('update_run', owner, rid, status, error=error) + await self.emit(owner, rid, 'run', {'run': run}) + return run + + async def finish(self, owner, rid, metadata, status, content=None, error=None): + detail = await self.call('get_run', owner, rid) + if detail['run']['status'] in TERMINAL: + return detail['run'] + if metadata.get('assistant_id'): + message = await self.call('update_message', owner, metadata['assistant_id'], content=content, status=status) + await self.emit(owner, rid, 'message', {'message': message}) + elif content is not None and status == 'completed': + message = await self.call('add_message', owner, metadata['conversation_id'], 'assistant', content, status='completed', run_id=rid) + await self.emit(owner, rid, 'message', {'message': message}) + run = await self.set_status(owner, rid, status, error) + await self.emit(owner, rid, 'done', {'run': run}) + return run + + async def monitor_authorization(self, owner, rid, metadata): + while True: + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if not await asyncio.to_thread(self.authorized, owner, metadata): + await self.cancel(owner, rid) + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if metadata.get('stop_requested') and metadata.get('upstream_id'): + try: + await self.adapter.stop(owner, metadata['upstream_id']) + state = await self.adapter.status(owner, metadata['upstream_id']) + if await self.consume_status(owner, rid, metadata, state): + self.confirmed_stops.add(rid) + task = self.tasks.get(rid) + if task and task is not asyncio.current_task(): + task.cancel() + return + except Exception: + pass + await asyncio.sleep(.25) + + async def execute(self, owner, rid): + watcher = None + try: + async with self.locks[owner]: + metadata = await asyncio.to_thread(self.metadata, owner, rid) + detail = await self.call('get_run', owner, rid) + if not metadata or (detail['run']['status'] in TERMINAL and not (metadata.get('stop_requested') and metadata.get('upstream_id') and not metadata.get('remote_terminal'))): + return + if not await asyncio.to_thread(self.authorized, owner, metadata): + if not metadata.get('upstream_id'): + await self.finish(owner, rid, metadata, 'cancelled') + return + metadata['stop_requested'] = True + await asyncio.to_thread(self.metadata, owner, rid, metadata) + watcher = asyncio.create_task(self.monitor_authorization(owner, rid, metadata)) + if not metadata['upstream_id']: + while True: + try: + started = await self.adapter.start(owner, rid, metadata['payload']) + upstream = started.get('run_id') + if not isinstance(upstream, str) or not upstream or len(upstream) > 200 or not all(c.isalnum() or c in '-_' for c in upstream): + raise RuntimeError('Invalid upstream run') + metadata['upstream_id'] = upstream + await asyncio.to_thread(self.metadata, owner, rid, metadata) + session_id = metadata['payload'].get('session_id') + if session_id: + await self.call('set_session', owner, metadata['conversation_id'], session_id, True) + break + except httpx.HTTPStatusError as exc: + if exc.response.status_code in (400, 401, 403, 404, 409, 422): + raise RuntimeError('Chat admission rejected') from None + await asyncio.sleep(1) + except (httpx.HTTPError, RuntimeError): + # Unknown outcome: retain account slot, retry exact + # durable payload/key, never create a second action. + await asyncio.sleep(1) + current = await self.call('get_run', owner, rid) + if current['run']['status'] in TERMINAL or not await asyncio.to_thread(self.authorized, owner, metadata): + metadata['stop_requested'] = True + await asyncio.to_thread(self.metadata, owner, rid, metadata) + try: + await self.adapter.stop(owner, metadata['upstream_id']) + except Exception: + pass + # Keep ownership of the execution slot until remote terminal. + else: + await self.set_status(owner, rid, 'running') + # Reconnect only this single queue-consuming relay; never expose upstream SSE. + while True: + metadata = await asyncio.to_thread(self.metadata, owner, rid) + try: + if metadata.get('stop_requested'): + await self.adapter.stop(owner, metadata['upstream_id']) + state = await self.adapter.status(owner, metadata['upstream_id']) + if await self.consume_status(owner, rid, metadata, state): + return + except Exception: + await asyncio.sleep(1) + continue + try: + async for event in self.adapter.events(owner, metadata['upstream_id']): + current = await self.call('get_run', owner, rid) + if current['run']['status'] in TERMINAL: + if await self.consume_status(owner, rid, metadata, {'status': event.get('event', '').removeprefix('run.'), 'output': event.get('output')}): + return + continue + latest = await asyncio.to_thread(self.metadata, owner, rid) + if latest.get('storage_failed'): + metadata = latest + continue + if await self.consume_event(owner, rid, metadata, event): + return + state = await self.adapter.status(owner, metadata['upstream_id']) + if await self.consume_status(owner, rid, metadata, state): + return + except (httpx.HTTPError, ValueError): + pass + await asyncio.sleep(1) + except asyncio.CancelledError: + if rid in self.confirmed_stops: + self.confirmed_stops.discard(rid) + return + # Shutdown detaches; durable metadata is recovered on startup. + raise + except Exception: + try: + metadata = await asyncio.to_thread(self.metadata, owner, rid) + await self.finish(owner, rid, metadata or {}, 'failed', error='Chat execution unavailable. Try again.') + except KeyError: + pass + finally: + if watcher: + watcher.cancel() + await asyncio.gather(watcher, return_exceptions=True) + + async def consume_status(self, owner, rid, metadata, state): + status = state.get('status') + if status in ('completed', 'failed', 'cancelled', 'interrupted'): + metadata['remote_terminal'] = True + await asyncio.to_thread(self.metadata, owner, rid, metadata) + if (await self.call('get_run', owner, rid))['run']['status'] in TERMINAL: + return True + if status in ('completed', 'failed', 'cancelled', 'interrupted') and metadata.get('storage_failed'): + if metadata.get('assistant_id'): + await self.call('update_message', owner, metadata['assistant_id'], status='failed') + run = await self.set_status(owner, rid, 'failed', 'Chat storage limit reached.') + await self.emit(owner, rid, 'done', {'run': run}) + return True + if status == 'waiting_for_approval': + approval = state.get('approval') + if isinstance(approval, dict) and isinstance(approval.get('request_id'), str): + await self.consume_event(owner, rid, metadata, {'event': 'approval.request', 'request_id': approval['request_id']}) + if status == 'completed': + output = state.get('output') + if not isinstance(output, str) or not output.strip(): + raise RuntimeError('Invalid final output') + await self.finish(owner, rid, metadata, 'completed', content=output[:32000]) + return True + if status in ('failed', 'cancelled', 'interrupted'): + await self.finish(owner, rid, metadata, 'cancelled' if status == 'cancelled' else 'failed', error=None if status == 'cancelled' else 'Chat execution failed.') + return True + return False + + async def consume_event(self, owner, rid, metadata, event): + name = event.get('event') + if name == 'message.delta': + delta = event.get('delta') + if isinstance(delta, str) and delta: + if not metadata.get('assistant_id'): + message = await self.call('add_message', owner, metadata['conversation_id'], 'assistant', '', status='streaming', run_id=rid) + metadata['assistant_id'] = message['id'] + metadata['content'] = '' + metadata['content'] = (metadata.get('content', '') + delta)[:32000] + now = time.monotonic() + if len(metadata['content']) - metadata.get('emitted_length', 0) >= 256 or now - metadata.get('emitted_at', 0) >= .1: + metadata['emitted_at'] = now + metadata['emitted_length'] = len(metadata['content']) + message = await self.call('update_message', owner, metadata['assistant_id'], content=metadata['content'], status='streaming') + await asyncio.to_thread(self.metadata, owner, rid, metadata) + await self.emit(owner, rid, 'message', {'message': message}) + elif name == 'approval.request': + approval_id = event.get('request_id') + if isinstance(approval_id, str) and 0 < len(approval_id) <= 200: + metadata['approval_id'] = approval_id + await asyncio.to_thread(self.metadata, owner, rid, metadata) + await self.set_status(owner, rid, 'awaiting_approval') + await self.emit(owner, rid, 'approval', {'approval_id': approval_id, 'prompt': 'Hermes requests permission to execute an action.'}) + elif name in ('tool.started', 'tool.completed', 'run.started'): + await self.emit(owner, rid, 'progress', {'status': 'working', 'message': 'Working on your request.'}) + elif name in ('run.completed', 'run.failed', 'run.cancelled'): + return await self.consume_status(owner, rid, metadata, {'status': name.split('.')[1], 'output': event.get('output')}) + # Unknown/reasoning/raw tools deliberately never enter the product stream. + return False + + async def cancel(self, owner, rid): + try: + detail = await self.call('get_run', owner, rid) + if detail['run']['status'] in TERMINAL: + return detail['run'] + metadata = await asyncio.to_thread(self.metadata, owner, rid) + if metadata: + metadata['stop_requested'] = True + await asyncio.to_thread(self.metadata, owner, rid, metadata) + if metadata.get('upstream_id'): + try: + await self.adapter.stop(owner, metadata['upstream_id']) + except Exception: + pass # Durable intent remains; never claim remote success. + return detail['run'] + # Queued/not-yet-admitted runs are safe to cancel locally. An admission + # in flight rechecks this terminal projection and stops its returned ID. + return await self.finish(owner, rid, metadata or {}, 'cancelled') + except KeyError: + raise HTTPException(404, 'Run not found') from None + + async def cancel_conversation(self, owner, cid): + from .chat_store import ChatConflict + detail = await self.call('get_conversation', owner, cid) + pending = False + for run in detail['runs']: + if run['status'] not in TERMINAL: + result = await self.cancel(owner, run['id']) + pending |= result['status'] not in TERMINAL + if pending: + raise ChatConflict('Remote cancellation is pending') + + async def approval(self, owner, rid, approval_id, decision): + if decision not in ('approve', 'deny') or not isinstance(approval_id, str) or not approval_id or len(approval_id) > 200: + raise HTTPException(400, 'Invalid approval') + if not self.capabilities()['approvals']: + raise HTTPException(409, 'Approvals unavailable') + metadata = await asyncio.to_thread(self.metadata, owner, rid) + try: + detail = await self.call('get_run', owner, rid) + if not metadata or detail['run']['status'] != 'awaiting_approval' or metadata.get('approval_id') != approval_id: + raise HTTPException(409, 'Approval no longer pending') + await self.adapter.approve(owner, metadata['upstream_id'], approval_id, decision) + return await self.set_status(owner, rid, 'running') + except KeyError: + raise HTTPException(404, 'Run not found') from None + except httpx.HTTPError: + raise HTTPException(503, 'Approval unavailable') from None + + async def stop(self): + self.closed = True + tasks = list(self.tasks.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + if hasattr(self.adapter, 'aclose'): + await self.adapter.aclose() diff --git a/tests/unit/test_chat_api.py b/tests/unit/test_chat_api.py new file mode 100644 index 0000000..0cf8550 --- /dev/null +++ b/tests/unit/test_chat_api.py @@ -0,0 +1,100 @@ +"""Offline durable API qualification using injected Hermes runs adapter.""" +import asyncio +import time +import pytest +from fastapi.testclient import TestClient +from control.app import Settings, create_app + +PASSWORD = 'offline-chat-password' + +class Provisioner: + async def ensure(self, user): + return {'container_name': user['container_name'], 'hermes_url': 'http://127.0.0.1:12345', 'hermes_api_key': 'secret-never-public'} + async def ready(self, user): + return True + async def remove(self, user): + pass + +class FakeRuns: + available = True + approvals = False + def __init__(self): + self.started = [] + self.stopped = [] + async def start(self, owner, rid, payload): + self.started.append((owner, rid, payload)) + return {'run_id': 'upstream-' + rid, 'status': 'started'} + async def status(self, owner, upstream): + return {'status': 'running'} + async def events(self, owner, upstream): + yield {'event': 'reasoning.delta', 'delta': 'hidden-secret'} + yield {'event': 'tool.completed', 'result': 'tool-secret', 'error': 'credential'} + yield {'event': 'message.delta', 'delta': 'Hello'} + yield {'event': 'run.completed', 'output': 'Hello world'} + async def stop(self, owner, upstream): + self.stopped.append(upstream) + +@pytest.fixture +def env(tmp_path): + adapter = FakeRuns() + app = create_app(Settings(db_path=str(tmp_path/'chat.sqlite'), admin_password=PASSWORD, allowed_origins=('https://testserver',), reconcile_interval=3600), Provisioner(), chat_adapter=adapter) + with TestClient(app, base_url='https://testserver') as client: + result = client.post('/api/auth/login', headers={'Origin':'https://testserver'}, json={'username':'admin','password':PASSWORD}) + headers = {'Authorization':'Bearer '+result.json()['token']} + yield client, app, adapter, headers + +def wait_run(client, headers, rid): + deadline = time.monotonic()+3 + while time.monotonic() < deadline: + value = client.get('/api/runs/'+rid, headers=headers).json() + if value['run']['status'] in ('completed','failed','cancelled'): + return value + raise AssertionError('Run did not finish') + +def test_metadata_crud_revision_and_sync(env): + client, app, adapter, headers = env + client.cookies.clear() + assert client.get('/api/conversations').status_code == 401 + workspace = client.post('/api/workspaces', headers=headers, json={'name':'Work'}).json()['workspace'] + conversation = client.post('/api/conversations', headers=headers, json={'title':'Durable','workspace_id':workspace['id']}).json()['conversation'] + cid = conversation['id'] + patch = {'revision':conversation['revision'],'pinned':True} + assert client.patch('/api/conversations/'+cid, headers=headers, json=patch).status_code == 200 + assert client.patch('/api/conversations/'+cid, headers=headers, json=patch).status_code == 409 + assert client.delete('/api/workspaces/'+workspace['id'], headers=headers).status_code == 200 + assert client.get('/api/conversations/'+cid, headers=headers).json()['conversation']['workspace_id'] is None + assert client.delete('/api/conversations/'+cid, headers=headers).status_code == 200 + assert client.get('/api/conversations/'+cid, headers=headers).status_code == 404 + assert client.get('/api/sync', headers=headers).json()['changes'] + +def test_idempotent_send_final_output_and_redacted_replay(env): + client, app, adapter, headers = env + cid = client.post('/api/conversations', headers=headers, json={}).json()['conversation']['id'] + url = '/api/conversations/'+cid+'/messages' + value = {'content':'Hi','idempotency_key':'same-key'} + response = client.post(url, headers=headers, json=value) + assert response.status_code == 202, response.text + rid = response.json()['run']['id'] + final = wait_run(client, headers, rid) + assert final['run']['status'] == 'completed', final + assert any(m['content']=='Hello world' for m in final['messages']) + repeat = client.post(url, headers=headers, json=value) + assert repeat.json()['run']['id'] == rid + assert client.post(url, headers=headers, json={**value,'content':'changed'}).status_code == 409 + assert len(adapter.started) == 1 + payload = adapter.started[0][2] + assert payload['input'] == 'Hi' + assert 'conversation_history' not in payload + stream = client.get('/api/runs/'+rid+'/events', headers=headers) + assert stream.status_code == 200 + assert 'event: done' in stream.text + assert 'hidden-secret' not in stream.text and 'tool-secret' not in stream.text + assert 'secret-never-public' not in stream.text + +def test_body_validation_and_honest_capabilities(env): + client, app, adapter, headers = env + caps = client.get('/api/capabilities', headers=headers).json() + assert caps['features']['text_chat'] is True + assert caps['features']['attachments'] is True + assert client.post('/api/conversations', headers=headers, json={'user_id':'fabricated'}).status_code == 400 + assert client.put('/api/preferences', headers=headers, json={'revision':True,'palette':'opal','appearance':'dark'}).status_code == 400 diff --git a/tests/unit/test_chat_runs.py b/tests/unit/test_chat_runs.py new file mode 100644 index 0000000..1fb0f22 --- /dev/null +++ b/tests/unit/test_chat_runs.py @@ -0,0 +1,52 @@ +"""Trusted transport protocol tests; HTTP mocked with no network.""" +import json +import asyncio +import functools + +def async_test(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + return asyncio.run(func(*args, **kwargs)) + return wrapper +import pytest +import httpx +from control.accounts import Accounts +from control.chat_runs import HermesRuns + +@async_test +async def test_native_runs_transport_exact_idempotency_and_session(tmp_path): + accounts = Accounts(str(tmp_path/'accounts.db')) + user = accounts.create('test-user','Test','admin','offline-password') + accounts.update(user['id'], status='ready', hermes_url='http://127.0.0.1:20000', hermes_api_key='private-key') + seen = [] + def handler(request): + seen.append(request) + if request.url.path.endswith('/events'): + return httpx.Response(200, text='data: {"event":"message.delta","delta":"Hi"}\n\ndata: {"event":"run.completed","output":"Hi"}\n\n') + return httpx.Response(202, json={'run_id':'native-run','status':'started'}) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + adapter = HermesRuns(accounts, client) + payload = {'input':'Hi','session_id':'stable-session'} + assert (await adapter.start(user['id'],'hope-run',payload))['run_id'] == 'native-run' + assert seen[0].url.path == '/v1/runs' + assert seen[0].headers['authorization'] == 'Bearer private-key' + assert seen[0].headers['idempotency-key'] == 'hope-run' + assert json.loads(seen[0].content) == payload + events = [e async for e in adapter.events(user['id'],'native-run')] + assert [e['event'] for e in events] == ['message.delta','run.completed'] + await adapter.stop(user['id'],'native-run') + assert seen[-1].url.path == '/v1/runs/native-run/stop' + await adapter.approve(user['id'],'native-run','approval-1','approve') + assert json.loads(seen[-1].content) == {'request_id':'approval-1','choice':'once'} + accounts.close() + +@async_test +async def test_transport_rejects_deleted_account_and_bad_trusted_url(tmp_path): + accounts = Accounts(str(tmp_path/'accounts.db')) + user = accounts.create('test-user','Test','admin','offline-password') + accounts.update(user['id'], status='ready', hermes_url='http://evil:password@localhost:2', hermes_api_key='private-key') + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: pytest.fail('Network requested'))) as client: + adapter = HermesRuns(accounts, client) + with pytest.raises(RuntimeError): + await adapter.start(user['id'],'rid',{'input':'Hi'}) + accounts.close() diff --git a/tests/unit/test_chat_runs_limits.py b/tests/unit/test_chat_runs_limits.py new file mode 100644 index 0000000..bb55bb7 --- /dev/null +++ b/tests/unit/test_chat_runs_limits.py @@ -0,0 +1,33 @@ +"""Offline storage-limit stop intent regression.""" +import asyncio +from control.accounts import Accounts, token_hash +from control.chat_store import ChatStore +from control.chat_runs import ChatRuns + +def test_event_capacity_failure_persists_stop_and_reserved_terminal(tmp_path): + async def scenario(): + accounts = Accounts(str(tmp_path/'limit.db')) + owner = accounts.create('test','Test','admin','offline-test-password') + store = ChatStore(accounts) + cid = store.create_conversation(owner['id'])['id'] + rid = store.send(owner['id'],cid,'Hi','key')['run']['id'] + store.update_run(owner['id'],rid,'running') + runs = ChatRuns(accounts,store,adapter=object()) + runs.metadata(owner['id'],rid,{'conversation_id':cid,'payload':{},'upstream_id':'upstream','assistant_id':None}) + original = store.append_event + def limited(owner,rid,event,data): + if event == 'progress': + raise ValueError('storage limit') + return original(owner,rid,event,data) + store.append_event = limited + try: + await runs.emit(owner['id'],rid,'progress',{'status':'working'}) + except ValueError: + pass + metadata = runs.metadata(owner['id'],rid) + assert metadata['stop_requested'] and metadata['storage_failed'] + assert await runs.consume_status(owner['id'],rid,metadata,{'status':'cancelled'}) + assert store.get_run(owner['id'],rid)['run']['status'] == 'failed' + assert store.events(owner['id'],rid)[-1]['event'] == 'done' + accounts.close() + asyncio.run(scenario())