feat(ai-builder): Harden the eval harness — crash-recovery journal, parity fixes, extension points (no-changelog) (#34747)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-08-28 17:22:01 +08:00 · 2026-07-23 12:25:55 +00:00
parent 66939681c9
commit b93a99945b
24 changed files with 1128 additions and 213 deletions
@@ -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 <manifest>` | skips the build; verifies existing workflows (score MCP/hand-built cohorts on the same verifier) |

@@ -1,3 +1,4 @@
 .output/
 eval-pr-comment.md
 eval-results.json
+eval-rows.jsonl
@@ -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 `<output-dir>/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.
@@ -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:
@@ -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<typeof import('../harness/runner')>();
-	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<typeof import('../build-expectations/verifier')>();
 	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' }),
+		);
+	});
+});
@@ -316,3 +316,141 @@ describe('createCasePipeline', () => {
 		expect(lane.tracedExecuteAgent).toHaveBeenCalledTimes(1);
 	});
 });
+
+function deferred(): { promise: Promise<unknown>; resolve: (v: unknown) => void } {
+	let resolve!: (v: unknown) => void;
+	const promise = new Promise<unknown>((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;
+	});
+});
@@ -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);
+	});
+});
@@ -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<typeof vi.fn> } {
+		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/);
+	});
+});
@@ -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<WrittenReport> {
+		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<WrittenReport>(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]);
+	});
+});
@@ -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 ?? '<sentinel>'}`)).toEqual([
+			'alpha/a1',
+			'beta/b1',
+			'alpha/a2',
+			'beta/b2',
+			'alpha/a3',
+			'gamma/<sentinel>',
+		]);
+	});
+
+	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([]);
+	});
+});
@@ -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,
@@ -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<void> {
@@ -52,6 +52,8 @@ async function main(): Promise<void> {
 		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<void> {
 					commitSha,
 					rerun: ciRerunHint(),
 					mcpBuildSpend,
+					rowSink,
+					testCasesWithFiles,
 				},
 				async (partialResults) => {
 					if (hasLangSmith) {
@@ -85,6 +89,7 @@ async function main(): Promise<void> {
 							cleanupBuiltWorkflows,
 							mcpBuildLogDir,
 							mcpBuildSpend,
+							rowSink,
 						});
 						return {
 							evaluation: langsmithRun.evaluation,
@@ -107,6 +112,7 @@ async function main(): Promise<void> {
 						mcpBuildLogDir,
 						mcpBuildSpend,
 						partialResults,
+						rowSink,
 					});
 					return { evaluation: directRun.evaluation, slugByTestCase: directRun.slugByTestCase };
 				},
@@ -128,18 +134,6 @@ async function main(): Promise<void> {
 	}
 }

-// ---------------------------------------------------------------------------
-// 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.
@@ -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);
@@ -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<LocalGatewayStatus> = 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[] {
@@ -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 --
@@ -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<string, Example>();
-	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<void> {
+	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
@@ -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<Awaited<ReturnType<typeof executeScenario>>>;
 	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 <type> 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)}`,
@@ -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<string, Promise<unknown>>();
+	async function withSerialSeeding<T>(key: string, fn: () => Promise<T>): Promise<T> {
+		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<TargetOutput> => {
 		const iteration = inputs._iteration ?? 0;
 		try {
@@ -140,7 +157,13 @@ export function createCasePipeline(deps: CasePipelineDeps): CasePipeline {
 		inputs: ScenarioRowInputs,
 		iteration: number,
 	): Promise<TargetOutput> => {
-		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<TargetOutput> => {
+			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 };
@@ -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<EvalSessionConfig, 'wrap'> {
 	/** 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<WorkflowTestCase, string>;
 }> {
-	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,
 		);

@@ -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(
@@ -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<TargetOutput> => {
+		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);
@@ -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<number, ReshapeRunRow[]>();
+	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<string, BuildExpectationResult[]>();
+	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(
@@ -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;
+}
@@ -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 <name>` 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<string> = new Set(['pr']);