diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts
index 46c3db0..bd0e9d2 100644
--- a/runtime/src/embedder/streams.ts
+++ b/runtime/src/embedder/streams.ts
@@ -21,6 +21,7 @@ import {
   hostStreamFor,
 } from "../exec/host_streams.ts";
 import {
+  abandonReasonOf,
   CopyResult,
   dropSharedForTeardown,
   type ErrorContext as InternalErrorContext,
@@ -80,7 +81,7 @@ const producerFailures = new WeakMap<object, StreamProducerError>();
 
 /**
  * Record on the handle and, if bound, the store's first-failure slot. Return
- * whether a store exists. Report before dropping the writer: driveAsync
+ * the site-named error. Report before dropping the writer: driveAsync
  * checks hostFailure before completion, so truncation cannot hide the cause.
  * The slot is store-wide, not an attribution to a particular consuming call.
  */
@@ -88,7 +89,7 @@ function reportProducerFailure(
   host: HostStream<unknown>,
   where: string,
   cause: unknown,
-): boolean {
+): StreamProducerError {
   // Brand, not class: a producer failure raised by another runtime copy
   // must not be re-wrapped into a second layer of the same error.
   const err = isStreamProducerError(cause)
@@ -101,9 +102,8 @@ function reportProducerFailure(
   const store = shared.boundStore;
   if (store != null && typeof store === "object") {
     if (store.hostFailure === undefined) store.hostFailure = err;
-    return true;
   }
-  return false;
+  return err;
 }
 
 /** @internal — raise a recorded producer failure, if any. */
@@ -273,7 +273,7 @@ export class Stream<T> implements ProtocolStream<T> {
     // instance trapped it means the retirement walk settled us — reject
     // instead of faking EOS (§"Streams and futures"). A non-empty chunk was really
     // copied before the trap and is delivered; the next read rejects.
-    if (raw.length === 0) throwIfPeerTrapped(host.value, where);
+    if (raw.length === 0) throwIfFailed(host.value, where);
     return this.#chunk(raw);
   }
 
@@ -310,7 +310,11 @@ export class Stream<T> implements ProtocolStream<T> {
     const n = await host.readable.readDirect(consume, info);
     // Preserve a callback-completed result. Otherwise report peer poisoning
     // with the acknowledged byte count, not a clean session end.
-    if (!info.endedByVerdict) throwIfPeerTrapped(host.value, where, n);
+    if (!info.endedByVerdict) {
+      const failure = producerFailures.get(host.value as object);
+      if (failure !== undefined) throw failure;
+      throwIfPeerTrapped(host.value, where, n);
+    }
     return n;
   }
 
@@ -620,11 +624,26 @@ export class Future<T> implements ProtocolFuture<T> {
     }
     this.#settled ??= (async () => {
       const host = await this.#hostP;
-      const { value, result } = await host.readResult();
+      let outcome: Awaited<ReturnType<HostFuture<T>["readResult"]>>;
+      try {
+        outcome = await 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;
+        throw e;
+      }
+      const { value, result } = outcome;
       if (result !== CopyResult.COMPLETED) {
-        // A drop caused by the writer's instance trapping is a fault, not a
-        // "no value" outcome — brand it (#66, §"Streams and futures").
-        throwIfPeerTrapped(host.value, this.#codec.where ?? "future read");
+        // Producer failure and peer poisoning both outrank the ordinary
+        // cancelled/dropped outcome; throwIfFailed checks them in that order.
+        throwIfFailed(host.value, this.#codec.where ?? "future read");
         throw new DroppedError(
           result === CopyResult.CANCELLED
             ? "the future read was cancelled"
@@ -985,14 +1004,22 @@ export function lowerFutureSource<T>(
       if (info?.progress === 0) codec.release?.(lowered);
     } catch (e) {
       // Report the producer cause rather than replace it with a generic
-      // abandonment trap. A bound store receives the failure; only an unbound
-      // future is dropped here. Reporting does not itself retire the future.
-      const reported = reportProducerFailure(
+      // abandonment trap. Reporting itself does not retire the future.
+      const failure = reportProducerFailure(
         { value: host.value } as unknown as HostStream<unknown>,
         codec.where ?? "future producer",
         e,
       );
-      if (!reported) host.drop();
+      try {
+        // CONTRACT: an unwritten bound future is abandoned, not completed
+        // DROPPED-shaped (embedder-api.md §"Streams and futures";
+        // definitions.py `WritableFutureEnd.drop`).
+        // Retire it with the producer error itself so pending guest and host
+        // readers wake with the same cause, and drop observers release activity.
+        host.fail(failure);
+      } catch {
+        // Reporting happened first; cleanup must not replace the producer fault.
+      }
     }
   })();
   return host.value;
diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts
index 0cda67d..0d65f39 100644
--- a/runtime/src/exec/host_streams.ts
+++ b/runtime/src/exec/host_streams.ts
@@ -1200,6 +1200,8 @@ export interface HostFuture<T> {
    *    cleanup.
    */
   drop(): void;
+  /** Retire an unwritten producer future while preserving its failure cause. */
+  fail(reason: Error): void;
   value: ComponentValue;
 }
 
@@ -1362,6 +1364,22 @@ function mkFuture<T>(
       activity.close();
       activity.pump();
     },
+    fail(reason: Error) {
+      // Unlike public drop(), producer failure must preserve its site-named
+      // cause. Record it before this call: abandonment notifies parked readers.
+      try {
+        if (!delivered && !shared.dropped) {
+          abandonSharedFuture(shared, reason);
+        } else {
+          shared.drop();
+        }
+      } finally {
+        // `dropSharedForTeardown` also runs observers in a finally block, but
+        // notification itself may throw; retention still ends in that case.
+        activity.close();
+      }
+      activity.pump();
+    },
     value,
   };
   return self;
diff --git a/runtime/tests/embedder/producer_failure_test.ts b/runtime/tests/embedder/producer_failure_test.ts
new file mode 100644
index 0000000..c6938ff
--- /dev/null
+++ b/runtime/tests/embedder/producer_failure_test.ts
@@ -0,0 +1,318 @@
+import { StreamProducerError } from "@polyengine/protocol";
+import { lowerFuture } from "../../src/cabi/async_values.ts";
+import type { ComponentValue } from "../../src/cabi/types.ts";
+import {
+  LiftLowerContext,
+  mkCanonicalOptions,
+} from "../../src/cabi/context.ts";
+import {
+  Future,
+  lowerFutureSource,
+  lowerStreamSource,
+  Stream,
+} from "../../src/embedder/streams.ts";
+import {
+  HostBuffer,
+  hostFutureFor,
+  hostStreamFor,
+} from "../../src/exec/host_streams.ts";
+import {
+  ComponentInstanceState,
+  type ReadableFutureEnd,
+  type SharedFutureImpl,
+  type SharedStreamImpl,
+  Store,
+} from "../../src/task/mod.ts";
+import { assertEq } from "../support/asserts.ts";
+
+const cause = new Error("producer failed");
+const codec = {
+  element: { kind: "u32" } as const,
+  where: "import 'test:producer/fail'.read",
+  toHost: (v: unknown) => v as number,
+  fromHost: (v: number) => v,
+};
+const bytesCodec = { ...codec, element: { kind: "u8" } as const };
+const futureType = { kind: "future", element: codec.element } as const;
+const turn = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+const asValue = (shared: SharedFutureImpl | SharedStreamImpl) =>
+  shared as unknown as ComponentValue;
+
+function deferred() {
+  let resolve!: () => void;
+  return {
+    promise: new Promise<void>((r) => resolve = r),
+    resolve,
+  };
+}
+
+async function rejected(p: PromiseLike<unknown>): Promise<unknown> {
+  return await Promise.resolve(p).then(
+    () => undefined,
+    (e) => e,
+  );
+}
+
+function assertProducerFailure(error: unknown): void {
+  assertEq(error instanceof StreamProducerError, true, String(error));
+  assertEq((error as StreamProducerError).cause, cause);
+  assertEq(String(error).includes(codec.where), true, String(error));
+}
+
+function producerFailure(error: unknown): StreamProducerError {
+  assertProducerFailure(error);
+  return error as StreamProducerError;
+}
+
+function failingStream<T>(first?: T) {
+  let reject!: (reason: unknown) => void;
+  let pulled = false;
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        next: () => {
+          if (!pulled && first !== undefined) {
+            pulled = true;
+            return Promise.resolve({ done: false, value: first });
+          }
+          return new Promise<IteratorResult<T>>((_, r) => reject = r);
+        },
+      };
+    },
+  };
+  return { source, fail: () => reject(cause) };
+}
+
+Deno.test("producer failure: failure before binding remains abandonment after guest lowering", async () => {
+  const raw = lowerFutureSource(Promise.reject(cause), codec);
+  const future = Future.fromLifted<number>(raw, codec);
+  await turn();
+  const error = producerFailure(await rejected(future));
+  assertEq(await rejected(future), error);
+
+  const store = new Store();
+  const inst = new ComponentInstanceState(0, store);
+  const index = lowerFuture(
+    new LiftLowerContext(mkCanonicalOptions(), inst),
+    raw as SharedFutureImpl,
+    futureType,
+  );
+  const end = inst.handles.get(index) as ReadableFutureEnd;
+  let result: unknown;
+  let guestError: unknown;
+  try {
+    end.copy(inst, new HostBuffer(codec.element, null, 1) as never, (r) => {
+      result = r;
+    });
+  } catch (e) {
+    guestError = e;
+  }
+  // `SharedFutureImpl.read` traps on abandonment; it must never manufacture
+  // the reference's valid CopyResult.DROPPED callback for this unwritten end.
+  assertEq(result, undefined);
+  assertEq((guestError as Error).cause, error);
+  // The failure predates binding, so no store existed to receive it.
+  assertEq(store.hostFailure, undefined);
+});
+
+Deno.test("producer failure: a bound future wakes its parked host reader and retires activity", async () => {
+  let fail!: (reason: unknown) => void;
+  const raw = lowerFutureSource(
+    new Promise<number>((_, reject) => fail = reject),
+    codec,
+  );
+  const store = new Store();
+  const inst = new ComponentInstanceState(0, store);
+  const cx = new LiftLowerContext(mkCanonicalOptions(), inst);
+  // Use the real canonical lower path so binding and host activity are genuine.
+  lowerFuture(cx, raw as SharedFutureImpl, futureType);
+  const future = Future.fromLifted<number>(
+    asValue(raw as SharedFutureImpl),
+    codec,
+  );
+  const pending = rejected(future);
+  await turn();
+  fail(cause);
+  const error = producerFailure(await pending);
+  assertEq(store.hostFailure, error);
+  assertEq((raw as SharedFutureImpl).pendingBuffer, null);
+  assertEq(store.pendingHostCalls.size, 0);
+  assertEq(await rejected(future), error);
+});
+
+Deno.test("producer failure: a bound rejection before host await retains identity and retires activity", async () => {
+  let fail!: (reason: unknown) => void;
+  const raw = lowerFutureSource(
+    new Promise<number>((_, reject) => fail = reject),
+    codec,
+  );
+  const store = new Store();
+  const inst = new ComponentInstanceState(0, store);
+  lowerFuture(
+    new LiftLowerContext(mkCanonicalOptions(), inst),
+    raw as SharedFutureImpl,
+    futureType,
+  );
+  fail(cause);
+  await turn();
+  const future = Future.fromLifted<number>(
+    asValue(raw as SharedFutureImpl),
+    codec,
+  );
+  const error = producerFailure(await rejected(future));
+  assertEq(store.hostFailure, error);
+  assertEq((raw as SharedFutureImpl).pendingBuffer, null);
+  assertEq(store.pendingHostCalls.size, 0);
+});
+
+Deno.test("producer failure: an unrelated read rejection remains primary", async () => {
+  let fail!: (reason: unknown) => void;
+  const raw = lowerFutureSource(
+    new Promise<number>((_, reject) => fail = reject),
+    codec,
+  );
+  const host = hostFutureFor<number>(raw);
+  const primary = new Error("primary read failure");
+  host.readResult = async () => {
+    fail(cause);
+    await turn(); // producer failure is recorded before this rejection resumes
+    throw primary;
+  };
+  const future = Future.fromLifted<number>(raw, codec);
+  assertEq(await rejected(future), primary);
+});
+
+Deno.test("producer failure: a COMPLETED future payload wins a later pump failure", async () => {
+  let resolve!: (value: number) => void;
+  const raw = lowerFutureSource(
+    new Promise<number>((r) => resolve = r),
+    codec,
+  ) as SharedFutureImpl;
+  const store: { hostFailure?: unknown } = {};
+  raw.boundStore = store;
+  const write = raw.write;
+  raw.write = (inst, src, _done) => {
+    // Complete the already-parked reader, but keep the producer write pending
+    // so the subsequent throw remains its failure rather than being ignored
+    // after Promise resolution.
+    write.call(raw, inst, src, () => {});
+    throw cause;
+  };
+  const host = hostFutureFor<number>(asValue(raw));
+  const readResult = host.readResult.bind(host);
+  const release = deferred();
+  host.readResult = async () => {
+    const result = await readResult();
+    await release.promise;
+    return result;
+  };
+  const future = Future.fromLifted<number>(asValue(raw), codec);
+  const completed = Promise.resolve(future);
+  await turn(); // park the host read before the producer writes and then fails
+  resolve(7);
+  await turn();
+  const error = producerFailure(store.hostFailure);
+  release.resolve();
+  assertEq(await completed, 7);
+  assertEq(
+    await rejected(Future.fromLifted<number>(asValue(raw), codec)),
+    error,
+  );
+});
+
+Deno.test("producer failure: pending read, iterator, and readable reject", async () => {
+  for (
+    const consume of [
+      (s: Stream<number>) => rejected(s.read(1)),
+      (s: Stream<number>) => rejected(s[Symbol.asyncIterator]().next()),
+      (s: Stream<number>) => rejected(s.readable().getReader().read()),
+    ]
+  ) {
+    const { source, fail } = failingStream<number>();
+    const stream = Stream.fromLifted<number>(
+      lowerStreamSource(source, codec),
+      codec,
+    );
+    const pending = consume(stream);
+    fail();
+    const error = producerFailure(await pending);
+    assertEq(await rejected(stream.read(1)), error);
+  }
+});
+
+Deno.test("producer failure: a completed chunk wins a fault recorded before its continuation", async () => {
+  const { source, fail } = failingStream([7]);
+  const raw = lowerStreamSource(source, codec) as SharedStreamImpl;
+  const store: { hostFailure?: unknown } = {};
+  raw.boundStore = store;
+  const host = hostStreamFor<number>(asValue(raw));
+  const read = host.readable.read.bind(host.readable);
+  const release = deferred();
+  host.readable.read = async (max) => {
+    const result = await read(max);
+    await release.promise;
+    return result;
+  };
+  const stream = Stream.fromLifted<number>(asValue(raw), codec);
+  const completed = stream.read(1);
+  await turn(); // first chunk completed below the gate; the next pull is parked
+  fail();
+  await turn();
+  const error = producerFailure(store.hostFailure);
+  release.resolve();
+  assertEq(await completed, [7]);
+  assertEq(await rejected(stream.read(1)), error);
+});
+
+Deno.test("producer failure: unfinished direct reads reject at zero and partial progress", async () => {
+  for (const first of [undefined, new Uint8Array([1, 2, 3])]) {
+    const { source, fail } = failingStream(first);
+    const stream = Stream.fromLifted<number>(
+      lowerStreamSource(source, bytesCodec),
+      bytesCodec,
+    );
+    let progress = 0;
+    const pending = stream.readDirect((src) => {
+      src.markRead(1);
+      progress++;
+      return "more";
+    });
+    await turn();
+    fail();
+    const error = producerFailure(await rejected(pending));
+    assertEq(progress, first === undefined ? 0 : 3);
+    assertEq(await rejected(stream.read(1)), error);
+  }
+});
+
+Deno.test("producer failure: direct session ended by done keeps its result", async () => {
+  const { source, fail } = failingStream(new Uint8Array([1, 2, 3]));
+  const raw = lowerStreamSource(source, bytesCodec) as SharedStreamImpl;
+  const store: { hostFailure?: unknown } = {};
+  raw.boundStore = store;
+  const host = hostStreamFor<number>(asValue(raw));
+  const readDirect = host.readable.readDirect.bind(host.readable);
+  const release = deferred();
+  host.readable.readDirect = async (consume, info) => {
+    const result = await readDirect(consume, info);
+    await release.promise;
+    return result;
+  };
+  const stream = Stream.fromLifted<number>(
+    asValue(raw),
+    bytesCodec,
+  );
+  const completed = stream.readDirect((src) => {
+    src.markRead(src.remaining().length);
+    return "done";
+  });
+  await turn(); // direct copy completed below the gate; the next pull is parked
+  fail();
+  await turn();
+  const error = producerFailure(store.hostFailure);
+  // Release the already-completed low-level direct result only after the real
+  // producer's next pull failed and recorded the fault.
+  release.resolve();
+  assertEq(await completed, 3);
+  assertEq(await rejected(stream.read(1)), error);
+});