From ebcaa7225ac4b7cf282cbabbacf94e768ecfc6c5 Mon Sep 17 00:00:00 2001 From: "n8n-assistant[bot]" <100856346+n8n-assistant[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 10:51:05 +0000 Subject: [PATCH] fix(core): Stop Instance AI follow-up runs from looping when they keep failing before the agent starts (backport to release-candidate/2.32.x) (#34772) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Raúl Gómez Morales --- .../__tests__/instance-ai.service.test.ts | 127 ++++++++++++++++++ .../instance-ai.service.threadPushRef.test.ts | 2 + .../instance-ai/instance-ai.service.ts | 64 +++++++++ 3 files changed, 193 insertions(+) diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts index 12929937ada..d33d6b98c14 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts @@ -660,6 +660,7 @@ type TerminalGuardOrderServiceInternals = { backgroundTasks: { getRunningTasks: Mock }; temporaryWorkflowService: { reapForRun: Mock }; creditService: { claimRunUsage: Mock }; + failedInternalFollowUpStreaks: Map; schedulePlannedTasks: Mock; drainPendingCheckpointReentries: Mock; taskProjector: { syncFromWorkflowLoop: Mock }; @@ -756,6 +757,7 @@ function createTerminalGuardOrderService(): TerminalGuardOrderServiceInternals { service.backgroundTasks = { getRunningTasks: vi.fn(() => []) }; service.temporaryWorkflowService = { reapForRun: vi.fn(async () => []) }; service.creditService = { claimRunUsage: vi.fn(async () => {}) }; + service.failedInternalFollowUpStreaks = new Map(); service.schedulePlannedTasks = vi.fn(async () => {}); service.drainPendingCheckpointReentries = vi.fn(async () => {}); service.preserveHitlOnShutdown = new Set(); @@ -3910,6 +3912,7 @@ describe('InstanceAiService — clearThreadState agent-builder cleanup', () => { runState: { clearThread: Mock }; backgroundTasks: { cancelThread: Mock }; schedulerLocks: Map; + failedInternalFollowUpStreaks: Map; liveness: { clearThreadState: Mock }; domainAccessTrackersByThread: Map; evalCredentialAllowlists: EvalThreadCredentialAllowlistService; @@ -3938,6 +3941,7 @@ describe('InstanceAiService — clearThreadState agent-builder cleanup', () => { service.runState = { clearThread: vi.fn(() => ({ active: undefined, suspended: undefined })) }; service.backgroundTasks = { cancelThread: vi.fn(() => []) }; service.schedulerLocks = new Map(); + service.failedInternalFollowUpStreaks = new Map(); service.liveness = { clearThreadState: vi.fn() }; service.domainAccessTrackersByThread = new Map(); service.evalCredentialAllowlists = new EvalThreadCredentialAllowlistService(); @@ -4065,3 +4069,126 @@ describe('createAgentMemoryOptions', () => { expect(service.creditService.claimRunUsage).toHaveBeenCalledTimes(1); }); }); + +type FollowUpStreakServiceInternals = { + failedInternalFollowUpStreaks: Map; + updateInternalFollowUpFailureStreak: ( + threadId: string, + status: 'completed' | 'cancelled' | 'error' | 'suspended' | undefined, + isInternalFollowUp: boolean, + ) => void; + startInternalFollowUpRun: (user: User, threadId: string, message: string) => Promise; + startExecuteRun: Mock; + defaultTimeZone: string; + runState: { + hasLiveRun: Mock; + startRun: Mock; + getTimeZone: Mock; + }; + logger: { warn: Mock; debug: Mock; error: Mock }; +}; + +function createFollowUpStreakService(): FollowUpStreakServiceInternals { + // Bypass the constructor — we only exercise the follow-up circuit breaker + // and the run-start path it gates. + const service = Object.create( + InstanceAiService.prototype, + ) as unknown as FollowUpStreakServiceInternals; + + service.failedInternalFollowUpStreaks = new Map(); + service.startExecuteRun = vi.fn(); + service.defaultTimeZone = 'UTC'; + service.runState = { + hasLiveRun: vi.fn(() => false), + startRun: vi.fn(() => ({ runId: 'follow-up-run', abortController: new AbortController() })), + getTimeZone: vi.fn(() => undefined), + }; + service.logger = { warn: vi.fn(), debug: vi.fn(), error: vi.fn() }; + + return service; +} + +describe('InstanceAiService — internal follow-up failure streak', () => { + describe('updateInternalFollowUpFailureStreak', () => { + it('counts consecutive errored internal follow-up runs per thread', () => { + const service = createFollowUpStreakService(); + + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + + expect(service.failedInternalFollowUpStreaks.get('thread-a')).toBe(2); + expect(service.failedInternalFollowUpStreaks.get('thread-b')).toBeUndefined(); + }); + + it('does not count errored runs that were not internal follow-ups', () => { + const service = createFollowUpStreakService(); + + service.updateInternalFollowUpFailureStreak('thread-a', 'error', false); + + expect(service.failedInternalFollowUpStreaks.get('thread-a')).toBeUndefined(); + }); + + it('resets the streak when a run completes or suspends', () => { + const service = createFollowUpStreakService(); + + service.failedInternalFollowUpStreaks.set('thread-a', 3); + service.updateInternalFollowUpFailureStreak('thread-a', 'completed', false); + expect(service.failedInternalFollowUpStreaks.get('thread-a')).toBeUndefined(); + + service.failedInternalFollowUpStreaks.set('thread-a', 3); + service.updateInternalFollowUpFailureStreak('thread-a', 'suspended', true); + expect(service.failedInternalFollowUpStreaks.get('thread-a')).toBeUndefined(); + }); + + it('keeps the streak unchanged on cancelled runs and missing terminal status', () => { + const service = createFollowUpStreakService(); + service.failedInternalFollowUpStreaks.set('thread-a', 2); + + service.updateInternalFollowUpFailureStreak('thread-a', 'cancelled', true); + service.updateInternalFollowUpFailureStreak('thread-a', undefined, true); + + expect(service.failedInternalFollowUpStreaks.get('thread-a')).toBe(2); + }); + }); + + describe('startInternalFollowUpRun circuit breaker', () => { + it('starts the follow-up while the streak is below the cap', async () => { + const service = createFollowUpStreakService(); + service.failedInternalFollowUpStreaks.set('thread-a', 2); + + const runId = await service.startInternalFollowUpRun(fakeUser, 'thread-a', 'verify'); + + expect(runId).toBe('follow-up-run'); + expect(service.startExecuteRun).toHaveBeenCalled(); + }); + + it('skips the follow-up once the streak reaches the cap', async () => { + const service = createFollowUpStreakService(); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + + const runId = await service.startInternalFollowUpRun(fakeUser, 'thread-a', 'verify'); + + expect(runId).toBe(''); + expect(service.startExecuteRun).not.toHaveBeenCalled(); + expect(service.logger.warn).toHaveBeenCalledWith( + 'Skipping internal follow-up: consecutive follow-up runs keep failing', + expect.objectContaining({ threadId: 'thread-a', failedStreak: 3 }), + ); + }); + + it('allows follow-ups again after a healthy run resets the streak', async () => { + const service = createFollowUpStreakService(); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'error', true); + service.updateInternalFollowUpFailureStreak('thread-a', 'completed', false); + + const runId = await service.startInternalFollowUpRun(fakeUser, 'thread-a', 'verify'); + + expect(runId).toBe('follow-up-run'); + expect(service.startExecuteRun).toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.threadPushRef.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.threadPushRef.test.ts index 319b53e2141..e5b7a7e1d8e 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.threadPushRef.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.threadPushRef.test.ts @@ -73,6 +73,7 @@ describe('InstanceAiService — threadPushRef lifetime', () => { runState: { clearThread: Mock }; backgroundTasks: { cancelThread: Mock }; schedulerLocks: Map; + failedInternalFollowUpStreaks: Map; liveness: { clearThreadState: Mock }; domainAccessTrackersByThread: Map; evalCredentialAllowlists: EvalThreadCredentialAllowlistService; @@ -100,6 +101,7 @@ describe('InstanceAiService — threadPushRef lifetime', () => { }; service.backgroundTasks = { cancelThread: vi.fn(() => []) }; service.schedulerLocks = new Map(); + service.failedInternalFollowUpStreaks = new Map(); service.liveness = { clearThreadState: vi.fn() }; service.domainAccessTrackersByThread = new Map(); service.evalCredentialAllowlists = new EvalThreadCredentialAllowlistService(); diff --git a/packages/cli/src/modules/instance-ai/instance-ai.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.service.ts index 7fe6da9db74..a4e1e6380ff 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -360,6 +360,14 @@ type RunFinishErrorInfo = { const MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD = 5; +/** + * Circuit breaker for machine-started follow-up runs (verification, synthesize, + * replan, …). A follow-up that dies before the agent can settle its trigger + * (e.g. sandbox setup fails on an exhausted quota) would otherwise be re-armed + * by its own post-run scheduler tick, producing an unbounded error loop. + */ +const MAX_CONSECUTIVE_FAILED_INTERNAL_FOLLOW_UPS = 3; + const TITLE_REFINE_HISTORY_LIMIT = 50; /** Collapse the frontend's typed confirmation union into the flat payload @@ -454,6 +462,14 @@ export class InstanceAiService { /** Per-thread promise chain that serializes schedulePlannedTasks calls. */ private readonly schedulerLocks = new Map>(); + /** + * Consecutive machine-started follow-up runs that errored, per thread. + * Gates `startInternalFollowUpRun` so a follow-up whose run keeps failing + * (its trigger left unsettled) cannot re-arm itself forever; reset by any + * run that completes or suspends, i.e. proves the thread is healthy again. + */ + private readonly failedInternalFollowUpStreaks = new Map(); + /** * Checkpoint re-entries that could not fire when their parent-tagged child * settled (an orchestrator run was live, or other parent siblings were @@ -1397,6 +1413,7 @@ export class InstanceAiService { }); this.schedulerLocks.delete(threadId); + this.failedInternalFollowUpStreaks.delete(threadId); this.domainAccessTrackersByThread.delete(threadId); this.evalCredentialAllowlists.clearThread(threadId); this.threadPushRef.delete(threadId); @@ -2767,6 +2784,29 @@ export class InstanceAiService { ); } + /** + * Feed the follow-up circuit breaker from a run's terminal status. Errored + * machine-started follow-ups (`isInternalFollowUp`) extend the streak; any + * run that completes or suspends proves the thread executes again and + * resets it. Cancelled runs carry no signal either way. + */ + private updateInternalFollowUpFailureStreak( + threadId: string, + status: MessageTraceFinalization['status'] | undefined, + isInternalFollowUp: boolean, + ): void { + if (status === 'completed' || status === 'suspended') { + this.failedInternalFollowUpStreaks.delete(threadId); + return; + } + if (status === 'error' && isInternalFollowUp) { + this.failedInternalFollowUpStreaks.set( + threadId, + (this.failedInternalFollowUpStreaks.get(threadId) ?? 0) + 1, + ); + } + } + private async startInternalFollowUpRun( user: User, threadId: string, @@ -2782,6 +2822,16 @@ export class InstanceAiService { return ''; } + const failedStreak = this.failedInternalFollowUpStreaks.get(threadId) ?? 0; + if (failedStreak >= MAX_CONSECUTIVE_FAILED_INTERNAL_FOLLOW_UPS) { + this.logger.warn('Skipping internal follow-up: consecutive follow-up runs keep failing', { + threadId, + failedStreak, + resumeReason: resumeReasonOverride, + }); + return ''; + } + const { runId, abortController } = this.runState.startRun({ threadId, user, @@ -3883,6 +3933,13 @@ export class InstanceAiService { this.liveness.consumeRunTimeout(runId); } } + // Must precede the reschedule below: the next tick consults the streak + // before re-arming another follow-up for the same unsettled trigger. + this.updateInternalFollowUpFailureStreak( + threadId, + messageTraceFinalization?.status, + resumeReason !== undefined, + ); // Post-run planned-task wiring (only when the run is actually ending, // not when it merely suspended for HITL): // 1. Checkpoint deadlock fallback — if this run was a checkpoint @@ -5029,6 +5086,13 @@ export class InstanceAiService { this.liveness.consumeRunTimeout(opts.runId); } } + // Resumed runs are user-driven, so they never extend the failure + // streak — but a healthy one resets it before the reschedule below. + this.updateInternalFollowUpFailureStreak( + opts.threadId, + messageTraceFinalization?.status, + false, + ); // Post-run planned-task wiring — mirror the executeRun finally. // Resumed ordinary-chat runs also need to drive the scheduler in case // a background task settled while they were active or suspended and