diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index d1cb650637..d968879762 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -12,6 +12,7 @@ import { 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"; @@ -21,7 +22,7 @@ import { migrateAgentStorage } from "./migrate-storage"; import type { AgentIdentity } from "./native"; import { ownerReads } from "./reads"; import { AgentRecords, type Member } from "./records"; -import { AgentRunners, type RunnerHost } from "./runner"; +import { AgentRunners, type Connection, type RunnerHost } from "./runner"; import type { Agent, ConfigBinding, RegisteredAgentType } from "./service"; import { soloChat } from "./solo-chat"; import { nameSoloChat } from "./solo-chat-name"; @@ -32,10 +33,6 @@ import type { RelaySession } from "../relay/session"; import type { Decisions } from "../decisions/service"; import { channelTabs } from "../../bundled/channels/useChannelTabState"; -/** The live session of an agent's community, as the Runner uses it. */ -export type RunnerBinding = NonNullable> & - Readonly<{ scope: string; session: RelaySession }>; - /** What the Runner asks of the Manager. */ export type RunnerManager = Pick< RunnerHost, @@ -61,11 +58,10 @@ export type RunnerManager = Pick< ownershipChanged(): void; /** A record or turn in progress changed what the screens show. */ changed(): void; - /** The live session of the agent's community, if it has one now. */ - binding( - agent: Pick, - ): RunnerBinding | undefined; - relay(key: AgentKey): RelayData | undefined; + /** The community's relay, while the viewer is in it. */ + open(origin: string): RelayData | undefined; + /** The agents running here, in every community. */ + running(): readonly Agent[]; /** The membership of this key, unless it was deleted. */ liveMember(key: AgentKey): Member | undefined; types(): readonly RegisteredAgentType[]; @@ -81,6 +77,8 @@ export class AgentsRunner { 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. */ private readonly configStore: AgentConfigStore; @@ -89,7 +87,7 @@ export class AgentsRunner { custody: RunnerHost["custody"], activity: LocalAgentActivity, decisions: Decisions | undefined, - manager: RunnerManager, + private readonly manager: RunnerManager, ) { // Each changes the copy at once and logs a failed save itself. void migrateAgentStorage(storage); @@ -110,8 +108,16 @@ export class AgentsRunner { }); 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(manager.relay(key), pubkeyOf(key)), + ownerReads(this.relay(key), pubkeyOf(key)), ); /** The catch-up progress this program last read or saved. */ let savedProgress: Checkpoints = {}; @@ -122,8 +128,10 @@ export class AgentsRunner { member: (agent, event) => manager.member(agent, event), dm: (agent, event) => manager.dm(agent, event), starting: (agent, channelId) => - manager.binding(agent)?.session.channelWrites.starting(channelId), - community: (agent) => manager.binding(agent), + this.connections + .binding(agent) + ?.session.channelWrites.starting(channelId), + community: (agent) => this.connections.binding(agent), records: () => this.records.all(), checkpoints: { read: () => (savedProgress = readCheckpoints(storage)), @@ -148,7 +156,7 @@ export class AgentsRunner { }, changed: () => manager.changed(), }, - scope: (agent) => manager.binding(agent)?.scope, + scope: (agent) => this.connections.binding(agent)?.scope, saved: (key) => this.records.saved(key), record: (key) => { const member = manager.liveMember(key); @@ -209,7 +217,7 @@ export class AgentsRunner { handoff: (agent, handoff) => manager.handoff(agent, handoff), templates: (agent) => manager.templates(agent), tabs: (agent) => { - const session = manager.binding(agent)?.session; + const session = this.connections.binding(agent)?.session; return session && channelTabs(session); }, classifier: () => manager.classifier(), @@ -224,7 +232,7 @@ export class AgentsRunner { custody, activity, soloChat: (agent, channel, signal) => - soloChat(() => manager.relay(agent.key), agent.owner, channel, signal), + soloChat(() => this.relay(agent.key), agent.owner, channel, signal), remembered: (key, slug, body) => this.coreMemory.wrote(key, slug, body), woke: (key, object) => this.records.history( @@ -236,7 +244,7 @@ export class AgentsRunner { finished: (key) => void this.coreMemory.refresh(key), named: (agent, event, title) => void nameSoloChat( - manager.relay(agent.key), + this.relay(agent.key), agent.owner, agent.pubkey, event, @@ -244,4 +252,67 @@ export class AgentsRunner { ).catch((error) => console.warn("Couldn't name the Solo chat", error)), }); } + + /** Connects each community where the viewer has an agent. */ + watch(identities: readonly AgentIdentity[], viewer: string | undefined) { + this.connections.watch(identities, viewer); + } + /** The running agent's community, while the Runner is connected to it. */ + relay(key: AgentKey) { + return this.connections.relay( + originOf(this.manager.agent(key)?.relay ?? ""), + ); + } + clear() { + this.connections.clear(); + this.runners.clear(); + } + + /** 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.manager.running()) + if (originOf(agent.relay) === origin) + void this.coreMemory.refresh(agent.key); + } } 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/service.ts b/src/features/agents/service.ts index fcef5b28a0..a9c4483117 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -11,7 +11,7 @@ import { sourceCompleted } from "./attention"; import { createLocalAgentActivity } from "./local-activity"; import type { SettingsSchema, Subject } from "./agent-config"; import type { Mode } from "./agent-start"; -import { type Connection, TIMEOUT_LIMIT_MS } from "./runner"; +import { TIMEOUT_LIMIT_MS } from "./runner"; import { ownerReads } from "./reads"; // Agents: agents that are plugins from the start. A plugin registers an agent // type (its summary line, peek view, settings tabs and `run`); each agent made @@ -45,7 +45,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, @@ -100,6 +100,7 @@ import { import { agentKey, pubkeyOf, type AgentKey } from "./membership"; import { EMPTY, type Member } from "./records"; import { AgentsRunner } from "./agents-runner"; +import { originOf } from "./connections"; import { AgentPresence } from "./presence"; import { readArchiveUnchecked, @@ -1184,7 +1185,6 @@ type Binding = { /** Aborts when this binding is replaced, so no work outlives its session. */ controller: AbortController; stopLive: () => void; - connection: Connection; }; /** A community the viewer has an agent in, and its session while live. */ type Bound = { @@ -1196,14 +1196,6 @@ type Bound = { }; /** A saved identity, which is the agent's membership in one community, with * that membership's key. */ -/** 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<{ @@ -1394,8 +1386,8 @@ export class AgentsService extends Service implements Agents { agent: (key) => this.agent(key), member: (agent, event) => this.member(agent, event), dm: (agent, event) => this.dm(agent, event), - binding: (agent) => this.bindingOf(agent), - relay: (key) => this.relay(key), + open: (origin) => this.communities.open(origin), + running: () => this.state.running, liveMember: (key) => this.liveMember(key), types: () => this.types(), defaults, @@ -1612,7 +1604,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.runner.runners.clear(); + this.runner.clear(); this.presence.clear(); // Clear what is shown without reconciling, which would make new runners. this.state = Object.freeze({ @@ -1993,10 +1985,7 @@ export class AgentsService extends Service implements Agents { note(key: AgentKey, channelId: string, text: string, root?: string) { 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.runner.coreMemory.core(key, signal); private localCache: @@ -2230,7 +2219,7 @@ export class AgentsService extends Service implements Agents { event: RelayEvent, ): Promise { const channelId = event.tags.find((tag) => tag[0] === "h")?.[1]; - const channels = this.bindingOf(agent)?.session.channels; + const channels = this.runner.connections.binding(agent)?.session.channels; const summary = channelId ? (channels?.list().channels.find((item) => item.id === channelId) ?? channels?.get?.(channelId)) @@ -2279,8 +2268,10 @@ export class AgentsService extends Service implements Agents { } private displayName(agent: Agent, pubkey: string) { return ( - this.bindingOf(agent)?.session.profiles?.snapshot().get(pubkey)?.name ?? - null + this.runner.connections + .binding(agent) + ?.session.profiles?.snapshot() + .get(pubkey)?.name ?? null ); } @@ -3599,7 +3590,7 @@ export class AgentsService extends Service implements Agents { /** 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 agent && this.runner.connections.binding(agent) ? agent : undefined; } /** Reads the agent list from native. Every change is made there first and @@ -3756,6 +3747,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); @@ -3828,14 +3822,9 @@ export class AgentsService extends Service implements Agents { entry.binding?.controller.abort(); this.listings.delete(origin); this.edits.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. @@ -3843,66 +3832,20 @@ 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); } } - /** 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.runner.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.runner.runners.catchUp(false); }); - // A roster the relay confirms decides the events held for it. - const stopRosters = session.channels.subscribeList?.(() => - this.runner.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.runner.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 alive: Agent[] = []; @@ -3979,7 +3922,9 @@ export class AgentsService extends Service implements Agents { } this.runner.runners.resume( new Set( - running.filter((agent) => this.bindingOf(agent)).map((a) => a.key), + running + .filter((agent) => this.runner.connections.binding(agent)) + .map((a) => a.key), ), ); this.presence.announce(); @@ -3992,7 +3937,7 @@ export class AgentsService extends Service implements Agents { /** 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; + const channels = this.runner.connections.binding(agent)?.session.channels; if (!channel || !channels) return undefined; return ( channels.list().channels.find((item) => item.id === channel) ??