mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0f808b3582 |
+82
-20
@@ -112,24 +112,81 @@ test.each(["foreground", "finish subscription"])(
|
||||
},
|
||||
);
|
||||
|
||||
test.each(["veto", "error"])(
|
||||
"a supported provider %s retains the runtime until a fresh idle hour passes",
|
||||
async (decision) => {
|
||||
const s = scenario();
|
||||
s.runtime.probe = async () => {
|
||||
if (decision === "error") throw new Error("Probe failed");
|
||||
return false;
|
||||
};
|
||||
const blocked = await s.sweep(61);
|
||||
expect(blocked.collected).toEqual([]);
|
||||
expect(blocked.failures).toHaveLength(decision === "error" ? 1 : 0);
|
||||
await s.sweep(119);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(120)).collected).toEqual([]);
|
||||
expect((await s.sweep(179)).collected).toEqual([]);
|
||||
expect((await s.sweep(180)).collected).toHaveLength(1);
|
||||
},
|
||||
);
|
||||
test("a provider veto retains the runtime until a fresh idle hour passes", async () => {
|
||||
const s = scenario();
|
||||
s.runtime.probe = async () => false;
|
||||
expect(await s.sweep(61)).toEqual({ collected: [], failures: [] });
|
||||
await s.sweep(119);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(120)).collected).toEqual([]);
|
||||
expect((await s.sweep(179)).collected).toEqual([]);
|
||||
expect((await s.sweep(180)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("repeated provider errors retry without postponing an already idle runtime", async () => {
|
||||
const s = scenario();
|
||||
const error = new Error("Probe failed");
|
||||
s.runtime.probe = async () => {
|
||||
throw error;
|
||||
};
|
||||
expect(await s.sweep(61)).toEqual({ collected: [], failures: [{ agentId: "agent", error }] });
|
||||
expect(await s.sweep(119)).toEqual({ collected: [], failures: [{ agentId: "agent", error }] });
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(120)).collected).toEqual([
|
||||
{ agentId: "agent", provider: "test", sessionId: "agent" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("a failed probe does not clear a genuine veto's idle clock", async () => {
|
||||
const s = scenario();
|
||||
s.runtime.probe = async () => false;
|
||||
await s.sweep(61);
|
||||
s.runtime.probe = async () => {
|
||||
throw new Error("Probe failed");
|
||||
};
|
||||
expect((await s.sweep(62)).failures).toHaveLength(1);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(63)).collected).toEqual([]);
|
||||
expect((await s.sweep(122)).collected).toEqual([]);
|
||||
expect((await s.sweep(123)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("a failed probe does not bypass the original one-hour idle period", async () => {
|
||||
const s = scenario();
|
||||
s.runtime.probe = async () => {
|
||||
throw new Error("Probe failed");
|
||||
};
|
||||
expect((await s.sweep(1)).failures).toHaveLength(1);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(59)).collected).toEqual([]);
|
||||
expect((await s.sweep(60)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("a child's failed probe retains its family only until safety is established", async () => {
|
||||
const s = scenario();
|
||||
const child = new Runtime("child");
|
||||
child.parentAgentId = s.runtime.id;
|
||||
const error = new Error("Child probe failed");
|
||||
child.probe = async () => {
|
||||
throw error;
|
||||
};
|
||||
s.runtimes.push(child);
|
||||
expect(await s.sweep(61)).toEqual({
|
||||
collected: [],
|
||||
failures: [
|
||||
{ agentId: "agent", error },
|
||||
{ agentId: "child", error },
|
||||
],
|
||||
});
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
expect(child.closed).toBe(false);
|
||||
child.probe = async () => true;
|
||||
expect((await s.sweep(62)).collected).toEqual([
|
||||
{ agentId: "agent", provider: "test", sessionId: "agent" },
|
||||
{ agentId: "child", provider: "test", sessionId: "child" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("an idle managed descendant with provider-held work vetoes its parent", async () => {
|
||||
const s = scenario();
|
||||
@@ -157,7 +214,7 @@ test("new caller protection during an async provider probe prevents stale collec
|
||||
expect((await s.sweep(122)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("a supported provider timeout retains ownership even if it later returns true", async () => {
|
||||
test("a provider timeout retains ownership until a fresh probe succeeds without restarting idleness", async () => {
|
||||
const s = scenario();
|
||||
const probe = deferred<boolean>();
|
||||
s.runtime.probe = () => probe.promise;
|
||||
@@ -165,8 +222,13 @@ test("a supported provider timeout retains ownership even if it later returns tr
|
||||
expect(result.collected).toEqual([]);
|
||||
expect(result.failures[0]?.error).toEqual(new Error("Runtime safety probe timed out for agent"));
|
||||
probe.resolve(true);
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
await probe.promise;
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(62)).collected).toEqual([
|
||||
{ agentId: "agent", provider: "test", sessionId: "agent" },
|
||||
]);
|
||||
expect(s.runtime.closed).toBe(true);
|
||||
}, 15_000);
|
||||
|
||||
test("the host opt-out disables explicit sweeps and scheduled ticks", async () => {
|
||||
|
||||
@@ -124,7 +124,7 @@ export function createIdleRuntimeCleanup(options: Options) {
|
||||
}
|
||||
} catch (error) {
|
||||
if (!isCurrent()) break;
|
||||
protectedAt.set(runtime.identity, { at: timestamp, protected: true });
|
||||
// An unknown safety result retains ownership, but is not evidence of activity.
|
||||
result.failures.push({ agentId: runtime.agentId, error });
|
||||
options.logger.warn(
|
||||
{ agentId: runtime.agentId, err: error },
|
||||
|
||||
@@ -20,7 +20,6 @@ import { DaemonConfigStore } from "../daemon-config-store.js";
|
||||
import { PluginService } from "./index.js";
|
||||
import { BuiltinPluginLoader } from "./builtin/index.js";
|
||||
import { ManagedPluginSources } from "./managed-source.js";
|
||||
import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
|
||||
import {
|
||||
startNpmRegistry,
|
||||
npmPluginPackages,
|
||||
@@ -180,55 +179,6 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
|
||||
return { runtime, started, releaseStart, starts };
|
||||
}
|
||||
|
||||
function createGatedRuntime(providers: Record<string, readonly PluginProviderMetadata[]> = {}) {
|
||||
const gates = new Map<string, PromiseWithResolvers<void>>();
|
||||
const gate = (pluginId: string) => {
|
||||
const existing = gates.get(pluginId);
|
||||
if (existing) return existing;
|
||||
const created = Promise.withResolvers<void>();
|
||||
gates.set(pluginId, created);
|
||||
return created;
|
||||
};
|
||||
const begun: string[] = [];
|
||||
const running = new Set<string>();
|
||||
const runtime: TestPluginRuntime = {
|
||||
...emptyUsageRuntime,
|
||||
catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
|
||||
invoke: async () => undefined,
|
||||
getLogs: () => [],
|
||||
clearLogs: () => undefined,
|
||||
getProviderRegistrations: (pluginId) =>
|
||||
running.has(pluginId) ? (providers[pluginId] ?? []) : [],
|
||||
startPlugin: async (pluginId) => {
|
||||
begun.push(pluginId);
|
||||
await gate(pluginId).promise;
|
||||
running.add(pluginId);
|
||||
},
|
||||
stopPluginById: async (pluginId) => running.delete(pluginId),
|
||||
stopAll: async () => {
|
||||
running.clear();
|
||||
},
|
||||
subscribe: () => () => undefined,
|
||||
bindPaseoSessionHost: () => undefined,
|
||||
};
|
||||
return {
|
||||
runtime,
|
||||
begun,
|
||||
finish: (pluginId: string) => gate(pluginId).resolve(),
|
||||
fail: (pluginId: string, error: Error) => gate(pluginId).reject(error),
|
||||
};
|
||||
}
|
||||
|
||||
function nextNotification(service: PluginService, pluginId: string): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const unsubscribe = service.subscribe((notified) => {
|
||||
if (notified !== pluginId) return;
|
||||
unsubscribe();
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
describe("PluginService", () => {
|
||||
it.each([
|
||||
{ shadow: false, runtime: "subprocess" },
|
||||
@@ -422,127 +372,15 @@ describe("PluginService", () => {
|
||||
"startup-first": { source: "directory", path: first },
|
||||
"startup-second": { source: "directory", path: second },
|
||||
});
|
||||
const notifications: { pluginId: string; running: boolean }[] = [];
|
||||
service.subscribe((pluginId) =>
|
||||
notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }),
|
||||
);
|
||||
const snapshots: string[][] = [];
|
||||
service.subscribe(() => snapshots.push(catalogIds(service)));
|
||||
|
||||
await service.start();
|
||||
|
||||
expect(notifications).toEqual([
|
||||
{ pluginId: "startup-first", running: true },
|
||||
{ pluginId: "startup-second", running: true },
|
||||
]);
|
||||
expect(catalogIds(service)).toEqual(["startup-first", "startup-second"]);
|
||||
expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
|
||||
await service.stopAllPlugins();
|
||||
}, 20_000);
|
||||
|
||||
it("starts configured plugins concurrently and publishes them in config order", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime();
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
alpha: { source: "directory", path: "/plugins/alpha" },
|
||||
beta: { source: "directory", path: "/plugins/beta" },
|
||||
gamma: { source: "directory", path: "/plugins/gamma" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const notified: string[] = [];
|
||||
service.subscribe((pluginId) => notified.push(pluginId));
|
||||
let started = false;
|
||||
const starting = (async () => {
|
||||
await service.start();
|
||||
started = true;
|
||||
})();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["alpha", "beta", "gamma"]);
|
||||
|
||||
gated.finish("gamma");
|
||||
const alphaPublished = nextNotification(service, "alpha");
|
||||
gated.finish("alpha");
|
||||
await alphaPublished;
|
||||
expect(notified).toEqual(["alpha"]);
|
||||
expect(started).toBe(false);
|
||||
|
||||
gated.finish("beta");
|
||||
await starting;
|
||||
expect(notified).toEqual(["alpha", "beta", "gamma"]);
|
||||
expect(catalogIds(service)).toEqual(["alpha", "beta", "gamma"]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("starts the other configured plugins when one fails while they are still starting", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime();
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
broken: { source: "directory", path: "/plugins/broken" },
|
||||
healthy: { source: "directory", path: "/plugins/healthy" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const notified: string[] = [];
|
||||
service.subscribe((pluginId) => notified.push(pluginId));
|
||||
const starting = service.start();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["broken", "healthy"]);
|
||||
const brokenPublished = nextNotification(service, "broken");
|
||||
gated.fail("broken", new Error("startup exploded"));
|
||||
await brokenPublished;
|
||||
gated.finish("healthy");
|
||||
await starting;
|
||||
|
||||
expect(notified).toEqual(["broken", "healthy"]);
|
||||
expect(catalogIds(service)).toEqual(["healthy"]);
|
||||
expect(await service.listPlugins()).toMatchObject([
|
||||
{ id: "broken", status: "failed", error: "startup exploded" },
|
||||
{ id: "healthy", status: "running" },
|
||||
]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("keeps a provider ID with the plugin listed first when both start at once", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime({
|
||||
first: [{ id: "shared-agent", label: "First agent" }],
|
||||
second: [{ id: "shared-agent", label: "Second agent" }],
|
||||
});
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
first: { source: "directory", path: "/plugins/first" },
|
||||
second: { source: "directory", path: "/plugins/second" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const starting = service.start();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["first", "second"]);
|
||||
gated.finish("second");
|
||||
gated.finish("first");
|
||||
await starting;
|
||||
|
||||
expect(service.getProviderRegistrations()).toMatchObject([
|
||||
{ id: "shared-agent", label: "First agent" },
|
||||
]);
|
||||
expect(catalogIds(service)).toEqual(["first"]);
|
||||
expect(await service.listPlugins()).toMatchObject([
|
||||
{ id: "first", status: "running" },
|
||||
{
|
||||
id: "second",
|
||||
status: "failed",
|
||||
error: 'Plugin second cannot register provider ID "shared-agent" twice',
|
||||
},
|
||||
]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("uses an explicit config key, exposes reload failure, and retries from disk", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
|
||||
@@ -230,18 +230,11 @@ export class PluginService {
|
||||
}
|
||||
});
|
||||
if (config.pluginsEnabled === true) {
|
||||
let earlierPublished: Promise<unknown> = Promise.resolve();
|
||||
await Promise.all(
|
||||
Object.entries(config.plugins ?? {})
|
||||
.filter(([, source]) => source.enabled !== false)
|
||||
.map(async ([pluginId]) => {
|
||||
const publishTurn = earlierPublished;
|
||||
const starting = this.startConfigured(pluginId, publishTurn);
|
||||
earlierPublished = Promise.all([publishTurn, starting]);
|
||||
await starting;
|
||||
this.notify(pluginId);
|
||||
}),
|
||||
);
|
||||
for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
|
||||
if (source.enabled === false) continue;
|
||||
await this.startConfigured(pluginId);
|
||||
this.notify(pluginId);
|
||||
}
|
||||
}
|
||||
this.configStore.onFieldChange("pluginsEnabled", (value) => {
|
||||
this.handleGlobalSwitch(value === true);
|
||||
@@ -567,12 +560,12 @@ export class PluginService {
|
||||
}).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins"));
|
||||
}
|
||||
|
||||
private async startConfigured(pluginId: string, publishTurn?: Promise<unknown>): Promise<void> {
|
||||
private async startConfigured(pluginId: string): Promise<void> {
|
||||
const source = this.configStore.get().plugins?.[pluginId];
|
||||
if (!source || source.enabled === false || !this.canPublish(pluginId)) return;
|
||||
this.errors.delete(pluginId);
|
||||
try {
|
||||
await this.startPlugin(pluginId, source.path, publishTurn);
|
||||
await this.startPlugin(pluginId, source.path);
|
||||
} catch (error) {
|
||||
if (this.canPublish(pluginId)) this.recordFailure(pluginId, error);
|
||||
}
|
||||
@@ -601,13 +594,8 @@ export class PluginService {
|
||||
);
|
||||
}
|
||||
|
||||
private async startPlugin(
|
||||
pluginId: string,
|
||||
sourcePath: string,
|
||||
publishTurn?: Promise<unknown>,
|
||||
): Promise<void> {
|
||||
private async startPlugin(pluginId: string, sourcePath: string): Promise<void> {
|
||||
await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId));
|
||||
await publishTurn;
|
||||
try {
|
||||
await this.publishProviderRegistrations(pluginId, sourcePath);
|
||||
this.publishUsageSources(pluginId);
|
||||
|
||||
Reference in New Issue
Block a user