feat(performance): add bounded monitoring contracts and benchmark harness

package.json

@@ -48,6 +48,7 @@
     "extract:conversations": "tsx --tsconfig tsconfig.node.json scripts/extract-conversation-corpus.mts",
     "check:conversation-fixtures": "tsx --tsconfig tsconfig.node.json scripts/extract-conversation-corpus.mts --verify-checked-in",
     "lifecycle:summarize": "tsx --tsconfig tsconfig.node.json scripts/summarize-lifecycle.mts",
+    "benchmark:monitoring": "tsx scripts/benchmark-monitoring.mts",
     "check:keybindings": "tsx --tsconfig tsconfig.web.json scripts/check-command-keybindings.mts",
     "check": "npm run test:contract && npm run check:worktree-live-fixtures && npm run check:conversation-fixtures && npm run check:keybindings && npm run typecheck && npm run check:live-resume-probe && npm test && npm run test:package",
     "audit:production": "npm audit --omit=dev --audit-level=high",

scripts/benchmark-monitoring.mts

@@ -0,0 +1,40 @@
+// Synthetic, content-free producer benchmark. This measures the code we own;
+// it cannot claim full packaged-app CPU, UI latency, or utility-process RSS.
+// The later Electron replay runner uses the same policy and reports those
+// independently. No credentials, sessions, userData, or diagnostic logs are read.
+import { performance } from 'node:perf_hooks'
+import { cpus, platform, arch } from 'node:os'
+import { BoundedQueue } from '../src/shared/performance/boundedQueue.ts'
+import { LatencyHistogram } from '../src/shared/performance/latencyHistogram.ts'
+import { MONITOR_POLICY } from '../src/shared/performance/monitorPolicy.ts'
+import { MONITOR_RECORD_BYTES, parseMonitorRendererRecord } from '../src/shared/performance/monitorContracts.ts'
+
+const iterations = 100_000
+const samples: number[] = []
+const queue = new BoundedQueue<unknown>(MONITOR_POLICY.sourceQueueRecords, MONITOR_POLICY.sourceQueueBytes)
+const histogram = new LatencyHistogram()
+const record = { kind: 'operation', name: 'transcript.fold', durationMs: 12, outcome: 'success', sessionId: 'synthetic-agent' }
+for (let i = 0; i < 10_000; i++) parseMonitorRendererRecord(record)
+const cpuBefore = process.cpuUsage()
+const started = performance.now()
+for (let i = 0; i < iterations; i++) {
+  const start = performance.now()
+  const admitted = parseMonitorRendererRecord(record)
+  if (!admitted) throw new Error('Synthetic fixture was rejected')
+  queue.push(admitted, MONITOR_RECORD_BYTES)
+  histogram.observe(record.durationMs)
+  if (i % 64 === 0) queue.drain(64, MONITOR_POLICY.batchBytes)
+  samples.push((performance.now() - start) * 1000)
+}
+const elapsedMs = performance.now() - started
+const cpu = process.cpuUsage(cpuBefore)
+samples.sort((a, b) => a - b)
+process.stdout.write(`${JSON.stringify({
+  scope: 'synthetic producer validation + bounded queue + histogram, not whole-app overhead',
+  node: process.version, platform: platform(), arch: arch(), logicalCpus: cpus().length,
+  iterations, elapsedMs, cpuMs: (cpu.user + cpu.system) / 1000,
+  p50Us: samples[Math.floor(samples.length * 0.50)],
+  p95Us: samples[Math.floor(samples.length * 0.95)],
+  p99Us: samples[Math.floor(samples.length * 0.99)],
+  queue: queue.stats,
+}, null, 2)}\n`)

src/shared/performance/boundedQueue.test.ts

@@ -0,0 +1,43 @@
+import { describe, expect, it } from 'vitest'
+import { BoundedQueue } from './boundedQueue'
+
+describe('monitoring backpressure', () => {
+  it('keeps the recent tail within both limits and preserves loss across drains', () => {
+    const queue = new BoundedQueue<string>(3, 10)
+    for (const value of ['a', 'b', 'c', 'd']) queue.push(value, 4)
+    expect(queue.stats).toEqual({ records: 2, bytes: 8, dropped: 2, droppedBytes: 8 })
+    expect(queue.drain(3, 5)).toEqual(['c'])
+    expect(queue.drain()).toEqual(['d'])
+    expect(queue.stats).toEqual({ records: 0, bytes: 0, dropped: 2, droppedBytes: 8 })
+  })
+
+  it('wraps under prolonged overload and releases drained values', () => {
+    const queue = new BoundedQueue<number>(7, 100)
+    for (let i = 0; i < 100_000; i++) queue.push(i, 1)
+    expect(queue.drain()).toEqual([99993, 99994, 99995, 99996, 99997, 99998, 99999])
+    expect(queue.stats.dropped).toBe(99993)
+    queue.push(1, 1)
+    queue.clear()
+    expect(queue.drain()).toEqual([])
+    expect(queue.stats.bytes).toBe(0)
+  })
+
+  it('rejects an oversized item without discarding useful queued evidence', () => {
+    const queue = new BoundedQueue<string>(2, 10)
+    queue.push('keep', 4)
+    expect(queue.push('too large', 11)).toBe(false)
+    expect(queue.push('invalid', NaN)).toBe(false)
+    expect(queue.drain()).toEqual(['keep'])
+    expect(queue.stats.dropped).toBe(2)
+  })
+
+  it('enforces record limits even when byte capacity remains', () => {
+    const queue = new BoundedQueue<number>(2, 100)
+    queue.push(1, 1)
+    queue.push(2, 1)
+    queue.push(3, 1)
+    expect(queue.drain()).toEqual([2, 3])
+    expect(() => new BoundedQueue(0, 1)).toThrow()
+    expect(() => queue.drain(-1)).toThrow()
+  })
+})

src/shared/performance/boundedQueue.ts

@@ -0,0 +1,68 @@
+type Entry<T> = { value: T; bytes: number }
+
+/** A ring, because shift() would make overload handling itself O(queue size). */
+export class BoundedQueue<T> {
+  private readonly entries: Array<Entry<T> | undefined>
+  private head = 0
+  private length = 0
+  private bytes = 0
+  private dropped = 0
+  private droppedBytes = 0
+
+  constructor(private readonly maxRecords: number, private readonly maxBytes: number) {
+    if (!Number.isSafeInteger(maxRecords) || maxRecords < 1
+      || !Number.isSafeInteger(maxBytes) || maxBytes < 1) throw new Error('Invalid queue limits')
+    this.entries = new Array(maxRecords)
+  }
+
+  get stats() {
+    return { records: this.length, bytes: this.bytes, dropped: this.dropped, droppedBytes: this.droppedBytes }
+  }
+
+  push(value: T, bytes: number): boolean {
+    if (!Number.isSafeInteger(bytes) || bytes < 1 || bytes > this.maxBytes) {
+      this.dropped++
+      if (Number.isSafeInteger(bytes) && bytes > 0) this.droppedBytes += bytes
+      return false
+    }
+    // The newest tail explains the current incident. Preserve cumulative loss
+    // separately so clearing/flushing the queue cannot make coverage look whole.
+    while (this.length >= this.maxRecords || this.bytes + bytes > this.maxBytes) {
+      const removed = this.remove()!
+      this.dropped++
+      this.droppedBytes += removed.bytes
+    }
+    this.entries[(this.head + this.length) % this.maxRecords] = { value, bytes }
+    this.length++
+    this.bytes += bytes
+    return true
+  }
+
+  drain(maxRecords = this.maxRecords, maxBytes = this.maxBytes): T[] {
+    if (!Number.isSafeInteger(maxRecords) || maxRecords < 0
+      || !Number.isSafeInteger(maxBytes) || maxBytes < 0) throw new Error('Invalid drain limits')
+    const result: T[] = []
+    let bytes = 0
+    while (this.length && result.length < maxRecords) {
+      const next = this.entries[this.head]!
+      if (bytes + next.bytes > maxBytes) break
+      bytes += next.bytes
+      result.push(this.remove()!.value)
+    }
+    return result
+  }
+
+  clear(): void {
+    while (this.length) this.remove()
+  }
+
+  private remove(): Entry<T> | undefined {
+    if (!this.length) return undefined
+    const entry = this.entries[this.head]!
+    this.entries[this.head] = undefined // Release references on drain, not eventual wraparound.
+    this.head = (this.head + 1) % this.maxRecords
+    this.length--
+    this.bytes -= entry.bytes
+    return entry
+  }
+}

src/shared/performance/latencyHistogram.test.ts

@@ -0,0 +1,55 @@
+import { describe, expect, it } from 'vitest'
+import { LatencyHistogram, latencyQuantile } from './latencyHistogram'
+
+describe('latency distribution', () => {
+  it('weights busy windows by samples when rolling up percentiles', () => {
+    const busy = new LatencyHistogram()
+    for (let i = 0; i < 100; i++) busy.observe(5)
+    const quiet = new LatencyHistogram()
+    quiet.observe(1500)
+    expect(busy.merge(quiet.snapshot())).toBe(true)
+    expect(latencyQuantile(busy.snapshot(), 0.95)).toEqual({ upperBoundMs: 8, overflow: false })
+    expect(busy.snapshot()).toMatchObject({ count: 101, sumMs: 2000, maxMs: 1500 })
+  })
+
+  it('marks unbounded and empty percentiles honestly', () => {
+    const histogram = new LatencyHistogram()
+    expect(latencyQuantile(histogram.snapshot(), 0.99)).toBeNull()
+    histogram.observe(120_000)
+    expect(latencyQuantile(histogram.snapshot(), 0.99)).toEqual({ upperBoundMs: null, overflow: true })
+    histogram.observe(NaN)
+    histogram.observe(-1)
+    histogram.observe(Infinity)
+    expect(histogram.snapshot().count).toBe(1)
+  })
+
+  it('rejects corrupted rollups without changing accumulated evidence', () => {
+    const histogram = new LatencyHistogram()
+    histogram.observe(50)
+    const before = histogram.snapshot()
+    expect(histogram.merge({ ...before, count: 100 })).toBe(false)
+    expect(histogram.merge({ ...before, counts: [1] })).toBe(false)
+    expect(histogram.snapshot()).toEqual(before)
+    before.counts[0] = 400
+    expect(histogram.snapshot().counts[0]).toBe(0)
+    expect(latencyQuantile(histogram.snapshot(), 0)).toBeNull()
+  })
+
+  it('rejects sparse and fabricated empty windows without losing genuine samples', () => {
+    const histogram = new LatencyHistogram()
+    histogram.observe(50)
+    const before = histogram.snapshot()
+    const sparse = new Array(before.counts.length)
+    sparse[0] = 1
+    expect(histogram.merge(structuredClone({ counts: sparse, count: 1, sumMs: 1, maxMs: 1 }))).toBe(false)
+    expect(histogram.merge({ counts: new Array(16), count: 0, sumMs: 0, maxMs: 0 })).toBe(false)
+    expect(histogram.merge({ counts: Array(16).fill(0), count: 0, sumMs: 60000, maxMs: 60000 })).toBe(false)
+    expect(histogram.merge({ ...before, maxMs: 60000 })).toBe(false)
+    const impossible = Array(16).fill(0)
+    impossible[14] = 100
+    expect(histogram.merge({ counts: impossible, count: 100, sumMs: 60000, maxMs: 60000 })).toBe(false)
+    expect(histogram.snapshot()).toEqual(before)
+    histogram.observe(5)
+    expect(latencyQuantile(histogram.snapshot(), 0.95)).toEqual({ upperBoundMs: 50, overflow: false })
+  })
+})

src/shared/performance/latencyHistogram.ts

@@ -0,0 +1,90 @@
+// Fixed, non-cumulative buckets are cheap to merge across workers and time
+// rollups. Averaging p95s would overweight quiet periods and invent a latency
+// distribution that no user experienced. The final bucket is overflow; its
+// percentile is explicitly unbounded, never mislabeled as 60 seconds.
+export const LATENCY_BUCKETS_MS = [1, 4, 8, 16, 32, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000] as const
+export type LatencyHistogramSnapshot = { counts: number[]; count: number; sumMs: number; maxMs: number }
+
+export class LatencyHistogram {
+  private counts: number[] = Array(LATENCY_BUCKETS_MS.length + 1).fill(0)
+  private count = 0
+  private sumMs = 0
+  private maxMs = 0
+
+  observe(ms: number): void {
+    if (!Number.isFinite(ms) || ms < 0 || !Number.isFinite(this.sumMs + ms)
+      || !Number.isSafeInteger(this.count + 1)) return
+    const index = LATENCY_BUCKETS_MS.findIndex(bound => ms <= bound)
+    this.counts[index < 0 ? LATENCY_BUCKETS_MS.length : index]++
+    this.count++
+    this.sumMs += ms
+    this.maxMs = Math.max(this.maxMs, ms)
+  }
+
+  merge(snapshot: LatencyHistogramSnapshot): boolean {
+    if (!isLatencyHistogram(snapshot)) return false
+    const sum = this.sumMs + snapshot.sumMs
+    if (!Number.isFinite(sum) || !Number.isSafeInteger(this.count + snapshot.count)) return false
+    this.counts = this.counts.map((count, index) => count + snapshot.counts[index])
+    this.count += snapshot.count
+    this.sumMs = sum
+    this.maxMs = Math.max(this.maxMs, snapshot.maxMs)
+    return true
+  }
+
+  snapshot(): LatencyHistogramSnapshot {
+    return { counts: [...this.counts], count: this.count, sumMs: this.sumMs, maxMs: this.maxMs }
+  }
+}
+
+export function isLatencyHistogram(input: unknown): input is LatencyHistogramSnapshot {
+  if (!input || typeof input !== 'object') return false
+  const value = input as LatencyHistogramSnapshot
+  if (!Array.isArray(value.counts) || value.counts.length !== LATENCY_BUCKETS_MS.length + 1
+    || !Number.isSafeInteger(value.count) || value.count < 0
+    || !Number.isFinite(value.sumMs) || value.sumMs < 0
+    || !Number.isFinite(value.maxMs) || value.maxMs < 0) return false
+  // Structured clone preserves sparse arrays. Array.every/reduce skip holes,
+  // which would admit missing buckets and permanently poison a merge with NaN.
+  // Validate every index before allowing any mutation of accumulated evidence.
+  let total = 0
+  let lastOccupied = -1
+  let minimumSum = 0
+  let maximumSum = 0
+  for (let i = 0; i < value.counts.length; i++) {
+    const count = value.counts[i]
+    if (!Number.isSafeInteger(count) || count < 0) return false
+    total += count
+    if (!Number.isSafeInteger(total)) return false
+    if (count > 0) lastOccupied = i
+    minimumSum += count * (i === 0 ? 0 : LATENCY_BUCKETS_MS[i - 1])
+    maximumSum += count * Math.min(value.maxMs, LATENCY_BUCKETS_MS[i] ?? value.maxMs)
+  }
+  if (total !== value.count) return false
+  if (total === 0) return value.sumMs === 0 && value.maxMs === 0
+  const maxBucket = LATENCY_BUCKETS_MS.findIndex(bound => value.maxMs <= bound)
+  // An empty window cannot contribute latency, and a nonempty window's maximum
+  // must belong to its last occupied bucket. Allow roundoff in the sum of many
+  // floating-point observations without accepting impossible aggregate totals.
+  const tolerance = Number.EPSILON * Math.max(1, value.sumMs) * value.count
+  // At least one sample actually attained maxMs. Bucket populations also
+  // constrain the sum: 100 samples in (30s, 60s] cannot have a 600ms mean.
+  minimumSum += value.maxMs - (lastOccupied === 0 ? 0 : LATENCY_BUCKETS_MS[lastOccupied - 1])
+  return lastOccupied === (maxBucket < 0 ? LATENCY_BUCKETS_MS.length : maxBucket)
+    && value.sumMs + tolerance >= minimumSum
+    && value.sumMs <= maximumSum + tolerance
+}
+
+export function latencyQuantile(snapshot: LatencyHistogramSnapshot, quantile: number): {
+  upperBoundMs: number | null; overflow: boolean
+} | null {
+  if (!isLatencyHistogram(snapshot) || snapshot.count === 0
+    || !Number.isFinite(quantile) || quantile <= 0 || quantile > 1) return null
+  const rank = Math.ceil(snapshot.count * quantile)
+  let seen = 0
+  for (let index = 0; index < snapshot.counts.length; index++) {
+    seen += snapshot.counts[index]
+    if (seen >= rank) return { upperBoundMs: LATENCY_BUCKETS_MS[index] ?? null, overflow: index === LATENCY_BUCKETS_MS.length }
+  }
+  return null
+}

src/shared/performance/monitorContracts.test.ts

@@ -0,0 +1,55 @@
+import { describe, expect, it } from 'vitest'
+import { MONITOR_POLICY } from './monitorPolicy'
+import { MONITOR_RECORD_BYTES, parseMonitorRendererBatch, parseMonitorRendererRecord } from './monitorContracts'
+import type { MonitorHeartbeat, MonitorOperation } from './monitorContracts'
+
+const operation: MonitorOperation = { kind: 'operation', name: 'transcript.fold', durationMs: 12, outcome: 'success', sessionId: 'session-1' }
+const heartbeat: MonitorHeartbeat = {
+  kind: 'heartbeat', monotonicMs: 1, timeOriginMs: 1000, lagMs: 0, visibility: 'visible',
+  longTaskCount: 0, longTaskTotalMs: 0, longTaskMaxMs: 0, heapUsedBytes: null,
+  heapLimitBytes: null, inputCount: 0, inputMaxMs: 0,
+}
+
+describe('content-minimized renderer ingress', () => {
+  it('admits supported numeric records and copies them at the trust boundary', () => {
+    expect(parseMonitorRendererBatch([operation, heartbeat])).toEqual([operation, heartbeat])
+    expect(parseMonitorRendererRecord(operation)).not.toBe(operation)
+  })
+
+  it.each([
+    { ...operation, data: { nested: { prompt: 'PRIVATE SENTINEL' } } },
+    { ...operation, name: 'PRIVATE SENTINEL' },
+    { ...operation, sessionId: '/Users/private/project' },
+    { ...operation, source: 'another-window' },
+    { ...operation, durationMs: NaN },
+    { ...operation, durationMs: Infinity },
+    { ...operation, outcome: 'PRIVATE SENTINEL' },
+    { ...heartbeat, longTaskCount: -1 },
+    { ...heartbeat, inputCount: 0.5 },
+    { ...heartbeat, heapUsedBytes: 'PRIVATE SENTINEL' },
+  ])('rejects unsupported fields, identities and measurements: %#', value => {
+    expect(parseMonitorRendererRecord(value)).toBeNull()
+  })
+
+  it('rejects oversized batches before processing their records', () => {
+    expect(parseMonitorRendererBatch(Array(MONITOR_POLICY.rendererBatchRecords + 1).fill(operation))).toBeNull()
+    expect(parseMonitorRendererBatch([operation, null])).toBeNull()
+    expect(parseMonitorRendererBatch({})).toBeNull()
+  })
+
+  it('conservatively charges even the largest admitted wire values', () => {
+    const largeHeartbeat = { ...heartbeat }
+    for (const key of Object.keys(largeHeartbeat) as Array<keyof MonitorHeartbeat>) {
+      if (typeof largeHeartbeat[key] === 'number' || largeHeartbeat[key] === null) {
+        Object.assign(largeHeartbeat, { [key]: Number.MAX_SAFE_INTEGER })
+      }
+    }
+    const largestOperation = { ...operation, sessionId: 'a'.repeat(96), durationMs: 86_400_000 }
+    for (const record of [largeHeartbeat, largestOperation]) {
+      expect(parseMonitorRendererRecord(record)).not.toBeNull()
+      expect(Buffer.byteLength(JSON.stringify(record))).toBeLessThanOrEqual(MONITOR_RECORD_BYTES)
+    }
+    expect(MONITOR_POLICY.rendererBatchRecords * (MONITOR_RECORD_BYTES + 1) + 2)
+      .toBeLessThanOrEqual(MONITOR_POLICY.batchBytes)
+  })
+})