diff --git a/CHANGELOG.md b/CHANGELOG.md index 2671cee87ee7..8872f5870e7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,7 +68,7 @@ Docs: https://docs.openclaw.ai ### Fixes -- **Control UI session companion:** load bounded selected-session context before answering, show Reading/Answering and retryable history errors, and prevent private companion context envelopes from appearing as answers. Fixes #120746. Thanks @shakkernerd. +- **Control UI session companion:** load bounded visible session context before answering, keep unavailable questions retryable, and prevent private companion reference wrappers from appearing as answers. Fixes #120746. Thanks @shakkernerd. - **Telegram live locations:** expose initial, moving, and stopped live-location updates through the channel-neutral `message_received` hook without starting agent turns for edits. - **Updater plugin convergence:** keep pre-plugin doctor passes from installing configured plugins before the updater's plugin sweep, while preserving the final post-plugin migration pass and preventing ambient update-phase state from leaking into fresh doctor processes. - **Control UI browser tab identity:** keep selected tab styling, accessibility, focus, address, and page snapshot aligned across in-place navigation and tab reordering. Fixes #120745. Thanks @shakkernerd. diff --git a/docs/tools/btw.md b/docs/tools/btw.md index 99a10627b0bd..092d574267fe 100644 --- a/docs/tools/btw.md +++ b/docs/tools/btw.md @@ -11,7 +11,7 @@ session** without adding it to conversation history. It is modeled after Claude Code's `/btw`, adapted to OpenClaw's Gateway and multi-channel architecture. -The two side-question contracts are deliberately separate. BTW is a one-shot question on the session's actual model, preserving harness behavior and Codex thread-fork continuity for channel ingress (WhatsApp, Telegram, and Discord), the TUI, and embedded `tui --local`; the TUI stays on BTW by design. The companion is a persistent, read-only RPC thread for Control UI-class clients. Its first question lazily prepares bounded selected-session context; a temporary history failure remains retryable and does not run as an empty session. Channels cannot use the companion because they do not have an RPC connection. +The two side-question contracts are deliberately separate. BTW is a one-shot question on the session's actual model, preserving harness behavior and Codex thread-fork continuity for channel ingress (WhatsApp, Telegram, and Discord), the TUI, and embedded `tui --local`; the TUI stays on BTW by design. The companion is a persistent, read-only RPC thread for Control UI-class clients. Its first question lazily prepares bounded visible context from the selected session; a temporary history failure remains retryable and does not run as an empty session. Channels cannot use the companion because they do not have an RPC connection. ```text /btw what changed? @@ -61,11 +61,11 @@ session companion RPCs and renders their bounded exchange state in the rail. ## Surface behavior -| Surface | Behavior | -| ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| TUI | Rendered inline in the chat log, visibly distinct from a normal reply, dismissible with `Enter` or `Esc`. | -| External channels | Delivered as a clearly labeled one-off reply (Telegram, WhatsApp, Discord have no local ephemeral overlay). | -| Control UI / web | Routes `/btw` and `/side` to the expanded session rail companion. The read-only thread is keyed by session, rehydrates from Gateway memory, shows Reading then Answering while the first context is prepared, and preserves a failed question for Retry. It can be cleared with the trash button. `Esc` collapses the rail. | +| Surface | Behavior | +| ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| TUI | Rendered inline in the chat log, visibly distinct from a normal reply, dismissible with `Enter` or `Esc`. | +| External channels | Delivered as a clearly labeled one-off reply (Telegram, WhatsApp, Discord have no local ephemeral overlay). | +| Control UI / web | Routes `/btw` and `/side` to the expanded session rail companion. The read-only thread is keyed by session, rehydrates from Gateway memory, and preserves a failed question for Retry. It can be cleared with the trash button. `Esc` collapses the rail. | ## Selection popup (Control UI) diff --git a/docs/web/control-ui.md b/docs/web/control-ui.md index 3b89bd032aae..abd4503c2e9b 100644 --- a/docs/web/control-ui.md +++ b/docs/web/control-ui.md @@ -16,7 +16,7 @@ It speaks **directly to the Gateway WebSocket** on the same port. While you watch a running session, the Gateway shows the model's latest safe preamble immediately as the session headline. When a utility model is available, it can replace that headline with a richer compact status digest after enough activity accumulates. Chat carries the result in a **session rail**: its compact pill shows the live digest, while the expanded rail shows the assessment, plan progress, pull requests, elapsed time, and a read-only companion thread. The rail can expand once when a run becomes stuck or needs input, and done or failed runs keep a frozen “finished” time based on the final digest. On wide chat panes the expanded rail docks as a 400 px right column; on narrower and mobile layouts it remains an overlay. -The companion answers questions about the selected session and its project without entering or interrupting the main agent run. On the first question, the Gateway lazily loads a bounded snapshot of the selected session before starting the utility model; the rail shows **Reading this session…** and then **Answering…**. If history is temporarily unavailable, the question stays visible with **Retry** instead of being treated as an empty session. The companion uses read-only access to the target session's history/search and agent workspace. Its bounded thread is held in Gateway memory, is restored when you switch sessions in the Control UI, and is cleared by the rail's trash button, a session reset, Gateway restart, or idle expiry. It never enters `chat.history`, and private context transport is not stored as operator dialogue. Type `/btw ` or `/side ` in the main Control UI composer to open the rail and ask there; other clients keep their existing BTW behavior. +The companion answers questions about the selected session and its project without entering or interrupting the main agent run. On the first question, the Gateway lazily loads a bounded visible snapshot of the selected session before starting the utility model. If history is temporarily unavailable, the question stays visible with **Retry** instead of being treated as an empty session. The companion uses read-only access to the target session's history/search and agent workspace. Its bounded thread is held in Gateway memory, is restored when you switch sessions in the Control UI, and is cleared by the rail's trash button, a session reset, Gateway restart, or idle expiry. It never enters `chat.history`, and private reference context is not stored as operator dialogue. Type `/btw ` or `/side ` in the main Control UI composer to open the rail and ask there; other clients keep their existing BTW behavior. Highlighting text in a chat message offers **More details**, which asks the companion immediately, and **Ask in side chat**, which opens the rail with a quoted draft ready to edit. diff --git a/packages/gateway-protocol/src/client-info.ts b/packages/gateway-protocol/src/client-info.ts index 078ef485fddc..415ef7c49bda 100644 --- a/packages/gateway-protocol/src/client-info.ts +++ b/packages/gateway-protocol/src/client-info.ts @@ -82,7 +82,6 @@ export const GATEWAY_CLIENT_CAPS = { EXEC_APPROVALS: "exec-approvals", INLINE_WIDGETS: "inline-widgets", RUN_TOOL_BINDINGS: "run-tool-bindings", - SESSION_COMPANION_PROGRESS: "session-companion-progress", SESSION_SCOPED_EVENTS: "session-scoped-events", PLUGIN_APPROVALS: "plugin-approvals", TASK_SUGGESTIONS: "task-suggestions", diff --git a/src/config/sessions/session-accessor.sqlite-active-events.test.ts b/src/config/sessions/session-accessor.sqlite-active-events.test.ts index 2205f60c899d..caa0ac975a1d 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.test.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.test.ts @@ -13,7 +13,6 @@ import { readRecentSessionTranscriptMessageEvents, readSessionTranscriptActiveLeafEvents, readSessionTranscriptActiveStats, - readSessionTranscriptBoundedContextMessageTailPage, readSessionTranscriptBoundedMessageTailPage, readSessionTranscriptMessageAnchorPage, readSessionTranscriptMessageEventById, @@ -873,40 +872,5 @@ describe("SQLite active transcript event projection", () => { await reconciliation; expect(order).toEqual(["event-loop-responsive", "live-write", "reconciled"]); expect(readSessionTranscriptMessageEventCount(scope)).toBe(100_001); - - await appendTranscriptEvent(scope, { - type: "compaction", - id: "large-context-boundary", - parentId: "m100001", - timestamp: "2026-08-11T00:00:00.000Z", - summary: "bounded large-history summary", - firstKeptEntryId: "m1", - tokensBefore: 100_000, - }); - await persistSessionTranscriptTurn(scope, { - messages: Array.from({ length: 40 }, (_, index) => ({ - eventId: `post-boundary-${index}`, - parentId: index === 0 ? "large-context-boundary" : `post-boundary-${index - 1}`, - message: { - role: index % 2 === 0 ? ("user" as const) : ("assistant" as const), - content: `post-boundary ${index}`, - }, - })), - touchSessionEntry: false, - }); - - const contextPage = readSessionTranscriptBoundedContextMessageTailPage(scope, { - maxBytes: 1024 * 1024, - maxMessages: 40, - maxScannedMessages: 4096, - }); - expect(contextPage).toMatchObject({ - authoritative: true, - contextSummary: { text: "bounded large-history summary" }, - empty: false, - }); - expect(contextPage.events).toHaveLength(40); - expect(contextPage.events.at(0)?.event).toMatchObject({ id: "post-boundary-0" }); - expect(contextPage.events.at(-1)?.event).toMatchObject({ id: "post-boundary-39" }); }, 60_000); }); diff --git a/src/config/sessions/session-accessor.sqlite-active-events.ts b/src/config/sessions/session-accessor.sqlite-active-events.ts index a9d418b02a6e..64cc99e816d4 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.ts @@ -14,7 +14,6 @@ import type { TranscriptEvent, } from "./session-accessor.sqlite-contract.js"; import { - readBoundedContextMessageTail, readVisibleMessageRange, resolveVisibleMessagePositionRange, resolveVisibleMessagePositions, @@ -61,13 +60,6 @@ export type SessionTranscriptBoundedMessageTailPage = SessionTranscriptMessageEv serializedBytes: number; }; -type SessionTranscriptBoundedContextMessageTailPage = { - authoritative: boolean; - contextSummary?: { text: string; ts: number }; - empty: boolean; - events: SessionTranscriptMessageEvent[]; -}; - function parseMessageEventRow(row: { event_json: string; message_position: number | null; @@ -447,17 +439,13 @@ export function readSessionTranscriptMessageEventPage( }); } -function readBoundedMessageTailPage( +/** Reads a tail page whose materialized event payloads fit a hard byte budget. */ +export function readSessionTranscriptBoundedMessageTailPage( scope: SessionTranscriptReadScope, options: { maxBytes: number; maxMessages: number; offset: number }, - visibility: { - resolvePositionRange: typeof resolveVisibleMessagePositionRange; - resolvePositions: typeof resolveVisibleMessagePositions; - }, - readExtra: (projection: Parameters[0]) => TExtra, -): SessionTranscriptBoundedMessageTailPage & TExtra { +): SessionTranscriptBoundedMessageTailPage { return withCurrentProjectionSnapshot(scope, (projection) => { - const visible = visibility.resolvePositions(projection); + const visible = resolveVisibleMessagePositions(projection); const totalMessages = visible.total; const offset = Math.min( Math.max(0, Math.floor(Number.isFinite(options.offset) ? options.offset : 0)), @@ -473,8 +461,7 @@ function readBoundedMessageTailPage( ); const endExclusive = Math.max(0, totalMessages - offset); const start = Math.max(0, endExclusive - maxMessages); - const positions = visibility.resolvePositionRange(projection, start, endExclusive); - const extra = readExtra(projection); + const positions = resolveVisibleMessagePositionRange(projection, start, endExclusive); if (positions.length === 0 || maxBytes === 0) { return { activeLeafEntryId: projection.state.leafEventId, @@ -482,7 +469,6 @@ function readBoundedMessageTailPage( scannedMessages: positions.length, serializedBytes: 0, totalMessages, - ...extra, }; } const db = getActiveTranscriptKysely(projection.database); @@ -536,40 +522,10 @@ function readBoundedMessageTailPage( scannedMessages: positions.length, serializedBytes, totalMessages, - ...extra, }; }); } -/** Reads a transcript-visible tail page whose payloads fit a hard byte budget. */ -export function readSessionTranscriptBoundedMessageTailPage( - scope: SessionTranscriptReadScope, - options: { maxBytes: number; maxMessages: number; offset: number }, -): SessionTranscriptBoundedMessageTailPage { - return readBoundedMessageTailPage( - scope, - options, - { - resolvePositionRange: resolveVisibleMessagePositionRange, - resolvePositions: resolveVisibleMessagePositions, - }, - () => ({}), - ); -} - -/** - * Reads a model-context tail without resurrecting messages discarded by the - * latest reset/compaction boundary. - */ -export function readSessionTranscriptBoundedContextMessageTailPage( - scope: SessionTranscriptReadScope, - options: { maxBytes: number; maxMessages: number; maxScannedMessages: number }, -): SessionTranscriptBoundedContextMessageTailPage { - return withCurrentProjectionSnapshot(scope, (projection) => - readBoundedContextMessageTail(projection, options), - ); -} - export function readSessionTranscriptMessageEventCount(scope: SessionTranscriptReadScope): number { return withCurrentProjectionSnapshot( scope, diff --git a/src/config/sessions/session-accessor.sqlite-reset-window.ts b/src/config/sessions/session-accessor.sqlite-reset-window.ts index 6f1d24650f05..b0e20a3a28a7 100644 --- a/src/config/sessions/session-accessor.sqlite-reset-window.ts +++ b/src/config/sessions/session-accessor.sqlite-reset-window.ts @@ -1,6 +1,4 @@ -// Reset and model-context boundaries project logical message windows without -// rewriting raw cursor positions. -import { sql } from "kysely"; +// Reset boundaries project a logical message window without rewriting raw cursor positions. import { executeSqliteQuerySync, executeSqliteQueryTakeFirstSync, @@ -38,11 +36,6 @@ type ResetWindowMessageEvent = { seq: number; }; -type ContextBoundarySummary = { - text: string; - ts: number; -}; - type ResetMessageWindow = { generation: string | undefined; indexedSeq: number; @@ -56,14 +49,8 @@ type ResetMessageWindowCacheEntry = { window: ResetMessageWindow | null; }; -type SessionTranscriptContextWindow = { - contextSummary?: ContextBoundarySummary; - scanStartActivePosition: number; -}; - const resetMessageWindowCache = new Map(); -const MAX_MESSAGE_WINDOW_CACHE = 64; -const MAX_CONTEXT_BOUNDARY_BYTES = 1024 * 1024; +const MAX_RESET_MESSAGE_WINDOW_CACHE = 64; function getResetWindowKysely(database: OpenClawAgentDatabase) { return getNodeSqliteKysely(database.db); @@ -109,7 +96,16 @@ function readMessageRange( ).rows.map(parseMessageEventRow); } -function messageWindowCacheKey(projection: ResetWindowProjection): string { +function parseTranscriptEventType(eventJson: string): string | undefined { + try { + const parsed = JSON.parse(eventJson) as { type?: unknown }; + return typeof parsed.type === "string" ? parsed.type : undefined; + } catch { + return undefined; + } +} + +function resetMessageWindowCacheKey(projection: ResetWindowProjection): string { return `${projection.database.path}\0${projection.resolved.sessionId}`; } @@ -123,142 +119,10 @@ function readTranscriptGeneration(projection: ResetWindowProjection): string | u )?.generation; } -function sqliteBoundarySerializedBytes() { - return /* kysely-allow-raw: boundary size is checked before scalar projection. */ sql`LENGTH(CAST(event.event_json AS BLOB))`; -} - -function sqliteBoundaryJsonValid() { - return /* kysely-allow-raw: boundary JSON validity is part of the fail-closed contract. */ sql`json_valid(event.event_json)`; -} - -function sqliteBoundaryFirstKeptEntryId() { - return /* kysely-allow-raw: project one bounded canonical boundary scalar. */ sql< - string | null - >`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`; -} - -function sqliteBoundarySummary() { - return /* kysely-allow-raw: project one bounded canonical boundary scalar. */ sql< - string | null - >`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`; -} - -function sqliteBoundaryTimestamp() { - return /* kysely-allow-raw: project one bounded canonical boundary scalar. */ sql< - string | number | null - >`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`; -} - -function sqliteContextMessageRole() { - return /* kysely-allow-raw: inspect the canonical role without loading payload JSON. */ sql< - string | null - >`CASE WHEN json_valid(event.event_json) - THEN json_extract(event.event_json, '$.message.role') ELSE NULL END`; -} - -function sqliteContextMessageSerializedBytes() { - return /* kysely-allow-raw: enforce the payload budget before materialization. */ sql`LENGTH(CAST(event.event_json AS BLOB)) + 1`; -} - -function readLatestActiveBoundaryByType( - projection: ResetWindowProjection, - eventType: "compaction" | "reset", -) { - const db = getResetWindowKysely(projection.database); - return executeSqliteQueryTakeFirstSync( - projection.database.db, - db - .selectFrom("session_transcript_active_events as active") - .innerJoin("transcript_event_identities as identity", (join) => - join - .onRef("identity.session_id", "=", "active.session_id") - .onRef("identity.seq", "=", "active.event_seq"), - ) - .innerJoin("transcript_events as event", (join) => - join - .onRef("event.session_id", "=", "active.session_id") - .onRef("event.seq", "=", "active.event_seq"), - ) - .select([ - "active.active_position", - "identity.event_type", - "identity.seq", - sqliteBoundarySerializedBytes().as("serialized_bytes"), - sqliteBoundaryJsonValid().as("json_valid"), - sqliteBoundaryFirstKeptEntryId().as("first_kept_entry_id"), - sqliteBoundarySummary().as("summary"), - sqliteBoundaryTimestamp().as("timestamp"), - ]) - .where("active.session_id", "=", projection.resolved.sessionId) - .where("identity.event_type", "=", eventType) - .orderBy("identity.seq", "desc") - .limit(1), - ); -} - -function readLatestActiveBoundary(projection: ResetWindowProjection) { - const reset = readLatestActiveBoundaryByType(projection, "reset"); - const compaction = readLatestActiveBoundaryByType(projection, "compaction"); - if (!reset) { - return compaction; - } - if (!compaction) { - return reset; - } - 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, - boundaryActivePosition: number, -): number | undefined { - if (typeof firstKeptEntryId !== "string") { - return undefined; - } - const db = getResetWindowKysely(projection.database); - const firstKept = executeSqliteQueryTakeFirstSync( - projection.database.db, - db - .selectFrom("transcript_event_identities as identity") - .innerJoin("session_transcript_active_events as active", (join) => - join - .onRef("active.session_id", "=", "identity.session_id") - .onRef("active.event_seq", "=", "identity.seq"), - ) - .select("active.active_position") - .where("identity.session_id", "=", projection.resolved.sessionId) - .where("identity.event_id", "=", firstKeptEntryId), - ); - return firstKept && firstKept.active_position < boundaryActivePosition - ? firstKept.active_position - : undefined; +function cacheResetMessageWindow(key: string, entry: ResetMessageWindowCacheEntry): void { + resetMessageWindowCache.delete(key); + resetMessageWindowCache.set(key, entry); + pruneMapToMaxSize(resetMessageWindowCache, MAX_RESET_MESSAGE_WINDOW_CACHE); } function findLatestResetMessageWindow( @@ -266,11 +130,29 @@ function findLatestResetMessageWindow( generation: string | undefined, ): ResetMessageWindow | null { const db = getResetWindowKysely(projection.database); - const latestBoundaryRow = readLatestActiveBoundary(projection); - if (!latestBoundaryRow || latestBoundaryRow.event_type !== "reset") { + const nonMessageRows = executeSqliteQuerySync( + projection.database.db, + db + .selectFrom("session_transcript_active_events as active") + .innerJoin("transcript_events as event", (join) => + join + .onRef("event.session_id", "=", "active.session_id") + .onRef("event.seq", "=", "active.event_seq"), + ) + .select(["active.active_position", "event.event_json"]) + .where("active.session_id", "=", projection.resolved.sessionId) + .where("active.message_position", "is", null) + .orderBy("active.active_position", "desc"), + ).rows; + const latestBoundaryRow = nonMessageRows.find((row) => { + const type = parseTranscriptEventType(row.event_json); + return type === "reset" || type === "compaction"; + }); + if (!latestBoundaryRow || parseTranscriptEventType(latestBoundaryRow.event_json) !== "reset") { return null; } - assertUsableBoundary(latestBoundaryRow); + const resetRow = latestBoundaryRow; + const reset = JSON.parse(resetRow.event_json) as { firstKeptEntryId?: unknown }; const postBoundaryMessagePosition = executeSqliteQueryTakeFirstSync( projection.database.db, @@ -278,44 +160,55 @@ function findLatestResetMessageWindow( .selectFrom("session_transcript_active_events") .select("message_position") .where("session_id", "=", projection.resolved.sessionId) - .where("active_position", ">", latestBoundaryRow.active_position) + .where("active_position", ">", resetRow.active_position) .where("message_position", "is not", null) .orderBy("active_position", "asc") .limit(1), )?.message_position ?? projection.state.activeMessageCount; let keptMessagePositions: number[] = []; - const firstKeptActivePosition = readFirstKeptActivePosition( - projection, - latestBoundaryRow.first_kept_entry_id, - latestBoundaryRow.active_position, - ); - if (firstKeptActivePosition !== undefined) { - keptMessagePositions = executeSqliteQuerySync( + if (typeof reset.firstKeptEntryId === "string") { + const firstKept = executeSqliteQueryTakeFirstSync( projection.database.db, db - .selectFrom("session_transcript_active_events as active") - .innerJoin("transcript_events as event", (join) => + .selectFrom("transcript_event_identities as identity") + .innerJoin("session_transcript_active_events as active", (join) => join - .onRef("event.session_id", "=", "active.session_id") - .onRef("event.seq", "=", "active.event_seq"), + .onRef("active.session_id", "=", "identity.session_id") + .onRef("active.event_seq", "=", "identity.seq"), ) - .select(["active.message_position", "event.event_json"]) - .where("active.session_id", "=", projection.resolved.sessionId) - .where("active.active_position", ">=", firstKeptActivePosition) - .where("active.active_position", "<", latestBoundaryRow.active_position) - .where("active.message_position", "is not", null) - .orderBy("active.active_position", "asc"), - ).rows.flatMap((row) => { - if (row.message_position === null) { - return []; - } - try { - const role = (JSON.parse(row.event_json) as { message?: { role?: unknown } }).message?.role; - return role === "user" || role === "assistant" ? [row.message_position] : []; - } catch { - return []; - } - }); + .select("active.active_position") + .where("identity.session_id", "=", projection.resolved.sessionId) + .where("identity.event_id", "=", reset.firstKeptEntryId), + ); + if (firstKept && firstKept.active_position < resetRow.active_position) { + keptMessagePositions = executeSqliteQuerySync( + projection.database.db, + db + .selectFrom("session_transcript_active_events as active") + .innerJoin("transcript_events as event", (join) => + join + .onRef("event.session_id", "=", "active.session_id") + .onRef("event.seq", "=", "active.event_seq"), + ) + .select(["active.message_position", "event.event_json"]) + .where("active.session_id", "=", projection.resolved.sessionId) + .where("active.active_position", ">=", firstKept.active_position) + .where("active.active_position", "<", resetRow.active_position) + .where("active.message_position", "is not", null) + .orderBy("active.active_position", "asc"), + ).rows.flatMap((row) => { + if (row.message_position === null) { + return []; + } + try { + const role = (JSON.parse(row.event_json) as { message?: { role?: unknown } }).message + ?.role; + return role === "user" || role === "assistant" ? [row.message_position] : []; + } catch { + return []; + } + }); + } } return { generation, @@ -325,40 +218,8 @@ function findLatestResetMessageWindow( }; } -function findContextMessageWindow( - projection: ResetWindowProjection, -): SessionTranscriptContextWindow | null { - const latestBoundaryRow = readLatestActiveBoundary(projection); - if (!latestBoundaryRow) { - return null; - } - assertUsableBoundary(latestBoundaryRow); - const retainedStartActivePosition = readFirstKeptActivePosition( - projection, - latestBoundaryRow.first_kept_entry_id, - latestBoundaryRow.active_position, - ); - return { - scanStartActivePosition: retainedStartActivePosition ?? latestBoundaryRow.active_position + 1, - ...(latestBoundaryRow.event_type === "compaction" && latestBoundaryRow.summary - ? { - contextSummary: { - text: latestBoundaryRow.summary, - ts: - typeof latestBoundaryRow.timestamp === "string" - ? Date.parse(latestBoundaryRow.timestamp) || 0 - : typeof latestBoundaryRow.timestamp === "number" && - Number.isFinite(latestBoundaryRow.timestamp) - ? latestBoundaryRow.timestamp - : 0, - }, - } - : {}), - }; -} - function resolveResetMessageWindow(projection: ResetWindowProjection): ResetMessageWindow | null { - const key = messageWindowCacheKey(projection); + const key = resetMessageWindowCacheKey(projection); const cached = resetMessageWindowCache.get(key); const generation = readTranscriptGeneration(projection); if (cached) { @@ -367,22 +228,14 @@ function resolveResetMessageWindow(projection: ResetWindowProjection): ResetMess } } const window = findLatestResetMessageWindow(projection, generation); - resetMessageWindowCache.delete(key); - resetMessageWindowCache.set(key, { + cacheResetMessageWindow(key, { generation, indexedSeq: projection.state.indexedSeq, window, }); - pruneMapToMaxSize(resetMessageWindowCache, MAX_MESSAGE_WINDOW_CACHE); return window; } -function resolveContextMessageWindow( - projection: ResetWindowProjection, -): SessionTranscriptContextWindow | null { - return findContextMessageWindow(projection); -} - export function resolveVisibleMessagePositions( projection: ResetWindowProjection, ): VisibleMessagePositions { @@ -427,12 +280,15 @@ export function readVisibleMessageRange( return [...keptEvents, ...postEvents]; } -/** Maps a logical transcript-visible range to materialized message positions. */ +/** Maps a logical visible-message range to its materialized message positions. */ export function resolveVisibleMessagePositionRange( projection: ResetWindowProjection, start: number, endExclusive: number, ): number[] { + if (endExclusive <= start) { + return []; + } const visible = resolveVisibleMessagePositions(projection); const boundedStart = Math.min(Math.max(0, start), visible.total); const boundedEnd = Math.min(Math.max(boundedStart, endExclusive), visible.total); @@ -445,82 +301,3 @@ export function resolveVisibleMessagePositionRange( } return positions; } - -/** Reads one authoritative bounded model-context tail from the active semantic window. */ -export function readBoundedContextMessageTail( - projection: ResetWindowProjection, - options: { maxBytes: number; maxMessages: number; maxScannedMessages: number }, -) { - const maxMessages = Math.max(0, Math.floor(options.maxMessages)); - const maxScannedMessages = Math.max(0, Math.floor(options.maxScannedMessages)); - const maxBytes = Math.max(0, Math.floor(options.maxBytes)); - const contextWindow = resolveContextMessageWindow(projection); - const db = getResetWindowKysely(projection.database); - const metadata = executeSqliteQuerySync( - projection.database.db, - db - .selectFrom("session_transcript_active_events as active") - .innerJoin("transcript_events as event", (join) => - join - .onRef("event.session_id", "=", "active.session_id") - .onRef("event.seq", "=", "active.event_seq"), - ) - .select([ - "active.message_position", - sqliteContextMessageRole().as("message_role"), - sqliteContextMessageSerializedBytes().as("serialized_bytes"), - ]) - .where("active.session_id", "=", projection.resolved.sessionId) - .where("active.message_position", "is not", null) - .$if(contextWindow !== null, (query) => - query.where("active.active_position", ">=", contextWindow?.scanStartActivePosition ?? 0), - ) - .orderBy("active.active_position", "desc") - .limit(maxScannedMessages + 1), - ).rows; - const selectedPositions: number[] = []; - let serializedBytes = 0; - let blockedByBytes = false; - for (const row of metadata.slice(0, maxScannedMessages)) { - if ( - row.message_position === null || - (row.message_role !== "assistant" && row.message_role !== "user") - ) { - continue; - } - if (selectedPositions.length >= maxMessages) { - break; - } - if (serializedBytes + row.serialized_bytes > maxBytes) { - blockedByBytes = true; - break; - } - selectedPositions.push(row.message_position); - serializedBytes += row.serialized_bytes; - } - const events = - selectedPositions.length === 0 - ? [] - : executeSqliteQuerySync( - projection.database.db, - db - .selectFrom("session_transcript_active_events as active") - .innerJoin("transcript_events as event", (join) => - join - .onRef("event.session_id", "=", "active.session_id") - .onRef("event.seq", "=", "active.event_seq"), - ) - .select(["active.message_position", "event.event_json"]) - .where("active.session_id", "=", projection.resolved.sessionId) - .where("active.message_position", "in", selectedPositions) - .orderBy("active.message_position", "asc"), - ).rows.map(parseMessageEventRow); - return { - authoritative: - !blockedByBytes && - (selectedPositions.length >= maxMessages || metadata.length <= maxScannedMessages), - ...(contextWindow?.contextSummary ? { contextSummary: contextWindow.contextSummary } : {}), - empty: metadata.length === 0 && !contextWindow?.contextSummary, - events, - }; -} diff --git a/src/gateway/server/ws-connection/authenticated-request-dispatch.abort.test.ts b/src/gateway/server/ws-connection/authenticated-request-dispatch.abort.test.ts index 3d82ec2eaef9..54677afdb43d 100644 --- a/src/gateway/server/ws-connection/authenticated-request-dispatch.abort.test.ts +++ b/src/gateway/server/ws-connection/authenticated-request-dispatch.abort.test.ts @@ -1,4 +1,4 @@ -/** Verifies transport disconnect cancels only its own paired-node invocation. */ +/** Verifies transport disconnect cancellation stays scoped to request-owned work. */ import { EventEmitter } from "node:events"; import { afterEach, describe, expect, it, vi } from "vitest"; import { WebSocket } from "ws"; @@ -104,7 +104,7 @@ function createDispatcher( return { client, dispatcher }; } -describe("paired-node WebSocket request cancellation", () => { +describe("authenticated WebSocket request cancellation", () => { it("forwards CLI socket closure to the actual first-party node cancel event", async () => { const socket = new EventEmitter(); const { registry, frames } = createPairedNode(); @@ -172,6 +172,38 @@ describe("paired-node WebSocket request cancellation", () => { expect(socket.listenerCount("close")).toBe(0); }); + it("cancels a session companion ask when its authenticated socket closes", async () => { + const socket = new EventEmitter(); + const { client, dispatcher } = createDispatcher(socket, { + id: GATEWAY_CLIENT_IDS.CONTROL_UI, + mode: GATEWAY_CLIENT_MODES.UI, + }); + let observedSignal: AbortSignal | undefined; + handleGatewayRequest.mockImplementation(async (options: GatewayRequestOptions) => { + observedSignal = options.signal; + await new Promise((resolve) => { + options.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + }); + + const request = dispatcher.dispatch( + { + type: "req", + id: "session-companion", + method: "sessions.companion.ask", + params: { sessionKey: "agent:main:main", question: "What changed?" }, + }, + client, + ); + await vi.waitFor(() => expect(socket.listenerCount("close")).toBe(1)); + + socket.emit("close", 1000, Buffer.alloc(0)); + + await request; + expect(observedSignal?.aborted).toBe(true); + expect(socket.listenerCount("close")).toBe(0); + }); + it.each([ { label: "control UI", diff --git a/src/gateway/server/ws-connection/authenticated-request-dispatch.ts b/src/gateway/server/ws-connection/authenticated-request-dispatch.ts index bfa7b3f73be2..1fd7fe9edb02 100644 --- a/src/gateway/server/ws-connection/authenticated-request-dispatch.ts +++ b/src/gateway/server/ws-connection/authenticated-request-dispatch.ts @@ -184,17 +184,17 @@ export function createGatewayAuthenticatedRequestDispatcher(params: { }; const executeRequest = async () => { - // One-shot CLI clients cancel by closing their authenticated socket; - // leave long-lived SDK/UI invocations independent of connection teardown. - const nodeInvocationController = - req.method === "node.invoke" && - client.connect.client.id === GATEWAY_CLIENT_IDS.CLI && - client.connect.client.mode === GATEWAY_CLIENT_MODES.CLI - ? new AbortController() - : undefined; - const cancelNodeInvocation = () => nodeInvocationController?.abort(); - if (nodeInvocationController) { - client.socket.once("close", cancelNodeInvocation); + // Most UI/SDK RPCs outlive a reconnect. Companion asks are the exception: + // without their requester there is no safe recipient for a late answer. + const cancelOnDisconnect = + req.method === "sessions.companion.ask" || + (req.method === "node.invoke" && + client.connect.client.id === GATEWAY_CLIENT_IDS.CLI && + client.connect.client.mode === GATEWAY_CLIENT_MODES.CLI); + const requestController = cancelOnDisconnect ? new AbortController() : undefined; + const cancelRequest = () => requestController?.abort(); + if (requestController) { + client.socket.once("close", cancelRequest); } try { const { handleGatewayRequest } = await loadGatewayServerMethods(); @@ -206,7 +206,7 @@ export function createGatewayAuthenticatedRequestDispatcher(params: { extraHandlers, methodRegistry: getMethodRegistry?.(), context, - ...(nodeInvocationController ? { signal: nodeInvocationController.signal } : {}), + ...(requestController ? { signal: requestController.signal } : {}), }); } catch (err) { // Failure diagnostics and responses belong to the same request trace as the handler. @@ -217,8 +217,8 @@ export function createGatewayAuthenticatedRequestDispatcher(params: { errorShape(ErrorCodes.UNAVAILABLE, formatForLog(err)), ); } finally { - if (nodeInvocationController) { - client.socket.off("close", cancelNodeInvocation); + if (requestController) { + client.socket.off("close", cancelRequest); } } }; diff --git a/src/gateway/session-companion-ask.ts b/src/gateway/session-companion-ask.ts index fd4d340d9389..11d32594ff56 100644 --- a/src/gateway/session-companion-ask.ts +++ b/src/gateway/session-companion-ask.ts @@ -15,7 +15,6 @@ import { buildSessionCompanionRunConfig, SESSION_COMPANION_TOOLS, } from "./session-companion-policy.js"; -import { notifySessionCompanionPrepared } from "./session-companion-progress.js"; import { trimSessionCompanionExchanges, type SessionCompanionThread, @@ -72,6 +71,7 @@ type SessionCompanionCancellationKind = | "backing-session-revoked" | "disposed" | "explicit-reset" + | "request-aborted" | "timeout"; type SessionCompanionActiveAsk = { @@ -264,12 +264,7 @@ function buildReferenceContext(params: { : "No bounded user/assistant transcript text was available; use the permitted session tools when needed." : params.thread.context.messages .map((message) => { - const label = - message.role === "summary" - ? "Compaction summary" - : message.role === "assistant" - ? "Assistant" - : "Operator"; + const label = message.role === "assistant" ? "Assistant" : "Operator"; return `${label}: ${escapeReferenceText(message.text)}`; }) .join("\n"); @@ -336,30 +331,13 @@ function composePromptMessages(params: { return messages; } -function isPrivateEnvelopeEcho(value: string): boolean { - if (value.includes(PRIVATE_REFERENCE_BEGIN) || value.includes(PRIVATE_REFERENCE_END)) { - return true; - } - try { - const parsed = JSON.parse(value) as Record; - const keys = Object.keys(parsed); - return ( - keys.length > 0 && - keys.every((key) => - ["inheritedSessionMessages", "observerDigestJson", "observerNotes", "question"].includes( - key, - ), - ) && - (keys.includes("inheritedSessionMessages") || keys.includes("observerNotes")) - ); - } catch { - return false; - } +function isPrivateReferenceEcho(value: string): boolean { + return value.includes(PRIVATE_REFERENCE_BEGIN) || value.includes(PRIVATE_REFERENCE_END); } function sanitizeAnswer(value: string): string { const redacted = redactToolPayloadText(value).trim(); - if (isPrivateEnvelopeEcho(redacted)) { + if (isPrivateReferenceEcho(redacted)) { return ""; } return truncateUtf16Safe(redacted, ANSWER_MAX_CHARS); @@ -379,7 +357,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt const setTimeoutFn = params.setTimeoutFn ?? setTimeout; const clearTimeoutFn = params.clearTimeoutFn ?? clearTimeout; const activeAsks = new Map(); - const preparations = new Map>(); const admissions: Array<{ connId: string; admittedAt: number }> = []; const resolveTarget = (sessionKey: string) => { @@ -398,69 +375,55 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt ): Promise => { const existing = params.threads.get(sessionKey); const { agentId, observerSnapshot } = resolveTarget(sessionKey); - if ( - existing && - currentSessionId(sessionKey, agentId) === existing.context.sessionId && - !signal.aborted - ) { + if (signal.aborted) { + throw new Error("session companion preparation was cancelled"); + } + if (existing && currentSessionId(sessionKey, agentId) === existing.context.sessionId) { return existing; } if (existing) { params.threads.delete(sessionKey); } - const activePreparation = preparations.get(sessionKey); - if (activePreparation) { - return await activePreparation; + const result = await contextReader.read({ agentId, sessionKey, signal }); + if (signal.aborted || params.isDisposed()) { + throw new Error("session companion preparation was cancelled"); } - const preparation = (async () => { - const result = await contextReader.read({ agentId, sessionKey, signal }); - if (signal.aborted || params.isDisposed()) { - throw new Error("session companion preparation was cancelled"); - } - if (result.kind === "missing") { - throw contextError("session-missing", "The selected session is no longer available."); - } - if (result.kind === "unavailable") { - throw contextError( - "context-unavailable", - "The selected session history could not be loaded.", - ); - } - if (currentSessionId(sessionKey, agentId) !== result.context.sessionId) { - throw contextError( - "context-unavailable", - "The selected session changed before its history was ready.", - ); - } - const thread: SessionCompanionThread = { - context: result.context, - digestText: formatObserverDigest(observerSnapshot), - exchanges: [], - lastNoteSequence: 0, - busy: false, - lastUsedAt: params.now(), - }; - params.threads.set(sessionKey, thread); - return thread; - })(); - preparations.set(sessionKey, preparation); - try { - return await preparation; - } finally { - if (preparations.get(sessionKey) === preparation) { - preparations.delete(sessionKey); - } + if (result.kind === "missing") { + throw contextError("session-missing", "The selected session is no longer available."); } + if (result.kind === "unavailable") { + throw contextError( + "context-unavailable", + "The selected session history could not be loaded.", + ); + } + if (currentSessionId(sessionKey, agentId) !== result.context.sessionId) { + throw contextError( + "context-unavailable", + "The selected session changed before its history was ready.", + ); + } + const thread: SessionCompanionThread = { + context: result.context, + digestText: formatObserverDigest(observerSnapshot), + exchanges: [], + lastNoteSequence: 0, + busy: false, + lastUsedAt: params.now(), + }; + params.threads.set(sessionKey, thread); + return thread; }; const ask = async (request: { sessionKey: string; question: string; connId: string; + signal?: AbortSignal; }): Promise<{ answer: string; ts: number }> => { const sessionKey = request.sessionKey.trim(); const question = request.question.trim(); - if (!sessionKey || !question || params.isDisposed()) { + if (!sessionKey || !question || params.isDisposed() || request.signal?.aborted) { throw new SessionCompanionAskError("unavailable", "Session companion is unavailable."); } const existing = params.threads.get(sessionKey); @@ -516,6 +479,12 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt activeAsk.cancellation = cancellation; controller.abort(); }; + const abortRequest = () => abort("request-aborted"); + if (request.signal?.aborted) { + abortRequest(); + } else { + request.signal?.addEventListener("abort", abortRequest, { once: true }); + } const timeout = setTimeoutFn(() => abort("timeout"), ASK_TIMEOUT_MS); const aborted = new Promise((_resolve, reject) => { controller.signal.addEventListener( @@ -524,7 +493,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt { once: true }, ); }); - let ownedAgentId: string | undefined; let ownedThread: SessionCompanionThread | undefined; const discardOwnedThread = () => { if (ownedThread && params.threads.get(sessionKey) === ownedThread) { @@ -534,11 +502,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt try { const thread = await prepareThread(sessionKey, controller.signal); ownedThread = thread; - notifySessionCompanionPrepared({ - connId: request.connId, - empty: thread.context.empty, - sessionKey, - }); if (thread.busy) { throw new SessionCompanionAskError( "busy", @@ -548,7 +511,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt thread.busy = true; thread.lastUsedAt = admittedAt; const { agentId, cfg } = resolveTarget(sessionKey); - ownedAgentId = agentId; if (currentSessionId(sessionKey, agentId) !== thread.context.sessionId) { params.threads.delete(sessionKey); throw contextError( @@ -632,19 +594,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt "The selected session changed before the companion could answer.", ); } - if (!activeAsk.cancellation && ownedThread) { - if ( - params.threads.get(sessionKey) !== ownedThread || - (ownedAgentId && - currentSessionId(sessionKey, ownedAgentId) !== ownedThread.context.sessionId) - ) { - discardOwnedThread(); - throw contextError( - "context-unavailable", - "The selected session changed before the companion could answer.", - ); - } - } companionLog.warn("session companion ask failed", { sessionKey, error }); throw new SessionCompanionAskError( "unavailable", @@ -656,6 +605,7 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt ); } finally { clearTimeoutFn(timeout); + request.signal?.removeEventListener("abort", abortRequest); if (activeAsks.get(sessionKey) === activeAsk) { activeAsks.delete(sessionKey); } @@ -687,7 +637,6 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt activeAsk.controller.abort(); } activeAsks.clear(); - preparations.clear(); admissions.length = 0; }, }; diff --git a/src/gateway/session-companion-context.test.ts b/src/gateway/session-companion-context.test.ts index dfa82cfed9fa..cba1271e8d0c 100644 --- a/src/gateway/session-companion-context.test.ts +++ b/src/gateway/session-companion-context.test.ts @@ -32,6 +32,25 @@ function createScope(prefix: string) { } describe("session companion context", () => { + it("distinguishes a missing session from an empty selected transcript", async () => { + const missing = createScope("companion-context-missing"); + await expect(defaultSessionCompanionContextReader.read(missing)).resolves.toEqual({ + kind: "missing", + }); + + const empty = createScope("companion-context-empty"); + await upsertSessionEntryCore(empty, { sessionId: empty.sessionId, updatedAt: 1 }); + const result = await defaultSessionCompanionContextReader.read(empty); + expect(result).toEqual({ + kind: "ready", + context: { + empty: true, + messages: [], + sessionId: empty.sessionId, + }, + }); + }); + it("reads a bounded active SQLite tail without decoding old transcript rows", async () => { const scope = createScope("companion-context-tail"); await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 1 }); @@ -176,7 +195,7 @@ describe("session companion context", () => { }); }); - it("preserves the latest compaction summary and retained context without resurrecting history", async () => { + it("keeps transcript-visible messages across compaction", async () => { const scope = createScope("companion-context-compaction"); await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 1 }); await persistSessionTranscriptTurn(scope, { @@ -226,17 +245,14 @@ describe("session companion context", () => { return; } expect(result.context.messages.map((message) => [message.role, message.text])).toEqual([ - ["summary", "older context was compacted"], + ["user", "discarded context"], ["user", "retained context"], ["assistant", "recent answer"], ["user", "visible current question"], ]); - expect(result.context.messages.some((message) => message.text === "discarded context")).toBe( - false, - ); }); - it("returns unavailable without materializing an oversized compaction boundary", async () => { + it("ignores oversized non-message compaction details", async () => { const scope = createScope("companion-context-oversized-boundary"); await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 1 }); await persistSessionTranscriptTurn(scope, { @@ -261,7 +277,12 @@ describe("session companion context", () => { }); await expect(defaultSessionCompanionContextReader.read(scope)).resolves.toEqual({ - kind: "unavailable", + kind: "ready", + context: { + empty: false, + messages: [{ role: "user", text: "retained context", ts: 1 }], + sessionId: scope.sessionId, + }, }); }); diff --git a/src/gateway/session-companion-context.ts b/src/gateway/session-companion-context.ts index 6eaaeb2a8ed3..bce1b1814cc2 100644 --- a/src/gateway/session-companion-context.ts +++ b/src/gateway/session-companion-context.ts @@ -5,7 +5,7 @@ import { } from "../agents/tools/chat-history-text.js"; import { isSessionTranscriptProjectionUnavailableError, - readSessionTranscriptBoundedContextMessageTailPage, + readSessionTranscriptBoundedMessageTailPage, } from "../config/sessions/session-accessor.sqlite-active-events.js"; import { redactToolPayloadText } from "../logging/redact.js"; import type { @@ -19,6 +19,7 @@ const CONTEXT_MAX_BYTES = 24 * 1024; const CONTEXT_MESSAGE_MAX_CHARS = 4000; const CONTEXT_READ_MAX_SCANNED_MESSAGES = 4096; const CONTEXT_READ_MAX_BYTES = 1024 * 1024; +const CONTEXT_READ_PAGE_MESSAGES = 128; type SessionCompanionContextReadResult = | { kind: "ready"; context: SessionCompanionPreparedContext } @@ -72,35 +73,28 @@ function readMessageTimestamp(message: unknown): number { return typeof value === "number" && Number.isFinite(value) ? Math.max(0, Math.floor(value)) : 0; } -function sanitizeContextMessages( - messages: unknown[], - contextSummary?: { text: string; ts: number }, -): SessionCompanionContextMessage[] { - const sanitized = stripToolMessages(messages) - .slice(-CONTEXT_MAX_MESSAGES) - .flatMap((message): SessionCompanionContextMessage[] => { - if (!message || typeof message !== "object") { - return []; - } - const role = (message as { role?: unknown }).role; - const text = - role === "assistant" - ? normalizeContextText(extractStoredAssistantText(message) ?? "") - : role === "user" - ? extractUserText(message) - : undefined; - return text && (role === "assistant" || role === "user") - ? [{ role, text, ts: readMessageTimestamp(message) }] - : []; - }); - const summaryText = contextSummary ? normalizeContextText(contextSummary.text) : ""; - const summaryMessage = summaryText - ? ({ role: "summary", text: summaryText, ts: contextSummary?.ts ?? 0 } as const) - : undefined; - const selected: SessionCompanionContextMessage[] = summaryMessage ? [summaryMessage] : []; - let bytes = - 2 + (summaryMessage ? Buffer.byteLength(JSON.stringify(summaryMessage), "utf8") + 1 : 0); - for (const message of sanitized.toReversed()) { +function sanitizeContextMessages(messages: unknown[]): SessionCompanionContextMessage[] { + return stripToolMessages(messages).flatMap((message): SessionCompanionContextMessage[] => { + if (!message || typeof message !== "object") { + return []; + } + const role = (message as { role?: unknown }).role; + const text = + role === "assistant" + ? normalizeContextText(extractStoredAssistantText(message) ?? "") + : role === "user" + ? extractUserText(message) + : undefined; + return text && (role === "assistant" || role === "user") + ? [{ role, text, ts: readMessageTimestamp(message) }] + : []; + }); +} + +function selectContextMessages(messages: SessionCompanionContextMessage[]) { + const selected: SessionCompanionContextMessage[] = []; + let bytes = 2; + for (const message of messages.toReversed()) { if (selected.length >= CONTEXT_MAX_MESSAGES) { break; } @@ -108,7 +102,7 @@ function sanitizeContextMessages( if (bytes + messageBytes > CONTEXT_MAX_BYTES) { break; } - selected.splice(summaryMessage ? 1 : 0, 0, message); + selected.unshift(message); bytes += messageBytes; } return selected; @@ -144,20 +138,46 @@ async function readSessionCompanionContext(params: { if (params.signal?.aborted) { return { kind: "unavailable" }; } - const page = readSessionTranscriptBoundedContextMessageTailPage(scope, { - maxBytes: CONTEXT_READ_MAX_BYTES, - maxMessages: CONTEXT_MAX_MESSAGES, - maxScannedMessages: CONTEXT_READ_MAX_SCANNED_MESSAGES, - }); - if (!page.authoritative || params.signal?.aborted) { + let offset = 0; + let rawBytes = 0; + let scannedMessages = 0; + let totalMessages = 0; + let contextMessages: SessionCompanionContextMessage[] = []; + while ( + contextMessages.length < CONTEXT_MAX_MESSAGES && + scannedMessages < CONTEXT_READ_MAX_SCANNED_MESSAGES + ) { + const page = readSessionTranscriptBoundedMessageTailPage(scope, { + maxBytes: CONTEXT_READ_MAX_BYTES - rawBytes, + maxMessages: Math.min( + CONTEXT_READ_PAGE_MESSAGES, + CONTEXT_READ_MAX_SCANNED_MESSAGES - scannedMessages, + ), + offset, + }); + if (params.signal?.aborted || page.events.length !== page.scannedMessages) { + return { kind: "unavailable" }; + } + totalMessages = page.totalMessages; + rawBytes += page.serializedBytes; + scannedMessages += page.scannedMessages; + offset += page.scannedMessages; + contextMessages = [ + ...sanitizeContextMessages(readPageMessages(page.events)), + ...contextMessages, + ].slice(-CONTEXT_MAX_MESSAGES); + if (page.scannedMessages === 0 || offset >= totalMessages) { + break; + } + } + if (contextMessages.length < CONTEXT_MAX_MESSAGES && offset < totalMessages) { return { kind: "unavailable" }; } - const selected = sanitizeContextMessages(readPageMessages(page.events), page.contextSummary); return { kind: "ready", context: { - empty: page.empty, - messages: selected, + empty: totalMessages === 0, + messages: selectContextMessages(contextMessages), sessionId, }, }; diff --git a/src/gateway/session-companion-progress.ts b/src/gateway/session-companion-progress.ts deleted file mode 100644 index 5014b22f9f8b..000000000000 --- a/src/gateway/session-companion-progress.ts +++ /dev/null @@ -1,39 +0,0 @@ -type SessionCompanionProgressListener = (payload: { empty: boolean }) => void; - -const listeners = new Map(); - -function progressKey(connId: string, sessionKey: string): string { - return `${connId}\0${sessionKey}`; -} - -export function registerSessionCompanionProgress(params: { - connId: string; - sessionKey: string; - listener: SessionCompanionProgressListener; -}): () => void { - const key = progressKey(params.connId, params.sessionKey); - // A duplicate busy ask must not steal the accepted phase from the request - // that already owns this connection/session slot. - if (listeners.has(key)) { - return () => {}; - } - listeners.set(key, params.listener); - return () => { - if (listeners.get(key) === params.listener) { - listeners.delete(key); - } - }; -} - -export function notifySessionCompanionPrepared(params: { - connId: string; - empty: boolean; - sessionKey: string; -}): void { - try { - listeners.get(progressKey(params.connId, params.sessionKey))?.({ empty: params.empty }); - } catch { - // Progress presentation is advisory; a callback failure cannot abort the - // authoritative companion request after context is ready. - } -} diff --git a/src/gateway/session-companion-rpc.test.ts b/src/gateway/session-companion-rpc.test.ts index 7ba4125386af..a1c2e84649f5 100644 --- a/src/gateway/session-companion-rpc.test.ts +++ b/src/gateway/session-companion-rpc.test.ts @@ -1,11 +1,6 @@ import { describe, expect, it, vi } from "vitest"; -import { GATEWAY_CLIENT_CAPS } from "../../packages/gateway-protocol/src/client-info.js"; import { GatewayErrorDetailCodes } from "../../packages/gateway-protocol/src/index.js"; import { SessionCompanionAskError } from "./session-companion-ask.js"; -import { - notifySessionCompanionPrepared, - registerSessionCompanionProgress, -} from "./session-companion-progress.js"; import { sessionCompanionHandlers } from "./session-companion-rpc.js"; async function invoke( @@ -16,10 +11,8 @@ async function invoke( state?: ReturnType; reset?: ReturnType; }, - client: { connId?: string; connect?: { caps?: string[] } } = { - connId: "conn-1", - connect: { caps: [] }, - }, + client: { connId?: string } = { connId: "conn-1" }, + signal?: AbortSignal, ) { const respond = vi.fn(); await sessionCompanionHandlers[method]?.({ @@ -27,40 +20,12 @@ async function invoke( client, context: { sessionCompanion: companion }, respond, + signal, } as never); return respond; } describe("session companion RPC", () => { - it("keeps the first progress owner and isolates callback failures", () => { - const first = vi.fn(() => { - throw new Error("presentation failed"); - }); - const second = vi.fn(); - const clearFirst = registerSessionCompanionProgress({ - connId: "conn-1", - sessionKey: "agent:main:main", - listener: first, - }); - const clearSecond = registerSessionCompanionProgress({ - connId: "conn-1", - sessionKey: "agent:main:main", - listener: second, - }); - - expect(() => - notifySessionCompanionPrepared({ - connId: "conn-1", - empty: false, - sessionKey: "agent:main:main", - }), - ).not.toThrow(); - expect(first).toHaveBeenCalledOnce(); - expect(second).not.toHaveBeenCalled(); - clearSecond(); - clearFirst(); - }); - it("dispatches a valid ask and returns its timestamp", async () => { const ask = vi.fn(async () => ({ answer: "It is checking the fix.", ts: 123 })); const respond = await invoke( @@ -80,34 +45,24 @@ describe("session companion RPC", () => { }); }); - it("emits progress only after context is ready, then returns the answer", async () => { - const ask = vi.fn(async () => { - notifySessionCompanionPrepared({ - connId: "conn-1", - empty: false, - sessionKey: "agent:main:main", - }); - return { answer: "It is checking the fix.", ts: 123 }; - }); + it("forwards the authenticated request lifetime and emits one final response", async () => { + const controller = new AbortController(); + const ask = vi.fn(async () => ({ answer: "Bound to this connection.", ts: 124 })); const respond = await invoke( "sessions.companion.ask", - { sessionKey: "agent:main:main", question: "What is happening?" }, + { sessionKey: "agent:main:main", question: "Who owns this ask?" }, { ask }, - { - connId: "conn-1", - connect: { caps: [GATEWAY_CLIENT_CAPS.SESSION_COMPANION_PROGRESS] }, - }, + { connId: "conn-1" }, + controller.signal, ); expect(ask).toHaveBeenCalledWith({ sessionKey: "agent:main:main", - question: "What is happening?", + question: "Who owns this ask?", connId: "conn-1", + signal: controller.signal, }); - expect(respond.mock.calls).toEqual([ - [true, { status: "accepted", empty: false }], - [true, { answer: "It is checking the fix.", ts: 123 }], - ]); + expect(respond.mock.calls).toEqual([[true, { answer: "Bound to this connection.", ts: 124 }]]); }); it.each([ @@ -173,10 +128,6 @@ describe("session companion RPC", () => { "sessions.companion.ask", { sessionKey: "agent:main:main", question: "Why?" }, { ask }, - { - connId: "conn-1", - connect: { caps: [GATEWAY_CLIENT_CAPS.SESSION_COMPANION_PROGRESS] }, - }, ); expect(respond).toHaveBeenCalledWith( false, diff --git a/src/gateway/session-companion-rpc.ts b/src/gateway/session-companion-rpc.ts index af147b9a3d42..0b4f1e0d2927 100644 --- a/src/gateway/session-companion-rpc.ts +++ b/src/gateway/session-companion-rpc.ts @@ -1,4 +1,3 @@ -import { GATEWAY_CLIENT_CAPS } from "../../packages/gateway-protocol/src/client-info.js"; import { ErrorCodes, errorShape, @@ -13,10 +12,9 @@ import { } from "../../packages/gateway-protocol/src/index.js"; import type { GatewayRequestHandlers } from "./server-methods/types.js"; import { SessionCompanionAskError } from "./session-companion-ask.js"; -import { registerSessionCompanionProgress } from "./session-companion-progress.js"; export const sessionCompanionHandlers: GatewayRequestHandlers = { - "sessions.companion.ask": async ({ params, respond, client, context }) => { + "sessions.companion.ask": async ({ params, respond, client, context, signal }) => { if (!validateSessionsCompanionAskParams(params)) { respond( false, @@ -53,20 +51,12 @@ export const sessionCompanionHandlers: GatewayRequestHandlers = { ); return; } - const unregisterProgress = client.connect.caps?.includes( - GATEWAY_CLIENT_CAPS.SESSION_COMPANION_PROGRESS, - ) - ? registerSessionCompanionProgress({ - connId: client.connId, - sessionKey, - listener: (prepared) => respond(true, { status: "accepted", ...prepared }), - }) - : undefined; try { const result = await context.sessionCompanion.ask({ sessionKey, question, connId: client.connId, + ...(signal ? { signal } : {}), }); respond(true, result); } catch (error) { @@ -99,8 +89,6 @@ export const sessionCompanionHandlers: GatewayRequestHandlers = { ...(error.retryAfterMs ? { retryAfterMs: error.retryAfterMs } : {}), }), ); - } finally { - unregisterProgress?.(); } }, "sessions.companion.state": ({ params, respond, context }) => { diff --git a/src/gateway/session-companion-state.ts b/src/gateway/session-companion-state.ts index f72f8e19463c..4fb8f9d64e97 100644 --- a/src/gateway/session-companion-state.ts +++ b/src/gateway/session-companion-state.ts @@ -1,7 +1,7 @@ import type { SessionCompanionExchange } from "../../packages/gateway-protocol/src/schema/sessions.js"; export type SessionCompanionContextMessage = { - role: "assistant" | "summary" | "user"; + role: "assistant" | "user"; text: string; ts: number; }; diff --git a/src/gateway/session-companion.test.ts b/src/gateway/session-companion.test.ts index 7b054e845eb2..c19419a29f76 100644 --- a/src/gateway/session-companion.test.ts +++ b/src/gateway/session-companion.test.ts @@ -227,17 +227,13 @@ describe("session companion asks", () => { missing.service.dispose(); }); - it("rejects private envelope echoes without rejecting requested JSON", async () => { + it("rejects the private reference wrapper without rejecting requested JSON", async () => { vi.useFakeTimers(); - const envelope = createHarness({ - run: async () => - JSON.stringify({ - inheritedSessionMessages: [], - observerDigestJson: "null", - }), + const wrapper = createHarness({ + run: async () => "private context", }); await expect( - envelope.service.ask({ + wrapper.service.ask({ sessionKey: "agent:main:main", question: "Return the first message.", connId: "conn-1", @@ -245,17 +241,25 @@ describe("session companion asks", () => { ).rejects.toMatchObject({ reason: "unavailable", } satisfies Partial); - expect(envelope.service.state("agent:main:main")).toEqual({ exchanges: [] }); - envelope.service.dispose(); + expect(wrapper.service.state("agent:main:main")).toEqual({ exchanges: [] }); + wrapper.service.dispose(); - const legitimate = createHarness({ run: async () => '{"status":"green"}' }); + const legitimate = createHarness({ + run: async () => + JSON.stringify({ + inheritedSessionMessages: [], + observerDigestJson: "null", + }), + }); await expect( legitimate.service.ask({ sessionKey: "agent:main:main", - question: "Return JSON with the build status.", + question: "Return JSON with these exact field names.", connId: "conn-1", }), - ).resolves.toMatchObject({ answer: '{"status":"green"}' }); + ).resolves.toMatchObject({ + answer: '{"inheritedSessionMessages":[],"observerDigestJson":"null"}', + }); legitimate.service.dispose(); }); @@ -493,6 +497,45 @@ describe("session companion asks", () => { harness.service.dispose(); }); + it("cancels a disconnected request before a late model result can commit", async () => { + vi.useFakeTimers(); + const pending = deferred(); + const controller = new AbortController(); + let runCount = 0; + const harness = createHarness({ + run: async () => (runCount++ === 0 ? "existing answer" : await pending.promise), + }); + await harness.service.ask({ + sessionKey: "agent:main:main", + question: "What is already known?", + connId: "conn-1", + }); + const active = harness.service.ask({ + sessionKey: "agent:main:main", + question: "Will a disconnected request commit?", + connId: "conn-1", + signal: controller.signal, + }); + await vi.waitFor(() => expect(harness.run).toHaveBeenCalledOnce()); + + controller.abort(); + pending.resolve("late answer"); + + await expect(active).rejects.toMatchObject({ + reason: "unavailable", + } satisfies Partial); + expect(harness.service.state("agent:main:main")).toEqual({ + exchanges: [ + { + question: "What is already known?", + answer: "existing answer", + ts: 100, + }, + ], + }); + harness.service.dispose(); + }); + it("disposal cancels an active ask without committing its late model result", async () => { vi.useFakeTimers(); const pending = deferred(); diff --git a/src/gateway/session-companion.ts b/src/gateway/session-companion.ts index 58bd998293a5..a9086770028e 100644 --- a/src/gateway/session-companion.ts +++ b/src/gateway/session-companion.ts @@ -14,6 +14,7 @@ export type SessionCompanionService = { sessionKey: string; question: string; connId: string; + signal?: AbortSignal; }) => Promise; state: (sessionKey: string) => SessionsCompanionStateResult; reset: (sessionKey: string) => void; diff --git a/ui/src/api/gateway.node.test.ts b/ui/src/api/gateway.node.test.ts index 64a139d0bdb2..de9c3bcfbbe1 100644 --- a/ui/src/api/gateway.node.test.ts +++ b/ui/src/api/gateway.node.test.ts @@ -18,7 +18,6 @@ import { readCloudSessionRecovery, writeCloudSessionRecovery, } from "../lib/sessions/cloud-recovery.ts"; -import { requestSessionCompanionAnswer } from "../pages/chat/chat-session-companion.ts"; import { createStorageMock } from "../test-helpers/storage.ts"; const wsInstances = vi.hoisted((): MockWebSocket[] => []); @@ -457,7 +456,6 @@ describe("GatewayBrowserClient", () => { GATEWAY_CLIENT_CAPS.TOOL_EVENTS, GATEWAY_CLIENT_CAPS.INLINE_WIDGETS, GATEWAY_CLIENT_CAPS.UI_COMMANDS, - GATEWAY_CLIENT_CAPS.SESSION_COMPANION_PROGRESS, ]); expect(connectFrame.params?.scopes).toEqual([...CONTROL_UI_OPERATOR_SCOPES]); }); @@ -720,7 +718,7 @@ describe("GatewayBrowserClient", () => { }); }); - it("correlates companion accepted progress with final and error responses", async () => { + it("settles a companion ask from one final response", async () => { const client = new GatewayBrowserClient({ url: "ws://127.0.0.1:18789", token: "shared-auth-token", @@ -733,87 +731,24 @@ describe("GatewayBrowserClient", () => { payload: { type: "hello-ok", protocol: 4, auth: { role: "operator", scopes: [] } }, }); - const prepared = vi.fn(); - const answer = requestSessionCompanionAnswer( - client, - "agent:main:main", - "What changed?", - prepared, + const answer = client.request( + "sessions.companion.ask", + { sessionKey: "agent:main:main", question: "What changed?" }, + { timeoutMs: 70_000 }, ); - const answerFrame = JSON.parse(ws.sent.at(-1) ?? "{}") as { id?: string; method?: string }; - expect(answerFrame.method).toBe("sessions.companion.ask"); - ws.emitMessage({ - type: "res", - id: answerFrame.id, - ok: true, - payload: { status: "accepted", empty: false }, - }); - expect(prepared).toHaveBeenCalledOnce(); - ws.emitMessage({ - type: "res", - id: answerFrame.id, - ok: true, - payload: { answer: "The fix changed.", ts: 4 }, - }); - await expect(answer).resolves.toEqual({ answer: "The fix changed.", ts: 4 }); - - const failedPrepared = vi.fn(); - const failed = requestSessionCompanionAnswer( - client, - "agent:main:main", - "Retry?", - failedPrepared, - ); - const failedFrame = JSON.parse(ws.sent.at(-1) ?? "{}") as { id?: string }; - ws.emitMessage({ - type: "res", - id: failedFrame.id, - ok: true, - payload: { status: "accepted", empty: false }, - }); - ws.emitMessage({ - type: "res", - id: failedFrame.id, - ok: false, - error: { - code: "UNAVAILABLE", - message: "history unavailable", - details: { reason: "context-unavailable" }, - retryable: true, - }, - }); - expect(failedPrepared).toHaveBeenCalledOnce(); - await expect(failed).rejects.toMatchObject({ - details: { reason: "context-unavailable" }, - retryable: true, - }); - }); - - it("accepts a legacy single final companion response without progress", async () => { - const client = new GatewayBrowserClient({ - url: "ws://127.0.0.1:18789", - token: "shared-auth-token", - }); - const { ws, connectFrame } = await startConnect(client); - ws.emitMessage({ - type: "res", - id: connectFrame.id, - ok: true, - payload: { type: "hello-ok", protocol: 4, auth: { role: "operator", scopes: [] } }, - }); - - const prepared = vi.fn(); - const answer = requestSessionCompanionAnswer(client, "agent:main:main", "Legacy?", prepared); - const frame = JSON.parse(ws.sent.at(-1) ?? "{}") as { id?: string }; + const frame = JSON.parse(ws.sent.at(-1) ?? "{}") as { id?: string; method?: string }; + expect(frame.method).toBe("sessions.companion.ask"); ws.emitMessage({ type: "res", id: frame.id, ok: true, - payload: { answer: "Still works.", ts: 5 }, + payload: { answer: "The companion was simplified.", ts: 4 }, }); - await expect(answer).resolves.toEqual({ answer: "Still works.", ts: 5 }); - expect(prepared).not.toHaveBeenCalled(); + await expect(answer).resolves.toEqual({ + answer: "The companion was simplified.", + ts: 4, + }); }); it("tracks inbound activity and delegates forced reconnect to the shared socket", async () => { diff --git a/ui/src/api/gateway.ts b/ui/src/api/gateway.ts index 2a2a0e5233b1..11cac8e26778 100644 --- a/ui/src/api/gateway.ts +++ b/ui/src/api/gateway.ts @@ -500,7 +500,6 @@ export class GatewayBrowserClient { GATEWAY_CLIENT_CAPS.TOOL_EVENTS, GATEWAY_CLIENT_CAPS.INLINE_WIDGETS, GATEWAY_CLIENT_CAPS.UI_COMMANDS, - GATEWAY_CLIENT_CAPS.SESSION_COMPANION_PROGRESS, ], auth: buildGatewayConnectAuth(selectedAuth), userAgent: navigator.userAgent, diff --git a/ui/src/i18n/locales/en.ts b/ui/src/i18n/locales/en.ts index 2cdb29a7cbe8..7adf239635dc 100644 --- a/ui/src/i18n/locales/en.ts +++ b/ui/src/i18n/locales/en.ts @@ -5028,8 +5028,7 @@ export const en: TranslationMap = { askLabel: "Ask the session companion", askPlaceholder: "Ask a question", askSubmit: "Ask", - askReading: "Reading this session…", - askAnswering: "Answering…", + askPending: "Answering from this session…", askBusy: "The companion is already answering a question.", askHistoryUnavailable: "Couldn't load this session's history.", askMissing: "This session is no longer available.", diff --git a/ui/src/pages/chat/chat-pane-base.ts b/ui/src/pages/chat/chat-pane-base.ts index 8bac0a6b06ee..67517f87b787 100644 --- a/ui/src/pages/chat/chat-pane-base.ts +++ b/ui/src/pages/chat/chat-pane-base.ts @@ -251,24 +251,8 @@ export abstract class ChatPaneBase extends OpenClawLightDomElement { return; } const client = state.client; - const connectionGeneration = this.connectionGeneration; - await this.sessionCompanionThreads.submit( - sessionKey, - question, - (key, value, onPrepared) => requestSessionCompanionAnswer(client, key, value, onPrepared), - () => - this.state === state && - state.connected && - 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); - }, + await this.sessionCompanionThreads.submit(sessionKey, question, (key, value) => + requestSessionCompanionAnswer(client, key, value), ); }; diff --git a/ui/src/pages/chat/chat-session-companion.ts b/ui/src/pages/chat/chat-session-companion.ts index 53ca2118cb26..73c9a8ae99b7 100644 --- a/ui/src/pages/chat/chat-session-companion.ts +++ b/ui/src/pages/chat/chat-session-companion.ts @@ -8,6 +8,7 @@ import type { GatewayBrowserClient } from "../../api/gateway.ts"; const COMPANION_BUSY_DETAIL_CODE = "SESSION_COMPANION_BUSY"; const MAX_COMPANION_EXCHANGES = 24; +const COMPANION_ASK_TIMEOUT_MS = 70_000; export type ChatSessionCompanionThread = { exchanges: SessionCompanionExchange[]; @@ -22,7 +23,6 @@ export type ChatSessionCompanionThread = { | "unavailable" | null; retryable?: boolean; - phase?: "answering" | "reading" | null; draft: string; }; @@ -67,7 +67,6 @@ function createThread(): MutableCompanionThread { failedQuestion: null, hint: null, retryable: false, - phase: null, draft: "", revision: 0, }; @@ -140,13 +139,7 @@ export class ChatSessionCompanionThreads { async submit( sessionKey: string, question: string, - ask: ( - sessionKey: string, - question: string, - onPrepared: () => void, - ) => Promise, - isCurrent: () => boolean = () => true, - reload?: (sessionKey: string) => Promise, + ask: (sessionKey: string, question: string) => Promise, ): Promise { const key = sessionKey.trim(); const normalized = question.trim(); @@ -161,66 +154,16 @@ export class ChatSessionCompanionThreads { thread.failedQuestion = null; thread.hint = null; thread.retryable = false; - thread.phase = "reading"; thread.draft = ""; thread.revision += 1; 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()) { - return; - } - thread.phase = "answering"; - thread.revision += 1; - this.notify(); - }); + const result = await ask(key, normalized); if (this.submissionTokens.get(key) !== token) { return; } - if (!isCurrent()) { - 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, { question: normalized, answer: result.answer, ts: result.ts }, @@ -229,16 +172,6 @@ export class ChatSessionCompanionThreads { if (this.submissionTokens.get(key) !== token) { return; } - if (!isCurrent()) { - 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 = @@ -253,12 +186,11 @@ export class ChatSessionCompanionThreads { : reason === "utility-model-unavailable" ? "model-unavailable" : "unavailable"; - thread.retryable = errorIsRetryable(error); + thread.retryable = errorIsRetryable(error) || reason === null; } finally { if (this.submissionTokens.get(key) === token) { this.submissionTokens.delete(key); thread.pendingQuestion = null; - thread.phase = null; thread.revision += 1; this.notify(); } @@ -295,12 +227,11 @@ export function requestSessionCompanionAnswer( client: Pick, sessionKey: string, question: string, - onPrepared: () => void, ): Promise { return client.request( "sessions.companion.ask", { sessionKey, question }, - { expectFinal: true, onAccepted: onPrepared }, + { timeoutMs: COMPANION_ASK_TIMEOUT_MS }, ); } diff --git a/ui/src/pages/chat/chat-session-rail.test.ts b/ui/src/pages/chat/chat-session-rail.test.ts index edc94d7721be..b7bcb85e7f2d 100644 --- a/ui/src/pages/chat/chat-session-rail.test.ts +++ b/ui/src/pages/chat/chat-session-rail.test.ts @@ -206,35 +206,26 @@ describe("ChatSessionRailState", () => { describe("ChatSessionCompanionThreads", () => { it("uses the exact companion RPC methods and payloads", async () => { - const onPrepared = vi.fn(); - const request = vi.fn( - async ( - method: string, - _params: Record, - options?: { onAccepted?: (payload: unknown) => void }, - ) => { - if (method === "sessions.companion.ask") { - options?.onAccepted?.({ status: "accepted", empty: false }); - return { answer: "Answer", ts: 1 }; - } - if (method === "sessions.companion.state") { - return { exchanges: [] }; - } - return { ok: true as const }; - }, - ); + const request = vi.fn(async (method: string) => { + if (method === "sessions.companion.ask") { + return { answer: "Answer", ts: 1 }; + } + if (method === "sessions.companion.state") { + return { exchanges: [] }; + } + return { ok: true as const }; + }); const client = { request: request as GatewayBrowserClient["request"] }; - await requestSessionCompanionAnswer(client, "one", "Question", onPrepared); + await requestSessionCompanionAnswer(client, "one", "Question"); await requestSessionCompanionState(client, "one"); await resetSessionCompanion(client, "one"); - expect(onPrepared).toHaveBeenCalledOnce(); expect(request.mock.calls).toEqual([ [ "sessions.companion.ask", { sessionKey: "one", question: "Question" }, - { expectFinal: true, onAccepted: onPrepared }, + { timeoutMs: 70_000 }, ], ["sessions.companion.state", { sessionKey: "one" }], ["sessions.companion.reset", { sessionKey: "one" }], @@ -261,32 +252,25 @@ describe("ChatSessionCompanionThreads", () => { }); it("moves a composer submission through pending to a timestamped answer", async () => { - let markPrepared!: () => void; let resolveAnswer!: (value: { answer: string; ts: number }) => void; const threads = new ChatSessionCompanionThreads(); threads.setDraft("one", "Why is it rerunning that test?"); const pending = threads.submit( "one", threads.view("one").draft, - (_sessionKey, _question, onPrepared) => { - markPrepared = onPrepared; - return new Promise((resolve) => { + () => + new Promise((resolve) => { resolveAnswer = resolve; - }); - }, + }), ); expect(threads.view("one").pendingQuestion).toBe("Why is it rerunning that test?"); - expect(threads.view("one").phase).toBe("reading"); expect(threads.view("one").draft).toBe(""); - markPrepared(); - await vi.waitFor(() => expect(threads.view("one").phase).toBe("answering")); resolveAnswer({ answer: "It is verifying the focused regression.", ts: 42 }); await pending; expect(threads.view("one")).toMatchObject({ pendingQuestion: null, - phase: null, exchanges: [ { question: "Why is it rerunning that test?", @@ -313,23 +297,34 @@ describe("ChatSessionCompanionThreads", () => { }); }); - it.each([ - { - reason: "rate-limited", + it("preserves a context failure for an explicit retry", async () => { + const threads = new ChatSessionCompanionThreads(); + await threads.submit("one", "What changed?", async () => { + throw Object.assign(new Error("history unavailable"), { + details: { reason: "context-unavailable" }, + retryable: true, + }); + }); + + expect(threads.view("one")).toMatchObject({ + failedQuestion: "What changed?", + hint: "history-unavailable", + pendingQuestion: null, retryable: true, - hint: "rate-limited", - }, - { - reason: "utility-model-unavailable", - retryable: false, - hint: "model-unavailable", - }, - { - reason: "unavailable", - retryable: false, - hint: "unavailable", - }, - ] as const)("maps $reason without falsely blaming session history", async (expected) => { + }); + await threads.hydrate("one", async () => ({ exchanges: [] })); + expect(threads.view("one")).toMatchObject({ + failedQuestion: "What changed?", + hint: "history-unavailable", + retryable: true, + }); + }); + + it.each([ + { reason: "rate-limited", retryable: true, hint: "rate-limited" }, + { reason: "utility-model-unavailable", retryable: false, hint: "model-unavailable" }, + { reason: "unavailable", retryable: false, hint: "unavailable" }, + ] as const)("maps $reason to its specific retry state", async (expected) => { const threads = new ChatSessionCompanionThreads(); await threads.submit("one", "What changed?", async () => { throw Object.assign(new Error(expected.reason), { @@ -344,55 +339,16 @@ describe("ChatSessionCompanionThreads", () => { }); }); - it("preserves an unavailable question for an explicit retry", async () => { + it("hydrates an answer committed before a disconnect response was lost", async () => { const threads = new ChatSessionCompanionThreads(); await threads.submit("one", "What changed?", async () => { - throw Object.assign(new Error("unavailable"), { - details: { reason: "context-unavailable" }, - retryable: true, - }); + throw new Error("socket closed"); }); expect(threads.view("one")).toMatchObject({ failedQuestion: "What changed?", - hint: "history-unavailable", - pendingQuestion: null, + hint: "unavailable", retryable: true, }); - await threads.hydrate("one", async () => ({ exchanges: [] })); - expect(threads.view("one")).toMatchObject({ - failedQuestion: "What changed?", - hint: "history-unavailable", - }); - - await threads.submit( - "one", - threads.view("one").failedQuestion ?? "", - async (_sessionKey, _question, prepared) => { - prepared(); - return { answer: "The focused test changed.", ts: 2 }; - }, - ); - expect(threads.view("one")).toMatchObject({ - failedQuestion: null, - hint: null, - exchanges: [ - { - question: "What changed?", - answer: "The focused test changed.", - ts: 2, - }, - ], - }); - }); - - it("clears a retry error when hydration confirms that exact answer committed", async () => { - const threads = new ChatSessionCompanionThreads(); - await threads.submit("one", "What changed?", async () => { - throw Object.assign(new Error("disconnected"), { - details: { reason: "context-unavailable" }, - retryable: true, - }); - }); await threads.hydrate("one", async () => ({ exchanges: [{ question: "What changed?", answer: "The fix committed.", ts: 4 }], @@ -401,112 +357,11 @@ describe("ChatSessionCompanionThreads", () => { expect(threads.view("one")).toMatchObject({ failedQuestion: null, hint: null, + retryable: false, exchanges: [{ question: "What changed?", answer: "The fix committed.", ts: 4 }], }); }); - it("rejects an answer that settles after the owning connection changes", async () => { - let current = true; - let resolveAnswer!: (value: { answer: string; ts: number }) => void; - const threads = new ChatSessionCompanionThreads(); - const pending = threads.submit( - "one", - "Which connection owns this?", - (_sessionKey, _question, prepared) => { - prepared(); - return new Promise((resolve) => { - resolveAnswer = resolve; - }); - }, - () => 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; - resolveAnswer({ answer: "stale answer", ts: 3 }); - await pending; - - expect(threads.view("one")).toMatchObject({ - exchanges: [ - { - question: "Which connection owns this?", - answer: "stale answer", - ts: 3, - }, - ], - failedQuestion: null, - hint: null, - pendingQuestion: null, - retryable: false, - }); - }); - - it("makes a rejected stale connection settlement retryable", async () => { - let current = true; - let rejectAnswer!: (error: Error) => void; - const threads = new ChatSessionCompanionThreads(); - const pending = threads.submit( - "one", - "Which connection rejected this?", - (_sessionKey, _question, prepared) => { - prepared(); - return new Promise((_resolve, reject) => { - rejectAnswer = reject; - }); - }, - () => current, - async () => ({ exchanges: [] }), - ); - await vi.waitFor(() => expect(threads.view("one").phase).toBe("answering")); - current = false; - rejectAnswer(new Error("old socket closed")); - await pending; - - expect(threads.view("one")).toMatchObject({ - exchanges: [], - failedQuestion: "Which connection rejected this?", - hint: "history-unavailable", - retryable: true, - }); - }); - - it.each(["resolve", "reject"] as const)( - "does not resurrect a reset pending request after late $outcome", - async (outcome) => { - let resolveAnswer!: (value: { answer: string; ts: number }) => void; - let rejectAnswer!: (error: Error) => void; - const threads = new ChatSessionCompanionThreads(); - const pending = threads.submit("one", "Will reset keep this?", () => { - return new Promise((resolve, reject) => { - resolveAnswer = resolve; - rejectAnswer = reject; - }); - }); - - await threads.reset("one", async () => ({ ok: true })); - if (outcome === "resolve") { - resolveAnswer({ answer: "late answer", ts: 5 }); - } else { - rejectAnswer(new Error("late error")); - } - await pending; - - expect(threads.view("one")).toMatchObject({ - exchanges: [], - failedQuestion: null, - pendingQuestion: null, - }); - }, - ); - it("clears local state only after the reset RPC succeeds", async () => { const threads = new ChatSessionCompanionThreads(); await threads.hydrate("one", async () => ({ @@ -522,6 +377,38 @@ describe("ChatSessionCompanionThreads", () => { await threads.reset("one", async () => ({ ok: true as const })); expect(threads.view("one").exchanges).toEqual([]); }); + + it.each(["resolve", "reject"] as const)( + "does not resurrect a reset request after a late $outcome", + async (outcome) => { + let resolveAnswer!: (value: { answer: string; ts: number }) => void; + let rejectAnswer!: (error: Error) => void; + const threads = new ChatSessionCompanionThreads(); + const pending = threads.submit( + "one", + "Will reset keep this?", + () => + new Promise((resolve, reject) => { + resolveAnswer = resolve; + rejectAnswer = reject; + }), + ); + + await threads.reset("one", async () => ({ ok: true as const })); + if (outcome === "resolve") { + resolveAnswer({ answer: "late answer", ts: 5 }); + } else { + rejectAnswer(new Error("late error")); + } + await pending; + + expect(threads.view("one")).toMatchObject({ + exchanges: [], + failedQuestion: null, + pendingQuestion: null, + }); + }, + ); }); describe("ChatSessionRailElement", () => { @@ -583,47 +470,31 @@ describe("ChatSessionRailElement", () => { expect(element.querySelector(".chat-session-rail__timestamp")?.textContent).toContain("as of"); }); - it("renders truthful reading and answering phases", async () => { - const element = await mount({ - companion: { - exchanges: [], - pendingQuestion: "What changed?", - failedQuestion: null, - hint: null, - phase: "reading", - draft: "", - }, - }); - - expect(element.textContent).toContain("Reading this session…"); - expect((element.querySelector(".chat-session-rail__input") as HTMLInputElement).disabled).toBe( - true, - ); - - element.companion = { ...element.companion, phase: "answering" }; - await element.updateComplete; - expect(element.textContent).toContain("Answering…"); - }); - - it("keeps a failed question visible and retries it from the error state", async () => { + it("renders one pending state and retries a retryable failure", async () => { const onSubmit = vi.fn(); const element = await mount({ onSubmit, companion: { exchanges: [], - pendingQuestion: null, - failedQuestion: "What changed?", - hint: "history-unavailable", - retryable: true, - phase: null, + pendingQuestion: "What changed?", + failedQuestion: null, + hint: null, draft: "", }, }); + expect(element.textContent).toContain("Answering from this session…"); + element.companion = { + exchanges: [], + pendingQuestion: null, + failedQuestion: "What changed?", + hint: "history-unavailable", + retryable: true, + draft: "", + }; + await element.updateComplete; expect(element.textContent).toContain("Couldn't load this session's history."); - const retry = element.querySelector(".chat-session-rail__retry") as HTMLButtonElement; - expect(retry.textContent?.trim()).toBe("Retry"); - retry.click(); + (element.querySelector(".chat-session-rail__retry") as HTMLButtonElement).click(); expect(onSubmit).toHaveBeenCalledWith("What changed?"); }); diff --git a/ui/src/pages/chat/components/chat-session-rail.ts b/ui/src/pages/chat/components/chat-session-rail.ts index 84cf6ca51f02..7921c7786541 100644 --- a/ui/src/pages/chat/components/chat-session-rail.ts +++ b/ui/src/pages/chat/components/chat-session-rail.ts @@ -235,7 +235,6 @@ export class ChatSessionRailElement extends OpenClawLightDomElement { failedQuestion: null, hint: null, retryable: false, - phase: null, draft: "", }; @property({ attribute: false }) connected = false; @@ -535,13 +534,7 @@ export class ChatSessionRailElement extends OpenClawLightDomElement { ? html`
${this.companion.pendingQuestion}
-
- ${t( - this.companion.phase === "answering" - ? "chat.rail.askAnswering" - : "chat.rail.askReading", - )} -
+
${t("chat.rail.askPending")}
` : nothing} @@ -698,11 +691,7 @@ export class ChatSessionRailElement extends OpenClawLightDomElement { aria-label=${t("chat.rail.askLabel")} .value=${this.companion.draft} placeholder=${this.companion.pendingQuestion - ? t( - this.companion.phase === "answering" - ? "chat.rail.askAnswering" - : "chat.rail.askReading", - ) + ? t("chat.rail.askPending") : t("chat.rail.askPlaceholder")} ?disabled=${!this.connected || this.companion.pendingQuestion !== null} @input=${(event: InputEvent) => {