officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pipeline: add activity timeout and waiting indicator for local models

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-03-12 10:56:39 +00:00
co-authored by Claude Opus 4.6
parent 92da03fcff
commit ea8113e61d
3 changed files with 56 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/tasks/pipeline-executor.ts
+38 -16
View File
@@ -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<void> {
}
}
// 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<MessageCost> {
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); });
}
});
}
src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx
+6
View File
@@ -690,6 +690,12 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => {
<span>{formatElapsed(pipeline.elapsed)}</span>
{totalTokens > 0 && <span>{totalTokens.toLocaleString()} tok</span>}
{rc.totalUSD > 0 && <span>${rc.totalUSD.toFixed(3)}</span>}
{pipeline.waitingStatus && (
<span className="ml-auto flex items-center gap-1.5 text-amber-500">
<Loader2 className="h-3 w-3 animate-spin" />
Waiting for model ({formatElapsed(pipeline.waitingStatus.elapsed)})
</span>
)}
</div>
)}
src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts
+12 -1
View File
@@ -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<Array<{ label: string; reason: string }>>([]);
const [elapsed, setElapsed] = useState(0);
const [waitingStatus, setWaitingStatus] = useState<{ stepIndex: number; elapsed: number; iterationLabel?: string } | null>(null);
const wsRef = useRef<WebSocket | null>(null);
const streamBufferRef = useRef('');
const startTimeRef = useRef<number>(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,
};
}