feat(agents2): show agents working: 👀💬 reactions, local typing, Codex settles

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<Delivery["workingUntil"]>();
   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 () => {
@@ -201,6 +213,7 @@ it("runs a mention on the model and effort its delivery picked", async () => {
     model: "haiku",
     effort: "low",
     signal: new AbortController().signal,
+    workingUntil: () => undefined,
   } as Delivery);
   await flush(10);
   const claude = claudes().find((process) => process.prompts.length);
@@ -352,7 +365,7 @@ it("says so in the thread when a turn fails", async () => {
       ["p", self],
     ],
   });
-  const { runtime, deliver, published, claudes } = setup();
+  const { runtime, deliver, published, claudes, workingUntil } = setup();
   runtime.sync([{ pubkey: self, config: DEFAULT_CONFIG } as Agent]);
   await flush(10);
   const [spare] = claudes();
@@ -361,8 +374,16 @@ it("says so in the thread when a turn fails", async () => {
   await deliver({ type: "mention", event: mention });
   await flush(10);
   expect(spare.prompts).toHaveLength(1);
+  // The host counts the agent as working until the turn it handed off ends.
+  let settled = false;
+  const done = workingUntil.mock.calls[0]?.[0];
+  if (!done) throw new Error("the turn was not handed to workingUntil");
+  void Promise.resolve(done).then(() => (settled = true));
+  await flush(10);
+  expect(settled).toBe(false);
   spare.exit(1, "Error: Invalid API key\n");
   await flush(10);
+  expect(settled).toBe(true);
   const report = published.find((event) => event.kind === 9);
   expect(report?.content).toBe(
     "⚠️ I couldn't finish that: Error: Invalid API key",
@@ -421,6 +442,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: () => undefined,
   } as unknown as Delivery);
   await flush(10);
   expect(claudes().some((process) => process.prompts.length)).toBe(true);

src/bundled/claude-code/runtime.ts

@@ -225,6 +225,7 @@ export class ClaudeRuntime {
     channelId,
     model,
     effort,
+    workingUntil,
   }: Delivery) {
     const settings = config(raw);
     if (
@@ -250,6 +251,7 @@ export class ClaudeRuntime {
         }),
         0,
       );
+      workingUntil(done);
       this.watch(done, () => undefined);
       return;
     }
@@ -284,6 +286,8 @@ export class ClaudeRuntime {
       fresh: async () => (await prompt(true)).text,
       pick: { ...(model ? { model } : {}), ...(effort ? { effort } : {}) },
     });
+    // The turn outlives `run`, which the host is told so it shows it as working.
+    workingUntil(done);
     this.notify();
     this.watch(done, async (error) => {
       if (error) await this.report(agent, channel, event, error);

src/bundled/codex/README.md

@@ -39,6 +39,9 @@ While work is active, new mentions steer it, incorporating the follow-up into th
 ongoing task. Messages arriving during startup wait for the turn to start and
 then steer it; idle conversations continue the saved session. If a turn finishes
 while a follow-up is being delivered, the follow-up starts another turn.
+A successful final send stops the request counting as working at once; Codex
+still ends its turn with one more model call. Interrupting the turn instead would
+leave a `<turn_aborted>` note in the session after every reply.
 By default, Buzz tools reply to the latest accepted steering message. The agent can also choose another thread or channel with tool arguments.
 
 There are no special chat commands. Asking the agent to stop is a steering request

src/bundled/codex/runtime.test.ts

@@ -230,6 +230,7 @@ function fixture(
     size: 12,
     type: mime,
   }));
+  const workingUntil = vi.fn<Delivery["workingUntil"]>();
   const agent = {
     pubkey,
     owner,
@@ -266,6 +267,7 @@ function fixture(
         instructions: "TEST_INSTRUCTIONS",
       },
       signal: new AbortController().signal,
+      workingUntil,
     };
     if (result.trigger.type !== "timer")
       events.push(result.trigger.event as EventData);
@@ -383,6 +385,7 @@ function fixture(
     read,
     storage,
     delivery,
+    workingUntil,
     starts,
     complete,
     interrupt,
@@ -701,8 +704,17 @@ it("serializes owner follow-ups and routes tool replies to the latest accepted r
   await vi.waitFor(() =>
     expect(f.sent.filter((w) => w.method === "turn/steer")).toHaveLength(2),
   );
+  // Each mention counts as working until the turn it was steered into answers.
+  let settled = 0;
+  for (const [done] of f.workingUntil.mock.calls)
+    void Promise.resolve(done).then(() => settled++);
+  expect(f.workingUntil).toHaveBeenCalledTimes(3);
+  // A macrotask lets every pending promise chain finish first.
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  expect(settled).toBe(0);
   f.complete();
   await vi.waitFor(() => expect(f.publish).toHaveBeenCalledTimes(1));
+  await vi.waitFor(() => expect(settled).toBe(3));
   const steering = f.sent.filter((wire) => wire.method === "turn/steer");
   expect(JSON.stringify(steering[0]?.params)).toContain("FIRST_STEER");
   expect(JSON.stringify(steering[1]?.params)).toContain("SECOND_STEER");
@@ -1416,6 +1428,14 @@ it("delivers pending owner steering after a final send and posts its own answer"
   await vi.waitFor(() =>
     expect(f.sent.some((wire) => wire.method === "turn/steer")).toBe(true),
   );
+  // The prior answer settles only the first request.
+  let steered = false;
+  void Promise.resolve(f.workingUntil.mock.calls[1]?.[0]).then(() => {
+    steered = true;
+  });
+  await f.workingUntil.mock.calls[0]?.[0];
+  await new Promise((resolve) => setTimeout(resolve, 0));
+  expect(steered).toBe(false);
   await f.complete(0, "NEW ANSWER");
   await vi.waitFor(() =>
     expect(f.runtime.sessions(pubkey)[0]?.status).toBe("Idle"),
@@ -1439,6 +1459,9 @@ it("finishes final sends normally and preserves the agent's background terminals
   expect(f.publish).toHaveBeenCalledTimes(1);
   expect(f.runtime.sessions(pubkey)[0]?.status).toBe("Working");
   expect(f.sent.some((wire) => wire.method === "turn/interrupt")).toBe(false);
+  // The request is answered, so it stops counting as working before Codex's
+  // closing model call ends the turn.
+  await f.workingUntil.mock.calls[0]?.[0];
   const threadId = (f.starts()[0]?.params as { threadId: string } | undefined)
     ?.threadId;
   f.emit({

src/bundled/codex/runtime.ts

@@ -45,6 +45,8 @@ type Request = {
   /** The model and effort the owner picked for this turn. */
   model?: string;
   effort?: string;
+  /** Called once a final send answers this request, or one steered after it. */
+  answered: () => void;
 };
 type Saved = { threadId: string; workspace: string };
 type Active = {
@@ -60,6 +62,8 @@ type Active = {
   signal: AbortSignal;
   version: number;
   sentVersion?: number;
+  /** Every request this turn has taken in, the first and each steered one. */
+  requests: Request[];
 };
 type Lane = {
   tail: Promise<void>;
@@ -338,6 +342,12 @@ export class CodexRuntime {
     const toolRoot =
       channel?.channelType === "dm" ? undefined : rootOf(trigger.event);
     const root = settings.scope === "channel" ? undefined : toolRoot;
+    let answered!: () => void;
+    // 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 replied = new Promise<void>((resolve) => {
+      answered = resolve;
+    });
     const request: Request = {
       scope: snapshot.scope ?? "",
       event: trigger.event,
@@ -353,6 +363,7 @@ export class CodexRuntime {
       },
       interest:
         trigger.type === "watch" ? (trigger.interest?.instructions ?? "") : "",
+      answered,
     };
     if (!validWorkspace(settings.workspace)) {
       await this.publish(
@@ -394,6 +405,7 @@ export class CodexRuntime {
             ),
           });
           active.reply = request;
+          active.requests.push(request);
           active.version++;
         } catch (error) {
           if (entry.abort.signal.aborted) return;
@@ -411,9 +423,16 @@ export class CodexRuntime {
       active.steering = accepted.catch((error) =>
         console.error("Codex steering failed", error),
       );
+      // The steered turn, or one started in its place, answers this mention.
+      const steered = lane;
+      delivery.workingUntil(
+        Promise.race([replied, accepted.then(() => steered.tail)]),
+      );
       return;
     }
-    this.start(request, text, key, lane, entry);
+    delivery.workingUntil(
+      Promise.race([replied, this.start(request, text, key, lane, entry)]),
+    );
   }
   private start(
     request: Request,
@@ -437,6 +456,7 @@ export class CodexRuntime {
       .finally(() => {
         if (current.tail === job && !current.active) entry.lanes.delete(key);
       });
+    return job;
   }
   private async history(request: Request, signal: AbortSignal) {
     signal.throwIfAborted();
@@ -489,6 +509,7 @@ export class CodexRuntime {
     const args = call.arguments as Record<string, unknown>;
     const version = active.version;
     const request = active.reply;
+    const answering = [...active.requests];
     const client = appClient(request.agent, {
       spawn: this.spawn,
       cwd: active.workspace,
@@ -530,6 +551,7 @@ export class CodexRuntime {
     const final = call.tool === "send" && args.final === true;
     if (final) {
       active.sentVersion = Math.max(active.sentVersion ?? -1, version);
+      for (const taken of answering) taken.answered();
     }
     return { success: true, contentItems: [{ type: "inputText", text }] };
   }
@@ -599,6 +621,7 @@ export class CodexRuntime {
       workspace: request.settings.workspace,
       signal,
       version: 0,
+      requests: [request],
     };
     void active.ready.catch(() => undefined);
     lane.active = active;

src/features/agents2/local-typing.test.ts

@@ -0,0 +1,74 @@
+// STOPGAP(agent-local-typing): delete with local-typing.ts.
+import { afterEach, expect, it, vi } from "vitest";
+import type { RelayEvent } from "../relay/events";
+import type { RelaySession } from "../relay/session";
+import { typeLocally } from "./local-typing";
+
+const agent = "a".repeat(64);
+const root = "b".repeat(64);
+const mention = {
+  id: "c".repeat(64),
+  pubkey: "d".repeat(64),
+  kind: 9,
+  created_at: 1,
+  content: "@Agent hi",
+  tags: [
+    ["h", "channel"],
+    ["e", root, "", "reply"],
+  ],
+} as unknown as RelayEvent;
+
+afterEach(() => vi.useRealTimers());
+
+it("pulses in the mention's thread every three seconds until stopped", () => {
+  vi.useFakeTimers();
+  vi.setSystemTime(10_000);
+  const local = vi.fn();
+  const stop = typeLocally(
+    () => ({ typing: { local } }) as unknown as RelaySession,
+    agent,
+    mention,
+    "channel",
+  );
+  expect(local).toHaveBeenCalledExactlyOnceWith({
+    id: "",
+    pubkey: agent,
+    kind: 20002,
+    created_at: 10,
+    content: "",
+    tags: [
+      ["h", "channel"],
+      ["e", root, "", "root"],
+      ["e", mention.id, "", "reply"],
+    ],
+  });
+  vi.advanceTimersByTime(3_000);
+  expect(local).toHaveBeenCalledTimes(2);
+  expect(local.mock.lastCall?.[0].created_at).toBe(13);
+  stop();
+  // Stopping ends it at once, then nothing more is sent.
+  expect(local).toHaveBeenCalledTimes(3);
+  expect(local.mock.lastCall?.[0]).toMatchObject({
+    kind: 9,
+    tags: local.mock.calls[0]?.[0].tags,
+  });
+  vi.advanceTimersByTime(9_000);
+  expect(local).toHaveBeenCalledTimes(3);
+});
+
+it("skips pulses while the community is not ready, and stops if one throws", () => {
+  vi.useFakeTimers();
+  const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+  let session: RelaySession | undefined;
+  const local = vi.fn(() => {
+    throw new Error("closed");
+  });
+  typeLocally(() => session, agent, mention, "channel");
+  session = { typing: { local } } as unknown as RelaySession;
+  vi.advanceTimersByTime(3_000);
+  expect(local).toHaveBeenCalledOnce();
+  vi.advanceTimersByTime(9_000);
+  expect(local).toHaveBeenCalledOnce();
+  expect(warn).toHaveBeenCalledOnce();
+  warn.mockRestore();
+});