feat(agents2): show agents as typing while they work on a mention

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<Vec<String>>,
     ) -> Result<serde_json::Value> {
         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

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<String> = 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()],

src/bundled/claude-code/runtime.test.ts

@@ -77,17 +77,29 @@ function setup(thread: readonly EventData[] = []) {
     upload: vi.fn(),
     remember: vi.fn(),
   };
+  const workingUntil = vi.fn();
   const deliver = (trigger: Delivery["trigger"]) =>
     runtime.run({
       trigger,
       channelId: channel,
       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);

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

src/features/agents2/native.ts

@@ -45,9 +45,9 @@ export type AgentsNative = {
   remove(pubkey: string): Promise<void>;
   /** Drops a deleted identity. */
   forget(pubkey: string): Promise<void>;
-  /** 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<RelayEvent>;
   /** Reads the agent's community as the agent: one to eight filters. */
   query(pubkey: string, filters: readonly object[]): Promise<RelayEvent[]>;

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<typeof fakeNative>) =>
+  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<void>((resolve) => (finish = resolve));
+  let release: () => void = () => {};
+  run.mockImplementationOnce(
+    ({ workingUntil }) =>
+      new Promise<void>((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);
+});