diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 3e8165146c..430e4ade74 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -50,7 +50,7 @@ import { useToast } from "@/contexts/toast-context"; import { toErrorMessage } from "@/utils/error-messages"; import { showProviderNoticeToast } from "@/utils/provider-notice-toast"; import { applyCheckoutStatusUpdateFromEvent } from "@/git/checkout-status-cache"; -import { useProviderSubagentStore } from "@/subagents/provider-store"; +import { resyncProviderSubagents, useProviderSubagentStore } from "@/subagents/provider-store"; // Re-export types from session-store and draft-store for backward compatibility export type { DraftInput } from "@/stores/draft-store"; @@ -532,9 +532,14 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } }); - const unsubProviderSubagentUpdate = onFeed("agent.provider_subagents.update", (message) => { - if (message.type !== "agent.provider_subagents.update") return; - useProviderSubagentStore.getState().applyUpdate(serverId, message.payload); + const unsubProviderSubagentUpdate = feeds.subscribe({ + snapshot: () => { + void resyncProviderSubagents(client, serverId); + }, + update: (message) => { + if (message.type !== "agent.provider_subagents.update") return; + useProviderSubagentStore.getState().applyUpdate(serverId, message.payload); + }, }); const unsubCheckoutStatusUpdate = onFeed("checkout_status_update", (message) => { diff --git a/packages/app/src/subagents/provider-store.test.ts b/packages/app/src/subagents/provider-store.test.ts index 96badfd354..01d6880d5b 100644 --- a/packages/app/src/subagents/provider-store.test.ts +++ b/packages/app/src/subagents/provider-store.test.ts @@ -1,8 +1,11 @@ import { DaemonConnectionError } from "@getpaseo/client/internal/daemon-client"; import { afterEach, describe, expect, test } from "vitest"; +import type { ProviderSubagentDescriptorPayload } from "@getpaseo/protocol/messages"; import { observeProviderSubagentTimeline, providerSubagentKey, + refreshProviderSubagents, + resyncProviderSubagents, useProviderSubagentStore, } from "./provider-store"; @@ -15,6 +18,7 @@ afterEach(() => { descriptors: new Map(), timelines: new Map(), hiddenFromTrack: new Set(), + trackedParents: new Set(), }); }); @@ -727,3 +731,96 @@ describe("projected child history", () => { expect(current().lastSeq).toBe(2000); }); }); + +describe("resyncing after the update feed resubscribes", () => { + function child( + id: string, + parentAgentId: string, + status: ProviderSubagentDescriptorPayload["status"], + ): ProviderSubagentDescriptorPayload { + return { + id, + parentAgentId, + provider: "claude", + title: "Explore", + description: null, + status, + createdAt: "2026-10-07T10:00:00.000Z", + updatedAt: "2026-10-07T10:00:00.000Z", + toolCallId: id, + }; + } + + function fakeDaemon(lists: Record) { + const requested: string[] = []; + return { + requested, + lists, + client: { + listProviderSubagents: async (parentAgentId: string) => { + requested.push(parentAgentId); + return { + requestId: "list", + parentAgentId, + subagents: lists[parentAgentId] ?? [], + error: null, + }; + }, + }, + }; + } + + const status = (parentAgentId: string, subagentId: string) => + useProviderSubagentStore + .getState() + .descriptors.get(providerSubagentKey(SERVER_ID, parentAgentId, subagentId))?.status; + + test("catches up on children that finished or started while updates were not delivered", async () => { + const daemon = fakeDaemon({ [PARENT_ID]: [child("child-1", PARENT_ID, "running")] }); + await refreshProviderSubagents(daemon.client, SERVER_ID, PARENT_ID); + expect(status(PARENT_ID, "child-1")).toBe("running"); + + // The connection dropped: the daemon finished child-1 and ran child-2 without telling us. + daemon.lists[PARENT_ID] = [ + child("child-1", PARENT_ID, "completed"), + child("child-2", PARENT_ID, "completed"), + ]; + await resyncProviderSubagents(daemon.client, SERVER_ID); + + expect(status(PARENT_ID, "child-1")).toBe("completed"); + expect(status(PARENT_ID, "child-2")).toBe("completed"); + }); + + test("catches up on a parent whose first list failed while disconnected", async () => { + const daemon = fakeDaemon({}); + const disconnected = { + listProviderSubagents: async () => { + throw new DaemonConnectionError("Transport not connected"); + }, + }; + await expect( + refreshProviderSubagents(disconnected, SERVER_ID, PARENT_ID), + ).rejects.toBeInstanceOf(DaemonConnectionError); + useProviderSubagentStore.getState().applyUpdate(SERVER_ID, { + kind: "upsert", + subagent: child("child-1", PARENT_ID, "running"), + }); + + // The connection dropped again: the daemon finished child-1 without telling us. + daemon.lists[PARENT_ID] = [child("child-1", PARENT_ID, "completed")]; + await resyncProviderSubagents(daemon.client, SERVER_ID); + + expect(status(PARENT_ID, "child-1")).toBe("completed"); + }); + + test("lists again only the parents this server was showing", async () => { + const daemon = fakeDaemon({}); + await refreshProviderSubagents(daemon.client, SERVER_ID, PARENT_ID); + await refreshProviderSubagents(daemon.client, "server-2", "other-parent"); + daemon.requested.length = 0; + + await resyncProviderSubagents(daemon.client, SERVER_ID); + + expect(daemon.requested).toEqual([PARENT_ID]); + }); +}); diff --git a/packages/app/src/subagents/provider-store.ts b/packages/app/src/subagents/provider-store.ts index 86e857d774..fd44dae8d2 100644 --- a/packages/app/src/subagents/provider-store.ts +++ b/packages/app/src/subagents/provider-store.ts @@ -28,6 +28,9 @@ interface ProviderSubagentState { descriptors: Map; timelines: Map; hiddenFromTrack: Set; + /** Parents whose list this client asked for, so a resubscribed update feed can catch them up. */ + trackedParents: Set; + trackParent(serverId: string, parentAgentId: string): void; hideFromTrack(serverId: string, parentAgentId: string, subagentIds: readonly string[]): void; replaceList( serverId: string, @@ -84,6 +87,7 @@ export function refreshProviderSubagents( const pending = clientRequests.get(requestKey); if (pending) return pending; + useProviderSubagentStore.getState().trackParent(serverId, parentAgentId); const request = client .listProviderSubagents(parentAgentId) .then((payload) => { @@ -101,6 +105,27 @@ function parentPrefix(serverId: string, parentAgentId: string): string { return `${serverId}\0${parentAgentId}\0`; } +/** + * Updates are not replayed across a reconnect, so a child that finished or started while the + * update feed was down stays stale until its parent is listed again. Call this once the feed has + * resubscribed, so every update after the list is delivered. + */ +export async function resyncProviderSubagents( + client: ProviderSubagentListClient, + serverId: string, +): Promise { + const serverPrefix = `${serverId}\0`; + const parentAgentIds = [...useProviderSubagentStore.getState().trackedParents] + .filter((key) => key.startsWith(serverPrefix)) + .map((key) => key.slice(serverPrefix.length, -1)); + await Promise.all( + parentAgentIds.map((parentAgentId) => + // A parent that was archived meanwhile has nothing left to catch up. + refreshProviderSubagents(client, serverId, parentAgentId).catch(() => undefined), + ), + ); +} + const EMPTY_TIMELINE: ProviderSubagentTimelineState = { tail: [], head: [], @@ -147,6 +172,14 @@ export const useProviderSubagentStore = create((set) => ( descriptors: new Map(), timelines: new Map(), hiddenFromTrack: new Set(), + trackedParents: new Set(), + trackParent(serverId, parentAgentId) { + set((state) => { + const prefix = parentPrefix(serverId, parentAgentId); + if (state.trackedParents.has(prefix)) return state; + return { trackedParents: new Set(state.trackedParents).add(prefix) }; + }); + }, hideFromTrack(serverId, parentAgentId, subagentIds) { set((state) => { const hiddenFromTrack = new Set(state.hiddenFromTrack); diff --git a/packages/website/e2e/plugins.spec.ts b/packages/website/e2e/plugins.spec.ts index 015d362338..26c0d9619a 100644 --- a/packages/website/e2e/plugins.spec.ts +++ b/packages/website/e2e/plugins.spec.ts @@ -388,19 +388,18 @@ async function expectPageMetadata(page: Page, title: string, path: string) { ); } -test("keeps the directory unlinked until the coordinated announcement", async ({ page }) => { +test("links the directory from the site navigation", async ({ page }) => { await page.goto("/"); await expect( - page.getByRole("banner").getByRole("link", { name: "Plugins", exact: true }), - ).toHaveCount(0); + page.getByRole("navigation").getByRole("link", { name: "Plugins", exact: true }), + ).toHaveAttribute("href", "/plugins"); await expect( page.getByRole("contentinfo").getByRole("link", { name: "Plugins", exact: true }), - ).toHaveCount(0); + ).toHaveAttribute("href", "/plugins"); await expect(page.getByRole("link", { name: "Browse plugins" })).toHaveAttribute( "href", - "https://paseo.cafe", + "/plugins", ); - await expect(page.locator('a[href="/plugins"]')).toHaveCount(0); const response = await page.goto("/plugins"); expect(response?.status()).toBe(200); await expect(page.getByRole("heading", { level: 1, name: /^Plugins/ })).toBeVisible();