From 5cfba2ff0c7f61395174eafe73761cb4497cf8aa Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 13 Oct 2026 11:53:20 -0400 Subject: [PATCH 1/2] refactor(claude-code): read the community through a host interface The Claude runtime read the channel, conversation, names and memory through the owner's app connection. It now asks an AgentReads port for them, which the app fills from that same connection (ownerReads), so its behavior is unchanged. A standalone host can give it reads made as the agent instead. The run part of the type (models, run and the hand-over deadline) moves to run.ts, which imports no views, and fromOwner moves next to the attestation check it uses, so the run code no longer imports the agents service. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- src/bundled/claude-code/index.ts | 14 ++-- src/bundled/claude-code/run.ts | 17 +++++ src/bundled/claude-code/runtime.test.ts | 7 +- src/bundled/claude-code/runtime.ts | 80 ++++++-------------- src/features/agents/owner-attestation.ts | 9 +++ src/features/agents/reads.ts | 94 ++++++++++++++++++++++++ src/features/agents/service.ts | 10 +-- 7 files changed, 158 insertions(+), 73 deletions(-) create mode 100644 src/bundled/claude-code/run.ts create mode 100644 src/features/agents/reads.ts diff --git a/src/bundled/claude-code/index.ts b/src/bundled/claude-code/index.ts index e10fb93d6f..0f225d8199 100644 --- a/src/bundled/claude-code/index.ts +++ b/src/bundled/claude-code/index.ts @@ -2,14 +2,14 @@ // this computer. Each conversation the agent is in is one continuing session; // Claude answers with the in-process Buzz tools. import { isTauri } from "@tauri-apps/api/core"; +import { ownerReads } from "../../features/agents/reads"; import type { PluginModule } from "../../plugins/api"; +import { claudeRun, TYPE_ID } from "./run"; import { ClaudeRuntime, type Config, config, DEFAULT_CONFIG } from "./runtime"; import { ClaudeSetup } from "./setup"; import { Create, createTabs } from "./tabs"; -export const TYPE_ID = "claude-code"; -/** A turn's run returns once Claude has it, so this bounds only the hand-over. */ -const HANDOVER_MS = 2 * 60_000; +export { TYPE_ID }; export const inject = ["agents", "host"]; export const apply: PluginModule["apply"] = (ctx) => { @@ -20,7 +20,9 @@ export const apply: PluginModule["apply"] = (ctx) => { const typeKey = `${ctx.pluginOwner?.id ?? "buzz.claude-code"}/${TYPE_ID}`; const spawn = host.spawn?.bind(host); const setup = new ClaudeSetup(spawn); - const runtime = new ClaudeRuntime(host, agents.relay); + const runtime = new ClaudeRuntime(host, (pubkey) => + ownerReads(agents.relay(pubkey), pubkey), + ); const sync = () => { const snapshot = agents.snapshot(); // Agents that stop running, the app signing out included, stop their @@ -57,8 +59,6 @@ export const apply: PluginModule["apply"] = (ctx) => { { id: "claude", title: "Claude", component: ClaudeTab }, { id: "settings", title: "Settings", component: SettingsTab }, ], - models: (agent) => runtime.models(agent.pubkey), - run: (delivery) => runtime.run(delivery), - timeoutMs: HANDOVER_MS, + ...claudeRun(runtime), }); }; diff --git a/src/bundled/claude-code/run.ts b/src/bundled/claude-code/run.ts new file mode 100644 index 0000000000..25a91397ad --- /dev/null +++ b/src/bundled/claude-code/run.ts @@ -0,0 +1,17 @@ +// The part of the Claude agent type that runs agents, without its views: the +// desktop app registers it with its tabs, and a standalone host runs it alone. +import type { AgentType } from "../../features/agents/service"; +import type { ClaudeRuntime, Config } from "./runtime"; + +export const TYPE_ID = "claude-code"; +/** A turn's run returns once Claude has it, so this bounds only the hand-over. */ +const HANDOVER_MS = 2 * 60_000; + +/** How agents of this type run on `runtime`. */ +export const claudeRun = ( + runtime: ClaudeRuntime, +): Pick, "models" | "run" | "timeoutMs"> => ({ + models: (agent) => runtime.models(agent.pubkey), + run: (delivery) => runtime.run(delivery), + timeoutMs: HANDOVER_MS, +}); diff --git a/src/bundled/claude-code/runtime.test.ts b/src/bundled/claude-code/runtime.test.ts index 5d695e939b..e1d3c3763d 100644 --- a/src/bundled/claude-code/runtime.test.ts +++ b/src/bundled/claude-code/runtime.test.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { schnorr } from "@noble/curves/secp256k1.js"; import { bytesToHex } from "nostr-tools/utils"; import { afterEach, expect, it, vi } from "vitest"; +import { ownerReads } from "../../features/agents/reads"; import type { Agent, Delivery } from "../../features/agents/service"; import type { Host } from "../../features/host/service"; import type { EventData, RelayEvent } from "../../features/relay/events"; @@ -71,7 +72,11 @@ function setup( getItem: (key: string) => data.get(key) ?? null, setItem: (key: string, value: string) => void data.set(key, value), } as Storage; - const runtime = new ClaudeRuntime(host, () => relay, storage); + const runtime = new ClaudeRuntime( + host, + (pubkey) => ownerReads(relay, pubkey), + storage, + ); const published: { kind: number; content: string; tags: string[][] }[] = []; const publish = vi.fn(async (template: (typeof published)[number]) => { published.push(template); diff --git a/src/bundled/claude-code/runtime.ts b/src/bundled/claude-code/runtime.ts index f5fc3ab460..fbfeb688d9 100644 --- a/src/bundled/claude-code/runtime.ts +++ b/src/bundled/claude-code/runtime.ts @@ -7,17 +7,17 @@ // only reports a turn that fails. import type { Context, Memory } from "../../buzz-mcp/client"; import { respond } from "../../buzz-mcp/rpc"; -import { - fromOwner, - type Agent, - type AgentHandle, - type Delivery, - type Effort, - type Models, +import { fromOwner } from "../../features/agents/owner-attestation"; +import type { AgentReads } from "../../features/agents/reads"; +import type { + Agent, + AgentHandle, + Delivery, + Effort, + Models, } from "../../features/agents/service"; import type { Host } from "../../features/host/service"; import type { EventData } from "../../features/relay/events"; -import type { RelayData } from "../../features/relay/service"; import { threadReference } from "../../features/relay/thread-reference"; import type { Settled, Spawn, ToolServer } from "./claude"; import { appClient } from "../../buzz-mcp/app-client"; @@ -65,7 +65,6 @@ export const config = (value: unknown): Config => ({ /** How long a read of the agent's core memory is reused. */ const MEMORY_TTL_MS = 5 * 60_000; -const MEMORY_TIMEOUT_MS = 5_000; const NAMES_TIMEOUT_MS = 2_000; const CHAT = [9, 40002]; @@ -111,9 +110,9 @@ export class ClaudeRuntime { ); constructor( - private readonly host: Host, - /** The owner's connection to each agent's own community. */ - private readonly relay: (pubkey: string) => RelayData | undefined, + private readonly host: Pick, + /** What each agent's turns read of its community, while it can be read. */ + private readonly reads: (pubkey: string) => AgentReads | undefined, private readonly storage: Storage = globalThis.localStorage, ) {} @@ -464,12 +463,9 @@ export class ClaudeRuntime { channelId: string, fresh = false, ) { - const snapshot = this.relay(self)?.snapshot(); - const session = snapshot?.status === "ready" ? snapshot.session : undefined; - const summary = - session?.channels.list().channels.find((item) => item.id === channelId) ?? - session?.channels.get?.(channelId); - const dm = summary?.channelType === "dm"; + const reads = this.reads(self); + const summary = await reads?.channel(channelId); + const dm = summary?.dm === true; const thread = threadReference(event); const rootId = thread?.rootId ?? event.id; const scope = entry.config.scope; @@ -477,16 +473,16 @@ export class ClaudeRuntime { dm || scope === "channel" ? channelId : `${channelId}/${rootId}`; const seen = fresh ? 0 : entry.sessions.seen(key); let earlier: readonly EventData[] = []; - if (session && (thread || dm)) + if (reads && (thread || dm)) try { - const events = await session.read( + const events = await reads.read( thread ? [ { ids: [rootId], limit: 1 }, { kinds: CHAT, "#h": [channelId], "#e": [rootId], limit: 200 }, ] : [{ kinds: CHAT, "#h": [channelId], limit: CONTEXT_LIMIT + 1 }], - { signal: AbortSignal.timeout(NAMES_TIMEOUT_MS * 2) }, + AbortSignal.timeout(NAMES_TIMEOUT_MS * 2), ); earlier = events .filter( @@ -512,9 +508,9 @@ export class ClaudeRuntime { .map((tag) => tag[1] ?? ""), ]), ].filter(Boolean); - if (session) + if (reads) await Promise.race([ - session.profiles.ensure(pubkeys, "foreground").catch(() => undefined), + reads.profiles(pubkeys).catch(() => undefined), new Promise((resolve) => setTimeout(resolve, NAMES_TIMEOUT_MS)), ]); const members = summary?.members ?? []; @@ -530,10 +526,7 @@ export class ClaudeRuntime { ...(thread ? { thread } : {}), context: unseen, total: earlier.length + 1, - name: (pubkey: string) => - session?.names.resolve(pubkey, undefined, members) ?? - session?.profiles.snapshot().get(pubkey)?.name ?? - undefined, + name: (pubkey: string) => reads?.name(pubkey, members), }; } @@ -626,8 +619,7 @@ export class ClaudeRuntime { return value; } - /** The agent's memory, read back through the owner's session: it is - * encrypted to the owner, so no agent key is needed to read it. */ + /** The agent's memory, as its community's reads give it. */ private memories(pubkey: string): Promise { // The owner's session holds few memory views at once, so calls share one. let reading = this.reading.get(pubkey); @@ -640,33 +632,9 @@ export class ClaudeRuntime { return reading; } private async readMemories(pubkey: string): Promise { - const snapshot = this.relay(pubkey)?.snapshot(); - if (snapshot?.status !== "ready") throw new Error("Buzz is not connected"); - const view = snapshot.session.agentMemories.open(pubkey); - try { - const done = new Promise((resolve) => { - const settled = () => - !["idle", "loading"].includes(view.snapshot().status) && resolve(); - view.subscribe(settled); - settled(); - }); - void view.refresh(); - await Promise.race([ - done, - new Promise((_, reject) => - setTimeout( - () => reject(new Error("Memory read timed out")), - MEMORY_TIMEOUT_MS, - ), - ), - ]); - const { status, listing } = view.snapshot(); - if (status !== "ready" || !listing) - throw new Error(`Memory unavailable (${status})`); - return listing.entries; - } finally { - view.dispose(); - } + const reads = this.reads(pubkey); + if (!reads) throw new Error("Buzz is not connected"); + return reads.memories(); } } diff --git a/src/features/agents/owner-attestation.ts b/src/features/agents/owner-attestation.ts index 31c8620717..68295f375a 100644 --- a/src/features/agents/owner-attestation.ts +++ b/src/features/agents/owner-attestation.ts @@ -84,3 +84,12 @@ function satisfied( : event.created_at > value; }); } + +/** Whether an agent answers `event`: its owner's, or one by another agent the + * owner attested (NIP-OA). Both bundled types answer nobody else. */ +export async function fromOwner( + event: Pick, + owner: string, +) { + return event.pubkey === owner || (await attestedOwner(event)) === owner; +} diff --git a/src/features/agents/reads.ts b/src/features/agents/reads.ts new file mode 100644 index 0000000000..853bdaa89d --- /dev/null +++ b/src/features/agents/reads.ts @@ -0,0 +1,94 @@ +// What an agent type reads of an agent's community beyond the agent's own +// handle: the channel a turn is in, the conversation before it, the names of +// the people in it, and the agent's memory. The desktop app reads these +// through the owner's connection to the community; a standalone host reads +// them as the agent. Both give a type the same answers, so the type runs the +// same code in either. +import type { Memory } from "../../buzz-mcp/client"; +import type { EventData, ReadFilter } from "../relay/events"; +import type { RelayData } from "../relay/service"; + +/** What a turn needs to know of its channel. */ +export type ChannelFacts = Readonly<{ + name?: string; + dm: boolean; + members: readonly string[]; +}>; + +export type AgentReads = Readonly<{ + /** The channel as the community lists it, or undefined when unknown. */ + channel(id: string): Promise; + read( + filters: readonly ReadFilter[], + signal: AbortSignal, + ): Promise; + /** Asks for these people's profiles, so `name` can name them. */ + profiles(pubkeys: readonly string[]): Promise; + /** A person's name as the channel shows it, if known yet. */ + name(pubkey: string, members: readonly string[]): string | undefined; + /** The agent's memory entries. */ + memories(): Promise; +}>; + +const MEMORY_TIMEOUT_MS = 5_000; + +/** The reads of the owner's connection to the agent's community, while it is + * ready; undefined otherwise. Memory is encrypted to the owner, so the owner's + * session reads it without the agent's key. */ +export function ownerReads( + relay: RelayData | undefined, + pubkey: string, +): AgentReads | undefined { + const snapshot = relay?.snapshot(); + if (snapshot?.status !== "ready") return undefined; + const { session } = snapshot; + return Object.freeze({ + channel: async (id: string) => { + const summary = + session.channels.list().channels.find((item) => item.id === id) ?? + session.channels.get?.(id); + return summary + ? { + ...(summary.name ? { name: summary.name } : {}), + dm: summary.channelType === "dm", + members: summary.members ?? [], + } + : undefined; + }, + read: (filters: readonly ReadFilter[], signal: AbortSignal) => + session.read(filters, { signal }), + profiles: (pubkeys: readonly string[]) => + session.profiles.ensure(pubkeys, "foreground"), + name: (person: string, members: readonly string[]) => + session.names.resolve(person, undefined, members) ?? + session.profiles.snapshot().get(person)?.name ?? + undefined, + memories: async () => { + const view = session.agentMemories.open(pubkey); + try { + const done = new Promise((resolve) => { + const settled = () => + !["idle", "loading"].includes(view.snapshot().status) && resolve(); + view.subscribe(settled); + settled(); + }); + void view.refresh(); + await Promise.race([ + done, + new Promise((_, reject) => + setTimeout( + () => reject(new Error("Memory read timed out")), + MEMORY_TIMEOUT_MS, + ), + ), + ]); + const { status, listing } = view.snapshot(); + if (status !== "ready" || !listing) + throw new Error(`Memory unavailable (${status})`); + return listing.entries; + } finally { + view.dispose(); + } + }, + }); +} diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 4ae17696da..7ff6b2d2d2 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -22,7 +22,6 @@ import { type Contribution, } from "../../plugins/contributions"; import type { LocalAgentsSnapshot } from "./local"; -import { attestedOwner } from "./owner-attestation"; import { deleteAgentRecord, publishAgentRecord, @@ -469,14 +468,7 @@ export function teamPrompt( }) .join("\n\n"); } -/** Whether an agent answers `event`: its owner's, or one by another agent the - * owner attested (NIP-OA). Both bundled types answer nobody else. */ -export async function fromOwner( - event: Pick, - owner: string, -) { - return event.pubkey === owner || (await attestedOwner(event)) === owner; -} +export { fromOwner } from "./owner-attestation"; /** What Agents needs of the communities: the viewer, the selected one, and * a connection to any joined one without selecting it. */ From d6928af8de519c2615d81af79f5ae6f545cc39e2 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 13 Oct 2026 12:01:45 -0400 Subject: [PATCH 2/2] refactor(codex): read the community through the host interface The Codex runtime read the channel, conversation and memory through the owner's app connection, like Claude did. It now uses the same AgentReads port, filled by ownerReads in the app, so a standalone host can run it with reads made as the agent. The port's memory read takes a signal; each runtime keeps its own bound (Claude five seconds, Codex its turn). The run part of the type moves to run.ts, as Claude's did. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- src/bundled/claude-code/runtime.ts | 12 +++++- src/bundled/codex/index.ts | 14 ++++--- src/bundled/codex/run.ts | 18 +++++++++ src/bundled/codex/runtime.test.ts | 7 +++- src/bundled/codex/runtime.ts | 59 +++++++++++------------------- src/features/agents/reads.ts | 40 ++++++++++---------- 6 files changed, 84 insertions(+), 66 deletions(-) create mode 100644 src/bundled/codex/run.ts diff --git a/src/bundled/claude-code/runtime.ts b/src/bundled/claude-code/runtime.ts index fbfeb688d9..5e68318a8e 100644 --- a/src/bundled/claude-code/runtime.ts +++ b/src/bundled/claude-code/runtime.ts @@ -65,6 +65,7 @@ export const config = (value: unknown): Config => ({ /** How long a read of the agent's core memory is reused. */ const MEMORY_TTL_MS = 5 * 60_000; +const MEMORY_TIMEOUT_MS = 5_000; const NAMES_TIMEOUT_MS = 2_000; const CHAT = [9, 40002]; @@ -634,7 +635,16 @@ export class ClaudeRuntime { private async readMemories(pubkey: string): Promise { const reads = this.reads(pubkey); if (!reads) throw new Error("Buzz is not connected"); - return reads.memories(); + const stop = new AbortController(); + const timer = setTimeout( + () => stop.abort(new Error("Memory read timed out")), + MEMORY_TIMEOUT_MS, + ); + try { + return await reads.memories(stop.signal); + } finally { + clearTimeout(timer); + } } } diff --git a/src/bundled/codex/index.ts b/src/bundled/codex/index.ts index 09041a4954..ada8bfef9f 100644 --- a/src/bundled/codex/index.ts +++ b/src/bundled/codex/index.ts @@ -1,6 +1,8 @@ import { isTauri } from "@tauri-apps/api/core"; +import { ownerReads } from "../../features/agents/reads"; import type { PluginModule } from "../../plugins/api"; import { config, defaults, validWorkspace } from "./config"; +import { codexRun, TYPE_ID } from "./run"; import { CodexRuntime } from "./runtime"; import { createTabs } from "./tabs"; export const inject = ["agents", "host"]; @@ -14,8 +16,10 @@ export const apply: PluginModule["apply"] = (ctx) => { (async () => { throw new Error("Codex agents run only in the desktop app"); }); - const runtime = new CodexRuntime(start, ctx.agents.relay); - const typeKey = `${ctx.pluginOwner?.id ?? "buzz.codex"}/codex`; + const runtime = new CodexRuntime(start, (pubkey) => + ownerReads(ctx.agents.relay(pubkey), pubkey), + ); + const typeKey = `${ctx.pluginOwner?.id ?? "buzz.codex"}/${TYPE_ID}`; // Agents that stop running stop their turns, and the last one stops the // server; switching communities or a reconnect does not. const sync = () => { @@ -31,7 +35,7 @@ export const apply: PluginModule["apply"] = (ctx) => { }); const { CodexTab, SettingsTab, Create } = createTabs(start, runtime); ctx.agents.register({ - id: "codex", + id: TYPE_ID, title: "Codex", description: "Codex on this computer, using its native coding tools and your account.", @@ -51,8 +55,6 @@ export const apply: PluginModule["apply"] = (ctx) => { { id: "codex", title: "Codex", component: CodexTab }, { id: "settings", title: "Settings", component: SettingsTab }, ], - models: (agent) => runtime.preview(agent), - run: (delivery) => runtime.run(delivery), - timeoutMs: 2 * 60_000, + ...codexRun(runtime), }); }; diff --git a/src/bundled/codex/run.ts b/src/bundled/codex/run.ts new file mode 100644 index 0000000000..a7a79e6224 --- /dev/null +++ b/src/bundled/codex/run.ts @@ -0,0 +1,18 @@ +// The part of the Codex agent type that runs agents, without its views: the +// desktop app registers it with its tabs, and a standalone host runs it alone. +import type { AgentType } from "../../features/agents/service"; +import type { Config } from "./config"; +import type { CodexRuntime } from "./runtime"; + +export const TYPE_ID = "codex"; +/** A turn's run returns once Codex has it, so this bounds only the hand-over. */ +const HANDOVER_MS = 2 * 60_000; + +/** How agents of this type run on `runtime`. */ +export const codexRun = ( + runtime: CodexRuntime, +): Pick, "models" | "run" | "timeoutMs"> => ({ + models: (agent) => runtime.preview(agent), + run: (delivery) => runtime.run(delivery), + timeoutMs: HANDOVER_MS, +}); diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index 85bc21a118..817ec5ee6f 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -3,6 +3,7 @@ import { schnorr } from "@noble/curves/secp256k1.js"; import { bytesToHex } from "nostr-tools/utils"; import { apply } from "./index"; import { describe, afterEach, expect, it, vi } from "vitest"; +import { ownerReads } from "../../features/agents/reads"; import type { Agent, Delivery } from "../../features/agents/service"; import type { RelayData } from "../../features/relay/service"; import type { EventData } from "../../features/relay/events"; @@ -203,7 +204,11 @@ function fixture( storage.set(key, value); }, } as Storage; - const runtime = new CodexRuntime(spawn, () => relay, store); + const runtime = new CodexRuntime( + spawn, + (pubkey) => ownerReads(relay, pubkey), + store, + ); runtimes.push(runtime); const events: EventData[] = []; const query = vi.fn(async (filters: readonly Record[]) => diff --git a/src/bundled/codex/runtime.ts b/src/bundled/codex/runtime.ts index d0cde7f3d2..1ac0ae104e 100644 --- a/src/bundled/codex/runtime.ts +++ b/src/bundled/codex/runtime.ts @@ -1,12 +1,12 @@ import type { WorkOutcome } from "../../features/agents/local-activity"; -import { - fromOwner, - type Agent, - type AgentHandle, - type Delivery, - type Models, +import { fromOwner } from "../../features/agents/owner-attestation"; +import type { AgentReads } from "../../features/agents/reads"; +import type { + Agent, + AgentHandle, + Delivery, + Models, } from "../../features/agents/service"; -import type { RelayData } from "../../features/relay/service"; import type { EventData } from "../../features/relay/events"; import { appClient } from "../../buzz-mcp/app-client"; import { TOOLS, callTool } from "../../buzz-mcp/tools"; @@ -118,8 +118,8 @@ export class CodexRuntime { private previews = new Map }>(); constructor( private readonly spawn: Spawn, - /** The owner's connection to each agent's own community. */ - private readonly relay: (pubkey: string) => RelayData | undefined, + /** What each agent's turns read of its community, while it can be read. */ + private readonly reads: (pubkey: string) => AgentReads | undefined, private readonly storage: Storage = localStorage, ) {} subscribe = (listener: () => void) => { @@ -345,17 +345,13 @@ export class CodexRuntime { delivery.ignored?.(); return; } - const snapshot = this.relay(agent.pubkey)?.snapshot(); - if (snapshot?.status !== "ready") - throw new Error("Connect to the agent's community first"); + const reads = this.reads(agent.pubkey); + if (!reads) throw new Error("Connect to the agent's community first"); const channelId = delivery.channelId; if (!channelId) throw new Error("The event has no unambiguous channel"); const settings = config(delivery.config); - const channel = snapshot.session.channels - .list() - .channels.find((row) => row.id === channelId); - const toolRoot = - channel?.channelType === "dm" ? undefined : rootOf(trigger.event); + const channel = await reads.channel(channelId); + const toolRoot = channel?.dm ? undefined : rootOf(trigger.event); const root = settings.scope === "channel" ? undefined : toolRoot; let finishActivity!: (result: WorkOutcome) => void; const replied = new Promise((resolve) => { @@ -368,7 +364,7 @@ export class CodexRuntime { // Codex makes one more model call after a final send before its turn // ends; the agent is done working on this request once it has replied. const request: Request = { - scope: snapshot.scope ?? "", + scope: reads.scope, event: trigger.event, agent, settings, @@ -516,11 +512,11 @@ export class CodexRuntime { } private async history(request: Request, signal: AbortSignal) { signal.throwIfAborted(); - const snapshot = this.relay(request.agent.pubkey)?.snapshot(); - if (snapshot?.status !== "ready" || snapshot.scope !== request.scope) + const reads = this.reads(request.agent.pubkey); + if (reads?.scope !== request.scope) throw new Error("The community changed"); return conversationHistory(request.event, request.conversation, (filters) => - snapshot.session.read(filters, { signal }), + reads.read(filters, signal), ); } // Tags the owner even when a sibling asked: tagging the sibling would wake @@ -572,23 +568,12 @@ export class CodexRuntime { check, signal: active.signal, memories: async () => { - const snapshot = this.relay(request.agent.pubkey)?.snapshot(); - if (snapshot?.status !== "ready" || snapshot.scope !== request.scope) + const reads = this.reads(request.agent.pubkey); + if (reads?.scope !== request.scope) throw new Error("Buzz is not connected to this community"); - const view = snapshot.session.agentMemories.open(request.agent.pubkey); - const abort = () => view.dispose(); - active.signal.addEventListener("abort", abort, { once: true }); - try { - await view.refresh(); - check(); - const { status, listing } = view.snapshot(); - if (status !== "ready" || !listing) - throw new Error(`Memory unavailable (${status})`); - return listing.entries; - } finally { - active.signal.removeEventListener("abort", abort); - view.dispose(); - } + const entries = await reads.memories(active.signal); + check(); + return entries; }, }); const reply = await callTool( diff --git a/src/features/agents/reads.ts b/src/features/agents/reads.ts index 853bdaa89d..82ba0d710a 100644 --- a/src/features/agents/reads.ts +++ b/src/features/agents/reads.ts @@ -16,6 +16,8 @@ export type ChannelFacts = Readonly<{ }>; export type AgentReads = Readonly<{ + /** The community these reads are of; it changes when the agent's does. */ + scope: string; /** The channel as the community lists it, or undefined when unknown. */ channel(id: string): Promise; read( @@ -26,12 +28,10 @@ export type AgentReads = Readonly<{ profiles(pubkeys: readonly string[]): Promise; /** A person's name as the channel shows it, if known yet. */ name(pubkey: string, members: readonly string[]): string | undefined; - /** The agent's memory entries. */ - memories(): Promise; + /** The agent's memory entries; `signal` stops the read. */ + memories(signal?: AbortSignal): Promise; }>; -const MEMORY_TIMEOUT_MS = 5_000; - /** The reads of the owner's connection to the agent's community, while it is * ready; undefined otherwise. Memory is encrypted to the owner, so the owner's * session reads it without the agent's key. */ @@ -43,6 +43,7 @@ export function ownerReads( if (snapshot?.status !== "ready") return undefined; const { session } = snapshot; return Object.freeze({ + scope: snapshot.scope ?? "", channel: async (id: string) => { const summary = session.channels.list().channels.find((item) => item.id === id) ?? @@ -63,30 +64,27 @@ export function ownerReads( session.names.resolve(person, undefined, members) ?? session.profiles.snapshot().get(person)?.name ?? undefined, - memories: async () => { + memories: async (signal?: AbortSignal) => { + signal?.throwIfAborted(); const view = session.agentMemories.open(pubkey); + // Stopping the read disposes the view, which ends its refresh. + let stop!: (reason: unknown) => void; + const stopped = new Promise((_, reject) => { + stop = reject; + }); + const abort = () => { + stop(signal?.reason); + view.dispose(); + }; + signal?.addEventListener("abort", abort, { once: true }); try { - const done = new Promise((resolve) => { - const settled = () => - !["idle", "loading"].includes(view.snapshot().status) && resolve(); - view.subscribe(settled); - settled(); - }); - void view.refresh(); - await Promise.race([ - done, - new Promise((_, reject) => - setTimeout( - () => reject(new Error("Memory read timed out")), - MEMORY_TIMEOUT_MS, - ), - ), - ]); + await Promise.race([view.refresh(), stopped]); const { status, listing } = view.snapshot(); if (status !== "ready" || !listing) throw new Error(`Memory unavailable (${status})`); return listing.entries; } finally { + signal?.removeEventListener("abort", abort); view.dispose(); } },