537 lines
31 KiB
Python
537 lines
31 KiB
Python
"""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')
|
|
operation = None
|
|
try:
|
|
# Coordinate payload admission with execution and recovery metadata.
|
|
async with self.admission_locks[owner]:
|
|
def prior_exists():
|
|
with self.accounts.lock:
|
|
return self.accounts.db.execute('SELECT 1 FROM chat_idempotency WHERE owner=? AND conversation_id=? AND key=?', (owner, cid, idempotency_key)).fetchone() is not None
|
|
for done_id in [key for key, task in self.tasks.items() if task.done()]:
|
|
self.tasks.pop(done_id, None)
|
|
self.task_owners.pop(done_id, None)
|
|
if not await asyncio.to_thread(prior_exists):
|
|
if len(self.tasks) >= 256 or sum(self.task_owners.get(rid) == owner for rid in self.tasks) >= 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 HTTPException:
|
|
if operation and await asyncio.to_thread(self.metadata, owner, operation['run']['id']) is None:
|
|
rid = operation['run']['id']
|
|
run = await self.set_status(owner, rid, 'failed', 'Chat admission could not be prepared. Start a new request.')
|
|
await self.emit(owner, rid, 'done', {'run': run})
|
|
raise
|
|
except ChatConflict:
|
|
if operation and await asyncio.to_thread(self.metadata, owner, operation['run']['id']) is None:
|
|
run = await self.set_status(owner, operation['run']['id'], 'failed', 'Chat admission could not be prepared. Start a new request.')
|
|
await self.emit(owner, operation['run']['id'], 'done', {'run': run})
|
|
raise HTTPException(409, 'Idempotency or operation conflict') from None
|
|
except KeyError:
|
|
if operation and await asyncio.to_thread(self.metadata, owner, operation['run']['id']) is None:
|
|
try:
|
|
run = await self.set_status(owner, operation['run']['id'], 'failed', 'Chat admission could not be prepared. Start a new request.')
|
|
await self.emit(owner, operation['run']['id'], 'done', {'run': run})
|
|
except KeyError:
|
|
pass # Owner/conversation was concurrently removed.
|
|
raise HTTPException(404, 'Chat resource not found') from None
|
|
except ValueError:
|
|
if operation and await asyncio.to_thread(self.metadata, owner, operation['run']['id']) is None:
|
|
run = await self.set_status(owner, operation['run']['id'], 'failed', 'Chat admission could not be prepared. Start a new request.')
|
|
await self.emit(owner, operation['run']['id'], 'done', {'run': run})
|
|
raise HTTPException(400, 'Invalid chat request') from None
|
|
except Exception:
|
|
if operation and await asyncio.to_thread(self.metadata, owner, operation['run']['id']) is None:
|
|
run = await self.set_status(owner, operation['run']['id'], 'failed', 'Chat admission could not be prepared. Start a new request.')
|
|
await self.emit(owner, operation['run']['id'], 'done', {'run': run})
|
|
raise HTTPException(503, 'Chat admission unavailable') 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']:
|
|
if not metadata.get('submission_attempted'):
|
|
session_state = await self.call('get_session', owner, metadata['conversation_id'])
|
|
if session_state['hermes_seeded']:
|
|
metadata['payload'].pop('conversation_history', None)
|
|
metadata['submission_attempted'] = True
|
|
await asyncio.to_thread(self.metadata, owner, rid, metadata)
|
|
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()
|