feat: implement durable authenticated chat run coordinator

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

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

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()