runtime: fix future cleanup and adapter cache writes

contracts/embedder-api.md

@@ -464,6 +464,13 @@ Obligations:
   examples/guests/resource-stream,
   runtime/tests/embedder/resource_stream_test.ts.
 
+**Owned future payloads** (`future<own<R>>`) follow the same delivery rule: the
+producer pump releases a successfully lowered value if the reader drops or the
+write ends without consuming it. Once consumed, ownership belongs to the
+receiver; a later pump failure must not dispose it again. Cleanup preserves an
+existing producer/write error and reports a standalone disposal failure. This
+rule covers top-level `own` payloads, not owns nested in composite values.
+
 **Stream and future values survive round trips.** Lifting an end the host
 already handled — a host-created stream passed back, or a guest stream on its
 second hop — is idempotent, yielding a handle over the same end. Hence:

docs/architecture.md

@@ -473,14 +473,10 @@ already converted to misses need not produce a callback. Component
 validation failures still propagate. Correctly populated entries can be
 read from a read-only cache, with a translator available for misses.
 
-**Current directory-cache limitation:** `put` preserves adapter names such
-as `adapters/0.wasm` beneath its own `adapters/` directory but does not
-create the additional nested directory. Normal adapter-bearing writes
-therefore fail and `translateCached` returns fresh artifacts without
-populating that entry. Adapter-free writes are not implicated, and
-correctly laid-out entries can still be read. Verify actual
-`fromCache: true` results after prewarming; successful translation alone
-does not establish that the cache was populated.
+The directory backend preserves validated adapter paths beneath its
+`adapters/` directory and creates their parent directories when writing.
+Verify actual `fromCache: true` results after prewarming: successful
+translation alone does not establish that cache writes succeeded.
 
 **Engine code caches.** These are independent, opportunistic platform
 optimizations; correctness and artifact caching do not depend on them.

docs/security.md

@@ -192,12 +192,9 @@ provider, not through other grants or a compromised host.
 **Prefer a pre-warmed, read-only cache for fixed deployments.** Translate
 in a trusted build step and deny the production process write access to
 the cache and application artifacts. **Verify actual hits after warming:**
-the current `dirCache.put` does not create the nested directory required
-by normal `adapters/<index>.wasm` names. Those adapter-bearing writes fail,
-but `translateCached` still returns a successful fresh translation.
-Adapter-free writes are unaffected by this limitation, and correctly
-laid-out entries can still be read. Check `fromCache: true` before relying
-on prewarming; this documentation does not resolve the backend limitation.
+`translateCached` returns successful fresh artifacts even if a cache write
+fails. Check `fromCache: true` before relying on prewarming, and use
+`onCacheError` to observe write failures.
 
 The intended permission split is:
 

runtime/src/cache/dir.ts

@@ -136,7 +136,11 @@ class DirCache implements ArtifactCache {
       );
       for (const [file, bytes] of artifacts.adapters) {
         const name = safeRelName(file);
-        await Deno.writeFile(`${tmp}/adapters/${name}`, bytes);
+        const path = `${tmp}/adapters/${name}`;
+        await Deno.mkdir(path.slice(0, path.lastIndexOf("/")), {
+          recursive: true,
+        });
+        await Deno.writeFile(path, bytes);
       }
 
       await rmIfExists(dir);

runtime/src/embedder/streams.ts

@@ -645,11 +645,13 @@ export class Future<T> implements ProtocolFuture<T> {
   }
 
   cancel(): void {
-    if (this.#host !== null) this.#host.cancel();
-    // A deferred future whose host end never materialized has nothing to
-    // cancel; swallow that rejection rather than let `cancel()` produce an
-    // unhandled one (issue #182 — mirrors `drop()` below).
-    else void this.#hostP.then((h) => h.cancel(), () => {});
+    if (this.#host !== null) {
+      try {
+        this.#host.cancel();
+      } catch {
+        // Pump failures remain recorded; public disposal is silent.
+      }
+    } else void this.#hostP.then((h) => h.cancel()).catch(() => {});
   }
 
   /**
@@ -668,11 +670,13 @@ export class Future<T> implements ProtocolFuture<T> {
   drop(): void {
     if (this.#dropped) return;
     this.#dropped = true;
-    if (this.#host !== null) this.#host.drop();
-    // A deferred future whose host end never materialized has nothing to
-    // release; swallow that rejection rather than let `drop()` produce an
-    // unhandled one.
-    else void this.#hostP.then((h) => h.drop(), () => {});
+    if (this.#host !== null) {
+      try {
+        this.#host.drop();
+      } catch {
+        // Pump failures remain recorded; public disposal is silent.
+      }
+    } else void this.#hostP.then((h) => h.drop()).catch(() => {});
   }
 
   /** @internal — see `Stream.dropForTeardown` (#66). */
@@ -964,7 +968,22 @@ export function lowerFutureSource<T>(
   void (async () => {
     try {
       const v = await (src as PromiseLike<T>);
-      await host.write(codec.fromHost(v) as unknown as T);
+      const lowered = codec.fromHost(v);
+      const info = codec.release === undefined ? undefined : { progress: 0 };
+      try {
+        await host.write(lowered as unknown as T, info);
+        if (info?.progress === 0) {
+          throwIfPeerTrapped(host.value, codec.where ?? "future producer", 0);
+        }
+      } catch (e) {
+        try {
+          if (info?.progress === 0) codec.release?.(lowered);
+        } catch {
+          // Preserve the write failure if cleanup also fails.
+        }
+        throw e;
+      }
+      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

runtime/src/exec/host_streams.ts

@@ -1169,8 +1169,8 @@ export function hostStreamFor<T>(value: ComponentValue): HostStream<T> {
 }
 
 export interface HostFuture<T> {
-  /** Deliver the future's single value. */
-  write(value: T): Promise<void>;
+  /** Deliver one value; optional internal output tracks source consumption, even on failure. */
+  write(value: T, info?: { progress: number }): Promise<void>;
   /** Await the future's single value. */
   read(): Promise<T | undefined>;
   /**
@@ -1270,7 +1270,8 @@ function mkFuture<T>(
     activity.notify();
   };
   const self: HostFuture<T> = {
-    write(v: T): Promise<void> {
+    write(v: T, info?: { progress: number }): Promise<void> {
+      if (info !== undefined) info.progress = 0;
       // One operation per direction; opposite ends may rendezvous after a
       // guest round trip. This is a busy guard, not a delivered-value guard.
       if (parked.write) {
@@ -1286,12 +1287,14 @@ function mkFuture<T>(
         parked.write = true;
         try {
           shared.write(writeInst, buf as never, (result: CopyResult) => {
+            if (info !== undefined) info.progress = buf.progress;
             settle("write", result);
             resolve();
           });
           activity.notify();
           activity.pump();
         } catch (e) {
+          if (info !== undefined) info.progress = buf.progress;
           reject(e);
           withdraw("write", buf);
         }

runtime/tests/cache_test.ts

@@ -6,6 +6,8 @@
 //       cargo build -p translator-shim --release --target wasm32-unknown-unknown
 //   - examples/guests/build/{hello,test-suite}.component.wasm
 //       ./examples/build.sh
+//   - harness/generated/values/transcode.0.wasm
+//       just corpus
 
 import { assertEq } from "./support/asserts.ts";
 import { Translator } from "../src/shim/mod.ts";
@@ -119,6 +121,60 @@ Deno.test("dirCache: round-trip, cache hit skips the translator entirely", async
   await roundTrip(cache, "dirCache");
 });
 
+Deno.test("dirCache: real adapters persist with full paths and execute on a warm hit", async () => {
+  const bytes = await readArtifact(
+    "harness/generated/values/transcode.0.wasm",
+    "just corpus",
+  );
+  const dir = await tmpDir();
+  try {
+    const spy = new SpyTranslator(await Translator.create(shimWasm));
+    const errors: unknown[] = [];
+    const opts = {
+      onCacheError: (_op: "get" | "put", err: unknown) => errors.push(err),
+    };
+    const first = await translateCached(spy, bytes, dirCache(dir), opts);
+    assertEq(first.fromCache, false, "cold miss");
+    assertEq(spy.translateCalls, 1, "cold miss translates once");
+    assert(first.adapters.size > 0, "fixture must emit real adapters");
+
+    const spy2 = new SpyTranslator(await Translator.create(shimWasm));
+    const second = await translateCached(spy2, bytes, dirCache(dir), opts);
+    assertEq(second.fromCache, true, "fresh backend reads persisted artifacts");
+    assertEq(spy2.translateCalls, 0, "warm hit never translates");
+    assertEq(errors, [], "cache writes and reads succeed");
+    assertEq(second.plan, first.plan, "cached plan matches translation");
+    assertEq(second.adapters.size, first.adapters.size);
+
+    const hex = await keyHex(await keyFor(spy, bytes));
+    for (const [file, adapter] of first.adapters) {
+      assertEq(
+        second.adapters.get(file),
+        adapter,
+        "cached adapter bytes match",
+      );
+      assertEq(
+        await Deno.readFile(`${dir}/${hex}/adapters/${file}`),
+        adapter,
+        "persisted layout preserves the full adapter name",
+      );
+    }
+
+    for (const result of [first, second]) {
+      const component = await instantiateComponent({
+        plan: result.plan,
+        componentBytes: bytes,
+        adapters: result.adapters,
+      });
+      const run = component.exports.run as () => number;
+      // The corpus guest checks transcoded string bytes on both sides.
+      assertEq(run(), 42, "fresh and cached adapters execute correctly");
+    }
+  } finally {
+    await Deno.remove(dir, { recursive: true });
+  }
+});
+
 Deno.test("webCache: round-trip, cache hit skips the translator entirely", async () => {
   const cache = webCache(`artifact-cache-test-${crypto.randomUUID()}`);
   await roundTrip(cache, "webCache");

runtime/tests/embedder/future-disposal.wat

@@ -0,0 +1,37 @@
+;; Reader abandonment legally wakes the parked writer; its callback then faults.
+(component
+  (type $F (future u32))
+  (core module $Mem (memory (export "mem") 1))
+  (core instance $mem (instantiate $Mem))
+  (canon future.new $F (core func $new))
+  (canon future.write $F async (memory $mem "mem") (core func $write))
+  (canon task.return (result $F) (core func $return))
+  (canon waitable-set.new (core func $set))
+  (canon waitable.join (core func $join))
+  (core module $M
+    (import "" "new" (func $new (result i64)))
+    (import "" "write" (func $write (param i32 i32) (result i32)))
+    (import "" "return" (func $return (param i32)))
+    (import "" "set" (func $set (result i32)))
+    (import "" "join" (func $join (param i32 i32)))
+    (func (export "idle") (result i32) (i32.wrap_i64 (call $new)))
+    (func (export "run") (result i32)
+      (local $ends i64) (local $tx i32) (local $ws i32)
+      (local.set $ends (call $new))
+      (local.set $tx (i32.wrap_i64 (i64.shr_u (local.get $ends) (i64.const 32))))
+      (call $return (i32.wrap_i64 (local.get $ends)))
+      (if (i32.ne (call $write (local.get $tx) (i32.const 0)) (i32.const -1))
+        (then unreachable))
+      (local.set $ws (call $set))
+      (call $join (local.get $tx) (local.get $ws))
+      (i32.or (i32.const 2) (i32.shl (local.get $ws) (i32.const 4))))
+    (func (export "cb") (param i32 i32 i32) (result i32) unreachable))
+  (core instance $m (instantiate $M (with "" (instance
+    (export "new" (func $new))
+    (export "write" (func $write))
+    (export "return" (func $return))
+    (export "set" (func $set))
+    (export "join" (func $join))))))
+  (func (export "run") async (result $F)
+    (canon lift (core func $m "run") async (callback (core func $m "cb"))))
+  (func (export "idle") (result $F) (canon lift (core func $m "idle"))))

runtime/tests/embedder/future-own.wat

@@ -0,0 +1,31 @@
+;; Readable-end abandonment, delivery, and peer faults for future<own<r>>.
+(component
+  (import "r" (type $R (sub resource)))
+  (type $F (future (own $R)))
+  (core module $Mem (memory (export "mem") 1))
+  (core instance $mem (instantiate $Mem))
+  (canon future.drop-readable $F (core func $drop))
+  (canon future.read $F async (memory $mem "mem") (core func $read))
+  (canon resource.drop $R (core func $drop-r))
+  (core module $M
+    (import "" "mem" (memory 1))
+    (import "" "drop" (func $drop (param i32)))
+    (import "" "read" (func $read (param i32 i32) (result i32)))
+    (import "" "drop-r" (func $drop-r (param i32)))
+    (func (export "drop") (param i32) (call $drop (local.get 0)))
+    (func (export "pass") (param i32) (result i32) (local.get 0))
+    (func (export "fail") (param i32) unreachable)
+    (func (export "take") (param i32 i32)
+      (if (call $read (local.get 0) (i32.const 0)) (then unreachable))
+      (call $drop-r (i32.load (i32.const 0)))
+      (call $drop (local.get 0))
+      (if (local.get 1) (then unreachable))))
+  (core instance $m (instantiate $M (with "" (instance
+    (export "mem" (memory $mem "mem"))
+    (export "drop" (func $drop))
+    (export "read" (func $read))
+    (export "drop-r" (func $drop-r))))))
+  (func (export "drop") (param "f" $F) (canon lift (core func $m "drop")))
+  (func (export "pass") (param "f" $F) (result $F) (canon lift (core func $m "pass")))
+  (func (export "fail") (param "f" $F) (canon lift (core func $m "fail")))
+  (func (export "take") (param "f" $F) (param "trap" bool) (canon lift (core func $m "take"))))