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