fix(pipeline): fail closed on incremental hash errors

Propagate incremental file-hash persistence failures instead of treating them as warning-only. The hash writer still attempts all current and mode-skipped rows so useful metadata is preserved where possible, but row or commit failures now make the incremental subpath return an error.

Add a public pipeline canary for injected hash-persist failure. The test verifies cbm_pipeline_run() falls back to a full rebuild, publishes the edited symbol, and leaves file-hash metadata available for the next incremental classification.

Validation: make -f Makefile.cbm build/c/test-runner; CBM_ONLY_SUITE=pipeline ./build/c/test-runner (227 passed); bash scripts/check-source-safety.sh; git diff --check; make -f Makefile.cbm cbm.
Signed-off-by: Andrew Hundt <ATHundt@gmail.com>

src/pipeline/pipeline_incremental.c

@@ -465,21 +465,22 @@ static void incr_free_edge_capture(cbm_edge_capture_t *cap) {
 /* Persist file hash rows for the current discovery and any mode-skipped
  * files preserved from the previous DB.
  *
- * Partial-failure policy: an `upsert` failure on any single row is logged
- * as a warning and the loop continues. We deliberately do NOT abort the
- * whole reindex on a single bad row — partial preservation is better than
- * total loss, and a transient failure on one file should not invalidate
- * the entire incremental update. The trade-off is that a silently-failed
- * row produces the same downstream effect as if the file were never
- * indexed at all (forced re-parse on the next run for current-files,
- * potential orphaned-node revival for mode_skipped). The warning surface
- * is the only signal that something went wrong. */
-static void persist_hashes(cbm_store_t *store, const char *project, cbm_file_info_t *files,
-                           int file_count, const cbm_file_hash_t *mode_skipped,
-                           int mode_skipped_count) {
+ * Partial-failure policy: continue writing all rows so one bad row does not
+ * discard useful metadata for the others, but return an error summary so the
+ * caller can fail closed instead of reporting a successful incremental run with
+ * stale classification metadata. */
+static int persist_hashes(cbm_store_t *store, const char *project, cbm_file_info_t *files,
+                          int file_count, const cbm_file_hash_t *mode_skipped,
+                          int mode_skipped_count) {
     int current_failed = 0;
     int ms_failed = 0;
 
+    if (incr_test_fail_phase_enabled(CBM_TEST_FAIL_INCREMENTAL_HASH_PERSIST)) {
+        cbm_log_error("incremental.err", "phase", CBM_TEST_FAIL_INCREMENTAL_HASH_PERSIST, "rc",
+                      itoa_buf_incr(CBM_STORE_ERR));
+        return CBM_STORE_ERR;
+    }
+
     /* Batch all hash upserts in one transaction: N files -> 1 COMMIT under
      * WAL instead of N autocommit fsyncs (a 10k-file reindex did 10k separate
      * commits). If BEGIN fails (e.g. store busy), fall back to per-row
@@ -530,13 +531,19 @@ static void persist_hashes(cbm_store_t *store, const char *project, cbm_file_inf
     }
 
     if (batched) {
-        (void)cbm_store_commit(store);
+        int commit_rc = cbm_store_commit(store);
+        if (commit_rc != CBM_STORE_OK) {
+            cbm_log_warn("incremental.persist_summary", "commit_failed", itoa_buf_incr(commit_rc));
+            return commit_rc;
+        }
     }
 
     if (current_failed > 0 || ms_failed > 0) {
         cbm_log_warn("incremental.persist_summary", "current_failed", itoa_buf_incr(current_failed),
                      "mode_skipped_failed", itoa_buf_incr(ms_failed));
+        return CBM_STORE_ERR;
     }
+    return CBM_STORE_OK;
 }
 
 /* ── Registry seed visitor ────────────────────────────────────────── */
@@ -772,7 +779,8 @@ static int dump_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *p
 
     cbm_store_t *hash_store = cbm_store_open_path(db_path);
     if (hash_store) {
-        persist_hashes(hash_store, project, files, file_count, mode_skipped, mode_skipped_count);
+        int hash_rc =
+            persist_hashes(hash_store, project, files, file_count, mode_skipped, mode_skipped_count);
 
         /* FTS5 rebuild after incremental dump.  The btree dump path bypasses
          * any triggers that could have kept nodes_fts synchronized, so we
@@ -789,6 +797,9 @@ static int dump_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *p
         }
 
         cbm_store_close(hash_store);
+        if (hash_rc != CBM_STORE_OK) {
+            return hash_rc;
+        }
     } else {
         cbm_log_error("incremental.err", "phase", "hash_store_open");
         return CBM_NOT_FOUND;

src/pipeline/pipeline_internal.h

@@ -59,6 +59,7 @@ int64_t cbm_pipeline_upsert_service_route(cbm_gbuf_t *gb, const char *path, cbm_
  * intentionally not user configuration. */
 #define CBM_TEST_FAIL_INCREMENTAL_PHASE "CBM_TEST_FAIL_INCREMENTAL_PHASE"
 #define CBM_TEST_FAIL_INCREMENTAL_POSTPASS "postpass"
+#define CBM_TEST_FAIL_INCREMENTAL_HASH_PERSIST "hash_persist"
 
 /* ── Pipeline context (internal) ─────────────────────────────────── */
 

tests/test_pipeline.c

@@ -5449,6 +5449,19 @@ static int pipeline_store_has_function_name(const char *db_path, const char *pro
     return found;
 }
 
+static int pipeline_store_file_hash_count(const char *db_path, const char *project) {
+    cbm_store_t *s = cbm_store_open_path_query(db_path);
+    if (!s) {
+        return CBM_STORE_ERR;
+    }
+    cbm_file_hash_t *hashes = NULL;
+    int count = 0;
+    int rc = cbm_store_get_file_hashes(s, project, &hashes, &count);
+    cbm_store_free_file_hashes(hashes, count);
+    cbm_store_close(s);
+    return rc == CBM_STORE_OK ? count : CBM_STORE_ERR;
+}
+
 /* ═══════════════════════════════════════════════════════════════════
  *  FastAPI Depends() edge tracking (PR #66, fix #27)
  * ═══════════════════════════════════════════════════════════════════ */
@@ -5946,6 +5959,51 @@ TEST(incremental_postpass_failure_keeps_existing_db) {
     PASS();
 }
 
+TEST(incremental_hash_persist_failure_falls_back_to_full) {
+    pipeline_env_snapshot_t fail_env = pipeline_env_save(CBM_TEST_FAIL_INCREMENTAL_PHASE);
+
+    if (setup_incremental_repo() != 0) {
+        FAIL("setup failed");
+    }
+
+    cbm_pipeline_t *p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(p);
+    ASSERT_EQ(cbm_pipeline_run(p), 0);
+    char *project = strdup(cbm_pipeline_project_name(p));
+    cbm_pipeline_free(p);
+    ASSERT_GTE(pipeline_store_file_hash_count(g_incr_dbpath, project), 2);
+    ASSERT(!pipeline_store_has_function_name(g_incr_dbpath, project, "NewFunc"));
+
+    char path[CBM_PATH_MAX];
+    snprintf(path, sizeof(path), "%s/helper.go", g_incr_tmpdir);
+    FILE *f = fopen(path, "w");
+    ASSERT_NOT_NULL(f);
+    fprintf(f, "package main\n\n"
+               "func Helper() string {\n\treturn \"hello\"\n}\n\n"
+               "func NewFunc() int {\n\treturn 42\n}\n");
+    fclose(f);
+
+    cbm_config_t *cfg = incremental_test_config(g_incr_tmpdir);
+    ASSERT_NOT_NULL(cfg);
+    p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(p);
+    cbm_pipeline_apply_config(p, cfg);
+
+    cbm_setenv(CBM_TEST_FAIL_INCREMENTAL_PHASE, CBM_TEST_FAIL_INCREMENTAL_HASH_PERSIST, 1);
+    int rc = cbm_pipeline_run(p);
+    pipeline_env_restore(&fail_env);
+
+    ASSERT_EQ(rc, 0);
+    ASSERT(pipeline_store_has_function_name(g_incr_dbpath, project, "NewFunc"));
+    ASSERT_GTE(pipeline_store_file_hash_count(g_incr_dbpath, project), 2);
+
+    cbm_pipeline_free(p);
+    free(project);
+    cbm_config_close(cfg);
+    cleanup_incremental_repo();
+    PASS();
+}
+
 TEST(incremental_detects_deleted_file) {
     /* Full index, delete a file, re-index → deleted file's nodes removed */
     if (setup_incremental_repo() != 0) {
@@ -7496,6 +7554,7 @@ SUITE(pipeline) {
     RUN_TEST(incremental_detects_changed_file);
     RUN_TEST(incremental_dump_failure_keeps_existing_db);
     RUN_TEST(incremental_postpass_failure_keeps_existing_db);
+    RUN_TEST(incremental_hash_persist_failure_falls_back_to_full);
     RUN_TEST(incremental_detects_deleted_file);
     RUN_TEST(incremental_new_file_added);
     RUN_TEST(incremental_fast_preserves_mode_skipped_tools_dir);