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,