From 193154ad2766b26146099291ab8d2df6088b95b7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 10:23:51 -0400 Subject: [PATCH 1/7] refactor(plugins): give plugins an agent entry A plugin can now export `agent = { inject, apply }` for what agents need while they run (agent type runtimes, tools, wakes, Attention sources), separate from its UI `apply`. Both parts start in the same page, under the same plugin identity, each with its own dependencies; the plugin is active once both are and fails if either does. GitHub's watch_pr moves into its agent entry. No behavior changes. This is the first step toward running agents in their own webview, so a hidden or suspended UI page cannot stall them. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- docs/plugin-architecture.md | 7 +++++ src/bundled/github/index.tsx | 7 +++-- src/plugins/api.ts | 7 +++++ src/plugins/modules.ts | 9 ++++-- src/plugins/runtime.test.mjs | 53 ++++++++++++++++++++++++++++++++++++ src/plugins/runtime.ts | 52 +++++++++++++++++++++-------------- 6 files changed, 110 insertions(+), 25 deletions(-) diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index 2b34165515..82dc5c04e5 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -238,6 +238,13 @@ 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 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/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/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; } From 2838db5e05a3792e7a8d36da10274ea5ff0a38b6 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 10:33:59 -0400 Subject: [PATCH 2/7] refactor(agents): split the Claude harness into runtime and UI An agent type can now come in two halves. The plugin's agent entry registers the runtime half with `agents.register`, and may pass `expose`: the handles its own UI reaches of the runtime (stores and async calls). The UI entry supplies icon, Create, tabs and provider dialogs with `agents.present(typeId, (exposed) => ...)`. The service merges both by type key, so readers of `types()` see one type, as before. Provider getters stay live, and a throwing `present` leaves the type's agents running. Claude moves its setup, runtime, process sync and update timer into its agent entry. Its tabs take narrow handles in place of the runtime classes. No behavior changes. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- docs/plugin-architecture.md | 8 ++ src/bundled/agents/AgentsPage.test.tsx | 1 + src/bundled/agents/ProviderDialog.tsx | 2 +- src/bundled/claude-code/index.test.ts | 16 ++-- src/bundled/claude-code/index.ts | 92 ++++++++++++-------- src/bundled/claude-code/tabs.test.tsx | 17 ++-- src/bundled/claude-code/tabs.tsx | 28 ++++-- src/features/agents/plugin-testing.ts | 24 ++++++ src/features/agents/service.test.ts | 63 ++++++++++++++ src/features/agents/service.ts | 115 +++++++++++++++++++++++-- 10 files changed, 296 insertions(+), 70 deletions(-) create mode 100644 src/features/agents/plugin-testing.ts diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index 82dc5c04e5..c7d2b32e89 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -245,6 +245,14 @@ 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/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/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/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..f0afa801eb 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -601,8 +601,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 +709,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 +807,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 +945,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 @@ -1287,6 +1337,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(); @@ -1384,6 +1445,7 @@ 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. @@ -1662,10 +1724,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 +1827,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 ?? "") || From 8f2f30c88ff7b500e6fc430ad3aa7e607f70e150 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 10:39:10 -0400 Subject: [PATCH 3/7] refactor(codex): split the Codex harness into runtime and UI Codex's setup, provider, runtime, process sync and update timer move into its agent entry, which registers the type and exposes handles for its UI. The UI entry presents the icon, Create, tabs and provider dialogs. The Activity tab no longer starts Codex itself to read a transcript: the runtime exposes a transcript read, so the UI holds no process capability. No behavior changes. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/bundled/codex/index.test.ts | 15 ++--- src/bundled/codex/index.ts | 99 ++++++++++++++++++++----------- src/bundled/codex/runtime.test.ts | 6 +- src/bundled/codex/tabs.test.tsx | 11 +++- src/bundled/codex/tabs.tsx | 52 ++++++++++++---- 5 files changed, 123 insertions(+), 60 deletions(-) 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 => From d04d1c2382606b7ecc018be93468cacdcf88340d Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 10:43:07 -0400 Subject: [PATCH 4/7] refactor(goose): split the goose harness into runtime and UI The goose agent entry registers the runtime half and exposes setup, session and history handles. The UI entry presents the icon, create form, provider dialog and tabs from those handles only. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/bundled/goose/index.test.ts | 16 +++---- src/bundled/goose/index.ts | 80 +++++++++++++++++++-------------- src/bundled/goose/tabs.tsx | 22 +++++++-- 3 files changed, 70 insertions(+), 48 deletions(-) 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); From d8171a399c4587a9e1e618a5b7552daf9f9e9018 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 10:44:31 -0400 Subject: [PATCH 5/7] refactor(pi): split the Pi harness into runtime and UI The Pi agent entry registers the runtime half and exposes the setup and runtime handles. The UI entry presents the icon, create form and tabs from those handles only. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/bundled/pi/index.test.ts | 23 +++++++++++++ src/bundled/pi/index.ts | 63 ++++++++++++++++++++++-------------- src/bundled/pi/tabs.tsx | 2 +- 3 files changed, 63 insertions(+), 25 deletions(-) create mode 100644 src/bundled/pi/index.test.ts 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 From 49c9b511f1490c5a18f69aee3fc55e10b7ffdbd3 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 15:05:30 -0400 Subject: [PATCH 6/7] Send local agent work to the screens over a JSON link Add the agents link: each side serves named topics, which the other watches, and named commands, which answer one call. Only JSON crosses, and either side may reload. The screens now read the local work that running agents report through this link, and Stop and hiding an error go back through it. Both sides still share one page. This is the first part of the Runner, which runs agents, moving to its own page. Behavior is the same, one task later. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/app/services.ts | 18 +- src/features/agents/link.test.ts | 191 ++++++++++++++++++ src/features/agents/link.ts | 254 ++++++++++++++++++++++++ src/features/agents/runner-link.test.ts | 78 ++++++++ src/features/agents/runner-link.ts | 65 ++++++ 5 files changed, 605 insertions(+), 1 deletion(-) create mode 100644 src/features/agents/link.test.ts create mode 100644 src/features/agents/link.ts create mode 100644 src/features/agents/runner-link.test.ts create mode 100644 src/features/agents/runner-link.ts 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/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/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 }; +} From 5add18de298dfc31b15e49d3cea3fdc54d5a75c1 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 15:12:10 -0400 Subject: [PATCH 7/7] Give agent records one owner that the screens edit with data Move the agents' saved records and each record's change queue out of AgentsService into AgentRecords. The Runner's own paths still change a record with a function. The screens' paths (save, add, leave, and the cleanup of records whose agent is gone) now send a RecordEdit, which is data only, so it can later cross to the Runner's page over the agents link. No behavior changes. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/records.test.ts | 97 ++++++++ src/features/agents/records.ts | 335 +++++++++++++++++++++++++++ src/features/agents/service.ts | 338 +++++----------------------- 3 files changed, 491 insertions(+), 279 deletions(-) create mode 100644 src/features/agents/records.test.ts create mode 100644 src/features/agents/records.ts 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/service.ts b/src/features/agents/service.ts index f0afa801eb..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, @@ -1144,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(); @@ -1265,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 { @@ -1362,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. */ @@ -1451,8 +1392,20 @@ export class AgentsService extends Service implements Agents { // 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); @@ -1481,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) => { @@ -1506,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( @@ -1690,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( @@ -2059,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); @@ -2105,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, { @@ -2799,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); @@ -2851,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; @@ -3017,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. @@ -3057,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: [], @@ -3209,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()]; } @@ -3309,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(); } @@ -3680,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( @@ -3814,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; @@ -3959,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. @@ -4241,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 =