"""Account-private history APIs. Private artifacts never enter Gradio's cache."""
from __future__ import annotations

import hashlib
import hmac
import json
import logging
import mimetypes
import os
from pathlib import Path
import re
import shutil
import threading
import time
from typing import Callable
from urllib.parse import unquote, urlsplit
import uuid

import anyio
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
from starlette.formparsers import MultiPartException
from starlette.responses import FileResponse, JSONResponse

from hf_space.history_auth import account_key, csrf_token, optional_identity, require_identity, require_mutation
from hf_space.history_oauth import install_oauth_routes
from hf_space import artifact_delivery


BASE = '/api/history'
PRIVATE_HEADERS = {
    'Cache-Control': 'private, no-store',
    'Pragma': 'no-cache',
    'X-Content-Type-Options': 'nosniff',
    'Cross-Origin-Resource-Policy': 'same-origin',
}
LOG = logging.getLogger(__name__)


class PrivateFileCache:
    """Bounded local mirror with leases, so another response cannot evict a read."""

    def __init__(self, root='/tmp/supratex-history-private', max_bytes=1024**3):
        self.root = Path(root)
        self.max_bytes = max_bytes
        self._guard = threading.RLock()
        self._locks = {}
        self._leases = {}
        self._reservations = {}

    def _release(self, key):
        with self._guard:
            count = self._leases.get(key, 0) - 1
            if count > 0:
                self._leases[key] = count
            else:
                self._leases.pop(key, None)

    def _make_room(self, needed, preserve):
        entries = []
        total = sum(self._reservations.values())
        for path in self.root.iterdir():
            if path.is_symlink() or not path.is_file() or path.name.startswith('.'):
                continue
            info = path.stat()
            entries.append((info.st_mtime_ns, path, info.st_size))
            total += info.st_size
        small = Path(preserve).suffix != '.glb'
        # A large model cannot evict the small images used by the entire UI.
        candidates = [item for item in entries if (item[1].suffix != '.glb') == small]
        partition_limit = min(self.max_bytes, 128*1024**2) if small else max(0, self.max_bytes-min(self.max_bytes//8,128*1024**2))
        partition_total = sum(item[2] for item in candidates)
        partition_total += sum(value for name,value in self._reservations.items()
                               if (Path(name).suffix != '.glb') == small)
        for _, path, size in sorted(candidates):
            if total + needed <= self.max_bytes and partition_total + needed <= partition_limit:
                break
            if path.name == preserve or self._leases.get(path.name):
                continue
            path.unlink(missing_ok=True)
            total -= size
            partition_total -= size
        if total + needed > self.max_bytes or partition_total + needed > partition_limit:
            raise HTTPException(503, 'Artifact downloads are busy. Please try again shortly.')

    def acquire(self, owner: str, source: Path, *, copy_from=None, progress=None, descriptor=None):
        source = Path(source)
        if descriptor is None and (source.is_symlink() or any(parent.is_symlink() for parent in source.absolute().parents)
                or not source.is_file()):
            raise HTTPException(404, 'File not found.')
        if descriptor is None:
            info = source.stat()
            size, version = info.st_size, str(info.st_mtime_ns)
            source_key = str(source.resolve())
        else:
            size, version = descriptor['size'], descriptor['version']
            source_key = os.path.abspath(source)
            if not isinstance(size,int) or size < 0 or source_key != os.path.abspath(descriptor['path']):
                raise HTTPException(404, 'File not found.')
        # The worker may seed the same bounded cache from its validated local
        # export. Authorization and the cache key still use the durable file.
        # This avoids writing to the Bucket and immediately reading it back.
        reader_source = Path(copy_from) if copy_from is not None else source
        if (copy_from is not None or descriptor is None) and (reader_source.is_symlink()
                or any(parent.is_symlink() for parent in reader_source.absolute().parents)
                or not reader_source.is_file() or reader_source.stat().st_size != size):
            raise HTTPException(404, 'File not found.')
        if size > self.max_bytes:
            raise HTTPException(413, 'This file exceeds the download size limit.')
        digest = hashlib.sha256((owner + '\0' + source_key + '\0' + str(size) + '\0' + version).encode()).hexdigest()
        suffix = source.suffix.lower() if source.suffix.lower() in ('.png', '.jpg', '.jpeg', '.webp', '.glb', '.npz', '.json') else '.bin'
        key = digest + suffix
        with self._guard:
            lock = self._locks.setdefault(key, threading.Lock())
        if not lock.acquire(timeout=30):
            raise HTTPException(503, 'Artifact downloads are busy. Please try again shortly.')
        try:
            self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
            destination = self.root / key
            with self._guard:
                self._leases[key] = self._leases.get(key, 0) + 1
                if not destination.is_file():
                    try:
                        self._make_room(size, key)
                        self._reservations[key] = size
                    except BaseException:
                        self._release(key)
                        raise
            try:
                if destination.is_symlink():
                    raise HTTPException(404, 'File not found.')
                if not destination.is_file():
                    temporary = self.root / ('.' + key + '.' + uuid.uuid4().hex + '.partial')
                    try:
                        direct = descriptor is not None and copy_from is None and artifact_delivery.bucket_url(source)
                        if direct:
                            artifact_delivery.download(source, temporary, size, progress=progress)
                        else:
                          if (reader_source.is_symlink() or not reader_source.is_file()
                                  or any(parent.is_symlink() for parent in reader_source.absolute().parents)):
                              raise HTTPException(404, 'File not found.')
                          with reader_source.open('rb') as reader, temporary.open('xb') as writer:
                            completed = 0
                            if progress:
                                progress(0, size)
                            while chunk := reader.read(8 * 1024 * 1024):
                                writer.write(chunk)
                                completed += len(chunk)
                                if progress:
                                    progress(completed, size)
                        if temporary.stat().st_size != size:
                            raise HTTPException(503, 'The artifact is being updated. Please try again.')
                        with self._guard:
                            temporary.replace(destination)
                            self._reservations.pop(key, None)
                    finally:
                        temporary.unlink(missing_ok=True)
                os.utime(destination, None)
                return destination, lambda: self._release(key)
            except BaseException:
                with self._guard:
                    self._reservations.pop(key, None)
                self._release(key)
                raise
        finally:
            lock.release()

    def seed(self, owner, source, local_source, *, progress=None, descriptor=None):
        """Prewarm a validated export without retaining an unbounded job copy."""
        _path, release = self.acquire(owner, source, copy_from=local_source, progress=progress, descriptor=descriptor)
        release()


class PrivateFileResponse(FileResponse):
    """Release download leases even when a client leaves mid-download."""

    def __init__(self, *args, release, **kwargs):
        super().__init__(*args, **kwargs)
        self._release = release

    async def __call__(self, scope, receive, send):
        try:
            await super().__call__(scope, receive, send)
        finally:
            self._release()


def _clean_resource_path(name: str) -> str:
    decoded = name
    for _ in range(4):
        next_value = unquote(decoded)
        if next_value == decoded:
            break
        decoded = next_value
    if (not decoded or '\\' in decoded or '\x00' in decoded or decoded.startswith('/')
            or ':' in decoded or any(part in ('', '.', '..') for part in decoded.split('/'))):
        raise HTTPException(404, 'File not found.')
    return decoded


def _public_allowed(path: Path, roots) -> bool:
    try:
        if path.is_symlink():
            return False
        resolved = path.resolve(strict=True)
        # Reject symlinks in any component, including a directory symlink.
        if any(parent.is_symlink() for parent in path.absolute().parents):
            return False
        for root in roots:
            root = Path(root).resolve(strict=True)
            if resolved == root or (root.is_dir() and resolved.is_relative_to(root)):
                return resolved.is_file()
    except (OSError, ValueError):
        pass
    return False


def install_routes(
    gradio_app: FastAPI,
    store,
    *,
    deployment_status: Callable,
    examples_provider: Callable | None = None,
    example_file_provider: Callable | None = None,
    prepare_example: Callable | None = None,
    prepare_models: Callable | None = None,
    preview_provider: Callable | None = None,
    public_gradio_paths=(),
    trusted_origins=(),
    max_upload_bytes=20 * 1024 * 1024,
    private_cache_root='/tmp/supratex-history-private',
    private_file_cache=None,
    csrf_secret: bytes | None = None,
):
    """Install on the mounted Gradio child app, which owns OAuth middleware.

    All providers and Store methods run in worker threads. The Store is the
    authorization authority and must check owner before looking up any file.
    """
    def require_account_binding(request: Request):
        supplied = request.headers.get('x-supra-account')
        if supplied is not None:
            identity = require_identity(request)
            if not re.fullmatch(r'[0-9a-f]{64}', supplied) or not hmac.compare_digest(supplied, account_key(identity, csrf_secret)):
                raise HTTPException(409, 'Your account changed. Verify your current account to continue.',
                                    headers={'X-Supra-Account-Changed': '1'})

    router = APIRouter(prefix=BASE, dependencies=[Depends(require_account_binding)])
    cache = private_file_cache if private_file_cache is not None else PrivateFileCache(private_cache_root)
    # Cold artifact reads can block on the mounted Bucket. Keep their workers
    # separate from metadata/status requests and bound the number of live reads.
    # Waiting requests acquire this token asynchronously. Normal AnyIO
    # cancellation keeps the token until the synchronous worker exits.
    artifact_limiter = anyio.CapacityLimiter(4)
    image_limiter = anyio.CapacityLimiter(4)
    upload_limiter = anyio.CapacityLimiter(2)
    mutation_limiter = anyio.CapacityLimiter(2)
    status_limiter = anyio.CapacityLimiter(2)
    delivery_limiter = anyio.CapacityLimiter(2)
    oauth_endpoints = install_oauth_routes(gradio_app)

    async def invoke(function, *args, _limiter=None, **kwargs):
        try:
            return await anyio.to_thread.run_sync(lambda: function(*args, **kwargs), limiter=_limiter)
        except HTTPException:
            raise
        except Exception as exc:
            # Do not expose storage paths or stack traces in an account API.
            status = {
                'NotFound': 404, 'Conflict': 409, 'NotReady': 503,
                'LimitExceeded': 413, 'ValidationError': 400,
                'FileNotFoundError': 404,
            }.get(type(exc).__name__)
            if status is not None:
                raise HTTPException(status, str(exc)) from None
            raise

    def available(*, read_only=False):
        if not (store.ready or (read_only and getattr(store, 'read_ready', False))):
            raise HTTPException(503, 'History is starting. Please try again shortly.')

    def workspace_status():
        # Memory-only: visitors must never initiate or wait on durable recovery.
        return {'stage': 'ready' if store.ready else 'failed' if getattr(store, 'error', None) else 'starting',
                'read_ready': bool(store.ready or getattr(store, 'read_ready', False))}

    async def require_ready():
        state = await invoke(deployment_status, _limiter=status_limiter)
        if (state.get('gpu_mode') != 'zerogpu'
                and (state.get('native_build', {}).get('stage') != 'compiled'
                     or state.get('models', {}).get('stage') != 'ready_released')):
            raise HTTPException(503, 'Models are loading in the background. Please wait until loading is complete.')
        if not store.ready:
            raise HTTPException(503, 'Preparing your workspace. Please wait a moment.')

    from hf_space.gallery_preview import install_asset_routes
    # The partial local test snapshot may omit the unchanged binary assets.
    if all(Path(root).exists() for root in public_gradio_paths):
        if public_gradio_paths:
            install_asset_routes(gradio_app, require_ready)

    def descriptor(owner, identifier, name, upload=False):
        lookup = getattr(store, 'get_file_descriptor', None)
        info = lookup(owner, identifier, name, upload=upload) if lookup else None
        if info:
            return info
        # Legacy files have no manifest. Resolve their *lexical* location only
        # after ownership/allowlist checks, then query bounded object HTTP.
        record = store.get_upload(owner,identifier) if upload else store.get_job(owner,identifier)
        if name in ('input.png','thumb.webp','display.webp','foreground.png'):
            path = store.root/('uploads' if upload else 'jobs')/identifier/name
        elif not upload and name == 'asset.glb' and record.get('export_status') == 'complete':
            path = store.asset_root/record['asset_id']/name
        elif not upload and re.fullmatch(r'frames/[a-z][a-z0-9_]{0,47}-[0-9]{2}\.webp',name):
            metadata = store.cached_preview(owner,identifier) if hasattr(store,'cached_preview') else store.get_preview(owner,identifier)
            mode, ordinal = name[7:-5].rsplit('-',1)
            if not metadata or mode not in metadata['modes'] or int(ordinal)>=metadata['views']:
                raise HTTPException(404,'File not found.')
            path = store.asset_root/record['asset_id']/'views'/name[7:]
        else:
            raise HTTPException(404,'File not found.')
        if artifact_delivery.bucket_url(path):
            try:
                info = artifact_delivery.describe(path)
            except FileNotFoundError:
                raise HTTPException(404,'File not found.') from None
            except Exception:
                raise HTTPException(503,'Artifact storage is temporarily unavailable. Please retry.') from None
        else:
            resolver = store.resolve_upload_file if upload else store.resolve_file
            path = resolver(owner,identifier,name)
            stat = path.stat()
            info = {'path':str(path),'size':stat.st_size,
                    'version':hashlib.sha256((str(path)+str(stat.st_mtime_ns)).encode()).hexdigest()}
        remember = getattr(store,'remember_file_descriptor',None)
        if remember:
            remember(owner,identifier,name,info,upload=upload)
        return info

    async def body(request, allowed):
        length = request.headers.get('content-length')
        if length and (not length.isdigit() or int(length) > 16 * 1024):
            raise HTTPException(413, 'Request is too large.')
        data = await request.body()
        if len(data) > 16 * 1024:
            raise HTTPException(413, 'Request is too large.')
        try:
            value = json.loads(data)
        except (ValueError, UnicodeError):
            raise HTTPException(400, 'Invalid JSON request.') from None
        if not isinstance(value, dict) or set(value) - set(allowed):
            raise HTTPException(400, 'Invalid request fields.')
        return value

    def key(value):
        if not isinstance(value, str) or not 8 <= len(value) <= 128:
            raise HTTPException(400, 'A valid submission key is required.')
        return value

    def snapshot(job):
        result = store.public_snapshot(job)
        job_id = result['id']
        result.update({
            'input_url': f'{BASE}/jobs/{job_id}/files/input.png',
            'display_url': f'{BASE}/jobs/{job_id}/files/display.webp' if job.get('files', {}).get('display.webp') else f'{BASE}/jobs/{job_id}/files/input.png',
            'thumbnail_url': f'{BASE}/jobs/{job_id}/files/thumb.webp',
            'preview_url': f'{BASE}/jobs/{job_id}/preview',
        })
        if result.get('foreground_prepared'):
            result['foreground_url'] = f'{BASE}/jobs/{job_id}/files/foreground.png'
        if result.get('glb_ready') or result.get('export_status') == 'complete':
            result['download_url'] = f'{BASE}/jobs/{job_id}/files/asset.glb'
        return result

    def upload_snapshot(upload):
        upload_id = upload.get('upload_id') or upload['id']
        result = {
            'upload_id': upload_id,
            'image_url': f'{BASE}/uploads/{upload_id}/files/input.png',
            'display_url': f'{BASE}/uploads/{upload_id}/files/display.webp' if upload.get('files', {}).get('display.webp') else f'{BASE}/uploads/{upload_id}/files/input.png',
            'thumbnail_url': f'{BASE}/uploads/{upload_id}/files/thumb.webp',
            'width': upload.get('width'), 'height': upload.get('height'),
        }
        if upload.get('foreground'):
            result['foreground'] = upload['foreground']
            result['status_url'] = f'{BASE}/uploads/{upload_id}'
            if upload['foreground'].get('status') == 'complete' and upload.get('files', {}).get('foreground.png'):
                result['foreground_url'] = f'{BASE}/uploads/{upload_id}/files/foreground.png'
        for field in ('gpu_mode', 'gpu_operation_id', 'gpu_operation_kind'):
            if field in upload:
                result[field] = upload[field]
        return result

    def prime_completed_upload(owner, upload):
        # Inputs with an existing alpha finish during upload, without passing
        # through Service.foreground's cache warmup. Serve those exact committed
        # bytes locally while the Bucket's HTTP view catches up with its mount.
        if upload.get('foreground', {}).get('status') != 'complete':
            return
        identifier = upload.get('upload_id') or upload['id']
        try:
            info = store.get_file_descriptor(owner, identifier, 'foreground.png', upload=True)
            if info:
                source = Path(info['path'])
                cache.seed(owner, source, source, descriptor=info)
        except Exception:
            # The upload was already durably accepted; never report it as an
            # unsuccessful POST and encourage duplicate processing.
            LOG.warning('Could not warm the completed upload image cache')

    @gradio_app.middleware('http')
    async def privacy_guard(request: Request, call_next):
        path = request.scope.get('path', '')
        if path.rstrip('/') in ('/login/huggingface', '/login/callback', '/logout') and '_target_url' in request.query_params:
            target = request.query_params.get('_target_url', '')
            decoded_target = target
            for _ in range(4):
                new_target = unquote(decoded_target)
                if new_target == decoded_target:
                    break
                decoded_target = new_target
            try:
                parsed = urlsplit(decoded_target)
                safe_target = (decoded_target.startswith('/') and not decoded_target.startswith('//')
                               and not parsed.scheme and not parsed.netloc and '\\' not in decoded_target
                               and not any(ord(c) < 32 for c in decoded_target))
            except ValueError:
                safe_target = False
            if not safe_target:
                return JSONResponse({'detail': 'Invalid return address.'}, 400, headers=PRIVATE_HEADERS)
        if path == BASE + '/upload' and request.method == 'POST':
            # Reject before multipart parsing/decoding can consume startup RAM.
            try:
                await require_ready()
            except HTTPException as exc:
                # This response guarantees that no upload body was processed.
                # A stale tab can retain its file and resume after readiness.
                return JSONResponse({'detail': exc.detail, 'code': 'upload_not_started'},
                                    exc.status_code, headers=PRIVATE_HEADERS)
            original_receive = request._receive
            received = 0

            async def limited_receive():
                nonlocal received
                message = await original_receive()
                if message.get('type') == 'http.request':
                    received += len(message.get('body', b''))
                    if received > max_upload_bytes + 64 * 1024:
                        request.scope['supra_upload_too_large'] = True
                        # The multipart parser closes already-spooled parts on
                        # this exception before converting it to an HTTP error.
                        raise MultiPartException('The uploaded image is too large.')
                return message

            request._receive = limited_receive
        if path.rstrip('/') == '/gradio_api/upload':
            return JSONResponse({'detail': 'Use the account upload endpoint.'}, 403, headers=PRIVATE_HEADERS)
        for prefix in ('/gradio_api/file=', '/file=', '/gradio_api/file/', '/file/'):
            if path.startswith(prefix):
                candidate = path[len(prefix):]
                for _ in range(4):
                    decoded = unquote(candidate)
                    if decoded == candidate:
                        break
                    candidate = decoded
                if not _public_allowed(Path(candidate), public_gradio_paths):
                    return JSONResponse({'detail': 'File not found.'}, 404, headers=PRIVATE_HEADERS)
                try:
                    await require_ready()
                except HTTPException as exc:
                    return JSONResponse({'detail': exc.detail}, exc.status_code, headers=PRIVATE_HEADERS)
        try:
            response = await call_next(request)
        except BaseException:
            if request.scope.get('supra_upload_too_large'):
                return JSONResponse({'detail': 'Please upload an image smaller than 20 MiB.'}, 413, headers=PRIVATE_HEADERS)
            raise
        if request.scope.get('supra_upload_too_large'):
            return JSONResponse({'detail': 'Please upload an image smaller than 20 MiB.'}, 413, headers=PRIVATE_HEADERS)
        if path.startswith(BASE):
            for name, value in PRIVATE_HEADERS.items():
                if name.lower() == 'cache-control' and '/examples/' in path and response.status_code in (200, 206, 304):
                    continue
                response.headers[name] = value
        return response

    @router.get('/me')
    async def me(request: Request):
        identity = optional_identity(request)
        return {'authenticated': identity is not None,
                'username': identity.username if identity else None,
                'account_key': account_key(identity, csrf_secret) if identity else None,
                'verification_key': (request.scope.get('supra_history_app_claims') or {}).get('check') if identity else None,
                'csrf_token': csrf_token(identity, csrf_secret) if identity else None,
                'login_url': oauth_endpoints['login_url'] if oauth_endpoints else '/login/huggingface',
                'silent_login_url': oauth_endpoints['silent_login_url'] if oauth_endpoints else None,
                'bridge_url': oauth_endpoints['bridge_url'] if oauth_endpoints else None,
                'popup_url': oauth_endpoints['popup_url'] if oauth_endpoints else None,
                'logout_url': '/logout'}

    @router.get('/status')
    async def status():
        state = await invoke(deployment_status, _limiter=status_limiter)
        return {**state, 'workspace': workspace_status()}

    @router.post('/gpu/prepare')
    async def prepare_gpu_task(request: Request):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        available()
        value = await body(request, ('operation_id',))
        pending = await invoke(store.get_operation, identity.owner, value.get('operation_id'),
                               _limiter=status_limiter)
        state = await invoke(deployment_status, _limiter=status_limiter)
        if state.get('gpu_mode') != 'zerogpu' or prepare_models is None:
            raise HTTPException(404, 'On-demand preparation is unavailable.')
        if pending['operation']['status'] == 'queued':
            kind = pending['operation']['kind']
            required = ('foreground',) if kind == 'foreground' else ('native_build', 'models')
            if any(state.get(key, {}).get('stage') == 'failed' for key in required):
                raise HTTPException(503, 'Model preparation failed. Please try again later.')
            # This callback only starts the shared CPU preparation controller;
            # GPU allocation remains in the visitor's later Gradio request.
            await invoke(prepare_models, kind, _limiter=mutation_limiter)
        return {'operation_id': pending['operation']['id'],
                'status': pending['operation']['status'], 'gpu_mode': 'zerogpu'}

    @router.get('/examples')
    async def examples():
        # This manifest only describes committed public assets. It does not
        # stage a model or access the mounted Bucket during GPU startup.
        value = await invoke(examples_provider, _limiter=image_limiter) if examples_provider else []
        return value if isinstance(value, dict) else {'items': value}

    @router.get('/examples/{name}/{kind}')
    async def example_resource(name: str, kind: str):
        if not example_file_provider or kind not in ('input', 'thumbnail', 'model'):
            raise HTTPException(404, 'Example not found.')
        if kind != 'thumbnail':
            await require_ready()
        name = _clean_resource_path(name)
        if '/' in name:
            raise HTTPException(404, 'Example not found.')
        path = Path(await invoke(example_file_provider, name, kind,
                                 _limiter=artifact_limiter if kind == 'model' else image_limiter))
        if path.is_symlink() or not path.is_file():
            raise HTTPException(404, 'Example not found.')
        media = 'model/gltf-binary' if kind == 'model' else ('image/webp' if kind == 'thumbnail' else 'image/png')
        return FileResponse(path, media_type=media, headers={'Cache-Control': 'public, max-age=3600', 'X-Content-Type-Options': 'nosniff'})

    @router.post('/upload')
    async def upload(request: Request):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        available()
        length = request.headers.get('content-length')
        if length and (not length.isdigit() or int(length) > max_upload_bytes + 64 * 1024):
            raise HTTPException(413, 'Please upload an image smaller than 20 MiB.')
        async with request.form(max_files=1, max_fields=0, max_part_size=max_upload_bytes) as form:
            item = form.get('file')
            if item is None or not hasattr(item, 'read'):
                raise HTTPException(400, 'Choose an image to upload.')
            data = await item.read(max_upload_bytes + 1)
            if not data or len(data) > max_upload_bytes:
                raise HTTPException(413, 'Please upload an image smaller than 20 MiB.')
            value = await invoke(store.save_upload, identity.owner, data, _limiter=upload_limiter)
        value = await invoke(store.request_foreground, identity.owner, value['upload_id'],
                             'foreground-upload-' + value['upload_id'], _limiter=mutation_limiter)
        await invoke(prime_completed_upload, identity.owner, value, _limiter=image_limiter)
        return upload_snapshot(value)

    @router.get('/uploads/{upload_id}')
    async def upload_status(upload_id: str, request: Request):
        identity = require_identity(request)
        available()
        value = await invoke(store.get_upload, identity.owner, upload_id, _limiter=status_limiter)
        return upload_snapshot(value)

    @router.post('/uploads/{upload_id}/foreground')
    async def retry_foreground(upload_id: str, request: Request):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        await require_ready()
        available()
        value = await body(request, ('idempotency_key',))
        result = await invoke(store.request_foreground, identity.owner, upload_id,
                              key(value.get('idempotency_key')), _limiter=mutation_limiter)
        await invoke(prime_completed_upload, identity.owner, result, _limiter=image_limiter)
        return upload_snapshot(result)

    @router.post('/submit')
    async def submit(request: Request):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        available()
        value = await body(request, ('upload_id', 'example', 'seed', 'idempotency_key'))
        request_key = key(value.get('idempotency_key'))
        if bool(value.get('upload_id')) == bool(value.get('example')):
            raise HTTPException(400, 'Choose one uploaded image or example.')
        state = await invoke(deployment_status)
        native = state.get('native_build', {}) if isinstance(state, dict) else {}
        models = state.get('models', {}) if isinstance(state, dict) else {}
        if native.get('stage') == 'failed' or models.get('stage') == 'failed':
            raise HTTPException(503, 'Model loading failed. Please try again later.')
        if state.get('gpu_mode') != 'zerogpu' and (native.get('stage') != 'compiled' or models.get('stage') != 'ready_released'):
            raise HTTPException(503, 'Models are loading in the background. Please wait until loading is complete.')
        upload_id = value.get('upload_id')
        if value.get('example'):
            if not prepare_example:
                raise HTTPException(404, 'Example not found.')
            name = _clean_resource_path(str(value['example']))
            if '/' in name:
                raise HTTPException(404, 'Example not found.')
            prepared = await invoke(prepare_example, identity.owner, name, _limiter=upload_limiter)
            upload_id = prepared.get('upload_id') or prepared['id']
        job = await invoke(store.submit, identity.owner, upload_id, value.get('seed', 42), request_key, _limiter=mutation_limiter)
        return await invoke(snapshot, job)

    @router.get('/jobs')
    async def jobs(request: Request, cursor: str | None = None, limit: int = 24, since: str | None = None):
        identity = require_identity(request)
        available(read_only=True)
        if not 1 <= limit <= 100 or (cursor is not None and len(cursor) > 256):
            raise HTTPException(400, 'Invalid page.')
        if since is not None and (cursor is not None or len(since)>128):
            raise HTTPException(400, 'Invalid update cursor.')
        result = await invoke(store.list_jobs, identity.owner, cursor=cursor, limit=limit,
                              **({'since':since} if since is not None else {}))
        # list_jobs already returns public snapshots; retrieve internal records
        # only if needed by public_snapshot implementations.
        items = []
        for item in result.get('items', result.get('jobs', [])):
            item = dict(item)
            job_id = item['id']
            item.update(input_url=f'{BASE}/jobs/{job_id}/files/input.png',
                        display_url=f'{BASE}/jobs/{job_id}/files/display.webp' if item.get('display_ready') else f'{BASE}/jobs/{job_id}/files/input.png',
                        thumbnail_url=f'{BASE}/jobs/{job_id}/files/thumb.webp',
                        preview_url=f'{BASE}/jobs/{job_id}/preview')
            if item.get('export_status') == 'complete':
                item['download_url'] = f'{BASE}/jobs/{job_id}/files/asset.glb'
            if item.get('foreground_prepared'):
                item['foreground_url'] = f'{BASE}/jobs/{job_id}/files/foreground.png'
            items.append(item)
        return {'items': items, 'next_cursor': result.get('next_cursor'),
                **{key:result[key] for key in ('revision','reset','deleted_ids') if key in result}}

    @router.delete('/jobs/{job_id}')
    async def delete_job(job_id: str, request: Request):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        available()
        return await invoke(store.delete_job, identity.owner, job_id, _limiter=mutation_limiter)

    @router.get('/jobs/{job_id}')
    async def job_detail(job_id: str, request: Request):
        identity = require_identity(request)
        available(read_only=True)
        job = await invoke(store.get_job, identity.owner, job_id)
        return await invoke(snapshot, job)

    @router.get('/jobs/{job_id}/preview')
    async def job_preview(job_id: str, request: Request, view: str = 'auto'):
        if view not in ('auto', 'render', 'glb'):
            raise HTTPException(422, 'Unknown preview view.')
        identity = require_identity(request)
        await require_ready()
        available()
        job = await invoke(store.get_job, identity.owner, job_id)
        if job.get('export_status') == 'complete' and view != 'render':
            await invoke(descriptor, identity.owner, job_id, 'asset.glb', _limiter=artifact_limiter)
        if not preview_provider:
            return {'html': '', 'job': await invoke(snapshot, job)}
        if view == 'auto':
            return await invoke(preview_provider, identity.owner, job, _limiter=image_limiter)
        return await invoke(preview_provider, identity.owner, job, view=view, _limiter=image_limiter)

    async def operation(request, job_id, kind):
        identity = require_mutation(request, trusted_origins, csrf_secret)
        await require_ready()
        available()
        value = await body(request, ('idempotency_key',))
        function = {'export': store.submit_export, 'recover': store.submit_recover,
                    'texture': store.submit_texture}[kind]
        job = await invoke(function, identity.owner, job_id, key(value.get('idempotency_key')), _limiter=mutation_limiter)
        return await invoke(snapshot, job)

    @router.post('/jobs/{job_id}/export')
    async def export(job_id: str, request: Request):
        return await operation(request, job_id, 'export')

    @router.post('/jobs/{job_id}/texture')
    async def texture(job_id: str, request: Request):
        return await operation(request, job_id, 'texture')

    @router.post('/jobs/{job_id}/recover')
    async def recover(job_id: str, request: Request):
        return await operation(request, job_id, 'recover')

    async def private_file(request, owner, job_id, name, is_upload=False):
        name = _clean_resource_path(name)
        # Startup history is deliberately limited to small owner-checked cards.
        # Original inputs, previews and models retain the full readiness gate.
        read_only = not is_upload and name == 'thumb.webp'
        if not read_only:
            await require_ready()
        available(read_only=read_only)
        resolver = store.resolve_upload_file if is_upload else store.resolve_file
        def prepare():
            # Ownership is checked by the resolver before touching cache/files.
            # Include its mounted-file stat calls in the same bounded I/O pool.
            info = descriptor(owner, job_id, name, is_upload)
            source = Path(info['path']) if info else resolver(owner, job_id, name)
            return cache.acquire(owner, source, descriptor=info)
        try:
            local, release = await invoke(prepare, _limiter=artifact_limiter if name.endswith('.glb') else image_limiter)
        except RuntimeError:
            raise HTTPException(503, 'Artifact storage is temporarily unavailable. Please retry.') from None
        media = {'.webp': 'image/webp', '.png': 'image/png'}.get(Path(name).suffix.lower())
        media = media or mimetypes.guess_type(name)[0] or 'application/octet-stream'
        if name.lower().endswith('.glb'):
            media = 'model/gltf-binary'
        return PrivateFileResponse(local, media_type=media, filename=Path(name).name,
                            content_disposition_type='inline' if media.startswith('image/') else 'attachment',
                            headers=PRIVATE_HEADERS, release=release)

    @router.get('/jobs/{job_id}/delivery')
    async def model_delivery(job_id: str, request: Request):
        identity = require_identity(request)
        await require_ready()
        available()
        info = await invoke(descriptor, identity.owner, job_id, 'asset.glb', _limiter=artifact_limiter)
        fallback = f'{BASE}/jobs/{job_id}/files/asset.glb'
        if not info:
            # Legacy records use the protected file path until indexed once.
            await invoke(store.resolve_file, identity.owner, job_id, 'asset.glb', _limiter=artifact_limiter)
            return {'url':fallback,'direct':False}
        grant = await invoke(artifact_delivery.cdn_grant, identity.owner, info, _limiter=delivery_limiter)
        return {'url':grant['url'] if grant else fallback, 'bytes':info['size'],
                'version':info['version'], 'direct':bool(grant),
                'expires_in':max(0,int(grant['expires_at']-time.time())) if grant else 0}

    @router.get('/jobs/{job_id}/files/{name:path}')
    async def job_file(job_id: str, name: str, request: Request):
        identity = require_identity(request)
        return await private_file(request, identity.owner, job_id, name)

    @router.get('/uploads/{upload_id}/files/{name:path}')
    async def upload_file(upload_id: str, name: str, request: Request):
        identity = require_identity(request)
        return await private_file(request, identity.owner, upload_id, name, True)

    gradio_app.include_router(router)
    return router