diff --git a/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts b/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts
index d3b7641d..229b12a6 100644
--- a/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts
+++ b/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts
@@ -5,8 +5,9 @@
  *
  * Opening a workflow in the designer and saving it keeps what the designer
  * does not edit — a pipe's timeout, the workflow's version, error handler,
- * display name, and the trust its origin records. Saving used to drop all
- * five: an imported workflow came back `authored`, renamed to its id.
+ * display name, the trust its origin records, and a retry policy beyond its
+ * attempts. Saving used to drop each: an imported workflow came back
+ * `authored`, renamed to its id, with the engine's default retry.
  *
  * Needs the web app at E2E_BASE_URL and workflow-server on :8081 with no
  * service token (loopback dev); skips when the engine is not there.
@@ -30,7 +31,17 @@ const doc = (id: string, extra: Record<string, unknown> = {}) => ({
     {
       id: 'main',
       name: 'Main',
-      pipes: [{ id: 'start', type: 'source.triggerPayload', role: 'source', config: {}, timeoutMs: 1234 }],
+      pipes: [
+        {
+          id: 'start',
+          type: 'source.triggerPayload',
+          role: 'source',
+          config: {},
+          timeoutMs: 1234,
+          // Beyond the attempts the canvas edits: a save used to drop the rest.
+          retry: { attempts: 5, backoff: 'fixed', maxDelayMs: 900, jitter: false, on: 'all' },
+        },
+      ],
       edges: [],
     },
   ],
@@ -99,6 +110,7 @@ describe('workflow designer round trip', () => {
       errorHandler: { workflowId: HANDLER },
     });
     expect(after.pipelines[0].pipes[0].timeoutMs).toBe(1234);
+    expect(after.pipelines[0].pipes[0].retry).toEqual({ attempts: 5, backoff: 'fixed', maxDelayMs: 900, jitter: false, on: 'all' });
   });
 
   it('shows only the trigger sections that have a form, and no dead buttons', async (ctx) => {
@@ -117,4 +129,17 @@ describe('workflow designer round trip', () => {
     ]);
     await driver.screenshot({ path: process.env.E2E_SHOT ?? '/tmp/workflow-trigger-dialog.png' });
   });
+
+  it('keeps the section you are in while you type in a trigger', async (ctx) => {
+    // The dialog used to re-sync its selection from props in an effect; with
+    // the triggers among its dependencies, every keystroke snapped it back to
+    // General. Mounted only while open, it starts once and stays put.
+    if (!engineUp) ctx.skip();
+    const section = driver.locator('[data-testid="workflow-trigger-config-section-input parameters"]');
+    await section.click();
+    await driver.locator(`${byTestId('workflow-trigger-config')} .monaco-editor .view-lines`).click();
+    await driver.keyboard.type('{"region": "west"}');
+    expect(await section.getAttribute('class')).toContain('active');
+    expect(await driver.locator(`${byTestId('workflow-trigger-config')} .monaco-editor`).isVisible()).toBe(true);
+  });
 });
diff --git a/apps/web/src/frontend/features/workflow/components/CredentialsPanel.tsx b/apps/web/src/frontend/features/workflow/components/CredentialsPanel.tsx
index 52cff2ac..3f9a96d9 100644
--- a/apps/web/src/frontend/features/workflow/components/CredentialsPanel.tsx
+++ b/apps/web/src/frontend/features/workflow/components/CredentialsPanel.tsx
@@ -25,8 +25,8 @@ const _kindsMatch: SameMembers<(typeof KINDS)[number], CredentialKind> = true;
 const _sourcesMatch: SameMembers<(typeof SOURCES)[number], Exclude<CredentialSource, 'foxschema'>> = true;
 void [_kindsMatch, _sourcesMatch];
 
-type Kind = (typeof KINDS)[number];
-type Source = (typeof SOURCES)[number];
+type Kind = CredentialKind;
+type Source = Exclude<CredentialSource, 'foxschema'>;
 
 const KIND_FIELDS: Record<Kind, Array<{ key: string; label: string; secret?: boolean }>> = {
   database: [
diff --git a/apps/web/src/frontend/features/workflow/components/DelimitedSourceEditor.tsx b/apps/web/src/frontend/features/workflow/components/DelimitedSourceEditor.tsx
index c289381c..c0bf0baa 100644
--- a/apps/web/src/frontend/features/workflow/components/DelimitedSourceEditor.tsx
+++ b/apps/web/src/frontend/features/workflow/components/DelimitedSourceEditor.tsx
@@ -12,10 +12,11 @@ import { compileZodSchema } from '../lib/zodSchema';
 import type {
   ColumnCheck as EngineColumnCheck,
   ColumnCheckKind,
+  ColumnRule,
   ColumnType,
   FixedWidthColumn,
 } from '@foxschema/workflow-engine/definitions';
-import type { KnowsEveryField, SameMembers } from '../lib/engineShapes';
+import type { SameMembers } from '../lib/engineShapes';
 
 /**
  * Config editor for the delimited file sources (`source.file.csv`,
@@ -59,33 +60,14 @@ const COLUMN_TYPES = ['string', 'integer', 'number', 'boolean', 'date'] as const
 const CHECK_KINDS = ['null', 'length', 'format'] as const;
 const _columnTypesMatch: SameMembers<(typeof COLUMN_TYPES)[number], ColumnType> = true;
 const _checkKindsMatch: SameMembers<(typeof CHECK_KINDS)[number], ColumnCheckKind> = true;
-const _checksKnown: KnowsEveryField<ColumnCheck, EngineColumnCheck> = true;
-const _fixedKnown: KnowsEveryField<FixedColumn, FixedWidthColumn> = true;
-void [_columnTypesMatch, _checkKindsMatch, _checksKnown, _fixedKnown];
-
-interface ColumnCheck {
-  pattern: string;
-  kind?: string;
-  message?: string;
-}
-
-interface ColumnRuleFields {
-  type?: string;
-  /** Legacy single regex — migrated into `checks` in the editor when present. */
-  pattern?: string;
-  checks?: ColumnCheck[];
-  required?: boolean;
-}
+void [_columnTypesMatch, _checkKindsMatch];
 
-interface FixedColumn extends ColumnRuleFields {
-  name: string;
-  start: number;
-  length: number;
-}
-
-interface FieldRule extends ColumnRuleFields {
-  name: string;
-}
+// The engine's own column shapes, type-only. Hand-kept copies here typed a
+// column's `type` as any string, so a value the engine rejects compiled.
+type ColumnCheck = EngineColumnCheck;
+type ColumnRuleFields = Omit<ColumnRule, 'name'>;
+type FixedColumn = FixedWidthColumn;
+type FieldRule = ColumnRule;
 
 const parseChecks = (value: unknown): ColumnCheck[] => {
   if (!Array.isArray(value)) return [];
@@ -103,6 +85,10 @@ const parseChecks = (value: unknown): ColumnCheck[] => {
   });
 };
 
+/** A column type the engine knows, or undefined — the engine refuses any other. */
+const asColumnType = (value: unknown): ColumnType | undefined =>
+  (COLUMN_TYPES as readonly unknown[]).includes(value) ? (value as ColumnType) : undefined;
+
 /** Normalize rule fields: fold legacy `pattern` into `checks` for editing. */
 const ruleFields = (raw: ColumnRuleFields): ColumnRuleFields => {
   const checks = parseChecks(raw.checks);
@@ -111,7 +97,7 @@ const ruleFields = (raw: ColumnRuleFields): ColumnRuleFields => {
     checks.unshift({ pattern: legacy });
   }
   return {
-    ...(raw.type ? { type: String(raw.type) } : {}),
+    ...(asColumnType(raw.type) ? { type: asColumnType(raw.type) } : {}),
     ...(checks.length > 0 ? { checks } : {}),
     ...(raw.required ? { required: true } : {}),
   };
@@ -209,7 +195,7 @@ function ColumnRuleEditor({
         <select data-testid={`workflow-source-rule-type-${name}`}
           title="Value type — validated and converted"
           value={fields.type ?? 'string'}
-          onChange={(ev) => onFieldsChange({ ...fields, type: ev.target.value })}
+          onChange={(ev) => onFieldsChange({ ...fields, type: asColumnType(ev.target.value) })}
         >
           {COLUMN_TYPES.map((type) => (
             <option key={type} value={type}>
diff --git a/apps/web/src/frontend/features/workflow/components/HttpRequestEditor.tsx b/apps/web/src/frontend/features/workflow/components/HttpRequestEditor.tsx
index 1af6fcb2..c282dbd8 100644
--- a/apps/web/src/frontend/features/workflow/components/HttpRequestEditor.tsx
+++ b/apps/web/src/frontend/features/workflow/components/HttpRequestEditor.tsx
@@ -16,16 +16,17 @@ import type {
   HttpRequestDef as EngineRequest,
   HttpSession as EngineSession,
   HttpTokenRefresh as EngineTokenRefresh,
+  HttpAuth as EngineAuth,
+  HttpBody as EngineBody,
 } from '@foxschema/workflow-engine/definitions';
-import type { KnowsEveryField, SameMembers } from '../lib/engineShapes';
+import type { SameMembers } from '../lib/engineShapes';
 import type { CredentialMeta } from '../api/engineClient';
 import { compileZodSchema } from '../lib/zodSchema';
 
 /**
- * The editor's own shapes for the engine's HttpRequestDef. Values stay local
- * (importing them pulls zod into the bundle); each list and shape is checked
- * against the engine's types below, so a field the engine adds fails the build
- * instead of being dropped on the next edit.
+ * The request model is the engine's own (`HttpRequestDef` and its parts),
+ * imported type-only. A hand-kept copy here had already drifted: it lacked a
+ * token refresh's headers and body, so every edit dropped them.
  */
 export const HTTP_KV_OPS = [
   'eq',
@@ -37,7 +38,7 @@ export const HTTP_KV_OPS = [
   'or',
   'range',
 ] as const;
-export type HttpKvOp = (typeof HTTP_KV_OPS)[number];
+export type HttpKvOp = EngineKvOp;
 
 /** UI labels + value hints per operator; serialized as `key[op]=value` (eq = plain). */
 const OP_META: Record<HttpKvOp, { label: string; placeholder: string }> = {
@@ -51,22 +52,9 @@ const OP_META: Record<HttpKvOp, { label: string; placeholder: string }> = {
   range: { label: 'range', placeholder: 'min,max' },
 };
 
-export type HttpKv = {
-  key: string;
-  value: string;
-  enabled: boolean;
-  op?: HttpKvOp;
-  /** Legacy path mapping — no longer editable in the UI, preserved on load. */
-  valueFrom?: string;
-  keyFrom?: string;
-};
+export type HttpKv = EngineKv;
 
-export type HttpAuth =
-  | { type: 'none' }
-  | { type: 'bearer'; token: string }
-  | { type: 'basic'; username: string; password: string }
-  | { type: 'apiKey'; key: string; value: string; in: 'header' | 'query' }
-  | { type: 'credential'; credentialId: string };
+export type HttpAuth = EngineAuth;
 
 export const HTTP_BODY_FIELD_TYPES = [
   'string',
@@ -74,14 +62,9 @@ export const HTTP_BODY_FIELD_TYPES = [
   'boolean',
   'json',
 ] as const;
-export type HttpBodyFieldType = (typeof HTTP_BODY_FIELD_TYPES)[number];
+export type HttpBodyFieldType = EngineBodyFieldType;
 
-export type HttpBodyField = {
-  key: string;
-  type: HttpBodyFieldType;
-  value: string;
-  enabled: boolean;
-};
+export type HttpBodyField = EngineBodyField;
 
 /** Optional JSON Schema — the runtime validates the body against it before sending. */
 type BodySchema = Record<string, unknown> | undefined;
@@ -93,47 +76,13 @@ type BodySchema = Record<string, unknown> | undefined;
  */
 type BodyValidation = { schema?: BodySchema; schemaSource?: string };
 
-export type HttpBody =
-  | { mode: 'none' }
-  | ({ mode: 'json'; json: unknown } & BodyValidation)
-  | ({ mode: 'raw'; raw: string; contentType: string } & BodyValidation)
-  | ({ mode: 'form'; form: HttpKv[] } & BodyValidation)
-  | ({ mode: 'fields'; fields: HttpBodyField[] } & BodyValidation);
+export type HttpBody = EngineBody;
 
-export type HttpSession = {
-  enabled: boolean;
-  persistToCredential: boolean;
-};
+export type HttpSession = EngineSession;
 
-export type HttpTokenRefresh = {
-  enabled: boolean;
-  url: string;
-  method: 'POST' | 'PUT';
-  /**
-   * Not edited here, but carried through: dropping them on every edit sent
-   * the engine's default refresh body instead of the one written by hand.
-   */
-  headers?: EngineTokenRefresh['headers'];
-  body?: EngineTokenRefresh['body'];
-  accessTokenPath: string;
-  refreshTokenPath?: string;
-  expiresInPath: string;
-  onStatus: number[];
-  skewSeconds: number;
-};
+export type HttpTokenRefresh = EngineTokenRefresh;
 
-export type HttpRequestValue = {
-  url: string;
-  method: 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE' | 'HEAD';
-  query: HttpKv[];
-  headers: HttpKv[];
-  auth: HttpAuth;
-  body: HttpBody;
-  timeoutMs: number;
-  variables: Record<string, unknown>;
-  session: HttpSession;
-  tokenRefresh?: HttpTokenRefresh;
-};
+export type HttpRequestValue = EngineRequest;
 
 export const EMPTY_HTTP_REQUEST: HttpRequestValue = {
   url: 'https://',
@@ -152,6 +101,9 @@ const DEFAULT_TOKEN_REFRESH: HttpTokenRefresh = {
   enabled: false,
   url: 'https://',
   method: 'POST',
+  // The engine's own defaults: no extra headers, the standard refresh_token body.
+  headers: [],
+  body: { mode: 'json', json: { refresh_token: '{{secrets.refreshToken}}', grant_type: 'refresh_token' }, form: [] },
   accessTokenPath: 'access_token',
   expiresInPath: 'expires_in',
   onStatus: [401],
@@ -1232,14 +1184,11 @@ function BodySchemaEditor({
 }
 
 /** Normalize pipe/trigger JSON into the editor value (legacy Record headers ok). */
-const _kvOpsMatch: SameMembers<HttpKvOp, EngineKvOp> = true;
-const _bodyFieldTypesMatch: SameMembers<HttpBodyFieldType, EngineBodyFieldType> = true;
-const _kvKnown: KnowsEveryField<HttpKv, EngineKv> = true;
-const _bodyFieldKnown: KnowsEveryField<HttpBodyField, EngineBodyField> = true;
-const _sessionKnown: KnowsEveryField<HttpSession, EngineSession> = true;
-const _tokenRefreshKnown: KnowsEveryField<HttpTokenRefresh, EngineTokenRefresh> = true;
-const _requestKnown: KnowsEveryField<HttpRequestValue, EngineRequest> = true;
-void [_kvOpsMatch, _bodyFieldTypesMatch, _kvKnown, _bodyFieldKnown, _sessionKnown, _tokenRefreshKnown, _requestKnown];
+// The two value lists stay local (importing them pulls zod into the bundle);
+// a value the engine adds or drops fails the build here.
+const _kvOpsMatch: SameMembers<(typeof HTTP_KV_OPS)[number], EngineKvOp> = true;
+const _bodyFieldTypesMatch: SameMembers<(typeof HTTP_BODY_FIELD_TYPES)[number], EngineBodyFieldType> = true;
+void [_kvOpsMatch, _bodyFieldTypesMatch];
 
 export function toHttpRequestValue(input: unknown): HttpRequestValue {
   if (!input || typeof input !== 'object') return { ...EMPTY_HTTP_REQUEST };
@@ -1390,8 +1339,10 @@ function coerceTokenRefresh(value: unknown): HttpTokenRefresh | undefined {
     enabled: raw.enabled === true,
     url: typeof raw.url === 'string' ? raw.url : DEFAULT_TOKEN_REFRESH.url,
     method: raw.method === 'PUT' ? 'PUT' : 'POST',
-    ...(Array.isArray(raw.headers) ? { headers: raw.headers as EngineTokenRefresh['headers'] } : {}),
-    ...(raw.body && typeof raw.body === 'object' ? { body: raw.body as EngineTokenRefresh['body'] } : {}),
+    // Not edited here, but carried through: dropping them on every edit sent
+    // the engine's default refresh body instead of the one written by hand.
+    headers: Array.isArray(raw.headers) ? (raw.headers as HttpTokenRefresh['headers']) : DEFAULT_TOKEN_REFRESH.headers,
+    body: raw.body && typeof raw.body === 'object' ? (raw.body as HttpTokenRefresh['body']) : DEFAULT_TOKEN_REFRESH.body,
     accessTokenPath:
       typeof raw.accessTokenPath === 'string'
         ? raw.accessTokenPath
diff --git a/apps/web/src/frontend/features/workflow/components/PipeNode.tsx b/apps/web/src/frontend/features/workflow/components/PipeNode.tsx
index b360a813..39ec35fe 100644
--- a/apps/web/src/frontend/features/workflow/components/PipeNode.tsx
+++ b/apps/web/src/frontend/features/workflow/components/PipeNode.tsx
@@ -8,6 +8,8 @@
 import { Handle, Position, type Node, type NodeProps } from '@xyflow/react';
 import { ArrowDownToLine, ArrowUpFromLine, Shuffle, Zap } from 'lucide-react';
 import { portLabel } from '../lib/ports';
+import type { z } from 'zod';
+import type { retrySchema } from '@foxschema/workflow-engine/definitions';
 
 export type PortInfo = { name: string; type: string };
 
@@ -24,6 +26,8 @@ export type PipeData = {
   credentialId: string;
   /** Not edited on the canvas; kept so a save does not drop it. */
   timeoutMs?: number;
+  /** The pipe's whole retry policy as loaded; the canvas edits only `retryAttempts`. */
+  retry?: z.input<typeof retrySchema>;
   inputs: PortInfo[];
   outputs: PortInfo[];
 };
diff --git a/apps/web/src/frontend/features/workflow/components/WorkflowDesigner.tsx b/apps/web/src/frontend/features/workflow/components/WorkflowDesigner.tsx
index f71b687b..45b395fa 100644
--- a/apps/web/src/frontend/features/workflow/components/WorkflowDesigner.tsx
+++ b/apps/web/src/frontend/features/workflow/components/WorkflowDesigner.tsx
@@ -38,7 +38,8 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
 import { useAuthStore } from '@/platform/authStore';
 import { useUiStore } from '@/platform/uiStore';
 import { toast } from '../lib/notify';
-import type { PipeDef, PipelineDef, WorkflowDef } from '@foxschema/workflow-engine/definitions';
+import type { z } from 'zod';
+import type { WorkflowDef, pipeSchema, pipelineSchema } from '@foxschema/workflow-engine/definitions';
 import { api, type RunStartResult } from '../api/engineClient';
 import type { KnowsEveryField } from '../lib/engineShapes';
 import {
@@ -139,17 +140,13 @@ type PipelineCanvas = Pipeline & {
   edges: Edge[];
 };
 type Dependency = WorkflowDependency;
-type SavedPipe = {
-  id: string;
-  role: PipeData['role'];
-  type: string;
-  intent?: string;
-  config?: Record<string, unknown>;
-  concurrency?: number;
-  retry?: { attempts?: number };
-  credentialId?: string;
-  timeoutMs?: number;
-};
+/**
+ * What the designer saves for a pipe: the engine's input shape — what its
+ * schema parses, defaults applied on that side. A hand-kept copy knew only
+ * `retry.attempts`, so saving dropped a policy's backoff, delay, jitter and
+ * `on`, and the run fell back to the defaults.
+ */
+type SavedPipe = z.input<typeof pipeSchema>;
 /**
  * Fields of a workflow the designer does not edit. Kept from the document that
  * was opened and written back on save — dropping them reset an imported
@@ -159,13 +156,7 @@ type CarriedFields = Pick<WorkflowDef, 'version' | 'origin' | 'errorHandler'> &
 type SavedWorkflow = Partial<CarriedFields> & {
   id: string;
   name: string;
-  pipelines: Array<{
-    id: string;
-    name: string;
-    task?: string;
-    pipes: SavedPipe[];
-    edges?: Array<{ from: string; to: string; fromPort?: string }>;
-  }>;
+  pipelines: Array<z.input<typeof pipelineSchema>>;
   dependencies?: Dependency[];
   triggers?: WorkflowTrigger[];
   onOverlap?: OverlapPolicy;
@@ -178,10 +169,8 @@ type SavedWorkflow = Partial<CarriedFields> & {
   middleware?: Array<{ name: string; config: Record<string, unknown> }>;
 };
 
-const _pipeKnown: KnowsEveryField<SavedPipe, PipeDef> = true;
-const _pipelineKnown: KnowsEveryField<SavedWorkflow['pipelines'][number], PipelineDef> = true;
 const _workflowKnown: KnowsEveryField<SavedWorkflow, WorkflowDef> = true;
-void [_pipeKnown, _pipelineKnown, _workflowKnown];
+void _workflowKnown;
 
 function parsePipeConfig(node: PipeNodeType): Record<string, unknown> {
   try {
@@ -200,8 +189,9 @@ function serializePipe(node: PipeNodeType): SavedPipe {
     ...(intent ? { intent } : {}),
     config: parsePipeConfig(node),
     concurrency: node.data.concurrency,
-    ...(node.data.retryAttempts !== 3
-      ? { retry: { attempts: node.data.retryAttempts } }
+    // The canvas edits only `attempts`; the rest of a policy rides along.
+    ...(node.data.retry || node.data.retryAttempts !== 3
+      ? { retry: { ...node.data.retry, attempts: node.data.retryAttempts } }
       : {}),
     ...(node.data.credentialId
       ? { credentialId: node.data.credentialId }
@@ -306,6 +296,7 @@ function toPipelineCanvas(
           config: JSON.stringify(pipe.config ?? {}, null, 2),
           concurrency: pipe.concurrency ?? 1,
           retryAttempts: pipe.retry?.attempts ?? 3,
+          ...(pipe.retry ? { retry: pipe.retry } : {}),
           credentialId: pipe.credentialId ?? '',
           ...(pipe.timeoutMs !== undefined ? { timeoutMs: pipe.timeoutMs } : {}),
           inputs: ports?.inputs ?? [],
diff --git a/apps/workflow-server/src/preview.test.ts b/apps/workflow-server/src/preview.test.ts
index 91dc51ae..cb59ddcb 100644
--- a/apps/workflow-server/src/preview.test.ts
+++ b/apps/workflow-server/src/preview.test.ts
@@ -105,6 +105,18 @@ describe('POST /api/preview/parse', () => {
     ]);
   });
 
+  it('numbers CSV records by the line they start on, past a quoted field that spans lines', async () => {
+    // preamble + header + index put the third record on line 5; it starts on 7.
+    const result = await parse({
+      kind: 'csv',
+      config: { skipLines: 1 },
+      sample: 'exported yesterday\nid,note\n1,"two\nlines"\n2,"three\nmore\nlines"\n3,plain\n',
+    });
+    expect(result.error).toBeNull();
+    expect(result.records.map((record) => record.id)).toEqual(['1', '2', '3']);
+    expect(result.lines).toEqual([3, 5, 8]);
+  });
+
   it('previews multi-delimiter text with offset and flat map', async () => {
     const result = await parse({
       kind: 'text',
diff --git a/apps/workflow-server/src/routes/preview.ts b/apps/workflow-server/src/routes/preview.ts
index a1797fcc..519f7bd6 100644
--- a/apps/workflow-server/src/routes/preview.ts
+++ b/apps/workflow-server/src/routes/preview.ts
@@ -5,7 +5,6 @@
  *
  * Workflow engine — moved from FoxAgent (apps/api/src/routes/preview.ts).
  */
-import { open } from 'node:fs/promises';
 import { z } from 'zod';
 import type { FastifyPluginAsync } from 'fastify';
 import type { ZodTypeProvider } from '../zod-provider.js';
@@ -16,7 +15,7 @@ import {
   parseTextRows,
   textConfigFields,
   textFormatRefine,
-  resolveWorkflowFile,
+  openWorkflowFile,
 } from '@foxschema/workflow-engine';
 
 /** At most this much of a file is read for a preview. */
@@ -49,7 +48,7 @@ const previewResponseSchema = z.object({
   records: z.array(z.record(z.string(), z.unknown())),
   /**
    * 1-based source line for each visible record (parallel to `records`).
-   * Text uses the physical file/sample line; CSV uses the record ordinal.
+   * The physical file/sample line the record starts on, for text and CSV alike.
    */
   lines: z.array(z.number().int().min(1)).default([]),
   invalid: z.array(
@@ -78,31 +77,6 @@ const failure = (error: string): PreviewResponse => ({
   error,
 });
 
-/**
- * Head of the file as text. When the file is larger than the window, the last
- * (possibly partial) line is dropped so the parser only sees complete lines.
- */
-async function readHead(
-  path: string,
-): Promise<{ text: string; truncated: boolean }> {
-  // eslint-disable-next-line security/detect-non-literal-fs-filename -- the workflow author's configured source file — the same path a run reads; the FoxSchema proxy requires workflow.design
-  const handle = await open(path, 'r');
-  try {
-    const { size } = await handle.stat();
-    const buffer = Buffer.alloc(Math.min(size, HEAD_BYTES));
-    await handle.read(buffer, 0, buffer.length, 0);
-    let text = buffer.toString('utf8');
-    const truncated = size > HEAD_BYTES;
-    if (truncated) {
-      const lastNewline = text.lastIndexOf('\n');
-      if (lastNewline !== -1) text = text.slice(0, lastNewline);
-    }
-    return { text, truncated };
-  } finally {
-    await handle.close();
-  }
-}
-
 /**
  * Parse a sample (or the head of the configured file) with the same code the
  * source pipes run, so the designer preview shows exactly what a run would
@@ -140,7 +114,11 @@ export function previewRoutes(): FastifyPluginAsync {
           const path = parsed.data.path;
           if (!path) return failure('provide a sample or set a file path');
           try {
-            ({ text, truncated: headTruncated } = await readHead(await resolveWorkflowFile(path)));
+            // The head of the file, through the same root-confined handle a
+            // run reads with; a partial last line is dropped.
+            ({ text, truncated: headTruncated } = await (await openWorkflowFile(path)).readHead(HEAD_BYTES, {
+              wholeLines: true,
+            }));
           } catch (error) {
             return failure(`read ${path}: ${(error as Error).message}`);
           }
@@ -159,11 +137,10 @@ export function previewRoutes(): FastifyPluginAsync {
           }[] = [];
           if (kind === 'csv') {
             const csv = csvConfigSchema.parse(config);
-            ({ headers: fields, records } = await parseCsvRecords(text, csv));
-            // Physical file line: skipped preamble + header row + 1-based data.
-            recordLines = records.map(
-              (_, index) => csv.skipLines + index + 2,
-            );
+            // Where each record starts, read by the scanner the run uses: a
+            // count of preamble + header + index was wrong after any quoted
+            // field that spans lines.
+            ({ headers: fields, records, lines: recordLines } = await parseCsvRecords(text, csv));
           } else {
             const opts = textConfigSchema.parse(config);
             ({
diff --git a/packages/workflow-engine/src/index.ts b/packages/workflow-engine/src/index.ts
index e8fc994c..93c5a4b2 100644
--- a/packages/workflow-engine/src/index.ts
+++ b/packages/workflow-engine/src/index.ts
@@ -33,6 +33,8 @@ export {
   ConditionPipe,
   createDefaultPipeRegistry,
   resolveWorkflowFile,
+  openWorkflowFile,
+  type WorkflowFile,
   workflowFilesRoot,
   FILES_DIR_ENV,
 } from './pipes/utility/index.js';
diff --git a/packages/workflow-engine/src/pipes/pipes-fs-boundary.test.ts b/packages/workflow-engine/src/pipes/pipes-fs-boundary.test.ts
new file mode 100644
index 00000000..78212acb
--- /dev/null
+++ b/packages/workflow-engine/src/pipes/pipes-fs-boundary.test.ts
@@ -0,0 +1,42 @@
+/**
+ * Fox Schema (foxschema)
+ * Copyright 2024-2026 Huy Phan <huyplb@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * File pipes reach the filesystem only through `openWorkflowFile`, which
+ * resolves the configured path inside the workflow files root first. A pipe
+ * that imported `node:fs` itself would read or write wherever its config
+ * pointed — `/etc/passwd`, `~/.ssh/authorized_keys` — as the engine's user,
+ * and nothing else would notice.
+ */
+import { readdirSync, readFileSync, statSync } from 'node:fs';
+import { join, relative } from 'node:path';
+import { describe, expect, it } from 'vitest';
+
+const PIPES = __dirname;
+
+/** The only files under pipes/ that may touch the filesystem directly. */
+const ALLOWED = new Set(['utility/file-root.ts', 'utility/workflow-files.ts']);
+
+function sources(dir: string): string[] {
+  return readdirSync(dir).flatMap((name) => {
+    const path = join(dir, name);
+    if (statSync(path).isDirectory()) return sources(path);
+    return /\.ts$/.test(name) && !/\.test\.ts$/.test(name) ? [path] : [];
+  });
+}
+
+const FS_IMPORT = /(?:from\s+|import\(\s*|require\(\s*)['"](?:node:)?fs(?:\/promises)?['"]/;
+
+describe('file pipes and the filesystem', () => {
+  it('finds the pipe sources it guards', () => {
+    expect(sources(PIPES).length).toBeGreaterThan(20);
+  });
+
+  it('lets only the root-confined module import node:fs', () => {
+    const offenders = sources(PIPES)
+      .map((path) => relative(PIPES, path))
+      .filter((rel) => !ALLOWED.has(rel) && FS_IMPORT.test(readFileSync(join(PIPES, rel), 'utf8')));
+    expect(offenders).toEqual([]);
+  });
+});
diff --git a/packages/workflow-engine/src/pipes/utility/batch-emitter.test.ts b/packages/workflow-engine/src/pipes/utility/batch-emitter.test.ts
new file mode 100644
index 00000000..0bbed9cd
--- /dev/null
+++ b/packages/workflow-engine/src/pipes/utility/batch-emitter.test.ts
@@ -0,0 +1,66 @@
+/**
+ * Fox Schema (foxschema)
+ * Copyright 2024-2026 Huy Phan <huyplb@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The batching CSV, text and JSON sources share. Each kept its own copy
+ * before, and the copies had drifted on exactly these points.
+ */
+import { describe, expect, it } from 'vitest';
+import type { RecordBatch } from '../../registry/index.js';
+import { BatchEmitter, type EmitterPositions } from './delimited.js';
+
+const ROWS: EmitterPositions = { cursorKey: 'row', positionKey: '_row', label: 'CSV row', cut: 'record' };
+const LINES: EmitterPositions = { cursorKey: 'line', positionKey: '_line', label: 'line', cut: 'boundary' };
+const schema = { type: 'object', properties: { n: { type: 'integer' } }, required: ['n'] };
+
+const shape = (batches: Iterable<RecordBatch>) =>
+  [...batches].map((b) => ({ id: b.id, port: b.port ?? 'out', cursor: b.cursor, size: b.records.length }));
+
+describe('BatchEmitter', () => {
+  it('cuts data and rejects at batchSize as they fill, in the order they filled', () => {
+    const emitter = new BatchEmitter('src', { batchSize: 2, onInvalid: 'reject', schema }, ROWS);
+    const out = [];
+    for (const [n, row] of [[1, 1], ['x', 2], [2, 3], ['y', 4], [3, 5]] as const) {
+      emitter.add({ n }, row);
+      out.push(...emitter.drain());
+    }
+    out.push(...emitter.flush(5));
+    expect(shape(out)).toEqual([
+      { id: 'src:0:1-3', port: 'out', cursor: { row: 3 }, size: 2 },
+      { id: 'src:rejects:2-4', port: 'rejects', cursor: undefined, size: 2 },
+      { id: 'src:0:5-5', port: 'out', cursor: { row: 5 }, size: 1 },
+    ]);
+  });
+
+  it('in boundary mode, never splits what one line produced across two batches', () => {
+    const emitter = new BatchEmitter('src', { batchSize: 2, onInvalid: 'skip' }, LINES);
+    // Line 1 flat-maps to three records: one batch of three, not two and one.
+    for (const n of [1, 2, 3]) emitter.add({ n }, 1);
+    expect(shape(emitter.drain())).toEqual([{ id: 'src:0:1-1', port: 'out', cursor: { line: 1 }, size: 3 }]);
+  });
+
+  it('points the final cursor past trailing refused rows, so a resume does not read them again', () => {
+    const emitter = new BatchEmitter('src', { batchSize: 10, onInvalid: 'skip', schema }, ROWS);
+    emitter.add({ n: 1 }, 1);
+    emitter.add({ n: 'bad' }, 2);
+    emitter.add({ n: 'bad' }, 3);
+    expect(shape(emitter.flush(3))).toEqual([{ id: 'src:0:1-3', port: 'out', cursor: { row: 3 }, size: 1 }]);
+  });
+
+  it('fails with the position as the source names it, and why', () => {
+    const csv = new BatchEmitter('src', { batchSize: 10, onInvalid: 'fail', schema }, ROWS);
+    expect(() => csv.add({ n: 'x' }, 7)).toThrow(/^CSV row 7 failed schema: /);
+    const text = new BatchEmitter('src', { batchSize: 10, onInvalid: 'fail' }, LINES);
+    expect(() => text.add({}, 3, { reason: 'validation', message: 'qty: not a number' })).toThrow(
+      'line 3 failed validation: qty: not a number',
+    );
+  });
+
+  it('keeps a rejected record as the caller asks, with its error and position', () => {
+    const emitter = new BatchEmitter('src', { batchSize: 10, onInvalid: 'reject' }, LINES);
+    emitter.add({ qty: Number.NaN }, 4, { reason: 'validation', message: 'qty: not a number' }, { qty: 'ten' });
+    const [rejects] = [...emitter.flush(4)];
+    expect(rejects!.records).toEqual([{ qty: 'ten', _error: 'qty: not a number', _line: 4 }]);
+  });
+});
diff --git a/packages/workflow-engine/src/pipes/utility/csv-parse.test.ts b/packages/workflow-engine/src/pipes/utility/csv-parse.test.ts
index 3ae4102b..33c75331 100644
--- a/packages/workflow-engine/src/pipes/utility/csv-parse.test.ts
+++ b/packages/workflow-engine/src/pipes/utility/csv-parse.test.ts
@@ -54,10 +54,10 @@ async function* referenceParse(chunks: Iterable<string>, delimiter: string): Asy
   }
 }
 
-async function collect(rows: AsyncIterable<string[]>): Promise<string[][] | string> {
+async function collect(rows: AsyncIterable<string[] | { values: string[] }>): Promise<string[][] | string> {
   const out: string[][] = [];
   try {
-    for await (const row of rows) out.push(row);
+    for await (const row of rows) out.push(Array.isArray(row) ? row : row.values);
   } catch (error) {
     return `error: ${(error as Error).message}`;
   }
@@ -115,4 +115,18 @@ describe('CSV scanning', () => {
       expect(await collect(parseCsv([text], delimiter)), `seed ${seed}, one chunk`).toEqual(expected);
     }
   });
+
+  it('reports the physical line each row starts on, across quoted line breaks and chunks', async () => {
+    const text = 'a,b\n"two\nlines",x\nplain,y\r\n"three\nmore\nlines",z\nlast,w';
+    const lines: number[] = [];
+    for await (const { line } of parseCsv(chunkedFixed(text, 3), ',')) lines.push(line);
+    // a,b on 1 · "two↵lines" starts 2 · plain 4 · "three↵more↵lines" 5 · last 8
+    expect(lines).toEqual([1, 2, 4, 5, 8]);
+  });
 });
+
+function chunkedFixed(text: string, size: number): string[] {
+  const parts: string[] = [];
+  for (let at = 0; at < text.length; at += size) parts.push(text.slice(at, at + size));
+  return parts;
+}
diff --git a/packages/workflow-engine/src/pipes/utility/csv.ts b/packages/workflow-engine/src/pipes/utility/csv.ts
index 3a5817e5..0c840d07 100644
--- a/packages/workflow-engine/src/pipes/utility/csv.ts
+++ b/packages/workflow-engine/src/pipes/utility/csv.ts
@@ -5,7 +5,6 @@
  *
  * Workflow engine — moved from FoxAgent (packages/pipes/utility/src/csv.ts).
  */
-import { createReadStream } from 'node:fs';
 import * as z from 'zod';
 import type {
   PipeContext,
@@ -20,9 +19,8 @@ import {
   rejectPolicyField,
   type PipeMetadata,
 } from '../../sdk/index.js';
-import { validateAgainstSchema } from '../../common/index.js';
-import { rejectBatch } from './delimited.js';
-import { resolveWorkflowFile } from './file-root.js';
+import { BatchEmitter } from './delimited.js';
+import { openWorkflowFile } from './workflow-files.js';
 
 /** Exported so the preview endpoint parses with the pipe's own field shape. */
 export const csvConfigFields = z.object({
@@ -64,66 +62,20 @@ export class CsvSourcePipe implements SourcePipe {
   async *read(context: PipeContext): AsyncIterable<RecordBatch> {
     const config = csvConfigFields.parse(context.pipe.config);
     const resumeAfter = Number(context.checkpoint?.cursor.row ?? 0);
-    let skipped = 0;
-    let headers: string[] | undefined;
+    const emitter = new BatchEmitter(context.pipe.id, config, {
+      cursorKey: 'row',
+      positionKey: '_row',
+      label: 'CSV row',
+      cut: 'record',
+    });
     let rowNumber = 0;
-    let records: Record<string, unknown>[] = [];
-    let firstRow = 0;
-    let rejects: Record<string, unknown>[] = [];
-    let firstReject = 0;
-    let lastReject = 0;
-
-    for await (const values of parseCsv(
-      // eslint-disable-next-line security/detect-non-literal-fs-filename -- the source file path is this pipe's own configuration, confined to the files root
-      createReadStream(await resolveWorkflowFile(config.path), { encoding: 'utf8' }),
-      config.delimiter,
-    )) {
-      if (skipped < config.skipLines) {
-        skipped++;
-        continue;
-      }
-      if (!headers) {
-        headers = deriveHeaders(values);
-        continue;
-      }
-      rowNumber++;
-      if (rowNumber <= resumeAfter) continue;
-      const record = Object.fromEntries(
-        headers.map((header, index) => [header, values[index] ?? '']),
-      );
-      if (config.schema) {
-        const message = validateAgainstSchema(config.schema, record);
-        if (message !== null) {
-          if (config.onInvalid === 'fail') {
-            throw new Error(`CSV row ${rowNumber} failed schema: ${message}`);
-          }
-          if (config.onInvalid === 'reject') {
-            if (rejects.length === 0) firstReject = rowNumber;
-            lastReject = rowNumber;
-            rejects.push({ ...record, _error: message, _row: rowNumber });
-            if (rejects.length === config.batchSize) {
-              yield rejectBatch(context.pipe.id, firstReject, lastReject, rejects);
-              rejects = [];
-            }
-          }
-          continue; // skip/reject — the row leaves the data stream either way
-        }
-      }
-      if (records.length === 0) firstRow = rowNumber;
-      records.push(record);
-      if (records.length === config.batchSize) {
-        yield makeBatch(context.pipe.id, firstRow, rowNumber, records);
-        records = [];
-      }
-    }
-    // Rejects flush before the final data batch so the run's last checkpoint
-    // (data cursors only) still covers everything that was emitted.
-    if (rejects.length > 0) {
-      yield rejectBatch(context.pipe.id, firstReject, lastReject, rejects);
-    }
-    if (records.length > 0) {
-      yield makeBatch(context.pipe.id, firstRow, rowNumber, records);
+    for await (const { record, row } of csvRows((await openWorkflowFile(config.path)).readStream(), config)) {
+      rowNumber = row;
+      if (row <= resumeAfter) continue;
+      emitter.add(record, row);
+      yield* emitter.drain();
     }
+    yield* emitter.flush(rowNumber);
   }
 }
 
@@ -141,45 +93,25 @@ export function deriveHeaders(values: string[]): string[] {
 }
 
 /**
- * Parse a full CSV text block into records — the preview path. Shares
- * `parseCsv`/`deriveHeaders` with the streaming pipe so preview and run agree.
+ * Parse a full CSV text block into records — the preview path, through the
+ * same {@link csvRows} the run reads with. `lines` is where each record
+ * starts in the text.
  */
 export async function parseCsvRecords(
   text: string,
   options: { delimiter?: string; skipLines?: number } = {},
-): Promise<{ headers: string[]; records: Record<string, string>[] }> {
-  const { delimiter = ',', skipLines = 0 } = options;
-  let skipped = 0;
-  let headers: string[] | undefined;
+): Promise<{ headers: string[]; records: Record<string, string>[]; lines: number[] }> {
+  let headers: string[] = [];
   const records: Record<string, string>[] = [];
-  for await (const values of parseCsv([text], delimiter)) {
-    if (skipped < skipLines) {
-      skipped++;
-      continue;
-    }
-    if (!headers) {
-      headers = deriveHeaders(values);
-      continue;
-    }
-    records.push(
-      Object.fromEntries(headers.map((header, index) => [header, values[index] ?? ''])),
-    );
+  const lines: number[] = [];
+  const rows = csvRows([text], { delimiter: options.delimiter ?? ',', skipLines: options.skipLines ?? 0 }, (found) => {
+    headers = found;
+  });
+  for await (const { record, line } of rows) {
+    records.push(record);
+    lines.push(line);
   }
-  return { headers: headers ?? [], records };
-}
-
-function makeBatch(
-  pipeId: string,
-  firstRow: number,
-  lastRow: number,
-  records: Record<string, unknown>[],
-): RecordBatch {
-  return {
-    id: `${pipeId}:0:${firstRow}-${lastRow}`,
-    partitionId: '0',
-    records,
-    cursor: { row: lastRow },
-  };
+  return { headers, records, lines };
 }
 
 /**
@@ -194,11 +126,15 @@ function makeBatch(
 export async function* parseCsv(
   chunks: AsyncIterable<string | Buffer> | Iterable<string>,
   delimiter: string,
-): AsyncIterable<string[]> {
+): AsyncIterable<{ values: string[]; line: number }> {
   let field = '';
   let row: string[] = [];
   let quoted = false;
   let pendingQuote = false;
+  // Physical lines: every `\n` read, inside quotes too, so a row's `line` is
+  // where it starts in the file even after a multi-line quoted field.
+  let newlines = 0;
+  let rowStart = 1;
 
   for await (const chunk of chunks) {
     const text = String(chunk);
@@ -217,11 +153,10 @@ export async function* parseCsv(
       }
       if (quoted) {
         const close = text.indexOf('"', at);
-        if (close === -1) {
-          field += text.slice(at);
-          break;
-        }
-        field += text.slice(at, close);
+        const end = close === -1 ? length : close;
+        newlines += countNewlines(text, at, end);
+        field += text.slice(at, end);
+        if (close === -1) break;
         pendingQuote = true;
         at = close + 1;
         continue;
@@ -236,7 +171,9 @@ export async function* parseCsv(
         at++;
       } else if (char === '\n') {
         row.push(field.replace(/\r$/, ''));
-        yield row;
+        newlines++;
+        yield { values: row, line: rowStart };
+        rowStart = newlines + 1;
         field = '';
         row = [];
         at++;
@@ -256,6 +193,50 @@ export async function* parseCsv(
   if (quoted && !pendingQuote) throw new Error('unterminated quoted CSV field');
   if (field.length > 0 || row.length > 0) {
     row.push(field.replace(/\r$/, ''));
-    yield row;
+    yield { values: row, line: rowStart };
+  }
+}
+
+function countNewlines(text: string, from: number, to: number): number {
+  let count = 0;
+  for (let at = text.indexOf('\n', from); at !== -1 && at < to; at = text.indexOf('\n', at + 1)) count++;
+  return count;
+}
+
+/** One CSV record: its 1-based ordinal after the header, and the physical line it starts on. */
+export interface CsvRow {
+  record: Record<string, string>;
+  row: number;
+  line: number;
+}
+
+/**
+ * The records a CSV source reads, after its preamble and header — the one
+ * reading the run and the designer preview share, so they cannot disagree on
+ * which row is which. `onHeaders` hears the column names even when no record
+ * follows them.
+ */
+export async function* csvRows(
+  chunks: AsyncIterable<string | Buffer> | Iterable<string>,
+  options: { delimiter: string; skipLines: number },
+  onHeaders?: (headers: string[]) => void,
+): AsyncIterable<CsvRow> {
+  let skipped = 0;
+  let headers: string[] | undefined;
+  let row = 0;
+  for await (const { values, line } of parseCsv(chunks, options.delimiter)) {
+    if (skipped < options.skipLines) {
+      skipped++;
+      continue;
+    }
+    if (!headers) {
+      headers = deriveHeaders(values);
+      onHeaders?.(headers);
+      continue;
+    }
+    row++;
+    const record: Record<string, string> = {};
+    for (let index = 0; index < headers.length; index++) record[headers[index]!] = values[index] ?? '';
+    yield { record, row, line };
   }
 }
diff --git a/packages/workflow-engine/src/pipes/utility/delimited.ts b/packages/workflow-engine/src/pipes/utility/delimited.ts
index 403c406f..088a2a45 100644
--- a/packages/workflow-engine/src/pipes/utility/delimited.ts
+++ b/packages/workflow-engine/src/pipes/utility/delimited.ts
@@ -451,10 +451,72 @@ export function parseLine(
   return lineToRecords(line, options);
 }
 
+/** One record a text line produced, and what the column rules made of it. */
+export interface TextRow {
+  /** The values as split, before any rule converted them. */
+  raw: Record<string, string>;
+  record: Record<string, unknown>;
+  /** Why a column rule refused it, or null. */
+  failure: string | null;
+  /** 1-based line it came from. */
+  line: number;
+}
+
+/**
+ * Text records, line by line: offset, empty lines, the header, the flat map
+ * and the column rules — the one reading the run and the designer preview
+ * share, so the preview shows what a run will read. `push` takes the next
+ * line and returns the records it produced (none for a skipped line).
+ */
+export class TextRowReader {
+  private lineNumber = 0;
+  private headerDone: boolean;
+  private readonly rules: ReturnType<typeof columnRules>;
+  private readonly compiled: ReturnType<typeof compileColumnRules>;
+
+  /** `resumeAfter`: lines a checkpoint already covers, which also puts the header behind it. */
+  constructor(
+    private readonly options: TextParseOptions,
+    private readonly resumeAfter = 0,
+  ) {
+    this.headerDone = (options.header ?? 'none') === 'none' || resumeAfter > 0;
+    this.rules = columnRules(options);
+    this.compiled = compileColumnRules(this.rules);
+  }
+
+  /** The last line number pushed. */
+  get line(): number {
+    return this.lineNumber;
+  }
+
+  push(text: string): TextRow[] {
+    this.lineNumber++;
+    const { offset = 0, skipEmpty = true, header = 'none' } = this.options;
+    if (this.lineNumber <= offset || this.lineNumber <= this.resumeAfter) return [];
+    if (skipEmpty && text.trim() === '') return [];
+    const parsed = parseLine(text, this.options);
+    if (!this.headerDone) {
+      this.headerDone = true;
+      if (header === 'skip') return [];
+      // auto: drop only when the line names its own columns.
+      if (parsed[0] && isHeaderRecord(parsed[0], configuredNames(this.options))) return [];
+    }
+    return parsed.map((raw) => {
+      const applied = applyColumnRules(raw, this.rules, this.compiled);
+      return {
+        raw,
+        record: applied.record,
+        failure: applied.failures.length > 0 ? describeFailures(applied.failures) : null,
+        line: this.lineNumber,
+      };
+    });
+  }
+}
+
 /**
  * Parse a whole text block and apply per-column rules, reporting both the
  * converted records and every column failure by record index — the preview
- * path, which must show what the run would produce *and* what it would reject.
+ * path, read by the same {@link TextRowReader} as a run.
  *
  * `lines[i]` is the 1-based source line for `records[i]`, so schema failures
  * (which only know the record index) can still report which row failed.
@@ -467,43 +529,20 @@ export function parseTextRows(
   invalid: InvalidRecord[];
   lines: number[];
 } {
-  const { offset = 0, skipEmpty = true, header = 'none' } = options;
-  const rules = columnRules(options);
-  const compiled = compileColumnRules(rules);
   // Lines as the run reads them (`readLines`): split on `\n`, a trailing `\r`
   // dropped, no line after a final newline. Splitting on a bare `\r` too made
   // a classic-Mac file preview as many rows and then run as one.
   const sourceLines = text.split('\n').map((line) => line.replace(/\r$/, ''));
   if (sourceLines[sourceLines.length - 1] === '') sourceLines.pop();
+  const reader = new TextRowReader(options);
   const records: Record<string, unknown>[] = [];
   const invalid: InvalidRecord[] = [];
   const lines: number[] = [];
-  let headerDone = header === 'none';
-
-  for (let i = offset; i < sourceLines.length; i++) {
-    const lineNumber = i + 1;
-    const line = sourceLines[i]!;
-    if (skipEmpty && line.trim() === '') continue;
-    const parsed = parseLine(line, options);
-    if (!headerDone) {
-      headerDone = true;
-      if (header === 'skip') continue;
-      // auto: drop only when the line names its own columns.
-      if (parsed[0] && isHeaderRecord(parsed[0], configuredNames(options))) {
-        continue;
-      }
-    }
-    for (const raw of parsed) {
-      const applied = applyColumnRules(raw, rules, compiled);
-      if (applied.failures.length > 0) {
-        invalid.push({
-          index: records.length,
-          line: lineNumber,
-          message: describeFailures(applied.failures),
-        });
-      }
-      records.push(applied.record);
-      lines.push(lineNumber);
+  for (const line of sourceLines) {
+    for (const row of reader.push(line)) {
+      if (row.failure !== null) invalid.push({ index: records.length, line: row.line, message: row.failure });
+      records.push(row.record);
+      lines.push(row.line);
     }
   }
   return { records, invalid, lines };
@@ -538,6 +577,128 @@ export function rejectBatch(
   };
 }
 
+/** How a source names a position in its errors and on rejected rows. */
+export interface EmitterPositions {
+  /** Checkpoint cursor key: `row`, `line`, `index`. */
+  cursorKey: string;
+  /** Key the position is stored under on a rejected row: `_row`, `_line`, `_index`. */
+  positionKey: string;
+  /** How an error names the position: `CSV row`, `line`, `index`. */
+  label: string;
+  /**
+   * `record` cuts a batch the moment it is full. `boundary` cuts only in
+   * {@link BatchEmitter.drain}: a text line can flat-map to several records
+   * and its cursor is the line, so a line must never straddle two batches.
+   */
+  cut: 'record' | 'boundary';
+}
+
+/** Why a record was refused before it reached the data stream. */
+export type RecordFailure = { reason: 'to parse' | 'validation' | 'schema'; message: string };
+
+/**
+ * The batching every file source shares: applies `onInvalid` to records that
+ * failed, cuts data batches (with a cursor) and reject batches (cursor-less,
+ * on the `rejects` port) at `batchSize`, and flushes rejects before the final
+ * data batch so the run's last checkpoint covers everything emitted.
+ *
+ * CSV, text and JSON each kept their own copy of this, and the copies had
+ * drifted — where they cut, what the last cursor named, how errors read.
+ */
+export class BatchEmitter {
+  private records: Record<string, unknown>[] = [];
+  private first = 0;
+  private last = 0;
+  private rejects: Record<string, unknown>[] = [];
+  private firstReject = 0;
+  private lastReject = 0;
+  private pending: RecordBatch[] = [];
+
+  constructor(
+    private readonly pipeId: string,
+    private readonly config: {
+      batchSize: number;
+      onInvalid: 'fail' | 'skip' | 'reject';
+      schema?: Record<string, unknown>;
+    },
+    private readonly positions: EmitterPositions,
+  ) {}
+
+  /**
+   * One record at `position`. `failure` is set when it already failed (did not
+   * parse, a column rule refused it); otherwise the schema, if any, decides.
+   * A rejected record is stored as `rejectAs` (the raw values, say) when given.
+   */
+  add(
+    record: Record<string, unknown>,
+    position: number,
+    failure: RecordFailure | null = null,
+    rejectAs: Record<string, unknown> = record,
+  ): void {
+    const schemaMessage = failure === null && this.config.schema ? validateAgainstSchema(this.config.schema, record) : null;
+    const refused = failure ?? (schemaMessage === null ? null : { reason: 'schema' as const, message: schemaMessage });
+    if (refused !== null) {
+      if (this.config.onInvalid === 'fail') {
+        throw new Error(`${this.positions.label} ${position} failed ${refused.reason}: ${refused.message}`);
+      }
+      if (this.config.onInvalid === 'reject') {
+        if (this.rejects.length === 0) this.firstReject = position;
+        this.lastReject = position;
+        this.rejects.push({ ...rejectAs, _error: refused.message, [this.positions.positionKey]: position });
+        if (this.positions.cut === 'record' && this.rejects.length === this.config.batchSize) this.cutRejects();
+      }
+      return; // skip and reject both leave the data stream
+    }
+    if (this.records.length === 0) this.first = position;
+    this.last = position;
+    this.records.push(record);
+    if (this.positions.cut === 'record' && this.records.length === this.config.batchSize) this.cutData(this.last);
+  }
+
+  /** Batches that are ready. In `boundary` mode, call at each boundary: this is where they are cut. */
+  *drain(): Iterable<RecordBatch> {
+    if (this.positions.cut === 'boundary') {
+      if (this.rejects.length >= this.config.batchSize) this.cutRejects();
+      if (this.records.length >= this.config.batchSize) this.cutData(this.last);
+    }
+    yield* this.pending.splice(0);
+  }
+
+  /**
+   * Everything left: rejects first, then the final data batch. Its cursor is
+   * `lastPosition` — the last position read — when given, so rows after the
+   * last good one that were skipped or rejected are not read again on resume.
+   */
+  *flush(lastPosition?: number): Iterable<RecordBatch> {
+    yield* this.pending.splice(0);
+    if (this.rejects.length > 0) this.cutRejects();
+    yield* this.pending.splice(0);
+    if (this.records.length > 0) {
+      yield this.dataBatch(Math.max(this.last, lastPosition ?? this.last));
+      this.records = [];
+    }
+  }
+
+  private cutRejects(): void {
+    this.pending.push(rejectBatch(this.pipeId, this.firstReject, this.lastReject, this.rejects));
+    this.rejects = [];
+  }
+
+  private cutData(end: number): void {
+    this.pending.push(this.dataBatch(end));
+    this.records = [];
+  }
+
+  private dataBatch(end: number): RecordBatch {
+    return {
+      id: `${this.pipeId}:0:${this.first}-${end}`,
+      partitionId: '0',
+      records: this.records,
+      cursor: { [this.positions.cursorKey]: end },
+    };
+  }
+}
+
 /**
  * Validate records against a JSON Schema (the designer authors it as Zod;
  * regex rules arrive as `pattern`). Returns the failures, not a filtered list —
diff --git a/packages/workflow-engine/src/pipes/utility/file-sink.ts b/packages/workflow-engine/src/pipes/utility/file-sink.ts
index 40cf18eb..fde3c36e 100644
--- a/packages/workflow-engine/src/pipes/utility/file-sink.ts
+++ b/packages/workflow-engine/src/pipes/utility/file-sink.ts
@@ -5,12 +5,10 @@
  *
  * Workflow engine — moved from FoxAgent (packages/pipes/utility/src/file-sink.ts).
  */
-import { appendFile, mkdir, stat, writeFile } from 'node:fs/promises';
-import { dirname } from 'node:path';
 import { z } from 'zod';
 import type { PipeContext, RecordBatch, SinkPipe } from '../../registry/index.js';
 import { definePipeMetadata, type PipeMetadata } from '../../sdk/index.js';
-import { resolveWorkflowFile } from './file-root.js';
+import { openWorkflowFile } from './workflow-files.js';
 
 /**
  * Write records to a delimited file.
@@ -148,17 +146,14 @@ export class FileSinkPipe implements SinkPipe {
   }
 
   async write(batch: RecordBatch, context: PipeContext): Promise<void> {
-    const parsedConfig = configSchema.parse(context.pipe.config);
+    const config = configSchema.parse(context.pipe.config);
     if (batch.records.length === 0) return;
-    const config = { ...parsedConfig, path: await resolveWorkflowFile(parsedConfig.path) };
+    const file = await openWorkflowFile(config.path);
 
-    const key = `${context.workflowRunId}:${config.path}`;
+    const key = `${context.workflowRunId}:${file.path}`;
     const first = !this.started.has(key);
     if (first) this.started.add(key);
 
-    // eslint-disable-next-line security/detect-non-literal-fs-filename -- the target path is this sink's own configuration
-    await mkdir(dirname(config.path), { recursive: true });
-
     const body =
       batch.records.map((record) => toRow(record, config)).join(config.newline) +
       config.newline;
@@ -171,24 +166,13 @@ export class FileSinkPipe implements SinkPipe {
     // Overwrite truncates once per run, not once per batch — otherwise batch
     // two erases batch one and the file only ever holds the last chunk.
     if (first && config.mode === 'overwrite') {
-      // eslint-disable-next-line security/detect-non-literal-fs-filename -- the target path is this sink's own configuration
-      await writeFile(config.path, header + body, 'utf8');
+      await file.write(header + body);
       return;
     }
 
     // Appending to a file that does not exist yet still needs its header;
     // appending to one that already has rows must not get a second.
-    const lead = first && header && (await isEmpty(config.path)) ? header : '';
-    // eslint-disable-next-line security/detect-non-literal-fs-filename -- the target path is this sink's own configuration
-    await appendFile(config.path, lead + body, 'utf8');
-  }
-}
-
-async function isEmpty(path: string): Promise<boolean> {
-  try {
-    // eslint-disable-next-line security/detect-non-literal-fs-filename -- the target path is this sink's own configuration
-    return (await stat(path)).size === 0;
-  } catch {
-    return true; // does not exist yet
+    const lead = first && header && (await file.isEmpty()) ? header : '';
+    await file.append(lead + body);
   }
 }
diff --git a/packages/workflow-engine/src/pipes/utility/index.ts b/packages/workflow-engine/src/pipes/utility/index.ts
index 7ce35ed2..fa44c656 100644
--- a/packages/workflow-engine/src/pipes/utility/index.ts
+++ b/packages/workflow-engine/src/pipes/utility/index.ts
@@ -13,6 +13,7 @@ export {
   type Verdict,
 } from './verify.js';
 export { resolveWorkflowFile, workflowFilesRoot, FILES_DIR_ENV } from './file-root.js';
+export { openWorkflowFile, type WorkflowFile } from './workflow-files.js';
 export { FileSinkPipe, escapeField, toRow, type FileSinkConfig } from './file-sink.js';
 export {
   CsvSourcePipe,
diff --git a/packages/workflow-engine/src/pipes/utility/json.ts b/packages/workflow-engine/src/pipes/utility/json.ts
index ba4c1cac..13fb8e78 100644
--- a/packages/workflow-engine/src/pipes/utility/json.ts
+++ b/packages/workflow-engine/src/pipes/utility/json.ts
@@ -5,8 +5,6 @@
  *
  * Workflow engine — moved from FoxAgent (packages/pipes/utility/src/json.ts).
  */
-import { createReadStream } from 'node:fs';
-import { open, readFile } from 'node:fs/promises';
 import * as z from 'zod';
 import type {
   PipeContext,
@@ -21,10 +19,9 @@ import {
   rejectPolicyField,
   type PipeMetadata,
 } from '../../sdk/index.js';
-import { validateAgainstSchema } from '../../common/index.js';
-import { rejectBatch } from './delimited.js';
+import { BatchEmitter } from './delimited.js';
 import { readLines } from './text.js';
-import { resolveWorkflowFile } from './file-root.js';
+import { openWorkflowFile, type WorkflowFile } from './workflow-files.js';
 
 const configSchema = z.object({
   path: z.string().min(1),
@@ -71,45 +68,53 @@ export class JsonSourcePipe implements SourcePipe {
   }
 
   async *read(context: PipeContext): AsyncIterable<RecordBatch> {
-    const parsedConfig = configSchema.parse(context.pipe.config);
-    const config = { ...parsedConfig, path: await resolveWorkflowFile(parsedConfig.path) };
+    const config = configSchema.parse(context.pipe.config);
+    const file = await openWorkflowFile(config.path);
     const format =
-      config.format === 'auto' ? await detectFormat(config.path) : config.format;
+      config.format === 'auto' ? await detectFormat(file) : config.format;
     yield* format === 'array'
-      ? this.readArray(context, config)
-      : this.readNdjson(context, config);
+      ? this.readArray(context, config, file)
+      : this.readNdjson(context, config, file);
   }
 
   private async *readArray(
     context: PipeContext,
     config: Config,
+    file: WorkflowFile,
   ): AsyncIterable<RecordBatch> {
-    // eslint-disable-next-line security/detect-non-literal-fs-filename -- the source file path is this pipe's own configuration
-    const parsed: unknown = JSON.parse(await readFile(config.path, 'utf8'));
+    const parsed: unknown = JSON.parse(await file.readText());
     if (!Array.isArray(parsed)) {
-      throw new Error(`${config.path}: expected a top-level JSON array`);
+      throw new Error(`${file.path}: expected a top-level JSON array`);
     }
     const resumeAfter = Number(context.checkpoint?.cursor.index ?? 0);
-    const emitter = new BatchEmitter(context.pipe.id, config, 'index', '_index');
+    const emitter = new BatchEmitter(context.pipe.id, config, {
+      cursorKey: 'index',
+      positionKey: '_index',
+      label: 'index',
+      cut: 'record',
+    });
     for (let i = 0; i < parsed.length; i++) {
       const position = i + 1; // 1-based, matching row/line cursors
       if (position <= resumeAfter) continue;
-      emitter.add(toRecord(parsed[i]), position, null);
+      emitter.add(toRecord(parsed[i]), position);
     }
-    yield* emitter.flush();
+    yield* emitter.flush(parsed.length);
   }
 
   private async *readNdjson(
     context: PipeContext,
     config: Config,
+    file: WorkflowFile,
   ): AsyncIterable<RecordBatch> {
     const resumeAfter = Number(context.checkpoint?.cursor.line ?? 0);
-    const emitter = new BatchEmitter(context.pipe.id, config, 'line', '_line');
+    const emitter = new BatchEmitter(context.pipe.id, config, {
+      cursorKey: 'line',
+      positionKey: '_line',
+      label: 'line',
+      cut: 'record',
+    });
     let lineNumber = 0;
-    for await (const line of readLines(
-      // eslint-disable-next-line security/detect-non-literal-fs-filename -- the source file path is this pipe's own configuration
-      createReadStream(config.path, { encoding: 'utf8' }),
-    )) {
+    for await (const line of readLines(file.readStream())) {
       lineNumber++;
       if (lineNumber <= resumeAfter) continue;
       if (line.trim() === '') continue;
@@ -117,14 +122,14 @@ export class JsonSourcePipe implements SourcePipe {
       try {
         value = JSON.parse(line);
       } catch (error) {
-        emitter.add({ _raw: line }, lineNumber, (error as Error).message);
+        emitter.add({ _raw: line }, lineNumber, { reason: 'to parse', message: (error as Error).message });
         yield* emitter.drain();
         continue;
       }
-      emitter.add(toRecord(value), lineNumber, null);
+      emitter.add(toRecord(value), lineNumber);
       yield* emitter.drain();
     }
-    yield* emitter.flush();
+    yield* emitter.flush(lineNumber);
   }
 }
 
@@ -135,107 +140,7 @@ function toRecord(value: unknown): Record<string, unknown> {
     : { value: value ?? null };
 }
 
-async function detectFormat(path: string): Promise<'array' | 'ndjson'> {
-  // eslint-disable-next-line security/detect-non-literal-fs-filename -- the source file path is this pipe's own configuration
-  const handle = await open(path, 'r');
-  try {
-    const buffer = Buffer.alloc(1024);
-    const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
-    const head = buffer.toString('utf8', 0, bytesRead).replace(/^\uFEFF/, '');
-    return head.trimStart().startsWith('[') ? 'array' : 'ndjson';
-  } finally {
-    await handle.close();
-  }
-}
-
-/**
- * Shared batching/validation for both JSON layouts: validates each record,
- * applies onInvalid, and cuts data batches (with a cursor) and reject batches
- * (cursor-less, `rejects` port) at `batchSize`.
- */
-class BatchEmitter {
-  private records: Record<string, unknown>[] = [];
-  private first = 0;
-  private last = 0;
-  private rejects: Record<string, unknown>[] = [];
-  private firstReject = 0;
-  private lastReject = 0;
-  private pending: RecordBatch[] = [];
-
-  constructor(
-    private readonly pipeId: string,
-    private readonly config: Config,
-    private readonly cursorKey: 'index' | 'line',
-    private readonly positionKey: '_index' | '_line',
-  ) {}
-
-  /** `parseError` marks records that never parsed (NDJSON bad lines). */
-  add(
-    record: Record<string, unknown>,
-    position: number,
-    parseError: string | null,
-  ): void {
-    const message =
-      parseError ??
-      (this.config.schema
-        ? validateAgainstSchema(this.config.schema, record)
-        : null);
-    if (message !== null) {
-      if (this.config.onInvalid === 'fail') {
-        throw new Error(
-          `${this.cursorKey} ${position} failed ${parseError ? 'to parse' : 'schema'}: ${message}`,
-        );
-      }
-      if (this.config.onInvalid === 'reject') {
-        if (this.rejects.length === 0) this.firstReject = position;
-        this.lastReject = position;
-        this.rejects.push({
-          ...record,
-          _error: message,
-          [this.positionKey]: position,
-        });
-        if (this.rejects.length === this.config.batchSize) {
-          this.pending.push(
-            rejectBatch(this.pipeId, this.firstReject, this.lastReject, this.rejects),
-          );
-          this.rejects = [];
-        }
-      }
-      return;
-    }
-    if (this.records.length === 0) this.first = position;
-    this.last = position;
-    this.records.push(record);
-    if (this.records.length === this.config.batchSize) {
-      this.pending.push(this.dataBatch());
-      this.records = [];
-    }
-  }
-
-  /** Batches that became full since the last call. */
-  *drain(): Iterable<RecordBatch> {
-    yield* this.pending.splice(0);
-  }
-
-  /** Everything left: rejects first, then the final (cursor-bearing) batch. */
-  *flush(): Iterable<RecordBatch> {
-    yield* this.drain();
-    if (this.rejects.length > 0) {
-      yield rejectBatch(this.pipeId, this.firstReject, this.lastReject, this.rejects);
-      this.rejects = [];
-    }
-    if (this.records.length > 0) {
-      yield this.dataBatch();
-      this.records = [];
-    }
-  }
-
-  private dataBatch(): RecordBatch {
-    return {
-      id: `${this.pipeId}:0:${this.first}-${this.last}`,
-      partitionId: '0',
-      records: this.records,
-      cursor: { [this.cursorKey]: this.last },
-    };
-  }
+async function detectFormat(file: WorkflowFile): Promise<'array' | 'ndjson'> {
+  const head = (await file.readHead(1024)).text.replace(/^\uFEFF/, '');
+  return head.trimStart().startsWith('[') ? 'array' : 'ndjson';
 }
diff --git a/packages/workflow-engine/src/pipes/utility/preview-run-agreement.test.ts b/packages/workflow-engine/src/pipes/utility/preview-run-agreement.test.ts
new file mode 100644
index 00000000..5cd7f622
--- /dev/null
+++ b/packages/workflow-engine/src/pipes/utility/preview-run-agreement.test.ts
@@ -0,0 +1,71 @@
+/**
+ * Fox Schema (foxschema)
+ * Copyright 2024-2026 Huy Phan <huyplb@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The designer's preview exists to show what a run will read. Both now read
+ * through the same row reader; these read one file both ways and compare.
+ */
+import { mkdtemp, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import type { PipeContext, RecordBatch } from '../../registry/index.js';
+import { CsvSourcePipe, parseCsvRecords } from './csv.js';
+import { parseTextRows } from './delimited.js';
+import { FILES_DIR_ENV } from './file-root.js';
+import { TextSourcePipe, textConfigFields } from './text.js';
+
+let root = '';
+let previous: string | undefined;
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'fox-preview-run-'));
+  previous = process.env[FILES_DIR_ENV];
+  process.env[FILES_DIR_ENV] = root;
+});
+afterEach(async () => {
+  if (previous === undefined) delete process.env[FILES_DIR_ENV];
+  else process.env[FILES_DIR_ENV] = previous;
+  await rm(root, { recursive: true, force: true });
+});
+
+async function runRecords(pipe: { read(context: PipeContext): AsyncIterable<RecordBatch> }, config: Record<string, unknown>) {
+  const records: Record<string, unknown>[] = [];
+  const context = { pipe: { id: 'src', config } } as unknown as PipeContext;
+  for await (const batch of pipe.read(context)) if (!batch.port) records.push(...batch.records);
+  return records;
+}
+
+describe('preview and run read the same rows', () => {
+  it('for text: offset, header, empty lines, a flat map and typed columns', async () => {
+    const text = 'exported yesterday\nname|qty\n\nada|2::grace|3\r\nlinus|x\n';
+    await writeFile(join(root, 'in.txt'), text);
+    const config = {
+      path: 'in.txt',
+      offset: 1,
+      header: 'auto',
+      delimiters: ['|'],
+      recordDelimiter: '::',
+      fields: ['name', 'qty'],
+      rules: [{ name: 'qty', type: 'integer' }],
+      onInvalid: 'skip',
+    };
+    // The preview parses with the source's own schema and defaults.
+    const preview = parseTextRows(text, textConfigFields.parse(config));
+    const valid = preview.records.filter((_, index) => !preview.invalid.some((failure) => failure.index === index));
+    expect(await runRecords(new TextSourcePipe(), config)).toEqual(valid);
+    expect(valid).toEqual([
+      { name: 'ada', qty: 2 },
+      { name: 'grace', qty: 3 },
+    ]);
+  });
+
+  it('for CSV: a preamble, and quoted fields that span lines', async () => {
+    const text = 'exported yesterday\nid,note\n1,"two\nlines"\n2,"three\nmore\nlines"\n3,plain\n';
+    await writeFile(join(root, 'in.csv'), text);
+    const config = { path: 'in.csv', skipLines: 1 };
+    const preview = await parseCsvRecords(text, { skipLines: 1 });
+    expect(await runRecords(new CsvSourcePipe(), config)).toEqual(preview.records);
+    expect(preview.records.map((record) => record.note)).toEqual(['two\nlines', 'three\nmore\nlines', 'plain']);
+  });
+});
diff --git a/packages/workflow-engine/src/pipes/utility/text.ts b/packages/workflow-engine/src/pipes/utility/text.ts
index dab41e70..4614620b 100644
--- a/packages/workflow-engine/src/pipes/utility/text.ts
+++ b/packages/workflow-engine/src/pipes/utility/text.ts
@@ -5,7 +5,6 @@
  *
  * Workflow engine — moved from FoxAgent (packages/pipes/utility/src/text.ts).
  */
-import { createReadStream } from 'node:fs';
 import * as z from 'zod';
 import type {
   PipeContext,
@@ -20,19 +19,14 @@ import {
   rejectPolicyField,
   type PipeMetadata,
 } from '../../sdk/index.js';
-import { validateAgainstSchema } from '../../common/index.js';
 import {
-  applyColumnRules,
   columnRuleFieldsSchema,
   columnRules,
   compileColumnRules,
-  configuredNames,
-  describeFailures,
-  isHeaderRecord,
-  parseLine,
-  rejectBatch,
+  BatchEmitter,
+  TextRowReader,
 } from './delimited.js';
-import { resolveWorkflowFile } from './file-root.js';
+import { openWorkflowFile } from './workflow-files.js';
 
 /**
  * The field shape, exported without the cross-field `.refine()` so callers
@@ -139,109 +133,31 @@ export class TextSourcePipe implements SourcePipe {
   async *read(context: PipeContext): AsyncIterable<RecordBatch> {
     const config = configSchema.parse(context.pipe.config);
     const resumeAfter = Number(context.checkpoint?.cursor.line ?? 0);
-    let lineNumber = 0;
-    let records: Record<string, unknown>[] = [];
-    let firstLine = 0;
-    let rejects: Record<string, unknown>[] = [];
-    let firstReject = 0;
-    let lastReject = 0;
-    const rules = columnRules(config);
-    const compiled = compileColumnRules(rules);
-    // A resumed run's cursor is always past the header — it was handled (or
-    // absent) in the run that wrote the checkpoint.
-    let headerDone = config.header === 'none' || resumeAfter > 0;
-
-    for await (const line of readLines(
-      // eslint-disable-next-line security/detect-non-literal-fs-filename -- the source file path is this pipe's own configuration, confined to the files root
-      createReadStream(await resolveWorkflowFile(config.path), { encoding: 'utf8' }),
-    )) {
-      lineNumber++;
-      if (lineNumber <= config.offset) continue;
-      if (lineNumber <= resumeAfter) continue;
-      if (config.skipEmpty && line.trim() === '') continue;
-
-      const parsed = parseLine(line, config);
-      if (!headerDone) {
-        headerDone = true;
-        if (config.header === 'skip') continue;
-        if (parsed[0] && isHeaderRecord(parsed[0], configuredNames(config))) {
-          continue;
-        }
-      }
-
-      for (const raw of parsed) {
-        // Column rules run first: they convert text into typed values, so the
-        // whole-record schema sees numbers/booleans rather than strings.
-        const applied = applyColumnRules(raw, rules, compiled);
-        const record = applied.record;
-        const columnError =
-          applied.failures.length > 0
-            ? describeFailures(applied.failures)
-            : null;
-        if (columnError !== null) {
-          if (config.onInvalid === 'fail') {
-            throw new Error(`line ${lineNumber} failed validation: ${columnError}`);
-          }
-          if (config.onInvalid === 'reject') {
-            if (rejects.length === 0) firstReject = lineNumber;
-            lastReject = lineNumber;
-            rejects.push({ ...raw, _error: columnError, _line: lineNumber });
-          }
-          continue;
-        }
-        if (config.schema) {
-          const message = validateAgainstSchema(config.schema, record);
-          if (message !== null) {
-            if (config.onInvalid === 'fail') {
-              throw new Error(`line ${lineNumber} failed schema: ${message}`);
-            }
-            if (config.onInvalid === 'reject') {
-              if (rejects.length === 0) firstReject = lineNumber;
-              lastReject = lineNumber;
-              rejects.push({ ...record, _error: message, _line: lineNumber });
-            }
-            continue; // skip/reject — the row leaves the data stream either way
-          }
+    // Cut at line boundaries: a line flat-maps to several records, its cursor
+    // is the line, and a line must never straddle two batches.
+    const emitter = new BatchEmitter(context.pipe.id, config, {
+      cursorKey: 'line',
+      positionKey: '_line',
+      label: 'line',
+      cut: 'boundary',
+    });
+    // The same reader the designer preview uses, so the two cannot disagree.
+    const reader = new TextRowReader(config, resumeAfter);
+    for await (const line of readLines((await openWorkflowFile(config.path)).readStream())) {
+      for (const row of reader.push(line)) {
+        // A record a column rule refused is kept, if rejected, as the raw values.
+        if (row.failure !== null) {
+          emitter.add(row.record, row.line, { reason: 'validation', message: row.failure }, row.raw);
+        } else {
+          emitter.add(row.record, row.line);
         }
-        if (records.length === 0) firstLine = lineNumber;
-        records.push(record);
       }
-      // Cut batches only at line boundaries so the line cursor stays exact —
-      // a flat-mapped line is never split across two batches.
-      if (rejects.length >= config.batchSize) {
-        yield rejectBatch(context.pipe.id, firstReject, lastReject, rejects);
-        rejects = [];
-      }
-      if (records.length >= config.batchSize) {
-        yield makeBatch(context.pipe.id, firstLine, lineNumber, records);
-        records = [];
-      }
-    }
-    // Rejects flush before the final data batch — checkpoints only ever come
-    // from data cursors, so everything emitted is covered by the last one.
-    if (rejects.length > 0) {
-      yield rejectBatch(context.pipe.id, firstReject, lastReject, rejects);
-    }
-    if (records.length > 0) {
-      yield makeBatch(context.pipe.id, firstLine, lineNumber, records);
+      yield* emitter.drain();
     }
+    yield* emitter.flush(reader.line);
   }
 }
 
-function makeBatch(
-  pipeId: string,
-  firstLine: number,
-  lastLine: number,
-  records: Record<string, unknown>[],
-): RecordBatch {
-  return {
-    id: `${pipeId}:0:${firstLine}-${lastLine}`,
-    partitionId: '0',
-    records,
-    cursor: { line: lastLine },
-  };
-}
-
 export async function* readLines(
   chunks: AsyncIterable<string | Buffer>,
 ): AsyncIterable<string> {
diff --git a/packages/workflow-engine/src/pipes/utility/workflow-files.test.ts b/packages/workflow-engine/src/pipes/utility/workflow-files.test.ts
new file mode 100644
index 00000000..30bbe543
--- /dev/null
+++ b/packages/workflow-engine/src/pipes/utility/workflow-files.test.ts
@@ -0,0 +1,53 @@
+/**
+ * Fox Schema (foxschema)
+ * Copyright 2024-2026 Huy Phan <huyplb@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+import { mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { openWorkflowFile } from './workflow-files.js';
+
+let root = '';
+let outside = '';
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'fox-files-root-'));
+  outside = await mkdtemp(join(tmpdir(), 'fox-files-outside-'));
+});
+afterEach(async () => {
+  await rm(root, { recursive: true, force: true });
+  await rm(outside, { recursive: true, force: true });
+});
+
+describe('openWorkflowFile', () => {
+  it('refuses a path outside the root before anything is read or written', async () => {
+    await expect(openWorkflowFile('../escape.csv', root)).rejects.toThrow(/outside the workflow files directory/);
+    await expect(openWorkflowFile(join(outside, 'x.csv'), root)).rejects.toThrow(/outside/);
+  });
+
+  it('refuses a link inside the root that points out of it', async () => {
+    await writeFile(join(outside, 'secret.txt'), 'nope');
+    await symlink(join(outside, 'secret.txt'), join(root, 'link.txt'));
+    await expect(openWorkflowFile('link.txt', root)).rejects.toThrow(/outside/);
+  });
+
+  it('writes, appends into a new folder, and reads back', async () => {
+    const file = await openWorkflowFile('out/report.csv', root);
+    expect(await file.isEmpty()).toBe(true);
+    await file.append('a,b\n');
+    await file.append('1,2\n');
+    expect(await file.readText()).toBe('a,b\n1,2\n');
+    expect(await file.isEmpty()).toBe(false);
+    await file.write('fresh\n');
+    expect(await readFile(join(root, 'out/report.csv'), 'utf8')).toBe('fresh\n');
+  });
+
+  it('reads a head, dropping a partial last line only when asked', async () => {
+    await writeFile(join(root, 'in.txt'), 'line one\nline two\nline three\n');
+    const file = await openWorkflowFile('in.txt', root);
+    expect(await file.readHead(13)).toEqual({ text: 'line one\nline', truncated: true });
+    expect(await file.readHead(13, { wholeLines: true })).toEqual({ text: 'line one', truncated: true });
+    expect(await file.readHead(1000)).toEqual({ text: 'line one\nline two\nline three\n', truncated: false });
+  });
+});
diff --git a/packages/workflow-engine/src/pipes/utility/workflow-files.ts b/packages/workflow-engine/src/pipes/utility/workflow-files.ts
new file mode 100644
index 00000000..1b7f8982
--- /dev/null
+++ b/packages/workflow-engine/src/pipes/utility/workflow-files.ts
@@ -0,0 +1,87 @@
+/**
+ * Fox Schema (foxschema)
+ * Copyright 2024-2026 Huy Phan <huyplb@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The one way a file pipe touches the filesystem.
+ *
+ * Designing a workflow is an editor-level permission, and a file source or
+ * sink names its path in config. Each pipe used to resolve that path inside
+ * the files root itself and then call `node:fs` — so the rule that keeps an
+ * editor from reading `/etc/passwd` or appending to `authorized_keys` held
+ * only while every new file pipe remembered one call, and forgetting it
+ * failed open. Here the path is resolved once, inside the root, and the
+ * handle that comes back is the only thing that can read or write it. A test
+ * (`pipes-fs-boundary.test.ts`) fails the build when anything else under
+ * `pipes/` imports `node:fs`.
+ */
+import { createReadStream } from 'node:fs';
+import { appendFile, mkdir, open, readFile, stat, writeFile } from 'node:fs/promises';
+import { dirname } from 'node:path';
+import { resolveWorkflowFile } from './file-root.js';
+
+/** A configured file, resolved inside the workflow files root. */
+export interface WorkflowFile {
+  /** The resolved absolute path — for keys and messages, not for `node:fs`. */
+  readonly path: string;
+  /** The file as a UTF-8 text stream. */
+  readStream(): NodeJS.ReadableStream;
+  /** The whole file as text. */
+  readText(): Promise<string>;
+  /**
+   * The first `bytes` bytes as text. `truncated` when the file is longer;
+   * with `wholeLines`, the last (possibly partial) line is then dropped.
+   */
+  readHead(bytes: number, options?: { wholeLines?: boolean }): Promise<{ text: string; truncated: boolean }>;
+  /** Replace the file's content, creating its folder if needed. */
+  write(text: string): Promise<void>;
+  /** Add to the end of the file, creating it and its folder if needed. */
+  append(text: string): Promise<void>;
+  /** True when the file does not exist yet or has no bytes. */
+  isEmpty(): Promise<boolean>;
+}
+
+/** Resolve `configuredPath` inside the files root (or `root`) and hand back its file. */
+export async function openWorkflowFile(configuredPath: string, root?: string): Promise<WorkflowFile> {
+  const path = await resolveWorkflowFile(configuredPath, root);
+  // Every operation below takes `path` as resolved above, inside the root.
+  /* eslint-disable security/detect-non-literal-fs-filename -- confined by resolveWorkflowFile, above */
+  return {
+    path,
+    readStream: () => createReadStream(path, { encoding: 'utf8' }),
+    readText: () => readFile(path, 'utf8'),
+    async readHead(bytes, options = {}) {
+      const handle = await open(path, 'r');
+      try {
+        const { size } = await handle.stat();
+        const buffer = Buffer.alloc(Math.min(size, bytes));
+        const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
+        let text = buffer.toString('utf8', 0, bytesRead);
+        const truncated = size > bytes;
+        if (truncated && options.wholeLines) {
+          const lastNewline = text.lastIndexOf('\n');
+          if (lastNewline !== -1) text = text.slice(0, lastNewline);
+        }
+        return { text, truncated };
+      } finally {
+        await handle.close();
+      }
+    },
+    async write(text) {
+      await mkdir(dirname(path), { recursive: true });
+      await writeFile(path, text, 'utf8');
+    },
+    async append(text) {
+      await mkdir(dirname(path), { recursive: true });
+      await appendFile(path, text, 'utf8');
+    },
+    async isEmpty() {
+      try {
+        return (await stat(path)).size === 0;
+      } catch {
+        return true; // does not exist yet
+      }
+    },
+  };
+  /* eslint-enable security/detect-non-literal-fs-filename */
+}