diff --git a/src/config/sessions/session-accessor.sqlite-reset-window.ts b/src/config/sessions/session-accessor.sqlite-reset-window.ts index f6fc49503036..6c62bdaa25a1 100644 --- a/src/config/sessions/session-accessor.sqlite-reset-window.ts +++ b/src/config/sessions/session-accessor.sqlite-reset-window.ts @@ -63,6 +63,7 @@ type SessionTranscriptContextWindow = { const resetMessageWindowCache = new Map(); const MAX_MESSAGE_WINDOW_CACHE = 64; +const MAX_CONTEXT_BOUNDARY_BYTES = 1024 * 1024; function getResetWindowKysely(database: OpenClawAgentDatabase) { return getNodeSqliteKysely(database.db); @@ -141,7 +142,35 @@ function readLatestActiveBoundaryByType( .onRef("event.session_id", "=", "active.session_id") .onRef("event.seq", "=", "active.event_seq"), ) - .select(["active.active_position", "identity.event_type", "identity.seq", "event.event_json"]) + .select([ + "active.active_position", + "identity.event_type", + "identity.seq", + /* kysely-allow-raw: reject oversized/malformed boundaries before projecting scalars. */ + sql`LENGTH(CAST(event.event_json AS BLOB))`.as("serialized_bytes"), + sql`json_valid(event.event_json)`.as("json_valid"), + sql`CASE + WHEN LENGTH(CAST(event.event_json AS BLOB)) <= ${MAX_CONTEXT_BOUNDARY_BYTES} + AND json_valid(event.event_json) + AND json_type(event.event_json, '$.firstKeptEntryId') = 'text' + THEN json_extract(event.event_json, '$.firstKeptEntryId') + ELSE NULL + END`.as("first_kept_entry_id"), + sql`CASE + WHEN LENGTH(CAST(event.event_json AS BLOB)) <= ${MAX_CONTEXT_BOUNDARY_BYTES} + AND json_valid(event.event_json) + AND json_type(event.event_json, '$.summary') = 'text' + THEN json_extract(event.event_json, '$.summary') + ELSE NULL + END`.as("summary"), + sql`CASE + WHEN LENGTH(CAST(event.event_json AS BLOB)) <= ${MAX_CONTEXT_BOUNDARY_BYTES} + AND json_valid(event.event_json) + AND json_type(event.event_json, '$.timestamp') IN ('integer', 'real', 'text') + THEN json_extract(event.event_json, '$.timestamp') + ELSE NULL + END`.as("timestamp"), + ]) .where("active.session_id", "=", projection.resolved.sessionId) .where("identity.event_type", "=", eventType) .orderBy("identity.seq", "desc") @@ -161,6 +190,14 @@ function readLatestActiveBoundary(projection: ResetWindowProjection) { return reset.seq > compaction.seq ? reset : compaction; } +function assertUsableBoundary( + boundary: NonNullable>, +): void { + if (boundary.serialized_bytes > MAX_CONTEXT_BOUNDARY_BYTES || boundary.json_valid !== 1) { + throw new Error("Active transcript boundary exceeds the bounded context contract"); + } +} + function readFirstKeptActivePosition( projection: ResetWindowProjection, firstKeptEntryId: unknown, @@ -197,7 +234,7 @@ function findLatestResetMessageWindow( if (!latestBoundaryRow || latestBoundaryRow.event_type !== "reset") { return null; } - const boundary = JSON.parse(latestBoundaryRow.event_json) as { firstKeptEntryId?: unknown }; + assertUsableBoundary(latestBoundaryRow); const postBoundaryMessagePosition = executeSqliteQueryTakeFirstSync( projection.database.db, @@ -213,7 +250,7 @@ function findLatestResetMessageWindow( let keptMessagePositions: number[] = []; const firstKeptActivePosition = readFirstKeptActivePosition( projection, - boundary.firstKeptEntryId, + latestBoundaryRow.first_kept_entry_id, latestBoundaryRow.active_position, ); if (firstKeptActivePosition !== undefined) { @@ -259,27 +296,24 @@ function findContextMessageWindow( if (!latestBoundaryRow) { return null; } - const boundary = JSON.parse(latestBoundaryRow.event_json) as { - firstKeptEntryId?: unknown; - summary?: unknown; - timestamp?: unknown; - }; + assertUsableBoundary(latestBoundaryRow); const retainedStartActivePosition = readFirstKeptActivePosition( projection, - boundary.firstKeptEntryId, + latestBoundaryRow.first_kept_entry_id, latestBoundaryRow.active_position, ); return { scanStartActivePosition: retainedStartActivePosition ?? latestBoundaryRow.active_position + 1, - ...(latestBoundaryRow.event_type === "compaction" && typeof boundary.summary === "string" + ...(latestBoundaryRow.event_type === "compaction" && latestBoundaryRow.summary ? { contextSummary: { - text: boundary.summary, + text: latestBoundaryRow.summary, ts: - typeof boundary.timestamp === "string" - ? Date.parse(boundary.timestamp) || 0 - : typeof boundary.timestamp === "number" && Number.isFinite(boundary.timestamp) - ? boundary.timestamp + typeof latestBoundaryRow.timestamp === "string" + ? Date.parse(latestBoundaryRow.timestamp) || 0 + : typeof latestBoundaryRow.timestamp === "number" && + Number.isFinite(latestBoundaryRow.timestamp) + ? latestBoundaryRow.timestamp : 0, }, } diff --git a/src/gateway/session-companion-context.test.ts b/src/gateway/session-companion-context.test.ts index c33211cff047..dfa82cfed9fa 100644 --- a/src/gateway/session-companion-context.test.ts +++ b/src/gateway/session-companion-context.test.ts @@ -236,6 +236,35 @@ describe("session companion context", () => { ); }); + it("returns unavailable without materializing an oversized compaction boundary", async () => { + const scope = createScope("companion-context-oversized-boundary"); + await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 1 }); + await persistSessionTranscriptTurn(scope, { + messages: [ + { + eventId: "retained", + parentId: null, + message: { role: "user" as const, content: "retained context", timestamp: 1 }, + }, + ], + touchSessionEntry: true, + }); + await appendTranscriptEvent(scope, { + type: "compaction", + id: "oversized-compaction", + parentId: "retained", + timestamp: "2026-08-11T00:00:00.000Z", + summary: "small summary", + firstKeptEntryId: "retained", + tokensBefore: 100, + details: { payload: "x".repeat(1024 * 1024) }, + }); + + await expect(defaultSessionCompanionContextReader.read(scope)).resolves.toEqual({ + kind: "unavailable", + }); + }); + it("returns unavailable rather than an empty context while the active projection is stale", async () => { const scope = createScope("companion-context-unavailable"); await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 1 }); diff --git a/ui/src/pages/chat/chat-pane-base.ts b/ui/src/pages/chat/chat-pane-base.ts index c9cff1f5592f..8bac0a6b06ee 100644 --- a/ui/src/pages/chat/chat-pane-base.ts +++ b/ui/src/pages/chat/chat-pane-base.ts @@ -262,6 +262,13 @@ export abstract class ChatPaneBase extends OpenClawLightDomElement { state.client === client && state.sessionKey === sessionKey && this.connectionGeneration === connectionGeneration, + async (key) => { + const current = this.state; + if (!current?.connected || !current.client) { + throw new Error("Session companion connection is unavailable."); + } + return await requestSessionCompanionState(current.client, key); + }, ); }; diff --git a/ui/src/pages/chat/chat-session-companion.ts b/ui/src/pages/chat/chat-session-companion.ts index 2e2cd3c79282..53ca2118cb26 100644 --- a/ui/src/pages/chat/chat-session-companion.ts +++ b/ui/src/pages/chat/chat-session-companion.ts @@ -146,6 +146,7 @@ export class ChatSessionCompanionThreads { onPrepared: () => void, ) => Promise, isCurrent: () => boolean = () => true, + reload?: (sessionKey: string) => Promise, ): Promise { const key = sessionKey.trim(); const normalized = question.trim(); @@ -166,6 +167,38 @@ export class ChatSessionCompanionThreads { const token = Symbol(key); this.submissionTokens.set(key, token); this.notify(); + const knownExchanges = new Set( + thread.exchanges.map(({ question: priorQuestion, answer, ts }) => + JSON.stringify([priorQuestion, answer, ts]), + ), + ); + const reconcileStale = async ( + expectedAnswer?: string, + ): Promise<"committed" | "missing" | "superseded" | "unavailable"> => { + if (!reload) { + return "unavailable"; + } + try { + const result = await reload(key); + if (this.submissionTokens.get(key) !== token) { + return "superseded"; + } + thread.exchanges = result.exchanges.map(({ question: nextQuestion, answer, ts }) => ({ + question: nextQuestion, + answer, + ts, + })); + const committed = thread.exchanges.some( + (exchange) => + exchange.question === normalized && + (expectedAnswer === undefined || exchange.answer === expectedAnswer) && + !knownExchanges.has(JSON.stringify([exchange.question, exchange.answer, exchange.ts])), + ); + return committed ? "committed" : "missing"; + } catch { + return "unavailable"; + } + }; try { const result = await ask(key, normalized, () => { if (this.submissionTokens.get(key) !== token || !isCurrent()) { @@ -179,10 +212,14 @@ export class ChatSessionCompanionThreads { return; } if (!isCurrent()) { - throw Object.assign(new Error("stale companion answer"), { - details: { reason: "context-unavailable" }, - retryable: true, - }); + const reconciliation = await reconcileStale(result.answer); + if (reconciliation === "committed" || reconciliation === "superseded") { + return; + } + thread.failedQuestion = normalized; + thread.hint = "unavailable"; + thread.retryable = false; + return; } thread.exchanges = [ ...thread.exchanges, @@ -192,12 +229,17 @@ export class ChatSessionCompanionThreads { if (this.submissionTokens.get(key) !== token) { return; } - thread.failedQuestion = normalized; if (!isCurrent()) { - thread.hint = "history-unavailable"; - thread.retryable = true; + const reconciliation = await reconcileStale(); + if (reconciliation === "committed" || reconciliation === "superseded") { + return; + } + thread.failedQuestion = normalized; + thread.hint = reconciliation === "missing" ? "history-unavailable" : "unavailable"; + thread.retryable = reconciliation === "missing"; return; } + thread.failedQuestion = normalized; const reason = errorDetailReason(error); thread.hint = errorDetailCode(error) === COMPANION_BUSY_DETAIL_CODE diff --git a/ui/src/pages/chat/chat-session-rail.test.ts b/ui/src/pages/chat/chat-session-rail.test.ts index 87b84e638f21..edc94d7721be 100644 --- a/ui/src/pages/chat/chat-session-rail.test.ts +++ b/ui/src/pages/chat/chat-session-rail.test.ts @@ -419,6 +419,15 @@ describe("ChatSessionCompanionThreads", () => { }); }, () => current, + async () => ({ + exchanges: [ + { + question: "Which connection owns this?", + answer: "stale answer", + ts: 3, + }, + ], + }), ); await vi.waitFor(() => expect(threads.view("one").phase).toBe("answering")); current = false; @@ -426,11 +435,17 @@ describe("ChatSessionCompanionThreads", () => { await pending; expect(threads.view("one")).toMatchObject({ - exchanges: [], - failedQuestion: "Which connection owns this?", - hint: "history-unavailable", + exchanges: [ + { + question: "Which connection owns this?", + answer: "stale answer", + ts: 3, + }, + ], + failedQuestion: null, + hint: null, pendingQuestion: null, - retryable: true, + retryable: false, }); }); @@ -448,6 +463,7 @@ describe("ChatSessionCompanionThreads", () => { }); }, () => current, + async () => ({ exchanges: [] }), ); await vi.waitFor(() => expect(threads.view("one").phase).toBe("answering")); current = false;