feat(harness): add server-authoritative approvals

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<ApprovalResolution>;
+  approve(approvalId: string): ApprovalResolution;
+  deny(approvalId: string): ApprovalResolution;
+  cancelForSession(sessionId: string): number;
+  cancelForRun(runId: string): number;
+}

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<ApprovalResolution>;
+  resolve: (resolution: ApprovalResolution) => void;
+  timer: ReturnType<typeof setTimeout>;
+  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<string, PendingApproval>();
+  readonly #resolved = new Map<string, ApprovalResolution>();
+
+  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<ApprovalResolution> {
+    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<ApprovalResolution>((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`);
+  }
+}

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";

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

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<Resolution>;
+  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<string, unknown>) => 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> = {}): 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" },
+    ]);
+  });
+});

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