LLM/moltbot
1
0
Fork 0
mirror of https://github.com/moltbot/moltbot.git synced 2026-05-22 06:08:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: require sqlite transcript locators at runtime

Browse Source
This commit is contained in:
Peter Steinberger
2026-05-09 08:31:48 +01:00
parent 2c892d0b89
commit bd2df4265e
6 changed files with 173 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/agents/pi-embedded-runner/compaction-successor-transcript.test.ts
View File

@@ -264,7 +264,7 @@ describe("rotateTranscriptAfterCompaction", () => {
expect(result.reason).toBe("no compaction entry");
});
it("does not create legacy jsonl successor files for unmigrated transcripts", async () => {
it("rejects filesystem transcript locators without creating successor files", async () => {
const dir = await createTmpDir();
const { manager } = createCompactedSession(dir);

8
src/agents/pi-embedded-runner/manual-compaction-boundary.ts
View File

@@ -72,15 +72,15 @@ function hasMessagesToSummarizeBeforeKeptTail(params: {
}
export async function hardenManualCompactionBoundary(params: {
sessionFile: string;
transcriptLocator: string;
preserveRecentTail?: boolean;
}): Promise<HardenedManualCompactionBoundary> {
const scope = resolveSqliteSessionTranscriptScopeForPath({
transcriptPath: params.sessionFile,
transcriptPath: params.transcriptLocator,
});
if (!scope) {
throw new Error(
`Legacy transcript has not been imported into SQLite: ${params.sessionFile}. Run "openclaw doctor --fix" to build the session database.`,
`SQLite transcript is missing from the state database: ${params.transcriptLocator}. Run "openclaw doctor --fix" if legacy transcript files still need import.`,
);
}
const events = loadSqliteSessionTranscriptEvents(scope).map((entry) => entry.event);
@@ -155,7 +155,7 @@ export async function hardenManualCompactionBoundary(params: {
});
replaceSqliteSessionTranscriptEvents({
...scope,
transcriptPath: params.sessionFile,
transcriptPath: params.transcriptLocator,
events: [header, ...replacedEntries],
});

137
src/agents/transcript/session-manager.test.ts
View File

@@ -13,14 +13,14 @@ import { openTranscriptSessionManager } from "./session-manager.js";
import { SessionManager } from "./session-transcript-contract.js";
import { replaceTranscriptStateEventsSync } from "./transcript-state.js";
async function makeTempSessionFile(name = "session.jsonl"): Promise<string> {
async function makeTempTranscriptLocator(name = "session.jsonl"): Promise<string> {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-transcript-session-"));
vi.stubEnv("OPENCLAW_STATE_DIR", dir);
return path.join(dir, name);
}
function readSessionEntries(sessionFile: string) {
const scope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: sessionFile });
function readSessionEntries(transcriptLocator: string) {
const scope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: transcriptLocator });
if (!scope) {
return [];
}
@@ -35,10 +35,10 @@ afterEach(() => {
describe("TranscriptSessionManager", () => {
it("exposes create, in-memory, list, continue, and fork through the contract value", async () => {
await makeTempSessionFile();
await makeTempTranscriptLocator();
const memory = SessionManager.inMemory("/tmp/memory-workspace");
expect(memory.isPersisted()).toBe(false);
expect(memory.getSessionFile()).toBeUndefined();
expect(memory.getTranscriptLocator()).toBeUndefined();
const memoryUserId = memory.appendMessage({
role: "user",
content: "in memory",
@@ -48,10 +48,10 @@ describe("TranscriptSessionManager", () => {
const created = SessionManager.create("/tmp/workspace");
created.appendMessage({ role: "user", content: "persist me", timestamp: 2 });
const sessionFile = created.getSessionFile();
expect(sessionFile).toBeTruthy();
if (!sessionFile) {
throw new Error("expected created session file");
const transcriptLocator = created.getTranscriptLocator();
expect(transcriptLocator).toBeTruthy();
if (!transcriptLocator) {
throw new Error("expected created transcript locator");
}
const listed = await SessionManager.list("/tmp/workspace");
@@ -60,33 +60,33 @@ describe("TranscriptSessionManager", () => {
const continued = SessionManager.continueRecent("/tmp/workspace");
expect(continued.getSessionId()).toBe(created.getSessionId());
const forked = SessionManager.forkFrom(sessionFile, "/tmp/forked-workspace");
const forked = SessionManager.forkFrom(transcriptLocator, "/tmp/forked-workspace");
expect(forked.getHeader()).toMatchObject({
cwd: "/tmp/forked-workspace",
parentSession: sessionFile,
parentSession: transcriptLocator,
});
expect(forked.buildSessionContext().messages).toMatchObject([
{ role: "user", content: "persist me" },
]);
});
it("rejects an unmigrated explicit legacy session file", async () => {
const sessionFile = await makeTempSessionFile();
it("rejects filesystem transcript locators at runtime", async () => {
const transcriptLocator = await makeTempTranscriptLocator();
expect(() =>
openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "session-1",
cwd: "/tmp/workspace",
}),
).toThrow(/Legacy transcript has not been imported into SQLite/);
).toThrow(/Transcript locator must be SQLite-backed/);
});
it("rejects runtime writes to unmigrated legacy session files", async () => {
const sessionFile = await makeTempSessionFile();
it("rejects runtime writes to filesystem transcript locators", async () => {
const transcriptLocator = await makeTempTranscriptLocator();
expect(() =>
replaceTranscriptStateEventsSync(sessionFile, [
replaceTranscriptStateEventsSync(transcriptLocator, [
{
type: "session",
version: 3,
@@ -95,30 +95,30 @@ describe("TranscriptSessionManager", () => {
cwd: "/tmp/workspace",
},
]),
).toThrow(/Legacy transcript has not been imported into SQLite/);
).toThrow(/Transcript locator must be SQLite-backed/);
});
it("opens virtual sqlite transcript locators without resolving them as filesystem paths", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "virtual-session",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "virtual-session",
cwd: "/tmp/workspace",
});
expect(sessionManager.getSessionFile()).toBe(sessionFile);
expect(sessionManager.getTranscriptLocator()).toBe(transcriptLocator);
expect(
resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: sessionFile }),
resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: transcriptLocator }),
).toMatchObject({
agentId: "main",
sessionId: "virtual-session",
});
expect(readSessionEntries(sessionFile)).toMatchObject([
expect(readSessionEntries(transcriptLocator)).toMatchObject([
{
type: "session",
id: "virtual-session",
@@ -128,20 +128,20 @@ describe("TranscriptSessionManager", () => {
});
it("uses the virtual sqlite transcript locator session id when no explicit id is supplied", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "locator-session",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
cwd: "/tmp/workspace",
});
sessionManager.appendMessage({ role: "user", content: "seed", timestamp: 1 });
expect(sessionManager.getSessionId()).toBe("locator-session");
expect(readSessionEntries(sessionFile)).toMatchObject([
expect(readSessionEntries(transcriptLocator)).toMatchObject([
{
type: "session",
id: "locator-session",
@@ -155,13 +155,13 @@ describe("TranscriptSessionManager", () => {
});
it("creates, branches, lists, and forks default sessions with virtual sqlite locators", async () => {
await makeTempSessionFile();
await makeTempTranscriptLocator();
const sessionManager = SessionManager.create("/tmp/sqlite-workspace");
const sessionFile = sessionManager.getSessionFile();
if (!sessionFile) {
throw new Error("expected session file");
const transcriptLocator = sessionManager.getTranscriptLocator();
if (!transcriptLocator) {
throw new Error("expected transcript locator");
}
expect(sessionFile).toMatch(/^sqlite-transcript:\/\/main\//);
expect(transcriptLocator).toMatch(/^sqlite-transcript:\/\/main\//);
const userId = sessionManager.appendMessage({
role: "user",
@@ -177,18 +177,18 @@ describe("TranscriptSessionManager", () => {
const listed = await SessionManager.list("/tmp/sqlite-workspace");
expect(listed.map((session) => session.id)).toContain(sessionManager.getSessionId());
const forked = SessionManager.forkFrom(sessionFile, "/tmp/sqlite-fork");
expect(forked.getSessionFile()).toMatch(/^sqlite-transcript:\/\/main\//);
const forked = SessionManager.forkFrom(transcriptLocator, "/tmp/sqlite-fork");
expect(forked.getTranscriptLocator()).toMatch(/^sqlite-transcript:\/\/main\//);
expect(forked.getHeader()).toMatchObject({
cwd: "/tmp/sqlite-fork",
parentSession: sessionFile,
parentSession: transcriptLocator,
});
});
it("allocates a fresh sqlite transcript locator when starting a new persisted session", async () => {
await makeTempSessionFile();
await makeTempTranscriptLocator();
const sessionManager = openTranscriptSessionManager({
sessionFile: createSqliteSessionTranscriptLocator({
transcriptLocator: createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "first-session",
}),
@@ -197,34 +197,34 @@ describe("TranscriptSessionManager", () => {
});
sessionManager.appendMessage({ role: "user", content: "first", timestamp: 1 });
const firstSessionFile = sessionManager.getSessionFile();
const secondSessionFile = sessionManager.newSession({ id: "second-session" });
const firstTranscriptLocator = sessionManager.getTranscriptLocator();
const secondTranscriptLocator = sessionManager.newSession({ id: "second-session" });
sessionManager.appendMessage({ role: "user", content: "second", timestamp: 2 });
expect(secondSessionFile).toBe(
expect(secondTranscriptLocator).toBe(
createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "second-session",
}),
);
expect(secondSessionFile).not.toBe(firstSessionFile);
expect(secondTranscriptLocator).not.toBe(firstTranscriptLocator);
expect(
readSessionEntries(firstSessionFile!).map((entry) => (entry as { id?: string }).id),
readSessionEntries(firstTranscriptLocator!).map((entry) => (entry as { id?: string }).id),
).toEqual(["first-session", expect.any(String)]);
expect(readSessionEntries(secondSessionFile!)).toMatchObject([
expect(readSessionEntries(secondTranscriptLocator!)).toMatchObject([
{ type: "session", id: "second-session" },
{ type: "message", message: { role: "user", content: "second" } },
]);
});
it("preserves non-main agent scope for virtual sqlite branches and forks", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "qa",
sessionId: "qa-source-session",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "qa-source-session",
cwd: "/tmp/qa-workspace",
});
@@ -242,23 +242,25 @@ describe("TranscriptSessionManager", () => {
agentId: "qa",
});
const forked = SessionManager.forkFrom(sessionFile, "/tmp/qa-fork");
expect(forked.getSessionFile()).toMatch(/^sqlite-transcript:\/\/qa\//);
const forked = SessionManager.forkFrom(transcriptLocator, "/tmp/qa-fork");
expect(forked.getTranscriptLocator()).toMatch(/^sqlite-transcript:\/\/qa\//);
expect(
resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: forked.getSessionFile()! }),
resolveSqliteSessionTranscriptScopeForPath({
transcriptPath: forked.getTranscriptLocator()!,
}),
).toMatchObject({
agentId: "qa",
});
});
it("persists initial user messages synchronously before the first assistant message", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "session-sync",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "session-sync",
cwd: "/tmp/workspace",
});
@@ -269,7 +271,7 @@ describe("TranscriptSessionManager", () => {
timestamp: 1,
});
const afterUser = readSessionEntries(sessionFile);
const afterUser = readSessionEntries(transcriptLocator);
expect(afterUser).toHaveLength(2);
expect(afterUser[1]).toMatchObject({
type: "message",
@@ -296,7 +298,7 @@ describe("TranscriptSessionManager", () => {
timestamp: 2,
});
const reopened = openTranscriptSessionManager({ sessionFile });
const reopened = openTranscriptSessionManager({ transcriptLocator });
expect(reopened.getBranch().map((entry) => entry.id)).toEqual([userId, assistantId]);
expect(reopened.buildSessionContext().messages.map((message) => message.role)).toEqual([
"user",
@@ -305,13 +307,13 @@ describe("TranscriptSessionManager", () => {
});
it("removes persisted tail entries through SQLite instead of rewriting JSONL", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "session-tail",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "session-tail",
cwd: "/tmp/workspace",
});
@@ -343,23 +345,22 @@ describe("TranscriptSessionManager", () => {
sessionManager.removeTailEntries((entry) => (entry as { id?: string }).id === assistantId),
).toBe(1);
const reopened = openTranscriptSessionManager({ sessionFile });
const reopened = openTranscriptSessionManager({ transcriptLocator });
expect(reopened.getEntry(assistantId)).toBeUndefined();
expect(reopened.getLeafId()).toBe(userId);
expect(readSessionEntries(sessionFile).map((entry) => (entry as { id?: string }).id)).toEqual([
"session-tail",
userId,
]);
expect(
readSessionEntries(transcriptLocator).map((entry) => (entry as { id?: string }).id),
).toEqual(["session-tail", userId]);
});
it("supports tree, label, name, and branch summary session APIs", async () => {
await makeTempSessionFile();
const sessionFile = createSqliteSessionTranscriptLocator({
await makeTempTranscriptLocator();
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: "main",
sessionId: "session-tree",
});
const sessionManager = openTranscriptSessionManager({
sessionFile,
transcriptLocator,
sessionId: "session-tree",
cwd: "/tmp/workspace",
});
@@ -386,7 +387,7 @@ describe("TranscriptSessionManager", () => {
children: [{ entry: { id: childId } }, { entry: { id: siblingId }, label: "alternate" }],
});
const reopened = openTranscriptSessionManager({ sessionFile });
const reopened = openTranscriptSessionManager({ transcriptLocator });
expect(reopened.getEntry(summaryId)).toMatchObject({
type: "branch_summary",
fromId: childId,

130
src/agents/transcript/session-manager.ts
View File

@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import path from "node:path";
import {
createSqliteSessionTranscriptLocator,
isSqliteSessionTranscriptLocator,
@@ -54,9 +53,14 @@ type SqliteTranscriptRecord = {
updatedAt: number;
};
function normalizeTranscriptLocator(sessionFile: string): string {
const trimmed = sessionFile.trim();
return isSqliteSessionTranscriptLocator(trimmed) ? trimmed : path.resolve(trimmed);
function normalizeTranscriptLocator(transcriptLocator: string): string {
const trimmed = transcriptLocator.trim();
if (isSqliteSessionTranscriptLocator(trimmed)) {
return trimmed;
}
throw new Error(
`Transcript locator must be SQLite-backed: ${trimmed}. Run "openclaw doctor --fix" to import legacy transcript files.`,
);
}
function createTranscriptLocator(header: SessionHeader, agentId = DEFAULT_AGENT_ID): string {
@@ -105,20 +109,16 @@ function appendTranscriptEntryToSqlite(scope: TranscriptSqliteScope, entry: Sess
});
}
function loadTranscriptState(params: { sessionFile: string; sessionId?: string; cwd?: string }): {
function loadTranscriptState(params: {
transcriptLocator: string;
sessionId?: string;
cwd?: string;
}): {
state: TranscriptState;
scope: TranscriptSqliteScope;
} {
const sessionFile = params.sessionFile.trim();
const transcriptPath = isSqliteSessionTranscriptLocator(sessionFile)
? sessionFile
: path.resolve(sessionFile);
const transcriptPath = normalizeTranscriptLocator(params.transcriptLocator);
const existingScope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath });
if (!isSqliteSessionTranscriptLocator(transcriptPath) && !existingScope) {
throw new Error(
`Legacy transcript has not been imported into SQLite: ${transcriptPath}. Run "openclaw doctor --fix" to build the session database.`,
);
}
const sessionId = existingScope?.sessionId ?? params.sessionId;
if (!sessionId) {
throw new Error(`SQLite transcript scope is missing session id for: ${transcriptPath}`);
@@ -175,7 +175,7 @@ function extractTextContent(message: { content: unknown }): string {
}
function buildSessionInfoFromState(
filePath: string,
transcriptLocator: string,
state: TranscriptState,
modifiedFallback: Date,
): SessionInfo | null {
@@ -221,7 +221,7 @@ function buildSessionInfoFromState(
}
const headerTime = Date.parse(header.timestamp);
return {
path: filePath,
path: transcriptLocator,
id: header.id,
cwd: header.cwd,
name: state.getSessionName(),
@@ -280,18 +280,18 @@ function loadTranscriptStateForRecord(record: SqliteTranscriptRecord): Transcrip
export class TranscriptSessionManager implements SessionManager {
private state: TranscriptState;
private sessionFile: string | undefined;
private transcriptLocator: string | undefined;
private persist: boolean;
private sqliteScope: TranscriptSqliteScope | undefined;
private constructor(params: {
state: TranscriptState;
sessionFile?: string;
transcriptLocator?: string;
persist: boolean;
sqliteScope?: TranscriptSqliteScope;
}) {
this.sessionFile = params.sessionFile
? normalizeTranscriptLocator(params.sessionFile)
this.transcriptLocator = params.transcriptLocator
? normalizeTranscriptLocator(params.transcriptLocator)
: undefined;
this.state = params.state;
this.persist = params.persist;
@@ -299,18 +299,18 @@ export class TranscriptSessionManager implements SessionManager {
}
static open(params: {
sessionFile: string;
transcriptLocator: string;
sessionId?: string;
cwd?: string;
}): TranscriptSessionManager {
const sessionFile = normalizeTranscriptLocator(params.sessionFile);
const transcriptLocator = normalizeTranscriptLocator(params.transcriptLocator);
const loaded = loadTranscriptState({
sessionFile,
transcriptLocator,
sessionId: params.sessionId,
cwd: params.cwd,
});
return new TranscriptSessionManager({
sessionFile,
transcriptLocator,
persist: true,
state: loaded.state,
sqliteScope: loaded.scope,
@@ -319,16 +319,16 @@ export class TranscriptSessionManager implements SessionManager {
static create(cwd: string): TranscriptSessionManager {
const header = createSessionHeader({ cwd });
const sessionFile = createTranscriptLocator(header);
const transcriptLocator = createTranscriptLocator(header);
const sqliteScope = {
agentId: resolveAgentIdFromTranscriptLocator(sessionFile),
agentId: resolveAgentIdFromTranscriptLocator(transcriptLocator),
sessionId: header.id,
transcriptPath: normalizeTranscriptLocator(sessionFile),
transcriptPath: normalizeTranscriptLocator(transcriptLocator),
};
const state = new TranscriptState({ header, entries: [] });
persistFullTranscriptStateToSqlite(sqliteScope, state);
return new TranscriptSessionManager({
sessionFile,
transcriptLocator,
persist: true,
state,
sqliteScope,
@@ -350,35 +350,35 @@ export class TranscriptSessionManager implements SessionManager {
return state.getCwd() === cwd;
});
if (newestSqlite) {
return TranscriptSessionManager.open({ sessionFile: newestSqlite.path, cwd });
return TranscriptSessionManager.open({ transcriptLocator: newestSqlite.path, cwd });
}
return TranscriptSessionManager.create(cwd);
}
static forkFrom(sourcePath: string, targetCwd: string): TranscriptSessionManager {
const sourceFile = normalizeTranscriptLocator(sourcePath);
const sourceScope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: sourceFile });
static forkFrom(sourceTranscriptLocator: string, targetCwd: string): TranscriptSessionManager {
const sourceTranscript = normalizeTranscriptLocator(sourceTranscriptLocator);
const sourceScope = resolveSqliteSessionTranscriptScopeForPath({
transcriptPath: sourceTranscript,
});
if (!sourceScope) {
throw new Error(
`Legacy transcript has not been imported into SQLite: ${sourceFile}. Run "openclaw doctor --fix" to build the session database.`,
);
throw new Error(`SQLite transcript is missing from the state database: ${sourceTranscript}`);
}
const sourceState = createTranscriptStateFromEvents(
loadSqliteSessionTranscriptEvents(sourceScope).map((entry) => entry.event),
);
const header = createSessionHeader({
cwd: targetCwd,
parentSession: sourceFile,
parentSession: sourceTranscript,
});
const sessionFile = createTranscriptLocator(header, sourceScope.agentId);
const transcriptLocator = createTranscriptLocator(header, sourceScope.agentId);
const state = new TranscriptState({ header, entries: sourceState.getEntries() });
const sqliteScope = {
agentId: sourceScope.agentId,
sessionId: header.id,
transcriptPath: normalizeTranscriptLocator(sessionFile),
transcriptPath: normalizeTranscriptLocator(transcriptLocator),
};
persistFullTranscriptStateToSqlite(sqliteScope, state);
return TranscriptSessionManager.open({ sessionFile, cwd: targetCwd });
return TranscriptSessionManager.open({ transcriptLocator, cwd: targetCwd });
}
static async list(cwd: string, onProgress?: SessionListProgress): Promise<SessionInfo[]> {
@@ -388,14 +388,14 @@ export class TranscriptSessionManager implements SessionManager {
}
static async listAll(onProgress?: SessionListProgress): Promise<SessionInfo[]> {
const files = listSqliteTranscriptRecords();
const records = listSqliteTranscriptRecords();
const sessions: SessionInfo[] = [];
let loaded = 0;
for (const file of files) {
const state = loadTranscriptStateForRecord(file);
for (const record of records) {
const state = loadTranscriptStateForRecord(record);
loaded += 1;
onProgress?.(loaded, files.length);
const info = buildSessionInfoFromState(file.path, state, new Date(file.updatedAt));
onProgress?.(loaded, records.length);
const info = buildSessionInfoFromState(record.path, state, new Date(record.updatedAt));
if (info) {
sessions.push(info);
}
@@ -403,11 +403,11 @@ export class TranscriptSessionManager implements SessionManager {
return sessions.toSorted((a, b) => b.modified.getTime() - a.modified.getTime());
}
setSessionFile(sessionFile: string): void {
this.sessionFile = normalizeTranscriptLocator(sessionFile);
setTranscriptLocator(transcriptLocator: string): void {
this.transcriptLocator = normalizeTranscriptLocator(transcriptLocator);
this.persist = true;
const loaded = loadTranscriptState({
sessionFile: this.sessionFile,
transcriptLocator: this.transcriptLocator,
cwd: this.getCwd(),
});
this.state = loaded.state;
@@ -422,15 +422,15 @@ export class TranscriptSessionManager implements SessionManager {
});
this.state = new TranscriptState({ header, entries: [] });
if (this.persist) {
this.sessionFile = createTranscriptLocator(header, this.sqliteScope?.agentId);
this.transcriptLocator = createTranscriptLocator(header, this.sqliteScope?.agentId);
this.sqliteScope = {
agentId: resolveAgentIdFromTranscriptLocator(this.sessionFile),
agentId: resolveAgentIdFromTranscriptLocator(this.transcriptLocator),
sessionId: header.id,
transcriptPath: normalizeTranscriptLocator(this.sessionFile),
transcriptPath: normalizeTranscriptLocator(this.transcriptLocator),
};
persistFullTranscriptStateToSqlite(this.sqliteScope, this.state);
}
return this.sessionFile;
return this.transcriptLocator;
}
isPersisted(): boolean {
@@ -445,8 +445,8 @@ export class TranscriptSessionManager implements SessionManager {
return this.state.getHeader()?.id ?? "";
}
getSessionFile(): string | undefined {
return this.sessionFile;
getTranscriptLocator(): string | undefined {
return this.transcriptLocator;
}
appendMessage(message: Parameters<SessionManager["appendMessage"]>[0]): string {
@@ -553,7 +553,7 @@ export class TranscriptSessionManager implements SessionManager {
options?: Parameters<SessionManager["removeTailEntries"]>[1],
): number {
const removed = this.state.removeTailEntries(shouldRemove, options);
if (removed > 0 && this.persist && this.sessionFile && this.sqliteScope) {
if (removed > 0 && this.persist && this.transcriptLocator && this.sqliteScope) {
persistFullTranscriptStateToSqlite(this.sqliteScope, this.state);
}
return removed;
@@ -577,9 +577,9 @@ export class TranscriptSessionManager implements SessionManager {
}
const header = createSessionHeader({
cwd: this.getCwd(),
parentSession: this.sessionFile,
parentSession: this.transcriptLocator,
});
const sessionFile = createSqliteSessionTranscriptLocator({
const transcriptLocator = createSqliteSessionTranscriptLocator({
agentId: this.sqliteScope?.agentId ?? DEFAULT_AGENT_ID,
sessionId: header.id,
});
@@ -592,17 +592,17 @@ export class TranscriptSessionManager implements SessionManager {
});
persistFullTranscriptStateToSqlite(
{
agentId: resolveAgentIdFromTranscriptLocator(sessionFile),
agentId: resolveAgentIdFromTranscriptLocator(transcriptLocator),
sessionId: header.id,
transcriptPath: normalizeTranscriptLocator(sessionFile),
transcriptPath: normalizeTranscriptLocator(transcriptLocator),
},
state,
);
return sessionFile;
return transcriptLocator;
}
private persistAppendedEntry(entry: SessionEntry): string {
if (!this.persist || !this.sessionFile || !this.sqliteScope) {
if (!this.persist || !this.transcriptLocator || !this.sqliteScope) {
return entry.id;
}
if (this.state.migrated) {
@@ -615,7 +615,7 @@ export class TranscriptSessionManager implements SessionManager {
}
export function openTranscriptSessionManager(params: {
sessionFile: string;
transcriptLocator: string;
sessionId?: string;
cwd?: string;
}): SessionManager {
@@ -624,16 +624,16 @@ export function openTranscriptSessionManager(params: {
export const SessionManagerValue = {
create: (cwd: string) => TranscriptSessionManager.create(cwd),
open: (sessionFile: string, cwdOverride?: string) => {
open: (transcriptLocator: string, cwdOverride?: string) => {
return TranscriptSessionManager.open({
sessionFile,
transcriptLocator,
cwd: cwdOverride,
});
},
continueRecent: (cwd: string) => TranscriptSessionManager.continueRecent(cwd),
inMemory: (cwd?: string) => TranscriptSessionManager.inMemory(cwd),
forkFrom: (sourcePath: string, targetCwd: string) =>
TranscriptSessionManager.forkFrom(sourcePath, targetCwd),
forkFrom: (sourceTranscriptLocator: string, targetCwd: string) =>
TranscriptSessionManager.forkFrom(sourceTranscriptLocator, targetCwd),
list: (cwd: string, onProgress?: SessionListProgress) =>
TranscriptSessionManager.list(cwd, onProgress),
listAll: (onProgress?: SessionListProgress) => TranscriptSessionManager.listAll(onProgress),

4
src/agents/transcript/session-transcript-contract.ts
View File

@@ -39,10 +39,10 @@ export type SessionManager = SessionManagerType;
export const SessionManager = SessionManagerValue as {
create(cwd: string): SessionManagerType;
open(path: string, cwdOverride?: string): SessionManagerType;
open(transcriptLocator: string, cwdOverride?: string): SessionManagerType;
continueRecent(cwd: string): SessionManagerType;
inMemory(cwd?: string): SessionManagerType;
forkFrom(sourcePath: string, targetCwd: string): SessionManagerType;
forkFrom(sourceTranscriptLocator: string, targetCwd: string): SessionManagerType;
list(cwd: string, onProgress?: SessionListProgress): Promise<SessionInfo[]>;
listAll(onProgress?: SessionListProgress): Promise<SessionInfo[]>;
};

63
src/agents/transcript/transcript-state.ts
View File

@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import path from "node:path";
import { isSqliteSessionTranscriptLocator } from "../../config/sessions/paths.js";
import {
appendSqliteSessionTranscriptEvent,
@@ -59,8 +58,8 @@ function transcriptStateFromEntries(fileEntries: FileEntry[]): TranscriptState {
return new TranscriptState({ header, entries, migrated });
}
function transcriptStateFromSqlite(sessionFile: string): TranscriptState | undefined {
const scope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: sessionFile });
function transcriptStateFromSqlite(transcriptLocator: string): TranscriptState | undefined {
const scope = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath: transcriptLocator });
if (!scope) {
return undefined;
}
@@ -74,19 +73,17 @@ function transcriptStateFromSqlite(sessionFile: string): TranscriptState | undef
}
function resolveTranscriptWriteScope(
sessionFile: string,
transcriptLocator: string,
entries: Array<SessionHeader | SessionEntry>,
): { agentId: string; sessionId: string; transcriptPath: string } | undefined {
const transcriptPath = isSqliteSessionTranscriptLocator(sessionFile)
? sessionFile
: path.resolve(sessionFile);
const header = entries.find((entry): entry is SessionHeader => entry.type === "session");
const existing = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath });
if (!isSqliteSessionTranscriptLocator(transcriptPath) && !existing) {
const transcriptPath = transcriptLocator.trim();
if (!isSqliteSessionTranscriptLocator(transcriptPath)) {
throw new Error(
`Legacy transcript has not been imported into SQLite: ${transcriptPath}. Run "openclaw doctor --fix" to build the session database.`,
`Transcript locator must be SQLite-backed: ${transcriptPath}. Run "openclaw doctor --fix" to import legacy transcript files.`,
);
}
const header = entries.find((entry): entry is SessionHeader => entry.type === "session");
const existing = resolveSqliteSessionTranscriptScopeForPath({ transcriptPath });
if (!existing) {
return undefined;
}
@@ -414,33 +411,35 @@ export class TranscriptState {
}
}
export async function readTranscriptState(sessionFile: string): Promise<TranscriptState> {
const sqliteState = transcriptStateFromSqlite(sessionFile);
export async function readTranscriptState(transcriptLocator: string): Promise<TranscriptState> {
const sqliteState = transcriptStateFromSqlite(transcriptLocator);
if (sqliteState) {
return sqliteState;
}
throw new Error(
`Transcript is not in SQLite: ${sessionFile}. Run "openclaw doctor --fix" to import legacy JSONL transcripts.`,
`Transcript is not in the SQLite state database: ${transcriptLocator}. Runtime transcript readers do not read transcript files; run "openclaw doctor --fix" if legacy files still need import.`,
);
}
export function readTranscriptStateSync(sessionFile: string): TranscriptState {
const sqliteState = transcriptStateFromSqlite(sessionFile);
export function readTranscriptStateSync(transcriptLocator: string): TranscriptState {
const sqliteState = transcriptStateFromSqlite(transcriptLocator);
if (sqliteState) {
return sqliteState;
}
throw new Error(
`Transcript is not in SQLite: ${sessionFile}. Run "openclaw doctor --fix" to import legacy JSONL transcripts.`,
`Transcript is not in the SQLite state database: ${transcriptLocator}. Runtime transcript readers do not read transcript files; run "openclaw doctor --fix" if legacy files still need import.`,
);
}
export async function replaceTranscriptStateEvents(
filePath: string,
transcriptLocator: string,
entries: Array<SessionHeader | SessionEntry>,
): Promise<void> {
const scope = resolveTranscriptWriteScope(filePath, entries);
const scope = resolveTranscriptWriteScope(transcriptLocator, entries);
if (!scope) {
throw new Error(`Cannot write SQLite transcript without a session header: ${filePath}`);
throw new Error(
`Cannot write SQLite transcript without a session header: ${transcriptLocator}`,
);
}
replaceSqliteSessionTranscriptEvents({
...scope,
@@ -449,12 +448,14 @@ export async function replaceTranscriptStateEvents(
}
export function replaceTranscriptStateEventsSync(
filePath: string,
transcriptLocator: string,
entries: Array<SessionHeader | SessionEntry>,
): void {
const scope = resolveTranscriptWriteScope(filePath, entries);
const scope = resolveTranscriptWriteScope(transcriptLocator, entries);
if (!scope) {
throw new Error(`Cannot write SQLite transcript without a session header: ${filePath}`);
throw new Error(
`Cannot write SQLite transcript without a session header: ${transcriptLocator}`,
);
}
replaceSqliteSessionTranscriptEvents({
...scope,
@@ -463,7 +464,7 @@ export function replaceTranscriptStateEventsSync(
}
export async function persistTranscriptStateMutation(params: {
sessionFile: string;
transcriptLocator: string;
state: TranscriptState;
appendedEntries: SessionEntry[];
}): Promise<void> {
@@ -471,19 +472,19 @@ export async function persistTranscriptStateMutation(params: {
return;
}
if (params.state.migrated) {
await replaceTranscriptStateEvents(params.sessionFile, [
await replaceTranscriptStateEvents(params.transcriptLocator, [
...(params.state.header ? [params.state.header] : []),
...params.state.entries,
]);
return;
}
const scope = resolveTranscriptWriteScope(params.sessionFile, [
const scope = resolveTranscriptWriteScope(params.transcriptLocator, [
...(params.state.header ? [params.state.header] : []),
...params.state.entries,
]);
if (!scope) {
throw new Error(
`Cannot append SQLite transcript without a session header: ${params.sessionFile}`,
`Cannot append SQLite transcript without a session header: ${params.transcriptLocator}`,
);
}
for (const entry of params.appendedEntries) {
@@ -492,7 +493,7 @@ export async function persistTranscriptStateMutation(params: {
}
export function persistTranscriptStateMutationSync(params: {
sessionFile: string;
transcriptLocator: string;
state: TranscriptState;
appendedEntries: SessionEntry[];
}): void {
@@ -500,19 +501,19 @@ export function persistTranscriptStateMutationSync(params: {
return;
}
if (params.state.migrated) {
replaceTranscriptStateEventsSync(params.sessionFile, [
replaceTranscriptStateEventsSync(params.transcriptLocator, [
...(params.state.header ? [params.state.header] : []),
...params.state.entries,
]);
return;
}
const scope = resolveTranscriptWriteScope(params.sessionFile, [
const scope = resolveTranscriptWriteScope(params.transcriptLocator, [
...(params.state.header ? [params.state.header] : []),
...params.state.entries,
]);
if (!scope) {
throw new Error(
`Cannot append SQLite transcript without a session header: ${params.sessionFile}`,
`Cannot append SQLite transcript without a session header: ${params.transcriptLocator}`,
);
}
for (const entry of params.appendedEntries) {