Merge pull request #2112 from DeusData/fix/decision-a-memory-budget-fail-whole

fix(pipeline,mcp,daemon): fail an over-budget index attempt whole with a named reason (#1997, #832)

src/daemon/application.c

@@ -181,7 +181,7 @@ struct cbm_daemon_application {
     cbm_daemon_application_update_ops_t update_ops;
     cbm_project_lock_manager_t *project_locks;
     size_t physical_job_limit;
-    size_t worker_memory_budget_bytes;
+    size_t aggregate_memory_budget_bytes;
     size_t active_mutations;
     size_t update_owners;
     cbm_daemon_application_update_worker_t update_worker;
@@ -693,6 +693,8 @@ static bool application_truncate_file(const char *path) {
 }
 
 static bool application_job_cancel_requested(cbm_daemon_application_job_t *job);
+static size_t application_worker_memory_slice_locked(cbm_daemon_application_t *application,
+                                                     size_t *active_jobs_out);
 
 static char **application_read_suspects(cbm_daemon_application_job_t *job, const char *path,
                                         int *count_out, bool *cancelled_out) {
@@ -1088,11 +1090,33 @@ static application_attempt_status_t application_job_run_attempt(cbm_daemon_appli
         return APPLICATION_ATTEMPT_CANCELLED;
     }
 
+    /* Decision 2 (#1997 #832): the slice is decided at spawn time from the jobs
+     * active right now (this job included), so a lone worker may use the whole
+     * aggregate and concurrent jobs split it. The divisor is logged so a
+     * fail-whole verdict can be read against the budget the worker really had. */
+    size_t active_jobs = 0;
+    cbm_mutex_lock(&application->mutex);
+    size_t memory_budget_bytes = application_worker_memory_slice_locked(application, &active_jobs);
+    size_t aggregate_memory_budget_bytes = application->aggregate_memory_budget_bytes;
+    cbm_mutex_unlock(&application->mutex);
+    if (memory_budget_bytes > 0) {
+        char active_text[32];
+        char aggregate_text[32];
+        char slice_text[32];
+        (void)snprintf(active_text, sizeof(active_text), "%zu", active_jobs);
+        (void)snprintf(aggregate_text, sizeof(aggregate_text), "%zu",
+                       aggregate_memory_budget_bytes / (1024U * 1024U));
+        (void)snprintf(slice_text, sizeof(slice_text), "%zu",
+                       memory_budget_bytes / (1024U * 1024U));
+        cbm_log_info("daemon.index.worker_budget", "project", job->project_key, "active_jobs",
+                     active_text, "aggregate_mb", aggregate_text, "slice_mb", slice_text);
+    }
+
     cbm_daemon_application_worker_t worker = NULL;
     application_tmp_lock();
-    int start_result = application->worker_ops.start(
-        application->worker_ops.context, job->args_json, application->worker_memory_budget_bytes,
-        marker_path, quarantine_path, &worker);
+    int start_result =
+        application->worker_ops.start(application->worker_ops.context, job->args_json,
+                                      memory_budget_bytes, marker_path, quarantine_path, &worker);
     application_tmp_unlock();
     if (start_result != 0 || !worker) {
         return application_job_cancel_requested(job) ? APPLICATION_ATTEMPT_CANCELLED
@@ -1564,6 +1588,22 @@ static size_t application_active_job_count_locked(cbm_daemon_application_t *appl
     return count;
 }
 
+/* Decision 2 (#1997 #832): a worker's memory slice is the aggregate budget
+ * divided by the jobs active at spawn time. The job being spawned is already
+ * in the table, so the divisor never drops below one; a zero aggregate means
+ * "no cap" and the worker falls back to its own RAM-fraction budget. */
+static size_t application_worker_memory_slice_locked(cbm_daemon_application_t *application,
+                                                     size_t *active_jobs_out) {
+    size_t active = application_active_job_count_locked(application);
+    if (active == 0) {
+        active = 1;
+    }
+    if (active_jobs_out) {
+        *active_jobs_out = active;
+    }
+    return application->aggregate_memory_budget_bytes / active;
+}
+
 /* Compare the effective index request, not its JSON spelling. yyjson's deep
  * equality treats object member order as insignificant, while the small
  * normalization below removes values that the index handler interprets as
@@ -2948,18 +2988,19 @@ cbm_daemon_application_t *cbm_daemon_application_new(
             application->ui_readiness_secret_set = true;
         }
     }
-    /* Equal fixed slices keep admission deterministic: starting fewer jobs does
-     * not let an early worker claim memory reserved for later concurrent jobs.
-     * The absurd sub-byte-per-slot case is made safe by reducing effective
-     * capacity before division; normal daemon budgets are many orders larger. */
+    /* The per-worker slice is decided at spawn time (see
+     * application_worker_memory_slice_locked): the aggregate divided by the
+     * jobs active then, so a lone worker may use the whole aggregate and
+     * concurrent jobs split it (decision 2, #1997 #832) — the former fixed
+     * aggregate/limit slice starved a lone job on hosts like #1864. The absurd
+     * sub-byte-per-slot case is still made safe by reducing effective capacity
+     * so every admitted job can receive at least one byte; normal daemon
+     * budgets are many orders larger. */
     if (aggregate_memory_budget_bytes > 0 &&
         application->physical_job_limit > aggregate_memory_budget_bytes) {
         application->physical_job_limit = aggregate_memory_budget_bytes;
     }
-    if (aggregate_memory_budget_bytes > 0 && application->physical_job_limit > 0) {
-        application->worker_memory_budget_bytes =
-            aggregate_memory_budget_bytes / application->physical_job_limit;
-    }
+    application->aggregate_memory_budget_bytes = aggregate_memory_budget_bytes;
     if (!application->worker_ops.start) {
         application->worker_ops = (cbm_daemon_application_worker_ops_t){
             .context = NULL,
@@ -3558,7 +3599,7 @@ size_t cbm_daemon_application_worker_memory_budget_bytes(cbm_daemon_application_
         return 0;
     }
     cbm_mutex_lock(&application->mutex);
-    size_t budget = application->worker_memory_budget_bytes;
+    size_t budget = application_worker_memory_slice_locked(application, NULL);
     cbm_mutex_unlock(&application->mutex);
     return budget;
 }

src/mcp/mcp.c

@@ -11209,6 +11209,49 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) {
         if (cbm_pipeline_had_format_migration(p)) {
             yyjson_mut_obj_add_bool(doc, root, "format_migration", true);
         }
+    } else if (rc == CBM_PIPELINE_ABORT_OVER_BUDGET) {
+        /* Decision A (#1997 #832): resident memory stayed above the budget
+         * after back-pressure and one confirmation cycle, so the run stopped
+         * before publication. Name the cause, the numbers and the fact that
+         * the previous index still serves — the generic "check repo_path"
+         * hint below sent people debugging a path that was fine. */
+        int budget_mb = (int)(cbm_mem_budget() / (1024 * 1024));
+        int peak_rss_mb = (int)(cbm_mem_peak_rss() / (1024 * 1024));
+        char budget_text[CBM_SZ_32];
+        char peak_text[CBM_SZ_32];
+        (void)snprintf(budget_text, sizeof(budget_text), "%d", budget_mb);
+        (void)snprintf(peak_text, sizeof(peak_text), "%d", peak_rss_mb);
+        cbm_log_error("index.abort", "reason", "over_memory_budget", "project", project_name,
+                      "budget_mb", budget_text, "peak_rss_mb", peak_text);
+        yyjson_mut_obj_add_str(doc, root, "status", "error");
+        yyjson_mut_obj_add_str(doc, root, "reason", "over_memory_budget");
+        yyjson_mut_obj_add_str(doc, root, "previous_index", "preserved");
+        yyjson_mut_obj_add_int(doc, root, "budget_mb", budget_mb);
+        yyjson_mut_obj_add_int(doc, root, "peak_rss_mb", peak_rss_mb);
+        yyjson_mut_obj_add_str(doc, root, "hint",
+                               "Indexing stopped: resident memory stayed above the budget after "
+                               "backpressure; no partial graph was published and the previous "
+                               "index still serves. Raise CBM_MEM_BUDGET_MB, lower CBM_WORKERS, "
+                               "or exclude large subtrees.");
+    } else if (rc == CBM_PIPELINE_ABORT_PRESERVE_DB) {
+        /* The truthful abort message (#2020): the old generic "check repo_path"
+         * hint sent people debugging a path that was fine, when the run
+         * aborted pre-publication (semantic inputs changed mid-run, or a
+         * discovery/manifest phase failed transiently) and the previous index
+         * is intact. A 99-test cascade on the Windows leg traced back to
+         * exactly this — the response said error, but not which kind. */
+        yyjson_mut_obj_add_str(doc, root, "status", "aborted_previous_preserved");
+        yyjson_mut_obj_add_str(doc, root, "hint",
+                               "Indexing aborted before publication; the previous index is "
+                               "intact and still serving. Causes: files changed while the run "
+                               "was in flight, or a discovery/manifest phase failed "
+                               "transiently. Retry; if it repeats, check the run log.");
+    } else if (rc == CBM_PIPELINE_PERSIST_FAILED) {
+        yyjson_mut_obj_add_str(doc, root, "status", "persist_failed");
+        yyjson_mut_obj_add_str(doc, root, "hint",
+                               "The validated staging database could not be published. Check "
+                               "free disk space and permissions on the cache directory; the "
+                               "previous index may have been rolled back.");
     } else {
         yyjson_mut_obj_add_str(doc, root, "status", "error");
         yyjson_mut_obj_add_str(doc, root, "hint",

src/pipeline/pass_parallel.c

@@ -30,7 +30,9 @@ enum {
     /* Extraction memory back-pressure: when the process is over its RSS budget,
      * a worker reclaims + naps before pulling another file so peers can finish
      * and return pages. Bounded spins avoid deadlock when the resident graph
-     * itself is near budget (then proceed with a soft overshoot). */
+     * itself is near budget: then ONE confirmation cycle decides — drained
+     * means continue, still over means the attempt fails whole (decision A,
+     * #1997 #832) instead of the former open-ended soft overshoot. */
     PP_BACKPRESSURE_MAX_SPINS = 40,
     PP_BACKPRESSURE_NAP_NS = 3000000, /* 3 ms */
 };
@@ -667,6 +669,11 @@ typedef struct {
      * While set, pulls skip the nap (the designed soft overshoot); the cheap
      * over-budget probe re-arms the gate once RSS drains under budget. */
     _Atomic int bp_futile;
+    /* Decision A (#1997 #832): set once futility was CONFIRMED by a second
+     * full cycle that still ended over budget. Every pull loop breaks on it
+     * and extract returns CBM_PIPELINE_ABORT_OVER_BUDGET. Deliberately not
+     * the shared cancel token — that one belongs to client cancellation. */
+    _Atomic int over_budget_abort;
 
     const CBMMacroTable *macro_table;            /* ObjectScript $$$macros (NULL if none) */
     const CBMReturnTypeTable *return_type_table; /* ObjectScript return types (NULL if none) */
@@ -722,6 +729,39 @@ static void log_extract_done(int pos, uint64_t ms, int defs, const char *path) {
     }
 }
 
+/* One back-pressure cycle: reclaim this thread's freed pages, then nap in
+ * bounded 3 ms steps while the process stays over budget. Counted for test
+ * observability. Returns true when the FULL cycle elapsed still over budget —
+ * the resident floor, not in-flight transients, holds the memory. */
+static bool pp_backpressure_cycle_still_over(extract_ctx_t *ec) {
+    cbm_mem_collect();
+    atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed);
+    int bp = 0;
+    for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() &&
+           !atomic_load_explicit(ec->cancelled, memory_order_relaxed) &&
+           !atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed);
+         bp++) {
+        struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS};
+        cbm_nanosleep(&nap, NULL);
+    }
+    return bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget();
+}
+
+/* Decision A (#1997 #832): back-pressure was futile and the confirmation
+ * cycle still ended over budget. The budget stops being advisory here: flag
+ * the attempt so every worker stops pulling and extract returns
+ * CBM_PIPELINE_ABORT_OVER_BUDGET. Cooperative on purpose — the failure must
+ * travel as a clean-exit worker response, never a kill, or the supervisor
+ * would quarantine innocent files. */
+static void pp_fail_whole_over_budget(extract_ctx_t *ec) {
+    if (atomic_exchange_explicit(&ec->over_budget_abort, 1, memory_order_relaxed) == 0) {
+        cbm_log_error("mem.budget.exceeded", "rss_mb",
+                      itoa_log((int)(cbm_mem_rss() / (1024 * 1024))), "budget_mb",
+                      itoa_log((int)(cbm_mem_budget() / (1024 * 1024))), "phase",
+                      "parallel_extract", "action", "fail_whole");
+    }
+}
+
 static void extract_worker(int worker_id, void *ctx_ptr) {
     extract_ctx_t *ec = ctx_ptr;
     extract_worker_state_t *ws = &ec->workers[worker_id];
@@ -739,7 +779,8 @@ static void extract_worker(int worker_id, void *ctx_ptr) {
             break;
         }
         cbm_scale_tick(&ec->scale, sort_pos);
-        if (atomic_load_explicit(ec->cancelled, memory_order_relaxed)) {
+        if (atomic_load_explicit(ec->cancelled, memory_order_relaxed) ||
+            atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed)) {
             break;
         }
 
@@ -750,39 +791,41 @@ static void extract_worker(int worker_id, void *ctx_ptr) {
          * near the budget instead of letting all workers parse their biggest
          * files at once. Self-disabling when the budget is unset (tests) or RSS
          * is under budget; bounded spins avoid deadlock when the resident graph
-         * is itself near budget (then proceed with a soft overshoot).
+         * is itself near budget.
          *
          * Futility latch: when a FULL nap cycle ends still over budget, the
          * resident floor — not transients — holds the memory; napping again on
          * the next pull cannot reclaim it and only idles workers (linux kernel:
-         * one full cycle per pull ≈ 390 s at 79% avg CPU). Latch bp_futile and
-         * proceed with the soft overshoot; the over-budget probe below re-arms
-         * the gate as soon as RSS drains under budget. */
+         * one full cycle per pull ≈ 390 s at 79% avg CPU). The worker that
+         * latches bp_futile (0→1) runs ONE confirmation cycle while its peers
+         * skip the nap. A confirmation that ends under budget re-arms the gate
+         * (RSS drained after all); one that still ends over budget fails the
+         * attempt whole (decision A, #1997 #832): every worker stops pulling,
+         * extract returns CBM_PIPELINE_ABORT_OVER_BUDGET, nothing is published
+         * and the previously serving index keeps answering. */
         if (cbm_mem_budget() > 0) {
             bool over = cbm_mem_over_budget();
             bool futile = atomic_load_explicit(&ec->bp_futile, memory_order_relaxed) != 0;
             if (over && !futile) {
-                cbm_mem_collect();
-                atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed);
-                int bp = 0;
-                for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() &&
-                       !atomic_load_explicit(ec->cancelled, memory_order_relaxed);
-                     bp++) {
-                    struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS};
-                    cbm_nanosleep(&nap, NULL);
-                }
-                if (bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget()) {
-                    /* Log only the 0→1 transition: all workers race into the
-                     * gate before anyone latches, so a plain store would WARN
-                     * once per worker (12 lines per latch event). */
-                    if (atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) {
-                        cbm_log_warn("mem.backpressure.futile", "action", "soft_overshoot");
+                /* Act only on the 0→1 transition: all workers race into the
+                 * gate before anyone latches, so a plain store would confirm
+                 * (and WARN) once per worker. */
+                if (pp_backpressure_cycle_still_over(ec) &&
+                    atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) {
+                    cbm_log_warn("mem.backpressure.futile", "action", "confirm");
+                    if (pp_backpressure_cycle_still_over(ec)) {
+                        pp_fail_whole_over_budget(ec);
+                    } else {
+                        atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed);
                     }
                 }
             } else if (!over && futile) {
                 atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed);
             }
         }
+        if (atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed)) {
+            break;
+        }
 
         int file_idx = ec->sorted[sort_pos].idx;
         const cbm_file_info_t *fi = &ec->files[file_idx];
@@ -1129,6 +1172,7 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file
     atomic_init(&ec.retain_cap_warned, 0);
     atomic_init(&ec.oversized_warned, 0);
     atomic_init(&ec.bp_futile, 0);
+    atomic_init(&ec.over_budget_abort, 0);
 
     /* Sub-phase: Dispatch workers (parse + extract per file, PARALLEL) */
     CBM_PROF_START(t_dispatch);
@@ -1176,6 +1220,12 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file
     free(sorted);
     cbm_macro_table_free(pp_macro_table); /* ObjectScript macro table (NULL-safe) */
 
+    /* The over-budget verdict outranks the cancel sentinel: a caller must be
+     * able to name the cause, and the orchestrator discards the staging DB on
+     * every non-zero code alike (the live generation is never touched). */
+    if (atomic_load(&ec.over_budget_abort)) {
+        return CBM_PIPELINE_ABORT_OVER_BUDGET;
+    }
     if (atomic_load(ctx->cancelled)) {
         return CBM_NOT_FOUND;
     }

src/pipeline/pipeline.h

@@ -63,9 +63,14 @@ void cbm_pipeline_free(cbm_pipeline_t *p);
  * need to know whether the PREVIOUS generation survived can distinguish the
  * failures by value: the run publishes by renaming a fully validated staging
  * database over the destination, so every abort before that rename leaves the
- * existing database in place. Those codes (CBM_PIPELINE_ABORT_PRESERVE_DB and
- * CBM_PIPELINE_PERSIST_FAILED) are defined in pipeline_internal.h alongside the
- * stages that raise them. */
+ * existing database in place. */
+#define CBM_PIPELINE_ABORT_PRESERVE_DB (-2) /* aborted pre-publication; previous DB intact */
+#define CBM_PIPELINE_PERSIST_FAILED (-4)    /* staging/rollback failure during publication */
+/* Resident memory stayed above the budget after back-pressure and one
+ * confirmation cycle: the run stops before publication, no partial graph is
+ * written, the previous DB is intact (#1997 #832). Distinct from the cancel
+ * sentinel so callers can name the cause instead of "pipeline failed". */
+#define CBM_PIPELINE_ABORT_OVER_BUDGET (-5)
 int cbm_pipeline_run(cbm_pipeline_t *p);
 
 /* Request cancellation of a running pipeline (thread-safe). */

src/pipeline/pipeline_internal.h

@@ -29,12 +29,12 @@
 /* Route node QN buffer size (must fit __route__METHOD__/full/url/path) */
 #define CBM_ROUTE_QN_SIZE 768
 
-/* Incremental integrity failure: abort the run and preserve the existing DB.
- * Distinct from CBM_NOT_FOUND, which the orchestrator uses as the normal
+/* CBM_PIPELINE_ABORT_PRESERVE_DB / CBM_PIPELINE_PERSIST_FAILED moved to
+ * pipeline.h — callers legitimately distinguish them (the header's contract
+ * always said so). FORCE_FULL_REINDEX never escapes the orchestrator and
+ * stays internal. All three are distinct from CBM_NOT_FOUND, the normal
  * "no incremental route; continue with a full index" sentinel. */
-#define CBM_PIPELINE_ABORT_PRESERVE_DB (-2)
 #define CBM_PIPELINE_FORCE_FULL_REINDEX (-3)
-#define CBM_PIPELINE_PERSIST_FAILED (-4)
 
 /* Canonicalize route-path parameter placeholders (":id", "{id}", "<id>",
  * "${...}") to a single "{}" token so that client call sites and server

tests/test_daemon_application.c

@@ -5245,12 +5245,157 @@ TEST(daemon_application_default_limit_admits_four_and_rejects_fifth) {
     ASSERT_EQ(atomic_load(&fake.starts), DEFAULT_CAP_RUNNING);
     ASSERT_EQ(atomic_load(&fake.cancels), DEFAULT_CAP_RUNNING);
     ASSERT_EQ(atomic_load(&fake.destroys), DEFAULT_CAP_RUNNING);
-    size_t assigned_budget = 0;
+    /* Decision 2 (#1997 #832): each worker's slice is the aggregate divided by
+     * the jobs active when IT was spawned. Four requests race through
+     * admission, so which divisor each saw is scheduling-dependent — but every
+     * slice is one of aggregate/1..4 and never exceeds the aggregate. */
     for (int i = 0; i < DEFAULT_CAP_RUNNING; i++) {
-        ASSERT_EQ(fake.memory_budgets[i], aggregate_budget / DEFAULT_CAP_RUNNING);
-        assigned_budget += fake.memory_budgets[i];
+        bool valid_slice = false;
+        for (size_t divisor = 1; divisor <= DEFAULT_CAP_RUNNING; divisor++) {
+            valid_slice = valid_slice || fake.memory_budgets[i] == aggregate_budget / divisor;
+        }
+        ASSERT_TRUE(valid_slice);
+        ASSERT_TRUE(fake.memory_budgets[i] <= aggregate_budget);
     }
-    ASSERT_TRUE(assigned_budget <= aggregate_budget);
+    PASS();
+}
+
+/* Decision 2 (#1997 #832): the worker slice is aggregate / jobs ACTIVE at
+ * spawn time. Sequenced admission makes the divisor deterministic: the first
+ * job spawns alone and receives the whole aggregate; the second spawns while
+ * the first still runs and receives half. */
+TEST(daemon_application_worker_slice_is_aggregate_over_active_jobs) {
+    const size_t aggregate_budget = 4099;
+    app_fake_worker_context_t fake;
+    app_fake_worker_context_init(&fake);
+    cbm_daemon_application_worker_ops_t worker_ops = {
+        .context = &fake,
+        .start = app_fake_worker_start,
+        .poll = app_fake_worker_poll,
+        .cancel = app_fake_worker_cancel,
+        .log_path = app_fake_worker_log_path,
+        .destroy = app_fake_worker_destroy,
+    };
+    cbm_daemon_application_config_t config = {
+        .worker_ops = &worker_ops,
+        .aggregate_memory_budget_bytes = aggregate_budget,
+    };
+    cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
+    char roots[2][APP_TEST_PATH_CAP];
+    bool roots_ok = true;
+    for (int i = 0; i < 2; i++) {
+        (void)snprintf(roots[i], sizeof(roots[i]), "%s/cbm-app-slice-%d-XXXXXX", cbm_tmpdir(), i);
+        roots_ok = roots_ok && cbm_mkdtemp(roots[i]) != NULL;
+    }
+    app_index_thread_t requests[2] = {
+        {.application = application, .project = "slice-first", .root = roots[0], .result = -1},
+        {.application = application, .project = "slice-second", .root = roots[1], .result = -1},
+    };
+    cbm_thread_t threads[2];
+    bool first_started = application && roots_ok &&
+                         cbm_thread_create(&threads[0], 0, app_index_thread, &requests[0]) == 0;
+    bool first_spawned = first_started && app_wait_for_atomic_int(&fake.starts, 1);
+    bool second_started =
+        first_spawned && cbm_thread_create(&threads[1], 0, app_index_thread, &requests[1]) == 0;
+    bool second_spawned = second_started && app_wait_for_atomic_int(&fake.starts, 2);
+    atomic_store(&fake.allow_completion, true);
+    if (first_started) {
+        (void)cbm_thread_join(&threads[0]);
+    }
+    if (second_started) {
+        (void)cbm_thread_join(&threads[1]);
+    }
+    bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
+    cbm_daemon_application_free(application);
+    for (int i = 0; i < 2; i++) {
+        (void)cbm_rmdir(roots[i]);
+    }
+
+    ASSERT_TRUE(roots_ok);
+    ASSERT_TRUE(first_spawned);
+    ASSERT_TRUE(second_spawned);
+    ASSERT_EQ(requests[0].result, 0);
+    ASSERT_EQ(requests[1].result, 0);
+    ASSERT_TRUE(stopped);
+    ASSERT_EQ(fake.memory_budgets[0], aggregate_budget);
+    ASSERT_EQ(fake.memory_budgets[1], aggregate_budget / 2);
+    PASS();
+}
+
+/* Decision A (#1997 #832): the over-budget verdict travels as a CLEAN worker
+ * exit carrying an error response. The daemon passes that response through
+ * verbatim on the first attempt — no crash/hang recovery loop, no quarantine
+ * of innocent files — because a healthy process reported an honest failure. */
+TEST(daemon_application_over_budget_response_passes_through_without_recovery) {
+    static const char over_budget_response[] =
+        "{\"content\":[{\"type\":\"text\",\"text\":\"{\\\"project\\\":\\\"budget\\\","
+        "\\\"status\\\":\\\"error\\\",\\\"reason\\\":\\\"over_memory_budget\\\","
+        "\\\"previous_index\\\":\\\"preserved\\\"}\"}],\"isError\":true}";
+    app_fake_worker_context_t fake;
+    app_fake_worker_context_init(&fake);
+    atomic_store(&fake.scripted, true);
+    fake.outcomes[0] = CBM_PROC_CLEAN;
+    fake.responses[0] = over_budget_response;
+    fake.outcomes[1] = CBM_PROC_CLEAN; /* a second start would be the recovery loop */
+    cbm_daemon_application_worker_ops_t worker_ops = {
+        .context = &fake,
+        .start = app_fake_worker_start,
+        .poll = app_fake_worker_poll,
+        .cancel = app_fake_worker_cancel,
+        .log_path = app_fake_worker_log_path,
+        .destroy = app_fake_worker_destroy,
+    };
+    cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
+    cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
+    cbm_daemon_runtime_application_callbacks_t callbacks =
+        cbm_daemon_application_runtime_callbacks(application);
+    cbm_daemon_runtime_application_session_t *session =
+        application ? app_test_open(&callbacks, 71) : NULL;
+    char root[APP_TEST_PATH_CAP];
+    snprintf(root, sizeof(root), "%s/cbm-app-over-budget-XXXXXX", cbm_tmpdir());
+    bool root_ok = cbm_mkdtemp(root) != NULL;
+    uint8_t *context = NULL;
+    uint32_t context_length = 0;
+    char args[APP_TEST_PATH_CAP + 32];
+    snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
+    uint8_t *tool = NULL;
+    uint32_t tool_length = 0;
+    bool setup = application && session && root_ok &&
+                 app_test_context_request(root, root, &context, &context_length) &&
+                 app_test_tool_request("index_repository", args, &tool, &tool_length);
+    uint8_t *response = NULL;
+    uint32_t response_length = 0;
+    if (setup) {
+        setup = app_test_request(&callbacks, session, context, context_length, &response,
+                                 &response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
+        free(response);
+        response = NULL;
+        response_length = 0;
+    }
+    cbm_daemon_runtime_application_status_t status =
+        setup
+            ? app_test_request(&callbacks, session, tool, tool_length, &response, &response_length)
+            : CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
+    bool passed_through = response && response_length > 0 &&
+                          strstr((char *)response, "over_memory_budget") != NULL &&
+                          strstr((char *)response, "\"isError\":true") != NULL;
+    free(response);
+    free(context);
+    free(tool);
+    if (session) {
+        callbacks.session_cancel(callbacks.context, session);
+        callbacks.session_close(callbacks.context, session);
+    }
+    bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
+    cbm_daemon_application_free(application);
+    (void)cbm_rmdir(root);
+
+    ASSERT_TRUE(setup);
+    ASSERT_EQ(status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
+    ASSERT_TRUE(passed_through);
+    ASSERT_EQ(atomic_load(&fake.starts), 1);
+    ASSERT_EQ(atomic_load(&fake.destroys), 1);
+    ASSERT_TRUE(stopped);
     PASS();
 }
 
@@ -5458,6 +5603,8 @@ SUITE(daemon_application) {
     RUN_TEST(daemon_application_thread_start_failure_rolls_back_job_reservation);
     RUN_TEST(daemon_application_queues_explicit_index_behind_physical_job_limit);
     RUN_TEST(daemon_application_default_limit_admits_four_and_rejects_fifth);
+    RUN_TEST(daemon_application_worker_slice_is_aggregate_over_active_jobs);
+    RUN_TEST(daemon_application_over_budget_response_passes_through_without_recovery);
     RUN_TEST(daemon_application_free_reports_retained_live_ownership);
     RUN_TEST(daemon_application_rejects_clean_exit_when_process_tree_is_not_contained);
 }

tests/test_incremental.c

@@ -107,7 +107,20 @@ static int reformat_files(const char *subdir, int max_files) {
 static char *index_repo(void) {
     char args[512];
     snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", g_repodir);
-    return cbm_mcp_handle_tool(g_srv, "index_repository", args);
+    char *resp = cbm_mcp_handle_tool(g_srv, "index_repository", args);
+    /* Fail-closed at the source: every caller asserts resp != NULL, but an
+     * error response (e.g. a transient pre-publication abort that preserves
+     * the previous index) used to slip through and cascade into dozens of
+     * mysterious count/query failures downstream — a 99-test pile-up on the
+     * Windows leg traced back to exactly this. Surface the real cause once,
+     * here, and let the existing non-NULL asserts fail loudly. */
+    if (resp && strstr(resp, "\"isError\":true")) {
+        fprintf(stderr, "  index_repo: index_repository returned an error response:\n  %.400s\n",
+                resp);
+        free(resp);
+        return NULL;
+    }
+    return resp;
 }
 
 /* Timed index: returns response, sets *elapsed_ms and *peak_rss_mb */

tests/test_mcp.c

@@ -9,6 +9,7 @@
 #include "../src/foundation/constants.h"
 #include "../src/foundation/log.h"
 #include "../src/foundation/platform.h" /* cbm_file_size */
+#include "../src/foundation/mem.h"      /* cbm_mem_set_budget_for_tests — over-budget seam */
 #include "../src/foundation/subprocess.h"
 #include "../src/foundation/workspace.h"
 #include "../src/mcp/compact_out.h"
@@ -17670,6 +17671,161 @@ static int idx823_supervised_name_override_check(const char *repo_dir, const cha
 }
 #endif
 
+/* Write the 64-file Go fixture (above MIN_FILES_FOR_PARALLEL, so the gated
+ * parallel extract path runs). generation 2 gives every file a second
+ * definition so a re-run must re-extract all of them — an unchanged repo is an
+ * incremental no-op and never reaches the memory gate. */
+static bool budget_fixture_write(const char *repo, int generation) {
+    for (int i = 0; i < 64; i++) {
+        char path[CBM_SZ_1K];
+        char body[CBM_SZ_256];
+        snprintf(path, sizeof(path), "%s/f%02d.go", repo, i);
+        if (generation == 1) {
+            snprintf(body, sizeof(body), "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n", i,
+                     i);
+        } else {
+            snprintf(body, sizeof(body),
+                     "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n\n"
+                     "func G%02d() int {\n\treturn F%02d() + 1\n}\n",
+                     i, i, i, i);
+        }
+        if (th_write_file(path, body) != 0) {
+            return false;
+        }
+    }
+    return true;
+}
+
+/* Parse the JSON object inside an MCP tool result. NULL when absent. */
+static yyjson_doc *budget_result_doc(const char *result) {
+    char *text = extract_text_content(result);
+    yyjson_doc *doc = text ? yyjson_read(text, strlen(text), 0) : NULL;
+    free(text);
+    return doc;
+}
+
+static const char *budget_doc_str(yyjson_doc *doc, const char *key) {
+    return doc ? yyjson_get_str(yyjson_obj_get(yyjson_doc_get_root(doc), key)) : NULL;
+}
+
+static int budget_doc_int(yyjson_doc *doc, const char *key, int missing) {
+    yyjson_val *val = doc ? yyjson_obj_get(yyjson_doc_get_root(doc), key) : NULL;
+    return val && yyjson_is_int(val) ? (int)yyjson_get_int(val) : missing;
+}
+
+/* Decision A (#1997 #832): an attempt that stays over the memory budget after
+ * back-pressure fails WHOLE and the response says so — reason, numbers, hint,
+ * and that the previous index still serves — instead of the generic "check
+ * repo_path" error. In-process: the runner is never a marked supervisor host,
+ * and the budget seam stands in for a 1 MiB machine. */
+TEST(index_repository_over_budget_reports_named_reason) {
+    char repo[CBM_SZ_1K];
+    char cache[CBM_SZ_1K];
+    snprintf(repo, sizeof(repo), "%s/cbm-mcp-budget-repo-XXXXXX", cbm_tmpdir());
+    snprintf(cache, sizeof(cache), "%s/cbm-mcp-budget-cache-XXXXXX", cbm_tmpdir());
+    if (!cbm_mkdtemp(repo)) {
+        FAIL("cbm_mkdtemp repo failed");
+    }
+    if (!cbm_mkdtemp(cache)) {
+        th_rmtree(repo);
+        FAIL("cbm_mkdtemp cache failed");
+    }
+    const char *saved_cache = getenv("CBM_CACHE_DIR");
+    char *saved_cache_copy = saved_cache ? strdup(saved_cache) : NULL;
+    const char *saved_workers = getenv("CBM_WORKERS");
+    char *saved_workers_copy = saved_workers ? strdup(saved_workers) : NULL;
+    cbm_setenv("CBM_CACHE_DIR", cache, 1);
+    cbm_setenv("CBM_WORKERS", "4", 1);
+
+    bool files_ok = budget_fixture_write(repo, 1);
+    cbm_mcp_server_t *srv = cbm_mcp_server_new(NULL);
+    bool had_server = srv != NULL;
+    char args[CBM_SZ_2K];
+    snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", repo);
+
+    /* Generation 1 at the process budget publishes. */
+    char *first = srv && files_ok ? cbm_mcp_handle_tool(srv, "index_repository", args) : NULL;
+    yyjson_doc *first_doc = budget_result_doc(first);
+    const char *first_status = budget_doc_str(first_doc, "status");
+    bool first_indexed = first_status && strcmp(first_status, "indexed") == 0;
+    int nodes_first = budget_doc_int(first_doc, "nodes", -1);
+    const char *first_project = budget_doc_str(first_doc, "project");
+    char *project = first_project ? strdup(first_project) : NULL;
+    yyjson_doc_free(first_doc);
+    free(first);
+    char db_path[CBM_SZ_2K];
+    snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project ? project : "missing");
+    long db_size_before = (long)cbm_file_size(db_path);
+
+    /* Generation 2 at 1 MiB over a changed repo fails whole. */
+    files_ok = files_ok && budget_fixture_write(repo, 2);
+    size_t saved_budget = cbm_mem_budget();
+    cbm_mem_set_budget_for_tests((size_t)1024 * 1024);
+    char *second = srv && files_ok ? cbm_mcp_handle_tool(srv, "index_repository", args) : NULL;
+    cbm_mem_set_budget_for_tests(saved_budget);
+    bool second_is_error = second && strstr(second, "\"isError\":true") != NULL;
+    yyjson_doc *second_doc = budget_result_doc(second);
+    const char *second_status = budget_doc_str(second_doc, "status");
+    const char *second_reason = budget_doc_str(second_doc, "reason");
+    const char *second_previous = budget_doc_str(second_doc, "previous_index");
+    const char *second_hint = budget_doc_str(second_doc, "hint");
+    bool status_error = second_status && strcmp(second_status, "error") == 0;
+    bool reason_named = second_reason && strcmp(second_reason, "over_memory_budget") == 0;
+    bool previous_preserved = second_previous && strcmp(second_previous, "preserved") == 0;
+    bool hint_names_knob = second_hint && strstr(second_hint, "CBM_MEM_BUDGET_MB") != NULL;
+    int budget_mb = budget_doc_int(second_doc, "budget_mb", -1);
+    int peak_rss_mb = budget_doc_int(second_doc, "peak_rss_mb", -1);
+    yyjson_doc_free(second_doc);
+    free(second);
+    long db_size_after = (long)cbm_file_size(db_path);
+
+    /* The previous generation still answers with its full graph. */
+    char status_args[CBM_SZ_2K];
+    snprintf(status_args, sizeof(status_args), "{\"project\":\"%s\"}", project ? project : "");
+    char *status = srv && project ? cbm_mcp_handle_tool(srv, "index_status", status_args) : NULL;
+    /* index_status answers in the tree format ("nodes: <n>" on its own line). */
+    char *status_text = extract_text_content(status);
+    const char *nodes_line = status_text ? strstr(status_text, "\nnodes: ") : NULL;
+    int nodes_after = nodes_line ? (int)strtol(nodes_line + strlen("\nnodes: "), NULL, 10) : -1;
+    if (nodes_after != nodes_first) {
+        printf("    index_status after the failed attempt: %.400s\n",
+               status_text ? status_text : "(null)");
+    }
+    free(status_text);
+    free(status);
+
+    cbm_mcp_server_free(srv);
+    cleanup_project_db(cache, project);
+    free(project);
+    if (saved_workers_copy) {
+        cbm_setenv("CBM_WORKERS", saved_workers_copy, 1);
+    } else {
+        cbm_unsetenv("CBM_WORKERS");
+    }
+    free(saved_workers_copy);
+    restore_cache_dir(saved_cache_copy);
+    free(saved_cache_copy);
+    th_rmtree(cache);
+    th_rmtree(repo);
+
+    ASSERT_TRUE(files_ok);
+    ASSERT_TRUE(had_server);
+    ASSERT_TRUE(first_indexed);
+    ASSERT_GT(nodes_first, 0);
+    ASSERT_TRUE(second_is_error);
+    ASSERT_TRUE(status_error);
+    ASSERT_TRUE(reason_named);
+    ASSERT_TRUE(previous_preserved);
+    ASSERT_TRUE(hint_names_knob);
+    ASSERT_EQ(budget_mb, 1);
+    ASSERT_GT(peak_rss_mb, budget_mb);
+    /* Preserved on disk and still served: same file size, same node count. */
+    ASSERT_GT(db_size_before, 0L);
+    ASSERT_EQ(db_size_after, db_size_before);
+    ASSERT_EQ(nodes_after, nodes_first);
+    PASS();
+}
+
 TEST(index_repository_cli_name_override_issue823) {
 #ifdef _WIN32
     SKIP_PLATFORM("POSIX fork harness required to isolate supervisor host mark");
@@ -20238,6 +20394,7 @@ SUITE(mcp) {
     RUN_TEST(index_repository_relative_path_uses_explicit_session_root);
     RUN_TEST(index_repository_supervisor_uses_canonical_session_path);
     RUN_TEST(index_repository_cli_name_override_issue823);
+    RUN_TEST(index_repository_over_budget_reports_named_reason);
     RUN_TEST(index_supervisor_unsafe_clean_is_never_fallback_or_recovery);
     RUN_TEST(index_supervisor_gate_requires_marked_host_issue845);
     RUN_TEST(index_supervisor_start_failure_is_fail_closed_in_real_host);

tests/test_pipeline.c

@@ -12871,23 +12871,82 @@ TEST(pipeline_committed_counts_match_persisted) {
  * MIN_FILES_FOR_PARALLEL (50) — else the run routes sequential, the gate never
  * fires, and the test would pass vacuously (cycles==0). The engagement assert
  * below (cycles >= 1) is a hard guard against that regressing silently. */
-TEST(pipeline_backpressure_futile_nap_disengages) {
-    /* 64 tiny files: > MIN_FILES_FOR_PARALLEL (50) so the parallel path (and its
-     * back-pressure gate) actually runs; old-code cycles (~64) >> the bound. */
+/* Shared fixture for the back-pressure tests: 64 tiny Go files, above
+ * MIN_FILES_FOR_PARALLEL (50) so the parallel extract path — the only phase
+ * with a memory gate — is the one that runs. */
+static bool write_backpressure_fixture(void) {
     snprintf(g_tmpdir, sizeof(g_tmpdir), "/tmp/cbm_test_XXXXXX");
     if (!cbm_mkdtemp(g_tmpdir)) {
-        FAIL("failed to create temp dir");
+        return false;
     }
     for (int i = 0; i < 64; i++) {
         char path[512];
         snprintf(path, sizeof(path), "%s/f%02d.go", g_tmpdir, i);
         FILE *f = fopen(path, "w");
         if (!f) {
-            FAIL("failed to create fixture file");
+            return false;
         }
         fprintf(f, "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n", i, i);
         fclose(f);
     }
+    return true;
+}
+
+/* Rewrite every fixture file with a second definition. An unchanged repo
+ * routes incremental_manifest → incremental.noop (nothing is re-extracted, so
+ * no gate can fire); changing all 64 files makes the next run re-extract them
+ * all through the parallel path. */
+static bool mutate_backpressure_fixture(void) {
+    for (int i = 0; i < 64; i++) {
+        char path[512];
+        snprintf(path, sizeof(path), "%s/f%02d.go", g_tmpdir, i);
+        FILE *f = fopen(path, "w");
+        if (!f) {
+            return false;
+        }
+        fprintf(f,
+                "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n\n"
+                "func G%02d() int {\n\treturn F%02d() + 1\n}\n",
+                i, i, i, i);
+        fclose(f);
+    }
+    return true;
+}
+
+/* Whole-file read for byte-identity assertions on a published database. */
+static unsigned char *read_file_bytes(const char *path, size_t *len_out) {
+    *len_out = 0;
+    FILE *f = fopen(path, "rb");
+    if (!f) {
+        return NULL;
+    }
+    long size = fseek(f, 0, SEEK_END) == 0 ? ftell(f) : -1;
+    if (size < 0) {
+        fclose(f);
+        return NULL;
+    }
+    rewind(f);
+    unsigned char *buf = malloc((size_t)size + 1);
+    if (!buf) {
+        fclose(f);
+        return NULL;
+    }
+    size_t got = fread(buf, 1, (size_t)size, f);
+    fclose(f);
+    if (got != (size_t)size) {
+        free(buf);
+        return NULL;
+    }
+    *len_out = got;
+    return buf;
+}
+
+TEST(pipeline_backpressure_futile_nap_disengages) {
+    /* 64 tiny files: > MIN_FILES_FOR_PARALLEL (50) so the parallel path (and its
+     * back-pressure gate) actually runs; old-code cycles (~64) >> the bound. */
+    if (!write_backpressure_fixture()) {
+        FAIL("failed to create fixture");
+    }
 
     /* 1 MB budget: over-budget on every pull, unreclaimable by napping.
      * Set via the test hook, NOT setenv + cbm_mem_init: the init-once guard
@@ -12942,7 +13001,10 @@ TEST(pipeline_backpressure_futile_nap_disengages) {
     teardown_test_repo();
 
     ASSERT_EQ(restore_workers_rc, 0);
-    ASSERT_EQ(rc, 0);
+    /* Decision A (#1997 #832): a budget that never drains is not a soft
+     * overshoot any more — after the latch and ONE confirmation cycle the
+     * attempt fails whole with the named code (rc==0 was the advisory era). */
+    ASSERT_EQ(rc, CBM_PIPELINE_ABORT_OVER_BUDGET);
     /* Engagement guard (anti-vacuous): the gate must have actually run — the
      * parallel path taken and the 1 MB budget exceeded on every pull. cycles==0
      * means the fixture routed sequential (or the gate was compiled out) and
@@ -12951,8 +13013,9 @@ TEST(pipeline_backpressure_futile_nap_disengages) {
         FAIL("back-pressure gate never engaged (cycles==0) — fixture routed sequential?");
     }
     /* Futile napping must disengage: at most one in-flight cycle per worker
-     * plus a small margin, never one per file (64). */
-    long bound = TEST_WORKERS + 2;
+     * plus the single confirmation cycle and a small margin, never one per
+     * file (64). */
+    long bound = TEST_WORKERS + 3;
     if (cycles > bound) {
         char msg[128];
         snprintf(msg, sizeof(msg), "nap cycles %ld > bound %ld (gate re-paid per pull)", cycles,
@@ -12962,6 +13025,102 @@ TEST(pipeline_backpressure_futile_nap_disengages) {
     PASS();
 }
 
+/* Decision A (#1997 #832): once back-pressure is futile AND one confirmation
+ * cycle still ends over budget, the attempt fails WHOLE with a named code —
+ * no partial graph is published, no stage residue remains, and the previously
+ * serving generation is byte-identical. */
+TEST(pipeline_over_budget_after_futility_fails_whole_and_preserves_db) {
+    if (!write_backpressure_fixture()) {
+        FAIL("failed to create fixture");
+    }
+    enum { TEST_WORKERS = 4 };
+    const char *old_workers = getenv("CBM_WORKERS");
+    char *saved_workers = old_workers ? strdup(old_workers) : NULL;
+    if ((old_workers && !saved_workers) || cbm_setenv("CBM_WORKERS", "4", 1) != 0) {
+        free(saved_workers);
+        teardown_test_repo();
+        FAIL("failed to pin CBM_WORKERS");
+    }
+
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/budget.db", g_tmpdir);
+
+    /* Generation 1 at the process budget: publishes normally. */
+    cbm_pipeline_t *p = cbm_pipeline_new(g_tmpdir, db_path, CBM_MODE_FULL);
+    int rc_first = p ? cbm_pipeline_run(p) : -1;
+    char *project = p ? strdup(cbm_pipeline_project_name(p)) : NULL;
+    cbm_pipeline_free(p);
+    int nodes_before = -1;
+    bool valid_before = false;
+    cbm_store_t *live = rc_first == 0 && project ? cbm_store_open_path(db_path) : NULL;
+    if (live) {
+        valid_before = cbm_store_check_integrity(live);
+        nodes_before = cbm_store_count_nodes(live, project);
+        cbm_store_close(live);
+    }
+    size_t before_len = 0;
+    unsigned char *before = read_file_bytes(db_path, &before_len);
+
+    /* Generation 2 at a 1 MiB budget over a CHANGED repo (every file gains a
+     * definition, so the run must re-extract all 64): over budget on every
+     * probe and unreclaimable by napping, so futility latches and the
+     * confirmation cycle still ends over budget. */
+    bool mutated = mutate_backpressure_fixture();
+    size_t saved_budget = cbm_mem_budget();
+    cbm_mem_set_budget_for_tests((size_t)1024 * 1024);
+    bool over_at_start = cbm_mem_over_budget();
+    cbm_pp_bp_nap_cycles_reset();
+    p = cbm_pipeline_new(g_tmpdir, db_path, CBM_MODE_FULL);
+    int rc_second = p ? cbm_pipeline_run(p) : -1;
+    long cycles = cbm_pp_bp_nap_cycles();
+    /* Restore the caller-visible budget BEFORE any assertion. */
+    cbm_mem_set_budget_for_tests(saved_budget);
+    cbm_pipeline_free(p);
+
+    size_t after_len = 0;
+    unsigned char *after = read_file_bytes(db_path, &after_len);
+    int stage_count = count_generation_stage_artifacts(g_tmpdir, "budget.db");
+    int nodes_after = -1;
+    bool valid_after = false;
+    live = cbm_store_open_path(db_path);
+    if (live) {
+        valid_after = cbm_store_check_integrity(live);
+        nodes_after = project ? cbm_store_count_nodes(live, project) : -1;
+        cbm_store_close(live);
+    }
+    bool bytes_identical =
+        before && after && before_len == after_len && memcmp(before, after, before_len) == 0;
+    free(before);
+    free(after);
+    free(project);
+    int restore_workers_rc =
+        saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS");
+    free(saved_workers);
+    teardown_test_repo();
+
+    ASSERT_EQ(restore_workers_rc, 0);
+    ASSERT_EQ(rc_first, 0);
+    ASSERT_TRUE(valid_before);
+    ASSERT_GT(nodes_before, 0);
+    ASSERT_TRUE(before_len > 0);
+    ASSERT_TRUE(mutated);
+    ASSERT_TRUE(over_at_start);
+    /* The named failure — not the cancel sentinel, not success. */
+    ASSERT_EQ(rc_second, CBM_PIPELINE_ABORT_OVER_BUDGET);
+    /* Engagement guard (anti-vacuous): fail-whole needs the latching cycle AND
+     * the confirmation cycle; fewer means the gate never decided anything. */
+    if (cycles < 2) {
+        FAIL("back-pressure gate never confirmed futility (cycles<2) — fixture routed sequential?");
+    }
+    /* The serving generation is untouched: same bytes, still valid, same
+     * graph; and the failed attempt left no stage residue behind. */
+    ASSERT_TRUE(bytes_identical);
+    ASSERT_TRUE(valid_after);
+    ASSERT_EQ(nodes_after, nodes_before);
+    ASSERT_EQ(stage_count, 0);
+    PASS();
+}
+
 /* TS cross-registry test hooks (ts_lsp.c) — extern to avoid pulling the
  * tree-sitter-typed ts_lsp.h into this store-level test. */
 extern long cbm_ts_full_registry_builds(void);
@@ -13774,6 +13933,7 @@ SUITE(pipeline) {
     RUN_TEST(pipeline_run_null);
     /* Extraction back-pressure */
     RUN_TEST(pipeline_backpressure_futile_nap_disengages);
+    RUN_TEST(pipeline_over_budget_after_futility_fails_whole_and_preserves_db);
     /* Sequential cross-LSP shared registry (ms-typescript quadratic) */
     RUN_TEST(pipeline_seq_ts_cross_uses_shared_registry);
     /* File persistence */