From b37ab3e158d65028a06f65b3092ef3f53efcc92c Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 4 Nov 2026 08:04:54 +0100 Subject: [PATCH] fix(plugins): useComposers holds the composer it names, and a composer lives as long as its agent PluginComposersSource.subscribe watched the stores directly and never the composer model, so a component using useComposers did not keep the model's catalog observer alive and reconnect repair skipped it. subscribe now takes the selector and holds the model it names, following focus. Dropping a composer when it went idle left a command a plugin had saved calling a dead model after the tab reopened. A composer is now dropped when its agent leaves the session and nothing watches it or has one mounted. --- docs/plugins.md | 8 +- .../src/composer/agent-composer/index.test.ts | 24 +++--- .../app/src/composer/agent-composer/index.ts | 46 ++++++------ .../src/plugins/composer-items/source.test.ts | 75 ++++++++++++++++++- .../app/src/plugins/composer-items/source.ts | 41 +++++++--- packages/plugin/src/client/composer.tsx | 13 +++- 6 files changed, 148 insertions(+), 59 deletions(-) diff --git a/docs/plugins.md b/docs/plugins.md index 13ee4e711c..5fa8f9abb9 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -679,9 +679,11 @@ implementations provide it: picker shows. The model loads the catalog itself and resolves with no models until it lands, as the picker does, so a read never differs from what the picker shows. While anything watches the model it holds a `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. The - model is shared while a watcher or a mounted composer holds it and dropped when the last one - lets go. The mounted + 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 `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 dd67b7af62..038356d9e3 100644 --- a/packages/app/src/composer/agent-composer/index.test.ts +++ b/packages/app/src/composer/agent-composer/index.test.ts @@ -374,23 +374,17 @@ describe("agent composer", () => { }, ); - it("is released when nothing watches it and no composer is mounted, and shared while something does", () => { + it("is shared while its agent is in the session, and released once the agent is gone and nothing holds it", () => { openSession(agent()); - const held = agentComposer(serverId, agentId); - const stop = held.subscribe(() => {}); - expect(agentComposer(serverId, agentId)).toBe(held); - stop(); + const shared = agentComposer(serverId, agentId); + expect(agentComposer(serverId, agentId)).toBe(shared); - // A tab that remounts holds the composer it rendered with, which is shared again. - const stopAgain = held.subscribe(() => {}); - expect(agentComposer(serverId, agentId)).toBe(held); - stopAgain(); - expect(agentComposer(serverId, agentId)).not.toBe(held); - - const mounted = agentComposer(serverId, agentId); - const detach = mounted.attachInput({ isLive: () => true, setText() {}, insertText() {} }); - expect(agentComposer(serverId, agentId)).toBe(mounted); + // 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)).not.toBe(mounted); + useSessionStore.getState().setAgents(serverId, new Map()); + expect(agentComposer(serverId, agentId)).not.toBe(shared); }); }); diff --git a/packages/app/src/composer/agent-composer/index.ts b/packages/app/src/composer/agent-composer/index.ts index 0c7d7d08b3..7b90c28c58 100644 --- a/packages/app/src/composer/agent-composer/index.ts +++ b/packages/app/src/composer/agent-composer/index.ts @@ -39,10 +39,6 @@ export interface AgentComposer extends ComposerModel { interface AgentComposerDeps { queryClient: QueryClient; cache?: ProviderSnapshotCache; - /** Something watches the composer or a composer is mounted. */ - onHeld?: () => void; - /** Nothing watches the composer and no composer is mounted, so it can be dropped. */ - onIdle?: () => void; } type Catalog = { status: "loading" } | { status: "ready"; entries: ProviderSnapshotEntry[] | null }; @@ -50,11 +46,16 @@ type Catalog = { status: "loading" } | { status: "ready"; entries: ProviderSnaps 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. */ + isReleasable(): boolean; +} + export function createAgentComposer( serverId: string, agentId: string, deps: AgentComposerDeps, -): AgentComposer { +): ManagedAgentComposer { const draftKey = buildDraftStoreKey({ serverId, agentId }); const listeners = new Set<() => void>(); // While something watches, the catalog has an observer: it fetches the catalog, and it is what @@ -189,15 +190,11 @@ export function createAgentComposer( return last.state; } - function releaseWhenIdle() { - if (listeners.size === 0 && mounted === null) deps.onIdle?.(); - } - return { + isReleasable: () => listeners.size === 0 && mounted === null && readAgent() === null, getState, subscribe(listener) { listeners.add(listener); - deps.onHeld?.(); observeCatalog(); getState(); const stops = [ @@ -211,15 +208,12 @@ export function createAgentComposer( listeners.delete(listener); for (const stop of stops) stop(); observeCatalog(); - releaseWhenIdle(); }; }, attachInput(input) { mounted = input; - deps.onHeld?.(); return () => { if (mounted === input) mounted = null; - releaseWhenIdle(); }; }, setPrompt(text) { @@ -267,24 +261,26 @@ function sameCatalog(left: Catalog, right: Catalog): boolean { ); } -const composers = new Map(); +const composers = new Map(); +let stopReleasing: (() => void) | null = null; -/** The composer of a running agent; the same one for every caller while something holds it. */ +/** + * 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. + */ export function agentComposer(serverId: string, agentId: string): AgentComposer { + stopReleasing ??= useSessionStore.subscribe(releaseComposersOfRemovedAgents); const key = `${serverId}:${agentId}`; let composer = composers.get(key); if (!composer) { - composer = createAgentComposer(serverId, agentId, { - queryClient: singletonQueryClient, - // A holder that outlives a release, such as a remounting tab, puts its composer back. - onHeld: () => { - if (composer && !composers.has(key)) composers.set(key, composer); - }, - onIdle: () => { - if (composers.get(key) === composer) composers.delete(key); - }, - }); + composer = createAgentComposer(serverId, agentId, { queryClient: singletonQueryClient }); composers.set(key, composer); } return composer; } + +function releaseComposersOfRemovedAgents() { + 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 c6e1ab1cd1..3d3dc30857 100644 --- a/packages/app/src/plugins/composer-items/source.test.ts +++ b/packages/app/src/plugins/composer-items/source.test.ts @@ -4,7 +4,9 @@ import { PluginTranslationRuntime } from "@getpaseo/plugin/client/host"; import { QueryClient } from "@tanstack/react-query"; import { z } from "zod"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { providersSnapshotQueryRoot } from "@/data/providers-snapshot"; import { queryClient } from "@/data/query-client"; +import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types"; import { useDraftStore } from "@/stores/draft-store"; import { useSessionStore, type Agent, type WorkspaceDescriptor } from "@/stores/session-store"; import { createPluginClientStateSource } from "../client-state/source"; @@ -122,7 +124,7 @@ describe("a plugin's composers by reference", () => { it("follows the agent the Command Center hands out, and is empty when there is none", () => { const source = pluginComposersSource(plugin()); let notified = 0; - const stop = source.subscribe(() => { + const stop = source.subscribe({ focused: true }, () => { notified += 1; }); expect(source.resolve({ focused: true })).toEqual([]); @@ -145,7 +147,7 @@ describe("a plugin's composers by reference", () => { it("is not told when another component creates a query while it renders", () => { const source = pluginComposersSource(plugin()); let notified = 0; - const stop = source.subscribe(() => { + const stop = source.subscribe({ agentId: "agent-1" }, () => { notified += 1; }); // `useQuery` does this during a render, and a subscriber that updates then makes React warn @@ -166,6 +168,75 @@ describe("a plugin's composers by reference", () => { }); }); +describe("a plugin component watching a composer by reference", () => { + const catalog = (defaultThinkingOptionId: string): ProviderSnapshotEntry[] => [ + { + provider: "claude", + status: "ready", + enabled: true, + models: [ + { + provider: "claude", + id: "opus", + label: "Opus", + isDefault: true, + defaultThinkingOptionId, + thinkingOptions: [ + { id: "low", label: "Low" }, + { id: "high", label: "High" }, + ], + }, + ], + }, + ]; + + afterEach(() => { + queryClient.clear(); + }); + + async function until(read: () => unknown, expected: unknown) { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (Object.is(read(), expected)) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + expect(read()).toBe(expected); + } + + it("sees the catalog the host changed while away once the host reconnects, with no picker open", async () => { + let hostDefault = "high"; + const client = { + async getProvidersSnapshot() { + return { + entries: catalog(hostDefault), + 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")]])); + + const source = pluginComposersSource(plugin()); + const stop = source.subscribe({ agentId: "agent-1" }, () => {}); + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "high"); + + // Reconnecting invalidates the provider catalogs, which refetches the ones something watches. + hostDefault = "low"; + await queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) }); + + await until(() => source.resolve({ agentId: "agent-1" })[0]?.agent.thinking, "low"); + stop(); + }); +}); + describe("a Command Center item reaching an agent's composer by id", () => { // The real client-state source reads these stores. A lookup that came back empty would make the // Command Center return before calling the item, and nothing would be reported. diff --git a/packages/app/src/plugins/composer-items/source.ts b/packages/app/src/plugins/composer-items/source.ts index 233c794baa..12c3d8853d 100644 --- a/packages/app/src/plugins/composer-items/source.ts +++ b/packages/app/src/plugins/composer-items/source.ts @@ -2,8 +2,6 @@ import type { PluginComposer, PluginComposerSelector } from "@getpaseo/plugin/cl import type { PluginComposersSource } from "@getpaseo/plugin/client/host"; import { agentComposer } from "@/composer/agent-composer"; import type { ComposerState } from "@/composer/model"; -import { queryClient } from "@/data/query-client"; -import { useDraftStore } from "@/stores/draft-store"; import { useSessionStore } from "@/stores/session-store"; import { getFocusedAgent, subscribeFocusedAgent } from "../focused-agent"; import type { InstalledPlugin } from "../types"; @@ -43,25 +41,44 @@ function createSource(plugin: InstalledPlugin): PluginComposersSource { return focused?.serverId === serverId ? focused.agentId : null; } + /** The running agent a selector names now, or null. */ + function namedAgentId(selector: PluginComposerSelector): string | null { + const requested = "agentId" in selector ? selector.agentId : focusedAgentId(); + return requested === null ? null : runningAgentId(requested); + } + return { - subscribe(listener) { + subscribe(selector, listener) { + // Watching a composer is what keeps its model catalog current, so the subscription holds + // the model the selector names now and moves with it as focus and agents change. + let held: { agentId: string; stop: () => void } | null = null; + function hold() { + const agentId = namedAgentId(selector); + if (held?.agentId === agentId) return; + held?.stop(); + held = agentId + ? { agentId, stop: agentComposer(serverId, agentId).subscribe(listener) } + : null; + } + hold(); const stops = [ - useSessionStore.subscribe(listener), - useDraftStore.subscribe(listener), - subscribeFocusedAgent(listener), - // The model catalog arrives as an update. A query another component creates while it - // renders is announced too, and a subscriber that updates then makes React warn. - queryClient.getQueryCache().subscribe((event) => { - if (event.type === "updated") listener(); + useSessionStore.subscribe(() => { + hold(); + listener(); + }), + subscribeFocusedAgent(() => { + hold(); + listener(); }), ]; return () => { for (const stop of stops) stop(); + held?.stop(); + held = null; }; }, resolve(selector: PluginComposerSelector) { - const requested = "agentId" in selector ? selector.agentId : focusedAgentId(); - const agentId = requested === null ? null : runningAgentId(requested); + const agentId = namedAgentId(selector); if (agentId === null) return NO_COMPOSERS; const model = agentComposer(serverId, agentId); const state = model.getState(); diff --git a/packages/plugin/src/client/composer.tsx b/packages/plugin/src/client/composer.tsx index cc9b415f1b..04f0cec142 100644 --- a/packages/plugin/src/client/composer.tsx +++ b/packages/plugin/src/client/composer.tsx @@ -1,6 +1,7 @@ import { createContext, useContext, + useMemo, useSyncExternalStore, type ComponentType, type ReactNode, @@ -91,7 +92,8 @@ export type PluginComposerSelector = { agentId: string } | { focused: true }; /** What `useComposers` and `composers(selector)` read. Paseo implements it per installed plugin. */ export interface PluginComposersSource { - subscribe(listener: () => void): () => void; + /** Tells `listener` when the composers `selector` names change, and keeps them live meanwhile. */ + subscribe(selector: PluginComposerSelector, listener: () => void): () => void; /** The composers `selector` names now: one for a known running agent, none otherwise. */ resolve(selector: PluginComposerSelector): readonly PluginComposer[]; } @@ -121,8 +123,15 @@ export function usePluginComposersSource(): PluginComposersSource | null { export function useComposers(selector: PluginComposerSelector): readonly PluginComposer[] { const source = useContext(PluginComposersContext); if (!source) throw new Error("useComposers must run inside a plugin component"); + // The selector is usually an inline literal; its content is what the subscription depends on. + const selection = "agentId" in selector ? selector.agentId : null; + const subscribe = useMemo( + () => (listener: () => void) => + source.subscribe(selection === null ? { focused: true } : { agentId: selection }, listener), + [source, selection], + ); return useSyncExternalStore( - source.subscribe, + subscribe, () => source.resolve(selector), () => source.resolve(selector), );