From 0ec1e4143f694d47f0a272ed16da0d768c91fcf6 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Mon, 19 Oct 2026 10:07:27 +0800 Subject: [PATCH 1/3] feat(reliability): measure observer overhead with a synthetic DSH session Add a deterministic bench that runs the DSH shadow observer against the same synthetic turn cycle with the observer off and on. It reports CPU, RSS, ledger append latency, ledger bytes, event loss, clock uncertainty, and full denominators, and writes public-safe JSON/Markdown reports. The tool measures observer component cost only; it is not a live DSH host measurement, does not qualify C1, and does not change the default-off runtime. Related to #5211 Signed-off-by: JasonBuildAI --- .../reliability_diagnostics/README.md | 21 +- .../reliability_diagnostics/README.zh-CN.md | 18 +- packages/dsh-loopx-plugin/bench/README.md | 77 +++++ .../bench/observer-overhead.ts | 149 +++++++++ .../dsh-loopx-plugin/bench/overhead-report.ts | 238 ++++++++++++++ .../dsh-loopx-plugin/bench/overhead-runner.ts | 290 ++++++++++++++++++ packages/dsh-loopx-plugin/package.json | 1 + .../tests/observer-overhead.spec.ts | 110 +++++++ 8 files changed, 900 insertions(+), 4 deletions(-) create mode 100644 packages/dsh-loopx-plugin/bench/README.md create mode 100644 packages/dsh-loopx-plugin/bench/observer-overhead.ts create mode 100644 packages/dsh-loopx-plugin/bench/overhead-report.ts create mode 100644 packages/dsh-loopx-plugin/bench/overhead-runner.ts create mode 100644 packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts diff --git a/loopx/capabilities/reliability_diagnostics/README.md b/loopx/capabilities/reliability_diagnostics/README.md index d675798ccb..6231e1a225 100644 --- a/loopx/capabilities/reliability_diagnostics/README.md +++ b/loopx/capabilities/reliability_diagnostics/README.md @@ -209,6 +209,22 @@ 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 latency, ledger bytes, event loss, +and clock uncertainty with full denominators. 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 --repeat 5 +``` + ## Validation ```bash @@ -231,5 +247,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..308cbc6671 100644 --- a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md +++ b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md @@ -177,6 +177,20 @@ live 部署验收。操作 regular file 前会验证 provider 目录与 CLI 的 旧文件仍可只读回查，但必须离线核对后才能恢复追加；方案说明了升级和回退步骤， 归属不明确时须暂停删除。冻结 writer 后一起升级 CLI 与插件。 +## 测量 observer 开销 + +`packages/dsh-loopx-plugin/bench/observer-overhead.ts` 在确定性的 synthetic DSH 事件流上 +测量 observer 组件开销：observer 关闭/开启两臂、CPU（user+system）、RSS、ledger append +延迟、ledger 字节数、event loss 与 clock uncertainty，并给出完整分母。输出 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 --repeat 5 +``` + ## 验证 ```bash @@ -195,5 +209,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..c596bcfa5b --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/README.md @@ -0,0 +1,77 @@ +# 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**: the same event stream with the observer disabled (baseline) and + enabled (observer). +- **CPU**: `process.cpuUsage()` user + system delta. +- **RSS**: process RSS before/after each trial. +- **Ledger append**: p50/p95 latency and bytes written per persisted envelope. +- **Event accounting**: observed, accepted, persisted, dropped, rejected, peak + buffer, and flush attempts. Counts must satisfy + `observed = accepted + rejected + dropped`. +- **Integrity**: no outbound endpoints, no worker-context influence, and no + scheduler-input influence. + +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 --repeat 5 --buffer-bound 256 \ + --json ../../.local/observer-overhead.json \ + --markdown ../../.local/observer-overhead.md +``` + +Defaults are 5,000 events, 3 repeats, and a buffer bound of 256. The tool runs a +short warmup, then alternates baseline and observer trials. Use `--json -` or +`--markdown -` to print a report to stdout. + +Example output from one Windows/Node 26 run (numbers are machine-specific and +are not a qualification): + +```text +| Metric | Baseline | Observer | Delta | +| --- | ---: | ---: | ---: | +| CPU total, median (ms) | 0.0 | 609.0 | 609.0 | +| CPU, median (ms / 1k events) | 0.0 | 121.9 | 121.9 | +| RSS delta, median | 204.0 KiB | 8.2 MiB | 8.0 MiB | +| Ledger append p95 (ms) | n/a | 2.68 | n/a | +| Ledger bytes / persisted event | n/a | 521.6 | n/a | +| Persisted envelopes, median | n/a | 5,000 | n/a | +``` + +The synthetic turn cycle has six events per turn and flushes on every turn +end, so it is deliberately flush-heavy. A real long turn flushes less often; +use the per-event numbers as a bound/signal, not a workload forecast. + + +## 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, count +conservation, append-failure accounting, 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..cc232785fa --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/observer-overhead.ts @@ -0,0 +1,149 @@ +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' + +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 +} + +function parseArgs(args: readonly string[]): CliOptions { + let events = 5_000 + let repeat = 3 + let bufferBound = 256 + 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 (default: 5000) + --repeat measured trials per arm (default: 3) + --buffer-bound observer buffer bound (default: 256) + --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) + 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, + 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` + + 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..eeab69eeaa --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/overhead-report.ts @@ -0,0 +1,238 @@ +import type { TrialMeasurement } from './overhead-runner.ts' + +export interface ArmSummary { + readonly 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 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 repeat: number + readonly buffer_bound: number + readonly event_mix: string +} + +export interface OverheadIntegrity { + readonly count_conservation: 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 rejected_events_median: number + readonly flush_attempts_median: number + readonly clock_uncertainty_max_ms_median: number +} + +export interface OverheadDelta { + readonly cpu_ms_per_1k_events: number + readonly rss_delta_bytes: number + readonly append_p95_ms: number +} + +export interface OverheadReport { + readonly schema: 'reliability_observer_overhead_report_v0' + readonly generated_at: 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); excludes Cordis hook dispatch, DSH scheduling, model, and network.', + 'The synthetic turn cycle has six events per turn and flushes on every turn end; a real long turn flushes less often, so the per-event append cost here is a flush-heavy microbenchmark, 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 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) +} + +export function summarizeTrials(trials: readonly TrialMeasurement[]): ArmSummary { + if (trials.length === 0) { + return { + trials: 0, + cpu_total_ms_median: 0, + cpu_total_ms_p95: 0, + cpu_ms_per_1k_events_median: 0, + rss_delta_bytes_median: 0, + ledger_bytes_per_event: 0, + append_p50_ms: 0, + append_p95_ms: 0, + persisted_events_median: 0, + } + } + const cpuPer1k = trials.map(trial => trial.workloadEvents === 0 + ? 0 + : (trial.cpuTotalMs / trial.workloadEvents) * 1_000) + const appendLatencies = trials.flatMap(trial => trial.appendLatenciesMs) + const totalLedgerBytes = trials.reduce((sum, trial) => sum + trial.ledgerBytes, 0) + const totalPersisted = trials.reduce((sum, trial) => sum + trial.persistedEvents, 0) + return { + trials: trials.length, + cpu_total_ms_median: median(trials.map(trial => trial.cpuTotalMs)), + cpu_total_ms_p95: percentile(trials.map(trial => trial.cpuTotalMs), 0.95), + cpu_ms_per_1k_events_median: median(cpuPer1k), + rss_delta_bytes_median: median(trials.map(trial => trial.rssDeltaBytes)), + ledger_bytes_per_event: totalPersisted === 0 ? 0 : totalLedgerBytes / totalPersisted, + append_p50_ms: percentile(appendLatencies, 0.5), + append_p95_ms: percentile(appendLatencies, 0.95), + persisted_events_median: median(trials.map(trial => trial.persistedEvents)), + } +} + +export function buildOverheadReport(input: OverheadReportInput): OverheadReport { + const baselineSummary = summarizeTrials(input.baseline) + const observerSummary = summarizeTrials(input.observer) + return { + schema: 'reliability_observer_overhead_report_v0', + generated_at: input.generatedAt ?? new Date().toISOString(), + 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, + append_p95_ms: observerSummary.append_p95_ms - baselineSummary.append_p95_ms, + }, + integrity: { + count_conservation: input.observer.every(trial => trial.observedEvents + === trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents), + 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(input.observer.map(trial => trial.droppedEvents)), + rejected_events_median: median(input.observer.map(trial => trial.rejectedEvents)), + flush_attempts_median: median(input.observer.map(trial => trial.flushAttempts)), + clock_uncertainty_max_ms_median: median(input.observer.map(trial => trial.clockUncertaintyMaxMs)), + }, + 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 + return `# Observer overhead (synthetic DSH session) + +- 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: ${formatInteger(report.config.events)} +- Repeats per arm: ${formatInteger(report.config.repeat)} +- Buffer bound: ${formatInteger(report.config.buffer_bound)} +- Event mix: ${report.config.event_mix} + +| Metric | Baseline | Observer | Delta | +| --- | ---: | ---: | ---: | +| CPU total, median (ms) | ${formatNumber(baseline.cpu_total_ms_median)} | ${formatNumber(observer.cpu_total_ms_median)} | ${formatNumber(observer.cpu_total_ms_median - baseline.cpu_total_ms_median)} | +| CPU, median (ms / 1k events) | ${formatNumber(baseline.cpu_ms_per_1k_events_median)} | ${formatNumber(observer.cpu_ms_per_1k_events_median)} | ${formatNumber(report.delta.cpu_ms_per_1k_events)} | +| RSS delta, median | ${formatBytes(baseline.rss_delta_bytes_median)} | ${formatBytes(observer.rss_delta_bytes_median)} | ${formatBytes(report.delta.rss_delta_bytes)} | +| Ledger append p50 (ms) | n/a | ${formatNumber(observer.append_p50_ms)} | n/a | +| Ledger append p95 (ms) | n/a | ${formatNumber(observer.append_p95_ms)} | n/a | +| Ledger bytes / persisted event | n/a | ${formatNumber(observer.ledger_bytes_per_event)} | n/a | +| Persisted envelopes, median | n/a | ${formatInteger(observer.persisted_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, median) + +| Observed | Accepted | Persisted | Dropped | Rejected | Flush attempts | +| ---: | ---: | ---: | ---: | ---: | ---: | +| ${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_median)} | ${formatInteger(report.event_accounting.rejected_events_median)} | ${formatInteger(report.event_accounting.flush_attempts_median)} | + +## Integrity + +- 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)} + +## 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..2faa091b27 --- /dev/null +++ b/packages/dsh-loopx-plugin/bench/overhead-runner.ts @@ -0,0 +1,290 @@ +import { appendFile, mkdir, mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, join } from 'node:path' +import { performance } from 'node:perf_hooks' +import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' +import { 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 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 clockUncertaintyMaxMs: number + 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) +} + +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, + 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, + clockUncertaintyMaxMs: 0, + 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-')) + let ledgerBytes = 0 + 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, + appendLines: async (path, lines) => { + if (options.failAppend === true) { + throw new Error('synthetic ledger append failure') + } + const start = performance.now() + const text = `${lines.join('\n')}\n` + await mkdir(dirname(path), { recursive: true }) + await appendFile(path, text, 'utf8') + ledgerBytes += Buffer.byteLength(text, 'utf8') + 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') { + // Let the observer's asynchronous flush drain between turns, as a + // real harness event loop would. + 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 clockUncertaintyMaxMs = 0 + try { + const text = await readFile(observer.path, 'utf8') + for (const line of text.split('\n')) { + if (!line.includes('reliability_observer_envelope_v0')) continue + persistedEvents += 1 + try { + const record = JSON.parse(line) as { clock?: { uncertainty_ms?: number } } + const uncertainty = record.clock?.uncertainty_ms + if (typeof uncertainty === 'number' && Number.isFinite(uncertainty)) { + clockUncertaintyMaxMs = Math.max(clockUncertaintyMaxMs, uncertainty) + } + } catch { + // A malformed line still counts as persisted; integrity reports it. + } + } + } catch { + persistedEvents = 0 + } + return { + workloadEvents, + observedEvents: stats.observed_event_count, + 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, + clockUncertaintyMaxMs, + 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/tests/observer-overhead.spec.ts b/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts new file mode 100644 index 0000000000..4055632a9b --- /dev/null +++ b/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts @@ -0,0 +1,110 @@ +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, + percentile, + renderMarkdown, + summarizeTrials, +} from '../bench/overhead-report.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', async () => { + const trial = await runObserverTrial({ inputEvents: 60, bufferBound: 8 }) + expect(trial.workloadEvents).toBe(60) + expect(trial.observedEvents).toBe(62) + expect(trial.observedEvents).toBe( + trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, + ) + expect(trial.persistedEvents).toBeGreaterThan(0) + expect(trial.ledgerBytes).toBeGreaterThan(0) + expect(trial.clockUncertaintyMaxMs).toBe(50) + expect(trial.appendLatenciesMs.length).toBeGreaterThan(0) + expect(trial.outboundEndpoints).toBe(0) + expect(trial.workerContextInfluence).toBe(false) + expect(trial.schedulerInputInfluence).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) + }) + + it('renders a public-safe report with both arms and integrity status', async () => { + const baseline = [runBaselineTrial({ inputEvents: 12, bufferBound: 8 })] + const observer = [await runObserverTrial({ inputEvents: 12, bufferBound: 8 })] + expect(summarizeTrials(baseline).trials).toBe(1) + expect(summarizeTrials(observer).trials).toBe(1) + const report = buildOverheadReport({ + config: { events: 12, repeat: 1, buffer_bound: 8, 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.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.clock_uncertainty_max_ms_median).toBe(50) + 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('Count conservation: passed') + 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') + }) +}) From 18b08550f28e8216f7add37f54483f22a57d5be1 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Mon, 19 Oct 2026 11:25:30 +0800 Subject: [PATCH 2/3] fix(reliability): surface lossy overhead trials and measure the shipped writer The observer-overhead bench summarized medians across trials and relied on an always-true count identity, so a trial that dropped thousands of events could still read as "Dropped 0 / integrity passed". It also timed a bench-local copy of the ledger writer, counted per-flush stats records as envelope bytes, and reported a structurally constant clock uncertainty. - Delegate the timed append to the shipped appendLedgerLines writer so the measured path cannot drift from production. - Split envelope and stats bytes; report bytes per envelope and stats bytes per flush. - Report per-trial loss totals and maxima, peak buffer, usable/excluded trials, and the harness-time/wall-clock envelope mix. - Exclude lossy, rejected, append-failed, or persisted-mismatched trials from the arm summaries, mark the report not usable, and print warnings in the Markdown and on CLI stderr. - Use a 4,096 buffer bound by default for the synthetic burst so a normal run does not spurious-drop; document that 256 mirrors the product default and exercises backpressure. - Document the --events normalization and print requested vs normalized counts. - Typecheck bench/**/*.ts and cover the CLI parser, loss marking, byte split, and public-safe rendering. Signed-off-by: JasonBuildAI --- .../reliability_diagnostics/README.md | 19 +- .../reliability_diagnostics/README.zh-CN.md | 17 +- packages/dsh-loopx-plugin/bench/README.md | 88 ++++++--- .../bench/observer-overhead.ts | 23 ++- .../dsh-loopx-plugin/bench/overhead-report.ts | 182 ++++++++++++++---- .../dsh-loopx-plugin/bench/overhead-runner.ts | 59 ++++-- packages/dsh-loopx-plugin/src/observer.ts | 7 +- .../tests/observer-overhead.spec.ts | 143 +++++++++++++- packages/dsh-loopx-plugin/tsconfig.tests.json | 1 + 9 files changed, 436 insertions(+), 103 deletions(-) diff --git a/loopx/capabilities/reliability_diagnostics/README.md b/loopx/capabilities/reliability_diagnostics/README.md index 6231e1a225..aad931329b 100644 --- a/loopx/capabilities/reliability_diagnostics/README.md +++ b/loopx/capabilities/reliability_diagnostics/README.md @@ -213,18 +213,27 @@ deletion. Upgrade CLI and plugin together with writers frozen. `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 latency, ledger bytes, event loss, -and clock uncertainty with full denominators. 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 +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. A trial that drops or +rejects events, fails an append, or persists fewer envelopes than it accepted is +excluded from the timing summary; the report is marked not usable and lists +warnings, so a lossy 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 --repeat 5 +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 diff --git a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md index 308cbc6671..50857ffb06 100644 --- a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md +++ b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md @@ -180,17 +180,24 @@ live 部署验收。操作 regular file 前会验证 provider 目录与 CLI 的 ## 测量 observer 开销 `packages/dsh-loopx-plugin/bench/observer-overhead.ts` 在确定性的 synthetic DSH 事件流上 -测量 observer 组件开销：observer 关闭/开启两臂、CPU（user+system）、RSS、ledger append -延迟、ledger 字节数、event loss 与 clock uncertainty，并给出完整分母。输出 public-safe -的 JSON + Markdown 报告，并声明限制；不测量真实 DSH host、模型、网络或 Cordis dispatch， -也不构成 C1 资格。见 [bench README](../../../packages/dsh-loopx-plugin/bench/README.md)。 +测量 observer 组件开销：observer 关闭/开启两臂、CPU（user+system）、RSS、经 shipped writer +的 ledger append 延迟、每条 envelope 的 ledger 字节与每次 flush 的 stats 字节、逐试验的 +event loss 与 buffer 峰值，以及 clock source 构成和完整分母。丢弃或拒绝事件、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 --repeat 5 +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 diff --git a/packages/dsh-loopx-plugin/bench/README.md b/packages/dsh-loopx-plugin/bench/README.md index c596bcfa5b..a22414b750 100644 --- a/packages/dsh-loopx-plugin/bench/README.md +++ b/packages/dsh-loopx-plugin/bench/README.md @@ -7,16 +7,26 @@ DSH host, model, or C1 qualification. ## What it measures -- **Two arms**: the same event stream with the observer disabled (baseline) and - enabled (observer). +- **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 and bytes written per persisted envelope. -- **Event accounting**: observed, accepted, persisted, dropped, rejected, peak - buffer, and flush attempts. Counts must satisfy +- **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`. -- **Integrity**: no outbound endpoints, no worker-context influence, and no - scheduler-input influence. +- **Clock**: the harness-time vs wall-clock envelope mix and the maximum + declared uncertainty. +- **Integrity**: no event loss, persisted envelopes equal accepted envelopes, no + outbound endpoints, no worker-context influence, and no scheduler-input + influence. + +A trial that drops or rejects events, fails an append, or persists fewer +envelopes than it accepted 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`. 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 @@ -28,33 +38,60 @@ From `packages/dsh-loopx-plugin`: ```bash node --experimental-strip-types bench/observer-overhead.ts \ - --events 10000 --repeat 5 --buffer-bound 256 \ + --events 10000 \ --json ../../.local/observer-overhead.json \ --markdown ../../.local/observer-overhead.md ``` -Defaults are 5,000 events, 3 repeats, and a buffer bound of 256. The tool runs a -short warmup, then alternates baseline and observer trials. Use `--json -` or -`--markdown -` to print a report to stdout. +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. -Example output from one Windows/Node 26 run (numbers are machine-specific and -are not a qualification): +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.0 | 609.0 | 609.0 | -| CPU, median (ms / 1k events) | 0.0 | 121.9 | 121.9 | -| RSS delta, median | 204.0 KiB | 8.2 MiB | 8.0 MiB | -| Ledger append p95 (ms) | n/a | 2.68 | n/a | -| Ledger bytes / persisted event | n/a | 521.6 | n/a | -| Persisted envelopes, median | n/a | 5,000 | n/a | +| CPU total, median (ms) | 0.00 | 1640.00 | 1640.00 | +| CPU, median (ms / 1k events) | 0.00 | 164.07 | 164.07 | +| RSS delta, median | 100.0 KiB | 12.2 MiB | 12.1 MiB | +| Ledger append p50 (ms) | n/a | 2.56 | n/a | +| Ledger append p95 (ms) | n/a | 3.84 | n/a | +| Ledger bytes / envelope | n/a | 490.38 | n/a | +| Ledger stats bytes / flush | n/a | 1326.31 | 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 | 251 | 78 | + +- No event loss: passed +- Persisted equals accepted: passed +- All trials usable: passed ``` -The synthetic turn cycle has six events per turn and flushes on every turn -end, so it is deliberately flush-heavy. A real long turn flushes less often; -use the per-event numbers as a bound/signal, not a workload forecast. - +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 @@ -73,5 +110,6 @@ and the [RFC](../../../docs/architecture/rfcs/long-running-agent-reliability-dia pnpm test -- observer-overhead ``` -The test covers the event-plan shape, percentile boundaries, count -conservation, append-failure accounting, and public-safe report rendering. +The test covers the event-plan shape, percentile boundaries, the shipped-writer +append path and byte split, loss and append-failure 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 index cc232785fa..cf2ed1a112 100644 --- a/packages/dsh-loopx-plugin/bench/observer-overhead.ts +++ b/packages/dsh-loopx-plugin/bench/observer-overhead.ts @@ -6,7 +6,7 @@ import { runBaselineTrial, runObserverTrial, normalizeEventCount } from './overh import type { TrialMeasurement } from './overhead-runner.ts' import { buildOverheadReport, renderMarkdown } from './overhead-report.ts' -interface CliOptions { +export interface CliOptions { readonly events: number readonly repeat: number readonly bufferBound: number @@ -23,10 +23,10 @@ function positiveInteger(value: string | undefined, flag: string): number { return parsed } -function parseArgs(args: readonly string[]): CliOptions { +export function parseArgs(args: readonly string[]): CliOptions { let events = 5_000 let repeat = 3 - let bufferBound = 256 + let bufferBound = 4_096 let json: string | undefined let markdown: string | undefined let help = false @@ -70,9 +70,11 @@ function printUsage(): void { process.stdout.write(`Usage: node --experimental-strip-types bench/observer-overhead.ts [options] Options: - --events synthetic session events per trial (default: 5000) + --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: 256) + --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 @@ -86,6 +88,11 @@ async function main(): Promise { 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 }) @@ -100,6 +107,7 @@ async function main(): Promise { const report = buildOverheadReport({ config: { events, + requested_events: options.events, repeat: options.repeat, buffer_bound: options.bufferBound, event_mix: 'turn-cycle-v1', @@ -117,6 +125,11 @@ async function main(): Promise { 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) diff --git a/packages/dsh-loopx-plugin/bench/overhead-report.ts b/packages/dsh-loopx-plugin/bench/overhead-report.ts index eeab69eeaa..259bf1a11b 100644 --- a/packages/dsh-loopx-plugin/bench/overhead-report.ts +++ b/packages/dsh-loopx-plugin/bench/overhead-report.ts @@ -2,11 +2,14 @@ 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 @@ -22,6 +25,7 @@ export interface OverheadReportEnvironment { export interface OverheadReportConfig { readonly events: number + readonly requested_events?: number readonly repeat: number readonly buffer_bound: number readonly event_mix: string @@ -29,6 +33,9 @@ export interface OverheadReportConfig { export interface OverheadIntegrity { readonly count_conservation: boolean + readonly no_event_loss: boolean + readonly persisted_equals_accepted: boolean + readonly all_trials_usable: boolean readonly outbound_endpoints_empty: boolean readonly worker_context_influence_absent: boolean readonly scheduler_input_influence_absent: boolean @@ -41,20 +48,29 @@ export interface OverheadEventAccounting { 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 } export interface OverheadDelta { readonly cpu_ms_per_1k_events: number readonly rss_delta_bytes: number - readonly append_p95_ms: 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: { @@ -77,10 +93,11 @@ export interface OverheadReportInput { 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); excludes Cordis hook dispatch, DSH scheduling, model, and network.', - 'The synthetic turn cycle has six events per turn and flushes on every turn end; a real long turn flushes less often, so the per-event append cost here is a flush-heavy microbenchmark, not a workload forecast.', + '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.', + '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, or loses persisted envelopes 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 @@ -96,45 +113,92 @@ 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) +} + +/** + * A trial whose timings can be compared with another trial: it lost no events, + * rejected none, failed no append, and persisted every accepted envelope. + */ +export function isUsableTrial(trial: TrialMeasurement): boolean { + return trial.droppedEvents === 0 + && trial.rejectedEvents === 0 + && trial.observerFailures === 0 + && trial.persistedEvents === trial.acceptedEvents +} + export function summarizeTrials(trials: readonly TrialMeasurement[]): ArmSummary { - if (trials.length === 0) { - return { - trials: 0, - cpu_total_ms_median: 0, - cpu_total_ms_p95: 0, - cpu_ms_per_1k_events_median: 0, - rss_delta_bytes_median: 0, - ledger_bytes_per_event: 0, - append_p50_ms: 0, - append_p95_ms: 0, - persisted_events_median: 0, - } - } - const cpuPer1k = trials.map(trial => trial.workloadEvents === 0 + const usable = trials.filter(isUsableTrial) + const cpuPer1k = usable.map(trial => trial.workloadEvents === 0 ? 0 : (trial.cpuTotalMs / trial.workloadEvents) * 1_000) - const appendLatencies = trials.flatMap(trial => trial.appendLatenciesMs) - const totalLedgerBytes = trials.reduce((sum, trial) => sum + trial.ledgerBytes, 0) - const totalPersisted = trials.reduce((sum, trial) => sum + trial.persistedEvents, 0) + 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, - cpu_total_ms_median: median(trials.map(trial => trial.cpuTotalMs)), - cpu_total_ms_p95: percentile(trials.map(trial => trial.cpuTotalMs), 0.95), + 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(trials.map(trial => trial.rssDeltaBytes)), - ledger_bytes_per_event: totalPersisted === 0 ? 0 : totalLedgerBytes / totalPersisted, + 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(trials.map(trial => trial.persistedEvents)), + 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.', + ) + } + 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: { @@ -146,11 +210,13 @@ export function buildOverheadReport(input: OverheadReportInput): OverheadReport - baselineSummary.cpu_ms_per_1k_events_median, rss_delta_bytes: observerSummary.rss_delta_bytes_median - baselineSummary.rss_delta_bytes_median, - append_p95_ms: observerSummary.append_p95_ms - baselineSummary.append_p95_ms, }, 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, 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), @@ -161,10 +227,18 @@ export function buildOverheadReport(input: OverheadReportInput): OverheadReport 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(input.observer.map(trial => trial.droppedEvents)), + 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, }, limits: REPORT_LIMITS, } @@ -193,44 +267,72 @@ function status(value: boolean): string { 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: ${formatInteger(report.config.events)} +- 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) | ${formatNumber(baseline.cpu_total_ms_median)} | ${formatNumber(observer.cpu_total_ms_median)} | ${formatNumber(observer.cpu_total_ms_median - baseline.cpu_total_ms_median)} | -| CPU, median (ms / 1k events) | ${formatNumber(baseline.cpu_ms_per_1k_events_median)} | ${formatNumber(observer.cpu_ms_per_1k_events_median)} | ${formatNumber(report.delta.cpu_ms_per_1k_events)} | -| RSS delta, median | ${formatBytes(baseline.rss_delta_bytes_median)} | ${formatBytes(observer.rss_delta_bytes_median)} | ${formatBytes(report.delta.rss_delta_bytes)} | -| Ledger append p50 (ms) | n/a | ${formatNumber(observer.append_p50_ms)} | n/a | -| Ledger append p95 (ms) | n/a | ${formatNumber(observer.append_p95_ms)} | n/a | -| Ledger bytes / persisted event | n/a | ${formatNumber(observer.ledger_bytes_per_event)} | n/a | -| Persisted envelopes, median | n/a | ${formatInteger(observer.persisted_events_median)} | n/a | +| 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, median) +## 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)} | -| Observed | Accepted | Persisted | Dropped | Rejected | Flush attempts | -| ---: | ---: | ---: | ---: | ---: | ---: | -| ${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_median)} | ${formatInteger(report.event_accounting.rejected_events_median)} | ${formatInteger(report.event_accounting.flush_attempts_median)} | +- 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) ## 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)} - 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 index 2faa091b27..d535f929d8 100644 --- a/packages/dsh-loopx-plugin/bench/overhead-runner.ts +++ b/packages/dsh-loopx-plugin/bench/overhead-runner.ts @@ -1,9 +1,9 @@ -import { appendFile, mkdir, mkdtemp, readFile, rm } from 'node:fs/promises' +import { mkdtemp, readFile, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { dirname, join } from 'node:path' +import { join } from 'node:path' import { performance } from 'node:perf_hooks' import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' -import { ShadowObserver } from '../src/observer.ts' +import { appendLedgerLines, ShadowObserver } from '../src/observer.ts' import type { ObserverRunIdentity, ShadowObserverConfig } from '../src/observer.ts' const GOAL_ID = 'goal-observer-overhead-fixture' @@ -54,7 +54,11 @@ export interface TrialMeasurement { 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 appendLatenciesMs: readonly number[] readonly outboundEndpoints: number readonly workerContextInfluence: boolean @@ -179,7 +183,11 @@ export function runBaselineTrial(options: TrialOptions): TrialMeasurement { peakBufferedEvents: 0, observerFailures: 0, ledgerBytes: 0, + envelopeBytes: 0, + statsBytes: 0, clockUncertaintyMaxMs: 0, + harnessTimeEvents: 0, + wallClockEvents: 0, appendLatenciesMs: [], outboundEndpoints: 0, workerContextInfluence: false, @@ -191,7 +199,6 @@ export async function runObserverTrial(options: TrialOptions): Promise EVENT_TIME_BASE, observerId: OBSERVER_ID, - appendLines: async (path, lines) => { + // 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() - const text = `${lines.join('\n')}\n` - await mkdir(dirname(path), { recursive: true }) - await appendFile(path, text, 'utf8') - ledgerBytes += Buffer.byteLength(text, 'utf8') + await appendLedgerLines(config, lines) appendLatenciesMs.push(performance.now() - start) }, }) @@ -227,8 +234,9 @@ export async function runObserverTrial(options: TrialOptions): 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 index 4055632a9b..6e9fd933e8 100644 --- a/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts +++ b/packages/dsh-loopx-plugin/tests/observer-overhead.spec.ts @@ -8,10 +8,12 @@ import { 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', @@ -48,20 +50,36 @@ describe('observer overhead event plan', () => { }) describe('observer overhead trials', () => { - it('measures observer component cost with full denominators', async () => { - const trial = await runObserverTrial({ inputEvents: 60, bufferBound: 8 }) + 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.observedEvents).toBe( trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, ) + expect(trial.persistedEvents).toBe(trial.acceptedEvents) expect(trial.persistedEvents).toBeGreaterThan(0) - expect(trial.ledgerBytes).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 () => { @@ -76,35 +94,144 @@ describe('observer overhead trials', () => { trial.acceptedEvents + trial.rejectedEvents + trial.droppedEvents, ) expect(trial.outboundEndpoints).toBe(0) + expect(isUsableTrial(trial)).toBe(false) }) - it('renders a public-safe report with both arms and integrity status', async () => { - const baseline = [runBaselineTrial({ inputEvents: 12, bufferBound: 8 })] - const observer = [await runObserverTrial({ inputEvents: 12, bufferBound: 8 })] + 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, repeat: 1, buffer_bound: 8, event_mix: 'turn-cycle-v1' }, + 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.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('Count conservation: passed') + 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**') + }) +}) + +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" ] From 2f87039a527b0f769195b40a0679c2a687b6edcf Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Mon, 19 Oct 2026 11:56:26 +0800 Subject: [PATCH 3/3] fix(reliability): fail the overhead bench when the planned record mix is not consumed The integrity checks only covered backpressure drops, rejects, append failures and persisted-vs-accepted counts, so a planned event kind could be retired (or substituted by an unknown kind) and the report would still read as usable with no warnings. Mutation testing confirmed that a retired turn/start kind silently shrank the observed stream while the report stayed clean. Each observer trial now carries the planned envelope-kind histogram, the runner records the persisted kind histogram from the ledger readback, and a trial is usable only when it observed every planned record with the planned kind mix. The report exposes observed_matches_plan / envelope_kind_mix_complete integrity rows, unconsumed and unsupported envelope totals, and targeted warnings. Docs and the regenerated bench example output describe the new exclusion. Signed-off-by: JasonBuildAI --- .../reliability_diagnostics/README.md | 10 +-- .../reliability_diagnostics/README.zh-CN.md | 7 +- packages/dsh-loopx-plugin/bench/README.md | 39 +++++----- .../dsh-loopx-plugin/bench/overhead-report.ts | 72 ++++++++++++++++++- .../dsh-loopx-plugin/bench/overhead-runner.ts | 37 +++++++++- .../tests/observer-overhead.spec.ts | 53 ++++++++++++++ 6 files changed, 191 insertions(+), 27 deletions(-) diff --git a/loopx/capabilities/reliability_diagnostics/README.md b/loopx/capabilities/reliability_diagnostics/README.md index aad931329b..d3b8560af4 100644 --- a/loopx/capabilities/reliability_diagnostics/README.md +++ b/loopx/capabilities/reliability_diagnostics/README.md @@ -215,10 +215,12 @@ deletion. Upgrade CLI and plugin together with writers frozen. 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. A trial that drops or -rejects events, fails an append, or persists fewer envelopes than it accepted is -excluded from the timing summary; the report is marked not usable and lists -warnings, so a lossy run cannot read as a clean result. It writes a public-safe +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). diff --git a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md index 50857ffb06..8dc69a8221 100644 --- a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md +++ b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md @@ -182,9 +182,10 @@ live 部署验收。操作 regular file 前会验证 provider 目录与 CLI 的 `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 构成和完整分母。丢弃或拒绝事件、append 失败、 -或 persisted 少于 accepted 的试验会从计时汇总中排除；报告标记为不可用并列出警告，因此有丢失 -的运行不会被读成干净结果。输出 public-safe 的 JSON + Markdown 报告，并声明限制；不测量真实 +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)。 diff --git a/packages/dsh-loopx-plugin/bench/README.md b/packages/dsh-loopx-plugin/bench/README.md index a22414b750..1d8076d5bf 100644 --- a/packages/dsh-loopx-plugin/bench/README.md +++ b/packages/dsh-loopx-plugin/bench/README.md @@ -18,15 +18,16 @@ DSH host, model, or C1 qualification. `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, no - outbound endpoints, no worker-context influence, and no scheduler-input - influence. +- **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, or persists fewer -envelopes than it accepted 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`. +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 @@ -67,24 +68,28 @@ machine-specific and are not a qualification): | Metric | Baseline | Observer | Delta | | --- | ---: | ---: | ---: | -| CPU total, median (ms) | 0.00 | 1640.00 | 1640.00 | -| CPU, median (ms / 1k events) | 0.00 | 164.07 | 164.07 | -| RSS delta, median | 100.0 KiB | 12.2 MiB | 12.1 MiB | -| Ledger append p50 (ms) | n/a | 2.56 | n/a | -| Ledger append p95 (ms) | n/a | 3.84 | n/a | +| 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.31 | 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 | 251 | 78 | +| 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 @@ -111,5 +116,5 @@ pnpm test -- observer-overhead ``` The test covers the event-plan shape, percentile boundaries, the shipped-writer -append path and byte split, loss and append-failure marking, CLI parsing, and -public-safe report rendering. +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/overhead-report.ts b/packages/dsh-loopx-plugin/bench/overhead-report.ts index 259bf1a11b..ad0509c4f0 100644 --- a/packages/dsh-loopx-plugin/bench/overhead-report.ts +++ b/packages/dsh-loopx-plugin/bench/overhead-report.ts @@ -36,6 +36,8 @@ export interface OverheadIntegrity { 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 @@ -59,6 +61,8 @@ export interface OverheadEventAccounting { 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 { @@ -94,10 +98,11 @@ export interface OverheadReportInput { 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, or loses persisted envelopes is excluded from the arm summaries; the report is marked not usable and prints warnings.', + '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 @@ -117,15 +122,43 @@ 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, and persisted every accepted envelope. + * 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 { @@ -186,6 +219,29 @@ function buildWarnings(observer: readonly TrialMeasurement[]): readonly string[] + '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 } @@ -217,6 +273,8 @@ export function buildOverheadReport(input: OverheadReportInput): OverheadReport 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), @@ -239,6 +297,12 @@ export function buildOverheadReport(input: OverheadReportInput): OverheadReport 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, } @@ -320,12 +384,16 @@ export function renderMarkdown(report: OverheadReport): string { - 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)} diff --git a/packages/dsh-loopx-plugin/bench/overhead-runner.ts b/packages/dsh-loopx-plugin/bench/overhead-runner.ts index d535f929d8..9fde55d0de 100644 --- a/packages/dsh-loopx-plugin/bench/overhead-runner.ts +++ b/packages/dsh-loopx-plugin/bench/overhead-runner.ts @@ -39,6 +39,7 @@ export interface TrialOptions { export interface TrialMeasurement { readonly workloadEvents: number readonly observedEvents: number + readonly expectedObservedEvents: number readonly acceptedEvents: number readonly persistedEvents: number readonly droppedEvents: number @@ -59,6 +60,8 @@ export interface TrialMeasurement { 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 @@ -69,6 +72,26 @@ 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, @@ -168,6 +191,7 @@ export function runBaselineTrial(options: TrialOptions): TrialMeasurement { return { workloadEvents, observedEvents: plan.length, + expectedObservedEvents: plan.length, cpuUserMs: cpu.userMs, cpuSystemMs: cpu.systemMs, cpuTotalMs: cpu.totalMs, @@ -188,6 +212,9 @@ export function runBaselineTrial(options: TrialOptions): TrialMeasurement { clockUncertaintyMaxMs: 0, harnessTimeEvents: 0, wallClockEvents: 0, + // The baseline driver runs no observer and persists no envelopes. + envelopeKinds: {}, + expectedEnvelopeKinds: {}, appendLatenciesMs: [], outboundEndpoints: 0, workerContextInfluence: false, @@ -254,6 +281,7 @@ export async function runObserverTrial(options: TrialOptions): Promise = {} try { const text = await readFile(observer.path, 'utf8') for (const line of text.split('\n')) { @@ -267,7 +295,8 @@ export async function runObserverTrial(options: TrialOptions): Promise { 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, ) @@ -123,6 +134,8 @@ describe('observer overhead trials', () => { 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) @@ -204,6 +217,46 @@ describe('observer overhead trials', () => { 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', () => {