diff --git a/docs/agents.md b/docs/agents.md index 1e97bfefa2..32f03e4dad 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -80,8 +80,8 @@ servers do not support these operations. In communities the build flag `BUZZ_OWNER_MCP_COMMUNITIES` enables, an agent's runs in this app on a bundled harness list `owner_channel_list`, `owner_channel_get`, -`owner_member_list`, `owner_read`, `owner_canvas_get` and `owner_canvas_history`. -They read the owner's channels, one channel's details and +`owner_member_list`, `owner_read`, `owner_canvas_get` and `owner_canvas_history`, +and the channel writes below. The reads read the owner's channels, one channel's details and the owner's rights in it, its roster, its messages, whole threads and searches, and its Canvas and earlier revisions, through the owner's own signed-in session, including private and archived channels, DMs and @@ -91,7 +91,7 @@ owner's channels and says `partial` while their list is incomplete. Other channels' matches still fill its page, so `limited` with few results means narrowing it by channel or time. A message shows its newest edit by its author or, as in the app, by the author's verified owner. One read fetches a page's edits; -when they fill it, the answer says `edits_limited` and some messages may show earlier text. They write +when they fill it, the answer says `edits_limited` and some messages may show earlier text. The reads write nothing: no membership, subscription, read state, post, agent start or Canvas change. Content comes from the relay's own answer, not the owner's open views, so it changes no unread counts. They do not save the owner's attachments: a @@ -108,6 +108,26 @@ model can read what they return. Protection reviews these calls and their results as it does other Buzz tools. See [configuration](configuration.md#owner-mcp-community-flag). +`owner_channel_create`, `owner_channel_update`, `owner_channel_archive`, +`owner_channel_unarchive` and `owner_channel_delete` manage the owner's +ordinary channels and Solo chats through the same session and the app's own +channel writers, never DMs or system channels. A created channel or Solo chat +has the owner as Owner and does not add the agent. Create takes a UUID the agent +chooses, and the same ID with other details is a conflict. To resume a create, +the agent repeats it with the `operation` it answered; that call never signs +another create, and answers `unconfirmed` when Buzz no longer holds that +operation and the relay shows no such channel. Update takes the +`version` that `owner_channel_get` answered and saves nothing if the channel has +changed since. Lifecycle tools take an exact channel ID; delete is permanent. +Each answer's `outcome` is `confirmed` (read back from the relay), `rejected` +(nothing changed; `reason` says why, such as `busy`, `conflict` or +`cancelled`) or `unconfirmed` (it may have taken effect; check with +`owner_channel_get` before trying again). A write is admitted only while the +calling turn runs, the agent still runs here as the owner's, and the owner's +account, community and connection are unchanged, checked again just before the +command is signed and sent. A harness that cannot report whether its turn is +still running gets no writes. + ## External app authorization The build flag `BUZZ_EXTERNAL_MCP_COMMUNITIES` lists the allowed community diff --git a/docs/buzz-help.md b/docs/buzz-help.md index 04c91598bf..8d31872c17 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -983,6 +983,12 @@ confirmation. **Delete team** asks first; it doesn't delete the team's agents. instructions for watched turns in both Chat and Agent mode. Watched messages are context for that Interest; later direct requests do not inherit it. Mesh agents respond only to their owner. +- **Owner tools**: in communities a build enables them, an agent running in this + app on a bundled harness can read its owner's channels with the owner's access + and create, edit, archive, unarchive or delete the owner's channels and Solo + chats. The owner stays the channel's Owner and the agent is not added. Each + change answers confirmed, rejected or unconfirmed; unconfirmed may still have + taken effect, so check the channel before asking again. Deleting is permanent. - **Solo workers**: where a plugin offers it, an owner can authorize one agent to start, message, read, stop, and archive the owner's workers in Solo chats. The agent asks the owner's Buzz in a private channel with only the two of them; diff --git a/docs/configuration.md b/docs/configuration.md index c73820f9c6..86c1608749 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -9,7 +9,7 @@ Build defaults are readable binary data, **never secret storage**. | Input | Consumer and boundary | | --- | --- | | `BUZZ_EXTERNAL_MCP_COMMUNITIES` | Frontend build: space-separated secure community origins allowed to authorize external MCP access. Empty disables access. See below. | -| `BUZZ_OWNER_MCP_COMMUNITIES` | Frontend build: space-separated secure community origins where an agent's app-hosted runs can read its owner's channels, members, messages and Canvases with the owner's access. Empty disables them. See below. | +| `BUZZ_OWNER_MCP_COMMUNITIES` | Frontend build: space-separated secure community origins where an agent's app-hosted runs can read its owner's channels, members, messages and Canvases, and create and manage the owner's channels, with the owner's access. Empty disables them. See below. | | `BUZZ_WORKFLOW_RESTRICTED_COMMUNITIES` | Frontend build: space-separated secure community origins where workflow messages cannot start or steer local agents. Empty keeps checked workflow delegation everywhere. See below. | | `BUZZ_BUILDERLAB_URL` | Shared frontend/native build: the live development broker's Builderlab, and whether a build without an added sign-in server keeps the built-in one; no runtime override. See below. | | `BUZZ_BUILDERLAB_APPS_DOMAIN` | Shared frontend/native build: exact Apps cookie domain paired with the login origin for native viewer SSO; no runtime override. | diff --git a/src/bundled/codex/runtime.test.ts b/src/bundled/codex/runtime.test.ts index d8373d36a1..85527f6af2 100644 --- a/src/bundled/codex/runtime.test.ts +++ b/src/bundled/codex/runtime.test.ts @@ -7078,6 +7078,11 @@ it.each([true, false])( "owner_read", "owner_canvas_get", "owner_canvas_history", + "owner_channel_create", + "owner_channel_update", + "owner_channel_archive", + "owner_channel_unarchive", + "owner_channel_delete", ].map((name) => expect.objectContaining({ name, deferLoading: true })) : [], ); diff --git a/src/buzz-mcp/app-client.test.ts b/src/buzz-mcp/app-client.test.ts index 548b401407..f2d3ccf90b 100644 --- a/src/buzz-mcp/app-client.test.ts +++ b/src/buzz-mcp/app-client.test.ts @@ -455,3 +455,70 @@ it("offers the handle's owner reads with the turn's signal, withholding answers expect(() => client.ownerAccess?.held(execution)).toThrow("ended"); expect(ownerAccess.held).toHaveBeenCalledOnce(); }); + +it("gives owner writes the turn's own caller check, and none to a harness without one", async () => { + const memories = async () => ({ entries: [], partial: false }); + const answer = { outcome: "confirmed" } as never; + const ownerAccess = { + createChannel: vi.fn(async (..._: unknown[]) => answer), + updateChannel: vi.fn(async (..._: unknown[]) => answer), + changeChannel: vi.fn(async (..._: unknown[]) => answer), + }; + let ended = false; + let protectedOut = false; + const controller = new AbortController(); + const handle = { + pubkey: "agent", + ownerAccess, + protection: { + assertAvailable: () => { + if (protectedOut) throw new Error("Protection is unavailable"); + }, + }, + } as unknown as AgentHandle; + const client = appClient(handle, { + read: vi.fn(), + memories, + signal: controller.signal, + check: () => { + if (ended) throw new Error("The turn has ended"); + }, + }); + const input = { id: "c", name: "plans", visibility: "open" as const }; + await expect(client.ownerAccess?.createChannel(input)).resolves.toBe(answer); + const [, caller] = ownerAccess.createChannel.mock.calls[0] as unknown as [ + unknown, + { signal?: AbortSignal; check(): void }, + ]; + expect(caller.signal).toBe(controller.signal); + expect(() => caller.check()).not.toThrow(); + // The caller is the turn itself: it fails once the turn, or its protection, + // no longer holds. + protectedOut = true; + expect(() => caller.check()).toThrow("Protection is unavailable"); + protectedOut = false; + ended = true; + expect(() => caller.check()).toThrow("ended"); + await expect( + client.ownerAccess?.changeChannel("archive", "c"), + ).rejects.toThrow("ended"); + expect(ownerAccess.changeChannel).not.toHaveBeenCalled(); + ended = false; + await client.ownerAccess?.updateChannel({ id: "c", version: "v" }); + await client.ownerAccess?.changeChannel("delete", "c"); + expect(ownerAccess.updateChannel.mock.calls[0]?.[0]).toEqual({ + id: "c", + version: "v", + }); + expect(ownerAccess.changeChannel.mock.calls[0]?.slice(0, 2)).toEqual([ + "delete", + "c", + ]); + + // A harness that can't say whether its turn still runs changes nothing. + const blind = appClient(handle, { read: vi.fn(), memories }); + await expect(blind.ownerAccess?.createChannel(input)).rejects.toThrow( + "unavailable in this harness", + ); + expect(ownerAccess.createChannel).toHaveBeenCalledOnce(); +}); diff --git a/src/buzz-mcp/app-client.ts b/src/buzz-mcp/app-client.ts index e400bd6938..6ae26b6307 100644 --- a/src/buzz-mcp/app-client.ts +++ b/src/buzz-mcp/app-client.ts @@ -2,7 +2,7 @@ // reads through its Agents handle, files are read as its plugin reads them, and // memory is read back through the owner's own session, to whom it is // encrypted. -import type { BuzzClient, Memories } from "./client"; +import type { BuzzClient, Memories, OwnerMutationCaller } from "./client"; import { prepareAttachment } from "../features/messages/prepare-attachment"; import type { AgentHandle, AgentHandoff } from "../features/agents/service"; import type { @@ -55,6 +55,24 @@ export function appClient( check(); return result; }; + /** An owner write, admitted only while this turn holds. A host that can't + * say whether its turn still runs gets none. */ + const changing = async ( + write: (caller: OwnerMutationCaller) => Promise, + ) => { + check(); + if (!options.signal && !options.check) + throw new Error( + "Owner changes are unavailable in this harness; nothing was changed.", + ); + return write({ + signal: options.signal, + check: () => { + check(); + agent.protection?.assertAvailable(); + }, + }); + }; return { ...(agent.protection ? { @@ -106,6 +124,12 @@ export function appClient( check(); ownerAccess.held(execution); }, + createChannel: (input) => + changing((caller) => ownerAccess.createChannel(input, caller)), + updateChannel: (input) => + changing((caller) => ownerAccess.updateChannel(input, caller)), + changeChannel: (action, id) => + changing((caller) => ownerAccess.changeChannel(action, id, caller)), }, }), relayAuthor: options.relayAuthor, diff --git a/src/buzz-mcp/client.ts b/src/buzz-mcp/client.ts index 4d96930f86..59aee7acd6 100644 --- a/src/buzz-mcp/client.ts +++ b/src/buzz-mcp/client.ts @@ -123,10 +123,13 @@ export type OwnerChannelDetails = Readonly<{ ttlSeconds?: number; creator?: string; createdAt?: number; + /** The metadata record read; owner_channel_update takes it. */ + version?: string; }>; /** From the relay-signed admin list and roster, read for this call. The * relay still authorizes every command. */ capabilities: Readonly<{ + edit?: boolean; manageRoles: boolean; removeMembers: boolean; archive: boolean; @@ -222,6 +225,71 @@ export type OwnerAccess = Readonly<{ /** Throws unless the binding that answered with `execution` still holds, * as checked just before its answer is released; reads nothing. */ held(execution: OwnerExecution): void; + /** Owner writes. The host supplies `caller`; a call without one changes + * nothing. */ + createChannel( + input: OwnerChannelCreate, + caller?: OwnerMutationCaller, + ): Promise; + updateChannel( + input: OwnerChannelUpdate, + caller?: OwnerMutationCaller, + ): Promise; + changeChannel( + action: OwnerLifecycleAction, + id: string, + caller?: OwnerMutationCaller, + ): Promise; +}>; + +/** The calling turn, as its host checks it: `check` throws once the turn has + * ended or been stopped. Host-supplied; no tool argument names it. */ +export type OwnerMutationCaller = Readonly<{ + signal?: AbortSignal | undefined; + check(): void; +}>; +/** `confirmed`: read back from the relay. `rejected`: refused before it was + * sent, or refused by the relay, so nothing changed. `unconfirmed`: sent, or + * possibly sent, with no answer yet; check before acting again. */ +export type OwnerWriteOutcome = "confirmed" | "rejected" | "unconfirmed"; +export type OwnerLifecycleAction = "archive" | "unarchive" | "delete"; +export type OwnerChannelCreate = Readonly<{ + /** The caller's own UUID for the new channel. */ + id: string; + /** The `operation` an earlier create of `id` answered. Given, the call only + * resumes or confirms that create and never signs another. */ + operation?: string | undefined; + name: string; + visibility: "open" | "private"; + description?: string | undefined; + ttlSeconds?: number | undefined; + /** A private Solo chat whose only member is the owner. */ + solo?: boolean | undefined; +}>; +export type OwnerChannelUpdate = Readonly<{ + id: string; + /** The `version` owner_channel_get answered; a newer one is a conflict. */ + version: string; + name?: string | undefined; + description?: string | undefined; + visibility?: "public" | "private" | undefined; + joinRole?: "member" | "guest" | undefined; + /** `null` makes a temporary channel ongoing. */ + ttlSeconds?: number | null | undefined; +}>; +export type OwnerChannelWrite = Readonly<{ + execution: OwnerExecution; + action: "create" | "update" | OwnerLifecycleAction; + channel: string; + outcome: OwnerWriteOutcome; + /** The signed command's event ID, where the writer exposes it. */ + operation?: string; + reason?: "busy" | "conflict" | "cancelled" | "denied" | "refused" | "failed"; + error?: string; + /** From the roster read back after a confirmed change. */ + ownerRole?: OwnerRole | "unknown" | null; + callerMember?: boolean; + version?: string; }>; export interface BuzzClient { diff --git a/src/buzz-mcp/owner.test.ts b/src/buzz-mcp/owner.test.ts index 74383cebf0..42e75f30e9 100644 --- a/src/buzz-mcp/owner.test.ts +++ b/src/buzz-mcp/owner.test.ts @@ -84,6 +84,24 @@ function owned() { revisions: [message("9", { kind: 40100, content: "# Plan\nShip it" })], })), held: vi.fn(), + createChannel: vi.fn(async (input) => ({ + execution: executedAs, + action: "create", + channel: input.id, + outcome: "confirmed", + })), + updateChannel: vi.fn(async (input) => ({ + execution: executedAs, + action: "update", + channel: input.id, + outcome: "confirmed", + })), + changeChannel: vi.fn(async (action, id) => ({ + execution: executedAs, + action, + channel: id, + outcome: "confirmed", + })), }; } const clientWith = (fields: Partial = {}) => @@ -211,6 +229,199 @@ describe("callOwnerTool", () => { expect(read).not.toHaveBeenCalled(); }); + it.each([ + ["owner_channel_create", { name: "plans", visibility: "open" }, "id must"], + [ + "owner_channel_create", + { id: "not-a-uuid", name: "plans", visibility: "open" }, + "id must", + ], + ["owner_channel_create", { id, name: "plans" }, "visibility must"], + ["owner_channel_create", { id, visibility: "open" }, "name is required"], + [ + "owner_channel_create", + { id, name: "me", solo: true, visibility: "open" }, + "Solo chat is private", + ], + [ + "owner_channel_create", + { id, name: "me", solo: true, description: "x" }, + "no description", + ], + [ + "owner_channel_create", + { id, name: "p", visibility: "open", ttl_seconds: 0 }, + "ttl_seconds", + ], + [ + "owner_channel_create", + { id, name: "p", visibility: "open", operation: "abc" }, + "operation must", + ], + [ + "owner_channel_create", + { id, name: "p", visibility: "open", as: executedAs.owner }, + "cannot be used", + ], + ["owner_channel_update", { channel: id, name: "x" }, "version"], + [ + "owner_channel_update", + { channel: id, version: hex("1") }, + "at least one", + ], + [ + "owner_channel_update", + { channel: id, version: hex("1"), visibility: "public" }, + "visibility must", + ], + [ + "owner_channel_update", + { channel: id, version: hex("1"), join_role: "owner" }, + "join_role must", + ], + [ + "owner_channel_update", + { channel: id, version: hex("1"), name: " " }, + "name must not be empty", + ], + ["owner_channel_archive", {}, "channel is required"], + ["owner_channel_delete", { channel: id, force: true }, "cannot be used"], + ])("rejects %s %j before changing anything", async (name, args, message) => { + const ownerAccess = owned(); + await expect( + callOwnerTool(clientWith({ ownerAccess }), name, args), + ).rejects.toThrow(message); + for (const each of Object.values(ownerAccess)) + expect(each).not.toHaveBeenCalled(); + }); + + it("creates, updates and changes a channel with exactly the given fields, answering who it ran as", async () => { + const ownerAccess = owned(); + const client = clientWith({ ownerAccess }); + const text = async (name: string, args: Record) => + JSON.parse((await callOwnerTool(client, name, args)).text); + expect( + await text("owner_channel_create", { + id, + name: "Plans", + visibility: "private", + description: "Next quarter", + ttl_seconds: 3600, + }), + ).toEqual({ + executedAs, + action: "create", + channel: id, + outcome: "confirmed", + }); + expect(ownerAccess.createChannel).toHaveBeenLastCalledWith({ + id, + name: "Plans", + visibility: "private", + description: "Next quarter", + ttlSeconds: 3600, + }); + await text("owner_channel_create", { id, name: "Me", solo: true }); + expect(ownerAccess.createChannel).toHaveBeenLastCalledWith({ + id, + name: "Me", + visibility: "private", + solo: true, + }); + await text("owner_channel_create", { + id, + name: "Plans", + visibility: "open", + operation: hex("2"), + }); + expect(ownerAccess.createChannel).toHaveBeenLastCalledWith({ + id, + operation: hex("2"), + name: "Plans", + visibility: "open", + }); + await text("owner_channel_update", { + channel: id, + version: hex("1"), + visibility: "open", + join_role: "guest", + ttl_seconds: null, + }); + expect(ownerAccess.updateChannel).toHaveBeenLastCalledWith({ + id, + version: hex("1"), + visibility: "public", + joinRole: "guest", + ttlSeconds: null, + }); + await text("owner_channel_update", { + channel: id, + version: hex("1"), + description: "", + }); + expect(ownerAccess.updateChannel).toHaveBeenLastCalledWith({ + id, + version: hex("1"), + description: "", + }); + for (const action of ["archive", "unarchive", "delete"] as const) + expect( + await text(`owner_channel_${action}`, { channel: id }), + ).toMatchObject({ action, channel: id, outcome: "confirmed" }); + expect(ownerAccess.changeChannel.mock.calls).toEqual([ + ["archive", id], + ["unarchive", id], + ["delete", id], + ]); + ownerAccess.changeChannel.mockResolvedValueOnce({ + execution: executedAs, + action: "delete", + channel: id, + outcome: "rejected", + reason: "busy", + error: "Another channel action is still in progress", + }); + expect(await text("owner_channel_delete", { channel: id })).toMatchObject({ + outcome: "rejected", + reason: "busy", + }); + }); + + it("says what a write did when its binding changed before the answer was released", async () => { + const ownerAccess = owned(); + ownerAccess.held.mockImplementation(() => { + throw new Error("changed"); + }); + ownerAccess.createChannel.mockResolvedValueOnce({ + execution: executedAs, + action: "create", + channel: id, + outcome: "unconfirmed", + operation: hex("e"), + }); + const sent = await callOwnerTool( + clientWith({ ownerAccess }), + "owner_channel_create", + { id, name: "plans", visibility: "open" }, + ); + expect(() => sent.held()).toThrow( + `owner_channel_create on ${id} (operation ${hex("e")}) may have taken effect`, + ); + ownerAccess.changeChannel.mockResolvedValueOnce({ + execution: executedAs, + action: "archive", + channel: id, + outcome: "rejected", + reason: "cancelled", + }); + const refused = await callOwnerTool( + clientWith({ ownerAccess }), + "owner_channel_archive", + { channel: id }, + ); + expect(() => refused.held()).toThrow("changed nothing"); + }); + it("reads nothing for a client without owner access", async () => { for (const { name } of OWNER_TOOLS) await expect( @@ -447,22 +658,35 @@ describe("callOwnerTool", () => { }); }); - it("names only the read tools", () => { - expect(OWNER_TOOLS.map(({ name }) => name)).toEqual([ + it("names the read tools and the channel writes, marking only delete and update destructive", () => { + const reads = [ "owner_channel_list", "owner_channel_get", "owner_member_list", "owner_read", "owner_canvas_get", "owner_canvas_history", + ]; + expect(OWNER_TOOLS.map(({ name }) => name)).toEqual([ + ...reads, + "owner_channel_create", + "owner_channel_update", + "owner_channel_archive", + "owner_channel_unarchive", + "owner_channel_delete", ]); expect( OWNER_TOOLS.every( (tool) => - tool.annotations?.readOnlyHint === true && + tool.annotations?.readOnlyHint === reads.includes(tool.name) && tool.inputSchema.additionalProperties === false, ), ).toBe(true); + expect( + OWNER_TOOLS.filter((tool) => tool.annotations?.destructiveHint).map( + ({ name }) => name, + ), + ).toEqual(["owner_channel_update", "owner_channel_delete"]); expect(isOwnerTool("owner_channel_list")).toBe(true); expect(isOwnerTool("owner_send")).toBe(false); }); diff --git a/src/buzz-mcp/owner.ts b/src/buzz-mcp/owner.ts index 7035c3caa0..ebbd264767 100644 --- a/src/buzz-mcp/owner.ts +++ b/src/buzz-mcp/owner.ts @@ -1,12 +1,15 @@ -// The owner tools (`owner_*`): read with the signed-in human's access instead -// of the agent's, only where the host gives the client `ownerAccess`. The host -// decides who the owner, agent and community are; no argument names them. +// The owner tools (`owner_*`): read and manage channels with the signed-in +// human's access instead of the agent's, only where the host gives the client +// `ownerAccess`. The host decides who the owner, agent and community are; no +// argument names them. import { CANVAS_HISTORY_PAGE } from "../features/channel-templates/canvas-limits"; import { parseBuzzLink } from "../features/navigation/buzz-links"; import type { BuzzClient, OwnerAccess, + OwnerChannelWrite, OwnerExecution, + OwnerLifecycleAction, OwnerReadRequest, } from "./client"; import { @@ -57,6 +60,31 @@ const tool = ( }, annotations: { readOnlyHint: true }, }); +const write = ( + name: string, + description: string, + properties: Readonly>, + required: readonly string[], + destructiveHint: boolean, +): Tool => ({ + ...tool(name, description, properties, required), + annotations: { readOnlyHint: false, destructiveHint }, +}); +const OUTCOME = + " outcome confirmed means read back; rejected, that nothing changed (see reason); unconfirmed, that it may have taken effect: check with owner_channel_get, never repeat it blindly."; +const exact = { + type: "string", + description: + "The channel's exact ID, from owner_channel_list or owner_channel_get.", +} as const; +const lifecycle = (action: OwnerLifecycleAction, description: string) => + write( + `owner_channel_${action}`, + `${description}${OUTCOME}`, + { channel: exact }, + ["channel"], + action === "delete", + ); export const OWNER_TOOLS: readonly Tool[] = [ tool( @@ -128,6 +156,65 @@ export const OWNER_TOOLS: readonly Tool[] = [ { channel, before, before_id: beforeId }, ["channel"], ), + write( + "owner_channel_create", + `Create a channel as your owner, who owns it; you are not added. Choose a new UUID as id. To resume a create that was not confirmed, call again with the same id, details and the operation it answered: that call never signs another create, and answers unconfirmed if Buzz can no longer tell.${OUTCOME}`, + { + id: { type: "string", description: "A new UUID you choose." }, + operation: { + type: "string", + description: "The operation an earlier create of this id answered.", + }, + name: { type: "string" }, + visibility: { + type: "string", + enum: ["open", "private"], + description: "Required unless solo.", + }, + description: { type: "string" }, + ttl_seconds: { + type: "integer", + description: "A temporary channel's lifetime; omit for ongoing.", + }, + solo: { + type: "boolean", + description: "A private Solo chat with only your owner in it.", + }, + }, + ["id", "name"], + false, + ), + write( + "owner_channel_update", + `Change a channel's name, description, visibility, join role or lifetime as your owner; other details stay. Give the version owner_channel_get answered: if the channel changed since, nothing is saved.${OUTCOME}`, + { + channel: exact, + version: { type: "string", description: "From owner_channel_get." }, + name: { type: "string" }, + description: { type: "string" }, + visibility: { type: "string", enum: ["open", "private"] }, + join_role: { + type: "string", + enum: ["member", "guest"], + description: "The role people who join it get.", + }, + ttl_seconds: { + type: ["integer", "null"], + description: "A temporary channel's lifetime; null makes it ongoing.", + }, + }, + ["channel", "version"], + true, + ), + lifecycle( + "archive", + "Archive one of your owner's channels: it stays readable, takes no new messages and can be unarchived.", + ), + lifecycle("unarchive", "Unarchive one of your owner's archived channels."), + lifecycle( + "delete", + "Delete one of your owner's channels and everything in it, permanently. Only for a channel your owner asked to delete, confirmed by its ID with owner_channel_get; archive one otherwise.", + ), ]; export const isOwnerTool = (name: string) => @@ -145,7 +232,7 @@ export async function callOwnerTool( const { ownerAccess } = client; if (!ownerAccess) throw new Error( - "Owner tools are not available to this agent here; nothing was read.", + "Owner tools are not available to this agent here; nothing was read or changed.", ); const extra = Object.keys(args).filter( (arg) => !Object.hasOwn(tool.inputSchema.properties, arg), @@ -169,9 +256,21 @@ export async function callOwnerTool( ...(list.error ? { error: list.error } : {}), }); } + if (name === "owner_channel_create") + return written(ownerAccess, await ownerAccess.createChannel(created(args))); if (name === "owner_read") return read(ownerAccess, args); const id = optionalText(args.channel, "channel"); if (!id) throw new Error("channel is required."); + if (name === "owner_channel_update") + return written( + ownerAccess, + await ownerAccess.updateChannel(updated(id, args)), + ); + if (name.startsWith("owner_channel_")) { + const action = name.slice("owner_channel_".length); + if (action === "archive" || action === "unarchive" || action === "delete") + return written(ownerAccess, await ownerAccess.changeChannel(action, id)); + } if (name === "owner_canvas_get") { const revision = eventId(args.id, "id"); const { execution, channel, canvas } = await ownerAccess.canvas( @@ -351,6 +450,121 @@ function answer( }; } +/** A write's answer. Released whatever its binding does next, as it reports + * what was done; past a change it says only that, never readback. */ +function written(ownerAccess: OwnerAccess, result: OwnerChannelWrite) { + const { execution, ...fields } = result; + return { + text: JSON.stringify({ executedAs: execution, ...fields }), + held: () => { + try { + ownerAccess.held(execution); + } catch { + throw new Error( + `Your owner's connection changed before this answer was returned: owner_channel_${result.action} on ${result.channel}${result.operation ? ` (operation ${result.operation})` : ""} ${result.outcome === "rejected" ? "changed nothing" : "may have taken effect"}. Check it with owner_channel_get before acting on it again.`, + ); + } + }, + }; +} + +function created(args: Readonly>) { + const solo = args.solo; + if (solo !== undefined && typeof solo !== "boolean") + throw new Error("solo must be true or false."); + const visibility = args.visibility; + if (solo) { + if (visibility !== undefined && visibility !== "private") + throw new Error("A Solo chat is private."); + } else if (visibility !== "open" && visibility !== "private") + throw new Error("visibility must be open or private."); + const name = optionalText(args.name, "name"); + if (!name) throw new Error("name is required."); + const description = optionalText(args.description, "description"); + const ttlSeconds = lifetime(args.ttl_seconds); + if (solo && description) throw new Error("A Solo chat has no description."); + return { + id: uuid(optionalText(args.id, "id"), "id"), + ...(args.operation !== undefined + ? { operation: eventId(args.operation, "operation") } + : {}), + name, + visibility: solo + ? ("private" as const) + : (visibility as "open" | "private"), + ...(description ? { description } : {}), + ...(typeof ttlSeconds === "number" ? { ttlSeconds } : {}), + ...(solo ? { solo } : {}), + }; +} + +function updated(id: string, args: Readonly>) { + const version = eventId(args.version, "version"); + if (!version) throw new Error("version is required, from owner_channel_get."); + const visibility = args.visibility; + if ( + visibility !== undefined && + visibility !== "open" && + visibility !== "private" + ) + throw new Error("visibility must be open or private."); + const joinRole = args.join_role; + if (joinRole !== undefined && joinRole !== "member" && joinRole !== "guest") + throw new Error("join_role must be member or guest."); + if (args.name !== undefined && !optionalText(args.name, "name")) + throw new Error("name must not be empty."); + if (args.description !== undefined && typeof args.description !== "string") + throw new Error("description must be text."); + const fields = [ + "name", + "description", + "visibility", + "join_role", + "ttl_seconds", + ]; + if (!fields.some((field) => args[field] !== undefined)) + throw new Error(`Give at least one of ${fields.join(", ")} to change.`); + const ttlSeconds = lifetime(args.ttl_seconds, true); + return { + id, + version, + ...(args.name !== undefined ? { name: args.name as string } : {}), + ...(args.description !== undefined + ? { description: args.description as string } + : {}), + ...(visibility + ? { + visibility: + visibility === "open" ? ("public" as const) : ("private" as const), + } + : {}), + ...(joinRole ? { joinRole: joinRole as "member" | "guest" } : {}), + ...(ttlSeconds !== undefined ? { ttlSeconds } : {}), + }; +} + +function lifetime(value: unknown, clearable = false) { + if (value === undefined || (clearable && value === null)) + return value as undefined | null; + if ( + typeof value !== "number" || + !Number.isInteger(value) || + value <= 0 || + value > 2_147_483_647 + ) + throw new Error( + "ttl_seconds must be a whole number of seconds, 1 or more.", + ); + return value; +} + +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; +function uuid(value: string | undefined, name: string): string { + if (!value || !UUID.test(value)) + throw new Error(`${name} must be a lowercase UUID.`); + return value; +} + /** A `before` and `before_id` cursor. */ function pageCursor(args: Readonly>) { const before = unixTime(args, "before"); diff --git a/src/buzz-mcp/session-tools.ts b/src/buzz-mcp/session-tools.ts index 0f25cb9512..1e0d92fe13 100644 --- a/src/buzz-mcp/session-tools.ts +++ b/src/buzz-mcp/session-tools.ts @@ -109,7 +109,7 @@ export function sessionHint(added: readonly AddedTool[] = [], owner = false) { ? ` and tools plugins add (${added.map((tool) => tool.name).join(", ")})` : ""; const owned = owner - ? ", your owner's channels, members, messages and canvases read with their access (owner_channel_list, owner_read)" + ? ", your owner's channels, members, messages and canvases read with their access (owner_channel_list, owner_read), and channels created and managed as them (owner_channel_create)" : ""; return `Read and act in Buzz as this agent. Search for the rest when you need them: your attention (Interests, event watches and timers; interest_set and watch_add say how they work), Nexus workers an owner lets you run (solo_worker_connections), this device's plugins (plugin_list; install, update, enable and remove are requests your owner confirms)${owned}${plugins}.`; } diff --git a/src/features/agents/owner-access.ts b/src/features/agents/owner-access.ts index a5d2304f5c..b6419b8048 100644 --- a/src/features/agents/owner-access.ts +++ b/src/features/agents/owner-access.ts @@ -1,13 +1,14 @@ -// The owner tools' host side: read-only operations the app runs with its -// signed-in human's own session, for that human's agent, in the agent's own -// community. Every call resolves the binding again and checks it once more -// before returning anything, so a sign-out, account switch or reconnect during -// a call returns nothing. +// The owner tools' host side: operations the app runs with its signed-in +// human's own session, for that human's agent, in the agent's own community. +// Every call resolves the binding again and checks it once more before +// returning anything, so a sign-out, account switch or reconnect during a read +// returns nothing. Writes are admitted by `ownerMutationActive`. import type { BuzzEvent, OwnerAccess, OwnerChannel, OwnerExecution, + OwnerMutationCaller, OwnerReadRequest, OwnerRole, } from "../../buzz-mcp/client"; @@ -26,6 +27,8 @@ import { } from "../relay/channel-lifecycle-protocol"; import { soloLabels } from "../solo/labels"; import type { RelaySession } from "../relay/session"; +import { detailsSettings } from "../relay/channel-details-protocol"; +import { ownerChannelWrites } from "./owner-channels"; /** Whether this build lets agents read with their owner's access in * `community`: its exact canonical origin is in the build's list. */ @@ -49,6 +52,9 @@ export type OwnerSession = Pick< | "profiles" | "channelLifecycle" | "snapshotRead" + | "channelWrites" + | "channelDetails" + | "outbox" >; /** The owner's live session for the agent's community, as the host binds it * now. `signal` aborts when that binding is replaced. */ @@ -59,6 +65,37 @@ export type OwnerBinding = Readonly<{ signal: AbortSignal; }>; +/** An owner write's admission: true only while the calling turn, the binding + * the write was admitted on, the agent's lifetime on this host and its + * recorded owner all still hold. Synchronous and side-effect free, so a writer + * can ask it right before signing and again right before publishing. */ +export function ownerMutationActive( + owner: string, + binding: OwnerBinding, + bind: () => OwnerBinding | undefined, + caller: OwnerMutationCaller, + lifetime?: AbortSignal, +): () => boolean { + return () => { + try { + caller.signal?.throwIfAborted(); + caller.check(); + if (binding.signal.aborted || lifetime?.aborted) return false; + const now = bind(); + return ( + !!now && + !now.signal.aborted && + now.session === binding.session && + now.viewer === binding.viewer && + now.origin === binding.origin && + now.viewer === owner + ); + } catch { + return false; + } + }; +} + /** One edit read serves a whole page; the relay reader's cap. */ const EDITS = 500; @@ -75,11 +112,13 @@ const changed = () => "Your owner's connection changed during the call, so nothing was returned. Try again.", ); -/** `agent`'s owner reads. `bind` answers the owner's current binding for the - * agent's community, or undefined when it may not be used now. */ +/** `agent`'s owner reads and writes. `bind` answers the owner's current + * binding for the agent's community, or undefined when it may not be used now; + * `lifetime` ends with the agent's run authority on this host. */ export function createOwnerAccess( agent: Readonly<{ pubkey: string; owner: string }>, bind: () => OwnerBinding | undefined, + lifetime?: AbortSignal, ): OwnerAccess { /** The binding each answer was read with. */ const answered = new WeakMap(); @@ -132,15 +171,14 @@ export function createOwnerAccess( return execution; }; /** The relay-signed metadata, admin list and roster, read fresh. */ - const authority = async ( + const records = async ( binding: OwnerBinding, id: string, signal: AbortSignal, ) => { - const { session, viewer } = binding; - const relayAuthor = session.relayAuthor; + const relayAuthor = binding.session.relayAuthor; if (!relayAuthor) throw unavailable(); - const events = await session.readRemote( + const events = await binding.session.readRemote( [39000, 39001, 39002].map((kind) => ({ kinds: [kind], consistency: "strong" as const, @@ -151,6 +189,18 @@ export function createOwnerAccess( { signal, priority: "foreground" }, ); signal.throwIfAborted(); + return events; + }; + /** The same records, for a channel that must exist and not be a system one. */ + const authority = async ( + binding: OwnerBinding, + id: string, + signal: AbortSignal, + ) => { + const { session, viewer } = binding; + const relayAuthor = session.relayAuthor; + if (!relayAuthor) throw unavailable(); + const events = await records(binding, id, signal); if (!events.some((event) => event.kind === 39000)) throw new Error( `Channel ${id} was not found or your owner cannot see it.`, @@ -281,7 +331,39 @@ export function createOwnerAccess( }; }; + /** A write's binding, admission and signal, for a host-supplied caller that + * still holds now. */ + const admit = (caller: OwnerMutationCaller | undefined) => { + if (!caller) + throw new Error( + "Owner changes are unavailable to this caller; nothing was changed.", + ); + caller.signal?.throwIfAborted(); + caller.check(); + if (lifetime?.aborted) throw unavailable(); + const binding = current(); + return { + binding, + active: ownerMutationActive(agent.owner, binding, bind, caller, lifetime), + signal: AbortSignal.any([ + binding.signal, + ...(caller.signal ? [caller.signal] : []), + ...(lifetime ? [lifetime] : []), + ]), + execution: execution(binding), + }; + }; + const writes = ownerChannelWrites({ + agent: agent.pubkey, + admit, + records, + known: (session, id) => known(session, channelId(id)), + }); + return Object.freeze({ + createChannel: writes.create, + updateChannel: writes.update, + changeChannel: writes.change, channels: (signal?: AbortSignal) => run(signal, (binding) => { const { channels } = binding.session; @@ -323,6 +405,18 @@ export function createOwnerAccess( ? await binding.session.channelLifecycle.load(id, bound, events) : undefined; bound.throwIfAborted(); + // Versioned as channel settings save it; unreadable details only omit it. + let details: ReturnType["details"] | undefined; + try { + ({ details } = detailsSettings( + events, + id, + viewer, + binding.session.relayAuthor ?? "", + )); + } catch { + details = undefined; + } return { execution: execution(binding), channel: { @@ -331,15 +425,34 @@ export function createOwnerAccess( archived: exactLifecycleTag(metadata, "archived") === "true", role: role ?? null, members: Object.keys(rights.roles).length, - ...(summary.description - ? { description: summary.description } - : {}), - ...(summary.joinRole ? { joinRole: summary.joinRole } : {}), - ...(summary.ttlSeconds ? { ttlSeconds: summary.ttlSeconds } : {}), + ...(details + ? { + name: details.name, + description: details.description, + visibility: + details.visibility === "public" + ? ("open" as const) + : ("private" as const), + ...(details.joinRole ? { joinRole: details.joinRole } : {}), + ...(details.ttlSeconds !== undefined + ? { ttlSeconds: details.ttlSeconds } + : {}), + version: details.version, + } + : { + ...(summary.description + ? { description: summary.description } + : {}), + ...(summary.joinRole ? { joinRole: summary.joinRole } : {}), + ...(summary.ttlSeconds + ? { ttlSeconds: summary.ttlSeconds } + : {}), + }), ...(summary.creator ? { creator: summary.creator } : {}), ...(summary.createdAt ? { createdAt: summary.createdAt } : {}), }, capabilities: { + edit: !!details?.canEdit, manageRoles: rights.canManage, removeMembers: rights.canRemove, archive: !!offer?.canArchive, diff --git a/src/features/agents/owner-channels.test.ts b/src/features/agents/owner-channels.test.ts new file mode 100644 index 0000000000..27012bf6e4 --- /dev/null +++ b/src/features/agents/owner-channels.test.ts @@ -0,0 +1,913 @@ +// Owner channel writes through a real relay session: its outbox, channel +// writes, details and lifecycle services, against a fixture relay that applies +// the commands it accepts to its own signed channel records. +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { EventTemplate } from "nostr-tools"; +import type { OwnerMutationCaller } from "../../buzz-mcp/client"; +import type { ReadFilter, RelayEvent } from "../relay/events"; +import { type OutgoingEvent, PublishRejected } from "../relay/outbox"; +import { createRelaySession } from "../relay/session"; +import { type Key, keypair, signed } from "../relay/testing"; +import { SOLO_LABEL } from "../solo/labels"; +import { + createOwnerAccess, + type OwnerBinding, + type OwnerSession, + ownerMutationActive, +} from "./owner-access"; + +const community = "https://relay.example.test"; +const existing = "11111111-1111-4111-8111-111111111111"; +const fresh = "22222222-2222-4222-8222-222222222222"; +const system = "33333333-3333-4333-8333-333333333333"; +const disposers: (() => void)[] = []; +afterEach(() => { + for (const dispose of disposers.splice(0)) dispose(); +}); + +function gate() { + let open!: () => void; + const promise = new Promise((resolve) => { + open = resolve; + }); + return { promise, open }; +} + +type Channel = { tags: string[][]; roles: Record }; + +/** A relay that applies the 9007, 9002 and 9008 commands it accepts to its + * own signed 39000–39002 records, and keeps each command it stored. */ +function fixtureRelay(owner: Key, relay: Key) { + const channels = new Map(); + const stored: RelayEvent[] = []; + const published: RelayEvent[] = []; + let time = 1_700_000_000; + const script: { + /** Holds publication until released. */ + hold?: Promise; + /** Thrown after the command is stored: a lost acknowledgment. */ + lose?: Error; + /** Thrown once publication starts, before the relay applies it. */ + drop?: Error; + reject?: string; + } = {}; + const roster = (roles: Channel["roles"]) => + Object.entries(roles).map(([key, role]) => ["p", key, "", role]); + let cache: { at: number; events: RelayEvent[] } | undefined; + const records = () => { + if (cache?.at === time) return cache.events; + const events = [...channels].flatMap(([id, channel]) => [ + signed(relay, { + kind: 39000, + created_at: time, + content: "", + tags: [["d", id], ...channel.tags], + }), + signed(relay, { + kind: 39001, + created_at: time, + content: "", + tags: [ + ["d", id], + ...Object.entries(channel.roles) + .filter(([, role]) => role === "owner" || role === "admin") + .map(([key, role]) => ["p", key, role]), + ], + }), + signed(relay, { + kind: 39002, + created_at: time, + content: "", + tags: [["d", id], ...roster(channel.roles)], + }), + ]); + cache = { at: time, events }; + return events; + }; + const replace = (tags: string[][], name: string, value?: string[]) => [ + ...tags.filter(([key]) => key !== name), + ...(value ? [value] : []), + ]; + function apply(event: RelayEvent) { + const tag = (name: string) => event.tags.find(([key]) => key === name)?.[1]; + const id = tag("h") ?? ""; + time += 1; + if (event.kind === 9007) { + if (channels.has(id)) + throw new PublishRejected("duplicate: channel exists"); + const labels = event.tags.filter(([key]) => key === "t"); + channels.set(id, { + tags: [ + ["name", tag("name") ?? ""], + ["t", tag("channel_type") ?? "stream"], + ...labels, + [tag("visibility") === "open" ? "public" : "private"], + ...(tag("about") ? [["about", tag("about") ?? ""]] : []), + ...(tag("ttl") ? [["ttl", tag("ttl") ?? ""]] : []), + ["P", event.pubkey], + ], + roles: { [event.pubkey]: "owner" }, + }); + return; + } + const channel = channels.get(id); + if (!channel) throw new PublishRejected("invalid: no such channel"); + if (event.kind === 9008) { + channels.delete(id); + return; + } + let tags = channel.tags; + for (const [name, value] of event.tags.slice(1)) { + if (name === "visibility") + tags = replace(replace(tags, "public"), "private", [ + value === "open" ? "public" : "private", + ]); + else if (name === "ttl") + tags = replace(tags, "ttl", value ? ["ttl", value] : undefined); + else tags = replace(tags, name ?? "", [name ?? "", value ?? ""]); + } + channel.tags = tags; + } + const sign = vi.fn(async (template: EventTemplate, _signal: AbortSignal) => + signed(owner, template), + ); + const writer = { + kinds: [9007, 9002, 9008], + sign, + publish: vi.fn(async (event: RelayEvent) => { + await script.hold; + if (script.reject) throw new PublishRejected(script.reject); + if (script.drop) throw script.drop; + published.push(event); + apply(event); + stored.push(event); + if (script.lose) throw script.lose; + }), + }; + const fixture = { + onQuery: undefined as + | ((filters: readonly ReadFilter[]) => void) + | undefined, + }; + const query = async (filters: readonly ReadFilter[]) => { + const events = [...records(), ...stored].filter((event) => + filters.some((filter) => matches(filter, event)), + ); + fixture.onQuery?.(filters); + return events; + }; + return Object.assign(fixture, { + channels, + script, + sign, + writer, + published, + query, + seed(id: string, tags: string[][], roles: Channel["roles"]) { + time += 1; + channels.set(id, { tags, roles }); + }, + }); +} + +function matches(filter: ReadFilter, event: RelayEvent) { + const tagged = (name: string) => { + const wanted = filter[`#${name}`]; + return ( + !wanted || + event.tags.some( + ([key, value]) => key === name && wanted.includes(value ?? ""), + ) + ); + }; + return ( + (!filter.ids || filter.ids.includes(event.id)) && + (!filter.kinds || filter.kinds.includes(event.kind)) && + (!filter.authors || filter.authors.includes(event.pubkey)) && + tagged("d") && + tagged("h") && + tagged("p") + ); +} + +/** The owner's real session on the fixture relay, an agent bound to it, and + * a turn the test can end. */ +async function setup() { + const owner = keypair(), + relay = keypair(), + agent = keypair(); + const fixture = fixtureRelay(owner, relay); + fixture.seed( + existing, + [["name", "Plans"], ["t", "stream"], ["private"], ["P", owner.pubkey]], + { [owner.pubkey]: "owner" }, + ); + fixture.seed(system, [["name", "App data"], ["t", "system"], ["private"]], { + [owner.pubkey]: "owner", + }); + let journal: readonly OutgoingEvent[] = []; + const outboxStorage = { + load: () => structuredClone(journal), + save: (next: readonly OutgoingEvent[]) => { + journal = structuredClone(next); + }, + }; + const live = createRelaySession( + { + viewer: owner.pubkey, + relayAuthor: relay.pubkey, + channelIdentity: true, + media: () => undefined, + query: fixture.query, + writer: fixture.writer, + channelDetails: fixture.writer, + channelLifecycle: fixture.writer, + }, + { outboxStorage }, + ); + disposers.push(live.dispose); + live.session.channels.ensureList(); + await vi.waitFor(() => + expect(live.session.channels.list().status).toBe("ready"), + ); + const replaced = new AbortController(); + const lifetime = new AbortController(); + const state = { + binding: { + viewer: owner.pubkey, + origin: community, + session: live.session as unknown as OwnerSession, + signal: replaced.signal, + } as OwnerBinding | undefined, + }; + const access = createOwnerAccess( + { pubkey: agent.pubkey, owner: owner.pubkey }, + () => state.binding, + lifetime.signal, + ); + const turn = { ended: false, controller: new AbortController() }; + const caller: OwnerMutationCaller = { + signal: turn.controller.signal, + check: () => { + if (turn.ended) throw new Error("The turn has ended"); + }, + }; + return { + owner, + agent, + fixture, + session: live.session, + dispose: live.dispose, + access, + caller, + turn, + state, + replaced, + lifetime, + }; +} + +const plans = { id: fresh, name: "Launch", visibility: "open" as const }; + +describe("owner_channel_create", () => { + it("creates the channel with the owner as its owner, without adding the agent", async () => { + const t = await setup(); + const result = await t.access.createChannel( + { ...plans, description: " Next quarter ", ttlSeconds: 3600 }, + t.caller, + ); + expect(result).toMatchObject({ + action: "create", + channel: fresh, + outcome: "confirmed", + ownerRole: "owner", + callerMember: false, + execution: { agent: t.agent.pubkey, owner: t.owner.pubkey, community }, + }); + expect(result.operation).toBe(t.fixture.published[0]?.id); + expect(t.fixture.published[0]?.tags).toEqual( + expect.arrayContaining([ + ["h", fresh], + ["name", "Launch"], + ["visibility", "open"], + ["about", "Next quarter"], + ["ttl", "3600"], + ]), + ); + expect(t.fixture.published[0]?.tags.some(([key]) => key === "p")).toBe( + false, + ); + // It lists in the owner's session, and its receipt is settled. + expect( + t.session.channels.list().channels.some(({ id }) => id === fresh), + ).toBe(true); + expect(t.session.outbox?.snapshot()).toEqual([]); + + // The same call again confirms it; nothing more is signed. + await expect( + t.access.createChannel( + { ...plans, description: "Next quarter", ttlSeconds: 3600 }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "confirmed", ownerRole: "owner" }); + // Different details under that ID are a conflict, never a second channel. + await expect( + t.access.createChannel({ ...plans, name: "Other" }, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "conflict" }); + expect(t.fixture.sign).toHaveBeenCalledOnce(); + }); + + it("creates a private Solo chat with only the owner in it", async () => { + const t = await setup(); + await expect( + t.access.createChannel( + { id: fresh, name: "Me", visibility: "private", solo: true }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "confirmed", callerMember: false }); + const tags = t.fixture.published[0]?.tags ?? []; + expect(tags).toContainEqual(["t", SOLO_LABEL]); + expect(tags).toContainEqual(["visibility", "private"]); + expect(tags.some(([key]) => key === "p")).toBe(false); + }); + + it("answers unconfirmed after a lost acknowledgment, and the same call resumes the same command", async () => { + const t = await setup(); + t.fixture.script.lose = new Error("connection lost"); + const first = await t.access.createChannel(plans, t.caller); + expect(first).toMatchObject({ outcome: "unconfirmed" }); + expect(first.operation).toBe(t.fixture.published[0]?.id); + delete t.fixture.script.lose; + // The relay already holds it: the retry confirms that command. + const again = await t.access.createChannel( + { ...plans, operation: first.operation }, + t.caller, + ); + expect(again).toMatchObject({ + outcome: "confirmed", + operation: first.operation, + ownerRole: "owner", + }); + expect(t.fixture.sign).toHaveBeenCalledOnce(); + expect(t.fixture.published).toHaveLength(1); + }); + + it("never signs a replacement for a possibly sent create whose receipt is gone", async () => { + const t = await setup(); + t.fixture.script.drop = new Error("connection lost"); + const first = await t.access.createChannel(plans, t.caller); + expect(first).toMatchObject({ outcome: "unconfirmed" }); + const operation = first.operation ?? ""; + expect(operation).toMatch(/^[0-9a-f]{64}$/); + // The owner clears the receipt; the relay shows no channel either. + await vi.waitFor(async () => { + await t.session.outbox?.dismiss(operation); + expect(t.session.outbox?.snapshot()).toEqual([]); + }); + delete t.fixture.script.drop; + const signedBefore = t.fixture.sign.mock.calls.length; + const publishedBefore = t.fixture.writer.publish.mock.calls.length; + + // A later call resuming that create cannot tell it from a new one. + const again = await t.access.createChannel( + { ...plans, operation }, + t.caller, + ); + expect(again).toMatchObject({ outcome: "unconfirmed", operation }); + expect(again.error).toContain("owner_channel_get"); + expect(t.fixture.sign).toHaveBeenCalledTimes(signedBefore); + expect(t.fixture.writer.publish).toHaveBeenCalledTimes(publishedBefore); + expect(t.fixture.channels.has(fresh)).toBe(false); + + // Absence proved nothing: once the channel appears, resuming confirms it. + t.fixture.seed( + fresh, + [["name", "Launch"], ["t", "stream"], ["public"], ["P", t.owner.pubkey]], + { [t.owner.pubkey]: "owner" }, + ); + await expect( + t.access.createChannel({ ...plans, operation }, t.caller), + ).resolves.toMatchObject({ + outcome: "confirmed", + operation, + ownerRole: "owner", + }); + expect(t.fixture.sign).toHaveBeenCalledTimes(signedBefore); + }); + + it.each(["system", "Solo with members"] as const)( + "does not resume a queued %s create as a narrower owner-tool request", + async (kind) => { + const t = await setup(); + // Another supported app path queued a broader create, but signing failed. + t.fixture.sign.mockRejectedValueOnce(new Error("signer unavailable")); + const operation = + kind === "system" + ? t.session.channelWrites.createChannel( + fresh, + plans.name, + "open", + undefined, + undefined, + undefined, + { system: true }, + ) + : t.session.channelWrites.create(fresh, plans.name, undefined, true, [ + t.agent.pubkey, + ]); + await vi.waitFor(() => + expect(t.session.outbox?.snapshot()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + event: expect.objectContaining({ id: operation }), + delivery: "failed", + }), + ]), + ), + ); + const signedBefore = t.fixture.sign.mock.calls.length; + await expect( + t.access.createChannel( + { + ...plans, + ...(kind === "Solo with members" + ? { solo: true, visibility: "private" as const } + : {}), + operation, + }, + t.caller, + ), + ).resolves.toMatchObject({ + outcome: "rejected", + reason: "conflict", + operation, + }); + expect(t.fixture.sign).toHaveBeenCalledTimes(signedBefore); + expect(t.fixture.writer.publish).not.toHaveBeenCalled(); + }, + ); + + it("signs one create for parallel calls with the same ID, and none for a different intent", async () => { + const t = await setup(); + const held = gate(); + t.fixture.script.hold = held.promise; + const first = t.access.createChannel(plans, t.caller); + const second = t.access.createChannel(plans, t.caller); + try { + await vi.waitFor(() => + expect(t.fixture.writer.publish).toHaveBeenCalledOnce(), + ); + // While that create is still queued, other details or another create's + // receipt under its ID conflict. + await expect( + t.access.createChannel({ ...plans, name: "Other" }, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "conflict" }); + await expect( + t.access.createChannel( + { ...plans, operation: "f".repeat(64) }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "rejected", reason: "conflict" }); + } finally { + held.open(); + } + const answers = await Promise.all([first, second]); + expect(answers.map(({ outcome }) => outcome)).toEqual([ + "confirmed", + "confirmed", + ]); + expect(answers[1]?.operation).toBe(answers[0]?.operation); + expect(t.fixture.sign).toHaveBeenCalledOnce(); + }); + + it("is rejected, changing nothing, when the relay refuses it", async () => { + const t = await setup(); + t.fixture.script.reject = "blocked: not allowed"; + await expect( + t.access.createChannel(plans, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "refused" }); + expect(t.fixture.channels.has(fresh)).toBe(false); + }); + + it("signs nothing once the turn ends before the command is queued", async () => { + const t = await setup(); + // The turn ends while the create's existence check reads the relay. + t.fixture.onQuery = (filters) => { + if (filters.some((filter) => filter["#d"]?.includes(fresh))) + t.turn.ended = true; + }; + await expect( + t.access.createChannel(plans, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "cancelled" }); + expect(t.fixture.sign).not.toHaveBeenCalled(); + }); + + it.each([ + [ + "the turn ends", + (t: Awaited>) => { + t.turn.ended = true; + }, + ], + [ + "the turn is stopped", + (t: Awaited>) => { + t.turn.controller.abort(); + }, + ], + [ + "the owner switches accounts", + (t: Awaited>) => { + t.state.binding = undefined; + }, + ], + [ + "the agent stops running here", + (t: Awaited>) => { + t.lifetime.abort(); + }, + ], + ])( + "publishes nothing when %s while the command is being signed", + async (_, end) => { + const t = await setup(); + const signing = gate(); + const release = gate(); + t.fixture.sign.mockImplementationOnce(async (template) => { + signing.open(); + await release.promise; + return signed(t.owner, template); + }); + const pending = t.access.createChannel(plans, t.caller); + try { + await signing.promise; + end(t); + } finally { + release.open(); + } + const result = await pending; + expect(result.outcome).toBe("rejected"); + expect(result.reason).toBe("cancelled"); + expect(t.fixture.writer.publish).not.toHaveBeenCalled(); + expect(t.fixture.channels.has(fresh)).toBe(false); + }, + ); + + it("refuses a caller the host did not supply, and an agent no longer running here", async () => { + const t = await setup(); + await expect(t.access.createChannel(plans)).rejects.toThrow( + "unavailable to this caller", + ); + t.lifetime.abort(); + await expect(t.access.createChannel(plans, t.caller)).rejects.toThrow(); + expect(t.fixture.sign).not.toHaveBeenCalled(); + }); +}); + +describe("owner_channel_update", () => { + it("pairs the fresh editable settings with their version while discovery still has old values", async () => { + const t = await setup(); + t.fixture.seed( + existing, + [ + ["name", "Changed elsewhere"], + ["t", "stream"], + ["public"], + ["about", "Current description"], + ["join_role", "guest"], + ["ttl", "7200"], + ["P", t.owner.pubkey], + ], + { [t.owner.pubkey]: "owner" }, + ); + expect( + t.session.channels.list().channels.find(({ id }) => id === existing) + ?.name, + ).toBe("Plans"); + const { channel } = await t.access.channel(existing); + expect(channel).toMatchObject({ + name: "Changed elsewhere", + description: "Current description", + visibility: "open", + joinRole: "guest", + ttlSeconds: 7200, + }); + expect(channel.version).toMatch(/^[0-9a-f]{64}$/); + await expect( + t.access.updateChannel( + { + id: existing, + version: channel.version ?? "", + description: "New description", + }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "confirmed" }); + expect(t.fixture.channels.get(existing)?.tags).toContainEqual([ + "name", + "Changed elsewhere", + ]); + }); + + it("saves exactly the given fields against the version read", async () => { + const t = await setup(); + const { channel } = await t.access.channel(existing); + expect(channel.version).toMatch(/^[0-9a-f]{64}$/); + const result = await t.access.updateChannel( + { + id: existing, + version: channel.version ?? "", + description: "Shipping", + visibility: "public", + }, + t.caller, + ); + expect(result).toMatchObject({ outcome: "confirmed", ownerRole: "owner" }); + expect(result.version).not.toBe(channel.version); + const tags = t.fixture.channels.get(existing)?.tags ?? []; + expect(tags).toEqual( + expect.arrayContaining([ + ["name", "Plans"], + ["about", "Shipping"], + ["public"], + ]), + ); + }); + + it("saves nothing over a newer version, and saves over the version it answered", async () => { + const t = await setup(); + const { channel } = await t.access.channel(existing); + const saved = await t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Plans 2" }, + t.caller, + ); + const stale = await t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Plans 3" }, + t.caller, + ); + expect(stale).toMatchObject({ outcome: "rejected", reason: "conflict" }); + expect(stale.version).toBe(saved.version); + expect(t.fixture.sign).toHaveBeenCalledOnce(); + await expect( + t.access.updateChannel( + { id: existing, version: saved.version ?? "", name: "Plans 3" }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "confirmed" }); + }); + + it("is cancelled before publication once the turn ends, and unconfirmed once it may have been sent", async () => { + const t = await setup(); + const { channel } = await t.access.channel(existing); + const signing = gate(); + const release = gate(); + t.fixture.sign.mockImplementationOnce(async (template) => { + signing.open(); + await release.promise; + return signed(t.owner, template); + }); + const pending = t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Renamed" }, + t.caller, + ); + try { + await signing.promise; + t.turn.ended = true; + } finally { + release.open(); + } + await expect(pending).resolves.toMatchObject({ + outcome: "rejected", + reason: "cancelled", + }); + expect(t.fixture.writer.publish).not.toHaveBeenCalled(); + + t.turn.ended = false; + t.fixture.script.lose = new Error("connection lost"); + // The relay applies it but the answer is lost: never "nothing happened". + const sent = await t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Renamed" }, + t.caller, + ); + expect(sent.outcome).toBe("unconfirmed"); + delete t.fixture.script.lose; + // Its readback settles it, so the next call sees the newer version. + await expect( + t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Again" }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "rejected", reason: "conflict" }); + }); + + it("stays busy while an earlier unconfirmed save cannot be read back", async () => { + const t = await setup(); + const { channel } = await t.access.channel(existing); + t.fixture.script.drop = new Error("connection lost"); + const change = { id: existing, version: channel.version ?? "", name: "X" }; + await expect( + t.access.updateChannel(change, t.caller), + ).resolves.toMatchObject({ outcome: "unconfirmed" }); + delete t.fixture.script.drop; + await expect( + t.access.updateChannel(change, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "busy" }); + expect(t.fixture.writer.publish).toHaveBeenCalledOnce(); + }); + + it("remains unconfirmed if the owner session closes after publication starts", async () => { + const t = await setup(); + const { channel } = await t.access.channel(existing); + const held = gate(); + t.fixture.script.hold = held.promise; + const pending = t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Renamed" }, + t.caller, + ); + try { + await vi.waitFor(() => + expect(t.fixture.writer.publish).toHaveBeenCalledOnce(), + ); + t.dispose(); + } finally { + held.open(); + } + await expect(pending).resolves.toMatchObject({ outcome: "unconfirmed" }); + expect(t.fixture.channels.get(existing)?.tags).toContainEqual([ + "name", + "Renamed", + ]); + }); +}); + +describe("owner channel lifecycle", () => { + it("creates, inspects, updates, archives, rediscovers, unarchives and deletes by exact ID", async () => { + const t = await setup(); + await expect( + t.access.createChannel(plans, t.caller), + ).resolves.toMatchObject({ outcome: "confirmed", callerMember: false }); + const { channel } = await t.access.channel(fresh); + await expect( + t.access.updateChannel( + { id: fresh, version: channel.version ?? "", name: "Launch plan" }, + t.caller, + ), + ).resolves.toMatchObject({ outcome: "confirmed", callerMember: false }); + await expect( + t.access.changeChannel("archive", fresh, t.caller), + ).resolves.toMatchObject({ outcome: "confirmed", ownerRole: "owner" }); + expect(t.fixture.channels.get(fresh)?.tags).toContainEqual([ + "archived", + "true", + ]); + expect((await t.access.channels()).channels).toEqual( + expect.arrayContaining([ + expect.objectContaining({ id: fresh, archived: true }), + ]), + ); + await expect( + t.access.changeChannel("unarchive", fresh, t.caller), + ).resolves.toMatchObject({ outcome: "confirmed" }); + await expect( + t.access.changeChannel("delete", fresh, t.caller), + ).resolves.toMatchObject({ action: "delete", outcome: "confirmed" }); + expect(t.fixture.channels.has(fresh)).toBe(false); + }); + + it("answers busy, not success, while another channel action runs", async () => { + const t = await setup(); + const held = gate(); + t.fixture.script.hold = held.promise; + const first = t.access.changeChannel("archive", existing, t.caller); + try { + await vi.waitFor(() => + expect(t.fixture.writer.publish).toHaveBeenCalledOnce(), + ); + await expect( + t.access.changeChannel("delete", existing, t.caller), + ).resolves.toMatchObject({ outcome: "rejected", reason: "busy" }); + } finally { + held.open(); + } + await expect(first).resolves.toMatchObject({ outcome: "confirmed" }); + expect(t.fixture.channels.has(existing)).toBe(true); + }); + + it("is unconfirmed when the command may have been sent", async () => { + const t = await setup(); + t.fixture.script.lose = new Error("connection lost"); + await expect( + t.access.changeChannel("delete", existing, t.caller), + ).resolves.toMatchObject({ outcome: "unconfirmed" }); + }); + + it("changes no system channel and no unknown action", async () => { + const t = await setup(); + await expect( + t.access.changeChannel("delete", system, t.caller), + ).rejects.toThrow(); + await expect( + t.access.changeChannel("hide" as "delete", existing, t.caller), + ).rejects.toThrow("No channel action"); + expect(t.fixture.sign).not.toHaveBeenCalled(); + }); +}); + +describe("owner channel signing admission", () => { + it.each(["update", "archive", "unarchive", "delete"] as const)( + "signs no %s command if the caller retires during authority reads", + async (action) => { + const t = await setup(); + if (action === "unarchive") + t.fixture.seed( + existing, + [ + ["name", "Plans"], + ["t", "stream"], + ["private"], + ["P", t.owner.pubkey], + ["archived", "true"], + ], + { [t.owner.pubkey]: "owner" }, + ); + const { channel } = await t.access.channel(existing); + t.fixture.onQuery = (filters) => { + if (filters.some((filter) => filter["#d"]?.includes(existing))) + t.turn.ended = true; + }; + const result = + action === "update" + ? await t.access.updateChannel( + { id: existing, version: channel.version ?? "", name: "Renamed" }, + t.caller, + ) + : await t.access.changeChannel(action, existing, t.caller); + expect(t.turn.ended).toBe(true); + expect(result).toMatchObject({ + outcome: "rejected", + reason: "cancelled", + }); + expect(t.fixture.sign).not.toHaveBeenCalled(); + expect(t.fixture.writer.publish).not.toHaveBeenCalled(); + }, + ); +}); + +describe("ownerMutationActive", () => { + const binding = (patch: Partial = {}): OwnerBinding => ({ + viewer: "owner", + origin: community, + session: {} as OwnerSession, + signal: new AbortController().signal, + ...patch, + }); + const caller = (check = () => {}): OwnerMutationCaller => ({ check }); + + it("holds only while the caller, binding, lifetime and recorded owner all hold", () => { + const admitted = binding(); + expect( + ownerMutationActive("owner", admitted, () => admitted, caller())(), + ).toBe(true); + const ended = caller(() => { + throw new Error("ended"); + }); + expect( + ownerMutationActive("owner", admitted, () => admitted, ended)(), + ).toBe(false); + const stopped = new AbortController(); + stopped.abort(); + expect( + ownerMutationActive("owner", admitted, () => admitted, { + signal: stopped.signal, + check: () => {}, + })(), + ).toBe(false); + expect( + ownerMutationActive( + "owner", + admitted, + () => admitted, + caller(), + stopped.signal, + )(), + ).toBe(false); + expect( + ownerMutationActive("owner", admitted, () => undefined, caller())(), + ).toBe(false); + for (const now of [ + binding({ session: {} as OwnerSession }), + binding({ viewer: "someone", session: admitted.session }), + binding({ + origin: "https://other.example.test", + session: admitted.session, + }), + binding({ signal: stopped.signal, session: admitted.session }), + ]) + expect( + ownerMutationActive("owner", admitted, () => now, caller())(), + ).toBe(false); + expect( + ownerMutationActive("another", admitted, () => admitted, caller())(), + ).toBe(false); + }); +}); diff --git a/src/features/agents/owner-channels.ts b/src/features/agents/owner-channels.ts new file mode 100644 index 0000000000..9e608f08d9 --- /dev/null +++ b/src/features/agents/owner-channels.ts @@ -0,0 +1,425 @@ +// Owner channel writes: create, update, archive, unarchive and delete through +// the owner's own session services. Each service checks admission before signing +// and publishing. Once a command may have been published, the answer is +// unconfirmed, never "nothing happened". +import type { + OwnerChannelCreate, + OwnerChannelUpdate, + OwnerChannelWrite, + OwnerExecution, + OwnerLifecycleAction, + OwnerMutationCaller, +} from "../../buzz-mcp/client"; +import { memberAuthority } from "../channel-members/administration-protocol"; +import { ChannelDetailsUnconfirmed } from "../relay/channel-details"; +import { + canonicalDetailsName, + detailsSettings, +} from "../relay/channel-details-protocol"; +import { channelIdentity } from "../relay/channel-labels"; +import { ChannelLifecycleUnconfirmed } from "../relay/channel-lifecycle"; +import { + channelTypeTag, + lifecycleRecord, +} from "../relay/channel-lifecycle-protocol"; +import type { RelayEvent } from "../relay/events"; +import { PublishRejected } from "../relay/outbox"; +import { SOLO_LABEL } from "../solo/labels"; +import type { OwnerBinding, OwnerSession } from "./owner-access"; + +type Admitted = Readonly<{ + binding: OwnerBinding; + active: () => boolean; + signal: AbortSignal; + execution: OwnerExecution; +}>; +type Answer = Omit; +type Intent = Readonly<{ + name: string; + visibility: "open" | "private"; + about: string; + ttl: number | undefined; + solo: boolean; +}>; + +export function ownerChannelWrites({ + agent, + admit, + records, + known, +}: Readonly<{ + /** The calling agent. */ + agent: string; + admit(caller: OwnerMutationCaller | undefined): Admitted; + /** The channel's relay-signed metadata, admin list and roster, read fresh. */ + records( + binding: OwnerBinding, + id: string, + signal: AbortSignal, + ): Promise; + /** Throws unless the owner's session lists channel `id`. */ + known(session: OwnerSession, id: string): unknown; +}>) { + const answer = ( + admitted: Admitted, + action: OwnerChannelWrite["action"], + channel: string, + fields: Answer, + ): OwnerChannelWrite => ({ + execution: admitted.execution, + action, + channel, + ...fields, + }); + /** Pre-publication refusals: nothing was sent. */ + const refused = (admitted: Admitted, error: unknown): Answer => ({ + outcome: "rejected", + reason: + error instanceof PublishRejected + ? "refused" + : !admitted.active() || + (error instanceof Error && error.name === "AbortError") + ? "cancelled" + : "failed", + error: message(error), + }); + /** The owner's role, the caller's membership and the metadata version, as + * the relay records them now. */ + const readback = async ( + { binding, signal }: Admitted, + id: string, + events?: readonly RelayEvent[], + ) => { + const { relayAuthor } = binding.session; + if (!relayAuthor) throw new Error("The owner's relay is unavailable."); + const read = events ?? (await records(binding, id, signal)); + const rights = memberAuthority(read, id, binding.viewer, relayAuthor); + const metadata = lifecycleRecord(read, 39000, id, relayAuthor); + return { + ownerRole: rights.roles[binding.viewer] ?? null, + callerMember: agent in rights.roles, + ...(metadata ? { version: metadata.id } : {}), + }; + }; + /** Channel `id`, one of the owner's ordinary channels or Solo chats: never + * a system channel or DM. */ + const target = async (admitted: Admitted, id: string) => { + const { binding, signal } = admitted; + known(binding.session, id); + const events = await records(binding, id, signal); + const metadata = binding.session.relayAuthor + ? lifecycleRecord(events, 39000, id, binding.session.relayAuthor) + : undefined; + const type = metadata && channelTypeTag(metadata); + if (type !== "stream" && type !== "forum") + throw new Error( + `Channel ${id} is not a channel or Solo chat that owner tools change.`, + ); + return events; + }; + + async function create( + input: OwnerChannelCreate, + caller?: OwnerMutationCaller, + ): Promise { + const admitted = admit(caller); + const { binding, active, signal } = admitted; + const { session } = binding; + const id = input.id; + const done = (fields: Answer) => answer(admitted, "create", id, fields); + const wanted: Intent = { + name: canonicalDetailsName(input.name), + visibility: input.solo ? "private" : input.visibility, + about: input.solo ? "" : (input.description?.trim() ?? ""), + ttl: input.ttlSeconds, + solo: !!input.solo, + }; + const writes = session.channelWrites; + if (!writes.available) + return done({ + outcome: "rejected", + reason: "failed", + error: "This community cannot create channels now.", + }); + // A create this owner already queued under this ID is resumed, never + // replaced: the same receipt, or a conflict when its intent differs. + const journaled = () => + session.outbox + ?.snapshot() + .find( + (item) => item.event.kind === 9007 && tag(item.event, "h") === id, + ); + const resumed = input.operation; + let queued = journaled(); + if (!queued) { + const events = await records(binding, id, signal); + if (events.some((event) => event.kind === 39000)) + return matches(binding, id, events, wanted) + ? done({ + outcome: "confirmed", + ...(resumed ? { operation: resumed } : {}), + ...(await readback(admitted, id)), + }) + : done({ + outcome: "rejected", + reason: "conflict", + error: `Channel ID ${id} is already in use.`, + }); + // A concurrent call may have queued it during that read. + queued = journaled(); + // Without its receipt or the channel, an earlier create may still be in + // flight or applied unseen; another create could not be told from it. + if (!queued && resumed) + return done({ + outcome: "unconfirmed", + operation: resumed, + error: `Buzz no longer holds create ${resumed}, and the relay shows no channel ${id} yet, so it may still appear. Check with owner_channel_get before creating it again under a new id.`, + }); + } + if (queued && resumed && queued.event.id !== resumed) + return done({ + outcome: "rejected", + reason: "conflict", + operation: queued.event.id, + error: `Channel ID ${id} is queued by create ${queued.event.id}, not ${resumed}.`, + }); + if (queued) { + if ( + tag(queued.event, "channel_type") !== "stream" || + queued.event.tags.some(([key]) => key === "p") || + !same(intentOf(queued.event.tags), wanted) + ) + return done({ + outcome: "rejected", + reason: "conflict", + operation: queued.event.id, + error: `Channel ID ${id} is already used by a different create.`, + }); + if (queued.delivery === "failed" && queued.refused) + return done({ + outcome: "rejected", + reason: "refused", + operation: queued.event.id, + error: queued.error ?? "The relay refused this create.", + }); + } + let operation: string; + try { + operation = + queued?.event.id ?? + (wanted.solo + ? writes.create(id, wanted.name, wanted.ttl, true, [], active) + : writes.createChannel( + id, + wanted.name, + wanted.visibility, + wanted.about || undefined, + wanted.ttl, + active, + )); + } catch (error) { + return done(refused(admitted, error)); + } + try { + await writes.delivered(operation, active); + // Lists the new channel in the owner's session, as the app's create does. + await writes.refresh(id, { member: binding.viewer }, false); + const events = await records(binding, id, signal); + if (!matches(binding, id, events, wanted)) + throw new Error("The created channel does not match this request."); + await session.outbox?.dismiss(operation); + return done({ + outcome: "confirmed", + operation, + ...(await readback(admitted, id, events)), + }); + } catch (error) { + const item = session.outbox + ?.snapshot() + .find((each) => each.event.id === operation); + // Failed without the relay's answer means it never reached transport; + // the receipt stays for a later call with this ID to resume. + if (item?.delivery === "failed") + return done({ + ...refused(admitted, error), + ...(item.refused ? { reason: "refused" as const } : {}), + operation, + ...(item.error ? { error: item.error } : {}), + }); + return done({ outcome: "unconfirmed", operation, error: message(error) }); + } + } + + async function update( + input: OwnerChannelUpdate, + caller?: OwnerMutationCaller, + ): Promise { + const admitted = admit(caller); + const { binding, signal } = admitted; + const id = input.id; + const done = (fields: Answer) => answer(admitted, "update", id, fields); + const details = binding.session.channelDetails; + await target(admitted, id); + // An earlier unconfirmed save blocks the next until a readback settles it. + if (details.snapshot(id)?.status === "unconfirmed") + await details.check(id, signal).catch(() => {}); + if (details.snapshot(id)) + return done({ + outcome: "rejected", + reason: "busy", + error: + "An earlier change to this channel is still saving or unconfirmed; check it in Buzz first.", + }); + let base: Awaited>; + try { + base = await details.load(id, signal); + } catch (error) { + return done(refused(admitted, error)); + } + if (base.version !== input.version) + return done({ + outcome: "rejected", + reason: "conflict", + version: base.version, + error: + "The channel changed since that version; read it again before changing it.", + }); + if (!base.canEdit) + return done({ + outcome: "rejected", + reason: "denied", + error: "Your owner cannot edit this channel.", + }); + const ttlSeconds = + input.ttlSeconds === null + ? undefined + : (input.ttlSeconds ?? base.ttlSeconds); + // Labels stay as they are; a renamed Solo chat keeps its chosen name. + const draft = { + name: + input.name !== undefined ? canonicalDetailsName(input.name) : base.name, + description: input.description?.trim() ?? base.description, + visibility: input.visibility ?? base.visibility, + ...((input.joinRole ?? base.joinRole) + ? { joinRole: input.joinRole ?? base.joinRole } + : {}), + ...(ttlSeconds !== undefined ? { ttlSeconds } : {}), + }; + try { + await details.save(base, draft, signal, admitted.active); + } catch (error) { + return done( + error instanceof ChannelDetailsUnconfirmed + ? { outcome: "unconfirmed", error: message(error) } + : refused(admitted, error), + ); + } + return done({ outcome: "confirmed", ...(await settled(admitted, id)) }); + } + + async function change( + action: OwnerLifecycleAction, + id: string, + caller?: OwnerMutationCaller, + ): Promise { + if (!["archive", "unarchive", "delete"].includes(action)) + throw new Error(`No channel action ${String(action)}.`); + const admitted = admit(caller); + const done = (fields: Answer) => answer(admitted, action, id, fields); + await target(admitted, id); + try { + await admitted.binding.session.channelLifecycle.run( + action, + id, + admitted.signal, + admitted.active, + ); + } catch (error) { + if (error instanceof ChannelLifecycleUnconfirmed) + return done({ outcome: "unconfirmed", error: message(error) }); + return done( + message(error).includes("still in progress") + ? { outcome: "rejected", reason: "busy", error: message(error) } + : message(error).includes("no longer permitted") + ? { outcome: "rejected", reason: "denied", error: message(error) } + : refused(admitted, error), + ); + } + // The lifecycle owner confirmed it from the relay's own records; a deleted + // channel has none left to read. + return done({ + outcome: "confirmed", + ...(action === "delete" ? {} : await settled(admitted, id)), + }); + } + + /** A confirmed change's readback, when it can still be read. */ + const settled = (admitted: Admitted, id: string) => + readback(admitted, id).catch(() => ({})); + + return { create, update, change }; +} + +function intentOf(tags: readonly (readonly string[])[]): Intent { + const value = (name: string) => tags.find(([key]) => key === name)?.[1]; + const ttl = value("ttl"); + return { + name: value("name") ?? "", + visibility: value("visibility") === "open" ? "open" : "private", + about: value("about") ?? "", + ttl: ttl === undefined ? undefined : Number(ttl), + solo: tags.some(([key, label]) => key === "t" && label === SOLO_LABEL), + }; +} + +/** Whether the relay's channel `id` is the one `wanted` describes, created by + * the owner, who holds it as owner. */ +function matches( + binding: OwnerBinding, + id: string, + events: readonly RelayEvent[], + wanted: Intent, +) { + const relayAuthor = binding.session.relayAuthor; + if (!relayAuthor) return false; + try { + const { details, metadata } = detailsSettings( + events, + id, + binding.viewer, + relayAuthor, + ); + const rights = memberAuthority(events, id, binding.viewer, relayAuthor); + const creator = channelIdentity(metadata).creator; + return ( + channelTypeTag(metadata) === "stream" && + (!creator || creator === binding.viewer) && + rights.roles[binding.viewer] === "owner" && + same( + { + name: details.name, + visibility: details.visibility === "public" ? "open" : "private", + about: details.description, + ttl: details.ttlSeconds, + solo: !!details.labels?.includes(SOLO_LABEL), + }, + wanted, + ) + ); + } catch { + return false; + } +} + +const same = (a: Intent, b: Intent) => + a.name === b.name && + a.visibility === b.visibility && + a.about === b.about && + a.ttl === b.ttl && + a.solo === b.solo; + +const tag = (event: { tags: readonly (readonly string[])[] }, name: string) => + event.tags.find(([key]) => key === name)?.[1]; + +const message = (error: unknown) => + error instanceof Error ? error.message : String(error); diff --git a/src/features/agents/runner.ts b/src/features/agents/runner.ts index e6044117e9..e362dfcb54 100644 --- a/src/features/agents/runner.ts +++ b/src/features/agents/runner.ts @@ -522,8 +522,9 @@ export type RunnerHost = { tabs(agent: Agent): ReturnType; /** Buzz settings on this device, where there are any. */ settings?(): DeviceSettings | undefined; - /** Owner reads for a run of the agent, where this build enables them. */ - ownerAccess?(agent: Agent): AgentHandle["ownerAccess"]; + /** Owner reads and writes for a run of the agent, where this build enables + * them; writes stop once `lifetime` aborts. */ + ownerAccess?(agent: Agent, lifetime: AbortSignal): AgentHandle["ownerAccess"]; /** Whether watch classifiers can run here. */ classifier(): ClassifierAvailability; /** The decisions service that runs classifiers, where there is one. */ @@ -2844,7 +2845,7 @@ export class AgentRunners { // with its owner's access. const ownerAccess = run && runner.type?.revision === "bundled" - ? this.host.ownerAccess?.(agent) + ? this.host.ownerAccess?.(agent, lifetime) : undefined; const handle = guarded( { diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 59263eaf6d..c184845a08 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -14669,7 +14669,16 @@ describe("full external agent access", () => { const run = runnerGate(); f.run.mockImplementationOnce(async ({ agent }) => run.resolve(agent)); f.plugin.agents.wake(bot, { channel, root: "e".repeat(64), text: "hi" }); - expect(await run.promise).toHaveProperty("ownerAccess"); + const handle = await run.promise; + expect(handle).toHaveProperty("ownerAccess"); + // Its writes take the calling turn from the host, never from a caller. + await expect( + handle.ownerAccess?.createChannel({ + id: "33333333-3333-4333-8333-333333333333", + name: "Plans", + visibility: "open", + }), + ).rejects.toThrow("unavailable to this caller"); const names = await f.names(); expect(names).toContain("send"); expect(names.filter((name) => name.startsWith("owner_"))).toEqual([]); diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index f974c561f9..9b6891ffe9 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -2460,21 +2460,27 @@ export class AgentsService extends Service implements Agents { return session && channelTabs(session, ctx.navigation); }, settings: () => this.deviceSettings, - ownerAccess: (agent) => + ownerAccess: (agent, lifetime) => ownerMcpEnabled(agent.relay) - ? createOwnerAccess(agent, () => { - const binding = this.bindingOf(agent); - return binding && - ownerMcpEnabled(binding.origin) && - this.communities.snapshot().viewer === agent.owner - ? { - viewer: binding.viewer, - origin: binding.origin, - session: binding.session, - signal: binding.controller.signal, - } - : undefined; - }) + ? createOwnerAccess( + agent, + () => { + const binding = this.bindingOf(agent); + return binding && + ownerMcpEnabled(binding.origin) && + this.communities.snapshot().viewer === agent.owner && + // The agent is still running here as this owner's. + this.agent(agent.key)?.owner === agent.owner + ? { + viewer: binding.viewer, + origin: binding.origin, + session: binding.session, + signal: binding.controller.signal, + } + : undefined; + }, + lifetime, + ) : undefined, classifier: () => this.classifier(), decisions: () => this.decisions, diff --git a/src/features/relay/channel-details.test.ts b/src/features/relay/channel-details.test.ts index 88edc4fe65..6458934570 100644 --- a/src/features/relay/channel-details.test.ts +++ b/src/features/relay/channel-details.test.ts @@ -318,7 +318,7 @@ it("publishes no edit once the caller's guard turns false during signing", async await h.owner.capability.save(base, draft); expect(h.publish).toHaveBeenCalledOnce(); }); -it("checks the guard only after the final re-authorization read settles", async () => { +it("rechecks the guard after the final re-authorization read settles", async () => { const h = harness(); const base = await h.owner.capability.load(id); let eligible = true; @@ -338,13 +338,13 @@ it("checks the guard only after the final re-authorization read settles", async .catch((error: unknown) => error); try { await started.promise; - expect(active).not.toHaveBeenCalled(); + expect(active).toHaveBeenCalledOnce(); eligible = false; } finally { gate.resolve(); } expect(await saving).toMatchObject({ name: "AbortError" }); - expect(active).toHaveBeenCalledOnce(); + expect(active).toHaveBeenCalledTimes(2); expect(h.publish).not.toHaveBeenCalled(); eligible = true; await h.owner.capability.save(base, draft, undefined, active); diff --git a/src/features/relay/channel-details.ts b/src/features/relay/channel-details.ts index 21c0a56735..5e099892a1 100644 --- a/src/features/relay/channel-details.ts +++ b/src/features/relay/channel-details.ts @@ -28,7 +28,7 @@ export interface ChannelDetailsCapability { base: ChannelDetails, draft: ChannelDetailsDraft, signal?: AbortSignal, - /** Checked synchronously right before publishing; false publishes nothing. */ + /** Checked before signing and publishing; false stops unpublished work. */ active?: () => boolean, ): Promise; check(id: string, signal?: AbortSignal): Promise; @@ -38,6 +38,13 @@ export interface ChannelDetailsCapability { const uncertain = "The change may have been saved. Check its status before trying again."; +export class ChannelDetailsUnconfirmed extends Error { + constructor() { + super(uncertain); + this.name = "ChannelDetailsUnconfirmed"; + } +} + /** Session owns uncertain intent through panel close/reopen; no automatic publication retry. */ export function createChannelDetails({ reader, @@ -236,6 +243,8 @@ export function createChannelDetails({ await authorize(); const template = detailsTemplate(id, draft, base); phase = "Could not sign the channel change"; + if (active?.() === false) + throw new DOMException("Channel edit cancelled", "AbortError"); const signed = await writer.sign(structuredClone(template), signal); signal.throwIfAborted(); if ( @@ -263,7 +272,7 @@ export function createChannelDetails({ if (publicationStarted && !(error instanceof PublishRejected)) { if (!closed) attempts.set(id, Object.freeze({ draft, status: "unconfirmed" })); - throw new Error(uncertain); + throw new ChannelDetailsUnconfirmed(); } attempts.delete(id); if (error instanceof Error && error.name === "AbortError") throw error; diff --git a/src/features/relay/channel-lifecycle.test.ts b/src/features/relay/channel-lifecycle.test.ts index 9d0fed26e4..d577a25a90 100644 --- a/src/features/relay/channel-lifecycle.test.ts +++ b/src/features/relay/channel-lifecycle.test.ts @@ -2264,8 +2264,10 @@ describe("caller publication guard", () => { const active = vi.fn(() => true); try { await h.owner.capability.run("archive", id, undefined, active); - expect(active).toHaveBeenCalledOnce(); - const [checked = 0] = active.mock.invocationCallOrder; + expect(active).toHaveBeenCalledTimes(2); + const [beforeSign = 0, checked = 0] = active.mock.invocationCallOrder; + const [signed = 0] = h.sign.mock.invocationCallOrder; + expect(beforeSign).toBeLessThan(signed); const [published = 0] = h.publish.mock.invocationCallOrder; const reads = h.read.mock.invocationCallOrder.filter( (at) => at < published, diff --git a/src/features/relay/channel-lifecycle.ts b/src/features/relay/channel-lifecycle.ts index 92c7521ac0..9bab53262b 100644 --- a/src/features/relay/channel-lifecycle.ts +++ b/src/features/relay/channel-lifecycle.ts @@ -49,7 +49,7 @@ export interface ChannelLifecycleCapability { action: ChannelLifecycleCommand, channelId: string, signal?: AbortSignal, - /** Checked synchronously right before publishing; false publishes nothing. */ + /** Checked before signing and publishing; false stops unpublished work. */ active?: () => boolean, ): Promise; snapshot(): DmVisibility; @@ -461,6 +461,8 @@ export function createChannelLifecycle({ ); }; if (await authorize()) return; + if (active?.() === false) + throw new DOMException("Channel action cancelled", "AbortError"); const fresh = lifecycleTemplate(action, id); const template = { ...fresh,