diff --git a/seed/tasks/build-discography/TASK.md b/seed/tasks/build-discography/TASK.md new file mode 100644 index 00000000..02adc82d --- /dev/null +++ b/seed/tasks/build-discography/TASK.md @@ -0,0 +1,23 @@ +--- +name: Build Discography +description: Identify all albums in an artist folder, rename directories, fetch detailed info and cover art for each. +version: 1 +mode: pipeline +triggers: + - type: directory +inputs: + artist_name: + type: string + description: Artist / band name + autofill: entry_name +steps: + - task: prepare-discography + inputs: + artist_name: ${artist_name} + - task: fetch-album-info + foreach: subdirectory + skip_if: album-info.md + inputs: + artist_name: ${artist_name} + album_name: ${folder_name} +--- diff --git a/seed/tasks/fetch-album-info/TASK.md b/seed/tasks/fetch-album-info/TASK.md new file mode 100644 index 00000000..ff1a47d1 --- /dev/null +++ b/seed/tasks/fetch-album-info/TASK.md @@ -0,0 +1,97 @@ +--- +name: Fetch Album Info +description: Search the web for album information and save a detailed info file. +version: 1 +mode: agentic +triggers: + - type: directory +inputs: + artist_name: + type: string + description: Artist name + autofill: entry_name + album_name: + type: string + description: Album name + autofill: entry_name +--- + +# Fetch Album Info + +You are given an artist name and album name. Your job is to find comprehensive information about this album from the web. + +## Process + +1. **Search** for the album using `web_search` with queries like `" wikipedia"`, `" musicbrainz"`, `" allmusic"`. +2. **Fetch** the most relevant pages using `web_fetch` to extract detailed information. +3. **Cross-reference** multiple sources to get the most accurate and complete data. +4. **Download** the front cover image and save it as `cover.{ext}` (matching the image format, e.g. `cover.jpg`) in the target directory. Overwrite if it already exists. +5. **Save** the result as an `album-info.md` file in the target directory (provided in Context). + +## Information to collect + +- **Artist** name (canonical/correct spelling) +- **Album** title (canonical/correct spelling) +- **Release year** (original release) +- **Genre(s)** and subgenres +- **Label** / record company +- **Tracklist** — for each track: number, title, duration. For multi-disc releases, organize by disc. +- **Total duration** +- **Front cover image URL** — find the highest quality cover art URL available (Wikipedia, MusicBrainz cover art archive) +- **Credits** — producer, engineer, notable musicians (if available) +- **Additional notes** — compilation info, remaster details, notable facts + +## Output format + +Write a markdown file named `album-info.md` in the **target directory** (from Context) with this structure: + +```markdown +# Artist — Album Title + +![Cover](url-to-cover-image) + +| Field | Value | +|-------|-------| +| Artist | ... | +| Album | ... | +| Year | ... | +| Genre | ... | +| Label | ... | +| Duration | ... | + +## Tracklist + +### Disc 1 +| # | Title | Duration | +|---|-------|----------| +| 1 | ... | 0:00 | + +### Disc 2 +... + +## Credits +- Producer: ... +- ... + +## Notes +... +``` + +If there is only one disc, omit the "Disc 1" heading and just use a flat tracklist. + +## Cover art download + +Once you have a cover image URL, download it to the target directory using curl: +``` +curl -sL -o "/cover." "" +``` +Use the correct extension based on the image format (jpg, png, etc.). Overwrite any existing file. +In the `album-info.md`, keep the original remote URL in the image tag: `![Cover]()`. + +## Important + +- Prefer Wikipedia for context, genre classification, credits, and tracklist data. +- Use MusicBrainz / Cover Art Archive for high-quality cover image URLs and precise edition identification. +- Use AllMusic as an additional source for credits and reviews. +- If information conflicts between sources, prefer Wikipedia for metadata and MusicBrainz for tracklist data. +- Do NOT make up information. If something is not found, omit it. diff --git a/seed/tasks/prepare-discography/TASK.md b/seed/tasks/prepare-discography/TASK.md new file mode 100644 index 00000000..f71511c5 --- /dev/null +++ b/seed/tasks/prepare-discography/TASK.md @@ -0,0 +1,117 @@ +--- +name: Prepare Discography +description: Scan an artist's album folders, identify exact release editions, and produce a discography summary. +version: 1 +mode: agentic +triggers: + - type: directory +inputs: + artist_name: + type: string + description: Artist / band name + autofill: entry_name +--- + +# Prepare Discography + +You are given an artist/band directory containing one subfolder per album. Your job is to identify the **exact release edition** of each album and produce a `discography.md` summary file. + +## Why this matters + +Album folders often contain special editions, anniversary reissues, deluxe multi-disc sets, etc. A naive search for "The Doors 1967" returns the original 11-track release, but the folder might contain a 50th Anniversary 3CD edition with 30+ tracks. You must identify the **specific edition** the user has. + +## Process + +For each album subfolder in the artist directory: + +### 1. Analyze local signals + +Gather as much information as possible from the folder itself **before** searching the web: + +- **Folder name**: often contains year and album title, sometimes edition hints like `(3CD)`, `(Deluxe)`, `(Remaster)` +- **Disc structure**: count `CD1/`, `CD2/`, etc. subdirectories — this tells you how many discs the release has +- **Track count per disc**: list audio files in each disc folder (or root if single-disc) +- **Track names**: the filenames often contain track titles (e.g. `01 - Break on Through.flac`) +- **Audio tags**: use the `mutagen` tool with `action: read` on 1-2 representative tracks per disc to get tagged metadata (artist, album, year, genre, label, etc.) +- **Existing files**: check for `album-info.md`, `Front.jpg`, `cover.jpg`, or `Scans/` — these provide additional context + +### 2. Build a search profile + +From the local signals, construct a profile: +- Artist name (from tags or folder name) +- Album title (from tags or folder name) +- Release year (from tags or folder name) +- Number of discs +- Track count per disc +- Any edition keywords (deluxe, remaster, anniversary, etc.) + +### 3. Search for the exact release + +Use `web_search` to find the specific edition: +- Search Wikipedia first: `" wikipedia"` — Wikipedia often has edition/reissue details +- Search MusicBrainz: `" musicbrainz"` — MusicBrainz catalogs every pressing and edition +- If the disc/track count doesn't match the first result, refine: `" deluxe 3CD"` or similar +- Use `web_fetch` on the most relevant pages to confirm the tracklist matches your local files + +### 4. Confirm the match + +Compare the web result against local signals: +- Does the disc count match? +- Does the track count per disc match (approximately)? +- Do track names align? +- If there's a mismatch, search for alternative editions until you find the best match + +### 5. Rename the album folder + +Once you have confidently identified the release, rename the album folder to the standardized format: + +``` +[YYYY] Album Title (Edition Marker) +``` + +- **YYYY** = original release year (NOT the reissue/special edition year) +- **Album Title** = canonical album name +- **Edition Marker** = only if it's not the standard original release. Examples: `50th Anniversary 3CD Deluxe Edition`, `2017 Remaster`, `Deluxe Edition` +- If it IS the plain original release, omit the parenthetical: `[1967] The Doors` + +Use `mv` via Bash to rename. Do this **before** writing discography.md so the file references the new folder names. + +## Output + +Write a `discography.md` file in the **artist directory** (the target directory from Context) with this structure: + +```markdown +# Artist Name — Discography + +## Albums + +### [YYYY] Album Title (Edition Marker) +- **Folder**: `[YYYY] Album Title (Edition Marker)/` +- **Format**: FLAC / MP3 / Mixed +- **Discs**: N +- **Tracks**: N (or N per disc: D1: X, D2: Y, ...) +- **Year**: YYYY (original release) / YYYY (this edition) +- **Label**: ... +- **Genre**: ... +- **Edition**: 50th Anniversary Deluxe Edition / Original / Remaster / etc. +- **Wikipedia**: https://en.wikipedia.org/wiki/... (if found) +- **MusicBrainz**: https://musicbrainz.org/release/... (if found) +- **Notes**: ... + +### [YYYY] Another Album +... +``` + +Sort albums chronologically by original release year. + +Include **direct URLs** to the sources used (Discogs release page, Wikipedia article, MusicBrainz release) — these will be used by downstream tasks to avoid redundant searching. + +## Important + +- **Accuracy over speed**: it's better to correctly identify 3 out of 4 albums than to guess all 4 wrong +- **Wikipedia and MusicBrainz are your primary sources** for edition identification +- **Use mutagen sparingly**: read 1-2 tracks per disc, not every file +- **Track count is the strongest signal** for distinguishing editions — a 3CD set with 10+12+8 tracks is very different from a single-disc 11-track original +- **Folder name hints**: `(3CD)`, `(2LP)`, `(Deluxe)`, `(Remaster)`, `(Anniversary)` in the folder name are strong clues +- **Do NOT make up information**. If you cannot confidently identify an edition, note what you found and flag the uncertainty +- The artist name in the folder may be formatted as `Last, First` or `Name, The` — normalize it when searching (e.g. `Doors, the` → `The Doors`) diff --git a/seed/templates/SCRIPTS.md b/seed/templates/SCRIPTS.md index 37120335..f13a0fec 100644 --- a/seed/templates/SCRIPTS.md +++ b/seed/templates/SCRIPTS.md @@ -20,6 +20,17 @@ All task inputs defined in the TASK.md frontmatter are available to the script i Inputs provided by context (e.g. `file_path` from the file browser) are auto-filled and hidden from the form. +### Autofill + +Inputs can declare `autofill` to pre-fill from the trigger context. Available context values: + +| Value | Source | +|-------|--------| +| `entry_name` | Name of the file or directory that triggered the task | +| `entry_path` | Full path of the file or directory | + +Autofilled inputs are visible and editable (unlike `file_path` which is hidden). + ## Templates ### 1. File Processor @@ -180,6 +191,7 @@ inputs: type: string # string, number, boolean description: What this input is default: value # optional — default value + autofill: entry_name # optional — pre-fill from context (entry_name or entry_path) options: # optional — renders as selectable pills in UI - option1 - option2 diff --git a/src/databases/officer_db/seed-tasks.ts b/src/databases/officer_db/seed-tasks.ts index cf13a08c..cb0b3973 100644 --- a/src/databases/officer_db/seed-tasks.ts +++ b/src/databases/officer_db/seed-tasks.ts @@ -99,6 +99,59 @@ function parseFrontmatter(content: string) { } } + // Parse pipeline steps + const stepsMatch = yaml.match(/^steps:\s*\n((?:[ \t]+.+\n?)*)/m); + if (stepsMatch) { + const stepsBlock = stepsMatch[1]!; + const steps: Record[] = []; + let currentStep: Record | null = null; + let inInputs = false; + let stepInputs: Record = {}; + + for (const line of stepsBlock.split('\n')) { + // New step entry (2 spaces + dash) + const stepStart = line.match(/^\s{2}-\s+task:\s*(.+)$/); + if (stepStart) { + if (currentStep) { + if (Object.keys(stepInputs).length > 0) currentStep.inputs = stepInputs; + steps.push(currentStep); + } + currentStep = { task: stepStart[1]!.trim() }; + stepInputs = {}; + inInputs = false; + continue; + } + if (!currentStep) continue; + + // Step-level properties (4 spaces) + const propMatch = line.match(/^\s{4}(\w[\w_-]*):\s*(.*)$/); + if (propMatch) { + const key = propMatch[1]!; + const value = propMatch[2]!.trim(); + if (key === 'inputs' && value === '') { + inInputs = true; + } else { + inInputs = false; + currentStep[key] = value; + } + continue; + } + + // Step input entries (6 spaces) + const inputMatch = line.match(/^\s{6}(\w[\w_-]*):\s*(.+)$/); + if (inputMatch && inInputs) { + stepInputs[inputMatch[1]!] = inputMatch[2]!.trim(); + } + } + if (currentStep) { + if (Object.keys(stepInputs).length > 0) currentStep.inputs = stepInputs; + steps.push(currentStep); + } + if (steps.length > 0) { + meta.config = { steps }; + } + } + return { meta, body }; } @@ -151,6 +204,7 @@ async function seedTasks() { args: (meta.args as string[]) || null, trigger: (meta.trigger as TriggerConfig[]) || null, inputs: (meta.inputs as Record) || null, + config: (meta.config as Record) || null, }; // Upsert: check by dirName + scope since unique constraint doesn't work with NULL userId @@ -174,6 +228,7 @@ async function seedTasks() { args: values.args, trigger: values.trigger, inputs: values.inputs, + config: values.config, updatedAt: new Date(), }) .where(eq(tasks.id, existing[0]!.id)); diff --git a/src/server.tsx b/src/server.tsx index 8e20ce03..d962c554 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -7,6 +7,7 @@ import { isTokenBlacklisted } from 'officerdb'; import { terminalWebsocket } from './servers/api/terminal/websocket'; import { piWebsocket } from './servers/api/pi/websocket'; import { taskRunnerWebsocket } from './servers/api/tasks/task-executor'; +import { pipelineWebsocket } from './servers/api/tasks/pipeline-executor'; import { cliampWebsocket } from './servers/api/cliamp/websocket'; import { cliampAudioWebsocket } from './servers/api/cliamp/audio-ws'; import { desktopWebsocket } from './servers/api/desktop/websocket'; @@ -24,7 +25,7 @@ type WSData = { email: string; username: string; role: string; - provider: 'terminal' | 'pi' | 'task-runner' | 'dev-server' | 'cliamp' | 'cliamp-audio' | 'desktop' | 'sidecar'; + provider: 'terminal' | 'pi' | 'task-runner' | 'pipeline' | 'dev-server' | 'cliamp' | 'cliamp-audio' | 'desktop' | 'sidecar'; sandboxed: boolean; sessionId?: string; cwd?: string; @@ -118,6 +119,7 @@ const handlers: Record = { terminal: terminalWebsocket, pi: piWebsocket, 'task-runner': taskRunnerWebsocket, + pipeline: pipelineWebsocket, cliamp: cliampWebsocket, 'cliamp-audio': cliampAudioWebsocket, desktop: desktopWebsocket, @@ -189,7 +191,7 @@ const devServerWebsocket = { }; handlers['dev-server'] = devServerWebsocket; -async function upgradeWs(req: Request, server: any, provider: 'terminal' | 'pi' | 'task-runner' | 'cliamp' | 'cliamp-audio' | 'desktop') { +async function upgradeWs(req: Request, server: any, provider: 'terminal' | 'pi' | 'task-runner' | 'pipeline' | 'cliamp' | 'cliamp-audio' | 'desktop') { const token = new URL(req.url).searchParams.get('token'); if (!token) return new Response('Unauthorized', { status: 401 }); @@ -274,6 +276,7 @@ const server = serve({ if (!ok) return new Response('Upgrade failed', { status: 500 }); }, '/api/tasks/run/ws': (req, server) => upgradeWs(req, server, 'task-runner'), + '/api/tasks/pipeline/ws': (req, server) => upgradeWs(req, server, 'pipeline'), '/api/terminal/ws': (req, server) => upgradeWs(req, server, 'terminal'), '/api/pi/chat/ws': (req, server) => upgradeWs(req, server, 'pi'), '/api/cliamp/ws': (req, server) => upgradeWs(req, server, 'cliamp'), diff --git a/src/servers/api/pi/pi-bridge.ts b/src/servers/api/pi/pi-bridge.ts index 6a84cdbf..4f884a90 100644 --- a/src/servers/api/pi/pi-bridge.ts +++ b/src/servers/api/pi/pi-bridge.ts @@ -14,10 +14,7 @@ import { getUserToolsDir, toShellUsername, } from '../../data-path'; -import { getServerIntegration, getUserIntegration, readConfigValue } from 'officerdb'; import { logger } from './logger'; -import { getRelayPort } from '../browser/relay'; -import { registerUserToken } from '../browser/relay-auth'; // Resolve pi as [node, cli.js] — Bun.spawn async pipes break with shebang scripts under pm2 const PI_CMD = (() => { @@ -71,32 +68,6 @@ function collectExtensionFlags(email: string): string[] { return flags; } -async function getApifyToken(): Promise { - try { - const integration = await getServerIntegration('apify'); - const config = integration?.config as Record | undefined; - return config?.apiToken ?? ''; - } catch { - return ''; - } -} - -async function getBrowserRelayEnv(userId: number): Promise> { - const port = getRelayPort(); - if (!port) return {}; - try { - const integration = await getUserIntegration(userId, 'browser-relay'); - const salt = (integration?.config as { tokenSalt?: string })?.tokenSalt; - if (!salt) return {}; - const token = registerUserToken(userId, port, salt); - return { - OFFICER_BROWSER_RELAY_PORT: String(port), - OFFICER_BROWSER_RELAY_TOKEN: token, - }; - } catch { - return {}; - } -} async function resolveApiKeyForModel(model: string): Promise { const provider = model.split('/')[0]; @@ -125,7 +96,6 @@ export async function spawnPi( onEvent: PiEventHandler, options?: SpawnPiOptions, ): Promise { - const searxngUrl = await readConfigValue('searxng-url', ''); const skillFlags = collectSkillFlags(email); const extensionFlags = collectExtensionFlags(email); @@ -151,8 +121,6 @@ export async function spawnPi( const homeDir = getHomeDirForRole(email, options?.role ?? null); const toolsDirs = [getGlobalToolsDir(), getUserToolsDir(email)].join(':'); - const browserRelayEnv = await getBrowserRelayEnv(userId); - const apifyToken = await getApifyToken(); const shellUsername = options?.username ?? toShellUsername('', email); const isServiceUser = (options?.username ?? toShellUsername('', email)) === (process.env.USER ?? ''); @@ -163,12 +131,9 @@ export async function spawnPi( OFFICER_USER_ROOT: join(DATA_PATH, email), PI_CODING_AGENT_DIR: isServiceUser ? PI_CONFIG_DIR : join(homeDir, '.pi', 'agent'), PI_TOOLS_DIRS: toolsDirs, - PI_SEARXNG_URL: searxngUrl, OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'), TERM: 'xterm-256color', PATH: process.env.PATH ?? '', - ...(apifyToken ? { OFFICER_APIFY_TOKEN: apifyToken } : {}), - ...browserRelayEnv, }; // For service user, keep real HOME so Pi finds its config @@ -453,9 +418,7 @@ export function killPi(process: Subprocess): void { } /** Build env vars needed by Officer tools when running on the host. */ -export async function buildHostToolEnv(userId: number, email: string, role?: string): Promise> { - const browserRelayEnv = await getBrowserRelayEnv(userId); - const apifyToken = await getApifyToken(); +export async function buildHostToolEnv(email: string, role?: string): Promise> { const homeDir = getHomeDirForRole(email, role ?? null); return { @@ -463,7 +426,5 @@ export async function buildHostToolEnv(userId: number, email: string, role?: str OFFICER_USER_HOME: homeDir, OFFICER_USER_ROOT: join(DATA_PATH, email), OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'), - ...(apifyToken ? { OFFICER_APIFY_TOKEN: apifyToken } : {}), - ...browserRelayEnv, }; } diff --git a/src/servers/api/tasks/pipeline-executor.ts b/src/servers/api/tasks/pipeline-executor.ts new file mode 100644 index 00000000..adc76dac --- /dev/null +++ b/src/servers/api/tasks/pipeline-executor.ts @@ -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 { + try { + const settings = await getUserSettings(userId); + const tasks = settings?.tasks as Record | 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; + foreach?: 'subdirectory'; + skip_if?: string; +}; + +type PipelineConfig = { + steps: PipelineStep[]; +}; + +type RunMessage = { + type: 'run'; + taskDirName: string; + inputs: Record; + 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 } + | { 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, { abort: () => void }>(); + +function send(ws: ServerWebSocket, msg: OutMessage) { + if (ws.readyState === 1) ws.send(JSON.stringify(msg)); +} + +type AbortSignal = { aborted: boolean }; + +type RunStepParams = { + ws: ServerWebSocket; + taskDirName: string; + prompt: string; + cwd: string; + abortSignal: AbortSignal; +}; + +async function runAgenticStep({ ws, taskDirName, prompt, cwd, abortSignal }: RunStepParams): Promise { + 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(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 { + return template.replace(/\$\{(\w+)\}/g, (_, key) => variables[key] ?? ''); +} + +function buildStepPrompt(taskBody: string, inputs: Record, 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, 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 = {}; + 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 = {}; + 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) {} + +export function message(ws: ServerWebSocket, 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) { + const active = activePipelines.get(ws); + if (active) { + active.abort(); + activePipelines.delete(ws); + } +} + +export const pipelineWebsocket = { + open, + message, + close, + drain() {}, +}; diff --git a/src/servers/api/tasks/tasks.ts b/src/servers/api/tasks/tasks.ts index 1b4c700e..2c1c9174 100644 --- a/src/servers/api/tasks/tasks.ts +++ b/src/servers/api/tasks/tasks.ts @@ -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, }); diff --git a/src/workspaces/officerdev/src/apps/Chat/components/CopyButton.tsx b/src/workspaces/officerdev/src/apps/Chat/components/CopyButton.tsx new file mode 100644 index 00000000..46e3f863 --- /dev/null +++ b/src/workspaces/officerdev/src/apps/Chat/components/CopyButton.tsx @@ -0,0 +1,28 @@ +import { useState } from 'react'; +import { Copy, Check } from 'lucide-react'; + +type CopyButtonProps = { + text: string; + className?: string; +}; + +export const CopyButton = ({ text, className = '' }: CopyButtonProps) => { + const [copied, setCopied] = useState(false); + + const handleCopy = () => { + navigator.clipboard.writeText(text.trim()); + setCopied(true); + setTimeout(() => setCopied(false), 2000); + }; + + return ( + + ); +}; diff --git a/src/workspaces/officerdev/src/apps/Chat/components/MessageBubble.tsx b/src/workspaces/officerdev/src/apps/Chat/components/MessageBubble.tsx index ed446dc6..e53fd7c7 100644 --- a/src/workspaces/officerdev/src/apps/Chat/components/MessageBubble.tsx +++ b/src/workspaces/officerdev/src/apps/Chat/components/MessageBubble.tsx @@ -9,6 +9,7 @@ import { ToolActivity } from './ToolActivity'; import { QuestionActivity } from './QuestionActivity'; import { getRawUrl } from '../../FileViewer/file-types'; import { useFilesAPI } from '../../../hooks/useFilesAPI'; +import { CopyButton } from './CopyButton'; const sanitizeSchema = { ...defaultSchema, @@ -135,7 +136,7 @@ export const MessageBubble = ({ message, onAnswer }: MessageBubbleProps) => { const assistantText = typeof message.text === 'string' ? message.text : ''; if (!assistantText) return null; return ( -
+
@@ -143,7 +144,8 @@ export const MessageBubble = ({ message, onAnswer }: MessageBubbleProps) => { {injectImages(assistantText)}
-
+
+
@@ -169,9 +171,12 @@ export const MessageBubble = ({ message, onAnswer }: MessageBubbleProps) => { case 'error': return ( -
-
+
+
{message.text} +
+ +
); diff --git a/src/workspaces/officerdev/src/apps/Chat/components/ToolActivity.tsx b/src/workspaces/officerdev/src/apps/Chat/components/ToolActivity.tsx index d484635c..e94b19d4 100644 --- a/src/workspaces/officerdev/src/apps/Chat/components/ToolActivity.tsx +++ b/src/workspaces/officerdev/src/apps/Chat/components/ToolActivity.tsx @@ -1,6 +1,7 @@ import { useState } from 'react'; import { FileText, Terminal, Pencil, Search, Globe, Wrench, ChevronRight } from 'lucide-react'; import type { ChatMessage } from '../types'; +import { CopyButton } from './CopyButton'; type ToolMessage = Extract; @@ -74,8 +75,16 @@ export const ToolActivity = ({ message }: ToolActivityProps) => { {open && (
-
-
Input
+
+
+
Input
+ `${k}: ${typeof v === 'string' ? v : JSON.stringify(v)}`).join('\n')} + className="!opacity-0 group-hover/input:!opacity-60 hover:!opacity-100" + /> +
{message.toolName === 'Bash' ? ( 
                 {(message.toolInput.command as string) ?? JSON.stringify(message.toolInput, null, 2)}
@@ -90,8 +99,11 @@ export const ToolActivity = ({ message }: ToolActivityProps) => {
{message.output !== undefined && ( -
-
Output
+
+
+
Output
+ +
)} diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx index d091a584..00535959 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx @@ -10,6 +10,7 @@ import { useUserVisibleModels } from 'state/useModels'; import { useClient } from 'hooks/useClient'; import type { TaskSummary } from '../../useTasks'; import { useTaskRunner } from './useTaskRunner'; +import { usePipelineRunner } from './usePipelineRunner'; const playDing = () => { const ctx = new AudioContext(); @@ -38,17 +39,43 @@ const playDing = () => { type Phase = 'ready' | 'running' | 'done'; type PiMonoInnerProps = { + taskDirName: string; defaultInput: string; cwd: { root?: string; path: string }; initialModel: string | null; taskInfo: TaskInfo; sandboxed?: boolean; + context: Record; }; -const PiMonoInner = ({ defaultInput, cwd, initialModel, taskInfo, sandboxed }: PiMonoInnerProps) => { +const PiMonoInner = ({ taskDirName, defaultInput, cwd, initialModel, taskInfo, sandboxed, context }: PiMonoInnerProps) => { const [phase, setPhase] = useState('ready'); const chat = usePiChat(undefined, initialModel, { replaceUrl: false, taskInfo }); const availableModels = useUserVisibleModels(); + const client = useClient(); + const [inputDefs, setInputDefs] = useState | null>(null); + const [formValues, setFormValues] = useState>({}); + const [taskBody, setTaskBody] = useState(null); + + // Fetch task detail for inputs and body + useEffect(() => { + client.get<{ inputs?: Record; body?: string }>(`/tasks/${taskDirName}`).then((task) => { + const defs = task.inputs ?? {}; + setInputDefs(defs); + setTaskBody(task.body ?? null); + // Initialize from defaults and autofill + const initial: Record = {}; + for (const [key, def] of Object.entries(defs)) { + if (def.autofill && context[def.autofill]) initial[key] = context[def.autofill]!; + else if (def.default !== undefined) initial[key] = def.default; + } + setFormValues(initial); + }); + }, [taskDirName]); + + const handleInputChange = (key: string, value: string) => { + setFormValues((prev) => ({ ...prev, [key]: value })); + }; // --- Independent message accumulator (never loses messages) --- const accRef = useRef([]); @@ -114,17 +141,42 @@ const PiMonoInner = ({ defaultInput, cwd, initialModel, taskInfo, sandboxed }: P }, [chat.isGenerating]); const handleRun = () => { + // Build prompt from task body + input values, or fall back to generic prompt + let prompt = defaultInput; + if (taskBody && inputDefs) { + const inputLines = Object.entries(formValues) + .filter(([, v]) => v.trim()) + .map(([key, value]) => { + const label = inputDefs[key]?.description ?? key; + return `- **${label}**: ${value}`; + }) + .join('\n'); + const contextLines: string[] = []; + if (context.entry_path) contextLines.push(`- **Target directory**: ${context.entry_path}`); + const contextSection = contextLines.length > 0 ? `\n\n## Context\n\n${contextLines.join('\n')}` : ''; + prompt = `${taskBody.trim()}\n\n## Inputs\n\n${inputLines}${contextSection}`; + } setPhase('running'); - chat.sendPrompt(defaultInput, undefined, undefined, cwd, undefined, sandboxed); + chat.sendPrompt(prompt, undefined, undefined, cwd, undefined, sandboxed); }; + const hasConfigurableInputs = inputDefs && Object.keys(inputDefs).length > 0; + if (phase === 'ready') { return ( <> + {hasConfigurableInputs && ( + + )}
+
+
+ ); + } + + return ( +
+ {/* Step progress header */} + {pipeline.currentStep && ( +
+
+ {pipeline.currentStep.taskName} + {pipeline.currentStep.iteration && ( + + ({pipeline.currentStep.iteration.current}/{pipeline.currentStep.iteration.total}) + {pipeline.currentStep.iteration.label} + + )} + {pipeline.currentStep.status === 'running' && ( + + )} + {pipeline.currentStep.status === 'complete' && ( + done + )} +
+
+ )} + + {/* Messages */} +
+ {pipeline.messages.map((msg, i) => ( +
+ {}} /> +
+ ))} + {pipeline.streamingText && ( +
+ +
+ )} +
+
+ + {/* Footer */} +
+ {pipeline.phase === 'running' ? ( + + ) : pipeline.hasError ? ( + + + Pipeline failed + + ) : ( + + + Pipeline complete + + )} + {pipeline.totalCost && ( + + ${pipeline.totalCost.totalUSD.toFixed(3)} · {pipeline.totalCost.inputTokens + pipeline.totalCost.outputTokens} tokens + + )} + {pipeline.skippedItems.length > 0 && ( + + {pipeline.skippedItems.length} skipped ({pipeline.skippedItems.map((s) => s.label).join(', ')}) + + )} +
+
+ ); +}; + type TaskRunnerModalProps = { open: boolean; onOpenChange: (open: boolean) => void; @@ -439,14 +646,22 @@ export const TaskRunnerModal = ({ open, onOpenChange, task, entryName, entryFull const taskSettings = settings.tasks; const entryRef = entryFullPath ?? entryName; const isScript = task.mode === 'script'; + const isPipeline = task.mode === 'pipeline'; - // Agentic mode prompt + // Agentic mode prompt (fallback if task has no body) const defaultInput = promptOverride ?? (entryRef && entryType ? `Execute the task "${task.name}" (${task.dirName}) on the ${entryType}: ${entryRef}` : `Execute the task "${task.name}" (${task.dirName})`); const taskInfo: TaskInfo = { taskName: task.name, taskDirName: task.dirName, entryName: entryName ?? '', entryType: entryType ?? 'file' }; + // Context values for autofill + // Build a ~/relative path for the agent (works inside bwrap sandbox) + const entryRelPath = entryName && cwd.path ? `~/${cwd.path}/${entryName}` : entryName ? `~/${entryName}` : undefined; + const autofillContext: Record = {}; + if (entryName) autofillContext.entry_name = entryName; + if (entryRelPath) autofillContext.entry_path = entryRelPath; + // Script mode: auto-filled inputs from context (e.g. file_path from file browser) const autoInputs: Record = {}; if (entryFullPath) autoInputs.file_path = entryFullPath; @@ -477,21 +692,31 @@ export const TaskRunnerModal = ({ open, onOpenChange, task, entryName, entryFull
{/* Task Runner — branch on mode */} - {isScript ? ( + {isPipeline ? ( + + ) : isScript ? ( ) : ( )} diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts new file mode 100644 index 00000000..1aa955e7 --- /dev/null +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts @@ -0,0 +1,185 @@ +import { useState, useEffect, useRef, useCallback } from 'react'; +import type { ChatMessage } from '../../../Chat'; + +type Phase = 'ready' | 'running' | 'done'; + +type StepDef = { + task: string; + foreach?: string; +}; + +type StepStatus = { + taskName: string; + iteration?: { current: number; total: number; label: string }; + status: 'pending' | 'running' | 'complete' | 'skipped'; + cost?: { inputTokens: number; outputTokens: number; totalUSD: number }; +}; + +type ServerMessage = + | { type: 'pipeline:init'; steps: StepDef[] } + | { type: 'step:start'; stepIndex: number; taskName: string; iteration?: { current: number; total: number; label: string } } + | { type: 'step:complete'; stepIndex: number; cost?: { inputTokens: number; outputTokens: number; totalUSD: number } } + | { 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 } + | { type: 'tool:result'; toolCallId: string; output: string; isError: boolean } + | { type: 'pipeline:complete'; totalCost: { inputTokens: number; outputTokens: number; totalUSD: number } } + | { type: 'error'; message: string } + | { type: 'stopped' }; + +export function usePipelineRunner() { + const [phase, setPhase] = useState('ready'); + const [isConnected, setIsConnected] = useState(false); + const [steps, setSteps] = useState([]); + const [currentStep, setCurrentStep] = useState(null); + const [messages, setMessages] = useState([]); + const [streamingText, setStreamingText] = useState(''); + const [totalCost, setTotalCost] = useState<{ inputTokens: number; outputTokens: number; totalUSD: number } | null>(null); + const [hasError, setHasError] = useState(false); + const [skippedItems, setSkippedItems] = useState>([]); + const wsRef = useRef(null); + const streamBufferRef = useRef(''); + + const flushStream = useCallback(() => { + const text = streamBufferRef.current; + if (text) { + setMessages((prev) => [...prev, { role: 'assistant', id: crypto.randomUUID(), text }]); + streamBufferRef.current = ''; + setStreamingText(''); + } + }, []); + + useEffect(() => { + const token = localStorage.getItem('BEARER_TOKEN'); + if (!token) return; + + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const url = `${protocol}//${window.location.host}/api/tasks/pipeline/ws?token=${token}`; + const ws = new WebSocket(url); + wsRef.current = ws; + + ws.addEventListener('open', () => setIsConnected(true)); + ws.addEventListener('close', () => setIsConnected(false)); + + ws.addEventListener('message', (ev) => { + try { + const msg = JSON.parse(ev.data) as ServerMessage; + + switch (msg.type) { + case 'pipeline:init': + setSteps(msg.steps); + break; + + case 'step:start': + // Flush any streaming text from the previous step + flushStream(); + // Clear messages for the new step iteration + setMessages([]); + setCurrentStep({ + taskName: msg.taskName, + iteration: msg.iteration, + status: 'running', + }); + break; + + case 'step:complete': + flushStream(); + setCurrentStep((prev) => prev ? { ...prev, status: 'complete', cost: msg.cost } : null); + break; + + case 'step:skip': + setSkippedItems((prev) => [...prev, { label: msg.label, reason: msg.reason }]); + break; + + case 'assistant:delta': + streamBufferRef.current += msg.text; + setStreamingText(streamBufferRef.current); + break; + + case 'assistant:text': { + const text = msg.text || streamBufferRef.current; + if (text) { + setMessages((prev) => [...prev, { role: 'assistant', id: crypto.randomUUID(), text }]); + } + streamBufferRef.current = ''; + setStreamingText(''); + break; + } + + case 'tool:start': + flushStream(); + setMessages((prev) => [ + ...prev, + { + role: 'tool' as const, + id: crypto.randomUUID(), + toolCallId: msg.toolCallId, + toolName: msg.toolName, + toolInput: msg.toolInput, + output: undefined, + isError: false, + }, + ]); + break; + + case 'tool:result': + setMessages((prev) => + prev.map((m) => + m.role === 'tool' && 'toolCallId' in m && m.toolCallId === msg.toolCallId + ? { ...m, output: msg.output, isError: msg.isError } + : m, + ), + ); + break; + + case 'pipeline:complete': + flushStream(); + setTotalCost(msg.totalCost); + setPhase('done'); + break; + + case 'error': + flushStream(); + setMessages((prev) => [...prev, { role: 'error' as const, id: crypto.randomUUID(), text: msg.message }]); + setHasError(true); + setPhase('done'); + break; + + case 'stopped': + flushStream(); + setPhase('done'); + break; + } + } catch { + // ignore + } + }); + + return () => { + ws.close(); + wsRef.current = null; + }; + }, []); + + const run = useCallback((taskDirName: string, inputs: Record, cwd?: string) => { + if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return; + + setPhase('running'); + setMessages([]); + setStreamingText(''); + setTotalCost(null); + setHasError(false); + setSkippedItems([]); + streamBufferRef.current = ''; + + wsRef.current.send(JSON.stringify({ type: 'run', taskDirName, inputs, cwd })); + }, []); + + const stop = useCallback(() => { + if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return; + wsRef.current.send(JSON.stringify({ type: 'stop' })); + }, []); + + return { phase, isConnected, steps, currentStep, messages, streamingText, totalCost, hasError, skippedItems, run, stop }; +} diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/useTasks.ts b/src/workspaces/officerdev/src/apps/FileBrowser/useTasks.ts index acd40285..e000ba54 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/useTasks.ts +++ b/src/workspaces/officerdev/src/apps/FileBrowser/useTasks.ts @@ -10,7 +10,7 @@ export type TaskSummary = { description: string; scope: string; triggers: TriggerConfig[]; - mode: 'script' | 'agentic'; + mode: 'script' | 'agentic' | 'pipeline'; userId: number | null; };