fix(task-graph): security & liveness fixes from PR #608 review (netstring, portless, race, watchdog)#615
Open
sroussey wants to merge 66 commits into
Open
fix(task-graph): security & liveness fixes from PR #608 review (netstring, portless, race, watchdog)#615sroussey wants to merge 66 commits into
sroussey wants to merge 66 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
…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
…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
… backings Filesystem/in-memory streaming backings resolve a cache-miss synchronously via an existence probe (openSync / Map lookup). Async stores (IndexedDB, SQL) cannot, so widen getOutputStreamByRef to optionally return a Promise and await it in the shared streamRefViaBacking helper. A dangling ref now resolves to undefined (a miss) for async backings instead of a truthy Promise mistaken for a live stream. Synchronous backings are unaffected (awaiting a non-Promise is identity). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
Persists a ref's bytes as ordered chunk rows keyed by [refKey, seq] with a manifest row (existence witness + size + createdAt). Streaming writes never accumulate; reads page the chunk store so memory stays bounded and each page runs in its own short-lived transaction. A sibling instance over the same dbName resolves refs (cross-instance read contract). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
A TaskOutputRepository whose JSON rows persist via IndexedDbTabularStorage and whose streamed port payloads persist as chunked blobs (IdbBlobChunkStore), referenced by CacheRef. Implements the full streaming surface (saveOutputStream/saveOutputStreamPort/getOutputByRef/getOutputStreamByRef/ deleteOutputByRef) with async reads; supportsStreaming/Reads/Ports all true and isDurable true. Round-trips every codec mode, keeps distinct ports at distinct refs, prunes blobs with rows, resolves refs cross-instance, and converts a dangling ref into a cache miss through streamRefViaBacking. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
The streaming-read contract now permits an async getOutputStreamByRef, so the base-typed contract suites (deterministic CacheStreamOut and run-private RunPrivateFsFolderStream) must await the reader before consuming it. Runtime is identical for the synchronous FsFolder/in-memory backings (awaiting a non-Promise is identity); the suites still assert the same bytes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
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. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 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.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QehfSG3zCtk6WPebaCuo8R
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 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. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 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. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 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. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> 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. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> 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. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019nPCEAuvb2Np5zqB6JLRna
fix(task-graph): route RunPrivateCacheRepo by-ref through *ForRun to prevent cross-run leak (follow-up to #611)
…-phase3-yg5msj Streaming phase 3
…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.
Coverage Report
File Coverage
|
||||||||||||||||||||||||||||||||||||||
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
Five focused fixes to the task-graph streaming framework surfaced during the PR #608 review. Each fix lands as its own commit with narrow test coverage; the changes stack on the same base so the two CRITICALs and three HIGHs land together.
Fix 1 — CRITICAL: netstring runScopePrefix (sanitize-collision cross-run leak)
sanitize()collapses:→-, sorunScopePrefix("session1")sanitized (__run-session1--) is a strict prefix ofrunScopePrefix("session1-")sanitized (__run-session1---). ARunPrivateCacheRepowrapper for one run then reads/deletes blobs belonging to another.Rewrite
runScopePrefix(runId)to a netstring-length prefix`__run:${runId.length}:${runId}::`— the length digits force any two distinct runIds to diverge before either one's content is compared, so no sanitized prefix can ever be a strict prefix of another's.runScopedTypesimplifies to`${runScopePrefix(runId)}${taskType}`;deleteRun/deleteRunOlderThan/sizeForRun/blobPathInRunScopeall inherit through the same helper.Test: extends
RunPrivateFsFolderStream.test.tswith aprefix-boundary collisiondescribe covering three(victim, attacker)runId pairs; each case verifies the victim's wrapper cannot read the attacker's ref, silently no-ops on delete (blob count unchanged), and the attacker still round-trips its own ref.Fix 2 — CRITICAL: same-backing CacheRegistry misconfiguration guard
If the
CacheRegistry'sprivateanddeterministicslots point at the same backing instance,TaskRunner.hydrateInputRefs' private-then-deterministic fallback lets a foreign-run ref thatRunPrivateCacheRepocorrectly rejects still resolve through the unscoped deterministic reader — reopening the cross-run leak at a different layer.Expose
RunPrivateCacheRepo.backingas a read-only getter (rename the private field to_backing) and add a config-time guard inTaskRunner.handleStart: whencacheRegistry.private instanceof RunPrivateCacheRepo && cacheRegistry.deterministic !== undefined && private.backing === deterministic, throwTaskConfigurationErrorbefore any input hydration. Document the invariant onCacheRegistry; the same-folder-path-but-different-instance variant is a residual, currently undetected case.Test: extends
TaskRunnerInputHydration.test.tswith three cases — throws on same backing, accepts distinct backings, accepts unset deterministic.Fix 3 — HIGH: schema-gated portless outputStream discovery
Portless discovery in
resolveJobOutputStreamused to deep-walk the ENTIRE output and stream anyCacheRefit found. Since a job may copy content from untrusted input into arbitrary output fields, a crafted branded ref shape smuggled through a non-streaming port would resolve against the backing — a whole-value side-channel.Gate portless discovery on the task's output schema.
resolveJobOutputStream/makeJobOutputStreamResolvertake an optionaloutputSchema?: DataPortSchema; portless discovery enumerates only ports the schema declares streamable viax-stream, checking each for aCacheRef(no deep walk). Zero refs at declared ports resolvesundefined; more than one throws the existing "explicit port" error; portless without a schema throwsTypeErrorup front rather than silently degrading.Test:
resolveJobOutputStream.test.ts— schema-required error case, ignore a ref at a non-streamable port, single-port auto-discovery, two-port throws, streaming port with no ref resolves undefined.Fix 4 — HIGH: BinaryStreamRouter push/end race
Two races in
BinaryStreamRouter:push()could interleave withend()/fail()such that a chunk was appended AFTER the gate closed — the fast-path closed check ran first, another turn ran end(), then push appended and woke the consumer, leaking a post-end chunk to the sink.gate.failure, so afail()mid-stream delivered a partial payload to the sink and only surfaced the error after — the sink treated a truncated payload as complete.·R·echeck
gate.closedafter the buffer append and un-stage the chunk if the router closed in the meantime; hoist thegate.failurecheck above the buffer drain in the iterable so a failure surfaces to the sink before any already-buffered chunk.Test: new
BinaryStreamRouterRace.test.ts— discard-after-end, 100-microtask interleaving with never-leaked post-end chunk, fail() surfaces to the sink even with a buffered chunk, doubleend()idempotency.Fix 5 — HIGH: passthrough edge gate liveness (abort + watchdog)
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 wedged the producer'spush()forever — the run appeared alive but made no progress.Thread the graph's
RunContextintobuildPassthroughEdgeGatesand, per gate, register an abort listener that closes the gate immediately and arm a watchdog timer that fails the gate when neither pull nor credit progresses withinstreamGateWatchdogMs(default 60_000 ms;0disables). The consumer-side wrapper rearms the watchdog on every pull and credit. InrunStreamingTask's finally, the abort listener is removed and the watchdog cleared. Also plumb the new option throughTaskGraphRunConfig→TaskGraphRunner→StreamingRunOptions→IRunConfig.Make
BackpressureGate.charge/awaitBelowMark/parkpropagate a stored failure: afail()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.Test: new
StreamPumpPassthroughLiveness.test.tscovering the gate primitives (close / fail / credit / post-fail charge) plus an end-to-end LiveConsumer under a strict watchdog, thestreamGateWatchdogMs=0disable path, and thenoAccumulation-off regression.Note on base
The plan targeted
main, butmainat HEAD does not yet contain the phase-3 streaming code (FsFolderTaskOutputRepository,BackpressureGate,resolveJobOutput, etc.) — the files these fixes touch. This branch is therefore based onb91ddec(PR #608 head, identical tree to the phase-3 branch tip) so the fixes apply cleanly. Merging intomainbrings both the phase-3 content and the security fixes together; PR #608 becomes redundant.Test plan
bunx turbo run build-types --filter=@workglow/task-graph— passes on every commitbunx turbo run build-types --filter=@workglow/test— passes on every commitbun scripts/test.ts graph vitest— 1053 tests / 116 files pass on final HEAD (up from 1036 at base — net +17 across the five new suites: 3 prefix-boundary + 3 same-backing + 3 portless-schema + 4 router-race + 8 liveness / gate primitives)Generated by Claude Code