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: | {