diff --git a/loopx/capabilities/reliability_diagnostics/README.md b/loopx/capabilities/reliability_diagnostics/README.md index d675798ccb..d3b8560af4 100644 --- a/loopx/capabilities/reliability_diagnostics/README.md +++ b/loopx/capabilities/reliability_diagnostics/README.md @@ -209,6 +209,33 @@ covers legacy readback and an explicit offline upgrade. Legacy files remain readable, but block new appends until reconciled; mixed ownership must hold deletion. Upgrade CLI and plugin together with writers frozen. +## Measure Observer Overhead + +`packages/dsh-loopx-plugin/bench/observer-overhead.ts` measures the observer +component cost on a deterministic synthetic DSH event stream: observer off/on +arms, CPU (user+system), RSS, ledger append through the shipped writer, ledger +bytes per envelope and stats bytes per flush, per-trial event loss and buffer +peaks, and the clock-source mix with full denominators. It also checks that +every planned record is observed with the planned envelope-kind mix. A trial +that drops or rejects events, fails an append, persists fewer envelopes than it +accepted, or misses that record mix is excluded from the timing summary; the +report is marked not usable and lists warnings, so a lossy or incompletely +consumed run cannot read as a clean result. It writes a public-safe +JSON + Markdown report and states its limits. It does not measure a live DSH +host, model, network, or Cordis dispatch, and it does not qualify C1. See the +[bench README](../../../packages/dsh-loopx-plugin/bench/README.md). + +Run it from `packages/dsh-loopx-plugin`: + +```bash +node --experimental-strip-types bench/observer-overhead.ts --events 10000 +``` + +`--events` is normalized down to a whole six-event turn cycle (5,000 → 4,998; +10,000 → 9,996); the report states both counts. The bench uses a 4,096 buffer +bound for the synthetic burst; pass `--buffer-bound 256` to mirror the observer's +product default. + ## Validation ```bash @@ -231,5 +258,6 @@ No dashboard surface, no L2 recommendations, no automatic recovery, no writeback into goals, todos, gates, or session runtime, and no change to the `loopx status` first screen. The adapter requires an externally pinned goal/session/run identity; it does not discover bindings through the Driver or -LoopX CLI. This prototype also does not provide matched native/L1 execution, -observer CPU/I/O/latency/storage measurement, or an eligible C1 run. +LoopX CLI. This prototype also does not provide matched native/L1 execution, a +live DSH host measurement, or an eligible C1 run. The bench tool measures +observer component cost on a synthetic stream; it is not a live qualification. diff --git a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md index 972ee51098..8dc69a8221 100644 --- a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md +++ b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md @@ -177,6 +177,28 @@ live 部署验收。操作 regular file 前会验证 provider 目录与 CLI 的 旧文件仍可只读回查,但必须离线核对后才能恢复追加;方案说明了升级和回退步骤, 归属不明确时须暂停删除。冻结 writer 后一起升级 CLI 与插件。 +## 测量 observer 开销 + +`packages/dsh-loopx-plugin/bench/observer-overhead.ts` 在确定性的 synthetic DSH 事件流上 +测量 observer 组件开销:observer 关闭/开启两臂、CPU(user+system)、RSS、经 shipped writer +的 ledger append 延迟、每条 envelope 的 ledger 字节与每次 flush 的 stats 字节、逐试验的 +event loss 与 buffer 峰值,以及 clock source 构成和完整分母;并校验计划记录与 envelope kind +构成是否被完整消费。丢弃或拒绝事件、append 失败、persisted 少于 accepted、或未覆盖计划记录构成 +的试验会从计时汇总中排除;报告标记为不可用并列出警告,因此有丢失或未完整消费的运行不会被读成 +干净结果。输出 public-safe 的 JSON + Markdown 报告,并声明限制;不测量真实 +DSH host、模型、网络或 Cordis dispatch,也不构成 C1 资格。见 +[bench README](../../../packages/dsh-loopx-plugin/bench/README.md)。 + +在 `packages/dsh-loopx-plugin` 下运行: + +```bash +node --experimental-strip-types bench/observer-overhead.ts --events 10000 +``` + +`--events` 会向下归一化为 6 事件 turn cycle 的整数倍(5,000 → 4,998;10,000 → 9,996), +报告同时给出请求值与归一化值。bench 对 synthetic 突发使用 4,096 buffer bound;传 +`--buffer-bound 256` 可对齐 observer 的产品默认值。 + ## 验证 ```bash @@ -195,5 +217,5 @@ projection 报告 `read` 上的重复、一次已恢复的错误、无 stall。 不做 dashboard、不做 L2 建议、不做自动恢复、不回写 goal / todo / gate / session runtime、 不改 `loopx status` 首屏。适配器要求外部固定 goal/session/run identity;不会通过 Driver -或 LoopX CLI 发现绑定。本原型也不提供 matched native/L1 执行、observer CPU/I/O/延迟/ -存储开销测量,或一次合格 C1 实跑。 +或 LoopX CLI 发现绑定。本原型也不提供 matched native/L1 执行、真实 DSH host 测量,或一次合格 C1 实跑。 +bench 工具只在 synthetic 事件流上测量 observer 组件开销,不构成 live 资格。 diff --git a/packages/dsh-loopx-plugin/bench/README.md b/packages/dsh-loopx-plugin/bench/README.md new file mode 100644 index 0000000000..1d8076d5bf --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/README.md @@ -0,0 +1,120 @@ +# Observer overhead bench + +Measure the DSH shadow observer's component cost on a deterministic synthetic +event stream. This is a development tool for the P0 "measured overhead" gap in +[#5211](https://github.com/loopx-project/loopx/issues/5211); it is not a live +DSH host, model, or C1 qualification. + +## What it measures + +- **Two arms**: a baseline driver that walks the same planned stream without an + observer (no serialization or appends), and the real `ShadowObserver`. +- **CPU**: `process.cpuUsage()` user + system delta. +- **RSS**: process RSS before/after each trial. +- **Ledger append**: p50/p95 latency through the shipped `appendLedgerLines` + writer, envelope bytes per persisted envelope, and stats bytes per flush. +- **Event accounting**: observed, accepted, persisted, dropped (total and max), + rejected, peak buffer, and flush attempts. Counts must satisfy + `observed = accepted + rejected + dropped`. +- **Clock**: the harness-time vs wall-clock envelope mix and the maximum + declared uncertainty. +- **Integrity**: no event loss, persisted envelopes equal accepted envelopes, + every planned record observed with the planned envelope-kind mix, no outbound + endpoints, no worker-context influence, and no scheduler-input influence. + +A trial that drops or rejects events, fails an append, persists fewer envelopes +than it accepted, or does not consume the planned record and envelope-kind mix +is excluded from the arm summaries. The report is then marked `NOT USABLE`, +prints the per-trial loss and peaks, and lists warnings. Do not use such a run +as an overhead sample: rerun with a larger `--buffer-bound` or fewer `--events` +(or fix the bench mix or observer when planned kinds stop being consumed). + +The report is public-safe: it contains aggregate numbers, the Node/platform +version, and limits. It does not contain goal ids, session ids, ledger paths, or +raw event content. + +## Run + +From `packages/dsh-loopx-plugin`: + +```bash +node --experimental-strip-types bench/observer-overhead.ts \ + --events 10000 \ + --json ../../.local/observer-overhead.json \ + --markdown ../../.local/observer-overhead.md +``` + +Defaults are 5,000 events, 3 repeats, and a buffer bound of 4,096. The bench +uses a larger bound than the observer's product default (256) because the +synthetic producer emits events far faster than a real session; pass +`--buffer-bound 256` to mirror the product default and exercise backpressure +drops. `--events` is normalized down to a whole six-event turn cycle: 5,000 +becomes 4,998 and 10,000 becomes 9,996, and the report prints both counts. The +tool runs a short warmup, then alternates baseline and observer trials. Use +`--json -` or `--markdown -` to print a report to stdout. + +The observer requests a flush at every turn end, but an append still in flight +coalesces later requests, so the number of append batches (and the peak buffer) +depends on I/O latency. The report prints the actual flush attempts and peak +buffer; treat the per-event numbers as a signal, not a workload forecast. + +Example output from one Windows/Node 26 run of `--events 10000` (numbers are +machine-specific and are not a qualification): + +```text +- Status: **usable** +- Events per trial: 9,996 (requested 10,000; normalized to whole 6-event turn cycles) +- Buffer bound: 4,096 + +| Metric | Baseline | Observer | Delta | +| --- | ---: | ---: | ---: | +| CPU total, median (ms) | 0.00 | 906.00 | 906.00 | +| CPU, median (ms / 1k events) | 0.00 | 90.64 | 90.64 | +| RSS delta, median | 448.0 KiB | 13.1 MiB | 12.7 MiB | +| Ledger append p50 (ms) | n/a | 1.49 | n/a | +| Ledger append p95 (ms) | n/a | 2.09 | n/a | +| Ledger bytes / envelope | n/a | 490.38 | n/a | +| Ledger stats bytes / flush | n/a | 1326.25 | n/a | +| Persisted envelopes, median | n/a | 9,998 | n/a | +| Clock source, median (harness / wall-clock) | n/a | 9,996 / 2 | n/a | +| Clock uncertainty max, median (ms) | n/a | 50.00 | n/a | + +| Observed | Accepted | Persisted | Dropped (total) | Dropped (max) | Rejected | Flush attempts | Peak buffer (max) | +| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| 9,998 | 9,998 | 9,998 | 0 | 0 | 0 | 218 | 84 | + +- Planned records not observed: 0 +- Unsupported envelopes: 0 +- No event loss: passed +- Persisted equals accepted: passed +- All trials usable: passed +- Observed matches plan: passed +- Planned envelope mix complete: passed +``` + +The clock row shows the split: workload events carry a harness timestamp, and +the session-created/disposed records use the observer wall clock with the +declared 50 ms uncertainty. That mix is the measured timing uncertainty; the +maximum is the observer's declared bound for those two lifecycle records and does +not vary with the stream. + +## What it deliberately does not measure + +- Cordis hook dispatch or DSH scheduling; +- model, network, or tool latency; +- a real session's worker outcome; +- C1 eligibility or P0 exit. + +Those require a separately authorized live run and maintainer acceptance. See +the [capability README](../../../loopx/capabilities/reliability_diagnostics/README.md) +and the [RFC](../../../docs/architecture/rfcs/long-running-agent-reliability-diagnostics-governed-delivery-v0.md). + +## Test + +```bash +pnpm test -- observer-overhead +``` + +The test covers the event-plan shape, percentile boundaries, the shipped-writer +append path and byte split, loss, append-failure, and planned-record-mix +marking, CLI parsing, and public-safe report rendering. diff --git a/packages/dsh-loopx-plugin/bench/observer-overhead.ts b/packages/dsh-loopx-plugin/bench/observer-overhead.ts new file mode 100644 index 0000000000..cf2ed1a112 --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/observer-overhead.ts @@ -0,0 +1,162 @@ +import { mkdir, writeFile } from 'node:fs/promises' +import { cpus, totalmem } from 'node:os' +import { dirname } from 'node:path' +import { pathToFileURL } from 'node:url' +import { runBaselineTrial, runObserverTrial, normalizeEventCount } from './overhead-runner.ts' +import type { TrialMeasurement } from './overhead-runner.ts' +import { buildOverheadReport, renderMarkdown } from './overhead-report.ts' + +export interface CliOptions { + readonly events: number + readonly repeat: number + readonly bufferBound: number + readonly json: string | undefined + readonly markdown: string | undefined + readonly help: boolean +} + +function positiveInteger(value: string | undefined, flag: string): number { + const parsed = Number.parseInt(value ?? '', 10) + if (!Number.isInteger(parsed) || parsed <= 0) { + throw new Error(`${flag} requires a positive integer`) + } + return parsed +} + +export function parseArgs(args: readonly string[]): CliOptions { + let events = 5_000 + let repeat = 3 + let bufferBound = 4_096 + let json: string | undefined + let markdown: string | undefined + let help = false + for (let index = 0; index < args.length; index += 1) { + const flag = args[index] + switch (flag) { + case '--help': + case '-h': + help = true + break + case '--events': + index += 1 + events = positiveInteger(args[index], flag) + break + case '--repeat': + index += 1 + repeat = positiveInteger(args[index], flag) + break + case '--buffer-bound': + index += 1 + bufferBound = positiveInteger(args[index], flag) + break + case '--json': + index += 1 + json = args[index] + if (json === undefined) throw new Error('--json requires a path or "-"') + break + case '--markdown': + index += 1 + markdown = args[index] + if (markdown === undefined) throw new Error('--markdown requires a path or "-"') + break + default: + throw new Error(`unknown argument: ${flag ?? ''}`) + } + } + return { events, repeat, bufferBound, json, markdown, help } +} + +function printUsage(): void { + process.stdout.write(`Usage: node --experimental-strip-types bench/observer-overhead.ts [options] + +Options: + --events synthetic session events per trial, normalized down to a + whole 6-event turn cycle (default: 5000) + --repeat measured trials per arm (default: 3) + --buffer-bound observer buffer bound (default: 4096; pass 256 to mirror + the observer's product default and exercise drops) + --json write the JSON report to path; "-" writes stdout + --markdown write the Markdown report to path; "-" writes stdout + --help show this help +`) +} + +async function main(): Promise { + const options = parseArgs(process.argv.slice(2)) + if (options.help) { + printUsage() + return + } + const events = normalizeEventCount(options.events) + if (events !== options.events) { + process.stderr.write( + `note: normalized --events ${options.events} to ${events} (whole 6-event turn cycles)\n`, + ) + } + const warmupEvents = Math.min(600, events) + runBaselineTrial({ inputEvents: warmupEvents, bufferBound: options.bufferBound }) + await runObserverTrial({ inputEvents: warmupEvents, bufferBound: options.bufferBound }) + + const baseline: TrialMeasurement[] = [] + const observer: TrialMeasurement[] = [] + for (let index = 0; index < options.repeat; index += 1) { + baseline.push(runBaselineTrial({ inputEvents: events, bufferBound: options.bufferBound })) + observer.push(await runObserverTrial({ inputEvents: events, bufferBound: options.bufferBound })) + } + + const report = buildOverheadReport({ + config: { + events, + requested_events: options.events, + repeat: options.repeat, + buffer_bound: options.bufferBound, + event_mix: 'turn-cycle-v1', + }, + baseline, + observer, + environment: { + node: process.version, + platform: process.platform, + arch: process.arch, + cpus: cpus().length, + total_memory_gib: Number((totalmem() / 1_024 ** 3).toFixed(1)), + }, + }) + const markdown = renderMarkdown(report) + const json = `${JSON.stringify(report, null, 2)}\n` + + if (!report.usable) { + process.stderr.write('warning: this report is not a usable overhead sample\n') + for (const warning of report.warnings) process.stderr.write(`- ${warning}\n`) + } + + let printed = false + if (options.json === '-') { + process.stdout.write(json) + printed = true + } else if (options.json !== undefined) { + await mkdir(dirname(options.json), { recursive: true }) + await writeFile(options.json, json, 'utf8') + process.stderr.write(`wrote ${options.json}\n`) + } + if (options.markdown === '-') { + process.stdout.write(markdown) + printed = true + } else if (options.markdown !== undefined) { + await mkdir(dirname(options.markdown), { recursive: true }) + await writeFile(options.markdown, markdown, 'utf8') + process.stderr.write(`wrote ${options.markdown}\n`) + } + if (!printed && options.markdown === undefined) { + process.stdout.write(markdown) + } +} + +if (process.argv[1] !== undefined && import.meta.url === pathToFileURL(process.argv[1]).href) { + try { + await main() + } catch (error: unknown) { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`) + process.exitCode = 1 + } +} diff --git a/packages/dsh-loopx-plugin/bench/overhead-report.ts b/packages/dsh-loopx-plugin/bench/overhead-report.ts new file mode 100644 index 0000000000..ad0509c4f0 --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/overhead-report.ts @@ -0,0 +1,408 @@ +import type { TrialMeasurement } from './overhead-runner.ts' + +export interface ArmSummary { + readonly trials: number + readonly usable_trials: number + readonly excluded_trials: number + readonly cpu_total_ms_median: number + readonly cpu_total_ms_p95: number + readonly cpu_ms_per_1k_events_median: number + readonly rss_delta_bytes_median: number + readonly ledger_bytes_per_event: number + readonly stats_bytes_per_flush: number + readonly append_p50_ms: number + readonly append_p95_ms: number + readonly persisted_events_median: number +} + +export interface OverheadReportEnvironment { + readonly node: string + readonly platform: string + readonly arch: string + readonly cpus: number + readonly total_memory_gib: number +} + +export interface OverheadReportConfig { + readonly events: number + readonly requested_events?: number + readonly repeat: number + readonly buffer_bound: number + readonly event_mix: string +} + +export interface OverheadIntegrity { + readonly count_conservation: boolean + readonly no_event_loss: boolean + readonly persisted_equals_accepted: boolean + readonly all_trials_usable: boolean + readonly observed_matches_plan: boolean + readonly envelope_kind_mix_complete: boolean + readonly outbound_endpoints_empty: boolean + readonly worker_context_influence_absent: boolean + readonly scheduler_input_influence_absent: boolean + readonly observer_failures: number + readonly persisted_envelopes: number +} + +export interface OverheadEventAccounting { + readonly observed_events_median: number + readonly accepted_events_median: number + readonly persisted_events_median: number + readonly dropped_events_median: number + readonly dropped_events_total: number + readonly dropped_events_max: number + readonly rejected_events_median: number + readonly flush_attempts_median: number + readonly peak_buffered_events_max: number + readonly clock_uncertainty_max_ms_median: number + readonly clock_harness_time_events_median: number + readonly clock_wall_clock_events_median: number + readonly trials_with_loss: number + readonly usable_trials: number + readonly excluded_trials: number + readonly unconsumed_events_total: number + readonly unsupported_envelopes_total: number +} + +export interface OverheadDelta { + readonly cpu_ms_per_1k_events: number + readonly rss_delta_bytes: number +} + +export interface OverheadReport { + readonly schema: 'reliability_observer_overhead_report_v0' + readonly generated_at: string + readonly usable: boolean + readonly warnings: readonly string[] + readonly environment: OverheadReportEnvironment + readonly config: OverheadReportConfig + readonly arms: { + readonly baseline: { readonly summary: ArmSummary; readonly trials: readonly TrialMeasurement[] } + readonly observer: { readonly summary: ArmSummary; readonly trials: readonly TrialMeasurement[] } + } + readonly delta: OverheadDelta + readonly integrity: OverheadIntegrity + readonly event_accounting: OverheadEventAccounting + readonly limits: readonly string[] +} + +export interface OverheadReportInput { + readonly config: OverheadReportConfig + readonly baseline: readonly TrialMeasurement[] + readonly observer: readonly TrialMeasurement[] + readonly environment: OverheadReportEnvironment + readonly generatedAt?: string +} + +const REPORT_LIMITS = [ + 'Synthetic event stream only; this is not a live DSH host, model, or network measurement.', + 'Measures observer component cost (event compaction, envelope construction, JSON serialization, and ledger append) through the shipped ledger writer; excludes Cordis hook dispatch, DSH scheduling, model, and network.', + 'Verifies that every planned record was observed and that the persisted envelope-kind mix matches the plan, so a retired or renamed planned event kind cannot read as a clean sample.', + 'The synthetic stream requests a flush at every turn end, but appends coalesce while one is in flight. The append-batch count, peak buffer, and per-event cost therefore depend on I/O latency; the report prints the actual counts and is not a workload forecast.', + 'Baseline is an empty synthetic driver; the delta is the observer component cost, not a claim about a real worker throughput.', + 'RSS is process-level and may move with GC; use repeats and medians, not a single sample.', + 'A trial that drops or rejects events, fails appends, loses persisted envelopes, or misses the planned record mix is excluded from the arm summaries; the report is marked not usable and prints warnings.', + 'A C1 eligibility or P0-exit claim requires a separately authorized live run and maintainer acceptance.', +] as const + +export function percentile(values: readonly number[], fraction: number): number { + if (values.length === 0) return 0 + const sorted = [...values].sort((left, right) => left - right) + const bounded = Math.min(1, Math.max(0, fraction)) + const index = Math.min(sorted.length - 1, Math.max(0, Math.ceil(bounded * sorted.length) - 1)) + return sorted[index] ?? 0 +} + +export function median(values: readonly number[]): number { + return percentile(values, 0.5) +} + +function maximum(values: readonly number[]): number { + return values.length === 0 ? 0 : Math.max(...values) +} + +interface EnvelopeKindDiff { + readonly missing: readonly string[] + readonly unexpected: readonly string[] +} + +/** Kinds whose persisted count is below the plan, and kinds the plan lacks. */ +export function envelopeKindDiff(trial: TrialMeasurement): EnvelopeKindDiff { + const missing: string[] = [] + const unexpected: string[] = [] + for (const [kind, expected] of Object.entries(trial.expectedEnvelopeKinds)) { + if ((trial.envelopeKinds[kind] ?? 0) < expected) missing.push(kind) + } + for (const [kind, actual] of Object.entries(trial.envelopeKinds)) { + if (actual > (trial.expectedEnvelopeKinds[kind] ?? 0)) unexpected.push(kind) + } + return { missing, unexpected } +} + +export function envelopeKindMixComplete(trial: TrialMeasurement): boolean { + const diff = envelopeKindDiff(trial) + return diff.missing.length === 0 && diff.unexpected.length === 0 +} + +/** + * A trial whose timings can be compared with another trial: it lost no events, + * rejected none, failed no append, persisted every accepted envelope, and + * consumed the planned record mix. A trial that silently skips a planned event + * kind (or turns it into an unknown kind) is not comparable either; the + * baseline arm plans no envelopes, so those checks hold trivially for it. + */ +export function isUsableTrial(trial: TrialMeasurement): boolean { + return trial.droppedEvents === 0 + && trial.rejectedEvents === 0 + && trial.observerFailures === 0 + && trial.persistedEvents === trial.acceptedEvents + && trial.observedEvents === trial.expectedObservedEvents + && envelopeKindMixComplete(trial) +} + +export function summarizeTrials(trials: readonly TrialMeasurement[]): ArmSummary { + const usable = trials.filter(isUsableTrial) + const cpuPer1k = usable.map(trial => trial.workloadEvents === 0 + ? 0 + : (trial.cpuTotalMs / trial.workloadEvents) * 1_000) + const appendLatencies = usable.flatMap(trial => trial.appendLatenciesMs) + const totalEnvelopeBytes = usable.reduce((sum, trial) => sum + trial.envelopeBytes, 0) + const totalStatsBytes = usable.reduce((sum, trial) => sum + trial.statsBytes, 0) + const totalPersisted = usable.reduce((sum, trial) => sum + trial.persistedEvents, 0) + const totalFlushes = usable.reduce((sum, trial) => sum + trial.flushAttempts, 0) + return { + trials: trials.length, + usable_trials: usable.length, + excluded_trials: trials.length - usable.length, + cpu_total_ms_median: median(usable.map(trial => trial.cpuTotalMs)), + cpu_total_ms_p95: percentile(usable.map(trial => trial.cpuTotalMs), 0.95), + cpu_ms_per_1k_events_median: median(cpuPer1k), + rss_delta_bytes_median: median(usable.map(trial => trial.rssDeltaBytes)), + ledger_bytes_per_event: totalPersisted === 0 ? 0 : totalEnvelopeBytes / totalPersisted, + stats_bytes_per_flush: totalFlushes === 0 ? 0 : totalStatsBytes / totalFlushes, + append_p50_ms: percentile(appendLatencies, 0.5), + append_p95_ms: percentile(appendLatencies, 0.95), + persisted_events_median: median(usable.map(trial => trial.persistedEvents)), + } +} + +function buildWarnings(observer: readonly TrialMeasurement[]): readonly string[] { + if (observer.length === 0) return ['No observer trial was measured.'] + const warnings: string[] = [] + const usableTrials = observer.filter(isUsableTrial).length + if (usableTrials !== observer.length) { + warnings.push(`Not a usable overhead sample: ${usableTrials} of ${observer.length} observer trials are usable.`) + } + const lossy = observer.filter(trial => trial.droppedEvents > 0) + if (lossy.length > 0) { + const dropped = lossy.map(trial => trial.droppedEvents) + warnings.push( + `${lossy.length} of ${observer.length} observer trials dropped events ` + + `(max ${maximum(dropped)}, total ${dropped.reduce((sum, value) => sum + value, 0)}); ` + + 'lossy trials are excluded from the arm summaries. Rerun with a larger --buffer-bound ' + + 'or fewer --events before treating this as an overhead sample.', + ) + } + const rejected = observer.reduce((sum, trial) => sum + trial.rejectedEvents, 0) + if (rejected > 0) { + warnings.push(`${rejected} observer event(s) were rejected; affected trials are excluded from the arm summaries.`) + } + const failures = observer.reduce((sum, trial) => sum + trial.observerFailures, 0) + if (failures > 0) { + warnings.push(`${failures} ledger append failure(s); affected trials are excluded from the arm summaries.`) + } + const mismatched = observer.filter(trial => trial.persistedEvents !== trial.acceptedEvents).length + if (mismatched > 0) { + warnings.push( + `Persisted envelopes do not match accepted envelopes in ${mismatched} trial(s); ` + + 'a write may have been lost.', + ) + } + const unconsumed = observer.reduce( + (sum, trial) => sum + Math.max(0, trial.expectedObservedEvents - trial.observedEvents), 0, + ) + if (unconsumed > 0) { + const affected = observer.filter(trial => trial.observedEvents !== trial.expectedObservedEvents).length + warnings.push( + `${affected} of ${observer.length} observer trial(s) did not observe every planned record ` + + `(${unconsumed} record(s) missing); a planned event kind may no longer be consumed. ` + + 'Update the bench event mix or the observer before treating this as an overhead sample.', + ) + } + const mixMismatches = observer.filter(trial => !envelopeKindMixComplete(trial)) + const firstMismatch = mixMismatches[0] + if (firstMismatch !== undefined) { + const diff = envelopeKindDiff(firstMismatch) + warnings.push( + `The persisted envelope kind mix does not match the plan in ${mixMismatches.length} of ` + + `${observer.length} observer trial(s); first mismatch is missing: ` + + `${diff.missing.length > 0 ? diff.missing.join(', ') : 'none'}, unexpected: ` + + `${diff.unexpected.length > 0 ? diff.unexpected.join(', ') : 'none'}. ` + + 'Such trials are excluded from the arm summaries.', + ) + } + return warnings +} + +export function buildOverheadReport(input: OverheadReportInput): OverheadReport { + const baselineSummary = summarizeTrials(input.baseline) + const observerSummary = summarizeTrials(input.observer) + const usable = input.observer.length > 0 && input.observer.every(isUsableTrial) + const dropped = input.observer.map(trial => trial.droppedEvents) + return { + schema: 'reliability_observer_overhead_report_v0', + generated_at: input.generatedAt ?? new Date().toISOString(), + usable, + warnings: buildWarnings(input.observer), + environment: input.environment, + config: input.config, + arms: { + baseline: { summary: baselineSummary, trials: input.baseline }, + observer: { summary: observerSummary, trials: input.observer }, + }, + delta: { + cpu_ms_per_1k_events: observerSummary.cpu_ms_per_1k_events_median + - baselineSummary.cpu_ms_per_1k_events_median, + rss_delta_bytes: observerSummary.rss_delta_bytes_median + - baselineSummary.rss_delta_bytes_median, + }, + integrity: { + count_conservation: input.observer.every(trial => trial.observedEvents + === trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents), + no_event_loss: input.observer.every(trial => trial.droppedEvents === 0), + persisted_equals_accepted: input.observer.every(trial => trial.persistedEvents === trial.acceptedEvents), + all_trials_usable: usable, + observed_matches_plan: input.observer.every(trial => trial.observedEvents === trial.expectedObservedEvents), + envelope_kind_mix_complete: input.observer.every(envelopeKindMixComplete), + outbound_endpoints_empty: input.observer.every(trial => trial.outboundEndpoints === 0), + worker_context_influence_absent: input.observer.every(trial => !trial.workerContextInfluence), + scheduler_input_influence_absent: input.observer.every(trial => !trial.schedulerInputInfluence), + observer_failures: input.observer.reduce((sum, trial) => sum + trial.observerFailures, 0), + persisted_envelopes: input.observer.reduce((sum, trial) => sum + trial.persistedEvents, 0), + }, + event_accounting: { + observed_events_median: median(input.observer.map(trial => trial.observedEvents)), + accepted_events_median: median(input.observer.map(trial => trial.acceptedEvents)), + persisted_events_median: median(input.observer.map(trial => trial.persistedEvents)), + dropped_events_median: median(dropped), + dropped_events_total: dropped.reduce((sum, value) => sum + value, 0), + dropped_events_max: maximum(dropped), + rejected_events_median: median(input.observer.map(trial => trial.rejectedEvents)), + flush_attempts_median: median(input.observer.map(trial => trial.flushAttempts)), + peak_buffered_events_max: maximum(input.observer.map(trial => trial.peakBufferedEvents)), + clock_uncertainty_max_ms_median: median(input.observer.map(trial => trial.clockUncertaintyMaxMs)), + clock_harness_time_events_median: median(input.observer.map(trial => trial.harnessTimeEvents)), + clock_wall_clock_events_median: median(input.observer.map(trial => trial.wallClockEvents)), + trials_with_loss: input.observer.filter(trial => trial.droppedEvents > 0).length, + usable_trials: observerSummary.usable_trials, + excluded_trials: observerSummary.excluded_trials, + unconsumed_events_total: input.observer.reduce( + (sum, trial) => sum + Math.max(0, trial.expectedObservedEvents - trial.observedEvents), 0, + ), + unsupported_envelopes_total: input.observer.reduce( + (sum, trial) => sum + (trial.envelopeKinds.unsupported ?? 0), 0, + ), + }, + limits: REPORT_LIMITS, + } +} + +function formatNumber(value: number, digits = 2): string { + return Number.isFinite(value) ? value.toFixed(digits) : 'n/a' +} + +function formatInteger(value: number): string { + return Number.isFinite(value) ? Math.round(value).toLocaleString('en-US') : 'n/a' +} + +function formatBytes(value: number): string { + const sign = value < 0 ? '-' : '' + const absolute = Math.abs(value) + if (absolute < 1_024) return `${sign}${absolute.toFixed(0)} B` + if (absolute < 1_024 ** 2) return `${sign}${(absolute / 1_024).toFixed(1)} KiB` + return `${sign}${(absolute / 1_024 ** 2).toFixed(1)} MiB` +} + +function status(value: boolean): string { + return value ? 'passed' : 'failed' +} + +export function renderMarkdown(report: OverheadReport): string { + const baseline = report.arms.baseline.summary + const observer = report.arms.observer.summary + const requested = report.config.requested_events + const eventsLine = requested !== undefined && requested !== report.config.events + ? `${formatInteger(report.config.events)} (requested ${formatInteger(requested)}; ` + + 'normalized to whole 6-event turn cycles)' + : formatInteger(report.config.events) + const warnings = report.warnings.length === 0 + ? '' + : `\n## Warnings\n\n${report.warnings.map(warning => `- ${warning}`).join('\n')}\n` + const baselineTiming = (value: number, digits = 2): string => + baseline.usable_trials === 0 ? 'n/a' : formatNumber(value, digits) + const observerTiming = (value: number, digits = 2): string => + observer.usable_trials === 0 ? 'n/a' : formatNumber(value, digits) + const baselineBytes = (value: number): string => + baseline.usable_trials === 0 ? 'n/a' : formatBytes(value) + const observerBytes = (value: number): string => + observer.usable_trials === 0 ? 'n/a' : formatBytes(value) + const observerInteger = (value: number): string => + observer.usable_trials === 0 ? 'n/a' : formatInteger(value) + return `# Observer overhead (synthetic DSH session) + +- Status: **${report.usable ? 'usable' : 'NOT USABLE'}** +- Generated: ${report.generated_at} +- Node: ${report.environment.node} +- Platform: ${report.environment.platform}/${report.environment.arch} +- CPUs: ${report.environment.cpus} +- Memory: ${formatNumber(report.environment.total_memory_gib, 1)} GiB +- Events per trial: ${eventsLine} +- Repeats per arm: ${formatInteger(report.config.repeat)} +- Observer trials used for timing: ${formatInteger(observer.usable_trials)} of ${formatInteger(observer.trials)} +- Buffer bound: ${formatInteger(report.config.buffer_bound)} +- Event mix: ${report.config.event_mix} + +| Metric | Baseline | Observer | Delta | +| --- | ---: | ---: | ---: | +| CPU total, median (ms) | ${baselineTiming(baseline.cpu_total_ms_median)} | ${observerTiming(observer.cpu_total_ms_median)} | ${observerTiming(observer.cpu_total_ms_median - baseline.cpu_total_ms_median)} | +| CPU, median (ms / 1k events) | ${baselineTiming(baseline.cpu_ms_per_1k_events_median)} | ${observerTiming(observer.cpu_ms_per_1k_events_median)} | ${observerTiming(report.delta.cpu_ms_per_1k_events)} | +| RSS delta, median | ${baselineBytes(baseline.rss_delta_bytes_median)} | ${observerBytes(observer.rss_delta_bytes_median)} | ${observerBytes(report.delta.rss_delta_bytes)} | +| Ledger append p50 (ms) | n/a | ${observerTiming(observer.append_p50_ms)} | n/a | +| Ledger append p95 (ms) | n/a | ${observerTiming(observer.append_p95_ms)} | n/a | +| Ledger bytes / envelope | n/a | ${observerTiming(observer.ledger_bytes_per_event)} | n/a | +| Ledger stats bytes / flush | n/a | ${observerTiming(observer.stats_bytes_per_flush)} | n/a | +| Persisted envelopes, median | n/a | ${observerInteger(observer.persisted_events_median)} | n/a | +| Clock source, median (harness / wall-clock) | n/a | ${formatInteger(report.event_accounting.clock_harness_time_events_median)} / ${formatInteger(report.event_accounting.clock_wall_clock_events_median)} | n/a | +| Clock uncertainty max, median (ms) | n/a | ${formatNumber(report.event_accounting.clock_uncertainty_max_ms_median)} | n/a | + +## Event accounting (observer, all trials) + +| Observed | Accepted | Persisted | Dropped (total) | Dropped (max) | Rejected | Flush attempts | Peak buffer (max) | +| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| ${formatInteger(report.event_accounting.observed_events_median)} | ${formatInteger(report.event_accounting.accepted_events_median)} | ${formatInteger(report.event_accounting.persisted_events_median)} | ${formatInteger(report.event_accounting.dropped_events_total)} | ${formatInteger(report.event_accounting.dropped_events_max)} | ${formatInteger(report.event_accounting.rejected_events_median)} | ${formatInteger(report.event_accounting.flush_attempts_median)} | ${formatInteger(report.event_accounting.peak_buffered_events_max)} | + +- Trials with loss: ${formatInteger(report.event_accounting.trials_with_loss)} of ${formatInteger(observer.trials)} +- Trials used for timing: ${formatInteger(report.event_accounting.usable_trials)} of ${formatInteger(observer.trials)} (${formatInteger(report.event_accounting.excluded_trials)} excluded) +- Planned records not observed: ${formatInteger(report.event_accounting.unconsumed_events_total)} +- Unsupported envelopes: ${formatInteger(report.event_accounting.unsupported_envelopes_total)} + +## Integrity + +- No event loss: ${status(report.integrity.no_event_loss)} +- Persisted equals accepted: ${status(report.integrity.persisted_equals_accepted)} +- All trials usable: ${status(report.integrity.all_trials_usable)} +- Observed matches plan: ${status(report.integrity.observed_matches_plan)} +- Planned envelope mix complete: ${status(report.integrity.envelope_kind_mix_complete)} +- Count conservation: ${status(report.integrity.count_conservation)} +- Outbound endpoints empty: ${status(report.integrity.outbound_endpoints_empty)} +- Worker-context influence absent: ${status(report.integrity.worker_context_influence_absent)} +- Scheduler-input influence absent: ${status(report.integrity.scheduler_input_influence_absent)} +- Observer failures: ${formatInteger(report.integrity.observer_failures)} +- Persisted envelopes: ${formatInteger(report.integrity.persisted_envelopes)} +${warnings} +## Limits + +${report.limits.map(limit => `- ${limit}`).join('\n')} +` +} diff --git a/packages/dsh-loopx-plugin/bench/overhead-runner.ts b/packages/dsh-loopx-plugin/bench/overhead-runner.ts new file mode 100644 index 0000000000..9fde55d0de --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/overhead-runner.ts @@ -0,0 +1,356 @@ +import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { performance } from 'node:perf_hooks' +import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' +import { appendLedgerLines, ShadowObserver } from '../src/observer.ts' +import type { ObserverRunIdentity, ShadowObserverConfig } from '../src/observer.ts' + +const GOAL_ID = 'goal-observer-overhead-fixture' +const SESSION_ID = 'session-observer-overhead-fixture' +const OBSERVER_ID = 'observer-overhead-fixture' +const EVENT_CYCLE = 6 +const EVENT_TIME_BASE = 1_756_728_000_000 + +const RUN_IDENTITY: ObserverRunIdentity = { + worker_id: 'worker-overhead-fixture', + model_id: 'model-overhead-fixture', + task_id: 'task-overhead-fixture', + environment_id: 'environment-overhead-fixture', + tools_id: 'tools-overhead-fixture', + budget_id: 'budget-overhead-fixture', + adapter_revision: 'adapter-overhead-fixture', + observer_revision: 'observer-overhead-fixture', +} + +const SESSION = { id: SESSION_ID } as unknown as Session + +export type PlannedEvent = + | { readonly kind: 'created' } + | { readonly kind: 'event'; readonly event: SessionEvent } + | { readonly kind: 'disposed' } + +export interface TrialOptions { + readonly inputEvents: number + readonly bufferBound: number + readonly failAppend?: boolean +} + +export interface TrialMeasurement { + readonly workloadEvents: number + readonly observedEvents: number + readonly expectedObservedEvents: number + readonly acceptedEvents: number + readonly persistedEvents: number + readonly droppedEvents: number + readonly rejectedEvents: number + readonly flushAttempts: number + readonly peakBufferedEvents: number + readonly observerFailures: number + readonly cpuUserMs: number + readonly cpuSystemMs: number + readonly cpuTotalMs: number + readonly rssBeforeBytes: number + readonly rssAfterBytes: number + readonly rssDeltaBytes: number + readonly wallMs: number + readonly ledgerBytes: number + readonly envelopeBytes: number + readonly statsBytes: number + readonly clockUncertaintyMaxMs: number + readonly harnessTimeEvents: number + readonly wallClockEvents: number + readonly envelopeKinds: Readonly> + readonly expectedEnvelopeKinds: Readonly> + readonly appendLatenciesMs: readonly number[] + readonly outboundEndpoints: number + readonly workerContextInfluence: boolean + readonly schedulerInputInfluence: boolean +} + +export function normalizeEventCount(inputEvents: number): number { + return Math.max(EVENT_CYCLE, Math.floor(inputEvents / EVENT_CYCLE) * EVENT_CYCLE) +} + +/** + * The record mix one trial must produce for its timings to be comparable: one + * lifecycle record per session and one record per planned event kind per turn + * cycle. Each trial carries this histogram so a silently retired event kind or + * an unknown-kind substitution can never read as a clean measurement. + */ +export function plannedEnvelopeKinds(workloadEvents: number): Readonly> { + const cycles = Math.floor(workloadEvents / EVENT_CYCLE) + return { + session_started: 1, + session_disposed: 1, + turn_started: cycles, + step_started: cycles, + tool_called: cycles, + tool_completed: cycles, + step_ended: cycles, + turn_ended: cycles, + } +} + +function sessionEvent(type: string, seq: number, data: unknown): SessionEvent { + return { + type, + seq, + time: EVENT_TIME_BASE + seq * 1_000, + data, + } as unknown as SessionEvent +} + +/** + * A deterministic turn cycle: one start/step/tool/result/end per turn, plus a + * terminal disposal. The tool intentionally avoids random data so two runs + * compare the same serialized payloads. + */ +export function buildEventPlan(inputEvents: number): readonly PlannedEvent[] { + const normalized = normalizeEventCount(inputEvents) + const cycles = normalized / EVENT_CYCLE + const plan: PlannedEvent[] = [{ kind: 'created' }] + let seq = 1 + for (let turn = 1; turn <= cycles; turn += 1) { + plan.push({ kind: 'event', event: sessionEvent('turn/start', seq, { turn }) }) + seq += 1 + plan.push({ kind: 'event', event: sessionEvent('step/start', seq, { turn, step: 1 }) }) + seq += 1 + plan.push({ + kind: 'event', + event: sessionEvent('tool/call', seq, { + turn, + step: 1, + callId: `call-${turn}`, + name: 'read', + }), + }) + seq += 1 + plan.push({ + kind: 'event', + event: sessionEvent('tool/result', seq, { + turn, + step: 1, + message: { source: { callId: `call-${turn}` } }, + }), + }) + seq += 1 + plan.push({ kind: 'event', event: sessionEvent('step/end', seq, { turn, step: 1 }) }) + seq += 1 + plan.push({ + kind: 'event', + event: sessionEvent('turn/end', seq, { turn, reason: { kind: 'completed' } }), + }) + seq += 1 + } + plan.push({ kind: 'disposed' }) + return plan +} + +function safeGc(): void { + const gc = (globalThis as { gc?: () => void }).gc + gc?.() +} + +function yieldToEventLoop(): Promise { + return new Promise(resolve => { + setImmediate(resolve) + }) +} + +function cpuMeasurement(cpu: NodeJS.CpuUsage): { + readonly userMs: number + readonly systemMs: number + readonly totalMs: number +} { + return { + userMs: cpu.user / 1_000, + systemMs: cpu.system / 1_000, + totalMs: (cpu.user + cpu.system) / 1_000, + } +} + +export function runBaselineTrial(options: TrialOptions): TrialMeasurement { + const plan = buildEventPlan(options.inputEvents) + const workloadEvents = normalizeEventCount(options.inputEvents) + safeGc() + const rssBeforeBytes = process.memoryUsage().rss + const cpuBefore = process.cpuUsage() + const wallStart = performance.now() + let checksum = 0 + for (const planned of plan) { + if (planned.kind === 'event') checksum += planned.event.seq + else checksum += 1 + } + const wallMs = performance.now() - wallStart + const cpu = cpuMeasurement(process.cpuUsage(cpuBefore)) + const rssAfterBytes = process.memoryUsage().rss + if (!Number.isFinite(checksum) || checksum <= 0) { + throw new Error('synthetic baseline event stream was empty') + } + return { + workloadEvents, + observedEvents: plan.length, + expectedObservedEvents: plan.length, + cpuUserMs: cpu.userMs, + cpuSystemMs: cpu.systemMs, + cpuTotalMs: cpu.totalMs, + rssBeforeBytes, + rssAfterBytes, + rssDeltaBytes: rssAfterBytes - rssBeforeBytes, + wallMs, + acceptedEvents: 0, + persistedEvents: 0, + droppedEvents: 0, + rejectedEvents: 0, + flushAttempts: 0, + peakBufferedEvents: 0, + observerFailures: 0, + ledgerBytes: 0, + envelopeBytes: 0, + statsBytes: 0, + clockUncertaintyMaxMs: 0, + harnessTimeEvents: 0, + wallClockEvents: 0, + // The baseline driver runs no observer and persists no envelopes. + envelopeKinds: {}, + expectedEnvelopeKinds: {}, + appendLatenciesMs: [], + outboundEndpoints: 0, + workerContextInfluence: false, + schedulerInputInfluence: false, + } +} + +export async function runObserverTrial(options: TrialOptions): Promise { + const plan = buildEventPlan(options.inputEvents) + const workloadEvents = normalizeEventCount(options.inputEvents) + const ledgerDir = await mkdtemp(join(tmpdir(), 'loopx-observer-overhead-')) + const appendLatenciesMs: number[] = [] + try { + const config: ShadowObserverConfig = { + goalId: GOAL_ID, + sessionId: SESSION_ID, + runIdentity: RUN_IDENTITY, + ledgerDir, + bufferBound: options.bufferBound, + } + const observer = new ShadowObserver({ + config, + now: () => EVENT_TIME_BASE, + observerId: OBSERVER_ID, + // Delegate to the shipped writer so the measured append path (including + // the legacy-ledger refusal and directory creation) cannot drift from + // what the observer actually does in production. Only the timing wraps it. + appendLines: async (_path, lines) => { + if (options.failAppend === true) { + throw new Error('synthetic ledger append failure') + } + const start = performance.now() + await appendLedgerLines(config, lines) + appendLatenciesMs.push(performance.now() - start) + }, + }) + safeGc() + const rssBeforeBytes = process.memoryUsage().rss + const cpuBefore = process.cpuUsage() + const wallStart = performance.now() + observer.observeSessionCreated(SESSION) + for (const planned of plan) { + if (planned.kind === 'created') continue + if (planned.kind === 'event') { + observer.observeSessionEvent(SESSION, planned.event) + if (planned.event.type === 'turn/end') { + // One macrotask tick lets a completed append resolve. An append still + // in flight coalesces with this turn's flush request instead of being + // awaited; the report prints the actual flush attempts. + await yieldToEventLoop() + } + } else { + observer.observeSessionDisposed(SESSION) + } + } + await observer.dispose() + const wallMs = performance.now() - wallStart + const cpu = cpuMeasurement(process.cpuUsage(cpuBefore)) + const rssAfterBytes = process.memoryUsage().rss + const stats = observer.stats() + let persistedEvents = 0 + let envelopeBytes = 0 + let statsBytes = 0 + let clockUncertaintyMaxMs = 0 + let harnessTimeEvents = 0 + let wallClockEvents = 0 + let envelopeKinds: Record = {} + try { + const text = await readFile(observer.path, 'utf8') + for (const line of text.split('\n')) { + if (line === '') continue + const lineBytes = Buffer.byteLength(line, 'utf8') + 1 + if (!line.includes('reliability_observer_envelope_v0')) { + statsBytes += lineBytes + continue + } + persistedEvents += 1 + envelopeBytes += lineBytes + try { + const record = JSON.parse(line) as { + clock?: { source?: string, uncertainty_ms?: number }, + event_kind?: string, + } + const source = record.clock?.source + if (source === 'harness_event_time') harnessTimeEvents += 1 + else if (typeof source === 'string') wallClockEvents += 1 + const uncertainty = record.clock?.uncertainty_ms + if (typeof uncertainty === 'number' && Number.isFinite(uncertainty)) { + clockUncertaintyMaxMs = Math.max(clockUncertaintyMaxMs, uncertainty) + } + const kind = record.event_kind + if (typeof kind === 'string') envelopeKinds[kind] = (envelopeKinds[kind] ?? 0) + 1 + } catch { + // A malformed line still counts as persisted; integrity reports it. + } + } + } catch { + persistedEvents = 0 + envelopeBytes = 0 + statsBytes = 0 + harnessTimeEvents = 0 + wallClockEvents = 0 + envelopeKinds = {} + } + return { + workloadEvents, + observedEvents: stats.observed_event_count, + expectedObservedEvents: plan.length, + cpuUserMs: cpu.userMs, + cpuSystemMs: cpu.systemMs, + cpuTotalMs: cpu.totalMs, + rssBeforeBytes, + rssAfterBytes, + rssDeltaBytes: rssAfterBytes - rssBeforeBytes, + wallMs, + acceptedEvents: stats.accepted_event_count, + persistedEvents, + droppedEvents: stats.backpressure_drop_count, + rejectedEvents: stats.rejected_event_count, + flushAttempts: stats.flush_attempt_count, + peakBufferedEvents: stats.peak_buffered_event_count, + observerFailures: stats.observer_failure_count, + ledgerBytes: envelopeBytes + statsBytes, + envelopeBytes, + statsBytes, + clockUncertaintyMaxMs, + harnessTimeEvents, + wallClockEvents, + envelopeKinds, + expectedEnvelopeKinds: plannedEnvelopeKinds(workloadEvents), + appendLatenciesMs, + outboundEndpoints: stats.outbound_endpoints.length, + workerContextInfluence: stats.observation_entered_worker_context, + schedulerInputInfluence: stats.observation_entered_scheduler_inputs, + } + } finally { + await rm(ledgerDir, { recursive: true, force: true }) + } +} diff --git a/packages/dsh-loopx-plugin/package.json b/packages/dsh-loopx-plugin/package.json index 2ed079a3ca..c041e12375 100644 --- a/packages/dsh-loopx-plugin/package.json +++ b/packages/dsh-loopx-plugin/package.json @@ -50,6 +50,7 @@ }, "scripts": { "build": "node scripts/build.mjs", + "bench:observer": "node --experimental-strip-types bench/observer-overhead.ts", "typecheck": "tsc -p tsconfig.host.json --noEmit && tsc -p tsconfig.client.json --noEmit && tsc -p tsconfig.tests.json", "test": "vitest run", "prepack": "node scripts/build.mjs", diff --git a/packages/dsh-loopx-plugin/src/observer.ts b/packages/dsh-loopx-plugin/src/observer.ts index e1fb05ccf0..bedc06025d 100644 --- a/packages/dsh-loopx-plugin/src/observer.ts +++ b/packages/dsh-loopx-plugin/src/observer.ts @@ -278,7 +278,12 @@ export function ledgerPath(config: ShadowObserverConfig): string { return join(config.ledgerDir, 'by-goal', `${digest}.ndjson`) } -async function appendLedgerLines(config: ShadowObserverConfig, lines: readonly string[]): Promise { +/** + * The shipped ledger writer. Exported so the overhead bench measures this exact + * path (legacy-ledger refusal, directory creation, and append) instead of a + * bench-local copy that would drift from the production writer. + */ +export async function appendLedgerLines(config: ShadowObserverConfig, lines: readonly string[]): Promise { if (lines.length === 0) return const legacy = join(config.ledgerDir, `${config.goalId.replaceAll(':', '_')}.ndjson`) if (lstatSync(legacy, { throwIfNoEntry: false })) { diff --git a/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts b/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts new file mode 100644 index 0000000000..2dde782dea --- /dev/null +++ b/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts @@ -0,0 +1,290 @@ +import { describe, expect, it } from 'vitest' +import { + buildEventPlan, + normalizeEventCount, + runBaselineTrial, + runObserverTrial, +} from '../bench/overhead-runner.ts' +import type { PlannedEvent } from '../bench/overhead-runner.ts' +import { + buildOverheadReport, + isUsableTrial, + percentile, + renderMarkdown, + summarizeTrials, +} from '../bench/overhead-report.ts' +import { parseArgs } from '../bench/observer-overhead.ts' + +const environment = { + node: 'v-test', + platform: 'test', + arch: 'test', + cpus: 1, + total_memory_gib: 1, +} + +describe('observer overhead event plan', () => { + it('normalizes to whole turn cycles and pins the stream shape', () => { + expect(normalizeEventCount(0)).toBe(6) + expect(normalizeEventCount(10)).toBe(6) + expect(normalizeEventCount(13)).toBe(12) + const plan = buildEventPlan(12) + expect(plan).toHaveLength(14) + expect(plan[0]?.kind).toBe('created') + expect(plan.at(-1)?.kind).toBe('disposed') + const eventTypes = plan + .filter((planned): planned is Extract => planned.kind === 'event') + .map(planned => planned.event.type) + expect(eventTypes).toEqual([ + 'turn/start', 'step/start', 'tool/call', 'tool/result', 'step/end', 'turn/end', + 'turn/start', 'step/start', 'tool/call', 'tool/result', 'step/end', 'turn/end', + ]) + }) + + it('keeps percentile boundaries deterministic', () => { + expect(percentile([], 0.5)).toBe(0) + expect(percentile([1, 2, 3, 4], 0.5)).toBe(2) + expect(percentile([1, 2, 3, 4], 0.95)).toBe(4) + expect(percentile([4, 1, 3, 2], 0.5)).toBe(2) + }) +}) + +describe('observer overhead trials', () => { + it('measures observer component cost with full denominators and a clock mix', async () => { + const trial = await runObserverTrial({ inputEvents: 60, bufferBound: 256 }) + expect(trial.workloadEvents).toBe(60) + expect(trial.observedEvents).toBe(62) + expect(trial.expectedObservedEvents).toBe(62) + expect(trial.envelopeKinds).toEqual({ + session_started: 1, + session_disposed: 1, + turn_started: 10, + step_started: 10, + tool_called: 10, + tool_completed: 10, + step_ended: 10, + turn_ended: 10, + }) + expect(trial.observedEvents).toBe( + trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, + ) + expect(trial.persistedEvents).toBe(trial.acceptedEvents) + expect(trial.persistedEvents).toBeGreaterThan(0) + expect(trial.ledgerBytes).toBe(trial.envelopeBytes + trial.statsBytes) + expect(trial.envelopeBytes).toBeGreaterThan(0) + expect(trial.statsBytes).toBeGreaterThan(0) + expect(trial.clockUncertaintyMaxMs).toBe(50) + expect(trial.harnessTimeEvents).toBe(60) + expect(trial.wallClockEvents).toBe(2) + expect(trial.appendLatenciesMs.length).toBeGreaterThan(0) + expect(trial.outboundEndpoints).toBe(0) + expect(trial.workerContextInfluence).toBe(false) + expect(trial.schedulerInputInfluence).toBe(false) + expect(isUsableTrial(trial)).toBe(true) + }) + + it('counts backpressure loss as an unusable trial', async () => { + const trial = await runObserverTrial({ inputEvents: 60, bufferBound: 4 }) + expect(trial.droppedEvents).toBeGreaterThan(0) + expect(trial.observedEvents).toBe( + trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, + ) + expect(trial.persistedEvents).toBe(trial.acceptedEvents) + expect(isUsableTrial(trial)).toBe(false) + }) + + it('counts append failures without throwing or persisting controls', async () => { + const trial = await runObserverTrial({ + inputEvents: 12, + bufferBound: 4, + failAppend: true, + }) + expect(trial.observerFailures).toBeGreaterThan(0) + expect(trial.persistedEvents).toBe(0) + expect(trial.observedEvents).toBe( + trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, + ) + expect(trial.outboundEndpoints).toBe(0) + expect(isUsableTrial(trial)).toBe(false) + }) + + it('renders a public-safe usable report with both arms and integrity status', async () => { + const baseline = [runBaselineTrial({ inputEvents: 12, bufferBound: 256 })] + const observer = [await runObserverTrial({ inputEvents: 12, bufferBound: 256 })] + expect(summarizeTrials(baseline).trials).toBe(1) + expect(summarizeTrials(baseline).usable_trials).toBe(1) + expect(summarizeTrials(observer).trials).toBe(1) + expect(summarizeTrials(observer).usable_trials).toBe(1) + const report = buildOverheadReport({ + config: { + events: 12, + requested_events: 13, + repeat: 1, + buffer_bound: 256, + event_mix: 'turn-cycle-v1', + }, + baseline, + observer, + environment, + generatedAt: '2026-10-19T00:00:00.000Z', + }) + expect(report.schema).toBe('reliability_observer_overhead_report_v0') + expect(report.usable).toBe(true) + expect(report.warnings).toEqual([]) + expect(report.integrity.no_event_loss).toBe(true) + expect(report.integrity.persisted_equals_accepted).toBe(true) + expect(report.integrity.all_trials_usable).toBe(true) + expect(report.integrity.observed_matches_plan).toBe(true) + expect(report.integrity.envelope_kind_mix_complete).toBe(true) + expect(report.integrity.count_conservation).toBe(true) + expect(report.integrity.outbound_endpoints_empty).toBe(true) + expect(report.integrity.worker_context_influence_absent).toBe(true) + expect(report.integrity.scheduler_input_influence_absent).toBe(true) + expect(report.event_accounting.observed_events_median).toBe(14) + expect(report.event_accounting.dropped_events_total).toBe(0) + expect(report.event_accounting.dropped_events_max).toBe(0) + expect(report.event_accounting.trials_with_loss).toBe(0) + expect(report.event_accounting.usable_trials).toBe(1) + expect(report.event_accounting.peak_buffered_events_max).toBeGreaterThan(0) + expect(report.event_accounting.clock_uncertainty_max_ms_median).toBe(50) + expect(report.event_accounting.clock_harness_time_events_median).toBe(12) + expect(report.event_accounting.clock_wall_clock_events_median).toBe(2) + expect(report.arms.observer.summary.ledger_bytes_per_event).toBeGreaterThan(0) + expect(report.arms.observer.summary.stats_bytes_per_flush).toBeGreaterThan(0) + expect(Number.isFinite(report.delta.cpu_ms_per_1k_events)).toBe(true) + expect(Number.isFinite(report.delta.rss_delta_bytes)).toBe(true) + const markdown = renderMarkdown(report) + expect(markdown).toContain('Observer overhead') + expect(markdown).toContain('Status: **usable**') + expect(markdown).toContain('No event loss: passed') + expect(markdown).toContain('Persisted equals accepted: passed') + expect(markdown).toContain('Events per trial: 12 (requested 13; normalized to whole 6-event turn cycles)') + expect(markdown).not.toContain('## Warnings') + expect(markdown).not.toContain('session-observer-overhead-fixture') + expect(markdown).not.toContain('goal-observer-overhead-fixture') + expect(JSON.stringify(report)).not.toContain('session-observer-overhead-fixture') + expect(JSON.stringify(report)).not.toContain('goal-observer-overhead-fixture') + expect(JSON.stringify(report)).not.toContain('loopx-observer-overhead-') + }) + + it('marks a lossy run unusable instead of hiding loss behind medians', async () => { + const baseline = [runBaselineTrial({ inputEvents: 60, bufferBound: 4 })] + const lossy = await runObserverTrial({ inputEvents: 60, bufferBound: 4 }) + const clean = await runObserverTrial({ inputEvents: 60, bufferBound: 256 }) + expect(lossy.droppedEvents).toBeGreaterThan(0) + expect(clean.droppedEvents).toBe(0) + const report = buildOverheadReport({ + config: { events: 60, repeat: 2, buffer_bound: 4, event_mix: 'turn-cycle-v1' }, + baseline, + observer: [lossy, clean], + environment, + generatedAt: '2026-10-19T00:00:00.000Z', + }) + expect(report.usable).toBe(false) + expect(report.integrity.no_event_loss).toBe(false) + expect(report.integrity.all_trials_usable).toBe(false) + expect(report.event_accounting.trials_with_loss).toBe(1) + expect(report.event_accounting.dropped_events_total).toBe(lossy.droppedEvents) + expect(report.event_accounting.dropped_events_max).toBe(lossy.droppedEvents) + expect(report.arms.observer.summary.trials).toBe(2) + expect(report.arms.observer.summary.usable_trials).toBe(1) + expect(report.arms.observer.summary.excluded_trials).toBe(1) + expect(report.arms.observer.summary.persisted_events_median).toBe(clean.persistedEvents) + expect(report.warnings.length).toBeGreaterThan(0) + const markdown = renderMarkdown(report) + expect(markdown).toContain('Status: **NOT USABLE**') + expect(markdown).toContain('No event loss: failed') + expect(markdown).toContain('Trials with loss: 1 of 2') + expect(markdown).toContain('## Warnings') + }) + + it('marks an append-failed run unusable without persisting', async () => { + const trial = await runObserverTrial({ + inputEvents: 12, + bufferBound: 4, + failAppend: true, + }) + const report = buildOverheadReport({ + config: { events: 12, repeat: 1, buffer_bound: 4, event_mix: 'turn-cycle-v1' }, + baseline: [runBaselineTrial({ inputEvents: 12, bufferBound: 4 })], + observer: [trial], + environment, + generatedAt: '2026-10-19T00:00:00.000Z', + }) + expect(report.usable).toBe(false) + expect(report.integrity.observer_failures).toBeGreaterThan(0) + expect(report.integrity.no_event_loss).toBe(false) + expect(report.warnings.length).toBeGreaterThan(0) + expect(renderMarkdown(report)).toContain('Status: **NOT USABLE**') + }) + + it('marks a trial unusable when the planned record mix is not consumed', async () => { + const trial = await runObserverTrial({ inputEvents: 60, bufferBound: 256 }) + expect(isUsableTrial(trial)).toBe(true) + // A compactor that silently skips one planned kind (retirement or rename) + // must not read as a usable overhead sample. + const skippedKind = { + ...trial, + observedEvents: trial.observedEvents - 10, + acceptedEvents: trial.acceptedEvents - 10, + persistedEvents: trial.persistedEvents - 10, + envelopeKinds: { ...trial.envelopeKinds, turn_started: 0 }, + } + expect(isUsableTrial(skippedKind)).toBe(false) + // A planned kind that no longer maps to a known kind still counts as an + // observed record, so only the kind histogram can catch it. + const substituted = { + ...trial, + envelopeKinds: { ...trial.envelopeKinds, turn_started: 0, unsupported: 10 }, + } + expect(isUsableTrial(substituted)).toBe(false) + const report = buildOverheadReport({ + config: { events: 60, repeat: 2, buffer_bound: 256, event_mix: 'turn-cycle-v1' }, + baseline: [runBaselineTrial({ inputEvents: 60, bufferBound: 256 })], + observer: [skippedKind, substituted], + environment, + generatedAt: '2026-10-19T00:00:00.000Z', + }) + expect(report.usable).toBe(false) + expect(report.integrity.observed_matches_plan).toBe(false) + expect(report.integrity.envelope_kind_mix_complete).toBe(false) + expect(report.event_accounting.unconsumed_events_total).toBe(10) + expect(report.event_accounting.unsupported_envelopes_total).toBe(10) + expect(report.arms.observer.summary.usable_trials).toBe(0) + const markdown = renderMarkdown(report) + expect(markdown).toContain('Status: **NOT USABLE**') + expect(markdown).toContain('Observed matches plan: failed') + expect(markdown).toContain('Planned envelope mix complete: failed') + expect(markdown).toContain('## Warnings') + }) +}) + +describe('observer overhead CLI arguments', () => { + it('parses defaults and explicit values', () => { + expect(parseArgs([])).toMatchObject({ + events: 5_000, + repeat: 3, + bufferBound: 4_096, + help: false, + }) + expect(parseArgs(['--events', '10', '--repeat', '2', '--buffer-bound', '8'])).toMatchObject({ + events: 10, + repeat: 2, + bufferBound: 8, + }) + expect(parseArgs(['--json', '-', '--markdown', 'report.md'])).toMatchObject({ + json: '-', + markdown: 'report.md', + }) + expect(parseArgs(['--help']).help).toBe(true) + expect(parseArgs(['-h']).help).toBe(true) + }) + + it('rejects missing or invalid values', () => { + expect(() => parseArgs(['--events', '0'])).toThrow(/positive integer/) + expect(() => parseArgs(['--repeat'])).toThrow(/positive integer/) + expect(() => parseArgs(['--buffer-bound', 'nope'])).toThrow(/positive integer/) + expect(() => parseArgs(['--json'])).toThrow(/requires a path or "-"/) + expect(() => parseArgs(['--unknown'])).toThrow(/unknown argument/) + }) +}) diff --git a/packages/dsh-loopx-plugin/tsconfig.tests.json b/packages/dsh-loopx-plugin/tsconfig.tests.json index 54fdb3369f..922e35e696 100644 --- a/packages/dsh-loopx-plugin/tsconfig.tests.json +++ b/packages/dsh-loopx-plugin/tsconfig.tests.json @@ -9,6 +9,7 @@ "include": [ "src/**/*.ts", "src/**/*.tsx", + "bench/**/*.ts", "tests/**/*.ts", "tests/**/*.tsx" ]