diff --git a/.changeset/uneval-stream.md b/.changeset/uneval-stream.md
new file mode 100644
index 00000000..2f9594bf
--- /dev/null
+++ b/.changeset/uneval-stream.md
@@ -0,0 +1,5 @@
+---
+'devalue': minor
+---
+
+feat: Add `unevalStream` for streaming asynchronous values
diff --git a/README.md b/README.md
index e92d8814..a7ef8d9d 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,7 @@ Like `JSON.stringify`, but handles
- `Temporal`
- custom types via replacers, reducers and revivers
- promises (via `stringifyAsync`)
+- streamed promises (via `unevalStream`)
Try it out [here](https://svelte.dev/repl/138d70def7a748ce9eda736ef1c71239?version=3.49.0).
@@ -87,6 +88,47 @@ devalue.parse(stringified); // { quick: 'data', slow: { ... } }
Promises are awaited and their resolved values are serialized. The output format is identical to `stringify`, so `parse` and `unflatten` work unchanged.
+### `unevalStream`
+
+`unevalStream` is a streaming version of `uneval`. It synchronously returns a `head` expression that recreates the value with every promise _pending_, and a `tail` of statement blocks that settle those promises as they settle on the server:
+
+```js
+const { head, tail } = devalue.unevalStream({
+ quick: 'data',
+ slow: fetch('/api/slow').then((r) => r.json())
+});
+
+res.write(``);
+for await (const block of tail) res.write(``);
+```
+
+Evaluate `head` once, then each block once, in order, in the same realm. Concatenating them into a single script also works. A settled value may contain further promises, which later blocks settle.
+
+- every promise is pending in `head`, regardless of timing
+- object identity is preserved across `head` and every block
+- the replacer is called at most once per object for the whole stream
+
+#### Options
+
+| Option | Description |
+| ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `id` | Session key (returned as `id`). Random by default; pass a stable value when `head` must be byte-identical across runs. |
+| `scope` | Trusted, assignable expression holding the client session table. Defaults to `globalThis.__d`. |
+| `signal` | An `AbortSignal` that stops the tail. The next `next()` yields one final block that rejects every promise still pending on the client with a generic `Error` and deletes the client session, then the tail is done. If the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. |
+| `transformError` | `(error) => unknown`, called with a rejection reason, or with the error that kept a resolved value from serializing. Its result (or what its returned promise resolves to) is what the client promise rejects with. As with a replacer, returning `undefined`, `null` or `false` keeps the default: a generic `Error`, so server errors never reach the client unless you opt in. Throwing, rejecting, or returning something that can't be serialized also falls back to it. |
+
+#### Constraints
+
+Don't mutate the input until the tail has finished, and don't mutate the revived value until every block has been evaluated — blocks address existing objects by path.
+
+Only native promises are streamed. Other thenables need a replacer that rebuilds them around a native promise:
+
+```js
+devalue.unevalStream(value, (v, js) => {
+ if (v instanceof MyThenable) return js`new MyThenable(${Promise.resolve(v)})`;
+});
+```
+
### `unflatten`
In the case where devalued data is one part of a larger JSON string, `unflatten` allows you to revive just the bit you need:
diff --git a/src/index.js b/src/index.js
index bdb3dba1..b34b49e4 100644
--- a/src/index.js
+++ b/src/index.js
@@ -1,4 +1,5 @@
export { uneval } from './uneval.js';
+export { unevalStream } from './uneval-stream.js';
export { parse, unflatten } from './parse.js';
export { stringify, stringifyAsync } from './stringify.js';
export {
diff --git a/src/types.ts b/src/types.ts
index 7740adaa..7b30c4f5 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -64,6 +64,39 @@ export type UnevalReplacer = (
js: JavaScriptTag
) => JavaScriptFragment | false | null | void;
+/** Options for `unevalStream`. */
+export interface UnevalStreamOptions {
+ /** Session key. Pass a stable value when heads must be byte-identical across runs. */
+ id?: string;
+ /** Trusted assignable expression holding the client session table. Default `globalThis.__d`. */
+ scope?: string;
+ /**
+ * Stops the tail: the next `next()` yields a final block that rejects every promise
+ * still pending on the client with a generic Error and deletes the client session,
+ * then the tail is done. If the signal is already aborted, or aborts during the head
+ * emission, `unevalStream` throws `signal.reason`.
+ */
+ signal?: AbortSignal;
+ /**
+ * Turns a rejection reason, or the error that kept a resolved value from serializing,
+ * into the value the client promise rejects with. May return a promise. Like a
+ * replacer, returning `undefined`, `null` or `false` keeps the default: the client
+ * promise rejects with a generic Error. So does throwing, rejecting, or returning
+ * something that can't be serialized.
+ */
+ transformError?: (error: unknown) => unknown;
+}
+
+/** The result of `unevalStream`. */
+export interface UnevalStreamResult {
+ /** A JavaScript expression that recreates the value, with every promise pending. */
+ head: string;
+ /** Statement blocks that settle the promises created by `head` and earlier blocks. */
+ tail: AsyncIterableIterator;
+ /** The session key. */
+ id: string;
+}
+
/**
* The introspection/extraction operations `stringify` performs on the value
* being serialized. Every dynamic operation — property reads, prototype
diff --git a/src/uneval-stream.js b/src/uneval-stream.js
new file mode 100644
index 00000000..c0496a19
--- /dev/null
+++ b/src/uneval-stream.js
@@ -0,0 +1,515 @@
+/**
+ * @import { UnevalReplacer, UnevalStreamOptions, UnevalStreamResult } from './types.js';
+ */
+import { js, JavaScriptSource } from './javascript-source.js';
+import { analyze } from './uneval.js';
+import { call_replacer, get_type, stringify_string } from './utils.js';
+
+const RESERVED = ['s', 'n'];
+const FAILED = 'new Error("devalue: failed to serialize asynchronous value")';
+const ABORTED = 'new Error("devalue: stream aborted")';
+// Stands in for an outcome that is sent as FAILED rather than serialized
+const FAILURE = Symbol();
+const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined });
+const identifier = /^[_$a-zA-Z][_$a-zA-Z0-9]*$/;
+const then = Promise.prototype.then;
+const noop = () => {};
+// Bounds the length of a path's property/bracket text (at most 3 UTF-8 bytes per unit),
+// independently of numeric slot width.
+const MAX_PATH_LENGTH = 128;
+
+/**
+ * Turn a value containing promises into a JavaScript expression (`head`) that
+ * recreates it with pending promises, and a stream of statement blocks (`tail`)
+ * that settle them as they settle on the server.
+ * @param {any} value
+ * @param {UnevalReplacer} [replacer]
+ * @param {UnevalStreamOptions} [options]
+ * @returns {UnevalStreamResult}
+ */
+export function unevalStream(value, replacer, options = {}) {
+ if (options.signal?.aborted) throw options.signal.reason;
+ const scope = options.scope ?? 'globalThis.__d';
+ const id = options.id ?? random_id();
+ const key = stringify_string(id);
+ const session = new Session(replacer, options, `${scope}[${key}]`);
+ const head = start(session, value, scope, key);
+ return { head, tail: session, id };
+}
+
+/**
+ * Emits the head for a new session. Assigned inside `Session` so it can reach private state.
+ * @type {(session: Session, value: any, scope: string, key: string) => string}
+ */
+let start;
+
+/** Persistent request state is owned here, never by the input-bearing setup scope. */
+class Session {
+ static {
+ start = (session, value, scope, key) => session.#start(value, scope, key);
+ }
+
+ /**
+ * @param {UnevalReplacer | undefined} replacer
+ * @param {UnevalStreamOptions} options
+ * @param {string} table
+ */
+ constructor(replacer, options, table) {
+ this.#replacer = replacer;
+ this.#transform_error = options.transformError;
+ this.#signal = options.signal;
+ this.#table = table;
+ this.#on_abort = this.#abort_stream.bind(this);
+ }
+
+ /** @type {UnevalReplacer | undefined} */
+ #replacer;
+ /** @type {UnevalStreamOptions['transformError']} */
+ #transform_error;
+ /** @type {AbortSignal | undefined} */
+ #signal;
+ /** @type {string} */
+ #table;
+ /** @type {() => void} */
+ #on_abort;
+
+ /**
+ * @param {any} value
+ * @param {string} scope
+ * @param {string} key
+ * @returns {string}
+ */
+ #start(value, scope, key) {
+ this.#signal?.addEventListener('abort', this.#on_abort);
+
+ let head;
+ try {
+ head = this.#emit_value(value, true);
+ } catch (error) {
+ this.#finish();
+ throw error;
+ }
+
+ // Aborted mid-emission: the head is dead on arrival, so fail like a pre-aborted signal
+ if (this.#abort) throw this.#abort.reason;
+
+ if (this.#pending === 0) {
+ this.#finish();
+ return head;
+ }
+
+ /**
+ * Wrap the head in an IIFE that sets up this stream's client-side state:
+ * - `n` is the shared `scope` table, created with a null prototype if missing
+ * - `s` is this stream's entry in `n`, keyed by `id`:
+ * - `s.o` holds slots for objects and primitives that later tail blocks reference
+ * - `s.p` holds the `[resolve, reject]` pair for each pending promise
+ * - `s.d(i)` creates deferred pending promise `i`, saving its resolvers in `s.p`. The
+ * no-op `catch` prevents unhandled rejections before user code attaches handlers
+ * - `s.r(i, k, v)` settles promise `i` with `v`, resolving when `k` is 0 and
+ * rejecting when `k` is 1, then drops its resolvers
+ * The head expression is then evaluated with `s` in scope
+ */
+ return (
+ `(()=>{let n=${scope}||(${scope}={__proto__:null}),s=(n[${key}]={o:[],p:[]});` +
+ `s.d=(i)=>{let p=new Promise((a,b)=>{s.p[i]=[a,b]});p.catch(()=>{});return p};` +
+ `s.r=(i,k,v)=>{s.p[i][k](v);delete s.p[i]};` +
+ `return ${head}})()`
+ );
+ }
+
+ /**
+ * Replacer results retained until termination, so each object is seen once
+ * @type {Map}
+ */
+ #replacer_cache = new Map();
+
+ /**
+ * Where each object that already exists on the client lives
+ * @type {Map}
+ */
+ #known = new Map();
+
+ /**
+ * Slots for long strings and bigints. We can vastly reduce the wire size of payloads
+ * containing repeated strings or bigints (whose textual length is essentially unbounded) by saving them
+ * in a slot, which lets us reuse them with very short syntax.
+ * @type {Map}
+ */
+ #primitives = new Map();
+
+ /**
+ * Promise settlements not yet emitted in a tail block
+ * @type {Array<{ index: number, ok: boolean, value: unknown }>}
+ */
+ #queue = [];
+
+ #next_slot = 0;
+ #next_promise = 0;
+ /** The number of pending promises that have been sent to the client but not yet resolved */
+ #pending = 0;
+ #done = false;
+
+ /**
+ * Clearing `current` when the stream ends prevents our pending promises from retaining no-longer-relevant state.
+ * Without this indirection, the callbacks passed to `then`/`enqueue` could strongly-reference our cache data or
+ * other unrelated stuff, which then can't be garbage-collected until the promise resolves. By passing them a pointer,
+ * we enable ourselves to null out that pointer when the stream is aborted or otherwise ends, which allows GC to occur
+ * even if there are latent promises.
+ * @type {{ current: Session | null }}
+ */
+ #delivery = { current: this };
+
+ /**
+ * In order to fulfill the AsyncIterableIterator contract (and avoid accidentally sending results out of order),
+ * we use a chain of promises to force subsequent calls to `#next` to resolve in the original call order.
+ */
+ #chain = Promise.resolve();
+
+ /**
+ * Resumes a `next()` waiting for a settlement or for the session to finish
+ * @type {(value?: unknown) => void}
+ */
+ #wake = noop;
+
+ /**
+ * Set by an abort and cleared once the final block that rejects the remaining
+ * client promises has been emitted
+ * @type {{ reason: unknown } | null}
+ */
+ #abort = null;
+
+ /**
+ * @param {any} value
+ * @param {boolean} head
+ */
+ #emit_value(value, head) {
+ const known = this.#known;
+ const replacer_cache = this.#replacer_cache;
+ const primitives = this.#primitives;
+ const replacer = this.#replacer;
+ const slot_start = this.#next_slot;
+ const promise_start = this.#next_promise;
+
+ // A failed head ends the session, so only tails need to stage their locations.
+ /** @type {Map} */
+ const locations = head ? known : new Map();
+ /** @type {Map} */
+ const retained = new Map();
+ /** @type {Map} */
+ const fresh = new Map();
+
+ /** @type {Map} */
+ const staged_primitives = new Map();
+ /** @type {Array<{ slot: number, literal: string }>} */
+ const primitive_assignments = [];
+
+ try {
+ const analysis = analyze(value, {
+ known: head ? undefined : (thing) => known.get(thing),
+ enter: (thing, parent, key) => {
+ if (parent !== undefined && key !== undefined) {
+ const up = /** @type {Location} */ (locations.get(parent));
+ const length = up.length + segment_length(key);
+ if (length <= MAX_PATH_LENGTH) {
+ locations.set(thing, { up, key, length });
+ return;
+ }
+ }
+ const slot = this.#next_slot++;
+ locations.set(thing, { up: null, key: slot, length: 0 });
+ // The root is assigned around the whole expression instead.
+ if (thing !== value) retained.set(thing, slot);
+ },
+ replace: (thing) => {
+ if (replacer) {
+ let source = replacer_cache.get(thing);
+ if (source === undefined) {
+ source = call_replacer(thing, replacer);
+ if (!this.#done) replacer_cache.set(thing, source);
+ }
+ if (source) return source;
+ }
+ if (get_type(thing) !== 'Promise' || !is_promise(thing)) return null;
+ const index = this.#next_promise++;
+ fresh.set(thing, index);
+ return JavaScriptSource.from(js`s.d(${index})`);
+ },
+ reserved: RESERVED
+ });
+
+ // a head without promises needs no session at all
+ if (head && fresh.size === 0) return analysis.render();
+
+ let code = analysis.render({
+ reference: expression,
+ retain: (thing) => {
+ const slot = retained.get(thing);
+ return slot === undefined ? undefined : `s.o[${slot}]`;
+ },
+ primitive: (thing, literal) => {
+ let slot = primitives.get(thing) ?? staged_primitives.get(thing);
+ if (slot === undefined) {
+ slot = this.#next_slot++;
+ staged_primitives.set(thing, slot);
+ primitive_assignments.push({ slot, literal });
+ }
+ return `s.o[${slot}]`;
+ }
+ });
+
+ const root = locations.get(value);
+ if (root !== undefined) code = `(s.o[${root.key}]=${code})`;
+ if (primitive_assignments.length) {
+ const initializers = primitive_assignments.map(
+ ({ slot, literal }) => `s.o[${slot}]=${literal}`
+ );
+ code = `(${initializers.join(',')},${code})`;
+ }
+ // Trusted callbacks can cancel synchronously during emission. Never publish
+ // request state or attach new delivery after terminal cleanup.
+ if (!this.#done) {
+ for (const [thing, slot] of staged_primitives) primitives.set(thing, slot);
+ if (!head) for (const [thing, location] of locations) known.set(thing, location);
+ for (const [promise, index] of fresh) {
+ this.#pending += 1;
+ Session.#subscribe(promise, index, this.#delivery);
+ }
+ }
+
+ return code;
+ } catch (error) {
+ this.#next_slot = slot_start;
+ this.#next_promise = promise_start;
+ throw error;
+ }
+ }
+
+ /**
+ * @param {number} index
+ * @param {boolean} ok
+ * @param {unknown} value
+ */
+ #enqueue(index, ok, value) {
+ if (this.#done) return;
+ this.#queue.push({ index, ok, value });
+ this.#wake();
+ }
+
+ /**
+ * @param {number} index
+ * @param {boolean} ok
+ * @param {unknown} value
+ */
+ #settle(index, ok, value) {
+ if (value === FAILURE) return `s.r(${index},1,${FAILED})`;
+ try {
+ return `s.r(${index},${ok ? 0 : 1},${this.#emit_value(value, false)})`;
+ } catch (error) {
+ // A transformed error that can't be serialized is not transformed again.
+ if (!ok) return `s.r(${index},1,${FAILED})`;
+ this.#fail(index, error);
+ return null;
+ }
+ }
+
+ /**
+ * Queues the rejection of promise `index` with `error` (its rejection reason, or the
+ * error that kept its value from serializing) passed through `transformError`.
+ * Without one, when it fails, or when it opts out like a replacer (`undefined`,
+ * `null` or `false`), the client receives FAILED.
+ * @param {number} index
+ * @param {unknown} error
+ */
+ #fail(index, error) {
+ let result;
+ try {
+ result = this.#transform_error?.(error);
+ } catch {
+ return this.#enqueue(index, false, FAILURE);
+ }
+
+ if (result instanceof Promise) {
+ // As in #subscribe, the reaction only holds the detachable delivery cell.
+ const delivery = this.#delivery;
+ then.call(
+ result,
+ (value) => Session.#deliver(delivery, index, false, or_failure(value)),
+ () => Session.#deliver(delivery, index, false, FAILURE)
+ );
+ } else {
+ this.#enqueue(index, false, or_failure(result));
+ }
+ }
+
+ #finish() {
+ if (this.#done) return;
+ this.#done = true;
+ this.#delivery.current = null;
+ this.#signal?.removeEventListener('abort', this.#on_abort);
+ this.#known.clear();
+ this.#replacer_cache.clear();
+ this.#primitives.clear();
+ this.#queue.length = 0;
+ this.#replacer = undefined;
+ this.#transform_error = undefined;
+ this.#signal = undefined;
+ this.#on_abort = noop;
+ const wake = this.#wake;
+ this.#wake = noop;
+ wake();
+ }
+
+ #abort_stream() {
+ if (this.#done) return;
+ this.#abort = { reason: this.#signal?.reason };
+ this.#finish();
+ }
+
+ /** @returns {Promise>} */
+ async #step() {
+ for (;;) {
+ while (!this.#done && this.#queue.length === 0) {
+ await new Promise((resolve) => (this.#wake = resolve));
+ this.#wake = noop;
+ }
+
+ if (this.#done && !this.#abort) return DONE;
+
+ const statements = [];
+ // Synchronously transformed failures are queued behind and join this block.
+ const queue = this.#queue;
+ for (let i = 0; i < queue.length; i += 1) {
+ const { index, ok, value } = queue[i];
+ const statement = this.#settle(index, ok, value);
+ if (statement === null) continue;
+ this.#pending -= 1;
+ statements.push(statement);
+ }
+ queue.length = 0;
+
+ // Checked after settling, since a replacer can abort while this block is emitted.
+ // Rejects everything still pending on the client, including any promises this
+ // block itself created after the abort.
+ const aborted = this.#abort !== null;
+ if (aborted) {
+ this.#abort = null;
+ statements.push(`s.p.forEach((p)=>p[1](${ABORTED}))`);
+ } else if (statements.length === 0) {
+ // Every outcome is waiting on an asynchronous transformError
+ continue;
+ }
+
+ const complete = aborted || this.#pending === 0;
+ if (complete) {
+ statements.push(`delete ${this.#table}`);
+ }
+
+ const value = `((s)=>{${statements.join(';')}})(${this.#table});`;
+ if (complete) this.#finish();
+ return { done: false, value };
+ }
+ }
+
+ /** @returns {Promise>} */
+ next() {
+ const result = this.#chain.then(this.#step.bind(this));
+ this.#chain = result.then(noop, noop);
+ return result;
+ }
+
+ /** @returns {Promise>} */
+ return() {
+ this.#finish();
+ return Promise.resolve(DONE);
+ }
+
+ [Symbol.asyncIterator]() {
+ return this;
+ }
+
+ /**
+ * Native reactions own only this detachable cell and the numeric index.
+ * @param {object} promise
+ * @param {number} index
+ * @param {{ current: Session | null }} delivery
+ */
+ static #subscribe(promise, index, delivery) {
+ then.call(
+ promise,
+ (value) => Session.#deliver(delivery, index, true, value),
+ (error) => {
+ const session = delivery.current;
+ if (session) session.#fail(index, error);
+ }
+ );
+ }
+
+ /**
+ * @param {{ current: Session | null }} delivery
+ * @param {number} index
+ * @param {boolean} ok
+ * @param {unknown} value
+ */
+ static #deliver(delivery, index, ok, value) {
+ const session = delivery.current;
+ if (session) session.#enqueue(index, ok, value);
+ }
+}
+
+/**
+ * `transformError` opts out with the same values as a replacer
+ * @param {unknown} value
+ */
+function or_failure(value) {
+ return value === undefined || value === null || value === false ? FAILURE : value;
+}
+
+/** @param {object} thing */
+function is_promise(thing) {
+ try {
+ then.call(thing, noop, noop);
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+/**
+ * A slot (`up` is null and `key` is its index) or a property `key` of another location.
+ * `length` is the length of the path text from the nearest slot.
+ * @typedef {{ up: Location | null, key: string | number, length: number }} Location
+ */
+
+/**
+ * @param {Location} location
+ * @returns {string}
+ */
+function expression(location) {
+ const segments = [];
+ while (location.up !== null) {
+ const key = location.key;
+ segments.push(typeof key === 'number' ? `[${key}]` : prop(key));
+ location = location.up;
+ }
+ return `s.o[${location.key}]${segments.reverse().join('')}`;
+}
+
+/**
+ * Length of the path segment for `key`, or Infinity past the cap
+ * @param {string | number} key
+ */
+function segment_length(key) {
+ if (typeof key === 'number') return String(key).length + 2;
+ if (key.length > MAX_PATH_LENGTH) return Infinity;
+ return identifier.test(key) ? key.length + 1 : prop(key).length;
+}
+
+/** @param {string} key */
+function prop(key) {
+ return identifier.test(key) ? `.${key}` : `[${stringify_string(key)}]`;
+}
+
+function random_id() {
+ return Math.random().toString(36).slice(2, 10);
+}
diff --git a/test/README.md b/test/README.md
index 49b7f2da..f5baec20 100644
--- a/test/README.md
+++ b/test/README.md
@@ -11,4 +11,6 @@ Unit tests are discovered from `src/**/*.test.js` and `test/**/*.test.js`. Brows
The base64 and string-escaping tests include independent decoder and UTF-8 transport checks. Keep these independent from encoder round trips so correlated encoding and decoding regressions cannot hide each other.
+`unevalStream` tests go through `expect_roundtrip(make, { replacer, globals, orders })` in `test/helpers/stream.js`. `make` receives a schedule whose `later(value)` and `later_reject(reason)` create promises that the harness settles in several orders (FIFO, LIFO, all at once, seeded shuffle). For each order, the harness evaluates the head and every tail block in fresh VM realms, both separately and concatenated into one script. It awaits every revived promise and checks structure and identity with `compare_topology`. It also asserts that the head is the same for every order, that the replacer was called once per object identity, and that the client session was deleted. New cases should be rows in the existing tables. Don't assert exact `head` strings except in escaping regressions.
+
Use `test.each` for independent input cases and `describe.each` for suites or nested case matrices. Wrap array-valued inputs in object rows so Vitest does not spread them into callback arguments, and create mutable values inside each test. Loops within a test are appropriate for building fixtures, checking one result's contents, or comparing growth across input sizes.
diff --git a/test/helpers/stream.js b/test/helpers/stream.js
new file mode 100644
index 00000000..36521cec
--- /dev/null
+++ b/test/helpers/stream.js
@@ -0,0 +1,281 @@
+import vm from 'node:vm';
+import { expect } from 'vitest';
+import { unevalStream } from '../../src/index.js';
+
+const DEFAULT_ORDERS = ['fifo', 'lifo', 'all-at-once', 'seeded-shuffle'];
+
+/**
+ * A fresh VM realm, with optional injected globals (e.g. classes used by a replacer)
+ * @param {Record} [globals]
+ */
+export function client(globals = {}) {
+ const context = vm.createContext({ ...globals });
+ context.globalThis = context;
+ return {
+ context,
+ /** @param {string} source */
+ head: (source) => vm.runInContext(`(${source})`, context),
+ /** @param {string} source */
+ block: (source) => vm.runInContext(source, context),
+ /**
+ * Evaluates the head and every block as one script
+ * @param {string} head
+ * @param {string[]} blocks
+ */
+ combined: (head, blocks) =>
+ vm.runInContext(`(function(){const root=(${head});${blocks.join('')}return root})()`, context)
+ };
+}
+
+/** @param {number} seed */
+export function seeded_random(seed) {
+ let state = seed >>> 0;
+ return () => {
+ state ^= state << 13;
+ state ^= state >>> 17;
+ state ^= state << 5;
+ return (state >>> 0) / 0x1_0000_0000;
+ };
+}
+
+/**
+ * @template T
+ * @param {T[]} values
+ * @param {() => number} random
+ */
+export function shuffle(values, random) {
+ for (let i = values.length - 1; i > 0; i -= 1) {
+ const j = Math.floor(random() * (i + 1));
+ [values[i], values[j]] = [values[j], values[i]];
+ }
+ return values;
+}
+
+const turn = () => new Promise((resolve) => setImmediate(resolve));
+
+/**
+ * `later(value, ms?)` / `later_reject(reason, ms?)` create promises that settle when
+ * `settle_in(order)` runs, or after `ms` milliseconds once it starts if `ms` is given
+ */
+export function create_schedule() {
+ /** @type {Array<() => void>} */
+ const ordered = [];
+ /** @type {Array<{ settle: () => void, ms: number }>} */
+ const timed = [];
+
+ /**
+ * @param {boolean} ok
+ * @param {unknown} value
+ * @param {number} [ms]
+ */
+ function create(ok, value, ms) {
+ const { promise, resolve, reject } = Promise.withResolvers();
+ // rejections nested in outcomes are only observed once the outcome is emitted
+ promise.catch(() => {});
+ const settle = () => (ok ? resolve(value) : reject(value));
+ if (ms === undefined) ordered.push(settle);
+ else timed.push({ settle, ms });
+ return promise;
+ }
+
+ return {
+ /** @param {unknown} value @param {number} [ms] */
+ later: (value, ms) => create(true, value, ms),
+ /** @param {unknown} reason @param {number} [ms] */
+ later_reject: (reason, ms) => create(false, reason, ms),
+ /** @param {string} order */
+ async settle_in(order) {
+ const timers = timed.map(({ settle, ms }) =>
+ new Promise((r) => setTimeout(r, ms)).then(settle)
+ );
+ const settles =
+ order === 'lifo'
+ ? ordered.slice().reverse()
+ : order === 'seeded-shuffle'
+ ? shuffle(ordered.slice(), seeded_random(ordered.length + 1))
+ : ordered;
+ if (order === 'all-at-once') {
+ for (const settle of settles) settle();
+ } else {
+ for (const settle of settles) {
+ settle();
+ await turn();
+ }
+ }
+ await Promise.all(timers);
+ }
+ };
+}
+
+/** @param {unknown} value */
+const type_of = (value) => Object.prototype.toString.call(value).slice(8, -1);
+
+/** @param {unknown} value */
+const is_object = (value) =>
+ value !== null && (typeof value === 'object' || typeof value === 'function');
+
+/** @param {PromiseLike} promise */
+const outcome = (promise) =>
+ Promise.resolve(promise).then(
+ (value) => ({ ok: true, value }),
+ (value) => ({ ok: false, value })
+ );
+
+/** @param {ArrayBufferLike} buffer @param {number} [offset] @param {number} [length] */
+const bytes = (buffer, offset, length) => Array.from(new Uint8Array(buffer, offset, length));
+
+/**
+ * Compares a server value with its revived client counterpart: structure, prototypes,
+ * key order, and identity (repeated server objects must map to one client object, and
+ * distinct ones must not collapse). Promises match when their settled outcomes match.
+ * @param {any} server
+ * @param {any} revived
+ * @param {string} [label]
+ * @param {Map} [server_to_client]
+ * @param {WeakMap} [client_to_server]
+ * @returns {Promise}
+ */
+export async function compare_topology(
+ server,
+ revived,
+ label = 'root',
+ server_to_client = new Map(),
+ client_to_server = new WeakMap()
+) {
+ if (!is_object(server)) {
+ expect(Object.is(revived, server), `${label}: primitive values differ`).toBe(true);
+ return;
+ }
+ if (server_to_client.has(server)) {
+ expect(revived, `${label}: repeated identity differs`).toBe(server_to_client.get(server));
+ return;
+ }
+ expect(is_object(revived), `${label}: expected an object`).toBe(true);
+ expect(client_to_server.has(revived), `${label}: distinct server nodes collapsed`).toBe(false);
+ server_to_client.set(server, revived);
+ client_to_server.set(revived, server);
+
+ /** @param {unknown} s @param {unknown} c @param {string} suffix */
+ const recurse = (s, c, suffix) =>
+ compare_topology(s, c, label + suffix, server_to_client, client_to_server);
+
+ const type = type_of(server);
+ expect(type_of(revived), `${label}: type differs`).toBe(type);
+
+ if (type === 'Promise') {
+ const [s, c] = await Promise.all([outcome(server), outcome(revived)]);
+ expect(c.ok, `${label}: settlement kind differs`).toBe(s.ok);
+ return recurse(s.value, c.value, s.ok ? '' : '');
+ }
+ if (['String', 'Number', 'Boolean', 'BigInt', 'Date'].includes(type)) {
+ expect(Object.is(revived.valueOf(), server.valueOf()), `${label}: boxed value differs`).toBe(
+ true
+ );
+ return;
+ }
+ if (['RegExp', 'URL', 'URLSearchParams'].includes(type)) {
+ expect(String(revived), `${label}: value differs`).toBe(String(server));
+ return;
+ }
+ if (type === 'ArrayBuffer') {
+ expect(bytes(revived), `${label}: buffer bytes differ`).toEqual(bytes(server));
+ return;
+ }
+ if (ArrayBuffer.isView(server)) {
+ expect(bytes(revived.buffer, revived.byteOffset, revived.byteLength)).toEqual(
+ bytes(server.buffer, server.byteOffset, server.byteLength)
+ );
+ // a Node Buffer's pooled backing store is never sent
+ if (Buffer.isBuffer(server)) return;
+ expect(revived.byteOffset, `${label}: view offset differs`).toBe(server.byteOffset);
+ return recurse(server.buffer, revived.buffer, '.buffer');
+ }
+ if (type === 'Map' || type === 'Set') {
+ const s = Array.from(server.entries());
+ const c = Array.from(revived.entries());
+ expect(c.length, `${label}: size differs`).toBe(s.length);
+ for (let i = 0; i < s.length; i += 1) {
+ await recurse(s[i][0], c[i][0], `.key[${i}]`);
+ if (type === 'Map') await recurse(s[i][1], c[i][1], `.value[${i}]`);
+ }
+ return;
+ }
+
+ const proto = Object.getPrototypeOf(server);
+ const revived_proto = Object.getPrototypeOf(revived);
+ if (proto === null || proto === Object.prototype || proto === Array.prototype) {
+ expect(revived_proto?.constructor?.name ?? null, `${label}: prototype differs`).toBe(
+ proto?.constructor.name ?? null
+ );
+ } else {
+ expect(revived_proto, `${label}: class differs`).toBe(proto);
+ }
+ if (Array.isArray(server)) expect(revived.length, `${label}: length differs`).toBe(server.length);
+
+ const keys = Object.keys(server);
+ expect(Object.keys(revived), `${label}: keys differ`).toEqual(keys);
+ for (const key of keys) await recurse(server[key], revived[key], `[${JSON.stringify(key)}]`);
+}
+
+/**
+ * Serializes `make(schedule)` once per settlement order, evaluates the result in fresh
+ * realms (block by block, and concatenated), and checks everything the contract promises.
+ * @param {(schedule: ReturnType) => unknown} make
+ * @param {{
+ * replacer?: import('../../src/types.js').UnevalReplacer,
+ * globals?: Record,
+ * orders?: string[]
+ * }} [options]
+ */
+export async function expect_roundtrip(make, { replacer, globals, orders = DEFAULT_ORDERS } = {}) {
+ /** @type {Set} */
+ const heads = new Set();
+ /** @type {{ head: string, blocks: string[], root: unknown }} */
+ let result = { head: '', blocks: [], root: undefined };
+
+ for (const order of orders) {
+ const schedule = create_schedule();
+ const value = make(schedule);
+
+ /** @type {Map} */
+ const calls = new Map();
+ /** @type {import('../../src/types.js').UnevalReplacer} */
+ const counted = (v, js) => {
+ calls.set(v, (calls.get(v) ?? 0) + 1);
+ return replacer?.(v, js);
+ };
+
+ // Rejections are checked like resolutions, so send their reasons as they are.
+ const { head, tail } = unevalStream(value, counted, {
+ id: 'roundtrip',
+ transformError: (error) => error
+ });
+ heads.add(head);
+
+ /** @type {string[]} */
+ const blocks = [];
+ const consumed = (async () => {
+ for await (const block of tail) blocks.push(block);
+ })();
+ await schedule.settle_in(order);
+ await consumed;
+
+ const separate = client(globals);
+ const root = separate.head(head);
+ for (const block of blocks) separate.block(block);
+ await compare_topology(value, root, `${order} (separate)`);
+
+ const concatenated = client(globals);
+ await compare_topology(value, concatenated.combined(head, blocks), `${order} (concatenated)`);
+
+ for (const [v, n] of calls) expect(n, `${order}: replacer calls for ${type_of(v)}`).toBe(1);
+ for (const { context } of [separate, concatenated]) {
+ expect(Object.keys(context.__d ?? {}), `${order}: session not deleted`).toEqual([]);
+ }
+
+ result = { head, blocks, root };
+ }
+
+ expect(heads.size, 'head depends on settlement order').toBe(1);
+ return result;
+}
diff --git a/test/helpers/subprocess.js b/test/helpers/subprocess.js
new file mode 100644
index 00000000..8c2cde6f
--- /dev/null
+++ b/test/helpers/subprocess.js
@@ -0,0 +1,27 @@
+import { execFileSync } from 'node:child_process';
+
+/**
+ * Keep strict unhandled-rejection failures (and explicit GC) out of the Vitest process.
+ * Fixtures are stringified, so they must not capture anything from their module.
+ * @param {(helpers: any) => unknown} fixture
+ * @param {{ expose_gc?: boolean, data?: unknown }} [options]
+ */
+export function in_subprocess(fixture, { expose_gc = false, data } = {}) {
+ execFileSync(
+ process.execPath,
+ [
+ '--unhandled-rejections=strict',
+ ...(expose_gc ? ['--expose-gc', '--max-old-space-size=128'] : []),
+ '--input-type=module',
+ '--eval',
+ `
+ import assert from 'node:assert/strict';
+ import vm from 'node:vm';
+ import { setImmediate as turn, setTimeout as delay } from 'node:timers/promises';
+ import { parse, stringify, stringifyAsync, unevalStream } from ${JSON.stringify(new URL('../../src/index.js', import.meta.url).href)};
+ await (${fixture})({ assert, vm, turn, delay, parse, stringify, stringifyAsync, unevalStream, data: ${JSON.stringify(data)} });
+ `
+ ],
+ { timeout: 5000, stdio: 'pipe' }
+ );
+}
diff --git a/test/stream-primitive-growth.test.js b/test/stream-primitive-growth.test.js
new file mode 100644
index 00000000..287528f9
--- /dev/null
+++ b/test/stream-primitive-growth.test.js
@@ -0,0 +1,202 @@
+import { setImmediate as turn } from 'node:timers/promises';
+import { expect, test } from 'vitest';
+import { parse, stringify, uneval, unevalStream } from '../src/index.js';
+import { client } from './helpers/stream.js';
+
+const primitives = [
+ { kind: 'string', make: (n) => 'x'.repeat(n * 8) },
+ { kind: 'escaped string', make: (n) => '\n"\\\0\u2028\u2029'.repeat(n) },
+ { kind: 'bigint', make: (n) => BigInt('9'.repeat(n * 4)) }
+];
+
+test.each(
+ ['batch', 'separate'].flatMap((mode) => primitives.map((primitive) => ({ mode, ...primitive })))
+)('$mode: repeated $kind settlements grow linearly', async ({ mode, make }) => {
+ const sizes = [];
+ for (const n of [128, 256]) {
+ const primitive = make(n);
+ const encoded = stringify(Array(n).fill(primitive));
+ const values = parse(encoded);
+ const controlled = mode === 'separate' ? values.map(() => Promise.withResolvers()) : [];
+ // These are distinct native promises, not repeated references to one promise.
+ const promises = values.map((value, i) =>
+ mode === 'separate' ? controlled[i].promise : Promise.resolve(value)
+ );
+ expect(new Set(promises).size).toBe(n);
+ const stream = unevalStream(promises);
+ const c = client();
+ const root = c.head(stream.head);
+ let source = stream.head;
+ const consuming = (async () => {
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+ })();
+ if (mode === 'separate') {
+ for (let i = 0; i < n; i += 1) {
+ controlled[i].resolve(values[i]);
+ await turn();
+ }
+ }
+ await consuming;
+
+ expect(root).toHaveLength(n);
+ expect(new Set(root).size).toBe(n);
+ for (const revived of await Promise.all(root)) expect(revived).toBe(primitive);
+ sizes.push({ n, input: encoded.length, output: source.length });
+ }
+
+ // All VM values are verified before checking budgets; do not prescribe batching.
+ expect
+ .soft(sizes[1].output / sizes[0].output, 'doubling the input must not quadruple output')
+ .toBeLessThan(2.5);
+ for (const { n, input, output } of sizes) {
+ expect
+ .soft(output, `n=${n}: output exceeds the linear input budget`)
+ .toBeLessThan(input * 8 + n * 192 + 4096);
+ }
+});
+
+test('control: primitive interning preserves custom cycles across discarded renderings', async () => {
+ class Cyclic {
+ constructor(text, meta) {
+ this.text = text;
+ this.meta = meta;
+ }
+ }
+ const text = '\n"\\\0\u2028\u2029'.repeat(128);
+ const value = new Cyclic(text, {});
+ value.meta.parent = value;
+ const first = Promise.withResolvers();
+ const second = Promise.withResolvers();
+ const stream = unevalStream({ first: first.promise, second: second.promise }, (v, js) =>
+ v instanceof Cyclic ? js`new Cyclic(${v.text},${v.meta})` : undefined
+ );
+ const c = client({ Cyclic });
+ const root = c.head(stream.head);
+
+ // Rendering text before meta's cycle can discard an outer constructor rendering.
+ // Its primitive-slot assignment must not be mistaken for an emitted assignment.
+ first.resolve(value);
+ c.block((await stream.tail.next()).value);
+ const revived = await root.first;
+ expect(revived).toBeInstanceOf(Cyclic);
+ expect(revived.text).toBe(text);
+ expect(revived.meta.parent).toBe(revived);
+ expect(revived.meta.parent.meta).toBe(revived.meta);
+
+ second.resolve(text);
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.second).toBe(text);
+});
+
+test('control: a failed emission does not poison a later settlement of the same primitive', async () => {
+ const text = '\n"\\\0\u2028\u2029'.repeat(128);
+ const first = Promise.withResolvers();
+ const second = Promise.withResolvers();
+ const stream = unevalStream({ first: first.promise, second: second.promise });
+ const c = client();
+ const root = c.head(stream.head);
+
+ // The eligible primitive is encountered before the unsupported late field.
+ first.resolve({ text, unsupported: Symbol('unsupported late property') });
+ c.block((await stream.tail.next()).value);
+ await expect(root.first).rejects.toThrow('devalue: failed to serialize asynchronous value');
+ second.resolve(text);
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.second).toBe(text);
+});
+
+test('a rendering failure discards staged primitives without losing earlier committed slots', async () => {
+ const committed = 'committed'.repeat(128);
+ const staged = 'staged'.repeat(128);
+ const first = Promise.withResolvers();
+ const second = Promise.withResolvers();
+ const stream = unevalStream({ committed, first: first.promise, second: second.promise });
+ const c = client();
+ const root = c.head(stream.head);
+ let reads = 0;
+ first.resolve({
+ staged,
+ get late() {
+ if (++reads > 1) throw new Error('failure after primitive rendering');
+ return null;
+ }
+ });
+ c.block((await stream.tail.next()).value);
+ expect(reads).toBe(2);
+ await expect(root.first).rejects.toThrow('devalue: failed to serialize asynchronous value');
+ second.resolve({ committed, staged });
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.second).toEqual({ committed, staged });
+});
+
+test.each(primitives)('head $kind slots are reused by tails', async ({ make }) => {
+ const primitive = make(128);
+ const deferred = Promise.withResolvers();
+ // Repetition exercises local primitive declarations as well as direct renders.
+ const stream = unevalStream({ values: [primitive, primitive], later: deferred.promise });
+ const c = client();
+ const root = c.head(stream.head);
+ expect(root.values).toEqual([primitive, primitive]);
+ deferred.resolve(primitive);
+ let source = '';
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+ expect(await root.later).toBe(primitive);
+ expect(source.length).toBeLessThan(primitive.toString().length / 2);
+});
+
+test('short strings and inexpensive bigints do not acquire primitive slots', async () => {
+ const short = '\n'.repeat(127); // Eligibility uses string length, not escaped length.
+ const bigint = BigInt('9'.repeat(126)); // 127 characters including the n suffix.
+ const deferred = Promise.withResolvers();
+ const stream = unevalStream({ short, bigint, later: deferred.promise });
+ const c = client();
+ const root = c.head(stream.head);
+ const session = c.context.__d[stream.id];
+ expect(
+ session.o.filter((value) => typeof value === 'string' || typeof value === 'bigint')
+ ).toEqual([]);
+ deferred.resolve({ short, bigint });
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.later).toEqual({ short, bigint });
+ expect(
+ session.o.filter((value) => typeof value === 'string' || typeof value === 'bigint')
+ ).toEqual([]);
+});
+
+test.each(primitives)('a no-promise $kind head preserves standalone output', async ({ make }) => {
+ const primitive = make(128);
+ const value = { primitive, repeated: [primitive, primitive] };
+ const stream = unevalStream(value);
+ expect(stream.head).toBe(uneval(value));
+ const c = client();
+ expect(c.head(stream.head)).toEqual(value);
+ expect(c.context.__d).toBeUndefined();
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+});
+
+test('primitive initialization composes with reference aliases and alternate slot bindings', async () => {
+ const key = 'path'.repeat(20);
+ const nodes = Array.from({ length: 32 }, (_, i) => ({ i }));
+ const text = 'primitive'.repeat(128);
+ const first = Promise.withResolvers();
+ const second = Promise.withResolvers();
+ const stream = unevalStream({ [key]: nodes, first: first.promise, second: second.promise });
+ const c = client();
+ const root = c.head(stream.head);
+ first.resolve({ refs: nodes.slice(), values: [text, text] });
+ c.block((await stream.tail.next()).value);
+ const revived = await root.first;
+ expect(revived.values).toEqual([text, text]);
+ for (let i = 0; i < nodes.length; i += 1) expect(revived.refs[i]).toBe(root[key][i]);
+ second.resolve({ refs: nodes.slice(), text });
+ for await (const block of stream.tail) c.block(block);
+ const reused = await root.second;
+ expect(reused.text).toBe(text);
+ for (let i = 0; i < nodes.length; i += 1) expect(reused.refs[i]).toBe(revived.refs[i]);
+});
diff --git a/test/stream-reference-growth.test.js b/test/stream-reference-growth.test.js
new file mode 100644
index 00000000..9718703c
--- /dev/null
+++ b/test/stream-reference-growth.test.js
@@ -0,0 +1,355 @@
+import { expect, test } from 'vitest';
+import { stringify, uneval, unevalStream } from '../src/index.js';
+import { client, compare_topology, expect_roundtrip } from './helpers/stream.js';
+
+/**
+ * @param {'long key' | 'deep chain' | 'short flat key'} kind
+ * @param {'head' | 'earlier tail'} origin
+ */
+async function reference_sizes(kind, origin) {
+ const sizes = [];
+ for (const n of [128, 256]) {
+ const nodes = Array.from({ length: n }, (_, i) => ({ i }));
+ let data;
+ let key;
+ if (kind === 'deep chain') {
+ for (let i = 0; i < nodes.length - 1; i += 1) nodes[i].next = nodes[i + 1];
+ data = nodes[0];
+ } else {
+ key = kind === 'long key' ? ''.repeat(n) : 'nodes';
+ data = { [key]: nodes };
+ }
+ // Include the fresh outcome array in the linear input budget, without promises.
+ const outcome = nodes.slice();
+ const encoded = stringify({ data, outcome });
+ const first = origin === 'earlier tail' ? Promise.withResolvers() : null;
+ const second = origin === 'earlier tail' ? Promise.withResolvers() : null;
+ const stream = unevalStream({
+ data: first ? first.promise : data,
+ outcome: second ? second.promise : Promise.resolve(outcome)
+ });
+ const c = client();
+ const root = c.head(stream.head);
+ let source = stream.head;
+ let revived_data = root.data;
+ if (first && second) {
+ first.resolve(data);
+ const block = await stream.tail.next();
+ expect(block.done).toBe(false);
+ source += block.value;
+ c.block(block.value);
+ revived_data = await root.data;
+ // Only settle the fresh shallow array after its ancestors exist on the client.
+ second.resolve(outcome);
+ }
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+
+ const revived = await root.outcome;
+ expect(revived).toHaveLength(n);
+ if (kind !== 'deep chain') expect(revived).not.toBe(revived_data[key]);
+ let node = kind === 'deep chain' ? revived_data : undefined;
+ for (let i = 0; i < n; i += 1) {
+ expect(revived[i]).toBe(kind === 'deep chain' ? node : revived_data[key][i]);
+ expect(revived[i].i).toBe(i);
+ if (kind === 'deep chain') node = node.next;
+ }
+ expect(new Set(revived).size).toBe(n);
+ sizes.push({ n, input: encoded.length, output: source.length });
+ }
+ return sizes;
+}
+
+test.each(
+ ['head', 'earlier tail'].flatMap((origin) =>
+ ['long key', 'deep chain', 'short flat key'].map((kind) => ({ origin, kind }))
+ )
+)(
+ '$origin: $kind references grow linearly across a later promise outcome',
+ async ({ origin, kind }) => {
+ // Identity checks for both sizes run before any output-size assertions.
+ const sizes = await reference_sizes(kind, origin);
+ expect
+ .soft(sizes[1].output / sizes[0].output, 'doubling the input must not quadruple output')
+ .toBeLessThan(2.5);
+ for (const { n, input, output } of sizes) {
+ // Allow escaping and generous per-node session/slot overhead, not repeated paths.
+ expect
+ .soft(output, `n=${n}: output exceeds the linear input budget`)
+ .toBeLessThan(input * 8 + n * 192 + 4096);
+ }
+ }
+);
+
+test('cheap single-use and repeated leaf paths do not create lazy slots', async () => {
+ const leaf = { value: 1 };
+ const stream = unevalStream({ leaf, outcome: Promise.resolve([leaf, leaf]) }, undefined, {
+ id: 'cheap'
+ });
+ const c = client();
+ const root = c.head(stream.head);
+ const slots = c.context.__d.cheap.o;
+ expect(slots).toHaveLength(1);
+ const blocks = [];
+ for await (const block of stream.tail) {
+ blocks.push(block);
+ c.block(block);
+ }
+ const outcome = await root.outcome;
+ expect(outcome[0]).toBe(root.leaf);
+ expect(outcome[1]).toBe(root.leaf);
+ // Only the new outcome array is retained, not its already-known leaf.
+ expect(slots).toHaveLength(2);
+ expect(blocks.join('')).toContain('s.o[0].leaf');
+});
+
+test('a long but bounded leaf read is repeated rather than given a slot', async () => {
+ const key = 'x'.repeat(80);
+ const leaf = { value: 1 };
+ const stream = unevalStream({ [key]: leaf, outcome: Promise.resolve(leaf) }, undefined, {
+ id: 'single'
+ });
+ const c = client();
+ const root = c.head(stream.head);
+ const slots = c.context.__d.single.o;
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.outcome).toBe(root[key]);
+ expect(slots).toHaveLength(1);
+});
+
+test('distinct descendants are reached by bounded paths from a numeric slot table', async () => {
+ const key = 'x'.repeat(80);
+ const nodes = Array.from({ length: 32 }, (_, i) => ({ i }));
+ const stream = unevalStream(
+ { [key]: nodes, outcome: Promise.resolve(nodes.slice()) },
+ undefined,
+ {
+ id: 'shared'
+ }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+ const slots = c.context.__d.shared.o;
+ expect(slots).toHaveLength(1); // below the ceiling: no speculative head slots
+ let source = stream.head;
+ let tail = '';
+ for await (const block of stream.tail) {
+ source += block;
+ tail += block;
+ c.block(block);
+ }
+ const outcome = await root.outcome;
+ for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]);
+ // Each reference repeats a path of at most MAX_PATH_LENGTH characters; compression absorbs the repeats.
+ expect(tail.split(key)).toHaveLength(nodes.length + 1);
+ expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i)));
+ expect(source.length).toBeLessThan(nodes.length * 128 + 1024);
+});
+
+test.each(['escaped segment', 'unicode segment', 'enormous segment', 'independent chains'])(
+ '%s forces ceiling anchors even for a single later read',
+ async (kind) => {
+ const leaf = { value: 42 };
+ let data = leaf;
+ if (kind === 'independent chains') {
+ for (let i = 0; i < 80; i += 1) data = { next: data };
+ data = { left: data, right: structuredClone(data) };
+ } else {
+ const key =
+ kind === 'unicode segment'
+ ? 'é'.repeat(129)
+ : '<'.repeat(kind === 'escaped segment' ? 24 : 4096);
+ data = { [key]: leaf };
+ }
+ const stream = unevalStream({ data, outcome: Promise.resolve(leaf) }, undefined, {
+ id: 'ceiling'
+ });
+ const c = client();
+ const root = c.head(stream.head);
+ const slots = c.context.__d.ceiling.o;
+ const before = slots.length;
+ expect(before).toBeGreaterThan(1);
+ let tail = '';
+ for await (const block of stream.tail) {
+ tail += block;
+ c.block(block);
+ }
+ let revived = root.data;
+ if (kind === 'independent chains') {
+ expect(before).toBeGreaterThanOrEqual(7);
+ revived = revived.left;
+ for (let i = 0; i < 80; i += 1) revived = revived.next;
+ } else revived = revived[Object.keys(data)[0]];
+ expect(await root.outcome).toBe(revived);
+ expect(slots).toHaveLength(before); // no profitable alias or new object
+ expect(tail.length).toBeLessThan(256);
+ expect(tail).not.toContain('\\u003C');
+ if (kind === 'enormous segment') {
+ // Includes the unavoidable original key literal, not just the tail.
+ expect(stream.head.length + tail.length).toBeLessThan(stringify(data).length * 8 + 4096);
+ }
+ }
+);
+
+test('no-promise heads remain byte-compatible even when paths exceed the ceiling', async () => {
+ const data = { ['<'.repeat(4096)]: { value: 1 } };
+ const stream = unevalStream(data);
+ expect(stream.head).toBe(uneval(data));
+ expect(stream.head).not.toContain('s.o');
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+});
+
+test('a failed emission rolls back forced slots, lazy aliases and promise indices', async () => {
+ const key = 'x'.repeat(80);
+ const nodes = Array.from({ length: 32 }, (_, i) => ({ i }));
+ const first = Promise.withResolvers();
+ const second = Promise.withResolvers();
+ const nested = Promise.resolve('nested');
+ const errors = [];
+ const stream = unevalStream(
+ { [key]: nodes, first: first.promise, second: second.promise },
+ undefined,
+ {
+ id: 'rollback',
+ transformError: (error) => {
+ errors.push(error);
+ throw error;
+ }
+ }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+ const slots = c.context.__d.rollback.o;
+ // Fail during rendering (the second read), after demand planning has staged aliases.
+ let reads = 0;
+ const failing = {
+ refs: nodes.slice(),
+ ['y'.repeat(256)]: { nested },
+ get bad() {
+ if (++reads === 2) throw new Error('render failed');
+ return 1;
+ }
+ };
+ first.resolve(failing);
+ const failed = await stream.tail.next();
+ c.block(failed.value);
+ await expect(root.first).rejects.toThrow('failed to serialize');
+ expect(errors).toHaveLength(1);
+ expect(slots).toHaveLength(1);
+ second.resolve({ refs: nodes.slice(), nested });
+ let tail = '';
+ for await (const block of stream.tail) {
+ tail += block;
+ c.block(block);
+ }
+ const outcome = await root.second;
+ for (let i = 0; i < nodes.length; i += 1) expect(outcome.refs[i]).toBe(root[key][i]);
+ expect(await outcome.nested).toBe('nested');
+ expect(tail).toContain('s.d(2)');
+ expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i)));
+});
+
+test('successful aliases in one block survive a later failing settlement', async () => {
+ const key = 'x'.repeat(80);
+ const nodes = Array.from({ length: 32 }, (_, i) => ({ i }));
+ const promises = Array.from({ length: 3 }, () => Promise.withResolvers());
+ const stream = unevalStream({ [key]: nodes, outcomes: promises.map((p) => p.promise) });
+ const c = client();
+ const root = c.head(stream.head);
+ promises[0].resolve(nodes.slice());
+ promises[1].resolve({ bad: () => {} });
+ await Promise.resolve();
+ const block = await stream.tail.next();
+ c.block(block.value);
+ const outcome = await root.outcomes[0];
+ await expect(root.outcomes[1]).rejects.toThrow('failed to serialize');
+ promises[2].resolve(nodes.slice());
+ for await (const block of stream.tail) c.block(block);
+ const reused = await root.outcomes[2];
+ for (let i = 0; i < nodes.length; i += 1) {
+ expect(outcome[i]).toBe(root[key][i]);
+ expect(reused[i]).toBe(outcome[i]);
+ }
+});
+
+test('forced anchors preserve cycles, sparse arrays and binary sharing without pooled bytes', async () => {
+ const buffer = new ArrayBuffer(16);
+ new Uint8Array(buffer).set([1, 2, 3, 4]);
+ const pooled = Buffer.allocUnsafe(3);
+ pooled.set([7, 8, 9]);
+ const sparse = [];
+ sparse.length = 0xffffffff;
+ sparse[4000000000] = { value: 1 };
+ const data = {
+ buffer,
+ view: new Uint16Array(buffer, 2, 3),
+ other: new Uint8Array(buffer, 1, 5),
+ data_view: new DataView(buffer, 3, 5),
+ pooled,
+ sparse,
+ set: new Set(),
+ map: new Map()
+ };
+ data.self = data;
+ data.set.add(data);
+ data.map.set(data, sparse);
+ const key = '<'.repeat(24);
+ const value = { [key]: data, outcome: Promise.resolve([data, buffer, data.view, sparse]) };
+ const stream = unevalStream(value);
+ const c = client();
+ const root = c.head(stream.head);
+ let source = stream.head;
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+ await compare_topology(value, root);
+ expect(root[key].pooled.buffer.byteLength).toBe(3);
+ expect(source.length).toBeLessThan(3000);
+});
+
+test('compact unshared typed-array emission is not changed by speculative buffer retention', async () => {
+ const view = new Float64Array([1.5, -0, 2.5]);
+ const stream = unevalStream({ view, outcome: Promise.resolve(view) });
+ expect(stream.head).toContain('new Float64Array([1.5,-0,2.5])');
+ expect(stream.head).not.toContain('new Uint8Array');
+ const c = client();
+ const root = c.head(stream.head);
+ for await (const block of stream.tail) c.block(block);
+ expect(await root.outcome).toBe(root.view);
+});
+
+test.each(['typed view', 'DataView'])(
+ '%s preserves its buffer when the buffer path independently crosses the ceiling',
+ async (kind) => {
+ const buffer = new ArrayBuffer(24);
+ new Uint8Array(buffer).set([1, 2, 3, 4, 5, 6, 7, 8]);
+ const view =
+ kind === 'typed view' ? new Float64Array(buffer, 8, 1) : new DataView(buffer, 3, 5);
+ const key = 'x'.repeat(124);
+ const value = { [key]: view, outcome: Promise.resolve([view, buffer]) };
+ const stream = unevalStream(value);
+ const c = client();
+ const root = c.head(stream.head);
+ expect(stream.head).toContain('new Uint8Array');
+ for await (const block of stream.tail) c.block(block);
+ await compare_topology(value, root);
+ expect((await root.outcome)[1]).toBe(root[key].buffer);
+ }
+);
+
+test('replaced containers retain identity', async () => {
+ class Box {
+ constructor(value) {
+ this.value = value;
+ }
+ }
+ const key = 'x'.repeat(80);
+ const nodes = Array.from({ length: 32 }, (_, i) => new Box({ i }));
+ await expect_roundtrip((schedule) => ({ [key]: nodes, outcome: schedule.later(nodes.slice()) }), {
+ replacer: (value, js) => value instanceof Box && js`new Box(${value.value})`,
+ globals: { Box }
+ });
+});
diff --git a/test/stream-regressions.test.js b/test/stream-regressions.test.js
new file mode 100644
index 00000000..945d59f2
--- /dev/null
+++ b/test/stream-regressions.test.js
@@ -0,0 +1,152 @@
+import { expect, test } from 'vitest';
+import { parse, stringify, unevalStream } from '../src/index.js';
+import { client, expect_roundtrip } from './helpers/stream.js';
+
+/** @param {string} source */
+function utf8_transport(source) {
+ const bytes = new TextEncoder().encode(source);
+ const decoder = new TextDecoder();
+ let reconstructed = '';
+ for (const byte of bytes) reconstructed += decoder.decode(Uint8Array.of(byte), { stream: true });
+ return reconstructed + decoder.decode();
+}
+
+test('survives UTF-8 byte boundaries and separate or concatenated VM evaluation', async () => {
+ const resolved = Promise.withResolvers();
+ const failed = Promise.withResolvers();
+ const shared = { label: 'shared-😀' };
+ const reason = 'late rejection 😀 \udfff';
+ const id = 'transport-001-😀--\ud800';
+ const value = {
+ numeric: { 0: '0', 12: '12', '001': '001' },
+ lone: '\ud800',
+ pair: '😀',
+ closing: '',
+ shared,
+ resolved: resolved.promise,
+ rejected: failed.promise
+ };
+ const result = unevalStream(value, undefined, { id, transformError: (error) => error });
+ expect(result.id).toBe(id);
+ resolved.resolve({ shared, again: shared });
+ const sources = [result.head, /** @type {string} */ ((await result.tail.next()).value)];
+ failed.reject(reason);
+ for await (const block of result.tail) sources.push(block);
+ expect(sources.length).toBe(3);
+ for (const source of sources) expect(utf8_transport(source)).toBe(source);
+ expect(sources.join('')).not.toMatch(/<\/script/i);
+
+ /** @param {any} root */
+ async function verify(root) {
+ expect(Object.keys(root.numeric)).toEqual(['0', '12', '001']);
+ expect(root.lone).toBe('\ud800');
+ expect(root.pair).toBe('😀');
+ expect(root.closing).toBe(value.closing);
+ const outcome = await root.resolved;
+ expect(outcome.shared).toBe(root.shared);
+ expect(outcome.again).toBe(root.shared);
+ await expect(root.rejected).rejects.toBe(reason);
+ }
+
+ const [head, ...blocks] = sources.map(utf8_transport);
+ const separate = client();
+ const root = separate.head(head);
+ for (const block of blocks) separate.block(block);
+ await verify(root);
+ await verify(client().combined(head, blocks));
+});
+
+/**
+ * @param {unknown} value
+ * @param {'head' | 'tail'} mode
+ * @param {import('../src/types.js').UnevalReplacer} [replacer]
+ * @param {Record} [globals]
+ */
+async function roundtrip(value, mode, replacer, globals) {
+ const { head, blocks, root } = await expect_roundtrip(
+ ({ later }) => (mode === 'head' ? value : later(value)),
+ { replacer, globals, orders: ['fifo'] }
+ );
+ return { restored: mode === 'head' ? root : await root, source: head + blocks.join('') };
+}
+
+class Box {
+ /** @param {unknown} value */
+ constructor(value) {
+ this.value = value;
+ }
+}
+
+test.each(
+ ['head', 'tail'].flatMap((mode) =>
+ [
+ { kind: 'string', make: (/** @type {number} */ n) => 'x'.repeat(n) },
+ {
+ kind: 'escaped string',
+ make: (/** @type {number} */ n) => '\n"\\\0\u2028\u2029'.repeat(n)
+ },
+ { kind: 'bigint', make: (/** @type {number} */ n) => BigInt('9'.repeat(n)) }
+ ].map((row) => ({ mode: /** @type {'head' | 'tail'} */ (mode), ...row }))
+ )
+)('$mode: repeated $kind output grows linearly', async ({ mode, make }) => {
+ let previous = 0;
+ for (const n of [2000, 4000]) {
+ const encoded = stringify(Array(n).fill(make(n)));
+ const { source } = await roundtrip(parse(encoded), mode);
+ expect(source.length).toBeLessThan(encoded.length * 4);
+ if (previous) expect(source.length).toBeLessThan(previous * 2.1);
+ previous = source.length;
+ expect(source).not.toContain('<');
+ }
+});
+
+test.each(
+ ['head', 'tail'].flatMap((mode) =>
+ ['inline', 'shared', 'cyclic', 'holes', 'custom-cycle'].map((kind) => ({
+ mode: /** @type {'head' | 'tail'} */ (mode),
+ kind
+ }))
+ )
+)('$mode: $kind sparse arrays avoid eager allocation', async ({ mode, kind }) => {
+ // eager allocation would need ~20GB; parse creates dictionary-mode arrays
+ const length = 1_000_000;
+ const arrays = Array.from({ length: 2500 }, () => parse(`[[-7,${length},0,1],42]`));
+ for (const array of arrays) {
+ if (kind === 'holes') delete array[0];
+ else
+ array[1] = kind === 'cyclic' ? array : kind === 'custom-cycle' ? new Box(array) : undefined;
+ }
+ const value = kind === 'shared' ? [arrays, arrays.slice()] : arrays;
+ const { restored } = await roundtrip(
+ value,
+ mode,
+ (v, js) => (v instanceof Box ? js`new Box(${v.value})` : undefined),
+ { Box }
+ );
+ const result = kind === 'shared' ? restored[0] : restored;
+ for (const i of [0, arrays.length - 1]) {
+ expect(result[i].length).toBe(length);
+ expect(Object.getOwnPropertyNames(result[i])).toEqual(
+ kind === 'holes' ? ['length'] : ['0', '1', 'length']
+ );
+ }
+});
+
+test.each([{ mode: /** @type {const} */ ('head') }, { mode: /** @type {const} */ ('tail') }])(
+ '$mode: Buffers preserve identity without exposing their shared pool',
+ async ({ mode }) => {
+ const allocation = Buffer.allocUnsafe(128).fill(255);
+ const first = allocation.subarray(8, 11);
+ const second = allocation.subarray(24, 26);
+ first.set([1, 2, 3]);
+ second.set([4, 5]);
+ /** @type {Record} */
+ const value = { first, again: first, second };
+ value.self = value;
+ const { restored, source } = await roundtrip(value, mode);
+ expect(restored.first.buffer.byteLength).toBe(3);
+ expect(restored.second.buffer.byteLength).toBe(2);
+ expect(restored.first.buffer).not.toBe(restored.second.buffer);
+ expect(source).not.toContain('255');
+ }
+);
diff --git a/test/stream-retention.test.js b/test/stream-retention.test.js
new file mode 100644
index 00000000..47a96962
--- /dev/null
+++ b/test/stream-retention.test.js
@@ -0,0 +1,309 @@
+import { test } from 'vitest';
+import { in_subprocess } from './helpers/subprocess.js';
+
+async function retention({ assert, turn, unevalStream, data: mode }) {
+ assert.equal(typeof global.gc, 'function');
+
+ function create_root() {
+ const source = Promise.withResolvers();
+ const graph = { nodes: Array.from({ length: 1000 }, (_, i) => ({ i })) };
+ graph.self = graph;
+ if (mode !== 'plain replacer') graph.pending = source.promise;
+ const weak = new WeakRef(graph);
+ const node_weak = new WeakRef(graph.nodes[500]);
+ const controller = new AbortController();
+ const result =
+ mode === 'sanity'
+ ? null
+ : unevalStream(graph, mode === 'plain replacer' ? () => undefined : undefined, {
+ signal: controller.signal
+ });
+ // Only the live control keeps an explicit input reference. This does not require
+ // active streams to use strong caches rather than an otherwise valid WeakMap.
+ return {
+ source,
+ weak,
+ node_weak,
+ controller,
+ result,
+ live_graph: mode === 'live' ? graph : null
+ };
+ }
+
+ async function cancel(fixture) {
+ const next = fixture.result.tail.next();
+ if (mode.startsWith('return')) {
+ assert.deepEqual(await fixture.result.tail.return(), { done: true, value: undefined });
+ assert.deepEqual(await next, { done: true, value: undefined });
+ } else {
+ fixture.controller.abort(new Error('abort stream'));
+ const last = await next;
+ assert.equal(last.done, false);
+ assert.match(last.value, /devalue: stream aborted/);
+ }
+ if (!mode.endsWith('retained')) fixture.result = null;
+ fixture.controller = null;
+ }
+
+ const fixture = create_root();
+ let settled = false;
+ fixture.source.promise.then(
+ () => {
+ settled = true;
+ },
+ () => {
+ settled = true;
+ }
+ );
+ if (mode.startsWith('return') || mode.startsWith('abort')) {
+ await cancel(fixture);
+ } else if (mode === 'complete') {
+ fixture.source.resolve(42);
+ for await (const block of fixture.result.tail) assert.equal(typeof block, 'string');
+ } else if (mode === 'plain replacer') {
+ assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined });
+ }
+
+ // A deref during this loop would pin the target for the current job. Check only
+ // after repeated GCs in distinct jobs, including a turn after WeakRef creation.
+ for (let i = 0; i < 20; i += 1) {
+ await turn();
+ global.gc();
+ }
+ assert.ok(fixture.source.promise instanceof Promise, 'source remains externally held');
+ assert.equal(settled, mode === 'complete', 'cancelled sources must remain pending');
+ if (mode === 'live') {
+ assert.ok(fixture.result.head.length);
+ assert.equal(fixture.weak.deref(), fixture.live_graph);
+ assert.ok(fixture.node_weak.deref());
+ } else {
+ if (mode === 'complete' || mode === 'plain replacer' || mode.endsWith('retained')) {
+ assert.ok(fixture.result.head.length, 'the finished result remains externally held');
+ }
+ assert.ok(fixture.weak.deref() === undefined, `${mode}: stream retained the input root`);
+ assert.ok(fixture.node_weak.deref() === undefined, `${mode}: stream retained an input node`);
+ }
+ // Only after proving collection with the source still held and pending, show
+ // that cancellation did not cancel the job and late delivery stays detached.
+ if (mode.startsWith('return') || mode.startsWith('abort')) {
+ fixture.source.resolve(42);
+ assert.equal(await fixture.source.promise, 42);
+ if (fixture.result) {
+ assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined });
+ }
+ }
+}
+
+test.each(['return', 'abort', 'return retained', 'abort retained'])(
+ '%s releases the graph while its native source stays pending',
+ (mode) => {
+ in_subprocess(retention, { expose_gc: true, data: mode });
+ }
+);
+
+test('natural completion releases the input graph with the result retained', () => {
+ in_subprocess(retention, { expose_gc: true, data: 'complete' });
+});
+
+test('a no-promise head with a replacer releases the graph with the result retained', () => {
+ in_subprocess(retention, { expose_gc: true, data: 'plain replacer' });
+});
+
+test('control: the same graph collects without a stream retaining it', () => {
+ in_subprocess(retention, { expose_gc: true, data: 'sanity' });
+});
+
+test('control: a live stream with an externally held input keeps its graph', () => {
+ in_subprocess(retention, { expose_gc: true, data: 'live' });
+});
+
+async function callback_retention({ assert, vm, turn, unevalStream, data: callback }) {
+ assert.equal(typeof global.gc, 'function');
+ let reports = 0;
+
+ function create_result() {
+ const graph = { nodes: Array.from({ length: 1000 }, (_, i) => ({ i })), visits: 0 };
+ const weak = new WeakRef(graph);
+ const replacer = () => {
+ graph.visits += 1;
+ return undefined;
+ };
+ const transformError = () => {
+ reports += graph.nodes.length;
+ };
+ const options = callback === 'transformError' ? { transformError } : {};
+ // The graph is callback-only data, never an input node or cached replacement.
+ const result = unevalStream(
+ callback === 'replacer' ? { value: 42 } : Promise.resolve(42),
+ callback === 'replacer' ? replacer : undefined,
+ options
+ );
+ if (callback === 'replacer') assert.ok(graph.visits > 0);
+ // No caller-held graph, callback or options reference escapes this helper.
+ return { weak, result };
+ }
+
+ const fixture = create_result();
+ const context = vm.createContext({});
+ const root = vm.runInContext(`(${fixture.result.head})`, context);
+ for await (const block of fixture.result.tail) vm.runInContext(block, context);
+ if (callback === 'replacer') assert.equal(root.value, 42);
+ else assert.equal(await root, 42);
+ assert.equal(reports, 0, 'supported outcomes must not invoke transformError');
+
+ // No reporter is active; a retained terminal result must release its callbacks.
+ // As above, do not deref until all GCs have run in separate jobs.
+ for (let i = 0; i < 20; i += 1) {
+ await turn();
+ global.gc();
+ }
+ assert.ok(fixture.result.head.length, 'the terminal result remains externally held');
+ assert.ok(
+ fixture.weak.deref() === undefined,
+ `terminal result retained the ${callback} callback's otherwise unreferenced graph`
+ );
+}
+
+test.each(['replacer', 'transformError'])(
+ 'a retained terminal result releases graph-capturing %s callbacks',
+ (callback) => in_subprocess(callback_retention, { expose_gc: true, data: callback })
+);
+
+test('control: a retained terminal result does not retain an unrelated helper graph', () => {
+ in_subprocess(callback_retention, { expose_gc: true, data: 'none' });
+});
+
+test('a live stream retains identities until the final delivery', () => {
+ in_subprocess(
+ async ({ assert, vm, turn, unevalStream }) => {
+ function create_result() {
+ const source = Promise.withResolvers();
+ const graph = { node: {}, pending: source.promise };
+ return { source, weak: new WeakRef(graph), result: unevalStream(graph) };
+ }
+ const fixture = create_result();
+ const context = vm.createContext({});
+ const root = vm.runInContext(`(${fixture.result.head})`, context);
+ for (let i = 0; i < 20; i += 1) {
+ await turn();
+ global.gc();
+ }
+ const graph = fixture.weak.deref();
+ assert.ok(graph, 'the active session must retain identities without caller-held input');
+ fixture.source.resolve({ root: graph, node: graph.node });
+ for await (const block of fixture.result.tail) vm.runInContext(block, context);
+ const delivered = await root.pending;
+ assert.equal(delivered.root, root);
+ assert.equal(delivered.node, root.node);
+ assert.equal(Object.keys(context.__d).length, 0, 'final block deletes the client session');
+ },
+ { expose_gc: true }
+ );
+});
+
+test.each(['return', 'abort'])(
+ '%s releases queued outcomes and wakes ordered concurrent readers',
+ (mode) => {
+ in_subprocess(
+ async ({ assert, turn, unevalStream, data: mode }) => {
+ function create_result(signal) {
+ const queued = Promise.withResolvers();
+ const pending = Promise.withResolvers();
+ const graph = { node: {} };
+ const result = unevalStream(
+ { queued: queued.promise, pending: pending.promise },
+ undefined,
+ { signal }
+ );
+ queued.resolve(graph);
+ return { result, pending, weak: new WeakRef(graph) };
+ }
+ const controller = new AbortController();
+ const fixture = create_result(controller.signal);
+ // Use a separate pending stream to check ordered wakeup, while the first
+ // stream has an undelivered graph-bearing outcome.
+ const waiting = unevalStream(fixture.pending.promise, undefined, {
+ signal: controller.signal
+ });
+ const readers = [waiting.tail.next(), waiting.tail.next(), waiting.tail.next()];
+ await turn();
+ if (mode === 'abort') {
+ controller.abort();
+ const last = await readers[0];
+ assert.equal(last.done, false);
+ assert.match(last.value, /devalue: stream aborted/);
+ } else {
+ await fixture.result.tail.return();
+ await waiting.tail.return();
+ assert.deepEqual(await readers[0], { done: true, value: undefined });
+ }
+ for (const reader of readers.slice(1)) {
+ assert.deepEqual(await reader, { done: true, value: undefined });
+ }
+ if (mode === 'abort') {
+ const last = await fixture.result.tail.next();
+ assert.equal(last.done, false);
+ assert.match(last.value, /devalue: stream aborted/);
+ assert.doesNotMatch(last.value, /node/, 'the queued outcome is dropped');
+ }
+ for (let i = 0; i < 20; i += 1) {
+ await turn();
+ global.gc();
+ }
+ assert.ok(fixture.weak.deref() === undefined, 'terminal queue releases its outcome');
+ assert.ok(fixture.pending.promise instanceof Promise, 'source is still externally held');
+ fixture.pending.resolve(42);
+ assert.equal(await fixture.pending.promise, 42);
+ assert.deepEqual(await waiting.tail.next(), { done: true, value: undefined });
+ },
+ { expose_gc: true, data: mode }
+ );
+ }
+);
+
+test.each(['return', 'abort'])(
+ 'synchronous %s during a replacer cannot republish terminal state',
+ (mode) => {
+ in_subprocess(
+ async ({ assert, turn, unevalStream, data: mode }) => {
+ function create_result() {
+ const source = Promise.withResolvers();
+ const pending = Promise.withResolvers();
+ const controller = new AbortController();
+ const graph = { node: {}, pending: pending.promise, text: 'long text '.repeat(100) };
+ let result;
+ result = unevalStream(
+ source.promise,
+ (thing) => {
+ if (thing === graph) {
+ if (mode === 'return') result.tail.return();
+ else controller.abort(0);
+ }
+ return undefined;
+ },
+ { signal: controller.signal }
+ );
+ source.resolve(graph);
+ return { result, pending, weak: new WeakRef(graph) };
+ }
+ const fixture = create_result();
+ const block = await fixture.result.tail.next();
+ if (mode === 'abort') {
+ // The block being emitted when the abort happened carries the final rejections
+ assert.match(block.value, /devalue: stream aborted/);
+ }
+ assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined });
+ for (let i = 0; i < 20; i += 1) {
+ await turn();
+ global.gc();
+ }
+ assert.ok(fixture.result.head.length, 'terminal result stays reachable');
+ assert.ok(fixture.weak.deref() === undefined, 'emission did not republish the graph');
+ assert.ok(fixture.pending.promise instanceof Promise, 'nested source is still held');
+ fixture.pending.resolve(42);
+ assert.equal(await fixture.pending.promise, 42);
+ },
+ { expose_gc: true, data: mode }
+ );
+ }
+);
diff --git a/test/stream.test.js b/test/stream.test.js
new file mode 100644
index 00000000..0ae0fa93
--- /dev/null
+++ b/test/stream.test.js
@@ -0,0 +1,611 @@
+import vm from 'node:vm';
+import { expect, test, vi } from 'vitest';
+import { unevalStream } from '../src/index.js';
+import { client, expect_roundtrip, seeded_random, shuffle } from './helpers/stream.js';
+
+test.each([
+ {
+ name: 'cycle',
+ make: () => {
+ const a = {};
+ a.self = a;
+ return a;
+ }
+ },
+ {
+ name: 'head → outcome identity',
+ make: ({ later }) => {
+ const s = {};
+ return { s, p: later(s) };
+ }
+ },
+ {
+ name: 'outcome → outcome identity',
+ make: ({ later }) => {
+ const s = {};
+ return [later(s), later({ s })];
+ }
+ },
+ { name: 'nested promises', make: ({ later }) => ({ p: later({ q: later({ r: later(1) }) }) }) },
+ {
+ name: 'cycle through outcome',
+ make: ({ later }) => {
+ const root = {};
+ root.p = later({ root });
+ return root;
+ }
+ },
+ {
+ name: 'rejection identity',
+ make: ({ later_reject }) => {
+ const e = { code: 1 };
+ return { e, p: later_reject(e) };
+ }
+ },
+ {
+ name: 'map key identity',
+ make: ({ later }) => {
+ const k = {};
+ return { m: new Map([[k, 1]]), p: later(k) };
+ }
+ },
+ {
+ name: 'set member identity',
+ make: ({ later }) => {
+ const k = {};
+ return { s: new Set([k]), p: later(k) };
+ }
+ },
+ {
+ name: 'typed view buffer identity',
+ make: ({ later }) => {
+ const b = new ArrayBuffer(4);
+ return { v: new Uint8Array(b), p: later(b) };
+ }
+ },
+ {
+ name: 'sparse + null-proto',
+ make: ({ later }) => {
+ const a = Array(5);
+ a[3] = 'x';
+ return { p: later(Object.assign(Object.create(null), { a })) };
+ }
+ },
+ { name: 'promise inside map value', make: ({ later }) => ({ m: new Map([['k', later(2)]]) }) },
+ {
+ name: 'same promise twice',
+ make: ({ later }) => {
+ const p = later(1);
+ return [p, p, { p }];
+ }
+ }
+])('$name', ({ make }) => expect_roundtrip(make));
+
+class Box {
+ /** @param {unknown} value */
+ constructor(value) {
+ this.value = value;
+ }
+}
+
+class Job {
+ /** @param {Promise} completion */
+ constructor(completion) {
+ this.completion = completion;
+ }
+}
+
+class Cyclic {
+ /** @param {unknown} [meta] */
+ constructor(meta) {
+ this.meta = meta;
+ }
+}
+
+class MyThenable {
+ /** @param {Promise} promise */
+ constructor(promise) {
+ this.promise = promise;
+ }
+ /** @type {Promise['then']} */
+ then(resolve, reject) {
+ return this.promise.then(resolve, reject);
+ }
+}
+
+/** @type {import('../src/types.js').UnevalReplacer} */
+const replacer = (v, js) => {
+ if (v instanceof Box) return js`new Box(${v.value})`;
+ if (v instanceof Job) return js`new Job(${v.completion})`;
+ if (v instanceof Cyclic) return js`new Cyclic(${v.meta})`;
+ if (v instanceof MyThenable) return js`new MyThenable(${Promise.resolve(v)})`;
+};
+
+test.each([
+ {
+ name: 'replaced object in head',
+ make: ({ later }) => {
+ const b = new Box({ x: 1 });
+ return { b, p: later(b) };
+ }
+ },
+ { name: 'replaced object in outcome', make: ({ later }) => ({ p: later(new Box({ x: 1 })) }) },
+ { name: 'replaced object wrapping a promise', make: ({ later }) => [new Job(later({ ok: 1 }))] },
+ {
+ name: 'replaced object whose hole is a head object',
+ make: ({ later }) => {
+ const s = {};
+ return { s, p: later(new Box(s)) };
+ }
+ },
+ {
+ name: 'cycle through a replaced object',
+ make: ({ later }) => {
+ const c = new Cyclic();
+ c.meta = { parent: c };
+ return { c, p: later(c.meta), q: later(new Cyclic([1])) };
+ }
+ },
+ {
+ name: 'custom thenable rebuilt on the client',
+ make: ({ later }) => {
+ const t = new MyThenable(later({ y: 2 }));
+ return { t, again: t, p: later(t) };
+ }
+ }
+])('replacer: $name', ({ make }) =>
+ expect_roundtrip(make, { replacer, globals: { Box, Job, Cyclic, MyThenable } })
+);
+
+class MatrixWrapper {
+ /** @param {string} kind @param {unknown} value */
+ constructor(kind, value) {
+ this.kind = kind;
+ this.value = value;
+ }
+}
+
+/**
+ * @param {number} seed
+ * @param {(value: unknown) => Promise} later
+ */
+function build_matrix_case(seed, later) {
+ const random = seeded_random(seed);
+ const nodes = Array.from({ length: 8 }, (_, index) => ({
+ index,
+ label: `node-${seed}-${index}`,
+ /** @type {unknown} */
+ next: null
+ }));
+ for (let i = 0; i < nodes.length; i += 1) {
+ nodes[i].next = nodes[(i + 1 + Math.floor(random() * 3)) % nodes.length];
+ }
+
+ const shared = nodes[Math.floor(random() * nodes.length)];
+ const distinct = { index: shared.index, label: shared.label, next: shared.next };
+ const null_object = Object.assign(Object.create(null), { kind: 'null', shared, distinct });
+ const sparse = Array(7);
+ sparse[1] = shared;
+ sparse[4] = null_object;
+ const buffer = new ArrayBuffer(12);
+ const bytes = new Uint8Array(buffer);
+ for (let i = 0; i < bytes.length; i += 1) bytes[i] = (seed * 17 + i * 29) & 255;
+ const view = new Uint16Array(buffer, 2, 4);
+ const wrapped = new MatrixWrapper(`wrapper-${seed}`, shared);
+ const map = new Map(
+ shuffle(
+ [
+ [shared, view],
+ [null_object, sparse],
+ [distinct, buffer]
+ ],
+ random
+ )
+ );
+ const set = new Set(shuffle([view, shared, null_object, distinct], random));
+
+ const outcome_a = {
+ kind: 'outcome-a',
+ shared,
+ wrapped,
+ map: new Map([[shared, nodes[(seed + 3) % nodes.length]]])
+ };
+ const outcome_b = Object.assign(Object.create(null), {
+ kind: 'outcome-b',
+ items: [outcome_a, shared, view],
+ set: new Set([distinct, outcome_a, buffer]),
+ inner: later(new MatrixWrapper('late', nodes[seed % nodes.length]))
+ });
+ const entries = shuffle(
+ [
+ ['seed', seed],
+ ['numeric', { 0: '0', 12: '12', '001': '001' }],
+ ['nodes', nodes],
+ ['shared', shared],
+ ['distinct', distinct],
+ ['null_object', null_object],
+ ['sparse', sparse],
+ ['map', map],
+ ['set', set],
+ ['buffer', buffer],
+ ['view', view],
+ ['wrapped', wrapped],
+ ['promises', [outcome_a, outcome_b, outcome_a].map((o) => later(o))]
+ ],
+ random
+ );
+ return Object.fromEntries(entries);
+}
+
+test.each(Array.from({ length: 20 }, (_, i) => ({ seed: i + 1 })))(
+ 'random graph (seed $seed)',
+ ({ seed }) =>
+ expect_roundtrip(({ later }) => build_matrix_case(seed, later), {
+ replacer: (v, js) =>
+ v instanceof MatrixWrapper ? js`new MatrixWrapper(${v.kind},${v.value})` : undefined,
+ globals: { MatrixWrapper }
+ })
+);
+
+test('recognizes only branded native promises across realms and subclasses', async () => {
+ class SubPromise extends Promise {}
+ await expect_roundtrip(() => [
+ Promise.resolve(1),
+ vm.runInNewContext('Promise.resolve(2)'),
+ SubPromise.resolve(3)
+ ]);
+
+ expect(() => unevalStream({ [Symbol.toStringTag]: 'Promise', then() {} })).toThrow(
+ /Cannot stringify/
+ );
+});
+
+test('calls the replacer once per identity, even after a failed outcome', async () => {
+ class X {
+ constructor() {
+ this.value = 42;
+ }
+ }
+ const x = new X();
+ /** @type {unknown[]} */
+ const seen = [];
+ const transformError = vi.fn(() => undefined);
+
+ const a = Promise.withResolvers();
+ const b = Promise.withResolvers();
+ const stream = unevalStream(
+ { p1: a.promise, p2: b.promise },
+ (value, js) => {
+ seen.push(value);
+ if (value instanceof X) return js`{value:${value.value}}`;
+ },
+ { transformError }
+ );
+
+ const c = client();
+ const root = c.head(stream.head);
+
+ a.resolve({ x, fn: () => {} });
+ c.block(/** @type {string} */ ((await stream.tail.next()).value));
+ b.resolve(x);
+ c.block(/** @type {string} */ ((await stream.tail.next()).value));
+
+ expect(seen.filter((v) => v === x)).toHaveLength(1);
+ await expect(root.p1).rejects.toThrow('devalue: failed to serialize asynchronous value');
+ expect(await root.p2).toEqual({ value: 42 });
+ expect(transformError).toHaveBeenCalledOnce();
+ expect((await stream.tail.next()).done).toBe(true);
+});
+
+test('head is deterministic regardless of settlement timing', () => {
+ const settled = unevalStream({ a: [1], p: Promise.resolve(1) }, undefined, { id: 'x' });
+ const unsettled = unevalStream({ a: [1], p: new Promise(() => {}) }, undefined, { id: 'x' });
+
+ expect(settled.head).toBe(unsettled.head);
+});
+
+test('folds settled promises via the replacer', async () => {
+ const unhandled = vi.fn();
+ process.on('unhandledRejection', unhandled);
+
+ try {
+ const shared = { x: 1 };
+ const ok = Promise.resolve(shared);
+ const bad = Promise.reject(shared);
+ bad.catch(() => {});
+
+ /** @type {Map} */
+ const settled = new Map([
+ [ok, { ok: true, value: shared }],
+ [bad, { ok: false, reason: shared }]
+ ]);
+ /**
+ * This is just a simple helper to create a rejected promise that doesn't throw an unhandled rejection
+ * @type {(reason: unknown, js: import('../src/types.js').JavaScriptTag) => any}
+ */
+ const rejected = (reason, js) => js`(p => (p.catch(() => {}), p))(Promise.reject(${reason}))`;
+
+ const { root, blocks } = await expect_roundtrip(() => ({ shared, ok, bad }), {
+ replacer: (v, js) => {
+ const r = settled.get(v);
+ if (r) return r.ok ? js`Promise.resolve(${r.value})` : rejected(r.reason, js);
+ },
+ orders: ['fifo']
+ });
+
+ expect(blocks).toHaveLength(0);
+ await new Promise((r) => setTimeout(r, 10));
+ expect(unhandled).not.toHaveBeenCalled();
+ const revived = /** @type {any} */ (root);
+ expect(await revived.ok).toBe(revived.shared);
+ } finally {
+ process.off('unhandledRejection', unhandled);
+ }
+});
+
+test('emits a plain expression when there are no promises', async () => {
+ const stream = unevalStream({ x: 1 });
+ const c = client();
+
+ expect(stream.head).not.toContain('__d');
+ expect(c.head(stream.head)).toEqual({ x: 1 });
+ expect((await stream.tail.next()).done).toBe(true);
+ expect(c.context.__d).toBeUndefined();
+});
+
+test('return() completes a pending next()', async () => {
+ const stream = unevalStream({ p: new Promise(() => {}) });
+ const next = stream.tail.next();
+
+ expect(await stream.tail.return()).toEqual({ done: true, value: undefined });
+ expect(await next).toEqual({ done: true, value: undefined });
+});
+
+test('abort emits a final block that rejects pending client promises', async () => {
+ const controller = new AbortController();
+ const settled = Promise.withResolvers();
+ const stream = unevalStream(
+ { a: settled.promise, b: new Promise(() => {}), c: new Promise(() => {}) },
+ undefined,
+ { signal: controller.signal }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+
+ settled.resolve(1);
+ const first = await stream.tail.next();
+ c.block(/** @type {string} */ (first.value));
+ expect(await root.a).toBe(1);
+
+ const next = stream.tail.next();
+ const later = stream.tail.next();
+ controller.abort(new Error('server-only reason'));
+
+ const last = await next;
+ expect(last.done).toBe(false);
+ c.block(/** @type {string} */ (last.value));
+ for (const p of [root.b, root.c]) {
+ await expect(p).rejects.toThrow('devalue: stream aborted');
+ }
+ expect(last.value).not.toContain('server-only reason');
+ expect(Object.keys(c.context.__d)).toEqual([]);
+ expect(await later).toEqual({ done: true, value: undefined });
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+});
+
+test('abort with no pending next() emits the final block on the next read', async () => {
+ const controller = new AbortController();
+ const queued = Promise.withResolvers();
+ const stream = unevalStream({ q: queued.promise, p: new Promise(() => {}) }, undefined, {
+ signal: controller.signal
+ });
+ const c = client();
+ const root = c.head(stream.head);
+
+ // An outcome queued but not yet emitted is rejected along with the rest.
+ queued.resolve(1);
+ await new Promise((r) => setTimeout(r, 0));
+ controller.abort();
+
+ const last = await stream.tail.next();
+ c.block(/** @type {string} */ (last.value));
+ await expect(root.q).rejects.toThrow('devalue: stream aborted');
+ await expect(root.p).rejects.toThrow('devalue: stream aborted');
+ expect(Object.keys(c.context.__d)).toEqual([]);
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+});
+
+test('abort after completion or return() emits nothing', async () => {
+ for (const mode of ['complete', 'return']) {
+ const controller = new AbortController();
+ const stream = unevalStream({ p: Promise.resolve(1) }, undefined, {
+ signal: controller.signal
+ });
+ if (mode === 'complete') expect((await stream.tail.next()).done).toBe(false);
+ else await stream.tail.return();
+ controller.abort();
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+ }
+});
+
+test('abort during a tail block rejects the promises that block creates', async () => {
+ const controller = new AbortController();
+ const source = Promise.withResolvers();
+ const graph = { nested: new Promise(() => {}) };
+ const stream = unevalStream(
+ source.promise,
+ (thing) => {
+ if (thing === graph) controller.abort();
+ },
+ { signal: controller.signal }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+
+ source.resolve(graph);
+ const block = await stream.tail.next();
+ c.block(/** @type {string} */ (block.value));
+
+ const revived = await root;
+ await expect(revived.nested).rejects.toThrow('devalue: stream aborted');
+ expect(Object.keys(c.context.__d)).toEqual([]);
+ expect(await stream.tail.next()).toEqual({ done: true, value: undefined });
+});
+
+test.each([{ promises: true }, { promises: false }])(
+ 'abort during head emission throws the exact reason (promises: $promises)',
+ ({ promises }) => {
+ const controller = new AbortController();
+ const reason = new Error('abort during head');
+ const value = {
+ ...(promises ? { p: new Promise(() => {}) } : {}),
+ get tripwire() {
+ controller.abort(reason);
+ return 1;
+ }
+ };
+
+ let result;
+ let thrown;
+ try {
+ result = unevalStream(value, undefined, { signal: controller.signal });
+ } catch (error) {
+ thrown = error;
+ }
+
+ expect(thrown).toBe(reason);
+ expect(result).toBeUndefined();
+ }
+);
+
+test('concurrent next() calls settle in order', async () => {
+ const a = Promise.withResolvers();
+ const b = Promise.withResolvers();
+ const stream = unevalStream({ a: a.promise, b: b.promise });
+ const c = client();
+ const root = c.head(stream.head);
+
+ /** @type {string[]} */
+ const order = [];
+ const first = stream.tail.next().then((r) => (order.push('first'), r));
+ const second = stream.tail.next().then((r) => (order.push('second'), r));
+ const third = stream.tail.next().then((r) => (order.push('third'), r));
+
+ b.resolve('b');
+ await new Promise((r) => setTimeout(r, 0));
+ a.resolve('a');
+
+ const results = await Promise.all([first, second, third]);
+ expect(order).toEqual(['first', 'second', 'third']);
+ expect(results.map((r) => r.done)).toEqual([false, false, true]);
+ for (const r of results.slice(0, 2)) c.block(/** @type {string} */ (r.value));
+ expect(await root.a).toBe('a');
+ expect(await root.b).toBe('b');
+});
+
+test('rejection reasons never reach the client by default', async () => {
+ const stream = unevalStream({
+ error: Promise.reject(new Error('db password=hunter2')),
+ object: Promise.reject({ secret: 'internal detail' }),
+ string: Promise.reject('stack trace')
+ });
+ let source = stream.head;
+ const c = client();
+ const root = c.head(stream.head);
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+ for (const key of ['error', 'object', 'string']) {
+ await expect(root[key]).rejects.toThrow('devalue: failed to serialize asynchronous value');
+ }
+ expect(source).not.toMatch(/hunter2|secret|internal|stack/);
+});
+
+test.each(['sync', 'async'])('transformError (%s) shapes rejections and failures', async (mode) => {
+ const shared = { shared: true };
+ /** @param {unknown} error */
+ const shape = (error) => ({
+ message: error instanceof Error ? error.message : String(error),
+ shared
+ });
+ const stream = unevalStream(
+ {
+ shared,
+ rejected: Promise.reject(new Error('nope')),
+ unserializable: Promise.resolve({ fn() {} }),
+ resolved: Promise.resolve(1)
+ },
+ undefined,
+ {
+ transformError: (error) => (mode === 'sync' ? shape(error) : Promise.resolve(shape(error)))
+ }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+ for await (const block of stream.tail) c.block(block);
+
+ await expect(root.rejected).rejects.toEqual({ message: 'nope', shared: root.shared });
+ const reason = await root.unserializable.catch((/** @type {any} */ e) => e);
+ expect(reason.message).toMatch(/Cannot stringify/);
+ expect(reason.shared).toBe(root.shared);
+ expect(await root.resolved).toBe(1);
+ expect(c.context.__d).toEqual({});
+});
+
+test.each([
+ [
+ 'throws',
+ () => {
+ throw new Error('transform failed');
+ }
+ ],
+ ['rejects', () => Promise.reject(new Error('transform failed'))],
+ ['returns something unserializable', () => ({ fn() {} })],
+ ['resolves to something unserializable', () => Promise.resolve(Symbol())],
+ ['returns undefined', () => undefined],
+ ['returns null', () => null],
+ ['returns false', () => false],
+ ['resolves to null', () => Promise.resolve(null)]
+])('falls back to the generic error when transformError %s', async (_, transformError) => {
+ const stream = unevalStream({ p: Promise.reject(new Error('secret')) }, undefined, {
+ transformError
+ });
+ const c = client();
+ const root = c.head(stream.head);
+ let source = '';
+ for await (const block of stream.tail) {
+ source += block;
+ c.block(block);
+ }
+ await expect(root.p).rejects.toThrow('devalue: failed to serialize asynchronous value');
+ expect(source).not.toContain('secret');
+ expect(c.context.__d).toEqual({});
+});
+
+test('a pending async transformError does not hold back other settlements', async () => {
+ const transform = Promise.withResolvers();
+ const stream = unevalStream(
+ { failed: Promise.reject(new Error('x')), ok: Promise.resolve(1) },
+ undefined,
+ { transformError: () => transform.promise }
+ );
+ const c = client();
+ const root = c.head(stream.head);
+ c.block(/** @type {string} */ ((await stream.tail.next()).value));
+ expect(await root.ok).toBe(1);
+ transform.resolve('shaped');
+ c.block(/** @type {string} */ ((await stream.tail.next()).value));
+ await expect(root.failed).rejects.toBe('shaped');
+ expect((await stream.tail.next()).done).toBe(true);
+});
+
+test('return() during an async transformError ends the stream', async () => {
+ const stream = unevalStream({ p: Promise.reject(new Error('x')) }, undefined, {
+ transformError: () => new Promise(() => {})
+ });
+ const next = stream.tail.next();
+ await new Promise((resolve) => setTimeout(resolve, 0));
+ await stream.tail.return();
+ expect(await next).toEqual({ done: true, value: undefined });
+});