fix(storage): wrapper tx-handle forwarding + real-pool mutex bypass

Addresses Copilot review on PR #458. Five related fixes flagged after the
real-pool dedicated-client refactor exposed gaps in the wrapper layer
and a perf regression in the per-method mutex.

PostgresTabularStorage — mutex bypass on real `pg.Pool`:
  The per-instance mutex was wrapping every public method, serializing
  reads and writes that previously fanned across separate pool clients
  in parallel. With the dedicated-client `withTransaction` path, the
  mutex is no longer needed for tx isolation on real pool — `pool.connect()`
  provides that. The mutex now short-circuits to a no-op when the handle
  exposes `connect()`, and only serializes on the single-connection path
  (PGlite / PGLitePool) where it is required to keep external traffic
  out of the open transaction.

TelemetryTabularStorage.withTransaction — forward inner tx handle:
  Previously called `inner.withTransaction(() => fn(this))`, throwing
  away the inner's tx proxy and passing the outer wrapper to `fn`. With
  the new createTxView proxy, `fn(this).put(...)` would re-enter the
  inner's PUBLIC put through the mutex-acquiring path and deadlock on
  single-connection backends, or run on the wrong connection on real
  pool. Now constructs a fresh `TelemetryTabularStorage` wrapping the
  inner's `tx` proxy and passes that to `fn`, so writes inside the
  callback flow through the transaction-bound resources.

ScopedTabularStorage.withTransaction — forward + defer events:
  Same forwarding fix as Telemetry. Additionally, since Scoped emits
  events on its own local emitter (not via the inner's `put` events),
  the inner's deferred-event flush does not cover Scoped's outer events.
  The tx wrapper's emitter is replaced with a per-tx buffer; entries
  are flushed to the parent's emitter only after `inner.withTransaction`
  resolves successfully (rollbacks discard via promise rejection).

CachedTabularStorage.withTransaction — explicit override:
  Previously inherited `BaseTabularStorage`'s no-op default, silently
  losing rollback / atomicity for callers using the cached wrapper.
  Now forwards to the durable store's `withTransaction` so writes inside
  `fn` are atomic. The cache layer is bypassed for the transaction's
  duration; documented post-tx staleness is repopulated on the first
  read miss.

ITabularStorage.withTransaction JSDoc — bring docs in step:
  Previous text claimed the API throws on a real `pg.Pool`, said the
  storage instance is not isolated from concurrent callers, and stated
  that `tx === this` for every backend — all out of date after the
  recent refactor. Updated to describe: real-pool path uses `pool.connect()`;
  concurrent calls are isolated (mutex on single-conn, dedicated client
  on real pool); the `tx` handle is a Proxy distinct from `this`, and
  callers MUST use it to avoid deadlock or wrong-connection writes.

767 storage tests passing locally.

Author: claude

packages/knowledge-base/src/knowledge-base/ScopedTabularStorage.ts

@@ -227,7 +227,42 @@ export class ScopedTabularStorage<
   }
 
   withTransaction<T>(fn: (tx: this) => Promise<T>): Promise<T> {
-    return this.inner.withTransaction(() => fn(this));
+    // Two responsibilities here:
+    //  1. Forward the inner's `tx` proxy into `fn`, so writes from the
+    //     callback go through the transaction-bound resources rather than
+    //     re-entering the public mutex path on the parent (which would
+    //     deadlock on single-connection backends, or run on the wrong
+    //     connection on real `pg.Pool`).
+    //  2. Defer events emitted by the scoped wrapper until the inner
+    //     transaction actually commits — Scoped emits on its own emitter
+    //     (separate from the inner's `put` events), so the inner's
+    //     deferred-event flush does not cover them.
+    const deferred: Array<[TabularEventName, unknown[]]> = [];
+    return this.inner
+      .withTransaction((innerTx: AnyTabularStorage) => {
+        const txWrapper = new ScopedTabularStorage<
+          Schema,
+          PrimaryKeyNames,
+          Entity,
+          PrimaryKey,
+          InsertType
+        >(innerTx, this.kbId);
+        // Override the tx wrapper's emitter to buffer instead of fan out;
+        // listeners on the original wrapper see events only after COMMIT.
+        (txWrapper as unknown as { events: { emit: (n: string, ...a: unknown[]) => void } }).events =
+          {
+            emit: (name: string, ...args: unknown[]) => {
+              deferred.push([name as TabularEventName, args]);
+            },
+          };
+        return fn(txWrapper as unknown as this);
+      })
+      .then((result) => {
+        for (const [name, args] of deferred) {
+          (this.events.emit as (n: TabularEventName, ...a: unknown[]) => void)(name, ...args);
+        }
+        return result;
+      });
   }
 
   // Lifecycle — no-op for shared storage

packages/postgres/src/storage/PostgresTabularStorage.ts

@@ -657,18 +657,26 @@ export class PostgresTabularStorage<
   }
 
   /**
-   * Per-instance promise-chain mutex. Every public read/write method awaits
-   * the previous holder before running and yields the lock when it settles;
-   * `withTransaction` holds it for the duration of the user's callback so
-   * concurrent calls from outside `fn` queue behind the transaction instead
-   * of slipping into it.
+   * Per-instance promise-chain mutex. Only meaningful on single-connection
+   * backends (PGlite / PGLitePool) — there `withTransaction` runs on the
+   * shared session and we need to keep external callers from slipping into
+   * the open transaction. On a real `pg.Pool` (anything exposing
+   * `connect()`) the mutex would serialize independent reads/writes that
+   * Postgres is happy to fan across separate pool clients, turning the
+   * pool's main benefit into a per-instance bottleneck — so we short-circuit
+   * to a no-op on that path. Real-pool isolation comes from
+   * `withTransaction` dedicating its own client via `pool.connect()`.
    *
    * The Proxy returned by {@link createTxView} routes back to the private
    * `_*Internal` methods directly, so calls made *through* the `tx` handle
    * inside `fn` do not deadlock against the mutex held by `withTransaction`.
    */
   private mutexChain: Promise<void> = Promise.resolve();
+  private get serializeOps(): boolean {
+    return typeof (this.db as unknown as { connect?: unknown }).connect !== "function";
+  }
   private async mutex<T>(fn: () => Promise<T>): Promise<T> {
+    if (!this.serializeOps) return fn();
     const prev = this.mutexChain;
     let release!: () => void;
     this.mutexChain = new Promise<void>((resolve) => {

packages/storage/src/tabular/CachedTabularStorage.ts

@@ -328,6 +328,31 @@ export class CachedTabularStorage<
     await this.cache.deleteSearch(criteria);
   }
 
+  /**
+   * Runs `fn` inside the durable store's transaction. The cache layer is
+   * intentionally bypassed for the transaction's duration — coordinating
+   * two-phase commit between the durable store and an in-memory cache is
+   * out of scope, and callers asking for `withTransaction` are asking for
+   * atomicity, which only the durable can provide. Inside `fn`, reads and
+   * writes go straight through the durable's transaction handle.
+   *
+   * After `fn` resolves and the durable commits, the cache may hold stale
+   * rows for any keys the transaction mutated; subsequent reads through
+   * this wrapper repopulate the cache on miss, so callers who need the
+   * cache hot for those rows should issue a read-through after the
+   * transaction resolves. Inheriting `BaseTabularStorage`'s no-op default
+   * here would silently lose the rollback / atomicity guarantee, since the
+   * default just runs `fn(this)` against the cached wrapper itself.
+   */
+  override async withTransaction<T>(fn: (tx: this) => Promise<T>): Promise<T> {
+    await this.initializeCache();
+    return this.durable.withTransaction(
+      fn as unknown as (
+        tx: ITabularStorage<Schema, PrimaryKeyNames, Entity, PrimaryKey>
+      ) => Promise<T>
+    );
+  }
+
   /**
    * Invalidates the cache by clearing it and resetting initialization flag
    */

packages/storage/src/tabular/ITabularStorage.ts

@@ -306,32 +306,44 @@ export interface ITabularStorage<
    *
    * Backends differ in how strong the guarantee is:
    *   - **SQLite**: real `BEGIN` / `COMMIT` / `ROLLBACK`.
-   *   - **PostgreSQL**: real `BEGIN` / `COMMIT` / `ROLLBACK` only when the
-   *     underlying handle is a single-connection wrapper (PGLitePool, raw
-   *     PGlite). On a multi-connection `pg.Pool` the call throws, because
-   *     methods on this storage dispatch each query through the pool
-   *     independently and the surrounding `BEGIN`/`COMMIT` cannot bracket
-   *     them. Use {@link putBulk} for atomic bulk inserts on a real pool, or
-   *     wrap the pool yourself.
+   *   - **PostgreSQL**: real `BEGIN` / `COMMIT` / `ROLLBACK`. On a real
+   *     `pg.Pool` (anything exposing `connect()`) the implementation
+   *     dedicates a client via `pool.connect()` and runs the transaction on
+   *     that client, leaving the parent's pool free for external traffic
+   *     in parallel. On single-connection wrappers (PGLitePool, raw PGlite)
+   *     the transaction runs on the shared session and concurrent calls on
+   *     the same instance are serialized behind a per-instance mutex so
+   *     they cannot slip into the open transaction.
    *   - **Supabase, in-memory, file system, IndexedDB**: best-effort. The
    *     callback runs to completion and rejection propagates, but partial
    *     writes are not rolled back because the backend does not expose a
    *     transaction surface usable by this API.
    *
-   * **Concurrency contract:** `withTransaction` does not isolate the storage
-   * instance from concurrent callers. While `fn` is awaiting, any unrelated
-   * operation invoked on the same storage instance from concurrent code will
-   * run on the same underlying connection (for SQLite and single-connection
-   * Postgres wrappers) and become part of this transaction. Do not invoke
-   * other methods on the same storage instance from outside `fn` while a
-   * `withTransaction` call is in flight; serialize that work yourself, or
-   * use a separate storage instance for the concurrent work.
+   * **Concurrency contract:**
+   *   - On backends with native transaction support (SQLite, PostgreSQL),
+   *     concurrent calls on the same storage instance are isolated from the
+   *     open transaction: SQLite and the single-connection Postgres path
+   *     serialize them through a per-instance mutex; the real-pool Postgres
+   *     path runs them on independent pool clients in parallel. Either way,
+   *     unrelated writes never accidentally commit or roll back along with
+   *     `fn`.
+   *   - On best-effort backends concurrent writes have no atomicity barrier
+   *     to begin with — the contract on those backends is "runs `fn`", not
+   *     "isolates `fn`".
    *
-   * The storage instance passed to `fn` is the same instance (`this`) for
-   * every backend in this codebase. The `tx` parameter is provided so the
-   * callback signature stays forward-compatible with a future backend that
-   * returns a transaction-scoped clone — callers should use the handle and
-   * not capture `this` directly.
+   * The `tx` handle passed to `fn` is **not** the same object as `this` for
+   * backends with native transaction support — it is a Proxy that routes
+   * writes through the transaction-bound resources (the dedicated client on
+   * real `pg.Pool`, the bypass-mutex internal methods on SQLite/PGlite) and
+   * routes events through the transaction's deferred-emit queue. Callers
+   * MUST use `tx` for everything inside `fn`. Capturing the outer `this` and
+   * calling methods on it from inside `fn` will deadlock against the held
+   * mutex (single-connection backends) or run on the wrong connection
+   * (`pg.Pool`), and is unsupported.
+   *
+   * Nested `withTransaction` calls — either via the original instance or
+   * via `tx` — throw rather than reusing the outer transaction implicitly.
+   * Use SAVEPOINT directly if you need nested rollback boundaries.
    */
   withTransaction<T>(fn: (tx: this) => Promise<T>): Promise<T>;
 

packages/storage/src/tabular/TelemetryTabularStorage.ts

@@ -137,8 +137,20 @@ export class TelemetryTabularStorage<
   }
 
   withTransaction<T>(fn: (tx: this) => Promise<T>): Promise<T> {
+    // Construct a tx-bound telemetry wrapper that forwards to the inner's
+    // transaction handle (`innerTx`) — the proxy from the inner's
+    // `createTxView`. Passing the outer `this` would re-enter `inner.put()`
+    // through the public, mutex-acquiring path and deadlock against the held
+    // mutex on single-connection backends, or run on the wrong connection on
+    // real `pg.Pool`.
     return traced("workglow.storage.tabular.withTransaction", this.storageName, () =>
-      this.inner.withTransaction(() => fn(this))
+      this.inner.withTransaction((innerTx) => {
+        const txWrapper = new TelemetryTabularStorage(
+          this.storageName,
+          innerTx as ITabularStorage<Schema, PrimaryKeyNames, Entity, PrimaryKey, InsertType>
+        ) as this;
+        return fn(txWrapper);
+      })
     );
   }