"""Authenticated advanced chat surfaces. Hope stores metadata; Hermes executes.

Construct FeatureStore with the application's Accounts and include
create_feature_router(store, provider). No provider operation is enabled merely
because a method exists: the adapter must explicitly qualify that capability.
All persistence and extraction is offloaded by the async public methods.
"""
from __future__ import annotations

import asyncio
import base64
import importlib.util
import hashlib
import json
import os
import re
import stat
import subprocess
import sys
import threading
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import quote

from fastapi import APIRouter, Depends, HTTPException, Request, Response

from .app import current_user

MIMES = ('text/plain', 'text/markdown', 'application/pdf', 'image/png', 'image/jpeg', 'image/webp')
ID = re.compile(r'[a-f0-9]{32}\Z')


@dataclass(frozen=True)
class FeatureLimits:
    upload_bytes: int = 8 * 1024 * 1024
    account_bytes: int = 64 * 1024 * 1024
    disk_bytes: int = 512 * 1024 * 1024
    extracted_chars: int = 100_000
    artifact_bytes: int = 512 * 1024
    artifact_count: int = 200
    account_artifact_bytes: int = 32 * 1024 * 1024
    global_artifact_bytes: int = 256 * 1024 * 1024
    artifact_versions: int = 100
    attachments_per_message: int = 8


class FeatureStore:
    def __init__(self, accounts, root_path, limits=None):
        self.accounts = accounts
        self.db = accounts.db
        self.lock = accounts.lock
        self.limits = limits or FeatureLimits()
        self.upload_admission = threading.BoundedSemaphore(2)
        raw = Path(root_path).absolute()
        if raw.is_symlink():
            raise ValueError('Private storage cannot be a symlink')
        raw.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.root = raw.resolve()
        os.chmod(self.root, 0o700)
        with self.lock, self.db:
            self.db.executescript('''
                CREATE TABLE IF NOT EXISTS hope_attachments (
                  id TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
                  filename TEXT NOT NULL, mime TEXT NOT NULL, size INTEGER NOT NULL,
                  extracted TEXT, extraction_status TEXT NOT NULL, created_at REAL NOT NULL,
                  deleted_at REAL);
                CREATE INDEX IF NOT EXISTS hope_attachments_owner ON hope_attachments(owner);
                CREATE TABLE IF NOT EXISTS hope_artifacts (
                  id TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
                  title TEXT NOT NULL, kind TEXT NOT NULL, revision INTEGER NOT NULL,
                  created_at REAL NOT NULL, updated_at REAL NOT NULL, deleted_at REAL);
                CREATE TABLE IF NOT EXISTS hope_artifact_versions (
                  artifact_id TEXT NOT NULL REFERENCES hope_artifacts(id) ON DELETE CASCADE,
                  revision INTEGER NOT NULL, content TEXT NOT NULL, created_at REAL NOT NULL,
                  PRIMARY KEY(artifact_id, revision));
                CREATE TABLE IF NOT EXISTS hope_feature_changes (
                  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
                  owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
                  kind TEXT NOT NULL, object_id TEXT NOT NULL, deleted INTEGER NOT NULL,
                  created_at REAL NOT NULL);
            ''')

    def _owner(self, owner):
        row = self.db.execute("SELECT id FROM users WHERE id=? AND desired!='deleted'", (owner,)).fetchone()
        if row is None:
            raise HTTPException(401, 'Account unavailable')

    def _path(self, identifier):
        if not isinstance(identifier, str) or not ID.fullmatch(identifier):
            raise HTTPException(404, 'Attachment not found')
        path = self.root / identifier
        if path.parent.resolve() != self.root or path.is_symlink():
            raise HTTPException(404, 'Attachment not found')
        return path

    def _change(self, owner, kind, identifier, deleted=False):
        self.db.execute('INSERT INTO hope_feature_changes(owner,kind,object_id,deleted,created_at) VALUES(?,?,?,?,?)',
                        (owner, kind, identifier, int(deleted), time.time()))

    @staticmethod
    def _attachment(row):
        return {k: row[k] for k in ('id', 'filename', 'mime', 'size', 'extraction_status', 'created_at')}

    def _extract(self, mime, data):
        if mime.startswith('text/'):
            try:
                text = data.decode('utf-8')
            except UnicodeDecodeError:
                raise HTTPException(415, 'Text must be UTF-8') from None
            if '\x00' in text:
                raise HTTPException(415, 'Binary text is not supported')
            return text[:self.limits.extracted_chars], 'truncated' if len(text) > self.limits.extracted_chars else 'ready'
        signatures = {'application/pdf': b'%PDF-', 'image/png': b'\x89PNG\r\n\x1a\n', 'image/jpeg': b'\xff\xd8\xff'}
        if mime in signatures and not data.startswith(signatures[mime]):
            raise HTTPException(415, 'File does not match MIME type')
        if mime == 'image/webp' and not (data.startswith(b'RIFF') and data[8:12] == b'WEBP'):
            raise HTTPException(415, 'File does not match MIME type')
        if mime == 'application/pdf':
            if importlib.util.find_spec('pypdf') is None:
                return None, 'unavailable'
            # Untrusted compressed PDF structures must not run in the control
            # process. Child has CPU/address-space/output limits and a wall timeout.
            script = '''import resource,sys,io,json
resource.setrlimit(resource.RLIMIT_AS,(384*1024*1024,384*1024*1024))
resource.setrlimit(resource.RLIMIT_CPU,(4,4))
from pypdf import PdfReader
limit=int(sys.argv[1]); reader=PdfReader(io.BytesIO(sys.stdin.buffer.read()))
if reader.is_encrypted:
 print(json.dumps([None,'encrypted'])); sys.exit(0)
parts=[]; size=0; truncated=False
for page in reader.pages:
 text=page.extract_text() or ''
 parts.append(text[:max(0,limit-size)])
 size+=len(text)
 if size>=limit:
  truncated=True; break
print(json.dumps(['\\n'.join(parts)[:limit],'truncated' if truncated else 'ready']))
'''
            try:
                result = subprocess.run([sys.executable, '-I', '-c', script, str(self.limits.extracted_chars)],
                                        input=data, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=6, check=True)
                text, status = json.loads(result.stdout)
                return text, status
            except Exception:
                return None, 'failed'
        return None, 'image'

    def _upload(self, owner, filename, mime, data):
        if mime not in MIMES:
            raise HTTPException(415, 'Unsupported MIME type')
        if not isinstance(data, bytes) or not 0 < len(data) <= self.limits.upload_bytes:
            raise HTTPException(413, 'Upload byte limit exceeded')
        if not isinstance(filename, str) or not 1 <= len(filename) <= 160 or any(c in filename for c in ('/', '\\', '\x00', '\r', '\n')) or any(ord(c) < 32 for c in filename):
            raise HTTPException(422, 'Invalid filename')
        # Reserve parser concurrency before quota/extraction work; fail fast.
        if not self.upload_admission.acquire(blocking=False):
            raise HTTPException(429, 'Upload processing busy')
        try:
            with self.lock:
                self._owner(owner)
                used = self.db.execute('SELECT COALESCE(SUM(size),0) FROM hope_attachments WHERE owner=? AND deleted_at IS NULL', (owner,)).fetchone()[0]
                disk = sum(p.stat(follow_symlinks=False).st_size for p in self.root.iterdir() if p.is_file() and not p.is_symlink())
                if used + len(data) > self.limits.account_bytes or disk + len(data) > self.limits.disk_bytes:
                    raise HTTPException(413, 'Private storage quota exceeded')
            text, status = self._extract(mime, data)
        finally:
            self.upload_admission.release()
        identifier = uuid.uuid4().hex
        path = self._path(identifier)
        with self.lock, self.db:
            self._owner(owner)
            used = self.db.execute('SELECT COALESCE(SUM(size),0) FROM hope_attachments WHERE owner=? AND deleted_at IS NULL', (owner,)).fetchone()[0]
            # Include retained/orphaned files after account deletion in global quota.
            disk = sum(p.stat(follow_symlinks=False).st_size for p in self.root.iterdir() if p.is_file() and not p.is_symlink())
            if used + len(data) > self.limits.account_bytes or disk + len(data) > self.limits.disk_bytes:
                raise HTTPException(413, 'Private storage quota exceeded')
            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, 'O_NOFOLLOW', 0), 0o600)
            try:
                with os.fdopen(fd, 'wb') as stream:
                    stream.write(data)
                    stream.flush()
                    os.fsync(stream.fileno())
                self.db.execute('INSERT INTO hope_attachments VALUES(?,?,?,?,?,?,?,?,NULL)',
                                (identifier, owner, filename, mime, len(data), text, status, time.time()))
                self._change(owner, 'attachment', identifier)
            except BaseException:
                # _path has verified the resolved parent and server-generated ID.
                path.unlink(missing_ok=True)
                raise
            return self._attachment(self._get_attachment(owner, identifier))

    def _get_attachment(self, owner, identifier):
        self._owner(owner)
        row = self.db.execute('SELECT * FROM hope_attachments WHERE owner=? AND id=? AND deleted_at IS NULL', (owner, identifier)).fetchone()
        if row is None:
            raise HTTPException(404, 'Attachment not found')
        return row

    def _read_attachment(self, owner, identifier):
        with self.lock:
            row = self._get_attachment(owner, identifier)
            path = self._path(identifier)
            try:
                fd = os.open(path, os.O_RDONLY | getattr(os, 'O_NOFOLLOW', 0))
                with os.fdopen(fd, 'rb') as stream:
                    info = os.fstat(stream.fileno())
                    if not stat.S_ISREG(info.st_mode) or info.st_size != row['size']:
                        raise HTTPException(404, 'Attachment unavailable')
                    data = stream.read(self.limits.upload_bytes + 1)
            except OSError:
                raise HTTPException(404, 'Attachment unavailable') from None
            return self._attachment(row), data

    def _delete_attachment(self, owner, identifier):
        with self.lock, self.db:
            self._get_attachment(owner, identifier)
            path = self._path(identifier)
            path.unlink(missing_ok=True)
            self.db.execute('UPDATE hope_attachments SET deleted_at=? WHERE owner=? AND id=?', (time.time(), owner, identifier))
            self._change(owner, 'attachment', identifier, True)
        return {'ok': True}

    async def upload(self, owner, filename, mime, data):
        return await asyncio.to_thread(self._upload, owner, filename, mime, data)

    async def read_attachment(self, owner, identifier):
        return await asyncio.to_thread(self._read_attachment, owner, identifier)

    async def delete_attachment(self, owner, identifier):
        return await asyncio.to_thread(self._delete_attachment, owner, identifier)

    async def provider_content(self, owner, identifiers, *, vision=False):
        def build():
            if not isinstance(identifiers, list) or len(identifiers) > self.limits.attachments_per_message or any(not isinstance(identifier, str) or not ID.fullmatch(identifier) for identifier in identifiers) or len(set(identifiers)) != len(identifiers):
                raise HTTPException(422, 'Invalid attachment selection')
            parts = []
            with self.lock:
                for identifier in identifiers:
                    row = self._get_attachment(owner, identifier)
                    if row['mime'].startswith('image/'):
                        if not vision:
                            raise HTTPException(409, 'Selected model is not qualified for image input')
                        _, data = self._read_attachment(owner, identifier)
                        parts.append({'type': 'image_url', 'image_url': {'url': 'data:' + row['mime'] + ';base64,' + base64.b64encode(data).decode('ascii')}})
                    elif row['extraction_status'] in ('ready', 'truncated'):
                        parts.append({'type': 'text', 'text': 'Attachment: ' + row['filename'] + '\n' + (row['extracted'] or '')})
                    else:
                        raise HTTPException(409, 'Attachment text extraction unavailable')
            return parts
        return await asyncio.to_thread(build)

    @staticmethod
    def _validate_artifact(title, kind, content, limit):
        if not isinstance(title, str) or not 1 <= len(title.strip()) <= 160:
            raise HTTPException(422, 'Invalid artifact title')
        if kind not in ('text', 'markdown', 'code'):
            raise HTTPException(422, 'Unsupported artifact kind')
        if not isinstance(content, str) or len(content.encode('utf-8')) > limit:
            raise HTTPException(413, 'Artifact byte limit exceeded')

    def _get_artifact(self, owner, identifier, revision=None):
        self._owner(owner)
        row = self.db.execute('SELECT * FROM hope_artifacts WHERE owner=? AND id=? AND deleted_at IS NULL', (owner, identifier)).fetchone()
        if row is None:
            raise HTTPException(404, 'Artifact not found')
        result = {k: row[k] for k in ('id', 'title', 'kind', 'revision', 'created_at', 'updated_at')}
        version = self.db.execute('SELECT content FROM hope_artifact_versions WHERE artifact_id=? AND revision=?', (identifier, revision if revision is not None else row['revision'])).fetchone()
        if version is None:
            raise HTTPException(404, 'Artifact version not found')
        result['content'] = version['content']
        if revision is not None:
            result['revision'] = revision
        return result

    def _save_artifact(self, owner, title, kind, content, identifier=None, revision=None):
        self._validate_artifact(title, kind, content, self.limits.artifact_bytes)
        now = time.time()
        with self.lock, self.db:
            self._owner(owner)
            used = self.db.execute('SELECT COALESCE(SUM(length(CAST(v.content AS BLOB))),0) FROM hope_artifact_versions v JOIN hope_artifacts a ON a.id=v.artifact_id WHERE a.owner=?', (owner,)).fetchone()[0]
            total = self.db.execute('SELECT COALESCE(SUM(length(CAST(content AS BLOB))),0) FROM hope_artifact_versions').fetchone()[0]
            incoming = len(content.encode('utf-8'))
            if used + incoming > self.limits.account_artifact_bytes or total + incoming > self.limits.global_artifact_bytes:
                raise HTTPException(413, 'Artifact storage quota exceeded')
            if identifier is None:
                count = self.db.execute('SELECT COUNT(*) FROM hope_artifacts WHERE owner=?', (owner,)).fetchone()[0]
                if count >= self.limits.artifact_count:
                    raise HTTPException(413, 'Artifact quota exceeded')
                identifier = uuid.uuid4().hex
                self.db.execute('INSERT INTO hope_artifacts VALUES(?,?,?,?,1,?,?,NULL)', (identifier, owner, title.strip(), kind, now, now))
                version = 1
            else:
                old = self._get_artifact(owner, identifier)
                if type(revision) is not int or revision != old['revision']:
                    raise HTTPException(409, 'Artifact revision conflict')
                if revision >= self.limits.artifact_versions:
                    raise HTTPException(413, 'Artifact version quota exceeded')
                version = revision + 1
                self.db.execute('UPDATE hope_artifacts SET title=?,kind=?,revision=?,updated_at=? WHERE owner=? AND id=?',
                                (title.strip(), kind, version, now, owner, identifier))
            self.db.execute('INSERT INTO hope_artifact_versions VALUES(?,?,?,?)', (identifier, version, content, now))
            self._change(owner, 'artifact', identifier)
            return self._get_artifact(owner, identifier)

    async def save_artifact(self, owner, title, kind, content, identifier=None, revision=None):
        return await asyncio.to_thread(self._save_artifact, owner, title, kind, content, identifier, revision)

    async def get_artifact(self, owner, identifier, revision=None):
        def get():
            with self.lock:
                return self._get_artifact(owner, identifier, revision)
        return await asyncio.to_thread(get)

    async def delete_artifact(self, owner, identifier):
        def remove():
            with self.lock, self.db:
                self._get_artifact(owner, identifier)
                self.db.execute('UPDATE hope_artifacts SET deleted_at=?,updated_at=? WHERE owner=? AND id=?', (time.time(), time.time(), owner, identifier))
                self._change(owner, 'artifact', identifier, True)
            return {'ok': True}
        return await asyncio.to_thread(remove)

    async def list_objects(self, owner, kind):
        def collect():
            with self.lock:
                self._owner(owner)
                if kind == 'attachments':
                    return [self._attachment(r) for r in self.db.execute('SELECT * FROM hope_attachments WHERE owner=? AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 200', (owner,))]
                return [dict(row) for row in self.db.execute('SELECT id,title,kind,revision,created_at,updated_at FROM hope_artifacts WHERE owner=? AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT 200', (owner,))]
        return await asyncio.to_thread(collect)

    async def changes(self, owner, after=0):
        def collect():
            with self.lock:
                self._owner(owner)
                rows = self.db.execute('SELECT * FROM hope_feature_changes WHERE owner=? AND sequence>? ORDER BY sequence LIMIT 200', (owner, after)).fetchall()
                return {'changes': [{'sequence': r['sequence'], 'kind': r['kind'], 'id': r['object_id'], 'deleted': bool(r['deleted']), 'created_at': r['created_at']} for r in rows], 'cursor': rows[-1]['sequence'] if rows else after}
        return await asyncio.to_thread(collect)


class FeatureProvider:
    """Fixed-route private gateway adapter; no client-selected URL or tool loop."""
    def __init__(self, accounts, http_client=None, state_dir=None):
        import httpx
        self.accounts = accounts
        self.http = http_client or httpx.AsyncClient(timeout=30, follow_redirects=False)
        self.owns_http = http_client is None
        self.state_dir = Path(state_dir or os.getenv('HOPE_HERMES_DATA_DIR', '.runtime/hermes')).absolute()

    async def aclose(self):
        if self.owns_http:
            await self.http.aclose()

    async def _account(self, owner):
        user = await asyncio.to_thread(self.accounts.get, owner)
        if not user or user['desired'] == 'deleted':
            raise HTTPException(401, 'Account unavailable')
        if user['status'] != 'ready' or not user.get('hermes_url') or not user.get('hermes_api_key'):
            raise HTTPException(503, 'Hermes not ready')
        return user

    async def toolset_status(self, owner):
        user = await self._account(owner)
        try:
            response = await self.http.get(user['hermes_url'].rstrip('/') + '/v1/toolsets', headers={'Authorization': 'Bearer ' + user['hermes_api_key']})
            if response.status_code != 200:
                raise ValueError()
            body = response.json()
            if not isinstance(body, dict) or not isinstance(body.get('data'), list):
                raise ValueError()
            return [{'name': row['name'], 'enabled': row.get('enabled') is True, 'configured': row.get('configured') is True,
                     'tools': [tool for tool in row.get('tools', []) if isinstance(tool, str)][:200]}
                    for row in body['data'][:100] if isinstance(row, dict) and isinstance(row.get('name'), str) and isinstance(row.get('tools'), list)]
        except Exception:
            raise HTTPException(503, 'Hermes tool capability status unavailable') from None

    async def research_status(self, owner):
        toolsets = await self.toolset_status(owner)
        available = any(row['name'] == 'web' and row['enabled'] and row['configured'] and
                        {'web_search', 'web_extract'} <= set(row['tools']) for row in toolsets)
        return {'available': available, 'mode': 'hermes_tools' if available else None,
                'reason': None if available else 'Hermes web search and extraction must be enabled and configured'}

    async def research_instruction(self, owner):
        status = await self.research_status(owner)
        if not status['available']:
            raise HTTPException(409, 'Research tools are unavailable for this account')
        return ('Research mode requested by the user. Use your Hermes web search and extraction tools to investigate the question. '
                'Prefer primary sources, corroborate material claims, distinguish evidence from uncertainty, and cite source URLs in the final answer. '
                'Do not claim sources were checked if tools failed or access was unavailable. Complete this research in this same Hermes run.')

    async def feature_capabilities(self, owner):
        await self._account(owner)
        settings = await asyncio.to_thread(self.accounts.get_model_settings, owner)
        # Exact pinned jobs API and previously qualified managed auto vision only.
        def memory_available():
            try:
                self._memory_file(owner, 'memory')
                self._memory_file(owner, 'user')
                return True
            except HTTPException:
                return False
        try:
            research = (await self.research_status(owner))['available']
        except HTTPException:
            research = False
        return {'scheduled_tasks': True, 'memory_controls': await asyncio.to_thread(memory_available),
                'image_input': settings['model'] == 'auto', 'connectors': False, 'research': research}

    async def feature_request(self, owner, method, path, *, json=None, params=None):
        user = await self._account(owner)
        permitted = ((path == '/api/jobs' and method in ('GET', 'POST')) or
                     (re.fullmatch(r'/api/jobs/[a-f0-9]{12}', path) and method in ('PATCH', 'DELETE')) or
                     (re.fullmatch(r'/api/jobs/[a-f0-9]{12}/(?:pause|resume|run)', path) and method == 'POST'))
        if not permitted:
            raise HTTPException(501, 'Unqualified provider operation')
        try:
            response = await self.http.request(method, user['hermes_url'].rstrip('/') + path,
                                               headers={'Authorization': 'Bearer ' + user['hermes_api_key']}, json=json, params=params)
            if response.status_code >= 400:
                # Remote detail can include config, paths or credentials.
                status = response.status_code if response.status_code in (400, 404, 409, 422, 429, 503) else 502
                raise HTTPException(status, 'Hermes task operation rejected')
            return response.json()
        except HTTPException:
            raise
        except Exception:
            raise HTTPException(502, 'Hermes task operation failed') from None

    def _memory_file(self, owner, section):
        if not ID.fullmatch(owner) or section not in ('memory', 'user'):
            raise HTTPException(404, 'Memory section not found')
        # Fixed layout shared with Provisioner._paths; never accept caller paths.
        components = (self.state_dir, self.state_dir / owner, self.state_dir / owner / 'home',
                      self.state_dir / owner / 'home' / 'memories')
        for component in components:
            if component.is_symlink():
                raise HTTPException(409, 'Unsafe memory storage')
        directory = components[-1]
        if not directory.is_dir():
            raise HTTPException(503, 'Hermes memory storage unavailable')
        path = directory / ('MEMORY.md' if section == 'memory' else 'USER.md')
        if path.is_symlink():
            raise HTTPException(409, 'Unsafe memory storage')
        return path

    def _memory_open(self, owner, section, flags):
        self._memory_file(owner, section)
        directory = os.open(self.state_dir, os.O_RDONLY | os.O_DIRECTORY | getattr(os, 'O_NOFOLLOW', 0))
        try:
            for name in (owner, 'home', 'memories'):
                next_directory = os.open(name, os.O_RDONLY | os.O_DIRECTORY | getattr(os, 'O_NOFOLLOW', 0), dir_fd=directory)
                os.close(directory)
                directory = next_directory
            fd = os.open('MEMORY.md' if section == 'memory' else 'USER.md', flags | getattr(os, 'O_NOFOLLOW', 0) | os.O_NONBLOCK, 0o600, dir_fd=directory)
            file_info = os.fstat(fd)
            if not stat.S_ISREG(file_info.st_mode) or file_info.st_nlink != 1:
                os.close(fd)
                raise HTTPException(409, 'Unsafe memory storage')
            if flags & os.O_CREAT and os.geteuid() == 0:
                info = os.fstat(directory)
                os.fchown(fd, info.st_uid, info.st_gid)
            return fd
        finally:
            os.close(directory)

    def _read_memory_section(self, owner, section):
        path = self._memory_file(owner, section)
        try:
            fd = self._memory_open(owner, section, os.O_RDONLY)
        except FileNotFoundError:
            data = b''
        else:
            with os.fdopen(fd, 'rb') as stream:
                if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
                    raise HTTPException(409, 'Unsafe memory storage')
                data = stream.read(65537)
        if len(data) > 65536:
            raise HTTPException(413, 'Memory section too large')
        try:
            content = data.decode('utf-8')
        except UnicodeDecodeError:
            raise HTTPException(409, 'Invalid memory encoding') from None
        return {'content': content, 'revision': hashlib.sha256(data).hexdigest()}

    async def read_memory(self, owner):
        await self._account(owner)
        def read():
            with self.accounts.lock:
                return {section: self._read_memory_section(owner, section) for section in ('memory', 'user')}
        return await asyncio.to_thread(read)

    async def write_memory(self, owner, section, content, revision):
        await self._account(owner)
        def write():
            with self.accounts.lock:
                current = self.accounts.get(owner)
                if not current or current['desired'] == 'deleted':
                    raise HTTPException(401, 'Account unavailable')
                old = self._read_memory_section(owner, section)
                if old['revision'] != revision:
                    raise HTTPException(409, 'Memory revision conflict')
                path = self._memory_file(owner, section)
                data = content.encode('utf-8')
                if len(data) > 65536:
                    raise HTTPException(413, 'Memory section too large')
                # Reopen without following symlinks and compare under the same
                # descriptor before in-place writing. Preserve runtime ownership.
                fd = self._memory_open(owner, section, os.O_RDWR | os.O_CREAT)
                with os.fdopen(fd, 'r+b') as stream:
                    info = os.fstat(stream.fileno())
                    if not stat.S_ISREG(info.st_mode) or hashlib.sha256(stream.read(65537)).hexdigest() != revision:
                        raise HTTPException(409, 'Memory revision conflict')
                    stream.seek(0)
                    stream.write(data)
                    stream.truncate()
                    stream.flush()
                    os.fsync(stream.fileno())
                return {'revision': hashlib.sha256(data).hexdigest()}
        return await asyncio.to_thread(write)


async def feature_json(request, limit):
    if request.headers.get('content-type', '').split(';')[0].lower() != 'application/json':
        raise HTTPException(415, 'JSON required')
    data = bytearray()
    async for chunk in request.stream():
        if len(data) + len(chunk) > limit:
            raise HTTPException(413, 'Request body too large')
        data.extend(chunk)
    try:
        result = json.loads(data)
    except (ValueError, UnicodeDecodeError):
        raise HTTPException(400, 'Invalid JSON') from None
    if not isinstance(result, dict):
        raise HTTPException(422, 'JSON object required')
    return result


def create_feature_router(store, provider=None):
    router = APIRouter()

    @router.post('/api/attachments', status_code=201)
    async def upload(request: Request, user=Depends(current_user)):
        data = bytearray()
        async for chunk in request.stream():
            if len(data) + len(chunk) > store.limits.upload_bytes:
                raise HTTPException(413, 'Upload byte limit exceeded')
            data.extend(chunk)
        user = await current_user(request)  # Revalidate after potentially slow body upload.
        attachment = await store.upload(user['id'], request.headers.get('x-filename', 'attachment'), request.headers.get('content-type', '').split(';')[0].strip().lower(), bytes(data))
        return {'attachment': attachment}

    @router.get('/api/attachments')
    async def attachments(user=Depends(current_user)):
        return {'attachments': await store.list_objects(user['id'], 'attachments'), 'next_cursor': None}

    @router.get('/api/attachments/{identifier}')
    async def attachment(identifier: str, user=Depends(current_user)):
        metadata, _ = await store.read_attachment(user['id'], identifier)
        return {'attachment': metadata}

    @router.get('/api/attachments/{identifier}/content')
    async def download(identifier: str, user=Depends(current_user)):
        metadata, data = await store.read_attachment(user['id'], identifier)
        return Response(data, media_type=metadata['mime'], headers={'Content-Disposition': "attachment; filename*=UTF-8''" + quote(metadata['filename'], safe=''), 'X-Content-Type-Options': 'nosniff', 'Cache-Control': 'no-store'})

    @router.delete('/api/attachments/{identifier}')
    async def delete_attachment(identifier: str, user=Depends(current_user)):
        return await store.delete_attachment(user['id'], identifier)

    @router.get('/api/artifacts')
    async def artifacts(user=Depends(current_user)):
        return {'artifacts': await store.list_objects(user['id'], 'artifacts'), 'next_cursor': None}

    @router.post('/api/artifacts', status_code=201)
    async def create_artifact(request: Request, user=Depends(current_user)):
        body = await feature_json(request, store.limits.artifact_bytes + 8192)
        if set(body) - {'title', 'kind', 'content'}:
            raise HTTPException(422, 'Unknown artifact fields')
        user = await current_user(request)
        return {'artifact': await store.save_artifact(user['id'], body.get('title'), body.get('kind', 'markdown'), body.get('content', ''))}

    @router.get('/api/artifacts/{identifier}')
    async def artifact(identifier: str, revision: int | None = None, user=Depends(current_user)):
        return {'artifact': await store.get_artifact(user['id'], identifier, revision)}

    @router.patch('/api/artifacts/{identifier}')
    async def edit_artifact(identifier: str, request: Request, user=Depends(current_user)):
        body = await feature_json(request, store.limits.artifact_bytes + 8192)
        if set(body) - {'revision', 'title', 'kind', 'content'}:
            raise HTTPException(422, 'Unknown artifact fields')
        user = await current_user(request)
        old = await store.get_artifact(user['id'], identifier)
        return {'artifact': await store.save_artifact(user['id'], body.get('title', old['title']), body.get('kind', old['kind']), body.get('content', old['content']), identifier, body.get('revision'))}

    @router.get('/api/artifacts/{identifier}/versions')
    async def artifact_versions(identifier: str, user=Depends(current_user)):
        def collect():
            with store.lock:
                store._get_artifact(user['id'], identifier)
                return [dict(row) for row in store.db.execute('SELECT revision,created_at FROM hope_artifact_versions WHERE artifact_id=? ORDER BY revision DESC', (identifier,))]
        return {'versions': await asyncio.to_thread(collect), 'next_cursor': None}

    @router.get('/api/artifacts/{identifier}/export')
    async def export_artifact(identifier: str, revision: int | None = None, user=Depends(current_user)):
        item = await store.get_artifact(user['id'], identifier, revision)
        suffix = '.md' if item['kind'] == 'markdown' else '.txt'
        return Response(item['content'].encode('utf-8'), media_type='text/plain', headers={'Content-Disposition': 'attachment; filename="artifact-' + item['id'] + suffix + '"', 'Cache-Control': 'no-store', 'X-Content-Type-Options': 'nosniff'})

    @router.delete('/api/artifacts/{identifier}')
    async def delete_artifact(identifier: str, user=Depends(current_user)):
        return await store.delete_artifact(user['id'], identifier)

    @router.get('/api/features/sync')
    async def sync(after: int = 0, user=Depends(current_user)):
        if after < 0:
            raise HTTPException(422, 'Invalid cursor')
        return await store.changes(user['id'], after)

    async def qualified(owner, capability):
        if provider is None or not hasattr(provider, 'feature_capabilities'):
            raise HTTPException(501, 'Operation is not qualified for this account')
        try:
            capabilities = await provider.feature_capabilities(owner)
        except Exception:
            raise HTTPException(503, 'Hermes capabilities unavailable') from None
        if not isinstance(capabilities, dict) or capabilities.get(capability) is not True:
            raise HTTPException(501, 'Operation is not qualified for this account')

    async def job_request(owner, method, suffix='', body=None, params=None):
        await qualified(owner, 'scheduled_tasks')
        try:
            return await provider.feature_request(owner, method, '/api/jobs' + suffix, json=body, params=params)
        except HTTPException:
            raise
        except Exception:
            raise HTTPException(502, 'Hermes task operation failed') from None

    def task(job):
        if not isinstance(job, dict) or not isinstance(job.get('id'), str) or not re.fullmatch(r'[a-f0-9]{12}', job['id']):
            raise HTTPException(502, 'Invalid Hermes task response')
        # Raw job rows contain upstream credentials/configuration. Never return them.
        keys = ('id', 'name', 'schedule', 'prompt', 'enabled', 'repeat', 'created_at', 'last_run_at', 'next_run_at')
        return {key: job[key] for key in keys if key in job}

    def task_id(identifier):
        if not re.fullmatch(r'[a-f0-9]{12}', identifier):
            raise HTTPException(404, 'Task not found')
        return identifier

    async def task_body(request, *, patch=False):
        body = await feature_json(request, 32768)
        allowed = {'name', 'schedule', 'prompt', 'repeat', 'enabled'} if patch else {'name', 'schedule', 'prompt', 'repeat', 'paused'}
        if set(body) - allowed or not body:
            raise HTTPException(422, 'Invalid task fields')
        for key, maximum in (('name', 200), ('schedule', 200), ('prompt', 5000)):
            if key in body and (not isinstance(body[key], str) or not 1 <= len(body[key].strip()) <= maximum):
                raise HTTPException(422, 'Invalid task ' + key)
            if not patch and key not in body:
                raise HTTPException(422, 'Task ' + key + ' required')
        if 'repeat' in body and (type(body['repeat']) is not int or body['repeat'] <= 0):
            raise HTTPException(422, 'Invalid task repeat')
        for key in ('enabled', 'paused'):
            if key in body and type(body[key]) is not bool:
                raise HTTPException(422, 'Invalid task ' + key)
        body['deliver'] = 'local'
        return body

    @router.get('/api/tasks')
    async def tasks(user=Depends(current_user)):
        result = await job_request(user['id'], 'GET', params={'include_disabled': 'true'})
        if not isinstance(result, dict) or not isinstance(result.get('jobs'), list):
            raise HTTPException(502, 'Invalid Hermes task response')
        return {'tasks': [task(job) for job in result['jobs']], 'next_cursor': None}

    @router.post('/api/tasks', status_code=201)
    async def create_task(request: Request, user=Depends(current_user)):
        body = await task_body(request)
        user = await current_user(request)
        result = await job_request(user['id'], 'POST', body=body)
        return {'task': task(result.get('job') if isinstance(result, dict) else None)}

    @router.patch('/api/tasks/{identifier}')
    async def update_task(identifier: str, request: Request, user=Depends(current_user)):
        body = await task_body(request, patch=True)
        user = await current_user(request)
        result = await job_request(user['id'], 'PATCH', '/' + task_id(identifier), body)
        return {'task': task(result.get('job') if isinstance(result, dict) else None)}

    @router.delete('/api/tasks/{identifier}')
    async def delete_task(identifier: str, user=Depends(current_user)):
        result = await job_request(user['id'], 'DELETE', '/' + task_id(identifier))
        if not isinstance(result, dict) or result.get('ok') is not True:
            raise HTTPException(502, 'Invalid Hermes task response')
        return {'ok': True}

    @router.post('/api/tasks/{identifier}/{action}')
    async def task_action(identifier: str, action: str, user=Depends(current_user)):
        if action not in ('pause', 'resume', 'run'):
            raise HTTPException(404, 'Task action not found')
        result = await job_request(user['id'], 'POST', '/' + task_id(identifier) + '/' + action)
        return {'task': task(result.get('job') if isinstance(result, dict) else None)}

    @router.get('/api/memory')
    async def memory(user=Depends(current_user)):
        await qualified(user['id'], 'memory_controls')
        try:
            result = await provider.read_memory(user['id'])
        except HTTPException:
            raise
        except Exception:
            raise HTTPException(502, 'Memory unavailable') from None
        # Dedicated trusted adapter returns only fixed sections, never paths/config.
        if not isinstance(result, dict):
            raise HTTPException(502, 'Invalid memory response')
        sections = []
        for section in ('memory', 'user'):
            value = result.get(section)
            if not isinstance(value, dict) or not isinstance(value.get('content'), str) or not isinstance(value.get('revision'), str):
                raise HTTPException(502, 'Invalid memory response')
            sections.append({'section': section, 'content': value['content'], 'revision': value['revision']})
        return {'sections': sections}

    @router.put('/api/memory/{section}')
    async def update_memory(section: str, request: Request, user=Depends(current_user)):
        if section not in ('memory', 'user'):
            raise HTTPException(404, 'Memory section not found')
        await qualified(user['id'], 'memory_controls')
        body = await feature_json(request, 131072)
        if set(body) != {'content', 'revision'} or not isinstance(body['content'], str) or len(body['content'].encode('utf-8')) > 65536 or not isinstance(body['revision'], str) or len(body['revision']) > 128:
            raise HTTPException(422, 'Invalid memory update')
        user = await current_user(request)
        try:
            result = await provider.write_memory(user['id'], section, body['content'], body['revision'])
        except HTTPException:
            raise
        except Exception:
            raise HTTPException(502, 'Memory update failed') from None
        if not isinstance(result, dict) or not isinstance(result.get('revision'), str):
            raise HTTPException(502, 'Invalid memory response')
        return {'section': section, 'content': body['content'], 'revision': result['revision']}

    @router.get('/api/connectors')
    async def connectors(user=Depends(current_user)):
        # The pinned gateway has no connector management API. Do not invent one.
        statuses = []
        if provider is not None and hasattr(provider, 'toolset_status'):
            try:
                statuses = await provider.toolset_status(user['id'])
            except HTTPException:
                pass
        return {'connectors': [], 'available': False, 'reason': 'Pinned Hermes has no connector credential/settings write API',
                'toolsets': [{'name': row['name'], 'enabled': row['enabled'], 'configured': row['configured']} for row in statuses]}

    @router.get('/api/research')
    async def research(user=Depends(current_user)):
        # Research remains normal Hermes execution, never a Hope reasoning loop.
        if provider is not None and hasattr(provider, 'research_status'):
            try:
                return await provider.research_status(user['id'])
            except HTTPException:
                pass
        return {'available': False, 'mode': None, 'reason': 'Hermes research tool availability could not be qualified'}

    return router


# Core merges these into its single /api/capabilities response.
def local_capabilities(store):
    return {'attachments': True, 'artifacts': True, 'pdf_extraction': importlib.util.find_spec('pypdf') is not None,
            'attachment_mimes': list(MIMES), 'upload_bytes': store.limits.upload_bytes,
            'account_attachment_bytes': store.limits.account_bytes, 'artifact_bytes': store.limits.artifact_bytes,
            'image_input': False, 'scheduled_tasks': False, 'memory_controls': False,
            'connectors': False, 'research': False}