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/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); 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;