mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix: agent stops replying when a session transcript has no header row (#115080)
* fix(sessions): stop rejecting header-less persisted transcripts as legacy Sessions whose SQLite transcript has no session header row were treated as version 1 and hard-failed every run with "Persisted legacy session transcripts require doctor/import migration before runtime use". Only an actual header now declares a legacy version. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(sessions): keep entry ids stable for header-less transcripts Rebuild the header at the current version instead of inferring v1, and route header-less transcripts that still hold legacy-shaped entries to doctor. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(sessions): repair headerless transcripts in doctor --------- Co-authored-by: Galin Iliev <galin.iliev@microsoft.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Galin Iliev
@@ -269,6 +269,13 @@ imported and archived out of the active sessions directory after successful
|
||||
import; archive-tier JSONL files remain support artifacts, not runtime
|
||||
fallbacks.
|
||||
|
||||
The regular `openclaw doctor` pass also reports canonical SQLite transcripts
|
||||
whose initial session header was never persisted. `openclaw doctor --fix`
|
||||
prepends a current header and rebuilds the transcript indexes in one
|
||||
transaction while preserving existing event IDs, parent links, row timestamps,
|
||||
and session-list recency. Headerless legacy or malformed transcripts remain
|
||||
rejected until their owning migration can validate them.
|
||||
|
||||
Modes:
|
||||
|
||||
| Mode | Behavior |
|
||||
|
||||
@@ -73,7 +73,9 @@ export class SessionManagerCore {
|
||||
entries: FileEntry[],
|
||||
): void {
|
||||
const partitioned = partitionSessionFileEntries(entries);
|
||||
if (partitioned.fileEntries.length === 0) {
|
||||
// Only a physically empty transcript may initialize lazily. Opaque persisted rows still need
|
||||
// a canonical header, or runtime would silently replace malformed history with a fresh session.
|
||||
if (partitioned.fileEntries.length === 0 && partitioned.opaqueEntries.length === 0) {
|
||||
this.persistenceTarget = target ? { ...target } : undefined;
|
||||
this.initializeSession({ id: target?.sessionId });
|
||||
this.persistenceHeaderPending = target !== undefined;
|
||||
|
||||
@@ -5,6 +5,7 @@ import { TextDecoder } from "node:util";
|
||||
import { isRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import { normalizeLoadedFileEntry, type FileEntry } from "../agents/sessions/session-manager.js";
|
||||
import type { TranscriptEvent } from "../config/sessions/session-accessor.js";
|
||||
import type { SqliteTranscriptStorageRow } from "../config/sessions/session-accessor.sqlite-read.js";
|
||||
import { resolveSqliteTargetFromSessionStorePath } from "../config/sessions/session-sqlite-target.js";
|
||||
import type { SessionStoreTarget } from "../config/sessions/targets.js";
|
||||
import type { SessionEntry } from "../config/sessions/types.js";
|
||||
@@ -432,3 +433,60 @@ export function readOnlySqliteTranscriptSnapshot(
|
||||
database?.close();
|
||||
}
|
||||
}
|
||||
|
||||
/** Reads exact row metadata for a guarded transcript replacement without opening a writer. */
|
||||
export function readOnlySqliteTranscriptStorageSnapshot(
|
||||
sqlitePath: string,
|
||||
sessionId: string,
|
||||
):
|
||||
| { ok: true; rows: SqliteTranscriptStorageRow[]; sessionKey?: string }
|
||||
| { ok: false; error: unknown } {
|
||||
if (!fs.existsSync(sqlitePath)) {
|
||||
return { ok: false, error: new Error(`SQLite database not found: ${sqlitePath}`) };
|
||||
}
|
||||
let database: DatabaseSync | undefined;
|
||||
try {
|
||||
database = openNodeSqliteDatabase(sqlitePath, { readOnly: true });
|
||||
const rows = database
|
||||
.prepare(
|
||||
"SELECT created_at, event_json, seq FROM transcript_events WHERE session_id = ? ORDER BY seq ASC",
|
||||
)
|
||||
.all(sessionId) as Array<{
|
||||
created_at?: unknown;
|
||||
event_json?: unknown;
|
||||
seq?: unknown;
|
||||
}>;
|
||||
const sessionKeyRow = database
|
||||
.prepare("SELECT session_key FROM session_windows WHERE session_id = ? LIMIT 1")
|
||||
.get(sessionId) as { session_key?: unknown } | undefined;
|
||||
const storageRows: SqliteTranscriptStorageRow[] = [];
|
||||
for (const row of rows) {
|
||||
if (
|
||||
typeof row.created_at !== "number" ||
|
||||
typeof row.event_json !== "string" ||
|
||||
typeof row.seq !== "number"
|
||||
) {
|
||||
return {
|
||||
ok: false,
|
||||
error: new Error(`Invalid transcript row metadata for session ${sessionId}`),
|
||||
};
|
||||
}
|
||||
storageRows.push({
|
||||
createdAt: row.created_at,
|
||||
eventJson: row.event_json,
|
||||
seq: row.seq,
|
||||
});
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
rows: storageRows,
|
||||
...(typeof sessionKeyRow?.session_key === "string"
|
||||
? { sessionKey: sessionKeyRow.session_key }
|
||||
: {}),
|
||||
};
|
||||
} catch (error) {
|
||||
return { ok: false, error };
|
||||
} finally {
|
||||
database?.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,363 @@
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { SessionManager } from "../agents/sessions/session-manager.js";
|
||||
import {
|
||||
loadTranscriptEventsSync,
|
||||
replaceTranscriptEventsSync,
|
||||
upsertSessionEntry,
|
||||
} from "../config/sessions/session-accessor.js";
|
||||
import { readSqliteTranscriptStorageRows } from "../config/sessions/session-accessor.sqlite-read.js";
|
||||
import { CURRENT_SESSION_VERSION } from "../config/sessions/version.js";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import {
|
||||
closeOpenClawAgentDatabasesForTest,
|
||||
openOpenClawAgentDatabase,
|
||||
runOpenClawAgentWriteTransaction,
|
||||
} from "../state/openclaw-agent-db.js";
|
||||
import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js";
|
||||
import {
|
||||
createOpenClawTestState,
|
||||
type OpenClawTestState,
|
||||
} from "../test-utils/openclaw-test-state.js";
|
||||
|
||||
const note = vi.hoisted(() => vi.fn());
|
||||
|
||||
vi.mock("../../packages/terminal-core/src/note.js", () => ({ note }));
|
||||
|
||||
import { noteSessionTranscriptHeaderHealth } from "./doctor-session-transcript-headers.js";
|
||||
|
||||
const AGENT_ID = "main";
|
||||
const SESSION_ID = "headerless-session";
|
||||
const SESSION_KEY = "agent:main:headerless-session";
|
||||
const SPAWNED_CWD = "/workspace/headerless-child";
|
||||
|
||||
describe("doctor SQLite session transcript header repair", () => {
|
||||
let state: OpenClawTestState;
|
||||
let cfg: OpenClawConfig;
|
||||
let scope: {
|
||||
agentId: string;
|
||||
env: NodeJS.ProcessEnv;
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
storePath: string;
|
||||
};
|
||||
|
||||
beforeEach(async () => {
|
||||
note.mockClear();
|
||||
state = await createOpenClawTestState({
|
||||
layout: "state-only",
|
||||
prefix: "openclaw-doctor-transcript-headers-",
|
||||
});
|
||||
cfg = {
|
||||
agents: { list: [{ id: AGENT_ID, workspace: state.workspaceDir }] },
|
||||
};
|
||||
scope = {
|
||||
agentId: AGENT_ID,
|
||||
env: state.env,
|
||||
sessionId: SESSION_ID,
|
||||
sessionKey: SESSION_KEY,
|
||||
storePath: path.join(state.sessionsDir(AGENT_ID), "sessions.json"),
|
||||
};
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await state.cleanup();
|
||||
});
|
||||
|
||||
async function seedHeaderlessTranscript(
|
||||
events: readonly unknown[],
|
||||
options: { spawnedCwd?: string } = { spawnedCwd: SPAWNED_CWD },
|
||||
): Promise<void> {
|
||||
await upsertSessionEntry(scope, {
|
||||
sessionId: SESSION_ID,
|
||||
...(options.spawnedCwd ? { spawnedCwd: options.spawnedCwd } : {}),
|
||||
updatedAt: 10,
|
||||
});
|
||||
expect(replaceTranscriptEventsSync(scope, [...events])).toBe(true);
|
||||
}
|
||||
|
||||
it("detects read-only, repairs atomically, and keeps event identities stable", async () => {
|
||||
await seedHeaderlessTranscript([
|
||||
{
|
||||
type: "model_change",
|
||||
id: "model-1",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.632Z",
|
||||
provider: "github-copilot",
|
||||
modelId: "claude-opus-4.8",
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: "model-1",
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "user", content: "Is all good?" },
|
||||
},
|
||||
{
|
||||
type: "leaf",
|
||||
id: "leaf-1",
|
||||
parentId: "user-1",
|
||||
targetId: "user-1",
|
||||
timestamp: "2026-07-15T21:23:03.699Z",
|
||||
},
|
||||
]);
|
||||
const databaseOptions = { agentId: AGENT_ID, env: state.env };
|
||||
const database = openOpenClawAgentDatabase(databaseOptions);
|
||||
const beforeRows = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
runOpenClawAgentWriteTransaction((transactionDatabase) => {
|
||||
transactionDatabase.db
|
||||
.prepare(
|
||||
"UPDATE session_windows SET transcript_updated_at = ?, transcript_observed_at = ? WHERE session_id = ?",
|
||||
)
|
||||
.run(1_000_000, 999_000, SESSION_ID);
|
||||
}, databaseOptions);
|
||||
|
||||
expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow(
|
||||
"require doctor/import migration before runtime use",
|
||||
);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: false }),
|
||||
).resolves.toEqual({ found: 1, repaired: 0 });
|
||||
expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(beforeRows);
|
||||
expect(note).toHaveBeenCalledWith(
|
||||
'- Found 1 canonical session transcript without a header.\n- Run "openclaw doctor --fix" to repair it before resuming the session.',
|
||||
"Session transcript headers",
|
||||
);
|
||||
|
||||
note.mockClear();
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 1, repaired: 1 });
|
||||
|
||||
const repairedRows = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
const repairedEvents = repairedRows.map((row) => JSON.parse(row.eventJson));
|
||||
expect(repairedEvents[0]).toMatchObject({
|
||||
type: "session",
|
||||
version: CURRENT_SESSION_VERSION,
|
||||
id: SESSION_ID,
|
||||
cwd: SPAWNED_CWD,
|
||||
timestamp: new Date(beforeRows[0]?.createdAt ?? 0).toISOString(),
|
||||
});
|
||||
expect(
|
||||
repairedEvents.slice(1).map((event) => ({
|
||||
id: event.id,
|
||||
parentId: event.parentId,
|
||||
targetId: event.targetId,
|
||||
})),
|
||||
).toEqual([
|
||||
{ id: "model-1", parentId: null, targetId: undefined },
|
||||
{ id: "user-1", parentId: "model-1", targetId: undefined },
|
||||
{ id: "leaf-1", parentId: "user-1", targetId: "user-1" },
|
||||
]);
|
||||
expect(repairedRows.slice(1).map((row) => row.createdAt)).toEqual(
|
||||
beforeRows.map((row) => row.createdAt),
|
||||
);
|
||||
expect(repairedRows.map((row) => row.seq)).toEqual([0, 1, 2, 3]);
|
||||
|
||||
const identities = database.db
|
||||
.prepare(
|
||||
"SELECT event_id, parent_id, seq FROM transcript_event_identities WHERE session_id = ? ORDER BY seq ASC",
|
||||
)
|
||||
.all(SESSION_ID);
|
||||
expect(identities).toEqual([
|
||||
{ event_id: SESSION_ID, parent_id: null, seq: 0 },
|
||||
{ event_id: "model-1", parent_id: null, seq: 1 },
|
||||
{ event_id: "user-1", parent_id: "model-1", seq: 2 },
|
||||
{ event_id: "leaf-1", parent_id: "user-1", seq: 3 },
|
||||
]);
|
||||
expect(
|
||||
database.db
|
||||
.prepare(
|
||||
"SELECT event_seq FROM session_transcript_active_events WHERE session_id = ? ORDER BY active_position ASC",
|
||||
)
|
||||
.all(SESSION_ID),
|
||||
).toEqual([{ event_seq: 1 }, { event_seq: 2 }]);
|
||||
expect(
|
||||
database.db
|
||||
.prepare("SELECT transcript_updated_at AS value FROM session_windows WHERE session_id = ?")
|
||||
.get(SESSION_ID),
|
||||
).toEqual({ value: 1_000_001 });
|
||||
const repairedManager = SessionManager.open(scope, state.workspaceDir);
|
||||
expect(repairedManager.getEntries().map((entry) => entry.id)).toEqual(["model-1", "user-1"]);
|
||||
expect(repairedManager.buildSessionContext().messages).toEqual([
|
||||
{ role: "user", content: "Is all good?" },
|
||||
]);
|
||||
expect(note).toHaveBeenCalledWith(
|
||||
"- Prepended current headers to 1 session transcript.",
|
||||
"Session transcript headers",
|
||||
);
|
||||
|
||||
note.mockClear();
|
||||
const afterFirstRepair = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 0, repaired: 0 });
|
||||
expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(afterFirstRepair);
|
||||
expect(note).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not admit legacy headerless rows into the current runtime shape", async () => {
|
||||
await seedHeaderlessTranscript([
|
||||
{ type: "message", message: { role: "user", content: "legacy message" } },
|
||||
{ type: "message", message: { role: "hookMessage", content: "legacy hook" } },
|
||||
]);
|
||||
const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env });
|
||||
const before = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 0, repaired: 0 });
|
||||
|
||||
expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before);
|
||||
expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow(
|
||||
"require doctor/import migration before runtime use",
|
||||
);
|
||||
expect(loadTranscriptEventsSync(scope)).toHaveLength(2);
|
||||
expect(note).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: "a malformed current entry",
|
||||
events: [
|
||||
{
|
||||
type: "message",
|
||||
id: "malformed-message",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "invalid", content: "not a current message" },
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
name: "an invalid leaf control",
|
||||
events: [
|
||||
{
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "user", content: "hello" },
|
||||
},
|
||||
{
|
||||
type: "leaf",
|
||||
id: "invalid-leaf",
|
||||
parentId: "user-1",
|
||||
},
|
||||
],
|
||||
},
|
||||
])("does not repair $name", async ({ events }) => {
|
||||
await seedHeaderlessTranscript(events);
|
||||
const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env });
|
||||
const before = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 0, repaired: 0 });
|
||||
|
||||
expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before);
|
||||
expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow(
|
||||
"require doctor/import migration before runtime use",
|
||||
);
|
||||
expect(note).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not repair duplicate event identities", async () => {
|
||||
await seedHeaderlessTranscript([
|
||||
{
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "user", content: "first" },
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "user-2",
|
||||
parentId: "user-1",
|
||||
timestamp: "2026-07-15T21:23:03.699Z",
|
||||
message: { role: "user", content: "second" },
|
||||
},
|
||||
]);
|
||||
const databaseOptions = { agentId: AGENT_ID, env: state.env };
|
||||
const database = openOpenClawAgentDatabase(databaseOptions);
|
||||
runOpenClawAgentWriteTransaction((transactionDatabase) => {
|
||||
const duplicate = {
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: "user-1",
|
||||
timestamp: "2026-07-15T21:23:03.699Z",
|
||||
message: { role: "user", content: "second" },
|
||||
};
|
||||
transactionDatabase.db
|
||||
.prepare("UPDATE transcript_events SET event_json = ? WHERE session_id = ? AND seq = 1")
|
||||
.run(JSON.stringify(duplicate), SESSION_ID);
|
||||
}, databaseOptions);
|
||||
const before = readSqliteTranscriptStorageRows(database, SESSION_ID);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 0, repaired: 0 });
|
||||
|
||||
expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before);
|
||||
expect(note).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("uses the configured agent workspace when the session has no spawned cwd", async () => {
|
||||
await seedHeaderlessTranscript(
|
||||
[
|
||||
{
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "user", content: "hello" },
|
||||
},
|
||||
],
|
||||
{},
|
||||
);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 1, repaired: 1 });
|
||||
|
||||
const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env });
|
||||
const header = JSON.parse(readSqliteTranscriptStorageRows(database, SESSION_ID)[0]!.eventJson);
|
||||
expect(header).toMatchObject({ type: "session", cwd: state.workspaceDir });
|
||||
});
|
||||
|
||||
it("does not borrow the spawned cwd from a newer session on the same key", async () => {
|
||||
await seedHeaderlessTranscript([
|
||||
{
|
||||
type: "message",
|
||||
id: "user-1",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-15T21:23:03.698Z",
|
||||
message: { role: "user", content: "hello" },
|
||||
},
|
||||
]);
|
||||
const databaseOptions = { agentId: AGENT_ID, env: state.env };
|
||||
runOpenClawAgentWriteTransaction((database) => {
|
||||
database.db
|
||||
.prepare(
|
||||
"UPDATE session_nodes SET current_session_id = ?, entry_json = ? WHERE session_key = ?",
|
||||
)
|
||||
.run(
|
||||
"newer-session",
|
||||
JSON.stringify({ sessionId: "newer-session", spawnedCwd: "/workspace/newer" }),
|
||||
SESSION_KEY,
|
||||
);
|
||||
}, databaseOptions);
|
||||
|
||||
await expect(
|
||||
noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }),
|
||||
).resolves.toEqual({ found: 1, repaired: 1 });
|
||||
|
||||
const database = openOpenClawAgentDatabase(databaseOptions);
|
||||
const header = JSON.parse(readSqliteTranscriptStorageRows(database, SESSION_ID)[0]!.eventJson);
|
||||
expect(header).toMatchObject({ type: "session", cwd: state.workspaceDir });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,329 @@
|
||||
import fs from "node:fs";
|
||||
import { note } from "../../packages/terminal-core/src/note.js";
|
||||
import { resolveAgentWorkspaceDir } from "../agents/agent-scope.js";
|
||||
import { isIndexedSessionEntry } from "../agents/sessions/session-manager-codec.js";
|
||||
import type { TranscriptEvent } from "../config/sessions/session-accessor.js";
|
||||
import {
|
||||
readSqliteTranscriptStorageRows,
|
||||
type SqliteTranscriptStorageRow,
|
||||
} from "../config/sessions/session-accessor.sqlite-read.js";
|
||||
import { getSessionKysely } from "../config/sessions/session-accessor.sqlite-scope.js";
|
||||
import { replaceSqliteTranscriptEventsInTransaction } from "../config/sessions/session-accessor.sqlite-transcript-store.js";
|
||||
import { resolveAllAgentSessionStoreTargetsSync } from "../config/sessions/targets.js";
|
||||
import { createSessionTranscriptHeader } from "../config/sessions/transcript-header.js";
|
||||
import {
|
||||
isCanonicalSessionTranscriptEntry,
|
||||
isSessionTranscriptLeafControl,
|
||||
} from "../config/sessions/transcript-tree.js";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import { formatErrorMessage } from "../infra/errors.js";
|
||||
import { executeSqliteQueryTakeFirstSync } from "../infra/kysely-sync.js";
|
||||
import { parseAgentSessionKey } from "../routing/session-key.js";
|
||||
import {
|
||||
runOpenClawAgentWriteTransaction,
|
||||
type OpenClawAgentDatabase,
|
||||
} from "../state/openclaw-agent-db.js";
|
||||
import {
|
||||
readOnlySqliteTranscriptSessionIds,
|
||||
readOnlySqliteTranscriptStorageSnapshot,
|
||||
resolveTargetSqlitePath,
|
||||
} from "./doctor-session-sqlite-readers.js";
|
||||
|
||||
const NOTE_TITLE = "Session transcript headers";
|
||||
|
||||
type HeaderRepairContext = {
|
||||
sessionKey: string;
|
||||
spawnedCwd?: string;
|
||||
};
|
||||
|
||||
type HeaderRepairReport = {
|
||||
found: number;
|
||||
repaired: number;
|
||||
};
|
||||
|
||||
function parseCanonicalHeaderlessEvents(
|
||||
rows: readonly SqliteTranscriptStorageRow[],
|
||||
sessionId: string,
|
||||
): TranscriptEvent[] | undefined {
|
||||
if (rows.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
const events: TranscriptEvent[] = [];
|
||||
const eventIds = new Set<string>([sessionId]);
|
||||
let indexedEntries = 0;
|
||||
for (const row of rows) {
|
||||
let event: TranscriptEvent;
|
||||
try {
|
||||
event = JSON.parse(row.eventJson) as TranscriptEvent;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
if (!event || typeof event !== "object" || Array.isArray(event)) {
|
||||
return undefined;
|
||||
}
|
||||
const record = event as Record<string, unknown>;
|
||||
if (record.type === "session") {
|
||||
return undefined;
|
||||
}
|
||||
// Known transcript entries must already satisfy the current runtime schema. Opaque plugin
|
||||
// rows remain uninterpreted, but malformed leaf controls and duplicate identities would make
|
||||
// a delete-and-reinsert repair lossy or ambiguous, so fail closed on those shapes.
|
||||
if (isCanonicalSessionTranscriptEntry(record)) {
|
||||
if (!isIndexedSessionEntry(event)) {
|
||||
return undefined;
|
||||
}
|
||||
indexedEntries += 1;
|
||||
} else if (record.type === "leaf" && !isSessionTranscriptLeafControl(record)) {
|
||||
return undefined;
|
||||
}
|
||||
if (typeof record.id === "string") {
|
||||
const eventId = record.id.trim();
|
||||
if (!eventId || eventIds.has(eventId)) {
|
||||
return undefined;
|
||||
}
|
||||
eventIds.add(eventId);
|
||||
}
|
||||
events.push(event);
|
||||
}
|
||||
return indexedEntries > 0 ? events : undefined;
|
||||
}
|
||||
|
||||
function snapshotsMatch(
|
||||
expected: readonly SqliteTranscriptStorageRow[],
|
||||
current: readonly SqliteTranscriptStorageRow[],
|
||||
): boolean {
|
||||
return (
|
||||
expected.length === current.length &&
|
||||
expected.every(
|
||||
(row, index) =>
|
||||
row.seq === current[index]?.seq &&
|
||||
row.createdAt === current[index]?.createdAt &&
|
||||
row.eventJson === current[index]?.eventJson,
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
function readHeaderRepairContext(
|
||||
database: OpenClawAgentDatabase,
|
||||
sessionId: string,
|
||||
): HeaderRepairContext | undefined {
|
||||
const db = getSessionKysely(database.db);
|
||||
const window = executeSqliteQueryTakeFirstSync(
|
||||
database.db,
|
||||
db
|
||||
.selectFrom("session_windows")
|
||||
.select("session_key")
|
||||
.where("session_id", "=", sessionId)
|
||||
.limit(1),
|
||||
);
|
||||
if (!window?.session_key) {
|
||||
return undefined;
|
||||
}
|
||||
const node = executeSqliteQueryTakeFirstSync(
|
||||
database.db,
|
||||
db
|
||||
.selectFrom("session_nodes")
|
||||
.select(["current_session_id", "entry_json"])
|
||||
.where("session_key", "=", window.session_key)
|
||||
.limit(1),
|
||||
);
|
||||
let spawnedCwd: string | undefined;
|
||||
// Historical windows can share a key; only the node's current session owns entry_json.
|
||||
if (node?.current_session_id === sessionId && node.entry_json) {
|
||||
try {
|
||||
const entry = JSON.parse(node.entry_json) as {
|
||||
sessionId?: unknown;
|
||||
spawnedCwd?: unknown;
|
||||
};
|
||||
if (
|
||||
entry.sessionId === sessionId &&
|
||||
typeof entry.spawnedCwd === "string" &&
|
||||
entry.spawnedCwd.trim()
|
||||
) {
|
||||
spawnedCwd = entry.spawnedCwd.trim();
|
||||
}
|
||||
} catch {
|
||||
// The transcript can still be repaired with the configured agent workspace.
|
||||
}
|
||||
}
|
||||
return { sessionKey: window.session_key, ...(spawnedCwd ? { spawnedCwd } : {}) };
|
||||
}
|
||||
|
||||
function formatHeaderTimestamp(createdAt: number): string | undefined {
|
||||
if (!Number.isFinite(createdAt)) {
|
||||
return undefined;
|
||||
}
|
||||
try {
|
||||
return new Date(createdAt).toISOString();
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function assertRepairPreservedEvents(params: {
|
||||
before: readonly SqliteTranscriptStorageRow[];
|
||||
database: OpenClawAgentDatabase;
|
||||
sessionId: string;
|
||||
}): void {
|
||||
const after = readSqliteTranscriptStorageRows(params.database, params.sessionId);
|
||||
if (after.length !== params.before.length + 1) {
|
||||
throw new Error(`header repair changed the event count for ${params.sessionId}`);
|
||||
}
|
||||
for (const [index, beforeRow] of params.before.entries()) {
|
||||
const afterRow = after[index + 1];
|
||||
if (!afterRow || afterRow.createdAt !== beforeRow.createdAt) {
|
||||
throw new Error(`header repair changed row timestamps for ${params.sessionId}`);
|
||||
}
|
||||
const beforeEvent = JSON.parse(beforeRow.eventJson) as Record<string, unknown>;
|
||||
const afterEvent = JSON.parse(afterRow.eventJson) as Record<string, unknown>;
|
||||
if (
|
||||
beforeEvent.id !== afterEvent.id ||
|
||||
beforeEvent.parentId !== afterEvent.parentId ||
|
||||
beforeEvent.targetId !== afterEvent.targetId ||
|
||||
beforeEvent.appendParentId !== afterEvent.appendParentId
|
||||
) {
|
||||
throw new Error(`header repair changed event identity for ${params.sessionId}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function formatCount(count: number, singular: string): string {
|
||||
return `${count} ${singular}${count === 1 ? "" : "s"}`;
|
||||
}
|
||||
|
||||
/** Reports or repairs canonical SQLite transcripts whose first header was never persisted. */
|
||||
export async function noteSessionTranscriptHeaderHealth(params: {
|
||||
cfg: OpenClawConfig;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
shouldRepair: boolean;
|
||||
}): Promise<HeaderRepairReport> {
|
||||
const env = params.env ?? process.env;
|
||||
let found = 0;
|
||||
let repaired = 0;
|
||||
|
||||
const targetsBySqlitePath = new Map<string, { agentId: string; storePath: string }>();
|
||||
for (const target of resolveAllAgentSessionStoreTargetsSync(params.cfg, { env })) {
|
||||
const sqlitePath = resolveTargetSqlitePath(target);
|
||||
if (!targetsBySqlitePath.has(sqlitePath)) {
|
||||
targetsBySqlitePath.set(sqlitePath, target);
|
||||
}
|
||||
}
|
||||
|
||||
for (const [sqlitePath, target] of targetsBySqlitePath) {
|
||||
if (!fs.existsSync(sqlitePath)) {
|
||||
continue;
|
||||
}
|
||||
const databaseOptions = { agentId: target.agentId, env, path: sqlitePath };
|
||||
try {
|
||||
for (const sessionId of readOnlySqliteTranscriptSessionIds(sqlitePath)) {
|
||||
const snapshot = readOnlySqliteTranscriptStorageSnapshot(sqlitePath, sessionId);
|
||||
if (!snapshot.ok) {
|
||||
const detail = formatErrorMessage(snapshot.error).replace(/\s+/g, " ").trim();
|
||||
note(
|
||||
`- Failed to read transcript ${sessionId} (${target.agentId}): ${detail}`,
|
||||
NOTE_TITLE,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if (!snapshot.sessionKey || !parseCanonicalHeaderlessEvents(snapshot.rows, sessionId)) {
|
||||
continue;
|
||||
}
|
||||
const headerTimestamp = formatHeaderTimestamp(snapshot.rows[0]?.createdAt ?? Number.NaN);
|
||||
if (!headerTimestamp) {
|
||||
note(
|
||||
`- Failed to repair transcript ${sessionId} (${target.agentId}): invalid first-row timestamp`,
|
||||
NOTE_TITLE,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
found += 1;
|
||||
if (!params.shouldRepair) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const logicalAgentId = parseAgentSessionKey(snapshot.sessionKey)?.agentId ?? target.agentId;
|
||||
const workspaceCwd = resolveAgentWorkspaceDir(params.cfg, logicalAgentId, env);
|
||||
try {
|
||||
runOpenClawAgentWriteTransaction(
|
||||
(database) => {
|
||||
const currentRows = readSqliteTranscriptStorageRows(database, sessionId);
|
||||
if (!snapshotsMatch(snapshot.rows, currentRows)) {
|
||||
throw new Error(
|
||||
`transcript changed while preparing header repair for ${sessionId}`,
|
||||
);
|
||||
}
|
||||
const events = parseCanonicalHeaderlessEvents(currentRows, sessionId);
|
||||
if (!events) {
|
||||
throw new Error(
|
||||
`transcript is no longer a canonical headerless session: ${sessionId}`,
|
||||
);
|
||||
}
|
||||
const context = readHeaderRepairContext(database, sessionId);
|
||||
if (!context || context.sessionKey !== snapshot.sessionKey) {
|
||||
throw new Error(
|
||||
`session binding changed while preparing header repair for ${sessionId}`,
|
||||
);
|
||||
}
|
||||
const header = createSessionTranscriptHeader({
|
||||
cwd: context.spawnedCwd ?? workspaceCwd,
|
||||
sessionId,
|
||||
timestamp: headerTimestamp,
|
||||
});
|
||||
replaceSqliteTranscriptEventsInTransaction(
|
||||
database,
|
||||
{
|
||||
agentId: target.agentId,
|
||||
env,
|
||||
path: sqlitePath,
|
||||
sessionId,
|
||||
sessionKey: context.sessionKey,
|
||||
},
|
||||
[header, ...events],
|
||||
{
|
||||
createdAtByIndex: [
|
||||
currentRows[0]?.createdAt ?? Date.parse(headerTimestamp),
|
||||
...currentRows.map((row) => row.createdAt),
|
||||
],
|
||||
preserveSessionWindowRecency: true,
|
||||
},
|
||||
);
|
||||
assertRepairPreservedEvents({ before: currentRows, database, sessionId });
|
||||
},
|
||||
databaseOptions,
|
||||
{ operationLabel: "doctor.session-transcript-headers" },
|
||||
);
|
||||
repaired += 1;
|
||||
} catch (error) {
|
||||
const detail = formatErrorMessage(error).replace(/\s+/g, " ").trim();
|
||||
note(
|
||||
`- Failed to repair transcript ${sessionId} (${target.agentId}): ${detail}`,
|
||||
NOTE_TITLE,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
const detail = formatErrorMessage(error).replace(/\s+/g, " ").trim();
|
||||
note(
|
||||
`- Failed to inspect transcript headers for ${target.agentId} (${sqlitePath}): ${detail}`,
|
||||
NOTE_TITLE,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (params.shouldRepair && repaired > 0) {
|
||||
note(
|
||||
`- Prepended current headers to ${formatCount(repaired, "session transcript")}.`,
|
||||
NOTE_TITLE,
|
||||
);
|
||||
} else if (!params.shouldRepair && found > 0) {
|
||||
note(
|
||||
[
|
||||
`- Found ${formatCount(found, "canonical session transcript")} without a header.`,
|
||||
`- Run "openclaw doctor --fix" to repair ${found === 1 ? "it" : "them"} before resuming the session.`,
|
||||
].join("\n"),
|
||||
NOTE_TITLE,
|
||||
);
|
||||
}
|
||||
|
||||
return { found, repaired };
|
||||
}
|
||||
@@ -30,6 +30,10 @@ export type SqliteTranscriptSnapshotRow = {
|
||||
seq: number;
|
||||
};
|
||||
|
||||
export type SqliteTranscriptStorageRow = SqliteTranscriptSnapshotRow & {
|
||||
createdAt: number;
|
||||
};
|
||||
|
||||
/** Loads raw transcript events from the additive SQLite transcript store. */
|
||||
export async function loadSqliteTranscriptEvents(
|
||||
scope: SessionTranscriptReadScope,
|
||||
@@ -193,6 +197,27 @@ export function readSqliteTranscriptEventRows(
|
||||
}));
|
||||
}
|
||||
|
||||
/** Reads exact transcript storage rows for guarded doctor rewrites. */
|
||||
export function readSqliteTranscriptStorageRows(
|
||||
database: OpenClawAgentDatabase,
|
||||
sessionId: string,
|
||||
): SqliteTranscriptStorageRow[] {
|
||||
const db = getSessionKysely(database.db);
|
||||
const rows = executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.selectFrom("transcript_events")
|
||||
.select(["created_at", "event_json", "seq"])
|
||||
.where("session_id", "=", sessionId)
|
||||
.orderBy("seq", "asc"),
|
||||
).rows;
|
||||
return rows.map((row) => ({
|
||||
createdAt: normalizeSqliteNumber(row.created_at),
|
||||
eventJson: row.event_json,
|
||||
seq: normalizeSqliteNumber(row.seq),
|
||||
}));
|
||||
}
|
||||
|
||||
function sqliteTranscriptJsonlByteSize() {
|
||||
return /* kysely-allow-raw: JSONL size includes event bytes plus newline separators. */ sql<number>`COALESCE(SUM(LENGTH(CAST(event_json AS BLOB))), 0)
|
||||
+ CASE WHEN COUNT(*) > 0 THEN COUNT(*) - 1 ELSE 0 END`.as("size_bytes");
|
||||
|
||||
@@ -322,19 +322,28 @@ export function replaceSqliteTranscriptEventsInTransaction(
|
||||
events: readonly TranscriptEvent[],
|
||||
options: {
|
||||
createdAtByIndex?: readonly number[];
|
||||
preserveSessionWindowTimestamps?: boolean;
|
||||
/** Keep maintenance rewrites at their existing recency while invalidating stale projections. */
|
||||
preserveSessionWindowRecency?: boolean;
|
||||
} = {},
|
||||
): void {
|
||||
const preservedTranscriptUpdatedAt =
|
||||
options.preserveSessionWindowRecency === true
|
||||
? readTranscriptMutationStateInTransaction(database, resolved.sessionId).updatedAt
|
||||
: undefined;
|
||||
const previousGeneration = readTranscriptGenerationInTransaction(database, resolved.sessionId);
|
||||
const deleted = deleteSqliteTranscriptEventsInTransaction(database, resolved.sessionId);
|
||||
if (events.length === 0) {
|
||||
if (deleted || previousGeneration) {
|
||||
rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
|
||||
touchTranscriptMutationInTransaction(database, resolved.sessionId);
|
||||
recordTranscriptReplacementMutation(
|
||||
database,
|
||||
resolved.sessionId,
|
||||
preservedTranscriptUpdatedAt,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (!deleted || options.preserveSessionWindowTimestamps !== true) {
|
||||
if (!deleted || options.preserveSessionWindowRecency !== true) {
|
||||
ensureTranscriptSessionRoot(database, resolved, readEventTimestamp(events[0]) ?? Date.now());
|
||||
}
|
||||
if (deleted || previousGeneration) {
|
||||
@@ -363,11 +372,27 @@ export function replaceSqliteTranscriptEventsInTransaction(
|
||||
}
|
||||
}
|
||||
if (deleted || seq > 0) {
|
||||
touchTranscriptMutationInTransaction(database, resolved.sessionId);
|
||||
recordTranscriptReplacementMutation(database, resolved.sessionId, preservedTranscriptUpdatedAt);
|
||||
reconcileSessionTranscriptIndexInTransaction(database.db, resolved.sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
function recordTranscriptReplacementMutation(
|
||||
database: OpenClawAgentDatabase,
|
||||
sessionId: string,
|
||||
preservedUpdatedAt: number | null | undefined,
|
||||
): void {
|
||||
if (preservedUpdatedAt === undefined || preservedUpdatedAt === null) {
|
||||
touchTranscriptMutationInTransaction(database, sessionId);
|
||||
return;
|
||||
}
|
||||
// Maintenance rewrites must invalidate in-flight projections without making an old session
|
||||
// look newly active. A one-tick advance preserves ordering while changing the snapshot key.
|
||||
advanceTranscriptMutationAtInTransaction(database, sessionId, preservedUpdatedAt, {
|
||||
strictly: true,
|
||||
});
|
||||
}
|
||||
|
||||
/** Rewrite existing transcript rows exactly, without append-time deduplication. */
|
||||
export function rewriteSqliteTranscriptEventRowsInTransaction(
|
||||
database: OpenClawAgentDatabase,
|
||||
|
||||
@@ -128,6 +128,18 @@ export async function runSessionTranscriptsHealth(ctx: DoctorHealthFlowContext):
|
||||
});
|
||||
}
|
||||
|
||||
export async function runSessionTranscriptHeadersHealth(
|
||||
ctx: DoctorHealthFlowContext,
|
||||
): Promise<void> {
|
||||
const { noteSessionTranscriptHeaderHealth } =
|
||||
await import("../commands/doctor-session-transcript-headers.js");
|
||||
await noteSessionTranscriptHeaderHealth({
|
||||
cfg: ctx.cfg,
|
||||
env: ctx.env ?? process.env,
|
||||
shouldRepair: ctx.prompter.shouldRepair,
|
||||
});
|
||||
}
|
||||
|
||||
export async function runSessionTranscriptLabelsHealth(
|
||||
ctx: DoctorHealthFlowContext,
|
||||
): Promise<void> {
|
||||
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
runSandboxHealth,
|
||||
runSessionLocksHealth,
|
||||
runSessionSnapshotsHealth,
|
||||
runSessionTranscriptHeadersHealth,
|
||||
runSessionTranscriptLabelsHealth,
|
||||
runSessionTranscriptsHealth,
|
||||
runStateIntegrityHealth,
|
||||
@@ -325,6 +326,11 @@ export function resolveInitialDoctorHealthContributions(params: {
|
||||
},
|
||||
run: runSessionTranscriptsHealth,
|
||||
}),
|
||||
createDoctorHealthContribution({
|
||||
id: "doctor:session-transcript-headers",
|
||||
label: "Session transcript headers",
|
||||
run: runSessionTranscriptHeadersHealth,
|
||||
}),
|
||||
createDoctorHealthContribution({
|
||||
id: "doctor:session-transcript-labels",
|
||||
label: "Session transcript labels",
|
||||
|
||||
Reference in New Issue
Block a user