add tag-album and clean-playlist-files tasks, script step support, stop fix, rename fix

- New tag-album task: renames tracks to NNN format, sets ID3 tags via mutagen - New clean-playlist-files task: deletes .cue, .m3u, .nfo and similar junk files - Pipeline executor now supports script-mode steps (runs directly, no agent) - Build discography pipeline: convert-audio → clean-playlist-files → prepare → fetch → tag - Fix stop button: abort signal now kills running agent processes - Fix job manager: broadcast stopped/error events to WebSocket viewers - Fix file browser rename: delay focus to avoid context menu close race, left-align input Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-10 04:10:14 +00:00
parent 5e38343f44
commit 53c78c76cb
7 changed files with 367 additions and 15 deletions
@@ -1,7 +1,8 @@
 import type { ServerWebSocket } from 'bun';
 import { randomUUID } from 'crypto';
-import { readdirSync, existsSync } from 'node:fs';
+import { readdirSync, existsSync, mkdirSync, writeFileSync, chmodSync, rmSync } from 'node:fs';
 import { join } from 'node:path';
+import { tmpdir } from 'node:os';
 import { getTaskByDirName, getUserSettings } from 'officerdb';
 import { getHomeDirForRole, getHomeDir } from '../../data-path';
 import { resolveBaseCwd } from '../pi/websocket';
@@ -104,20 +105,31 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom
          emit({ type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError, stepIndex, iterationLabel });
          break;
        case 'result':
+          clearInterval(abortPoll);
          cleanup?.();
          resolve(event.cost);
          break;
        case 'error':
+          clearInterval(abortPoll);
          cleanup?.();
          reject(new Error(event.message));
          break;
        case 'stopped':
+          clearInterval(abortPoll);
          cleanup?.();
          reject(new Error('Step was stopped'));
          break;
      }
    };

+    // Poll for abort signal to kill the running agent
+    const abortPoll = setInterval(() => {
+      if (abortSignal.aborted && cleanup) {
+        clearInterval(abortPoll);
+        cleanup();
+      }
+    }, 500);
+
    try {
      if (isClaudeCode) {
        const handle = await sendClaudeCodeStreaming({
@@ -144,7 +156,14 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom
        await sidecar.spawnPi({ sessionId, email, userId, username, role, cwd, model });
        sidecar.sendPiPrompt(sessionId, prompt, randomUUID());
      }
+
+      // If already aborted while setting up, kill immediately
+      if (abortSignal.aborted) {
+        clearInterval(abortPoll);
+        cleanup?.();
+      }
    } catch (err) {
+      clearInterval(abortPoll);
      cleanup?.();
      reject(err);
    }
@@ -180,6 +199,121 @@ function buildStepPrompt(taskBody: string, inputs: Record<string, string>, targe
  return `${taskBody.trim()}\n\n## Inputs\n\n${inputLines}${contextSection}`;
 }

+// ── Script step execution ──
+
+type RunScriptStepParams = {
+  email: string;
+  role: string;
+  task: { name: string; implementation: string; language: string; args?: string[] | null };
+  inputs: Record<string, string>;
+  cwd: string;
+  abortSignal: AbortSignal;
+  emit: EmitEvent;
+  stepIndex: number;
+};
+
+function getRunner(language: string): string[] {
+  switch (language) {
+    case 'bash': return ['bash'];
+    case 'python': return ['python3'];
+    case 'typescript': return ['bun', 'run'];
+    case 'javascript': return ['node'];
+    default: return ['bash'];
+  }
+}
+
+function getFileName(language: string): string {
+  switch (language) {
+    case 'bash': return 'run.sh';
+    case 'python': return 'run.py';
+    case 'typescript': return 'index.ts';
+    case 'javascript': return 'index.js';
+    default: return 'run.sh';
+  }
+}
+
+async function runScriptStep({ email, role, task, inputs, cwd, abortSignal, emit, stepIndex }: RunScriptStepParams): Promise<void> {
+  const language = task.language ?? 'bash';
+
+  // Write script to temp file
+  const dir = join(tmpdir(), `officer-task-${Date.now()}-${Math.random().toString(36).slice(2)}`);
+  mkdirSync(dir, { recursive: true });
+  const fileName = getFileName(language);
+  const scriptPath = join(dir, fileName);
+  writeFileSync(scriptPath, task.implementation);
+  chmodSync(scriptPath, 0o755);
+
+  const cleanup = () => {
+    try { rmSync(dir, { recursive: true, force: true }); } catch { /* best effort */ }
+  };
+
+  // Build env vars from inputs
+  const inputEnv: Record<string, string> = {};
+  for (const [key, value] of Object.entries(inputs)) {
+    inputEnv[`INPUT_${key.toUpperCase()}`] = value;
+  }
+
+  // Build positional args
+  const positionalArgs = task.args?.map((name) => inputs[name] ?? '') ?? [];
+
+  const runner = getRunner(language);
+  const cmd = [...runner, scriptPath, ...positionalArgs];
+
+  const spawnEnv = { ...process.env as Record<string, string>, ...inputEnv };
+
+  console.log(`[pipeline] running script step ${stepIndex}: ${task.name} (cwd=${cwd})`);
+
+  try {
+    const proc = Bun.spawn(cmd, {
+      cwd,
+      env: spawnEnv,
+      stdout: 'pipe',
+      stderr: 'pipe',
+    });
+
+    const decoder = new TextDecoder();
+
+    const readStream = async (reader: ReadableStreamDefaultReader<Uint8Array>) => {
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          const text = decoder.decode(value);
+          emit({ type: 'assistant:delta', text, stepIndex });
+        }
+      } catch {
+        // stream closed
+      }
+    };
+
+    // Check abort periodically
+    const abortCheck = setInterval(() => {
+      if (abortSignal.aborted) {
+        try { proc.kill(); } catch { /* already dead */ }
+      }
+    }, 500);
+
+    const [, , exitCode] = await Promise.all([
+      readStream(proc.stdout.getReader()),
+      readStream(proc.stderr.getReader()),
+      proc.exited,
+    ]);
+
+    clearInterval(abortCheck);
+    cleanup();
+
+    // Flush the streamed text as a complete message
+    emit({ type: 'assistant:text', text: '', stepIndex });
+
+    if (exitCode !== 0 && !abortSignal.aborted) {
+      throw new Error(`Script exited with code ${exitCode}`);
+    }
+  } catch (err) {
+    cleanup();
+    throw err;
+  }
+}
+
 // ── Parallel foreach ──

 type ParallelForeachParams = {
@@ -251,7 +385,7 @@ async function runParallelForeach({
      const cwdRelative = cwd ? `${cwd}/${subdir}` : subdir;
      const resolvedCwd = resolveBaseCwd(email, role, cwdRelative);
      const targetDir = toAgentPath(resolvedCwd, email, role);
-      const prompt = buildStepPrompt(stepTask.body, iterInputs, targetDir);
+      const prompt = buildStepPrompt(stepTask.body!,iterInputs, targetDir);

      try {
        const cost = await runAgenticStep({
@@ -341,7 +475,9 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
      emit({ type: 'error', message: `Step task not found: ${step.task}` });
      return;
    }
-    if (!stepTask.body) {
+
+    const isScript = stepTask.mode === 'script';
+    if (!isScript && !stepTask.body) {
      emit({ type: 'error', message: `Step task "${step.task}" has no body` });
      return;
    }
@@ -353,6 +489,35 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
      }
    }

+    // Script-mode steps run directly (no agent), only single execution supported
+    if (isScript) {
+      if (!stepTask.implementation) {
+        emit({ type: 'error', message: `Script task "${step.task}" has no implementation` });
+        return;
+      }
+
+      emit({ type: 'step:start', stepIndex: stepIdx, taskName: stepTask.name });
+
+      try {
+        await runScriptStep({
+          email, role,
+          task: { name: stepTask.name, implementation: stepTask.implementation, language: stepTask.language ?? 'bash', args: stepTask.args as string[] | null },
+          inputs: resolvedInputs,
+          cwd: baseCwd,
+          abortSignal,
+          emit,
+          stepIndex: stepIdx,
+        });
+        emit({ type: 'step:complete', stepIndex: stepIdx });
+      } catch (err) {
+        const message = err instanceof Error ? err.message : String(err);
+        emit({ type: 'error', message: `Script step "${step.task}" failed: ${message}` });
+        return;
+      }
+
+      continue;
+    }
+
    if (step.foreach === 'subdirectory') {
      let subdirs: string[];
      try {
@@ -371,7 +536,7 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
      if (concurrency > 1) {
        await runParallelForeach({
          userId, email, username, role,
-          stepIdx, step, stepTask: { name: stepTask.name, body: stepTask.body },
+          stepIdx, step, stepTask: { name: stepTask.name, body: stepTask.body! },
          subdirs, baseCwd, inputs, cwd, abortSignal, totalCost, emit,
        });
      } else {
@@ -405,7 +570,7 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
          const cwdRelative = cwd ? `${cwd}/${subdir}` : subdir;
          const resolvedCwd = resolveBaseCwd(email, role, cwdRelative);
          const targetDir = toAgentPath(resolvedCwd, email, role);
-          const prompt = buildStepPrompt(stepTask.body, iterInputs, targetDir);
+          const prompt = buildStepPrompt(stepTask.body!,iterInputs, targetDir);

          const cost = await runAgenticStep({
            userId, email, username, role,
@@ -430,7 +595,7 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
      emit({ type: 'step:start', stepIndex: stepIdx, taskName: stepTask.name });

      const targetDir = toAgentPath(baseCwd, email, role);
-      const prompt = buildStepPrompt(stepTask.body, resolvedInputs, targetDir);
+      const prompt = buildStepPrompt(stepTask.body!,resolvedInputs, targetDir);

      const cost = await runAgenticStep({
        userId, email, username, role,
@@ -450,9 +615,11 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
    }
  }

-  if (!abortSignal.aborted) {
-    emit({ type: 'pipeline:complete', totalCost });
+  if (abortSignal.aborted) {
+    throw new Error('Pipeline was stopped');
  }
+
+  emit({ type: 'pipeline:complete', totalCost });
 }

 // ── WebSocket handler (thin layer) ──
@@ -161,6 +161,7 @@ export async function startJob(params: StartJobParams): Promise<string> {
    clearInterval(flushInterval);
    const message = err instanceof Error ? err.message : String(err);
    const isStopped = job.abortSignal.aborted;
+    broadcast(job, isStopped ? { type: 'stopped' } : { type: 'error', message });
    await updatePipelineJob(jobId, {
      status: isStopped ? 'stopped' : 'failed',
      progress: job.lastProgress as Record<string, unknown>,