add pipeline job management with /jobs pages and per-step output viewer
- Pipeline jobs now persist to DB with progress tracking and cost accumulation - Jobs survive WebSocket disconnects with in-memory event buffer replay - New /jobs list page with search, status badges, and cost display - New /jobs/:id detail page with live WebSocket attachment and REST fallback - Two-column layout using WorkspaceLayout for resizable steps/output panels - Streaming messages tagged with stepIndex/iterationLabel for per-step output grouping - TaskRunnerModal links to job detail page once job is created - Dock entry added for Jobs page Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -3,11 +3,13 @@ import { randomUUID } from 'crypto';
|
||||
import { readdirSync, existsSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { getTaskByDirName, getUserSettings } from 'officerdb';
|
||||
import { getHomeDirForRole } from '../../data-path';
|
||||
import { getHomeDirForRole, getHomeDir } from '../../data-path';
|
||||
import { resolveBaseCwd } from '../pi/websocket';
|
||||
import { SANDBOX_HOME } from '../../sidecar/sandbox';
|
||||
import * as sidecar from '../../sidecar-registry';
|
||||
import { sendClaudeCodeStreaming } from '../../channels/send-claude-code';
|
||||
import type { PiEvent, MessageCost } from '../pi/types';
|
||||
import * as jobManager from './pipeline-job-manager';
|
||||
|
||||
const DEFAULT_MODEL = 'claude-code';
|
||||
|
||||
@@ -21,72 +23,64 @@ async function resolveModel(userId: number): Promise<string> {
|
||||
}
|
||||
}
|
||||
|
||||
type WSData = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
sandboxed: boolean;
|
||||
};
|
||||
// ── Types ──
|
||||
|
||||
type PipelineStep = {
|
||||
task: string;
|
||||
inputs?: Record<string, string>;
|
||||
foreach?: 'subdirectory';
|
||||
skip_if?: string;
|
||||
concurrency?: string;
|
||||
};
|
||||
|
||||
type PipelineConfig = {
|
||||
steps: PipelineStep[];
|
||||
};
|
||||
|
||||
type RunMessage = {
|
||||
type: 'run';
|
||||
taskDirName: string;
|
||||
inputs: Record<string, string>;
|
||||
cwd?: string;
|
||||
};
|
||||
|
||||
type ClientMessage = RunMessage | { type: 'stop' };
|
||||
|
||||
// Messages sent to client
|
||||
type OutMessage =
|
||||
| { type: 'pipeline:init'; steps: Array<{ task: string; foreach?: string }> }
|
||||
export type OutMessage =
|
||||
| { type: 'pipeline:init'; steps: Array<{ task: string; foreach?: string; concurrency?: number }> }
|
||||
| { type: 'step:start'; stepIndex: number; taskName: string; iteration?: { current: number; total: number; label: string } }
|
||||
| { type: 'step:complete'; stepIndex: number; cost?: MessageCost }
|
||||
| { type: 'step:skip'; stepIndex: number; label: string; reason: string }
|
||||
| { type: 'assistant:delta'; text: string }
|
||||
| { type: 'assistant:text'; text: string }
|
||||
| { type: 'tool:start'; toolCallId: string; toolName: string; toolInput: Record<string, unknown> }
|
||||
| { type: 'tool:result'; toolCallId: string; output: string; isError: boolean }
|
||||
| { type: 'step:parallel'; stepIndex: number; taskName: string; iterations: string[]; concurrency: number }
|
||||
| { type: 'iteration:start'; stepIndex: number; label: string }
|
||||
| { type: 'iteration:complete'; stepIndex: number; label: string; cost?: MessageCost }
|
||||
| { type: 'iteration:error'; stepIndex: number; label: string; error: string }
|
||||
| { type: 'assistant:delta'; text: string; stepIndex: number; iterationLabel?: string }
|
||||
| { type: 'assistant:text'; text: string; stepIndex: number; iterationLabel?: string }
|
||||
| { type: 'tool:start'; toolCallId: string; toolName: string; toolInput: Record<string, unknown>; stepIndex: number; iterationLabel?: string }
|
||||
| { type: 'tool:result'; toolCallId: string; output: string; isError: boolean; stepIndex: number; iterationLabel?: string }
|
||||
| { type: 'pipeline:complete'; totalCost: MessageCost }
|
||||
| { type: 'error'; message: string }
|
||||
| { type: 'stopped' };
|
||||
|
||||
// Active pipelines per WebSocket
|
||||
const activePipelines = new WeakMap<ServerWebSocket<WSData>, { abort: () => void }>();
|
||||
|
||||
function send(ws: ServerWebSocket<WSData>, msg: OutMessage) {
|
||||
if (ws.readyState === 1) ws.send(JSON.stringify(msg));
|
||||
}
|
||||
type EmitEvent = (msg: OutMessage) => void;
|
||||
|
||||
type AbortSignal = { aborted: boolean };
|
||||
|
||||
// ── Step execution ──
|
||||
|
||||
type RunStepParams = {
|
||||
ws: ServerWebSocket<WSData>;
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
taskDirName: string;
|
||||
prompt: string;
|
||||
cwd: string;
|
||||
abortSignal: AbortSignal;
|
||||
emit: EmitEvent;
|
||||
stepIndex: number;
|
||||
iterationLabel?: string;
|
||||
};
|
||||
|
||||
async function runAgenticStep({ ws, taskDirName, prompt, cwd, abortSignal }: RunStepParams): Promise<MessageCost> {
|
||||
const { email, username, userId, role } = ws.data;
|
||||
async function runAgenticStep({ userId, email, username, role, taskDirName, prompt, cwd, abortSignal, emit, stepIndex, iterationLabel }: RunStepParams): Promise<MessageCost> {
|
||||
const sessionId = randomUUID();
|
||||
const model = await resolveModel(userId);
|
||||
const isClaudeCode = model.startsWith('claude-code');
|
||||
|
||||
console.log(`[pipeline] starting step for session ${sessionId} (model=${model})`);
|
||||
console.log(`[pipeline] starting step ${stepIndex} for session ${sessionId} (model=${model})${iterationLabel ? ` [${iterationLabel}]` : ''}`);
|
||||
|
||||
return new Promise<MessageCost>(async (resolve, reject) => {
|
||||
if (abortSignal.aborted) return reject(new Error('Pipeline aborted'));
|
||||
@@ -98,16 +92,16 @@ async function runAgenticStep({ ws, taskDirName, prompt, cwd, abortSignal }: Run
|
||||
|
||||
switch (event.type) {
|
||||
case 'delta':
|
||||
send(ws, { type: 'assistant:delta', text: event.text });
|
||||
emit({ type: 'assistant:delta', text: event.text, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'text':
|
||||
send(ws, { type: 'assistant:text', text: event.text });
|
||||
emit({ type: 'assistant:text', text: event.text, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'tool:start':
|
||||
send(ws, { type: 'tool:start', toolCallId: event.toolCallId, toolName: event.toolName, toolInput: event.toolInput });
|
||||
emit({ type: 'tool:start', toolCallId: event.toolCallId, toolName: event.toolName, toolInput: event.toolInput, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'tool:result':
|
||||
send(ws, { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError });
|
||||
emit({ type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'result':
|
||||
cleanup?.();
|
||||
@@ -157,10 +151,22 @@ async function runAgenticStep({ ws, taskDirName, prompt, cwd, abortSignal }: Run
|
||||
});
|
||||
}
|
||||
|
||||
// ── Helpers ──
|
||||
|
||||
function resolveInputTemplate(template: string, variables: Record<string, string>): string {
|
||||
return template.replace(/\$\{(\w+)\}/g, (_, key) => variables[key] ?? '');
|
||||
}
|
||||
|
||||
/** Convert a host-side absolute path to the path the agent sees inside the sandbox. */
|
||||
function toAgentPath(hostPath: string, email: string, role: string): string {
|
||||
if (role === 'Super Admin') return hostPath;
|
||||
const hostHome = getHomeDir(email);
|
||||
if (hostPath.startsWith(hostHome)) {
|
||||
return SANDBOX_HOME + hostPath.slice(hostHome.length);
|
||||
}
|
||||
return hostPath;
|
||||
}
|
||||
|
||||
function buildStepPrompt(taskBody: string, inputs: Record<string, string>, targetDir?: string): string {
|
||||
const inputLines = Object.entries(inputs)
|
||||
.filter(([, v]) => v.trim())
|
||||
@@ -174,93 +180,214 @@ function buildStepPrompt(taskBody: string, inputs: Record<string, string>, targe
|
||||
return `${taskBody.trim()}\n\n## Inputs\n\n${inputLines}${contextSection}`;
|
||||
}
|
||||
|
||||
async function handleRun(ws: ServerWebSocket<WSData>, msg: RunMessage) {
|
||||
const { email, role, userId } = ws.data;
|
||||
// ── Parallel foreach ──
|
||||
|
||||
const pipelineTask = await getTaskByDirName(msg.taskDirName, userId);
|
||||
type ParallelForeachParams = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
stepIdx: number;
|
||||
step: PipelineStep;
|
||||
stepTask: { name: string; body: string };
|
||||
subdirs: string[];
|
||||
baseCwd: string;
|
||||
inputs: Record<string, string>;
|
||||
cwd?: string;
|
||||
abortSignal: AbortSignal;
|
||||
totalCost: MessageCost;
|
||||
emit: EmitEvent;
|
||||
};
|
||||
|
||||
async function runParallelForeach({
|
||||
userId, email, username, role, stepIdx, step, stepTask, subdirs, baseCwd,
|
||||
inputs, cwd, abortSignal, totalCost, emit,
|
||||
}: ParallelForeachParams) {
|
||||
const concurrency = Math.max(1, parseInt(step.concurrency ?? '1', 10));
|
||||
|
||||
// Determine skip vs run
|
||||
const toSkip: string[] = [];
|
||||
const toRun: string[] = [];
|
||||
for (const subdir of subdirs) {
|
||||
const subdirPath = join(baseCwd, subdir);
|
||||
if (step.skip_if && existsSync(join(subdirPath, step.skip_if))) {
|
||||
toSkip.push(subdir);
|
||||
} else {
|
||||
toRun.push(subdir);
|
||||
}
|
||||
}
|
||||
|
||||
// Send skips
|
||||
for (const label of toSkip) {
|
||||
emit({ type: 'step:skip', stepIndex: stepIdx, label, reason: `${step.skip_if} already exists` });
|
||||
}
|
||||
|
||||
// Tell UI about parallel step
|
||||
emit({
|
||||
type: 'step:parallel',
|
||||
stepIndex: stepIdx,
|
||||
taskName: stepTask.name,
|
||||
iterations: toRun,
|
||||
concurrency,
|
||||
});
|
||||
|
||||
// Pool-based execution
|
||||
const executing = new Set<Promise<void>>();
|
||||
|
||||
for (const subdir of toRun) {
|
||||
if (abortSignal.aborted) break;
|
||||
|
||||
const run = async () => {
|
||||
emit({ type: 'iteration:start', stepIndex: stepIdx, label: subdir });
|
||||
|
||||
const iterVars = { ...inputs, folder_name: subdir };
|
||||
const iterInputs: Record<string, string> = {};
|
||||
if (step.inputs) {
|
||||
for (const [key, template] of Object.entries(step.inputs)) {
|
||||
iterInputs[key] = resolveInputTemplate(template, iterVars);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
try {
|
||||
const cost = await runAgenticStep({
|
||||
userId, email, username, role,
|
||||
taskDirName: step.task,
|
||||
prompt,
|
||||
cwd: resolvedCwd,
|
||||
abortSignal,
|
||||
emit,
|
||||
stepIndex: stepIdx,
|
||||
iterationLabel: subdir,
|
||||
});
|
||||
|
||||
totalCost.inputTokens += cost.inputTokens;
|
||||
totalCost.outputTokens += cost.outputTokens;
|
||||
totalCost.totalUSD += cost.totalUSD;
|
||||
|
||||
emit({ type: 'iteration:complete', stepIndex: stepIdx, label: subdir, cost });
|
||||
} catch (err) {
|
||||
if (!abortSignal.aborted) {
|
||||
emit({ type: 'iteration:error', stepIndex: stepIdx, label: subdir, error: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const p = run().then(() => { executing.delete(p); });
|
||||
executing.add(p);
|
||||
|
||||
if (executing.size >= concurrency) {
|
||||
await Promise.race(executing);
|
||||
}
|
||||
}
|
||||
|
||||
await Promise.all(executing);
|
||||
}
|
||||
|
||||
// ── Pipeline orchestrator ──
|
||||
|
||||
export type ExecutePipelineParams = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
taskDirName: string;
|
||||
inputs: Record<string, string>;
|
||||
cwd?: string;
|
||||
abortSignal: AbortSignal;
|
||||
emit: EmitEvent;
|
||||
};
|
||||
|
||||
export async function executePipeline({ userId, email, username, role, taskDirName, inputs, cwd, abortSignal, emit }: ExecutePipelineParams): Promise<void> {
|
||||
const pipelineTask = await getTaskByDirName(taskDirName, userId);
|
||||
if (!pipelineTask) {
|
||||
send(ws, { type: 'error', message: `Task not found: ${msg.taskDirName}` });
|
||||
emit({ type: 'error', message: `Task not found: ${taskDirName}` });
|
||||
return;
|
||||
}
|
||||
if (pipelineTask.mode !== 'pipeline') {
|
||||
send(ws, { type: 'error', message: 'Task is not a pipeline-mode task' });
|
||||
emit({ type: 'error', message: 'Task is not a pipeline-mode task' });
|
||||
return;
|
||||
}
|
||||
|
||||
const config = pipelineTask.config as PipelineConfig | null;
|
||||
if (!config?.steps?.length) {
|
||||
send(ws, { type: 'error', message: 'Pipeline has no steps defined' });
|
||||
emit({ type: 'error', message: 'Pipeline has no steps defined' });
|
||||
return;
|
||||
}
|
||||
|
||||
const abortSignal = { aborted: false };
|
||||
activePipelines.set(ws, {
|
||||
abort: () => { abortSignal.aborted = true; },
|
||||
});
|
||||
|
||||
const baseCwd = resolveBaseCwd(email, role, msg.cwd);
|
||||
const baseCwd = resolveBaseCwd(email, role, cwd);
|
||||
const totalCost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
|
||||
// Send pipeline init
|
||||
send(ws, {
|
||||
emit({
|
||||
type: 'pipeline:init',
|
||||
steps: config.steps.map((s) => ({ task: s.task, foreach: s.foreach })),
|
||||
steps: config.steps.map((s) => ({
|
||||
task: s.task,
|
||||
foreach: s.foreach,
|
||||
concurrency: s.concurrency ? parseInt(s.concurrency, 10) : undefined,
|
||||
})),
|
||||
});
|
||||
|
||||
try {
|
||||
for (let stepIdx = 0; stepIdx < config.steps.length; stepIdx++) {
|
||||
if (abortSignal.aborted) break;
|
||||
for (let stepIdx = 0; stepIdx < config.steps.length; stepIdx++) {
|
||||
if (abortSignal.aborted) break;
|
||||
|
||||
const step = config.steps[stepIdx]!;
|
||||
const step = config.steps[stepIdx]!;
|
||||
|
||||
// Resolve the referenced task
|
||||
const stepTask = await getTaskByDirName(step.task, userId);
|
||||
if (!stepTask) {
|
||||
send(ws, { type: 'error', message: `Step task not found: ${step.task}` });
|
||||
const stepTask = await getTaskByDirName(step.task, userId);
|
||||
if (!stepTask) {
|
||||
emit({ type: 'error', message: `Step task not found: ${step.task}` });
|
||||
return;
|
||||
}
|
||||
if (!stepTask.body) {
|
||||
emit({ type: 'error', message: `Step task "${step.task}" has no body` });
|
||||
return;
|
||||
}
|
||||
|
||||
const resolvedInputs: Record<string, string> = {};
|
||||
if (step.inputs) {
|
||||
for (const [key, template] of Object.entries(step.inputs)) {
|
||||
resolvedInputs[key] = resolveInputTemplate(template, inputs);
|
||||
}
|
||||
}
|
||||
|
||||
if (step.foreach === 'subdirectory') {
|
||||
let subdirs: string[];
|
||||
try {
|
||||
subdirs = readdirSync(baseCwd, { withFileTypes: true })
|
||||
.filter((d) => d.isDirectory())
|
||||
.map((d) => d.name)
|
||||
.filter((name) => !name.startsWith('.'))
|
||||
.sort();
|
||||
} catch {
|
||||
emit({ type: 'error', message: `Cannot read directory: ${baseCwd}` });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!stepTask.body) {
|
||||
send(ws, { type: 'error', message: `Step task "${step.task}" has no body` });
|
||||
return;
|
||||
}
|
||||
|
||||
// Resolve input templates using pipeline inputs
|
||||
const resolvedInputs: Record<string, string> = {};
|
||||
if (step.inputs) {
|
||||
for (const [key, template] of Object.entries(step.inputs)) {
|
||||
resolvedInputs[key] = resolveInputTemplate(template, msg.inputs);
|
||||
}
|
||||
}
|
||||
|
||||
if (step.foreach === 'subdirectory') {
|
||||
// Iterate over subdirectories
|
||||
let subdirs: string[];
|
||||
try {
|
||||
subdirs = readdirSync(baseCwd, { withFileTypes: true })
|
||||
.filter((d) => d.isDirectory())
|
||||
.map((d) => d.name)
|
||||
.filter((name) => !name.startsWith('.'))
|
||||
.sort();
|
||||
} catch {
|
||||
send(ws, { type: 'error', message: `Cannot read directory: ${baseCwd}` });
|
||||
return;
|
||||
}
|
||||
const concurrency = parseInt(step.concurrency ?? '1', 10);
|
||||
|
||||
if (concurrency > 1) {
|
||||
await runParallelForeach({
|
||||
userId, email, username, role,
|
||||
stepIdx, step, stepTask: { name: stepTask.name, body: stepTask.body },
|
||||
subdirs, baseCwd, inputs, cwd, abortSignal, totalCost, emit,
|
||||
});
|
||||
} else {
|
||||
// Sequential
|
||||
for (let i = 0; i < subdirs.length; i++) {
|
||||
if (abortSignal.aborted) break;
|
||||
|
||||
const subdir = subdirs[i]!;
|
||||
const subdirPath = join(baseCwd, subdir);
|
||||
|
||||
// Check skip condition
|
||||
if (step.skip_if && existsSync(join(subdirPath, step.skip_if))) {
|
||||
send(ws, { type: 'step:skip', stepIndex: stepIdx, label: subdir, reason: `${step.skip_if} already exists` });
|
||||
emit({ type: 'step:skip', stepIndex: stepIdx, label: subdir, reason: `${step.skip_if} already exists` });
|
||||
continue;
|
||||
}
|
||||
|
||||
// Resolve folder-specific variables
|
||||
const iterVars = { ...msg.inputs, folder_name: subdir };
|
||||
const iterVars = { ...inputs, folder_name: subdir };
|
||||
const iterInputs: Record<string, string> = {};
|
||||
if (step.inputs) {
|
||||
for (const [key, template] of Object.entries(step.inputs)) {
|
||||
@@ -268,87 +395,145 @@ async function handleRun(ws: ServerWebSocket<WSData>, msg: RunMessage) {
|
||||
}
|
||||
}
|
||||
|
||||
send(ws, {
|
||||
emit({
|
||||
type: 'step:start',
|
||||
stepIndex: stepIdx,
|
||||
taskName: stepTask.name,
|
||||
iteration: { current: i + 1, total: subdirs.length, label: subdir },
|
||||
});
|
||||
|
||||
// Build relative path for cwd (sandbox-safe)
|
||||
const stepCwd = subdirPath;
|
||||
const cwdRelative = msg.cwd ? `${msg.cwd}/${subdir}` : subdir;
|
||||
const cwdRelative = cwd ? `${cwd}/${subdir}` : subdir;
|
||||
const resolvedCwd = resolveBaseCwd(email, role, cwdRelative);
|
||||
|
||||
// Build ~/relative path for context
|
||||
const targetDir = msg.cwd ? `~/${msg.cwd}/${subdir}` : `~/${subdir}`;
|
||||
const targetDir = toAgentPath(resolvedCwd, email, role);
|
||||
const prompt = buildStepPrompt(stepTask.body, iterInputs, targetDir);
|
||||
|
||||
const cost = await runAgenticStep({
|
||||
ws,
|
||||
userId, email, username, role,
|
||||
taskDirName: step.task,
|
||||
prompt,
|
||||
cwd: resolvedCwd,
|
||||
abortSignal,
|
||||
emit,
|
||||
stepIndex: stepIdx,
|
||||
iterationLabel: subdir,
|
||||
});
|
||||
|
||||
totalCost.inputTokens += cost.inputTokens;
|
||||
totalCost.outputTokens += cost.outputTokens;
|
||||
totalCost.totalUSD += cost.totalUSD;
|
||||
|
||||
send(ws, { type: 'step:complete', stepIndex: stepIdx, cost });
|
||||
emit({ type: 'step:complete', stepIndex: stepIdx, cost });
|
||||
}
|
||||
} else {
|
||||
// Single execution step
|
||||
send(ws, { type: 'step:start', stepIndex: stepIdx, taskName: stepTask.name });
|
||||
|
||||
const targetDir = msg.cwd ? `~/${msg.cwd}` : '~';
|
||||
const prompt = buildStepPrompt(stepTask.body, resolvedInputs, targetDir);
|
||||
|
||||
const cost = await runAgenticStep({
|
||||
ws,
|
||||
taskDirName: step.task,
|
||||
prompt,
|
||||
cwd: baseCwd,
|
||||
abortSignal,
|
||||
});
|
||||
|
||||
totalCost.inputTokens += cost.inputTokens;
|
||||
totalCost.outputTokens += cost.outputTokens;
|
||||
totalCost.totalUSD += cost.totalUSD;
|
||||
|
||||
send(ws, { type: 'step:complete', stepIndex: stepIdx, cost });
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Single execution step
|
||||
emit({ type: 'step:start', stepIndex: stepIdx, taskName: stepTask.name });
|
||||
|
||||
if (!abortSignal.aborted) {
|
||||
send(ws, { type: 'pipeline:complete', totalCost });
|
||||
const targetDir = toAgentPath(baseCwd, email, role);
|
||||
const prompt = buildStepPrompt(stepTask.body, resolvedInputs, targetDir);
|
||||
|
||||
const cost = await runAgenticStep({
|
||||
userId, email, username, role,
|
||||
taskDirName: step.task,
|
||||
prompt,
|
||||
cwd: baseCwd,
|
||||
abortSignal,
|
||||
emit,
|
||||
stepIndex: stepIdx,
|
||||
});
|
||||
|
||||
totalCost.inputTokens += cost.inputTokens;
|
||||
totalCost.outputTokens += cost.outputTokens;
|
||||
totalCost.totalUSD += cost.totalUSD;
|
||||
|
||||
emit({ type: 'step:complete', stepIndex: stepIdx, cost });
|
||||
}
|
||||
} catch (err) {
|
||||
if (!abortSignal.aborted) {
|
||||
send(ws, { type: 'error', message: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
} finally {
|
||||
activePipelines.delete(ws);
|
||||
}
|
||||
|
||||
if (!abortSignal.aborted) {
|
||||
emit({ type: 'pipeline:complete', totalCost });
|
||||
}
|
||||
}
|
||||
|
||||
// ── WebSocket handler (thin layer) ──
|
||||
|
||||
type WSData = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
sandboxed: boolean;
|
||||
};
|
||||
|
||||
type ClientMessage =
|
||||
| { type: 'run'; taskDirName: string; inputs: Record<string, string>; cwd?: string }
|
||||
| { type: 'stop'; jobId: string }
|
||||
| { type: 'attach'; jobId: string }
|
||||
| { type: 'list' };
|
||||
|
||||
function send(ws: ServerWebSocket<WSData>, msg: Record<string, unknown>) {
|
||||
if (ws.readyState === 1) ws.send(JSON.stringify(msg));
|
||||
}
|
||||
|
||||
export function open(_ws: ServerWebSocket<WSData>) {}
|
||||
|
||||
export function message(ws: ServerWebSocket<WSData>, raw: string | Buffer) {
|
||||
export async 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 = activePipelines.get(ws);
|
||||
if (active) {
|
||||
active.abort();
|
||||
activePipelines.delete(ws);
|
||||
send(ws, { type: 'stopped' });
|
||||
switch (msg.type) {
|
||||
case 'run': {
|
||||
const { userId, email, username, role } = ws.data;
|
||||
|
||||
// Resolve task name for the DB record
|
||||
const task = await getTaskByDirName(msg.taskDirName, userId);
|
||||
if (!task) {
|
||||
send(ws, { type: 'error', message: `Task not found: ${msg.taskDirName}` });
|
||||
return;
|
||||
}
|
||||
|
||||
const jobId = await jobManager.startJob({
|
||||
userId,
|
||||
email,
|
||||
username,
|
||||
role,
|
||||
taskDirName: msg.taskDirName,
|
||||
taskName: task.name,
|
||||
inputs: msg.inputs,
|
||||
cwd: msg.cwd,
|
||||
config: task.config,
|
||||
});
|
||||
|
||||
send(ws, { type: 'job:created', jobId });
|
||||
jobManager.attachViewer(jobId, ws);
|
||||
break;
|
||||
}
|
||||
|
||||
case 'stop': {
|
||||
jobManager.stopJob(msg.jobId);
|
||||
break;
|
||||
}
|
||||
|
||||
case 'attach': {
|
||||
const attached = jobManager.attachViewer(msg.jobId, ws);
|
||||
if (!attached) {
|
||||
// Job not live — send the DB state
|
||||
const job = await jobManager.getJob(msg.jobId);
|
||||
if (job) {
|
||||
send(ws, { type: 'job:state', jobId: msg.jobId, status: job.status, progress: job.progress, cost: job.totalCost });
|
||||
} else {
|
||||
send(ws, { type: 'error', message: `Job not found: ${msg.jobId}` });
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'list': {
|
||||
const jobs = await jobManager.getJobsForUser(ws.data.userId);
|
||||
send(ws, { type: 'job:list', jobs: jobs.map((j) => ({ id: j.id, taskDirName: j.taskDirName, taskName: j.taskName, status: j.status, isLive: j.isLive, totalCost: j.totalCost, createdAt: j.createdAt, completedAt: j.completedAt })) });
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
@@ -357,11 +542,8 @@ export function message(ws: ServerWebSocket<WSData>, raw: string | Buffer) {
|
||||
}
|
||||
|
||||
export function close(ws: ServerWebSocket<WSData>) {
|
||||
const active = activePipelines.get(ws);
|
||||
if (active) {
|
||||
active.abort();
|
||||
activePipelines.delete(ws);
|
||||
}
|
||||
// Detach from all jobs but don't stop them
|
||||
jobManager.detachAllForViewer(ws);
|
||||
}
|
||||
|
||||
export const pipelineWebsocket = {
|
||||
|
||||
@@ -0,0 +1,293 @@
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
import { randomUUID } from 'crypto';
|
||||
import { EventEmitter } from 'events';
|
||||
import {
|
||||
createPipelineJob,
|
||||
getPipelineJob,
|
||||
updatePipelineJob,
|
||||
getPipelineJobsForUser,
|
||||
markInterruptedJobs,
|
||||
} from 'officerdb';
|
||||
import { executePipeline } from './pipeline-executor';
|
||||
import type { OutMessage } from './pipeline-executor';
|
||||
|
||||
type WSData = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
sandboxed: boolean;
|
||||
};
|
||||
|
||||
type LiveJob = {
|
||||
jobId: string;
|
||||
userId: number;
|
||||
abortSignal: { aborted: boolean };
|
||||
emitter: EventEmitter;
|
||||
eventBuffer: OutMessage[];
|
||||
viewers: Set<ServerWebSocket<WSData>>;
|
||||
progressDirty: boolean;
|
||||
lastProgress: unknown;
|
||||
lastCost: unknown;
|
||||
};
|
||||
|
||||
const EVENT_BUFFER_SIZE = 500;
|
||||
const PROGRESS_FLUSH_MS = 3000;
|
||||
|
||||
const liveJobs = new Map<string, LiveJob>();
|
||||
|
||||
function sendToViewer(ws: ServerWebSocket<WSData>, jobId: string, event: OutMessage) {
|
||||
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') {
|
||||
job.eventBuffer.push(event);
|
||||
if (job.eventBuffer.length > EVENT_BUFFER_SIZE) {
|
||||
job.eventBuffer.shift();
|
||||
}
|
||||
}
|
||||
|
||||
// Forward to all viewers
|
||||
for (const ws of job.viewers) {
|
||||
sendToViewer(ws, job.jobId, event);
|
||||
}
|
||||
}
|
||||
|
||||
type StartJobParams = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
taskDirName: string;
|
||||
taskName: string;
|
||||
inputs: Record<string, string>;
|
||||
cwd?: string;
|
||||
config: unknown;
|
||||
};
|
||||
|
||||
export async function startJob(params: StartJobParams): Promise<string> {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Create DB record
|
||||
await createPipelineJob({
|
||||
id: jobId,
|
||||
userId: params.userId,
|
||||
taskDirName: params.taskDirName,
|
||||
taskName: params.taskName,
|
||||
status: 'running',
|
||||
inputs: params.inputs,
|
||||
cwd: params.cwd ?? null,
|
||||
config: params.config,
|
||||
startedAt: new Date(),
|
||||
});
|
||||
|
||||
// Create live state
|
||||
const job: LiveJob = {
|
||||
jobId,
|
||||
userId: params.userId,
|
||||
abortSignal: { aborted: false },
|
||||
emitter: new EventEmitter(),
|
||||
eventBuffer: [],
|
||||
viewers: new Set(),
|
||||
progressDirty: false,
|
||||
lastProgress: null,
|
||||
lastCost: null,
|
||||
};
|
||||
liveJobs.set(jobId, job);
|
||||
|
||||
// Emit callback — broadcasts to viewers and tracks progress
|
||||
const emit = (event: OutMessage) => {
|
||||
broadcast(job, event);
|
||||
|
||||
// Track progress for DB persistence (skip streaming messages — only structural events)
|
||||
if (event.type === 'pipeline:init' || event.type.startsWith('step:') || event.type.startsWith('iteration:')) {
|
||||
job.progressDirty = true;
|
||||
job.lastProgress = extractProgress(event, job.lastProgress);
|
||||
}
|
||||
if (event.type === 'step:complete' || event.type === 'iteration:complete') {
|
||||
const cost = 'cost' in event ? event.cost : undefined;
|
||||
if (cost) {
|
||||
const prev = (job.lastCost as { inputTokens: number; outputTokens: number; totalUSD: number } | null) ?? { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
job.lastCost = {
|
||||
inputTokens: prev.inputTokens + cost.inputTokens,
|
||||
outputTokens: prev.outputTokens + cost.outputTokens,
|
||||
totalUSD: prev.totalUSD + cost.totalUSD,
|
||||
};
|
||||
job.progressDirty = true;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Periodic progress flush to DB
|
||||
const flushInterval = setInterval(async () => {
|
||||
if (job.progressDirty) {
|
||||
job.progressDirty = false;
|
||||
try {
|
||||
await updatePipelineJob(jobId, {
|
||||
progress: job.lastProgress as Record<string, unknown>,
|
||||
totalCost: job.lastCost as Record<string, unknown>,
|
||||
});
|
||||
} catch (err) {
|
||||
console.error(`[pipeline-jobs] failed to flush progress for ${jobId}:`, err);
|
||||
}
|
||||
}
|
||||
}, 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,
|
||||
abortSignal: job.abortSignal,
|
||||
emit,
|
||||
}).then(async () => {
|
||||
clearInterval(flushInterval);
|
||||
await updatePipelineJob(jobId, {
|
||||
status: 'completed',
|
||||
progress: job.lastProgress as Record<string, unknown>,
|
||||
totalCost: job.lastCost as Record<string, unknown>,
|
||||
completedAt: new Date(),
|
||||
}).catch((err) => console.error(`[pipeline-jobs] failed to finalize ${jobId}:`, err));
|
||||
liveJobs.delete(jobId);
|
||||
}).catch(async (err) => {
|
||||
clearInterval(flushInterval);
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
const isStopped = job.abortSignal.aborted;
|
||||
await updatePipelineJob(jobId, {
|
||||
status: isStopped ? 'stopped' : 'failed',
|
||||
progress: job.lastProgress as Record<string, unknown>,
|
||||
totalCost: job.lastCost as Record<string, unknown>,
|
||||
error: isStopped ? undefined : message,
|
||||
completedAt: new Date(),
|
||||
}).catch((e) => console.error(`[pipeline-jobs] failed to finalize ${jobId}:`, e));
|
||||
liveJobs.delete(jobId);
|
||||
});
|
||||
|
||||
return jobId;
|
||||
}
|
||||
|
||||
export function attachViewer(jobId: string, ws: ServerWebSocket<WSData>) {
|
||||
const job = liveJobs.get(jobId);
|
||||
if (!job) return false;
|
||||
|
||||
job.viewers.add(ws);
|
||||
|
||||
// Replay buffered events so the viewer catches up
|
||||
for (const event of job.eventBuffer) {
|
||||
sendToViewer(ws, jobId, event);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
export function detachViewer(jobId: string, ws: ServerWebSocket<WSData>) {
|
||||
const job = liveJobs.get(jobId);
|
||||
if (job) job.viewers.delete(ws);
|
||||
}
|
||||
|
||||
export function detachAllForViewer(ws: ServerWebSocket<WSData>) {
|
||||
for (const job of liveJobs.values()) {
|
||||
job.viewers.delete(ws);
|
||||
}
|
||||
}
|
||||
|
||||
export function stopJob(jobId: string): boolean {
|
||||
const job = liveJobs.get(jobId);
|
||||
if (!job) return false;
|
||||
job.abortSignal.aborted = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
export function isJobLive(jobId: string): boolean {
|
||||
return liveJobs.has(jobId);
|
||||
}
|
||||
|
||||
export function getLiveJobForUser(userId: number): string | null {
|
||||
for (const [jobId, job] of liveJobs) {
|
||||
if (job.userId === userId) return jobId;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function getJobsForUser(userId: number) {
|
||||
const jobs = await getPipelineJobsForUser(userId);
|
||||
return jobs.map((j) => ({
|
||||
...j,
|
||||
isLive: liveJobs.has(j.id),
|
||||
}));
|
||||
}
|
||||
|
||||
export async function getJob(jobId: string) {
|
||||
const job = await getPipelineJob(jobId);
|
||||
if (!job) return null;
|
||||
return { ...job, isLive: liveJobs.has(job.id) };
|
||||
}
|
||||
|
||||
export async function cleanupOnStartup() {
|
||||
const count = await markInterruptedJobs();
|
||||
if (count > 0) {
|
||||
console.log(`[pipeline-jobs] marked ${count} interrupted jobs on startup`);
|
||||
}
|
||||
}
|
||||
|
||||
/** Extract minimal progress info from events for DB storage. */
|
||||
function extractProgress(event: OutMessage, prev: unknown): unknown {
|
||||
const p = (prev as Record<string, unknown>) ?? {};
|
||||
|
||||
switch (event.type) {
|
||||
case 'pipeline:init':
|
||||
return { ...p, steps: event.steps, currentStepIndex: -1 };
|
||||
|
||||
case 'step:start':
|
||||
return { ...p, currentStepIndex: event.stepIndex, currentIteration: event.iteration ?? null };
|
||||
|
||||
case 'step:complete':
|
||||
return { ...p, currentStepIndex: event.stepIndex };
|
||||
|
||||
case 'step:parallel':
|
||||
return {
|
||||
...p,
|
||||
currentStepIndex: event.stepIndex,
|
||||
parallel: { taskName: event.taskName, concurrency: event.concurrency, iterations: event.iterations.map((l) => ({ label: l, status: 'pending' })) },
|
||||
};
|
||||
|
||||
case 'iteration:start':
|
||||
if (p.parallel && Array.isArray((p.parallel as Record<string, unknown>).iterations)) {
|
||||
const parallel = p.parallel as { iterations: Array<{ label: string; status: string }> };
|
||||
return {
|
||||
...p,
|
||||
parallel: {
|
||||
...parallel,
|
||||
iterations: parallel.iterations.map((it) => it.label === event.label ? { ...it, status: 'running' } : it),
|
||||
},
|
||||
};
|
||||
}
|
||||
return p;
|
||||
|
||||
case 'iteration:complete':
|
||||
case 'iteration:error':
|
||||
if (p.parallel && Array.isArray((p.parallel as Record<string, unknown>).iterations)) {
|
||||
const parallel = p.parallel as { iterations: Array<{ label: string; status: string }> };
|
||||
const status = event.type === 'iteration:complete' ? 'complete' : 'error';
|
||||
return {
|
||||
...p,
|
||||
parallel: {
|
||||
...parallel,
|
||||
iterations: parallel.iterations.map((it) => it.label === event.label ? { ...it, status } : it),
|
||||
},
|
||||
};
|
||||
}
|
||||
return p;
|
||||
|
||||
default:
|
||||
return p;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
import { createRouter } from '../../create-router';
|
||||
import { getPipelineJobsForUser, getPipelineJob } from 'officerdb';
|
||||
|
||||
export const pipelineJobsRouter = createRouter();
|
||||
|
||||
// GET /pipeline-jobs — list user's pipeline jobs
|
||||
pipelineJobsRouter.get('/', async (c) => {
|
||||
const user = c.get('user');
|
||||
const jobs = await getPipelineJobsForUser(user.id);
|
||||
return c.json(jobs.map((j) => ({
|
||||
id: j.id,
|
||||
taskDirName: j.taskDirName,
|
||||
taskName: j.taskName,
|
||||
status: j.status,
|
||||
totalCost: j.totalCost,
|
||||
createdAt: j.createdAt,
|
||||
startedAt: j.startedAt,
|
||||
completedAt: j.completedAt,
|
||||
error: j.error,
|
||||
})));
|
||||
});
|
||||
|
||||
// GET /pipeline-jobs/:id — single job detail
|
||||
pipelineJobsRouter.get('/:id', async (c) => {
|
||||
const user = c.get('user');
|
||||
const job = await getPipelineJob(c.req.param('id'));
|
||||
if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404);
|
||||
return c.json(job);
|
||||
});
|
||||
Reference in New Issue
Block a user