script-mode task execution — database-backed tasks with direct script runner
Tasks now live in the database (mode: script or agentic). Script-mode tasks bypass the agent entirely — the implementation is materialized to a temp file and executed directly, with stdout/stderr streamed to the UI via WebSocket. Includes convert-to-mp3 as the first native script task. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,255 @@
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
import { join } from 'node:path';
|
||||
import { mkdirSync, writeFileSync, chmodSync, rmSync } from 'node:fs';
|
||||
import { getTaskByDirName } from 'officerdb';
|
||||
import { getHomeDirForRole, DATA_PATH } from '../../data-path';
|
||||
import { buildSandboxPrefix, buildRunuserSuffix } from '../../sidecar/sandbox';
|
||||
|
||||
type WSData = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
sandboxed: boolean;
|
||||
};
|
||||
|
||||
type RunMessage = {
|
||||
type: 'run';
|
||||
taskDirName: string;
|
||||
inputs: Record<string, string>;
|
||||
cwd?: string;
|
||||
};
|
||||
|
||||
type StopMessage = {
|
||||
type: 'stop';
|
||||
};
|
||||
|
||||
type ClientMessage = RunMessage | StopMessage;
|
||||
|
||||
type OutMessage =
|
||||
| { type: 'started'; taskName: string }
|
||||
| { type: 'stdout'; data: string }
|
||||
| { type: 'stderr'; data: string }
|
||||
| { type: 'exit'; code: number }
|
||||
| { type: 'error'; message: string };
|
||||
|
||||
// Active processes per WebSocket
|
||||
const activeProcs = new WeakMap<ServerWebSocket<WSData>, { proc: ReturnType<typeof Bun.spawn>; kill: () => void }>();
|
||||
|
||||
import { tmpdir } from 'node:os';
|
||||
|
||||
function send(ws: ServerWebSocket<WSData>, msg: OutMessage) {
|
||||
if (ws.readyState === 1) ws.send(JSON.stringify(msg));
|
||||
}
|
||||
|
||||
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';
|
||||
}
|
||||
}
|
||||
|
||||
// Write implementation to a temp file for execution, cleaned up after
|
||||
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 fileName = getFileName(language);
|
||||
const filePath = join(dir, fileName);
|
||||
|
||||
writeFileSync(filePath, implementation);
|
||||
chmodSync(filePath, 0o755);
|
||||
|
||||
return filePath;
|
||||
}
|
||||
|
||||
function 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;
|
||||
}
|
||||
|
||||
function buildArgs(inputs: Record<string, string>, argsOrder?: string[] | null): string[] {
|
||||
if (!argsOrder || argsOrder.length === 0) return [];
|
||||
return argsOrder.map((name) => inputs[name] ?? '');
|
||||
}
|
||||
|
||||
async function handleRun(ws: ServerWebSocket<WSData>, msg: RunMessage) {
|
||||
const { email, role, sandboxed, userId } = ws.data;
|
||||
|
||||
// Resolve task from database
|
||||
const task = await getTaskByDirName(msg.taskDirName, userId);
|
||||
if (!task) {
|
||||
send(ws, { type: 'error', message: `Task not found: ${msg.taskDirName}` });
|
||||
return;
|
||||
}
|
||||
|
||||
if (task.mode !== 'script') {
|
||||
send(ws, { type: 'error', message: 'Task is not a script-mode task' });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!task.implementation) {
|
||||
send(ws, { type: 'error', message: `Task ${msg.taskDirName} has no implementation` });
|
||||
return;
|
||||
}
|
||||
|
||||
const language = task.language ?? 'bash';
|
||||
|
||||
// Write script to temp dir for execution
|
||||
const scriptPath = materializeScript(language, task.implementation);
|
||||
|
||||
// Build env vars from inputs
|
||||
const inputEnv = buildInputEnv(msg.inputs);
|
||||
|
||||
// Build positional args
|
||||
const positionalArgs = buildArgs(msg.inputs, task.args);
|
||||
|
||||
// Build the command
|
||||
const runner = getRunner(language);
|
||||
const cmd = [...runner, scriptPath, ...positionalArgs];
|
||||
|
||||
// Resolve cwd
|
||||
const homeDir = getHomeDirForRole(email, role);
|
||||
const cwd = msg.cwd ?? homeDir;
|
||||
|
||||
let spawnCmd: string[];
|
||||
let spawnEnv: Record<string, string>;
|
||||
let spawnCwd: string;
|
||||
|
||||
if (sandboxed) {
|
||||
const prefix = buildSandboxPrefix(email);
|
||||
const suffix = buildRunuserSuffix();
|
||||
|
||||
// Translate paths in inputs and args: DATA_PATH/{email}/... → /data/...
|
||||
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));
|
||||
}
|
||||
|
||||
// Translate positional args too
|
||||
const sandboxCmd = cmd.map((arg) => translatePath(arg));
|
||||
|
||||
// Script is in /tmp which is a tmpfs inside bwrap — need to bind-mount the host tmp dir
|
||||
const scriptDir = join(scriptPath, '..');
|
||||
const extraMounts = ['--ro-bind', scriptDir, scriptDir];
|
||||
|
||||
spawnCmd = [...prefix, ...extraMounts, ...envArgs, ...suffix, ...sandboxCmd];
|
||||
spawnEnv = {};
|
||||
spawnCwd = '/';
|
||||
} else {
|
||||
spawnCmd = cmd;
|
||||
spawnEnv = { ...process.env as Record<string, string>, ...inputEnv };
|
||||
spawnCwd = cwd;
|
||||
}
|
||||
|
||||
const cleanup = () => {
|
||||
try { rmSync(join(scriptPath, '..'), { recursive: true, force: true }); } catch { /* best effort */ }
|
||||
};
|
||||
|
||||
send(ws, { type: 'started', taskName: task.name });
|
||||
|
||||
try {
|
||||
const proc = Bun.spawn(spawnCmd, {
|
||||
cwd: spawnCwd,
|
||||
env: spawnEnv,
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
});
|
||||
|
||||
activeProcs.set(ws, {
|
||||
proc,
|
||||
kill: () => {
|
||||
try { proc.kill(); } catch { /* already dead */ }
|
||||
},
|
||||
});
|
||||
|
||||
const stdoutReader = proc.stdout.getReader();
|
||||
const stderrReader = proc.stderr.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
|
||||
const readStream = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
send(ws, { type, data: decoder.decode(value) });
|
||||
}
|
||||
} catch {
|
||||
// stream closed
|
||||
}
|
||||
};
|
||||
|
||||
const [, , exitCode] = await Promise.all([
|
||||
readStream(stdoutReader, 'stdout'),
|
||||
readStream(stderrReader, 'stderr'),
|
||||
proc.exited,
|
||||
]);
|
||||
|
||||
activeProcs.delete(ws);
|
||||
cleanup();
|
||||
send(ws, { type: 'exit', code: exitCode });
|
||||
} catch (err) {
|
||||
activeProcs.delete(ws);
|
||||
cleanup();
|
||||
send(ws, { type: 'error', message: `Failed to spawn: ${err instanceof Error ? err.message : String(err)}` });
|
||||
}
|
||||
}
|
||||
|
||||
export function open(_ws: ServerWebSocket<WSData>) {
|
||||
// nothing to do
|
||||
}
|
||||
|
||||
export function message(ws: ServerWebSocket<WSData>, raw: string | Buffer) {
|
||||
const data = typeof raw === 'string' ? raw : raw.toString();
|
||||
|
||||
try {
|
||||
const msg = JSON.parse(data) as ClientMessage;
|
||||
|
||||
if (msg.type === 'run') {
|
||||
handleRun(ws, msg);
|
||||
} else if (msg.type === 'stop') {
|
||||
const active = activeProcs.get(ws);
|
||||
if (active) {
|
||||
active.kill();
|
||||
activeProcs.delete(ws);
|
||||
send(ws, { type: 'exit', code: -1 });
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
send(ws, { type: 'error', message: 'Failed to parse message' });
|
||||
}
|
||||
}
|
||||
|
||||
export function close(ws: ServerWebSocket<WSData>) {
|
||||
const active = activeProcs.get(ws);
|
||||
if (active) {
|
||||
active.kill();
|
||||
activeProcs.delete(ws);
|
||||
}
|
||||
}
|
||||
|
||||
export const taskRunnerWebsocket = {
|
||||
open,
|
||||
message,
|
||||
close,
|
||||
drain() {},
|
||||
};
|
||||
Reference in New Issue
Block a user