fix: replies fail after transcript projection repair (#121025)

* fix(reply): rebind prepared session ownership

* test(mcp): correlate queued wait events

* fix(sessions): await transcript admission projection

* fix(recovery): distinguish admission failure stages

* refactor(sessions): split runtime transcript projection

* fix(ci): preserve transcript API baseline

* fix(sessions): await target projection only

src/auto-reply/reply/get-reply-run-admission.ts

@@ -280,21 +280,20 @@ export async function prepareReplyRunAdmission(context: PreparedReplyRunContext)
     resolveCommandTurnTargetSessionKey(ctx) !== undefined
       ? sessionKey
       : undefined;
-  if (
-    commandTurnContinuationTargetKey === undefined &&
-    providedReplyOperation !== undefined &&
-    providedReplyOperation.result === null &&
-    providedReplyOperation.phase === "queued" &&
-    sessionId !== undefined &&
-    sessionId !== providedReplyOperation.sessionId
-  ) {
-    // Dispatch reserves a queued operation before session init. If stale init
-    // rotates the session, move the reservation so later steer/abort paths
-    // target the session that will actually run. Command-turn continuations
-    // rebind after slot adoption below: rebinding first would collide with a
-    // still-active target operation that owns the same session ID.
-    providedReplyOperation.updateSessionId(sessionId);
-  }
+  const rebindProvidedReplyOperation = (nextSessionId: string) => {
+    if (
+      commandTurnContinuationTargetKey === undefined &&
+      providedReplyOperation !== undefined &&
+      providedReplyOperation.result === null &&
+      providedReplyOperation.phase === "queued" &&
+      nextSessionId !== providedReplyOperation.sessionId
+    ) {
+      // Dispatch can reserve a queued operation before session init discovers the
+      // authoritative row. Keep steer/abort and durable admission on that session.
+      // Command continuations rebind only after adopting the target slot below.
+      providedReplyOperation.updateSessionId(nextSessionId);
+    }
+  };
   const isOwnPreDispatchOperationSession = (candidateSessionId: string | undefined): boolean =>
     providedReplyOperation !== undefined &&
     providedReplyOperation.result === null &&
@@ -318,6 +317,7 @@ export async function prepareReplyRunAdmission(context: PreparedReplyRunContext)
           sessionEntry)
         : sessionEntry;
     const latestSessionId = latestSessionEntry?.sessionId ?? sessionIdFinal;
+    rebindProvidedReplyOperation(latestSessionId);
     opts?.onSessionPrepared?.({ sessionKey, sessionId: latestSessionId, storePath });
     const sessionFile = storePath
       ? formatSqliteSessionFileMarker({ agentId, sessionId: latestSessionId, storePath })

src/auto-reply/reply/get-reply-run.media-only.test.ts

@@ -2136,6 +2136,43 @@ describe("runPreparedReply media-only handling", () => {
     }
   });
 
+  it("rebinds a provisional pre-dispatch operation to a discovered existing session", async () => {
+    const operation = createReplyOperation({
+      sessionId: "provisional-session",
+      sessionKey: "session-key",
+      resetTriggered: false,
+    });
+    const sessionStore: Record<string, SessionEntry> = {
+      "session-key": {
+        sessionId: "existing-session",
+        sessionFile: "/tmp/existing-session.jsonl",
+        updatedAt: 1,
+      },
+    };
+
+    try {
+      await expect(
+        runPreparedReply(
+          baseParams({
+            isNewSession: false,
+            sessionEntry: undefined,
+            sessionId: undefined,
+            sessionStore,
+            storePath: "/tmp/sessions.json",
+            opts: { replyOperation: operation } as never,
+          }),
+        ),
+      ).resolves.toEqual({ text: "ok" });
+
+      const call = requireLastRunReplyAgentCall();
+      expect(operation.sessionId).toBe("existing-session");
+      expect(call.replyOperation).toBe(operation);
+      expect(call.followupRun.run.sessionId).toBe("existing-session");
+    } finally {
+      operation.complete();
+    }
+  });
+
   it("does not interrupt its provided pre-dispatch reply operation for reset turns", async () => {
     const queueSettings = await import("./queue/settings-runtime.js");
     const embeddedAgentRuntime = await import("../../agents/embedded-agent.runtime.js");

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

@@ -26,6 +26,7 @@ import {
   reconcileSessionTranscriptIndexes,
   startSessionTranscriptIndexReconcile,
   waitForSessionTranscriptIndexReconcile,
+  waitForSessionTranscriptProjection,
 } from "./session-transcript-reconcile.js";
 
 const queuedSessionWrite = vi.hoisted(() => vi.fn());
@@ -454,6 +455,50 @@ describe("SQLite active transcript event projection", () => {
     ).toEqual([]);
   });
 
+  it("resolves one session before unrelated projection repair completes", async () => {
+    const secondScope = { ...scope, sessionId: "session-slow", sessionKey: "agent:main:slow" };
+    await persistSessionTranscriptTurn(scope, {
+      messages: [
+        { eventId: "target", parentId: null, message: { role: "user", content: "target" } },
+      ],
+      touchSessionEntry: false,
+    });
+    await persistSessionTranscriptTurn(secondScope, {
+      messages: Array.from({ length: 5_000 }, (_, index) => ({
+        eventId: `slow-${index}`,
+        parentId: index === 0 ? null : `slow-${index - 1}`,
+        message: { role: "toolResult", content: "slow" },
+      })),
+      touchSessionEntry: false,
+    });
+    const databaseOptions = { agentId: scope.agentId, env: scope.env };
+    const database = openOpenClawAgentDatabase(databaseOptions);
+    const markDirty = database.db.prepare(
+      "UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?",
+    );
+    markDirty.run(scope.sessionId);
+    markDirty.run(secondScope.sessionId);
+
+    startSessionTranscriptIndexReconcile({
+      ...databaseOptions,
+      preferredSessionId: scope.sessionId,
+    });
+    let allReconciled = false;
+    const allReconciliation = waitForSessionTranscriptIndexReconcile(databaseOptions).then(() => {
+      allReconciled = true;
+    });
+
+    await waitForSessionTranscriptProjection(scope);
+
+    expect(allReconciled).toBe(false);
+    expect(
+      database.db
+        .prepare("SELECT needs_rebuild FROM session_transcript_index_state WHERE session_id = ?")
+        .get(scope.sessionId),
+    ).toEqual({ needs_rebuild: 0 });
+    await allReconciliation;
+  }, 30_000);
+
   it("keeps projection state and rows on one snapshot during a concurrent append", async () => {
     await persistSessionTranscriptTurn(scope, {
       messages: [

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

@@ -77,6 +77,23 @@ function readSessionTranscriptProjectionState(
   };
 }
 
+export function sessionTranscriptIndexNeedsReconcile(db: DatabaseSync, sessionId: string): boolean {
+  const latest = executeSqliteQueryTakeFirstSync(
+    db,
+    getIndexKysely(db)
+      .selectFrom("transcript_events")
+      .select("seq")
+      .where("session_id", "=", sessionId)
+      .orderBy("seq", "desc")
+      .limit(1),
+  );
+  if (!latest) {
+    return false;
+  }
+  const state = readSessionTranscriptProjectionState(db, sessionId);
+  return !state || state.needsRebuild || state.indexedSeq !== latest.seq;
+}
+
 function writeWatermark(
   db: DatabaseSync,
   sessionId: string,
@@ -371,8 +388,7 @@ export function reconcileSessionTranscriptIndexInTransaction(
     deleteSessionTranscriptIndexInTransaction(db, sessionId);
     return false;
   }
-  const state = readSessionTranscriptProjectionState(db, sessionId);
-  if (state && !state.needsRebuild && state.indexedSeq === latest.seq) {
+  if (!sessionTranscriptIndexNeedsReconcile(db, sessionId)) {
     return false;
   }
   const rows = executeSqliteQuerySync(

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

@@ -2,10 +2,12 @@
 // request paths may only schedule it and return a bounded retryable response.
 import { randomInt } from "node:crypto";
 import path from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
 import { fileURLToPath, pathToFileURL } from "node:url";
 import { Worker, type WorkerOptions } from "node:worker_threads";
 import { createSubsystemLogger } from "../../logging/subsystem.js";
 import {
+  openOpenClawAgentDatabase,
   resolveOpenClawAgentSqlitePath,
   runOpenClawAgentWriteTransaction,
   type OpenClawAgentDatabase,
@@ -20,6 +22,7 @@ import {
 import {
   deleteOrphanedTranscriptIndexRowsInTransaction,
   listSessionsNeedingTranscriptIndexReconcile,
+  sessionTranscriptIndexNeedsReconcile,
 } from "./session-transcript-index.js";
 import {
   appendPreparedSessionTranscriptProjectionChunkInTransaction,
@@ -36,6 +39,7 @@ import type {
 
 const log = createSubsystemLogger("sessions/transcript-index");
 const PROJECTION_WRITE_CHUNK_ROWS = 512;
+const PROJECTION_READY_POLL_MS = 10;
 
 type RunningReconcile = {
   pending: boolean;
@@ -413,10 +417,17 @@ export async function waitForSessionTranscriptIndexReconcile(
   await runningReconciles.get(reconcileKey(params))?.promise;
 }
 
-/** Waits for a projection rebuild already scheduled by a failed transcript read. */
+/** Waits only until the requested session's scheduled projection rebuild settles. */
 export async function waitForSessionTranscriptProjection(
   scope: SessionTranscriptReadScope,
 ): Promise<void> {
   const resolved = resolveSqliteTranscriptReadScope(scope);
-  await waitForSessionTranscriptIndexReconcile(toDatabaseOptions(resolved));
+  const databaseOptions = toDatabaseOptions(resolved);
+  const database = openOpenClawAgentDatabase(databaseOptions);
+  while (
+    isSessionTranscriptIndexReconcileRunning(databaseOptions) &&
+    sessionTranscriptIndexNeedsReconcile(database.db, resolved.sessionId)
+  ) {
+    await delay(PROJECTION_READY_POLL_MS);
+  }
 }

src/sessions/user-turn-transcript.test.ts

@@ -5,7 +5,11 @@ import path from "node:path";
 import { castAgentMessage } from "openclaw/plugin-sdk/test-fixtures";
 import { afterEach, describe, expect, it } from "vitest";
 import { formatSqliteSessionFileMarker } from "../config/sessions/legacy-sqlite-marker.js";
-import { loadTranscriptEvents } from "../config/sessions/session-accessor.js";
+import {
+  loadTranscriptEvents,
+  persistSessionTranscriptTurn,
+  replaceSessionEntry,
+} from "../config/sessions/session-accessor.js";
 import {
   buildLateMediaAttachedProjection,
   createUserTurnTranscriptRecorder,
@@ -552,6 +556,48 @@ describe("user turn transcript persistence", () => {
       });
     });
 
+    it("waits for a deferred projection rebuild before returning admission identity", async () => {
+      const dir = createTempDir("openclaw-user-turn-recorder-projection-");
+      const target = createSqliteTranscriptTarget({ dir });
+      await replaceSessionEntry(
+        { storePath: target.storePath, sessionKey: target.sessionKey },
+        {
+          sessionId: target.sessionId,
+          sessionFile: target.sqliteMarker,
+          updatedAt: 1,
+        },
+      );
+      await persistSessionTranscriptTurn(target, {
+        messages: [
+          { eventId: "root", parentId: null, message: { role: "user", content: "root" } },
+          {
+            eventId: "inactive",
+            parentId: "root",
+            message: { role: "assistant", content: "inactive" },
+          },
+          {
+            eventId: "active",
+            parentId: "root",
+            message: { role: "assistant", content: "active" },
+          },
+        ],
+        touchSessionEntry: false,
+      });
+      const recorder = createUserTurnTranscriptRecorder({
+        input: { text: "admit after rebuild", idempotencyKey: "projection:user" },
+        target,
+      });
+
+      const persisted = await recorder.persistApproved({ expectedSessionId: target.sessionId });
+
+      expect(persisted).toBeDefined();
+      expect(persisted?.admission).toMatchObject({
+        entryId: persisted?.messageId,
+        sessionId: target.sessionId,
+        idempotencyKey: "projection:user",
+      });
+    });
+
     it("preserves distinct text supplied with late-resolved media", async () => {
       const dir = createTempDir("openclaw-user-turn-recorder-late-caption-");
       const target = createSqliteTranscriptTarget({ dir });

src/sessions/user-turn-transcript.ts

@@ -6,9 +6,11 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coe
 import type { AgentMessage } from "../../packages/agent-core/src/types.js";
 import {
   persistSessionTranscriptTurn,
+  readActiveTranscriptEntryAnchor,
   type TranscriptEntryAnchor,
   type SessionTranscriptTurnPersistOptions,
 } from "../config/sessions/session-accessor.js";
+import { waitForSessionTranscriptProjection } from "../config/sessions/session-transcript-reconcile.js";
 import { readPersistedMediaFacts, type MediaFact } from "../media/media-facts.js";
 import { applyInputProvenanceToUserMessage, normalizeInputProvenance } from "./input-provenance.js";
 import { resolveUserTurnTranscriptAdmission } from "./user-turn-transcript-admission.js";
@@ -189,13 +191,7 @@ export function buildPersistedUserTurnMessage(params: UserTurnInput): PersistedU
 function resolvePersistedUserTurnMessage(
   params: Pick<UserTurnMessagePersistenceParams, "input" | "message">,
 ): PersistedUserTurnMessage | undefined {
-  if (params.message) {
-    return params.message;
-  }
-  if (!params.input) {
-    return undefined;
-  }
-  return buildPersistedUserTurnMessage(params.input);
+  return params.message ?? (params.input ? buildPersistedUserTurnMessage(params.input) : undefined);
 }
 
 function isUserMessage(message: AgentMessage): message is PersistedUserTurnMessage {
@@ -410,7 +406,7 @@ async function persistUserTurnTranscript(
       ],
     },
   );
-  const appended = turn.messages[0] as
+  let appended = turn.messages[0] as
     | {
         anchor?: Omit<UserTurnTranscriptAdmissionReceipt, "logicalTurnId" | "role">;
         appended: boolean;
@@ -418,6 +414,11 @@ async function persistUserTurnTranscript(
         message: PersistedUserTurnMessage;
       }
     | undefined;
+  if (appended && !appended.anchor && appended.message.role === "user") {
+    await waitForSessionTranscriptProjection(params);
+    const anchor = readActiveTranscriptEntryAnchor({ ...params, entryId: appended.messageId });
+    appended = anchor ? { ...appended, anchor } : appended;
+  }
   if (!appended?.anchor || appended.message.role !== "user") {
     return undefined;
   }

test/e2e/qa-lab/runtime/mcp-channels-docker-client.ts

@@ -234,17 +234,38 @@ async function main() {
       "expected one seeded attachment",
     );
 
+    let waitCursor = 0;
+    let lastWaitEvent: Record<string, unknown> | undefined;
     const waitMessage = `wait event ${randomUUID()}`;
-    const [waited, waitRun] = await Promise.all([
-      callTool<{
-        structuredContent?: { event?: Record<string, unknown> };
-      }>({
-        name: "events_wait",
-        arguments: {
-          session_key: "agent:main:main",
-          after_cursor: 0,
-          timeout_ms: 120_000,
+    const [waitEvent, waitRun] = await Promise.all([
+      waitFor(
+        "correlated events_wait user event",
+        async () => {
+          const waited = await callTool<{
+            structuredContent?: { event?: Record<string, unknown> };
+          }>({
+            name: "events_wait",
+            arguments: {
+              session_key: "agent:main:main",
+              after_cursor: waitCursor,
+              timeout_ms: 15_000,
+            },
+          });
+          const event = waited.structuredContent?.event;
+          if (!event) {
+            return undefined;
+          }
+          assert(typeof event.cursor === "number", "expected events_wait cursor");
+          waitCursor = event.cursor;
+          lastWaitEvent = event;
+          return event.text === waitMessage ? event : undefined;
         },
+        120_000,
+      ).catch((error: unknown) => {
+        throw new Error(
+          `events_wait did not return the expected user event: ${JSON.stringify(lastWaitEvent)}`,
+          { cause: error },
+        );
       }),
       gateway.request<{ runId?: string; status?: string }>("chat.send", {
         sessionKey: "agent:main:main",
@@ -252,12 +273,9 @@ async function main() {
         idempotencyKey: randomUUID(),
       }),
     ]);
-    const waitEvent = waited.structuredContent?.event;
-    assert(waitEvent, "expected events_wait result");
     assert(waitEvent.type === "message", "expected message event");
     assert(waitEvent.role === "user", "expected user event role");
     assert(waitEvent.text === waitMessage, "expected wait event text");
-    const waitCursor = typeof waitEvent.cursor === "number" ? waitEvent.cursor : 0;
     assert(
       waitRun.status === "started" && typeof waitRun.runId === "string",
       `chat.send did not start: ${JSON.stringify(waitRun)}`,