diff --git a/.github/pr-assets/compaction-checkpoints/sessions-checkpoints-inline.png b/.github/pr-assets/compaction-checkpoints/sessions-checkpoints-inline.png new file mode 100644 index 000000000000..ec61f9603b79 Binary files /dev/null and b/.github/pr-assets/compaction-checkpoints/sessions-checkpoints-inline.png differ diff --git a/.github/pr-assets/compaction-checkpoints/sessions-overview-inline.png b/.github/pr-assets/compaction-checkpoints/sessions-overview-inline.png new file mode 100644 index 000000000000..6efc15fc0708 Binary files /dev/null and b/.github/pr-assets/compaction-checkpoints/sessions-overview-inline.png differ diff --git a/CHANGELOG.md b/CHANGELOG.md index 336d2fd2ff3e..513c6aa7196c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ Docs: https://docs.openclaw.ai - ACP/ACPX plugin: bump the bundled `acpx` pin to `0.5.1` so plugin-local installs and strict version checks pick up the latest published runtime release. (#62148) Thanks @onutc. - Tools/media generation: auto-fallback across auth-backed image, music, and video providers by default, and remap fallback size, aspect ratio, resolution, and duration hints to the closest supported option instead of dropping intent on provider switches. - Tools/media generation: report applied fallback geometry and duration settings consistently in tool results, add a shared normalization contract for image/music/video runtimes, and simplify the bundled image-generation-core runtime test to only verify the plugin-sdk re-export seam. +- Gateway/sessions: add persisted compaction checkpoints plus Sessions UI branch/restore actions so operators can inspect and recover pre-compaction session state. (#62146) Thanks @scoootscooob. ### Fixes diff --git a/src/agents/pi-embedded-runner/compact.ts b/src/agents/pi-embedded-runner/compact.ts index b768b33b0076..06306b498744 100644 --- a/src/agents/pi-embedded-runner/compact.ts +++ b/src/agents/pi-embedded-runner/compact.ts @@ -15,6 +15,13 @@ import { ensureContextEnginesInitialized, resolveContextEngine, } from "../../context-engine/index.js"; +import { + captureCompactionCheckpointSnapshot, + cleanupCompactionCheckpointSnapshot, + persistSessionCompactionCheckpoint, + resolveSessionCompactionCheckpointReason, + type CapturedCompactionCheckpointSnapshot, +} from "../../gateway/session-compaction-checkpoints.js"; import { resolveHeartbeatSummaryForAgent } from "../../infra/heartbeat-summary.js"; import { getMachineDisplayName } from "../../infra/machine-name.js"; import { generateSecureToken } from "../../infra/secure-random.js"; @@ -108,6 +115,7 @@ import { applyExtraParamsToAgent } from "./extra-params.js"; import { getDmHistoryLimitFromSessionKey, limitHistoryTurns } from "./history.js"; import { resolveGlobalLane, resolveSessionLane } from "./lanes.js"; import { log } from "./logger.js"; +import { hardenManualCompactionBoundary } from "./manual-compaction-boundary.js"; import { buildEmbeddedMessageActionDiscoveryInput } from "./message-action-discovery-input.js"; import { readPiModelContextTokens } from "./model-context-tokens.js"; import { buildModelAliasLines, resolveModelAsync } from "./model.js"; @@ -415,6 +423,8 @@ export async function compactEmbeddedPiSessionDirect( let restoreSkillEnv: (() => void) | undefined; let compactionSessionManager: unknown = null; + let checkpointSnapshot: CapturedCompactionCheckpointSnapshot | null = null; + let checkpointSnapshotRetained = false; try { const { shouldLoadSkillEntries, skillEntries } = resolveEmbeddedRunSkillEntries({ workspaceDir: effectiveWorkspace, @@ -730,6 +740,10 @@ export async function compactEmbeddedPiSessionDirect( allowSyntheticToolResults: transcriptPolicy.allowSyntheticToolResults, allowedToolNames, }); + checkpointSnapshot = captureCompactionCheckpointSnapshot({ + sessionManager, + sessionFile: params.sessionFile, + }); compactionSessionManager = sessionManager; trackSessionManagerAccess(params.sessionFile); const settingsManager = createPreparedEmbeddedPiSettingsManager({ @@ -960,6 +974,28 @@ export async function compactEmbeddedPiSessionDirect( sessionKey: params.sessionKey, sessionFile: params.sessionFile, }); + let effectiveFirstKeptEntryId = result.firstKeptEntryId; + let postCompactionLeafId = + typeof sessionManager.getLeafId === "function" + ? (sessionManager.getLeafId() ?? undefined) + : undefined; + if (params.trigger === "manual") { + try { + const hardenedBoundary = await hardenManualCompactionBoundary({ + sessionFile: params.sessionFile, + }); + if (hardenedBoundary.applied) { + effectiveFirstKeptEntryId = + hardenedBoundary.firstKeptEntryId ?? effectiveFirstKeptEntryId; + postCompactionLeafId = hardenedBoundary.leafId ?? postCompactionLeafId; + session.agent.state.messages = hardenedBoundary.messages; + } + } catch (err) { + log.warn("[compaction] failed to harden manual compaction boundary", { + errorMessage: err instanceof Error ? err.message : String(err), + }); + } + } // Estimate tokens after compaction by summing token estimates for remaining messages const tokensAfter = estimateTokensAfterCompaction({ messagesAfter: session.messages, @@ -969,6 +1005,32 @@ export async function compactEmbeddedPiSessionDirect( }); const messageCountAfter = session.messages.length; const compactedCount = Math.max(0, messageCountCompactionInput - messageCountAfter); + if (params.config && params.sessionKey && checkpointSnapshot) { + try { + const storedCheckpoint = await persistSessionCompactionCheckpoint({ + cfg: params.config, + sessionKey: params.sessionKey, + sessionId: params.sessionId, + reason: resolveSessionCompactionCheckpointReason({ + trigger: params.trigger, + }), + snapshot: checkpointSnapshot, + summary: result.summary, + firstKeptEntryId: effectiveFirstKeptEntryId, + tokensBefore: observedTokenCount ?? result.tokensBefore, + tokensAfter, + postSessionFile: params.sessionFile, + postLeafId: postCompactionLeafId, + postEntryId: postCompactionLeafId, + createdAt: compactStartedAt, + }); + checkpointSnapshotRetained = storedCheckpoint !== null; + } catch (err) { + log.warn("failed to persist compaction checkpoint", { + errorMessage: err instanceof Error ? err.message : String(err), + }); + } + } const postMetrics = diagEnabled ? summarizeCompactionMessages(session.messages) : undefined; @@ -1000,7 +1062,7 @@ export async function compactEmbeddedPiSessionDirect( sessionFile: params.sessionFile, summaryLength: typeof result.summary === "string" ? result.summary.length : undefined, tokensBefore: result.tokensBefore, - firstKeptEntryId: result.firstKeptEntryId, + firstKeptEntryId: effectiveFirstKeptEntryId, }); // Truncate session file to remove compacted entries (#39953) if (params.config?.agents?.defaults?.compaction?.truncateAfterCompaction) { @@ -1032,7 +1094,7 @@ export async function compactEmbeddedPiSessionDirect( compacted: true, result: { summary: result.summary, - firstKeptEntryId: result.firstKeptEntryId, + firstKeptEntryId: effectiveFirstKeptEntryId, tokensBefore: observedTokenCount ?? result.tokensBefore, tokensAfter, details: result.details, @@ -1091,6 +1153,9 @@ export async function compactEmbeddedPiSessionDirect( }); return fail(reason); } finally { + if (!checkpointSnapshotRetained) { + await cleanupCompactionCheckpointSnapshot(checkpointSnapshot); + } restoreSkillEnv?.(); } } @@ -1116,6 +1181,8 @@ export async function compactEmbeddedPiSession( }); ensureContextEnginesInitialized(); const contextEngine = await resolveContextEngine(params.config); + let checkpointSnapshot: CapturedCompactionCheckpointSnapshot | null = null; + let checkpointSnapshotRetained = false; try { const agentDir = params.agentDir ?? resolveOpenClawAgentDir(); const resolvedCompactionTarget = resolveEmbeddedCompactionTarget({ @@ -1150,6 +1217,12 @@ export async function compactEmbeddedPiSession( // Fire before_compaction / after_compaction hooks here so plugin subscribers // are notified regardless of which engine is active. const engineOwnsCompaction = contextEngine.info.ownsCompaction === true; + checkpointSnapshot = engineOwnsCompaction + ? captureCompactionCheckpointSnapshot({ + sessionManager: SessionManager.open(params.sessionFile), + sessionFile: params.sessionFile, + }) + : null; const hookRunner = engineOwnsCompaction ? asCompactionHookRunner(getGlobalHookRunner()) : null; @@ -1222,6 +1295,33 @@ export async function compactEmbeddedPiSession( runtimeContext, }); if (result.ok && result.compacted) { + if (params.config && params.sessionKey && checkpointSnapshot) { + try { + const postCompactionSession = SessionManager.open(params.sessionFile); + const postLeafId = postCompactionSession.getLeafId() ?? undefined; + const storedCheckpoint = await persistSessionCompactionCheckpoint({ + cfg: params.config, + sessionKey: params.sessionKey, + sessionId: params.sessionId, + reason: resolveSessionCompactionCheckpointReason({ + trigger: params.trigger, + }), + snapshot: checkpointSnapshot, + summary: result.result?.summary, + firstKeptEntryId: result.result?.firstKeptEntryId, + tokensBefore: result.result?.tokensBefore, + tokensAfter: result.result?.tokensAfter, + postSessionFile: params.sessionFile, + postLeafId, + postEntryId: postLeafId, + }); + checkpointSnapshotRetained = storedCheckpoint !== null; + } catch (err) { + log.warn("failed to persist compaction checkpoint", { + errorMessage: err instanceof Error ? err.message : String(err), + }); + } + } await runContextEngineMaintenance({ contextEngine, sessionId: params.sessionId, @@ -1275,6 +1375,9 @@ export async function compactEmbeddedPiSession( : undefined, }; } finally { + if (!checkpointSnapshotRetained) { + await cleanupCompactionCheckpointSnapshot(checkpointSnapshot); + } await contextEngine.dispose?.(); } }), @@ -1287,6 +1390,7 @@ export const __testing = { containsRealConversationMessages, estimateTokensAfterCompaction, buildBeforeCompactionHookMetrics, + hardenManualCompactionBoundary, runBeforeCompactionHooks, runAfterCompactionHooks, runPostCompactionSideEffects, diff --git a/src/agents/pi-embedded-runner/manual-compaction-boundary.test.ts b/src/agents/pi-embedded-runner/manual-compaction-boundary.test.ts new file mode 100644 index 000000000000..eb977385ba49 --- /dev/null +++ b/src/agents/pi-embedded-runner/manual-compaction-boundary.test.ts @@ -0,0 +1,133 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { AgentMessage } from "@mariozechner/pi-agent-core"; +import type { AssistantMessage } from "@mariozechner/pi-ai"; +import { SessionManager } from "@mariozechner/pi-coding-agent"; +import { afterEach, describe, expect, it } from "vitest"; +import { hardenManualCompactionBoundary } from "./manual-compaction-boundary.js"; + +let tmpDir = ""; + +async function makeTmpDir(): Promise { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "manual-compaction-boundary-")); + return tmpDir; +} + +afterEach(async () => { + if (tmpDir) { + await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}); + tmpDir = ""; + } +}); + +function createAssistantTextMessage(text: string, timestamp: number): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "responses", + provider: "openai", + model: "gpt-test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop", + timestamp, + }; +} + +function messageText(message: AgentMessage): string { + if (!("content" in message)) { + return ""; + } + const content = message.content; + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .map((block) => + block && typeof block === "object" && "text" in block && typeof block.text === "string" + ? block.text + : "", + ) + .filter(Boolean) + .join(" "); +} + +describe("hardenManualCompactionBoundary", () => { + it("turns manual compaction into a true checkpoint for rebuilt context", async () => { + const dir = await makeTmpDir(); + const session = SessionManager.create(dir, dir); + + session.appendMessage({ role: "user", content: "old question", timestamp: 1 }); + session.appendMessage(createAssistantTextMessage("very long old answer", 2)); + const firstKeepId = session.getBranch().at(-1)?.id; + expect(firstKeepId).toBeTruthy(); + session.appendCompaction("old summary", firstKeepId!, 100); + + session.appendMessage({ role: "user", content: "new question", timestamp: 3 }); + session.appendMessage( + createAssistantTextMessage("detailed new answer that should be summarized away", 4), + ); + const secondKeepId = session.getBranch().at(-1)?.id; + expect(secondKeepId).toBeTruthy(); + const latestCompactionId = session.appendCompaction("fresh summary", secondKeepId!, 200); + const sessionFile = session.getSessionFile(); + expect(sessionFile).toBeTruthy(); + + const before = SessionManager.open(sessionFile!); + const beforeTexts = before + .buildSessionContext() + .messages.map((message) => messageText(message)); + expect(beforeTexts.join("\n")).toContain("detailed new answer"); + + const hardened = await hardenManualCompactionBoundary({ sessionFile: sessionFile! }); + expect(hardened.applied).toBe(true); + expect(hardened.firstKeptEntryId).toBe(latestCompactionId); + expect(hardened.messages.map((message) => message.role)).toEqual(["compactionSummary"]); + + const reopened = SessionManager.open(sessionFile!); + const latest = reopened.getLeafEntry(); + expect(latest?.type).toBe("compaction"); + if (!latest || latest.type !== "compaction") { + throw new Error("expected latest leaf to be a compaction entry"); + } + expect(latest.firstKeptEntryId).toBe(latestCompactionId); + + reopened.appendMessage({ role: "user", content: "what was happening?", timestamp: 5 }); + const after = SessionManager.open(sessionFile!); + const afterTexts = after.buildSessionContext().messages.map((message) => messageText(message)); + expect(after.buildSessionContext().messages.map((message) => message.role)).toEqual([ + "compactionSummary", + "user", + ]); + expect(afterTexts.join("\n")).not.toContain("detailed new answer"); + }); + + it("is a no-op when the latest leaf is not a compaction entry", async () => { + const dir = await makeTmpDir(); + const session = SessionManager.create(dir, dir); + session.appendMessage({ role: "user", content: "hello", timestamp: 1 }); + session.appendMessage(createAssistantTextMessage("hi", 2)); + const sessionFile = session.getSessionFile(); + expect(sessionFile).toBeTruthy(); + + const result = await hardenManualCompactionBoundary({ sessionFile: sessionFile! }); + expect(result.applied).toBe(false); + expect(result.messages.map((message) => message.role)).toEqual(["user", "assistant"]); + }); +}); diff --git a/src/agents/pi-embedded-runner/manual-compaction-boundary.ts b/src/agents/pi-embedded-runner/manual-compaction-boundary.ts new file mode 100644 index 000000000000..2fc8ef5f1eb8 --- /dev/null +++ b/src/agents/pi-embedded-runner/manual-compaction-boundary.ts @@ -0,0 +1,103 @@ +import fs from "node:fs/promises"; +import type { AgentMessage } from "@mariozechner/pi-agent-core"; +import { SessionManager } from "@mariozechner/pi-coding-agent"; + +type SessionManagerLike = ReturnType; +type SessionEntry = ReturnType[number]; +type SessionHeader = NonNullable>; +type CompactionEntry = Extract; + +export type HardenedManualCompactionBoundary = { + applied: boolean; + firstKeptEntryId?: string; + leafId?: string; + messages: AgentMessage[]; +}; + +function serializeSessionFile(header: SessionHeader, entries: SessionEntry[]): string { + return ( + [JSON.stringify(header), ...entries.map((entry) => JSON.stringify(entry))].join("\n") + "\n" + ); +} + +function replaceLatestCompactionBoundary(params: { + entries: SessionEntry[]; + compactionEntryId: string; +}): SessionEntry[] { + return params.entries.map((entry) => { + if (entry.type !== "compaction" || entry.id !== params.compactionEntryId) { + return entry; + } + return { + ...entry, + // Manual /compact is an explicit checkpoint request, so make the + // rebuilt context start from the summary itself instead of preserving + // an upstream "recent tail" that can keep large prior turns alive. + firstKeptEntryId: entry.id, + } satisfies CompactionEntry; + }); +} + +export async function hardenManualCompactionBoundary(params: { + sessionFile: string; +}): Promise { + const sessionManager = SessionManager.open(params.sessionFile) as Partial; + if ( + typeof sessionManager.getHeader !== "function" || + typeof sessionManager.getLeafEntry !== "function" || + typeof sessionManager.buildSessionContext !== "function" || + typeof sessionManager.getEntries !== "function" + ) { + return { + applied: false, + messages: [], + }; + } + + const header = sessionManager.getHeader(); + const leaf = sessionManager.getLeafEntry(); + if (!header || leaf?.type !== "compaction") { + const sessionContext = sessionManager.buildSessionContext(); + return { + applied: false, + leafId: + typeof sessionManager.getLeafId === "function" + ? (sessionManager.getLeafId() ?? undefined) + : undefined, + messages: sessionContext.messages, + }; + } + + if (leaf.firstKeptEntryId === leaf.id) { + const sessionContext = sessionManager.buildSessionContext(); + return { + applied: false, + firstKeptEntryId: leaf.id, + leafId: + typeof sessionManager.getLeafId === "function" + ? (sessionManager.getLeafId() ?? undefined) + : undefined, + messages: sessionContext.messages, + }; + } + + const content = serializeSessionFile( + header, + replaceLatestCompactionBoundary({ + entries: sessionManager.getEntries(), + compactionEntryId: leaf.id, + }), + ); + const tmpFile = `${params.sessionFile}.manual-compaction-tmp`; + await fs.writeFile(tmpFile, content, "utf-8"); + await fs.rename(tmpFile, params.sessionFile); + + const refreshed = SessionManager.open(params.sessionFile); + const sessionContext = refreshed.buildSessionContext(); + return { + applied: true, + firstKeptEntryId: leaf.id, + leafId: refreshed.getLeafId() ?? undefined, + messages: sessionContext.messages, + }; +} diff --git a/src/config/sessions/types.ts b/src/config/sessions/types.ts index bc37ee4899f1..c57d31ec8ed7 100644 --- a/src/config/sessions/types.ts +++ b/src/config/sessions/types.ts @@ -75,6 +75,33 @@ export type CliSessionBinding = { mcpConfigHash?: string; }; +export type SessionCompactionCheckpointReason = + | "manual" + | "auto-threshold" + | "overflow-retry" + | "timeout-retry"; + +export type SessionCompactionTranscriptReference = { + sessionId: string; + sessionFile?: string; + leafId?: string; + entryId?: string; +}; + +export type SessionCompactionCheckpoint = { + checkpointId: string; + sessionKey: string; + sessionId: string; + createdAt: number; + reason: SessionCompactionCheckpointReason; + tokensBefore?: number; + tokensAfter?: number; + summary?: string; + firstKeptEntryId?: string; + preCompaction: SessionCompactionTranscriptReference; + postCompaction: SessionCompactionTranscriptReference; +}; + export type SessionEntry = { /** * Last delivered heartbeat payload (used to suppress duplicate heartbeat notifications). @@ -182,6 +209,7 @@ export type SessionEntry = { fallbackNoticeReason?: string; contextTokens?: number; compactionCount?: number; + compactionCheckpoints?: SessionCompactionCheckpoint[]; memoryFlushAt?: number; memoryFlushCompactionCount?: number; memoryFlushContextHash?: string; diff --git a/src/gateway/method-scopes.ts b/src/gateway/method-scopes.ts index ac2abc110496..64c3702502fe 100644 --- a/src/gateway/method-scopes.ts +++ b/src/gateway/method-scopes.ts @@ -87,6 +87,8 @@ const METHOD_SCOPE_GROUPS: Record = { "sessions.get", "sessions.preview", "sessions.resolve", + "sessions.compaction.list", + "sessions.compaction.get", "sessions.subscribe", "sessions.unsubscribe", "sessions.messages.subscribe", @@ -129,6 +131,7 @@ const METHOD_SCOPE_GROUPS: Record = { "sessions.send", "sessions.steer", "sessions.abort", + "sessions.compaction.branch", "push.test", "node.pending.enqueue", ], @@ -149,6 +152,7 @@ const METHOD_SCOPE_GROUPS: Record = { "sessions.reset", "sessions.delete", "sessions.compact", + "sessions.compaction.restore", "connect", "chat.inject", "web.login.start", diff --git a/src/gateway/protocol/index.ts b/src/gateway/protocol/index.ts index e18b91df5615..692745d12b78 100644 --- a/src/gateway/protocol/index.ts +++ b/src/gateway/protocol/index.ts @@ -200,6 +200,14 @@ import { SessionsAbortParamsSchema, type SessionsCompactParams, SessionsCompactParamsSchema, + type SessionsCompactionBranchParams, + SessionsCompactionBranchParamsSchema, + type SessionsCompactionGetParams, + SessionsCompactionGetParamsSchema, + type SessionsCompactionListParams, + SessionsCompactionListParamsSchema, + type SessionsCompactionRestoreParams, + SessionsCompactionRestoreParamsSchema, type SessionsCreateParams, SessionsCreateParamsSchema, type SessionsDeleteParams, @@ -376,6 +384,18 @@ export const validateSessionsDeleteParams = ajv.compile( export const validateSessionsCompactParams = ajv.compile( SessionsCompactParamsSchema, ); +export const validateSessionsCompactionListParams = ajv.compile( + SessionsCompactionListParamsSchema, +); +export const validateSessionsCompactionGetParams = ajv.compile( + SessionsCompactionGetParamsSchema, +); +export const validateSessionsCompactionBranchParams = ajv.compile( + SessionsCompactionBranchParamsSchema, +); +export const validateSessionsCompactionRestoreParams = ajv.compile( + SessionsCompactionRestoreParamsSchema, +); export const validateSessionsUsageParams = ajv.compile(SessionsUsageParamsSchema); export const validateConfigGetParams = ajv.compile(ConfigGetParamsSchema); @@ -551,6 +571,10 @@ export { SessionsListParamsSchema, SessionsPreviewParamsSchema, SessionsResolveParamsSchema, + SessionsCompactionListParamsSchema, + SessionsCompactionGetParamsSchema, + SessionsCompactionBranchParamsSchema, + SessionsCompactionRestoreParamsSchema, SessionsCreateParamsSchema, SessionsSendParamsSchema, SessionsAbortParamsSchema, diff --git a/src/gateway/protocol/schema/protocol-schemas.ts b/src/gateway/protocol/schema/protocol-schemas.ts index 534dcaac816c..e04c18b9bbb0 100644 --- a/src/gateway/protocol/schema/protocol-schemas.ts +++ b/src/gateway/protocol/schema/protocol-schemas.ts @@ -155,6 +155,15 @@ import { import { SessionsAbortParamsSchema, SessionsCompactParamsSchema, + SessionsCompactionBranchParamsSchema, + SessionsCompactionBranchResultSchema, + SessionsCompactionGetParamsSchema, + SessionsCompactionGetResultSchema, + SessionsCompactionListParamsSchema, + SessionsCompactionListResultSchema, + SessionsCompactionRestoreParamsSchema, + SessionsCompactionRestoreResultSchema, + SessionCompactionCheckpointSchema, SessionsCreateParamsSchema, SessionsDeleteParamsSchema, SessionsListParamsSchema, @@ -224,6 +233,15 @@ export const ProtocolSchemas = { SessionsListParams: SessionsListParamsSchema, SessionsPreviewParams: SessionsPreviewParamsSchema, SessionsResolveParams: SessionsResolveParamsSchema, + SessionCompactionCheckpoint: SessionCompactionCheckpointSchema, + SessionsCompactionListParams: SessionsCompactionListParamsSchema, + SessionsCompactionGetParams: SessionsCompactionGetParamsSchema, + SessionsCompactionBranchParams: SessionsCompactionBranchParamsSchema, + SessionsCompactionRestoreParams: SessionsCompactionRestoreParamsSchema, + SessionsCompactionListResult: SessionsCompactionListResultSchema, + SessionsCompactionGetResult: SessionsCompactionGetResultSchema, + SessionsCompactionBranchResult: SessionsCompactionBranchResultSchema, + SessionsCompactionRestoreResult: SessionsCompactionRestoreResultSchema, SessionsCreateParams: SessionsCreateParamsSchema, SessionsSendParams: SessionsSendParamsSchema, SessionsMessagesSubscribeParams: SessionsMessagesSubscribeParamsSchema, diff --git a/src/gateway/protocol/schema/sessions.ts b/src/gateway/protocol/schema/sessions.ts index 5252e7c72cf1..7d8945ec1cd2 100644 --- a/src/gateway/protocol/schema/sessions.ts +++ b/src/gateway/protocol/schema/sessions.ts @@ -1,6 +1,40 @@ import { Type } from "@sinclair/typebox"; import { NonEmptyString, SessionLabelString } from "./primitives.js"; +export const SessionCompactionCheckpointReasonSchema = Type.Union([ + Type.Literal("manual"), + Type.Literal("auto-threshold"), + Type.Literal("overflow-retry"), + Type.Literal("timeout-retry"), +]); + +export const SessionCompactionTranscriptReferenceSchema = Type.Object( + { + sessionId: NonEmptyString, + sessionFile: Type.Optional(NonEmptyString), + leafId: Type.Optional(NonEmptyString), + entryId: Type.Optional(NonEmptyString), + }, + { additionalProperties: false }, +); + +export const SessionCompactionCheckpointSchema = Type.Object( + { + checkpointId: NonEmptyString, + sessionKey: NonEmptyString, + sessionId: NonEmptyString, + createdAt: Type.Integer({ minimum: 0 }), + reason: SessionCompactionCheckpointReasonSchema, + tokensBefore: Type.Optional(Type.Integer({ minimum: 0 })), + tokensAfter: Type.Optional(Type.Integer({ minimum: 0 })), + summary: Type.Optional(Type.String()), + firstKeptEntryId: Type.Optional(NonEmptyString), + preCompaction: SessionCompactionTranscriptReferenceSchema, + postCompaction: SessionCompactionTranscriptReferenceSchema, + }, + { additionalProperties: false }, +); + export const SessionsListParamsSchema = Type.Object( { limit: Type.Optional(Type.Integer({ minimum: 1 })), @@ -163,6 +197,90 @@ export const SessionsCompactParamsSchema = Type.Object( { additionalProperties: false }, ); +export const SessionsCompactionListParamsSchema = Type.Object( + { + key: NonEmptyString, + }, + { additionalProperties: false }, +); + +export const SessionsCompactionGetParamsSchema = Type.Object( + { + key: NonEmptyString, + checkpointId: NonEmptyString, + }, + { additionalProperties: false }, +); + +export const SessionsCompactionBranchParamsSchema = Type.Object( + { + key: NonEmptyString, + checkpointId: NonEmptyString, + }, + { additionalProperties: false }, +); + +export const SessionsCompactionRestoreParamsSchema = Type.Object( + { + key: NonEmptyString, + checkpointId: NonEmptyString, + }, + { additionalProperties: false }, +); + +export const SessionsCompactionListResultSchema = Type.Object( + { + ok: Type.Literal(true), + key: NonEmptyString, + checkpoints: Type.Array(SessionCompactionCheckpointSchema), + }, + { additionalProperties: false }, +); + +export const SessionsCompactionGetResultSchema = Type.Object( + { + ok: Type.Literal(true), + key: NonEmptyString, + checkpoint: SessionCompactionCheckpointSchema, + }, + { additionalProperties: false }, +); + +export const SessionsCompactionBranchResultSchema = Type.Object( + { + ok: Type.Literal(true), + sourceKey: NonEmptyString, + key: NonEmptyString, + sessionId: NonEmptyString, + checkpoint: SessionCompactionCheckpointSchema, + entry: Type.Object( + { + sessionId: NonEmptyString, + updatedAt: Type.Integer({ minimum: 0 }), + }, + { additionalProperties: true }, + ), + }, + { additionalProperties: false }, +); + +export const SessionsCompactionRestoreResultSchema = Type.Object( + { + ok: Type.Literal(true), + key: NonEmptyString, + sessionId: NonEmptyString, + checkpoint: SessionCompactionCheckpointSchema, + entry: Type.Object( + { + sessionId: NonEmptyString, + updatedAt: Type.Integer({ minimum: 0 }), + }, + { additionalProperties: true }, + ), + }, + { additionalProperties: false }, +); + export const SessionsUsageParamsSchema = Type.Object( { /** Specific session key to analyze; if omitted returns all sessions. */ diff --git a/src/gateway/protocol/schema/types.ts b/src/gateway/protocol/schema/types.ts index f3eb9725093a..0515ca140f61 100644 --- a/src/gateway/protocol/schema/types.ts +++ b/src/gateway/protocol/schema/types.ts @@ -41,6 +41,15 @@ export type PushTestResult = SchemaType<"PushTestResult">; export type SessionsListParams = SchemaType<"SessionsListParams">; export type SessionsPreviewParams = SchemaType<"SessionsPreviewParams">; export type SessionsResolveParams = SchemaType<"SessionsResolveParams">; +export type SessionCompactionCheckpoint = SchemaType<"SessionCompactionCheckpoint">; +export type SessionsCompactionListParams = SchemaType<"SessionsCompactionListParams">; +export type SessionsCompactionGetParams = SchemaType<"SessionsCompactionGetParams">; +export type SessionsCompactionBranchParams = SchemaType<"SessionsCompactionBranchParams">; +export type SessionsCompactionRestoreParams = SchemaType<"SessionsCompactionRestoreParams">; +export type SessionsCompactionListResult = SchemaType<"SessionsCompactionListResult">; +export type SessionsCompactionGetResult = SchemaType<"SessionsCompactionGetResult">; +export type SessionsCompactionBranchResult = SchemaType<"SessionsCompactionBranchResult">; +export type SessionsCompactionRestoreResult = SchemaType<"SessionsCompactionRestoreResult">; export type SessionsCreateParams = SchemaType<"SessionsCreateParams">; export type SessionsSendParams = SchemaType<"SessionsSendParams">; export type SessionsMessagesSubscribeParams = SchemaType<"SessionsMessagesSubscribeParams">; diff --git a/src/gateway/server-methods-list.ts b/src/gateway/server-methods-list.ts index ecc6a02d1227..0176150f9f8c 100644 --- a/src/gateway/server-methods-list.ts +++ b/src/gateway/server-methods-list.ts @@ -68,6 +68,10 @@ const BASE_METHODS = [ "sessions.messages.subscribe", "sessions.messages.unsubscribe", "sessions.preview", + "sessions.compaction.list", + "sessions.compaction.get", + "sessions.compaction.branch", + "sessions.compaction.restore", "sessions.create", "sessions.send", "sessions.abort", diff --git a/src/gateway/server-methods/agent.ts b/src/gateway/server-methods/agent.ts index 2d3658f09bc2..285002ea7bd2 100644 --- a/src/gateway/server-methods/agent.ts +++ b/src/gateway/server-methods/agent.ts @@ -177,6 +177,8 @@ function emitSessionsChanged( startedAt: sessionRow.startedAt, endedAt: sessionRow.endedAt, runtimeMs: sessionRow.runtimeMs, + compactionCheckpointCount: sessionRow.compactionCheckpointCount, + latestCompactionCheckpoint: sessionRow.latestCompactionCheckpoint, } : {}), }, diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index c865fc7131c3..10e85c19b0fc 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/src/gateway/server-methods/sessions.ts @@ -1,13 +1,15 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import path from "node:path"; -import { CURRENT_SESSION_VERSION } from "@mariozechner/pi-coding-agent"; -import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; +import { CURRENT_SESSION_VERSION, SessionManager } from "@mariozechner/pi-coding-agent"; +import { resolveAgentWorkspaceDir, resolveDefaultAgentId } from "../../agents/agent-scope.js"; import { abortEmbeddedPiRun, isEmbeddedPiRunActive, waitForEmbeddedPiRunEnd, } from "../../agents/pi-embedded-runner/runs.js"; +import { compactEmbeddedPiSession } from "../../agents/pi-embedded.js"; +import { normalizeReasoningLevel, normalizeThinkLevel } from "../../auto-reply/thinking.js"; import { clearSessionQueues } from "../../auto-reply/reply/queue/cleanup.js"; import { loadConfig } from "../../config/config.js"; import { @@ -36,6 +38,10 @@ import { errorShape, validateSessionsAbortParams, validateSessionsCompactParams, + validateSessionsCompactionBranchParams, + validateSessionsCompactionGetParams, + validateSessionsCompactionListParams, + validateSessionsCompactionRestoreParams, validateSessionsCreateParams, validateSessionsDeleteParams, validateSessionsListParams, @@ -47,6 +53,10 @@ import { validateSessionsResolveParams, validateSessionsSendParams, } from "../protocol/index.js"; +import { + getSessionCompactionCheckpoint, + listSessionCompactionCheckpoints, +} from "../session-compaction-checkpoints.js"; import { reactivateCompletedSubagentSession } from "../session-subagent-reactivation.js"; import { archiveFileOnDisk, @@ -185,6 +195,8 @@ function emitSessionsChanged( startedAt: sessionRow.startedAt, endedAt: sessionRow.endedAt, runtimeMs: sessionRow.runtimeMs, + compactionCheckpointCount: sessionRow.compactionCheckpointCount, + latestCompactionCheckpoint: sessionRow.latestCompactionCheckpoint, } : {}), }, @@ -220,6 +232,47 @@ function buildDashboardSessionKey(agentId: string): string { return `agent:${agentId}:dashboard:${randomUUID()}`; } +function cloneCheckpointSessionEntry(params: { + currentEntry: SessionEntry; + nextSessionId: string; + nextSessionFile: string; + label?: string; + parentSessionKey?: string; + totalTokens?: number; + preserveCompactionCheckpoints?: boolean; +}): SessionEntry { + return { + ...params.currentEntry, + sessionId: params.nextSessionId, + sessionFile: params.nextSessionFile, + updatedAt: Date.now(), + systemSent: false, + abortedLastRun: false, + startedAt: undefined, + endedAt: undefined, + runtimeMs: undefined, + status: undefined, + inputTokens: undefined, + outputTokens: undefined, + cacheRead: undefined, + cacheWrite: undefined, + estimatedCostUsd: undefined, + totalTokens: + typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) + ? params.totalTokens + : undefined, + totalTokensFresh: + typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) + ? true + : undefined, + label: params.label ?? params.currentEntry.label, + parentSessionKey: params.parentSessionKey ?? params.currentEntry.parentSessionKey, + compactionCheckpoints: params.preserveCompactionCheckpoints + ? params.currentEntry.compactionCheckpoints + : undefined, + }; +} + function ensureSessionTranscriptFile(params: { sessionId: string; storePath: string; @@ -644,6 +697,74 @@ export const sessionsHandlers: GatewayRequestHandlers = { } respond(true, { ok: true, key: resolved.key }, undefined); }, + "sessions.compaction.list": ({ params, respond }) => { + if ( + !assertValidParams( + params, + validateSessionsCompactionListParams, + "sessions.compaction.list", + respond, + ) + ) { + return; + } + const key = requireSessionKey((params as { key?: unknown }).key, respond); + if (!key) { + return; + } + const { entry, canonicalKey } = loadSessionEntry(key); + respond( + true, + { + ok: true, + key: canonicalKey, + checkpoints: listSessionCompactionCheckpoints(entry), + }, + undefined, + ); + }, + "sessions.compaction.get": ({ params, respond }) => { + if ( + !assertValidParams( + params, + validateSessionsCompactionGetParams, + "sessions.compaction.get", + respond, + ) + ) { + return; + } + const p = params; + const key = requireSessionKey(p.key, respond); + if (!key) { + return; + } + const checkpointId = + typeof p.checkpointId === "string" && p.checkpointId.trim() ? p.checkpointId.trim() : ""; + if (!checkpointId) { + respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "checkpointId required")); + return; + } + const { entry, canonicalKey } = loadSessionEntry(key); + const checkpoint = getSessionCompactionCheckpoint({ entry, checkpointId }); + if (!checkpoint) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `checkpoint not found: ${checkpointId}`), + ); + return; + } + respond( + true, + { + ok: true, + key: canonicalKey, + checkpoint, + }, + undefined, + ); + }, "sessions.create": async ({ req, params, respond, context, client, isWebchatConnect }) => { if (!assertValidParams(params, validateSessionsCreateParams, "sessions.create", respond)) { return; @@ -831,6 +952,228 @@ export const sessionsHandlers: GatewayRequestHandlers = { }); } }, + "sessions.compaction.branch": async ({ params, respond, context }) => { + if ( + !assertValidParams( + params, + validateSessionsCompactionBranchParams, + "sessions.compaction.branch", + respond, + ) + ) { + return; + } + const p = params; + const key = requireSessionKey(p.key, respond); + if (!key) { + return; + } + const checkpointId = + typeof p.checkpointId === "string" && p.checkpointId.trim() ? p.checkpointId.trim() : ""; + if (!checkpointId) { + respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "checkpointId required")); + return; + } + const loaded = loadSessionEntry(key); + const { cfg, entry, canonicalKey } = loaded; + const target = resolveGatewaySessionStoreTarget({ cfg, key: canonicalKey }); + if (!entry?.sessionId) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `session not found: ${key}`), + ); + return; + } + const checkpoint = getSessionCompactionCheckpoint({ entry, checkpointId }); + if (!checkpoint?.preCompaction.sessionFile) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `checkpoint not found: ${checkpointId}`), + ); + return; + } + if (!fs.existsSync(checkpoint.preCompaction.sessionFile)) { + respond( + false, + undefined, + errorShape(ErrorCodes.UNAVAILABLE, "checkpoint snapshot transcript is missing"), + ); + return; + } + + const snapshotSession = SessionManager.open( + checkpoint.preCompaction.sessionFile, + path.dirname(checkpoint.preCompaction.sessionFile), + ); + const branchedSession = SessionManager.forkFrom( + checkpoint.preCompaction.sessionFile, + snapshotSession.getCwd(), + path.dirname(checkpoint.preCompaction.sessionFile), + ); + const branchedSessionFile = branchedSession.getSessionFile(); + if (!branchedSessionFile) { + respond( + false, + undefined, + errorShape(ErrorCodes.UNAVAILABLE, "failed to create checkpoint branch transcript"), + ); + return; + } + const nextKey = buildDashboardSessionKey(target.agentId); + const label = entry.label?.trim() ? `${entry.label.trim()} (checkpoint)` : "Checkpoint branch"; + const nextEntry = cloneCheckpointSessionEntry({ + currentEntry: entry, + nextSessionId: branchedSession.getSessionId(), + nextSessionFile: branchedSessionFile, + label, + parentSessionKey: canonicalKey, + totalTokens: checkpoint.tokensBefore, + }); + + await updateSessionStore(target.storePath, (store) => { + store[nextKey] = nextEntry; + }); + + respond( + true, + { + ok: true, + sourceKey: canonicalKey, + key: nextKey, + sessionId: nextEntry.sessionId, + checkpoint, + entry: nextEntry, + }, + undefined, + ); + emitSessionsChanged(context, { + sessionKey: canonicalKey, + reason: "checkpoint-branch", + }); + emitSessionsChanged(context, { + sessionKey: nextKey, + reason: "checkpoint-branch", + }); + }, + "sessions.compaction.restore": async ({ + req, + params, + respond, + context, + client, + isWebchatConnect, + }) => { + if ( + !assertValidParams( + params, + validateSessionsCompactionRestoreParams, + "sessions.compaction.restore", + respond, + ) + ) { + return; + } + const p = params; + const key = requireSessionKey(p.key, respond); + if (!key) { + return; + } + const checkpointId = + typeof p.checkpointId === "string" && p.checkpointId.trim() ? p.checkpointId.trim() : ""; + if (!checkpointId) { + respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "checkpointId required")); + return; + } + const loaded = loadSessionEntry(key); + const { entry, canonicalKey, storePath } = loaded; + if (!entry?.sessionId) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `session not found: ${key}`), + ); + return; + } + const checkpoint = getSessionCompactionCheckpoint({ entry, checkpointId }); + if (!checkpoint?.preCompaction.sessionFile) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `checkpoint not found: ${checkpointId}`), + ); + return; + } + if (!fs.existsSync(checkpoint.preCompaction.sessionFile)) { + respond( + false, + undefined, + errorShape(ErrorCodes.UNAVAILABLE, "checkpoint snapshot transcript is missing"), + ); + return; + } + + const interruptResult = await interruptSessionRunIfActive({ + req, + context, + client, + isWebchatConnect, + requestedKey: key, + canonicalKey, + sessionId: entry.sessionId, + }); + if (interruptResult.error) { + respond(false, undefined, interruptResult.error); + return; + } + + const snapshotSession = SessionManager.open( + checkpoint.preCompaction.sessionFile, + path.dirname(checkpoint.preCompaction.sessionFile), + ); + const restoredSession = SessionManager.forkFrom( + checkpoint.preCompaction.sessionFile, + snapshotSession.getCwd(), + path.dirname(checkpoint.preCompaction.sessionFile), + ); + const restoredSessionFile = restoredSession.getSessionFile(); + if (!restoredSessionFile) { + respond( + false, + undefined, + errorShape(ErrorCodes.UNAVAILABLE, "failed to restore checkpoint transcript"), + ); + return; + } + const nextEntry = cloneCheckpointSessionEntry({ + currentEntry: entry, + nextSessionId: restoredSession.getSessionId(), + nextSessionFile: restoredSessionFile, + totalTokens: checkpoint.tokensBefore, + preserveCompactionCheckpoints: true, + }); + + await updateSessionStore(storePath, (store) => { + store[canonicalKey] = nextEntry; + }); + + respond( + true, + { + ok: true, + key: canonicalKey, + sessionId: nextEntry.sessionId, + checkpoint, + entry: nextEntry, + }, + undefined, + ); + emitSessionsChanged(context, { + sessionKey: canonicalKey, + reason: "checkpoint-restore", + }); + }, "sessions.send": async ({ req, params, respond, context, client, isWebchatConnect }) => { await handleSessionSend({ method: "sessions.send", @@ -1122,7 +1465,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { const messages = limit < allMessages.length ? allMessages.slice(-limit) : allMessages; respond(true, { messages }, undefined); }, - "sessions.compact": async ({ params, respond, context }) => { + "sessions.compact": async ({ req, params, respond, context, client, isWebchatConnect }) => { if (!assertValidParams(params, validateSessionsCompactParams, "sessions.compact", respond)) { return; } @@ -1135,7 +1478,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { const maxLines = typeof p.maxLines === "number" && Number.isFinite(p.maxLines) ? Math.max(1, Math.floor(p.maxLines)) - : 400; + : undefined; const { cfg, target, storePath } = resolveGatewaySessionTargetFromKey(key); // Lock + read in a short critical section; transcript work happens outside. @@ -1179,6 +1522,88 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } + if (maxLines === undefined) { + const interruptResult = await interruptSessionRunIfActive({ + req, + context, + client, + isWebchatConnect, + requestedKey: key, + canonicalKey: target.canonicalKey, + sessionId, + }); + if (interruptResult.error) { + respond(false, undefined, interruptResult.error); + return; + } + + const resolvedModel = resolveSessionModelRef(cfg, entry, target.agentId); + const workspaceDir = + entry?.spawnedWorkspaceDir?.trim() || resolveAgentWorkspaceDir(cfg, target.agentId); + const result = await compactEmbeddedPiSession({ + sessionId, + sessionKey: target.canonicalKey, + allowGatewaySubagentBinding: true, + sessionFile: filePath, + workspaceDir, + config: cfg, + provider: resolvedModel.provider, + model: resolvedModel.model, + thinkLevel: normalizeThinkLevel(entry?.thinkingLevel), + reasoningLevel: normalizeReasoningLevel(entry?.reasoningLevel), + bashElevated: { + enabled: false, + allowed: false, + defaultLevel: "off", + }, + trigger: "manual", + }); + + if (result.ok && result.compacted) { + await updateSessionStore(storePath, (store) => { + const entryKey = compactTarget.primaryKey; + const entryToUpdate = store[entryKey]; + if (!entryToUpdate) { + return; + } + entryToUpdate.updatedAt = Date.now(); + entryToUpdate.compactionCount = Math.max(0, entryToUpdate.compactionCount ?? 0) + 1; + delete entryToUpdate.inputTokens; + delete entryToUpdate.outputTokens; + if ( + typeof result.result?.tokensAfter === "number" && + Number.isFinite(result.result.tokensAfter) + ) { + entryToUpdate.totalTokens = result.result.tokensAfter; + entryToUpdate.totalTokensFresh = true; + } else { + delete entryToUpdate.totalTokens; + delete entryToUpdate.totalTokensFresh; + } + }); + } + + respond( + true, + { + ok: result.ok, + key: target.canonicalKey, + compacted: result.compacted, + reason: result.reason, + result: result.result, + }, + undefined, + ); + if (result.ok) { + emitSessionsChanged(context, { + sessionKey: target.canonicalKey, + reason: "compact", + compacted: result.compacted, + }); + } + return; + } + const raw = fs.readFileSync(filePath, "utf-8"); const lines = raw.split(/\r?\n/).filter((l) => l.trim().length > 0); if (lines.length <= maxLines) { diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index cd2688d9db7e..040493a22ffc 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -1059,6 +1059,8 @@ export async function startGatewayServer( startedAt: sessionRow.startedAt, endedAt: sessionRow.endedAt, runtimeMs: sessionRow.runtimeMs, + compactionCheckpointCount: sessionRow.compactionCheckpointCount, + latestCompactionCheckpoint: sessionRow.latestCompactionCheckpoint, } : {}; const message = attachOpenClawTranscriptMeta(update.message, { @@ -1160,6 +1162,8 @@ export async function startGatewayServer( startedAt: sessionRow.startedAt, endedAt: sessionRow.endedAt, runtimeMs: sessionRow.runtimeMs, + compactionCheckpointCount: sessionRow.compactionCheckpointCount, + latestCompactionCheckpoint: sessionRow.latestCompactionCheckpoint, } : {}), }, diff --git a/src/gateway/server.sessions.gateway-server-sessions-a.test.ts b/src/gateway/server.sessions.gateway-server-sessions-a.test.ts index 0187529449a9..6a04155f96a9 100644 --- a/src/gateway/server.sessions.gateway-server-sessions-a.test.ts +++ b/src/gateway/server.sessions.gateway-server-sessions-a.test.ts @@ -1,6 +1,9 @@ +import fsSync from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import type { AssistantMessage, UserMessage } from "@mariozechner/pi-ai"; +import { SessionManager } from "@mariozechner/pi-coding-agent"; import { afterAll, beforeAll, beforeEach, describe, expect, test, vi } from "vitest"; import { WebSocket } from "ws"; import { clearConfigCache, clearRuntimeConfigSnapshot } from "../config/config.js"; @@ -221,6 +224,68 @@ async function writeSingleLineSession(dir: string, sessionId: string, content: s ); } +function createCheckpointFixture(dir: string) { + const session = SessionManager.create(dir, dir); + const userMessage: UserMessage = { + role: "user", + content: "before compaction", + timestamp: Date.now(), + }; + const assistantMessage: AssistantMessage = { + role: "assistant", + content: [{ type: "text", text: "working on it" }], + api: "responses", + provider: "openai", + model: "gpt-test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; + session.appendMessage(userMessage); + session.appendMessage(assistantMessage); + const preCompactionLeafId = session.getLeafId(); + if (!preCompactionLeafId) { + throw new Error("expected persisted session leaf before compaction"); + } + const sessionFile = session.getSessionFile(); + if (!sessionFile) { + throw new Error("expected persisted session file"); + } + const preCompactionSessionFile = path.join( + dir, + `${path.parse(sessionFile).name}.checkpoint-test.jsonl`, + ); + fsSync.copyFileSync(sessionFile, preCompactionSessionFile); + const preCompactionSession = SessionManager.open(preCompactionSessionFile, dir); + session.appendCompaction("checkpoint summary", preCompactionLeafId, 123, { ok: true }); + const postCompactionLeafId = session.getLeafId(); + if (!postCompactionLeafId) { + throw new Error("expected post-compaction leaf"); + } + return { + session, + sessionId: session.getSessionId(), + sessionFile, + preCompactionSession, + preCompactionSessionFile, + preCompactionLeafId, + postCompactionLeafId, + }; +} + async function seedActiveMainSession() { const { dir, storePath } = await createSessionStoreDir(); await writeSingleLineSession(dir, "sess-main", "hello"); @@ -1227,6 +1292,211 @@ describe("gateway server sessions", () => { ws.close(); }); + test("sessions.compaction.* lists checkpoints and branches or restores from pre-compaction snapshots", async () => { + const { dir, storePath } = await createSessionStoreDir(); + const fixture = createCheckpointFixture(dir); + await writeSessionStore({ + entries: { + main: { + sessionId: fixture.sessionId, + sessionFile: fixture.sessionFile, + updatedAt: Date.now(), + compactionCheckpoints: [ + { + checkpointId: "checkpoint-1", + sessionKey: "agent:main:main", + sessionId: fixture.sessionId, + createdAt: Date.now(), + reason: "manual", + tokensBefore: 123, + tokensAfter: 45, + summary: "checkpoint summary", + firstKeptEntryId: fixture.preCompactionLeafId, + preCompaction: { + sessionId: fixture.preCompactionSession.getSessionId(), + sessionFile: fixture.preCompactionSessionFile, + leafId: fixture.preCompactionLeafId, + }, + postCompaction: { + sessionId: fixture.sessionId, + sessionFile: fixture.sessionFile, + leafId: fixture.postCompactionLeafId, + entryId: fixture.postCompactionLeafId, + }, + }, + ], + }, + }, + }); + + const { ws } = await openClient(); + + const listedSessions = await rpcReq<{ + sessions: Array<{ + key: string; + compactionCheckpointCount?: number; + latestCompactionCheckpoint?: { + checkpointId: string; + reason: string; + tokensBefore?: number; + tokensAfter?: number; + }; + }>; + }>(ws, "sessions.list", {}); + expect(listedSessions.ok).toBe(true); + const main = listedSessions.payload?.sessions.find( + (session) => session.key === "agent:main:main", + ); + expect(main?.compactionCheckpointCount).toBe(1); + expect(main?.latestCompactionCheckpoint?.checkpointId).toBe("checkpoint-1"); + expect(main?.latestCompactionCheckpoint?.reason).toBe("manual"); + + const listedCheckpoints = await rpcReq<{ + ok: true; + key: string; + checkpoints: Array<{ checkpointId: string; summary?: string; tokensBefore?: number }>; + }>(ws, "sessions.compaction.list", { key: "main" }); + expect(listedCheckpoints.ok).toBe(true); + expect(listedCheckpoints.payload?.key).toBe("agent:main:main"); + expect(listedCheckpoints.payload?.checkpoints).toHaveLength(1); + expect(listedCheckpoints.payload?.checkpoints[0]).toMatchObject({ + checkpointId: "checkpoint-1", + summary: "checkpoint summary", + tokensBefore: 123, + }); + + const checkpoint = await rpcReq<{ + ok: true; + key: string; + checkpoint: { checkpointId: string; preCompaction: { sessionFile: string } }; + }>(ws, "sessions.compaction.get", { + key: "main", + checkpointId: "checkpoint-1", + }); + expect(checkpoint.ok).toBe(true); + expect(checkpoint.payload?.checkpoint.checkpointId).toBe("checkpoint-1"); + expect(checkpoint.payload?.checkpoint.preCompaction.sessionFile).toBe( + fixture.preCompactionSessionFile, + ); + + const branched = await rpcReq<{ + ok: true; + sourceKey: string; + key: string; + entry: { sessionId: string; sessionFile?: string; parentSessionKey?: string }; + }>(ws, "sessions.compaction.branch", { + key: "main", + checkpointId: "checkpoint-1", + }); + expect(branched.ok).toBe(true); + expect(branched.payload?.sourceKey).toBe("agent:main:main"); + expect(branched.payload?.entry.parentSessionKey).toBe("agent:main:main"); + const branchedSessionFile = branched.payload?.entry.sessionFile; + expect(branchedSessionFile).toBeTruthy(); + const branchedSession = SessionManager.open(branchedSessionFile!, dir); + expect(branchedSession.getEntries()).toHaveLength( + fixture.preCompactionSession.getEntries().length, + ); + + const storeAfterBranch = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + { + parentSessionKey?: string; + compactionCheckpoints?: unknown[]; + sessionId?: string; + } + >; + const branchedEntry = storeAfterBranch[branched.payload!.key]; + expect(branchedEntry?.parentSessionKey).toBe("agent:main:main"); + expect(branchedEntry?.compactionCheckpoints).toBeUndefined(); + + const restored = await rpcReq<{ + ok: true; + key: string; + sessionId: string; + entry: { sessionId: string; sessionFile?: string; compactionCheckpoints?: unknown[] }; + }>(ws, "sessions.compaction.restore", { + key: "main", + checkpointId: "checkpoint-1", + }); + expect(restored.ok).toBe(true); + expect(restored.payload?.key).toBe("agent:main:main"); + expect(restored.payload?.sessionId).not.toBe(fixture.sessionId); + expect(restored.payload?.entry.compactionCheckpoints).toHaveLength(1); + const restoredSessionFile = restored.payload?.entry.sessionFile; + expect(restoredSessionFile).toBeTruthy(); + const restoredSession = SessionManager.open(restoredSessionFile!, dir); + expect(restoredSession.getEntries()).toHaveLength( + fixture.preCompactionSession.getEntries().length, + ); + + const storeAfterRestore = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + { compactionCheckpoints?: unknown[]; sessionId?: string } + >; + expect(storeAfterRestore["agent:main:main"]?.sessionId).toBe(restored.payload?.sessionId); + expect(storeAfterRestore["agent:main:main"]?.compactionCheckpoints).toHaveLength(1); + + ws.close(); + }); + + test("sessions.compact without maxLines runs embedded manual compaction for checkpoint-capable flows", async () => { + const { dir, storePath } = await createSessionStoreDir(); + await fs.writeFile( + path.join(dir, "sess-main.jsonl"), + `${JSON.stringify({ role: "user", content: "hello" })}\n`, + "utf-8", + ); + await writeSessionStore({ + entries: { + main: { + sessionId: "sess-main", + updatedAt: Date.now(), + thinkingLevel: "medium", + reasoningLevel: "stream", + }, + }, + }); + + const { ws } = await openClient(); + const compacted = await rpcReq<{ + ok: true; + key: string; + compacted: boolean; + result?: { tokensAfter?: number }; + }>(ws, "sessions.compact", { + key: "main", + }); + + expect(compacted.ok).toBe(true); + expect(compacted.payload?.key).toBe("agent:main:main"); + expect(compacted.payload?.compacted).toBe(true); + expect(embeddedRunMock.compactEmbeddedPiSession).toHaveBeenCalledTimes(1); + expect(embeddedRunMock.compactEmbeddedPiSession).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "sess-main", + sessionKey: "agent:main:main", + sessionFile: expect.stringMatching(/sess-main\.jsonl$/), + config: expect.any(Object), + provider: expect.any(String), + model: expect.any(String), + thinkLevel: "medium", + reasoningLevel: "stream", + trigger: "manual", + }), + ); + + const store = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + { compactionCount?: number; totalTokens?: number; totalTokensFresh?: boolean } + >; + expect(store["agent:main:main"]?.compactionCount).toBe(1); + expect(store["agent:main:main"]?.totalTokens).toBe(80); + expect(store["agent:main:main"]?.totalTokensFresh).toBe(true); + + ws.close(); + }); + test("sessions.preview returns transcript previews", async () => { const { dir } = await createSessionStoreDir(); const sessionId = "sess-preview"; diff --git a/src/gateway/session-compaction-checkpoints.test.ts b/src/gateway/session-compaction-checkpoints.test.ts new file mode 100644 index 000000000000..63a64804bfe2 --- /dev/null +++ b/src/gateway/session-compaction-checkpoints.test.ts @@ -0,0 +1,84 @@ +import fsSync from "node:fs"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { AssistantMessage, UserMessage } from "@mariozechner/pi-ai"; +import { SessionManager } from "@mariozechner/pi-coding-agent"; +import { afterEach, describe, expect, test } from "vitest"; +import { + captureCompactionCheckpointSnapshot, + cleanupCompactionCheckpointSnapshot, +} from "./session-compaction-checkpoints.js"; + +const tempDirs: string[] = []; + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); +}); + +describe("session-compaction-checkpoints", () => { + test("capture stores the copied pre-compaction transcript path and cleanup removes only the copy", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-")); + tempDirs.push(dir); + + const session = SessionManager.create(dir, dir); + const userMessage: UserMessage = { + role: "user", + content: "before compaction", + timestamp: Date.now(), + }; + const assistantMessage: AssistantMessage = { + role: "assistant", + content: [{ type: "text", text: "working on it" }], + api: "responses", + provider: "openai", + model: "gpt-test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; + session.appendMessage(userMessage); + session.appendMessage(assistantMessage); + + const sessionFile = session.getSessionFile(); + const leafId = session.getLeafId(); + expect(sessionFile).toBeTruthy(); + expect(leafId).toBeTruthy(); + + const originalBefore = await fs.readFile(sessionFile!, "utf-8"); + const snapshot = captureCompactionCheckpointSnapshot({ + sessionManager: session, + sessionFile: sessionFile!, + }); + + expect(snapshot).not.toBeNull(); + expect(snapshot?.leafId).toBe(leafId); + expect(snapshot?.sessionFile).not.toBe(sessionFile); + expect(snapshot?.sessionFile).toContain(".checkpoint."); + expect(fsSync.existsSync(snapshot!.sessionFile)).toBe(true); + expect(await fs.readFile(snapshot!.sessionFile, "utf-8")).toBe(originalBefore); + + session.appendCompaction("checkpoint summary", leafId!, 123, { ok: true }); + + expect(await fs.readFile(snapshot!.sessionFile, "utf-8")).toBe(originalBefore); + expect(await fs.readFile(sessionFile!, "utf-8")).not.toBe(originalBefore); + + await cleanupCompactionCheckpointSnapshot(snapshot); + + expect(fsSync.existsSync(snapshot!.sessionFile)).toBe(false); + expect(fsSync.existsSync(sessionFile!)).toBe(true); + }); +}); diff --git a/src/gateway/session-compaction-checkpoints.ts b/src/gateway/session-compaction-checkpoints.ts new file mode 100644 index 000000000000..142db040deab --- /dev/null +++ b/src/gateway/session-compaction-checkpoints.ts @@ -0,0 +1,207 @@ +import { randomUUID } from "node:crypto"; +import fsSync from "node:fs"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { SessionManager } from "@mariozechner/pi-coding-agent"; +import type { OpenClawConfig } from "../config/config.js"; +import { updateSessionStore } from "../config/sessions.js"; +import type { + SessionCompactionCheckpoint, + SessionCompactionCheckpointReason, + SessionEntry, +} from "../config/sessions.js"; +import { createSubsystemLogger } from "../logging/subsystem.js"; +import { resolveGatewaySessionStoreTarget } from "./session-utils.js"; + +const log = createSubsystemLogger("gateway/session-compaction-checkpoints"); +const MAX_COMPACTION_CHECKPOINTS_PER_SESSION = 25; + +export type CapturedCompactionCheckpointSnapshot = { + sessionId: string; + sessionFile: string; + leafId: string; +}; + +function trimSessionCheckpoints( + checkpoints: SessionCompactionCheckpoint[] | undefined, +): SessionCompactionCheckpoint[] | undefined { + if (!Array.isArray(checkpoints) || checkpoints.length === 0) { + return undefined; + } + return checkpoints.slice(-MAX_COMPACTION_CHECKPOINTS_PER_SESSION); +} + +function sessionStoreCheckpoints( + entry: Pick | undefined, +): SessionCompactionCheckpoint[] { + return Array.isArray(entry?.compactionCheckpoints) ? [...entry.compactionCheckpoints] : []; +} + +export function resolveSessionCompactionCheckpointReason(params: { + trigger?: "budget" | "overflow" | "manual"; + timedOut?: boolean; +}): SessionCompactionCheckpointReason { + if (params.trigger === "manual") { + return "manual"; + } + if (params.timedOut) { + return "timeout-retry"; + } + if (params.trigger === "overflow") { + return "overflow-retry"; + } + return "auto-threshold"; +} + +export function captureCompactionCheckpointSnapshot(params: { + sessionManager: Pick; + sessionFile: string; +}): CapturedCompactionCheckpointSnapshot | null { + const getLeafId = + params.sessionManager && typeof params.sessionManager.getLeafId === "function" + ? params.sessionManager.getLeafId.bind(params.sessionManager) + : null; + const sessionFile = params.sessionFile.trim(); + if (!getLeafId || !sessionFile) { + return null; + } + const leafId = getLeafId(); + if (!leafId) { + return null; + } + const parsedSessionFile = path.parse(sessionFile); + const snapshotFile = path.join( + parsedSessionFile.dir, + `${parsedSessionFile.name}.checkpoint.${randomUUID()}${parsedSessionFile.ext || ".jsonl"}`, + ); + try { + fsSync.copyFileSync(sessionFile, snapshotFile); + } catch { + return null; + } + let snapshotSession: SessionManager; + try { + snapshotSession = SessionManager.open(snapshotFile, path.dirname(snapshotFile)); + } catch { + try { + fsSync.unlinkSync(snapshotFile); + } catch { + // Best-effort cleanup if the copied transcript cannot be reopened. + } + return null; + } + const getSessionId = + snapshotSession && typeof snapshotSession.getSessionId === "function" + ? snapshotSession.getSessionId.bind(snapshotSession) + : null; + if (!getSessionId) { + return null; + } + return { + sessionId: getSessionId(), + sessionFile: snapshotFile, + leafId, + }; +} + +export async function cleanupCompactionCheckpointSnapshot( + snapshot: CapturedCompactionCheckpointSnapshot | null | undefined, +): Promise { + if (!snapshot?.sessionFile) { + return; + } + try { + await fs.unlink(snapshot.sessionFile); + } catch { + // Best-effort cleanup; retained snapshots are harmless and easier to debug. + } +} + +export async function persistSessionCompactionCheckpoint(params: { + cfg: OpenClawConfig; + sessionKey: string; + sessionId: string; + reason: SessionCompactionCheckpointReason; + snapshot: CapturedCompactionCheckpointSnapshot; + summary?: string; + firstKeptEntryId?: string; + tokensBefore?: number; + tokensAfter?: number; + postSessionFile?: string; + postLeafId?: string; + postEntryId?: string; + createdAt?: number; +}): Promise { + const target = resolveGatewaySessionStoreTarget({ + cfg: params.cfg, + key: params.sessionKey, + }); + const createdAt = params.createdAt ?? Date.now(); + const checkpoint: SessionCompactionCheckpoint = { + checkpointId: randomUUID(), + sessionKey: target.canonicalKey, + sessionId: params.sessionId, + createdAt, + reason: params.reason, + ...(typeof params.tokensBefore === "number" ? { tokensBefore: params.tokensBefore } : {}), + ...(typeof params.tokensAfter === "number" ? { tokensAfter: params.tokensAfter } : {}), + ...(params.summary?.trim() ? { summary: params.summary.trim() } : {}), + ...(params.firstKeptEntryId?.trim() + ? { firstKeptEntryId: params.firstKeptEntryId.trim() } + : {}), + preCompaction: { + sessionId: params.snapshot.sessionId, + sessionFile: params.snapshot.sessionFile, + leafId: params.snapshot.leafId, + }, + postCompaction: { + sessionId: params.sessionId, + ...(params.postSessionFile?.trim() ? { sessionFile: params.postSessionFile.trim() } : {}), + ...(params.postLeafId?.trim() ? { leafId: params.postLeafId.trim() } : {}), + ...(params.postEntryId?.trim() ? { entryId: params.postEntryId.trim() } : {}), + }, + }; + + let stored = false; + await updateSessionStore(target.storePath, (store) => { + const existing = store[target.canonicalKey]; + if (!existing?.sessionId) { + return; + } + const checkpoints = sessionStoreCheckpoints(existing); + checkpoints.push(checkpoint); + store[target.canonicalKey] = { + ...existing, + updatedAt: Math.max(existing.updatedAt ?? 0, createdAt), + compactionCheckpoints: trimSessionCheckpoints(checkpoints), + }; + stored = true; + }); + + if (!stored) { + log.warn("skipping compaction checkpoint persist: session not found", { + sessionKey: params.sessionKey, + }); + return null; + } + return checkpoint; +} + +export function listSessionCompactionCheckpoints( + entry: Pick | undefined, +): SessionCompactionCheckpoint[] { + return sessionStoreCheckpoints(entry).toSorted((a, b) => b.createdAt - a.createdAt); +} + +export function getSessionCompactionCheckpoint(params: { + entry: Pick | undefined; + checkpointId: string; +}): SessionCompactionCheckpoint | undefined { + const checkpointId = params.checkpointId.trim(); + if (!checkpointId) { + return undefined; + } + return listSessionCompactionCheckpoints(params.entry).find( + (checkpoint) => checkpoint.checkpointId === checkpointId, + ); +} diff --git a/src/gateway/session-reset-service.ts b/src/gateway/session-reset-service.ts index 93e5f61e4515..bb9cb83e33b7 100644 --- a/src/gateway/session-reset-service.ts +++ b/src/gateway/session-reset-service.ts @@ -472,6 +472,8 @@ export async function performGatewaySessionReset(params: { model: resolvedModel.model, modelProvider: resolvedModel.provider, contextTokens: resetEntry?.contextTokens, + compactionCount: currentEntry?.compactionCount, + compactionCheckpoints: currentEntry?.compactionCheckpoints, sendPolicy: currentEntry?.sendPolicy, queueMode: currentEntry?.queueMode, queueDebounceMs: currentEntry?.queueDebounceMs, diff --git a/src/gateway/session-utils.ts b/src/gateway/session-utils.ts index 2ce224a47eb4..d844d3a4dddb 100644 --- a/src/gateway/session-utils.ts +++ b/src/gateway/session-utils.ts @@ -213,6 +213,18 @@ function resolveNonNegativeNumber(value: number | null | undefined): number | un return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; } +function resolveLatestCompactionCheckpoint( + entry?: Pick | null, +): NonNullable[number] | undefined { + const checkpoints = entry?.compactionCheckpoints; + if (!Array.isArray(checkpoints) || checkpoints.length === 0) { + return undefined; + } + return checkpoints.reduce((latest, checkpoint) => + !latest || checkpoint.createdAt > latest.createdAt ? checkpoint : latest, + ); +} + function resolveEstimatedSessionCostUsd(params: { cfg: OpenClawConfig; provider?: string; @@ -1268,6 +1280,7 @@ export function buildGatewaySessionRow(params: { ? true : transcriptUsage?.totalTokensFresh === true; const childSessions = resolveChildSessionKeys(key, store); + const latestCompactionCheckpoint = resolveLatestCompactionCheckpoint(entry); const estimatedCostUsd = resolveEstimatedSessionCostUsd({ cfg, @@ -1354,6 +1367,8 @@ export function buildGatewaySessionRow(params: { lastTo: deliveryFields.lastTo ?? entry?.lastTo, lastAccountId: deliveryFields.lastAccountId ?? entry?.lastAccountId, lastThreadId: deliveryFields.lastThreadId ?? entry?.lastThreadId, + compactionCheckpointCount: entry?.compactionCheckpoints?.length, + latestCompactionCheckpoint, }; } diff --git a/src/gateway/session-utils.types.ts b/src/gateway/session-utils.types.ts index 5c1d255edee2..893e3876ec36 100644 --- a/src/gateway/session-utils.types.ts +++ b/src/gateway/session-utils.types.ts @@ -1,5 +1,6 @@ import type { ChatType } from "../channels/chat-type.js"; import type { SessionEntry } from "../config/sessions.js"; +import type { SessionCompactionCheckpoint } from "../config/sessions.js"; import type { GatewayAgentRow as SharedGatewayAgentRow, SessionsListResultBase, @@ -64,6 +65,8 @@ export type GatewaySessionRow = { lastTo?: string; lastAccountId?: string; lastThreadId?: SessionEntry["lastThreadId"]; + compactionCheckpointCount?: number; + latestCompactionCheckpoint?: SessionCompactionCheckpoint; }; export type GatewayAgentRow = SharedGatewayAgentRow; diff --git a/src/gateway/test-helpers.mocks.ts b/src/gateway/test-helpers.mocks.ts index d44e543351e8..68222bdeeea1 100644 --- a/src/gateway/test-helpers.mocks.ts +++ b/src/gateway/test-helpers.mocks.ts @@ -54,6 +54,8 @@ const gatewayTestHoisted = getGatewayTestHoistedState(); function createEmbeddedRunMockExports() { return { + compactEmbeddedPiSession: (...args: unknown[]) => + embeddedRunMock.compactEmbeddedPiSession(...args), isEmbeddedPiRunActive: (sessionId: string) => embeddedRunMock.activeIds.has(sessionId), abortEmbeddedPiRun: (sessionId: string) => { embeddedRunMock.abortCalls.push(sessionId); diff --git a/src/gateway/test-helpers.runtime-state.ts b/src/gateway/test-helpers.runtime-state.ts index b509801cbcf8..01a2177ee455 100644 --- a/src/gateway/test-helpers.runtime-state.ts +++ b/src/gateway/test-helpers.runtime-state.ts @@ -22,6 +22,7 @@ export type AgentCommandFn = (...args: unknown[]) => Promise; export type SendWhatsAppFn = (...args: unknown[]) => Promise<{ messageId: string; toJid: string }>; export type RunBtwSideQuestionFn = (...args: unknown[]) => Promise; export type DispatchInboundMessageFn = (...args: unknown[]) => Promise; +export type CompactEmbeddedPiSessionFn = (...args: unknown[]) => Promise; const GATEWAY_TEST_CONFIG_ROOT_KEY = Symbol.for("openclaw.gatewayTestHelpers.configRoot"); @@ -49,6 +50,7 @@ export type GatewayTestHoistedState = { abortCalls: string[]; waitCalls: string[]; waitResults: Map; + compactEmbeddedPiSession: Mock; }; testTailscaleWhois: { value: TailscaleWhoisIdentity | null }; getReplyFromConfig: Mock; @@ -99,6 +101,16 @@ const gatewayTestHoisted = vi.hoisted(() => { abortCalls: [], waitCalls: [], waitResults: new Map(), + compactEmbeddedPiSession: vi.fn().mockResolvedValue({ + ok: true, + compacted: true, + result: { + summary: "summary", + firstKeptEntryId: "entry-1", + tokensBefore: 120, + tokensAfter: 80, + }, + }), }, testTailscaleWhois: { value: null }, getReplyFromConfig: vi.fn().mockResolvedValue(undefined), diff --git a/src/gateway/test-helpers.server.ts b/src/gateway/test-helpers.server.ts index 21897620a6e6..6e3d0866db8f 100644 --- a/src/gateway/test-helpers.server.ts +++ b/src/gateway/test-helpers.server.ts @@ -332,6 +332,17 @@ async function resetGatewayTestState(options: { uniqueConfigRoot: boolean }) { embeddedRunMock.abortCalls = []; embeddedRunMock.waitCalls = []; embeddedRunMock.waitResults.clear(); + embeddedRunMock.compactEmbeddedPiSession.mockReset(); + embeddedRunMock.compactEmbeddedPiSession.mockResolvedValue({ + ok: true, + compacted: true, + result: { + summary: "summary", + firstKeptEntryId: "entry-1", + tokensBefore: 120, + tokensAfter: 80, + }, + }); for (const sessionKey of resolveGatewayTestMainSessionKeys()) { drainSystemEvents(sessionKey); } @@ -411,6 +422,17 @@ async function resetGatewayTestRuntimeOnly() { embeddedRunMock.abortCalls = []; embeddedRunMock.waitCalls = []; embeddedRunMock.waitResults.clear(); + embeddedRunMock.compactEmbeddedPiSession.mockReset(); + embeddedRunMock.compactEmbeddedPiSession.mockResolvedValue({ + ok: true, + compacted: true, + result: { + summary: "summary", + firstKeptEntryId: "entry-1", + tokensBefore: 120, + tokensAfter: 80, + }, + }); clearSessionStoreCacheForTest(); await persistTestSessionConfig(); for (const sessionKey of resolveGatewayTestMainSessionKeys()) { diff --git a/ui/src/ui/app-render.ts b/ui/src/ui/app-render.ts index f5addb5c466a..6298cd680fdf 100644 --- a/ui/src/ui/app-render.ts +++ b/ui/src/ui/app-render.ts @@ -79,7 +79,14 @@ import { import { loadLogs } from "./controllers/logs.ts"; import { loadNodes } from "./controllers/nodes.ts"; import { loadPresence } from "./controllers/presence.ts"; -import { deleteSessionsAndRefresh, loadSessions, patchSession } from "./controllers/sessions.ts"; +import { + branchSessionFromCheckpoint, + deleteSessionsAndRefresh, + loadSessions, + patchSession, + restoreSessionFromCheckpoint, + toggleSessionCompactionCheckpoints, +} from "./controllers/sessions.ts"; import { closeClawHubDetail, installFromClawHub, @@ -840,6 +847,11 @@ export function renderApp(state: AppViewState) { page: state.sessionsPage, pageSize: state.sessionsPageSize, selectedKeys: state.sessionsSelectedKeys, + expandedCheckpointKey: state.sessionsExpandedCheckpointKey, + checkpointItemsByKey: state.sessionsCheckpointItemsByKey, + checkpointLoadingKey: state.sessionsCheckpointLoadingKey, + checkpointBusyKey: state.sessionsCheckpointBusyKey, + checkpointErrorByKey: state.sessionsCheckpointErrorByKey, onFiltersChange: (next) => { state.sessionsFilterActive = next.activeMinutes; state.sessionsFilterLimit = next.limit; @@ -905,6 +917,21 @@ export function renderApp(state: AppViewState) { switchChatSession(state, sessionKey); state.setTab("chat" as import("./navigation.ts").Tab); }, + onToggleCheckpointDetails: (sessionKey) => + toggleSessionCompactionCheckpoints(state, sessionKey), + onBranchFromCheckpoint: async (sessionKey, checkpointId) => { + const nextKey = await branchSessionFromCheckpoint( + state, + sessionKey, + checkpointId, + ); + if (nextKey) { + switchChatSession(state, nextKey); + state.setTab("chat" as import("./navigation.ts").Tab); + } + }, + onRestoreCheckpoint: (sessionKey, checkpointId) => + restoreSessionFromCheckpoint(state, sessionKey, checkpointId), }), ) : nothing} diff --git a/ui/src/ui/app-view-state.ts b/ui/src/ui/app-view-state.ts index d80270fc77d6..bf159357015c 100644 --- a/ui/src/ui/app-view-state.ts +++ b/ui/src/ui/app-view-state.ts @@ -209,6 +209,11 @@ export type AppViewState = { sessionsPage: number; sessionsPageSize: number; sessionsSelectedKeys: Set; + sessionsExpandedCheckpointKey: string | null; + sessionsCheckpointItemsByKey: Record; + sessionsCheckpointLoadingKey: string | null; + sessionsCheckpointBusyKey: string | null; + sessionsCheckpointErrorByKey: Record; usageLoading: boolean; usageResult: SessionsUsageResult | null; usageCostSummary: CostUsageSummary | null; diff --git a/ui/src/ui/app.ts b/ui/src/ui/app.ts index 236812755c3e..a0959d27584a 100644 --- a/ui/src/ui/app.ts +++ b/ui/src/ui/app.ts @@ -89,6 +89,7 @@ import type { ModelCatalogEntry, PresenceEntry, ChannelsStatusSnapshot, + SessionCompactionCheckpoint, SessionsListResult, SkillStatusReport, StatusSummary, @@ -312,6 +313,11 @@ export class OpenClawApp extends LitElement { @state() sessionsPage = 0; @state() sessionsPageSize = 25; @state() sessionsSelectedKeys: Set = new Set(); + @state() sessionsExpandedCheckpointKey: string | null = null; + @state() sessionsCheckpointItemsByKey: Record = {}; + @state() sessionsCheckpointLoadingKey: string | null = null; + @state() sessionsCheckpointBusyKey: string | null = null; + @state() sessionsCheckpointErrorByKey: Record = {}; @state() usageLoading = false; @state() usageResult: import("./types.js").SessionsUsageResult | null = null; diff --git a/ui/src/ui/chat/slash-command-executor.ts b/ui/src/ui/chat/slash-command-executor.ts index 49f13614f6fb..7534ca43186f 100644 --- a/ui/src/ui/chat/slash-command-executor.ts +++ b/ui/src/ui/chat/slash-command-executor.ts @@ -146,8 +146,24 @@ async function executeCompact( sessionKey: string, ): Promise { try { - await client.request("sessions.compact", { key: sessionKey }); - return { content: "Context compacted successfully.", action: "refresh" }; + const result = await client.request<{ + compacted?: boolean; + reason?: string; + result?: { tokensBefore?: number; tokensAfter?: number }; + }>("sessions.compact", { key: sessionKey }); + if (result?.compacted) { + const before = result.result?.tokensBefore; + const after = result.result?.tokensAfter; + const tokenSummary = + typeof before === "number" && typeof after === "number" + ? ` (${before.toLocaleString()} -> ${after.toLocaleString()} tokens)` + : ""; + return { content: `Context compacted successfully${tokenSummary}.`, action: "refresh" }; + } + if (typeof result?.reason === "string" && result.reason.trim()) { + return { content: `Compaction skipped: ${result.reason}`, action: "refresh" }; + } + return { content: "Compaction skipped.", action: "refresh" }; } catch (err) { return { content: `Compaction failed: ${String(err)}` }; } diff --git a/ui/src/ui/controllers/sessions.test.ts b/ui/src/ui/controllers/sessions.test.ts index ad7d8f564855..bfed20ce153a 100644 --- a/ui/src/ui/controllers/sessions.test.ts +++ b/ui/src/ui/controllers/sessions.test.ts @@ -1,5 +1,10 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import { deleteSessionsAndRefresh, subscribeSessions, type SessionsState } from "./sessions.ts"; +import { + deleteSessionsAndRefresh, + loadSessions, + subscribeSessions, + type SessionsState, +} from "./sessions.ts"; type RequestFn = (method: string, params?: unknown) => Promise; @@ -22,6 +27,11 @@ function createState(request: RequestFn, overrides: Partial = {}) sessionsFilterLimit: "0", sessionsIncludeGlobal: true, sessionsIncludeUnknown: true, + sessionsExpandedCheckpointKey: null, + sessionsCheckpointItemsByKey: {}, + sessionsCheckpointLoadingKey: null, + sessionsCheckpointBusyKey: null, + sessionsCheckpointErrorByKey: {}, ...overrides, }; } @@ -120,3 +130,91 @@ describe("deleteSessionsAndRefresh", () => { expect(request).not.toHaveBeenCalled(); }); }); + +describe("loadSessions", () => { + it("refreshes expanded checkpoint cards when the row summary changes", async () => { + const request = vi.fn(async (method: string) => { + if (method === "sessions.list") { + return { + ts: 1, + path: "(multiple)", + count: 1, + defaults: {}, + sessions: [ + { + key: "agent:main:main", + kind: "direct", + updatedAt: 1, + compactionCheckpointCount: 1, + latestCompactionCheckpoint: { + checkpointId: "checkpoint-new", + createdAt: 20, + }, + }, + ], + }; + } + if (method === "sessions.compaction.list") { + return { + ok: true, + key: "agent:main:main", + checkpoints: [ + { + checkpointId: "checkpoint-new", + sessionKey: "agent:main:main", + sessionId: "session-1", + createdAt: 20, + reason: "manual", + }, + ], + }; + } + throw new Error(`unexpected method: ${method}`); + }); + const state = createState(request, { + sessionsExpandedCheckpointKey: "agent:main:main", + sessionsResult: { + ts: 0, + path: "(multiple)", + count: 1, + defaults: {}, + sessions: [ + { + key: "agent:main:main", + kind: "direct", + updatedAt: 0, + compactionCheckpointCount: 3, + latestCompactionCheckpoint: { + checkpointId: "checkpoint-old", + createdAt: 10, + }, + }, + ], + } as never, + sessionsCheckpointItemsByKey: { + "agent:main:main": [ + { + checkpointId: "checkpoint-old", + sessionKey: "agent:main:main", + sessionId: "session-old", + createdAt: 10, + reason: "manual", + }, + ] as never, + }, + }); + + await loadSessions(state); + + expect(request).toHaveBeenNthCalledWith(1, "sessions.list", { + includeGlobal: true, + includeUnknown: true, + }); + expect(request).toHaveBeenNthCalledWith(2, "sessions.compaction.list", { + key: "agent:main:main", + }); + expect( + state.sessionsCheckpointItemsByKey["agent:main:main"]?.map((item) => item.checkpointId), + ).toEqual(["checkpoint-new"]); + }); +}); diff --git a/ui/src/ui/controllers/sessions.ts b/ui/src/ui/controllers/sessions.ts index 4821f521b31c..d800579a85fb 100644 --- a/ui/src/ui/controllers/sessions.ts +++ b/ui/src/ui/controllers/sessions.ts @@ -1,6 +1,12 @@ import { toNumber } from "../format.ts"; import type { GatewayBrowserClient } from "../gateway.ts"; -import type { SessionsListResult } from "../types.ts"; +import type { + SessionCompactionCheckpoint, + SessionsCompactionBranchResult, + SessionsCompactionListResult, + SessionsCompactionRestoreResult, + SessionsListResult, +} from "../types.ts"; import { formatMissingOperatorReadScopeMessage, isMissingOperatorReadScopeError, @@ -16,8 +22,74 @@ export type SessionsState = { sessionsFilterLimit: string; sessionsIncludeGlobal: boolean; sessionsIncludeUnknown: boolean; + sessionsExpandedCheckpointKey: string | null; + sessionsCheckpointItemsByKey: Record; + sessionsCheckpointLoadingKey: string | null; + sessionsCheckpointBusyKey: string | null; + sessionsCheckpointErrorByKey: Record; }; +function checkpointSignature( + row: + | { + key: string; + compactionCheckpointCount?: number; + latestCompactionCheckpoint?: { checkpointId?: string; createdAt?: number } | null; + } + | undefined, +): string { + return JSON.stringify({ + key: row?.key ?? "", + count: row?.compactionCheckpointCount ?? 0, + latestCheckpointId: row?.latestCompactionCheckpoint?.checkpointId ?? "", + latestCreatedAt: row?.latestCompactionCheckpoint?.createdAt ?? 0, + }); +} + +function invalidateCheckpointCacheForKey(state: SessionsState, key: string) { + if ( + !(key in state.sessionsCheckpointItemsByKey) && + !(key in state.sessionsCheckpointErrorByKey) + ) { + return; + } + const nextItems = { ...state.sessionsCheckpointItemsByKey }; + const nextErrors = { ...state.sessionsCheckpointErrorByKey }; + delete nextItems[key]; + delete nextErrors[key]; + state.sessionsCheckpointItemsByKey = nextItems; + state.sessionsCheckpointErrorByKey = nextErrors; +} + +async function fetchSessionCompactionCheckpoints(state: SessionsState, key: string) { + state.sessionsCheckpointLoadingKey = key; + state.sessionsCheckpointErrorByKey = { + ...state.sessionsCheckpointErrorByKey, + [key]: "", + }; + try { + const result = await state.client?.request( + "sessions.compaction.list", + { key }, + ); + if (result) { + state.sessionsCheckpointItemsByKey = { + ...state.sessionsCheckpointItemsByKey, + [key]: result.checkpoints ?? [], + }; + } + } catch (err) { + state.sessionsCheckpointErrorByKey = { + ...state.sessionsCheckpointErrorByKey, + [key]: String(err), + }; + } finally { + if (state.sessionsCheckpointLoadingKey === key) { + state.sessionsCheckpointLoadingKey = null; + } + } +} + export async function subscribeSessions(state: SessionsState) { if (!state.client || !state.connected) { return; @@ -47,6 +119,9 @@ export async function loadSessions( state.sessionsLoading = true; state.sessionsError = null; try { + const previousRows = new Map( + (state.sessionsResult?.sessions ?? []).map((row) => [row.key, row] as const), + ); const includeGlobal = overrides?.includeGlobal ?? state.sessionsIncludeGlobal; const includeUnknown = overrides?.includeUnknown ?? state.sessionsIncludeUnknown; const activeMinutes = overrides?.activeMinutes ?? toNumber(state.sessionsFilterActive, 0); @@ -64,6 +139,30 @@ export async function loadSessions( const res = await state.client.request("sessions.list", params); if (res) { state.sessionsResult = res; + const nextKeys = new Set(res.sessions.map((row) => row.key)); + for (const key of Object.keys(state.sessionsCheckpointItemsByKey)) { + if (!nextKeys.has(key)) { + invalidateCheckpointCacheForKey(state, key); + } + } + let expandedNeedsRefetch = false; + for (const row of res.sessions) { + const previous = previousRows.get(row.key); + if (checkpointSignature(previous) !== checkpointSignature(row)) { + invalidateCheckpointCacheForKey(state, row.key); + if (state.sessionsExpandedCheckpointKey === row.key) { + expandedNeedsRefetch = true; + } + } + } + const expandedKey = state.sessionsExpandedCheckpointKey; + if ( + expandedKey && + nextKeys.has(expandedKey) && + (expandedNeedsRefetch || !state.sessionsCheckpointItemsByKey[expandedKey]) + ) { + await fetchSessionCompactionCheckpoints(state, expandedKey); + } } } catch (err) { if (isMissingOperatorReadScopeError(err)) { @@ -156,3 +255,81 @@ export async function deleteSessionsAndRefresh( } return deleted; } + +export async function toggleSessionCompactionCheckpoints(state: SessionsState, key: string) { + const trimmedKey = key.trim(); + if (!trimmedKey) { + return; + } + if (state.sessionsExpandedCheckpointKey === trimmedKey) { + state.sessionsExpandedCheckpointKey = null; + return; + } + state.sessionsExpandedCheckpointKey = trimmedKey; + if (state.sessionsCheckpointItemsByKey[trimmedKey]) { + return; + } + await fetchSessionCompactionCheckpoints(state, trimmedKey); +} + +export async function branchSessionFromCheckpoint( + state: SessionsState, + key: string, + checkpointId: string, +): Promise { + if (!state.client || !state.connected) { + return null; + } + const confirmed = window.confirm( + "Create a new child session from this pre-compaction checkpoint?", + ); + if (!confirmed) { + return null; + } + state.sessionsCheckpointBusyKey = checkpointId; + try { + const result = await state.client.request( + "sessions.compaction.branch", + { key, checkpointId }, + ); + await loadSessions(state); + return result?.key ?? null; + } catch (err) { + state.sessionsError = String(err); + return null; + } finally { + if (state.sessionsCheckpointBusyKey === checkpointId) { + state.sessionsCheckpointBusyKey = null; + } + } +} + +export async function restoreSessionFromCheckpoint( + state: SessionsState, + key: string, + checkpointId: string, +) { + if (!state.client || !state.connected) { + return; + } + const confirmed = window.confirm( + "Restore this session to the selected pre-compaction checkpoint?\n\nThis replaces the current active transcript for the session key.", + ); + if (!confirmed) { + return; + } + state.sessionsCheckpointBusyKey = checkpointId; + try { + await state.client.request("sessions.compaction.restore", { + key, + checkpointId, + }); + await loadSessions(state); + } catch (err) { + state.sessionsError = String(err); + } finally { + if (state.sessionsCheckpointBusyKey === checkpointId) { + state.sessionsCheckpointBusyKey = null; + } + } +} diff --git a/ui/src/ui/types.ts b/ui/src/ui/types.ts index 4b7ce526ab56..c65e0c2086fd 100644 --- a/ui/src/ui/types.ts +++ b/ui/src/ui/types.ts @@ -369,6 +369,33 @@ export type AgentsFilesSetResult = { export type SessionRunStatus = "running" | "done" | "failed" | "killed" | "timeout"; +export type SessionCompactionCheckpointReason = + | "manual" + | "auto-threshold" + | "overflow-retry" + | "timeout-retry"; + +export type SessionCompactionTranscriptReference = { + sessionId: string; + sessionFile?: string; + leafId?: string; + entryId?: string; +}; + +export type SessionCompactionCheckpoint = { + checkpointId: string; + sessionKey: string; + sessionId: string; + createdAt: number; + reason: SessionCompactionCheckpointReason; + tokensBefore?: number; + tokensAfter?: number; + summary?: string; + firstKeptEntryId?: string; + preCompaction: SessionCompactionTranscriptReference; + postCompaction: SessionCompactionTranscriptReference; +}; + export type GatewaySessionRow = { key: string; spawnedBy?: string; @@ -400,10 +427,47 @@ export type GatewaySessionRow = { model?: string; modelProvider?: string; contextTokens?: number; + compactionCheckpointCount?: number; + latestCompactionCheckpoint?: SessionCompactionCheckpoint; }; export type SessionsListResult = SessionsListResultBase; +export type SessionsCompactionListResult = { + ok: true; + key: string; + checkpoints: SessionCompactionCheckpoint[]; +}; + +export type SessionsCompactionGetResult = { + ok: true; + key: string; + checkpoint: SessionCompactionCheckpoint; +}; + +export type SessionsCompactionBranchResult = { + ok: true; + sourceKey: string; + key: string; + sessionId: string; + checkpoint: SessionCompactionCheckpoint; + entry: { + sessionId: string; + updatedAt: number; + } & Record; +}; + +export type SessionsCompactionRestoreResult = { + ok: true; + key: string; + sessionId: string; + checkpoint: SessionCompactionCheckpoint; + entry: { + sessionId: string; + updatedAt: number; + } & Record; +}; + export type SessionsPatchResult = SessionsPatchResultBase<{ sessionId: string; updatedAt?: number; diff --git a/ui/src/ui/views/sessions.test.ts b/ui/src/ui/views/sessions.test.ts index 0f18a179b1b3..62367dfc2c91 100644 --- a/ui/src/ui/views/sessions.test.ts +++ b/ui/src/ui/views/sessions.test.ts @@ -41,6 +41,11 @@ function buildProps(result: SessionsListResult): SessionsProps { page: 0, pageSize: 10, selectedKeys: new Set(), + expandedCheckpointKey: null, + checkpointItemsByKey: {}, + checkpointLoadingKey: null, + checkpointBusyKey: null, + checkpointErrorByKey: {}, onFiltersChange: () => undefined, onSearchChange: () => undefined, onSortChange: () => undefined, @@ -53,6 +58,9 @@ function buildProps(result: SessionsListResult): SessionsProps { onDeselectPage: () => undefined, onDeselectAll: () => undefined, onDeleteSelected: () => undefined, + onToggleCheckpointDetails: () => undefined, + onBranchFromCheckpoint: () => undefined, + onRestoreCheckpoint: () => undefined, }; } diff --git a/ui/src/ui/views/sessions.ts b/ui/src/ui/views/sessions.ts index 5c8fc9568980..7b5f73368ba5 100644 --- a/ui/src/ui/views/sessions.ts +++ b/ui/src/ui/views/sessions.ts @@ -4,7 +4,11 @@ import { formatRelativeTimestamp } from "../format.ts"; import { icons } from "../icons.ts"; import { pathForTab } from "../navigation.ts"; import { formatSessionTokens } from "../presenter.ts"; -import type { GatewaySessionRow, SessionsListResult } from "../types.ts"; +import type { + GatewaySessionRow, + SessionCompactionCheckpoint, + SessionsListResult, +} from "../types.ts"; export type SessionsProps = { loading: boolean; @@ -21,6 +25,11 @@ export type SessionsProps = { page: number; pageSize: number; selectedKeys: Set; + expandedCheckpointKey: string | null; + checkpointItemsByKey: Record; + checkpointLoadingKey: string | null; + checkpointBusyKey: string | null; + checkpointErrorByKey: Record; onFiltersChange: (next: { activeMinutes: string; limit: string; @@ -48,6 +57,9 @@ export type SessionsProps = { onDeselectAll: () => void; onDeleteSelected: () => void; onNavigateToChat?: (sessionKey: string) => void; + onToggleCheckpointDetails: (sessionKey: string) => void; + onBranchFromCheckpoint: (sessionKey: string, checkpointId: string) => void | Promise; + onRestoreCheckpoint: (sessionKey: string, checkpointId: string) => void | Promise; }; const THINK_LEVELS = ["", "off", "minimal", "low", "medium", "high", "xhigh"] as const; @@ -182,6 +194,36 @@ function paginateRows(rows: T[], page: number, pageSize: number): T[] { return rows.slice(start, start + pageSize); } +function formatCheckpointReason(reason: SessionCompactionCheckpoint["reason"]): string { + switch (reason) { + case "manual": + return "manual"; + case "auto-threshold": + return "auto-threshold"; + case "overflow-retry": + return "overflow retry"; + case "timeout-retry": + return "timeout retry"; + default: + return reason; + } +} + +function formatCheckpointDelta(checkpoint: SessionCompactionCheckpoint): string { + if ( + typeof checkpoint.tokensBefore === "number" && + typeof checkpoint.tokensAfter === "number" && + Number.isFinite(checkpoint.tokensBefore) && + Number.isFinite(checkpoint.tokensAfter) + ) { + return `${checkpoint.tokensBefore.toLocaleString()} → ${checkpoint.tokensAfter.toLocaleString()} tokens`; + } + if (typeof checkpoint.tokensBefore === "number" && Number.isFinite(checkpoint.tokensBefore)) { + return `${checkpoint.tokensBefore.toLocaleString()} tokens before`; + } + return "token delta unavailable"; +} + export function renderSessions(props: SessionsProps) { const rawRows = props.result?.sessions ?? []; const filtered = filterRows(rawRows, props.searchQuery); @@ -349,6 +391,7 @@ export function renderSessions(props: SessionsProps) { Label ${sortHeader("kind", "Kind")} ${sortHeader("updated", "Updated")} ${sortHeader("tokens", "Tokens")} + Compaction Thinking Fast Verbose @@ -360,24 +403,14 @@ export function renderSessions(props: SessionsProps) { ? html` No sessions found. ` - : paginated.map((row) => - renderRow( - row, - props.basePath, - props.onPatch, - props.selectedKeys.has(row.key), - props.onToggleSelect, - props.loading, - props.onNavigateToChat, - ), - )} + : paginated.flatMap((row) => renderRows(row, props))} @@ -416,15 +449,7 @@ export function renderSessions(props: SessionsProps) { `; } -function renderRow( - row: GatewaySessionRow, - basePath: string, - onPatch: SessionsProps["onPatch"], - selected: boolean, - onToggleSelect: SessionsProps["onToggleSelect"], - disabled: boolean, - onNavigateToChat?: (sessionKey: string) => void, -) { +function renderRows(row: GatewaySessionRow, props: SessionsProps) { const updated = row.updatedAt ? formatRelativeTimestamp(row.updatedAt) : t("common.na"); const rawThinking = row.thinkingLevel ?? ""; const isBinaryThinking = isBinaryThinkingProvider(row.modelProvider); @@ -436,6 +461,11 @@ function renderRow( const verboseLevels = withCurrentLabeledOption(VERBOSE_LEVELS, verbose); const reasoning = row.reasoningLevel ?? ""; const reasoningLevels = withCurrentOption(REASONING_LEVELS, reasoning); + const latestCheckpoint = row.latestCompactionCheckpoint; + const checkpointCount = row.compactionCheckpointCount ?? 0; + const isExpanded = props.expandedCheckpointKey === row.key; + const checkpointItems = props.checkpointItemsByKey[row.key] ?? []; + const checkpointError = props.checkpointErrorByKey[row.key]; const displayName = typeof row.displayName === "string" && row.displayName.trim().length > 0 ? row.displayName.trim() @@ -447,7 +477,7 @@ function renderRow( ); const canLink = row.kind !== "global"; const chatUrl = canLink - ? `${pathForTab("chat", basePath)}?session=${encodeURIComponent(row.key)}` + ? `${pathForTab("chat", props.basePath)}?session=${encodeURIComponent(row.key)}` : null; const badgeClass = row.kind === "direct" @@ -458,13 +488,13 @@ function renderRow( ? "data-table-badge--global" : "data-table-badge--unknown"; - return html` - + return [ + html` onToggleSelect(row.key)} + .checked=${props.selectedKeys.has(row.key)} + @change=${() => props.onToggleSelect(row.key)} aria-label="Select session" /> @@ -485,9 +515,9 @@ function renderRow( ) { return; } - if (onNavigateToChat) { + if (props.onNavigateToChat) { e.preventDefault(); - onNavigateToChat(row.key); + props.onNavigateToChat(row.key); } }} >${row.key} { const value = (e.target as HTMLInputElement).value.trim(); - onPatch(row.key, { label: value || null }); + props.onPatch(row.key, { label: value || null }); }} /> @@ -515,13 +545,37 @@ function renderRow( ${updated} ${formatSessionTokens(row)} + +
+ + ${checkpointCount > 0 + ? `${checkpointCount} checkpoint${checkpointCount === 1 ? "" : "s"}` + : "none"} + + ${latestCheckpoint + ? html` + + ${formatCheckpointReason(latestCheckpoint.reason)} · + ${formatRelativeTimestamp(latestCheckpoint.createdAt)} + + ` + : nothing} + +
+ { const value = (e.target as HTMLSelectElement).value; - onPatch(row.key, { fastMode: value === "" ? null : value === "on" }); + props.onPatch(row.key, { fastMode: value === "" ? null : value === "on" }); }} > ${fastLevels.map( @@ -553,11 +607,11 @@ function renderRow( { const value = (e.target as HTMLSelectElement).value; - onPatch(row.key, { reasoningLevel: value || null }); + props.onPatch(row.key, { reasoningLevel: value || null }); }} > ${reasoningLevels.map( @@ -585,6 +639,77 @@ function renderRow( )} - - `; + `, + ...(isExpanded + ? [ + html` + +
+ ${props.checkpointLoadingKey === row.key + ? html`
Loading checkpoints…
` + : checkpointError + ? html`
${checkpointError}
` + : checkpointItems.length === 0 + ? html`
+ No compaction checkpoints recorded for this session. +
` + : html` +
+ ${checkpointItems.map( + (checkpoint) => html` +
+
+ + ${formatCheckpointReason(checkpoint.reason)} · + ${formatRelativeTimestamp(checkpoint.createdAt)} + + + ${formatCheckpointDelta(checkpoint)} + +
+ ${checkpoint.summary + ? html`
+ ${checkpoint.summary} +
` + : html`
No summary captured.
`} +
+ + +
+
+ `, + )} +
+ `} +
+ + `, + ] + : []), + ]; }