diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index d91ecc9..46a7de8 100644
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@@ -234,6 +234,8 @@ The one exception: an update carrying the **unknown** version (an unparseable AC
 
 **The store closes off the runtime.** However the watch ends — shutdown, stream close, or any error — `watch_applied` hands its store to a dedicated thread to drop, and waits up to 30 s for it (`close_off_runtime`). A store's destructor may block: fjall 3.1.x's `Database::drop` can deadlock joining its worker pool. Dropped on the runtime, that parked the only tokio worker of a `CPUQuota=100%` service and froze the whole process before it could act on the watch's error. A normal close finishes inside the wait, so the caller can reopen the same path; a hung close leaves its thread behind and the watch returns anyway. A std thread, not `spawn_blocking`, because runtime shutdown waits on blocking tasks (`a_store_that_hangs_on_close_cannot_freeze_the_runtime`, `the_store_is_closed_before_watch_applied_returns`).
 
+**Resuming in-process (`watch_applied_owned`).** Closing on every exit makes a watch error fatal to a caller whose readers hold the store (a fold's `FjallReader` keeps its database alive). Such a caller can't re-open the path while those handles live, so its only recovery was a process restart, which a NATS heartbeat miss forced. `watch_applied_owned` is the same loop without the close. Its `WatchEnd` carries the result, the applied cursor (committed with the store's data), and the store itself, intact on every exit. The caller resumes by calling again with the same store and `resume = applied`: nothing is re-applied out of order and nothing is skipped, and cursor expiry and its repair take the same path as a fresh start. A store that comes back `None` was lost (a flush task panicked), and only a fresh open recovers it. `watch_applied` is `watch_applied_owned` followed by `close_off_runtime` (`owned_watch_hands_the_store_back_and_resumes_in_process`).
+
 **Cursor-expiry repair.** <a id="cursor-expiry-repair"></a> The cursor expires at resume (`CursorExpired` from the `*_from` watch) or mid-watch (the live floor guard ends an All-scope watch with `CursorExpired` when retention overruns it). Both take one path. Correct means **every write minus every real delete** — not "whatever NATS still lists"; the two agree only when retention never evicts current values. The repair (`ExpiryRepair`, `src/repair.rs`) is chosen accordingly:
 
 - **Key-listing diff** (`Relist`; `Auto` on a bucket that keeps current values). The watch task hands the main loop the reader and waits for an ack; the main loop lists the bucket's live keys, sorting them externally (`KeySorter`: 64 MiB runs spilled to the restore source's scratch dir under `Auto`, else `TMPDIR`), merge-joins that against the fold's own key-ordered scan — so neither the listing nor the fold is held in memory, only the stale keys — and runs a synthetic `KvUpdate::Delete` (unknown version — never advances the cursor) through `parse`/`apply`/store for each key that vanished; only then does the full-scope re-list (`watch_all` / `watch_prefix` / `watch_prefixes`) start. That ack ordering is the invariant: a synthetic delete always precedes the re-list put for the same key, so delete-then-recreate during the gap converges. (With the listing taken by the main loop after its drain, `tests/model_repair.rs` shows axiom 6 alone already keeps it correct; the barrier is defense in depth.) **Refused** — the watch fails — when the watcher reports retention that evicts current values (`KvWatcher::retention`, read live from the stream config), or the fold remembers seeing it do so (below): there it deletes valid keys. `tests/eviction.rs` reproduced exactly that against a live server before the fix.
diff --git a/Cargo.lock b/Cargo.lock
index 6517acc..6dd135c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
 
 [[package]]
 name = "beyond-slipstream"
-version = "0.8.4"
+version = "0.8.5"
 dependencies = [
  "aho-corasick",
  "async-nats",
diff --git a/Cargo.toml b/Cargo.toml
index d7e1155..89f6deb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,7 +8,7 @@ rust-version = "1.92"
 
 [package]
 name = "beyond-slipstream"
-version = "0.8.4"
+version = "0.8.5"
 edition.workspace = true
 license.workspace = true
 rust-version.workspace = true
diff --git a/src/applied.rs b/src/applied.rs
index 942cb25..a1facce 100644
--- a/src/applied.rs
+++ b/src/applied.rs
@@ -226,6 +226,78 @@ impl Default for BatchConfig {
 // would either box the closures or force a single generic bundle, losing that.
 #[allow(clippy::too_many_arguments)]
 pub async fn watch_applied<U, S, P, A, O, R>(
+    watcher: Arc<dyn KvWatcher>,
+    scope: WatchScope,
+    resume: Option<WatchCursor>,
+    // How an expired cursor is repaired (see the function docs). Only
+    // consulted on expiry (and, for a restore, at a cursor-less start on a
+    // bucket that has already evicted current values) — the hot path never
+    // touches it.
+    repair: R,
+    store: Option<S>,
+    // `Some(rx)` arms an export-request arm in the select loop: each
+    // [`ExportRequest`] is handled between flushes (pending batch flushed
+    // first), so the exported artifact's cursor is the applied cursor (or,
+    // across a transiently failed store flush, the store's own lagging but
+    // self-consistent cursor — never a cursor past unfolded data). The
+    // artifact's manifest records this watch's scope, which a restore checks.
+    // `None` (or dropping the paired sender) leaves the loop's behavior
+    // unchanged.
+    exports: Option<mpsc::Receiver<ExportRequest>>,
+    config: BatchConfig,
+    parse: P,
+    apply: A,
+    on_applied: O,
+    shutdown: watch::Receiver<bool>,
+) -> Result<WatchCursor, KvError>
+where
+    U: Send,
+    // `Send + 'static`: each flush moves `store` onto a blocking task to run its
+    // (potentially blocking) `apply`, then takes it back — the same offload the
+    // append log's compaction always used.
+    S: SnapshotStore + Send + 'static,
+    P: FnMut(&KvUpdate) -> Option<U> + Send,
+    A: FnMut(Vec<U>) + Send,
+    O: FnMut(WatchCursor) + Send,
+    R: Into<ExpiryRepair<S>>,
+{
+    let end = watch_applied_owned(
+        watcher, scope, resume, repair, store, exports, config, parse, apply, on_applied, shutdown,
+    )
+    .await;
+    if let Some(store) = end.store {
+        close_off_runtime(store).await;
+    }
+    end.result
+}
+
+/// How a [`watch_applied_owned`] run ended.
+pub struct WatchEnd<S> {
+    /// The run's result: the applied cursor on a clean end or a shutdown, the
+    /// error otherwise.
+    pub result: Result<WatchCursor, KvError>,
+    /// The applied cursor at the end, error or not: everything up to it was
+    /// applied and, with a store, committed with it. The `resume` for the next
+    /// run.
+    pub applied: WatchCursor,
+    /// The store, intact. `None` when none was given, or it was lost (a
+    /// flush task panicked).
+    pub store: Option<S>,
+}
+
+/// [`watch_applied`], handing the store back instead of closing it.
+///
+/// On every exit, a watch error included, the [`WatchEnd`] carries the store
+/// intact and the applied cursor it committed with (data and cursor commit
+/// together). A caller that keeps readers open on the store can therefore
+/// resume in-process: call again with the same store and `resume` set to
+/// [`WatchEnd::applied`]. Nothing the store holds is re-applied out of order and
+/// nothing is skipped. A store that comes back `None` was lost (a flush task
+/// panicked); only a fresh open recovers from that. Closing a returned store
+/// is the caller's job: drop it off the runtime (fjall's `Database::drop` can
+/// block; see [`watch_applied`]).
+#[allow(clippy::too_many_arguments)]
+pub async fn watch_applied_owned<U, S, P, A, O, R>(
     watcher: Arc<dyn KvWatcher>,
     scope: WatchScope,
     resume: Option<WatchCursor>,
@@ -249,7 +321,7 @@ pub async fn watch_applied<U, S, P, A, O, R>(
     apply: A,
     on_applied: O,
     mut shutdown: watch::Receiver<bool>,
-) -> Result<WatchCursor, KvError>
+) -> WatchEnd<S>
 where
     U: Send,
     // `Send + 'static`: each flush moves `store` onto a blocking task to run its
@@ -270,14 +342,27 @@ where
     // store to run it against (both repairs edit the fold). The watch task
     // sends the repair here and waits for the ack/reply before starting its
     // next watch.
-    let (repair, mut repairs) = repair::arm(repair.into(), store.is_some())?;
+    let start = resume.clone().unwrap_or_else(WatchCursor::none);
+    let early = |e: KvError, store: Option<S>| WatchEnd {
+        result: Err(e),
+        applied: start.clone(),
+        store,
+    };
+    let (repair, mut repairs) = match repair::arm(repair.into(), store.is_some()) {
+        Ok(armed) => armed,
+        Err(e) => return early(e, store),
+    };
 
     // A fold with data but no cursor can't be re-listed blind (see
     // `repair::run_watch`).
     let unanchored = match &store {
-        Some(st) if resume.as_ref().is_none_or(WatchCursor::is_none) => st
-            .has_entries()
-            .map_err(|e| KvError::WatchError(format!("reading the fold at start failed: {e}")))?,
+        Some(st) if resume.as_ref().is_none_or(WatchCursor::is_none) => match st.has_entries() {
+            Ok(has) => has,
+            Err(e) => {
+                let e = KvError::WatchError(format!("reading the fold at start failed: {e}"));
+                return early(e, store);
+            }
+        },
         _ => false,
     };
 
@@ -405,10 +490,11 @@ where
         }
     }
     .await;
-    if let Some(store) = fold.take_store() {
-        close_off_runtime(store).await;
+    WatchEnd {
+        result,
+        applied: fold.applied().clone(),
+        store: fold.take_store(),
     }
-    result
 }
 
 /// How long [`watch_applied`] waits for its store to close before returning
@@ -1400,6 +1486,116 @@ mod tests {
         assert_eq!(snap.entries["node.a"].value, b"1b");
     }
 
+    /// A watch that dies mid-stream (a NATS heartbeat miss) hands its store
+    /// back with the cursor it committed, and a second run on the same store,
+    /// resumed from that cursor, folds what changed during the gap: the caller
+    /// never has to close the store and re-open it.
+    #[tokio::test]
+    async fn owned_watch_hands_the_store_back_and_resumes_in_process() {
+        struct DiesAfter(Mutex<Option<Vec<KvUpdate>>>);
+        #[async_trait]
+        impl KvWatcher for DiesAfter {
+            async fn watch_all(&self, tx: Sender<KvUpdate>) -> Result<(), KvError> {
+                let updates = self.0.lock().unwrap().take().unwrap_or_default();
+                for u in updates {
+                    let _ = tx.send(u).await;
+                }
+                Err(KvError::WatchError(
+                    "watcher consumer error: missed idle heartbeat".into(),
+                ))
+            }
+            async fn watch_prefix(&self, _p: &str, tx: Sender<KvUpdate>) -> Result<(), KvError> {
+                self.watch_all(tx).await
+            }
+            async fn watch_prefixes(
+                &self,
+                _p: &[&str],
+                tx: Sender<KvUpdate>,
+            ) -> Result<(), KvError> {
+                self.watch_all(tx).await
+            }
+        }
+
+        let dir = tempfile::TempDir::new().unwrap();
+        let path = dir.path().join("owned.snap");
+        let (_r, store) = AppendLogSnapshot::open(&path, u64::MAX).unwrap();
+        let (_sd_tx, sd_rx) = watch::channel(false);
+
+        // Run 1: folds a@1 and b@2, then the watch dies.
+        let end = watch_applied_owned(
+            Arc::new(DiesAfter(Mutex::new(Some(vec![
+                put("node.a", b"1", 1),
+                put("node.b", b"2", 2),
+            ])))),
+            WatchScope::All,
+            None,
+            None::<Arc<dyn KvReader>>,
+            Some(store),
+            None,
+            BatchConfig::default(),
+            |_u: &KvUpdate| Some(()),
+            |_b: Vec<()>| {},
+            |_| {},
+            sd_rx.clone(),
+        )
+        .await;
+        assert!(
+            matches!(&end.result, Err(KvError::WatchError(m)) if m.contains("missed idle heartbeat")),
+            "the watch error is surfaced"
+        );
+        assert_eq!(
+            end.applied.as_u64(),
+            Some(2),
+            "everything delivered was applied"
+        );
+        let store = end.store.expect("the store comes back intact");
+        assert_eq!(store.get("node.a").unwrap().unwrap().value, b"1");
+
+        // Run 2, same store, resumed at the applied cursor: during the gap b
+        // was deleted and c written. The stream then closes cleanly.
+        let resumed = MockWatcher {
+            full: Mutex::new(Some(vec![])),
+            from: Mutex::new(Some(vec![
+                KvUpdate::Delete {
+                    key: "node.b".into(),
+                    version: VersionToken::from_u64(3),
+                },
+                put("node.c", b"3", 4),
+            ])),
+            from_expires: false,
+            hold: false,
+        };
+        let end = watch_applied_owned(
+            Arc::new(resumed),
+            WatchScope::All,
+            Some(end.applied),
+            None::<Arc<dyn KvReader>>,
+            Some(store),
+            None,
+            BatchConfig::default(),
+            |_u: &KvUpdate| Some(()),
+            |_b: Vec<()>| {},
+            |_| {},
+            sd_rx,
+        )
+        .await;
+        assert_eq!(end.result.unwrap().as_u64(), Some(4));
+        let store = end.store.expect("the store comes back on a clean end too");
+        assert!(
+            store.get("node.b").unwrap().is_none(),
+            "the gap's delete landed"
+        );
+        assert_eq!(store.get("node.c").unwrap().unwrap().value, b"3");
+        assert_eq!(store.get("node.a").unwrap().unwrap().value, b"1");
+        drop(store);
+        let snap = crate::snapshot::load(&path).unwrap().unwrap();
+        assert_eq!(
+            snap.cursor.as_u64(),
+            Some(4),
+            "the cursor committed with the data"
+        );
+    }
+
     /// A prefix-scoped resync diffs only in-scope keys: an out-of-scope key the
     /// fold holds survives, the in-scope stale key is deleted, and a flush
     /// containing only synthetic deletes leaves the cursor untouched.
diff --git a/src/lib.rs b/src/lib.rs
index 179cfe4..0a69642 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -42,7 +42,9 @@ mod stores;
 #[cfg(feature = "transport")]
 mod transport;
 
-pub use applied::{BatchConfig, ExportRequest, WatchScope, watch_applied};
+pub use applied::{
+    BatchConfig, ExportRequest, WatchEnd, WatchScope, watch_applied, watch_applied_owned,
+};
 pub use artifact::{ARTIFACT_SCHEMA_VERSION, ArtifactFile, ExportManifest, MANIFEST_FILE};
 pub use export_lease::{ExportLease, LeaseGuard, LeaseRecord};
 pub use kv::{