fix(libs): janitor live-run guard, Container dispose-on-replace, vector default parity, StreamPump cancel

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

@@ -14,7 +14,7 @@ import type {
 } from "@workglow/job-queue";
 import { JobStatus, validateLeaseMs } from "@workglow/job-queue";
 import { HybridSubscriptionManager } from "@workglow/storage";
-import { createServiceToken, deepEqual, makeFingerprint, uuid4 } from "@workglow/util";
+import { createServiceToken, deepEqual, getLogger, makeFingerprint, uuid4 } from "@workglow/util";
 import { IndexedDbMigrationRunner } from "../migrations/IndexedDbMigrationRunner";
 import { indexedDbQueueMigrations } from "../migrations/indexedDbQueueMigrations";
 import { openIdb } from "../storage/openIdb";
@@ -419,9 +419,10 @@ export class IndexedDbQueueStorage<Input, Output> implements IQueueStorage<Input
           | (JobStorageFormat<Input, Output> & Record<string, unknown>)
           | undefined;
         if (!existing || existing.queue !== this.queueName || !this.matchesPrefixes(existing)) {
-          reject(
-            new Error(`Job ${job.id} not found or does not belong to queue ${this.queueName}`)
-          );
+          // Contract: complete() must silently no-op (not throw) when the row is
+          // missing or belongs to another queue. Lease-expiry races can legitimately
+          // make the target row disappear between claim and complete, so the
+          // transaction is allowed to commit with no write and resolve.
           return;
         }
         const currentAttempts = existing.attempts ?? 0;
@@ -673,7 +674,12 @@ export class IndexedDbQueueStorage<Input, Output> implements IQueueStorage<Input
     details: Record<string, any> | null
   ): Promise<void> {
     const job = await this.get(id);
-    if (!job) throw new Error(`Job ${id} not found`);
+    if (!job) {
+      // Contract: progress updates for a missing job silently no-op — the job may
+      // have already completed/been removed, or a lease-expiry race removed it.
+      getLogger().warn("Job not found for progress update", { id });
+      return;
+    }
 
     job.progress = progress;
     job.progress_message = message;

packages/indexeddb/src/storage/IndexedDbVectorStorage.ts

@@ -17,8 +17,10 @@ import {
   getMetadataProperty,
   getVectorProperty,
   matchesFilter,
+  safeEmit,
   validateVectorEntities,
 } from "@workglow/storage";
+import type { EventEmitter } from "@workglow/util";
 import { createServiceToken } from "@workglow/util";
 import type {
   DataPortSchemaObject,
@@ -130,6 +132,10 @@ export class IndexedDbVectorStorage<
     options: VectorSearchOptions<Record<string, unknown>> = {}
   ) {
     assertVectorShape(query, this.vectorDimensions, "query");
+    // Default to 0 to match every SQL backend (Sqlite / Postgres / Supabase /
+    // SqliteAi). Cosine similarity ranges over [-1, 1]; a negatively-correlated
+    // hit is "not a match" for relevance retrieval. Callers that want every
+    // result regardless of correlation pass `scoreThreshold: -Infinity`.
     const { topK = 10, filter, scoreThreshold = 0 } = options;
     const results: Array<Entity & { score: number }> = [];
 
@@ -138,9 +144,10 @@ export class IndexedDbVectorStorage<
     for (const entity of allEntities) {
       // IndexedDB stores TypedArrays natively via structured clone
       const vector = entity[this.vectorPropertyName] as TypedArray;
-      const metadata = this.metadataPropertyName
-        ? (entity[this.metadataPropertyName] as Metadata)
-        : ({} as Metadata);
+      // A present-but-null metadata column value coalesces to `{}` so a filtered
+      // search treats the row as a non-match rather than dereferencing null.
+      const rawMetadata = this.metadataPropertyName ? entity[this.metadataPropertyName] : undefined;
+      const metadata = (rawMetadata ?? {}) as Metadata;
 
       if (filter && !matchesFilter(metadata, filter)) {
         continue;
@@ -160,6 +167,19 @@ export class IndexedDbVectorStorage<
 
     results.sort((a, b) => b.score - a.score);
     const topResults = results.slice(0, topK);
+    // The inherited `events` emitter is typed for the tabular event surface;
+    // `similaritySearch` lives on the vector extension of that surface. The
+    // emitter instance is the same object, so widen the view to a record that
+    // carries the event so it can be emitted type-safely.
+    type SimilaritySearchEvents = {
+      similaritySearch: (query: TypedArray, results: (Entity & { score: number })[]) => void;
+    };
+    safeEmit(
+      this.events as unknown as EventEmitter<SimilaritySearchEvents>,
+      "similaritySearch",
+      query,
+      topResults
+    );
 
     return topResults;
   }

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

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import { getLogger } from "@workglow/util";
 import { JobError } from "./JobError";
 
 /**
@@ -57,9 +58,9 @@ export function registerErrorCodeReconstructor(
       return; // idempotent — same fn, same prefix
     }
 
-    console.warn(
-      `registerErrorCodeReconstructor: overwriting existing reconstructor for prefix "${prefix}"`
-    );
+    getLogger().warn("registerErrorCodeReconstructor: overwriting existing reconstructor", {
+      prefix,
+    });
     reconstructors[existingIdx] = { prefix, reconstructor };
     return;
   }

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

@@ -173,23 +173,7 @@ export class JobQueueClient<Input, Output> {
       readonly timeoutSeconds?: number;
     }
   ): Promise<JobHandle<Output>> {
-    const job: JobStorageFormat<Input, Output> = {
-      queue: this.queueName,
-      input,
-      job_run_id: options?.jobRunId,
-      fingerprint: options?.fingerprint,
-      max_attempts: options?.maxAttempts ?? 10,
-      visible_at:
-        options?.delaySeconds != null
-          ? new Date(Date.now() + options.delaySeconds * 1000).toISOString()
-          : new Date().toISOString(),
-      deadline_at:
-        options?.timeoutSeconds != null
-          ? new Date(Date.now() + options.timeoutSeconds * 1000).toISOString()
-          : null,
-      completed_at: null,
-      status: JobStatus.PENDING,
-    };
+    const job = this.buildJobBody(input, options);
 
     const id = await this.messageQueue.send(job, {
       fingerprint: options?.fingerprint,
@@ -208,21 +192,77 @@ export class JobQueueClient<Input, Output> {
   }
 
   /**
-   * Send multiple jobs to the queue
+   * Send multiple jobs to the queue in a single batched insert.
+   *
+   * Delegates to {@link IMessageQueue.sendBatch} so N jobs become one storage
+   * round-trip and a single worker wake, instead of N inserts + N wakes.
+   *
+   * Per the {@link IMessageQueue.sendBatch} contract, a batch-wide
+   * `fingerprint` is intentionally NOT accepted (it would dedup every body
+   * against the first row); use {@link send} per job for fingerprinted dedup.
+   * The other options — `jobRunId`, `maxAttempts`, `delaySeconds`,
+   * `timeoutSeconds` — apply uniformly to every body and are forwarded (the
+   * previous per-item loop silently dropped `delaySeconds`/`timeoutSeconds`).
    */
   public async sendBatch(
     inputs: readonly Input[],
     options?: {
       readonly jobRunId?: string;
       readonly maxAttempts?: number;
+      /** Delay in seconds before every job becomes visible for processing */
+      readonly delaySeconds?: number;
+      /** Timeout in seconds after which every job's deadline is exceeded */
+      readonly timeoutSeconds?: number;
     }
   ): Promise<readonly JobHandle<Output>[]> {
-    const handles: JobHandle<Output>[] = [];
-    for (const input of inputs) {
-      const handle = await this.send(input, options);
-      handles.push(handle);
+    if (inputs.length === 0) return [];
+
+    const bodies = inputs.map((input) => this.buildJobBody(input, options));
+    const ids = await this.messageQueue.sendBatch(bodies, {
+      jobRunId: options?.jobRunId,
+      maxAttempts: options?.maxAttempts,
+      delaySeconds: options?.delaySeconds,
+      timeoutSeconds: options?.timeoutSeconds,
+    });
+
+    // Single wake for the whole batch — avoids the N-wake thundering herd the
+    // per-item loop caused.
+    this.server?.handleJobAdded(ids[ids.length - 1]);
+
+    return ids.map((id) => this.createJobHandle(id));
+  }
+
+  /**
+   * Build a {@link JobStorageFormat} body from an input + send options. Shared
+   * by {@link send} and {@link sendBatch} so both produce identical rows.
+   */
+  private buildJobBody(
+    input: Input,
+    options?: {
+      readonly jobRunId?: string;
+      readonly fingerprint?: string;
+      readonly maxAttempts?: number;
+      readonly delaySeconds?: number;
+      readonly timeoutSeconds?: number;
     }
-    return handles;
+  ): JobStorageFormat<Input, Output> {
+    return {
+      queue: this.queueName,
+      input,
+      job_run_id: options?.jobRunId,
+      fingerprint: options?.fingerprint,
+      max_attempts: options?.maxAttempts ?? 10,
+      visible_at:
+        options?.delaySeconds != null
+          ? new Date(Date.now() + options.delaySeconds * 1000).toISOString()
+          : new Date().toISOString(),
+      deadline_at:
+        options?.timeoutSeconds != null
+          ? new Date(Date.now() + options.timeoutSeconds * 1000).toISOString()
+          : null,
+      completed_at: null,
+      status: JobStatus.PENDING,
+    };
   }
 
   /**

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

@@ -21,6 +21,14 @@ import { storageToClass } from "./JobStorageConverters";
  * Statistics tracked for the job queue
  */
 export interface JobQueueStats {
+  /**
+   * Total number of execution ATTEMPTS started, not distinct jobs. Incremented
+   * once per `job_start`, which the worker emits on every execution — so a job
+   * that retries N times contributes N to this counter. It therefore will not
+   * equal `completedJobs + failedJobs + disabledJobs` in any run with retries;
+   * use {@link retriedJobs} to reconcile. Renamed semantics, not the field, to
+   * preserve the public stats shape.
+   */
   readonly totalJobs: number;
   readonly completedJobs: number;
   readonly failedJobs: number;
@@ -39,7 +47,7 @@ export type JobQueueServerEventListeners<Input, Output> = {
   server_stop: (queueName: string) => void;
   job_start: (queueName: string, jobId: unknown) => void;
   job_complete: (queueName: string, jobId: unknown, output: Output) => void;
-  job_error: (queueName: string, jobId: unknown, error: string) => void;
+  job_error: (queueName: string, jobId: unknown, error: string, errorCode?: string) => void;
   job_disabled: (queueName: string, jobId: unknown) => void;
   job_retry: (queueName: string, jobId: unknown, visibleAt: Date) => void;
   job_progress: (
@@ -414,6 +422,8 @@ export class JobQueueServer<
 
     // Forward worker events to server and clients
     worker.on("job_start", (jobId) => {
+      // job_start fires once per execution attempt, so totalJobs is
+      // attempt-weighted (see JobQueueStats.totalJobs docs).
       this.stats = { ...this.stats, totalJobs: this.stats.totalJobs + 1 };
       this.events.emit("job_start", this.queueName, jobId);
       this.forwardToClients("handleJobStart", jobId);
@@ -428,7 +438,11 @@ export class JobQueueServer<
       // Immediate deletion when configured
       if (this.deleteAfterCompletionMs === 0) {
         this.jobStore.delete(jobId).catch((err) => {
-          console.error("Error deleting job after completion:", err);
+          getLogger().error("Error deleting job after completion", {
+            error: err,
+            jobId,
+            queueName: this.queueName,
+          });
         });
       }
 
@@ -438,13 +452,17 @@ export class JobQueueServer<
 
     worker.on("job_error", (jobId, error, errorCode) => {
       this.stats = { ...this.stats, failedJobs: this.stats.failedJobs + 1 };
-      this.events.emit("job_error", this.queueName, jobId, error);
+      this.events.emit("job_error", this.queueName, jobId, error, errorCode);
       this.forwardToClients("handleJobError", jobId, error, errorCode);
 
       // Immediate deletion when configured
       if (this.deleteAfterFailureMs === 0) {
         this.jobStore.delete(jobId).catch((err) => {
-          console.error("Error deleting job after error:", err);
+          getLogger().error("Error deleting job after error", {
+            error: err,
+            jobId,
+            queueName: this.queueName,
+          });
         });
       }
 
@@ -460,7 +478,11 @@ export class JobQueueServer<
       // Immediate deletion when configured
       if (this.deleteAfterDisabledMs === 0) {
         this.jobStore.delete(jobId).catch((err) => {
-          console.error("Error deleting job after disabling:", err);
+          getLogger().error("Error deleting job after disabling", {
+            error: err,
+            jobId,
+            queueName: this.queueName,
+          });
         });
       }
 
@@ -562,7 +584,7 @@ export class JobQueueServer<
         await this.jobStore.deleteByStatusAndAge(JobStatus.DISABLED, this.deleteAfterDisabledMs);
       }
     } catch (error) {
-      console.error("Error in cleanup:", error);
+      getLogger().error("Error in cleanup", { error, queueName: this.queueName });
     }
   }