diff --git a/src/agents/tools/embedded-gateway-stub.test.ts b/src/agents/tools/embedded-gateway-stub.test.ts index f35aeb709377..6a3ead831baf 100644 --- a/src/agents/tools/embedded-gateway-stub.test.ts +++ b/src/agents/tools/embedded-gateway-stub.test.ts @@ -131,6 +131,7 @@ describe("embedded gateway stub", () => { mode: "recent", maxMessages: 200, maxBytes: 1024 * 1024, + allowResetArchiveFallback: true, }, ); expect(result.messages).toEqual(projectedMessages); @@ -193,6 +194,7 @@ describe("embedded gateway stub", () => { mode: "recent", maxMessages: 1, maxBytes: 1024 * 1024, + allowResetArchiveFallback: true, }, ); }); @@ -222,6 +224,7 @@ describe("embedded gateway stub", () => { mode: "recent", maxMessages: 2, maxBytes: 1024 * 1024, + allowResetArchiveFallback: true, }, ); }); diff --git a/src/agents/tools/embedded-gateway-stub.ts b/src/agents/tools/embedded-gateway-stub.ts index a9a4c257221a..cd1b2c3a977d 100644 --- a/src/agents/tools/embedded-gateway-stub.ts +++ b/src/agents/tools/embedded-gateway-stub.ts @@ -162,6 +162,7 @@ async function handleChatHistory(params: Record): Promise<{ mode: "recent", maxMessages: max, maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024), + allowResetArchiveFallback: true, }, ) : []; diff --git a/src/gateway/managed-image-attachments.test.ts b/src/gateway/managed-image-attachments.test.ts index 9f51db9d8f26..1d6a11285eb2 100644 --- a/src/gateway/managed-image-attachments.test.ts +++ b/src/gateway/managed-image-attachments.test.ts @@ -19,6 +19,7 @@ const resolveOpenAiCompatibleHttpOperatorScopesMock = vi.fn(); const resolveOpenAiCompatibleHttpSenderIsOwnerMock = vi.fn(); const loadSessionEntryMock = vi.fn(); const readSessionMessagesMock = vi.fn(); +const resolveSessionHistoryTranscriptPathMock = vi.fn(); const getRuntimeConfigMock = vi.fn(() => ({})); vi.mock("../config/config.js", () => ({ @@ -34,6 +35,11 @@ vi.mock("./http-utils.js", () => ({ vi.mock("./session-utils.js", () => ({ loadSessionEntry: loadSessionEntryMock, readSessionMessagesAsync: readSessionMessagesMock, + readSessionMessagesWithSourceAsync: async (...args: unknown[]) => ({ + messages: await readSessionMessagesMock(...args), + transcriptPath: await resolveSessionHistoryTranscriptPathMock(...args), + }), + resolveSessionHistoryTranscriptPathAsync: resolveSessionHistoryTranscriptPathMock, })); const { @@ -144,6 +150,8 @@ async function requestManagedImage(params: { headers?: Record; transcriptMessages?: Record[]; sessionEntry?: { sessionId: string; sessionFile?: string }; + resolvedTranscriptPath?: string | null; + onReadTranscriptMessages?: () => Promise | void; }) { authorizeGatewayHttpRequestOrReplyMock.mockImplementation(async ({ res }) => { if (params.denyAuth) { @@ -167,21 +175,27 @@ async function requestManagedImage(params: { storePath: path.join(params.stateDir, "gateway-sessions.json"), entry: params.sessionEntry ?? { sessionId: "sess-1", sessionFile: "session.jsonl" }, }); - readSessionMessagesMock.mockReturnValue( - params.transcriptMessages ?? [ - { - role: "assistant", - content: [ - { - type: "image", - url: params.pathName, - openUrl: params.pathName, - }, - ], - __openclaw: { id: "msg-1" }, - }, - ], + resolveSessionHistoryTranscriptPathMock.mockResolvedValue( + params.resolvedTranscriptPath ?? params.sessionEntry?.sessionFile ?? "session.jsonl", ); + readSessionMessagesMock.mockImplementation(async () => { + await params.onReadTranscriptMessages?.(); + return ( + params.transcriptMessages ?? [ + { + role: "assistant", + content: [ + { + type: "image", + url: params.pathName, + openUrl: params.pathName, + }, + ], + __openclaw: { id: "msg-1" }, + }, + ] + ); + }); const auth = { mode: "test" } as never; const server = http.createServer((req, res) => { @@ -272,6 +286,12 @@ describe("handleManagedOutgoingImageHttpRequest", () => { expect(result.headers["content-type"]).toBe("image/png"); expect(result.headers["content-disposition"]).toContain("inline"); expect(result.body.toString("utf-8")).toBe("original-image"); + expect(readSessionMessagesMock).toHaveBeenCalledWith( + "sess-1", + path.join(stateDir, "gateway-sessions.json"), + "session.jsonl", + expect.objectContaining({ allowResetArchiveFallback: true }), + ); }); it("rejects unauthenticated requests before serving bytes", async () => { @@ -404,6 +424,99 @@ describe("handleManagedOutgoingImageHttpRequest", () => { expect(third.result.statusCode).toBe(200); expect(readSessionMessagesMock).toHaveBeenCalledTimes(2); }); + + it("reuses the session attachment index for archive-backed requests", async () => { + const { attachmentId, sessionKey } = await createFixture(stateDir); + const archiveFile = path.join( + stateDir, + "sessions", + "sess-main.jsonl.reset.2026-02-16T22-26-34.000Z", + ); + await fs.mkdir(path.dirname(archiveFile), { recursive: true }); + await fs.writeFile(archiveFile, '{"message":{}}\n', "utf-8"); + + const transcriptMessages = [ + { + __openclaw: { id: "msg-1" }, + content: [ + { + type: "image", + url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`, + openUrl: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`, + }, + ], + }, + ]; + + const pathName = `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`; + const first = await requestManagedImage({ + stateDir, + pathName, + authResponse: { authMethod: "token" }, + sessionEntry: { sessionId: "sess-main" }, + resolvedTranscriptPath: archiveFile, + transcriptMessages, + }); + const second = await requestManagedImage({ + stateDir, + pathName, + authResponse: { authMethod: "token" }, + sessionEntry: { sessionId: "sess-main" }, + resolvedTranscriptPath: archiveFile, + transcriptMessages, + }); + + expect(first.result.statusCode).toBe(200); + expect(second.result.statusCode).toBe(200); + expect(readSessionMessagesMock).toHaveBeenCalledTimes(1); + }); + + it("does not cache a session attachment index when the transcript changes during the read", async () => { + const { attachmentId, sessionKey } = await createFixture(stateDir); + const sessionFile = path.join(stateDir, "sessions", "sess-main.jsonl"); + await fs.mkdir(path.dirname(sessionFile), { recursive: true }); + await fs.writeFile(sessionFile, '{"message":{}}\n', "utf-8"); + + const transcriptMessages = [ + { + __openclaw: { id: "msg-1" }, + content: [ + { + type: "image", + url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`, + openUrl: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`, + }, + ], + }, + ]; + + let mutatedTranscript = false; + const pathName = `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`; + const first = await requestManagedImage({ + stateDir, + pathName, + authResponse: { authMethod: "token" }, + sessionEntry: { sessionId: "sess-main", sessionFile }, + transcriptMessages, + onReadTranscriptMessages: async () => { + if (!mutatedTranscript) { + mutatedTranscript = true; + await fs.appendFile(sessionFile, '{"message":{"content":"updated"}}\n', "utf-8"); + } + }, + }); + const second = await requestManagedImage({ + stateDir, + pathName, + authResponse: { authMethod: "token" }, + sessionEntry: { sessionId: "sess-main", sessionFile }, + transcriptMessages, + }); + + expect(first.result.statusCode).toBe(200); + expect(second.result.statusCode).toBe(200); + expect(readSessionMessagesMock).toHaveBeenCalledTimes(2); + }); }); describe("createManagedOutgoingImageBlocks", () => { @@ -942,6 +1055,12 @@ describe("cleanupManagedOutgoingImageRecords", () => { expect(result.deletedFileCount).toBe(0); expect(result.retainedCount).toBe(1); await expect(fs.access(fixture.originalPath)).resolves.toBeUndefined(); + expect(readSessionMessagesMock).toHaveBeenCalledWith( + "sess-main", + path.join(stateDir, "gateway-sessions.json"), + "/tmp/sess-main.jsonl", + expect.objectContaining({ allowResetArchiveFallback: true }), + ); }); it("reads each session transcript once while evaluating committed records", async () => { diff --git a/src/gateway/managed-image-attachments.ts b/src/gateway/managed-image-attachments.ts index 9c2adccdfeea..f40400605899 100644 --- a/src/gateway/managed-image-attachments.ts +++ b/src/gateway/managed-image-attachments.ts @@ -28,7 +28,11 @@ import { resolveOpenAiCompatibleHttpSenderIsOwner, } from "./http-utils.js"; import { authorizeOperatorScopesForMethod } from "./method-scopes.js"; -import { loadSessionEntry, readSessionMessagesAsync } from "./session-utils.js"; +import { + loadSessionEntry, + readSessionMessagesWithSourceAsync, + resolveSessionHistoryTranscriptPathAsync, +} from "./session-utils.js"; const OUTGOING_IMAGE_ROUTE_PREFIX = "/api/chat/media/outgoing"; const DEFAULT_TRANSIENT_OUTGOING_IMAGE_TTL_MS = 15 * 60 * 1000; @@ -99,6 +103,10 @@ type SessionManagedOutgoingAttachmentIndexCacheEntry = { size: number; index: SessionManagedOutgoingAttachmentIndex; }; +type SessionManagedOutgoingAttachmentTranscriptStat = Omit< + SessionManagedOutgoingAttachmentIndexCacheEntry, + "index" +>; const sessionManagedOutgoingAttachmentIndexCache = new Map< string, @@ -638,7 +646,7 @@ function getCachedSessionManagedOutgoingAttachmentIndex( function setCachedSessionManagedOutgoingAttachmentIndex( sessionKey: string, agentId: string | undefined, - stat: { transcriptPath: string; mtimeMs: number; size: number }, + stat: SessionManagedOutgoingAttachmentTranscriptStat, index: SessionManagedOutgoingAttachmentIndex, ) { sessionManagedOutgoingAttachmentIndexCache.set( @@ -662,6 +670,17 @@ function setCachedSessionManagedOutgoingAttachmentIndex( } } +function sameManagedOutgoingAttachmentTranscriptStat( + left: SessionManagedOutgoingAttachmentTranscriptStat | null, + right: SessionManagedOutgoingAttachmentTranscriptStat | null, +): boolean { + return ( + left?.transcriptPath === right?.transcriptPath && + left?.mtimeMs === right?.mtimeMs && + left?.size === right?.size + ); +} + async function getSessionManagedOutgoingAttachmentIndex( sessionKey: string, cache?: Map, @@ -681,13 +700,18 @@ async function getSessionManagedOutgoingAttachmentIndex( return null; } - let transcriptStat: { transcriptPath: string; mtimeMs: number; size: number } | null = null; - const transcriptPath = typeof entry?.sessionFile === "string" ? entry.sessionFile.trim() : ""; - if (transcriptPath) { + let transcriptStat: SessionManagedOutgoingAttachmentTranscriptStat | null = null; + const resolvedTranscriptPath = await resolveSessionHistoryTranscriptPathAsync( + sessionId, + storePath, + entry.sessionFile, + { allowResetArchiveFallback: true }, + ); + if (resolvedTranscriptPath) { try { - const stat = await fs.stat(transcriptPath); + const stat = await fs.stat(resolvedTranscriptPath); transcriptStat = { - transcriptPath, + transcriptPath: resolvedTranscriptPath, mtimeMs: stat.mtimeMs, size: stat.size, }; @@ -703,12 +727,42 @@ async function getSessionManagedOutgoingAttachmentIndex( } catch { sessionManagedOutgoingAttachmentIndexCache.delete(cacheKey); } + } else { + sessionManagedOutgoingAttachmentIndexCache.delete(cacheKey); } - const messages = await readSessionMessagesAsync(sessionId, storePath, entry.sessionFile, { - mode: "full", - reason: "managed outgoing attachment index", - }); + const readResult = await readSessionMessagesWithSourceAsync( + sessionId, + storePath, + entry.sessionFile, + { + mode: "full", + reason: "managed outgoing attachment index", + allowResetArchiveFallback: true, + }, + ); + const messages = readResult.messages; + const preReadTranscriptStat = transcriptStat; + if (readResult.transcriptPath) { + try { + const stat = await fs.stat(readResult.transcriptPath); + const postReadTranscriptStat = { + transcriptPath: readResult.transcriptPath, + mtimeMs: stat.mtimeMs, + size: stat.size, + }; + transcriptStat = sameManagedOutgoingAttachmentTranscriptStat( + preReadTranscriptStat, + postReadTranscriptStat, + ) + ? postReadTranscriptStat + : null; + } catch { + transcriptStat = null; + } + } else { + transcriptStat = null; + } const index: SessionManagedOutgoingAttachmentIndex = new Set(); for (const message of messages) { const meta = (message as { __openclaw?: { id?: string } } | null)?.["__openclaw"]; diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index e16b55c460a9..48b2a419d99d 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -2442,6 +2442,7 @@ async function isChatMessageIdVisibleAfterHistoryFilters(params: { sessionFile: string | undefined; messageId: string; sessionStartedAt?: number; + allowResetArchiveFallback?: boolean; }): Promise { if (params.sessionStartedAt === undefined) { return true; @@ -2453,6 +2454,7 @@ async function isChatMessageIdVisibleAfterHistoryFilters(params: { { mode: "full", reason: "chat.message.get visibility", + ...(params.allowResetArchiveFallback === true ? { allowResetArchiveFallback: true } : {}), }, ); return dropPreSessionStartAnnouncePairs(messages, params.sessionStartedAt).some( @@ -2565,6 +2567,7 @@ async function handleChatHistoryRequest({ ? await readRecentSessionMessagesAsync(sessionId, storePath, entry?.sessionFile, { ...localHistoryReadOptions, maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024), + allowResetArchiveFallback: true, }) : []; const overreadContextMessage = @@ -2746,6 +2749,7 @@ export const chatHandlers: GatewayRequestHandlers = { storePath, entry?.sessionFile, messageId, + { allowResetArchiveFallback: true }, ); if (!resolved.found) { respond(true, { ok: false, unavailableReason: "not_found" }); @@ -2758,6 +2762,7 @@ export const chatHandlers: GatewayRequestHandlers = { messageId, sessionStartedAt: typeof entry?.sessionStartedAt === "number" ? entry.sessionStartedAt : undefined, + allowResetArchiveFallback: true, }); if (!visible) { respond(true, { ok: false, unavailableReason: "not_found" }); diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index aa50946cdf97..977c3cab1d3a 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/src/gateway/server-methods/sessions.ts @@ -2436,6 +2436,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { { maxMessages: limit, maxLines: limit * 20 + 20, + allowResetArchiveFallback: true, }, ); respond(true, { messages }, undefined); diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index 1947b2272da4..01f54cf92e20 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -2721,6 +2721,39 @@ describe("gateway server chat", () => { }); }); + test("chat.message.get returns archive-backed rows surfaced by history", async () => { + await withGatewayChatHarness(async ({ ws, createSessionDir }) => { + const sessionDir = await prepareMainHistoryHarness({ ws, createSessionDir }); + await fs.writeFile( + path.join(sessionDir, "sess-main.jsonl.reset.2026-02-16T22-26-34.000Z"), + [ + JSON.stringify({ type: "session", version: 1, id: "sess-main" }), + JSON.stringify({ + id: "msg-archive-full-assistant", + message: { + role: "assistant", + content: [{ type: "text", text: "archive abcdefghij" }], + timestamp: Date.now(), + }, + }), + ].join("\n"), + "utf-8", + ); + + const historyMessages = await fetchHistoryMessages(ws, { maxChars: 12 }); + expect(JSON.stringify(historyMessages)).toContain("archive abcd\\n...(truncated)..."); + + const full = await fetchChatMessage(ws, { + sessionKey: "main", + messageId: "msg-archive-full-assistant", + }); + expect(full.ok).toBe(true); + expect(full.unavailableReason).toBeUndefined(); + expect(JSON.stringify(full.message)).toContain("archive abcdefghij"); + expect(JSON.stringify(full.message)).not.toContain("...(truncated)..."); + }); + }); + test("chat.message.get accepts the selected agent for global sessions", async () => { await withGatewayChatHarness(async ({ ws, createSessionDir }) => { await writeGatewayConfig({ diff --git a/src/gateway/session-history-state.ts b/src/gateway/session-history-state.ts index c4ab69572415..87b422b593b5 100644 --- a/src/gateway/session-history-state.ts +++ b/src/gateway/session-history-state.ts @@ -5,10 +5,11 @@ import { DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS, projectChatDisplayMessages, } from "./chat-display-projection.js"; +import { resolveTranscriptPathForComparison } from "./session-transcript-path.js"; import { attachOpenClawTranscriptMeta, readRecentSessionMessagesWithStatsAsync, - readSessionMessagesAsync, + readSessionMessagesWithSourceAsync, } from "./session-utils.js"; // Session history state owns the SSE-friendly projection of transcript JSONL: @@ -50,6 +51,7 @@ type SessionHistoryRawSnapshot = { rawMessages: unknown[]; rawTranscriptSeq?: number; totalRawMessages?: number; + transcriptPath?: string; }; /** Computes an oversized raw transcript tail window for projected chat history. */ @@ -181,12 +183,14 @@ export class SessionHistorySseState { private readonly cursor: string | undefined; private sentHistory: PaginatedSessionHistory; private rawTranscriptSeq: number; + private transcriptPath: string | undefined; static fromRawSnapshot(params: { target: SessionHistoryTranscriptTarget; rawMessages: unknown[]; rawTranscriptSeq?: number; totalRawMessages?: number; + transcriptPath?: string; maxChars?: number; limit?: number; cursor?: string; @@ -199,6 +203,7 @@ export class SessionHistorySseState { initialRawMessages: params.rawMessages, rawTranscriptSeq: params.rawTranscriptSeq, totalRawMessages: params.totalRawMessages, + transcriptPath: params.transcriptPath, }); } @@ -210,6 +215,7 @@ export class SessionHistorySseState { initialRawMessages: unknown[]; rawTranscriptSeq?: number; totalRawMessages?: number; + transcriptPath?: string; }) { this.target = params.target; this.maxChars = params.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS; @@ -226,6 +232,7 @@ export class SessionHistorySseState { }); this.sentHistory = snapshot.history; this.rawTranscriptSeq = snapshot.rawTranscriptSeq; + this.transcriptPath = normalizeTranscriptPathForComparison(params.transcriptPath); } snapshot(): PaginatedSessionHistory { @@ -322,10 +329,16 @@ export class SessionHistorySseState { }; } + shouldRefreshForTranscriptPath(updatePath: string | undefined): boolean { + const nextPath = normalizeTranscriptPathForComparison(updatePath); + return Boolean(this.transcriptPath && nextPath && this.transcriptPath !== nextPath); + } + async refreshAsync(): Promise { const rawSnapshot = await this.readRawSnapshotAsync(); const snapshot = this.buildSnapshot(rawSnapshot); this.rawTranscriptSeq = snapshot.rawTranscriptSeq; + this.transcriptPath = normalizeTranscriptPathForComparison(rawSnapshot.transcriptPath); this.sentHistory = snapshot.history; return snapshot.history; } @@ -353,24 +366,33 @@ export class SessionHistorySseState { this.target.sessionFile, { ...resolveSessionHistoryTailReadOptions(this.limit), + allowResetArchiveFallback: true, }, ); return { rawMessages: snapshot.messages, rawTranscriptSeq: snapshot.totalMessages, totalRawMessages: snapshot.totalMessages, + transcriptPath: snapshot.transcriptPath, }; } + const snapshot = await readSessionMessagesWithSourceAsync( + this.target.sessionId, + this.target.storePath, + this.target.sessionFile, + { + mode: "full", + reason: "session history cursor pagination", + allowResetArchiveFallback: true, + }, + ); return { - rawMessages: await readSessionMessagesAsync( - this.target.sessionId, - this.target.storePath, - this.target.sessionFile, - { - mode: "full", - reason: "session history cursor pagination", - }, - ), + rawMessages: snapshot.messages, + transcriptPath: snapshot.transcriptPath, }; } } + +function normalizeTranscriptPathForComparison(filePath: string | undefined): string | undefined { + return typeof filePath === "string" ? resolveTranscriptPathForComparison(filePath) : undefined; +} diff --git a/src/gateway/session-transcript-files.fs.ts b/src/gateway/session-transcript-files.fs.ts index 45b4469dd7cf..e7282c7dbb50 100644 --- a/src/gateway/session-transcript-files.fs.ts +++ b/src/gateway/session-transcript-files.fs.ts @@ -18,11 +18,83 @@ import { resolveRequiredHomeDir } from "../infra/home-dir.js"; import { emitSessionTranscriptUpdate } from "../sessions/transcript-events.js"; type ArchiveFileReason = SessionArchiveReason; +type ResetArchiveCandidate = { archivePath: string; name: string; timestamp: number }; export type ArchivedSessionTranscript = { sourcePath: string; archivedPath: string; }; +const MAX_RESET_ARCHIVE_DISCOVERY_CACHE_ENTRIES = 2048; +const MAX_RESET_ARCHIVE_HEADER_MATCH_CACHE_ENTRIES = 4096; +const MAX_RESET_ARCHIVE_CANDIDATES_PER_TRANSCRIPT = 128; + +const resetArchiveDiscoveryCache = new Map< + string, + { + dirMtimeMs: number; + dirSize: number; + archives: ResetArchiveCandidate[]; + } +>(); +const resetArchiveHeaderMatchCache = new Map< + string, + { + mtimeMs: number; + size: number; + matches: boolean; + } +>(); + +function clearSessionTranscriptResetArchiveDiscoveryCache(): void { + resetArchiveDiscoveryCache.clear(); + resetArchiveHeaderMatchCache.clear(); +} + +function deleteResetArchiveHeaderMatchesForArchives(archives: ResetArchiveCandidate[]): void { + if (archives.length === 0 || resetArchiveHeaderMatchCache.size === 0) { + return; + } + const archivePaths = new Set(archives.map((archive) => archive.archivePath)); + for (const cacheKey of resetArchiveHeaderMatchCache.keys()) { + const archivePath = cacheKey.slice(cacheKey.indexOf("\0") + 1); + if (archivePaths.has(archivePath)) { + resetArchiveHeaderMatchCache.delete(cacheKey); + } + } +} + +function setResetArchiveDiscoveryCacheEntry( + cacheKey: string, + entry: { dirMtimeMs: number; dirSize: number; archives: ResetArchiveCandidate[] }, +): void { + resetArchiveDiscoveryCache.set(cacheKey, entry); + while (resetArchiveDiscoveryCache.size > MAX_RESET_ARCHIVE_DISCOVERY_CACHE_ENTRIES) { + const oldestKey = resetArchiveDiscoveryCache.keys().next().value; + if (typeof oldestKey !== "string") { + break; + } + const oldestEntry = resetArchiveDiscoveryCache.get(oldestKey); + if (oldestEntry) { + deleteResetArchiveHeaderMatchesForArchives(oldestEntry.archives); + } + resetArchiveDiscoveryCache.delete(oldestKey); + } +} + +function setResetArchiveHeaderMatchCacheEntry( + cacheKey: string, + entry: { mtimeMs: number; size: number; matches: boolean }, +): void { + resetArchiveHeaderMatchCache.set(cacheKey, entry); + while (resetArchiveHeaderMatchCache.size > MAX_RESET_ARCHIVE_HEADER_MATCH_CACHE_ENTRIES) { + const oldestKey = resetArchiveHeaderMatchCache.keys().next().value; + if (typeof oldestKey !== "string") { + break; + } + resetArchiveHeaderMatchCache.delete(oldestKey); + } +} + function classifySessionTranscriptCandidate( sessionId: string, sessionFile?: string, @@ -129,10 +201,189 @@ export function resolveSessionTranscriptCandidates( return uniqueStrings(candidates); } +async function resetArchiveHeaderMatchesSessionId( + sessionId: string, + archivePath: string, +): Promise { + const stat = await fs.promises.stat(archivePath).catch(() => null); + if (!stat?.isFile()) { + return false; + } + const cacheKey = `${sessionId}\0${archivePath}`; + const cached = resetArchiveHeaderMatchCache.get(cacheKey); + if (cached && cached.mtimeMs === stat.mtimeMs && cached.size === stat.size) { + resetArchiveHeaderMatchCache.delete(cacheKey); + resetArchiveHeaderMatchCache.set(cacheKey, cached); + return cached.matches; + } + + let matches = false; + const handle = await fs.promises.open(archivePath, "r").catch(() => null); + if (!handle) { + return false; + } + try { + const buffer = Buffer.alloc(64 * 1024); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0); + const lines = buffer.toString("utf-8", 0, bytesRead).split(/\r?\n/); + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) { + continue; + } + const record = JSON.parse(trimmed) as unknown; + matches = + Boolean(record) && + typeof record === "object" && + !Array.isArray(record) && + (record as { type?: unknown; id?: unknown }).type === "session" && + (record as { type?: unknown; id?: unknown }).id === sessionId; + return matches; + } + return false; + } catch { + return false; + } finally { + await handle.close().catch(() => undefined); + setResetArchiveHeaderMatchCacheEntry(cacheKey, { + mtimeMs: stat.mtimeMs, + size: stat.size, + matches, + }); + } +} + +async function listResetArchiveCandidatesForTranscriptAsync( + transcriptPath: string, +): Promise { + const base = path.basename(transcriptPath); + if (!base.endsWith(".jsonl")) { + return undefined; + } + const dir = path.dirname(transcriptPath); + const dirStat = await fs.promises.stat(dir).catch(() => null); + if (!dirStat?.isDirectory()) { + return undefined; + } + const cacheKey = `${dir}\0${base}`; + const cached = resetArchiveDiscoveryCache.get(cacheKey); + if (cached && cached.dirMtimeMs === dirStat.mtimeMs && cached.dirSize === dirStat.size) { + resetArchiveDiscoveryCache.delete(cacheKey); + resetArchiveDiscoveryCache.set(cacheKey, cached); + return cached.archives; + } + + const archives: ResetArchiveCandidate[] = []; + try { + for (const entry of await fs.promises.readdir(dir, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.startsWith(`${base}.reset.`)) { + continue; + } + const timestamp = parseSessionArchiveTimestamp(entry.name, "reset"); + if (timestamp == null) { + continue; + } + archives.push({ archivePath: path.join(dir, entry.name), name: entry.name, timestamp }); + } + } catch { + return undefined; + } + archives.sort( + (left, right) => right.timestamp - left.timestamp || right.name.localeCompare(left.name), + ); + const boundedArchives = archives.slice(0, MAX_RESET_ARCHIVE_CANDIDATES_PER_TRANSCRIPT); + setResetArchiveDiscoveryCacheEntry(cacheKey, { + dirMtimeMs: dirStat.mtimeMs, + dirSize: dirStat.size, + archives: boundedArchives, + }); + return boundedArchives; +} + +async function resolveLatestResetArchiveForTranscriptAsync( + sessionId: string, + transcriptPath: string, + opts?: { requireSessionHeader?: boolean }, +): Promise { + const archives = await listResetArchiveCandidatesForTranscriptAsync(transcriptPath); + if (!archives) { + return undefined; + } + if (opts?.requireSessionHeader !== true) { + return archives[0]; + } + for (const archive of archives) { + if (await resetArchiveHeaderMatchesSessionId(sessionId, archive.archivePath)) { + return archive; + } + } + return undefined; +} + +function transcriptArchiveIdentity( + sessionId: string, + transcriptPath: string, +): { key: string; requireSessionHeader: boolean } | undefined { + const generatedSessionId = extractGeneratedTranscriptSessionId(transcriptPath); + return { + key: path.basename(transcriptPath), + requireSessionHeader: !generatedSessionId || generatedSessionId !== sessionId, + }; +} + +export async function resolveSessionTranscriptResetArchiveCandidatesAsync( + sessionId: string, + storePath: string | undefined, + sessionFile?: string, + agentId?: string, +): Promise { + const candidatesByIdentity = new Map< + string, + Array<{ path: string; requireSessionHeader: boolean }> + >(); + for (const candidate of resolveSessionTranscriptCandidates( + sessionId, + storePath, + sessionFile, + agentId, + )) { + const identity = transcriptArchiveIdentity(sessionId, candidate); + if (!identity) { + continue; + } + candidatesByIdentity.set(identity.key, [ + ...(candidatesByIdentity.get(identity.key) ?? []), + { path: candidate, requireSessionHeader: identity.requireSessionHeader }, + ]); + } + const archives = ( + await Promise.all( + Array.from(candidatesByIdentity.values(), (candidates) => + Promise.all( + candidates.map((candidate) => + resolveLatestResetArchiveForTranscriptAsync(sessionId, candidate.path, { + requireSessionHeader: candidate.requireSessionHeader, + }), + ), + ), + ), + ) + ).flatMap((identityArchives) => + identityArchives + .flatMap((archive) => (archive ? [archive] : [])) + .sort( + (left, right) => right.timestamp - left.timestamp || right.name.localeCompare(left.name), + ) + .slice(0, 1), + ); + return uniqueStrings(archives.map((archive) => archive.archivePath)); +} + export function archiveFileOnDisk(filePath: string, reason: ArchiveFileReason): string { const ts = formatSessionArchiveTimestamp(); const archived = `${filePath}.${reason}.${ts}`; fs.renameSync(filePath, archived); + clearSessionTranscriptResetArchiveDiscoveryCache(); // Notify the session transcript subscribers (memory index, sessions-history // HTTP, etc.) that a mutation landed on a session-owned path. Without this // emit the memory sync's incremental path never learns the new archive diff --git a/src/gateway/session-utils.fs.test.ts b/src/gateway/session-utils.fs.test.ts index 130a08563c6b..cb0ed301178a 100644 --- a/src/gateway/session-utils.fs.test.ts +++ b/src/gateway/session-utils.fs.test.ts @@ -79,6 +79,17 @@ function writeTranscript(tmpDir: string, sessionId: string, lines: unknown[]): s return transcriptPath; } +function writeResetArchive( + tmpDir: string, + sessionId: string, + timestamp: string, + lines: unknown[], +): string { + const archivePath = path.join(tmpDir, `${sessionId}.jsonl.reset.${timestamp}`); + fs.writeFileSync(archivePath, lines.map((line) => JSON.stringify(line)).join("\n"), "utf-8"); + return archivePath; +} + function appendBlockedUserMessageWithSessionManager(params: { sessionFile: string; originalText?: string; @@ -782,6 +793,277 @@ describe("readSessionMessages", () => { } }); + test("falls back to the latest reset archive when the active transcript is missing", async () => { + const sessionId = "test-session-reset-archive-fallback"; + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-33.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "older archive" } }, + ]); + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "user", content: "restored prompt" } }, + { message: { role: "assistant", content: "restored archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, undefined, { + mode: "full", + reason: "test reset archive fallback", + allowResetArchiveFallback: true, + }); + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "restored prompt", + "restored archive", + ]); + await expect(readSessionMessageCountAsync(sessionId, storePath)).resolves.toBe(0); + + const recent = await readRecentSessionMessagesWithStatsAsync(sessionId, storePath, undefined, { + maxMessages: 1, + maxBytes: 2048, + allowResetArchiveFallback: true, + }); + expect(recent.totalMessages).toBe(2); + expect(recent.messages).toHaveLength(1); + expectMessageFields(recent.messages[0], { + role: "assistant", + content: "restored archive", + openclaw: { seq: 2 }, + }); + }); + + test("uses the active transcript if it appears during reset archive discovery", async () => { + const sessionId = "test-session-reset-archive-active-race"; + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "stale archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const originalReaddir = fs.promises.readdir.bind(fs.promises) as typeof fs.promises.readdir; + let wroteActiveTranscript = false; + const readdirSpy = vi.spyOn(fs.promises, "readdir").mockImplementation((async ( + ...args: unknown[] + ) => { + const result = await (originalReaddir as (...readdirArgs: unknown[]) => Promise)( + ...args, + ); + if (!wroteActiveTranscript) { + wroteActiveTranscript = true; + writeTranscript(tmpDir, sessionId, [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "active transcript" } }, + ]); + clearSessionTranscriptIndexCache(); + } + return result; + }) as typeof fs.promises.readdir); + + try { + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, undefined, { + mode: "full", + reason: "test active transcript race", + allowResetArchiveFallback: true, + }); + + expect(readdirSpy).toHaveBeenCalled(); + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "active transcript", + ]); + } finally { + readdirSpy.mockRestore(); + } + }); + + test("caches reset archive discovery for repeated missing-active reads", async () => { + const sessionId = "test-session-reset-archive-cache"; + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "cached archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const readdirSpy = vi.spyOn(fs.promises, "readdir"); + try { + const firstMessages = await readSessionMessagesAsync(sessionId, storePath, undefined, { + mode: "full", + reason: "test first cached archive read", + allowResetArchiveFallback: true, + }); + const readdirCallsAfterFirstRead = readdirSpy.mock.calls.length; + + const secondMessages = await readSessionMessagesAsync(sessionId, storePath, undefined, { + mode: "full", + reason: "test second cached archive read", + allowResetArchiveFallback: true, + }); + + expect(readdirCallsAfterFirstRead).toBeGreaterThan(0); + expect(readdirSpy.mock.calls).toHaveLength(readdirCallsAfterFirstRead); + expect(firstMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "cached archive", + ]); + expect(secondMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "cached archive", + ]); + } finally { + readdirSpy.mockRestore(); + } + }); + + test("chooses the newest reset archive across candidate roots", async () => { + const sessionId = "test-session-reset-archive-cross-root"; + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-33.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "older store archive" } }, + ]); + const legacySessionsDir = path.join(tmpDir, ".openclaw", "sessions"); + fs.mkdirSync(legacySessionsDir, { recursive: true }); + writeResetArchive(legacySessionsDir, sessionId, "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "newer legacy archive" } }, + ]); + clearSessionTranscriptIndexCache(); + vi.stubEnv("OPENCLAW_HOME", tmpDir); + try { + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, undefined, { + mode: "full", + reason: "test cross-root reset archive fallback", + allowResetArchiveFallback: true, + }); + + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "newer legacy archive", + ]); + } finally { + vi.unstubAllEnvs(); + } + }); + + test("does not use stale generated session archives for reset archive fallback", async () => { + const sessionId = "00000000-0000-4000-8000-000000000001"; + const staleSessionId = "00000000-0000-4000-8000-000000000002"; + const staleSessionFile = path.join(tmpDir, `${staleSessionId}.jsonl`); + writeResetArchive(tmpDir, staleSessionId, "2026-02-16T22-26-35.000Z", [ + { type: "session", version: 1, id: staleSessionId }, + { message: { role: "assistant", content: "wrong stale archive" } }, + ]); + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "current archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, staleSessionFile, { + mode: "full", + reason: "test stale archive fallback rejection", + allowResetArchiveFallback: true, + }); + + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "current archive", + ]); + }); + + test("accepts stale generated session archives when the header matches the current session", async () => { + const sessionId = "00000000-0000-4000-8000-000000000006"; + const staleSessionId = "00000000-0000-4000-8000-000000000007"; + const staleSessionFile = `${staleSessionId}.jsonl`; + writeResetArchive(tmpDir, staleSessionId, "2026-02-16T22-26-35.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "valid stale-name archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, staleSessionFile, { + mode: "full", + reason: "test stale generated archive header recovery", + allowResetArchiveFallback: true, + }); + + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "valid stale-name archive", + ]); + }); + + test("preserves explicit transcript variant priority for reset archive fallback", async () => { + const sessionId = "00000000-0000-4000-8000-000000000003"; + const topicSessionFile = "custom-topic-alpha.jsonl"; + writeResetArchive(tmpDir, sessionId, "2026-02-16T22-26-35.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "newer canonical archive" } }, + ]); + writeResetArchive(tmpDir, "custom-topic-alpha", "2026-02-16T22-26-34.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "preferred topic archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, topicSessionFile, { + mode: "full", + reason: "test explicit archive variant priority", + allowResetArchiveFallback: true, + }); + + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "preferred topic archive", + ]); + }); + + test("rejects custom reset archives from a previous session id", async () => { + const sessionId = "00000000-0000-4000-8000-000000000004"; + const previousSessionId = "00000000-0000-4000-8000-000000000005"; + const sessionFile = "shared-topic.jsonl"; + writeResetArchive(tmpDir, "shared-topic", "2026-02-16T22-26-36.000Z", [ + { type: "session", version: 1, id: previousSessionId }, + { message: { role: "assistant", content: "previous session archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, sessionFile, { + mode: "full", + reason: "test previous custom archive rejection", + allowResetArchiveFallback: true, + }); + expect(fullMessages).toEqual([]); + + const recent = await readRecentSessionMessagesWithStatsAsync( + sessionId, + storePath, + sessionFile, + { + maxMessages: 1, + maxBytes: 2048, + allowResetArchiveFallback: true, + }, + ); + expect(recent).toEqual({ messages: [], totalMessages: 0 }); + }); + + test("uses the newest custom reset archive whose header matches the session", async () => { + const sessionId = "00000000-0000-4000-8000-000000000008"; + const previousSessionId = "00000000-0000-4000-8000-000000000009"; + const sessionFile = "shared-topic-valid-latest.jsonl"; + writeResetArchive(tmpDir, "shared-topic-valid-latest", "2026-02-16T22-26-35.000Z", [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "assistant", content: "older valid archive" } }, + ]); + writeResetArchive(tmpDir, "shared-topic-valid-latest", "2026-02-16T22-26-36.000Z", [ + { type: "session", version: 1, id: previousSessionId }, + { message: { role: "assistant", content: "newer invalid archive" } }, + ]); + clearSessionTranscriptIndexCache(); + + const fullMessages = await readSessionMessagesAsync(sessionId, storePath, sessionFile, { + mode: "full", + reason: "test newest valid custom archive", + allowResetArchiveFallback: true, + }); + + expect(fullMessages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "older valid archive", + ]); + }); + test("keeps async active branch rows when imported parent links are incomplete", async () => { const sessionId = "test-session-tree-async-incomplete-parent"; writeTranscript(tmpDir, sessionId, [ diff --git a/src/gateway/session-utils.fs.ts b/src/gateway/session-utils.fs.ts index f7f0b16b62b5..04cd6afd6642 100644 --- a/src/gateway/session-utils.fs.ts +++ b/src/gateway/session-utils.fs.ts @@ -16,7 +16,10 @@ import { estimateStringChars, estimateTokensFromChars } from "../utils/cjk-chars import { stripInlineDirectiveTagsForDisplay } from "../utils/directive-tags.js"; import { extractToolCallNames, hasToolCall } from "../utils/transcript-tools.js"; import { stripEnvelope } from "./chat-sanitize.js"; -import { resolveSessionTranscriptCandidates } from "./session-transcript-files.fs.js"; +import { + resolveSessionTranscriptCandidates, + resolveSessionTranscriptResetArchiveCandidatesAsync, +} from "./session-transcript-files.fs.js"; import { readSessionTranscriptIndex, type IndexedTranscriptEntry, @@ -153,9 +156,7 @@ export function readSessionMessages( storePath: string | undefined, sessionFile?: string, ): unknown[] { - const candidates = resolveSessionTranscriptCandidates(sessionId, storePath, sessionFile); - - const filePath = candidates.find((p) => fs.existsSync(p)); + const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile); if (!filePath) { return []; } @@ -167,12 +168,14 @@ export type ReadRecentSessionMessagesOptions = { maxMessages: number; maxBytes?: number; maxLines?: number; + allowResetArchiveFallback?: boolean; }; export type ReadSessionMessagesAsyncOptions = | { mode: "full"; reason: string; + allowResetArchiveFallback?: boolean; } | ({ mode: "recent"; @@ -181,6 +184,12 @@ export type ReadSessionMessagesAsyncOptions = type ReadRecentSessionMessagesResult = { messages: unknown[]; totalMessages: number; + transcriptPath?: string; +}; + +type ReadSessionMessagesResult = { + messages: unknown[]; + transcriptPath?: string; }; const RECENT_SESSION_MESSAGES_DEFAULT_MAX_BYTES = 8 * 1024 * 1024; @@ -582,16 +591,38 @@ export async function readSessionMessagesAsync( sessionFile: string | undefined, opts: ReadSessionMessagesAsyncOptions, ): Promise { + const result = await readSessionMessagesWithSourceAsync(sessionId, storePath, sessionFile, opts); + return result.messages; +} + +export async function readSessionMessagesWithSourceAsync( + sessionId: string, + storePath: string | undefined, + sessionFile: string | undefined, + opts: ReadSessionMessagesAsyncOptions, +): Promise { if (opts.mode === "recent") { const { mode: _modeValue, ...recentOpts } = opts; - return await readRecentSessionMessagesAsync(sessionId, storePath, sessionFile, recentOpts); + const result = await readRecentSessionMessagesWithSourceAsync( + sessionId, + storePath, + sessionFile, + recentOpts, + ); + return result; } - const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile); + const filePath = + opts.allowResetArchiveFallback === true + ? await findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile) + : findExistingTranscriptPath(sessionId, storePath, sessionFile); if (!filePath) { - return []; + return { messages: [] }; } const index = await readSessionTranscriptIndex(filePath); - return index?.entries.flatMap((entry) => indexedTranscriptEntryToMessages(entry)) ?? []; + return { + messages: index?.entries.flatMap((entry) => indexedTranscriptEntryToMessages(entry)) ?? [], + transcriptPath: filePath, + }; } export async function readSessionMessageByIdAsync( @@ -599,8 +630,12 @@ export async function readSessionMessageByIdAsync( storePath: string | undefined, sessionFile: string | undefined, messageId: string, + opts?: { allowResetArchiveFallback?: boolean }, ): Promise<{ message?: unknown; seq?: number; oversized: boolean; found: boolean }> { - const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile); + const filePath = + opts?.allowResetArchiveFallback === true + ? await findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile) + : findExistingTranscriptPath(sessionId, storePath, sessionFile); if (!filePath) { return { oversized: false, found: false }; } @@ -691,16 +726,45 @@ export async function readRecentSessionMessagesAsync( sessionFile?: string, opts?: ReadRecentSessionMessagesOptions, ): Promise { + const result = await readRecentSessionMessagesWithSourceAsync( + sessionId, + storePath, + sessionFile, + opts, + ); + return result.messages; +} + +async function readRecentSessionMessagesWithSourceAsync( + sessionId: string, + storePath: string | undefined, + sessionFile?: string, + opts?: ReadRecentSessionMessagesOptions, +): Promise { const normalized = normalizeRecentSessionReadOptions(opts); const { maxMessages } = normalized; if (maxMessages === 0) { - return []; + return { messages: [] }; } - const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile); + const filePath = + opts?.allowResetArchiveFallback === true + ? await findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile) + : findExistingTranscriptPath(sessionId, storePath, sessionFile); if (!filePath) { - return []; + return { messages: [] }; } + return { + messages: await readRecentSessionMessagesFromPathAsync(filePath, normalized), + transcriptPath: filePath, + }; +} + +async function readRecentSessionMessagesFromPathAsync( + filePath: string, + opts: ReturnType, +): Promise { + const { maxMessages } = opts; let stat: fs.Stats; try { @@ -712,7 +776,7 @@ export async function readRecentSessionMessagesAsync( return []; } const lines = await readRecentTranscriptTailLinesAsync(filePath, stat, { - ...normalized, + ...opts, }); return parseRecentTranscriptTailMessages(lines, maxMessages); } @@ -723,13 +787,23 @@ export async function readRecentSessionMessagesWithStatsAsync( sessionFile: string | undefined, opts: ReadRecentSessionMessagesOptions, ): Promise { - const totalMessages = await readSessionMessageCountAsync(sessionId, storePath, sessionFile); - const messages = await readRecentSessionMessagesAsync(sessionId, storePath, sessionFile, opts); + const filePath = + opts.allowResetArchiveFallback === true + ? await findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile) + : findExistingTranscriptPath(sessionId, storePath, sessionFile); + if (!filePath) { + return { messages: [], totalMessages: 0 }; + } + const totalMessages = await readSessionMessageCountFromPathAsync(filePath); + const messages = await readRecentSessionMessagesFromPathAsync( + filePath, + normalizeRecentSessionReadOptions(opts), + ); const firstSeq = Math.max(1, totalMessages - messages.length + 1); const messagesWithSeq = messages.map((message, index) => attachOpenClawTranscriptMeta(message, { seq: firstSeq + index }), ); - return { messages: messagesWithSeq, totalMessages }; + return { messages: messagesWithSeq, totalMessages, transcriptPath: filePath }; } export function readRecentSessionTranscriptLines(params: { @@ -820,6 +894,7 @@ export { archiveSessionTranscripts, cleanupArchivedSessionTranscripts, resolveSessionTranscriptCandidates, + resolveSessionTranscriptResetArchiveCandidatesAsync, } from "./session-transcript-files.fs.js"; export function capArrayByJsonBytes( @@ -855,8 +930,7 @@ export function readSessionTitleFieldsFromTranscript( agentId?: string, opts?: { includeInterSession?: boolean }, ): SessionTitleFields { - const candidates = resolveSessionTranscriptCandidates(sessionId, storePath, sessionFile, agentId); - const filePath = candidates.find((p) => fs.existsSync(p)); + const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile, agentId); if (!filePath) { return { firstUserMessage: null, lastMessagePreview: null }; } @@ -927,8 +1001,7 @@ export async function readSessionTitleFieldsFromTranscriptAsync( agentId?: string, opts?: { includeInterSession?: boolean }, ): Promise { - const candidates = resolveSessionTranscriptCandidates(sessionId, storePath, sessionFile, agentId); - const filePath = candidates.find((p) => fs.existsSync(p)); + const filePath = findExistingTranscriptPath(sessionId, storePath, sessionFile, agentId); if (!filePath) { return { firstUserMessage: null, lastMessagePreview: null }; } @@ -1068,6 +1141,69 @@ function findExistingTranscriptPath( return candidates.find((p) => fs.existsSync(p)) ?? null; } +async function findExistingTranscriptHistoryPathAsync( + sessionId: string, + storePath: string | undefined, + sessionFile?: string, + agentId?: string, +): Promise { + const activePath = findExistingTranscriptPath(sessionId, storePath, sessionFile, agentId); + if (activePath) { + return activePath; + } + for (const archivePath of await resolveSessionTranscriptResetArchiveCandidatesAsync( + sessionId, + storePath, + sessionFile, + agentId, + )) { + const stat = await fs.promises.stat(archivePath).catch(() => null); + if (stat?.isFile()) { + const refreshedActivePath = findExistingTranscriptPath( + sessionId, + storePath, + sessionFile, + agentId, + ); + if (refreshedActivePath) { + return refreshedActivePath; + } + return archivePath; + } + } + return null; +} + +export async function resolveSessionHistoryTranscriptPathAsync( + sessionId: string, + storePath: string | undefined, + sessionFile?: string, + opts?: { agentId?: string; allowResetArchiveFallback?: boolean }, +): Promise { + return opts?.allowResetArchiveFallback === true + ? findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile, opts.agentId) + : findExistingTranscriptPath(sessionId, storePath, sessionFile, opts?.agentId); +} + +async function readSessionMessageCountFromPathAsync(filePath: string): Promise { + let stat: fs.Stats | null = null; + try { + stat = await fs.promises.stat(filePath); + const cached = getCachedTranscriptMessageCount(filePath, stat); + if (typeof cached === "number") { + return cached; + } + } catch { + // Count from the transcript index below when stat metadata is unavailable. + } + const index = await readSessionTranscriptIndex(filePath); + const count = index?.entries.length ?? 0; + if (stat) { + setCachedTranscriptMessageCount(filePath, stat, count); + } + return count; +} + function withOpenTranscriptFd(filePath: string, read: (fd: number) => T | null): T | null { let fd: number | null = null; try { diff --git a/src/gateway/session-utils.ts b/src/gateway/session-utils.ts index 85592669614e..827ee64f7f98 100644 --- a/src/gateway/session-utils.ts +++ b/src/gateway/session-utils.ts @@ -129,6 +129,8 @@ export { readSessionTitleFieldsFromTranscriptAsync, readSessionPreviewItemsFromTranscript, readSessionMessagesAsync, + readSessionMessagesWithSourceAsync, + resolveSessionHistoryTranscriptPathAsync, visitSessionMessagesAsync, resolveSessionTranscriptCandidates, } from "./session-utils.fs.js"; diff --git a/src/gateway/sessions-history-http.revocation.test.ts b/src/gateway/sessions-history-http.revocation.test.ts index 8eeb0f0f5a78..545b5be504f8 100644 --- a/src/gateway/sessions-history-http.revocation.test.ts +++ b/src/gateway/sessions-history-http.revocation.test.ts @@ -94,6 +94,7 @@ vi.mock("./session-utils.js", () => ({ sessionFile: "/tmp/session-1.jsonl", }), readSessionMessagesAsync: async () => [], + readSessionMessagesWithSourceAsync: async () => ({ messages: [] }), resolveSessionTranscriptCandidates: () => ["/tmp/session-1.jsonl"], })); @@ -109,6 +110,7 @@ vi.mock("./session-history-state.js", () => ({ messageSeq: 1, messageId, }), + shouldRefreshForTranscriptPath: () => false, refreshAsync: async () => ({ items: [], nextCursor: null, messages: [] }), }), }, diff --git a/src/gateway/sessions-history-http.test.ts b/src/gateway/sessions-history-http.test.ts index 42fde2dd2b41..788e0d4ec1fd 100644 --- a/src/gateway/sessions-history-http.test.ts +++ b/src/gateway/sessions-history-http.test.ts @@ -67,6 +67,26 @@ async function seedSession(params?: { text?: string }) { return { storePath }; } +async function writeResetArchiveTranscript(params: { + dir: string; + sessionId: string; + timestamp: string; + texts: string[]; +}) { + await fs.writeFile( + path.join(params.dir, `${params.sessionId}.jsonl.reset.${params.timestamp}`), + [ + JSON.stringify({ type: "session", version: 1, id: params.sessionId }), + ...params.texts.map((text) => + JSON.stringify({ + message: { role: "assistant", content: [{ type: "text", text }] }, + }), + ), + ].join("\n"), + "utf-8", + ); +} + function makeTranscriptAssistantMessage(params: { text: string; content?: AssistantMessage["content"]; @@ -369,6 +389,90 @@ describe("session history HTTP endpoints", () => { }); }); + test("returns session history from the latest reset archive when the active transcript is missing", async () => { + const storePath = await createSessionStoreFile(); + const sessionId = "sess-reset-main"; + const dir = path.dirname(storePath); + await writeResetArchiveTranscript({ + dir, + sessionId, + timestamp: "2026-02-16T22-26-33.000Z", + texts: ["older archived history"], + }); + await writeResetArchiveTranscript({ + dir, + sessionId, + timestamp: "2026-02-16T22-26-34.000Z", + texts: ["restored first", "restored latest"], + }); + await writeSessionStoreForTestAsync(storePath, { + "agent:main:main": { + sessionId, + updatedAt: 1, + }, + }); + + await withGatewayHarness(async (harness) => { + const body = await readSessionHistoryBody(harness.port, "agent:main:main", { + query: "?limit=1", + }); + expect(body.sessionKey).toBe("agent:main:main"); + expect(body.messages?.map((message) => message.content?.[0]?.text)).toEqual([ + "restored latest", + ]); + expect(body.hasMore).toBe(true); + expect(body.nextCursor).toBe("2"); + expectOpenClawMetadata(body.messages?.[0]?.["__openclaw"], { + seq: 2, + }); + }); + }); + + test("refreshes unbounded SSE when an active transcript replaces reset archive history", async () => { + const storePath = await createSessionStoreFile(); + const sessionId = "sess-reset-sse-takeover"; + const dir = path.dirname(storePath); + await writeResetArchiveTranscript({ + dir, + sessionId, + timestamp: "2026-02-16T22-26-34.000Z", + texts: ["archived before reset"], + }); + await writeSessionStoreForTestAsync(storePath, { + "agent:main:main": { + sessionId, + updatedAt: 1, + }, + }); + + await withGatewayHarness(async (harness) => { + const stream = await openSessionHistorySse(harness.port, "agent:main:main"); + try { + await expectHistoryEventTexts(stream, ["archived before reset"]); + + const activeTranscriptPath = path.join(dir, `${sessionId}.jsonl`); + const activeMessage = makeTranscriptAssistantMessage({ text: "active after reset" }); + await fs.writeFile( + activeTranscriptPath, + [ + JSON.stringify({ type: "session", version: 1, id: sessionId }), + JSON.stringify({ message: activeMessage }), + ].join("\n"), + "utf-8", + ); + emitSessionTranscriptUpdate({ + sessionFile: activeTranscriptPath, + sessionKey: "agent:main:main", + message: activeMessage, + }); + + await expectHistoryEventTexts(stream, ["active after reset"]); + } finally { + await stream.reader.cancel(); + } + }); + }); + test("matches direct REST history paths without trusting malformed Host headers", async () => { await seedSession({ text: "history with bad host" }); await withGatewayHarness(async (harness) => { diff --git a/src/gateway/sessions-history-http.ts b/src/gateway/sessions-history-http.ts index 79fecad8db7f..04d4a9081677 100644 --- a/src/gateway/sessions-history-http.ts +++ b/src/gateway/sessions-history-http.ts @@ -33,7 +33,7 @@ import { import { resolveTranscriptPathForComparison } from "./session-transcript-path.js"; import { readRecentSessionMessagesWithStatsAsync, - readSessionMessagesAsync, + readSessionMessagesWithSourceAsync, resolveFreshestSessionEntryFromStoreKeys, resolveGatewaySessionStoreTarget, resolveSessionTranscriptCandidates, @@ -148,19 +148,28 @@ export async function handleSessionHistoryHttpRequest( entry.sessionId, target.storePath, entry.sessionFile, - resolveSessionHistoryTailReadOptions(limit), + { + ...resolveSessionHistoryTailReadOptions(limit), + allowResetArchiveFallback: true, + }, ) : undefined; // Cursor reads still need an arbitrary historical window. The common first // page path is bounded above so `limit=1` cannot materialize huge transcripts. - const rawSnapshot = - boundedSnapshot?.messages ?? - (entry?.sessionId - ? await readSessionMessagesAsync(entry.sessionId, target.storePath, entry.sessionFile, { - mode: "full", - reason: "session history cursor pagination", - }) - : []); + const fullSnapshot = + boundedSnapshot === undefined && entry?.sessionId + ? await readSessionMessagesWithSourceAsync( + entry.sessionId, + target.storePath, + entry.sessionFile, + { + mode: "full", + reason: "session history cursor pagination", + allowResetArchiveFallback: true, + }, + ) + : undefined; + const rawSnapshot = boundedSnapshot?.messages ?? fullSnapshot?.messages ?? []; const historySnapshot = buildSessionHistorySnapshot({ rawMessages: rawSnapshot, maxChars: effectiveMaxChars, @@ -202,6 +211,7 @@ export async function handleSessionHistoryHttpRequest( rawMessages: rawSnapshot, rawTranscriptSeq: boundedSnapshot?.totalMessages, totalRawMessages: boundedSnapshot?.totalMessages, + transcriptPath: boundedSnapshot?.transcriptPath ?? fullSnapshot?.transcriptPath, maxChars: effectiveMaxChars, limit, cursor, @@ -309,6 +319,14 @@ export async function handleSessionHistoryHttpRequest( } if (update.message !== undefined) { if (limit === undefined && cursor === undefined) { + if (sseState.shouldRefreshForTranscriptPath(updatePath)) { + sentHistory = await sseState.refreshAsync(); + sseWrite(res, "history", { + sessionKey: target.canonicalKey, + ...sentHistory, + }); + return; + } const nextEvent = sseState.appendInlineMessage({ message: update.message, messageId: update.messageId, diff --git a/src/tui/embedded-backend.test.ts b/src/tui/embedded-backend.test.ts index 8e6e563745fe..e044378a78d0 100644 --- a/src/tui/embedded-backend.test.ts +++ b/src/tui/embedded-backend.test.ts @@ -39,6 +39,14 @@ const getRuntimeConfigMock = vi.fn(() => ({})); const loadGatewayModelCatalogMock = vi.fn( (_params?: unknown): Array<{ id: string; name: string; provider: string }> => [], ); +const readSessionMessagesAsyncMock = vi.fn( + async ( + _sessionId?: string, + _storePath?: string, + _sessionFile?: string, + _opts?: unknown, + ): Promise => [], +); type LoadSessionEntryMockResult = { cfg: Record; canonicalKey: string; @@ -168,7 +176,8 @@ vi.mock("../gateway/session-utils.js", () => ({ loadSessionEntry: (sessionKey: string, opts?: { agentId?: string }) => loadSessionEntryMock(sessionKey, opts), migrateAndPruneGatewaySessionStoreKey: ({ key }: { key: string }) => ({ primaryKey: key }), - readSessionMessagesAsync: async () => [], + readSessionMessagesAsync: (...args: Parameters) => + readSessionMessagesAsyncMock(...args), resolveGatewaySessionStoreTarget: ({ key }: { key: string }) => ({ canonicalKey: key, storePath: "/tmp/openclaw-sessions.json", @@ -264,6 +273,8 @@ describe("EmbeddedTuiBackend", () => { getRuntimeConfigMock.mockReturnValue({}); loadGatewayModelCatalogMock.mockReset(); loadGatewayModelCatalogMock.mockReturnValue([]); + readSessionMessagesAsyncMock.mockReset(); + readSessionMessagesAsyncMock.mockResolvedValue([]); loadSessionEntryMock.mockReset(); loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ cfg: {}, @@ -625,6 +636,32 @@ describe("EmbeddedTuiBackend", () => { expect(loadSessionEntryMock).toHaveBeenCalledWith("global", { agentId: "work" }); }); + it("uses reset-archive fallback for embedded TUI history reads", async () => { + loadSessionEntryMock.mockReturnValue({ + cfg: {}, + canonicalKey: "agent:main:main", + storePath: "/tmp/openclaw-sessions.json", + entry: { sessionId: "sess-main" }, + }); + + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const backend = new EmbeddedTuiBackend(); + + await backend.loadHistory({ sessionKey: "agent:main:main" }); + + expect(readSessionMessagesAsyncMock).toHaveBeenCalledWith( + "sess-main", + "/tmp/openclaw-sessions.json", + undefined, + { + mode: "recent", + maxMessages: 200, + maxBytes: 1024 * 1024, + allowResetArchiveFallback: true, + }, + ); + }); + it("loads runtime plugins for the send-path workspace before returning embedded history", async () => { const cfg = { agents: { list: [{ id: "main" }] } }; loadSessionEntryMock.mockReturnValue({ diff --git a/src/tui/embedded-backend.ts b/src/tui/embedded-backend.ts index c2174789a592..7dbf8a3998e8 100644 --- a/src/tui/embedded-backend.ts +++ b/src/tui/embedded-backend.ts @@ -449,6 +449,7 @@ export class EmbeddedTuiBackend implements TuiBackend { mode: "recent", maxMessages: max, maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024), + allowResetArchiveFallback: true, }) : []; const rawMessages = augmentChatHistoryWithCliSessionImports({