Full stream - #621
Merged
Merged
Conversation
Coverage Report
File CoverageNo changed files found. |
Collaborator
Author
Automated review — must-fix findingsFindings from an automated review of this PR: 3 HIGH. H1 — Fix:
H2 — Fix:
H3 — StreamMode
— posted by scheduled review routine Generated by Claude Code |
sroussey
force-pushed
the
full-stream
branch
2 times, most recently
from
August 5, 2026 23:48
8a6087c to
2226ee7
Compare
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.
The column definitions and the ON CONFLICT clause already quote identifiers, but the CREATE TABLE PRIMARY KEY clause did not — so a schema with a camelCase primary-key column (e.g. TaskOutputSchema's taskType) created a quoted "taskType" column while the PK clause referenced the folded tasktype, failing with 'column "tasktype" named in key does not exist'. Quote the PK list consistently; snake_case schemas are unaffected. Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
TabularBlobChunkStore persists a ref's bytes as ordered (refKey, seq, bytes)
chunk rows plus a manifest row over any ITabularStorage, with keyset-paged
bounded-memory reads. TabularStreamingTaskOutputRepository is the shared base
(rows + blob store + <scheme>://<refKey> refs); Streaming{Postgres,Sqlite}
TaskOutputRepository are thin subclasses over bytea / BLOB chunk tables. Tested
with InMemory (store logic), in-process PGlite (Postgres), and better-sqlite3.
Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
Parity with the PostgresTabularStorage fix: the exec_sql bootstrap DDL quoted the column definitions but not the PRIMARY KEY clause, so a camelCase-PK schema created a quoted column while the PK referenced the folded lowercase name. Only the local-dev/test exec_sql bootstrap path is affected (production tables are owned by Supabase migrations); snake_case schemas are unaffected. Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
StreamingSupabaseTaskOutputRepository is a thin TabularStreamingTaskOutputRepository subclass over SupabaseTabularStorage (bytea chunk tables via TabularBlobChunkStore), tested through the PGlite-backed Supabase mock. Two mock fixes make a camelCase schema faithful to real PostgREST (which quotes identifiers): quote the ON CONFLICT target in upsert, and map delete-chain neq(col, null) to IS NOT NULL so deleteAll actually clears. No external Supabase required. Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
…itory Adds three optional runId-scoped by-ref methods to the base interface: getOutputByRefForRun, getOutputStreamByRefForRun, deleteOutputByRefForRun. Each carries a JSDoc contract mirroring the base by-ref idempotency: a ref not produced by a *ForRun writer for the given runId resolves to undefined (readers) or is a no-op (delete). Interface only, no implementations touched. Claude-Session: https://claude.ai/code/session_019nPCEAuvb2Np5zqB6JLRna
Adds getOutputByRefForRun / getOutputStreamByRefForRun / deleteOutputByRefForRun to FsFolderTaskOutputRepository. Each parses the ref through REF_PATTERN and requires the captured blob name to start with sanitize(runScopePrefix(runId)); otherwise the reader returns undefined and the delete is a no-op. Shared sidecar-reading logic factored into a small private helper (blobPathInRunScope). The RunPrivateCacheRepo wrapper still forwards through the unscoped by-ref methods; the follow-up commit rewires it to close the cross-run leak. Claude-Session: https://claude.ai/code/session_019nPCEAuvb2Np5zqB6JLRna
…prevent cross-run leak RunPrivateCacheRepo previously forwarded getOutputByRef / getOutputStreamByRef / deleteOutputByRef straight to the backing's runId-agnostic methods, so a CacheRef minted by one run's wrapper resolved (and could be deleted) through another run's wrapper on the same backing — the private tier's whole point is that runs cannot see each other, and this made them observable and mutable across the boundary. Rewire the wrapper to route by-ref reads and deletes exclusively through the new *ForRun variants, threading its own runId. A foreign ref (another run's blob, an unscoped deterministic write, malformed / foreign-scheme) resolves to undefined at the backing and delete becomes a no-op — matching the base contract's cache-miss idempotency. The unscoped variants are no longer touched by the wrapper. New tests cover cross-run read/delete rejection, positive same-run regression, malformed / foreign-scheme rejection, and a strengthened clearRun() assertion that a run's ref does not resolve through another run's wrapper after cleanup. Claude-Session: https://claude.ai/code/session_019nPCEAuvb2Np5zqB6JLRna
…llision
The sanitize function collapses `:` to `-`, which meant one runId's sanitized
scope prefix could be a strict prefix of another's — e.g. sanitize("__run:a::")
== "__run-a--" is a prefix of sanitize("__run:a-::") == "__run-a---".
Rewriting runScopePrefix to use a netstring length-prefix ("__run:${len}:${runId}::")
forces two distinct runIds to diverge at the length digits (which sanitize to
themselves) before either one's content is compared, so no sanitized prefix
can ever be a strict prefix of another's.
Add a `prefix-boundary collision` describe block in RunPrivateFsFolderStream.test.ts
covering three flavors of the pre-fix collision. Each case verifies that the
victim's wrapper cannot read the attacker's ref, cannot delete it (silent no-op,
blob count unchanged), and that the attacker can still round-trip its own ref.
…t run start If the CacheRegistry's `private` and `deterministic` slots point at the same backing repository instance, `TaskRunner.hydrateInputRefs`' private-then- deterministic fallback lets a foreign-run ref that the RunPrivateCacheRepo wrapper correctly rejects still resolve through the unscoped deterministic reader, reopening the cross-run leak the wrapper exists to close. Expose `RunPrivateCacheRepo.backing` as a read-only getter (rename the private field to `_backing`) and add a config-time guard in `TaskRunner.handleStart`: when `cacheRegistry.private instanceof RunPrivateCacheRepo` and `private.backing === deterministic`, throw `TaskConfigurationError` before any input hydration for the run. Document the invariant on `CacheRegistry`; the same-folder-path-but-different-instance variant is a residual, currently undetected case.
… streamable ports Portless discovery used to deep-walk the ENTIRE output and stream any CacheRef it found. Because a job may copy content from untrusted input into arbitrary output fields, a crafted branded ref shape smuggled through a non-streaming port would be resolved against the backing — a whole-value side-channel. Gate portless discovery on the task's output schema. When the caller doesn't pass a port, `resolveJobOutputStream` / `makeJobOutputStreamResolver` now require a `DataPortSchema` and only enumerate ports the schema declares streamable via `x-stream`. Zero refs at declared ports resolves undefined; >1 throws the existing "explicit port" error. Portless discovery without a schema throws TypeError up front rather than silently degrading to the deep walk. Update resolveJobOutputStream.test.ts accordingly and add cases for the schema-required error, ignoring a ref at a non-streamable port, and single- port auto-discovery.
`push()` could interleave with `end()` / `fail()` such that a chunk was appended to the buffer AFTER the gate closed — the fast-path closed check ran first, then another turn ran end(), then push appended and woke the consumer, leaking a post-end chunk to the sink. The consumer iterable also drained the buffer BEFORE checking `gate.failure`, so a fail() mid-stream delivered a partially-buffered payload to the sink and only surfaced the error after — the sink treated a truncated payload as complete. Recheck `gate.closed` after the buffer append and un-stage the chunk if the router closed in the meantime; and hoist the `gate.failure` check above the buffer drain so a failure surfaces to the sink before any already-buffered chunk. Add BinaryStreamRouterRace.test.ts with four cases: discard-after-end, 100-microtask interleaving with a never-leaked post-end chunk, fail() surfacing to the sink even with a buffered chunk, and double end() idempotency.
…chdog liveness The no-accumulation passthrough gate could park a producer indefinitely: nothing in the graph runner released it on ctx.abortController.abort(), and nothing surfaced a wedged consumer as an error. A dead consumer therefore wedged the producer's push() forever — the run appeared alive but made no progress. Thread the graph's RunContext into buildPassthroughEdgeGates and, per gate, register an abort listener that closes the gate immediately and arm a watchdog timer that fails the gate when neither a pull nor a credit progresses within streamGateWatchdogMs (default 60_000 ms; `0` disables). The consumer-side wrapper rearms the watchdog on every pull and credit, so a live consumer keeps resetting the deadline. In runStreamingTask's finally, the abort listener is removed and the watchdog cleared so a terminated run never leaks listeners or timers. Make BackpressureGate.charge/awaitBelowMark/park propagate a stored failure: a fail() on a parked producer now rejects its promise (previously silently resolved), and post-fail charges reject immediately. Producers must see the watchdog / abort error, not silently resume as if the buffer drained. Add StreamPumpPassthroughLiveness.test.ts covering the gate primitives (close / fail / credit / post-fail charges) and an end-to-end LiveConsumer run under a strict watchdog + the streamGateWatchdogMs=0 disable path + the noAccumulation-off regression.
postStreamChunk now extracts transferables (mirroring postResult) so binary stream payloads — binary-delta buffers and snapshot image bytes — move zero-copy from worker to main thread instead of being structure-cloned. Non-binary events (text-delta, object-delta, phase, finish) yield an empty transfer list and clone byte-for-byte exactly as before. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
Add optional publishStreamChunk/subscribeToStream to IMessageQueue and IQueueStorage (capability-gated like subscribeToChanges), the StreamChunkRow row type, and a StreamReassembler that orders rows by monotonic per-job seq, buffers gaps, and drops duplicates — the client-side ordering unit for the cross-process stream side-channel. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
Implement publishStreamChunk/subscribeToStream on InMemoryQueueStorage: a per-job ordered append log plus live subscribers, with seq>sinceSeq replay for late/reconnecting subscribers and eviction of a job's stream on row deletion. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
emitStreamEvent assigns a monotonic per-job seq synchronously and best-effort publishes each event to a channel-capable message queue, keeping the in-memory fast path unchanged; the seq counter clears on job cleanup. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
…nto onStream A client with no attached server on a channel-capable queue lazily opens a subscribeToStream subscription per job with a stream listener, feeds rows through a StreamReassembler, and dispatches ordered events through the existing handleJobStream path. onStream is now exposed whenever the server OR the queue's stream channel can deliver; server-attached clients keep the direct fast path (no double-delivery). Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
Describe the worker-thread transferable path and the channel-capable IMessageQueue stream side-stream (publishStreamChunk/subscribeToStream, per-job seq, StreamReassembler reassembly, onStream capability gating), the InMemory reference carrier, and the deferred durable carriers. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
A stream emits many payloads over a job's life. Transferring a partial view — a subarray chunk, or a small Node Buffer carved from the shared allocation pool — detaches a backing buffer that a later chunk (or worker-retained state) still aliases, throwing DataCloneError on the next emit or neutering the worker's Buffer pool. extractTransferables gains an ownedBuffersOnly mode that transfers a TypedArray's buffer only when the view spans the whole buffer; postStreamChunk uses it, so partial views clone byte-for-byte instead. postResult keeps the default. Fresh full buffers (decoded images) still transfer zero-copy. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
- Client resumes from the last delivered seq on re-subscribe: a listener removed and re-added while a job is still running no longer replays the whole log and double-delivers every prior event (a per-job cursor, persisted across teardown, drives sinceSeq). - TelemetryQueueStorage forwards the optional publishStreamChunk/subscribeToStream so a telemetry-wrapped channel-capable storage keeps onStream (decorator transparency, matching findActiveByFingerprint). - attach()/detach() reconcile channel subscriptions: attaching a server tears down channel subs (fast path takes over — no double-delivery); detaching re-opens them for jobs with listeners. - WrappedMessageQueue stream forwarders are readonly (assigned once). - Document that the per-worker seq restarts on a cross-worker retry (deferred to the durable carriers, where the store assigns seq); single-attempt and the InMemory reference are unaffected. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
Turns the three deferred code-review findings into real fixes: - Carrier-assigned seq: publishStreamChunk(jobId, event) — the carrier owns the per-job seq counter, not the worker. A retry claimed by a different worker continues the same job's sequence instead of restarting at 1 and colliding with the prior attempt (which the subscriber's reassembler would drop). Drops the worker-side streamSeq map. - Reassembler skip-forward: once more than maxGapBuffer rows pile up ahead of a missing seq, the dropped seq is skipped (logged) and delivery resumes, so a lost row on a best-effort carrier can't stall the stream forever or grow the gap buffer unbounded. - Teardown decoupled from completion: the channel subscription tears down on the terminal finish/error event (driven by the stream itself); on job completion a grace window defers teardown so a trailing event still in flight on an async carrier — which raced and lost to the storage-completion signal — is not dropped. attach() cancels a pending grace timer. Claude-Session: https://claude.ai/code/session_01Cp2W5rBaUyT9mtQCWcuMbJ
Replaying the streaming branch commit-by-commit onto current main let a few superseding hunks land on drifted context, leaving earlier-commit state in the final tree. Align those files with the branch's pre-rebase content: - FsFolderTaskOutputRepository: drop the ns-prefix cascade overrides (superseded by run-scoped deleteRun/blob sidecar cleanup) and their tests in CacheStreamOut.test - TaskOutputRepositoryStream.test: run-private wrapper no longer exposes the deterministic streaming sink - StreamPump: release edge streams on ABORTING as well - CacheCoordinator: hoist the instance-schema lookup out of the try so the stream-ref replay path can read it - TaskRunner/JobQueueServer: import tidy-ups Claude-Session: https://claude.ai/code/session_015uTj4xMYarAPCUpo4UCL9w
- force accumulation when a streamable task's cache sinks resolve
undefined (policy-routed slot missing or stream-wired downgrade), so
delta output is never silently dropped
- propagate cache-sink failures to the binary stream router's gate and
reject post-failure pushes, so a dead sink fails the run instead of
wedging a parked producer
- passthrough watchdog re-arms instead of failing when the producer is
not parked (slow-but-live producers no longer trip it)
- replace-mode refusal streams complete with output.refusal again
instead of throwing replace-mode-no-value
- cache-hit stream replay degrades decode/read errors to a cache miss,
and releases every opened stream on partial failure (fd leak)
- format-less binary ports serialize through the mode-derived codec so
Blobs no longer persist as '{}' rows
- compound runners (GraphAsTask, Iterator/Map/Reduce, While, Fallback)
forward noAccumulation/streamHighWaterBytes/streamGateWatchdogMs into
subgraph runs
- collapse binaryHighWaterBytes into streamHighWaterBytes and the dual
sink deps into one StreamSink map; length-based gate costs replace the
per-delta re-encode and the derivable pendingCosts FIFO
- clear stale runner.inputStreams when a run has no live streaming edges
- hydrate input CacheRefs only at ports whose schema admits one
Claude-Session: https://claude.ai/code/session_015uTj4xMYarAPCUpo4UCL9w
- advance the per-job stream cursor with the delivered row's real seq (fast path advances in emission order), so re-subscribes resume from the true position instead of replaying delivered events - serialize per-job publishStreamChunk calls so async carriers cannot assign inverted seqs to in-flight publishes - forward publishStreamChunk/subscribeToStream through InMemoryMessageQueue so createInMemoryQueue-built queues stream - disconnect() finalizes stream subscriptions/listeners/cursors instead of re-opening channel subscriptions that could never be torn down - detect teardown fired during a synchronous terminal replay inside subscribeToStream and unsubscribe the just-created subscription instead of registering a zombie - server-attached clients subscribe to a channel-capable carrier too (channel authoritative; fast path suppressed while open), so jobs claimed by another process's worker still stream - terminal stream rows stamp a retention deadline on the InMemory log, swept lazily, bounding retained stream bytes - document that portless outputStream() requires a schema-built resolver; tighten the two-ref test assertion accordingly Claude-Session: https://claude.ai/code/session_015uTj4xMYarAPCUpo4UCL9w
- hex-encode the runId inside the FsFolder run-scope prefix so sanitize is the identity on it: distinct runIds can never collide into one blob namespace (closes the same-length sanitize collision) and row/blob comparisons now agree - hoist the anti-traversal ref-key ceremony (sanitize, mintRefKey, makeRefPattern) into CacheRef so the contract lives once; use uuid4 - refuse detached/empty buffers in the worker transferable scan and fall back to a clone post when a transfer fails; document that emitted fully-owned binary payloads are transferred - IdbBlobChunkStore: clean up chunk rows when a stream write fails, prune orphaned chunks lacking a manifest, and batch chunk writes into paged transactions; TabularBlobChunkStore batches pages via putBulk - share one bytesToBase64/base64ToBytes pair from @workglow/util (block- wise browser encode) between BinaryPortCodec and imageCacheCodec - retype the tabular streaming base on a structural IBlobChunkStore and reduce StreamingIndexedDbTaskOutputRepository to a thin subclass - collapse FsFolder's duplicated ByRef/ForRun method bodies into shared path-resolving helpers; use openIdb for IDB setup - drop the unused supportsStreamingReads probe and the never-passed resolveOutput filter option Claude-Session: https://claude.ai/code/session_015uTj4xMYarAPCUpo4UCL9w
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
1 participant
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.
No description provided.