diff --git a/tests/persistence/loaded_index_growth.rs b/tests/persistence/loaded_index_growth.rs
new file mode 100644
index 0000000..f6ab4e2
--- /dev/null
+++ b/tests/persistence/loaded_index_growth.rs
@@ -0,0 +1,171 @@
+use std::collections::BTreeSet;
+use std::time::Duration;
+
+use tokio::time::timeout;
+
+use crate::remove_dir_if_exists;
+use worktable::prelude::PersistedWorkTable;
+use worktable::prelude::*;
+use worktable::worktable;
+
+/*
+ * Regression test: growing a LOADED table's primary index.
+ *
+ * Tables built up from empty grow their index files fine. Tables loaded from
+ * disk are mapped at their on-disk size, and on 2026-08-01 an insert that
+ * needed the primary index to grow past that size wrote past the mapping
+ * instead of extending it or refusing: SIGBUS mid-write, torn table, every
+ * subsequent open of the store dead. It killed two production tables the
+ * same day with the same signature — `primary.wt.idx` frozen at exactly the
+ * size it was loaded with (65536 in one table, 229376 in the other) while
+ * `.wt.data` kept growing.
+ *
+ * The schema mirrors the table that died: String primary key of uuid shape,
+ * a String secondary index with heavy duplication, and a ~1K payload so the
+ * data:index ratio stays honest.
+ */
+worktable!(
+    name: LoadedIndexGrowth,
+    persist: true,
+    columns: {
+        id: String primary_key,
+        project_id: String,
+        body: String,
+    },
+    indexes: {
+        project_idx: project_id,
+    },
+);
+
+/// Enough rows that the primary index spans several growth steps: the table
+/// that died at 65536 bytes of index held 442 rows of this key shape.
+const ROWS_BEFORE_RELOAD: u64 = 1_500;
+const ROWS_AFTER_RELOAD: u64 = 1_500;
+
+fn key(i: u64) -> String {
+    // Same length and shape as the uuid-suffixed ids the dead table held.
+    format!("msg-00000000-0000-4000-8000-{i:012}")
+}
+
+fn row(i: u64) -> LoadedIndexGrowthRow {
+    LoadedIndexGrowthRow {
+        id: key(i),
+        project_id: format!("proj-{:02}", i % 3),
+        body: "x".repeat(1_000),
+    }
+}
+
+fn primary_idx_size(dir: &str) -> u64 {
+    let path = format!("{dir}/{}/primary.wt.idx", LoadedIndexGrowthWorkTable::name_snake_case());
+    std::fs::metadata(&path)
+        .unwrap_or_else(|error| panic!("no primary index at {path}: {error}"))
+        .len()
+}
+
+/// Build a store, close it, LOAD it, and append until the primary index must
+/// grow. The bug makes the append phase die of SIGBUS the moment the loaded
+/// index's capacity is exhausted; fixed, the index file grows exactly as it
+/// does for a fresh table and every row stays addressable across one more
+/// reload.
+#[test]
+fn test_primary_index_grows_on_a_loaded_table() {
+    let dir = "tests/data/loaded_index_growth/persisted";
+    let config = DiskConfig::new_with_table_name(
+        dir,
+        LoadedIndexGrowthWorkTable::name_snake_case(),
+        LoadedIndexGrowthWorkTable::version(),
+    );
+
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    runtime.block_on(async {
+        remove_dir_if_exists(dir.to_string()).await;
+
+        // Phase 1: build from empty. Growth on this path has always worked;
+        // this phase exists to leave a store whose index is well past its
+        // initial allocation, so phase 2 cannot fit inside leftover headroom.
+        {
+            let engine = LoadedIndexGrowthPersistenceEngine::new(config.clone()).await.unwrap();
+            let table = LoadedIndexGrowthWorkTable::load(engine).await.unwrap();
+            for i in 0..ROWS_BEFORE_RELOAD {
+                table.insert(row(i)).unwrap();
+            }
+            timeout(Duration::from_secs(30), table.wait_for_ops())
+                .await
+                .expect("persistence stalled building the initial store");
+        }
+
+        let idx_when_loaded = primary_idx_size(dir);
+
+        // Phase 2: load the store and append the same volume again. The
+        // primary index MUST grow past the size it was mapped at; the bug
+        // kills the process right here instead.
+        {
+            let engine = LoadedIndexGrowthPersistenceEngine::new(config.clone()).await.unwrap();
+            let table = LoadedIndexGrowthWorkTable::load(engine).await.unwrap();
+            for i in ROWS_BEFORE_RELOAD..(ROWS_BEFORE_RELOAD + ROWS_AFTER_RELOAD) {
+                table
+                    .insert(row(i))
+                    .unwrap_or_else(|error| panic!("insert {i} into the loaded table was refused: {error:?}"));
+            }
+            timeout(Duration::from_secs(30), table.wait_for_ops())
+                .await
+                .expect("persistence stalled appending to the loaded store");
+        }
+
+        let idx_after_appends = primary_idx_size(dir);
+        assert!(
+            idx_after_appends > idx_when_loaded,
+            "the primary index never grew while loaded ({idx_when_loaded} -> \
+             {idx_after_appends} bytes): the workload no longer exercises \
+             growth-on-a-loaded-table, which is the whole regression"
+        );
+
+        // Phase 3: reload and hold the table to the exact id set, not a
+        // count. A torn-but-openable store is the failure mode this bug
+        // ships in production; every row must come back and stay
+        // addressable through the secondary index too.
+        {
+            let engine = LoadedIndexGrowthPersistenceEngine::new(config.clone()).await.unwrap();
+            let table = LoadedIndexGrowthWorkTable::load(engine).await.unwrap();
+
+            let expected: BTreeSet<String> = (0..ROWS_BEFORE_RELOAD + ROWS_AFTER_RELOAD).map(key).collect();
+            let got: BTreeSet<String> = table
+                .select_all()
+                .execute()
+                .unwrap()
+                .into_iter()
+                .map(|r| r.id)
+                .collect();
+            assert_eq!(
+                got, expected,
+                "rows lost or duplicated across grow-while-loaded and reload"
+            );
+
+            for project in 0..3u64 {
+                let per_project = table
+                    .select_by_project_id(format!("proj-{project:02}"))
+                    .execute()
+                    .unwrap()
+                    .len() as u64;
+                assert_eq!(
+                    per_project,
+                    (ROWS_BEFORE_RELOAD + ROWS_AFTER_RELOAD) / 3,
+                    "secondary index lost rows for proj-{project:02}"
+                );
+            }
+
+            // And the grown, reloaded table must still be writable: the
+            // production stores died on exactly this insert.
+            table.insert(row(ROWS_BEFORE_RELOAD + ROWS_AFTER_RELOAD)).unwrap();
+            timeout(Duration::from_secs(30), table.wait_for_ops())
+                .await
+                .expect("persistence stalled on the post-reload insert");
+        }
+    })
+}
diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs
index 67e293b..4a96533 100644
--- a/tests/persistence/mod.rs
+++ b/tests/persistence/mod.rs
@@ -7,10 +7,12 @@ mod concurrent;
 mod duplicate_key_index_reload;
 mod failure;
 mod index_page;
+mod loaded_index_growth;
 mod read;
 mod space_index;
 mod sync;
 mod toc;
+mod torn_shutdown;
 mod vacuum;
 
 #[cfg(feature = "s3-support")]
diff --git a/tests/persistence/torn_shutdown.rs b/tests/persistence/torn_shutdown.rs
new file mode 100644
index 0000000..faf397e
--- /dev/null
+++ b/tests/persistence/torn_shutdown.rs
@@ -0,0 +1,276 @@
+use std::collections::BTreeSet;
+use std::time::Duration;
+
+use tokio::time::timeout;
+
+use crate::remove_dir_if_exists;
+use worktable::prelude::PersistedWorkTable;
+use worktable::prelude::*;
+use worktable::worktable;
+
+/*
+ * Regression test: a store torn by an abrupt process death must fail CLEANLY
+ * at the next load.
+ *
+ * The persistence engine writes pages in place and multi-step operations are
+ * not atomic (a data page can be half-written while its index events are
+ * abandoned — the Drop impl on PersistenceTask says as much). A process that
+ * dies mid-write (SIGKILL, crash, quit without wait_for_ops) therefore
+ * leaves torn bytes on disk. Today those bytes are read back through
+ * `rkyv::access_unchecked`, so a torn store does not fail at load: it loads
+ * as garbage, and the garbage's wild relative pointers blow up later as a
+ * SIGBUS in whatever operation happens to walk them. On 2026-08-01 that
+ * pattern killed a production store four times: each session ended abruptly,
+ * each next session opened "fine" and died mid-append, and each death tore
+ * the store further.
+ *
+ * The test kills a writer child mid-write repeatedly and then loads the
+ * store. Acceptable outcomes at load or scan: Ok with a consistent prefix of
+ * the data, or Err naming corruption. Unacceptable: SIGBUS / UB-check abort,
+ * which is what `access_unchecked` turns torn bytes into.
+ */
+worktable!(
+    name: TornShutdown,
+    persist: true,
+    columns: {
+        id: String primary_key,
+        project_id: String,
+        body: String,
+    },
+    indexes: {
+        project_idx: project_id,
+    },
+);
+
+const DIR: &str = "tests/data/torn_shutdown/persisted";
+pub const WRITER_ENV: &str = "WT_TORN_SHUTDOWN_WRITER";
+
+fn key(i: u64) -> String {
+    format!("msg-00000000-0000-4000-8000-{i:012}")
+}
+
+fn row(i: u64) -> TornShutdownRow {
+    TornShutdownRow {
+        id: key(i),
+        project_id: format!("proj-{:02}", i % 3),
+        body: "x".repeat(1_000),
+    }
+}
+
+async fn open_table() -> TornShutdownWorkTable {
+    let config = DiskConfig::new_with_table_name(
+        DIR,
+        TornShutdownWorkTable::name_snake_case(),
+        TornShutdownWorkTable::version(),
+    );
+    let engine = TornShutdownPersistenceEngine::new(config).await.unwrap();
+    TornShutdownWorkTable::load(engine).await.unwrap()
+}
+
+/// The writer half, run as a child process: appends rows forever without
+/// ever draining, so a SIGKILL lands mid-write with high probability. Not a
+/// test of anything by itself; the env gate keeps it inert in normal runs.
+#[test]
+fn torn_shutdown_writer() {
+    let Ok(start) = std::env::var(WRITER_ENV) else {
+        return; // Normal test run: nothing to do.
+    };
+    let start: u64 = start.parse().unwrap();
+
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+    runtime.block_on(async {
+        let table = open_table().await;
+        for i in start.. {
+            // Errors are tolerated, aborts are not: the parent only checks
+            // how this process DIES, and it must die by the parent's signal,
+            // not by its own reading of what the last kill left behind.
+            let _ = table.insert(row(i));
+            if i % 64 == 0 {
+                tokio::time::sleep(Duration::from_millis(1)).await;
+            }
+        }
+    });
+}
+
+/// Kill a writer mid-write N times, then hold the survivors to account: the
+/// store must load and scan without dying of a signal, and every row it does
+/// return must be one the writers actually inserted.
+///
+/// Ignored because it FAILS today, by design: it is the executable repro for
+/// the open crash-consistency bug. Run it with
+/// `cargo test -- --ignored test_store_survives_torn_shutdowns`. Observed
+/// failure modes so far: a phantom all-zero row returned by the scan, and a
+/// load that dies inside page parsing (`data_bucket` `parse_general_header`).
+/// Un-ignore it the day the engine gets crash-consistent writes or
+/// validated-and-refusing loads.
+#[test]
+#[ignore = "executable repro for the open torn-shutdown crash-consistency bug"]
+fn test_store_survives_torn_shutdowns() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    runtime.block_on(async {
+        remove_dir_if_exists(DIR.to_string()).await;
+
+        // A clean base, so the writers start from a real store rather than
+        // an empty directory.
+        {
+            let table = open_table().await;
+            for i in 0..200 {
+                table.insert(row(i)).unwrap();
+            }
+            timeout(Duration::from_secs(30), table.wait_for_ops())
+                .await
+                .expect("persistence stalled building the base store");
+        }
+    });
+
+    // Tear the store: run the writer, kill it mid-write, several rounds.
+    // Each round loads the store the previous kill tore.
+    let exe = std::env::current_exe().unwrap();
+    for round in 0..5u64 {
+        let mut child = std::process::Command::new(&exe)
+            .arg("--exact")
+            .arg("persistence::torn_shutdown::torn_shutdown_writer")
+            .arg("--nocapture")
+            .env(WRITER_ENV, (1_000 + round * 10_000).to_string())
+            .stdout(std::process::Stdio::null())
+            .stderr(std::process::Stdio::piped())
+            .spawn()
+            .unwrap();
+        // Long enough to be mid-write, short enough to stay a unit test.
+        std::thread::sleep(Duration::from_millis(700));
+        let status = match child.try_wait().unwrap() {
+            // Still writing, as intended: kill it mid-flight.
+            None => {
+                child.kill().unwrap();
+                child.wait().unwrap()
+            }
+            /*
+             * Already dead without being killed: the previous round's tear
+             * took it down at load or insert. That is exactly the disease —
+             * fail here, with the child's stderr as the diagnosis.
+             */
+            Some(status) => {
+                let mut stderr = String::new();
+                use std::io::Read;
+                if let Some(mut pipe) = child.stderr.take() {
+                    let _ = pipe.read_to_string(&mut stderr);
+                }
+                panic!(
+                    "writer round {round} died on its own ({status}) instead of being \
+                     killed: the store the previous kill left behind is torn beyond \
+                     loading. Child stderr:\n{stderr}"
+                );
+            }
+        };
+        assert!(
+            !status.success(),
+            "the writer exited cleanly; it is meant to write until killed"
+        );
+    }
+
+    // The reckoning: load and scan the torn store IN THIS PROCESS. A clean
+    // Err from load would also be acceptable behavior for a torn store; what
+    // must not happen is the process dying of SIGBUS/UB, which is what
+    // unchecked access turns torn bytes into — and if this test dies here,
+    // that is the failure the harness reports.
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+    runtime.block_on(async {
+        let table = open_table().await;
+        let rows = table.select_all().execute().unwrap();
+        let legal_projects: BTreeSet<String> = (0..3).map(|p| format!("proj-{p:02}")).collect();
+        for row in &rows {
+            assert!(
+                row.id.starts_with("msg-00000000-0000-4000-8000-"),
+                "scan returned a row no writer ever inserted (id {:?}): torn bytes \
+                 were read as data",
+                &row.id[..row.id.len().min(60)]
+            );
+            assert!(
+                legal_projects.contains(&row.project_id),
+                "row {} carries project {:?}, which no writer ever wrote",
+                row.id,
+                row.project_id
+            );
+        }
+        // And the survivor must still accept writes and a drain.
+        table.insert(row(9_000_000)).unwrap();
+        timeout(Duration::from_secs(30), table.wait_for_ops())
+            .await
+            .expect("persistence stalled appending to the survivor store");
+    });
+}
+
+/// The clean-shutdown sibling: many short load-append-drain-close sessions,
+/// no kill anywhere, then one full scan. The production table that died had
+/// lived exactly this life — dozens of small sessions, each ended with a
+/// drained quit — so if this fails, the corruption needs no crash at all:
+/// the load-append path drifts on its own, one generation at a time.
+#[test]
+fn test_many_clean_sessions_stay_readable() {
+    const DIR: &str = "tests/data/torn_shutdown/clean_sessions";
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    runtime.block_on(async {
+        remove_dir_if_exists(DIR.to_string()).await;
+        let open = || async {
+            let config = DiskConfig::new_with_table_name(
+                DIR,
+                TornShutdownWorkTable::name_snake_case(),
+                TornShutdownWorkTable::version(),
+            );
+            let engine = TornShutdownPersistenceEngine::new(config).await.unwrap();
+            TornShutdownWorkTable::load(engine).await.unwrap()
+        };
+
+        // Forty generations of a handful of appends each: the shape of a
+        // long-lived store that is opened, written a little, and closed.
+        let mut next_id = 0u64;
+        for session in 0..40u64 {
+            let table = open().await;
+            for _ in 0..8 {
+                table
+                    .insert(row(next_id))
+                    .unwrap_or_else(|error| panic!("session {session}: insert {next_id} refused: {error:?}"));
+                next_id += 1;
+            }
+            timeout(Duration::from_secs(30), table.wait_for_ops())
+                .await
+                .unwrap_or_else(|_| panic!("session {session}: drain stalled"));
+        }
+
+        let table = open().await;
+        let got: BTreeSet<String> = table
+            .select_all()
+            .execute()
+            .unwrap()
+            .into_iter()
+            .map(|r| r.id)
+            .collect();
+        let expected: BTreeSet<String> = (0..next_id).map(key).collect();
+        assert_eq!(
+            got, expected,
+            "rows lost, duplicated, or invented across clean load-append-close generations"
+        );
+    })
+}