fix(acpx-engine): bound the permission observer's ledger and log volume

The agent process is untrusted. It can send an unbounded number of
session/request_permission frames, and each one grew the observer's
ledger and produced one run-log event. Add a fixed per-run budget: 256
ledger entries, 256 observed events, and 256 settled-or-unsettled
events. The ledger entry and the observed event use separate budgets
so an agent cannot spend the whole budget on one event type and starve
the other. A run that hits any limit emits one summary event at
finalization with the three suppressed counts, and no other value.

Co-authored-by: Paperclip <noreply@paperclip.ing>

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", () => {

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();
+  });
+});

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,
       });
     }
   };