jobs: run script tasks as background jobs via executeScript (phase 1b)

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

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

@@ -0,0 +1,145 @@
+import { join, isAbsolute } from 'node:path';
+import { mkdirSync, writeFileSync, chmodSync, rmSync, createWriteStream } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { getTaskByDirName } from './task-files';
+import { getHomeDirForRole, DATA_PATH } from '../../data-path';
+import { buildSandboxPrefix, buildRunuserSuffix } from '../../sidecar/sandbox';
+import { killTree } from './process-tree';
+
+// Script-job event stream. `stdout`/`stderr` are high-frequency (live-only + persisted to the log
+// file); `started`/`exit`/`error` are structural and get buffered for viewer replay.
+export type ScriptEvent =
+  | { type: 'started'; taskName: string }
+  | { type: 'stdout'; data: string }
+  | { type: 'stderr'; data: string }
+  | { type: 'exit'; code: number }
+  | { type: 'error'; message: string };
+
+export type ExecuteScriptParams = {
+  jobId: string;
+  email: string;
+  role: string;
+  sandboxed: boolean;
+  taskDirName: string;
+  inputs: Record<string, string>;
+  cwd?: string;
+  abortSignal: { aborted: boolean };
+  emit: (event: ScriptEvent) => void;
+};
+
+const getRunner = (language: string): string[] =>
+  language === 'python' ? ['python3'] : language === 'typescript' ? ['bun', 'run'] : language === 'javascript' ? ['node'] : ['bash'];
+const getFileName = (language: string): string =>
+  language === 'python' ? 'run.py' : language === 'typescript' ? 'index.ts' : language === 'javascript' ? 'index.js' : 'run.sh';
+
+function materializeScript(language: string, implementation: string): string {
+  const dir = join(tmpdir(), `officer-task-${Date.now()}-${Math.random().toString(36).slice(2)}`);
+  mkdirSync(dir, { recursive: true });
+  const filePath = join(dir, getFileName(language));
+  writeFileSync(filePath, implementation);
+  chmodSync(filePath, 0o755);
+  return filePath;
+}
+
+const buildInputEnv = (inputs: Record<string, string>): Record<string, string> => {
+  const env: Record<string, string> = {};
+  for (const [key, value] of Object.entries(inputs)) env[`INPUT_${key.toUpperCase()}`] = value;
+  return env;
+};
+
+const buildArgs = (inputs: Record<string, string>, argsOrder?: string[] | null): string[] =>
+  !argsOrder || argsOrder.length === 0 ? [] : argsOrder.map((name) => inputs[name] ?? '');
+
+// The persisted per-job output log — served by GET /jobs/:id/log for backfill / unattended viewing.
+export const jobLogPath = (jobId: string) => join(DATA_PATH, 'jobs', `${jobId}.log`);
+
+// Run a script-mode task as a background job. Emits ScriptEvents (for live viewers) AND appends all
+// output to a durable log file. Resolves with the process exit code; throws only on spawn failure or
+// when aborted (the manager maps those to failed/stopped).
+export async function executeScript(params: ExecuteScriptParams): Promise<{ exitCode: number }> {
+  const { jobId, email, role, sandboxed, inputs, abortSignal, emit } = params;
+
+  const task = await getTaskByDirName(params.taskDirName);
+  if (!task) throw new Error(`Task not found: ${params.taskDirName}`);
+  if (task.mode !== 'script') throw new Error('Task is not a script-mode task');
+  if (!task.implementation) throw new Error(`Task ${params.taskDirName} has no implementation`);
+
+  const language = task.language ?? 'bash';
+  const scriptPath = materializeScript(language, task.implementation);
+  const inputEnv = buildInputEnv(inputs);
+  const positionalArgs = buildArgs(inputs, task.args);
+  const cmd = [...getRunner(language), scriptPath, ...positionalArgs];
+
+  const homeDir = getHomeDirForRole(email, role);
+  const cwd = params.cwd ? (isAbsolute(params.cwd) ? params.cwd : join(homeDir, params.cwd)) : homeDir;
+
+  let spawnCmd: string[];
+  let spawnEnv: Record<string, string>;
+  let spawnCwd: string;
+
+  if (sandboxed) {
+    const prefix = buildSandboxPrefix(email);
+    const suffix = buildRunuserSuffix();
+    const userDataPrefix = join(DATA_PATH, email);
+    const translatePath = (v: string) => (v.startsWith(userDataPrefix) ? '/data' + v.slice(userDataPrefix.length) : v);
+    const envArgs: string[] = [];
+    for (const [key, value] of Object.entries(inputEnv)) envArgs.push('--setenv', key, translatePath(value));
+    const sandboxCmd = cmd.map((arg) => translatePath(arg));
+    const scriptDir = join(scriptPath, '..');
+    spawnCmd = [...prefix, '--ro-bind', scriptDir, scriptDir, ...envArgs, ...suffix, ...sandboxCmd];
+    spawnEnv = {};
+    spawnCwd = '/';
+  } else {
+    spawnCmd = cmd;
+    spawnEnv = { ...(process.env as Record<string, string>), ...inputEnv };
+    spawnCwd = cwd;
+  }
+
+  mkdirSync(join(DATA_PATH, 'jobs'), { recursive: true });
+  const log = createWriteStream(jobLogPath(jobId), { flags: 'w' });
+  const cleanup = () => {
+    try { rmSync(join(scriptPath, '..'), { recursive: true, force: true }); } catch { /* best effort */ }
+  };
+
+  emit({ type: 'started', taskName: task.name });
+
+  const proc = Bun.spawn(spawnCmd, { cwd: spawnCwd, env: spawnEnv, stdout: 'pipe', stderr: 'pipe' });
+
+  // Cooperative cancellation: poll the shared abort flag and tear down the whole process tree.
+  const abortPoll = setInterval(() => {
+    if (abortSignal.aborted) {
+      clearInterval(abortPoll);
+      try { killTree(proc.pid); } catch { /* already dead */ }
+    }
+  }, 500);
+
+  const decoder = new TextDecoder();
+  const pump = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
+    try {
+      for (;;) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        const data = decoder.decode(value);
+        emit({ type, data });
+        log.write(data);
+      }
+    } catch {
+      // stream closed
+    }
+  };
+
+  const [, , exitCode] = await Promise.all([
+    pump(proc.stdout.getReader(), 'stdout'),
+    pump(proc.stderr.getReader(), 'stderr'),
+    proc.exited,
+  ]);
+
+  clearInterval(abortPoll);
+  await new Promise<void>((resolve) => log.end(resolve));
+  cleanup();
+
+  if (abortSignal.aborted) throw new Error('Job was stopped');
+
+  emit({ type: 'exit', code: exitCode });
+  return { exitCode };
+}

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

@@ -10,6 +10,11 @@ import {
 } from 'officerdb';
 import { executePipeline } from './pipeline-executor';
 import type { OutMessage } from './pipeline-executor';
+import { executeScript, type ScriptEvent } from './execute-script';
+
+// Everything a job can stream — pipeline structural events plus script stdout/stderr/exit.
+type JobEvent = OutMessage | ScriptEvent;
+type JobMode = 'pipeline' | 'script' | 'agentic';
 
 type WSData = {
   userId: number;
@@ -24,7 +29,7 @@ type LiveJob = {
   userId: number;
   abortSignal: { aborted: boolean };
   emitter: EventEmitter;
-  eventBuffer: OutMessage[];
+  eventBuffer: JobEvent[];
   viewers: Set<ServerWebSocket<WSData>>;
   progressDirty: boolean;
   lastProgress: unknown;
@@ -36,15 +41,16 @@ const PROGRESS_FLUSH_MS = 3000;
 
 const liveJobs = new Map<string, LiveJob>();
 
-function sendToViewer(ws: ServerWebSocket<WSData>, jobId: string, event: OutMessage) {
+function sendToViewer(ws: ServerWebSocket<WSData>, jobId: string, event: JobEvent) {
   if (ws.readyState === 1) {
     ws.send(JSON.stringify({ jobId, ...event }));
   }
 }
 
-function broadcast(job: LiveJob, event: OutMessage) {
-  // Buffer for replay — skip high-frequency deltas (only useful for live streaming)
-  if (event.type !== 'assistant:delta') {
+function broadcast(job: LiveJob, event: JobEvent) {
+  // Buffer for replay — skip high-frequency streams (deltas / script stdout+stderr). Those are
+  // live-only; a late viewer backfills script output from the persisted log file instead.
+  if (event.type !== 'assistant:delta' && event.type !== 'stdout' && event.type !== 'stderr') {
     job.eventBuffer.push(event);
     if (job.eventBuffer.length > EVENT_BUFFER_SIZE) {
       job.eventBuffer.shift();
@@ -62,6 +68,8 @@ type StartJobParams = {
   email: string;
   username: string;
   role: string;
+  mode?: JobMode; // defaults to 'pipeline' for back-compat with the existing pipeline caller
+  sandboxed?: boolean; // script jobs only
   taskDirName: string;
   taskName: string;
   inputs: Record<string, string>;
@@ -73,6 +81,7 @@ type StartJobParams = {
 
 export async function startJob(params: StartJobParams): Promise<string> {
   const jobId = randomUUID();
+  const mode: JobMode = params.mode ?? 'pipeline';
 
   // Create DB record
   await createPipelineJob({
@@ -80,6 +89,7 @@ export async function startJob(params: StartJobParams): Promise<string> {
     userId: params.userId,
     taskDirName: params.taskDirName,
     taskName: params.taskName,
+    mode,
     status: 'running',
     inputs: params.inputs,
     cwd: params.cwd ?? null,
@@ -102,7 +112,7 @@ export async function startJob(params: StartJobParams): Promise<string> {
   liveJobs.set(jobId, job);
 
   // Emit callback — broadcasts to viewers and tracks progress
-  const emit = (event: OutMessage) => {
+  const emit = (event: JobEvent) => {
     broadcast(job, event);
 
     // Track progress for DB persistence (skip streaming messages — only structural events)
@@ -139,25 +149,45 @@ export async function startJob(params: StartJobParams): Promise<string> {
     }
   }, PROGRESS_FLUSH_MS);
 
-  // Run the pipeline in the background
-  executePipeline({
-    userId: params.userId,
-    email: params.email,
-    username: params.username,
-    role: params.role,
-    taskDirName: params.taskDirName,
-    inputs: params.inputs,
-    cwd: params.cwd,
-    model: params.model,
-    startAt: params.startAt,
-    abortSignal: job.abortSignal,
-    emit,
-  }).then(async () => {
+  // Run the job in the background — dispatch by mode.
+  const runner: Promise<void | { exitCode: number }> =
+    mode === 'script'
+      ? executeScript({
+          jobId,
+          email: params.email,
+          role: params.role,
+          sandboxed: params.sandboxed ?? false,
+          taskDirName: params.taskDirName,
+          inputs: params.inputs,
+          cwd: params.cwd,
+          abortSignal: job.abortSignal,
+          emit,
+        })
+      : executePipeline({
+          userId: params.userId,
+          email: params.email,
+          username: params.username,
+          role: params.role,
+          taskDirName: params.taskDirName,
+          inputs: params.inputs,
+          cwd: params.cwd,
+          model: params.model,
+          startAt: params.startAt,
+          abortSignal: job.abortSignal,
+          emit,
+        });
+
+  runner.then(async (result) => {
     clearInterval(flushInterval);
+    // Script jobs resolve with an exit code — a non-zero exit is a failure. Pipelines resolve void.
+    const exitCode = result && typeof result === 'object' && 'exitCode' in result ? result.exitCode : null;
+    const failed = exitCode !== null && exitCode !== 0;
     await updatePipelineJob(jobId, {
-      status: 'completed',
+      status: failed ? 'failed' : 'completed',
+      exitCode,
       progress: job.lastProgress as Record<string, unknown>,
       totalCost: job.lastCost as Record<string, unknown>,
+      error: failed ? `Script exited with code ${exitCode}` : undefined,
       completedAt: new Date(),
     }).catch((err) => console.error(`[pipeline-jobs] failed to finalize ${jobId}:`, err));
     liveJobs.delete(jobId);
@@ -243,8 +273,8 @@ export async function cleanupOnStartup() {
   }
 }
 
-/** Extract minimal progress info from events for DB storage. */
-function extractProgress(event: OutMessage, prev: unknown): unknown {
+/** Extract minimal progress info from events for DB storage. Script events fall through to `prev`. */
+function extractProgress(event: JobEvent, prev: unknown): unknown {
   const p = (prev as Record<string, unknown>) ?? {};
 
   switch (event.type) {

src/servers/api/tasks/process-tree.ts

@@ -0,0 +1,53 @@
+import { readdirSync, readFileSync } from 'node:fs';
+
+// Every descendant PID of `root`, from a single /proc walk (Linux). Bun's proc.kill() signals only
+// the direct child (bash), so a running grandchild like ffmpeg is left orphaned and keeps going.
+export function descendantPids(root: number): number[] {
+  const childrenOf = new Map<number, number[]>();
+  let entries: string[];
+  try {
+    entries = readdirSync('/proc');
+  } catch {
+    return [];
+  }
+  for (const entry of entries) {
+    const pid = Number(entry);
+    if (!Number.isInteger(pid)) continue;
+    try {
+      const stat = readFileSync(`/proc/${pid}/stat`, 'utf8');
+      // "pid (comm) state ppid …" — comm may contain spaces/parens, so parse after the last ')'.
+      const fields = stat.slice(stat.lastIndexOf(')') + 2).split(' ');
+      const ppid = Number(fields[1]);
+      if (!Number.isInteger(ppid)) continue;
+      const list = childrenOf.get(ppid);
+      if (list) list.push(pid);
+      else childrenOf.set(ppid, [pid]);
+    } catch {
+      // process vanished mid-scan
+    }
+  }
+  const out: number[] = [];
+  const stack = [root];
+  while (stack.length > 0) {
+    const parent = stack.pop()!;
+    for (const child of childrenOf.get(parent) ?? []) {
+      out.push(child);
+      stack.push(child);
+    }
+  }
+  return out;
+}
+
+// Terminate a process and its whole subtree: SIGTERM everything, then SIGKILL stragglers after a
+// grace period. Killing the root (bash) too stops it from spawning the next file mid-batch.
+export function killTree(root: number) {
+  const pids = [root, ...descendantPids(root)];
+  for (const pid of pids) {
+    try { process.kill(pid, 'SIGTERM'); } catch { /* already gone */ }
+  }
+  setTimeout(() => {
+    for (const pid of pids) {
+      try { process.kill(pid, 'SIGKILL'); } catch { /* gone */ }
+    }
+  }, 2000);
+}