From b3bf239f488adc321d9537e652fd08d55d359944 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sat, 22 Aug 2026 19:56:47 -0400 Subject: [PATCH] streams: direct-access byte edges for host stream ends (A21, #128) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wasmtime DirectSource/DirectDestination-shaped. StreamWriter.writeDirect / Stream.readDirect (and the low-level HostWritableEnd/HostReadableEnd forms) park a direct session: at every rendezvous with a peer operation of nonzero capacity the callback runs exactly once, synchronously, inside the rendezvous, against a scoped view of the peer's bytes — guest linear memory when the peer is a guest, so an external byte mover's own set() IS the single canonical-ABI copy. "more"/"done" is wasmtime's poll cadence spelled event-style; "done" with zero marked retracts (the speculative-park correction), and marks acknowledge on clean return only, so no zero-progress COMPLETED copy is ever emitted. The rendezvous half is a seam in task/streams.ts that collapses to the reference's dst.write(src.read(n)) verbatim whenever neither side is direct; the no-copy outcomes route to branch shapes definitions.py already produces. The session half (scope, mark accounting, verdict cadence, promise) lives in exec/host_streams.ts. Host-to-host stays at the one-copy floor against the chunk forms (scratch-becomes-chunk / view-of-offered-chunk); two direct sessions refuse loudly. Zero-length probes complete without invoking the callback — the armed session is the readiness claim. Contract: embedder-api amendment A21; architecture §7 gains the scoped exception to "views into guest memory are never exposed". --- contracts/embedder-api.md | 104 ++- docs/architecture.md | 7 +- runtime/src/embedder/mod.ts | 6 + runtime/src/embedder/streams.ts | 106 +++ runtime/src/exec/host_streams.ts | 528 +++++++++++ runtime/src/task/streams.ts | 257 +++++- runtime/tests/direct_streams_test.ts | 819 ++++++++++++++++++ runtime/tests/embedder/direct_streams_test.ts | 241 ++++++ 8 files changed, 2064 insertions(+), 4 deletions(-) create mode 100644 runtime/tests/direct_streams_test.ts create mode 100644 runtime/tests/embedder/direct_streams_test.ts diff --git a/contracts/embedder-api.md b/contracts/embedder-api.md index 4dc8ed8..6f353fc 100644 --- a/contracts/embedder-api.md +++ b/contracts/embedder-api.md @@ -505,12 +505,28 @@ interface Stream { readable(): ReadableStream>; // web-native; Chunk = Uint8Array, else T[] [Symbol.asyncIterator](): AsyncIterator>; read(max: number): Promise>; // low-level; empty chunk = end + readDirect( // stream only — amendment A21 + consume: (src: DirectSource) => "more" | "done", + ): Promise; cancelRead(): void; drop(): void; // [Symbol.dispose] alias } interface Future extends PromiseLike { // await it directly drop(): void; cancel(): void; } +// Direct-access byte edges (amendment A21, stream only). The writer-side +// mirror lives on StreamWriter: +// writeDirect(produce: (dest: DirectDestination) => "more" | "done"): Promise +// Both objects are DEAD once the callback returns — every later method call +// throws. +interface DirectDestination { + remaining(): Uint8Array; // scoped view over the reader's unfilled landing zone + markWritten(n: number): void; // cumulative within the invocation +} +interface DirectSource { + remaining(): Uint8Array; // scoped view of the writer's unread bytes; read-only by contract + markRead(n: number): void; +} class ErrorContext { readonly message: string } // lift-only constructor-wise (C2 amendment); lowering also accepts any branded string-`message` carrier by minting a fresh local context (A20) class DroppedError extends Error { … } // awaiting a dropped future rejects with this ``` @@ -660,7 +676,7 @@ class DroppedError extends Error { … } // awaiting a dropped future rejects - Writer-side host ends (`hostStream()`-era API) remain the low-level seam underneath; the conventions layer exposes them as `Stream.create(): { stream: Stream, writer: StreamWriter }` - with `write`/`writeAll`/`cancelWrite`/`close`. + with `write`/`writeAll`/`writeDirect`/`cancelWrite`/`close`. - **Component faults are loud on stream/future operations** (amendment A7). When the component instance holding the peer end traps, its live ends are retired: a parked host `read`/`write`/`writeAll`/future-await @@ -703,6 +719,92 @@ class DroppedError extends Error { … } // awaiting a dropped future rejects iterator present as clean EOS. The canceller is the same code observing the end, so no discriminated signal is warranted; pinned by test. (A *peer* fault is never presented this way — that is A7's rule.) +- **Direct-access byte edges** (amendment A21, 2026-08-22, polyengine#128 — + wasmtime `DirectSource`/`DirectDestination`-shaped, `component::concurrent` + 47.0.3). For `stream` only, both host ends gain a form whose last hop + *is* the single canonical-ABI copy, so external buffer movers (websocket + frames, SAB-ring segments, transferred `ArrayBuffer`s) never pay a second + copy inside the runtime: + - `StreamWriter.writeDirect(produce)` and `Stream.readDirect(consume)` + (with the same methods on the low-level `HostWritableEnd`/ + `HostReadableEnd` seam). Each parks a **direct session**: at every + rendezvous with a peer operation of nonzero capacity, the callback runs + **exactly once, synchronously, inside the rendezvous** — the guest's + copy trampoline, or the host call that arrived second. `produce` + receives a `DirectDestination` whose `remaining()` is the reader's + unfilled landing zone; `consume` receives a `DirectSource` whose + `remaining()` is the writer's unread bytes. When the peer is a guest, + that view aliases **guest linear memory**: the embedder's own + `set()`/`subarray` copy is the ABI copy. The callback's verdict is + wasmtime's poll cadence spelled event-style: `"more"` keeps the session + parked for the next rendezvous; `"done"` ends it, resolving the promise + with the session's total byte count. + - **Scope is the validity window.** The `DirectDestination`/`DirectSource` + object dies when the callback returns; every later method call throws a + `TypeError` naming the scoping rule. Views are re-derived per + `remaining()` call (a `memory.grow` between rendezvous never yields a + stale view), and retaining one past the callback is misuse. Inside the + callback, calls that can run guest code or operate this stream are + forbidden (reentrancy); the one-in-flight-per-end rule (A7) covers the + stream's own operations, and `writeDirect`/`readDirect` participate in + it exactly as `write`/`read` do. + - **Marks acknowledge on clean return only.** `markWritten`/`markRead` + accumulate within the invocation (over-marking throws). A callback that + returns having marked ≥ 1 byte completes the peer's copy with that + count. Returning `"done"` with **zero** marked is *retraction*: the + session ends (promise resolves with its running total), the peer's + operation stays parked, and no event is delivered — the speculative-park + pattern (demand arrived while the producer's ring happened to be empty; + re-arm when it fills). Zero marked with `"more"` is misuse: the session + rejects with a `TypeError`. A callback that **throws** rejects the + session with that error, and the invocation's marks are discarded — + bytes physically written past the acknowledged progress are + unobservable to the peer. In every outcome the peer's parked operation + survives and the stream stays alive (the host still holds its end and + may fall back to chunk forms); a runtime never emits a zero-progress + COMPLETED copy, which is unreachable in definitions.py for a + nonzero-capacity operation and which a guest may lawfully misread as + end-of-stream. + - **Zero-length-read readiness position** (Concurrency.md "Stream + Readiness"): a parked direct session answers a zero-length probe with + immediate COMPLETED — the armed session is the readiness claim — and + the callback is **not** invoked. A producer that parks speculatively + while knowingly empty is stretching that claim; the retraction path + above is its correction. + - **Host↔host at the same floor.** A direct session rendezvousing with a + peer *chunk* end still costs one copy: `produce` against a host + `read(max)` writes into a fresh scratch that becomes the delivered + chunk (ownership passes with it); `consume` against a parked chunk + `write` gets a scoped view of the offered chunk itself (the A5 borrow, + scoped to the callback). Two direct sessions cannot rendezvous with + each other — neither side owns memory — so the arriving side throws a + `TypeError`: at least one side of a host↔host rendezvous uses chunk + forms. + - **Interplay with the existing rules, all inherited:** a peer trap + rejects the session with `PeerTrappedError` carrying the delivered byte + count, while a session the callback already completed keeps its result + (A7 precision); reader/writer drop resolves the session with its total + (the `write`/`writeAll` convention — a resolution the producer's own + `"done"` did not cause is the reader-gone signal); `cancelWrite`/ + `cancelRead` retract a parked session (A8's indistinguishability + caveats unchanged); the A15 transfer guard applies to `readDirect` as + to `read`; a parked session is retention, so the deadlock-verdict arm + stays live. `writeDirect` on an unbound `Stream.create()` writer parks + until the lowering site binds the element type, then requires u8; + `readDirect` on an unbound or non-u8 stream throws, as `read`'s + refusals do. + - **What is deliberately absent:** no ownership-transfer variant of the + chunk forms — `write`/`writeAll`'s borrowed-until-settled contract (A5) + already meets the one-copy floor, and `HostBuffer`'s `taken()` already + passes a sole chunk through unsliced; no `list` intake/output form + (same question, tracked separately); no conduit, credit, or realm + machinery (the #128 scope ruling: deltic provides the byte edge, not + the mover). SAB-backed `Uint8Array`s are legal on the embedder's side + of every copy in both directions — the embedder performs the copy, so + nothing here can reject them. The #97 `HostBuffer` length bound applies + to the buffered path only: a direct session's capacity IS the peer's + actual buffer size, already bounded by the guest's own `MAX_LENGTH` + trap. ## Module wiring and instantiation diff --git a/docs/architecture.md b/docs/architecture.md index 9000d10..d76df07 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -422,7 +422,12 @@ decide deliberately and document here. tests forced it immediately; wit-bindgen guests themselves use utf8). - **Numbers.** `u64`/`s64` ↔ `BigInt`; everything else ↔ `number`. `list` ↔ `Uint8Array` (copy; views into guest memory are never - exposed). Both directions are bulk copies: lift via a `Uint8Array` slice, + exposed — with one deliberate, scoped exception: the `stream` + direct-access sessions of embedder-api amendment A21 hand the callback a + view over the peer guest's landing zone or unread bytes, valid only for + that synchronous callback, so an external byte mover's last hop can BE + the one ABI copy). Both directions are bulk copies: lift via a + `Uint8Array` slice, lower via `Uint8Array.set` (issue #54 — the per-element interpreted store cost ~45 ns/byte and capped host→guest byte traffic at ~22 MB/s). Stream payload copies share these paths, and u8 stream chunks stay `Uint8Array` diff --git a/runtime/src/embedder/mod.ts b/runtime/src/embedder/mod.ts index 0b9a9eb..7977f70 100644 --- a/runtime/src/embedder/mod.ts +++ b/runtime/src/embedder/mod.ts @@ -86,6 +86,12 @@ export { export { type Chunk, + // Direct-access byte edges (amendment A21, polyengine#128): the two scoped + // callback objects `StreamWriter.writeDirect` / `Stream.readDirect` hand + // out, plus their verdict type. + type DirectDestination, + type DirectSource, + type DirectVerdict, type ElemCodec, ErrorContext, Future, diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts index 6d11992..5d1b37e 100644 --- a/runtime/src/embedder/streams.ts +++ b/runtime/src/embedder/streams.ts @@ -12,6 +12,10 @@ import type { ValType } from "../cabi/types.ts"; import { despecialize } from "../cabi/types.ts"; import type { ComponentValue } from "../cabi/types.ts"; import { + type DirectDestination, + type DirectSessionInfo, + type DirectSource, + type DirectVerdict, type HostFuture, hostFuture, hostFutureFor, @@ -146,6 +150,27 @@ export function isU8Element(element: ValType | null): boolean { return element !== null && despecialize(element).kind === "u8"; } +// Re-exported so embedders reach the A21 callback shapes from this layer too +// (contracts/embedder-api.md §"Streams and futures", amendment A21, #128). +export type { + DirectDestination, + DirectSource, + DirectVerdict, +} from "../exec/host_streams.ts"; + +/** + * A21 (#128): the direct-access byte edges are `stream` only. A + * zero-width element type (`t === null`) is not u8 either. + */ +function requireU8Direct(codec: ElemCodec | null, who: string): void { + if (codec === null || !isU8Element(codec.element)) { + throw new TypeError( + `${who} is available on stream only (embedder-api amendment A21, ` + + `polyengine#128); use write()/read() for other element types`, + ); + } +} + /** * A stream handle. * @@ -283,6 +308,47 @@ export class Stream { return this.#chunk(raw); } + /** + * Consume the writer's bytes in place, without an intermediate chunk + * (`stream` only — contracts/embedder-api.md amendment A21, + * polyengine#128). + * + * At every rendezvous with a writer of nonzero capacity, `consume` runs + * exactly once, synchronously, with a `DirectSource` over the writer's + * unread bytes — guest linear memory when the peer is a guest, so the + * consumer's own `set()`/`subarray` copy IS the canonical-ABI copy. + * `"more"` keeps the session parked for the next rendezvous; `"done"` ends + * it. Resolves with the session's total byte count. Marking a prefix is + * normal: the writer re-offers the rest on its own schedule. + * + * `"done"` with zero bytes marked *retracts*: the session ends and the + * writer's operation stays parked, with no event delivered. `"more"` with + * zero marked, and a throwing callback, reject — and in both cases the + * writer's parked operation survives and the stream stays alive. + * + * Refusals mirror `read`: an unbound `Stream.create()` handle and a handle + * already passed to a guest (the A15 transfer guard) both throw, as does a + * non-`u8` element type. + */ + async readDirect( + consume: (src: DirectSource) => DirectVerdict, + ): Promise { + const host = this.#require(); + const where = this.#codec?.where ?? "stream read"; + throwIfFailed(host.value, where); + requireU8Direct(this.#codec, "readDirect"); + const info: DirectSessionInfo = { endedByVerdict: false }; + const n = await host.readable.readDirect(consume, info); + // A7 precision, `read`'s rule adapted: a session the CONSUMER itself + // ended with `"done"` genuinely completed and keeps its resolution. Any + // other way out (the writer dropped, the session was cancelled, the + // retirement walk settled us) is a settle-path this consumer did not + // cause — so if the peer's instance trapped, reject with the delivered + // count rather than fake a clean end. + if (!info.endedByVerdict) throwIfPeerTrapped(host.value, where, n); + return n; + } + #chunk(raw: ComponentValue[] | Uint8Array): Chunk { const codec = this.#codec!; if (isU8Element(codec.element)) { @@ -416,6 +482,46 @@ export class StreamWriter { return n; } + /** + * Fill the reader's landing zone in place, without an intermediate chunk + * (`stream` only — contracts/embedder-api.md amendment A21, + * polyengine#128). + * + * At every rendezvous with a reader of nonzero capacity, `produce` runs + * exactly once, synchronously, with a `DirectDestination` over the reader's + * unfilled landing zone — guest linear memory when the peer is a guest, so + * the producer's own `set()` IS the canonical-ABI copy and an external byte + * mover (a websocket frame, a SAB ring segment, a transferred + * `ArrayBuffer`) never pays a second copy inside the runtime. `"more"` + * keeps the session parked for the next rendezvous; `"done"` ends it. + * Resolves with the session's total byte count. + * + * `"done"` with zero bytes marked *retracts* (the session ends, the + * reader's operation stays parked, no event — the speculative-park + * correction); `"more"` with zero marked, and a throwing callback, reject. + * + * Parks until the element type is known, exactly as `write` does — a + * `Stream.create()` writer has no element type until the lowering site + * binds one — and then requires `u8`. + */ + async writeDirect( + produce: (dest: DirectDestination) => DirectVerdict, + ): Promise { + await this.#stream.whenBound(); + const host = hostOf(this.#stream); + const where = this.#stream.codec?.where ?? "stream write"; + throwIfFailed(host.value, where); + requireU8Direct(this.#stream.codec, "writeDirect"); + const info: DirectSessionInfo = { endedByVerdict: false }; + const n = await host.writable.writeDirect(produce, info); + // A7 precision, `write`'s short-take rule adapted: a session the PRODUCER + // itself ended with `"done"` keeps its resolution; every other way out is + // a settle-path the producer did not cause, so a trapped peer rejects + // here carrying the delivered count. + if (!info.endedByVerdict) throwIfPeerTrapped(host.value, where, n); + return n; + } + /** Offer values until all are taken or the reader goes away. */ async writeAll(values: Chunk): Promise { await this.#stream.whenBound(); diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index eac9dfd..2df4c91 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -79,8 +79,11 @@ import { import { abandonSharedFuture, BUFFER_MAX_LENGTH, + type ByteWindow, type ComponentInstanceState, CopyResult, + type DirectBuffer, + type DirectOutcome, markHostActivityArm, type PayloadChunk, sameElemType, @@ -210,6 +213,68 @@ export class HostBuffer { } return out; } + + // --- A21 `ByteWindow` (embedder-api amendment A21, polyengine#128) --- + // + // A host buffer can be the PEER of a direct session on the other end of a + // host↔host rendezvous. Which of the two shapes it takes follows from the + // direction it was built for, exactly as `read`/`write` above do: + // + // * SOURCE (`values !== null`, a parked `write`): the window is a view of + // the offered chunk itself — the A5 borrow, scoped to the callback. No + // extra copy at all. + // * DESTINATION (`values === null`, a parked/arriving `read(max)`): there + // is no landing zone to view, so the window is a fresh scratch; the + // marked prefix becomes the delivered chunk (ownership passes with it, + // and `taken()` hands a sole chunk through unsliced). + + /** The synthesized destination window, live for one direct invocation. */ + #scratch: Uint8Array | null = null; + + byteView(n: number): Uint8Array { + assert_(n <= this.remain(), "host direct window beyond remaining"); + if (this.values === null) { + // Stable for the whole invocation: `remaining()` re-derives on every + // call and the producer's earlier `set()`s must survive that. + if (this.#scratch === null || this.#scratch.length !== n) { + this.#scratch = new Uint8Array(n); + } + return this.#scratch; + } + assert_( + this.values instanceof Uint8Array, + "host direct window on a non-u8 chunk", + ); + return (this.values as Uint8Array).subarray( + this.progress, + this.progress + n, + ); + } + + advanceBytes(k: number): void { + assert_( + k >= 0 && k <= this.remain(), + "host direct advance beyond remaining", + ); + if (this.values === null) { + // A callback may mark bytes it never actually looked at the window to + // write (nonsense, but the runtime must stay total rather than trip an + // internal assertion). The acknowledged prefix is then whatever the + // synthesized landing zone held — zeroes — which is the faithful + // analogue of the guest-peer case, where it would be whatever the + // reader's memory already contained. + const scratch = this.#scratch ?? new Uint8Array(k); + // Delivered as an owned chunk; `write` is the same call the reference + // copy would have made, so `remain()`/`taken()` stay consistent. + this.write(scratch.subarray(0, k)); + } else { + this.progress += k; + } + } + + endWindow(): void { + this.#scratch = null; + } } /** @@ -447,6 +512,311 @@ class HostActivity { } } +// --------------------------------------------------------------------------- +// Direct-access byte edges (embedder-api amendment A21, 2026-08-22, #128) +// --------------------------------------------------------------------------- +// +// wasmtime `DirectSource`/`DirectDestination`-shaped (`component::concurrent`, +// 47.0.3). For `stream` only, a host end may park a *direct session* +// instead of a chunk: at every rendezvous with a peer operation of nonzero +// capacity the session's callback runs exactly once, synchronously, inside the +// rendezvous, against a scoped view of the peer's bytes — so an external +// buffer mover's own `set()` IS the single canonical-ABI copy. +// +// The rendezvous half lives in task/streams.ts (`rendezvousCopy` and the two +// call sites it collapses to `dst.write(src.read(n))` for every non-direct +// path). This half owns the session: the callback scope, mark accounting, +// the verdict cadence, and the promise. + +/** The scoped landing zone handed to a `writeDirect` producer (A21, #128). */ +export interface DirectDestination { + /** + * The reader's still-unfilled bytes. Re-derived on every call (a + * `memory.grow` between two rendezvous of one session never yields a stale + * view) and shrinking by whatever has been marked so far in THIS + * invocation. DEAD once the callback returns. + */ + remaining(): Uint8Array; + /** + * Acknowledge bytes written into the view. Cumulative within the + * invocation; acknowledged only if the callback then returns cleanly. + */ + markWritten(n: number): void; +} + +/** The scoped view handed to a `readDirect` consumer (A21, #128). */ +export interface DirectSource { + /** + * The writer's unread bytes; read-only by contract. Same scoping and + * re-derivation rules as `DirectDestination.remaining`. + */ + remaining(): Uint8Array; + /** Acknowledge bytes consumed from the view. See `markWritten`. */ + markRead(n: number): void; +} + +/** The callback's poll cadence, spelled event-style (A21). */ +export type DirectVerdict = "more" | "done"; + +/** + * Out-parameter of the low-level direct forms: `true` iff the session ended + * because the callback itself returned `"done"`, rather than because the peer + * dropped / the operation was cancelled / the peer's instance trapped. + * + * The conventions layer needs the distinction for A7 precision — a session + * the producer already completed keeps its resolution even if the peer then + * trapped — and `Promise` is the contract's return shape, so it rides + * here rather than in the resolved value. + */ +export interface DirectSessionInfo { + endedByVerdict: boolean; +} + +/** + * The `DirectDestination`/`DirectSource` object itself. One per INVOCATION, + * not per session: "the object dies when the callback returns" is the + * contract's validity window, and every later method call throws a + * `TypeError` naming the rule. + */ +class DirectScope implements DirectDestination, DirectSource { + marked = 0; + #live = true; + + constructor( + private readonly peer: ByteWindow, + /** The peer's actual remaining capacity — never the parked sentinel. */ + private readonly capacity: number, + ) {} + + remaining(): Uint8Array { + this.#check(); + // Re-derived per call: `byteView` is grow-safe for a guest peer, and the + // `subarray` accounts for the marks made so far in this invocation. + return this.peer.byteView(this.capacity).subarray(this.marked); + } + + markWritten(n: number): void { + this.#mark(n, "markWritten"); + } + + markRead(n: number): void { + this.#mark(n, "markRead"); + } + + #mark(n: number, who: string): void { + this.#check(); + if (!Number.isInteger(n) || n < 0) { + throw new TypeError( + `${who}(${n}): a direct-access mark must be a non-negative integer`, + ); + } + if (this.marked + n > this.capacity) { + throw new TypeError( + `${who}(${n}) would take the invocation's cumulative mark to ` + + `${this.marked + n}, past the ${this.capacity} byte(s) the view ` + + `held on entry (embedder-api amendment A21)`, + ); + } + this.marked += n; + } + + #check(): void { + if (!this.#live) { + throw new TypeError( + "this direct-access view is dead: a DirectDestination/DirectSource " + + "is scoped to the synchronous callback invocation it was passed " + + "to, and retaining one past its return is misuse (embedder-api " + + "amendment A21, polyengine#128)", + ); + } + } + + /** + * End of the invocation: the object is dead, and every later method call + * throws. Releasing the peer's synthesized window is the caller's job + * (`DirectSession.runDirect`), because it must happen strictly after the + * acknowledged marks are applied. + */ + die(): void { + this.#live = false; + } +} + +/** + * A parked direct session, as both halves see it: a `DirectBuffer` to the + * rendezvous (task/streams.ts) and a promise to the embedder. + * + * It presents the ordinary buffer surface so the reference control flow keeps + * working unchanged — `remain()` answers a positive SENTINEL while the session + * is live, which only ever feeds the rendezvous' `min()` and so resolves to + * the peer's real capacity — but `read`/`write` are unreachable: the seam + * routes a direct buffer through `runDirect` instead. + */ +class DirectSession implements DirectBuffer { + readonly direct = true as const; + /** Bytes acknowledged across the whole session. */ + total = 0; + /** The callback said `"done"`, or the session failed / was settled. */ + ended = false; + /** `ended` because the callback said so (A7 precision; see `DirectSessionInfo`). */ + endedByVerdict = false; + /** Installed in the shared object's pending slot right now. */ + pending = false; + /** `cancelWrite`/`cancelRead` arrived; stop at the next loop top. */ + cancelled = false; + + #settle: ((step: "done" | "reissue") => void) | null = null; + #reject: ((e: unknown) => void) | null = null; + + constructor( + readonly t: ValType | null, + private readonly invoke: (scope: DirectScope) => DirectVerdict, + ) {} + + // --- buffer surface (definitions.py `Buffer`) --- + + remain(): number { + // The sentinel is `Buffer.MAX_LENGTH`, the largest value the rendezvous + // can legally see; it never surfaces to the embedder because the scope is + // built from `min(peer.remain(), sentinel)`. + return this.ended ? 0 : BUFFER_MAX_LENGTH; + } + + isZeroLength(): boolean { + return false; + } + + read(_n: number): PayloadChunk { + throw new Error("internal: a direct session must go through the A21 seam"); + } + + write(_vs: PayloadChunk): void { + throw new Error("internal: a direct session must go through the A21 seam"); + } + + // --- the direct protocol --- + + runDirect(peer: ByteWindow, n: number): DirectOutcome { + const scope = new DirectScope(peer, n); + try { + return this.#runDirect(scope, peer); + } finally { + // Release any window the peer SYNTHESIZED (a `HostBuffer` destination's + // scratch). Strictly after `advanceBytes`, which is what turns the + // marked prefix of that scratch into the delivered chunk. + peer.endWindow?.(); + } + } + + #runDirect(scope: DirectScope, peer: ByteWindow): DirectOutcome { + let verdict: DirectVerdict; + try { + verdict = this.invoke(scope); + } catch (e) { + // "A callback that throws rejects the session with that error, and the + // invocation's marks are discarded" — so nothing touches `peer`. + scope.die(); + this.#fail(e); + return "failed"; + } + scope.die(); + if (verdict !== "more" && verdict !== "done") { + this.#fail( + new TypeError( + `a direct-access callback must return "more" or "done", got ` + + `${JSON.stringify(verdict)} (embedder-api amendment A21)`, + ), + ); + return "failed"; + } + const k = scope.marked; + if (k === 0) { + if (verdict === "done") { + // Retraction: the speculative-park correction. The session ends with + // its running total and the peer's operation stays parked. + this.ended = true; + this.endedByVerdict = true; + return "retracted"; + } + this.#fail( + new TypeError( + 'a direct-access callback returned "more" without marking any ' + + "bytes; a session that has nothing to offer retracts by " + + 'returning "done" (embedder-api amendment A21, polyengine#128)', + ), + ); + return "failed"; + } + // Marks acknowledge ON CLEAN RETURN ONLY: this is the first and only + // place the peer's progress moves, and it completes the copy with `k`. + peer.advanceBytes(k); + this.total += k; + if (verdict === "done") { + this.ended = true; + this.endedByVerdict = true; + } + return "copied"; + } + + failDirect(error: Error): void { + this.#fail(error); + } + + // --- promise plumbing --- + + /** Arm the settle hooks for one issuance of this session. */ + arm( + settle: (step: "done" | "reissue") => void, + reject: (e: unknown) => void, + ): void { + this.#settle = settle; + this.#reject = reject; + } + + #take(): [ + ((s: "done" | "reissue") => void) | null, + ((e: unknown) => void) | null, + ] { + const s = this.#settle, r = this.#reject; + this.#settle = null; + this.#reject = null; + return [s, r]; + } + + #fail(e: unknown): void { + this.ended = true; + this.pending = false; + const [, r] = this.#take(); + r?.(e); + } + + /** The session is over; the driving loop resolves with `total`. */ + finish(): void { + this.ended = true; + this.pending = false; + const [s] = this.#take(); + s?.("done"); + } + + /** This issuance rendezvoused but the session lives; re-issue it. */ + reissue(): void { + this.pending = false; + const [s] = this.#take(); + s?.("reissue"); + } +} + +/** A21 is `stream` only; `null` (zero-width) is not u8 either. */ +function requireU8Element(t: ValType | null, who: string): void { + if (t === null || despecialize(t).kind !== "u8") { + throw new TypeError( + `${who} is available on stream only; this stream's element type ` + + `is ${t === null ? "the zero-width payload" : despecialize(t).kind} ` + + `(embedder-api amendment A21, polyengine#128)`, + ); + } +} + /** Host end the embedder WRITES; the guest reads. */ export interface HostWritableEnd { /** @@ -474,6 +844,27 @@ export interface HostWritableEnd { * if the reader dropped. */ writeAll(values: T[]): Promise; + /** + * Park a **direct session** on this end (`stream` only — embedder-api + * amendment A21, polyengine#128). + * + * At every rendezvous with a reader of nonzero capacity, `produce` runs + * exactly once, synchronously, inside the rendezvous, with a + * `DirectDestination` over the reader's unfilled landing zone — guest linear + * memory when the peer is a guest, so the producer's own `set()` is the + * canonical-ABI copy. `"more"` keeps the session parked for the next + * rendezvous; `"done"` ends it. Resolves with the session's total. + * + * Marks acknowledge on clean return only. `"done"` with zero marked is + * *retraction* (the session ends, the reader's operation stays parked, no + * event); `"more"` with zero marked, and a throwing callback, reject. + * + * Participates in the one-in-flight-per-end rule exactly as `write` does. + */ + writeDirect( + produce: (dest: DirectDestination) => DirectVerdict, + info?: DirectSessionInfo, + ): Promise; /** * Cancel an in-flight `write`/`writeAll` (definitions.py * `SharedStreamImpl.cancel` -> `CopyResult.CANCELLED`). No-op when nothing @@ -502,6 +893,18 @@ export interface HostReadableEnd { * array. */ read(max: number): Promise; + /** + * Park a **direct session** on this end (`stream` only — embedder-api + * amendment A21, polyengine#128). The mirror of + * `HostWritableEnd.writeDirect`: `consume` receives a `DirectSource` over + * the writer's unread bytes (a view of guest memory, or of the offered + * host chunk itself) and may take a prefix — a partial take is normal, and + * the writer re-offers on its own schedule. + */ + readDirect( + consume: (src: DirectSource) => DirectVerdict, + info?: DirectSessionInfo, + ): Promise; /** Cancel an in-flight `read`; see `HostWritableEnd.cancelWrite`. */ cancelRead(): void; drop(): void; @@ -631,6 +1034,87 @@ function mkStreamEnds( if (result === CopyResult.DROPPED) activity.close(); else activity.notify(); }; + /** The live direct session on each end, if any (A21, polyengine#128). */ + const direct: { read: DirectSession | null; write: DirectSession | null } = { + read: null, + write: null, + }; + /** + * Drive one direct session from park to end (A21). + * + * Two shapes reach us, and the difference is *which side arrived second*: + * + * * the session is the PENDING side — every rendezvous fires `onCopy`, and + * the `"more"` verdict simply declines to `reclaim()`, so the session + * stays in the pending slot for the next peer operation. This is + * `write()`'s "stay parked until the offer is exhausted" mechanism, with + * the callback's verdict in place of `buf.remain() > 0`. + * * the session ARRIVED second — the rendezvous completes it with + * `onCopyDone(COMPLETED)`, so a `"more"` verdict has to re-issue. The + * re-issue rides the loop below (one `await` apart), which is exactly + * `writeAll`'s re-offer shape and therefore inherits its ordering: the + * peer's pending event is delivered and its buffer reclaimed before we + * can rendezvous against it a second time. + */ + const runDirectSession = async ( + side: "read" | "write", + session: DirectSession, + ): Promise => { + parked[side] = true; + direct[side] = session; + try { + for (;;) { + if (session.cancelled) break; + const step = await new Promise<"done" | "reissue">((res, rej) => { + session.arm(res, rej); + session.pending = true; + const onCopy = (reclaim: () => void): void => { + if (!session.ended) return; // "more": stay parked + reclaim(); + activity.notify(); + session.finish(); + }; + const onCopyDone = (result: CopyResult): void => { + session.pending = false; + settle(result); + // COMPLETED with the session still live == the arriving-side + // rendezvous above; anything else (DROPPED, CANCELLED, or the + // retraction path through `reset_and_notify_pending`) ends it. + if (result === CopyResult.COMPLETED && !session.ended) { + session.reissue(); + } else { + session.finish(); + } + }; + if (side === "write") { + shared.write(writeInst, session as never, onCopy, onCopyDone); + } else { + shared.read(readInst, session as never, onCopy, onCopyDone); + } + activity.notify(); + activity.pump(); + }); + if (step === "done") break; + } + } finally { + parked[side] = false; + direct[side] = null; + } + return session.total; + }; + /** Shared tail of `cancelWrite`/`cancelRead` for a parked direct session. */ + const cancelDirect = (session: DirectSession): void => { + // A21: cancelling RETRACTS the session — it resolves with its running + // total (A8's indistinguishability caveats unchanged). `shared.cancel()` + // only when the session actually holds the pending slot: a session caught + // between two issuances holds nothing, and `SharedBase.cancel` asserts + // that something is pending. + session.cancelled = true; + if (session.pending) shared.cancel(); + else session.finish(); + activity.notify(); + activity.pump(); + }; return { writable: { write(values: T[]): Promise { @@ -702,8 +1186,31 @@ function mkStreamEnds( } return sent; }, + writeDirect( + produce: (dest: DirectDestination) => DirectVerdict, + info?: DirectSessionInfo, + ): Promise { + // Same one-in-flight-per-end rule, same wording shape as `write`: + // `writeDirect` participates in it exactly as `write` does (A21). + if (parked.write) { + throw new TypeError( + "a write is already in flight on this stream's writable end; " + + "await it or cancelWrite() first", + ); + } + requireU8Element(shared.t, "writeDirect"); + const session = new DirectSession(shared.t, (scope) => produce(scope)); + const p = runDirectSession("write", session); + if (info === undefined) return p; + return p.then((n) => { + info.endedByVerdict = session.endedByVerdict; + return n; + }); + }, cancelWrite() { if (!parked.write) return; + const session = direct.write; + if (session !== null) return cancelDirect(session); parked.write = false; shared.cancel(); activity.notify(); @@ -751,6 +1258,25 @@ function mkStreamEnds( activity.pump(); }); }, + readDirect( + consume: (src: DirectSource) => DirectVerdict, + info?: DirectSessionInfo, + ): Promise { + if (parked.read) { + throw new TypeError( + "a read is already in flight on this stream's readable end; " + + "await it or cancelRead() first", + ); + } + requireU8Element(shared.t, "readDirect"); + const session = new DirectSession(shared.t, (scope) => consume(scope)); + const p = runDirectSession("read", session); + if (info === undefined) return p; + return p.then((n) => { + info.endedByVerdict = session.endedByVerdict; + return n; + }); + }, cancelRead() { // #97, DELIBERATE AND PINNED: cancelling resolves the in-flight // `read` promise with whatever the buffer took so far — for a read @@ -763,6 +1289,8 @@ function mkStreamEnds( // already knows which of the two happened. Nothing else can reach // this state — a guest cannot cancel the host's read. if (!parked.read) return; + const session = direct.read; + if (session !== null) return cancelDirect(session); parked.read = false; shared.cancel(); activity.notify(); diff --git a/runtime/src/task/streams.ts b/runtime/src/task/streams.ts index 357d8d2..146e513 100644 --- a/runtime/src/task/streams.ts +++ b/runtime/src/task/streams.ts @@ -47,6 +47,7 @@ import { defineBrand, ERROR_CONTEXT } from "@polyengine/protocol"; import { assert_, Trap, trapIf } from "../cabi/trap.ts"; import { LiftLowerContext } from "../cabi/context.ts"; +import { bytesOf } from "../cabi/memory.ts"; import { loadListFromValidRange } from "../cabi/load.ts"; import { storeListIntoValidRange } from "../cabi/store.ts"; import { alignment, alignTo, elemSize } from "../cabi/layout.ts"; @@ -169,6 +170,174 @@ export class GuestBuffer { } this.progress += vs.length; } + + // --- A21 direct-access byte edges (embedder-api amendment A21, #128) --- + // + // `ByteWindow`, implemented for the `stream` case only. The two methods + // together are the copy `read`/`write` would have done, split so that the + // *peer's* callback performs it: `byteView` hands out the range, and + // `advanceBytes` records the bytes that actually moved. They are role-blind + // (destination or source) because `this.ptr` already advances on BOTH + // `read` and `write` above, and `elemSize(u8) === 1`. + + /** + * A fresh view over the next `n` bytes of this buffer's remaining range. + * + * Fresh on every call, via `bytesOf` (cabi/memory.ts:195) over the + * `LiveMemory` getters — so a `memory.grow` between two rendezvous of one + * parked direct session never yields a view onto the detached buffer. + */ + byteView(n: number): Uint8Array { + assert_( + this.t !== null && despecialize(this.t).kind === "u8", + "direct byte window on a non-u8 buffer", + ); + assert_(n <= this.remain(), "direct byte window beyond remaining"); + const mem = this.cx.opts.memory; + assert_(mem !== null, "direct byte window requires a memory"); + return bytesOf(mem!, this.ptr, n); + } + + /** + * Advance by `k` WITHOUT copying: the bytes already moved through the view + * `byteView` handed out. Called by the seam only after the direct callback + * returned cleanly, which is what makes marks acknowledge-on-clean-return. + */ + advanceBytes(k: number): void { + assert_(k >= 0 && k <= this.remain(), "direct advance beyond remaining"); + this.ptr += k; // elemSize(u8) === 1 + this.progress += k; + } +} + +// --------------------------------------------------------------------------- +// The direct-access seam (embedder-api amendment A21, 2026-08-22, #128) +// --------------------------------------------------------------------------- +// +// A21 lets ONE side of a rendezvous be a *direct session*: instead of handing +// the rendezvous a buffer to copy out of / into, the host parks a callback +// that runs synchronously inside the rendezvous and performs the canonical +// copy itself, against a scoped view of the peer's memory. +// +// The seam below is the whole of it inside this file. It exists so that the +// rendezvous keeps mirroring definitions.py `SharedStreamImpl.read`/`.write` +// (lines 1032/1050) line for line for every non-direct path: when neither +// side is direct, `rendezvousCopy` IS `dst.write(src.read(n))`, unchanged. +// +// Nothing here imports from `exec/`: a direct session is recognised +// structurally (`direct === true`) and driven through two small optional +// protocols — `DirectBuffer` (the session) and `ByteWindow` (the peer). + +/** + * The buffer surface the rendezvous actually uses (definitions.py `Buffer`, + * line 918). `GuestBuffer` and the host layer's `HostBuffer` both satisfy it. + */ +export interface RendezvousBuffer { + remain(): number; + isZeroLength(): boolean; + read(n: number): PayloadChunk; + write(vs: PayloadChunk): void; +} + +/** + * A21: the peer half of a direct rendezvous — a buffer that can expose its + * remaining range as bytes and be advanced without a copy. + * + * Implemented by `GuestBuffer` (a view into guest linear memory: the + * embedder's own `set()` becomes the one ABI copy) and by `HostBuffer` (a + * view of the offered chunk when it is the source; a synthesized scratch that + * becomes the delivered chunk when it is the destination). + */ +export interface ByteWindow { + /** + * A view over the next `n` bytes. May be called several times within one + * direct invocation (`remaining()` re-derives on every call); an + * implementation that *synthesizes* the window must return the same + * storage for the whole invocation and release it in `endWindow`. + */ + byteView(n: number): Uint8Array; + /** Record `k` bytes as moved. Called only after a clean callback return. */ + advanceBytes(k: number): void; + /** End of one direct invocation; drop any synthesized window. */ + endWindow?(): void; +} + +/** + * A21: the parked direct session, as the rendezvous sees it. It presents the + * ordinary buffer surface (so `remain()`/`isZeroLength()` keep the reference + * control flow working) but its `read`/`write` are never called — the seam + * routes it through `runDirect` instead. + */ +export interface DirectBuffer extends RendezvousBuffer { + readonly direct: true; + /** + * Run this session's callback exactly once against the peer's window, + * with `n` bytes of capacity. Applies the acknowledged marks to `peer` + * itself, and settles the session on failure — the seam only routes the + * rendezvous state that follows. + */ + runDirect(peer: ByteWindow, n: number): DirectOutcome; + /** + * Reject this session out-of-band (the two-direct-sessions rendezvous, + * where neither side owns memory). + */ + failDirect(error: Error): void; +} + +/** + * What the seam did, and hence how the rendezvous must continue. + * + * * `"chunk"` — no direct session was involved: the reference copy ran. + * * `"copied"` — the callback acknowledged ≥ 1 byte; continue exactly as + * after a reference copy (fire the pending side's `on_copy`). + * * `"retracted"` — `"done"` with zero marked. Continue as if the direct + * side's buffer had had `remain() == 0` all along, which is a state + * definitions.py already routes. + * * `"failed"` — misuse or a throwing callback; the session has already + * rejected. No copy, no event, the peer's parked operation survives. + * * `"both-direct"` — neither side owns memory; the ARRIVING side is + * rejected by the caller and the parked side is left undisturbed. + */ +export type DirectOutcome = "copied" | "retracted" | "failed"; +export type RendezvousOutcome = DirectOutcome | "chunk" | "both-direct"; + +function isDirectBuffer(b: RendezvousBuffer): b is DirectBuffer { + return (b as { direct?: unknown }).direct === true; +} + +/** + * The one copy site, shared by `SharedStreamImpl.read` and `.write`. + * + * Collapses to definitions.py's `dst_buffer.write(src_buffer.read(n))` + * whenever neither side is a direct session — which is every guest↔guest, + * guest↔host-chunk and host-chunk↔host-chunk rendezvous, i.e. everything + * that existed before A21. + */ +function rendezvousCopy( + src: RendezvousBuffer, + dst: RendezvousBuffer, + n: number, +): RendezvousOutcome { + const srcDirect = isDirectBuffer(src); + const dstDirect = isDirectBuffer(dst); + if (!srcDirect && !dstDirect) { + dst.write(src.read(n)); + return "chunk"; + } + if (srcDirect && dstDirect) return "both-direct"; + return srcDirect + ? src.runDirect(dst as unknown as ByteWindow, n) + : (dst as DirectBuffer).runDirect(src as unknown as ByteWindow, n); +} + +/** The A21 rejection for a rendezvous of two direct sessions. */ +function bothDirectError(): TypeError { + return new TypeError( + "at least one side of a host-to-host rendezvous must use the chunk " + + "forms: two direct-access sessions cannot rendezvous with each other " + + "because neither side owns the memory the other would write into " + + "(embedder-api amendment A21, polyengine#128)", + ); } /** @@ -326,7 +495,28 @@ export class SharedStreamImpl implements SharedBase { if (this.pendingBuffer.remain() > 0) { if (dstBuffer.remain() > 0) { const n = Math.min(dstBuffer.remain(), this.pendingBuffer.remain()); - dstBuffer.write(this.pendingBuffer.read(n)); + // A21 seam (#128). `"chunk"` is the reference line verbatim. + const pendingIsDirect = isDirectBuffer(this.pendingBuffer); + const out = rendezvousCopy(this.pendingBuffer, dstBuffer, n); + if (out === "both-direct") { + // The ARRIVING side (here the reader) is the one refused; the + // parked session keeps the pending slot, undisturbed. + (dstBuffer as unknown as DirectBuffer).failDirect( + bothDirectError(), + ); + return; + } + if (out === "retracted" || out === "failed") { + this.#routeDirectNoCopy( + out, + pendingIsDirect, + inst, + dstBuffer, + onCopy, + onCopyDone, + ); + return; + } this.pendingOnCopy!(() => this.resetPending()); } onCopyDone(CopyResult.COMPLETED); @@ -355,7 +545,28 @@ export class SharedStreamImpl implements SharedBase { if (this.pendingBuffer.remain() > 0) { if (srcBuffer.remain() > 0) { const n = Math.min(srcBuffer.remain(), this.pendingBuffer.remain()); - this.pendingBuffer.write(srcBuffer.read(n)); + // A21 seam (#128). `"chunk"` is the reference line verbatim. + const pendingIsDirect = isDirectBuffer(this.pendingBuffer); + const out = rendezvousCopy(srcBuffer, this.pendingBuffer, n); + if (out === "both-direct") { + // The ARRIVING side (here the writer) is refused; the parked + // session keeps the pending slot. + (srcBuffer as unknown as DirectBuffer).failDirect( + bothDirectError(), + ); + return; + } + if (out === "retracted" || out === "failed") { + this.#routeDirectNoCopy( + out, + pendingIsDirect, + inst, + srcBuffer, + onCopy, + onCopyDone, + ); + return; + } this.pendingOnCopy!(() => this.resetPending()); } onCopyDone(CopyResult.COMPLETED); @@ -373,6 +584,48 @@ export class SharedStreamImpl implements SharedBase { } } + /** + * A21 (#128): route a rendezvous whose direct session did NOT copy. + * + * Two outcomes land here, and both share one invariant: the peer's parked + * operation survives, no event is delivered, and the stream is not dropped + * — a runtime never emits a zero-progress COMPLETED copy, which is + * unreachable in definitions.py for a nonzero-capacity operation. + * + * * `"retracted"` — `"done"` with zero marked. The session ends and + * resolves with its running total, through the ordinary + * `on_copy_done(COMPLETED)` channel. + * * `"failed"` — misuse or a throwing callback. The session has ALREADY + * rejected (`DirectSession.#fail`), so it must be retired silently: + * its rejection is its notification. + * + * Which side was the session decides where each goes, and both shapes are + * states definitions.py already produces: + * + * * PARKED session ⇒ the "the parked side had nothing left" branch + * (definitions.py:1043/1063): retire it and park the arriving + * operation, which gets no event either way. + * * ARRIVING session ⇒ the "arriving buffer of zero capacity" state + * (definitions.py:1041/1057): the pending side is left untouched with + * its `on_copy` unfired, and the arriving side completes. + */ + #routeDirectNoCopy( + out: "retracted" | "failed", + pendingIsDirect: boolean, + inst: unknown, + arriving: GuestBuffer, + onCopy: OnCopy, + onCopyDone: OnCopyDone, + ): void { + if (pendingIsDirect) { + if (out === "retracted") this.resetAndNotifyPending(CopyResult.COMPLETED); + else this.resetPending(); + this.setPending(inst, arriving, onCopy, onCopyDone); + return; + } + if (out === "retracted") onCopyDone(CopyResult.COMPLETED); + } + #assertSameElemType(b: GuestBuffer): void { // Structural, not identity: definitions.py compares dataclass types with // `==`, and our `ValType`s are fresh objects per table (the plan's type diff --git a/runtime/tests/direct_streams_test.ts b/runtime/tests/direct_streams_test.ts new file mode 100644 index 0000000..657684e --- /dev/null +++ b/runtime/tests/direct_streams_test.ts @@ -0,0 +1,819 @@ +// Direct-access byte edges at the RAW seam (contracts/embedder-api.md +// §"Streams and futures", amendment A21, 2026-08-22, polyengine#128). +// +// WHAT A21 IS +// =========== +// +// For `stream` only, a host end may park a *direct session* instead of a +// chunk: at every rendezvous with a peer operation of nonzero capacity, the +// session's callback runs exactly once, synchronously, inside the rendezvous, +// against a scoped view of the peer's bytes. When the peer is a guest, that +// view aliases guest linear memory, so the embedder's own `set()` IS the +// single canonical-ABI copy. +// +// The implementation is split in two, and so are these tests: +// +// * task/streams.ts holds the SEAM — `rendezvousCopy` plus the routing of +// its three direct outcomes at the two copy sites of +// `SharedStreamImpl.read`/`.write` (definitions.py:1032/1050). Every +// non-direct path must stay byte-identical to the reference, which is +// what the `"chunk"` outcome is: `dst.write(src.read(n))`, verbatim. +// * exec/host_streams.ts holds the SESSION — the scoped callback object, +// mark accounting, the `"more"`/`"done"` cadence and the promise. +// +// HOW THE "GUEST" IS MODELLED HERE +// ================================ +// +// These tests drive `SharedStreamImpl` directly with real `GuestBuffer`s over +// a real `WebAssembly.Memory`, standing in for the guest's `stream.read` / +// `stream.write` trampoline (intrinsics/stream_builtins.ts `streamCopy`). +// The one simplification: the stand-in reclaims the pending buffer inside +// `on_copy` rather than at event-delivery time. That models the post-pump +// state a real guest reaches — `HostActivity.pump()` runs synchronously +// inside every host operation and delivers the armed event — and it is the +// same model `HostBuffer`-backed host ends already use. + +import { assertEq } from "./support/asserts.ts"; +import { CopyResult, GuestBuffer, SharedStreamImpl } from "../src/task/mod.ts"; +import { LiftLowerContext, mkCanonicalOptions } from "../src/cabi/context.ts"; +import type { ValType } from "../src/cabi/types.ts"; +import type { + DirectDestination, + DirectSource, +} from "../src/exec/host_streams.ts"; +import { hostStream } from "../src/exec/mod.ts"; + +const U8: ValType = { kind: "u8" }; +const U32: ValType = { kind: "u32" }; + +function assert(cond: boolean, msg: string): asserts cond { + if (!cond) throw new Error(`assertion failed: ${msg}`); +} + +async function caught(p: PromiseLike): Promise { + try { + await p; + } catch (e) { + return e; + } + return undefined; +} + +function caughtSync(fn: () => unknown): unknown { + try { + fn(); + } catch (e) { + return e; + } + return undefined; +} + +// --------------------------------------------------------------------------- +// A live `MemInst` over a real `WebAssembly.Memory` +// --------------------------------------------------------------------------- +// +// Structurally what exec/boundary.ts `LiveMemory` is: `bytes`/`view` are +// GETTERS that re-derive from `memory.buffer`, so a `memory.grow` (which +// detaches the old ArrayBuffer) is invisible to holders of the MemInst. That +// is exactly the property A21's "views are re-derived per `remaining()` call" +// rule depends on. + +function mkMemory(initial = 1) { + const memory = new WebAssembly.Memory({ initial }); + const view = { + addrType: "i32" as const, + get bytes() { + return new Uint8Array(memory.buffer); + }, + get view() { + return new DataView(memory.buffer); + }, + get length() { + return memory.buffer.byteLength; + }, + ptrType: () => "i32" as const, + ptrSize: () => 4 as const, + }; + return { memory, view }; +} + +function mkCx(view: unknown): LiftLowerContext { + // deno-lint-ignore no-explicit-any + return new LiftLowerContext(mkCanonicalOptions({ memory: view as any })); +} + +/** One instance sentinel per side; the rendezvous compares them by identity. */ +const GUEST_A = Object.freeze({ guest: "a" }); +const GUEST_B = Object.freeze({ guest: "b" }); + +interface GuestOp { + buf: GuestBuffer; + /** Events the guest end observed, in order (result + progress at delivery). */ + events: { result: CopyResult; progress: number }[]; +} + +function guestOp( + shared: SharedStreamImpl, + kind: "read" | "write", + cx: LiftLowerContext, + ptr: number, + len: number, + inst: unknown = GUEST_A, + t: ValType | null = U8, +): GuestOp { + const buf = new GuestBuffer(t, cx, ptr, len); + const events: { result: CopyResult; progress: number }[] = []; + const onCopy = (reclaim: () => void) => { + reclaim(); + events.push({ result: CopyResult.COMPLETED, progress: buf.progress }); + }; + const onCopyDone = (result: CopyResult) => + events.push({ result, progress: buf.progress }); + if (kind === "read") shared.read(inst, buf, onCopy, onCopyDone); + else shared.write(inst, buf, onCopy, onCopyDone); + return { buf, events }; +} + +/** The bytes of guest memory at `[ptr, ptr+len)`. */ +function memBytes(memory: WebAssembly.Memory, ptr: number, len: number) { + return [...new Uint8Array(memory.buffer, ptr, len)]; +} + +function fill(memory: WebAssembly.Memory, ptr: number, vs: number[]) { + new Uint8Array(memory.buffer).set(Uint8Array.from(vs), ptr); +} + +// =========================================================================== +// 1. The two arrival orders, both directions +// =========================================================================== + +Deno.test("A21 writeDirect: parked session, guest read arrives", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + let aliased = false; + let capacity = -1; + const session = hs.writable.writeDirect((dest: DirectDestination) => { + const win = dest.remaining(); + // ONE-COPY PIN: the view the producer is handed IS guest linear memory, + // so its `set()` is the canonical-ABI copy and nothing else copies. + aliased = win.buffer === memory.buffer; + capacity = win.length; + win.set(Uint8Array.from([7, 8, 9])); + dest.markWritten(3); + return "done"; + }); + + // The session parks synchronously; only now does the reader arrive. + const g = guestOp(shared, "read", cx, 64, 5); + assertEq(await session, 3, "session total"); + assert(aliased, "the destination view aliases the guest's memory buffer"); + // The BUFFER_MAX_LENGTH sentinel the parked session reports to the + // rendezvous never surfaces: the capacity is the reader's actual remaining. + assertEq(capacity, 5, "capacity is the reader's remaining, not the sentinel"); + assertEq(memBytes(memory, 64, 5), [7, 8, 9, 0, 0], "bytes land at ptr"); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + assertEq(g.buf.progress, 3, "the guest read completes with progress 3"); +}); + +Deno.test("A21 writeDirect: guest read parked, session arrives", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const g = guestOp(shared, "read", cx, 128, 4); + let capacity = -1; + const n = await hs.writable.writeDirect((dest) => { + capacity = dest.remaining().length; + dest.remaining().set(Uint8Array.from([1, 2, 3, 4])); + dest.markWritten(4); + return "done"; + }); + assertEq(n, 4, "session total"); + assertEq(capacity, 4); + assertEq(memBytes(memory, 128, 4), [1, 2, 3, 4]); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 4 }]); +}); + +Deno.test("A21 readDirect: parked session, guest write arrives", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + fill(memory, 256, [10, 20, 30]); + + let aliased = false; + const got: number[] = []; + const session = hs.readable.readDirect((src: DirectSource) => { + const win = src.remaining(); + aliased = win.buffer === memory.buffer; + got.push(...win); + src.markRead(win.length); + return "done"; + }); + + const g = guestOp(shared, "write", cx, 256, 3); + assertEq(await session, 3); + assert(aliased, "the source view aliases the guest's memory buffer"); + assertEq(got, [10, 20, 30]); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); +}); + +Deno.test("A21 readDirect: guest write parked, session arrives (partial take)", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + fill(memory, 300, [1, 2, 3, 4, 5, 6]); + + const g = guestOp(shared, "write", cx, 300, 6); + const got: number[] = []; + const n = await hs.readable.readDirect((src) => { + // A PARTIAL take is normal Component Model behaviour: the writer's copy + // completes with the marked count and it re-offers on its own schedule. + got.push(...src.remaining().subarray(0, 2)); + src.markRead(2); + return "done"; + }); + assertEq(n, 2, "session total is the marked prefix"); + assertEq(got, [1, 2]); + assertEq( + g.events, + [{ result: CopyResult.COMPLETED, progress: 2 }], + "the guest write completes with progress 2, not 6", + ); +}); + +// =========================================================================== +// 2. Multi-rendezvous sessions +// =========================================================================== + +Deno.test('A21: a "more" session drains across several guest reads', async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const payload = Uint8Array.from([1, 2, 3, 4, 5, 6, 7]); + let sent = 0; + const caps: number[] = []; + const session = hs.writable.writeDirect((dest) => { + const win = dest.remaining(); + caps.push(win.length); + const k = Math.min(win.length, payload.length - sent); + win.set(payload.subarray(sent, sent + k)); + dest.markWritten(k); + sent += k; + return sent < payload.length ? "more" : "done"; + }); + + // Three guest reads of 3, 3 and 3: the third finds only one byte left. + const g1 = guestOp(shared, "read", cx, 400, 3); + const g2 = guestOp(shared, "read", cx, 410, 3); + const g3 = guestOp(shared, "read", cx, 420, 3); + assertEq(await session, 7, "session total across three rendezvous"); + assertEq(caps, [3, 3, 3], "each invocation sees the reader's own capacity"); + assertEq(memBytes(memory, 400, 3), [1, 2, 3]); + assertEq(memBytes(memory, 410, 3), [4, 5, 6]); + assertEq(memBytes(memory, 420, 3), [7, 0, 0]); + assertEq(g1.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + assertEq(g2.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + assertEq(g3.events, [{ result: CopyResult.COMPLETED, progress: 1 }]); +}); + +Deno.test('A21: a "more" readDirect session drains across several guest writes', async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + fill(memory, 500, [1, 2, 3]); + fill(memory, 510, [4, 5, 6]); + + const got: number[] = []; + const session = hs.readable.readDirect((src) => { + const win = src.remaining(); + got.push(...win); + src.markRead(win.length); + return got.length >= 6 ? "done" : "more"; + }); + const g1 = guestOp(shared, "write", cx, 500, 3); + const g2 = guestOp(shared, "write", cx, 510, 3); + assertEq(await session, 6); + assertEq(got, [1, 2, 3, 4, 5, 6]); + assertEq(g1.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + assertEq(g2.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); +}); + +// =========================================================================== +// 3. Retraction — "done" with zero marked +// =========================================================================== + +Deno.test("A21 retraction: the arriving reader stays parked, with no event", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const session = hs.writable.writeDirect((dest) => { + // The speculative-park correction: demand arrived while the ring + // happened to be empty. Nothing marked, so nothing is acknowledged. + dest.remaining(); + return "done"; + }); + const g = guestOp(shared, "read", cx, 600, 4); + assertEq(await session, 0, "the session resolves with its running total"); + assertEq(g.events, [], "no event is delivered to the parked reader"); + + // The reader is still PARKED: a later chunk write completes it. + assertEq(await hs.writable.write(Uint8Array.from([5, 6]) as never), 2); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); + assertEq(memBytes(memory, 600, 4), [5, 6, 0, 0]); +}); + +Deno.test("A21 retraction: the parked guest reader is untouched when the session arrives second", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const g = guestOp(shared, "read", cx, 700, 4); + assertEq(await hs.writable.writeDirect(() => "done"), 0); + assertEq(g.events, [], "no event for the parked reader"); + assertEq(await hs.writable.write(Uint8Array.from([9]) as never), 1); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 1 }]); + assertEq(memBytes(memory, 700, 2), [9, 0]); +}); + +Deno.test("A21 retraction: readDirect leaves a parked guest writer parked", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + fill(memory, 800, [3, 4]); + + const g = guestOp(shared, "write", cx, 800, 2); + assertEq(await hs.readable.readDirect(() => "done"), 0); + assertEq(g.events, []); + // Still parked: a chunk read drains it. + assertEq([...(await hs.readable.read(8)) as unknown as Uint8Array], [3, 4]); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); +}); + +// =========================================================================== +// 4. Misuse — the session fails, the peer survives, the stream lives +// =========================================================================== + +Deno.test('A21 misuse: "more" with zero marked rejects TypeError', async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const session = hs.writable.writeDirect(() => "more"); + const g = guestOp(shared, "read", cx, 900, 4); + const e = await caught(session); + assert(e instanceof TypeError, `TypeError, got ${e}`); + assert( + String(e.message).includes("without marking any"), + `names the rule: ${e}`, + ); + assertEq(g.events, [], "no event: never a zero-progress COMPLETED copy"); + assertEq(shared.dropped, false, "the stream stays alive"); + // The reader is still parked and the host may fall back to the chunk form. + assertEq(await hs.writable.write(Uint8Array.from([1, 2]) as never), 2); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); +}); + +Deno.test("A21 misuse: a throwing callback rejects and discards its marks", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + const boom = new Error("producer blew up"); + + const session = hs.writable.writeDirect((dest) => { + // Bytes physically written past the acknowledged progress must be + // unobservable to the peer: the mark is discarded with the throw. + dest.remaining().set(Uint8Array.from([0xff, 0xff, 0xff])); + dest.markWritten(3); + throw boom; + }); + const g = guestOp(shared, "read", cx, 1000, 4); + assertEq(await caught(session), boom, "rejects with the callback's error"); + assertEq(g.events, [], "no event"); + assertEq(g.buf.progress, 0, "the peer's progress did not move"); + + // The parked reader survives, and a chunk write delivers UNPOLLUTED data: + // whatever the failed callback scribbled is past the acknowledged + // progress, so the reader's own copy overwrites it. + assertEq(await hs.writable.write(Uint8Array.from([1, 2]) as never), 2); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); + assertEq(memBytes(memory, 1000, 2), [1, 2], "peer sees only its own copy"); +}); + +Deno.test("A21 misuse: the same failures leave a PARKED guest reader parked", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const g = guestOp(shared, "read", cx, 1100, 4); + const e = await caught(hs.writable.writeDirect(() => "more")); + assert(e instanceof TypeError, `TypeError, got ${e}`); + assertEq(g.events, [], "the parked reader got no event"); + assertEq(await hs.writable.write(Uint8Array.from([4, 5]) as never), 2); + assertEq(g.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); +}); + +Deno.test("A21 misuse: over-marking throws inside the callback", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + let inner: unknown; + const session = hs.writable.writeDirect((dest) => { + dest.markWritten(2); + inner = caughtSync(() => dest.markWritten(3)); // 2 + 3 > 4 + return "done"; + }); + guestOp(shared, "read", cx, 1200, 4); + assertEq(await session, 2, "only the legal marks were acknowledged"); + assert(inner instanceof TypeError, `TypeError, got ${inner}`); + assert( + String((inner as Error).message).includes("cumulative"), + `names the rule: ${inner}`, + ); + + // Negative / non-integer marks are refused the same way. + const hs2 = hostStream(U8); + const shared2 = hs2.value as unknown as SharedStreamImpl; + let bad: unknown[] = []; + const s2 = hs2.writable.writeDirect((dest) => { + bad = [ + caughtSync(() => dest.markWritten(-1)), + caughtSync(() => dest.markWritten(1.5)), + ]; + dest.markWritten(1); + return "done"; + }); + guestOp(shared2, "read", cx, 1300, 4); + assertEq(await s2, 1); + assert(bad.every((e) => e instanceof TypeError), `both refused: ${bad}`); +}); + +Deno.test("A21 scoping: the view is dead once the callback returns", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + let escaped: DirectDestination | null = null; + const session = hs.writable.writeDirect((dest) => { + escaped = dest; + dest.markWritten(1); + return "done"; + }); + guestOp(shared, "read", cx, 1400, 4); + assertEq(await session, 1); + + const d = escaped as unknown as DirectDestination; + for ( + const [what, fn] of [ + ["remaining", () => d.remaining()], + ["markWritten", () => d.markWritten(1)], + ] as const + ) { + const e = caughtSync(fn); + assert(e instanceof TypeError, `${what} after return: TypeError, got ${e}`); + assert( + String(e.message).includes("scoped to the synchronous callback"), + `${what} names the scoping rule: ${e}`, + ); + } +}); + +Deno.test("A21: non-u8 and zero-width element types are refused", () => { + for (const [label, t] of [["u32", U32], ["zero-width", null]] as const) { + const hs = hostStream(t); + const w = caughtSync(() => hs.writable.writeDirect(() => "done")); + const r = caughtSync(() => hs.readable.readDirect(() => "done")); + for (const [who, e] of [["writeDirect", w], ["readDirect", r]] as const) { + assert(e instanceof TypeError, `${label} ${who}: TypeError, got ${e}`); + assert( + String(e.message).includes("stream only"), + `${label} ${who} names A21's scope: ${e}`, + ); + } + } +}); + +Deno.test("A21: the one-in-flight-per-end rule covers the direct forms", async () => { + const hs = hostStream(U8); + + // A parked chunk write blocks writeDirect, and vice versa. + const w = hs.writable.write(Uint8Array.from([1]) as never); + const e1 = caughtSync(() => hs.writable.writeDirect(() => "done")); + assert(e1 instanceof TypeError, `write||writeDirect: ${e1}`); + hs.writable.cancelWrite(); + await w; + + const wd = hs.writable.writeDirect((d) => { + d.markWritten(0); + return "done"; + }); + const e2 = caughtSync(() => hs.writable.write(Uint8Array.from([1]) as never)); + assert(e2 instanceof TypeError, `writeDirect||write: ${e2}`); + const e3 = caughtSync(() => hs.writable.writeDirect(() => "done")); + assert(e3 instanceof TypeError, `writeDirect||writeDirect: ${e3}`); + hs.writable.cancelWrite(); + await wd; + + // The read end is independent (that is the pass-through data plane) but + // has the same rule within itself. + const r = hs.readable.read(4); + const e4 = caughtSync(() => hs.readable.readDirect(() => "done")); + assert(e4 instanceof TypeError, `read||readDirect: ${e4}`); + hs.readable.cancelRead(); + await r; + + const rd = hs.readable.readDirect(() => "done"); + const e5 = caughtSync(() => hs.readable.read(4)); + assert(e5 instanceof TypeError, `readDirect||read: ${e5}`); + const e6 = caughtSync(() => hs.readable.readDirect(() => "done")); + assert(e6 instanceof TypeError, `readDirect||readDirect: ${e6}`); + hs.readable.cancelRead(); + await rd; +}); + +// =========================================================================== +// 5. Zero-length probes (Concurrency.md "Stream Readiness") +// =========================================================================== + +Deno.test("A21: a zero-length probe completes without invoking the callback", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + + // Reader direction: a parked writeDirect answers a zero-length read with + // immediate COMPLETED — the armed session IS the readiness claim. + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + let calls = 0; + const wsession = hs.writable.writeDirect((d) => { + calls++; + d.markWritten(1); + return "done"; + }); + const probe = guestOp(shared, "read", cx, 1500, 0); + assertEq(calls, 0, "the producer was not invoked by the probe"); + assertEq(probe.events, [{ result: CopyResult.COMPLETED, progress: 0 }]); + // The session is still parked, and a real read still drives it. + const real = guestOp(shared, "read", cx, 1500, 4); + assertEq(await wsession, 1); + assertEq(calls, 1); + assertEq(real.events, [{ result: CopyResult.COMPLETED, progress: 1 }]); + + // Writer direction: a zero-length write against a parked readDirect. + const hs2 = hostStream(U8); + const shared2 = hs2.value as unknown as SharedStreamImpl; + let consumed = 0; + const rsession = hs2.readable.readDirect((s) => { + consumed++; + s.markRead(1); + return "done"; + }); + const probe2 = guestOp(shared2, "write", cx, 1600, 0); + assertEq(consumed, 0, "the consumer was not invoked by the probe"); + assertEq(probe2.events, [{ result: CopyResult.COMPLETED, progress: 0 }]); + guestOp(shared2, "write", cx, 1600, 2); + assertEq(await rsession, 1); + assertEq(consumed, 1); +}); + +// =========================================================================== +// 6. Teardown: drop and cancel +// =========================================================================== + +Deno.test("A21: a peer drop mid-session resolves with the running total", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const session = hs.writable.writeDirect((d) => { + d.markWritten(2); + return "more"; + }); + guestOp(shared, "read", cx, 1700, 2); + guestOp(shared, "read", cx, 1710, 2); + // The reader goes away while the session is still parked. + shared.drop(); + assertEq(await session, 4, "resolves with everything acknowledged so far"); +}); + +Deno.test("A21: cancelWrite/cancelRead retract a parked session", async () => { + const { view } = mkMemory(); + const cx = mkCx(view); + + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + const w = hs.writable.writeDirect((d) => { + d.markWritten(1); + return "more"; + }); + guestOp(shared, "read", cx, 1800, 1); + hs.writable.cancelWrite(); + assertEq(await w, 1, "cancel resolves with the running total"); + + const hs2 = hostStream(U8); + const r = hs2.readable.readDirect(() => "more"); + hs2.readable.cancelRead(); + assertEq(await r, 0, "a never-rendezvoused session cancels to zero"); +}); + +// =========================================================================== +// 7. External byte movers: a SharedArrayBuffer-backed source +// =========================================================================== + +Deno.test("A21: a SAB-backed producer copies straight into the guest view", async () => { + const { memory, view } = mkMemory(); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + // The motivating shape: bytes live in a shared ring the runtime must never + // see a second copy of. `set()` from a SAB-backed view into the guest view + // is the one ABI copy. + const ring = new Uint8Array(new SharedArrayBuffer(4)); + ring.set([11, 12, 13, 14]); + const session = hs.writable.writeDirect((dest) => { + dest.remaining().set(ring); + dest.markWritten(ring.length); + return "done"; + }); + guestOp(shared, "read", cx, 1900, 4); + assertEq(await session, 4); + assertEq(memBytes(memory, 1900, 4), [11, 12, 13, 14]); +}); + +// =========================================================================== +// 8. memory.grow between two rendezvous of one parked session +// =========================================================================== + +Deno.test("A21: views are re-derived, so memory.grow between rendezvous is safe", async () => { + const { memory, view } = mkMemory(1); + const cx = mkCx(view); + const hs = hostStream(U8); + const shared = hs.value as unknown as SharedStreamImpl; + + const buffers: ArrayBufferLike[] = []; + let round = 0; + const session = hs.writable.writeDirect((dest) => { + const win = dest.remaining(); + buffers.push(win.buffer); + win.set(Uint8Array.from([round + 1, round + 1])); + dest.markWritten(2); + round++; + return round < 2 ? "more" : "done"; + }); + + const g1 = guestOp(shared, "read", cx, 2000, 2); + const before = memory.buffer; + // Growing DETACHES the old ArrayBuffer; a view cached across the rendezvous + // would be zero-length and its writes would go nowhere. + memory.grow(1); + assert(memory.buffer !== before, "grow produced a fresh buffer"); + const hiPtr = 65536 + 16; + const g2 = guestOp(shared, "read", cx, hiPtr, 2); + assertEq(await session, 4); + + assertEq(buffers[0] === before, true, "first view was over the old buffer"); + assertEq( + buffers[1] === memory.buffer, + true, + "second view is over the NEW buffer", + ); + assertEq(g1.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); + assertEq(g2.events, [{ result: CopyResult.COMPLETED, progress: 2 }]); + assertEq(memBytes(memory, hiPtr, 2), [2, 2], "the second copy landed"); +}); + +// =========================================================================== +// 9. Host <-> host (contract: "at the same floor") +// =========================================================================== + +Deno.test("A21 host<->host: writeDirect against a chunk read(max)", async () => { + // Both arrival orders; the marked prefix of the scratch becomes the + // delivered chunk, handed through `taken()` unsliced. + for (const order of ["read-first", "direct-first"] as const) { + const hs = hostStream(U8); + let cap = -1; + const produce = (d: DirectDestination) => { + cap = d.remaining().length; + d.remaining().set(Uint8Array.from([1, 2, 3])); + d.markWritten(3); + return "done" as const; + }; + let chunk: Promise, session: Promise; + if (order === "read-first") { + chunk = hs.readable.read(8) as unknown as Promise; + session = hs.writable.writeDirect(produce); + } else { + session = hs.writable.writeDirect(produce); + chunk = hs.readable.read(8) as unknown as Promise; + } + const got = await chunk as unknown as Uint8Array; + assertEq(await session, 3, `${order}: session total`); + assertEq(cap, 8, `${order}: capacity is the reader's max`); + assertEq(got instanceof Uint8Array, true, `${order}: u8 chunk shape`); + assertEq([...got], [1, 2, 3], `${order}: payload`); + } +}); + +Deno.test("A21 host<->host: readDirect against a chunk write borrows the offered chunk", async () => { + for (const order of ["write-first", "direct-first"] as const) { + const hs = hostStream(U8); + const offered = Uint8Array.from([4, 5, 6, 7]); + let aliased = false; + const got: number[] = []; + const consume = (s: DirectSource) => { + const win = s.remaining(); + // ZERO extra copy: the window IS the offered chunk (the A5 borrow, + // scoped to the callback). + aliased = win.buffer === offered.buffer; + got.push(...win); + s.markRead(win.length); + return "done" as const; + }; + let w: Promise, session: Promise; + if (order === "write-first") { + w = hs.writable.write(offered as never); + session = hs.readable.readDirect(consume); + } else { + session = hs.readable.readDirect(consume); + w = hs.writable.write(offered as never); + } + assertEq(await session, 4, `${order}: session total`); + assertEq(await w, 4, `${order}: the chunk write completes`); + assert(aliased, `${order}: the view aliases the offered chunk`); + assertEq(got, [4, 5, 6, 7], `${order}: payload`); + } +}); + +Deno.test("A21 host<->host: two direct sessions cannot rendezvous", async () => { + const hs = hostStream(U8); + let produced = 0; + const w = hs.writable.writeDirect((d) => { + produced++; + d.remaining().set(Uint8Array.from([42])); + d.markWritten(1); + return "more"; + }); + // The ARRIVING side is the one refused. + const e = await caught(hs.readable.readDirect(() => "done")); + assert(e instanceof TypeError, `TypeError, got ${e}`); + assert( + String(e.message).includes("at least one side"), + `names the rule: ${e}`, + ); + assertEq(produced, 0, "the parked producer was never invoked"); + + // The parked session is undisturbed: a chunk read still drives it. + const got = await hs.readable.read(4) as unknown as Uint8Array; + assertEq([...got], [42], "the parked writeDirect still serves chunk reads"); + assertEq(produced, 1); + hs.writable.cancelWrite(); + assertEq(await w, 1); +}); + +// =========================================================================== +// 10. Parity: the seam collapses to the reference copy when nobody is direct +// =========================================================================== + +Deno.test("A21 parity: guest<->guest and chunk paths are unchanged", async () => { + const { memory, view } = mkMemory(); + const cxA = mkCx(view); + const shared = new SharedStreamImpl(U8); + fill(memory, 2100, [1, 2, 3, 4, 5]); + + // guest -> guest, partial: `min(remain, remain)` and both sides notified. + const w = guestOp(shared, "write", cxA, 2100, 5, GUEST_A); + const r = guestOp(shared, "read", cxA, 2200, 3, GUEST_B); + assertEq(memBytes(memory, 2200, 3), [1, 2, 3]); + assertEq(w.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + assertEq(r.events, [{ result: CopyResult.COMPLETED, progress: 3 }]); + + // A non-u8 element type still rendezvouses (the seam is u8-blind for + // chunks; only the DIRECT forms are u8-only). + const s32 = new SharedStreamImpl(U32); + const hs32 = hostStream(U32); + const shared32 = hs32.value as unknown as SharedStreamImpl; + void s32; + const pending = hs32.readable.read(4); + new DataView(memory.buffer).setUint32(2300, 42, true); + guestOp(shared32, "write", cxA, 2300, 1, GUEST_A, U32); + assertEq([...await pending], [42]); +}); diff --git a/runtime/tests/embedder/direct_streams_test.ts b/runtime/tests/embedder/direct_streams_test.ts new file mode 100644 index 0000000..ac884db --- /dev/null +++ b/runtime/tests/embedder/direct_streams_test.ts @@ -0,0 +1,241 @@ +// Direct-access byte edges through the CONVENTIONS facade +// (contracts/embedder-api.md §"Streams and futures", amendment A21, +// 2026-08-22, polyengine#128). +// +// The raw seam and session live in runtime/tests/direct_streams_test.ts; +// this file pins what the *handle* layer adds on top: +// +// * `StreamWriter.writeDirect` parks until the lowering site binds the +// element type, then requires `u8` — the `write` refusal shape; +// * `Stream.readDirect` goes through the same `#require()` gate as `read` +// (unbound refusal, and the A15 post-transfer guard), plus the `u8` check; +// * the A7 interplay: a peer trap rejects the session with +// `PeerTrappedError` carrying the delivered byte count. +// +// Fixture note: `examples/guests/stream-echo` is `stream`, so it cannot +// carry A21 (which is `stream` only). The u8 fixture is +// `examples/guests/stream-pass` — `take` (guest consumes a host-fed +// `stream`), `open-then-trap` (guest produces bytes then traps) and +// `pass-through-text` (a non-u8 element type, for the refusal). No new +// fixtures were built. + +import { assertEq } from "../support/asserts.ts"; +import { caught, guest, haveFixture, instantiateFixture } from "./support.ts"; +import { + type DirectDestination, + type DirectSource, + PeerTrappedError, + Stream, +} from "../../src/embedder/mod.ts"; + +const FIXTURE = guest("stream-pass"); +const ready = await haveFixture(FIXTURE); + +function assert(cond: boolean, msg: string): asserts cond { + if (!cond) throw new Error(`assertion failed: ${msg}`); +} + +Deno.test({ + name: "A21 e2e: StreamWriter.writeDirect feeds a real guest", + ignore: !ready, + fn: async () => { + // `take: async func(input: stream, count: u32) -> u64` reads `count` + // elements and returns their sum. The producer never builds a chunk: it + // writes straight into the guest's landing zone. + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const data = Uint8Array.from([9, 8, 7, 6, 5, 4, 3, 2]); + const expected = BigInt(data[0] + data[1] + data[2]); + + let sent = 0; + let aliasedSomething = false; + // Issued BEFORE the stream is passed to the guest: the writer parks until + // the lowering site binds the element type. + const session = writer.writeDirect((dest: DirectDestination) => { + const win = dest.remaining(); + // The landing zone is guest linear memory, not a runtime scratch. + aliasedSomething ||= win.byteLength > 0; + const k = Math.min(win.length, data.length - sent); + win.set(data.subarray(sent, sent + k)); + dest.markWritten(k); + sent += k; + return sent < data.length ? "more" : "done"; + }); + + assertEq(await c.exports.take(stream, 3), expected, "sum of the first 3"); + const total = await session; + assert(aliasedSomething, "the producer saw a non-empty landing zone"); + assert( + total >= 3 && total <= data.length, + `session total is bounded by the offer (got ${total})`, + ); + }, +}); + +Deno.test({ + name: + "A21 e2e: Stream.readDirect consumes guest output, and a peer trap rejects with the delivered count", + ignore: !ready, + fn: async () => { + // `open-then-trap: async func(n: u32) -> stream` writes n bytes from + // a background task and then traps, so the write end dies in the + // poisoned handle table while our session is parked. Bytes copied BEFORE + // the fault are delivered; the session then rejects rather than faking a + // clean end (amendment A7, inherited by A21). + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const out = await c.exports.openThenTrap(3) as Stream; + const got: number[] = []; + const e = await caught(() => + out.readDirect((src: DirectSource) => { + const win = src.remaining(); + got.push(...win); + src.markRead(win.length); + return "more"; // never volunteers to stop: the trap ends the session + }) + ); + assert(e instanceof PeerTrappedError, `branded: ${e}`); + assertEq(got.length, 3, "bytes written before the trap were delivered"); + assertEq(e.progress, 3, "PeerTrappedError carries the session total"); + }, +}); + +Deno.test({ + name: + "A21: readDirect inherits read's refusals (unbound, and the A15 transfer guard)", + ignore: !ready, + fn: async () => { + // Unbound: `Stream.create()` has no element type until a lowering site + // supplies one. `read`'s refusal, verbatim — unlike the WRITER, which + // parks. + const fresh = Stream.create().stream; + const unbound = await caught(() => fresh.readDirect(() => "done")); + assert(unbound instanceof TypeError, `unbound: TypeError, got ${unbound}`); + assert( + String(unbound.message).includes("has not been " + "passed to a guest"), + `unbound names the reason: ${unbound}`, + ); + + // A15 (#162): once the handle's shared object has been passed to a guest, + // the guest owns the readable end and a host read here would operate a + // phantom duplicate. + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + const e = await caught(() => stream.readDirect(() => "done")); + assert(e instanceof TypeError, `transferred: TypeError, got ${e}`); + assert( + String(e.message).includes("already been passed to a guest"), + `names the A15 guard: ${e}`, + ); + await writer.close(); + out.drop(); + }, +}); + +Deno.test({ + name: "A21: a non-u8 element type is refused on both direct forms", + ignore: !ready, + fn: async () => { + // `pass-through-text` is `stream`: the writer parks until that + // lowering binds the element type, and THEN discovers it is not u8. + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const out = await c.exports.passThroughText(stream) as Stream; + for ( + const [who, e] of [ + ["writeDirect", await caught(() => writer.writeDirect(() => "done"))], + ["readDirect", await caught(() => out.readDirect(() => "done"))], + ] as const + ) { + assert(e instanceof TypeError, `${who}: TypeError, got ${e}`); + assert( + String(e.message).includes("stream only"), + `${who} names A21's scope: ${e}`, + ); + } + await writer.close(); + out.drop(); + }, +}); + +Deno.test({ + name: + "A21: host<->host through a round trip — two direct sessions are refused", + ignore: !ready, + fn: async () => { + // `pass-through` hands the stream straight back (A5 identity), so both + // endpoints end up host-side. A chunk form on either side is fine; two + // direct sessions are not, and the ARRIVING one is what fails. + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + + const session = writer.writeDirect((dest) => { + dest.remaining().set(Uint8Array.from([1, 2, 3])); + dest.markWritten(3); + return "done"; + }); + // Let the producer's session actually park before the consumer arrives: + // `writeDirect` awaits `whenBound()` first, so which side is "arriving" + // is otherwise decided by microtask ordering. + await new Promise((r) => setTimeout(r, 0)); + + const e = await caught(() => out.readDirect(() => "done")); + assert(e instanceof TypeError, `TypeError, got ${e}`); + assert( + String(e.message).includes("at least one side"), + `names the rule: ${e}`, + ); + + // The parked session is undisturbed: the chunk form still drains it. + const got = await out.read(16); + assertEq([...got], [1, 2, 3], "the chunk read completes the session"); + assertEq(await session, 3); + out.drop(); + }, +}); + +Deno.test({ + name: + "A21: host<->host at the same floor — each direct form against the peer chunk form", + ignore: !ready, + fn: async () => { + // Both mixed rows of the host↔host matrix, through the conventions layer + // and across a real boundary round trip (`pass-through`). + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + + // writeDirect vs a chunk read: the marked prefix becomes the chunk. + { + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + const w = writer.writeDirect((dest: DirectDestination) => { + const win = dest.remaining(); + win.set(Uint8Array.from([5, 6, 7, 8]).subarray(0, win.length)); + dest.markWritten(Math.min(4, win.length)); + return "done"; + }); + const got = await out.read(16); + assertEq([...got], [5, 6, 7, 8], "the direct producer's bytes arrive"); + assertEq(await w, 4); + out.drop(); + } + + // readDirect vs a chunk write: the view IS the offered chunk (A5 borrow). + { + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + const w = writer.write(Uint8Array.from([1, 2, 3])); + const got: number[] = []; + const n = await out.readDirect((src: DirectSource) => { + const win = src.remaining(); + got.push(...win); + src.markRead(win.length); + return "done"; + }); + assertEq(n, 3, "the consumer's session total"); + assertEq(got, [1, 2, 3]); + assertEq(await w, 3, "the chunk write completes"); + out.drop(); + } + }, +});