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 e99d35d81d6e..58edd7082696 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.test.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.test.ts @@ -11,7 +11,7 @@ import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db import { appendTranscriptEvent, persistSessionTranscriptTurn } from "./session-accessor.js"; import { readRecentSessionTranscriptMessageEvents, - readSessionTranscriptActiveLeafEvents, + readSessionTranscriptActivePathEntryRelation, readSessionTranscriptActiveStats, readSessionTranscriptBoundedMessageTailPage, readSessionTranscriptMessageAnchorPage, @@ -113,9 +113,8 @@ describe("SQLite active transcript event projection", () => { "root", "active", ]); - expect(readSessionTranscriptActiveLeafEvents(scope)).toEqual([ - expect.objectContaining({ id: "active" }), - ]); + expect(readSessionTranscriptActivePathEntryRelation(scope, "active")).toBe("exact"); + expect(readSessionTranscriptActivePathEntryRelation(scope, "root")).toBe("ancestor"); expect(page.events.map((entry) => entry.seq)).toEqual([1, 2]); expect(page.totalMessages).toBe(2); expect( @@ -343,9 +342,8 @@ describe("SQLite active transcript event projection", () => { maxMessages: 1, }).activeLeafEntryId, ).toBe("newer-compaction"); - expect(readSessionTranscriptActiveLeafEvents(scope)).toEqual([ - expect.objectContaining({ id: "newer-compaction" }), - ]); + expect(readSessionTranscriptActivePathEntryRelation(scope, "newer-compaction")).toBe("exact"); + expect(readSessionTranscriptActivePathEntryRelation(scope, "post-reset")).toBe("ancestor"); expect(readSessionTranscriptMessageEventCount(scope)).toBe(5); expect(readSessionTranscriptMessageEventById(scope, "old")).toBeDefined(); }); diff --git a/src/config/sessions/session-accessor.sqlite-active-events.ts b/src/config/sessions/session-accessor.sqlite-active-events.ts index 06581933e223..24bcf3daefa1 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.ts @@ -90,34 +90,31 @@ export function readSessionTranscriptMessageEvents( }); } -/** Reads the projected active leaf without materializing the transcript. */ -export function readSessionTranscriptActiveLeafEvents( +/** Classifies one entry against the authoritative active path and leaf. */ +export function readSessionTranscriptActivePathEntryRelation( scope: SessionTranscriptReadScope, -): TranscriptEvent[] { + entryId: string | null, +): "exact" | "ancestor" | "off-path" { return withCurrentProjectionSnapshot(scope, (projection) => { - const leafEventId = projection.state.leafEventId; - if (!leafEventId) { - return []; + if (projection.state.leafEventId === entryId || entryId === null) { + return projection.state.leafEventId === entryId ? "exact" : "off-path"; } const db = getActiveTranscriptKysely(projection.database); const row = executeSqliteQueryTakeFirstSync( projection.database.db, db .selectFrom("transcript_event_identities as identity") - .innerJoin("transcript_events as event", (join) => + .innerJoin("session_transcript_active_events as active", (join) => join - .onRef("event.session_id", "=", "identity.session_id") - .onRef("event.seq", "=", "identity.seq"), + .onRef("active.session_id", "=", "identity.session_id") + .onRef("active.event_seq", "=", "identity.seq"), ) - .select("event.event_json") + .select("identity.seq") .where("identity.session_id", "=", projection.resolved.sessionId) - .where("identity.event_id", "=", leafEventId) + .where("identity.event_id", "=", entryId) .limit(1), ); - if (!row) { - throw new Error(`Active transcript leaf event is missing: ${leafEventId}`); - } - return [JSON.parse(row.event_json) as TranscriptEvent]; + return row ? "ancestor" : "off-path"; }); } diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index 61c35bc46147..27fc2d757f3b 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -241,7 +241,7 @@ export { readSessionTranscriptActiveStats, readSessionTranscriptBoundedMessageTailPage, readRecentSessionTranscriptMessageEvents, - readSessionTranscriptActiveLeafEvents, + readSessionTranscriptActivePathEntryRelation, readSessionTranscriptMessageAnchorPage, readSessionTranscriptMessageEventById, readSessionTranscriptMessageEventCount, diff --git a/src/gateway/server-methods/chat-send-admission.ts b/src/gateway/server-methods/chat-send-admission.ts index 4c3687ec9f3a..83a86a258c10 100644 --- a/src/gateway/server-methods/chat-send-admission.ts +++ b/src/gateway/server-methods/chat-send-admission.ts @@ -8,10 +8,7 @@ import { } from "../../auto-reply/reply/reply-run-registry.js"; import { resolveSessionWorkStartError } from "../../config/sessions.js"; import { SESSION_ROUTING_CHANGED_ERROR_REASON } from "../../config/sessions/main-session.js"; -import { - readSessionTranscriptActiveLeafEvents, - resolveSessionTranscriptActiveLeafEntryId, -} from "../../config/sessions/session-accessor.js"; +import { readSessionTranscriptActivePathEntryRelation } from "../../config/sessions/session-accessor.js"; import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js"; import { claimAgentRunContext, clearAgentRunContext } from "../../infra/agent-run-registry.js"; import { beginSessionWorkAdmission } from "../../sessions/session-lifecycle-admission.js"; @@ -224,20 +221,28 @@ export async function admitChatSend(params: { if (commitOutcome && expectedLeafEntryId !== undefined && !resolvedInjectionTarget) { // Runtime session identity resolves through the canonical SQLite accessor; // legacy/reset-archive files are read-only history fallbacks, never send targets. - const currentLeafEntryId = latestEntry?.sessionId - ? resolveSessionTranscriptActiveLeafEntryId( - readSessionTranscriptActiveLeafEvents({ + const activePathRelation = latestEntry?.sessionId + ? readSessionTranscriptActivePathEntryRelation( + { agentId, sessionId: latestEntry.sessionId, sessionKey: latestSession.canonicalKey, sessionEntry: latestEntry, storePath: latestSession.storePath, - }), + }, + expectedLeafEntryId, ) - : undefined; - // The lifecycle admission fence also blocks branch switching. Check the canonical - // transcript under that fence so a stale pane cannot dispatch onto another branch. - if ((currentLeafEntryId ?? null) !== expectedLeafEntryId) { + : expectedLeafEntryId === null + ? "exact" + : "off-path"; + // Branch switches preserve entry ids while rotating session ids. A supplied session id + // must fence exact and ancestor matches; omission remains legacy exact-only compatibility. + const matchesRequestedSession = + requestedSessionId === undefined || requestedSessionId === latestEntry?.sessionId; + const matchesActivePath = + activePathRelation === "exact" || + (activePathRelation === "ancestor" && requestedSessionId !== undefined); + if (!matchesRequestedSession || !matchesActivePath) { throw new Error(ACTIVE_LEAF_CHANGED_ERROR_REASON); } } diff --git a/src/gateway/server-methods/chat.directive-tags.test.ts b/src/gateway/server-methods/chat.directive-tags.test.ts index 2cab1dc6d23f..93ce1f5b9f62 100644 --- a/src/gateway/server-methods/chat.directive-tags.test.ts +++ b/src/gateway/server-methods/chat.directive-tags.test.ts @@ -33,10 +33,13 @@ import { import { testing as replyRunRegistryTesting } from "../../auto-reply/reply/reply-run-registry.test-support.js"; import type { MsgContext } from "../../auto-reply/templating.js"; import { + appendTranscriptEvent, appendTranscriptMessage, loadSessionEntry as loadSqliteSessionEntry, loadTranscriptEventsSync, replaceSessionEntry, + resolveSessionTranscriptActiveLeafEntryId, + switchSessionBranch, type SessionAccessScope, type SessionTranscriptReadScope, upsertSessionEntryCore, @@ -1405,6 +1408,182 @@ describe("chat directive tag stripping for non-streaming final payloads", () => expect(loadTranscriptEventsSync(transcriptScope())).toEqual(before); }); + it.each([ + { + name: "allows a same-session expected ancestor that remains on the active path", + sessionId: "current", + accepted: true, + }, + { + name: "rejects an active-path ancestor from a different requested session generation", + sessionId: "different-session-generation", + accepted: false, + }, + ])("$name", async ({ sessionId, accepted }) => { + await createGatewayUserTurnSqliteFixture(`openclaw-chat-send-active-ancestor-${sessionId}-`); + await appendTranscriptMessage(transcriptScope(), { + eventId: "rendered-leaf", + message: { role: "assistant", content: "rendered" }, + now: 1, + parentId: null, + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "memory-flush-user", + message: { role: "user", content: "maintenance prompt", display: false }, + now: 2, + parentId: "rendered-leaf", + }); + await appendTranscriptEvent(transcriptScope(), { + type: "compaction", + id: "background-compaction", + parentId: "memory-flush-user", + timestamp: "2026-08-10T00:00:00.000Z", + summary: "background maintenance", + firstKeptEntryId: "rendered-leaf", + tokensBefore: 10, + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "background-leaf", + message: { role: "assistant", content: "background append", display: false }, + now: 3, + parentId: "background-compaction", + }); + const before = loadTranscriptEventsSync(transcriptScope()); + const { context, respond, send } = createChatRequestFixture(); + + await send({ + idempotencyKey: `idem-active-ancestor-${sessionId}`, + requestParams: { + expectedLeafEntryId: "rendered-leaf", + sessionId: sessionId === "current" ? mockState.sessionId : sessionId, + }, + waitFor: "none", + }); + + const response = expectDefined( + lastRespondCall(respond), + "active ancestor response test invariant", + ); + expect(response[0]).toBe(accepted); + expect(context.addChatRun).toHaveBeenCalledTimes(accepted ? 1 : 0); + if (accepted) { + expect(response[1]).toEqual(expect.objectContaining({ status: "started" })); + } else { + expect(response[2]).toEqual( + expect.objectContaining({ details: { reason: "active-leaf-changed" } }), + ); + expect(loadTranscriptEventsSync(transcriptScope())).toEqual(before); + } + }); + + it("rejects a copied exact leaf from the session before a branch switch", async () => { + await createGatewayUserTurnSqliteFixture("openclaw-chat-send-rotated-exact-leaf-"); + await appendTranscriptMessage(transcriptScope(), { + eventId: "branch-root", + message: { role: "user", content: "root" }, + now: 1, + parentId: null, + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "copied-leaf", + message: { role: "assistant", content: "selected branch" }, + now: 2, + parentId: "branch-root", + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "active-sibling", + message: { role: "assistant", content: "active branch" }, + now: 3, + parentId: "branch-root", + }); + await waitForSessionTranscriptIndexReconcile({ + agentId: "main", + env: suiteFixtureEnv, + path: suiteDatabasePath, + }); + const staleSessionId = mockState.sessionId; + const switched = await switchSessionBranch({ + agentId: "main", + env: suiteFixtureEnv, + leafEntryId: "copied-leaf", + sessionKey: "agent:main:main", + storePath: suiteDatabasePath, + }); + expect(switched.status).toBe("created"); + if (switched.status !== "created") { + throw new Error("expected branch switch test invariant"); + } + expect(switched.entry.sessionId).not.toBe(staleSessionId); + mockState.sessionId = switched.entry.sessionId; + const before = loadTranscriptEventsSync(transcriptScope()); + expect(resolveSessionTranscriptActiveLeafEntryId(before)).toBe("copied-leaf"); + const { context, respond, send } = createChatRequestFixture(); + + await send({ + idempotencyKey: "idem-rotated-exact-leaf", + requestParams: { + expectedLeafEntryId: "copied-leaf", + sessionId: staleSessionId, + }, + waitFor: "none", + }); + + expect(lastRespondCall(respond)).toEqual([ + false, + undefined, + expect.objectContaining({ details: { reason: "active-leaf-changed" } }), + ]); + expect(context.addChatRun).not.toHaveBeenCalled(); + expect(mockState.lastDispatchCtx).toBeUndefined(); + expect(loadTranscriptEventsSync(transcriptScope())).toEqual(before); + }); + + it("rejects an expected sibling that is off the active path", async () => { + await createGatewayUserTurnSqliteFixture("openclaw-chat-send-off-path-sibling-"); + await appendTranscriptMessage(transcriptScope(), { + eventId: "branch-root", + message: { role: "user", content: "root" }, + now: 1, + parentId: null, + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "off-path-sibling", + message: { role: "assistant", content: "abandoned" }, + now: 2, + parentId: "branch-root", + }); + await appendTranscriptMessage(transcriptScope(), { + eventId: "active-sibling", + message: { role: "assistant", content: "active" }, + now: 3, + parentId: "branch-root", + }); + await waitForSessionTranscriptIndexReconcile({ + agentId: "main", + env: suiteFixtureEnv, + path: suiteDatabasePath, + }); + const before = loadTranscriptEventsSync(transcriptScope()); + const { context, respond, send } = createChatRequestFixture(); + + await send({ + idempotencyKey: "idem-off-path-sibling", + requestParams: { + expectedLeafEntryId: "off-path-sibling", + sessionId: mockState.sessionId, + }, + waitFor: "none", + }); + + expect(lastRespondCall(respond)).toEqual([ + false, + undefined, + expect.objectContaining({ details: { reason: "active-leaf-changed" } }), + ]); + expect(context.addChatRun).not.toHaveBeenCalled(); + expect(loadTranscriptEventsSync(transcriptScope())).toEqual(before); + }); + it("allows an expected empty leaf when the transcript is still empty", async () => { await createGatewayUserTurnSqliteFixture("openclaw-chat-send-matching-empty-leaf-"); const { context, respond, send } = createChatRequestFixture();