From ff944ed8650a005c0fcfab8ddceb2fc906e40c84 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 11:53:37 -0400 Subject: [PATCH] Give plugin agent tools and attention sources a Runner owner RunnerPlugins now keeps the tools plugins add for agent turns and the attention sources that wake agents, and runs those sources against the Runner's running agents. AgentsRunner tells its subscribers when what runs or who holds it changes, so sources and agent commands follow the Runner rather than every Manager change. AgentsService forwards the plugin calls with the calling plugin's context. Behavior is unchanged. 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.ts | 26 ++- src/features/agents/runner-plugins.ts | 322 ++++++++++++++++++++++++++ src/features/agents/service.ts | 298 +----------------------- 3 files changed, 353 insertions(+), 293 deletions(-) create mode 100644 src/features/agents/runner-plugins.ts diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index d5e0a2c1e0..6596ddee4d 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -89,6 +89,7 @@ export class AgentsRunner { }> = { running: [], interrupted: [] }; /** The memberships it last decided from. */ private identities: readonly Member[] = []; + private readonly listeners = new Set<() => void>(); constructor( storage: AgentStore, @@ -130,7 +131,10 @@ export class AgentsRunner { /** The catch-up progress this program last read or saved. */ let savedProgress: Checkpoints = {}; this.runners = new AgentRunners({ - ownershipChanged: () => manager.ownershipChanged(), + ownershipChanged: () => { + this.notify(); + manager.ownershipChanged(); + }, ready: (key) => this.ready(key), agent: (key) => this.agent(key), member: (agent, event) => this.member(agent, event), @@ -262,6 +266,16 @@ export class AgentsRunner { }); } + /** Hears each change to what runs here or who holds it. */ + subscribe(listener: () => void) { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + private notify() { + for (const listener of [...this.listeners]) listener(); + } /** Connects each community where the viewer has an agent. */ watch(identities: readonly AgentIdentity[], viewer: string | undefined) { this.connections.watch(identities, viewer); @@ -317,14 +331,16 @@ export class AgentsRunner { const interrupted = running.filter( (agent) => left.has(agent.key) && this.runners.runs(agent.key), ); + const prior = this.state; this.state = { - running: same(running, this.state.running) - ? this.state.running + running: same(running, prior.running) + ? prior.running : Object.freeze(running), - interrupted: same(interrupted, this.state.interrupted) - ? this.state.interrupted + interrupted: same(interrupted, prior.interrupted) + ? prior.interrupted : Object.freeze(interrupted), }; + if (this.state.running !== prior.running) this.notify(); this.runners.resume( new Set( running diff --git a/src/features/agents/runner-plugins.ts b/src/features/agents/runner-plugins.ts new file mode 100644 index 0000000000..16f79cbf1a --- /dev/null +++ b/src/features/agents/runner-plugins.ts @@ -0,0 +1,322 @@ +// What plugins' agent entries add to the Runner: tools that agent turns can +// call, and attention sources that wake agents. Both run with the agents, so +// they are the Runner's. +import type { Context } from "@deepseek-ai/cordis"; +import { appClient } from "../../buzz-mcp/app-client"; +import { connectMcp } from "../../buzz-mcp/connection"; +import { respond } from "../../buzz-mcp/rpc"; +import { type AddedTool, BUZZ_SESSION } from "../../buzz-mcp/session-tools"; +import { + createContributions, + type Contribution, +} from "../../plugins/contributions"; +import type { + HostMcpServer, + HostProcess, + HostProcessOptions, +} from "../host/service"; +import type { AgentsRunner } from "./agents-runner"; +import { sourceCompleted } from "./attention"; +import { + AttentionSources, + type AttentionProducer, + type AttentionSource, +} from "./attention-sources"; +import { spawnAgentCommand } from "./command"; +import { pubkeyOf, type AgentKey } from "./membership"; +import { ownerReads } from "./reads"; +import { soloChat } from "./solo-chat"; + +const serializable = (value: unknown) => { + try { + JSON.stringify(value); + return true; + } catch { + return false; + } +}; + +export class RunnerPlugins { + private readonly addedTools; + /** Every added tool's name, its plugin active or still starting. */ + private readonly toolNames = new Set(); + private readonly attentionSources; + private readonly sourceTails = new Map>(); + + constructor( + ctx: Context, + private readonly runner: Pick< + AgentsRunner, + "runners" | "records" | "running" | "relay" | "subscribe" + >, + ) { + this.addedTools = createContributions(ctx); + this.attentionSources = createContributions(ctx); + ctx.effect(() => { + const running = new Map, () => void>(); + const reconcile = () => { + const active = this.attentionSources.snapshot(); + for (const [source, stop] of running) { + if (active.includes(source)) continue; + stop(); + running.delete(source); + } + for (const source of active) { + if (!running.has(source)) + running.set(source, this.startAttentionSource(source)); + } + }; + const unsubscribe = this.attentionSources.subscribe(reconcile); + reconcile(); + return () => { + unsubscribe(); + for (const stop of running.values()) stop(); + }; + }); + } + + addTool(ctx: Context, tool: AddedTool) { + if ( + !/^[a-z][a-z0-9_]{0,63}$/.test(tool.name ?? "") || + BUZZ_SESSION.tools.some((core) => core.name === tool.name) || + !tool.description || + tool.inputSchema?.type !== "object" || + !serializable(tool.inputSchema) || + typeof tool.call !== "function" + ) + throw new Error( + "Agent tools need a lowercase name no Buzz tool has, a description, a JSON object inputSchema and call", + ); + // Reserved from registration, not activation: plugins starting together + // are not active yet, and the second to take a name must fail. + if (this.toolNames.has(tool.name)) + throw new Error(`Another plugin adds agent tool ${tool.name}`); + this.addedTools.register(ctx, { ...tool, id: tool.name }); + ctx.effect(() => { + this.toolNames.add(tool.name); + return () => this.toolNames.delete(tool.name); + }); + } + addAttentionSource(ctx: Context, source: AttentionSource) { + const plugin = ctx.pluginOwner?.id; + if ( + !plugin || + !/^[a-z0-9][a-z0-9._-]*$/.test(source.id) || + typeof source.run !== "function" || + (source.description !== undefined && + (typeof source.description !== "string" || + !source.description.trim())) || + (source.configExample !== undefined && + (!source.configExample || + typeof source.configExample !== "object" || + Array.isArray(source.configExample) || + !serializable(source.configExample))) + ) + throw new Error( + "A plugin source needs an id, run and valid discovery metadata", + ); + const connect = async ( + key: AgentKey, + options: Readonly<{ + signal: AbortSignal; + channel?: string | undefined; + root?: string | undefined; + cwd?: string; + }>, + ): Promise => { + if (!ctx.pluginOwner) + throw new Error("Only an installed plugin connects as an agent"); + options.signal.throwIfAborted(); + const handle = this.runner.runners.handle(key); + if (!handle) throw new Error("This agent is not running here"); + const host = ctx.host; + if (!host?.serveMcp) + throw new Error("Agent connections require a desktop MCP host"); + const reads = () => { + const source = ownerReads(this.runner.relay(key), pubkeyOf(key)); + if (!source) throw new Error("The agent's community is not connected"); + return source; + }; + const source = reads(); + const topLevel = + options.channel !== undefined && + (await soloChat( + () => this.runner.relay(key), + handle.owner, + options.channel, + options.signal, + )); + return connectMcp( + host.serveMcp.bind(host), + (signal) => { + const client = appClient(handle, { + signal, + relayAuthor: source.relayAuthor?.(), + signingAuthority: source.workflowAuthority(), + check: () => { + handle.attention.enabled(); + if (reads().connection !== source.connection) + throw new Error("The agent's community connection changed"); + }, + read: async (path) => { + if (!host.readFile) + throw new Error("This plugin cannot read local files"); + return host.readFile( + path, + options.cwd === undefined ? undefined : { cwd: options.cwd }, + ); + }, + memories: (slug) => reads().memoryListing(signal, slug), + }); + return (message) => + respond( + client, + { + ...(options.channel === undefined + ? {} + : { channel: options.channel }), + ...(topLevel + ? { topLevel: true } + : options.root === undefined + ? {} + : { root: options.root }), + }, + message, + ); + }, + options.signal, + ); + }; + const spawn = async ( + key: AgentKey, + process: string, + options: Omit & { + signal: AbortSignal; + channel?: string | undefined; + root?: string | undefined; + }, + ): Promise => { + options.signal.throwIfAborted(); + const host = ctx.host; + if (!host?.spawn) + throw new Error("Agent commands require a desktop process host"); + const handle = this.runner.runners.handle(key); + if (!handle) throw new Error("This agent is not running here"); + const source = ownerReads(this.runner.relay(key), pubkeyOf(key)); + if (!source) throw new Error("The agent's community is not connected"); + const spawn = host.spawn.bind(host); + return spawnAgentCommand( + { + connect: (signal) => connect(key, { ...options, signal }), + spawn: (settings) => spawn(process, settings), + subscribe: (changed) => { + const agent = this.runner.subscribe(changed); + const community = this.runner.relay(key)?.subscribe(changed); + return () => { + agent(); + community?.(); + }; + }, + check: () => { + if (!handle.attention.enabled()) + throw new Error("Attention is off"); + if ( + ownerReads(this.runner.relay(key), pubkeyOf(key))?.connection !== + source.connection + ) + throw new Error("The agent's community connection changed"); + }, + }, + options, + ); + }; + this.attentionSources.register(ctx, { + ...source, + run: async (input) => { + const { + agent, + watch: { channel, root }, + } = input; + const completed = new AbortController(); + const signal = AbortSignal.any([input.signal, completed.signal]); + try { + await source.run({ + ...input, + spawn: (process, options) => + spawn(agent, process, { + ...options, + signal, + channel, + root, + }), + }); + } finally { + completed.abort(); + } + }, + ...(source.configExample === undefined + ? {} + : { configExample: JSON.parse(JSON.stringify(source.configExample)) }), + }); + } + attentionProviders = () => + this.attentionSources + .snapshot() + .map(({ key: provider, description, configExample }) => ({ + provider, + ...(description === undefined ? {} : { description }), + ...(configExample === undefined + ? {} + : { configExample: JSON.parse(JSON.stringify(configExample)) }), + })); + private startAttentionSource(source: Contribution) { + const sources = new AttentionSources( + { + completed: (key, object) => + sourceCompleted(this.runner.records.saved(key)?.sourceRuns, object), + finish: (key, object) => + this.runner.runners.completeSource(key, object), + emit: (key, object, text, current) => { + const agent = this.runner + .running() + .find((agent) => agent.key === key); + if ( + !agent?.attentionEnabled || + agent.attention[object.slug] !== object || + !object.value.enabled + ) + throw new Error("This source watch is no longer enabled"); + this.runner.records.history(key, object, text); + this.runner.runners.wakeSource( + key, + { + type: "source", + slug: object.slug, + watch: object.value, + plugin: source.pluginId, + text, + }, + object, + current, + ); + }, + }, + this.sourceTails, + ); + const update = () => + sources.reconcile( + source.key, + source, + this.runner + .running() + .filter((agent) => this.runner.runners.runs(agent.key)), + ); + const unsubscribe = this.runner.subscribe(update); + update(); + return () => { + unsubscribe(); + sources.stop(); + }; + } + tools = () => this.addedTools.snapshot(); +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index ae6c9c4b8f..5626a06f04 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1,18 +1,7 @@ -import { appClient } from "../../buzz-mcp/app-client"; -import { respond } from "../../buzz-mcp/rpc"; -import { connectMcp } from "../../buzz-mcp/connection"; -import type { - HostMcpServer, - HostProcess, - HostProcessOptions, -} from "../host/service"; -import { spawnAgentCommand } from "./command"; -import { sourceCompleted } from "./attention"; import { createLocalAgentActivity } from "./local-activity"; import type { SettingsSchema, Subject } from "./agent-config"; import type { Mode } from "./agent-start"; import { TIMEOUT_LIMIT_MS } from "./runner"; -import { ownerReads } from "./reads"; // 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 @@ -96,6 +85,7 @@ import { import { agentKey, pubkeyOf, type AgentKey } from "./membership"; import { EMPTY, type Member } from "./records"; import { AgentsRunner } from "./agents-runner"; +import { RunnerPlugins } from "./runner-plugins"; import { originOf } from "./connections"; import { AgentPresence } from "./presence"; import { @@ -111,9 +101,7 @@ import { type Team, type Unfinished, } from "./store"; -import { soloChat } from "./solo-chat"; -import { AttentionSources, type AttentionProducer } from "./attention-sources"; -import { type AddedTool, BUZZ_SESSION } from "../../buzz-mcp/session-tools"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; import { forgetDefaultAgent } from "../solo/default-agent"; /** Where an agent runs: `here`, with its key on this device; `hosted`, by its @@ -1104,14 +1092,6 @@ declare module "@deepseek-ai/cordis" { } /** Whether `value` can be sent as JSON, as every session's tool list is. */ -const serializable = (value: unknown) => { - try { - JSON.stringify(value); - return true; - } catch { - return false; - } -}; /** Bounds one cleanup attempt, whose relay steps otherwise wait on the outbox. */ const CLEANUP_TIMEOUT_MS = 60_000; const TIMER_TICK_MS = 5_000; @@ -1264,11 +1244,6 @@ export class AgentsService extends Service implements Agents { 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(); - private readonly attentionSources; - private readonly sourceTails = new Map>(); private readonly listeners = new Set<() => void>(); private requests: readonly AgentCreationRequest[] = []; private updates: readonly AgentUpdateRequest[] = []; @@ -1282,6 +1257,8 @@ export class AgentsService extends Service implements Agents { 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; /** Counts list requests, so only the latest one is applied. */ private listed = 0; /** Each running cleanup of a deleted membership, by key, how to stop it, @@ -1371,8 +1348,6 @@ 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); this.runner = new AgentsRunner(storage, native, localActivity, decisions, { ownershipChanged: () => this.notify(), changed: () => this.update(), @@ -1497,6 +1472,7 @@ export class AgentsService extends Service implements Agents { if (generation === this.keyGeneration) this.setKey(key); }, }); + this.plugins = new RunnerPlugins(ctx, this.runner); this.unfinished = Object.freeze(readUnfinished(storage)); this.unarchiving = readArchiveUnchecked(storage); this.teamList = Object.freeze(readTeams(storage)); @@ -1509,27 +1485,6 @@ export class AgentsService extends Service implements Agents { interrupted: [], teams: this.teamList, }); - ctx.effect(() => { - const running = new Map, () => void>(); - const reconcile = () => { - const active = this.attentionSources.snapshot(); - for (const [source, stop] of running) { - if (active.includes(source)) continue; - stop(); - running.delete(source); - } - for (const source of active) { - if (!running.has(source)) - running.set(source, this.startAttentionSource(source)); - } - }; - const unsubscribe = this.attentionSources.subscribe(reconcile); - reconcile(); - return () => { - unsubscribe(); - for (const stop of running.values()) stop(); - }; - }); ctx.effect(() => { const stops = [ ...[this.contributions, this.presentations].map((registry) => @@ -1674,247 +1629,13 @@ export class AgentsService extends Service implements Agents { return result; }; addTool(tool: AddedTool) { - if ( - !/^[a-z][a-z0-9_]{0,63}$/.test(tool.name ?? "") || - BUZZ_SESSION.tools.some((core) => core.name === tool.name) || - !tool.description || - tool.inputSchema?.type !== "object" || - !serializable(tool.inputSchema) || - typeof tool.call !== "function" - ) - throw new Error( - "Agent tools need a lowercase name no Buzz tool has, a description, a JSON object inputSchema and call", - ); - // Reserved from registration, not activation: plugins starting together - // are not active yet, and the second to take a name must fail. - if (this.toolNames.has(tool.name)) - throw new Error(`Another plugin adds agent tool ${tool.name}`); - this.addedTools.register(this.ctx, { ...tool, id: tool.name }); - this.ctx.effect(() => { - this.toolNames.add(tool.name); - return () => this.toolNames.delete(tool.name); - }); + this.plugins.addTool(this.ctx, tool); } addAttentionSource(source: import("./attention-sources").AttentionSource) { - const plugin = this.ctx.pluginOwner?.id; - if ( - !plugin || - !/^[a-z0-9][a-z0-9._-]*$/.test(source.id) || - typeof source.run !== "function" || - (source.description !== undefined && - (typeof source.description !== "string" || - !source.description.trim())) || - (source.configExample !== undefined && - (!source.configExample || - typeof source.configExample !== "object" || - Array.isArray(source.configExample) || - !serializable(source.configExample))) - ) - throw new Error( - "A plugin source needs an id, run and valid discovery metadata", - ); - const connect = async ( - key: AgentKey, - options: Readonly<{ - signal: AbortSignal; - channel?: string | undefined; - root?: string | undefined; - cwd?: string; - }>, - ): Promise => { - if (!this.ctx.pluginOwner) - throw new Error("Only an installed plugin connects as an agent"); - options.signal.throwIfAborted(); - const handle = this.runner.runners.handle(key); - if (!handle) throw new Error("This agent is not running here"); - const host = this.ctx.host; - if (!host?.serveMcp) - throw new Error("Agent connections require a desktop MCP host"); - const reads = () => { - const source = ownerReads(this.relay(key), pubkeyOf(key)); - if (!source) throw new Error("The agent's community is not connected"); - return source; - }; - const source = reads(); - const topLevel = - options.channel !== undefined && - (await soloChat( - () => this.relay(key), - handle.owner, - options.channel, - options.signal, - )); - return connectMcp( - host.serveMcp.bind(host), - (signal) => { - const client = appClient(handle, { - signal, - relayAuthor: source.relayAuthor?.(), - signingAuthority: source.workflowAuthority(), - check: () => { - handle.attention.enabled(); - if (reads().connection !== source.connection) - throw new Error("The agent's community connection changed"); - }, - read: async (path) => { - if (!host.readFile) - throw new Error("This plugin cannot read local files"); - return host.readFile( - path, - options.cwd === undefined ? undefined : { cwd: options.cwd }, - ); - }, - memories: (slug) => reads().memoryListing(signal, slug), - }); - return (message) => - respond( - client, - { - ...(options.channel === undefined - ? {} - : { channel: options.channel }), - ...(topLevel - ? { topLevel: true } - : options.root === undefined - ? {} - : { root: options.root }), - }, - message, - ); - }, - options.signal, - ); - }; - const spawn = async ( - key: AgentKey, - process: string, - options: Omit & { - signal: AbortSignal; - channel?: string | undefined; - root?: string | undefined; - }, - ): Promise => { - options.signal.throwIfAborted(); - const host = this.ctx.host; - if (!host?.spawn) - throw new Error("Agent commands require a desktop process host"); - const handle = this.runner.runners.handle(key); - if (!handle) throw new Error("This agent is not running here"); - const source = ownerReads(this.relay(key), pubkeyOf(key)); - if (!source) throw new Error("The agent's community is not connected"); - const spawn = host.spawn.bind(host); - return spawnAgentCommand( - { - connect: (signal) => connect(key, { ...options, signal }), - spawn: (settings) => spawn(process, settings), - subscribe: (changed) => { - const agent = this.subscribe(changed); - const community = this.relay(key)?.subscribe(changed); - return () => { - agent(); - community?.(); - }; - }, - check: () => { - if (!handle.attention.enabled()) - throw new Error("Attention is off"); - if ( - ownerReads(this.relay(key), pubkeyOf(key))?.connection !== - source.connection - ) - throw new Error("The agent's community connection changed"); - }, - }, - options, - ); - }; - this.attentionSources.register(this.ctx, { - ...source, - run: async (input) => { - const { - agent, - watch: { channel, root }, - } = input; - const completed = new AbortController(); - const signal = AbortSignal.any([input.signal, completed.signal]); - try { - await source.run({ - ...input, - spawn: (process, options) => - spawn(agent, process, { - ...options, - signal, - channel, - root, - }), - }); - } finally { - completed.abort(); - } - }, - ...(source.configExample === undefined - ? {} - : { configExample: JSON.parse(JSON.stringify(source.configExample)) }), - }); + this.plugins.addAttentionSource(this.ctx, source); } - attentionProviders = () => - this.attentionSources - .snapshot() - .map(({ key: provider, description, configExample }) => ({ - provider, - ...(description === undefined ? {} : { description }), - ...(configExample === undefined - ? {} - : { configExample: JSON.parse(JSON.stringify(configExample)) }), - })); - private startAttentionSource(source: Contribution) { - const sources = new AttentionSources( - { - completed: (key, object) => - sourceCompleted(this.runner.records.saved(key)?.sourceRuns, object), - finish: (key, object) => - this.runner.runners.completeSource(key, object), - emit: (key, object, text, current) => { - const agent = this.state.running.find((agent) => agent.key === key); - if ( - !agent?.attentionEnabled || - agent.attention[object.slug] !== object || - !object.value.enabled - ) - throw new Error("This source watch is no longer enabled"); - this.runner.records.history(key, object, text); - this.runner.runners.wakeSource( - key, - { - type: "source", - slug: object.slug, - watch: object.value, - plugin: source.pluginId, - text, - }, - object, - current, - ); - }, - }, - this.sourceTails, - ); - const update = () => - sources.reconcile( - source.key, - source, - this.state.running.filter((agent) => - this.runner.runners.runs(agent.key), - ), - ); - const unsubscribe = this.subscribe(update); - update(); - return () => { - unsubscribe(); - sources.stop(); - }; - } - tools = () => this.addedTools.snapshot(); + attentionProviders = () => this.plugins.attentionProviders(); + tools = () => this.plugins.tools(); wake(key: AgentKey, wake: AgentWake) { const plugin = this.ctx.pluginOwner?.id; if (!plugin) throw new Error("Only a plugin wakes an agent"); @@ -1951,6 +1672,7 @@ export class AgentsService extends Service implements Agents { relay = (key: AgentKey) => this.runner.relay(key); core = (key: AgentKey, signal?: AbortSignal) => this.runner.coreMemory.core(key, signal); + private localCache: | { identities: readonly AgentIdentity[];