Fix pipeline SSE delivery, add LLM timeout, and redesign progress UI

amoghbanta · amoghbanta · commit a5d081e22cd2 · 2026-02-10T16:14:50.000+01:00
- Fix SSE event loss: switch to queue-based writes with proper await
  on each writeSSE() call (was fire-and-forget, causing final events
  to be lost before stream closed)
- Fix SSE race condition: re-check job status after subscribing
  listener to catch completions during the gap
- Add 90s AbortSignal timeout to LLM calls to prevent indefinite hangs
  when OpenAI rate-limits concurrent requests
- Add per-page error handling in pipeline runner so one page timeout
  doesn't block remaining pages
- Redesign pipeline progress from vertical list to tile grid showing
  all parallel steps simultaneously with individual progress bars
- Fix step state detection to show all active parallel steps, not just
  the most recent one
diff --git a/apps/api/src/routes/pipeline.ts b/apps/api/src/routes/pipeline.ts
@@ -1,7 +1,7 @@
 import { Hono } from "hono"
 import { streamSSE } from "hono/streaming"
 import { HTTPException } from "hono/http-exception"
-import type { PipelineService } from "../services/pipeline-service.js"
+import type { PipelineService, PipelineSSEEvent } from "../services/pipeline-service.js"
 
 export function createPipelineRoutes(
   service: PipelineService,
@@ -91,47 +91,75 @@ export function createPipelineRoutes(
           return
         }
 
-        // Stream real-time events
-        let closed = false
+        // Queue-based SSE: listener pushes events, loop drains with awaited writes
+        const queue: PipelineSSEEvent[] = []
+        let done = false
+
         const unsubscribe = service.addListener(label, (event) => {
-          if (closed) return
-          try {
-            if (event.type === "progress") {
-              stream.writeSSE({
-                event: "progress",
-                data: JSON.stringify(event.data),
-              })
-            } else if (event.type === "pipeline-complete") {
-              stream.writeSSE({
-                event: "complete",
-                data: JSON.stringify({ label: event.label }),
-              })
-              closed = true
-            } else if (event.type === "pipeline-error") {
-              stream.writeSSE({
-                event: "error",
-                data: JSON.stringify({
-                  label: event.label,
-                  error: event.error,
-                }),
-              })
-              closed = true
-            }
-          } catch {
-            // Stream write failed (client disconnected)
-            closed = true
-          }
+          if (done) return
+          queue.push(event)
         })
 
-        // Keep stream alive until pipeline completes or client disconnects
+        // Re-check status after subscribing to avoid race where pipeline
+        // completes between the initial check and listener registration
+        const jobAfterSubscribe = service.getStatus(label)
+        if (
+          jobAfterSubscribe?.status === "completed" ||
+          jobAfterSubscribe?.status === "failed"
+        ) {
+          const event =
+            jobAfterSubscribe.status === "completed" ? "complete" : "error"
+          const data =
+            jobAfterSubscribe.status === "completed"
+              ? { label }
+              : { label, error: jobAfterSubscribe.error }
+          await stream.writeSSE({ event, data: JSON.stringify(data) })
+          unsubscribe()
+          return
+        }
+
         stream.onAbort(() => {
-          closed = true
+          done = true
           unsubscribe()
         })
 
-        // Wait for completion
-        while (!closed) {
-          await new Promise((r) => setTimeout(r, 100))
+        // Drain queue with proper await on each write
+        while (!done) {
+          while (queue.length > 0) {
+            const event = queue.shift()!
+            try {
+              if (event.type === "progress") {
+                await stream.writeSSE({
+                  event: "progress",
+                  data: JSON.stringify(event.data),
+                })
+              } else if (event.type === "pipeline-complete") {
+                await stream.writeSSE({
+                  event: "complete",
+                  data: JSON.stringify({ label: event.label }),
+                })
+                done = true
+                break
+              } else if (event.type === "pipeline-error") {
+                await stream.writeSSE({
+                  event: "error",
+                  data: JSON.stringify({
+                    label: event.label,
+                    error: event.error,
+                  }),
+                })
+                done = true
+                break
+              }
+            } catch {
+              // Stream write failed (client disconnected)
+              done = true
+              break
+            }
+          }
+          if (!done) {
+            await new Promise((r) => setTimeout(r, 50))
+          }
         }
 
         unsubscribe()
diff --git a/apps/api/src/services/pipeline-runner.ts b/apps/api/src/services/pipeline-runner.ts
@@ -150,68 +150,86 @@ export function createPipelineRunner(): PipelineRunner {
         let completedSection = 0
         let completedRender = 0
 
+        const failedPages: string[] = []
+
         await processWithConcurrency(
           pages,
           effectiveConcurrency,
           async (page: PageData) => {
-            await processPage(
-              page,
-              storage,
-              {
-                textClassifyConfig,
-                imageClassifyConfig,
-                sectioningConfig,
-                renderConfig,
-              },
-              llmModel,
-              progress,
-              totalPages,
-              {
-                onClassifyImages: () => {
-                  completedClassifyImages++
-                  progress.emit({
-                    type: "step-progress",
-                    step: "image-classification",
-                    message: `${completedClassifyImages}/${totalPages}`,
-                    page: completedClassifyImages,
-                    totalPages,
-                  })
-                },
-                onClassifyText: () => {
-                  completedClassifyText++
-                  progress.emit({
-                    type: "step-progress",
-                    step: "text-classification",
-                    message: `${completedClassifyText}/${totalPages}`,
-                    page: completedClassifyText,
-                    totalPages,
-                  })
-                },
-                onSection: () => {
-                  completedSection++
-                  progress.emit({
-                    type: "step-progress",
-                    step: "page-sectioning",
-                    message: `${completedSection}/${totalPages}`,
-                    page: completedSection,
-                    totalPages,
-                  })
+            try {
+              await processPage(
+                page,
+                storage,
+                {
+                  textClassifyConfig,
+                  imageClassifyConfig,
+                  sectioningConfig,
+                  renderConfig,
                 },
-                onRender: () => {
-                  completedRender++
-                  progress.emit({
-                    type: "step-progress",
-                    step: "web-rendering",
-                    message: `${completedRender}/${totalPages}`,
-                    page: completedRender,
-                    totalPages,
-                  })
-                },
-              }
-            )
+                llmModel,
+                progress,
+                totalPages,
+                {
+                  onClassifyImages: () => {
+                    completedClassifyImages++
+                    progress.emit({
+                      type: "step-progress",
+                      step: "image-classification",
+                      message: `${completedClassifyImages}/${totalPages}`,
+                      page: completedClassifyImages,
+                      totalPages,
+                    })
+                  },
+                  onClassifyText: () => {
+                    completedClassifyText++
+                    progress.emit({
+                      type: "step-progress",
+                      step: "text-classification",
+                      message: `${completedClassifyText}/${totalPages}`,
+                      page: completedClassifyText,
+                      totalPages,
+                    })
+                  },
+                  onSection: () => {
+                    completedSection++
+                    progress.emit({
+                      type: "step-progress",
+                      step: "page-sectioning",
+                      message: `${completedSection}/${totalPages}`,
+                      page: completedSection,
+                      totalPages,
+                    })
+                  },
+                  onRender: () => {
+                    completedRender++
+                    progress.emit({
+                      type: "step-progress",
+                      step: "web-rendering",
+                      message: `${completedRender}/${totalPages}`,
+                      page: completedRender,
+                      totalPages,
+                    })
+                  },
+                }
+              )
+            } catch (err) {
+              const msg = err instanceof Error ? err.message : String(err)
+              failedPages.push(`${page.pageId}: ${msg}`)
+              progress.emit({
+                type: "step-error",
+                step: "web-rendering",
+                error: `${page.pageId} failed: ${msg}`,
+              })
+            }
           }
         )
 
+        if (failedPages.length > 0) {
+          throw new Error(
+            `${failedPages.length} page(s) failed:\n${failedPages.join("\n")}`
+          )
+        }
+
         // Emit completion for all per-page steps
         progress.emit({
           type: "step-complete",
diff --git a/apps/studio/src/components/pipeline/PipelineProgress.tsx b/apps/studio/src/components/pipeline/PipelineProgress.tsx
@@ -27,6 +27,7 @@ function getStepState(
   progress: PipelineProgressState
 ): "pending" | "active" | "completed" | "error" {
   if (progress.completedSteps.has(step)) return "completed"
+  if (progress.stepProgress.has(step)) return "active"
   if (progress.currentStep === step) return "active"
   if (progress.error?.startsWith(step)) return "error"
   return "pending"
@@ -62,7 +63,7 @@ export function PipelineProgress({
       </CardHeader>
       <CardContent>
         {(isRunning || isComplete || error) && (
-          <div className="mb-4 space-y-0.5">
+          <div className="mb-4 grid grid-cols-2 gap-2 sm:grid-cols-3">
             {STEP_ORDER.map((step) => (
               <StepIndicator
                 key={step}
diff --git a/apps/studio/src/components/pipeline/StepIndicator.tsx b/apps/studio/src/components/pipeline/StepIndicator.tsx
@@ -40,7 +40,7 @@ function StepIcon({ state }: { state: StepState }) {
     case "error":
       return <AlertCircle className="h-4 w-4 text-destructive" />
     default:
-      return <Circle className="h-4 w-4 text-muted-foreground/40" />
+      return <Circle className="h-4 w-4 text-muted-foreground/30" />
   }
 }
 
@@ -49,44 +49,64 @@ export function StepIndicator({
   state,
   progress,
 }: StepIndicatorProps) {
+  const pct =
+    state === "active" && progress?.totalPages
+      ? Math.round(((progress.page ?? 0) / progress.totalPages) * 100)
+      : state === "completed"
+        ? 100
+        : 0
+
   return (
-    <div className="flex items-center gap-3 py-1.5">
-      <StepIcon state={state} />
-      <div className="flex-1">
-        <div
+    <div
+      className={cn(
+        "rounded-lg border p-3 transition-all",
+        state === "active" && "border-blue-200 bg-blue-50/50",
+        state === "completed" && "border-green-200 bg-green-50/30",
+        state === "error" && "border-destructive/30 bg-destructive/5",
+        state === "pending" && "border-border/50 bg-muted/20 opacity-60"
+      )}
+    >
+      <div className="mb-2 flex items-center justify-between">
+        <span
           className={cn(
-            "text-sm",
-            state === "active" && "font-medium text-foreground",
-            state === "completed" && "text-muted-foreground",
-            state === "pending" && "text-muted-foreground/60",
-            state === "error" && "text-destructive"
+            "text-sm font-medium",
+            state === "completed" && "text-green-700",
+            state === "error" && "text-destructive",
+            state === "pending" && "text-muted-foreground"
           )}
         >
           {label}
-        </div>
+        </span>
+        <StepIcon state={state} />
+      </div>
+
+      {/* Progress bar — always visible for active/completed */}
+      <div className="h-1.5 rounded-full bg-muted/60">
+        <div
+          className={cn(
+            "h-full rounded-full transition-all duration-300",
+            state === "active" && "bg-blue-600",
+            state === "completed" && "bg-green-500",
+            state === "error" && "bg-destructive",
+            state === "pending" && "bg-transparent"
+          )}
+          style={{ width: `${pct}%` }}
+        />
+      </div>
+
+      {/* Status text */}
+      <div className="mt-1.5 text-xs text-muted-foreground">
         {state === "active" && progress?.totalPages && (
-          <div className="mt-1">
-            <div className="flex justify-between text-xs text-muted-foreground">
-              <span>
-                {progress.page ?? 0} / {progress.totalPages} pages
-              </span>
-              <span>
-                {Math.round(
-                  ((progress.page ?? 0) / progress.totalPages) * 100
-                )}
-                %
-              </span>
-            </div>
-            <div className="mt-0.5 h-1.5 rounded-full bg-muted">
-              <div
-                className="h-full rounded-full bg-blue-600 transition-all"
-                style={{
-                  width: `${((progress.page ?? 0) / progress.totalPages) * 100}%`,
-                }}
-              />
-            </div>
-          </div>
+          <span>
+            {progress.page ?? 0} / {progress.totalPages} pages
+          </span>
+        )}
+        {state === "active" && !progress?.totalPages && (
+          <span>Processing...</span>
         )}
+        {state === "completed" && <span>Done</span>}
+        {state === "error" && <span>Failed</span>}
+        {state === "pending" && <span>Waiting</span>}
       </div>
     </div>
   )
diff --git a/packages/llm/src/client.ts b/packages/llm/src/client.ts
@@ -209,6 +209,7 @@ async function callLLM<T>(
     schema: opts.schema,
     system,
     messages: coreMessages,
+    abortSignal: AbortSignal.timeout(90_000),
   }
   if (opts.maxTokens) {
     generateOpts.maxTokens = opts.maxTokens

Original file line number	Diff line number	Diff line change
`@@ -209,6 +209,7 @@ async function callLLM<T>(`
`209`	`209`	`schema: opts.schema,`
`210`	`210`	`system,`
`211`	`211`	`messages: coreMessages,`
	`212`	`+ abortSignal: AbortSignal.timeout(90_000),`
`212`	`213`	`}`
`213`	`214`	`if (opts.maxTokens) {`
`214`	`215`	`generateOpts.maxTokens = opts.maxTokens`