From 3a48e7d869082c4c6957dcba2c35c755f068954b Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 19:06:49 -0400 Subject: [PATCH] Build the Runner as its own service, apart from the Manager The Runner now has its own class: the agent types that run here, what plugins' agent entries add, its timers and its side of the link. The Manager builds it in this page for now and keeps hosted types, presentations, presence and reconnects. The Runner no longer calls the Manager directly when something changes; the link already carries it. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.test.ts | 1 - src/features/agents/agents-runner.ts | 13 +- src/features/agents/runner-service.test.ts | 60 +++++++ src/features/agents/runner-service.ts | 184 +++++++++++++++++++++ src/features/agents/service.ts | 146 ++++------------ 5 files changed, 278 insertions(+), 126 deletions(-) create mode 100644 src/features/agents/runner-service.test.ts create mode 100644 src/features/agents/runner-service.ts diff --git a/src/features/agents/agents-runner.test.ts b/src/features/agents/agents-runner.test.ts index c9904cf816..a56e1f2906 100644 --- a/src/features/agents/agents-runner.test.ts +++ b/src/features/agents/agents-runner.test.ts @@ -29,7 +29,6 @@ function saved() { function runner(storage: ReturnType) { const manager = { ownershipChanged: () => {}, - changed: () => {}, open: () => undefined, types: () => [], defaults: () => { diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index 4d3db5bafa..940747bee9 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -63,8 +63,6 @@ export type RunnerManager = Pick< | "setClassifierKey" > & Readonly<{ - /** A record or turn in progress changed what the screens show. */ - changed(): void; /** The community's relay, while the viewer is in it. */ open(origin: string): RelayData | undefined; types(): readonly RegisteredAgentType[]; @@ -149,7 +147,7 @@ export class AgentsRunner { }, changed: () => { for (const listener of [...this.recordListeners]) listener(); - this.changed(); + this.update(); }, // Its timers waited for this change, saved or failed. settled: (key) => this.runners.wake(key), @@ -163,7 +161,6 @@ export class AgentsRunner { changed: (origin, renewed) => { this.update(); if (renewed) this.refreshCore(origin); - this.manager.changed(); }, }); this.coreMemory = new CoreMemoryCache((key) => @@ -204,7 +201,7 @@ export class AgentsRunner { console.warn("Agent turns in progress were not saved", error), ); }, - changed: () => this.changed(), + changed: () => this.update(), }, scope: (agent) => this.connections.binding(agent)?.scope, saved: (key) => this.records.saved(key), @@ -376,12 +373,6 @@ export class AgentsRunner { console.warn(`Agent ${key} settings were not removed`, error), ); } - /** Decides again what runs once a record, a turn in progress or one of its - * sessions changed, and lets the Manager show it. */ - private changed() { - this.update(); - this.manager.changed(); - } /** Agent types were loaded or unloaded. */ typesChanged() { this.update(); diff --git a/src/features/agents/runner-service.test.ts b/src/features/agents/runner-service.test.ts new file mode 100644 index 0000000000..8b6b681071 --- /dev/null +++ b/src/features/agents/runner-service.test.ts @@ -0,0 +1,60 @@ +import { Context } from "@deepseek-ai/cordis"; +import { expect, it, vi } from "vitest"; +import { link } from "./link"; +import { createLocalAgentActivity } from "./local-activity"; +import { AgentsRunnerService } from "./runner-service"; +import type { AgentCommunities } from "./service"; +import { memoryStorage } from "./test-fakes"; + +it("runs without the Manager, which hears its agent types over the link", async () => { + const ctx = new Context(); + ctx.provide("pluginStatus", { + isActive: () => true, + subscribe: () => () => {}, + }); + const plugin = ctx.extend({ + pluginOwner: Object.freeze({ id: "example", revision: "one" }), + }); + const channel = new MessageChannel(); + const communities = { + snapshot: () => ({ viewer: undefined }), + subscribe: () => () => {}, + open: () => undefined, + } as unknown as AgentCommunities; + const runner = new AgentsRunnerService(ctx, channel.port1, { + communities, + native: undefined, + storage: memoryStorage(), + localActivity: createLocalAgentActivity(), + decisions: undefined, + }); + const manager = link(channel.port2); + manager.watch("runner.types", null, () => {}); + runner.register(plugin, { + id: "echo", + title: "Echo", + defaults: () => ({ config: {} }), + run: () => {}, + }); + await vi.waitFor(() => + expect(manager.read("runner.types")).toEqual([ + expect.objectContaining({ key: "example/echo", title: "Echo" }), + ]), + ); + // A hosted type is the Manager's. + expect(() => + runner.register(plugin, { + id: "hosted", + title: "Hosted", + defaults: () => ({ config: {} }), + remote: { + provision: async () => ({}), + activate: async () => {}, + profile: async () => {}, + remove: async () => {}, + }, + } as never), + ).toThrow("A hosted agent type runs elsewhere"); + manager.close(); + await ctx.fiber.dispose(); +}); diff --git a/src/features/agents/runner-service.ts b/src/features/agents/runner-service.ts new file mode 100644 index 0000000000..0f5ec8dde2 --- /dev/null +++ b/src/features/agents/runner-service.ts @@ -0,0 +1,184 @@ +// The Runner: it runs agents, owns the agent storage they keep, and has what +// plugins' agent entries add for them. It reaches the Manager only over its +// link, so a page can build it without the screens. +import type { Context } from "@deepseek-ai/cordis"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import { createContributions } from "../../plugins/contributions"; +import type { Decisions } from "../decisions/service"; +import { AgentsRunner } from "./agents-runner"; +import type { AttentionSource } from "./attention-sources"; +import type { RuntimeLink } from "./exposed"; +import type { LinkPort } from "./link"; +import type { LocalAgentActivity } from "./local-activity"; +import type { AgentKey } from "./membership"; +import type { AgentsNative } from "./native"; +import { TIMEOUT_LIMIT_MS } from "./runner"; +import { routed, runnerSide } from "./runner-link"; +import { RunnerPlugins } from "./runner-plugins"; +import type { + Agent, + AgentCommunities, + AgentType, + AgentWake, + ConfigBinding, + RegisteredAgentType, +} from "./service"; +import { PRESENTED } from "./shown-types"; +import type { AgentStore } from "./state"; +import { setAttention, type AgentRecord } from "./store"; + +/** How often each runner is woken, to start any timer that is due. */ +const TIMER_TICK_MS = 5_000; + +/** A new agent's settings: its type's defaults. */ +export function defaults(pubkey: string, type: RegisteredAgentType) { + const { config, attention = {} } = type.defaults(); + let record: AgentRecord = { pubkey, attention: {}, config }; + for (const [slug, value] of Object.entries(attention)) + record = setAttention(record, slug, value); + return record; +} + +/** Throws unless `type` is a whole agent type: one that runs here, or a + * hosted one. */ +export function checkType(type: AgentType) { + if ( + !/^[a-z0-9][a-z0-9._-]*$/.test(type.id ?? "") || + !type.title || + typeof type.defaults !== "function" + ) + throw new Error("Agent types need an id, a title and defaults"); + if ( + type.timeoutMs !== undefined && + !( + Number.isInteger(type.timeoutMs) && + type.timeoutMs >= 1 && + type.timeoutMs <= TIMEOUT_LIMIT_MS + ) + ) + throw new Error( + `Agent type timeoutMs must be a whole number from 1 to ${TIMEOUT_LIMIT_MS}`, + ); + const { remote } = type; + if ( + remote && + (type.run || + type.models || + type.timeoutMs !== undefined || + (["provision", "activate", "profile", "remove"] as const).some( + (step) => typeof remote[step] !== "function", + )) + ) + throw new Error( + "A hosted agent type has provision, activate, profile and remove, and no run, models or timeoutMs", + ); + // A type that runs here may run in another page than its UI. + const shown = remote + ? undefined + : (PRESENTED.find((field) => type[field] !== undefined) ?? + (type.sessionControls?.scope && "sessionControls.scope") ?? + (type.provider?.DialogBody && "provider.DialogBody") ?? + (type.provider?.DefaultControl && "provider.DefaultControl")); + if (shown) + throw new Error( + `An agent type that runs here gives ${shown} from its plugin's present`, + ); +} + +export type RunnerOptions = Readonly<{ + communities: AgentCommunities; + native: AgentsNative | undefined; + storage: AgentStore; + localActivity: LocalAgentActivity; + decisions: Decisions | undefined; + /** Which artifact types an agent has synced. */ + configBinding?: ((agent: Agent) => ConfigBinding) | undefined; +}>; + +/** The Runner, with its side of the link on `port`. What a plugin calls + * takes that plugin's `ctx`. */ +export class AgentsRunnerService { + /** The agent types that run here. */ + private readonly types; + private readonly runner: AgentsRunner; + private readonly plugins: RunnerPlugins; + private readonly link: ReturnType; + + constructor(ctx: Context, port: LinkPort, options: RunnerOptions) { + const { communities, storage, native, localActivity, decisions } = options; + this.types = createContributions(ctx); + this.runner = new AgentsRunner(storage, native, localActivity, decisions, { + open: (origin) => communities.open(origin), + types: () => this.types.snapshot(), + defaults, + configBinding: options.configBinding, + // What the Runner asks of the Manager crosses the link. + createAgent: (agent, input) => this.link.createAgent(agent, input), + updateAgent: (agent, input) => this.link.updateAgent(agent, input), + templates: (agent) => this.link.templates(agent), + keyStatus: () => this.link.keyStatus(), + classifierKey: () => this.link.classifierKey(), + classifierKeyGeneration: () => this.link.classifierKeyGeneration(), + setClassifierKey: (key, generation) => + this.link.setClassifierKey(key, generation), + }); + this.plugins = new RunnerPlugins(ctx, this.runner); + this.link = runnerSide(port, this.runner); + ctx.effect(() => { + const stops = [ + this.types.subscribe(() => this.runner.typesChanged()), + communities.subscribe(() => + this.runner.communitiesChanged(communities.snapshot().viewer), + ), + ]; + const tick = setInterval(() => { + this.runner.runners.tick(); + this.runner.runners.catchUp(true); + }, TIMER_TICK_MS); + return () => { + clearInterval(tick); + for (const stop of stops) stop(); + // Nothing the link sends now may start runners again. + this.link.close(); + this.runner.clear(); + }; + }); + } + + register(ctx: Context, type: AgentType) { + checkType(type); + if (type.remote) throw new Error("A hosted agent type runs elsewhere"); + this.types.register(ctx, type as AgentType); + } + /** The link to what `ctx`'s plugin's UI shares. */ + shared(ctx: Context) { + const plugin = ctx.pluginOwner?.id; + if (!plugin) throw new Error("Only a plugin reaches what its UI shares"); + const reached = routed(this.link, "plugin", { plugin }); + ctx.effect(() => () => reached.close()); + return reached as RuntimeLink; + } + addTool(ctx: Context, tool: AddedTool) { + this.plugins.addTool(ctx, tool); + } + addAttentionSource(ctx: Context, source: AttentionSource) { + this.plugins.addAttentionSource(ctx, source); + } + attentionProviders = () => this.plugins.attentionProviders(); + tools = () => this.plugins.tools(); + wake(ctx: Context, key: AgentKey, wake: AgentWake) { + const plugin = ctx.pluginOwner?.id; + if (!plugin) throw new Error("Only a plugin wakes an agent"); + this.runner.wake(key, wake, { plugin }); + } + handle(ctx: Context, key: AgentKey) { + const owner = ctx.pluginOwner; + return owner && this.runner.runners.handle(key, owner.id); + } + note(key: AgentKey, channelId: string, text: string, root?: string) { + return this.runner.runners.note(key, channelId, text, root); + } + relay = (key: AgentKey) => this.runner.relay(key); + core = (key: AgentKey, signal?: AbortSignal) => + this.runner.coreMemory.core(key, signal); +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 953149c536..fc5a1b0819 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1,7 +1,6 @@ import { createLocalAgentActivity } from "./local-activity"; import type { SettingsSchema, Subject } from "./agent-config"; import type { Mode } from "./agent-start"; -import { TIMEOUT_LIMIT_MS } from "./runner"; // Agents: agents that are plugins from the start. A plugin registers an agent // type (its summary line, peek view, settings tabs and `run`); each agent made // from it has its own native-held key, an `agent-attention/v1` configuration the @@ -83,17 +82,13 @@ import { } from "./native"; import { agentKey, pubkeyOf, type AgentKey } from "./membership"; import { EMPTY, type Member } from "./records"; -import { - AgentsRunner, - classifierAvailability, - type RunnerStatus, -} from "./agents-runner"; -import { RunnerPlugins } from "./runner-plugins"; +import { classifierAvailability, type RunnerStatus } from "./agents-runner"; +import { AgentsRunnerService, checkType, defaults } from "./runner-service"; export { teamPrompt } from "./team-prompt"; -import { managerSide, routed, runnerSide, type TypeInfo } from "./runner-link"; +import { managerSide, type TypeInfo } from "./runner-link"; import type { RuntimeLink } from "./exposed"; import type { Served } from "./link"; -import { PRESENTED, shownType } from "./shown-types"; +import { shownType } from "./shown-types"; type ManagerSide = ReturnType; import { originOf } from "./connections"; @@ -102,7 +97,6 @@ import { readArchiveUnchecked, readTeams, readUnfinished, - setAttention, writeArchiveUnchecked, writeTeams, writeUnfinished, @@ -1109,14 +1103,6 @@ const CLEANUP_TIMEOUT_MS = 60_000; const TIMER_TICK_MS = 5_000; /** How often a failed connection to an unselected community is tried again. */ const RETRY_MS = 30_000; -/** A new agent's settings: its type's defaults. */ -function defaults(pubkey: string, type: RegisteredAgentType) { - const { config, attention = {} } = type.defaults(); - let record: AgentRecord = { pubkey, attention: {}, config }; - for (const [slug, value] of Object.entries(attention)) - record = setAttention(record, slug, value); - return record; -} const message = (error: unknown) => String(error instanceof Error ? error.message : error); export { fromOwner } from "./owner-attestation"; @@ -1207,6 +1193,7 @@ export class AgentsService extends Service implements Agents { private keyGeneration = 0; /** Whether the saved key's first read ended, or there is none to read. */ private keyChecked = false; + /** The hosted agent types; the Runner keeps those that run here. */ private readonly contributions; private readonly presentations; private readonly shares; @@ -1237,12 +1224,9 @@ export class AgentsService extends Service implements Agents { private state: AgentsSnapshot; /** By origin, so each agent listens in its own community, selected or not. */ private bound = new Map(); - /** Runs the agents, and owns the agent storage they keep. */ - private readonly runner: AgentsRunner; - /** The tools and attention sources plugins add for agents: the Runner's. */ - private readonly plugins: RunnerPlugins; - /** The Runner's side of the link between them. */ - private readonly runnerLink: ReturnType; + /** Runs the agents, and owns the agent storage they keep, with what + * plugins' agent entries add for them. */ + private readonly runner: AgentsRunnerService; /** The Manager's side: what it sends the Runner, and what the Runner * decided, as the link last sent it. */ private readonly decided: ManagerSide; @@ -1332,34 +1316,24 @@ export class AgentsService extends Service implements Agents { private readonly decisions: Decisions | undefined = undefined, /** Which artifact types an agent has synced, from its config channel * binding. Unset, every agent starts `local`. */ - private readonly configBinding?: (agent: Agent) => ConfigBinding, + configBinding?: (agent: Agent) => ConfigBinding, ) { super(ctx, "agents"); this.contributions = createContributions(ctx); this.presentations = createContributions(ctx); this.shares = createContributions(ctx); - this.runner = new AgentsRunner(storage, native, localActivity, decisions, { - changed: () => this.update(), - open: (origin) => this.communities.open(origin), - types: () => this.contributions.snapshot(), - defaults, - configBinding: this.configBinding, - // What the Runner asks of the Manager crosses the link. - createAgent: (agent, input) => this.runnerLink.createAgent(agent, input), - updateAgent: (agent, input) => this.runnerLink.updateAgent(agent, input), - templates: (agent) => this.runnerLink.templates(agent), - keyStatus: () => this.runnerLink.keyStatus(), - classifierKey: () => this.runnerLink.classifierKey(), - classifierKeyGeneration: () => this.runnerLink.classifierKeyGeneration(), - setClassifierKey: (key, generation) => - this.runnerLink.setClassifierKey(key, generation), - }); - this.plugins = new RunnerPlugins(ctx, this.runner); // The Runner decides from the memberships the Manager sends over the // link, and the screens show what it sends back. Only JSON crosses, as it // will between pages. const channel = new MessageChannel(); - this.runnerLink = runnerSide(channel.port1, this.runner); + this.runner = new AgentsRunnerService(ctx, channel.port1, { + communities, + native, + storage, + localActivity, + decisions, + configBinding, + }); this.decided = managerSide(channel.port2, { members: () => ({ loaded: this.loaded, @@ -1509,7 +1483,6 @@ export class AgentsService extends Service implements Agents { const stops = [ ...[this.contributions, this.presentations].map((registry) => registry.subscribe(() => { - this.runner.typesChanged(); this.update(); this.notify(); }), @@ -1530,7 +1503,6 @@ export class AgentsService extends Service implements Agents { this.updates = updates; this.notify(); } - this.runner.communitiesChanged(viewer); this.watch(); this.update(); }), @@ -1558,12 +1530,9 @@ export class AgentsService extends Service implements Agents { found("unreadable"); }) .finally(checked); - // Wakes each runner, which starts any timer that is due. const tick = setInterval(() => { - this.runner.runners.tick(); this.reconnect(); this.presence.announce(); - this.runner.runners.catchUp(true); }, TIMER_TICK_MS); return () => { clearInterval(tick); @@ -1571,10 +1540,7 @@ export class AgentsService extends Service implements Agents { this.listed++; for (const stop of stops) stop(); for (const origin of [...this.bound.keys()]) this.unbind(origin); - // Nothing the link sends now may start runners again. - this.runnerLink.close(); this.decided.close(); - this.runner.clear(); this.presence.clear(); // Clear what is shown without reconciling, which would make new runners. this.state = Object.freeze({ @@ -1589,48 +1555,11 @@ export class AgentsService extends Service implements Agents { } register(type: AgentType) { - if ( - !/^[a-z0-9][a-z0-9._-]*$/.test(type.id ?? "") || - !type.title || - typeof type.defaults !== "function" - ) - throw new Error("Agent types need an id, a title and defaults"); - if ( - type.timeoutMs !== undefined && - !( - Number.isInteger(type.timeoutMs) && - type.timeoutMs >= 1 && - type.timeoutMs <= TIMEOUT_LIMIT_MS - ) - ) - throw new Error( - `Agent type timeoutMs must be a whole number from 1 to ${TIMEOUT_LIMIT_MS}`, - ); - const { remote } = type; - if ( - remote && - (type.run || - type.models || - type.timeoutMs !== undefined || - (["provision", "activate", "profile", "remove"] as const).some( - (step) => typeof remote[step] !== "function", - )) - ) - throw new Error( - "A hosted agent type has provision, activate, profile and remove, and no run, models or timeoutMs", - ); - // A type that runs here may run in another page than its UI. - const shown = remote - ? undefined - : (PRESENTED.find((field) => type[field] !== undefined) ?? - (type.sessionControls?.scope && "sessionControls.scope") ?? - (type.provider?.DialogBody && "provider.DialogBody") ?? - (type.provider?.DefaultControl && "provider.DefaultControl")); - if (shown) - throw new Error( - `An agent type that runs here gives ${shown} from its plugin's present`, - ); - this.contributions.register(this.ctx, type as unknown as AgentType); + const registered = type as unknown as AgentType; + checkType(registered); + // A type that runs here is the Runner's; the Manager keeps hosted ones. + if (registered.remote) this.contributions.register(this.ctx, registered); + else this.runner.register(this.ctx, registered); } present( typeId: string, @@ -1649,11 +1578,7 @@ export class AgentsService extends Service implements Agents { this.shares.register(this.ctx, { id: "shared", served }); } shared() { - const plugin = this.ctx.pluginOwner?.id; - if (!plugin) throw new Error("Only a plugin reaches what its UI shares"); - const reached = routed(this.runnerLink, "plugin", { plugin }); - this.ctx.effect(() => () => reached.close()); - return reached as RuntimeLink; + return this.runner.shared(this.ctx); } /** The agent types as the screens and management see them: each that runs * here as the Runner says, with what its plugin presents, then each hosted @@ -1694,25 +1619,20 @@ export class AgentsService extends Service implements Agents { known.runtime?.close(); this.presentedTypes.delete(info); } - const result = Object.freeze([ - ...local, - ...registered.filter((type) => type.remote), - ]); + const result = Object.freeze([...local, ...registered]); this.merged = { infos, registered, shown, result }; return result; }; addTool(tool: AddedTool) { - this.plugins.addTool(this.ctx, tool); + this.runner.addTool(this.ctx, tool); } addAttentionSource(source: import("./attention-sources").AttentionSource) { - this.plugins.addAttentionSource(this.ctx, source); + this.runner.addAttentionSource(this.ctx, source); } - attentionProviders = () => this.plugins.attentionProviders(); - tools = () => this.plugins.tools(); + attentionProviders = () => this.runner.attentionProviders(); + tools = () => this.runner.tools(); wake(key: AgentKey, wake: AgentWake) { - const plugin = this.ctx.pluginOwner?.id; - if (!plugin) throw new Error("Only a plugin wakes an agent"); - this.runner.wake(key, wake, { plugin }); + this.runner.wake(this.ctx, key, wake); } snapshot = () => this.state; subscribe = (listener: () => void) => { @@ -1738,15 +1658,13 @@ export class AgentsService extends Service implements Agents { find = (pubkey: string) => this.state.agents.find((agent) => agent.pubkey === pubkey); handle(key: AgentKey) { - const owner = this.ctx.pluginOwner; - return owner && this.runner.runners.handle(key, owner.id); + return this.runner.handle(this.ctx, key); } note(key: AgentKey, channelId: string, text: string, root?: string) { - return this.runner.runners.note(key, channelId, text, root); + return this.runner.note(key, channelId, text, root); } relay = (key: AgentKey) => this.runner.relay(key); - core = (key: AgentKey, signal?: AbortSignal) => - this.runner.coreMemory.core(key, signal); + core = (key: AgentKey, signal?: AbortSignal) => this.runner.core(key, signal); private localCache: | {