diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 5dc6887608..6af6c9f98f 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -2620,6 +2620,64 @@ describe("ACPX engine permission-observer wiring", () => { expect(settled).toMatchObject({ toolCallId: "tool-1", sessionId: "backend-session", outcome: "completed" }); expect(events.find((event) => event.type === "acpx.permission_unsettled")).toBeUndefined(); }); + + it("bounds the observer's run-log event count when the agent sends more than 256 permission requests", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const logs: Array<{ stream: string; text: string }> = []; + const REQUEST_COUNT = 300; + const execute = createAcpxEngineExecutor({ + createRuntime: (options) => { + for (let i = 0; i < REQUEST_COUNT; i += 1) { + void options.onPermissionRequest?.( + { + sessionId: "backend-session", + inferredKind: "execute", + raw: { sessionId: "backend-session", toolCall: { toolCallId: `tool-${i}` }, options: [] }, + } as never, + { signal: new AbortController().signal }, + ); + } + return { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: async () => {}, + } as never; + }, + }); + + const result = await execute({ + runId: "run-permission-wiring-bounded", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + + const events = parseLogEvents(logs); + const observed = events.filter((event) => event.type === "acpx.permission_observed"); + const unsettled = events.filter((event) => event.type === "acpx.permission_unsettled"); + const summaries = events.filter((event) => event.type === "acpx.permission_observer_truncated"); + expect(observed.length).toBeLessThanOrEqual(256); + expect(unsettled.length).toBeLessThanOrEqual(256); + expect(summaries).toHaveLength(1); + }); }); describe("findAncestorBin", () => { diff --git a/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts b/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts index 1a4b4cd05b..51a9233a1f 100644 --- a/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts +++ b/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts @@ -296,3 +296,126 @@ describe("createAcpPermissionObserver — ledger lifecycle", () => { expect(events).toHaveLength(0); }); }); + +describe("createAcpPermissionObserver — per-run budgets", () => { + function buildRequestFor(sessionId: string, toolCallId: string): AcpPermissionRequest { + return buildRequest({ + sessionId, + raw: { sessionId, toolCall: { toolCallId }, options: [] } as AcpPermissionRequest["raw"], + }); + } + + it("caps the ledger and the observed-event count when more than 256 requests never settle", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => events.push(event), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + for (let i = 0; i < 300; i += 1) { + await observer.handlePermissionRequest(buildRequestFor(`session-${i}`, `tool-${i}`), { signal }); + } + + const observed = events.filter((event) => event.type === "acpx.permission_observed"); + expect(observed.length).toBeLessThanOrEqual(256); + + events.length = 0; + await observer.finalizeRun(); + const unsettled = events.filter((event) => event.type === "acpx.permission_unsettled"); + expect(unsettled.length).toBeLessThanOrEqual(256); + const summaries = events.filter((event) => event.type === "acpx.permission_observer_truncated"); + expect(summaries).toHaveLength(1); + }); + + it("bounds cumulative emissions at 512 detailed events across more than 1000 open-and-settle cycles", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => events.push(event), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + for (let i = 0; i < 1_100; i += 1) { + const sessionId = `session-${i}`; + const toolCallId = `tool-${i}`; + await observer.handlePermissionRequest(buildRequestFor(sessionId, toolCallId), { signal }); + observer.noteToolCallEvent(sessionId, { toolCallId, status: "completed" }); + } + + const detailed = events.filter((event) => event.type !== "acpx.permission_observer_truncated"); + expect(detailed.length).toBeLessThanOrEqual(512); + + events.length = 0; + await observer.finalizeRun(); + expect(events.filter((event) => event.type === "acpx.permission_observer_truncated")).toHaveLength(1); + }); + + it("keeps the settled-event budget reachable after the observed budget is spent", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => events.push(event), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + const opened: Array<{ sessionId: string; toolCallId: string }> = []; + for (let i = 0; i < 256; i += 1) { + const sessionId = `session-${i}`; + const toolCallId = `tool-${i}`; + opened.push({ sessionId, toolCallId }); + await observer.handlePermissionRequest(buildRequestFor(sessionId, toolCallId), { signal }); + } + // The observed budget is now fully spent. A further request emits no + // "acpx.permission_observed" event. + await observer.handlePermissionRequest(buildRequestFor("session-extra", "tool-extra"), { signal }); + expect(events.filter((event) => event.type === "acpx.permission_observed")).toHaveLength(256); + + events.length = 0; + for (const { sessionId, toolCallId } of opened) { + observer.noteToolCallEvent(sessionId, { toolCallId, status: "completed" }); + } + + const settled = events.filter((event) => event.type === "acpx.permission_settled"); + expect(settled.length).toBe(256); + }); + + it("emits a summary event with only the three counters and the type, using an exact key match", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => events.push(event), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + for (let i = 0; i < 300; i += 1) { + await observer.handlePermissionRequest(buildRequestFor(`session-${i}`, `tool-${i}`), { signal }); + } + + events.length = 0; + await observer.finalizeRun(); + const summary = events.find((event) => event.type === "acpx.permission_observer_truncated"); + expect(summary).toBeDefined(); + expect(Object.keys(summary as object).sort()).toEqual( + ["suppressedLedgerEntries", "suppressedObservedEvents", "suppressedTerminalEvents", "type"].sort(), + ); + expect(typeof summary?.suppressedLedgerEntries).toBe("number"); + expect(typeof summary?.suppressedObservedEvents).toBe("number"); + expect(typeof summary?.suppressedTerminalEvents).toBe("number"); + }); + + it("still resolves handlePermissionRequest to undefined, and throws nothing, once every cap is reached", async () => { + const observer = createAcpPermissionObserver({ + emitLog: () => {}, + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + for (let i = 0; i < 300; i += 1) { + await observer.handlePermissionRequest(buildRequestFor(`session-${i}`, `tool-${i}`), { signal }); + } + const result = await observer.handlePermissionRequest(buildRequestFor("session-last", "tool-last"), { signal }); + expect(result).toBeUndefined(); + await expect(observer.finalizeRun()).resolves.toBeUndefined(); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/permission-observer.ts b/packages/adapter-utils/src/acpx-engine/permission-observer.ts index f17e33c001..7da8f15573 100644 --- a/packages/adapter-utils/src/acpx-engine/permission-observer.ts +++ b/packages/adapter-utils/src/acpx-engine/permission-observer.ts @@ -13,6 +13,22 @@ import type { AcpPermissionDecision, AcpPermissionRequest } from "acpx/runtime"; const MAX_IDENTIFIER_LENGTH = 200; const MAX_AGE_MS = 24 * 60 * 60 * 1000; +// The agent process is untrusted. It picks how many permission requests it +// sends, so the observer must cap its own memory and log volume against that +// input. Each run gets its own budget: every counter below lives inside the +// observer closure, not at module scope. +// +// The ledger, the "observed" count, and the "settled or unsettled" count each +// get a separate budget of 256, instead of one shared budget of 512. A shared +// budget lets an agent that sends 512 requests spend the whole budget on +// "observed" events, so the observer would then emit no "unsettled" event at +// finalization — the one signal this observer exists to produce. Two budgets +// keep that signal reachable no matter how the agent spends the "observed" +// side. +const MAX_LEDGER_ENTRIES = 256; +const MAX_OBSERVED_EVENTS = 256; +const MAX_TERMINAL_EVENTS = 256; + export const PERMISSION_OBSERVER_METHODS = ["session/request_permission"] as const; export type PermissionObserverMethod = (typeof PERMISSION_OBSERVER_METHODS)[number] | "unknown"; @@ -106,7 +122,11 @@ export interface PermissionObserverToolCallEvent { } export interface PermissionObserverLogEvent { - type: "acpx.permission_observed" | "acpx.permission_settled" | "acpx.permission_unsettled"; + type: + | "acpx.permission_observed" + | "acpx.permission_settled" + | "acpx.permission_unsettled" + | "acpx.permission_observer_truncated"; [key: string]: unknown; } @@ -155,6 +175,15 @@ export function createAcpPermissionObserver(options: AcpPermissionObserverOption const permissionMode = mapPermissionObserverPermissionMode(options.permissionMode); const transport = mapPermissionObserverTransport(options.transport); + // Per-run budget state. Each counter tracks a cumulative count of emitted + // events, not the live ledger size, so an agent cannot reset a counter by + // opening and settling requests in a loop (the churn case). + let observedEventCount = 0; + let terminalEventCount = 0; + let suppressedLedgerEntries = 0; + let suppressedObservedEvents = 0; + let suppressedTerminalEvents = 0; + const ledgerKey = (sessionId: string, toolCallId: string) => sessionId + "\u0000" + toolCallId; const handlePermissionRequest: AcpPermissionObserver["handlePermissionRequest"] = async (request) => { @@ -170,28 +199,42 @@ export function createAcpPermissionObserver(options: AcpPermissionObserverOption if (typeof rawSessionId === "string" && typeof rawToolCallId === "string") { const key = ledgerKey(sessionId, toolCallId); if (!ledger.has(key)) { - // The ledger entry always opens as "requested": the observer has not - // yet seen a settlement for this tool call, whatever status the - // permission request itself carried. - ledger.set(key, { - sessionId, - toolCallId, - openedAtMs: now(), - toolKind, - lastStage: "requested", - }); + if (ledger.size < MAX_LEDGER_ENTRIES) { + // The ledger entry always opens as "requested": the observer has + // not yet seen a settlement for this tool call, whatever status + // the permission request itself carried. + ledger.set(key, { + sessionId, + toolCallId, + openedAtMs: now(), + toolKind, + lastStage: "requested", + }); + } else { + suppressedLedgerEntries += 1; + } } } - options.emitLog({ - type: "acpx.permission_observed", - sessionId, - toolCallId, - method: mapPermissionObserverMethod("session/request_permission"), - toolKind, - stage: "requested", - permissionMode, - transport, - }); + // Count every emission against the budget, not the live ledger size, so + // an agent cannot refill the budget by settling old requests. + if (observedEventCount < MAX_OBSERVED_EVENTS) { + observedEventCount += 1; + options.emitLog({ + type: "acpx.permission_observed", + sessionId, + toolCallId, + method: mapPermissionObserverMethod("session/request_permission"), + toolKind, + stage: "requested", + permissionMode, + transport, + }); + } else { + // Emit nothing, not even a reduced event: that keeps every + // attacker-controlled identifier out of the log once the budget runs + // out. + suppressedObservedEvents += 1; + } } catch { // The observer is diagnostic only. An internal error here must never // affect the permission handoff, so every failure resolves the same @@ -212,14 +255,22 @@ export function createAcpPermissionObserver(options: AcpPermissionObserverOption entry.lastStage = stage; const outcome = mapPermissionObserverOutcome(event?.status); if (outcome === "unknown") return; + // Delete the entry (freeing the ledger memory) even when the terminal + // budget below is spent: the ledger must not hold a settled entry just + // because the observer could not log its settlement. ledger.delete(key); - options.emitLog({ - type: "acpx.permission_settled", - sessionId: entry.sessionId, - toolCallId: entry.toolCallId, - outcome, - ageMs: boundedAgeMs(now() - entry.openedAtMs), - }); + if (terminalEventCount < MAX_TERMINAL_EVENTS) { + terminalEventCount += 1; + options.emitLog({ + type: "acpx.permission_settled", + sessionId: entry.sessionId, + toolCallId: entry.toolCallId, + outcome, + ageMs: boundedAgeMs(now() - entry.openedAtMs), + }); + } else { + suppressedTerminalEvents += 1; + } } catch { // Diagnostic only; never let a logging failure surface into the event // loop that drains the turn's tool-call events. @@ -230,12 +281,29 @@ export function createAcpPermissionObserver(options: AcpPermissionObserverOption const openEntries = [...ledger.values()]; ledger.clear(); for (const entry of openEntries) { + if (terminalEventCount < MAX_TERMINAL_EVENTS) { + terminalEventCount += 1; + options.emitLog({ + type: "acpx.permission_unsettled", + sessionId: entry.sessionId, + toolCallId: entry.toolCallId, + stage: entry.lastStage, + ageMs: boundedAgeMs(now() - entry.openedAtMs), + }); + } else { + suppressedTerminalEvents += 1; + } + } + // Emit one summary event for the whole run, and only when the observer + // suppressed something. It carries only the three counters: no session + // identifier, no tool-call identifier, and no other agent-controlled + // value. + if (suppressedLedgerEntries > 0 || suppressedObservedEvents > 0 || suppressedTerminalEvents > 0) { options.emitLog({ - type: "acpx.permission_unsettled", - sessionId: entry.sessionId, - toolCallId: entry.toolCallId, - stage: entry.lastStage, - ageMs: boundedAgeMs(now() - entry.openedAtMs), + type: "acpx.permission_observer_truncated", + suppressedLedgerEntries, + suppressedObservedEvents, + suppressedTerminalEvents, }); } };