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 { getOwnerHomeDir, DATA_PATH } from '../../data-path'; import { killTree } from './process-tree'; import { buildTaskApiEnv } from './task-api-env'; // 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; userId: number; email: string; username?: string | null; 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, userId, email, username, 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 = getOwnerHomeDir(email); const cwd = params.cwd ? (isAbsolute(params.cwd) ? params.cwd : join(homeDir, params.cwd)) : homeDir; const spawnCmd = cmd; const apiEnv = await buildTaskApiEnv({ id: userId, email, username }); const spawnEnv = { ...(process.env as Record), ...apiEnv, ...inputEnv }; const 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 }; }