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 = {