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..5e68318a8e 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"; @@ -111,9 +111,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 +464,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 +474,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 +509,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 +527,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 +620,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,32 +633,17 @@ 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); + const reads = this.reads(pubkey); + if (!reads) throw new Error("Buzz is not connected"); + const stop = new AbortController(); + const timer = setTimeout( + () => stop.abort(new Error("Memory read timed out")), + MEMORY_TIMEOUT_MS, + ); 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; + return await reads.memories(stop.signal); } finally { - view.dispose(); + 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/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..82ba0d710a --- /dev/null +++ b/src/features/agents/reads.ts @@ -0,0 +1,92 @@ +// 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 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( + 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; `signal` stops the read. */ + memories(signal?: AbortSignal): Promise; +}>; + +/** 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({ + scope: snapshot.scope ?? "", + 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 (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 { + 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(); + } + }, + }); +} 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. */