From 87a2b6dc05b3587cad26562e712cfabc784f3822 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 10:19:29 -0400 Subject: [PATCH 1/4] Let the Runner decide which agents run AgentsRunner.update now picks the running agents from the identities, its records and its own community connections, reconciles the runners, finds the interrupted turns and resumes and catches up. The Manager shows what it decided and no longer answers the Runner's agent, ready or running questions. Behavior is unchanged. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.ts | 145 +++++++++++++++++++++++---- src/features/agents/service.ts | 110 +++++++------------- 2 files changed, 167 insertions(+), 88 deletions(-) diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index d968879762..43fb72c8da 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -1,9 +1,9 @@ -// The Runner: the part of agents that runs them. It 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. +// 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 { AgentConfigStore } from "./agent-config"; import { isSynced, resolveStart } from "./agent-start"; import { @@ -21,13 +21,14 @@ import { migrateWatchFilters } from "./migrate-filter"; import { migrateAgentStorage } from "./migrate-storage"; import type { AgentIdentity } from "./native"; import { ownerReads } from "./reads"; -import { AgentRecords, type Member } from "./records"; +import { AgentRecords, EMPTY, type Member } from "./records"; 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"; import type { AgentStore } from "./state"; import type { AgentRecord } from "./store"; +import { relayPartition } from "../relay/partition"; import type { RelayData } from "../relay/service"; import type { RelaySession } from "../relay/session"; import type { Decisions } from "../decisions/service"; @@ -36,8 +37,6 @@ import { channelTabs } from "../../bundled/channels/useChannelTabState"; /** What the Runner asks of the Manager. */ export type RunnerManager = Pick< RunnerHost, - | "ready" - | "agent" | "member" | "dm" | "team" @@ -60,8 +59,6 @@ export type RunnerManager = Pick< changed(): void; /** 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 +78,12 @@ export class AgentsRunner { readonly connections: RunnerConnections; /** Confirmed synced config copies, the only synced launch source. */ private 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: [] }; constructor( storage: AgentStore, @@ -123,8 +126,8 @@ export class AgentsRunner { let savedProgress: Checkpoints = {}; this.runners = new AgentRunners({ ownershipChanged: () => manager.ownershipChanged(), - ready: (key) => manager.ready(key), - agent: (key) => manager.agent(key), + ready: (key) => this.ready(key), + agent: (key) => this.agent(key), member: (agent, event) => manager.member(agent, event), dm: (agent, event) => manager.dm(agent, event), starting: (agent, channelId) => @@ -257,15 +260,99 @@ export class AgentsRunner { 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) { + const live = identities.filter( + (identity) => !identity.deleted && !identity.deleting, + ); + const running: Agent[] = []; + 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.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) && + prior.config === record.config + ? prior + : view(identity, record), + ); + } + this.runners.reconcile( + live.map(({ key, pubkey, relay, type, owner }) => ({ + key, + pubkey, + relay, + type, + 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), + ); + this.state = { + running: same(running, this.state.running) + ? this.state.running + : Object.freeze(running), + interrupted: same(interrupted, this.state.interrupted) + ? this.state.interrupted + : Object.freeze(interrupted), + }; + 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 running agent's community, while the Runner is connected to it. */ relay(key: AgentKey) { - return this.connections.relay( - originOf(this.manager.agent(key)?.relay ?? ""), - ); + 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 @@ -311,8 +398,32 @@ export class AgentsRunner { /** 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()) + for (const agent of this.state.running) if (originOf(agent.relay) === origin) void this.coreMemory.refresh(agent.key); } } + +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. */ +function view(identity: Member, record: AgentRecord): 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", + attention: record.attention, + skipped: record.skipped ?? EMPTY, + timers: record.timers ?? EMPTY, + ...(record.sourceRuns ? { sourceRuns: record.sourceRuns } : {}), + attentionHistory: record.attentionHistory, + attentionEnabled: record.attentionEnabled === true, + config: record.config, + }); +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index a9c4483117..9308a5f826 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1382,12 +1382,9 @@ export class AgentsService extends Service implements Agents { this.runner = new AgentsRunner(storage, native, localActivity, decisions, { ownershipChanged: () => this.notify(), changed: () => this.update(), - ready: (key) => this.ready(key), - agent: (key) => this.agent(key), member: (agent, event) => this.member(agent, event), dm: (agent, event) => this.dm(agent, event), open: (origin) => this.communities.open(origin), - running: () => this.state.running, liveMember: (key) => this.liveMember(key), types: () => this.types(), defaults, @@ -3585,12 +3582,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.runner.connections.binding(agent) ? agent : undefined; + return this.runner.agent(key); } /** Reads the agent list from native. Every change is made there first and @@ -3847,88 +3839,60 @@ export class AgentsService extends Service implements Agents { } 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; + 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 fits = (prior: Agent | undefined) => + prior?.name === identity.name && + prior.picture === identity.picture && + prior.type === identity.type && + 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) && + prior.config === record.config && + 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.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) && - prior.config === record.config && - 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.runner.runners.reconcile( - identities - .filter((identity) => !identity.deleting) - .map(({ key, pubkey, relay, type, owner }) => ({ - key, - pubkey, - relay, - type, - scope: relayPartition(originOf(relay) ?? relay, owner), - })), - new Map(this.types().map((type) => [type.key, type])), - ); - const left = this.runner.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.runner.runners.runs(agent.key), - ); - const same = (next: readonly Agent[], prior: readonly Agent[]) => - next.length === prior.length && - next.every((agent, index) => agent === prior[index]); + const running = this.runner.running(); + const interrupted = this.runner.interrupted(); if ( !same(agents, this.state.agents) || - !same(running, this.state.running) || - !same(interrupted, this.state.interrupted) + running !== this.state.running || + interrupted !== this.state.interrupted ) { this.state = Object.freeze({ ...this.state, agents: Object.freeze(agents), - running: Object.freeze(running), - interrupted: Object.freeze(interrupted), + running, + interrupted, }); this.notify(); } - this.runner.runners.resume( - new Set( - running - .filter((agent) => this.runner.connections.binding(agent)) - .map((a) => a.key), - ), - ); this.presence.announce(); - this.runner.runners.catchUp(false); } private notify() { @@ -3970,6 +3934,10 @@ export class AgentsService extends Service implements Agents { } } +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) { From 09a0d6a068def05c0a5f3f4a59a47ebeb2fe98f7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 10:23:09 -0400 Subject: [PATCH 2/4] Let the Runner read its own event context Whether an agent is in an event's channel, whether the channel is its DM, a sender's display name and the classifier's context read only the Runner's connections and agent custody, so they move into AgentsRunner. RunnerManager loses member, dm, context and displayName. Behavior is unchanged. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.ts | 112 ++++++++++++++++++++++++--- src/features/agents/service.ts | 106 ------------------------- 2 files changed, 103 insertions(+), 115 deletions(-) diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index 43fb72c8da..bd25ae4896 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -5,6 +5,7 @@ // `RunnerManager`; those facts and requests are what will cross the agents // link when the Runner moves to its own page. import { AgentConfigStore } from "./agent-config"; +import { excerpt, mentionFacts, type ClassifierContext } from "./classifier"; import { isSynced, resolveStart } from "./agent-start"; import { readCheckpoints, @@ -29,7 +30,9 @@ 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"; @@ -37,8 +40,6 @@ import { channelTabs } from "../../bundled/channels/useChannelTabState"; /** What the Runner asks of the Manager. */ export type RunnerManager = Pick< RunnerHost, - | "member" - | "dm" | "team" | "createAgent" | "updateAgent" @@ -49,8 +50,6 @@ export type RunnerManager = Pick< | "classifierKey" | "classifierKeyGeneration" | "setClassifierKey" - | "context" - | "displayName" > & Readonly<{ /** Who runs which agent changed. */ @@ -87,7 +86,7 @@ export class AgentsRunner { constructor( storage: AgentStore, - custody: RunnerHost["custody"], + private readonly custody: RunnerHost["custody"], activity: LocalAgentActivity, decisions: Decisions | undefined, private readonly manager: RunnerManager, @@ -128,8 +127,8 @@ export class AgentsRunner { ownershipChanged: () => manager.ownershipChanged(), ready: (key) => this.ready(key), agent: (key) => this.agent(key), - member: (agent, event) => manager.member(agent, event), - dm: (agent, event) => manager.dm(agent, event), + member: (agent, event) => this.member(agent, event), + dm: (agent, event) => this.dm(agent, event), starting: (agent, channelId) => this.connections .binding(agent) @@ -230,8 +229,8 @@ export class AgentsRunner { classifierKeyGeneration: () => manager.classifierKeyGeneration(), setClassifierKey: (key, generation) => manager.setClassifierKey(key, generation), - context: (agent, event) => manager.context(agent, event), - displayName: (agent, pubkey) => manager.displayName(agent, pubkey), + context: (agent, event) => this.context(agent, event), + displayName: (agent, pubkey) => this.displayName(agent, pubkey), custody, activity, soloChat: (agent, channel, signal) => @@ -402,8 +401,103 @@ export class AgentsRunner { 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]); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 9308a5f826..a999ac957f 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -55,7 +55,6 @@ 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 type { channelTabs } from "../../bundled/channels/useChannelTabState"; import type { RelayData } from "../relay/service"; import type { RelaySession } from "../relay/session"; @@ -76,10 +75,7 @@ import type { WriteOptions, } from "./attention-objects"; import { - excerpt, - mentionFacts, nativeClassifier, - type ClassifierContext, type ClassifierNative, type ClassifierOutcome, } from "./classifier"; @@ -1129,8 +1125,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 @@ -1382,8 +1376,6 @@ export class AgentsService extends Service implements Agents { this.runner = new AgentsRunner(storage, native, localActivity, decisions, { ownershipChanged: () => this.notify(), changed: () => this.update(), - member: (agent, event) => this.member(agent, event), - dm: (agent, event) => this.dm(agent, event), open: (origin) => this.communities.open(origin), liveMember: (key) => this.liveMember(key), types: () => this.types(), @@ -1507,8 +1499,6 @@ export class AgentsService extends Service implements Agents { setClassifierKey: (key, generation) => { if (generation === this.keyGeneration) this.setKey(key); }, - context: (agent, event) => this.context(agent, event), - displayName: (agent, pubkey) => this.displayName(agent, pubkey), }); this.unfinished = Object.freeze(readUnfinished(storage)); this.unarchiving = readArchiveUnchecked(storage); @@ -2210,68 +2200,6 @@ export class AgentsService extends Service implements Agents { this.state = Object.freeze({ ...this.state, classifier, classifierKey }); 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.runner.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.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.runner.connections - .binding(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( @@ -3898,40 +3826,6 @@ export class AgentsService extends Service implements Agents { 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.runner.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) - ); - } } const same = (next: readonly Agent[], prior: readonly Agent[]) => From 6441dad7acdd373d75e05ee339e965fc9e3be2c7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 10:25:28 -0400 Subject: [PATCH 3/4] Let the Runner find memberships and queue wakes itself The Runner keeps the memberships it last decided from, so its record changes find the membership without asking the Manager. Plugin wakes and an agent's own handoff are queued by the Runner too. RunnerManager loses liveMember and handoff. Behavior is unchanged. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/agents-runner.ts | 52 ++++++++++++++++++++++++---- src/features/agents/service.ts | 26 +------------- 2 files changed, 46 insertions(+), 32 deletions(-) diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index bd25ae4896..d5e0a2c1e0 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -4,6 +4,7 @@ // 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 { isSynced, resolveStart } from "./agent-start"; @@ -24,7 +25,13 @@ 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, ConfigBinding, RegisteredAgentType } from "./service"; +import type { + Agent, + AgentWake, + ConfigBinding, + HandoffSource, + RegisteredAgentType, +} from "./service"; import { soloChat } from "./solo-chat"; import { nameSoloChat } from "./solo-chat-name"; import type { AgentStore } from "./state"; @@ -43,7 +50,6 @@ export type RunnerManager = Pick< | "team" | "createAgent" | "updateAgent" - | "handoff" | "templates" | "classifier" | "keyStatus" @@ -58,8 +64,6 @@ export type RunnerManager = Pick< changed(): void; /** The community's relay, while the viewer is in it. */ open(origin: string): RelayData | undefined; - /** The membership of this key, unless it was deleted. */ - liveMember(key: AgentKey): Member | undefined; types(): readonly RegisteredAgentType[]; /** The settings a new agent of this type starts with. */ defaults(pubkey: string, type: RegisteredAgentType): AgentRecord; @@ -83,6 +87,8 @@ export class AgentsRunner { running: readonly Agent[]; interrupted: readonly Agent[]; }> = { running: [], interrupted: [] }; + /** The memberships it last decided from. */ + private identities: readonly Member[] = []; constructor( storage: AgentStore, @@ -161,13 +167,13 @@ export class AgentsRunner { scope: (agent) => this.connections.binding(agent)?.scope, saved: (key) => this.records.saved(key), record: (key) => { - const member = manager.liveMember(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 = manager.liveMember(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)); @@ -216,7 +222,8 @@ export class AgentsRunner { .map(({ key, title }) => ({ key, title })), createAgent: (agent, input) => manager.createAgent(agent, input), updateAgent: (agent, input) => manager.updateAgent(agent, input), - handoff: (agent, handoff) => manager.handoff(agent, handoff), + handoff: (agent, { from, ...wake }) => + this.wake(agent.key, wake, { from }), templates: (agent) => manager.templates(agent), tabs: (agent) => { const session = this.connections.binding(agent)?.session; @@ -264,6 +271,7 @@ export class AgentsRunner { * 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, ); @@ -343,6 +351,36 @@ export class AgentsRunner { 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 ?? "")); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index a999ac957f..ae6c9c4b8f 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1377,7 +1377,6 @@ export class AgentsService extends Service implements Agents { ownershipChanged: () => this.notify(), changed: () => this.update(), open: (origin) => this.communities.open(origin), - liveMember: (key) => this.liveMember(key), types: () => this.types(), defaults, configBinding: this.configBinding, @@ -1472,8 +1471,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. @@ -1921,28 +1918,7 @@ export class AgentsService extends Service implements Agents { 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.runner.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) => { From 1184a1947586d4f4ab855b7f8ac53a0bb3e253fc Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 28 Oct 2026 11:53:32 -0400 Subject: [PATCH 4/4] Start restart catch-up tests at their fixed second Three catch-up tests captured the second, then set up the first copy, then froze the clock at that second. When setup crossed into the next second, the copy's connection started a second late, so catch-up read the second of the unheard message and the restarted copy skipped it. Freeze the clock before setup instead. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/features/agents/service.test.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 4eafa41597..2a21e23b98 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -5177,9 +5177,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, @@ -5749,9 +5751,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 }, @@ -5833,9 +5837,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 },