diff --git a/native-lib/node/src/dataweave.ts b/native-lib/node/src/dataweave.ts
new file mode 100644
index 0000000..9956862
--- /dev/null
+++ b/native-lib/node/src/dataweave.ts
@@ -0,0 +1,196 @@
+import * as ffi from "./ffi";
+import { findLibrary, buildInputsJson } from "./utils";
+import { parseNativeResponse } from "./result";
+import { createChunkReader } from "./reader";
+import { streamFromNative } from "./stream";
+import { DataWeaveError, DataWeaveScriptError } from "./errors";
+import type { ExecutionResult, StreamingResult, Inputs, TransformOptions } from "./types";
+
+/**
+ * A handle to the DataWeave native runtime for executing scripts.
+ *
+ * Wraps the native `dwlib` shared library via the FFI addon. Call
+ * {@link DataWeave.initialize} once before running scripts and
+ * {@link DataWeave.cleanup} when done. For most callers the module-level
+ * {@link run} / {@link runStreaming} / {@link runTransform} functions — backed
+ * by a lazily-initialized singleton — are more convenient than constructing an
+ * instance directly.
+ */
+export class DataWeave {
+  private readonly libPath: string;
+  private initialized = false;
+
+  /**
+   * @param libPath - Absolute path to the `dwlib` shared library. When omitted,
+   *   it is discovered via {@link findLibrary} (env var, packaged, or dev-build).
+   */
+  constructor(libPath?: string) {
+    this.libPath = libPath ?? findLibrary();
+  }
+
+  /**
+   * Loads and initializes the native runtime. Idempotent — a no-op if already
+   * initialized.
+   *
+   * @throws DataWeaveError if the native library fails to load or initialize.
+   */
+  initialize(): void {
+    if (this.initialized) return;
+    try {
+      ffi.initialize(this.libPath);
+    } catch (e: unknown) {
+      throw new DataWeaveError(`Failed to initialize: ${e instanceof Error ? e.message : e}`);
+    }
+    this.initialized = true;
+  }
+
+  /**
+   * Releases the native runtime. Idempotent — a no-op if not initialized. After
+   * cleanup the instance can be re-initialized via {@link DataWeave.initialize}.
+   */
+  cleanup(): void {
+    if (!this.initialized) return;
+    ffi.cleanup();
+    this.initialized = false;
+  }
+
+  /**
+   * Executes a script and returns its result in a single (non-streaming) call.
+   *
+   * @param script - The DataWeave script to run.
+   * @param inputs - Named inputs made available to the script (e.g. `payload`).
+   * @param opts - When `raiseOnError` is set, an unsuccessful result is thrown
+   *   as a {@link DataWeaveScriptError} instead of being returned.
+   * @returns The {@link ExecutionResult} carrying the output payload or error.
+   * @throws DataWeaveError if the runtime is not initialized.
+   * @throws DataWeaveScriptError if the script fails and `opts.raiseOnError` is set.
+   */
+  run(script: string, inputs?: Inputs, opts?: { raiseOnError?: boolean }): ExecutionResult {
+    this.ensureInitialized();
+    const inputsJson = buildInputsJson(inputs ?? {});
+    const raw = ffi.runScript(script, inputsJson);
+    const result = parseNativeResponse(raw);
+
+    if (opts?.raiseOnError && !result.success) {
+      throw new DataWeaveScriptError(result);
+    }
+    return result;
+  }
+
+  /**
+   * Executes a script and streams its output as it is produced.
+   *
+   * Yields output chunks in order; the generator's return value is the terminal
+   * {@link StreamingResult} with the final status and content metadata.
+   *
+   * @param script - The DataWeave script to run.
+   * @param inputs - Named inputs made available to the script.
+   * @returns An async generator of output chunks, returning the streaming metadata.
+   * @throws DataWeaveError if the runtime is not initialized.
+   */
+  async *runStreaming(script: string, inputs?: Inputs): AsyncGenerator<Buffer, StreamingResult, undefined> {
+    this.ensureInitialized();
+    const inputsJson = buildInputsJson(inputs ?? {});
+    return yield* streamFromNative((chunkCb) => ffi.runScriptStreaming(script, inputsJson, chunkCb));
+  }
+
+  /**
+   * Executes a script over a streamed primary input, streaming the output.
+   *
+   * The `input` chunks are fed to the script as the primary input (named
+   * `opts.inputName`, default `payload`); output chunks are yielded as they are
+   * produced and the generator returns the terminal {@link StreamingResult}.
+   * Sync iterables are consumed on demand; async iterables are pre-buffered (see
+   * {@link createChunkReader}).
+   *
+   * @param script - The DataWeave script to run.
+   * @param input - A sync or async iterable of byte chunks for the primary input.
+   * @param opts - Primary-input framing (`inputName`, `mimeType`, `charset`) and
+   *   any additional named `inputs`.
+   * @returns An async generator of output chunks, returning the streaming metadata.
+   * @throws DataWeaveError if the runtime is not initialized.
+   */
+  async *runTransform(
+    script: string,
+    input: AsyncIterable<Buffer | Uint8Array> | Iterable<Buffer | Uint8Array>,
+    opts?: TransformOptions
+  ): AsyncGenerator<Buffer, StreamingResult, undefined> {
+    this.ensureInitialized();
+
+    const inputName = opts?.inputName ?? "payload";
+    const inputMimeType = opts?.mimeType ?? "application/json";
+    const inputCharset = opts?.charset ?? null;
+    const extraInputs = opts?.inputs ?? {};
+    const inputsJson = Object.keys(extraInputs).length > 0 ? buildInputsJson(extraInputs) : "{}";
+
+    const readCb = await createChunkReader(input);
+
+    return yield* streamFromNative((writeCb) =>
+      ffi.runScriptTransform(script, inputsJson, inputName, inputMimeType, inputCharset, readCb, writeCb)
+    );
+  }
+
+  private ensureInitialized(): void {
+    if (!this.initialized) {
+      throw new DataWeaveError("DataWeave runtime not initialized. Call initialize() first.");
+    }
+  }
+}
+
+// Module-level convenience API with lazy singleton
+let globalInstance: DataWeave | null = null;
+
+/**
+ * Returns the process-wide {@link DataWeave} singleton, creating and
+ * initializing it (and registering a process-exit cleanup hook) on first use.
+ */
+function getGlobalInstance(): DataWeave {
+  if (!globalInstance) {
+    globalInstance = new DataWeave();
+    globalInstance.initialize();
+    process.on("exit", () => cleanup());
+  }
+  return globalInstance;
+}
+
+/**
+ * Executes a script on the shared {@link DataWeave} singleton.
+ * @see DataWeave.run
+ */
+export function run(script: string, inputs?: Inputs, opts?: { raiseOnError?: boolean }): ExecutionResult {
+  return getGlobalInstance().run(script, inputs, opts);
+}
+
+/**
+ * Streams a script's output on the shared {@link DataWeave} singleton.
+ * @see DataWeave.runStreaming
+ */
+export function runStreaming(
+  script: string,
+  inputs?: Inputs
+): AsyncGenerator<Buffer, StreamingResult, undefined> {
+  return getGlobalInstance().runStreaming(script, inputs);
+}
+
+/**
+ * Streams a transform over a streamed input on the shared {@link DataWeave} singleton.
+ * @see DataWeave.runTransform
+ */
+export function runTransform(
+  script: string,
+  input: AsyncIterable<Buffer | Uint8Array> | Iterable<Buffer | Uint8Array>,
+  opts?: TransformOptions
+): AsyncGenerator<Buffer, StreamingResult, undefined> {
+  return getGlobalInstance().runTransform(script, input, opts);
+}
+
+/**
+ * Releases the shared {@link DataWeave} singleton, if one was created. A fresh
+ * singleton is created lazily on the next convenience-API call.
+ */
+export function cleanup(): void {
+  if (globalInstance) {
+    globalInstance.cleanup();
+    globalInstance = null;
+  }
+}
\ No newline at end of file
diff --git a/native-lib/node/src/errors.ts b/native-lib/node/src/errors.ts
new file mode 100644
index 0000000..9e4eb36
--- /dev/null
+++ b/native-lib/node/src/errors.ts
@@ -0,0 +1,22 @@
+import type { ExecutionResult } from "./types";
+
+/** Base error for all failures raised by the DataWeave binding. */
+export class DataWeaveError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "DataWeaveError";
+  }
+}
+
+/**
+ * Raised when a script executes but reports a failure, when the caller opted in
+ * via `raiseOnError`. Carries the full {@link ExecutionResult} for inspection.
+ */
+export class DataWeaveScriptError extends DataWeaveError {
+  result: ExecutionResult;
+  constructor(result: ExecutionResult) {
+    super(result.error ?? "Script execution failed");
+    this.name = "DataWeaveScriptError";
+    this.result = result;
+  }
+}
\ No newline at end of file
diff --git a/native-lib/node/src/index.ts b/native-lib/node/src/index.ts
index 762ab35..05711e2 100644
--- a/native-lib/node/src/index.ts
+++ b/native-lib/node/src/index.ts
@@ -1,12 +1,5 @@
-import * as ffi from "./ffi";
-import { findLibrary, buildInputsJson } from "./utils";
-import { parseNativeResponse, parseStreamingResult } from "./result";
-import type {
-  ExecutionResult,
-  StreamingResult,
-  Inputs,
-  TransformOptions,
-} from "./types";
+export { DataWeave, run, runStreaming, runTransform, cleanup } from "./dataweave";
+export { DataWeaveError, DataWeaveScriptError } from "./errors";
 
 export type {
   ExecutionResult,
@@ -15,277 +8,4 @@ export type {
   InputValue,
   InputEntry,
   TransformOptions,
-} from "./types";
-
-export class DataWeaveError extends Error {
-  constructor(message: string) {
-    super(message);
-    this.name = "DataWeaveError";
-  }
-}
-
-export class DataWeaveScriptError extends DataWeaveError {
-  result: ExecutionResult;
-  constructor(result: ExecutionResult) {
-    super(result.error ?? "Script execution failed");
-    this.name = "DataWeaveScriptError";
-    this.result = result;
-  }
-}
-
-export class DataWeave {
-  private libPath: string;
-  private initialized = false;
-
-  constructor(libPath?: string) {
-    this.libPath = libPath ?? findLibrary();
-  }
-
-  initialize(): void {
-    if (this.initialized) return;
-    try {
-      ffi.initialize(this.libPath);
-    } catch (e: unknown) {
-      throw new DataWeaveError(`Failed to initialize: ${e instanceof Error ? e.message : e}`);
-    }
-    this.initialized = true;
-  }
-
-  cleanup(): void {
-    if (!this.initialized) return;
-    ffi.cleanup();
-    this.initialized = false;
-  }
-
-  run(script: string, inputs?: Inputs, opts?: { raiseOnError?: boolean }): ExecutionResult {
-    this.ensureInitialized();
-    const inputsJson = buildInputsJson(inputs ?? {});
-    const raw = ffi.runScript(script, inputsJson);
-    const result = parseNativeResponse(raw);
-
-    if (opts?.raiseOnError && !result.success) {
-      throw new DataWeaveScriptError(result);
-    }
-    return result;
-  }
-
-  async *runStreaming(script: string, inputs?: Inputs): AsyncGenerator<Buffer, StreamingResult, undefined> {
-    this.ensureInitialized();
-    const inputsJson = buildInputsJson(inputs ?? {});
-
-    const chunks: Buffer[] = [];
-    const pendingResolves: Array<() => void> = [];
-    let done = false;
-    let metaRaw: string | null = null;
-
-    const chunkCb = (chunk: Buffer) => {
-      chunks.push(chunk);
-      // Resolve one waiting consumer if any
-      const resolve = pendingResolves.shift();
-      if (resolve) {
-        resolve();
-      }
-    };
-
-    const metaPromise = ffi.runScriptStreaming(script, inputsJson, chunkCb).then((raw) => {
-      metaRaw = raw;
-      done = true;
-      // Wake all waiting consumers
-      while (pendingResolves.length > 0) {
-        const resolve = pendingResolves.shift();
-        if (resolve) resolve();
-      }
-    });
-
-    while (true) {
-      if (chunks.length > 0) {
-        yield chunks.shift()!;
-        continue;
-      }
-      if (done) break;
-      await new Promise<void>((resolve) => { pendingResolves.push(resolve); });
-    }
-
-    // Drain remaining chunks
-    while (chunks.length > 0) {
-      yield chunks.shift()!;
-    }
-
-    await metaPromise;
-    return parseStreamingResult(metaRaw ?? "");
-  }
-
-  async *runTransform(
-    script: string,
-    input: AsyncIterable<Buffer | Uint8Array> | Iterable<Buffer | Uint8Array>,
-    opts?: TransformOptions
-  ): AsyncGenerator<Buffer, StreamingResult, undefined> {
-    this.ensureInitialized();
-
-    const inputName = opts?.inputName ?? "payload";
-    const inputMimeType = opts?.mimeType ?? "application/json";
-    const inputCharset = opts?.charset ?? null;
-    const extraInputs = opts?.inputs ?? {};
-    const inputsJson = Object.keys(extraInputs).length > 0 ? buildInputsJson(extraInputs) : "{}";
-
-    const isAsync = Symbol.asyncIterator in (input as object);
-
-    let readCb: (bufSize: number) => Buffer | null;
-
-    if (isAsync) {
-      // Async iterables must be pre-buffered because the native read callback
-      // is invoked synchronously on the JS main thread and cannot await.
-      const inputBuffers: (Buffer | null)[] = [];
-      const asyncIter = (input as AsyncIterable<Buffer | Uint8Array>)[Symbol.asyncIterator]();
-      try {
-        while (true) {
-          const { value, done: d } = await asyncIter.next();
-          if (d) break;
-          inputBuffers.push(Buffer.isBuffer(value) ? value : Buffer.from(value));
-        }
-      } catch { /* input error = EOF */ }
-
-      let bufIdx = 0;
-      let currentBuf: Buffer | null = null;
-      let readOffset = 0;
-
-      readCb = (bufSize: number): Buffer | null => {
-        while (true) {
-          if (currentBuf && readOffset < currentBuf.length) {
-            const n = Math.min(currentBuf.length - readOffset, bufSize);
-            const slice = currentBuf.subarray(readOffset, readOffset + n);
-            readOffset += n;
-            if (readOffset >= currentBuf.length) {
-              currentBuf = null;
-              readOffset = 0;
-            }
-            return Buffer.from(slice);
-          }
-          if (bufIdx < inputBuffers.length) {
-            currentBuf = inputBuffers[bufIdx];
-            inputBuffers[bufIdx] = null; // Release memory as we consume
-            bufIdx++;
-            readOffset = 0;
-            continue;
-          }
-          return null;
-        }
-      };
-    } else {
-      // Sync iterables are consumed on-demand — constant memory, no pre-buffering.
-      const syncIter = (input as Iterable<Buffer | Uint8Array>)[Symbol.iterator]();
-      let currentBuf: Buffer | null = null;
-      let readOffset = 0;
-      let iterDone = false;
-
-      readCb = (bufSize: number): Buffer | null => {
-        while (true) {
-          if (currentBuf && readOffset < currentBuf.length) {
-            const n = Math.min(currentBuf.length - readOffset, bufSize);
-            const slice = currentBuf.subarray(readOffset, readOffset + n);
-            readOffset += n;
-            if (readOffset >= currentBuf.length) {
-              currentBuf = null;
-              readOffset = 0;
-            }
-            return Buffer.from(slice);
-          }
-          if (iterDone) return null;
-          const { value, done: d } = syncIter.next();
-          if (d) {
-            iterDone = true;
-            return null;
-          }
-          currentBuf = Buffer.isBuffer(value) ? value : Buffer.from(value);
-          readOffset = 0;
-        }
-      };
-    }
-
-    const chunks: Buffer[] = [];
-    const pendingResolves: Array<() => void> = [];
-    let done = false;
-    let metaRaw: string | null = null;
-
-    const writeCb = (chunk: Buffer) => {
-      chunks.push(chunk);
-      // Resolve one waiting consumer if any
-      const resolve = pendingResolves.shift();
-      if (resolve) {
-        resolve();
-      }
-    };
-
-    const metaPromise = ffi.runScriptTransform(
-      script, inputsJson, inputName, inputMimeType, inputCharset, readCb, writeCb
-    ).then((raw) => {
-      metaRaw = raw;
-      done = true;
-      // Wake all waiting consumers
-      while (pendingResolves.length > 0) {
-        const resolve = pendingResolves.shift();
-        if (resolve) resolve();
-      }
-    });
-
-    while (true) {
-      if (chunks.length > 0) {
-        yield chunks.shift()!;
-        continue;
-      }
-      if (done) break;
-      await new Promise<void>((resolve) => { pendingResolves.push(resolve); });
-    }
-
-    while (chunks.length > 0) {
-      yield chunks.shift()!;
-    }
-
-    await metaPromise;
-    return parseStreamingResult(metaRaw ?? "");
-  }
-
-  private ensureInitialized(): void {
-    if (!this.initialized) {
-      throw new DataWeaveError("DataWeave runtime not initialized. Call initialize() first.");
-    }
-  }
-}
-
-// Module-level convenience API with lazy singleton
-let globalInstance: DataWeave | null = null;
-
-function getGlobalInstance(): DataWeave {
-  if (!globalInstance) {
-    globalInstance = new DataWeave();
-    globalInstance.initialize();
-    process.on("exit", () => cleanup());
-  }
-  return globalInstance;
-}
-
-export function run(script: string, inputs?: Inputs, opts?: { raiseOnError?: boolean }): ExecutionResult {
-  return getGlobalInstance().run(script, inputs, opts);
-}
-
-export function runStreaming(
-  script: string,
-  inputs?: Inputs
-): AsyncGenerator<Buffer, StreamingResult, undefined> {
-  return getGlobalInstance().runStreaming(script, inputs);
-}
-
-export function runTransform(
-  script: string,
-  input: AsyncIterable<Buffer | Uint8Array> | Iterable<Buffer | Uint8Array>,
-  opts?: TransformOptions
-): AsyncGenerator<Buffer, StreamingResult, undefined> {
-  return getGlobalInstance().runTransform(script, input, opts);
-}
-
-export function cleanup(): void {
-  if (globalInstance) {
-    globalInstance.cleanup();
-    globalInstance = null;
-  }
-}
+} from "./types";
\ No newline at end of file
diff --git a/native-lib/node/src/reader.ts b/native-lib/node/src/reader.ts
new file mode 100644
index 0000000..1aae7ed
--- /dev/null
+++ b/native-lib/node/src/reader.ts
@@ -0,0 +1,93 @@
+/**
+ * A pull-based byte reader: `readCb(bufSize)` returns up to `bufSize` bytes as a
+ * fresh {@link Buffer}, or `null` once the input is exhausted. The native
+ * transform layer drives it synchronously to pull input on demand.
+ */
+export type ChunkReader = (bufSize: number) => Buffer | null;
+
+/**
+ * Builds a {@link ChunkReader} over an input iterable of byte chunks.
+ *
+ * Async iterables are fully pre-buffered up front because the native read
+ * callback is invoked synchronously on the JS main thread and cannot await;
+ * consumed buffers are released as they are read to bound memory. Sync
+ * iterables are pulled on demand for constant-memory streaming.
+ *
+ * @param input - The source of byte chunks (Buffers or Uint8Arrays).
+ * @returns A reader that yields the input as `bufSize`-bounded Buffers, then `null`.
+ */
+export async function createChunkReader(
+  input: AsyncIterable<Buffer | Uint8Array> | Iterable<Buffer | Uint8Array>
+): Promise<ChunkReader> {
+  const isAsync = Symbol.asyncIterator in (input as object);
+
+  if (isAsync) {
+    // Async iterables must be pre-buffered because the native read callback
+    // is invoked synchronously on the JS main thread and cannot await.
+    const inputBuffers: (Buffer | null)[] = [];
+    const asyncIter = (input as AsyncIterable<Buffer | Uint8Array>)[Symbol.asyncIterator]();
+    try {
+      while (true) {
+        const { value, done: d } = await asyncIter.next();
+        if (d) break;
+        inputBuffers.push(Buffer.isBuffer(value) ? value : Buffer.from(value));
+      }
+    } catch { /* input error = EOF */ }
+
+    let bufIdx = 0;
+    let currentBuf: Buffer | null = null;
+    let readOffset = 0;
+
+    return (bufSize: number): Buffer | null => {
+      while (true) {
+        if (currentBuf && readOffset < currentBuf.length) {
+          const n = Math.min(currentBuf.length - readOffset, bufSize);
+          const slice = currentBuf.subarray(readOffset, readOffset + n);
+          readOffset += n;
+          if (readOffset >= currentBuf.length) {
+            currentBuf = null;
+            readOffset = 0;
+          }
+          return Buffer.from(slice);
+        }
+        if (bufIdx < inputBuffers.length) {
+          currentBuf = inputBuffers[bufIdx];
+          inputBuffers[bufIdx] = null; // Release memory as we consume
+          bufIdx++;
+          readOffset = 0;
+          continue;
+        }
+        return null;
+      }
+    };
+  }
+
+  // Sync iterables are consumed on-demand — constant memory, no pre-buffering.
+  const syncIter = (input as Iterable<Buffer | Uint8Array>)[Symbol.iterator]();
+  let currentBuf: Buffer | null = null;
+  let readOffset = 0;
+  let iterDone = false;
+
+  return (bufSize: number): Buffer | null => {
+    while (true) {
+      if (currentBuf && readOffset < currentBuf.length) {
+        const n = Math.min(currentBuf.length - readOffset, bufSize);
+        const slice = currentBuf.subarray(readOffset, readOffset + n);
+        readOffset += n;
+        if (readOffset >= currentBuf.length) {
+          currentBuf = null;
+          readOffset = 0;
+        }
+        return Buffer.from(slice);
+      }
+      if (iterDone) return null;
+      const { value, done: d } = syncIter.next();
+      if (d) {
+        iterDone = true;
+        return null;
+      }
+      currentBuf = Buffer.isBuffer(value) ? value : Buffer.from(value);
+      readOffset = 0;
+    }
+  };
+}
\ No newline at end of file
diff --git a/native-lib/node/src/stream.ts b/native-lib/node/src/stream.ts
new file mode 100644
index 0000000..855807f
--- /dev/null
+++ b/native-lib/node/src/stream.ts
@@ -0,0 +1,65 @@
+import { parseStreamingResult } from "./result";
+import type { StreamingResult } from "./types";
+
+/**
+ * Starts a native streaming call, wiring its chunk callback to `chunkCb` and
+ * resolving to the raw trailing metadata JSON once the stream completes.
+ */
+export type StartStreaming = (chunkCb: (chunk: Buffer) => void) => Promise<string>;
+
+/**
+ * Bridges a native push-based streaming call into a pull-based async generator.
+ *
+ * The native side pushes output chunks through the callback while
+ * {@link StartStreaming} runs; this generator buffers them and yields in order,
+ * parking the consumer when no chunk is ready and waking it on the next push or
+ * on completion. After all chunks drain, it awaits the native promise and
+ * returns the parsed {@link StreamingResult}.
+ *
+ * @param start - Launches the native call and returns its metadata promise.
+ * @returns An async generator of output chunks whose return value is the terminal metadata.
+ */
+export async function* streamFromNative(
+  start: StartStreaming
+): AsyncGenerator<Buffer, StreamingResult, undefined> {
+  const chunks: Buffer[] = [];
+  const pendingResolves: Array<() => void> = [];
+  let done = false;
+  let metaRaw: string | null = null;
+
+  const chunkCb = (chunk: Buffer) => {
+    chunks.push(chunk);
+    // Resolve one waiting consumer if any
+    const resolve = pendingResolves.shift();
+    if (resolve) {
+      resolve();
+    }
+  };
+
+  const metaPromise = start(chunkCb).then((raw) => {
+    metaRaw = raw;
+    done = true;
+    // Wake all waiting consumers
+    while (pendingResolves.length > 0) {
+      const resolve = pendingResolves.shift();
+      if (resolve) resolve();
+    }
+  });
+
+  while (true) {
+    if (chunks.length > 0) {
+      yield chunks.shift()!;
+      continue;
+    }
+    if (done) break;
+    await new Promise<void>((resolve) => { pendingResolves.push(resolve); });
+  }
+
+  // Drain remaining chunks
+  while (chunks.length > 0) {
+    yield chunks.shift()!;
+  }
+
+  await metaPromise;
+  return parseStreamingResult(metaRaw ?? "");
+}
\ No newline at end of file