From ffb2974a585e2d67e8dd59568391e8705a261733 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 15:27:45 -0400 Subject: [PATCH 1/2] Give the Runner its own community connections The Runner now keeps its own live session of each community it runs agents in, and its own listener that delivers those events to the agents. The Manager keeps its connections for listings and cleanup, and syncs again on reconnect itself. RunnerManager no longer hands the Runner live sessions: it gives only the community relay to open and the running agents. The Runner binds first on every change, so the Manager's update sees its new session, as before. No behavior changes. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.ts | 105 +++++++++++++++---- src/features/agents/connections.ts | 146 +++++++++++++++++++++++++++ src/features/agents/service.ts | 107 +++++--------------- 3 files changed, 259 insertions(+), 99 deletions(-) create mode 100644 src/features/agents/connections.ts diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index d1cb650637..6adc567976 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,14 @@ 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), + bound: (origin) => this.refreshCore(origin), + }); 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 +126,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 +154,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 +215,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 +230,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 +242,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 +250,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.ts b/src/features/agents/connections.ts new file mode 100644 index 0000000000..cff785a1eb --- /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 or was replaced. */ + bound(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), + }; + if (entry.binding) this.host.bound(origin); + } +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index fcef5b28a0..ec69967586 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 first, so each change binds its session before this one + // shows and reconciles what runs. + 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) ?? From a5b59183dd20e10ca6c66f933bfe253c278e78e7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 15:32:55 -0400 Subject: [PATCH 2/2] Reconcile again when the Runner's session changes The Manager's relay subscription can be older than the Runner's: a community selected before its first agent, or one whose agents were all deleted for a time. Its update then ran before the Runner's session changed, and nothing reconciled after it. Queued work waited for the next tick. The Runner now asks for an update each time its session starts, is replaced or ends, and no longer refreshes core memory a second time. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.ts | 4 +- src/features/agents/connections.test.ts | 71 +++++++++++++++++++++++++ src/features/agents/connections.ts | 6 +-- src/features/agents/service.ts | 4 +- 4 files changed, 79 insertions(+), 6 deletions(-) create mode 100644 src/features/agents/connections.test.ts diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index 6adc567976..d968879762 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -112,7 +112,9 @@ export class AgentsRunner { open: (origin) => manager.open(origin), listen: (session, origin, connection) => this.listen(session, origin, connection), - bound: (origin) => this.refreshCore(origin), + // 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)), 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 index cff785a1eb..590ec501b0 100644 --- a/src/features/agents/connections.ts +++ b/src/features/agents/connections.ts @@ -43,8 +43,8 @@ export type ConnectionsHost = Readonly<{ origin: string, connection: Connection, ): () => void; - /** The community's live session started or was replaced. */ - bound(origin: string): void; + /** The community's live session started, was replaced or ended. */ + changed(origin: string): void; }>; export class RunnerConnections { @@ -141,6 +141,6 @@ export class RunnerConnections { connection, stopLive: this.host.listen(live.session, origin, connection), }; - if (entry.binding) this.host.bound(origin); + this.host.changed(origin); } } diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index ec69967586..a9c4483117 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -3747,8 +3747,8 @@ export class AgentsService extends Service implements Agents { * finish; lets go of the rest. */ private watch() { const { viewer, selected } = this.communities.snapshot(); - // The Runner's first, so each change binds its session before this one - // shows and reconciles what runs. + // 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 ?? "");