fix(task-graph,storage): cache restart-resume + SharedInMemory sync barrier

packages/storage/src/tabular/SharedInMemoryTabularStorage.ts

@@ -60,6 +60,14 @@ export class SharedInMemoryTabularStorage<
   private isInitialized = false;
   private syncInProgress = false;
   private pendingMessages: BroadcastMessage[] = [];
+  /**
+   * Resolves when the current `syncFromOtherTabs()` settles, either because a
+   * peer tab sent a `SYNC_RESPONSE` (drained, then resolved) or because the
+   * sync timeout fired. `setupDatabase({ awaitTabSync: true })` awaits this so
+   * callers that immediately read after setup do not race the replay.
+   */
+  private syncSettled: Promise<void> | null = null;
+  private syncSettledResolver: (() => void) | null = null;
 
   constructor(
     channelName: string = "tabular_store",
@@ -151,6 +159,7 @@ export class SharedInMemoryTabularStorage<
         }
         this.syncInProgress = false;
         await this.drainPendingMessages();
+        this.resolveSyncSettled();
         break;
 
       case "PUT":
@@ -196,18 +205,33 @@ export class SharedInMemoryTabularStorage<
     if (!this.channel) return;
 
     this.syncInProgress = true;
+    this.syncSettled = new Promise<void>((resolve) => {
+      this.syncSettledResolver = resolve;
+    });
     this.channel.postMessage({ type: "SYNC_REQUEST" } as BroadcastMessage);
 
     setTimeout(() => {
       if (this.syncInProgress) {
         this.syncInProgress = false;
-        void this.drainPendingMessages().catch((error) => {
-          console.error("Failed to drain pending messages after sync timeout", error);
-        });
+        void this.drainPendingMessages()
+          .catch((error) => {
+            console.error("Failed to drain pending messages after sync timeout", error);
+          })
+          .finally(() => {
+            this.resolveSyncSettled();
+          });
       }
     }, SYNC_TIMEOUT);
   }
 
+  private resolveSyncSettled(): void {
+    const resolver = this.syncSettledResolver;
+    if (resolver) {
+      this.syncSettledResolver = null;
+      resolver();
+    }
+  }
+
   private async copyDataFromArray(entities: Entity[]): Promise<void> {
     if (entities.length === 0) return;
     await this.inMemoryRepo.deleteAll();
@@ -220,12 +244,23 @@ export class SharedInMemoryTabularStorage<
     }
   }
 
-  public override async setupDatabase(): Promise<void> {
+  /**
+   * Initialize the cross-tab sync barrier and apply any pending migrations.
+   *
+   * Defaults to `awaitTabSync: true` so that callers that immediately read or
+   * write after `setupDatabase()` resolves see other tabs' replayed rows; the
+   * fire-and-forget shape regressed in 64591e3 and caused first-call races.
+   * Pass `awaitTabSync: false` only for latency-sensitive startup paths that
+   * explicitly tolerate a stale initial view.
+   */
+  public override async setupDatabase(options: { awaitTabSync?: boolean } = {}): Promise<void> {
     if (this.isInitialized) return;
     this.isInitialized = true;
-    // Fire-and-forget: posts a SYNC_REQUEST and drains responses on a timeout
-    // (see syncFromOtherTabs); there is nothing to await.
+    const awaitTabSync = options.awaitTabSync ?? true;
     this.syncFromOtherTabs();
+    if (awaitTabSync && this.syncSettled) {
+      await this.syncSettled;
+    }
     if (this.tabularMigrations && this.tabularMigrations.length > 0) {
       await this.applyTabularMigrations();
     }

packages/task-graph/src/cache/RunPrivateCacheRepo.ts

@@ -23,17 +23,49 @@ export interface RunPrivateCacheRepoOptions {
  * - Two wrappers with the same `runId` (e.g., a restart after a crash) see each
  *   other's writes via the backing store — that's the restart-survival contract.
  * - Two wrappers with different `runId`s see only their own entries.
+ *
+ * Durable restart-resume requires deterministic task ids. When a task does not
+ * pin `config.id` to a stable value the default constructor mints a fresh v4
+ * UUID per process, so the cache key would change across restarts and orphan
+ * every row. In that case {@link CacheCoordinator} keys by task type instead
+ * and signals the fallback via {@link noteFallbackKey}; this wrapper warns once
+ * per process to tell the operator that the cache is best-effort intra-process
+ * only until task ids are pinned.
  */
 export class RunPrivateCacheRepo extends TaskOutputRepository {
+  private static fallbackWarned = false;
+
   private readonly backing: TaskOutputRepository;
   private readonly runId: string;
+  private observedFallback: boolean = false;
 
   constructor({ backing, runId }: RunPrivateCacheRepoOptions) {
     super({ outputCompression: backing.outputCompression });
     this.backing = backing;
     this.runId = runId;
   }
 
+  /**
+   * Called by {@link CacheCoordinator} the first time it has to key a private
+   * cache entry by task type because the task id is autogenerated. Emits a
+   * single `console.warn` per process so deployments without pinned ids learn
+   * that crash-resume is degraded; subsequent fallbacks are silent.
+   */
+  public noteFallbackKey(): void {
+    this.observedFallback = true;
+    if (RunPrivateCacheRepo.fallbackWarned) return;
+    RunPrivateCacheRepo.fallbackWarned = true;
+    console.warn(
+      "[RunPrivateCacheRepo] Private cache fell back to taskType keying because " +
+        "task.id is autogenerated; pin task.id to enable crash-resume."
+    );
+  }
+
+  /** @internal Test-only hook for inspecting whether any fallback was observed. */
+  public hasObservedFallback(): boolean {
+    return this.observedFallback;
+  }
+
   private ns(cacheIdentity: string): string {
     return `__run:${this.runId}::${cacheIdentity}`;
   }

packages/task-graph/src/task/CacheCoordinator.ts

@@ -7,6 +7,7 @@
 import { getPortCodec } from "@workglow/util";
 import { type CachePolicy, isPolicyCached, isPolicyPrivate } from "../cache/CachePolicy";
 import type { CacheRegistry } from "../cache/CacheRegistry";
+import { RunPrivateCacheRepo } from "../cache/RunPrivateCacheRepo";
 import type { TaskOutputRepository } from "../storage/TaskOutputRepository";
 import type { ITask } from "./ITask";
 import type { StreamEvent } from "./StreamTypes";
@@ -63,11 +64,20 @@ export class CacheCoordinator<Input extends TaskInput, Output extends TaskOutput
    */
   /**
    * Cache identity for the `taskType` axis of {@link TaskOutputRepository}.
-   * Deterministic entries key by task class; private (run-resume) entries key by
-   * task instance id so two nodes of the same type in one graph do not collide.
+   *
+   * Deterministic entries key by task class. Private (run-resume) entries
+   * normally key by task instance id so two nodes of the same type in one
+   * graph do not collide — but only when the id is stable across restarts.
+   * When the id is the autogenerated v4 UUID default (no caller pin), a
+   * restart would mint a fresh id and orphan every prior row, so we fall
+   * back to the task type to preserve a best-effort intra-type crash-resume.
+   * The fallback is announced once per process via {@link RunPrivateCacheRepo}.
    */
-  private cacheIdentityKey(policy: CachePolicy): string {
-    return isPolicyPrivate(policy) ? String(this.task.id) : this.task.type;
+  private cacheIdentityKey(policy: CachePolicy, outputCache: TaskOutputRepository): string {
+    if (!isPolicyPrivate(policy)) return this.task.type;
+    if (this.task.hasDeterministicId()) return String(this.task.id);
+    if (outputCache instanceof RunPrivateCacheRepo) outputCache.noteFallbackKey();
+    return this.task.type;
   }
 
   async lookup(
@@ -79,7 +89,10 @@ export class CacheCoordinator<Input extends TaskInput, Output extends TaskOutput
   ): Promise<Output | undefined> {
     if (!outputCache || !this.task.cacheable) return undefined;
 
-    const cached = await outputCache.getOutput(this.cacheIdentityKey(policy), keyInputs);
+    const cached = await outputCache.getOutput(
+      this.cacheIdentityKey(policy, outputCache),
+      keyInputs
+    );
     if (cached === undefined) return undefined;
 
     const outputSchema = (this.task.constructor as typeof Task).outputSchema();
@@ -118,7 +131,11 @@ export class CacheCoordinator<Input extends TaskInput, Output extends TaskOutput
       output as Record<string, unknown>,
       outputSchema as unknown as SchemaProperties
     );
-    await outputCache.saveOutput(this.cacheIdentityKey(policy), keyInputs, wireOutputs as Output);
+    await outputCache.saveOutput(
+      this.cacheIdentityKey(policy, outputCache),
+      keyInputs,
+      wireOutputs as Output
+    );
   }
 
   // ========================================================================

packages/task-graph/src/task/ITask.ts

@@ -283,6 +283,13 @@ export interface ITaskSerialization {
 export interface ITaskState<Config extends TaskConfig = TaskConfig> {
   readonly config: Config;
   get id(): unknown;
+  /**
+   * Returns true when `id` is a stable, caller-pinned value (not the
+   * autogenerated v4 UUID default). The run-private cache uses this to
+   * decide whether to key by instance id (deterministic) or fall back to
+   * keying by task type so a crash-restart can still find prior rows.
+   */
+  hasDeterministicId(): boolean;
   status: TaskStatus;
   /** 0..100 for measured progress, or `undefined` for indeterminate (in-progress, percentage unknown). */
   progress: number | undefined;

packages/task-graph/src/task/Task.ts

@@ -31,6 +31,13 @@ import { TaskRunner } from "./TaskRunner";
 import type { TaskConfig, TaskIdType, TaskInput, TaskOutput, TaskTypeName } from "./TaskTypes";
 import { TaskConfigSchema, TaskStatus } from "./TaskTypes";
 
+/**
+ * Matches a canonical lowercase RFC 4122 v4 UUID. Used to detect the
+ * autogenerated default id produced when no `config.id` was supplied so the
+ * cache layer can fall back to a deterministic per-type key.
+ */
+const UUID_V4_REGEX = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
+
 /**
  * Base class for all tasks that implements the ITask interface.
  * This abstract class provides common functionality for both simple and compound tasks.
@@ -393,6 +400,20 @@ export class Task<
     return this.config.id;
   }
 
+  /**
+   * Whether this task instance has a caller-pinned (deterministic) id. The
+   * default constructor mints a fresh v4 UUID when no id is supplied, so a
+   * v4-UUID-shaped id is treated as autogenerated — not deterministic — and
+   * the run-private cache will fall back to keying by task type to preserve
+   * a best-effort crash-resume across restarts. Override (or pin `config.id`
+   * to a stable string) to enable exact row reuse across restarts.
+   */
+  public hasDeterministicId(): boolean {
+    const id = this.config.id;
+    if (typeof id !== "string" || id.length === 0) return false;
+    return !UUID_V4_REGEX.test(id);
+  }
+
   /**
    * Runtime configuration (not serialized with the task).
    * Set via the constructor's third argument or mutated by the graph runner.

packages/test/src/test/storage-tabular/SharedInMemoryTabularStorageSyncBarrier.test.ts

@@ -0,0 +1,102 @@
+/**
+ * @license
+ * Copyright 2026 Steven Roussey <sroussey@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { SharedInMemoryTabularStorage } from "@workglow/storage";
+import type { DataPortSchemaObject } from "@workglow/util/schema";
+import { describe, expect, it } from "vitest";
+
+const ItemSchema = {
+  type: "object",
+  properties: {
+    id: { type: "string" },
+    value: { type: "string" },
+  },
+  required: ["id", "value"],
+  additionalProperties: false,
+} as const satisfies DataPortSchemaObject;
+
+const ItemPK = ["id"] as const;
+
+/**
+ * Builds a fresh, isolated channel name so parallel test runs cannot collide.
+ */
+function uniqueChannel(label: string): string {
+  return `sync_barrier_${label}_${Date.now()}_${Math.random().toString(36).slice(2)}`;
+}
+
+describe("SharedInMemoryTabularStorage.setupDatabase awaits tab-sync by default", () => {
+  it("first tab's first get sees rows another tab wrote before setupDatabase() returns", async () => {
+    const channelName = uniqueChannel("default_await");
+
+    // Simulate a peer tab that already has rows. `peer` constructed first so it
+    // owns the canonical row set before the first tab calls setupDatabase().
+    const peer = new SharedInMemoryTabularStorage<typeof ItemSchema, typeof ItemPK>(
+      channelName,
+      ItemSchema,
+      ItemPK
+    );
+    await peer.setupDatabase();
+    await peer.put({ id: "preexisting", value: "from-peer" });
+
+    // Fresh tab joins the channel after the peer has populated rows.
+    const fresh = new SharedInMemoryTabularStorage<typeof ItemSchema, typeof ItemPK>(
+      channelName,
+      ItemSchema,
+      ItemPK
+    );
+    // Default is awaitTabSync: true — should not resolve until SYNC_RESPONSE
+    // has been applied to the inner store.
+    await fresh.setupDatabase();
+
+    const row = await fresh.get({ id: "preexisting" } as any);
+    expect(row).toBeDefined();
+    expect(row).toMatchObject({ id: "preexisting", value: "from-peer" });
+
+    fresh.destroy();
+    peer.destroy();
+  });
+
+  it("explicit awaitTabSync: false matches the prior fire-and-forget behavior (opt-out)", async () => {
+    const channelName = uniqueChannel("opt_out");
+
+    const fresh = new SharedInMemoryTabularStorage<typeof ItemSchema, typeof ItemPK>(
+      channelName,
+      ItemSchema,
+      ItemPK
+    );
+    // No peer is on the channel, so there is nothing to wait for; the opt-out
+    // path must still resolve promptly and the storage must still work.
+    await fresh.setupDatabase({ awaitTabSync: false });
+
+    await fresh.put({ id: "self", value: "from-self" });
+    const row = await fresh.get({ id: "self" } as any);
+    expect(row).toMatchObject({ id: "self", value: "from-self" });
+
+    fresh.destroy();
+  });
+
+  it("setupDatabase() resolves even when no peer responds (timeout path)", async () => {
+    const channelName = uniqueChannel("no_peer");
+
+    const solo = new SharedInMemoryTabularStorage<typeof ItemSchema, typeof ItemPK>(
+      channelName,
+      ItemSchema,
+      ItemPK
+    );
+
+    // With no peer broadcaster on the channel SYNC_RESPONSE never arrives;
+    // setupDatabase must still resolve via the SYNC_TIMEOUT fallback rather
+    // than hang. Bounded by a generous outer timeout in this test.
+    await Promise.race([
+      solo.setupDatabase(),
+      new Promise<never>((_, reject) =>
+        setTimeout(() => reject(new Error("setupDatabase never resolved")), 5000)
+      ),
+    ]);
+
+    solo.destroy();
+  });
+});