From 82732426f6004d5522625cae37418ead00d4537c Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Thu, 8 Oct 2026 21:56:55 -0700 Subject: [PATCH 1/2] feat(agents2): show agents as typing while they work on a mention The Agents2 host now sends a typing pulse (kind 20002) when a mention-started run begins and every 3 s after, in the mention's channel and threaded where the agent replies. Pulses stop when the run returns, or, for a type that hands its work off and returns early, when every promise it passed to the new `Delivery.workingUntil` has settled. They also stop when the agent goes away, after 30 minutes, or after a pulse fails, so a relay that refuses typing is asked once per run. Native may now sign kind 20002 for an agent. Pulses are signed through native directly and the agent's echoed pulses are not recorded as something it wrote, so they cannot push real messages out of that set. Needs the relay to accept typing over HTTP POST /events (block/buzz#8207). Signed-off-by: Bradley Axen Co-authored-by: ART Co-authored-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 --- crates/agent-controller/src/app_agent.rs | 8 +- .../agent-controller/src/app_agent/tests.rs | 2 +- src/features/agents2/native.ts | 6 +- src/features/agents2/service.test.ts | 159 +++++++++++++++++- src/features/agents2/service.ts | 66 +++++++- 5 files changed, 230 insertions(+), 11 deletions(-) diff --git a/crates/agent-controller/src/app_agent.rs b/crates/agent-controller/src/app_agent.rs index e07c14580..f4529f031 100644 --- a/crates/agent-controller/src/app_agent.rs +++ b/crates/agent-controller/src/app_agent.rs @@ -10,10 +10,10 @@ use serde::{Deserialize, Serialize}; use std::path::PathBuf; /// What a plugin agent may publish: a deletion (5), a reaction (7), a message -/// (9), an edit of its own message (40003), a +/// (9), a typing pulse (20002), an edit of its own message (40003), a /// channel canvas (40100) or a request to open a DM (41010). Its profile (0) /// is the app's, and its memory (30174) is written by `memory`. -const KINDS: [u16; 6] = [5, 7, 9, 40003, 40100, 41010]; +const KINDS: [u16; 7] = [5, 7, 9, 20002, 40003, 40100, 41010]; /// One saved identity. The key itself stays in the OS credential store. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -77,7 +77,9 @@ impl AppAgent { tags: Vec>, ) -> Result { if !KINDS.contains(&kind) { - return Err("Agents can sign messages, edits, reactions and deletions only".into()); + return Err( + "Agents can sign messages, edits, reactions, deletions and typing only".into(), + ); } if content.len() > 64 * 1024 || tags.len() > 256 diff --git a/crates/agent-controller/src/app_agent/tests.rs b/crates/agent-controller/src/app_agent/tests.rs index 219ee8a2d..c28e96dfb 100644 --- a/crates/agent-controller/src/app_agent/tests.rs +++ b/crates/agent-controller/src/app_agent/tests.rs @@ -211,7 +211,7 @@ fn signs_only_bounded_kinds_with_the_saved_attestation() { let (_, agent) = created(dir.path(), &credentials); let key = agent.read_key(&credentials).unwrap(); let saved: Vec = serde_json::from_str(&agent.auth).unwrap(); - for kind in [5, 7, 9, 40003, 40100, 41010] { + for kind in [5, 7, 9, 20002, 40003, 40100, 41010] { let tags = vec![ vec!["auth".to_owned(), "forged".to_owned()], vec!["h".to_owned(), "channel".to_owned()], diff --git a/src/features/agents2/native.ts b/src/features/agents2/native.ts index ea6309a29..bb273accb 100644 --- a/src/features/agents2/native.ts +++ b/src/features/agents2/native.ts @@ -45,9 +45,9 @@ export type AgentsNative = { remove(pubkey: string): Promise; /** Drops a deleted identity. */ forget(pubkey: string): Promise; - /** Signs one event as the agent (deletions, reactions, messages, edits, - * canvases and DM opens only) with its owner attestation and posts it to the - * agent's community. */ + /** Signs one event as the agent (deletions, reactions, messages, typing + * pulses, edits, canvases and DM opens only) with its owner attestation and + * posts it to the agent's community. */ publish(pubkey: string, event: AgentEventTemplate): Promise; /** Reads the agent's community as the agent: one to eight filters. */ query(pubkey: string, filters: readonly object[]): Promise; diff --git a/src/features/agents2/service.test.ts b/src/features/agents2/service.test.ts index e810347ed..74adf4043 100644 --- a/src/features/agents2/service.test.ts +++ b/src/features/agents2/service.test.ts @@ -4,7 +4,7 @@ import type { RelayEvent } from "../relay/events"; import type { LiveBatch, LiveListener } from "../relay/incoming"; import type { RelayData, RelaySnapshot } from "../relay/service"; import type { RelaySession } from "../relay/session"; -import type { AgentIdentity, AgentsNative } from "./native"; +import type { AgentEventTemplate, AgentIdentity, AgentsNative } from "./native"; import { Agents2Service, type AgentType, type Delivery } from "./service"; import { memoryStorage } from "./test-fakes"; @@ -112,6 +112,7 @@ function fakeRelay() { function fakeNative(identities: AgentIdentity[] = []) { let published = 0; + let typed = 0; const keys = [bot, sibling]; const at = (pubkey: string) => identities.findIndex((identity) => identity.pubkey === pubkey); @@ -145,8 +146,11 @@ function fakeNative(identities: AgentIdentity[] = []) { steps.push("forget"); identities.splice(at(pubkey), 1); }), - publish: vi.fn(async (pubkey: string, template: { kind: number }) => - event(`p${++published}`, { pubkey, kind: template.kind }), + publish: vi.fn(async (pubkey: string, template: AgentEventTemplate) => + event(template.kind === 20002 ? `t${++typed}` : `p${++published}`, { + pubkey, + kind: template.kind, + }), ), publishProfile: vi.fn(async (_: string) => {}), query: vi.fn(async () => []), @@ -930,3 +934,152 @@ it("keeps an Interest while a watch still uses it", async () => { }); expect(service.find(bot)?.attention).toEqual({}); }); + +const typingCalls = (native: ReturnType) => + native.publish.mock.calls.filter(([, template]) => template.kind === 20002); + +it("shows a mention run as typing in its thread until the run and its handed-off work end", async () => { + vi.useFakeTimers({ now: 1_000_000 }); + const { service, native, run, emit } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + const root = "a1".padEnd(64, "0"); + const mention = event("m", { + tags: [ + ["h", channel], + ["e", root, "", "root"], + ["e", root, "", "reply"], + ["p", bot], + ], + }); + let finish: () => void = () => {}; + const work = new Promise((resolve) => (finish = resolve)); + let release: () => void = () => {}; + run.mockImplementationOnce( + ({ workingUntil }) => + new Promise((resolve) => { + workingUntil(work); + release = resolve; + }), + ); + emit({ events: [mention], channelId: channel }); + await vi.advanceTimersByTimeAsync(0); + expect(typingCalls(native)).toEqual([ + [ + bot, + { + kind: 20002, + content: "", + tags: [ + ["h", channel], + ["e", root, "", "root"], + ["e", mention.id, "", "reply"], + ], + }, + ], + ]); + await vi.advanceTimersByTimeAsync(3_000); + expect(typingCalls(native)).toHaveLength(2); + // `run` returning does not end typing while the handed-off work goes on. + release(); + await vi.advanceTimersByTimeAsync(3_000); + expect(typingCalls(native)).toHaveLength(3); + finish(); + await vi.advanceTimersByTimeAsync(9_000); + expect(typingCalls(native)).toHaveLength(3); + // A top-level mention is the root of the thread the reply starts. + emit({ + events: [ + event("b2", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + }); + await vi.advanceTimersByTimeAsync(0); + expect(typingCalls(native)[3]?.[1].tags).toEqual([ + ["h", channel], + ["e", "b2".padEnd(64, "0"), "", "reply"], + ]); + await vi.advanceTimersByTimeAsync(9_000); + expect(typingCalls(native)).toHaveLength(4); +}); + +it("sends no typing for watch runs, and stops a run's typing when a pulse fails or the agent goes", async () => { + vi.useFakeTimers({ now: 1_000_000 }); + const { service, native, run, emit } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + // Every run hands off work that never ends. + run.mockImplementation(({ workingUntil }) => { + workingUntil(new Promise(() => {})); + }); + emit({ events: [event("w", { content: "deploy" })], channelId: channel }); + await vi.advanceTimersByTimeAsync(9_000); + expect(run).toHaveBeenCalledTimes(1); + expect(typingCalls(native)).toHaveLength(0); + // A relay that refuses typing is asked once a run. + native.publish.mockRejectedValueOnce(new Error("unknown event kind")); + emit({ + events: [ + event("m", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + }); + await vi.advanceTimersByTimeAsync(9_000); + expect(run).toHaveBeenCalledTimes(2); + expect(typingCalls(native)).toHaveLength(1); + emit({ + events: [ + event("n", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + }); + await vi.advanceTimersByTimeAsync(3_000); + expect(typingCalls(native)).toHaveLength(3); + await service.remove(bot); + await vi.advanceTimersByTimeAsync(9_000); + expect(typingCalls(native)).toHaveLength(3); +}); + +it("does not count its own typing pulses as something it wrote", async () => { + const { service, run, emit } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + const pulse = event("own", { pubkey: bot, kind: 20002 }); + emit({ events: [pulse] }); + emit({ events: [event("r", { tags: [["e", pulse.id]] })] }); + await settle(); + expect(run).not.toHaveBeenCalled(); +}); + +it("stops typing for handed-off work after the longest run allowed", async () => { + vi.useFakeTimers({ now: 1_000_000 }); + const { service, native, run, emit } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + run.mockImplementation(({ workingUntil }) => { + workingUntil(new Promise(() => {})); + }); + emit({ + events: [ + event("m", { + tags: [ + ["h", channel], + ["p", bot], + ], + }), + ], + }); + await vi.advanceTimersByTimeAsync(30 * 60_000 - 1); + const sent = typingCalls(native).length; + expect(sent).toBe(600); + await vi.advanceTimersByTimeAsync(9_000); + expect(typingCalls(native)).toHaveLength(sent); +}); diff --git a/src/features/agents2/service.ts b/src/features/agents2/service.ts index 770c35194..ebf6fce6c 100644 --- a/src/features/agents2/service.ts +++ b/src/features/agents2/service.ts @@ -17,6 +17,7 @@ import type { LiveBatch } from "../relay/incoming"; import { relayPartition } from "../relay/partition"; import type { RelayData } from "../relay/service"; import type { RelaySession } from "../relay/session"; +import { threadReference } from "../relay/thread-reference"; import { addressedTo, CHAT_KINDS, @@ -131,6 +132,10 @@ export type Delivery = Readonly<{ /** Aborts on the run's deadline, or when the agent is removed or its type is * replaced. Advisory: the handle keeps working, so finishing up is harmless. */ signal: AbortSignal; + /** Keeps the agent shown as typing after `run` returns, until `done` + * settles, for a type that hands its work off and returns early. Only + * mention runs show typing; elsewhere this does nothing. */ + workingUntil(done: PromiseLike): void; }>; export type AgentType = { id: string; @@ -186,6 +191,8 @@ const RATE_LIMIT = 60; const RATE_WINDOW_MS = 60_000; const TIMEOUT_LIMIT_MS = 30 * 60_000; const TIMER_TICK_MS = 5_000; +/** Clients show a typing pulse for 8 s, so one every 3 s keeps it steady. */ +const TYPING_PULSE_MS = 3_000; const EMPTY = Object.freeze({}); const blank = (pubkey: string): AgentRecord => ({ pubkey, @@ -708,7 +715,8 @@ export class Agents2Service extends Service implements Agents2 { bounded(runner.seen, event.id); // An agent never hears itself; its own events make replies addressed. if (event.pubkey === agent.pubkey) { - bounded(runner.wrote, event.id); + // Its typing pulses would push real messages out of `wrote`. + if (event.kind !== 20002) bounded(runner.wrote, event.id); continue; } for (const trigger of this.heard(runner, agent, event)) @@ -851,6 +859,7 @@ export class Agents2Service extends Service implements Agents2 { remember: (slug: string, body: string, after: number) => this.require().remember(agent.pubkey, slug, body, after), }); + const typing = this.typing(runner, job); try { await Promise.race([ Promise.resolve().then(() => @@ -861,6 +870,8 @@ export class Agents2Service extends Service implements Agents2 { agent: handle, config: agent.config, signal, + workingUntil: (done: PromiseLike) => + typing?.hold(done), }), ), ), @@ -874,11 +885,64 @@ export class Agents2Service extends Service implements Agents2 { } catch (error) { if (!lifetime.aborted) console.error(`Agent run failed: ${agent.name}`, error); + } finally { + typing?.release(); } } runner.running = false; } + /** Shows a mention run as typing where the agent will reply: a pulse now and + * every few seconds until the run and everything it handed to `workingUntil` + * have settled, the agent goes away, or the longest run allowed has passed. + * Pulses skip `publish` so they never count as something the agent wrote. A + * failed pulse ends them, so a relay that refuses typing is asked once a run. */ + private typing(runner: Runner, { trigger, channelId }: Job) { + const native = this.native; + if (!native || trigger.type !== "mention") return undefined; + const { event } = trigger; + const channel = channelId ?? event.tags.find((tag) => tag[0] === "h")?.[1]; + if (!channel) return undefined; + const root = threadReference(event)?.rootId ?? event.id; + const tags = [ + ["h", channel], + ...(root === event.id ? [] : [["e", root, "", "root"]]), + ["e", event.id, "", "reply"], + ]; + const lifetime = runner.controller.signal; + const started = Date.now(); + let holds = 1; + let sending = false; + const pulse = () => { + if (Date.now() - started >= TIMEOUT_LIMIT_MS) return stop(); + if (sending) return; + sending = true; + native.publish(runner.pubkey, { kind: 20002, content: "", tags }).then( + () => (sending = false), + () => stop(), + ); + }; + const timer = setInterval(pulse, TYPING_PULSE_MS); + const stop = () => { + holds = 0; + clearInterval(timer); + lifetime.removeEventListener("abort", stop); + }; + lifetime.addEventListener("abort", stop, { once: true }); + pulse(); + const release = () => { + if (holds > 0 && --holds === 0) stop(); + }; + return { + hold(done: PromiseLike) { + if (holds === 0) return; + holds++; + Promise.resolve(done).then(release, release); + }, + release, + }; + } + private require() { if (!this.native) throw new Error("Agents run only in the desktop app"); return this.native; From ade61ce341bef38c0aef9aea0c209c840dec71eb Mon Sep 17 00:00:00 2001 From: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Date: Fri, 9 Oct 2026 13:48:40 -0700 Subject: [PATCH 2/2] feat(claude-code): show the agent as typing for its whole turn Claude Code's run returns as soon as the turn is handed to its session, so typing stopped after the first pulse. Hand the turn's promise to `workingUntil` so the pulses last until the turn settles. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> --- src/bundled/claude-code/runtime.test.ts | 41 ++++++++++++++++++++++++- src/bundled/claude-code/runtime.ts | 10 +++++- 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/src/bundled/claude-code/runtime.test.ts b/src/bundled/claude-code/runtime.test.ts index 437c2410f..2c2e4bcf9 100644 --- a/src/bundled/claude-code/runtime.test.ts +++ b/src/bundled/claude-code/runtime.test.ts @@ -77,6 +77,7 @@ function setup(thread: readonly EventData[] = []) { upload: vi.fn(), remember: vi.fn(), }; + const workingUntil = vi.fn(); const deliver = (trigger: Delivery["trigger"]) => runtime.run({ trigger, @@ -84,10 +85,21 @@ function setup(thread: readonly EventData[] = []) { agent, config: DEFAULT_CONFIG, signal: new AbortController().signal, + workingUntil, } as Delivery); const claudes = () => fake.processes.filter((process) => process.id === "claude"); - return { runtime, deliver, published, claudes, fake, read, data, agent }; + return { + runtime, + deliver, + published, + claudes, + fake, + read, + data, + agent, + workingUntil, + }; } it("hands a mention to its thread's session", async () => { @@ -247,6 +259,32 @@ it("shows a thread's later turns only what the session has not seen", async () = ); }); +it("keeps the agent shown as typing until its turn settles", async () => { + const mention = message("2", "@Claude hi", { + tags: [ + ["h", channel], + ["p", self], + ], + }); + const { runtime, deliver, claudes, workingUntil } = setup(); + runtime.sync([{ pubkey: self, config: DEFAULT_CONFIG } as Agent]); + await flush(10); + const [spare] = claudes(); + if (!spare) throw new Error("no spare"); + spare.hold = true; + await deliver({ type: "mention", event: mention }); + expect(workingUntil).toHaveBeenCalledTimes(1); + let settled = false; + void Promise.resolve(workingUntil.mock.calls[0]?.[0]).finally( + () => (settled = true), + ); + await flush(10); + expect(settled).toBe(false); + spare.exit(1, "Error: Invalid API key\n"); + await flush(10); + expect(settled).toBe(true); +}); + it("says so in the thread when a turn fails", async () => { const mention = message("2", "@Claude hi", { tags: [ @@ -322,6 +360,7 @@ it("acts only on its owner's messages unless told to answer anyone", async () => }, config: { ...DEFAULT_CONFIG, respondTo: "anyone" }, signal: new AbortController().signal, + workingUntil: vi.fn(), } as unknown as Delivery); await flush(10); expect(claudes().some((process) => process.prompts.length)).toBe(true); diff --git a/src/bundled/claude-code/runtime.ts b/src/bundled/claude-code/runtime.ts index 9f79b0c8f..c3507166a 100644 --- a/src/bundled/claude-code/runtime.ts +++ b/src/bundled/claude-code/runtime.ts @@ -139,7 +139,13 @@ export class ClaudeRuntime { this.agents.clear(); } - async run({ trigger, agent, config: raw, channelId }: Delivery) { + async run({ + trigger, + agent, + config: raw, + channelId, + workingUntil, + }: Delivery) { const settings = config(raw); if ( trigger.type !== "timer" && @@ -194,6 +200,8 @@ export class ClaudeRuntime { steer: steerPrompt(text), fresh: async () => (await prompt(true)).text, }); + // The turn outlives `run`; keep the agent shown as typing until it settles. + workingUntil(done); this.notify(); this.watch(done, async (error) => { if (error) await this.report(agent, channel, event, error);