streams: direct-access byte edges for host stream<u8> ends (A21, #128)

contracts/embedder-api.md

@@ -505,12 +505,28 @@ interface Stream<T> {
   readable(): ReadableStream<Chunk<T>>;    // web-native; Chunk<u8> = Uint8Array, else T[]
   [Symbol.asyncIterator](): AsyncIterator<Chunk<T>>;
   read(max: number): Promise<Chunk<T>>;    // low-level; empty chunk = end
+  readDirect(                              // stream<u8> only — amendment A21
+    consume: (src: DirectSource) => "more" | "done",
+  ): Promise<number>;
   cancelRead(): void;
   drop(): void;                            // [Symbol.dispose] alias
 }
 interface Future<T> extends PromiseLike<T> {  // await it directly
   drop(): void; cancel(): void;
 }
+// Direct-access byte edges (amendment A21, stream<u8> only). The writer-side
+// mirror lives on StreamWriter:
+//   writeDirect(produce: (dest: DirectDestination) => "more" | "done"): Promise<number>
+// 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<T>(): { stream: Stream<T>, writer: StreamWriter<T> }`
-  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<u8>` 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<u8>` 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
 

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<u8>` ↔ `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<u8>`
+  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`

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,

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<u8>` only. A
+ * zero-width element type (`t === null`) is not u8 either.
+ */
+function requireU8Direct<T>(codec: ElemCodec<T> | null, who: string): void {
+  if (codec === null || !isU8Element(codec.element)) {
+    throw new TypeError(
+      `${who} is available on stream<u8> only (embedder-api amendment A21, ` +
+        `polyengine#128); use write()/read() for other element types`,
+    );
+  }
+}
+
 /**
  * A stream handle.
  *
@@ -283,6 +308,47 @@ export class Stream<T> {
     return this.#chunk(raw);
   }
 
+  /**
+   * Consume the writer's bytes in place, without an intermediate chunk
+   * (`stream<u8>` 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<number> {
+    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<T> {
     const codec = this.#codec!;
     if (isU8Element(codec.element)) {
@@ -416,6 +482,46 @@ export class StreamWriter<T> {
     return n;
   }
 
+  /**
+   * Fill the reader's landing zone in place, without an intermediate chunk
+   * (`stream<u8>` 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<number> {
+    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<T>): Promise<number> {
     await this.#stream.whenBound();