jobs: video/audio download as a two-phase job (backend)

Turns the downloader into a server-side job on the existing jobs spine (Postgres
persistence, live WS viewers + replay, abort, /jobs UI) — but with its own
executor and its own lane, since it's deterministic scripting, not an agent, and
a multi-hour playlist mustn't block agentic jobs.

- reclip-client.ts (new, shared): reclipInfo / reclipPlaylist / reclipDownloadOne
  (single download → streams the file to a dir, abort-aware). Extracted so both
  the file-browser endpoints and the job executor use one client.
- execute-download.ts (new): the two-phase executor —
  phase 1 metadata (expand playlist, fetch each info, keep survivors, skip
  errors), phase 2 download (each survivor in the chosen format; skip download
  errors). Emits a compact `download:progress` snapshot (counters, not per-item
  events — playlists are thousands of items). Throws on abort / fatal.
- job manager: `download` mode dispatch → executeDownload; persists
  download:progress; adds execution LANES (download vs default) so the two run
  independently and each serializes on its own; promoteNext fills both lanes.
- POST /api/tasks/jobs/download { url, format, dir, root?, label? } — enqueues a
  download job (own lane, no capability task needed; traversal-guarded target).
- schema: `download` added to the mode enum (drizzle text-enum — no DB migration);
  getPendingJobs() query for lane filling.

Verified the executor with a mocked ReClip client: two-phase filtering, skip-on-
error counts, and abort-throws all correct.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/servers/api/file-browser/router.ts

@@ -43,7 +43,7 @@ function getUserDataDir(email: string): string {
   return join(DATA_PATH, email);
 }
 
-function getRootDir(user: UserCtx, root?: string): string {
+export function getRootDir(user: UserCtx, root?: string): string {
   if (!root || root === 'home') return getOwnerHomeDir(user.email);
   if (root === 'user-data') return getUserDataDir(user.email);
   throw errors.BAD_REQUEST(`Invalid root: ${root}`);
@@ -54,7 +54,7 @@ function getRootDir(user: UserCtx, root?: string): string {
 // passes a `/home/br` check — which is how `..` segments escaped.
 const isInside = (root: string, target: string): boolean => target === root || target.startsWith(root + sep);
 
-function resolveUserPath(rootDir: string, relPath: string): string {
+export function resolveUserPath(rootDir: string, relPath: string): string {
   const resolved = resolve(rootDir, relPath.replace(/^\/+/, ''));
   if (!isInside(rootDir, resolved)) throw errors.FORBIDDEN('Path outside root directory');
   return resolved;

src/servers/api/tasks/execute-download.ts

@@ -0,0 +1,110 @@
+import { reclipInfo, reclipPlaylist, reclipDownloadOne, type ReclipInfo } from '../../reclip-client';
+
+// The download-job executor — pure scripting, no agent. Two phases:
+//   1. metadata — expand the playlist, fetch each item's info sequentially, keep the ones that resolve
+//                 (skip the errors: private/deleted/unavailable).
+//   2. download — download every survivor in the chosen format (audio/video); skip anything that fails.
+// Emits a compact `download:progress` snapshot (counters, not per-item events — a playlist can be
+// thousands of items). Throws on abort or a fatal error (playlist expansion) so the job manager marks it
+// stopped/failed; per-item errors are counted and skipped, never fatal.
+
+type Counts = { done: number; failed: number; total: number };
+export type DownloadProgress = {
+  phase: 'expanding' | 'metadata' | 'download' | 'done';
+  meta: Counts;
+  dl: Counts;
+  current?: string; // title of the item currently downloading
+};
+
+export type DownloadEvent = { type: 'download:progress'; progress: DownloadProgress };
+
+export type ExecuteDownloadParams = {
+  jobId: string;
+  userId: number;
+  email: string;
+  username?: string | null;
+  inputs: Record<string, string>; // { url, format: 'audio'|'video', absDir }
+  cwd?: string;
+  abortSignal: { aborted: boolean };
+  emit: (event: DownloadEvent) => void;
+};
+
+const EMIT_THROTTLE_MS = 750;
+
+export async function executeDownload(params: ExecuteDownloadParams): Promise<void> {
+  const { url, format, absDir } = params.inputs;
+  if (!url || !absDir) throw new Error('download job missing url or target directory');
+  const audioOnly = format !== 'video'; // default to audio
+
+  const progress: DownloadProgress = {
+    phase: 'expanding',
+    meta: { done: 0, failed: 0, total: 0 },
+    dl: { done: 0, failed: 0, total: 0 },
+  };
+  let lastEmit = 0;
+  const emit = (force = false) => {
+    const now = Date.now();
+    if (!force && now - lastEmit < EMIT_THROTTLE_MS) return;
+    lastEmit = now;
+    params.emit({
+      type: 'download:progress',
+      progress: { ...progress, meta: { ...progress.meta }, dl: { ...progress.dl } },
+    });
+  };
+  const checkAbort = () => {
+    if (params.abortSignal.aborted) throw new Error('aborted');
+  };
+
+  emit(true);
+
+  // ── Expand ──
+  let urls = [url];
+  if (url.includes('list=')) {
+    const pl = await reclipPlaylist(url);
+    if (pl.error) throw new Error(pl.error);
+    if (pl.urls?.length) urls = pl.urls;
+  }
+
+  // ── Phase 1: metadata (keep survivors) ──
+  progress.phase = 'metadata';
+  progress.meta.total = urls.length;
+  emit(true);
+  const valid: Array<{ url: string; title: string }> = [];
+  for (const u of urls) {
+    checkAbort();
+    const info = await reclipInfo(u).catch((): ReclipInfo => ({ error: 'fetch failed' }));
+    if (info && !info.error) valid.push({ url: u, title: info.title ?? '' });
+    else progress.meta.failed++;
+    progress.meta.done++;
+    emit();
+  }
+  emit(true);
+
+  // ── Phase 2: download survivors ──
+  progress.phase = 'download';
+  progress.dl.total = valid.length;
+  emit(true);
+  for (const item of valid) {
+    checkAbort();
+    progress.current = item.title || item.url;
+    emit(true);
+    try {
+      await reclipDownloadOne({
+        url: item.url,
+        destDir: absDir,
+        audioOnly,
+        title: item.title,
+        signal: params.abortSignal,
+      });
+      progress.dl.done++;
+    } catch {
+      checkAbort(); // an abort surfaces as a throw here — re-check so it stops instead of counting as a skip
+      progress.dl.failed++;
+    }
+    emit();
+  }
+
+  progress.phase = 'done';
+  progress.current = undefined;
+  emit(true);
+}

src/servers/api/tasks/pipeline-job-manager.ts

@@ -6,7 +6,7 @@ import {
   getPipelineJob,
   updatePipelineJob,
   getPipelineJobsForUser,
-  getOldestPendingJob,
+  getPendingJobs,
   countPendingJobs,
   deletePipelineJob,
   deleteTerminalJobsForUser,
@@ -18,10 +18,16 @@ import { toShellUsername } from '../../data-path';
 import { executePipeline } from './pipeline-executor';
 import type { OutMessage } from './pipeline-executor';
 import { executeScript, jobLogPath, type ScriptEvent } from './execute-script';
+import { executeDownload, type DownloadEvent } from './execute-download';
 
-// Everything a job can stream — pipeline structural events plus script stdout/stderr/exit.
-type JobEvent = OutMessage | ScriptEvent;
-type JobMode = 'pipeline' | 'script' | 'agentic';
+// Everything a job can stream — pipeline structural events, script stdout/stderr/exit, download progress.
+type JobEvent = OutMessage | ScriptEvent | DownloadEvent;
+type JobMode = 'pipeline' | 'script' | 'agentic' | 'download';
+
+// Execution lanes: downloads run independently of agentic/script jobs so a multi-hour playlist doesn't
+// block capability jobs. Each lane runs one job at a time (single user).
+type Lane = 'download' | 'default';
+const laneOf = (mode: JobMode): Lane => (mode === 'download' ? 'download' : 'default');
 
 type WSData = {
   userId: number;
@@ -32,6 +38,7 @@ type WSData = {
 type LiveJob = {
   jobId: string;
   userId: number;
+  lane: Lane;
   abortSignal: { aborted: boolean };
   emitter: EventEmitter;
   eventBuffer: JobEvent[];
@@ -46,6 +53,12 @@ const PROGRESS_FLUSH_MS = 3000;
 
 const liveJobs = new Map<string, LiveJob>();
 
+const runningInLane = (lane: Lane): number => {
+  let n = 0;
+  for (const job of liveJobs.values()) if (job.lane === lane) n++;
+  return n;
+};
+
 function sendToViewer(ws: ServerWebSocket<WSData>, jobId: string, event: JobEvent) {
   if (ws.readyState === 1) {
     ws.send(JSON.stringify({ jobId, ...event }));
@@ -94,7 +107,8 @@ export async function enqueueJob(
 ): Promise<{ jobId: string; status: 'running' | 'pending' }> {
   const jobId = randomUUID();
   const mode: JobMode = params.mode ?? 'pipeline';
-  const run = action === 'start' || runningCount() === 0;
+  // Queue within the job's lane: run now if forced, or if that lane is idle.
+  const run = action === 'start' || runningInLane(laneOf(mode)) === 0;
   await createPipelineJob({
     id: jobId,
     userId: params.userId,
@@ -122,6 +136,7 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
   const job: LiveJob = {
     jobId,
     userId: params.userId,
+    lane: laneOf(mode),
     abortSignal: { aborted: false },
     emitter: new EventEmitter(),
     eventBuffer: [],
@@ -141,6 +156,10 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
       job.progressDirty = true;
       job.lastProgress = extractProgress(event, job.lastProgress);
     }
+    if (event.type === 'download:progress') {
+      job.progressDirty = true;
+      job.lastProgress = event.progress;
+    }
     if (event.type === 'step:complete' || event.type === 'iteration:complete') {
       const cost = 'cost' in event ? event.cost : undefined;
       if (cost) {
@@ -176,30 +195,41 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
 
   // Run the job in the background — dispatch by mode.
   const runner: Promise<void | { exitCode: number }> =
-    mode === 'script'
-      ? executeScript({
+    mode === 'download'
+      ? executeDownload({
           jobId,
           userId: params.userId,
           email: params.email,
           username: params.username,
-          taskDirName: params.taskDirName,
           inputs: params.inputs,
           cwd: params.cwd,
           abortSignal: job.abortSignal,
           emit,
         })
-      : executePipeline({
-          userId: params.userId,
-          email: params.email,
-          username: params.username,
-          taskDirName: params.taskDirName,
-          inputs: params.inputs,
-          cwd: params.cwd,
-          model: params.model,
-          startAt: params.startAt,
-          abortSignal: job.abortSignal,
-          emit,
-        });
+      : mode === 'script'
+        ? executeScript({
+            jobId,
+            userId: params.userId,
+            email: params.email,
+            username: params.username,
+            taskDirName: params.taskDirName,
+            inputs: params.inputs,
+            cwd: params.cwd,
+            abortSignal: job.abortSignal,
+            emit,
+          })
+        : executePipeline({
+            userId: params.userId,
+            email: params.email,
+            username: params.username,
+            taskDirName: params.taskDirName,
+            inputs: params.inputs,
+            cwd: params.cwd,
+            model: params.model,
+            startAt: params.startAt,
+            abortSignal: job.abortSignal,
+            emit,
+          });
 
   runner
     .then(async (result) => {
@@ -235,32 +265,40 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
     });
 }
 
-// When a job finishes (and nothing else is running), promote the oldest queued job. Also called on
-// startup to resume a backlog. Rebuilds the user context from userId since there's no request here.
+// When a job finishes (or on startup), fill each idle lane with its oldest queued job. Rebuilds the user
+// context from userId since there's no request here. Lanes are independent — a running download doesn't
+// hold back an agentic job and vice-versa.
 async function promoteNext(): Promise<void> {
-  if (runningCount() > 0) return;
-  const next = await getOldestPendingJob();
-  if (!next) return;
-  const user = await getUserById(next.userId);
-  if (!user) {
-    await updatePipelineJob(next.id, { status: 'failed', error: 'user not found', completedAt: new Date() }).catch(
-      () => {},
-    );
-    return promoteNext();
+  const pending = await getPendingJobs(); // oldest first
+  const launchedNow = new Set<string>();
+  for (const lane of ['default', 'download'] as Lane[]) {
+    if (runningInLane(lane) > 0) continue;
+    for (const next of pending) {
+      const nextMode = (next.mode as JobMode) ?? 'pipeline';
+      if (laneOf(nextMode) !== lane || launchedNow.has(next.id)) continue;
+      const user = await getUserById(next.userId);
+      if (!user) {
+        await updatePipelineJob(next.id, { status: 'failed', error: 'user not found', completedAt: new Date() }).catch(
+          () => {},
+        );
+        continue; // skip to the next candidate in this lane
+      }
+      await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() });
+      launchedNow.add(next.id);
+      launch(next.id, nextMode, {
+        userId: next.userId,
+        email: user.email,
+        username: toShellUsername(user.username ?? '', user.email),
+        mode: nextMode,
+        taskDirName: next.taskDirName,
+        taskName: next.taskName,
+        inputs: next.inputs as Record<string, string>,
+        cwd: next.cwd ?? undefined,
+        config: next.config,
+      });
+      break; // lane filled
+    }
   }
-  await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() });
-  const nextMode = (next.mode as JobMode) ?? 'pipeline';
-  launch(next.id, nextMode, {
-    userId: next.userId,
-    email: user.email,
-    username: toShellUsername(user.username ?? '', user.email),
-    mode: nextMode,
-    taskDirName: next.taskDirName,
-    taskName: next.taskName,
-    inputs: next.inputs as Record<string, string>,
-    cwd: next.cwd ?? undefined,
-    config: next.config,
-  });
 }
 
 export function attachViewer(jobId: string, ws: ServerWebSocket<WSData>) {

src/servers/api/tasks/pipeline-jobs-routes.ts

@@ -5,6 +5,7 @@ import { getPipelineJobsForUser, getPipelineJob } from 'officerdb';
 import { getTaskByDirName } from './task-files';
 import * as jobManager from './pipeline-job-manager';
 import { jobLogPath } from './execute-script';
+import { getRootDir, resolveUserPath } from '../file-browser/router';
 
 export const pipelineJobsRouter = createRouter();
 
@@ -70,6 +71,39 @@ pipelineJobsRouter.post('/', async (c) => {
   return c.json({ jobId, status });
 });
 
+// POST /download — enqueue a video/audio download job (ReClip). Runs in its own lane, needs no capability
+// task. Body: { url (video or playlist), format:'audio'|'video', dir (target folder, home-relative),
+// root?, label? }. The job expands + fetches metadata (phase 1) then downloads survivors (phase 2).
+pipelineJobsRouter.post('/download', async (c) => {
+  const user = c.get('user');
+  const body = await c.req.json<{
+    url: string;
+    format?: 'audio' | 'video';
+    dir?: string;
+    root?: string;
+    label?: string;
+  }>();
+  if (!body.url) throw errors.BAD_REQUEST('url is required');
+  const rootDir = getRootDir(user, body.root);
+  const absDir = resolveUserPath(rootDir, body.dir ?? '/'); // traversal-guarded
+  const format = body.format === 'video' ? 'video' : 'audio';
+  const { jobId, status } = await jobManager.enqueueJob(
+    {
+      userId: user.id,
+      email: user.email,
+      username: user.username ?? '',
+      mode: 'download',
+      taskDirName: 'video-download',
+      taskName: body.label || (format === 'audio' ? 'Audio download' : 'Video download'),
+      inputs: { url: body.url, format, absDir },
+      cwd: body.dir ?? '/',
+      config: {},
+    },
+    'queue',
+  );
+  return c.json({ jobId, status });
+});
+
 // GET /counts — header-badge summary { running, runningJobId, queued }. Before /:id so it isn't
 // captured as an id.
 pipelineJobsRouter.get('/counts', async (c) => {