diff --git a/src/framework.ts b/src/framework.ts index 5583822..a25deca 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -322,7 +322,7 @@ export class AgentFramework { private maintenanceRunId = 0; private currentMaintenanceRun: ContextMaintenanceRun | null = null; private maintenanceHistory: ContextMaintenanceRun[] = []; - /** Last time we console-warned about stale (busy-requeued) inference requests, per agent. */ + /** Last time we reported stale (busy-requeued) inference requests, per agent. */ private staleWarnAt = new Map(); /** Per-agent last inference activity (epoch ms), for /healthz + doctor tooling. */ private lastInferenceAt = new Map(); @@ -2852,12 +2852,21 @@ export class AgentFramework { // Check for inference requests await this.processInferenceRequests(); - // Yield to the event loop between iterations. - // Full 10ms sleep when truly idle; minimal yield when streams are active - // (needed to let stream microtasks and tool-call callbacks execute). - if (!event && this.pendingRequests.length === 0) { + // Yield to the event loop between iterations. A pending inference request + // is not necessarily runnable: while its agent is streaming or waiting for + // tools, processInferenceRequests() deliberately requeues it. Treating that + // requeued request as "work made progress" creates a microtask-only polling + // loop. If activeStreams bookkeeping is absent/stale at the same time, the + // old code had no await at all and starved the tool-result and HTTP I/O that + // could make the agent runnable again (the Sol outage, 2026-07-15). + // + // No queue event means no foreground progress, so always take the normal + // polling backoff. After an event, retain the low-latency macrotask yield + // while background work remains. Both paths give Bun/Node a real event-loop + // turn; neither can recurse forever through already-resolved promises. + if (!event) { await new Promise((resolve) => setTimeout(resolve, 10)); - } else if (this.activeStreams.size > 0) { + } else if (this.pendingRequests.length > 0 || this.activeStreams.size > 0) { await new Promise((resolve) => setTimeout(resolve, 0)); } } @@ -3654,7 +3663,15 @@ export class AgentFramework { if (agent.state.status === 'inferring' || agent.state.status === 'streaming' || agent.state.status === 'waiting_for_tools') { // Re-queue requests, but warn if they've been pending too long const oldest = Math.min(...requests.map(r => r.timestamp)); - if (now - oldest > STALE_REQUEST_MS) { + if ( + now - oldest > STALE_REQUEST_MS && + (this.staleWarnAt.get(agentName) ?? 0) < now - 60_000 + ) { + // Trace and stderr share the throttle. The old code throttled only + // stderr, so a long-running tool call generated one trace per poll + // (up to 100/sec after the scheduler backoff), adding avoidable work + // precisely while the agent was already under pressure. + this.staleWarnAt.set(agentName, now); this.emitTrace({ type: 'inference:request_stale', agentName, @@ -3662,14 +3679,10 @@ export class AgentFramework { requestCount: requests.length, oldestRequestAge: now - oldest, }); - // Loud (but throttled) note that requests are waiting on a busy agent. - if ((this.staleWarnAt.get(agentName) ?? 0) < now - 60_000) { - this.staleWarnAt.set(agentName, now); - console.error( - `[inference-stale] agent=${agentName} busy (${agent.state.status}) — ` + - `${requests.length} request(s) waiting ${Math.round((now - oldest) / 1000)}s`, - ); - } + console.error( + `[inference-stale] agent=${agentName} busy (${agent.state.status}) — ` + + `${requests.length} request(s) waiting ${Math.round((now - oldest) / 1000)}s`, + ); } this.pendingRequests.push(...requests); continue; diff --git a/test/busy-agent-scheduler.test.ts b/test/busy-agent-scheduler.test.ts new file mode 100644 index 0000000..b4ed7d6 --- /dev/null +++ b/test/busy-agent-scheduler.test.ts @@ -0,0 +1,89 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { AgentFramework } from '../src/framework.js'; + +interface FrameworkHarness { + queue: { tryPop(): unknown | null }; + pendingRequests: Array<{ + agentName: string; + reason: string; + source: string; + timestamp: number; + }>; + agents: Map; + activeStreams: Map>; + staleWarnAt: Map; + sweepExpiredConversations(): void; + createFrameworkState(): Record; + emitTrace(): void; + handleProcessEvent(event: unknown): Promise; + processNextEvent(): Promise; + processInferenceRequests(): Promise; +} + +function busyAgentHarness(): FrameworkHarness { + const framework = Object.create(AgentFramework.prototype) as FrameworkHarness; + framework.queue = { tryPop: () => null }; + framework.pendingRequests = [{ + agentName: 'Sol', + reason: 'discord-message', + source: 'discord', + timestamp: Date.now(), + }]; + framework.agents = new Map([['Sol', { state: { status: 'waiting_for_tools' } }]]); + framework.activeStreams = new Map(); + framework.staleWarnAt = new Map(); + framework.sweepExpiredConversations = () => {}; + framework.createFrameworkState = () => ({}); + framework.emitTrace = () => {}; + framework.handleProcessEvent = async () => {}; + return framework; +} + +test('busy-agent inference requests yield to timers when no active stream is registered', async () => { + const framework = busyAgentHarness(); + let timerRan = false; + setTimeout(() => { + timerRan = true; + }, 0); + + await framework.processNextEvent(); + + assert.equal(timerRan, true, 'scheduler must yield a macrotask before polling the busy agent again'); + assert.equal(framework.pendingRequests.length, 1, 'the deferred wake remains queued'); +}); + +test('continuous queue traffic still yields when a busy-agent wake remains deferred', async () => { + const framework = busyAgentHarness(); + framework.queue = { tryPop: () => ({ type: 'diagnostic-event' }) }; + let timerRan = false; + setTimeout(() => { + timerRan = true; + }, 0); + + await framework.processNextEvent(); + + assert.equal(timerRan, true, 'queue traffic must not starve tool-result or HTTP I/O'); + assert.equal(framework.pendingRequests.length, 1, 'the busy-agent wake remains deferred'); +}); + +test('stale busy-agent diagnostics are throttled with the warning', async () => { + const framework = busyAgentHarness(); + framework.pendingRequests[0].timestamp = Date.now() - 31_000; + let traceCount = 0; + framework.emitTrace = () => { + traceCount++; + }; + + const originalError = console.error; + console.error = () => {}; + try { + await framework.processInferenceRequests(); + await framework.processInferenceRequests(); + } finally { + console.error = originalError; + } + + assert.equal(traceCount, 1, 'the stale trace shares the once-per-minute warning throttle'); + assert.equal(framework.pendingRequests.length, 1, 'diagnostics do not consume the deferred wake'); +});