diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts
index bd0e9d2..37d524a 100644
--- a/runtime/src/embedder/streams.ts
+++ b/runtime/src/embedder/streams.ts
@@ -16,9 +16,11 @@ import {
   type HostFuture,
   hostFuture,
   hostFutureFor,
+  hostFutureReadableState,
   type HostStream,
   hostStream,
   hostStreamFor,
+  hostStreamReadableBusy,
 } from "../exec/host_streams.ts";
 import {
   abandonReasonOf,
@@ -170,6 +172,7 @@ export class Stream<T> implements ProtocolStream<T> {
   #dropped = false;
   /** Waiters parked in `Stream.create()` until an element type is known. */
   #binders: (() => void)[] = [];
+  #boundListeners: (() => void)[] = [];
 
   private constructor(host: HostStream<T> | null, codec: ElemCodec<T> | null) {
     this.#host = host;
@@ -207,24 +210,53 @@ export class Stream<T> implements ProtocolStream<T> {
     this.#codec = codec;
     this.#host = hostStream<T>(codec.element);
     publishHostStream(this, this.#host);
+    if (this.#dropped) this.#host.readable.drop();
     const waiters = this.#binders;
     this.#binders = [];
     for (const w of waiters) w();
+    for (const listener of this.#boundListeners) listener();
   }
 
   /** @internal — resolve once this handle has a shared object. */
   whenBound(): Promise<void> {
-    if (this.#host !== null) return Promise.resolve();
+    if (this.#host !== null || this.#dropped) return Promise.resolve();
     return new Promise<void>((r) => this.#binders.push(r));
   }
 
+  /** @internal — StreamWriter installs one stable lazy-binding hook. */
+  onBound(listener: () => void): void {
+    this.#boundListeners.push(listener);
+  }
+
   /** @internal */
   get bound(): boolean {
     return this.#host !== null;
   }
 
+  /** @internal */
+  get dropped(): boolean {
+    return this.#dropped;
+  }
+
   /** @internal — the shared value to hand to a lowering site. */
   takeValue(codec: ElemCodec<T>): ComponentValue {
+    if (this.#dropped) {
+      throw new TypeError(
+        "this Stream has been dropped and cannot be passed to a guest",
+      );
+    }
+    // CONTRACT: transfer is the host spelling of lift_async_value's IDLE
+    // precondition (definitions.py:1504-1511). Check the cached low-level
+    // wrapper so aliases/round trips cannot bypass it.
+    if (
+      this.#host !== null &&
+      hostStreamReadableBusy(this.#host as HostStream<unknown>)
+    ) {
+      throw new TypeError(
+        "this Stream's readable end has a read in flight; await it or " +
+          "cancelRead() before passing the stream to a guest",
+      );
+    }
     this.bindElement(codec);
     if (this.#consumed) {
       throw new TypeError(
@@ -262,13 +294,42 @@ export class Stream<T> implements ProtocolStream<T> {
   }
 
   /** Low-level read: up to `max` elements; an empty chunk means end-of-stream. */
-  async read(max: number): Promise<Chunk<T>> {
-    const host = this.#require();
+  read(max: number): Promise<Chunk<T>> {
+    let host: HostStream<T>;
+    try {
+      host = this.#require();
+    } catch (e) {
+      return Promise.reject(e);
+    }
+    if (hostStreamReadableBusy(host as HostStream<unknown>)) {
+      throw new TypeError(
+        "a read is already in flight on this stream's readable end; " +
+          "await it or cancelRead() first",
+      );
+    }
     const where = this.#codec?.where ?? "stream read";
-    throwIfFailed(host.value, where);
-    const raw = await host.readable.read(max) as unknown as
-      | ComponentValue[]
-      | Uint8Array;
+    try {
+      throwIfFailed(host.value, where);
+    } catch (e) {
+      return Promise.reject(e);
+    }
+    let pending: Promise<T[]>;
+    try {
+      pending = host.readable.read(max);
+    } catch (e) {
+      // Only the busy exclusion above is synchronously observable. Capacity
+      // and other issuance failures retain the Promise-shaped API.
+      return Promise.reject(e);
+    }
+    return this.#finishRead(pending, host, where);
+  }
+
+  async #finishRead(
+    pending: Promise<T[]>,
+    host: HostStream<T>,
+    where: string,
+  ): Promise<Chunk<T>> {
+    const raw = await pending as unknown as ComponentValue[] | Uint8Array;
     // An empty chunk normally means clean end-of-stream; when the peer's
     // instance trapped it means the retirement walk settled us — reject
     // instead of faking EOS (§"Streams and futures"). A non-empty chunk was really
@@ -299,15 +360,45 @@ export class Stream<T> implements ProtocolStream<T> {
    * already passed to a guest both throw, as does a
    * non-`u8` element type.
    */
-  async readDirect(
+  readDirect(
     consume: (src: DirectSource) => DirectVerdict,
   ): Promise<number> {
-    const host = this.#require();
+    let host: HostStream<T>;
+    try {
+      host = this.#require();
+      requireU8Direct(this.#codec, "readDirect");
+    } catch (e) {
+      return Promise.reject(e);
+    }
+    if (hostStreamReadableBusy(host as HostStream<unknown>)) {
+      throw new TypeError(
+        "a read is already in flight on this stream's readable end; " +
+          "await it or cancelRead() first",
+      );
+    }
     const where = this.#codec?.where ?? "stream read";
-    throwIfFailed(host.value, where);
-    requireU8Direct(this.#codec, "readDirect");
+    try {
+      throwIfFailed(host.value, where);
+    } catch (e) {
+      return Promise.reject(e);
+    }
     const info: DirectSessionInfo = { endedByVerdict: false };
-    const n = await host.readable.readDirect(consume, info);
+    let pending: Promise<number>;
+    try {
+      pending = host.readable.readDirect(consume, info);
+    } catch (e) {
+      return Promise.reject(e);
+    }
+    return this.#finishReadDirect(pending, info, host, where);
+  }
+
+  async #finishReadDirect(
+    pending: Promise<number>,
+    info: DirectSessionInfo,
+    host: HostStream<T>,
+    where: string,
+  ): Promise<number> {
+    const n = await pending;
     // Preserve a callback-completed result. Otherwise report peer poisoning
     // with the acknowledged byte count, not a clean session end.
     if (!info.endedByVerdict) {
@@ -345,9 +436,17 @@ export class Stream<T> implements ProtocolStream<T> {
   drop(): void {
     if (this.#dropped) return;
     this.#dropped = true;
+    const waiters = this.#binders;
+    this.#binders = [];
+    for (const w of waiters) w();
+    for (const listener of this.#boundListeners) listener();
     // Both ends of a host wrapper name the same shared object; dropping once
     // is enough (`SharedStreamImpl.drop` is idempotent).
-    this.#host?.readable.drop();
+    try {
+      this.#host?.readable.drop();
+    } catch {
+      // Pump failures remain recorded; public disposal is total and silent.
+    }
   }
 
   /**
@@ -396,12 +495,23 @@ export class Stream<T> implements ProtocolStream<T> {
 /** How many elements a convenience read asks for at a time. */
 const READ_CHUNK = 4096;
 
+interface WriterOperation {
+  cancelled: boolean;
+  lowStarted: boolean;
+  started?: boolean;
+  run(op: WriterOperation): Promise<number>;
+  resolve(n: number): void;
+  reject(error: unknown): void;
+}
+
 /** Writer half of `Stream.create()`. */
 export class StreamWriter<T> implements ProtocolStreamWriter<T> {
   #stream: Stream<T>;
+  #active: WriterOperation | null = null;
 
   constructor(stream: Stream<T>) {
     this.#stream = stream;
+    stream.onBound(() => this.#launch());
     // realm boundary realm-local pill (see Stream's constructor above for rationale).
     defineRealmLocal(this);
   }
@@ -417,37 +527,93 @@ export class StreamWriter<T> implements ProtocolStreamWriter<T> {
    * window is misuse. Plain-array chunks are lowered (copied) up front.
    */
   write(values: Chunk<T>): Promise<number> {
-    return this.#write(values, false);
+    return this.#start(() => this.#write(values, false));
   }
 
-  async #write(values: Chunk<T>, all: boolean): Promise<number> {
-    await this.#stream.whenBound();
+  #start(run: (op: WriterOperation) => Promise<number>): Promise<number> {
+    if (this.#active !== null) {
+      throw new TypeError(
+        "a write is already in flight on this stream's writable end; " +
+          "await it or cancelWrite() first",
+      );
+    }
+    let resolve!: (n: number) => void;
+    let reject!: (e: unknown) => void;
+    const promise = new Promise<number>((res, rej) => {
+      resolve = res;
+      reject = rej;
+    });
+    const op: WriterOperation = {
+      cancelled: false,
+      lowStarted: false,
+      run,
+      resolve,
+      reject,
+    };
+    this.#active = op;
+    this.#launch();
+    return promise;
+  }
+
+  #launch(): void {
+    const op = this.#active;
+    if (op === null || op.started) return;
+    if (this.#stream.dropped) return this.#resolve(op, 0);
+    if (!this.#stream.bound) return;
+    op.started = true;
+    void op.run(op).then(
+      (n) => this.#resolve(op, n),
+      (e) => this.#reject(op, e),
+    );
+  }
+
+  async #write(
+    values: Chunk<T>,
+    all: boolean,
+  ): Promise<number> {
+    const op = this.#active!;
+    const codec = this.#stream.codec!;
     const host = hostOf(this.#stream);
-    const where = this.#stream.codec?.where ?? "stream write";
+    const where = codec.where ?? "stream write";
     throwIfFailed(host.value, where);
-    const codec = this.#stream.codec!;
     const lowered = packChunk(values, codec);
+    if (op.cancelled) {
+      releaseUntaken(lowered, 0, codec);
+      return 0;
+    }
     const info = codec.release === undefined ? undefined : { progress: 0 };
+    op.lowStarted = true;
     let n: number;
     try {
       n = await host.writable[all ? "writeAll" : "write"](
         lowered as unknown as T[],
         info,
       );
-      // Full takes completed before a later peer fault and keep their result.
       if (n < values.length) throwIfPeerTrapped(host.value, where, n);
     } catch (e) {
       try {
         releaseUntaken(lowered, info?.progress ?? 0, codec);
       } catch {
-        // Cleanup attempted every tail element; preserve the write failure.
+        // Preserve the operation failure after attempting every release.
       }
       throw e;
     }
+    // Keep successful cleanup outside the operation catch: if it throws,
+    // each untaken resource has still been attempted exactly once.
     releaseUntaken(lowered, n, codec);
     return n;
   }
 
+  #resolve(op: WriterOperation, n: number): void {
+    if (this.#active === op) this.#active = null;
+    op.resolve(n);
+  }
+
+  #reject(op: WriterOperation, error: unknown): void {
+    if (this.#active === op) this.#active = null;
+    op.reject(error);
+  }
+
   /**
    * Fill the reader's landing zone in place, without an intermediate chunk
    * (`stream<u8>` only — contracts/embedder-api.md §"Streams and futures" ("Direct-access byte edges"),
@@ -471,36 +637,40 @@ export class StreamWriter<T> implements ProtocolStreamWriter<T> {
    * `Stream.create()` writer has no element type until the lowering site
    * binds one — and then requires `u8`.
    */
-  async writeDirect(
+  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);
-    // Callback completion survives a later peer trap; other endings report
-    // poisoning with the acknowledged byte count.
-    if (!info.endedByVerdict) throwIfPeerTrapped(host.value, where, n);
-    return n;
+    return this.#start(async (op) => {
+      const host = hostOf(this.#stream);
+      const where = this.#stream.codec?.where ?? "stream write";
+      throwIfFailed(host.value, where);
+      requireU8Direct(this.#stream.codec, "writeDirect");
+      if (op.cancelled) return 0;
+      const info: DirectSessionInfo = { endedByVerdict: false };
+      op.lowStarted = true;
+      const n = await host.writable.writeDirect(produce, info);
+      if (!info.endedByVerdict) throwIfPeerTrapped(host.value, where, n);
+      return n;
+    });
   }
 
   /** Offer values until all are taken or the reader goes away. */
   writeAll(values: Chunk<T>): Promise<number> {
-    return this.#write(values, true);
+    return this.#start(() => this.#write(values, true));
   }
 
   cancelWrite(): void {
-    if (!this.#stream.bound) return;
-    hostOf(this.#stream).writable.cancelWrite();
+    const op = this.#active;
+    if (op === null) return;
+    op.cancelled = true;
+    if (op.lowStarted) hostOf(this.#stream).writable.cancelWrite();
+    else if (!op.started) this.#resolve(op, 0);
   }
 
   /** End-of-stream. */
   async close(): Promise<void> {
     await this.#stream.whenBound();
-    hostOf(this.#stream).writable.drop();
+    if (this.#stream.bound) hostOf(this.#stream).writable.drop();
   }
 }
 
@@ -537,6 +707,12 @@ export class Future<T> implements ProtocolFuture<T> {
   #consumed = false;
   #dropped = false;
   #settled: Promise<T> | null = null;
+  /** Low-level read installed synchronously during deferred adoption. */
+  #adoptedRead:
+    | Promise<Awaited<ReturnType<HostFuture<T>["readResult"]>>>
+    | null = null;
+  /** Await has started but its low-level read may still be behind #hostP. */
+  #reading = false;
 
   private constructor(
     host: HostFuture<T> | null,
@@ -592,6 +768,13 @@ export class Future<T> implements ProtocolFuture<T> {
   /** @internal */
   adopt(h: HostFuture<T>): void {
     this.#host = h;
+    if (this.#reading && this.#adoptedRead === null) {
+      try {
+        this.#adoptedRead = h.readResult();
+      } catch (e) {
+        this.#adoptedRead = Promise.reject(e);
+      }
+    }
   }
 
   /** @internal */
@@ -606,6 +789,22 @@ export class Future<T> implements ProtocolFuture<T> {
         "this Future handle has already been passed to a guest",
       );
     }
+    const state = hostFutureReadableState(
+      this.#host as HostFuture<unknown>,
+    );
+    if (this.#reading) {
+      throw new TypeError(
+        "this Future's readable end has an operation in flight; await or " +
+          "cancel it before passing the future to a guest",
+      );
+    }
+    if (state !== "idle") {
+      throw new TypeError(
+        state === "busy"
+          ? "this Future's readable end has an operation in flight; await or cancel it before passing the future to a guest"
+          : "this Future's value has already been consumed and cannot be passed to a guest again",
+      );
+    }
     this.#consumed = true;
     return this.#host.value;
   }
@@ -622,24 +821,35 @@ export class Future<T> implements ProtocolFuture<T> {
         ),
       );
     }
-    this.#settled ??= (async () => {
-      const host = await this.#hostP;
-      let outcome: Awaited<ReturnType<HostFuture<T>["readResult"]>>;
+    if (this.#settled !== null) return this.#settled;
+    // Reserve before awaiting deferred materialization. takeValue() observes
+    // this initiation window even when #hostP has just adopted a host end.
+    this.#reading = true;
+    if (this.#host !== null) {
+      // Do not cross a microtask for an already materialized future: aliases
+      // consult the shared wrapper's busy state synchronously.
+      const host = this.#host;
       try {
-        outcome = await host.readResult();
+        this.#settled = this.#finishRead(host, host.readResult());
       } catch (e) {
-        // Only replace the trap manufactured by producer-failure abandonment.
-        // A distinct read/pump/busy failure remains primary even if the
-        // producer failure happens to be recorded before this continuation.
-        const failure = producerFailures.get(host.value as object);
-        if (
-          failure !== undefined && abandonReasonOf(host.value) === failure &&
-          typeof e === "object" && e !== null &&
-          (e as { cause?: unknown }).cause === failure
-        ) throw failure;
+        this.#reading = false;
         throw e;
       }
-      const { value, result } = outcome;
+    } else {
+      this.#settled = this.#hostP.then((host) =>
+        this.#finishRead(host, this.#adoptedRead ?? host.readResult())
+      );
+    }
+    this.#settled = this.#settled.finally(() => this.#reading = false);
+    return this.#settled;
+  }
+
+  async #finishRead(
+    host: HostFuture<T>,
+    pending: Promise<Awaited<ReturnType<HostFuture<T>["readResult"]>>>,
+  ): Promise<T> {
+    try {
+      const { value, result } = await pending;
       if (result !== CopyResult.COMPLETED) {
         // Producer failure and peer poisoning both outrank the ordinary
         // cancelled/dropped outcome; throwIfFailed checks them in that order.
@@ -651,8 +861,16 @@ export class Future<T> implements ProtocolFuture<T> {
         );
       }
       return this.#codec.toHost(value as ComponentValue);
-    })();
-    return this.#settled;
+    } catch (e) {
+      // Only replace the trap manufactured by producer-failure abandonment.
+      const failure = producerFailures.get(host.value as object);
+      if (
+        failure !== undefined && abandonReasonOf(host.value) === failure &&
+        typeof e === "object" && e !== null &&
+        (e as { cause?: unknown }).cause === failure
+      ) throw failure;
+      throw e;
+    }
   }
 
   then<R1 = T, R2 = never>(
diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts
index 0d65f39..f878ede 100644
--- a/runtime/src/exec/host_streams.ts
+++ b/runtime/src/exec/host_streams.ts
@@ -817,7 +817,11 @@ function bindOnLower(
 function mkStreamEnds<T>(
   shared: SharedStreamImpl,
   activity: HostActivity,
-): { readable: HostReadableEnd<T>; writable: HostWritableEnd<T> } {
+): {
+  readable: HostReadableEnd<T>;
+  writable: HostWritableEnd<T>;
+  readableBusy: () => boolean;
+} {
   // Distinct rendezvous identities per end — see `hostEndInstance`.
   const writeInst = hostEndInstance("write");
   const readInst = hostEndInstance("read");
@@ -966,6 +970,7 @@ function mkStreamEnds<T>(
     });
   };
   return {
+    readableBusy: () => parked.read,
     writable: {
       write(values: T[], info?: { progress: number }): Promise<number> {
         // One operation per direction prevents write-against-write rendezvous.
@@ -1135,14 +1140,42 @@ function mkStreamEnds<T>(
  */
 const streamWrappers = new WeakMap<object, HostStream<unknown>>();
 const futureWrappers = new WeakMap<object, HostFuture<unknown>>();
+const streamReadableStates = new WeakMap<HostStream<unknown>, () => boolean>();
+const futureReadableStates = new WeakMap<
+  HostFuture<unknown>,
+  () => "idle" | "busy" | "done"
+>();
+
+/** @internal — facade transfer exclusion without widening the end interfaces. */
+export function hostStreamReadableBusy(host: HostStream<unknown>): boolean {
+  return streamReadableStates.get(host)?.() ?? false;
+}
+
+/** @internal — facade transfer/single-consumption state. */
+export function hostFutureReadableState(
+  host: HostFuture<unknown>,
+): "idle" | "busy" | "done" {
+  return futureReadableStates.get(host)?.() ?? "idle";
+}
 
 /** Create a host-owned stream of `element` (`null` = zero-width payload). */
 export function hostStream<T>(element: ValType | null): HostStream<T> {
   const shared = new SharedStreamImpl(element);
   const activity = new HostActivity();
   bindOnLower(shared, activity, "created");
-  const ends = mkStreamEnds<T>(shared, activity);
-  const wrapper = { ...ends, value: shared as unknown as ComponentValue };
+  const { readable, writable, readableBusy } = mkStreamEnds<T>(
+    shared,
+    activity,
+  );
+  const wrapper = {
+    readable,
+    writable,
+    value: shared as unknown as ComponentValue,
+  };
+  streamReadableStates.set(
+    wrapper as HostStream<unknown>,
+    readableBusy,
+  );
   streamWrappers.set(shared, wrapper as HostStream<unknown>);
   return wrapper;
 }
@@ -1162,8 +1195,15 @@ export function hostStreamFor<T>(value: ComponentValue): HostStream<T> {
   if (cached !== undefined) return cached as HostStream<T>;
   const activity = new HostActivity();
   bindOnLower(shared, activity, "lifted");
-  const ends = mkStreamEnds<T>(shared, activity);
-  const wrapper = { ...ends, value };
+  const { readable, writable, readableBusy } = mkStreamEnds<T>(
+    shared,
+    activity,
+  );
+  const wrapper = { readable, writable, value };
+  streamReadableStates.set(
+    wrapper as HostStream<unknown>,
+    readableBusy,
+  );
   streamWrappers.set(shared, wrapper as HostStream<unknown>);
   return wrapper;
 }
@@ -1310,6 +1350,11 @@ function mkFuture<T>(
             "await it or cancel() first",
         );
       }
+      if (delivered) {
+        throw new TypeError(
+          "this future's single value has already been consumed",
+        );
+      }
       // definitions.py `SharedFutureImpl.read` asserts `not self.dropped`, so
       // a read after the write end went away must be answered here rather
       // than by tripping an internal assertion.
@@ -1382,6 +1427,10 @@ function mkFuture<T>(
     },
     value,
   };
+  futureReadableStates.set(
+    self as HostFuture<unknown>,
+    () => parked.read ? "busy" : delivered ? "done" : "idle",
+  );
   return self;
 }
 
diff --git a/runtime/tests/embedder/busy-read.wasm b/runtime/tests/embedder/busy-read.wasm
new file mode 100644
index 0000000..098cbd9
Binary files /dev/null and b/runtime/tests/embedder/busy-read.wasm differ
diff --git a/runtime/tests/embedder/busy-read.wat b/runtime/tests/embedder/busy-read.wat
new file mode 100644
index 0000000..8ae67aa
--- /dev/null
+++ b/runtime/tests/embedder/busy-read.wat
@@ -0,0 +1,36 @@
+;; A receiving guest which performs a real canonical stream/future read.
+;; Transfer must be refused before these exports are entered when the shared
+;; readable end is busy.
+(component
+  (type $s (stream u8))
+  (type $f (future u8))
+  (core module $mem (memory (export "memory") 1))
+  (core instance $memi (instantiate $mem))
+  (canon stream.read $s (memory $memi "memory") async (core func $sr))
+  (canon future.read $f (memory $memi "memory") async (core func $fr))
+  (core module $m
+    (import "" "memory" (memory 1))
+    (import "" "sr" (func $sr (param i32 i32 i32) (result i32)))
+    (import "" "fr" (func $fr (param i32 i32) (result i32)))
+    (func (export "pass") (param i32) (result i32) local.get 0)
+    (func (export "ping") (result i32) i32.const 42)
+    (func (export "stream-read") (param i32) (result i32)
+      (drop (call $sr (local.get 0) (i32.const 0) (i32.const 1)))
+      (i32.const 0))
+    (func (export "future-read") (param i32) (result i32)
+      (drop (call $fr (local.get 0) (i32.const 0)))
+      (i32.const 0)))
+  (core instance $i (instantiate $m (with "" (instance
+    (export "memory" (memory $memi "memory"))
+    (export "sr" (func $sr))
+    (export "fr" (func $fr))))))
+  (func (export "pass-stream") (param "s" $s) (result $s)
+    (canon lift (core func $i "pass")))
+  (func (export "pass-future") (param "f" $f) (result $f)
+    (canon lift (core func $i "pass")))
+  (func (export "read-stream") (param "s" $s) (result u32)
+    (canon lift (core func $i "stream-read")))
+  (func (export "read-future") (param "f" $f) (result u32)
+    (canon lift (core func $i "future-read")))
+  (func (export "ping") (result u32)
+    (canon lift (core func $i "ping"))))
diff --git a/runtime/tests/embedder/passthrough_test.ts b/runtime/tests/embedder/passthrough_test.ts
index 513a177..6cec5df 100644
--- a/runtime/tests/embedder/passthrough_test.ts
+++ b/runtime/tests/embedder/passthrough_test.ts
@@ -348,13 +348,12 @@ Deno.test({
 
 Deno.test({
   name:
-    "deadlock-verdict suppression: a Future read memoized BEFORE the transfer still resolves",
+    "deadlock-verdict suppression: a Future read memoized before transfer preserves the read on refusal",
   ignore: false,
   fn: async () => {
-    // The read genuinely happened while the host owned the end; only reads
-    // STARTED after the transfer are refused. Modelled on a LIFTED future
-    // (the host holds the readable end) with a guest-shaped write completing
-    // the rendezvous.
+    // The read genuinely started while the host owned the end. Transfer is
+    // refused while it is busy, but the memoized read remains live and may be
+    // observed repeatedly after its guest-shaped peer write completes.
     const codec = {
       element: { kind: "u32" } as ValType,
       toHost: (v: ComponentValue) => v as number,
@@ -369,7 +368,13 @@ Deno.test({
     // `Promise.resolve(f)` would adopt the thenable a microtask later, i.e.
     // after `takeValue()`.
     const pending = new Promise<number>((res, rej) => f.then(res, rej));
-    f.takeValue();
+    let transferError: unknown;
+    try {
+      f.takeValue();
+    } catch (e) {
+      transferError = e;
+    }
+    assertEq(transferError instanceof TypeError, true);
 
     // The guest delivers the value into the parked read.
     let progress = 0;
@@ -387,7 +392,8 @@ Deno.test({
       src as never,
       () => {},
     );
-    assertEq(await pending, 7, "the pre-transfer read resolves");
+    assertEq(await pending, 7);
+    assertEq(await f, 7, "the completed read remains memoized");
   },
 });
 
diff --git a/runtime/tests/embedder/producer_failure_test.ts b/runtime/tests/embedder/producer_failure_test.ts
index c6938ff..ccbdd36 100644
--- a/runtime/tests/embedder/producer_failure_test.ts
+++ b/runtime/tests/embedder/producer_failure_test.ts
@@ -214,10 +214,9 @@ Deno.test("producer failure: a COMPLETED future payload wins a later pump failur
   const error = producerFailure(store.hostFailure);
   release.resolve();
   assertEq(await completed, 7);
-  assertEq(
-    await rejected(Future.fromLifted<number>(asValue(raw), codec)),
-    error,
-  );
+  const reread = await rejected(Future.fromLifted<number>(asValue(raw), codec));
+  assertEq(reread instanceof TypeError, true, `single-use reread: ${reread}`);
+  assertEq(store.hostFailure, error, "the later producer failure is preserved");
 });
 
 Deno.test("producer failure: pending read, iterator, and readable reject", async () => {
diff --git a/runtime/tests/embedder/stream_lifecycle_test.ts b/runtime/tests/embedder/stream_lifecycle_test.ts
new file mode 100644
index 0000000..a245711
--- /dev/null
+++ b/runtime/tests/embedder/stream_lifecycle_test.ts
@@ -0,0 +1,399 @@
+// Regression coverage for facade lifecycle/exclusion (#324, #326, #327,
+// #331). The shared wrapper is the authority for alias exclusion; facade
+// reservation additionally covers work queued before lazy binding.
+
+import { assertEq } from "../support/asserts.ts";
+import { Future, Stream } from "../../src/embedder/streams.ts";
+import {
+  hostFuture,
+  hostStream,
+  hostStreamFor,
+} from "../../src/exec/host_streams.ts";
+import type { DirectDestination } from "@polyengine/protocol";
+import { instantiate } from "../../src/embedder/mod.ts";
+import { artifactsOf, caught, haveFixture } from "./support.ts";
+
+const u8 = {
+  element: { kind: "u8" } as const,
+  toHost: (v: unknown) => v as number,
+  fromHost: (v: number) => v,
+};
+
+function throwsTypeError(f: () => unknown): boolean {
+  try {
+    f();
+    return false;
+  } catch (e) {
+    return e instanceof TypeError;
+  }
+}
+
+for (const kind of ["write", "writeAll", "writeDirect"] as const) {
+  Deno.test(`${kind}: reserves synchronously while binding is pending`, async () => {
+    const { stream, writer } = Stream.create<number>();
+    const first = kind === "write"
+      ? writer.write(new Uint8Array([1]))
+      : kind === "writeAll"
+      ? writer.writeAll(new Uint8Array([1]))
+      : writer.writeDirect((_dest: DirectDestination) => "done");
+    assertEq(
+      throwsTypeError(() => writer.write(new Uint8Array([2]))),
+      true,
+    );
+    writer.cancelWrite();
+    assertEq(await first, 0, "pre-bind cancellation settles promptly");
+
+    stream.bindElement(u8);
+    const retry = writer.write(new Uint8Array([3]));
+    assertEq([...(await stream.read(1))], [3]);
+    assertEq(await retry, 1, "old continuation cannot affect reuse");
+    stream.drop();
+  });
+}
+
+for (const kind of ["write", "writeAll", "writeDirect"] as const) {
+  Deno.test(`${kind}: immediate cancellation on an already-bound stream`, async () => {
+    const { stream, writer } = Stream.create<number>();
+    stream.bindElement(u8);
+    const pending = kind === "write"
+      ? writer.write(new Uint8Array([1]))
+      : kind === "writeAll"
+      ? writer.writeAll(new Uint8Array([1]))
+      : writer.writeDirect((dest) => {
+        dest.markWritten(1);
+        return "done";
+      });
+    writer.cancelWrite();
+    assertEq(await pending, 0);
+    const retry = writer.write(new Uint8Array([2]));
+    assertEq([...(await stream.read(1))], [2]);
+    assertEq(await retry, 1);
+    stream.drop();
+  });
+}
+
+Deno.test("writer modes contend pairwise synchronously", async () => {
+  const starts = [
+    (w: ReturnType<typeof Stream.create<number>>["writer"]) =>
+      w.write(new Uint8Array([1])),
+    (w: ReturnType<typeof Stream.create<number>>["writer"]) =>
+      w.writeAll(new Uint8Array([1])),
+    (w: ReturnType<typeof Stream.create<number>>["writer"]) =>
+      w.writeDirect(() => "done"),
+  ];
+  for (const first of starts) {
+    for (const second of starts) {
+      const { stream, writer } = Stream.create<number>();
+      stream.bindElement(u8);
+      const pending = first(writer);
+      assertEq(throwsTypeError(() => second(writer)), true);
+      writer.cancelWrite();
+      await pending;
+      stream.drop();
+    }
+  }
+});
+
+Deno.test("successful resource cleanup runs once even when release throws", async () => {
+  const released: number[] = [];
+  const codec = {
+    element: { kind: "u32" } as const,
+    toHost: (v: unknown) => v as number,
+    fromHost: (v: number) => v,
+    release: (v: unknown) => {
+      released.push(v as number);
+      throw new Error("release failed");
+    },
+  };
+  const { stream, writer } = Stream.create<number>();
+  stream.bindElement(codec);
+  const read = stream.read(1);
+  const error = await caught(() => writer.write([1, 2, 3]));
+  assertEq(await read, [1]);
+  assertEq(String(error).includes("release failed"), true);
+  assertEq(released, [2, 3], "each untaken resource is attempted once");
+  stream.drop();
+});
+
+Deno.test("cancel during conversion preserves conversion and cleanup failures", async () => {
+  for (const thrown of [undefined, new Error("conversion failed")]) {
+    let releases = 0;
+    const pair = Stream.create<number>();
+    const writer = pair.writer;
+    const codec = {
+      element: { kind: "u32" } as const,
+      toHost: (v: unknown) => v as number,
+      fromHost: (v: number) => {
+        if (v === 2) {
+          writer.cancelWrite();
+          throw thrown;
+        }
+        return v;
+      },
+      release: () => {
+        releases++;
+        throw new Error("cleanup failed");
+      },
+    };
+    pair.stream.bindElement(codec);
+    let rejected = false;
+    let reason: unknown;
+    await writer.write([1, 2]).then(() => {}, (e) => {
+      rejected = true;
+      reason = e;
+    });
+    assertEq(rejected, true);
+    assertEq(reason, thrown);
+    assertEq(releases, 1, "lowered prefix cleanup is attempted once");
+    pair.stream.drop();
+  }
+});
+
+Deno.test("arbitrary undefined write failure remains a rejection", async () => {
+  const codec = {
+    element: { kind: "u32" } as const,
+    toHost: (v: unknown) => v as number,
+    fromHost: (_v: number): never => {
+      throw undefined;
+    },
+  };
+  const { stream, writer } = Stream.create<number>();
+  stream.bindElement(codec);
+  let fulfilled = false;
+  let rejected = false;
+  let reason: unknown;
+  await writer.write([1]).then(
+    () => fulfilled = true,
+    (e) => {
+      rejected = true;
+      reason = e;
+    },
+  );
+  assertEq(fulfilled, false);
+  assertEq(rejected, true);
+  assertEq(reason, undefined);
+  stream.drop();
+});
+
+Deno.test("read/readDirect busy refusal is synchronous and cancellation permits reuse", async () => {
+  const { stream, writer } = Stream.create<number>();
+  stream.bindElement(u8);
+  const read = stream.read(1);
+  assertEq(throwsTypeError(() => stream.read(1)), true);
+  assertEq(throwsTypeError(() => stream.readDirect(() => "done")), true);
+  stream.cancelRead();
+  assertEq([...(await read)], []);
+
+  const direct = stream.readDirect(() => "done");
+  assertEq(throwsTypeError(() => stream.read(1)), true);
+  stream.cancelRead();
+  assertEq(await direct, 0);
+
+  const write = writer.write(new Uint8Array([4]));
+  assertEq(
+    [...(await stream.read(1))],
+    [4],
+    "opposite directions remain legal",
+  );
+  assertEq(await write, 1);
+  stream.drop();
+});
+
+Deno.test("invalid read capacity rejects, while busy takes synchronous precedence", async () => {
+  const { stream } = Stream.create<number>();
+  stream.bindElement(u8);
+
+  let invalid: Promise<unknown> | undefined;
+  assertEq(
+    throwsTypeError(() => invalid = stream.read(-1)),
+    false,
+    "idle capacity failure is not thrown synchronously",
+  );
+  const error = await caught(() => invalid!);
+  assertEq(error instanceof RangeError, true, `expected RangeError: ${error}`);
+
+  const active = stream.read(1);
+  assertEq(
+    throwsTypeError(() => stream.read(-1)),
+    true,
+    "busy exclusion wins before capacity validation",
+  );
+  stream.cancelRead();
+  await active;
+  stream.drop();
+});
+
+Deno.test("busy transfer checks the shared wrapper across aliases", async () => {
+  const host = hostFuture<number>(u8.element);
+  const first = Future.fromHostFuture(host, u8);
+  const alias = Future.fromLifted<number>(host.value, u8);
+  const read = Promise.resolve(first.then((v) => v));
+  assertEq(throwsTypeError(() => alias.takeValue()), true);
+  first.cancel();
+  await read.catch(() => {});
+  first.drop();
+
+  const hs = hostStreamFor<number>(
+    new (await import("../../src/task/mod.ts"))
+      .SharedStreamImpl(u8.element) as never,
+  );
+  const a = Stream.fromHostStream(hs, u8);
+  const b = Stream.fromLifted<number>(hs.value, u8);
+  const pending = a.read(1);
+  assertEq(throwsTypeError(() => b.takeValue(u8)), true);
+  a.cancelRead();
+  assertEq([...(await pending)], []);
+  a.drop();
+});
+
+Deno.test("direct read reservation excludes transfer across a more gap", async () => {
+  const host = hostStream<number>(u8.element);
+  const stream = Stream.fromHostStream(host, u8);
+  const alias = Stream.fromLifted<number>(host.value, u8);
+  const firstWrite = host.writable.write([1]);
+  const direct = stream.readDirect((src) => {
+    src.markRead(1);
+    return "more";
+  });
+  await firstWrite;
+  assertEq(throwsTypeError(() => alias.takeValue(u8)), true);
+  assertEq(throwsTypeError(() => stream.takeValue(u8)), true);
+  const secondWrite = host.writable.write([2]);
+  await secondWrite;
+  stream.cancelRead();
+  assertEq(await direct, 2, "the refused transfer preserves the session");
+  stream.drop();
+});
+
+Deno.test("completed Future payload cannot be transferred, while awaits stay memoized", async () => {
+  const host = hostFuture<number>(u8.element);
+  const future = Future.fromHostFuture(host, u8);
+  const writing = host.write(7);
+  assertEq(await future, 7);
+  await writing;
+  assertEq(await future, 7);
+  assertEq(throwsTypeError(() => future.takeValue()), true);
+  future.drop();
+});
+
+Deno.test("deferred Future reserves before host adoption", async () => {
+  let adopt!: (value: unknown) => void;
+  const host = hostFuture<number>(u8.element);
+  const future = Future.deferred<number>(
+    new Promise((resolve) => adopt = resolve) as never,
+    u8,
+  );
+  const pending = future.then((v) => v);
+  adopt(host.value);
+  await Promise.resolve();
+  const alias = Future.fromLifted<number>(host.value, u8);
+  assertEq(throwsTypeError(() => alias.takeValue()), true);
+  const write = host.write(8);
+  assertEq(await pending, 8);
+  await write;
+  future.drop();
+});
+
+Deno.test("completed stream read permits a later transfer", async () => {
+  const host = hostStream<number>(u8.element);
+  const stream = Stream.fromHostStream(host, u8);
+  const write = host.writable.write([5]);
+  assertEq([...(await stream.read(1))], [5]);
+  await write;
+  assertEq(stream.takeValue(u8) === host.value, true);
+});
+
+Deno.test("drop before binding settles queued work and cannot resurrect", async () => {
+  const { stream, writer } = Stream.create<number>();
+  const pending = writer.write(new Uint8Array([8]));
+  stream.drop();
+  stream.drop();
+  assertEq(await pending, 0);
+  assertEq(throwsTypeError(() => stream.takeValue(u8)), true);
+  await writer.close();
+});
+
+Deno.test("drop before any write and immediately after binding is terminal", async () => {
+  const fresh = Stream.create<number>();
+  fresh.stream.drop();
+  assertEq(throwsTypeError(() => fresh.stream.takeValue(u8)), true);
+  assertEq(await fresh.writer.close(), undefined);
+
+  const bound = Stream.create<number>();
+  bound.stream.bindElement(u8);
+  bound.stream.drop();
+  assertEq(await bound.writer.write(new Uint8Array([1])), 0);
+});
+
+const wasmReady = await haveFixture("runtime/tests/embedder/busy-read.wasm");
+
+Deno.test({
+  name: "busy stream transfer is refused before real Wasm entry",
+  ignore: !wasmReady,
+  fn: async () => {
+    const c = await instantiate(
+      await artifactsOf("runtime/tests/embedder/busy-read.wasm"),
+      {},
+      { jspi: false },
+    );
+    const { stream, writer } = Stream.create<number>();
+    const lifted = await c.exports.passStream(stream) as Stream<number>;
+    const active = lifted.read(1);
+    const e = await caught(() => c.exports.readStream(lifted));
+    assertEq(e instanceof TypeError, true, `expected transfer refusal: ${e}`);
+    assertEq(await c.exports.ping(), 42, "guest was not poisoned");
+    const write = writer.write(new Uint8Array([9]));
+    assertEq([...(await active)], [9], "active read remains intact");
+    assertEq(await write, 1);
+    lifted.drop();
+  },
+});
+
+Deno.test({
+  name: "busy direct stream transfer is refused before real Wasm entry",
+  ignore: !wasmReady,
+  fn: async () => {
+    const c = await instantiate(
+      await artifactsOf("runtime/tests/embedder/busy-read.wasm"),
+      {},
+      { jspi: false },
+    );
+    const { stream, writer } = Stream.create<number>();
+    const lifted = await c.exports.passStream(stream) as Stream<number>;
+    const active = lifted.readDirect((src) => {
+      src.markRead(src.remaining().length);
+      return "done";
+    });
+    const e = await caught(() => c.exports.readStream(lifted));
+    assertEq(e instanceof TypeError, true, `expected transfer refusal: ${e}`);
+    const write = writer.write(new Uint8Array([6]));
+    assertEq(await active, 1);
+    assertEq(await write, 1);
+    assertEq(await c.exports.ping(), 42);
+    lifted.drop();
+  },
+});
+
+Deno.test({
+  name: "busy future transfer is refused before real Wasm entry",
+  ignore: !wasmReady,
+  fn: async () => {
+    const c = await instantiate(
+      await artifactsOf("runtime/tests/embedder/busy-read.wasm"),
+      {},
+      { jspi: false },
+    );
+    const host = hostFuture<number>(u8.element);
+    const lifted = c.exports.passFuture(
+      Future.fromHostFuture(host, u8),
+    ) as Future<number>;
+    const active = lifted.then((v) => v);
+    const e = await caught(() => c.exports.readFuture(lifted));
+    assertEq(e instanceof TypeError, true, `expected transfer refusal: ${e}`);
+    const write = host.write(7);
+    assertEq(await active, 7);
+    await write;
+    assertEq(await c.exports.ping(), 42);
+    lifted.drop();
+  },
+});