Add streaming binary output support and cache-ref passthrough

packages/ai/src/task/base/AiTask.ts

@@ -370,7 +370,10 @@ export class AiTask<
   /**
    * Validates that model inputs are valid ModelConfig objects.
    */
-  public override async validateInput(input: Input): Promise<boolean> {
+  public override async validateInput(
+    input: Input,
+    skipPorts?: ReadonlySet<string>
+  ): Promise<boolean> {
     const inputSchema = this.inputSchema();
     if (typeof inputSchema === "boolean") {
       if (inputSchema === false) {
@@ -422,7 +425,7 @@ export class AiTask<
       }
     }
 
-    return super.validateInput(input);
+    return super.validateInput(input, skipPorts);
   }
 
   public override async narrowInput(input: Input, registry: ServiceRegistry): Promise<Input> {

packages/ai/src/task/generation/ImageEditTask.ts

@@ -93,8 +93,11 @@ export class ImageEditTask extends AiImageOutputTask<ImageEditTaskInput, ImageEd
     return ImageEditOutputSchema as DataPortSchema;
   }
 
-  public override async validateInput(input: ImageEditTaskInput): Promise<boolean> {
-    const ok = await super.validateInput(input);
+  public override async validateInput(
+    input: ImageEditTaskInput,
+    skipPorts?: ReadonlySet<string>
+  ): Promise<boolean> {
+    const ok = await super.validateInput(input, skipPorts);
     if (!ok) return false;
     await this.validateProviderImageInput(input);
     return true;

packages/ai/src/task/generation/ImageGenerateTask.ts

@@ -66,8 +66,11 @@ export class ImageGenerateTask extends AiImageOutputTask<
     return ImageGenerateOutputSchema as DataPortSchema;
   }
 
-  public override async validateInput(input: ImageGenerateTaskInput): Promise<boolean> {
-    const ok = await super.validateInput(input);
+  public override async validateInput(
+    input: ImageGenerateTaskInput,
+    skipPorts?: ReadonlySet<string>
+  ): Promise<boolean> {
+    const ok = await super.validateInput(input, skipPorts);
     if (!ok) return false;
     await this.validateProviderImageInput(input);
     return true;

packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts

@@ -159,8 +159,7 @@ export class IndexedDbQueueStorage<Input, Output> implements IQueueStorage<Input
     return new Promise((resolve, reject) => {
       request.onsuccess = () => {
         const job = request.result as
-          | (JobStorageFormat<Input, Output> & Record<string, unknown>)
-          | undefined;
+          (JobStorageFormat<Input, Output> & Record<string, unknown>) | undefined;
         if (job && job.queue === this.queueName && this.matchesPrefixes(job)) {
           resolve(job);
         } else {
@@ -416,8 +415,7 @@ export class IndexedDbQueueStorage<Input, Output> implements IQueueStorage<Input
       const getReq = store.get(job.id as string);
       getReq.onsuccess = () => {
         const existing = getReq.result as
-          | (JobStorageFormat<Input, Output> & Record<string, unknown>)
-          | undefined;
+          (JobStorageFormat<Input, Output> & Record<string, unknown>) | undefined;
         if (!existing || existing.queue !== this.queueName || !this.matchesPrefixes(existing)) {
           // Contract: complete() must silently no-op (not throw) when the row is
           // missing or belongs to another queue. Lease-expiry races can legitimately
@@ -654,8 +652,7 @@ export class IndexedDbQueueStorage<Input, Output> implements IQueueStorage<Input
     return new Promise((resolve, reject) => {
       request.onsuccess = () => {
         const job = request.result as
-          | (JobStorageFormat<Input, Output> & Record<string, unknown>)
-          | undefined;
+          (JobStorageFormat<Input, Output> & Record<string, unknown>) | undefined;
         if (job && this.matchesPrefixes(job)) {
           resolve(job.output ?? null);
         } else {

packages/job-queue/src/job/Job.ts

@@ -6,7 +6,7 @@
 
 import { JobStatus } from "../queue-storage/IQueueStorage";
 import { JobError } from "./JobError";
-import type { JobProgressListener } from "./JobQueueEventListeners";
+import type { JobProgressListener, StreamEventLike } from "./JobQueueEventListeners";
 
 export { JobStatus };
 
@@ -20,6 +20,12 @@ export interface IJobExecuteContext {
     message?: string,
     details?: Record<string, any> | null
   ) => Promise<void>;
+  /**
+   * OPTIONAL. Present only when the worker's transport can deliver stream
+   * events. Jobs MUST NOT retain references to chunk buffers after calling
+   * this (buffers may be transferred across a worker boundary and detached).
+   */
+  emitStreamEvent?: (event: StreamEventLike) => void;
 }
 
 /**

packages/job-queue/src/job/JobQueueClient.ts

@@ -26,6 +26,8 @@ import {
   JobQueueEventListeners,
   JobQueueEventParameters,
   JobQueueEvents,
+  JobStreamListener,
+  type StreamEventLike,
 } from "./JobQueueEventListeners";
 import type { JobQueueServer } from "./JobQueueServer";
 import { storageToClass } from "./JobStorageConverters";
@@ -38,15 +40,50 @@ export interface JobHandle<Output> {
   waitFor(): Promise<Output>;
   abort(): Promise<void>;
   onProgress(callback: JobProgressListener): () => void;
+  /**
+   * OPTIONAL — present only when this handle's transport can deliver stream
+   * events (a same-process server-attached queue). Absent on storage-only
+   * backends; callers branch on `typeof handle.onStream === "function"`.
+   */
+  onStream?(callback: JobStreamListener): () => void;
+  /**
+   * OPTIONAL — present only when the client was configured with an
+   * `outputStreamResolver` (an output cache backing reachable from this
+   * process). Awaits the job's completion, then streams the binary result
+   * back out of the output cache without materializing it. `port` selects the
+   * output port; omit it when the output holds exactly one cache ref.
+   * Resolves `undefined` when there is nothing binary to stream (or the
+   * cache entry was evicted).
+   */
+  outputStream?(port?: string): Promise<AsyncIterable<Uint8Array> | undefined>;
 }
 
+/**
+ * Resolves a completed job's output value to a byte stream. Injected via
+ * {@link JobQueueClientOptions.outputStreamResolver} because the cache layer
+ * that understands output refs lives above this package in the dependency
+ * graph (`@workglow/task-graph` exports `makeJobOutputStreamResolver` to
+ * build one from a cache backing).
+ */
+export type JobOutputStreamResolver = (
+  output: unknown,
+  port?: string
+) => Promise<AsyncIterable<Uint8Array> | undefined>;
+
 /**
  * Options for creating a JobQueueClient
  */
 export interface JobQueueClientOptions<Input, Output> {
   readonly messageQueue: IMessageQueue<JobStorageFormat<Input, Output>>;
   readonly jobStore: IJobStore<Input, Output>;
   readonly queueName: string;
+  /**
+   * OPTIONAL — enables `JobHandle.outputStream` on handles from this client.
+   * Deployments whose output cache backing is reachable from this process
+   * inject a resolver (see `makeJobOutputStreamResolver` in
+   * `@workglow/task-graph`); without it, handles omit the method.
+   */
+  readonly outputStreamResolver?: JobOutputStreamResolver;
 }
 
 /**
@@ -61,6 +98,7 @@ export class JobQueueClient<Input, Output> {
   protected readonly events = new EventEmitter<JobQueueEventListeners<Input, Output>>();
   protected server: JobQueueServer<Input, Output> | null = null;
   protected storageUnsubscribe: (() => void) | null = null;
+  protected readonly outputStreamResolver: JobOutputStreamResolver | undefined;
 
   /**
    * Map of job IDs to their pending promise resolvers
@@ -78,6 +116,11 @@ export class JobQueueClient<Input, Output> {
    */
   protected readonly jobProgressListeners: Map<unknown, Set<JobProgressListener>> = new Map();
 
+  /**
+   * Map of job IDs to their stream listeners
+   */
+  protected readonly jobStreamListeners: Map<unknown, Set<JobStreamListener>> = new Map();
+
   /**
    * Last known progress state for each job
    */
@@ -94,6 +137,7 @@ export class JobQueueClient<Input, Output> {
     this.queueName = options.queueName;
     this.messageQueue = options.messageQueue;
     this.jobStore = options.jobStore;
+    this.outputStreamResolver = options.outputStreamResolver;
   }
 
   /**
@@ -431,6 +475,27 @@ export class JobQueueClient<Input, Output> {
     };
   }
 
+  /**
+   * Subscribe to stream events for a specific job
+   */
+  public onJobStream(jobId: unknown, listener: JobStreamListener): () => void {
+    if (!this.jobStreamListeners.has(jobId)) {
+      this.jobStreamListeners.set(jobId, new Set());
+    }
+    const listeners = this.jobStreamListeners.get(jobId)!;
+    listeners.add(listener);
+
+    return () => {
+      const listeners = this.jobStreamListeners.get(jobId);
+      if (listeners) {
+        listeners.delete(listener);
+        if (listeners.size === 0) {
+          this.jobStreamListeners.delete(jobId);
+        }
+      }
+    };
+  }
+
   // ========================================================================
   // Event handling
   // ========================================================================
@@ -564,23 +629,60 @@ export class JobQueueClient<Input, Output> {
     }
   }
 
+  /**
+   * Called by server when a job emits a stream event. Listener throws are
+   * isolated per-listener — one misbehaving subscriber does not interrupt
+   * delivery to the rest or abort the dispatch itself.
+   * @internal
+   */
+  public handleJobStream(jobId: unknown, event: StreamEventLike): void {
+    this.events.emit("job_stream", this.queueName, jobId, event);
+
+    const listeners = this.jobStreamListeners.get(jobId);
+    if (!listeners) return;
+    for (const listener of listeners) {
+      try {
+        listener(event);
+      } catch (err) {
+        getLogger().error("JobHandle.onStream listener threw", {
+          jobId,
+          error: err instanceof Error ? err.message : String(err),
+        });
+      }
+    }
+  }
+
   // ========================================================================
   // Private helpers
   // ========================================================================
 
   private createJobHandle(id: unknown): JobHandle<Output> {
-    return {
+    const handle: JobHandle<Output> = {
       id,
       waitFor: () => this.waitFor(id),
       abort: () => this.abort(id),
       onProgress: (callback: JobProgressListener) => this.onJobProgress(id, callback),
     };
+    // Stream delivery requires a same-process server-attached transport — the
+    // same signal `connect()` uses. Storage-only backends omit `onStream`, so
+    // callers branch on `typeof handle.onStream === "function"`.
+    if (this.server) {
+      handle.onStream = (callback: JobStreamListener) => this.onJobStream(id, callback);
+    }
+    // Streaming result reads require a cache backing reachable from this
+    // process; the injected resolver is the capability signal.
+    const resolver = this.outputStreamResolver;
+    if (resolver) {
+      handle.outputStream = async (port?: string) => resolver(await this.waitFor(id), port);
+    }
+    return handle;
   }
 
   private cleanupJob(jobId: unknown): void {
     this.activeJobPromises.delete(jobId);
     this.lastKnownProgress.delete(jobId);
     this.jobProgressListeners.delete(jobId);
+    this.jobStreamListeners.delete(jobId);
   }
 
   private handleStorageChange(change: QueueChangePayload<Input, Output>): void {

packages/job-queue/src/job/JobQueueEventListeners.ts

@@ -25,6 +25,7 @@ export type JobQueueEventListeners<Input, Output> = {
     message: string,
     details: Record<string, any> | null
   ) => void;
+  job_stream: (queueName: string, jobId: unknown, event: StreamEventLike) => void;
 };
 
 export type JobQueueEvents = keyof JobQueueEventListeners<any, any>;
@@ -46,3 +47,16 @@ export type JobProgressListener = (
   message: string,
   details: Record<string, any> | null
 ) => void;
+
+/**
+ * Minimal structural shape of a stream event crossing the job-queue boundary.
+ *
+ * `@workglow/job-queue` sits below `@workglow/task-graph` in the dependency
+ * graph, so it cannot import task-graph's `StreamEvent`. This structural type
+ * captures just what the queue plumbing needs; task-graph's `StreamEvent` is
+ * assignable to it, so real stream producers interoperate transparently.
+ */
+export type StreamEventLike = { type: string; port?: string; [k: string]: unknown };
+
+/** Listener for cross-process stream events emitted by an executing job. */
+export type JobStreamListener = (event: StreamEventLike) => void;

packages/job-queue/src/job/JobQueueServer.ts

@@ -14,6 +14,7 @@ import { JobStatus } from "../queue-storage/IQueueStorage";
 import type { DeadLetter } from "./DeadLetter";
 import { Job, JobClass } from "./Job";
 import { JobQueueClient } from "./JobQueueClient";
+import type { StreamEventLike } from "./JobQueueEventListeners";
 import { JobQueueWorker } from "./JobQueueWorker";
 import { storageToClass } from "./JobStorageConverters";
 
@@ -57,6 +58,7 @@ export type JobQueueServerEventListeners<Input, Output> = {
     message: string,
     details: Record<string, unknown> | null
   ) => void;
+  job_stream: (queueName: string, jobId: unknown, event: StreamEventLike) => void;
 };
 
 export type JobQueueServerEvents = keyof JobQueueServerEventListeners<unknown, unknown>;
@@ -501,6 +503,11 @@ export class JobQueueServer<
       this.forwardToClients("handleJobProgress", jobId, progress, message, details);
     });
 
+    worker.on("job_stream", (jobId, event) => {
+      this.events.emit("job_stream", this.queueName, jobId, event);
+      this.forwardToClients("handleJobStream", jobId, event);
+    });
+
     return worker;
   }
 
@@ -524,6 +531,11 @@ export class JobQueueServer<
     message: string,
     details: Record<string, unknown> | null
   ): void;
+  protected forwardToClients(
+    method: "handleJobStream",
+    jobId: unknown,
+    event: StreamEventLike
+  ): void;
   protected forwardToClients(method: string, ...args: unknown[]): void {
     for (const client of this.clients) {
       const fn = (client as any)[method];

packages/job-queue/src/job/JobQueueWorker.ts

@@ -31,6 +31,7 @@ import {
   RetryableJobError,
 } from "./JobError";
 import { withJobErrorDiagnostics } from "./JobErrorDiagnostics";
+import type { StreamEventLike } from "./JobQueueEventListeners";
 import { storageToClass } from "./JobStorageConverters";
 
 /**
@@ -71,6 +72,7 @@ export type JobQueueWorkerEventListeners<Input, Output> = {
     message: string,
     details: Record<string, unknown> | null
   ) => void;
+  job_stream: (jobId: unknown, event: StreamEventLike) => void;
   worker_start: () => void;
   worker_stop: () => void;
 };
@@ -856,6 +858,7 @@ export class JobQueueWorker<
     return await job.execute(job.input, {
       signal,
       updateProgress: this.updateProgress.bind(this, job.id),
+      emitStreamEvent: (event) => this.emitStreamEvent(job.id, event),
     });
   }
 
@@ -877,6 +880,17 @@ export class JobQueueWorker<
     this.events.emit("job_progress", jobId, progress, message, details);
   }
 
+  /**
+   * Emit a cross-process stream event for a job.
+   *
+   * Mirrors {@link updateProgress}: stream events are delivered in-memory via
+   * the `job_stream` event and forwarded by an attached `JobQueueServer` to
+   * subscribed clients. Storage is not touched.
+   */
+  protected emitStreamEvent(jobId: unknown, event: StreamEventLike): void {
+    this.events.emit("job_stream", jobId, event);
+  }
+
   /** Internal — resolve the active claim for a job id, throw if missing. */
   private getClaim(jobId: unknown): IClaim<JobStorageFormat<Input, Output>> | undefined {
     return this.activeClaims.get(jobId);

packages/storage/src/tabular/HttpTabularProxyStorage.ts

@@ -44,8 +44,7 @@ export interface HttpTabularProxyOptions<
   readonly schema: Schema;
   readonly primaryKey: PrimaryKeyNames;
   readonly indexes?: readonly (
-    | keyof Schema["properties"]
-    | readonly (keyof Schema["properties"])[]
+    keyof Schema["properties"] | readonly (keyof Schema["properties"])[]
   )[];
   /** Optional base path. Defaults to `/api/storage`. Trailing slashes are stripped. */
   readonly basePath?: string;