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

@@ -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();
+    }
+  });
+});

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<SessionsListResult>();
+    const sharedList = createDeferred<SessionsListResult>();
+    const queuedList = createDeferred<SessionsListResult>();
+    const request = vi.fn(async (method: string, params?: Record<string, unknown>) => {
+      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<SessionsListResult>((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<string, unknown>) => {
       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();
   });
 

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";

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();

ui/src/lib/sessions/list-options.test.ts

@@ -23,7 +23,7 @@ function deferred<T>() {
   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) => {

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;

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<Record<string, unknown>> & { 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();
     }