From d4b31cedb105e9ba7a927a39b63534ebaaa8182a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 4 Nov 2026 08:28:12 +0100 Subject: [PATCH] fix(plugins): a composer refreshes a stale catalog on resumed demand, shares its store watches, and is released with its last holder Watching again, or reading once, after the catalog was invalidated or failed while nothing watched now starts one deduplicated refresh; the observer refetches on mount. Reconnect repair only refetches watched queries, so an unwatched catalog stayed stale. Each subscribe installed its own session and draft store callbacks, so N watchers received N squared notifications, including for other agents' drafts. One pair of subscriptions now serves all listeners, and the draft callback ignores drafts of other agents. A composer whose agent was removed while it was held was only released on the next session change. The last holder letting go now checks too, and an archived agent's idle composer is releasable. --- docs/plugins.md | 10 ++- .../src/composer/agent-composer/index.test.ts | 60 ++++++++++++-- .../app/src/composer/agent-composer/index.ts | 83 +++++++++++++------ .../src/plugins/composer-items/source.test.ts | 67 +++++++++++++++ 4 files changed, 184 insertions(+), 36 deletions(-) diff --git a/docs/plugins.md b/docs/plugins.md index 5fa8f9abb9..4a045f29df 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -681,9 +681,13 @@ implementations provide it: `QueryObserver` on the agent's catalog: that is what lets reconnect repair, which refetches only active queries, refresh a catalog nobody has a picker open for, and retry one that failed. `PluginComposersSource.subscribe(selector, listener)` is what holds it for `useComposers`, so - the subscription must go through the model's own `subscribe`. The model is shared for as long as - its agent is in the session, so a command a plugin saved keeps reaching the composer a later tab - mounts; it is dropped once the agent is gone and nothing watches it or has one mounted. The mounted + the subscription must go through the model's own `subscribe`. Watching again after the catalog + was invalidated or failed refetches it, and a read with no watcher starts one deduplicated + refresh and shows the cached catalog until it lands. The model keeps one pair of store + subscriptions however many components watch it. It is shared for as long as its agent is in the + session and not archived, so a command a plugin saved keeps reaching the composer a later tab + mounts; it is dropped when the agent is gone or archived and nothing watches it or has one + mounted, checked on session changes and when the last holder lets go. The mounted `AgentControls` plugs its input in with `attachInput` so `insertText` lands at the cursor. Without a live input the model writes the stored prompt with `replaceDraftText`, which also asks a mounted input to show it. A workspace the user left stays mounted, display none, with no diff --git a/packages/app/src/composer/agent-composer/index.test.ts b/packages/app/src/composer/agent-composer/index.test.ts index 038356d9e3..b668893df8 100644 --- a/packages/app/src/composer/agent-composer/index.test.ts +++ b/packages/app/src/composer/agent-composer/index.test.ts @@ -336,6 +336,27 @@ describe("agent composer", () => { stop(); }); + it("watches the stores once however many components watch the composer, and only for its own draft", () => { + openSession(agent()); + const model = composer(); + const calls = [0, 0, 0]; + const stops: Array<() => void> = []; + for (const index of calls.keys()) { + stops.push(model.subscribe(() => void (calls[index] += 1))); + } + const before = [...calls]; + model.insertText("typed"); + expect(calls.map((count, index) => count - before[index])).toEqual([1, 1, 1]); + + const unrelated = [...calls]; + useDraftStore.getState().replaceDraftText({ + draftKey: buildDraftStoreKey({ serverId, agentId: "agent-2" }), + text: "elsewhere", + }); + expect(calls).toEqual(unrelated); + for (const stop of stops) stop(); + }); + function pickerShows(running: Agent, catalog: ProviderSnapshotEntry[] | null) { const picker = resolveAgentModelSelection({ models: filterSelectableModels(catalog?.[0]?.models ?? null), @@ -374,17 +395,42 @@ describe("agent composer", () => { }, ); - it("is shared while its agent is in the session, and released once the agent is gone and nothing holds it", () => { + it("is shared while its agent is in the session, across closing and reopening its tab", () => { openSession(agent()); const shared = agentComposer(serverId, agentId); - expect(agentComposer(serverId, agentId)).toBe(shared); - - // A mounted composer or a watcher keeps it, even if the agent is briefly missing. const detach = shared.attachInput({ isLive: () => true, setText() {}, insertText() {} }); - useSessionStore.getState().setAgents(serverId, new Map()); - expect(agentComposer(serverId, agentId)).toBe(shared); detach(); + expect(agentComposer(serverId, agentId)).toBe(shared); + const stop = shared.subscribe(() => {}); + stop(); + expect(agentComposer(serverId, agentId)).toBe(shared); + }); + + it("is released when the last holder lets go of a composer whose agent was removed meanwhile", () => { + openSession(agent()); + const mounted = agentComposer(serverId, agentId); + const detach = mounted.attachInput({ isLive: () => true, setText() {}, insertText() {} }); useSessionStore.getState().setAgents(serverId, new Map()); - expect(agentComposer(serverId, agentId)).not.toBe(shared); + // Held, so the session change did not release it, and nothing changes in the session after. + expect(agentComposer(serverId, agentId)).toBe(mounted); + detach(); + expect(agentComposer(serverId, agentId)).not.toBe(mounted); + + useSessionStore.getState().setAgents(serverId, new Map([[agentId, agent()]])); + const watched = agentComposer(serverId, agentId); + const stop = watched.subscribe(() => {}); + useSessionStore.getState().clearSession(serverId); + expect(agentComposer(serverId, agentId)).toBe(watched); + stop(); + expect(agentComposer(serverId, agentId)).not.toBe(watched); + }); + + it("is released when its agent is archived and nothing holds it", () => { + openSession(agent()); + const idle = agentComposer(serverId, agentId); + useSessionStore + .getState() + .setAgents(serverId, new Map([[agentId, agent({ archivedAt: new Date() })]])); + expect(agentComposer(serverId, agentId)).not.toBe(idle); }); }); diff --git a/packages/app/src/composer/agent-composer/index.ts b/packages/app/src/composer/agent-composer/index.ts index 7b90c28c58..e7aaf6e833 100644 --- a/packages/app/src/composer/agent-composer/index.ts +++ b/packages/app/src/composer/agent-composer/index.ts @@ -39,6 +39,8 @@ export interface AgentComposer extends ComposerModel { interface AgentComposerDeps { queryClient: QueryClient; cache?: ProviderSnapshotCache; + /** The last watcher or mounted composer let go, so the composer may be releasable. */ + onUnheld?: () => void; } type Catalog = { status: "loading" } | { status: "ready"; entries: ProviderSnapshotEntry[] | null }; @@ -47,7 +49,7 @@ const NO_VALUES = Object.freeze({}); const NO_ATTACHMENTS: readonly UserComposerAttachment[] = Object.freeze([]); interface ManagedAgentComposer extends AgentComposer { - /** The agent left the session and nothing watches the composer or has one mounted. */ + /** The agent is gone or archived, and nothing watches the composer or has one mounted. */ isReleasable(): boolean; } @@ -61,6 +63,8 @@ export function createAgentComposer( // While something watches, the catalog has an observer: it fetches the catalog, and it is what // makes the catalog an active query that reconnect repair refetches, a failed one included. let observing: { cwd: string; stop: () => void } | null = null; + // One pair of store subscriptions serves every listener, from the first to the last. + let stopWatchingStores: (() => void) | null = null; let mounted: MountedComposerInput | null = null; let last: { agent: Agent | null; @@ -107,9 +111,23 @@ export function createAgentComposer( observing?.stop(); observing = null; if (cwd === null) return; - observing = { - cwd, - stop: new QueryObserver(deps.queryClient, catalogQuery(cwd)).subscribe(notify), + // Watching again after the catalog was invalidated or failed while nothing watched refreshes it. + const options = { ...catalogQuery(cwd), refetchOnMount: true }; + observing = { cwd, stop: new QueryObserver(deps.queryClient, options).subscribe(notify) }; + } + + function watchStores() { + const stops = [ + useSessionStore.subscribe(() => { + observeCatalog(); + notify(); + }), + useDraftStore.subscribe((state, previous) => { + if (state.drafts[draftKey] !== previous.drafts[draftKey]) notify(); + }), + ]; + return () => { + for (const stop of stops) stop(); }; } @@ -120,18 +138,19 @@ export function createAgentComposer( return { status: "ready", entries: null }; } const query = catalogQuery(agent.cwd); - const data = deps.queryClient.getQueryData<{ entries: ProviderSnapshotEntry[] }>( + const state = deps.queryClient.getQueryState<{ entries: ProviderSnapshotEntry[] }>( query.queryKey, ); - if (data) return { status: "ready", entries: data.entries }; - if (deps.queryClient.getQueryState(query.queryKey)?.status === "error") { - return { status: "ready", entries: null }; + // Nothing observes while nothing watches, so a read starts the one deduplicated load or + // refresh it needs and shows the cached catalog meanwhile. A read can run in another + // component's render and creating the query tells every query-cache subscriber, so it starts + // after the render. + const loads = state?.isInvalidated === true || (!state?.data && state?.status !== "error"); + if (loads && !observing && session.client) { + queueMicrotask(() => void deps.queryClient.prefetchQuery(query)); } - if (!session.client) return { status: "loading" }; - // Nothing watches, so nothing observes: a command reads once and the load starts after it. - // Creating the query tells every query-cache subscriber, and a read can run in another - // component's render. - if (!observing) queueMicrotask(() => void deps.queryClient.prefetchQuery(query)); + if (state?.data) return { status: "ready", entries: state.data.entries }; + if (state?.status === "error") return { status: "ready", entries: null }; return { status: "loading" }; } @@ -190,30 +209,37 @@ export function createAgentComposer( return last.state; } + function unheld() { + if (listeners.size === 0 && mounted === null) deps.onUnheld?.(); + } + return { - isReleasable: () => listeners.size === 0 && mounted === null && readAgent() === null, + isReleasable() { + if (listeners.size > 0 || mounted !== null) return false; + const agent = readAgent(); + return agent === null || agent.archivedAt != null; + }, getState, subscribe(listener) { listeners.add(listener); + stopWatchingStores ??= watchStores(); observeCatalog(); getState(); - const stops = [ - useSessionStore.subscribe(() => { - observeCatalog(); - notify(); - }), - useDraftStore.subscribe(notify), - ]; return () => { listeners.delete(listener); - for (const stop of stops) stop(); + if (listeners.size === 0) { + stopWatchingStores?.(); + stopWatchingStores = null; + } observeCatalog(); + unheld(); }; }, attachInput(input) { mounted = input; return () => { if (mounted === input) mounted = null; + unheld(); }; }, setPrompt(text) { @@ -266,20 +292,25 @@ let stopReleasing: (() => void) | null = null; /** * The composer of a running agent; the same one for every caller for as long as the agent is in the - * session, so a command a plugin saved keeps reaching the composer a later tab mounts. + * session and not archived, so a command a plugin saved keeps reaching the composer a later tab + * mounts. It is released when the agent is gone or archived and nothing holds it, checked when the + * session changes and when the last holder lets go. */ export function agentComposer(serverId: string, agentId: string): AgentComposer { - stopReleasing ??= useSessionStore.subscribe(releaseComposersOfRemovedAgents); + stopReleasing ??= useSessionStore.subscribe(releaseReleasableComposers); const key = `${serverId}:${agentId}`; let composer = composers.get(key); if (!composer) { - composer = createAgentComposer(serverId, agentId, { queryClient: singletonQueryClient }); + composer = createAgentComposer(serverId, agentId, { + queryClient: singletonQueryClient, + onUnheld: releaseReleasableComposers, + }); composers.set(key, composer); } return composer; } -function releaseComposersOfRemovedAgents() { +function releaseReleasableComposers() { for (const [key, composer] of composers) { if (composer.isReleasable()) composers.delete(key); } diff --git a/packages/app/src/plugins/composer-items/source.test.ts b/packages/app/src/plugins/composer-items/source.test.ts index 3d3dc30857..6531d7aef8 100644 --- a/packages/app/src/plugins/composer-items/source.test.ts +++ b/packages/app/src/plugins/composer-items/source.test.ts @@ -157,6 +157,19 @@ describe("a plugin's composers by reference", () => { stop(); }); + it("delivers a draft change once to each of several components watching the same agent", () => { + const source = pluginComposersSource(plugin()); + const delivered = [0, 0, 0]; + const stops: Array<() => void> = []; + for (const index of delivered.keys()) { + stops.push(source.subscribe({ agentId: "agent-1" }, () => void (delivered[index] += 1))); + } + const before = [...delivered]; + source.resolve({ agentId: "agent-1" })[0]?.insertText("typed"); + expect(delivered.map((count, index) => count - before[index])).toEqual([1, 1, 1]); + for (const stop of stops) stop(); + }); + it("returns the same array until a composer changes", () => { const source = pluginComposersSource(plugin()); const first = source.resolve({ agentId: "agent-1" }); @@ -202,6 +215,60 @@ describe("a plugin component watching a composer by reference", () => { expect(read()).toBe(expected); } + function hostWithDefaultThinking(initial: string) { + const host = { defaultThinking: initial }; + const client = { + async getProvidersSnapshot() { + return { + entries: catalog(host.defaultThinking), + generatedAt: "2026-11-04T10:00:00.000Z", + requestId: "snapshot", + }; + }, + } as unknown as DaemonClient; + const store = useSessionStore.getState(); + store.clearSession(serverId); + store.initializeSession(serverId, client); + store.updateSessionServerInfo(serverId, { + serverId, + hostname: null, + version: null, + features: { providersSnapshot: true }, + } as never); + store.setAgents(serverId, new Map([["agent-1", agent("agent-1")]])); + return host; + } + + it("refreshes a catalog that was invalidated while nothing watched, when watching resumes", async () => { + const host = hostWithDefaultThinking("high"); + const source = pluginComposersSource(plugin()); + const stopFirst = source.subscribe({ agentId: "agent-1" }, () => {}); + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "high"); + stopFirst(); + + // Reconnect repair only refetches what is watched, so this catalog is invalidated and kept. + host.defaultThinking = "low"; + await queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) }); + + const stopAgain = source.subscribe({ agentId: "agent-1" }, () => {}); + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "low"); + stopAgain(); + }); + + it("refreshes an invalidated catalog for a command that reads once", async () => { + const host = hostWithDefaultThinking("high"); + const source = pluginComposersSource(plugin()); + const stopFirst = source.subscribe({ agentId: "agent-1" }, () => {}); + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "high"); + stopFirst(); + host.defaultThinking = "low"; + await queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) }); + + // A command cannot wait, so this read still shows the old catalog and starts the refresh. + source.resolve({ agentId: "agent-1" }); + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "low"); + }); + it("sees the catalog the host changed while away once the host reconnects, with no picker open", async () => { let hostDefault = "high"; const client = {