diff --git a/apps/agent/src/agent-runner.ts b/apps/agent/src/agent-runner.ts index f349b76d..01e6ff38 100644 --- a/apps/agent/src/agent-runner.ts +++ b/apps/agent/src/agent-runner.ts @@ -82,7 +82,8 @@ import { startTurnTrace, } from './langfuse.js'; import { withOpenRouterAttribution } from './agent-runner/openrouter-attribution.js'; -import { buildUserFacingModelError } from './agent-runner/user-facing-error.js'; +import { buildUserFacingModelError, classifyModelError } from './agent-runner/user-facing-error.js'; +import type { ModelErrorKind } from './agent-runner/user-facing-error.js'; import { withAmikoTwinAttribution } from './agent-runner/amiko-attribution.js'; import { awaitTriggeredTurn, @@ -142,6 +143,22 @@ const addUserIdToList = (existing: string[], userId: string | undefined): string return existing.includes(userId) ? existing : [...existing, userId]; }; +/** + * Raised by `runScheduledJob` when a cron firing's turn ended in a model error + * (`stopReason==='error'`). A model error means the run produced no reply, so + * it must be recorded as a *failed* run — the central scheduler then applies + * exponential backoff and auto-recovers on the next success, instead of + * treating the empty turn as success and re-firing at full cadence (which, on + * an account-wide 402 credit outage, turned every frequent schedule into a + * per-tick error storm). The `kind` classifies the underlying provider error. + */ +export class ScheduledRunModelError extends Error { + constructor(readonly kind: ModelErrorKind, rawMessage: string) { + super(`scheduled run failed (${kind}): ${rawMessage}`); + this.name = 'ScheduledRunModelError'; + } +} + export class AgentRunner implements SessionRuntime { readonly events = new SessionEventBroker(); /** Per-agent typed event bus — subscribed to by future plugins. */ @@ -254,6 +271,18 @@ export class AgentRunner implements SessionRuntime { () => this.postMessage(sessionId, { text: transformed.prompt, metadata }), () => this.waitForSessionIdle(sessionId), ); + + // A model error leaves the turn with stopReason==='error' instead of + // throwing, so without this the run would be recorded as a success and + // the schedule would keep firing at full cadence while every firing + // fails (an account-wide 402 credit outage then storms every tick). + // Surface it as a failed run so the central scheduler backs off and + // auto-recovers. Scoped to cron: `once` firings don't re-fire, so their + // completion semantics are left unchanged. + const turnError = this.sessions.get(sessionId)?.lastTurnModelError; + if (turnError && schedule.type === 'cron') { + throw new ScheduledRunModelError(turnError.kind, turnError.message); + } } catch (error) { runFailed = true; throw error; @@ -1444,6 +1473,7 @@ export class AgentRunner implements SessionRuntime { session.reasoningTagStream = undefined; session.reasoningCarryTagName = undefined; session.consecutiveToolFailures = 0; + delete session.lastTurnModelError; if (message.messageId !== undefined) { session.currentTurnCorrelationId = message.messageId; } else { @@ -3386,6 +3416,10 @@ export class AgentRunner implements SessionRuntime { // Handle error responses from the model provider. if (assistantMessage.stopReason === 'error') { const errorMsg = assistantMessage.errorMessage ?? 'Model returned an error.'; + // Record the failure synchronously (not via a side-effect) so it is + // visible the moment the turn's queue settles — runScheduledJob reads + // it right after waitForSessionIdle to decide success vs. failed run. + session.lastTurnModelError = { kind: classifyModelError(errorMsg), message: errorMsg }; const ts = new Date().toISOString(); session.updatedAt = ts; void this.queueSideEffect(session, () => this.persistSessionIndex(session)); diff --git a/apps/agent/src/agent-runner/types.ts b/apps/agent/src/agent-runner/types.ts index d45886e3..1f75849f 100644 --- a/apps/agent/src/agent-runner/types.ts +++ b/apps/agent/src/agent-runner/types.ts @@ -6,6 +6,7 @@ import type { LangfuseClientLike, LangfuseTurnContext } from '../langfuse.js'; import type { SessionDescriptor } from '../runtime.js'; import type { ApprovalGate } from './approval-gate.js'; import type { ReasoningTagStreamState, SpeakerTagStreamState } from './message-utils.js'; +import type { ModelErrorKind } from './user-facing-error.js'; export interface RunnerSession extends SessionDescriptor { agent: Agent; @@ -56,6 +57,13 @@ export interface RunnerSession extends SessionDescriptor { * when this reaches `MAX_CONSECUTIVE_TOOL_FAILURES` to prevent the * model from looping forever against a broken tool. */ consecutiveToolFailures: number; + /** Set at message_end when the turn ended in a model error + * (`stopReason==='error'`), cleared at turn start. `runScheduledJob` reads + * it so a cron run that produced no reply is recorded as a *failed* run + * (→ scheduler exponential backoff, auto-recovering on the next success) + * instead of a silent success that keeps re-firing at full cadence — the + * behaviour that turned an account-wide 402 outage into a per-tick storm. */ + lastTurnModelError?: { kind: ModelErrorKind; message: string }; } export interface AgentRunnerOptions { diff --git a/apps/agent/src/agent-runner/user-facing-error.ts b/apps/agent/src/agent-runner/user-facing-error.ts index 846eb76f..0e7cdb05 100644 --- a/apps/agent/src/agent-runner/user-facing-error.ts +++ b/apps/agent/src/agent-runner/user-facing-error.ts @@ -25,7 +25,7 @@ const CLASSIFIERS: Array<{ kind: ModelErrorKind; pattern: RegExp }> = [ { kind: 'quota', pattern: /key limit exceeded|insufficient[\s\w]{0,20}credit|out of credit|quota exceeded|spend(ing)? limit|payment required|\b402\b/i }, { kind: 'rate_limit', pattern: /\b429\b|rate.?limit|too many requests/i }, { kind: 'auth', pattern: /invalid.{0,10}(api.?key|token)|unauthorized|authentication|\b401\b|no auth credentials/i }, - { kind: 'unavailable', pattern: /\b5\d\d\b|overloaded|service unavailable|timed?.?out|econn|network error|internal (server )?error|bad gateway/i }, + { kind: 'unavailable', pattern: /\b5\d\d\b|overloaded|service unavailable|timed?.?out|econn|network error|internal (server )?error|bad gateway|no endpoints found|\b404\b/i }, ]; export const classifyModelError = (raw: string): ModelErrorKind => { diff --git a/apps/agent/test/scheduled-turn.test.ts b/apps/agent/test/scheduled-turn.test.ts index cd7fe1b3..3d1f68cb 100644 --- a/apps/agent/test/scheduled-turn.test.ts +++ b/apps/agent/test/scheduled-turn.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import type { ScheduleRecord } from '@openhermit/store'; -import { AgentRunner } from '../src/agent-runner.js'; +import { AgentRunner, ScheduledRunModelError } from '../src/agent-runner.js'; import { awaitTriggeredTurn, surfaceRunError, @@ -233,6 +233,94 @@ test('runScheduledJob preserves the run error when ephemeral teardown also fails ); }); +test('runScheduledJob fails a cron run whose turn ended in a model error (in-band, no throw)', async () => { + // A 402 leaves the turn with stopReason==='error' — the turn "completes" + // (waitForSessionIdle resolves) but produced no reply. Without surfacing it, + // the scheduler records success and re-fires at full cadence. lastTurnModelError + // is what handleAgentEvent stamps on the session at message_end. + const schedule: ScheduleRecord = { + agentId: 'agent-1', + scheduleId: 'schedule-1', + type: 'cron', + status: 'active', + cronExpression: '* * * * *', + prompt: 'scan feed', + sessionMode: { kind: 'dedicated' }, + delivery: { kind: 'silent' }, + policy: {}, + createdAt: '2026-09-01T00:00:00.000Z', + updatedAt: '2026-09-01T00:00:00.000Z', + runCount: 0, + consecutiveErrors: 0, + }; + const sessionId = 'schedule:schedule-1'; + const session = { + status: 'running', + queue: Promise.resolve(), + lastTurnModelError: { kind: 'quota' as const, message: '402 Insufficient credits' }, + }; + const fakeRunner = { + scope: { agentId: 'agent-1' }, + sessions: new Map([[sessionId, session]]), + bus: { + transform: async (_event: string, payload: Record) => payload, + }, + openSession: async () => undefined, + postMessage: async () => ({ sessionId, triggered: true }), + // Turn completes normally; the error is in-band, not thrown. + waitForSessionIdle: async () => undefined, + }; + + await assert.rejects( + (AgentRunner.prototype.runScheduledJob as Function).call(fakeRunner, schedule, sessionId), + (err: unknown) => err instanceof ScheduledRunModelError && err.kind === 'quota', + ); + // Dedicated cron: the queue is healed so later checkpoints aren't skipped. + await assert.doesNotReject(session.queue); +}); + +test('runScheduledJob leaves a once run untouched on a model error', async () => { + // once firings never re-fire, so their completion semantics must not change. + const schedule = { + agentId: 'agent-1', + scheduleId: 'schedule-1', + type: 'once', + status: 'active', + prompt: 'one-shot', + sessionMode: { kind: 'dedicated' }, + delivery: { kind: 'silent' }, + policy: {}, + createdAt: '2026-09-01T00:00:00.000Z', + updatedAt: '2026-09-01T00:00:00.000Z', + runCount: 0, + consecutiveErrors: 0, + } as unknown as ScheduleRecord; + const sessionId = 'schedule:schedule-1'; + const session = { + status: 'running', + lastTurnModelError: { kind: 'quota' as const, message: '402 Insufficient credits' }, + }; + const fakeRunner = { + scope: { agentId: 'agent-1' }, + sessions: new Map([[sessionId, session]]), + bus: { + transform: async (_event: string, payload: Record) => payload, + emit: async () => undefined, + }, + openSession: async () => undefined, + postMessage: async () => ({ sessionId, triggered: true }), + waitForSessionIdle: async () => undefined, + clearIdleSummaryTimer: () => undefined, + persistSessionIndex: async () => undefined, + }; + + await assert.doesNotReject( + (AgentRunner.prototype.runScheduledJob as Function).call(fakeRunner, schedule, sessionId), + ); + // once => teardown drops the in-memory session. + assert.equal(fakeRunner.sessions.has(sessionId), false); +}); + test('runScheduledJob propagates an ephemeral teardown failure after a successful run', async () => { const schedule: ScheduleRecord = { agentId: 'agent-1', diff --git a/apps/agent/test/user-facing-error.test.ts b/apps/agent/test/user-facing-error.test.ts index 052d99e6..791f617e 100644 --- a/apps/agent/test/user-facing-error.test.ts +++ b/apps/agent/test/user-facing-error.test.ts @@ -32,6 +32,13 @@ test('classifies auth failures', () => { assert.equal(classifyModelError('Invalid API key provided'), 'auth'); }); +test('classifies a 404 dead-model / no-endpoints error as unavailable, not generic', () => { + assert.equal( + classifyModelError('404 No endpoints found for anthropic/claude-3.5-haiku.'), + 'unavailable', + ); +}); + test('classifies context overflow before quota-ish words', () => { assert.equal( classifyModelError('This model\'s maximum context length is 128000 tokens'),