Skip to content

Commit 4752524

Browse files
jalehmanjalehman
authored
refactor: add transcript update identity contract (#89912)
1 parent d38fb74 commit 4752524

21 files changed

Lines changed: 672 additions & 53 deletions

‎extensions/codex/src/app-server/transcript-mirror.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -360,6 +360,15 @@ export async function mirrorCodexAppServerTranscript(params: {
360360
sessionFile: params.sessionFile,
361361
...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
362362
...(params.agentId ? { agentId: params.agentId } : {}),
363+
...(params.sessionId && params.sessionKey && params.agentId
364+
? {
365+
target: {
366+
agentId: params.agentId,
367+
sessionId: params.sessionId,
368+
sessionKey: params.sessionKey,
369+
},
370+
}
371+
: {}),
363372
message: update.message,
364373
messageId: update.messageId,
365374
messageSeq: update.messageSeq,

‎extensions/copilot/src/attempt.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1018,6 +1018,7 @@ export async function runCopilotAttempt(
10181018
await dualWriteCopilotTranscriptBestEffort({
10191019
sessionFile: sessionFileForMirror,
10201020
sessionKey: readString((input as { sessionKey?: unknown }).sessionKey),
1021+
sessionId: readString(input.sessionId),
10211022
agentId: readString(input.agentId),
10221023
messages: taggedMessages,
10231024
idempotencyScope: sessionIdForScope ? `copilot:${sessionIdForScope}` : undefined,

‎extensions/copilot/src/dual-write-transcripts.ts‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@ function buildMirrorDedupeIdentity(message: MirroredAgentMessage): string {
9696
export interface MirrorCopilotTranscriptParams {
9797
sessionFile: string;
9898
sessionKey?: string;
99+
sessionId?: string;
99100
agentId?: string;
100101
messages: AgentMessage[];
101102
/**
@@ -168,7 +169,20 @@ export async function mirrorCopilotTranscript(
168169
}
169170

170171
if (params.sessionKey) {
171-
emitSessionTranscriptUpdate({ sessionFile: params.sessionFile, sessionKey: params.sessionKey });
172+
emitSessionTranscriptUpdate({
173+
sessionFile: params.sessionFile,
174+
sessionKey: params.sessionKey,
175+
...(params.agentId ? { agentId: params.agentId } : {}),
176+
...(params.sessionId && params.agentId
177+
? {
178+
target: {
179+
agentId: params.agentId,
180+
sessionId: params.sessionId,
181+
sessionKey: params.sessionKey,
182+
},
183+
}
184+
: {}),
185+
});
172186
} else {
173187
emitSessionTranscriptUpdate(params.sessionFile);
174188
}

‎extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import fs from "node:fs/promises";
33
import os from "node:os";
44
import path from "node:path";
55
import type { DatabaseSync } from "node:sqlite";
6+
import { emitSessionTranscriptUpdate } from "openclaw/plugin-sdk/agent-harness-runtime";
67
import {
78
resolveSessionTranscriptsDirForAgent,
89
type OpenClawConfig,
@@ -33,6 +34,25 @@ type SyncParams = {
3334
progress?: (update: MemorySyncProgressUpdate) => void;
3435
};
3536

37+
type MemorySessionTranscriptUpdate = {
38+
agentId?: string;
39+
sessionFile?: string;
40+
sessionKey?: string;
41+
target?: {
42+
agentId: string;
43+
sessionId: string;
44+
sessionKey: string;
45+
};
46+
};
47+
48+
type MemoryTranscriptUpdateSubscriber = (
49+
listener: (update: MemorySessionTranscriptUpdate) => void,
50+
) => () => void;
51+
52+
const MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY = Symbol.for(
53+
"openclaw.memoryCore.sessionTranscriptUpdateSubscriber",
54+
);
55+
3656
type SourceStateRow = { path: string; hash: string; mtime: number; size: number };
3757

3858
class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
@@ -111,10 +131,27 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
111131
return Array.from(this.sessionsDirtyFiles);
112132
}
113133

134+
getPendingSessionTargets(): MemorySyncParams["sessions"] {
135+
return Array.from(this.sessionPendingTargets.values());
136+
}
137+
138+
getPendingSessionFiles(): string[] {
139+
return Array.from(this.sessionPendingFiles);
140+
}
141+
114142
isSessionsDirty(): boolean {
115143
return this.sessionsDirty;
116144
}
117145

146+
startTranscriptListener(): void {
147+
this.ensureSessionListener();
148+
}
149+
150+
stopTranscriptListener(): void {
151+
this.sessionUnsubscribe?.();
152+
this.sessionUnsubscribe = null;
153+
}
154+
118155
protected computeProviderKey(): string {
119156
return "test";
120157
}
@@ -162,6 +199,8 @@ describe("session startup catch-up", () => {
162199
});
163200

164201
afterEach(async () => {
202+
vi.clearAllTimers();
203+
vi.useRealTimers();
165204
vi.unstubAllEnvs();
166205
await fs.rm(stateDir, { recursive: true, force: true });
167206
});
@@ -356,4 +395,84 @@ describe("session startup catch-up", () => {
356395

357396
expect(harness.indexedPaths).toEqual([]);
358397
});
398+
399+
it("queues transcript update identity without requiring a session file", async () => {
400+
vi.useFakeTimers();
401+
const harness = new SessionStartupCatchupHarness([]);
402+
const originalSubscriber = (globalThis as Record<symbol, unknown>)[
403+
MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY
404+
];
405+
let transcriptListener: ((update: MemorySessionTranscriptUpdate) => void) | undefined;
406+
(globalThis as Record<symbol, unknown>)[MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY] = ((
407+
listener,
408+
) => {
409+
transcriptListener = listener;
410+
return () => {
411+
if (transcriptListener === listener) {
412+
transcriptListener = undefined;
413+
}
414+
};
415+
}) satisfies MemoryTranscriptUpdateSubscriber;
416+
harness.startTranscriptListener();
417+
418+
try {
419+
transcriptListener?.({
420+
target: {
421+
agentId: "main",
422+
sessionId: "thread",
423+
sessionKey: "agent:main:thread",
424+
},
425+
});
426+
427+
expect(harness.getPendingSessionTargets()).toEqual([
428+
{ agentId: "main", sessionId: "thread", sessionKey: "agent:main:thread" },
429+
]);
430+
} finally {
431+
harness.stopTranscriptListener();
432+
if (originalSubscriber === undefined) {
433+
delete (globalThis as Record<symbol, unknown>)[
434+
MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY
435+
];
436+
} else {
437+
(globalThis as Record<symbol, unknown>)[MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY] =
438+
originalSubscriber;
439+
}
440+
}
441+
});
442+
443+
it("keeps canonical path transcript update compatibility", async () => {
444+
vi.useFakeTimers();
445+
const session = await writeSessionFile("thread.jsonl");
446+
const harness = new SessionStartupCatchupHarness([]);
447+
harness.startTranscriptListener();
448+
449+
emitSessionTranscriptUpdate({
450+
sessionFile: session.filePath,
451+
sessionKey: "agent:main:thread",
452+
});
453+
454+
expect(harness.getPendingSessionFiles()).toEqual([session.filePath]);
455+
expect(harness.getPendingSessionTargets()).toEqual([]);
456+
harness.stopTranscriptListener();
457+
});
458+
459+
it("prefers transcript update path compatibility before identity", async () => {
460+
vi.useFakeTimers();
461+
const session = await writeSessionFile("thread.jsonl");
462+
const harness = new SessionStartupCatchupHarness([]);
463+
harness.startTranscriptListener();
464+
465+
emitSessionTranscriptUpdate({
466+
sessionFile: session.filePath,
467+
target: {
468+
agentId: "main",
469+
sessionId: "identity-target",
470+
sessionKey: "agent:main:identity-target",
471+
},
472+
});
473+
474+
expect(harness.getPendingSessionFiles()).toEqual([session.filePath]);
475+
expect(harness.getPendingSessionTargets()).toEqual([]);
476+
harness.stopTranscriptListener();
477+
});
359478
});

‎extensions/memory-core/src/memory/manager-sync-ops.ts‎

Lines changed: 59 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -170,9 +170,27 @@ const IGNORED_MEMORY_WATCH_DIR_NAMES = new Set([
170170
]);
171171

172172
const log = createSubsystemLogger("memory");
173+
const MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY = Symbol.for(
174+
"openclaw.memoryCore.sessionTranscriptUpdateSubscriber",
175+
);
173176
const TEST_MEMORY_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryWatchFactory");
174177
const TEST_MEMORY_NATIVE_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryNativeWatchFactory");
175178

179+
type MemorySessionTranscriptUpdate = {
180+
agentId?: string;
181+
sessionFile?: string;
182+
sessionKey?: string;
183+
target?: {
184+
agentId: string;
185+
sessionId: string;
186+
sessionKey: string;
187+
};
188+
};
189+
190+
type MemoryTranscriptUpdateSubscriber = (
191+
listener: (update: MemorySessionTranscriptUpdate) => void,
192+
) => () => void;
193+
176194
function memoryTableExists(db: DatabaseSync, tableName: string): boolean {
177195
return Boolean(
178196
db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(tableName),
@@ -191,6 +209,18 @@ type LinuxMemoryDirectoryWatcher = {
191209
ino: number;
192210
};
193211

212+
function subscribeMemorySessionTranscriptUpdates(
213+
listener: (update: MemorySessionTranscriptUpdate) => void,
214+
): () => void {
215+
const injected = (globalThis as Record<symbol, unknown>)[
216+
MEMORY_CORE_TRANSCRIPT_UPDATE_SUBSCRIBER_KEY
217+
];
218+
if (typeof injected === "function") {
219+
return (injected as MemoryTranscriptUpdateSubscriber)(listener);
220+
}
221+
return onSessionTranscriptUpdate(listener);
222+
}
223+
194224
function resolveMemoryWatchFactory(): typeof chokidar.watch {
195225
if (process.env.VITEST === "true" || process.env.NODE_ENV === "test") {
196226
const override = (globalThis as Record<PropertyKey, unknown>)[TEST_MEMORY_WATCH_FACTORY_KEY];
@@ -1422,20 +1452,22 @@ export abstract class MemoryManagerSyncOps {
14221452
if (!this.sources.has("sessions") || this.sessionUnsubscribe) {
14231453
return;
14241454
}
1425-
this.sessionUnsubscribe = onSessionTranscriptUpdate((update) => {
1455+
this.sessionUnsubscribe = subscribeMemorySessionTranscriptUpdates((update) => {
14261456
if (this.closed) {
14271457
return;
14281458
}
14291459
const sessionFile = update.sessionFile;
1430-
if (!this.isSessionFileForAgent(sessionFile)) {
1460+
if (sessionFile && isSessionArchiveArtifactName(path.basename(sessionFile))) {
1461+
return;
1462+
}
1463+
if (sessionFile && this.isSessionFileForAgent(sessionFile)) {
1464+
this.scheduleSessionDirty(sessionFile);
14311465
return;
14321466
}
14331467
const target = this.resolveSessionTranscriptUpdateSyncTarget(update);
14341468
if (target) {
14351469
this.scheduleSessionDirty(target);
1436-
return;
14371470
}
1438-
this.scheduleSessionDirty(sessionFile);
14391471
});
14401472
}
14411473

@@ -1703,13 +1735,30 @@ export abstract class MemoryManagerSyncOps {
17031735
return resolvedFile.startsWith(`${resolvedDir}${path.sep}`);
17041736
}
17051737

1706-
private resolveSessionTranscriptUpdateSyncTarget(update: {
1707-
agentId?: string;
1708-
sessionFile: string;
1709-
sessionKey?: string;
1710-
}): MemorySessionSyncTarget | null {
1738+
private resolveSessionTranscriptUpdateSyncTarget(
1739+
update: MemorySessionTranscriptUpdate,
1740+
): MemorySessionSyncTarget | null {
1741+
if (update.sessionFile && isSessionArchiveArtifactName(path.basename(update.sessionFile))) {
1742+
return null;
1743+
}
1744+
if (update.target) {
1745+
const agentId = update.target.agentId.trim();
1746+
const sessionId = update.target.sessionId.trim();
1747+
const sessionKey = update.target.sessionKey.trim();
1748+
if (!agentId || !sessionId || normalizeAgentId(agentId) !== normalizeAgentId(this.agentId)) {
1749+
return null;
1750+
}
1751+
return {
1752+
agentId,
1753+
sessionId,
1754+
...(sessionKey ? { sessionKey } : {}),
1755+
};
1756+
}
1757+
if (!update.sessionFile) {
1758+
return null;
1759+
}
17111760
const parsed = parseCanonicalSessionSyncTargetFromPath(update.sessionFile);
1712-
if (!parsed || isSessionArchiveArtifactName(path.basename(update.sessionFile))) {
1761+
if (!parsed) {
17131762
return null;
17141763
}
17151764
const agentId = update.agentId?.trim() || parsed.agentId;

‎extensions/telegram/src/bot-message-dispatch.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -393,6 +393,11 @@ async function mirrorTelegramAssistantReplyToTranscript(params: {
393393
sessionFile,
394394
sessionKey: params.sessionKey,
395395
agentId: params.route.agentId,
396+
target: {
397+
agentId: params.route.agentId,
398+
sessionId: sessionEntry.sessionId,
399+
sessionKey: params.sessionKey,
400+
},
396401
message: appendedMessage,
397402
messageId,
398403
});

‎src/agents/embedded-agent-runner/tool-result-truncation.ts‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -942,6 +942,15 @@ function truncateOversizedToolResultsInExistingSessionManager(params: {
942942
sessionFile: params.sessionFile,
943943
sessionKey: params.sessionKey,
944944
...(params.agentId ? { agentId: params.agentId } : {}),
945+
...(params.sessionId && params.sessionKey && params.agentId
946+
? {
947+
target: {
948+
agentId: params.agentId,
949+
sessionId: params.sessionId,
950+
sessionKey: params.sessionKey,
951+
},
952+
}
953+
: {}),
945954
});
946955
}
947956

@@ -1013,6 +1022,15 @@ async function truncateOversizedToolResultsInTranscriptState(params: {
10131022
sessionFile: params.sessionFile,
10141023
sessionKey: params.sessionKey,
10151024
...(params.agentId ? { agentId: params.agentId } : {}),
1025+
...(params.sessionId && params.sessionKey && params.agentId
1026+
? {
1027+
target: {
1028+
agentId: params.agentId,
1029+
sessionId: params.sessionId,
1030+
sessionKey: params.sessionKey,
1031+
},
1032+
}
1033+
: {}),
10161034
});
10171035
}
10181036

@@ -1079,6 +1097,7 @@ export async function truncateOversizedToolResultsInSession(params: {
10791097
sessionFile,
10801098
sessionId: params.sessionId,
10811099
sessionKey: params.sessionKey,
1100+
agentId: params.agentId,
10821101
});
10831102
} catch (err) {
10841103
const errMsg = formatErrorMessage(err);

0 commit comments

Comments
 (0)