diff --git a/packages/@n8n/agents/src/runtime/__tests__/observation-log-observer-retry.test.ts b/packages/@n8n/agents/src/runtime/__tests__/observation-log-observer-retry.test.ts new file mode 100644 index 00000000000..a39d32b4425 --- /dev/null +++ b/packages/@n8n/agents/src/runtime/__tests__/observation-log-observer-retry.test.ts @@ -0,0 +1,201 @@ +import { APICallError } from 'ai'; +import { MockLanguageModelV3 } from 'ai/test'; + +import type { ExecutionOptions, RunOptions } from '../../types/sdk/agent'; +import type { AgentRuntimeConfig } from '../loop/agent-runtime'; +import { MemoryOrchestrator } from '../memory/memory-orchestrator'; +import { InMemoryMemory } from '../memory/memory-store'; +import { + createObservationLogObserveFn, + DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES, +} from '../memory/observation-log-defaults'; +import type { ScopedMemoryTaskEvent } from '../memory/scoped-memory-task-runner'; +import { AgentMessageList } from '../model/message-list'; +import { BackgroundTaskTracker } from '../state/background-task-tracker'; +import { AgentEventBus } from '../state/event-bus'; +import { RuntimeTelemetry } from '../telemetry/runtime-telemetry'; + +type GenerateResult = Awaited>; + +const THREAD_ID = 'thread-1'; +const RESOURCE_ID = 'user-1'; +// Covers the SDK backoff for every retry (2s, 4s, 8s) with room to spare. +const BACKOFF_WINDOW_MS = 60_000; + +function apiError(statusCode: number): APICallError { + return new APICallError({ + message: `Provider returned ${statusCode}`, + url: 'https://provider.test/v1/messages', + requestBodyValues: {}, + statusCode, + }); +} + +function textResult(text: string): GenerateResult { + return { + content: [{ type: 'text', text }], + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 10, noCache: 10, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 5, text: 5, reasoning: 0 }, + }, + warnings: [], + }; +} + +function runOptions(): RunOptions & ExecutionOptions { + return { persistence: { threadId: THREAD_ID, resourceId: RESOURCE_ID } }; +} + +function setup(doGenerate: () => Promise) { + const store = new InMemoryMemory(); + const model = new MockLanguageModelV3({ doGenerate }); + const events: ScopedMemoryTaskEvent[] = []; + const config = { + name: 'observer-retry-agent', + memory: store, + observationalMemory: { + observerThresholdTokens: 10, + observationLogTailLimit: 20, + observe: createObservationLogObserveFn(model), + }, + onMemoryTaskEvent: (event: ScopedMemoryTaskEvent) => events.push(event), + } as unknown as AgentRuntimeConfig; + const tracker = new BackgroundTaskTracker(); + const orchestrator = new MemoryOrchestrator( + config, + tracker, + new AgentEventBus(), + new RuntimeTelemetry(config), + async (text) => await Promise.resolve(text.length), + ); + return { store, model, events, orchestrator, tracker }; +} + +/** Run one mid-run boundary and let the SDK retry backoff elapse. */ +async function observeMidRun( + orchestrator: MemoryOrchestrator, + list: AgentMessageList, +): Promise { + const boundary = orchestrator.maybeObserveMidRun(list, runOptions()); + await vi.advanceTimersByTimeAsync(BACKOFF_WINDOW_MS); + await boundary; +} + +function turn(text: string): AgentMessageList { + const list = new AgentMessageList(); + list.addInput([{ role: 'user', content: [{ type: 'text', text }] }]); + list.addResponse([{ role: 'assistant', content: [{ type: 'text', text: 'Noted.' }] }]); + return list; +} + +describe('Observer retry for transient model failures', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('records observations once after a transient 503', async () => { + const doGenerate = vi + .fn<() => Promise>() + .mockRejectedValueOnce(apiError(503)) + .mockResolvedValue(textResult('* CRITICAL (14:30) User chose the violet theme.')); + const { store, orchestrator } = setup(doGenerate); + const list = turn('Use the violet theme for the launch page.'); + + await observeMidRun(orchestrator, list); + + expect(doGenerate).toHaveBeenCalledTimes(2); + expect(await store.getActiveObservationLog({ observationScopeId: THREAD_ID })).toMatchObject([ + { marker: 'critical', text: 'User chose the violet theme.' }, + ]); + const cursor = await store.getCursor(THREAD_ID); + expect(cursor?.lastObservedMessageId).toBe(list.messages().at(-1)?.id); + }); + + it('stops at the retry limit, keeps memory unchanged, and observes again on a new run', async () => { + const doGenerate = vi.fn<() => Promise>().mockRejectedValue(apiError(503)); + const { store, events, orchestrator, tracker } = setup(doGenerate); + const list = turn('Use the violet theme for the launch page.'); + + await expect(observeMidRun(orchestrator, list)).resolves.toBeUndefined(); + // A later boundary in the same run does not start another attempt. + await observeMidRun(orchestrator, list); + await orchestrator.saveToMemory(list, runOptions()); + await tracker.flush(); + + expect(doGenerate).toHaveBeenCalledTimes(DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES + 1); + expect(events).toContainEqual(expect.objectContaining({ type: 'failed' })); + expect(await store.getActiveObservationLog({ observationScopeId: THREAD_ID })).toEqual([]); + expect(await store.getCursor(THREAD_ID)).toBeNull(); + expect(list.llmVisibleMessages()).toEqual(list.messages()); + + doGenerate.mockReset(); + doGenerate.mockResolvedValue(textResult('* CRITICAL (14:30) User chose the violet theme.')); + const next = new AgentMessageList(); + await orchestrator.loadInto(next, runOptions()); + next.addInput([{ role: 'user', content: [{ type: 'text', text: 'Continue the work.' }] }]); + await observeMidRun(orchestrator, next); + + expect(doGenerate).toHaveBeenCalledTimes(1); + expect(await store.getActiveObservationLog({ observationScopeId: THREAD_ID })).toMatchObject([ + { text: 'User chose the violet theme.' }, + ]); + expect(await store.getCursor(THREAD_ID)).not.toBeNull(); + }); + + it.each([400, 401])('does not retry a permanent %i error', async (statusCode) => { + const doGenerate = vi + .fn<() => Promise>() + .mockRejectedValue(apiError(statusCode)); + const { store, events, orchestrator } = setup(doGenerate); + const list = turn('Use the violet theme for the launch page.'); + + await observeMidRun(orchestrator, list); + + expect(doGenerate).toHaveBeenCalledTimes(1); + expect(events).toContainEqual(expect.objectContaining({ type: 'failed' })); + expect(await store.getCursor(THREAD_ID)).toBeNull(); + }); + + it('keeps the validation of invalid responses after a transient 503', async () => { + const doGenerate = vi + .fn<() => Promise>() + .mockRejectedValueOnce(apiError(503)) + .mockResolvedValue(textResult('Here are my notes about the conversation.')); + const { store, orchestrator } = setup(doGenerate); + const list = turn('Use the violet theme for the launch page.'); + + await observeMidRun(orchestrator, list); + + expect(doGenerate).toHaveBeenCalledTimes(2); + expect(await store.getActiveObservationLog({ observationScopeId: THREAD_ID })).toEqual([]); + expect(await store.getCursor(THREAD_ID)).toBeNull(); + expect(list.llmVisibleMessages()).toEqual(list.messages()); + }); + + it('records the post-turn observation after a transient 503', async () => { + const doGenerate = vi + .fn<() => Promise>() + .mockRejectedValueOnce(apiError(503)) + .mockResolvedValue(textResult('* IMPORTANT (14:30) Launch page work is open.')); + const { store, orchestrator, tracker } = setup(doGenerate); + const list = turn('Use the violet theme for the launch page.'); + + await orchestrator.saveToMemory(list, runOptions()); + const flushed = tracker.flush(); + await vi.advanceTimersByTimeAsync(BACKOFF_WINDOW_MS); + await flushed; + + expect(doGenerate).toHaveBeenCalledTimes(2); + expect(await store.getActiveObservationLog({ observationScopeId: THREAD_ID })).toMatchObject([ + { text: 'Launch page work is open.' }, + ]); + expect((await store.getCursor(THREAD_ID))?.lastObservedMessageId).toBe( + list.messages().at(-1)?.id, + ); + }); +}); diff --git a/packages/@n8n/agents/src/runtime/memory/observation-log-defaults.ts b/packages/@n8n/agents/src/runtime/memory/observation-log-defaults.ts index 2b27e0290f4..1a030fa3edc 100644 --- a/packages/@n8n/agents/src/runtime/memory/observation-log-defaults.ts +++ b/packages/@n8n/agents/src/runtime/memory/observation-log-defaults.ts @@ -24,6 +24,7 @@ export const DEFAULT_OBSERVATION_LOG_TAIL_LIMIT = 20; export const DEFAULT_OBSERVATION_LOG_REFLECTOR_THRESHOLD_TOKENS = 60_000; export const DEFAULT_OBSERVATION_LOG_RENDER_TOKEN_BUDGET = 67_500; export const DEFAULT_OBSERVATION_LOG_LOCK_TTL_MS = 30_000; +export const DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES = 3; export const DEFAULT_OBSERVATION_LOG_OBSERVER_PROMPT = `You observe a conversation between a user and an agent. Extract only durable facts that the agent needs to continue correctly. The agent can receive your observations after the transcript is removed. @@ -126,6 +127,11 @@ When durable facts exist, return only observation bullets whose lines start with export interface CreateObservationLogObserveFnOptions { observerPrompt?: string; + /** + * Retries for a transient model failure before the call fails. + * Defaults to {@link DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES}. + */ + maxRetries?: number; /** Called with normalized token usage after each observer LLM call. */ onUsage?: (report: MemoryTaskUsageReport) => void | Promise; } @@ -155,6 +161,7 @@ export function createObservationLogObserveFn( model: createModel(model), instructions: options.observerPrompt ?? DEFAULT_OBSERVATION_LOG_OBSERVER_PROMPT, prompt: buildObservationLogObserverPrompt(input), + maxRetries: options.maxRetries ?? DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES, ...buildAiSdkTelemetry(input.telemetry, { functionSuffix: 'memory-observer' }), }); incrementTokenCountFromUsage(input.executionCounter, usage);