mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
feat(ui): session-reset dividers + boundary markers for DB-backed sessions (#122222)
* feat(ui): show session-reset dividers and fix boundary markers on DB-backed sessions /reset now leaves a durable 'Session reset' divider at the transcript boundary in the Control UI. Root-cause fix underneath: the SQLite transcript projection only selected message events, so compaction (and now reset) markers never reached clients for DB-backed sessions; marker synthesis now has one owner (session-transcript-message.ts) consumed by both storage backends across full/recent/paged/by-id/anchor reads. Additive __openclaw marker kind 'reset' documented in clients.md. * fix(gateway): keep history readers out of the plugin SDK barrel and fix CI gates Direct imports for the sqlite history readers (the session-accessor barrel is SDK-reachable via session-transcript-lock-runtime); reset marker added to the kept-tail chat.history expectation; lint naming fixes; marker tests split into session-transcript-readers.markers.test.ts.
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
7eed2c3f21
commit
a1846dbebc
@@ -196,6 +196,9 @@ Rows returned by `chat.history` can carry an `__openclaw` metadata envelope:
|
||||
`kind: "compaction"` and may include `tokensBefore` and `tokensAfter` when a
|
||||
matching checkpoint recorded those metrics.
|
||||
|
||||
A session reset boundary uses `kind: "reset"`. It has no checkpoint token
|
||||
metrics.
|
||||
|
||||
Page backward with the response's `hasMore` and `nextOffset` values. Numeric
|
||||
offsets describe the current transcript projection, so do not persist them as
|
||||
long-lived bookmarks across reset or compaction. Persist `__openclaw.id` instead.
|
||||
|
||||
@@ -23,7 +23,7 @@ type ActiveTranscriptDatabase = Pick<
|
||||
| "transcript_events"
|
||||
>;
|
||||
|
||||
type CurrentTranscriptProjection = {
|
||||
export type CurrentTranscriptProjection = {
|
||||
database: OpenClawAgentDatabase;
|
||||
resolved: ReturnType<typeof resolveSqliteTranscriptReadScope>;
|
||||
state: SessionTranscriptProjectionState;
|
||||
|
||||
@@ -0,0 +1,328 @@
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
} from "../../infra/kysely-sync.js";
|
||||
import type {
|
||||
SessionTranscriptMessageAnchorPage,
|
||||
SessionTranscriptMessageEvent,
|
||||
SessionTranscriptMessageEventPage,
|
||||
} from "./session-accessor.sqlite-active-events.js";
|
||||
import {
|
||||
getActiveTranscriptKysely,
|
||||
withCurrentProjectionSnapshot,
|
||||
type CurrentTranscriptProjection,
|
||||
} from "./session-accessor.sqlite-active-projection.js";
|
||||
import type {
|
||||
SessionTranscriptReadScope,
|
||||
TranscriptEvent,
|
||||
} from "./session-accessor.sqlite-contract.js";
|
||||
import {
|
||||
readVisibleMessageRange,
|
||||
resolveVisibleMessagePositions,
|
||||
} from "./session-accessor.sqlite-reset-window.js";
|
||||
import { MAX_VISIBLE_MESSAGE_MAX_MESSAGES } from "./session-accessor.sqlite-visible-cursor.js";
|
||||
|
||||
type VisibleHistoryBoundary = {
|
||||
displayPosition: number;
|
||||
event: TranscriptEvent;
|
||||
messagePosition: number;
|
||||
};
|
||||
|
||||
type VisibleHistoryProjection = {
|
||||
boundaries: VisibleHistoryBoundary[];
|
||||
total: number;
|
||||
};
|
||||
|
||||
function resolveVisibleHistoryProjection(
|
||||
projection: CurrentTranscriptProjection,
|
||||
): VisibleHistoryProjection {
|
||||
const visibleMessages = resolveVisibleMessagePositions(projection);
|
||||
const db = getActiveTranscriptKysely(projection.database);
|
||||
const rows = executeSqliteQuerySync(
|
||||
projection.database.db,
|
||||
db
|
||||
.selectFrom("session_transcript_active_events as active")
|
||||
.innerJoin("transcript_event_identities as identity", (join) =>
|
||||
join
|
||||
.onRef("identity.session_id", "=", "active.session_id")
|
||||
.onRef("identity.seq", "=", "active.event_seq"),
|
||||
)
|
||||
.innerJoin("transcript_events as event", (join) =>
|
||||
join
|
||||
.onRef("event.session_id", "=", "active.session_id")
|
||||
.onRef("event.seq", "=", "active.event_seq"),
|
||||
)
|
||||
.select(["identity.event_type", "event.event_json"])
|
||||
.select((eb) =>
|
||||
eb
|
||||
.selectFrom("session_transcript_active_events as next")
|
||||
.select((nextEb) => nextEb.fn.min<number>("next.message_position").as("position"))
|
||||
.whereRef("next.session_id", "=", "active.session_id")
|
||||
.whereRef("next.active_position", ">", "active.active_position")
|
||||
.where("next.message_position", "is not", null)
|
||||
.as("next_message_position"),
|
||||
)
|
||||
.where("active.session_id", "=", projection.resolved.sessionId)
|
||||
.where("identity.event_type", "in", ["compaction", "reset"])
|
||||
.orderBy("active.active_position", "asc"),
|
||||
).rows;
|
||||
const latestBoundaryIsReset = rows.at(-1)?.event_type === "reset";
|
||||
const visibleRows = latestBoundaryIsReset ? rows.slice(-1) : rows;
|
||||
let priorBoundaries = 0;
|
||||
const boundaries = visibleRows.map((row): VisibleHistoryBoundary => {
|
||||
const messagePosition = latestBoundaryIsReset
|
||||
? visibleMessages.kept.length
|
||||
: Math.min(
|
||||
row.next_message_position ?? projection.state.activeMessageCount,
|
||||
visibleMessages.total,
|
||||
);
|
||||
return {
|
||||
displayPosition: messagePosition + priorBoundaries++,
|
||||
event: JSON.parse(row.event_json) as TranscriptEvent,
|
||||
messagePosition,
|
||||
};
|
||||
});
|
||||
return {
|
||||
boundaries,
|
||||
total: visibleMessages.total + boundaries.length,
|
||||
};
|
||||
}
|
||||
|
||||
function readVisibleHistoryRange(
|
||||
projection: CurrentTranscriptProjection,
|
||||
start: number,
|
||||
endExclusive: number,
|
||||
history = resolveVisibleHistoryProjection(projection),
|
||||
): SessionTranscriptMessageEvent[] {
|
||||
const boundedStart = Math.min(Math.max(0, start), history.total);
|
||||
const boundedEnd = Math.min(Math.max(boundedStart, endExclusive), history.total);
|
||||
if (boundedEnd <= boundedStart) {
|
||||
return [];
|
||||
}
|
||||
const boundaries = new Map(
|
||||
history.boundaries.map((boundary) => [boundary.displayPosition, boundary] as const),
|
||||
);
|
||||
const boundariesBefore = history.boundaries.filter(
|
||||
(boundary) => boundary.displayPosition < boundedStart,
|
||||
).length;
|
||||
const selectedBoundaryCount = history.boundaries.filter(
|
||||
(boundary) => boundary.displayPosition >= boundedStart && boundary.displayPosition < boundedEnd,
|
||||
).length;
|
||||
const messageStart = boundedStart - boundariesBefore;
|
||||
const messageEnd = messageStart + boundedEnd - boundedStart - selectedBoundaryCount;
|
||||
const messages = readVisibleMessageRange(projection, messageStart, messageEnd);
|
||||
let messageIndex = 0;
|
||||
const events: SessionTranscriptMessageEvent[] = [];
|
||||
for (let displayPosition = boundedStart; displayPosition < boundedEnd; displayPosition += 1) {
|
||||
const boundary = boundaries.get(displayPosition);
|
||||
if (boundary) {
|
||||
events.push({ event: boundary.event, seq: displayPosition + 1 });
|
||||
continue;
|
||||
}
|
||||
const message = messages[messageIndex++];
|
||||
if (message) {
|
||||
events.push({ event: message.event, seq: displayPosition + 1 });
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
function readVisibleMessageById(
|
||||
projection: CurrentTranscriptProjection,
|
||||
eventId: string,
|
||||
): SessionTranscriptMessageEvent | undefined {
|
||||
const db = getActiveTranscriptKysely(projection.database);
|
||||
const row = executeSqliteQueryTakeFirstSync(
|
||||
projection.database.db,
|
||||
db
|
||||
.selectFrom("transcript_event_identities as identity")
|
||||
.innerJoin("session_transcript_active_events as active", (join) =>
|
||||
join
|
||||
.onRef("active.session_id", "=", "identity.session_id")
|
||||
.onRef("active.event_seq", "=", "identity.seq"),
|
||||
)
|
||||
.innerJoin("transcript_events as event", (join) =>
|
||||
join
|
||||
.onRef("event.session_id", "=", "active.session_id")
|
||||
.onRef("event.seq", "=", "active.event_seq"),
|
||||
)
|
||||
.select(["active.message_position", "event.event_json"])
|
||||
.where("identity.session_id", "=", projection.resolved.sessionId)
|
||||
.where("identity.event_id", "=", eventId)
|
||||
.where("active.message_position", "is not", null),
|
||||
);
|
||||
if (!row || row.message_position === null) {
|
||||
return undefined;
|
||||
}
|
||||
const visible = resolveVisibleMessagePositions(projection);
|
||||
const logicalPosition =
|
||||
row.message_position >= visible.postStart
|
||||
? visible.kept.length + row.message_position - visible.postStart
|
||||
: visible.kept.indexOf(row.message_position);
|
||||
return logicalPosition < 0
|
||||
? undefined
|
||||
: { event: JSON.parse(row.event_json) as TranscriptEvent, seq: logicalPosition + 1 };
|
||||
}
|
||||
|
||||
function resolveHistoryEventById(
|
||||
projection: CurrentTranscriptProjection,
|
||||
eventId: string,
|
||||
history = resolveVisibleHistoryProjection(projection),
|
||||
): SessionTranscriptMessageEvent | undefined {
|
||||
const boundary = history.boundaries.find(
|
||||
(candidate) => (candidate.event as { id?: unknown }).id === eventId,
|
||||
);
|
||||
if (boundary) {
|
||||
return { event: boundary.event, seq: boundary.displayPosition + 1 };
|
||||
}
|
||||
const message = readVisibleMessageById(projection, eventId);
|
||||
if (!message) {
|
||||
return undefined;
|
||||
}
|
||||
const messagePosition = message.seq - 1;
|
||||
const precedingBoundaries = history.boundaries.filter(
|
||||
(candidate) => candidate.messagePosition <= messagePosition,
|
||||
).length;
|
||||
return {
|
||||
event: message.event,
|
||||
seq: message.seq + precedingBoundaries,
|
||||
};
|
||||
}
|
||||
|
||||
export function readSessionTranscriptHistoryEvents(
|
||||
scope: SessionTranscriptReadScope,
|
||||
): SessionTranscriptMessageEvent[] {
|
||||
return withCurrentProjectionSnapshot(scope, (projection) => {
|
||||
const history = resolveVisibleHistoryProjection(projection);
|
||||
return readVisibleHistoryRange(projection, 0, history.total, history);
|
||||
});
|
||||
}
|
||||
|
||||
export function readRecentSessionTranscriptHistoryEvents(
|
||||
scope: SessionTranscriptReadScope,
|
||||
options: { maxBytes: number; maxLines: number; maxMessages: number },
|
||||
): SessionTranscriptMessageEventPage {
|
||||
return withCurrentProjectionSnapshot(scope, (projection) => {
|
||||
const history = resolveVisibleHistoryProjection(projection);
|
||||
const maxMessages = Math.min(
|
||||
MAX_VISIBLE_MESSAGE_MAX_MESSAGES,
|
||||
Math.max(0, Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 0)),
|
||||
);
|
||||
const maxLines = Math.max(
|
||||
0,
|
||||
Math.floor(Number.isFinite(options.maxLines) ? options.maxLines : 0),
|
||||
);
|
||||
if (maxMessages === 0 || maxLines === 0) {
|
||||
return {
|
||||
activeLeafEntryId: projection.state.leafEventId,
|
||||
events: [],
|
||||
totalMessages: history.total,
|
||||
};
|
||||
}
|
||||
const maxBytes = Math.max(
|
||||
1024,
|
||||
Math.floor(Number.isFinite(options.maxBytes) ? options.maxBytes : 8 * 1024 * 1024),
|
||||
);
|
||||
const candidates = readVisibleHistoryRange(
|
||||
projection,
|
||||
Math.max(0, history.total - maxLines),
|
||||
history.total,
|
||||
history,
|
||||
);
|
||||
const selected: SessionTranscriptMessageEvent[] = [];
|
||||
let bytes = 0;
|
||||
for (const event of candidates.toReversed()) {
|
||||
const eventBytes = Buffer.byteLength(JSON.stringify(event.event)) + 1;
|
||||
if (
|
||||
selected.length >= maxMessages ||
|
||||
(selected.length > 0 && bytes + eventBytes > maxBytes)
|
||||
) {
|
||||
break;
|
||||
}
|
||||
selected.push(event);
|
||||
bytes += eventBytes;
|
||||
}
|
||||
return {
|
||||
activeLeafEntryId: projection.state.leafEventId,
|
||||
events: selected.toReversed(),
|
||||
totalMessages: history.total,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export function readSessionTranscriptHistoryEventPage(
|
||||
scope: SessionTranscriptReadScope,
|
||||
options: { maxMessages: number; offset: number },
|
||||
): SessionTranscriptMessageEventPage {
|
||||
return withCurrentProjectionSnapshot(scope, (projection) => {
|
||||
const history = resolveVisibleHistoryProjection(projection);
|
||||
const offset = Math.min(
|
||||
Math.max(0, Math.floor(Number.isFinite(options.offset) ? options.offset : 0)),
|
||||
history.total,
|
||||
);
|
||||
const maxMessages = Math.max(
|
||||
0,
|
||||
Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 0),
|
||||
);
|
||||
const endExclusive = Math.max(0, history.total - offset);
|
||||
const start = Math.max(0, endExclusive - maxMessages);
|
||||
return {
|
||||
activeLeafEntryId: projection.state.leafEventId,
|
||||
events: readVisibleHistoryRange(projection, start, endExclusive, history),
|
||||
totalMessages: history.total,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export function readSessionTranscriptHistoryEventCount(scope: SessionTranscriptReadScope): number {
|
||||
return withCurrentProjectionSnapshot(
|
||||
scope,
|
||||
(projection) => resolveVisibleHistoryProjection(projection).total,
|
||||
);
|
||||
}
|
||||
|
||||
export function readSessionTranscriptHistoryEventById(
|
||||
scope: SessionTranscriptReadScope,
|
||||
eventId: string,
|
||||
): SessionTranscriptMessageEvent | undefined {
|
||||
return withCurrentProjectionSnapshot(scope, (projection) =>
|
||||
resolveHistoryEventById(projection, eventId),
|
||||
);
|
||||
}
|
||||
|
||||
export function readSessionTranscriptHistoryAnchorPage(
|
||||
scope: SessionTranscriptReadScope,
|
||||
options: { maxMessages: number; messageId: string },
|
||||
): SessionTranscriptMessageAnchorPage {
|
||||
return withCurrentProjectionSnapshot(scope, (projection) => {
|
||||
const history = resolveVisibleHistoryProjection(projection);
|
||||
const anchor = resolveHistoryEventById(projection, options.messageId, history);
|
||||
if (!anchor) {
|
||||
return {
|
||||
events: [],
|
||||
found: false,
|
||||
hasOverreadContext: false,
|
||||
offset: 0,
|
||||
totalMessages: history.total,
|
||||
};
|
||||
}
|
||||
const pageSize = Math.max(
|
||||
1,
|
||||
Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 1),
|
||||
);
|
||||
const anchorPosition = anchor.seq - 1;
|
||||
const newerMessages = Math.floor(pageSize / 2);
|
||||
const olderMessages = pageSize - newerMessages - 1;
|
||||
const latestStart = Math.max(0, history.total - pageSize);
|
||||
const start = Math.min(Math.max(0, anchorPosition - olderMessages), latestStart);
|
||||
const endExclusive = Math.min(history.total, start + pageSize);
|
||||
const readStart = Math.max(0, start - 1);
|
||||
return {
|
||||
events: readVisibleHistoryRange(projection, readStart, endExclusive, history),
|
||||
found: true,
|
||||
hasOverreadContext: readStart < start,
|
||||
offset: history.total - endExclusive,
|
||||
totalMessages: history.total,
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -830,7 +830,7 @@ describe("gateway server chat", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("chat.history applies the reset boundary kept-tail cut", async () => {
|
||||
test("chat.history applies the reset kept-tail cut and preserves its marker", async () => {
|
||||
await withMainSessionStore(async () => {
|
||||
const storePath = testState.sessionStorePath;
|
||||
if (!storePath) {
|
||||
@@ -883,6 +883,7 @@ describe("gateway server chat", () => {
|
||||
expect(collectHistoryTextValues(history.payload?.messages ?? [])).toEqual([
|
||||
"kept question",
|
||||
"kept answer",
|
||||
"Reset",
|
||||
"new turn",
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
import {
|
||||
readSessionTranscriptMessageAnchorPage,
|
||||
type SessionTranscriptReadScope,
|
||||
} from "../config/sessions/session-accessor.js";
|
||||
import type { SessionTranscriptReadScope } from "../config/sessions/session-accessor.js";
|
||||
import { readSessionTranscriptHistoryAnchorPage } from "../config/sessions/session-accessor.sqlite-history-events.js";
|
||||
import { projectTranscriptEntryMessage } from "./session-transcript-message.js";
|
||||
import {
|
||||
resolveTranscriptReadTarget,
|
||||
sqliteMessageEventWithSeq,
|
||||
toTranscriptReadScope,
|
||||
type ReadRecentSessionMessagesResult,
|
||||
} from "./session-transcript-readers.js";
|
||||
@@ -28,7 +26,7 @@ export async function readSessionMessagesAroundIdWithStatsAsync(
|
||||
scope.sessionEntry.sessionId !== scope.sessionId
|
||||
? undefined
|
||||
: target.sessionFile;
|
||||
const page = readSessionTranscriptMessageAnchorPage(toTranscriptReadScope(target), opts);
|
||||
const page = readSessionTranscriptHistoryAnchorPage(toTranscriptReadScope(target), opts);
|
||||
if (!page.found) {
|
||||
if (opts.allowResetArchiveFallback === true) {
|
||||
return await new ArchivedTranscriptReader({
|
||||
@@ -51,7 +49,7 @@ export async function readSessionMessagesAroundIdWithStatsAsync(
|
||||
found: true,
|
||||
hasOverreadContext: page.hasOverreadContext,
|
||||
messages: page.events.flatMap((entry) => {
|
||||
const message = sqliteMessageEventWithSeq(entry);
|
||||
const message = projectTranscriptEntryMessage(entry.event, entry.seq);
|
||||
return message === undefined ? [] : [message];
|
||||
}),
|
||||
offset: page.offset,
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
/** Attach OpenClaw metadata to a transcript message without dropping existing metadata. */
|
||||
export function attachOpenClawTranscriptMeta(
|
||||
message: unknown,
|
||||
meta: Record<string, unknown>,
|
||||
): unknown {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return message;
|
||||
}
|
||||
const record = message as Record<string, unknown>;
|
||||
const existing =
|
||||
record["__openclaw"] &&
|
||||
typeof record["__openclaw"] === "object" &&
|
||||
!Array.isArray(record["__openclaw"])
|
||||
? (record["__openclaw"] as Record<string, unknown>)
|
||||
: {};
|
||||
return {
|
||||
...record,
|
||||
__openclaw: {
|
||||
...existing,
|
||||
...meta,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function readTranscriptMessageIdempotencyKey(message: unknown): string | undefined {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return undefined;
|
||||
}
|
||||
const value = (message as Record<string, unknown>).idempotencyKey;
|
||||
return typeof value === "string" && value.trim() ? value : undefined;
|
||||
}
|
||||
|
||||
/** Project one stored transcript entry onto the client-visible chat history shape. */
|
||||
export function projectTranscriptEntryMessage(entry: unknown, seq: number): unknown {
|
||||
if (!entry || typeof entry !== "object" || Array.isArray(entry)) {
|
||||
return null;
|
||||
}
|
||||
const record = entry as Record<string, unknown>;
|
||||
if (record.message) {
|
||||
const recordTimestampMs =
|
||||
typeof record.timestamp === "string"
|
||||
? Date.parse(record.timestamp)
|
||||
: typeof record.timestamp === "number"
|
||||
? record.timestamp
|
||||
: Number.NaN;
|
||||
const idempotencyKey = readTranscriptMessageIdempotencyKey(record.message);
|
||||
return attachOpenClawTranscriptMeta(record.message, {
|
||||
...(typeof record.id === "string" ? { id: record.id } : {}),
|
||||
...(idempotencyKey ? { idempotencyKey } : {}),
|
||||
...(Number.isFinite(recordTimestampMs) ? { recordTimestampMs } : {}),
|
||||
seq,
|
||||
});
|
||||
}
|
||||
if (record.type !== "compaction" && record.type !== "reset") {
|
||||
return null;
|
||||
}
|
||||
const kind = record.type;
|
||||
const parsedTimestamp =
|
||||
typeof record.timestamp === "string" ? Date.parse(record.timestamp) : Number.NaN;
|
||||
return {
|
||||
role: "system",
|
||||
content: [{ type: "text", text: kind === "compaction" ? "Compaction" : "Reset" }],
|
||||
timestamp: Number.isFinite(parsedTimestamp) ? parsedTimestamp : Date.now(),
|
||||
__openclaw: {
|
||||
kind,
|
||||
id: typeof record.id === "string" ? record.id : undefined,
|
||||
seq,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, test } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
|
||||
import { replaceTranscriptEvents } from "../config/sessions/session-accessor.js";
|
||||
import { closeOpenClawAgentDatabasesForTest } from "../state/openclaw-agent-db.js";
|
||||
import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js";
|
||||
import { captureEnv, setTestEnvValue } from "../test-utils/env.js";
|
||||
import { readSessionMessagesAroundIdWithStatsAsync } from "./session-transcript-anchor-reader.js";
|
||||
import {
|
||||
readRecentSessionMessagesWithStatsAsync,
|
||||
readSessionMessageByIdAsync,
|
||||
readSessionMessageCountAsync,
|
||||
readSessionMessagesAsync,
|
||||
readSessionMessagesPageWithStatsAsync,
|
||||
type SessionTranscriptReadScope,
|
||||
} from "./session-transcript-readers.js";
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
describe("session transcript reader marker projection", () => {
|
||||
let tempDir: string;
|
||||
let storePath: string;
|
||||
let envSnapshot: ReturnType<typeof captureEnv>;
|
||||
|
||||
beforeEach(() => {
|
||||
envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]);
|
||||
tempDir = tempDirs.make("openclaw-transcript-markers-");
|
||||
storePath = path.join(tempDir, "sessions.json");
|
||||
setTestEnvValue("OPENCLAW_STATE_DIR", tempDir);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
envSnapshot.restore();
|
||||
});
|
||||
|
||||
async function writeTranscript(
|
||||
sessionId: string,
|
||||
events: unknown[],
|
||||
): Promise<SessionTranscriptReadScope> {
|
||||
const scope = {
|
||||
agentId: "main",
|
||||
sessionId,
|
||||
sessionKey: `agent:main:${sessionId}`,
|
||||
storePath,
|
||||
};
|
||||
await replaceTranscriptEvents(scope, events);
|
||||
return scope;
|
||||
}
|
||||
|
||||
test.each([
|
||||
{
|
||||
name: "compaction",
|
||||
sessionId: "reader-compaction-boundary",
|
||||
markerId: "compaction-boundary",
|
||||
markerKind: "compaction",
|
||||
markerText: "Compaction",
|
||||
events: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
{
|
||||
type: "message",
|
||||
id: "before-compaction",
|
||||
parentId: null,
|
||||
message: { role: "user", content: "before compaction" },
|
||||
},
|
||||
{
|
||||
type: "compaction",
|
||||
id: "compaction-boundary",
|
||||
parentId: "before-compaction",
|
||||
timestamp: "2026-08-11T18:00:00.000Z",
|
||||
summary: "summary",
|
||||
firstKeptEntryId: "before-compaction",
|
||||
tokensBefore: 100,
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "after-compaction",
|
||||
parentId: "compaction-boundary",
|
||||
message: { role: "assistant", content: "after compaction" },
|
||||
},
|
||||
],
|
||||
expected: ["before compaction", "compaction", "after compaction"],
|
||||
},
|
||||
{
|
||||
name: "reset",
|
||||
sessionId: "reader-reset-boundary",
|
||||
markerId: "reset-boundary",
|
||||
markerKind: "reset",
|
||||
markerText: "Reset",
|
||||
events: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
{
|
||||
type: "message",
|
||||
id: "old",
|
||||
parentId: null,
|
||||
message: { role: "user", content: "hidden old turn" },
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "kept-user",
|
||||
parentId: "old",
|
||||
message: { role: "user", content: "kept question" },
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "kept-assistant",
|
||||
parentId: "kept-user",
|
||||
message: { role: "assistant", content: "kept answer" },
|
||||
},
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: "kept-assistant",
|
||||
timestamp: "2026-08-11T18:00:00.000Z",
|
||||
reason: "reset",
|
||||
firstKeptEntryId: "kept-user",
|
||||
},
|
||||
{
|
||||
type: "message",
|
||||
id: "post-reset",
|
||||
parentId: "reset-boundary",
|
||||
message: { role: "assistant", content: "new answer" },
|
||||
},
|
||||
],
|
||||
expected: ["kept question", "kept answer", "reset", "new answer"],
|
||||
},
|
||||
])("projects $name boundaries through every SQLite history read", async (fixture) => {
|
||||
const scope = await writeTranscript(fixture.sessionId, fixture.events(fixture.sessionId));
|
||||
const summarize = (messages: unknown[]) =>
|
||||
messages.map((message) => {
|
||||
const record = message as { content?: unknown; __openclaw?: { kind?: string } };
|
||||
return record["__openclaw"]?.kind ?? record.content;
|
||||
});
|
||||
|
||||
const full = await readSessionMessagesAsync(scope, {
|
||||
mode: "full",
|
||||
reason: `${fixture.name} boundary projection test`,
|
||||
});
|
||||
const recent = await readRecentSessionMessagesWithStatsAsync(scope, {
|
||||
maxBytes: 16_384,
|
||||
maxLines: 10,
|
||||
maxMessages: 10,
|
||||
});
|
||||
const markerIndex = fixture.expected.indexOf(fixture.markerKind);
|
||||
const page = await readSessionMessagesPageWithStatsAsync(scope, {
|
||||
maxMessages: 1,
|
||||
offset: fixture.expected.length - markerIndex - 1,
|
||||
});
|
||||
const byId = await readSessionMessageByIdAsync(scope, fixture.markerId);
|
||||
const anchored = await readSessionMessagesAroundIdWithStatsAsync(scope, {
|
||||
messageId: fixture.markerId,
|
||||
maxMessages: 10,
|
||||
});
|
||||
|
||||
expect(summarize(full)).toEqual(fixture.expected);
|
||||
expect(summarize(recent.messages)).toEqual(fixture.expected);
|
||||
expect(recent.totalMessages).toBe(fixture.expected.length);
|
||||
expect(summarize(page.messages)).toEqual([fixture.markerKind]);
|
||||
expect(page.totalMessages).toBe(fixture.expected.length);
|
||||
expect(await readSessionMessageCountAsync(scope)).toBe(fixture.expected.length);
|
||||
expect(byId).toMatchObject({
|
||||
found: true,
|
||||
message: {
|
||||
role: "system",
|
||||
content: [{ type: "text", text: fixture.markerText }],
|
||||
timestamp: Date.parse("2026-08-11T18:00:00.000Z"),
|
||||
__openclaw: {
|
||||
kind: fixture.markerKind,
|
||||
id: fixture.markerId,
|
||||
seq: markerIndex + 1,
|
||||
},
|
||||
},
|
||||
seq: markerIndex + 1,
|
||||
});
|
||||
expect(anchored.found).toBe(true);
|
||||
expect(summarize(anchored.messages)).toEqual(fixture.expected);
|
||||
expect(anchored.totalMessages).toBe(fixture.expected.length);
|
||||
});
|
||||
});
|
||||
@@ -3,9 +3,6 @@ import { parseDateFirstTimestampMs } from "@openclaw/normalization-core/number-c
|
||||
import {
|
||||
isSessionTranscriptProjectionUnavailableError,
|
||||
readRecentSessionTranscriptMessageEvents,
|
||||
readSessionTranscriptMessageEventById,
|
||||
readSessionTranscriptMessageEventCount,
|
||||
readSessionTranscriptMessageEventPage,
|
||||
readSessionTranscriptMessageEvents,
|
||||
resolveConcreteSessionStorePath,
|
||||
resolveSessionTranscriptReadTarget,
|
||||
@@ -14,8 +11,19 @@ import {
|
||||
type SessionTranscriptReadScope,
|
||||
type TranscriptEvent,
|
||||
} from "../config/sessions/session-accessor.js";
|
||||
import {
|
||||
readRecentSessionTranscriptHistoryEvents,
|
||||
readSessionTranscriptHistoryEventById,
|
||||
readSessionTranscriptHistoryEventCount,
|
||||
readSessionTranscriptHistoryEventPage,
|
||||
readSessionTranscriptHistoryEvents,
|
||||
} from "../config/sessions/session-accessor.sqlite-history-events.js";
|
||||
import { resolveAgentIdFromSessionKey } from "../routing/session-key.js";
|
||||
import { aggregateSqliteUsageSnapshots } from "./session-transcript-derived-readers.js";
|
||||
import {
|
||||
attachOpenClawTranscriptMeta,
|
||||
projectTranscriptEntryMessage,
|
||||
} from "./session-transcript-message.js";
|
||||
import type {
|
||||
ReadRecentSessionMessagesOptions,
|
||||
ReadSessionMessagesAsyncOptions,
|
||||
@@ -23,14 +31,14 @@ import type {
|
||||
} from "./session-utils.fs.js";
|
||||
import {
|
||||
ArchivedTranscriptReader,
|
||||
attachOpenClawTranscriptMeta,
|
||||
buildSessionPreviewItems,
|
||||
readLatestSessionUsageFromTranscriptFileAsync,
|
||||
} from "./session-utils.fs.js";
|
||||
import type { SessionPreviewItem } from "./session-utils.types.js";
|
||||
|
||||
export type { ReadSessionMessagesAsyncOptions };
|
||||
export { attachOpenClawTranscriptMeta, capArrayByJsonBytes } from "./session-utils.fs.js";
|
||||
export { capArrayByJsonBytes } from "./session-utils.fs.js";
|
||||
export { attachOpenClawTranscriptMeta } from "./session-transcript-message.js";
|
||||
export { readSessionTranscriptVisibleMessageDeltaCore } from "../config/sessions/session-accessor.js";
|
||||
|
||||
export type { SessionTranscriptReadScope };
|
||||
@@ -148,6 +156,19 @@ async function readSqliteMessageRecords(
|
||||
);
|
||||
}
|
||||
|
||||
function projectSqliteHistoryEvents(entries: readonly SessionTranscriptMessageEvent[]): unknown[] {
|
||||
return entries.flatMap((entry) => {
|
||||
const message = projectTranscriptEntryMessage(entry.event, entry.seq);
|
||||
return message ? [message] : [];
|
||||
});
|
||||
}
|
||||
|
||||
async function readSqliteHistoryMessages(target: ResolvedTranscriptReadTarget): Promise<unknown[]> {
|
||||
return projectSqliteHistoryEvents(
|
||||
readSessionTranscriptHistoryEvents(toTranscriptReadScope(target)),
|
||||
);
|
||||
}
|
||||
|
||||
function readSqliteMessagesSync(target: ResolvedTranscriptReadTarget): unknown[] {
|
||||
return readSqliteMessageRecordsSync(target).map(sqliteRecordMessageWithSeq);
|
||||
}
|
||||
@@ -171,17 +192,17 @@ async function readRecentSqliteMessageRecords(
|
||||
opts?: Partial<ReadRecentSessionMessagesOptions>,
|
||||
): Promise<{
|
||||
activeLeafEntryId?: string | null;
|
||||
records: SqliteMessageRecord[];
|
||||
messages: unknown[];
|
||||
transcriptEvents: TranscriptEvent[];
|
||||
totalMessages: number;
|
||||
}> {
|
||||
const normalized = normalizeRecentSqliteReadOptions(opts);
|
||||
const page = readRecentSessionTranscriptMessageEvents(toTranscriptReadScope(target), normalized);
|
||||
const page = readRecentSessionTranscriptHistoryEvents(toTranscriptReadScope(target), normalized);
|
||||
return {
|
||||
...(Object.hasOwn(page, "activeLeafEntryId")
|
||||
? { activeLeafEntryId: page.activeLeafEntryId }
|
||||
: {}),
|
||||
records: extractMessageRecordsFromEventEntries(page.events),
|
||||
messages: projectSqliteHistoryEvents(page.events),
|
||||
transcriptEvents: page.events.map((entry) => entry.event),
|
||||
totalMessages: page.totalMessages,
|
||||
};
|
||||
@@ -222,8 +243,7 @@ function sqliteRecordMessageWithSeq(record: {
|
||||
}
|
||||
|
||||
export function sqliteMessageEventWithSeq(entry: SessionTranscriptMessageEvent): unknown {
|
||||
const record = extractMessageRecord(entry.event);
|
||||
return record ? sqliteRecordMessageWithSeq({ ...record, seq: entry.seq }) : undefined;
|
||||
return projectTranscriptEntryMessage(entry.event, entry.seq);
|
||||
}
|
||||
|
||||
export function extractMessageRole(message: unknown): string | undefined {
|
||||
@@ -279,19 +299,19 @@ export async function readSessionMessagesAsync(
|
||||
): Promise<unknown[]> {
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
if (opts.mode === "recent") {
|
||||
const { records } = await readRecentSqliteMessageRecords(target, opts);
|
||||
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
const { messages } = await readRecentSqliteMessageRecords(target, opts);
|
||||
if (messages.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
return (await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true }))
|
||||
.messages;
|
||||
}
|
||||
return records.map(sqliteRecordMessageWithSeq);
|
||||
return messages;
|
||||
}
|
||||
const records = await readSqliteMessageRecords(target);
|
||||
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
const messages = await readSqliteHistoryMessages(target);
|
||||
if (messages.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
return (await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true }))
|
||||
.messages;
|
||||
}
|
||||
return records.map(sqliteRecordMessageWithSeq);
|
||||
return messages;
|
||||
}
|
||||
|
||||
/** Reads display messages with source metadata through the reader seam. */
|
||||
@@ -300,15 +320,15 @@ export async function readSessionMessagesWithSourceAsync(
|
||||
opts: ReadSessionMessagesAsyncOptions,
|
||||
): Promise<ReadSessionMessagesResult> {
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
const records =
|
||||
const messages =
|
||||
opts.mode === "recent"
|
||||
? (await readRecentSqliteMessageRecords(target, opts)).records
|
||||
: await readSqliteMessageRecords(target);
|
||||
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
? (await readRecentSqliteMessageRecords(target, opts)).messages
|
||||
: await readSqliteHistoryMessages(target);
|
||||
if (messages.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
return await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true });
|
||||
}
|
||||
return {
|
||||
messages: records.map(sqliteRecordMessageWithSeq),
|
||||
messages,
|
||||
transcriptPath: target.sessionFile,
|
||||
};
|
||||
}
|
||||
@@ -320,13 +340,17 @@ export async function readSessionMessageByIdAsync(
|
||||
opts?: { allowResetArchiveFallback?: boolean },
|
||||
): Promise<ReadSessionMessageByIdResult> {
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
const foundEvent = readSessionTranscriptMessageEventById(
|
||||
const foundEvent = readSessionTranscriptHistoryEventById(
|
||||
toTranscriptReadScope(target),
|
||||
messageId,
|
||||
);
|
||||
const found = foundEvent ? extractMessageRecordsFromEventEntries([foundEvent]).at(0) : undefined;
|
||||
if (found) {
|
||||
return { found: true, message: found.message, oversized: false, seq: found.seq };
|
||||
if (foundEvent) {
|
||||
return {
|
||||
found: true,
|
||||
message: projectTranscriptEntryMessage(foundEvent.event, foundEvent.seq),
|
||||
oversized: false,
|
||||
seq: foundEvent.seq,
|
||||
};
|
||||
}
|
||||
if (opts?.allowResetArchiveFallback === true) {
|
||||
return await archivedTranscriptReader(target).readById(messageId, {
|
||||
@@ -359,7 +383,7 @@ export async function readSessionMessageCountAsync(
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
const transcriptScope = toTranscriptReadScope(target);
|
||||
try {
|
||||
return readSessionTranscriptMessageEventCount(transcriptScope);
|
||||
return readSessionTranscriptHistoryEventCount(transcriptScope);
|
||||
} catch (error) {
|
||||
if (!isSessionTranscriptProjectionUnavailableError(error)) {
|
||||
throw error;
|
||||
@@ -367,7 +391,7 @@ export async function readSessionMessageCountAsync(
|
||||
// The failed read already scheduled the rebuild; wait before assigning
|
||||
// a sequence so a concurrent send cannot fail or reuse a stale count.
|
||||
await waitForSessionTranscriptProjection(transcriptScope);
|
||||
return readSessionTranscriptMessageEventCount(transcriptScope);
|
||||
return readSessionTranscriptHistoryEventCount(transcriptScope);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -377,9 +401,9 @@ export async function readRecentSessionMessagesWithStatsAsync(
|
||||
opts: ReadRecentSessionMessagesOptions,
|
||||
): Promise<ReadRecentSessionMessagesResult> {
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
const { activeLeafEntryId, records, transcriptEvents, totalMessages } =
|
||||
const { activeLeafEntryId, messages, transcriptEvents, totalMessages } =
|
||||
await readRecentSqliteMessageRecords(target, opts);
|
||||
if (totalMessages === 0 && records.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
if (totalMessages === 0 && messages.length === 0 && opts.allowResetArchiveFallback === true) {
|
||||
return await archivedTranscriptReader(target).readRecentWithStats({
|
||||
...opts,
|
||||
resetArchiveOnly: true,
|
||||
@@ -387,7 +411,7 @@ export async function readRecentSessionMessagesWithStatsAsync(
|
||||
}
|
||||
return {
|
||||
...(activeLeafEntryId !== undefined ? { activeLeafEntryId } : {}),
|
||||
messages: records.map(sqliteRecordMessageWithSeq),
|
||||
messages,
|
||||
transcriptEvents,
|
||||
totalMessages,
|
||||
transcriptPath: target.sessionFile,
|
||||
@@ -401,7 +425,7 @@ export async function readSessionMessagesPageWithStatsAsync(
|
||||
opts: { offset: number; maxMessages: number; allowResetArchiveFallback?: boolean },
|
||||
): Promise<ReadRecentSessionMessagesResult> {
|
||||
const target = resolveTranscriptReadTarget(scope);
|
||||
const page = readSessionTranscriptMessageEventPage(toTranscriptReadScope(target), opts);
|
||||
const page = readSessionTranscriptHistoryEventPage(toTranscriptReadScope(target), opts);
|
||||
if (page.totalMessages === 0 && opts.allowResetArchiveFallback === true) {
|
||||
return await archivedTranscriptReader(target).readPage({ ...opts, resetArchiveOnly: true });
|
||||
}
|
||||
@@ -409,7 +433,7 @@ export async function readSessionMessagesPageWithStatsAsync(
|
||||
...(Object.hasOwn(page, "activeLeafEntryId")
|
||||
? { activeLeafEntryId: page.activeLeafEntryId }
|
||||
: {}),
|
||||
messages: extractMessageRecordsFromEventEntries(page.events).map(sqliteRecordMessageWithSeq),
|
||||
messages: projectSqliteHistoryEvents(page.events),
|
||||
transcriptEvents: page.events.map((entry) => entry.event),
|
||||
totalMessages: page.totalMessages,
|
||||
transcriptPath: target.sessionFile,
|
||||
|
||||
@@ -458,31 +458,149 @@ describe("readSessionMessages", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("applies reset kept-tail projection to file-backed history", async () => {
|
||||
const sessionId = "test-session-reset-boundary";
|
||||
writeTranscript(tmpDir, sessionId, [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
createTranscriptMessage("old", null, "user", "old"),
|
||||
createTranscriptMessage("kept-user", "old", "user", "kept question"),
|
||||
createTranscriptMessage("kept-tool", "kept-user", "toolResult", "hidden tool"),
|
||||
createTranscriptMessage("kept-assistant", "kept-tool", "assistant", "kept answer"),
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: "kept-assistant",
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
reason: "new",
|
||||
firstKeptEntryId: "kept-user",
|
||||
},
|
||||
createTranscriptMessage("post-reset", "reset-boundary", "user", "new turn"),
|
||||
]);
|
||||
test.each([
|
||||
{
|
||||
name: "keeps the reset marker first when no earlier entries survive",
|
||||
sessionId: "test-session-reset-no-kept",
|
||||
entries: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
createTranscriptMessage("old", null, "user", "old"),
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: "old",
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
reason: "reset",
|
||||
},
|
||||
createTranscriptMessage("post-reset", "reset-boundary", "user", "new turn"),
|
||||
],
|
||||
expected: ["reset", "new turn"],
|
||||
},
|
||||
{
|
||||
name: "places the reset marker between retained and new turns",
|
||||
sessionId: "test-session-reset-kept-tail",
|
||||
entries: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
createTranscriptMessage("old", null, "user", "old"),
|
||||
createTranscriptMessage("kept-user", "old", "user", "kept question"),
|
||||
createTranscriptMessage("kept-tool", "kept-user", "toolResult", "hidden tool"),
|
||||
createTranscriptMessage("kept-assistant", "kept-tool", "assistant", "kept answer"),
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: "kept-assistant",
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
reason: "new",
|
||||
firstKeptEntryId: "kept-user",
|
||||
},
|
||||
createTranscriptMessage("post-reset", "reset-boundary", "user", "new turn"),
|
||||
],
|
||||
expected: ["kept question", "kept answer", "reset", "new turn"],
|
||||
},
|
||||
{
|
||||
name: "drops an earlier compaction marker at a later reset boundary",
|
||||
sessionId: "test-session-compaction-then-reset",
|
||||
entries: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
createTranscriptMessage("old", null, "user", "old"),
|
||||
{
|
||||
type: "compaction",
|
||||
id: "compaction-before-reset",
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
},
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: "compaction-before-reset",
|
||||
timestamp: "2026-07-22T00:01:00.000Z",
|
||||
reason: "reset",
|
||||
},
|
||||
createTranscriptMessage("post-reset", "reset-boundary", "assistant", "new answer"),
|
||||
],
|
||||
expected: ["reset", "new answer"],
|
||||
},
|
||||
{
|
||||
name: "preserves reset then compaction marker order",
|
||||
sessionId: "test-session-reset-then-compaction",
|
||||
entries: (sessionId: string) => [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-boundary",
|
||||
parentId: null,
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
reason: "reset",
|
||||
},
|
||||
createTranscriptMessage("post-reset", "reset-boundary", "user", "new turn"),
|
||||
{
|
||||
type: "compaction",
|
||||
id: "compaction-after-reset",
|
||||
parentId: "post-reset",
|
||||
timestamp: "2026-07-22T00:01:00.000Z",
|
||||
},
|
||||
createTranscriptMessage(
|
||||
"post-compaction",
|
||||
"compaction-after-reset",
|
||||
"assistant",
|
||||
"new answer",
|
||||
),
|
||||
],
|
||||
expected: ["reset", "new turn", "compaction", "new answer"],
|
||||
},
|
||||
])("$name", async ({ sessionId, entries, expected }) => {
|
||||
writeTranscript(tmpDir, sessionId, entries(sessionId));
|
||||
|
||||
const messages = await readSessionMessagesAsync(sessionId, storePath, undefined, {
|
||||
const project = (messages: unknown[]) =>
|
||||
messages.map((message) => {
|
||||
const record = message as { content?: unknown; __openclaw?: { kind?: string } };
|
||||
return record["__openclaw"]?.kind ?? record.content;
|
||||
});
|
||||
const full = await readSessionMessagesAsync(sessionId, storePath, undefined, {
|
||||
mode: "full",
|
||||
reason: "test reset boundary",
|
||||
});
|
||||
const recent = await readRecentSessionMessagesWithStatsAsync(sessionId, storePath, undefined, {
|
||||
maxMessages: 10,
|
||||
maxBytes: 16_384,
|
||||
});
|
||||
|
||||
expectMessageContents(messages, ["kept question", "kept answer", "new turn"]);
|
||||
expect(project(full)).toEqual(expected);
|
||||
expect(project(recent.messages)).toEqual(expected);
|
||||
});
|
||||
|
||||
test("keeps reset markers reachable through pagination", async () => {
|
||||
const sessionId = "paginated-branch-with-reset";
|
||||
const sessionFile = writeTranscript(tmpDir, sessionId, [
|
||||
{ type: "session", version: 3, id: sessionId },
|
||||
createTranscriptMessage("old-user", null, "user", "old prompt"),
|
||||
{
|
||||
type: "reset",
|
||||
id: "reset-1",
|
||||
parentId: "old-user",
|
||||
timestamp: "2026-07-22T00:00:00.000Z",
|
||||
reason: "reset",
|
||||
},
|
||||
createTranscriptMessage("active-user", "reset-1", "user", "active prompt"),
|
||||
createTranscriptMessage("active-assistant", "active-user", "assistant", "active answer"),
|
||||
]);
|
||||
const newest = await readSessionMessagesPageWithStatsAsync(sessionId, storePath, sessionFile, {
|
||||
offset: 0,
|
||||
maxMessages: 2,
|
||||
});
|
||||
const oldest = await readSessionMessagesPageWithStatsAsync(sessionId, storePath, sessionFile, {
|
||||
offset: 2,
|
||||
maxMessages: 1,
|
||||
});
|
||||
|
||||
expect(newest.totalMessages).toBe(3);
|
||||
expectMessageFields(newest.messages[0], { content: "active prompt", openclaw: { seq: 2 } });
|
||||
expectMessageFields(newest.messages[1], { content: "active answer", openclaw: { seq: 3 } });
|
||||
expect(oldest.totalMessages).toBe(3);
|
||||
expectMessageFields(oldest.messages[0], {
|
||||
role: "system",
|
||||
content: [{ type: "text", text: "Reset" }],
|
||||
openclaw: { kind: "reset", id: "reset-1", seq: 1 },
|
||||
});
|
||||
});
|
||||
|
||||
test("keeps parentless linear history after a leaf control", async () => {
|
||||
|
||||
@@ -39,40 +39,12 @@ import {
|
||||
extractJsonStringFieldPrefix,
|
||||
readNonBlankStringPreservingWhitespace,
|
||||
} from "./session-transcript-json.js";
|
||||
import {
|
||||
attachOpenClawTranscriptMeta,
|
||||
projectTranscriptEntryMessage,
|
||||
} from "./session-transcript-message.js";
|
||||
import type { SessionPreviewItem } from "./session-utils.types.js";
|
||||
|
||||
/** Attach OpenClaw metadata to a transcript message without dropping existing metadata. */
|
||||
export function attachOpenClawTranscriptMeta(
|
||||
message: unknown,
|
||||
meta: Record<string, unknown>,
|
||||
): unknown {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return message;
|
||||
}
|
||||
const record = message as Record<string, unknown>;
|
||||
const existing =
|
||||
record["__openclaw"] &&
|
||||
typeof record["__openclaw"] === "object" &&
|
||||
!Array.isArray(record["__openclaw"])
|
||||
? (record["__openclaw"] as Record<string, unknown>)
|
||||
: {};
|
||||
return {
|
||||
...record,
|
||||
__openclaw: {
|
||||
...existing,
|
||||
...meta,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function readTranscriptMessageIdempotencyKey(message: unknown): string | undefined {
|
||||
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
||||
return undefined;
|
||||
}
|
||||
const value = (message as Record<string, unknown>).idempotencyKey;
|
||||
return typeof value === "string" && value.trim() ? value : undefined;
|
||||
}
|
||||
|
||||
export type ReadRecentSessionMessagesOptions = {
|
||||
maxMessages: number;
|
||||
maxBytes?: number;
|
||||
@@ -469,14 +441,16 @@ function parseRecentTranscriptTailSnapshot(
|
||||
const entry = parseTranscriptRecord(line);
|
||||
return entry ? [entry] : [];
|
||||
});
|
||||
const selected = selectSessionTranscriptActiveEntries({
|
||||
entries,
|
||||
recordOf: (entry) => entry.record,
|
||||
failClosedOnInvalidLeafControl: true,
|
||||
});
|
||||
const selected = projectResetBoundary(
|
||||
selectSessionTranscriptActiveEntries({
|
||||
entries,
|
||||
recordOf: (entry) => entry.record,
|
||||
failClosedOnInvalidLeafControl: true,
|
||||
}),
|
||||
);
|
||||
const messages: unknown[] = [];
|
||||
for (const entry of selected) {
|
||||
const message = parsedSessionEntryToMessage(entry.record, messages.length + 1);
|
||||
const message = projectTranscriptEntryMessage(entry.record, messages.length + 1);
|
||||
if (message) {
|
||||
messages.push(message);
|
||||
}
|
||||
@@ -488,7 +462,7 @@ function parseRecentTranscriptTailSnapshot(
|
||||
}
|
||||
|
||||
function isVisibleTranscriptRecord(record: Record<string, unknown>): boolean {
|
||||
return Boolean(record.message) || record.type === "compaction";
|
||||
return Boolean(record.message) || record.type === "compaction" || record.type === "reset";
|
||||
}
|
||||
|
||||
function projectResetBoundary(entries: TranscriptRecord[]): TranscriptRecord[] {
|
||||
@@ -510,7 +484,7 @@ function projectResetBoundary(entries: TranscriptRecord[]): TranscriptRecord[] {
|
||||
const role = (record.message as { role?: unknown } | undefined)?.role;
|
||||
return role === "user" || role === "assistant";
|
||||
});
|
||||
return [...kept, ...entries.slice(boundaryIndex + 1)];
|
||||
return [...kept, ...entries.slice(boundaryIndex)];
|
||||
}
|
||||
|
||||
function toIndexedEntries(entries: TranscriptRecord[]): IndexedTranscriptEntry[] {
|
||||
@@ -861,48 +835,8 @@ async function readRecentSessionSnapshotFromPathAsync(
|
||||
return parseRecentTranscriptTailSnapshot(lines, maxMessages);
|
||||
}
|
||||
|
||||
function parsedSessionEntryToMessage(parsed: unknown, seq: number): unknown {
|
||||
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
|
||||
return null;
|
||||
}
|
||||
const entry = parsed as Record<string, unknown>;
|
||||
if (entry.message) {
|
||||
const recordTimestampMs =
|
||||
typeof entry.timestamp === "string"
|
||||
? Date.parse(entry.timestamp)
|
||||
: typeof entry.timestamp === "number"
|
||||
? entry.timestamp
|
||||
: Number.NaN;
|
||||
const idempotencyKey = readTranscriptMessageIdempotencyKey(entry.message);
|
||||
return attachOpenClawTranscriptMeta(entry.message, {
|
||||
...(typeof entry.id === "string" ? { id: entry.id } : {}),
|
||||
...(idempotencyKey ? { idempotencyKey } : {}),
|
||||
...(Number.isFinite(recordTimestampMs) ? { recordTimestampMs } : {}),
|
||||
seq,
|
||||
});
|
||||
}
|
||||
|
||||
// Compaction entries are not "message" records, but they're useful context for debugging.
|
||||
// Emit a lightweight synthetic message that the Web UI can render as a divider.
|
||||
if (entry.type === "compaction") {
|
||||
const ts = typeof entry.timestamp === "string" ? Date.parse(entry.timestamp) : Number.NaN;
|
||||
const timestamp = Number.isFinite(ts) ? ts : Date.now();
|
||||
return {
|
||||
role: "system",
|
||||
content: [{ type: "text", text: "Compaction" }],
|
||||
timestamp,
|
||||
__openclaw: {
|
||||
kind: "compaction",
|
||||
id: typeof entry.id === "string" ? entry.id : undefined,
|
||||
seq,
|
||||
},
|
||||
};
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function indexedTranscriptEntryToMessage(entry: IndexedTranscriptEntry): unknown {
|
||||
return parsedSessionEntryToMessage(entry.record, entry.seq);
|
||||
return projectTranscriptEntryMessage(entry.record, entry.seq);
|
||||
}
|
||||
|
||||
function indexedTranscriptEntryToMessages(entry: IndexedTranscriptEntry): unknown[] {
|
||||
|
||||
@@ -219,6 +219,8 @@ export const toolIcons = {
|
||||
<path d="m9 12 2 2 4-4" />`),
|
||||
refresh: strokeIcon(svg` <path d="M21 12a9 9 0 1 1-9-9c2.52 0 4.93 1 6.74 2.74L21 8" />
|
||||
<path d="M21 3v5h-5" />`),
|
||||
rotateCcw: strokeIcon(svg`<path d="M3 12a9 9 0 1 0 9-9 9.75 9.75 0 0 0-6.74 2.74L3 8" />
|
||||
<path d="M3 3v5h5" />`),
|
||||
trash: strokeIcon(svg` <path d="M3 6h18" />
|
||||
<path d="M19 6v14c0 1-1 2-2 2H7c-1 0-2-1-2-2V6" />
|
||||
<path d="M8 6V4c0-1 1-2 2-2h4c1 0 2 1 2 2v2" />
|
||||
|
||||
@@ -4770,6 +4770,10 @@ export const en: TranslationMap = {
|
||||
description: "The compacted transcript is preserved as a checkpoint.",
|
||||
openCheckpoints: "Open checkpoints",
|
||||
},
|
||||
sessionReset: {
|
||||
label: "Session reset",
|
||||
description: "The earlier conversation was cleared.",
|
||||
},
|
||||
systemNotice: {
|
||||
restartRecovery: {
|
||||
label: "System · restart recovery",
|
||||
|
||||
@@ -50,6 +50,24 @@ export function buildCompactionDividerItem(
|
||||
};
|
||||
}
|
||||
|
||||
export function buildResetDividerItem(
|
||||
marker: Record<string, unknown>,
|
||||
timestamp: number,
|
||||
index: number,
|
||||
): Extract<ChatItem, { kind: "divider" }> {
|
||||
return {
|
||||
kind: "divider",
|
||||
key:
|
||||
typeof marker.id === "string"
|
||||
? `divider:reset:${marker.id}`
|
||||
: `divider:reset:${timestamp}:${index}`,
|
||||
label: t("chat.sessionReset.label"),
|
||||
icon: "rotateCcw",
|
||||
description: t("chat.sessionReset.description"),
|
||||
timestamp,
|
||||
};
|
||||
}
|
||||
|
||||
export function shouldRenderQueuedSendInThread(item: ChatQueueItem): boolean {
|
||||
// Page-local submit timing is not persisted; durable attempts keep restored prompts visible.
|
||||
const sendStarted = typeof item.sendSubmittedAtMs === "number" || (item.sendAttempts ?? 0) > 0;
|
||||
|
||||
@@ -23,6 +23,7 @@ import { areUiSessionKeysEquivalent } from "../../lib/sessions/session-key.ts";
|
||||
import { normalizeOptionalString } from "../../lib/string-coerce.ts";
|
||||
import {
|
||||
buildCompactionDividerItem,
|
||||
buildResetDividerItem,
|
||||
clearWorkingProgress,
|
||||
resolveWorkingProgress,
|
||||
shouldRenderQueuedSendInThread,
|
||||
@@ -231,6 +232,10 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
|
||||
items.push(buildCompactionDividerItem(marker, normalized.timestamp ?? Date.now(), i));
|
||||
continue;
|
||||
}
|
||||
if (marker && marker.kind === "reset") {
|
||||
items.push(buildResetDividerItem(marker, normalized.timestamp ?? Date.now(), i));
|
||||
continue;
|
||||
}
|
||||
|
||||
const role = normalizeRoleForGrouping(normalized.role);
|
||||
if (role === "system") {
|
||||
|
||||
@@ -169,6 +169,14 @@ function compactionMessage(id: string, metrics: Record<string, unknown> = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
function resetMessage(id: string) {
|
||||
return {
|
||||
role: "system",
|
||||
timestamp: 2_000,
|
||||
__openclaw: { kind: "reset", id },
|
||||
};
|
||||
}
|
||||
|
||||
function canvasToolOutput(viewId: string, title: string, preferredHeight: number): string {
|
||||
return JSON.stringify({
|
||||
kind: "canvas",
|
||||
@@ -3504,6 +3512,25 @@ describe("buildCachedChatItems", () => {
|
||||
metric: "saved 875.3k tokens",
|
||||
});
|
||||
});
|
||||
|
||||
it("explains reset boundaries without compaction-only details", () => {
|
||||
const items = buildCachedChatItems(
|
||||
createProps({
|
||||
messages: [resetMessage("reset-1")],
|
||||
}),
|
||||
);
|
||||
|
||||
expect(items).toHaveLength(1);
|
||||
expect(items[0]).toMatchObject({
|
||||
kind: "divider",
|
||||
key: "divider:reset:reset-1",
|
||||
label: "Session reset",
|
||||
icon: "rotateCcw",
|
||||
description: "The earlier conversation was cleared.",
|
||||
});
|
||||
expect(items[0]).not.toHaveProperty("metric");
|
||||
expect(items[0]).not.toHaveProperty("action");
|
||||
});
|
||||
});
|
||||
|
||||
describe("tool expansion state", () => {
|
||||
|
||||
@@ -92,14 +92,25 @@ const buildChatItemsMock = vi.fn(
|
||||
runWorking?: boolean;
|
||||
loading?: boolean;
|
||||
}): ReturnType<typeof chatThread.buildCachedChatItems> => {
|
||||
if (
|
||||
props.messages.some(
|
||||
(message) =>
|
||||
typeof message === "object" &&
|
||||
message !== null &&
|
||||
(message as { __testDivider?: unknown })["__testDivider"] === true,
|
||||
)
|
||||
) {
|
||||
const testDivider = props.messages.find(
|
||||
(message) =>
|
||||
typeof message === "object" &&
|
||||
message !== null &&
|
||||
typeof (message as { testDividerMarker?: unknown }).testDividerMarker === "string",
|
||||
) as { testDividerMarker: string } | undefined;
|
||||
if (testDivider) {
|
||||
if (testDivider.testDividerMarker === "reset") {
|
||||
return [
|
||||
{
|
||||
kind: "divider",
|
||||
key: "divider:reset:test",
|
||||
icon: "rotateCcw",
|
||||
label: "Session reset",
|
||||
description: "The earlier conversation was cleared.",
|
||||
timestamp: 1,
|
||||
},
|
||||
] as ReturnType<typeof chatThread.buildCachedChatItems>;
|
||||
}
|
||||
return [
|
||||
{
|
||||
kind: "divider",
|
||||
@@ -894,7 +905,7 @@ describe("chat compaction divider", () => {
|
||||
it("renders checkpoint recovery copy and action", () => {
|
||||
const onOpenSessionCheckpoints = vi.fn();
|
||||
const container = renderChatView({
|
||||
messages: [{ __testDivider: true }],
|
||||
messages: [{ testDividerMarker: "compaction" }],
|
||||
onOpenSessionCheckpoints,
|
||||
});
|
||||
|
||||
@@ -911,6 +922,14 @@ describe("chat compaction divider", () => {
|
||||
|
||||
expect(onOpenSessionCheckpoints).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("renders the session reset divider title", () => {
|
||||
const container = renderChatView({
|
||||
messages: [{ testDividerMarker: "reset" }],
|
||||
});
|
||||
|
||||
expect(container.querySelector(".chat-divider__title")?.textContent).toBe("Session reset");
|
||||
});
|
||||
});
|
||||
|
||||
describe("cloud workspace conflict notice", () => {
|
||||
|
||||
Reference in New Issue
Block a user