@@ -3,6 +3,7 @@ import fs from "node:fs/promises";
33import os from "node:os";
44import path from "node:path";
55import type { DatabaseSync } from "node:sqlite";
6+import { emitSessionTranscriptUpdate } from "openclaw/plugin-sdk/agent-harness-runtime";
67import {
78resolveSessionTranscriptsDirForAgent,
89type OpenClawConfig,
@@ -33,6 +34,25 @@ type SyncParams = {
3334progress?: (update: MemorySyncProgressUpdate) => void;
3435};
353637+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+3656type SourceStateRow = { path: string; hash: string; mtime: number; size: number };
37573858class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
@@ -111,10 +131,27 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
111131return Array.from(this.sessionsDirtyFiles);
112132}
113133134+getPendingSessionTargets(): MemorySyncParams["sessions"] {
135+return Array.from(this.sessionPendingTargets.values());
136+}
137+138+getPendingSessionFiles(): string[] {
139+return Array.from(this.sessionPendingFiles);
140+}
141+114142isSessionsDirty(): boolean {
115143return this.sessionsDirty;
116144}
117145146+startTranscriptListener(): void {
147+this.ensureSessionListener();
148+}
149+150+stopTranscriptListener(): void {
151+this.sessionUnsubscribe?.();
152+this.sessionUnsubscribe = null;
153+}
154+118155protected computeProviderKey(): string {
119156return "test";
120157}
@@ -162,6 +199,8 @@ describe("session startup catch-up", () => {
162199});
163200164201afterEach(async () => {
202+vi.clearAllTimers();
203+vi.useRealTimers();
165204vi.unstubAllEnvs();
166205await fs.rm(stateDir, { recursive: true, force: true });
167206});
@@ -356,4 +395,84 @@ describe("session startup catch-up", () => {
356395357396expect(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});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。