fix(artifact): snapshot exports consistently and refuse corrupt imports

The FAST export path (the watcher/incremental auto-update) read the
raw main-file bytes of a live WAL-mode store: committed rows still in
the -wal were missing entirely and mid-checkpoint reads produced torn
snapshots (field artifact: four short indexes + btreeInitPage error 11
across ~12 nodes-table pages). Import then installed the torn file
byte-for-byte — the integrity check only sanity-checked the projects
table.

- Export: both quality levels now snapshot via VACUUM INTO (the
  consistent copy); only the index-stripping remains BEST-only, FAST
  keeps zstd-3. The fast path pays a full-copy VACUUM INTO now —
  correctness over speed; a torn team-shared artifact costs far more.
- Import: new cbm_store_check_integrity_deep (shallow check + PRAGMA
  quick_check) — used only on this rare path, hot-path callers keep
  the cheap shallow check.

Reproduce-first: artifact_fast_export_snapshots_live_wal_store
exports FAST while the writer connection is open — RED on the old
code with 0 of 60 nodes surviving the round-trip (all rows lived in
the WAL); GREEN now. store_deep_integrity_detects_page_corruption
pins the import guard both directions (healthy passes, zero-banded
pages refused).

Closes #895

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>

src/pipeline/artifact.c

@@ -435,8 +435,12 @@ static const char *DROP_INDEXES_SQL = "DROP INDEX IF EXISTS idx_nodes_label;"
 /* ── Export helpers ───────────────────────────────────────────────── */
 
 /* Prepare a stripped DB copy for best-quality export.
- * VACUUM INTO → drop indexes → VACUUM. Returns malloc'd buffer or NULL. */
-static char *prepare_stripped_db(const char *db_path, size_t *out_size) {
+ * VACUUM INTO → (optionally) drop indexes → VACUUM. Returns malloc'd buffer
+ * or NULL. VACUUM INTO runs on BOTH quality levels: it is the consistent
+ * snapshot — the store runs in WAL mode, so raw main-file bytes miss
+ * committed transactions still in the -wal and can be mid-checkpoint torn
+ * (#895). Only the index-stripping is BEST-only. */
+static char *prepare_snapshot_db(const char *db_path, size_t *out_size, bool strip_indexes) {
     char tmp_path[CBM_SZ_4K];
     snprintf(tmp_path, sizeof(tmp_path), "%s/cbm_artifact_tmp.db", cbm_tmpdir());
     cbm_unlink(tmp_path);
@@ -464,12 +468,14 @@ static char *prepare_stripped_db(const char *db_path, size_t *out_size) {
         return NULL;
     }
 
-    /* Strip indexes from the copy for better compression. */
-    sqlite3 *tmp_db = NULL;
-    if (sqlite3_open_v2(tmp_path, &tmp_db, SQLITE_OPEN_READWRITE, NULL) == SQLITE_OK) {
-        sqlite3_exec(tmp_db, DROP_INDEXES_SQL, NULL, NULL, NULL);
-        sqlite3_exec(tmp_db, "VACUUM;", NULL, NULL, NULL);
-        sqlite3_close(tmp_db);
+    /* Strip indexes from the copy for better compression (BEST only). */
+    if (strip_indexes) {
+        sqlite3 *tmp_db = NULL;
+        if (sqlite3_open_v2(tmp_path, &tmp_db, SQLITE_OPEN_READWRITE, NULL) == SQLITE_OK) {
+            sqlite3_exec(tmp_db, DROP_INDEXES_SQL, NULL, NULL, NULL);
+            sqlite3_exec(tmp_db, "VACUUM;", NULL, NULL, NULL);
+            sqlite3_close(tmp_db);
+        }
     }
 
     char *data = read_file_alloc(tmp_path, out_size);
@@ -519,9 +525,12 @@ int cbm_artifact_export(const char *db_path, const char *repo_path, const char *
 
     if (quality == CBM_ARTIFACT_BEST) {
         compression_level = ART_ZSTD_BEST;
-        db_data = prepare_stripped_db(db_path, &db_size);
+        db_data = prepare_snapshot_db(db_path, &db_size, true);
     } else {
-        db_data = read_file_alloc(db_path, &db_size);
+        /* FAST keeps zstd-3 and its indexes, but still snapshots via
+         * VACUUM INTO: the raw main-file bytes of a live WAL store are a
+         * torn copy (#895). */
+        db_data = prepare_snapshot_db(db_path, &db_size, false);
     }
 
     if (!db_data || db_size == 0) {
@@ -689,8 +698,10 @@ int cbm_artifact_import(const char *repo_path, const char *cache_db_path) {
         return CBM_NOT_FOUND;
     }
 
-    /* Integrity check — refuse corrupted artifacts */
-    if (!cbm_store_check_integrity(store)) {
+    /* Deep integrity check — refuse corrupted artifacts. The shallow check
+     * only sanity-checks the projects table, so page-corrupted (torn)
+     * artifacts installed cleanly (#895); quick_check catches them. */
+    if (!cbm_store_check_integrity_deep(store)) {
         cbm_log_error("artifact.import", "err", "integrity_check_failed");
         cbm_store_close(store);
         cbm_unlink(tmp_path);

src/store/store.c

@@ -828,6 +828,31 @@ cbm_store_t *cbm_store_open_path_query(const char *db_path) {
 
 /* ── Integrity check ───────────────────────────────────────────── */
 
+/* Deep integrity: the shallow check above only sanity-checks the projects
+ * table, so page-level corruption (torn artifacts, #895) sails through.
+ * quick_check(1) walks the btrees and stops at the first error. Used on
+ * rare paths only (artifact import) — cost is proportional to DB size. */
+bool cbm_store_check_integrity_deep(cbm_store_t *s) {
+    if (!cbm_store_check_integrity(s)) {
+        return false;
+    }
+    sqlite3_stmt *stmt = NULL;
+    if (sqlite3_prepare_v2(s->db, "PRAGMA quick_check(1);", CBM_NOT_FOUND, &stmt, NULL) !=
+        SQLITE_OK) {
+        return false;
+    }
+    bool ok = false;
+    if (sqlite3_step(stmt) == SQLITE_ROW) {
+        const char *res = (const char *)sqlite3_column_text(stmt, 0);
+        ok = res && strcmp(res, "ok") == 0;
+        if (!ok) {
+            (void)fprintf(stderr, "ERROR store.corrupt quick_check=%s\n", res ? res : "(null)");
+        }
+    }
+    sqlite3_finalize(stmt);
+    return ok;
+}
+
 bool cbm_store_check_integrity(cbm_store_t *s) {
     if (!s || !s->db) {
         return false;

src/store/store.h

@@ -215,6 +215,9 @@ const char *cbm_store_db_path(const cbm_store_t *s);
  * (projects table has correct types, no corruption indicators).
  * Returns false if corruption is detected — caller should delete and re-index. */
 bool cbm_store_check_integrity(cbm_store_t *s);
+/* Shallow check + PRAGMA quick_check — catches page-level corruption.
+ * O(db size); use on rare paths (artifact import), not hot opens. */
+bool cbm_store_check_integrity_deep(cbm_store_t *s);
 
 /* Open database for a named project in the default cache dir. */
 cbm_store_t *cbm_store_open(const char *project);

tests/test_artifact.c

@@ -443,7 +443,109 @@ TEST(artifact_repo_path_shell_safe_rejects_cmd_metachars_on_windows) {
     PASS();
 }
 
+/* #895: the FAST export path (watcher/incremental auto-update) read the
+ * raw main-file bytes of a live WAL-mode store — committed rows still in
+ * the -wal were missing and mid-checkpoint reads produced torn snapshots
+ * that imported as page-corrupted caches. Export must snapshot
+ * consistently (VACUUM INTO) on BOTH quality levels. */
+TEST(artifact_fast_export_snapshots_live_wal_store) {
+    setup_artifact_test();
+    enum { WAL_NODES = 60 };
+
+    /* Live store: rows committed but NOT checkpointed into the main file —
+     * exactly the state the watcher export runs against. */
+    cbm_store_t *s = cbm_store_open_path(g_db);
+    ASSERT_NOT_NULL(s);
+    cbm_store_upsert_project(s, "test-proj", "/tmp/test");
+    for (int i = 0; i < WAL_NODES; i++) {
+        char name[64];
+        char qn[128];
+        snprintf(name, sizeof(name), "walnode_%03d", i);
+        snprintf(qn, sizeof(qn), "test-proj.mod.%s", name);
+        cbm_node_t n = {.project = "test-proj",
+                        .label = "Function",
+                        .name = name,
+                        .qualified_name = qn,
+                        .file_path = "mod.py",
+                        .start_line = i + 1,
+                        .end_line = i + 2};
+        ASSERT_TRUE(cbm_store_upsert_node(s, &n) > 0);
+    }
+
+    /* Export WHILE the writer connection is still open. */
+    ASSERT_EQ(cbm_artifact_export(g_db, g_repo, "test-proj", CBM_ARTIFACT_FAST), 0);
+    cbm_store_close(s);
+
+    char import_db[1024];
+    snprintf(import_db, sizeof(import_db), "%s/imported_wal.db", g_tmpdir);
+    ASSERT_EQ(cbm_artifact_import(g_repo, import_db), 0);
+
+    cbm_store_t *imp = cbm_store_open_path(import_db);
+    ASSERT_NOT_NULL(imp);
+    /* Torn snapshot = the WAL-resident rows are missing. */
+    ASSERT_EQ(cbm_store_count_nodes(imp, "test-proj"), WAL_NODES);
+    cbm_store_close(imp);
+    PASS();
+}
+
+/* #895 (import half): page-level corruption must be refused at import.
+ * The shallow integrity check only sanity-checks the projects table; the
+ * deep variant runs PRAGMA quick_check and catches corrupt pages. */
+TEST(store_deep_integrity_detects_page_corruption) {
+    setup_artifact_test();
+    enum { DEEP_NODES = 800, PAGE = 4096, ZERO_PAGES = 10 };
+    char db2[1024];
+    snprintf(db2, sizeof(db2), "%s/deep.db", g_tmpdir);
+    cbm_store_t *s = cbm_store_open_path(db2);
+    ASSERT_NOT_NULL(s);
+    cbm_store_upsert_project(s, "deep", "/tmp/deep");
+    for (int i = 0; i < DEEP_NODES; i++) {
+        char name[64];
+        char qn[192];
+        snprintf(name, sizeof(name), "deep_probe_%04d", i);
+        snprintf(qn, sizeof(qn), "deep.rather.long.module.path.for.page.fill.%s_pad_pad_pad",
+                 name);
+        cbm_node_t n = {.project = "deep",
+                        .label = "Function",
+                        .name = name,
+                        .qualified_name = qn,
+                        .file_path = "deep.py",
+                        .start_line = i + 1,
+                        .end_line = i + 2};
+        ASSERT_TRUE(cbm_store_upsert_node(s, &n) > 0);
+    }
+    cbm_store_close(s);
+
+    /* Healthy file passes the deep check. */
+    cbm_store_t *ok = cbm_store_open_path(db2);
+    ASSERT_NOT_NULL(ok);
+    ASSERT_TRUE(cbm_store_check_integrity_deep(ok));
+    cbm_store_close(ok);
+
+    /* Zero a mid-file band and the deep check must refuse. */
+    FILE *f = fopen(db2, "rb+");
+    ASSERT_NOT_NULL(f);
+    (void)fseek(f, 0, SEEK_END);
+    long pages = ftell(f) / PAGE;
+    ASSERT_TRUE(pages > ZERO_PAGES + 6);
+    char zero[PAGE];
+    memset(zero, 0, sizeof(zero));
+    (void)fseek(f, (pages / 2) * (long)PAGE, SEEK_SET);
+    for (int i = 0; i < ZERO_PAGES; i++) {
+        ASSERT_EQ(fwrite(zero, 1, PAGE, f), (size_t)PAGE);
+    }
+    (void)fclose(f);
+
+    cbm_store_t *bad = cbm_store_open_path(db2);
+    ASSERT_NOT_NULL(bad);
+    ASSERT_FALSE(cbm_store_check_integrity_deep(bad));
+    cbm_store_close(bad);
+    PASS();
+}
+
 SUITE(artifact) {
+    RUN_TEST(artifact_fast_export_snapshots_live_wal_store);
+    RUN_TEST(store_deep_integrity_detects_page_corruption);
     RUN_TEST(artifact_repo_path_shell_safe_accepts_plain_and_spaced);
     RUN_TEST(artifact_repo_path_shell_safe_rejects_injection);
     RUN_TEST(artifact_repo_path_shell_safe_rejects_cmd_metachars_on_windows);