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

fix(sessions): preserve concurrent appends across reset (#127842)

Browse Source
This commit is contained in:
Vyctor H. Brzezowski
2026-08-23 13:15:06 -03:00
committed by GitHub
parent 3d75c27103
commit f9b5657cbd
9 changed files with 160 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
+1 -1
View File
@@ -2763,7 +2763,7 @@ src/config/sessions/session-accessor.transcript-range.ts 3
src/config/sessions/session-accessor.transcript-turn.ts 1
src/config/sessions/session-entry-json.ts 1
src/config/sessions/session-history-eviction.ts 2
src/config/sessions/session-reset-boundary-event.ts 11
src/config/sessions/session-reset-boundary-event.ts 5
src/config/sessions/session-sharing-store.ts 2
src/config/sessions/session-snapshot-merge.ts 26
src/config/sessions/session-sqlite-target.ts 3
src/config/sessions/session-accessor.reset-boundary-race.test.ts
+123
View File
@@ -0,0 +1,123 @@
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js";
import * as agentDatabase from "../../state/openclaw-agent-db.js";
import {
applySessionEntryLifecycleMutation,
loadTranscriptEvents,
resetSessionEntryLifecycle,
upsertSessionEntryCore,
} from "./session-accessor.js";
import {
readRecentSessionTranscriptActiveEvents,
waitForSessionTranscriptProjection,
} from "./session-accessor.sqlite-active-events.js";
import { appendTranscriptMessageSync } from "./session-accessor.sqlite-transcript-write.js";
const transactionInjection = vi.hoisted(() => ({ run: null as (() => void) | null }));
vi.mock("../../state/openclaw-agent-db.js", async (importOriginal) => {
const actual = await importOriginal<typeof agentDatabase>();
return {
...actual,
runOpenClawAgentWriteTransaction: <T>(
run: Parameters<typeof actual.runOpenClawAgentWriteTransaction<T>>[0],
options: Parameters<typeof actual.runOpenClawAgentWriteTransaction<T>>[1],
) => {
const inject = transactionInjection.run;
transactionInjection.run = null;
inject?.();
return actual.runOpenClawAgentWriteTransaction(run, options);
},
};
});
describe("reset boundary concurrency", () => {
const tempDirs: string[] = [];
let tempDir: string;
let storePath: string;
beforeEach(() => {
tempDir = makeTempDir(tempDirs, "openclaw-reset-boundary-race-");
storePath = path.join(tempDir, "sessions.json");
});
afterEach(() => {
transactionInjection.run = null;
agentDatabase.closeOpenClawAgentDatabasesForTest();
cleanupTempDirs(tempDirs);
});
it.each([
{
name: "single reset",
reset: async (scope: { sessionId: string; sessionKey: string; storePath: string }) =>
resetSessionEntryLifecycle({
buildNextEntry: () => ({ sessionId: "next-single", updatedAt: 20 }),
resetBoundaryReason: "reset",
storePath: scope.storePath,
target: { canonicalKey: scope.sessionKey, storeKeys: [scope.sessionKey] },
}),
},
{
name: "bulk lifecycle reset",
reset: async (scope: { sessionId: string; sessionKey: string; storePath: string }) =>
applySessionEntryLifecycleMutation({
skipMaintenance: true,
storePath: scope.storePath,
upserts: [
{
entry: { sessionId: "next-bulk", updatedAt: 20 },
resetBoundaryReason: "reset",
sessionKey: scope.sessionKey,
},
],
}),
},
])("parents the $name boundary after a concurrent accepted message", async ({ reset }) => {
const scope = {
sessionId: "current-session",
sessionKey: "agent:main:reset-race",
storePath,
};
await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 10 });
appendTranscriptMessageSync(scope, {
eventId: "initial",
message: { role: "user", content: "initial" },
parentId: null,
});
transactionInjection.run = () => {
appendTranscriptMessageSync(scope, {
eventId: "concurrent",
message: { role: "user", content: "accepted concurrently" },
parentId: "initial",
});
};
await reset(scope);
const raw = await loadTranscriptEvents(scope);
const boundary = raw.find(
(event) =>
event !== null &&
typeof event === "object" &&
!Array.isArray(event) &&
(event as { type?: unknown }).type === "reset",
);
expect(boundary).toMatchObject({ parentId: "concurrent" });
await waitForSessionTranscriptProjection(scope);
expect(
readRecentSessionTranscriptActiveEvents(scope, 10).map(
(event) => (event as { id?: unknown }).id,
),
).toContain("concurrent");
agentDatabase.closeOpenClawAgentDatabasesForTest();
await waitForSessionTranscriptProjection(scope);
expect(
readRecentSessionTranscriptActiveEvents(scope, 10).map(
(event) => (event as { id?: unknown }).id,
),
).toContain("concurrent");
});
});
src/config/sessions/session-accessor.sqlite-lifecycle-state.ts
+1 -10
View File
@@ -35,11 +35,9 @@ import type {
SessionEntryRemovalPlan,
} from "./session-accessor.sqlite-lifecycle-types.js";
import { coerceSqliteNumber } from "./session-accessor.sqlite-normalize.js";
import { loadTranscriptEventsFromDatabase } from "./session-accessor.sqlite-read.js";
import { collectSessionStateIdsForEntry } from "./session-accessor.sqlite-references.js";
import { cloneSessionEntry, getSessionKysely } from "./session-accessor.sqlite-scope.js";
import { parseSessionEntryJson as parseSessionEntryRow } from "./session-accessor.sqlite-status.js";
import { buildSessionResetBoundaryPlan } from "./session-reset-boundary-event.js";
import { deleteSessionTranscriptIndexInTransaction } from "./session-transcript-index.js";
import type { SessionEntry } from "./types.js";
@@ -394,19 +392,12 @@ export async function projectSessionEntryLifecycleMutation(
const cloned = cloneSessionEntry(entry);
store[sessionKey] = cloned;
changedSessionKeys.add(sessionKey);
const resetBoundaryPlan =
upsert.resetBoundaryReason && expectedEntry?.sessionId
? await buildSessionResetBoundaryPlan({
events: loadTranscriptEventsFromDatabase(database, expectedEntry.sessionId),
reason: upsert.resetBoundaryReason,
})
: undefined;
upsertedEntries.push({
expectedEntry,
sessionKey,
entry: cloned,
...(upsert.routeContext !== undefined ? { routeContext: upsert.routeContext } : {}),
...(resetBoundaryPlan ? { resetBoundaryPlan } : {}),
...(upsert.resetBoundaryReason ? { resetBoundaryReason: upsert.resetBoundaryReason } : {}),
});
}
const referencedSessionIds = collectProjectedReferencedSessionIds({
src/config/sessions/session-accessor.sqlite-lifecycle-types.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import type { ConversationRouteContext } from "./conversation-route-context.js";
import type { SessionLifecycleArchivedTranscript } from "./session-accessor.lifecycle-types.js";
import type { SessionStateDeletePlan } from "./session-accessor.sqlite-archive.js";
import type { SessionEntryLifecycleRemoval } from "./session-accessor.sqlite-contract.js";
import type { SessionResetBoundaryPlan } from "./session-reset-boundary-event.js";
import type { SessionResetBoundaryReason } from "./session-reset-boundary-event.js";
import type { SessionEntry } from "./types.js";
// Shared plan shapes only. Runtime ownership stays in maintenance and lifecycle-state.
@@ -39,7 +39,7 @@ export type ProjectedLifecycleMutation = {
entry: SessionEntry;
expectedEntry: SessionEntry | undefined;
routeContext?: ConversationRouteContext | null;
resetBoundaryPlan?: SessionResetBoundaryPlan;
resetBoundaryReason?: SessionResetBoundaryReason;
sessionKey: string;
}>;
};
src/config/sessions/session-accessor.sqlite-lifecycle.ts
+10 -12
View File
@@ -67,7 +67,7 @@ import {
collectAdmissionProtectedSessionIds,
kickSessionHistoryDiskBudgetMaintenance,
} from "./session-history-eviction.js";
import { buildSessionResetBoundaryPlan } from "./session-reset-boundary-event.js";
import { buildSessionResetBoundaryEvent } from "./session-reset-boundary-event.js";
import type { InternalSessionEntry as SessionEntry } from "./types.js";
// Single-target lifecycle owner: cleanup, reset, guarded delete, and trusted rollback.
@@ -189,15 +189,10 @@ export async function resetSessionEntryLifecycle(
currentEntry: current ? cloneSessionEntry(current.entry) : undefined,
primaryKey: params.target.canonicalKey,
});
const resetBoundaryPlan =
const shouldAppendResetBoundary =
params.resetBoundaryReason &&
current?.entry.sessionId &&
!sqliteSessionEntriesEqual(current.entry, nextEntry)
? await buildSessionResetBoundaryPlan({
events: loadTranscriptEventsFromDatabase(database, current.entry.sessionId),
reason: params.resetBoundaryReason,
})
: undefined;
!sqliteSessionEntriesEqual(current.entry, nextEntry);
const mutation: ResetSessionEntryLifecycleMutation = {
nextEntry: cloneSessionEntry(nextEntry),
...(current ? { previousEntry: cloneSessionEntry(current.entry) } : {}),
@@ -205,8 +200,11 @@ export async function resetSessionEntryLifecycle(
};
runOpenClawAgentWriteTransaction((transactionDb) => {
assertLifecycleTargetUnchanged(transactionDb, params.target, current?.entry, "reset");
if (resetBoundaryPlan && current?.entry.sessionId) {
const events = [...resetBoundaryPlan.seedEvents, resetBoundaryPlan.event];
if (shouldAppendResetBoundary && current?.entry.sessionId && params.resetBoundaryReason) {
const event = buildSessionResetBoundaryEvent({
events: loadTranscriptEventsFromDatabase(transactionDb, current.entry.sessionId),
reason: params.resetBoundaryReason,
});
const appended = appendTranscriptEventsInTransaction(
transactionDb,
{
@@ -214,9 +212,9 @@ export async function resetSessionEntryLifecycle(
sessionId: current.entry.sessionId,
sessionKey: current.key,
},
events,
[event],
);
if (appended !== events.length) {
if (appended !== 1) {
throw new Error(`Failed to append reset boundary for ${current.key}`);
}
}
src/config/sessions/session-accessor.sqlite-projection.ts
+10 -5
View File
@@ -67,6 +67,7 @@ import {
applySessionEntryMaintenance,
finalizeSessionEntryMaintenancePlansAfterWriterReleaseBestEffort,
} from "./session-accessor.sqlite-maintenance.js";
import { loadTranscriptEventsFromDatabase } from "./session-accessor.sqlite-read.js";
import { applySessionEntryExactReplacements } from "./session-accessor.sqlite-replacement-projection.js";
import {
cloneSessionEntry,
@@ -77,6 +78,7 @@ import {
toDatabaseOptions,
} from "./session-accessor.sqlite-scope.js";
import { appendTranscriptEventsInTransaction } from "./session-accessor.sqlite-transcript-store.js";
import { buildSessionResetBoundaryEvent } from "./session-reset-boundary-event.js";
import { resolveMaintenanceConfig } from "./store-maintenance-runtime.js";
import type { ResolvedSessionMaintenanceConfig } from "./store-maintenance.js";
import type { SessionEntry } from "./types.js";
@@ -340,7 +342,7 @@ export async function applySessionEntryLifecycleMutation(params: {
entry,
expectedEntry,
routeContext,
resetBoundaryPlan,
resetBoundaryReason,
} of projected.upsertedEntries) {
const sameKeyRemoval = validatedRemovals.find(
(removal) => removal.sessionKey === sessionKey,
@@ -363,14 +365,17 @@ export async function applySessionEntryLifecycleMutation(params: {
if (sameKeyRemoval && !shouldRemoveSessionEntry(currentEntry, sameKeyRemoval.removal)) {
throw new Error(`SQLite session entry has stale lifecycle state for ${sessionKey}`);
}
if (resetBoundaryPlan && expectedEntry?.sessionId) {
const events = [...resetBoundaryPlan.seedEvents, resetBoundaryPlan.event];
if (resetBoundaryReason && expectedEntry?.sessionId) {
const event = buildSessionResetBoundaryEvent({
events: loadTranscriptEventsFromDatabase(transactionDb, expectedEntry.sessionId),
reason: resetBoundaryReason,
});
const appended = appendTranscriptEventsInTransaction(
transactionDb,
{ ...resolved, sessionId: expectedEntry.sessionId, sessionKey },
events,
[event],
);
if (appended !== events.length) {
if (appended !== 1) {
throw new Error(`Failed to append reset boundary for ${sessionKey}`);
}
}
src/config/sessions/session-reset-boundary-event.test.ts
+9 -123
View File
@@ -1,8 +1,5 @@
import fs from "node:fs/promises";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { withTestDir } from "../../test-helpers/temp-dir.js";
import { buildSessionResetBoundaryPlan } from "./session-reset-boundary-event.js";
import { buildSessionResetBoundaryEvent } from "./session-reset-boundary-event.js";
function message(params: {
id: string;
@@ -60,12 +57,10 @@ describe("reset boundary planning", () => {
};
expect(
(
await buildSessionResetBoundaryPlan({
events: [oldUser, oldAssistant, keptUser, keptAssistant, firstReset],
reason: "reset",
})
).event,
buildSessionResetBoundaryEvent({
events: [oldUser, oldAssistant, keptUser, keptAssistant, firstReset],
reason: "reset",
}),
).toMatchObject({
parentId: firstReset.id,
firstKeptEntryId: keptUser.id,
@@ -105,122 +100,13 @@ describe("reset boundary planning", () => {
};
expect(
(
await buildSessionResetBoundaryPlan({
events: [discarded, keptUser, keptAssistant, compaction],
reason: "new",
})
).event,
buildSessionResetBoundaryEvent({
events: [discarded, keptUser, keptAssistant, compaction],
reason: "new",
}),
).toMatchObject({
parentId: compaction.id,
firstKeptEntryId: keptUser.id,
});
});
it("seeds only the bounded replay tail from a legacy transcript", async () => {
await withTestDir({ prefix: "openclaw-reset-boundary-" }, async (dir) => {
const sessionFile = path.join(dir, "legacy.jsonl");
const records = Array.from({ length: 20 }, (_, index) =>
message({
id: `message-${index}`,
parentId: index === 0 ? null : `message-${index - 1}`,
role: index % 2 === 0 ? "user" : "assistant",
content: `message ${index}`,
second: index,
}),
);
await fs.writeFile(
sessionFile,
`${records.map((entry) => JSON.stringify(entry)).join("\n")}\n`,
);
const plan = await buildSessionResetBoundaryPlan({
events: [],
legacySessionFile: sessionFile,
reason: "new",
});
expect(plan.seedEvents).toHaveLength(6);
expect(plan.seedEvents.map((entry) => (entry as { id?: string }).id)).toEqual(
records.slice(-6).map((entry) => entry.id),
);
expect(plan.event.firstKeptEntryId).toBe("message-14");
const metadataOnlyPlan = await buildSessionResetBoundaryPlan({
events: [{ type: "model_change", id: "metadata-only", parentId: null }],
legacySessionFile: sessionFile,
reason: "new",
});
expect(metadataOnlyPlan.seedEvents).toHaveLength(6);
expect(metadataOnlyPlan.event.firstKeptEntryId).toBe("message-14");
});
});
it("respects legacy reset cuts and reparents the selected tail", async () => {
await withTestDir({ prefix: "openclaw-reset-boundary-" }, async (dir) => {
const sessionFile = path.join(dir, "legacy-reset.jsonl");
const oldUser = message({
id: "legacy-old-user",
parentId: null,
role: "user",
content: "discarded",
second: 1,
});
const oldAssistant = message({
id: "legacy-old-assistant",
parentId: oldUser.id,
role: "assistant",
content: "discarded answer",
second: 2,
});
const keptUser = message({
id: "legacy-kept-user",
parentId: oldAssistant.id,
role: "user",
content: "kept",
second: 3,
});
const toolResult = {
type: "message",
id: "legacy-tool-result",
parentId: keptUser.id,
timestamp: "2026-07-22T00:00:04.000Z",
message: { role: "toolResult", content: "tool" },
};
const keptAssistant = message({
id: "legacy-kept-assistant",
parentId: toolResult.id,
role: "assistant",
content: "kept answer",
second: 5,
});
const reset = {
type: "reset",
id: "legacy-reset",
parentId: keptAssistant.id,
timestamp: "2026-07-22T00:00:06.000Z",
reason: "new",
firstKeptEntryId: keptUser.id,
};
await fs.writeFile(
sessionFile,
`${[oldUser, oldAssistant, keptUser, toolResult, keptAssistant, reset]
.map((entry) => JSON.stringify(entry))
.join("\n")}\n`,
);
const plan = await buildSessionResetBoundaryPlan({
events: [],
legacySessionFile: sessionFile,
reason: "reset",
});
expect(plan.seedEvents).toEqual([
expect.objectContaining({ id: keptUser.id, parentId: null }),
expect.objectContaining({ id: keptAssistant.id, parentId: keptUser.id }),
]);
expect(JSON.stringify(plan.seedEvents)).not.toContain("discarded");
expect(plan.event.firstKeptEntryId).toBe(keptUser.id);
});
});
});
src/config/sessions/session-reset-boundary-event.ts
+2 -96
View File
@@ -1,11 +1,5 @@
import { randomUUID } from "node:crypto";
import path from "node:path";
import {
DEFAULT_REPLAY_MAX_MESSAGES,
replayableTranscriptRole,
selectRecentUserAssistantReplayRecords,
} from "./transcript-replay.js";
import { streamSessionTranscriptLinesReverse } from "./transcript-stream.js";
import { selectRecentUserAssistantReplayRecords } from "./transcript-replay.js";
import { selectSessionTranscriptLeafControlledPath } from "./transcript-tree.js";
export type SessionResetBoundaryReason = "new" | "reset" | "idle" | "daily" | "cron-stale";
@@ -19,11 +13,6 @@ type SessionResetBoundaryEvent = {
firstKeptEntryId?: string;
};
export type SessionResetBoundaryPlan = {
event: SessionResetBoundaryEvent;
seedEvents: unknown[];
};
function recordId(record: unknown): string | undefined {
if (!record || typeof record !== "object" || Array.isArray(record)) {
return undefined;
@@ -73,7 +62,7 @@ function projectLatestBoundaryWindow(entries: readonly unknown[]): unknown[] {
return [...kept, ...entries.slice(boundaryIndex + 1)];
}
function buildSessionResetBoundaryEvent(params: {
export function buildSessionResetBoundaryEvent(params: {
events: readonly unknown[];
reason: SessionResetBoundaryReason;
}): SessionResetBoundaryEvent {
@@ -98,86 +87,3 @@ function buildSessionResetBoundaryEvent(params: {
...(firstKeptEntryId ? { firstKeptEntryId } : {}),
};
}
async function readLegacyTranscriptEvents(sessionFile: string | undefined): Promise<unknown[]> {
const filePath = sessionFile?.trim();
if (!filePath || !path.isAbsolute(filePath) || !filePath.endsWith(".jsonl")) {
return [];
}
try {
const newestFirst: unknown[] = [];
let boundaryFirstKeptEntryId: string | undefined;
let foundBoundary = false;
for await (const line of streamSessionTranscriptLinesReverse(filePath)) {
let record: unknown;
try {
record = JSON.parse(line) as unknown;
} catch {
continue;
}
const type =
record && typeof record === "object" && !Array.isArray(record)
? (record as { type?: unknown }).type
: undefined;
if (!foundBoundary && (type === "reset" || type === "compaction")) {
foundBoundary = true;
const firstKept = (record as { firstKeptEntryId?: unknown }).firstKeptEntryId;
boundaryFirstKeptEntryId =
typeof firstKept === "string" && firstKept.trim() ? firstKept : undefined;
if (!boundaryFirstKeptEntryId) {
break;
}
continue;
}
if (foundBoundary && (type === "reset" || type === "compaction")) {
break;
}
if (replayableTranscriptRole(record as Parameters<typeof replayableTranscriptRole>[0])) {
newestFirst.push(record);
}
if (
newestFirst.length >= DEFAULT_REPLAY_MAX_MESSAGES ||
(foundBoundary && recordId(record) === boundaryFirstKeptEntryId)
) {
break;
}
}
const selected = selectRecentUserAssistantReplayRecords(newestFirst.toReversed());
return selected.map((record, index) =>
Object.assign({}, record as Record<string, unknown>, {
parentId: index === 0 ? null : (recordId(selected[index - 1]) ?? null),
}),
);
} catch {
return [];
}
}
export async function buildSessionResetBoundaryPlan(params: {
events: readonly unknown[];
legacySessionFile?: string;
reason: SessionResetBoundaryReason;
}): Promise<SessionResetBoundaryPlan> {
const hasConversationEvents = params.events.some((event) => {
const type =
event !== null && typeof event === "object" && !Array.isArray(event)
? (event as { type?: unknown }).type
: undefined;
return type === "message" || type === "compaction" || type === "reset";
});
const legacyEvents = hasConversationEvents
? []
: await readLegacyTranscriptEvents(params.legacySessionFile);
const seedEvents = legacyEvents.filter(
(event) =>
event !== null &&
typeof event === "object" &&
!Array.isArray(event) &&
(event as { type?: unknown }).type !== "session",
);
const events = seedEvents.length > 0 ? [...params.events, ...seedEvents] : params.events;
return {
event: buildSessionResetBoundaryEvent({ events, reason: params.reason }),
seedEvents,
};
}
src/config/sessions/transcript-replay.ts
+2 -4
View File
@@ -1,7 +1,7 @@
// Selects safe user/assistant tails for in-log lifecycle boundaries.
/** Tail kept so DM continuity survives silent session rotations. */
export const DEFAULT_REPLAY_MAX_MESSAGES = 6;
const DEFAULT_REPLAY_MAX_MESSAGES = 6;
type SessionRecord = {
type?: unknown;
@@ -19,9 +19,7 @@ function isValidReplayTimestamp(value: unknown): boolean {
return typeof value === "string" && value.trim().length > 0;
}
export function replayableTranscriptRole(
record: SessionRecord | null,
): "user" | "assistant" | undefined {
function replayableTranscriptRole(record: SessionRecord | null): "user" | "assistant" | undefined {
if (
!record ||
record.type !== "message" ||