---
name: framework-fs-split
description: >-
  Guides correct use of `split`, `join` and `read` from @t4h.framework/fs:
  fanning one large file out into a batch of workflows (one per
  slice), folding the batch results back into a single file, and reading small
  byte windows (headers, trailers) into the workflow — with O(1) parent
  history and bounded memory regardless of file size. Use whenever a workflow
  must process a large file record by record — positional files with
  header/trailer, plain-text lines, NDJSON, batch imports, per-record fan-out
  with an aggregated output (e.g. a CSV report) — or when the user mentions
  splitting/chunking a file, trimming a header/trailer, reading a file header,
  filtering batch results, running a workflow per line/record, waiting for a
  batch, collecting batch results, or the symbols `split`, `join`, `read`,
  `Split`, `FileSplitActivity`, `FileJoinActivity`, `FileReadActivity`,
  `WorkflowBatchRef`, `WorkflowBatchClaim`. Prefer this skill over
  hand-rolling stream loops + `startWorkflow` calls.
---

# `split` / `join` / `read` — file fan-out, fold and windows for workflows

`split` turns **one file into a batch of workflows**: it streams the file once,
cuts it into slices, starts one child workflow per slice, and **suspends the parent until every child is
terminal**. `join` is the inverse: it folds the batch results back into a
single file, streaming one result at a time.

```typescript
import { read, split, join, Split } from '@t4h.framework/fs'

const processRecord = new Workflow(
  { id: 'process-record', schema: Type.String() },
  async (record: string) => {
    // record is the raw slice content — no Binary handling needed
    return toCsvRow(parsePositional(record))
  },
)

const main = new Workflow({ id: 'main', schema: fileSchema }, async (file: Binary) => {
  const header = await read(file, { range: { start: 0, end: 240 } })
  // header: string — recorded in history, never re-read on replay

  const batch = await split(file, processRecord, {
    delimiter: Split.char('\n10', { next: 2 }),
    trim: { start: 241, end: 241 }, // header/trailer stay out of the batch
    timeoutInMs: 3_600_000,
  })
  // this line only runs once the WHOLE batch is terminal;
  // batch: WorkflowBatchRef — { id: string, count: number }

  return await join(batch, { format: 'lines', header: 'id;value;date' })
  // Binary: the aggregated file — no slice ever crossed workflow memory
})
```

Package path: `framework/packages/fs`. Import only from the package root.

---

## The memory model (the golden rule)

Everything here exists to keep the parent history **O(1)** whatever the file
size:

1. `split` reads the source once, worker side: it cuts records, drops the ones
   `where` rejects, groups the rest by `every` and hands each slice to the
   batch as a decoded string as soon as it is complete. Nothing collects the
   slices, and no copy of the file is written.
2. Each child receives its slice as a plain string — what its
   `Type.String()` schema declares.
3. The parent resumes with only `WorkflowBatchRef { id, count }`. Per-child
   results are **never aggregated onto the ref**: they live with each child
   run.
4. `join` streams results one at a time through `WorkflowBatchClaim.outputs`
   into `FileSystemClaim.write` with backpressure; only the resulting `Binary`
   reaches history.

**Golden rule:** volume proportional to the batch size only ever flows inside
an activity, through a claim. Into workflow history goes a fold result or a
bounded slice — never the N items.

---

## `split`

```typescript
async function split(
  binary: Binary,
  workflow: Workflow<any, any>,
  options: FileSplitOptions,
): Promise<WorkflowBatchRef>
```

The child workflow declares its **real input type** (e.g. `Type.String()`) —
it never sees `Binary`, ranges, or storage. `await split(...)` suspends the
parent (one pending activity, one resume, one replay) until every child is
terminal — **regardless of each child succeeding or failing**: a rejected
child never rejects the batch (allSettled semantics). An empty batch resolves
immediately.

### Delimiters (`options.delimiter`)

The cut is a serializable spec. When the delimiter is found at
`[p, p + length)`, the previous slice ends at `p + keep.previous` and the next
slice starts at `p + length - keep.next`; the bytes in between are discarded
(defaults discard the whole delimiter — classic line semantics).

```typescript
Split.line()                     // cut at every '\n', delimiter discarded
Split.char(';')                  // cut at any byte sequence
Split.char(';', { previous: 1 }) // keep the ';' with the previous slice
Split.char('\n10', { next: 2 })  // positional files, see below
```

Positional files where a logical record spans multiple physical lines and a
new record only starts when a line begins with a marker (the record type
code):

```
10JOHN;123\n20detail-a\n20detail-b\n10MARY;456\n20detail-c
└──────────── record 1 ───────────┘└─────── record 2 ──────┘
```

`Split.char('\n10', { next: 2 })`: the newline between records is discarded
and the `10` marker — which is **data**, not separator — stays at the start of
the next slice. Forgetting `{ next: 2 }` silently strips the record type code
from every slice.

Offsets are always computed over **raw bytes** (never decoded characters), so
multi-byte encodings cannot skew the cuts.

### Other options

| Option | Default | Behavior |
|--------|---------|----------|
| `every` | `1` | Cut only every N delimiter occurrences — groups N records per child. One million lines with `every: 10_000` start 100 children, each receiving a raw block (inner delimiters untouched) to iterate locally. `keep` arithmetic applies only at actual cut points. A slice over 3 MiB makes the split throw a `RangeError`; lower `every`. |
| `trim` | none | `{ start?, end? }` **bytes** cut off each end of the file before scanning — a positional header/trailer. Byte-based on purpose: a record-start delimiter (`'\n10'`) fuses the trailer into the last record's slice, so a record-count skip would drop a valid record along with it. Trimmed content never becomes a slice nor counts toward `every`; fetch it with `read`. Positional layouts make the byte counts constants — getting them right is the caller's business. |
| `skipEmpty` | `true` | Drops zero-length slices (consecutive delimiters, file starting/ending with one) so they never become child workflows. Turn off when slice index must align 1:1 with record position. "Empty" means zero bytes — a whitespace-only line is a regular slice. |
| `encoding` | `'utf8'` | Converts the delimiter to bytes and decodes each slice into the string the child receives. |
| `timeoutInMs` | none | Max milliseconds to await the batch; on expiry the activity **rejects** (reason `timeout`) instead of suspending forever — the safety net for a child that never reaches a terminal state. |

---

## `join`

```typescript
async function join(
  batch: WorkflowBatchRef,
  options: FileJoinOptions,
): Promise<Binary>
```

Formats (content is the **child's** job — the child returns the final line;
`join` only owns structure):

| Format | Behavior |
|--------|----------|
| `lines` | Concatenates raw child outputs, one per line. Options: `header`, `separator` (default `'\n'`). Non-string outputs are JSON-stringified. |
| `entries` | NDJSON dump of `{ index, status, output }`, one per line — the format-agnostic view of the batch, rejected children included by default. |
| `csv` | Projects object outputs into RFC 4180 escaped columns. Options: `columns` (required; entries are `{ key, header? }` — `key` selects the value on the child output (a dot-path like `'meta.total'` reaches nested fields, matching the `where` semantics; a literal top-level key with dots wins over the nested path), `header` labels the header row, plain string is shorthand for `{ key }`), `delimiter` (default `','`), `header` (default `true`). Missing keys become empty cells; nested values selected whole are JSON inside the cell. Value *formatting* belongs in the child workflow (sandboxed user code); the join only owns layout. |

**Status filter** (`status?: ('fulfilled' | 'rejected')[]`) selects which
results make it into the file — there is no error policy. Defaults:
`['fulfilled']` for `lines`/`csv`, both for `entries`. Filtering is silent by
design; a reprocessing file of failures is one call away:

```typescript
const failures = await join(batch, { format: 'entries', status: ['rejected'] })
```

**Where filter** (`where?`) narrows results with a **sift (MongoDB-style)
query** over each entry's output, applied after `status`. Whitelisted subset
only — declarative, JSON-only operators: `$eq`, `$ne`, `$lt`, `$lte`, `$gt`,
`$gte`, `$in`, `$nin`, `$exists`, `$not`, `$and`, `$or`, `$nor`, plus plain
equality and dot-paths for nested fields. Pass the child output type as a
generic to narrow top-level keys/values (dot-paths accepted, not narrowed):

```typescript
const relevant = await join<InvoiceResult>(batch, {
  format: 'csv',
  columns: ['id', 'total'],
  where: { valid: true, total: { $gt: 100 }, 'meta.situacao': 'aprovada' },
})
```

For an arbitrary predicate, write a custom activity in project code, so the
function lives in versioned code. MongoDB semantics apply: array fields match
by containment (`{ tags: 'a' }` matches `['a', 'b']`), object values compare
deeply, no type coercion. The query compiles once and runs inside the existing
streaming loop over already-hydrated outputs — same memory profile; app-side
(decides what enters the file, not what leaves the database).

---

## `read`

```typescript
async function read(
  binary: Binary,
  options?: { range?: { start: number; end: number }; encoding?: BufferEncoding },
): Promise<string>
```

Reads a byte window of a `Binary` into the workflow as a decoded string. The
output is recorded in history (replay never re-reads the file). Use it for
headers, trailers and single records; a data file flows through `split`.
`range` defaults to the whole binary; `encoding` defaults to `'utf8'`.

Two paths under the hood:

1. **Fast path** — `FileSystemClaim.slice`: on a file-backed binary a trailer
   read is a ranged download, whatever the file size.
2. **Streaming fallback** — binaries the claim cannot slice (e.g. an SFTP
   reference): the stream is consumed up to the window and destroyed as soon
   as it is filled. A header read pulls the first chunks only, whatever the
   file size; a **trailing window over a non-sliceable binary** is the one
   expensive case — the stream traverses the whole file to reach it.

Reading a slice is **relative to the slice** (`range: { start: 0, end: 10 }`
returns its first 10 bytes, not the underlying file's) — slice ranges compose.

The positional-file recipe — header/trailer in the parent, records in the
batch:

```typescript
const header = await read(file, { range: { start: 0, end: HEADER } })
const trailer = await read(file, {
  range: { start: file.contentLength - TRAILER, end: file.contentLength },
})

const batch = await split(file, processInvoice, {
  delimiter: Split.char('\n10', { next: 2 }),
  trim: { start: HEADER + 1, end: TRAILER + 1 }, // +1: boundary newlines
})

// e.g. validate the trailer's record count against batch.count
```

---

## Runtime requirements

- Claim providers: `FileSystemClaim` (with `slice`, for `read` and `join`),
  `WorkflowClaim` (with `startMany` reading an `AsyncIterable`), and
  `WorkflowBatchClaim` (from `@t4h.framework/core`).
- Production whitelist: `FileSplitActivity`, `FileJoinActivity`,
  `FileReadActivity` and core's `StartManyWorkflowsActivity` must be
  registered via `runtime.addActivity`.
- `FileSystemClaim.slice` implementations must **compose ranges** when slicing
  an already-sliced binary (relative in, absolute against the underlying file
  out) — `read` and slice consumers rely on that semantics.
- `FileSplitActivity` is an **AsyncActivity**: the heavy work happens in
  `bootstrap`, executed worker side atomically with the pending suspension —
  children cannot finish before the parent is awaiting them.
- `FileJoinActivity` is a **SyncActivity**: batches are only consumed after
  the completion barrier resolved, so there is nothing to await.

---

## Testing

Drive `bootstrap`/`run` directly with fake claims inside `Claim.run` (patterns
from `src/activities/__tests__/`):

```typescript
import { AsyncActivityResultPending, Claim, WorkflowBatchClaim, WorkflowBatchRef, WorkflowClaim } from '@t4h.framework/core'
import { FileSplitActivity, Split } from '@t4h.framework/fs'

const inputs: unknown[] = []
const startMany = vi.fn(async (_wf, slices: AsyncIterable<unknown>) => {
  for await (const slice of slices) inputs.push(slice) // must consume the stream
  return new WorkflowBatchRef('batch-1', inputs.length)
})

const result = await Claim.run(
  [{ provide: WorkflowClaim, value: { startMany } as never }],
  () =>
    new FileSplitActivity().bootstrap({ binary, workflow: 'child', options: { delimiter: Split.line() } }),
)

expect(result).toBeInstanceOf(AsyncActivityResultPending)
expect(inputs).toEqual(['foo', 'bar', 'baz'])
```

For `join`, provide `WorkflowBatchClaim` as an async generator:

```typescript
const outputs = vi.fn(async function* () {
  yield { index: 0, status: 'fulfilled', output: 'line-1' }
})
```

Behaviors worth asserting because they regress easily: the source is streamed
once; `keep` arithmetic on positional cuts; the empty batch resolving immediately;
CSV escaping (quotes doubled, delimiter/newline wrapped); the status filter.

---

## Anti-patterns

- **Paging batch results into the workflow.** A `readBatch(ref, cursor)` loop
  in workflow code turns every page into a replayed history entry — O(N) is
  back. Aggregation belongs inside `join` (or a framework activity consuming
  `WorkflowBatchClaim`).
- **Expecting per-child outputs on the ref.** `WorkflowBatchRef` is
  deliberately `{ id, count }` only. Results are read through `join` — e.g.
  `format: 'entries'` for the faithful dump.
- **Treating a rejected child as a batch failure.** The batch always resolves
  (allSettled). Decide what failures mean: compare against `batch.count`,
  generate the `status: ['rejected']` file, or tolerate.
- **Baking record format into `join`.** The child returns the final line
  (`lines`) or a projectable object (`csv`); custom formats live in the child
  workflow — sandboxed user code — not in new join options.
- **Splitting when you need the file whole.** If downstream wants the blob
  intact (hash it, upload it), pass the `Binary` through.
- **Forgetting `keep` on positional delimiters.** `Split.char('\n10')` without
  `{ next: 2 }` silently discards the record type code from every slice.
- **Splitting a positional file without `trim`.** The header rides inside the
  first slice (with `every: N` it silently consumes a record slot) and the
  trailer fuses into the last one. Trim both ends; fetch them with `read`.
- **Reading a data file with `read`.** The output lands in workflow history.
  Records flow through `split`. `read` throws a `RangeError` past 1 MiB.
- **A non-serializable `where`.** `tsc` rejects a function through
  `WorkflowBatchWhere`. `toInput` throws a `TypeError` for a function,
  `$where`, a `RegExp`, `undefined` or an operator outside the subset.
- **Hand-rolling the loop.** A stream-and-`startWorkflow` loop inside an
  activity re-parses on replay, misses the barrier, and pushes N entries into
  history.

---

## Related skills

- `framework-fs` — `FileSystemClaim` contract (including `slice`), `Binary`,
  provider wiring.
- `framework-core` — `startWorkflowMany` (the array-input sibling of `split`),
  `WorkflowBatchRef`, `WorkflowBatchClaim`, `Claim.run`, activity model,
  testing harness.