perf(gateway): skip clean transcript workers (#117342)

src/config/sessions/session-accessor.sqlite-active-events.test.ts

@@ -571,11 +571,12 @@ describe("SQLite active transcript event projection", () => {
     }
   });
 
-  it("awaits queued completion work after the preparation worker exits", async () => {
+  it("skips the preparation worker when the projection is already current", async () => {
     await persistSessionTranscriptTurn(scope, {
       messages: [{ eventId: "seed", message: { role: "user", content: "seed" } }],
       touchSessionEntry: false,
     });
+    queuedSessionWrite.mockClear();
     let resolveCompletionQueued!: () => void;
     const completionQueued = new Promise<void>((resolve) => {
       resolveCompletionQueued = resolve;
@@ -601,21 +602,26 @@ describe("SQLite active transcript event projection", () => {
       },
     );
     await entered;
+    const createWorker = vi.fn(() => {
+      throw new Error("clean projection must not spawn a worker");
+    });
     const outcome = reconcileSessionTranscriptIndexes({
       agentId: scope.agentId,
+      createWorker,
       env: scope.env,
     }).then(
       (value) => ({ value }),
       (error: unknown) => ({ error }),
     );
 
-    // The second queued write is the orphan sweep issued after the worker's done message.
+    // The second queued write is the preflight transaction waiting behind the held writer.
     await completionQueued;
     expect(queuedSessionWrite).toHaveBeenCalledTimes(2);
     releaseWriter();
     await heldWriter;
 
     expect(await outcome).toEqual({ value: { reconciledSessions: 0 } });
+    expect(createWorker).not.toHaveBeenCalled();
   }, 10_000);
 
   it("keeps dirty batch appends off the synchronous writer stack", async () => {

src/config/sessions/session-transcript-reconcile.ts

@@ -3,7 +3,7 @@
 import { randomInt } from "node:crypto";
 import path from "node:path";
 import { fileURLToPath, pathToFileURL } from "node:url";
-import { Worker } from "node:worker_threads";
+import { Worker, type WorkerOptions } from "node:worker_threads";
 import { createSubsystemLogger } from "../../logging/subsystem.js";
 import {
   resolveOpenClawAgentSqlitePath,
@@ -17,7 +17,10 @@ import {
   runExclusiveSqliteSessionWrite,
   toDatabaseOptions,
 } from "./session-accessor.sqlite-scope.js";
-import { deleteOrphanedTranscriptIndexRowsInTransaction } from "./session-transcript-index.js";
+import {
+  deleteOrphanedTranscriptIndexRowsInTransaction,
+  listSessionsNeedingTranscriptIndexReconcile,
+} from "./session-transcript-index.js";
 import {
   appendPreparedSessionTranscriptProjectionChunkInTransaction,
   claimPreparedSessionTranscriptProjectionInTransaction,
@@ -47,6 +50,7 @@ export type SessionTranscriptReconcileResult = {
 };
 
 type SessionTranscriptReconcileParams = OpenClawAgentDatabaseOptions & {
+  createWorker?: (filename: string | URL, options: WorkerOptions) => Worker;
   preferredSessionId?: string;
 };
 
@@ -198,7 +202,7 @@ async function finalizePreparedProjection(
 }
 
 /** Prepares full trees off-thread, then commits bounded chunks through the runtime writer owner. */
-export function reconcileSessionTranscriptIndexes(
+export async function reconcileSessionTranscriptIndexes(
   params: SessionTranscriptReconcileParams,
 ): Promise<SessionTranscriptReconcileResult> {
   const databasePath = resolveOpenClawAgentSqlitePath(params);
@@ -207,6 +211,19 @@ export function reconcileSessionTranscriptIndexes(
     ...(params.env ? { env: params.env } : {}),
     path: databasePath,
   };
+  // The SQLite owner can cheaply prove a clean projection before paying for a
+  // Worker. Keep the post-worker sweep too, because request-time writers may race.
+  const needsWorker = await runProjectionWrite(
+    databaseOptions,
+    "sessions.transcript-index.preflight",
+    (database) => {
+      deleteOrphanedTranscriptIndexRowsInTransaction(database.db);
+      return listSessionsNeedingTranscriptIndexReconcile(database.db).length > 0;
+    },
+  );
+  if (!needsWorker) {
+    return { reconciledSessions: 0 };
+  }
   const workerUrl = resolveSessionTranscriptReconcileWorkerUrl();
   const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined;
   const input: SessionTranscriptReconcileWorkerInput = {
@@ -216,9 +233,12 @@ export function reconcileSessionTranscriptIndexes(
   };
   let worker: Worker;
   try {
-    worker = new Worker(workerUrl, { workerData: input, execArgv: sourceWorkerExecArgv });
+    worker = (params.createWorker ?? ((filename, options) => new Worker(filename, options)))(
+      workerUrl,
+      { workerData: input, execArgv: sourceWorkerExecArgv },
+    );
   } catch (error) {
-    return Promise.reject(normalizeReconcileError(error));
+    throw normalizeReconcileError(error);
   }
 
   return new Promise<SessionTranscriptReconcileResult>((resolve, reject) => {