diff --git a/seed/tasks/build-discography/TASK.md b/seed/tasks/build-discography/TASK.md index 02adc82d..dd48e03e 100644 --- a/seed/tasks/build-discography/TASK.md +++ b/seed/tasks/build-discography/TASK.md @@ -16,6 +16,7 @@ steps: artist_name: ${artist_name} - task: fetch-album-info foreach: subdirectory + concurrency: 5 skip_if: album-info.md inputs: artist_name: ${artist_name} diff --git a/src/apps/officer-web/App.tsx b/src/apps/officer-web/App.tsx index b51e99b6..614e027f 100644 --- a/src/apps/officer-web/App.tsx +++ b/src/apps/officer-web/App.tsx @@ -64,6 +64,8 @@ export function App() { } /> } /> } /> + } /> + } /> } /> } /> } /> diff --git a/src/apps/officer-web/Screens/Dashboard/Jobs/JobDetail.tsx b/src/apps/officer-web/Screens/Dashboard/Jobs/JobDetail.tsx new file mode 100644 index 00000000..f97d955a --- /dev/null +++ b/src/apps/officer-web/Screens/Dashboard/Jobs/JobDetail.tsx @@ -0,0 +1,774 @@ +import { useState, useEffect, useRef, useCallback, useMemo, createContext, useContext } from 'react'; +import { useParams, useNavigate, Link } from 'react-router'; +import { + ArrowLeft, CheckCircle2, AlertCircle, Loader2, StopCircle, AlertTriangle, Clock, Square, + ChevronRight, Wrench, +} from 'lucide-react'; +import { useClient } from 'hooks/useClient'; +import { Card } from '@/components/Card'; +import { Button } from '@/components/ui/button'; +import { WorkspaceLayout } from 'officerdev'; +import type { LayoutNode, PanelComponents } from 'officerdev'; + +type Cost = { inputTokens: number; outputTokens: number; totalUSD: number }; + +type StepDef = { task: string; foreach?: string; concurrency?: number }; + +type IterationStatus = { + label: string; + status: 'pending' | 'running' | 'complete' | 'error'; + error?: string; + cost?: Cost; +}; + +type ParallelStep = { + stepIndex: number; + taskName: string; + concurrency: number; + iterations: IterationStatus[]; +}; + +type JobStatus = 'pending' | 'running' | 'completed' | 'failed' | 'stopped' | 'interrupted'; + +type ProgressData = { + steps?: StepDef[]; + currentStepIndex?: number; + parallel?: { taskName: string; concurrency: number; iterations: Array<{ label: string; status: string }> }; +}; + +type JobData = { + id: string; + taskDirName: string; + taskName: string; + status: JobStatus; + totalCost: Cost | null; + progress: ProgressData | null; + error: string | null; + createdAt: string; + startedAt: string | null; + completedAt: string | null; +}; + +// Output entries for the right panel +type OutputEntry = + | { id: string; type: 'text'; text: string } + | { id: string; type: 'tool'; toolCallId: string; toolName: string; toolInput: Record; output?: string; isError?: boolean }; + +type ServerMessage = + | { jobId: string; type: 'pipeline:init'; steps: StepDef[] } + | { jobId: string; type: 'step:start'; stepIndex: number; taskName: string; iteration?: { current: number; total: number; label: string } } + | { jobId: string; type: 'step:complete'; stepIndex: number; cost?: Cost } + | { jobId: string; type: 'step:skip'; stepIndex: number; label: string; reason: string } + | { jobId: string; type: 'step:parallel'; stepIndex: number; taskName: string; iterations: string[]; concurrency: number } + | { jobId: string; type: 'iteration:start'; stepIndex: number; label: string } + | { jobId: string; type: 'iteration:complete'; stepIndex: number; label: string; cost?: Cost } + | { jobId: string; type: 'iteration:error'; stepIndex: number; label: string; error: string } + | { jobId: string; type: 'assistant:delta'; text: string; stepIndex: number; iterationLabel?: string } + | { jobId: string; type: 'assistant:text'; text: string; stepIndex: number; iterationLabel?: string } + | { jobId: string; type: 'tool:start'; toolCallId: string; toolName: string; toolInput: Record; stepIndex: number; iterationLabel?: string } + | { jobId: string; type: 'tool:result'; toolCallId: string; output: string; isError: boolean; stepIndex: number; iterationLabel?: string } + | { jobId: string; type: 'pipeline:complete'; totalCost: Cost } + | { jobId: string; type: 'error'; message: string } + | { jobId: string; type: 'stopped' } + | { type: 'job:state'; jobId: string; status: string; progress: unknown; cost: unknown }; + +const formatDate = (iso: string) => { + const d = new Date(iso); + return d.toLocaleDateString(undefined, { month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit' }); +}; + +const formatElapsed = (seconds: number) => { + const m = Math.floor(seconds / 60); + const s = seconds % 60; + return m > 0 ? `${m}m ${s}s` : `${s}s`; +}; + +const formatCost = (cost: number) => `$${cost.toFixed(4)}`; + +const formatTokens = (n: number) => n >= 1000 ? `${(n / 1000).toFixed(1)}k` : String(n); + +/** Build a unique key for grouping output by step/iteration */ +const outputKey = (stepIndex: number, iterationLabel?: string) => + iterationLabel ? `${stepIndex}:${iterationLabel}` : `${stepIndex}`; + +const StatusIcon = ({ status, className }: { status: string; className?: string }) => { + const base = className ?? 'h-4 w-4 shrink-0'; + switch (status) { + case 'completed': + case 'complete': + return ; + case 'failed': + case 'error': + return ; + case 'running': + return ; + case 'stopped': + return ; + case 'interrupted': + return ; + default: + return ; + } +}; + +type ToolCallEntryProps = { + entry: Extract; +}; + +const ToolCallEntry = ({ entry }: ToolCallEntryProps) => { + const [expanded, setExpanded] = useState(false); + + return ( +
+ + {expanded && ( +
+
+
Input
+ 
+              {JSON.stringify(entry.toolInput, null, 2)}
+
+
+ {entry.output !== undefined && ( +
+
Output
+ 
+                {entry.output}
+
+
+ )} +
+ )} +
+ ); +}; + +// Context to share state between workspace panels +type JobPanelContext = { + displaySteps: StepDef[]; + isLive: boolean; + isRunning: boolean; + completedSteps: Set; + activeStepIndex: number; + progressStepIndex: number; + jobStatus: JobStatus; + selectedKey: string | null; + selectOutput: (key: string) => void; + displayParallel: ParallelStep | null; + skippedItems: Array<{ label: string; reason: string }>; + outputMap: Map; + streamingMap: Map; + outputPanelRef: React.RefObject; +}; + +const JobPanelCtx = createContext(null); +const useJobPanel = () => useContext(JobPanelCtx)!; + +const DEFAULT_LAYOUT: LayoutNode = { + type: 'group', + id: 'root', + direction: 'horizontal', + children: [ + { node: { type: 'panel', id: 'steps', appType: null }, size: 30 }, + { node: { type: 'panel', id: 'output', appType: null }, size: 70 }, + ], +}; + +const StepsPanel = () => { + const ctx = useJobPanel(); + const { + displaySteps, isLive, isRunning, completedSteps, activeStepIndex, + progressStepIndex, jobStatus, selectedKey, selectOutput, displayParallel, + skippedItems, outputMap, + } = ctx; + + return ( +
+
+

Steps

+
+
+ {displaySteps.map((step, i) => { + let status: string = 'pending'; + + if (isLive || isRunning) { + if (completedSteps.has(i)) { + status = 'complete'; + } else if (activeStepIndex === i) { + status = 'running'; + } else if (i < activeStepIndex) { + status = 'complete'; + } + } else { + if (jobStatus === 'completed') { + status = 'complete'; + } else if (progressStepIndex >= 0 && i < progressStepIndex) { + status = 'complete'; + } else if (i === progressStepIndex) { + status = jobStatus === 'failed' || jobStatus === 'interrupted' ? 'error' : 'complete'; + } + } + + const stepKey = outputKey(i); + const isSelected = selectedKey === stepKey && !displayParallel?.stepIndex; + const hasOutput = outputMap.has(stepKey); + + return ( +
+ + + {/* Parallel iterations nested under their step */} + {displayParallel && displayParallel.stepIndex === i && ( +
+ {displayParallel.iterations.map((it) => { + const itKey = outputKey(i, it.label); + const itSelected = selectedKey === itKey; + const itHasOutput = outputMap.has(itKey); + + return ( + + ); + })} + {skippedItems.length > 0 && ( +
+ {skippedItems.length} skipped +
+ )} +
+ )} +
+ ); + })} +
+
+ ); +}; + +const OutputPanel = () => { + const ctx = useJobPanel(); + const { selectedKey, outputMap, streamingMap, outputPanelRef } = ctx; + + const selectedOutputs = selectedKey ? (outputMap.get(selectedKey) ?? []) : []; + const selectedStreaming = selectedKey ? (streamingMap.get(selectedKey) ?? '') : ''; + + return ( +
+
+

Output

+ {selectedKey && ( + {selectedKey.includes(':') ? selectedKey.split(':')[1] : `step ${Number(selectedKey) + 1}`} + )} +
+
+ {!selectedKey && ( +
+ Select a step or sub-task to view output +
+ )} + {selectedKey && selectedOutputs.length === 0 && !selectedStreaming && ( +
+ No output yet +
+ )} + {selectedOutputs.map((entry) => { + if (entry.type === 'text') { + return ( +
+ {entry.text} +
+ ); + } + return ( + + ); + })} + {selectedStreaming && ( +
+ {selectedStreaming} + | +
+ )} +
+
+ ); +}; + +export const JobDetail = () => { + const { id } = useParams<{ id: string }>(); + const navigate = useNavigate(); + const client = useClient(); + + const [job, setJob] = useState(null); + const [isLoading, setIsLoading] = useState(true); + const [isLive, setIsLive] = useState(false); + + // Live state — track by step index + const [steps, setSteps] = useState([]); + const [activeStepIndex, setActiveStepIndex] = useState(-1); + const [completedSteps, setCompletedSteps] = useState>(new Set()); + const [parallelStep, setParallelStep] = useState(null); + const [runningCost, setRunningCost] = useState({ inputTokens: 0, outputTokens: 0, totalUSD: 0 }); + const [totalCost, setTotalCost] = useState(null); + const [hasError, setHasError] = useState(false); + const [skippedItems, setSkippedItems] = useState>([]); + const [elapsed, setElapsed] = useState(0); + const [liveStatus, setLiveStatus] = useState<'running' | 'done'>('running'); + + // Per-step/iteration output tracking + const [outputMap, setOutputMap] = useState>(new Map()); + const [streamingMap, setStreamingMap] = useState>(new Map()); + const [selectedKey, setSelectedKey] = useState(null); + const streamBuffers = useRef>(new Map()); + const outputPanelRef = useRef(null); + const autoFollowRef = useRef(true); + + const [layout, setLayout] = useState(DEFAULT_LAYOUT); + const wsRef = useRef(null); + const startTimeRef = useRef(Date.now()); + const timerRef = useRef | null>(null); + + const stopTimer = useCallback(() => { + if (timerRef.current) { clearInterval(timerRef.current); timerRef.current = null; } + }, []); + + const addCost = useCallback((cost: Cost) => { + setRunningCost((prev) => ({ + inputTokens: prev.inputTokens + cost.inputTokens, + outputTokens: prev.outputTokens + cost.outputTokens, + totalUSD: prev.totalUSD + cost.totalUSD, + })); + }, []); + + const appendOutput = useCallback((key: string, entry: OutputEntry) => { + setOutputMap((prev) => { + const next = new Map(prev); + const arr = next.get(key) ?? []; + next.set(key, [...arr, entry]); + return next; + }); + }, []); + + const updateToolOutput = useCallback((key: string, toolCallId: string, output: string, isError: boolean) => { + setOutputMap((prev) => { + const arr = prev.get(key); + if (!arr) return prev; + const next = new Map(prev); + next.set(key, arr.map((e) => + e.type === 'tool' && e.toolCallId === toolCallId ? { ...e, output, isError } : e, + )); + return next; + }); + }, []); + + // Auto-scroll output panel + useEffect(() => { + if (outputPanelRef.current) { + outputPanelRef.current.scrollTop = outputPanelRef.current.scrollHeight; + } + }, [outputMap, streamingMap, selectedKey]); + + // Fetch job from REST API + useEffect(() => { + if (!id) return; + client + .get(`/pipeline-jobs/${id}`) + .then((data) => { + setJob(data); + setIsLoading(false); + }) + .catch(() => { + setIsLoading(false); + }); + }, [id]); + + // Connect WebSocket for live attachment + useEffect(() => { + if (!id || !job) return; + if (job.status !== 'running' && job.status !== 'pending') return; + + 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; + + startTimeRef.current = job.startedAt ? new Date(job.startedAt).getTime() : Date.now(); + setElapsed(Math.floor((Date.now() - startTimeRef.current) / 1000)); + timerRef.current = setInterval(() => { + setElapsed(Math.floor((Date.now() - startTimeRef.current) / 1000)); + }, 1000); + + ws.addEventListener('open', () => { + ws.send(JSON.stringify({ type: 'attach', jobId: id })); + }); + + ws.addEventListener('message', (ev) => { + try { + const msg = JSON.parse(ev.data) as ServerMessage; + handleEvent(msg); + } catch { + // ignore + } + }); + + ws.addEventListener('close', () => { + setIsLive(false); + }); + + return () => { + ws.close(); + wsRef.current = null; + stopTimer(); + }; + }, [id, job?.status]); + + const handleEvent = useCallback((msg: ServerMessage) => { + switch (msg.type) { + case 'job:state': + if (msg.status === 'completed' || msg.status === 'failed' || msg.status === 'stopped' || msg.status === 'interrupted') { + setLiveStatus('done'); + if (msg.cost) setTotalCost(msg.cost as Cost); + if (msg.status === 'failed' || msg.status === 'interrupted') setHasError(true); + stopTimer(); + if (id) { + client.get(`/pipeline-jobs/${id}`).then(setJob).catch(() => {}); + } + } + if (msg.progress) { + const p = msg.progress as ProgressData; + if (p?.steps) setSteps(p.steps); + } + break; + + case 'pipeline:init': + setIsLive(true); + setSteps(msg.steps); + break; + + case 'step:start': { + setParallelStep(null); + setActiveStepIndex(msg.stepIndex); + const key = msg.iteration + ? outputKey(msg.stepIndex, msg.iteration.label) + : outputKey(msg.stepIndex); + if (autoFollowRef.current) setSelectedKey(key); + break; + } + + case 'step:complete': + setCompletedSteps((prev) => new Set(prev).add(msg.stepIndex)); + setParallelStep(null); + setActiveStepIndex(-1); + if (msg.cost) addCost(msg.cost); + // Flush any remaining stream buffer for this step + flushStreamBuffer(outputKey(msg.stepIndex)); + break; + + case 'step:skip': + setSkippedItems((prev) => [...prev, { label: msg.label, reason: msg.reason }]); + break; + + case 'step:parallel': + setActiveStepIndex(msg.stepIndex); + setParallelStep({ + stepIndex: msg.stepIndex, + taskName: msg.taskName, + concurrency: msg.concurrency, + iterations: msg.iterations.map((label) => ({ label, status: 'pending' })), + }); + // Auto-select first iteration + if (autoFollowRef.current && msg.iterations.length > 0) { + setSelectedKey(outputKey(msg.stepIndex, msg.iterations[0])); + } + break; + + case 'iteration:start': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'running' } : it, + ), + }; + }); + break; + + case 'iteration:complete': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'complete', cost: msg.cost } : it, + ), + }; + }); + if (msg.cost) addCost(msg.cost); + flushStreamBuffer(outputKey(msg.stepIndex, msg.label)); + break; + + case 'iteration:error': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'error', error: msg.error } : it, + ), + }; + }); + flushStreamBuffer(outputKey(msg.stepIndex, msg.label)); + break; + + case 'assistant:delta': { + const key = outputKey(msg.stepIndex, msg.iterationLabel); + const buf = streamBuffers.current; + buf.set(key, (buf.get(key) ?? '') + msg.text); + setStreamingMap((prev) => new Map(prev).set(key, buf.get(key)!)); + break; + } + + case 'assistant:text': { + const key = outputKey(msg.stepIndex, msg.iterationLabel); + const text = msg.text || streamBuffers.current.get(key) || ''; + if (text) { + appendOutput(key, { id: crypto.randomUUID(), type: 'text', text }); + } + streamBuffers.current.delete(key); + setStreamingMap((prev) => { const n = new Map(prev); n.delete(key); return n; }); + break; + } + + case 'tool:start': { + const key = outputKey(msg.stepIndex, msg.iterationLabel); + // Flush any streaming text before the tool call + flushStreamBuffer(key); + appendOutput(key, { + id: crypto.randomUUID(), + type: 'tool', + toolCallId: msg.toolCallId, + toolName: msg.toolName, + toolInput: msg.toolInput, + }); + break; + } + + case 'tool:result': { + const key = outputKey(msg.stepIndex, msg.iterationLabel); + updateToolOutput(key, msg.toolCallId, msg.output, msg.isError); + break; + } + + case 'pipeline:complete': + setTotalCost(msg.totalCost); + setLiveStatus('done'); + setCompletedSteps((prev) => { + const next = new Set(prev); + setSteps((s) => { s.forEach((_, i) => next.add(i)); return s; }); + return next; + }); + setActiveStepIndex(-1); + setParallelStep(null); + stopTimer(); + if (id) { + client.get(`/pipeline-jobs/${id}`).then(setJob).catch(() => {}); + } + break; + + case 'error': + setHasError(true); + setLiveStatus('done'); + stopTimer(); + break; + + case 'stopped': + setLiveStatus('done'); + stopTimer(); + break; + } + }, [id, stopTimer, addCost, appendOutput, updateToolOutput]); + + const flushStreamBuffer = useCallback((key: string) => { + const text = streamBuffers.current.get(key); + if (text) { + appendOutput(key, { id: crypto.randomUUID(), type: 'text', text }); + streamBuffers.current.delete(key); + setStreamingMap((prev) => { const n = new Map(prev); n.delete(key); return n; }); + } + }, [appendOutput]); + + const handleStop = useCallback(() => { + if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN && id) { + wsRef.current.send(JSON.stringify({ type: 'stop', jobId: id })); + } + }, [id]); + + const selectOutput = useCallback((key: string) => { + autoFollowRef.current = false; + setSelectedKey(key); + }, []); + + const panelComponents: PanelComponents = useMemo(() => ({ + steps: StepsPanel, + output: OutputPanel, + }), []); + + const displayStatus = job ? (isLive && liveStatus === 'running' ? 'running' : job.status) : 'pending'; + const isRunning = displayStatus === 'running'; + const displaySteps = steps.length > 0 ? steps : (job?.progress?.steps ?? []); + const jobDone = !isRunning && !isLive; + const progressStepIndex = job?.progress?.currentStepIndex ?? -1; + + const displayParallel: ParallelStep | null = parallelStep ?? (jobDone && job?.progress?.parallel ? { + stepIndex: progressStepIndex, + taskName: job.progress.parallel.taskName, + concurrency: job.progress.parallel.concurrency, + iterations: job.progress.parallel.iterations.map((it) => ({ + label: it.label, + status: it.status as IterationStatus['status'], + })), + } : null); + + const panelCtx = useMemo(() => ({ + displaySteps, isLive, isRunning, completedSteps, activeStepIndex, + progressStepIndex, jobStatus: job?.status ?? 'pending', selectedKey, selectOutput, + displayParallel, skippedItems, outputMap, streamingMap, outputPanelRef, + }), [ + displaySteps, isLive, isRunning, completedSteps, activeStepIndex, + progressStepIndex, job?.status, selectedKey, selectOutput, + displayParallel, skippedItems, outputMap, streamingMap, + ]); + + if (isLoading) { + return ( +
Loading...
+ ); + } + + if (!job) { + return ( +
+ Job not found + Back to jobs +
+ ); + } + + const displayCost = totalCost ?? (runningCost.totalUSD > 0 ? runningCost : null) ?? job.totalCost; + const displayElapsed = isRunning + ? elapsed + : job.startedAt && job.completedAt + ? Math.floor((new Date(job.completedAt).getTime() - new Date(job.startedAt).getTime()) / 1000) + : elapsed > 0 ? elapsed : null; + + return ( +
+ {/* Header + stats bar */} + +
+ +
+
+ +

{job.taskName}

+
+
+ {formatDate(job.createdAt)} + {job.completedAt && ` — ${formatDate(job.completedAt)}`} +
+
+ {isRunning && ( + + )} +
+ {(displayCost || displayElapsed !== null) && ( +
+ {displayElapsed !== null && ( +
+ + {formatElapsed(displayElapsed)} +
+ )} + {displayCost && ( + <> +
+ Tokens: + + {formatTokens(displayCost.inputTokens)} in / {formatTokens(displayCost.outputTokens)} out + +
+
+ Cost: + {formatCost(displayCost.totalUSD)} +
+ + )} +
+ )} + + + {/* Error message */} + {(hasError || job.error) && ( + +
+ + {job.error ?? 'An error occurred during execution'} +
+ + )} + + {/* Two-column: steps list + output viewer */} +
+ + + +
+
+ ); +}; + diff --git a/src/apps/officer-web/Screens/Dashboard/Jobs/JobsScreen.tsx b/src/apps/officer-web/Screens/Dashboard/Jobs/JobsScreen.tsx new file mode 100644 index 00000000..1365baee --- /dev/null +++ b/src/apps/officer-web/Screens/Dashboard/Jobs/JobsScreen.tsx @@ -0,0 +1,138 @@ +import { useState, useEffect } from 'react'; +import { useNavigate } from 'react-router'; +import { Search, CheckCircle2, AlertCircle, Clock, Loader2, StopCircle, AlertTriangle } from 'lucide-react'; +import { useClient } from 'hooks/useClient'; +import { Card } from '@/components/Card'; + +type JobSummary = { + id: string; + taskDirName: string; + taskName: string; + status: 'pending' | 'running' | 'completed' | 'failed' | 'stopped' | 'interrupted'; + totalCost: { inputTokens: number; outputTokens: number; totalUSD: number } | null; + createdAt: string; + startedAt: string | null; + completedAt: string | null; + error: string | null; +}; + +const formatDate = (iso: string) => { + const d = new Date(iso); + return d.toLocaleDateString(undefined, { month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit' }); +}; + +const formatCost = (cost: number) => `$${cost.toFixed(4)}`; + +const StatusIcon = ({ status }: { status: JobSummary['status'] }) => { + switch (status) { + case 'completed': + return ; + case 'failed': + return ; + case 'running': + return ; + case 'stopped': + return ; + case 'interrupted': + return ; + case 'pending': + return ; + } +}; + +const StatusBadge = ({ status }: { status: JobSummary['status'] }) => { + const colors: Record = { + completed: 'bg-green-100 dark:bg-green-900/40 text-green-700 dark:text-green-300', + failed: 'bg-red-100 dark:bg-red-900/40 text-red-700 dark:text-red-300', + running: 'bg-blue-100 dark:bg-blue-900/40 text-blue-700 dark:text-blue-300', + stopped: 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300', + interrupted: 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300', + pending: 'bg-gray-100 dark:bg-gray-800/40 text-gray-600 dark:text-gray-400', + }; + + return ( + + {status} + + ); +}; + +export const JobsScreen = () => { + const client = useClient(); + const navigate = useNavigate(); + const [jobs, setJobs] = useState([]); + const [search, setSearch] = useState(''); + const [isLoading, setIsLoading] = useState(true); + + useEffect(() => { + client + .get('/pipeline-jobs') + .then((data) => { + setJobs(data); + setIsLoading(false); + }) + .catch(() => setIsLoading(false)); + }, []); + + const filtered = search + ? jobs.filter((j) => { + const q = search.toLowerCase(); + return j.taskName.toLowerCase().includes(q) || j.taskDirName.toLowerCase().includes(q) || j.status.includes(q); + }) + : jobs; + + return ( +
+ +
+

Pipeline Jobs

+
+ + setSearch(ev.target.value)} + className="w-full pl-8 pr-3 py-1.5 text-sm rounded-md border border-duck-dark/15 bg-background/60 text-duck-dark placeholder:text-duck-dark/30 focus:outline-none focus:ring-1 focus:ring-duck-teal/40" + /> +
+
+ +
+ {isLoading && ( +
Loading...
+ )} + {!isLoading && filtered.length === 0 && ( +
+ {search ? 'No jobs match your search' : 'No pipeline jobs yet'} +
+ )} + {filtered.map((job) => ( + + ))} +
+ +
+ ); +}; diff --git a/src/apps/officer-web/Screens/Dashboard/Jobs/index.tsx b/src/apps/officer-web/Screens/Dashboard/Jobs/index.tsx new file mode 100644 index 00000000..13076ead --- /dev/null +++ b/src/apps/officer-web/Screens/Dashboard/Jobs/index.tsx @@ -0,0 +1,2 @@ +export { JobsScreen } from './JobsScreen'; +export { JobDetail } from './JobDetail'; diff --git a/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx b/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx index 0ac5da3b..12a6bc61 100644 --- a/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx +++ b/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx @@ -110,7 +110,7 @@ export const Dock = ({ items, className }: DockProps) => { }; -import { Home, MessageCircle, FileText, FolderOpen, Code, LayoutGrid, Bot, ScrollText, FolderKanban, Monitor, Mail, Globe, MonitorSmartphone } from 'lucide-react'; +import { Home, MessageCircle, FileText, FolderOpen, Code, LayoutGrid, Bot, ScrollText, FolderKanban, Monitor, Mail, Globe, MonitorSmartphone, Workflow } from 'lucide-react'; export const ALL_DOCK_ITEMS: DockItem[] = [ { label: 'Home', to: '/', icon: Home, color: '#f59e0b' }, @@ -120,6 +120,7 @@ export const ALL_DOCK_ITEMS: DockItem[] = [ { label: 'Editor', to: '/code-editor', icon: Code, color: '#a78bfa' }, { label: 'Plans', to: '/plans', icon: FileText, color: '#f472b6' }, { label: 'Automation', to: '/automation', icon: Bot, color: '#2dd4bf' }, + { label: 'Jobs', to: '/jobs', icon: Workflow, color: '#14b8a6' }, { label: 'Logs', to: '/task-logs', icon: ScrollText, color: '#94a3b8' }, { label: 'Terminal', to: '/terminal', icon: Monitor, color: '#f97316' }, { label: 'Projects', to: '/projects', icon: FolderKanban, color: '#10b981' }, diff --git a/src/apps/officer-web/Screens/Dashboard/index.tsx b/src/apps/officer-web/Screens/Dashboard/index.tsx index 1145e61d..83cece9e 100644 --- a/src/apps/officer-web/Screens/Dashboard/index.tsx +++ b/src/apps/officer-web/Screens/Dashboard/index.tsx @@ -20,3 +20,4 @@ export * from './Terminal'; export * from './Email'; export * from './Browser'; export * from './Desktop'; +export * from './Jobs'; diff --git a/src/databases/officer_db/migrations/0003_perpetual_james_howlett.sql b/src/databases/officer_db/migrations/0003_perpetual_james_howlett.sql new file mode 100644 index 00000000..cfba48a2 --- /dev/null +++ b/src/databases/officer_db/migrations/0003_perpetual_james_howlett.sql @@ -0,0 +1,20 @@ +CREATE TABLE "pipeline_jobs" ( + "id" text PRIMARY KEY NOT NULL, + "user_id" integer NOT NULL, + "task_dir_name" text NOT NULL, + "task_name" text NOT NULL, + "status" text DEFAULT 'pending' NOT NULL, + "inputs" jsonb DEFAULT '{}'::jsonb NOT NULL, + "cwd" text, + "config" jsonb NOT NULL, + "progress" jsonb, + "total_cost" jsonb, + "error" text, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "started_at" timestamp with time zone, + "completed_at" timestamp with time zone +); +--> statement-breakpoint +ALTER TABLE "pipeline_jobs" ADD CONSTRAINT "pipeline_jobs_user_id_users_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +CREATE INDEX "idx_pipeline_jobs_user_created" ON "pipeline_jobs" USING btree ("user_id","created_at");--> statement-breakpoint +CREATE INDEX "idx_pipeline_jobs_status" ON "pipeline_jobs" USING btree ("status"); \ No newline at end of file diff --git a/src/databases/officer_db/migrations/meta/0003_snapshot.json b/src/databases/officer_db/migrations/meta/0003_snapshot.json new file mode 100644 index 00000000..8637bd1d --- /dev/null +++ b/src/databases/officer_db/migrations/meta/0003_snapshot.json @@ -0,0 +1,2500 @@ +{ + "id": "6efe8a17-2fc4-4e8d-9c97-459149255bb8", + "prevId": "9b81457f-9d9e-495c-8a22-0de8a220e29e", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.passkey_challenges": { + "name": "passkey_challenges", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "origin": { + "name": "origin", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "challenge": { + "name": "challenge", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "passkey_challenges_user_id_users_id_fk": { + "name": "passkey_challenges_user_id_users_id_fk", + "tableFrom": "passkey_challenges", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.passkeys": { + "name": "passkeys", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "origin": { + "name": "origin", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "credential_id": { + "name": "credential_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "public_key": { + "name": "public_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "counter": { + "name": "counter", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "passkeys_user_id_users_id_fk": { + "name": "passkeys_user_id_users_id_fk", + "tableFrom": "passkeys", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.token_blacklist": { + "name": "token_blacklist", + "schema": "", + "columns": { + "jti": { + "name": "jti", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "idx_token_blacklist_expires": { + "name": "idx_token_blacklist_expires", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.users": { + "name": "users", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "password": { + "name": "password", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'Member'" + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'Unverified'" + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "username": { + "name": "username", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "avatar": { + "name": "avatar", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "password_changed_at": { + "name": "password_changed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "users_email_unique": { + "name": "users_email_unique", + "nullsNotDistinct": false, + "columns": [ + "email" + ] + }, + "users_username_unique": { + "name": "users_username_unique", + "nullsNotDistinct": false, + "columns": [ + "username" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.dock_configs": { + "name": "dock_configs", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": true, + "notNull": true + }, + "paths": { + "name": "paths", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "dock_configs_user_id_users_id_fk": { + "name": "dock_configs_user_id_users_id_fk", + "tableFrom": "dock_configs", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user_integrations": { + "name": "user_integrations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "server_integration_id": { + "name": "server_integration_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "user_integrations_user_id_users_id_fk": { + "name": "user_integrations_user_id_users_id_fk", + "tableFrom": "user_integrations", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "user_integrations_server_integration_id_server_integrations_id_fk": { + "name": "user_integrations_server_integration_id_server_integrations_id_fk", + "tableFrom": "user_integrations", + "tableTo": "server_integrations", + "columnsFrom": [ + "server_integration_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_user_integrations_user_provider": { + "name": "uq_user_integrations_user_provider", + "nullsNotDistinct": false, + "columns": [ + "user_id", + "provider" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user_settings": { + "name": "user_settings", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": true, + "notNull": true + }, + "settings": { + "name": "settings", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "user_settings_user_id_users_id_fk": { + "name": "user_settings_user_id_users_id_fk", + "tableFrom": "user_settings", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user_state": { + "name": "user_state", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": true, + "notNull": true + }, + "state": { + "name": "state", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "user_state_user_id_users_id_fk": { + "name": "user_state_user_id_users_id_fk", + "tableFrom": "user_state", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.saved_sessions": { + "name": "saved_sessions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "context": { + "name": "context", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "context_id": { + "name": "context_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "summary": { + "name": "summary", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "raw_messages": { + "name": "raw_messages", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "cost": { + "name": "cost", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_saved_sessions_user_created": { + "name": "idx_saved_sessions_user_created", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "saved_sessions_user_id_users_id_fk": { + "name": "saved_sessions_user_id_users_id_fk", + "tableFrom": "saved_sessions", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.dashboard_defaults": { + "name": "dashboard_defaults", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "terminals": { + "name": "terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "host_terminals": { + "name": "host_terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "dashboard_defaults_user_id_users_id_fk": { + "name": "dashboard_defaults_user_id_users_id_fk", + "tableFrom": "dashboard_defaults", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "dashboard_defaults_user_id_unique": { + "name": "dashboard_defaults_user_id_unique", + "nullsNotDistinct": false, + "columns": [ + "user_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.dashboards": { + "name": "dashboards", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "layout": { + "name": "layout", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "terminals": { + "name": "terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "host_terminals": { + "name": "host_terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "sort_order": { + "name": "sort_order", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "dashboards_user_id_users_id_fk": { + "name": "dashboards_user_id_users_id_fk", + "tableFrom": "dashboards", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_dashboards_user_id": { + "name": "uq_dashboards_user_id", + "nullsNotDistinct": false, + "columns": [ + "user_id", + "id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.projects": { + "name": "projects", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "slug": { + "name": "slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "meta": { + "name": "meta", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "layout": { + "name": "layout", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "terminals": { + "name": "terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "host_terminals": { + "name": "host_terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "projects_user_id_users_id_fk": { + "name": "projects_user_id_users_id_fk", + "tableFrom": "projects", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_projects_user_slug": { + "name": "uq_projects_user_slug", + "nullsNotDistinct": false, + "columns": [ + "user_id", + "slug" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.screens": { + "name": "screens", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "layout": { + "name": "layout", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "terminals": { + "name": "terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "host_terminals": { + "name": "host_terminals", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "screens_user_id_users_id_fk": { + "name": "screens_user_id_users_id_fk", + "tableFrom": "screens", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_screens_user_name": { + "name": "uq_screens_user_name", + "nullsNotDistinct": false, + "columns": [ + "user_id", + "name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.extensions": { + "name": "extensions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "dir_name": { + "name": "dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "implementation": { + "name": "implementation", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_extensions_scope": { + "name": "idx_extensions_scope", + "columns": [ + { + "expression": "scope", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_extensions_user": { + "name": "idx_extensions_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "extensions_user_id_users_id_fk": { + "name": "extensions_user_id_users_id_fk", + "tableFrom": "extensions", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_extensions_scope_user_dir": { + "name": "uq_extensions_scope_user_dir", + "nullsNotDistinct": false, + "columns": [ + "scope", + "user_id", + "dir_name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.item_chats": { + "name": "item_chats", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "item_type": { + "name": "item_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "item_id": { + "name": "item_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "messages": { + "name": "messages", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_item_chats_type_item": { + "name": "idx_item_chats_type_item", + "columns": [ + { + "expression": "item_type", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "item_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_item_chats_type_item": { + "name": "uq_item_chats_type_item", + "nullsNotDistinct": false, + "columns": [ + "item_type", + "item_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.processes": { + "name": "processes", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "dir_name": { + "name": "dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_processes_scope": { + "name": "idx_processes_scope", + "columns": [ + { + "expression": "scope", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_processes_user": { + "name": "idx_processes_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "processes_user_id_users_id_fk": { + "name": "processes_user_id_users_id_fk", + "tableFrom": "processes", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_processes_scope_user_dir": { + "name": "uq_processes_scope_user_dir", + "nullsNotDistinct": false, + "columns": [ + "scope", + "user_id", + "dir_name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.skills": { + "name": "skills", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "dir_name": { + "name": "dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_skills_scope": { + "name": "idx_skills_scope", + "columns": [ + { + "expression": "scope", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_skills_user": { + "name": "idx_skills_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "skills_user_id_users_id_fk": { + "name": "skills_user_id_users_id_fk", + "tableFrom": "skills", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_skills_scope_user_dir": { + "name": "uq_skills_scope_user_dir", + "nullsNotDistinct": false, + "columns": [ + "scope", + "user_id", + "dir_name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tasks": { + "name": "tasks", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "dir_name": { + "name": "dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "mode": { + "name": "mode", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'agentic'" + }, + "language": { + "name": "language", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "implementation": { + "name": "implementation", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "args": { + "name": "args", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "tags": { + "name": "tags", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "tools": { + "name": "tools", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "skills": { + "name": "skills", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "inputs": { + "name": "inputs", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "outputs": { + "name": "outputs", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "dependencies": { + "name": "dependencies", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "trigger": { + "name": "trigger", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_tasks_scope": { + "name": "idx_tasks_scope", + "columns": [ + { + "expression": "scope", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_tasks_user": { + "name": "idx_tasks_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "tasks_user_id_users_id_fk": { + "name": "tasks_user_id_users_id_fk", + "tableFrom": "tasks", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_tasks_scope_user_dir": { + "name": "uq_tasks_scope_user_dir", + "nullsNotDistinct": false, + "columns": [ + "scope", + "user_id", + "dir_name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tools": { + "name": "tools", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "dir_name": { + "name": "dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "label": { + "name": "label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "language": { + "name": "language", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "inputs": { + "name": "inputs", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "implementation": { + "name": "implementation", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_tools_scope": { + "name": "idx_tools_scope", + "columns": [ + { + "expression": "scope", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_tools_user": { + "name": "idx_tools_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "tools_user_id_users_id_fk": { + "name": "tools_user_id_users_id_fk", + "tableFrom": "tools", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_tools_scope_user_dir": { + "name": "uq_tools_scope_user_dir", + "nullsNotDistinct": false, + "columns": [ + "scope", + "user_id", + "dir_name" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.queue_jobs": { + "name": "queue_jobs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "lane": { + "name": "lane", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "type": { + "name": "type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'queued'" + }, + "current_step": { + "name": "current_step", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "steps": { + "name": "steps", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "meta": { + "name": "meta", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "completed_at": { + "name": "completed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_queue_jobs_status_lane": { + "name": "idx_queue_jobs_status_lane", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "lane", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_queue_jobs_user": { + "name": "idx_queue_jobs_user", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "queue_jobs_user_id_users_id_fk": { + "name": "queue_jobs_user_id_users_id_fk", + "tableFrom": "queue_jobs", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.task_logs": { + "name": "task_logs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "task_name": { + "name": "task_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "task_dir_name": { + "name": "task_dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "entry_name": { + "name": "entry_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "entry_type": { + "name": "entry_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "is_error": { + "name": "is_error", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "messages": { + "name": "messages", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "completed_at": { + "name": "completed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_task_logs_user_started": { + "name": "idx_task_logs_user_started", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "started_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "task_logs_user_id_users_id_fk": { + "name": "task_logs_user_id_users_id_fk", + "tableFrom": "task_logs", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.terminal_containers": { + "name": "terminal_containers", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": true, + "notNull": true + }, + "docker_id": { + "name": "docker_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "port": { + "name": "port", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "terminal_containers_user_id_users_id_fk": { + "name": "terminal_containers_user_id_users_id_fk", + "tableFrom": "terminal_containers", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.server_config": { + "name": "server_config", + "schema": "", + "columns": { + "key": { + "name": "key", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "value": { + "name": "value", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.server_integrations": { + "name": "server_integrations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "enabled": { + "name": "enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "server_integrations_provider_unique": { + "name": "server_integrations_provider_unique", + "nullsNotDistinct": false, + "columns": [ + "provider" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.email_accounts": { + "name": "email_accounts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "display_name": { + "name": "display_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "imap_host": { + "name": "imap_host", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "imap_port": { + "name": "imap_port", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "imap_secure": { + "name": "imap_secure", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "auth_type": { + "name": "auth_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "credentials": { + "name": "credentials", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "enabled": { + "name": "enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'connected'" + }, + "sync_meta": { + "name": "sync_meta", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "email_accounts_user_id_users_id_fk": { + "name": "email_accounts_user_id_users_id_fk", + "tableFrom": "email_accounts", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_email_accounts_user_email": { + "name": "uq_email_accounts_user_email", + "nullsNotDistinct": false, + "columns": [ + "user_id", + "email" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.pipeline_jobs": { + "name": "pipeline_jobs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "task_dir_name": { + "name": "task_dir_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "task_name": { + "name": "task_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "inputs": { + "name": "inputs", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "progress": { + "name": "progress", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "total_cost": { + "name": "total_cost", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "completed_at": { + "name": "completed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_pipeline_jobs_user_created": { + "name": "idx_pipeline_jobs_user_created", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_pipeline_jobs_status": { + "name": "idx_pipeline_jobs_status", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "pipeline_jobs_user_id_users_id_fk": { + "name": "pipeline_jobs_user_id_users_id_fk", + "tableFrom": "pipeline_jobs", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/src/databases/officer_db/migrations/meta/_journal.json b/src/databases/officer_db/migrations/meta/_journal.json index 45830a64..43016983 100644 --- a/src/databases/officer_db/migrations/meta/_journal.json +++ b/src/databases/officer_db/migrations/meta/_journal.json @@ -22,6 +22,13 @@ "when": 1772985087180, "tag": "0002_cute_doorman", "breakpoints": true + }, + { + "idx": 3, + "version": "7", + "when": 1773040399200, + "tag": "0003_perpetual_james_howlett", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 528ab813..2aa1b14d 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -81,5 +81,13 @@ export { deleteTask, } from './queries/tasks'; +export { + createPipelineJob, + getPipelineJob, + updatePipelineJob, + getPipelineJobsForUser, + markInterruptedJobs, +} from './queries/pipeline-jobs'; + export { db } from './db'; export * as schema from './schema'; diff --git a/src/databases/officer_db/src/queries/pipeline-jobs.ts b/src/databases/officer_db/src/queries/pipeline-jobs.ts new file mode 100644 index 00000000..4aa3638c --- /dev/null +++ b/src/databases/officer_db/src/queries/pipeline-jobs.ts @@ -0,0 +1,36 @@ +import { eq, and, inArray, desc } from 'drizzle-orm'; +import { db } from '../db'; +import { pipelineJobs } from '../schema/pipeline-jobs'; +import type { PipelineJobInsert } from '../types'; + +export async function createPipelineJob(data: PipelineJobInsert) { + const rows = await db.insert(pipelineJobs).values(data).returning(); + return rows[0]!; +} + +export async function getPipelineJob(id: string) { + const rows = await db.select().from(pipelineJobs).where(eq(pipelineJobs.id, id)).limit(1); + return rows[0] ?? null; +} + +export async function updatePipelineJob(id: string, data: Partial) { + await db.update(pipelineJobs).set(data).where(eq(pipelineJobs.id, id)); +} + +export async function getPipelineJobsForUser(userId: number, limit = 50) { + return db + .select() + .from(pipelineJobs) + .where(eq(pipelineJobs.userId, userId)) + .orderBy(desc(pipelineJobs.createdAt)) + .limit(limit); +} + +export async function markInterruptedJobs() { + const result = await db + .update(pipelineJobs) + .set({ status: 'interrupted', completedAt: new Date() }) + .where(inArray(pipelineJobs.status, ['pending', 'running'])) + .returning({ id: pipelineJobs.id }); + return result.length; +} diff --git a/src/databases/officer_db/src/schema/index.ts b/src/databases/officer_db/src/schema/index.ts index 35fd13c3..ce9fb65a 100644 --- a/src/databases/officer_db/src/schema/index.ts +++ b/src/databases/officer_db/src/schema/index.ts @@ -6,3 +6,4 @@ export * from './agent-items'; export * from './operations'; export * from './server'; export * from './email'; +export * from './pipeline-jobs'; diff --git a/src/databases/officer_db/src/schema/pipeline-jobs.ts b/src/databases/officer_db/src/schema/pipeline-jobs.ts new file mode 100644 index 00000000..ba5993fc --- /dev/null +++ b/src/databases/officer_db/src/schema/pipeline-jobs.ts @@ -0,0 +1,22 @@ +import { pgTable, text, integer, timestamp, jsonb, index } from 'drizzle-orm/pg-core'; +import { users } from './auth'; + +export const pipelineJobs = pgTable('pipeline_jobs', { + id: text('id').primaryKey(), + userId: integer('user_id').notNull().references(() => users.id, { onDelete: 'cascade' }), + taskDirName: text('task_dir_name').notNull(), + taskName: text('task_name').notNull(), + status: text('status', { enum: ['pending', 'running', 'completed', 'failed', 'stopped', 'interrupted'] }).notNull().default('pending'), + inputs: jsonb('inputs').notNull().default({}), + cwd: text('cwd'), + config: jsonb('config').notNull(), + progress: jsonb('progress'), + totalCost: jsonb('total_cost'), + error: text('error'), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + startedAt: timestamp('started_at', { withTimezone: true }), + completedAt: timestamp('completed_at', { withTimezone: true }), +}, (table) => [ + index('idx_pipeline_jobs_user_created').on(table.userId, table.createdAt), + index('idx_pipeline_jobs_status').on(table.status), +]); diff --git a/src/databases/officer_db/src/types.ts b/src/databases/officer_db/src/types.ts index 1a190695..a9602d57 100644 --- a/src/databases/officer_db/src/types.ts +++ b/src/databases/officer_db/src/types.ts @@ -103,3 +103,8 @@ export type ServerConfigInsert = typeof Schema.serverConfig.$inferInsert; export type ServerIntegrationSelect = typeof Schema.serverIntegrations.$inferSelect; export type ServerIntegrationInsert = typeof Schema.serverIntegrations.$inferInsert; + +// ── Pipeline Jobs ── + +export type PipelineJobSelect = typeof Schema.pipelineJobs.$inferSelect; +export type PipelineJobInsert = typeof Schema.pipelineJobs.$inferInsert; diff --git a/src/server.tsx b/src/server.tsx index d962c554..de57cb92 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -323,6 +323,10 @@ try { import { initQueue, enqueueJob as queueEnqueue, cancelJob as queueCancel, listAllJobs as queueList, readJob as queueGet } from './servers/queue/init'; initQueue().catch((err) => console.error('[queue] failed to initialize:', err)); +// Mark any orphaned pipeline jobs from previous server run +import { cleanupOnStartup } from './servers/api/tasks/pipeline-job-manager'; +cleanupOnStartup().catch((err) => console.error('[pipeline-jobs] startup cleanup failed:', err)); + // Ensure PulseAudio is running with virtual sink for cliamp audio streaming (async () => { const pulseaudio = Bun.which('pulseaudio'); diff --git a/src/servers/api/tasks/pipeline-executor.ts b/src/servers/api/tasks/pipeline-executor.ts index adc76dac..afbd12c2 100644 --- a/src/servers/api/tasks/pipeline-executor.ts +++ b/src/servers/api/tasks/pipeline-executor.ts @@ -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 { } } -type WSData = { - userId: number; - email: string; - username: string; - role: string; - sandboxed: boolean; -}; +// ── Types ── type PipelineStep = { task: string; inputs?: Record; foreach?: 'subdirectory'; skip_if?: string; + concurrency?: 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 }> } +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 } - | { 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; 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, { abort: () => void }>(); - -function send(ws: ServerWebSocket, 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; + 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 { - const { email, username, userId, role } = ws.data; +async function runAgenticStep({ userId, email, username, role, taskDirName, prompt, cwd, abortSignal, emit, stepIndex, iterationLabel }: RunStepParams): Promise { 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(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 { 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, targetDir?: string): string { const inputLines = Object.entries(inputs) .filter(([, v]) => v.trim()) @@ -174,93 +180,214 @@ function buildStepPrompt(taskBody: string, inputs: Record, targe return `${taskBody.trim()}\n\n## Inputs\n\n${inputLines}${contextSection}`; } -async function handleRun(ws: ServerWebSocket, 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; + 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>(); + + 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 = {}; + 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; + cwd?: string; + abortSignal: AbortSignal; + emit: EmitEvent; +}; + +export async function executePipeline({ userId, email, username, role, taskDirName, inputs, cwd, abortSignal, emit }: ExecutePipelineParams): Promise { + 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 = {}; + 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 = {}; - 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 = {}; if (step.inputs) { for (const [key, template] of Object.entries(step.inputs)) { @@ -268,87 +395,145 @@ async function handleRun(ws: ServerWebSocket, 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; cwd?: string } + | { type: 'stop'; jobId: string } + | { type: 'attach'; jobId: string } + | { type: 'list' }; + +function send(ws: ServerWebSocket, msg: Record) { + if (ws.readyState === 1) ws.send(JSON.stringify(msg)); } export function open(_ws: ServerWebSocket) {} -export function message(ws: ServerWebSocket, raw: string | Buffer) { +export async 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' }); + 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, raw: string | Buffer) { } export function close(ws: ServerWebSocket) { - 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 = { diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts new file mode 100644 index 00000000..f89498ca --- /dev/null +++ b/src/servers/api/tasks/pipeline-job-manager.ts @@ -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>; + progressDirty: boolean; + lastProgress: unknown; + lastCost: unknown; +}; + +const EVENT_BUFFER_SIZE = 500; +const PROGRESS_FLUSH_MS = 3000; + +const liveJobs = new Map(); + +function sendToViewer(ws: ServerWebSocket, 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; + cwd?: string; + config: unknown; +}; + +export async function startJob(params: StartJobParams): Promise { + 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, + totalCost: job.lastCost as Record, + }); + } 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, + totalCost: job.lastCost as Record, + 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, + totalCost: job.lastCost as Record, + 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) { + 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) { + const job = liveJobs.get(jobId); + if (job) job.viewers.delete(ws); +} + +export function detachAllForViewer(ws: ServerWebSocket) { + 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) ?? {}; + + 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).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).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; + } +} diff --git a/src/servers/api/tasks/pipeline-jobs-routes.ts b/src/servers/api/tasks/pipeline-jobs-routes.ts new file mode 100644 index 00000000..c856f31e --- /dev/null +++ b/src/servers/api/tasks/pipeline-jobs-routes.ts @@ -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); +}); diff --git a/src/servers/hono.ts b/src/servers/hono.ts index 41aabe37..dfc065f6 100644 --- a/src/servers/hono.ts +++ b/src/servers/hono.ts @@ -30,6 +30,7 @@ import { desktopRouter } from './api/desktop/rest'; import { appsRouter, appServeRouter } from './api/apps'; import { bugReportRouter } from './api/bug-report/bug-report'; import { savedSessionsRouter } from './api/saved-sessions/saved-sessions'; +import { pipelineJobsRouter } from './api/tasks/pipeline-jobs-routes'; import { broadcastPanelRefresh } from './api/terminal/websocket'; import { CustomError } from './custom-errors'; import { userMiddleware, bodyParser, isOriginAllowed, superAdminMiddleware } from './_middlewares'; @@ -100,6 +101,7 @@ protectedRouter.route('/browser', browserRouter); protectedRouter.route('/apps', appsRouter); protectedRouter.route('/bug-report', bugReportRouter); protectedRouter.route('/saved-sessions', savedSessionsRouter); +protectedRouter.route('/pipeline-jobs', pipelineJobsRouter); desktopRouter.use(superAdminMiddleware); protectedRouter.route('/desktop', desktopRouter); protectedRouter.route('/', piRestRouter); 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 00535959..cb202a2c 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx @@ -1,5 +1,5 @@ import { useState, useEffect, useRef } from 'react'; -import { X, Play, Square, CircleCheck, CircleX, Copy, Check } from 'lucide-react'; +import { X, Play, Square, CircleCheck, CircleX, Copy, Check, Loader2, AlertCircle, ExternalLink } from 'lucide-react'; import { Dialog, DialogOverlay, DialogPortal } from '@/components/ui/dialog'; import * as DialogPrimitive from '@radix-ui/react-dialog'; import { cardStyle } from '@/components/Card'; @@ -512,7 +512,7 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { // Auto-scroll useEffect(() => { bottomRef.current?.scrollIntoView({ behavior: 'smooth' }); - }, [pipeline.messages, pipeline.streamingText, pipeline.currentStep]); + }, [pipeline.messages, pipeline.streamingText, pipeline.currentStep, pipeline.parallelStep]); // Ding on completion const prevPhaseRef = useRef(pipeline.phase); @@ -554,9 +554,25 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { ); } + const formatElapsed = (s: number) => { + const m = Math.floor(s / 60); + const sec = s % 60; + return m > 0 ? `${m}m ${sec}s` : `${sec}s`; + }; + + const rc = pipeline.runningCost; + const totalTokens = rc.inputTokens + rc.outputTokens; + const ps = pipeline.parallelStep; + + // Parallel progress counts + const pDone = ps?.iterations.filter((it) => it.status === 'complete').length ?? 0; + const pRunning = ps?.iterations.filter((it) => it.status === 'running').length ?? 0; + const pError = ps?.iterations.filter((it) => it.status === 'error').length ?? 0; + const pTotal = ps?.iterations.length ?? 0; + return (
- {/* Step progress header */} + {/* Step progress header — sequential */} {pipeline.currentStep && (
@@ -577,8 +593,65 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => {
)} - {/* Messages */} + {/* Step progress header — parallel */} + {ps && ( +
+
+ {ps.taskName} + + {pDone}/{pTotal} done + {pRunning > 0 && · {pRunning} running} + {pError > 0 && · {pError} failed} + + {pDone + pError < pTotal && ( + + )} + {pDone + pError === pTotal && pTotal > 0 && ( + done + )} +
+
+ )} + + {/* Running stats */} + {pipeline.phase === 'running' && ( +
+ {formatElapsed(pipeline.elapsed)} + {totalTokens > 0 && {totalTokens.toLocaleString()} tok} + {rc.totalUSD > 0 && ${rc.totalUSD.toFixed(3)}} +
+ )} + + {/* Content area */}
+ {/* Parallel iteration grid */} + {ps && ( +
+ {ps.iterations.map((it) => ( +
+ {it.status === 'pending' && } + {it.status === 'running' && } + {it.status === 'complete' && } + {it.status === 'error' && } + + {it.label} + + {it.cost && ( + + ${it.cost.totalUSD.toFixed(3)} + + )} + {it.error && ( + + {it.error} + + )} +
+ ))} +
+ )} + + {/* Sequential messages */} {pipeline.messages.map((msg, i) => (
{}} /> @@ -614,8 +687,8 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { )} {pipeline.totalCost && ( - - ${pipeline.totalCost.totalUSD.toFixed(3)} · {pipeline.totalCost.inputTokens + pipeline.totalCost.outputTokens} tokens + + {formatElapsed(pipeline.elapsed)} · {(pipeline.totalCost.inputTokens + pipeline.totalCost.outputTokens).toLocaleString()} tokens · ${pipeline.totalCost.totalUSD.toFixed(3)} )} {pipeline.skippedItems.length > 0 && ( @@ -623,6 +696,15 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { {pipeline.skippedItems.length} skipped ({pipeline.skippedItems.map((s) => s.label).join(', ')}) )} + {pipeline.jobId && ( + + + View in Jobs + + )}
); @@ -656,8 +738,9 @@ export const TaskRunnerModal = ({ open, onOpenChange, task, entryName, entryFull 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; + // Build absolute path the agent sees (sandboxed: /data/home/..., non-sandboxed: ~/...) + const homePrefix = sandboxed ? '/data/home' : '~'; + const entryRelPath = entryName && cwd.path ? `${homePrefix}/${cwd.path}/${entryName}` : entryName ? `${homePrefix}/${entryName}` : undefined; const autofillContext: Record = {}; if (entryName) autofillContext.entry_name = entryName; if (entryRelPath) autofillContext.entry_path = entryRelPath; diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts index 1aa955e7..b228de35 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts @@ -6,6 +6,7 @@ type Phase = 'ready' | 'running' | 'done'; type StepDef = { task: string; foreach?: string; + concurrency?: number; }; type StepStatus = { @@ -15,31 +16,60 @@ type StepStatus = { cost?: { inputTokens: number; outputTokens: number; totalUSD: number }; }; +export type IterationStatus = { + label: string; + status: 'pending' | 'running' | 'complete' | 'error'; + error?: string; + cost?: { inputTokens: number; outputTokens: number; totalUSD: number }; +}; + +type ParallelStep = { + stepIndex: number; + taskName: string; + concurrency: number; + iterations: IterationStatus[]; +}; + 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' }; + | { jobId: string; type: 'pipeline:init'; steps: StepDef[] } + | { jobId: string; type: 'step:start'; stepIndex: number; taskName: string; iteration?: { current: number; total: number; label: string } } + | { jobId: string; type: 'step:complete'; stepIndex: number; cost?: { inputTokens: number; outputTokens: number; totalUSD: number } } + | { jobId: string; type: 'step:skip'; stepIndex: number; label: string; reason: string } + | { jobId: string; type: 'step:parallel'; stepIndex: number; taskName: string; iterations: string[]; concurrency: number } + | { jobId: string; type: 'iteration:start'; stepIndex: number; label: string } + | { jobId: string; type: 'iteration:complete'; stepIndex: number; label: string; cost?: { inputTokens: number; outputTokens: number; totalUSD: number } } + | { jobId: string; type: 'iteration:error'; stepIndex: number; label: string; error: string } + | { jobId: string; type: 'assistant:delta'; text: string; iterationLabel?: string } + | { jobId: string; type: 'assistant:text'; text: string; iterationLabel?: string } + | { jobId: string; type: 'tool:start'; toolCallId: string; toolName: string; toolInput: Record; iterationLabel?: string } + | { jobId: string; type: 'tool:result'; toolCallId: string; output: string; isError: boolean; iterationLabel?: string } + | { jobId: string; type: 'pipeline:complete'; totalCost: { inputTokens: number; outputTokens: number; totalUSD: number } } + | { jobId: string; type: 'error'; message: string } + | { jobId: string; type: 'stopped' } + | { type: 'job:created'; jobId: string } + | { type: 'job:state'; jobId: string; status: string; progress: unknown; cost: unknown } + | { type: 'error'; message: string }; export function usePipelineRunner() { const [phase, setPhase] = useState('ready'); const [isConnected, setIsConnected] = useState(false); + const [jobId, setJobId] = useState(null); const [steps, setSteps] = useState([]); const [currentStep, setCurrentStep] = useState(null); + const [parallelStep, setParallelStep] = useState(null); const [messages, setMessages] = useState([]); const [streamingText, setStreamingText] = useState(''); const [totalCost, setTotalCost] = useState<{ inputTokens: number; outputTokens: number; totalUSD: number } | null>(null); + const [runningCost, setRunningCost] = useState({ inputTokens: 0, outputTokens: 0, totalUSD: 0 }); const [hasError, setHasError] = useState(false); const [skippedItems, setSkippedItems] = useState>([]); + const [elapsed, setElapsed] = useState(0); const wsRef = useRef(null); const streamBufferRef = useRef(''); + const startTimeRef = useRef(0); + const timerRef = useRef | null>(null); + const jobIdRef = useRef(null); + const inParallelRef = useRef(false); const flushStream = useCallback(() => { const text = streamBufferRef.current; @@ -50,6 +80,183 @@ export function usePipelineRunner() { } }, []); + const stopTimer = useCallback(() => { + if (timerRef.current) { clearInterval(timerRef.current); timerRef.current = null; } + }, []); + + const addCost = useCallback((cost: { inputTokens: number; outputTokens: number; totalUSD: number }) => { + setRunningCost((prev) => ({ + inputTokens: prev.inputTokens + cost.inputTokens, + outputTokens: prev.outputTokens + cost.outputTokens, + totalUSD: prev.totalUSD + cost.totalUSD, + })); + }, []); + + const handleEvent = useCallback((msg: ServerMessage) => { + // Filter events by jobId (ignore events from other jobs) + if ('jobId' in msg && msg.jobId && jobIdRef.current && msg.jobId !== jobIdRef.current) return; + + switch (msg.type) { + case 'job:created': + jobIdRef.current = msg.jobId; + setJobId(msg.jobId); + break; + + case 'job:state': + // Reconnection to a completed/failed job + if (msg.status === 'completed' || msg.status === 'failed' || msg.status === 'stopped' || msg.status === 'interrupted') { + setPhase('done'); + if (msg.cost) setTotalCost(msg.cost as { inputTokens: number; outputTokens: number; totalUSD: number }); + if (msg.status === 'failed' || msg.status === 'interrupted') setHasError(true); + stopTimer(); + } + break; + + case 'pipeline:init': + setSteps(msg.steps); + break; + + case 'step:start': + flushStream(); + setMessages([]); + setParallelStep(null); + inParallelRef.current = false; + setCurrentStep({ + taskName: msg.taskName, + iteration: msg.iteration, + status: 'running', + }); + break; + + case 'step:complete': + flushStream(); + setCurrentStep((prev) => prev ? { ...prev, status: 'complete', cost: msg.cost } : null); + if (msg.cost) addCost(msg.cost); + break; + + case 'step:skip': + setSkippedItems((prev) => [...prev, { label: msg.label, reason: msg.reason }]); + break; + + case 'step:parallel': + flushStream(); + setMessages([]); + setCurrentStep(null); + inParallelRef.current = true; + setParallelStep({ + stepIndex: msg.stepIndex, + taskName: msg.taskName, + concurrency: msg.concurrency, + iterations: msg.iterations.map((label) => ({ label, status: 'pending' })), + }); + break; + + case 'iteration:start': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'running' } : it, + ), + }; + }); + break; + + case 'iteration:complete': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'complete', cost: msg.cost } : it, + ), + }; + }); + if (msg.cost) addCost(msg.cost); + break; + + case 'iteration:error': + setParallelStep((prev) => { + if (!prev) return prev; + return { + ...prev, + iterations: prev.iterations.map((it) => + it.label === msg.label ? { ...it, status: 'error', error: msg.error } : it, + ), + }; + }); + break; + + case 'assistant:delta': + // Skip messages from parallel sub-agents (shown in iteration grid instead) + if (inParallelRef.current && msg.iterationLabel) break; + streamBufferRef.current += msg.text; + setStreamingText(streamBufferRef.current); + break; + + case 'assistant:text': { + if (inParallelRef.current && msg.iterationLabel) break; + const text = msg.text || streamBufferRef.current; + if (text) { + setMessages((prev) => [...prev, { role: 'assistant', id: crypto.randomUUID(), text }]); + } + streamBufferRef.current = ''; + setStreamingText(''); + break; + } + + case 'tool:start': + if (inParallelRef.current && msg.iterationLabel) break; + 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': + if (inParallelRef.current && msg.iterationLabel) break; + 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'); + stopTimer(); + break; + + case 'error': + flushStream(); + setMessages((prev) => [...prev, { role: 'error' as const, id: crypto.randomUUID(), text: msg.message }]); + setHasError(true); + setPhase('done'); + stopTimer(); + break; + + case 'stopped': + flushStream(); + setPhase('done'); + stopTimer(); + break; + } + }, [flushStream, stopTimer, addCost]); + useEffect(() => { const token = localStorage.getItem('BEARER_TOKEN'); if (!token) return; @@ -59,98 +266,18 @@ export function usePipelineRunner() { const ws = new WebSocket(url); wsRef.current = ws; - ws.addEventListener('open', () => setIsConnected(true)); + ws.addEventListener('open', () => { + setIsConnected(true); + if (jobIdRef.current) { + ws.send(JSON.stringify({ type: 'attach', jobId: jobIdRef.current })); + } + }); 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; - } + handleEvent(msg); } catch { // ignore } @@ -159,6 +286,7 @@ export function usePipelineRunner() { return () => { ws.close(); wsRef.current = null; + stopTimer(); }; }, []); @@ -169,17 +297,32 @@ export function usePipelineRunner() { setMessages([]); setStreamingText(''); setTotalCost(null); + setRunningCost({ inputTokens: 0, outputTokens: 0, totalUSD: 0 }); setHasError(false); setSkippedItems([]); + setCurrentStep(null); + setParallelStep(null); + setElapsed(0); + setJobId(null); + jobIdRef.current = null; streamBufferRef.current = ''; + startTimeRef.current = Date.now(); + if (timerRef.current) clearInterval(timerRef.current); + timerRef.current = setInterval(() => { + setElapsed(Math.floor((Date.now() - startTimeRef.current) / 1000)); + }, 1000); + 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' })); + if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN || !jobIdRef.current) return; + wsRef.current.send(JSON.stringify({ type: 'stop', jobId: jobIdRef.current })); }, []); - return { phase, isConnected, steps, currentStep, messages, streamingText, totalCost, hasError, skippedItems, run, stop }; + return { + phase, isConnected, jobId, steps, currentStep, parallelStep, messages, streamingText, + totalCost, runningCost, hasError, skippedItems, elapsed, run, stop, + }; }