-
+
@@ -156,6 +190,31 @@ export const ScriptJobDetail = () => {
)}
+ {isDownloadProgress(job.progress) && (
+
+
+
+ {running && job.progress.phase === 'download' && job.progress.current && (
+
+
+ {job.progress.current}
+
+ )}
+
+ )}
+
users.id, { onDelete: 'cascade' }),
taskDirName: text('task_dir_name').notNull(),
taskName: text('task_name').notNull(),
- // Job kind — pipeline (multi-step agentic), script (single bash/py/ts task), download (video/audio via
- // ReClip — its own lane so a long download doesn't block agentic jobs), later agentic.
- mode: text('mode', { enum: ['pipeline', 'script', 'agentic', 'download'] })
+ // Job kind — pipeline (multi-step agentic), script (single bash/py/ts task), later agentic. (Legacy
+ // rows may still carry mode 'download' from before video/audio downloads moved to a script capability.)
+ mode: text('mode', { enum: ['pipeline', 'script', 'agentic'] })
.notNull()
.default('pipeline'),
status: text('status', { enum: ['pending', 'running', 'completed', 'failed', 'stopped', 'interrupted'] })
diff --git a/src/servers/api/tasks/execute-download.ts b/src/servers/api/tasks/execute-download.ts
deleted file mode 100644
index dc4c7763..00000000
--- a/src/servers/api/tasks/execute-download.ts
+++ /dev/null
@@ -1,163 +0,0 @@
-import { writeFile } from 'node:fs/promises';
-import { join } from 'node:path';
-import { reclipInfo, reclipPlaylist, reclipDownloadOne, type ReclipInfo } from '../../reclip-client';
-
-// Write a per-item description sidecar next to its media file — same base name, `.txt` (e.g.
-// "Song Name.mp3" → "Song Name.txt"). Best-effort; skipped when the description is empty.
-async function writeDescriptionSidecar(dir: string, mediaFilename: string, description: string): Promise {
- const desc = description.trim();
- if (!desc) return;
- const base = mediaFilename.replace(/\.[^./\\]+$/, ''); // strip the media extension only
- try {
- await writeFile(join(dir, `${base}.txt`), desc, 'utf8');
- } catch {
- /* best-effort — a missing sidecar shouldn't fail the download */
- }
-}
-
-const FS_ILLEGAL = /[/\\:*?"<>|\u0000-\u001f]/g;
-const safeName = (s: string) => s.replace(FS_ILLEGAL, '_').replace(/\s+/g, ' ').trim().slice(0, 180);
-
-// On a FAILED download there's no media file to pair with — still drop a reference so the missed video is
-// recoverable: a `.txt` (or `<videoId>.txt` when untitled) holding the URL + the description. Always
-// written (even with an empty description — the URL is the reference).
-async function writeMissedSidecar(
- dir: string,
- item: { url: string; title: string; description: string },
-): Promise<void> {
- const vid = item.url.match(/[?&]v=([\w-]+)/)?.[1];
- const name = safeName(item.title) || vid || 'missed';
- const content = [item.url, item.description.trim()].filter(Boolean).join('\n\n');
- try {
- await writeFile(join(dir, `${name}.txt`), content, 'utf8');
- } catch {
- /* best-effort */
- }
-}
-
-// The download-job executor — pure scripting, no agent. Two phases:
-// 1. metadata — fetch each item's info (title + validity); keep the ones that resolve, skip the errors
-// (private / deleted / unavailable). The title is required: ReClip names the output file
-// from it (no title → a hash filename), so we cannot skip this pass.
-// 2. download — download every survivor in the chosen format (audio/video), passing its title; skip
-// anything that fails.
-// The item list comes from the caller (inputs.urls — the exact list the panel expanded, so a Mix/radio
-// playlist that returns a different set each call can't drift); falls back to expanding inputs.url. 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; per-item errors are counted + skipped.
-
-type Counts = { done: number; failed: number; total: number };
-export type DownloadProgress = {
- phase: 'expanding' | 'metadata' | 'download' | 'done';
- meta: Counts; // done = kept (title fetched), failed = skipped
- dl: Counts; // done = saved, failed = failed
- 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>; // { urls (JSON) | 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, urls: urlsJson, format, absDir } = params.inputs;
- if (!absDir) throw new Error('download job missing 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);
-
- // ── Resolve the item list: the caller's exact list, else expand the url server-side ──
- let urls: string[];
- if (urlsJson) {
- urls = JSON.parse(urlsJson) as string[];
- } else if (url) {
- if (url.includes('list=')) {
- const pl = await reclipPlaylist(url);
- if (pl.error) throw new Error(pl.error);
- urls = pl.urls?.length ? pl.urls : [url];
- } else {
- urls = [url];
- }
- } else {
- throw new Error('download job missing url(s)');
- }
-
- // ── Phase 1: metadata (title + validity) ──
- progress.phase = 'metadata';
- progress.meta.total = urls.length;
- emit(true);
- const valid: Array<{ url: string; title: string; description: 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 ?? '', description: info.description ?? '' });
- progress.meta.done++;
- } else {
- progress.meta.failed++;
- }
- emit();
- }
- emit(true);
-
- // ── Phase 2: download survivors (title → real filename) ──
- 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,
- // Filename is known before the file transfers — write the description sidecar alongside it, in
- // parallel with the download stream. Same base name guarantees they pair up.
- onFilename: (filename) => void writeDescriptionSidecar(absDir, filename, item.description),
- });
- 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++;
- void writeMissedSidecar(absDir, item); // no media file — leave a reference to what we missed
- }
- emit();
- }
-
- progress.phase = 'done';
- progress.current = undefined;
- emit(true);
-}
diff --git a/src/servers/api/tasks/execute-script.ts b/src/servers/api/tasks/execute-script.ts
index a3a0809d..0b3c956a 100644
--- a/src/servers/api/tasks/execute-script.ts
+++ b/src/servers/api/tasks/execute-script.ts
@@ -12,9 +12,15 @@ export type ScriptEvent =
| { type: 'started'; taskName: string }
| { type: 'stdout'; data: string }
| { type: 'stderr'; data: string }
+ | { type: 'progress'; progress: unknown }
| { type: 'exit'; code: number }
| { type: 'error'; message: string };
+// A script can emit machine-readable progress on stdout: a line `@@officer:progress@@ {json}`. Those
+// lines are plucked out (parsed into a `progress` event, kept OUT of the human log); everything else
+// streams verbatim.
+const PROGRESS_SENTINEL = '@@officer:progress@@';
+
export type ExecuteScriptParams = {
jobId: string;
userId: number;
@@ -92,6 +98,23 @@ export async function executeScript(params: ExecuteScriptParams): Promise<{ exit
mkdirSync(join(DATA_PATH, 'jobs'), { recursive: true });
const log = createWriteStream(jobLogPath(jobId), { flags: 'w' });
+ // The pumps must keep draining the child's stdout no matter what: a throw from a viewer broadcast or
+ // a log-stream error must not stop consumption, or the child's stdout pipe backs up and the whole job
+ // silently wedges (log + progress freeze while it keeps running). So both side-effects are isolated.
+ const safeEmit = (event: ScriptEvent) => {
+ try {
+ emit(event);
+ } catch {
+ /* a dead viewer / broadcast error must never kill the pump */
+ }
+ };
+ const safeLog = (data: string) => {
+ try {
+ log.write(data);
+ } catch {
+ /* a log-stream error must never kill the pump */
+ }
+ };
const cleanup = () => {
try {
rmSync(join(scriptPath, '..'), { recursive: true, force: true });
@@ -116,24 +139,63 @@ export async function executeScript(params: ExecuteScriptParams): Promise<{ exit
}
}, 500);
- const decoder = new TextDecoder();
- const pump = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
+ // stderr: raw passthrough. Each pump owns its decoder (stateful across chunks; must not be shared).
+ const pumpRaw = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
+ const decoder = new TextDecoder();
try {
for (;;) {
const { done, value } = await reader.read();
if (done) break;
- const data = decoder.decode(value);
- emit({ type, data });
- log.write(data);
+ const data = decoder.decode(value, { stream: true });
+ if (!data) continue;
+ safeEmit({ type, data });
+ safeLog(data);
}
} catch {
// stream closed
}
};
+ // stdout: line-buffered so progress-sentinel lines can be extracted from the human log + stream.
+ const pumpStdout = async (reader: ReadableStreamDefaultReader<Uint8Array>) => {
+ const decoder = new TextDecoder();
+ let buf = '';
+ const handleLine = (line: string, hadNewline: boolean) => {
+ if (line.startsWith(PROGRESS_SENTINEL)) {
+ let parsed: unknown;
+ try {
+ parsed = JSON.parse(line.slice(PROGRESS_SENTINEL.length).trim());
+ } catch {
+ return; // malformed progress payload — drop it, don't leak the sentinel into the log
+ }
+ safeEmit({ type: 'progress', progress: parsed });
+ return;
+ }
+ const out = hadNewline ? `${line}\n` : line;
+ if (!out) return;
+ safeEmit({ type: 'stdout', data: out });
+ safeLog(out);
+ };
+ try {
+ for (;;) {
+ const { done, value } = await reader.read();
+ if (done) break;
+ buf += decoder.decode(value, { stream: true });
+ let nl: number;
+ while ((nl = buf.indexOf('\n')) !== -1) {
+ handleLine(buf.slice(0, nl), true);
+ buf = buf.slice(nl + 1);
+ }
+ }
+ } catch {
+ // stream closed
+ }
+ if (buf) handleLine(buf, false); // trailing partial line (no newline)
+ };
+
const [, , exitCode] = await Promise.all([
- pump(proc.stdout.getReader(), 'stdout'),
- pump(proc.stderr.getReader(), 'stderr'),
+ pumpStdout(proc.stdout.getReader()),
+ pumpRaw(proc.stderr.getReader(), 'stderr'),
proc.exited,
]);
diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts
index c3097b37..c954d6ca 100644
--- a/src/servers/api/tasks/pipeline-job-manager.ts
+++ b/src/servers/api/tasks/pipeline-job-manager.ts
@@ -18,16 +18,10 @@ 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, 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');
+// Everything a job can stream — pipeline structural events, script stdout/stderr/exit/progress.
+type JobEvent = OutMessage | ScriptEvent;
+type JobMode = 'pipeline' | 'script' | 'agentic';
type WSData = {
userId: number;
@@ -38,7 +32,6 @@ type WSData = {
type LiveJob = {
jobId: string;
userId: number;
- lane: Lane;
abortSignal: { aborted: boolean };
emitter: EventEmitter;
eventBuffer: JobEvent[];
@@ -53,15 +46,15 @@ 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) {
+ // readyState can flip to closing between this check and send (e.g. a page refresh), so send() may
+ // still throw. A dead viewer must never propagate into the job's output pump, so swallow it here.
if (ws.readyState === 1) {
- ws.send(JSON.stringify({ jobId, ...event }));
+ try {
+ ws.send(JSON.stringify({ jobId, ...event }));
+ } catch {
+ /* viewer went away mid-send */
+ }
}
}
@@ -107,8 +100,8 @@ export async function enqueueJob(
): Promise<{ jobId: string; status: 'running' | 'pending' }> {
const jobId = randomUUID();
const mode: JobMode = params.mode ?? 'pipeline';
- // Queue within the job's lane: run now if forced, or if that lane is idle.
- const run = action === 'start' || runningInLane(laneOf(mode)) === 0;
+ // One global queue (single user): run now if forced, or if nothing is currently running.
+ const run = action === 'start' || liveJobs.size === 0;
await createPipelineJob({
id: jobId,
userId: params.userId,
@@ -136,7 +129,6 @@ 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: [],
@@ -156,7 +148,7 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
job.progressDirty = true;
job.lastProgress = extractProgress(event, job.lastProgress);
}
- if (event.type === 'download:progress') {
+ if (event.type === 'progress') {
job.progressDirty = true;
job.lastProgress = event.progress;
}
@@ -195,41 +187,30 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
// Run the job in the background — dispatch by mode.
const runner: Promise<void | { exitCode: number }> =
- mode === 'download'
- ? executeDownload({
+ 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,
})
- : 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,
- });
+ : 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) => {
@@ -265,39 +246,33 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
});
}
-// 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.
+// When a job finishes (or on startup), launch the oldest queued job — one global queue, one job at a
+// time (single user). Rebuilds the user context from userId since there's no request here.
async function promoteNext(): Promise<void> {
+ if (liveJobs.size > 0) return; // something is already running
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
+ for (const next of pending) {
+ const nextMode = (next.mode as JobMode) ?? 'pipeline';
+ 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
}
+ await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() });
+ 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,
+ });
+ return; // one at a time
}
}
diff --git a/src/servers/api/tasks/pipeline-jobs-routes.ts b/src/servers/api/tasks/pipeline-jobs-routes.ts
index e50b36cb..7fea9c05 100644
--- a/src/servers/api/tasks/pipeline-jobs-routes.ts
+++ b/src/servers/api/tasks/pipeline-jobs-routes.ts
@@ -5,7 +5,6 @@ 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();
@@ -71,44 +70,6 @@ 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: { urls[] (the exact list to download — preferred), OR url (expanded server-side),
-// format:'audio'|'video', dir (target folder, home-relative), root?, label? }. Passing `urls` locks the
-// job to the list the panel already expanded, so a Mix/radio playlist can't drift between fetch + job.
-pipelineJobsRouter.post('/download', async (c) => {
- const user = c.get('user');
- const body = await c.req.json<{
- url?: string;
- urls?: string[];
- format?: 'audio' | 'video';
- dir?: string;
- root?: string;
- label?: string;
- }>();
- if (!body.url && !body.urls?.length) throw errors.BAD_REQUEST('url or urls is required');
- const rootDir = getRootDir(user, body.root);
- const absDir = resolveUserPath(rootDir, body.dir ?? '/'); // traversal-guarded
- const format = body.format === 'video' ? 'video' : 'audio';
- const inputs: Record<string, string> = body.urls?.length
- ? { urls: JSON.stringify(body.urls), format, absDir }
- : { url: body.url!, format, absDir };
- 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,
- 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) => {
diff --git a/src/servers/reclip-client.ts b/src/servers/reclip-client.ts
deleted file mode 100644
index 1135ca2a..00000000
--- a/src/servers/reclip-client.ts
+++ /dev/null
@@ -1,117 +0,0 @@
-import { join } from 'node:path';
-import { mkdir } from 'node:fs/promises';
-
-// Shared client for the ReClip download service (its own yt-dlp). The platform is a pure proxy: it never
-// runs yt-dlp itself. Used by the file-browser download endpoints AND the download-job executor.
-
-export const RECLIP_BASE = process.env.RECLIP_URL ?? 'http://localhost:8899';
-
-const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
-
-export type ReclipInfo = {
- title?: string;
- description?: string;
- thumbnail?: string;
- duration?: number;
- uploader?: string;
- formats?: Array<{ height?: number; id?: string; label?: string }>;
- error?: string;
-};
-
-/** Single-video metadata (title/thumbnail/duration/uploader/formats). Returns { error } inline. */
-export async function reclipInfo(url: string): Promise<ReclipInfo> {
- const res = await fetch(`${RECLIP_BASE}/api/info`, {
- method: 'POST',
- headers: { 'Content-Type': 'application/json' },
- body: JSON.stringify({ url }),
- signal: AbortSignal.timeout(90_000),
- }).catch(() => null);
- if (!res) return { error: `Could not reach ReClip at ${RECLIP_BASE}` };
- return (await res.json().catch(() => ({}))) as ReclipInfo;
-}
-
-/** Expand a playlist URL into its individual video URLs. */
-export async function reclipPlaylist(url: string): Promise<{ urls?: string[]; error?: string }> {
- const res = await fetch(`${RECLIP_BASE}/api/playlist`, {
- method: 'POST',
- headers: { 'Content-Type': 'application/json' },
- body: JSON.stringify({ url }),
- signal: AbortSignal.timeout(120_000),
- }).catch(() => null);
- if (!res) return { error: `Could not reach ReClip at ${RECLIP_BASE}` };
- return (await res.json().catch(() => ({}))) as { urls?: string[]; error?: string };
-}
-
-type DownloadOpts = {
- url: string;
- destDir: string; // absolute directory to write the finished file into
- audioOnly: boolean;
- title?: string; // optional override; omit and ReClip names the file from the video title itself
- onPhase?: (phase: 'transferring') => void;
- onFilename?: (filename: string) => void; // fired as soon as the final filename is known (before streaming)
- signal?: { aborted: boolean };
-};
-
-/**
- * Download ONE video/audio via ReClip and stream the finished file into `destDir`. Resolves with the
- * saved filename; throws on any failure (unreachable / rejected / job error / timeout / abort). Respects
- * a cooperative `signal.aborted` between polls and while streaming. No metadata prefetch — ReClip derives
- * the filename from the video title — so this is a single request per item.
- */
-export async function reclipDownloadOne(opts: DownloadOpts): Promise<string> {
- const { url, destDir, audioOnly, signal } = opts;
- const aborted = () => signal?.aborted === true;
-
- const dlRes = await fetch(`${RECLIP_BASE}/api/download`, {
- method: 'POST',
- headers: { 'Content-Type': 'application/json' },
- body: JSON.stringify({ url, format: audioOnly ? 'audio' : 'video', title: opts.title ?? '' }),
- signal: AbortSignal.timeout(30_000),
- }).catch(() => {
- throw new Error(`Could not reach ReClip at ${RECLIP_BASE}`);
- });
- if (!dlRes.ok) throw new Error('ReClip rejected the download request');
- const reclipJob = ((await dlRes.json()) as { job_id?: string }).job_id;
- if (!reclipJob) throw new Error('ReClip did not return a job id');
-
- // Poll ReClip until done/error (generous deadline; ReClip enforces its own per-download cap).
- const deadline = Date.now() + 60 * 60_000;
- let filename = '';
- for (;;) {
- if (aborted()) throw new Error('aborted');
- if (Date.now() > deadline) throw new Error('Download timed out');
- await sleep(2000);
- const stRes = await fetch(`${RECLIP_BASE}/api/status/${reclipJob}`, { signal: AbortSignal.timeout(15_000) }).catch(
- () => null,
- );
- if (!stRes?.ok) continue;
- const st = (await stRes.json()) as { status: string; error?: string | null; filename?: string | null };
- if (st.status === 'error') throw new Error(st.error || 'ReClip download failed');
- if (st.status === 'done') {
- filename = st.filename || `${reclipJob}.${audioOnly ? 'mp3' : 'mp4'}`;
- break;
- }
- }
-
- opts.onFilename?.(filename); // exact name known — the caller can write a sidecar while we stream
- opts.onPhase?.('transferring');
-
- // Stream the finished file into the destination folder (filename is title-sanitized by ReClip).
- const fileRes = await fetch(`${RECLIP_BASE}/api/file/${reclipJob}`, { signal: AbortSignal.timeout(600_000) });
- const body = fileRes.body;
- if (!fileRes.ok || !body) throw new Error('Failed to fetch the downloaded file from ReClip');
- await mkdir(destDir, { recursive: true });
- const sink = Bun.file(join(destDir, filename)).writer();
- const reader = body.getReader();
- try {
- for (;;) {
- if (aborted()) throw new Error('aborted');
- const { done, value } = await reader.read();
- if (done) break;
- sink.write(value);
- }
- } finally {
- await sink.end();
- }
- return filename;
-}
diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx
index 07c17cd1..bb011790 100644
--- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx
+++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx
@@ -1774,7 +1774,13 @@ export const TaskRunnerModal = ({
taskDirName={task.dirName}
autoInputs={autoInputs}
context={autofillContext}
- cwd={cwd.path || undefined}
+ cwd={
+ entryType === 'directory' && entryName
+ ? cwd.path
+ ? `${cwd.path}/${entryName}`
+ : entryName
+ : cwd.path || undefined
+ }
entryType={effectiveEntryType}
filePath={
multi
diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/VideoDownloadPanel.tsx b/src/workspaces/officerdev/src/apps/FileBrowser/VideoDownloadPanel.tsx
index ebe3ccd4..eb3e3a00 100644
--- a/src/workspaces/officerdev/src/apps/FileBrowser/VideoDownloadPanel.tsx
+++ b/src/workspaces/officerdev/src/apps/FileBrowser/VideoDownloadPanel.tsx
@@ -245,16 +245,17 @@ export const VideoDownloadPanel = () => {
setFetching(false);
};
- // Job path: hand the whole playlist to a server-side two-phase download job (own lane, survives the
- // panel closing). The server re-expands + fetches metadata (phase 1) then downloads survivors (phase 2).
+ // Job path: hand the whole playlist to the `download-media` script capability as a background job (it
+ // survives the panel closing). We pass the exact expanded video URLs — already individual, no `list=`,
+ // so the task won't re-expand and a Mix/radio playlist can't drift to a different set. cwd is
+ // home-relative (no leading slash) — the capability writes into it.
const startJob = async () => {
try {
- const res = await client.post<{ jobId: string; status: string }>('/jobs/download', {
- urls: expandedUrls, // the exact list we counted — a Mix playlist won't re-expand to a different set
- format: jobFormat,
- dir: targetDir(),
- root,
- label: `${jobFormat === 'audio' ? 'Audio' : 'Video'} · ${expandedUrls.length} items`,
+ const res = await client.post<{ jobId: string; status: string }>('/jobs', {
+ taskDirName: 'download-media',
+ inputs: { url: expandedUrls.join('\n'), format: jobFormat },
+ cwd: targetDir().replace(/^\/+/, ''),
+ action: 'queue',
});
lastDlDone.current = 0;
setJobId(res.jobId);