diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index 2b34165515..c7d2b32e89 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -238,6 +238,21 @@ active registered types appear there; plugin activation remains owned by Settings → Plugins → Manage plugins. Provider defaults are versioned device preferences, not agent-record migrations; blank per-agent values inherit them. +Code that agents need while they run, such as an agent type's runtime, tools, +wakes and Attention sources, belongs in the plugin's optional `agent` export: +`export const agent = { inject, apply(ctx) }`, with its own `inject`. `apply` +stays the plugin's UI. Both parts now start in the same page, under the same +plugin identity; the plugin is active once both are, and fails if either does. +This separation lets agents later run apart from the UI page. + +A harness plugin registers its type from its `agent` entry and presents it +from `apply`: `ctx.agents.present(typeId, (exposed) => ({ icon, Peek, Create, +CreateModel, tabs, provider: { DialogBody, DefaultControl } }))`. The type's +readers see one type with both halves. `exposed` is the type's `expose`: the +handles its UI reaches of the runtime. Each is JSON, a store (`snapshot` and +`subscribe`) or a call returning a promise, so it can later cross to another +page. Claude is the first harness split this way. + A plugin, bundled or installed, can add a tool to every agent session in the app with `ctx.agents.addTool({ name, description, inputSchema, call })`. `call` gets the agent's Buzz client and where it is working, so the tool acts as the diff --git a/src/app/services.ts b/src/app/services.ts index 9fc686611a..af262d1e18 100644 --- a/src/app/services.ts +++ b/src/app/services.ts @@ -1,6 +1,8 @@ import { createAdminAccess } from "../bundled/admin-access"; import { nativeRelayStaff } from "../features/relay-staff/native"; import { createLocalAgentActivity } from "../features/agents/local-activity"; +import { link } from "../features/agents/link"; +import { linkedActivity, serveActivity } from "../features/agents/runner-link"; // FOUNDATION: Compose the bundled distribution, plugin runtime, and services here. import { createIdentity, @@ -52,6 +54,20 @@ export function createServices() { }); const localAgents = createLocalAgents(); const localActivity = createLocalAgentActivity(); + // The Manager, which the screens use, and the Runner, which runs agents, + // share this page for now. Only JSON crosses the link between them. + const agentsLink = new MessageChannel(); + const runnerLink = link( + agentsLink.port1, + serveActivity(localActivity.reader), + ); + const managerLink = link(agentsLink.port2); + const runnerActivity = linkedActivity(managerLink); + ctx.effect(() => () => { + runnerActivity.stop(); + managerLink.close(); + runnerLink.close(); + }); const navigationHost = provideNavigation(ctx); const navigation = navigationHost.navigation; const browser = new BrowserService(ctx); @@ -72,7 +88,7 @@ export function createServices() { localAgents.agents, identity?.ready, identity ? connectNativeTransport : undefined, - localActivity.reader, + runnerActivity.source, ); ctx.provide("communityReader", { snapshot: communities.snapshot, diff --git a/src/bundled/agents/AgentsPage.test.tsx b/src/bundled/agents/AgentsPage.test.tsx index 8d1da7866f..f4473de98c 100644 --- a/src/bundled/agents/AgentsPage.test.tsx +++ b/src/bundled/agents/AgentsPage.test.tsx @@ -281,6 +281,7 @@ function setup( session: vi.fn(async () => {}), register: vi.fn(), addTool: vi.fn(), + present: vi.fn(), tools: () => [], create: vi.fn(async () => { // `shown: false`: it was made for a community no longer selected. diff --git a/src/bundled/agents/ProviderDialog.tsx b/src/bundled/agents/ProviderDialog.tsx index 96761eb77c..f480357e24 100644 --- a/src/bundled/agents/ProviderDialog.tsx +++ b/src/bundled/agents/ProviderDialog.tsx @@ -82,7 +82,7 @@ function ManagedProviderDialog({ finalFocus={() => finalFocus.current} > - + {Body && } ); diff --git a/src/bundled/claude-code/index.test.ts b/src/bundled/claude-code/index.test.ts index f11b432574..94f6200eb2 100644 --- a/src/bundled/claude-code/index.test.ts +++ b/src/bundled/claude-code/index.test.ts @@ -1,8 +1,8 @@ // @vitest-environment jsdom import { afterEach, expect, it, vi } from "vitest"; -import type { AgentType } from "../../features/agents/service"; import { ClaudeRuntime } from "./runtime"; -import { apply } from "./index"; +import * as claude from "./index"; +import { appliedType } from "../../features/agents/plugin-testing"; vi.mock("@tauri-apps/api/core", () => ({ isTauri: () => true })); afterEach(() => vi.restoreAllMocks()); @@ -26,18 +26,12 @@ it("observes saved sessions with the live status of running ones", () => { { key: "room/r", busy: true }, { key: "dm", busy: false }, ]); - let type: AgentType | undefined; - apply({ + const type = appliedType(claude, { pluginOwner: { id: "buzz.claude-code", revision: "r1" }, host: { request: vi.fn() }, - agents: { - register: (registered: AgentType) => { - type = registered; - }, - relay: () => undefined, - }, + agents: { relay: () => undefined }, effect: vi.fn(), - } as never); + }); expect(type?.sessions?.list(agent)).toEqual([ { channelId: "room", root: "r", at: 3, status: "Working" }, { channelId: "dm", at: 2, status: "Idle" }, diff --git a/src/bundled/claude-code/index.ts b/src/bundled/claude-code/index.ts index 83c6dd3709..00ec7e50a0 100644 --- a/src/bundled/claude-code/index.ts +++ b/src/bundled/claude-code/index.ts @@ -18,11 +18,56 @@ export { TYPE_ID }; /** How often Buzz's copy of Claude looks for a new release. */ const UPDATE_MS = 6 * 60 * 60_000; -export const inject = ["agents", "host"]; +// Only the desktop app can start `claude`, and setup and file uploads run +// `bash` and `base64`, which stock Windows lacks; no unusable type elsewhere. +const supported = () => isTauri() && /Mac|Linux/i.test(navigator.platform); + +/** What the type's UI reaches of its runtime. */ +export type ClaudeExposed = Parameters[0]; + +export const inject = ["agents"]; export const apply: PluginModule["apply"] = (ctx) => { - // Only the desktop app can start `claude`, and setup and file uploads run - // `bash` and `base64`, which stock Windows lacks; no unusable type elsewhere. - if (!isTauri() || !/Mac|Linux/i.test(navigator.platform)) return; + if (!supported()) return; + ctx.agents.present(TYPE_ID, (exposed) => { + const { + Create, + CreateModel, + DefaultControl, + ProviderDialogBody, + SettingsTab, + ActivityTab, + } = createTabs(exposed); + return { + icon: ClaudeLogoIcon, + Create, + CreateModel, + provider: { DialogBody: ProviderDialogBody, DefaultControl }, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + splitSettings: true, + component: SettingsTab, + }, + { + id: "activity", + title: "Activity", + role: "activity", + component: ActivityTab, + }, + ], + }; + }); +}; + +// The runtime, which runs where the agents do. +export const agent: NonNullable = { + inject: ["agents", "host"], + apply: (ctx) => applyAgent(ctx), +}; +function applyAgent(ctx: Parameters[0]) { + if (!supported()) return; const { agents, host } = ctx; const typeKey = `${ctx.pluginOwner?.id ?? "buzz.claude-code"}/${TYPE_ID}`; const spawn = host.spawn && withScript(host.spawn.bind(host)); @@ -72,27 +117,17 @@ export const apply: PluginModule["apply"] = (ctx) => { }); } - const { - Create, - CreateModel, - DefaultControl, - ProviderDialogBody, - SettingsTab, - ActivityTab, - } = createTabs({ - setup, - defaults: providerDefaults, - runtime, - }); agents.register({ id: TYPE_ID, title: "Claude", - icon: ClaudeLogoIcon, description: "Claude on this computer. Each thread it is mentioned in is one continuing session.", defaults: () => ({ config: DEFAULT_CONFIG }), - Create, - CreateModel, + expose: { + setup, + defaults: providerDefaults, + runtime, + } satisfies ClaudeExposed, adopt: (carry, saved) => ({ ...saved, instructions: carry.instructions ?? saved.instructions, @@ -145,8 +180,6 @@ export const apply: PluginModule["apply"] = (ctx) => { ? () => void setup.run("login") : undefined; }, - DialogBody: ProviderDialogBody, - DefaultControl, }, sessions: { list: (key) => { @@ -169,22 +202,7 @@ export const apply: PluginModule["apply"] = (ctx) => { }, subscribe: (listener) => runtime.subscribe(listener), }, - tabs: [ - { - id: "settings", - title: "Settings", - role: "settings", - splitSettings: true, - component: SettingsTab, - }, - { - id: "activity", - title: "Activity", - role: "activity", - component: ActivityTab, - }, - ], // A mention during the first download waits for it, within its deadline. ...claudeRun(runtime, (signal) => setup.installed(signal)), }); -}; +} diff --git a/src/bundled/claude-code/tabs.test.tsx b/src/bundled/claude-code/tabs.test.tsx index 4a5ac3469a..7b33b20e9f 100644 --- a/src/bundled/claude-code/tabs.test.tsx +++ b/src/bundled/claude-code/tabs.test.tsx @@ -25,7 +25,8 @@ import { } from "../../features/agents/state"; import type { HostProcessOptions } from "../../features/host/service"; import { ClaudeLogoIcon } from "../../shared/design-system/icons"; -import { apply, inject } from "./index"; +import * as claude from "./index"; +import { appliedType } from "../../features/agents/plugin-testing"; import { type Config, DEFAULT_CONFIG } from "./runtime"; import { ClaudeSetup } from "./setup"; import { ClaudeDefaults } from "./defaults"; @@ -126,7 +127,6 @@ function applied({ /** Whether Claude lists its models when asked. */ catalog?: { listed: boolean; ready?: Promise }; } = {}) { - let type: AgentType | undefined; const spawn = vi.fn(async (id: string, options: HostProcessOptions = {}) => { // Claude and its install steps run as steps of claude.sh: its script, $0 // and step come first. @@ -246,16 +246,13 @@ function applied({ }, agents: { storage, - register: (registered: AgentType) => { - type = registered; - }, snapshot: () => ({ status: "ready", agents: [], running: [] }), relay: () => undefined, subscribe: () => () => undefined, }, effect: vi.fn(), }; - apply(ctx as never); + const type = appliedType(claude, ctx); return { type, spawn }; } function installed(options?: Parameters[0]) { @@ -269,8 +266,9 @@ function tab(type: AgentType, index: number) { return component; } function providerSetup(type: AgentType) { - if (!type.provider) throw new Error("no provider contribution"); - return type.provider.DialogBody; + const body = type.provider?.DialogBody; + if (!body) throw new Error("no provider contribution"); + return body; } async function choose(label: string, option: string) { const user = userEvent.setup(); @@ -405,7 +403,8 @@ it("finds the thread's Claude session another build started", async () => { }); it("is an agent type on the desktop app", () => { - expect(inject).toEqual(["agents", "host"]); + expect(claude.inject).toEqual(["agents"]); + expect(claude.agent.inject).toEqual(["agents", "host"]); const { type } = installed(); expect(type).toMatchObject({ id: "claude-code", diff --git a/src/bundled/claude-code/tabs.tsx b/src/bundled/claude-code/tabs.tsx index a727f97dc1..372d72abd4 100644 --- a/src/bundled/claude-code/tabs.tsx +++ b/src/bundled/claude-code/tabs.tsx @@ -50,6 +50,24 @@ import { import type { ClaudeModel } from "./claude"; import type { ClaudeSetup } from "./setup"; import type { ClaudeDefaults } from "./defaults"; + +// What the tabs reach of the runtime: stores and calls returning promises, +// as an agent type's `expose` allows. +type SetupHandle = Pick< + ClaudeSetup, + | "snapshot" + | "subscribe" + | "check" + | "loadModels" + | "run" + | "answer" + | "cancel" +>; +type DefaultsHandle = Pick; +type ActivityHandle = Pick< + ClaudeRuntime, + "subscribe" | "sessions" | "conversations" | "find" | "transcript" +>; import { TURN_LIMIT } from "./transcript"; const message = (error: unknown) => @@ -139,7 +157,7 @@ function ClaudeSignIn({ setup, disabled, }: { - setup: ClaudeSetup; + setup: SetupHandle; disabled: boolean; }) { const [open, setOpen] = useState(false); @@ -180,7 +198,7 @@ function ClaudeSetupPanel({ disabled = false, variant = "full", }: { - setup: ClaudeSetup; + setup: SetupHandle; disabled?: boolean; variant?: "full" | "compact"; }) { @@ -324,9 +342,9 @@ export function createTabs({ defaults, runtime, }: Readonly<{ - setup: ClaudeSetup; - defaults: ClaudeDefaults; - runtime: ClaudeRuntime; + setup: SetupHandle; + defaults: DefaultsHandle; + runtime: ActivityHandle; }>) { function useSetupCatalog() { const state = useSyncExternalStore( diff --git a/src/bundled/codex/index.test.ts b/src/bundled/codex/index.test.ts index 518f375e86..bad5863eda 100644 --- a/src/bundled/codex/index.test.ts +++ b/src/bundled/codex/index.test.ts @@ -1,11 +1,11 @@ // @vitest-environment jsdom import { afterEach, beforeEach, expect, it, vi } from "vitest"; -import type { AgentType } from "../../features/agents/service"; import { CodexLogoIcon } from "../../shared/design-system/icons"; import { CodexRuntime } from "./runtime"; import { CodexSetup } from "./setup"; import { CodexProvider } from "./tabs"; -import { apply } from "./index"; +import * as codex from "./index"; +import { appliedType } from "../../features/agents/plugin-testing"; const desktop = vi.hoisted(() => ({ tauri: true })); vi.mock("@tauri-apps/api/core", () => ({ isTauri: () => desktop.tauri })); @@ -21,23 +21,18 @@ afterEach(() => { /** Applies the plugin to a stand-in context and returns the registered type. */ function applied() { - let type: AgentType | undefined; const subscribe = () => () => undefined; const ctx = { pluginOwner: { id: "buzz.codex", revision: "r1" }, host: { spawn: vi.fn(), request: vi.fn() }, agents: { - register: (registered: AgentType) => { - type = registered; - }, snapshot: () => ({ status: "ready", agents: [], running: [] }), relay: () => undefined, subscribe, }, effect: vi.fn(), }; - apply(ctx as never); - return type; + return appliedType(codex, ctx); } it("registers only where the desktop app can run codex and base64", () => { @@ -123,7 +118,7 @@ it("stops agents that stop running even when the agent list could not be read", const agent = { pubkey: "agent", type: "buzz.codex/codex" }; let state = { status: "ready", agents: [agent], running: [agent] }; let notify = () => undefined; - apply({ + codex.agent.apply({ pluginOwner: { id: "buzz.codex" }, host: { spawn: vi.fn(), request: vi.fn() }, agents: { @@ -187,7 +182,7 @@ it("starts the server once Codex is updated and checked, only while a Codex agen notify(); }; let cleanup: () => void = () => undefined; - apply({ + codex.agent.apply({ pluginOwner: { id: "buzz.codex" }, host: { spawn: vi.fn(), request: vi.fn() }, agents: { diff --git a/src/bundled/codex/index.ts b/src/bundled/codex/index.ts index 39206660f5..4db1e8585f 100644 --- a/src/bundled/codex/index.ts +++ b/src/bundled/codex/index.ts @@ -8,15 +8,68 @@ import { type Config, config, defaults, validWorkspace } from "./config"; import { codexRun, TYPE_ID } from "./run"; import { CodexRuntime } from "./runtime"; import { CodexSetup, withScript } from "./setup"; -import { CodexProvider, createTabs } from "./tabs"; +import { CodexProvider, createTabs, transcripts } from "./tabs"; import { readUsage } from "./usage"; /** How often Buzz's copy of Codex looks for a new release. */ const UPDATE_MS = 6 * 60 * 60_000; -export const inject = ["agents", "host"]; +// Only the desktop app can start `codex`, and setup and file uploads run +// `bash` and `base64`, which stock Windows lacks; no unusable type elsewhere. +const supported = () => isTauri() && /Mac|Linux/i.test(navigator.platform); + +/** What the type's UI reaches of its runtime. */ +type CodexExposed = Readonly<{ + provider: Parameters[0]; + runtime: Parameters[1]; + transcript: Parameters[2]; + setup: Parameters[3]; +}>; + +export const inject = ["agents"]; export const apply: PluginModule["apply"] = (ctx) => { - // Only the desktop app can start `codex`, and setup and file uploads run - // `bash` and `base64`, which stock Windows lacks; no unusable type elsewhere. - if (!isTauri() || !/Mac|Linux/i.test(navigator.platform)) return; + if (!supported()) return; + ctx.agents.present( + TYPE_ID, + ({ provider, runtime, transcript, setup }) => { + const { + Create, + CreateModel, + DefaultControl, + ProviderDialogBody, + SettingsTab, + ActivityTab, + } = createTabs(provider, runtime, transcript, setup); + return { + icon: CodexLogoIcon, + Create, + CreateModel, + provider: { DialogBody: ProviderDialogBody, DefaultControl }, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + splitSettings: true, + component: SettingsTab, + }, + { + id: "activity", + title: "Activity", + role: "activity", + component: ActivityTab, + }, + ], + }; + }, + ); +}; + +// The runtime, which runs where the agents do. +export const agent: NonNullable = { + inject: ["agents", "host"], + apply: (ctx) => applyAgent(ctx), +}; +function applyAgent(ctx: Parameters[0]) { + if (!supported()) return; const spawn = ctx.host.spawn && withScript(ctx.host.spawn.bind(ctx.host)); const readFile = ctx.host.readFile?.bind(ctx.host); const setup = new CodexSetup(spawn, (input) => ctx.host.request(input)); @@ -89,23 +142,18 @@ export const apply: PluginModule["apply"] = (ctx) => { runtime.dispose(); setup.dispose(); }); - const { - Create, - CreateModel, - DefaultControl, - ProviderDialogBody, - SettingsTab, - ActivityTab, - } = createTabs(providerController, runtime, start, setup); ctx.agents.register({ id: TYPE_ID, title: "Codex", - icon: CodexLogoIcon, description: "Codex on this computer, using its native coding tools and your account.", defaults: () => ({ config: { ...defaults } }), - Create, - CreateModel, + expose: { + provider: providerController, + runtime, + transcript: transcripts(start, runtime), + setup, + } satisfies CodexExposed, importShared: (source) => source.runtime === "codex" && (!source.provider || source.provider === "openai") @@ -155,8 +203,6 @@ export const apply: PluginModule["apply"] = (ctx) => { snapshot: providerController.snapshot, subscribe: providerController.subscribe, check: () => providerController.check(), - DialogBody: ProviderDialogBody, - DefaultControl, }, sessions: { list: (key) => { @@ -177,21 +223,6 @@ export const apply: PluginModule["apply"] = (ctx) => { }, subscribe: runtime.subscribe, }, - tabs: [ - { - id: "settings", - title: "Settings", - role: "settings", - splitSettings: true, - component: SettingsTab, - }, - { - id: "activity", - title: "Activity", - role: "activity", - component: ActivityTab, - }, - ], // A mention during the first download waits for it, within its deadline. ...codexRun(runtime, async (signal) => { await setup.installed(signal); @@ -199,4 +230,4 @@ export const apply: PluginModule["apply"] = (ctx) => { await providerController.check(); }), }); -}; +} diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index 78452aab95..9aef47f211 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -1,7 +1,7 @@ import { createHash } from "node:crypto"; import { schnorr } from "@noble/curves/secp256k1.js"; import { bytesToHex } from "nostr-tools/utils"; -import { apply } from "./index"; +import { agent as codexAgent } from "./index"; import { describe, afterEach, expect, it, vi } from "vitest"; import { pubkeyOf } from "../../features/agents/membership"; import { ownerReads } from "../../features/agents/reads"; @@ -436,7 +436,7 @@ function fixture( if (plugin) { vi.stubGlobal("localStorage", local); vi.stubGlobal("navigator", { platform: "MacIntel" }); - apply({ + codexAgent.apply({ host: { spawn }, agents: { snapshot: () => ({ status: "ready", agents: [], running }), @@ -454,7 +454,7 @@ function fixture( effect: (effect: () => () => void) => { pluginDispose = effect(); }, - } as unknown as Parameters[0]); + } as unknown as Parameters[0]); } const starts = () => sent.filter((wire) => wire.method === "turn/start"); const interrupt = () => { diff --git a/src/bundled/codex/tabs.test.tsx b/src/bundled/codex/tabs.test.tsx index d6b0ae6aeb..a1fe89c0e7 100644 --- a/src/bundled/codex/tabs.test.tsx +++ b/src/bundled/codex/tabs.test.tsx @@ -30,7 +30,7 @@ import type { RelayData } from "../../features/relay/service"; import { PanelWorkspace } from "../../features/panels/PanelWorkspace"; import { defaults, type Config } from "./config"; import { sessionTag } from "./prompt"; -import { CodexProvider, createTabs } from "./tabs"; +import { CodexProvider, createTabs, transcripts } from "./tabs"; import { CodexRuntime } from "./runtime"; import type { Wire } from "./rpc"; import { CodexSetup } from "./setup"; @@ -200,7 +200,12 @@ function fixture({ sessions, (raw) => provider.resolve(raw), ); - const tabs = createTabs(provider, runtime, spawn, setup); + const tabs = createTabs( + provider, + runtime, + transcripts(spawn, runtime), + setup, + ); const props = (settings = defaults, save = vi.fn(async () => {})) => ({ agent: { @@ -1291,7 +1296,7 @@ it("installs and signs in Buzz's copy of Codex from its provider settings", asyn const { ProviderDialogBody } = createTabs( provider, f.runtime, - f.spawn, + transcripts(f.spawn, f.runtime), f.setup, ); render(); diff --git a/src/bundled/codex/tabs.tsx b/src/bundled/codex/tabs.tsx index 685b7532b2..82d8164a96 100644 --- a/src/bundled/codex/tabs.tsx +++ b/src/bundled/codex/tabs.tsx @@ -52,7 +52,12 @@ import { import { AppServer, listModels, needsSignIn, type Spawn } from "./rpc"; import type { CodexRuntime, SavedConversation } from "./runtime"; import type { CodexSetup } from "./setup"; -import { readTranscript, TURN_LIMIT, type CodexTranscript } from "./transcript"; +import { + readTranscript, + TURN_LIMIT, + type CodexTranscript, + type Lookup, +} from "./transcript"; const label = ({ channelId, name, root }: SavedConversation) => `${name ? `#${name}` : channelId.slice(0, 8)} · ${root ? `thread ${root.slice(0, 8)}` : "channel"}`; @@ -310,7 +315,7 @@ function CodexSignIn({ check, disabled, }: { - login: CodexSetup; + login: SetupHandle; step: "install" | "login"; checking: boolean; check(): void; @@ -412,11 +417,39 @@ function CodexSignIn({ ); } +// What the tabs reach of the runtime: stores and calls returning promises, +// as an agent type's `expose` allows. +type ProviderHandle = Pick< + CodexProvider, + "snapshot" | "subscribe" | "check" | "resolve" | "saveDefaults" +>; +type ActivityHandle = Pick< + CodexRuntime, + "subscribe" | "sessions" | "conversations" | "find" +>; +type SetupHandle = Pick< + CodexSetup, + "snapshot" | "subscribe" | "run" | "cancel" +>; +/** Reads one conversation's Codex thread. */ +export type TranscriptRead = ( + agent: Lookup["agent"], + conversation: SavedConversation, + signal: AbortSignal, +) => Promise; +/** Transcripts read with `spawn`, named by the runtime. */ +export const transcripts = + (spawn: Spawn, runtime: Pick): TranscriptRead => + (agent, conversation, signal) => + readTranscript(spawn, { agent, ...conversation }, signal, (pubkey) => + runtime.name(agent.key, pubkey), + ); + export function createTabs( - provider: CodexProvider, - runtime: CodexRuntime, - spawn: Spawn, - setup: CodexSetup, + provider: ProviderHandle, + runtime: ActivityHandle, + transcript: TranscriptRead, + setup: SetupHandle, ) { const login = setup; @@ -738,11 +771,10 @@ export function createTabs( useEffect(() => { const abort = new AbortController(); setLoading(true); - void readTranscript( - spawn, - { agent: { key: agent.key, name: agent.name }, ...conversation }, + void transcript( + { key: agent.key, name: agent.name }, + conversation, abort.signal, - (pubkey) => runtime.name(agent.key, pubkey), ) .then( (transcript): Read => diff --git a/src/bundled/github/index.tsx b/src/bundled/github/index.tsx index 8077fac3cd..9b3d01968a 100644 --- a/src/bundled/github/index.tsx +++ b/src/bundled/github/index.tsx @@ -155,8 +155,11 @@ export const apply: PluginModule["apply"] = (ctx) => { : "Unlinked this chat's pull request."; }, }); - // Agents can watch pull requests where the app runs them. - ctx.inject(["agents", "host"], watchPulls); +}; +// Agents can watch pull requests where the app runs them. +export const agent: NonNullable = { + inject: ["agents", "host"], + apply: watchPulls, }; const labels = { repository: "Repository", diff --git a/src/bundled/goose/index.test.ts b/src/bundled/goose/index.test.ts index 7fcb83b261..213827209f 100644 --- a/src/bundled/goose/index.test.ts +++ b/src/bundled/goose/index.test.ts @@ -1,8 +1,8 @@ // @vitest-environment jsdom import { afterEach, expect, it, vi } from "vitest"; -import type { AgentType } from "../../features/agents/service"; import { GooseRuntime } from "./runtime"; -import { apply } from "./index"; +import * as goose from "./index"; +import { appliedType } from "../../features/agents/plugin-testing"; vi.mock("@tauri-apps/api/core", () => ({ isTauri: () => true })); afterEach(() => vi.restoreAllMocks()); @@ -22,18 +22,12 @@ it("observes saved conversations with the status of open sessions", () => { // A one-off turn joins no conversation. { key: "timer/x #1", busy: true }, ]); - let type: AgentType | undefined; - apply({ + const type = appliedType(goose, { pluginOwner: { id: "buzz.goose", revision: "r1" }, host: { request: vi.fn() }, - agents: { - register: (registered: AgentType) => { - type = registered; - }, - relay: () => undefined, - }, + agents: { relay: () => undefined }, effect: vi.fn(), - } as never); + }); // A reload stops its turns, so the owner is asked to resume them. expect(type?.resumes).toBe(true); expect( diff --git a/src/bundled/goose/index.ts b/src/bundled/goose/index.ts index 25053b0bbb..55f7008cb0 100644 --- a/src/bundled/goose/index.ts +++ b/src/bundled/goose/index.ts @@ -15,9 +15,52 @@ import { readUsage } from "./usage"; export const TYPE_ID = "goose"; -export const inject = ["agents", "host"]; +/** What the type's UI reaches of its runtime. */ +type GooseExposed = Parameters[0]; + +export const inject = ["agents"]; export const apply: PluginModule["apply"] = (ctx) => { // Only the desktop app can start `goose-acp`; no unusable type elsewhere. + if (!isTauri()) return; + ctx.agents.present(TYPE_ID, (exposed) => { + const { + Create, + CreateModel, + SettingsTab, + ActivityTab, + DefaultControl, + ProviderDialogBody, + } = createTabs(exposed); + return { + icon: GooseLogoIcon, + Create, + CreateModel, + provider: { DialogBody: ProviderDialogBody, DefaultControl }, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + splitSettings: true, + component: SettingsTab, + }, + { + id: "activity", + title: "Activity", + role: "activity", + component: ActivityTab, + }, + ], + }; + }); +}; + +// The runtime, which runs where the agents do. +export const agent: NonNullable = { + inject: ["agents", "host"], + apply: (ctx) => applyAgent(ctx), +}; +function applyAgent(ctx: Parameters[0]) { if (!isTauri()) return; const { agents, host } = ctx; const typeKey = `${ctx.pluginOwner?.id ?? "buzz.goose"}/${TYPE_ID}`; @@ -38,18 +81,6 @@ export const apply: PluginModule["apply"] = (ctx) => { : undefined; const setup = new GooseSetup(desktop, () => runtime.defaultsChanged()); const history = new GooseHistory(desktop); - const { - Create, - CreateModel, - SettingsTab, - ActivityTab, - DefaultControl, - ProviderDialogBody, - } = createTabs({ - setup, - runtime, - history, - }); const sync = () => { const snapshot = agents.snapshot(); if (snapshot.status === "loading") return; @@ -67,7 +98,6 @@ export const apply: PluginModule["apply"] = (ctx) => { agents.register({ id: TYPE_ID, title: "Goose", - icon: GooseLogoIcon, description: "goose on this computer, with any provider goose supports. Each thread it is mentioned in is one continuing session.", defaults: () => ({ config: DEFAULT_CONFIG }), @@ -79,14 +109,11 @@ export const apply: PluginModule["apply"] = (ctx) => { provider: source.provider ?? "", } : undefined, - Create, - CreateModel, + expose: { setup, runtime, history } satisfies GooseExposed, provider: { snapshot: setup.snapshot, subscribe: setup.subscribe, check: setup.check, - DialogBody: ProviderDialogBody, - DefaultControl, }, adopt: (carry, saved) => ({ ...saved, @@ -141,22 +168,7 @@ export const apply: PluginModule["apply"] = (ctx) => { signal, ), }), - tabs: [ - { - id: "settings", - title: "Settings", - role: "settings", - splitSettings: true, - component: SettingsTab, - }, - { - id: "activity", - title: "Activity", - role: "activity", - component: ActivityTab, - }, - ], models: (agent) => setup.picks(config(agent.config)), ...gooseRun(runtime), }); -}; +} diff --git a/src/bundled/goose/tabs.tsx b/src/bundled/goose/tabs.tsx index e5440198df..51c3a29a60 100644 --- a/src/bundled/goose/tabs.tsx +++ b/src/bundled/goose/tabs.tsx @@ -132,14 +132,30 @@ const placeOf = (key: string): Place => { const titled = (name: string, title: string) => title.startsWith(`${name} · `) ? title.slice(name.length + 3) : title; +// What the tabs reach of the runtime: stores and calls returning promises, +// as an agent type's `expose` allows. +type SetupHandle = Pick< + GooseSetup, + | "snapshot" + | "subscribe" + | "check" + | "providers" + | "models" + | "efforts" + | "saved" + | "save" + | "saveDefaults" + | "test" +>; + export function createTabs({ setup, runtime, history, }: Readonly<{ - setup: GooseSetup; - runtime: GooseRuntime; - history: GooseHistory; + setup: SetupHandle; + runtime: Pick; + history: Pick; }>) { function DefaultControl({ onConfigure }: { onConfigure(): void }) { const state = useSyncExternalStore(setup.subscribe, setup.snapshot); diff --git a/src/bundled/pi/index.test.ts b/src/bundled/pi/index.test.ts new file mode 100644 index 0000000000..f3eeff49c7 --- /dev/null +++ b/src/bundled/pi/index.test.ts @@ -0,0 +1,23 @@ +// @vitest-environment jsdom +import { expect, it, vi } from "vitest"; +import { appliedType } from "../../features/agents/plugin-testing"; +import * as pi from "./index"; + +vi.mock("@tauri-apps/api/core", () => ({ isTauri: () => true })); + +it("shows the Pi type's screens from the runtime its agent entry registers", () => { + // The UI needs no host; only the runtime starts Pi. + expect(pi.inject).toEqual(["agents"]); + expect(pi.agent.inject).toEqual(["agents", "host"]); + const type = appliedType(pi, { + pluginOwner: { id: "buzz.pi", revision: "r1" }, + host: { request: vi.fn() }, + agents: { relay: () => undefined }, + effect: vi.fn(), + }); + expect(type).toMatchObject({ id: "pi", title: "Pi", timeoutMs: 120_000 }); + expect(type?.Create).toBeTypeOf("function"); + expect(type?.CreateModel).toBeTypeOf("function"); + expect(type?.icon).toBeDefined(); + expect(type?.tabs?.map((tab) => tab.id)).toEqual(["settings", "activity"]); +}); diff --git a/src/bundled/pi/index.ts b/src/bundled/pi/index.ts index 6449c7e243..b816450be9 100644 --- a/src/bundled/pi/index.ts +++ b/src/bundled/pi/index.ts @@ -16,9 +16,45 @@ export const TYPE_ID = "pi"; * which includes the first install of Pi's packages. */ const HANDOVER_MS = 2 * 60_000; -export const inject = ["agents", "host"]; +/** What the type's UI reaches of its runtime. */ +type PiExposed = Parameters[0]; + +export const inject = ["agents"]; export const apply: PluginModule["apply"] = (ctx) => { // Only the desktop app can start the Pi host; no unusable type elsewhere. + if (!isTauri()) return; + ctx.agents.present(TYPE_ID, (exposed) => { + const { ActivityTab, Create, CreateModel, SettingsTab } = + createTabs(exposed); + return { + icon: PiLogoIcon, + Create, + CreateModel, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + splitSettings: true, + component: SettingsTab, + }, + { + id: "activity", + title: "Activity", + role: "activity", + component: ActivityTab, + }, + ], + }; + }); +}; + +// The runtime, which runs where the agents do. +export const agent: NonNullable = { + inject: ["agents", "host"], + apply: (ctx) => applyAgent(ctx), +}; +function applyAgent(ctx: Parameters[0]) { if (!isTauri()) return; const { agents, host } = ctx; const typeKey = `${ctx.pluginOwner?.id ?? "buzz.pi"}/${TYPE_ID}`; @@ -30,10 +66,6 @@ export const apply: PluginModule["apply"] = (ctx) => { repositoriesContext, (key) => agents.handle(key), ); - const { ActivityTab, Create, CreateModel, SettingsTab } = createTabs({ - setup: runtime.setup, - runtime, - }); const sync = () => { const snapshot = agents.snapshot(); if (snapshot.status === "loading") return; @@ -51,12 +83,10 @@ export const apply: PluginModule["apply"] = (ctx) => { agents.register({ id: TYPE_ID, title: "Pi", - icon: PiLogoIcon, description: "Pi on this computer, with any provider Pi supports. Each thread it is mentioned in is one continuing conversation.", defaults: () => ({ config: DEFAULT_CONFIG }), - Create, - CreateModel, + expose: { setup: runtime.setup, runtime } satisfies PiExposed, adopt: (carry, saved) => ({ ...saved, instructions: carry.instructions ?? saved.instructions, @@ -81,21 +111,6 @@ export const apply: PluginModule["apply"] = (ctx) => { list: (key) => runtime.sessions(key), subscribe: (listener) => runtime.subscribe(listener), }, - tabs: [ - { - id: "settings", - title: "Settings", - role: "settings", - splitSettings: true, - component: SettingsTab, - }, - { - id: "activity", - title: "Activity", - role: "activity", - component: ActivityTab, - }, - ], sessionControls: { scope: (agent) => config(agent.config).scope, state: (agent, place) => runtime.sessionState(agent, place), @@ -107,4 +122,4 @@ export const apply: PluginModule["apply"] = (ctx) => { runOnce: (turn) => runtime.runOnce(turn), timeoutMs: HANDOVER_MS, }); -}; +} diff --git a/src/bundled/pi/tabs.tsx b/src/bundled/pi/tabs.tsx index aabc413304..cc4c24075b 100644 --- a/src/bundled/pi/tabs.tsx +++ b/src/bundled/pi/tabs.tsx @@ -182,7 +182,7 @@ export function createTabs({ setup, runtime, }: Readonly<{ - setup: PiSetup; + setup: Pick; runtime: Pick; }>) { /** The provider, its key and its model, and a test that they answer. As diff --git a/src/features/agents/link.test.ts b/src/features/agents/link.test.ts new file mode 100644 index 0000000000..97d3ef5c5c --- /dev/null +++ b/src/features/agents/link.test.ts @@ -0,0 +1,191 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { type Json, type Link, link, type LinkPort, type Served } from "./link"; + +const open: Link[] = []; +afterEach(() => { + for (const side of open.splice(0)) side.close(); +}); + +/** A store a topic serves, as a side's own state would be. */ +function store(value: T) { + const listeners = new Set<() => void>(); + return { + get: () => value, + set(next: T) { + value = next; + for (const listener of [...listeners]) listener(); + }, + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + }; +} + +/** The two ends of one channel, as two pages are. */ +function pages() { + const channel = new MessageChannel(); + const side = (port: LinkPort, served?: Served) => { + const one = link(port, served); + open.push(one); + return one; + }; + return { + first: (served?: Served) => side(channel.port1, served), + second: (served?: Served) => side(channel.port2, served), + }; +} + +it("sends a watched topic's value, and again only when it changes", async () => { + const count = store(1); + const { first, second } = pages(); + first({ topics: { count: () => count } }); + const runner = second(); + const heard = vi.fn(); + runner.watch("count", null, heard); + await vi.waitFor(() => expect(runner.read("count")).toBe(1)); + heard.mockClear(); + count.set(2); + count.set(3); + await vi.waitFor(() => expect(runner.read("count")).toBe(3)); + // Two changes in one task are one message. + expect(heard).toHaveBeenCalledTimes(1); + count.set(3); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(heard).toHaveBeenCalledTimes(1); +}); + +it("serves a topic for each set of arguments, and stops when nobody watches", async () => { + const subscribed = vi.fn(); + const stopped = vi.fn(); + const { first, second } = pages(); + first({ + topics: { + double: (args) => ({ + get: () => (args as number) * 2, + subscribe() { + subscribed(args); + return () => stopped(args); + }, + }), + }, + }); + const client = second(); + const stop = client.watch("double", 2, () => {}); + const again = client.watch("double", 2, () => {}); + client.watch("double", 5, () => {}); + await vi.waitFor(() => { + expect(client.read("double", 2)).toBe(4); + expect(client.read("double", 5)).toBe(10); + }); + expect(subscribed.mock.calls).toEqual([[2], [5]]); + stop(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(stopped).not.toHaveBeenCalled(); + again(); + await vi.waitFor(() => expect(stopped).toHaveBeenCalledWith(2)); + expect(client.read("double", 2)).toBeUndefined(); +}); + +it("answers a command with its value or its error", async () => { + const { first, second } = pages(); + first({ + commands: { + add: (args) => { + const [a, b] = args as [number, number]; + return a + b; + }, + fail: async () => { + throw new Error("No such agent on this device"); + }, + }, + }); + const client = second(); + await expect(client.call("add", [2, 3])).resolves.toBe(5); + await expect(client.call("fail")).rejects.toThrow( + "No such agent on this device", + ); + await expect(client.call("missing")).rejects.toThrow( + "The agents link has no command missing", + ); +}); + +it("either side serves, so each can watch and call the other", async () => { + const records = store({ a: 1 }); + const status = store("idle"); + const { first, second } = pages(); + const manager = first({ topics: { records: () => records } }); + const runner = second({ + topics: { status: () => status }, + commands: { ping: () => "pong" }, + }); + manager.watch("status", null, () => {}); + runner.watch("records", null, () => {}); + await vi.waitFor(() => { + expect(manager.read("status")).toBe("idle"); + expect(runner.read("records")).toEqual({ a: 1 }); + }); + await expect(manager.call("ping")).resolves.toBe("pong"); +}); + +/** Two ends whose second page can be replaced while the first keeps going. */ +function reloading() { + let to: MessagePort | undefined; + const first: LinkPort = { + onmessage: null, + postMessage: (message) => to?.postMessage(message), + close() {}, + }; + const load = (served?: Served) => { + const channel = new MessageChannel(); + to = channel.port1; + channel.port1.onmessage = (event) => first.onmessage?.(event); + const page = link(channel.port2, served); + open.push(page); + return page; + }; + return { first, load }; +} + +it("watches again, and ends its waiting calls, when the other page reloads", async () => { + const { first, load } = reloading(); + const manager = link(first); + open.push(manager); + let release: (value: Json) => void = () => {}; + const old = load({ + topics: { status: () => store("old") }, + commands: { slow: () => new Promise((done) => (release = done)) }, + }); + manager.watch("status", null, () => {}); + await vi.waitFor(() => expect(manager.read("status")).toBe("old")); + const waiting = manager.call("slow"); + await new Promise((resolve) => setTimeout(resolve, 10)); + load({ topics: { status: () => store("new") } }); + await expect(waiting).rejects.toThrow("The agents page reloaded"); + await vi.waitFor(() => expect(manager.read("status")).toBe("new")); + // The old page's late answer reaches nobody. + old.close(); + release(1); +}); + +it("does not answer a call for a page that has since reloaded", async () => { + const { first, load } = reloading(); + let release: (value: Json) => void = () => {}; + const runner = link(first, { + commands: { + slow: () => new Promise((done) => (release = done)), + fast: () => "fast", + }, + }); + open.push(runner); + const old = load(); + void old.call("slow").catch(() => {}); + await new Promise((resolve) => setTimeout(resolve, 10)); + const page = load(); + // The new page's first call has the same id as the old page's. + const answer = page.call("fast"); + release("slow"); + await expect(answer).resolves.toBe("fast"); +}); diff --git a/src/features/agents/link.ts b/src/features/agents/link.ts new file mode 100644 index 0000000000..54f47ff5f4 --- /dev/null +++ b/src/features/agents/link.ts @@ -0,0 +1,254 @@ +// The link between the agents' two owners: the Manager, which the screens use, +// and the Runner, which runs agents. They may run in different pages, so only +// JSON data crosses: no functions, no shared objects. Each side serves named +// topics, values the other side watches and is sent again when they change, +// and named commands, which answer one call. Either side may serve both. +// +// Either page may reload. Each side says hello with a new instance when it +// starts. A side that hears a new instance from its peer watches again what +// it watched, ends the calls it was waiting on, and stops serving what the +// old peer watched. + +export type Json = + | null + | boolean + | number + | string + | readonly Json[] + | { readonly [key: string]: Json }; + +/** A `MessagePort`, or a link to a page in another webview. It delivers each + * message once, in the order sent. */ +export type LinkPort = { + onmessage: ((event: MessageEvent) => void) | null; + postMessage(message: unknown): void; + close(): void; +}; + +/** A value the other side can watch, for one set of arguments. */ +export type Topic = (args: Json) => Readonly<{ + get(): Json; + subscribe(listener: () => void): () => void; +}>; +export type Command = (args: Json) => Json | Promise; +export type Served = Readonly<{ + topics?: Readonly>; + commands?: Readonly>; +}>; + +type Frame = + | Readonly<{ kind: "hello"; instance: string }> + | Readonly<{ + kind: "watch"; + from: string; + key: string; + topic: string; + args: Json; + }> + | Readonly<{ kind: "unwatch"; from: string; key: string }> + | Readonly<{ + kind: "call"; + from: string; + id: number; + command: string; + args: Json; + }> + // What a side sends for a watch or call names the instance it is for, so a + // page that reloaded does not take what was meant for its last load. + | Readonly<{ kind: "value"; for: string; key: string; value: Json }> + | Readonly<{ kind: "result"; for: string; id: number; value: Json }> + | Readonly<{ kind: "error"; for: string; id: number; message: string }>; + +const keyOf = (topic: string, args: Json) => `${topic}:${JSON.stringify(args)}`; +const messageOf = (error: unknown) => + error instanceof Error ? error.message : String(error); + +export type Link = ReturnType; + +/** One side of the link over `port`, serving `served` to the other side. */ +export function link(port: LinkPort, served: Served = {}) { + const instance = crypto.randomUUID(); + let peer: string | undefined; + let closed = false; + const post = (frame: Frame) => { + if (!closed) port.postMessage(frame); + }; + + // What this side watches of the other's. + const watched = new Map< + string, + { + topic: string; + args: Json; + value: Json | undefined; + listeners: Set<() => void>; + } + >(); + // What this side serves to the other's watches. + const serving = new Map< + string, + Readonly<{ watcher: string; stop(): void }> + >(); + const calls = new Map< + number, + Readonly<{ resolve(value: Json): void; reject(error: Error): void }> + >(); + let next = 0; + + const stopServing = () => { + for (const { stop } of serving.values()) stop(); + serving.clear(); + }; + const serve = (watcher: string, key: string, name: string, args: Json) => { + serving.get(key)?.stop(); + const topic = served.topics?.[name]; + if (!topic) { + console.warn(`The agents link has no topic ${name}`); + return; + } + const source = topic(args); + let sent: string | undefined; + let queued = false; + const send = () => { + queued = false; + if (serving.get(key) !== entry) return; + const value = source.get(); + const json = JSON.stringify(value); + // Several changes in one task are one message, and none is sent for a + // change that leaves the value as it was. + if (json === sent) return; + sent = json; + post({ kind: "value", for: watcher, key, value }); + }; + const unsubscribe = source.subscribe(() => { + if (queued) return; + queued = true; + queueMicrotask(send); + }); + const entry = Object.freeze({ + watcher, + stop() { + unsubscribe(); + if (serving.get(key) === entry) serving.delete(key); + }, + }); + serving.set(key, entry); + send(); + }; + const answer = async ( + asker: string, + id: number, + name: string, + args: Json, + ) => { + try { + const command = served.commands?.[name]; + if (!command) throw new Error(`The agents link has no command ${name}`); + const value = (await command(args)) ?? null; + post({ kind: "result", for: asker, id, value }); + } catch (error) { + post({ kind: "error", for: asker, id, message: messageOf(error) }); + } + }; + // The other side reloaded or started: what it watched and was owed is + // gone with its last page. + const meet = (other: string) => { + const reloaded = peer !== undefined; + peer = other; + // It may not have heard this side's hello. + post({ kind: "hello", instance }); + // What was sent before the first hello reached the page that answers it; + // what was sent to a page that reloaded did not. + if (!reloaded) return; + stopServing(); + for (const pending of calls.values()) + pending.reject(new Error("The agents page reloaded")); + calls.clear(); + for (const [key, { topic, args }] of watched) + post({ kind: "watch", from: instance, key, topic, args }); + }; + + port.onmessage = ({ data }: MessageEvent) => { + if (closed) return; + switch (data.kind) { + case "hello": + if (data.instance !== peer) meet(data.instance); + return; + case "watch": + serve(data.from, data.key, data.topic, data.args); + return; + case "unwatch": { + const entry = serving.get(data.key); + if (entry?.watcher === data.from) entry.stop(); + return; + } + case "value": { + if (data.for !== instance) return; + const entry = watched.get(data.key); + if (!entry) return; + entry.value = data.value; + for (const listener of [...entry.listeners]) listener(); + return; + } + case "call": + void answer(data.from, data.id, data.command, data.args); + return; + case "result": + if (data.for !== instance) return; + calls.get(data.id)?.resolve(data.value); + calls.delete(data.id); + return; + case "error": + if (data.for !== instance) return; + calls.get(data.id)?.reject(new Error(data.message)); + calls.delete(data.id); + return; + } + }; + post({ kind: "hello", instance }); + + return { + /** The value of the other side's `topic`, as last sent, while watched. */ + read(topic: string, args: Json = null): Json | undefined { + return watched.get(keyOf(topic, args))?.value; + }, + /** Watches the other side's `topic` until the returned stop. */ + watch(topic: string, args: Json, listener: () => void) { + const key = keyOf(topic, args); + let entry = watched.get(key); + if (!entry) { + entry = { topic, args, value: undefined, listeners: new Set() }; + watched.set(key, entry); + post({ kind: "watch", from: instance, key, topic, args }); + } + const listeners = entry.listeners; + listeners.add(listener); + return () => { + listeners.delete(listener); + if (listeners.size || watched.get(key)?.listeners !== listeners) return; + watched.delete(key); + post({ kind: "unwatch", from: instance, key }); + }; + }, + /** Calls the other side's `command`. */ + call(command: string, args: Json = null): Promise { + if (closed) return Promise.reject(new Error("The agents link is closed")); + const id = ++next; + return new Promise((resolve, reject) => { + calls.set(id, { resolve, reject }); + post({ kind: "call", from: instance, id, command, args }); + }); + }, + close() { + if (closed) return; + closed = true; + stopServing(); + for (const pending of calls.values()) + pending.reject(new Error("The agents link is closed")); + calls.clear(); + watched.clear(); + port.onmessage = null; + port.close(); + }, + }; +} diff --git a/src/features/agents/plugin-testing.ts b/src/features/agents/plugin-testing.ts new file mode 100644 index 0000000000..6de02adcbc --- /dev/null +++ b/src/features/agents/plugin-testing.ts @@ -0,0 +1,24 @@ +import type { PluginModule } from "../../plugins/api"; +import { presented, type AgentPresentation, type AgentType } from "./service"; + +/** Starts a harness plugin's agent entry, then its UI entry, on `ctx`, and + * returns the type it registered as the service shows it, with what it + * presents. Undefined when it registered none. */ +export function appliedType( + module: Pick, + ctx: Readonly<{ agents: object }> & Record, +): AgentType | undefined { + let type: AgentType | undefined; + let present: ((exposed: unknown) => AgentPresentation) | undefined; + Object.assign(ctx.agents, { + register: (registered: AgentType) => { + type = registered; + }, + present: (_id: string, shown: typeof present) => { + present = shown; + }, + }); + void module.agent?.apply(ctx as never); + void module.apply(ctx as never); + return type && present ? presented(type, present(type.expose)) : type; +} diff --git a/src/features/agents/records.test.ts b/src/features/agents/records.test.ts new file mode 100644 index 0000000000..863d58a4a1 --- /dev/null +++ b/src/features/agents/records.test.ts @@ -0,0 +1,97 @@ +import { expect, it, vi } from "vitest"; +import { agentKey } from "./membership"; +import { AgentRecords, type Member } from "./records"; +import { recordKey } from "./store"; +import { memoryStorage } from "./test-fakes"; + +const pubkey = "a".repeat(64); +const home: Member = { + key: pubkey, + pubkey, + relay: "wss://home.example", + owner: "b".repeat(64), + type: "test/agent", + name: "Ada", + deleted: false, +}; +const away: Member = { + ...home, + key: agentKey(pubkey, "https://away.example", "https://home.example"), + relay: "wss://away.example", + home: "wss://home.example", +}; + +function records() { + const storage = memoryStorage(); + const changed = vi.fn(); + const settled = vi.fn(); + const owner = new AgentRecords(storage, { + defaults: ({ pubkey }) => ({ + pubkey, + attention: {}, + config: { model: "a" }, + }), + changed, + settled, + }); + return { storage, owner, changed, settled }; +} + +it("keeps config in the home record and the rest in the membership's own", async () => { + const { storage, owner, settled } = records(); + await owner.edit(away.key, { + kind: "settings", + member: away, + change: { config: { model: "b" }, attentionEnabled: true }, + }); + expect(owner.saved(pubkey)?.config).toEqual({ model: "b" }); + expect(owner.saved(away.key)).toMatchObject({ + pubkey: away.key, + attentionEnabled: true, + config: undefined, + }); + expect( + JSON.parse(storage.get(recordKey(away.key)) ?? "{}"), + ).not.toHaveProperty("config"); + // The membership sees its agent's config. + expect(owner.recordOf(away).config).toEqual({ model: "b" }); + expect(settled).toHaveBeenCalledWith(pubkey); + expect(settled).toHaveBeenCalledWith(away.key); +}); + +it("applies each edit to the record the one before it saved", async () => { + const { owner, changed } = records(); + const created = owner.edit(pubkey, { + kind: "create", + record: { pubkey, attention: {}, config: { model: "c" } }, + }); + const enabled = owner.edit(pubkey, { + kind: "settings", + member: home, + change: { attentionEnabled: true }, + }); + await Promise.all([created, enabled]); + expect(owner.saved(pubkey)).toMatchObject({ + config: { model: "c" }, + attentionEnabled: true, + }); + expect(changed).toHaveBeenCalledTimes(2); +}); + +it("forgets attention when the agent leaves, and the record when it goes", async () => { + const { storage, owner } = records(); + await owner.edit(pubkey, { + kind: "settings", + member: home, + change: { attentionEnabled: true }, + }); + await owner.edit(pubkey, { kind: "leave" }); + expect(owner.saved(pubkey)).toEqual({ + pubkey, + attention: {}, + config: { model: "a" }, + }); + await owner.edit(pubkey, { kind: "remove" }); + expect(owner.saved(pubkey)).toBeUndefined(); + expect(storage.get(recordKey(pubkey))).toBeNull(); +}); diff --git a/src/features/agents/records.ts b/src/features/agents/records.ts new file mode 100644 index 0000000000..77bdab3d0f --- /dev/null +++ b/src/features/agents/records.ts @@ -0,0 +1,335 @@ +// The agents' saved settings: one record for each membership, and the one +// queue each record's changes take. The Runner, which runs agents, owns them +// and is their only writer. The Manager, which the screens use, asks for its +// changes as `RecordEdit` data, so they can later cross to the Runner's page. +import { parseSlug, type AttentionObject } from "./attention"; +import { pubkeyOf, type AgentKey } from "./membership"; +import type { AgentIdentity } from "./native"; +import type { AgentChange } from "./service"; +import type { AgentStore } from "./state"; +import { + readRecords, + recordKey, + removeRecord, + setAttention, + splitLegacyRecords, + withProgress, + writeRecord, + type AgentRecord, + type TimerProgress, +} from "./store"; + +export type Member = AgentIdentity & Readonly<{ key: AgentKey }>; + +/** A change the Manager asks of a membership's record. */ +export type RecordEdit = + /** A new agent's first settings. */ + | Readonly<{ kind: "create"; record: AgentRecord }> + /** The owner's change to its config or attention. Its config is its home + * record's, in every community; the rest is this membership's own. */ + | Readonly<{ + kind: "settings"; + member: Member; + change: Pick< + AgentChange, + "config" | "attentionEnabled" | "attention" | "restart" + >; + }> + /** It left this community: its home record keeps the agent's settings, + * but not what it paid attention to there. */ + | Readonly<{ kind: "leave" }> + /** Its agent is gone, or it left this community. */ + | Readonly<{ kind: "remove" }>; + +export const EMPTY = Object.freeze({}); +export const blank = (key: AgentKey): AgentRecord => ({ + pubkey: key, + attention: EMPTY, + config: undefined, +}); +/** What a membership other than the agent's home keeps of `record`: its + * attention and timers. */ +const ownRecord = ( + key: AgentKey, + { config: _, ...record }: AgentRecord, +): AgentRecord => ({ ...record, pubkey: key, config: undefined }); +/** `record` with the owner's `change` to its settings; throws when the + * change is not valid for it. */ +function changed(record: AgentRecord, change: AgentChange): AgentRecord { + let next = record; + if (change.config !== undefined) next = { ...next, config: change.config }; + if (change.attentionEnabled !== undefined) { + const { attentionEnabled: _, ...rest } = next; + next = change.attentionEnabled ? { ...rest, attentionEnabled: true } : rest; + } + for (const [slug, value] of Object.entries(change.attention ?? {})) + next = setAttention( + next, + slug, + value, + undefined, + change.restart?.includes(slug) ? "restart" : undefined, + ); + // As the spec's writer: an Interest stays while anything still uses it. + for (const [slug, value] of Object.entries(change.attention ?? {})) { + const id = parseSlug(slug)?.id; + if (value !== null || !slug.startsWith("interest/") || !id) continue; + if ( + Object.values(next.attention).some( + (object) => + object.value.type !== "interest" && object.value.interest_id === id, + ) + ) + throw new Error( + "Remove or move the watches and timers that use this Interest first", + ); + } + return next; +} + +export type RecordsHost = Readonly<{ + /** The settings a new agent of this type starts with, if its type is here. */ + defaults( + identity: Pick, + ): AgentRecord | undefined; + /** A record changed and is shown. */ + changed(): void; + /** The record's changes all settled, saved or failed. */ + settled(key: AgentKey): void; +}>; + +export class AgentRecords { + private records: Readonly>; + private readonly changes = new Map>(); + private readonly changing = new Map(); + private readonly progressing = new Map(); + + constructor( + private readonly storage: AgentStore, + private readonly host: RecordsHost, + ) { + this.records = readRecords(storage); + void splitLegacyRecords(storage, this.records); + } + + all() { + return Object.values(this.records); + } + keys() { + return Object.keys(this.records); + } + saved(key: AgentKey): AgentRecord | undefined { + return this.records[key]; + } + /** A change to it is waiting or being saved. */ + isChanging(key: AgentKey) { + return this.changing.has(key); + } + + /** Applies the Manager's `edit` to the membership's record. */ + async edit(key: AgentKey, edit: RecordEdit) { + switch (edit.kind) { + case "create": + await this.change(key, () => edit.record); + return; + case "remove": + await this.change(key, () => null); + return; + case "leave": + await this.change(key, (saved) => { + if (!saved) return null; + const { + skipped: _, + timers: __, + attentionEnabled: ___, + ...rest + } = saved; + return { ...rest, attention: {} }; + }); + return; + case "settings": { + const { member, change } = edit; + const home = key === member.pubkey; + if (!home && change.config !== undefined) + await this.change(member.pubkey, (saved) => + changed(saved ?? this.settingsOf(member), { + config: change.config, + }), + ); + const own = home ? change : { ...change, config: undefined }; + if ( + own.config !== undefined || + own.attention || + own.attentionEnabled !== undefined + ) + await this.change(key, (saved) => + changed(saved ?? this.recordOf(member), own), + ); + return; + } + } + } + + /** Changes one agent's record. This is the only way a record changes: + * owner settings, the agent's attention tools, timer progress, a provider + * change and removal all send a change here. One agent's changes run in + * the order they are sent, each after the one before it is saved or has + * failed, so each one applies to the newest saved record, never to a copy + * another change replaces. `apply` gets that record (or undefined) and + * returns the next one, or null to remove it; it may throw to change + * nothing. The owner and the agent see a change once it is saved, and the + * promise settles then. When the save fails, the promise rejects and they + * keep the saved record, so a form keeps its unsaved changes. `saved` + * runs as the change is shown, before anyone is told. Only the Runner + * passes a function; the Manager sends an `edit`. */ + change( + key: AgentKey, + apply: ( + record: AgentRecord | undefined, + ) => AgentRecord | null | Promise, + saved?: (record: AgentRecord | null) => void, + ): Promise { + this.changing.set(key, (this.changing.get(key) ?? 0) + 1); + return this.queued(key, async () => { + try { + const base = this.records[key]; + const next = await apply(base); + await this.commit(key, base, next); + saved?.(next); + this.host.changed(); + return next; + } finally { + const left = (this.changing.get(key) ?? 1) - 1; + if (left) this.changing.set(key, left); + else { + this.changing.delete(key); + // Its timers waited for this change, saved or failed. + this.host.settled(key); + } + } + }); + } + /** Records that timers ran or started counting, as a change to their run + * state only. It applies to whatever the record is when its turn comes, so + * it never undoes a change sent before it: when a timer was rearmed or + * restarted meanwhile, its new deadline stays and only the occurrences it + * used are added. Timers see it at once (`timersOf`), and memory keeps it + * even when its save fails, so a timer does not run again at once. */ + progress(key: AgentKey, ran: TimerProgress) { + const pending = this.progressing.get(key) ?? []; + this.progressing.set(key, [...pending, ran]); + void this.queued(key, async () => { + try { + const base = this.records[key]; + const next = base && withProgress(base, ran); + if (!base || !next) return; + await this.commit(key, base, next).catch((error) => { + console.warn(`Agent ${key} timer state was not saved`, error); + this.records = { ...this.records, [key]: next }; + }); + this.host.changed(); + } finally { + const left = (this.progressing.get(key) ?? []).filter( + (entry) => entry !== ran, + ); + if (left.length) this.progressing.set(key, left); + else this.progressing.delete(key); + } + }); + } + /** Keeps what an attention object's last wake and its source printed, while + * the object is the same. */ + history( + key: AgentKey, + object: AttentionObject, + output?: string, + lastWokeAt?: number, + ) { + void this.queued(key, async () => { + const record = this.records[key]; + if (!record || record.attention[object.slug] !== object) return; + const prior = record.attentionHistory?.[object.slug]; + const entry = { + ...prior, + ...(lastWokeAt !== undefined ? { lastWokeAt } : {}), + ...(output !== undefined + ? { + output: [prior?.output, output] + .filter((part) => part !== undefined && part !== "") + .join(prior?.output?.endsWith("\n") ? "" : "\n") + .slice(-16384) + .split("\n") + .slice(-40) + .join("\n"), + } + : {}), + }; + await this.commit(key, record, { + ...record, + attentionHistory: { ...record.attentionHistory, [object.slug]: entry }, + }); + this.host.changed(); + }).catch((error) => console.warn("Attention history was not saved", error)); + } + /** The agent's timers' run state as its timers go by: saved, with the + * progress still being saved. */ + timersOf(key: AgentKey) { + const record = this.records[key]; + const pending = this.progressing.get(key); + if (!record || !pending) return record?.timers ?? EMPTY; + return ( + pending.reduce((next, ran) => withProgress(next, ran) ?? next, record) + .timers ?? EMPTY + ); + } + /** The agent's home record, which keeps its config: as saved or, without + * one, its type's defaults, kept in memory until a save stores them. + * Without its type there are no defaults yet. */ + settingsOf(identity: Pick) { + const { pubkey } = identity; + if (!this.records[pubkey]) { + const record = this.host.defaults(identity); + if (record) this.records = { ...this.records, [pubkey]: record }; + } + return this.records[pubkey] ?? blank(pubkey); + } + /** The membership's settings as its owner and agent see them: its own + * attention, starting with none, and the agent's config. */ + recordOf(member: Member): AgentRecord { + const settings = this.settingsOf(member); + if (member.key === member.pubkey) return settings; + const own = this.records[member.key] ?? blank(member.key); + return { ...own, config: settings.config }; + } + + /** Runs `step` after the agent's earlier changes have settled. */ + private queued(key: AgentKey, step: () => Promise) { + const run = (this.changes.get(key) ?? Promise.resolve()).then(step); + const tail = run.then( + () => {}, + () => {}, + ); + this.changes.set(key, tail); + void tail.then(() => { + if (this.changes.get(key) === tail) this.changes.delete(key); + }); + return run; + } + /** Saves `next` over `base` in the membership's own key, then keeps it. + * Only this key: others may be another program's to change. A record + * other than the agent's home record keeps no config, which is the + * agent's, never one community's. */ + private async commit( + key: AgentKey, + base: AgentRecord | undefined, + given: AgentRecord | null, + ) { + if (given === base) return; + const next = given && key !== pubkeyOf(key) ? ownRecord(key, given) : given; + if (next) await writeRecord(this.storage, next); + else if (this.storage.get(recordKey(key)) !== null) + await removeRecord(this.storage, key); + const { [key]: _, ...others } = this.records; + this.records = next ? { ...others, [key]: next } : others; + } +} diff --git a/src/features/agents/runner-link.test.ts b/src/features/agents/runner-link.test.ts new file mode 100644 index 0000000000..79ef36e7d2 --- /dev/null +++ b/src/features/agents/runner-link.test.ts @@ -0,0 +1,78 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { link, type Link } from "./link"; +import { createLocalAgentActivity } from "./local-activity"; +import { linkedActivity, serveActivity } from "./runner-link"; + +const open: Link[] = []; +afterEach(() => { + for (const side of open.splice(0)) side.close(); +}); + +function linked() { + const activity = createLocalAgentActivity(); + const channel = new MessageChannel(); + const runner = link(channel.port1, serveActivity(activity.reader)); + const manager = link(channel.port2); + open.push(runner, manager); + return { activity, source: linkedActivity(manager).source }; +} + +const request = { + scope: "wss://relay.example", + agent: "a".repeat(64), + channelId: "channel", + messageId: "message", + threadRootId: "message", +}; + +it("shows the Runner's local work, and keeps rows until they change", async () => { + const { activity, source } = linked(); + const heard = vi.fn(); + source.subscribeWork(heard); + const work = activity.begin(request); + work.started(); + work.step("Read", "notes.md"); + await vi.waitFor(() => + expect(source.work()).toMatchObject([ + { ...request, state: "working", step: { tool: "Read" } }, + ]), + ); + const rows = source.work(); + expect(Object.isFrozen(rows[0])).toBe(true); + heard.mockClear(); + // A step with the same text is no change, so nothing is sent. + work.step("Read", "notes.md"); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(heard).not.toHaveBeenCalled(); + expect(source.work()).toBe(rows); +}); + +it("stops a turn and hides an error through the Runner", async () => { + const { activity, source } = linked(); + const stop = vi.fn(async () => {}); + const running = activity.begin(request); + running.started(); + running.interruptible(stop); + await vi.waitFor(() => + expect(source.work()).toMatchObject([{ stoppable: true }]), + ); + await source.interrupt?.(request); + expect(stop).toHaveBeenCalledTimes(1); + await vi.waitFor(() => + expect(source.work()).toMatchObject([{ state: "stopped" }]), + ); + + const failed = activity.begin({ ...request, messageId: "other" }); + failed.finish({ ok: false, error: "No credits left" }); + await vi.waitFor(() => + expect(source.work()).toContainEqual( + expect.objectContaining({ messageId: "other", state: "error" }), + ), + ); + source.dismissError?.({ ...request, messageId: "other" }); + await vi.waitFor(() => + expect(source.work()).toContainEqual( + expect.objectContaining({ messageId: "other", dismissed: true }), + ), + ); +}); diff --git a/src/features/agents/runner-link.ts b/src/features/agents/runner-link.ts new file mode 100644 index 0000000000..a1c2fef57b --- /dev/null +++ b/src/features/agents/runner-link.ts @@ -0,0 +1,65 @@ +// What the Runner serves the Manager over the agents link. Each name here is +// part of the contract between the two owners, so add to it with care: only +// JSON crosses, and either side may be a page that reloads. +import type { Json, Link, Served } from "./link"; +import type { LocalAgentSource, LocalWork } from "./local-activity"; + +type WorkKey = Pick; + +const workKey = (args: Json): WorkKey => { + const { scope, agent, messageId } = args as WorkKey; + return { scope, agent, messageId }; +}; + +/** The Runner's side: the local work its agents report. */ +export function serveActivity(source: LocalAgentSource): Served { + return { + topics: { + "activity.work": () => ({ + get: () => source.work() as unknown as Json, + subscribe: (listener) => source.subscribeWork(listener), + }), + }, + commands: { + "activity.interrupt": async (args) => { + await source.interrupt?.(workKey(args)); + return null; + }, + "activity.dismissError": (args) => { + source.dismissError?.(workKey(args)); + return null; + }, + }, + }; +} + +/** The Manager's side: the Runner's local work, as the screens read it. Rows + * keep their identity until the Runner sends a change. */ +export function linkedActivity(runner: Link) { + let work: readonly LocalWork[] = Object.freeze([]); + const listeners = new Set<() => void>(); + const stop = runner.watch("activity.work", null, () => { + const rows = (runner.read("activity.work") ?? []) as unknown as LocalWork[]; + work = Object.freeze(rows.map((row) => Object.freeze(row))); + for (const listener of [...listeners]) listener(); + }); + const source: LocalAgentSource = { + work: () => work, + subscribeWork(listener) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + dismissError: (row) => + void runner + .call("activity.dismissError", workKey(row)) + .catch((error: unknown) => + console.warn("Could not hide this error", error), + ), + interrupt: async (row) => { + await runner.call("activity.interrupt", workKey(row)); + }, + }; + return { source, stop }; +} diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 3c0f1095d3..4eafa41597 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -263,6 +263,69 @@ beforeEach(() => { }); afterEach(() => vi.useRealTimers()); +it("shows a type with what its plugin presents, from the handles the type exposes", async () => { + const { service, plugin } = await setup(); + const ready = { readiness: "ready" as const }; + let attention: (() => void) | undefined; + const handles = { count: 1 }; + const shown = vi.fn((exposed: typeof handles) => ({ + icon: () => null, + tabs: [{ id: "t", title: "T", component: () => null }], + provider: { DialogBody: () => null }, + // Only what a type shows is taken from `present`. + run: () => {}, + title: "Wrong", + exposed, + })); + plugin.agents.register({ + id: "shown", + title: "Shown", + defaults: () => ({ config: {} }), + expose: handles, + provider: { + snapshot: () => ready, + subscribe: () => () => {}, + check: () => {}, + get attentionAction() { + return attention; + }, + }, + }); + const runtimeOnly = service.types().find((t) => t.key === "example/shown"); + expect(runtimeOnly?.tabs).toBeUndefined(); + plugin.agents.present("shown", shown as never); + const type = service.types().find((t) => t.key === "example/shown"); + expect(shown).toHaveBeenCalledExactlyOnceWith(handles); + expect(type).toMatchObject({ title: "Shown", tabs: [{ id: "t" }] }); + expect(type?.run).toBeUndefined(); + expect(type?.icon).toBeTypeOf("function"); + expect(type?.provider?.DialogBody).toBeTypeOf("function"); + // The runtime's provider stays live, and readers see one stable type. + expect(type?.provider?.attentionAction).toBeUndefined(); + attention = () => {}; + expect(type?.provider?.attentionAction).toBe(attention); + expect(service.types().find((t) => t.key === "example/shown")).toBe(type); + expect(shown).toHaveBeenCalledOnce(); +}); + +it("still runs a type whose present throws", async () => { + const { service, plugin } = await setup(); + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + plugin.agents.register({ + id: "broken", + title: "Broken", + defaults: () => ({ config: {} }), + }); + plugin.agents.present("broken", () => { + throw new Error("no UI"); + }); + expect(service.types().find((t) => t.key === "example/broken")).toMatchObject( + { title: "Broken" }, + ); + expect(error).toHaveBeenCalledOnce(); + error.mockRestore(); +}); + it("creates an agent in the community shown, with its type's defaults, and publishes its profile", async () => { const { service, native } = await setup(); const agent = await service.create({ type: "example/echo", name: " Echo " }); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index de4273857f..2f4ff25d6b 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -66,14 +66,13 @@ import { threadReference } from "../relay/thread-reference"; import { channelTabs } from "../../bundled/channels/useChannelTabState"; import type { RelayData } from "../relay/service"; import type { RelaySession } from "../relay/session"; -import { - parseSlug, - type AttentionObject, - type AttentionValue, - type EventWatch, - type Interest, - type TimerState, - type TimerWatch, +import type { + AttentionObject, + AttentionValue, + EventWatch, + Interest, + TimerState, + TimerWatch, } from "./attention"; import type { ClassifierAvailability, @@ -106,22 +105,16 @@ import { type PortSource, } from "./native"; import { agentKey, pubkeyOf, type AgentKey } from "./membership"; +import { AgentRecords, EMPTY, type Member } from "./records"; import { readArchiveUnchecked, - readRecords, - recordKey, readTeams, readUnfinished, setAttention, writeArchiveUnchecked, - removeRecord, - splitLegacyRecords, - writeRecord, writeTeams, writeUnfinished, - withProgress, type AgentRecord, - type TimerProgress, type SkippedObject, type Team, type Unfinished, @@ -601,8 +594,9 @@ export type AgentProvider = Readonly<{ check(): void | Promise; /** Handles the current attention state directly; absent opens DialogBody. */ attentionAction?: (() => void) | undefined; - /** Provider-owned recovery/progress controls and optional saved defaults. */ - DialogBody: ComponentType<{ onClose(): void }>; + /** Provider-owned recovery/progress controls and optional saved defaults, + * from the type's `present`. */ + DialogBody?: ComponentType<{ onClose(): void }>; /** Compact device default picker; persistence remains provider-owned. */ DefaultControl?: ComponentType<{ onConfigure(): void }>; }>; @@ -708,7 +702,46 @@ export type AgentType = { /** Makes the type hosted: its provider runs each agent, so it has no `run`, * `models` or `timeoutMs`, and its agents have no attention. */ remote?: RemoteAgentType; + /** Handles for the type's own UI, given to its plugin's `present`. Each is + * JSON, a store (`snapshot` and `subscribe`) or a call returning a promise, + * so that agents can later run in another page than their UI. */ + expose?: unknown; }; +/** What an agent type shows, from its plugin's UI entry (`present`); the + * rest of the type comes from its agent entry (`register`). */ +export type AgentPresentation = Pick< + AgentType, + "icon" | "Peek" | "Create" | "CreateModel" | "tabs" +> & + Readonly<{ + provider?: Pick; + }>; +const PRESENTED = ["icon", "Peek", "Create", "CreateModel", "tabs"] as const; +/** The type as its readers see it: `type` with what `present` shows. */ +export function presented( + type: AgentType, + shown: AgentPresentation, +): AgentType { + const merged: AgentType = { ...type }; + for (const field of PRESENTED) + if (shown[field] !== undefined) + Object.assign(merged, { [field]: shown[field] }); + if (type.provider) { + // Keep getters, such as `attentionAction`, live. + const provider = Object.defineProperties( + {}, + Object.getOwnPropertyDescriptors(type.provider), + ) as AgentProvider; + if (shown.provider?.DialogBody) + Object.assign(provider, { DialogBody: shown.provider.DialogBody }); + if (shown.provider?.DefaultControl) + Object.assign(provider, { + DefaultControl: shown.provider.DefaultControl, + }); + merged.provider = provider; + } + return merged; +} /** A hosted type's provider: it holds each agent's key and runs it. The app * calls these in order to create one (`provision`, then `activate`) and owns * the owner's agent list; every step is asked again after a failure, so each @@ -767,6 +800,10 @@ export type RemoteAgentType = Readonly<{ ): Promise; }>; export type RegisteredAgentType = Contribution; +type Presentation = Readonly<{ + id: string; + present(exposed: unknown): AgentPresentation; +}>; export type RetypePreview = Readonly<{ /** Destination settings after applying the target type's migration policy. */ config: unknown; @@ -901,6 +938,12 @@ export type Agents = { configExample?: Readonly>; }>[]; register(type: AgentType): void; + /** Supplies what this plugin's type `typeId` shows, from its UI entry. + * `present` gets the type's `expose`. */ + present( + typeId: string, + present: (exposed: Exposed) => AgentPresentation, + ): void; types(): readonly RegisteredAgentType[]; /** Adds a tool to every agent session in the app, beside the Buzz tools. * The tool acts as the agent, like the rest of a trusted plugin's code @@ -1094,51 +1137,6 @@ const RETRY_MS = 30_000; const PRESENCE_MS = 60_000; /** The most agents one presence event names. */ const PRESENCE_BATCH = 256; -const EMPTY = Object.freeze({}); -const blank = (key: AgentKey): AgentRecord => ({ - pubkey: key, - attention: EMPTY, - config: undefined, -}); -/** What a membership other than the agent's home keeps of `record`: its - * attention and timers. */ -const ownRecord = ( - key: AgentKey, - { config: _, ...record }: AgentRecord, -): AgentRecord => ({ ...record, pubkey: key, config: undefined }); -/** `record` with the owner's `change` to its settings; throws when the - * change is not valid for it. */ -function changed(record: AgentRecord, change: AgentChange): AgentRecord { - let next = record; - if (change.config !== undefined) next = { ...next, config: change.config }; - if (change.attentionEnabled !== undefined) { - const { attentionEnabled: _, ...rest } = next; - next = change.attentionEnabled ? { ...rest, attentionEnabled: true } : rest; - } - for (const [slug, value] of Object.entries(change.attention ?? {})) - next = setAttention( - next, - slug, - value, - undefined, - change.restart?.includes(slug) ? "restart" : undefined, - ); - // As the spec's writer: an Interest stays while anything still uses it. - for (const [slug, value] of Object.entries(change.attention ?? {})) { - const id = parseSlug(slug)?.id; - if (value !== null || !slug.startsWith("interest/") || !id) continue; - if ( - Object.values(next.attention).some( - (object) => - object.value.type !== "interest" && object.value.interest_id === id, - ) - ) - throw new Error( - "Remove or move the watches and timers that use this Interest first", - ); - } - return next; -} /** A new agent's settings: its type's defaults. */ function defaults(pubkey: string, type: RegisteredAgentType) { const { config, attention = {} } = type.defaults(); @@ -1215,7 +1213,6 @@ type Bound = { }; /** A saved identity, which is the agent's membership in one community, with * that membership's key. */ -type Member = AgentIdentity & Readonly<{ key: AgentKey }>; /** The origin a relay URL or community ID names, if it is a valid one. */ const originOf = (value: string) => { try { @@ -1287,6 +1284,17 @@ export class AgentsService extends Service implements Agents { * shown over the newer one. */ private keyGeneration = 0; private readonly contributions; + private readonly presentations; + /** Each registered type with what it shows, worked out once per pair. */ + private readonly presentedTypes = new WeakMap< + RegisteredAgentType, + Readonly<{ from: Presentation; type: RegisteredAgentType }> + >(); + private merged: Readonly<{ + types: readonly RegisteredAgentType[]; + shown: readonly Contribution[]; + result: readonly RegisteredAgentType[]; + }> = { types: [], shown: [], result: [] }; private readonly addedTools; /** Every added tool's name, its plugin active or still starting. */ private readonly toolNames = new Set(); @@ -1301,13 +1309,7 @@ export class AgentsService extends Service implements Agents { private identities: readonly Member[] = []; /** Each membership's settings as saved, by key, and as the owner and the * agent see them. Its home record also keeps the agent's config. */ - private records: Record; - /** Each membership's last change in its queue, until it settles. */ - private readonly changes = new Map>(); - /** How many changes other than timer progress each membership has queued. */ - private readonly changing = new Map(); - /** Each membership's timer progress that is not saved yet. */ - private readonly progressing = new Map(); + private readonly records: AgentRecords; private teamList: readonly Team[]; private state: AgentsSnapshot; /** By origin, so each agent listens in its own community, selected or not. */ @@ -1384,13 +1386,26 @@ export class AgentsService extends Service implements Agents { ) { super(ctx, "agents"); this.contributions = createContributions(ctx); + this.presentations = createContributions(ctx); this.addedTools = createContributions(ctx); this.attentionSources = createContributions(ctx); // Each changes the copy at once and logs a failed save itself. void migrateAgentStorage(storage); void migrateWatchFilters(storage); - this.records = readRecords(storage); - void splitLegacyRecords(storage, this.records); + this.records = new AgentRecords(storage, { + defaults: ({ pubkey, type }) => { + const kind = this.types().find((entry) => entry.key === type); + if (kind) + try { + return defaults(pubkey, kind); + } catch (error) { + console.warn(`Agent type ${type} has invalid defaults`, error); + } + }, + changed: () => this.update(), + // Its timers waited for this change, saved or failed. + settled: (key) => this.runners.wake(key), + }); void splitLegacyCheckpoints(storage); this.unfinished = Object.freeze(readUnfinished(storage)); this.unarchiving = readArchiveUnchecked(storage); @@ -1419,7 +1434,7 @@ export class AgentsService extends Service implements Agents { starting: (agent, channelId) => this.bindingOf(agent)?.session.channelWrites.starting(channelId), community: (agent) => this.bindingOf(agent), - records: () => Object.values(this.records), + records: () => this.records.all(), checkpoints: { read: () => (savedProgress = readCheckpoints(this.storage)), write: (checkpoints) => { @@ -1444,25 +1459,25 @@ export class AgentsService extends Service implements Agents { changed: () => this.update(), }, scope: (agent) => this.bindingOf(agent)?.scope, - saved: (key) => this.records[key], + saved: (key) => this.records.saved(key), record: (key) => { const member = this.liveMember(key); if (!member) throw new Error("No such agent on this device"); - return this.recordOf(member); + return this.records.recordOf(member); }, change: async (key, apply) => { - const record = await this.change(key, () => { + const record = await this.records.change(key, () => { const member = this.liveMember(key); if (!member) throw new Error("No such agent on this device"); // In turn, so this is the newest saved record. - return apply(this.recordOf(member)); + return apply(this.records.recordOf(member)); }); if (!record) throw new Error("No such agent on this device"); return record; }, - changing: (key) => this.changing.has(key), - timers: (key) => this.timersOf(key), - progress: (key, ran) => this.progress(key, ran), + changing: (key) => this.records.isChanging(key), + timers: (key) => this.records.timersOf(key), + progress: (key, ran) => this.records.progress(key, ran), start: (agent) => { const { settings, instructions } = this.configBinding?.(agent) ?? {}; const merge = this.types().find( @@ -1628,7 +1643,12 @@ export class AgentsService extends Service implements Agents { soloChat(() => this.relay(agent.key), agent.owner, channel, signal), remembered: (key, slug, body) => this.coreMemory.wrote(key, slug, body), woke: (key, object) => - this.history(key, object, undefined, Math.floor(Date.now() / 1000)), + this.records.history( + key, + object, + undefined, + Math.floor(Date.now() / 1000), + ), finished: (key) => void this.coreMemory.refresh(key), named: (agent, event, title) => void nameSoloChat( @@ -1662,10 +1682,12 @@ export class AgentsService extends Service implements Agents { }); ctx.effect(() => { const stops = [ - this.contributions.subscribe(() => { - this.update(); - this.notify(); - }), + ...[this.contributions, this.presentations].map((registry) => + registry.subscribe(() => { + this.update(); + this.notify(); + }), + ), communities.subscribe(() => { const { viewer } = communities.snapshot(); const kept = this.requests.filter( @@ -1763,7 +1785,44 @@ export class AgentsService extends Service implements Agents { ); this.contributions.register(this.ctx, type as unknown as AgentType); } - types = () => this.contributions.snapshot(); + present( + typeId: string, + present: (exposed: Exposed) => AgentPresentation, + ) { + if (typeof present !== "function") + throw new Error("present needs a function"); + this.presentations.register(this.ctx, { + id: typeId, + present: present as Presentation["present"], + }); + } + types = () => { + const types = this.contributions.snapshot(); + const shown = this.presentations.snapshot(); + if (types === this.merged.types && shown === this.merged.shown) + return this.merged.result; + const result = Object.freeze( + types.map((type) => { + const from = shown.find((entry) => entry.key === type.key); + if (!from) return type; + const known = this.presentedTypes.get(type); + if (known?.from === from) return known.type; + let merged = type; + try { + merged = Object.freeze( + presented(type, from.present(type.expose)), + ) as RegisteredAgentType; + } catch (error) { + // Its agents still run; only what the type shows is missing. + console.error(`Agent type ${type.key} could not be shown`, error); + } + this.presentedTypes.set(type, { from, type: merged }); + return merged; + }), + ); + this.merged = { types, shown, result }; + return result; + }; addTool(tool: AddedTool) { if ( !/^[a-z][a-z0-9_]{0,63}$/.test(tool.name ?? "") || @@ -1958,43 +2017,11 @@ export class AgentsService extends Service implements Agents { ? {} : { configExample: JSON.parse(JSON.stringify(configExample)) }), })); - private history( - key: AgentKey, - object: AttentionObject, - output?: string, - lastWokeAt?: number, - ) { - void this.queued(key, async () => { - const record = this.records[key]; - if (!record || record.attention[object.slug] !== object) return; - const prior = record.attentionHistory?.[object.slug]; - const entry = { - ...prior, - ...(lastWokeAt !== undefined ? { lastWokeAt } : {}), - ...(output !== undefined - ? { - output: [prior?.output, output] - .filter((part) => part !== undefined && part !== "") - .join(prior?.output?.endsWith("\n") ? "" : "\n") - .slice(-16384) - .split("\n") - .slice(-40) - .join("\n"), - } - : {}), - }; - await this.commit(key, record, { - ...record, - attentionHistory: { ...record.attentionHistory, [object.slug]: entry }, - }); - this.update(); - }).catch((error) => console.warn("Attention history was not saved", error)); - } private startAttentionSource(source: Contribution) { const sources = new AttentionSources( { completed: (key, object) => - sourceCompleted(this.records[key]?.sourceRuns, object), + sourceCompleted(this.records.saved(key)?.sourceRuns, object), finish: (key, object) => this.runners.completeSource(key, object), emit: (key, object, text, current) => { const agent = this.state.running.find((agent) => agent.key === key); @@ -2004,7 +2031,7 @@ export class AgentsService extends Service implements Agents { !object.value.enabled ) throw new Error("This source watch is no longer enabled"); - this.history(key, object, text); + this.records.history(key, object, text); this.runners.wakeSource( key, { @@ -2698,7 +2725,7 @@ export class AgentsService extends Service implements Agents { record = { ...record, config: made.config }; if (carry && kind.adopt) record = { ...record, config: kind.adopt(carry, record.config) }; - await this.change(identity.pubkey, () => record); + await this.records.edit(identity.pubkey, { kind: "create", record }); } catch (error) { // The agent exists either way; unsaved settings start from the defaults. console.warn(`Agent ${identity.name} settings were not saved`, error); @@ -2750,22 +2777,12 @@ export class AgentsService extends Service implements Agents { throw new Error(`${identity.name}'s provider is already changing`); const name = change.name?.trim(); if (name === "") throw new Error("Name the agent"); - // Its config is its home record's, in every community; the rest is this - // membership's own. - const home = key === pubkey; - if (!home && change.config !== undefined) - await this.change(pubkey, (saved) => - changed(saved ?? this.settingsOf(identity), { config: change.config }), - ); - const own = home ? change : { ...change, config: undefined }; - if ( - own.config !== undefined || - own.attention || - own.attentionEnabled !== undefined - ) - await this.change(key, (saved) => - changed(saved ?? this.recordOf(identity), own), - ); + const { name: _, picture: __, ...settings } = change; + await this.records.edit(key, { + kind: "settings", + member: identity, + change: settings, + }); const picture = change.picture === undefined ? undefined : change.picture || null; const renamed = name !== undefined && name !== identity.name; @@ -2916,10 +2933,10 @@ export class AgentsService extends Service implements Agents { ); await native.setType(pubkey, type); try { - await this.change( + await this.records.change( pubkey, (saved) => ({ - ...(saved ?? this.settingsOf(identity)), + ...(saved ?? this.records.settingsOf(identity)), config: change ? change.config : adopted, }), // Shown with its new type, only once its config is saved. @@ -2956,7 +2973,7 @@ export class AgentsService extends Service implements Agents { throw new Error("A hosted provider cannot take over a local agent key"); if (type === identity.type) return { - config: this.settingsOf(identity).config, + config: this.records.settingsOf(identity).config, moved: [], dropped: [], settings: [], @@ -3108,7 +3125,7 @@ export class AgentsService extends Service implements Agents { continue; // By its home, where it has one; else the first community it is in. if (shown.has(pubkey) && member.key !== pubkey) continue; - shown.set(pubkey, this.view(member, this.recordOf(member))); + shown.set(pubkey, this.view(member, this.records.recordOf(member))); } return [...shown.values()]; } @@ -3208,17 +3225,7 @@ export class AgentsService extends Service implements Agents { ); // Its home record keeps the agent's settings, but not what it paid // attention to there. - if (key === pubkey) - await this.change(key, (saved) => { - if (!saved) return null; - const { - skipped: _, - timers: __, - attentionEnabled: ___, - ...rest - } = saved; - return { ...rest, attention: {} }; - }); + if (key === pubkey) await this.records.edit(key, { kind: "leave" }); await this.refresh(); } @@ -3579,113 +3586,6 @@ export class AgentsService extends Service implements Agents { return agent; } - /** Changes one agent's record. This is the only way a record changes: - * owner settings, the agent's attention tools, timer progress, a provider - * change and removal all send a change here. One agent's changes run in - * the order they are sent, each after the one before it is saved or has - * failed, so each one applies to the newest saved record, never to a copy - * another change replaces. `apply` gets that record (or undefined) and - * returns the next one, or null to remove it; it may throw to change - * nothing. The owner and the agent see a change once it is saved, and the - * promise settles then. When the save fails, the promise rejects and they - * keep the saved record, so a form keeps its unsaved changes. `saved` - * runs as the change is shown, before anyone is told. */ - private change( - key: AgentKey, - apply: ( - record: AgentRecord | undefined, - ) => AgentRecord | null | Promise, - saved?: (record: AgentRecord | null) => void, - ): Promise { - this.changing.set(key, (this.changing.get(key) ?? 0) + 1); - return this.queued(key, async () => { - try { - const base = this.records[key]; - const next = await apply(base); - await this.commit(key, base, next); - saved?.(next); - this.update(); - return next; - } finally { - const left = (this.changing.get(key) ?? 1) - 1; - if (left) this.changing.set(key, left); - else { - this.changing.delete(key); - // Its timers waited for this change, saved or failed. - this.runners.wake(key); - } - } - }); - } - /** Records that timers ran or started counting, as a change to their run - * state only. It applies to whatever the record is when its turn comes, so - * it never undoes a change sent before it: when a timer was rearmed or - * restarted meanwhile, its new deadline stays and only the occurrences it - * used are added. Timers see it at once (`timersOf`), and memory keeps it - * even when its save fails, so a timer does not run again at once. */ - private progress(key: AgentKey, ran: TimerProgress) { - const pending = this.progressing.get(key) ?? []; - this.progressing.set(key, [...pending, ran]); - void this.queued(key, async () => { - try { - const base = this.records[key]; - const next = base && withProgress(base, ran); - if (!base || !next) return; - await this.commit(key, base, next).catch((error) => { - console.warn(`Agent ${key} timer state was not saved`, error); - this.records = { ...this.records, [key]: next }; - }); - this.update(); - } finally { - const left = (this.progressing.get(key) ?? []).filter( - (entry) => entry !== ran, - ); - if (left.length) this.progressing.set(key, left); - else this.progressing.delete(key); - } - }); - } - /** Runs `step` after the agent's earlier changes have settled. */ - private queued(key: AgentKey, step: () => Promise) { - const run = (this.changes.get(key) ?? Promise.resolve()).then(step); - const tail = run.then( - () => {}, - () => {}, - ); - this.changes.set(key, tail); - void tail.then(() => { - if (this.changes.get(key) === tail) this.changes.delete(key); - }); - return run; - } - /** Saves `next` over `base` in the membership's own key, then keeps it. - * Only this key: others may be another program's to change. A record - * other than the agent's home record keeps no config, which is the - * agent's, never one community's. */ - private async commit( - key: AgentKey, - base: AgentRecord | undefined, - given: AgentRecord | null, - ) { - if (given === base) return; - const next = given && key !== pubkeyOf(key) ? ownRecord(key, given) : given; - if (next) await writeRecord(this.storage, next); - else if (this.storage.get(recordKey(key)) !== null) - await removeRecord(this.storage, key); - const { [key]: _, ...others } = this.records; - this.records = next ? { ...others, [key]: next } : others; - } - /** The agent's timers' run state as its timers go by: saved, with the - * progress still being saved. */ - private timersOf(key: AgentKey) { - const record = this.records[key]; - const pending = this.progressing.get(key); - if (!record || !pending) return record?.timers ?? EMPTY; - return ( - pending.reduce((next, ran) => withProgress(next, ran) ?? next, record) - .timers ?? EMPTY - ); - } /** The agent's memberships, or every agent's, that were not deleted. */ private liveRows(pubkey?: string) { return this.identities.filter( @@ -3713,33 +3613,12 @@ export class AgentsService extends Service implements Agents { }), ); } - /** The agent's home record, which keeps its config: as saved or, without - * one, its type's defaults, kept in memory until a save stores them. - * Without its type there are no defaults yet. */ - private settingsOf({ pubkey, type }: Pick) { - const kind = this.types().find((entry) => entry.key === type); - if (!this.records[pubkey] && kind) - try { - this.records = { ...this.records, [pubkey]: defaults(pubkey, kind) }; - } catch (error) { - console.warn(`Agent type ${type} has invalid defaults`, error); - } - return this.records[pubkey] ?? blank(pubkey); - } - /** The membership's settings as its owner and agent see them: its own - * attention, starting with none, and the agent's config. */ - private recordOf(member: Member): AgentRecord { - const settings = this.settingsOf(member); - if (member.key === member.pubkey) return settings; - const own = this.records[member.key] ?? blank(member.key); - return { ...own, config: settings.config }; - } private retypePlan( identity: AgentIdentity, source: RegisteredAgentType | undefined, kind: RegisteredAgentType, ) { - const record = this.settingsOf(identity); + const record = this.records.settingsOf(identity); const carry = source?.carry?.(record.config) ?? {}; const { config } = kind.defaults(); const adopted = kind.adopt ? kind.adopt(carry, config) : config; @@ -3858,11 +3737,13 @@ export class AgentsService extends Service implements Agents { .filter((saved) => !saved.deleted) .flatMap((saved) => [saved.key, saved.pubkey]), ); - const gone = Object.keys(this.records).filter((key) => !live.has(key)); + const gone = this.records.keys().filter((key) => !live.has(key)); for (const key of gone) - void this.change(key, () => null).catch((error) => - console.warn(`Agent ${key} settings were not removed`, error), - ); + void this.records + .edit(key, { kind: "remove" }) + .catch((error) => + console.warn(`Agent ${key} settings were not removed`, error), + ); // Nor do change requests, which would hold review slots unseen. this.updates = this.updates.filter((request) => live.has(request.agent)); // A deleted agent leaves its teams; the teams stay. @@ -4140,7 +4021,7 @@ export class AgentsService extends Service implements Agents { const alive: Agent[] = []; for (const identity of identities) { if (!this.alive(identity)) continue; - const record = this.recordOf(identity); + const record = this.records.recordOf(identity); // Reuse the frozen agent while nothing visible changed, so subscribers see // a stable snapshot. const prior = diff --git a/src/plugins/api.ts b/src/plugins/api.ts index c5df871048..0f2a248a1f 100644 --- a/src/plugins/api.ts +++ b/src/plugins/api.ts @@ -50,6 +50,13 @@ export type PluginModule = { // Cordis starts apply only when these services are available. inject?: Inject; apply: (ctx: Context) => void | Promise; + /** What agents need while they run: agent types' runtimes, tools, wakes and + * Attention sources. It starts with `apply`, under the same plugin identity, + * with its own dependencies, so it can later run apart from the UI page. */ + agent?: Readonly<{ + inject?: Inject; + apply: (ctx: Context) => void | Promise; + }>; }; // Installation identity follows child Cordis scopes so services can own contributions. diff --git a/src/plugins/modules.ts b/src/plugins/modules.ts index 2f7a63eb13..b3363e8ad1 100644 --- a/src/plugins/modules.ts +++ b/src/plugins/modules.ts @@ -30,9 +30,12 @@ export function createModuleLoader( } if (!module || typeof module !== "object") throw new Error("Invalid plugin module"); - if ("apply" in module && typeof module.apply === "function") - return module as PluginModule; - throw new Error("Plugin must export apply(ctx)"); + if (!("apply" in module) || typeof module.apply !== "function") + throw new Error("Plugin must export apply(ctx)"); + const { agent } = module as { agent?: { apply?: unknown } | null }; + if (agent !== undefined && typeof agent?.apply !== "function") + throw new Error("Plugin agent must have apply(ctx)"); + return module as PluginModule; })(); modules.set(key, promise); void promise.catch(() => { diff --git a/src/plugins/runtime.test.mjs b/src/plugins/runtime.test.mjs index cea79e06e9..1a86b7e432 100644 --- a/src/plugins/runtime.test.mjs +++ b/src/plugins/runtime.test.mjs @@ -104,6 +104,59 @@ test("pages wait for dependencies and completion of async activation", async () } }); +test("an agent entry has its own dependencies and the plugin waits for both parts", async () => { + const stopped = []; + const h = harness(async () => ({ + apply(ctx) { + ctx.pages.register({ id: "main", title: "Main", component: Page }); + ctx.effect(() => () => stopped.push("ui")); + }, + agent: { + inject: ["connection"], + apply(ctx) { + assert.equal(ctx.pluginOwner.id, "example.page"); + ctx.effect(() => () => stopped.push("agent")); + }, + }, + })); + try { + h.runtime.reconcile(desired([plugin()])); + await settle(); + assert.equal(h.states["example.page"].status, "starting"); + assert.deepEqual(h.pages.snapshot(), []); + h.root.provide("connection", {}); + await h.wait((s) => s["example.page"]?.status === "active"); + assert.equal(h.pages.snapshot().length, 1); + h.runtime.reconcile([]); + await settle(); + assert.deepEqual(stopped.sort(), ["agent", "ui"]); + } finally { + await h.dispose(); + } +}); + +test("a failed agent entry fails its plugin and stops its other part", async () => { + const h = harness(async () => ({ + apply(ctx) { + ctx.pages.register({ id: "main", title: "Main", component: Page }); + }, + agent: { + apply() { + throw new Error("Agent part failed"); + }, + }, + })); + try { + h.runtime.reconcile(desired([plugin()])); + await h.wait((s) => s["example.page"]?.status === "failed"); + assert.match(h.states["example.page"].error, /Agent part failed/); + await settle(); + assert.deepEqual(h.pages.snapshot(), []); + } finally { + await h.dispose(); + } +}); + test("missing dependencies time out and cannot activate late", async () => { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); let starts = 0; diff --git a/src/plugins/runtime.ts b/src/plugins/runtime.ts index fb35dfe9ae..304bb5a301 100644 --- a/src/plugins/runtime.ts +++ b/src/plugins/runtime.ts @@ -14,7 +14,8 @@ type Entry = { plugin: PluginInfo; cancelled: boolean; failed: boolean; - fiber?: Fiber; + /** The plugin's own fiber, then its agent entry's, when it has one. */ + fibers: Fiber[]; unwatch?: () => void; stopped?: Promise; predecessor: Promise; @@ -98,6 +99,7 @@ export class PluginRuntime { plugin, cancelled: false, failed: false, + fibers: [], predecessor: previous ? this.stop(previous) : Promise.resolve(), }; this.entries.set(id, entry); @@ -150,32 +152,40 @@ export class PluginRuntime { revision: entry.plugin.revision, }), }); - const fiber = scope.plugin({ - name: entry.plugin.manifest.id, - inject: module.inject ?? [], - apply: (ctx: Context) => module.apply(ctx), - }); - entry.fiber = fiber; - this.watch(entry, fiber); + const id = entry.plugin.manifest.id; + // Both parts run in this page for now; the plugin is active once both are. + const parts = [ + { name: id, inject: module.inject, apply: module.apply }, + ...(module.agent ? [{ name: `${id}/agent`, ...module.agent }] : []), + ]; + entry.fibers = parts.map(({ name, inject, apply }) => + scope.plugin({ + name, + inject: inject ?? [], + apply: (ctx: Context) => apply(ctx), + }), + ); + this.watch(entry, entry.fibers); } catch (reason) { this.fail(entry, reason); } } - private watch(entry: Entry, fiber: Fiber) { + private watch(entry: Entry, fibers: readonly Fiber[]) { let timer: ReturnType | undefined; const refresh = () => { if (entry.cancelled || entry.failed) return; - if (fiber.state === cordisState.active) { - clearTimeout(timer); - timer = undefined; - this.update(entry, "active", null); - } else if (fiber.state === cordisState.failed) { + const failed = fibers.find((fiber) => fiber.state === cordisState.failed); + if (failed) { // await() retrieves the startup error, but does not wait for missing // dependencies. Only the ACTIVE state means contributions are ready. - void fiber.await().catch((reason) => this.fail(entry, reason)); - } else if (fiber.state === cordisState.disposed) { + void failed.await().catch((reason) => this.fail(entry, reason)); + } else if (fibers.some((fiber) => fiber.state === cordisState.disposed)) { this.fail(entry, new Error("Plugin was disposed")); + } else if (fibers.every((fiber) => fiber.state === cordisState.active)) { + clearTimeout(timer); + timer = undefined; + this.update(entry, "active", null); } else { this.update(entry, "starting", null); timer ??= setTimeout( @@ -186,7 +196,7 @@ export class PluginRuntime { }; const unsubscribe = this.root.on("internal/status", (changed) => { // ctx.plugin() returns a wrapper; events carry the underlying fiber. - if (changed.ctx === fiber.ctx) refresh(); + if (fibers.some((fiber) => changed.ctx === fiber.ctx)) refresh(); }); entry.unwatch = () => { clearTimeout(timer); @@ -202,8 +212,10 @@ export class PluginRuntime { this.update(entry, "failed", String(reason)); // Failure stays visible until an explicit retry. Retain the cleanup barrier // so replacement also waits for disposal started by a timeout or error. - if (entry.fiber) { - entry.stopped ??= entry.fiber.dispose(); + if (entry.fibers.length) { + entry.stopped ??= Promise.all( + entry.fibers.map((fiber) => fiber.dispose()), + ).then(() => {}); void entry.stopped.catch(() => {}); } } @@ -213,7 +225,7 @@ export class PluginRuntime { entry.unwatch?.(); entry.stopped ??= Promise.all([ entry.predecessor, - entry.fiber?.dispose(), + ...entry.fibers.map((fiber) => fiber.dispose()), ]).then(() => {}); return entry.stopped; }