add pipeline task system, agentic music tasks, copy buttons, and pi-bridge cleanup
- Pipeline mode: new task mode that chains agentic tasks sequentially with foreach/subdirectory iteration and skip_if conditions - Pipeline executor backend (WebSocket at /api/tasks/pipeline/ws) with support for both Pi and Claude Code models - Frontend PipelineRunner component with step progress, streaming output, and aggregate cost tracking - New agentic tasks: prepare-discography, fetch-album-info, build-discography (pipeline combining both) - Seed parser extended to handle pipeline steps in frontmatter config - CopyButton component added to assistant bubbles, error bubbles, and tool input/output sections - Removed obsolete SearXNG/Apify/browser relay code from pi-bridge Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,372 @@
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
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 { resolveBaseCwd } from '../pi/websocket';
|
||||
import * as sidecar from '../../sidecar-registry';
|
||||
import { sendClaudeCodeStreaming } from '../../channels/send-claude-code';
|
||||
import type { PiEvent, MessageCost } from '../pi/types';
|
||||
|
||||
const DEFAULT_MODEL = 'claude-code';
|
||||
|
||||
async function resolveModel(userId: number): Promise<string> {
|
||||
try {
|
||||
const settings = await getUserSettings(userId);
|
||||
const tasks = settings?.tasks as Record<string, unknown> | undefined;
|
||||
return (tasks?.defaultModel as string) || DEFAULT_MODEL;
|
||||
} catch {
|
||||
return DEFAULT_MODEL;
|
||||
}
|
||||
}
|
||||
|
||||
type WSData = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
role: string;
|
||||
sandboxed: boolean;
|
||||
};
|
||||
|
||||
type PipelineStep = {
|
||||
task: string;
|
||||
inputs?: Record<string, string>;
|
||||
foreach?: 'subdirectory';
|
||||
skip_if?: 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 }> }
|
||||
| { 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: '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 AbortSignal = { aborted: boolean };
|
||||
|
||||
type RunStepParams = {
|
||||
ws: ServerWebSocket<WSData>;
|
||||
taskDirName: string;
|
||||
prompt: string;
|
||||
cwd: string;
|
||||
abortSignal: AbortSignal;
|
||||
};
|
||||
|
||||
async function runAgenticStep({ ws, taskDirName, prompt, cwd, abortSignal }: RunStepParams): Promise<MessageCost> {
|
||||
const { email, username, userId, role } = ws.data;
|
||||
const sessionId = randomUUID();
|
||||
const model = await resolveModel(userId);
|
||||
const isClaudeCode = model.startsWith('claude-code');
|
||||
|
||||
console.log(`[pipeline] starting step for session ${sessionId} (model=${model})`);
|
||||
|
||||
return new Promise<MessageCost>(async (resolve, reject) => {
|
||||
if (abortSignal.aborted) return reject(new Error('Pipeline aborted'));
|
||||
|
||||
let cleanup: (() => void) | null = null;
|
||||
|
||||
const onEvent = (event: PiEvent) => {
|
||||
if (abortSignal.aborted) return;
|
||||
|
||||
switch (event.type) {
|
||||
case 'delta':
|
||||
send(ws, { type: 'assistant:delta', text: event.text });
|
||||
break;
|
||||
case 'text':
|
||||
send(ws, { type: 'assistant:text', text: event.text });
|
||||
break;
|
||||
case 'tool:start':
|
||||
send(ws, { type: 'tool:start', toolCallId: event.toolCallId, toolName: event.toolName, toolInput: event.toolInput });
|
||||
break;
|
||||
case 'tool:result':
|
||||
send(ws, { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError });
|
||||
break;
|
||||
case 'result':
|
||||
cleanup?.();
|
||||
resolve(event.cost);
|
||||
break;
|
||||
case 'error':
|
||||
cleanup?.();
|
||||
reject(new Error(event.message));
|
||||
break;
|
||||
case 'stopped':
|
||||
cleanup?.();
|
||||
reject(new Error('Step was stopped'));
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
if (isClaudeCode) {
|
||||
const handle = await sendClaudeCodeStreaming({
|
||||
userId,
|
||||
email,
|
||||
username,
|
||||
prompt,
|
||||
sessionKey: sessionId,
|
||||
cwd,
|
||||
model,
|
||||
role,
|
||||
onEvent,
|
||||
});
|
||||
cleanup = handle.kill;
|
||||
} else {
|
||||
const unsub = sidecar.onPiEvent((evtSessionId, event) => {
|
||||
if (evtSessionId === sessionId) onEvent(event);
|
||||
});
|
||||
cleanup = () => {
|
||||
unsub();
|
||||
sidecar.killPi(sessionId);
|
||||
};
|
||||
|
||||
await sidecar.spawnPi({ sessionId, email, userId, username, role, cwd, model });
|
||||
sidecar.sendPiPrompt(sessionId, prompt, randomUUID());
|
||||
}
|
||||
} catch (err) {
|
||||
cleanup?.();
|
||||
reject(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function resolveInputTemplate(template: string, variables: Record<string, string>): string {
|
||||
return template.replace(/\$\{(\w+)\}/g, (_, key) => variables[key] ?? '');
|
||||
}
|
||||
|
||||
function buildStepPrompt(taskBody: string, inputs: Record<string, string>, targetDir?: string): string {
|
||||
const inputLines = Object.entries(inputs)
|
||||
.filter(([, v]) => v.trim())
|
||||
.map(([key, value]) => `- **${key}**: ${value}`)
|
||||
.join('\n');
|
||||
|
||||
const contextLines: string[] = [];
|
||||
if (targetDir) contextLines.push(`- **Target directory**: ${targetDir}`);
|
||||
const contextSection = contextLines.length > 0 ? `\n\n## Context\n\n${contextLines.join('\n')}` : '';
|
||||
|
||||
return `${taskBody.trim()}\n\n## Inputs\n\n${inputLines}${contextSection}`;
|
||||
}
|
||||
|
||||
async function handleRun(ws: ServerWebSocket<WSData>, msg: RunMessage) {
|
||||
const { email, role, userId } = ws.data;
|
||||
|
||||
const pipelineTask = await getTaskByDirName(msg.taskDirName, userId);
|
||||
if (!pipelineTask) {
|
||||
send(ws, { type: 'error', message: `Task not found: ${msg.taskDirName}` });
|
||||
return;
|
||||
}
|
||||
if (pipelineTask.mode !== 'pipeline') {
|
||||
send(ws, { 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' });
|
||||
return;
|
||||
}
|
||||
|
||||
const abortSignal = { aborted: false };
|
||||
activePipelines.set(ws, {
|
||||
abort: () => { abortSignal.aborted = true; },
|
||||
});
|
||||
|
||||
const baseCwd = resolveBaseCwd(email, role, msg.cwd);
|
||||
const totalCost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
|
||||
// Send pipeline init
|
||||
send(ws, {
|
||||
type: 'pipeline:init',
|
||||
steps: config.steps.map((s) => ({ task: s.task, foreach: s.foreach })),
|
||||
});
|
||||
|
||||
try {
|
||||
for (let stepIdx = 0; stepIdx < config.steps.length; stepIdx++) {
|
||||
if (abortSignal.aborted) break;
|
||||
|
||||
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}` });
|
||||
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;
|
||||
}
|
||||
|
||||
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` });
|
||||
continue;
|
||||
}
|
||||
|
||||
// Resolve folder-specific variables
|
||||
const iterVars = { ...msg.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);
|
||||
}
|
||||
}
|
||||
|
||||
send(ws, {
|
||||
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 resolvedCwd = resolveBaseCwd(email, role, cwdRelative);
|
||||
|
||||
// Build ~/relative path for context
|
||||
const targetDir = msg.cwd ? `~/${msg.cwd}/${subdir}` : `~/${subdir}`;
|
||||
const prompt = buildStepPrompt(stepTask.body, iterInputs, targetDir);
|
||||
|
||||
const cost = await runAgenticStep({
|
||||
ws,
|
||||
taskDirName: step.task,
|
||||
prompt,
|
||||
cwd: resolvedCwd,
|
||||
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
|
||||
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 });
|
||||
}
|
||||
}
|
||||
|
||||
if (!abortSignal.aborted) {
|
||||
send(ws, { type: 'pipeline:complete', totalCost });
|
||||
}
|
||||
} catch (err) {
|
||||
if (!abortSignal.aborted) {
|
||||
send(ws, { type: 'error', message: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
} finally {
|
||||
activePipelines.delete(ws);
|
||||
}
|
||||
}
|
||||
|
||||
export function open(_ws: ServerWebSocket<WSData>) {}
|
||||
|
||||
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 = activePipelines.get(ws);
|
||||
if (active) {
|
||||
active.abort();
|
||||
activePipelines.delete(ws);
|
||||
send(ws, { type: 'stopped' });
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
send(ws, { type: 'error', message: 'Failed to parse message' });
|
||||
}
|
||||
}
|
||||
|
||||
export function close(ws: ServerWebSocket<WSData>) {
|
||||
const active = activePipelines.get(ws);
|
||||
if (active) {
|
||||
active.abort();
|
||||
activePipelines.delete(ws);
|
||||
}
|
||||
}
|
||||
|
||||
export const pipelineWebsocket = {
|
||||
open,
|
||||
message,
|
||||
close,
|
||||
drain() {},
|
||||
};
|
||||
@@ -47,6 +47,7 @@ tasksRouter.get('/:name', async (ctx) => {
|
||||
inputs: task.inputs,
|
||||
args: task.args,
|
||||
trigger: task.trigger,
|
||||
config: task.config,
|
||||
version: task.version,
|
||||
userId: task.userId,
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user