streams: invalid direct callback diagnostics reject the peer and leave the session parked #346

Summary

When a direct-access byte-edge callback returns an invalid verdict that
itself is unserializable (e.g. bigint), the diagnostic-message
construction throws before the session is failed. The result violates the
documented contract for invalid-verdict handling
(contracts/embedder-api.md §"Streams and futures" ("Direct-access byte
edges")): instead of failing the direct session and leaving the peer's
parked operation undisturbed, the peer (the side that happened to be
arriving) is rejected with a raw serialization TypeError, and the
direct-session side is left permanently pending — never resolved, never
rejected — until the caller manually cancels and drops the stream.

Where

runtime/src/exec/host_streams.ts:555-563
(#runDirect, baseline 1a4f5e1):

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.md §"Streams and futures" ("Direct-access byte edges"))`,
    ),
  );
  return "failed";
}

JSON.stringify(verdict) is evaluated as part of building the argument to
new TypeError(...), i.e. before this.#fail(...) is called. If
verdict is a bigint (or contains a bigint, a circular reference, etc.
— anything JSON.stringify itself throws on), the TypeError allocation
throws first. That throw is not caught by the try { verdict = this.invoke(scope) } catch (e) { ... }
block a few lines above (lines 545-553) — it only wraps the callback
invocation itself, not the subsequent verdict-validation code — so it
escapes #runDirect entirely, uncaught, and propagates through
runtime/src/task/streams.ts:269-270
(rendezvousCopy's src.runDirect(...) / dst.runDirect(...) call) into
whichever SharedStreamImpl.read/.write
(runtime/src/task/streams.ts:417-518)
call happened to trigger the rendezvous — i.e. into the arriving peer's
call frame, not the direct session's.

Contract

Per #routeDirectNoCopy's own doc comment
(runtime/src/task/streams.ts:520-526):

Retire a direct session without reporting a copy to its peer. Retraction
resolves the session with its total; failure has already rejected it.
[...] Neither path drops the stream or emits a zero-progress completion
to the peer's nonzero-capacity operation.

and the "failed" DirectOutcome variant's own doc
(runtime/src/task/streams.ts:238-239):

"failed" — misuse or a throwing callback; the session has already
rejected. No copy, no event, the peer's parked operation survives.

The contract is: an invalid-callback-verdict failure rejects the direct
session and leaves the peer's parked operation untouched (to be
re-satisfied normally later). This bug instead: (1) never rejects the
direct session (#fail is never reached), and (2) rejects the peer
with an unrelated internal TypeError ("Do not know how to serialize a
BigInt") that has nothing to do with the peer's own operation.

Expected vs actual

Expected (per contract above): the direct-callback side's pending
operation rejects with the "a direct-access callback must return..."
diagnostic; the peer's parked operation is left undisturbed.

Actual, reproduced against 1a4f5e1c6f3fc906ce576acdc4499ef2745d6e41:

{"writerOutcome":"pending","readerOutcome":"TypeError:Do not know how to serialize a BigInt"}
writer retry: TypeError
after cleanup: {"writerOutcome":"resolved:0","readerOutcome":"TypeError:Do not know how to serialize a BigInt"}

• The reader (the arriving peer, uninvolved in the invalid callback)
  rejects with a raw JS serialization error instead of any documented
  diagnostic.
• The writer (the direct session whose callback misbehaved) never settles
  on its own: a same-session retry write (writer.write(...)) throws
  "stream busy", confirming the session is stuck pending, not failed.
  It only resolves — with 0 progress — after the caller calls
  cancelWrite() and drop() from outside.

Repro

Standalone, repo-relative (run from repo root):

deno run --allow-read --config runtime/deno.json <(cat <<'EOF'
import { Stream } from "./runtime/src/embedder/streams.ts";
const codec = {
  element: { kind: "u8" } as const,
  fromHost: (v: number) => v,
  toHost: (v: unknown) => v as number,
};
const { stream, writer } = Stream.create<number>();
stream.bindElement(codec);
let writerOutcome = "pending", readerOutcome = "pending";
// Invalid JS callback result: it must fail the direct session, not its peer.
const write = writer.writeDirect((() => 1n) as never)
  .then((v) => writerOutcome = "resolved:" + v, (e) => writerOutcome = e.name + ":" + e.message);
const read = stream.read(1)
  .then((v) => readerOutcome = "resolved:" + JSON.stringify([...v]), (e) => readerOutcome = e.name + ":" + e.message);
await new Promise((r) => setTimeout(r, 30));
console.log(JSON.stringify({ writerOutcome, readerOutcome }));
try {
  writer.write(new Uint8Array([9]));
} catch (e) {
  console.log("writer retry:", e.name);
}
writer.cancelWrite();
stream.drop();
await Promise.all([write, read]);
console.log("after cleanup:", JSON.stringify({ writerOutcome, readerOutcome }));
EOF
)

(Or save the block above as a .ts file anywhere and run
deno run --allow-read --config runtime/deno.json <file> — the import
path is relative to wherever the file lives, so adjust
./runtime/src/embedder/streams.ts accordingly.)

Verified twice against the same commit; deterministic both runs.

Suggested fix direction

Build the diagnostic TypeError message using a serialization that
cannot itself throw — e.g. guard JSON.stringify(verdict) with a
try/catch (falling back to String(verdict) or typeof verdict), or
String(verdict) directly, since the message only needs to name what was
returned, not round-trip it. That keeps the existing
this.#fail(...); return "failed"; path reachable for any invalid
verdict, matching the "misuse or a throwing callback; the session has
already rejected" contract for all misuse shapes, not just the ones whose
verdict happens to be JSON-serializable.