From b390bd66e1508bc4f88a3d04bb2cbcd57cc271a5 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 20 Aug 2026 17:48:27 -0700 Subject: [PATCH] feat(statistics): reconstruct a sequential e-process after a restart --- CHANGELOG.md | 6 +- clients/python/pyproject.toml | 2 +- clients/python/src/agent_eval_rpc/__init__.py | 2 +- clients/python/uv.lock | 2 +- docs/design/statistics-decisions.md | 2 +- package.json | 2 +- src/analyst/benchmark-implementation.ts | 4 +- src/campaign/gates/sequential.test.ts | 202 ++++++++++++++++- src/campaign/gates/sequential.ts | 80 ++++++- src/campaign/index.ts | 1 + src/statistics/sequential-eprocess.test.ts | 214 ++++++++++++++++++ src/statistics/sequential-eprocess.ts | 97 +++++++- 12 files changed, 597 insertions(+), 17 deletions(-) create mode 100644 src/statistics/sequential-eprocess.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f0e45b87..6c795c14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,11 @@ All notable changes to `@tangle-network/agent-eval` and its sibling `agent-eval- --- -## [Unreleased] +## [0.154.0] — 2026-08-21 + +### Added + +- Sequential state reconstruction after a process restart (#411, criterion 2). `EProcessState` now carries the running sums `sumX` and `varSum` next to wealth, n, and the parameters, so the snapshot `state()` returns is sufficient to continue the betting test-martingale. `eProcess({ resume })` rebuilds a process from such a snapshot, and `sequentialPairedGate({ resume })` rebuilds the gate's observe-stream from the new exported `SequentialStreamState` (the e-process state plus the gate decision). A process or gate interrupted at any n and resumed from a JSON round-trip of its state produces the identical wealth sequence, observation sequence, decision, and final state as an uninterrupted run. The snapshot never supplies the parameters: a snapshot recorded under a different alpha, maxBet, null boundary, or threshold is refused with a `ValidationError`, as is one whose fields cannot all be true at once (running sums out of range, a latch without its n, a gate decision this configuration could not have reached). `sequentialDecide` needs no snapshot: it is replayed from the generation history it is handed. --- diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml index 4a6be63d..8a180b68 100644 --- a/clients/python/pyproject.toml +++ b/clients/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "agent-eval-rpc" -version = "0.153.0" +version = "0.154.0" description = "Python RPC client, official optimizer bridge, and DSPy metric adapter for @tangle-network/agent-eval." readme = "README.md" requires-python = ">=3.10" diff --git a/clients/python/src/agent_eval_rpc/__init__.py b/clients/python/src/agent_eval_rpc/__init__.py index 0924e790..bbc7bc5e 100644 --- a/clients/python/src/agent_eval_rpc/__init__.py +++ b/clients/python/src/agent_eval_rpc/__init__.py @@ -53,7 +53,7 @@ try: __version__ = version("agent-eval-rpc") except PackageNotFoundError: - __version__ = "0.153.0" + __version__ = "0.154.0" __all__ = [ "Client", diff --git a/clients/python/uv.lock b/clients/python/uv.lock index 6c87e03c..a20f9dc4 100644 --- a/clients/python/uv.lock +++ b/clients/python/uv.lock @@ -34,7 +34,7 @@ conflicts = [[ [[package]] name = "agent-eval-rpc" -version = "0.153.0" +version = "0.154.0" source = { editable = "." } dependencies = [ { name = "filelock" }, diff --git a/docs/design/statistics-decisions.md b/docs/design/statistics-decisions.md index 3e00924d..af463338 100644 --- a/docs/design/statistics-decisions.md +++ b/docs/design/statistics-decisions.md @@ -90,7 +90,7 @@ These 17 matched their references with zero mismatches and stay as they are, pin | `wilson` | keep-with-CI-oracle | Matched the closed form to `< 1e-8` on all 13 `(successes, n)` pairs including `0/1`, `10/10`, `1/1000`, `0/0`. Correctly asymmetric at the boundaries: `wilson(0, 10)` returns `[0, 0.27753280]`. | | `passAtK` | keep-with-CI-oracle | Chen et al. 2021 unbiased estimator, exhaustively verified for every `(n, c, k)` with `n = 1..6` against exact integer `math.comb`: zero mismatches. Stable at scale (`n=1000, c=3, k=100` agreed to `1.11e-16`). `passAtK(10, 3, 5) = 0.9166666667`. | | `corpusInterRaterAgreement` | keep-with-CI-oracle | The ICC(2,1) it surfaces matched a hand-derived two-way random-effects ANOVA reference to `< 1e-9` on 5 matrices, including the inverted case at `−1.959459`. This is a genuinely different and correct computation from `interRaterReliability`: it pivots to a proper items × judges matrix and delegates to `continuousAgreement`. Its fail-loud contract behaves — empty input, fewer than two judges, fewer than two common items, duplicate records, and absent dimensions all throw `ValidationError`. | -| `eProcess` | keep-with-CI-oracle | Betting test-martingale. Empirical type-I over 4,000 sequences of 200 observations at `α = 0.05`: 2.40 % at the null boundary, 0.00 % in the null interior, 1.33 % on continuous uniform — all inside Ville's bound. Power at `E[x] = 0.7` is 99.775 %. The predictability invariant holds: the first update leaves wealth at exactly 1. | +| `eProcess` | keep-with-CI-oracle | Betting test-martingale. Empirical type-I over 4,000 sequences of 200 observations at `α = 0.05`: 2.40 % at the null boundary, 0.00 % in the null interior, 1.33 % on continuous uniform — all inside Ville's bound. Power at `E[x] = 0.7` is 99.775 %. The predictability invariant holds: the first update leaves wealth at exactly 1. Restart reconstruction: `state()` carries the running sums (`sumX`, `varSum`) next to wealth and n, and `eProcess({ resume })` rebuilds the process from that snapshot; a process interrupted at any n and resumed from a JSON round-trip of its state reproduces the uninterrupted wealth sequence and decision exactly (`sequential-eprocess.test.ts`, interruptions at n = 0, 1, 7, 30, 119 and at the crossing), and `sequentialPairedGate({ resume })` does the same for the gate's observe-stream. A snapshot recorded under other parameters, or one whose fields cannot all be true at once, is refused. | | `holm` | keep-with-CI-oracle | Matched `statsmodels.multipletests` to `1e-9` on a 6-value reference vector, and uses `≤` at the boundary, which is the correct rule. It also validates both `alpha` and the p range, which `bonferroni` does not. | | `zQuantile` | keep-with-CI-oracle | Acklam inverse-normal, structurally independent of `normalCdf`. This independence is why the sample-size functions were untouched by the CDF defect, and why `mcnemarPower`'s failure to invert `mcnemarRequiredN` was a valid detector of it. | | `mcnemarRequiredN` | keep-with-CI-oracle | Matched the Lachin closed form exactly on all 4 parameter sets: `234`, `77`, `155`, `Infinity`. Unchanged by the CDF fix. Round-trip against the repaired `mcnemarPower` now holds across 16 configurations: power at `requiredN` meets the target with overshoot `≤ 0.0049`, and power at `requiredN − 1` is below target in all 16. | diff --git a/package.json b/package.json index 1a4632a9..de144870 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/agent-eval", - "version": "0.153.0", + "version": "0.154.0", "description": "Evaluate and improve AI agents from runs, traces, judges, and feedback. Compare candidates, cluster failures, measure lift, and gate releases.", "homepage": "https://github.com/tangle-network/agent-eval#readme", "repository": { diff --git a/src/analyst/benchmark-implementation.ts b/src/analyst/benchmark-implementation.ts index 2ec95b91..ee470dff 100644 --- a/src/analyst/benchmark-implementation.ts +++ b/src/analyst/benchmark-implementation.ts @@ -10,7 +10,7 @@ export const ANALYST_BENCHMARK_DEPENDENCY_LOCK_FILES = Object.freeze([ ]) export const ANALYST_BENCHMARK_DEPENDENCY_LOCK_SHA256 = - 'eba5216907300465abf4fdb084c04156d4c247a122645a2240690c5349500aff' + 'c306327eaf924a0f51ede71f5cd5e4efc17fc1dc4920d80e03a71a6c98a6e87d' /** The published benchmark evidence was produced at this package version, by * the retired one-shot direct runner, before trace analysts moved to the @@ -137,7 +137,7 @@ export const ANALYST_BENCHMARK_IMPLEMENTATION_FILES = Object.freeze([ ]) export const ANALYST_BENCHMARK_IMPLEMENTATION_SHA256 = - 'a42b96f7d7329cc5b1f4f7ee81d387dd60664f6719a03686f1a74b2c315e08e9' + '3bae068f5efdeb37ea3af721b0771848a28148e1a2b21f2d22afd3292ffa1041' export function analystBenchmarkImplementationDigest() { return ANALYST_BENCHMARK_IMPLEMENTATION_SHA256 diff --git a/src/campaign/gates/sequential.test.ts b/src/campaign/gates/sequential.test.ts index e8289bf5..5a33260d 100644 --- a/src/campaign/gates/sequential.test.ts +++ b/src/campaign/gates/sequential.test.ts @@ -2,7 +2,13 @@ import { describe, expect, it } from 'vitest' import { type HypothesisManifest, signManifest } from '../../pre-registration' import { eProcess, mulberry32 } from '../../statistics' import type { GateContext, GenerationRecord, JudgeScore, Scenario } from '../types' -import { sequentialDecide, sequentialPairedGate } from './sequential' +import { + type SequentialObservation, + type SequentialPairedGateOptions, + type SequentialStreamState, + sequentialDecide, + sequentialPairedGate, +} from './sequential' // ── Fixtures ────────────────────────────────────────────────────────── @@ -329,6 +335,200 @@ describe('sequentialPairedGate.decide — gate contract', () => { }) }) +// ── sequentialPairedGate — restart reconstruction ───────────────────── + +describe('sequentialPairedGate.resume — reconstruction after a process restart', () => { + const opts = { alpha: 0.05, minN: 5, maxN: 80, maxBet: 0.5, scale: 1 } + + /** Paired deltas with a real edge (≈ +0.3 on scale 1) so the reference + * promotes inside the budget. */ + function deltas(seed: number, length: number): number[] { + const rng = mulberry32(seed) + return Array.from({ length }, () => Math.min(1, Math.max(-1, 0.3 + (rng() - 0.5) * 1.2))) + } + + function observeAll( + ds: number[], + o: SequentialPairedGateOptions = opts, + ): SequentialObservation[] { + const gate = sequentialPairedGate(o) + const out: SequentialObservation[] = [] + for (const d of ds) out.push(gate.observe(d)) + return out + } + + /** Observe the first k deltas, persist `state()` through JSON, rebuild + * the gate with `resume`, and observe the rest. */ + function observeWithRestart( + ds: number[], + k: number, + o: SequentialPairedGateOptions = opts, + ): { observations: SequentialObservation[]; final: SequentialStreamState } { + const first = sequentialPairedGate(o) + const observations: SequentialObservation[] = [] + for (const d of ds.slice(0, k)) observations.push(first.observe(d)) + const persisted = JSON.parse(JSON.stringify(first.state())) as SequentialStreamState + const second = sequentialPairedGate({ ...o, resume: persisted }) + for (const d of ds.slice(k)) observations.push(second.observe(d)) + return { observations, final: second.state() } + } + + const ds = deltas(99, 60) + const reference = observeAll(ds) + const promoteAt = reference.findIndex((o) => o.decision === 'promote') + 1 + + it('the reference stream promotes inside the budget (fixture has an edge)', () => { + expect(promoteAt).toBeGreaterThan(opts.minN) + expect(promoteAt).toBeLessThan(ds.length) + }) + + it.each([0, 1, 4, 5, 20, 59])( + 'interrupting at n=%i and resuming from state() yields the identical observation sequence and final state', + (k) => { + const resumed = observeWithRestart(ds, k) + expect(resumed.observations).toEqual(reference) + const uninterrupted = sequentialPairedGate(opts) + for (const d of ds) uninterrupted.observe(d) + expect(resumed.final).toEqual(uninterrupted.state()) + }, + ) + + it('interrupting around the promote crossing keeps the sticky decision and its n', () => { + for (const k of [promoteAt - 1, promoteAt, promoteAt + 1]) { + const resumed = observeWithRestart(ds, k) + expect(resumed.observations).toEqual(reference) + expect(resumed.final.decision).toBe('promote') + } + }) + + it('a stream resumed after promote keeps the sticky decision against contrary evidence', () => { + const first = sequentialPairedGate(opts) + for (const d of ds.slice(0, promoteAt + 3)) first.observe(d) + const second = sequentialPairedGate({ ...opts, resume: first.state() }) + const next = second.observe(-0.9) + expect(next.decision).toBe('promote') + expect(next.n).toBe(promoteAt + 4) + }) + + it('a stream resumed at undecided-at-maxN refuses further observations exactly like the original', () => { + const flat = Array.from({ length: 6 }, () => 0) + const small = { alpha: 0.05, minN: 5, maxN: 6 } + const first = sequentialPairedGate(small) + for (const d of flat) first.observe(d) + expect(first.state().decision).toBe('undecided-at-maxN') + const second = sequentialPairedGate({ ...small, resume: first.state() }) + expect(second.state()).toEqual(first.state()) + expect(() => second.observe(0)).toThrow(/optional stopping/) + }) + + it('a resumed gate reaches undecided-at-maxN at the same n as the uninterrupted gate', () => { + const noise = Array.from({ length: 30 }, (_, i) => (i % 2 === 0 ? 0.05 : -0.05)) + const small = { alpha: 0.05, minN: 5, maxN: 30 } + const ref = observeAll(noise, small) + expect(ref[ref.length - 1]!.decision).toBe('undecided-at-maxN') + const resumed = observeWithRestart(noise, 17, small) + expect(resumed.observations).toEqual(ref) + }) + + it('resume under a bound manifest continues the registered statistic', async () => { + const signed = await signManifest(manifestBase()) + const bound = { preRegistration: signed, minN: 5 } + const ref = observeAll(ds.slice(0, 30), bound) + const first = sequentialPairedGate(bound) + for (const d of ds.slice(0, 9)) first.observe(d) + const second = sequentialPairedGate({ ...bound, resume: first.state() }) + const tail = ds.slice(9, 30).map((d) => second.observe(d)) + expect([...ds.slice(0, 9).map((_, i) => ref[i]!), ...tail]).toEqual(ref) + }) + + it('decide(ctx) is unaffected by resume — it always runs its own fresh stream', async () => { + const first = sequentialPairedGate(opts) + for (const d of ds.slice(0, 10)) first.observe(d) + const resumed = sequentialPairedGate({ ...opts, resume: first.state() }) + const fresh = sequentialPairedGate(opts) + const ctx = ctxFrom( + Array.from({ length: 12 }, (_, i) => ({ + scenarioId: `s${i}`, + reps: 2, + candidate: 0.8, + baseline: 0.5, + })), + ) + const [a, b] = await Promise.all([resumed.decide(ctx), fresh.decide(ctx)]) + expect(a).toEqual(b) + expect(resumed.state().n).toBe(10) + }) + + it('refuses a snapshot recorded under different e-process parameters (alpha / minEffect / maxBet)', async () => { + const first = sequentialPairedGate(opts) + for (const d of ds.slice(0, 10)) first.observe(d) + const snap = first.state() + expect(() => sequentialPairedGate({ ...opts, alpha: 0.01, resume: snap })).toThrow( + /cannot resume — snapshot alpha=0\.05 differs from the process alpha=0\.01/, + ) + expect(() => sequentialPairedGate({ ...opts, maxBet: 0.3, resume: snap })).toThrow( + /snapshot maxBet=0\.5 differs/, + ) + // A manifest with minEffect shifts the null boundary: a snapshot taken at + // minEffect 0 does not continue a minEffect 0.2 statistic. + const shifted = await signManifest({ ...manifestBase(), minEffect: 0.2 }) + expect(() => sequentialPairedGate({ preRegistration: shifted, minN: 5, resume: snap })).toThrow( + /snapshot nullMean=0\.5 differs from the process nullMean=0\.6/, + ) + expect(() => sequentialPairedGate({ ...opts, resume: { ...snap, threshold: 7 } })).toThrow( + /snapshot threshold=7 differs/, + ) + }) + + it('refuses a snapshot whose gate decision this configuration could not have reached', () => { + const first = sequentialPairedGate(opts) + for (const d of ds.slice(0, 10)) first.observe(d) + const snap = first.state() + expect(snap.decision).toBe('continue') + expect(() => sequentialPairedGate({ ...opts, maxN: 9, resume: snap })).toThrow( + /snapshot n=10 exceeds the pre-registered maxN=9/, + ) + expect(() => sequentialPairedGate({ ...opts, maxN: 10, resume: snap })).toThrow( + /decision 'continue' at n=10 with maxN=10; the stream is finished/, + ) + expect(() => + sequentialPairedGate({ + ...opts, + resume: { ...snap, decision: 'promote', n: 3, sumX: 1, varSum: 0.1 }, + }), + ).toThrow(/decision 'promote' at n=3 is below minN=5/) + expect(() => + sequentialPairedGate({ ...opts, resume: { ...snap, decision: 'undecided-at-maxN' } }), + ).toThrow(/decision 'undecided-at-maxN' at n=10 does not match maxN=80/) + expect(() => + sequentialPairedGate({ + ...opts, + resume: { ...snap, decision: 'continue', wealth: 25, decided: true, decidedAtN: 9 }, + }), + ).toThrow( + /decision 'continue' at n=10 ≥ minN=5 with e-value 25 ≥ 1\/α=20; this stream would have promoted/, + ) + expect(() => + sequentialPairedGate({ + ...opts, + resume: { ...snap, decision: 'later' as unknown as SequentialStreamState['decision'] }, + }), + ).toThrow(/unknown decision 'later'/) + }) + + it('refuses a tampered e-process snapshot through the core validator', () => { + const first = sequentialPairedGate(opts) + for (const d of ds.slice(0, 10)) first.observe(d) + const snap = first.state() + expect(() => sequentialPairedGate({ ...opts, resume: { ...snap, sumX: 11 } })).toThrow( + /eProcess: cannot resume — sumX must lie in \[0, n=10\]/, + ) + expect(() => sequentialPairedGate({ ...opts, resume: { ...snap, wealth: -2 } })).toThrow( + /wealth must be a finite positive number/, + ) + }) +}) + // ── Pre-registration binding ────────────────────────────────────────── describe('sequentialPairedGate — pre-registration binding', () => { diff --git a/src/campaign/gates/sequential.ts b/src/campaign/gates/sequential.ts index 3871e1a1..493626a6 100644 --- a/src/campaign/gates/sequential.ts +++ b/src/campaign/gates/sequential.ts @@ -75,8 +75,21 @@ export interface SequentialPairedGateOptions { preRegistration?: SignedManifest /** Override the gate name in reports. */ name?: string + /** Continue the observe-stream from a `state()` snapshot taken before a + * restart. The gate's parameters still resolve from this options object + * (and the manifest); the snapshot must have been recorded under the same + * alpha, maxBet, and null boundary, and its decision must be one this + * configuration could have reached at its n, or construction throws. + * `decide(ctx)` is unaffected — it always runs on a fresh stream. */ + resume?: SequentialStreamState } +/** Snapshot of one observe-stream: the e-process state plus the gate's own + * decision, which applies `minN` and `maxN` on top of the core latch. A + * gate rebuilt with `resume` from this value continues exactly where the + * snapshot was taken. */ +export type SequentialStreamState = EProcessState & { decision: SequentialDecision } + export interface SequentialPairedGate extends Gate { /** Streaming entry point: feed one paired per-scenario delta @@ -87,8 +100,9 @@ export interface SequentialPairedGate { + throw new Error(`sequentialPairedGate: cannot resume — ${detail}`) + } + if (!SEQUENTIAL_DECISIONS.includes(resume.decision)) { + fail(`unknown decision '${String(resume.decision)}'`) + } + if (resume.n > cfg.maxN) { + fail(`snapshot n=${resume.n} exceeds the pre-registered maxN=${cfg.maxN}`) + } + const threshold = 1 / cfg.alpha + switch (resume.decision) { + case 'promote': + if (resume.n < cfg.minN) { + fail(`decision 'promote' at n=${resume.n} is below minN=${cfg.minN}`) + } + break + case 'undecided-at-maxN': + if (resume.n !== cfg.maxN) { + fail(`decision 'undecided-at-maxN' at n=${resume.n} does not match maxN=${cfg.maxN}`) + } + break + case 'continue': + if (resume.n >= cfg.maxN) { + fail(`decision 'continue' at n=${resume.n} with maxN=${cfg.maxN}; the stream is finished`) + } + if (resume.n >= cfg.minN && resume.wealth >= threshold) { + fail( + `decision 'continue' at n=${resume.n} ≥ minN=${cfg.minN} with e-value ` + + `${resume.wealth} ≥ 1/α=${threshold}; this stream would have promoted`, + ) + } + break + } } -function makeStream(cfg: ResolvedConfig): SequentialStream { - const proc = eProcess({ alpha: cfg.alpha, maxBet: cfg.maxBet, nullMean: cfg.nullMean }) +function makeStream(cfg: ResolvedConfig, resume?: SequentialStreamState): SequentialStream { + if (resume !== undefined) assertResumableStream(resume, cfg) + const proc = eProcess({ + alpha: cfg.alpha, + maxBet: cfg.maxBet, + nullMean: cfg.nullMean, + ...(resume === undefined ? {} : { resume }), + }) const threshold = 1 / cfg.alpha - let terminal: SequentialDecision | undefined + let terminal: SequentialDecision | undefined = + resume === undefined || resume.decision === 'continue' ? undefined : resume.decision return { observe(delta: number): SequentialObservation { if (terminal === 'undecided-at-maxN') { @@ -296,7 +362,7 @@ export function sequentialPairedGate Math.min(1, Math.max(0, 0.7 + (rng() - 0.5) * 0.6))) +} + +interface Trace { + wealths: number[] + decidedFlags: boolean[] + final: EProcessState +} + +function runUninterrupted(xs: number[], opts: Parameters[0]): Trace { + const p = eProcess(opts) + const wealths: number[] = [] + const decidedFlags: boolean[] = [] + for (const x of xs) { + const step = p.update(x) + wealths.push(step.wealth) + decidedFlags.push(step.decided) + } + return { wealths, decidedFlags, final: p.state() } +} + +/** Run the first k observations, snapshot `state()`, serialize it through + * JSON (what a restart-safe store does), rebuild, and run the rest. */ +function runWithRestart(xs: number[], opts: Parameters[0], k: number): Trace { + const first = eProcess(opts) + const wealths: number[] = [] + const decidedFlags: boolean[] = [] + for (const x of xs.slice(0, k)) { + const step = first.update(x) + wealths.push(step.wealth) + decidedFlags.push(step.decided) + } + const persisted = JSON.parse(JSON.stringify(first.state())) as EProcessState + const second = eProcess({ ...opts, resume: persisted }) + for (const x of xs.slice(k)) { + const step = second.update(x) + wealths.push(step.wealth) + decidedFlags.push(step.decided) + } + return { wealths, decidedFlags, final: second.state() } +} + +// ── state(): complete and plain ─────────────────────────────────────── + +describe('eProcess.state — the running sums the next bet is computed from', () => { + it('reports the sums the bet formula reads', () => { + const p = eProcess({ alpha: 0.1, maxBet: 0.4, nullMean: 0.5 }) + for (const x of [0.2, 0.9, 0.6]) p.update(x) + const s = p.state() + expect(s.sumX).toBeCloseTo(0.2 + 0.9 + 0.6, 12) + // varSum = Σ (x_i − μ̂_i)² with the shrunk running mean μ̂_i = (1/2 + Σ_{j≤i} x_j)/(i+1). + const mu1 = (0.5 + 0.2) / 2 + const mu2 = (0.5 + 0.2 + 0.9) / 3 + const mu3 = (0.5 + 0.2 + 0.9 + 0.6) / 4 + expect(s.varSum).toBeCloseTo((0.2 - mu1) ** 2 + (0.9 - mu2) ** 2 + (0.6 - mu3) ** 2, 12) + }) + + it('a JSON round-trip of the snapshot rebuilds the same process', () => { + const p = eProcess() + for (const x of [0.8, 0.4, 0.95]) p.update(x) + const snapshot = p.state() + const persisted = JSON.parse(JSON.stringify(snapshot)) as EProcessState + expect(eProcess({ resume: persisted }).state()).toEqual(snapshot) + }) +}) + +// ── Restart reconstruction ──────────────────────────────────────────── + +describe('eProcess.resume — reconstruction after a restart', () => { + const opts = { alpha: 0.05, maxBet: 0.5, nullMean: 0.5 } + const xs = edgeStream(2024, 120) + const reference = runUninterrupted(xs, opts) + + it('the uninterrupted reference decides inside the stream (the fixture has an edge)', () => { + expect(reference.final.decided).toBe(true) + expect(reference.final.decidedAtN).toBeGreaterThan(1) + expect(reference.final.decidedAtN).toBeLessThan(xs.length) + }) + + it.each([0, 1, 7, 30, 119])( + 'interrupting at n=%i and resuming from state() reproduces the wealth sequence and decision exactly', + (k) => { + const resumed = runWithRestart(xs, opts, k) + expect(resumed.wealths).toEqual(reference.wealths) + expect(resumed.decidedFlags).toEqual(reference.decidedFlags) + expect(resumed.final).toEqual(reference.final) + }, + ) + + it('interrupting exactly at the decision crossing keeps decidedAtN and the latch', () => { + const k = reference.final.decidedAtN as number + for (const at of [k - 1, k, k + 1]) { + const resumed = runWithRestart(xs, opts, at) + expect(resumed.wealths).toEqual(reference.wealths) + expect(resumed.final.decidedAtN).toBe(k) + expect(resumed.final.decided).toBe(true) + } + }) + + it('resuming twice in a row (two restarts) is still exact', () => { + const first = eProcess(opts) + for (const x of xs.slice(0, 20)) first.update(x) + const second = eProcess({ ...opts, resume: first.state() }) + for (const x of xs.slice(20, 55)) second.update(x) + const third = eProcess({ ...opts, resume: JSON.parse(JSON.stringify(second.state())) }) + const wealths = [ + ...reference.wealths.slice(0, 55), + ...xs.slice(55).map((x) => third.update(x).wealth), + ] + expect(wealths).toEqual(reference.wealths) + expect(third.state()).toEqual(reference.final) + }) + + it('a resumed process keeps the predictability invariant: the next bet depends only on the snapshot', () => { + // Two different continuations of the same snapshot agree on the first + // post-restart wealth factor whenever their first observation agrees, + // and diverge only from the second observation on. + const p = eProcess(opts) + for (const x of xs.slice(0, 10)) p.update(x) + const snap = p.state() + const a = eProcess({ ...opts, resume: snap }) + const b = eProcess({ ...opts, resume: snap }) + expect(a.update(0.9).wealth).toBe(b.update(0.9).wealth) + expect(a.update(1).wealth).not.toBe(b.update(0).wealth) + }) + + it('with non-default parameters, resume still matches the uninterrupted run', () => { + const custom = { alpha: 0.2, maxBet: 0.9, nullMean: 0.6 } + const ref = runUninterrupted(xs, custom) + const resumed = runWithRestart(xs, custom, 41) + expect(resumed.wealths).toEqual(ref.wealths) + expect(resumed.final).toEqual(ref.final) + }) +}) + +// ── Refusals ────────────────────────────────────────────────────────── + +describe('eProcess.resume — refuses an incompatible or inconsistent snapshot', () => { + function snapshot(n = 12): EProcessState { + const p = eProcess({ alpha: 0.05 }) + for (const x of edgeStream(7, n)) p.update(x) + return p.state() + } + + it('a snapshot taken under different parameters is refused, naming the field', () => { + const s = snapshot() + expect(() => eProcess({ alpha: 0.01, resume: s })).toThrow(ValidationError) + expect(() => eProcess({ alpha: 0.01, resume: s })).toThrow( + /alpha=0\.05 differs from the process alpha=0\.01/, + ) + expect(() => eProcess({ nullMean: 0.6, resume: s })).toThrow(/nullMean=0\.5 differs/) + expect(() => eProcess({ maxBet: 0.25, resume: s })).toThrow(/maxBet=0\.5 differs/) + // The defaults are the parameters when none are passed: a snapshot with + // a non-default alpha needs that alpha passed explicitly. + const tight = eProcess({ alpha: 0.01 }).state() + expect(() => eProcess({ resume: tight })).toThrow( + /alpha=0\.01 differs from the process alpha=0\.05/, + ) + expect(() => eProcess({ alpha: 0.01, resume: tight })).not.toThrow() + }) + + it('a tampered threshold is refused even when alpha matches', () => { + const s = { ...snapshot(), threshold: 50 } + expect(() => eProcess({ resume: s })).toThrow( + /threshold=50 differs from the process threshold=20/, + ) + }) + + it.each<[string, Partial]>([ + ['negative n', { n: -1 }], + ['fractional n', { n: 2.5 }], + ['zero wealth', { wealth: 0 }], + ['negative wealth', { wealth: -1 }], + ['sumX above n', { sumX: 13 }], + ['negative sumX', { sumX: -0.1 }], + ['negative varSum', { varSum: -1e-9 }], + ['decided without decidedAtN', { decided: true }], + ['decidedAtN set while undecided', { decidedAtN: 3 }], + ['decidedAtN above n', { decided: true, decidedAtN: 99 }], + ['decidedAtN zero', { decided: true, decidedAtN: 0 }], + ['undecided with wealth at the threshold', { wealth: 20 }], + ])('refuses %s', (_label, patch) => { + const s = { ...snapshot(), ...patch } + expect(() => eProcess({ resume: s })).toThrow(ValidationError) + expect(() => eProcess({ resume: s })).toThrow(/cannot resume/) + }) + + it('an n=0 snapshot must be the initial state', () => { + const fresh = eProcess().state() + expect(eProcess({ resume: fresh }).state()).toEqual(fresh) + expect(() => eProcess({ resume: { ...fresh, wealth: 1.5 } })).toThrow(/n=0 requires/) + expect(() => eProcess({ resume: { ...fresh, sumX: 0.3 } })).toThrow( + /sumX must lie in \[0, n=0\]/, + ) + }) + + it('a truncated snapshot (missing running sums) is refused, not silently re-zeroed', () => { + const { sumX: _sumX, varSum: _varSum, ...legacy } = snapshot() + void _sumX + void _varSum + expect(() => eProcess({ resume: legacy as EProcessState })).toThrow(/sumX must lie in/) + }) +}) diff --git a/src/statistics/sequential-eprocess.ts b/src/statistics/sequential-eprocess.ts index 8ddb8e29..618df9cd 100644 --- a/src/statistics/sequential-eprocess.ts +++ b/src/statistics/sequential-eprocess.ts @@ -14,6 +14,14 @@ export interface EProcessOptions { * (the paired-delta encoding x = (d+1)/2 maps "no effect" to 1/2). * A pre-registered minEffect shifts this — see `sequentialPairedGate`. */ nullMean?: number + /** Continue a process from a `state()` snapshot taken before a restart. + * The snapshot never supplies the parameters: `alpha`, `maxBet`, and + * `nullMean` resolve from this options object exactly as for a fresh + * process, and a snapshot recorded under different parameters is refused + * (`ValidationError`) — re-deciding the same stream under new parameters + * would reopen optional stopping. A snapshot whose fields are not + * mutually consistent (tampered or truncated) is refused the same way. */ + resume?: EProcessState } export interface EProcessStep { @@ -25,6 +33,14 @@ export interface EProcessStep { decided: boolean } +/** + * Complete snapshot of an e-process. Together with the parameters it is + * sufficient to continue the process after a restart: `sumX` and `varSum` + * are the running sums the next bet is computed from, so a process rebuilt + * from a snapshot produces the same wealth sequence and decision as one that + * was never interrupted. Plain data: a JSON round-trip preserves it + * (`decidedAtN` is undefined, and therefore omitted, until decided). + */ export interface EProcessState extends EProcessStep { alpha: number maxBet: number @@ -33,6 +49,11 @@ export interface EProcessState extends EProcessStep { threshold: number /** Observation count at the first threshold crossing; undefined until decided. */ decidedAtN?: number + /** Σ x_i over the n observations consumed. */ + sumX: number + /** Σ (x_i − μ̂_i)² over the n observations consumed, μ̂_i the shrunk running + * mean after observation i. */ + varSum: number } export interface EProcess { @@ -94,6 +115,15 @@ export function eProcess(opts: EProcessOptions = {}): EProcess { // the next x, so every bet is predictable. let sumX = 0 let varSum = 0 + if (opts.resume !== undefined) { + const r = assertResumable(opts.resume, { alpha, maxBet, nullMean, threshold }) + wealth = r.wealth + n = r.n + decided = r.decided + decidedAtN = r.decidedAtN + sumX = r.sumX + varSum = r.varSum + } return { update(x: number): EProcessStep { if (typeof x !== 'number' || !Number.isFinite(x) || x < 0 || x > 1) { @@ -118,7 +148,72 @@ export function eProcess(opts: EProcessOptions = {}): EProcess { return { wealth, n, decided } }, state(): EProcessState { - return { wealth, n, decided, alpha, maxBet, nullMean, threshold, decidedAtN } + return { wealth, n, decided, alpha, maxBet, nullMean, threshold, decidedAtN, sumX, varSum } }, } } + +interface ResolvedEProcessParameters { + alpha: number + maxBet: number + nullMean: number + threshold: number +} + +/** + * Refuse a snapshot that was not produced by an e-process with exactly these + * parameters, or whose fields cannot all be true at once. Every check names + * the field, so a refused restart reports what disagrees. + */ +function assertResumable(resume: EProcessState, params: ResolvedEProcessParameters): EProcessState { + const fail = (detail: string): never => { + throw new ValidationError(`eProcess: cannot resume — ${detail}`) + } + if (typeof resume !== 'object' || resume === null) { + fail(`resume must be an EProcessState object, got ${String(resume)}`) + } + for (const key of ['alpha', 'maxBet', 'nullMean', 'threshold'] as const) { + if (resume[key] !== params[key]) { + fail( + `snapshot ${key}=${resume[key]} differs from the process ${key}=${params[key]}; ` + + 'a snapshot continues only the process it was taken from', + ) + } + } + const { wealth, n, decided, decidedAtN, sumX, varSum } = resume + if (!Number.isInteger(n) || n < 0) fail(`n must be a non-negative integer, got ${n}`) + if (!Number.isFinite(wealth) || wealth <= 0) { + fail(`wealth must be a finite positive number, got ${wealth}`) + } + if (!Number.isFinite(sumX) || sumX < 0 || sumX > n) { + fail(`sumX must lie in [0, n=${n}] for observations in [0,1], got ${sumX}`) + } + if (!Number.isFinite(varSum) || varSum < 0) { + fail(`varSum must be a finite non-negative number, got ${varSum}`) + } + if (typeof decided !== 'boolean') fail(`decided must be a boolean, got ${String(decided)}`) + if (n === 0 && (wealth !== 1 || sumX !== 0 || varSum !== 0 || decided)) { + fail('n=0 requires wealth=1, sumX=0, varSum=0, decided=false') + } + if (decided) { + if ( + decidedAtN === undefined || + !Number.isInteger(decidedAtN) || + decidedAtN < 1 || + decidedAtN > n + ) { + fail(`decided requires decidedAtN to be an integer in [1, n=${n}], got ${decidedAtN}`) + } + } else { + if (decidedAtN !== undefined) { + fail(`decidedAtN=${decidedAtN} is set while decided=false`) + } + if (wealth >= params.threshold) { + fail( + `wealth ${wealth} ≥ threshold ${params.threshold} while decided=false; ` + + 'decided latches at the first crossing', + ) + } + } + return resume +}