refactor(core): replace plugin catalog push with per-request freshness

- Drop pluginCommands.subscribe and the plugins.commands.changed event. Hub
  runtime hosts no longer open a global event stream, so the per-listener
  event ID dedupe in NodeHubClient is removed with it.
- Reduce PluginCommandManager to a per-workspace catalog that fingerprints
  plugin sources and settings on every request, keeps healthy commands
  through a failed load, and retries that load after 30 seconds. File
  watchers, debounced source checks, backoff timers, and per-path recovery
  are removed.
- Release workspace catalogs that have been idle for 10 minutes so the
  long-lived Hub does not keep plugin workers for every workspace it served.
- CLI suggestions reload on mount and active session change; desktop
  composers re-list plugin and instruction commands when the slash menu
  opens. Client retry loops are removed.

Co-authored-by: Saoud Rizwan <saoudrizwan@users.noreply.github.com>
This commit is contained in:
Saoud Rizwan
2026-10-20 17:51:34 +00:00
co-authored by Saoud Rizwan
parent b757b75bca
commit 0c89932af8
30 changed files with 374 additions and 1552 deletions
@@ -197,7 +197,6 @@ describe("handleConnectorUserTurn", () => {
list: vi
.fn()
.mockResolvedValue({ status: "ready", commands: [{ name: "echo" }] }),
subscribe: vi.fn(),
},
});
await Promise.all(
@@ -144,7 +144,7 @@ function makeManager() {
});
return {
start,
pluginCommands: { list: vi.fn(), run: vi.fn(), subscribe: vi.fn() },
pluginCommands: { list: vi.fn(), run: vi.fn() },
stop: vi.fn(async () => {}),
send: vi.fn(),
getAccumulatedUsage: vi.fn(),
+2 -4
View File
@@ -230,13 +230,11 @@ export async function runInteractive(
}));
};
// Session commands come from the active session's plugins, so a session
// change can change the catalog without any workspace catalog event.
// change can change the catalog.
const activeSessionListeners = new Set<() => void>();
const subscribeAdditionalSlashCommands = async (listener: () => void) => {
const stop = (await ensurePluginHost()).subscribe(listener);
const subscribeAdditionalSlashCommands = (listener: () => void) => {
activeSessionListeners.add(listener);
return () => {
stop();
activeSessionListeners.delete(listener);
};
};
+42 -123
View File
@@ -34,142 +34,61 @@ function unmount() {
for (const cleanup of reactMock.cleanups.splice(0)) cleanup();
}
beforeEach(() => {
vi.useFakeTimers();
reactMock.setCommands.mockClear();
});
afterEach(() => {
unmount();
vi.useRealTimers();
});
describe("plugin slash command catalog lifecycle", () => {
it("retries a failed subscription and receives updates after recovery", async () => {
let listener = () => {};
it("reloads when notified and keeps the last catalog when a reload fails", async () => {
let notify = () => {};
const stop = vi.fn();
const load = vi.fn().mockResolvedValue(first);
const subscribe = vi
const load = vi
.fn()
.mockRejectedValueOnce(new Error("startup failed"))
.mockImplementation(async (refresh) => {
listener = refresh;
return stop;
});
useTestSlashCommands(load, subscribe);
await vi.advanceTimersByTimeAsync(0);
expect(subscribe).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1000);
expect(subscribe).toHaveBeenCalledTimes(2);
load.mockResolvedValue(next);
listener();
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next);
.mockResolvedValueOnce(first)
.mockRejectedValueOnce(new Error("hub unavailable"))
.mockResolvedValueOnce(next);
useTestSlashCommands(load, (listener) => {
notify = listener;
return stop;
});
await vi.waitFor(() =>
expect(reactMock.setCommands).toHaveBeenLastCalledWith(first),
);
notify();
await vi.waitFor(() => expect(load).toHaveBeenCalledTimes(2));
await Promise.resolve();
expect(reactMock.setCommands).toHaveBeenCalledTimes(1);
notify();
await vi.waitFor(() =>
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next),
);
unmount();
expect(stop).toHaveBeenCalledTimes(1);
});
it("retains the loaded catalog during a failed refresh and retries without an event", async () => {
let listener = () => {};
const load = vi.fn().mockResolvedValue(first);
useTestSlashCommands(load, async (refresh) => {
listener = refresh;
it("ignores an older load that resolves after a newer one", async () => {
let notify = () => {};
let resolveOlder!: (commands: typeof first) => void;
const load = vi
.fn()
.mockImplementationOnce(
() =>
new Promise((resolve) => {
resolveOlder = resolve;
}),
)
.mockResolvedValueOnce(next);
useTestSlashCommands(load, (listener) => {
notify = listener;
return () => {};
});
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(first);
const successfulUpdates = reactMock.setCommands.mock.calls.length;
load
.mockRejectedValueOnce(new Error("connection lost"))
.mockResolvedValue(next);
listener();
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenCalledTimes(successfulUpdates);
await vi.advanceTimersByTimeAsync(1000);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next);
});
it("backs off refresh retries and cancels retries on unmount", async () => {
const load = vi.fn().mockRejectedValue(new Error("offline"));
useTestSlashCommands(load);
await vi.advanceTimersByTimeAsync(0);
await vi.advanceTimersByTimeAsync(1000);
expect(load).toHaveBeenCalledTimes(2);
await vi.advanceTimersByTimeAsync(1999);
expect(load).toHaveBeenCalledTimes(2);
await vi.advanceTimersByTimeAsync(1);
expect(load).toHaveBeenCalledTimes(3);
unmount();
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(60_000);
expect(load).toHaveBeenCalledTimes(3);
});
it("cancels failed subscription retries on unmount", async () => {
const load = vi.fn().mockResolvedValue(first);
const subscribe = vi.fn().mockRejectedValue(new Error("offline"));
useTestSlashCommands(load, subscribe);
await vi.advanceTimersByTimeAsync(0);
unmount();
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(60_000);
expect(subscribe).toHaveBeenCalledTimes(1);
});
it("stops a subscription that resolves after unmount", async () => {
let resolve!: (stop: () => void) => void;
const pending = new Promise<() => void>((done) => {
resolve = done;
});
const stop = vi.fn();
useTestSlashCommands(vi.fn().mockResolvedValue(first), () => pending);
unmount();
resolve(stop);
await vi.advanceTimersByTimeAsync(0);
expect(stop).toHaveBeenCalledTimes(1);
expect(reactMock.setCommands).not.toHaveBeenCalled();
});
it("ignores an older refresh that resolves after the subscribed catalog", async () => {
let resolve!: (commands: typeof first) => void;
const pending = new Promise<typeof first>((done) => {
resolve = done;
});
const load = vi.fn().mockReturnValueOnce(pending).mockResolvedValue(next);
useTestSlashCommands(load, async () => () => {});
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next);
resolve(first);
await vi.advanceTimersByTimeAsync(0);
notify();
await vi.waitFor(() =>
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next),
);
resolveOlder(first);
await Promise.resolve();
expect(reactMock.setCommands).toHaveBeenCalledTimes(1);
});
it("publishes the initial catalog while the subscription refresh stalls", async () => {
let resolve!: (commands: typeof first) => void;
const initial = new Promise<typeof first>((done) => {
resolve = done;
});
const stalled = new Promise<typeof first>(() => {});
const load = vi
.fn()
.mockReturnValueOnce(initial)
.mockReturnValueOnce(stalled);
useTestSlashCommands(load, async () => () => {});
await vi.advanceTimersByTimeAsync(0);
expect(load).toHaveBeenCalledTimes(2);
resolve(first);
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(first);
});
it("keeps an initial success when the subscription refresh fails and retries", async () => {
let resolve!: (commands: typeof first) => void;
const initial = new Promise<typeof first>((done) => {
resolve = done;
});
const load = vi
.fn()
.mockReturnValueOnce(initial)
.mockRejectedValueOnce(new Error("refresh failed"))
.mockResolvedValue(next);
useTestSlashCommands(load, async () => () => {});
await vi.advanceTimersByTimeAsync(0);
resolve(first);
await vi.advanceTimersByTimeAsync(0);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(first);
await vi.advanceTimersByTimeAsync(1000);
expect(reactMock.setCommands).toHaveBeenLastCalledWith(next);
});
});
+10 -51
View File
@@ -30,61 +30,20 @@ export function useSlashCommands(input: {
}
let cancelled = false;
let requestId = 0;
let publishedRequestId = 0;
let refreshRetry: ReturnType<typeof setTimeout> | undefined;
let subscribeRetry: ReturnType<typeof setTimeout> | undefined;
let refreshDelay = 1000;
let subscribeDelay = 1000;
const refresh = async () => {
if (cancelled) return;
clearTimeout(refreshRetry);
const refresh = () => {
const id = ++requestId;
try {
const commands = await loadAdditionalSlashCommands();
if (!cancelled && id > publishedRequestId) {
// A newer pending/failed request must not discard a usable catalog.
publishedRequestId = id;
if (id === requestId) refreshDelay = 1000;
setAdditionalSlashCommands(commands);
}
} catch {
if (!cancelled && id === requestId) {
// Keep the last catalog visible while the runtime reconnects.
refreshRetry = setTimeout(() => {
void refresh();
}, refreshDelay);
refreshDelay = Math.min(refreshDelay * 2, 30_000);
}
}
void loadAdditionalSlashCommands()
.then((commands) => {
if (!cancelled && id === requestId)
setAdditionalSlashCommands(commands);
})
// Keep the last catalog; the next refresh tries again.
.catch(() => {});
};
void refresh();
let unsubscribe: (() => void) | undefined;
const subscribe = async () => {
if (cancelled || !subscribeAdditionalSlashCommands) return;
try {
const stop = await subscribeAdditionalSlashCommands(() => {
void refresh();
});
if (cancelled) stop();
else {
unsubscribe = stop;
// Catch any catalog change missed before subscription was established.
void refresh();
}
} catch {
if (!cancelled) {
subscribeRetry = setTimeout(() => {
void subscribe();
}, subscribeDelay);
subscribeDelay = Math.min(subscribeDelay * 2, 30_000);
}
}
};
void subscribe();
refresh();
const unsubscribe = subscribeAdditionalSlashCommands?.(refresh);
return () => {
cancelled = true;
clearTimeout(refreshRetry);
clearTimeout(subscribeRetry);
unsubscribe?.();
};
}, [loadAdditionalSlashCommands, subscribeAdditionalSlashCommands]);
+2 -3
View File
@@ -157,9 +157,8 @@ export interface TuiProps {
initialRepoStatus?: RepoStatus;
workflowSlashCommands?: InteractiveSlashCommand[];
loadAdditionalSlashCommands?: () => Promise<InteractiveSlashCommand[]>;
subscribeAdditionalSlashCommands?: (
listener: () => void,
) => Promise<() => void>;
/** Notifies when additional commands may have changed (the active session changed). */
subscribeAdditionalSlashCommands?: (listener: () => void) => () => void;
loadWelcomeLine?: () => Promise<string | undefined>;
loadClineAccount: () => Promise<ClineAccountSnapshot>;
loadIndividualSubscriptionPlans?: () => Promise<ClineSubscriptionPlan[]>;
@@ -130,7 +130,6 @@ describe("plugin chat commands", () => {
const commands: PluginCommandsApi = {
list: vi.fn().mockRejectedValue(new Error("hub unavailable")),
run: vi.fn().mockRejectedValue(new Error("must not execute")),
subscribe: vi.fn(),
};
const logger = { log: vi.fn(), debug: vi.fn() };
const { host } = await createWorkspaceChatCommandHost({
@@ -162,7 +161,6 @@ describe("plugin chat commands", () => {
error: "sibling failed",
}),
run: vi.fn(),
subscribe: vi.fn(),
};
const { host } = await createWorkspaceChatCommandHost({
commands,
@@ -192,7 +190,6 @@ describe("plugin chat commands", () => {
error: "sibling failed",
}),
run: vi.fn().mockRejectedValue(new Error("handler failed")),
subscribe: vi.fn(),
};
const { host } = await createWorkspaceChatCommandHost({
commands,
@@ -55,10 +55,5 @@ export async function createWorkspaceChatCommandHost(input: {
input.logger?.log(`Plugin commands unavailable: ${catalog.error}`);
return catalog.commands;
},
subscribe(listener: () => void) {
return commands.subscribe((catalog) => {
if (catalog.workspacePath === target().workspacePath) listener();
});
},
};
}
+1 -4
View File
@@ -936,10 +936,7 @@ export function handleHubLiveEvent(
});
return;
}
if (
event.event === "settings.changed" ||
event.event === "plugins.commands.changed"
) {
if (event.event === "settings.changed") {
sendEvent(ctx, event.event, event.payload ?? {});
return;
}
@@ -402,73 +402,39 @@ describe("ChatInputBar", () => {
expect(container.textContent).toContain("Publish workflow");
});
it("retries rejected plugin reads without an event and stops retrying after recovery", async () => {
vi.useFakeTimers();
it("re-reads plugin commands when the slash menu reopens after a failure", async () => {
let available = false;
const invoke = vi
.spyOn(desktopClient, "invoke")
.mockImplementation(async (name) => {
if (name !== "list_plugin_commands") return { runtimeCommands: [] };
if (!available) throw new Error("hub connection lost");
return {
workspacePath: "/workspace/cline",
status: "ready",
commands: [{ name: "restored", description: "Recovered command" }],
};
});
try {
await renderVoiceComposer({ prompt: "/rest", executionTarget: "local" });
expect(container.textContent).toContain("Plugin commands unavailable");
await act(async () => {
await vi.advanceTimersByTimeAsync(1000);
});
expect(
invoke.mock.calls.filter(([name]) => name === "list_plugin_commands"),
).toHaveLength(2);
available = true;
await act(async () => {
await vi.advanceTimersByTimeAsync(2000);
});
expect(container.textContent).toContain("Recovered command");
expect(container.textContent).not.toContain(
"Plugin commands unavailable",
);
await act(async () => {
await vi.advanceTimersByTimeAsync(30000);
});
expect(
invoke.mock.calls.filter(([name]) => name === "list_plugin_commands"),
).toHaveLength(3);
} finally {
vi.useRealTimers();
}
vi.spyOn(desktopClient, "invoke").mockImplementation(async (name) => {
if (name !== "list_plugin_commands") return { runtimeCommands: [] };
if (!available) throw new Error("hub connection lost");
return {
workspacePath: "/workspace/cline",
status: "ready",
commands: [{ name: "restored", description: "Recovered command" }],
};
});
await renderVoiceComposer({
prompt: "/rest",
promptVersion: 1,
executionTarget: "local",
});
expect(container.textContent).toContain("Plugin commands unavailable");
available = true;
await renderVoiceComposer({
prompt: "ordinary text",
promptVersion: 2,
executionTarget: "local",
});
await renderVoiceComposer({
prompt: "/rest",
promptVersion: 3,
executionTarget: "local",
});
expect(container.textContent).toContain("Recovered command");
expect(container.textContent).not.toContain("Plugin commands unavailable");
});
it("cancels failed plugin read retries when the composer unmounts", async () => {
vi.useFakeTimers();
const invoke = vi
.spyOn(desktopClient, "invoke")
.mockImplementation(async (name) => {
if (name === "list_plugin_commands")
throw new Error("hub connection lost");
return { runtimeCommands: [] };
});
try {
await renderVoiceComposer({ prompt: "/rest", executionTarget: "local" });
await act(async () => root.unmount());
root = createRoot(container);
await act(async () => {
await vi.advanceTimersByTimeAsync(30000);
});
expect(
invoke.mock.calls.filter(([name]) => name === "list_plugin_commands"),
).toHaveLength(1);
} finally {
vi.useRealTimers();
}
});
it("refreshes instruction files on reopen without resetting plugin or cached suggestions", async () => {
it("refreshes commands on reopen without resetting cached suggestions", async () => {
const updated = deferred<{
runtimeCommands: Array<{
name: string;
@@ -524,7 +490,7 @@ describe("ChatInputBar", () => {
expect(instructionRequests).toBe(2);
expect(
invoke.mock.calls.filter(([name]) => name === "list_plugin_commands"),
).toHaveLength(1);
).toHaveLength(2);
await act(async () =>
updated.resolve({
runtimeCommands: [
@@ -541,45 +507,6 @@ describe("ChatInputBar", () => {
expect(container.textContent).toContain("Plugin suggestion");
});
it("refreshes the open slash menu when the hub catalog recovers", async () => {
let available = false;
let changed: ((payload: unknown) => void) | undefined;
vi.spyOn(desktopClient, "subscribe").mockImplementation(
(name, listener) => {
if (name === "plugins.commands.changed") changed = listener;
return vi.fn();
},
);
vi.spyOn(desktopClient, "invoke").mockImplementation(async (name) => {
if (name === "list_plugin_commands")
return {
workspacePath: "/workspace/cline",
status: available ? "ready" : "error",
commands: available
? [{ name: "upload-history", description: "Upload history" }]
: [],
};
return { runtimeCommands: [] };
});
await renderVoiceComposer({ prompt: "/upload-", executionTarget: "local" });
expect(container.textContent).toContain("Plugin commands unavailable");
available = true;
await act(async () => {
changed?.({
environmentId: "other-host",
catalog: { workspacePath: "/workspace/cline" },
});
});
expect(container.textContent).not.toContain("Upload history");
await act(async () => {
changed?.({
environmentId: "local",
catalog: { workspacePath: "/workspace/cline" },
});
});
expect(container.textContent).toContain("Upload history");
});
it("prevents sending from a read-only session", async () => {
const onSend = vi.fn();
await renderVoiceComposer({ prompt: "Test", readOnly: true, onSend });
@@ -655,20 +655,29 @@ function ChatInputBarImpl({
environmentId: string;
response: UserInstructionConfigResponse;
}>();
const [pluginCommands, setPluginCommands] = useState<SlashCommand[]>([]);
const slashCommands = useMemo(
() => [
const [pluginResponse, setPluginResponse] = useState<{
workspaceRoot: string;
environmentId: string;
commands: SlashCommand[];
}>();
const slashCommands = useMemo(() => {
const current = <
T extends { workspaceRoot: string; environmentId: string },
>(
target: T | undefined,
) =>
target?.workspaceRoot === workspaceRoot &&
target.environmentId === environmentId
? target
: undefined;
return [
...BUILTIN_SLASH_COMMANDS,
...buildUserInstructionSlashCommands(
instructionResponse?.workspaceRoot === workspaceRoot &&
instructionResponse.environmentId === environmentId
? instructionResponse.response
: {},
pluginCommands,
current(instructionResponse)?.response ?? {},
current(pluginResponse)?.commands ?? [],
),
],
[instructionResponse, pluginCommands, workspaceRoot, environmentId],
);
];
}, [instructionResponse, pluginResponse, workspaceRoot, environmentId]);
const [pluginLoading, setPluginLoading] = useState(false);
const [instructionLoading, setInstructionLoading] = useState(false);
const [pluginError, setPluginError] = useState<string>();
@@ -1028,128 +1037,63 @@ function ChatInputBarImpl({
?.scrollIntoView({ block: "nearest" });
}, [mentionOpen, mentionSelectedIndex]);
// Runtime-owned plugin catalogs stay loaded across menu toggles.
// Refresh user and plugin commands whenever the slash menu opens, keeping
// the last results while fetching so suggestions do not flash away. Each
// request settles independently so one failure cannot hide the other.
useEffect(() => {
setPluginCommands([]);
setPluginError(undefined);
if (executionTarget !== "local") return;
if (!slashOpen) {
setInstructionLoading(false);
setPluginLoading(false);
return;
}
let cancelled = false;
let requestId = 0;
let retryCount = 0;
let retry: ReturnType<typeof setTimeout> | undefined;
setInstructionLoading(true);
setPluginLoading(true);
const refresh = async () => {
clearTimeout(retry);
const id = ++requestId;
try {
const catalog = await desktopClient.invoke<PluginCommandCatalog>(
"list_plugin_commands",
{
workspacePath: workspaceRoot,
environmentId,
sessionId: sessionId ?? undefined,
},
);
if (cancelled || id !== requestId) return;
retryCount = 0;
setPluginCommands(catalog.commands);
desktopClient
.invoke<UserInstructionConfigResponse>("list_user_instruction_configs", {
workspacePath: workspaceRoot,
environmentId,
})
.then((response) => {
if (cancelled) return;
setInstructionResponse({ workspaceRoot, environmentId, response });
setInstructionError(undefined);
})
.catch(() => {
if (!cancelled) setInstructionError("Instruction commands unavailable");
})
.finally(() => {
if (!cancelled) setInstructionLoading(false);
});
desktopClient
.invoke<PluginCommandCatalog>("list_plugin_commands", {
workspacePath: workspaceRoot,
environmentId,
sessionId: sessionId ?? undefined,
})
.then((catalog) => {
if (cancelled) return;
setPluginResponse({
workspaceRoot,
environmentId,
commands: catalog.commands,
});
setPluginError(
catalog.status === "error"
? "Plugin commands unavailable"
: undefined,
);
} catch {
if (!cancelled && id === requestId) {
setPluginError("Plugin commands unavailable");
retry = setTimeout(
() => void refresh(),
Math.min(1000 * 2 ** retryCount++, 30000),
);
}
} finally {
if (!cancelled && id === requestId) setPluginLoading(false);
}
};
const unsubscribe = desktopClient.subscribe(
"plugins.commands.changed",
(payload) => {
const event = payload as {
environmentId?: string;
catalog?: PluginCommandCatalog;
};
if (
event.environmentId === environmentId &&
event.catalog?.workspacePath === workspaceRoot
)
void refresh();
},
);
const unsubscribeSettings = desktopClient.subscribe(
"settings.changed",
(payload) => {
if (
(payload as { environmentId?: string }).environmentId ===
environmentId
)
void refresh();
},
);
void refresh();
})
.catch(() => {
if (!cancelled) setPluginError("Plugin commands unavailable");
})
.finally(() => {
if (!cancelled) setPluginLoading(false);
});
return () => {
cancelled = true;
clearTimeout(retry);
unsubscribe();
unsubscribeSettings();
};
}, [workspaceRoot, environmentId, sessionId, executionTarget]);
// Instruction files have no catalog-change event. Refresh them on open,
// retaining the last response while fetching so suggestions do not flash away.
useEffect(() => {
if (!slashOpen) {
setInstructionLoading(false);
return;
}
let cancelled = false;
let requestId = 0;
const refresh = async () => {
const id = ++requestId;
setInstructionLoading(true);
try {
const response =
await desktopClient.invoke<UserInstructionConfigResponse>(
"list_user_instruction_configs",
{
workspacePath: workspaceRoot,
environmentId,
},
);
if (cancelled || id !== requestId) return;
setInstructionResponse({ workspaceRoot, environmentId, response });
setInstructionError(undefined);
} catch {
if (!cancelled && id === requestId)
setInstructionError("Instruction commands unavailable");
} finally {
if (!cancelled && id === requestId) setInstructionLoading(false);
}
};
const unsubscribe = desktopClient.subscribe(
"settings.changed",
(payload) => {
if (
(payload as { environmentId?: string }).environmentId ===
environmentId
)
void refresh();
},
);
void refresh();
return () => {
cancelled = true;
unsubscribe();
};
}, [slashOpen, workspaceRoot, environmentId]);
}, [slashOpen, workspaceRoot, environmentId, sessionId]);
// Filtered slash commands based on the current query.
const filteredSlashCommands = useMemo(() => {
+23 -49
View File
@@ -799,58 +799,32 @@ runtime flags (including system CA trust) intact. Generic script sandboxes
still use plain Bun mode when reusing a compiled runtime.
Command ownership is independent of how the host packages or locates the plugin
bootstrap and SDK resources. Resource loading supplies the existing plugin
runtime; it does not introduce another command service or command payload.
`LocalRuntimeHost.pluginCommands` owns plugin command discovery and execution,
exposed to clients through Hub RPC (`plugins.commands.list` /
`plugins.commands.run`). The desktop sidecar, CLI, and connectors reuse their
existing runtime connections and do not load plugins themselves. Plugin command
execution can recover a connection before dispatch, but never resends a
dispatched action whose reply was lost.
`LocalRuntimeHost.pluginCommands` owns plugin command discovery and execution.
Hub clients deduplicate overlapping stream event IDs per listener before dispatch,
while retaining replay for newly subscribed listeners. Plugin command execution
can recover a connection before dispatch, but cannot replay a dispatched action
whose reply was lost.
Hub RPC (`plugins.commands.list` / `plugins.commands.run`) and
`plugins.commands.changed` expose the same service to every client. The desktop
sidecar, CLI, and connectors reuse their existing hub connections. Workspace
catalogs are initialized when a composer first requests them and refreshed by
runtime-owned plugin/settings watchers, including for non-Git folders. Watches
follow directory creation and replacement independently of source fingerprints,
so an empty `.cline` directory cannot hide later plugin installation events.
A debounced source check also reconciles writes whose native events coalesce
during watch registration, without restarting unchanged plugins. Failed
initialization retries in the background and publishes explicit error/recovery
state. Workspace catalogs use their own worker registry so catalog failures do not disable resident session instances. Partial retries reload only failed plugin paths and retain
healthy instances; source/settings changes reload the workspace catalog. Git
queries and individual UI mounts do not own plugin lifetimes. Catalog requests
also fingerprint the imported source graph, including helpers outside watched
directories, before serving commands. Only discoverable entries, their imported
source graph, and plugin settings participate in this fingerprint; setup logs and
runtime state files beside a plugin do not. Requests allow at most two
reconciliation passes and reject if sources keep changing, so a setup cannot
trap a request in a refresh loop. Partial recovery follows the original
plugin path precedence and releases overridden workers independently.
A workspace catalog owns a discovery sandbox; a resident session instead lists
and executes its own contribution registry, retaining the same plugin instance
and context as its hooks. Existing session registrations remain fixed for that
session, while changed workspace catalogs apply to new sessions. Reload and
disposal wait for command handlers before terminating their workers. Hub status and replacement checks count active command requests even without a chat participant; draining refuses new catalog and command requests.
A resident session lists and executes commands from its own contribution
registry, retaining the plugin instance and context shared with its hooks.
Session registrations keep their session lifetime. Without a resident session,
the runtime uses a per-workspace catalog. Each request fingerprints discoverable
plugin entries, their imported source graph, and plugin settings, and reloads
the catalog when that changes; setup logs and state files beside a plugin do
not. A failed load keeps healthy commands, reports `status: "error"`, and is
retried by a request after 30 seconds. Catalogs idle for 10 minutes release
their workers. Reload, eviction, and disposal wait for running handlers. Clients
re-list to stay current (for example when a slash menu opens or the active
session changes); there is no catalog push event.
The shared command executor records `task.tool_used` once per matched handler,
including failures, for both direct runtime and Hub calls. Resident execution
uses session/provider/model and agent identity; workspace execution uses an
action correlation ID without inventing a session. Catalog reads and unmatched
commands do not count as tool use. CLI interactive initialization shares one
pending core across catalog discovery and session startup and waits for it
during cleanup. CLI fallback checks command ownership before execution:
discovery failures leave skills/workflows available, while errors from owned
plugin handlers remain visible to the caller.
Telemetry failures cannot replace a command's result or original error. The
CLI suggestion hook preserves its last successful catalog while refresh or
subscription initialization retries with capped exponential backoff, and
unmount cancels retries and releases subscriptions that finish late. Desktop
composers also retry failed catalog reads with capped backoff, cancel retries
on target changes or unmount, and stop retrying after a successful read.
including failures, for both direct runtime and Hub calls. Telemetry failures
cannot replace a command's result or original error. Hub status and replacement
checks count active command requests even without a chat participant; draining
refuses new catalog and command requests. CLI and desktop fallbacks treat a
failed catalog read as "not a plugin command" so skills and workflows still
work, while errors from a listed plugin's handler remain visible.
Sandboxed plugin subprocesses and workers are session-local but lazily recreatable. Core
reclaims a sandbox after 30 minutes without an in-flight RPC call (configurable
+15 -32
View File
@@ -3,51 +3,34 @@
Use `core.pluginCommands` on an existing `ClineCore` (local or hub-backed), or
`client.pluginCommands` on an existing `HubSessionClient`. Hosts with a
`NodeHubClient` can use `createHubPluginCommandsApi(client)` from `@cline/core`.
These APIs reuse the runtime and connection; clients do not load plugins or
own a separate command sandbox. Pass the active `sessionId` when invoking a
command during a session so it runs in that session's plugin instance and retains
the state and context shared with its hooks.
Bootstrap/SDK resource loading is configured separately from this API. Changing
how a host ships those resources does not change command ownership or payloads.
These APIs reuse the runtime and connection; clients do not load plugins or own
a separate command sandbox. Pass the active `sessionId` so a command runs in
that session's plugin instance and retains the state and context shared with
its hooks.
If a connection closes after a command is dispatched, execution rejects without
automatically resending it: the command may already have changed external state.
```ts
const target = { workspacePath: workspaceRoot, sessionId };
const stop = core.pluginCommands.subscribe((catalog) => {
if (catalog.workspacePath === workspaceRoot) refreshCommandMenu();
});
const catalog = await core.pluginCommands.list(target);
const result = await core.pluginCommands.run({ ...target, prompt: "/upload-history" });
// Display result?.reply; submit only result?.submitPrompt to the agent.
// undefined means no registered command handled the input.
stop();
```
Omit `sessionId` for workspace discovery before a session exists. The runtime
initializes that workspace catalog on its first request, watches plugin sources
and settings, and publishes changes. A failed load produces `status: "error"`
with diagnostics and retries in the background with bounded backoff. When only
some plugins fail, the catalog still lists and executes healthy commands.
Background retries load only the failed paths, preserving healthy plugin
instances and command state. Source/settings changes still reload the workspace catalog. It never
requires a Git query or a running agent turn. Dispose the owning runtime to
release catalog watchers and sandboxes; unsubscribe when a view closes.
Catalog invalidation follows discoverable plugin entries, imported source files,
and plugin settings; runtime log/state output does not invalidate a catalog.
If source files keep changing during initialization, a request rejects after
two reconciliation passes instead of waiting indefinitely. Retry after edits settle.
For a resident session, both listing and execution use its initialized
contribution registry and existing plugin sandbox, including session context.
Session plugin registrations retain their existing session lifetime: workspace
file changes refresh discovery for new sessions, not an active session's hooks
or commands. A non-resident session can preview workspace commands; execution
requires restoring the session first. Built-in command precedence and rendering
remain host responsibilities. Preserve the original arguments when dispatching.
Omit `sessionId`, or pass one the runtime does not have, to use the workspace
catalog. Every call checks plugin sources and settings and reloads the catalog
when they change, so call `list` again whenever the command menu needs to be
current. A failed load lists healthy commands with `status: "error"` and is
retried after 30 seconds. Workspace catalogs idle for 10 minutes release their
plugin workers.
For a resident session, listing and execution use its initialized contribution
registry and existing plugin sandbox. Workspace file changes apply to new
sessions, not an active session's commands. Built-in command precedence and
rendering remain host responsibilities. Preserve the original arguments when
dispatching.
## Oversized tool result recovery
@@ -96,6 +96,5 @@ export function createClineCorePluginCommandsApi(
return {
list: (target) => service().list(target),
run: (input) => service().run(input),
subscribe: (listener) => service().subscribe(listener),
};
}
@@ -336,23 +336,6 @@ export interface ResolveAndLoadAgentPluginsOptions
export async function resolveAndLoadAgentPlugins(
options: ResolveAndLoadAgentPluginsOptions = {},
) {
const resolved = resolveAgentPluginPathsWithDiagnostics(options);
return loadResolvedAgentPlugins({
...options,
pluginPaths: resolved.paths,
disabledPluginPaths: resolved.disabledPaths,
discoveryFailures: resolved.discoveryFailures,
});
}
/** Load only the supplied resolved paths, without rediscovering other plugins. */
export async function loadResolvedAgentPlugins(
options: ResolveAndLoadAgentPluginsOptions & {
pluginPaths: string[];
disabledPluginPaths?: string[];
discoveryFailures?: Array<{ pluginPath: string; error: unknown }>;
},
): Promise<
{
extensions: AgentPlugin[];
@@ -364,11 +347,7 @@ export async function loadResolvedAgentPlugins(
> {
const mode = resolvePluginExecutionMode(options.mode);
if (mode === "worker") {
const resolved = {
paths: options.pluginPaths,
disabledPaths: options.disabledPluginPaths ?? [],
discoveryFailures: options.discoveryFailures ?? [],
};
const resolved = resolveAgentPluginPathsWithDiagnostics(options);
if (
resolved.paths.length === 0 &&
resolved.disabledPaths.length === 0 &&
@@ -421,7 +400,7 @@ export async function loadResolvedAgentPlugins(
// The subprocess sandbox does not apply the per-session `policy`; it is
// an opt-in escape hatch and loads every discovered plugin.
const paths = options.pluginPaths;
const paths = resolveAgentPluginPaths(options);
if (paths.length === 0) {
return {
extensions: [],
@@ -194,42 +194,6 @@ describe("NodeHubClient", () => {
}
});
it("deduplicates overlapping events per listener while retaining a new listener's replay", async () => {
vi.stubGlobal("WebSocket", MockWebSocket);
const client = new NodeHubClient({ url: "ws://127.0.0.1:25463/hub" });
try {
await client.connect();
const global = vi.fn();
const session = vi.fn();
client.subscribe(global);
const deliver = (eventId: string) =>
MockWebSocket.instances[0].emit("message", {
data: JSON.stringify({
kind: "event",
envelope: {
version: "v1",
event: "assistant.delta",
eventId,
sessionId: "session",
payload: { text: "same text" },
},
}),
});
deliver("event-1");
client.subscribe(session, { sessionId: "session" });
deliver("event-1");
deliver("event-1");
expect(global).toHaveBeenCalledTimes(1);
expect(session).toHaveBeenCalledTimes(1);
deliver("event-2");
deliver("event-2");
expect(global).toHaveBeenCalledTimes(2);
expect(session).toHaveBeenCalledTimes(2);
} finally {
await client.dispose();
}
});
it("re-subscribes global listeners without sending the wildcard sentinel", async () => {
vi.stubGlobal("WebSocket", MockWebSocket);
+1 -18
View File
@@ -56,13 +56,9 @@ type HubCommandOptions = {
onDispatch?: (requestId: string) => void;
};
// Bounded per listener so overlapping wire streams/replay cannot repeat a
// callback, while newly added listeners can still receive their own replay.
const MAX_RECENT_EVENT_IDS = 4096;
type SubscriptionEntry = {
listener: (event: HubEventEnvelope) => void;
sessionId?: string;
deliveredEventIds: Set<string>;
};
function resolveDefaultHubOwnerContext(): HubOwnerContext {
@@ -708,11 +704,7 @@ export class NodeHubClient {
options?: { sessionId?: string },
): () => void {
const sessionId = options?.sessionId?.trim() || undefined;
const entry: SubscriptionEntry = {
listener,
sessionId,
deliveredEventIds: new Set(),
};
const entry = { listener, sessionId };
this.listeners.add(entry);
this.adjustSubscriptionCount(sessionId, 1);
return () => {
@@ -1085,15 +1077,6 @@ export class NodeHubClient {
) {
continue;
}
const eventId = frame.envelope.eventId;
if (eventId) {
if (entry.deliveredEventIds.has(eventId)) continue;
entry.deliveredEventIds.add(eventId);
if (entry.deliveredEventIds.size > MAX_RECENT_EVENT_IDS) {
const oldest = entry.deliveredEventIds.values().next().value;
if (oldest !== undefined) entry.deliveredEventIds.delete(oldest);
}
}
entry.listener(frame.envelope);
}
return;
@@ -29,9 +29,9 @@ async function commandIfSupported(
}
}
/** Uses the caller's existing connection and subscription lifecycle. */
/** Uses the caller's existing connection. */
export function createHubPluginCommandsApi(
client: Pick<NodeHubClient, "command" | "subscribe">,
client: Pick<NodeHubClient, "command">,
): PluginCommandsApi {
return {
async list(target) {
@@ -64,11 +64,5 @@ export function createHubPluginCommandsApi(
);
return reply.payload?.result as PluginCommandResult | undefined;
},
subscribe(listener) {
return client.subscribe((event) => {
if (event.event === "plugins.commands.changed")
listener(event.payload?.catalog as unknown as PluginCommandCatalog);
});
},
};
}
@@ -1,92 +0,0 @@
import { once } from "node:events";
import type { AddressInfo } from "node:net";
import type { HubCommandEnvelope, HubEventEnvelope } from "@cline/shared";
import { expect, it, vi } from "vitest";
import { WebSocketServer } from "ws";
import { HubRuntimeHost } from "./hub-runtime-host";
it("delivers overlapping catalog and session wire events once", async () => {
const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
await once(server, "listening");
const streams = new Set<string>();
server.on("connection", (socket) => {
socket.on("message", (raw) => {
const frame = JSON.parse(String(raw));
if (frame.kind === "stream.subscribe")
streams.add(frame.sessionId ?? "global");
if (frame.kind === "stream.unsubscribe")
streams.delete(frame.sessionId ?? "global");
if (frame.kind !== "command") return;
const request = frame.envelope as HubCommandEnvelope;
socket.send(
JSON.stringify({
kind: "reply",
envelope: {
version: "v1",
requestId: request.requestId,
ok: true,
payload:
request.command === "session.create"
? {
session: {
sessionId: "sess-1",
status: "running",
createdAt: Date.now(),
updatedAt: Date.now(),
},
}
: {},
},
}),
);
});
});
const host = new HubRuntimeHost({
url: `ws://127.0.0.1:${(server.address() as AddressInfo).port}/hub`,
});
try {
const events = vi.fn();
const catalogs = vi.fn();
host.subscribe(events);
host.pluginCommands.subscribe(catalogs);
await host.startSession({
config: {
sessionId: "sess-1",
cwd: "/tmp/project",
workspaceRoot: "/tmp/project",
providerId: "openai-native",
modelId: "gpt-4o-mini",
apiKey: "test",
systemPrompt: "test",
enableTools: false,
enableSpawnAgent: false,
enableAgentTeams: false,
},
interactive: true,
});
await vi.waitFor(() => expect(streams.size).toBeGreaterThan(0));
const delta: HubEventEnvelope = {
version: "v1",
event: "assistant.delta",
eventId: "delta-1",
sessionId: "sess-1",
payload: { text: "hello" },
};
for (const socket of server.clients) {
for (const stream of streams) {
if (stream === "global" || stream === delta.sessionId)
socket.send(JSON.stringify({ kind: "event", envelope: delta }));
}
}
await vi.waitFor(() => expect(events).toHaveBeenCalled());
expect(streams).toEqual(new Set(["global", "sess-1"]));
expect(events).toHaveBeenCalledTimes(1);
expect(catalogs).not.toHaveBeenCalled();
} finally {
await host.dispose();
for (const socket of server.clients) socket.terminate();
await new Promise<void>((resolve, reject) =>
server.close((error) => (error ? reject(error) : resolve())),
);
}
});
@@ -769,9 +769,6 @@ export class HubRuntimeHost implements RuntimeHost {
public readonly pendingPrompts: PendingPromptsServiceApi;
public readonly pluginCommands: PluginCommandsApi;
private client: NodeHubClient;
private readonly pluginCommandListeners = new Set<
(event: HubEventEnvelope) => void
>();
private readonly clientOptions: Omit<HubClientOptions, "url">;
private readonly clientContext?: { workspaceRoot?: string; cwd?: string };
private readonly events = new RuntimeHostEventBus();
@@ -816,22 +813,11 @@ export class HubRuntimeHost implements RuntimeHost {
this.client = this.createClient(options.url);
this.pluginCommands = createHubPluginCommandsApi({
command: (...args) => this.client.command(...args),
subscribe: (listener) => {
this.pluginCommandListeners.add(listener);
return () => {
this.pluginCommandListeners.delete(listener);
};
},
});
}
private createClient(url: string): NodeHubClient {
const client = new NodeHubClient({ ...this.clientOptions, url });
client.subscribe((event) => {
if (event.event === "plugins.commands.changed")
for (const listener of this.pluginCommandListeners) listener(event);
});
return client;
return new NodeHubClient({ ...this.clientOptions, url });
}
private async replaceClient(url: string): Promise<void> {
@@ -1357,7 +1343,6 @@ export class HubRuntimeHost implements RuntimeHost {
// Best-effort detach during shutdown.
}
}
this.pluginCommandListeners.clear();
this.sessionSubscriptions.clear();
this.sessionCapabilities.clear();
this.agentDoneEmittedForCurrentRunBySession.clear();
@@ -141,21 +141,6 @@ export class BrowserWebSocketHubAdapter {
};
const onEvent = (envelope: HubEventEnvelope): void => {
if (
envelope.event === "plugins.commands.changed" &&
!authority?.crossWorkspace
) {
const catalog = envelope.payload?.catalog as
| { workspacePath?: unknown }
| undefined;
const root = authority?.workspaceContext?.workspaceRoot;
if (
!root ||
typeof catalog?.workspacePath !== "string" ||
resolve(catalog.workspacePath) !== resolve(root)
)
return;
}
sendFrame({ kind: "event", envelope });
};
@@ -295,7 +295,6 @@ export class HubServerTransport implements NativeHubTransport {
private readonly pendingPluginCommands = new Set<Promise<unknown>>();
private readonly plugins: PluginRegistry = getProcessPluginRegistry();
private unsubscribePluginStatus?: () => void;
private readonly unsubscribePluginCommands?: () => void;
constructor(readonly options: HubWebSocketServerOptions) {
this.plugins.setLogger(hubPluginLogger);
@@ -476,11 +475,6 @@ export class HubServerTransport implements NativeHubTransport {
createTasksPromptExtension({ todoEnabled: AGENDA_TODO_TOOL_ENABLED }),
);
this.settings = options.settingsService ?? new CoreSettingsService();
this.unsubscribePluginCommands = (
this.sessionHost as RuntimeHost & Partial<PluginCommandsRuntimeService>
).pluginCommands?.subscribe((catalog) => {
this.publish(buildHubEvent("plugins.commands.changed", { catalog }));
});
if (options.cronOptions) {
this.cronService = new CronService({
runtimeHandlers: options.runtimeHandlers,
@@ -770,7 +764,6 @@ export class HubServerTransport implements NativeHubTransport {
);
await this.sessionSearch.dispose();
await this.tasks.dispose();
this.unsubscribePluginCommands?.();
await this.sessionHost.dispose("hub_server_stop");
await this.schedules.dispose();
if (this.cronService) {
@@ -156,55 +156,6 @@ describe("hub-owned plugin commands", () => {
expect(stopped).toBe(true);
});
it("recovers a failed sandbox plugin without restarting its healthy sibling", async () => {
const readyPath = join(root, "allow-recovery");
await writeFile(
join(root, ".cline", "plugins", "retry.js"),
`import {existsSync} from 'node:fs';
export default {name:'retry',manifest:{capabilities:['commands']},setup(api){
if (!existsSync(${JSON.stringify(readyPath)})) throw new Error('not ready');
api.registerCommand({name:'recovered',handler:()=> 'ready'});
}};`,
);
const target = { workspacePath: root };
await runtime.pluginCommands.list(target);
expect(
await runtime.pluginCommands.run({ ...target, prompt: "/counter first" }),
).toMatchObject({ reply: "workspace:1:first" });
const updates = vi.fn();
const stop = runtime.pluginCommands.subscribe(updates);
await vi.waitFor(
() =>
expect(updates).toHaveBeenCalledWith(
expect.objectContaining({ status: "error" }),
),
{ timeout: 5000 },
);
expect(
await runtime.pluginCommands.run({
...target,
prompt: "/counter second",
}),
).toMatchObject({ reply: "workspace:2:second" });
await writeFile(readyPath, "ready");
await vi.waitFor(
() =>
expect(updates).toHaveBeenLastCalledWith(
expect.objectContaining({ status: "ready" }),
),
{ timeout: 7000 },
);
expect(
await runtime.pluginCommands.run({ ...target, prompt: "/recovered" }),
).toMatchObject({ reply: "ready" });
expect(
await runtime.pluginCommands.run({ ...target, prompt: "/counter third" }),
).toMatchObject({ reply: "workspace:3:third" });
expect((await readFile(join(root, "setups.txt"), "utf8")).trim()).toBe(
"workspace",
);
stop();
});
it("keeps healthy commands available while a sibling plugin fails", async () => {
const broken = join(root, ".cline", "plugins", "broken.js");
await writeFile(
@@ -223,48 +174,28 @@ api.registerCommand({name:'recovered',handler:()=> 'ready'});
prompt: "/counter healthy",
}),
).toMatchObject({ reply: "workspace:1:healthy" });
const updates = vi.fn();
const stop = runtime.pluginCommands.subscribe(updates);
await rm(broken);
await vi.waitFor(
() =>
expect(updates).toHaveBeenLastCalledWith(
expect.objectContaining({
status: "ready",
commands: [{ name: "counter" }],
}),
),
{ timeout: 5000 },
);
stop();
expect(
await runtime.pluginCommands.list({ workspacePath: root }),
).toMatchObject({ status: "ready", commands: [{ name: "counter" }] });
});
it("delivers invalidations and current commands through the shared client API", async () => {
it("serves newly added commands through the shared client API", async () => {
const api = createHubPluginCommandsApi({
command: (name, payload, sessionId) =>
command(name, payload ?? {}, sessionId),
subscribe: (listener) => transport.subscribe("observer", listener),
});
const updates = vi.fn();
const stop = api.subscribe(updates);
await api.list({ workspacePath: root });
await writeFile(
join(root, ".cline", "plugins", "extra.js"),
`export default {name:'extra',manifest:{capabilities:['commands']},setup(api){api.registerCommand({name:'extra',handler:()=> 'new command'});}};`,
);
await vi.waitFor(
() =>
expect(updates).toHaveBeenLastCalledWith(
expect.objectContaining({
commands: expect.arrayContaining([{ name: "extra" }]),
}),
),
{ timeout: 5000 },
expect((await api.list({ workspacePath: root })).commands).toEqual(
expect.arrayContaining([{ name: "extra" }]),
);
expect(await api.run({ workspacePath: root, prompt: "/extra" })).toEqual({
reply: "new command",
submitPrompt: undefined,
});
stop();
});
it("treats a Hub without plugin command RPCs as having no plugin commands", async () => {
const api = createHubPluginCommandsApi({
@@ -275,7 +206,6 @@ api.registerCommand({name:'recovered',handler:()=> 'ready'});
`Unsupported hub schedule command: ${name}`,
);
},
subscribe: () => () => {},
});
expect(await api.list({ workspacePath: root })).toEqual({
workspacePath: root,
@@ -394,7 +394,6 @@ export class LocalRuntimeHost implements RuntimeHost {
),
);
},
subscribe: this.pluginCommandManager.subscribe,
};
// A caller-owned telemetry service may already be identified to an
// authenticated account (the long-lived Hub daemon is one example).
@@ -1,21 +0,0 @@
import { randomUUID } from "node:crypto";
import { appendFileSync, existsSync, readFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
const id = randomUUID();
const directory = dirname(fileURLToPath(import.meta.url));
export default {
name: "duplicate",
manifest: { capabilities: ["commands"] },
setup(api, ctx) {
const settings = JSON.parse(
readFileSync(join(directory, "fixture-settings.json"), "utf8"),
);
if (settings.failedPlugin === "earlier" && !existsSync(settings.gate))
throw new Error("not ready");
ctx.onDispose(() => appendFileSync(settings.disposed, `earlier:${id}\n`));
api.registerCommand({ name: "same", handler: () => `earlier:${id}` });
api.registerCommand({ name: "obsolete", handler: () => "obsolete" });
},
};
@@ -1,20 +0,0 @@
import { randomUUID } from "node:crypto";
import { appendFileSync, existsSync, readFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
const id = randomUUID();
const directory = dirname(fileURLToPath(import.meta.url));
export default {
name: "duplicate",
manifest: { capabilities: ["commands"] },
setup(api, ctx) {
const settings = JSON.parse(
readFileSync(join(directory, "fixture-settings.json"), "utf8"),
);
if (settings.failedPlugin === "later" && !existsSync(settings.gate))
throw new Error("not ready");
ctx.onDispose(() => appendFileSync(settings.disposed, `later:${id}\n`));
api.registerCommand({ name: "same", handler: () => `later:${id}` });
},
};
@@ -27,13 +27,15 @@ export interface PluginCommandCatalog extends PluginCommandTarget {
commands: PluginSlashCommand[];
error?: string;
}
/** Owned by the runtime. Clients neither load plugins nor own their sandboxes. */
/**
* Owned by the runtime. Clients neither load plugins nor own their sandboxes.
* Every call checks plugin sources, so re-listing is how clients stay current.
*/
export interface PluginCommandsApi {
list(target: PluginCommandTarget): Promise<PluginCommandCatalog>;
run(
input: PluginCommandTarget & { prompt: string },
): Promise<PluginCommandResult | undefined>;
subscribe(listener: (catalog: PluginCommandCatalog) => void): () => void;
}
export interface PluginCommandsRuntimeService {
readonly pluginCommands: PluginCommandsApi;
@@ -1,13 +1,4 @@
import { EventEmitter } from "node:events";
import type { watch } from "node:fs";
import {
copyFile,
mkdir,
mkdtemp,
readFile,
rm,
writeFile,
} from "node:fs/promises";
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type {
@@ -31,6 +22,7 @@ beforeEach(async () => {
});
const managers: PluginCommandManager[] = [];
afterEach(async () => {
vi.useRealTimers();
await Promise.all(managers.splice(0).map((m) => m.dispose()));
setHomeDir(originalHome ?? "~");
setClineDir(originalClineDir ?? join(originalHome ?? "~", ".cline"));
@@ -58,7 +50,11 @@ function loaded(
},
],
pluginPaths: [],
failures: [],
failures: [] as Array<{
pluginPath: string;
phase: "setup";
message: string;
}>,
warnings: [],
issues: [],
shutdown: vi.fn(async () => {}),
@@ -87,112 +83,39 @@ describe("runtime plugin catalogs", () => {
manager.run({ workspacePath, prompt: "/unknown" }),
).resolves.toBeUndefined();
});
it("retries only failed paths while retaining healthy command state", async () => {
it("keeps healthy commands through a failure and retries the load after a delay", async () => {
vi.useFakeTimers();
const workspacePath = await workspace();
const brokenPath = join(workspacePath, "broken.js");
const failures = [
{ pluginPath: brokenPath, phase: "setup" as const, message: "broken" },
];
let counter = 0;
let recover = false;
const healthy = loaded(async () => String(++counter));
const recovered = loaded();
recovered.extensions[0].name = "recovered-plugin";
recovered.extensions[0].setup = (api) =>
api.registerCommand({ name: "fixed", handler: () => "recovered" });
const load = vi.fn(async (_options: { pluginPaths: string[] }) => {
if (load.mock.calls.length === 1) return { ...healthy, failures };
return recover ? recovered : { ...loaded(), extensions: [], failures };
const failing = loaded();
failing.failures.push({
pluginPath: join(workspacePath, "broken.js"),
phase: "setup",
message: "broken",
});
const manager = new PluginCommandManager({ load, retryDelayMs: 10 });
managers.push(manager);
const updates = vi.fn();
manager.subscribe(updates);
await manager.list({ workspacePath });
expect(await manager.run({ workspacePath, prompt: "/echo" })).toMatchObject(
{ reply: "1" },
);
await vi.waitFor(
() => expect(load.mock.calls.length).toBeGreaterThanOrEqual(3),
{ timeout: 5000 },
);
expect(healthy.shutdown).not.toHaveBeenCalled();
expect(await manager.run({ workspacePath, prompt: "/echo" })).toMatchObject(
{ reply: "2" },
);
for (const [options] of load.mock.calls.slice(1))
expect(options.pluginPaths).toEqual([brokenPath]);
recover = true;
await vi.waitFor(
() =>
expect(updates).toHaveBeenLastCalledWith(
expect.objectContaining({ status: "ready" }),
),
{ timeout: 5000 },
);
expect(
await manager.run({ workspacePath, prompt: "/fixed" }),
).toMatchObject({ reply: "recovered" });
expect(await manager.run({ workspacePath, prompt: "/echo" })).toMatchObject(
{ reply: "3" },
);
expect(healthy.shutdown).not.toHaveBeenCalled();
});
it.each([
"earlier",
"later",
])("a recovered %s duplicate preserves priority and releases only the replaced plugin", async (failed) => {
const workspacePath = await workspace();
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
const earlier = join(directory, "a.js");
const later = join(directory, "b.js");
const disposed = join(workspacePath, "disposed.txt");
const gate = join(workspacePath, "ready.txt");
await copyFile(
new URL("./fixtures/command-priority/earlier.js", import.meta.url),
earlier,
);
await copyFile(
new URL("./fixtures/command-priority/later.js", import.meta.url),
later,
);
await writeFile(
join(directory, "fixture-settings.json"),
JSON.stringify({ gate, disposed, failedPlugin: failed }),
);
const manager = new PluginCommandManager({ retryDelayMs: 10 });
const load = vi
.fn()
.mockResolvedValueOnce(failing)
.mockImplementation(async () => loaded());
const manager = new PluginCommandManager({ load });
managers.push(manager);
expect(await manager.list({ workspacePath })).toMatchObject({
status: "error",
error: "broken",
commands: [{ name: "echo" }],
});
const initial = (await manager.run({ workspacePath, prompt: "/same" }))
?.reply;
expect(initial).toMatch(failed === "later" ? /^earlier:/ : /^later:/);
await writeFile(gate, "ready");
await vi.waitFor(
async () => {
expect(await manager.list({ workspacePath })).toMatchObject({
status: "ready",
commands: [{ name: "same" }],
});
},
{ timeout: 5000 },
);
const winner = (await manager.run({ workspacePath, prompt: "/same" }))
?.reply;
expect(winner).toMatch(/^later:/);
if (failed === "earlier") expect(winner).toBe(initial);
expect(
await manager.run({ workspacePath, prompt: "/obsolete" }),
).toBeUndefined();
const releases = (await readFile(disposed, "utf8")).trim().split("\n");
expect(releases.filter((id) => id === initial)).toHaveLength(
failed === "later" ? 1 : 0,
);
await manager.run({ workspacePath, prompt: "/echo" });
expect(load).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(30_000);
expect(await manager.list({ workspacePath })).toMatchObject({
status: "ready",
commands: [{ name: "echo" }],
});
expect(load).toHaveBeenCalledTimes(2);
expect(failing.shutdown).toHaveBeenCalledTimes(1);
});
it("rechecks imported helpers outside watched plugin directories", async () => {
it("rechecks imported helpers outside plugin directories", async () => {
const workspacePath = await workspace();
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
@@ -203,12 +126,7 @@ describe("runtime plugin catalogs", () => {
`import {value} from '../../helpers.js';
export default {name:'helper',manifest:{capabilities:['commands']},setup(api){api.registerCommand({name:'helper',description:value,handler:()=>value})}};`,
);
const manager = new PluginCommandManager({
watch: (() =>
Object.assign(new EventEmitter(), {
close() {},
})) as unknown as typeof watch,
});
const manager = new PluginCommandManager();
managers.push(manager);
expect(
await manager.run({ workspacePath, prompt: "/helper" }),
@@ -222,7 +140,7 @@ export default {name:'helper',manifest:{capabilities:['commands']},setup(api){ap
).toMatchObject({ reply: "later" });
});
it("reconciles edits made during initial catalog loading", async () => {
it("reloads edits made during initial catalog loading on the next request", async () => {
const workspacePath = await workspace();
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
@@ -233,13 +151,7 @@ export default {name:'helper',manifest:{capabilities:['commands']},setup(api){ap
if (load.mock.calls.length === 1) await writeFile(plugin, "// changed");
return loaded(async () => contents);
});
const manager = new PluginCommandManager({
load,
watch: (() =>
Object.assign(new EventEmitter(), {
close() {},
})) as unknown as typeof watch,
});
const manager = new PluginCommandManager({ load });
managers.push(manager);
await manager.list({ workspacePath });
expect(await manager.run({ workspacePath, prompt: "/echo" })).toMatchObject(
@@ -259,250 +171,45 @@ export default {name:'logging',manifest:{capabilities:['commands']},setup(api){a
);
const manager = new PluginCommandManager();
managers.push(manager);
// A broken refresh implementation must still be interruptible so this
// regression test can dispose its workers instead of hanging the suite.
const deadline = setTimeout(() => void manager.dispose(), 5000);
try {
expect(await manager.list({ workspacePath })).toMatchObject({
status: "ready",
commands: [{ name: "counter" }],
});
expect(
await manager.run({ workspacePath, prompt: "/counter" }),
).toMatchObject({ reply: "1" });
await writeFile(join(directory, "state.json"), '{"count":1}');
await manager.list({ workspacePath });
expect(
await manager.run({ workspacePath, prompt: "/counter" }),
).toMatchObject({ reply: "2" });
expect(await readFile(join(directory, "setup.log"), "utf8")).toBe(
"setup\n",
);
} finally {
clearTimeout(deadline);
await manager.dispose();
}
});
it("bounds reconciliation when imported source keeps changing during setup", async () => {
const workspacePath = await workspace();
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
const plugin = join(directory, "changing.js");
await writeFile(plugin, "// initial");
const load = vi.fn(async () => {
if (load.mock.calls.length <= 5)
await writeFile(plugin, `// edit ${load.mock.calls.length}`);
return loaded();
expect(await manager.list({ workspacePath })).toMatchObject({
status: "ready",
commands: [{ name: "counter" }],
});
const manager = new PluginCommandManager({
load,
watch: (() =>
Object.assign(new EventEmitter(), {
close() {},
})) as unknown as typeof watch,
});
managers.push(manager);
await expect(manager.list({ workspacePath })).rejects.toThrow(
"Plugin sources changed repeatedly",
expect(
await manager.run({ workspacePath, prompt: "/counter" }),
).toMatchObject({ reply: "1" });
await writeFile(join(directory, "state.json"), '{"count":1}');
await manager.list({ workspacePath });
expect(
await manager.run({ workspacePath, prompt: "/counter" }),
).toMatchObject({ reply: "2" });
expect(await readFile(join(directory, "setup.log"), "utf8")).toBe(
"setup\n",
);
expect(load).toHaveBeenCalledTimes(2);
await manager.dispose();
});
it("recovers from initialization timeout without another client request", async () => {
it("releases idle workspace catalogs and reloads them on demand", async () => {
vi.useFakeTimers();
const workspacePath = await workspace();
let recover!: (value: ReturnType<typeof loaded>) => void;
const recovered = new Promise<ReturnType<typeof loaded>>((resolve) => {
recover = resolve;
});
const first = loaded();
const load = vi
.fn()
.mockRejectedValueOnce(new Error("plugin-sandbox initialize timed out"))
.mockImplementation(() => recovered);
const manager = new PluginCommandManager({ load, retryDelayMs: 10 });
managers.push(manager);
const updates = vi.fn();
manager.subscribe(updates);
const initial = await manager.list({ workspacePath });
recover(loaded());
expect(initial).toMatchObject({
status: "error",
commands: [],
error: expect.stringContaining("timed out"),
});
await vi.waitFor(
() =>
expect(updates).toHaveBeenLastCalledWith(
expect.objectContaining({
status: "ready",
commands: [{ name: "echo", description: "Echo" }],
}),
),
{ timeout: 5000 },
);
await expect(
manager.run({ workspacePath, prompt: "/echo recovered" }),
).resolves.toMatchObject({ reply: "echo:recovered" });
});
it("watches newly created plugin directories in non-Git workspaces", async () => {
const workspacePath = await workspace();
const load = vi.fn(async () => loaded());
.mockResolvedValueOnce(first)
.mockImplementation(async () => loaded());
const manager = new PluginCommandManager({ load });
managers.push(manager);
await manager.list({ workspacePath });
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
await writeFile(join(directory, "echo.js"), "export default {};");
await vi.waitFor(() => expect(load.mock.calls.length).toBeGreaterThan(1), {
timeout: 3000,
});
const calls = load.mock.calls.length;
await writeFile(
join(directory, "echo.js"),
"export default {name:'changed'};",
);
await vi.waitFor(
() => expect(load.mock.calls.length).toBeGreaterThan(calls),
{ timeout: 3000 },
);
});
it("moves ancestor watches toward a new plugin directory before its source fingerprint changes", async () => {
const workspacePath = await workspace();
const callbacks: Array<{
directory: string;
listener: (event: string, file: string) => void;
closed: boolean;
}> = [];
const watchFiles = vi.fn(
(
directory: string,
_options: unknown,
listener: (event: string, file: string) => void,
) => {
const watched = { directory, listener, closed: false };
callbacks.push(watched);
return Object.assign(new EventEmitter(), {
close: () => {
watched.closed = true;
},
});
},
);
const load = vi.fn(async () => loaded());
const manager = new PluginCommandManager({
load,
watch: watchFiles as unknown as typeof watch,
});
managers.push(manager);
await vi.advanceTimersByTimeAsync(9 * 60_000);
await manager.list({ workspacePath });
const currentWatch = (directory: string) => {
const watched = [...callbacks]
.reverse()
.find((item) => item.directory === directory && !item.closed);
if (!watched) throw new Error(`Missing active watcher for ${directory}`);
return watched;
};
const config = join(workspacePath, ".cline");
await mkdir(config);
currentWatch(workspacePath).listener("rename", ".cline");
// An empty config directory changes watch topology, not plugin sources.
const configWatch = currentWatch(config);
expect(load).toHaveBeenCalledTimes(1);
const plugins = join(config, "plugins");
await mkdir(plugins);
configWatch.listener("rename", "plugins");
const pluginWatch = currentWatch(plugins);
await writeFile(join(plugins, "echo.js"), "export default {};");
pluginWatch.listener("rename", "echo.js");
await vi.waitFor(() => expect(load.mock.calls.length).toBeGreaterThan(1));
});
it("reconciles a plugin written after the only ancestor event", async () => {
const workspacePath = await workspace();
let ancestorEvent: ((event: string, file: string) => void) | undefined;
const watchFiles = vi.fn(
(
directory: string,
_options: unknown,
listener: (event: string, file: string) => void,
) => {
if (directory === workspacePath) ancestorEvent = listener;
return Object.assign(new EventEmitter(), { close: vi.fn() });
},
);
const load = vi.fn(async () => loaded());
const manager = new PluginCommandManager({
load,
watch: watchFiles as unknown as typeof watch,
});
managers.push(manager);
await manager.list({ workspacePath });
vi.useFakeTimers();
try {
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(join(workspacePath, ".cline"));
if (!ancestorEvent) throw new Error("Missing ancestor watcher");
ancestorEvent("rename", ".cline");
// Complete the immediate empty-source check before writing the plugin.
await manager.list({ workspacePath });
await mkdir(directory);
await writeFile(join(directory, "echo.js"), "export default {};");
// Native events can coalesce while a recursive watch is installed.
// No second event is delivered; the debounce must reconcile this write.
await vi.advanceTimersByTimeAsync(100);
} finally {
vi.useRealTimers();
}
await vi.waitFor(() => expect(load.mock.calls.length).toBeGreaterThan(1));
});
it.each([
"add",
"edit",
"remove",
"unchanged",
"watch-throw",
])("reconciles %s during watcher outages without resetting unchanged plugins", async (change) => {
const workspacePath = await workspace();
const directory = join(workspacePath, ".cline", "plugins");
await mkdir(directory, { recursive: true });
const file = join(directory, "echo.js");
if (change !== "add") await writeFile(file, "before");
const watchers: EventEmitter[] = [];
let unavailable = change === "watch-throw";
const watchFiles = vi.fn(() => {
if (unavailable) throw new Error("watch unavailable");
const watcher = Object.assign(new EventEmitter(), { close: vi.fn() });
watchers.push(watcher);
return watcher;
});
const initial = loaded();
const load = vi.fn(async () => initial);
const manager = new PluginCommandManager({
load,
watch: watchFiles as unknown as typeof watch,
});
managers.push(manager);
const updates = vi.fn();
manager.subscribe(updates);
await manager.list({ workspacePath });
const attempts = watchFiles.mock.calls.length;
watchers[0]?.emit("error", new Error("watch lost"));
if (change === "remove") await rm(file);
else if (change !== "unchanged") await writeFile(file, "after");
unavailable = false;
await vi.waitFor(
() => expect(watchFiles.mock.calls.length).toBeGreaterThan(attempts),
{ timeout: 3000 },
);
if (change === "unchanged") {
expect(load).toHaveBeenCalledTimes(1);
expect(initial.shutdown).not.toHaveBeenCalled();
} else {
await vi.waitFor(() => expect(updates).toHaveBeenCalledTimes(2));
expect(load).toHaveBeenCalledTimes(2);
expect(initial.shutdown).toHaveBeenCalledTimes(1);
}
expect(first.shutdown).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(10 * 60_000);
expect(first.shutdown).toHaveBeenCalledTimes(1);
await expect(
manager.run({ workspacePath, prompt: "/echo again" }),
).resolves.toMatchObject({ reply: "echo:again" });
expect(load).toHaveBeenCalledTimes(2);
});
it("waits for active handlers before disposing the sandbox", async () => {
const workspacePath = await workspace();
let finish!: (result: string) => void;
@@ -526,6 +233,7 @@ export default {name:'logging',manifest:{capabilities:['commands']},setup(api){a
expect(plugins.shutdown).toHaveBeenCalledTimes(1);
await expect(manager.list({ workspacePath })).rejects.toThrow("disposed");
});
it("isolates workspaces and propagates handler errors", async () => {
const a = await workspace();
const b = await workspace();
@@ -554,6 +262,7 @@ export default {name:'logging',manifest:{capabilities:['commands']},setup(api){a
),
).rejects.toThrow("handler boom");
});
it("parses slash input without changing internal whitespace", () => {
expect(parsePluginCommand(" \n/GOAL a b\nc ")).toEqual({
name: "goal",
+101 -358
View File
@@ -1,7 +1,6 @@
import { createHash } from "node:crypto";
import { existsSync, type FSWatcher, statSync, watch } from "node:fs";
import { readFile } from "node:fs/promises";
import { dirname, resolve, sep } from "node:path";
import { resolve } from "node:path";
import {
type AgentExtensionCommand,
type AgentTool,
@@ -12,122 +11,118 @@ import {
} from "@cline/shared";
import { resolveGlobalSettingsPath } from "@cline/shared/storage";
import {
loadResolvedAgentPlugins,
resolveAgentPluginPaths,
resolvePluginConfigSearchPaths,
resolveAndLoadAgentPlugins,
} from "../extensions/plugin/plugin-config-loader";
import { fingerprintPluginSources } from "../extensions/plugin/plugin-module-import";
import { PluginRegistry } from "../extensions/plugin/plugin-registry";
import {
executePluginCommand,
listPluginCommands,
type PluginCommandCatalog,
type PluginCommandsApi,
type PluginCommandResult,
type PluginCommandTarget,
} from "./plugin-command-api";
const MAX_CATALOG_REFRESH_PASSES = 2;
// A failed load may be transient (sandbox startup timeout, I/O hiccup), so it
// is retried on a later request instead of waiting for a source change.
const FAILED_LOAD_RETRY_MS = 30_000;
// The Hub is long-lived and shared across workspaces; release plugin workers
// for workspaces that stop asking for commands.
const IDLE_EVICTION_MS = 10 * 60_000;
type LoadedPlugins = Awaited<ReturnType<typeof loadResolvedAgentPlugins>>;
type LoadedPlugin = {
path: string;
loaded: LoadedPlugins;
contributions: Array<{ name: string; commands: AgentExtensionCommand[] }>;
};
type Entry = {
workspacePath: string;
registry: PluginRegistry;
key?: string;
catalog: PluginCommandCatalog;
commands: AgentExtensionCommand[];
loaded: LoadedPlugin[];
pluginPaths: string[];
failedPaths?: string[];
retryDue: boolean;
pending?: Promise<void>;
dirty: boolean;
retryCount: number;
timer?: ReturnType<typeof setTimeout>;
watchers: FSWatcher[];
watchRetry?: ReturnType<typeof setTimeout>;
sourceCheckTimer?: ReturnType<typeof setTimeout>;
fingerprint?: string;
/** Serializes execution with reload/disposal so a handler cannot lose its sandbox. */
failedAt?: number;
shutdown?: () => Promise<void>;
/** Serializes loads, execution, and disposal so a handler cannot lose its sandbox. */
tail: Promise<unknown>;
idleTimer?: ReturnType<typeof setTimeout>;
};
/** Runtime-owned workspace catalogs. Session execution uses the session registry instead. */
export class PluginCommandManager implements PluginCommandsApi {
export class PluginCommandManager {
private readonly entries = new Map<string, Entry>();
private readonly listeners = new Set<
(catalog: PluginCommandCatalog) => void
>();
private disposed = false;
constructor(
private readonly options: {
logger?: BasicLogger;
telemetry?: ITelemetryService;
load?: typeof loadResolvedAgentPlugins;
retryDelayMs?: number;
watch?: typeof watch;
load?: typeof resolveAndLoadAgentPlugins;
} = {},
) {}
subscribe = (
listener: (catalog: PluginCommandCatalog) => void,
): (() => void) => {
this.listeners.add(listener);
return () => {
this.listeners.delete(listener);
};
};
async list(target: PluginCommandTarget): Promise<PluginCommandCatalog> {
return this.withEntry(target, (entry) => structuredClone(entry.catalog));
}
private entry(target: PluginCommandTarget): Entry {
async run(
input: PluginCommandTarget & { prompt: string },
): Promise<PluginCommandResult | undefined> {
return this.withEntry(input, (entry) =>
executePluginCommand(entry.commands, input.prompt, {
telemetry: this.options.telemetry,
}),
);
}
async dispose(): Promise<void> {
this.disposed = true;
const entries = [...this.entries.values()];
this.entries.clear();
await Promise.all(entries.map((entry) => this.release(entry)));
}
private withEntry<T>(
target: PluginCommandTarget,
operation: (entry: Entry) => T | Promise<T>,
): Promise<T> {
if (this.disposed) throw new Error("Plugin command manager is disposed");
const workspacePath = resolve(target.workspacePath);
let entry = this.entries.get(workspacePath);
if (!entry) {
entry = {
workspacePath,
registry: new PluginRegistry({ logger: this.options.logger }),
catalog: { workspacePath, status: "ready", commands: [] },
commands: [],
loaded: [],
pluginPaths: [],
retryDue: false,
dirty: true,
retryCount: 0,
watchers: [],
tail: Promise.resolve(),
};
this.entries.set(workspacePath, entry);
}
return entry;
const current = entry;
clearTimeout(current.idleTimer);
current.idleTimer = setTimeout(() => {
if (this.entries.get(workspacePath) !== current) return;
this.entries.delete(workspacePath);
void this.release(current);
}, IDLE_EVICTION_MS);
current.idleTimer.unref?.();
const result = current.tail.then(async () => {
await this.refresh(current);
return operation(current);
});
current.tail = result.catch(() => {});
return result;
}
private schedule(entry: Entry, delay: number, retryOnly = false): void {
if (this.disposed) return;
clearTimeout(entry.timer);
entry.timer = setTimeout(() => {
if (retryOnly) entry.retryDue = true;
else entry.dirty = true;
void this.refresh(entry);
}, delay);
entry.timer.unref?.();
}
private watchRoots(entry: Entry): Set<string> {
return new Set([
...resolvePluginConfigSearchPaths(entry.workspacePath),
...entry.pluginPaths.map(dirname),
resolveGlobalSettingsPath(),
]);
private async release(entry: Entry): Promise<void> {
clearTimeout(entry.idleTimer);
await entry.tail;
const shutdown = entry.shutdown;
entry.shutdown = undefined;
await shutdown?.().catch(() => {});
}
/** Fingerprint discoverable entries, their import graphs, and plugin settings.
* Runtime output beside a plugin is not source and must not reset its state. */
private async fingerprint(entry: Entry): Promise<string> {
private async fingerprint(workspacePath: string): Promise<string> {
const hash = createHash("sha256");
try {
const paths = resolveAgentPluginPaths({
cwd: entry.workspacePath,
workspacePath: entry.workspacePath,
cwd: workspacePath,
workspacePath,
});
hash.update(JSON.stringify(paths));
for (const path of paths) hash.update(fingerprintPluginSources(path));
@@ -143,300 +138,48 @@ export class PluginCommandManager implements PluginCommandsApi {
}
return hash.digest("hex");
}
private scheduleSourceCheck(entry: Entry): void {
clearTimeout(entry.sourceCheckTimer);
entry.sourceCheckTimer = setTimeout(() => {
void this.fingerprint(entry).then((fingerprint) => {
if (this.disposed || fingerprint === entry.fingerprint) return;
entry.dirty = true;
void this.refresh(entry);
});
}, 100);
entry.sourceCheckTimer.unref?.();
}
private reconnect(entry: Entry): void {
if (this.disposed) return;
clearTimeout(entry.watchRetry);
entry.watchRetry = setTimeout(() => {
if (this.disposed) return;
// Install watchers first, then reconcile changes missed during the outage.
this.watch(entry);
}, 1000);
entry.watchRetry.unref?.();
}
private watch(entry: Entry): void {
clearTimeout(entry.watchRetry);
if (this.disposed) return;
for (const watcher of entry.watchers) watcher.close();
entry.watchers = [];
const settings = resolveGlobalSettingsPath();
const roots = this.watchRoots(entry);
for (const root of roots) {
const targetDirectory = root === settings ? dirname(root) : root;
let directory = targetDirectory;
while (!existsSync(directory) && dirname(directory) !== directory)
directory = dirname(directory);
try {
const recursive =
directory === root && statSync(directory).isDirectory();
const watcher = (this.options.watch ?? watch)(
directory,
{ recursive, persistent: false },
(event, file) => {
const changed = file ? resolve(directory, String(file)) : directory;
if (String(file).split(/[\\/]/).includes("node_modules")) return;
if (
changed !== root &&
!changed.startsWith(`${root}${sep}`) &&
!root.startsWith(`${changed}${sep}`)
)
return;
// Directory creation/removal changes where we must listen even when
// no discoverable plugin exists yet. Install the closer watch before
// checking sources so later child events cannot fall through the gap.
if (event === "rename" || directory !== targetDirectory)
this.watch(entry);
this.scheduleSourceCheck(entry);
},
);
watcher.on("error", (error) => {
this.options.logger?.debug?.(
"Plugin command watcher failed; reconnecting",
{ directory, error },
);
this.reconnect(entry);
});
entry.watchers.push(watcher);
} catch (error) {
this.options.logger?.debug?.("Plugin command watcher unavailable", {
directory,
error,
});
this.reconnect(entry);
}
}
// Native events may coalesce while recursive watches are being installed.
// Reconcile the completed write batch, without restarting unchanged plugins.
this.scheduleSourceCheck(entry);
}
private async loadPlugin(
entry: Entry,
pluginPaths: string[],
retryOnly: boolean,
): Promise<{ failedPaths: string[]; errors: string[] }> {
const failedPaths: string[] = [];
const errors: string[] = [];
let loaded: LoadedPlugins | undefined;
private async refresh(entry: Entry): Promise<void> {
const key = await this.fingerprint(entry.workspacePath);
if (
key === entry.key &&
(entry.failedAt === undefined ||
Date.now() - entry.failedAt < FAILED_LOAD_RETRY_MS)
)
return;
await entry.shutdown?.().catch(() => {});
entry.key = key;
entry.shutdown = undefined;
entry.commands = [];
entry.failedAt = undefined;
let error: string | undefined;
try {
if (retryOnly && !this.options.load) {
for (const path of pluginPaths) {
if (
entry.registry.get(path).some((status) => status.state === "failed")
)
await entry.registry.reload(path);
}
}
loaded = await (this.options.load ?? loadResolvedAgentPlugins)({
const loaded = await (this.options.load ?? resolveAndLoadAgentPlugins)({
cwd: entry.workspacePath,
workspacePath: entry.workspacePath,
pluginPaths,
registry: entry.registry,
});
const contributions: LoadedPlugin["contributions"] = [];
for (const extension of loaded.extensions) {
const registry = createContributionRegistry<
typeof extension,
AgentTool,
Message[]
>({ extensions: [extension] });
await registry.initialize();
contributions.push({
name: extension.name,
commands: registry.getRegistrySnapshot().commands,
});
}
failedPaths.push(...loaded.failures.map((failure) => failure.pluginPath));
errors.push(...loaded.failures.map((failure) => failure.message));
if (contributions.length)
entry.loaded.push({
path: pluginPaths[0] ?? "",
loaded,
contributions,
});
else await loaded.shutdown?.();
entry.shutdown = loaded.shutdown;
const registry = createContributionRegistry<
(typeof loaded.extensions)[number],
AgentTool,
Message[]
>({ extensions: loaded.extensions });
await registry.initialize();
entry.commands = registry.getRegistrySnapshot().commands;
if (loaded.failures.length)
error = loaded.failures.map((failure) => failure.message).join("; ");
} catch (cause) {
await loaded?.shutdown?.().catch(() => {});
failedPaths.push(...pluginPaths);
errors.push(cause instanceof Error ? cause.message : String(cause));
this.options.logger?.error?.(
"Plugin command discovery failed; retrying",
{ error: cause },
);
}
return { failedPaths, errors };
}
private async selectCommands(entry: Entry): Promise<void> {
const selected = new Map<
string,
{ owner: LoadedPlugin; commands: AgentExtensionCommand[] }
>();
entry.loaded.sort(
(a, b) =>
entry.pluginPaths.indexOf(a.path) - entry.pluginPaths.indexOf(b.path),
);
for (const owner of entry.loaded) {
for (const contribution of owner.contributions) {
selected.delete(contribution.name);
selected.set(contribution.name, {
owner,
commands: contribution.commands,
});
}
}
const owners = new Set([...selected.values()].map(({ owner }) => owner));
await Promise.all(
entry.loaded
.filter((owner) => !owners.has(owner))
.map(({ loaded }) => loaded.shutdown?.()),
);
entry.loaded = entry.loaded.filter((owner) => owners.has(owner));
entry.commands = [...selected.values()].flatMap(({ commands }) => commands);
}
private refresh(entry: Entry): Promise<void> {
if (entry.pending) return entry.pending;
if ((!entry.dirty && !entry.retryDue) || this.disposed)
return Promise.resolve();
const retryOnly = !entry.dirty;
entry.dirty = false;
entry.retryDue = false;
clearTimeout(entry.timer);
entry.pending = entry.tail
.then(async () => {
if (this.disposed) return;
if (!retryOnly) {
await Promise.all(
entry.loaded.map(({ loaded }) =>
loaded.shutdown?.().catch(() => {}),
),
);
entry.loaded = [];
entry.commands = [];
entry.failedPaths = undefined;
}
const errors: string[] = [];
if (!retryOnly) entry.fingerprint = await this.fingerprint(entry);
try {
if (!entry.failedPaths) {
entry.pluginPaths = resolveAgentPluginPaths({
cwd: entry.workspacePath,
workspacePath: entry.workspacePath,
});
entry.failedPaths = entry.pluginPaths;
// Cover edits during initialization, as well as later watcher outages.
entry.fingerprint = await this.fingerprint(entry);
this.watch(entry);
}
const paths = entry.failedPaths;
entry.failedPaths = [];
// Each path owns its worker lifetime so a recovered override can
// release its predecessor without restarting unrelated plugins.
const results = await Promise.all(
(paths.length ? paths.map((path) => [path]) : [[]]).map((paths) =>
this.loadPlugin(entry, paths, retryOnly),
),
);
entry.failedPaths = [
...new Set(results.flatMap((result) => result.failedPaths)),
];
errors.push(...results.flatMap((result) => result.errors));
await this.selectCommands(entry);
if (!errors.length) entry.retryCount = 0;
} catch (cause) {
errors.push(cause instanceof Error ? cause.message : String(cause));
}
const error = errors.join("; ") || undefined;
entry.catalog = {
workspacePath: entry.workspacePath,
status: error ? "error" : "ready",
error,
commands: listPluginCommands(entry.commands),
};
if (error)
this.schedule(
entry,
Math.min(
(this.options.retryDelayMs ?? 1000) * 2 ** entry.retryCount++,
30_000,
),
true,
);
if (!this.disposed) {
this.watch(entry);
for (const listener of this.listeners) {
try {
listener(structuredClone(entry.catalog));
} catch (error) {
this.options.logger?.error?.("Plugin command subscriber failed", {
error,
});
}
}
}
})
.finally(() => {
entry.pending = undefined;
if (entry.dirty || entry.retryDue)
this.schedule(entry, entry.dirty ? 100 : 0, !entry.dirty);
error = cause instanceof Error ? cause.message : String(cause);
this.options.logger?.error?.("Plugin command discovery failed", {
error: cause,
});
entry.tail = entry.pending.catch(() => {});
return entry.pending;
}
private async ensureFresh(entry: Entry): Promise<void> {
for (let pass = 0; pass < MAX_CATALOG_REFRESH_PASSES; pass++) {
await this.refresh(entry);
if (this.disposed) return;
// Watchers keep menus current, but imports can reach outside their
// roots. Check the loaded graph before serving either API request.
const fingerprint = await this.fingerprint(entry);
if (fingerprint === entry.fingerprint) return;
entry.dirty = true;
}
throw new Error(
"Plugin sources changed repeatedly during command discovery. Retry after edits settle.",
);
}
async list(target: PluginCommandTarget): Promise<PluginCommandCatalog> {
const entry = this.entry(target);
await this.ensureFresh(entry);
return structuredClone(entry.catalog);
}
async run(input: PluginCommandTarget & { prompt: string }) {
const entry = this.entry(input);
await this.ensureFresh(entry);
if (this.disposed) throw new Error("Plugin command manager is disposed");
const run = entry.tail.then(() =>
executePluginCommand(entry.commands, input.prompt, {
telemetry: this.options.telemetry,
}),
);
entry.tail = run.catch(() => {});
return await run;
}
async dispose(): Promise<void> {
this.disposed = true;
this.listeners.clear();
await Promise.all(
[...this.entries.values()].map(async (entry) => {
clearTimeout(entry.timer);
clearTimeout(entry.watchRetry);
clearTimeout(entry.sourceCheckTimer);
for (const watcher of entry.watchers) watcher.close();
await entry.tail;
await Promise.all(
entry.loaded.map(({ loaded }) => loaded.shutdown?.()),
);
}),
);
this.entries.clear();
if (error) entry.failedAt = Date.now();
entry.catalog = {
workspacePath: entry.workspacePath,
status: error ? "error" : "ready",
error,
commands: listPluginCommands(entry.commands),
};
}
}
-1
View File
@@ -813,7 +813,6 @@ export type HubEventName =
| "task.run.completed"
| "task.run.failed"
| "task.automation.updated"
| "plugins.commands.changed"
| "settings.changed"
| "plugin.status_changed"
| "ui.notify"