diff --git a/.agents/skills/create-instance-ai-eval/running-evals.md b/.agents/skills/create-instance-ai-eval/running-evals.md index ecdb47df936..b07d21a0595 100644 --- a/.agents/skills/create-instance-ai-eval/running-evals.md +++ b/.agents/skills/create-instance-ai-eval/running-evals.md @@ -41,7 +41,7 @@ instance's Instance AI config and the README's environment-variables section. | Mode | How | Produces | |---|---|---| -| **Direct loop** | no `LANGSMITH_API_KEY` | `eval-results.json` + HTML report locally | +| **Direct driver** | no `LANGSMITH_API_KEY` | `eval-results.json` + HTML report locally — same pipeline and row order as the LangSmith driver (TRUST-261), row concurrency follows `--concurrency` | | **LangSmith** | `LANGSMITH_API_KEY` set | also records an experiment and auto-compares against the baseline | | **Prebuilt** | `--prebuilt-workflows ` | skips the build; verifies existing workflows (score MCP/hand-built cohorts on the same verifier) | diff --git a/packages/@n8n/instance-ai/.gitignore b/packages/@n8n/instance-ai/.gitignore index 44923d7e8be..9a59c8dbb7a 100644 --- a/packages/@n8n/instance-ai/.gitignore +++ b/packages/@n8n/instance-ai/.gitignore @@ -1,3 +1,4 @@ .output/ eval-pr-comment.md eval-results.json +eval-rows.jsonl diff --git a/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md b/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md new file mode 100644 index 00000000000..230217d9e07 --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md @@ -0,0 +1,74 @@ +# Eval harness architecture + +One pipeline, two thin drivers (TRUST-261). Every phase of a run lives behind a +named seam in `evaluations/run/`; `cli/index.ts` is a ~150-line composition +root, and the rest of `cli/` is the sibling entrypoints (`pairwise`, +`compare-pairwise`, `report`, `build-mcp-manifest`, `langtracer-push`) plus +their helpers (`args.ts`, `mcp-builder.ts`). + +``` +cli/index.ts (composition root) + → run/case-selection.ts narrow loaded cases (prebuilt coverage, MCP limits) + → run/lane-setup.ts one authenticated lane per --base-url (+ final cleanup) + → driver: + run/langsmith-driver.ts LANGSMITH_API_KEY set — evaluate(), experiments, + per-run pass metrics, gate/baseline comparison + run/direct-driver.ts keyless — the LangTracer dispatcher's mode; same + rows, same pipeline, same local artifacts — just + no LangSmith experiments/feedback/comparison + → run/eval-session.ts shared assembly: lane wrappers (tracing hook), + work-stealing allocator, orchestrator, pipeline, + side-band resolution, end-of-run artifact drain + → run/build-orchestrator.ts getOrBuild: per-(iteration, case) build + cache, transient retry across lanes, + side-band capture (transcript, expectation + verdicts incl. artifactContext, run debug) + → run/case-pipeline.ts runRow: sentinel / build-fail / agent / + workflow dispatch, transient retry, seed- + table serialization, framework_issue + guard, eager per-build cleanup + → run/rows.ts single source of row set + order for BOTH drivers + → run/reshape.ts rows → WorkflowTestCaseResult[][] (driver-agnostic) + → run/aggregator.ts pass@k / pass^k aggregation + → run/persist.ts always-write eval-results.json / eval-pr-comment.md, + row journal (eval-rows.jsonl) + crash recovery + → comparison/* baseline comparison + the absolute gate + → run/reporters.ts console paths, HTML reports, terminal summary, noise advisory +``` + +The tracing hook is the only driver-specific piece of the session: the +LangSmith driver wraps lane functions with `traceable()` (span names +`workflow_build` / `scenario_execution` / `agent_scenario_execution` — the +analytics pipeline reads them), the direct driver passes identity. + +## Where to add things + +| You want to… | Touch exactly | +|---|---| +| Add a grader / judge / check wiring | `run/case-pipeline.ts` (row-side) or `run/build-orchestrator.ts` (build-side capture) — it then runs in CI **and** dispatcher mode | +| Add a tier | nothing — tiers are free-form strings in a case's `datasets` field; only a tier that should assert the absolute green bar registers in `run/tiers.ts` | +| Add a case source | `data/source.ts` (`--source` dispatch) | +| Add a driver | compose `createEvalSession()` + feed rows from `run/rows.ts` | +| Change persisted output | `run/persist.ts` — and extend `__tests__/eval-results-dispatcher-contract.test.ts` | + +## External contracts (do not drift silently) + +- **`eval-results.json`** is ingested by the LangTracer dispatcher, which runs + this CLI keyless per case. The exact field set is pinned by + `__tests__/eval-results-dispatcher-contract.test.ts`. +- **`eval-pr-comment.md`** is posted verbatim by CI; the comment is found by its + `### Instance AI Workflow Eval` prefix. +- **LangSmith feedback keys** (`scenario_pass`, `failure_category`, + `evals.workflows.*`, `pass_at_k`, `pass_hat_k`) and the traced span names + feed the LangSmith→BigQuery analytics. +- **`pnpm eval:*` script names** are invoked by CI workflows and + `run-eval-lanes.sh`. + +## Crash recovery + +Both drivers journal every completed row to `/eval-rows.jsonl` +(`run/persist.ts`). If the run dies, `runEvalAndPersist` reshapes the journal's +*complete* iterations into `eval-results.json` — incomplete iterations are +dropped, never stubbed, so a crash artifact cannot fabricate failures for rows +that never ran. This is also the merge seam for sharding runs (TRUST-152): +concatenate journals, reshape once. diff --git a/packages/@n8n/instance-ai/evaluations/README.md b/packages/@n8n/instance-ai/evaluations/README.md index 3ebe9601df0..6e1000006c8 100644 --- a/packages/@n8n/instance-ai/evaluations/README.md +++ b/packages/@n8n/instance-ai/evaluations/README.md @@ -1,5 +1,7 @@ # Workflow evaluation framework +> Module layout, extension points and external contracts: [ARCHITECTURE.md](./ARCHITECTURE.md). + Tests whether workflows built by Instance AI actually work by executing them with LLM-generated mock HTTP responses. No real credentials or external services are involved. Five harnesses live here: diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts index c5b1407658c..454061149e9 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts @@ -1,5 +1,7 @@ +import { verifyBuildExpectations } from '../build-expectations/verifier'; import type { CliArgs } from '../cli/args'; import type { N8nClient } from '../clients/n8n-client'; +import { resolveArtifactContext } from '../harness/artifacts/artifact-context'; import type { EvalLogger } from '../harness/logger'; import { runWorkflowChecks, type BuildResult } from '../harness/runner'; import { @@ -18,13 +20,21 @@ import type { WorkflowTestCase } from '../types'; vi.mock('../harness/runner', async (importOriginal) => { const actual = await importOriginal(); - return { ...actual, runWorkflowChecks: vi.fn().mockResolvedValue([]) }; + return { + ...actual, + runWorkflowChecks: vi.fn().mockResolvedValue([]), + fetchAgentScenarioContext: vi.fn().mockResolvedValue('AGENT CONTEXT'), + }; }); vi.mock('../harness/capture-run-debug', () => ({ captureThreadRunDebug: vi.fn().mockResolvedValue([]), })); +vi.mock('../harness/artifacts/artifact-context', () => ({ + resolveArtifactContext: vi.fn().mockResolvedValue('RESOLVED ARTIFACTS'), +})); + vi.mock('../build-expectations/verifier', async (importOriginal) => { const actual = await importOriginal(); return { @@ -273,3 +283,59 @@ describe('createBuildOrchestrator', () => { ]); }); }); + +describe('expectation judging context', () => { + it('threads the rendered agent artifact into the expectation judge', async () => { + const tracedBuild = vi.fn().mockResolvedValue( + okBuild({ + threadId: 'thread-9', + transcript: [] as never, + artifactRefs: [{ type: 'agent', id: 'agent-1' }] as never, + }), + ); + const deps = makeDeps([makeLane(1, tracedBuild)], { + testCaseByFileSlug: new Map([ + ['case-a', baseCase({ outcomeExpectations: ['the agent has a Slack tool'] })], + ]), + }); + const orchestrator = createBuildOrchestrator(deps); + + await orchestrator.getOrBuild(0, 'case-a'); + await deps.buildExpectationsByKey.get('0:case-a'); + + expect(vi.mocked(resolveArtifactContext)).toHaveBeenCalledWith( + expect.objectContaining({ artifactRefs: [{ type: 'agent', id: 'agent-1' }] }), + ); + expect(vi.mocked(verifyBuildExpectations)).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ artifactContext: 'RESOLVED ARTIFACTS' }), + ); + }); + + it('resolves non-agent artifacts (config-eval) for the judge too — not just agent refs', async () => { + const tracedBuild = vi.fn().mockResolvedValue( + okBuild({ + threadId: 'thread-9', + transcript: [] as never, + artifactRefs: [{ type: 'config-eval', id: 'ce-1' }] as never, + }), + ); + const deps = makeDeps([makeLane(1, tracedBuild)], { + testCaseByFileSlug: new Map([ + ['case-a', baseCase({ outcomeExpectations: ['a config eval exists'] })], + ]), + }); + const orchestrator = createBuildOrchestrator(deps); + + await orchestrator.getOrBuild(0, 'case-a'); + await deps.buildExpectationsByKey.get('0:case-a'); + + expect(vi.mocked(resolveArtifactContext)).toHaveBeenCalledWith( + expect.objectContaining({ artifactRefs: [{ type: 'config-eval', id: 'ce-1' }] }), + ); + expect(vi.mocked(verifyBuildExpectations)).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ artifactContext: 'RESOLVED ARTIFACTS' }), + ); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/case-pipeline.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/case-pipeline.test.ts index 0fc4534c7d5..8e1f96028a4 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/case-pipeline.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/case-pipeline.test.ts @@ -316,3 +316,141 @@ describe('createCasePipeline', () => { expect(lane.tracedExecuteAgent).toHaveBeenCalledTimes(1); }); }); + +function deferred(): { promise: Promise; resolve: (v: unknown) => void } { + let resolve!: (v: unknown) => void; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; +} + +describe('seed-table scenarios (TRUST-311 parity)', () => { + function seededCase(names: string[]): WorkflowTestCase { + return { + ...scenarioCase(names), + executionScenarios: names.map((name) => ({ + name, + description: 'd', + dataSetup: 's', + successCriteria: 'c', + seedDataTables: [ + { + id: 'seed-table-1', + name: 'Jobs', + columns: [{ name: 'id', type: 'string' as const }], + rows: [{ id: 'row_001' }], + }, + ], + })), + }; + } + + it('merges the authored scenario and passes the build seed context to execution', async () => { + const lane = makeLane(); + vi.mocked(lane.tracedExecute).mockResolvedValue({ + success: true, + score: 1, + reasoning: 'ok', + } as never); + const orchestrator = makeOrchestrator({ + build: okBuild({ threadId: 'thread-1', seededScenarioTableIdsByName: { Jobs: 'dt-real-1' } }), + lane, + buildDurationMs: 1, + }); + const pipeline = createCasePipeline( + makeDeps(orchestrator, { + testCaseByFileSlug: new Map([['case-a', seededCase(['happy-path'])]]), + }), + ); + + await pipeline.runRow(rowInputs('happy-path')); + + expect(lane.tracedExecute).toHaveBeenCalledWith( + expect.objectContaining({ + scenario: expect.objectContaining({ + seedDataTables: [expect.objectContaining({ name: 'Jobs' })], + }) as unknown, + seedContext: { threadId: 'thread-1', tableIdsByName: { Jobs: 'dt-real-1' } }, + }), + ); + }); + + it('refuses to run a seeded scenario when the build carries no seeded-table mapping', async () => { + const lane = makeLane(); + // MCP/prebuilt-shaped build: success, but no thread or table mapping — + // running would grade the workflow against empty tables. + const orchestrator = makeOrchestrator({ build: okBuild(), lane, buildDurationMs: 1 }); + const pipeline = createCasePipeline( + makeDeps(orchestrator, { + testCaseByFileSlug: new Map([['case-a', seededCase(['happy-path'])]]), + }), + ); + + const output = await pipeline.runRow(rowInputs('happy-path')); + + expect(lane.tracedExecute).not.toHaveBeenCalled(); + expect(output).toMatchObject({ + passed: false, + failureCategory: 'framework_issue', + reasoning: expect.stringContaining('no seeded-table mapping') as unknown, + }); + }); + + it('serializes rows of a seeded case so table reseeding cannot interleave', async () => { + const lane = makeLane(); + const started: string[] = []; + const first = deferred(); + const second = deferred(); + vi.mocked(lane.tracedExecute).mockImplementation(((execArgs: { + scenario: { name: string }; + }) => { + started.push(execArgs.scenario.name); + return (started.length === 1 ? first.promise : second.promise) as never; + }) as never); + const orchestrator = makeOrchestrator({ + build: okBuild({ threadId: 'thread-1', seededScenarioTableIdsByName: { Jobs: 'dt-real-1' } }), + lane, + buildDurationMs: 1, + }); + const pipeline = createCasePipeline( + makeDeps(orchestrator, { + testCaseByFileSlug: new Map([['case-a', seededCase(['s1', 's2'])]]), + }), + ); + + const rows = Promise.all([pipeline.runRow(rowInputs('s1')), pipeline.runRow(rowInputs('s2'))]); + await vi.waitFor(() => expect(started).toHaveLength(1)); + for (let i = 0; i < 5; i++) await Promise.resolve(); + // The second seeded row must not start while the first still runs. + expect(started).toEqual(['s1']); + + first.resolve({ success: true, score: 1, reasoning: 'ok' }); + await vi.waitFor(() => expect(started).toHaveLength(2)); + second.resolve({ success: true, score: 1, reasoning: 'ok' }); + await rows; + }); + + it('does not serialize rows of an unseeded case', async () => { + const lane = makeLane(); + const started: string[] = []; + const first = deferred(); + const second = deferred(); + vi.mocked(lane.tracedExecute).mockImplementation(((execArgs: { + scenario: { name: string }; + }) => { + started.push(execArgs.scenario.name); + return (started.length === 1 ? first.promise : second.promise) as never; + }) as never); + const orchestrator = makeOrchestrator({ build: okBuild(), lane, buildDurationMs: 1 }); + const pipeline = createCasePipeline( + makeDeps(orchestrator, { + testCaseByFileSlug: new Map([['case-a', scenarioCase(['s1', 's2'])]]), + }), + ); + + const rows = Promise.all([pipeline.runRow(rowInputs('s1')), pipeline.runRow(rowInputs('s2'))]); + await vi.waitFor(() => expect(started).toHaveLength(2)); + first.resolve({ success: true, score: 1, reasoning: 'ok' }); + second.resolve({ success: true, score: 1, reasoning: 'ok' }); + await rows; + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/data-discovery.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/data-discovery.test.ts index 8e139fb98f3..28ccdf94dd5 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/data-discovery.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/data-discovery.test.ts @@ -7,7 +7,7 @@ * runner. */ -import { loadDiscoveryTestCasesWithFiles } from '../data/discovery'; +import { discoveryTestCaseSchema, loadDiscoveryTestCasesWithFiles } from '../data/discovery'; import { runExpectedToolsInvokedCheck } from '../discovery/expected-tools-invoked'; describe('loadDiscoveryTestCasesWithFiles', () => { @@ -69,3 +69,40 @@ describe('loadDiscoveryTestCasesWithFiles', () => { expect(negative.length).toBeGreaterThan(0); }); }); + +describe('discoveryTestCaseSchema', () => { + const valid = { + id: 'my-scenario', + userMessage: 'do the thing', + expectedToolInvocations: { anyOf: ['build-workflow'] }, + }; + + it('accepts a minimal valid case', () => { + expect(discoveryTestCaseSchema.safeParse(valid).success).toBe(true); + }); + + it('rejects a typo-d key instead of passing vacuously', () => { + const typo = { ...valid, expectedToolInvocation: { anyOf: ['x'] } }; + expect(discoveryTestCaseSchema.safeParse(typo).success).toBe(false); + }); + + it('rejects empty expectations — a case must assert something', () => { + const empty = { ...valid, expectedToolInvocations: {} }; + expect(discoveryTestCaseSchema.safeParse(empty).success).toBe(false); + }); + + it('rejects an empty expectation list — dead config must fail at load time', () => { + const emptyList = { ...valid, expectedToolInvocations: { allOfToolCalls: [] } }; + expect(discoveryTestCaseSchema.safeParse(emptyList).success).toBe(false); + }); + + it.each([ + ['connected with capabilities', { status: 'connected', capabilities: ['screenshot'] }, true], + ['connected without capabilities', { status: 'connected' }, false], + ['disabled', { status: 'disabled' }, true], + ['an unknown status', { status: 'on-fire' }, false], + ])('validates instanceState.localGateway strictly: %s', (_name, localGateway, ok) => { + const withGateway = { ...valid, instanceState: { localGateway } }; + expect(discoveryTestCaseSchema.safeParse(withGateway).success).toBe(ok); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/dataset-sync.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/dataset-sync.test.ts index 14abbdd7064..75ab538b639 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/dataset-sync.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/dataset-sync.test.ts @@ -5,7 +5,12 @@ import type { Mock } from 'vitest'; import type { WorkflowTestCaseWithFile } from '../data/workflows'; import type { EvalLogger } from '../harness/logger'; -import { ARCHIVED_SPLIT, BUILD_ONLY_SCENARIO_NAME, syncDataset } from '../langsmith/dataset-sync'; +import { + ARCHIVED_SPLIT, + BUILD_ONLY_SCENARIO_NAME, + syncDataset, + ensureExamplesVisible, +} from '../langsmith/dataset-sync'; function scenarioFixture(testCaseFile: string, scenarioName: string): WorkflowTestCaseWithFile { return { @@ -294,3 +299,56 @@ describe('syncDataset', () => { expect(second.createExamples).not.toHaveBeenCalled(); }); }); + +describe('ensureExamplesVisible', () => { + const silent = { + info: () => {}, + verbose: () => {}, + success: () => {}, + warn: vi.fn(), + error: () => {}, + isVerbose: false, + }; + const caseWithOneScenario = { + testCase: { + conversation: [{ role: 'user' as const, text: 'build it' }], + complexity: 'simple' as const, + tags: [], + datasets: ['full'], + executionScenarios: [{ name: 's1', description: 'd', dataSetup: 's', successCriteria: 'c' }], + }, + fileSlug: 'case-a', + }; + + function clientListing(counts: number[]): { listExamples: ReturnType } { + let call = 0; + return { + listExamples: vi.fn().mockImplementation(() => { + const n = counts[Math.min(call++, counts.length - 1)]; + return (async function* () { + await Promise.resolve(); + for (let i = 0; i < n; i++) yield { id: String(i) }; + })(); + }), + }; + } + + it('passes once the split-scoped count covers the synced rows', async () => { + const client = clientListing([0, 1]); + await ensureExamplesVisible(client as never, 'ds', [caseWithOneScenario], silent as never, { + baseDelayMs: 1, + }); + expect(client.listExamples).toHaveBeenCalledWith({ datasetName: 'ds', splits: ['case-a'] }); + expect(client.listExamples).toHaveBeenCalledTimes(2); + }); + + it('throws loudly instead of running a partial experiment', async () => { + const client = clientListing([0]); + await expect( + ensureExamplesVisible(client as never, 'ds', [caseWithOneScenario], silent as never, { + attempts: 2, + baseDelayMs: 1, + }), + ).rejects.toThrow(/0\/1 synced example/); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/row-sink.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/row-sink.test.ts new file mode 100644 index 00000000000..cb283224a8c --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/__tests__/row-sink.test.ts @@ -0,0 +1,150 @@ +import { appendFileSync, mkdtempSync, readFileSync } from 'fs'; +import { jsonParse } from 'n8n-workflow'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import type { WorkflowTestCaseWithFile } from '../data/workflows'; +import type { EvalLogger } from '../harness/logger'; +import type { ScenarioRowInputs } from '../run/case-pipeline'; +import { createRowSink, runEvalAndPersist } from '../run/persist'; +import type { TargetOutput } from '../run/reshape'; +import type { WorkflowTestCase } from '../types'; + +// The row sink is the crash-recovery journal both drivers feed: one JSON line +// per completed row. On the crash path runEvalAndPersist reshapes COMPLETE +// iterations back into eval-results.json — never-run rows must not surface as +// fabricated failures in the artifact. + +const silentLogger: EvalLogger = { + info: () => {}, + verbose: () => {}, + success: () => {}, + warn: () => {}, + error: () => {}, + isVerbose: false, +}; + +const testCase: WorkflowTestCase = { + conversation: [{ role: 'user', text: 'build it' }], + complexity: 'simple', + tags: [], + datasets: ['full'], + executionScenarios: [ + { name: 'happy-path', description: 'd', dataSetup: 's', successCriteria: 'c' }, + ], +}; +const testCasesWithFiles: WorkflowTestCaseWithFile[] = [{ testCase, fileSlug: 'case-a' }]; + +function rowInputs(iteration: number): ScenarioRowInputs { + return { + testCaseFile: 'case-a', + scenarioName: 'happy-path', + scenarioDescription: 'd', + dataSetup: 's', + successCriteria: 'c', + _iteration: iteration, + }; +} + +function rowOutputs(passed: boolean): TargetOutput { + return { + buildSuccess: true, + workflowId: 'wf-1', + passed, + score: passed ? 1 : 0, + reasoning: passed ? 'ok' : 'nope', + execErrors: [], + buildDurationMs: 1, + execDurationMs: 1, + nodeCount: 0, + }; +} + +describe('createRowSink', () => { + it('round-trips appended rows and truncates the previous run', () => { + const dir = mkdtempSync(join(tmpdir(), 'row-sink-')); + const first = createRowSink(dir); + first.append({ run: { inputs: rowInputs(0), outputs: rowOutputs(true) } }); + expect(first.readRows()).toHaveLength(1); + + // A new sink for the same dir starts empty — recovery never mixes runs. + const second = createRowSink(dir); + expect(second.readRows()).toHaveLength(0); + }); + + it('tolerates a torn final line from a hard crash', () => { + const dir = mkdtempSync(join(tmpdir(), 'row-sink-')); + const sink = createRowSink(dir); + sink.append({ run: { inputs: rowInputs(0), outputs: rowOutputs(true) } }); + appendFileSync(sink.path, '{"run":{"inputs":{"trunc'); + + expect(sink.readRows()).toHaveLength(1); + }); +}); + +describe('runEvalAndPersist crash recovery from the sink', () => { + interface WrittenReport { + testCases: Array<{ + testCaseFile?: string; + scenarios: Array<{ runs: unknown[] }>; + buildExpectationResultsPerRun?: unknown[]; + }>; + } + + async function crashWith( + rows: Array<{ iteration: number }>, + outputs: TargetOutput = rowOutputs(true), + ): Promise { + const dir = mkdtempSync(join(tmpdir(), 'row-sink-recovery-')); + const rowSink = createRowSink(dir); + await expect( + runEvalAndPersist( + { + logger: silentLogger, + outputDir: dir, + startTime: 1, + iterations: 3, + tier: undefined, + commitSha: undefined, + rerun: undefined, + mcpBuildSpend: [], + rowSink, + testCasesWithFiles, + }, + async () => { + for (const { iteration } of rows) { + rowSink.append({ run: { inputs: rowInputs(iteration), outputs } }); + } + return await Promise.reject(new Error('lane meltdown')); + }, + ), + ).rejects.toThrow('lane meltdown'); + return jsonParse(readFileSync(join(dir, 'eval-results.json'), 'utf8')); + } + + it('recovers completed iterations into the crash artifact', async () => { + const report = await crashWith([{ iteration: 0 }, { iteration: 1 }]); + + expect(report.testCases).toHaveLength(1); + expect(report.testCases[0].testCaseFile).toBe('case-a'); + expect(report.testCases[0].scenarios[0].runs).toHaveLength(2); + }); + + it('writes an empty artifact when no iteration completed — no fabricated failures', async () => { + // The single-scenario case needs one row per iteration; an empty sink means + // nothing finished, so nothing may be reported. + const report = await crashWith([]); + + expect(report.testCases).toHaveLength(0); + }); + + it('recovers embedded expectation verdicts — reshape reads only the side-band map', async () => { + const verdicts = [{ expectation: 'sends a digest', pass: true, reason: 'ok' }]; + const report = await crashWith([{ iteration: 0 }], { + ...rowOutputs(true), + expectationResults: verdicts, + }); + + expect(report.testCases[0].buildExpectationResultsPerRun).toEqual([verdicts]); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/rows.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/rows.test.ts new file mode 100644 index 00000000000..ba04f4c6c73 --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/__tests__/rows.test.ts @@ -0,0 +1,51 @@ +import type { WorkflowTestCaseWithFile } from '../data/workflows'; +import { roundRobinCaseRows } from '../run/rows'; +import type { WorkflowTestCase } from '../types'; + +// Pins the row set + order both drivers evaluate (run/rows.ts is the single +// flattening source for the LangSmith dataset sync AND the direct driver). + +function caseWith(fileSlug: string, scenarioNames: string[]): WorkflowTestCaseWithFile { + const testCase: WorkflowTestCase = { + conversation: [{ role: 'user', text: `build ${fileSlug}` }], + complexity: 'simple', + tags: [], + datasets: ['full'], + executionScenarios: scenarioNames.map((name) => ({ + name, + description: 'd', + dataSetup: 's', + successCriteria: 'c', + })), + }; + return { testCase, fileSlug }; +} + +describe('roundRobinCaseRows', () => { + it('orders scenario #n of every case before scenario #n+1, sentinels last', () => { + const rows = roundRobinCaseRows([ + caseWith('alpha', ['a1', 'a2', 'a3']), + caseWith('beta', ['b1', 'b2']), + caseWith('gamma', []), + ]); + + expect(rows.map((r) => `${r.testCaseFile}/${r.scenario?.name ?? ''}`)).toEqual([ + 'alpha/a1', + 'beta/b1', + 'alpha/a2', + 'beta/b2', + 'alpha/a3', + 'gamma/', + ]); + }); + + it('emits exactly one sentinel row per scenario-less case and none otherwise', () => { + const rows = roundRobinCaseRows([caseWith('one', ['s']), caseWith('two', [])]); + expect(rows.filter((r) => r.scenario === null)).toHaveLength(1); + expect(rows.find((r) => r.scenario === null)?.testCaseFile).toBe('two'); + }); + + it('returns an empty list for no cases', () => { + expect(roundRobinCaseRows([])).toEqual([]); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/cli/args.ts b/packages/@n8n/instance-ai/evaluations/cli/args.ts index 87f000675d0..e8a05b9d159 100644 --- a/packages/@n8n/instance-ai/evaluations/cli/args.ts +++ b/packages/@n8n/instance-ai/evaluations/cli/args.ts @@ -63,6 +63,9 @@ export interface CliArgs { outputDir?: string; /** LangSmith dataset name (synced from JSON test cases before each run) */ dataset: string; + /** True when `--source langtracer` auto-forked the dataset name off the suite + * (no explicit --dataset) — cohort isolation, surfaced loudly by the driver. */ + datasetAutoForked: boolean; /** Max concurrent target() calls in LangSmith evaluate(). Build concurrency is * enforced separately by the LaneAllocator (cap=4 per lane). */ concurrency: number; @@ -212,13 +215,17 @@ export function parseCliArgs(argv: string[]): CliArgs { // In langtracer mode, default the dataset + baseline to a suite-scoped, eval-tagged // name so runs don't pollute the shared cohort and re-runs upsert one stable dataset. let dataset = validated.dataset; + let datasetAutoForked = false; let baselinePrefix = validated.baselinePrefix; if (validated.source === 'langtracer' && validated.suite) { const suiteSlug = validated.suite .toLowerCase() .replace(/[^a-z0-9]+/g, '-') .replace(/^-+|-+$/g, ''); - if (!raw.datasetProvided) dataset = `instance-ai-langtracer-${suiteSlug}`; + if (!raw.datasetProvided) { + dataset = `instance-ai-langtracer-${suiteSlug}`; + datasetAutoForked = true; + } if (!raw.baselineProvided) baselinePrefix = `instance-ai-langtracer-${suiteSlug}-baseline-`; } @@ -235,6 +242,7 @@ export function parseCliArgs(argv: string[]): CliArgs { deletePrebuiltWorkflows: validated.deletePrebuiltWorkflows, outputDir: validated.outputDir, dataset, + datasetAutoForked, concurrency: validated.concurrency, experimentName: validated.experimentName, iterations: validated.iterations, diff --git a/packages/@n8n/instance-ai/evaluations/cli/index.ts b/packages/@n8n/instance-ai/evaluations/cli/index.ts index f666fd04f60..efa3b600892 100644 --- a/packages/@n8n/instance-ai/evaluations/cli/index.ts +++ b/packages/@n8n/instance-ai/evaluations/cli/index.ts @@ -5,8 +5,8 @@ // Parses args, selects cases, sets up lanes, then hands the run to one of two // drivers over the shared session/pipeline in evaluations/run/: the LangSmith // driver (evaluate() + experiments) when LANGSMITH_API_KEY is set, else the -// direct driver (same rows, same pipeline, eval-results.json only — the mode -// the LangTracer dispatcher invokes). +// direct driver (same rows, same pipeline and local artifacts, no LangSmith +// experiment tracking — the mode the LangTracer dispatcher invokes). // --------------------------------------------------------------------------- import { mkdirSync } from 'fs'; @@ -20,7 +20,7 @@ import { selectCases } from '../run/case-selection'; import { runDirect } from '../run/direct-driver'; import { cleanupLanes, setupLanes } from '../run/lane-setup'; import { runWithLangSmith } from '../run/langsmith-driver'; -import { ciRerunHint, runEvalAndPersist } from '../run/persist'; +import { ciRerunHint, createRowSink, runEvalAndPersist } from '../run/persist'; import { emitRunReports } from '../run/reporters'; async function main(): Promise { @@ -52,6 +52,8 @@ async function main(): Promise { args.deletePrebuiltWorkflows || (args.buildViaMcp && !args.keepWorkflows); const mcpBuildSpend: McpBuildSpend[] = []; + // Every completed row is journaled so a crashed run still persists verdicts. + const rowSink = createRowSink(args.outputDir); const commitSha = process.env.LANGSMITH_REVISION_ID ?? process.env.GITHUB_SHA; try { @@ -72,6 +74,8 @@ async function main(): Promise { commitSha, rerun: ciRerunHint(), mcpBuildSpend, + rowSink, + testCasesWithFiles, }, async (partialResults) => { if (hasLangSmith) { @@ -85,6 +89,7 @@ async function main(): Promise { cleanupBuiltWorkflows, mcpBuildLogDir, mcpBuildSpend, + rowSink, }); return { evaluation: langsmithRun.evaluation, @@ -107,6 +112,7 @@ async function main(): Promise { mcpBuildLogDir, mcpBuildSpend, partialResults, + rowSink, }); return { evaluation: directRun.evaluation, slugByTestCase: directRun.slugByTestCase }; }, @@ -128,18 +134,6 @@ async function main(): Promise { } } -// --------------------------------------------------------------------------- -// LangSmith mode: evaluate() with dataset sync, tracing, experiments -// --------------------------------------------------------------------------- - -// --------------------------------------------------------------------------- -// eval-results.json output (same shape as CI PR comment expects) -// --------------------------------------------------------------------------- - -// --------------------------------------------------------------------------- -// Comparison vs the pinned baseline experiment -// --------------------------------------------------------------------------- - // Only auto-run as the CLI entry point. Importing this module (e.g. from a unit // test that exercises the exported runEvalAndPersist / writeEvalResults seams) // must not kick off a real eval run against process.argv. diff --git a/packages/@n8n/instance-ai/evaluations/comparison/gate.ts b/packages/@n8n/instance-ai/evaluations/comparison/gate.ts index 07600901cf7..4779719ea37 100644 --- a/packages/@n8n/instance-ai/evaluations/comparison/gate.ts +++ b/packages/@n8n/instance-ai/evaluations/comparison/gate.ts @@ -14,11 +14,13 @@ // nothing to judge. // --------------------------------------------------------------------------- +import { GATED_TIER_NAMES } from '../run/tiers'; import type { MultiRunEvaluation, WorkflowTestCase } from '../types'; // Tiers whose runs assert an absolute green bar instead of comparing to a -// baseline. Keep this the single source of truth for "is this a gated run". -export const GATED_TIERS = new Set(['pr']); +// baseline — declared in run/tiers.ts, the single source of truth for +// "is this a gated run". Tier names are otherwise free-form (`datasets`). +export const GATED_TIERS = GATED_TIER_NAMES; export function isGatedTier(tier?: string): boolean { return tier !== undefined && GATED_TIERS.has(tier); diff --git a/packages/@n8n/instance-ai/evaluations/data/discovery/index.ts b/packages/@n8n/instance-ai/evaluations/data/discovery/index.ts index d6eef217fb4..2ee409646f9 100644 --- a/packages/@n8n/instance-ai/evaluations/data/discovery/index.ts +++ b/packages/@n8n/instance-ai/evaluations/data/discovery/index.ts @@ -1,8 +1,60 @@ import { readFileSync, readdirSync } from 'fs'; import { basename, join } from 'path'; +import { z } from 'zod'; +import type { LocalGatewayStatus } from '../../../src/types'; import type { DiscoveryTestCase } from '../../discovery/types'; +const forbiddenToolCallSchema = z + .object({ + toolName: z.string().min(1), + argsContainAny: z.array(z.string().min(1)).min(1).optional(), + }) + .strict(); + +/** Mirrors `LocalGatewayStatus` (src/types.ts) — the annotation makes tsc flag + * this schema when the source union drifts. */ +const localGatewayStatusSchema: z.ZodType = z.discriminatedUnion('status', [ + z.object({ status: z.literal('connected'), capabilities: z.array(z.string()) }).strict(), + z.object({ status: z.literal('disabledGlobally') }).strict(), + z.object({ status: z.literal('disconnected') }).strict(), + z.object({ status: z.literal('disabled') }).strict(), +]); + +/** Strict authoring schema for discovery cases — a typo'd key or an empty + * expectation must fail at load time, not pass vacuously at run time (the + * workflow-case loader has had this guarantee for a while; discovery cases + * were a blind JSON.parse cast until TRUST-261's cleanup). */ +export const discoveryTestCaseSchema = z + .object({ + id: z.string().min(1), + userMessage: z.string().min(1), + instanceState: z + .object({ + localGateway: localGatewayStatusSchema.optional(), + browserAvailable: z.boolean().optional(), + }) + .strict() + .optional(), + expectedToolInvocations: z + .object({ + // min(1) on every list: an empty expectation array is dead config that + // would otherwise pass here and only surface as a run-time failure. + anyOf: z.array(z.string().min(1)).min(1).optional(), + noneOf: z.array(z.string().min(1)).min(1).optional(), + anyOfToolCalls: z.array(forbiddenToolCallSchema).min(1).optional(), + allOfToolCalls: z.array(forbiddenToolCallSchema).min(1).optional(), + noneOfToolCalls: z.array(forbiddenToolCallSchema).min(1).optional(), + }) + .strict() + .refine((expectations) => Object.values(expectations).some((v) => v !== undefined), { + message: 'expectedToolInvocations needs at least one expectation key', + }), + rationale: z.string().optional(), + maxSteps: z.number().int().positive().optional(), + }) + .strict(); + export interface DiscoveryTestCaseWithFile { testCase: DiscoveryTestCase; /** Filename without extension, e.g. "slack-oauth-credential-setup" */ @@ -11,13 +63,22 @@ export interface DiscoveryTestCaseWithFile { function parseTestCaseFile(filePath: string): DiscoveryTestCase { const content = readFileSync(filePath, 'utf-8'); + let raw: unknown; try { - return JSON.parse(content) as DiscoveryTestCase; + raw = JSON.parse(content); } catch (error) { throw new Error( `Failed to parse discovery test case ${filePath}: ${error instanceof Error ? error.message : String(error)}`, ); } + const parsed = discoveryTestCaseSchema.safeParse(raw); + if (!parsed.success) { + const issues = parsed.error.issues + .map((issue) => `${issue.path.join('.') || '(root)'}: ${issue.message}`) + .join('; '); + throw new Error(`Invalid discovery test case ${filePath}: ${issues}`); + } + return parsed.data; } function parseSubstringList(value: string | undefined): string[] { diff --git a/packages/@n8n/instance-ai/evaluations/index.ts b/packages/@n8n/instance-ai/evaluations/index.ts index bb6c34b3104..f0b1a0f216d 100644 --- a/packages/@n8n/instance-ai/evaluations/index.ts +++ b/packages/@n8n/instance-ai/evaluations/index.ts @@ -2,7 +2,8 @@ // Public API for the instance-ai workflow evaluation framework // // This module exports the domain logic used by the CLI (evaluations/cli/) -// and available for custom orchestration (e.g. LangSmith evaluate). +// and available for custom orchestration. The run phases themselves live in +// evaluations/run/ — see evaluations/ARCHITECTURE.md. // --------------------------------------------------------------------------- // -- Client & Auth -- diff --git a/packages/@n8n/instance-ai/evaluations/langsmith/dataset-sync.ts b/packages/@n8n/instance-ai/evaluations/langsmith/dataset-sync.ts index 016ade12287..9e94f737813 100644 --- a/packages/@n8n/instance-ai/evaluations/langsmith/dataset-sync.ts +++ b/packages/@n8n/instance-ai/evaluations/langsmith/dataset-sync.ts @@ -20,6 +20,7 @@ import { z } from 'zod'; import type { WorkflowTestCaseWithFile } from '../data/workflows'; import type { EvalLogger } from '../harness/logger'; +import { BUILD_ONLY_SCENARIO_NAME, roundRobinCaseRows } from '../run/rows'; /** * Shape of the inputs passed to the target function for each scenario. @@ -102,9 +103,15 @@ export async function syncDataset( logger.info(`Created dataset: ${datasetName}`); } - // List existing examples, keyed by derived ID (testCaseFile/scenarioName from inputs). + // List existing examples, keyed by derived ID (testCaseFile/scenarioName from + // inputs). Scoped to the synced cases' slug splits: every mutation below only + // touches these slugs, and a scoped read keeps concurrent syncs of disjoint + // cases (the LangTracer dispatcher pattern) and sync cost independent of + // dataset size. Already-archived examples carry only the 'archived' split, so + // they fall out of the read — which keeps re-archiving idempotent for free. + const slugSplits = [...new Set(testCasesWithFiles.map((tc) => tc.fileSlug))]; const existingByDerivedId = new Map(); - for await (const example of lsClient.listExamples({ datasetId })) { + for await (const example of lsClient.listExamples({ datasetId, splits: slugSplits })) { const inputs = existingInputsSchema.safeParse(example.inputs); if (!inputs.success) continue; existingByDerivedId.set(`${inputs.data.testCaseFile}/${inputs.data.scenarioName}`, example); @@ -229,14 +236,45 @@ export async function syncDataset( return datasetName; } +/** Read-after-write guard: freshly created examples can lag the immediate + * list. Verify the split-scoped count covers what was just synced before a + * driver starts an experiment — an invisible example silently produces an + * empty or partial run (the dispatcher's historical "no results" failure). */ +export async function ensureExamplesVisible( + lsClient: Client, + datasetName: string, + testCasesWithFiles: WorkflowTestCaseWithFile[], + logger: EvalLogger, + opts: { attempts?: number; baseDelayMs?: number } = {}, +): Promise { + const expected = roundRobinCaseRows(testCasesWithFiles).length; + if (expected === 0) return; + const attempts = opts.attempts ?? 3; + const baseDelayMs = opts.baseDelayMs ?? 2_000; + const splits = [...new Set(testCasesWithFiles.map((tc) => tc.fileSlug))]; + for (let attempt = 1; ; attempt++) { + let count = 0; + for await (const _example of lsClient.listExamples({ datasetName, splits })) count++; + if (count >= expected) return; + if (attempt >= attempts) { + throw new Error( + `Dataset "${datasetName}" lists ${String(count)}/${String(expected)} synced example(s) after ${String(attempts)} attempt(s) — read-after-write lag or split drift; refusing to run a partial experiment.`, + ); + } + logger.warn( + `Dataset "${datasetName}" lists ${String(count)}/${String(expected)} synced example(s); retrying (${String(attempt)}/${String(attempts)})…`, + ); + await new Promise((resolve) => setTimeout(resolve, baseDelayMs * attempt)); + } +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- -/** Scenario name for the single "build-only" row a 0-scenario case emits, so the - * workflow still builds and its process/outcome expectations get judged. Shared - * with target() and reshape so all three agree on the sentinel. */ -export const BUILD_ONLY_SCENARIO_NAME = '__build_only__'; +// Home moved to run/rows.ts (single row-flattening source for both drivers); +// re-exported here so existing importers keep working. +export { BUILD_ONLY_SCENARIO_NAME }; interface FlatScenario { testCaseFile: string; @@ -258,49 +296,17 @@ interface FlatScenario { * Output: [tc1/s1, tc2/s1, tc3/s1, tc1/s2, tc2/s2, tc1/s3] */ function buildRoundRobinScenarios(testCasesWithFiles: WorkflowTestCaseWithFile[]): FlatScenario[] { - const result: FlatScenario[] = []; - const maxScenarios = Math.max( - ...testCasesWithFiles.map((tc) => (tc.testCase.executionScenarios ?? []).length), - 0, - ); - - for (let i = 0; i < maxScenarios; i++) { - for (const { testCase, fileSlug } of testCasesWithFiles) { - const scenario = testCase.executionScenarios?.[i]; - if (scenario) { - result.push({ - testCaseFile: fileSlug, - scenarioName: scenario.name, - scenarioDescription: scenario.description, - dataSetup: scenario.dataSetup, - successCriteria: scenario.successCriteria, - complexity: testCase.complexity, - tags: testCase.tags, - triggerType: testCase.triggerType, - datasets: testCase.datasets, - }); - } - } - } - - // Build-only cases (0 scenarios) emit one sentinel row so the workflow still builds and its expectations get judged. - for (const { testCase, fileSlug } of testCasesWithFiles) { - if ((testCase.executionScenarios?.length ?? 0) === 0) { - result.push({ - testCaseFile: fileSlug, - scenarioName: BUILD_ONLY_SCENARIO_NAME, - scenarioDescription: '', - dataSetup: '', - successCriteria: '', - complexity: testCase.complexity, - tags: testCase.tags, - triggerType: testCase.triggerType, - datasets: testCase.datasets, - }); - } - } - - return result; + return roundRobinCaseRows(testCasesWithFiles).map(({ testCase, testCaseFile, scenario }) => ({ + testCaseFile, + scenarioName: scenario?.name ?? BUILD_ONLY_SCENARIO_NAME, + scenarioDescription: scenario?.description ?? '', + dataSetup: scenario?.dataSetup ?? '', + successCriteria: scenario?.successCriteria ?? '', + complexity: testCase.complexity, + tags: testCase.tags, + triggerType: testCase.triggerType, + datasets: testCase.datasets, + })); } // Schemas for reading existing LangSmith example data, which is typed as an diff --git a/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts b/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts index 47e02ed25ef..4cd5baf1673 100644 --- a/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts +++ b/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts @@ -16,6 +16,7 @@ import { allFailVerdicts, verifyBuildExpectations } from '../build-expectations/ import type { CliArgs } from '../cli/args'; import { buildWorkflowViaMcp, type McpBuildSettings } from '../cli/mcp-builder'; import type { N8nClient } from '../clients/n8n-client'; +import { resolveArtifactContext } from '../harness/artifacts/artifact-context'; import { captureThreadRunDebug } from '../harness/capture-run-debug'; import type { EvalLogger } from '../harness/logger'; import { @@ -31,6 +32,7 @@ import { type BuildResult, type executeAgentScenario, type executeScenario, + type ScenarioSeedContext, } from '../harness/runner'; import { isTransientNetworkError } from '../harness/transient-error'; import type { @@ -96,6 +98,7 @@ export interface LaneState { workflowJsons: BuildResult['workflowJsons']; buildTrace?: BuildResult['buildTrace']; timeoutMs: number; + seedContext?: ScenarioSeedContext; }) => Promise>>; tracedExecuteAgent: (execArgs: { agentId: string; @@ -290,6 +293,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche function stashBuildExpectations( key: string, fileSlug: string, + client: N8nClient, build: BuildResult, isPrebuilt: boolean, ): void { @@ -305,11 +309,21 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche if (expectations.length === 0) return; buildExpectationsByKey.set( key, - verifyBuildExpectations(expectations, { - transcript, - workflowJson: build.workflowJsons[0], - metrics: build.conversationMetrics, - }).catch((error: unknown) => + (async () => + await verifyBuildExpectations(expectations, { + transcript, + workflowJson: build.workflowJsons[0], + metrics: build.conversationMetrics, + // Rendered non-workflow artifacts (agent AND config-eval), sectioned + // with "(no produced)" fallbacks, so outcome expectations can + // judge artifact existence, absence and content — parity with the + // retired direct loop, which always threaded resolveArtifactContext. + artifactContext: await resolveArtifactContext({ + artifactRefs: build.artifactRefs ?? [], + client, + logger, + }), + }))().catch((error: unknown) => allFailVerdicts( expectations, `judge error: ${error instanceof Error ? error.message : String(error)}`, @@ -364,7 +378,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche stashTranscript(build); // isPrebuilt=true: MCP builds have no build transcript, so only // outcome expectations are judged (against the workflow), like prebuilt. - stashBuildExpectations(key, fileSlug, build, true); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); stashRunDebug(lane.runner.client, build); if (build.success && !build.workflowChecks) { build.workflowChecks = await runWorkflowChecks({ @@ -390,7 +404,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche const buildDurationMs = Date.now() - start; buildDurations.set(key, buildDurationMs); stashTranscript(build); - stashBuildExpectations(key, fileSlug, build, true); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); stashRunDebug(lane.runner.client, build); if (build.success && !build.workflowChecks) { // No transcript in prebuilt mode, but the authored conversation still @@ -452,8 +466,8 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche } buildDurations.set(key, buildDurationMs); stashTranscript(build); - stashBuildExpectations(key, fileSlug, build, false); stashAgentContext(key, lane.runner.client, build); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, false); stashRunDebug(lane.runner.client, build); logger.info( `[lane ${String(lane.laneNum)}] built ${fileSlug} (iteration ${String(iteration)}) thread=${build.threadId ?? 'none'} success=${String(build.success)}`, diff --git a/packages/@n8n/instance-ai/evaluations/run/case-pipeline.ts b/packages/@n8n/instance-ai/evaluations/run/case-pipeline.ts index 92f357cc47c..d4659323cd8 100644 --- a/packages/@n8n/instance-ai/evaluations/run/case-pipeline.ts +++ b/packages/@n8n/instance-ai/evaluations/run/case-pipeline.ts @@ -20,7 +20,9 @@ import { cleanupBuild, effectiveTimeoutMs, findAgentArtifactRef, + scenariosRequireSerialSeeding, warnAgentSeedDataTablesIgnored, + type ScenarioSeedContext, } from '../harness/runner'; import { classifyScenarioExecutionError, @@ -106,6 +108,21 @@ export function createCasePipeline(deps: CasePipelineDeps): CasePipeline { } } + // Per-build-key execution chain for seed-table cases (see runScenarioRow). + const serialExecutionByKey = new Map>(); + async function withSerialSeeding(key: string, fn: () => Promise): Promise { + const prev = serialExecutionByKey.get(key) ?? Promise.resolve(); + const next = prev.then(fn, fn); + serialExecutionByKey.set( + key, + next.then( + () => undefined, + () => undefined, + ), + ); + return await next; + } + const runRow = async (inputs: ScenarioRowInputs): Promise => { const iteration = inputs._iteration ?? 0; try { @@ -140,7 +157,13 @@ export function createCasePipeline(deps: CasePipelineDeps): CasePipeline { inputs: ScenarioRowInputs, iteration: number, ): Promise => { - const scenario: ExecutionScenario = { + // Rows carry only the per-scenario prose; the authored scenario is the + // source of truth for typed extras (seedDataTables) — merge it back so + // table-backed scenarios seed their declared rows in both drivers. + const authoredScenarios = testCaseByFileSlug.get(inputs.testCaseFile)?.executionScenarios ?? []; + const scenario: ExecutionScenario = authoredScenarios.find( + (s) => s.name === inputs.scenarioName, + ) ?? { name: inputs.scenarioName, description: inputs.scenarioDescription, dataSetup: inputs.dataSetup, @@ -304,88 +327,128 @@ export function createCasePipeline(deps: CasePipelineDeps): CasePipeline { // and guards the workflow path below. throw new Error(`No runnable artifact for scenario ${scenario.name}`); } + // Captured as a const so the narrowing survives into the closure below. + const workflowId = build.workflowId; - const execStart = Date.now(); - const nodeCount = build.workflowJsons[0]?.nodes.length ?? 0; - let result; - for (let attempt = 1; ; attempt++) { - try { - result = await builtOnLane.tracedExecute({ - workflowId: build.workflowId, - scenario, - workflowJsons: build.workflowJsons, - buildTrace: build.buildTrace, - timeoutMs: effectiveTimeoutMs( - testCaseByFileSlug.get(inputs.testCaseFile)?.complexity, - args.timeoutMs, - ), - }); - break; - } catch (error: unknown) { - const errorMessage = extractErrorMessage(error); - if (shouldRetryScenarioExecution(errorMessage, attempt)) { - logger.warn( - ` [${scenario.name}] execution attempt ${attempt}/${MAX_EXEC_ATTEMPTS} failed (${errorMessage}); retrying`, - ); - await new Promise((resolve) => setTimeout(resolve, 500 * attempt)); - continue; - } - // Mirror direct mode's per-scenario guard — without this, n8n API errors, - // verifier timeouts, or a per-iteration budget abort from - // executeWithLlmMock / verifyChecklist would escape to the driver, come - // back as a Run with null outputs, and be misclassified as builder - // regressions by the feedback extractor. classifyScenarioExecutionError - // stamps framework_issue + a timeout-flavoured rootCause for budget aborts. - logger.error(` ERROR [${scenario.name}]: ${errorMessage}`); - const classified = classifyScenarioExecutionError(errorMessage); - return await attachExpectations({ - buildSuccess: true, - workflowId: build.workflowId, - passed: false, - score: 0, - reasoning: classified.reasoning, - failureCategory: classified.failureCategory, - rootCause: classified.rootCause, - execErrors: [errorMessage], - buildDurationMs, - execDurationMs: Date.now() - execStart, - nodeCount, - threadId: build.threadId, - workflowChecks: build.workflowChecks, - workflowJson: build.workflowJsons[0], - buildTrace: build.buildTrace, - planRejections: build.proxyDecisionStats?.rejection ?? 0, - }); - } + // Mirrors the retired direct loop (TRUST-311): a seeded row resets + seeds + const seedContext: ScenarioSeedContext | undefined = + build.threadId && build.seededScenarioTableIdsByName + ? { threadId: build.threadId, tableIdsByName: build.seededScenarioTableIdsByName } + : undefined; + // A scenario that declares seed tables must not run without them (MCP and + // prebuilt builds never seed data tables) — executing anyway would grade the + // workflow against empty tables and report the miss as a builder failure. + if ((scenario.seedDataTables?.length ?? 0) > 0 && !seedContext) { + const reason = + 'Scenario declares seedDataTables but the build provided no seeded-table mapping ' + + '(MCP/prebuilt builds do not seed data tables) — refusing to run without the declared rows'; + logger.error(` ERROR [${scenario.name}]: ${reason}`); + return await attachExpectations({ + buildSuccess: true, + workflowId, + passed: false, + score: 0, + reasoning: reason, + failureCategory: 'framework_issue', + execErrors: [reason], + buildDurationMs, + execDurationMs: 0, + nodeCount: build.workflowJsons[0]?.nodes.length ?? 0, + threadId: build.threadId, + buildTrace: build.buildTrace, + planRejections: build.proxyDecisionStats?.rejection ?? 0, + }); } - const execDurationMs = Date.now() - execStart; + const runWorkflowScenario = async (): Promise => { + const execStart = Date.now(); + const nodeCount = build.workflowJsons[0]?.nodes.length ?? 0; + let result; + for (let attempt = 1; ; attempt++) { + try { + result = await builtOnLane.tracedExecute({ + workflowId, + scenario, + workflowJsons: build.workflowJsons, + buildTrace: build.buildTrace, + seedContext, + timeoutMs: effectiveTimeoutMs( + testCaseByFileSlug.get(inputs.testCaseFile)?.complexity, + args.timeoutMs, + ), + }); + break; + } catch (error: unknown) { + const errorMessage = extractErrorMessage(error); + if (shouldRetryScenarioExecution(errorMessage, attempt)) { + logger.warn( + ` [${scenario.name}] execution attempt ${attempt}/${MAX_EXEC_ATTEMPTS} failed (${errorMessage}); retrying`, + ); + await new Promise((resolve) => setTimeout(resolve, 500 * attempt)); + continue; + } + // Mirror direct mode's per-scenario guard — without this, n8n API errors, + // verifier timeouts, or a per-iteration budget abort from + // executeWithLlmMock / verifyChecklist would escape to the driver, come + // back as a Run with null outputs, and be misclassified as builder + // regressions by the feedback extractor. classifyScenarioExecutionError + // stamps framework_issue + a timeout-flavoured rootCause for budget aborts. + logger.error(` ERROR [${scenario.name}]: ${errorMessage}`); + const classified = classifyScenarioExecutionError(errorMessage); + return await attachExpectations({ + buildSuccess: true, + workflowId: build.workflowId, + passed: false, + score: 0, + reasoning: classified.reasoning, + failureCategory: classified.failureCategory, + rootCause: classified.rootCause, + execErrors: [errorMessage], + buildDurationMs, + execDurationMs: Date.now() - execStart, + nodeCount, + threadId: build.threadId, + workflowChecks: build.workflowChecks, + workflowJson: build.workflowJsons[0], + buildTrace: build.buildTrace, + planRejections: build.proxyDecisionStats?.rejection ?? 0, + }); + } + } + const execDurationMs = Date.now() - execStart; - // Strip failure fields on pass: the verifier sometimes returns "." - // placeholders instead of omitting them. - const failureCategory = result.success ? undefined : result.failureCategory; - const rootCause = result.success ? undefined : result.rootCause; + // Strip failure fields on pass: the verifier sometimes returns "." + // placeholders instead of omitting them. + const failureCategory = result.success ? undefined : result.failureCategory; + const rootCause = result.success ? undefined : result.rootCause; - return await attachExpectations({ - buildSuccess: true, - workflowId: build.workflowId, - scenarioWorkflowId: result.workflowId, - passed: result.success, - score: result.score, - reasoning: result.reasoning, - failureCategory, - rootCause, - ...(result.incomplete ? { incomplete: true } : {}), - execErrors: result.evalResult?.errors ?? [], - evalResult: result.evalResult, - buildDurationMs, - execDurationMs, - nodeCount, - threadId: build.threadId, - workflowChecks: build.workflowChecks, - workflowJson: build.workflowJsons[0], - buildTrace: build.buildTrace, - planRejections: build.proxyDecisionStats?.rejection ?? 0, - }); + return await attachExpectations({ + buildSuccess: true, + workflowId: build.workflowId, + scenarioWorkflowId: result.workflowId, + passed: result.success, + score: result.score, + reasoning: result.reasoning, + failureCategory, + rootCause, + ...(result.incomplete ? { incomplete: true } : {}), + execErrors: result.evalResult?.errors ?? [], + evalResult: result.evalResult, + buildDurationMs, + execDurationMs, + nodeCount, + threadId: build.threadId, + workflowChecks: build.workflowChecks, + workflowJson: build.workflowJsons[0], + buildTrace: build.buildTrace, + planRejections: build.proxyDecisionStats?.rejection ?? 0, + }); + }; + // Scenarios of one case share tables by name, so seeded rows must not + // interleave — the retired direct loop ran them at concurrency 1; rows now + // arrive independently, so the gate is a per-build-key chain instead. + return scenariosRequireSerialSeeding(authoredScenarios) + ? await withSerialSeeding(cacheKey, runWorkflowScenario) + : await runWorkflowScenario(); }; return { runRow }; diff --git a/packages/@n8n/instance-ai/evaluations/run/direct-driver.ts b/packages/@n8n/instance-ai/evaluations/run/direct-driver.ts index f5e92b96855..e5b9a6ea0ea 100644 --- a/packages/@n8n/instance-ai/evaluations/run/direct-driver.ts +++ b/packages/@n8n/instance-ai/evaluations/run/direct-driver.ts @@ -12,61 +12,38 @@ import { aggregateResults } from './aggregator'; import type { ScenarioRowInputs } from './case-pipeline'; import { createEvalSession, type EvalSessionConfig } from './eval-session'; import { expandWithIterations } from './iterations'; +import { type RowSink } from './persist'; import { reshapeLangSmithRuns, type ReshapeRunRow } from './reshape'; +import { BUILD_ONLY_SCENARIO_NAME, roundRobinCaseRows } from './rows'; import type { WorkflowTestCaseWithFile } from '../data/workflows'; import { runWithConcurrency } from '../harness/runner'; -import { BUILD_ONLY_SCENARIO_NAME } from '../langsmith/dataset-sync'; import type { MultiRunEvaluation, WorkflowTestCase, WorkflowTestCaseResult } from '../types'; export interface DirectRunConfig extends Omit { /** Sink for per-iteration results as reshape produces them, so an abort in * aggregation/persistence still leaves runEvalAndPersist the completed rows. */ partialResults?: WorkflowTestCaseResult[][]; + /** Journal of completed rows for crash recovery (see run/persist.ts). */ + rowSink?: RowSink; } -/** Mirror of the dataset sync's round-robin ordering: scenario #1 of every - * case, then scenario #2, …, then one build-only sentinel row per - * scenario-less case — so builds diversify early instead of burning all - * concurrency slots on one test case. */ +/** Same flattening as the LangSmith dataset sync (run/rows.ts), projected to + * the per-row input shape the pipeline consumes. */ function roundRobinRows(testCasesWithFiles: WorkflowTestCaseWithFile[]): ScenarioRowInputs[] { - const rows: ScenarioRowInputs[] = []; - const maxScenarios = Math.max( - ...testCasesWithFiles.map(({ testCase }) => (testCase.executionScenarios ?? []).length), - 0, - ); - for (let i = 0; i < maxScenarios; i++) { - for (const { testCase, fileSlug } of testCasesWithFiles) { - const scenario = testCase.executionScenarios?.[i]; - if (scenario) { - rows.push({ - testCaseFile: fileSlug, - scenarioName: scenario.name, - scenarioDescription: scenario.description, - dataSetup: scenario.dataSetup, - successCriteria: scenario.successCriteria, - }); - } - } - } - for (const { testCase, fileSlug } of testCasesWithFiles) { - if ((testCase.executionScenarios?.length ?? 0) === 0) { - rows.push({ - testCaseFile: fileSlug, - scenarioName: BUILD_ONLY_SCENARIO_NAME, - scenarioDescription: '', - dataSetup: '', - successCriteria: '', - }); - } - } - return rows; + return roundRobinCaseRows(testCasesWithFiles).map(({ testCaseFile, scenario }) => ({ + testCaseFile, + scenarioName: scenario?.name ?? BUILD_ONLY_SCENARIO_NAME, + scenarioDescription: scenario?.description ?? '', + dataSetup: scenario?.dataSetup ?? '', + successCriteria: scenario?.successCriteria ?? '', + })); } export async function runDirect(config: DirectRunConfig): Promise<{ evaluation: MultiRunEvaluation; slugByTestCase: Map; }> { - const { args, lanes, logger, testCasesWithFiles, partialResults } = config; + const { args, lanes, logger, testCasesWithFiles, partialResults, rowSink } = config; if (testCasesWithFiles.length === 0) { console.log('No workflow test cases selected (check --source / --filter / --exclude / --tier)'); @@ -97,7 +74,12 @@ export async function runDirect(config: DirectRunConfig): Promise<{ // flight, builds capped per lane by the allocator (MAX_CONCURRENT_BUILDS). const completed: ReshapeRunRow[] = await runWithConcurrency( rows, - async (row) => ({ run: { inputs: row, outputs: await session.pipeline.runRow(row) } }), + async (row) => { + const outputs = await session.pipeline.runRow(row); + const completedRow = { run: { inputs: row, outputs } }; + rowSink?.append(completedRow); + return completedRow; + }, args.concurrency, ); diff --git a/packages/@n8n/instance-ai/evaluations/run/eval-session.ts b/packages/@n8n/instance-ai/evaluations/run/eval-session.ts index 36c07f6b8fe..c26419dc761 100644 --- a/packages/@n8n/instance-ai/evaluations/run/eval-session.ts +++ b/packages/@n8n/instance-ai/evaluations/run/eval-session.ts @@ -32,6 +32,7 @@ import { executeScenario, workflowExpectedForCase, type BuildResult, + type ScenarioSeedContext, } from '../harness/runner'; import type { BuildExpectationResult, @@ -161,6 +162,7 @@ export function createEvalSession(config: EvalSessionConfig): EvalSession { workflowJsons: BuildResult['workflowJsons']; buildTrace?: BuildResult['buildTrace']; timeoutMs: number; + seedContext?: ScenarioSeedContext; }) => await executeScenario( lane.client, @@ -172,6 +174,7 @@ export function createEvalSession(config: EvalSessionConfig): EvalSession { undefined, execArgs.buildTrace, args.pinAiRoots, + execArgs.seedContext, ), ), tracedExecuteAgent: wrap( diff --git a/packages/@n8n/instance-ai/evaluations/run/langsmith-driver.ts b/packages/@n8n/instance-ai/evaluations/run/langsmith-driver.ts index 64ff9ee2976..a543f42f78b 100644 --- a/packages/@n8n/instance-ai/evaluations/run/langsmith-driver.ts +++ b/packages/@n8n/instance-ai/evaluations/run/langsmith-driver.ts @@ -15,11 +15,17 @@ import { traceable } from 'langsmith/traceable'; import { aggregateResults, passAtK, passHatK } from './aggregator'; import { type Lane, type McpBuildSpend } from './build-orchestrator'; +import type { ScenarioRowInputs } from './case-pipeline'; import { buildCIMetadata, computeExperimentPrefix } from './ci-metadata'; import { createEvalSession, MAX_CONCURRENT_BUILDS } from './eval-session'; import { expandWithIterations } from './iterations'; -import { computePassRatePerIter, summarizeMcpBuildSpend } from './persist'; -import { isPlainObject, parseTargetOutput, reshapeLangSmithRuns } from './reshape'; +import { computePassRatePerIter, summarizeMcpBuildSpend, type RowSink } from './persist'; +import { + isPlainObject, + parseTargetOutput, + reshapeLangSmithRuns, + type TargetOutput, +} from './reshape'; import { partialIsolationWarning } from '../cli/args'; import type { CliArgs } from '../cli/args'; import { bucketFromEvaluation } from '../comparison/bucket-from-evaluation'; @@ -30,8 +36,8 @@ import type { WorkflowTestCaseWithFile } from '../data/workflows'; import { EVAL_WORKSPACE_NAME, resolveEvalWorkspaceId } from '../harness/langsmith-seed'; import type { EvalLogger } from '../harness/logger'; import type { PrebuiltManifest } from '../harness/prebuilt-workflows'; -import { syncDataset } from '../langsmith/dataset-sync'; -import type { MultiRunEvaluation, WorkflowTestCase, WorkflowTestCaseResult } from '../types'; +import { ensureExamplesVisible, syncDataset } from '../langsmith/dataset-sync'; +import type { MultiRunEvaluation, WorkflowTestCase } from '../types'; export interface RunConfig { args: CliArgs; @@ -51,10 +57,8 @@ export interface RunConfig { * LangSmith experiment metadata and eval-results.json — the run's only spend * record beyond raw session logs, for a suite that's manual-only due to cost. */ mcpBuildSpend: McpBuildSpend[]; - /** Optional sink the direct loop pushes each completed iteration's results into - * as they finish, so an abort that rejects the run still leaves the caller - * (runEvalAndPersist) with the scenarios that already completed. */ - partialResults?: WorkflowTestCaseResult[][]; + /** Journal of completed rows for crash recovery (see run/persist.ts). */ + rowSink?: RowSink; } export async function runWithLangSmith(config: RunConfig): Promise<{ @@ -90,6 +94,11 @@ export async function runWithLangSmith(config: RunConfig): Promise<{ // isolation; overriding only one silently touches shared Instance AI data. const isolationWarning = partialIsolationWarning(args.dataset, args.baselinePrefix); if (isolationWarning) logger.warn(isolationWarning); + if (args.datasetAutoForked) { + logger.warn( + `--source langtracer auto-forked this run to dataset "${args.dataset}" (baseline prefix "${args.baselinePrefix}") for cohort isolation. Pass --dataset/--baseline-prefix explicitly to target a shared cohort.`, + ); + } // Pin eval writes to the eval workspace; our PAT would otherwise default to Prod. const workspaceId = await resolveEvalWorkspaceId(); @@ -100,6 +109,7 @@ export async function runWithLangSmith(config: RunConfig): Promise<{ } const lsClient = new Client(workspaceId ? { workspaceId } : {}); const datasetName = await syncDataset(lsClient, args.dataset, logger, testCasesWithFiles); + await ensureExamplesVisible(lsClient, datasetName, testCasesWithFiles, logger); // Shared per-run assembly (lanes → allocator → build orchestrator → case // pipeline). The traceable() hook around the lane functions is the only @@ -124,7 +134,11 @@ export async function runWithLangSmith(config: RunConfig): Promise<{ }) as typeof fn, }); const { buildDurations } = session.orchestrator; - const target = session.pipeline.runRow; + const target = async (inputs: ScenarioRowInputs): Promise => { + const outputs = await session.pipeline.runRow(inputs); + config.rowSink?.append({ run: { inputs, outputs } }); + return outputs; + }; const feedbackExtractor = ({ run }: { run: Run }): EvaluationResult[] => { const output = parseTargetOutput(run.outputs); diff --git a/packages/@n8n/instance-ai/evaluations/run/persist.ts b/packages/@n8n/instance-ai/evaluations/run/persist.ts index 9d59f22757c..cfc7691b7a6 100644 --- a/packages/@n8n/instance-ai/evaluations/run/persist.ts +++ b/packages/@n8n/instance-ai/evaluations/run/persist.ts @@ -5,20 +5,28 @@ // crash path that still writes whatever completed when the run threw. // --------------------------------------------------------------------------- -import { mkdirSync, writeFileSync } from 'fs'; +import { appendFileSync, mkdirSync, readFileSync, writeFileSync } from 'fs'; import { join } from 'path'; import { aggregateResults } from './aggregator'; import type { McpBuildSpend } from './build-orchestrator'; +import { parseTargetOutput, reshapeLangSmithRuns, type ReshapeRunRow } from './reshape'; +import { roundRobinCaseRows } from './rows'; import { aggregateWorkflowChecks, statusMap } from '../binaryChecks/aggregate'; import type { CliArgs } from '../cli/args'; import type { ComparisonOutcome, ComparisonResult } from '../comparison/compare'; import { formatComparisonMarkdown, type RerunHint } from '../comparison/format'; import { evaluateGate, isGatedTier, type GateResult } from '../comparison/gate'; +import type { WorkflowTestCaseWithFile } from '../data/workflows'; import type { EvalLogger } from '../harness/logger'; import { extractErrorMessage } from '../harness/transient-error'; import { rollupCaseVerification } from '../summary'; -import type { MultiRunEvaluation, WorkflowTestCase, WorkflowTestCaseResult } from '../types'; +import type { + BuildExpectationResult, + MultiRunEvaluation, + WorkflowTestCase, + WorkflowTestCaseResult, +} from '../types'; import { caseDisplayPrompt } from '../utils/conversation-text'; /** @@ -127,6 +135,108 @@ export function ciRerunHint(): RerunHint | undefined { /** What `runEval` produces on success — the aggregation plus the LangSmith-only * comparison metadata (undefined in direct-loop mode). */ + +// --------------------------------------------------------------------------- +// Row sink — crash recovery +// --------------------------------------------------------------------------- + +/** Journals one JSON line per completed row so a run that dies mid-flight + * (budget abort, lane meltdown, OOM) still leaves recoverable verdicts. + * Both drivers feed it; `runEvalAndPersist` reads it back on the crash path. */ +export interface RowSink { + path: string; + append: (row: ReshapeRunRow) => void; + readRows: () => ReshapeRunRow[]; +} + +export function createRowSink(outputDir: string | undefined): RowSink { + const dir = outputDir ?? process.cwd(); + mkdirSync(dir, { recursive: true }); + const path = join(dir, 'eval-rows.jsonl'); + // Truncate any previous run's file so recovery never mixes runs. + writeFileSync(path, ''); + return { + path, + append(row) { + try { + appendFileSync(path, `${JSON.stringify(row)}\n`); + } catch { + // The sink must never fail a row. + } + }, + readRows() { + try { + return readFileSync(path, 'utf8') + .split('\n') + .filter(Boolean) + .flatMap((line) => { + try { + return [JSON.parse(line) as ReshapeRunRow]; + } catch { + return []; // torn final line from a hard crash + } + }); + } catch { + return []; + } + }, + }; +} + +/** Reshape sink rows back into per-iteration results, keeping only COMPLETE + * iterations (every expected row present) — the retired direct loop's + * guarantee that persisted iterations are whole and index-aligned, so + * never-run scenarios are not misreported as failures in a crash artifact. */ +function recoverCompleteIterations( + sinkRows: ReshapeRunRow[], + testCasesWithFiles: WorkflowTestCaseWithFile[], +): WorkflowTestCaseResult[][] { + const rowsPerIteration = roundRobinCaseRows(testCasesWithFiles).length; + if (rowsPerIteration === 0) return []; + const byIteration = new Map(); + for (const row of sinkRows) { + const inputs = row.run.inputs as { _iteration?: number } | undefined; + const iteration = inputs?._iteration ?? 0; + const group = byIteration.get(iteration) ?? []; + group.push(row); + byIteration.set(iteration, group); + } + const complete = [...byIteration.entries()] + .filter(([, rows]) => rows.length === rowsPerIteration) + .sort(([a], [b]) => a - b); + if (complete.length === 0) return []; + // Renumber to dense iterations so reshape doesn't stub the gaps. + const renumbered = complete.flatMap(([, rows], denseIndex) => + rows.map((row) => ({ + run: { + inputs: { ...(row.run.inputs ?? {}), _iteration: denseIndex }, + outputs: row.run.outputs, + }, + })), + ); + // Rebuild the judge-verdict side band from the verdicts each row embeds — + // reshape reads only this map (the in-memory one died with the run), and + // without it recovered rows would lose their expectation units. + const buildExpectationsByKey = new Map(); + for (const row of renumbered) { + const inputs = row.run.inputs as { _iteration?: number; testCaseFile?: string }; + const output = parseTargetOutput(row.run.outputs); + if (!inputs.testCaseFile || !output?.expectationResults?.length) continue; + const key = `${String(inputs._iteration ?? 0)}:${inputs.testCaseFile}`; + if (!buildExpectationsByKey.has(key)) + buildExpectationsByKey.set(key, output.expectationResults); + } + return reshapeLangSmithRuns( + renumbered, + testCasesWithFiles, + complete.length, + new Map(), + buildExpectationsByKey, + undefined, + new Map(), + ); +} + export interface EvalRunOutput { evaluation: MultiRunEvaluation; experimentName?: string; @@ -144,6 +254,10 @@ export interface PersistEvalConfig { commitSha: string | undefined; rerun: RerunHint | undefined; mcpBuildSpend: McpBuildSpend[]; + /** Journal of completed rows; read back to recover a crashed run. */ + rowSink?: RowSink; + /** Needed to reshape recovered sink rows on the crash path. */ + testCasesWithFiles?: WorkflowTestCaseWithFile[]; } export interface PersistedEval extends EvalRunOutput { @@ -193,10 +307,21 @@ export async function runEvalAndPersist( } finally { if (!persisted) { try { + // Prefer sink recovery — row-granular and fed by BOTH drivers; fall + // back to the direct driver's per-iteration channel. + const sinkRows = config.rowSink?.readRows() ?? []; + const recovered = + sinkRows.length > 0 && config.testCasesWithFiles + ? recoverCompleteIterations(sinkRows, config.testCasesWithFiles) + : []; + const runResults = recovered.length > 0 ? recovered : partialResults; const evaluation: MultiRunEvaluation = - partialResults.length > 0 - ? aggregateResults(partialResults, partialResults.length) + runResults.length > 0 + ? aggregateResults(runResults, runResults.length) : { totalRuns: config.iterations, testCases: [] }; + const recoverySlugByTestCase = config.testCasesWithFiles + ? new Map(config.testCasesWithFiles.map(({ testCase, fileSlug }) => [testCase, fileSlug])) + : undefined; const { jsonPath } = writeEvalResults( evaluation, Date.now() - config.startTime, @@ -204,14 +329,14 @@ export async function runEvalAndPersist( undefined, undefined, config.commitSha, - undefined, + recoverySlugByTestCase, config.rerun, undefined, config.mcpBuildSpend, undefined, ); config.logger.error( - `Eval run did not finish cleanly — wrote partial results (${String(partialResults.length)} iteration(s)) to ${jsonPath}`, + `Eval run did not finish cleanly — wrote partial results (${String(runResults.length)} iteration(s)) to ${jsonPath}`, ); } catch (writeError: unknown) { config.logger.error( diff --git a/packages/@n8n/instance-ai/evaluations/run/rows.ts b/packages/@n8n/instance-ai/evaluations/run/rows.ts new file mode 100644 index 00000000000..6766c398dab --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/run/rows.ts @@ -0,0 +1,51 @@ +// --------------------------------------------------------------------------- +// Row flattening — the single source of truth for which rows a run evaluates +// and in what order. The LangSmith dataset sync and the direct driver both +// flatten cases through here, so keyless (dispatcher) and CI runs can never +// drift to different row sets or orders. +// --------------------------------------------------------------------------- + +import type { WorkflowTestCaseWithFile } from '../data/workflows'; +import type { ExecutionScenario, WorkflowTestCase } from '../types'; + +/** Scenario name for the single "build-only" row a 0-scenario case emits, so + * the workflow still builds and its process/outcome expectations get judged. + * Shared by the dataset sync, both drivers and reshape so all agree on the + * sentinel. */ +export const BUILD_ONLY_SCENARIO_NAME = '__build_only__'; + +export interface CaseRow { + testCase: WorkflowTestCase; + testCaseFile: string; + /** null = the build-only sentinel row of a scenario-less case. */ + scenario: ExecutionScenario | null; +} + +/** + * Flatten test cases into rows ordered round-robin across cases: scenario #1 + * of every case, then scenario #2, …, then one build-only sentinel row per + * scenario-less case — so builds diversify early instead of burning all + * concurrency slots on one test case. + * + * Input: [tc1(s1,s2,s3), tc2(s1,s2), tc3()] + * Output: [tc1/s1, tc2/s1, tc1/s2, tc2/s2, tc1/s3, tc3/sentinel] + */ +export function roundRobinCaseRows(testCasesWithFiles: WorkflowTestCaseWithFile[]): CaseRow[] { + const rows: CaseRow[] = []; + const maxScenarios = Math.max( + ...testCasesWithFiles.map(({ testCase }) => (testCase.executionScenarios ?? []).length), + 0, + ); + for (let i = 0; i < maxScenarios; i++) { + for (const { testCase, fileSlug } of testCasesWithFiles) { + const scenario = testCase.executionScenarios?.[i]; + if (scenario) rows.push({ testCase, testCaseFile: fileSlug, scenario }); + } + } + for (const { testCase, fileSlug } of testCasesWithFiles) { + if ((testCase.executionScenarios?.length ?? 0) === 0) { + rows.push({ testCase, testCaseFile: fileSlug, scenario: null }); + } + } + return rows; +} diff --git a/packages/@n8n/instance-ai/evaluations/run/tiers.ts b/packages/@n8n/instance-ai/evaluations/run/tiers.ts new file mode 100644 index 00000000000..4b783fd7602 --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/run/tiers.ts @@ -0,0 +1,14 @@ +// --------------------------------------------------------------------------- +// Gated tiers — the only tier names that carry behavior. +// +// Tiers themselves are deliberately free-form: a case opts into any grouping +// via its `datasets` field, `--tier ` filters by it, and the name flows +// to LangSmith as an example split. There is no tier registry on purpose — +// a catalog nothing enforces would only drift. The single piece of tier +// knowledge the harness carries is which tiers assert the absolute pass@k +// green bar (comparison/gate.ts) instead of a baseline comparison; declare +// those here. +// --------------------------------------------------------------------------- + +/** Tiers whose runs assert the absolute gate instead of comparing to a baseline. */ +export const GATED_TIER_NAMES: ReadonlySet = new Set(['pr']);