mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 02:06:43 +00:00
improve: stream native sessions as hosts finish (#110211)
* perf: stream native session catalogs by host * fix: satisfy session catalog CI contracts * fix: retain changed-session refresh timing * fix: keep catalog refresh helpers private
This commit is contained in:
Peter Steinberger
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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`).
|
||||
|
||||
@@ -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<unknown>((resolve) => {
|
||||
resolveInvoke = resolve;
|
||||
});
|
||||
const invoke = vi.fn<PluginRuntime["nodes"]["invoke"]>(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;
|
||||
|
||||
@@ -993,6 +993,7 @@ function parseGatewayQuery(value: unknown): {
|
||||
async function listClaudeSessionCatalog(params: {
|
||||
runtime: PluginRuntime;
|
||||
query?: unknown;
|
||||
onHost?: (host: ClaudeSessionCatalogHost) => void;
|
||||
}): Promise<ClaudeSessionCatalogResult> {
|
||||
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({
|
||||
|
||||
@@ -83,6 +83,7 @@ export async function listPairedNode(params: {
|
||||
node: CatalogNode;
|
||||
query: CodexSessionCatalogParams;
|
||||
adoptedSessions: ReadonlyMap<string, AdoptedSessionEntry>;
|
||||
onHost?: (host: CodexSessionCatalogHost) => void;
|
||||
}): Promise<CodexSessionCatalogHost> {
|
||||
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,
|
||||
|
||||
@@ -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<unknown>((resolve) => {
|
||||
resolveInvoke = resolve;
|
||||
});
|
||||
const invoke = vi.fn<PluginRuntime["nodes"]["invoke"]>(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: [
|
||||
|
||||
@@ -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<CodexSessionCatalogResult> {
|
||||
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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -12,6 +12,7 @@ export type {
|
||||
SessionsCatalogContinueResult,
|
||||
SessionsCatalogListParams,
|
||||
SessionsCatalogListResult,
|
||||
SessionsCatalogHostEvent,
|
||||
SessionsCatalogReadParams,
|
||||
SessionsCatalogReadResult,
|
||||
} from "./schema/sessions-catalog.js";
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<typeof SessionCatalogHostSchema>;
|
||||
export type SessionCatalog = Static<typeof SessionCatalogSchema>;
|
||||
export type SessionsCatalogListParams = Static<typeof SessionsCatalogListParamsSchema>;
|
||||
export type SessionsCatalogListResult = Static<typeof SessionsCatalogListResultSchema>;
|
||||
export type SessionsCatalogHostEvent = Static<typeof SessionsCatalogHostEventSchema>;
|
||||
export type SessionCatalogTranscriptItem = Static<typeof SessionCatalogTranscriptItemSchema>;
|
||||
export type SessionsCatalogReadParams = Static<typeof SessionsCatalogReadParamsSchema>;
|
||||
export type SessionsCatalogReadResult = Static<typeof SessionsCatalogReadResultSchema>;
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -56,6 +56,7 @@ const EVENT_SCOPE_GUARDS: Record<string, string[]> = {
|
||||
"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],
|
||||
|
||||
@@ -53,14 +53,15 @@ async function call(
|
||||
method: keyof typeof sessionCatalogHandlers,
|
||||
params: unknown,
|
||||
config: Record<string, unknown> = {},
|
||||
client?: { connect?: { scopes?: string[] } },
|
||||
client?: { connect?: { scopes?: string[] }; connId?: string },
|
||||
contextOverrides: Record<string, unknown> = {},
|
||||
) {
|
||||
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: [] }));
|
||||
|
||||
@@ -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<SessionCatalog> => {
|
||||
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) {
|
||||
|
||||
@@ -12,6 +12,8 @@ export type SessionCatalogListProviderParams = {
|
||||
limitPerHost?: number;
|
||||
hostIds?: string[];
|
||||
cursors?: Record<string, string>;
|
||||
/** Publishes completed hosts without waiting for slower machines in the same list. */
|
||||
onHost?: (host: SessionCatalogHost) => void;
|
||||
};
|
||||
export type SessionCatalogReadProviderParams = Omit<SessionsCatalogReadParams, "catalogId">;
|
||||
export type SessionCatalogContinueProviderParams = Omit<
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
const catalog = event.catalog;
|
||||
if (!catalog || typeof catalog !== "object") {
|
||||
return false;
|
||||
}
|
||||
const catalogRecord = catalog as Record<string, unknown>;
|
||||
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<string, unknown>).hostId === "string" &&
|
||||
Array.isArray((host as Record<string, unknown>).sessions)
|
||||
);
|
||||
}
|
||||
|
||||
/** Tracks one sidebar's progressive list streams and adaptive refresh lifecycle. */
|
||||
export class SessionCatalogLiveState {
|
||||
timer: ReturnType<typeof globalThis.setTimeout> | null = null;
|
||||
requestGeneration: number | null = null;
|
||||
sawChange = false;
|
||||
refreshPending = false;
|
||||
progressive = true;
|
||||
|
||||
private activationTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
|
||||
private connectionEpoch = 0;
|
||||
private refetchOwner: symbol | null = null;
|
||||
private presenceSignature: string | null = null;
|
||||
private progressSequence = 0;
|
||||
private readonly progressSequences = new Map<string, number>();
|
||||
private readonly hostProgressSequences = new Map<string, number>();
|
||||
private readonly hostIdsByCatalog = new Map<string, ReadonlySet<string>>();
|
||||
private readonly requestChangedHostKeys = new Set<string>();
|
||||
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<SessionsCatalogListResult> {
|
||||
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<string>();
|
||||
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<string, "connected" | "offline">();
|
||||
for (const entry of presence) {
|
||||
if (!entry || typeof entry !== "object") {
|
||||
continue;
|
||||
}
|
||||
const record = entry as Record<string, unknown>;
|
||||
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<string, number>;
|
||||
}): { 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<string, number>;
|
||||
connected: () => boolean;
|
||||
applyFinal: (catalogs: SessionCatalog[], revisedCatalogIds: ReadonlySet<string>) => 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,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<SessionCatalog["error"]>;
|
||||
|
||||
@@ -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<string, number>;
|
||||
isCurrent: () => boolean;
|
||||
}): Promise<SessionCatalog[]> {
|
||||
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<SessionsCatalogListResult>(
|
||||
"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 };
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -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<typeof globalThis.setTimeout> | 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<string, number>();
|
||||
private readonly sessionCatalogRevisions = new Map<string, number>();
|
||||
|
||||
@@ -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<SessionsCatalogListResult>("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<SessionCatalog[]> {
|
||||
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<SessionsCatalogListResult>(
|
||||
"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();
|
||||
|
||||
@@ -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<ReturnType<SessionCapability["groupsRename"]>>;
|
||||
@@ -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<SessionsCatalogListResult>();
|
||||
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<SessionsCatalogListResult>();
|
||||
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<SessionsCatalogListResult>();
|
||||
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<SessionsCatalogListResult>();
|
||||
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<SessionsCatalogListResult>();
|
||||
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<HTMLButtonElement>('[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 {
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user