feat: add remote API automation and service updates

.env.example

@@ -39,6 +39,11 @@ CELERY_PUB_CONCURRENCY=1
 # Build with real ASR (faster-whisper)
 INSTALL_ASR=1
 
+# Optional: refresh yt-dlp at image build time.
+# - latest: install newest release available when this layer is rebuilt
+# - or pin a specific version, e.g. 2026.03.03
+YTDLP_VERSION=latest
+
 # Optional: allow browser origins (comma-separated)
 CORS_ALLOW_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
 

Dockerfile

@@ -12,6 +12,7 @@ RUN apt-get update \
 COPY pyproject.toml README.md ./
 
 ARG INSTALL_ASR=0
+ARG YTDLP_VERSION=latest
 
 # Install dependencies in a cache-friendly layer so editing source code doesn't
 # force re-downloading everything on every docker build.
@@ -25,6 +26,14 @@ COPY docker ./docker
 
 RUN pip install --no-cache-dir -e . --no-deps
 
+RUN if [ -n "$YTDLP_VERSION" ]; then \
+      if [ "$YTDLP_VERSION" = "latest" ]; then \
+        pip install --no-cache-dir -U "yt-dlp[default]"; \
+      else \
+        pip install --no-cache-dir -U "yt-dlp[default]==${YTDLP_VERSION}"; \
+      fi; \
+    fi
+
 RUN chmod +x /app/docker/entrypoint.sh
 
 CMD ["/app/docker/entrypoint.sh"]

compose.yml

@@ -35,6 +35,7 @@ services:
       context: .
       args:
         INSTALL_ASR: ${INSTALL_ASR:-1}
+        YTDLP_VERSION: ${YTDLP_VERSION:-latest}
     image: videoroll:prod
     environment:
       DATABASE_URL: ${DATABASE_URL}

docker-compose.yml

@@ -35,6 +35,7 @@ services:
       context: .
       args:
         INSTALL_ASR: ${INSTALL_ASR:-1}
+        YTDLP_VERSION: ${YTDLP_VERSION:-latest}
     image: videoroll:prod
     environment:
       DATABASE_URL: ${DATABASE_URL}

skills/openclaw-videoroll/SKILL.md

@@ -0,0 +1,79 @@
+---
+name: openclaw-videoroll
+description: Collect YouTube video URLs on an external machine and submit them to VideoRoll through the remote auto-ingest API. Use when OpenClaw needs to find canonical YouTube watch links, deduplicate them, and push each URL to VideoRoll with the configured token so the project starts auto mode immediately.
+---
+
+# OpenClaw -> VideoRoll
+
+## Overview
+
+Use this skill only for one job: get valid YouTube video links and send them to VideoRoll's remote API.
+
+## Required config
+
+- `VIDEOROLL_REMOTE_URL`
+  Full remote endpoint URL. Usually:
+  `https://your-host/api/remote/auto/youtube`
+- `VIDEOROLL_REMOTE_TOKEN`
+  Token configured in VideoRoll's `Settings · API`
+- Optional: `VIDEOROLL_SOURCE_LICENSE`
+  Default `authorized`
+- Optional: `VIDEOROLL_PROOF_URL`
+  Authorization proof link if needed
+
+## Workflow
+
+1. Collect candidate YouTube links from the requested source.
+2. Normalize every link to a canonical watch URL:
+   `https://www.youtube.com/watch?v=<video_id>`
+3. Drop non-video links, duplicates, shorts landing pages without a resolvable video id, and obviously broken URLs.
+4. For each final URL, call VideoRoll remote API once.
+5. Return the API result for each URL, especially:
+   `task_id`, `pipeline_job_id`, `deduped`, `source_id`
+
+## API call
+
+Send query parameters:
+
+- `token`
+- `url`
+- Optional `license`
+- Optional `proof_url`
+
+Example:
+
+```bash
+curl -G "$VIDEOROLL_REMOTE_URL" \
+  --data-urlencode "token=$VIDEOROLL_REMOTE_TOKEN" \
+  --data-urlencode "url=https://www.youtube.com/watch?v=dQw4w9WgXcQ" \
+  --data-urlencode "license=${VIDEOROLL_SOURCE_LICENSE:-authorized}"
+```
+
+Success response shape:
+
+```json
+{
+  "task_id": "...",
+  "pipeline_job_id": "...",
+  "deduped": false,
+  "source_id": "..."
+}
+```
+
+## Rules
+
+- Only submit actual YouTube video URLs.
+- Preserve the original video id exactly.
+- Do not invent metadata that the API does not need.
+- If the API returns `deduped=true`, report it instead of retrying.
+- If token/auth fails, stop and surface the exact error.
+- Do not attempt to bypass paywalls, private videos, or platform restrictions.
+
+## Output format
+
+When reporting back, list each processed URL with:
+- normalized URL
+- API status
+- `task_id` if created
+- `pipeline_job_id` if started
+- whether it was deduped

skills/openclaw-videoroll/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "OpenClaw VideoRoll"
+  short_description: "Collect YouTube URLs and push them to VideoRoll"
+  default_prompt: "Use $openclaw-videoroll to collect YouTube watch URLs on an external machine and submit them to VideoRoll through the remote API."

src/videoroll/apps/bilibili_publisher/worker.py

@@ -2,8 +2,10 @@ from __future__ import annotations
 
 import json
 import logging
+import os
 import random
 import tempfile
+import threading
 import traceback
 import uuid
 from datetime import datetime, timezone
@@ -31,6 +33,8 @@ from videoroll.apps.subtitle_service.translate_settings_store import get_transla
 
 settings = get_bilibili_publisher_settings()
 logger = logging.getLogger(__name__)
+_DB_READY_LOCK = threading.Lock()
+_DB_READY_PID: int | None = None
 celery_app = Celery("bilibili_publisher", broker=settings.redis_url, backend=settings.redis_url)
 celery_app.conf.update(
     task_serializer="json",
@@ -47,9 +51,17 @@ def _db() -> Session:
 
 
 def _ensure_db() -> None:
-    engine = get_engine(settings.database_url)
-    Base.metadata.create_all(engine)
-    auto_migrate(settings.database_url)
+    global _DB_READY_PID
+    pid = os.getpid()
+    if _DB_READY_PID == pid:
+        return
+    with _DB_READY_LOCK:
+        if _DB_READY_PID == pid:
+            return
+        engine = get_engine(settings.database_url)
+        Base.metadata.create_all(engine)
+        auto_migrate(settings.database_url)
+        _DB_READY_PID = pid
 
 
 def _utcnow() -> datetime:

src/videoroll/apps/orchestrator_api/main.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import hashlib
 import hmac
 import json
+import logging
 import os
 import tempfile
 import threading
@@ -18,6 +19,7 @@ from botocore.exceptions import ClientError
 from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, Response, StreamingResponse
+from starlette.concurrency import run_in_threadpool
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.types import ASGIApp
 from sqlalchemy import func
@@ -26,7 +28,7 @@ from sqlalchemy.orm import Session
 from videoroll.config import OrchestratorSettings, get_orchestrator_settings
 from videoroll.db.base import Base
 from videoroll.db.auto_migrate import auto_migrate
-from videoroll.db.models import AppSetting, Asset, AssetKind, PublishJob, Subtitle, SubtitleJob, SubtitleJobStatus, Task, TaskStatus
+from videoroll.db.models import AppSetting, Asset, AssetKind, PublishJob, SourceLicense, Subtitle, SubtitleJob, SubtitleJobStatus, Task, TaskStatus
 from videoroll.db.session import db_session, get_engine, get_sessionmaker
 from videoroll.storage.s3 import S3Store
 from videoroll.utils.internal_api_token import internal_api_token
@@ -41,6 +43,8 @@ from videoroll.apps.orchestrator_api.schemas import (
     ConvertedVideoItem,
     PublishActionRequest,
     PublishJobSummary,
+    RemoteAPISettingsRead,
+    RemoteAPISettingsUpdate,
     RemoteJobResponse,
     RemotePublishResponse,
     StorageRetentionSettingsRead,
@@ -70,6 +74,7 @@ from videoroll.apps.orchestrator_api.admin_auth_store import (
     verify_password_hash,
 )
 from videoroll.apps.orchestrator_api.youtube_downloader import (
+    YtDlpRuntimeError,
     download_youtube_video,
     download_thumbnail_jpg,
     extract_youtube_metadata,
@@ -79,6 +84,14 @@ from videoroll.apps.orchestrator_api.storage_retention_store import (
     get_storage_retention_settings,
     update_storage_retention_settings,
 )
+from videoroll.apps.orchestrator_api.remote_api_settings_store import (
+    REMOTE_API_TOKEN_QUERY_PARAM,
+    REMOTE_AUTO_YOUTUBE_PATH,
+    get_remote_api_settings,
+    remote_api_token_is_configured,
+    update_remote_api_settings,
+    verify_remote_api_token,
+)
 from videoroll.apps.subtitle_service.bilibili_tags_store import get_task_bilibili_tags
 from videoroll.apps.subtitle_service.task_title_store import get_task_display_title_with_s3
 from videoroll.apps.youtube_settings_store import (
@@ -89,6 +102,8 @@ from videoroll.apps.youtube_settings_store import (
     update_youtube_settings,
 )
 
+logger = logging.getLogger(__name__)
+
 
 def get_settings() -> OrchestratorSettings:
     return get_orchestrator_settings()
@@ -99,9 +114,7 @@ def get_db(settings: OrchestratorSettings = Depends(get_settings)) -> Generator[
 
 
 def get_s3(settings: OrchestratorSettings = Depends(get_settings)) -> S3Store:
-    store = S3Store(settings)
-    store.ensure_bucket()
-    return store
+    return S3Store(settings)
 
 
 def _sha256_bytes(data: bytes) -> str:
@@ -300,6 +313,228 @@ def _write_s3_json(s3: S3Store, key: str, obj: dict[str, Any]) -> bytes:
     return payload
 
 
+def _write_s3_text(s3: S3Store, key: str, text: str) -> bytes:
+    payload = str(text).encode("utf-8")
+    s3.put_bytes(payload, key, content_type="text/plain; charset=utf-8")
+    return payload
+
+
+def _store_task_log_asset(
+    db: Session,
+    s3: S3Store,
+    *,
+    task_id: uuid.UUID,
+    log_key: str,
+    text: str,
+) -> Asset:
+    payload = _write_s3_text(s3, log_key, text)
+    asset = Asset(
+        task_id=task_id,
+        kind=AssetKind.log,
+        storage_key=log_key,
+        sha256=_sha256_bytes(payload),
+        size_bytes=len(payload),
+    )
+    db.add(asset)
+    db.commit()
+    db.refresh(asset)
+    return asset
+
+
+def _build_youtube_download_failure_log(
+    *,
+    task_id: uuid.UUID,
+    url: str,
+    error_message: str,
+    hint: str | None,
+    diagnostics: list[str] | None,
+) -> str:
+    body = "\n".join([str(line or "") for line in (diagnostics or [])]).strip()
+    sections = [
+        body or "videoroll yt-dlp diagnostics unavailable",
+        "\n".join(
+            [
+                "---- videoroll error summary ----",
+                f"task_id={task_id}",
+                f"url={url}",
+                f"error={error_message}",
+            ]
+        ),
+    ]
+    hint_text = str(hint or "").strip()
+    if hint_text:
+        sections.append(f"---- videoroll hint ----\n{hint_text}")
+    return "\n\n".join(sections).rstrip() + "\n"
+
+
+def _youtube_download_log_key(task_id: uuid.UUID) -> str:
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    return f"log/{task_id}/youtube_download_{stamp}_{uuid.uuid4().hex[:8]}.log"
+
+
+def _extract_metadata_title(raw: bytes) -> str:
+    if not raw:
+        return ""
+    try:
+        parsed = json.loads(raw.decode("utf-8"))
+    except Exception:
+        return ""
+    info = parsed if isinstance(parsed, dict) else {}
+    return str(info.get("title") or info.get("fulltitle") or info.get("alt_title") or "").strip()
+
+
+def _task_title_key(task_id: uuid.UUID) -> str:
+    return f"task.title.{task_id}"
+
+
+def _load_task_display_titles(
+    db: Session,
+    task_ids: list[uuid.UUID],
+    *,
+    s3: S3Store | None = None,
+    allow_s3_fallback: bool,
+) -> dict[uuid.UUID, str]:
+    title_map: dict[uuid.UUID, str] = {}
+    if not task_ids:
+        return title_map
+
+    rows = db.query(AppSetting).filter(AppSetting.key.in_([_task_title_key(tid) for tid in task_ids])).all()
+    by_key = {str(r.key): _as_dict(getattr(r, "value_json", None)) for r in rows}
+    for tid in task_ids:
+        data = by_key.get(_task_title_key(tid)) or {}
+        for key in ("translated_title", "source_title"):
+            value = data.get(key)
+            if isinstance(value, str) and value.strip():
+                title_map[tid] = value.strip()
+                break
+
+    if not allow_s3_fallback or s3 is None:
+        return title_map
+
+    missing = [tid for tid in task_ids if tid not in title_map]
+    if not missing:
+        return title_map
+
+    assets = (
+        db.query(Asset)
+        .filter(Asset.task_id.in_(missing), Asset.kind == AssetKind.metadata_json)
+        .order_by(Asset.created_at.desc())
+        .all()
+    )
+    picked: dict[uuid.UUID, Asset] = {}
+    for asset in assets:
+        if asset.task_id not in picked:
+            picked[asset.task_id] = asset
+    for tid, asset in picked.items():
+        try:
+            title = _extract_metadata_title(_read_s3_bytes(s3, asset.storage_key))
+        except Exception:
+            continue
+        if title:
+            title_map[tid] = title
+    return title_map
+
+
+def _stream_upload_to_tempfile(
+    file_obj: Any,
+    *,
+    prefix: str,
+    suffix: str,
+) -> tuple[Path, str, int]:
+    tmp_path: Path | None = None
+    sha256 = hashlib.sha256()
+    size_bytes = 0
+    try:
+        with tempfile.NamedTemporaryFile(prefix=prefix, suffix=suffix, delete=False) as tmp:
+            tmp_path = Path(tmp.name)
+            while True:
+                chunk = file_obj.read(1024 * 1024)
+                if not chunk:
+                    break
+                if not isinstance(chunk, (bytes, bytearray)):
+                    raise TypeError("uploaded file stream returned non-bytes content")
+                sha256.update(chunk)
+                size_bytes += len(chunk)
+                tmp.write(chunk)
+        return tmp_path, sha256.hexdigest(), size_bytes
+    except Exception:
+        if tmp_path is not None:
+            try:
+                tmp_path.unlink(missing_ok=True)
+            except Exception:
+                pass
+        raise
+
+
+def _safe_unlink(path: Path | None) -> None:
+    if path is None:
+        return
+    try:
+        path.unlink(missing_ok=True)
+    except Exception:
+        pass
+
+
+async def _store_uploaded_task_asset(
+    *,
+    task: Task,
+    file: UploadFile,
+    s3: S3Store,
+    db: Session,
+    temp_prefix: str,
+    default_suffix: str,
+    key_prefix: str,
+    object_name_prefix: str,
+    asset_kind: AssetKind,
+    update_task_status: TaskStatus | None = None,
+) -> Asset:
+    suffix = Path(file.filename or "").suffix or default_suffix
+    tmp_path: Path | None = None
+    uploaded_key: str | None = None
+
+    try:
+        await file.seek(0)
+        tmp_path, sha256, size_bytes = await run_in_threadpool(
+            _stream_upload_to_tempfile,
+            file.file,
+            prefix=temp_prefix,
+            suffix=suffix,
+        )
+        uploaded_key = f"{key_prefix}/{task.id}/{object_name_prefix}_{sha256[:16]}{suffix}"
+        await run_in_threadpool(s3.upload_file, tmp_path, uploaded_key, file.content_type or None)
+
+        asset = Asset(
+            task_id=task.id,
+            kind=asset_kind,
+            storage_key=uploaded_key,
+            sha256=sha256,
+            size_bytes=size_bytes,
+        )
+        db.add(asset)
+        if update_task_status is not None:
+            task.status = update_task_status
+            db.add(task)
+        db.commit()
+        db.refresh(asset)
+        return asset
+    except HTTPException:
+        raise
+    except Exception as e:
+        db.rollback()
+        if uploaded_key:
+            try:
+                await run_in_threadpool(s3.delete_object, uploaded_key)
+            except Exception:
+                logger.exception("failed to roll back uploaded S3 object", extra={"storage_key": uploaded_key})
+        raise HTTPException(status_code=500, detail=f"upload failed: {e}") from e
+    finally:
+        await run_in_threadpool(_safe_unlink, tmp_path)
+        try:
+            await file.close()
+        except Exception:
+            pass
+
+
 def _internal_header_token(settings: OrchestratorSettings) -> str:
     return internal_api_token(settings.s3_secret_access_key)
 
@@ -334,6 +569,8 @@ class _AdminAuthMiddleware(BaseHTTPMiddleware):
             return await call_next(request)
         if path.startswith("/auth"):
             return await call_next(request)
+        if path.rstrip("/") == REMOTE_AUTO_YOUTUBE_PATH:
+            return await call_next(request)
 
         pw_hash = _get_admin_password_hash(request)
         if not pw_hash:
@@ -422,48 +659,59 @@ def _cleanup_storage_once(settings: OrchestratorSettings) -> dict[str, int]:
 
         cutoff = _utcnow() - timedelta(days=ttl_days)
         store = S3Store(settings)
-        store.ensure_bucket()
 
         deleted_objects = 0
         deleted_assets = 0
         deleted_subtitles = 0
+        failed_keys: set[str] = set()
 
         # 1) Delete fully-expired keys (no Asset rows >= cutoff).
         while True:
-            rows = (
-                db.query(Asset.storage_key)
-                .group_by(Asset.storage_key)
-                .having(func.max(Asset.created_at) < cutoff)
-                .limit(200)
-                .all()
-            )
+            query = db.query(Asset.storage_key).group_by(Asset.storage_key).having(func.max(Asset.created_at) < cutoff)
+            if failed_keys:
+                query = query.filter(~Asset.storage_key.in_(failed_keys))
+            rows = query.limit(200).all()
             keys = [r[0] for r in rows if r and r[0]]
             if not keys:
                 break
 
+            deleted_keys: list[str] = []
             for key in keys:
                 try:
                     store.delete_object(key)
                     deleted_objects += 1
+                    deleted_keys.append(key)
                 except Exception:
-                    # Best-effort: still delete DB rows to honor TTL.
-                    pass
+                    failed_keys.add(key)
+                    logger.exception("failed to delete expired S3 object", extra={"storage_key": key})
 
-            deleted_assets += (
-                db.query(Asset)
-                .filter(Asset.storage_key.in_(keys))
-                .delete(synchronize_session=False)
-            )
-            deleted_subtitles += (
-                db.query(Subtitle)
-                .filter(Subtitle.storage_key.in_(keys))
-                .delete(synchronize_session=False)
+            if not deleted_keys:
+                continue
+
+            deleted_assets += db.query(Asset).filter(Asset.storage_key.in_(deleted_keys)).delete(synchronize_session=False)
+            deleted_subtitles += db.query(Subtitle).filter(Subtitle.storage_key.in_(deleted_keys)).delete(
+                synchronize_session=False
             )
             db.commit()
 
-        # 2) Delete old duplicate rows for keys that still exist (object kept).
-        deleted_assets += db.query(Asset).filter(Asset.created_at < cutoff).delete(synchronize_session=False)
-        deleted_subtitles += db.query(Subtitle).filter(Subtitle.created_at < cutoff).delete(synchronize_session=False)
+        # 2) Delete old duplicate rows only for keys that still have a newer Asset row.
+        active_key_subq = (
+            db.query(Asset.storage_key.label("storage_key"))
+            .group_by(Asset.storage_key)
+            .having(func.max(Asset.created_at) >= cutoff)
+            .subquery()
+        )
+        active_key_query = db.query(active_key_subq.c.storage_key)
+        deleted_assets += (
+            db.query(Asset)
+            .filter(Asset.created_at < cutoff, Asset.storage_key.in_(active_key_query))
+            .delete(synchronize_session=False)
+        )
+        deleted_subtitles += (
+            db.query(Subtitle)
+            .filter(Subtitle.created_at < cutoff, Subtitle.storage_key.in_(active_key_query))
+            .delete(synchronize_session=False)
+        )
         db.commit()
 
         return {
@@ -481,7 +729,7 @@ def _cleanup_loop() -> None:
         try:
             _cleanup_storage_once(settings)
         except Exception:
-            pass
+            logger.exception("storage cleanup loop failed")
         _cleanup_stop.wait(timeout=max(30, _cleanup_interval_seconds))
 
 
@@ -694,22 +942,25 @@ def auth_logout(request: Request, db: Session = Depends(get_db)) -> JSONResponse
     return resp
 
 
-@app.post("/auto/youtube", response_model=AutoYouTubeResponse)
-def auto_youtube(
-    payload: AutoYouTubeRequest,
-    settings: OrchestratorSettings = Depends(get_settings),
+def _start_auto_youtube_pipeline(
+    *,
+    url: str,
+    license: SourceLicense,
+    proof_url: str | None,
+    settings: OrchestratorSettings,
 ) -> AutoYouTubeResponse:
-    url = (payload.url or "").strip()
+    url = str(url or "").strip()
     if not url:
         raise HTTPException(status_code=400, detail="url is required")
     if not _is_youtube_url(url):
         raise HTTPException(status_code=400, detail="url is not a valid youtube url")
+    proof = str(proof_url or "").strip() or None
 
     try:
         with httpx.Client(timeout=30.0, headers=_internal_http_headers(settings)) as client:
             resp = client.post(
                 f"{settings.youtube_ingest_url}/youtube/ingest",
-                json={"url": url, "license": payload.license.value, "proof_url": payload.proof_url},
+                json={"url": url, "license": license.value, "proof_url": proof},
             )
             resp.raise_for_status()
             data = resp.json()
@@ -734,6 +985,41 @@ def auto_youtube(
     )
 
 
+@app.post("/auto/youtube", response_model=AutoYouTubeResponse)
+def auto_youtube(
+    payload: AutoYouTubeRequest,
+    settings: OrchestratorSettings = Depends(get_settings),
+) -> AutoYouTubeResponse:
+    return _start_auto_youtube_pipeline(
+        url=payload.url,
+        license=payload.license,
+        proof_url=payload.proof_url,
+        settings=settings,
+    )
+
+
+@app.get(REMOTE_AUTO_YOUTUBE_PATH, response_model=AutoYouTubeResponse)
+@app.post(REMOTE_AUTO_YOUTUBE_PATH, response_model=AutoYouTubeResponse)
+def remote_auto_youtube(
+    url: str | None = Query(default=None),
+    token: str | None = Query(default=None, alias=REMOTE_API_TOKEN_QUERY_PARAM),
+    license: SourceLicense = Query(default=SourceLicense.authorized),
+    proof_url: str | None = Query(default=None),
+    settings: OrchestratorSettings = Depends(get_settings),
+    db: Session = Depends(get_db),
+) -> AutoYouTubeResponse:
+    if not remote_api_token_is_configured(db):
+        raise HTTPException(status_code=403, detail="remote api token is not set")
+    if not verify_remote_api_token(db, str(token or "")):
+        raise HTTPException(status_code=401, detail="invalid remote api token")
+    return _start_auto_youtube_pipeline(
+        url=str(url or ""),
+        license=license,
+        proof_url=proof_url,
+        settings=settings,
+    )
+
+
 @app.post("/tasks", response_model=TaskRead)
 def create_task(payload: TaskCreate, db: Session = Depends(get_db)) -> Task:
     task = Task(
@@ -755,7 +1041,6 @@ def create_task(payload: TaskCreate, db: Session = Depends(get_db)) -> Task:
 def list_converted_videos(
     limit: int = Query(default=50, ge=1, le=200),
     db: Session = Depends(get_db),
-    s3: S3Store = Depends(get_s3),
 ) -> list[dict[str, Any]]:
     # Fetch recent video_final assets and dedupe by task_id (keep latest per task).
     fetch_n = min(1000, max(limit, 1) * 5)
@@ -767,25 +1052,48 @@ def list_converted_videos(
         .all()
     )
 
-    items: list[dict[str, Any]] = []
+    final_assets: list[Asset] = []
     seen: set[uuid.UUID] = set()
-    for a in assets:
-        if a.task_id in seen:
+    for asset in assets:
+        if asset.task_id in seen:
             continue
-        seen.add(a.task_id)
-        task = db.get(Task, a.task_id)
+        seen.add(asset.task_id)
+        final_assets.append(asset)
+        if len(final_assets) >= limit:
+            break
+
+    task_ids = [asset.task_id for asset in final_assets]
+    if not task_ids:
+        return []
+
+    task_map = {task.id: task for task in db.query(Task).filter(Task.id.in_(task_ids)).all()}
+    cover_assets = (
+        db.query(Asset)
+        .filter(Asset.task_id.in_(task_ids), Asset.kind == AssetKind.cover_image)
+        .order_by(Asset.created_at.desc())
+        .all()
+    )
+    cover_by_task: dict[uuid.UUID, Asset] = {}
+    for asset in cover_assets:
+        if asset.task_id not in cover_by_task:
+            cover_by_task[asset.task_id] = asset
+
+    title_map = _load_task_display_titles(db, task_ids, allow_s3_fallback=False)
+
+    items: list[dict[str, Any]] = []
+    for asset in final_assets:
+        task = task_map.get(asset.task_id)
         if not task:
             continue
-        cover_asset = (
-            db.query(Asset)
-            .filter(Asset.task_id == a.task_id, Asset.kind == AssetKind.cover_image)
-            .order_by(Asset.created_at.desc())
-            .first()
+        display_title = str(title_map.get(asset.task_id) or "").strip()
+        items.append(
+            {
+                "task": task,
+                "final_asset": asset,
+                "cover_asset": cover_by_task.get(asset.task_id),
+                "display_title": display_title or None,
+            }
         )
-        display_title = get_task_display_title_with_s3(db, str(a.task_id), s3=s3)
-        items.append({"task": task, "final_asset": a, "cover_asset": cover_asset, "display_title": display_title or None})
-        if len(items) >= limit:
-            break
     return items
 
 
@@ -796,55 +1104,13 @@ def list_tasks(
     db: Session = Depends(get_db),
     s3: S3Store = Depends(get_s3),
 ) -> list[dict[str, Any]]:
-    def _title_key(tid: uuid.UUID) -> str:
-        return f"task.title.{tid}"
-
     q = db.query(Task).order_by(Task.created_at.desc())
     if status is not None:
         q = q.filter(Task.status == status)
     tasks = q.limit(limit).all()
 
     ids = [t.id for t in tasks]
-
-    # Batch load translated/source titles from AppSetting first, then fall back to YouTube meta title.
-    title_map: dict[uuid.UUID, str] = {}
-    if ids:
-        keys = [_title_key(tid) for tid in ids]
-        rows = db.query(AppSetting).filter(AppSetting.key.in_(keys)).all()
-        by_key: dict[str, dict[str, Any]] = {}
-        for r in rows:
-            by_key[str(r.key)] = _as_dict(getattr(r, "value_json", None))
-
-        for tid in ids:
-            data = by_key.get(_title_key(tid)) or {}
-            for k in ("translated_title", "source_title"):
-                v = data.get(k)
-                if isinstance(v, str) and v.strip():
-                    title_map[tid] = v.strip()
-                    break
-
-        missing = [tid for tid in ids if tid not in title_map]
-        if missing:
-            assets = (
-                db.query(Asset)
-                .filter(Asset.task_id.in_(missing), Asset.kind == AssetKind.metadata_json)
-                .order_by(Asset.created_at.desc())
-                .all()
-            )
-            picked: dict[uuid.UUID, Asset] = {}
-            for a in assets:
-                if a.task_id not in picked:
-                    picked[a.task_id] = a
-            for tid, asset in picked.items():
-                try:
-                    raw = _read_s3_bytes(s3, asset.storage_key)
-                    parsed = json.loads(raw.decode("utf-8")) if raw else {}
-                    info = parsed if isinstance(parsed, dict) else {}
-                except Exception:
-                    continue
-                title = str(info.get("title") or info.get("fulltitle") or info.get("alt_title") or "").strip()
-                if title:
-                    title_map[tid] = title
+    title_map = _load_task_display_titles(db, ids, s3=s3, allow_s3_fallback=True)
 
     out: list[dict[str, Any]] = []
     for t in tasks:
@@ -870,6 +1136,17 @@ def put_storage_settings(payload: StorageRetentionSettingsUpdate, db: Session =
     return StorageRetentionSettingsRead(**cfg)
 
 
+@app.get("/settings/api", response_model=RemoteAPISettingsRead)
+def get_remote_api_settings_view(db: Session = Depends(get_db)) -> RemoteAPISettingsRead:
+    return RemoteAPISettingsRead(**get_remote_api_settings(db))
+
+
+@app.put("/settings/api", response_model=RemoteAPISettingsRead)
+def put_remote_api_settings_view(payload: RemoteAPISettingsUpdate, db: Session = Depends(get_db)) -> RemoteAPISettingsRead:
+    cfg = update_remote_api_settings(db, payload.model_dump(exclude_unset=True))
+    return RemoteAPISettingsRead(**cfg)
+
+
 @app.get("/settings/youtube", response_model=YouTubeSettingsRead)
 def get_youtube_settings_view(settings: OrchestratorSettings = Depends(get_settings), db: Session = Depends(get_db)) -> YouTubeSettingsRead:
     cfg = get_youtube_settings(db, default_proxy=settings.youtube_proxy)
@@ -973,39 +1250,18 @@ async def upload_task_video(
     task = db.get(Task, task_id)
     if not task:
         raise HTTPException(status_code=404, detail="task not found")
-
-    suffix = Path(file.filename or "").suffix or ".mp4"
-
-    with tempfile.NamedTemporaryFile(prefix="videoroll_", suffix=suffix, delete=False) as tmp:
-        tmp_path = Path(tmp.name)
-        while True:
-            chunk = await file.read(1024 * 1024)
-            if not chunk:
-                break
-            tmp.write(chunk)
-
-    sha256 = sha256_file(tmp_path)
-    size_bytes = tmp_path.stat().st_size
-    key = f"raw/{task_id}/video_{sha256[:16]}{suffix}"
-    s3.upload_file(tmp_path, key)
-    try:
-        tmp_path.unlink(missing_ok=True)
-    except Exception:
-        pass
-
-    asset = Asset(
-        task_id=task_id,
-        kind=AssetKind.video_raw,
-        storage_key=key,
-        sha256=sha256,
-        size_bytes=size_bytes,
+    return await _store_uploaded_task_asset(
+        task=task,
+        file=file,
+        s3=s3,
+        db=db,
+        temp_prefix="videoroll_",
+        default_suffix=".mp4",
+        key_prefix="raw",
+        object_name_prefix="video",
+        asset_kind=AssetKind.video_raw,
+        update_task_status=TaskStatus.downloaded,
     )
-    db.add(asset)
-    task.status = TaskStatus.downloaded
-    db.add(task)
-    db.commit()
-    db.refresh(asset)
-    return asset
 
 
 @app.post("/tasks/{task_id}/upload/cover", response_model=AssetRead)
@@ -1021,37 +1277,17 @@ async def upload_task_cover(
 
     if file.content_type and not file.content_type.startswith("image/"):
         raise HTTPException(status_code=400, detail="cover must be an image")
-
-    suffix = Path(file.filename or "").suffix or ".jpg"
-
-    with tempfile.NamedTemporaryFile(prefix="videoroll_cover_", suffix=suffix, delete=False) as tmp:
-        tmp_path = Path(tmp.name)
-        while True:
-            chunk = await file.read(1024 * 1024)
-            if not chunk:
-                break
-            tmp.write(chunk)
-
-    sha256 = sha256_file(tmp_path)
-    size_bytes = tmp_path.stat().st_size
-    key = f"final/{task_id}/cover_{sha256[:16]}{suffix}"
-    s3.upload_file(tmp_path, key)
-    try:
-        tmp_path.unlink(missing_ok=True)
-    except Exception:
-        pass
-
-    asset = Asset(
-        task_id=task_id,
-        kind=AssetKind.cover_image,
-        storage_key=key,
-        sha256=sha256,
-        size_bytes=size_bytes,
+    return await _store_uploaded_task_asset(
+        task=task,
+        file=file,
+        s3=s3,
+        db=db,
+        temp_prefix="videoroll_cover_",
+        default_suffix=".jpg",
+        key_prefix="final",
+        object_name_prefix="cover",
+        asset_kind=AssetKind.cover_image,
     )
-    db.add(asset)
-    db.commit()
-    db.refresh(asset)
-    return asset
 
 
 @app.get("/tasks/{task_id}/assets", response_model=list[AssetRead])
@@ -1376,6 +1612,24 @@ def download_youtube(
                 video_path, info, meta = download_youtube_video(url, yt_settings, work_dir=tmp_dir)
             except Exception as e:
                 hint = _youtube_bot_check_hint(str(e), yt_settings=yt_settings, db=db)
+                diagnostics = e.diagnostics if isinstance(e, YtDlpRuntimeError) else []
+                try:
+                    _store_task_log_asset(
+                        db,
+                        s3,
+                        task_id=task_id,
+                        log_key=_youtube_download_log_key(task_id),
+                        text=_build_youtube_download_failure_log(
+                            task_id=task_id,
+                            url=url,
+                            error_message=str(e),
+                            hint=hint,
+                            diagnostics=diagnostics,
+                        ),
+                    )
+                except Exception:
+                    db.rollback()
+                    logger.exception("failed to persist youtube download diagnostics log for task %s", task_id)
                 detail = f"youtube download failed: {e}"
                 if hint:
                     detail = f"{detail}\n\n{hint}"

src/videoroll/apps/orchestrator_api/remote_api_settings_store.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from videoroll.apps.orchestrator_api.admin_auth_store import encode_password_hash, verify_password_hash
+from videoroll.db.models import AppSetting
+
+
+REMOTE_API_SETTINGS_KEY = "remote.api"
+
+REMOTE_AUTO_YOUTUBE_PATH = "/remote/auto/youtube"
+REMOTE_API_TOKEN_QUERY_PARAM = "token"
+REMOTE_AUTO_URL_QUERY_PARAM = "url"
+REMOTE_AUTO_LICENSE_QUERY_PARAM = "license"
+REMOTE_AUTO_PROOF_URL_QUERY_PARAM = "proof_url"
+
+_MIN_TOKEN_LEN = 8
+_MAX_TOKEN_LEN = 512
+
+
+def _as_dict(v: Any) -> dict[str, Any]:
+    return v if isinstance(v, dict) else {}
+
+
+def _get_row(db: Session) -> AppSetting:
+    row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY)
+    if row:
+        return row
+    row = AppSetting(key=REMOTE_API_SETTINGS_KEY, value_json={})
+    db.add(row)
+    try:
+        db.commit()
+        db.refresh(row)
+    except IntegrityError:
+        db.rollback()
+        row2 = db.get(AppSetting, REMOTE_API_SETTINGS_KEY)
+        if row2:
+            return row2
+        raise
+    return row
+
+
+def _normalize_token(token: Any) -> str:
+    value = str(token or "").strip()
+    if not value:
+        return ""
+    if len(value) < _MIN_TOKEN_LEN:
+        raise ValueError(f"token too short (min {_MIN_TOKEN_LEN} chars)")
+    if len(value) > _MAX_TOKEN_LEN:
+        raise ValueError(f"token too long (max {_MAX_TOKEN_LEN} chars)")
+    return value
+
+
+def get_remote_api_settings(db: Session) -> dict[str, Any]:
+    row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY)
+    stored = dict(_as_dict(row.value_json)) if row else {}
+    token_hash = str(stored.get("token_hash") or "").strip()
+    token_updated_at = str(stored.get("token_updated_at") or "").strip() or None
+    return {
+        "token_set": bool(token_hash),
+        "token_updated_at": token_updated_at,
+        "endpoint_path": REMOTE_AUTO_YOUTUBE_PATH,
+        "token_query_param": REMOTE_API_TOKEN_QUERY_PARAM,
+        "url_query_param": REMOTE_AUTO_URL_QUERY_PARAM,
+        "license_query_param": REMOTE_AUTO_LICENSE_QUERY_PARAM,
+        "proof_url_query_param": REMOTE_AUTO_PROOF_URL_QUERY_PARAM,
+    }
+
+
+def update_remote_api_settings(db: Session, update: dict[str, Any]) -> dict[str, Any]:
+    row = _get_row(db)
+    stored = dict(_as_dict(row.value_json))
+
+    if "token" in update:
+        token_raw = update.get("token")
+        if token_raw is not None:
+            token = _normalize_token(token_raw)
+            if token:
+                stored["token_hash"] = encode_password_hash(token)
+                stored["token_updated_at"] = datetime.now(timezone.utc).isoformat()
+            else:
+                stored.pop("token_hash", None)
+                stored.pop("token_updated_at", None)
+
+    row.value_json = stored
+    db.add(row)
+    db.commit()
+
+    return get_remote_api_settings(db)
+
+
+def remote_api_token_is_configured(db: Session) -> bool:
+    row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY)
+    stored = dict(_as_dict(row.value_json)) if row else {}
+    return bool(str(stored.get("token_hash") or "").strip())
+
+
+def verify_remote_api_token(db: Session, token: str) -> bool:
+    value = str(token or "").strip()
+    if not value:
+        return False
+    row = db.get(AppSetting, REMOTE_API_SETTINGS_KEY)
+    stored = dict(_as_dict(row.value_json)) if row else {}
+    token_hash = str(stored.get("token_hash") or "").strip()
+    if not token_hash:
+        return False
+    return verify_password_hash(value, token_hash)

src/videoroll/apps/orchestrator_api/schemas.py

@@ -188,6 +188,20 @@ class StorageRetentionSettingsUpdate(BaseModel):
     asset_ttl_days: Optional[int] = None
 
 
+class RemoteAPISettingsRead(BaseModel):
+    token_set: bool = False
+    token_updated_at: Optional[str] = None
+    endpoint_path: str = "/remote/auto/youtube"
+    token_query_param: str = "token"
+    url_query_param: str = "url"
+    license_query_param: str = "license"
+    proof_url_query_param: str = "proof_url"
+
+
+class RemoteAPISettingsUpdate(BaseModel):
+    token: Optional[str] = None
+
+
 class YouTubeSettingsRead(BaseModel):
     proxy: str = ""
     cookies_set: bool = False

src/videoroll/apps/orchestrator_api/youtube_downloader.py

@@ -6,7 +6,7 @@ import subprocess
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Optional
-from urllib.parse import urlparse
+from urllib.parse import urlparse, urlsplit, urlunsplit
 
 import httpx
 import yt_dlp
@@ -25,6 +25,12 @@ class YouTubeMeta:
     duration: Optional[int] = None
 
 
+class YtDlpRuntimeError(RuntimeError):
+    def __init__(self, message: str, *, diagnostics: list[str] | None = None) -> None:
+        super().__init__(message)
+        self.diagnostics = diagnostics or []
+
+
 def _run(cmd: list[str]) -> None:
     subprocess.run(cmd, check=True)
 
@@ -66,10 +72,85 @@ def _extractor_args(settings: OrchestratorSettings) -> dict[str, Any] | None:
     return parsed
 
 
-def build_ydl_opts(settings: OrchestratorSettings, *, outtmpl: str | None = None, for_download: bool) -> dict[str, Any]:
+def _redact_url(value: str) -> str:
+    raw = str(value or "").strip()
+    if not raw:
+        return ""
+    try:
+        parts = urlsplit(raw)
+    except Exception:
+        return raw
+    hostname = parts.hostname or ""
+    if not hostname:
+        return raw
+    auth = ""
+    if parts.username:
+        auth = parts.username
+        if parts.password is not None:
+            auth += ":***"
+        auth += "@"
+    netloc = auth + hostname
+    if parts.port is not None:
+        netloc += f":{parts.port}"
+    return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
+
+
+def _sanitize_diagnostic_line(line: str, *, proxy: str | None) -> str:
+    text = str(line or "").rstrip()
+    if proxy:
+        text = text.replace(proxy, _redact_url(proxy))
+    return text
+
+
+def _yt_dlp_diagnostics_header(
+    *,
+    url: str,
+    settings: OrchestratorSettings,
+    outtmpl: str | None,
+) -> list[str]:
+    extractor_args = _extractor_args(settings)
+    return [
+        "videoroll yt-dlp diagnostics",
+        f"url={url}",
+        f"proxy={_redact_url(str(settings.youtube_proxy or '').strip()) or '(none)'}",
+        f"cookiefile={(str(settings.youtube_cookie_file or '').strip() or '(none)')}",
+        f"ffmpeg_location={(str(settings.ffmpeg_path or '').strip() or '(default PATH)')}",
+        f"outtmpl={outtmpl or '(none)'}",
+        f"extractor_args={json.dumps(extractor_args, ensure_ascii=False, sort_keys=True) if extractor_args else '(none)'}",
+        "---- yt-dlp verbose ----",
+    ]
+
+
+class _DiagnosticsLogger:
+    def __init__(self, diagnostics: list[str], *, proxy: str | None) -> None:
+        self._diagnostics = diagnostics
+        self._proxy = str(proxy or "").strip() or None
+
+    def debug(self, msg: str) -> None:
+        self._diagnostics.append(_sanitize_diagnostic_line(f"[debug] {msg}", proxy=self._proxy))
+
+    def info(self, msg: str) -> None:
+        self._diagnostics.append(_sanitize_diagnostic_line(str(msg), proxy=self._proxy))
+
+    def warning(self, msg: str) -> None:
+        self._diagnostics.append(_sanitize_diagnostic_line(f"WARNING: {msg}", proxy=self._proxy))
+
+    def error(self, msg: str) -> None:
+        self._diagnostics.append(_sanitize_diagnostic_line(f"ERROR: {msg}", proxy=self._proxy))
+
+
+def build_ydl_opts(
+    settings: OrchestratorSettings,
+    *,
+    outtmpl: str | None = None,
+    for_download: bool,
+    logger: Any | None = None,
+    verbose: bool = False,
+) -> dict[str, Any]:
     opts: dict[str, Any] = {
         "quiet": True,
         "no_warnings": True,
+        "ignoreconfig": True,
         "noplaylist": True,
         "retries": 3,
         "fragment_retries": 3,
@@ -103,6 +184,10 @@ def build_ydl_opts(settings: OrchestratorSettings, *, outtmpl: str | None = None
 
     if outtmpl:
         opts["outtmpl"] = outtmpl
+    if logger is not None:
+        opts["logger"] = logger
+    if verbose:
+        opts["verbose"] = True
 
     return opts
 
@@ -284,10 +369,20 @@ def _resolve_downloaded_file(info: dict[str, Any], *, work_dir: Path) -> Path:
     raise RuntimeError("yt-dlp download succeeded but no output file was found")
 
 
-def _download_once(url: str, settings: OrchestratorSettings, *, work_dir: Path) -> tuple[Path, dict[str, Any], YouTubeMeta]:
+def _download_once(
+    url: str,
+    settings: OrchestratorSettings,
+    *,
+    work_dir: Path,
+    diagnostics: list[str] | None = None,
+) -> tuple[Path, dict[str, Any], YouTubeMeta]:
     work_dir.mkdir(parents=True, exist_ok=True)
     outtmpl = str(work_dir / "%(id)s.%(ext)s")
-    with yt_dlp.YoutubeDL(build_ydl_opts(settings, outtmpl=outtmpl, for_download=True)) as ydl:
+    ydl_logger = None
+    if diagnostics is not None:
+        diagnostics.extend(_yt_dlp_diagnostics_header(url=url, settings=settings, outtmpl=outtmpl))
+        ydl_logger = _DiagnosticsLogger(diagnostics, proxy=settings.youtube_proxy)
+    with yt_dlp.YoutubeDL(build_ydl_opts(settings, outtmpl=outtmpl, for_download=True, logger=ydl_logger, verbose=diagnostics is not None)) as ydl:
         info_raw = ydl.extract_info(url, download=True)
         info = _pick_video_info(info_raw)
         if not info:
@@ -306,9 +401,10 @@ def download_youtube_video(url: str, settings: OrchestratorSettings, *, work_dir
       - sanitized info dict
       - meta summary
     """
+    diagnostics: list[str] = []
     try:
-        return _download_once(url, settings, work_dir=work_dir)
+        return _download_once(url, settings, work_dir=work_dir, diagnostics=diagnostics)
     except DownloadError as e:
-        raise RuntimeError(str(e)) from e
+        raise YtDlpRuntimeError(str(e), diagnostics=diagnostics) from e
     except Exception as e:
-        raise RuntimeError(str(e)) from e
+        raise YtDlpRuntimeError(str(e), diagnostics=diagnostics) from e

src/videoroll/apps/subtitle_service/main.py

@@ -632,8 +632,29 @@ def _read_task_queue(db: Session, *, limit: int) -> TaskQueueRead:
         )
 
     unlocked = (Task.lock_owner != TASK_QUEUE_LOCK_OWNER) | (Task.lock_until.is_(None)) | (Task.lock_until <= now)
-    # queued_count: distinct tasks with queued subtitle/render jobs (best-effort, capped).
-    queued_task_ids: set[uuid.UUID] = set()
+    orphaned_render_by_task: dict[uuid.UUID, RenderJob] = {}
+    orphaned_subtitle_by_task: dict[uuid.UUID, SubtitleJob] = {}
+    for rj in (
+        db.query(RenderJob)
+        .join(Task, Task.id == RenderJob.task_id)
+        .filter(RenderJob.status == RenderJobStatus.running, unlocked)
+        .order_by(RenderJob.updated_at.asc(), RenderJob.created_at.asc())
+        .limit(5000)
+        .all()
+    ):
+        orphaned_render_by_task.setdefault(rj.task_id, rj)
+    for sj in (
+        db.query(SubtitleJob)
+        .join(Task, Task.id == SubtitleJob.task_id)
+        .filter(SubtitleJob.status == SubtitleJobStatus.running, unlocked)
+        .order_by(SubtitleJob.updated_at.asc(), SubtitleJob.created_at.asc())
+        .limit(5000)
+        .all()
+    ):
+        orphaned_subtitle_by_task.setdefault(sj.task_id, sj)
+
+    # queued_count: distinct tasks with queued jobs, plus orphaned running jobs that lost their task lock.
+    queued_task_ids: set[uuid.UUID] = set(orphaned_render_by_task) | set(orphaned_subtitle_by_task)
     for tid, in (
         db.query(SubtitleJob.task_id)
         .join(Task, Task.id == SubtitleJob.task_id)
@@ -659,33 +680,74 @@ def _read_task_queue(db: Session, *, limit: int) -> TaskQueueRead:
     queued_items: list[TaskQueueItemRead] = []
     seen: set[uuid.UUID] = set()
 
-    fetch_n = min(5000, max(50, remaining * 20))
-    for sj in (
-        db.query(SubtitleJob)
-        .join(Task, Task.id == SubtitleJob.task_id)
-        .filter(SubtitleJob.status == SubtitleJobStatus.queued, unlocked)
-        .order_by(SubtitleJob.created_at.asc())
-        .limit(fetch_n)
-        .all()
-    ):
-        if sj.task_id in seen:
+    for rj in orphaned_render_by_task.values():
+        if rj.task_id in seen:
             continue
-        seen.add(sj.task_id)
+        seen.add(rj.task_id)
         queued_items.append(
             TaskQueueItemRead(
-                task_id=sj.task_id,
+                task_id=rj.task_id,
                 state="queued",
-                stage="subtitle",
-                subtitle_job_id=sj.id,
-                progress=int(sj.progress or 0),
-                error_message=sj.error_message,
-                created_at=sj.created_at,
-                updated_at=sj.updated_at,
+                stage="recover_render",
+                render_job_id=rj.id,
+                subtitle_job_id=rj.subtitle_job_id,
+                progress=int(rj.progress or 0),
+                error_message=rj.error_message,
+                created_at=rj.created_at,
+                updated_at=rj.updated_at,
             )
         )
         if len(queued_items) >= remaining:
             break
 
+    if len(queued_items) < remaining:
+        for sj in orphaned_subtitle_by_task.values():
+            if sj.task_id in seen:
+                continue
+            seen.add(sj.task_id)
+            queued_items.append(
+                TaskQueueItemRead(
+                    task_id=sj.task_id,
+                    state="queued",
+                    stage="recover_subtitle",
+                    subtitle_job_id=sj.id,
+                    progress=int(sj.progress or 0),
+                    error_message=sj.error_message,
+                    created_at=sj.created_at,
+                    updated_at=sj.updated_at,
+                )
+            )
+            if len(queued_items) >= remaining:
+                break
+
+    fetch_n = min(5000, max(50, remaining * 20))
+    if len(queued_items) < remaining:
+        for sj in (
+            db.query(SubtitleJob)
+            .join(Task, Task.id == SubtitleJob.task_id)
+            .filter(SubtitleJob.status == SubtitleJobStatus.queued, unlocked)
+            .order_by(SubtitleJob.created_at.asc())
+            .limit(fetch_n)
+            .all()
+        ):
+            if sj.task_id in seen:
+                continue
+            seen.add(sj.task_id)
+            queued_items.append(
+                TaskQueueItemRead(
+                    task_id=sj.task_id,
+                    state="queued",
+                    stage="subtitle",
+                    subtitle_job_id=sj.id,
+                    progress=int(sj.progress or 0),
+                    error_message=sj.error_message,
+                    created_at=sj.created_at,
+                    updated_at=sj.updated_at,
+                )
+            )
+            if len(queued_items) >= remaining:
+                break
+
     if len(queued_items) < remaining:
         for rj in (
             db.query(RenderJob)

src/videoroll/apps/subtitle_service/worker.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import logging
+import os
 import re
 import shutil
 import threading
@@ -70,6 +71,8 @@ settings = get_subtitle_settings()
 _ORCH_INTERNAL_TOKEN = internal_api_token(settings.s3_secret_access_key)
 _ORCH_INTERNAL_HEADERS = {INTERNAL_TOKEN_HEADER: _ORCH_INTERNAL_TOKEN} if _ORCH_INTERNAL_TOKEN else {}
 logger = logging.getLogger(__name__)
+_DB_READY_LOCK = threading.Lock()
+_DB_READY_PID: int | None = None
 celery_app = Celery("subtitle_service", broker=settings.redis_url, backend=settings.redis_url)
 celery_app.conf.update(
     task_serializer="json",
@@ -144,9 +147,17 @@ def _db() -> Session:
 
 
 def _ensure_db() -> None:
-    engine = get_engine(settings.database_url)
-    Base.metadata.create_all(engine)
-    auto_migrate(settings.database_url)
+    global _DB_READY_PID
+    pid = os.getpid()
+    if _DB_READY_PID == pid:
+        return
+    with _DB_READY_LOCK:
+        if _DB_READY_PID == pid:
+            return
+        engine = get_engine(settings.database_url)
+        Base.metadata.create_all(engine)
+        auto_migrate(settings.database_url)
+        _DB_READY_PID = pid
 
 
 def _now() -> datetime:
@@ -173,6 +184,18 @@ def _task_queue_unlock(task: Task) -> None:
     task.lock_until = None
 
 
+def _task_queue_join_message(message: str | None, detail: str, *, limit: int = 2000) -> str:
+    head = str(message or "").strip()
+    tail = str(detail or "").strip()
+    if head and tail:
+        out = f"{head}\n{tail}"
+    else:
+        out = head or tail
+    if len(out) > limit:
+        out = out[: limit - 1] + "…"
+    return out
+
+
 def _task_queue_lock_settings_row(db: Session) -> None:
     """
     Serialize queue ticks by locking the settings row.
@@ -1178,6 +1201,53 @@ def task_queue_tick() -> dict[str, Any]:
         except Exception:
             unlocked_expired = 0
 
+        unlocked = or_(
+            Task.lock_owner != TASK_QUEUE_LOCK_OWNER,
+            Task.lock_until.is_(None),
+            Task.lock_until <= now,
+        )
+
+        # Recover jobs that were actively running but lost their task slot/heartbeat.
+        # This is the "ghost running" case: task disappears from the queue while the job row
+        # still says running forever.
+        orphaned_subtitle = (
+            db.query(SubtitleJob)
+            .join(Task, Task.id == SubtitleJob.task_id)
+            .filter(SubtitleJob.status == SubtitleJobStatus.running, unlocked)
+            .order_by(SubtitleJob.updated_at.asc(), SubtitleJob.created_at.asc())
+            .with_for_update(skip_locked=True)
+            .all()
+        )
+        for j in orphaned_subtitle:
+            detail = "检测到任务队列锁已丢失：subtitle job 已自动按 resume 方式重新排队。"
+            req = dict(j.request_json) if isinstance(j.request_json, dict) else {}
+            req["resume"] = True
+            j.request_json = req
+            j.error_message = _task_queue_join_message(j.error_message, detail)
+            j.status = SubtitleJobStatus.queued
+            j.progress = 0
+            db.add(j)
+            recovered_subtitle += 1
+
+        orphaned_render = (
+            db.query(RenderJob)
+            .join(Task, Task.id == RenderJob.task_id)
+            .filter(RenderJob.status == RenderJobStatus.running, unlocked)
+            .order_by(RenderJob.updated_at.asc(), RenderJob.created_at.asc())
+            .with_for_update(skip_locked=True)
+            .all()
+        )
+        for j in orphaned_render:
+            detail = "检测到任务队列锁已丢失：render job 已自动重新排队。"
+            j.error_message = _task_queue_join_message(j.error_message, detail)
+            j.retry_count = int(j.retry_count or 0) + 1
+            j.status = RenderJobStatus.queued
+            j.progress = 0
+            j.started_at = None
+            j.finished_at = None
+            db.add(j)
+            recovered_render += 1
+
         # Recover render jobs that were marked as running by the scheduler but never claimed by a worker.
         cutoff = now - timedelta(seconds=60)
         # Subtitle jobs: scheduler sets progress=1; worker bumps it to >=2 once claimed.

src/videoroll/db/session.py

@@ -4,13 +4,25 @@ from functools import lru_cache
 from typing import Generator
 
 from sqlalchemy import create_engine
-from sqlalchemy.engine import Engine
+from sqlalchemy.engine import Engine, make_url
 from sqlalchemy.orm import Session, sessionmaker
 
 
+def _engine_connect_args(database_url: str) -> dict[str, object]:
+    try:
+        url = make_url(database_url)
+    except Exception:
+        return {}
+    if url.get_backend_name() == "postgresql" and url.get_driver_name() == "psycopg":
+        # PgBouncer/transaction-pooling setups can fail with
+        # "prepared statement already exists" when psycopg auto-prepares queries.
+        return {"prepare_threshold": None}
+    return {}
+
+
 @lru_cache
 def get_engine(database_url: str) -> Engine:
-    return create_engine(database_url, pool_pre_ping=True)
+    return create_engine(database_url, pool_pre_ping=True, connect_args=_engine_connect_args(database_url))
 
 
 def get_sessionmaker(database_url: str) -> sessionmaker[Session]:
@@ -25,4 +37,3 @@ def db_session(database_url: str) -> Generator[Session, None, None]:
         yield db
     finally:
         db.close()
-

src/videoroll/storage/s3.py

@@ -42,11 +42,31 @@ class S3Store:
     def bucket(self) -> str:
         return self._bucket
 
+    @staticmethod
+    def _is_missing_bucket_error(exc: ClientError) -> bool:
+        err = exc.response.get("Error") or {}
+        code = str(err.get("Code") or "").strip()
+        status = int((exc.response.get("ResponseMetadata") or {}).get("HTTPStatusCode") or 0)
+        return code in {"404", "NoSuchBucket", "NotFound"} or status == 404
+
+    @staticmethod
+    def _is_bucket_already_exists_error(exc: ClientError) -> bool:
+        err = exc.response.get("Error") or {}
+        code = str(err.get("Code") or "").strip()
+        return code in {"BucketAlreadyOwnedByYou", "BucketAlreadyExists"}
+
     def ensure_bucket(self) -> None:
         try:
             self._client.head_bucket(Bucket=self._bucket)
-        except ClientError:
-            self._client.create_bucket(Bucket=self._bucket)
+        except ClientError as e:
+            if not self._is_missing_bucket_error(e):
+                raise
+            try:
+                self._client.create_bucket(Bucket=self._bucket)
+            except ClientError as create_error:
+                if self._is_bucket_already_exists_error(create_error):
+                    return
+                raise
 
     def upload_file(self, path: Path, key: str, content_type: Optional[str] = None) -> PutResult:
         if content_type is None:

src/web/src/App.tsx

@@ -12,6 +12,7 @@ import YouTubeSourcesPage from "./pages/YouTubeSourcesPage";
 import SettingsASRPage from "./pages/SettingsASRPage";
 import SettingsYouTubePage from "./pages/SettingsYouTubePage";
 import SettingsStoragePage from "./pages/SettingsStoragePage";
+import SettingsApiPage from "./pages/SettingsApiPage";
 import SettingsTranslatePage from "./pages/SettingsTranslatePage";
 import SettingsBilibiliPage from "./pages/SettingsBilibiliPage";
 import SettingsAutoPage from "./pages/SettingsAutoPage";
@@ -97,6 +98,7 @@ export default function App() {
               <NavItem to="/settings/asr" label="Settings · ASR" />
               <NavItem to="/settings/youtube" label="Settings · YouTube" />
               <NavItem to="/settings/storage" label="Settings · Storage" />
+              <NavItem to="/settings/api" label="Settings · API" />
               <NavItem to="/settings/auto" label="Settings · Auto" />
               <NavItem to="/settings/translate" label="Settings · Translate" />
               <NavItem to="/settings/bilibili" label="Settings · Bilibili" />
@@ -118,6 +120,7 @@ export default function App() {
               <Route path="/settings/asr" element={<SettingsASRPage />} />
               <Route path="/settings/youtube" element={<SettingsYouTubePage />} />
               <Route path="/settings/storage" element={<SettingsStoragePage />} />
+              <Route path="/settings/api" element={<SettingsApiPage />} />
               <Route path="/settings/auto" element={<SettingsAutoPage />} />
               <Route path="/settings/translate" element={<SettingsTranslatePage />} />
               <Route path="/settings/bilibili" element={<SettingsBilibiliPage />} />

src/web/src/lib/http.ts

@@ -2,7 +2,18 @@ export async function fetchJson<T>(url: string, init?: RequestInit): Promise<T>
   const resp = await fetch(url, { credentials: "include", ...init });
   if (!resp.ok) {
     const text = await resp.text().catch(() => "");
-    throw new Error(`${resp.status} ${resp.statusText}${text ? ` - ${text}` : ""}`);
+    let detail = text;
+    if (text) {
+      try {
+        const parsed = JSON.parse(text) as { detail?: unknown; message?: unknown };
+        if (typeof parsed.detail === "string" && parsed.detail.trim()) {
+          detail = parsed.detail.trim();
+        } else if (typeof parsed.message === "string" && parsed.message.trim()) {
+          detail = parsed.message.trim();
+        }
+      } catch {}
+    }
+    throw new Error(`${resp.status} ${resp.statusText}${detail ? ` - ${detail}` : ""}`);
   }
   return (await resp.json()) as T;
 }

src/web/src/pages/SettingsApiPage.tsx

@@ -0,0 +1,193 @@
+import { useEffect, useMemo, useState } from "react";
+import { fetchJson } from "../lib/http";
+import { ORCHESTRATOR_URL } from "../lib/urls";
+
+type RemoteApiSettings = {
+  token_set: boolean;
+  token_updated_at?: string | null;
+  endpoint_path: string;
+  token_query_param: string;
+  url_query_param: string;
+  license_query_param: string;
+  proof_url_query_param: string;
+};
+
+const DEFAULT_SETTINGS: RemoteApiSettings = {
+  token_set: false,
+  token_updated_at: null,
+  endpoint_path: "/remote/auto/youtube",
+  token_query_param: "token",
+  url_query_param: "url",
+  license_query_param: "license",
+  proof_url_query_param: "proof_url",
+};
+
+function orchestratorBaseUrl(): string {
+  const raw = ORCHESTRATOR_URL.replace(/\/+$/, "");
+  if (raw.startsWith("http://") || raw.startsWith("https://")) return raw;
+  if (typeof window === "undefined") return raw;
+  return new URL(raw || "/api", window.location.origin).toString().replace(/\/+$/, "");
+}
+
+export default function SettingsApiPage() {
+  const [settings, setSettings] = useState<RemoteApiSettings | null>(null);
+  const [tokenInput, setTokenInput] = useState("");
+  const [error, setError] = useState<string | null>(null);
+  const [busy, setBusy] = useState(false);
+
+  async function refresh() {
+    setError(null);
+    try {
+      const s = await fetchJson<RemoteApiSettings>(`${ORCHESTRATOR_URL}/settings/api`);
+      setSettings(s);
+    } catch (e: unknown) {
+      setError(e instanceof Error ? e.message : String(e));
+    }
+  }
+
+  useEffect(() => {
+    refresh();
+  }, []);
+
+  const effective = settings ?? DEFAULT_SETTINGS;
+  const baseUrl = useMemo(() => orchestratorBaseUrl(), []);
+  const remoteEndpoint = useMemo(() => `${baseUrl}${effective.endpoint_path}`, [baseUrl, effective.endpoint_path]);
+  const sampleUrl = useMemo(() => {
+    const qs = new URLSearchParams();
+    qs.set(effective.token_query_param, tokenInput.trim() || "YOUR_TOKEN");
+    qs.set(effective.url_query_param, "https://www.youtube.com/watch?v=dQw4w9WgXcQ");
+    qs.set(effective.license_query_param, "authorized");
+    return `${remoteEndpoint}?${qs.toString()}`;
+  }, [effective.license_query_param, effective.token_query_param, effective.url_query_param, remoteEndpoint, tokenInput]);
+  const sampleCurl = useMemo(
+    () =>
+      [
+        `curl -G "${remoteEndpoint}" \\`,
+        `  --data-urlencode "${effective.token_query_param}=${tokenInput.trim() || "YOUR_TOKEN"}" \\`,
+        `  --data-urlencode "${effective.url_query_param}=https://www.youtube.com/watch?v=dQw4w9WgXcQ" \\`,
+        `  --data-urlencode "${effective.license_query_param}=authorized"`,
+      ].join("\n"),
+    [effective.license_query_param, effective.token_query_param, effective.url_query_param, remoteEndpoint, tokenInput],
+  );
+
+  return (
+    <div className="space-y-4">
+      <div className="rounded border bg-white p-4">
+        <div className="text-lg font-semibold">Settings · API</div>
+        <div className="mt-1 text-sm text-slate-600">配置远程管理 token。外部请求带上 token 和 YouTube `url` 参数后，会直接按 Auto Mode 自动创建任务并开始处理。</div>
+        {error ? <div className="mt-3 whitespace-pre-wrap break-words text-sm text-rose-700">{error}</div> : null}
+      </div>
+
+      <div className="rounded border bg-white p-4">
+        <div className="flex items-center justify-between">
+          <div className="text-sm font-semibold">当前配置</div>
+          <button onClick={() => refresh()} className="rounded border px-3 py-2 text-sm hover:bg-slate-50">
+            刷新
+          </button>
+        </div>
+        {!settings ? (
+          <div className="mt-2 text-sm text-slate-500">加载中…</div>
+        ) : (
+          <div className="mt-3 grid gap-3 md:grid-cols-2">
+            <div className="rounded border p-3">
+              <div className="text-xs text-slate-500">remote token</div>
+              <div className="mt-1 text-sm">{settings.token_set ? "已设置" : "未设置"}</div>
+            </div>
+            <div className="rounded border p-3">
+              <div className="text-xs text-slate-500">token_updated_at</div>
+              <div className="mt-1 font-mono text-sm">{settings.token_updated_at || "-"}</div>
+            </div>
+            <div className="rounded border p-3 md:col-span-2">
+              <div className="text-xs text-slate-500">remote endpoint</div>
+              <div className="mt-1 break-all font-mono text-sm">{remoteEndpoint}</div>
+            </div>
+          </div>
+        )}
+        <div className="mt-3 text-xs text-slate-500">说明：当前远程入口仅支持 YouTube 视频链接，行为等价于网页里的 “YouTube 自动模式”。</div>
+      </div>
+
+      <div className="rounded border bg-white p-4">
+        <div className="text-sm font-semibold">保存 token</div>
+        <div className="mt-2 grid gap-3 md:grid-cols-2">
+          <label className="block md:col-span-2">
+            <div className="mb-1 text-xs text-slate-600">token（最少 8 个字符）</div>
+            <input
+              className="w-full rounded border px-3 py-2 text-sm"
+              value={tokenInput}
+              onChange={(e) => setTokenInput(e.target.value)}
+              placeholder={settings?.token_set ? "已设置（输入新 token 会覆盖旧值）" : "输入新的远程调用 token"}
+            />
+          </label>
+        </div>
+
+        <div className="mt-3 flex flex-wrap items-center gap-2">
+          <button
+            disabled={busy || !tokenInput.trim()}
+            className="rounded bg-slate-900 px-3 py-2 text-sm text-white hover:bg-slate-800 disabled:opacity-50"
+            onClick={async () => {
+              setBusy(true);
+              setError(null);
+              try {
+                await fetchJson(`${ORCHESTRATOR_URL}/settings/api`, {
+                  method: "PUT",
+                  headers: { "Content-Type": "application/json" },
+                  body: JSON.stringify({ token: tokenInput.trim() }),
+                });
+                setTokenInput("");
+                await refresh();
+              } catch (e: unknown) {
+                setError(e instanceof Error ? e.message : String(e));
+              } finally {
+                setBusy(false);
+              }
+            }}
+          >
+            保存
+          </button>
+          <button
+            disabled={busy || !settings?.token_set}
+            className="rounded border px-3 py-2 text-sm hover:bg-slate-50 disabled:opacity-50"
+            onClick={async () => {
+              if (!window.confirm("确认清空远程管理 token？清空后外部将无法再调用该接口。")) return;
+              setBusy(true);
+              setError(null);
+              try {
+                await fetchJson(`${ORCHESTRATOR_URL}/settings/api`, {
+                  method: "PUT",
+                  headers: { "Content-Type": "application/json" },
+                  body: JSON.stringify({ token: "" }),
+                });
+                setTokenInput("");
+                await refresh();
+              } catch (e: unknown) {
+                setError(e instanceof Error ? e.message : String(e));
+              } finally {
+                setBusy(false);
+              }
+            }}
+          >
+            清空 token
+          </button>
+        </div>
+
+        <div className="mt-3 text-xs text-slate-500">提示：保存后不会回显明文 token。请自行保管；需要更换时直接输入新 token 覆盖即可。</div>
+      </div>
+
+      <div className="rounded border bg-white p-4">
+        <div className="text-sm font-semibold">调用示例</div>
+        <div className="mt-2 text-xs text-slate-500">
+          必填参数：<span className="font-mono">{effective.token_query_param}</span>、<span className="font-mono">{effective.url_query_param}</span>
+          。可选参数：<span className="font-mono">{effective.license_query_param}</span>、<span className="font-mono">{effective.proof_url_query_param}</span>。
+        </div>
+        <div className="mt-3 rounded border bg-slate-50 p-3">
+          <div className="text-xs text-slate-500">示例 URL</div>
+          <pre className="mt-2 whitespace-pre-wrap break-all font-mono text-xs text-slate-800">{sampleUrl}</pre>
+        </div>
+        <div className="mt-3 rounded border bg-slate-50 p-3">
+          <div className="text-xs text-slate-500">curl</div>
+          <pre className="mt-2 whitespace-pre-wrap break-all font-mono text-xs text-slate-800">{sampleCurl}</pre>
+        </div>
+      </div>
+    </div>
+  );
+}

src/web/src/pages/TaskDetailPage.tsx

@@ -427,8 +427,13 @@ export default function TaskDetailPage() {
   const finalAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "video_final"), [assets]);
   const coverAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "cover_image"), [assets]);
   const logAssets = useMemo(() => (assets ?? []).filter((x) => x.kind === "log"), [assets]);
+  const youtubeDownloadLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/youtube_download_")), [logAssets]);
   const subtitleLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/subtitle_")), [logAssets]);
   const renderLogAssets = useMemo(() => logAssets.filter((x) => x.storage_key.includes("/render_")), [logAssets]);
+  const latestYouTubeDownloadLog = useMemo(
+    () => (youtubeDownloadLogAssets.length ? youtubeDownloadLogAssets[youtubeDownloadLogAssets.length - 1] : null),
+    [youtubeDownloadLogAssets],
+  );
   const latestSubtitleLog = useMemo(
     () => (subtitleLogAssets.length ? subtitleLogAssets[subtitleLogAssets.length - 1] : null),
     [subtitleLogAssets],
@@ -541,6 +546,10 @@ export default function TaskDetailPage() {
       await refresh();
       alert(`${opts.resume ? "已继续字幕任务" : "已提交字幕任务"}：${resp.job_id}`);
     } catch (e: unknown) {
+      try {
+        await refresh({ silent: true });
+        await loadLogs({ silent: true });
+      } catch {}
       setError(e instanceof Error ? e.message : String(e));
     } finally {
       setBusy(false);
@@ -578,6 +587,9 @@ export default function TaskDetailPage() {
         const maxBytes = 200_000;
         if (logSelection === "combined") {
           const parts: Array<{ title: string; asset: Asset }> = [];
+          if (latestYouTubeDownloadLog) {
+            parts.push({ title: `YouTube Download · ${latestYouTubeDownloadLog.storage_key}`, asset: latestYouTubeDownloadLog });
+          }
           if (latestSubtitleLog) parts.push({ title: `Subtitle · ${latestSubtitleLog.storage_key}`, asset: latestSubtitleLog });
           if (latestRenderLog) parts.push({ title: `Render · ${latestRenderLog.storage_key}`, asset: latestRenderLog });
           if (!parts.length) {
@@ -605,7 +617,7 @@ export default function TaskDetailPage() {
         if (!silent) setLogBusy(false);
       }
     },
-    [taskId, logSelection, latestSubtitleLog, latestRenderLog, selectedLogAsset],
+    [taskId, logSelection, latestYouTubeDownloadLog, latestSubtitleLog, latestRenderLog, selectedLogAsset],
   );
 
   useEffect(() => {
@@ -679,7 +691,7 @@ export default function TaskDetailPage() {
           </button>
         </div>
 
-        {error ? <div className="mt-3 text-sm text-rose-700">{error}</div> : null}
+        {error ? <div className="mt-3 whitespace-pre-wrap break-words text-sm text-rose-700">{error}</div> : null}
         {!task ? <div className="mt-3 text-sm text-slate-500">加载中…</div> : null}
         {task ? (
           <div className="mt-3 grid gap-3 md:grid-cols-2">
@@ -748,6 +760,10 @@ export default function TaskDetailPage() {
                   }
                   await refresh();
                 } catch (e: unknown) {
+                  try {
+                    await refresh({ silent: true });
+                    await loadLogs({ silent: true });
+                  } catch {}
                   setError(e instanceof Error ? e.message : String(e));
                 } finally {
                   setBusy(false);
@@ -996,7 +1012,7 @@ export default function TaskDetailPage() {
         </div>
         <div className="mt-2 flex flex-wrap items-center gap-2">
           <select className="rounded border px-3 py-2 text-sm" value={logSelection} onChange={(e) => setLogSelection(e.target.value)}>
-            <option value="combined">合并（最新字幕 + 压制）</option>
+            <option value="combined">合并（最新 YouTube 下载 + 字幕 + 压制）</option>
             {[...logAssets].reverse().map((a) => (
               <option key={a.id} value={a.id}>
                 {a.storage_key}