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.
This commit is contained in:
Mohamed Boudra
2026-11-04 08:10:29 +01:00
parent e59183f2de
commit b37ab3e158
6 changed files with 148 additions and 59 deletions
+5 -3
View File
@@ -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
@@ -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);
});
});
@@ -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<string, AgentComposer>();
const composers = new Map<string, ManagedAgentComposer>();
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);
}
}
@@ -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.
@@ -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();
+11 -2
View File
@@ -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),
);