From 44b026f547436444966052940024cdebb8aacad8 Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Sun, 11 Oct 2026 11:13:52 -0700 Subject: [PATCH] feat(agents2): say the agents this copy runs are online App-hosted agents have no socket of their own, so they always read as offline, and an owner cannot tell from another device whether the laptop running them is awake. Each origin now sends one owner-signed kind 20003 batch for the agents this copy holds and runs: on a new session, when the set changes, and every minute otherwise (presence expires after three minutes). Agents it stops running are said offline, unless another copy has taken them over. A refused send is retried after 30 seconds rather than every tick. Relays without kind 20003 ignore it. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Bradley Axen --- src/features/agents/service.test.ts | 71 ++++++++++++++++++++++ src/features/agents/service.ts | 92 +++++++++++++++++++++++++++++ src/features/relay/session.ts | 21 +++++++ 3 files changed, 184 insertions(+) diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index b92556510..7d1841632 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -96,6 +96,13 @@ function fakeRelay(origin = "https://relay.example.test") { live.add(listener); return () => live.delete(listener); }, + publishAgentPresence: vi.fn( + async ( + _: "online" | "offline", + __: readonly string[], + ___: AbortSignal, + ) => {}, + ), } as unknown as RelaySession; let snapshot: RelaySnapshot = { status: "ready", @@ -145,6 +152,9 @@ function fakeRelay(origin = "https://relay.example.test") { archives, listed, session, + presence: session.publishAgentPresence as ReturnType< + typeof vi.fn + >, origin, }; } @@ -357,6 +367,7 @@ async function setup({ session: fake.session, authorize, plugin, + presence: fake.presence, second, select: client.select, leave: client.leave, @@ -2030,3 +2041,63 @@ it("keeps what it lists when the community's records cannot be read", async () = place: "hosted", }); }); + +/** What each presence call said, as `status:agents`. */ +const said = (presence: { mock: { calls: unknown[][] } }) => + presence.mock.calls.map( + ([status, agents]) => `${status}:${(agents as string[]).join(",")}`, + ); + +it("says the agents it runs are online, again every minute and on a new session, and offline once removed", async () => { + vi.useFakeTimers(); + const { service, presence, connect } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + await vi.advanceTimersByTimeAsync(5_000); + expect(said(presence)).toEqual([`online:${bot}`]); + await vi.advanceTimersByTimeAsync(50_000); + expect(said(presence)).toEqual([`online:${bot}`]); + await vi.advanceTimersByTimeAsync(10_000); + expect(said(presence)).toEqual([`online:${bot}`, `online:${bot}`]); + connect(true); + await vi.advanceTimersByTimeAsync(0); + expect(said(presence)).toHaveLength(3); + await service.remove(bot); + await vi.advanceTimersByTimeAsync(0); + expect(said(presence).at(-1)).toBe(`offline:${bot}`); + await vi.advanceTimersByTimeAsync(120_000); + expect(said(presence)).toHaveLength(4); +}); + +it("leaves an agent another copy of the app holds to that copy, and does not say it went offline", async () => { + vi.useFakeTimers(); + const { service, presence, native } = await setup(); + native.claim.mockResolvedValue(false); + await service.create({ type: "example/echo", name: "Echo" }); + await vi.advanceTimersByTimeAsync(10_000); + expect(said(presence)).toEqual([]); + native.claim.mockResolvedValue(true); + await vi.advanceTimersByTimeAsync(10_000); + expect(said(presence)).toEqual([`online:${bot}`]); + native.claim.mockResolvedValue(false); + await vi.advanceTimersByTimeAsync(120_000); + expect(said(presence)).toEqual([`online:${bot}`]); +}); + +it("tries a refused presence again after half a minute, not every tick", async () => { + vi.useFakeTimers(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { service, presence } = await setup(); + presence.mockRejectedValueOnce(new Error("rate-limited")); + await service.create({ type: "example/echo", name: "Echo" }); + await vi.advanceTimersByTimeAsync(5_000); + expect(presence).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(20_000); + expect(presence).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(10_000); + expect(presence).toHaveBeenCalledTimes(2); + expect(warn).toHaveBeenCalledWith( + "Agent presence was not sent", + expect.any(Error), + ); + warn.mockRestore(); +}); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index f20a896e1..9ae2a7b03 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -338,6 +338,10 @@ const TIMEOUT_LIMIT_MS = 30 * 60_000; const TIMER_TICK_MS = 5_000; /** How often a failed connection to an unselected community is tried again. */ const RETRY_MS = 30_000; +/** How often agents are said to be online; the relay keeps it 180 seconds. */ +const PRESENCE_MS = 60_000; +/** The most agents one presence event names. */ +const PRESENCE_BATCH = 256; const EMPTY = Object.freeze({}); const blank = (pubkey: string): AgentRecord => ({ pubkey, @@ -545,6 +549,9 @@ type Runner = { /** Whether native may hold the agent for this copy; true at first, since * native outlives a page reload. */ mayHold: boolean; + /** Whether a claim has been answered, so `mayHold` is known rather than + * assumed. */ + asked?: boolean; compiled?: { attention: Agent["attention"]; watches: readonly CompiledWatch[]; @@ -586,6 +593,13 @@ export class AgentsService extends Service implements Agents { private busy = new Set(); /** Hosted and elsewhere agents as last shown, kept while unchanged. */ private shown = new Map(); + /** By origin: the agents last said to be online there, and when. */ + private presence = new Map< + string, + { session: RelaySession; online: ReadonlySet; at: number } + >(); + /** Origins whose presence is being sent. */ + private announcing = new Set(); constructor( ctx: Context, @@ -626,6 +640,7 @@ export class AgentsService extends Service implements Agents { const tick = setInterval(() => { for (const runner of this.runners.values()) void this.drain(runner); this.reconnect(); + this.announce(); }, TIMER_TICK_MS); return () => { clearInterval(tick); @@ -635,6 +650,7 @@ export class AgentsService extends Service implements Agents { for (const origin of [...this.bound.keys()]) this.unbind(origin); for (const runner of this.runners.values()) this.retire(runner); this.runners.clear(); + this.presence.clear(); // Clear what is shown without reconciling, which would make new runners. this.state = Object.freeze({ ...this.state, @@ -1602,6 +1618,80 @@ export class AgentsService extends Service implements Agents { ) void this.drain(runner); this.connected = ready; + this.announce(); + } + + /** Tells each community which agents this copy runs there: online every + * minute while it does, and offline once it stops, unless another copy took + * over. Offline is best effort; the relay lets a lapsed online expire. */ + private announce() { + const online = new Map>(); + for (const agent of this.state.running) { + const origin = originOf(agent.relay); + if (origin && this.bindingOf(agent) && this.runs(agent.pubkey)) + online.set(origin, (online.get(origin) ?? new Set()).add(agent.pubkey)); + } + for (const origin of this.presence.keys()) + if (!this.bound.has(origin)) this.presence.delete(origin); + const now = Date.now(); + for (const [origin, entry] of this.bound) { + const binding = entry.binding; + if (!binding || this.announcing.has(origin)) continue; + const next = online.get(origin) ?? new Set(); + const prior = this.presence.get(origin); + const was = prior?.online ?? new Set(); + // Another copy that took an agent over says it is online; offline from + // this one could arrive after that. + const gone = [...was].filter( + (pubkey) => !next.has(pubkey) && !this.yielded(pubkey), + ); + const due = + prior?.session !== binding.session || now - prior.at >= PRESENCE_MS; + const changed = + gone.length > 0 || [...next].some((pubkey) => !was.has(pubkey)); + if (!changed && (!due || !next.size)) continue; + this.announcing.add(origin); + this.presence.set(origin, { + session: binding.session, + online: next, + at: now, + }); + void (async () => { + for (const [status, agents] of [ + ["offline", gone], + ["online", [...next]], + ] as const) + for (let at = 0; at < agents.length; at += PRESENCE_BATCH) + await binding.session.publishAgentPresence( + status, + agents.slice(at, at + PRESENCE_BATCH), + binding.controller.signal, + ); + })() + .catch((error) => { + if (binding.controller.signal.aborted) return; + console.warn("Agent presence was not sent", error); + // Try again sooner than the minute, but not every tick. + const current = this.presence.get(origin); + if (current?.online === next) + this.presence.set(origin, { + ...current, + at: Date.now() - PRESENCE_MS + RETRY_MS, + }); + }) + .finally(() => this.announcing.delete(origin)); + } + } + /** Whether this copy runs the agent: its type can run it, and no other + * copy of the app holds it. */ + private runs(pubkey: string) { + const runner = this.runners.get(pubkey); + return !!runner?.type?.run && !!runner.asked && runner.mayHold; + } + /** Whether another copy of the app runs an agent this one could. */ + private yielded(pubkey: string) { + const runner = this.runners.get(pubkey); + return !!runner?.type?.run && !runner.mayHold && !!this.ready(pubkey); } private notify() { for (const listener of this.listeners) listener(); @@ -1905,9 +1995,11 @@ export class AgentsService extends Service implements Agents { private async claim(runner: Runner) { try { runner.mayHold = (await this.native?.claim(runner.pubkey)) ?? false; + runner.asked = true; return runner.mayHold; } catch (error) { runner.mayHold = true; + runner.asked = true; // Answering twice beats not answering at all. console.warn( `Agent ${this.agent(runner.pubkey)?.name} was not claimed`, diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index d235309d0..b410908d5 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -1595,6 +1595,27 @@ export function createRelaySession( }; }, typing: typing.capability, + /** Says the viewer's `agents` are online or offline, on the live socket: + * ephemeral, so nothing is queued or retried. The relay keeps only the + * agents the viewer owns. */ + async publishAgentPresence( + status: "online" | "offline", + agents: readonly string[], + signal: AbortSignal, + ) { + if (closed || !writer || !traffic?.publish) + throw new Error("Agent presence needs a live connection"); + const event = await writer.sign( + { + kind: 20003, + content: status, + tags: agents.map((agent) => ["p", agent]), + created_at: Math.floor(Date.now() / 1000), + }, + signal, + ); + await traffic.publish(event, signal); + }, channelCreation, channelKit: channelKit.capability, canvas: channelKit.canvas,