From b0e8a985ee11ac94aa76513f4da6b5693334c409 Mon Sep 17 00:00:00 2001 From: RoboClaw Date: Mon, 24 Aug 2026 11:50:36 -0700 Subject: [PATCH] feat(ui): load owner sessions before shared roster (#128767) * feat(ui): load owner sessions before shared roster Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> * fix(ui): preserve owner roster across refreshes Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> --------- Co-authored-by: roboclaw-bot <309084314+roboclaw-bot@users.noreply.github.com> Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> --- ui/src/e2e/session-owner-first.e2e.test.ts | 109 ++++++++++++++ .../lib/sessions/connection-hydration.test.ts | 134 +++++++++++++++++- .../lib/sessions/index.event-refresh.test.ts | 68 ++++++++- ui/src/lib/sessions/index.ts | 7 +- ui/src/lib/sessions/list-options.test.ts | 28 +++- ui/src/lib/sessions/session-capability.ts | 1 + ui/src/lib/sessions/session-roster-refresh.ts | 77 ++++++++-- 7 files changed, 404 insertions(+), 20 deletions(-) create mode 100644 ui/src/e2e/session-owner-first.e2e.test.ts diff --git a/ui/src/e2e/session-owner-first.e2e.test.ts b/ui/src/e2e/session-owner-first.e2e.test.ts new file mode 100644 index 000000000000..1890d5e32b14 --- /dev/null +++ b/ui/src/e2e/session-owner-first.e2e.test.ts @@ -0,0 +1,109 @@ +import { mkdir } from "node:fs/promises"; +import path from "node:path"; +import type { Page } from "playwright"; +import { expect, it } from "vitest"; +import { installMockGateway } from "../test-helpers/control-ui-e2e.ts"; +import { createControlUiE2eSuite } from "./control-ui-e2e-suite.test-support.ts"; + +const suite = createControlUiE2eSuite({ name: "Control UI owner-first session roster" }); +const captureProof = process.env.OPENCLAW_CAPTURE_UI_PROOF === "1"; +const proofDir = path.join(process.cwd(), ".artifacts", "control-ui-e2e", "session-owner-stack"); + +function sessionRoster(ownerId: string, key: string, label: string, updatedAt: number) { + const owner = { + type: "human" as const, + id: ownerId, + label: ownerId === "profile-ada" ? "Ada" : "Bob", + }; + return { + key, + kind: "direct" as const, + label, + createdActor: owner, + owner: { actor: owner }, + updatedAt, + }; +} + +function sessionsList() { + const sessions = [ + sessionRoster("profile-ada", "agent:main:ada", "Ada research", 2), + sessionRoster("profile-bob", "agent:main:bob", "Bob operations", 1), + ]; + return { + count: sessions.length, + owners: sessions.map((session) => session.owner.actor), + defaults: { contextTokens: null, model: null, modelProvider: null }, + path: "", + sessions, + ts: 1, + }; +} + +async function captureSidebar(page: Page, fileName: string) { + if (!captureProof) { + return; + } + await mkdir(proofDir, { recursive: true }); + await page.locator(".sidebar-sessions").screenshot({ + animations: "disabled", + path: path.join(proofDir, fileName), + }); +} + +suite.define(() => { + it("publishes the signed-in owner's sessions before the shared roster", async () => { + const context = await suite.browser.newContext({ viewport: { height: 800, width: 1200 } }); + const page = await context.newPage(); + const sharedRoster = sessionsList(); + const ownerRoster = { + ...sharedRoster, + count: 1, + owners: sharedRoster.owners.slice(0, 1), + sessions: sharedRoster.sessions.slice(0, 1), + }; + const gateway = await installMockGateway(page, { + deferredMethods: ["sessions.list"], + presenceUsers: [{ self: true, id: "profile-ada", name: "Ada" }], + sessionKey: "agent:main:ada", + methodResponses: { + "sessions.list": { + cases: [ + { match: { ownerId: "profile-ada" }, response: ownerRoster }, + { response: sharedRoster }, + ], + }, + }, + }); + + try { + await page.goto(`${suite.server?.baseUrl ?? ""}chat`); + await expect + .poll(async () => + (await gateway.getRequests("sessions.list")).some( + (request) => + (request.params as { ownerId?: unknown } | undefined)?.ownerId === "profile-ada", + ), + ) + .toBe(true); + await gateway.deferNext("sessions.list", { agentId: "main", limit: 60 }); + await gateway.resolveDeferred("sessions.list", ownerRoster); + + const adaRow = page.locator('[data-session-key="agent:main:ada"]'); + const bobRow = page.locator('[data-session-key="agent:main:bob"]'); + await adaRow.waitFor(); + await expect.poll(() => bobRow.count()).toBe(0); + await expect + .poll(async () => (await gateway.getRequests("sessions.list")).length) + .toBeGreaterThanOrEqual(2); + await captureSidebar(page, "owner-first-roster.png"); + + await gateway.resolveDeferred("sessions.list", sharedRoster); + await bobRow.waitFor(); + await expect.poll(() => adaRow.count()).toBe(1); + await captureSidebar(page, "owner-first-shared-roster.png"); + } finally { + await context.close(); + } + }); +}); diff --git a/ui/src/lib/sessions/connection-hydration.test.ts b/ui/src/lib/sessions/connection-hydration.test.ts index f325954cc5b3..c22de9441203 100644 --- a/ui/src/lib/sessions/connection-hydration.test.ts +++ b/ui/src/lib/sessions/connection-hydration.test.ts @@ -5,6 +5,7 @@ import { type GatewayProtocolSocketHandlers, } from "@openclaw/gateway-client/browser"; import { describe, expect, it, vi } from "vitest"; +import { createDeferred } from "../../../../test/helpers/promise.js"; import { GatewayRequestError, type GatewayBrowserClient, @@ -113,6 +114,116 @@ const targetedSessionReconciliationCases = [ ]; describe("session connection hydration", () => { + it("publishes the signed-in owner's roster before loading the shared roster", async () => { + const ownerResult: SessionsListResult = { + ...emptySessionsResult(), + count: 2, + sessions: [ + { + key: "agent:main:mine-new", + kind: "direct", + updatedAt: 3, + createdActor: { type: "human", id: "operator" }, + }, + { + key: "agent:main:mine-old", + kind: "direct", + updatedAt: 1, + createdActor: { type: "human", id: "operator" }, + }, + ], + }; + const sharedResult: SessionsListResult = { + ...emptySessionsResult(), + count: 2, + sessions: [ + ownerResult.sessions[0]!, + { + key: "agent:main:theirs", + kind: "direct", + updatedAt: 2, + createdActor: { type: "human", id: "collaborator" }, + }, + ], + }; + const ownerList = createDeferred(); + const sharedList = createDeferred(); + const queuedList = createDeferred(); + const request = vi.fn(async (method: string, params?: Record) => { + if (method === "sessions.subscribe") { + return { subscribed: true }; + } + if (method === "sessions.list") { + if (params?.ownerId === "operator") { + return await ownerList.promise; + } + return params?.search === "queued" ? await queuedList.promise : await sharedList.promise; + } + throw new Error(`Unexpected request: ${method}`); + }); + const client = { request } as unknown as GatewayBrowserClient; + let snapshot = { + client: null as GatewayBrowserClient | null, + phase: "reconnecting" as "connected" | "reconnecting", + sessionKey: "agent:main:mine-new", + assistantAgentId: "main" as string | null, + hello: null as GatewayHelloOk | null, + selfUser: { id: "operator", name: "Operator" }, + }; + let gatewayListener: ((next: typeof snapshot) => void) | undefined; + const sessions = createSessionCapability({ + get snapshot() { + return snapshot; + }, + subscribe(listener) { + gatewayListener = listener; + return () => undefined; + }, + subscribeEvents: () => () => undefined, + }); + + snapshot = { ...snapshot, client, phase: "connected" }; + gatewayListener?.(snapshot); + + await waitForFast(() => + expect(request).toHaveBeenCalledWith( + "sessions.list", + expect.objectContaining({ ownerId: "operator", limit: 60 }), + ), + ); + expect(request.mock.calls.filter(([method]) => method === "sessions.list")).toHaveLength(1); + const queuedRefresh = sessions.refresh({ agentId: "other", search: "queued", force: true }); + + ownerList.resolve(ownerResult); + await waitForFast(() => expect(sessions.state.result).toBe(ownerResult)); + await waitForFast(() => + expect(request.mock.calls.filter(([method]) => method === "sessions.list")).toHaveLength(2), + ); + expect(sessions.state.loading).toBe(false); + expect(request.mock.calls.at(-1)?.[1]).toEqual( + expect.objectContaining({ agentId: "main", limit: 60 }), + ); + expect(request.mock.calls.at(-1)?.[1]).not.toHaveProperty("ownerId"); + + sharedList.resolve(sharedResult); + await waitForFast(() => + expect(sessions.state.result?.sessions.map((row) => row.key)).toEqual([ + "agent:main:mine-new", + "agent:main:mine-old", + "agent:main:theirs", + ]), + ); + await waitForFast(() => + expect(request.mock.calls.filter(([method]) => method === "sessions.list")).toHaveLength(3), + ); + expect(request.mock.calls.at(-1)?.[1]).toEqual( + expect.objectContaining({ agentId: "other", search: "queued" }), + ); + queuedList.resolve(emptySessionsResult()); + await queuedRefresh; + sessions.dispose(); + }); + it("uses the selected agent before a session key is available", async () => { const result: SessionsListResult = { ts: 1, @@ -164,7 +275,7 @@ describe("session connection hydration", () => { sessions.dispose(); }); - it("ignores same-connection gateway metadata snapshots during hydration", async () => { + it("rehydrates owner sessions when identity arrives on the same connection", async () => { let resolveList: (result: SessionsListResult) => void = () => undefined; const pendingList = new Promise((resolve) => { resolveList = resolve; @@ -177,7 +288,7 @@ describe("session connection hydration", () => { defaults: { modelProvider: null, model: null, contextTokens: null }, sessions: [], }; - const request = vi.fn(async (method: string) => { + const request = vi.fn(async (method: string, _params?: Record) => { if (method === "sessions.subscribe") { return { subscribed: true }; } @@ -215,14 +326,25 @@ describe("session connection hydration", () => { snapshot = { ...snapshot, canvasPluginSurfaceUrl: "https://gateway.example.test/canvas" }; gatewayListener?.(snapshot); + await Promise.resolve(); + expect(listCalls).toBe(1); + snapshot = { ...snapshot, selfUser: { id: "operator", name: "Operator" } }; gatewayListener?.(snapshot); resolveList(result); - await waitForFast(() => expect(sessions.state.result).toBe(result)); - await Promise.resolve(); - await Promise.resolve(); + await waitForFast(() => expect(listCalls).toBe(3)); - expect(listCalls).toBe(1); + expect( + request.mock.calls + .filter(([method]) => method === "sessions.list") + .map(([, params]) => params), + ).toEqual([ + expect.not.objectContaining({ ownerId: expect.anything() }), + expect.objectContaining({ ownerId: "operator", limit: 60 }), + expect.not.objectContaining({ ownerId: expect.anything() }), + ]); + expect(sessions.state.result?.sessions).toEqual([]); + expect(sessions.state.agentId).toBe("main"); sessions.dispose(); }); diff --git a/ui/src/lib/sessions/index.event-refresh.test.ts b/ui/src/lib/sessions/index.event-refresh.test.ts index 45546baafb23..0238a143b30c 100644 --- a/ui/src/lib/sessions/index.event-refresh.test.ts +++ b/ui/src/lib/sessions/index.event-refresh.test.ts @@ -36,7 +36,7 @@ function sessionChangedEvent(key: string): GatewayEventFrame { }; } -function createHarness(request: GatewayBrowserClient["request"]) { +function createHarness(request: GatewayBrowserClient["request"], ownerId?: string) { const client = { request } as GatewayBrowserClient; let eventListener: ((event: GatewayEventFrame) => void) | undefined; const sessions = createSessionCapability({ @@ -46,6 +46,7 @@ function createHarness(request: GatewayBrowserClient["request"]) { sessionKey: "agent:main:main", assistantAgentId: "main", hello: null, + selfUser: ownerId ? { id: ownerId } : null, }, subscribe: () => () => undefined, subscribeEvents(listener) { @@ -582,6 +583,71 @@ describe("event-driven session list refresh", () => { } }); + it("retains owner and appended shared pages when an event replaces the list", async () => { + vi.useFakeTimers(); + const ownerId = "profile-ada"; + const ownerTail = { + key: "agent:main:owner-tail", + kind: "direct" as const, + updatedAt: 1, + createdActor: { type: "human" as const, id: ownerId }, + }; + const ownerHead = { + key: "agent:main:owner-head", + kind: "direct" as const, + updatedAt: 3, + createdActor: { type: "human" as const, id: ownerId }, + }; + const sharedRows = [ + ownerHead, + ...Array.from({ length: 119 }, (_, index) => ({ + key: `agent:main:shared-${index}`, + kind: "direct" as const, + updatedAt: 119 - index, + createdActor: { type: "human" as const, id: "profile-bob" }, + })), + ]; + const request = vi.fn( + async (method: string, params?: { limit?: number; offset?: number; ownerId?: string }) => { + if (method !== "sessions.list") { + throw new Error(`Unexpected request: ${method}`); + } + if (params?.ownerId === ownerId) { + return sessionsResult(1, [ownerHead, ownerTail]); + } + const offset = params?.offset ?? 0; + return sessionsResult(2, sharedRows.slice(offset, offset + (params?.limit ?? 50))); + }, + ); + const { sessions, emitEvent } = createHarness( + request as unknown as GatewayBrowserClient["request"], + ownerId, + ); + + try { + await sessions.refresh({ agentId: "main", limit: 60, force: true }); + await sessions.refresh({ agentId: "main", limit: 60, offset: 60, append: true, force: true }); + expect(sessions.state.result?.sessions).toHaveLength(121); + + emitEvent(sessionChangedEvent(sharedRows[1]!.key)); + await vi.advanceTimersByTimeAsync(SESSION_EVENT_REFRESH_DEBOUNCE_MS); + + expect(request.mock.calls).toHaveLength(5); + expect(request.mock.calls.map(([, params]) => params)).toEqual([ + expect.objectContaining({ ownerId, limit: 60 }), + expect.objectContaining({ limit: 60 }), + expect.objectContaining({ limit: 60, offset: 60 }), + expect.objectContaining({ ownerId, limit: 60 }), + expect.objectContaining({ limit: 120 }), + ]); + expect(sessions.state.result?.sessions).toHaveLength(121); + expect(sessions.state.result?.sessions.map((row) => row.key)).toContain(ownerTail.key); + } finally { + sessions.dispose(); + vi.useRealTimers(); + } + }); + it("clears a recreated session's prior deletion before the debounced refresh", async () => { vi.useFakeTimers(); const key = "agent:main:recreated-thread"; diff --git a/ui/src/lib/sessions/index.ts b/ui/src/lib/sessions/index.ts index 1b32ac7b0f28..086c0f5f5820 100644 --- a/ui/src/lib/sessions/index.ts +++ b/ui/src/lib/sessions/index.ts @@ -104,6 +104,7 @@ export function createSessionCapability(gateway: SessionGateway): SessionCapabil const createdListeners = new Set<(key: string) => void>(); let canonicalListRevision = 0; let hydratedClient: SessionGateway["snapshot"]["client"] = null; + let hydratedSelfUserId: string | null = null; let connectionClient = gateway.snapshot.client; let sessionEventSubscriptionError: string | null = null; let publishedErrorSource: "session-observer" | "operation" | null = null; @@ -340,6 +341,7 @@ export function createSessionCapability(gateway: SessionGateway): SessionCapabil const stopGateway = gateway.subscribe((next) => { const previousClient = connectionClient; const connected = next.phase === "connected"; + const selfUserId = next.selfUser?.id.trim() || null; const connectionChanged = connection.transition(next); connectionClient = next.client; if (connectionChanged) { @@ -360,6 +362,7 @@ export function createSessionCapability(gateway: SessionGateway): SessionCapabil } if (!connected || !next.client) { hydratedClient = null; + hydratedSelfUserId = null; publish({ result: null, agentId: null, @@ -373,12 +376,13 @@ export function createSessionCapability(gateway: SessionGateway): SessionCapabil }); return; } - if (hydratedClient !== next.client) { + if (hydratedClient !== next.client || hydratedSelfUserId !== selfUserId) { const scope = connection.capture(); if (!scope) { return; } hydratedClient = scope.client; + hydratedSelfUserId = selfUserId; void (async () => { await sessionEventSubscription.ensure(scope); if (connection.isCurrent(scope)) { @@ -571,6 +575,7 @@ export function createSessionCapability(gateway: SessionGateway): SessionCapabil connection.dispose(); groups.dispose(); hydratedClient = null; + hydratedSelfUserId = null; mutations.dispose(); swarmActivity.clear(); pullRequestSummaries.clear(); diff --git a/ui/src/lib/sessions/list-options.test.ts b/ui/src/lib/sessions/list-options.test.ts index 4e3af3c6d12c..439f1b430996 100644 --- a/ui/src/lib/sessions/list-options.test.ts +++ b/ui/src/lib/sessions/list-options.test.ts @@ -23,7 +23,7 @@ function deferred() { return { promise, reject, resolve }; } -function createSessions(client: GatewayBrowserClient, key: string) { +function createSessions(client: GatewayBrowserClient, key: string, ownerId?: string) { return createSessionCapability({ snapshot: { client, @@ -31,6 +31,7 @@ function createSessions(client: GatewayBrowserClient, key: string) { sessionKey: key, assistantAgentId: "main", hello: null, + selfUser: ownerId ? { id: ownerId } : null, }, subscribe: () => () => undefined, subscribeEvents: () => () => undefined, @@ -38,6 +39,31 @@ function createSessions(client: GatewayBrowserClient, key: string) { } describe("session list replacement options", () => { + it.each([ + { filter: "owner", options: { ownerId: "profile-bob" } }, + { filter: "involving me", options: { involvingMe: true } }, + { filter: "search", options: { search: "release" } }, + ])("keeps an explicit $filter query single-phase", async ({ options }) => { + const request = vi.fn(async (method: string, _params?: unknown) => { + if (method === "sessions.list") { + return sessionsResult([], 1); + } + throw new Error(`Unexpected request: ${method}`); + }); + const sessions = createSessions( + { request } as unknown as GatewayBrowserClient, + "agent:main:main", + "profile-ada", + ); + + await sessions.refresh({ agentId: "main", ...options, force: true }); + + const listCalls = request.mock.calls.filter(([method]) => method === "sessions.list"); + expect(listCalls).toHaveLength(1); + expect(listCalls[0]?.[1]).toEqual(expect.objectContaining(options)); + sessions.dispose(); + }); + it("preserves sidebar metadata hydration when refreshing after session patches", async () => { const key = "agent:main:untitled"; const request = vi.fn(async (method: string, _params?: unknown) => { diff --git a/ui/src/lib/sessions/session-capability.ts b/ui/src/lib/sessions/session-capability.ts index d9278e7f4311..020423afd45a 100644 --- a/ui/src/lib/sessions/session-capability.ts +++ b/ui/src/lib/sessions/session-capability.ts @@ -126,6 +126,7 @@ export type SessionGateway = { hello: GatewayHelloOk | null; assistantAgentId?: string | null; sessionKey?: string; + selfUser?: { readonly id: string } | null; }; subscribe: (listener: (snapshot: SessionGateway["snapshot"]) => void) => () => void; subscribeEvents: (listener: (event: GatewayEventFrame) => void) => () => void; diff --git a/ui/src/lib/sessions/session-roster-refresh.ts b/ui/src/lib/sessions/session-roster-refresh.ts index 41fc8cdf8aed..5867c3fd33ba 100644 --- a/ui/src/lib/sessions/session-roster-refresh.ts +++ b/ui/src/lib/sessions/session-roster-refresh.ts @@ -41,6 +41,13 @@ type ManagedSessionListRefresh = { invalidated?: true; }; +type SessionRosterLoadOptions = SessionRefreshOptions & { + mergeExisting?: boolean; + provisional?: boolean; +}; + +const OWNER_FIRST_SESSION_LIST_LIMIT = 60; + type ManagedSessionListQuery = Readonly> & { readonly limit: number }; type ManagedSessionList = { @@ -224,22 +231,29 @@ export function createSessionRosterRefresh(host: SessionRosterRefreshHost) { } }; - const load = async (options: SessionRefreshOptions) => { + const load = async (options: SessionRosterLoadOptions) => { const scope = host.connection.capture(); if (!scope) { return; } - const { append = false, force: _force, backgroundHydrate = false, ...requestOptions } = options; + const { + append = false, + force: _force, + backgroundHydrate = false, + mergeExisting = false, + provisional = false, + ...requestOptions + } = options; // Every canonical roster replaces visible session names, so omitted title // enrichment must inherit the UI default instead of publishing fallback ids. requestOptions.includeDerivedTitles ??= true; const durableListOptions: SessionListOptions = { ...requestOptions }; // Pagination is request-local; replacements retain filters but restart at page one. delete durableListOptions.offset; - if (!backgroundHydrate) { + if (!backgroundHydrate && !provisional) { lastListOptions = durableListOptions; hasForegroundListOptions = true; - } else if (!hasForegroundListOptions && !hasSeededListOptions) { + } else if (!provisional && !hasForegroundListOptions && !hasSeededListOptions) { lastListOptions = durableListOptions; hasSeededListOptions = true; } @@ -256,17 +270,26 @@ export function createSessionRosterRefresh(host: SessionRosterRefreshHost) { return; } const currentState = host.readState(); + const mergeWithCurrent = + mergeExisting || (append && typeof requestOptions.offset === "number"); let nextResult = - result && append && requestOptions.offset && currentState.result + result && mergeWithCurrent && currentState.result ? appendSessionResults(currentState.result, result) : reconcileRosterPresentationMetadata(result, currentState.result); if (append && nextResult && !backgroundHydrate) { - // Canonical event refreshes must retain all previously appended visible pages. + const ownerFirstPage = + Boolean(host.snapshot().selfUser?.id.trim()) && + isPrimarySessionListQuery(durableListOptions); + const retainedListLimit = + ownerFirstPage && result && typeof requestOptions.offset === "number" + ? requestOptions.offset + result.sessions.length + : nextResult.sessions.length; + // Retain the shared pagination window, excluding owner rows merged ahead of it. lastListOptions = { ...durableListOptions, limit: Math.max( durableListOptions.limit ?? DEFAULT_SESSION_LIST_QUERY.limit, - nextResult.sessions.length, + retainedListLimit, ), }; } @@ -303,7 +326,9 @@ export function createSessionRosterRefresh(host: SessionRosterRefreshHost) { } } nextResult = host.decorate(nextResult); - host.onCanonicalList(nextResult); + if (!provisional) { + host.onCanonicalList(nextResult); + } const state = host.readState(); const error = host.observerError(); host.publish( @@ -361,6 +386,34 @@ export function createSessionRosterRefresh(host: SessionRosterRefreshHost) { return { ...lastListOptions, force: true }; }; + const refreshPlan = (options: SessionRefreshOptions): SessionRosterLoadOptions[] => { + const ownerId = host.snapshot().selfUser?.id.trim(); + if (!ownerId || options.append === true || !isPrimarySessionListQuery(options)) { + return [options]; + } + const sharedLimit = Math.max( + OWNER_FIRST_SESSION_LIST_LIMIT, + typeof options.limit === "number" && options.limit > 0 + ? Math.floor(options.limit) + : DEFAULT_SESSION_LIST_QUERY.limit, + ); + // Keep owner-first and shared loads atomic in the existing refresh queue. + // Only the shared phase advances canonical membership and durable options. + return [ + { + ...options, + ownerId, + limit: OWNER_FIRST_SESSION_LIST_LIMIT, + provisional: true, + }, + { + ...options, + limit: sharedLimit, + mergeExisting: true, + }, + ]; + }; + const drainRefreshQueue = async (options: SessionRefreshOptions) => { const scope = host.connection.capture(); if (!scope) { @@ -368,9 +421,11 @@ export function createSessionRosterRefresh(host: SessionRosterRefreshHost) { } let next: SessionRefreshOptions | null = options; while (next) { - await load(next); - if (!host.connection.isCurrent(scope)) { - return; + for (const loadOptions of refreshPlan(next)) { + await load(loadOptions); + if (!host.connection.isCurrent(scope)) { + return; + } } next = takeNextQueuedRefresh(); }