diff --git a/src/features/agents/port.ts b/src/features/agents/port.ts index 33523ec5d4..e9c2434c2a 100644 --- a/src/features/agents/port.ts +++ b/src/features/agents/port.ts @@ -1,13 +1,16 @@ // The agents service and the screens that use it can run in different pages. // The service answers over a MessagePort; the client holds what screens read // synchronously, as the service last pushed it, and sends everything else. -import type { - Agent, - AgentCreationRequest, - Agents, - AgentsClient, - AgentsSnapshot, - AgentUpdateRequest, +import { + type Agent, + type AgentCreationRequest, + type Agents, + type AgentsClient, + type AgentsSnapshot, + type AgentUpdateRequest, + retypePreview, + type SessionPlace, + type SessionState, } from "./service"; import { serveTypes, @@ -59,6 +62,8 @@ type State = Readonly<{ creationRequests: readonly AgentCreationRequest[]; updateRequests: readonly AgentUpdateRequest[]; types: TypesState; + /** Each session state a screen asked for, by `placeKey`. */ + sessions: Readonly>; }>; type ToClient = | Readonly<{ kind: "state"; state: State }> @@ -69,18 +74,22 @@ type ToService = | Readonly<{ kind: "send"; method: Send; args: unknown[] }> | Readonly<{ kind: "type"; id: number; call: TypeCall; args: unknown[] }> | Readonly<{ kind: "watch"; watch: TypeWatch }> + | Readonly<{ kind: "session"; key: string; place: SessionPlace }> | Readonly<{ kind: "abort"; signal: number }>; -/** What the client carries of `AgentsClient` so far, and `present` for what - * this page shows of each type (by type key). Session state and the retype - * preview read service state that a later change pushes. */ -export type AgentsPortClient = Omit< - AgentsClient, - "sessionState" | "previewRetype" -> & +/** `AgentsClient`, and `present` for what this page shows of each type (by + * type key). */ +export type AgentsPortClient = AgentsClient & Readonly<{ present: ReturnType["present"] }>; -function stateOf(agents: Agents, types: TypesState): State { +const placeKey = (key: string, { channelId, root }: SessionPlace) => + JSON.stringify([key, channelId, root ?? ""]); + +function stateOf( + agents: Agents, + types: TypesState, + places: ReadonlyMap, +): State { const snapshot = agents.snapshot(); const memberships: Record> = {}; const joinable = agents.joinable(); @@ -107,6 +116,12 @@ function stateOf(agents: Agents, types: TypesState): State { creationRequests: agents.creationRequests(), updateRequests: agents.updateRequests(), types, + sessions: Object.fromEntries( + [...places].map(([id, [key, place]]) => [ + id, + agents.sessionState(key, place), + ]), + ), }; } @@ -114,6 +129,7 @@ function stateOf(agents: Agents, types: TypesState): State { * each answer, so a call's result never arrives before what it changed. */ export function serveAgents(agents: Agents, port: MessagePort) { const types = serveTypes(agents); + const places = new Map(); // The types' own stores, which change without the service's notice. const stores = new Map<(listener: () => void) => () => void, () => void>(); let pending = true; @@ -130,7 +146,7 @@ export function serveAgents(agents: Agents, port: MessagePort) { if (!stores.has(subscribe)) stores.set(subscribe, subscribe(changed)); port.postMessage({ kind: "state", - state: stateOf(agents, types.state()), + state: stateOf(agents, types.state(), places), } satisfies ToClient); }; // Several changes in one task are one push. @@ -152,6 +168,11 @@ export function serveAgents(agents: Agents, port: MessagePort) { changed(); return; } + if (data.kind === "session") { + places.set(placeKey(data.key, data.place), [data.key, data.place]); + changed(); + return; + } if (data.kind === "abort") { types.abort(data.signal); return; @@ -212,6 +233,7 @@ const STARTING: State = Object.freeze({ values: Object.freeze({}), scopes: Object.freeze({}), }), + sessions: Object.freeze({}), }); /** `next`, keeping each part of `prior` it equals, so screens that compare by @@ -266,6 +288,7 @@ export function connectAgents(port: MessagePort) { let next = 0; let closed = false; const listeners = new Set<() => void>(); + const asked = new Set(); const waiting = new Map< number, Readonly<{ resolve(value: unknown): void; reject(error: Error): void }> @@ -363,6 +386,33 @@ export function connectAgents(port: MessagePort) { interruptible: (pubkey: string) => state.interruptible.includes(pubkey), creationRequests: () => state.creationRequests, updateRequests: () => state.updateRequests, + // A place reads as unavailable until the service has pushed it. + sessionState: (key: string, place: SessionPlace) => { + const id = placeKey(key, place); + if (!asked.has(id) && !closed) { + asked.add(id); + post({ kind: "session", key, place }); + } + return state.sessions[id] ?? "unavailable"; + }, + // The service's preview, from the agent's config and the types' config + // functions, which both pages have. + previewRetype: (key: string, type: string) => { + const agent = state.snapshot.agents.find((item) => item.key === key); + if (!agent) throw new Error("No such agent on this device"); + const kinds = types.types(); + const kind = kinds.find((item) => item.key === type); + if (!kind) throw new Error("That agent type is not available"); + if (kind.remote) + throw new Error("A hosted provider cannot take over a local agent key"); + if (type === agent.type) + return { config: agent.config, moved: [], dropped: [], settings: [] }; + return retypePreview( + agent.config, + kinds.find((item) => item.key === agent.type), + kind, + ); + }, } as AgentsPortClient; return Object.freeze({ ...client, diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index f85d8eb0c6..c772a30b1c 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -105,7 +105,13 @@ async function setup({ jev = undefined as FakeJev | undefined, binding = undefined as ((agent: Agent) => ConfigBinding) | undefined, sessionControls = undefined as - | Pick["sessionControls"]>, "state" | "run"> + | (Pick< + NonNullable["sessionControls"]>, + "state" | "run" + > & + Partial< + Pick["sessionControls"]>, "subscribe"> + >) | undefined, resumes = false, /** How `run` answers, from the first delivery on. */ @@ -1857,6 +1863,44 @@ it("drops its events while another copy of the app holds the agent, and runs onc await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(2)); }); +it("serves a session's state over the port once a screen asks for it", async () => { + let state: "idle" | "compacting" = "idle"; + const listeners = new Set<() => void>(); + const { service } = await setup({ + sessionControls: { + state: () => state, + run: async () => {}, + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }, + }); + await service.create({ type: "example/echo", name: "Echo" }); + await vi.waitFor(() => + expect(service.sessionState(bot, { channelId: channel })).toBe("idle"), + ); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(service, port1); + const client = connectAgents(port2); + try { + const place = { channelId: channel }; + expect(client.sessionState(bot, place)).toBe("unavailable"); + await vi.waitFor(() => + expect(client.sessionState(bot, place)).toBe("idle"), + ); + // The harness's own change notice reaches the screens' page. + state = "compacting"; + for (const listener of listeners) listener(); + await vi.waitFor(() => + expect(client.sessionState(bot, place)).toBe("compacting"), + ); + } finally { + client.close(); + stop(); + } +}); + it.each([false, new Error("Custody unavailable")])( "rechecks native custody before an immediate session mutation: %s", async (denied) => { @@ -2762,6 +2806,35 @@ it("changes an agent's type between turns, carrying what the new type adopts", a { id: "sessions", label: "Sessions", outcome: "reset" }, ], }); + // Screens in another page compute the same preview from the same config + // functions, which each type's UI entry presents there too. + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(service, port1); + const client = connectAgents(port2); + try { + for (const type of service.types()) + client.present(type.key, () => { + const { carry, adopt, describeMigration } = type; + return { + ...(carry && { carry }), + ...(adopt && { adopt }), + ...(describeMigration && { describeMigration }), + }; + }); + await vi.waitFor(() => expect(client.types()).toHaveLength(2)); + expect(client.previewRetype(bot, "other/relay")).toEqual( + service.previewRetype(bot, "other/relay"), + ); + expect(client.previewRetype(bot, "example/echo")).toEqual( + service.previewRetype(bot, "example/echo"), + ); + expect(() => client.previewRetype(bot, "other/missing")).toThrow( + "That agent type is not available", + ); + } finally { + client.close(); + stop(); + } expect(native.setType).not.toHaveBeenCalled(); await expect(service.retype(bot, "other/relay")).rejects.toThrow( "Echo is answering",