From 7746bb2359a84743ae7f4e7dc6483fbf4242dbc5 Mon Sep 17 00:00:00 2001
From: huyplb <huyplb@live.com>
Date: Sat, 17 Oct 2026 10:39:18 -0600
Subject: [PATCH] refactor(workflow): one file boundary, one batcher, one
 reader per format, and the engine's own types

The workflow review's remaining structural items.

File boundary. Every file pipe resolved its configured path inside the
files root and then called node:fs itself, so the rule that keeps an editor
from reading /etc/passwd held only while each pipe remembered one call.
openWorkflowFile resolves once and hands back the only handle that can read
or write; CSV, text, JSON, the CSV sink and the preview route use it, and
pipes-fs-boundary.test.ts fails the build when anything else under pipes/
imports node:fs.

One batcher. CSV, text and JSON each kept their own onInvalid/reject
batching and had drifted on where they cut, what the last cursor named and
how errors read. BatchEmitter in delimited.ts does it once: `record` cuts
for CSV/JSON, `boundary` cuts for text (a line never straddles batches),
and the final cursor names the last position read, so trailing refused rows
are not read again on resume (JSON now gets that too).

One reader per format. The preview re-implemented each source's loop and
had diverged: CSV numbered rows as preamble + header + index, wrong after
any quoted field spanning lines. csvRows and TextRowReader are shared by
run and preview; CSV rows carry the physical line they start on.

Engine types. The HTTP request model, column rules, credential kinds and
the designer's saved pipes now come from @foxschema/workflow-engine
(type-only; z.input where the designer writes what the engine parses). The
swap surfaced two more drops on save: a pipe's retry policy beyond
`attempts`, and a column type typed as any string. Both fixed; the e2e
round trip now covers retry, and the trigger dialog keeping its section
while typing.

Verified: full vitest (timing tests pass alone), apps/web tsc, eslint +
security lint, build, bundle 146.2 KB; e2e workflow suite in a real
browser; every new test fails on the code before it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../tests/workflow-designer-roundtrip.test.ts |  31 ++-
 .../workflow/components/CredentialsPanel.tsx  |   4 +-
 .../components/DelimitedSourceEditor.tsx      |  44 ++--
 .../workflow/components/HttpRequestEditor.tsx | 103 +++-----
 .../features/workflow/components/PipeNode.tsx |   4 +
 .../workflow/components/WorkflowDesigner.tsx  |  39 ++--
 apps/workflow-server/src/preview.test.ts      |  12 +
 apps/workflow-server/src/routes/preview.ts    |  45 +---
 packages/workflow-engine/src/index.ts         |   2 +
 .../src/pipes/pipes-fs-boundary.test.ts       |  42 ++++
 .../src/pipes/utility/batch-emitter.test.ts   |  66 ++++++
 .../src/pipes/utility/csv-parse.test.ts       |  18 +-
 .../workflow-engine/src/pipes/utility/csv.ts  | 187 +++++++--------
 .../src/pipes/utility/delimited.ts            | 221 +++++++++++++++---
 .../src/pipes/utility/file-sink.ts            |  30 +--
 .../src/pipes/utility/index.ts                |   1 +
 .../workflow-engine/src/pipes/utility/json.ts | 159 +++----------
 .../utility/preview-run-agreement.test.ts     |  71 ++++++
 .../workflow-engine/src/pipes/utility/text.ts | 128 ++--------
 .../src/pipes/utility/workflow-files.test.ts  |  53 +++++
 .../src/pipes/utility/workflow-files.ts       |  87 +++++++
 21 files changed, 788 insertions(+), 559 deletions(-)
 create mode 100644 packages/workflow-engine/src/pipes/pipes-fs-boundary.test.ts
 create mode 100644 packages/workflow-engine/src/pipes/utility/batch-emitter.test.ts
 create mode 100644 packages/workflow-engine/src/pipes/utility/preview-run-agreement.test.ts
 create mode 100644 packages/workflow-engine/src/pipes/utility/workflow-files.test.ts
 create mode 100644 packages/workflow-engine/src/pipes/utility/workflow-files.ts

diff --git a/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts b/apps/e2e/src/tests/workflow-designer-roundtrip.test.ts
index d3b7641df..229b12a63 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 52cff2ac4..3f9a96d93 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 c289381c9..c0bf0baa4 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 1af6fcb23..c282dbd81 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 b360a8138..39ec35fe5 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 f71b687bb..45b395fa8 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 91dc51ae2..cb59ddcbf 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 a1797fcce..519f7bd63 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 e8fc994c3..93c5a4b28 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 000000000..78212acbd
--- /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 000000000..0bbed9cd4
--- /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 3ae4102bb..33c75331c 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 3a5817e5f..0c840d075 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 403c406fc..088a2a45f 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 40cf18eb9..fde3c36e7 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 7ce35ed2a..fa44c656d 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 ba4c1cacb..13fb8e788 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 000000000..5cd7f6223
--- /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 dab41e701..4614620b6 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 000000000..30bbe543d
--- /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 000000000..1b7f8982f
--- /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 */
+}