diff --git a/src/features/agents/agents-runner.test.ts b/src/features/agents/agents-runner.test.ts new file mode 100644 index 000000000..c9904cf81 --- /dev/null +++ b/src/features/agents/agents-runner.test.ts @@ -0,0 +1,60 @@ +import { expect, it, vi } from "vitest"; +import { AgentsRunner, type RunnerManager } from "./agents-runner"; +import { createLocalAgentActivity } from "./local-activity"; +import type { Member } from "./records"; +import { memoryStorage } from "./test-fakes"; + +const owner = "b".repeat(64); +const ada: Member = { + key: "a".repeat(64), + pubkey: "a".repeat(64), + relay: "wss://relay.example", + owner, + type: "test/agent", + name: "Ada", + deleted: false, +}; +const gone = "c".repeat(64); + +function saved() { + const storage = memoryStorage(); + for (const pubkey of [ada.pubkey, gone]) + storage.setItem( + `buzz.agent.v1:${pubkey}`, + JSON.stringify({ pubkey, attention: {}, config: {} }), + ); + return storage; +} + +function runner(storage: ReturnType) { + const manager = { + ownershipChanged: () => {}, + changed: () => {}, + open: () => undefined, + types: () => [], + defaults: () => { + throw new Error("No types here"); + }, + } as unknown as RunnerManager; + return new AgentsRunner( + storage, + undefined, + createLocalAgentActivity(), + undefined, + manager, + ); +} + +it("forgets records no membership keeps only once the whole list is read", async () => { + const storage = saved(); + const agents = runner(storage); + expect(agents.records.keys().sort()).toEqual([ada.pubkey, gone]); + // Before native's list is read, an empty list names nobody. + agents.members([], owner, false); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(agents.records.keys().sort()).toEqual([ada.pubkey, gone]); + agents.members([ada], owner, true); + await vi.waitFor(() => expect(agents.records.keys()).toEqual([ada.pubkey])); + expect(storage.getItem(`buzz.agent.v1:${gone}`)).toBeNull(); + agents.clear(); +}); diff --git a/src/features/agents/agents-runner.ts b/src/features/agents/agents-runner.ts index 287f2ee7d..6daeaaaa3 100644 --- a/src/features/agents/agents-runner.ts +++ b/src/features/agents/agents-runner.ts @@ -89,7 +89,10 @@ export class AgentsRunner { /** The memberships it last decided from, and their viewer. */ private identities: readonly Member[] = []; private viewer: string | undefined; + /** The last full list it forgot records by. */ + private list: string | undefined; private readonly listeners = new Set<() => void>(); + private readonly recordListeners = new Set<() => void>(); constructor( storage: AgentStore, @@ -111,7 +114,10 @@ export class AgentsRunner { console.warn(`Agent type ${type} has invalid defaults`, error); } }, - changed: () => this.changed(), + changed: () => { + for (const listener of [...this.recordListeners]) listener(); + this.changed(); + }, // Its timers waited for this change, saved or failed. settled: (key) => this.runners.wake(key), }); @@ -276,16 +282,51 @@ export class AgentsRunner { this.listeners.delete(listener); }; } + /** Hears each change to a record. */ + subscribeRecords(listener: () => void) { + this.recordListeners.add(listener); + return () => { + this.recordListeners.delete(listener); + }; + } private notify() { for (const listener of [...this.listeners]) listener(); } - /** The viewer's memberships, as the Manager last sent them: connects each - * community where the viewer has an agent, and decides again what runs. */ - members(identities: readonly Member[], viewer: string | undefined) { + /** The memberships on this device, as the Manager last sent them, and + * whether they are all of them (`loaded`): connects each community where + * the viewer has an agent, decides again what runs, and forgets records + * no membership keeps. */ + members( + identities: readonly Member[], + viewer: string | undefined, + loaded: boolean, + ) { + // Only a new list can name a record no membership keeps. + const list = JSON.stringify(identities); + const listed = loaded && list !== this.list; + if (loaded) this.list = list; this.identities = identities; this.viewer = viewer; this.connections.watch(identities, viewer); this.update(); + if (listed) this.forget(); + } + /** Removes the settings whose agent is gone, or that a community it left + * kept: they do nothing. Its home record keeps its config while it is + * anywhere. */ + private forget() { + const live = new Set( + this.identities + .filter((saved) => !saved.deleted) + .flatMap((saved) => [saved.key, saved.pubkey]), + ); + for (const key of this.records.keys()) + if (!live.has(key)) + void this.records + .edit(key, { kind: "remove" }) + .catch((error) => + console.warn(`Agent ${key} settings were not removed`, error), + ); } /** Decides again what runs once a record, a turn in progress or one of its * sessions changed, and lets the Manager show it. */ @@ -445,6 +486,7 @@ export class AgentsRunner { this.runners.clear(); this.identities = []; this.viewer = undefined; + this.list = undefined; this.state = { running: [], interrupted: [] }; } diff --git a/src/features/agents/link.test.ts b/src/features/agents/link.test.ts index 97d3ef5c5..6f84ba4c7 100644 --- a/src/features/agents/link.test.ts +++ b/src/features/agents/link.test.ts @@ -189,3 +189,23 @@ it("does not answer a call for a page that has since reloaded", async () => { release("slow"); await expect(answer).resolves.toBe("fast"); }); + +it("sends what a command changed before its result", async () => { + const count = store(1); + const { first, second } = pages(); + first({ + topics: { count: () => count }, + commands: { + bump: async () => { + await Promise.resolve(); + count.set(count.get() + 1); + return "done"; + }, + }, + }); + const manager = second(); + manager.watch("count", null, () => {}); + await vi.waitFor(() => expect(manager.read("count")).toBe(1)); + await expect(manager.call("bump")).resolves.toBe("done"); + expect(manager.read("count")).toBe(2); +}); diff --git a/src/features/agents/runner-link.test.ts b/src/features/agents/runner-link.test.ts index 61a26ac66..9a9153b4a 100644 --- a/src/features/agents/runner-link.test.ts +++ b/src/features/agents/runner-link.test.ts @@ -1,7 +1,8 @@ import { afterEach, expect, it, vi } from "vitest"; import { link, type Link, type LinkPort } from "./link"; import { createLocalAgentActivity } from "./local-activity"; -import type { Member } from "./records"; +import { agentKey } from "./membership"; +import type { Member, RecordEdit } from "./records"; import { linkedActivity, managerSide, @@ -9,6 +10,7 @@ import { serveActivity, } from "./runner-link"; import type { Agent } from "./service"; +import type { AgentRecord } from "./store"; const open: Link[] = []; afterEach(() => { @@ -111,21 +113,25 @@ function fakeRunner() { const listeners = new Set<() => void>(); let running: readonly Agent[] = []; const left = new Set(); + const recordListeners = new Set<() => void>(); + let saved: Record = {}; const runner = { - members: vi.fn((identities: readonly Member[], _viewer?: string) => { - const next = identities.map( - (identity) => - running.find( - (agent) => - agent.key === identity.key && agent.name === identity.name, - ) ?? agentOf(identity), - ); - const changed = - next.length !== running.length || - next.some((agent, index) => agent !== running[index]); - running = next; - if (changed) for (const listener of [...listeners]) listener(); - }), + members: vi.fn( + (identities: readonly Member[], _viewer?: string, _loaded?: boolean) => { + const next = identities.map( + (identity) => + running.find( + (agent) => + agent.key === identity.key && agent.name === identity.name, + ) ?? agentOf(identity), + ); + const changed = + next.length !== running.length || + next.some((agent, index) => agent !== running[index]); + running = next; + if (changed) for (const listener of [...listeners]) listener(); + }, + ), running: () => running, interrupted: () => running.filter((agent) => left.has(agent.key)), subscribe(listener: () => void) { @@ -134,21 +140,48 @@ function fakeRunner() { listeners.delete(listener); }; }, + records: { + all: () => Object.values(saved), + edit: vi.fn(async (key: string, edit: RecordEdit) => { + await Promise.resolve(); + if (edit.kind === "remove") { + const { [key]: _, ...rest } = saved; + saved = rest; + } else if (edit.kind === "create") + saved = { ...saved, [key]: edit.record }; + for (const listener of [...recordListeners]) listener(); + }), + }, + subscribeRecords(listener: () => void) { + recordListeners.add(listener); + return () => { + recordListeners.delete(listener); + }; + }, }; return { runner, left }; } +/** A new agent's settings, as its type gives them. */ +const defaults = vi.fn( + ({ pubkey }: Pick): AgentRecord => ({ + pubkey, + attention: {}, + config: { reply: "default" }, + }), +); + function split(identities: Member[] = []) { const { runner, left } = fakeRunner(); const channel = new MessageChannel(); let shown: Member[] = identities; const changed = vi.fn(); const runnerLink = runnerSide(channel.port1, runner); - const manager = managerSide( - channel.port2, - () => ({ identities: shown, viewer: owner }), + const manager = managerSide(channel.port2, { + members: () => ({ loaded: true, identities: shown, viewer: owner }), + defaults, changed, - ); + }); open.push(runnerLink); closing.push(manager); return { @@ -172,7 +205,7 @@ it("lets the Runner decide from the Manager's memberships, and waits for it", as const { runner, manager, show } = split(); show([ada]); await manager.decided(); - expect(runner.members).toHaveBeenLastCalledWith([ada], owner); + expect(runner.members).toHaveBeenLastCalledWith([ada], owner, true); expect(manager.running()).toEqual([agentOf(ada)]); expect(manager.agent(ada.key)?.name).toBe("Ada"); expect(Object.isFrozen(manager.running()[0])).toBe(true); @@ -196,6 +229,7 @@ it("settles a wait even when nothing that runs changed", async () => { expect(runner.members).toHaveBeenLastCalledWith( [{ ...ada, deleting: true }], owner, + true, ); expect(manager.running()).toBe(shown); }); @@ -257,11 +291,15 @@ it("does not take what the Runner decided for a page that reloaded as an answer" const channel = new MessageChannel(); to = channel.port1; channel.port1.onmessage = (event) => port.onmessage?.(event); - const page = managerSide( - channel.port2, - () => ({ identities: identities(), viewer: owner }), - () => {}, - ); + const page = managerSide(channel.port2, { + members: () => ({ + loaded: true, + identities: identities(), + viewer: owner, + }), + defaults: () => undefined, + changed: () => {}, + }); closing.push(page); return page; }; @@ -275,11 +313,65 @@ it("does not take what the Runner decided for a page that reloaded as an answer" shown = [ada, bo]; first.changed(); await first.decided(); - expect(runner.members).toHaveBeenLastCalledWith([ada, bo], owner); + expect(runner.members).toHaveBeenLastCalledWith([ada, bo], owner, true); // The reloaded page's first revision is not the Runner's last one. const second = load(() => [bo]); second.changed(); await second.decided(); - expect(runner.members).toHaveBeenLastCalledWith([bo], owner); + expect(runner.members).toHaveBeenLastCalledWith([bo], owner, true); expect(second.running().map((agent) => agent.name)).toEqual(["Bo"]); }); + +it("shows the Runner's records, and settles an edit once it is shown", async () => { + const ada = member("a".repeat(64), "Ada"); + const { runner, manager, show, changed } = split(); + show([ada]); + await manager.decided(); + // Until the Runner keeps one, its type's defaults, the same each time. + const first = manager.records.settingsOf(ada); + expect(first.config).toEqual({ reply: "default" }); + expect(manager.records.settingsOf(ada)).toBe(first); + expect(manager.records.keys()).toEqual([]); + const record = { pubkey: ada.key, attention: {}, config: { reply: "mine" } }; + changed.mockClear(); + await manager.records.edit(ada.key, { kind: "create", record }); + expect(runner.records.edit).toHaveBeenCalledWith(ada.key, { + kind: "create", + record, + }); + expect(changed).toHaveBeenCalled(); + expect(manager.records.keys()).toEqual([ada.key]); + const kept = manager.records.recordOf(ada); + expect(kept).toEqual(record); + // An unchanged record keeps its identity when another changes. + const bo = member("c".repeat(64), "Bo"); + await manager.records.edit(bo.key, { + kind: "create", + record: { ...record, pubkey: bo.key }, + }); + expect(manager.records.recordOf(ada)).toBe(kept); + await manager.records.edit(ada.key, { kind: "remove" }); + expect(manager.records.keys()).toEqual([bo.key]); +}); + +it("gives a membership outside its home its own attention and the agent's config", async () => { + const ada = member("a".repeat(64), "Ada"); + const away: Member = { + ...ada, + key: agentKey(ada.pubkey, "https://other.example", "https://relay.example"), + relay: "wss://other.example", + home: "wss://relay.example", + }; + const { manager, show } = split(); + show([ada, away]); + await manager.decided(); + await manager.records.edit(ada.key, { + kind: "create", + record: { pubkey: ada.key, attention: {}, config: { reply: "home" } }, + }); + expect(manager.records.recordOf(away)).toEqual({ + pubkey: away.key, + attention: {}, + config: { reply: "home" }, + }); +}); diff --git a/src/features/agents/runner-link.ts b/src/features/agents/runner-link.ts index 9e0384b0c..33b62e832 100644 --- a/src/features/agents/runner-link.ts +++ b/src/features/agents/runner-link.ts @@ -5,7 +5,8 @@ import type { AgentsRunner } from "./agents-runner"; import { link, type Json, type Link, type LinkPort, type Served } from "./link"; import type { LocalAgentSource, LocalWork } from "./local-activity"; import type { AgentKey } from "./membership"; -import type { Member } from "./records"; +import { blank, type Member, type RecordEdit } from "./records"; +import type { AgentRecord } from "./store"; import type { Agent } from "./service"; type WorkKey = Pick; @@ -74,6 +75,8 @@ export function linkedActivity(runner: Link) { type Members = Readonly<{ from: string; revision: number; + /** The list was read, so it names every membership on this device. */ + loaded: boolean; identities: readonly Member[]; viewer: string | null; }>; @@ -93,8 +96,9 @@ export function runnerSide( port: LinkPort, runner: Pick< AgentsRunner, - "members" | "running" | "interrupted" | "subscribe" - >, + "members" | "running" | "interrupted" | "subscribe" | "subscribeRecords" + > & + Readonly<{ records: Pick }>, ) { let from: string | null = null; let revision = 0; @@ -118,6 +122,20 @@ export function runnerSide( }; }, }), + "runner.records": () => ({ + get: () => runner.records.all() as unknown as Json, + subscribe: (listener) => runner.subscribeRecords(listener), + }), + }, + commands: { + "records.edit": async (args) => { + const { key, edit } = args as unknown as Readonly<{ + key: AgentKey; + edit: RecordEdit; + }>; + await runner.records.edit(key, edit); + return null; + }, }, }); side.watch("manager.members", null, () => { @@ -126,7 +144,11 @@ export function runnerSide( | undefined; if (!members) return; try { - runner.members(members.identities, members.viewer ?? undefined); + runner.members( + members.identities, + members.viewer ?? undefined, + members.loaded, + ); } catch (error) { console.error("The Runner could not decide what runs", error); } @@ -140,13 +162,22 @@ export function runnerSide( } /** The Manager's side: it sends the Runner the viewer's memberships, and - * shows what the Runner decided. An agent keeps its identity until the Runner - * sends a change to it. */ + * shows what the Runner decided and the records it keeps. An agent or record + * keeps its identity until the Runner sends a change to it. */ export function managerSide( port: LinkPort, - members: () => Omit, - changed: () => void, + host: Readonly<{ + members(): Omit; + /** The settings a new agent of this type starts with, if its type is + * here. */ + defaults( + identity: Pick, + ): AgentRecord | undefined; + /** What the Runner decided, or a record, changed. */ + changed(): void; + }>, ) { + const { members, changed } = host; const from = crypto.randomUUID(); let revision = 0; let closed = false; @@ -163,6 +194,10 @@ export function managerSide( }; const waits = new Set>(); const shown = new Map(); + let records: Readonly> = {}; + const recordJson = new Map(); + // A new agent's defaults, until the Runner sends a record for it. + const filled = new Map(); const side = link(port, { topics: { "manager.members": () => ({ @@ -214,6 +249,35 @@ export function managerSide( } changed(); }); + side.watch("runner.records", null, () => { + const all = side.read("runner.records") as unknown as + | readonly AgentRecord[] + | undefined; + if (!all) return; + const next: Record = {}; + for (const record of all) { + const json = JSON.stringify(record); + const prior = records[record.pubkey]; + next[record.pubkey] = + prior && recordJson.get(record.pubkey) === json ? prior : record; + recordJson.set(record.pubkey, json); + filled.delete(record.pubkey); + } + for (const key of recordJson.keys()) if (!next[key]) recordJson.delete(key); + records = next; + changed(); + }); + /** The agent's home record, which keeps its config: as the Runner keeps it + * or, without one, its type's defaults. */ + const settingsOf = (identity: Pick) => { + const { pubkey } = identity; + const kept = records[pubkey] ?? filled.get(pubkey); + if (kept) return kept; + const record = host.defaults(identity); + if (!record) return blank(pubkey); + filled.set(pubkey, record); + return record; + }; return { /** The memberships may have changed: sends them if they did. */ changed() { @@ -234,6 +298,27 @@ export function managerSide( interrupted: () => state.interrupted, /** A running agent, in any community. */ agent: (key: AgentKey) => state.running.find((agent) => agent.key === key), + /** The records the Runner keeps, as it last sent them. */ + records: { + keys: () => Object.keys(records), + settingsOf, + /** The membership's settings as its owner and agent see them: its own + * attention, starting with none, and the agent's config. */ + recordOf(member: Member): AgentRecord { + const settings = settingsOf(member); + if (member.key === member.pubkey) return settings; + const own = records[member.key] ?? blank(member.key); + return { ...own, config: settings.config }; + }, + /** Asks the Runner to apply `edit`; settles once it is saved, and + * shown here. */ + async edit(key: AgentKey, edit: RecordEdit) { + await side.call("records.edit", { + key, + edit, + } as unknown as Json); + }, + }, close() { closed = true; side.close(); diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 3836190d7..451fcaadb 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -9524,8 +9524,10 @@ it.each(["abort", "community"])( } finally { release(); } + // The save answers over the Runner link, so the refusal can come first. + const refused = expect(pending).rejects.toThrow(); await ownerSave; - await expect(pending).rejects.toThrow(); + await refused; expect(service.find(bot)?.attention["interest/revoked"]).toBeUndefined(); }, ); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 27af5a00f..849525ccb 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -1267,6 +1267,8 @@ export class AgentsService extends Service implements Agents { private readonly decided: ReturnType; /** Counts list requests, so only the latest one is applied. */ private listed = 0; + /** Native's agent list was read, so `identities` is every membership. */ + private loaded = false; /** Each running cleanup of a deleted membership, by key, how to stop it, * and whether it finished. */ private cleaning = new Map< @@ -1484,14 +1486,23 @@ export class AgentsService extends Service implements Agents { // will between pages. const channel = new MessageChannel(); this.runnerLink = runnerSide(channel.port1, this.runner); - this.decided = managerSide( - channel.port2, - () => ({ + this.decided = managerSide(channel.port2, { + members: () => ({ + loaded: this.loaded, identities: this.identities, viewer: this.communities.snapshot().viewer ?? null, }), - () => this.update(), - ); + defaults: ({ pubkey, type }) => { + const kind = this.types().find((entry) => entry.key === type); + if (kind) + try { + return defaults(pubkey, kind); + } catch (error) { + console.warn(`Agent type ${type} has invalid defaults`, error); + } + }, + changed: () => this.update(), + }); this.unfinished = Object.freeze(readUnfinished(storage)); this.unarchiving = readArchiveUnchecked(storage); this.teamList = Object.freeze(readTeams(storage)); @@ -2238,7 +2249,7 @@ export class AgentsService extends Service implements Agents { record = { ...record, config: made.config }; if (carry && kind.adopt) record = { ...record, config: kind.adopt(carry, record.config) }; - await this.runner.records.edit(identity.pubkey, { + await this.decided.records.edit(identity.pubkey, { kind: "create", record, }); @@ -2294,7 +2305,7 @@ export class AgentsService extends Service implements Agents { const name = change.name?.trim(); if (name === "") throw new Error("Name the agent"); const { name: _, picture: __, ...settings } = change; - await this.runner.records.edit(key, { + await this.decided.records.edit(key, { kind: "settings", member: identity, change: settings, @@ -2456,7 +2467,7 @@ export class AgentsService extends Service implements Agents { await this.runner.records.change( pubkey, (saved) => ({ - ...(saved ?? this.runner.records.settingsOf(identity)), + ...(saved ?? this.decided.records.settingsOf(identity)), config: change ? change.config : adopted, }), // Shown with its new type, only once its config is saved. @@ -2493,7 +2504,7 @@ export class AgentsService extends Service implements Agents { throw new Error("A hosted provider cannot take over a local agent key"); if (type === identity.type) return { - config: this.runner.records.settingsOf(identity).config, + config: this.decided.records.settingsOf(identity).config, moved: [], dropped: [], settings: [], @@ -2647,7 +2658,7 @@ export class AgentsService extends Service implements Agents { if (shown.has(pubkey) && member.key !== pubkey) continue; shown.set( pubkey, - this.view(member, this.runner.records.recordOf(member)), + this.view(member, this.decided.records.recordOf(member)), ); } return [...shown.values()]; @@ -2748,7 +2759,7 @@ export class AgentsService extends Service implements Agents { ); // Its home record keeps the agent's settings, but not what it paid // attention to there. - if (key === pubkey) await this.runner.records.edit(key, { kind: "leave" }); + if (key === pubkey) await this.decided.records.edit(key, { kind: "leave" }); await this.refresh(); } @@ -3141,7 +3152,7 @@ export class AgentsService extends Service implements Agents { source: RegisteredAgentType | undefined, kind: RegisteredAgentType, ) { - const record = this.runner.records.settingsOf(identity); + const record = this.decided.records.settingsOf(identity); const carry = source?.carry?.(record.config) ?? {}; const { config } = kind.defaults(); const adopted = kind.adopt ? kind.adopt(carry, config) : config; @@ -3248,20 +3259,12 @@ export class AgentsService extends Service implements Agents { // A read that succeeds ends the failure the last one reported. const { error: _, ...state } = this.state; this.state = Object.freeze({ ...state, status: "ready" }); - // Settings whose agent is gone, or that a community it left kept, do - // nothing. Its home record keeps its config while it is anywhere. + this.loaded = true; const live = new Set( identities .filter((saved) => !saved.deleted) .flatMap((saved) => [saved.key, saved.pubkey]), ); - const gone = this.runner.records.keys().filter((key) => !live.has(key)); - for (const key of gone) - void this.runner.records - .edit(key, { kind: "remove" }) - .catch((error) => - console.warn(`Agent ${key} settings were not removed`, error), - ); // Nor do change requests, which would hold review slots unseen. this.updates = this.updates.filter((request) => live.has(request.agent)); // A deleted agent leaves its teams; the teams stay. @@ -3490,7 +3493,7 @@ export class AgentsService extends Service implements Agents { const alive: Agent[] = []; for (const identity of this.identities) { if (identity.deleted || !this.alive(identity)) continue; - const record = this.runner.records.recordOf(identity); + const record = this.decided.records.recordOf(identity); // Reuse the frozen agent while nothing visible changed, so subscribers see // a stable snapshot. const fits = (prior: Agent | undefined) =>