Merge pull request #788 from workglow-dev/claude/libs-issues-triage-o2hmfl

feat(task-graph): stream N binary output ports to cache on the unflagged path

Author: sroussey

packages/task-graph/src/EXECUTION_MODEL.md

@@ -296,9 +296,9 @@ Binary output ports whose bytes were piped into a stream-capable cache carry a b
 
 **Rows store the wire form**: the cached row always carries the `CacheRef`, never inline bytes — JSON-row backings would destroy an inline `Blob`/`ArrayBuffer` (`JSON.stringify(Blob)` is `{}`). Below-threshold hydration to inline bytes applies to the value **returned to the caller**, identically on fresh runs and cache hits.
 
-**Single binary port**: outside the per-port path below the runner drives a single binary sink, so this path supports exactly one binary output port. Tasks with multiple binary ports take the accumulation path (enforced in both `StreamPump.canStreamBinaryToCache` and `CacheCoordinator.getBinaryRefSinksByPolicy`) — unless the run opts into the per-port sink path below, which has no such limit.
+**Binary-only, any number of ports**: this path takes one _or more_ binary output ports — each gets its own sink and its own `CacheRef`, and `mintRefKey(taskType, fingerprint, port)` keeps the blobs distinct. What it does not take is a **mix**: a task whose streaming ports are not all binary falls back to accumulation (enforced in both `StreamPump.canStreamBinaryToCache` and `CacheCoordinator.getBinaryRefSinksByPolicy`), because only the binary ports get sinks here, so an `append` / `object` port alongside them would have neither a sink nor — with accumulation skipped — an accumulator, and its deltas would be dropped into an empty finish payload. Such mixed tasks stream every port only under the per-port path below.
 
-**One writer, two gating rules**: a backing declares exactly one streaming writer, `saveOutputStreamPort(taskType, inputs, port, mode, chunks, metadata)`, probed by `supportsStreaming()`. The single-binary case is that writer called with the task's one binary port and `mode: "binary"`; `CacheCoordinator.getBinaryRefSinksByPolicy` adapts it there, since that is the one place the port id is already known. What still differs between the two paths is when the runner _uses_ the writer, not what a backing has to implement: a single binary port streams to the cache unconditionally, while `append` / `object` ports stream only under `noAccumulation`.
+**One writer, two gating rules**: a backing declares exactly one streaming writer, `saveOutputStreamPort(taskType, inputs, port, mode, chunks, metadata)`, probed by `supportsStreaming()`. The binary case is that writer called per binary port with `mode: "binary"`; `CacheCoordinator.getBinaryRefSinksByPolicy` selects those ports and shares one writer closure with the all-mode builder, so the two paths differ in which ports they select, never in how a port's bytes are written. What still differs is when the runner _uses_ the writer, not what a backing has to implement: binary ports stream to the cache unconditionally, while `append` / `object` ports stream only under `noAccumulation`.
 
 **Self-healing dangling refs**: when a ref needed for replay or hydration no longer resolves (blob evicted, cache cleared), the hit converts into a **miss** — the task re-executes and rewrites both the row and the bytes. No events are emitted before all refs are validated.
 
@@ -346,7 +346,7 @@ The **SQL** backings share one implementation. `TabularBlobChunkStore` persists
 
 ### Per-port stream sinks and the no-accumulation passthrough
 
-Everything above generalizes from "one binary port" to **every delta stream mode**
+Everything above generalizes from "binary ports only" to **every delta stream mode**
 (`append`, `object`, `binary`) behind an opt-in run flag,
 `TaskGraphRunConfig.noAccumulation` (default off — off is byte-identical to the
 accumulation path).

packages/task-graph/src/task-graph/StreamPump.ts

@@ -404,10 +404,10 @@ export class StreamPump {
   ): boolean {
     if (outputCache) {
       // Relaxation: when the cache can ingest a byte stream, the task streams
-      // ONLY binary, and no downstream edge needs the materialized value, the
-      // bytes are piped straight to the cache sink instead of being buffered
-      // into an enriched finish event. This is the memory win for large binary
-      // outputs (e.g. file/image producers).
+      // ONLY binary (any number of such ports), and no downstream edge needs
+      // the materialized value, the bytes are piped straight to one cache sink
+      // per port instead of being buffered into an enriched finish event. This
+      // is the memory win for large binary outputs (e.g. file/image producers).
       if (StreamPump.canStreamBinaryToCache(this.graph, task, outputCache)) return false;
       // No-accumulation passthrough: under the opt-in flag, a cacheable task
       // whose streamable ports can each be sunk per-port (and no consumer needs
@@ -454,7 +454,9 @@ export class StreamPump {
    *    `RunPrivateCacheRepo` may expose a concrete writer while its
    *    `supportsStreaming()` reflects the BACKING repo, so the duck-type would
    *    falsely report `true` over a non-streaming backing store).
-   * 2. The task's only streaming output port(s) are binary.
+   * 2. The task's streaming output ports are ALL binary — one or many. Each
+   *    gets its own sink and its own {@link CacheRef}, so a two-artifact
+   *    producer is not pushed back onto full in-memory accumulation.
    * 3. No downstream dataflow edge needs the materialized value (every consumer
    *    accepts the raw binary stream, or there are no consumers).
    *
@@ -479,21 +481,25 @@ export class StreamPump {
 
     const outSchema = task.outputSchema();
     const streamingPorts = getStreamingPorts(outSchema);
-    // Exactly ONE binary port: outside the no-accumulation path the runner
-    // drives a single binary sink, so only one port can pipe to the cache.
-    // With accumulation skipped, any additional binary port would have neither
-    // a sink nor an accumulator and its chunks would be silently dropped —
-    // multi-port tasks must take the accumulation path instead.
-    if (streamingPorts.length !== 1 || streamingPorts[0].mode !== "binary") return false;
+    // At least one streaming port, and EVERY one of them binary. The count is
+    // free (each binary port gets its own sink and its own CacheRef), but the
+    // binary-only half is load-bearing: `getBinaryRefSinksByPolicy` builds
+    // sinks for binary ports only, so an `append`/`object` port alongside them
+    // would have neither a sink nor — with accumulation skipped — an
+    // accumulator, and its deltas would be silently dropped. Such mixed tasks
+    // take the accumulation path unless the caller opts into `noAccumulation`,
+    // where `canStreamAllPortsToCache` sinks every delta-mode port.
+    if (streamingPorts.length === 0) return false;
+    if (!streamingPorts.every((p) => p.mode === "binary")) return false;
 
     return !StreamPump.anyConsumerNeedsMaterialized(graph, task);
   }
 
   /**
    * All-mode analogue of {@link canStreamBinaryToCache} for the opt-in
    * no-accumulation path. The two share a capability probe but NOT a gating
-   * rule: a single binary port streams to the cache unconditionally, while
-   * append / object ports only do so under the flag. True when the flag is on,
+   * rule: binary ports stream to the cache unconditionally, while append /
+   * object ports only do so under the flag. True when the flag is on,
    * the task is cacheable, the cache reports `supportsStreaming()`, every streaming
    * output port is a delta mode (`append` / `object` / `binary`), and no
    * downstream edge needs a materialized value. Then each port is sunk