From 32d6448f85f59ffa422e768362ddd3bdde60ef52 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Thu, 29 Oct 2026 08:41:11 -0400 Subject: [PATCH] Give the Runner its own owner of agent state, in the same page The Runner, which runs agents, now owns what it decides from: agent records, presence, runner storage, its own community connections, which agents run, and plugin agent tools and Attention sources. Local agent work reaches the screens over a JSON link. Inert: the Runner still runs in the screens' page. Squashes #1748, #1749, #1750, #1752, #1755, #1824 and #1826. Signed-off-by: Logan Johnson Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 --- src/app/services.ts | 18 +- src/features/agents/agents-runner.ts | 571 +++++++++ src/features/agents/attention-command.ts | 12 + src/features/agents/connections.test.ts | 71 ++ src/features/agents/connections.ts | 146 +++ src/features/agents/link.test.ts | 191 +++ src/features/agents/link.ts | 254 ++++ src/features/agents/presence.ts | 114 ++ src/features/agents/records.test.ts | 97 ++ src/features/agents/records.ts | 365 ++++++ src/features/agents/runner-link.test.ts | 78 ++ src/features/agents/runner-link.ts | 65 + src/features/agents/runner-plugins.ts | 348 ++++++ src/features/agents/service.test.ts | 23 +- src/features/agents/service.ts | 1423 ++++------------------ 15 files changed, 2588 insertions(+), 1188 deletions(-) create mode 100644 src/features/agents/agents-runner.ts create mode 100644 src/features/agents/connections.test.ts create mode 100644 src/features/agents/connections.ts create mode 100644 src/features/agents/link.test.ts create mode 100644 src/features/agents/link.ts create mode 100644 src/features/agents/presence.ts create mode 100644 src/features/agents/records.test.ts create mode 100644 src/features/agents/records.ts create mode 100644 src/features/agents/runner-link.test.ts create mode 100644 src/features/agents/runner-link.ts create mode 100644 src/features/agents/runner-plugins.ts diff --git a/src/app/services.ts b/src/app/services.ts index 0d6173d830..a3205b4c4a 100644 --- a/src/app/services.ts +++ b/src/app/services.ts @@ -2,6 +2,8 @@ import { ArtifactStorageService } from "../features/artifact-storage/service"; 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, @@ -61,6 +63,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); @@ -82,7 +98,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/agents-runner.ts b/src/features/agents/agents-runner.ts new file mode 100644 index 0000000000..ced2fc15a1 --- /dev/null +++ b/src/features/agents/agents-runner.ts @@ -0,0 +1,571 @@ +// The Runner: the part of agents that runs them. It decides which agents run, +// and owns the agent runners and the agent storage they keep, which is +// records, catch-up progress, turns in progress and synced config copies. +// What it needs from the Manager, the part the screens use, it asks through +// `RunnerManager`; those facts and requests are what will cross the agents +// link when the Runner moves to its own page. +import { bytesToHex } from "nostr-tools/utils"; +import { AgentConfigStore } from "./agent-config"; +import { excerpt, mentionFacts, type ClassifierContext } from "./classifier"; +import type { Start } from "./agent-start"; +import { + readCheckpoints, + splitLegacyCheckpoints, + writeCheckpoints, + type Checkpoints, +} from "./catch-up"; +import { originOf, RunnerConnections } from "./connections"; +import { CoreMemoryCache } from "./core-memory-cache"; +import { readInterrupted, writeInterrupted } from "./interrupted"; +import type { LocalAgentActivity } from "./local-activity"; +import { pubkeyOf, type AgentKey } from "./membership"; +import { migrateWatchFilters } from "./migrate-filter"; +import { migrateWatchKeywords } from "./migrate-keywords"; +import { migrateAgentStorage } from "./migrate-storage"; +import type { AgentIdentity } from "./native"; +import { ownerReads } from "./reads"; +import { AgentRecords, EMPTY, type Member } from "./records"; +import { AgentRunners, type Connection, type RunnerHost } from "./runner"; +import type { + Agent, + AgentWake, + HandoffSource, + RegisteredAgentType, +} from "./service"; +import { soloChat } from "./solo-chat"; +import { nameSoloChat } from "./solo-chat-name"; +import type { AgentStore } from "./state"; +import type { AgentRecord } from "./store"; +import { relayPartition } from "../relay/partition"; +import type { RelayEvent } from "../relay/events"; +import type { RelayData } from "../relay/service"; +import { threadReference } from "../relay/thread-reference"; +import type { RelaySession } from "../relay/session"; +import type { Decisions } from "../decisions/service"; +import { channelTabs } from "../../bundled/channels/useChannelTabState"; +import type { Navigation } from "../navigation/controller"; + +/** What the Runner asks of the Manager. */ +export type RunnerManager = Pick< + RunnerHost, + | "team" + | "createAgent" + | "updateAgent" + | "templates" + | "classifier" + | "attentionShown" + | "pluginsSettled" + | "protection" +> & + Readonly<{ + /** Who runs which agent changed. */ + ownershipChanged(): void; + /** 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[]; + /** The settings a new agent of this type starts with. */ + defaults(pubkey: string, type: RegisteredAgentType): AgentRecord; + /** What the agent starts with: `local`, its home record's config, with + * each synced side's confirmed copy over it. */ + resolve(agent: Agent, local: unknown): Start; + /** A run of the agent ended. */ + finished(key: AgentKey): void; + /** Where an agent's closed tabs leave the viewer. */ + navigation?: Navigation | undefined; + }>; + +export class AgentsRunner { + /** 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. */ + readonly records: AgentRecords; + readonly coreMemory: CoreMemoryCache; + readonly runners: AgentRunners; + /** The Runner's own live session of each community it runs agents in. */ + readonly connections: RunnerConnections; + /** Confirmed synced config copies, the only synced launch source. */ + readonly configStore: AgentConfigStore; + /** The agents running here, in every community, and those of them with + * turns the app left working last time. */ + private state: Readonly<{ + running: readonly Agent[]; + interrupted: readonly Agent[]; + }> = { running: [], interrupted: [] }; + /** The memberships it last decided from. */ + private identities: readonly Member[] = []; + private readonly listeners = new Set<() => void>(); + + constructor( + storage: AgentStore, + private readonly custody: RunnerHost["custody"], + activity: LocalAgentActivity, + decisions: Decisions | undefined, + private readonly manager: RunnerManager, + ) { + // Each changes the copy at once and logs a failed save itself. + void migrateAgentStorage(storage); + void migrateWatchFilters(storage); + void migrateWatchKeywords(storage); + this.records = new AgentRecords(storage, { + defaults: ({ pubkey, type }) => { + const kind = manager.types().find((entry) => entry.key === type); + if (kind) + try { + return manager.defaults(pubkey, kind); + } catch (error) { + console.warn(`Agent type ${type} has invalid defaults`, error); + } + }, + changed: () => manager.changed(), + // Its timers waited for this change, saved or failed. + settled: (key) => this.runners.wake(key), + }); + void splitLegacyCheckpoints(storage); + this.configStore = new AgentConfigStore(storage); + this.connections = new RunnerConnections({ + open: (origin) => manager.open(origin), + listen: (session, origin, connection) => + this.listen(session, origin, connection), + // The Manager's own session may change first: what runs is shown and + // reconciled again once this one has. + changed: () => manager.changed(), + }); + this.coreMemory = new CoreMemoryCache((key) => + ownerReads(this.relay(key), pubkeyOf(key)), + ); + /** The catch-up progress this program last read or saved. */ + let savedProgress: Checkpoints = {}; + this.runners = new AgentRunners({ + ownershipChanged: () => { + this.notify(); + manager.ownershipChanged(); + }, + ready: (key) => this.ready(key), + agent: (key) => this.agent(key), + member: (agent, event) => this.member(agent, event), + dm: (agent, event) => this.dm(agent, event), + starting: (agent, channelId) => + this.connections + .binding(agent) + ?.session.channelWrites.starting(channelId), + community: (agent) => this.connections.binding(agent), + records: () => this.records.all(), + checkpoints: { + read: () => (savedProgress = readCheckpoints(storage)), + write: (checkpoints) => { + void writeCheckpoints(storage, checkpoints, savedProgress).catch( + (error) => { + console.warn("Agent catch-up progress was not saved", error); + // The copy went back to what is saved: the next save compares + // with that. + savedProgress = readCheckpoints(storage); + }, + ); + savedProgress = checkpoints; + }, + }, + interrupted: { + read: () => readInterrupted(storage), + write: (key, turns) => { + void writeInterrupted(storage, key, turns).catch((error) => + console.warn("Agent turns in progress were not saved", error), + ); + }, + changed: () => manager.changed(), + }, + scope: (agent) => this.connections.binding(agent)?.scope, + saved: (key) => this.records.saved(key), + attentionShown: () => manager.attentionShown(), + pluginsSettled: () => manager.pluginsSettled(), + record: (key) => { + const member = this.liveMember(key); + if (!member) throw new Error("No such agent on this device"); + return this.records.recordOf(member); + }, + change: async (key, apply) => { + 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.records.recordOf(member)); + }); + if (!record) throw new Error("No such agent on this device"); + return record; + }, + changing: (key) => this.records.isChanging(key), + timers: (key) => this.records.timersOf(key), + progress: (key, ran) => this.records.progress(key, ran), + start: (agent) => + manager.resolve( + agent, + (this.records.saved(agent.pubkey) ?? agent).config, + ), + team: (key) => manager.team(key), + protection: (agent) => manager.protection?.(agent), + agentTypes: () => + manager + .types() + .filter((type) => !type.remote) + .map(({ key, title }) => ({ key, title })), + createAgent: (agent, input) => manager.createAgent(agent, input), + updateAgent: (agent, input) => manager.updateAgent(agent, input), + handoff: (agent, { from, ...wake }) => + this.wake(agent.key, wake, { from }), + templates: (agent) => manager.templates(agent), + tabs: (agent) => { + const session = this.connections.binding(agent)?.session; + return session && channelTabs(session, manager.navigation); + }, + classifier: () => manager.classifier(), + decisions: () => decisions, + context: (agent, event) => this.context(agent, event), + displayName: (agent, pubkey) => this.displayName(agent, pubkey), + custody, + activity, + soloChat: (agent, channel, signal) => + soloChat(() => this.relay(agent.key), agent.owner, channel, signal), + remembered: (key, slug, body) => this.coreMemory.wrote(key, slug, body), + woke: (key, object, events) => + this.records.history( + key, + object, + events + ?.map( + (event) => + `${new Date(event.created_at * 1000).toISOString()} · ${event.pubkey}\n${event.content}`, + ) + .join("\n\n"), + Math.floor(Date.now() / 1000), + ), + finished: (key) => { + void this.coreMemory.refresh(key); + manager.finished(key); + }, + named: (agent, event, title) => + void nameSoloChat( + this.relay(agent.key), + agent.owner, + agent.pubkey, + event, + title, + ).catch((error) => console.warn("Couldn't name the Solo chat", error)), + }); + } + + /** 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); + } + /** Decides which agents run: each of the viewer's agents whose community is + * joined and not disconnected. A reconnect does not stop one. An unfinished + * Delete or Remove stops it, and its types' plugins stop what they keep for + * it. Then each runner goes on with what it can do now. */ + update(identities: readonly Member[], viewer: string | undefined) { + this.identities = identities; + const live = identities.filter( + (identity) => !identity.deleted && !identity.deleting, + ); + const running: Agent[] = []; + const shown = this.manager.attentionShown(); + for (const identity of live) { + const status = this.connections + .relay(originOf(identity.relay)) + ?.snapshot().status; + if (identity.owner !== viewer || !status || status === "disconnected") + continue; + const record = this.records.recordOf(identity); + // Reuse the frozen agent while nothing it shows changed, so subscribers + // see a stable snapshot. + const prior = this.agent(identity.key); + running.push( + prior && + prior.name === identity.name && + prior.picture === identity.picture && + prior.type === identity.type && + prior.protectionRevision === (identity.protectionRevision ?? 0) && + prior.attention === record.attention && + prior.skipped === (record.skipped ?? EMPTY) && + prior.timers === (record.timers ?? EMPTY) && + prior.sourceRuns === record.sourceRuns && + prior.attentionHistory === record.attentionHistory && + prior.attentionEnabled === + (record.attentionEnabled === true && shown) && + prior.config === record.config + ? prior + : view(identity, record, shown), + ); + } + this.runners.reconcile( + live.map(({ key, pubkey, relay, type, owner, protectionRevision }) => ({ + key, + pubkey, + relay, + type, + protectionRevision, + scope: relayPartition(originOf(relay) ?? relay, owner), + })), + new Map(this.manager.types().map((type) => [type.key, type])), + ); + const left = this.runners.interrupted(); + // Only the copy that holds an agent may resume or forget its turns. + const interrupted = running.filter( + (agent) => left.has(agent.key) && this.runners.runs(agent.key), + ); + const prior = this.state; + this.state = { + running: same(running, prior.running) + ? prior.running + : Object.freeze(running), + interrupted: same(interrupted, prior.interrupted) + ? prior.interrupted + : Object.freeze(interrupted), + }; + if (this.state.running !== prior.running) this.notify(); + this.runners.resume( + new Set( + running + .filter((agent) => this.connections.binding(agent)) + .map((agent) => agent.key), + ), + ); + this.runners.catchUp(false); + } + /** The agents running here, in every community. */ + running() { + return this.state.running; + } + /** The running agents with turns the app left working last time. */ + interrupted() { + return this.state.interrupted; + } + /** A running agent, in any community. */ + agent(key: AgentKey) { + return this.state.running.find((agent) => agent.key === key); + } + /** The agent, while its community's session is live to run it in. */ + ready(key: AgentKey) { + const agent = this.agent(key); + return agent && this.connections.binding(agent) ? agent : undefined; + } + /** The membership of this key, unless it was deleted. */ + liveMember(key: AgentKey) { + return this.identities.find( + (identity) => identity.key === key && !identity.deleted, + ); + } + /** Queues a plugin's wake, or the agent's own handoff, in the agent's + * mention lane. Throws when this copy does not run the agent, or the lane + * is full. */ + wake( + key: AgentKey, + { channel, root, text }: AgentWake, + source: Readonly<{ plugin: string } | { from: HandoffSource }>, + ) { + if (!channel || !text.trim()) + throw new Error("A wake needs a channel and text"); + // Harnesses find the thread from a well-formed id; any other would leave + // them answering at the wake's own id. + if (root !== undefined && !/^[0-9a-f]{64}$/i.test(root)) + throw new Error("A wake's root must be the thread's 64-hex event id"); + this.runners.wakeIn(key, { + type: "plugin", + ...source, + channel, + ...(root ? { root: root.toLowerCase() } : {}), + text, + id: bytesToHex(crypto.getRandomValues(new Uint8Array(32))), + at: Date.now(), + }); + } + /** The running agent's community, while the Runner is connected to it. */ + relay(key: AgentKey) { + return this.connections.relay(originOf(this.agent(key)?.relay ?? "")); + } + /** Stops it all without reconciling, which would make new runners. */ + clear() { + this.connections.clear(); + this.runners.clear(); + this.state = { running: [], interrupted: [] }; + } + + /** Delivers the session's live events, and retries what failed whenever its + * connection recovers, which happens within the same session. A subscription + * that comes back, after a reconnect or a failed route, first sends its + * recent history again: what in it is new since this listener started was + * missed while it was down, and is delivered now (`seen` drops the rest). + * Older history is catch-up's, not replay's. */ + private listen( + session: RelaySession, + origin: string, + connection: Connection, + ) { + let connected = session.live.snapshot().status === "connected"; + const since = connection.start; + if (connected) Object.assign(connection, { epoch: 1, since }); + const heard = (agent: Agent) => originOf(agent.relay) === origin; + const stopEvents = session.subscribeLive( + (batch) => this.runners.dispatch(batch, heard, since), + { replay: true }, + ); + const stopStatus = session.live.subscribe(() => { + const now = session.live.snapshot().status === "connected"; + if (now !== connected) { + // Catch-up reads what this connect may have missed, up to now. + connection.epoch += now ? 1 : 0; + connection.since = now ? Math.floor(Date.now() / 1000) : undefined; + } + if (now && !connected) this.refreshCore(origin); + connected = now; + this.runners.catchUp(false); + }); + // A roster the relay confirms decides the events held for it. + const stopRosters = session.channels.subscribeList?.(() => + this.runners.releaseHeld(heard), + ); + return () => { + stopEvents(); + stopStatus(); + stopRosters?.(); + }; + } + /** Refreshes the cached `core` memory of each agent running in the + * community, which may have changed elsewhere while it was not connected. */ + refreshCore(origin: string) { + for (const agent of this.state.running) + if (originOf(agent.relay) === origin) + void this.coreMemory.refresh(agent.key); + } + /** The event's surroundings for the classifier, from what the app knows. */ + private async context( + agent: Agent, + event: RelayEvent, + ): Promise { + const channelId = event.tags.find((tag) => tag[0] === "h")?.[1]; + const channels = this.connections.binding(agent)?.session.channels; + const summary = channelId + ? (channels?.list().channels.find((item) => item.id === channelId) ?? + channels?.get?.(channelId)) + : undefined; + const parent = threadReference(event)?.parentId ?? null; + let found: RelayEvent | undefined; + const native = this.custody; + if (parent && native) + try { + // Bounded: classified events release in order, so a stuck lookup + // would hold back every later one. + let timer: ReturnType | undefined; + [found] = await Promise.race([ + native.query(agent, [{ ids: [parent], limit: 1 }]), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error("The parent lookup timed out")), + PARENT_LOOKUP_MS, + ); + }), + ]).finally(() => clearTimeout(timer)); + } catch { + // The parent stays unknown; the classifier still runs. + } + return { + channel: { name: summary?.name ?? null }, + thread: { + is_reply: !!parent, + parent: found + ? { + author: { + role: + found.pubkey === agent.pubkey + ? "self" + : found.pubkey === agent.owner + ? "owner" + : "other", + name: this.displayName(agent, found.pubkey), + }, + content: excerpt(found.content), + } + : null, + }, + mentions: mentionFacts(event.tags, agent.pubkey, agent.owner), + }; + } + private displayName(agent: Agent, pubkey: string) { + return ( + this.connections.binding(agent)?.session.profiles?.snapshot().get(pubkey) + ?.name ?? null + ); + } + /** The event's channel as the agent's community knows it, if at all. */ + private channelOf(agent: Agent, event: Pick) { + const channel = event.tags.find((tag) => tag[0] === "h")?.[1]; + const channels = this.connections.binding(agent)?.session.channels; + if (!channel || !channels) return undefined; + return ( + channels.list().channels.find((item) => item.id === channel) ?? + channels.get?.(channel) + ); + } + /** Whether the agent is in the event's channel, by its relay-signed roster. + * The stream is the owner's, so a watch, even of "all", sees only what the + * agent itself could read; an event with no channel, or whose roster is + * unknown, wakes no watch. A roster only loaded from disk decides nothing + * (undefined): the agent may have been removed or added since it was saved, + * so the event waits until the relay confirms it. */ + private member( + agent: Agent, + event: Pick, + ): boolean | undefined { + const summary = this.channelOf(agent, event); + if (summary?.cached) return undefined; + return !!summary?.members?.includes(agent.pubkey); + } + /** Whether the event's channel is a DM of exactly the agent and one other, + * by the same roster. A group DM, or one whose roster is unknown, is not. */ + private dm(agent: Agent, event: RelayEvent) { + const summary = this.channelOf(agent, event); + return ( + summary?.channelType === "dm" && + summary.members?.length === 2 && + summary.members.includes(agent.pubkey) + ); + } +} + +/** How long the classifier waits for a reply's parent before it goes on. */ +const PARENT_LOOKUP_MS = 5_000; +const same = (next: readonly Agent[], prior: readonly Agent[]) => + next.length === prior.length && + next.every((agent, index) => agent === prior[index]); +/** A membership as it runs here. */ +/** `shown`: whether a plugin shows Attention, without which it is off. */ +function view(identity: Member, record: AgentRecord, shown: boolean): Agent { + return Object.freeze({ + key: identity.key, + pubkey: identity.pubkey, + name: identity.name, + ...(identity.picture && { picture: identity.picture }), + type: identity.type, + owner: identity.owner, + relay: identity.relay, + place: "here", + ...(identity.protection ? { protection: identity.protection } : {}), + protectionRevision: identity.protectionRevision ?? 0, + attention: record.attention, + skipped: record.skipped ?? EMPTY, + timers: record.timers ?? EMPTY, + ...(record.sourceRuns ? { sourceRuns: record.sourceRuns } : {}), + attentionHistory: record.attentionHistory, + attentionEnabled: record.attentionEnabled === true && shown, + config: record.config, + }); +} diff --git a/src/features/agents/attention-command.ts b/src/features/agents/attention-command.ts index ce03117683..75915acddc 100644 --- a/src/features/agents/attention-command.ts +++ b/src/features/agents/attention-command.ts @@ -8,6 +8,18 @@ export const bundledAttention = { ], } as const; +/** The provider a plugin's source `id` names in its watches: a bundled + * provider's own name, or the contribution's key. */ +export function attentionProvider( + source: Readonly<{ key: string; pluginId: string; id: string }>, +) { + return ( + bundledAttention.providers.find( + ({ plugin, id }) => source.pluginId === plugin && source.id === id, + )?.provider ?? source.key + ); +} + type Sources = { addAttentionSource(source: AttentionSource): void; }; diff --git a/src/features/agents/connections.test.ts b/src/features/agents/connections.test.ts new file mode 100644 index 0000000000..33843f3d28 --- /dev/null +++ b/src/features/agents/connections.test.ts @@ -0,0 +1,71 @@ +import { expect, it, vi } from "vitest"; +import type { RelayData } from "../relay/service"; +import type { RelaySession } from "../relay/session"; +import { RunnerConnections } from "./connections"; +import type { AgentIdentity } from "./native"; + +const viewer = "b".repeat(64); +const identity: AgentIdentity = { + pubkey: "a".repeat(64), + relay: "wss://relay.example", + owner: viewer, + type: "test/agent", + name: "Ada", + deleted: false, +}; + +/** A community relay whose session goes live and down on demand. */ +function community() { + const listeners = new Set<() => void>(); + const session = {} as RelaySession; + let live = false; + const relay = { + snapshot: () => + live + ? { + status: "ready", + scope: "scope", + origin: "https://relay.example", + viewer, + session, + } + : { status: "connecting" }, + subscribe(listener: () => void) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + } as unknown as RelayData; + const set = (next: boolean) => { + live = next; + for (const listener of [...listeners]) listener(); + }; + return { relay, session, set }; +} + +it("hears a session while it is live, and says each time it starts or ends", () => { + const { relay, session, set } = community(); + const stopLive = vi.fn(); + const listen = vi.fn(() => stopLive); + const changed = vi.fn(); + const connections = new RunnerConnections({ + open: () => relay, + listen, + changed, + }); + connections.watch([identity], viewer); + expect(connections.binding(identity)).toBeUndefined(); + set(true); + expect(listen).toHaveBeenCalledWith( + session, + "https://relay.example", + expect.objectContaining({ epoch: 0 }), + ); + expect(changed).toHaveBeenCalledTimes(1); + // The same session again changes nothing. + set(true); + expect(changed).toHaveBeenCalledTimes(1); + set(false); + expect(stopLive).toHaveBeenCalled(); + expect(connections.binding(identity)).toBeUndefined(); + expect(changed).toHaveBeenCalledTimes(2); +}); diff --git a/src/features/agents/connections.ts b/src/features/agents/connections.ts new file mode 100644 index 0000000000..590ec501b0 --- /dev/null +++ b/src/features/agents/connections.ts @@ -0,0 +1,146 @@ +// The Runner's own connections: the live session of each community it runs +// agents in, and the listener that hears that community's events for them. +// The Manager keeps its own connections for listings and cleanup. +import { communityDestination } from "../communities/destination"; +import { relayPartition } from "../relay/partition"; +import type { RelayData } from "../relay/service"; +import type { RelaySession } from "../relay/session"; +import type { AgentIdentity } from "./native"; +import type { Connection } from "./runner"; + +/** A community's session while it is live, as the Runner's agents use it. */ +export type RunnerBinding = Readonly<{ + scope: string; + origin: string; + viewer: string; + session: RelaySession; + /** Aborts when this binding is replaced, so no work outlives its session. */ + controller: AbortController; + connection: Connection; + stopLive: () => void; +}>; +/** The origin a relay URL or community ID names, if it is a valid one. */ +export const originOf = (value: string) => { + try { + return communityDestination(value).url; + } catch { + return undefined; + } +}; + +type Entry = { + relay: RelayData; + stop: () => void; + binding?: RunnerBinding | undefined; +}; + +export type ConnectionsHost = Readonly<{ + /** The community's relay, while the viewer is in it. */ + open(origin: string): RelayData | undefined; + /** Hears the live session's events until the returned stop. */ + listen( + session: RelaySession, + origin: string, + connection: Connection, + ): () => void; + /** The community's live session started, was replaced or ended. */ + changed(origin: string): void; +}>; + +export class RunnerConnections { + private readonly bound = new Map(); + + constructor(private readonly host: ConnectionsHost) {} + + /** Connects each community where the viewer has an agent; lets go of the + * rest. */ + watch(identities: readonly AgentIdentity[], viewer: string | undefined) { + const origins = new Set(); + for (const identity of identities) { + const origin = + !identity.deleted && + identity.owner === viewer && + originOf(identity.relay); + if (origin) origins.add(origin); + } + // Left, or left and joined again with a new session. + for (const [origin, entry] of this.bound) + if (!origins.has(origin) || this.host.open(origin) !== entry.relay) + this.unbind(origin); + for (const origin of origins) { + if (this.bound.has(origin)) continue; + const relay = this.host.open(origin); + if (!relay) continue; + this.bound.set(origin, { + relay, + stop: relay.subscribe(() => this.bind(origin)), + }); + this.bind(origin); + } + } + /** The live binding of the agent's community, if it has one now. */ + binding({ relay, owner }: Pick) { + const origin = originOf(relay); + const binding = origin ? this.bound.get(origin)?.binding : undefined; + return binding?.scope === relayPartition(origin ?? "", owner) + ? binding + : undefined; + } + /** The community's relay, while the Runner is connected to it. */ + relay(origin: string | undefined) { + return origin ? this.bound.get(origin)?.relay : undefined; + } + /** Each community with its live binding, if it has one. */ + entries() { + return [...this.bound].map( + ([origin, entry]) => [origin, entry.binding] as const, + ); + } + clear() { + for (const origin of [...this.bound.keys()]) this.unbind(origin); + } + + private unbind(origin: string) { + const entry = this.bound.get(origin); + this.bound.delete(origin); + entry?.stop(); + entry?.binding?.stopLive(); + entry?.binding?.controller.abort(); + } + private bind(origin: string) { + const entry = this.bound.get(origin); + if (!entry) return; + const snapshot = entry.relay.snapshot(); + const live = + snapshot.status === "ready" && + !snapshot.cached && + snapshot.scope && + snapshot.origin && + snapshot.viewer + ? { + scope: snapshot.scope, + origin: snapshot.origin, + viewer: snapshot.viewer, + session: snapshot.session, + controller: new AbortController(), + } + : undefined; + if ( + entry.binding?.session === live?.session && + entry.binding?.scope === live?.scope + ) + return; + entry.binding?.stopLive(); + entry.binding?.controller.abort(); + const connection: Connection = { + epoch: 0, + start: Math.floor(Date.now() / 1000), + }; + entry.binding = live && { + ...live, + connection, + stopLive: this.host.listen(live.session, origin, connection), + }; + this.host.changed(origin); + } +} 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/presence.ts b/src/features/agents/presence.ts new file mode 100644 index 0000000000..b842390431 --- /dev/null +++ b/src/features/agents/presence.ts @@ -0,0 +1,114 @@ +// Agent presence: the Runner tells each community which agents this copy of +// the app runs there. It is the Runner's, as it knows what runs. +import type { RelaySession } from "../relay/session"; + +/** How often agents are said to be online; the relay keeps it 180 seconds. */ +const PRESENCE_MS = 60_000; +/** How soon a failed presence is sent again. */ +const RETRY_MS = 30_000; +/** The most agents one presence event names. */ +const PRESENCE_BATCH = 256; + +/** A community's live connection, while it has one. */ +export type PresenceConnection = Readonly<{ + session: RelaySession; + controller: AbortController; +}>; +export type PresenceHost = Readonly<{ + /** By origin, the pubkeys of the agents this copy runs there. */ + running(): ReadonlyMap>; + /** Each community an agent is in, by origin, with its live connection. */ + communities(): Iterable; + /** Another copy of the app took over the agent's membership there. */ + yielded(origin: string, pubkey: string): boolean; +}>; + +export class AgentPresence { + /** By origin: the agents last said to be online there, and when. */ + private readonly sent = new Map< + string, + { + session: RelaySession; + online: ReadonlySet; + at: number; + /** The session cannot send presence, so it is not tried again. */ + unable?: boolean; + } + >(); + /** Origins whose presence is being sent. */ + private readonly announcing = new Set(); + + constructor(private readonly host: PresenceHost) {} + + /** Tells each community which agents this copy runs there: online every + * minute while it does, and offline once it stops, unless another copy took + * over. Offline is best effort; the relay lets a lapsed online expire. */ + announce() { + const online = this.host.running(); + const communities = new Map(this.host.communities()); + for (const origin of this.sent.keys()) + if (!communities.has(origin)) this.sent.delete(origin); + const now = Date.now(); + for (const [origin, binding] of communities) { + if (!binding || this.announcing.has(origin)) continue; + const next = online.get(origin) ?? new Set(); + const prior = this.sent.get(origin); + const same = prior?.session === binding.session; + if (same && prior.unable) continue; + // Another copy that took an agent over says it is online; offline from + // this one could arrive after that, so forget it was online here. + const was = new Set( + [...(prior?.online ?? [])].filter( + (pubkey) => next.has(pubkey) || !this.host.yielded(origin, pubkey), + ), + ); + if (prior && was.size < prior.online.size) + this.sent.set(origin, { ...prior, online: was }); + const gone = [...was].filter((pubkey) => !next.has(pubkey)); + const due = !same || now - prior.at >= PRESENCE_MS; + const changed = + gone.length > 0 || [...next].some((pubkey) => !was.has(pubkey)); + if (!changed && (!due || !next.size)) continue; + this.announcing.add(origin); + const sent = { session: binding.session, online: next, at: now }; + this.sent.set(origin, sent); + void (async () => { + for (const [status, agents] of [ + ["offline", gone], + ["online", [...next]], + ] as const) + for (let at = 0; at < agents.length; at += PRESENCE_BATCH) + if ( + !(await binding.session.publishAgentPresence( + status, + agents.slice(at, at + PRESENCE_BATCH), + binding.controller.signal, + )) + ) + return false; + return true; + })() + .then((ok) => { + if (ok || this.sent.get(origin) !== sent) return; + console.warn("Agent presence cannot be sent on this connection"); + this.sent.set(origin, { ...sent, unable: true }); + }) + .catch((error) => { + if (binding.controller.signal.aborted) return; + console.warn("Agent presence was not sent", error); + // Try again sooner than the minute, but not every tick. + if (this.sent.get(origin) === sent) + this.sent.set(origin, { + ...sent, + at: Date.now() - PRESENCE_MS + RETRY_MS, + }); + }) + .finally(() => this.announcing.delete(origin)); + } + } + + /** Forgets what was sent, as the app stops. */ + clear() { + this.sent.clear(); + } +} 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..8c4b1d008d --- /dev/null +++ b/src/features/agents/records.ts @@ -0,0 +1,365 @@ +// 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, + sourceCompleted, + sourceSignature, + 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" | "watchEnabled" | "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, + ); + for (const [slug, { enabled, expected }] of Object.entries( + change.watchEnabled ?? {}, + )) { + const current = next.attention[slug]; + if (!current || JSON.stringify(current) !== JSON.stringify(expected)) + throw new Error( + "This watch changed. Review its current settings and try again.", + ); + if (current.value.type === "interest") throw new Error("Choose a watch"); + if (enabled && current.value.type === "timer" && !current.value.enabled) + throw new Error("Ask the agent to rearm this timer."); + if (current.value.enabled !== enabled) { + const history = next.attentionHistory; + const completed = sourceCompleted(next.sourceRuns, current); + next = setAttention(next, slug, { ...current.value, enabled }); + if (history) next = { ...next, attentionHistory: history }; + const updated = next.attention[slug]; + if (completed && updated) + next = { + ...next, + sourceRuns: { ...next.sourceRuns, [slug]: sourceSignature(updated) }, + }; + } + } + // 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.watchEnabled || + own.attentionEnabled !== undefined + ) + await this.change(key, (saved) => + changed(saved ?? this.recordOf(member), own), + ); + return; + } + } + } + + /** Changes one agent's record. This is the only way a record changes: + * owner settings, the agent's attention tools, timer progress, a provider + * change and removal all send a change here. One agent's changes run in + * the order they are sent, each after the one before it is saved or has + * failed, so each one applies to the newest saved record, never to a copy + * another change replaces. `apply` gets that record (or undefined) and + * returns the next one, or null to remove it; it may throw to change + * nothing. The owner and the agent see a change once it is saved, and the + * promise settles then. When the save fails, the promise rejects and they + * keep the saved record, so a form keeps its unsaved changes. `saved` + * runs as the change is shown, before anyone is told. Only the Runner + * passes a function; the Manager sends an `edit`. */ + change( + key: AgentKey, + apply: ( + record: AgentRecord | undefined, + ) => AgentRecord | null | Promise, + saved?: (record: AgentRecord | null) => void, + ): Promise { + this.changing.set(key, (this.changing.get(key) ?? 0) + 1); + return this.queued(key, async () => { + try { + const base = this.records[key]; + const next = await apply(base); + await this.commit(key, base, next); + saved?.(next); + this.host.changed(); + return next; + } finally { + const left = (this.changing.get(key) ?? 1) - 1; + if (left) this.changing.set(key, left); + else { + this.changing.delete(key); + // Its timers waited for this change, saved or failed. + this.host.settled(key); + } + } + }); + } + /** Records that timers ran or started counting, as a change to their run + * state only. It applies to whatever the record is when its turn comes, so + * it never undoes a change sent before it: when a timer was rearmed or + * restarted meanwhile, its new deadline stays and only the occurrences it + * used are added. Timers see it at once (`timersOf`), and memory keeps it + * even when its save fails, so a timer does not run again at once. */ + progress(key: AgentKey, ran: TimerProgress) { + const pending = this.progressing.get(key) ?? []; + this.progressing.set(key, [...pending, ran]); + void this.queued(key, async () => { + try { + const base = this.records[key]; + const next = base && withProgress(base, ran); + if (!base || !next) return; + await this.commit(key, base, next).catch((error) => { + console.warn(`Agent ${key} timer state was not saved`, error); + this.records = { ...this.records, [key]: next }; + }); + this.host.changed(); + } finally { + const left = (this.progressing.get(key) ?? []).filter( + (entry) => entry !== ran, + ); + if (left.length) this.progressing.set(key, left); + else this.progressing.delete(key); + } + }); + } + /** Keeps what an attention object's last wake and its source printed, while + * the object is the same. */ + history( + key: AgentKey, + object: AttentionObject, + output?: string, + lastWokeAt?: number, + ) { + void this.queued(key, async () => { + const record = this.records[key]; + if (!record || record.attention[object.slug] !== object) return; + const prior = record.attentionHistory?.[object.slug]; + const entry = { + ...prior, + ...(lastWokeAt !== undefined ? { lastWokeAt } : {}), + ...(output !== undefined + ? { + output: [prior?.output, output] + .filter((part) => part !== undefined && part !== "") + .join(prior?.output?.endsWith("\n") ? "" : "\n") + .slice(-16384) + .split("\n") + .slice(-40) + .join("\n"), + } + : {}), + }; + await this.commit(key, record, { + ...record, + attentionHistory: { ...record.attentionHistory, [object.slug]: entry }, + }); + this.host.changed(); + }).catch((error) => console.warn("Attention history was not saved", error)); + } + /** The agent's timers' run state as its timers go by: saved, with the + * progress still being saved. */ + timersOf(key: AgentKey) { + const record = this.records[key]; + const pending = this.progressing.get(key); + if (!record || !pending) return record?.timers ?? EMPTY; + return ( + pending.reduce((next, ran) => withProgress(next, ran) ?? next, record) + .timers ?? EMPTY + ); + } + /** The agent's home record, which keeps its config: as saved or, without + * one, its type's defaults, kept in memory until a save stores them. + * Without its type there are no defaults yet. */ + settingsOf(identity: Pick) { + const { pubkey } = identity; + if (!this.records[pubkey]) { + const record = this.host.defaults(identity); + if (record) this.records = { ...this.records, [pubkey]: record }; + } + return this.records[pubkey] ?? blank(pubkey); + } + /** The membership's settings as its owner and agent see them: its own + * attention, starting with none, and the agent's config. */ + recordOf(member: Member): AgentRecord { + const settings = this.settingsOf(member); + if (member.key === member.pubkey) return settings; + const own = this.records[member.key] ?? blank(member.key); + return { ...own, config: settings.config }; + } + + /** Runs `step` after the agent's earlier changes have settled. */ + private queued(key: AgentKey, step: () => Promise) { + const run = (this.changes.get(key) ?? Promise.resolve()).then(step); + const tail = run.then( + () => {}, + () => {}, + ); + this.changes.set(key, tail); + void tail.then(() => { + if (this.changes.get(key) === tail) this.changes.delete(key); + }); + return run; + } + /** Saves `next` over `base` in the membership's own key, then keeps it. + * Only this key: others may be another program's to change. A record + * other than the agent's home record keeps no config, which is the + * agent's, never one community's. */ + private async commit( + key: AgentKey, + base: AgentRecord | undefined, + given: AgentRecord | null, + ) { + if (given === base) return; + const next = given && key !== pubkeyOf(key) ? ownRecord(key, given) : given; + if (next) await writeRecord(this.storage, next); + else if (this.storage.get(recordKey(key)) !== null) + await removeRecord(this.storage, key); + const { [key]: _, ...others } = this.records; + this.records = next ? { ...others, [key]: next } : others; + } +} diff --git a/src/features/agents/runner-link.test.ts b/src/features/agents/runner-link.test.ts new file mode 100644 index 0000000000..79ef36e7d2 --- /dev/null +++ b/src/features/agents/runner-link.test.ts @@ -0,0 +1,78 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { link, type Link } from "./link"; +import { createLocalAgentActivity } from "./local-activity"; +import { linkedActivity, serveActivity } from "./runner-link"; + +const open: Link[] = []; +afterEach(() => { + for (const side of open.splice(0)) side.close(); +}); + +function linked() { + const activity = createLocalAgentActivity(); + const channel = new MessageChannel(); + const runner = link(channel.port1, serveActivity(activity.reader)); + const manager = link(channel.port2); + open.push(runner, manager); + return { activity, source: linkedActivity(manager).source }; +} + +const request = { + scope: "wss://relay.example", + agent: "a".repeat(64), + channelId: "channel", + messageId: "message", + threadRootId: "message", +}; + +it("shows the Runner's local work, and keeps rows until they change", async () => { + const { activity, source } = linked(); + const heard = vi.fn(); + source.subscribeWork(heard); + const work = activity.begin(request); + work.started(); + work.step("Read", "notes.md"); + await vi.waitFor(() => + expect(source.work()).toMatchObject([ + { ...request, state: "working", step: { tool: "Read" } }, + ]), + ); + const rows = source.work(); + expect(Object.isFrozen(rows[0])).toBe(true); + heard.mockClear(); + // A step with the same text is no change, so nothing is sent. + work.step("Read", "notes.md"); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(heard).not.toHaveBeenCalled(); + expect(source.work()).toBe(rows); +}); + +it("stops a turn and hides an error through the Runner", async () => { + const { activity, source } = linked(); + const stop = vi.fn(async () => {}); + const running = activity.begin(request); + running.started(); + running.interruptible(stop); + await vi.waitFor(() => + expect(source.work()).toMatchObject([{ stoppable: true }]), + ); + await source.interrupt?.(request); + expect(stop).toHaveBeenCalledTimes(1); + await vi.waitFor(() => + expect(source.work()).toMatchObject([{ state: "stopped" }]), + ); + + const failed = activity.begin({ ...request, messageId: "other" }); + failed.finish({ ok: false, error: "No credits left" }); + await vi.waitFor(() => + expect(source.work()).toContainEqual( + expect.objectContaining({ messageId: "other", state: "error" }), + ), + ); + source.dismissError?.({ ...request, messageId: "other" }); + await vi.waitFor(() => + expect(source.work()).toContainEqual( + expect.objectContaining({ messageId: "other", dismissed: true }), + ), + ); +}); diff --git a/src/features/agents/runner-link.ts b/src/features/agents/runner-link.ts new file mode 100644 index 0000000000..a1c2fef57b --- /dev/null +++ b/src/features/agents/runner-link.ts @@ -0,0 +1,65 @@ +// What the Runner serves the Manager over the agents link. Each name here is +// part of the contract between the two owners, so add to it with care: only +// JSON crosses, and either side may be a page that reloads. +import type { Json, Link, Served } from "./link"; +import type { LocalAgentSource, LocalWork } from "./local-activity"; + +type WorkKey = Pick; + +const workKey = (args: Json): WorkKey => { + const { scope, agent, messageId } = args as WorkKey; + return { scope, agent, messageId }; +}; + +/** The Runner's side: the local work its agents report. */ +export function serveActivity(source: LocalAgentSource): Served { + return { + topics: { + "activity.work": () => ({ + get: () => source.work() as unknown as Json, + subscribe: (listener) => source.subscribeWork(listener), + }), + }, + commands: { + "activity.interrupt": async (args) => { + await source.interrupt?.(workKey(args)); + return null; + }, + "activity.dismissError": (args) => { + source.dismissError?.(workKey(args)); + return null; + }, + }, + }; +} + +/** The Manager's side: the Runner's local work, as the screens read it. Rows + * keep their identity until the Runner sends a change. */ +export function linkedActivity(runner: Link) { + let work: readonly LocalWork[] = Object.freeze([]); + const listeners = new Set<() => void>(); + const stop = runner.watch("activity.work", null, () => { + const rows = (runner.read("activity.work") ?? []) as unknown as LocalWork[]; + work = Object.freeze(rows.map((row) => Object.freeze(row))); + for (const listener of [...listeners]) listener(); + }); + const source: LocalAgentSource = { + work: () => work, + subscribeWork(listener) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + dismissError: (row) => + void runner + .call("activity.dismissError", workKey(row)) + .catch((error: unknown) => + console.warn("Could not hide this error", error), + ), + interrupt: async (row) => { + await runner.call("activity.interrupt", workKey(row)); + }, + }; + return { source, stop }; +} diff --git a/src/features/agents/runner-plugins.ts b/src/features/agents/runner-plugins.ts new file mode 100644 index 0000000000..fcce011e4d --- /dev/null +++ b/src/features/agents/runner-plugins.ts @@ -0,0 +1,348 @@ +// 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, readFile } 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 { attentionProvider, bundledAttention } from "./attention-command"; +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 ( + bundledAttention.providers.some( + ({ provider }) => provider === `${plugin}/${source.id}`, + ) + ) + throw new Error( + "Attention command sources are built into Buzz. Update the Attention UI plugin.", + ); + 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 (handle.protection) { + if (!host.spawn) + throw new Error("Protected file reading is unavailable"); + return readFile( + host.spawn.bind(host), + options.cwd ?? "~", + path, + signal, + handle.protection.processProtection, + ); + } + 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"); + if (handle.protection) + throw new Error( + "Protected agents cannot run external Attention commands until that execution path supports complete protection", + ); + 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((source) => { + const { description, configExample } = source; + return { + provider: attentionProvider(source), + ...(attentionProvider(source) !== source.key ? { builtIn: true } : {}), + ...(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( + attentionProvider(source), + 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.test.ts b/src/features/agents/service.test.ts index 6a23fb0cfb..552a7d6fec 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -336,11 +336,16 @@ function runnerGate() { * microtask queue is empty, so this does not depend on how fast runners are. */ const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); /** The service's synced config state, which tests set as the relay would. */ -const internals = (service: AgentsService) => - service as unknown as { - configStore: AgentConfigStore; +const internals = (service: AgentsService) => { + const parts = service as unknown as { + runner: { configStore: AgentConfigStore }; communityConfigs: Map; }; + return { + configStore: parts.runner.configStore, + communityConfigs: parts.communityConfigs, + }; +}; /** The events native published as an agent, of `kind`, in order. */ async function published(native: ReturnType, kind: number) { @@ -6563,9 +6568,11 @@ it("catches a kind 9 watch up on what it missed, oldest first in batches, and no vi.useFakeTimers({ toFake: ["Date"] }); try { const start = nowSeconds(); + // Before setup, so its connection starts at `start` even when the clock + // reaches the next second meanwhile. + vi.setSystemTime(start * 1000); const storage = memoryStorage(); const first = await setup({ storage, identities: [echo] }); - vi.setSystemTime(start * 1000); // Only this watch: the type's default one reads the channel too. await first.service.save(bot, { attentionEnabled: true, @@ -7135,9 +7142,11 @@ it("wakes nothing for what came before a watch was first caught up, as after an vi.useFakeTimers({ toFake: ["Date"] }); try { const start = nowSeconds(); + // Before setup, so its connection starts at `start` even when the clock + // reaches the next second meanwhile. + vi.setSystemTime(start * 1000); const storage = memoryStorage(); const first = await setup({ storage, identities: [echo] }); - vi.setSystemTime(start * 1000); await first.service.save(bot, { attentionEnabled: true, attention: { ...chat, "watch/channel": null }, @@ -7219,9 +7228,11 @@ it("does not catch up after a restart on a message it already heard live", async vi.useFakeTimers({ toFake: ["Date"] }); try { const start = nowSeconds(); + // Before setup, so its connection starts at `start` even when the clock + // reaches the next second meanwhile. + vi.setSystemTime(start * 1000); const storage = memoryStorage(); const first = await setup({ storage, identities: [echo] }); - vi.setSystemTime(start * 1000); await first.service.save(bot, { attentionEnabled: true, attention: { ...chat, "watch/channel": null }, diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index b0fb45e954..a7e8c219db 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1,15 +1,5 @@ import type {} from "../navigation/service"; -import { appClient, readFile } 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 { bundledAttention } from "./attention-command"; -import { sourceCompleted, sourceSignature } from "./attention"; +import { attentionProvider } from "./attention-command"; import { AgentProtectionService, type AgentProtection, @@ -17,12 +7,7 @@ import { } from "../agent-protection/service"; import { createLocalAgentActivity } from "./local-activity"; import { bindKeepAwake } from "./keep-awake"; -import { - AgentConfigStore, - type SettingsSchema, - type Subject, -} from "./agent-config"; -import { isSynced, type Mode, resolveStart } from "./agent-start"; +import { isSynced, resolveStart } from "./agent-start"; import { type CommunityConfig, type ConfigAgent, @@ -46,15 +31,15 @@ import type { Expected, SnapshotRead } from "./agent-config-artifacts"; import type { revertTemplate, SettingsCheck } from "./agent-config-history"; import type { RevertOutcome } from "./AgentConfigHistory"; import { nativeRelayInfo } from "../relay/native"; -import { AgentRunners, type Connection, TIMEOUT_LIMIT_MS } from "./runner"; -import { CoreMemoryCache } from "./core-memory-cache"; import { createAgentHooks, dispatchAgentHooks, type AgentHook, type AgentHookContext, } from "./hooks"; -import { ownerReads } from "./reads"; +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 @@ -68,9 +53,6 @@ import { ownerReads } from "./reads"; // attestation. Every agent of either kind is listed in the owner's kind 30177 // records with its type as the `runtime` tag, so every device of the owner's // can show it: as here, hosted, or on another device. -import { migrateAgentStorage } from "./migrate-storage"; -import { migrateWatchFilters } from "./migrate-filter"; -import { migrateWatchKeywords } from "./migrate-keywords"; import { agentStorage, type AgentStore } from "./state"; import { Service, type Context } from "@deepseek-ai/cordis"; import { npubEncode } from "nostr-tools/nip19"; @@ -91,7 +73,7 @@ import { removeAgentFromChannels, removeRelayAgent, } from "./relay-removal"; -import { communityDestination, relayOrigin } from "../communities/destination"; +import { relayOrigin } from "../communities/destination"; import type { Communities } from "../communities/service"; import { prepareRemoteAgentAuthorization, @@ -101,18 +83,16 @@ import { avatarSource } from "../../shared/avatar-source"; import { publicKeyLabels } from "../../shared/identity/public-key"; import { newer, type EventData, type RelayEvent } from "../relay/events"; import { relayPartition } from "../relay/partition"; -import { threadReference } from "../relay/thread-reference"; -import { channelTabs } from "../../bundled/channels/useChannelTabState"; +import type { 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, @@ -121,12 +101,7 @@ import type { WatchSummary, WriteOptions, } from "./attention-objects"; -import { - excerpt, - mentionFacts, - type ClassifierContext, - type ClassifierOutcome, -} from "./classifier"; +import type { ClassifierOutcome } from "./classifier"; import type { Decisions } from "../decisions/service"; import { nativeAgents, @@ -142,37 +117,25 @@ import { type PortSource, } from "./native"; 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 { readArchiveUnchecked, - readRecords, - recordKey, readTeams, readUnfinished, setAttention, writeArchiveUnchecked, - removeRecord, - splitLegacyRecords, - writeRecord, writeTeams, writeUnfinished, - withProgress, type AgentRecord, - type TimerProgress, type SkippedObject, type Team, type Unfinished, } from "./store"; -import { soloChat } from "./solo-chat"; -import { - readCheckpoints, - splitLegacyCheckpoints, - writeCheckpoints, - type Checkpoints, -} from "./catch-up"; -import { readInterrupted, writeInterrupted } from "./interrupted"; -import { AttentionSources, type AttentionProducer } from "./attention-sources"; -import { type AddedTool, BUZZ_SESSION } from "../../buzz-mcp/session-tools"; -import { nameSoloChat } from "./solo-chat-name"; +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 @@ -1315,92 +1278,11 @@ 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; /** How often a failed connection to an unselected community is tried again. */ const RETRY_MS = 30_000; -/** How often agents are said to be online; the relay keeps it 180 seconds. */ -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, - ); - for (const [slug, { enabled, expected }] of Object.entries( - change.watchEnabled ?? {}, - )) { - const current = next.attention[slug]; - if (!current || JSON.stringify(current) !== JSON.stringify(expected)) - throw new Error( - "This watch changed. Review its current settings and try again.", - ); - if (current.value.type === "interest") throw new Error("Choose a watch"); - if (enabled && current.value.type === "timer" && !current.value.enabled) - throw new Error("Ask the agent to rearm this timer."); - if (current.value.enabled !== enabled) { - const history = next.attentionHistory; - const completed = sourceCompleted(next.sourceRuns, current); - next = setAttention(next, slug, { ...current.value, enabled }); - if (history) next = { ...next, attentionHistory: history }; - const updated = next.attention[slug]; - if (completed && updated) - next = { - ...next, - sourceRuns: { ...next.sourceRuns, [slug]: sourceSignature(updated) }, - }; - } - } - // 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(); @@ -1409,8 +1291,6 @@ function defaults(pubkey: string, type: RegisteredAgentType) { record = setAttention(record, slug, value); return record; } -/** How long the classifier waits for a reply's parent before it goes on. */ -const PARENT_LOOKUP_MS = 5_000; const message = (error: unknown) => String(error instanceof Error ? error.message : error); /** The team sections for agent `self`: each team it is on, its members that @@ -1466,7 +1346,6 @@ type Binding = { /** Aborts when this binding is replaced, so no work outlives its session. */ controller: AbortController; stopLive: () => void; - connection: Connection; /** Its config support while no config channels are up: being checked, the * check failed (and when), or the relay has none. */ support?: "checking" | "unsupported" | Readonly<{ failed: number }>; @@ -1481,15 +1360,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 { - return communityDestination(value).url; - } catch { - return undefined; - } -}; /** One of the viewer's kind 30177 agent records. Ones without a `runtime` * are earlier agent systems', not shown; replacing one keeps its settings. */ type Listing = Readonly<{ @@ -1592,11 +1462,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[] = []; @@ -1604,15 +1469,6 @@ export class AgentsService extends Service implements Agents { /** Every saved membership, deleted ones included: native's list is the * agent list, and `records` only adds settings to it. */ 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 teamList: readonly Team[]; /** Synced teams as discovery last found them, by home community. */ private readonly syncedTeams: SyncedTeams; @@ -1623,8 +1479,10 @@ 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(); - private readonly runners: AgentRunners; - private readonly coreMemory: CoreMemoryCache; + /** 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, @@ -1666,23 +1524,41 @@ export class AgentsService extends Service implements Agents { private busy = new Set(); /** Hosted and elsewhere agents as last shown, kept while unchanged. */ private shown = new Map(); - /** By origin: the agents last said to be online there, and when. */ - private presence = new Map< - string, - { - session: RelaySession; - online: ReadonlySet; - at: number; - /** The session cannot send presence, so it is not tried again. */ - unable?: boolean; - } - >(); - /** Confirmed synced config copies, the only synced launch source. */ - private readonly configStore: AgentConfigStore; /** Each bound community's config channels, while config sync is on. */ private readonly communityConfigs = new Map(); - /** Origins whose presence is being sent. */ - private announcing = new Set(); + /** Says which agents this copy runs in each community. */ + private readonly presence = new AgentPresence({ + running: () => { + const online = new Map>(); + for (const agent of this.state.running) { + const origin = originOf(agent.relay); + if ( + origin && + this.bindingOf(agent) && + this.runner.runners.runs(agent.key) + ) + online.set( + origin, + (online.get(origin) ?? new Set()).add(agent.pubkey), + ); + } + return online; + }, + communities: () => + [...this.bound].map( + ([origin, entry]) => [origin, entry.binding] as const, + ), + // Each pubkey said online there is one membership's, the one there. + yielded: (origin, pubkey) => { + const key = this.identities.find( + (identity) => + identity.pubkey === pubkey && + !identity.deleted && + originOf(identity.relay) === origin, + )?.key; + return !!key && this.runner.runners.yielded(key); + }, + }); private readonly protection: AgentProtectionService; constructor( @@ -1709,27 +1585,6 @@ export class AgentsService extends Service implements Agents { this.presentations = createContributions(ctx); this.sourcePresentations = createContributions(ctx); - this.addedTools = createContributions(ctx); - this.attentionSources = createContributions(ctx); - // Each changes the copy at once and logs a failed save itself. - void migrateAgentStorage(storage); - void migrateWatchFilters(storage); - void migrateWatchKeywords(storage); - this.records = readRecords(storage); - void splitLegacyRecords(storage, this.records); - void splitLegacyCheckpoints(storage); - this.unfinished = Object.freeze(readUnfinished(storage)); - this.unarchiving = readArchiveUnchecked(storage); - this.teamList = Object.freeze(readTeams(storage)); - this.syncedTeams = new SyncedTeams(storage); - this.state = Object.freeze({ - status: native ? "loading" : "unavailable", - classifier: this.availability(), - agents: [], - running: [], - interrupted: [], - teams: this.teamList, - }); this.protection = new AgentProtectionService( ctx, () => ({ @@ -1767,7 +1622,7 @@ export class AgentsService extends Service implements Agents { throw new Error( "This runtime cannot safely change protection while Buzz is open", ); - await this.runners.retype(agent.key, agent.name, async () => { + await this.runner.runners.retype(agent.key, agent.name, async () => { for (const member of members) await type.retire?.(member.key); await setProtection(pubkey, revision, provider, policy); await this.refresh(); @@ -1775,75 +1630,36 @@ export class AgentsService extends Service implements Agents { }, }, ); - this.configStore = new AgentConfigStore(storage); - this.teamsChanged(); - // A synced copy that changes reaches running agents as a local save does. - this.configStore.subscribe(() => { - this.teamsChanged(); - this.update(); - }); - this.coreMemory = new CoreMemoryCache((key) => - ownerReads(this.relay(key), pubkeyOf(key)), - ); - /** The catch-up progress this program last read or saved. */ - let savedProgress: Checkpoints = {}; - this.runners = new AgentRunners({ - ownershipChanged: () => this.notify(), - ready: (key) => this.ready(key), - agent: (key) => this.agent(key), - member: (agent, event) => this.member(agent, event), - dm: (agent, event) => this.dm(agent, event), - starting: (agent, channelId) => - this.bindingOf(agent)?.session.channelWrites.starting(channelId), - community: (agent) => this.bindingOf(agent), - records: () => Object.values(this.records), - checkpoints: { - read: () => (savedProgress = readCheckpoints(this.storage)), - write: (checkpoints) => { - void writeCheckpoints(this.storage, checkpoints, savedProgress).catch( - (error) => { - console.warn("Agent catch-up progress was not saved", error); - // The copy went back to what is saved: the next save compares - // with that. - savedProgress = readCheckpoints(this.storage); - }, - ); - savedProgress = checkpoints; - }, - }, - interrupted: { - read: () => readInterrupted(this.storage), - write: (key, turns) => { - void writeInterrupted(this.storage, key, turns).catch((error) => - console.warn("Agent turns in progress were not saved", error), + // A claim names the protection the agent is saved with; a refused one + // may mean it changed. + const custody = native && { + ...native, + claim: async (pubkey: string) => { + try { + return await native.claim( + pubkey, + this.identities.find( + (agent) => agent.pubkey === pubkey && !agent.deleted, + )?.protectionRevision ?? 0, ); - }, - changed: () => this.update(), + } catch (error) { + void this.refresh().catch(() => undefined); + throw error; + } }, - scope: (agent) => this.bindingOf(agent)?.scope, - saved: (key) => this.records[key], + }; + this.runner = new AgentsRunner(storage, custody, localActivity, decisions, { + ownershipChanged: () => this.notify(), + changed: () => this.update(), + open: (origin) => this.communities.open(origin), + types: () => this.types(), + defaults, + resolve: (agent, local) => this.resolve(agent, local), + finished: (key) => this.retakeType(key), + navigation: ctx.navigation, attentionShown: () => this.attentionShown(), pluginsSettled: () => this.ctx.pluginStatus.settled?.() ?? true, - record: (key) => { - const member = this.liveMember(key); - if (!member) throw new Error("No such agent on this device"); - return this.recordOf(member); - }, - change: async (key, apply) => { - const record = await this.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)); - }); - 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), - start: (agent) => - this.resolve(agent, (this.records[agent.pubkey] ?? agent).config), + protection: (agent) => this.protection.bind(agent), team: (key) => { const member = this.liveMember(key); const origin = member && originOf(member.relay); @@ -1852,11 +1668,6 @@ export class AgentsService extends Service implements Agents { ? teamPrompt(this.allTeams(), this.teammates(origin), member.pubkey) : ""; }, - protection: (agent) => this.protection.bind(agent), - agentTypes: () => - this.types() - .filter((type) => !type.remote) - .map(({ key, title }) => ({ key, title })), createAgent: async (agent, input) => { const binding = this.bindingOf(agent); if (!binding || !this.liveMember(agent.key)) @@ -1933,8 +1744,6 @@ export class AgentsService extends Service implements Agents { this.notify(); return { requestId: id, saved: false }; }, - handoff: (agent, { from, ...wake }) => - this.wakeIn(agent.key, wake, { from }), templates: (agent) => { // The agents Agents shows for this community, as the create dialog // offers a template's. @@ -1953,83 +1762,30 @@ export class AgentsService extends Service implements Agents { agents: templateAgents(template, this.allTeams(), here), })); }, - tabs: (agent) => { - const session = this.bindingOf(agent)?.session; - return session && channelTabs(session, ctx.navigation); - }, classifier: () => this.classifier(), - decisions: () => this.decisions, - context: (agent, event) => this.context(agent, event), - displayName: (agent, pubkey) => this.displayName(agent, pubkey), - custody: native && { - ...native, - claim: async (pubkey) => { - try { - return await native.claim( - pubkey, - this.identities.find( - (agent) => agent.pubkey === pubkey && !agent.deleted, - )?.protectionRevision ?? 0, - ); - } catch (error) { - void this.refresh().catch(() => undefined); - throw error; - } - }, - }, - activity: localActivity, - soloChat: (agent, channel, signal) => - soloChat(() => this.relay(agent.key), agent.owner, channel, signal), - remembered: (key, slug, body) => this.coreMemory.wrote(key, slug, body), - woke: (key, object, events) => - this.history( - key, - object, - events - ?.map( - (event) => - `${new Date(event.created_at * 1000).toISOString()} · ${event.pubkey}\n${event.content}`, - ) - .join("\n\n"), - Math.floor(Date.now() / 1000), - ), - finished: (key) => { - void this.coreMemory.refresh(key); - this.retakeType(key); - }, - named: (agent, event, title) => - void nameSoloChat( - this.relay(agent.key), - agent.owner, - agent.pubkey, - event, - title, - ).catch((error) => console.warn("Couldn't name the Solo chat", error)), }); + this.plugins = new RunnerPlugins(ctx, this.runner); if (native) ctx.effect(() => - bindKeepAwake(this.runners, (hold) => native.keepAwake(hold)), + bindKeepAwake(this.runner.runners, (hold) => native.keepAwake(hold)), ); - 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(); - }; + this.unfinished = Object.freeze(readUnfinished(storage)); + this.unarchiving = readArchiveUnchecked(storage); + this.teamList = Object.freeze(readTeams(storage)); + this.syncedTeams = new SyncedTeams(storage); + this.state = Object.freeze({ + status: native ? "loading" : "unavailable", + classifier: this.availability(), + agents: [], + running: [], + interrupted: [], + teams: this.teamList, + }); + this.teamsChanged(); + // A synced copy that changes reaches running agents as a local save does. + this.runner.configStore.subscribe(() => { + this.teamsChanged(); + this.update(); }); ctx.effect(() => { const stops = [ @@ -2081,10 +1837,10 @@ export class AgentsService extends Service implements Agents { this.updateClassifier(); // Wakes each runner, which starts any timer that is due. const tick = setInterval(() => { - this.runners.tick(); + this.runner.runners.tick(); this.reconnect(); - this.announce(); - this.runners.catchUp(true); + this.presence.announce(); + this.runner.runners.catchUp(true); }, TIMER_TICK_MS); return () => { clearInterval(tick); @@ -2092,7 +1848,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); - this.runners.clear(); + this.runner.clear(); this.presence.clear(); // Clear what is shown without reconciling, which would make new runners. this.state = Object.freeze({ @@ -2199,242 +1955,18 @@ export class AgentsService extends Service implements Agents { /** Whether a plugin that shows Attention is on, so Attention may run. */ private attentionShown = () => this.attentionControls.snapshot().length > 0; 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 ( - bundledAttention.providers.some( - ({ provider }) => provider === `${plugin}/${source.id}`, - ) - ) - throw new Error( - "Attention command sources are built into Buzz. Update the Attention UI plugin.", - ); - 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.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 (handle.protection) { - if (!host.spawn) - throw new Error("Protected file reading is unavailable"); - return readFile( - host.spawn.bind(host), - options.cwd ?? "~", - path, - signal, - handle.protection.processProtection, - ); - } - 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.runners.handle(key); - if (!handle) throw new Error("This agent is not running here"); - if (handle.protection) - throw new Error( - "Protected agents cannot run external Attention commands until that execution path supports complete protection", - ); - 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); } - private attentionProvider(source: Contribution) { - return ( - bundledAttention.providers.find( - ({ plugin, id }) => source.pluginId === plugin && source.id === id, - )?.provider ?? source.key - ); - } - attentionProviders = () => - this.attentionSources.snapshot().map((source) => { - const { description, configExample } = source; - return { - provider: this.attentionProvider(source), - ...(this.attentionProvider(source) !== source.key - ? { builtIn: true } - : {}), - ...(description === undefined ? {} : { description }), - ...(configExample === undefined - ? {} - : { configExample: JSON.parse(JSON.stringify(configExample)) }), - }; - }); + attentionProviders = () => this.plugins.attentionProviders(); + tools = () => this.plugins.tools(); presentWatch = (watch: import("./attention").SourceWatch, title: string) => { - const source = this.attentionSources + // Whichever page runs the source, its UI entry presents it here. + const shown = this.sourcePresentations .snapshot() - .find((source) => this.attentionProvider(source) === watch.provider); - const shown = - source && - this.sourcePresentations - .snapshot() - .find((entry) => entry.key === source.key); + .find((entry) => attentionProvider(entry) === watch.provider); try { return shown?.present(watch, title); } catch (error) { @@ -2446,108 +1978,10 @@ export class AgentsService extends Service implements Agents { return undefined; } }; - 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), - finish: (key, object) => this.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.history(key, object, text); - this.runners.wakeSource( - key, - { - type: "source", - slug: object.slug, - watch: object.value, - plugin: source.pluginId, - text, - }, - object, - current, - ); - }, - }, - this.sourceTails, - ); - const update = () => - sources.reconcile( - this.attentionProvider(source), - source, - this.state.running.filter((agent) => this.runners.runs(agent.key)), - ); - const unsubscribe = this.subscribe(update); - update(); - return () => { - unsubscribe(); - sources.stop(); - }; - } - tools = () => this.addedTools.snapshot(); wake(key: AgentKey, wake: AgentWake) { const plugin = this.ctx.pluginOwner?.id; if (!plugin) throw new Error("Only a plugin wakes an agent"); - this.wakeIn(key, wake, { plugin }); - } - private wakeIn( - key: AgentKey, - { channel, root, text }: AgentWake, - source: Readonly<{ plugin: string } | { from: HandoffSource }>, - ) { - if (!channel || !text.trim()) - throw new Error("A wake needs a channel and text"); - // Harnesses find the thread from a well-formed id; any other would leave - // them answering at the wake's own id. - if (root !== undefined && !/^[0-9a-f]{64}$/i.test(root)) - throw new Error("A wake's root must be the thread's 64-hex event id"); - this.runners.wakeIn(key, { - type: "plugin", - ...source, - channel, - ...(root ? { root: root.toLowerCase() } : {}), - text, - id: bytesToHex(crypto.getRandomValues(new Uint8Array(32))), - at: Date.now(), - }); + this.runner.wake(key, wake, { plugin }); } snapshot = () => this.state; subscribe = (listener: () => void) => { @@ -2557,11 +1991,11 @@ export class AgentsService extends Service implements Agents { }; }; resumeInterrupted = () => { - this.runners.resumeInterrupted(this.interruptedKeys()); + this.runner.runners.resumeInterrupted(this.interruptedKeys()); this.update(); }; dismissInterrupted = () => { - this.runners.dismissInterrupted(this.interruptedKeys()); + this.runner.runners.dismissInterrupted(this.interruptedKeys()); this.update(); }; /** The agents the notice names, and only they, are resumed or forgotten. */ @@ -2572,17 +2006,15 @@ export class AgentsService extends Service implements Agents { this.state.agents.find((agent) => agent.pubkey === pubkey); handle(key: AgentKey) { const owner = this.ctx.pluginOwner; - return owner && this.runners.handle(key, owner.id); + return owner && this.runner.runners.handle(key, owner.id); } note(key: AgentKey, channelId: string, text: string, root?: string) { - return this.runners.note(key, channelId, text, root); + return this.runner.runners.note(key, channelId, text, root); } - relay = (key: AgentKey) => { - const origin = originOf(this.agent(key)?.relay ?? ""); - return origin ? this.bound.get(origin)?.relay : undefined; - }; + relay = (key: AgentKey) => this.runner.relay(key); core = (key: AgentKey, signal?: AbortSignal) => - this.coreMemory.core(key, signal); + this.runner.coreMemory.core(key, signal); + private localCache: | { identities: readonly AgentIdentity[]; @@ -2795,66 +2227,6 @@ export class AgentsService extends Service implements Agents { this.state = Object.freeze({ ...this.state, classifier }); this.notify(); } - /** The event's surroundings for the classifier, from what the app knows. */ - private async context( - agent: Agent, - event: RelayEvent, - ): Promise { - const channelId = event.tags.find((tag) => tag[0] === "h")?.[1]; - const channels = this.bindingOf(agent)?.session.channels; - const summary = channelId - ? (channels?.list().channels.find((item) => item.id === channelId) ?? - channels?.get?.(channelId)) - : undefined; - const parent = threadReference(event)?.parentId ?? null; - let found: RelayEvent | undefined; - const native = this.native; - if (parent && native) - try { - // Bounded: classified events release in order, so a stuck lookup - // would hold back every later one. - let timer: ReturnType | undefined; - [found] = await Promise.race([ - native.query(agent, [{ ids: [parent], limit: 1 }]), - new Promise((_, reject) => { - timer = setTimeout( - () => reject(new Error("The parent lookup timed out")), - PARENT_LOOKUP_MS, - ); - }), - ]).finally(() => clearTimeout(timer)); - } catch { - // The parent stays unknown; the classifier still runs. - } - return { - channel: { name: summary?.name ?? null }, - thread: { - is_reply: !!parent, - parent: found - ? { - author: { - role: - found.pubkey === agent.pubkey - ? "self" - : found.pubkey === agent.owner - ? "owner" - : "other", - name: this.displayName(agent, found.pubkey), - }, - content: excerpt(found.content), - } - : null, - }, - mentions: mentionFacts(event.tags, agent.pubkey, agent.owner), - }; - } - private displayName(agent: Agent, pubkey: string) { - return ( - this.bindingOf(agent)?.session.profiles?.snapshot().get(pubkey)?.name ?? - null - ); - } - /** Has the provider make the agent, then sets it up. From here on it is * kept, unfinished until set up, so a failure never makes a second one. */ private async provision( @@ -3215,7 +2587,10 @@ 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.runner.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); @@ -3292,30 +2667,20 @@ export class AgentsService extends Service implements Agents { return this.exclusive(key, () => this.saveHosted(key, change)); if (!this.native) throw new Error("No such agent on this device"); const { pubkey } = identity; - if (change.config !== undefined && this.runners.transitioning(key)) + if (change.config !== undefined && this.runner.runners.transitioning(key)) 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. Synced sides save to the relay first: an edit that - // isn't saved there changes nothing here and stays in the editor. - const home = key === pubkey; + // Synced sides save to the relay first: an edit that isn't saved there + // changes nothing here and stays in the editor. if (change.config !== undefined) await this.saveSynced(identity, change.config as object, change.base); - 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.watchEnabled || - own.attentionEnabled !== undefined - ) - await this.change(key, (saved) => - changed(saved ?? this.recordOf(identity), own), - ); + const { name: _, picture: __, base: ___, ...settings } = change; + await this.runner.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; @@ -3331,7 +2696,8 @@ export class AgentsService extends Service implements Agents { * names another type. */ private otherType(identity: AgentIdentity) { const settings = this.configBinding(identity).settings; - const head = settings && this.configStore.state(settings.subject).head; + const head = + settings && this.runner.configStore.state(settings.subject).head; const named = head && settingsType(head); return named && named !== identity.type ? `This agent's settings are for its ${named} type now; change its type here to edit them` @@ -3354,9 +2720,14 @@ export class AgentsService extends Service implements Agents { const subject = binding[side]?.subject; return ( subject && - isSynced(this.configStore, binding[side]) && + isSynced(this.runner.configStore, binding[side]) && applies(side) && - changes(side, agent, config, this.configStore.state(subject).head) + changes( + side, + agent, + config, + this.runner.configStore.state(subject).head, + ) ); }); if (!edited.length) return; @@ -3369,7 +2740,10 @@ export class AgentsService extends Service implements Agents { problem && edited.some((side) => { const subject = binding[side]?.subject; - return !!subject && this.configStore.state(subject).state !== "present"; + return ( + !!subject && + this.runner.configStore.state(subject).state !== "present" + ); }) ) throw new Error(problem); @@ -3443,8 +2817,8 @@ export class AgentsService extends Service implements Agents { if ( !agent || (agent.protection && type?.protection !== true) || - !this.runners.runs(key) || - this.runners.transitioning(key) + !this.runner.runners.runs(key) || + this.runner.runners.transitioning(key) ) return "unavailable"; return type?.sessionControls?.state(agent, place) ?? "unavailable"; @@ -3466,7 +2840,7 @@ export class AgentsService extends Service implements Agents { throw new Error( "Use session controls in the Buzz app that runs this agent.", ); - await this.runners.control(key, async () => { + await this.runner.runners.control(key, async () => { const current = shown(); if (current?.type !== agent.type || !this.types().includes(type)) throw new Error("The agent's harness changed. Try again."); @@ -3546,7 +2920,7 @@ export class AgentsService extends Service implements Agents { await this.applyType(identity, type, config); return preview.dropped; }; - return this.runners.retype(key, identity.name, () => + return this.runner.runners.retype(key, identity.name, () => this.typeChange(identity, retyped), ); } @@ -3591,8 +2965,8 @@ export class AgentsService extends Service implements Agents { config: object, ) { const settings = this.configBinding(identity).settings; - if (!settings || !isSynced(this.configStore, settings)) return; - const head = this.configStore.state(settings.subject).head; + if (!settings || !isSynced(this.runner.configStore, settings)) return; + const head = this.runner.configStore.state(settings.subject).head; if (head && settingsType(head) === type) return; const live = this.communityConfigs.get(this.configHome(identity)); if (!live) throw new Error("Connect to the community to change its type"); @@ -3620,9 +2994,12 @@ export class AgentsService extends Service implements Agents { const { pubkey } = identity; await native.setType(pubkey, type); try { - await this.change( + await this.runner.records.change( pubkey, - (saved) => ({ ...(saved ?? this.settingsOf(identity)), config }), + (saved) => ({ + ...(saved ?? this.runner.records.settingsOf(identity)), + config, + }), // Shown with its new type, only once its config is saved. () => { this.identities = this.identities.map((saved) => @@ -3652,14 +3029,14 @@ export class AgentsService extends Service implements Agents { private followType(identity: Member) { const settings = this.configBinding(identity).settings; if (settings?.mode !== "synced") return; - const head = this.configStore.state(settings.subject).head; + const head = this.runner.configStore.state(settings.subject).head; const type = head && settingsType(head); if ( !head || !type || type === identity.type || this.followedType.get(identity.pubkey) === head.id || - this.runners.transitioning(identity.key) || + this.runner.runners.transitioning(identity.key) || // The change running now calls `update` as it ends. this.retyping.has(identity.pubkey) ) @@ -3678,8 +3055,8 @@ export class AgentsService extends Service implements Agents { this.typeChange(identity, () => this.applyType(identity, type, adopted)); // The copy that runs it holds its runners while its type changes. void ( - this.runners.runs(identity.key) - ? this.runners.retype(identity.key, identity.name, exclusive) + this.runner.runners.runs(identity.key) + ? this.runner.runners.retype(identity.key, identity.name, exclusive) : exclusive() ).catch(failed); } @@ -3704,7 +3081,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.runner.records.settingsOf(identity).config, moved: [], dropped: [], settings: [], @@ -3897,7 +3274,10 @@ 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.runner.records.recordOf(member)), + ); } return [...shown.values()]; } @@ -3997,17 +3377,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.runner.records.edit(key, { kind: "leave" }); await this.refresh(); } @@ -4184,7 +3554,8 @@ export class AgentsService extends Service implements Agents { return this.syncedTeams.all().flatMap(({ owner, community, team }) => { if (owner !== viewer) return []; const subject = teamArtifact(owner, community, team); - const head = !team.archived && this.configStore.state(subject).head; + const head = + !team.archived && this.runner.configStore.state(subject).head; const body = head && teamBody(head.content); return body ? [ @@ -4532,113 +3903,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( @@ -4698,33 +3962,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 saved = this.settingsOf(identity); + const saved = this.runner.records.settingsOf(identity); // Its synced fields are its config's, not the local copy's. const record = { ...saved, @@ -4839,8 +4082,9 @@ export class AgentsService extends Service implements Agents { channel: "", type: SIDE_TYPE[side], }; - const channel = this.configStore.boundChannel(unbound); - return channel !== undefined || this.configStore.hadVersion(unbound) + const channel = this.runner.configStore.boundChannel(unbound); + return channel !== undefined || + this.runner.configStore.hadVersion(unbound) ? { mode: "synced" as const, subject: { ...unbound, channel: channel ?? "" }, @@ -4871,14 +4115,14 @@ export class AgentsService extends Service implements Agents { const merge = this.types().find( (type) => type.key === agent.type, )?.syncedSettings; - if (!merge && isSynced(this.configStore, settings)) + if (!merge && isSynced(this.runner.configStore, settings)) return { ok: false as const, state: "invalid" as const, reason: "This agent type has no synced settings", }; return resolveStart({ - store: this.configStore, + store: this.runner.configStore, local: local as object, agentType: agent.type, ...(settings && merge @@ -4903,7 +4147,7 @@ export class AgentsService extends Service implements Agents { private editorConfig(identity: AgentIdentity, local: object) { const binding = this.configBinding(identity); const sides = SIDES.filter((side) => - isSynced(this.configStore, binding[side]), + isSynced(this.runner.configStore, binding[side]), ); if (!sides.length) return undefined; const base: ConfigBase = Object.fromEntries( @@ -4911,7 +4155,7 @@ export class AgentsService extends Service implements Agents { const subject = binding[side]?.subject; return [ side, - (subject && this.configStore.state(subject).head?.id) ?? null, + (subject && this.runner.configStore.state(subject).head?.id) ?? null, ]; }), ); @@ -4971,12 +4215,7 @@ export class AgentsService extends Service implements Agents { } /** A running agent, in any community. */ private agent(key: AgentKey) { - return this.state.running.find((agent) => agent.key === key); - } - /** The agent, while its community's session is live to run it in. */ - private ready(key: AgentKey) { - const agent = this.agent(key); - return agent && this.bindingOf(agent) ? agent : undefined; + return this.runner.agent(key); } /** Reads the agent list from native. Every change is made there first and @@ -5001,11 +4240,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.runner.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.runner.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. @@ -5131,6 +4372,9 @@ export class AgentsService extends Service implements Agents { * finish; lets go of the rest. */ private watch() { const { viewer, selected } = this.communities.snapshot(); + // The Runner's own sessions; each change to one shows and reconciles + // what runs again. + this.runner.watch(this.identities, viewer); const origins = new Set(); const shown = viewer && originOf(selected ?? ""); if (shown) origins.add(shown); @@ -5218,14 +4462,9 @@ export class AgentsService extends Service implements Agents { this.edits.delete(origin); this.communityConfigs.get(origin)?.dispose(); this.communityConfigs.delete(origin); - const connection: Connection = { - epoch: 0, - start: Math.floor(Date.now() / 1000), - }; entry.binding = live && { ...live, - connection, - stopLive: this.listen(live.session, origin, connection), + stopLive: this.syncOnReconnect(live.session), }; } // Disconnecting stops its agents even when no binding changed. @@ -5233,7 +4472,7 @@ export class AgentsService extends Service implements Agents { // Its sync reads the community's agent records too. if (changed && entry.binding) { this.sync(); - this.refreshCore(origin); + this.runner.refreshCore(origin); void this.startConfig(origin, entry.binding); } } @@ -5247,7 +4486,7 @@ export class AgentsService extends Service implements Agents { session: binding.session, owner: binding.viewer, community: origin, - store: this.configStore, + store: this.runner.configStore, teams: this.syncedTeams, info: (signal) => nativeRelayInfo(origin, signal), changed: () => { @@ -5268,129 +4507,66 @@ export class AgentsService extends Service implements Agents { this.update(); } - /** Delivers the session's live events, and retries what failed whenever its - * connection recovers, which happens within the same session. A subscription - * that comes back, after a reconnect or a failed route, first sends its - * recent history again: what in it is new since this listener started was - * missed while it was down, and is delivered now (`seen` drops the rest). - * Older history is catch-up's, not replay's. */ - private listen( - session: RelaySession, - origin: string, - connection: Connection, - ) { + /** Syncs the relay again whenever the session's connection recovers. */ + private syncOnReconnect(session: RelaySession) { let connected = session.live.snapshot().status === "connected"; - const since = connection.start; - if (connected) Object.assign(connection, { epoch: 1, since }); - const heard = (agent: Agent) => originOf(agent.relay) === origin; - const stopEvents = session.subscribeLive( - (batch) => this.runners.dispatch(batch, heard, since), - { replay: true }, - ); - const stopStatus = session.live.subscribe(() => { + return session.live.subscribe(() => { const now = session.live.snapshot().status === "connected"; - if (now !== connected) { - // Catch-up reads what this connect may have missed, up to now. - connection.epoch += now ? 1 : 0; - connection.since = now ? Math.floor(Date.now() / 1000) : undefined; - } - if (now && !connected) { - this.sync(); - this.refreshCore(origin); - } + if (now && !connected) this.sync(); connected = now; - this.runners.catchUp(false); }); - // A roster the relay confirms decides the events held for it. - const stopRosters = session.channels.subscribeList?.(() => - this.runners.releaseHeld(heard), - ); - return () => { - stopEvents(); - stopStatus(); - stopRosters?.(); - }; - } - - /** Refreshes the cached `core` memory of each agent running in the - * community, which may have changed elsewhere while it was not connected. */ - private refreshCore(origin: string) { - for (const agent of this.state.running) - if (originOf(agent.relay) === origin) - void this.coreMemory.refresh(agent.key); } - // Joins identities, records and the connections into the running and visible - // agents, and keeps one runner per saved agent. Runners outlive edits and - // reconnects; one is retired only when its agent is deleted or its Remove - // starts, and its run is aborted when the agent's type is replaced. private update() { - const identities = this.identities.filter((identity) => !identity.deleted); + const { viewer, selected } = this.communities.snapshot(); + // The Runner decides what runs; the screens show what it decided. + this.runner.update(this.identities, viewer); const alive: Agent[] = []; - for (const identity of identities) { - if (!this.alive(identity)) continue; - const record = this.recordOf(identity); + for (const identity of this.identities) { + if (identity.deleted || !this.alive(identity)) continue; + const record = this.runner.records.recordOf(identity); // Reuse the frozen agent while nothing visible changed, so subscribers see - // a stable snapshot. - const prior = - this.agent(identity.key) ?? - this.state.agents.find( - (agent) => agent.key === identity.key && agent.deleting, - ); + // a stable snapshot: as the Runner runs it or, when that is not what is + // shown, as it was shown. const editor = this.editorConfig(identity, record.config as object); if (identity.key === identity.pubkey) this.followType(identity); + const fits = (prior: Agent | undefined) => + prior?.name === identity.name && + prior.picture === identity.picture && + prior.type === identity.type && + prior.protectionRevision === (identity.protectionRevision ?? 0) && + prior.attention === record.attention && + prior.skipped === (record.skipped ?? EMPTY) && + prior.timers === (record.timers ?? EMPTY) && + prior.sourceRuns === record.sourceRuns && + prior.attentionHistory === record.attentionHistory && + prior.attentionEnabled === + (record.attentionEnabled === true && this.attentionShown()) && + sameConfig(prior, editor?.config ?? record.config, editor?.base) && + prior.configProblem === editor?.problem && + prior.archiveUnchecked === this.archiveUnchecked(identity.key) && + prior.deleting === identity.deleting; + const before = this.state.agents.find( + (agent) => agent.key === identity.key && agent.place === "here", + ); alive.push( - prior && - prior.name === identity.name && - prior.picture === identity.picture && - prior.type === identity.type && - prior.protectionRevision === (identity.protectionRevision ?? 0) && - prior.attention === record.attention && - prior.skipped === (record.skipped ?? EMPTY) && - prior.timers === (record.timers ?? EMPTY) && - prior.sourceRuns === record.sourceRuns && - prior.attentionHistory === record.attentionHistory && - prior.attentionEnabled === - (record.attentionEnabled === true && this.attentionShown()) && - sameConfig(prior, editor?.config ?? record.config, editor?.base) && - prior.configProblem === editor?.problem && - prior.archiveUnchecked === this.archiveUnchecked(identity.key) && - prior.deleting === identity.deleting - ? prior - : this.view(identity, record), + [this.agent(identity.key), before].find(fits) ?? + this.view(identity, record), ); } - // An unfinished Delete or Remove stays listed, but nothing runs it: its - // types' plugins stop what they keep for it. - const running = alive.filter((agent) => !agent.deleting); - const selected = originOf(this.communities.snapshot().selected ?? ""); + const shown = originOf(selected ?? ""); const agents = [ ...alive.filter( - (agent) => originOf(agent.relay) === selected && this.bindingOf(agent), + (agent) => originOf(agent.relay) === shown && this.bindingOf(agent), ), - ...this.listedAgents(selected), + ...this.listedAgents(shown), ]; - this.runners.reconcile( - identities - .filter((identity) => !identity.deleting) - .map(({ key, pubkey, relay, type, owner, protectionRevision }) => ({ - key, - pubkey, - relay, - type, - protectionRevision, - scope: relayPartition(originOf(relay) ?? relay, owner), - })), - new Map(this.types().map((type) => [type.key, type])), - ); - const left = this.runners.interrupted(); - // Only the copy that holds an agent may resume or forget its turns. - const interrupted = running.filter( - (agent) => left.has(agent.key) && this.runners.runs(agent.key), - ); - const same = (next: readonly Agent[], prior: readonly Agent[]) => - next.length === prior.length && - next.every((agent, index) => agent === prior[index]); + // Each as shown: with its config as its editor has it. + const views = new Map(alive.map((agent) => [agent.key, agent])); + const asShown = (ran: readonly Agent[]) => + ran.map((agent) => views.get(agent.key) ?? agent); + const running = asShown(this.runner.running()); + const interrupted = asShown(this.runner.interrupted()); const syncsTeams = !!selected && this.communityConfigs.has(selected); if ( !same(agents, this.state.agents) || @@ -5407,13 +4583,7 @@ export class AgentsService extends Service implements Agents { }); this.notify(); } - this.runners.resume( - new Set( - running.filter((agent) => this.bindingOf(agent)).map((a) => a.key), - ), - ); - this.announce(); - this.runners.catchUp(false); + this.presence.announce(); for (const [origin, config] of this.communityConfigs) void config.attach( running @@ -5426,131 +4596,22 @@ export class AgentsService extends Service implements Agents { .map((agent) => this.configAgent( agent, - (this.records[agent.pubkey] ?? agent).config as object, + (this.runner.records.saved(agent.pubkey) ?? agent) + .config as object, ), ), ); } - /** Tells each community which agents this copy runs there: online every - * minute while it does, and offline once it stops, unless another copy took - * over. Offline is best effort; the relay lets a lapsed online expire. */ - private announce() { - const online = new Map>(); - for (const agent of this.state.running) { - const origin = originOf(agent.relay); - if (origin && this.bindingOf(agent) && this.runners.runs(agent.key)) - online.set(origin, (online.get(origin) ?? new Set()).add(agent.pubkey)); - } - for (const origin of this.presence.keys()) - if (!this.bound.has(origin)) this.presence.delete(origin); - const now = Date.now(); - for (const [origin, entry] of this.bound) { - const binding = entry.binding; - if (!binding || this.announcing.has(origin)) continue; - const next = online.get(origin) ?? new Set(); - const prior = this.presence.get(origin); - const same = prior?.session === binding.session; - if (same && prior.unable) continue; - // Another copy that took an agent over says it is online; offline from - // this one could arrive after that, so forget it was online here. - // Each pubkey said online here is one membership's, the one here. - const keyOf = (pubkey: string) => - this.identities.find( - (identity) => - identity.pubkey === pubkey && - !identity.deleted && - originOf(identity.relay) === origin, - )?.key; - const was = new Set( - [...(prior?.online ?? [])].filter((pubkey) => { - const key = keyOf(pubkey); - return next.has(pubkey) || !key || !this.runners.yielded(key); - }), - ); - if (prior && was.size < prior.online.size) - this.presence.set(origin, { ...prior, online: was }); - const gone = [...was].filter((pubkey) => !next.has(pubkey)); - const due = !same || now - prior.at >= PRESENCE_MS; - const changed = - gone.length > 0 || [...next].some((pubkey) => !was.has(pubkey)); - if (!changed && (!due || !next.size)) continue; - this.announcing.add(origin); - const sent = { session: binding.session, online: next, at: now }; - this.presence.set(origin, sent); - void (async () => { - for (const [status, agents] of [ - ["offline", gone], - ["online", [...next]], - ] as const) - for (let at = 0; at < agents.length; at += PRESENCE_BATCH) - if ( - !(await binding.session.publishAgentPresence( - status, - agents.slice(at, at + PRESENCE_BATCH), - binding.controller.signal, - )) - ) - return false; - return true; - })() - .then((ok) => { - if (ok || this.presence.get(origin) !== sent) return; - console.warn("Agent presence cannot be sent on this connection"); - this.presence.set(origin, { ...sent, unable: true }); - }) - .catch((error) => { - if (binding.controller.signal.aborted) return; - console.warn("Agent presence was not sent", error); - // Try again sooner than the minute, but not every tick. - if (this.presence.get(origin) === sent) - this.presence.set(origin, { - ...sent, - at: Date.now() - PRESENCE_MS + RETRY_MS, - }); - }) - .finally(() => this.announcing.delete(origin)); - } - } private notify() { for (const listener of this.listeners) listener(); } - /** The event's channel as the agent's community knows it, if at all. */ - private channelOf(agent: Agent, event: Pick) { - const channel = event.tags.find((tag) => tag[0] === "h")?.[1]; - const channels = this.bindingOf(agent)?.session.channels; - if (!channel || !channels) return undefined; - return ( - channels.list().channels.find((item) => item.id === channel) ?? - channels.get?.(channel) - ); - } - /** Whether the agent is in the event's channel, by its relay-signed roster. - * The stream is the owner's, so a watch, even of "all", sees only what the - * agent itself could read; an event with no channel, or whose roster is - * unknown, wakes no watch. A roster only loaded from disk decides nothing - * (undefined): the agent may have been removed or added since it was saved, - * so the event waits until the relay confirms it. */ - private member( - agent: Agent, - event: Pick, - ): boolean | undefined { - const summary = this.channelOf(agent, event); - if (summary?.cached) return undefined; - return !!summary?.members?.includes(agent.pubkey); - } - /** Whether the event's channel is a DM of exactly the agent and one other, - * by the same roster. A group DM, or one whose roster is unknown, is not. */ - private dm(agent: Agent, event: RelayEvent) { - const summary = this.channelOf(agent, event); - return ( - summary?.channelType === "dm" && - summary.members?.length === 2 && - summary.members.includes(agent.pubkey) - ); - } } +const same = (next: readonly Agent[], prior: readonly Agent[]) => + next.length === prior.length && + next.every((agent, index) => agent === prior[index]); + /** An old Buzz team's copy for one owner in one community (`scope`), the same * from every import. */ async function importedTeamId(team: OldTeam, scope: string) {