diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index fe74309213a..352f95c66e6 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -3370,6 +3370,7 @@ public struct SessionCatalogTranscriptItem: Codable, Sendable { public struct SessionsCatalogListParams: Codable, Sendable { public let catalogid: String? public let agentid: String? + public let progressid: String? public let search: String? public let limitperhost: Int? public let hostids: [String]? @@ -3378,6 +3379,7 @@ public struct SessionsCatalogListParams: Codable, Sendable { public init( catalogid: String? = nil, agentid: String? = nil, + progressid: String? = nil, search: String? = nil, limitperhost: Int? = nil, hostids: [String]? = nil, @@ -3385,6 +3387,7 @@ public struct SessionsCatalogListParams: Codable, Sendable { { self.catalogid = catalogid self.agentid = agentid + self.progressid = progressid self.search = search self.limitperhost = limitperhost self.hostids = hostids @@ -3394,6 +3397,7 @@ public struct SessionsCatalogListParams: Codable, Sendable { private enum CodingKeys: String, CodingKey { case catalogid = "catalogId" case agentid = "agentId" + case progressid = "progressId" case search case limitperhost = "limitPerHost" case hostids = "hostIds" diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index a80e4e83f8d..4b73746bb92 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -7e717544b401b5ca3447d5383e241f678b6c343c089b3f430c2fc48244f1b3c8 plugin-sdk-api-baseline.json -0b4bc1cfc71177b5f5e0d146c48d86d48ff0eea1a075358adb7583ee35af54b9 plugin-sdk-api-baseline.jsonl +5e2ce14f9b24efbaa43886bc2f82226f6ca83ba1e02520dc8812da839fa20e2f plugin-sdk-api-baseline.json +62a3c8c10295440d956d257b4e8cb981b0ed37a16f9976b9417ac3227fed9242 plugin-sdk-api-baseline.jsonl diff --git a/docs/plugins/codex-supervision.md b/docs/plugins/codex-supervision.md index b77c2b73c1a..e71e9bdb768 100644 --- a/docs/plugins/codex-supervision.md +++ b/docs/plugins/codex-supervision.md @@ -137,6 +137,12 @@ closed instead of risking the node or Gateway connection. Open the **Codex** group in the normal sessions sidebar. It lists the same sessions grouped by host. **Load more sessions** appends the next page from each host that has older rows, and those appended rows survive the sidebar's periodic refresh. +Each host appears as soon as its own native listing settles. The visible page +reconciles after node-connectivity changes, when it regains focus, and at most +every 30 seconds; a changed result gets a faster follow-up pass. Sessions created +in Codex Desktop, the CLI, or another native client therefore appear without a +full page reload. The first page follows Codex's own most-recently-updated order, +so a newly created native session is eligible immediately. Each returned search page scans a bounded number of native pages per host rather than sending the query to App Server, because native search can also match transcript previews. diff --git a/docs/plugins/sdk-entrypoints.md b/docs/plugins/sdk-entrypoints.md index cd593fb9563..de2d7f89037 100644 --- a/docs/plugins/sdk-entrypoints.md +++ b/docs/plugins/sdk-entrypoints.md @@ -147,7 +147,10 @@ export default definePluginEntry({ `openclaw/plugin-sdk/session-catalog` and `api.registerSessionCatalog({ id, label, list, read, continueSession?, archive? })`. Core owns the `sessions.catalog.*` Gateway methods; providers return host, - session, and normalized transcript projections without registering RPCs. + session, and normalized transcript projections without registering RPCs. A + list provider should call the optional `onHost(host)` callback as each host + settles; the returned host array remains required as the final compatibility + snapshot. - `kind` is deprecated: declare an exclusive slot (`"memory"` or `"context-engine"`) in the `openclaw.plugin.json` manifest `kind` field instead. Runtime-entry `kind` remains only as a compatibility fallback for diff --git a/docs/providers/anthropic.md b/docs/providers/anthropic.md index fbb1473801b..0848b4de01b 100644 --- a/docs/providers/anthropic.md +++ b/docs/providers/anthropic.md @@ -217,12 +217,15 @@ is bundled and enabled by default; a native macOS node advertises the read-only Claude session commands when the local `~/.claude/projects/` directory exists. Approve the node pairing upgrade when those commands first appear. -The sidebar groups rows by their Gateway or paired-node host, starts with the -newest bounded page from each host, and refreshes on the normal 30-second -cadence. Use **Load more sessions** below a catalog group to append the next page -for every host that has more history; appended rows stay visible and are -re-fetched to the same depth across refreshes. Catalog clients use -`sessions.catalog.list`; opening a row uses `sessions.catalog.read`. +The sidebar groups rows by their Gateway or paired-node host and shows each +host's newest bounded page as soon as that computer answers. It reconciles again +after host-connectivity changes, when the page regains focus, and at most every +30 seconds while visible, so Claude sessions created outside OpenClaw appear +without a reload. A changed catalog gets a faster follow-up pass. Use **Load more +sessions** below a catalog group to append the next page for every host that has +more history; appended rows stay visible and are re-fetched to the same depth +across refreshes. Catalog clients use `sessions.catalog.list`; opening a row uses +`sessions.catalog.read`. Terminal takeover resolves `claude` from the owning host user's login-shell PATH before the service/daemon PATH. This keeps app-launched sessions aligned diff --git a/docs/specs/codex-supervision.md b/docs/specs/codex-supervision.md index 4f3741631b1..46ef7c9f430 100644 --- a/docs/specs/codex-supervision.md +++ b/docs/specs/codex-supervision.md @@ -150,6 +150,13 @@ local App Server does not erase healthy hosts from the page. Connectivity is a host property, not a thread status: a failed host result contains no fresh session rows and does not project `offline` onto native threads. +The Control UI requests progressive catalog updates. Each local or paired host +appears when its own App Server listing settles; the aggregate response remains +the compatibility and recovery snapshot. The visible page reconciles after +connectivity changes, on focus, and at most every 30 seconds, with a faster pass +after changes. Native Codex sessions created in another client are therefore +eventually discovered without importing them into OpenClaw storage. + Catalog discovery is passive. Listing or reading metadata must not call `thread/resume`, subscribe the OpenClaw client to live thread requests, or answer an approval. diff --git a/docs/web/control-ui.md b/docs/web/control-ui.md index 42b61113bda..051424842c3 100644 --- a/docs/web/control-ui.md +++ b/docs/web/control-ui.md @@ -201,6 +201,7 @@ A **Search** field at the top of the sidebar opens the command palette (⌘K). C - Channels: built-in plus bundled/external plugin channels status, QR login, and per-channel config (`channels.status`, `web.login.*`, `config.patch`). - Channel probe refreshes keep the previous snapshot visible while slow provider checks finish, and label partial snapshots when a probe or audit exceeds its UI budget. - Sessions (a settings page under **Agents & Tools**, `/settings/sessions`): list configured-agent sessions by default, pin frequent sessions, rename them, archive or restore inactive sessions, fall back from stale unconfigured agent session keys, and apply per-session model/thinking/fast/verbose/trace/reasoning overrides (`sessions.list`, `sessions.patch`). Pinned sessions sort above recent unpinned sessions; archived sessions live in the Sessions page's archived view and keep their transcripts. Rows show an unread dot for sessions with activity since their last read, with mark-unread/mark-read actions (`sessions.patch { unread }`), and a Fork action that branches the transcript into a new session (`sessions.create { parentSessionKey, fork: true }`). Overview tiles above the table summarize the loaded roster (session count, live runs, unread sessions, total tokens), each row carries a kind glyph with a live-run dot, status renders as a plain dot plus label, and the Tokens column shows a context-window usage meter when the session reports token and context sizes. Row management actions live in a per-row menu (kebab button or right-click) mirroring the sidebar's session menu, and the row drawer carries the agent runtime and run duration alongside the other session details. + - Native Claude and Codex sidebar catalogs stream one host at a time, then reconcile after node connectivity changes, on page focus, and at most every 30 seconds while visible. Catalog changes trigger a faster follow-up pass, so sessions created in the native tools appear without reloading the Control UI. - Session grouping: a Group by control organizes the sessions table into sections by custom groups, channel, kind, agent, or date. Custom groups persist per session via `sessions.patch` (`category`), so sessions started from message channels (Discord, Telegram, WhatsApp, ...) can be categorized too; assign groups by dragging rows onto a section, or with the per-row group selector, and create groups with the New group action. - Memory (a tab on the Agents page, scoped to the selected agent): dreaming status, enable/disable toggle, and Dream Diary reader (`doctor.memory.status`, `doctor.memory.dreamDiary`, `config.patch`). - Import Memory (a settings page under **Agents & Tools**, `/settings/memory-import`): preview and copy local Claude Code auto-memory, Codex consolidated memory, or Hermes memory files into the selected agent workspace (`migrations.memory.plan`, `migrations.memory.apply`). diff --git a/extensions/anthropic/session-catalog.test.ts b/extensions/anthropic/session-catalog.test.ts index 35f82321cc5..fcfd006a4d7 100644 --- a/extensions/anthropic/session-catalog.test.ts +++ b/extensions/anthropic/session-catalog.test.ts @@ -1559,6 +1559,64 @@ describe("Claude session catalog", () => { } }); + it("publishes a paired-node page that finishes after the fail-soft response", async () => { + vi.useFakeTimers(); + try { + let resolveInvoke!: (value: unknown) => void; + const invokeResult = new Promise((resolve) => { + resolveInvoke = resolve; + }); + const invoke = vi.fn(async () => await invokeResult); + const provider = captureCatalogProvider({ + nodes: { + list: vi.fn().mockResolvedValue({ + nodes: [ + { + nodeId: "slow-node", + displayName: "Slow node", + connected: true, + commands: [CLAUDE_SESSIONS_LIST_COMMAND], + }, + ], + }), + invoke, + }, + } as unknown as PluginRuntime); + const onHost = vi.fn(); + const pending = provider.list({ hostIds: ["node:slow-node"], onHost }); + + await vi.advanceTimersByTimeAsync(8_000); + await expect(pending).resolves.toEqual([ + expect.objectContaining({ error: expect.objectContaining({ code: "NODE_INVOKE_FAILED" }) }), + ]); + expect(onHost).not.toHaveBeenCalled(); + + resolveInvoke({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: "late-thread", + status: "stored", + source: "claude-cli", + modelProvider: "anthropic", + archived: false, + }, + ], + }), + }); + await vi.advanceTimersByTimeAsync(0); + + expect(onHost).toHaveBeenCalledWith( + expect.objectContaining({ + hostId: "node:slow-node", + sessions: [expect.objectContaining({ threadId: "late-thread" })], + }), + ); + } finally { + vi.useRealTimers(); + } + }); + it("starts paired-node discovery while the local catalog is still reading", async () => { const home = await createHome(); process.env.HOME = home; diff --git a/extensions/anthropic/session-catalog.ts b/extensions/anthropic/session-catalog.ts index 023ece08b99..63b0d892813 100644 --- a/extensions/anthropic/session-catalog.ts +++ b/extensions/anthropic/session-catalog.ts @@ -993,6 +993,7 @@ function parseGatewayQuery(value: unknown): { async function listClaudeSessionCatalog(params: { runtime: PluginRuntime; query?: unknown; + onHost?: (host: ClaudeSessionCatalogHost) => void; }): Promise { const query = parseGatewayQuery(params.query); const requested = query.hostIds ? new Set(query.hostIds) : undefined; @@ -1030,6 +1031,11 @@ async function listClaudeSessionCatalog(params: { })(), ] : []; + for (const host of localHosts) { + if (params.onHost) { + void host.then(params.onHost).catch(() => undefined); + } + } const wantsNodes = !requested || query.hostIds?.some((hostId) => hostId.startsWith("node:")); if (!wantsNodes) { return { hosts: await Promise.all(localHosts) }; @@ -1038,18 +1044,17 @@ async function listClaudeSessionCatalog(params: { try { nodes = (await params.runtime.nodes.list()).nodes; } catch (error) { + const registryHost: ClaudeSessionCatalogHost = { + hostId: "node:registry", + label: "Paired nodes", + kind: "node", + connected: false, + sessions: [], + error: createNodeListFailedError(error), + }; + params.onHost?.(registryHost); return { - hosts: [ - ...(await Promise.all(localHosts)), - { - hostId: "node:registry", - label: "Paired nodes", - kind: "node", - connected: false, - sessions: [], - error: createNodeListFailedError(error), - }, - ], + hosts: [...(await Promise.all(localHosts)), registryHost], }; } const eligible = nodes @@ -1078,14 +1083,16 @@ async function listClaudeSessionCatalog(params: { ...catalogTerminal.claudeNodeTerminalCapability(node), }; if (node.connected !== true) { - return Object.assign(common, { + const host: ClaudeSessionCatalogHost = Object.assign({}, common, { sessions: [], error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, }); + params.onHost?.(host); + return host; } - try { - const raw = await withTimeout( - params.runtime.nodes.invoke({ + const eventualHost = Promise.resolve() + .then(async () => { + const raw = await params.runtime.nodes.invoke({ nodeId: node.nodeId, command: CLAUDE_SESSIONS_LIST_COMMAND, params: { @@ -1095,13 +1102,30 @@ async function listClaudeSessionCatalog(params: { }, timeoutMs: NODE_INVOKE_TIMEOUT_MS, scopes: ["operator.write"], - }), - NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS, - { message: "paired node Claude session catalog timed out" }, + }); + return Object.assign({}, common, parseCatalogPage(unwrapNodePayload(raw))); + }) + .catch( + (): ClaudeSessionCatalogHost => + Object.assign({}, common, { + sessions: [], + error: { + code: "NODE_INVOKE_FAILED", + message: "Paired node Claude sessions are unavailable", + }, + }), ); - return Object.assign(common, parseCatalogPage(unwrapNodePayload(raw))); + if (params.onHost) { + // The fail-soft response can finish first; the original node invoke still + // publishes its authoritative host page whenever cold discovery completes. + void eventualHost.then(params.onHost).catch(() => undefined); + } + try { + return await withTimeout(eventualHost, NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS, { + message: "paired node Claude session catalog timed out", + }); } catch { - return Object.assign(common, { + return Object.assign({}, common, { sessions: [], error: { code: "NODE_INVOKE_FAILED", @@ -1444,8 +1468,15 @@ export function registerClaudeSessionCatalog(api: OpenClawPluginApi): void { list: async (query) => { const adopted = listBoundClaudeSessions(api); const localCliAvailable = catalogTerminal.isClaudeCliAvailable(); - const result = await listClaudeSessionCatalog({ runtime: api.runtime, query }); - return result.hosts.map((host) => toGenericClaudeHost(host, adopted, localCliAvailable)); + const { onHost, ...gatewayQuery } = query; + const mapHost = (host: ClaudeSessionCatalogHost) => + toGenericClaudeHost(host, adopted, localCliAvailable); + const result = await listClaudeSessionCatalog({ + runtime: api.runtime, + query: gatewayQuery, + ...(onHost ? { onHost: (host) => onHost(mapHost(host)) } : {}), + }); + return result.hosts.map(mapHost); }, read: async (request) => { const page = await readClaudeSessionTranscript({ diff --git a/extensions/codex/src/session-catalog-node-continue.ts b/extensions/codex/src/session-catalog-node-continue.ts index ec8cd0383e3..5dfaeda33b6 100644 --- a/extensions/codex/src/session-catalog-node-continue.ts +++ b/extensions/codex/src/session-catalog-node-continue.ts @@ -83,6 +83,7 @@ export async function listPairedNode(params: { node: CatalogNode; query: CodexSessionCatalogParams; adoptedSessions: ReadonlyMap; + onHost?: (host: CodexSessionCatalogHost) => void; }): Promise { const hostId = `node:${params.node.nodeId}`; const common = { @@ -93,16 +94,18 @@ export async function listPairedNode(params: { canContinueCodex: canContinueCodexOnNode(params.node), }; if (params.node.connected !== true) { - return { + const host = { ...common, connected: false, sessions: [], error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, }; + params.onHost?.(host); + return host; } - try { - const raw = await withTimeout( - params.runtime.nodes.invoke({ + const eventualHost = Promise.resolve() + .then(async () => { + const raw = await params.runtime.nodes.invoke({ nodeId: params.node.nodeId, command: CODEX_APP_SERVER_THREADS_LIST_COMMAND, params: { @@ -112,23 +115,40 @@ export async function listPairedNode(params: { }, timeoutMs: NODE_INVOKE_TIMEOUT_MS, scopes: ["operator.write"], - }), + }); + const page = filterCatalogPageByTitle( + parseCatalogPage(unwrapNodeInvokePayload(raw)), + params.query.search, + ); + return { + ...common, + connected: true, + ...page, + sessions: page.sessions.map((session) => { + const adopted = params.adoptedSessions.get(adoptedSourceKey(hostId, session.threadId)); + return adopted + ? Object.assign({}, session, { openClawSessionKey: adopted.key }) + : session; + }), + }; + }) + .catch((error: unknown) => ({ + ...common, + connected: true, + sessions: [], + error: catalogError("NODE_INVOKE_FAILED", error), + })); + if (params.onHost) { + // Keep the 8s aggregate response while allowing cold app-server discovery + // to replace that fail-soft page as soon as the node invoke really settles. + void eventualHost.then(params.onHost).catch(() => undefined); + } + try { + return await withTimeout( + eventualHost, NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS, "paired node Codex session catalog timed out", ); - const page = filterCatalogPageByTitle( - parseCatalogPage(unwrapNodeInvokePayload(raw)), - params.query.search, - ); - return { - ...common, - connected: true, - ...page, - sessions: page.sessions.map((session) => { - const adopted = params.adoptedSessions.get(adoptedSourceKey(hostId, session.threadId)); - return adopted ? Object.assign({}, session, { openClawSessionKey: adopted.key }) : session; - }), - }; } catch (error) { return { ...common, diff --git a/extensions/codex/src/session-catalog.test.ts b/extensions/codex/src/session-catalog.test.ts index f0ba04e32c9..9275f69eaf6 100644 --- a/extensions/codex/src/session-catalog.test.ts +++ b/extensions/codex/src/session-catalog.test.ts @@ -448,7 +448,7 @@ describe("Codex supervision catalog", () => { archived: false, limit: 25, modelProviders: [], - sortKey: "recency_at", + sortKey: "updated_at", sortDirection: "desc", cwd: "/workspace/one", }, @@ -1028,6 +1028,50 @@ describe("Codex supervision catalog", () => { } }); + it("publishes a paired-node page that finishes after the fail-soft response", async () => { + vi.useFakeTimers(); + try { + let resolveInvoke!: (value: unknown) => void; + const invokeResult = new Promise((resolve) => { + resolveInvoke = resolve; + }); + const invoke = vi.fn(async () => await invokeResult); + const onHost = vi.fn(); + const pending = listPairedNode({ + runtime: { nodes: { invoke } } as unknown as PluginRuntime, + node: { + nodeId: "slow-node", + displayName: "Slow node", + connected: true, + commands: [CODEX_APP_SERVER_THREADS_LIST_COMMAND], + }, + query: { limitPerHost: 40 }, + adoptedSessions: new Map(), + onHost, + }); + + await vi.advanceTimersByTimeAsync(8_000); + await expect(pending).resolves.toMatchObject({ error: { code: "NODE_INVOKE_FAILED" } }); + expect(onHost).not.toHaveBeenCalled(); + + resolveInvoke({ + payloadJSON: JSON.stringify({ + sessions: [{ threadId: "late-thread", status: "idle", archived: false }], + }), + }); + await vi.advanceTimersByTimeAsync(0); + + expect(onHost).toHaveBeenCalledWith( + expect.objectContaining({ + hostId: "node:slow-node", + sessions: [expect.objectContaining({ threadId: "late-thread" })], + }), + ); + } finally { + vi.useRealTimers(); + } + }); + it("serves one bounded transcript page from the node host command", async () => { const listTurnPage = vi.fn(async () => ({ data: [ diff --git a/extensions/codex/src/session-catalog.ts b/extensions/codex/src/session-catalog.ts index f076f586ced..54f3032f7bf 100644 --- a/extensions/codex/src/session-catalog.ts +++ b/extensions/codex/src/session-catalog.ts @@ -164,7 +164,9 @@ function createCodexSessionCatalogControlFromRequests(params: { archived: false, limit: remaining, modelProviders: [], - sortKey: "recency_at", + // Match Codex's resume picker/latest-session ordering so a session + // created outside OpenClaw enters the first catalog page immediately. + sortKey: "updated_at", sortDirection: "desc", ...(cwd ? { cwd } : {}), ...(cursor ? { cursor } : {}), @@ -390,6 +392,7 @@ async function listCodexSessionCatalog(params: { runtime: PluginRuntime; control: CodexSessionCatalogControl; query?: CodexSessionCatalogParams; + onHost?: (host: CodexSessionCatalogHost) => void; }): Promise { const query = readGatewayParams(params.query); const requestedHostIds = query.hostIds ? new Set(query.hostIds) : undefined; @@ -405,6 +408,11 @@ async function listCodexSessionCatalog(params: { }), ] : []; + for (const host of localHosts) { + if (params.onHost) { + void host.then(params.onHost).catch(() => undefined); + } + } const wantsNodes = !requestedHostIds || query.hostIds?.some((hostId) => hostId.startsWith("node:")); if (!wantsNodes) { @@ -420,18 +428,17 @@ async function listCodexSessionCatalog(params: { ) .slice(0, MAX_HOST_COUNT - localHosts.length); } catch (error) { + const registryHost: CodexSessionCatalogHost = { + hostId: "node:registry", + label: "Paired nodes", + kind: "node", + connected: false, + sessions: [], + error: catalogError("NODE_LIST_FAILED", error), + }; + params.onHost?.(registryHost); return { - hosts: [ - ...(await Promise.all(localHosts)), - { - hostId: "node:registry", - label: "Paired nodes", - kind: "node", - connected: false, - sessions: [], - error: catalogError("NODE_LIST_FAILED", error), - }, - ], + hosts: [...(await Promise.all(localHosts)), registryHost], }; } const adoptedNodeSessions = listNodeAdoptedSessionEntries({ @@ -444,6 +451,7 @@ async function listCodexSessionCatalog(params: { node, query, adoptedSessions: adoptedNodeSessions, + ...(params.onHost ? { onHost: params.onHost } : {}), }); return Object.assign(host, codexNodeTerminalCapability(node)); }); @@ -1239,15 +1247,19 @@ function registerCodexSessionCatalog(params: { label: "Codex", list: async (query) => { const localTerminalAvailable = resolveLocalCodexTerminalExecutable() !== undefined; + const { onHost, ...gatewayQuery } = query; + const mapHost = (host: CodexSessionCatalogHost) => + toGenericCatalogHost(host, localTerminalAvailable); return ( await listCodexSessionCatalog({ bindingStore: params.bindingStore, config: params.getRuntimeConfig(), runtime: params.api.runtime, control: params.control, - query, + query: gatewayQuery, + ...(onHost ? { onHost: (host) => onHost(mapHost(host)) } : {}), }) - ).hosts.map((host) => toGenericCatalogHost(host, localTerminalAvailable)); + ).hosts.map(mapHost); }, read: async (request) => { const page = await readCodexSessionTranscript({ diff --git a/packages/gateway-protocol/src/index.ts b/packages/gateway-protocol/src/index.ts index 1328cedd056..5a93b1322ae 100644 --- a/packages/gateway-protocol/src/index.ts +++ b/packages/gateway-protocol/src/index.ts @@ -413,6 +413,7 @@ import { SessionsCatalogContinueResultSchema, SessionsCatalogListParamsSchema, SessionsCatalogListResultSchema, + SessionsCatalogHostEventSchema, SessionsCatalogReadParamsSchema, SessionsCatalogReadResultSchema, SessionsMessagesSubscribeParamsSchema, @@ -671,6 +672,7 @@ export const validateSecretsResolveParams = lazyCompile(SecretsResolveParamsSche export const validateSecretsResolveResult = lazyCompile(SecretsResolveResultSchema); export const validateSessionsListParams = lazyCompile(SessionsListParamsSchema); export const validateSessionsCatalogListParams = lazyCompile(SessionsCatalogListParamsSchema); +export const validateSessionsCatalogHostEvent = lazyCompile(SessionsCatalogHostEventSchema); export const validateSessionsCatalogReadParams = lazyCompile(SessionsCatalogReadParamsSchema); export const validateSessionsCatalogContinueParams = lazyCompile( SessionsCatalogContinueParamsSchema, @@ -1033,6 +1035,7 @@ export { SessionCatalogTranscriptItemSchema, SessionsCatalogListParamsSchema, SessionsCatalogListResultSchema, + SessionsCatalogHostEventSchema, SessionsCatalogReadParamsSchema, SessionsCatalogReadResultSchema, SessionsCatalogContinueParamsSchema, diff --git a/packages/gateway-protocol/src/public-session-catalog.ts b/packages/gateway-protocol/src/public-session-catalog.ts index 9592b45b3b0..cacea919138 100644 --- a/packages/gateway-protocol/src/public-session-catalog.ts +++ b/packages/gateway-protocol/src/public-session-catalog.ts @@ -12,6 +12,7 @@ export type { SessionsCatalogContinueResult, SessionsCatalogListParams, SessionsCatalogListResult, + SessionsCatalogHostEvent, SessionsCatalogReadParams, SessionsCatalogReadResult, } from "./schema/sessions-catalog.js"; diff --git a/packages/gateway-protocol/src/schema/sessions-catalog.test.ts b/packages/gateway-protocol/src/schema/sessions-catalog.test.ts index 112bb13a4ee..20767ebef19 100644 --- a/packages/gateway-protocol/src/schema/sessions-catalog.test.ts +++ b/packages/gateway-protocol/src/schema/sessions-catalog.test.ts @@ -1,6 +1,7 @@ import { Value } from "typebox/value"; import { describe, expect, it } from "vitest"; import { + SessionsCatalogHostEventSchema, SessionsCatalogListParamsSchema, SessionsCatalogListResultSchema, } from "./sessions-catalog.js"; @@ -45,6 +46,15 @@ describe("SessionsCatalogListResultSchema", () => { }); describe("SessionsCatalogListParamsSchema", () => { + it("accepts an optional progressive stream id without a catalog selector", () => { + expect( + Value.Check(SessionsCatalogListParamsSchema, { + agentId: "main", + progressId: "progress-1", + }), + ).toBe(true); + }); + it("accepts an optional agent scope", () => { expect( Value.Check(SessionsCatalogListParamsSchema, { @@ -66,3 +76,41 @@ describe("SessionsCatalogListParamsSchema", () => { ).toBe(true); }); }); + +describe("SessionsCatalogHostEventSchema", () => { + it("accepts one completed host and rejects unknown fields", () => { + const event = { + progressId: "progress-1", + agentId: "main", + catalog: { + id: "codex", + label: "Codex", + capabilities: { continueSession: true, archive: true }, + hosts: [ + { + hostId: "gateway:local", + label: "Local Codex", + kind: "gateway", + connected: true, + sessions: [], + }, + ], + }, + }; + + expect(Value.Check(SessionsCatalogHostEventSchema, event)).toBe(true); + expect(Value.Check(SessionsCatalogHostEventSchema, { ...event, unexpected: true })).toBe(false); + expect( + Value.Check(SessionsCatalogHostEventSchema, { + ...event, + catalog: { ...event.catalog, hosts: [] }, + }), + ).toBe(false); + expect( + Value.Check(SessionsCatalogHostEventSchema, { + ...event, + catalog: { ...event.catalog, hosts: [event.catalog.hosts[0], event.catalog.hosts[0]] }, + }), + ).toBe(false); + }); +}); diff --git a/packages/gateway-protocol/src/schema/sessions-catalog.ts b/packages/gateway-protocol/src/schema/sessions-catalog.ts index 2cf4c29893c..33ab9bf266c 100644 --- a/packages/gateway-protocol/src/schema/sessions-catalog.ts +++ b/packages/gateway-protocol/src/schema/sessions-catalog.ts @@ -59,6 +59,7 @@ export const SessionCatalogSchema = closedObject({ const SessionsCatalogListCommonProperties = { agentId: Type.Optional(NonEmptyString), + progressId: Type.Optional(Type.String({ minLength: 1, maxLength: 128 })), search: Type.Optional(Type.String()), limitPerHost: Type.Optional(Type.Integer({ minimum: 1 })), hostIds: Type.Optional(Type.Array(NonEmptyString)), @@ -80,6 +81,17 @@ export const SessionsCatalogListResultSchema = closedObject({ catalogs: Type.Array(SessionCatalogSchema), }); +const SessionsCatalogHostEventCatalogSchema = closedObject({ + ...SessionCatalogSchema.properties, + hosts: Type.Array(SessionCatalogHostSchema, { minItems: 1, maxItems: 1 }), +}); + +export const SessionsCatalogHostEventSchema = closedObject({ + progressId: Type.String({ minLength: 1, maxLength: 128 }), + agentId: NonEmptyString, + catalog: SessionsCatalogHostEventCatalogSchema, +}); + export const SessionCatalogTranscriptItemSchema = closedObject({ id: Type.Optional(Type.String()), type: Type.Union([ @@ -137,6 +149,7 @@ export type SessionCatalogHost = Static; export type SessionCatalog = Static; export type SessionsCatalogListParams = Static; export type SessionsCatalogListResult = Static; +export type SessionsCatalogHostEvent = Static; export type SessionCatalogTranscriptItem = Static; export type SessionsCatalogReadParams = Static; export type SessionsCatalogReadResult = Static; diff --git a/src/gateway/gateway-misc.test.ts b/src/gateway/gateway-misc.test.ts index 812b63cf305..d12239a6c41 100644 --- a/src/gateway/gateway-misc.test.ts +++ b/src/gateway/gateway-misc.test.ts @@ -484,6 +484,24 @@ describe("gateway broadcaster", () => { expect(readSocket.send).not.toHaveBeenCalled(); }); + it("requires operator.read for progressive session catalog events", () => { + const { pairingSocket, nodeSocket, readSocket, writeSocket, adminSocket, broadcastToConnIds } = + makeScopedBroadcastContext(); + const targets = new Set(["c-pairing", "c-node", "c-read", "c-write", "c-admin"]); + + broadcastToConnIds( + "sessions.catalog.host", + { progressId: "progress-1", agentId: "main", catalog: { id: "codex", hosts: [] } }, + targets, + ); + + expect(pairingSocket.send).not.toHaveBeenCalled(); + expect(nodeSocket.send).not.toHaveBeenCalled(); + expectSentEvents(readSocket, ["sessions.catalog.host"]); + expectSentEvents(writeSocket, ["sessions.catalog.host"]); + expectSentEvents(adminSocket, ["sessions.catalog.host"]); + }); + it("requires operator.read for task ledger broadcast events", () => { const { pairingSocket, nodeSocket, readSocket, writeSocket, adminSocket, broadcast } = makeScopedBroadcastContext(); diff --git a/src/gateway/server-broadcast.ts b/src/gateway/server-broadcast.ts index 6c916f1f327..3272d3a2c2a 100644 --- a/src/gateway/server-broadcast.ts +++ b/src/gateway/server-broadcast.ts @@ -56,6 +56,7 @@ const EVENT_SCOPE_GUARDS: Record = { "node.pair.requested": [PAIRING_SCOPE], "node.pair.resolved": [PAIRING_SCOPE], "node.presence": [READ_SCOPE], + "sessions.catalog.host": [READ_SCOPE], "sessions.changed": [READ_SCOPE], "session.approval": [APPROVALS_SCOPE], "session.message": [READ_SCOPE], diff --git a/src/gateway/server-methods/session-catalog.test.ts b/src/gateway/server-methods/session-catalog.test.ts index 34b3c62a95e..1b5e5970bca 100644 --- a/src/gateway/server-methods/session-catalog.test.ts +++ b/src/gateway/server-methods/session-catalog.test.ts @@ -53,14 +53,15 @@ async function call( method: keyof typeof sessionCatalogHandlers, params: unknown, config: Record = {}, - client?: { connect?: { scopes?: string[] } }, + client?: { connect?: { scopes?: string[] }; connId?: string }, + contextOverrides: Record = {}, ) { const respond = vi.fn(); await sessionCatalogHandlers[method]?.({ params, respond, client, - context: { getRuntimeConfig: () => config }, + context: { getRuntimeConfig: () => config, ...contextOverrides }, } as never); return respond; } @@ -98,6 +99,50 @@ describe("session catalog Gateway methods", () => { }); }); + it("streams completed hosts to only the requesting connection", async () => { + const broadcastToConnIds = vi.fn(); + const host = { + hostId: "node:fast", + label: "Fast node", + kind: "node" as const, + connected: true, + nodeId: "fast", + sessions: [], + }; + hoisted.activeRegistry.sessionCatalogs = [ + { + provider: provider("codex", { + list: vi.fn(async ({ onHost }) => { + onHost?.(host); + return [host]; + }), + }), + }, + ]; + + const respond = await call( + "sessions.catalog.list", + { progressId: "progress-1" }, + {}, + { connId: "requester", connect: {} }, + { broadcastToConnIds }, + ); + + expect(broadcastToConnIds).toHaveBeenCalledWith( + "sessions.catalog.host", + { + progressId: "progress-1", + agentId: "main", + catalog: expect.objectContaining({ id: "codex", hosts: [host] }), + }, + new Set(["requester"]), + { dropIfSlow: true }, + ); + expect(respond).toHaveBeenCalledWith(true, { + catalogs: [expect.objectContaining({ id: "codex", hosts: [host] })], + }); + }); + it("uses the pinned Gateway catalog runtime after active registry churn", async () => { const previousNodesRuntime = gatewaySubagentState.nodes; const listNodes = vi.fn(async () => ({ nodes: [] })); diff --git a/src/gateway/server-methods/session-catalog.ts b/src/gateway/server-methods/session-catalog.ts index 18f03e8dab4..49adca1ac58 100644 --- a/src/gateway/server-methods/session-catalog.ts +++ b/src/gateway/server-methods/session-catalog.ts @@ -156,7 +156,7 @@ function catalogResult( } export const sessionCatalogHandlers: GatewayRequestHandlers = { - "sessions.catalog.list": async ({ params, respond, context }) => { + "sessions.catalog.list": async ({ params, respond, context, client }) => { if ( !assertValidParams( params, @@ -189,16 +189,35 @@ export const sessionCatalogHandlers: GatewayRequestHandlers = { return; } const search = normalizeSessionCatalogSearch(request.search); + const progressId = request.progressId; + const progressConnId = progressId && client?.connId ? client.connId : undefined; const catalogList = await Promise.all( selected.map(async (provider): Promise => { const createTarget = resolveProviderCreateTarget(provider, resolvedAgent.agentId); const createSession = createTarget.ok ? { model: createTarget.target.model } : undefined; + const onHost = progressConnId + ? (host: SessionCatalog["hosts"][number]) => { + // Progressive frames are an optimization. The final RPC response remains + // authoritative when a slow client drops an intermediate host update. + context.broadcastToConnIds( + "sessions.catalog.host", + { + progressId, + agentId: resolvedAgent.agentId, + catalog: catalogResult(provider, [host], undefined, createSession), + }, + new Set([progressConnId]), + { dropIfSlow: true }, + ); + } + : undefined; try { const hosts = await provider.list({ search, limitPerHost: request.limitPerHost, hostIds: request.hostIds, ...("cursors" in request ? { cursors: request.cursors } : {}), + ...(onHost ? { onHost } : {}), }); return catalogResult(provider, hosts, undefined, createSession); } catch (error) { diff --git a/src/plugins/session-catalog.ts b/src/plugins/session-catalog.ts index f6a7a581507..4a2475853aa 100644 --- a/src/plugins/session-catalog.ts +++ b/src/plugins/session-catalog.ts @@ -12,6 +12,8 @@ export type SessionCatalogListProviderParams = { limitPerHost?: number; hostIds?: string[]; cursors?: Record; + /** Publishes completed hosts without waiting for slower machines in the same list. */ + onHost?: (host: SessionCatalogHost) => void; }; export type SessionCatalogReadProviderParams = Omit; export type SessionCatalogContinueProviderParams = Omit< diff --git a/ui/src/components/app-sidebar-session-catalog-live.ts b/ui/src/components/app-sidebar-session-catalog-live.ts new file mode 100644 index 00000000000..820d4679602 --- /dev/null +++ b/ui/src/components/app-sidebar-session-catalog-live.ts @@ -0,0 +1,494 @@ +import type { + SessionCatalog, + SessionsCatalogHostEvent, + SessionsCatalogListResult, +} from "../../../packages/gateway-protocol/src/index.ts"; +import { GatewayRequestError, type GatewayBrowserClient } from "../api/gateway.ts"; +import type { ApplicationGatewaySnapshot } from "../app/gateway.ts"; +import { isGatewayMethodAdvertised } from "../lib/gateway-methods.ts"; +import { normalizeAgentId } from "../lib/sessions/session-key.ts"; +import { generateUUID } from "../lib/uuid.ts"; +import { + preserveExpandedCatalogHost, + refetchExpandedSessionCatalogPages, +} from "./app-sidebar-session-catalog-state.ts"; +import { sessionCatalogHostKey } from "./app-sidebar-session-types.ts"; + +export const SESSION_CATALOG_CHANGED_REFRESH_MS = 5_000; +const SESSION_CATALOG_STABLE_REFRESH_MS = 30_000; + +function sessionCatalogSnapshot(catalogs: readonly SessionCatalog[]): string { + return JSON.stringify(catalogs); +} + +export function sessionCatalogListClient( + snapshot: ApplicationGatewaySnapshot | undefined, + connected: boolean, +): GatewayBrowserClient | null { + if ( + !connected || + !snapshot?.connected || + !snapshot.client || + isGatewayMethodAdvertised(snapshot, "sessions.catalog.list") !== true + ) { + return null; + } + return snapshot.client; +} + +async function requestSessionCatalogList(params: { + client: GatewayBrowserClient; + agentId: string; + progressId: string; + progressive: boolean; +}): Promise<{ result: SessionsCatalogListResult; progressive: boolean }> { + const baseParams = { agentId: params.agentId, limitPerHost: 40 }; + if (!params.progressive) { + return { + result: await params.client.request("sessions.catalog.list", baseParams), + progressive: false, + }; + } + try { + return { + result: await params.client.request("sessions.catalog.list", { + ...baseParams, + progressId: params.progressId, + }), + progressive: true, + }; + } catch (error) { + if (!(error instanceof GatewayRequestError) || error.gatewayCode !== "INVALID_REQUEST") { + throw error; + } + // Older Gateways advertise the list method but reject the additive field. + // Retry once without streaming, then keep that connection on final pages. + return { + result: await params.client.request("sessions.catalog.list", baseParams), + progressive: false, + }; + } +} + +function isSessionsCatalogHostEvent(value: unknown): value is SessionsCatalogHostEvent { + if (!value || typeof value !== "object") { + return false; + } + const event = value as Record; + const catalog = event.catalog; + if (!catalog || typeof catalog !== "object") { + return false; + } + const catalogRecord = catalog as Record; + const hosts = Array.isArray(catalogRecord.hosts) ? catalogRecord.hosts : []; + const host = hosts[0]; + return ( + typeof event.progressId === "string" && + event.progressId.length > 0 && + typeof event.agentId === "string" && + event.agentId.length > 0 && + typeof catalogRecord.id === "string" && + typeof catalogRecord.label === "string" && + catalogRecord.capabilities !== null && + typeof catalogRecord.capabilities === "object" && + hosts.length === 1 && + host !== null && + typeof host === "object" && + typeof (host as Record).hostId === "string" && + Array.isArray((host as Record).sessions) + ); +} + +/** Tracks one sidebar's progressive list streams and adaptive refresh lifecycle. */ +export class SessionCatalogLiveState { + timer: ReturnType | null = null; + requestGeneration: number | null = null; + sawChange = false; + refreshPending = false; + progressive = true; + + private activationTimer: ReturnType | null = null; + private connectionEpoch = 0; + private refetchOwner: symbol | null = null; + private presenceSignature: string | null = null; + private progressSequence = 0; + private readonly progressSequences = new Map(); + private readonly hostProgressSequences = new Map(); + private readonly hostIdsByCatalog = new Map>(); + private readonly requestChangedHostKeys = new Set(); + private requestOwner: symbol | null = null; + + cancelTimer() { + if (this.timer !== null) { + globalThis.clearTimeout(this.timer); + this.timer = null; + } + } + + clear() { + this.cancelScheduledRefreshes(); + this.requestGeneration = null; + this.requestOwner = null; + this.progressSequences.clear(); + this.hostProgressSequences.clear(); + this.hostIdsByCatalog.clear(); + this.requestChangedHostKeys.clear(); + this.presenceSignature = null; + this.sawChange = false; + this.refreshPending = false; + this.refetchOwner = null; + } + + resetConnection() { + this.clear(); + this.connectionEpoch += 1; + this.progressive = true; + } + + async requestList( + client: GatewayBrowserClient, + agentId: string, + progressId: string, + ): Promise { + const connectionEpoch = this.connectionEpoch; + const response = await requestSessionCatalogList({ + client, + agentId, + progressId, + progressive: this.progressive, + }); + if (connectionEpoch === this.connectionEpoch) { + this.progressive = response.progressive; + } + return response.result; + } + + mergeFinal(catalogs: SessionCatalog[], currentCatalogs: readonly SessionCatalog[]) { + const current = new Map(currentCatalogs.map((catalog) => [catalog.id, catalog])); + return catalogs.map((catalog) => { + const currentHosts = new Map( + current.get(catalog.id)?.hosts.map((host) => [host.hostId, host]) ?? [], + ); + return { + ...catalog, + hosts: catalog.hosts.map((host) => { + const progressiveHost = currentHosts.get(host.hostId); + const changed = this.requestChangedHostKeys.has( + sessionCatalogHostKey(catalog.id, host.hostId), + ); + return host.error && changed && progressiveHost && !progressiveHost.error + ? progressiveHost + : host; + }), + }; + }); + } + + get refetching() { + return this.refetchOwner !== null; + } + + beginRefetch(active: boolean): symbol | null { + if (!active) { + return null; + } + const owner = Symbol("session-catalog-refetch"); + this.refetchOwner = owner; + return owner; + } + + endRefetch(owner: symbol | null) { + if (owner !== null && this.refetchOwner === owner) { + this.refetchOwner = null; + } + } + + beginRequest(generation: number): { + progressId: string; + progressSequence: number; + requestOwner: symbol; + } { + this.requestGeneration = generation; + const requestOwner = Symbol("session-catalog-request"); + this.requestOwner = requestOwner; + this.sawChange = false; + this.requestChangedHostKeys.clear(); + const progressId = generateUUID(); + const progressSequence = ++this.progressSequence; + this.progressSequences.set(progressId, progressSequence); + if (this.progressSequences.size > 8) { + const oldest = this.progressSequences.keys().next().value; + if (oldest) { + this.progressSequences.delete(oldest); + } + } + return { progressId, progressSequence, requestOwner }; + } + + ownsRequest(owner: symbol) { + return this.requestOwner === owner; + } + + markFinal(params: { + catalogs: readonly SessionCatalog[]; + hadCatalogs: boolean; + previousSnapshot: string; + progressSequence: number; + }) { + this.sawChange = + params.hadCatalogs && params.previousSnapshot !== sessionCatalogSnapshot(params.catalogs); + const finalCatalogIds = new Set(params.catalogs.map((catalog) => catalog.id)); + for (const [catalogId, previousHostIds] of this.hostIdsByCatalog) { + if (finalCatalogIds.has(catalogId)) { + continue; + } + for (const hostId of previousHostIds) { + this.hostProgressSequences.set( + sessionCatalogHostKey(catalogId, hostId), + params.progressSequence, + ); + } + this.hostIdsByCatalog.delete(catalogId); + } + for (const catalog of params.catalogs) { + const previousHostIds = this.hostIdsByCatalog.get(catalog.id) ?? new Set(); + const finalHostIds = new Set(catalog.hosts.map((host) => host.hostId)); + for (const hostId of previousHostIds) { + if (!finalHostIds.has(hostId)) { + this.hostProgressSequences.set( + sessionCatalogHostKey(catalog.id, hostId), + params.progressSequence, + ); + } + } + for (const host of catalog.hosts) { + if (host.error) { + continue; + } + const key = sessionCatalogHostKey(catalog.id, host.hostId); + this.hostProgressSequences.set( + key, + Math.max(this.hostProgressSequences.get(key) ?? -1, params.progressSequence), + ); + } + this.hostIdsByCatalog.set(catalog.id, finalHostIds); + } + } + + observePresence(payload: unknown): boolean { + const presence = + payload && typeof payload === "object" + ? (payload as { presence?: unknown }).presence + : undefined; + if (!Array.isArray(presence)) { + return false; + } + const states = new Map(); + for (const entry of presence) { + if (!entry || typeof entry !== "object") { + continue; + } + const record = entry as Record; + const rawId = typeof record.deviceId === "string" ? record.deviceId : record.instanceId; + const id = typeof rawId === "string" ? rawId.trim().toLowerCase() : ""; + const mode = typeof record.mode === "string" ? record.mode.trim().toLowerCase() : ""; + if (!id || mode === "gateway") { + continue; + } + const reason = typeof record.reason === "string" ? record.reason.trim().toLowerCase() : ""; + states.set(id, reason === "disconnect" ? "offline" : "connected"); + } + const signature = JSON.stringify( + [...states].toSorted(([left], [right]) => left.localeCompare(right)), + ); + const previous = this.presenceSignature; + this.presenceSignature = signature; + return previous === null ? states.size > 0 : previous !== signature; + } + + applyHost(params: { + payload: unknown; + agentId: string; + catalogs: SessionCatalog[]; + pageDepths: ReadonlyMap; + }): { catalogs: SessionCatalog[]; catalogId: string } | null { + if (!isSessionsCatalogHostEvent(params.payload)) { + return null; + } + const event = params.payload; + if (normalizeAgentId(event.agentId) !== normalizeAgentId(params.agentId)) { + return null; + } + const sequence = this.progressSequences.get(event.progressId); + const freshHost = event.catalog.hosts[0]; + if (sequence === undefined || !freshHost) { + return null; + } + const hostKey = sessionCatalogHostKey(event.catalog.id, freshHost.hostId); + if (sequence < (this.hostProgressSequences.get(hostKey) ?? -1)) { + return null; + } + this.hostProgressSequences.set(hostKey, sequence); + if (this.requestGeneration !== null) { + this.requestChangedHostKeys.add(hostKey); + } + const currentCatalog = params.catalogs.find((catalog) => catalog.id === event.catalog.id); + let catalogs: SessionCatalog[]; + if (!currentCatalog) { + catalogs = [...params.catalogs, { ...event.catalog, hosts: [freshHost] }].toSorted( + (left, right) => left.id.localeCompare(right.id), + ); + } else { + const currentHost = currentCatalog.hosts.find((host) => host.hostId === freshHost.hostId); + const mergedHost = + (params.pageDepths.get(hostKey) ?? 0) > 0 + ? preserveExpandedCatalogHost(freshHost, currentHost) + : freshHost; + const hosts = currentHost + ? currentCatalog.hosts.map((host) => (host.hostId === freshHost.hostId ? mergedHost : host)) + : [...currentCatalog.hosts, mergedHost]; + catalogs = params.catalogs.map((catalog) => + catalog.id === event.catalog.id + ? { + ...event.catalog, + hosts: hosts.toSorted((left, right) => left.label.localeCompare(right.label)), + } + : catalog, + ); + } + if (sessionCatalogSnapshot(catalogs) === sessionCatalogSnapshot(params.catalogs)) { + return null; + } + this.sawChange = true; + return { catalogs, catalogId: event.catalog.id }; + } + + schedule(delayMs: number, isConnected: boolean, refresh: () => void) { + if (document.visibilityState === "hidden" || !isConnected) { + return; + } + this.cancelTimer(); + this.timer = globalThis.setTimeout(() => { + this.timer = null; + refresh(); + }, delayMs); + } + + requestRefresh(params: { + visible: boolean; + connected: boolean; + generation: number; + refresh: () => void; + }) { + if (!params.visible || !params.connected) { + return; + } + this.cancelTimer(); + if (this.requestGeneration === params.generation) { + this.refreshPending = true; + return; + } + params.refresh(); + } + + cancelActivation() { + if (this.activationTimer !== null) { + globalThis.clearTimeout(this.activationTimer); + this.activationTimer = null; + } + } + + cancelScheduledRefreshes() { + this.cancelTimer(); + this.cancelActivation(); + } + + scheduleActivation(refresh: () => void) { + if (this.activationTimer !== null) { + return; + } + // Browsers fire visibilitychange and focus as one foregrounding pair. + // The short window prevents that pair from triggering two fleet scans. + this.activationTimer = globalThis.setTimeout(() => { + this.activationTimer = null; + refresh(); + }, 50); + } +} + +/** Runs one progressive list/reconciliation cycle without binding it to a Lit element. */ +export async function refreshSessionCatalogsLive(params: { + live: SessionCatalogLiveState; + client: GatewayBrowserClient; + agentId: string; + generation: number; + revision: number; + currentGeneration: () => number; + currentRevision: () => number; + currentClient: () => GatewayBrowserClient | null; + catalogs: () => SessionCatalog[]; + pageDepths: ReadonlyMap; + connected: () => boolean; + applyFinal: (catalogs: SessionCatalog[], revisedCatalogIds: ReadonlySet) => void; + refresh: () => void; +}) { + const { live, client, generation, revision } = params; + if (live.requestGeneration === generation) { + return; + } + const { progressId, progressSequence, requestOwner } = live.beginRequest(generation); + const hadCatalogs = params.catalogs().length > 0; + const previousSnapshot = sessionCatalogSnapshot(params.catalogs()); + let refetchOwner: symbol | null = null; + const requestIsCurrent = () => + live.ownsRequest(requestOwner) && + generation === params.currentGeneration() && + client === params.currentClient(); + const revisionIsCurrent = () => requestIsCurrent() && revision === params.currentRevision(); + try { + const result = await live.requestList(client, params.agentId, progressId); + if (!requestIsCurrent()) { + return; + } + refetchOwner = live.beginRefetch(params.pageDepths.size > 0); + const previousCatalogs = params.catalogs(); + const catalogs = await refetchExpandedSessionCatalogPages({ + catalogs: live.mergeFinal(result.catalogs, previousCatalogs), + previousCatalogs, + client, + agentId: params.agentId, + pageDepths: params.pageDepths, + isCurrent: revisionIsCurrent, + }); + if (!revisionIsCurrent()) { + return; + } + const revisedCatalogIds = new Set([ + ...params.catalogs().map((catalog) => catalog.id), + ...catalogs.map((catalog) => catalog.id), + ]); + params.applyFinal(catalogs, revisedCatalogIds); + live.markFinal({ catalogs, hadCatalogs, previousSnapshot, progressSequence }); + } catch { + // A transient poll failure must not collapse already visible or expanded pages. + } finally { + live.endRefetch(refetchOwner); + const ownsRequest = live.ownsRequest(requestOwner); + if (ownsRequest) { + live.requestGeneration = null; + } + if (ownsRequest && requestIsCurrent() && params.connected()) { + const pending = live.refreshPending; + live.refreshPending = false; + live.schedule( + pending + ? 0 + : live.sawChange + ? SESSION_CATALOG_CHANGED_REFRESH_MS + : SESSION_CATALOG_STABLE_REFRESH_MS, + params.connected(), + params.refresh, + ); + } + } +} diff --git a/ui/src/components/app-sidebar-session-catalog-state.ts b/ui/src/components/app-sidebar-session-catalog-state.ts index 87df3f2ec67..99d7998624d 100644 --- a/ui/src/components/app-sidebar-session-catalog-state.ts +++ b/ui/src/components/app-sidebar-session-catalog-state.ts @@ -2,8 +2,10 @@ import type { SessionCatalog, SessionCatalogHost, SessionCatalogSession, + SessionsCatalogListResult, } from "../../../packages/gateway-protocol/src/index.ts"; -import { GatewayRequestError } from "../api/gateway.ts"; +import { GatewayRequestError, type GatewayBrowserClient } from "../api/gateway.ts"; +import { sessionCatalogHostKey } from "./app-sidebar-session-types.ts"; type SessionCatalogError = NonNullable; @@ -14,7 +16,7 @@ export function sessionCatalogRequestError(error: unknown): SessionCatalogError }; } -export function mergeCatalogSessionRows( +function mergeCatalogSessionRows( first: readonly SessionCatalogSession[], second: readonly SessionCatalogSession[], ): SessionCatalogSession[] { @@ -77,3 +79,68 @@ export function mergeSessionCatalogPage(params: { advancedHostIds, }; } + +export async function refetchExpandedSessionCatalogPages(params: { + catalogs: SessionCatalog[]; + previousCatalogs: readonly SessionCatalog[]; + client: GatewayBrowserClient; + agentId: string; + pageDepths: ReadonlyMap; + isCurrent: () => boolean; +}): Promise { + const previousCatalogs = new Map(params.previousCatalogs.map((catalog) => [catalog.id, catalog])); + return Promise.all( + params.catalogs.map(async (catalog) => { + const previousHosts = new Map( + previousCatalogs.get(catalog.id)?.hosts.map((host) => [host.hostId, host]) ?? [], + ); + const hosts = await Promise.all( + catalog.hosts.map(async (host) => { + const pageDepth = + params.pageDepths.get(sessionCatalogHostKey(catalog.id, host.hostId)) ?? 0; + if (pageDepth === 0) { + return host; + } + const previous = previousHosts.get(host.hostId); + if (host.error) { + return preserveExpandedCatalogHost(host, previous); + } + let sessions = host.sessions; + let nextCursor = host.nextCursor; + for (let loadedPages = 0; loadedPages < pageDepth && nextCursor; loadedPages += 1) { + let result: SessionsCatalogListResult; + try { + result = await params.client.request( + "sessions.catalog.list", + { + agentId: params.agentId, + catalogId: catalog.id, + cursors: { [host.hostId]: nextCursor }, + }, + ); + } catch { + return previous ?? host; + } + if (!params.isCurrent()) { + return previous ?? host; + } + const pageHost = result.catalogs + .find((candidate) => candidate.id === catalog.id) + ?.hosts.find((candidate) => candidate.hostId === host.hostId); + if (!pageHost) { + return previous ?? host; + } + if (pageHost.error) { + return preserveExpandedCatalogHost({ ...host, ...pageHost }, previous ?? host); + } + sessions = mergeCatalogSessionRows(sessions, pageHost.sessions); + nextCursor = pageHost.nextCursor; + } + const { nextCursor: _cursor, sessions: _sessions, ...freshHost } = host; + return { ...freshHost, sessions, ...(nextCursor ? { nextCursor } : {}) }; + }), + ); + return { ...catalog, hosts }; + }), + ); +} diff --git a/ui/src/components/app-sidebar-session-data.ts b/ui/src/components/app-sidebar-session-data.ts index dd7c26eb0b3..05a81b7a79d 100644 --- a/ui/src/components/app-sidebar-session-data.ts +++ b/ui/src/components/app-sidebar-session-data.ts @@ -7,7 +7,6 @@ import type { GatewayBrowserClient } from "../api/gateway.ts"; import type { GatewaySessionRow, SessionsListResult } from "../api/types.ts"; import type { RouteId } from "../app-route-paths.ts"; import type { ApplicationContext } from "../app/context.ts"; -import { isGatewayMethodAdvertised } from "../lib/gateway-methods.ts"; import { CATALOG_SESSION_CONTINUED_EVENT, type CatalogSessionContinuedDetail, @@ -23,9 +22,13 @@ import { mergeChildSessionRows, } from "./app-sidebar-child-session-data.ts"; import { - mergeCatalogSessionRows, + refreshSessionCatalogsLive, + SESSION_CATALOG_CHANGED_REFRESH_MS, + SessionCatalogLiveState, + sessionCatalogListClient, +} from "./app-sidebar-session-catalog-live.ts"; +import { mergeSessionCatalogPage, - preserveExpandedCatalogHost, sessionCatalogRequestError, } from "./app-sidebar-session-catalog-state.ts"; import { bindAdoptedCatalogSession } from "./app-sidebar-session-catalogs.ts"; @@ -74,11 +77,10 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { private sessionMutationEpoch = 0; private sessionsScrollElement: HTMLElement | null = null; private sessionsScrollResizeObserver: ResizeObserver | null = null; - private sessionCatalogTimer: ReturnType | null = null; + private readonly sessionCatalogLive = new SessionCatalogLiveState(); private sessionCatalogAgentId: string | null = null; private sessionCatalogGeneration = 0; private sessionCatalogRevision = 0; - private sessionCatalogRequestGeneration: number | null = null; private readonly sessionCatalogPageDepths = new Map(); private readonly sessionCatalogRevisions = new Map(); @@ -104,6 +106,22 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { () => this.context?.sessions, (sessions) => sessions.subscribeCreated((key) => this.promoteCreatedSession(key)), ) + .effect( + () => this.context?.gateway, + (gateway) => + gateway.subscribeEvents((event) => { + if (event.event === "sessions.catalog.host") { + this.applySessionCatalogHostEvent(event.payload); + return; + } + if ( + event.event === "presence" && + this.sessionCatalogLive.observePresence(event.payload) + ) { + this.requestSessionCatalogRefresh(); + } + }), + ) .watch( () => this.context?.agents, (agents, notify) => agents.subscribe(notify), @@ -122,6 +140,8 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { CATALOG_SESSION_CONTINUED_EVENT, this.handleCatalogSessionContinued as EventListener, ); + document.addEventListener("visibilitychange", this.handleSessionCatalogPageActivation); + globalThis.addEventListener("focus", this.handleSessionCatalogPageActivation); } override disconnectedCallback() { @@ -129,19 +149,18 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { CATALOG_SESSION_CONTINUED_EVENT, this.handleCatalogSessionContinued as EventListener, ); + document.removeEventListener("visibilitychange", this.handleSessionCatalogPageActivation); + globalThis.removeEventListener("focus", this.handleSessionCatalogPageActivation); this.dismissTransientMenus(); this.invalidateSessionMutations(); this.gatewaySource = null; this.gatewayClient = null; this.gatewayConnected = false; this.sessionCatalogGeneration += 1; + this.sessionCatalogLive.clear(); this.sessionsScrollResizeObserver?.disconnect(); this.sessionsScrollResizeObserver = null; this.sessionsScrollElement = null; - if (this.sessionCatalogTimer) { - globalThis.clearTimeout(this.sessionCatalogTimer); - this.sessionCatalogTimer = null; - } if (this.activeSessionLineageRetryTimer) { globalThis.clearTimeout(this.activeSessionLineageRetryTimer); this.activeSessionLineageRetryTimer = null; @@ -156,11 +175,9 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { this.synchronizeSessionCatalogAgent(this.expandedAgentId()); } if ( - !snapshot?.connected || - !snapshot.client || - isGatewayMethodAdvertised(snapshot, "sessions.catalog.list") !== true || - this.sessionCatalogTimer || - this.sessionCatalogRequestGeneration === this.sessionCatalogGeneration + !sessionCatalogListClient(snapshot, this.connected) || + this.sessionCatalogLive.timer || + this.sessionCatalogLive.requestGeneration === this.sessionCatalogGeneration ) { return; } @@ -174,11 +191,8 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { this.sessionCatalogAgentId = agentId; this.sessionCatalogGeneration += 1; this.sessionCatalogRevision += 1; + this.sessionCatalogLive.clear(); this.loadingMoreSessionCatalogIds = new Set(); - if (this.sessionCatalogTimer) { - globalThis.clearTimeout(this.sessionCatalogTimer); - this.sessionCatalogTimer = null; - } if (this.sessionCatalogs.some((catalog) => catalog.capabilities.createSession)) { this.sessionCatalogs = this.sessionCatalogs.map((catalog) => { const { createSession: _createSession, ...capabilities } = catalog.capabilities; @@ -204,143 +218,79 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { ); }; + private readonly handleSessionCatalogPageActivation = () => { + if (document.visibilityState === "hidden") { + this.sessionCatalogLive.cancelScheduledRefreshes(); + return; + } + this.sessionCatalogLive.scheduleActivation(() => this.requestSessionCatalogRefresh()); + }; + + private requestSessionCatalogRefresh() { + const snapshot = this.context?.gateway.snapshot; + this.sessionCatalogLive.requestRefresh({ + visible: document.visibilityState !== "hidden", + connected: this.isConnected && Boolean(sessionCatalogListClient(snapshot, this.connected)), + generation: this.sessionCatalogGeneration, + refresh: () => void this.refreshSessionCatalogs(), + }); + } + + private applySessionCatalogHostEvent(payload: unknown) { + const update = this.sessionCatalogLive.applyHost({ + payload, + agentId: this.sessionCatalogAgentId ?? "", + catalogs: this.sessionCatalogs, + pageDepths: this.sessionCatalogPageDepths, + }); + if (!update) { + return; + } + this.sessionCatalogs = update.catalogs; + this.sessionCatalogRevision += this.sessionCatalogLive.refetching ? 1 : 0; + const catalogRevision = this.sessionCatalogRevisions.get(update.catalogId) ?? 0; + this.sessionCatalogRevisions.set(update.catalogId, catalogRevision + 1); + if (this.sessionCatalogLive.requestGeneration !== this.sessionCatalogGeneration) { + this.sessionCatalogLive.schedule( + SESSION_CATALOG_CHANGED_REFRESH_MS, + this.isConnected, + () => void this.refreshSessionCatalogs(), + ); + } + } + private async refreshSessionCatalogs() { - const client = this.context?.gateway.snapshot.client; - if (!client || !this.connected) { + const client = sessionCatalogListClient(this.context?.gateway.snapshot, this.connected); + if (!client) { return; } const generation = this.sessionCatalogGeneration; const revision = this.sessionCatalogRevision; const agentId = this.sessionCatalogAgentId ?? this.expandedAgentId(); - if (this.sessionCatalogRequestGeneration === generation) { - return; - } - this.sessionCatalogRequestGeneration = generation; - try { - const result = await client.request("sessions.catalog.list", { - agentId, - limitPerHost: 40, - }); - if (generation !== this.sessionCatalogGeneration || client !== this.gatewayClient) { - return; - } - const catalogs = await this.refetchSessionCatalogPages({ - catalogs: result.catalogs, - client, - generation, - agentId, - }); - if ( - generation !== this.sessionCatalogGeneration || - revision !== this.sessionCatalogRevision || - client !== this.gatewayClient - ) { - return; - } - const revisedCatalogIds = new Set([ - ...this.sessionCatalogs.map((catalog) => catalog.id), - ...catalogs.map((catalog) => catalog.id), - ]); - this.sessionCatalogs = catalogs; - for (const catalogId of revisedCatalogIds) { - this.sessionCatalogRevisions.set( - catalogId, - (this.sessionCatalogRevisions.get(catalogId) ?? 0) + 1, - ); - } - this.sessionCatalogRevision += 1; - } catch { - // A transient poll failure must not collapse already visible or expanded pages. - } finally { - if (this.sessionCatalogRequestGeneration === generation) { - this.sessionCatalogRequestGeneration = null; - } - if ( - generation === this.sessionCatalogGeneration && - client === this.gatewayClient && - this.isConnected - ) { - this.sessionCatalogTimer = globalThis.setTimeout(() => { - this.sessionCatalogTimer = null; - void this.refreshSessionCatalogs(); - }, 30_000); - } - } - } - - private async refetchSessionCatalogPages(params: { - catalogs: SessionCatalog[]; - client: GatewayBrowserClient; - generation: number; - agentId: string; - }): Promise { - const previousCatalogs = new Map(this.sessionCatalogs.map((catalog) => [catalog.id, catalog])); - return Promise.all( - params.catalogs.map(async (catalog) => { - const previousHosts = new Map( - previousCatalogs.get(catalog.id)?.hosts.map((host) => [host.hostId, host]) ?? [], - ); - const hosts = await Promise.all( - catalog.hosts.map(async (host) => { - const key = sessionCatalogHostKey(catalog.id, host.hostId); - const pageDepth = this.sessionCatalogPageDepths.get(key) ?? 0; - if (pageDepth === 0) { - return host; - } - const previous = previousHosts.get(host.hostId); - if (host.error) { - return preserveExpandedCatalogHost(host, previous); - } - let sessions = host.sessions; - let nextCursor = host.nextCursor; - let loadedPages = 0; - for (; loadedPages < pageDepth && nextCursor; loadedPages += 1) { - let result: SessionsCatalogListResult; - try { - result = await params.client.request( - "sessions.catalog.list", - { - agentId: params.agentId, - catalogId: catalog.id, - cursors: { [host.hostId]: nextCursor }, - }, - ); - } catch { - return previous ?? host; - } - if ( - params.generation !== this.sessionCatalogGeneration || - params.client !== this.gatewayClient - ) { - return previous ?? host; - } - const pageHost = result.catalogs - .find((candidate) => candidate.id === catalog.id) - ?.hosts.find((candidate) => candidate.hostId === host.hostId); - if (!pageHost) { - return previous ?? host; - } - if (pageHost.error) { - return preserveExpandedCatalogHost({ ...host, ...pageHost }, previous ?? host); - } - sessions = mergeCatalogSessionRows(sessions, pageHost.sessions); - nextCursor = pageHost.nextCursor; - } - const { - nextCursor: _firstPageCursor, - sessions: _firstPageSessions, - ...freshHost - } = host; - return { - ...freshHost, - sessions, - ...(nextCursor ? { nextCursor } : {}), - }; - }), - ); - return { ...catalog, hosts }; - }), - ); + await refreshSessionCatalogsLive({ + live: this.sessionCatalogLive, + client, + agentId, + generation, + revision, + currentGeneration: () => this.sessionCatalogGeneration, + currentRevision: () => this.sessionCatalogRevision, + currentClient: () => this.gatewayClient, + catalogs: () => this.sessionCatalogs, + pageDepths: this.sessionCatalogPageDepths, + connected: () => this.isConnected, + applyFinal: (catalogs, revisedCatalogIds) => { + this.sessionCatalogs = catalogs; + for (const catalogId of revisedCatalogIds) { + this.sessionCatalogRevisions.set( + catalogId, + (this.sessionCatalogRevisions.get(catalogId) ?? 0) + 1, + ); + } + this.sessionCatalogRevision += 1; + }, + refresh: () => void this.refreshSessionCatalogs(), + }); } protected async loadMoreSessionCatalog(catalogId: string) { @@ -539,10 +489,7 @@ export abstract class AppSidebarSessionDataElement extends AppSidebarBase { this.clearSessionCache(); this.sessionCatalogGeneration += 1; this.sessionCatalogRevision += 1; - if (this.sessionCatalogTimer) { - globalThis.clearTimeout(this.sessionCatalogTimer); - this.sessionCatalogTimer = null; - } + this.sessionCatalogLive.resetConnection(); this.sessionCatalogs = []; this.loadingMoreSessionCatalogIds = new Set(); this.sessionCatalogPageDepths.clear(); diff --git a/ui/src/components/app-sidebar.test.ts b/ui/src/components/app-sidebar.test.ts index 3f01c750507..2554a28b896 100644 --- a/ui/src/components/app-sidebar.test.ts +++ b/ui/src/components/app-sidebar.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { SessionCatalog, + SessionsCatalogHostEvent, SessionsCatalogListResult, } from "../../../packages/gateway-protocol/src/index.ts"; import { GatewayRequestError, type GatewayBrowserClient } from "../api/gateway.ts"; @@ -17,12 +18,13 @@ import { CATALOG_SESSION_CONTINUED_EVENT } from "../lib/sessions/catalog-key.ts" import type { SessionCapability } from "../lib/sessions/index.ts"; import { createApplicationContextProvider } from "../test-helpers/application-context.ts"; import { createStorageMock } from "../test-helpers/storage.ts"; +import { SessionCatalogLiveState } from "./app-sidebar-session-catalog-live.ts"; +import "./app-sidebar.ts"; import { LOBSTER_LOGO_VISIT_EVENT, createLobsterPetLook, type LobsterLogoVisitDetail, } from "./lobster-pet.ts"; -import "./app-sidebar.ts"; import { TERMINAL_PANEL_TOGGLE_EVENT } from "./panel-toggle-contract.ts"; type SessionGroupMutationResult = Awaited>; @@ -78,6 +80,7 @@ function createGatewayHarness(client: GatewayBrowserClient) { lastErrorCode: null, }; const listeners = new Set<(next: ApplicationGatewaySnapshot) => void>(); + const eventListeners = new Set<(event: { event: string; payload: unknown }) => void>(); const gateway = { get snapshot() { return snapshot; @@ -87,6 +90,10 @@ function createGatewayHarness(client: GatewayBrowserClient) { listeners.add(listener); return () => listeners.delete(listener); }, + subscribeEvents(listener: (event: { event: string; payload: unknown }) => void) { + eventListeners.add(listener); + return () => eventListeners.delete(listener); + }, } as unknown as ApplicationGateway; return { gateway, @@ -96,6 +103,11 @@ function createGatewayHarness(client: GatewayBrowserClient) { listener(snapshot); } }, + publishEvent(event: string, payload: unknown) { + for (const listener of eventListeners) { + listener({ event, payload }); + } + }, }; } @@ -1383,6 +1395,27 @@ describe("AppSidebar session scroll fade", () => { }); describe("AppSidebar session catalog pagination", () => { + it("keeps the current refetch guard when an older request finishes", () => { + const live = new SessionCatalogLiveState(); + const older = live.beginRefetch(true); + const current = live.beginRefetch(true); + + live.endRefetch(older); + expect(live.refetching).toBe(true); + live.endRefetch(current); + expect(live.refetching).toBe(false); + }); + + it("invalidates request ownership when live state is cleared", () => { + const live = new SessionCatalogLiveState(); + const first = live.beginRequest(1); + live.clear(); + const second = live.beginRequest(1); + + expect(live.ownsRequest(first.requestOwner)).toBe(false); + expect(live.ownsRequest(second.requestOwner)).toBe(true); + }); + const catalogPage = ( sessions: Array<{ threadId: string; name: string }>, nextCursor?: string, @@ -1684,6 +1717,496 @@ describe("AppSidebar session catalog pagination", () => { } }); + it("renders each catalog host before the aggregate list response finishes", async () => { + vi.useFakeTimers(); + try { + const pending = deferred(); + const request = vi.fn().mockReturnValue(pending.promise); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + const progressId = (request.mock.calls[0]?.[1] as { progressId?: string })?.progressId; + expect(progressId).toEqual(expect.any(String)); + const catalog = catalogPage([{ threadId: "thread-fast", name: "Fast host" }]).catalogs[0]; + if (!progressId || !catalog) { + throw new Error("progressive catalog fixture is incomplete"); + } + gateway.publishEvent("sessions.catalog.host", { + progressId, + agentId: "main", + catalog, + } satisfies SessionsCatalogHostEvent); + await sidebar.updateComplete; + + expect(sidebar.textContent).toContain("Fast host"); + expect(request).toHaveBeenCalledTimes(1); + + pending.resolve(catalogPage([{ threadId: "thread-fast", name: "Fast host" }])); + await vi.advanceTimersByTimeAsync(0); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects a late host event after a newer catalog request starts", async () => { + vi.useFakeTimers(); + try { + const initialPage = catalogPage([]); + const initialCatalog = initialPage.catalogs[0]; + const localHost = initialCatalog?.hosts[0]; + if (!initialCatalog || !localHost) { + throw new Error("initial catalog fixture is incomplete"); + } + const removedHost = { + ...localHost, + hostId: "node:removed", + label: "Removed node", + kind: "node" as const, + sessions: [], + error: { code: "NODE_INVOKE_FAILED", message: "Node timed out" }, + }; + const request = vi + .fn() + .mockResolvedValueOnce({ + catalogs: [{ ...initialCatalog, hosts: [localHost, removedHost] }], + }) + .mockResolvedValue(catalogPage([])); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + const oldProgressId = (request.mock.calls[0]?.[1] as { progressId?: string })?.progressId; + expect(oldProgressId).toEqual(expect.any(String)); + await vi.advanceTimersByTimeAsync(30_000); + expect(request).toHaveBeenCalledTimes(2); + + const staleCatalog = catalogPage([{ threadId: "thread-obsolete", name: "Obsolete session" }]) + .catalogs[0]; + const staleHost = staleCatalog?.hosts[0]; + if (!oldProgressId || !staleCatalog || !staleHost) { + throw new Error("stale catalog fixture is incomplete"); + } + gateway.publishEvent("sessions.catalog.host", { + progressId: oldProgressId, + agentId: "main", + catalog: { + ...staleCatalog, + hosts: [{ ...staleHost, hostId: "node:removed", label: "Removed node", kind: "node" }], + }, + } satisfies SessionsCatalogHostEvent); + await sidebar.updateComplete; + + expect(sidebar.textContent).not.toContain("Obsolete session"); + expect(sidebar.sessionCatalogs[0]?.hosts).toHaveLength(1); + } finally { + vi.useRealTimers(); + } + }); + + it("accepts a slow older host when a newer scan has only timed out", async () => { + vi.useFakeTimers(); + try { + const timeoutPage = catalogPage([]); + const catalog = timeoutPage.catalogs[0]; + const localHost = catalog?.hosts[0]; + if (!catalog || !localHost) { + throw new Error("slow catalog fixture is incomplete"); + } + const timeoutHost = { + ...localHost, + hostId: "node:slow", + label: "Slow node", + kind: "node" as const, + sessions: [], + error: { code: "NODE_INVOKE_FAILED", message: "Node timed out" }, + }; + const finalPage = { + catalogs: [{ ...catalog, hosts: [localHost, timeoutHost] }], + } satisfies SessionsCatalogListResult; + const pendingTimeout = deferred(); + const request = vi + .fn() + .mockResolvedValueOnce(finalPage) + .mockReturnValueOnce(pendingTimeout.promise) + .mockResolvedValue(finalPage); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + const oldProgressId = (request.mock.calls[0]?.[1] as { progressId?: string })?.progressId; + await vi.advanceTimersByTimeAsync(30_000); + if (!oldProgressId) { + throw new Error("first catalog request has no progress id"); + } + gateway.publishEvent("sessions.catalog.host", { + progressId: oldProgressId, + agentId: "main", + catalog: { + ...catalog, + hosts: [ + { + ...timeoutHost, + sessions: [ + { + threadId: "thread-eventual", + name: "Eventually ready", + status: "idle", + archived: false, + canContinue: true, + canArchive: true, + }, + ], + error: undefined, + }, + ], + }, + } satisfies SessionsCatalogHostEvent); + await sidebar.updateComplete; + + expect(sidebar.textContent).toContain("Eventually ready"); + pendingTimeout.resolve(finalPage); + await vi.advanceTimersByTimeAsync(0); + await sidebar.updateComplete; + expect(sidebar.textContent).toContain("Eventually ready"); + } finally { + vi.useRealTimers(); + } + }); + + it("discovers an external session on the stable poll and then follows changes quickly", async () => { + vi.useFakeTimers(); + try { + const empty = catalogPage([]); + const discovered = catalogPage([{ threadId: "thread-external", name: "External session" }]); + const request = vi + .fn() + .mockResolvedValueOnce(empty) + .mockResolvedValueOnce(discovered) + .mockResolvedValue(discovered); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + expect(request).toHaveBeenCalledTimes(1); + expect(sidebar.textContent).not.toContain("External session"); + + await vi.advanceTimersByTimeAsync(30_000); + await sidebar.updateComplete; + expect(request).toHaveBeenCalledTimes(2); + expect(sidebar.textContent).toContain("External session"); + + await vi.advanceTimersByTimeAsync(4_999); + expect(request).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + expect(request).toHaveBeenCalledTimes(3); + } finally { + vi.useRealTimers(); + } + }); + + it("returns to the stable cadence when only a progressive host order changed", async () => { + vi.useFakeTimers(); + try { + const basePage = catalogPage([{ threadId: "thread-local", name: "Local session" }]); + const catalog = basePage.catalogs[0]; + const localHost = catalog?.hosts[0]; + if (!catalog || !localHost) { + throw new Error("ordered catalog fixture is incomplete"); + } + const pairedHost = { + ...localHost, + hostId: "node:paired", + label: "A paired node", + kind: "node" as const, + sessions: [], + }; + const stablePage: SessionsCatalogListResult = { + catalogs: [ + { + ...catalog, + hosts: [{ ...localHost, label: "Z local Gateway" }, pairedHost], + }, + ], + }; + const pending = deferred(); + const request = vi + .fn() + .mockResolvedValueOnce(stablePage) + .mockReturnValueOnce(pending.promise) + .mockResolvedValue(stablePage); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + await vi.advanceTimersByTimeAsync(30_000); + const progressId = (request.mock.calls[1]?.[1] as { progressId?: string })?.progressId; + if (!progressId) { + throw new Error("second catalog request has no progress id"); + } + gateway.publishEvent("sessions.catalog.host", { + progressId, + agentId: "main", + catalog: { ...catalog, hosts: [pairedHost] }, + } satisfies SessionsCatalogHostEvent); + pending.resolve(stablePage); + await vi.advanceTimersByTimeAsync(0); + + await vi.advanceTimersByTimeAsync(5_000); + expect(request).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(25_000); + expect(request).toHaveBeenCalledTimes(3); + } finally { + vi.useRealTimers(); + } + }); + + it("refreshes immediately when paired-node presence changes", async () => { + vi.useFakeTimers(); + try { + const request = vi.fn().mockResolvedValue(catalogPage([])); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + gateway.publishEvent("presence", { + presence: [{ deviceId: "node-1", mode: "node", reason: "connect" }], + }); + expect(request).toHaveBeenCalledTimes(2); + + gateway.publishEvent("presence", { + presence: [{ deviceId: "node-1", mode: "node", reason: "disconnect" }], + }); + await vi.advanceTimersByTimeAsync(0); + expect(request).toHaveBeenCalledTimes(3); + } finally { + vi.useRealTimers(); + } + }); + + it("coalesces the visibility and focus events from one tab activation", async () => { + vi.useFakeTimers(); + let visibility: DocumentVisibilityState = "visible"; + const visibilitySpy = vi + .spyOn(document, "visibilityState", "get") + .mockImplementation(() => visibility); + try { + const request = vi.fn().mockResolvedValue(catalogPage([])); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + visibility = "hidden"; + document.dispatchEvent(new Event("visibilitychange")); + visibility = "visible"; + document.dispatchEvent(new Event("visibilitychange")); + globalThis.dispatchEvent(new Event("focus")); + await vi.advanceTimersByTimeAsync(49); + expect(request).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(request).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(0); + expect(request).toHaveBeenCalledTimes(2); + } finally { + visibilitySpy.mockRestore(); + vi.useRealTimers(); + } + }); + + it("does not poll an older Gateway that does not advertise session catalogs", async () => { + vi.useFakeTimers(); + try { + const request = vi.fn().mockResolvedValue(catalogPage([])); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + globalThis.dispatchEvent(new Event("focus")); + gateway.publishEvent("presence", { + presence: [{ deviceId: "node-1", mode: "node", reason: "connect" }], + }); + gateway.publishEvent("presence", { + presence: [{ deviceId: "node-1", mode: "node", reason: "disconnect" }], + }); + await vi.advanceTimersByTimeAsync(30_000); + + expect(request).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("falls back to final pages when an older Gateway rejects progressId", async () => { + vi.useFakeTimers(); + try { + const request = vi + .fn() + .mockRejectedValueOnce( + new GatewayRequestError({ code: "INVALID_REQUEST", message: "invalid params" }), + ) + .mockResolvedValue(catalogPage([])); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + expect(request).toHaveBeenNthCalledWith(1, "sessions.catalog.list", { + agentId: "main", + limitPerHost: 40, + progressId: expect.any(String), + }); + expect(request).toHaveBeenNthCalledWith(2, "sessions.catalog.list", { + agentId: "main", + limitPerHost: 40, + }); + + await vi.advanceTimersByTimeAsync(30_000); + expect(request).toHaveBeenNthCalledWith(3, "sessions.catalog.list", { + agentId: "main", + limitPerHost: 40, + }); + } finally { + vi.useRealTimers(); + } + }); + + it("ignores a legacy fallback that settles after the Gateway client changes", async () => { + vi.useFakeTimers(); + try { + const legacyFallback = deferred(); + const legacyRequest = vi + .fn() + .mockRejectedValueOnce( + new GatewayRequestError({ code: "INVALID_REQUEST", message: "invalid params" }), + ) + .mockReturnValueOnce(legacyFallback.promise); + const legacyGateway = createGatewayHarness({ + request: legacyRequest, + } as unknown as GatewayBrowserClient); + legacyGateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const sessions = createSessions("main", ["agent:main:main"]); + const { provider, sidebar } = await mountSidebar(legacyGateway.gateway, sessions); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + expect(legacyRequest).toHaveBeenCalledTimes(2); + + const currentRequest = vi.fn().mockResolvedValue(catalogPage([])); + const currentGateway = createGatewayHarness({ + request: currentRequest, + } as unknown as GatewayBrowserClient); + currentGateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + provider.setContext(createContext(currentGateway.gateway, sessions)); + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + legacyFallback.resolve(catalogPage([])); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(30_000); + + expect(currentRequest).toHaveBeenCalledTimes(2); + expect(currentRequest).toHaveBeenNthCalledWith(2, "sessions.catalog.list", { + agentId: "main", + limitPerHost: 40, + progressId: expect.any(String), + }); + } finally { + vi.useRealTimers(); + } + }); + it("refreshes catalog creation capability for the expanded agent", async () => { vi.useFakeTimers(); try { @@ -1712,6 +2235,7 @@ describe("AppSidebar session catalog pagination", () => { expect(request).toHaveBeenNthCalledWith(1, "sessions.catalog.list", { agentId: "main", limitPerHost: 40, + progressId: expect.any(String), }); const selection = context.agentSelection.state as { @@ -1727,6 +2251,7 @@ describe("AppSidebar session catalog pagination", () => { expect(request).toHaveBeenNthCalledWith(2, "sessions.catalog.list", { agentId: "research", limitPerHost: 40, + progressId: expect.any(String), }); } finally { vi.useRealTimers(); @@ -2078,6 +2603,7 @@ describe("AppSidebar session catalog pagination", () => { expect(request).toHaveBeenNthCalledWith(3, "sessions.catalog.list", { agentId: "main", limitPerHost: 40, + progressId: expect.any(String), }); expect(request).toHaveBeenNthCalledWith(4, "sessions.catalog.list", { agentId: "main", @@ -2105,6 +2631,66 @@ describe("AppSidebar session catalog pagination", () => { } }); + it("keeps a progressive host update that arrives during expanded-page refetch", async () => { + vi.useFakeTimers(); + try { + const pageOne = catalogPage([{ threadId: "thread-1", name: "Newest" }], "page-2"); + const pageTwo = catalogPage([{ threadId: "thread-2", name: "Older" }]); + const pendingRefetch = deferred(); + const request = vi + .fn() + .mockResolvedValueOnce(pageOne) + .mockResolvedValueOnce(pageTwo) + .mockResolvedValueOnce(pageOne) + .mockReturnValueOnce(pendingRefetch.promise) + .mockResolvedValue(pageOne); + const gateway = createGatewayHarness({ request } as unknown as GatewayBrowserClient); + gateway.publish({ + hello: { + features: { methods: ["sessions.catalog.list"] }, + } as ApplicationGatewaySnapshot["hello"], + }); + const { sidebar } = await mountSidebar( + gateway.gateway, + createSessions("main", ["agent:main:main"]), + ); + sidebar.connected = true; + await sidebar.updateComplete; + await vi.advanceTimersByTimeAsync(0); + + sidebar.querySelector('[data-session-catalog-load-more="codex"]')?.click(); + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(30_000); + expect(request).toHaveBeenCalledTimes(4); + + const progressId = (request.mock.calls[2]?.[1] as { progressId?: string })?.progressId; + const catalog = pageOne.catalogs[0]; + const host = catalog?.hosts[0]; + if (!progressId || !catalog || !host) { + throw new Error("expanded progressive fixture is incomplete"); + } + gateway.publishEvent("sessions.catalog.host", { + progressId, + agentId: "main", + catalog: { + ...catalog, + hosts: [{ ...host, label: "Progressive Gateway" }], + }, + } satisfies SessionsCatalogHostEvent); + await sidebar.updateComplete; + expect(sidebar.textContent).toContain("Progressive Gateway"); + + pendingRefetch.resolve(pageTwo); + await vi.advanceTimersByTimeAsync(0); + await sidebar.updateComplete; + + expect(sidebar.textContent).toContain("Progressive Gateway"); + expect(request).toHaveBeenCalledTimes(4); + } finally { + vi.useRealTimers(); + } + }); + it("discards a load-more response after a poll replaces its cursor", async () => { vi.useFakeTimers(); try { diff --git a/ui/src/e2e/codex-sessions.e2e.test.ts b/ui/src/e2e/codex-sessions.e2e.test.ts index 26176699fd0..723fa5b7bcd 100644 --- a/ui/src/e2e/codex-sessions.e2e.test.ts +++ b/ui/src/e2e/codex-sessions.e2e.test.ts @@ -2,6 +2,7 @@ import { mkdir } from "node:fs/promises"; import path from "node:path"; import { chromium, type Browser } from "playwright"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { SessionsCatalogHostEvent } from "../../../packages/gateway-protocol/src/index.ts"; import { canRunPlaywrightChromium, installMockGateway, @@ -90,6 +91,67 @@ suite("Codex native session catalog", () => { await page.close(); }); + it("shows a completed host while the aggregate catalog request is still pending", async () => { + const page = await browser.newPage({ viewport: { height: 900, width: 1280 } }); + const gateway = await installMockGateway(page, { + deferredMethods: ["sessions.catalog.list"], + featureMethods: ["chat.metadata", "chat.startup", "sessions.catalog.list"], + }); + + try { + await page.goto(`${server.baseUrl}chat`); + const request = await gateway.waitForRequest("sessions.catalog.list"); + const progressId = (request.params as { progressId?: string })?.progressId; + expect(progressId).toEqual(expect.any(String)); + if (!progressId) { + throw new Error("catalog request did not opt in to progressive host events"); + } + await gateway.emitGatewayEvent("sessions.catalog.host", { + progressId, + agentId: "main", + catalog: { + id: "codex", + label: "Codex", + capabilities: { continueSession: true, archive: true }, + hosts: [ + { + hostId: "node:fast", + label: "Fast Mac", + kind: "node", + connected: true, + nodeId: "fast", + sessions: [ + { + threadId: "thread-fast", + name: "Progressive node result", + status: "idle", + archived: false, + canContinue: true, + canArchive: false, + }, + ], + }, + ], + }, + } satisfies SessionsCatalogHostEvent); + + await page.getByText("Progressive node result", { exact: true }).waitFor(); + expect((await gateway.getRequests("sessions.catalog.list")).length).toBe(1); + if (captureUiProofEnabled) { + await mkdir(uiProofArtifactDir, { recursive: true }); + await page.screenshot({ + animations: "disabled", + fullPage: true, + path: path.join(uiProofArtifactDir, "05-progressive-host-result.png"), + }); + } + + await gateway.resolveDeferred("sessions.catalog.list", { catalogs: [] }); + } finally { + await page.close(); + } + }); + it("groups sessions by host and hides empty offline nodes", async () => { const page = await browser.newPage({ viewport: { height: 1100, width: 1440 } }); await page.addInitScript((key) => localStorage.removeItem(key), catalogGroupingStorageKey);