# Bug Audit — 2026-05-02

Scope: `net/crates/net/` (477 Rust files). Four parallel-hunter passes; every
finding below is numbered sequentially `#1`…`#74` across all passes for easy
cross-reference.

Severity ladder:
- **Critical** — panic / corruption / security failure, no caller-visible signal
- **High** — incorrect behavior on a realistic path
- **Medium** — edge case or contention-only failure
- **Low** — latent / footgun / by-design / astronomical

| Pass | Scope | High | Medium | Low |
|------|-------|------|--------|-----|
| 1 | cortex / channel / redex / compute / traversal / contested / continuity / sdk / bindings / state / subnet / behavior / identity / netdb / config | 10 (#1–#10) | 11 (#11–#21) | 13 (#22–#34) |
| 2 | top-level `src/` (bus, shard, consumer), `adapter/jetstream`, `adapter/redis*`, `adapter/dedup_state`, `adapter/net/{pool,crypto,protocol}`, `ffi/` | 2 (#35–#36) | 5 (#37–#41) | 2 (#42–#43) |
| 3 | `bus.rs` (~120K), `event.rs` / `config.rs` / `error.rs` / `timestamp.rs`, full `adapter/` tree, `consumer/`, `shard/`, `ffi/` | 4 (#44–#47) | 11 (#48–#58) | 8 (#59–#66) |
| 4 | `consumer/`, `adapter/`, `ffi/` | 1 (#67) | 4 (#68–#71) | 3 (#72–#74) |

---

## Status — bugfixes-8 branch (as of session end)

All 74 audit items are addressed on branch `bugfixes-8` (67 commits).
Status legend used in the per-bug sections below:

| Marker | Meaning |
|--------|---------|
| ✅ | Fixed with code change + regression test where unit-testable |
| 📝 | Fixed with documentation only (trade-off accepted, future-readers warned) |
| ✓ | Already correct in the current tree (audit was written against an older revision) |
| ❌ | Investigated and rejected as a non-bug (rationale in commit) |
| ⏭ | Skipped — semantics-change risk or API redesign too large for this batch |

| Bug | Status | Notes |
|-----|--------|-------|
| #1 | ✅ | Lost events on partial replay failure |
| #2 | ✅ | Silent eviction during tail backfill |
| #3 | ✅ | Index task exits permanently after Lagged |
| #4 | ✅ | Node binding u32→u8 truncation |
| #5 | ✅ | NetBuilder::identity silent override |
| #6 | ✅ | Python bindings GIL release |
| #7 | ✅ | Lossy as_f64 numeric ordering |
| #8 | ✅ | Snapshot retention high-water mark |
| #9 | ✅ | Subnet config fallible try_* constructors |
| #10 | ✅ | SdkError::Ingestion structured variants |
| #11 | ✅ | RedexFile::next_seq holds state lock |
| #12 | 📝 | Disk I/O under parking_lot Mutex (latency, not correctness) |
| #13 | 📝 | ingest_typed single-origin invariant documented |
| #14 | ✅ | start_snapshot drops DashMap guard before I/O |
| #15 | ✅ | window_failures sort+dedup, deterministic order |
| #16 | ✅ | SubscribeOpts intervals clamped to non-zero |
| #17 | ✅ | record_execution skips count for non-rate-limited rules |
| #18 | ✅ | NetAdapterConfig upper bounds + heartbeat floor |
| #19 | 📝 | StreamWindow consumer-side clamp (already enforced; doc'd) |
| #20 | ✅ | Partition next_id saturates at u64::MAX |
| #21 | ✅ | Rule timestamps use saturating u64::try_from |
| #22 | ✅ | PermissionToken::try_issue rejects ttl=0 |
| #23 | ✅ | Age-retention boundary inclusive (breaking change) |
| #24 | ✅ | new_heap rejects RedexFlags::INLINE |
| #25 | ✅ | Membership decoder rejects trailing bytes (all paths) |
| #26 | ✅ | PublishReport::all_delivered vacuously true |
| #27 | ✅ | append_batch returns Option<u64> (breaking change) |
| #28 | ❌ | condition_count for Not — consistent with And/Or, not a bug |
| #29 | 📝 | OriginStamp threat model documented |
| #30 | ✅ | Cortex Stop policy emits final changes_tx notify |
| #31 | ✅ | drain_pending uses saturating_add |
| #32 | 📝 | on_node_recovery overlapping-partition documented |
| #33 | ✅ | Removed unused `kind: String` from Node group structs |
| #34 | ✅ | Saturating doubling in stream backoff |
| #35 | ✅ | activate_shard doesn't double-count num_shards |
| #36 | ✅ | PacketBuilder rejects events.len() > MAX |
| #37 | ✅ | Anti-replay bitmap reset emits structured warn |
| #38 | ✅ | pushes_since_drain_start uses Acquire |
| #39 | ✅ | JetStream rejects shard_id outside u16 range |
| #40 | ✅ | dedup_state propagates fsync errors |
| #41 | ✅ | Redis is_healthy enforces command_timeout |
| #42 | ✅ | RedisStreamDedup capacity clamped at MAX_CAPACITY |
| #43 | ❌ | net_num_shards 0-on-null sentinel is unambiguous (validated >0) |
| #44 | ✅ | Activation-failure rollback uses graceful teardown |
| #45 | ✅ | Same as #44 (paired fix) |
| #46 | ✅ | Final drain sweep requires two consecutive zero passes |
| #47 | ❌ | Redis MULTI/EXEC — redis-rs 1.2 already checks ServerError |
| #48 | ✅ | Lossy shutdown surfaces via stats flag |
| #49 | 📝 | Drop has no partial-destruction UB (clarifying doc) |
| #50 | 📝 | poll() topology-change visibility documented |
| #51 | ✓ | Already correct — events_dispatched IS bumped on stranded-flush |
| #52 | ✅ | dispatch_batch drops tagged with structured `reason` |
| #53 | ✅ | JetStream re-serialize errors surfaced as Fatal |
| #54 | ✅ | JetStream poll bails after consecutive NotFounds |
| #55 | ❌ | `&mut self` precludes the audit's "concurrent" race |
| #56 | ✅ | Lemire unbiased shard selection |
| #57 | ✅ | per_shard_limit cap surfaced in response |
| #58 | ✅ | TimestampGenerator captures construction-time TSC baseline |
| #59 | ✅ | in_flight_ingests widened to AtomicU64 |
| #60 | ✅ | drain_finalize_ready store promoted to SeqCst |
| #61 | ✅ | TimestampGenerator panics preemptively at u64::MAX |
| #62 | 📝 | high_throughput max_shards cap documented as intentional |
| #63 | ✅ | Cursor::encode panics on schema break, no silent rewind |
| #64 | ✅ | PollMerger surfaces stalled_shards in response |
| #65 | ✅ | velocity_samples capped by count |
| #66 | 📝 | add_events(empty) timeout-flush side effect documented |
| #67 | ✅ | PollMerger uses live shard id set, not 0..num_shards |
| #68 | ✅ | dedup_state writable handle for cross-platform fsync |
| #69 | ✅ | JetStream cursor seq+1 saturates at u64::MAX |
| #70 | 📝 | FFI runtime-context restriction documented |
| #71 | ✅ | Adapter limit+1 saturates at usize::MAX |
| #72 | 📝 | last_scaling RwLock implicit serialization documented |
| #73 | ✅ | RedisStreamDedup docs accurately describe FIFO |
| #74 | ✅ | dispatch_batch retries jittered + exponential backoff |

**Tally:** 51 ✅ + 13 📝 + 1 ✓ + 4 ❌ + 0 ⏭ = 69 addressed in code/docs, 5 categorized
as already-correct/non-bug (#43, #47, #51, #55, #28). Net: 70 commits, +4571/-407
lines, 1427 lib tests passing (was 1391 baseline; 36 new regression tests).

**Breaking changes (#23, #27)** were applied on caller request:
- `RedexFile::append_batch` / `append_batch_ordered` /
  `TypedRedexFile::append_batch` / `OrderedAppender::append_batch` now return
  `Result<Option<u64>, RedexError>` (was `Result<u64, RedexError>`); empty input
  yields `Ok(None)`.
- Age-retention boundary changed from `>` to `>=` (inclusive); an entry exactly
  `max_age_ns` old is now retained instead of dropped. Existing pinned tests
  updated.

**Branch:** `bugfixes-8` (off `master`).

---

## Pass 1 — Subsystem audit

### Critical / High

#### #1 — Lost events on partial replay failure
**File:** `src/adapter/net/compute/migration_target.rs:400-426`
```rust
while let Some(event) = state.pending_events.remove(&next_seq) {
    to_replay.push(event);   // events removed BEFORE delivery
    next_seq += 1;
}
...
for event in &to_replay {
    self.daemon_registry.deliver(...)?;   // early-return on failure
}
if let Some(last) = to_replay.last() { state.replayed_through = last.link.sequence; }
```
**Failure:** If `deliver()` fails on event N of M during `replay_events` /
`buffer_event`, the `?` returns early. All events in `to_replay` were already
removed from `pending_events`, but `replayed_through` is never updated. On retry,
those events are gone. Daemon state is permanently corrupted. **Severity: high.**

#### #2 — Silent eviction during tail backfill (no `Lagged`)
**File:** `src/adapter/net/redex/file.rs:820–892`
```rust
let start = state.index.partition_point(|e| e.seq < from_seq);
let backfill_count = state.index.len() - start;
```
**Failure:** A subscriber requesting `from_seq=0` after retention has dropped the
head gets only the currently retained tail with no `Lagged` signal —
`partition_point` returns the first retained entry with `seq >= from_seq`, so the
"history was lost" condition is never detected. Documented behavior on
`read_range` but `tail`'s doc explicitly promises `Lagged` on backfill overflow
only. Should also fire on seq-below-retained. **Severity: high (silent data loss
for resume-from-snapshot consumers).**

#### #3 — Index task exits permanently after `Lagged`
**File:** `src/adapter/net/redex/index.rs:131–139`
```rust
Some(Err(e)) => { tracing::debug!(error = %e, "RedexIndex tail error; continuing"); }
```
**Failure:** `RedexError::Lagged` is logged at `debug` and the loop continues.
After `Lagged` the underlying watcher is dropped from the file, so the next
`tail.next()` yields `None` and the index task exits permanently with the index
permanently incomplete. Keys missing post-`Lagged` stay missing forever.
**Severity: high (silent index corruption).**

#### #4 — Node binding `u32 → u8` truncation on member index
**File:** `bindings/node/src/groups.rs:664, 672`
```rust
pub fn member_role(&self, index: u32) -> Option<String> {
    self.inner.member_role(index as u8)...
pub fn synced_through(&self, index: u32) -> Option<BigInt> {
    self.inner.synced_through(index as u8)...
```
**Failure:** JS caller invokes `group.memberRole(256)` expecting `null`
(out-of-range). `as u8` wraps to `0`, so it returns the role for member 0. Same
for `syncedThrough` — caller polling sync progress for index 257 silently gets
the lag of member 1, makes an incorrect failover decision. The crate already has
a `count_u8()` helper for this — these two calls bypass it. **Severity: high.**

#### #5 — `NetBuilder::identity()` silently overrides `entity_keypair`
**File:** `sdk/src/config.rs:160-165`
```rust
if let (Some(id), AdapterConfig::Net(ref mut net_cfg)) =
    (self.identity.as_ref(), &mut adapter)
{
    net_cfg.entity_keypair = Some((**id.keypair()).clone());
}
```
**Failure:** A user calls
`.mesh(NetAdapterConfig::initiator(...).with_entity_keypair(kp_a)).identity(id_b)`
(in any order — `mesh()` doesn't take `&mut`). Both inputs are caller-supplied
keypairs; no error is raised, no precedence is documented, `id_b` silently wins.
The two identities are loadbearing for `node_id` / `entity_id` — picking the
wrong one routes the node under the wrong identity. **Severity: high — silent
contract violation on cryptographic identity selection.**

#### #6 — Python bindings hold GIL across blocking compute ops
**File:** `bindings/python/src/groups.rs:257-272, 354-366, 464-476`
```rust
fn scale_to(&self, n: u8) -> PyResult<()> { self.inner.scale_to(n).map_err(group_err) }
fn on_node_failure(&self, failed_node_id: u64) -> PyResult<Vec<u8>> { ... }
fn sync_standbys(&self) -> PyResult<u64> { ... }
fn promote(&self) -> PyResult<u32> { ... }
```
**Failure:** Module docstring promises sync because "PyO3's `Python::attach` is
safe from any tokio worker." But `scale_to`, `on_node_failure`, `promote`,
`sync_standbys` invoke scheduler placement, registry operations, snapshot
serialization — all of which can block on internal locks for ms–s. None wrap
with `py.allow_threads(...)`. A Python program with a watchdog thread or
asyncio loop is frozen for the duration. **Severity: high.**

#### #7 — Lossy `as_f64` for all numeric ordering in rules
**File:** `src/adapter/net/behavior/rules.rs:118-120`
```rust
let a = a.as_f64()?;
let b = b.as_f64()?;
a.partial_cmp(&b)
```
**Failure:** Rule conditions `Lt`/`Gt`/`Le`/`Ge` against `serde_json` u64/i64
fields >2^53 (e.g., timestamps in ns, byte counts) silently truncate. Two
adjacent values like `9_007_199_254_740_992` and `9_007_199_254_740_993` compare
as `Equal`; a `Gt` guard on a quota or sequence number returns `false` and the
rule fails to fire. NaN / Infinity (legal under arbitrary_precision) yield
`None`, silently masking the bug. **Severity: high.**

#### #8 — Snapshot-store retention drops high-water mark on `remove`
**File:** `src/adapter/net/state/snapshot.rs:592-594`
```rust
pub fn remove(&self, entity_id: &EntityId) -> Option<StateSnapshot> {
    self.snapshots.remove(entity_id.as_bytes()).map(|(_, s)| s)
}
```
**Failure:** A stale producer racing a retention-`remove` re-stores an older
snapshot at lower `through_seq` than the just-removed one — concrete ABA / state
rollback. The regression test
`cr17_remove_does_not_carry_forward_through_seq_high_water_mark` pins this as a
documented limitation; multi-producer paths can hit it without the locking the
test enforces. **Severity: high.**

#### #9 — Untrusted-input panics in subnet config
**Files:**
- `src/adapter/net/subnet/assignment.rs:73-77, 124-128`
- `src/adapter/net/subnet/id.rs:43-48`
```rust
assert!(rule.level < 4, ...);          // SubnetPolicy::add_rule
assert!(level_value != 0, ...);        // SubnetRule::map
assert!(levels.len() <= MAX_DEPTH ...) // SubnetId::new
```
**Failure:** Subnet policies typically come from config / FFI / JSON. A config
file with `level: 4` or `level_value: 0` or `>4 entries` panics the daemon
loader. No fallible constructor exists. **Severity: high.**

#### #10 — `SdkError::Ingestion(String)` flattens structured `IngestionError`
**File:** `sdk/src/error.rs:85-89`
```rust
impl From<net::error::IngestionError> for SdkError {
    fn from(e: net::error::IngestionError) -> Self {
        SdkError::Ingestion(e.to_string())
    }
}
```
**Failure:** `IngestionError` has structured variants (`Backpressure`, `Sampled`,
`Unrouted`, shutdown). The SDK already exports `SdkError::Backpressure` and
`SdkError::Shutdown` with rich semantics for stream-side errors, but
ingestion-side mappings stringify everything. A user calling `node.emit(...)` on
a backpressured bus gets `SdkError::Ingestion("backpressure: ring buffer full")`
and must string-match to distinguish from `Sampled` (event deliberately dropped,
retry pointless) or `Unrouted` (topology transient). **Severity: high — looks
ergonomic but forces wrong remediation.**

### Medium

#### #11 — Observable seq rollback via `next_seq()`
**File:** `src/adapter/net/redex/file.rs:393, 440, 509, 608, 648, 706`
```rust
let seq = self.inner.next_seq.fetch_add(1, Ordering::AcqRel);
// ... disk write ...
self.inner.next_seq.fetch_sub(1, Ordering::AcqRel); // on disk failure
```
A reader calling `RedexFile::next_seq()` (line 358 — `Ordering::Acquire` outside
the state lock) can observe the temporarily incremented value before the
rollback completes on disk failure. Window where external observers (metrics,
snapshot logic, an `IndexStart::FromSeq(next_seq())` re-tail) believe a seq
exists that was never durably appended.

#### #12 — Disk I/O held under `parking_lot::Mutex`
**Files:** `src/adapter/net/redex/file.rs:475–565, 678–758, :987`
`append_batch` calls `disk.append_entries_at(...)` and `sweep_retention` calls
`disk.compact_to(...)` while holding `state.lock()` (a non-yielding parking_lot
Mutex). All concurrent appenders, `tail` registrations, `read_range`, `len`,
`is_empty`, and `close` block on the same lock during disk I/O. Latency /
starvation, not correctness.

#### #13 — `ingest_typed` post-`open` race vs foreign same-origin writers
**Files:**
- `src/adapter/net/cortex/memories/adapter.rs:116-117`
- `src/adapter/net/cortex/tasks/adapter.rs:119-120, 340-341`
```rust
let file = redex.open_file(&name, redex_config)?;
let next_seq = file.next_seq();
```
Concurrent ingests on this channel can append events between
`CortexAdapter::open` (which spawned the fold task with `file.tail(start_seq)`)
and `next_seq()` here. `wait_for_seq(next_seq - 1)` then waits for a moving
target — there is a window where post-read appends with our `origin_hash` (e.g.
crashed sibling adapter resumed) can land *after* `wait_for_seq` returns yet
*before* the user's first `ingest_typed`, and collisions can occur. Doc-comment
on `MemoriesAdapter::open` overstates the guarantee.

#### #14 — DashMap entry guard held across registry I/O
**File:** `src/adapter/net/compute/migration_source.rs:73-101`
```rust
let entry = match self.migrations.entry(daemon_origin) { ... Vacant(entry) => entry };
if !self.daemon_registry.contains(daemon_origin) { return Err(...); }
let snapshot = self.daemon_registry.snapshot(daemon_origin)?...;
entry.insert(...);
```
The DashMap shard write guard from `entry()` is held across
`daemon_registry.snapshot()`, which itself locks an inner `Mutex<DaemonHost>`.
Combined with another caller acquiring the daemon-host mutex first then touching
this dashmap shard, deadlock risk. Also, `take_snapshot` may run user code
(`MeshDaemon::snapshot()`) under the held shard guard, blocking concurrent
migrations on co-hashed origins.

#### #15 — Non-deterministic verdict ordering
**File:** `src/adapter/net/contested/correlation.rs:158-164`
```rust
let window_failures: Vec<u64> = self.recent_failures.iter().map(|e| e.node_id)
    .collect::<std::collections::HashSet<_>>().into_iter().collect();
```
`HashSet<u64>` iteration order is randomized per process. `window_failures` is
the `failed_nodes` list passed verbatim into `PartitionRecord::other_side`
(`partition.rs:160`). Two nodes with identical inputs assemble different
`other_side` orderings, causing divergent serialization, divergent reconcile
ordering, and replays that can't be cross-validated.

#### #16 — Stream tight-loop spin
**File:** `sdk/src/stream.rs:158, 174-185`
```rust
this.current_interval = (this.current_interval * 2).min(this.opts.max_backoff);
let mut sleep = Box::pin(tokio::time::sleep(this.current_interval));
match sleep.as_mut().poll(cx) {
    Poll::Ready(()) => { cx.waker().wake_by_ref(); Poll::Pending }
```
`SubscribeOpts::default().poll_interval(Duration::ZERO).max_backoff(Duration::ZERO)`
with no events resolves the zero-duration sleep instantly, immediately re-wakes,
and hammers `bus.poll()` at 100% CPU. No validation rejects zero.

#### #17 — `record_execution` window-reset across rule reload
**File:** `src/adapter/net/behavior/rules.rs:1184-1200`
```rust
state.last_execution = Some(now);
state.execution_count += 1;
...
if now.duration_since(state.window_start) >= window_duration {
    state.window_start = now;
    state.execution_count = 1;
}
```
`record_execution` is called for rules whose *current* `rate_limit` is `None`
(`if let Some(ref limit)` guards only the reset, not the increment). A rule that
toggled rate-limited → unlimited carries its old `execution_count` forever; on
toggle back with the same window, it is immediately blocked because
`count >= max`. Concrete on hot-reload of a rule's limits.

#### #18 — `NetAdapterConfig::validate` accepts pathological values
**File:** `src/adapter/net/config.rs:230-249`
Only zero/ordering checks; `packet_pool_size = usize::MAX`,
`handshake_retries = usize::MAX`, `heartbeat_interval = 1ns`,
`session_timeout = 2ns` all pass. A 2-byte typo
(`with_pool_size(1_000_000_000)`) walks past validation and OOMs at first
allocation.

#### #19 — `StreamWindow.decode` unbounded `total_consumed`
**File:** `src/adapter/net/subprotocol/stream_window.rs`
Codec accepts any `total_consumed: u64`. Doc-comment formula
`tx_credit_remaining = tx_window - (tx_bytes_sent - total_consumed)` underflows
/ wraps if peer sends `total_consumed > tx_bytes_sent`. The codec's promise of
"self-healing" gives a false sense of safety if the consumer doesn't clamp —
verify in caller.

#### #20 — Partition `next_id` reuses ID 0 on `u64::MAX` overflow
**File:** `src/adapter/net/contested/partition.rs:153-154`
```rust
let id = self.next_id;
self.next_id += 1;
```
At `u64::MAX`, panics in debug, wraps in release reusing ID 0 — `confirm`/`get`
could operate on the wrong record. Astronomical in practice.

#### #21 — `Rule::new` lossily casts `u128` millis to `u64`
**File:** `src/adapter/net/behavior/rules.rs:579, RuleSet::new:1237`
```rust
.as_millis() as u64;
```
Silent truncation past year ~584,554,051 — not operational. `as` cast on
`unwrap_or_default()`'d `Duration::ZERO` (clock pre-epoch / VM clock-skew)
silently produces `created_at = 0`, which then collides across rules and breaks
any "newer-rule-wins" logic.

### Low

#### #22 — `ttl_seconds = 0` token mints expired
**Files:**
- `bindings/node/src/identity.rs:222`
- `bindings/python/src/identity.rs:221`
- `sdk/src/identity.rs:139`
Caller mints a token with TTL 0; `not_after == not_before` so it's expired
before it leaves the call. Token verifies signature-OK but every check fails at
the receiver. No diagnostic.

#### #23 — Age-retention off-by-one
**File:** `src/adapter/net/redex/retention.rs:75`
```rust
let cutoff = now_ns.saturating_sub(max_age_ns);
...
if ts > cutoff { break; }
age_drop += 1;
```
An entry with `ts == cutoff` is dropped (the condition is `ts > cutoff`); should
be `>=` to retain the boundary.

#### #24 — `new_heap` accepts `RedexFlags::INLINE`
**File:** `src/adapter/net/redex/entry.rs:71`
```rust
flags_and_checksum: (flags & FLAG_MASK) | (checksum & CHECKSUM_MASK),
```
A caller passing `RedexFlags::INLINE` to `new_heap` produces an entry whose
`is_inline()` returns true but whose `payload_offset`/`payload_len` are real
heap fields rather than payload bytes. `materialize` would then invoke
`inline_payload()` and reinterpret the offset/length as data — silent
corruption. Currently not exercised (callers all pass `flags=0`), but unguarded
footgun.

#### #25 — Channel decoder accepts trailing bytes on UNSUBSCRIBE/ACK
**File:** `src/adapter/net/channel/membership.rs:218-219`
For both `MSG_UNSUBSCRIBE` and `MSG_ACK` the decoder returns `Ok` without
checking `cur.remaining() == 0`. Pre-existing tests enforce strict-trailer
rejection only on the SUBSCRIBE path (regression test at line 309). A malformed
peer can append arbitrary bytes after a valid `Unsubscribe` or `Ack`.
Defense-in-depth, not exploitable.

#### #26 — `Publisher::all_delivered()` returns false for empty roster
**File:** `src/adapter/net/channel/publisher.rs:130`
```rust
self.attempted > 0 && self.delivered == self.attempted && self.errors.is_empty()
```
A publish to a channel with zero subscribers is `is_empty() == true` AND
`all_delivered() == false`. Callers branching on `all_delivered()` for "all
good" treat empty rosters as failure. Common bug after `remove_peer` GC.

#### #27 — `append_batch` empty-input returns plausible-looking seq
**File:** `src/adapter/net/redex/file.rs:475–479`
```rust
if payloads.is_empty() {
    return Ok(self.inner.next_seq.load(Ordering::Acquire));
}
```
Returns the next-to-assign seq for an empty batch; caller cannot distinguish
"wrote zero entries" from "wrote one entry with that seq." Footgun.

#### #28 — `condition_count` off-by-one for `Not`
**File:** `src/adapter/net/behavior/rules.rs:294`
```rust
ConditionExpr::Not(c) => c.condition_count(),
```
The `Not` node itself is dropped from the count — a `Not(big_or)` evades a
per-rule condition cap by 1. Concrete only if a caller enforces a hard cap with
this method.

#### #29 — `OriginStamp` has no per-packet binding
**File:** `src/adapter/net/identity/origin.rs:42-46`
`origin_hash` is written verbatim into headers; tamper resistance is delegated
to AEAD. Any peer with the session key can mint packets claiming arbitrary
`origin_hash` values within the channel — by-design ("zero per-packet crypto")
but worth flagging since the audit explicitly looked at origin spoofability.

#### #30 — `Stop` policy halts without final `changes_tx` notify
**File:** `src/adapter/net/cortex/adapter.rs:516-525`
```rust
match policy {
    FoldErrorPolicy::Stop if !recoverable_decode => true,
    ...
}
```
When `Stop` halts on a non-recoverable error, the loop breaks without emitting
any `changes_tx.send` for that seq. Subscribers using `changes_with_lag()` get
no terminal signal — they only learn the task halted by separately polling
`is_running()`. Documented but UX rough.

#### #31 — Migration `next_seq` overflow
**File:** `src/adapter/net/compute/migration_target.rs:398`
```rust
let mut next_seq = state.replayed_through + 1;
```
If `replayed_through == u64::MAX`, `+1` panics in debug (wraps in release,
immediately mis-keys lookups). Astronomical.

#### #32 — `on_node_recovery` does not break after first matching partition
**File:** `src/adapter/net/contested/partition.rs:185-213`
A node that simultaneously appears in `other_side` of two overlapping partition
records is added to BOTH `reappeared` lists, and BOTH may transition to `Healed`
from one recovery signal. If healing actions fire per partition, the recovery
is double-counted. Overlapping partitions are atypical but not prevented.

#### #33 — Node-binding groups carry an unused `kind: String` field
**File:** `bindings/node/src/groups.rs:399, 488, 576`
The `kind` is captured at spawn but never read — `Arc<SdkReplicaGroup>` holds
its own copy. Trivial waste, but a maintenance trap: anyone who later "uses"
`self.kind` will read a snapshot that can drift from the SDK-side kind.

#### #34 — Stream backoff overflow on absurd `poll_interval`
**File:** `sdk/src/stream.rs:158`
`this.current_interval * 2` panics on `Duration` overflow. Not realistic for
sane inputs, but the SDK accepts any `Duration` from `poll_interval()`.

### Pass 1 — top 3 to fix first
1. **#1** `migration_target.rs` lost-events on `?` early-return — silent
   permanent desync.
2. **#2** `redex/file.rs` silent backfill below retention — silent data loss
   for resume-from-snapshot consumers.
3. **#3** `redex/index.rs` permanent index halt after `Lagged` — silent index
   corruption, keys never reappear.

All three cause silent data divergence with no caller-visible signal.

---

## Pass 2 — Core / Adapter / FFI

Second audit pass covering areas the original sweep didn't focus on: top-level
`src/` (bus, shard, consumer), `adapter/jetstream.rs`, `adapter/redis*.rs`,
`adapter/dedup_state.rs`, `adapter/net/pool.rs` / `crypto.rs` / `protocol.rs`,
and `ffi/`. Four parallel hunters; high-severity claims spot-checked against
source.

### High

#### #35 — `ShardManager::activate_shard` double-counts on idempotent calls
**File:** `src/shard/mod.rs:792-800` (with `src/shard/mapper.rs:760-806`)
```rust
pub fn activate_shard(&self, shard_id: u16) -> Result<(), ScalingError> {
    let mapper = ...;
    mapper.activate(shard_id)?;          // returns Ok(()) early when already Active
    self.num_shards
        .fetch_add(1, std::sync::atomic::Ordering::Release);  // unconditional
    Ok(())
}
```
**Failure:** Docstring at line 791 promises "Idempotent: calling on an
already-`Active` shard is `Ok(())`." The mapper honours this — when state is
already `Active` it returns `Ok(())` at `mapper.rs:766` without bumping its own
`active_count`. But `ShardManager` then unconditionally `fetch_add(1)`s
`num_shards`. After repeated activate calls, `num_shards` exceeds the mapper's
`active_count` and the actual shard count, breaking modulo-based shard
selection (`select_shard`) and producing stale routing decisions.
**Severity: high.**

#### #36 — Silent `event_count` truncation in packet builder
**File:** `src/adapter/net/pool.rs:144, 208`
```rust
let header = NetHeader::new(
    self.session_id, stream_id, sequence, [0u8; NONCE_SIZE],
    self.payload.len() as u16,
    events.len() as u16,            // silent wrap if > 65_535
    flags,
)
```
**Failure:** Protocol-level cap is `MAX_EVENTS_PER_PACKET = 2027`
(`protocol.rs:443`), but the packet builder doesn't enforce it before the cast.
A caller passing >65 535 events wraps to `len % 65536`. The receiver's
`validate()` (`protocol.rs:451`) accepts any `event_count <= 2027`, so a
wrapped value like 34 464 fails validation noisily — but a wrapped value below
2027 (e.g. caller passed 67 562 events) passes validation, and the receiver
mis-parses the payload because the actual event_count on the wire mismatches
what's encoded. Add `debug_assert!(events.len() <= MAX_EVENTS_PER_PACKET as
usize)` plus a runtime guard returning an error. **Severity: high (silent
corruption).**

### Medium

#### #37 — Anti-replay window cleared on large legitimate jumps
**File:** `src/adapter/net/crypto.rs:417, 469-476`
```rust
fn is_valid(&self, received: u64) -> bool {
    if received >= self.rx_counter {
        received.saturating_sub(self.rx_counter) <= Self::MAX_FORWARD  // 65_536
    } ...
}
fn shift_bitmap_up(&mut self, shift: u64) {
    ...
    if shift >= (Self::BITMAP_WORDS as u64) * 64 {  // 1024
        self.bitmap = [0; Self::BITMAP_WORDS];      // wipes prior tracking
        return;
    }
    ...
}
```
**Failure:** `MAX_FORWARD = 65 536` allows accepting future counters far beyond
`WINDOW_SIZE = 1024`. When a legitimate packet jumps the counter forward by
more than 1024, `shift_bitmap_up` zeros the bitmap, leaving the prior ~64 000
counters unprotected. Any captured packets in that window can be replayed
undetected. Either tie `MAX_FORWARD` to `WINDOW_SIZE`, or document the tradeoff
as accepted. **Severity: medium (replay window of seen packets).**

#### #38 — `pushes_since_drain_start` mismatched atomic ordering
**File:** `src/shard/mapper.rs:213-215` (read), `:203-204` (writer)
The reader getter uses `Ordering::Relaxed`; the writer (`set_draining(true)`)
uses `SeqCst` for the reset. On weakly-ordered hardware (ARM),
`finalize_draining` can read a stale counter and falsely conclude the drain
has flushed while producers are still pushing. Reader should be `Acquire` to
pair with the release of the reset. **Severity: medium
(architecture-specific).**

#### #39 — Silent `shard_id` truncation in JetStream deserialize
**File:** `src/adapter/jetstream.rs:110`
```rust
let shard_id = value.get("s").and_then(|v| v.as_u64()).unwrap_or(0) as u16;
```
A stored `shard_id` of 100 000 silently becomes 34 464 (`100_000 % 65_536`).
No warning. Mis-routes events under the wrong shard at consume time. Either
reject the message or log a deser error; better, store the value as the same
width it was written with.

#### #40 — `dedup_state` silently swallows fsync failures
**File:** `src/adapter/dedup_state.rs:221-235`
```rust
let _ = f.sync_all();
```
Disk-full or I/O errors during `sync_all` are discarded. The producer-nonce
file may be partially persisted; the next restart resumes with a stale nonce,
defeating the cross-restart dedup guarantee that's the entire point of the
file. Propagate the error (or at minimum log at warn).

#### #41 — `Redis::is_healthy` PING has no enforced timeout
**File:** `src/adapter/redis.rs:465-478`
```rust
async fn is_healthy(&self) -> bool {
    if !self.initialized.load(Ordering::Acquire) { return false; }
    if let Ok(mut conn) = self.get_conn().await {
        redis::cmd("PING").query_async::<String>(&mut conn).await.is_ok()
    } else { false }
}
```
Health entirely depends on `ConnectionManager`'s configured request timeout. If
that's unset or large, a network partition leaves the health check blocked
indefinitely, making the adapter appear "pending" rather than "unhealthy". Wrap
in `tokio::time::timeout(self.config.request_timeout, …)`.

### Low

#### #42 — `RedisStreamDedup::new` accepts unbounded capacity
**File:** `src/adapter/net/../redis_dedup.rs:103-109`
`capacity.max(1)` has no upper clamp. A misconfigured `usize::MAX`
pre-allocates `VecDeque`/`HashSet` and OOMs on construction. Cheap to clamp.

#### #43 — `net_num_shards` returns 0 on null handle
**File:** `src/ffi/mod.rs:1069-1078`
Returns `0` for null handle — also a valid (if unusual) shard count.
Inconsistent with sibling FFI functions that return negative `NetError` codes;
C callers can't distinguish null handle from a 0-shard bus. Return
`NetError::InvalidHandle.into()`.

### Pass 2 — claims investigated and rejected
Listed so future audits don't re-litigate them.

- **Off-by-one null-terminator in `net_poll`** (`ffi/mod.rs:806/817`) — the
  bound check `response_json.len() + 1 > buffer_len` is correct: when
  `len + 1 == buffer_len`, the write at `out_buffer.add(len)` lands at the last
  valid byte.
- **`OnceCell` caches stream-init errors forever** (`adapter/jetstream.rs:131`)
  — the docstring at line 129-130 and `OnceCell::get_or_try_init` semantics
  confirm errors leave the cell uninitialized; subsequent calls retry.
- **`shift_bitmap_up` UB at `bit_shift == 64`** (`adapter/net/crypto.rs:495`) —
  guarded by the `if bit_shift == 0` branch at line 479; the right-shift is
  always in `[1, 63]`.
- **`session_prefix_from_id` XOR collisions**
  (`adapter/net/crypto.rs:511-515`) — irrelevant to security: keys are
  per-session-direction, so nonce uniqueness only needs `(key, nonce)`
  uniqueness within one key.
- **JetStream batch ordering across concurrent `on_batch`**
  (`jetstream.rs:262-277`) — the per-shard FIFO contract is enforced upstream
  by the shard manager serializing per-shard dispatch; this layer only sees
  one batch at a time per shard.

---

## Pass 3 — bus / event / config / adapter / consumer / shard / ffi

Six parallel auditors over `bus.rs` (~120K), `event.rs` / `config.rs` /
`error.rs` / `timestamp.rs`, the full `adapter/` tree, `consumer/`, `shard/`,
and `ffi/`. New findings below; duplicates of items already in pass 2
(`activate_shard` double-count, JetStream `shard_id as u16` truncation,
`RedisStreamDedup::new` unbounded capacity, `net_num_shards` null-handle
ambiguity) are not re-listed.

### High

#### #44 — Activation-failure abort drops drain-worker scratch buffer
**File:** `src/bus.rs:420-463`
If `activate_shard()` fails after the ingest + drain workers have been spawned,
both tasks are `.abort()`-ed. Any events the drain worker has already pulled
from the ring buffer into its `scratch: Vec<InternalEvent>` are never sent to
the adapter. The comment at lines 441-446 acknowledges the scenario without
addressing the loss path. Drain the scratch synchronously on the failure path
(or post the events back to a recovery channel) before aborting.
**Severity: high (silent loss on partial activation failure).**

#### #45 — Batch worker abort drops in-memory `current_batch`
**File:** `src/bus.rs:1554-1643`
The batch worker holds an in-memory `BatchWorker.current_batch` between flush
ticks. On `.abort()` (used during failed activation, scale-down, and shutdown
timeouts), `BatchWorker::flush()` is never called, so any events already pulled
from the channel into the batch are lost. Channel-resident events survive (the
close signal triggers a final flush), but the worker's own buffered window does
not. Replace abort with a graceful `flush()` on shutdown / activation-failure
paths. **Severity: high (silent loss of one batch worth of events per worker).**

#### #46 — Shutdown drain race past `in_flight_ingests`
**File:** `src/bus.rs:1685-1735` (with shutdown at `:1065-1083`, gate at
`:1088-1094`)
The drain worker's final sweep gates on `drain_finalize_ready`, which is set
only after `in_flight_ingests` reaches zero. A producer that has already
incremented `in_flight_ingests` but stalls (lock contention, scheduler) before
reaching the shard-lock body can push *after* the in-flight counter is
observed at zero — by an unlucky drop ordering of the guard — stranding events
past the final sweep. The SeqCst pair orders the `push` w.r.t. the final read,
but does not prevent the producer's push from being delayed past the final
sweep iteration. Either widen the in-flight window to span lock acquisition,
or run the final sweep loop until two consecutive zero-event passes.
**Severity: high (silent loss on shutdown under contention).**

#### #47 — Redis MULTI/EXEC partial failures swallowed
**File:** `src/adapter/redis.rs:399-405`
```rust
let fut = pipe.query_async::<()>(&mut conn);
```
The unit-typed query discards the EXEC reply array. In Redis, individual XADD
commands inside a successful `MULTI/EXEC` can fail (e.g. wrong type, OOM) while
the transaction itself returns OK. The adapter currently treats this as full
success — the caller's `events_dispatched` counter ticks, the events vanish.
Decode as `Vec<Value>` and inspect each reply, or switch to non-transactional
pipelining and validate per-command results.
**Severity: high (silent data loss on transient backend errors).**

### Medium

#### #48 — `shutdown()` returns `Ok(())` after timeout-with-drops
**File:** `src/bus.rs:1065-1083`
When the in-flight wait times out at the configured grace deadline, the code
logs a warning and bumps `events_dropped` for the stranded count, but still
returns `Ok(())`. Callers can't distinguish clean from lossy shutdown. Return a
`ShutdownError::Lossy { dropped: u64 }` or surface a structured field on the
success path. **Severity: medium (operator-visibility hole).**

#### #49 — `Drop` releases shutdown gates synchronously while workers hold `Arc<Self>`
**File:** `src/bus.rs:1273-1282`
`Drop::drop()` synchronously sets `shutdown=true` and
`drain_finalize_ready=true` without awaiting the workers. Worker tasks holding
`Arc<Self>` may observe the gate during partial destruction; the comment notes
"events are lost" but doesn't address the partial-Drop UB risk. Workers should
be `take()`-ed into a join set in an explicit shutdown path; `Drop` should at
minimum verify they were already joined.
**Severity: medium (correctness fragility on rare drop-without-shutdown).**

#### #50 — `poll_merger` ArcSwap leaves polls operating on stale topology
**File:** `src/bus.rs:472-475, 626, 777-778`
Concurrent `poll()` callers `.load()` the merger Arc; a topology change
(`add_shard` / `remove_shard`) `.store()`s a fresh merger. Polls already in
flight continue against the old shard list and can return incomplete results
during scaling. Pin the merger version into `ConsumeResponse` and let the
caller retry on mismatch, or block topology changes until in-flight polls
complete. **Severity: medium (intermittent missed events during scaling).**

#### #51 — `flush()` baseline excludes events flushed via `remove_shard_internal`
**File:** `src/bus.rs:876-925`
`flush()` snapshots `events_ingested` then waits for
`events_dispatched + events_dropped >= target`. If a scale-down occurs
mid-flush, `remove_shard_internal` flushes the worker's pending batch
*directly to the adapter*, bypassing the `events_dispatched` increment at
`:1611`. The flush can return early before those redirected events are durably
persisted. Either bump the counter on the redirect path, or block scale-downs
while a flush is in flight.
**Severity: medium (premature flush success during concurrent scale-down).**

#### #52 — `dispatch_batch` final attempt collapses error reasons
**File:** `src/bus.rs:1499-1509`
The final attempt after retries returns `false` for all of: retryable
exhausted, non-retryable, and timeout — with one log line "Failed to dispatch
batch, dropping". Operators can't tell expected drops from unexpected fatal
errors in metrics. Tag the drop reason and emit distinct counters
(`drops_retry_exhausted` / `drops_fatal` / `drops_timeout`).
**Severity: medium (observability hole, not correctness).**

#### #53 — `unwrap_or_default()` on JetStream remote JSON
**File:** `src/adapter/jetstream.rs:104-118`
```rust
let raw = value.get("r").cloned().unwrap_or(serde_json::Value::Null);
let raw_bytes = Bytes::from(serde_json::to_vec(&raw).unwrap_or_default());
```
A malformed/oversized `r` field whose serialization fails silently becomes
empty bytes. The event is then "delivered" with an empty payload instead of
surfacing the corruption. Fail loudly: return
`AdapterError::Fatal("malformed event")`.
**Severity: medium (silent corruption on bad stream entries).**

#### #54 — Cold-stream poll walks `fetch_limit * 10` round-trips
**File:** `src/adapter/jetstream.rs:426-432`
```rust
let (max_seq, first_seq) = match stream.info().await {
    Ok(info) => (info.state.last_sequence, info.state.first_sequence),
    Err(_)   => (start_seq.saturating_add(span), 0),  // first_seq = 0
};
```
On `info()` failure for an empty/cold stream, `first_seq=0` causes the loop at
`:437` to walk every sequence from 1 to `max_seq` collecting `NotFound`s. With
`fetch_limit=101`, that's ~1010 RTTs per poll. Cache the previous `info`
result, or treat consecutive `NotFound`s as end-of-stream after a small budget.
**Severity: medium (pathological latency, easy to reproduce).**

#### #55 — Redis dedup eviction race exceeds capacity
**File:** `src/adapter/redis_dedup.rs:156-174`
```rust
if self.seen.len() >= self.capacity {
    if let Some(evicted) = self.order.pop_front() {
        self.seen.remove(&evicted);
    }
}
self.seen.insert(...);
```
The check-then-evict-then-insert sequence isn't atomic; concurrent calls can
each see `len == capacity`, both evict, both insert, ending at
`len > capacity`. With `capacity == 1`, two threads racing on different ids
both see `len == 0`, skip eviction, and both insert. Evict-while-full inside a
single mutex-guarded section, or switch to an LRU implementation that handles
capacity atomically. **Severity: medium (LRU bound violation, residual memory
growth).**

#### #56 — Modulo bias in equal-weight candidate selection
**File:** `src/shard/mapper.rs:514`
```rust
let idx = (event_hash as usize) % candidates.len();
```
For `candidates.len()` not a power of two, low-bucket indices receive slightly
more events than high ones. With u64 hashes the bias is small but non-zero and
biases hot-shard selection over time. Replace with rejection sampling
(`fastrand`-style) or `((hash as u128 * len as u128) >> 64) as usize` (Lemire's
technique). **Severity: medium (sustained skew, hard to debug from metrics).**

#### #57 — `per_shard_limit` silently capped at 10 000
**File:** `src/consumer/merge.rs:296-303`
```rust
let per_shard_limit = request.limit
    .div_ceil(shards.len()).max(1)
    .saturating_mul(over_fetch_factor)
    .min(10_000);
```
The hard cap silently truncates per-shard fetches. A consumer with a
caller-set `limit=200_000` over 10 shards expects 20 000/shard plus over-fetch;
it gets 10 000 with no warning. Either make the cap configurable (carry it on
`ConsumeRequest`) or surface a structured
`response.truncated_at_per_shard_cap = true`.
**Severity: medium (silent under-delivery on large polls).**

#### #58 — `TimestampGenerator` uses hard-coded baseline `0` for `delta_as_nanos`
**File:** `src/timestamp.rs:85`
```rust
let now = self.clock.delta_as_nanos(0, raw);
```
`quanta::Clock::new()` calibrates against an internal raw-TSC offset that is
*not* zero. Two `TimestampGenerator`s instantiated at different times produce
timestamps that differ by the calibration drift, breaking the documented
monotonicity invariant *across* generators. Use `self.clock.start_raw()` (or
whatever quanta exposes) as the baseline, or store the baseline at generator
construction.
**Severity: medium (cross-generator timestamp ordering corrupted).**

### Low

#### #59 — `in_flight_ingests` is `AtomicU32` with no saturating semantics
**File:** `src/bus.rs:650-652`
A 4-billion-in-flight wrap is not realistic in production, but the counter
participates in the shutdown protocol and the wrap would invert the protocol's
ordering check. Either widen to `AtomicU64` or use `fetch_add`/`fetch_sub` with
a `debug_assert` on the returned value.

#### #60 — `drain_finalize_ready` Release pairs only via implicit fence
**File:** `src/bus.rs:1088-1094`
The Release store on `drain_finalize_ready` is ordered after the
`in_flight_ingests` SeqCst spin via the implicit fence carried by SeqCst
operations. Correct today, but a future change to the spin's ordering would
silently break it. Add an explicit `fence(Ordering::Release)` or upgrade the
`drain_finalize_ready` store to SeqCst.

#### #61 — `TimestampGenerator` monotonicity stalls before the documented panic
**File:** `src/timestamp.rs:93-96`
Once `last == u64::MAX-1`, calls with `now < MAX` return `MAX` (via `.max()` on
the result), and only the call after that panics. The monotonicity contract
briefly stalls before failure. Panic preemptively when `last == u64::MAX-1`
and the input is non-strictly-greater.

#### #62 — `cpus.saturating_mul(2)` caps `max_shards: u16` at 65 535
**File:** `src/config.rs:744`
On hosts with >32 767 CPUs, `high_throughput` silently caps at 65 535 instead
of the documented 2× CPU count. Edge-case-only; either widen `max_shards` to
`u32` or cap deliberately and document.

#### #63 — `Cursor::encode` returns empty cursor on serialization failure
**File:** `src/consumer/merge.rs:100-103`
```rust
let json = serde_json::to_string(&self.positions).unwrap_or_default();
```
Should be unreachable for the current `positions` schema
(`HashMap<u16, Arc<str>>`), but the silent fallback to an empty cursor would
restart the consumer from the beginning on the next poll. Propagate the error.

#### #64 — `has_more=true` from a stalled adapter is silently suppressed
**File:** `src/consumer/merge.rs:568-574`
The merger detects "adapter reports `has_more` with no progress" and
suppresses `has_more` to avoid caller infinite loops. Correctness is preserved,
but the underlying adapter bug is invisible to the caller — they see "no more
events" and exit. Surface a `stalled_shards: Vec<u16>` on the response so
callers/operators can alert.

#### #65 — `velocity_samples` `VecDeque` bounded only by time, not count
**File:** `src/shard/batch.rs:29, 62-72`
At 100 k events/s with a 100 ms `velocity_window`, the deque grows to ~10 k
entries; `calculate_velocity` is then O(n) per call. Cap by sample count as
well, or sample at fixed `tick_granularity`.

#### #66 — `add_events(vec![])` flushes as a side effect
**File:** `src/shard/batch.rs:241-252`
`add_events` with an empty vector calls `check_timeout`, which can trigger a
flush of the open batch. A caller treating `add_events([])` as a no-op is
surprised. Either skip `check_timeout` on empty input or rename / document the
side-effect.

### Pass 3 — top 3 to fix first
1. **#44 + #45** `bus.rs` activation-failure scratch loss + batch-worker abort
   flush (`:420-463`, `:1554-1643`) — paired silent-loss bugs on every
   error/shutdown path.
2. **#47** `redis.rs:399-405` MULTI/EXEC partial-failure swallow — silent data
   loss on transient backend errors.
3. **#54** `jetstream.rs:426-432` cold-stream poll walking 1000+ RTTs — easy
   reproducer, observable as "JetStream is slow" in production.

---

## Pass 4 — consumer / adapter / ffi

### High

#### #67 — `PollMerger` default shard list is wrong after dynamic scale-down — silent event loss
**File:** `src/consumer/merge.rs:286-289`
Constructed at `src/bus.rs:236-237` and re-stored at `src/bus.rs:472-475`.
```rust
let shards: Vec<u16> = request
    .shards
    .clone()
    .unwrap_or_else(|| (0..self.num_shards).collect());
```
`PollMerger::new` is given only a *count* (`shard_manager.num_shards()`), not
the actual id set. `ShardMapper::scale_down` removes shards by lowest weight
(`src/shard/mapper.rs:907`), not by highest id, so after a scale-down the
active id set can be sparse — e.g. `{1, 2}` after id 0 is drained. With
`num_shards == 2`, the default-shards path generates `[0, 1]`:
- Shard 0 (drained / nonexistent) is polled. JetStream's
  `get_or_create_stream(0)` will recreate the stream and return empty / stale
  data.
- Shard 2 (the one with live events) is never polled — events ingested into it
  via the bus are invisible to consumers using the default shard list.

Static-mode (no scaling) is unaffected because ids are always
`0..num_shards` there.
Fix: pass the active id set (e.g. `shard_manager.shard_ids()`) into
`PollMerger`, or resolve at `poll()` time.

### Medium

#### #68 — Producer-nonce `sync_all` is a silent no-op on Windows
**File:** `src/adapter/dedup_state.rs:227-229`
```rust
if let Ok(f) = fs::File::open(&tmp_path) {
    let _ = f.sync_all();
}
```
`fs::File::open` opens read-only. On Windows, `File::sync_all` calls
`FlushFileBuffers`, which returns `ERROR_ACCESS_DENIED` on a read-only handle.
The error is swallowed by `let _ =`. The atomic-rename idiom assumes the data
has been fsync'd before rename; on Windows that guarantee is silently absent.
Power-loss between rename and the OS's own background flush can leave the
nonce file undurable — on next start, `load_or_create` reads zeros / partial
bytes / nothing, surfacing either as `InvalidData` or, with bad luck, as a
wrong-but-valid nonce, silently breaking JetStream / Redis cross-restart
dedup.
Fix: keep the writable handle from the original write (use
`OpenOptions::new().write(true).open(&tmp_path)` and `sync_all` on that), so
the fsync runs on a write-capable handle on every platform.

#### #69 — JetStream `from_id` cursor `seq + 1` overflows
**File:** `src/adapter/jetstream.rs:395-398`
```rust
let start_seq = from_id
    .and_then(|id| id.parse::<u64>().ok())
    .map(|seq| seq + 1)
    .unwrap_or(1);
```
A caller-supplied cursor of `"18446744073709551615"` (`u64::MAX`) parses
successfully and `seq + 1` panics in debug or wraps to `0` in release,
silently restarting polling from the start of the stream and re-delivering
everything. Cursors are produced by the adapter today, so not reachable
internally — but `bus.poll()` accepts any base64 `CompositeCursor`, so a
hand-crafted cursor lands here. Use `seq.checked_add(1).unwrap_or(seq)`
(saturate at `u64::MAX`).

#### #70 — `Runtime::block_on` from `extern "C"` shims unwinds across FFI
**Files:**
- `src/ffi/mod.rs:772, 916, 1048`
- `src/ffi/cortex.rs:419-453, 497-499`
- `src/ffi/mesh.rs:468`
```rust
rt.block_on(async move { ... })
```
`Runtime::block_on` panics with "Cannot start a runtime from within a runtime"
if the calling thread is already inside a tokio runtime context. The function
is `extern "C"`, so the panic unwinds across the FFI boundary into C / Go-cgo
/ Python — UB. The pattern is fine for purely-C/Go callers (the common case),
but a Rust caller that loads the cdylib and calls these functions from inside
its own tokio runtime triggers UB. `net_init` (`mod.rs:284-316`) already
hardens against this for runtime *construction*; the steady-state ops
(`net_poll`, `net_flush`, `net_shutdown`, `net_redex_*`, `net_mesh_new`) do
not. Wrap each `block_on` call in `std::panic::catch_unwind` to convert into
an error code, or document the runtime-context restriction prominently.

#### #71 — Redis & JetStream `limit + 1` overflow on adversarial limits
**Files:**
- `src/adapter/redis.rs:444`
- `src/adapter/jetstream.rs:401`
```rust
let fetch_limit = limit + 1;
```
A caller-supplied `limit == usize::MAX` (reachable through the FFI poll-request
JSON, which does `usize::try_from` but does not bound the value) panics in
debug or wraps to `0` in release, returning an empty result with no error.
Use `limit.saturating_add(1)`.

### Low

#### #72 — `mapper.rs` cooldown check + scale mutation aren't atomic w.r.t. each other
**File:** `src/shard/mapper.rs:854-923`
`scale_down` reads `last_scaling` to gate the cooldown, then later writes
`last_scaling`, both under the same outer `shards.write()` — sound today
because mutual exclusion is preserved across the gap. Brittle if a future
refactor narrows the lock scope on `shards`. Fold the cooldown gate into the
same lock as `shards`, or use a single `compare_exchange`-style update on
`last_scaling`.

#### #73 — `RedisStreamDedup` is FIFO eviction, not LRU as documented
**File:** `src/adapter/redis_dedup.rs:82-86, 156-186`
The doc/comments describe "LRU-bounded" but `is_duplicate` never moves a
re-observed id to the back of the queue; the queue is strictly
insertion-ordered. The eviction is FIFO. Dedup correctness is intact; only
callers expecting "frequently-observed ids stay tracked longer" are surprised.
Either rename the type / docs or implement true LRU semantics.

#### #74 — `dispatch_batch` retry sleep has no jitter / backoff
**File:** `src/bus.rs:1496`
```rust
tokio::time::sleep(Duration::from_millis(100)).await;
```
Under a partial backend outage (Redis / JetStream slow but not dead), every
shard's `BatchWorker` retries on the same 100 ms cadence with no jitter,
producing a synchronized retry storm that amplifies load while the backend is
recovering. Add jitter or capped exponential backoff.

### Pass 4 — top 3 to fix first
1. **#67** `consumer/merge.rs:286-289` — `PollMerger` default shard list is
   `0..num_shards` and lies after dynamic scale-down; consumers using the
   default (`request.shards == None`) lose events from any shard whose id ≥
   current count. The single highest-impact finding in this pass.
2. **#68** `adapter/dedup_state.rs:227-229` — Windows fsync silently disabled,
   breaking cross-restart dedup durability promise.
3. **#69 + #71** `adapter/{redis,jetstream}.rs` — `limit + 1` and `seq + 1`
   overflows on attacker-supplied poll inputs; trivial to harden with
   `saturating_add` / `checked_add`.

---

## Global priority (cross-pass)

Highest-impact silent-data-loss bugs spanning all four passes:

1. **#67** `PollMerger` default-shard-list lie after scale-down — consumers
   using default shards silently miss events on every poll.
2. **#1** `migration_target.rs` lost-events on `?` early-return — permanent
   per-daemon desync.
3. **#44 + #45 + #46** `bus.rs` shutdown / activation-failure / scratch /
   batch-worker silent loss — paired bugs on every error path.
4. **#47** Redis MULTI/EXEC partial-failure swallow — silent loss on transient
   backend errors.
5. **#2 + #3** Redex retention drops + index-task halt after `Lagged` —
   silent corruption that never recovers.

---

## Files audited with no bugs found

**Cortex:** `envelope.rs`, `meta.rs`, `error.rs`, `config.rs`, `mod.rs`,
`watermark.rs`, `memories/fold.rs`, `tasks/fold.rs`, `memories/state.rs`,
`tasks/state.rs`, `*/types.rs`, `*/dispatch.rs`, `memories/query.rs`,
`tasks/query.rs`, `memories/filter.rs`, `tasks/filter.rs`, `memories/watch.rs`,
`tasks/watch.rs` (well-defended; explicit prior-bug pins #134, #141, #142,
#144, #148, CR-20).

**Other:** `behavior/broadcast.rs` (only a `SUBPROTOCOL_CAPABILITY_ANN`
constant — real broadcast logic lives on `MeshNode`), `consumer/mod.rs`
(re-export shim), `noop.rs` (counter-only test adapter).
