puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-04 15:34:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugins): a running agent's composer validates and refreshes against the agent's own catalog

Browse Source 
setAgentModel and setAgentThinkingOption checked the provider's global catalog,
while the picker reads the catalog of the agent's directory, so a model only
the project lists was rejected. The validation input now carries the directory.

The composer read the catalog once with no observer, so reconnect repair, which
refetches only active queries, never refreshed it and a failed load never
retried. While anything watches, the composer now observes the catalog query.

The shared composer map kept every agent's composer for the session. It is now
dropped when nothing watches it and no composer is mounted.
This commit is contained in:
Mohamed Boudra
2026-11-04 08:10:29 +01:00
parent dad466d696
commit e59183f2de
8 changed files with 230 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/composer/agent-composer/index.test.ts
+79 -4
View File
@@ -7,7 +7,8 @@ import { filterSelectableModels } from "@/provider-selection/model-catalog";
import { buildDraftStoreKey } from "@/stores/draft-keys";
import { useDraftStore } from "@/stores/draft-store";
import { useSessionStore, type Agent } from "@/stores/session-store";
import { createAgentComposer } from "./index";
import { providersSnapshotQueryRoot } from "@/data/providers-snapshot";
import { agentComposer, createAgentComposer } from "./index";
const serverId = "host-1";
const agentId = "agent-1";
@@ -78,7 +79,12 @@ function fakeClient() {
const calls: string[] = [];
const client = {
async getProvidersSnapshot() {
return { entries, generatedAt: "2026-11-04T10:00:00.000Z", requestId: "snapshot" };
if (hostCatalog.failing) throw new Error("The host is busy.");
return {
entries: hostCatalog.entries,
generatedAt: "2026-11-04T10:00:00.000Z",
requestId: "snapshot",
};
},
async setAgentModel(id: string, model: string | null) {
calls.push(`model:${id}:${model}`);
@@ -102,6 +108,12 @@ function fakeClient() {
return { client: client as unknown as DaemonClient, calls };
}
/** What the host answers the next catalog request with. */
const hostCatalog: { entries: ProviderSnapshotEntry[]; failing: boolean } = {
entries,
failing: false,
};
function openSession(running: Agent) {
const { client, calls } = fakeClient();
const store = useSessionStore.getState();
@@ -124,11 +136,20 @@ async function until(read: () => unknown, expected: unknown) {
expect(read()).toBe(expected);
}
let queryClient: QueryClient;
function runtimeModel(model: string): Agent["runtimeInfo"] {
return { provider: "claude", sessionId: null, model };
}
function composer() {
return createAgentComposer(serverId, agentId, { queryClient: new QueryClient() });
return createAgentComposer(serverId, agentId, { queryClient });
}
beforeEach(() => {
queryClient = new QueryClient();
hostCatalog.entries = entries;
hostCatalog.failing = false;
useDraftStore.setState({ drafts: {}, textReplacementRequestByDraftKey: {} });
});
@@ -148,7 +169,6 @@ describe("agent composer", () => {
[other.id, other],
]),
);
const queryClient = new QueryClient();
const watched = createAgentComposer(serverId, running.id, { queryClient });
const reader = createAgentComposer(serverId, other.id, { queryClient });
let notifications = 0;
@@ -281,6 +301,41 @@ describe("agent composer", () => {
);
});
it("takes the catalog the host changed while the picker was not open, once the host reconnects", async () => {
openSession(agent({ runtimeInfo: runtimeModel("opus") }));
const model = composer();
const stop = model.subscribe(() => {});
await until(() => model.getState().thinking, "high");
// The host changes the catalog while the client is away. Reconnecting invalidates the provider
// catalogs, which refetches the ones something is watching.
hostCatalog.entries = [
{
...entries[0],
models: [{ ...entries[0].models![0], defaultThinkingOptionId: "low" }],
},
] as ProviderSnapshotEntry[];
await queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) });
await until(() => model.getState().thinking, "low");
stop();
});
it("asks for the catalog again after a failed load once the host reconnects", async () => {
hostCatalog.failing = true;
openSession(agent({ runtimeInfo: runtimeModel("opus") }));
const model = composer();
const stop = model.subscribe(() => {});
await new Promise((resolve) => setTimeout(resolve, 100));
expect(model.getState().thinking).toBeNull();
hostCatalog.failing = false;
await queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) });
await until(() => model.getState().thinking, "high");
stop();
});
function pickerShows(running: Agent, catalog: ProviderSnapshotEntry[] | null) {
const picker = resolveAgentModelSelection({
models: filterSelectableModels(catalog?.[0]?.models ?? null),
@@ -318,4 +373,24 @@ describe("agent composer", () => {
stop();
},
);
it("is released when nothing watches it and no composer is mounted, and shared while something does", () => {
openSession(agent());
const held = agentComposer(serverId, agentId);
const stop = held.subscribe(() => {});
expect(agentComposer(serverId, agentId)).toBe(held);
stop();
// 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);
detach();
expect(agentComposer(serverId, agentId)).not.toBe(mounted);
});
});
packages/app/src/composer/agent-composer/index.ts
+55 -28
View File
@@ -1,4 +1,4 @@
import type { QueryClient } from "@tanstack/react-query";
import { QueryObserver, type QueryClient } from "@tanstack/react-query";
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
import { providersSnapshotQuery } from "@/data/providers-snapshot";
import { replicaQueryOptions } from "@/data/query";
@@ -39,6 +39,10 @@ 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 };
@@ -53,8 +57,9 @@ export function createAgentComposer(
): AgentComposer {
const draftKey = buildDraftStoreKey({ serverId, agentId });
const listeners = new Set<() => void>();
const catalogFailures = new Set<string>();
const catalogRequests = new Set<string>();
// 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;
let mounted: MountedComposerInput | null = null;
let last: {
agent: Agent | null;
@@ -90,15 +95,21 @@ export function createAgentComposer(
for (const listener of listeners) listener();
}
async function loadCatalog(query: ReturnType<typeof catalogQuery>, key: string) {
try {
await deps.queryClient.ensureQueryData(query);
} catch {
catalogFailures.add(key);
} finally {
catalogRequests.delete(key);
notify();
}
function observeCatalog() {
const agent = listeners.size > 0 ? readAgent() : null;
const session = useSessionStore.getState().sessions[serverId];
const cwd =
agent && session?.client && session.serverInfo?.features?.providersSnapshot === true
? agent.cwd
: null;
if ((observing?.cwd ?? null) === cwd) return;
observing?.stop();
observing = null;
if (cwd === null) return;
observing = {
cwd,
stop: new QueryObserver(deps.queryClient, catalogQuery(cwd)).subscribe(notify),
};
}
function readCatalog(agent: Agent): Catalog {
@@ -112,15 +123,14 @@ export function createAgentComposer(
query.queryKey,
);
if (data) return { status: "ready", entries: data.entries };
const key = JSON.stringify(query.queryKey);
if (catalogFailures.has(key)) return { status: "ready", entries: null };
if (!session.client) return { status: "loading" };
if (!catalogRequests.has(key)) {
catalogRequests.add(key);
// Creating the query tells every query-cache subscriber. A read can run in another
// component's render, so the load starts after it.
queueMicrotask(() => void loadCatalog(query, key));
if (deps.queryClient.getQueryState(query.queryKey)?.status === "error") {
return { status: "ready", entries: null };
}
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));
return { status: "loading" };
}
@@ -179,29 +189,37 @@ export function createAgentComposer(
return last.state;
}
function releaseWhenIdle() {
if (listeners.size === 0 && mounted === null) deps.onIdle?.();
}
return {
getState,
subscribe(listener) {
listeners.add(listener);
deps.onHeld?.();
observeCatalog();
getState();
const stops = [
useSessionStore.subscribe(notify),
useDraftStore.subscribe(notify),
// Only a query's update changes the catalog. A query another component creates while it
// renders is announced too, and a subscriber that updates then makes React warn.
deps.queryClient.getQueryCache().subscribe((event) => {
if (event.type === "updated") notify();
useSessionStore.subscribe(() => {
observeCatalog();
notify();
}),
useDraftStore.subscribe(notify),
];
return () => {
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) {
@@ -251,12 +269,21 @@ function sameCatalog(left: Catalog, right: Catalog): boolean {
const composers = new Map<string, AgentComposer>();
/** The composer of a running agent; the same one for every caller. */
/** The composer of a running agent; the same one for every caller while something holds it. */
export function agentComposer(serverId: string, agentId: string): AgentComposer {
const key = `${serverId}:${agentId}`;
let composer = composers.get(key);
if (!composer) {
composer = createAgentComposer(serverId, agentId, { queryClient: singletonQueryClient });
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);
},
});
composers.set(key, composer);
}
return composer;