From 3d7164d7d273bf1d18a58aa03377581b571ce62d Mon Sep 17 00:00:00 2001 From: Luka Zivkovic Date: Fri, 16 Oct 2026 19:38:54 +0200 Subject: [PATCH] refactor(core): Isolate agent eval run settlement --- packages/@n8n/api-types/src/index.ts | 1 + .../src/schemas/instance-ai.schema.ts | 4 +- .../agent-evals/agent-eval-runner.service.ts | 239 ++++++++++-------- 3 files changed, 143 insertions(+), 101 deletions(-) diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index c30c3a6e3aa..0b745d664e1 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/packages/@n8n/api-types/src/index.ts @@ -502,6 +502,7 @@ export { InstanceAiSendMessageRequest, InstanceAiEvalExecutionRequest, InstanceAiEvalAgentExecutionRequest, + INSTANCE_AI_EVAL_AGENT_MAX_TIMEOUT_MS, InstanceAiEvalCredentialAllowlistRequest, InstanceAiEvalCredentialTestTargetRequest, INSTANCE_AI_MEMORY_TASK_WAIT_TIMEOUT_MS, diff --git a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts index 6a5441040c3..622fad95057 100644 --- a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts +++ b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts @@ -3013,6 +3013,8 @@ export interface InstanceAiEvalAgentExecutionResult { llmUsage?: InstanceAiEvalLlmUsage[]; } +export const INSTANCE_AI_EVAL_AGENT_MAX_TIMEOUT_MS = 3_600_000; + export class InstanceAiEvalAgentExecutionRequest extends Z.class({ /** Project the agent lives in (agent routes are project-scoped). */ projectId: z.string().min(1), @@ -3021,7 +3023,7 @@ export class InstanceAiEvalAgentExecutionRequest extends Z.class({ * Overall run budget. Server default applies when omitted. Shares the workflow * variant's ceiling — the old 900_000 cap truncated a `complex` case's budget. */ - timeoutMs: z.number().int().min(30_000).max(3_600_000).optional(), + timeoutMs: z.number().int().min(30_000).max(INSTANCE_AI_EVAL_AGENT_MAX_TIMEOUT_MS).optional(), }) {} export class InstanceAiEvalCredentialAllowlistRequest extends Z.class({ diff --git a/packages/cli/src/modules/agent-evals/agent-eval-runner.service.ts b/packages/cli/src/modules/agent-evals/agent-eval-runner.service.ts index 28becdc504b..92576e33824 100644 --- a/packages/cli/src/modules/agent-evals/agent-eval-runner.service.ts +++ b/packages/cli/src/modules/agent-evals/agent-eval-runner.service.ts @@ -1,5 +1,9 @@ import { setTimeout as sleep } from 'node:timers/promises'; -import type { AgentEvalVerdict, AgentEvalRunSummary } from '@n8n/api-types'; +import { + INSTANCE_AI_EVAL_AGENT_MAX_TIMEOUT_MS, + type AgentEvalVerdict, + type AgentEvalRunSummary, +} from '@n8n/api-types'; import { Logger, ModuleRegistry } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; import type { AgentEvalDataset, AgentEvalResult, User } from '@n8n/db'; @@ -48,8 +52,8 @@ const MAX_CASES = 500; // `setTimeout` holds its delay in a 32-bit int and silently coerces anything // larger to 1ms, so a deadline past ~24.8 days must be clamped, not passed on. const MAX_TIMER_DELAY_MS = 2_147_483_647; -// Reruns allow a one-hour execution budget. Leave five minutes for judging. -const RESULT_SETTLEMENT_TIMEOUT_MS = 65 * 60_000; +// Cover the maximum agent execution budget, plus five minutes for judging. +const RESULT_SETTLEMENT_TIMEOUT_MS = INSTANCE_AI_EVAL_AGENT_MAX_TIMEOUT_MS + 5 * 60_000; const RESULT_SETTLEMENT_POLL_MS = 1_000; /** A dataset row resolved into a runnable case via the dataset's column mapping. */ @@ -67,6 +71,12 @@ interface CaseUsage { outputTokens: number; } +interface RunDeadline { + hasExpired: () => boolean; + clear: () => void; + deadlineMinutes: number; +} + /** * Executes an agent-eval dataset against a real agent and persists per-case * results, then aggregates run-level status. @@ -433,7 +443,6 @@ export class AgentEvalRunnerService { dispatchFailures.push(message); unfinishedOriginals.set(seeded[index].id, message); } - let wasCancelled = cancelObserved; if (dispatchFailures.length > 0) { this.logger.error( @@ -442,112 +451,142 @@ export class AgentEvalRunnerService { ); } - // A settled sibling can be rerun while the original pool is active. - // Fence the terminal write against those attempts, including other mains. - const settlementDeadline = Date.now() + RESULT_SETTLEMENT_TIMEOUT_MS; - while (true) { - for (const [resultId, message] of unfinishedOriginals) { - try { - await this.resultRepository.markUnfinishedAsError( - resultId, - originalAttempts.get(resultId), - message, - ); - unfinishedOriginals.delete(resultId); - } catch (error) { - this.logger.warn(`[AgentEvalRunner] Could not settle original case ${resultId}`, { - error, - }); - } - } - if (!wasCancelled) { - try { - wasCancelled = await this.runRepository.isCancellationRequested(runId); - } catch (error) { - dispatchFailures.push(error instanceof Error ? error.message : String(error)); - } - } - const counts = await this.resultRepository.countByStatus(runId); - const metrics: IDataObject = { ...toSummaryCounts(counts), usage: { ...totalUsage } }; - if (Date.now() >= settlementDeadline) { - if (wasCancelled) { - await this.runRepository.markAsCancelled(runId, metrics); - } else { - await this.runRepository.markAsError( - runId, - 'result_settlement_timeout', - { - message: 'A result did not settle after the original batch finished.', - }, - metrics, - ); - } - break; - } - if (counts.new > 0 || counts.running > 0) { - await sleep(RESULT_SETTLEMENT_POLL_MS); - continue; - } - // Dispatch failures use the same claim fence as successful completion. - const finalCounts = counts; - - let settled; - // Every branch records the tally under `metrics`, so pollers never find a - // settled run whose counts are missing or filed elsewhere. - if (wasCancelled) { - // A user stop outranks the deadline: it's intent, not a failure. - settled = await this.runRepository.markAsCancelled(runId, metrics, finalCounts); - } else if (deadline.hasExpired() && stoppedCases > 0) { - settled = await this.runRepository.markAsError( - runId, - 'timeout', - { - message: `Run exceeded its ${deadline.deadlineMinutes}-minute deadline; ${stoppedCases} case(s) were not started.`, - }, - metrics, - finalCounts, - ); - } else if (dispatchFailures.length > 0) { - settled = await this.runRepository.markAsError( - runId, - 'case_dispatch_failed', - { - message: `${dispatchFailures.length} failure(s) occurred outside case execution.`, - errors: dispatchFailures, - }, - metrics, - finalCounts, - ); - } else { - if (deadline.hasExpired()) { - // Overran, but nothing was left to skip. Logged for anyone tuning the - // limit, since the run itself still succeeded. - this.logger.debug( - `[AgentEvalRunner] Run ${runId} overran its ${deadline.deadlineMinutes}-minute deadline, but every case had finished`, - ); - } - settled = await this.runRepository.markAsCompleted(runId, metrics, finalCounts); - } - if (settled) break; - const current = await this.runRepository.findById(runId); - if (!current || current.status !== 'running') break; - await sleep(RESULT_SETTLEMENT_POLL_MS); - } + await this.settleRun({ + runId, + unfinishedOriginals, + originalAttempts, + dispatchFailures, + cancelObserved, + totalUsage, + deadline, + stoppedCases, + }); } catch (error) { const message = error instanceof Error ? error.message : String(error); await this.failRun(runId, message); } } + private async settleRun(ctx: { + runId: string; + unfinishedOriginals: Map; + originalAttempts: Map; + dispatchFailures: string[]; + cancelObserved: boolean; + totalUsage: CaseUsage; + deadline: RunDeadline; + stoppedCases: number; + }): Promise { + const { + runId, + unfinishedOriginals, + originalAttempts, + dispatchFailures, + cancelObserved, + totalUsage, + deadline, + stoppedCases, + } = ctx; + let wasCancelled = cancelObserved; + + // A settled sibling can be rerun while the original pool is active. + // Fence the terminal write against those attempts, including other mains. + const settlementDeadline = Date.now() + RESULT_SETTLEMENT_TIMEOUT_MS; + while (true) { + for (const [resultId, message] of unfinishedOriginals) { + try { + await this.resultRepository.markUnfinishedAsError( + resultId, + originalAttempts.get(resultId), + message, + ); + unfinishedOriginals.delete(resultId); + } catch (error) { + this.logger.warn(`[AgentEvalRunner] Could not settle original case ${resultId}`, { + error, + }); + } + } + if (!wasCancelled) { + try { + wasCancelled = await this.runRepository.isCancellationRequested(runId); + } catch (error) { + dispatchFailures.push(error instanceof Error ? error.message : String(error)); + } + } + const counts = await this.resultRepository.countByStatus(runId); + const metrics: IDataObject = { ...toSummaryCounts(counts), usage: { ...totalUsage } }; + if (Date.now() >= settlementDeadline) { + if (wasCancelled) { + await this.runRepository.markAsCancelled(runId, metrics); + } else { + await this.runRepository.markAsError( + runId, + 'result_settlement_timeout', + { + message: 'A result did not settle after the original batch finished.', + }, + metrics, + ); + } + break; + } + if (counts.new > 0 || counts.running > 0) { + await sleep(RESULT_SETTLEMENT_POLL_MS); + continue; + } + // Dispatch failures use the same claim fence as successful completion. + const finalCounts = counts; + + let settled; + // Every branch records the tally under `metrics`, so pollers never find a + // settled run whose counts are missing or filed elsewhere. + if (wasCancelled) { + // A user stop outranks the deadline: it's intent, not a failure. + settled = await this.runRepository.markAsCancelled(runId, metrics, finalCounts); + } else if (deadline.hasExpired() && stoppedCases > 0) { + settled = await this.runRepository.markAsError( + runId, + 'timeout', + { + message: `Run exceeded its ${deadline.deadlineMinutes}-minute deadline; ${stoppedCases} case(s) were not started.`, + }, + metrics, + finalCounts, + ); + } else if (dispatchFailures.length > 0) { + settled = await this.runRepository.markAsError( + runId, + 'case_dispatch_failed', + { + message: `${dispatchFailures.length} failure(s) occurred outside case execution.`, + errors: dispatchFailures, + }, + metrics, + finalCounts, + ); + } else { + if (deadline.hasExpired()) { + // Overran, but nothing was left to skip. Logged for anyone tuning the + // limit, since the run itself still succeeded. + this.logger.debug( + `[AgentEvalRunner] Run ${runId} overran its ${deadline.deadlineMinutes}-minute deadline, but every case had finished`, + ); + } + settled = await this.runRepository.markAsCompleted(runId, metrics, finalCounts); + } + if (settled) break; + const current = await this.runRepository.findById(runId); + if (current?.status !== 'running') break; + await sleep(RESULT_SETTLEMENT_POLL_MS); + } + } + /** * Arm the run-level deadline; expiring trips the run's abort. Non-positive * disables. Bounds when the last case may *start*, not when the run ends. */ - private startRunDeadline(abort: AbortController): { - hasExpired: () => boolean; - clear: () => void; - deadlineMinutes: number; - } { + private startRunDeadline(abort: AbortController): RunDeadline { const deadlineMinutes = this.globalConfig.evaluation.agentEvalsRunTimeoutMinutes; if (deadlineMinutes <= 0) { return { hasExpired: () => false, clear: () => undefined, deadlineMinutes };