ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-25 11:55:47 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: send follow-ups after background transcript updates (#121332)

Browse Source 
* fix(gateway): allow sends after linear transcript advance

* fix(gateway): keep active-path check internal

* fix(gateway): fence exact leaves by session generation

* test(gateway): use canonical branch-switch key

* refactor: consolidate active path relation reader

---------

Co-authored-by: scotthuang <scotthuang@tencent.com>
Co-authored-by: fuller-stack-dev <263060202+fuller-stack-dev@users.noreply.github.com>
This commit is contained in:
scotthuang
2026-08-15 02:13:01 +08:00
committed by GitHub
parent 3f006ba0fc
commit 79a4d512d4
5 changed files with 214 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/sessions/session-accessor.sqlite-active-events.test.ts
+5 -7
View File
@@ -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();
});
src/config/sessions/session-accessor.sqlite-active-events.ts
+12 -15
View File
@@ -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";
});
}
src/config/sessions/session-accessor.ts
+1 -1
View File
@@ -241,7 +241,7 @@ export {
readSessionTranscriptActiveStats,
readSessionTranscriptBoundedMessageTailPage,
readRecentSessionTranscriptMessageEvents,
readSessionTranscriptActiveLeafEvents,
readSessionTranscriptActivePathEntryRelation,
readSessionTranscriptMessageAnchorPage,
readSessionTranscriptMessageEventById,
readSessionTranscriptMessageEventCount,
src/gateway/server-methods/chat-send-admission.ts
+17 -12
View File
@@ -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);
}
}
src/gateway/server-methods/chat.directive-tags.test.ts
+179
View File
@@ -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();