diff --git a/packages/harness/src/approvals/approval-broker.ts b/packages/harness/src/approvals/approval-broker.ts new file mode 100644 index 0000000..31f1d52 --- /dev/null +++ b/packages/harness/src/approvals/approval-broker.ts @@ -0,0 +1,24 @@ +export type ApprovalStatus = "approved" | "denied" | "expired" | "cancelled"; + +export type ApprovalRequest = { + approvalId: string; + sessionId: string; + runId: string; + toolCallId: string; + toolName: string; + arguments: unknown; + riskLevel: "low" | "medium" | "high"; +}; + +export type ApprovalResolution = { + approvalId: string; + status: ApprovalStatus; +}; + +export interface ApprovalBroker { + request(input: ApprovalRequest, signal?: AbortSignal): Promise; + approve(approvalId: string): ApprovalResolution; + deny(approvalId: string): ApprovalResolution; + cancelForSession(sessionId: string): number; + cancelForRun(runId: string): number; +} diff --git a/packages/harness/src/approvals/in-memory-approval-broker.ts b/packages/harness/src/approvals/in-memory-approval-broker.ts new file mode 100644 index 0000000..127de97 --- /dev/null +++ b/packages/harness/src/approvals/in-memory-approval-broker.ts @@ -0,0 +1,148 @@ +import type { EventStore } from "../events/event-store.js"; +import { HarnessError } from "../errors/harness-error.js"; +import { redactAndLimit } from "../security/redact.js"; +import type { + ApprovalBroker, + ApprovalRequest, + ApprovalResolution, + ApprovalStatus, +} from "./approval-broker.js"; + +type PendingApproval = { + request: ApprovalRequest; + promise: Promise; + resolve: (resolution: ApprovalResolution) => void; + timer: ReturnType; + signal?: AbortSignal; + abortListener?: () => void; +}; + +export type InMemoryApprovalBrokerOptions = { + eventStore: EventStore; + timeoutMs?: number; + maxArgumentBytes?: number; +}; + +export class InMemoryApprovalBroker implements ApprovalBroker { + readonly #eventStore: EventStore; + readonly #timeoutMs: number; + readonly #maxArgumentBytes: number; + readonly #pending = new Map(); + readonly #resolved = new Map(); + + public constructor(options: InMemoryApprovalBrokerOptions) { + this.#eventStore = options.eventStore; + this.#timeoutMs = options.timeoutMs ?? 30_000; + this.#maxArgumentBytes = options.maxArgumentBytes ?? 16_384; + assertPositiveInteger(this.#timeoutMs, "timeoutMs"); + assertPositiveInteger(this.#maxArgumentBytes, "maxArgumentBytes"); + } + + public request(input: ApprovalRequest, signal?: AbortSignal): Promise { + if (this.#pending.has(input.approvalId) || this.#resolved.has(input.approvalId)) { + throw new HarnessError("APPROVAL_DUPLICATE", `Approval already exists: ${input.approvalId}`); + } + + let resolvePromise!: (resolution: ApprovalResolution) => void; + const promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + const expiresAt = new Date(Date.now() + this.#timeoutMs).toISOString(); + const pending: PendingApproval = { + request: input, + promise, + resolve: resolvePromise, + timer: setTimeout(() => this.#resolve(input.approvalId, "expired"), this.#timeoutMs), + ...(signal === undefined ? {} : { signal }), + }; + this.#pending.set(input.approvalId, pending); + this.#eventStore.append(input.sessionId, { + type: "approval.required", + runId: input.runId, + payload: { + approvalId: input.approvalId, + toolCallId: input.toolCallId, + toolName: input.toolName, + arguments: redactAndLimit(input.arguments, this.#maxArgumentBytes), + riskLevel: input.riskLevel, + expiresAt, + }, + }); + + if (signal !== undefined) { + const abortListener = (): void => { + this.#resolve(input.approvalId, "cancelled"); + }; + pending.abortListener = abortListener; + signal.addEventListener("abort", abortListener, { once: true }); + if (signal.aborted) abortListener(); + } + return promise; + } + + public approve(approvalId: string): ApprovalResolution { + return this.#decide(approvalId, "approved"); + } + + public deny(approvalId: string): ApprovalResolution { + return this.#decide(approvalId, "denied"); + } + + public cancelForSession(sessionId: string): number { + return this.#cancelWhere((request) => request.sessionId === sessionId); + } + + public cancelForRun(runId: string): number { + return this.#cancelWhere((request) => request.runId === runId); + } + + #decide(approvalId: string, status: "approved" | "denied"): ApprovalResolution { + const prior = this.#resolved.get(approvalId); + if (prior !== undefined) { + if (prior.status === status) return prior; + throw new HarnessError( + "APPROVAL_ALREADY_RESOLVED", + `Approval ${approvalId} was already resolved as ${prior.status}`, + { approvalId, status: prior.status }, + ); + } + if (!this.#pending.has(approvalId)) { + throw new HarnessError("APPROVAL_NOT_FOUND", `Unknown approval: ${approvalId}`); + } + return this.#resolve(approvalId, status)!; + } + + #cancelWhere(predicate: (request: ApprovalRequest) => boolean): number { + const approvalIds = [...this.#pending.values()] + .filter((pending) => predicate(pending.request)) + .map((pending) => pending.request.approvalId); + for (const approvalId of approvalIds) this.#resolve(approvalId, "cancelled"); + return approvalIds.length; + } + + #resolve(approvalId: string, status: ApprovalStatus): ApprovalResolution | undefined { + const pending = this.#pending.get(approvalId); + if (pending === undefined) return this.#resolved.get(approvalId); + + this.#pending.delete(approvalId); + clearTimeout(pending.timer); + if (pending.signal !== undefined && pending.abortListener !== undefined) { + pending.signal.removeEventListener("abort", pending.abortListener); + } + const resolution = { approvalId, status } satisfies ApprovalResolution; + this.#resolved.set(approvalId, resolution); + this.#eventStore.append(pending.request.sessionId, { + type: "approval.resolved", + runId: pending.request.runId, + payload: resolution, + }); + pending.resolve(resolution); + return resolution; + } +} + +function assertPositiveInteger(value: number, name: string): void { + if (!Number.isSafeInteger(value) || value < 1) { + throw new RangeError(`${name} must be a positive integer`); + } +} diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index 78a5efc..c3b6b47 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -1,6 +1,9 @@ +export * from "./approvals/approval-broker.js"; +export * from "./approvals/in-memory-approval-broker.js"; export * from "./events/event-store.js"; export * from "./events/in-memory-event-store.js"; export * from "./errors/harness-error.js"; +export * from "./security/redact.js"; export * from "./skills/skill-catalog.js"; export * from "./skills/skill-event-inference.js"; export * from "./tools/mock-ads-metrics.js"; diff --git a/packages/harness/src/security/redact.ts b/packages/harness/src/security/redact.ts new file mode 100644 index 0000000..b94995e --- /dev/null +++ b/packages/harness/src/security/redact.ts @@ -0,0 +1,57 @@ +const REDACTED = "[REDACTED]"; + +export type TruncatedValue = { + type: "truncated"; + byteLength: number; + maxBytes: number; +}; + +export function redactAndLimit(value: unknown, maxBytes: number): unknown { + if (!Number.isSafeInteger(maxBytes) || maxBytes < 1) { + throw new RangeError("maxBytes must be a positive integer"); + } + + const redacted = redact(value, new WeakSet()); + const serialized = JSON.stringify(redacted) ?? "null"; + const byteLength = Buffer.byteLength(serialized, "utf8"); + if (byteLength <= maxBytes) return redacted; + return { type: "truncated", byteLength, maxBytes } satisfies TruncatedValue; +} + +function redact(value: unknown, ancestors: WeakSet): unknown { + if ( + value === null || + typeof value === "string" || + typeof value === "boolean" || + typeof value === "number" + ) { + return value; + } + if (typeof value === "bigint") return value.toString(); + if (typeof value === "undefined") return null; + if (typeof value === "function" || typeof value === "symbol") return "[UNSERIALIZABLE]"; + if (value instanceof Date) return value.toISOString(); + if (ancestors.has(value)) return "[CIRCULAR]"; + + ancestors.add(value); + try { + if (Array.isArray(value)) { + return value.map((entry) => redact(entry, ancestors)); + } + + const entries = Object.entries(value).map(([key, entry]) => [ + key, + isSensitiveKey(key) ? REDACTED : redact(entry, ancestors), + ]); + return Object.fromEntries(entries); + } finally { + ancestors.delete(value); + } +} + +function isSensitiveKey(key: string): boolean { + const normalized = key.toLowerCase().replaceAll(/[^a-z0-9]/g, ""); + return ["apikey", "authorization", "token", "secret", "password", "credential"].some( + (suffix) => normalized === suffix || normalized.endsWith(suffix), + ); +} diff --git a/packages/harness/tests/approval-broker.test.ts b/packages/harness/tests/approval-broker.test.ts new file mode 100644 index 0000000..7002f4e --- /dev/null +++ b/packages/harness/tests/approval-broker.test.ts @@ -0,0 +1,124 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as harness from "../src/index.js"; + +type Resolution = { approvalId: string; status: string }; +type Request = { + approvalId: string; + sessionId: string; + runId: string; + toolCallId: string; + toolName: string; + arguments: unknown; + riskLevel: "low" | "medium" | "high"; +}; +type Broker = { + request(input: Request, signal?: AbortSignal): Promise; + approve(approvalId: string): Resolution; + deny(approvalId: string): Resolution; + cancelForSession(sessionId: string): number; + cancelForRun(runId: string): number; +}; + +function createBroker(timeoutMs = 1_000): { + broker: Broker; + store: { listAfter(sessionId: string): { events: Array<{ type: string; payload: unknown }> } }; +} { + const Store = Reflect.get(harness, "InMemoryEventStore") as new () => { + listAfter(sessionId: string): { events: Array<{ type: string; payload: unknown }> }; + }; + const Broker = Reflect.get(harness, "InMemoryApprovalBroker") as unknown as + | (new (options: Record) => Broker) + | undefined; + expect(Broker, "InMemoryApprovalBroker must be exported").toBeDefined(); + const store = new Store(); + return { broker: new Broker!({ eventStore: store, timeoutMs }), store }; +} + +function request(overrides: Partial = {}): Request { + return { + approvalId: "ap_1", + sessionId: "ses_1", + runId: "run_1", + toolCallId: "call_1", + toolName: "save_report_draft", + arguments: { body: "report", apiKey: "never-log-me" }, + riskLevel: "medium", + ...overrides, + }; +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("InMemoryApprovalBroker", () => { + it("resolves an approved request and treats the same repeated decision as idempotent", async () => { + const { broker, store } = createBroker(); + const waiting = broker.request(request()); + + const first = broker.approve("ap_1"); + const second = broker.approve("ap_1"); + + await expect(waiting).resolves.toEqual({ approvalId: "ap_1", status: "approved" }); + expect(second).toEqual(first); + expect(store.listAfter("ses_1").events.map((event) => event.type)).toEqual([ + "approval.required", + "approval.resolved", + ]); + expect(JSON.stringify(store.listAfter("ses_1"))).not.toContain("never-log-me"); + }); + + it("rejects a conflicting second decision", async () => { + const { broker } = createBroker(); + const waiting = broker.request(request()); + broker.approve("ap_1"); + await waiting; + + expect(() => broker.deny("ap_1")).toThrowError( + expect.objectContaining({ code: "APPROVAL_ALREADY_RESOLVED" }), + ); + }); + + it("expires pending approval once when its deadline passes", async () => { + vi.useFakeTimers(); + const { broker, store } = createBroker(50); + const waiting = broker.request(request()); + + await vi.advanceTimersByTimeAsync(50); + + await expect(waiting).resolves.toEqual({ approvalId: "ap_1", status: "expired" }); + expect(store.listAfter("ses_1").events.map((event) => event.type)).toEqual([ + "approval.required", + "approval.resolved", + ]); + }); + + it("cancels on abort and removes the abort listener after resolution", async () => { + const { broker, store } = createBroker(); + const controller = new AbortController(); + const waiting = broker.request(request(), controller.signal); + + controller.abort(); + + await expect(waiting).resolves.toEqual({ approvalId: "ap_1", status: "cancelled" }); + expect(store.listAfter("ses_1").events).toHaveLength(2); + controller.abort(); + expect(store.listAfter("ses_1").events).toHaveLength(2); + }); + + it("cancels all pending approvals for a session or run", async () => { + const { broker } = createBroker(); + const first = broker.request(request()); + const second = broker.request(request({ approvalId: "ap_2", runId: "run_2", toolCallId: "call_2" })); + const third = broker.request(request({ approvalId: "ap_3", sessionId: "ses_2", toolCallId: "call_3" })); + + expect(broker.cancelForRun("run_2")).toBe(1); + expect(broker.cancelForSession("ses_1")).toBe(1); + expect(broker.cancelForSession("ses_2")).toBe(1); + await expect(Promise.all([first, second, third])).resolves.toEqual([ + { approvalId: "ap_1", status: "cancelled" }, + { approvalId: "ap_2", status: "cancelled" }, + { approvalId: "ap_3", status: "cancelled" }, + ]); + }); +}); diff --git a/packages/harness/tests/redact.test.ts b/packages/harness/tests/redact.test.ts new file mode 100644 index 0000000..dd6bb07 --- /dev/null +++ b/packages/harness/tests/redact.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import * as harness from "../src/index.js"; + +describe("redactAndLimit", () => { + it("recursively redacts sensitive keys without mutating the source", () => { + const redactAndLimit = Reflect.get(harness, "redactAndLimit") as + | ((value: unknown, maxBytes: number) => unknown) + | undefined; + expect(redactAndLimit, "redactAndLimit must be exported").toBeDefined(); + const source = { + apiKey: "sk-secret", + nested: [{ Authorization: "Bearer secret" }, { refresh_token: "refresh" }], + safe: "visible", + }; + + expect(redactAndLimit!(source, 10_000)).toEqual({ + apiKey: "[REDACTED]", + nested: [{ Authorization: "[REDACTED]" }, { refresh_token: "[REDACTED]" }], + safe: "visible", + }); + expect(source.apiKey).toBe("sk-secret"); + }); + + it("replaces oversized serialized values with a typed byte summary", () => { + const redactAndLimit = Reflect.get(harness, "redactAndLimit") as + | ((value: unknown, maxBytes: number) => unknown) + | undefined; + const result = redactAndLimit?.({ text: "中".repeat(100) }, 32); + + expect(result).toMatchObject({ + type: "truncated", + maxBytes: 32, + }); + expect(Reflect.get(result as object, "byteLength")).toBeGreaterThan(32); + expect(JSON.stringify(result)).not.toContain("中"); + }); +});