Add streaming binary output support and cache-ref passthrough#608
Open
sroussey wants to merge 46 commits into
Open
Add streaming binary output support and cache-ref passthrough#608sroussey wants to merge 46 commits into
sroussey wants to merge 46 commits into
Conversation
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).
Adds a same-process channel so a holder of a `JobHandle` can subscribe
to a running job's stream events (text deltas, object deltas,
binary-delta chunks, snapshot, finish, error, phase) instead of only
the terminal result.
Worker side
-----------
- `IJobExecuteContext` gains an optional `emitStreamEvent(event)` method.
- `JobQueueWorker` plumbs a per-job event emitter through into the
execute context so a run-fn can call `ctx.emitStreamEvent(...)` to
publish stream chunks as they're produced.
Server side
-----------
- `JobQueueServer.forwardToClients("handleJobStream", jobId, event)`
fans the event to every attached client by direct method invocation —
pure in-memory, no `postMessage`, no serialization, no worker thread.
The channel is intentionally same-process only; storage-backed cross-
process clients see state transitions through `subscribeToChanges`
but receive no incremental stream events.
Client side
-----------
- `JobHandle.onStream(callback)` is exposed only when the client is
server-attached (`this.server` set); callers branch on
`typeof handle.onStream === "function"`.
- Each listener invocation is wrapped in try/catch so one throwing
subscriber does not abort delivery to the rest or break the dispatch.
Tests
-----
- `JobQueueStream.test.ts` proves end-to-end same-process delivery: a
worker's `emitStreamEvent` calls reach every `JobHandle.onStream`
listener in order.
- `JobQueueStreamWorker.integration.test.ts` (+ its `.fixture.mjs`)
validates the underlying Node `worker_threads` transfer mechanism
the design relies on for any future cross-thread queue host: binary
chunks emitted from a worker thread transfer (not copy) to the host
per `WorkerServerBase.extractTransferables`. The docblock spells
out that this is a Node-primitive validation, NOT a test of the
current package's behavior — today's queue channel is entirely
same-process and the test exists as a navigational marker for a
future hosted-in-thread variant.
…ly collisions
`CacheRef` was discriminated by shape alone: any object with `{ $ref: string }`
satisfied `isCacheRef`, including JSON-Schema `$ref` pointers embedded in
metadata. The cache-ref resolver walks task outputs and calls
`getOutputByRef(ref)` on every match — so any code path that surfaces an
attacker-influenced `{$ref: "cache://OTHER_RUN/secret"}` shape (e.g. a tool
result, an AI structured-output field, a parsed-JSON document) could trick
`resolveJobOutput` / `resolveOutput` into reading bytes from another run or
tenant's private cache slot.
This patch adds a literal `kind: "task-graph/CacheRef"` brand discriminator
that:
- survives JSON serialization across queue rows / IPC (Symbol-based brands
would be erased by `JSON.stringify` and break cross-process resolution);
- is checked by `isCacheRef` alongside the `$ref` string;
- is applied uniformly by a new `makeCacheRef(...)` helper that callers
use to construct refs.
`CacheCoordinator.getBinaryRefSinksByPolicy` and
`RunPrivateCacheRepo.saveOutputStream` now defensively re-wrap the value
returned by legacy backings (`isCacheRef(raw) ? raw : makeCacheRef(raw)`),
so a backing that pre-dates the brand still produces a discriminator-bearing
ref when seen through the framework. In-tree test repositories and callers
are updated to use `makeCacheRef`.
Test coverage:
- `CacheRef.test.ts` now expects shape-only `{$ref: string}` to be rejected
and exercises JSON round-trip preserving the brand.
- `resolveOutput.test.ts` adds a case where a JSON-Schema-shaped
`{schema: {$ref: "#/\$defs/Foo"}}` is left untouched and the resolver is
never called (identity preserved).
- `resolveJobOutput.test.ts` adds the cross-tenant attack case: an
attacker-supplied `{note: {\$ref: "cache://OTHER_RUN/secret"}}` never
invokes `getOutputByRef`.
…b"|"binary"
`materializeBinary` previously accepted any string and silently coerced
unknown values (including casing typos like `"Blob"`) to the ArrayBuffer
branch. A task author writing `format: "Blob"` would unknowingly produce an
ArrayBuffer where every downstream consumer expected a Blob — and the
mismatch only surfaced at the consumer (often as a misleading runtime
error during streaming, or worse, silent data corruption when the consumer
duck-typed both shapes).
This patch establishes a canonical `BinaryFormat = "blob" | "binary"` type
and routes every binary-port consumer through a single
`assertBinaryFormat(schema, port)` helper:
- `undefined` and `"blob"` resolve to `"blob"` (the documented default);
- `"binary"` resolves to `"binary"`;
- anything else throws with the allowed vocabulary in the message.
`materializeBinary` now takes the canonical `BinaryFormat` directly and
`StreamProcessor` / `CacheCoordinator.hydrateRefsBelowThreshold` both call
`assertBinaryFormat` before invoking it.
`TaskRegistry.registerTask` runs the same check at registration time over
every output port with `x-stream: "binary"`, so the typo fails near the
task definition rather than during a streaming run. The task is not added
to the registry when the check fails.
Test coverage:
- `StreamBinaryTypes.test.ts` replaces the now-removed "unknown format =
binary" behavior with `assertBinaryFormat` cases for `"blob"`,
`"binary"`, undefined-default, the casing typo `"Blob"`, and an unknown
value (`"wat"`).
- `TaskRegistry.test.ts` adds cases asserting registration throws on a
binary port with `format: "Blob"`, and succeeds on `"blob"` /
`"binary"`.
- `Spec2QueueRowAndRehydrate.test.ts` adds symmetric rehydration cases:
`format: "blob"` rehydrates into a `Blob`, `format: "binary"` into an
`ArrayBuffer`.
…efault 8 MiB)
The streaming binary router buffered chunks without bound. A fast producer
(e.g. an AI image / audio generator yielding 1 MiB chunks) feeding a slow
sink (remote object store, throttled FS) would let the producer race ahead
and accumulate the entire payload in memory before the sink saw the first
chunk — turning a notionally O(1) streaming path into peak-residency O(N).
The old comment even acknowledged the issue ("backpressure: there is none")
and offloaded the problem onto the sink author.
This patch:
- Introduces `DEFAULT_BINARY_HIGH_WATER_BYTES = 8 MiB` in `StreamTypes.ts`.
- `BinaryStreamRouter` now tracks `bufferedBytes` (sum of un-consumed
chunk sizes). `push(chunk)` returns a Promise that resolves
immediately while `bufferedBytes < highWaterMarkBytes`, and parks the
producer until the consumer drains under the mark otherwise. `end()`
and `fail()` BOTH release any parked producer so an abort mid-park
does not leak the Promise.
- `StreamProcessor` `await router.push(...)` on every `binary-delta`
yield, so the byte-bounded backpressure applies for tasks running
through the standard streaming path.
- `IRunConfig.binaryHighWaterBytes` lets callers override per-run.
Threaded through `TaskRunner` → `StreamProcessor.run` deps.
- `IExecuteContext.binaryBackpressure?: () => Promise<void>` is a
cooperative hook for tasks that emit via a side channel and cannot
use the awaited `push` path; the StreamProcessor and StreamPump
install router-aware implementations, and an absent runtime supplies
a no-op (free for tasks that don't call it).
- `StreamPump.pipeBinaryToCache` (the EventEmitter path used for the
cache-ingest tee) gets the same byte-counted queue and returns a
`backpressure()` function alongside `promise` / `detach`.
Test coverage in `StreamingBackpressure.test.ts` adds a "binary
backpressure" describe block:
- 100 × 1 MiB through a slow (50 ms / chunk) sink with a 4 MiB
high-water mark: peak buffer stays at or below `mark + 1 chunk`
and every byte is delivered.
- End-to-end: 100 MiB through `StreamProcessor.run` with the same
high-water mark, asserting full delivery without drops.
- Abort-while-parked: a producer parked at the high-water mark sees
its `push()` Promise settle within 100 ms of `r.end()`.
Adds supportsStreamingReads() to TaskOutputRepository (mirrored by RunPrivateCacheRepo), plus streamRefViaBacking/byteIterableFromBlob and the RefStreamBacking shape in resolveRef. Extracts the shared in-memory streaming repo test double into packages/test bindings.
…om cache Streams a completed job's binary output out of the cache backing by port or single-ref discovery, adapting inline Blob/ArrayBuffer/Uint8Array values. makeJobOutputStreamResolver produces the injectable resolver shape for job-queue (which cannot depend on task-graph).
…it replay StreamPump.anyConsumerAcceptsBinaryStream inspects outgoing edges for binary-to-binary stream pass-through; the graph runner threads the result into each task run as IRunConfig.hasStreamingConsumers.
On a cache hit whose binary ports hold CacheRefs, CacheCoordinator now mirrors the fresh-run event contract: cached bytes replay as chunked binary-delta events for stream-capable consumers and hydrate into the enriched finish event for materializing consumers, while the returned output keeps the ref. Dangling refs convert the hit into a miss so the task re-executes and rewrites the entry. Consumer needs are graph-computed (anyConsumerNeedsMaterialized / anyConsumerAcceptsBinaryStream) and threaded through IRunConfig.
Branded CacheRefs reaching a task's resolved inputs are resolved against the run's cache registry (private first, then deterministic) and inlined per the port's format annotation before validation and cache-key computation. Binary-streaming ports with a live input stream are skipped; unresolvable refs fail the task with a named-port error.
…decar files First production streaming-capable output cache (node/bun, exported via a new common-server entry): JSON rows through FsFolderTabularStorage, binary payloads as sidecar blob files written incrementally and published by atomic rename. Deterministic blob naming from (taskType, input fingerprint) overwrites instead of leaking; refs from foreign or path-traversal shaped $refs never resolve. Includes a generic stream-out contract suite run against both the in-memory and FS repositories, and an end-to-end cache-hit replay through the FS backing.
…inary results JobQueueClientOptions accepts an injected outputStreamResolver (built via task-graph's makeJobOutputStreamResolver — the dependency edge points the other way, so the resolver is structural). When configured, handles expose outputStream(port?) which awaits completion and streams the binary result out of the output cache without materializing it.
…tory Review findings: prefix-scoped row deletions (RunPrivateCacheRepo.clearRun, CacheJanitor sweeps) now cascade to blob sidecar files instead of leaking them; blob names fingerprint the raw taskType so lossy sanitization cannot make two task types share a blob file; a failed write or rename removes its .tmp instead of stranding it. Also clears the shared test repo between job-queue outputStream tests.
…am listeners on abort/error pipeBinaryToCache had no production callers — StreamProcessor's BinaryStreamRouter owns live byte delivery to the cache sink — so the duplicate queue/backpressure implementation and its test scaffolding are gone. createStreamFromTaskEvents now also terminates on the task's abort/error events (which never emit stream_end), closing the edge stream and detaching listeners instead of leaking them and leaving downstream readers waiting forever.
…ngle-binary-port streaming
Review findings: (1) saveByPolicy now runs before below-threshold
rehydration so JSON-row backings persist the serializable CacheRef
envelope instead of an inline Blob that stringifies to {} — and cache
hits apply the same hydration before returning, so small outputs come
back inline on both paths. (2) canStreamBinaryToCache and
getBinaryRefSinksByPolicy both require exactly one binary streaming
port; multi-port tasks fall back to accumulation instead of silently
dropping every port without a sink.
…ackings
Review follow-up (H-1): a cacheable task with a binary output port on a
NON-streaming backing accumulates an inline Blob/ArrayBuffer that
JSON.stringify silently turns into {} in the row. New BinaryPortCodec
registers default codecs for format blob/binary that encode inline
bytes as a base64 BinaryPortWire envelope and decode back to the
port's declared type. CacheRefs and unknown shapes pass through
unchanged in both directions so streaming-backed rows keep their ref
envelopes verbatim. Also adds the review's docstring hardening notes:
explicit-port guidance for portless resolveJobOutputStream, bounded
chunk requirement on getOutputStreamByRef, size population on
saveOutputStream refs, and capability-probing rules for
RunPrivateCacheRepo.
… subtrees The walker recursed into every reachable object without a visited set, so a self-referential output or a shared sub-tree stack-overflowed the resolver loop. Thread a `WeakSet<object>` through `walk`, `hasMatchingRef`, and `collectCacheRefs`; revisited objects short-circuit by reference rather than recursing. Cycles preserve their topology — refs inside the cycle are not rewritten on the back-edge.
Errors carry `message` / `stack` as own non-enumerable properties and `URL` exposes everything via prototype accessors. The generic `Object.keys()` clone in `walk` would have dropped that data while preserving the prototype, leaving a hollow shell. Add `Error` and `URL` to `isLeaf` (and the matching skip in `collectCacheRefs`) so they pass through by reference instead of being restructurally cloned.
…OutputRepository The atomic-rename pattern only guarantees a published name pointing at the right inode; it doesn't guarantee that the data has reached storage. On a crash between the rename and the OS flushing dirty pages, the published blob name can resolve to zero bytes — the very partial-blob scenario the rename was meant to avoid. Add `handle.sync()` before close so the data is durable when the rename announces it. Skip the directory fsync (cache semantics tolerate a renamed-but-unflushed-directory crash; it just forces a recompute).
… row commit fails A streaming binary save is a two-phase operation: the sink writes the blob (producing a CacheRef) and then the row commit points at it. If the row commit failed (or the process died between the two), the blob persisted on disk with nothing referencing it, and the row-driven cleanup paths would never find it. Add an optional `deleteOutputByRef` hook on `TaskOutputRepository`, implement it in `FsFolderTaskOutputRepository`, and expose a `CacheCoordinator.cleanupOrphanBlobsForBinaryPorts` helper that the runner calls on `saveByPolicy` failure to drop just-written blobs before re-throwing. Document that periodic `clearOlderThan` is still required to catch the hard-kill case that races the in-band cleanup.
…putRepository deterministic tier Blob names in the deterministic-cache path are `(sanitize(taskType), fingerprint(inputs))` with no tenant axis. Two tenants on a shared backing with identical inputs resolve to the same name — a blob-existence side channel for sensitive inputs. Document the single-tenant assumption on the class and the fingerprinting site, and point operators at the supported wrappers (per-tenant folder/prefix or `RunPrivateCacheRepo`) for the multi-tenant case. Behavior is unchanged.
The previous walker walked any object with prototype != Object.prototype
when isLeaf opted them in (e.g. Error, URL). Class instances whose data
lives on the prototype (accessors) or in private slots — Headers,
Request, Response, FormData, URLSearchParams, ReadableStream, and any
user-defined class — were still walked via Object.keys() and silently
cloned to empty objects.
Invert the policy: walk only plain objects (Object.prototype / null
prototype), Array, Map, and Set. Every class instance is opaque and
returned by reference. The cycle/short-circuit logic in hasMatchingRef
and walk now reach the plain-object branch only for plain objects, so
the prototype-preserving Object.create(proto) branch in walk collapses
to {}.
Mirror the same opaque-by-default policy in collectCacheRefs in
resolveJobOutput.ts so both walkers stop at the same boundary.
…s between rename and dir-metadata flush On ext4 `data=ordered` (and similar journaled filesystems) the rename of the temp blob to its published name is not durable until the parent directory's metadata is also flushed. A crash between the rename and that metadata flush can leave the published name visible but pointing at stale (zero-byte) content — the file handle was already fsync'd, but the directory entry change is lost. After the rename, open the blobs directory and call `sync()` on the handle, then close it. Run this best-effort: swallow `EPERM`, `EINVAL`, `ENOTSUP`, `EISDIR` from the dir-open for filesystems / platforms that reject opening a directory for fsync (the rename is still the durability boundary; on a recompute the cache simply re-runs the task). Add an integration test that exercises the happy path round-trip and a 16-way concurrent-write scenario to confirm the dir-sync does not break normal flow or serialize writes incorrectly. The unsupported-FS error codes can't be naturally produced on a Linux tmp dir, so they're covered by the swallow list in code review.
…rphan-cleanup races The blob name was `<sanitized-taskType>_<fingerprint>.bin`, derived entirely from `(taskType, inputs)`. Two runners executing the same task with the same inputs would both write to the same path. If runner A's row-commit failed and triggered the orphan-blob cleanup (`cleanupOrphanBlobsForBinaryPorts` -> `deleteOutputByRef`), A would unlink the blob that runner B's successfully-committed row was still pointing at — a silent data-loss race. Append a per-write `randomUUID()` suffix to the blob filename so each `saveOutputStream` invocation lands at a unique path. Concurrent writers no longer share a file, so a row-cleanup on one path can't touch the other writer's blob. The published `$ref` still carries the sanitized taskType prefix, so prefix-scoped pruning (`deleteByTaskTypePrefix` / `clearOlderThanWithTaskTypePrefix`) keeps cascading correctly. The existing REF_PATTERN regex already accepts both the new `<taskType>_<fingerprint>_<uuid>.bin` shape and the legacy `<taskType>_<fingerprint>.bin` shape, so old refs written by previous versions of this repository still resolve via `getOutputByRef`. Tests cover (1) two concurrent writers with identical inputs producing distinct readable refs, (2) the cleanup race — A's `deleteOutputByRef` leaves B's blob intact, and (3) backward compatibility with legacy un-suffixed blob filenames.
Pull the byte-bounded park/wake accounting out of BinaryStreamRouter into a cost-agnostic BackpressureGate (charge / credit / close / fail / awaitBelowMark) with its own unit tests. The router now delegates its buffer accounting to the gate and keeps only its chunk buffer and consumer-wake; behavior is unchanged (existing binary streaming and cache stream-out tests pass as-is). The gate is the reusable primitive a future no-accumulation streaming path will build transport backpressure on once streams are no longer materialized at full speed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
The rebase that linearized the binary-streaming branch onto main preserved a stale `classToStorage` import: current main removed both the export from JobStorageConverters and the symbol's use here, keeping only storageToClass. The leftover import failed the type build (TS2305 unresolved member, TS6133 unused). Import only storageToClass; the StreamEventLike import is unaffected. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
A streaming task that declares x-stream:"replace" finished with an empty
payload and no preceding snapshot would silently return {} — clearing its
own output. Surface a clear error in both the accumulation and
no-accumulation finish branches instead. A final snapshot, a non-empty
finish payload, or a binary ref still resolves normally.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Per-port stream sinks need a single cached row to carry more than one ref (binary and text/object). Add optional `port` and `mode` to ICacheRef and makeCacheRef. Both are optional: legacy portless binary refs are still recognized by isCacheRef and resolve exactly as before. Wire format only — no resolver/replay behavior change yet. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Add cache/streamCodec.ts: per-port/per-mode codecs (append=UTF-8 text, object=NDJSON delta log, binary=identity) that encode a port's deltas to bytes, decode them back to deltas for replay, and materialize them to the port value. Extract the object-delta fold into a shared foldObjectDelta in StreamTypes (used by both the live accumulator and the codec, single source of truth). Add optional saveOutputStreamPort(taskType, inputs, port, mode, chunks, metadata) to TaskOutputRepository and implement it on FsFolder (port-named sidecar, mode/port tagged on the returned CacheRef; shared writeSidecar with the binary path — behavior-preserving). Not wired into the runner yet (that's the sink/no-accumulation task). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Mirror of #600's StreamEventAccumulator "no clobber" test on the streaming (.run via executeStream) path: an object-mode toolCalls delta must survive a static {text:"",toolCalls:[]} finish scaffold, and the absent text port falls back to the scaffold default. Guards the B1 class against drift in the rewritten StreamProcessor finish-enrichment. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Extends the binary-only ref-sink path to every streamable mode so a
multi-port streaming task can pipe each port straight to the cache as a
CacheRef instead of accumulating in memory.
- StreamSink { mode, write } generalizes BinaryRefSink; StreamProcessor
routes append/object/binary deltas to a per-port sink via the mode's
codec (single-event encode), mirroring the binary router/backpressure
path. binaryRefSinks stays as a binary-mode shim.
- StreamPortCodec gains encodeEvent (per-delta primitive); encode folds
over it.
- CacheCoordinator.getRefSinksByPolicy builds a sink per streamable port
via saveOutputStreamPort (replace excluded). supportsStreamingPorts()
capability flag; RunPrivateCacheRepo forwards saveOutputStreamPort with
runId namespacing.
Purely additive: nothing wires getRefSinksByPolicy/refSinks into the
runner yet, so behavior is unchanged until the no-accumulation flag lands.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Adds IRunConfig.noAccumulation + streamHighWaterBytes (and the matching TaskGraphRunConfig fields), a TaskGraphRunner field set from the run config, and threads both through StreamingRunOptions onto each streaming task's run config so the flag reaches TaskRunner — the seam the no-accumulation passthrough path (skip-drain + per-port edge gate) builds on. Pure plumbing, default off: nothing consumes the flag behaviorally yet, so every edge takes today's drain and behavior is byte-identical. Verified against the full graph (931) and task (1117) vitest suites. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
…lidation Whole-value input validation is a settled-value concept: a port fed by a live event stream has no settled value to validate this run (its slot may hold only a CacheRef pointer). validateInput now takes an optional skipPorts set that drops those ports from both the validated object and a derived schema (properties + required), so neither a type mismatch on a ref nor a `required` check rejects a valid streaming run. The runner computes the skip set value-awarely: a stream-wired port is exempted only when its slot is a CacheRef/undefined; a port carrying a settled value is still validated. So off the no-accumulation path (drain materializes the value) behavior is byte-identical. `x-validate-stream: true` opts a port back in (forces materialize-and-validate). hydrateInputRefs likewise skips any stream-wired port with a live stream (was binary-only), leaving the ref as the durable pointer. This is the seam the no-accumulation passthrough path needs: the consumer's static input slot can hold the upstream CacheRef for all modes without tripping the validator. No-op today (verified graph 936 + task 1117 green). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
…drain Wires the opt-in no-accumulation path end-to-end for a same-mode streaming passthrough edge (append/object/binary), so producer → (consumer + per-port cache sink) flows without the full-speed materialize drain: - StreamPump.canStreamAllPortsToCache: all-mode analogue of the binary relaxation — a cacheable task whose streamable ports can each be sunk per-port (cache implements saveOutputStreamPort) and whose consumers don't need a materialized value skips accumulation under the flag. - StreamPump.isNoAccumulationPassthroughEdge: the precise passthrough predicate (same delta mode both ends, no transforms, single consumer, not x-validate-stream). awaitStreamInputs skips the drain for such edges; EdgeMaterializer sets the per-port CacheRef as the edge value instead of skipping the streaming edge. - TaskRunner selects getRefSinksByPolicy (all modes) under noAccumulation; honors streamHighWaterBytes. - CacheCoordinator.hydrateRefsBelowThreshold generalized to append/object (codec materialize) alongside binary, so a below-threshold ref rehydrates to its value on fresh run and cache hit. The consumer reads the live stream (tee'd to executeStream); its static input slot holds the upstream ref, exempt from validation by the prior seam. Off the flag, every edge takes today's drain (byte-identical). New end-to-end append test; graph 938 + task 1117 green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Generalizes the cache-hit stream-out so an append/object CacheRef replays like a fresh run, completing the no-accumulation path's read side: - CacheCoordinator.replayBinaryRefs -> replayStreamRefs: append/object refs decode through the per-mode codec (text-delta / object-delta events for a stream-capable consumer; codec.materialize for the finish payload), binary keeps its chunk-by-chunk path (memory-bounded, format-aware). - StreamPump.anyConsumerAcceptsStream: all-mode analogue of the binary-only consumer probe, so a cache hit replays deltas for append/object streaming consumers too (hasStreamingConsumers hint). The binary-only helper stays for its existing unit tests. Cache hit ≡ fresh run for every mode now. New CacheHitReplayParity test (append + object): the second run serves from cache without re-executing the source, and the consumer output matches the fresh run. graph 940 + task 1117 green; existing CacheHitReplay + StreamBinaryPump suites unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
…h edge Pace a streaming producer to its passthrough consumer's read rate so peak buffered bytes stay bounded for a fast-producer / slow-consumer pair: - streamEventCost(event): per-event byte cost (UTF-8 for text deltas, JSON length for object deltas, byte length for binary) shared by charge/credit. - StreamPump builds one BackpressureGate per passthrough source port (high-water mark = streamHighWaterBytes, default 8 MiB). The edge stream charges the gate as events are enqueued; a credit-on-read wrapper around the consumer's stream credits it back, and end/abort/error/consumer termination all close the gate so a parked producer is never orphaned. - StreamProcessor awaits the graph-installed edgeBackpressure thunk after each delta (per-port), keeping the task layer edge-agnostic; the cooperative ctx hook is generalized to backpressure() (awaits cache-sink routers AND edge gates) with binaryBackpressure kept as a deprecated alias. - prepareStreamingInputs no longer tees a passthrough edge: its materialize copy is never drained, so the tee branch would silently retain every event. - Gates engage only on isNoAccumulationPassthroughEdge; fan-out (2+ consumers of one port) keeps the tee'd drain — in-order delivery to every consumer, best-effort pacing by design. Flag off = unchanged behavior. Tests: StreamBackpressureEngaged (peak producer lead bounded by the mark, off-path runs free), StreamMixedModeFanout (append+object ports pace independently per-port; fan-out delivers all events in order to both consumers). EXECUTION_MODEL.md documents per-port sinks, cache-as-tee, skippable materialization, all-mode backpressure, the noAccumulation + streamHighWaterBytes knobs, x-validate-stream, the single-consumer scope, and inline-only backings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T89ffUgYbtp6UztpEXWyVW
…struction TS 5.7+ types Uint8Array over ArrayBufferLike, which no longer satisfies BlobPart; the in-memory repo doubles never hold SharedArrayBuffer-backed views, so the assertion is safe. Fixes @workglow/test build-types. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T89ffUgYbtp6UztpEXWyVW
Liveness and cache-correctness fixes on the streaming/cache paths:
- Cache policy: a run consuming a live stream at an unsettled input port
now runs uncached (kind none) — the streamed content cannot contribute
to the cache key, so two runs differing only in stream payload would
collide on one entry (stale hits, poisoned rows).
- Edge gate liveness: a passthrough gate is built only when the consumer
can reach its stream reads while the producer parks; any other edge into
the consumer sourced from the producer or its descendants (drained,
mode-mismatched, or static edges) falls back to ungated passthrough
instead of deadlocking the pair. Credit reuses the charged cost via a
FIFO (no double JSON.stringify), and a read failure also closes the gate.
- Passthrough predicate: the target must itself be a streamable,
non-subgraph task — only streamable tasks receive ctx.inputStreams, and
subgraph hosts need the drain for their inner tasks' settled inputs.
- canStreamBinaryToCache now requires task.cacheable (matching the sink
builder); a non-cacheable binary streamer no longer silently drops its
output to {}.
- Per-port edge filtering treats binary-delta as a port delta, so one
port's bytes can no longer leak into another port's edge stream.
- Cache-hit replay honors the consumer-edge gate per delta and decodes
append/object logs in a single pass (emit + fold) instead of buffering
the whole byte log and decoding twice; dangling-ref misses release
already-opened streams; lookup/save/key use the instance schemas so
dynamic-schema tasks replay the ports they wrote.
- Producer failure enqueues an in-stream error event before closing edge
streams, so an early-dispatched consumer fails instead of completing
(and caching) on truncated input; abort keeps the graceful close.
- StreamProcessor fails (not ends) its cache-sink routers when the stream
ends without a finish event, so an aborted/truncated run discards the
partial blob instead of publishing it.
- Input hydration decodes append/object refs through their stream codec
(string / folded object) instead of handing byte Blobs to string ports,
resolves ports concurrently, and inline string/object job outputs adapt
to streams like their surviving-ref forms.
- Orphan-blob cleanup covers all delta modes (was binary-only) and works
on the private tier (RunPrivateCacheRepo now forwards deleteOutputByRef).
- FsFolder: stream reads open the fd eagerly (ENOENT → clean miss instead
of mid-iteration throw; no prune TOCTOU) and short writes are looped to
completion so ref.size never overstates the file.
- resolveOutput resolves shared subtrees once via promise memoization
(second occurrence previously got the original, unresolved object);
cyclic values keep the conservative visited-set behavior.
- TaskRegistry validates binary port formats on input schemas too;
AiTask/ImageEditTask/ImageGenerateTask forward the validateInput
skip-ports parameter.
- Cleanups: born-deprecated binaryBackpressure alias, dead
CacheCoordinator.saveStream, anyConsumerAcceptsBinaryStream, and
getBinaryPortId removed; isDeltaStreamMode/portForcesStreamValidation
helpers replace six hand-expanded copies; shared TextEncoder and
chunked browser base64 on hot paths; spec references removed from test
comments per repo rule.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T89ffUgYbtp6UztpEXWyVW
main refactored the run-private cache scoping model (removed the ns() taskType-prefix + deleteByTaskTypePrefix cleanup) in favor of a first-class run-scoped backing contract (saveOutputForRun / getOutputForRun / deleteRun / sizeForRun / deleteRunOlderThan) fronted by a dedicated RunPrivateTaskOutputRepository. This branch's streaming persistence layer was built on the old prefix model, so the merge required adapting it: - Conflicts resolved: CacheCoordinator + TaskRunner imports (specific-module form main adopted; kept isCacheRef/resolveReferenceThreshold/DataPortSchema); StreamPump edge-stream teardown (took main's cleanup()/onStatus/cancel refactor, dropped the old onTerminate/detach path); RunPrivateCacheRepo (took main's version wholesale). - CacheCoordinator.lookup: hoisted outputSchema above main's new decode try/catch so replayStreamRefs still sees it. - Scoped reconciliation: streaming is a deterministic-cache capability; RunPrivateCacheRepo no longer forwards streaming (run-private streaming degrades to accumulation, since main's run-private backing is not stream-capable). FsFolderTaskOutputRepository drops the obsolete deleteByTaskTypePrefix / clearOlderThanWithTaskTypePrefix overrides; blob cleanup runs through clearOlderThan. Run-private-streaming and prefix-cascade tests updated to the new contract. Verified: task-graph typecheck clean; graph 958 + task 1119 + storage/queue 1987 vitest green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
main's TypeScript bump makes Uint8Array generic (Uint8Array<ArrayBufferLike>), which is no longer directly assignable to BlobPart. The two in-test streaming cache repos' getOutputByRef built `new Blob([bytes])` — cast to BlobPart as the codebase does elsewhere (streamCodec). Fixes the build-types / typecheck CI failures surfaced only by tsc (vitest/esbuild skips type-check). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
The streaming framework + no-accumulation passthrough grew task-graph's TypeScript instantiations to 70814 (from the 59917 baseline, +18%), tripping the typecheck-budget +15% guard. This is expected growth for the feature, so re-baseline the single regressed package to its measured value (others are untouched and within budget). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
Coverage Report
File Coverage
|
||||||||||||||||||||||||||||||||||||||
…eaming phase-2 branch Conflict resolutions combine both sides' intents: - CacheCoordinator.lookup: keep the parent's degrade-to-miss try/catch around getOutput/deserialize AND this branch's instance output schema (dynamic-schema tasks replay the ports they wrote). - TaskRunner imports: parent's specific-module cache imports plus this branch's streamRefViaBacking / getStreamPortCodec. - StreamPump.createStreamFromTaskEvents: parent's hoisted idempotent cleanup + terminal-status listener + reader-cancel teardown, with this branch's passthrough-gate close folded into cleanup (every teardown path wakes a parked producer) and the in-stream error event emitted on FAILED before cleanup so drained consumers fail instead of settling on truncated data; abort stays a graceful close. - RunPrivateCacheRepo: take the parent's first-class run-scoped design (saveOutputForRun/getOutputForRun/deleteRun); the wrapper no longer forwards streaming sinks, so the private tier deliberately falls back to accumulation and this branch's deleteOutputByRef forward is moot. - Test doubles: parent's BlobPart cast form. Verified post-merge: graph 962 green, task 1119 green (24 skipped), tsc --noEmit clean, build:types clean (36/36). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T89ffUgYbtp6UztpEXWyVW
…-phase2-mbeczk Implement per-port backpressure gating for no-accumulation passthrough edges
…backing Restore end-to-end streaming for kind: "private" tasks. Streaming lives in the blob sidecar, not the tier, so FsFolderTaskOutputRepository now also implements the run-scoped *ForRun contract (rows + streaming) by folding the runId into the taskType axis — the same sidecar path serves both the deterministic and the private cache tier, and a run's rows and blobs stay prefix-selectable for cleanup. - TaskOutputRepository: add optional saveOutputStreamForRun / saveOutputStreamPortForRun to the base surface. - FsFolderTaskOutputRepository: implement saveOutputForRun / getOutputForRun / saveOutputStreamForRun / saveOutputStreamPortForRun / deleteRun / deleteRunOlderThan / sizeForRun via a `__run:<runId>::` namespace prefix; deleteRun reclaims both output rows and sidecar blobs by that prefix. - RunPrivateCacheRepo: forward the stream writers to the backing's *ForRun methods and forward the opaque by-ref readers, but only when the backing is a run-scoped streaming backing — a tabular (no-sidecar) run-private backing leaves the streaming surface undefined and the tier degrades to accumulation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements comprehensive streaming binary output support and a no-accumulation passthrough path for streaming ports. Tasks can now emit binary data as ordered byte chunks (
binary-deltaevents), which are persisted to a streaming-capable cache backing and referenced viaCacheRefinstead of being accumulated in memory. This enables efficient handling of large binary outputs (audio, video, images) without materializing full payloads.Key Changes
Core Streaming Binary Support
StreamTypes.ts: Addedbinarystream mode alongside existingappend,replace,objectmodes. Binary streams emitbinary-deltaevents carryingUint8Arraychunks that concatenate into a completeBlob/ArrayBuffer.CacheRef.ts(new): Branded reference type for cached outputs, with optionalportandmodediscriminators for streaming ports. Survives JSON serialization across queue rows and IPC boundaries.streamCodec.ts(new): Per-mode codecs that encode/decode streaming deltas to ordered byte streams.append→ UTF-8,object→ NDJSON,binary→ identity bytes.BinaryPortCodec.ts(new): JSON-safe wire form for inline binary port values, preventing silent corruption of cacheable tasks with binary outputs.No-Accumulation Passthrough Path
StreamProcessor.ts: AddedBinaryRefSinkandStreamSinkinterfaces for per-port streaming sinks. When registered, the processor routes a port's deltas directly to the sink (encoded by the port's codec) instead of accumulating in memory. At finish, the sink's returnedCacheRefreplaces the port's slot in the output object.BackpressureGate.ts(new): Cost-agnostic park/wake primitive for producer/consumer backpressure. Producers charge buffered cost; the gate parks them when crossing a high-water mark and releases once consumers credit enough back.StreamPump.ts: AddednoAccumulationflag andreferenceBytesThresholdtoStreamingRunOptions. Enables the passthrough path when conditions are met (streaming-capable cache, binary-only leaf, no value-needing consumer).TaskRunner.ts: Integrated per-port ref-sink path. WiresStreamSinkinstances onto the run config so the runner can choose between accumulation and passthrough per port.Cache Integration
resolveRef.ts(new):CacheRefResolverandCacheRefStreamResolvertypes for by-ref readers.RefStreamBackinginterface carries optional streaming retrieval capabilities.resolveJobOutput.ts(new): Resolves job outputs containingCacheRefvalues by hydrating them through a resolver.CacheCoordinator.ts: AddedCacheReplayContextto drive cache-hit behavior for binary refs: hydrate bytes into enriched finish events for materializing consumers, replay chunkedbinary-deltaevents for stream-capable consumers.RunPrivateCacheRepo.ts: Extended to support streaming output paths viasaveOutputStreamPort.TaskOutputRepository.ts: AddedsaveOutputStreamPort,getOutputByRef,getOutputStreamByRef, andsupportsStreamingReads()methods. Implementations can opt into streaming persistence.FsFolderTaskOutputRepository.ts(new): Filesystem-backed task output repo with streaming support. Persists one output port's codec-encoded byte stream as a sidecar blob.Job Queue Integration
JobQueueClient.ts: AddedonStreamlistener for streaming events during job execution.JobQueueEventListeners.ts: Addedjob_streamevent type andStreamEventLikeinterface.JobQueueWorker.ts/JobQueueServer.ts: Wired streaming event emission through the queue.Comprehensive Test Coverage
StreamBinaryPump.test.ts: Regression guard for binary sources feeding non-binary consumers (must materialize across the edgehttps://claude.ai/code/session_013fSp8GisRRDbtTvbUrS1c5