sqlite: dead-watcher subscribe fails closed + initial-stat identity-unknown baseline (review 004 items 2+3) — the poll loop sets a dead sign before the watcher-fatal exit and subscribe() fails closed on it (and the taken watcher slot) atomically with enrollment; listen/stream subscribe map it to Error::Database, never park on a dead feed; the identity baseline is Option-tracked so a failed startup stat no longer seeds the (0,0) false identity that a transient recovery would trip into death; engine-sqlite.md posture covers both arms; PROVENANCE D-39; task -> completed

alkstore-sqlite/src/notify.rs

@@ -25,7 +25,11 @@
 //! (it never reopens); `try_recv` / `recv_timeout` carry
 //! [`Error::Closed`], distinguishable from idle's `Ok(None)`;
 //! `recv_timeout`'s timeout expiry without a wake is `Ok(None)` — the
-//! documented idle arm, never a timeout signal.
+//! documented idle arm, never a timeout signal. A `listen` issued
+//! *after* a watcher death fails closed with
+//! [`Error::Database`](alkstore::Error) — no new enrollment onto a
+//! dead feed (which would park forever); recovery is a fresh store
+//! open.
 //!
 //! Unsubscribe is the receiver's own [`Drop`]: the substrate prunes
 //! disconnected subscribers lazily (at the next fanout), so the
@@ -40,7 +44,7 @@ use serde_json::Value;
 
 use alkstore::{BoxedFuture, Error, Result, Wake, WakeReceiver};
 
-use crate::seam::{sqlite_error, with_writer};
+use crate::seam::{database_error, sqlite_error, with_writer};
 use crate::substrate::{SharedUpdateWatcher, Writer};
 
 /// The bridge's tokio-channel capacity. The substrate's 1-slot sync
@@ -80,7 +84,10 @@ pub(crate) fn notify(
 /// The listen path (see the module docs). Name validation fires at the
 /// entry point; a listen against a closed store fails closed with
 /// [`Error::Database`](alkstore::Error) (the engine-wide closed-store
-/// posture — `begin_tx`/`notify` yield the same shape).
+/// posture — `begin_tx`/`notify` yield the same shape); a listen
+/// against a store whose watcher has *died* (the db file was replaced)
+/// fails closed the same way — a new subscription never enrolls onto a
+/// dead feed to park forever (review 004 item 2's new-subscriber arm).
 pub(crate) fn listen(
     watcher: Arc<SharedUpdateWatcher>,
     closed: Arc<AtomicBool>,
@@ -92,7 +99,7 @@ pub(crate) fn listen(
             "the store is closed",
         )));
     }
-    let (id, rx) = watcher.subscribe();
+    let (id, rx) = watcher.subscribe().map_err(database_error)?;
     let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(WAKE_CHANNEL_CAPACITY);
     let channel = channel.to_string();
     let _bridge = tokio::task::spawn_blocking(move || {

alkstore-sqlite/src/store.rs

@@ -13,12 +13,13 @@
 //! bootstrap.
 //!
 //! Watcher spawn is fallible in the substrate (ADR-012 §4's W-2); the
-//! failure maps into [`Error::Database`](alkstore::Error::Database)
+//! failure maps into [`Error::Database`](alkstore::Error) engine-side
-//! engine-side and `open` fails if the watcher cannot start. On
+//! and `open` fails if the watcher cannot start. On watcher death the
-//! watcher death the substrate's `WatcherDeathGuard` closes every
+//! substrate's `WatcherDeathGuard` closes every subscriber (ADR-006)
-//! subscriber (ADR-006) — the death-guard plumbing exists from open
+//! and new subscriptions fail closed with `Error::Database` (the
-//! onward; the consumer-visible receiver bridging arrives with
+//! dead-watcher sign at the substrate's `subscribe`; review 004
-//! `listen`.
+//! item 2) — a listen or stream subscribe on a dead-watcher store
+//! never parks.
 //!
 //! Drop/close: the watcher's poll thread is joined (releasing its
 //! connection and the db file handle) and the writer slot + reader

alkstore-sqlite/src/store/notify_tests.rs

@@ -262,6 +262,50 @@ async fn watcher_death_closes_the_receiver_terminally() {
     cleanup(&dir);
 }
 
+/// Acceptance: `listen` *after* a watcher death fails closed with
+/// `Error::Database` — a new subscription never enrolls onto a dead
+/// feed (which would park forever; review 004 item 2's
+/// new-subscriber arm). Driven through the substrate's real
+/// dead-man's switch (db-file replacement), unix-gated like the other
+/// file-replacement drives. The pre-death subscribers' terminals stay
+/// pinned above (`watcher_death_closes_the_receiver_terminally`).
+#[tokio::test(flavor = "multi_thread")]
+#[cfg(unix)]
+async fn listen_after_watcher_death_fails_closed() {
+    let dir = temp_dir("death-listen");
+    let path = temp_path("death-listen");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    // Kill the watcher: replace the db file (the identity mismatch
+    // drives the substrate's watcher-fatal death).
+    let other = dir.join("store.db.replacement");
+    std::fs::File::create(&other).unwrap();
+    std::fs::rename(&other, &path).unwrap();
+    until("death-listen", || store.watcher.is_dead()).await;
+
+    let err = match store.listen("c").await {
+        Err(e) => e,
+        Ok(_) => panic!("listen on a dead-watcher store must fail closed, not park"),
+    };
+    match &err {
+        Error::Database(source) => {
+            let message = source.to_string();
+            assert!(
+                message.contains("the update watcher has died"),
+                "the source chain must name the watcher death: {message}"
+            );
+        }
+        other => panic!("listen after watcher death is Database, got {other:?}"),
+    }
+    assert_eq!(
+        store.watcher.subscriber_count(),
+        0,
+        "the failed listen enrolled no subscription"
+    );
+    store.close();
+    cleanup(&dir);
+}
+
 /// Acceptance: `try_recv` distinguishes closed from idle pre-close —
 /// `Ok(None)` with no wake pending (idle), `Some` when one arrived,
 /// `Err(Closed)` only after the source closed (the pinned arms).

alkstore-sqlite/src/store/open_tests.rs

@@ -82,7 +82,7 @@ async fn open_leaves_the_watcher_running_and_fanning_out() {
     let path = dir.join("store.db");
     let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap();
 
-    let (_id, rx) = store.watcher.subscribe();
+    let (_id, rx) = store.watcher.subscribe().unwrap();
 
     let writer_conn = store.writer.acquire().unwrap();
     writer_conn
@@ -124,7 +124,7 @@ async fn closing_the_store_closes_watcher_subscribers() {
     let path = dir.join("store.db");
     let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap();
 
-    let (_id, rx) = store.watcher.subscribe();
+    let (_id, rx) = store.watcher.subscribe().unwrap();
     store.close();
     assert!(
         rx.recv().is_err(),
@@ -356,7 +356,7 @@ async fn drop_runs_the_full_teardown() {
     let path = dir.join("store.db");
     {
         let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap();
-        let (_id, rx) = store.watcher.subscribe();
+        let (_id, rx) = store.watcher.subscribe().unwrap();
         drop(store);
         assert!(
             rx.recv().is_err(),

alkstore-sqlite/src/store/stream_tests.rs

@@ -417,6 +417,51 @@ async fn watcher_death_closes_the_subscription_terminally() {
     cleanup(&dir);
 }
 
+/// Acceptance: a stream `subscribe` *after* a watcher death fails
+/// closed with `Error::Database` — a new subscription never enrolls
+/// onto a dead feed (which would park forever; review 004 item 2's
+/// new-subscriber arm). Driven through the substrate's real
+/// dead-man's switch (db-file replacement), unix-gated like the other
+/// file-replacement drives. The pre-death subscribers' terminals stay
+/// pinned above (`watcher_death_closes_the_subscription_terminally`).
+#[tokio::test(flavor = "multi_thread")]
+#[cfg(unix)]
+async fn subscribe_after_watcher_death_fails_closed() {
+    let dir = temp_dir("death-subscribe");
+    let path = temp_path("death-subscribe");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let stream = store.stream("s").await.unwrap();
+
+    // Kill the watcher: replace the db file (the identity mismatch
+    // drives the substrate's watcher-fatal death).
+    let other = dir.join("store.db.replacement");
+    std::fs::File::create(&other).unwrap();
+    std::fs::rename(&other, &path).unwrap();
+    until("death-subscribe", || store.watcher.is_dead()).await;
+
+    let err = match stream.subscribe("c").await {
+        Err(e) => e,
+        Ok(_) => panic!("subscribe on a dead-watcher store must fail closed, not park"),
+    };
+    match &err {
+        Error::Database(source) => {
+            let message = source.to_string();
+            assert!(
+                message.contains("the update watcher has died"),
+                "the source chain must name the watcher death: {message}"
+            );
+        }
+        other => panic!("subscribe after watcher death is Database, got {other:?}"),
+    }
+    assert_eq!(
+        store.watcher.subscriber_count(),
+        0,
+        "the failed subscribe enrolled no subscription"
+    );
+    store.close();
+    cleanup(&dir);
+}
+
 /// Acceptance: a subscribe attached while the stream is at tail idles
 /// (`Ok(None)`), then delivers when the next publish lands.
 #[tokio::test(flavor = "multi_thread")]

alkstore-sqlite/src/stream.rs

@@ -27,9 +27,11 @@
 //!   stored offset to the current tail, then wake-driven re-reads off
 //!   the watcher fanout (the same bridge shape as `listen`, yielding
 //!   `Result<StreamEvent>` per `EventReceiver`). Watcher death closes
-//!   it terminally (`recv() -> None`); recovery is a fresh
+//!   it terminally (`recv() -> None`); a subscribe issued after a
-//!   `subscribe(consumer)` resuming from the saved checkpoint
+//!   watcher death fails closed with
-//!   (ADR-021 §5).
+//!   ([`Error::Database`](alkstore::Error)) (never a parked enrollment
+//!   onto a dead feed); recovery is a fresh `subscribe` on a reopened
+//!   store, resuming from the saved checkpoint (ADR-021 §5).
 //!
 //! Subscribe specifics:
 //!
@@ -70,8 +72,8 @@ use alkstore::{
 };
 
 use crate::seam::{
-    ReaderLoopExit as LoopExit, blocking_read, closed_store_error, sqlite_error, with_reader,
+    ReaderLoopExit as LoopExit, blocking_read, closed_store_error, database_error, sqlite_error,
-    with_writer,
+    with_reader, with_writer,
 };
 use crate::substrate::{
     Readers, SharedUpdateWatcher, Writer, stream_get_offset, stream_publish, stream_read_since,
@@ -347,7 +349,12 @@ impl StreamHandle for SqliteStreamHandle {
 /// The `subscribe` bridge construction: the watcher subscription is
 /// taken before any read, the bridge thread then replays and parks on
 /// the wake feed. A thread-spawn failure unsubscribes and fails with
-/// `Database` (no panics cross the seam, the W-2 posture).
+/// `Database` (no panics cross the seam, the W-2 posture). A watcher
+/// that has *died* (db-file replacement) fails the subscription
+/// closed with `Database` too — a new subscription never enrolls onto
+/// a dead feed to park forever (the death-guard closes *pre-death*
+/// subscribers; this closes the *new-subscriber* arm, review 004
+/// item 2).
 fn spawn_bridge(
     name: String,
     consumer: String,
@@ -356,7 +363,7 @@ fn spawn_bridge(
     watcher: Arc<SharedUpdateWatcher>,
     store_closed: Arc<AtomicBool>,
 ) -> Result<SqliteEventReceiver> {
-    let (id, wake_rx) = watcher.subscribe();
+    let (id, wake_rx) = watcher.subscribe().map_err(database_error)?;
     let wake_rx = Arc::new(Mutex::new(wake_rx));
     let (event_tx, event_rx) =
         tokio::sync::mpsc::channel::<Result<StreamEvent>>(EVENT_CHANNEL_CAPACITY);

alkstore-sqlite/src/substrate/PROVENANCE.md

@@ -131,6 +131,7 @@ Cherry-picks: empty at scaffold, append-only forever.
 | D-36 | drop | `UpdateWatcher::spawn` and `SharedUpdateWatcher::new` cut (the default-cadence convenience constructors): upstream's embedding harnesses opened watchers without explicit config; this engine's single open path always constructs a `WatcherConfig` — from `SqliteOpts` (ADR-023 §4's `poll_interval` knob, `None` = the 1 ms default) — and passes it through `*_with_config`; a config-less twin would re-create two open postures (the mixed-flag inconsistency D-29 resolved at the URI layer). `WatcherConfig::default()` remains the default's single owner; the tests driving the cut constructors now build the config explicitly, pinning the same behavior | `watcher.rs` (both constructors + five watcher-test call sites) | ADR-023 §4 (the cadence knob rides the engine's opts — one open posture carries config); ADR-012 §2 (one construction path the engine reviews); wave-3 lint-removal obligation — cut, not suppressed | ours |
 | D-37 | re-derivation | visibility-only widening on the re-derived `@every` surface: `parse_every_interval` (`pub(crate)` → `pub`) and `skip_forward` (module-private → `pub`) in `queue_ops.rs`, plus this subtree's facade re-export (`mod.rs`: `pub use queue_ops::{parse_every_interval, skip_forward}`), so the engine crate root can `#[doc(hidden)] pub use` both as the wave-7 fuzzing differential target's drive surface (`fuzz-parser-exposure`, target `every_parse`). Zero logic change, zero signature change, zero name change; `mod substrate` stays private, so containment holds — nothing leaves the crate except the doc-hidden root re-exports. A `pub use` re-export mechanically requires item-level `pub` (E0364/E0603), which is why the facade line is in `mod.rs`; the module doc's "nothing re-exported" boundary sentence was amended in the same commit to name this sole exception | `queue_ops.rs` (two visibility keywords); `mod.rs` (facade re-export line + boundary-doc exception sentence) | ADR-012 §3 (fidelity governs the ported honker body — `watcher.rs`/`schema.rs`/`ops.rs` untouched; `queue_ops.rs` is the re-derived half, owned contract-v1 code, so a visibility-only delta is legitimate); docs/research/fuzzing.md §3.2 (option (b): the `#[doc(hidden)]` narrow exposure); ADR-017 §3 (doc-hidden test surface, no stability guarantee — the `Job::from_row` precedent) | ours |
 | D-38 | port delta | degenerate/adversarial-triple hardening on `skip_forward` (two legs, both aligning to the pg engine's committed clamp shape where it existed and extending pg itself): (1) the `interval` is clamped (`interval.max(1) as i128`) — pre-fix, `interval == 0` panicked on i128 division-by-zero, reachable only through degenerate/non-grammar callers (the parsers reject `n <= 0`, so every engine-internal call site passes `interval >= 1`); (2) the gap is computed in i128 (`now as i128 - next_fire_at as i128`) — pre-fix, `now - next_fire_at` was an **i64 subtraction** that panics in debug (overflow) / silently wraps in release when `(next_fire_at, now)` straddle the full domain (`now=i64::MIN`-adjacent, `next_fire_at=i64::MAX`-adjacent), feeding a wrapped gap to the 128-bit arithmetic and breaking the saturation guarantee. Found by the `every_parse` differential target (wave 7, tasks/fuzz-target-every-parse) driving the adversarial `(next_fire_at, now, interval)` space; pg needed leg (2) as well. Observable behavior unaffected in the reachable domain (|gap| < i64 range), saturation posture now holds over the whole domain. Zero signature change; lineage body untouched (`skip_forward` is the re-derived scheduler half, D-12's own) | `queue_ops.rs` (`skip_forward`, two lines) | docs/research/fuzzing.md §3.2 (the overflow-posture invariant: reject-or-saturate, never wrap or panic, over the whole domain); wave-6's register-overflow alignment (tasks/review-wave-6); alkstore-postgres/src/scheduler.rs `skip_forward` (the pinned clamp shape) | ours |
+| D-39 | port delta | watcher-death state machine (review 004 items 2+3, two legs on the ported watcher): (1) **identity-unknown baseline** — the loop's initial identity `stat` failure seeds an unknown baseline instead of the false `(0, 0)`; the first *successful* stat captures it without driving the death path (pre-fix, a transient startup stat hiccup killed the watcher — and closed every subscriber — when the first later stat succeeded); (2) **dead-watcher fail-closed subscribe** — the loop sets a `dead` sign on the caller (`SharedUpdateWatcher`, an `Arc<AtomicBool>`) before exiting through the watcher-fatal path, and `subscribe` becomes fallible, failing closed on the sign (or a taken watcher slot) checked atomically with the enrollment under the senders lock — new subscriptions after a death fail instead of enrolling onto a feed with no poller alive to park forever (the death-guard's clear of *pre-death* subscribers unchanged, D-07 still holds); the engine maps the failure into `Error::Database` at `listen`/stream `subscribe` (ADR-008 §5's opaque fallback). Lineage behavior for both legs was the (0,0)-seed and the infallible enroll | `watcher.rs` (`run_poll_loop`'s baseline/identity arms, `SharedUpdateWatcher`'s dead sign + fallible `subscribe`); `notify.rs`/`stream.rs` (the engine-side mapping) | engine-sqlite.md's watcher-death posture ("new subscriptions fail closed"; identity unknown until the first successful stat); review 004 items 2–3 (tasks/sqlite-fix-watcher-death-subscribe); ADR-006 (the close/fail-closed posture); ADR-012 §4 (delta discipline; D-07's death path refined) | ours |
 
 ### Cherry-picks
 

alkstore-sqlite/src/substrate/watcher.rs

@@ -135,15 +135,23 @@ fn open_watcher_conn(path: &Path) -> rusqlite::Result<Connection> {
 ///    vanished db file produces bounded open attempts (W-1), not the
 ///    lineage's one attempt per millisecond.
 /// 3. **Identity check (about every 100 ms):** `stat(db_path)` to
-///    compare `(dev, ino)`. If the file was replaced, log the precise
+///    compare `(dev, ino)` against the baseline. If the file was
-///    diagnostic and exit through the ordinary death path (the fork's
+///    replaced, log the precise diagnostic and exit through the
-///    dead-man's-switch delta — the lineage panicked here); continuing
+///    ordinary death path (the fork's dead-man's-switch delta — the
-///    would silently watch stale data. `WatcherDeathGuard` closes
+///    lineage panicked here); continuing would silently watch stale
-///    every subscriber exactly as it does for any watcher death.
+///    data. `WatcherDeathGuard` closes every subscriber exactly as it
+///    does for any watcher death, and the caller's `dead` flag is set
+///    before the exit — new subscriptions on the dead store fail
+///    closed, they never park on a dead feed. The baseline starts
+///    unknown when the initial stat fails (the first *successful*
+///    stat captures it, without driving death — a failed startup stat
+///    must not seed a false identity that a later transient recovery
+///    would trip).
 pub(crate) fn run_poll_loop<F>(
     db_path: PathBuf,
     on_change: F,
     stop: Arc<AtomicBool>,
+    dead: Arc<AtomicBool>,
     ready: std::sync::mpsc::SyncSender<()>,
     poll_interval: Duration,
     open_conn_fn: ReconnectOpenFn,
@@ -161,11 +169,16 @@ pub(crate) fn run_poll_loop<F>(
         .as_ref()
         .and_then(|c| poll_data_version(c).ok())
         .unwrap_or(0);
-    let initial_identity = match stat_identity(&db_path) {
+    // The file identity baseline, seeded by the loop's initial stat:
-        Ok(id) => id,
+    // `None` while unknown — a failed startup stat must not seed the
+    // `(0, 0)` false identity that the first successful later stat
+    // would trip into the death path; the first successful stat
+    // captures the baseline instead, without driving death.
+    let mut baseline = match stat_identity(&db_path) {
+        Ok(id) => Some(id),
         Err(e) => {
             eprintln!("alkstore: failed to stat database for identity check: {e}");
-            (0, 0)
+            None
         }
     };
     let mut next_identity_check = Instant::now() + UPDATE_WATCHER_IDENTITY_INTERVAL;
@@ -211,19 +224,23 @@ pub(crate) fn run_poll_loop<F>(
         if now >= next_identity_check {
             next_identity_check = now + UPDATE_WATCHER_IDENTITY_INTERVAL;
             match stat_identity(&db_path) {
-                Ok(current) => {
+                Ok(current) => match baseline {
-                    if current != initial_identity {
+                    Some(expected) => {
-                        eprintln!(
+                        if current != expected {
-                            "alkstore: database file replaced: \
+                            eprintln!(
-                             expected (dev={}, ino={}), \
+                                "alkstore: database file replaced: \
-                             found (dev={}, ino={}) at {:?}. \
+                                 expected (dev={}, ino={}), \
-                             The watcher cannot recover; it has stopped and \
+                                 found (dev={}, ino={}) at {:?}. \
-                             closed all subscribers; close the store and reopen.",
+                                 The watcher cannot recover; it has stopped and \
-                            initial_identity.0, initial_identity.1, current.0, current.1, db_path
+                                 closed all subscribers; close the store and reopen.",
-                        );
+                                expected.0, expected.1, current.0, current.1, db_path
-                        return;
+                            );
+                            dead.store(true, Ordering::Release);
+                            return;
+                        }
                     }
-                }
+                    None => baseline = Some(current),
+                },
                 Err(e) => {
                     eprintln!("alkstore: stat identity check failed: {e}");
                     conn = None;
@@ -256,6 +273,7 @@ impl UpdateWatcher {
         db_path: PathBuf,
         on_change: F,
         config: WatcherConfig,
+        dead: Arc<AtomicBool>,
     ) -> Result<Self, String>
     where
         F: Fn() + Send + 'static,
@@ -269,6 +287,7 @@ impl UpdateWatcher {
             stop_t,
             ready_tx,
             config.poll_interval,
+            dead,
             None,
         )?;
         let _ = ready_rx.recv();
@@ -288,6 +307,7 @@ impl UpdateWatcher {
         stop_t: Arc<AtomicBool>,
         ready_tx: std::sync::mpsc::SyncSender<()>,
         poll_interval: Duration,
+        dead: Arc<AtomicBool>,
         stack_size: Option<usize>,
     ) -> Result<std::thread::JoinHandle<()>, String>
     where
@@ -304,6 +324,7 @@ impl UpdateWatcher {
                     db_path,
                     on_change,
                     stop_t,
+                    dead,
                     ready_tx,
                     poll_interval,
                     open_watcher_conn,
@@ -345,6 +366,12 @@ impl Drop for WatcherDeathGuard {
 pub(crate) struct SharedUpdateWatcher {
     watcher: Mutex<Option<UpdateWatcher>>,
     senders: Arc<Mutex<HashMap<u64, SyncSender<()>>>>,
+    /// The death sign the poll loop sets before exiting through the
+    /// watcher-fatal path (db-file replacement). New subscriptions
+    /// fail closed on it — they must never enroll onto a feed with no
+    /// poller alive (the death-guard clears *existing* subscribers;
+    /// this closes the *new-subscriber* arm).
+    dead: Arc<AtomicBool>,
     next_id: AtomicU64,
 }
 
@@ -358,6 +385,7 @@ impl SharedUpdateWatcher {
         let senders: Arc<Mutex<HashMap<u64, SyncSender<()>>>> =
             Arc::new(Mutex::new(HashMap::new()));
         let senders_t = senders.clone();
+        let dead: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
         let death_guard = WatcherDeathGuard {
             senders: senders.clone(),
         };
@@ -372,10 +400,12 @@ impl SharedUpdateWatcher {
                 });
             },
             config,
+            dead.clone(),
         )?;
         Ok(Self {
             watcher: Mutex::new(Some(watcher)),
             senders,
+            dead,
             next_id: AtomicU64::new(0),
         })
     }
@@ -388,11 +418,32 @@ impl SharedUpdateWatcher {
     ///
     /// Channel capacity is 1: bursts coalesce into one wake per drain
     /// cycle. Wakes are "go re-read state" signals.
-    pub(crate) fn subscribe(&self) -> (u64, std::sync::mpsc::Receiver<()>) {
+    ///
+    /// Fallible, failing closed: on a dead watcher (the poll thread
+    /// exited through the watcher-fatal path — the store's db file was
+    /// replaced; close the store and reopen) or a closed watcher (the
+    /// watcher slot was taken by [`close`](Self::close)). Both checks
+    /// hold the senders lock together with the enrollment, so the
+    /// sign and the enroll are one step — the death-guard's clear
+    /// races the check as a pre-death subscriber (cleared → the
+    /// receiver sees the disconnect), never as a parked ghost. The
+    /// engine layer maps the failure into the contract taxonomy.
+    pub(crate) fn subscribe(&self) -> Result<(u64, std::sync::mpsc::Receiver<()>), String> {
+        let mut senders = self.senders.lock();
+        if self.dead.load(Ordering::Acquire) {
+            return Err(
+                "the update watcher has died: the database file was replaced; \
+                 new subscriptions fail closed — close the store and reopen"
+                    .to_string(),
+            );
+        }
+        if self.watcher.lock().is_none() {
+            return Err("the update watcher is closed".to_string());
+        }
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (tx, rx) = std::sync::mpsc::sync_channel(1);
-        self.senders.lock().insert(id, tx);
+        senders.insert(id, tx);
-        (id, rx)
+        Ok((id, rx))
     }
 
     /// Remove a subscriber. The corresponding receiver sees
@@ -407,6 +458,11 @@ impl SharedUpdateWatcher {
         self.senders.lock().len()
     }
 
+    #[cfg(test)]
+    pub(crate) fn is_dead(&self) -> bool {
+        self.dead.load(Ordering::Acquire)
+    }
+
     /// Disconnect all subscribers and synchronously join the poll
     /// thread. The thread owns the watcher's connection; joining drops
     /// that connection and releases the file handle. Idempotent — safe
@@ -464,6 +520,7 @@ mod watcher_tests {
             Arc::new(AtomicBool::new(false)),
             tx,
             Duration::from_millis(1),
+            Arc::new(AtomicBool::new(false)),
             Some(usize::MAX),
         )
         .expect_err("an oversized stack reservation must fail to spawn");
@@ -494,7 +551,7 @@ mod watcher_tests {
         let shared =
             SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap();
         let subs: Vec<(u64, std::sync::mpsc::Receiver<()>)> =
-            (0..50).map(|_| shared.subscribe()).collect();
+            (0..50).map(|_| shared.subscribe().unwrap()).collect();
 
         let writer = Connection::open(&tmp).unwrap();
         writer
@@ -532,7 +589,7 @@ mod watcher_tests {
 
         let shared =
             SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap();
-        let (id, rx) = shared.subscribe();
+        let (id, rx) = shared.subscribe().unwrap();
         assert_eq!(shared.subscriber_count(), 1);
 
         shared.unsubscribe(id);
@@ -564,7 +621,7 @@ mod watcher_tests {
 
         let shared =
             SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap();
-        let (_id, rx) = shared.subscribe();
+        let (_id, rx) = shared.subscribe().unwrap();
 
         // Let the watcher snapshot the initial identity.
         std::thread::sleep(Duration::from_millis(200));
@@ -586,6 +643,10 @@ mod watcher_tests {
                  WatcherDeathGuard didn't fire?"
             );
         }
+        assert!(
+            shared.is_dead(),
+            "the death sign must be set once the poll thread died"
+        );
 
         shared.close().unwrap();
         let _ = std::fs::remove_file(&tmp);
@@ -603,8 +664,13 @@ mod watcher_tests {
         let tmp = temp_db("watcher-replace");
         wal_db(&tmp);
 
-        let watcher =
+        let watcher = UpdateWatcher::spawn_with_config(
-            UpdateWatcher::spawn_with_config(tmp.clone(), || {}, WatcherConfig::default()).unwrap();
+            tmp.clone(),
+            || {},
+            WatcherConfig::default(),
+            Arc::new(AtomicBool::new(false)),
+        )
+        .unwrap();
 
         std::thread::sleep(Duration::from_millis(200));
 
@@ -634,7 +700,7 @@ mod watcher_tests {
         let shared =
             SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap();
         {
-            let _subs: Vec<_> = (0..10).map(|_| shared.subscribe()).collect();
+            let _subs: Vec<_> = (0..10).map(|_| shared.subscribe().unwrap()).collect();
             assert_eq!(shared.subscriber_count(), 10);
         }
 
@@ -858,6 +924,8 @@ mod watcher_tests {
         let wakes_t = wakes.clone();
         let stop = Arc::new(AtomicBool::new(false));
         let stop_t = stop.clone();
+        let dead = Arc::new(AtomicBool::new(false));
+        let dead_t = dead.clone();
         let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1);
         let db_path = tmp.clone();
         let handle = std::thread::spawn(move || {
@@ -867,6 +935,7 @@ mod watcher_tests {
                     wakes_t.fetch_add(1, Ordering::Relaxed);
                 },
                 stop_t,
+                dead_t,
                 ready_tx,
                 Duration::from_millis(5),
                 instrumented_open,
@@ -930,12 +999,15 @@ mod watcher_tests {
         ));
         let stop = Arc::new(AtomicBool::new(false));
         let stop_t = stop.clone();
+        let dead = Arc::new(AtomicBool::new(false));
+        let dead_t = dead.clone();
         let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1);
         let handle = std::thread::spawn(move || {
             run_poll_loop(
                 db_path,
                 || {},
                 stop_t,
+                dead_t,
                 ready_tx,
                 Duration::from_millis(tick_ms),
                 counting_open,
@@ -968,4 +1040,163 @@ mod watcher_tests {
              window at {tick_ms}ms/tick; the backoff admits at most {bound}"
         );
     }
+
+    /// Review 004 item 2 (substrate arm): after the poll thread exits
+    /// through the watcher-fatal path (file replacement), `subscribe`
+    /// fails closed — no new sender enrolls onto a feed with no poller
+    /// alive (which would park its receiver forever). Driven through
+    /// the same seam as the death-signal test: a pre-death subscriber
+    /// observes the death first, then the dead store rejects a fresh
+    /// subscription.
+    #[test]
+    #[cfg(unix)]
+    fn subscribe_fails_closed_after_watcher_death() {
+        let tmp = temp_db("death-subscribe");
+        wal_db(&tmp);
+
+        let shared =
+            SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap();
+        let (_id, rx) = shared.subscribe().unwrap();
+
+        std::thread::sleep(Duration::from_millis(200));
+
+        let other = temp_db("death-subscribe-other");
+        let _ = std::fs::File::create(&other).unwrap();
+        std::fs::rename(&other, &tmp).unwrap();
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
+        loop {
+            match rx.recv_timeout(Duration::from_millis(100)) {
+                Ok(()) => continue,
+                Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
+                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "watcher died but subscriber's channel never disconnected"
+            );
+        }
+        assert!(
+            shared.is_dead(),
+            "the death sign must be set once the poll thread died"
+        );
+
+        let err = shared
+            .subscribe()
+            .expect_err("a subscription on a dead-watcher store must fail closed");
+        assert!(
+            err.contains("the update watcher has died"),
+            "the failure must name the dead watcher: {err}"
+        );
+        assert_eq!(
+            shared.subscriber_count(),
+            0,
+            "a rejected subscribe must enroll nothing"
+        );
+
+        shared.close().unwrap();
+        let _ = std::fs::remove_file(&tmp);
+        let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
+        let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
+    }
+
+    /// Review 004 item 3: a failed identity stat at watcher start must
+    /// leave the baseline *unknown*, not seeded with the `(0, 0)`
+    /// false identity — the first successful later stat captures the
+    /// baseline and the watcher survives, and only a later true
+    /// mismatch drives the death path. Driven through `run_poll_loop`
+    /// with the db file absent at start (the initial stat genuinely
+    /// fails on unix; on non-unix `stat_identity` never fails, so the
+    /// scenario cannot arise there and this test is unix-only), the
+    /// file then created and committed to (alive: the wake lands —
+    /// with the `(0, 0)` seeding the watcher would have died on the
+    /// first successful stat and no wake could arrive), and replaced
+    /// afterwards (the death path still fires on a real mismatch).
+    #[test]
+    #[cfg(unix)]
+    fn initial_stat_failure_does_not_seed_a_false_identity() {
+        let tmp = temp_db("identity-unknown");
+
+        let wakes = Arc::new(AtomicU64::new(0));
+        let wakes_t = wakes.clone();
+        let stop = Arc::new(AtomicBool::new(false));
+        let stop_t = stop.clone();
+        let dead = Arc::new(AtomicBool::new(false));
+        let dead_t = dead.clone();
+        let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1);
+        let db_path = tmp.clone();
+        let handle = std::thread::spawn(move || {
+            run_poll_loop(
+                db_path,
+                move || {
+                    wakes_t.fetch_add(1, Ordering::Relaxed);
+                },
+                stop_t,
+                dead_t,
+                ready_tx,
+                Duration::from_millis(5),
+                open_watcher_conn,
+            )
+        });
+        ready_rx
+            .recv_timeout(Duration::from_secs(2))
+            .expect("watcher loop baseline capture");
+
+        // The file is absent: several identity checks fail; the
+        // baseline must stay unknown (never seeded with (0, 0)).
+        std::thread::sleep(Duration::from_millis(300));
+        assert!(
+            !dead.load(Ordering::Acquire),
+            "identity-unknown must not drive the death path"
+        );
+
+        // First successful stat: the baseline is captured. The watcher
+        // survives it — a later commit's wake lands. (Wakes counted
+        // *before* the snapshot include the transient on_change calls
+        // the failing identity stats fired; the proof is relative to
+        // the post-recovery snapshot.)
+        wal_db(&tmp);
+        let writer = Connection::open(&tmp).unwrap();
+        writer
+            .execute("CREATE TABLE IF NOT EXISTS _test_ident(x INTEGER)", [])
+            .unwrap();
+
+        std::thread::sleep(Duration::from_millis(400));
+        assert!(
+            !dead.load(Ordering::Acquire),
+            "the first successful stat must not trigger the death path"
+        );
+        let before = wakes.load(Ordering::Relaxed);
+        writer
+            .execute("INSERT INTO _test_ident(x) VALUES (1)", [])
+            .unwrap();
+        let deadline = std::time::Instant::now() + Duration::from_secs(3);
+        while wakes.load(Ordering::Relaxed) == before && std::time::Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(5));
+        }
+        assert!(
+            wakes.load(Ordering::Relaxed) > before,
+            "the watcher must still be delivering wakes after the \
+             identity-unknown recovery"
+        );
+
+        // A real mismatch still drives the death path.
+        let tmp2 = temp_db("identity-replacement");
+        let _ = std::fs::File::create(&tmp2).unwrap();
+        std::fs::rename(&tmp2, &tmp).unwrap();
+        let deadline = std::time::Instant::now() + Duration::from_secs(3);
+        while !dead.load(Ordering::Acquire) && std::time::Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        assert!(
+            dead.load(Ordering::Acquire),
+            "a genuine identity mismatch must still drive the death path"
+        );
+
+        stop.store(true, Ordering::Release);
+        handle.join().unwrap();
+        let _ = std::fs::remove_file(&tmp);
+        let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
+        let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
+    }
 }

docs/architecture/engine-sqlite.md

@@ -97,7 +97,16 @@ per-contract obligations live in the core spec and are not restated here.
 - Watcher failure handling is inherited: on watcher death, every
   subscriber's receiver closes (`WatcherDeathGuard` behavior) —
   consumers see the close, never a silent hang
-  ([ADR-006](decisions/006-wake-and-delivery-contract.md)).
+  ([ADR-006](decisions/006-wake-and-delivery-contract.md)). New
+  subscriptions issued *after* a watcher death (a `listen()` or a
+  stream `subscribe()`) fail closed with the opaque `Error::Database`
+  fallback — they never enroll onto a dead feed to park forever
+  (review 004 item 2); recovery is a fresh store open. The watcher
+  also treats the db file's identity as unknown until its first
+  successful stat: a failed startup stat does not seed a false
+  identity that the first later stat would spuriously trip into the
+  death path (review 004 item 3 — a transient stat hiccup at start
+  closes nothing).
 - Wake coalescing: bursts inside one poll tick produce one wake;
   correctness is preserved by the re-read contract
   (POC-pinned: missed-wake stress with correct post-burst re-reads).

tasks/sqlite-fix-watcher-death-subscribe.md

@@ -1,7 +1,7 @@
 ---
 id: sqlite-fix-watcher-death-subscribe
 name: Dead-watcher subscribe hang + initial-stat self-kill (review 004 items 2+3)
-status: pending
+status: completed
 depends_on: []
 scope: moderate
 risk: medium
@@ -46,17 +46,17 @@ subscribers stay open.
 
 ## Acceptance Criteria
 
-- [ ] Subscribing (listen or stream) after watcher death fails closed
+- [x] Subscribing (listen or stream) after watcher death fails closed
       with a typed error instead of parking — pinned by test
-- [ ] Pre-death subscribers still close at watcher death — the
+- [x] Pre-death subscribers still close at watcher death — the
       existing pins green unchanged
-- [ ] An initial identity-stat failure no longer seeds a false
+- [x] An initial identity-stat failure no longer seeds a false
       identity: the first successful stat does not trigger the death
       path and subscribers stay open — pinned by injected-failure test
-- [ ] engine-sqlite.md's watcher-death posture text covers the
+- [x] engine-sqlite.md's watcher-death posture text covers the
       new-subscriber arm (and the identity-unknown state if it changes
       observable behavior)
-- [ ] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt
+- [x] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt
       clean; workspace gates green server-less
 
 ## References
@@ -71,6 +71,79 @@ subscribers stay open.
 
 > To be filled by implementation agent
 
+Decisions of record the description didn't pin:
+
+- The dead sign lives on `SharedUpdateWatcher` as an
+  `Arc<AtomicBool>`, set by `run_poll_loop` itself just before the
+  watcher-fatal return — it is set *before* the death-guard's senders
+  clear (which happens after the loop returns), so the guard-side
+  ordering is fixed by program order.
+- `SharedUpdateWatcher::subscribe` became fallible
+  (`Result<_, String>`) and checks **two** closed arms atomically with
+  the enrollment under the senders lock: the dead sign *and* the
+  taken watcher slot (the post-`close` enroll-onto-empty-watcher race
+  of the same shape). Failing closed on the taken slot is new
+  belt-and-braces beyond the finding's shape; the store's `closed`
+  flag already guarded the normal path. Rationale for checking under
+  the lock: a check-then-enroll done outside the lock could interleave
+  between the death and the guard's clear and enroll a ghost.
+- The substrate stays contract-blind: the failure is a `String`
+  naming the death, mapped into the taxonomy engine-side —
+  `Error::Database` (the opaque fallback with the source chain), the
+  same shape the engine's closed-store checks use. `Error::Closed`
+  was rejected: the contract docs scope that variant to *receiver*
+  closed arms (`try_recv`/`recv_timeout`), not entry points.
+- The initial-identity fix is an in-place re-derivation of the same
+  code (`Option<(u64, u64)>` baseline vs. the `(0, 0)` seed); the
+  death-log message keeps its precise diagnostic (now printing the
+  captured baseline).
+- The injected-failure test for (b) is the substrate-level drive
+  (`run_poll_loop` with the db file absent at start — the initial
+  stat genuinely fails; on non-unix `stat_identity` never fails, so
+  the scenario cannot arise and the test is unix-gated like the other
+  file-replacement drives); no `cfg(test)` injection seam was added to
+  `stat_identity` itself. Aliveness after the first successful stat is
+  proven by wake delivery relative to a pre-insert snapshot (failing
+  identity stats fire `on_change`, so absolute wake counts are not a
+  clean liveness signal).
+- Also updated: engine-sqlite.md's watcher-failure posture bullet
+  (both arms), notify.rs/stream.rs/store.rs module docs, open_tests'
+  three direct `watcher.subscribe()` call sites (now `.unwrap()`d),
+  and one PROVENANCE.md row (D-39 — the register wants divergence to
+  the ported watcher recorded even though the task text didn't name it).
+
 ## Summary
 
 > To be filled on completion
+
+Both defects landed with tests:
+
+- (a) Post-watcher-death subscribe hang: the poll loop sets the dead
+  sign before exiting through the identity-replacement death path;
+  `SharedUpdateWatcher::subscribe` fails closed on it (and on a taken
+  watcher slot) atomically with enrollment under the senders lock;
+  `notify.rs::listen` and `stream.rs::spawn_bridge` map the failure to
+  `Error::Database` with the death named in the source chain. Pinned
+  by four tests: substrate-level
+  (`subscribe_fails_closed_after_watcher_death`, unix file-replacement
+  drive) and engine-level (`listen_after_watcher_death_fails_closed`,
+  `subscribe_after_watcher_death_fails_closed` — same drive through
+  the store surface, error-shape + no-enrollment assertions). The
+  pre-death subscriber close-at-death pins
+  (`watcher_death_closes_the_receiver_terminally`,
+  `watcher_death_closes_the_subscription_terminally`, and the
+  substrate death-signal tests) are green unchanged — the death-signal
+  substrate test also now pins the dead sign being set.
+- (b) Initial-stat self-kill: the loop tracks the identity baseline as
+  `Option<(u64, u64)>`; a failed startup stat leaves it unknown, the
+  first successful stat captures it without driving death, and a later
+  genuine mismatch still kills. Pinned by
+  `initial_stat_failure_does_not_seed_a_false_identity`
+  (`run_poll_loop` drive: absent file → unknown baseline → file
+  created → no death + wake delivery lands → replacement → death
+  fires; unix-gated).
+- Verified: `cargo build`, `cargo test` (workspace, all green),
+  `cargo test -p alkstore-sqlite` (201 tests), fuzz gate
+  (`cargo test --manifest-path fuzz/shared/Cargo.toml`), clippy
+  `-D warnings` clean, `cargo fmt --check` clean, rustdoc
+  `-D warnings` doc gate clean.