11// Session store facade coordinates reads, writes, maintenance, delivery metadata, and exports.
22import fs from "node:fs";
33import path from "node:path";
4+import {
5+normalizeLowercaseStringOrEmpty,
6+normalizeOptionalString,
7+} from "@openclaw/normalization-core/string-coerce";
48import type { MsgContext } from "../../auto-reply/templating.js";
59import { writeTextAtomic } from "../../infra/json-files.js";
610import { createSubsystemLogger } from "../../logging/subsystem.js";
@@ -86,6 +90,25 @@ export type {
8690} from "./store-cache.js";
8791export { normalizeStoreSessionKey, resolveSessionStoreEntry } from "./store-entry.js";
889293+export type SessionEntryPatchProjectionSnapshot = {
94+entries: ReadonlyArray<{ sessionKey: string; entry: SessionEntry }>;
95+};
96+97+export type SessionEntryPatchProjectionTarget = {
98+candidateKeys?: readonly string[];
99+primaryKey: string;
100+};
101+102+export type SessionEntryPatchProjectionContext = SessionEntryPatchProjectionSnapshot &
103+SessionEntryPatchProjectionTarget & {
104+existingEntry?: SessionEntry;
105+};
106+107+export type SessionEntryPatchProjectionFailure = { ok: false };
108+109+export type SessionEntryPatchProjectionResult<TFailure extends SessionEntryPatchProjectionFailure> =
110+{ ok: true; entry: SessionEntry } | TFailure;
111+89112const log = createSubsystemLogger("sessions/store");
90113let sessionArchiveRuntimePromise: Promise<
91114typeof import("../../gateway/session-archive.runtime.js")
@@ -926,6 +949,129 @@ export async function updateSessionStore<T>(
926949});
927950}
928951952+function cloneSessionEntryProjectionSnapshot(
953+store: Record<string, SessionEntry>,
954+): SessionEntryPatchProjectionSnapshot {
955+return {
956+entries: Object.entries(store).map(([sessionKey, entry]) => ({
957+ sessionKey,
958+entry: cloneSessionEntry(entry),
959+})),
960+};
961+}
962+963+function resolveFreshestProjectedEntry(params: {
964+store: Record<string, SessionEntry>;
965+candidateKeys: readonly string[];
966+}): SessionEntry | undefined {
967+let freshest: SessionEntry | undefined;
968+const keys = new Set<string>();
969+for (const candidateKey of params.candidateKeys) {
970+const trimmed = normalizeOptionalString(candidateKey) ?? "";
971+if (!trimmed) {
972+continue;
973+}
974+keys.add(trimmed);
975+for (const match of findSessionStoreKeysIgnoreCase(params.store, trimmed)) {
976+keys.add(match);
977+}
978+}
979+for (const key of keys) {
980+const entry = params.store[key];
981+if (!entry) {
982+continue;
983+}
984+if (!freshest || (entry.updatedAt ?? 0) > (freshest.updatedAt ?? 0)) {
985+freshest = entry;
986+}
987+}
988+return freshest;
989+}
990+991+function findSessionStoreKeysIgnoreCase(
992+store: Record<string, SessionEntry>,
993+targetKey: string,
994+): string[] {
995+const lowered = normalizeLowercaseStringOrEmpty(targetKey);
996+return Object.keys(store).filter((key) => normalizeLowercaseStringOrEmpty(key) === lowered);
997+}
998+999+function migrateSessionEntryProjectionTarget(params: {
1000+store: Record<string, SessionEntry>;
1001+target: SessionEntryPatchProjectionTarget;
1002+}): void {
1003+const candidateKeys = params.target.candidateKeys ?? [params.target.primaryKey];
1004+const freshest = resolveFreshestProjectedEntry({
1005+store: params.store,
1006+ candidateKeys,
1007+});
1008+const currentPrimary = params.store[params.target.primaryKey];
1009+if (
1010+freshest &&
1011+(!currentPrimary || (freshest.updatedAt ?? 0) > (currentPrimary.updatedAt ?? 0))
1012+) {
1013+params.store[params.target.primaryKey] = freshest;
1014+}
1015+1016+const keysToDelete = new Set<string>();
1017+for (const candidateKey of candidateKeys) {
1018+const trimmed = normalizeOptionalString(candidateKey) ?? "";
1019+if (!trimmed) {
1020+continue;
1021+}
1022+if (trimmed !== params.target.primaryKey) {
1023+keysToDelete.add(trimmed);
1024+}
1025+for (const match of findSessionStoreKeysIgnoreCase(params.store, trimmed)) {
1026+if (match !== params.target.primaryKey) {
1027+keysToDelete.add(match);
1028+}
1029+}
1030+}
1031+for (const key of keysToDelete) {
1032+delete params.store[key];
1033+}
1034+}
1035+1036+/**
1037+ * Applies a storage-neutral entry projection inside the session-store writer.
1038+ * The projection receives a cloned snapshot and returns the replacement entry;
1039+ * it cannot mutate the backing whole store.
1040+ */
1041+export async function applySessionEntryPatchProjection<
1042+TFailure extends SessionEntryPatchProjectionFailure,
1043+>(params: {
1044+storePath: string;
1045+resolveTarget: (
1046+snapshot: SessionEntryPatchProjectionSnapshot,
1047+) => SessionEntryPatchProjectionTarget;
1048+project: (
1049+context: SessionEntryPatchProjectionContext,
1050+) =>
1051+| Promise<SessionEntryPatchProjectionResult<TFailure>>
1052+| SessionEntryPatchProjectionResult<TFailure>;
1053+}): Promise<SessionEntryPatchProjectionResult<TFailure>> {
1054+return await runExclusiveSessionStoreWrite(params.storePath, async () => {
1055+const store = loadMutableSessionStoreForWriter(params.storePath);
1056+const target = params.resolveTarget(cloneSessionEntryProjectionSnapshot(store));
1057+migrateSessionEntryProjectionTarget({ store, target });
1058+const projected = await params.project({
1059+ ...target,
1060+entries: cloneSessionEntryProjectionSnapshot(store).entries,
1061+ ...(store[target.primaryKey]
1062+ ? { existingEntry: cloneSessionEntry(store[target.primaryKey]) }
1063+ : {}),
1064+});
1065+if (projected.ok) {
1066+store[target.primaryKey] = cloneSessionEntry(projected.entry);
1067+}
1068+await saveSessionStoreUnlocked(params.storePath, store, {
1069+activeSessionKey: target.primaryKey,
1070+});
1071+return projected.ok ? { ...projected, entry: cloneSessionEntry(projected.entry) } : projected;
1072+});
1073+}
1074+9291075async function archiveUnreferencedLifecycleTranscriptArtifacts(params: {
9301076storePath: string;
9311077transcriptContentMarker: string;
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。