From ea8113e61d4f857d7a11f70a99f21f88577644c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 12 Mar 2026 10:56:39 +0000 Subject: [PATCH] pipeline: add activity timeout and waiting indicator for local models Agentic steps now have a 5-minute inactivity timeout (non-Claude-Code models only) so stalled local LLM requests fail with a clear error instead of hanging forever. A "Waiting for model" indicator is emitted every 10s and shown in the pipeline UI stats bar, clearing as soon as the model starts responding. Co-Authored-By: Claude Opus 4.6 --- src/servers/api/tasks/pipeline-executor.ts | 54 +++++++++++++------ .../components/TaskRunnerModal.tsx | 6 +++ .../components/usePipelineRunner.ts | 13 ++++- 3 files changed, 56 insertions(+), 17 deletions(-) diff --git a/src/servers/api/tasks/pipeline-executor.ts b/src/servers/api/tasks/pipeline-executor.ts index 3f7dde8a..9c8d6e54 100644 --- a/src/servers/api/tasks/pipeline-executor.ts +++ b/src/servers/api/tasks/pipeline-executor.ts @@ -45,6 +45,7 @@ export type OutMessage = | { type: 'step:complete'; stepIndex: number; cost?: MessageCost } | { type: 'step:skip'; stepIndex: number; label: string; reason: string } | { type: 'step:parallel'; stepIndex: number; taskName: string; iterations: string[]; concurrency: number } + | { type: 'step:waiting'; stepIndex: number; iterationLabel?: string; elapsed: 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 } @@ -86,6 +87,11 @@ async function refreshProxyToken(): Promise { } } +// How long (ms) to wait with zero activity before timing out an agentic step. +// Only applies to non-Claude-Code models (local LLMs etc). +const ACTIVITY_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes +const WAITING_INTERVAL_MS = 10 * 1000; // emit "waiting" every 10s + async function runAgenticStep({ userId, email, username, role, taskDirName, prompt, cwd, model, abortSignal, emit, stepIndex, iterationLabel }: RunStepParams): Promise { const sessionId = randomUUID(); const isClaudeCode = model.startsWith('claude-code'); @@ -99,9 +105,21 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom if (abortSignal.aborted) return reject(new Error('Pipeline aborted')); let cleanup: (() => void) | null = null; + let settled = false; + let lastActivity = Date.now(); + const stepStart = Date.now(); + + const settle = (fn: () => void) => { + if (settled) return; + settled = true; + clearInterval(abortPoll); + clearInterval(waitingPoll); + fn(); + }; const onEvent = (event: PiEvent) => { if (abortSignal.aborted) return; + lastActivity = Date.now(); switch (event.type) { case 'delta': @@ -117,32 +135,38 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom emit({ type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError, stepIndex, iterationLabel }); break; case 'result': - clearInterval(abortPoll); - cleanup?.(); - resolve(event.cost); + settle(() => { cleanup?.(); resolve(event.cost); }); break; case 'error': - clearInterval(abortPoll); - cleanup?.(); - reject(new Error(event.message)); + settle(() => { cleanup?.(); reject(new Error(event.message)); }); break; case 'stopped': - clearInterval(abortPoll); - cleanup?.(); - reject(new Error('Step was stopped')); + settle(() => { cleanup?.(); reject(new Error('Step was stopped')); }); break; } }; - // Poll for abort signal to kill the running agent + // Poll for abort signal and activity timeout const abortPoll = setInterval(() => { if (abortSignal.aborted) { - clearInterval(abortPoll); - cleanup?.(); - reject(new Error('Pipeline was stopped')); + settle(() => { cleanup?.(); reject(new Error('Pipeline was stopped')); }); + return; + } + // Activity timeout (skip for Claude Code which has its own mechanisms) + if (!isClaudeCode && Date.now() - lastActivity > ACTIVITY_TIMEOUT_MS) { + const elapsed = Math.round((Date.now() - stepStart) / 1000); + console.error(`[pipeline] step ${stepIndex} timed out after ${elapsed}s of inactivity (session=${sessionId})`); + settle(() => { cleanup?.(); reject(new Error(`Step timed out — no response from model for ${Math.round(ACTIVITY_TIMEOUT_MS / 1000)}s`)); }); } }, 500); + // Emit periodic "waiting" events so the frontend knows the step is alive + const waitingPoll = setInterval(() => { + if (settled) return; + const elapsed = Math.round((Date.now() - stepStart) / 1000); + emit({ type: 'step:waiting', stepIndex, iterationLabel, elapsed }); + }, WAITING_INTERVAL_MS); + try { if (isClaudeCode) { const handle = await sendClaudeCodeStreaming({ @@ -171,9 +195,7 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom } } catch (err) { - clearInterval(abortPoll); - cleanup?.(); - reject(err); + settle(() => { cleanup?.(); reject(err); }); } }); } 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 09f8f3c7..7e4156d4 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx @@ -690,6 +690,12 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { {formatElapsed(pipeline.elapsed)} {totalTokens > 0 && {totalTokens.toLocaleString()} tok} {rc.totalUSD > 0 && ${rc.totalUSD.toFixed(3)}} + {pipeline.waitingStatus && ( + + + Waiting for model ({formatElapsed(pipeline.waitingStatus.elapsed)}) + + )} )} 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 0d0873f7..76d96494 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts @@ -36,6 +36,7 @@ type ServerMessage = | { 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: 'step:waiting'; stepIndex: number; iterationLabel?: string; elapsed: 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 } @@ -64,6 +65,7 @@ export function usePipelineRunner() { const [hasError, setHasError] = useState(false); const [skippedItems, setSkippedItems] = useState>([]); const [elapsed, setElapsed] = useState(0); + const [waitingStatus, setWaitingStatus] = useState<{ stepIndex: number; elapsed: number; iterationLabel?: string } | null>(null); const wsRef = useRef(null); const streamBufferRef = useRef(''); const startTimeRef = useRef(0); @@ -120,6 +122,7 @@ export function usePipelineRunner() { flushStream(); setMessages([]); setParallelStep(null); + setWaitingStatus(null); inParallelRef.current = false; setCurrentStep({ taskName: msg.taskName, @@ -130,6 +133,7 @@ export function usePipelineRunner() { case 'step:complete': flushStream(); + setWaitingStatus(null); setCurrentStep((prev) => prev ? { ...prev, status: 'complete', cost: msg.cost } : null); if (msg.cost) addCost(msg.cost); break; @@ -138,6 +142,10 @@ export function usePipelineRunner() { setSkippedItems((prev) => [...prev, { label: msg.label, reason: msg.reason }]); break; + case 'step:waiting': + setWaitingStatus({ stepIndex: msg.stepIndex, elapsed: msg.elapsed, iterationLabel: msg.iterationLabel }); + break; + case 'step:parallel': flushStream(); setMessages([]); @@ -191,6 +199,7 @@ export function usePipelineRunner() { case 'assistant:delta': // Skip messages from parallel sub-agents (shown in iteration grid instead) if (inParallelRef.current && msg.iterationLabel) break; + setWaitingStatus(null); streamBufferRef.current += msg.text; setStreamingText(streamBufferRef.current); break; @@ -208,6 +217,7 @@ export function usePipelineRunner() { case 'tool:start': if (inParallelRef.current && msg.iterationLabel) break; + setWaitingStatus(null); flushStream(); setMessages((prev) => [ ...prev, @@ -302,6 +312,7 @@ export function usePipelineRunner() { setSkippedItems([]); setCurrentStep(null); setParallelStep(null); + setWaitingStatus(null); setElapsed(0); setJobId(null); jobIdRef.current = null; @@ -323,6 +334,6 @@ export function usePipelineRunner() { return { phase, isConnected, jobId, steps, currentStep, parallelStep, messages, streamingText, - totalCost, runningCost, hasError, skippedItems, elapsed, run, stop, + totalCost, runningCost, hasError, skippedItems, elapsed, waitingStatus, run, stop, }; }