diff --git a/src/bundled/claude-code/runtime.test.ts b/src/bundled/claude-code/runtime.test.ts index ab58a6eda..ff0e2a751 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 () => { @@ -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); diff --git a/src/bundled/claude-code/runtime.ts b/src/bundled/claude-code/runtime.ts index dd5a91757..a8e076e06 100644 --- a/src/bundled/claude-code/runtime.ts +++ b/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); diff --git a/src/bundled/codex/README.md b/src/bundled/codex/README.md index dd92278f5..8eed47317 100644 --- a/src/bundled/codex/README.md +++ b/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 `` 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 diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index cbc1c7d38..20677e425 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -230,6 +230,7 @@ function fixture( size: 12, type: mime, })); + const workingUntil = vi.fn(); 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({ diff --git a/src/bundled/codex/runtime.ts b/src/bundled/codex/runtime.ts index 47559abbf..a043bf69a 100644 --- a/src/bundled/codex/runtime.ts +++ b/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; @@ -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((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; 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; diff --git a/src/features/agents2/local-typing.test.ts b/src/features/agents2/local-typing.test.ts new file mode 100644 index 000000000..08f4d8e85 --- /dev/null +++ b/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(); +}); diff --git a/src/features/agents2/local-typing.ts b/src/features/agents2/local-typing.ts new file mode 100644 index 000000000..f7c0604a9 --- /dev/null +++ b/src/features/agents2/local-typing.ts @@ -0,0 +1,59 @@ +// STOPGAP(agent-local-typing): DELETE once agents' typing pulses reach the +// relay. Every piece is tagged STOPGAP(agent-local-typing): grep for it and +// delete this file and its test, its import and use in service.ts, and +// `typing.local` in relay/typing.ts with its test and the fake in +// agents2/service.test.ts. +// +// While an agent this app runs works on a mention, it shows as typing there, +// on this device only. +import type { RelayEvent } from "../relay/events"; +import type { RelaySession } from "../relay/session"; +import { threadReference } from "../relay/thread-reference"; + +/** Clients show a pulse for 8 s, so one every 3 s keeps it steady. */ +const PULSE_MS = 3_000; + +/** Shows `pubkey` typing where it answers `event` until the returned function + * is called. `session` is read on every pulse, as the community may reconnect. */ +export function typeLocally( + session: () => RelaySession | undefined, + pubkey: string, + event: RelayEvent, + channelId: string, +): () => void { + const root = threadReference(event)?.rootId ?? event.id; + const tags = [ + ["h", channelId], + ...(root === event.id ? [] : [["e", root, "", "root"]]), + ["e", event.id, "", "reply"], + ]; + const send = (kind: number) => + session()?.typing.local({ + id: "", + pubkey, + kind, + created_at: Math.floor(Date.now() / 1000), + content: "", + tags, + }); + const pulse = () => { + try { + send(20002); + } catch (error) { + // A stopgap must never break the run it decorates. + clearInterval(timer); + console.warn("Agent typing was not shown", error); + } + }; + const timer = setInterval(pulse, PULSE_MS); + pulse(); + return () => { + clearInterval(timer); + // Ends it now rather than when the last pulse would expire. + try { + send(9); + } catch (error) { + console.warn("Agent typing was not ended", error); + } + }; +} diff --git a/src/features/agents2/service.test.ts b/src/features/agents2/service.test.ts index 6f111fbac..33534c7c5 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 AgentCommunities, @@ -65,6 +65,8 @@ function fakeRelay(origin = "https://relay.example.test") { const statuses = new Set<() => void>(); const session = { archives, + // STOPGAP(agent-local-typing): delete with local-typing.ts. + typing: { local: vi.fn() }, live: { snapshot: () => ({ status }), subscribe(listener: () => void) { @@ -231,8 +233,13 @@ 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(`p${++published}`, { + pubkey, + kind: template.kind, + content: template.content, + tags: (template.tags ?? []).map((tag) => [...tag]), + }), ), publishProfile: vi.fn(async (_: string) => {}), query: vi.fn(async () => []), @@ -325,6 +332,14 @@ async function setup({ * microtask queue is empty, so this does not depend on how fast runners are. */ const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); +/** The events native published as an agent, of `kind`, in order. */ +async function published(native: ReturnType, kind: number) { + const events = await Promise.all( + native.publish.mock.results.map(({ value }) => value), + ); + return events.filter((event) => event.kind === kind); +} + beforeEach(() => { vi.useRealTimers(); steps.length = 0; @@ -415,6 +430,71 @@ it("delivers mentions and matching watches once, and never the agent's own event expect(first?.config).toEqual({ reply: "ok" }); }); +// STOPGAP(agent-working-reactions): delete with working-reactions.ts. +it("reacts 👀 and 💬 to a mention until the work its run handed off settles", async () => { + const { service, run, emit, native, relay } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + let finish = () => {}; + const turn = new Promise((resolve) => { + finish = resolve; + }); + run.mockImplementationOnce(async ({ workingUntil }) => workingUntil(turn)); + const mention = event("m", { + tags: [ + ["h", channel], + ["p", bot], + ], + }); + emit({ events: [mention], channelId: channel }); + await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(1)); + await settle(); + const reactions = await published(native, 7); + expect(reactions.map(({ content, tags }) => [content, tags])).toEqual( + ["👀", "💬"].map((emoji) => [ + emoji, + [ + ["h", channel], + ["e", mention.id], + ], + ]), + ); + // STOPGAP(agent-local-typing): this device shows the agent typing in the + // mention's thread meanwhile. + expect(relay.snapshot().session.typing.local).toHaveBeenCalledWith( + expect.objectContaining({ + pubkey: bot, + kind: 20002, + tags: [ + ["h", channel], + ["e", mention.id, "", "reply"], + ], + }), + ); + // `run` has returned, but the turn it handed off is still going. + expect(await published(native, 5)).toEqual([]); + finish(); + await vi.waitFor(async () => + expect(await published(native, 5)).toHaveLength(2), + ); + // The relay takes one target per deletion. + expect((await published(native, 5)).map(({ tags }) => tags)).toEqual( + reactions.map(({ id }) => [ + ["h", channel], + ["e", id], + ["k", "7"], + ]), + ); + // Only mentions are reacted to, and a failed run takes its reactions back. + run.mockRejectedValueOnce(new Error("boom")); + emit({ events: [event("m2", { tags: mention.tags })], channelId: channel }); + emit({ events: [event("w", { content: "deploy" })], channelId: channel }); + await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(3)); + await vi.waitFor(async () => + expect(await published(native, 5)).toHaveLength(4), + ); + expect(await published(native, 7)).toHaveLength(4); +}); + it("treats a reply to something the agent published as addressed", async () => { const { service, run, emit } = await setup(); await service.create({ type: "example/echo", name: "Echo" }); @@ -736,7 +816,7 @@ it("is not woken by reactions, deletions or DMs", async () => { }); it("does not let one owner's agents wake each other by replying", async () => { - const { service, run, emit } = await setup(); + const { service, run, emit, native } = await setup(); await service.create({ type: "example/echo", name: "Echo" }); await service.create({ type: "example/echo", name: "Sibling" }); run.mockImplementation(async ({ agent, trigger }) => { @@ -753,7 +833,7 @@ it("does not let one owner's agents wake each other by replying", async () => { await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(1)); // The sibling replies to Echo's message, naming it, and also says "deploy", // which Echo's watch matches. Neither wakes Echo. - const echoMessage = "p1".padEnd(64, "0"); + const echoMessage = (await published(native, 9))[0]?.id; emit({ events: [ event("reply", { diff --git a/src/features/agents2/service.ts b/src/features/agents2/service.ts index a75a9ba29..94f5454a9 100644 --- a/src/features/agents2/service.ts +++ b/src/features/agents2/service.ts @@ -52,6 +52,10 @@ import { type AgentRecord, type SkippedObject, } from "./store"; +// STOPGAP(agent-local-typing): delete with local-typing.ts. +import { typeLocally } from "./local-typing"; +// STOPGAP(agent-working-reactions): delete with working-reactions.ts. +import { reactWhileWorking } from "./working-reactions"; /** One of the viewer's agents, in the community at `relay`. */ export type Agent = Readonly<{ @@ -145,6 +149,9 @@ 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 run counted as working after `run` returns, until `done` + * settles, for a type that hands its work off and returns early. */ + workingUntil(done: PromiseLike): void; }>; export type Effort = Readonly<{ id: string; label: string }>; /** A model with its effort levels, lowest first. */ @@ -1154,6 +1161,7 @@ export class Agents2Service extends Service implements Agents2 { remember: (slug: string, body: string, after: number) => this.require().remember(agent.pubkey, slug, body, after), }); + const working = this.working(runner, job); try { await Promise.race([ Promise.resolve().then(async () => { @@ -1167,6 +1175,7 @@ export class Agents2Service extends Service implements Agents2 { config: agent.config, ...pick, signal, + workingUntil: working.hold, }), ); }), @@ -1180,11 +1189,64 @@ export class Agents2Service extends Service implements Agents2 { } catch (error) { if (!lifetime.aborted) console.error(`Agent run failed: ${agent.name}`, error); + } finally { + working.release(); } } runner.running = false; } + /** Counts a run as working until it and everything it handed to + * `workingUntil` have settled, or the longest run allowed has passed. A + * mention run is reacted to, and shown typing here, meanwhile: both are + * STOPGAPs, and `done` is empty without them. */ + private working(runner: Runner, { trigger, channelId }: Job) { + const native = this.native; + const event = trigger.type === "mention" ? trigger.event : undefined; + const channel = + event && (channelId ?? event.tags.find((tag) => tag[0] === "h")?.[1]); + const done = + native && event && channel + ? [ + // STOPGAP(agent-working-reactions): delete with working-reactions.ts. + reactWhileWorking( + (template) => native.publish(runner.pubkey, template), + event, + channel, + ), + // STOPGAP(agent-local-typing): delete with local-typing.ts. + typeLocally( + () => { + const relay = this.relay(runner.pubkey)?.snapshot(); + return relay?.status === "ready" ? relay.session : undefined; + }, + runner.pubkey, + event, + channel, + ), + ] + : []; + let holds = 1; + const stop = () => { + if (holds < 0) return; + holds = -1; + clearTimeout(timer); + for (const end of done) end(); + }; + const timer = setTimeout(stop, TIMEOUT_LIMIT_MS); + const release = () => { + if (holds > 0 && --holds === 0) stop(); + }; + return { + hold(work: PromiseLike) { + if (holds <= 0) return; + holds++; + Promise.resolve(work).then(release, release); + }, + release, + }; + } + private async claim(runner: Runner) { try { runner.mayHold = (await this.native?.claim(runner.pubkey)) ?? false; diff --git a/src/features/agents2/working-reactions.ts b/src/features/agents2/working-reactions.ts new file mode 100644 index 000000000..193c4181c --- /dev/null +++ b/src/features/agents2/working-reactions.ts @@ -0,0 +1,58 @@ +// STOPGAP(agent-working-reactions): DELETE once Agents2 agents show as typing +// through the relay. Every piece is tagged STOPGAP(agent-working-reactions): +// grep for it and delete this file, its import and use in service.ts, and its +// test. Keep `Delivery.workingUntil`, which typing indicators need too. +// +// While an agent works on a mention it reacts to it with 👀 and 💬, then takes +// both back. +import type { RelayEvent } from "../relay/events"; +import type { AgentEventTemplate } from "./native"; + +const EMOJI = ["👀", "💬"] as const; + +/** Reacts to `event` as the agent now; the returned function takes the + * reactions back once they have been published. */ +export function reactWhileWorking( + publish: (event: AgentEventTemplate) => Promise, + event: RelayEvent, + channelId: string, +): () => void { + const reactions = Promise.allSettled( + EMOJI.map(async (emoji) => + publish({ + kind: 7, + content: emoji, + tags: [ + ["h", channelId], + ["e", event.id], + ], + }), + ), + ); + return () => + void reactions + .then((results) => { + const ids = results.flatMap((result) => + result.status === "fulfilled" ? [result.value.id] : [], + ); + if (ids.length < results.length) + console.warn("Agent working reactions were not all published"); + // The relay takes one target per deletion. + return Promise.all( + ids.map((id) => + publish({ + kind: 5, + content: "", + tags: [ + ["h", channelId], + ["e", id], + ["k", "7"], + ], + }), + ), + ); + }) + .catch((error) => + console.warn("Agent working reactions were not removed", error), + ); +} diff --git a/src/features/relay/typing.test.ts b/src/features/relay/typing.test.ts index fb72dd79c..a6e59e68c 100644 --- a/src/features/relay/typing.test.ts +++ b/src/features/relay/typing.test.ts @@ -309,3 +309,43 @@ describe("completion capacity", () => { expect(vi.getTimerCount()).toBe(0); }); }); +// STOPGAP(agent-local-typing): delete with agents2/local-typing.ts. +it("shows a local pulse like a live one, through the agent's posts until a local end", () => { + const { owner, snapshot } = setup(); + const local = { + id: "", + pubkey: agent.pubkey, + content: "", + created_at: epoch, + }; + owner.capability.local({ ...local, kind: 7, tags: [["h", "a"]] }); + expect(snapshot()).toEqual([]); + owner.capability.local({ ...local, kind: 20002, tags: [["h", "a"]] }); + expect(snapshot()).toEqual([{ channelId: "a", pubkey: agent.pubkey }]); + // A progress post leaves it typing, and the next pulse still counts. + owner.accept([message(agent, "a", "progress", epoch)], true); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(3_000); + owner.capability.local({ + ...local, + created_at: epoch + 3, + kind: 20002, + tags: [["h", "a"]], + }); + vi.advanceTimersByTime(6_000); + expect(snapshot()).toHaveLength(1); + // The local end hides it at once, and the agent's posts end it as usual + // once this app has stopped driving it. + owner.capability.local({ + ...local, + created_at: epoch + 9, + kind: 9, + tags: [["h", "a"]], + }); + expect(snapshot()).toEqual([]); + vi.advanceTimersByTime(3_000); + owner.accept([pulse(undefined, epoch + 12)], true); + expect(snapshot()).toHaveLength(1); + owner.accept([message(agent, "a", "done", epoch + 12)], true); + expect(snapshot()).toEqual([]); +}); diff --git a/src/features/relay/typing.ts b/src/features/relay/typing.ts index 458206886..131dd534c 100644 --- a/src/features/relay/typing.ts +++ b/src/features/relay/typing.ts @@ -1,4 +1,4 @@ -import type { RelayEvent } from "./events"; +import type { EventData, RelayEvent } from "./events"; import { threadReference } from "./thread-reference"; const ACTIVITY_LIFETIME_MS = 8_000; @@ -15,6 +15,9 @@ type ParticipantActivity = { lastMessageAt: number; visibleUntil: number; quietUntil: number; + // STOPGAP(agent-local-typing): delete with agents2/local-typing.ts. + /** Driven by this app's own pulses, which end only with a local end. */ + local?: boolean; }; /** Session-owned activity from verified events; no event payloads or persistence. */ @@ -86,7 +89,7 @@ export function createTyping( record.visibleUntil = 0; record.quietUntil = now + POST_MESSAGE_QUIET_MS; } - function receive(event: RelayEvent, now: number) { + function receive(event: RelayEvent, now: number, local = false) { const typing = event.kind === 20002; if (event.pubkey === viewer || !/^[0-9a-f]{64}$/.test(event.pubkey)) return; const at = event.created_at * 1000; @@ -139,10 +142,19 @@ export function createTyping( }; records.set(key, record); } + // STOPGAP(agent-local-typing): an agent's progress posts would hide its + // typing between pulses; the local end says when the turn is over. + if (local) record.local = typing; + else if (!typing && record.local) return; if (typing) receiveActivity(record, at, now); else recordCompletion(record, at, now); } - function accept(events: readonly RelayEvent[], live = false) { + function accept( + events: readonly RelayEvent[], + live = false, + // STOPGAP(agent-local-typing) + local = false, + ) { if (closed) return; const now = Date.now(); // Keep completion evidence until old pulses can no longer be current. @@ -150,12 +162,12 @@ export function createTyping( expireSilence(now); for (const event of events) { if (event.kind === 9 || event.kind === 40002 || event.kind === 40008) - receive(event, now); + receive(event, now, local); } // Completion wins even when activity came first in the batch. if (live) { for (const event of events) { - if (event.kind === 20002) receive(event, now); + if (event.kind === 20002) receive(event, now, local); } } publish(); @@ -167,6 +179,15 @@ export function createTyping( return { capability: Object.freeze({ snapshot: () => snapshot, + // STOPGAP(agent-local-typing): delete with agents2/local-typing.ts. + /** Takes a pulse (20002) this app made for an agent it runs, or the + * end of its turn (9, as if the agent had replied), on this device only, + * as if it came live, until agents' pulses reach the relay. Until that + * end, the agent's own posts there leave it typing. */ + local(event: EventData) { + if (event.kind === 20002 || event.kind === 9) + accept([event as RelayEvent], true, true); + }, subscribe(listener: () => void) { if (closed) return () => {}; listeners.add(listener);