From 234a24ddaf5974e567af34b6450e41b52e96efb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 23 Jul 2026 13:42:57 +0000 Subject: [PATCH] jobs: run script tasks as background jobs via executeScript (phase 1b) Co-Authored-By: Claude Opus 4.8 --- docs/jobs-unification.md | 9 +- src/servers/api/tasks/execute-script.ts | 145 ++++++++++++++++++ src/servers/api/tasks/pipeline-job-manager.ts | 76 ++++++--- src/servers/api/tasks/process-tree.ts | 53 +++++++ 4 files changed, 257 insertions(+), 26 deletions(-) create mode 100644 src/servers/api/tasks/execute-script.ts create mode 100644 src/servers/api/tasks/process-tree.ts diff --git a/docs/jobs-unification.md b/docs/jobs-unification.md index b20e537f..fd22edbc 100644 --- a/docs/jobs-unification.md +++ b/docs/jobs-unification.md @@ -58,9 +58,12 @@ Favor power-user affordances over guardrails. See memory `sole-user-assume-compe - One `notifyJobDone(job)` hook at finalize → push to the phone app. ## Progress -- [ ] 1a data model (migration + schema/types) -- [ ] 1b executeScript + manager dispatch -- [ ] 1c scheduler / queue +- [x] 1a data model — `mode` + `exit_code` columns (schema + applied to DB) +- [x] 1b executeScript + manager dispatch — `execute-script.ts` (spawn/sandbox/killTree port, log file, + abort poll, returns exitCode), `process-tree.ts` (shared killTree), `pipeline-job-manager` now + dispatches by `mode` and finalizes script jobs by exit code. *Compiles; runtime-untested until + a REST caller + restart exist.* +- [ ] 1c scheduler / queue (FIFO pending → promote on finalize; promoteNext on startup) - [ ] 2 REST job API + WS consolidation - [ ] 3 /jobs/new page + JobDetail script branch + retire modal + header indicator - [ ] 4 push notifications diff --git a/src/servers/api/tasks/execute-script.ts b/src/servers/api/tasks/execute-script.ts new file mode 100644 index 00000000..42a65126 --- /dev/null +++ b/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; + 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): Record => { + const env: Record = {}; + for (const [key, value] of Object.entries(inputs)) env[`INPUT_${key.toUpperCase()}`] = value; + return env; +}; + +const buildArgs = (inputs: Record, 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; + 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), ...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, 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((resolve) => log.end(resolve)); + cleanup(); + + if (abortSignal.aborted) throw new Error('Job was stopped'); + + emit({ type: 'exit', code: exitCode }); + return { exitCode }; +} diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts index 8799d1f2..9f624c27 100644 --- a/src/servers/api/tasks/pipeline-job-manager.ts +++ b/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>; progressDirty: boolean; lastProgress: unknown; @@ -36,15 +41,16 @@ const PROGRESS_FLUSH_MS = 3000; const liveJobs = new Map(); -function sendToViewer(ws: ServerWebSocket, jobId: string, event: OutMessage) { +function sendToViewer(ws: ServerWebSocket, 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; @@ -73,6 +81,7 @@ type StartJobParams = { export async function startJob(params: StartJobParams): Promise { 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 { 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 { 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 { } }, 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 = + 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, totalCost: job.lastCost as Record, + 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) ?? {}; switch (event.type) { diff --git a/src/servers/api/tasks/process-tree.ts b/src/servers/api/tasks/process-tree.ts new file mode 100644 index 00000000..0e386dce --- /dev/null +++ b/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(); + 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); +}