diff --git a/src/auto-reply/reply/get-reply-run-admission.ts b/src/auto-reply/reply/get-reply-run-admission.ts index 14d19b96588f..9a6d42aef5f7 100644 --- a/src/auto-reply/reply/get-reply-run-admission.ts +++ b/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 }) diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index 591641c8cd38..b9800cc6ec6f 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/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 = { + "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"); diff --git a/src/config/sessions/session-accessor.sqlite-active-events.test.ts b/src/config/sessions/session-accessor.sqlite-active-events.test.ts index 48c050d6b7f1..caa0ac975a1d 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.test.ts +++ b/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: [ diff --git a/src/config/sessions/session-transcript-index.ts b/src/config/sessions/session-transcript-index.ts index 6dd76dbb1bff..8f3ad96b45c1 100644 --- a/src/config/sessions/session-transcript-index.ts +++ b/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( diff --git a/src/config/sessions/session-transcript-reconcile.ts b/src/config/sessions/session-transcript-reconcile.ts index 0dc9d3a2d74f..e316a53af961 100644 --- a/src/config/sessions/session-transcript-reconcile.ts +++ b/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 { 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); + } } diff --git a/src/sessions/user-turn-transcript.test.ts b/src/sessions/user-turn-transcript.test.ts index 92b925b791e4..908318567a6d 100644 --- a/src/sessions/user-turn-transcript.test.ts +++ b/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 }); diff --git a/src/sessions/user-turn-transcript.ts b/src/sessions/user-turn-transcript.ts index 19964fd34520..f1588237e292 100644 --- a/src/sessions/user-turn-transcript.ts +++ b/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, ): 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; 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; } diff --git a/test/e2e/qa-lab/runtime/mcp-channels-docker-client.ts b/test/e2e/qa-lab/runtime/mcp-channels-docker-client.ts index 656cee485aa0..d2f4a8da9f52 100644 --- a/test/e2e/qa-lab/runtime/mcp-channels-docker-client.ts +++ b/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 | undefined; const waitMessage = `wait event ${randomUUID()}`; - const [waited, waitRun] = await Promise.all([ - callTool<{ - structuredContent?: { event?: Record }; - }>({ - 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 }; + }>({ + 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)}`,