import { useState, useEffect, useRef, useCallback, useMemo, createContext, useContext } from 'react'; import { useParams, 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'; import { randomId } from 'helpers/random-id'; 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 PipelineJobDetail = () => { const { id } = useParams<{ id: string }>(); 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: randomId(), 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: randomId(), 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: randomId(), 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 */}
{/* The other two job detail panes already do this; this was the last imperative back in Jobs. */}

{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 */}
); };