From 2c48dd227734bf0d1e94ec08db014a22e20178bb Mon Sep 17 00:00:00 2001 From: charles-openclaw Date: Tue, 2 Jun 2026 11:02:09 +0000 Subject: [PATCH] fix(sessions): preserve corrupt-header transcripts Fixes #89037. Co-authored-by: Charles --- .../session-manager-init.test.ts | 116 ++++++++++++++++++ .../session-manager-init.ts | 38 ++++++ src/agents/sessions/session-manager.test.ts | 108 ++++++++++++++++ src/agents/sessions/session-manager.ts | 83 +++++++++++-- 4 files changed, 336 insertions(+), 9 deletions(-) create mode 100644 src/agents/sessions/session-manager.test.ts diff --git a/src/agents/embedded-agent-runner/session-manager-init.test.ts b/src/agents/embedded-agent-runner/session-manager-init.test.ts index 47d4bebdebb9..6175c53e864d 100644 --- a/src/agents/embedded-agent-runner/session-manager-init.test.ts +++ b/src/agents/embedded-agent-runner/session-manager-init.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; +import { SessionManager } from "../sessions/session-manager.js"; import { prepareSessionManagerForRun } from "./session-manager-init.js"; const tempPaths: string[] = []; @@ -146,4 +147,119 @@ describe("prepareSessionManagerForRun", () => { ); expect(JSON.parse(assistantLine ?? "{}")).toEqual(assistantEntry); }); + + it("does not truncate an existing transcript with a corrupted header", async () => { + const sessionFile = await makeTempFile(); + const originalTranscript = + [ + '{"type":"session","id":"broken"', + JSON.stringify({ + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-05-27T00:00:01.000Z", + message: { role: "user", content: "persisted prompt" }, + }), + ].join("\n") + "\n"; + await fs.writeFile(sessionFile, originalTranscript, "utf-8"); + const sessionManager = { + sessionId: "fresh-session", + cwd: "/srv/openclaw/main", + flushed: true, + fileEntries: [ + { + type: "session", + id: "fresh-session", + cwd: "/srv/openclaw/main", + }, + { + type: "message", + message: { role: "user" }, + }, + ], + byId: new Map([["user-1", {}]]), + labelsById: new Map(), + leafId: "user-1", + }; + + await expect( + prepareSessionManagerForRun({ + sessionManager, + sessionFile, + hadSessionFile: true, + sessionId: "new-session", + cwd: "/tmp/task-repo", + }), + ).rejects.toThrow("Refusing to reset session transcript with unreadable header"); + + expect(await fs.readFile(sessionFile, "utf-8")).toBe(originalTranscript); + expect(sessionManager.fileEntries).toEqual([ + { + type: "session", + id: "fresh-session", + cwd: "/srv/openclaw/main", + }, + { + type: "message", + message: { role: "user" }, + }, + ]); + expect(sessionManager.flushed).toBe(true); + }); + + it("keeps recovered user-only transcripts through open and run preparation", async () => { + const sessionFile = await makeTempFile(); + const userEntry = { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-05-27T00:00:01.000Z", + message: { role: "user", content: "persisted prompt" }, + }; + await fs.writeFile( + sessionFile, + ['{"type":"session","id":"broken"', JSON.stringify(userEntry)].join("\n") + "\n", + "utf-8", + ); + + const sessionManager = SessionManager.open(sessionFile, path.dirname(sessionFile), "/old/cwd"); + + await prepareSessionManagerForRun({ + sessionManager, + sessionFile, + hadSessionFile: true, + sessionId: "new-session", + cwd: "/tmp/task-repo", + }); + sessionManager.appendMessage({ + role: "assistant", + content: [{ type: "text", text: "response" }], + api: "messages", + provider: "anthropic", + model: "sonnet-4.6", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }); + + const entries = (await fs.readFile(sessionFile, "utf-8")) + .trim() + .split("\n") + .map( + (line) => JSON.parse(line) as { type: string; id?: string; message?: { role?: string } }, + ); + expect(entries.map((entry) => entry.type)).toEqual(["session", "message", "message"]); + expect(entries[0]).toEqual( + expect.objectContaining({ type: "session", id: "new-session", cwd: "/tmp/task-repo" }), + ); + expect(entries[1]).toEqual(userEntry); + expect(entries[2]?.message?.role).toBe("assistant"); + }); }); diff --git a/src/agents/embedded-agent-runner/session-manager-init.ts b/src/agents/embedded-agent-runner/session-manager-init.ts index c12a62f1aa35..bd937322cad1 100644 --- a/src/agents/embedded-agent-runner/session-manager-init.ts +++ b/src/agents/embedded-agent-runner/session-manager-init.ts @@ -4,6 +4,30 @@ import { serializeJsonlLine, writeJsonlLines } from "../../config/sessions/trans type SessionHeaderEntry = { type: "session"; id?: string; cwd?: string }; type SessionMessageEntry = { type: "message"; message?: { role?: string } }; +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +async function assertExistingHeaderIsReadable(sessionFile: string): Promise { + const content = await fs.readFile(sessionFile, "utf-8"); + const firstLine = content.split("\n").find((line) => line.trim()); + if (!firstLine) { + return; + } + + let parsed: unknown; + try { + parsed = JSON.parse(firstLine); + } catch (error) { + throw new Error(`Refusing to reset session transcript with unreadable header: ${sessionFile}`, { + cause: error, + }); + } + if (!isRecord(parsed) || parsed.type !== "session") { + throw new Error(`Refusing to reset session transcript with invalid header: ${sessionFile}`); + } +} + /** * session runtime SessionManager persistence quirk: * - If the file exists but has no assistant message, SessionManager marks itself `flushed=true` @@ -29,6 +53,7 @@ export async function prepareSessionManagerForRun(params: { byId?: Map; labelsById?: Map; leafId?: string | null; + wasRecoveredFromCorruptHeader?: () => boolean; }; const header = sm.fileEntries.find((e): e is SessionHeaderEntry => e.type === "session"); @@ -45,7 +70,20 @@ export async function prepareSessionManagerForRun(params: { } if (params.hadSessionFile && header && !hasAssistant) { + if (sm.wasRecoveredFromCorruptHeader?.()) { + header.id = params.sessionId; + header.cwd = params.cwd; + sm.sessionId = params.sessionId; + sm.cwd = params.cwd; + await writeJsonlLines(params.sessionFile, sm.fileEntries.map(serializeJsonlLine), { + mode: 0o600, + }); + sm.flushed = true; + return; + } + // Reset file so the first assistant flush includes header+user+assistant in order. + await assertExistingHeaderIsReadable(params.sessionFile); await fs.writeFile(params.sessionFile, "", "utf-8"); header.id = params.sessionId; header.cwd = params.cwd; diff --git a/src/agents/sessions/session-manager.test.ts b/src/agents/sessions/session-manager.test.ts new file mode 100644 index 000000000000..cb52f56bc903 --- /dev/null +++ b/src/agents/sessions/session-manager.test.ts @@ -0,0 +1,108 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { SessionManager } from "./session-manager.js"; + +const tempPaths: string[] = []; + +async function makeTempDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-manager-")); + tempPaths.push(dir); + return dir; +} + +describe("SessionManager.open", () => { + afterEach(async () => { + await Promise.all( + tempPaths.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), + ); + }); + + it("recovers a corrupted first-line header without truncating later messages", async () => { + const dir = await makeTempDir(); + const sessionFile = path.join(dir, "session.jsonl"); + const originalHeader = { + type: "session", + version: 3, + id: "original-session", + timestamp: "2026-05-27T00:00:00.000Z", + cwd: "/srv/openclaw/main", + }; + const userEntry = { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-05-27T00:00:01.000Z", + message: { role: "user", content: "important question" }, + }; + const assistantEntry = { + type: "message", + id: "assistant-1", + parentId: "user-1", + timestamp: "2026-05-27T00:00:02.000Z", + message: { role: "assistant", content: "important answer" }, + }; + const originalTranscript = + [ + JSON.stringify(originalHeader).slice(0, 30), + JSON.stringify(userEntry), + JSON.stringify(assistantEntry), + ].join("\n") + "\n"; + await fs.writeFile(sessionFile, originalTranscript, "utf8"); + if (process.platform !== "win32") { + await fs.chmod(sessionFile, 0o600); + } + + const sessionManager = SessionManager.open(sessionFile, dir, "/tmp/task-repo"); + + expect(sessionManager.getEntries()).toEqual([userEntry, assistantEntry]); + expect(await fs.readFile(sessionFile, "utf8")).toContain("important question"); + expect(await fs.readFile(sessionFile, "utf8")).toContain("important answer"); + await expect(fs.readFile(sessionFile, "utf8")).resolves.not.toBe(originalTranscript); + + const backupFiles = (await fs.readdir(dir)).filter((file) => file.includes(".corrupt-")); + expect(backupFiles).toHaveLength(1); + await expect(fs.readFile(path.join(dir, backupFiles[0] ?? ""), "utf8")).resolves.toBe( + originalTranscript, + ); + if (process.platform !== "win32") { + const backupStat = await fs.stat(path.join(dir, backupFiles[0] ?? "")); + expect(backupStat.mode & 0o777).toBe(0o600); + } + }); + + it("does not duplicate the header after recovering a header-only corrupt file", async () => { + const dir = await makeTempDir(); + const sessionFile = path.join(dir, "session.jsonl"); + await fs.writeFile(sessionFile, '{"type":"session","version":3,"id":"sess', "utf8"); + + const sessionManager = SessionManager.open(sessionFile, dir, "/tmp/task-repo"); + sessionManager.appendMessage({ role: "user", content: "hello", timestamp: Date.now() }); + sessionManager.appendMessage({ + role: "assistant", + content: [{ type: "text", text: "hi" }], + api: "messages", + provider: "anthropic", + model: "sonnet-4.6", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }); + + const entries = (await fs.readFile(sessionFile, "utf8")) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { type: string }); + + expect(entries.map((entry) => entry.type)).toEqual(["session", "message", "message"]); + expect(entries.filter((entry) => entry.type === "session")).toHaveLength(1); + }); +}); diff --git a/src/agents/sessions/session-manager.ts b/src/agents/sessions/session-manager.ts index 03ccc6008c16..e49b4ab159d6 100644 --- a/src/agents/sessions/session-manager.ts +++ b/src/agents/sessions/session-manager.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { closeSync, + chmodSync, existsSync, mkdirSync, openSync, @@ -8,11 +9,11 @@ import { readFileSync, readSync, statSync, + writeFileSync, } from "node:fs"; import { readdir, readFile, stat } from "node:fs/promises"; import { join, resolve } from "node:path"; import { - appendJsonlEntriesSync, appendJsonlEntrySync, writeJsonlEntriesSync, } from "../../config/sessions/transcript-jsonl.js"; @@ -387,6 +388,21 @@ export function loadEntriesFromFile(filePath: string): FileEntry[] { } const content = readFileSync(filePath, "utf8"); + const entries = parseJsonlEntries(content); + + // Validate session header + if (entries.length === 0) { + return entries; + } + const header = entries[0]; + if (header.type !== "session" || typeof (header as { id?: unknown }).id !== "string") { + return []; + } + + return entries; +} + +function parseJsonlEntries(content: string): FileEntry[] { const entries: FileEntry[] = []; const lines = content.trim().split("\n"); @@ -402,16 +418,46 @@ export function loadEntriesFromFile(filePath: string): FileEntry[] { } } - // Validate session header - if (entries.length === 0) { - return entries; - } + return entries; +} + +function hasReadableSessionHeader(entries: FileEntry[]): boolean { const header = entries[0]; - if (header.type !== "session" || typeof (header as { id?: unknown }).id !== "string") { - return []; + return header?.type === "session" && typeof (header as { id?: unknown }).id === "string"; +} + +function buildCorruptSessionBackupPath(filePath: string): string { + const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); + return `${filePath}.corrupt-${timestamp}-${randomUUID().slice(0, 8)}.jsonl`; +} + +function recoverCorruptSessionEntries(filePath: string, cwd: string): FileEntry[] | null { + const content = readFileSync(filePath, "utf8"); + if (content.trim().length === 0) { + return null; } - return entries; + const parsedEntries = parseJsonlEntries(content); + const recoveredHeader = parsedEntries.find( + (entry): entry is SessionHeader => + entry.type === "session" && typeof (entry as { id?: unknown }).id === "string", + ); + const header: SessionHeader = + recoveredHeader ?? + ({ + type: "session", + version: CURRENT_SESSION_VERSION, + id: createSessionId(), + timestamp: new Date().toISOString(), + cwd, + } satisfies SessionHeader); + const recoveredEntries = parsedEntries.filter((entry) => entry.type !== "session"); + + const backupPath = buildCorruptSessionBackupPath(filePath); + const backupMode = statSync(filePath).mode & 0o777; + writeFileSync(backupPath, content, { encoding: "utf8", mode: backupMode || 0o600 }); + chmodSync(backupPath, backupMode || 0o600); + return [header, ...recoveredEntries]; } function isValidSessionFile(filePath: string): boolean { @@ -695,6 +741,7 @@ export class SessionManager { private labelsById: Map = new Map(); private labelTimestampsById: Map = new Map(); private leafId: string | null = null; + private recoveredCorruptHeader = false; private constructor( cwd: string, @@ -719,12 +766,25 @@ export class SessionManager { /** Switch to a different session file (used for resume and branching) */ setSessionFile(sessionFile: string): void { this.sessionFile = resolve(sessionFile); + this.recoveredCorruptHeader = false; if (existsSync(this.sessionFile)) { this.fileEntries = loadEntriesFromFile(this.sessionFile); // If file was empty or corrupted (no valid header), truncate and start fresh // to avoid appending messages without a session header (which breaks the session) if (this.fileEntries.length === 0) { + const recoveredEntries = recoverCorruptSessionEntries(this.sessionFile, this.cwd); + if (recoveredEntries && hasReadableSessionHeader(recoveredEntries)) { + this.fileEntries = recoveredEntries; + const header = this.fileEntries.find((e) => e.type === "session"); + this.sessionId = header?.id ?? createSessionId(); + this.buildIndex(); + this.rewriteFile(); + this.recoveredCorruptHeader = true; + this.flushed = true; + return; + } + const explicitPath = this.sessionFile; this.newSession(); this.sessionFile = explicitPath; @@ -750,6 +810,7 @@ export class SessionManager { } newSession(options?: NewSessionOptions): string | undefined { + this.recoveredCorruptHeader = false; this.sessionId = options?.id ?? createSessionId(); const timestamp = new Date().toISOString(); const header: SessionHeader = { @@ -819,6 +880,10 @@ export class SessionManager { return this.sessionId; } + wasRecoveredFromCorruptHeader(): boolean { + return this.recoveredCorruptHeader; + } + getSessionFile(): string | undefined { return this.sessionFile; } @@ -838,7 +903,7 @@ export class SessionManager { } if (!this.flushed) { - appendJsonlEntriesSync(this.sessionFile, this.fileEntries); + writeJsonlEntriesSync(this.sessionFile, this.fileEntries); this.flushed = true; } else { appendJsonlEntrySync(this.sessionFile, entry);