Skip to content

Commit c85bd45

Browse files
jalehmanjalehman
authored
clawdbot-d02.1.9.1.16: add session patch projection seam (#93739)
1 parent 402c85b commit c85bd45

9 files changed

Lines changed: 395 additions & 50 deletions

File tree

‎scripts/check-session-accessor-boundary.mjs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ export const migratedSessionAccessorFiles = new Set([
6868
"src/gateway/sessions-resolve.ts",
6969
"src/gateway/server-methods/sessions.ts",
7070
"src/infra/outbound/message-action-tts.ts",
71+
"src/tui/embedded-backend.ts",
7172
]);
7273

7374
export const migratedBundledPluginSessionAccessorFiles = new Set([
@@ -98,6 +99,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
9899
"src/auto-reply/reply/session-reset-model.ts",
99100
"src/auto-reply/reply/session-updates.ts",
100101
"src/auto-reply/reply/session-usage.ts",
102+
"src/tui/embedded-backend.ts",
101103
]);
102104

103105
export const migratedTranscriptWriterFiles = new Set([
@@ -290,8 +292,13 @@ export async function main() {
290292
"src/cron",
291293
"src/gateway",
292294
"src/infra",
295+
"src/tui",
296+
]);
297+
const writeSourceRoots = resolveSourceRoots(repoRoot, [
298+
"src/agents",
299+
"src/auto-reply",
300+
"src/tui",
293301
]);
294-
const writeSourceRoots = resolveSourceRoots(repoRoot, ["src/agents", "src/auto-reply"]);
295302
const transcriptWriterSourceRoots = resolveSourceRoots(repoRoot, [
296303
"src/agents/command",
297304
"src/agents/embedded-agent-runner",

‎src/config/sessions/session-accessor.test.ts‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js";
66
import {
77
appendTranscriptMessage,
88
appendTranscriptEvent,
9+
applySessionPatchProjection,
910
cleanupSessionLifecycleArtifacts,
1011
createSessionEntryWithTranscript,
1112
listSessionEntries,
@@ -323,6 +324,93 @@ describe("session accessor file-backed seam", () => {
323324
});
324325
});
325326

327+
it("applies projected session patches after migrating legacy candidate keys", async () => {
328+
fs.writeFileSync(
329+
storePath,
330+
JSON.stringify({
331+
"agent:main:main": {
332+
sessionId: "canonical-session",
333+
updatedAt: 10,
334+
},
335+
"AGENT:MAIN:MAIN": {
336+
sessionId: "legacy-session",
337+
updatedAt: 20,
338+
},
339+
}),
340+
"utf8",
341+
);
342+
343+
const projected = await applySessionPatchProjection({
344+
storePath,
345+
resolveTarget: () => ({
346+
primaryKey: "agent:main:main",
347+
candidateKeys: ["agent:main:main"],
348+
}),
349+
project: ({ entries, existingEntry, primaryKey }) => {
350+
expect(primaryKey).toBe("agent:main:main");
351+
expect(existingEntry?.sessionId).toBe("legacy-session");
352+
expect(entries.map((entry) => entry.sessionKey)).toEqual(["agent:main:main"]);
353+
return {
354+
ok: true as const,
355+
entry: {
356+
...existingEntry,
357+
label: "Projected",
358+
} as SessionEntry,
359+
};
360+
},
361+
});
362+
363+
expect(projected).toMatchObject({
364+
ok: true,
365+
entry: {
366+
label: "Projected",
367+
sessionId: "legacy-session",
368+
},
369+
});
370+
expect(loadSessionStore(storePath)).toEqual({
371+
"agent:main:main": expect.objectContaining({
372+
label: "Projected",
373+
sessionId: "legacy-session",
374+
}),
375+
});
376+
});
377+
378+
it("persists legacy key pruning when projected session patches fail validation", async () => {
379+
fs.writeFileSync(
380+
storePath,
381+
JSON.stringify({
382+
"agent:main:main": {
383+
sessionId: "canonical-session",
384+
updatedAt: 10,
385+
},
386+
"AGENT:MAIN:MAIN": {
387+
sessionId: "legacy-session",
388+
updatedAt: 20,
389+
},
390+
}),
391+
"utf8",
392+
);
393+
394+
const projected = await applySessionPatchProjection({
395+
storePath,
396+
resolveTarget: () => ({
397+
primaryKey: "agent:main:main",
398+
candidateKeys: ["agent:main:main"],
399+
}),
400+
project: () => ({
401+
ok: false as const,
402+
error: "invalid patch",
403+
}),
404+
});
405+
406+
expect(projected).toEqual({ ok: false, error: "invalid patch" });
407+
expect(loadSessionStore(storePath)).toEqual({
408+
"agent:main:main": expect.objectContaining({
409+
sessionId: "legacy-session",
410+
}),
411+
});
412+
});
413+
326414
it("cleans scoped lifecycle entries and unreferenced transcript artifacts", async () => {
327415
const nowMs = Date.now();
328416
const oldDate = new Date(nowMs - 600_000);

‎src/config/sessions/session-accessor.ts‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,17 @@ import {
2525
cleanupSessionLifecycleArtifacts as cleanupFileSessionLifecycleArtifacts,
2626
listSessionEntries as listFileSessionEntries,
2727
loadSessionStore,
28+
applySessionEntryPatchProjection as applyFileSessionEntryPatchProjection,
2829
patchSessionEntry as patchFileSessionEntry,
2930
readSessionUpdatedAt as readFileSessionUpdatedAt,
3031
resolveSessionStoreEntry,
3132
updateSessionStore,
3233
updateSessionStoreEntry as updateFileSessionStoreEntry,
34+
type SessionEntryPatchProjectionContext,
35+
type SessionEntryPatchProjectionFailure,
36+
type SessionEntryPatchProjectionResult,
37+
type SessionEntryPatchProjectionSnapshot,
38+
type SessionEntryPatchProjectionTarget,
3339
type SessionLifecycleArtifactCleanupParams,
3440
type SessionLifecycleArtifactCleanupResult,
3541
} from "./store.js";
@@ -267,6 +273,13 @@ type CreatedSessionTranscriptResult =
267273
| { ok: true; sessionFile: string }
268274
| { ok: false; error: string; phase: "transcript" };
269275

276+
export type SessionPatchProjectionContext = SessionEntryPatchProjectionContext;
277+
export type SessionPatchProjectionFailure = SessionEntryPatchProjectionFailure;
278+
export type SessionPatchProjectionResult<TFailure extends SessionPatchProjectionFailure> =
279+
SessionEntryPatchProjectionResult<TFailure>;
280+
export type SessionPatchProjectionSnapshot = SessionEntryPatchProjectionSnapshot;
281+
export type SessionPatchProjectionTarget = SessionEntryPatchProjectionTarget;
282+
270283
export type { SessionLifecycleArtifactCleanupParams, SessionLifecycleArtifactCleanupResult };
271284

272285
/** Returns the entry for a canonical or alias session key, if one exists. */
@@ -484,6 +497,23 @@ export async function updateSessionEntry(
484497
});
485498
}
486499

500+
/**
501+
* Applies a session patch projection through the accessor boundary.
502+
* The resolver sees a read-only snapshot and names the persisted key set; the
503+
* projector returns one replacement entry without receiving the mutable store.
504+
*/
505+
export async function applySessionPatchProjection<
506+
TFailure extends SessionPatchProjectionFailure,
507+
>(params: {
508+
storePath: string;
509+
resolveTarget: (snapshot: SessionPatchProjectionSnapshot) => SessionPatchProjectionTarget;
510+
project: (
511+
context: SessionPatchProjectionContext,
512+
) => Promise<SessionPatchProjectionResult<TFailure>> | SessionPatchProjectionResult<TFailure>;
513+
}): Promise<SessionPatchProjectionResult<TFailure>> {
514+
return await applyFileSessionEntryPatchProjection(params);
515+
}
516+
487517
/** Removes entries and orphan transcript artifacts owned by a named session lifecycle. */
488518
export async function cleanupSessionLifecycleArtifacts(
489519
params: SessionLifecycleArtifactCleanupParams,

‎src/config/sessions/store.ts‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
// Session store facade coordinates reads, writes, maintenance, delivery metadata, and exports.
22
import fs from "node:fs";
33
import path from "node:path";
4+
import {
5+
normalizeLowercaseStringOrEmpty,
6+
normalizeOptionalString,
7+
} from "@openclaw/normalization-core/string-coerce";
48
import type { MsgContext } from "../../auto-reply/templating.js";
59
import { writeTextAtomic } from "../../infra/json-files.js";
610
import { createSubsystemLogger } from "../../logging/subsystem.js";
@@ -86,6 +90,25 @@ export type {
8690
} from "./store-cache.js";
8791
export { normalizeStoreSessionKey, resolveSessionStoreEntry } from "./store-entry.js";
8892

93+
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+
89112
const log = createSubsystemLogger("sessions/store");
90113
let sessionArchiveRuntimePromise: Promise<
91114
typeof import("../../gateway/session-archive.runtime.js")
@@ -926,6 +949,129 @@ export async function updateSessionStore<T>(
926949
});
927950
}
928951

952+
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+
9291075
async function archiveUnreferencedLifecycleTranscriptArtifacts(params: {
9301076
storePath: string;
9311077
transcriptContentMarker: string;

0 commit comments

Comments
 (0)