diff --git a/AGENTS.md b/AGENTS.md index e4a968022..2916966ac 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -189,7 +189,9 @@ server writes the instruction from the plan, rechecks write access as for per document. An MCP `invoke_planner` call posts its instruction as the caller's member message and runs under the channel's existing Planner owner; only a caller with a live -browser login can claim an unowned channel. +browser login can claim an unowned channel. With `as: "system"`, an automatic +follow-up limited to the document's creator or its Planner's holder and to 10 +an hour, it goes through `instruct()` and records only a system notice instead. `Chat.Send.focus` names the child a member has open. The server resolves it against storage and treats it as context, never as an authorization grant. A diff --git a/apps/server/src/chat/service.ts b/apps/server/src/chat/service.ts index 19b277e24..4d3312e32 100644 --- a/apps/server/src/chat/service.ts +++ b/apps/server/src/chat/service.ts @@ -115,6 +115,8 @@ export type Waiting = Wire.Waiting & { focus?: Focus; /** The child holding `thread`, when the thread is not in this document. */ threadChannel?: string; + /** See `Instruction`. */ + quiet?: boolean; }; /** A request for implementation tasks, live while its turn is queued or running. */ @@ -169,6 +171,8 @@ export type Instruction = { focus?: Focus; /** The child holding `thread`, when the thread is not in this document. */ threadChannel?: string; + /** Kept out of the visible queue: its notice says why, and its text is for the Planner alone. */ + quiet?: boolean; }; /** @@ -553,7 +557,7 @@ function responded(chat: Chat, server: Server, room: string, text: s * anyone sent, so they stay off the wire entirely. */ function visible(chat: Chat): Wire.Waiting[] { - return chat.waiting.filter(item => !item.draft && !item.job).map(( + return chat.waiting.filter(item => !item.draft && !item.job && !item.quiet).map(( { handle, id, text, references }, ) => ({ handle, @@ -694,7 +698,9 @@ export function send(context: Room, ws: Socket, msg: Request): Promis * * The caller is attributed on the posted message, and the turn runs under the * channel's Planner owner. A checkout matters only to the atomic harness, which - * verifies it before anything is posted and remembers it for the channel. + * verifies it before anything is posted and remembers it for the channel. With a + * `notice`, an automatic follow-up, Chat shows that notice instead of a message + * and the text goes to the Planner alone, as for any other instruction. */ export async function invoke( context: Room, @@ -704,6 +710,7 @@ export async function invoke( focus?: Focus, /** Called once, when the invoked turn ends or is dropped; never when this refuses. */ settled?: () => void, + notice?: string, ): Promise< | "planner-unavailable" | "planner-not-attached" @@ -732,6 +739,15 @@ export async function invoke( } if (chat.closed) return "planner-unavailable" as const; if (chat.busy && chat.waiting.length >= MAX_QUEUE) return "planner-queue-full" as const; + if (notice !== undefined) { + if (verified) rememberCheckout(room, verified); + await instruct(context, user.login, text, notice, { + quiet: true, + ...(focus ? { focus } : {}), + ...(settled ? { ended: () => settled() } : {}), + }); + return undefined; + } let entry: Wire.Entry = { id: ulid(), author: { kind: "member", handle: user.login }, diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index ca52b62c4..2fc861068 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -75,7 +75,7 @@ import { JobRunner } from "./jobs/runner"; import { researchAnswerDefinition, researchEvidenceDefinition } from "./jobs/research-workspace"; import { JobService } from "./jobs/service"; import { DocumentSummaryCoordinator } from "./jobs/summary-coordinator"; -import { invokeRouted } from "./mcp/invoke"; +import { invokeRouted, SystemTurns } from "./mcp/invoke"; import { liveClassifier } from "./tasks/live-gate"; import { LiveSyncCoordinator } from "./tasks/live-sync"; import { heuristicJudge, jevJudge, SpikeScout } from "./experiments/spike-scout"; @@ -174,6 +174,8 @@ let sessionCleanup: ReturnType | undefined; let cleaningSessions: Promise | undefined; let ownerBindings: ActiveOwnerBindings | undefined; let plannerAttachments: PlannerAttachments | undefined; +/** Each document's recent automatic follow-ups through `invoke_planner`, for their hourly limit. */ +let systemTurns = new SystemTurns(); /** Browser sockets by login, so a login that ends closes them before another broadcast. */ let sessionSockets = new SessionSockets(sessionId => hostedAuth.sessions.isLive(sessionId)); let jobRunner: JobRunner | undefined; @@ -2475,6 +2477,8 @@ registerMcpRoutes(router, hostedAuth, { deleting: id => deletingChannels.has(id) || archivingChannels.has(id), transition: withDocumentTransition, open: rootId => openRootChat(rootId, input.session?.session.id, input.repository), + plannerAccount: rootId => plannerAttachments?.holding(rootId)?.userId, + systemTurns, }, input), plannerNotAttached, archiveChannel, diff --git a/apps/server/src/mcp.ts b/apps/server/src/mcp.ts index 43b4df4ab..98fe245f0 100644 --- a/apps/server/src/mcp.ts +++ b/apps/server/src/mcp.ts @@ -149,7 +149,13 @@ export type RestoreDocument = { >; }; -export type InvokePlannerInput = { id: string; instruction: string; checkout?: string }; +export type InvokePlannerInput = { + id: string; + instruction: string; + checkout?: string; + /** An automatic follow-up: Chat shows a system notice rather than the caller's message. */ + as?: "system"; +}; export type InvokePlannerError = | "document-unavailable" @@ -159,7 +165,9 @@ export type InvokePlannerError = | "planner-owner-unavailable" | "checkout-unverified" | "planner-unavailable" - | "planner-queue-full"; + | "planner-queue-full" + | "system-turn-forbidden" + | "system-turn-limited"; export type InvokePlanner = { invoke(caller: Caller, input: InvokePlannerInput): Promise< @@ -524,7 +532,8 @@ export const TOOLS: Tool[] = [ + "owner, and otherwise the call is refused. Where the Planner runs in an attached " + "client, a document with no attached Planner refuses the call with " + "planner-not-attached and a message saying how to attach one. Questions appear in " - + "Decisions.", + + 'Decisions. With as: "system", for an automatic follow-up, Chat shows a short ' + + "system notice instead of the caller's message.", inputSchema: { type: "object", properties: { @@ -547,6 +556,14 @@ export const TOOLS: Tool[] = [ + "repository. Used only by HARNESS=atomic, which verifies its origin and " + "remembers it as the document's Planner working directory; ignored otherwise.", }, + as: { + type: "string", + enum: ["system"], + description: "An automatic follow-up nobody typed, such as a finished workflow run: " + + "Chat shows a system notice naming the caller, and the instruction goes to the " + + "Planner alone. Only the document's creator, or the account whose client holds " + + "its Planner, may use it, at most 10 times per document per hour.", + }, }, required: ["id", "instruction"], additionalProperties: false, @@ -574,6 +591,8 @@ export const TOOLS: Tool[] = [ "checkout-unverified", "planner-unavailable", "planner-queue-full", + "system-turn-forbidden", + "system-turn-limited", ], }, message: { type: "string" }, @@ -1100,19 +1119,20 @@ export function handler( if (tool.name === "invoke_planner" && options.invoke) { let args = tool.arguments; if ( - Object.keys(args).some(key => !["id", "instruction", "checkout"].includes(key)) + Object.keys(args).some(key => !["id", "instruction", "checkout", "as"].includes(key)) || !isLocator(args.id) || typeof args.instruction !== "string" || !args.instruction.trim() || Buffer.byteLength(args.instruction) > MAX_MESSAGE_BYTES || (Object.hasOwn(args, "checkout") && (typeof args.checkout !== "string" || !isAbsolute(args.checkout))) + || (Object.hasOwn(args, "as") && args.as !== "system") ) { return notification ? undefined : error( call.id, -32602, - "invoke_planner requires an id or URL, instruction, and optional absolute checkout", + "invoke_planner requires an id or URL, instruction, optional absolute checkout, and optional as: system", ); } let result = await options.invoke.invoke(caller, args as InvokePlannerInput); diff --git a/apps/server/src/mcp/hosted.test.ts b/apps/server/src/mcp/hosted.test.ts index 35b419348..82791ec06 100644 --- a/apps/server/src/mcp/hosted.test.ts +++ b/apps/server/src/mcp/hosted.test.ts @@ -276,6 +276,10 @@ describe("the hosted MCP adapter", () => { expect(started[1]).toMatchObject({ session: { session: { id: session.id }, user: { id: caller.user.id } }, }); + expect(Object.hasOwn(started[1]!, "system")).toBe(false); + refusal = undefined; + await adapter.invoke!.invoke(caller, { ...input, as: "system" }); + expect(started[2]).toMatchObject({ instruction: input.instruction, system: true }); } finally { await Service.close(opened.plan); } diff --git a/apps/server/src/mcp/hosted.ts b/apps/server/src/mcp/hosted.ts index c166bd1e5..43c214d3f 100644 --- a/apps/server/src/mcp/hosted.ts +++ b/apps/server/src/mcp/hosted.ts @@ -52,6 +52,8 @@ export type PlannerInvocation = { instruction: string; /** Unverified; only the atomic harness reads it. */ checkout?: string; + /** An automatic follow-up shown in Chat as a system notice rather than the caller's message. */ + system?: boolean; }; export type ImplementationPersistence = { lease(): Lease }; @@ -341,6 +343,7 @@ export function hosted( session: await auth.sessions.forUser(caller.user.id), instruction: input.instruction, ...(input.checkout === undefined ? {} : { checkout: input.checkout }), + ...(input.as === "system" ? { system: true } : {}), }); let url = new URL(await documentUrl(channel, auth.storage.channels), auth.config.origin).href; diff --git a/apps/server/src/mcp/invoke.test.ts b/apps/server/src/mcp/invoke.test.ts index 9d2e1f8ab..7156a4795 100644 --- a/apps/server/src/mcp/invoke.test.ts +++ b/apps/server/src/mcp/invoke.test.ts @@ -3,7 +3,7 @@ import { afterEach, expect, test } from "bun:test"; import * as Chat from "../chat/service"; import { handler, TOOLS } from "../mcp"; import { rooted } from "../testing/rooted"; -import { invokeRouted } from "./invoke"; +import { invokeRouted, MAX_SYSTEM_TURNS_PER_HOUR, SystemTurns } from "./invoke"; let cleanups: Array<() => Promise> = []; afterEach(async () => { @@ -82,6 +82,8 @@ test("invoke_planner validates strict arguments and counts instruction bytes wit { ...valid, checkout: null }, { ...valid, checkout: "" }, { ...valid, extra: true }, + { ...valid, as: "member" }, + { ...valid, as: true }, ] ) { let result = await call(mcp, "tools/call", { name: "invoke_planner", arguments: args }); @@ -93,6 +95,11 @@ test("invoke_planner validates strict arguments and counts instruction bytes wit (await call(mcp, "tools/call", { name: "invoke_planner", arguments: exact })).result.isError, ).toBeUndefined(); expect(received).toEqual([exact]); + let system = { ...valid, as: "system" }; + expect( + (await call(mcp, "tools/call", { name: "invoke_planner", arguments: system })).result.isError, + ).toBeUndefined(); + expect(received).toEqual([exact, system]); }); test("invoke_planner exposes each refusal as an object code", async () => { @@ -106,6 +113,8 @@ test("invoke_planner exposes each refusal as an object code", async () => { "checkout-unverified", "planner-unavailable", "planner-queue-full", + "system-turn-forbidden", + "system-turn-limited", ] as const ) { let mcp = endpoint({ invoke: async () => ({ kind: "refused", code }) }); @@ -137,7 +146,7 @@ test("invoke_planner carries a refusal's message beside its code", async () => { }); /** A root and child, with `invokePlanner` routed as `main.ts` routes it. */ -async function routed() { +async function routed(options: { plannerAccount?: string; now?: () => number } = {}) { let fixture = await rooted({ root: "# Root\n\nRoot prose.\n", child: "# Child\n\nFinding.\n" }); cleanups.push(fixture.close); let opened: string[] = []; @@ -151,9 +160,20 @@ async function routed() { if (rootId !== fixture.root.channel.id) throw new Error(`unexpected room ${rootId}`); return { context: fixture.root.context, release: () => released++ }; }, + plannerAccount: () => options.plannerAccount, + systemTurns: new SystemTurns(options.now), }; - let invoke = async (channelId: string, repositoryId = fixture.repository.id) => { + let invoke = async ( + channelId: string, + repositoryId = fixture.repository.id, + as: { + user?: { id: string; login: string }; + system?: boolean; + instruction?: string; + } = {}, + ) => { let channel = (await fixture.storage.channels.get(channelId))!; + let user = as.user ?? fixture.user; return invokeRouted(route, { channel, repository: { @@ -161,13 +181,88 @@ async function routed() { id: repositoryId, permissions: { push: true, pull: true, admin: false }, }, - user: { id: fixture.user.id, login: fixture.user.login, avatarUrl: "" }, - instruction: "Summarize the open findings.", + user: { id: user.id, login: user.login, avatarUrl: "" }, + instruction: as.instruction ?? "Summarize the open findings.", + ...(as.system ? { system: true } : {}), } as never); }; return { ...fixture, invoke, opened, released: () => released }; } +const RIVAL = { id: "U_rival", login: "rival" }; +const FOLLOW_UP = "A workflow run finished. Update the document from its results."; + +test("a system invocation starts a Planner turn with its instruction behind a system notice", async () => { + let { invoke, root, turns } = await routed(); + let system = { system: true, instruction: FOLLOW_UP }; + expect(await invoke(root.channel.id, undefined, system)).toBeUndefined(); + await turns.started(); + expect(turns.prompts.at(-1)).toContain(FOLLOW_UP); + expect(root.plan.chat.entries).toEqual([ + expect.objectContaining({ + author: { kind: "system" }, + text: "Automatic follow-up requested by @ana.", + }), + ]); + expect(root.plan.chat.entries.some(entry => entry.text.includes(FOLLOW_UP))).toBe(false); +}); + +test("a system invocation behind a running turn waits out of the visible queue", async () => { + let { events, invoke, released, root, turns, user } = await routed(); + expect(await Chat.invoke(root.context, user, "First")).toBeUndefined(); + await turns.started(1); + expect(await invoke(root.channel.id, undefined, { system: true, instruction: FOLLOW_UP })) + .toBeUndefined(); + let queues = events.filter(event => event.frame.kind === "chat:queue"); + expect(JSON.stringify(queues)).not.toContain(FOLLOW_UP); + + await turns.finish(); + await turns.started(2); + expect(turns.prompts[1]).toContain(FOLLOW_UP); + expect(released()).toBe(0); + await turns.finish(); + await root.plan.chat.running; + expect(released()).toBe(1); +}); + +test("only the document's creator or the account holding its Planner may start a system turn", async () => { + let refused = await routed(); + let system = { user: RIVAL, system: true, instruction: FOLLOW_UP }; + expect(await refused.invoke(refused.root.channel.id, undefined, system)) + .toBe("system-turn-forbidden"); + expect(await refused.invoke(refused.child.channel.id, undefined, system)) + .toBe("system-turn-forbidden"); + expect(refused.root.plan.chat.entries).toEqual([]); + expect(refused.opened).toEqual([]); + + let holder = await routed({ plannerAccount: RIVAL.id }); + expect(await holder.invoke(holder.root.channel.id, undefined, system)).toBeUndefined(); + await holder.turns.started(); + expect(holder.root.plan.chat.entries.at(-1)).toMatchObject({ + author: { kind: "system" }, + text: "Automatic follow-up requested by @rival.", + }); +}); + +test("system turns are limited per document per hour", async () => { + let now = 0; + let { invoke, root, turns } = await routed({ now: () => now }); + let system = { system: true, instruction: FOLLOW_UP }; + for (let i = 0; i < MAX_SYSTEM_TURNS_PER_HOUR; i++) { + expect(await invoke(root.channel.id, undefined, system)).toBeUndefined(); + } + let notices = root.plan.chat.entries.length; + expect(await invoke(root.channel.id, undefined, system)).toBe("system-turn-limited"); + expect(root.plan.chat.entries).toHaveLength(notices); + // An ordinary instruction is not counted against the limit. + expect(await invoke(root.channel.id)).toBeUndefined(); + + now += 60 * 60_000; + turns.finishAll(); + await root.plan.chat.running; + expect(await invoke(root.channel.id, undefined, system)).toBeUndefined(); +}); + test("invoke_planner on a child posts to the root's Chat with the child as focus", async () => { let { child, invoke, opened, root, turns } = await routed(); expect(await invoke(child.channel.id)).toBeUndefined(); diff --git a/apps/server/src/mcp/invoke.ts b/apps/server/src/mcp/invoke.ts index 172d99180..e09048b43 100644 --- a/apps/server/src/mcp/invoke.ts +++ b/apps/server/src/mcp/invoke.ts @@ -5,6 +5,11 @@ * its instruction runs on its root's, with the child as the turn's focus. The * child's parentage and repository are asked of storage again first; a child * that no longer belongs to a live root in the caller's repository is refused. + * + * A system invocation is an automatic follow-up nobody typed: Chat shows a + * notice naming the caller instead of their message. Only the root's creator, + * or the account whose client holds its Planner, may start one, and each root + * takes a bounded number an hour. */ import { resolveChild } from "../chat/children"; import * as Chat from "../chat/service"; @@ -23,8 +28,44 @@ export type InvocationRoute = { rootId: string, input: PlannerInvocation, ) => Promise<{ context: Chat.Room; release: () => void }>; + /** The account whose client holds the document's Planner, if one does. */ + plannerAccount: (rootId: string) => string | undefined; + systemTurns: SystemTurns; }; +export const MAX_SYSTEM_TURNS_PER_HOUR = 10; +const HOUR_MS = 60 * 60_000; + +/** When each root's recent system turns were accepted, for its hourly limit. */ +export class SystemTurns { + #accepted = new Map(); + #now: () => number; + + constructor(now: () => number = Date.now) { + this.#now = now; + } + + /** Counts one against `rootId`'s limit, or returns false when the limit is spent. */ + take(rootId: string): boolean { + let now = this.#now(); + let recent = (this.#accepted.get(rootId) ?? []).filter(at => now - at < HOUR_MS); + if (recent.length >= MAX_SYSTEM_TURNS_PER_HOUR) { + this.#accepted.set(rootId, recent); + return false; + } + recent.push(now); + this.#accepted.set(rootId, recent); + return true; + } + + /** Gives back the latest one taken, for a turn that was refused after all. */ + refund(rootId: string): void { + let recent = this.#accepted.get(rootId); + recent?.pop(); + if (recent?.length === 0) this.#accepted.delete(rootId); + } +} + export async function invokeRouted( route: InvocationRoute, input: PlannerInvocation, @@ -43,23 +84,43 @@ export async function invokeRouted( if (!child) return "document-unavailable"; focus = { channelId: child.id, title: child.title }; } - let { context, release } = await route.open(rootId, input); + if (input.system) { + if (input.user.id !== root.createdBy && input.user.id !== route.plannerAccount(rootId)) { + console.warn(`[mcp] @${input.user.login} may not start a system turn on ${rootId}`); + return "system-turn-forbidden"; + } + if (!route.systemTurns.take(rootId)) { + console.warn(`[mcp] system turns on ${rootId} are limited; @${input.user.login} refused`); + return "system-turn-limited"; + } + } let code: InvokePlannerError | undefined; try { - // Released once the invoked turn itself ends, even when it waits behind others. - code = await Chat.invoke( - context, - input.user, - input.instruction, - input.checkout, - focus, - release, - ); + let { context, release } = await route.open(rootId, input); + try { + // Released once the invoked turn itself ends, even when it waits behind others. + code = await Chat.invoke( + context, + input.user, + input.instruction, + input.checkout, + focus, + release, + input.system ? `Automatic follow-up requested by @${input.user.login}.` : undefined, + ); + } catch (err) { + release(); + throw err; + } + if (code) release(); } catch (err) { - release(); + if (input.system) route.systemTurns.refund(rootId); throw err; } - if (code) release(); + if (input.system) { + if (code) route.systemTurns.refund(rootId); + else console.info(`[mcp] @${input.user.login} started a system turn on ${rootId}`); + } return code; }); } diff --git a/docs/agent-connection.md b/docs/agent-connection.md index 8bf44461b..c5d68483b 100644 --- a/docs/agent-connection.md +++ b/docs/agent-connection.md @@ -397,7 +397,10 @@ how `invoke_planner` returns it. Once a client is attached, both arrive as fresh ids, such as UUIDs. - `runs` reports the session's workflow runs whenever they change. Report them before `turn-end` so the server keeps the session while runs are live. Reports - only change run cards and Chat notices; they never start a Planner turn. + only change run cards and Chat notices; they never start a Planner turn. To + have the Planner act on a run that ended, call `invoke_planner` with + `as: "system"` and a GitHub token of the account holding the Planner; see + [Automatic follow-ups](local-agent-mcp.md#automatic-follow-ups). - `control-result` answers `stop` or `resume`. - `detach` closes the connection with code 1000. The account keeps the document's Planner for the grace period in [Ownership](#ownership). diff --git a/docs/authentication.md b/docs/authentication.md index 645da9952..54c8a7227 100644 --- a/docs/authentication.md +++ b/docs/authentication.md @@ -313,6 +313,8 @@ lends the caller's bearer to the Planner: the turn runs under the channel's existing Planner owner, and a channel without one is claimed only for the caller's own live browser login, which must already have passed the installation-gated owner checks. +Its `as: "system"` form, an automatic follow-up shown in Chat as a system notice, +is further limited to the document's creator or the account holding its Planner. The authorization-code flow uses state, S256 PKCE, the exact configured callback, and the App client secret. It does not request OAuth scopes; diff --git a/docs/hosted-agent.md b/docs/hosted-agent.md index 43865a066..bbdea7ed7 100644 --- a/docs/hosted-agent.md +++ b/docs/hosted-agent.md @@ -102,6 +102,11 @@ call follows the same rules. Its instruction is posted as the caller's own message; the turn runs under the channel's current owner, whoever that is. A channel without one is claimed for the caller's live browser login, hosted or local, and without such a login the call is refused and nothing is posted. +With `as: "system"` the call is an +[automatic follow-up](local-agent-mcp.md#automatic-follow-ups): Chat shows a +system notice naming the caller instead of their message, and the instruction +goes to the Planner alone. Only the document's creator or the account holding +its Planner may send one, at most 10 an hour per document. Under `HARNESS=remote` only attaching claims ownership. An `@chopin` message or `invoke_planner` call for a document without an attached Planner is refused with `planner-not-attached`, and with one it runs as a turn on that Planner under its diff --git a/docs/local-agent-mcp.md b/docs/local-agent-mcp.md index 6d421510c..92892a763 100644 --- a/docs/local-agent-mcp.md +++ b/docs/local-agent-mcp.md @@ -155,6 +155,31 @@ browser need not already have the document open: Chopin keeps its room alive while the invoked turn and queue run. An interrupted turn is not replayed automatically on restart. +### Automatic follow-ups + +Some instructions are not something anyone said. A client that watched a +workflow run finish, for example, wants the Planner to update the document from +its results. Add `"as": "system"` for those: + +```json +{ + "id": "/documents/octo-org/score/release-readiness", + "instruction": "The workflow run finished. Read its results and update the document.", + "as": "system" +} +``` + +Chat then shows a short system notice, `Automatic follow-up requested by +@octocat.`, instead of a message from you. The instruction goes to the Planner +alone and stays out of the visible queue while it waits. Ownership, +`planner-not-attached`, the queue, `checkout`, and the result work as for any +other instruction. + +Only the document's creator, or the account whose client holds its Planner +under `HARNESS=remote`, may send one; for a child document that means its root. +Each document takes at most 10 an hour, counted in memory until the server +restarts. Chopin logs each one it accepts or refuses. + Refusals return an object with `code`, and `planner-not-attached` also a `message`: @@ -169,6 +194,10 @@ Refusals return an object with `code`, and `planner-not-attached` also a repository origin. - `planner-unavailable`: the Planner is disabled or the room is closing. - `planner-queue-full`: wait for queued work to drain before invoking again. +- `system-turn-forbidden`: `as: "system"` came from someone other than the + document's creator or the account holding its Planner. +- `system-turn-limited`: the document has taken 10 automatic follow-ups in the + past hour; try again later. - `document-unavailable`, `document-archived`, and `repository-forbidden` retain their ordinary document/access meanings.