github/openclaw-openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-08 10:02:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve sqlite command session updates

Browse Source
This commit is contained in:
Josh Lehman
2026-07-06 07:43:58 -07:00
parent 518e23a716
commit 77665163bb
21 changed files with 975 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/auto-reply/reply/agent-runner-session-reset.test.ts
View File

@@ -80,6 +80,13 @@ describe("resetReplyRunSession", () => {
fallbackNoticeSelectedModel: "anthropic/claude",
fallbackNoticeActiveModel: "openai/gpt",
fallbackNoticeReason: "rate limit",
compactionCount: 4,
memoryFlushAt: 50,
memoryFlushCompactionCount: 3,
memoryFlushContextHash: "context-hash",
memoryFlushFailureCount: 2,
memoryFlushLastFailedAt: 60,
memoryFlushLastFailureError: "memory failed",
systemPromptReport: {
source: "run",
generatedAt: 1,
@@ -126,6 +133,13 @@ describe("resetReplyRunSession", () => {
expect(activeSessionEntry?.fallbackNoticeActiveModel).toBeUndefined();
expect(activeSessionEntry?.fallbackNoticeReason).toBeUndefined();
expect(activeSessionEntry?.systemPromptReport).toBeUndefined();
expect(activeSessionEntry?.compactionCount).toBe(0);
expect(activeSessionEntry?.memoryFlushAt).toBeUndefined();
expect(activeSessionEntry?.memoryFlushCompactionCount).toBeUndefined();
expect(activeSessionEntry?.memoryFlushContextHash).toBeUndefined();
expect(activeSessionEntry?.memoryFlushFailureCount).toBeUndefined();
expect(activeSessionEntry?.memoryFlushLastFailedAt).toBeUndefined();
expect(activeSessionEntry?.memoryFlushLastFailureError).toBeUndefined();
expect(refreshQueuedFollowupSessionMock).toHaveBeenCalledWith({
key: "main",
previousSessionId: "session",
@@ -138,6 +152,8 @@ describe("resetReplyRunSession", () => {
expect(persisted?.sessionId).toBe(activeSessionEntry?.sessionId);
expect(persisted?.contextBudgetStatus).toBeUndefined();
expect(persisted?.fallbackNoticeReason).toBeUndefined();
expect(persisted?.compactionCount).toBe(0);
expect(persisted?.memoryFlushAt).toBeUndefined();
});
it("cleans up the old transcript when requested", async () => {

7
src/auto-reply/reply/agent-runner-session-reset.ts
View File

@@ -79,6 +79,13 @@ export async function resetReplyRunSession(params: {
fallbackNoticeSelectedModel: undefined,
fallbackNoticeActiveModel: undefined,
fallbackNoticeReason: undefined,
compactionCount: 0,
memoryFlushAt: undefined,
memoryFlushCompactionCount: undefined,
memoryFlushContextHash: undefined,
memoryFlushFailureCount: undefined,
memoryFlushLastFailedAt: undefined,
memoryFlushLastFailureError: undefined,
};
const agentId = resolveAgentIdFromSessionKey(params.sessionKey);
const nextSessionFile = formatSqliteSessionFileMarker({

17
src/auto-reply/reply/commands-core.ts
View File

@@ -35,7 +35,20 @@ export async function handleCommands(params: HandleCommandsParams): Promise<Comm
if (HANDLERS === null) {
HANDLERS = (await loadCommandHandlersRuntime()).loadCommandHandlers();
}
const resetResult = await maybeHandleResetCommand(params);
const allowCreateSessionEntry = params.allowCreateSessionEntry === true;
const initialSessionEntry =
params.initialSessionEntry ??
(allowCreateSessionEntry
? undefined
: params.sessionEntry
? { ...params.sessionEntry }
: undefined);
const commandParams: HandleCommandsParams = {
...params,
initialSessionEntry,
allowCreateSessionEntry,
};
const resetResult = await maybeHandleResetCommand(commandParams);
if (resetResult) {
return normalizeCommandHandlerResult(resetResult);
}
@@ -47,7 +60,7 @@ export async function handleCommands(params: HandleCommandsParams): Promise<Comm
});
for (const handler of HANDLERS) {
const result = await handler(params, allowTextCommands);
const result = await handler(commandParams, allowTextCommands);
if (result) {
return normalizeCommandHandlerResult(result);
}

5
src/auto-reply/reply/commands-dock.ts
View File

@@ -175,7 +175,10 @@ export const handleDockCommand: CommandHandler = async (params, allowTextCommand
sessionEntry.lastTo = target.peerId;
sessionEntry.lastAccountId = resolveTargetChannelAccountId(params, targetChannel);
params.sessionEntry = sessionEntry;
const persisted = await persistSessionEntry(params);
const persisted = await persistSessionEntry({
...params,
touchedFields: ["lastChannel", "lastTo", "lastAccountId"],
});
if (!persisted) {
return {
shouldContinue: false,

248
src/auto-reply/reply/commands-session-store.test.ts
View File

@@ -1,7 +1,7 @@
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { describe, expect, it, vi } from "vitest";
import { loadSessionEntry, replaceSessionEntry } from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import { persistAbortTargetEntry, persistSessionEntry } from "./commands-session-store.js";
@@ -16,6 +16,63 @@ async function withTempStore<T>(run: (storePath: string) => Promise<T>): Promise
}
describe("commands session store persistence", () => {
it("creates a missing row for the first command-only session mutation", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:first-command";
const entry: SessionEntry = {
sessionId: "first-command-session",
updatedAt: 1,
responseUsage: "tokens",
};
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
await expect(
persistSessionEntry({
allowCreateSessionEntry: true,
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
touchedFields: ["responseUsage"],
}),
).resolves.toBe(true);
const persisted = loadSessionEntry({ storePath, sessionKey });
expect(persisted).toMatchObject({
sessionId: "first-command-session",
responseUsage: "tokens",
});
expect(sessionStore[sessionKey]).toMatchObject({
sessionId: "first-command-session",
responseUsage: "tokens",
});
});
});
it("does not recreate a missing row without explicit create ownership", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:missing-existing";
const entry: SessionEntry = {
sessionId: "missing-existing-session",
updatedAt: 1,
responseUsage: "tokens",
};
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
await expect(
persistSessionEntry({
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
touchedFields: ["responseUsage"],
}),
).resolves.toBe(false);
expect(loadSessionEntry({ storePath, sessionKey })).toBeUndefined();
});
});
it("persists a single command session entry through the accessor", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:command";
@@ -29,7 +86,14 @@ describe("commands session store persistence", () => {
sessionId: "other-session",
updatedAt: 2,
};
await replaceSessionEntry({ storePath, sessionKey }, { ...entry });
const seedEntry = { ...entry };
await persistSessionEntry({
allowCreateSessionEntry: true,
sessionEntry: seedEntry,
sessionStore: { [sessionKey]: seedEntry },
sessionKey,
storePath,
});
await replaceSessionEntry({ storePath, sessionKey: otherKey }, { ...otherEntry });
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
@@ -44,7 +108,11 @@ describe("commands session store persistence", () => {
const persisted = loadSessionEntry({ storePath, sessionKey });
const persistedOther = loadSessionEntry({ storePath, sessionKey: otherKey });
expect(sessionStore[sessionKey]).toBe(entry);
expect(sessionStore[sessionKey]).toMatchObject({
sessionId: "command-session",
model: "gpt-5.5",
});
expect(sessionStore[sessionKey]?.updatedAt).toBeGreaterThanOrEqual(entry.updatedAt);
expect(entry.updatedAt).not.toBe(1);
expect(persisted).toMatchObject({
sessionId: "command-session",
@@ -55,6 +123,180 @@ describe("commands session store persistence", () => {
});
});
it("persists command state without reverting concurrent session management", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:command";
const otherKey = "agent:main:other";
const entry: SessionEntry = {
sessionId: "command-session",
updatedAt: 1,
model: "gpt-5.5",
label: "Before rename",
pinnedAt: 100,
};
const otherEntry: SessionEntry = {
sessionId: "other-session",
updatedAt: 2,
};
const concurrentUpdatedAt = 300;
const concurrentEntry = {
...entry,
updatedAt: concurrentUpdatedAt,
label: "After rename",
pinnedAt: undefined,
};
await persistSessionEntry({
allowCreateSessionEntry: true,
sessionEntry: concurrentEntry,
sessionStore: { [sessionKey]: concurrentEntry },
sessionKey,
storePath,
});
await replaceSessionEntry({ storePath, sessionKey: otherKey }, { ...otherEntry });
const sessionStore: Record<string, SessionEntry> = { [sessionKey]: entry };
const nowSpy = vi.spyOn(Date, "now").mockReturnValueOnce(200).mockReturnValue(400);
try {
await expect(
persistSessionEntry({
sessionEntry: entry,
sessionStore,
sessionKey,
storePath,
}),
).resolves.toBe(true);
} finally {
nowSpy.mockRestore();
}
const persisted = loadSessionEntry({ storePath, sessionKey });
const persistedOther = loadSessionEntry({ storePath, sessionKey: otherKey });
expect(entry.updatedAt).not.toBe(1);
expect(sessionStore[sessionKey]).toMatchObject({
sessionId: "command-session",
label: "After rename",
model: "gpt-5.5",
});
expect(sessionStore[sessionKey]?.updatedAt).toBeGreaterThanOrEqual(concurrentUpdatedAt);
expect(sessionStore[sessionKey]?.pinnedAt).toBeUndefined();
expect(persisted).toMatchObject({
sessionId: "command-session",
label: "After rename",
model: "gpt-5.5",
});
expect(persisted?.updatedAt).toBeGreaterThanOrEqual(concurrentUpdatedAt);
expect(persisted?.pinnedAt).toBeUndefined();
expect(persistedOther).toStrictEqual(otherEntry);
});
});
it("rejects command persistence after the session rotates", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:command";
const initialEntry: SessionEntry = {
sessionId: "session-1",
updatedAt: 1,
queueMode: "collect",
};
const sessionEntry: SessionEntry = {
...initialEntry,
queueMode: "followup",
};
const rotatedEntry: SessionEntry = {
sessionId: "session-2",
updatedAt: 3,
queueMode: "interrupt",
};
await replaceSessionEntry({ storePath, sessionKey }, rotatedEntry);
const sessionStore = { [sessionKey]: sessionEntry };
await expect(
persistSessionEntry({
initialSessionEntry: initialEntry,
sessionEntry,
sessionStore,
sessionKey,
storePath,
}),
).resolves.toBe(false);
expect(sessionStore[sessionKey]).toEqual(rotatedEntry);
expect(loadSessionEntry({ storePath, sessionKey })).toEqual(rotatedEntry);
});
});
it("rejects an explicit same-value command after a concurrent change", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:command";
const initialEntry: SessionEntry = {
sessionId: "session-1",
updatedAt: 1,
sendPolicy: "deny",
};
const sessionEntry = { ...initialEntry };
const concurrentEntry: SessionEntry = {
...initialEntry,
updatedAt: 2,
sendPolicy: "allow",
};
await replaceSessionEntry({ storePath, sessionKey }, concurrentEntry);
const sessionStore = { [sessionKey]: sessionEntry };
await expect(
persistSessionEntry({
initialSessionEntry: initialEntry,
sessionEntry,
sessionStore,
sessionKey,
storePath,
touchedFields: ["sendPolicy"],
}),
).resolves.toBe(false);
expect(sessionStore[sessionKey]).toMatchObject({
sessionId: "session-1",
sendPolicy: "allow",
});
});
});
it("rejects a grouped command before committing any non-conflicting field", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:command";
const initialEntry: SessionEntry = {
sessionId: "session-1",
updatedAt: 1,
groupActivation: "mention",
groupActivationNeedsSystemIntro: true,
};
const sessionEntry: SessionEntry = {
...initialEntry,
groupActivation: "always",
};
const concurrentEntry: SessionEntry = {
...initialEntry,
updatedAt: 2,
groupActivationNeedsSystemIntro: false,
};
await replaceSessionEntry({ storePath, sessionKey }, concurrentEntry);
const sessionStore = { [sessionKey]: sessionEntry };
await expect(
persistSessionEntry({
initialSessionEntry: initialEntry,
sessionEntry,
sessionStore,
sessionKey,
storePath,
touchedFields: ["groupActivation", "groupActivationNeedsSystemIntro"],
}),
).resolves.toBe(false);
expect(sessionStore[sessionKey]).toEqual(concurrentEntry);
expect(loadSessionEntry({ storePath, sessionKey })).toEqual(concurrentEntry);
});
});
it("falls back to the supplied abort target when the persisted row is missing", async () => {
await withTempStore(async (storePath) => {
const sessionKey = "agent:main:abort-target";

53
src/auto-reply/reply/commands-session-store.ts
View File

@@ -1,14 +1,21 @@
// Shared session-store helpers for command handlers that mutate sessions.
import { resolveSessionStoreEntry, type SessionEntry } from "../../config/sessions.js";
import { patchSessionEntry } from "../../config/sessions/session-accessor.js";
import { sessionSnapshotChangesApplied } from "../../config/sessions/session-snapshot-merge.js";
import { applyAbortCutoffToSessionEntry, type AbortCutoff } from "./abort-cutoff.js";
import type { CommandHandler } from "./commands-types.js";
import type { CommandHandler, CommandHandlerResult } from "./commands-types.js";
import { persistReplySessionEntry } from "./session-entry-persistence.js";
type CommandParams = Parameters<CommandHandler>[0];
type PersistSessionEntryParams = Pick<
CommandParams,
"sessionEntry" | "sessionStore" | "sessionKey" | "storePath"
>;
| "allowCreateSessionEntry"
| "initialSessionEntry"
| "sessionEntry"
| "sessionKey"
| "sessionStore"
| "storePath"
> & { touchedFields?: ReadonlyArray<keyof SessionEntry> };
/** Resolves a command target entry through canonical and legacy session keys. */
export function resolveCommandSessionEntryForKey(
@@ -33,25 +40,47 @@ export async function persistSessionEntry(params: PersistSessionEntryParams): Pr
return false;
}
const sessionEntry = params.sessionEntry;
const creatingSession = params.allowCreateSessionEntry === true;
const initialEntry = params.initialSessionEntry ?? { ...sessionEntry };
sessionEntry.updatedAt = Date.now();
params.sessionStore[params.sessionKey] = sessionEntry;
if (params.storePath) {
// Slash commands mutate one known session entry; skipping global session
// maintenance avoids scanning the whole sessions directory for simple
// command-only writes.
await patchSessionEntry(
{ storePath: params.storePath, sessionKey: params.sessionKey },
() => sessionEntry,
{
fallbackEntry: sessionEntry,
replaceEntry: true,
skipMaintenance: true,
},
);
const persistence = await persistReplySessionEntry({
storePath: params.storePath,
sessionKey: params.sessionKey,
allowCreate: creatingSession,
initialEntry,
entry: sessionEntry,
skipMaintenance: true,
touchedFields: params.touchedFields,
});
if (persistence.status === "lifecycle-invalidated") {
if (persistence.entry) {
params.sessionStore[params.sessionKey] = persistence.entry;
}
return false;
}
params.sessionStore[params.sessionKey] = persistence.entry;
return sessionSnapshotChangesApplied({
initial: initialEntry,
next: sessionEntry,
current: persistence.entry,
touchedFields: params.touchedFields,
});
}
return true;
}
export function sessionEntryPersistenceConflictReply(): CommandHandlerResult {
return {
shouldContinue: false,
reply: { text: "⚠️ Session changed before this setting could be saved. Retry the command." },
};
}
export async function persistAbortTargetEntry(params: {
entry?: SessionEntry;
key?: string;

5
src/auto-reply/reply/commands-session.activation.test.ts
View File

@@ -90,6 +90,9 @@ describe("handleActivationCommand", () => {
});
expect(params.sessionEntry?.groupActivation).toBe("always");
expect(params.sessionEntry?.groupActivationNeedsSystemIntro).toBe(true);
expect(persistSessionEntryMock).toHaveBeenCalledWith(params);
expect(persistSessionEntryMock).toHaveBeenCalledWith({
...params,
touchedFields: ["groupActivation", "groupActivationNeedsSystemIntro"],
});
});
});

58
src/auto-reply/reply/commands-session.ts
View File

@@ -48,7 +48,10 @@ import {
import { resolveCommandSurfaceChannel } from "./channel-context.js";
import { rejectNonOwnerCommand, rejectUnauthorizedCommand } from "./command-gates.js";
import { handleAbortTrigger, handleStopCommand } from "./commands-session-abort.js";
import { persistSessionEntry } from "./commands-session-store.js";
import {
persistSessionEntry,
sessionEntryPersistenceConflictReply,
} from "./commands-session-store.js";
import type { CommandHandler, HandleCommandsParams } from "./commands-types.js";
import { resolveConversationBindingContextFromAcpCommand } from "./conversation-binding-input.js";
@@ -230,7 +233,14 @@ export const handleActivationCommand: CommandHandler = async (params, allowTextC
if (params.sessionEntry && params.sessionStore && params.sessionKey) {
params.sessionEntry.groupActivation = activationCommand.mode;
params.sessionEntry.groupActivationNeedsSystemIntro = true;
await persistSessionEntry(params);
if (
!(await persistSessionEntry({
...params,
touchedFields: ["groupActivation", "groupActivationNeedsSystemIntro"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
}
return {
shouldContinue: false,
@@ -268,7 +278,9 @@ export const handleSendPolicyCommand: CommandHandler = async (params, allowTextC
} else {
params.sessionEntry.sendPolicy = sendPolicyCommand.mode;
}
await persistSessionEntry(params);
if (!(await persistSessionEntry({ ...params, touchedFields: ["sendPolicy"] }))) {
return sessionEntryPersistenceConflictReply();
}
}
const label =
sendPolicyCommand.mode === "inherit"
@@ -357,7 +369,15 @@ export const handleUsageCommand: CommandHandler = async (params, allowTextComman
if (targetSessionEntry && params.sessionStore && params.sessionKey) {
delete targetSessionEntry.responseUsage;
params.sessionStore[params.sessionKey] = targetSessionEntry;
await persistSessionEntry({ ...params, sessionEntry: targetSessionEntry });
if (
!(await persistSessionEntry({
...params,
sessionEntry: targetSessionEntry,
touchedFields: ["responseUsage"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
}
return {
shouldContinue: false,
@@ -377,7 +397,15 @@ export const handleUsageCommand: CommandHandler = async (params, allowTextComman
if (targetSessionEntry && params.sessionStore && params.sessionKey) {
targetSessionEntry.responseUsage = next;
params.sessionStore[params.sessionKey] = targetSessionEntry;
await persistSessionEntry({ ...params, sessionEntry: targetSessionEntry });
if (
!(await persistSessionEntry({
...params,
sessionEntry: targetSessionEntry,
touchedFields: ["responseUsage"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
}
return {
@@ -437,7 +465,15 @@ export const handleFastCommand: CommandHandler = async (params, allowTextCommand
if (resetsToDefault) {
if (targetSessionEntry && params.sessionStore && params.sessionKey) {
delete targetSessionEntry.fastMode;
await persistSessionEntry({ ...params, sessionEntry: targetSessionEntry });
if (
!(await persistSessionEntry({
...params,
sessionEntry: targetSessionEntry,
touchedFields: ["fastMode"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
}
return {
shouldContinue: false,
@@ -452,7 +488,15 @@ export const handleFastCommand: CommandHandler = async (params, allowTextCommand
if (targetSessionEntry && params.sessionStore && params.sessionKey) {
targetSessionEntry.fastMode = nextMode;
await persistSessionEntry({ ...params, sessionEntry: targetSessionEntry });
if (
!(await persistSessionEntry({
...params,
sessionEntry: targetSessionEntry,
touchedFields: ["fastMode"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
}
return {

26
src/auto-reply/reply/commands-tts.ts
View File

@@ -33,7 +33,10 @@ import {
} from "../../tts/tts.js";
import { isSilentReplyPayloadText } from "../tokens.js";
import type { ReplyPayload } from "../types.js";
import { persistSessionEntry } from "./commands-session-store.js";
import {
persistSessionEntry,
sessionEntryPersistenceConflictReply,
} from "./commands-session-store.js";
import type { CommandHandler } from "./commands-types.js";
type ParsedTtsCommand = {
@@ -228,17 +231,23 @@ export const handleTtsCommands: CommandHandler = async (params, allowTextCommand
}
if (requested === "on") {
params.sessionEntry.ttsAuto = "always";
await persistSessionEntry(params);
if (!(await persistSessionEntry({ ...params, touchedFields: ["ttsAuto"] }))) {
return sessionEntryPersistenceConflictReply();
}
return { shouldContinue: false, reply: { text: "🔊 TTS enabled for this chat." } };
}
if (requested === "off") {
params.sessionEntry.ttsAuto = "off";
await persistSessionEntry(params);
if (!(await persistSessionEntry({ ...params, touchedFields: ["ttsAuto"] }))) {
return sessionEntryPersistenceConflictReply();
}
return { shouldContinue: false, reply: { text: "🔇 TTS disabled for this chat." } };
}
if (requested === "default" || requested === "inherit" || requested === "clear") {
delete params.sessionEntry.ttsAuto;
await persistSessionEntry(params);
if (!(await persistSessionEntry({ ...params, touchedFields: ["ttsAuto"] }))) {
return sessionEntryPersistenceConflictReply();
}
return { shouldContinue: false, reply: { text: "🔊 TTS chat override cleared." } };
}
return { shouldContinue: false, reply: ttsUsage() };
@@ -289,7 +298,14 @@ export const handleTtsCommands: CommandHandler = async (params, allowTextCommand
params.sessionEntry.lastTtsReadLatestHash = hash;
params.sessionEntry.lastTtsReadLatestAt = Date.now();
await persistSessionEntry(params);
if (
!(await persistSessionEntry({
...params,
touchedFields: ["lastTtsReadLatestHash", "lastTtsReadLatestAt"],
}))
) {
return sessionEntryPersistenceConflictReply();
}
return { shouldContinue: false, reply: audio.reply };
}

6
src/auto-reply/reply/commands-types.ts
View File

@@ -1,9 +1,9 @@
import type { FastMode } from "@openclaw/normalization-core/string-coerce";
/** Shared command handler context and result contracts. */
import type { BlockReplyChunking } from "../../agents/embedded-agent-block-chunker.js";
import type { ChannelId } from "../../channels/plugins/types.public.js";
import type { SessionEntry, SessionScope } from "../../config/sessions.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import type { FastMode } from "@openclaw/normalization-core/string-coerce";
import type { SkillCommandSpec } from "../../skills/types.js";
import type { MsgContext } from "../templating.js";
import type { ElevatedLevel, ReasoningLevel, ThinkLevel, VerboseLevel } from "../thinking.js";
@@ -49,6 +49,10 @@ export type HandleCommandsParams = {
failures: Array<{ gate: string; key: string }>;
};
sessionEntry?: SessionEntry;
/** Snapshot captured before command handlers mutate the active entry. */
initialSessionEntry?: SessionEntry;
/** True only when the current command owns first creation of this session row. */
allowCreateSessionEntry?: boolean;
previousSessionEntry?: SessionEntry;
sessionStore?: Record<string, SessionEntry>;
sessionKey: string;

6
src/auto-reply/reply/get-reply-fast-path.ts
View File

@@ -7,9 +7,10 @@ import {
import { normalizeChatType } from "../../channels/chat-type.js";
import { normalizeAnyChannelId } from "../../channels/registry.js";
import { applyMergePatch } from "../../config/merge-patch.js";
import { resolveSessionTranscriptPath, resolveStorePath } from "../../config/sessions/paths.js";
import { resolveStorePath } from "../../config/sessions/paths.js";
import { loadSessionEntry, listSessionEntries } from "../../config/sessions/session-accessor.js";
import { resolveSessionKey } from "../../config/sessions/session-key.js";
import { formatSqliteSessionFileMarker } from "../../config/sessions/sqlite-marker.js";
import type { SessionEntry, SessionScope } from "../../config/sessions/types.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { resolveCommandTurnTargetSessionKey } from "../command-turn-context.js";
@@ -246,7 +247,7 @@ export function initFastReplySessionState(params: {
const sessionFile =
!resetTriggered && existingEntry?.sessionFile
? existingEntry.sessionFile
: resolveSessionTranscriptPath(sessionId, agentId);
: formatSqliteSessionFileMarker({ agentId, sessionId, storePath });
const sessionEntry: SessionEntry = {
...(!resetTriggered ? existingEntry : undefined),
sessionId,
@@ -299,6 +300,7 @@ export function initFastReplySessionState(params: {
return {
sessionCtx,
sessionEntry,
initialSessionEntry: existingEntry ? { ...existingEntry } : undefined,
sessionEntryHandle,
sessionStore,
sessionKey,

6
src/auto-reply/reply/get-reply-inline-actions.ts
View File

@@ -178,6 +178,8 @@ export async function handleInlineActions(params: {
agentId: string;
agentDir?: string;
sessionEntry?: SessionEntry;
initialSessionEntry?: SessionEntry;
allowCreateSessionEntry?: boolean;
previousSessionEntry?: SessionEntry;
sessionStore?: Record<string, SessionEntry>;
sessionKey: string;
@@ -220,6 +222,8 @@ export async function handleInlineActions(params: {
agentId,
agentDir,
sessionEntry,
initialSessionEntry,
allowCreateSessionEntry,
previousSessionEntry,
sessionStore,
sessionKey,
@@ -516,6 +520,8 @@ export async function handleInlineActions(params: {
failures: elevatedFailures,
},
sessionEntry: targetSessionEntry,
initialSessionEntry,
allowCreateSessionEntry,
previousSessionEntry,
sessionStore,
sessionKey,

33
src/auto-reply/reply/get-reply-native-slash-fast-path.ts
View File

@@ -28,6 +28,7 @@ import { resolveReplyDirectives } from "./get-reply-directives.js";
import { initFastReplySessionState } from "./get-reply-fast-path.js";
import { handleInlineActions } from "./get-reply-inline-actions.js";
import { stripStructuralPrefixes } from "./mentions.js";
import { persistReplySessionEntry } from "./session-entry-persistence.js";
import type { createTypingController } from "./typing.js";
type AgentDefaults = NonNullable<NonNullable<OpenClawConfig["agents"]>["defaults"]> | undefined;
@@ -137,6 +138,34 @@ export async function maybeResolveNativeSlashCommandFastReply(params: {
commandAuthorized: params.commandAuthorized,
workspaceDir: params.workspaceDir,
});
if (params.commandAuthorized) {
const creatingSession = sessionState.initialSessionEntry === undefined;
const initializationEntry = sessionState.initialSessionEntry ?? sessionState.sessionEntry;
const persistence = await persistReplySessionEntry({
storePath: sessionState.storePath,
sessionKey: sessionState.sessionKey,
allowCreate: creatingSession,
initialEntry: initializationEntry,
entry: sessionState.sessionEntry,
skipMaintenance: !creatingSession,
});
if (persistence.status === "lifecycle-invalidated") {
params.typing.cleanup();
return {
handled: true,
reply: markCommandReplyForDelivery({
text: persistence.error,
}),
};
}
const persistedInitialEntry = persistence.entry;
// Commit the synthesized activity/channel touch before commands or directives
// capture their own mutation baseline.
sessionState.sessionEntry = persistedInitialEntry;
sessionState.sessionEntryHandle.replaceCurrent(persistedInitialEntry);
sessionState.sessionStore[sessionState.sessionKey] = persistedInitialEntry;
sessionState.sessionId = persistedInitialEntry.sessionId;
}
const command = buildCommandContext({
ctx: params.ctx,
cfg: params.cfg,
@@ -291,6 +320,10 @@ export async function maybeResolveNativeSlashCommandFastReply(params: {
agentId: params.agentId,
agentDir: params.agentDir,
sessionEntry: sessionState.sessionEntry,
...(sessionState.initialSessionEntry
? { initialSessionEntry: sessionState.initialSessionEntry }
: {}),
allowCreateSessionEntry: sessionState.initialSessionEntry === undefined,
previousSessionEntry: sessionState.previousSessionEntry,
sessionStore: sessionState.sessionStore,
sessionKey: sessionState.sessionKey,

41
src/auto-reply/reply/get-reply.fast-path.test.ts
View File

@@ -7,6 +7,7 @@ import { testing as cliBackendsTesting } from "../../agents/cli-backends.js";
import type { OpenClawConfig } from "../../config/config.js";
import { getSessionEntry, type SessionEntry } from "../../config/sessions.js";
import { loadSessionEntry, replaceSessionEntry } from "../../config/sessions/session-accessor.js";
import { formatSqliteSessionFileMarker } from "../../config/sessions/sqlite-marker.js";
import { getReplyPayloadMetadata } from "../reply-payload.js";
import {
buildFastReplyCommandContext,
@@ -646,13 +647,14 @@ describe("getReplyFromConfig fast test bootstrap", () => {
});
it("uses native command target session keys during fast bootstrap", () => {
const storePath = "/tmp/sessions.json";
const result = initFastReplySessionState({
ctx: buildGetReplyCtx({
SessionKey: "telegram:slash:123",
CommandSource: "native",
CommandTargetSessionKey: "agent:main:main",
}),
cfg: { session: { store: "/tmp/sessions.json" } } as OpenClawConfig,
cfg: { session: { store: storePath } } as OpenClawConfig,
agentId: "main",
commandAuthorized: true,
workspaceDir: "/tmp/workspace",
@@ -660,6 +662,13 @@ describe("getReplyFromConfig fast test bootstrap", () => {
expect(result.sessionKey).toBe("agent:main:main");
expect(result.sessionCtx.SessionKey).toBe("agent:main:main");
expect(result.sessionEntry.sessionFile).toBe(
formatSqliteSessionFileMarker({
agentId: "main",
sessionId: result.sessionId,
storePath,
}),
);
});
it("preserves usage footer mode during fast reset bootstrap", async () => {
@@ -691,6 +700,36 @@ describe("getReplyFromConfig fast test bootstrap", () => {
expect(result.sessionEntry.responseUsage).toBe("full");
});
it("captures the initial SQLite session entry during fast bootstrap", async () => {
const home = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-fast-initial-entry-"));
const storePath = path.join(home, "sessions.json");
const sessionKey = "agent:main:telegram:123";
await seedFastPathSessionStore(storePath, {
[sessionKey]: {
sessionId: "existing-fast-initial",
updatedAt: Date.now(),
responseUsage: "tokens",
},
});
const result = initFastReplySessionState({
ctx: buildGetReplyCtx({
Body: "hello",
RawBody: "hello",
CommandBody: "hello",
SessionKey: sessionKey,
}),
cfg: { session: { store: storePath } } as OpenClawConfig,
agentId: "main",
commandAuthorized: true,
workspaceDir: home,
});
expect(result.initialSessionEntry?.sessionId).toBe("existing-fast-initial");
expect(result.initialSessionEntry?.responseUsage).toBe("tokens");
expect(result.initialSessionEntry).not.toBe(result.sessionEntry);
});
it("maps explicit gateway origin into command context", () => {
const command = buildFastReplyCommandContext({
ctx: buildGetReplyCtx({

3
src/auto-reply/reply/get-reply.ts
View File

@@ -495,6 +495,7 @@ export async function getReplyFromConfig(
const {
sessionCtx,
sessionEntry,
initialSessionEntry,
sessionEntryHandle,
previousSessionEntry,
sessionStore,
@@ -859,6 +860,8 @@ export async function getReplyFromConfig(
agentId,
agentDir,
sessionEntry,
...(initialSessionEntry ? { initialSessionEntry } : {}),
allowCreateSessionEntry: useFastTestBootstrap && initialSessionEntry === undefined,
previousSessionEntry,
sessionStore,
sessionKey,

88
src/auto-reply/reply/session-entry-persistence.ts Normal file
View File

@@ -0,0 +1,88 @@
// Atomic persistence for broad auto-reply session snapshots.
import type { SessionEntry } from "../../config/sessions.js";
import { resolveSessionWorkStartError } from "../../config/sessions/lifecycle.js";
import { patchSessionEntry } from "../../config/sessions/session-accessor.js";
import {
projectSessionSnapshotChanges,
sessionSnapshotTouchedFieldsConflict,
} from "../../config/sessions/session-snapshot-merge.js";
type PersistReplySessionEntryParams = {
allowCreate?: boolean;
entry: SessionEntry;
initialEntry: SessionEntry;
reassertLiveModelSwitchPending?: boolean;
sessionKey: string;
skipMaintenance?: boolean;
storePath: string;
touchedFields?: ReadonlyArray<keyof SessionEntry>;
};
export type PersistReplySessionEntryResult =
| { status: "current"; entry: SessionEntry }
| { status: "lifecycle-invalidated"; error: string; entry?: SessionEntry };
/** Persists reply-owned state without reverting concurrent session management. */
export async function persistReplySessionEntry(
params: PersistReplySessionEntryParams,
): Promise<PersistReplySessionEntryResult> {
let lifecycleError: string | undefined;
let lifecycleEntry: SessionEntry | undefined;
const persisted = await patchSessionEntry(
{ sessionKey: params.sessionKey, storePath: params.storePath },
(_entry, context) => {
if (!context.existingEntry) {
if (params.allowCreate !== true) {
lifecycleError = resolveSessionWorkStartError(params.sessionKey, undefined, {
expectedSessionId: params.initialEntry.sessionId,
});
return null;
}
return params.entry;
}
lifecycleError = resolveSessionWorkStartError(params.sessionKey, context.existingEntry, {
expectedSessionId: params.initialEntry.sessionId,
});
if (lifecycleError) {
lifecycleEntry = context.existingEntry;
return null;
}
if (
sessionSnapshotTouchedFieldsConflict({
initial: params.initialEntry,
next: params.entry,
current: context.existingEntry,
touchedFields: params.touchedFields,
})
) {
return null;
}
// Reply flows persist broad snapshots. Project only reply-owned changes
// so concurrent lifecycle, policy, and privacy updates remain authoritative.
return projectSessionSnapshotChanges({
initial: params.initialEntry,
next: params.entry,
current: context.existingEntry,
reassertLiveModelSwitchPending: params.reassertLiveModelSwitchPending,
});
},
{
fallbackEntry: params.entry,
skipMaintenance: params.skipMaintenance,
},
);
if (lifecycleError) {
return {
status: "lifecycle-invalidated",
error: lifecycleError,
...(lifecycleEntry ? { entry: lifecycleEntry } : {}),
};
}
if (!persisted) {
return {
status: "lifecycle-invalidated",
error: `Session "${params.sessionKey}" changed while starting work. Retry.`,
};
}
return { status: "current", entry: persisted };
}

42
src/auto-reply/reply/session-updates.test.ts
View File

@@ -199,4 +199,46 @@ describe("ensureSkillSnapshot", () => {
expect(result.systemSent).toBe(false);
expect(sessionStore[sessionKey]).toEqual(reboundEntry);
});
it("persists first-turn skill snapshots as a guarded partial update", async () => {
vi.stubEnv("OPENCLAW_TEST_FAST", "0");
const sessionKey = "agent:main:main";
const sessionEntry = {
sessionId: "session-1",
updatedAt: 10,
modelOverride: "gpt-5.5",
};
const sessionStore = { [sessionKey]: sessionEntry };
updateSessionEntryMock.mockImplementationOnce(async (_scope, update) => {
const patch = await update({
...sessionEntry,
updatedAt: 20,
modelOverride: "sonnet-4.6",
});
expect(patch).toMatchObject({
sessionId: "session-1",
systemSent: true,
});
expect(patch).not.toHaveProperty("modelOverride");
return {
...sessionEntry,
...patch,
modelOverride: "sonnet-4.6",
};
});
const result = await ensureSkillSnapshot({
sessionEntry,
sessionStore,
sessionKey,
sessionId: "session-1",
storePath: "/tmp/sessions.json",
isFirstTurnInSession: true,
workspaceDir: TEST_WORKSPACE_DIR,
cfg: {},
});
expect(result.sessionEntry?.modelOverride).toBe("sonnet-4.6");
expect(sessionStore[sessionKey]?.modelOverride).toBe("sonnet-4.6");
});
});

14
src/auto-reply/reply/session-updates.ts
View File

@@ -28,6 +28,7 @@ async function persistSessionEntryUpdate(params: {
sessionKey?: string;
storePath?: string;
nextEntry: SessionEntry;
updates: Partial<SessionEntry>;
}): Promise<SessionEntry | undefined> {
if (!params.sessionEntryHandle && (!params.sessionStore || !params.sessionKey)) {
return undefined;
@@ -48,7 +49,7 @@ async function persistSessionEntryUpdate(params: {
storePath: params.storePath,
sessionKey: params.sessionKey,
},
(entry) => (entry.sessionId === params.expectedSessionId ? params.nextEntry : null),
(entry) => (entry.sessionId === params.expectedSessionId ? params.updates : null),
);
if (persistedEntry) {
if (params.sessionEntryHandle) {
@@ -220,6 +221,12 @@ export async function ensureSkillSnapshot(params: {
sessionKey,
storePath,
nextEntry,
updates: {
sessionId: nextEntry.sessionId,
updatedAt: nextEntry.updatedAt,
systemSent: nextEntry.systemSent,
skillsSnapshot: nextEntry.skillsSnapshot,
},
});
nextEntry = persistedEntry;
systemSent = persistedEntry?.systemSent ?? systemSent;
@@ -258,6 +265,11 @@ export async function ensureSkillSnapshot(params: {
sessionKey,
storePath,
nextEntry,
updates: {
sessionId: nextEntry.sessionId,
updatedAt: nextEntry.updatedAt,
skillsSnapshot: nextEntry.skillsSnapshot,
},
});
}

1
src/auto-reply/reply/session.ts
View File

@@ -166,6 +166,7 @@ function hasProviderOwnedSession(entry: SessionEntry | undefined): boolean {
export type SessionInitResult = {
sessionCtx: TemplateContext;
sessionEntry: SessionEntry;
initialSessionEntry?: SessionEntry;
previousSessionEntry?: SessionEntry;
sessionEntryHandle: ReplySessionEntryHandle;
sessionStore: Record<string, SessionEntry>;

46
src/config/sessions/lifecycle.ts
View File

@@ -15,6 +15,52 @@ type SessionLifecycleEntry = Pick<
"sessionId" | "sessionFile" | "sessionStartedAt" | "lastInteractionAt" | "updatedAt"
>;
type SessionWorkStartEntry = Pick<SessionEntry, "sessionId">;
type SessionWorkStartOptions = {
expectedSessionId?: string;
};
/** Stable Gateway error detail for stale session lifecycle requests. */
export const SESSION_LIFECYCLE_CHANGED_ERROR_REASON = "session-changed";
export const SESSION_WORK_START_INVALIDATED_ERROR_CODE = "SESSION_WORK_START_INVALIDATED";
export class SessionWorkStartInvalidatedError extends Error {
readonly code = SESSION_WORK_START_INVALIDATED_ERROR_CODE;
constructor(message: string) {
super(message);
this.name = "SessionWorkStartInvalidatedError";
}
}
export function isSessionWorkStartInvalidatedError(
error: unknown,
): error is SessionWorkStartInvalidatedError {
return (
error instanceof SessionWorkStartInvalidatedError ||
(typeof error === "object" &&
error !== null &&
"code" in error &&
error.code === SESSION_WORK_START_INVALIDATED_ERROR_CODE)
);
}
/** Stale session starts are rejected after another lifecycle owner moves the row. */
export function resolveSessionWorkStartError(
sessionKey: string,
entry: SessionWorkStartEntry | null | undefined,
options?: SessionWorkStartOptions,
): string | undefined {
if (options?.expectedSessionId && !entry) {
return `Session "${sessionKey}" was deleted while starting work. Retry.`;
}
if (options?.expectedSessionId && entry?.sessionId !== options.expectedSessionId) {
return `Session "${sessionKey}" changed while starting work. Retry.`;
}
return undefined;
}
// Transcript headers are read lazily to recover startedAt without parsing full files.
type TerminalMainSessionTranscriptRegistryParams = {

290
src/config/sessions/session-snapshot-merge.ts Normal file
View File

@@ -0,0 +1,290 @@
import { isDeepStrictEqual } from "node:util";
import type { SessionEntry } from "./types.js";
type SessionEntryRecord = Partial<Record<keyof SessionEntry, unknown>>;
export const SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS = [
"providerOverride",
"modelOverride",
"agentRuntimeOverride",
"modelOverrideSource",
"modelOverrideFallbackOriginProvider",
"modelOverrideFallbackOriginModel",
"authProfileOverride",
"authProfileOverrideSource",
"authProfileOverrideCompactionCount",
] as const satisfies ReadonlyArray<keyof SessionEntry>;
const MODEL_ROUTE_OVERRIDE_FIELDS = [
"providerOverride",
"modelOverride",
"agentRuntimeOverride",
] as const satisfies ReadonlyArray<keyof SessionEntry>;
const MODEL_OVERRIDE_RUNTIME_FIELDS = [
"modelProvider",
"model",
"fallbackNoticeSelectedModel",
"fallbackNoticeActiveModel",
"fallbackNoticeReason",
"contextTokens",
"contextBudgetStatus",
] as const satisfies ReadonlyArray<keyof SessionEntry>;
const MODEL_OVERRIDE_RUNTIME_FIELD_SET = new Set<keyof SessionEntry>(MODEL_OVERRIDE_RUNTIME_FIELDS);
const MODEL_OVERRIDE_DEPENDENT_FIELDS = new Set<keyof SessionEntry>([
...MODEL_OVERRIDE_RUNTIME_FIELDS,
"liveModelSwitchPending",
"thinkingLevel",
]);
const MODEL_OVERRIDE_CONFLICT_DEPENDENT_FIELDS = ["thinkingLevel"] as const satisfies ReadonlyArray<
keyof SessionEntry
>;
function anySessionFieldChanged(
before: SessionEntryRecord,
after: SessionEntryRecord,
fields: ReadonlyArray<keyof SessionEntry>,
): boolean {
return fields.some((field) => !isDeepStrictEqual(before[field], after[field]));
}
/** Projects run-local snapshot changes without restoring concurrently changed fields. */
export function projectSessionSnapshotChanges(params: {
initial: SessionEntry;
next: SessionEntry;
current: SessionEntry;
reassertLiveModelSwitchPending?: boolean;
}): Partial<SessionEntry> {
if (params.current.sessionId !== params.initial.sessionId) {
return {};
}
const initial = params.initial as SessionEntryRecord;
const next = params.next as SessionEntryRecord;
const current = params.current as SessionEntryRecord;
const patch: Partial<SessionEntry> = {};
const patchRecord = patch as SessionEntryRecord;
const fields = new Set<keyof SessionEntry>([
...(Object.keys(params.initial) as Array<keyof SessionEntry>),
...(Object.keys(params.next) as Array<keyof SessionEntry>),
]);
const modelOverrideChanged = anySessionFieldChanged(
initial,
next,
SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS,
);
const modelRouteOverrideChanged = anySessionFieldChanged(
initial,
next,
MODEL_ROUTE_OVERRIDE_FIELDS,
);
const modelOverrideChangedConcurrently = anySessionFieldChanged(
initial,
current,
SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS,
);
if (modelOverrideChanged && !modelOverrideChangedConcurrently) {
// Model, provenance, and auth overrides form one selection transaction.
// Project the whole family or none of it so concurrent switches cannot hybridize rows.
for (const field of SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS) {
patchRecord[field] = Object.hasOwn(params.next, field) ? next[field] : undefined;
}
if (modelRouteOverrideChanged) {
// A winning route switch invalidates any runtime facts written for the
// old model, including facts that appeared after this snapshot began.
for (const field of MODEL_OVERRIDE_RUNTIME_FIELDS) {
patchRecord[field] = Object.hasOwn(params.next, field) ? next[field] : undefined;
}
}
for (const field of MODEL_OVERRIDE_DEPENDENT_FIELDS) {
if (modelRouteOverrideChanged && MODEL_OVERRIDE_RUNTIME_FIELD_SET.has(field)) {
continue;
}
if (!isDeepStrictEqual(initial[field], next[field])) {
patchRecord[field] = Object.hasOwn(params.next, field) ? next[field] : undefined;
}
}
if (params.reassertLiveModelSwitchPending) {
// A second explicit switch can inherit true from the first snapshot while
// the active runner concurrently clears it. Re-arm the winning transaction.
patch.liveModelSwitchPending = true;
}
}
const runtimeModelChanged =
!isDeepStrictEqual(initial.model, next.model) ||
!isDeepStrictEqual(initial.modelProvider, next.modelProvider);
const runtimeModelChangedConcurrently =
!isDeepStrictEqual(current.model, initial.model) ||
!isDeepStrictEqual(current.modelProvider, initial.modelProvider);
if (
runtimeModelChanged &&
!runtimeModelChangedConcurrently &&
!modelOverrideChanged &&
!modelOverrideChangedConcurrently
) {
// Runtime model metadata is one pair. A model-only patch intentionally
// clears provider state, so snapshot projection must emit both fields.
patch.model = params.next.model;
patch.modelProvider = params.next.modelProvider;
}
for (const field of fields) {
if (
field === "model" ||
field === "modelProvider" ||
SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS.includes(
field as (typeof SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS)[number],
)
) {
continue;
}
if (isDeepStrictEqual(initial[field], next[field])) {
continue;
}
if (field === "updatedAt") {
patch.updatedAt = Math.max(params.current.updatedAt, params.next.updatedAt);
continue;
}
if (
(modelOverrideChanged || modelOverrideChangedConcurrently) &&
MODEL_OVERRIDE_DEPENDENT_FIELDS.has(field)
) {
continue;
}
// The latest row wins conflicts. This prevents a stale run snapshot from
// reviving policy or privacy fields cleared while the run was active.
if (!isDeepStrictEqual(current[field], initial[field])) {
continue;
}
patchRecord[field] = Object.hasOwn(params.next, field) ? next[field] : undefined;
}
return patch;
}
/** Reports whether every caller-owned snapshot delta is present in the current row. */
export function sessionSnapshotChangesApplied(params: {
initial: SessionEntry;
next: SessionEntry;
current: SessionEntry;
touchedFields?: ReadonlyArray<keyof SessionEntry>;
}): boolean {
if (params.current.sessionId !== params.initial.sessionId) {
return false;
}
const initial = params.initial as SessionEntryRecord;
const next = params.next as SessionEntryRecord;
const current = params.current as SessionEntryRecord;
const fields = new Set<keyof SessionEntry>([
...(Object.keys(params.initial) as Array<keyof SessionEntry>),
...(Object.keys(params.next) as Array<keyof SessionEntry>),
...(params.touchedFields ?? []),
]);
fields.delete("updatedAt");
for (const field of fields) {
const explicitlyTouched = params.touchedFields?.includes(field) === true;
if (
(explicitlyTouched || !isDeepStrictEqual(initial[field], next[field])) &&
!isDeepStrictEqual(current[field], next[field])
) {
return false;
}
}
return true;
}
/** Reports an explicit grouped-write conflict before any snapshot fields are committed. */
export function sessionSnapshotTouchedFieldsConflict(params: {
initial: SessionEntry;
next: SessionEntry;
current: SessionEntry;
touchedFields?: ReadonlyArray<keyof SessionEntry>;
}): boolean {
if (params.current.sessionId !== params.initial.sessionId) {
return true;
}
const initial = params.initial as SessionEntryRecord;
const next = params.next as SessionEntryRecord;
const current = params.current as SessionEntryRecord;
const fields = new Set(params.touchedFields ?? []);
if (SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS.some((field) => fields.has(field))) {
for (const field of MODEL_OVERRIDE_CONFLICT_DEPENDENT_FIELDS) {
if (!isDeepStrictEqual(initial[field], next[field])) {
fields.add(field);
}
}
}
return [...fields].some(
(field) =>
!isDeepStrictEqual(current[field], initial[field]) &&
!isDeepStrictEqual(current[field], next[field]),
);
}
/** Replaces a caller-held snapshot with the latest persisted row in place. */
export function adoptPersistedSessionSnapshot(target: SessionEntry, current: SessionEntry): void {
const targetRecord = target as SessionEntryRecord;
const currentRecord = current as SessionEntryRecord;
for (const field of Object.keys(target) as Array<keyof SessionEntry>) {
if (!Object.hasOwn(current, field)) {
delete targetRecord[field];
}
}
for (const field of Object.keys(current) as Array<keyof SessionEntry>) {
targetRecord[field] = currentRecord[field];
}
}
/** Reports whether a model/auth selection transaction is the persisted winner. */
export function sessionModelOverrideChangesApplied(params: {
initial: SessionEntry;
next: SessionEntry;
current: SessionEntry;
reassertLiveModelSwitchPending?: boolean;
}): boolean {
if (params.current.sessionId !== params.initial.sessionId) {
return false;
}
const next = params.next as SessionEntryRecord;
const current = params.current as SessionEntryRecord;
const initial = params.initial as SessionEntryRecord;
const changedDependentFields = [...MODEL_OVERRIDE_DEPENDENT_FIELDS].filter(
(field) => !isDeepStrictEqual(initial[field], next[field]),
);
const modelRouteOverrideChanged = anySessionFieldChanged(
initial,
next,
MODEL_ROUTE_OVERRIDE_FIELDS,
);
const fields = [
...SESSION_MODEL_OVERRIDE_TRANSACTION_FIELDS,
...(modelRouteOverrideChanged ? MODEL_OVERRIDE_RUNTIME_FIELDS : []),
...changedDependentFields,
];
if (fields.some((field) => !isDeepStrictEqual(current[field], next[field]))) {
return false;
}
return !params.reassertLiveModelSwitchPending || params.current.liveModelSwitchPending === true;
}
/** Merges run-local snapshot changes into the latest persisted session row. */
export function mergeSessionSnapshotChanges(params: {
initial: SessionEntry;
next: SessionEntry;
current: SessionEntry;
reassertLiveModelSwitchPending?: boolean;
}): SessionEntry {
const merged = { ...params.current };
const mergedRecord = merged as SessionEntryRecord;
const patch = projectSessionSnapshotChanges(params) as SessionEntryRecord;
for (const field of Object.keys(patch) as Array<keyof SessionEntry>) {
if (patch[field] === undefined) {
delete mergedRecord[field];
} else {
mergedRecord[field] = patch[field];
}
}
return merged;
}