diff --git a/doc/run-log-events.md b/doc/run-log-events.md index c65aab2775..72d986a143 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -146,6 +146,99 @@ The payload never carries a command, an argument, a path, an environment value, or a raw identifier. The event rides the `ctx.onEvent` run-event bridge and is run-log-only. It needs no OTLP endpoint. +## ACP Permission Handoff Observer Run-Log Events + +Paperclip writes four run-log event types for the ACP permission handoff. The +producer is `createAcpPermissionObserver` in +`packages/adapter-utils/src/acpx-engine/permission-observer.ts`. These events +are run-log records, not first-party telemetry events. The generated +telemetry contract does not cover them, so this section is their canonical +contract. + +The observer is observation-only. It never answers, approves, or denies a +permission request. It only records receipt and settlement, so an operator +can tell a stalled handoff from a normal wait. An internal error in the +observer resolves the hook to `undefined`, the same as a normal observation; +the observer never blocks or changes the permission decision. + +Each field passes through a closed enumeration or a bounded scalar. The +event never carries the raw ACP frame, the tool input, or another free-form +payload. A session ID and a tool-call ID are capped at 200 characters each +before they enter the payload. + +### `acpx.permission_observed` + +Paperclip writes this event when the engine receives a permission request. + +| Field | Type | Meaning | +| --- | --- | --- | +| `sessionId` | string | The session ID, capped at 200 characters. | +| `toolCallId` | string | The tool-call ID, capped at 200 characters. | +| `method` | string | The ACP method name, from a closed allowlist (`session/request_permission` or `unknown`). | +| `toolKind` | string | The inferred tool kind, from a closed allowlist (`read`, `edit`, `delete`, `move`, `search`, `execute`, `think`, `fetch`, `switch_mode`, `other`, or `unknown`). | +| `stage` | string | Always `requested` for this event. | +| `permissionMode` | string | The run's effective permission mode, from a closed allowlist (`approve-all`, `approve-reads`, `deny-all`, or `unknown`). | +| `transport` | string | The run's execution transport, from a closed allowlist (`local`, `ssh`, `sandbox`, or `unknown`). | + +### `acpx.permission_settled` + +Paperclip writes this event when a tracked tool call reaches a terminal +status (`completed` or `failed`). + +| Field | Type | Meaning | +| --- | --- | --- | +| `sessionId` | string | The session ID, capped at 200 characters. | +| `toolCallId` | string | The tool-call ID, capped at 200 characters. | +| `outcome` | string | The terminal outcome (`completed` or `failed`). | +| `ageMs` | number | The time from receipt to settlement, in milliseconds, clamped to a maximum of 24 hours. | + +### `acpx.permission_unsettled` + +Paperclip writes one of these events for each permission request still open +when the run finalizes. This is the signal that a handoff stalled. + +| Field | Type | Meaning | +| --- | --- | --- | +| `sessionId` | string | The session ID, capped at 200 characters. | +| `toolCallId` | string | The tool-call ID, capped at 200 characters. | +| `stage` | string | The last known tool-call stage, from a closed allowlist (`requested`, `pending`, `in_progress`, `completed`, `failed`, or `unknown`). | +| `ageMs` | number | The time from receipt to run finalization, in milliseconds, clamped to a maximum of 24 hours. | + +### `acpx.permission_observer_truncated` + +Paperclip writes this event once per run, only when the observer suppressed +an entry or an event. It carries no session ID, no tool-call ID, and no other +agent-controlled value. + +| Field | Type | Meaning | +| --- | --- | --- | +| `suppressedLedgerEntries` | number | The count of open permission requests the observer could not track because the ledger was full. | +| `suppressedObservedEvents` | number | The count of `acpx.permission_observed` events the observer dropped because that event's own budget was full. | +| `suppressedSettledEvents` | number | The count of `acpx.permission_settled` events the observer dropped because that event's own budget was full. | +| `suppressedUnsettledEvents` | number | The count of `acpx.permission_unsettled` events the observer dropped because that event's own budget was full. | + +### Bounded output + +The agent process is untrusted, so it picks how many permission requests it +sends. The observer bounds its own memory and log volume against that input +instead of trusting a limit the agent could exceed. + +The observer tracks open requests in a ledger capped at 256 entries. It also +gives each event type its own emission budget of 256 events for the run. The +four budgets are separate: `acpx.permission_settled` fires once per tool call +the agent completes, so a normal long run can spend a shared budget before +the run ends. A separate budget for `acpx.permission_unsettled` keeps that +signal reachable even when the agent's normal traffic would otherwise spend +a shared budget first. + +Each budget counts the cumulative number of emitted events for the run, not +the live ledger size, so an agent cannot refill a budget by opening and +settling requests in a loop. When a budget is spent, the observer emits +nothing further for that event type; it never emits a reduced event. The run +finalization step emits at most one `acpx.permission_unsettled` event per +still-open ledger entry, and at most one `acpx.permission_observer_truncated` +event for the whole run. + ## Related instrumentation The sandbox duplex transport also writes one run-log event as one of its three diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index a91503b9c9..cebd4a6580 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -2562,6 +2562,198 @@ describe("shared ACPX engine runtime behavior", () => { }); }); +describe("ACPX engine permission-observer wiring", () => { + function parseLogEvents(logs: Array<{ stream: string; text: string }>): Array> { + return logs + .filter((entry) => entry.stream === "stdout") + .map((entry) => { + try { + return JSON.parse(entry.text) as Record; + } catch { + return null; + } + }) + .filter((parsed): parsed is Record => parsed !== null); + } + + it("passes the observer's hook as onPermissionRequest, and the settle step reports a request the turn never saw settle", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const logs: Array<{ stream: string; text: string }> = []; + const runtimeOptions: Array<{ onPermissionRequest?: (...args: unknown[]) => unknown }> = []; + const execute = createAcpxEngineExecutor({ + createRuntime: (options) => { + runtimeOptions.push(options as unknown as { onPermissionRequest?: (...args: unknown[]) => unknown }); + // The engine wires this before it calls createRuntime, so calling it + // here reproduces the same order acpx uses: request arrives, then the + // turn's tool-call events drain (or, as here, never arrive). + void options.onPermissionRequest?.( + { + sessionId: "backend-session", + inferredKind: "execute", + raw: { sessionId: "backend-session", toolCall: { toolCallId: "tool-never-closes" }, 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-unsettled", + 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); + expect(typeof runtimeOptions[0]?.onPermissionRequest).toBe("function"); + + const events = parseLogEvents(logs); + const observed = events.find((event) => event.type === "acpx.permission_observed"); + expect(observed).toMatchObject({ toolCallId: "tool-never-closes", sessionId: "backend-session" }); + const unsettled = events.find((event) => event.type === "acpx.permission_unsettled"); + expect(unsettled).toMatchObject({ toolCallId: "tool-never-closes", sessionId: "backend-session" }); + }); + + it("settles a permission request when the tool-call drain loop reports a matching terminal event", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const logs: Array<{ stream: string; text: string }> = []; + const execute = createAcpxEngineExecutor({ + createRuntime: (options) => { + void options.onPermissionRequest?.( + { + sessionId: "backend-session", + inferredKind: "execute", + raw: { sessionId: "backend-session", toolCall: { toolCallId: "tool-1" }, 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: "tool_call", + text: "Bash (completed)", + title: "Bash", + status: "completed", + toolCallId: "tool-1", + tag: "tool_call", + }; + 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-settled", + 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 settled = events.find((event) => event.type === "acpx.permission_settled"); + 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", () => { async function writeFakeBin(dir: string, name: string) { const binDir = path.join(dir, "node_modules", ".bin"); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index e61bf94ae2..f4668215fd 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -117,6 +117,7 @@ import type { TurnCompletion, } from "./run-contracts.js"; import { createRunResourceLedger } from "./run-resource-ledger.js"; +import { createAcpPermissionObserver, type AcpPermissionObserver } from "./permission-observer.js"; import { settleAcpRun, type SettlementSteps } from "./settlement-sequence.js"; import { runAttempt, @@ -3998,6 +3999,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { let prepared!: AcpxPreparedRuntime; let runtime!: AcpRuntime; let sessionHandle!: AcpRuntimeHandle; + // Observes the ACP permission handoff for the run log; it never answers + // a permission request. Assigned once `prepared` is ready (it reads the + // effective permission mode and the execution transport), so it stays + // undefined on a build failure that never reaches that point — the + // ledger would hold no entries at that point anyway. + let permissionObserver: AcpPermissionObserver | undefined; let childStderrState!: ChildStderrState; let processIdentitySink!: AcpxProcessIdentitySink; let resumedSession = false; @@ -4129,6 +4136,21 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }), ); buildRuntimeSettled = true; + const observedExecutionTarget = ctx.executionTarget; + permissionObserver = createAcpPermissionObserver({ + permissionMode: prepared.permissionMode, + // A missing execution target means the local environment, the same + // convention `describeAdapterExecutionTarget` already uses. + transport: !observedExecutionTarget || observedExecutionTarget.kind === "local" + ? "local" + : observedExecutionTarget.transport, + // Return the write's promise; do not swallow it here. The observer + // never awaits this on the permission critical path, but it keeps + // the promise so `finalizeRun` can drain every pending write before + // it returns, so the run never finalizes with a write still in + // flight. + emitLog: (payload) => emitAcpxLog(ctx, payload), + }); // Capture the run's staging lease release now that the runtime built. The // run root `finally` releases it as the final settlement act. releaseStagingLease = prepared.sessionStagingLeaseRelease; @@ -4215,6 +4237,9 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { agentRegistry: prepared.agentRegistry, permissionMode: prepared.permissionMode, nonInteractivePermissions: prepared.nonInteractivePermissions, + // Observation only: this hook always resolves to `undefined`, so it + // never changes which permission option acpx resolves on its own. + onPermissionRequest: permissionObserver?.handlePermissionRequest, mcpServers: prepared.mcpServers, timeoutMs: prepared.timeoutSec > 0 ? prepared.timeoutSec * 1000 : undefined, // Scope ACPX runtime verbose logs to the claude agent only. Codex @@ -4813,6 +4838,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { kind: event.kind ?? previous?.kind, status: event.status ?? previous?.status, }); + permissionObserver?.noteToolCallEvent(sessionHandle.backendSessionId, { + toolCallId: event.toolCallId, + status: event.status, + }); } } if (event.type === "text_delta" && event.stream !== "thought") { @@ -5333,6 +5362,11 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const report = await settleAcpRun(runResourceLedger, cause, settlementSteps); recordDispositionReport(report); if (childStderrState) flushChildStderr(childStderrState); + // Report every permission request the run never saw settle. Not on + // the permission critical path. This call waits for every queued + // log write, including one still in flight from an earlier + // permission event, to reach durable storage before it returns. + await permissionObserver?.finalizeRun(); }, reproduceResult: async (): Promise => { if (!capturedResult) { diff --git a/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts b/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts new file mode 100644 index 0000000000..bc41f57989 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/permission-observer.test.ts @@ -0,0 +1,580 @@ +import { describe, expect, it } from "vitest"; +import type { AcpPermissionRequest } from "acpx/runtime"; +import { + createAcpPermissionObserver, + mapPermissionObserverMethod, + mapPermissionObserverOutcome, + mapPermissionObserverToolKind, + type PermissionObserverLogEvent, +} from "./permission-observer.js"; + +function buildRequest(overrides: Partial = {}): AcpPermissionRequest { + return { + sessionId: "session-1", + inferredKind: "execute", + raw: { + sessionId: "session-1", + toolCall: { toolCallId: "tool-1" }, + options: [], + }, + ...overrides, + } as AcpPermissionRequest; +} + +describe("permission-observer closed-enum mappers", () => { + it("maps a recognized method and returns 'unknown' for anything else", () => { + expect(mapPermissionObserverMethod("session/request_permission")).toBe("session/request_permission"); + expect(mapPermissionObserverMethod("session/other_method")).toBe("unknown"); + expect(mapPermissionObserverMethod(undefined)).toBe("unknown"); + expect(mapPermissionObserverMethod(42)).toBe("unknown"); + }); + + it("maps a recognized tool kind and returns 'unknown' for anything else", () => { + expect(mapPermissionObserverToolKind("execute")).toBe("execute"); + expect(mapPermissionObserverToolKind("teleport")).toBe("unknown"); + expect(mapPermissionObserverToolKind(undefined)).toBe("unknown"); + }); + + it("maps a terminal outcome and returns 'unknown' for a non-terminal or unrecognized status", () => { + expect(mapPermissionObserverOutcome("completed")).toBe("completed"); + expect(mapPermissionObserverOutcome("failed")).toBe("failed"); + expect(mapPermissionObserverOutcome("pending")).toBe("unknown"); + expect(mapPermissionObserverOutcome("something-else")).toBe("unknown"); + }); +}); + +describe("createAcpPermissionObserver — handlePermissionRequest", () => { + it("resolves to undefined for a normal request", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const result = await observer.handlePermissionRequest(buildRequest(), { signal: new AbortController().signal }); + expect(result).toBeUndefined(); + }); + + it("resolves to undefined for a request that carries unknown fields", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const request = buildRequest({ + inferredKind: "not-a-real-kind" as unknown as AcpPermissionRequest["inferredKind"], + raw: { + sessionId: "session-1", + toolCall: { toolCallId: "tool-1" }, + options: [], + _meta: { anExtraField: "some value nobody declared" }, + } as AcpPermissionRequest["raw"], + }); + const result = await observer.handlePermissionRequest(request, { signal: new AbortController().signal }); + expect(result).toBeUndefined(); + }); + + it("resolves to undefined even when the injected log sink throws", async () => { + const observer = createAcpPermissionObserver({ + emitLog: () => { + throw new Error("sink failure"); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const result = await observer.handlePermissionRequest(buildRequest(), { signal: new AbortController().signal }); + expect(result).toBeUndefined(); + }); + + it("never awaits the log write on the hook path", async () => { + let logCalled = false; + const observer = createAcpPermissionObserver({ + // A log sink that returns a promise which never resolves. If the hook + // awaited it internally, the assertion below would never run. + emitLog: () => { + logCalled = true; + return new Promise(() => {}) as unknown as void; + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + let settled = false; + const promise = observer.handlePermissionRequest(buildRequest(), { signal: new AbortController().signal }); + void promise.then(() => { + settled = true; + }); + // Flush one microtask turn. A synchronous hook body resolves within it; + // an awaited I/O call would not. + await Promise.resolve(); + await Promise.resolve(); + expect(settled).toBe(true); + expect(logCalled).toBe(true); + }); + + it("emits only allow-listed scalar fields, never the raw request payload", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const longString = "x".repeat(10_000); + const request = buildRequest({ + raw: { + sessionId: "session-1", + toolCall: { + toolCallId: "tool-1", + rawInput: { command: "rm -rf /", nested: { secret: "s3cr3t" } }, + }, + options: [], + _meta: { note: longString }, + } as unknown as AcpPermissionRequest["raw"], + }); + (request as unknown as { error: unknown }).error = { data: { stack: "leaked stack trace" } }; + await observer.handlePermissionRequest(request, { signal: new AbortController().signal }); + + expect(events).toHaveLength(1); + const [event] = events; + const serialized = JSON.stringify(event); + expect(serialized).not.toContain("rm -rf"); + expect(serialized).not.toContain("s3cr3t"); + expect(serialized).not.toContain("leaked stack trace"); + expect(serialized).not.toContain(longString); + expect(Object.keys(event).sort()).toEqual( + ["method", "permissionMode", "sessionId", "stage", "toolCallId", "toolKind", "transport", "type"].sort(), + ); + }); + + it("maps an unmapped tool kind to exactly 'unknown', dropping the source string", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const request = buildRequest({ inferredKind: "levitate" as unknown as AcpPermissionRequest["inferredKind"] }); + await observer.handlePermissionRequest(request, { signal: new AbortController().signal }); + expect(events[0]?.toolKind).toBe("unknown"); + expect(JSON.stringify(events[0])).not.toContain("levitate"); + }); +}); + +describe("createAcpPermissionObserver — ledger lifecycle", () => { + it("opens an entry on receipt, closes it on the terminal tool_call_update, and reports the age", async () => { + let clock = 1_000; + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + now: () => clock, + }); + await observer.handlePermissionRequest(buildRequest(), { signal: new AbortController().signal }); + clock += 4_200; + observer.noteToolCallEvent("session-1", { toolCallId: "tool-1", status: "completed" }); + + const settled = events.find((event) => event.type === "acpx.permission_settled"); + expect(settled).toMatchObject({ + type: "acpx.permission_settled", + sessionId: "session-1", + toolCallId: "tool-1", + outcome: "completed", + ageMs: 4_200, + }); + + // A settled entry must not resurface at finalization. + events.length = 0; + await observer.finalizeRun(); + expect(events).toHaveLength(0); + }); + + it("reports one unsettled event per still-open entry at run finalization, with the last observed stage", async () => { + let clock = 0; + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + now: () => clock, + }); + await observer.handlePermissionRequest(buildRequest(), { signal: new AbortController().signal }); + await observer.handlePermissionRequest( + buildRequest({ + raw: { sessionId: "session-1", toolCall: { toolCallId: "tool-2" }, options: [] } as AcpPermissionRequest["raw"], + }), + { signal: new AbortController().signal }, + ); + // tool-1 advances to "in_progress" but never reaches a terminal status. + observer.noteToolCallEvent("session-1", { toolCallId: "tool-1", status: "in_progress" }); + clock = 27 * 60 * 1000; + + await observer.finalizeRun(); + + const unsettled = events.filter((event) => event.type === "acpx.permission_unsettled"); + expect(unsettled).toHaveLength(2); + const byToolCallId = Object.fromEntries(unsettled.map((event) => [event.toolCallId, event])); + expect(byToolCallId["tool-1"]).toMatchObject({ stage: "in_progress", ageMs: 27 * 60 * 1000 }); + expect(byToolCallId["tool-2"]).toMatchObject({ stage: "requested", ageMs: 27 * 60 * 1000 }); + + // Finalization drains the ledger; a second call reports nothing new. + events.length = 0; + await observer.finalizeRun(); + expect(events).toHaveLength(0); + }); + + it("does not close an entry on a non-terminal tool_call status", () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + observer.noteToolCallEvent("session-1", { toolCallId: "tool-1", status: "pending" }); + expect(events.filter((event) => event.type === "acpx.permission_settled")).toHaveLength(0); + }); + + it("ignores a tool_call event for a toolCallId it never opened", () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + observer.noteToolCallEvent("session-1", { toolCallId: "never-opened", status: "completed" }); + expect(events).toHaveLength(0); + }); + + it("does not open a ledger entry for a request with no tool-call identifier", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const request = buildRequest({ + raw: { sessionId: "session-1", toolCall: {}, options: [] } as unknown as AcpPermissionRequest["raw"], + }); + await observer.handlePermissionRequest(request, { signal: new AbortController().signal }); + + const observed = events.filter((event) => event.type === "acpx.permission_observed"); + expect(observed).toHaveLength(1); + expect(observed[0]).toMatchObject({ toolCallId: "unknown" }); + + events.length = 0; + await observer.finalizeRun(); + expect(events).toHaveLength(0); + }); + + it("does not let two requests with a missing session identifier share one ledger entry", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const firstRequest = buildRequest({ + sessionId: undefined as unknown as string, + raw: { toolCall: { toolCallId: "tool-1" }, options: [] } as unknown as AcpPermissionRequest["raw"], + }); + const secondRequest = buildRequest({ + sessionId: undefined as unknown as string, + raw: { toolCall: { toolCallId: "tool-2" }, options: [] } as unknown as AcpPermissionRequest["raw"], + }); + await observer.handlePermissionRequest(firstRequest, { signal: new AbortController().signal }); + await observer.handlePermissionRequest(secondRequest, { signal: new AbortController().signal }); + + // Neither request carried a real session identifier, so neither one opened + // a ledger entry. A terminal tool_call event for either tool call must + // find nothing to settle. + observer.noteToolCallEvent(undefined, { toolCallId: "tool-1", status: "completed" }); + observer.noteToolCallEvent(undefined, { toolCallId: "tool-2", status: "completed" }); + expect(events.filter((event) => event.type === "acpx.permission_settled")).toHaveLength(0); + + events.length = 0; + await observer.finalizeRun(); + 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 settled emissions at 256 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 settled = events.filter((event) => event.type === "acpx.permission_settled"); + expect(settled.length).toBeLessThanOrEqual(256); + + events.length = 0; + await observer.finalizeRun(); + expect(events.filter((event) => event.type === "acpx.permission_observer_truncated")).toHaveLength(1); + }); + + it("keeps the unsettled-event budget reachable after a normal run spends the settled budget", async () => { + const events: PermissionObserverLogEvent[] = []; + const observer = createAcpPermissionObserver({ + emitLog: (event) => { + events.push(event); + }, + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + // A normal run: 256 requests open and settle during the turn. + for (let i = 0; i < 256; i += 1) { + const sessionId = `session-${i}`; + const toolCallId = `tool-${i}`; + await observer.handlePermissionRequest(buildRequestFor(sessionId, toolCallId), { signal }); + observer.noteToolCallEvent(sessionId, { toolCallId, status: "completed" }); + } + // One more request opens and never settles. This is the stall. + await observer.handlePermissionRequest(buildRequestFor("session-stall", "tool-stall"), { signal }); + + events.length = 0; + await observer.finalizeRun(); + const unsettled = events.filter((event) => event.type === "acpx.permission_unsettled"); + expect(unsettled).toHaveLength(1); + expect(unsettled[0]).toMatchObject({ toolCallId: "tool-stall" }); + }); + + it("bounds cumulative unsettled emissions at 256 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 }); + } + + events.length = 0; + await observer.finalizeRun(); + const unsettled = events.filter((event) => event.type === "acpx.permission_unsettled"); + expect(unsettled.length).toBeLessThanOrEqual(256); + }); + + it("emits exactly one truncated summary event across two finalizeRun calls", 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(); + 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 four 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", + "suppressedSettledEvents", + "suppressedUnsettledEvents", + "type", + ].sort(), + ); + expect(typeof summary?.suppressedLedgerEntries).toBe("number"); + expect(typeof summary?.suppressedObservedEvents).toBe("number"); + expect(typeof summary?.suppressedSettledEvents).toBe("number"); + expect(typeof summary?.suppressedUnsettledEvents).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(); + }); +}); + +describe("createAcpPermissionObserver — asynchronous log persistence", () => { + it("waits for a write still pending from an earlier event before finalizeRun resolves", async () => { + const durable: PermissionObserverLogEvent[] = []; + const releaseWrite: Array<() => void> = []; + const observer = createAcpPermissionObserver({ + // A durable log sink that starts a write and only completes it when the + // test calls the matching release function. This stands in for a real + // write to storage that takes more than one microtask turn. + emitLog: (event) => + new Promise((resolve) => { + releaseWrite.push(() => { + durable.push(event); + resolve(); + }); + }), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + // This request's "observed" write starts but does not finish. The tool + // call it opened never settles, so the entry is still open at + // finalization. + await observer.handlePermissionRequest(buildRequest(), { signal }); + expect(durable).toHaveLength(0); + + const finalizePromise = observer.finalizeRun(); + // finalizeRun has started its own "unsettled" write for the still-open + // entry. That write is also pending. Flush a few microtask turns: if + // finalizeRun dropped either write's promise, it would resolve here even + // though neither write has completed. + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + expect(durable).toHaveLength(0); + + // Release both writes, the one queued before finalizeRun ran and the one + // finalizeRun queued itself. + for (const release of releaseWrite) release(); + await finalizePromise; + + expect(durable.map((event) => event.type).sort()).toEqual( + ["acpx.permission_observed", "acpx.permission_unsettled"].sort(), + ); + }); + + it("resolves finalizeRun without throwing when a queued write rejects", async () => { + const observer = createAcpPermissionObserver({ + emitLog: () => Promise.reject(new Error("durable sink unavailable")), + permissionMode: "approve-all", + transport: "sandbox", + }); + const signal = new AbortController().signal; + await observer.handlePermissionRequest(buildRequest(), { signal }); + 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 new file mode 100644 index 0000000000..33652b4fb5 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/permission-observer.ts @@ -0,0 +1,359 @@ +import type { AcpPermissionDecision, AcpPermissionRequest } from "acpx/runtime"; + +/** + * Observation-only hook into the ACP permission handoff. It never answers a + * request. It records receipt and settlement in the run log so an operator + * can tell a stalled handoff from a normal wait. + * + * Every emitted field passes through a closed enumeration or a bounded + * scalar. The module never forwards the raw ACP frame, the tool input, or + * any other free-form payload. + */ + +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, the "settled" count, and the "unsettled" +// count each get a separate budget of 256, instead of one shared budget. A +// diagnostic event class must never share a budget with an event class that +// the agent drives. "acpx.permission_settled" fires once for each tool call +// the agent completes, so a normal long run can spend a shared budget before +// the run ends. "acpx.permission_unsettled" is the one signal this observer +// exists to produce: it tells an operator that a handoff stalled. A shared +// budget would let a normal run silence that signal at the exact time it +// matters most. Four separate budgets keep each event class reachable no +// matter how the agent spends the others. +const MAX_LEDGER_ENTRIES = 256; +const MAX_OBSERVED_EVENTS = 256; +const MAX_SETTLED_EVENTS = 256; +const MAX_UNSETTLED_EVENTS = 256; + +export const PERMISSION_OBSERVER_METHODS = ["session/request_permission"] as const; +export type PermissionObserverMethod = (typeof PERMISSION_OBSERVER_METHODS)[number] | "unknown"; + +export const PERMISSION_OBSERVER_TOOL_KINDS = [ + "read", + "edit", + "delete", + "move", + "search", + "execute", + "think", + "fetch", + "switch_mode", + "other", +] as const; +export type PermissionObserverToolKind = (typeof PERMISSION_OBSERVER_TOOL_KINDS)[number] | "unknown"; + +// "requested" is not part of the raw ACP tool-call status enum. The observer +// assigns it itself, at receipt, before any tool-call status update arrives. +export const PERMISSION_OBSERVER_STAGES = ["requested", "pending", "in_progress", "completed", "failed"] as const; +export type PermissionObserverStage = (typeof PERMISSION_OBSERVER_STAGES)[number] | "unknown"; + +// A settlement outcome is always one of the two terminal tool-call statuses. +export const PERMISSION_OBSERVER_OUTCOMES = ["completed", "failed"] as const; +export type PermissionObserverOutcome = (typeof PERMISSION_OBSERVER_OUTCOMES)[number] | "unknown"; + +export const PERMISSION_OBSERVER_TRANSPORTS = ["local", "ssh", "sandbox"] as const; +export type PermissionObserverTransport = (typeof PERMISSION_OBSERVER_TRANSPORTS)[number] | "unknown"; + +export const PERMISSION_OBSERVER_PERMISSION_MODES = ["approve-all", "approve-reads", "deny-all"] as const; +export type PermissionObserverPermissionMode = (typeof PERMISSION_OBSERVER_PERMISSION_MODES)[number] | "unknown"; + +function mapClosedEnum(allowed: T, value: unknown): T[number] | "unknown" { + if (typeof value === "string" && (allowed as readonly string[]).includes(value)) { + return value as T[number]; + } + return "unknown"; +} + +export function mapPermissionObserverMethod(value: unknown): PermissionObserverMethod { + return mapClosedEnum(PERMISSION_OBSERVER_METHODS, value); +} + +export function mapPermissionObserverToolKind(value: unknown): PermissionObserverToolKind { + return mapClosedEnum(PERMISSION_OBSERVER_TOOL_KINDS, value); +} + +function mapPermissionObserverToolCallStatus(value: unknown): PermissionObserverStage { + return mapClosedEnum(PERMISSION_OBSERVER_STAGES, value); +} + +export function mapPermissionObserverOutcome(value: unknown): PermissionObserverOutcome { + return mapClosedEnum(PERMISSION_OBSERVER_OUTCOMES, value); +} + +export function mapPermissionObserverTransport(value: unknown): PermissionObserverTransport { + return mapClosedEnum(PERMISSION_OBSERVER_TRANSPORTS, value); +} + +export function mapPermissionObserverPermissionMode(value: unknown): PermissionObserverPermissionMode { + return mapClosedEnum(PERMISSION_OBSERVER_PERMISSION_MODES, value); +} + +/** + * Cap and type-guard a correlation identifier before it enters the log + * payload. Paperclip keeps this value in the local run log only; it never + * forwards it to an external sink. + */ +function capIdentifier(value: unknown): string { + if (typeof value !== "string") return "unknown"; + return value.length > MAX_IDENTIFIER_LENGTH ? value.slice(0, MAX_IDENTIFIER_LENGTH) : value; +} + +function boundedAgeMs(ms: number): number { + if (!Number.isFinite(ms) || ms < 0) return 0; + return Math.min(Math.round(ms), MAX_AGE_MS); +} + +interface PermissionLedgerEntry { + sessionId: string; + toolCallId: string; + openedAtMs: number; + toolKind: PermissionObserverToolKind; + lastStage: PermissionObserverStage; +} + +/** A minimal, structural view of the tool-call event the engine already consumes. */ +export interface PermissionObserverToolCallEvent { + toolCallId?: string; + status?: string; +} + +export interface PermissionObserverLogEvent { + type: + | "acpx.permission_observed" + | "acpx.permission_settled" + | "acpx.permission_unsettled" + | "acpx.permission_observer_truncated"; + [key: string]: unknown; +} + +export interface AcpPermissionObserverOptions { + /** + * Starts the durable log write and returns a promise for it. The observer + * never awaits this promise on the permission critical path — call it and + * return, do not `await` it inside `handlePermissionRequest`. The observer + * still tracks the promise so `finalizeRun` can wait for it later. + */ + emitLog: (event: PermissionObserverLogEvent) => Promise | void; + /** The engine's effective permission mode for this run. */ + permissionMode: unknown; + /** The run's execution transport. */ + transport: unknown; + now?: () => number; +} + +export interface AcpPermissionObserver { + /** + * The `onPermissionRequest` hook. Always resolves to `undefined`: it never + * answers, approves, denies, or maps a request to an option. Every branch, + * including an internal error, resolves to `undefined` and never throws. + */ + handlePermissionRequest: ( + request: AcpPermissionRequest, + hookCtx: { signal: AbortSignal }, + ) => Promise; + /** + * Feed a `tool_call` runtime event so the ledger can close a matching open + * entry once its status turns terminal. Call this for every such event on + * the current session; the ledger uses the pair (sessionId, toolCallId) as + * the entry's key. + */ + noteToolCallEvent: (sessionId: string | undefined, event: PermissionObserverToolCallEvent) => void; + /** + * Emit one `acpx.permission_unsettled` event per entry still open, then + * wait for every log write this observer started — including a write + * still in flight from an earlier `handlePermissionRequest` or + * `noteToolCallEvent` call — to finish. Call this once, at run + * finalization. A caller that awaits this method sees every one of the + * observer's log writes land before it returns. + */ + finalizeRun: () => Promise; +} + +export function createAcpPermissionObserver(options: AcpPermissionObserverOptions): AcpPermissionObserver { + const now = options.now ?? Date.now; + const ledger = new Map(); + 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 settledEventCount = 0; + let unsettledEventCount = 0; + let suppressedLedgerEntries = 0; + let suppressedObservedEvents = 0; + let suppressedSettledEvents = 0; + let suppressedUnsettledEvents = 0; + let hasFinalized = false; + + // Every write `emitLog` starts stays in this set until it settles. A write + // started from the permission critical path (`handlePermissionRequest`, + // `noteToolCallEvent`) is never awaited there, so it can still be pending + // when `finalizeRun` runs. `finalizeRun` drains this whole set before it + // returns, so a caller that awaits `finalizeRun` never sees it resolve + // before every queued write, including its own unsettled and truncation + // records, has reached the log sink. + const pendingWrites = new Set>(); + + const queueLog = (event: PermissionObserverLogEvent): void => { + const write = (async () => { + await options.emitLog(event); + })().catch(() => { + // The log sink is diagnostic only; a write failure must never surface + // into the permission critical path or into `finalizeRun`. + }); + pendingWrites.add(write); + void write.finally(() => pendingWrites.delete(write)); + }; + + const ledgerKey = (sessionId: string, toolCallId: string) => sessionId + "\u0000" + toolCallId; + + const handlePermissionRequest: AcpPermissionObserver["handlePermissionRequest"] = async (request) => { + try { + const rawSessionId = request?.sessionId; + const rawToolCallId = request?.raw?.toolCall?.toolCallId; + const sessionId = capIdentifier(rawSessionId); + const toolCallId = capIdentifier(rawToolCallId); + const toolKind = mapPermissionObserverToolKind(request?.inferredKind); + // A ledger entry can only settle later if it opens with a real session + // identifier and a real tool-call identifier. Open it only then, so a + // missing field never produces an entry that can never close. + if (typeof rawSessionId === "string" && typeof rawToolCallId === "string") { + const key = ledgerKey(sessionId, toolCallId); + if (!ledger.has(key)) { + 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; + } + } + } + // 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; + queueLog({ + 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 + // way as a normal observation: `undefined`. + } + return undefined; + }; + + const noteToolCallEvent: AcpPermissionObserver["noteToolCallEvent"] = (sessionIdInput, event) => { + try { + const sessionId = capIdentifier(sessionIdInput); + const toolCallId = capIdentifier(event?.toolCallId); + const key = ledgerKey(sessionId, toolCallId); + const entry = ledger.get(key); + if (!entry) return; + const stage = mapPermissionObserverToolCallStatus(event?.status); + if (stage === "unknown") return; + entry.lastStage = stage; + const outcome = mapPermissionObserverOutcome(event?.status); + if (outcome === "unknown") return; + // Delete the entry (freeing the ledger memory) even when the settled + // budget below is spent: the ledger must not hold a settled entry just + // because the observer could not log its settlement. + ledger.delete(key); + if (settledEventCount < MAX_SETTLED_EVENTS) { + settledEventCount += 1; + queueLog({ + type: "acpx.permission_settled", + sessionId: entry.sessionId, + toolCallId: entry.toolCallId, + outcome, + ageMs: boundedAgeMs(now() - entry.openedAtMs), + }); + } else { + suppressedSettledEvents += 1; + } + } catch { + // Diagnostic only; never let a logging failure surface into the event + // loop that drains the turn's tool-call events. + } + }; + + const finalizeRun: AcpPermissionObserver["finalizeRun"] = async () => { + // The engine's settle step can call finalizeRun more than one time for + // the same run. Only the first call may drain the ledger and emit the + // summary event; every later call is a no-op. + if (hasFinalized) return; + hasFinalized = true; + + const openEntries = [...ledger.values()]; + ledger.clear(); + for (const entry of openEntries) { + if (unsettledEventCount < MAX_UNSETTLED_EVENTS) { + unsettledEventCount += 1; + queueLog({ + type: "acpx.permission_unsettled", + sessionId: entry.sessionId, + toolCallId: entry.toolCallId, + stage: entry.lastStage, + ageMs: boundedAgeMs(now() - entry.openedAtMs), + }); + } else { + suppressedUnsettledEvents += 1; + } + } + // Emit one summary event for the whole run, and only when the observer + // suppressed something. It carries only the four counters and the type + // field: no session identifier, no tool-call identifier, and no other + // agent-controlled value. + if ( + suppressedLedgerEntries > 0 || + suppressedObservedEvents > 0 || + suppressedSettledEvents > 0 || + suppressedUnsettledEvents > 0 + ) { + queueLog({ + type: "acpx.permission_observer_truncated", + suppressedLedgerEntries, + suppressedObservedEvents, + suppressedSettledEvents, + suppressedUnsettledEvents, + }); + } + // Wait for every write this observer started, including a write still in + // flight from an earlier call, so the caller sees every record reach the + // log sink before this method returns. + await Promise.allSettled([...pendingWrites]); + }; + + return { handlePermissionRequest, noteToolCallEvent, finalizeRun }; +}