Implement per-port backpressure gating for no-accumulation passthrough edges

[sroussey]

Summary

Adds per-port backpressure gating to the no-accumulation passthrough streaming path, enabling fine-grained pacing of producers to match consumer read rates on individual ports. Previously, passthrough edges had no pacing mechanism; producers could race ahead unbounded. This change introduces BackpressureGate coordination between producer and consumer streams, with separate gates per source port feeding independent consumers.

Key Changes

• Per-port edge gates (EdgeGateState): StreamPump now builds a BackpressureGate for each source port that feeds a no-accumulation passthrough edge. Gates track pending event costs in a FIFO and coordinate producer/consumer pacing.

• Producer-side pacing: StreamProcessor receives an edgeBackpressure callback from the graph runner. After emitting each delta, the processor awaits the gate for that port, parking until the consumer has read enough to bring buffered bytes below the high-water mark.

• Consumer-side credit: The passthrough stream's reader charges the gate as events are enqueued and credits it as the consumer reads, using streamEventCost to measure bytes per event.

• Safety cleanup: If a passthrough consumer reaches a terminal state (COMPLETED, FAILED, DISABLED, ABORTING) without fully reading its stream, the gate is closed to release any parked producer; every edge-stream teardown path (stream end, terminal producer status, reader cancel) also closes the gate.

• Deadlock liveness guard: a gate is built only when the consumer can make read progress while the producer is parked — if any other edge into the consumer is sourced from the producer or one of its DAG descendants (drained, mode-mismatched, or static edges), that edge settles only after the producer finishes, so gating would deadlock the pair. Such consumers keep the ungated passthrough (correct, just unpaced).

• Stream-fed consumers run uncached: a run consuming a live stream at an unsettled input port computes its cache key without the streamed content, so caching is disabled (kind: "none") for that run — preventing stale hits and poisoned rows when two runs differ only in stream payload. Drained edges keep caching.

• Producer failure propagates in-stream: a producer failure enqueues an in-stream error event before edge streams close, so an early-dispatched consumer fails instead of completing (and caching) on truncated input; abort keeps the graceful close.

• Mixed-mode and fan-out test coverage: Two new comprehensive test suites validate the gating behavior:

  • StreamMixedModeFanout.test.ts: One source with separate append and object ports, each feeding its own passthrough consumer with independent gates; also tests fan-out (one port → two consumers) which correctly falls back to tee'd drain without a gate.
  • StreamBackpressureEngaged.test.ts: Fast append producer feeding a slow consumer over a small high-water mark; verifies producer is parked and peak lead stays bounded.

• Cache hit replay pacing: CacheCoordinator now accepts edgeBackpressure in CacheReplayContext, allowing cache-hit replay to respect the same per-port gates as fresh runs so slow consumers pace both paths identically; append/object logs decode in a single pass instead of buffering the whole byte log.

• Validation and schema improvements:

  • Added isDeltaStreamMode() helper to identify modes that can be encoded to byte streams (append, object, binary).
  • Added portForcesStreamValidation() to read the x-validate-stream opt-in, allowing ports to opt out of both validation exemption and passthrough pacing.
  • Updated TaskRegistry to validate binary format on both input and output ports.
  • AiTask / ImageEditTask / ImageGenerateTask forward the validateInput skip-ports parameter.

• Instance schema support: CacheCoordinator and TaskRunner now use instance schemas (via task.inputSchema() / task.outputSchema()) instead of static schemas, enabling dynamic-schema tasks to add ports at runtime without breaking cache key normalization, sink setup, or cache-hit replay.

• Streaming output hydration: resolveJobOutput now adapts inline append/object values to byte streams (UTF-8 text, NDJSON) for consistent caller behavior whether the value came from a ref or inline; input-ref hydration decodes append/object refs through their stream codecs (string / folded object) instead of handing byte Blobs to string ports.

• Cycle detection in ref resolution: resolveOutput now detects cyclic values and uses memoization for acyclic ones, ensuring shared subtrees resolve once and both slots receive the same resolved copy.

• File write robustness: FsFolderTaskOutputRepository now loops on partial writes to ensure the full chunk reaches disk before returning the size stamped into the CacheRef, and opens stream reads eagerly so a pruned blob reports a clean cache miss instead of throwing ENOENT mid-iteration.

• Orphan-blob cleanup: covers all delta modes (was binary-only); a stream ending without a finish event fails its cache sinks so truncated blobs are discarded, not published. (The run-private tier does not build streaming sinks after the run-scoping reconciliation from main — it falls back to accumulation — so no orphan refs can arise there.)

Notable Implementation Details

• Gates are installed only for edges that satisfy isNoAccumulationPassthroughEdge(): single consumer, same mode on both ends, consumer is a streamable non-subgraph task, no validation opt-in — plus the liveness guard above.
• The edgeBackpressure callback is threaded through IRunConfig → StreamProcessor → task's IExecuteContext backpressure() hook, allowing both direct emitters and StreamProcessor-routed deltas to respect pacing.
• Passthrough streams are never tee'd (no materialize copy), avoiding the accumulation the passthrough exists to prevent; the edge keeps the same stream reference so "this edge is streaming" checks still hold.
• Non-qualifying edges (transforms, mode mismatch, fan-out, * edges, non-streamable or subgraph targets) fall back to the existing drain path — correct, just without the memory and pacing win. With noAccumulation off, behavior is unchanged.
• The branch merges claude/relaxed-faraday-ql3olr after its reconciliation with main (run-scoped private cache, degrade-to-miss cache decode, edge-stream reader-cancel teardown); conflict resolutions combine both sides — see the merge commit message for details.

Verification (post-merge head)

• bun scripts/test.ts graph vitest green (962 tests); bun scripts/test.ts task vitest green (1119, 24 skipped); job-queue + FsFolder storage spot suites green
• npx tsc -p packages/task-graph/tsconfig.json --noEmit clean; bun run build:types clean across all 36 packages
• Policy-OFF runs identical to pre-change behavior (explicit assertions in test files)

https://claude.ai/code/session_01T89ffUgYbtp6UztpEXWyVW

[github-actions]

Coverage Report

Status	Category	Percentage	Covered / Total
🔵	Lines	63.29%	26521 / 41898
🔵	Statements	63.13%	27433 / 43454
🔵	Functions	64.58%	5017 / 7768
🔵	Branches	52.27%	13131 / 25117

File Coverage

File	Stmts	Branches	Functions	Lines	Uncovered Lines
Changed Files
packages/storage/src/tabular/ITabularStorage.ts	20%	0%	0%	28.57%	226-230

Generated in workflow #2643 for commit d443441 by the Vitest Coverage Report Action