feat(task-graph): binary-streaming framework + result-as-reference

[sroussey]

Spec 1 — binary-delta streaming framework

Adds a binary-delta variant to StreamEvent (analogous to text-delta /
object-delta) plus an x-stream: "binary" annotation on output port
schemas, so a task can executeStream byte chunks the same way it streams
text or structured objects. New port helpers (getBinaryPortId,
getBinaryPortFormat, getStreamingPorts), a materializeBinary
assembler (Blob for format: "blob"/absent, ArrayBuffer for
format: "binary"), and a getOutputStreamMode adopter let downstream
code branch cleanly on binary mode without reaching for any.

StreamProcessor accumulates binary-delta chunks per port and merges
them into the enriched finish event so downstream dataflows see the
materialized payload (or, for explicit binary finish payloads, the
artifact wins per Spec 1's precedence rule).

StreamPump adds the graph-aware decision (canStreamBinaryToCache,
anyConsumerNeedsMaterialized) and the pipeBinaryToCache assembly
helper that turns a task's binary-delta events into an AsyncIterable
ready to drive a streaming cache sink.

TaskOutputRepository gains an optional saveOutputStream sink so
file-backed (or other stream-capable) caches can ingest bytes without
materializing the full payload; supportsStreaming() and the
RunPrivateCacheRepo wrapper forward the capability correctly.

Spec 2 — result-as-reference

Builds on Spec 1 to close the queue-row-bloat hole: when the cache backing
supports streaming, the runner pipes the binary bytes straight to the
cache and places a CacheRef placeholder in Output at the port slot.
Downstream Output consumers (and the queue row) see a small envelope
({ \$ref, size?, mime? }) instead of the full payload, while the bytes
live in the cache for hydration on demand.

Pieces:

• CacheRef type + isCacheRef type guard (cache/CacheRef.ts).
• resolveOutput walker (cache/resolveRef.ts) — pure recursive walker
  that hydrates refs through a caller-supplied resolver. Identity is
  preserved when no descendant matches the optional filter; class
  instances (Error, URL, custom classes) survive with prototype
  intact; Map/Set are walked through so nested refs resolve; opaque
  leaves are Blob/ArrayBuffer/TypedArray/Date/RegExp/Promise.
• resolveJobOutput queue-boundary bridge (cache/resolveJobOutput.ts)
  accepting either a CacheRefResolver function or any object exposing
  getOutputByRef (TaskOutputRepository shape).
• IRunConfig.referenceThresholdBytes (default 64 KiB; 0 forces ref
  for every binary output).
• TaskOutputRepository.saveOutputStream now returns Promise<CacheRef>;
  new getOutputByRef / getOutputStreamByRef readers complete the
  contract.
• CacheCoordinator.getBinaryRefSinksByPolicy derives a per-port
  BinaryRefSink map; hydrateRefsBelowThreshold rehydrates refs whose
  committed size falls below the configured threshold (schema-restricted
  to binary streaming ports so legitimate {\$ref: string} fields in
  non-binary slots are not mistakenly hit against the cache).
• StreamProcessor routes binary-delta chunks to a BinaryRefSink
  via a small BinaryStreamRouter producer-consumer pump.
• TaskRunner reads the threshold, builds sinks, threads them through
  StreamProcessor, and rehydrates below-threshold refs in the post-run
  pass — saveByPolicy then writes the small ref-bearing Output.
• StreamProcessor TEES when both an accumulator and a router exist for
  a port (graph context where the cache can stream AND a downstream
  edge needs materialized bytes): the emitted finish event carries the
  materialized Blob/ArrayBuffer for edge consumers; finalOutput
  carries the CacheRef so the queue/cache row stays small.
• RunPrivateCacheRepo forwards all three new optional methods,
  mirroring the backing's true capability on the wrapper instance
  (assigning undefined when the backing lacks them) so callers
  probing typeof === "function" see the truth.

Tests cover binary-delta accumulation + explicit-finish-payload
precedence, port helpers, cache decision + assembly, runner pipe + force-ref +
threshold rehydrate, tee for the graph + materializing-consumer case,
saved-row size + cross-process serialization round-trip + dangling-ref
best-effort, and the walker / resolveOutput / resolveJobOutput
surface (class instances, Map/Set, sparse-ref filter, concurrency bound,
identity preservation).

[github-actions]

Coverage Report

Status	Category	Percentage	Covered / Total
🔵	Lines	62.52%	25544 / 40853
🔵	Statements	62.36%	26421 / 42363
🔵	Functions	63.5%	4822 / 7593
🔵	Branches	51.34%	12601 / 24543

File Coverage

No changed files found.

Generated in workflow #2569 for commit 7c67ab4 by the Vitest Coverage Report Action

[sroussey]

@cursoragent /review

[cursor]

You need to increase your spend limit or enable usage-based billing to run background agents. Go to Cursor

[sroussey]

Outside review — one HIGH follow-up before merge

Pulled this PR into a focused review (security / DX / correctness). The 337b688 fix is correct for the streaming-capable backings (the cache row carries the wire-form CacheRef; cache-hit applies the same below-threshold hydration as a fresh run, so semantics are symmetric). Brand discrimination on CacheRef, the FsFolder REF_PATTERN, and the schema-restricted hydration path together close the obvious cross-tenant / unscoped-resolve risks within a single trust boundary.

One HIGH item that survives the 337b688 fix — flagging here because it requires this branch's code to reproduce, and the fix is small enough to land in this PR rather than as a follow-up.

H-1 — Silent Blob → "{}" corruption in non-streaming cache backings

Where: packages/task-graph/src/task/CacheCoordinator.ts ~218-227 (the saveByPolicy → serializeOutputPorts path).

Symptom: When a cacheable task has an x-stream: "binary" output port AND the configured TaskOutputRepository does NOT implement saveOutputStream (TaskOutputTabularRepository directly → SQLite/Postgres/InMemory; IndexedDbTaskOutputRepository), the StreamProcessor accumulation path materializes binary deltas into a Blob. Then serializeOutputPorts looks up a PortCodec for format: "blob" / "binary". No such codec is registered (grep confirms only imageCacheCodec and this PR's own codecs), so the lookup returns undefined and the raw Blob is handed to JSON.stringify → "{}". Cache row is empty; next cache hit returns nothing useful.

The 337b688 fix only rescues backings that DO implement saveOutputStream (by turning the binary output into a CacheRef so the row carries the envelope, not the Blob). Backings that don't stream still hit the corruption path.

Suggested fix — default PortCodec for blob / binary (Option A)

Preserves user-facing contract (existing non-streaming consumers keep working), matches the precedent set by imageCacheCodec.

New file: packages/task-graph/src/cache/BinaryPortCodec.ts

/**
 * @license
 * Copyright 2026 Steven Roussey <sroussey@gmail.com>
 * SPDX-License-Identifier: Apache-2.0
 */

import { registerPortCodec } from "@workglow/util";

export interface BinaryPortWire {
  readonly __binaryPortWire: 1;
  readonly base64: string;
  readonly size: number;
  readonly mime: string | undefined;
}

function isBinaryPortWire(v: unknown): v is BinaryPortWire {
  if (v === null || typeof v !== "object") return false;
  const o = v as Record<string, unknown>;
  return o.__binaryPortWire === 1 && typeof o.base64 === "string" && typeof o.size === "number";
}

function bytesToBase64(bytes: Uint8Array): string {
  if (typeof Buffer !== "undefined") {
    return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength).toString("base64");
  }
  let bin = "";
  for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i] ?? 0);
  return btoa(bin);
}

function base64ToBytes(b64: string): Uint8Array {
  if (typeof Buffer !== "undefined") {
    const buf = Buffer.from(b64, "base64");
    return new Uint8Array(buf.buffer, buf.byteOffset, buf.byteLength);
  }
  const bin = atob(b64);
  const out = new Uint8Array(bin.length);
  for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
  return out;
}

async function serializeBinary(value: unknown): Promise<BinaryPortWire> {
  if (value instanceof Blob) {
    const bytes = new Uint8Array(await value.arrayBuffer());
    return {
      __binaryPortWire: 1,
      base64: bytesToBase64(bytes),
      size: bytes.byteLength,
      mime: value.type === "" ? undefined : value.type,
    };
  }
  if (value instanceof ArrayBuffer) {
    const bytes = new Uint8Array(value);
    return { __binaryPortWire: 1, base64: bytesToBase64(bytes), size: bytes.byteLength, mime: undefined };
  }
  throw new Error("BinaryPortCodec.serialize: value is not a Blob or ArrayBuffer");
}

registerPortCodec<Blob | ArrayBuffer, BinaryPortWire>("blob", {
  async serialize(value) { return serializeBinary(value); },
  async deserialize(wire): Promise<Blob> {
    if (!isBinaryPortWire(wire)) throw new Error("BinaryPortCodec(blob).deserialize: input is not a BinaryPortWire");
    const bytes = base64ToBytes(wire.base64);
    return new Blob([bytes.buffer as ArrayBuffer], wire.mime ? { type: wire.mime } : undefined);
  },
});

registerPortCodec<Blob | ArrayBuffer, BinaryPortWire>("binary", {
  async serialize(value) { return serializeBinary(value); },
  async deserialize(wire): Promise<ArrayBuffer> {
    if (!isBinaryPortWire(wire)) throw new Error("BinaryPortCodec(binary).deserialize: input is not a BinaryPortWire");
    const bytes = base64ToBytes(wire.base64);
    return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength);
  },
});

Wire registration: packages/task-graph/src/common.ts — side-effect import next to imageCacheCodec's:

import "./cache/BinaryPortCodec";

No change needed to CacheCoordinator.ts — the existing getPortCodec(prop.format) lookup will now hit the new codecs.

Tests (suggested) — packages/test/src/test/task-graph/CacheSerialization.test.ts

1. Blob round-trip via non-streaming backing. Task with format: "blob" port returns new Blob([new Uint8Array([1,2,3,4])], { type: "application/octet-stream" }). Run against SpyRepo. Assert JSON.stringify(repo.saved[0]) is NOT '{"bytes":{}}' AND repo.saved[0].bytes.__binaryPortWire === 1. Re-run → cache hit → assert rehydrated Blob bytes deep-equal [1,2,3,4] and blob.type === "application/octet-stream".
2. ArrayBuffer round-trip (format: "binary"). Same shape; rehydrated value is ArrayBuffer (not Blob); bytes match.
3. Streaming backing still produces a CacheRef. SpyRepo with saveOutputStream; assert repo.saved[0].bytes matches isCacheRef(...) and is NOT a BinaryPortWire — confirms codec doesn't interfere with the ref path.

Notes

• Memory: base64 is ~1.33× the binary size + UTF-16 spike inside JSON.stringify. The PR's IRunConfig.referenceThresholdBytes only governs ref-vs-inline on streaming-capable backings; non-streaming backings have no equivalent guard. Mitigation is out of scope here, but the codec docstring could call this out (small payloads only — use a streaming-capable backing for large blobs).
• Cross-runtime: Blob is available in Node 18+, Bun, and all browsers; Buffer is gated behind typeof Buffer !== "undefined". Same gating as imageCacheCodec.
• Registration ordering: last-writer-wins, same contract as imageCacheCodec.

Happy to send a follow-up PR against this branch if it's easier than folding it in here — let me know.

---

Other findings (not blockers; included as a heads-up — see below)

MEDIUM / LOW findings from the review

M-1 — resolveJobOutputStream(handle, backing) without explicit port resolves any branded CacheRef reachable in the output — cache/resolveJobOutput.ts:64-83. collectCacheRefs walks every property including user-controlled fields; a hand-crafted {kind: "task-graph/CacheRef", $ref: "..."} embedded in metadata would stream through the backing. The same protection hydrateRefsBelowThreshold applies (schema-binary-port allow-list) is missing here. Either pass the task's output schema and filter to declared binary ports, or document that callers must pass port when the producer is not fully trusted.

M-2 — BinaryStreamRouter.wakeDrain releases ALL chained drain waiters at once — task/StreamProcessor.ts:444-456, 520-528. Even when consuming a single chunk only freed a small fraction of the buffer. Producers re-park immediately if still over the mark, but the thundering herd briefly pushes bufferedBytes above highWaterMarkBytes by N chained chunk-pushes before the next park. Wake one waiter at a time, or have each waiter re-check the mark before resolving its outer Promise.

M-3 — resolveOutput walker doesn't preserve Symbol-keyed / non-enumerable properties or properly reconstitute class instances with private fields — cache/resolveRef.ts:179-195. Object.keys(source) is string-enumerable-only; Object.create(proto) skips the constructor, so a class Foo { #x = 1 } instance loses its private field. Real-world impact is small (task outputs are usually plain data) but the docstring claims "the returned clone preserves their prototype" without noting these caveats. Narrow the docstring, or fall back to leaf when proto !== Object.prototype && proto !== null.

M-4 — RunPrivateCacheRepo shadows prototype overrides with undefined — cache/RunPrivateCacheRepo.ts:53-60. A caller using repo.saveOutputStream?.(args) is fine; repo.saveOutputStream(args) without the check gets TypeError. Worth a one-line docstring note: "capability probing MUST use typeof === 'function' or ?.()".

M-5 — FsFolderTaskOutputRepository.clearOlderThan — storage/FsFolderTaskOutputRepository.ts:135-142. deleteBlobsByPrefix("", cutoff) deletes all blob files whose mtime is older than cutoff with no row-presence check; a pruned-row whose blob is newer (concurrent write) gets the row removed but the blob kept until the next sweep. Self-healing; worth a docstring note or row-presence check.

M-6 — replayBinaryRefs emits binary-delta events synchronously with no backpressure — task/CacheCoordinator.ts:200-213. Stream-reading from FsFolder is paced by createReadStream's default 64 KiB chunks, so memory is bounded for that backing — but a future getOutputStreamByRef implementation that yields the entire payload as one chunk would blow up consumer memory. Document the implementation requirement, or pass the consumer's high-water mark through.

L-1 — CacheRef.size is "best-effort" but FsFolderTaskOutputRepository is the only first-party backing that populates it; third-party saveOutputStream implementations may silently bypass below-threshold inlining. Flag in saveOutputStream's docstring.
L-2 — resolveOutput's Promise.all parallel walk doesn't short-circuit on a thrown resolver error. Use Promise.allSettled + first-error-throw, or accept it.
L-3 — RunPrivateCacheRepo.fallbackWarned is process-static; tests exercising the fallback twice get one warning total. _resetWarnedForTests would help isolation.
L-5 — BinaryStreamRouter private getters _bufferedBytes / _highWaterMarkBytes are exposed as test hooks via public get. Collapse to a single _getMetrics() to avoid leaking internal state into the public type.

---

Overall this is a large, carefully-architected slice with thorough test coverage. The brand discrimination on CacheRef, the FsFolder REF_PATTERN, and the schema-restricted hydration close the obvious risks. H-1 above is the one item I'd hold the merge for; everything else is non-blocking.

https://claude.ai/code/session_017TXwbp2GD6jvXrELz6msZW

---

Generated by Claude Code

[sroussey]

H-1 addressed in 1d2f730: default blob/binary port codecs (cache/BinaryPortCodec.ts, registered via side-effect import in the package entry) encode inline binary values as a base64 BinaryPortWire envelope so non-streaming JSON-row backings round-trip instead of persisting {}.

One deliberate deviation from the suggested codec: serialize/deserialize pass CacheRefs and unknown shapes through unchanged rather than throwing — serializeOutputPorts/deserializeOutputPorts apply the codec to every format: "blob" port, including ports whose value is the ref envelope on the streaming path, so a strict-throw codec would break ref rows (and the suggested test 3).

Tests: Blob and ArrayBuffer round-trips through a non-streaming JSON-row backing (cache hit returns original bytes + mime), and ref-path non-interference (streaming rows still store the branded ref, no wire envelope). The old "contrast" test asserting the raw-Blob bloat row now asserts the encoded envelope. Full task-graph+task suites: 1987 passed.

Also landed the doc-hardening notes for M-1 (explicit port for untrusted producers), M-4 (capability probing contract), M-6 (bounded chunk requirement on getOutputStreamByRef), and L-1 (populate size on refs). M-2/M-3/L-2/L-3/L-5 left as non-blockers per the review.

---

Generated by Claude Code

[sroussey]

Pushed 5 HIGH-priority follow-up fixes on top of 1d2f730. All packages/test/src/test/task-graph tests (875) pass on this branch; one unrelated pre-existing failure remains in JobQueueStreamWorker.integration.test.ts (Node worker_threads transfer semantics) that is identical with or without these commits.

• b04a42d — fix(task-graph): guard resolveOutput walker against cycles and shared subtrees. Threads a WeakSet<object> through walk, hasMatchingRef, and collectCacheRefs so a self-referential output or a shared sub-tree no longer stack-overflows the resolver. Cycles preserve their topology (the back-edge is returned by reference rather than rewritten).
• a846500 — fix(task-graph): treat Error and URL as opaque leaves in ref walker. Error.message/stack are own non-enumerable, URL is all prototype accessors — the generic Object.keys() clone was silently dropping their data while keeping the prototype. Adds both to isLeaf (and the matching collectCacheRefs skip) so they pass through by reference.
• 4fd581f — fix(task-graph): fsync blob temp handle before rename in FsFolderTaskOutputRepository. Adds await handle.sync() before close in saveOutputStream so the atomic-rename pattern actually delivers the durability promise its JSDoc claims. Directory fsync intentionally skipped (cache semantics tolerate a rename loss; cost not justified).
• 16bb667 — fix(task-graph): clean up orphan blobs when stream-write succeeds but row commit fails. Adds optional deleteOutputByRef on TaskOutputRepository, implements it in FsFolderTaskOutputRepository, and a CacheCoordinator.cleanupOrphanBlobsForBinaryPorts helper the runner invokes on saveByPolicy rejection to best-effort delete just-written blobs before re-throwing. clearOlderThan JSDoc updated to note that periodic janitor sweeps are still required to catch hard-kill cases that race the in-band cleanup. Covered by a new TaskRunnerRefPath test that injects a failing saveOutput and asserts the blob is dropped.
• 0fca60d — fix(task-graph): document single-tenant assumption of FsFolderTaskOutputRepository deterministic tier. Doc-only. Calls out that the deterministic blob naming (sanitize(taskType), fingerprint(inputs)) carries no tenant axis and shares blobs across tenants with identical inputs — an existence side channel. Points operators at per-tenant prefixes or RunPrivateCacheRepo for the multi-tenant case. Behavior unchanged.

No departures from the suggested fixes; the one tweak worth flagging is in b04a42d — hasMatchingRef is called with a fresh WeakSet from inside walk (separate from walk's own visited set) so the pre-scan still terminates on cycles without forcing the walker to share state with the containment check.

---

Generated by Claude Code

[sroussey]

Three follow-up review fixes pushed on top of the current branch:

1. b4fe120 — Ref walker: opaque-by-default for class instances

The previous walker only opted Error and URL into the opaque set. Every other class instance (Headers, Request, Response, FormData, URLSearchParams, ReadableStream, user-defined classes with private fields, etc.) was still walked via Object.keys() and silently cloned to an empty object — data lives on the prototype or in private slots and Object.keys() cannot see it.

Inverted the policy in resolveRef.ts and mirrored it in resolveJobOutput.collectCacheRefs: walk only plain objects (prototype Object.prototype / null), Array, Map, Set. Every class instance is opaque and returned by reference. The prototype-preserving Object.create(proto) branch collapses now that only plain objects reach it. New tests cover Headers, Request, Response, FormData, URLSearchParams, Error, URL, and a user class with private fields.

2. 87d9f4e — Directory fsync after rename

The temp blob file handle was fsynced before rename, but the containing directory was not. On ext4 data=ordered (and similar journaled filesystems) a crash between the rename and the directory metadata flush can leave the published name pointing at zero bytes.

After the rename, open(blobsDir, "r") → dir.sync() → dir.close(), wrapped in try/catch that swallows EPERM, EINVAL, ENOTSUP, EISDIR for filesystems that reject opening a directory for fsync. Integration test covers the happy round-trip + 16-way concurrent writes; the unsupported-FS error codes can't naturally fire on a Linux tmp dir, so they're covered by the swallow list in code review.

3. 7c67ab4 — Per-write unique blob suffix (the CRITICAL race)

Blob name was <sanitized-taskType>_<fingerprint>.bin, derived entirely from (taskType, inputs). Two concurrent runners with the same (taskType, inputs) wrote to the same path. If runner A's row-commit failed and triggered the orphan cleanup (cleanupOrphanBlobsForBinaryPorts → deleteOutputByRef), A would unlink the blob that runner B's successfully-committed row still pointed at — silent data loss.

Appended a per-write randomUUID() suffix: <sanitized-taskType>_<fingerprint>_<uuid>.bin. Each saveOutputStream lands at a unique path; concurrent writers can't race on the same file. The sanitized taskType prefix is unchanged, so prefix-scoped pruning still cascades correctly. The existing REF_PATTERN regex already matches both the new and legacy shapes, so old refs written by previous versions continue to resolve.

New tests cover: concurrent writers with identical inputs producing distinct refs, A's deleteOutputByRef cleanup leaving B's blob intact, and backward compatibility with legacy un-suffixed blob names.

Verification: bun scripts/test.ts graph vitest → 91 files / 890 tests passing; bun run build:types clean.

---

Generated by Claude Code

[sroussey]

Two HIGH findings from a fresh review pass on this PR — both narrowly scoped, both currently uncovered by tests. Patches and test sketches below.

HIGH 1 — RunPrivateCacheRepo does not forward deleteOutputByRef

File: packages/task-graph/src/cache/RunPrivateCacheRepo.ts (constructor ~lines 42–67; missing method).

Failure mode: CacheCoordinator.cleanupOrphanBlobsForBinaryPorts (CacheCoordinator.ts:332) guards typeof cache.deleteOutputByRef !== "function" and silently returns when missing. RunPrivateCacheRepo overrides saveOutputStream / getOutputByRef / getOutputStreamByRef but never overrides deleteOutputByRef; the base class declares it optional. When a run-private cache writes a streamed blob successfully and the row-commit then fails (the exact case commit 16bb667 was added to fix), orphan cleanup becomes a silent no-op — the blob lingers on disk until the periodic janitor sweep.

Patch (sketch):

In the constructor, immediately after the getOutputStreamByRef capability-shadow block (around line 65):

if (typeof backing.deleteOutputByRef !== "function") {
  (this as { deleteOutputByRef?: unknown }).deleteOutputByRef = undefined;
}

New method, after getOutputStreamByRef (~line 154):

public override deleteOutputByRef(ref: CacheRef): Promise<void> {
  if (typeof this.backing.deleteOutputByRef !== "function") return Promise.resolve();
  return this.backing.deleteOutputByRef(ref);
}

Do not inspect the ref string for the run id — refs are opaque to the wrapper, and cross-run isolation is already structural via the __run:<runId>::<taskType> namespacing.

Tests to add (extend packages/test/src/test/task-graph-cache/RunPrivateCacheRepo.test.ts):

• Backing supports stream: save → exists → deleteOutputByRef(ref) → blob gone, getOutputByRef(ref) returns undefined.
• Regression pin: FsFolder backing wrapped in RunPrivateCacheRepo; monkey-patch saveOutput to throw after a successful streamed write; drive cleanupOrphanBlobsForBinaryPorts; assert the blob file no longer exists on disk. (This is the test that fails on the current PR head.)
• Backing without deleteOutputByRef: assert typeof wrapper.deleteOutputByRef === "undefined".
• Cross-run isolation: two wrappers run-A/run-B sharing one FsFolder backing; A's delete leaves B's blob intact.

HIGH 2 — BinaryStreamRouter swallows sink rejection → producer hangs forever

File: packages/task-graph/src/task/StreamProcessor.ts (constructor ~lines 430–440; push parks at high-water mark ~lines 448–464; fail ~lines 475–481).

Failure mode: The constructor invokes this.refPromise = sink(this.iterable()) then attaches this.refPromise.catch(() => {}), suppressing the unhandled rejection but NOT propagating it. If a sink rejects BEFORE iterating (e.g. FsFolderTaskOutputRepository.saveOutputStream's await open(tmpPath, "w") throws EACCES / ENOSPC, or RunPrivateCacheRepo.saveOutputStream throws because the backing's stream method is missing), the iterable's next() is never called, chunkNotify is never set, and the producer's await router.push(chunk) parks at the 8 MiB high-water mark forever. executeStream's top-level catch only runs after the for-await exits — but the loop is itself awaiting that very push. The task hangs until external abort.

Patch:

Replace the swallowed catch in the constructor with a real handler:

this.refPromise.catch((err: unknown) => {
  this.fail(err instanceof Error ? err : new Error(String(err)));
});

fail() (lines 475–481) already sets failure, flips finished = true, and calls wakeChunk() + wakeDrain(), so the parked producer wakes and subsequent pushes early-return at the existing if (this.finished) return Promise.resolve(); check. The attached .catch returns a discarded chain — the original this.refPromise rejection still surfaces through await router.ref(). No caller-side change needed.

Tests to add (extend packages/test/src/test/task-graph/StreamProcessorBinaryRefSink.test.ts, each with vitest { timeout: 2000 }):

• Synchronous sink rejection before iteration: sink that returns Promise.reject(new Error("EACCES")) without iterating. Drive a producer that pushes ~10 MiB as small chunks (above the 8 MiB high-water mark). Assert (i) loop completes — 2s timeout is the failure signal, (ii) await router.ref() rejects with "EACCES".
• Asynchronous sink rejection after partial iteration: sink consumes 3 chunks then throws; assert producer eventually returns and await router.ref() rejects.
• Happy-path regression: existing sink behavior unchanged.

Verify BinaryStreamRouter is exported from packages/task-graph/src/index.ts; if not, add the named export so tests can import it.

---

Happy to push these if it's preferable to land them here vs as a follow-up PR.

---

Generated by Claude Code