mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-09 03:22:40 +00:00
fix(sessions): restore reset archive fallback reads
Fall back to valid reset transcript archives when active async session transcripts are missing, while keeping active transcript priority and choosing the newest valid archive across roots. Validation: - node scripts/run-vitest.mjs src/gateway/session-utils.fs.test.ts src/gateway/sessions-history-http.test.ts src/gateway/sessions-history-http.revocation.test.ts src/gateway/session-history-state.test.ts src/gateway/server.chat.gateway-server-chat-b.test.ts src/gateway/managed-image-attachments.test.ts src/agents/tools/embedded-gateway-stub.test.ts src/tui/embedded-backend.test.ts - node scripts/run-tsgo.mjs -p test/tsconfig/tsconfig.test.src.json --incremental --tsBuildInfoFile .artifacts/tsgo-cache/test-src-pr92879.tsbuildinfo - git diff --check origin/main...HEAD && git diff --check - autoreview --mode branch --base origin/main: clean Direct-landed from #92879 because the source branch has maintainer edits disabled and the landed diff needed maintainer repair before merge. Co-authored-by: Masato Hoshino <246810661+masatohoshino@users.noreply.github.com> Co-authored-by: Hu Yitao <39733381+CadanHu@users.noreply.github.com> Co-authored-by: Vincent Koc <25068+vincentkoc@users.noreply.github.com>
This commit is contained in:
openclaw-clownfish[bot]
committed by
Vincent Koc
Vincent Koc
parent
a57b4c513f
commit
b9e8e6d66e
@@ -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,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
@@ -162,6 +162,7 @@ async function handleChatHistory(params: Record<string, unknown>): Promise<{
|
||||
mode: "recent",
|
||||
maxMessages: max,
|
||||
maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024),
|
||||
allowResetArchiveFallback: true,
|
||||
},
|
||||
)
|
||||
: [];
|
||||
|
||||
@@ -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<string, string>;
|
||||
transcriptMessages?: Record<string, unknown>[];
|
||||
sessionEntry?: { sessionId: string; sessionFile?: string };
|
||||
resolvedTranscriptPath?: string | null;
|
||||
onReadTranscriptMessages?: () => Promise<void> | 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 () => {
|
||||
|
||||
@@ -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<string, SessionManagedOutgoingAttachmentIndex | null>,
|
||||
@@ -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"];
|
||||
|
||||
@@ -2442,6 +2442,7 @@ async function isChatMessageIdVisibleAfterHistoryFilters(params: {
|
||||
sessionFile: string | undefined;
|
||||
messageId: string;
|
||||
sessionStartedAt?: number;
|
||||
allowResetArchiveFallback?: boolean;
|
||||
}): Promise<boolean> {
|
||||
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" });
|
||||
|
||||
@@ -2436,6 +2436,7 @@ export const sessionsHandlers: GatewayRequestHandlers = {
|
||||
{
|
||||
maxMessages: limit,
|
||||
maxLines: limit * 20 + 20,
|
||||
allowResetArchiveFallback: true,
|
||||
},
|
||||
);
|
||||
respond(true, { messages }, undefined);
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<PaginatedSessionHistory> {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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<boolean> {
|
||||
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<ResetArchiveCandidate[] | undefined> {
|
||||
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<ResetArchiveCandidate | undefined> {
|
||||
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<string[]> {
|
||||
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
|
||||
|
||||
@@ -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<unknown>)(
|
||||
...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, [
|
||||
|
||||
@@ -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<unknown[]> {
|
||||
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<ReadSessionMessagesResult> {
|
||||
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<unknown[]> {
|
||||
const result = await readRecentSessionMessagesWithSourceAsync(
|
||||
sessionId,
|
||||
storePath,
|
||||
sessionFile,
|
||||
opts,
|
||||
);
|
||||
return result.messages;
|
||||
}
|
||||
|
||||
async function readRecentSessionMessagesWithSourceAsync(
|
||||
sessionId: string,
|
||||
storePath: string | undefined,
|
||||
sessionFile?: string,
|
||||
opts?: ReadRecentSessionMessagesOptions,
|
||||
): Promise<ReadSessionMessagesResult> {
|
||||
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<typeof normalizeRecentSessionReadOptions>,
|
||||
): Promise<unknown[]> {
|
||||
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<ReadRecentSessionMessagesResult> {
|
||||
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<T>(
|
||||
@@ -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<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 };
|
||||
}
|
||||
@@ -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<string | null> {
|
||||
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<string | null> {
|
||||
return opts?.allowResetArchiveFallback === true
|
||||
? findExistingTranscriptHistoryPathAsync(sessionId, storePath, sessionFile, opts.agentId)
|
||||
: findExistingTranscriptPath(sessionId, storePath, sessionFile, opts?.agentId);
|
||||
}
|
||||
|
||||
async function readSessionMessageCountFromPathAsync(filePath: string): Promise<number> {
|
||||
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<T>(filePath: string, read: (fd: number) => T | null): T | null {
|
||||
let fd: number | null = null;
|
||||
try {
|
||||
|
||||
@@ -129,6 +129,8 @@ export {
|
||||
readSessionTitleFieldsFromTranscriptAsync,
|
||||
readSessionPreviewItemsFromTranscript,
|
||||
readSessionMessagesAsync,
|
||||
readSessionMessagesWithSourceAsync,
|
||||
resolveSessionHistoryTranscriptPathAsync,
|
||||
visitSessionMessagesAsync,
|
||||
resolveSessionTranscriptCandidates,
|
||||
} from "./session-utils.fs.js";
|
||||
|
||||
@@ -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: [] }),
|
||||
}),
|
||||
},
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<unknown[]> => [],
|
||||
);
|
||||
type LoadSessionEntryMockResult = {
|
||||
cfg: Record<string, unknown>;
|
||||
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<typeof readSessionMessagesAsyncMock>) =>
|
||||
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({
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in New Issue
Block a user