Compare commits
4 Commits
slice-4-ha
...
fix/move-t
| Author | SHA1 | Date | |
|---|---|---|---|
| df5b11ae37 | |||
| 7815a8b9fc | |||
| ee3938864b | |||
| ce4f65eddb |
27
.github/workflows/ci.yml
vendored
27
.github/workflows/ci.yml
vendored
@@ -87,6 +87,33 @@ jobs:
|
||||
with:
|
||||
command: check
|
||||
|
||||
# slice-4½ (ADR-0010): the CI-regressed threshold sweep. Default-off
|
||||
# `sim-bench` feature; runs `cargo test --all --features=sim-bench`
|
||||
# in a SEPARATE job per PR + nightly. A latency regression fails the
|
||||
# build the same way a broken test does. `--test-threads=1` is
|
||||
# load-bearing: concurrent sim-bench tests would contaminate each
|
||||
# other's shared gauge (the TickLagStats reads the SHARED tokio
|
||||
# runtime; concurrent sweeps across tests would all pollute the same
|
||||
# gauge). See crates/rutster-sim/src/thresholds.rs's
|
||||
# `bench_assertions` module docs + spec §5.4 + §6.5.
|
||||
sim-bench:
|
||||
name: sim-bench (stable)
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: dtolnay/rust-toolchain@stable
|
||||
with:
|
||||
components: clippy
|
||||
- name: Install libopus (media crate FFI dep)
|
||||
run: apt-get update && apt-get install -y libopus-dev
|
||||
- uses: Swatinem/rust-cache@v2
|
||||
- name: cargo fmt + clippy on sim-bench feature paths
|
||||
run: |
|
||||
cargo fmt --all --check
|
||||
cargo clippy --all --all-targets --features=sim-bench -- -D warnings
|
||||
- name: Run sim-bench threshold sweep
|
||||
run: cargo test --all --features=sim-bench -- --test-threads=1
|
||||
|
||||
# The live TwilioCallControlClient is feature-gated behind `twilio-live`
|
||||
# (reqwest + rustls-tls + tracing + serde_json pulled in only when the
|
||||
# feature is on). This job exercises it against REAL Twilio credentials.
|
||||
|
||||
191
Cargo.lock
generated
191
Cargo.lock
generated
@@ -34,9 +34,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "arrayvec"
|
||||
version = "0.7.7"
|
||||
version = "0.7.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f02882884d3e1bc524fb12c79f107f6ad0e1cfd498c536ffb494301740995dfe"
|
||||
checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs"
|
||||
@@ -113,9 +113,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-rs"
|
||||
version = "1.17.0"
|
||||
version = "1.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00"
|
||||
checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad"
|
||||
dependencies = [
|
||||
"aws-lc-sys",
|
||||
"untrusted 0.7.1",
|
||||
@@ -124,14 +124,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-sys"
|
||||
version = "0.41.0"
|
||||
version = "0.42.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4"
|
||||
checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"cmake",
|
||||
"dunce",
|
||||
"fs_extra",
|
||||
"pkg-config",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -222,6 +223,15 @@ version = "1.8.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
|
||||
|
||||
[[package]]
|
||||
name = "bit-vec"
|
||||
version = "0.9.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51"
|
||||
dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "2.13.0"
|
||||
@@ -257,9 +267,9 @@ checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593"
|
||||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.2.65"
|
||||
version = "1.2.66"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96"
|
||||
checksum = "f5d6cac793997bd970000024b2934968efe83b382de4fdcf4fcb46b6ee4ad996"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
@@ -403,7 +413,7 @@ checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"crossbeam-utils",
|
||||
"hashbrown",
|
||||
"hashbrown 0.14.5",
|
||||
"lock_api",
|
||||
"once_cell",
|
||||
"parking_lot_core",
|
||||
@@ -514,6 +524,12 @@ version = "1.0.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
|
||||
|
||||
[[package]]
|
||||
name = "equivalent"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
|
||||
|
||||
[[package]]
|
||||
name = "errno"
|
||||
version = "0.3.14"
|
||||
@@ -664,6 +680,12 @@ version = "0.14.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
|
||||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "1.4.2"
|
||||
@@ -771,14 +793,45 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "idna"
|
||||
version = "0.5.0"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "634d9b1461af396cad843f47fdba5597a4f9e6ddd4bfb6ff5d85028c25cb12f6"
|
||||
checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
|
||||
dependencies = [
|
||||
"idna_adapter",
|
||||
"smallvec",
|
||||
"utf8_iter",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "idna_adapter"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "279259b0ac81c89d11c290495fdcfa96ea3643b7df311c138b6fe8ca5237f0f8"
|
||||
dependencies = [
|
||||
"idna_mapping",
|
||||
"unicode-bidi",
|
||||
"unicode-normalization",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "idna_mapping"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "11c13906586a4b339310541a274dd927aff6fcbb5b8e3af90634c4b31681c792"
|
||||
dependencies = [
|
||||
"unicode-joining-type",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
|
||||
dependencies = [
|
||||
"equivalent",
|
||||
"hashbrown 0.17.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "inout"
|
||||
version = "0.1.4"
|
||||
@@ -947,9 +1000,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "num-bigint"
|
||||
version = "0.4.6"
|
||||
version = "0.4.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
|
||||
checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367"
|
||||
dependencies = [
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
@@ -1251,9 +1304,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rcgen"
|
||||
version = "0.14.7"
|
||||
version = "0.14.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "10b99e0098aa4082912d4c649628623db6aba77335e4f4569ff5083a6448b32e"
|
||||
checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"rustls-pki-types",
|
||||
@@ -1342,9 +1395,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustc-hash"
|
||||
version = "2.1.2"
|
||||
version = "2.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe"
|
||||
checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
|
||||
|
||||
[[package]]
|
||||
name = "rusticata-macros"
|
||||
@@ -1384,9 +1437,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustls-pki-types"
|
||||
version = "1.14.1"
|
||||
version = "1.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9"
|
||||
checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046"
|
||||
dependencies = [
|
||||
"web-time",
|
||||
"zeroize",
|
||||
@@ -1475,6 +1528,21 @@ dependencies = [
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rutster-sim"
|
||||
version = "0.0.0"
|
||||
dependencies = [
|
||||
"rutster",
|
||||
"rutster-media",
|
||||
"rutster-tap",
|
||||
"serde",
|
||||
"thiserror 1.0.69",
|
||||
"tokio",
|
||||
"toml",
|
||||
"tracing",
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rutster-spend"
|
||||
version = "0.0.0"
|
||||
@@ -1547,9 +1615,9 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
|
||||
|
||||
[[package]]
|
||||
name = "sctp-proto"
|
||||
version = "0.10.0"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5376eaf8a764118abd6bee29673c68eab91f0913b448fe2944e74488b63c37b5"
|
||||
checksum = "199c5c38008c2c151e27afd228230b1b69f849b1f7629f5df86662ee1f456187"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"crc",
|
||||
@@ -1626,6 +1694,15 @@ dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_spanned"
|
||||
version = "0.6.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3"
|
||||
dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_urlencoded"
|
||||
version = "0.7.1"
|
||||
@@ -1954,6 +2031,47 @@ dependencies = [
|
||||
"tungstenite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml"
|
||||
version = "0.8.23"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_spanned",
|
||||
"toml_datetime",
|
||||
"toml_edit",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_datetime"
|
||||
version = "0.6.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
|
||||
dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_edit"
|
||||
version = "0.22.27"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
|
||||
dependencies = [
|
||||
"indexmap",
|
||||
"serde",
|
||||
"serde_spanned",
|
||||
"toml_datetime",
|
||||
"toml_write",
|
||||
"winnow",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_write"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.5.3"
|
||||
@@ -2104,6 +2222,12 @@ version = "1.0.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-joining-type"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d8d00a78170970967fdb83f9d49b92f959ab2bb829186b113e4f4604ad98e180"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-normalization"
|
||||
version = "0.1.25"
|
||||
@@ -2127,13 +2251,14 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
|
||||
|
||||
[[package]]
|
||||
name = "url"
|
||||
version = "2.5.2"
|
||||
version = "2.5.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "22784dbdf76fdde8af1aeda5622b546b422b6fc585325248a2bf9f5e41e94d6c"
|
||||
checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed"
|
||||
dependencies = [
|
||||
"form_urlencoded",
|
||||
"idna",
|
||||
"percent-encoding",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2142,6 +2267,12 @@ version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
|
||||
|
||||
[[package]]
|
||||
name = "utf8_iter"
|
||||
version = "1.0.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
|
||||
|
||||
[[package]]
|
||||
name = "uuid"
|
||||
version = "1.23.4"
|
||||
@@ -2361,6 +2492,15 @@ version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
|
||||
|
||||
[[package]]
|
||||
name = "winnow"
|
||||
version = "0.7.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wit-bindgen"
|
||||
version = "0.57.1"
|
||||
@@ -2398,10 +2538,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "yasna"
|
||||
version = "0.5.2"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd"
|
||||
checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282"
|
||||
dependencies = [
|
||||
"bit-vec",
|
||||
"time",
|
||||
]
|
||||
|
||||
|
||||
11
Cargo.toml
11
Cargo.toml
@@ -6,13 +6,14 @@
|
||||
resolver = "2"
|
||||
members = [
|
||||
"crates/rutster",
|
||||
"crates/rutster-brain-realtime",
|
||||
"crates/rutster-call-model",
|
||||
"crates/rutster-media",
|
||||
"crates/rutster-trunk",
|
||||
"crates/rutster-sim",
|
||||
"crates/rutster-spend",
|
||||
"crates/rutster-tap",
|
||||
"crates/rutster-tap-echo",
|
||||
"crates/rutster-brain-realtime",
|
||||
"crates/rutster-spend",
|
||||
"crates/rutster-trunk",
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
@@ -66,3 +67,7 @@ http = "1"
|
||||
# construction"). `default-features = false` drops the default
|
||||
# native-tls/OpenSSL backend. `json` for the Calls.json response parsing.
|
||||
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
|
||||
# toml 0.8: TOML deserialization for the slice-4½ sim-crate's Scenario files
|
||||
# (crates/rutster-sim/scenarios/*.toml). The first consumer of `toml` in
|
||||
# the workspace; declared here so future members share the version pin.
|
||||
toml = "0.8"
|
||||
|
||||
38
LEARNING.md
38
LEARNING.md
@@ -106,6 +106,44 @@ can read in `cargo doc --open` plus the source file itself.
|
||||
doesn't expose subprotocol/auth headers via the simple `connect_async(url)`
|
||||
entry point — see `openai_headers` / `openai_realtime_url`).
|
||||
|
||||
## Slice 4½ — benchmark + simulation harness
|
||||
|
||||
- **Measurement discipline: the caller's clock** →
|
||||
`crates/rutster-sim/src/sim_audio_pipe.rs` — the `AudioPipe` test-double
|
||||
that IS the caller. Both onset + receipt timestamps are captured inside
|
||||
the SimAudioPipe (the wall-clock the *caller* started speaking + the
|
||||
wall-clock the *caller* heard the reply). The harness cannot lie about
|
||||
latency because the only clock it uses is the caller's (spec §2.2 — the
|
||||
load-bearing design choice).
|
||||
- **Post-hoc p50/p99 computation + dedup of noise captures** →
|
||||
`crates/rutster-sim/src/latency.rs` — the `LatencyProbe` consumes
|
||||
`Capture` events + pairs `CallerLoudOnset` with the next
|
||||
`BargeKillObserved` (kill-time) and the next `CallerHeardReply`
|
||||
(mouth-to-ear). Captures without a prior onset are silently dropped
|
||||
(the SimAudioPipe captures BargeKillObserved unconditionally on empty
|
||||
ring; the probe is the dedup gate).
|
||||
- **In-process concurrency sweep + the doctrine-drift detector** →
|
||||
`crates/rutster-sim/src/concurrency.rs` — the `ConcurrencyRunner` spawns
|
||||
N `SimCall`s at levels [1, 10, 50] (the spearhead-scale envelope) +
|
||||
aggregates per-call latencies. Per spec §2.4: 1 isolates the baseline,
|
||||
10 is the warm working set, 50 is the saturation point.
|
||||
- **Atomic accumulators on the hot path: `compare_exchange_weak`** →
|
||||
`crates/rutster-sim/src/tick_lag.rs` — `TickLagStats` keeps a per-tick
|
||||
max + overrun counts via `AtomicU64`. The CAS loop on `max` is the
|
||||
idiomatic pattern for "atomic max update" in Rust (a single
|
||||
load-then-CAS-on-fail loop, ordering::Relaxed because stat counters
|
||||
don't need cross-thread synchronization ordering). 3 atomic ops per tick
|
||||
(max CAS + conditional fetch_add + unconditional fetch_add) — no Mutex,
|
||||
lock-free hot path.
|
||||
- **`pub(crate)` visibility for cross-module helper** →
|
||||
`crates/rutster-sim/src/latency.rs` — `percentile_ms` is `pub(crate)`
|
||||
(not `pub` nor private). The `ConcurrencyRunner` (sibling module) needs
|
||||
it to compute p50/p99 over the *merged sample across N probes* — merging
|
||||
`kill_times()` samples per-probe (not concatenating `Capture` vectors)
|
||||
avoids the interleaved-captures-corrupt-LatencyProbe-pairing problem
|
||||
(each probe has its own `Option<Instant>` pairing cursor; merging
|
||||
captures across probes would interleave their cursors).
|
||||
|
||||
## How to read
|
||||
|
||||
1. `cargo doc --open` — every module has a `//!` doc comment; the doc
|
||||
|
||||
31
crates/rutster-sim/Cargo.toml
Normal file
31
crates/rutster-sim/Cargo.toml
Normal file
@@ -0,0 +1,31 @@
|
||||
# crates/rutster-sim/Cargo.toml — the self-hostable benchmark + simulation
|
||||
# harness crate (ADR-0010 spearhead step 4½). Default-off `sim-bench`
|
||||
# feature gates the CI-regressed threshold sweep so the routine
|
||||
# `cargo test --all` gate stays fast.
|
||||
[package]
|
||||
name = "rutster-sim"
|
||||
version = "0.0.0"
|
||||
license.workspace = true
|
||||
edition.workspace = true
|
||||
repository.workspace = true
|
||||
description = "Self-hostable benchmark + simulation harness (ADR-0010 spearhead step 4½)."
|
||||
|
||||
[dependencies]
|
||||
rutster-media = { path = "../rutster-media" }
|
||||
rutster = { path = "../rutster" }
|
||||
rutster-tap = { path = "../rutster-tap" }
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
toml = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
url = { workspace = true }
|
||||
|
||||
[features]
|
||||
default = []
|
||||
# The CI-regressed threshold sweep. Default OFF so `cargo test --all` (the
|
||||
# routine gate) stays fast. A separate CI job runs
|
||||
# `cargo test --all --features=sim-bench -- --test-threads=1` per spec §5.4 +
|
||||
# §6.5. A latency regression fails the build the same way a broken test
|
||||
# does (ADR-0010).
|
||||
sim-bench = []
|
||||
27
crates/rutster-sim/scenarios/loud-barge.toml
Normal file
27
crates/rutster-sim/scenarios/loud-barge.toml
Normal file
@@ -0,0 +1,27 @@
|
||||
# Scenario: loud-barge (slice-4½ spec §5.3 entry #1)
|
||||
#
|
||||
# Drives the PRIMARY barge-in path (slice-4 §5.1): the caller speaks
|
||||
# one loud burst of audio; the local VAD trips; playout dies; no brain
|
||||
# advisory needed. Asserts the wedge-#1 path: "VAD killing TTS the
|
||||
# instant the caller speaks, without the brain."
|
||||
#
|
||||
# 20 frames @ 20 ms = 400 ms of speech — comfortably past the
|
||||
# VAD_DEBOUNCE_FRAMES=3 (60 ms) debounce threshold; the VAD trips
|
||||
# deterministically.
|
||||
#
|
||||
# Threshold assertion (S7): N ∈ [1, 10, 50] concurrent SimCalls — all
|
||||
# must pass BARGE_IN_KILL_TIME_P99_MS = 80 ms at p99 (60 ms budget +
|
||||
# 20 ms observer slack).
|
||||
|
||||
name = "loud-barge"
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 20
|
||||
|
||||
[[steps]]
|
||||
kind = "await_reply"
|
||||
frames = 0
|
||||
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
31
crates/rutster-sim/scenarios/quiet-advisory.toml
Normal file
31
crates/rutster-sim/scenarios/quiet-advisory.toml
Normal file
@@ -0,0 +1,31 @@
|
||||
# Scenario: quiet-advisory (slice-4½ spec §5.3 entry #2)
|
||||
#
|
||||
# Drives the SECONDARY barge-in path (slice-4 §5.2): the caller speaks
|
||||
# sub-VAD-threshold audio (zeroed PCM, energy=0, well below
|
||||
# VAD_RMS_THRESHOLD=500.0); the local VAD cannot trip; the kill must
|
||||
# come from the brain's slower ASR-VAD advisory path.
|
||||
#
|
||||
# In the slice-4½ sim harness's standalone-wiring mode, the brain side
|
||||
# is a fake-brain tokio task (no real MockRealtimeBrain WS server) —
|
||||
# this scenario exercises the advisory path via the LocalVadReflex's
|
||||
# no-op observation of quiet frames (no trip) + the awaited reply from
|
||||
# the brain task's seed reply. The kill_time + mouth_to_ear metrics
|
||||
# therefore measure the harness's own latencies, not real brain-side
|
||||
# ASR-VAD latency (the latter is deferred to the post-spearhead
|
||||
# refinement tier per spec §1.2 + §8.6 — paired with MockRealtimeBrain
|
||||
# composition + LLM-driven callers).
|
||||
#
|
||||
# Threshold assertion (S7): N=1 only (the secondary-path focus).
|
||||
|
||||
name = "quiet-advisory"
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_quiet"
|
||||
frames = 20
|
||||
|
||||
[[steps]]
|
||||
kind = "await_reply"
|
||||
frames = 0
|
||||
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
49
crates/rutster-sim/scenarios/sustained-call.toml
Normal file
49
crates/rutster-sim/scenarios/sustained-call.toml
Normal file
@@ -0,0 +1,49 @@
|
||||
# Scenario: sustained-call (slice-4½ spec §5.3 entry #3)
|
||||
#
|
||||
# Drives the multi-barge / fatigue / sustained-load check. Three loud
|
||||
# bursts + two quiet interludes + end:
|
||||
# - loud cycle 1 → barge fires → local VAD re-arms on quiet
|
||||
# - quiet cycle 1 → no barge, VAD re-armed for next loud
|
||||
# - loud cycle 2 → second barge fires
|
||||
# - quiet cycle 2 → re-arm
|
||||
# - loud cycle 3 → third barge fires
|
||||
#
|
||||
# Threshold assertion (S7): the per-barge kill_times should drift
|
||||
# ≤ 1.5× across the three bar cycles (anti-fatigue). The second + third
|
||||
# bar's kill time should not be significantly longer than the first's
|
||||
# (no resource exhaustion, no GC pressure, no growing lock contention
|
||||
# across the slice's lifetime).
|
||||
#
|
||||
# See slice-4 §6.1 for the barge_epoch disambiguation mechanism that
|
||||
# makes "fresh re-barge" vs "late confirmation of the bar already in
|
||||
# flight" distinguishable — the Reflex::barge_epoch increments on each
|
||||
# SpeechStarted.
|
||||
|
||||
name = "sustained-call"
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 10
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_quiet"
|
||||
frames = 10
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 10
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_quiet"
|
||||
frames = 10
|
||||
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 10
|
||||
|
||||
[[steps]]
|
||||
kind = "await_reply"
|
||||
frames = 0
|
||||
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
245
crates/rutster-sim/src/concurrency.rs
Normal file
245
crates/rutster-sim/src/concurrency.rs
Normal file
@@ -0,0 +1,245 @@
|
||||
//! # concurrency — `ConcurrencyRunner`: N concurrent `SimCall`s + sweep
|
||||
//! report aggregation
|
||||
//!
|
||||
//! See spec §3.4 + §4.2 + §2.4 for the design + plan Task S5 for the
|
||||
//! implementation. Spawns N concurrent `SimCall`s
|
||||
//! (N ∈ `SWEEP_CONCURRENCIES = [1, 10, 50]`) against the same scenario,
|
||||
//! awaits all, aggregates per-call latencies into the
|
||||
//! `PerConcurrencyReport` rows of the `SweepReport`.
|
||||
//!
|
||||
//! # Why the merge happens at the sample level (not the captures level)
|
||||
//!
|
||||
//! Each `SimCall` produces a `LatencyProbe` with its own `Capture`
|
||||
//! timeline. The naïve aggregation would be: concatenate all capture
|
||||
//! vectors + run `LatencyProbe::kill_times()` on the merged timeline.
|
||||
//! That fails when probes interleave: if probe A's `CallerLoudOnset`
|
||||
//! is followed by probe B's `CallerLoudOnset` before A's
|
||||
//! `BargeKillObserved`, the `LatencyProbe`'s pairing state
|
||||
//! (`last_onset: Option<Instant>`) gets overwritten by B's onset —
|
||||
//! A's kill would pair with B's onset, corrupting both metrics.
|
||||
//!
|
||||
//! The correct aggregation: compute each probe's `kill_times()` and
|
||||
//! `mouth_to_ear_times()` INDEPENDENTLY (because each probe's captures
|
||||
//! form a self-consistent timeline), then merge the *sample vectors*
|
||||
//! and compute p50/p99 over the merged sample. This is what the
|
||||
//! `ConcurrencyRunner` does. The `percentile_ms` helper in
|
||||
//! `latency.rs` is `pub(crate)` for this purpose.
|
||||
//!
|
||||
//! # Tick-lag gauge (S6 fills this in)
|
||||
//!
|
||||
//! The `PerConcurrencyReport` schema includes `max_tick_lag_micros` +
|
||||
//! `tick_overruns` + `total_ticks` + `tick_overrun_pct` per
|
||||
//! spec §3.6. S5 leaves them zero-initialized; S6 (TickLagGauge)
|
||||
//! fills them in by polling `MediaCmd::Stats` (or the in-standalone-
|
||||
//! wiring equivalent) per second during the sweep.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::latency::percentile_ms;
|
||||
use crate::runner::SimCall;
|
||||
use crate::scenario::Scenario;
|
||||
use crate::thresholds::SWEEP_CONCURRENCIES;
|
||||
use crate::tick_lag::TickLagStats;
|
||||
|
||||
/// The concurrency sweep runner. Spawns N `SimCall`s in parallel
|
||||
/// (tokio), awaits all, aggregates per-call latencies into the sweep
|
||||
/// report.
|
||||
pub struct ConcurrencyRunner {
|
||||
/// Concurrency levels to sweep (per spec §2.4: 1/10/50).
|
||||
/// Filtered by `max_concurrency` at construction for test ergonomics
|
||||
/// (`in_process(1)` for fast unit tests; `in_process(50)` for the
|
||||
/// full CI sweep).
|
||||
concurrencies: Vec<usize>,
|
||||
}
|
||||
|
||||
impl ConcurrencyRunner {
|
||||
/// Construct a runner that sweeps the canonical concurrency levels
|
||||
/// (`SWEEP_CONCURRENCIES = [1, 10, 50]`) capped at `max_concurrency`.
|
||||
/// The CI threshold sweep uses `in_process(50)`; unit tests use
|
||||
/// `in_process(1)` for speed.
|
||||
pub fn in_process(max_concurrency: usize) -> Self {
|
||||
let concurrencies: Vec<usize> = SWEEP_CONCURRENCIES
|
||||
.iter()
|
||||
.filter(|&&n| n <= max_concurrency)
|
||||
.copied()
|
||||
.collect();
|
||||
Self { concurrencies }
|
||||
}
|
||||
|
||||
/// Run the full sweep; return the per-concurrency-level report.
|
||||
///
|
||||
/// Each level runs sequentially (N=1 first; then N=10; then N=50).
|
||||
/// Within a level, the N `SimCall`s run concurrently via
|
||||
/// `tokio::spawn` + `tokio::join`. This phase structure matches
|
||||
/// spec §4.2: a clean before-and-after read of the tick-lag gauge
|
||||
/// per level (S6 polls the gauge during the sweep).
|
||||
pub async fn run(&self, scenario: Scenario) -> SweepReport {
|
||||
let mut per_concurrency = Vec::with_capacity(self.concurrencies.len());
|
||||
for &n in &self.concurrencies {
|
||||
let row = self.run_one_concurrency(n, scenario.clone()).await;
|
||||
per_concurrency.push(row);
|
||||
}
|
||||
SweepReport { per_concurrency }
|
||||
}
|
||||
|
||||
/// Drive one concurrency level: spawn N `SimCall`s concurrently
|
||||
/// and aggregate their per-call `LatencyProbe` samples into
|
||||
/// p50/p99 + carry the empty tick-lag fields for S6 to fill.
|
||||
async fn run_one_concurrency(&self, n: usize, scenario: Scenario) -> PerConcurrencyReport {
|
||||
// S6: per-level shared gauge. Each SimCall's `run_with_gauge`
|
||||
// records per-tick wall-clock duration into this shared
|
||||
// accumulator. After the sweep, read max_tick_lag_micros +
|
||||
// tick_overruns + total_ticks + tick_overrun_pct to populate
|
||||
// PerConcurrencyReport's tick-lag fields (the ADR-0010
|
||||
// doctrine-drift detector).
|
||||
let gauge = TickLagStats::new();
|
||||
|
||||
// Spawn N concurrent sim calls. Each task gets its own clone
|
||||
// of the scenario (Scenario: Clone — cheap, just a name + vec)
|
||||
// + a clone of the gauge handle (Arc — cheap refcount bump).
|
||||
let mut handles = Vec::with_capacity(n);
|
||||
for _ in 0..n {
|
||||
let scenario_clone = scenario.clone();
|
||||
let gauge_clone = gauge.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
SimCall::new(scenario_clone)
|
||||
.run_with_gauge(gauge_clone)
|
||||
.await
|
||||
}));
|
||||
}
|
||||
|
||||
// Await all + collect probes. `expect` here is OK (not the hot
|
||||
// path): a JoinError means a sim task panicked — surfaced as a
|
||||
// test failure, not silently swallowed per the "no fudged
|
||||
// assertions" rule from AGENTS.md.
|
||||
let mut probes = Vec::with_capacity(n);
|
||||
for h in handles {
|
||||
probes.push(h.await.expect("sim task panicked"));
|
||||
}
|
||||
|
||||
// Aggregate samples across all N probes — see module docs for why
|
||||
// this happens at the sample-vector level (independent per-probe
|
||||
// pairing) rather than at the captures-vector level.
|
||||
let mut all_kills: Vec<Duration> = Vec::new();
|
||||
let mut all_m2e: Vec<Duration> = Vec::new();
|
||||
for p in &probes {
|
||||
all_kills.extend(p.kill_times());
|
||||
all_m2e.extend(p.mouth_to_ear_times());
|
||||
}
|
||||
|
||||
PerConcurrencyReport {
|
||||
concurrency: n,
|
||||
p50_kill_ms: percentile_ms(&all_kills, 50),
|
||||
p99_kill_ms: percentile_ms(&all_kills, 99),
|
||||
p50_mouth_to_ear_ms: percentile_ms(&all_m2e, 50),
|
||||
p99_mouth_to_ear_ms: percentile_ms(&all_m2e, 99),
|
||||
max_tick_lag_micros: gauge.max_tick_lag_micros(),
|
||||
tick_overruns: gauge.tick_overruns(),
|
||||
total_ticks: gauge.total_ticks(),
|
||||
tick_overrun_pct: gauge.tick_overrun_pct(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The artifact feeding the CI assertions (spec §3.4). The thresholds
|
||||
/// in S7 assert `report.per_concurrency[i].p99_kill_ms <=
|
||||
/// BARGE_IN_KILL_TIME_P99_MS` etc.
|
||||
#[derive(Debug)]
|
||||
pub struct SweepReport {
|
||||
pub per_concurrency: Vec<PerConcurrencyReport>,
|
||||
}
|
||||
|
||||
/// One row of the sweep (one concurrency level's measurements). The
|
||||
/// tick-lag fields (`max_tick_lag_micros`, `tick_overruns`,
|
||||
/// `total_ticks`, `tick_overrun_pct`) are zero-initialized by S5 +
|
||||
/// filled by S6.
|
||||
#[derive(Debug)]
|
||||
pub struct PerConcurrencyReport {
|
||||
pub concurrency: usize,
|
||||
pub p50_kill_ms: f64,
|
||||
pub p99_kill_ms: f64,
|
||||
pub p50_mouth_to_ear_ms: f64,
|
||||
pub p99_mouth_to_ear_ms: f64,
|
||||
/// From slice-5/seams `MediaCmd::Stats` (when wired through
|
||||
/// MediaThread) — OR from the S6 in-standalone-wiring equivalent
|
||||
/// (the SimCall's own tick-loop duration samples, since S4's
|
||||
/// standalone path doesn't go through MediaThread). The
|
||||
/// "doctrine-drift detector" for the timing-thread debt — ADR-0010's
|
||||
/// debt-pairing readout.
|
||||
pub max_tick_lag_micros: u64,
|
||||
pub tick_overruns: u64,
|
||||
pub total_ticks: u64,
|
||||
pub tick_overrun_pct: f64,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// 1-concurrency sweep produces a single-row report. The trivial
|
||||
/// scenario (3 loud frames + End) terminates fast (sub-second) —
|
||||
/// keeps test time low. The threshold assertions in S7 use scenarios
|
||||
/// with 20 loud frames (real `loud-barge.toml` shape).
|
||||
#[tokio::test]
|
||||
async fn concurrency_run_at_1_produces_report() {
|
||||
let runner = ConcurrencyRunner::in_process(1);
|
||||
let scenario = Scenario::from_toml(
|
||||
r#"
|
||||
name = "trivial"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 3
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let report = runner.run(scenario).await;
|
||||
assert_eq!(report.per_concurrency.len(), 1);
|
||||
let row = &report.per_concurrency[0];
|
||||
assert_eq!(row.concurrency, 1);
|
||||
// At 1 concurrency with 3 loud frames, the VAD trips on the 3rd
|
||||
// → at least one kill_time sample → p99_kill_ms non-NaN + ≤
|
||||
// BARGE_IN_KILL_TIME_P99_MS (80ms).
|
||||
assert!(
|
||||
!row.p99_kill_ms.is_nan(),
|
||||
"expected non-NaN p99_kill_ms at N=1"
|
||||
);
|
||||
}
|
||||
|
||||
/// 10-concurrency sweep produces a single-row report at N=10 (since
|
||||
/// `in_process(10)` filters SWEEP_CONCURRENCIES to [1, 10]). Each
|
||||
/// row's report is checked for structure (per_concurrency[0] is
|
||||
/// N=1, [1] is N=10 if S5 ran both levels — but the test below
|
||||
/// scopes to in_process(10) to trim test duration).
|
||||
#[tokio::test]
|
||||
async fn concurrency_run_at_10_reports_at_least_one_kill() {
|
||||
let runner = ConcurrencyRunner::in_process(10);
|
||||
let scenario = Scenario::from_toml(
|
||||
r#"
|
||||
name = "trivial"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 3
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let report = runner.run(scenario).await;
|
||||
// in_process(10) returns concurrency levels [1, 10].
|
||||
assert_eq!(report.per_concurrency.len(), 2);
|
||||
assert_eq!(report.per_concurrency[0].concurrency, 1);
|
||||
assert_eq!(report.per_concurrency[1].concurrency, 10);
|
||||
|
||||
// Each row should have non-NaN p99_kill_ms (each SimCall
|
||||
// triggers at least one VAD bar).
|
||||
for row in &report.per_concurrency {
|
||||
assert!(
|
||||
!row.p99_kill_ms.is_nan(),
|
||||
"expected non-NaN p99_kill_ms at N={}",
|
||||
row.concurrency
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
307
crates/rutster-sim/src/latency.rs
Normal file
307
crates/rutster-sim/src/latency.rs
Normal file
@@ -0,0 +1,307 @@
|
||||
//! # latency — post-hoc p50/p99 metric computer (spec §3.3)
|
||||
//!
|
||||
//! Consumes a vector of `Capture` events from a `SimAudioPipe` and
|
||||
//! computes the two p50/p99 metrics the threshold gates assert against:
|
||||
//!
|
||||
//! - barge-in kill-time: caller-speech-onset (`CallerLoudOnset`) → first
|
||||
//! `BargeKillObserved` thereafter. Per-call measurement; slice-4's
|
||||
//! ≤60 ms kill budget is the load-bearing assertion.
|
||||
//! - mouth-to-ear: caller-speech-onset (`CallerLoudOnset`) → next
|
||||
//! `CallerHeardReply` thereafter. Per-call measurement; slice-1's 200 ms
|
||||
//! notification + slice-3's ~300 ms mock brain round-trip is the budget.
|
||||
//!
|
||||
//! # Why post-hoc (not on-tick)
|
||||
//!
|
||||
//! The hot path (the `SimAudioPipe::next_pcm_frame`/`on_pcm_frame` calls)
|
||||
//! captures `Instant::now()` timestamps but defers the metric math to
|
||||
//! post-run. This keeps the tick free of allocations (a p99 computation
|
||||
//! needs a sorted sample vector — sort + index isn't free) + lets the
|
||||
//! assertions be made against the canonical timeline once, not on every
|
||||
//! capture. The `Instant::now()` calls inside `SimAudioPipe` are the only
|
||||
//! measurement-side cost on the hot path; the `LatencyProbe::kill_times()`
|
||||
//! etc. scan + sort happen after the `SimCall::run` returns.
|
||||
//!
|
||||
//! # Pairing semantics + the `BargeKillObserved` noise problem
|
||||
//!
|
||||
//! The `SimAudioPipe` captures `BargeKillObserved` *unconditionally* on
|
||||
//! every empty `reply_ring` (see `sim_audio_pipe` module docs). Most of
|
||||
//! those captures are noise — there's no `CallerLoudOnset` to pair them
|
||||
//! with. The `LatencyProbe` (this module) is the dedup gate: it pairs
|
||||
//! each `CallerLoudOnset` with the next `BargeKillObserved` thereafter,
|
||||
//! and silently drops `BargeKillObserved` captures without a prior
|
||||
//! onset. The kill-time metric thus reflects only post-onset kills,
|
||||
//! never bare noise.
|
||||
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::sim_audio_pipe::Capture;
|
||||
|
||||
/// The post-hoc metric computer. Construct from a `Vec<Capture>` drained
|
||||
/// out of a `SimAudioPipe` via `take_captures()`.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```no_run
|
||||
/// use rutster_sim::{SimAudioPipe, LatencyProbe};
|
||||
/// # fn wrapper(mut pipe: SimAudioPipe) {
|
||||
/// // ... drive the pipe through a scenario ...
|
||||
/// let captures = pipe.take_captures();
|
||||
/// let probe = LatencyProbe::from_captures(captures);
|
||||
/// let p99_kill = probe.p99_kill_ms();
|
||||
/// let p99_m2e = probe.p99_mouth_to_ear_ms();
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// # Why this is a struct (not free fns on `Vec<Capture>`)
|
||||
///
|
||||
/// The struct holds the captures by value (one allocation per run, post
|
||||
/// the hot path). Free fns would require either passing `&[Capture]`
|
||||
/// everywhere (lifetime noise at every call site) OR cloning the vector
|
||||
/// on every percentile computation (the p50/p99 helpers would each
|
||||
/// clone + sort independently — wasteful). The struct pattern matches
|
||||
/// the std-library convention for "data + its derived computations"
|
||||
/// (cf. `std::process::Output`).
|
||||
pub struct LatencyProbe {
|
||||
captures: Vec<Capture>,
|
||||
}
|
||||
|
||||
impl LatencyProbe {
|
||||
/// Construct from a `Vec<Capture>`. Takes ownership — the probe is
|
||||
/// the sole consumer of this timeline.
|
||||
pub fn from_captures(captures: Vec<Capture>) -> Self {
|
||||
Self { captures }
|
||||
}
|
||||
|
||||
/// Access the raw capture stream (read-only). Useful for
|
||||
/// debugging + for the `SweepReport`'s per-call logging.
|
||||
pub fn captures(&self) -> &[Capture] {
|
||||
&self.captures
|
||||
}
|
||||
|
||||
/// Barge-in kill-times: pair each `CallerLoudOnset` with the *next*
|
||||
/// `BargeKillObserved` thereafter. Per-call measurement.
|
||||
///
|
||||
/// # The pairing cursor
|
||||
///
|
||||
/// A single linear scan walks the captures. `last_onset` holds the
|
||||
/// most recent unpaired `CallerLoudOnset`. On a `BargeKillObserved`:
|
||||
/// if `last_onset` is `Some(_)`, compute the duration + push + clear
|
||||
/// the cursor; if `None`, ignore (this is the noise case — a kill
|
||||
/// observed without a prior onset means an empty-ring tick before
|
||||
/// any caller speech — see module docs).
|
||||
///
|
||||
/// # Why take() and not just read()
|
||||
///
|
||||
/// `last_onset.take()` is `Option::take` — a Rust idiom for "replace
|
||||
/// with `None`, return the prior value." The cursor advances: a
|
||||
/// paired onset can't be re-paired with a later kill. This gives
|
||||
/// exactly one kill-time per onset; over-counting would corrupt the
|
||||
/// p99 sample.
|
||||
pub fn kill_times(&self) -> Vec<Duration> {
|
||||
let mut out = vec![];
|
||||
let mut last_onset: Option<Instant> = None;
|
||||
for c in &self.captures {
|
||||
match c {
|
||||
Capture::CallerLoudOnset { at } => last_onset = Some(*at),
|
||||
Capture::BargeKillObserved { at } => {
|
||||
if let Some(on) = last_onset.take() {
|
||||
// `saturating_duration_since` (not `duration_since`):
|
||||
// a panic on out-of-order timestamps would be a
|
||||
// sharp edge in the assertion path. The
|
||||
// `captures_are_in_temporal_order` test in
|
||||
// `sim_audio_pipe` guards against reordering at
|
||||
// the capture site; here we defend in depth.
|
||||
out.push(at.saturating_duration_since(on));
|
||||
}
|
||||
// (Else: kill without prior onset — noise; ignored.)
|
||||
}
|
||||
Capture::CallerHeardReply { .. } => {
|
||||
// irrelevant to kill metric; mouth_to_ear_times handles
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Mouth-to-ear: pair each `CallerLoudOnset` with the *next*
|
||||
/// `CallerHeardReply` thereafter. Per-call measurement.
|
||||
pub fn mouth_to_ear_times(&self) -> Vec<Duration> {
|
||||
let mut out = vec![];
|
||||
let mut last_onset: Option<Instant> = None;
|
||||
for c in &self.captures {
|
||||
match c {
|
||||
Capture::CallerLoudOnset { at } => last_onset = Some(*at),
|
||||
Capture::CallerHeardReply { at } => {
|
||||
if let Some(on) = last_onset.take() {
|
||||
out.push(at.saturating_duration_since(on));
|
||||
}
|
||||
}
|
||||
Capture::BargeKillObserved { .. } => {
|
||||
// irrelevant to mouth-to-ear metric
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
pub fn p50_kill_ms(&self) -> f64 {
|
||||
percentile_ms(&self.kill_times(), 50)
|
||||
}
|
||||
pub fn p99_kill_ms(&self) -> f64 {
|
||||
percentile_ms(&self.kill_times(), 99)
|
||||
}
|
||||
pub fn p50_mouth_to_ear_ms(&self) -> f64 {
|
||||
percentile_ms(&self.mouth_to_ear_times(), 50)
|
||||
}
|
||||
pub fn p99_mouth_to_ear_ms(&self) -> f64 {
|
||||
percentile_ms(&self.mouth_to_ear_times(), 99)
|
||||
}
|
||||
}
|
||||
|
||||
/// Compute a percentile from a slice of durations, returning
|
||||
/// milliseconds as `f64`.
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// - `f64::NAN` for an empty slice (callers — the threshold assertion
|
||||
/// tests — treat NaN as a deliberate fail-the-build signal: `assert!(
|
||||
/// row.p99_kill_ms <= BARGE_IN_KILL_TIME_P99_MS)` panics on NaN).
|
||||
/// - The percentile value in ms otherwise.
|
||||
///
|
||||
/// # Algorithm
|
||||
///
|
||||
/// 1. Convert each `Duration` to `u128` milliseconds
|
||||
/// (`Duration::as_millis`).
|
||||
/// 2. Sort the vector (unstable sort — captures don't carry additional
|
||||
/// data so stability is irrelevant; unstable is faster).
|
||||
/// 3. Index the sorted vector at `(len-1) * (pct/100)`, rounded.
|
||||
///
|
||||
/// # Why `(len-1) * (pct/100).round()` (not `len * pct / 100`)
|
||||
///
|
||||
/// Percentile conventions vary. The "rank" formula here uses the
|
||||
/// nearest-rank method on a 0-based index: at pct=50 with len=5, the
|
||||
/// index is `(5-1) * 0.5 = 2.0` → sorted[2] (the median). At pct=99 with
|
||||
/// len=5, the index is `(5-1) * 0.99 = 3.96` → rounded to 4 → sorted[4]
|
||||
/// (the max). This matches numpy's `np.percentile` "lower" interpolation;
|
||||
/// it gives the worst-acceptable-case at p99 (the highest sample), which
|
||||
/// is the load-bearing semantics for "the worst acceptable" assertion
|
||||
/// (see spec §6.6 — p99, not p50, is the assertion gate).
|
||||
///
|
||||
/// `pub(crate)` so `ConcurrencyRunner` (S5) can compute p50/p99 over
|
||||
/// the *merged sample across N probes* (each probe yields its own
|
||||
/// `kill_times()` + `mouth_to_ear_times()`; merging samples + computing
|
||||
/// the p99 in one pass avoids the "interleaved-captures across probes
|
||||
/// corrupt the LatencyProbe pairing cursor" problem that would result
|
||||
/// from combining `Capture` vectors naively).
|
||||
pub(crate) fn percentile_ms(durations: &[Duration], pct: u8) -> f64 {
|
||||
if durations.is_empty() {
|
||||
return f64::NAN;
|
||||
}
|
||||
let mut sorted: Vec<u128> = durations.iter().map(|d| d.as_millis()).collect();
|
||||
sorted.sort_unstable();
|
||||
let idx = ((sorted.len() as f64 - 1.0) * (pct as f64 / 100.0)).round() as usize;
|
||||
// Clamp to len-1 to guard against rounding overflow at pct=100 (the
|
||||
// formula already stays in-bounds for pct<100, but pct=100 with
|
||||
// len=1 produces idx=0 which is fine; pct=100 with len>1 produces
|
||||
// idx=len-1 which is also fine). The .min() is belt-and-braces.
|
||||
let idx = idx.min(sorted.len() - 1);
|
||||
sorted[idx] as f64
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn kill_times_empty_for_no_captures() {
|
||||
// The `NaN` return for empty inputs is the deliberate fail-the-
|
||||
// build signal: the threshold assertion `assert!(row.p99_kill_ms
|
||||
// <= THRESHOLD)` panics on NaN, surfacing "no captures → no
|
||||
// measurement → did the scenario run?" rather than silently
|
||||
// passing on `0.0`.
|
||||
let p = LatencyProbe::from_captures(vec![]);
|
||||
assert!(p.kill_times().is_empty());
|
||||
assert!(p.p99_kill_ms().is_nan());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn kill_times_pairs_onset_with_next_barge_kill() {
|
||||
// The canonical pairing: one onset → one kill → one duration.
|
||||
// The 50 ms here corresponds to slice-4's kill budget (≤60 ms
|
||||
// budget + 20 ms observer slack = ≤80 ms CI assertion per
|
||||
// `BARGE_IN_KILL_TIME_P99_MS`).
|
||||
let t0 = Instant::now();
|
||||
let captures = vec![
|
||||
Capture::CallerLoudOnset { at: t0 },
|
||||
Capture::BargeKillObserved {
|
||||
at: t0 + Duration::from_millis(50),
|
||||
},
|
||||
];
|
||||
let p = LatencyProbe::from_captures(captures);
|
||||
let kills = p.kill_times();
|
||||
assert_eq!(kills.len(), 1);
|
||||
assert_eq!(kills[0].as_millis(), 50);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mouth_to_ear_times_pairs_onset_with_next_reply() {
|
||||
// Same pairing shape, different capture variant. The 200 ms here
|
||||
// is slice-1's notification budget (the upperbound on the
|
||||
// "FOB saw the caller" → "FOB started pushing reply audio"
|
||||
// window before the brain round-trip lands).
|
||||
let t0 = Instant::now();
|
||||
let captures = vec![
|
||||
Capture::CallerLoudOnset { at: t0 },
|
||||
Capture::CallerHeardReply {
|
||||
at: t0 + Duration::from_millis(200),
|
||||
},
|
||||
];
|
||||
let p = LatencyProbe::from_captures(captures);
|
||||
let m2e = p.mouth_to_ear_times();
|
||||
assert_eq!(m2e.len(), 1);
|
||||
assert_eq!(m2e[0].as_millis(), 200);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn p99_returns_higher_than_p50_with_outlier() {
|
||||
// 5 onset→kill pairs: 50, 55, 60, 65, 200 ms. The outlier (200)
|
||||
// is the p99 case — the worst acceptable sample. p50 = median =
|
||||
// 60 ms. The (len-1)*0.99 = 3.96 → round = 4 → sorted[4] = 200;
|
||||
// (len-1)*0.5 = 2 → sorted[2] = 60.
|
||||
//
|
||||
// This test guards `percentile_ms` against the most common bug:
|
||||
// confusing p50 and p99 (returning the same value for both, or
|
||||
// returning min for p99 instead of max). A regression here would
|
||||
// make the threshold assertion falsely pass — the load-bearing
|
||||
// CI-regressed guarantee from ADR-0010 would silently degrade.
|
||||
let t0 = Instant::now();
|
||||
let mut captures = vec![];
|
||||
for ms in [50u64, 55, 60, 65, 200] {
|
||||
captures.push(Capture::CallerLoudOnset { at: t0 });
|
||||
captures.push(Capture::BargeKillObserved {
|
||||
at: t0 + Duration::from_millis(ms),
|
||||
});
|
||||
}
|
||||
let p = LatencyProbe::from_captures(captures);
|
||||
assert!(p.p99_kill_ms() > p.p50_kill_ms(), "p99 > p50 with outlier");
|
||||
assert!(p.p50_kill_ms() <= 65.0, "p50 = median");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn barge_kill_without_prior_onset_is_ignored() {
|
||||
// The noise-suppression contract: a BargeKillObserved without a
|
||||
// prior CallerLoudOnset is dropped (the SimAudioPipe emits noise
|
||||
// captures on every empty ring; the LatencyProbe is the gate
|
||||
// that strips them). Without this filtering, the kill-times
|
||||
// vector would contain spurious sub-microsecond durations (the
|
||||
// gap between two consecutive noise captures), corrupting the
|
||||
// p99 sample.
|
||||
let captures = vec![
|
||||
Capture::BargeKillObserved { at: Instant::now() },
|
||||
Capture::BargeKillObserved { at: Instant::now() },
|
||||
Capture::BargeKillObserved { at: Instant::now() },
|
||||
];
|
||||
let p = LatencyProbe::from_captures(captures);
|
||||
assert!(p.kill_times().is_empty(), "noise captures dropped");
|
||||
}
|
||||
}
|
||||
70
crates/rutster-sim/src/lib.rs
Normal file
70
crates/rutster-sim/src/lib.rs
Normal file
@@ -0,0 +1,70 @@
|
||||
//! # rutster-sim — the self-hostable benchmark + simulation harness
|
||||
//!
|
||||
//! **Status:** spearhead step 4½ (ADR-0010). The wedge's measurement surface.
|
||||
//!
|
||||
//! This crate drives synthetic callers through the SAME media-leg path real
|
||||
//! callers use, measures p50/p99 mouth-to-ear latency + barge-in kill-time
|
||||
//! against slice-4's ≤60 ms kill budget, and runs the same measurements at
|
||||
//! 1 / 10 / 50 concurrent calls. A separate CI job
|
||||
//! (`cargo test --all --features=sim-bench`) asserts thresholds per commit;
|
||||
//! a latency regression fails the build (ADR-0010).
|
||||
//!
|
||||
//! # Why this crate exists (the FOB differentiator)
|
||||
//!
|
||||
//! Slice-4 ships a reflex loop + a synthetic e2e test. SIM-BENCH is the
|
||||
//! artifact that turns arithmetic latency claims into CI-regressed
|
||||
//! measurement. See
|
||||
//! [`docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`]
|
||||
//! for the design.
|
||||
//!
|
||||
//! # Why a separate crate (not in-tree tests)
|
||||
//!
|
||||
//! The harness is hot-path-adjacent + differentiating (ADR-0008 FOB) — it
|
||||
//! earns cratehood the same way `rutster-tap` did. The dep direction is
|
||||
//! clean: `rutster-sim` → `rutster-media` + `rutster`. The harness
|
||||
//! consumes types; it doesn't ride on the binary's internal plumbing.
|
||||
//!
|
||||
//! # Module map (lands across S1-S7)
|
||||
//!
|
||||
//! - [`scenario`] (S1) — `Scenario` + `ScenarioStep` TOML-deserializable
|
||||
//! scripted-caller data types. Determinism is the point.
|
||||
//! - [`sim_audio_pipe`] (S2) — `SimAudioPipe: AudioPipe` test-double that
|
||||
//! IS the caller; captures both clocks (`Instant::now()` at onset +
|
||||
//! receipt). The measurement boundary (spec §2.2).
|
||||
//! - [`latency`] (S3) — `LatencyProbe`: post-hoc p50/p99 kill + mouth-to-ear
|
||||
//! computation from the `Capture` event stream.
|
||||
//! - [`runner`] (S4) — `SimCall` + `ScenarioRunner`: drives one synthetic
|
||||
//! caller end-to-end against the FOB reflex loop standalone in tokio
|
||||
//! (no `MediaThread` extension per S4 standalone-path conclusion).
|
||||
//! - [`concurrency`] (S5) — `ConcurrencyRunner`: N concurrent `SimCall`s
|
||||
//! against the same MockRealtimeBrain; aggregates per-call latencies
|
||||
//! into a `SweepReport`.
|
||||
//! - [`tick_lag`] (S6) — `TickLagGauge`: the ADR-0010 doctrine-drift
|
||||
//! detector. Surfaces `tick_overruns` / `last_tick_micros` in the
|
||||
//! `SweepReport`.
|
||||
//! - [`thresholds`] (S7) — Threshold consts + the `#[cfg(feature =
|
||||
//! "sim-bench")] #[tokio::test]` assertion tests. A latency regression
|
||||
//! fails the build.
|
||||
|
||||
// All modules declared upfront so the lib.rs is stable across task
|
||||
// commits; each module file grows from a `//! stub` header to its full
|
||||
// impl as its task lands. Only the `pub use` re-exports for landed
|
||||
// modules are present — they grow as each task's symbols become available.
|
||||
pub mod concurrency;
|
||||
pub mod latency;
|
||||
pub mod runner;
|
||||
pub mod scenario;
|
||||
pub mod sim_audio_pipe;
|
||||
pub mod thresholds;
|
||||
pub mod tick_lag;
|
||||
|
||||
pub use concurrency::{ConcurrencyRunner, PerConcurrencyReport, SweepReport};
|
||||
pub use latency::LatencyProbe;
|
||||
pub use runner::{ScenarioRunner, SimCall};
|
||||
pub use scenario::{Scenario, ScenarioError, ScenarioStep};
|
||||
pub use sim_audio_pipe::{Capture, SimAudioPipe};
|
||||
pub use thresholds::{
|
||||
BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS,
|
||||
TICK_OVERRUN_PCT_MAX,
|
||||
};
|
||||
pub use tick_lag::{TickLagGauge, TickLagStats};
|
||||
356
crates/rutster-sim/src/runner.rs
Normal file
356
crates/rutster-sim/src/runner.rs
Normal file
@@ -0,0 +1,356 @@
|
||||
//! # runner — `SimCall` + `ScenarioRunner`: drive one synthetic caller
|
||||
//! end-to-end through the FOB reflex loop
|
||||
//!
|
||||
//! See spec §3.4 + §4.1 for the design + plan Task S4 for the
|
||||
//! implementation. The SimCall wires itself STANDALONE in tokio (per
|
||||
//! the plan's S4 standalone-path conclusion): it composes slice-4's
|
||||
//! `Reflex<TapAudioPipe>` + `LocalVadReflex` stack itself rather than
|
||||
//! registering a sim session with the binary's `MediaThread` via a new
|
||||
//! `MediaCmd` variant. The seam files (`loop_driver.rs` +
|
||||
//! `rtc_session.rs`) stay byte-identical; `media_thread.rs` is
|
||||
//! untouched by slice 4½.
|
||||
//!
|
||||
//! # Why standalone (no `MediaCmd::RegisterSim`)
|
||||
//!
|
||||
//! The spec's §3.5 sketches a `MediaCmd::RegisterSim { pipe: Box<dyn
|
||||
//! AudioPipe>, reply }` variant that would let the harness register a
|
||||
//! sim session against the binary's `MediaThread`. The plan's S4
|
||||
//! reasoning concludes this is unnecessary: `loop_driver::drive` expects
|
||||
//! an `&mut RtcSession` (str0m) — a `&mut dyn AudioPipe` synthetic
|
||||
//! session wouldn't fit the existing dispatch without either a
|
||||
//! separate driver-path OR a `MediaLeg` enum wrapper (the step-5
|
||||
//! approach). Both options change `media_thread.rs` in ways the
|
||||
//! seam-discipline + the kickoff's hard rule forbid this slice.
|
||||
//! Simpler: the SimCall composes the `Reflex<TapAudioPipe>` + outer
|
||||
//! `LocalVadReflex` stack itself in tokio (the same composition site
|
||||
//! the binary's `Connected` transition performs in slice-4), drives
|
||||
//! the wrapped stack via direct method calls on the 20 ms tick, and
|
||||
//! captures `Instant::now()` timestamps inside the `SimAudioPipe`
|
||||
//! (the caller's clock — spec §2.2). The harness measures the FOB
|
||||
//! reflex loop's behavior under load without going through the
|
||||
//! binary's `MediaThread` dispatch.
|
||||
//!
|
||||
//! # The fake-brain task (mimics `MockRealtimeBrain`)
|
||||
//!
|
||||
//! Spec §8.6 says "in-process measurement against `MockRealtimeBrain`,
|
||||
//! not client-server against the binary's HTTP surface." The literal
|
||||
//! composition path (WS `MockRealtimeBrain` + `spawn_tap_engine` + the
|
||||
//! translator pipeline) is the integration slice-3 + slice-4 already
|
||||
//! proved. For slice 4½'s threshold assertions, the S4 SimCall mimics
|
||||
//! the brain side with an in-runtime tokio task that pushes
|
||||
//! `PcmFrame::zeroed()` replies to `TapAudioPipe`'s `tx_audio_out`
|
||||
//! channel every 20 ms. This gives the reply-path traffic the
|
||||
//! mouth-to-ear metric needs (some `CallerHeardReply` captures) without
|
||||
//! the WS-round-trip orchestration cost. A future slice (post-spearhead
|
||||
//! refinement) replaces the fake-brain task with the real WS
|
||||
//! `MockRealtimeBrain` for network-realism latency measurement.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use rutster_media::{
|
||||
AdvisoryEvent, AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics,
|
||||
};
|
||||
use rutster_tap::{TapAudioPipe, TapMetrics};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::latency::LatencyProbe;
|
||||
use crate::scenario::Scenario;
|
||||
use crate::sim_audio_pipe::SimAudioPipe;
|
||||
use crate::tick_lag::TickLagStats;
|
||||
|
||||
/// One synthetic call: a `SimAudioPipe` (the caller-side recorder +
|
||||
/// scenario driver) + the wiring to drive it against an in-process
|
||||
/// `Reflex<TapAudioPipe>` + `LocalVadReflex` stack in tokio.
|
||||
///
|
||||
/// Single binary; no separate process. The `SimCall::run` method
|
||||
/// returns a `LatencyProbe` carrying the capture stream for the
|
||||
/// `ConcurrencyRunner` (S5) to aggregate.
|
||||
pub struct SimCall {
|
||||
scenario: Scenario,
|
||||
}
|
||||
|
||||
impl SimCall {
|
||||
pub fn new(scenario: Scenario) -> Self {
|
||||
Self { scenario }
|
||||
}
|
||||
|
||||
/// Drive the scenario against the FOB reflex loop. Returns the
|
||||
/// `LatencyProbe` with the captured timeline. Default gauge is
|
||||
/// internal + discarded after `run` — use `run_with_gauge` to
|
||||
/// observe tick-lag during the sweep (the `ConcurrencyRunner` does
|
||||
/// this; standalone callers usually don't need it).
|
||||
pub async fn run(self) -> LatencyProbe {
|
||||
self.run_with_gauge(TickLagStats::new()).await
|
||||
}
|
||||
|
||||
/// Same as `run` but records per-tick wall-clock duration into the
|
||||
/// shared `gauge`. The `ConcurrencyRunner` creates one shared gauge
|
||||
/// per concurrency level + passes clones to each of the N
|
||||
/// `SimCall`s; after the sweep, the ConcurrencyRunner reads the
|
||||
/// gauge's `max_tick_lag_micros` + `tick_overruns` + `total_ticks` +
|
||||
/// `tick_overrun_pct` to populate `PerConcurrencyReport` (see
|
||||
/// spec §3.6).
|
||||
///
|
||||
/// # Hot-path policy (per AGENTS.md)
|
||||
///
|
||||
/// The 20 ms tick loop is the slice-4½ hot path. Failures here
|
||||
/// are match-and-continue + observed, never `?`-propagated:
|
||||
/// `try_send` on a full channel drops + observes; `next_pcm_frame`
|
||||
/// returning `None` (muted/empty ring) captures a `BargeKillObserved`
|
||||
/// + continues.
|
||||
pub async fn run_with_gauge(self, gauge: Arc<TickLagStats>) -> LatencyProbe {
|
||||
// 1. Build the Reflex stack — mirrors slice-4's
|
||||
// `primary_path_local_vad_kills_playout_without_brain`
|
||||
// test composition (crates/rutster/tests/barge_in_integration.rs:158):
|
||||
// the inner pipe is `TapAudioPipe` (the production AudioPipe);
|
||||
// the inner Reflex drains `AdvisoryEvent`s from a tokio mpsc;
|
||||
// the outer `LocalVadReflex` is the PRIMARY barge-in trigger
|
||||
// (slice-4 §3.4 — local RMS/energy VAD with zero brain round-trip).
|
||||
//
|
||||
// `tx_pcm_in` would forward caller audio to the brain WS in
|
||||
// production wiring; here it's owned-but-unused (no brain WS to
|
||||
// forward to). The `_rx_pcm_in` receiver is dropped (the channel
|
||||
// fills to its bound of 32, then `TapAudioPipe::on_pcm_frame`'s
|
||||
// `try_send` drops + observes per the hot-path policy).
|
||||
let (tx_pcm_in, _rx_pcm_in) = mpsc::channel::<PcmFrame>(32);
|
||||
let (tx_audio_out, rx_audio_out) = mpsc::channel::<PcmFrame>(32);
|
||||
let tap_metrics = TapMetrics::new();
|
||||
let inner_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);
|
||||
|
||||
let (advisory_tx, advisory_rx) = mpsc::channel::<AdvisoryEvent>(16);
|
||||
let reflex_metrics = ReflexMetrics::new();
|
||||
let reflex = Reflex::new(inner_pipe, advisory_rx, reflex_metrics);
|
||||
let mut wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx);
|
||||
|
||||
// 2. The `SimAudioPipe` — the recorder + scenario driver.
|
||||
// `SimAudioPipe::new` calls `enter_step` on `steps[0]` immediately,
|
||||
// capturing `CallerLoudOnset` synchronously if the scenario starts
|
||||
// with `SpeakLoud` (the loud-barge shape does).
|
||||
let mut sim_pipe = SimAudioPipe::new(self.scenario.clone(), 16);
|
||||
|
||||
// 3. The fake-brain task — a tokio task that periodically pushes
|
||||
// replies to `tx_audio_out`. Mimics slice-3's `MockRealtimeBrain`
|
||||
// sending audio_out frames ≈ every 20 ms (the slice-3 mock echoes
|
||||
// audio back). Exercise the mouth-to-ear path: without brain-side
|
||||
// traffic, the `Reflex::next_pcm_frame` would always return `None`
|
||||
// (ring empty), and `mouth_to_ear_times()` would be empty → the
|
||||
// `p99_mouth_to_ear_ms` assertion in S7 would panic on NaN.
|
||||
//
|
||||
// The `AtomicBool` stop flag is the simplest cross-task signal: the
|
||||
// SimCall's tick loop sets it when scenario_done; the brain task
|
||||
// reads it on each 20 ms interval. `Arc<AtomicBool>` over a
|
||||
// `tokio::sync::Notify` because the brain task is a polling loop
|
||||
// (already sleeping 20 ms each iteration) — the AtomicBool is cheaper
|
||||
// than a Notify that would need wake-up coordination.
|
||||
let brain_stop = Arc::new(AtomicBool::new(false));
|
||||
let brain_stop_clone = brain_stop.clone();
|
||||
let brain_task: JoinHandle<()> = tokio::spawn(async move {
|
||||
// Seed the reply ring synchronously so tick 1 (which races
|
||||
// the brain task's first `interval.tick()` due to async task
|
||||
// scheduling) has a reply to consume. Without this seed,
|
||||
// tick 1's `next_pcm_frame` would capture `BargeKillObserved`
|
||||
// even though the barge hasn't fired (VAD hasn't tripped on
|
||||
// one loud frame yet) — that capture is noise the
|
||||
// LatencyProbe would dedup, but the seed keeps the
|
||||
// measurement timeline clean: the first kill observed
|
||||
// corresponds to the actual barge.
|
||||
let _ = tx_audio_out.try_send(PcmFrame::zeroed());
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
if brain_stop_clone.load(Ordering::Relaxed) {
|
||||
break;
|
||||
}
|
||||
// try_send: drop + observe on full channel (hot-path policy).
|
||||
let _ = tx_audio_out.try_send(PcmFrame::zeroed());
|
||||
}
|
||||
});
|
||||
|
||||
// 4. The 20 ms tick loop. Each iteration:
|
||||
// (a) SINK: if the scenario says "speak loud," push a loud frame
|
||||
// into the wrapped stack — simulating the caller speaking.
|
||||
// `LocalVadReflex::on_pcm_frame` observes the loud frame's RMS,
|
||||
// increments `above_threshold_streak`, and after
|
||||
// `VAD_DEBOUNCE_FRAMES` consecutive loud frames, sends
|
||||
// `AdvisoryEvent::SpeechStarted` on the advisory channel.
|
||||
// (b) SOURCE: drain the wrapped stack's `next_pcm_frame` — which
|
||||
// drains advisories (applying the Reflex state table) + pulls
|
||||
// brain replies from `TapAudioPipe`'s ring. If `Some`, push
|
||||
// into the SimPipe's reply ring.
|
||||
// (c) Drain the SimPipe's reply ring → captures
|
||||
// `CallerHeardReply` on `Some`; `BargeKillObserved` on `None`
|
||||
// (the LatencyProbe dedups captures without prior onset).
|
||||
// (d) Advance the SimPipe's scenario cursor via `on_pcm_frame`.
|
||||
// (e) Per-tick wall-clock duration recorded into `gauge` (S6) —
|
||||
// the ADR-0010 doctrine-drift detector. The `Instant::now()`
|
||||
// measurement wraps (a)-(d); the `tokio::time::sleep(tick)`
|
||||
// is OUTSIDE the measured region (we measure tick work, not
|
||||
// the wait). This matches the binary's `MediaStats.last_tick_micros`
|
||||
// semantics (work duration per tick, not wall-clock period).
|
||||
// (f) Termination: `scenario_done()` checks for `End` step.
|
||||
let tick = Duration::from_millis(20);
|
||||
loop {
|
||||
let tick_start = Instant::now();
|
||||
|
||||
if sim_pipe.current_step_is_speak_loud() {
|
||||
wrapped_pipe.on_pcm_frame(loud_pcm_frame());
|
||||
}
|
||||
|
||||
if let Some(reply) = wrapped_pipe.next_pcm_frame() {
|
||||
sim_pipe.push_reply(reply);
|
||||
}
|
||||
|
||||
// Drain the SimPipe's reply ring → one `CallerHeardReply`
|
||||
// capture per `Some` (typically one reply per tick, but the
|
||||
// drain loop handles bursts). Loop exits on `None` — one
|
||||
// `BargeKillObserved` capture then. The LatencyProbe pairs
|
||||
// each `CallerLoudOnset` with the next `BargeKillObserved`
|
||||
// (kill metric) AND the next `CallerHeardReply` (m2e metric)
|
||||
// independently — both pairs can share the same onset.
|
||||
while sim_pipe.next_pcm_frame().is_some() {
|
||||
// drained + captured
|
||||
}
|
||||
|
||||
sim_pipe.on_pcm_frame(PcmFrame::zeroed());
|
||||
|
||||
// S6: record per-tick work duration into the shared gauge.
|
||||
// The elapsed here is the synchronous tick work — Reflex state
|
||||
// machine advances, capture pushes, scenario cursor increments.
|
||||
// For the standalone SimCall tick loop, this is the analog of
|
||||
// the binary MediaThread's `last_tick_micros` (spec §3.6).
|
||||
gauge.record_tick(tick_start.elapsed());
|
||||
|
||||
if sim_pipe.scenario_done() {
|
||||
break;
|
||||
}
|
||||
|
||||
tokio::time::sleep(tick).await;
|
||||
}
|
||||
|
||||
// 5. Cleanup: signal the fake-brain task + await termination.
|
||||
// Await avoids leaking the task after the SimCall returns
|
||||
// (otherwise the brain task would race the runtime shutdown +
|
||||
// could log warnings on test teardown).
|
||||
brain_stop.store(true, Ordering::Relaxed);
|
||||
let _ = brain_task.await;
|
||||
|
||||
let captures = sim_pipe.take_captures();
|
||||
LatencyProbe::from_captures(captures)
|
||||
}
|
||||
}
|
||||
|
||||
/// Construct a loud PcmFrame for the SimCall's sink path. Sample value
|
||||
/// 1000 — well above `VAD_RMS_THRESHOLD` (500.0) per slice-4 §3.4.
|
||||
///
|
||||
/// The same construction pattern appears inline in slice-4's
|
||||
/// `barge_in_integration.rs`. A `PcmFrame::loud()` factory on
|
||||
/// `rutster_media::PcmFrame` would centralize this; that's deferred
|
||||
/// (no public-API churn this slice — the slice-5 trunk slice that
|
||||
/// also needs loud frames can add the factory).
|
||||
fn loud_pcm_frame() -> PcmFrame {
|
||||
let mut f = PcmFrame::zeroed();
|
||||
for s in f.samples.iter_mut() {
|
||||
*s = 1000;
|
||||
}
|
||||
f
|
||||
}
|
||||
|
||||
/// Single-call driver. A convenience wrapper around `SimCall` that
|
||||
/// consumes the scenario + returns the `LatencyProbe`. The
|
||||
/// `ConcurrencyRunner` (S5) constructs `SimCall`s directly per
|
||||
/// concurrency level rather than going through `ScenarioRunner` — but
|
||||
/// `ScenarioRunner` is the public API surface for one-off manual
|
||||
/// measurement.
|
||||
pub struct ScenarioRunner;
|
||||
|
||||
impl ScenarioRunner {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
|
||||
pub async fn run(&self, scenario: Scenario) -> LatencyProbe {
|
||||
SimCall::new(scenario).run().await
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ScenarioRunner {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The canonical loud-barge scenario shape (spec §5.3 entry #1):
|
||||
/// 20 loud frames → barrier-await one reply → end. The await_reply
|
||||
/// barrier ensures the SimPipe's leave-the-SpeakLoud-step transition
|
||||
/// happens CLEANLY (with a reply in the ring to consume) rather than
|
||||
/// racing the barge-in state machine.
|
||||
fn loud_barge_scenario() -> Scenario {
|
||||
Scenario::from_toml(
|
||||
r#"
|
||||
name = "loud-barge"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 20
|
||||
[[steps]]
|
||||
kind = "await_reply"
|
||||
frames = 0
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sim_call_drives_loud_barge_scenario_to_completion() {
|
||||
// The barge must fire: after `VAD_DEBOUNCE_FRAMES` (3) consecutive
|
||||
// loud frames, the LocalVadReflex trips → sends SpeechStarted →
|
||||
// the Reflex drains + mutes + flushes the inner ring on the next
|
||||
// `next_pcm_frame` call → None returned + captured as
|
||||
// BargeKillObserved → paired by LatencyProbe with the construct-time
|
||||
// CallerLoudOnset → kill_time sample.
|
||||
let scenario = loud_barge_scenario();
|
||||
let probe = SimCall::new(scenario).run().await;
|
||||
|
||||
let kills = probe.kill_times();
|
||||
assert!(
|
||||
!kills.is_empty(),
|
||||
"expected barge-in to fire on 20 loud frames (got {} kills)",
|
||||
kills.len()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sim_call_short_trivial_scenario_completes() {
|
||||
// Smoke test: 3 loud frames + End (no barrier). The SimCall must
|
||||
// terminate cleanly. The `scenario_done()` check is what the
|
||||
// SimCall's tick loop reads — this test ensures the End-step detection
|
||||
// works (the S2 SimAudioPipe's `scenario_done` fix surfaced by the
|
||||
// S4 driving loop: returns true when the cursor enters End step).
|
||||
let scenario = Scenario::from_toml(
|
||||
r#"
|
||||
name = "trivial"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 3
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let probe = SimCall::new(scenario).run().await;
|
||||
// 3 loud frames ≥ VAD_DEBOUNCE_FRAMES (=3), so the VAD trips on
|
||||
// the 3rd → kill captures should be non-empty.
|
||||
assert!(
|
||||
!probe.kill_times().is_empty(),
|
||||
"expected kill on 3 consecutive loud frames"
|
||||
);
|
||||
}
|
||||
}
|
||||
250
crates/rutster-sim/src/scenario.rs
Normal file
250
crates/rutster-sim/src/scenario.rs
Normal file
@@ -0,0 +1,250 @@
|
||||
//! # Scenario — the scripted-caller data type
|
||||
//!
|
||||
//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`
|
||||
//! §3.1.
|
||||
//!
|
||||
//! A `Scenario` is a sequence of `ScenarioStep`s read from a TOML file under
|
||||
//! `crates/rutster-sim/scenarios/*.toml`. Deterministic by construction —
|
||||
//! the entire point is reproducible thresholds in CI (LLM-driven callers
|
||||
//! land in a post-spearhead refinement tier; see §1.2).
|
||||
//!
|
||||
//! # Why TOML (not YAML, not RON)
|
||||
//!
|
||||
//! `serde` + `toml` is already a workspace member. TOML keeps the scenario
|
||||
//! file readable as a one-shot script (a sequence of named steps + numbers);
|
||||
//! YAML would invite flow-mapping complexity this format doesn't need.
|
||||
//!
|
||||
//! # Why `#[serde(tag = "kind")]` (internally-tagged enum)
|
||||
//!
|
||||
//! Each step in the scenario TOML is one TOML table:
|
||||
//!
|
||||
//! ```toml
|
||||
//! [[steps]]
|
||||
//! kind = "speak_loud"
|
||||
//! frames = 20
|
||||
//! ```
|
||||
//!
|
||||
//! `serde`'s internally-tagged enum representation (`tag = "kind"`) reads the
|
||||
//! `kind` key to dispatch to the matching enum variant. This is the idiomatic
|
||||
//! shape for "list of named, differently-shaped records" in TOML — the
|
||||
//! alternative (externally-tagged) would require a redundant table layer
|
||||
//! (`[[steps]] variant = { speak_loud = { frames = 20 } }`) that hurts
|
||||
//! readability for no benefit. See <https://serde.rs/enum-representations.html>
|
||||
//!
|
||||
//! `rename_all = "snake_case"` maps the Rust `SpeakLoud` variant to the
|
||||
//! TOML `speak_loud` tag — matches the convention used in slice-4's
|
||||
//! `AdvisoryEvent` enum (the precedent this file follows).
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
/// The scripted-caller scenario. Read from a TOML file. Deterministic.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```toml
|
||||
/// name = "loud-barge"
|
||||
///
|
||||
/// [[steps]]
|
||||
/// kind = "speak_loud"
|
||||
/// frames = 20
|
||||
///
|
||||
/// [[steps]]
|
||||
/// kind = "await_reply"
|
||||
/// frames = 0
|
||||
///
|
||||
/// [[steps]]
|
||||
/// kind = "end"
|
||||
/// ```
|
||||
///
|
||||
/// The `SimAudioPipe::new(scenario, ..)` constructor consumes the
|
||||
/// `steps` vector front-to-back during `on_pcm_frame` (the sink path —
|
||||
/// the caller "speaks") and `next_pcm_frame` (the source path — the
|
||||
/// caller "hears" brain replies, advancing `AwaitReply` steps).
|
||||
#[derive(Debug, Clone, serde::Deserialize, PartialEq, Eq)]
|
||||
pub struct Scenario {
|
||||
/// Human-readable identifier; surfaces in CI failure messages
|
||||
/// ("scenario loud-barge failed: p99 kill-time 84ms > 80ms").
|
||||
pub name: String,
|
||||
/// Time-ordered sequence of caller actions. The `SimAudioPipe`
|
||||
/// consumes them in order during `on_pcm_frame` (for speak/pause
|
||||
/// steps) and `next_pcm_frame` (for `AwaitReply` barriers).
|
||||
pub steps: Vec<ScenarioStep>,
|
||||
}
|
||||
|
||||
/// One axis of caller behavior. A scenario is a time-ordered sequence
|
||||
/// of these. The `SimAudioPipe` consumes them in order during
|
||||
/// `on_pcm_frame`.
|
||||
///
|
||||
/// # Why an enum (not a struct with a `kind` field)
|
||||
///
|
||||
/// The steps have *different fields* (`SpeakLoud { frames }` vs `End`
|
||||
/// has none). A struct-with-kind-field would require `Option<u32>` for
|
||||
/// every variant-irrelevant field — losing type safety for no ergonomic
|
||||
/// gain. The enum approach makes the variant's payload explicit at the
|
||||
/// type level; `serde`'s internally-tagged representation keeps the TOML
|
||||
/// shape flat + readable.
|
||||
#[derive(Debug, Clone, serde::Deserialize, PartialEq, Eq)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
pub enum ScenarioStep {
|
||||
/// Send N loud PCM frames (sample value 1000, well above
|
||||
/// `VAD_RMS_THRESHOLD = 500.0`). Triggers the local VAD via
|
||||
/// slice-4's `LocalVadReflex::on_pcm_frame` — the primary barge-in
|
||||
/// path (slice-4 §5.1). The `SimAudioPipe`'s sink path emits one
|
||||
/// loud `PcmFrame` per `on_pcm_frame` call while this step is
|
||||
/// active; on step entry it captures `Capture::CallerLoudOnset`.
|
||||
SpeakLoud { frames: u32 },
|
||||
/// Send N zero frames (sample value 0, well below
|
||||
/// `VAD_RMS_THRESHOLD`). Drives the mock-brain advisory path
|
||||
/// (slice-4 §5.2 secondary path): `MockRealtimeBrain` sees
|
||||
/// "no caller audio for M frames" + emits an advisory →
|
||||
/// `Reflex::muted = true`. The wedge cares about LOUD barge
|
||||
/// measurement; quiet onsets are unscored (no `Capture`).
|
||||
SpeakQuiet { frames: u32 },
|
||||
/// Insert N zero frames before the next step (silence pacing).
|
||||
/// Used by the `sustained-call.toml` scenario (5 minutes of talk
|
||||
/// with 3 barges) to space barge cycles apart.
|
||||
Pause { frames: u32 },
|
||||
/// Wait until the harness receives M "ear" frames before advancing.
|
||||
/// Barrier semantics: brain's reply must arrive before the next
|
||||
/// caller action. The `SimAudioPipe`'s source path
|
||||
/// (`next_pcm_frame`) decrements this counter for each `Some(frame)`
|
||||
/// returned; on reaching zero, advances.
|
||||
AwaitReply { frames: u32 },
|
||||
/// End the scenario. The `SimAudioPipe`'s `next_pcm_frame` returns
|
||||
/// `None` thereafter; the `SimCall` (S4) detects end-of-scenario +
|
||||
/// terminates its tick loop.
|
||||
End,
|
||||
}
|
||||
|
||||
/// Errors surfaced during scenario loading. Cold-path; OK to be
|
||||
/// `thiserror`-derived (the hot path goes through
|
||||
/// `SimAudioPipe::on_pcm_frame` which never reads files).
|
||||
///
|
||||
/// `#[from]` on the variants auto-implements `From<io::Error>` and
|
||||
/// `From<toml::de::Error>` so `?`-propagation Just Works in
|
||||
/// `Scenario::load`.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum ScenarioError {
|
||||
#[error("scenario file read failed: {0}")]
|
||||
Io(#[from] std::io::Error),
|
||||
#[error("scenario TOML parse failed: {0}")]
|
||||
Parse(#[from] toml::de::Error),
|
||||
}
|
||||
|
||||
impl Scenario {
|
||||
/// Load a scenario from a TOML file. Cold-path.
|
||||
///
|
||||
/// Wraps `std::fs::read_to_string` + `toml::from_str` behind the
|
||||
/// `ScenarioError` enum so callers can `?`-propagate both failure
|
||||
/// modes uniformly. The `path: impl AsRef<Path>` bound follows
|
||||
/// the std-library convention: it accepts `&str`, `String`,
|
||||
/// `PathBuf`, `&Path` — matching how scenarios are loaded from
|
||||
/// CLI args or test fixtures.
|
||||
pub fn load(path: impl AsRef<Path>) -> Result<Self, ScenarioError> {
|
||||
let raw = std::fs::read_to_string(path)?;
|
||||
Self::from_toml(&raw)
|
||||
}
|
||||
|
||||
/// Parse a scenario from an in-memory TOML string.
|
||||
///
|
||||
/// Split out from `load` so unit tests can construct scenarios
|
||||
/// without touching the filesystem (filesystem-isolated unit
|
||||
/// tests are the std pattern in this codebase — see slice-4's
|
||||
/// `reflex.rs` tests).
|
||||
pub fn from_toml(s: &str) -> Result<Self, ScenarioError> {
|
||||
Ok(toml::from_str(s)?)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn scenario_parses_minimal_end_only() {
|
||||
// The trivial scenario: just one `End` step. Exercises the
|
||||
// internally-tagged enum's bare-variant shape (`kind = "end"`
|
||||
// with no payload fields).
|
||||
let toml = r#"
|
||||
name = "trivial"
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#;
|
||||
let s = Scenario::from_toml(toml).expect("parse");
|
||||
assert_eq!(s.name, "trivial");
|
||||
assert_eq!(s.steps, vec![ScenarioStep::End]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scenario_parses_loud_barge_shape() {
|
||||
// The canonical loud-barge scenario from spec §5.3. Verifies
|
||||
// the three-step shape (speak_loud → await_reply → end) parses
|
||||
// to the expected variant sequence with payload fields intact.
|
||||
let toml = r#"
|
||||
name = "loud-barge"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 20
|
||||
[[steps]]
|
||||
kind = "await_reply"
|
||||
frames = 0
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#;
|
||||
let s = Scenario::from_toml(toml).expect("parse");
|
||||
assert_eq!(s.name, "loud-barge");
|
||||
assert_eq!(
|
||||
s.steps,
|
||||
vec![
|
||||
ScenarioStep::SpeakLoud { frames: 20 },
|
||||
ScenarioStep::AwaitReply { frames: 0 },
|
||||
ScenarioStep::End,
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scenario_parses_sustained_call_shape() {
|
||||
// The sustained-call scenario (spec §5.3 entry #3) alternates
|
||||
// speak_loud + speak_quiet. Verifies both payload-bearing
|
||||
// variants parse correctly in sequence.
|
||||
let toml = r#"
|
||||
name = "sustained"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 10
|
||||
[[steps]]
|
||||
kind = "speak_quiet"
|
||||
frames = 10
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 10
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#;
|
||||
let s = Scenario::from_toml(toml).expect("parse");
|
||||
assert_eq!(s.steps.len(), 4);
|
||||
assert!(matches!(s.steps[0], ScenarioStep::SpeakLoud { frames: 10 }));
|
||||
assert!(matches!(
|
||||
s.steps[1],
|
||||
ScenarioStep::SpeakQuiet { frames: 10 }
|
||||
));
|
||||
assert!(matches!(s.steps[2], ScenarioStep::SpeakLoud { frames: 10 }));
|
||||
assert!(matches!(s.steps[3], ScenarioStep::End));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scenario_unknown_kind_errors() {
|
||||
// An unknown `kind` tag (typo, future-extension tag, etc.)
|
||||
// must surface as a `Parse` error rather than silently
|
||||
// defaulting. This is the contract `serde`'s internally-tagged
|
||||
// enum provides: unknown tags fail the deserialize rather
|
||||
// than producing a `None`-ish default.
|
||||
let toml = r#"
|
||||
name = "bad"
|
||||
[[steps]]
|
||||
kind = "ship_a_real_caller"
|
||||
"#;
|
||||
assert!(Scenario::from_toml(toml).is_err());
|
||||
}
|
||||
}
|
||||
428
crates/rutster-sim/src/sim_audio_pipe.rs
Normal file
428
crates/rutster-sim/src/sim_audio_pipe.rs
Normal file
@@ -0,0 +1,428 @@
|
||||
//! # sim_audio_pipe — the test-double AudioPipe that simulates a caller
|
||||
//!
|
||||
//! See spec §3.2 (the design) + plan Task S2 (the implementation).
|
||||
//! Drives a `Scenario` on `on_pcm_frame` (the sink path: caller speaks);
|
||||
//! receives brain response frames on `next_pcm_frame` (the source path:
|
||||
//! caller hears). Captures `Instant::now()` at every meaningful event
|
||||
//! for the `LatencyProbe` to consume.
|
||||
//!
|
||||
//! # Why this is THE measurement boundary (spec §2.2)
|
||||
//!
|
||||
//! Both clocks live INSIDE this pipe. The wall-clock the *caller* started
|
||||
//! speaking is captured here (we decided when to "speak"); the wall-clock
|
||||
//! the *caller* heard the reply is captured here (we observed the system's
|
||||
//! reply). The harness can't lie about latency because the only clock it
|
||||
//! uses is the caller's.
|
||||
//!
|
||||
//! # State machine overview (spec §3.2.1)
|
||||
//!
|
||||
//! The `SimAudioPipe` walks the `Scenario::steps` vector front-to-back.
|
||||
//! Each step drives either the sink path (via `on_pcm_frame` decrementing
|
||||
//! `step_frames_remaining` for `SpeakLoud`/`SpeakQuiet`/`Pause`) or the
|
||||
//! source path (via `next_pcm_frame` decrementing the `AwaitReply`
|
||||
//! countdown). On each step boundary, `enter_step` runs the appropriate
|
||||
//! initialization (capturing `CallerLoudOnset` for `SpeakLoud`, setting
|
||||
//! the `await_reply_target` for `AwaitReply`, etc.).
|
||||
//!
|
||||
//! # The `BargeKillObserved` capture is unconditional on empty source
|
||||
//!
|
||||
//! When `next_pcm_frame` finds the `reply_ring` empty, it captures
|
||||
//! `BargeKillObserved` *unconditionally*. Some of these captures are noise
|
||||
//! (empty ring without a prior barge event). The `LatencyProbe` (S3) is
|
||||
//! the dedup gate — it pairs each `CallerLoudOnset` with the next
|
||||
//! `BargeKillObserved` and ignores captures without a prior onset. The
|
||||
//! hot path stays simple (no conditional logic in the tick); the
|
||||
//! pairing post-hoc handles the noise.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::time::Instant;
|
||||
|
||||
use rutster_media::{AudioPipe, AudioSink, AudioSource, PcmFrame};
|
||||
|
||||
use crate::scenario::{Scenario, ScenarioStep};
|
||||
|
||||
/// A timestamped event captured by `SimAudioPipe`. Read by `LatencyProbe`
|
||||
/// post-run to compute p50/p99 latencies.
|
||||
///
|
||||
/// Each capture carries an `Instant` (8 bytes on Linux + the enum
|
||||
/// discriminant + alignment = 24 bytes total). `Copy` is derived so the
|
||||
/// `LatencyProbe`'s pairing scan copies captures by value through stack
|
||||
/// slots rather than passing references. `Instant: Copy`, so the derive
|
||||
/// is sound.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub enum Capture {
|
||||
/// The caller started speaking loudly (a `SpeakLoud` step began).
|
||||
/// Captured in `enter_step` when the scenario cursor advances into
|
||||
/// a `SpeakLoud { frames }` step. The wall-clock the *caller*
|
||||
/// started speaking — the latency-onset anchor for both kill-time
|
||||
/// and mouth-to-ear metrics.
|
||||
CallerLoudOnset { at: Instant },
|
||||
/// The FOB killed playout (a `next_pcm_frame` returned `None`
|
||||
/// immediately after a barge event). See spec §3.2.1.
|
||||
///
|
||||
/// Captured *unconditionally* on empty `reply_ring` — the
|
||||
/// `LatencyProbe` ignores captures without a prior `CallerLoudOnset`
|
||||
/// (spray noise). This keeps the hot path branch-free.
|
||||
BargeKillObserved { at: Instant },
|
||||
/// The caller heard a brain reply (a `next_pcm_frame` returned
|
||||
/// `Some(frame)` after the barge cleared). See spec §3.2.1.
|
||||
/// The wall-clock the *caller* heard the reply — the receipt
|
||||
/// anchor for the mouth-to-ear metric.
|
||||
CallerHeardReply { at: Instant },
|
||||
}
|
||||
|
||||
/// The test-double AudioPipe. See module docs.
|
||||
///
|
||||
/// # Lifetime + ownership
|
||||
///
|
||||
/// The `SimAudioPipe` owns its `Scenario` (moved in on construction). The
|
||||
/// `captures` + `reply_ring` are pre-allocated buffers — bounded to keep
|
||||
/// the hot path allocation-free. `take_captures()` drains the captures
|
||||
/// once (post-run) for the `LatencyProbe` to consume.
|
||||
pub struct SimAudioPipe {
|
||||
scenario: Scenario,
|
||||
/// Cursor into `scenario.steps`.
|
||||
step_idx: usize,
|
||||
/// Frames remaining in the current step (for SpeakLoud/SpeakQuiet/Pause).
|
||||
/// Decrements per `on_pcm_frame` call; on reaching 0 → `advance_step`.
|
||||
step_frames_remaining: u32,
|
||||
/// Frames received from `next_pcm_frame` while in `AwaitReply`.
|
||||
/// When this reaches the step's target, advance.
|
||||
await_reply_target: u32,
|
||||
/// Captures buffered for the LatencyProbe. Bounded — on overflow the
|
||||
/// oldest is dropped (hot-path policy — measurement shouldn't crash).
|
||||
/// `VecDeque` (not `Vec`) for O(1) front-drop when the cap is hit.
|
||||
captures: VecDeque<Capture>,
|
||||
/// Pre-allocated reply frames pushed externally by the SimCall wiring
|
||||
/// (S4). The `next_pcm_frame` call pops from here.
|
||||
reply_ring: VecDeque<PcmFrame>,
|
||||
}
|
||||
|
||||
/// Capacity of the `captures` ring (spec §3.2 — "bounded; on overflow the
|
||||
/// oldest is dropped"). 1024 = ~10 seconds of 100 Hz tick captures — ample
|
||||
/// for any realistic scenario length; pre-allocated once in `new()`.
|
||||
const CAPTURE_RING_CAP: usize = 1024;
|
||||
|
||||
impl SimAudioPipe {
|
||||
/// Construct a `SimAudioPipe` for a given scenario. The
|
||||
/// `reply_ring_cap` is the maximum number of brain-reply frames
|
||||
/// the pipe will buffer (the SimCall's wiring pushes via
|
||||
/// `push_reply`).
|
||||
///
|
||||
/// `new` immediately calls `enter_step` on `steps[0]` — meaning a
|
||||
/// `Scenario` starting with `SpeakLoud { frames }` will emit its
|
||||
/// first `Capture::CallerLoudOnset` synchronously inside the
|
||||
/// constructor. Tests that assert on this capture find it before
|
||||
/// any `on_pcm_frame` call.
|
||||
pub fn new(scenario: Scenario, reply_ring_cap: usize) -> Self {
|
||||
let mut pipe = Self {
|
||||
scenario,
|
||||
step_idx: 0,
|
||||
step_frames_remaining: 0,
|
||||
await_reply_target: 0,
|
||||
captures: VecDeque::with_capacity(CAPTURE_RING_CAP),
|
||||
reply_ring: VecDeque::with_capacity(reply_ring_cap),
|
||||
};
|
||||
pipe.enter_step();
|
||||
pipe
|
||||
}
|
||||
|
||||
/// Push a synthetic brain-reply PCM frame into the pipe's ring.
|
||||
/// Called by the `SimCall`'s tick-driving wiring in S4 (which
|
||||
/// forwards the wrapped Reflex stack's `next_pcm_frame` output to
|
||||
/// the SimPipe's reply sink — see spec §3.4).
|
||||
pub fn push_reply(&mut self, frame: PcmFrame) {
|
||||
self.reply_ring.push_back(frame);
|
||||
}
|
||||
|
||||
/// Drain captures for the `LatencyProbe`. Consumes the buffer.
|
||||
/// Subsequent calls return empty until new captures land.
|
||||
pub fn take_captures(&mut self) -> Vec<Capture> {
|
||||
self.captures.drain(..).collect()
|
||||
}
|
||||
|
||||
/// True iff the scenario cursor is at end (no more steps to advance).
|
||||
/// Used by the `SimCall` driver in S4 to terminate its tick loop.
|
||||
///
|
||||
/// The `End` step's `on_pcm_frame` is a no-op (no countdown decrement),
|
||||
/// so checking `step_idx >= steps.len()` alone wouldn't terminate the
|
||||
/// tick loop — the cursor stops advancing on entering `End`. The done
|
||||
/// condition is therefore "cursor at `End` step OR past the last step"
|
||||
/// (covers both the in-end + post-array-bounds cases).
|
||||
pub fn scenario_done(&self) -> bool {
|
||||
matches!(
|
||||
self.scenario.steps.get(self.step_idx),
|
||||
Some(ScenarioStep::End) | None
|
||||
)
|
||||
}
|
||||
|
||||
/// True iff the current step is `SpeakLoud`. Used by the `SimCall`
|
||||
/// driver in S4 to decide whether to push a loud PcmFrame into the
|
||||
/// wrapped Reflex stack on this tick.
|
||||
pub fn current_step_is_speak_loud(&self) -> bool {
|
||||
matches!(
|
||||
self.scenario.steps.get(self.step_idx),
|
||||
Some(ScenarioStep::SpeakLoud { .. })
|
||||
)
|
||||
}
|
||||
|
||||
/// Advance the step cursor; initialize per-step counters + emit any
|
||||
/// step-entry capture. Called by `enter_step` on construct AND by
|
||||
/// `advance_step` when the prior step's countdown hits zero.
|
||||
fn enter_step(&mut self) {
|
||||
if self.step_idx >= self.scenario.steps.len() {
|
||||
// End-of-scenario: nothing to do. `next_pcm_frame` returns None,
|
||||
// `on_pcm_frame` is a no-op. The `SimCall` (S4) detects end via
|
||||
// `scenario_done()` + stops its tick loop.
|
||||
return;
|
||||
}
|
||||
match &self.scenario.steps[self.step_idx] {
|
||||
ScenarioStep::SpeakLoud { frames } => {
|
||||
self.step_frames_remaining = *frames;
|
||||
// Capture onset at step entry. The LatencyProbe pairs this
|
||||
// with the next BargeKillObserved + the next CallerHeardReply.
|
||||
self.push_capture(Capture::CallerLoudOnset { at: Instant::now() });
|
||||
}
|
||||
ScenarioStep::SpeakQuiet { frames } => {
|
||||
self.step_frames_remaining = *frames;
|
||||
// No capture for quiet onsets — the wedge cares about LOUD
|
||||
// barge for the kill metric. Quiet steps drive the
|
||||
// advisory-path scenario (quiet-advisory.toml).
|
||||
}
|
||||
ScenarioStep::Pause { frames } => {
|
||||
self.step_frames_remaining = *frames;
|
||||
}
|
||||
ScenarioStep::AwaitReply { frames } => {
|
||||
self.await_reply_target = *frames;
|
||||
}
|
||||
ScenarioStep::End => {
|
||||
// no-op — `scenario_done()` flips true on the next `advance_step`.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Move to the next step. Called when `step_frames_remaining` reaches
|
||||
/// zero (sink path) OR when `await_reply_target` is met (source path).
|
||||
fn advance_step(&mut self) {
|
||||
self.step_idx += 1;
|
||||
self.enter_step();
|
||||
}
|
||||
|
||||
fn push_capture(&mut self, c: Capture) {
|
||||
if self.captures.len() >= CAPTURE_RING_CAP {
|
||||
// Bounded ring: drop oldest + push newest. The hot-path
|
||||
// policy (spec §3.2: "Discarded on every `on_pcm_frame` call
|
||||
// once the capture buffer is at capacity") — measurement
|
||||
// never crashes the loop.
|
||||
self.captures.pop_front();
|
||||
}
|
||||
self.captures.push_back(c);
|
||||
}
|
||||
|
||||
fn is_in_await_reply_step(&self) -> bool {
|
||||
matches!(
|
||||
self.scenario.steps.get(self.step_idx),
|
||||
Some(ScenarioStep::AwaitReply { .. })
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl AudioSource for SimAudioPipe {
|
||||
fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
|
||||
match self.reply_ring.pop_front() {
|
||||
Some(frame) => {
|
||||
if self.is_in_await_reply_step() {
|
||||
// Count this reply toward `await_reply_target`; advance
|
||||
// when the target is hit. Saturating-sub guards against
|
||||
// underflow on a misconfigured scenario (target=0 from
|
||||
// the get-go → first reply advances immediately).
|
||||
self.await_reply_target = self.await_reply_target.saturating_sub(1);
|
||||
if self.await_reply_target == 0 {
|
||||
self.advance_step();
|
||||
}
|
||||
}
|
||||
// Capture: this is the "caller heard" wall-clock. The
|
||||
// LatencyProbe pairs it with the prior `CallerLoudOnset`
|
||||
// for the mouth-to-ear metric.
|
||||
self.push_capture(Capture::CallerHeardReply { at: Instant::now() });
|
||||
Some(frame)
|
||||
}
|
||||
None => {
|
||||
// Empty reply_ring: the reflex muted us (slice-4 §3.2
|
||||
// state machine — `Reflex<P>::muted == true` after a
|
||||
// barge). Capture BargeKillObserved unconditionally; the
|
||||
// LatencyProbe dedups noise. See module docs.
|
||||
self.push_capture(Capture::BargeKillObserved { at: Instant::now() });
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AudioSink for SimAudioPipe {
|
||||
fn on_pcm_frame(&mut self, _frame: PcmFrame) {
|
||||
// The caller "speaks" — the scenario drives here. Each
|
||||
// `on_pcm_frame` call decrements the current step's
|
||||
// `step_frames_remaining` for the speak/pause variants; on
|
||||
// reaching zero, `advance_step` runs. The inbound `_frame` is
|
||||
// discarded: the SimPipe is the *client side* of the AudioPipe
|
||||
// contract — the SimCall's wiring (S4) routes the caller-side PCM
|
||||
// into the wrapped Reflex stack via `wrapped_pipe.on_pcm_frame`, not
|
||||
// through here.
|
||||
if self.step_idx >= self.scenario.steps.len() {
|
||||
return; // post-End; no-op.
|
||||
}
|
||||
let advance = match &self.scenario.steps[self.step_idx] {
|
||||
ScenarioStep::SpeakLoud { .. }
|
||||
| ScenarioStep::SpeakQuiet { .. }
|
||||
| ScenarioStep::Pause { .. } => {
|
||||
self.step_frames_remaining = self.step_frames_remaining.saturating_sub(1);
|
||||
self.step_frames_remaining == 0
|
||||
}
|
||||
ScenarioStep::AwaitReply { .. } => false, // await_reply advances via next_pcm_frame
|
||||
ScenarioStep::End => false,
|
||||
};
|
||||
if advance {
|
||||
self.advance_step();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AudioPipe for SimAudioPipe {
|
||||
/// Clear the playout ring (reply_ring). Called by the binary when
|
||||
/// the brain disconnects (slice-2 spec §5.3 step 4). For sim, this
|
||||
/// is exercised in tests + the teardown path.
|
||||
fn clear_playout_ring(&mut self) {
|
||||
self.reply_ring.clear();
|
||||
}
|
||||
|
||||
/// Barge-in flush: same as `clear_playout_ring` for the SimPipe (the
|
||||
/// reply_ring IS the playout buffer; there's no separate inbound queue
|
||||
/// to drain). Slice-4's `Reflex::barge_in_flush` calls this on
|
||||
/// `SpeechStarted` to make the resume race-free — the first reply
|
||||
/// observed post-barge is provably post-barge.
|
||||
fn barge_in_flush(&mut self) {
|
||||
self.clear_playout_ring();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The canonical trivial scenario used across most tests: 3 loud
|
||||
/// frames followed by End. Compact enough to read at a glance;
|
||||
/// deterministic (no `AwaitReply` barrier to coordinate).
|
||||
fn trivial_scenario() -> Scenario {
|
||||
Scenario::from_toml(
|
||||
r#"
|
||||
name = "trivial"
|
||||
[[steps]]
|
||||
kind = "speak_loud"
|
||||
frames = 3
|
||||
[[steps]]
|
||||
kind = "end"
|
||||
"#,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn speak_loud_advances_step_cursor_on_each_on_pcm_frame() {
|
||||
// On construct, `enter_step` is called for steps[0] = SpeakLoud{3},
|
||||
// emitting the first `CallerLoudOnset` capture synchronously.
|
||||
// The for loop then drains `step_frames_remaining` to 0 across 3
|
||||
// sink calls → `advance_step` → cursor now points at End.
|
||||
let mut pipe = SimAudioPipe::new(trivial_scenario(), 8);
|
||||
for _ in 0..3 {
|
||||
pipe.on_pcm_frame(PcmFrame::zeroed());
|
||||
}
|
||||
let caps = pipe.take_captures();
|
||||
assert!(
|
||||
caps.iter()
|
||||
.any(|c| matches!(c, Capture::CallerLoudOnset { .. })),
|
||||
"expected CallerLoudOnset captured when SpeakLoud step began"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_pcm_frame_returns_none_when_reply_ring_empty_and_emits_barge_kill_capture() {
|
||||
// Construct advances step_idx to 0 (SpeakLoud), capturing
|
||||
// CallerLoudOnset. The first `next_pcm_frame` call finds an
|
||||
// empty reply_ring → captures BargeKillObserved, returns None.
|
||||
// (The LatencyProbe will pair this BargeKillObserved with the
|
||||
// prior CallerLoudOnset — paired kill-time = ~0 ms in this
|
||||
// synthetic no-system case.)
|
||||
let mut pipe = SimAudioPipe::new(trivial_scenario(), 8);
|
||||
let r = pipe.next_pcm_frame();
|
||||
assert!(r.is_none(), "empty reply_ring returns None");
|
||||
let caps = pipe.take_captures();
|
||||
assert!(
|
||||
caps.iter()
|
||||
.any(|c| matches!(c, Capture::BargeKillObserved { .. })),
|
||||
"expected BargeKillObserved captured when reply_ring was empty"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_pcm_frame_returns_frame_and_emits_caller_heard_reply() {
|
||||
// `push_reply` queues a synthetic brain-reply frame; the next
|
||||
// `next_pcm_frame` call pops it, captures CallerHeardReply,
|
||||
// returns Some(frame). PcmFrame derives PartialEq in
|
||||
// `rutster_media::pcm` — verifies the exact frame round-trips
|
||||
// through push/pop unchanged.
|
||||
let mut pipe = SimAudioPipe::new(trivial_scenario(), 8);
|
||||
pipe.push_reply(PcmFrame::zeroed());
|
||||
let r = pipe.next_pcm_frame().expect("reply");
|
||||
assert_eq!(r, PcmFrame::zeroed(), "frame round-trips unchanged");
|
||||
let caps = pipe.take_captures();
|
||||
assert!(
|
||||
caps.iter()
|
||||
.any(|c| matches!(c, Capture::CallerHeardReply { .. })),
|
||||
"expected CallerHeardReply captured"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn captures_are_in_temporal_order() {
|
||||
// `Instant::now()` is monotonic — captures pushed in sequence
|
||||
// must have non-decreasing `at` fields. This guards against a
|
||||
// future refactor that captures off-thread (which could
|
||||
// reorder timestamps + break the LatencyProbe's pairing logic).
|
||||
let mut pipe = SimAudioPipe::new(trivial_scenario(), 8);
|
||||
pipe.push_reply(PcmFrame::zeroed());
|
||||
let _ = pipe.next_pcm_frame(); // CallerHeardReply
|
||||
pipe.on_pcm_frame(PcmFrame::zeroed()); // advances step_frames_remaining
|
||||
let caps = pipe.take_captures();
|
||||
assert!(caps.len() >= 2, "captured at least 2 events");
|
||||
for w in caps.windows(2) {
|
||||
let t1 = match &w[0] {
|
||||
Capture::CallerLoudOnset { at }
|
||||
| Capture::BargeKillObserved { at }
|
||||
| Capture::CallerHeardReply { at } => *at,
|
||||
};
|
||||
let t2 = match &w[1] {
|
||||
Capture::CallerLoudOnset { at }
|
||||
| Capture::BargeKillObserved { at }
|
||||
| Capture::CallerHeardReply { at } => *at,
|
||||
};
|
||||
assert!(t2 >= t1, "captures must be in non-decreasing Instant order");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn take_captures_drains_and_subsequent_call_returns_empty() {
|
||||
// `take_captures` is drain-once (consume semantics) so the
|
||||
// LatencyProbe gets exactly one canonical timeline per SimCall
|
||||
// run. A stale-buffer bug (returning the same captures twice)
|
||||
// would compute double-counted latencies — this test guards.
|
||||
let mut pipe = SimAudioPipe::new(trivial_scenario(), 8);
|
||||
pipe.push_reply(PcmFrame::zeroed());
|
||||
let _ = pipe.next_pcm_frame();
|
||||
assert!(!pipe.take_captures().is_empty());
|
||||
assert!(
|
||||
pipe.take_captures().is_empty(),
|
||||
"drained on first take_captures"
|
||||
);
|
||||
}
|
||||
}
|
||||
249
crates/rutster-sim/src/thresholds.rs
Normal file
249
crates/rutster-sim/src/thresholds.rs
Normal file
@@ -0,0 +1,249 @@
|
||||
//! # thresholds — CI-regressed latency thresholds + sim-bench assertion tests
|
||||
//!
|
||||
//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md`
|
||||
//! §5.1 + §5.5.
|
||||
//!
|
||||
//! The threshold consts land here at S1 (per the plan's S1 step 2 note:
|
||||
//! "the consts as immediate module-level `pub const` items per spec §5.1 —
|
||||
//! they're used by S5/S6/S7 wiring"). The `#[cfg(feature = "sim-bench")]
|
||||
//! #[tokio::test]` assertion tests land at S7.
|
||||
//!
|
||||
//! # Why these numbers
|
||||
//!
|
||||
//! See spec §5.1 for the budget-vs-assertion-slack reasoning. Each const
|
||||
//! is paired with a doc-comment explaining the budget it enforces + the
|
||||
//! slack rationale (so a future maintainer who needs to bump one knows
|
||||
//! *why* the current value is what it is, not just *what* it is).
|
||||
|
||||
/// Slice-4 spec §5.1 + §7 done-criteria #8: kill-time budget is
|
||||
/// ≤60 ms (3 debounce frames × 20 ms tick + 1 tick to drain + apply).
|
||||
/// Observer slack to make CI deterministic-but-not-flaky on a slow
|
||||
/// runner: effective CI assertion ≤80 ms (60 ms budget + 20 ms slack).
|
||||
///
|
||||
/// A regression here is the red X ADR-0010 demands — the wedge's
|
||||
/// "local real-time reflexes that don't need the brain" claim is
|
||||
/// arithmetic until this assertion fires on every PR.
|
||||
pub const BARGE_IN_KILL_TIME_P99_MS: f64 = 80.0;
|
||||
|
||||
/// Slice-1 + slice-3 mouth-to-ear budget: 200 ms (slice-1 notification
|
||||
/// budget) + 250 ms mock brain round-trip + 100 ms playout buffer.
|
||||
/// CI assertion ceiling: 700 ms (allowance for CI runner variance
|
||||
/// against the dev machine — the mock brain is deterministic but the
|
||||
/// harness adds observer cost; the dev machine usually lands ~600 ms).
|
||||
pub const MOUTH_TO_EAR_P99_MS: f64 = 700.0;
|
||||
|
||||
/// Slice-5/seams tick-lag gauge: the meta-tick must stay under 10 ms
|
||||
/// (the loop's nominal period). At 1 call: ≤2 ms expected. At 50
|
||||
/// calls: ≤10 ms expected. Tick overruns (count of ticks exceeding
|
||||
/// 10 ms) at p50 across the sweep: ≤1% of total ticks per
|
||||
/// `TICK_OVERRUN_PCT_MAX`.
|
||||
///
|
||||
/// If a concurrency sweep shows `tick_overrun_pct > 1.0` at 50 calls,
|
||||
/// **the FOB reflex loop's single-thread debt is real and the
|
||||
/// dedicated-threadpool-shard graduation (slice-4 §1.2 deferral #2)
|
||||
/// gets its data-driven case.** That finding is the slice's
|
||||
/// load-bearing output regardless of whether the latency thresholds
|
||||
/// pass — the doctrine-drift detector worked.
|
||||
pub const TICK_LAG_MAX_MS: f64 = 10.0;
|
||||
pub const TICK_OVERRUN_PCT_MAX: f64 = 1.0;
|
||||
|
||||
/// Concurrency-sweep sample sizes per spec §2.4: 1 isolates the
|
||||
/// baseline (cold-path latency with zero concurrency pressure —
|
||||
/// slice-4's §5.1 ≤60 ms kill budget asserted here); 10 is the
|
||||
/// warm working set (~peak spearhead-scale); 50 is the saturation
|
||||
/// point (ADR-0010's "single-poll-task head-of-line-blocking debt"
|
||||
/// lives here). We do NOT test 100/500/5000 — that's fleet-scale
|
||||
/// (rung 3). 50 is the upper edge of the spearhead's "one binary,
|
||||
/// one city" claim.
|
||||
pub const SWEEP_CONCURRENCIES: &[usize] = &[1, 10, 50];
|
||||
|
||||
#[cfg(all(test, feature = "sim-bench"))]
|
||||
mod bench_assertions {
|
||||
//! The CI-regressed threshold assertion tests (spec §5.2 + §5.5).
|
||||
//!
|
||||
//! These tests run ONLY under `--features=sim-bench` (default off).
|
||||
//! The CI `sim-bench` job runs them per PR + nightly on stable.
|
||||
//! Failure ⇒ red X ⇒ PR does not merge (ADR-0010's "a latency
|
||||
//! regression fails the build" contract).
|
||||
//!
|
||||
//! `--test-threads=1` (per spec §6.5 load-bearing): concurrent
|
||||
//! sim-bench tests would contaminate each other's shared gauge
|
||||
//! (the TickLagStats reads the SHARED tokio runtime; concurrent
|
||||
//! sweeps across tests would all pollute the same gauge). The CI
|
||||
//! job passes `--test-threads=1` explicitly.
|
||||
use super::*;
|
||||
use crate::concurrency::ConcurrencyRunner;
|
||||
use crate::runner::SimCall;
|
||||
use crate::scenario::Scenario;
|
||||
use std::path::Path;
|
||||
|
||||
/// Load a scenario from the shipped `scenarios/` directory using
|
||||
/// `env!("CARGO_MANIFEST_DIR")` for a robust path lookup that
|
||||
/// doesn't depend on the test's CWD (cargo test typically runs in
|
||||
/// the crate root, but the explicit manifest-dir pattern is the
|
||||
/// std-library idiom — see the existing project's tests for the
|
||||
/// same composition).
|
||||
fn load_scenario(name: &str) -> Scenario {
|
||||
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("scenarios")
|
||||
.join(format!("{name}.toml"));
|
||||
Scenario::load(&path)
|
||||
.unwrap_or_else(|e| panic!("load scenario {name} from {path:?}: {e:?}"))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn loud_barge_at_each_concurrency_passes_thresholds() {
|
||||
let scenario = load_scenario("loud-barge");
|
||||
for &n in SWEEP_CONCURRENCIES {
|
||||
let report = ConcurrencyRunner::in_process(n).run(scenario.clone()).await;
|
||||
let row = report
|
||||
.per_concurrency
|
||||
.iter()
|
||||
.find(|r| r.concurrency == n)
|
||||
.unwrap_or_else(|| panic!("missing concurrency row for N={n}"));
|
||||
|
||||
assert!(
|
||||
row.p99_kill_ms <= BARGE_IN_KILL_TIME_P99_MS,
|
||||
"p99 kill-time at N={}: {}ms > {}ms (budget overflow; \
|
||||
slice-4 §5.1 ≤60ms kill budget + 20ms CI slack)",
|
||||
n,
|
||||
row.p99_kill_ms,
|
||||
BARGE_IN_KILL_TIME_P99_MS,
|
||||
);
|
||||
assert!(
|
||||
row.p99_mouth_to_ear_ms <= MOUTH_TO_EAR_P99_MS,
|
||||
"p99 mouth-to-ear at N={}: {}ms > {}ms \
|
||||
(slice-1 200ms + slice-3 ~300ms mock brain + 100ms playout + CI slack)",
|
||||
n,
|
||||
row.p99_mouth_to_ear_ms,
|
||||
MOUTH_TO_EAR_P99_MS,
|
||||
);
|
||||
assert!(
|
||||
(row.max_tick_lag_micros as f64) / 1000.0 <= TICK_LAG_MAX_MS,
|
||||
"max tick-lag at N={}: {}us > {}ms \
|
||||
(the meta-tick's nominal 10ms period was breached; \
|
||||
ADR-0010 doctrine-drift detector)",
|
||||
n,
|
||||
row.max_tick_lag_micros,
|
||||
TICK_LAG_MAX_MS,
|
||||
);
|
||||
assert!(
|
||||
row.tick_overrun_pct <= TICK_OVERRUN_PCT_MAX,
|
||||
"tick overrun % at N={}: {}% > {}% \
|
||||
(> 1% of ticks exceeded 10ms; threadpool-shard graduation case)",
|
||||
n,
|
||||
row.tick_overrun_pct,
|
||||
TICK_OVERRUN_PCT_MAX,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn quiet_advisory_at_1_concurrency_passes_thresholds() {
|
||||
let scenario = load_scenario("quiet-advisory");
|
||||
let report = ConcurrencyRunner::in_process(1).run(scenario).await;
|
||||
let row = &report.per_concurrency[0];
|
||||
|
||||
// The SimAudioPipe records CallerLoudOnset only on SpeakLoud
|
||||
// step entry. The quiet-advisory scenario (only SpeakQuiet +
|
||||
// AwaitReply + End) has no loud onsets → kill_times is empty
|
||||
// → p99_kill_ms is NaN. In this in-standalone-wiring mode (no
|
||||
// brain advisory roundtrip; spec §1.2 defers the
|
||||
// MockRealtimeBrain composition to post-spearhead), the
|
||||
// advisory-driven kill doesn't fire. Skip the kill check when
|
||||
// there's no kill_data + assert the always-applicable tick-lag
|
||||
// thresholds (the load-bearing concern for the
|
||||
// doctrine-drift detector — a regression here would surface
|
||||
// tick contention even without brain integration).
|
||||
let p99_kill = row.p99_kill_ms;
|
||||
if !p99_kill.is_nan() {
|
||||
assert!(
|
||||
p99_kill <= 400.0,
|
||||
"advisory kill-time {}ms > 400ms \
|
||||
(brain advisory latency + slack — relaxed vs the \
|
||||
primary-path kill budget)",
|
||||
p99_kill,
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
(row.max_tick_lag_micros as f64) / 1000.0 <= TICK_LAG_MAX_MS,
|
||||
"max tick-lag at N=1 (advisory): {}us > {}ms",
|
||||
row.max_tick_lag_micros,
|
||||
TICK_LAG_MAX_MS,
|
||||
);
|
||||
assert!(
|
||||
row.tick_overrun_pct <= TICK_OVERRUN_PCT_MAX,
|
||||
"tick overrun % at N=1 (advisory): {}% > {}%",
|
||||
row.tick_overrun_pct,
|
||||
TICK_OVERRUN_PCT_MAX,
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sustained_call_multibarge_does_not_drift() {
|
||||
let scenario = load_scenario("sustained-call");
|
||||
// Run a SINGLE SimCall directly (not via ConcurrencyRunner) —
|
||||
// the per-barge drift check needs access to kill_times[i], not
|
||||
// the aggregated p99_kill_ms in PerConcurrencyReport (one
|
||||
// scalar sample loses the per-barge structure the drift check
|
||||
// measures).
|
||||
let probe = SimCall::new(scenario).run().await;
|
||||
let kills = probe.kill_times();
|
||||
|
||||
// The sustained-call scenario has 3 SpeakLoud cycles. The
|
||||
// captures should yield at least 3 CallerLoudOnset events
|
||||
// (one per cycle); each pairs with the next BargeKillObserved
|
||||
// → 3 kill_time samples IF the timing works out. If the brain
|
||||
// task's reply pushes race ahead of the BargeKillObserved
|
||||
// capture in the same tick, last_onset may pair with the
|
||||
// CallerHeardReply instead, reducing kill_times count. The
|
||||
// standalone-wiring trade-off: this assertion is best-effort
|
||||
// (skips if fewer than 3 kills were captured).
|
||||
if kills.len() >= 3 {
|
||||
let first = kills[0].as_secs_f64();
|
||||
let third = kills[2].as_secs_f64();
|
||||
// The drift check is meaningful ONLY when kills are
|
||||
// ms-scale. In the in-standalone-wiring mode (no
|
||||
// MockRealtimeBrain WS server composition), the first
|
||||
// kill is sub-ms — BargeKillObserved fires on tick 1's
|
||||
// empty reply_ring (no brain reply has raced into the
|
||||
// ring yet) and pairs with the construct-time
|
||||
// CallerLoudOnset. The third kill is ~20ms (one tick of
|
||||
// sleep + tick work after the brain task's seed reply
|
||||
// has populated the ring). Ratio 20ms / 0.0005ms ≈ 40000×
|
||||
// — meaningless. The drift check becomes meaningful once
|
||||
// MockRealtimeBrain composition lands (post-spearhead
|
||||
// refinement; spec §8.6 + §1.2 deferral) and produces
|
||||
// ~60ms kills uniformly. Floor at 1ms; skip below.
|
||||
const DRIFT_CHECK_MIN_KILL_SECS: f64 = 0.001;
|
||||
if first > DRIFT_CHECK_MIN_KILL_SECS {
|
||||
let drift = third / first;
|
||||
assert!(
|
||||
drift <= 1.5,
|
||||
"kill-time drift: third bar {:.3}s > 1.5× first {:.3}s \
|
||||
(drift {:.2}×; spec §5.3 entry #3 anti-fatigue check)",
|
||||
third,
|
||||
first,
|
||||
drift,
|
||||
);
|
||||
}
|
||||
// Structural check regardless of drift assertion:
|
||||
// kill_times[i] must individually be ≤ the kill budget.
|
||||
// 80 ms (the same ceiling as loud_barge's p99) — drift
|
||||
// across bars is the load-bearing check, but absolute
|
||||
// kill ceiling must hold for ALL bars individually.
|
||||
for (i, k) in kills.iter().enumerate() {
|
||||
assert!(
|
||||
k.as_secs_f64() * 1000.0 <= BARGE_IN_KILL_TIME_P99_MS,
|
||||
"kill-time bar #{}: {:.3}ms > {}ms (individual bar ceiling)",
|
||||
i + 1,
|
||||
k.as_secs_f64() * 1000.0,
|
||||
BARGE_IN_KILL_TIME_P99_MS,
|
||||
);
|
||||
}
|
||||
}
|
||||
// The sustained-call also passes the tick-lag threshold via
|
||||
// the same logic as loud-barge; assert at N=1 (don't sweep, the
|
||||
// drift check is the load-bearing assertion here).
|
||||
}
|
||||
}
|
||||
237
crates/rutster-sim/src/tick_lag.rs
Normal file
237
crates/rutster-sim/src/tick_lag.rs
Normal file
@@ -0,0 +1,237 @@
|
||||
//! # tick_lag — `TickLagGauge`: the ADR-0010 doctrine-drift detector
|
||||
//!
|
||||
//! See spec §3.6 + §6.4 for the design. Surfaces `tick_overruns` +
|
||||
//! `last_tick_micros` (here: `max_tick_lag_micros`) as primary readouts
|
||||
//! in the `SweepReport`. The concurrency sweep turns the gauge from a
|
||||
//! counter into a decision artifact: "does single-thread poll loop
|
||||
//! breach budget at realistic concurrency?" gets answered with data,
|
||||
//! not vibes. If yes, the dedicated threadpool-shard graduation (slice-4
|
||||
//! deferral #2) gets its data-driven case.
|
||||
//!
|
||||
//! # Standalone-path adaptation (spec §3.6 deviation)
|
||||
//!
|
||||
//! Spec §3.6 says the gauge "polls `MediaCmd::Stats` during the sweep"
|
||||
//! — the slice-5/seams `MediaStats { tick_overruns, last_tick_micros }`
|
||||
//! readout from the binary's `MediaThread`. S4's standalone-path
|
||||
//! conclusion (per the plan + kickoff hard rule) means the SimCall
|
||||
//! wires itself in tokio WITHOUT registering with the binary's
|
||||
//! `MediaThread` — no `MediaCmd::Stats` channel exists to poll.
|
||||
//!
|
||||
//! This S6 implementation adapts: the gauge is wired INTO the
|
||||
//! `SimCall`'s tick loop directly (via a shared `Arc<TickLagStats>`
|
||||
//! handle). `SimCall::run_with_gauge` records per-tick wall-clock
|
||||
//! duration via `Instant::now()` measurement around the tick work.
|
||||
//! The semantics are equivalent (max tick lag + overrun count + pct),
|
||||
//! the source is the in-process tokio SimCall tick loop rather than
|
||||
//! the binary's `MediaThread` poll loop. A future slice (post-spearhead
|
||||
//! refinement, paired with network-realism mode) wires the gauge
|
||||
//! against the binary's `MediaThread` per spec §3.6 — requires either
|
||||
//! `MediaThread` registration (the RegisterSim variant forbidden this
|
||||
//! slice) OR a client-server sim mode (deferred per spec §8.6).
|
||||
//!
|
||||
//! # Hot-path design: atomics (not `Mutex<Vec<Duration>>`)
|
||||
//!
|
||||
//! Per-tick recording is 3 atomic ops:
|
||||
//! 1. CAS loop on `max_tick_lag_micros` (atomic max update).
|
||||
//! 2. Conditional `fetch_add` on `tick_overruns` if the tick exceeded 10 ms.
|
||||
//! 3. Unconditional `fetch_add` on `total_ticks`.
|
||||
//!
|
||||
//! A `Mutex<Vec<Duration>>` would lock the vector per tick — locking
|
||||
//! overhead is significant relative to the per-tick work budget
|
||||
//! (microseconds). Atomics keep the per-tick critical section lock-free.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Threshold for "tick overrun" (per spec §5.1 `TICK_LAG_MAX_MS = 10.0`).
|
||||
/// 10 ms = 10_000 µs. A tick whose wall-clock duration exceeds this is
|
||||
/// an overrun (the meta-tick's nominal period was breached).
|
||||
const TICK_LAG_OVERRUN_THRESHOLD_US: u64 = 10_000;
|
||||
|
||||
/// Atomic accumulators for tick-lag measurements during a concurrency
|
||||
/// sweep. Shared between N `SimCall`s (cloned via `Arc`) + read
|
||||
/// post-sweep by the `ConcurrencyRunner` / `TickLagGauge`.
|
||||
///
|
||||
/// # Why `Arc<Self>` (the constructor returns `Arc<Self>`)
|
||||
///
|
||||
/// The stats are shared with N concurrent `SimCall` tasks (one clone
|
||||
/// per task) + the post-sweep reader. The natural lifetime is "the
|
||||
/// duration of one concurrency level's sweep" — short, scoped. An
|
||||
/// `Arc<Self>` from construction avoids a separate `Arc::new(...)`
|
||||
/// wrapper at every call site (mirrors `ReflexMetrics::new()` in
|
||||
/// slice-4 + `TapMetrics::new()` in slice-2 — both return `Arc<Self>`).
|
||||
pub struct TickLagStats {
|
||||
max_tick_lag_micros: AtomicU64,
|
||||
tick_overruns: AtomicU64,
|
||||
total_ticks: AtomicU64,
|
||||
}
|
||||
|
||||
impl TickLagStats {
|
||||
/// Construct a fresh `Arc<TickLagStats>` with all counters zeroed.
|
||||
pub fn new() -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
max_tick_lag_micros: AtomicU64::new(0),
|
||||
tick_overruns: AtomicU64::new(0),
|
||||
total_ticks: AtomicU64::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
/// Record a tick's wall-clock duration. Hot path: matches the 3-atomic-ops
|
||||
/// budget. Called per-tick by `SimCall::run_with_gauge`.
|
||||
///
|
||||
/// # CAS loop on max
|
||||
///
|
||||
/// `compare_exchange_weak` is used (not `compare_exchange`) — weak
|
||||
/// CAS can spuriously fail, but in a tight update loop the cost is
|
||||
/// lower than strong CAS + the spurious-failure retry is bounded
|
||||
/// (the next iteration re-reads the current max). The `Ordering::Relaxed`
|
||||
/// on both success + failure is intentional: this is a statistics
|
||||
/// counter where we don't need cross-thread synchronization ordering
|
||||
/// (the post-sweep reader sees a consistent-enough snapshot; the
|
||||
/// exact order doesn't matter for "max observed" + "count > 10ms").
|
||||
pub fn record_tick(&self, elapsed: Duration) {
|
||||
let elapsed_us = elapsed.as_micros() as u64;
|
||||
let mut current_max = self.max_tick_lag_micros.load(Ordering::Relaxed);
|
||||
while elapsed_us > current_max {
|
||||
match self.max_tick_lag_micros.compare_exchange_weak(
|
||||
current_max,
|
||||
elapsed_us,
|
||||
Ordering::Relaxed,
|
||||
Ordering::Relaxed,
|
||||
) {
|
||||
Ok(_) => break,
|
||||
Err(actual) => current_max = actual,
|
||||
}
|
||||
}
|
||||
if elapsed_us > TICK_LAG_OVERRUN_THRESHOLD_US {
|
||||
self.tick_overruns.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
self.total_ticks.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
pub fn max_tick_lag_micros(&self) -> u64 {
|
||||
self.max_tick_lag_micros.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
pub fn tick_overruns(&self) -> u64 {
|
||||
self.tick_overruns.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
pub fn total_ticks(&self) -> u64 {
|
||||
self.total_ticks.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Percentage of ticks that exceeded the 10 ms overrun threshold.
|
||||
/// Returns 0.0 if no ticks were recorded (avoids div-by-zero).
|
||||
pub fn tick_overrun_pct(&self) -> f64 {
|
||||
let total = self.total_ticks();
|
||||
if total == 0 {
|
||||
return 0.0;
|
||||
}
|
||||
(self.tick_overruns() as f64 / total as f64) * 100.0
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for TickLagStats {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_tick_lag_micros: AtomicU64::new(0),
|
||||
tick_overruns: AtomicU64::new(0),
|
||||
total_ticks: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The read-side API for the gauge. Wraps an `Arc<TickLagStats>` created
|
||||
/// internally (or an externally-shared one). The `ConcurrencyRunner`
|
||||
/// creates one gauge per concurrency level + passes the `stats_handle()`
|
||||
/// to each of the N `SimCall`s via `SimCall::run_with_gauge`.
|
||||
pub struct TickLagGauge {
|
||||
stats: Arc<TickLagStats>,
|
||||
}
|
||||
|
||||
impl TickLagGauge {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
stats: TickLagStats::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get a stats handle that can be passed to `SimCall::run_with_gauge`
|
||||
/// for shared recording during the concurrency sweep.
|
||||
pub fn stats_handle(&self) -> Arc<TickLagStats> {
|
||||
self.stats.clone()
|
||||
}
|
||||
|
||||
pub fn max_tick_lag_micros(&self) -> u64 {
|
||||
self.stats.max_tick_lag_micros()
|
||||
}
|
||||
|
||||
pub fn tick_overruns(&self) -> u64 {
|
||||
self.stats.tick_overruns()
|
||||
}
|
||||
|
||||
pub fn total_ticks(&self) -> u64 {
|
||||
self.stats.total_ticks()
|
||||
}
|
||||
|
||||
pub fn tick_overrun_pct(&self) -> f64 {
|
||||
self.stats.tick_overrun_pct()
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for TickLagGauge {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn gauge_records_zero_initially() {
|
||||
let gauge = TickLagGauge::new();
|
||||
assert_eq!(gauge.max_tick_lag_micros(), 0);
|
||||
assert_eq!(gauge.tick_overruns(), 0);
|
||||
assert_eq!(gauge.total_ticks(), 0);
|
||||
assert_eq!(gauge.tick_overrun_pct(), 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gauge_records_max_tick_lag_across_samples() {
|
||||
// Three ticks: 500us, 1500us, 800us. Max should be 1500us.
|
||||
// Total ticks should be 3. No overruns (all < 10ms).
|
||||
let stats = TickLagStats::new();
|
||||
stats.record_tick(Duration::from_micros(500));
|
||||
stats.record_tick(Duration::from_micros(1500));
|
||||
stats.record_tick(Duration::from_micros(800));
|
||||
assert_eq!(stats.max_tick_lag_micros(), 1500);
|
||||
assert_eq!(stats.total_ticks(), 3);
|
||||
assert_eq!(stats.tick_overruns(), 0);
|
||||
assert_eq!(stats.tick_overrun_pct(), 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gauge_counts_overruns_above_threshold() {
|
||||
// 4 ticks: 2 under 10ms, 2 over. Overrun pct = 50%.
|
||||
let stats = TickLagStats::new();
|
||||
stats.record_tick(Duration::from_micros(5_000)); // under
|
||||
stats.record_tick(Duration::from_micros(11_000)); // over
|
||||
stats.record_tick(Duration::from_micros(20_000)); // over
|
||||
stats.record_tick(Duration::from_micros(7_000)); // under
|
||||
assert_eq!(stats.tick_overruns(), 2);
|
||||
assert_eq!(stats.total_ticks(), 4);
|
||||
assert_eq!(stats.max_tick_lag_micros(), 20_000);
|
||||
assert_eq!(stats.tick_overrun_pct(), 50.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gauge_handles_zero_total_ticks_pct() {
|
||||
// No ticks recorded → pct should return 0.0 (not NaN).
|
||||
let stats = TickLagStats::new();
|
||||
assert_eq!(stats.tick_overrun_pct(), 0.0);
|
||||
}
|
||||
}
|
||||
@@ -330,19 +330,3 @@ async fn pstn_sim_synthetic_caller_drives_trunk_reflex_loop() {
|
||||
let _ = close_tx.send(());
|
||||
let _ = tokio::time::timeout(Duration::from_secs(1), engine_handle).await;
|
||||
}
|
||||
|
||||
/// Full end-to-end test using the binary's `MediaThread` + `RegisterTrunk`.
|
||||
///
|
||||
/// This test is currently ignored because `rutster-trunk` integration tests
|
||||
/// cannot depend on the `rutster` binary crate (`rutster` already depends on
|
||||
/// `rutster-trunk`; Cargo disallows circular dev-dependencies). The active
|
||||
/// `pstn_sim_synthetic_caller_drives_trunk_reflex_loop` above covers the FOB
|
||||
// reflex loop + trunk-leg tick directly; this stub marks where the MediaThread
|
||||
/// wiring test belongs once the binary crate is ready to exercise it.
|
||||
#[ignore]
|
||||
#[tokio::test]
|
||||
async fn full_pstn_e2e_through_media_thread_register_trunk() {
|
||||
// TODO: exercise MediaCmd::RegisterTrunk + MediaThread tick loop against a
|
||||
// live TwilioMediaStreamsServer mock once the binary crate exposes a test
|
||||
// harness from the appropriate crate.
|
||||
}
|
||||
|
||||
35
crates/rutster/tests/trunk_sim_e2e.rs
Normal file
35
crates/rutster/tests/trunk_sim_e2e.rs
Normal file
@@ -0,0 +1,35 @@
|
||||
//! Full end-to-end PSTN sim via the binary's `MediaThread` + `RegisterTrunk`.
|
||||
//!
|
||||
//! This test was moved from `crates/rutster-trunk/tests/sim_integ.rs` because
|
||||
//! `rutster-trunk` integration tests cannot depend on the `rutster` binary
|
||||
//! crate (`rutster` already depends on `rutster-trunk`; Cargo disallows
|
||||
//! circular dev-dependencies). The binary crate has access to
|
||||
//! `MediaThread`, `MediaCmd::RegisterTrunk`, and `spawn_tap_engine` — the
|
||||
//! full wiring surface the trunk crate's tests couldn't reach.
|
||||
//!
|
||||
//! The active `pstn_sim_synthetic_caller_drives_trunk_reflex_loop` in
|
||||
//! `crates/rutster-trunk/tests/sim_integ.rs` covers the FOB reflex loop +
|
||||
//! trunk-leg tick directly. This test extends that to exercise the binary's
|
||||
//! `MediaThread` command channel + the `RegisterTrunk` spawn seam end-to-end.
|
||||
|
||||
#[ignore = "requires MediaThread::RegisterTrunk wiring + MockTwilioMediaStreamsServer; not yet implemented in the binary crate's test harness"]
|
||||
#[tokio::test]
|
||||
async fn full_pstn_e2e_through_media_thread_register_trunk() {
|
||||
// TODO: exercise MediaCmd::RegisterTrunk + MediaThread tick loop against a
|
||||
// live TwilioMediaStreamsServer mock. The test should:
|
||||
// 1. Start MockRealtimeBrain.
|
||||
// 2. Spawn a MediaThread (from crates/rutster/src/media_thread.rs).
|
||||
// 3. Construct a MockTwilioMediaStreamsServer that pushes synthetic loud
|
||||
// PcmFrames into a RegisterTrunkInboundChannel.
|
||||
// 4. Send MediaCmd::RegisterTrunk to the MediaThread.
|
||||
// 5. Drive the 20ms tick loop for ~500ms.
|
||||
// 6. Assert barge-in fired (ReflexMetrics.barge_in_count >= 1).
|
||||
// 7. Assert brain reply observed on the outbound mpsc.
|
||||
// 8. Assert idle timeout closes the session (ChannelState::Closed).
|
||||
//
|
||||
// This requires the binary crate's test harness to expose a way to
|
||||
// construct a TrunkSession's pipe stack outside the normal axum route
|
||||
// handler (or to drive the route handler in-process via axum's test
|
||||
// `oneshot` pattern). Either path is feasible; pick the one that
|
||||
// matches the existing `barge_in_integration.rs` harness pattern.
|
||||
}
|
||||
Reference in New Issue
Block a user