mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix: simplify session companion grounding
This commit is contained in:
Shakker
+1
-1
@@ -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.
|
||||
|
||||
+6
-6
@@ -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)
|
||||
|
||||
|
||||
@@ -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 <question>` or `/side <question>` 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 <question>` or `/side <question>` 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.
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
@@ -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<TExtra extends object>(
|
||||
/** 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<typeof resolveVisibleMessagePositions>[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<TExtra extends object>(
|
||||
);
|
||||
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<TExtra extends object>(
|
||||
scannedMessages: positions.length,
|
||||
serializedBytes: 0,
|
||||
totalMessages,
|
||||
...extra,
|
||||
};
|
||||
}
|
||||
const db = getActiveTranscriptKysely(projection.database);
|
||||
@@ -536,40 +522,10 @@ function readBoundedMessageTailPage<TExtra extends object>(
|
||||
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,
|
||||
|
||||
@@ -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<string, ResetMessageWindowCacheEntry>();
|
||||
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<ResetWindowDatabase>(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<number>`LENGTH(CAST(event.event_json AS BLOB))`;
|
||||
}
|
||||
|
||||
function sqliteBoundaryJsonValid() {
|
||||
return /* kysely-allow-raw: boundary JSON validity is part of the fail-closed contract. */ sql<number>`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<number>`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<ReturnType<typeof readLatestActiveBoundary>>,
|
||||
): 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,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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<void>((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",
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
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<string, SessionCompanionActiveAsk>();
|
||||
const preparations = new Map<string, Promise<SessionCompanionThread>>();
|
||||
const admissions: Array<{ connId: string; admittedAt: number }> = [];
|
||||
|
||||
const resolveTarget = (sessionKey: string) => {
|
||||
@@ -398,69 +375,55 @@ export function createSessionCompanionAskRuntime(params: SessionCompanionAskRunt
|
||||
): Promise<SessionCompanionThread> => {
|
||||
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<never>((_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;
|
||||
},
|
||||
};
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
};
|
||||
|
||||
@@ -1,39 +0,0 @@
|
||||
type SessionCompanionProgressListener = (payload: { empty: boolean }) => void;
|
||||
|
||||
const listeners = new Map<string, SessionCompanionProgressListener>();
|
||||
|
||||
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.
|
||||
}
|
||||
}
|
||||
@@ -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<typeof vi.fn>;
|
||||
reset?: ReturnType<typeof vi.fn>;
|
||||
},
|
||||
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,
|
||||
|
||||
@@ -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 }) => {
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
@@ -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-session-reference>private context</private-session-reference>",
|
||||
});
|
||||
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<SessionCompanionAskError>);
|
||||
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<string>();
|
||||
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<SessionCompanionAskError>);
|
||||
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<string>();
|
||||
|
||||
@@ -14,6 +14,7 @@ export type SessionCompanionService = {
|
||||
sessionKey: string;
|
||||
question: string;
|
||||
connId: string;
|
||||
signal?: AbortSignal;
|
||||
}) => Promise<SessionsCompanionAskResult>;
|
||||
state: (sessionKey: string) => SessionsCompanionStateResult;
|
||||
reset: (sessionKey: string) => void;
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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),
|
||||
);
|
||||
};
|
||||
|
||||
|
||||
@@ -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<SessionsCompanionAskResult>,
|
||||
isCurrent: () => boolean = () => true,
|
||||
reload?: (sessionKey: string) => Promise<SessionsCompanionStateResult>,
|
||||
ask: (sessionKey: string, question: string) => Promise<SessionsCompanionAskResult>,
|
||||
): Promise<void> {
|
||||
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<GatewayBrowserClient, "request">,
|
||||
sessionKey: string,
|
||||
question: string,
|
||||
onPrepared: () => void,
|
||||
): Promise<SessionsCompanionAskResult> {
|
||||
return client.request<SessionsCompanionAskResult>(
|
||||
"sessions.companion.ask",
|
||||
{ sessionKey, question },
|
||||
{ expectFinal: true, onAccepted: onPrepared },
|
||||
{ timeoutMs: COMPANION_ASK_TIMEOUT_MS },
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string, unknown>,
|
||||
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?");
|
||||
});
|
||||
|
||||
|
||||
@@ -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`
|
||||
<article class="chat-session-rail__exchange chat-session-rail__exchange--pending">
|
||||
<div class="chat-session-rail__question">${this.companion.pendingQuestion}</div>
|
||||
<div class="chat-session-rail__hint">
|
||||
${t(
|
||||
this.companion.phase === "answering"
|
||||
? "chat.rail.askAnswering"
|
||||
: "chat.rail.askReading",
|
||||
)}
|
||||
</div>
|
||||
<div class="chat-session-rail__hint">${t("chat.rail.askPending")}</div>
|
||||
</article>
|
||||
`
|
||||
: 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) => {
|
||||
|
||||
Reference in New Issue
Block a user