Merge pull request #2111 from DeusData/fix/1839-1864-stage-orphans

fix(pipeline): sweep orphaned/nested .db.stage files and stop the false invalid_existing_db route (#1839, #1864)

src/foundation/compat_fs.c

@@ -23,11 +23,13 @@
 #include <windows.h>
 
 #include <aclapi.h>
-#include <direct.h> /* _wmkdir */
-#include <errno.h>  /* errno for spawn-failure logging */
-#include <fcntl.h>  /* _O_RDONLY */
-#include <io.h>     /* _wunlink, _open_osfhandle, _close */
-#include <stdint.h> /* intptr_t */
+#include <direct.h>   /* _wmkdir */
+#include <errno.h>    /* errno for spawn-failure logging */
+#include <fcntl.h>    /* _O_RDONLY */
+#include <io.h>       /* _wunlink, _open_osfhandle, _close */
+#include <share.h>    /* _SH_DENYRW */
+#include <sys/stat.h> /* _S_IREAD */
+#include <stdint.h>   /* intptr_t */
 #include "foundation/log.h"
 #include "foundation/win_utf8.h"
 
@@ -588,6 +590,26 @@ int cbm_rmdir(const char *path) {
     return ret;
 }
 
+int cbm_lockfile_open(const char *path, bool create) {
+    wchar_t *wpath = cbm_path_to_wide(path);
+    if (!wpath) {
+        errno = EINVAL;
+        return -1;
+    }
+    int flags = _O_RDWR | _O_BINARY | _O_NOINHERIT | (create ? _O_CREAT : 0);
+    /* _SH_DENYRW: every other open of this file, from any process, fails
+     * with EACCES until this descriptor closes -- including at death. */
+    int fd = _wsopen(wpath, flags, _SH_DENYRW, _S_IREAD | _S_IWRITE);
+    free(wpath);
+    return fd;
+}
+
+void cbm_lockfile_close(int fd) {
+    if (fd >= 0) {
+        (void)_close(fd);
+    }
+}
+
 /* Build a properly-quoted Windows command line from an argv array.
  * Returns a heap-allocated wide string, or NULL on allocation failure.
  * Quoting follows the MSVC CRT convention: arguments containing spaces,
@@ -737,6 +759,7 @@ int cbm_exec_no_shell(const char *const *argv) {
 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <sys/file.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
 #include <unistd.h>
@@ -1043,6 +1066,34 @@ int cbm_rmdir(const char *path) {
     return rmdir(path);
 }
 
+int cbm_lockfile_open(const char *path, bool create) {
+    int flags = O_RDWR | O_CLOEXEC | O_NOFOLLOW | (create ? O_CREAT : 0);
+    int fd;
+    do {
+        fd = open(path, flags, S_IRUSR | S_IWUSR);
+    } while (fd < 0 && errno == EINTR);
+    if (fd < 0) {
+        return -1;
+    }
+    int rc;
+    do {
+        rc = flock(fd, LOCK_EX | LOCK_NB);
+    } while (rc != 0 && errno == EINTR);
+    if (rc != 0) {
+        int saved = errno;
+        (void)close(fd);
+        errno = saved;
+        return -1;
+    }
+    return fd;
+}
+
+void cbm_lockfile_close(int fd) {
+    if (fd >= 0) {
+        (void)close(fd);
+    }
+}
+
 int cbm_exec_no_shell(const char *const *argv) {
     if (!argv || !argv[0]) {
         return CBM_NOT_FOUND;

src/foundation/compat_fs.h

@@ -113,6 +113,18 @@ int cbm_canonical_path(const char *path, char *out, size_t out_sz);
 /* Delete an empty directory. Returns 0 on success. */
 int cbm_rmdir(const char *path);
 
+/* Exclusive lock file. Opens `path` and takes an exclusive lock the KERNEL
+ * releases on any process death -- POSIX open(O_CLOEXEC|O_NOFOLLOW) +
+ * flock(LOCK_EX|LOCK_NB); Windows _wsopen with _SH_DENYRW (deny every other
+ * open) and _O_NOINHERIT -- so ownership never outlives its holder and is
+ * never inherited by a spawned child. `create` false never creates the file.
+ * Returns the descriptor, or -1 with errno: EWOULDBLOCK/EAGAIN (POSIX) or
+ * EACCES (Windows sharing violation) when another holder is live, ENOENT
+ * when absent, otherwise the open error. Never use this on a SQLite file:
+ * on macOS an flock conflicts with SQLite's own fcntl byte locks. */
+int cbm_lockfile_open(const char *path, bool create);
+void cbm_lockfile_close(int fd);
+
 /* Open a file by UTF-8 path.
  * On Windows, converts to wide-char and calls _wfopen so paths with
  * non-ASCII characters (accents, CJK, etc.) are handled correctly.

src/pipeline/pipeline.c

@@ -37,7 +37,9 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6 };
 #include "foundation/compat_thread.h"
 #include "foundation/profile.h"
 #include "foundation/mem.h"
+#include "foundation/secure_random.h"
 
+#include <ctype.h>
 #include <errno.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -72,6 +74,8 @@ static atomic_bool g_persist_test_cancel_after_destination_prepare = false;
 static atomic_bool g_persist_test_fail_adr_capture = false;
 static cbm_pipeline_test_hook_fn g_persist_test_before_final_manifest = NULL;
 static void *g_persist_test_before_final_manifest_userdata = NULL;
+static cbm_pipeline_test_hook_fn g_persist_test_after_stage_created = NULL;
+static void *g_persist_test_after_stage_created_userdata = NULL;
 
 void cbm_pipeline_incremental_test_fail_after_stage_dump_once(void) {
     atomic_store(&g_persist_test_fail_after_stage_dump, true);
@@ -105,6 +109,29 @@ void cbm_pipeline_persist_test_run_before_final_manifest(void) {
     }
 }
 
+void cbm_pipeline_incremental_test_after_stage_created_once(cbm_pipeline_test_hook_fn hook,
+                                                            void *userdata) {
+    g_persist_test_after_stage_created = hook;
+    g_persist_test_after_stage_created_userdata = userdata;
+}
+
+/* Fired by create_staging_path() right after the stage's main file is created
+ * with O_EXCL -- and, in the current lock-before-visible ordering, after its
+ * sidecar lock is already held. A test hook installed here can run a
+ * concurrent sweep (another cbm_pipeline_run() against the same final_path) at
+ * this instant to prove the just-created stage survives it. Under the OLD
+ * create-then-lock ordering this was the unlocked TOCTOU window, so the same
+ * hook binds RED if that ordering ever regresses. */
+void cbm_pipeline_persist_test_run_after_stage_created(void) {
+    cbm_pipeline_test_hook_fn hook = g_persist_test_after_stage_created;
+    void *userdata = g_persist_test_after_stage_created_userdata;
+    g_persist_test_after_stage_created = NULL;
+    g_persist_test_after_stage_created_userdata = NULL;
+    if (hook) {
+        hook(userdata);
+    }
+}
+
 bool cbm_pipeline_persist_test_take_failure_after_stage_dump(void) {
     return atomic_exchange(&g_persist_test_fail_after_stage_dump, false);
 }
@@ -124,6 +151,8 @@ void cbm_pipeline_persist_test_reset_faults(void) {
     atomic_store(&g_persist_test_fail_adr_capture, false);
     g_persist_test_before_final_manifest = NULL;
     g_persist_test_before_final_manifest_userdata = NULL;
+    g_persist_test_after_stage_created = NULL;
+    g_persist_test_after_stage_created_userdata = NULL;
 }
 #endif
 
@@ -195,6 +224,15 @@ struct cbm_pipeline {
      * full rebuild, so the MCP response can surface the migration. */
     bool format_migration;
 
+    /* Recorded by cbm_pipeline_run for the staged run beneath it: whether
+     * the destination existed, and whether it was copied into the stage so
+     * that an incremental route has a real previous generation to work
+     * from. Without a copy the stage is the run's empty placeholder, and
+     * probing THAT for integrity is what reported every first index as
+     * "invalid_existing_db" (#1864). */
+    bool final_existed;
+    bool existing_generation;
+
     /* ADR (project_summaries) captured before a full-reindex DB delete, so it
      * can be restored after the rebuild. NULL when no ADR existed. Issue #516. */
     char *saved_adr;
@@ -1402,6 +1440,16 @@ static int capture_existing_adr(cbm_pipeline_t *p, const char *db_path) {
 static int try_incremental_or_delete_db(cbm_pipeline_t *p, cbm_file_info_t *files, int file_count,
                                         const cbm_file_hash_t *baseline_manifest,
                                         int baseline_count, bool force_full_on_mismatch) {
+    if (!p->existing_generation) {
+        /* Nothing to be incremental against: a first index, or a
+         * destination that could not be copied (already reported as
+         * backup_failed_full_rebuild). The stage is an empty placeholder,
+         * not a database, so it is not probed -- "invalid_existing_db"
+         * stays reserved for a real copy that fails its integrity check. */
+        cbm_log_info("pipeline.route", "path", "full", "reason",
+                     p->final_existed ? "existing_db_backup_failed" : "no_existing_db");
+        return CBM_PIPELINE_FORCE_FULL_REINDEX;
+    }
     char *db_path = resolve_db_path(p);
     if (!db_path) {
         return CBM_PIPELINE_FORCE_FULL_REINDEX;
@@ -1530,12 +1578,148 @@ static bool promote_mode_to_existing_coverage(cbm_pipeline_t *p) {
 /* Defined below, next to the other publication helpers. */
 static char *create_staging_path(const char *final_path);
 
+/* ── Stage ownership (#1839) ─────────────────────────────────────
+ *
+ * A stage used to be recognisable only by its name: the mkstemp descriptor
+ * was closed at once and nothing marked who was writing it. A worker killed
+ * mid-run (the daemon cancels with SIGTERM then SIGKILL after one second of
+ * grace, which a gigabyte backup or clone never finishes inside) left its
+ * full-size stage behind forever, and no later run could tell a dead stage
+ * from a live one -- so none tried.
+ *
+ * Ownership is now an exclusive kernel lock on the sidecar "<stage>.lock",
+ * held from minting until the stage is discarded or renamed into place. The
+ * kernel releases it on any death, so "can I take this lock?" is exactly
+ * "is this stage dead?" -- no pid, no mtime, no grace period. The lock lives
+ * on a sidecar rather than the stage itself because on macOS an flock on a
+ * file conflicts with SQLite's fcntl byte locks on that same file.
+ *
+ * The stage path is passed around as a plain string through publish and
+ * finalize, so the descriptor is kept in this per-process registry keyed by
+ * path, and released by the same helpers that remove the file. */
+typedef struct stage_owner {
+    char *stage_path;
+    int lock_fd;
+    struct stage_owner *next;
+} stage_owner_t;
+
+static stage_owner_t *g_stage_owners = NULL;
+static atomic_flag g_stage_owners_spin = ATOMIC_FLAG_INIT;
+
+static void stage_owners_lock(void) {
+    while (atomic_flag_test_and_set_explicit(&g_stage_owners_spin, memory_order_acquire)) {}
+}
+
+static void stage_owners_unlock(void) {
+    atomic_flag_clear_explicit(&g_stage_owners_spin, memory_order_release);
+}
+
+static char *stage_lock_sidecar_path(const char *stage_path) {
+    static const char suffix[] = ".lock";
+    size_t len = strlen(stage_path);
+    if (len > SIZE_MAX - sizeof(suffix)) {
+        return NULL;
+    }
+    char *sidecar = (char *)malloc(len + sizeof(suffix));
+    if (!sidecar) {
+        return NULL;
+    }
+    memcpy(sidecar, stage_path, len);
+    memcpy(sidecar + len, suffix, sizeof(suffix));
+    return sidecar;
+}
+
+int cbm_pipeline_stage_lock_hold(const char *stage_path) {
+    if (!stage_path) {
+        return -1;
+    }
+    char *sidecar = stage_lock_sidecar_path(stage_path);
+    if (!sidecar) {
+        return -1;
+    }
+    int fd = cbm_lockfile_open(sidecar, true);
+    free(sidecar);
+    return fd;
+}
+
+void cbm_pipeline_stage_lock_drop(const char *stage_path, int lock_fd) {
+    if (!stage_path || lock_fd < 0) {
+        return;
+    }
+    char *sidecar = stage_lock_sidecar_path(stage_path);
+    /* Close before unlinking: Windows refuses to delete an open file. The
+     * sidecar exists unlocked for that instant, but by every drop the stage
+     * itself is already gone (discarded or renamed), so there is nothing a
+     * sweeper could take from us. */
+    cbm_lockfile_close(lock_fd);
+    if (sidecar) {
+        (void)cbm_unlink(sidecar);
+        free(sidecar);
+    }
+}
+
+/* Record an already-held stage lock in the per-process owner table, keyed by
+ * path so publish/finalize/discard can release it later. On success the table
+ * owns lock_fd; on failure the caller still does and must drop it.
+ *
+ * The lock must ALREADY be held: create_staging_path() takes it before the
+ * stage's main file is created (so the file is never visible on disk without
+ * its lock), then hands the descriptor here. Re-taking the lock in this helper
+ * would self-conflict -- both flock() and Windows _SH_DENYRW deny a second
+ * acquire of the same sidecar even from this same process. */
+static bool stage_owner_adopt(const char *stage_path, int lock_fd) {
+    stage_owner_t *owner = (stage_owner_t *)malloc(sizeof(*owner));
+    char *path_copy = strdup(stage_path);
+    if (!owner || !path_copy) {
+        free(owner);
+        free(path_copy);
+        return false;
+    }
+    owner->stage_path = path_copy;
+    owner->lock_fd = lock_fd;
+    stage_owners_lock();
+    owner->next = g_stage_owners;
+    g_stage_owners = owner;
+    stage_owners_unlock();
+    return true;
+}
+
+/* Release ownership of a stage that no longer exists under this name. A path
+ * this process never registered is a no-op. */
+static void stage_owner_release(const char *stage_path) {
+    if (!stage_path) {
+        return;
+    }
+    stage_owner_t *found = NULL;
+    stage_owners_lock();
+    for (stage_owner_t **link = &g_stage_owners; *link; link = &(*link)->next) {
+        if (strcmp((*link)->stage_path, stage_path) == 0) {
+            found = *link;
+            *link = found->next;
+            break;
+        }
+    }
+    stage_owners_unlock();
+    if (!found) {
+        return;
+    }
+    cbm_pipeline_stage_lock_drop(found->stage_path, found->lock_fd);
+    free(found->stage_path);
+    free(found);
+}
+
+/* Remove a stage's main file and SQLite sidecars, keeping ownership. */
+static void remove_stage_files(const char *stage_path) {
+    (void)cbm_unlink(stage_path);
+    (void)cbm_remove_db_sidecars(stage_path);
+}
+
 static void discard_generation_stage(const char *stage_path) {
     if (!stage_path) {
         return;
     }
-    cbm_unlink(stage_path);
-    cbm_remove_db_sidecars(stage_path);
+    remove_stage_files(stage_path);
+    stage_owner_release(stage_path);
 }
 
 typedef struct {
@@ -1929,6 +2113,7 @@ int cbm_pipeline_finalize_staged_generation(char *stage_path, const char *final_
         discard_generation_stage(stage_path);
         return CBM_PIPELINE_PERSIST_FAILED;
     }
+    stage_owner_release(stage_path);
     cbm_log_info("finalize.timing", "block", "rename", "elapsed_ms",
                  itoa_buf((int)elapsed_ms(t_fin)));
     return 0;
@@ -2386,8 +2571,8 @@ static void cleanup_staging_db(const char *path) {
     if (!path) {
         return;
     }
-    (void)cbm_unlink(path);
-    (void)cbm_remove_db_sidecars(path);
+    remove_stage_files(path);
+    stage_owner_release(path);
 }
 
 static bool ensure_db_parent(const char *path) {
@@ -2415,12 +2600,56 @@ static bool ensure_db_parent(const char *path) {
     return ok;
 }
 
+/* Length of the path a stage was minted for: the input itself unless its
+ * basename has exactly the minted shape "<name>.stage.<6 alphanumerics>", in
+ * which case the root is <name>. The outer run rewrites the pipeline's db_path
+ * to its stage, so the inner publication (dump and delta clone) used to mint
+ * ITS stage from that stage: <db>.stage.A.stage.B, with -wal/-shm beside it
+ * (#1839). Minting from the root keeps every generation's stage a sibling of
+ * the live database. Only the exact minted shape is recognised: a database
+ * named "x.stage.y.db" is not a stage and keeps its full name. */
+enum { CBM_STAGE_SUFFIX_RANDOM_CHARS = 6 };
+static const char cbm_stage_marker[] = ".stage.";
+
+static bool stage_suffix_at(const char *tail) {
+    if (strncmp(tail, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) {
+        return false;
+    }
+    const char *random = tail + sizeof(cbm_stage_marker) - 1;
+    for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) {
+        if (!isalnum((unsigned char)random[i])) {
+            return false;
+        }
+    }
+    return random[CBM_STAGE_SUFFIX_RANDOM_CHARS] == '\0';
+}
+
+static size_t stage_root_length(const char *path) {
+    size_t len = strlen(path);
+    const size_t suffix_len = sizeof(cbm_stage_marker) - 1 + CBM_STAGE_SUFFIX_RANDOM_CHARS;
+    if (len <= suffix_len) {
+        return len;
+    }
+    size_t root_len = len - suffix_len;
+    /* The marker must sit inside the basename, never span a separator. */
+    for (size_t i = root_len; i < len; i++) {
+        if (path[i] == '/'
+#ifdef _WIN32
+            || path[i] == '\\'
+#endif
+        ) {
+            return len;
+        }
+    }
+    return stage_suffix_at(path + root_len) ? root_len : len;
+}
+
 static char *create_staging_path(const char *final_path) {
     if (!final_path) {
         return NULL;
     }
     static const char suffix[] = ".stage.XXXXXX";
-    size_t final_len = strlen(final_path);
+    size_t final_len = stage_root_length(final_path);
     if (final_len > SIZE_MAX - sizeof(suffix)) {
         return NULL;
     }
@@ -2440,17 +2669,79 @@ static char *create_staging_path(const char *final_path) {
     }
     memcpy(path, final_path, final_len);
     memcpy(path + final_len, suffix, sizeof(suffix));
-    int fd = cbm_mkstemp(path);
-    if (fd < 0) {
-        free(path);
-        return NULL;
-    }
-#ifdef _WIN32
-    _close(fd);
-#else
-    close(fd);
+    /* The six random chars sit directly after the ".stage." marker; each
+     * attempt overwrites the "XXXXXX" template in place. Alphanumerics only,
+     * matching stage_suffix_at()/stage_entry_stage_length() so the sweep
+     * recognises the minted name and its sidecars. */
+    char *random_at = path + final_len + (sizeof(cbm_stage_marker) - 1);
+    static const char alphabet[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
+
+    /* Lock BEFORE the stage becomes visible: take the sidecar lock first, then
+     * create the stage's main file with O_EXCL. The main file is therefore
+     * never present on disk without its lock already held, so a concurrent
+     * run's sweep_orphan_stages() against the same final_path can only ever
+     * find this stage lock-held -- it reaches the stage through the .lock
+     * sidecar too (stage_entry_stage_length() matches it), probes the lock,
+     * sees a live holder, and keeps it. That closes the old create->register
+     * window that let a racing sweep delete a live-but-unlocked stage (POSIX)
+     * or collide on the sidecar with EACCES (Windows) -- #2111's windows-guards
+     * red. A pre-lock-era orphan minted by an OLDER binary still carries no
+     * lock, so the sweep's ENOENT path still removes it (#1839 preserved).
+     *
+     * A minted suffix collides with an existing stage only about 1 in 62^6;
+     * retry a bounded number of times, the way mkstemp/mkdtemp do, then fail. */
+    for (int attempt = 0; attempt < 128; attempt++) {
+        unsigned char rnd[CBM_STAGE_SUFFIX_RANDOM_CHARS];
+        if (!cbm_secure_random(rnd, sizeof(rnd))) {
+            free(path);
+            errno = EIO;
+            return NULL;
+        }
+        for (size_t i = 0; i < sizeof(rnd); i++) {
+            random_at[i] = alphabet[rnd[i] % (sizeof(alphabet) - 1)];
+        }
+        errno = 0;
+        int lock_fd = cbm_pipeline_stage_lock_hold(path);
+        if (lock_fd < 0) {
+            /* A live twin already owns this exact suffix's sidecar (EAGAIN /
+             * EACCES), or the sidecar could not be created. Mint a fresh suffix
+             * and try again rather than contend for this one. */
+            continue;
+        }
+        FILE *main_file = cbm_fopen(path, "wbx");
+        if (!main_file) {
+            /* The suffix collided with a lock-less orphan's main file -- its
+             * sidecar was takeable, so it is not a live writer. Never inherit a
+             * stranger's bytes: drop the lock, remove the sidecar we just took,
+             * and mint a fresh suffix. The orphan's main file is left for a
+             * later sweep, which removes it as a pre-lock-era orphan. */
+            cbm_pipeline_stage_lock_drop(path, lock_fd);
+            continue;
+        }
+        (void)fclose(main_file);
+#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API
+        /* Main file now exists and its lock is already held. Under the OLD
+         * create-then-lock ordering this was the unlocked window; the
+         * concurrent-sweep test fires here to prove the stage now survives a
+         * racing sweep, and to bind RED if that ordering ever regresses. */
+        cbm_pipeline_persist_test_run_after_stage_created();
 #endif
-    return path;
+        if (!stage_owner_adopt(path, lock_fd)) {
+            cbm_pipeline_stage_lock_drop(path, lock_fd);
+            (void)cbm_unlink(path);
+            free(path);
+            return NULL;
+        }
+        return path;
+    }
+    /* Every attempt failed to take a lock -- keep the observability the old
+     * stage_owner_register() emitted for a lock failure. */
+    char errno_text[16];
+    (void)snprintf(errno_text, sizeof(errno_text), "%d", errno);
+    cbm_log_warn("pipeline.stage", "action", "lock_failed", "errno", errno_text, "path", path);
+    free(path);
+    errno = EEXIST;
+    return NULL;
 }
 
 /* A backup-failed destination may still have the only recoverable WAL or
@@ -2554,6 +2845,200 @@ static int export_after_publish(cbm_pipeline_t *p, const char *final_path) {
     return 0;
 }
 
+/* ── Orphan sweep (#1839) ────────────────────────────────────────
+ *
+ * Nothing on the worker-death path ever cleaned a stage up, so the sweep
+ * runs at the start of every run, before this run mints its own stage. It
+ * considers ONLY names of the exact minted shape for THIS database --
+ * "<basename>.stage.<6 alphanumerics>" plus that stage's -wal/-shm/-journal
+ * and .lock sidecars -- never the live database, a quarantined .corrupt, or
+ * another project's files. A stage is removed when its ownership lock can be
+ * taken (its writer is dead, or the stage predates ownership) and kept when
+ * a live writer holds the lock. The pre-ownership case is the one honest
+ * gap: a stage an OLDER binary is still writing against this database has
+ * no lock and is swept; that writer's final rename then fails and it
+ * discards. The live database is never named here on either path. */
+
+static const char *const cbm_stage_sidecar_tails[] = {"", "-wal", "-shm", "-journal", ".lock"};
+
+/* If `name` is "<base>.stage.<6 alphanumerics><known tail>", return the
+ * length of the stage name proper (without the tail); 0 otherwise. */
+static size_t stage_entry_stage_length(const char *name, const char *base, size_t base_len) {
+    if (strncmp(name, base, base_len) != 0) {
+        return 0;
+    }
+    const char *at = name + base_len;
+    if (strncmp(at, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) {
+        return 0;
+    }
+    at += sizeof(cbm_stage_marker) - 1;
+    for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) {
+        /* NUL is not alphanumeric, so a short name fails here too. */
+        if (!isalnum((unsigned char)at[i])) {
+            return 0;
+        }
+    }
+    at += CBM_STAGE_SUFFIX_RANDOM_CHARS;
+    for (size_t i = 0; i < sizeof(cbm_stage_sidecar_tails) / sizeof(cbm_stage_sidecar_tails[0]);
+         i++) {
+        if (strcmp(at, cbm_stage_sidecar_tails[i]) == 0) {
+            return (size_t)(at - name);
+        }
+    }
+    return 0;
+}
+
+typedef struct {
+    char **names;
+    int count;
+    int cap;
+} stage_name_list_t;
+
+/* Add a stage name once, however many of its files were listed. */
+static void stage_name_list_add(stage_name_list_t *list, const char *name, size_t len) {
+    for (int i = 0; i < list->count; i++) {
+        if (strlen(list->names[i]) == len && strncmp(list->names[i], name, len) == 0) {
+            return;
+        }
+    }
+    if (list->count == list->cap) {
+        int cap = list->cap ? list->cap * 2 : 8;
+        char **grown = (char **)realloc(list->names, (size_t)cap * sizeof(*grown));
+        if (!grown) {
+            return;
+        }
+        list->names = grown;
+        list->cap = cap;
+    }
+    char *copy = (char *)malloc(len + 1);
+    if (!copy) {
+        return;
+    }
+    memcpy(copy, name, len);
+    copy[len] = '\0';
+    list->names[list->count++] = copy;
+}
+
+static int64_t stage_bytes_on_disk(const char *stage_path) {
+    static const char *const files[] = {"", "-wal", "-shm", "-journal"};
+    int64_t total = 0;
+    char side[CBM_SZ_4K];
+    for (size_t i = 0; i < sizeof(files) / sizeof(files[0]); i++) {
+        int n = snprintf(side, sizeof(side), "%s%s", stage_path, files[i]);
+        if (n <= 0 || (size_t)n >= sizeof(side)) {
+            continue;
+        }
+        cbm_path_info_t info;
+        if (cbm_path_info_utf8(side, &info) == CBM_PATH_INFO_OK && info.is_regular) {
+            total += info.size;
+        }
+    }
+    return total;
+}
+
+static void sweep_one_stage(const char *stage_path) {
+    char *sidecar = stage_lock_sidecar_path(stage_path);
+    if (!sidecar) {
+        return;
+    }
+    errno = 0;
+    int lock_fd = cbm_lockfile_open(sidecar, false);
+    int probe_errno = errno;
+    free(sidecar);
+    if (lock_fd < 0 && probe_errno != ENOENT) {
+        bool live = probe_errno == EAGAIN || probe_errno == EACCES;
+#if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN
+        live = live || probe_errno == EWOULDBLOCK;
+#endif
+        if (live) {
+            cbm_log_info("pipeline.stage", "action", "orphan_kept", "reason", "live_writer", "path",
+                         stage_path);
+            return;
+        }
+        char errno_text[16];
+        (void)snprintf(errno_text, sizeof(errno_text), "%d", probe_errno);
+        cbm_log_warn("pipeline.stage", "action", "orphan_kept", "reason", "lock_probe_failed",
+                     "errno", errno_text, "path", stage_path);
+        return;
+    }
+    /* No owner: absent sidecar (an ENOENT probe) or a lock the kernel released
+     * with its writer. Ours now, from the lock down.
+     *
+     * The absent-sidecar (ENOENT) case is exactly a pre-lock-era orphan: a
+     * stage an OLDER binary minted with no sidecar at all (#1839 pins that
+     * these ARE swept). It is NOT an in-flight stage of a current run: since
+     * create_staging_path() now takes the sidecar lock BEFORE the stage's main
+     * file becomes visible on disk, a live stage always has its sidecar, so a
+     * concurrent sweep landing here for one would instead find the lock held
+     * above and keep it. Removing on ENOENT therefore reclaims genuine orphans
+     * without ever deleting a live stage (the create->register race behind
+     * #2111's windows-guards red is closed at the source). */
+    int64_t bytes = stage_bytes_on_disk(stage_path);
+    remove_stage_files(stage_path);
+    if (lock_fd >= 0) {
+        cbm_pipeline_stage_lock_drop(stage_path, lock_fd);
+    }
+    char bytes_text[32];
+    (void)snprintf(bytes_text, sizeof(bytes_text), "%lld", (long long)bytes);
+    cbm_log_info("pipeline.stage", "action", "orphan_removed", "bytes", bytes_text, "path",
+                 stage_path);
+}
+
+static void sweep_orphan_stages(const char *final_path) {
+    /* Directory part INCLUDING its trailing separator, so the stage paths
+     * are joined exactly as the final path was spelled. */
+    size_t prefix_len = 0;
+    for (const char *c = final_path; *c; c++) {
+        if (*c == '/'
+#ifdef _WIN32
+            || *c == '\\'
+#endif
+        ) {
+            prefix_len = (size_t)(c - final_path) + 1;
+        }
+    }
+    const char *base = final_path + prefix_len;
+    size_t base_len = strlen(base);
+    if (base_len == 0) {
+        return;
+    }
+    char *dir_path = prefix_len ? (char *)malloc(prefix_len + 1) : strdup(".");
+    if (!dir_path) {
+        return;
+    }
+    if (prefix_len) {
+        memcpy(dir_path, final_path, prefix_len);
+        dir_path[prefix_len] = '\0';
+    }
+    cbm_dir_t *dir = cbm_opendir(dir_path);
+    if (!dir) {
+        free(dir_path);
+        return;
+    }
+    stage_name_list_t list = {0};
+    cbm_dirent_t *entry;
+    while ((entry = cbm_readdir(dir)) != NULL) {
+        size_t stage_len = stage_entry_stage_length(entry->name, base, base_len);
+        if (stage_len) {
+            stage_name_list_add(&list, entry->name, stage_len);
+        }
+    }
+    cbm_closedir(dir);
+    for (int i = 0; i < list.count; i++) {
+        size_t name_len = strlen(list.names[i]);
+        char *stage_path = (char *)malloc(prefix_len + name_len + 1);
+        if (stage_path) {
+            memcpy(stage_path, final_path, prefix_len);
+            memcpy(stage_path + prefix_len, list.names[i], name_len + 1);
+            sweep_one_stage(stage_path);
+            free(stage_path);
+        }
+        free(list.names[i]);
+    }
+    free(list.names);
+    free(dir_path);
+}
+
 int cbm_pipeline_run(cbm_pipeline_t *p) {
     if (!p) {
         return CBM_NOT_FOUND;
@@ -2565,6 +3050,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
     }
     struct stat final_st;
     bool final_existed = stat(final_path, &final_st) == 0;
+    sweep_orphan_stages(final_path);
     char *staging_path = create_staging_path(final_path);
     if (!staging_path) {
         free(final_path);
@@ -2577,10 +3063,15 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
         if (!backup_succeeded) {
             cbm_log_warn("pipeline.stage", "action", "backup_failed_full_rebuild", "path",
                          final_path);
-            cleanup_staging_db(staging_path);
+            /* The copy is gone but the NAME stays ours: the rebuilt
+             * generation is renamed over it by the inner finalize and then
+             * published from it below, so its lock is held to the end. */
+            remove_stage_files(staging_path);
         }
     }
 
+    p->final_existed = final_existed;
+    p->existing_generation = final_existed && backup_succeeded;
     char *configured_db_path = p->db_path;
     p->db_path = strdup(staging_path);
     if (!p->db_path) {
@@ -2660,6 +3151,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
         return CBM_PIPELINE_PERSIST_FAILED;
     }
 
+    stage_owner_release(staging_path);
     rc = export_after_publish(p, final_path);
     free(staging_path);
     free(final_path);

src/pipeline/pipeline_internal.h

@@ -792,6 +792,17 @@ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_
  * executor; the dump path uses it internally). malloc'd, caller frees. */
 char *cbm_pipeline_create_staging_path(const char *final_path);
 
+/* Stage ownership (#1839). Every stage minted by cbm_pipeline_create_staging_path
+ * is owned through an exclusive kernel lock on the sidecar "<stage>.lock" for
+ * as long as the stage exists; the lock -- and the sidecar -- go away when the
+ * stage is discarded or renamed into place, and the kernel drops the lock
+ * when the writer dies. Hold/drop are the same primitive, exposed so a test
+ * can stand in for a live writer. hold returns a descriptor >= 0, or -1 when
+ * another holder is live or the sidecar cannot be created. drop releases the
+ * lock and unlinks the sidecar. */
+int cbm_pipeline_stage_lock_hold(const char *stage_path);
+void cbm_pipeline_stage_lock_drop(const char *stage_path, int lock_fd);
+
 /* ── Delta-repair staging primitives (pipeline_delta.c) ──────────
  * Closure-route-only subsystem: clone the live generation, patch exactly
  * the repaired node/edge set, publish through the shared finalize leg. */
@@ -897,6 +908,15 @@ void cbm_pipeline_incremental_test_fail_adr_capture_once(void);
 typedef void (*cbm_pipeline_test_hook_fn)(void *userdata);
 void cbm_pipeline_incremental_test_before_final_manifest_once(cbm_pipeline_test_hook_fn hook,
                                                               void *userdata);
+/* Fires from create_staging_path(), right after the stage's main file is
+ * created with O_EXCL. In the current lock-before-visible ordering its sidecar
+ * lock is already held at this point, so a test hook installed here can run a
+ * concurrent sweep (via another cbm_pipeline_run() against the same
+ * final_path) and confirm the just-created stage survives it. Under the OLD
+ * create-then-lock ordering this was the unlocked window, so the hook also
+ * binds RED if that ordering regresses. */
+void cbm_pipeline_incremental_test_after_stage_created_once(cbm_pipeline_test_hook_fn hook,
+                                                            void *userdata);
 cbm_incremental_route_t cbm_pipeline_incremental_test_last_route(void);
 void cbm_pipeline_incremental_test_reset_faults(void);
 
@@ -906,6 +926,7 @@ bool cbm_pipeline_persist_test_take_failure_after_stage_dump(void);
 bool cbm_pipeline_persist_test_take_cancel_after_predump(void);
 bool cbm_pipeline_persist_test_take_cancel_after_destination_prepare(void);
 void cbm_pipeline_persist_test_run_before_final_manifest(void);
+void cbm_pipeline_persist_test_run_after_stage_created(void);
 void cbm_pipeline_persist_test_reset_faults(void);
 #endif
 

tests/test_pipeline.c

@@ -2278,7 +2278,7 @@ TEST(pipeline_complexity_props_independent_of_worker_order) {
     ASSERT_EQ(sequential_rc, 0);
     ASSERT_NOT_NULL(sequential_sig);
     ASSERT_GTE(sequential_funcs, copied); /* at least the one function per fixture file */
-    ASSERT_TRUE(cycles_detected);        /* the cycles must reach the pass at all */
+    ASSERT_TRUE(cycles_detected);         /* the cycles must reach the pass at all */
     if (mismatch_run >= 0) {
         printf("\n    parallel run %d diverges from sequential: %s\n", mismatch_run, diff);
         FAIL("complexity props depend on worker id order");
@@ -3416,6 +3416,481 @@ TEST(pipeline_publication_never_uses_a_predictable_staging_path) {
     PASS();
 }
 
+static int count_substring(const char *haystack, const char *needle) {
+    int count = 0;
+    size_t needle_len = strlen(needle);
+    for (const char *at = strstr(haystack, needle); at; at = strstr(at + needle_len, needle)) {
+        count++;
+    }
+    return count;
+}
+
+static int count_nested_stage_entries(const char *dir_path, const char *db_basename) {
+    cbm_dir_t *dir = cbm_opendir(dir_path);
+    if (!dir) {
+        return -1;
+    }
+    size_t base_len = strlen(db_basename);
+    int count = 0;
+    cbm_dirent_t *entry;
+    while ((entry = cbm_readdir(dir)) != NULL) {
+        if (strncmp(entry->name, db_basename, base_len) == 0 &&
+            count_substring(entry->name + base_len, ".stage.") >= 2) {
+            count++;
+        }
+    }
+    cbm_closedir(dir);
+    return count;
+}
+
+/* #1839: the outer run rewrites the pipeline's db_path to its stage, so the
+ * inner publication (dump or delta clone) minted ITS stage from a stage:
+ * <db>.stage.A.stage.B, plus -wal/-shm under WAL. A generation's stage is a
+ * sibling of the live database, whichever path it is minted from; a database
+ * whose own basename merely contains ".stage." is not a stage and keeps its
+ * full name. */
+TEST(pipeline_stage_names_never_nest) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_nesting_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+    char odd_db[512];
+    snprintf(odd_db, sizeof(odd_db), "%s/x.stage.y.db", tmp);
+
+    char *outer = cbm_pipeline_create_staging_path(db_path);
+    ASSERT_NOT_NULL(outer);
+    char *inner = cbm_pipeline_create_staging_path(outer);
+    ASSERT_NOT_NULL(inner);
+    char *odd_stage = cbm_pipeline_create_staging_path(odd_db);
+    ASSERT_NOT_NULL(odd_stage);
+
+    size_t db_len = strlen(db_path);
+    size_t odd_len = strlen(odd_db);
+    int outer_tokens = count_substring(outer, ".stage.");
+    int inner_tokens = count_substring(inner, ".stage.");
+    bool inner_is_sibling =
+        strncmp(inner, db_path, db_len) == 0 && strncmp(inner + db_len, ".stage.", 7) == 0;
+    bool inner_distinct = strcmp(inner, outer) != 0;
+    bool odd_keeps_basename =
+        strncmp(odd_stage, odd_db, odd_len) == 0 && strncmp(odd_stage + odd_len, ".stage.", 7) == 0;
+    int nested_entries = count_nested_stage_entries(tmp, "generation.db");
+
+    cbm_pipeline_discard_stage(inner);
+    cbm_pipeline_discard_stage(outer);
+    cbm_pipeline_discard_stage(odd_stage);
+    int leftover_entries = count_generation_stage_artifacts(tmp, "generation.db") +
+                           count_generation_stage_artifacts(tmp, "x.stage.y.db");
+    free(inner);
+    free(outer);
+    free(odd_stage);
+    th_rmtree(tmp);
+
+    ASSERT_EQ(outer_tokens, 1);
+    ASSERT_EQ(inner_tokens, 1);
+    ASSERT_TRUE(inner_is_sibling);
+    ASSERT_TRUE(inner_distinct);
+    ASSERT_TRUE(odd_keeps_basename);
+    ASSERT_EQ(nested_entries, 0);
+    ASSERT_EQ(leftover_entries, 0);
+    PASS();
+}
+
+static bool path_exists(const char *path) {
+    cbm_path_info_t info;
+    return cbm_path_info_utf8(path, &info) == CBM_PATH_INFO_OK;
+}
+
+/* #1839: a minted stage is OWNED through an exclusive kernel lock on its
+ * "<stage>.lock" sidecar for exactly as long as the stage exists. A second
+ * holder cannot take it while the writer is live; discarding the stage frees
+ * the name and removes the sidecar, so nothing stays behind. */
+TEST(pipeline_minted_stage_is_owned_until_released) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_owner_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+
+    char *stage = cbm_pipeline_create_staging_path(db_path);
+    ASSERT_NOT_NULL(stage);
+    char lock_path[600];
+    snprintf(lock_path, sizeof(lock_path), "%s.lock", stage);
+    bool lock_present_while_live = path_exists(lock_path);
+    int hold_while_live = cbm_pipeline_stage_lock_hold(stage);
+    if (hold_while_live >= 0) {
+        cbm_pipeline_stage_lock_drop(stage, hold_while_live);
+    }
+
+    cbm_pipeline_discard_stage(stage);
+    bool stage_gone = !path_exists(stage);
+    bool lock_gone = !path_exists(lock_path);
+
+    int hold_after_discard = cbm_pipeline_stage_lock_hold(stage);
+    int second_holder = cbm_pipeline_stage_lock_hold(stage);
+    if (second_holder >= 0) {
+        cbm_pipeline_stage_lock_drop(stage, second_holder);
+    }
+    cbm_pipeline_stage_lock_drop(stage, hold_after_discard);
+    int hold_after_drop = cbm_pipeline_stage_lock_hold(stage);
+    cbm_pipeline_stage_lock_drop(stage, hold_after_drop);
+    int leftovers = count_generation_stage_artifacts(tmp, "generation.db");
+    free(stage);
+    th_rmtree(tmp);
+
+    ASSERT_TRUE(lock_present_while_live);
+    ASSERT_EQ(hold_while_live, -1);
+    ASSERT_TRUE(stage_gone);
+    ASSERT_TRUE(lock_gone);
+    ASSERT_TRUE(hold_after_discard >= 0);
+    ASSERT_EQ(second_holder, -1);
+    ASSERT_TRUE(hold_after_drop >= 0);
+    ASSERT_EQ(leftovers, 0);
+    PASS();
+}
+
+static bool file_has_content(const char *path, const char *expected) {
+    FILE *f = cbm_fopen(path, "rb");
+    if (!f) {
+        return false;
+    }
+    char buf[256] = {0};
+    size_t n = fread(buf, 1, sizeof(buf) - 1, f);
+    (void)fclose(f);
+    return n == strlen(expected) && memcmp(buf, expected, n) == 0;
+}
+
+/* #1839 / #1864: a stage a dead writer left beside a VALID database is swept
+ * by the next run and never influences its route. The 0-byte shape is what a
+ * worker killed before its backup wrote a page leaves behind; a stale -shm
+ * beside it is swept with it. */
+TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_stale_stage_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 1\n");
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(baseline);
+    ASSERT_EQ(cbm_pipeline_run(baseline), 0);
+    char project[256];
+    snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(baseline));
+    cbm_pipeline_free(baseline);
+
+    char stale_stage[600];
+    char stale_shm[600];
+    snprintf(stale_stage, sizeof(stale_stage), "%s.stage.deadbe", db_path);
+    snprintf(stale_shm, sizeof(stale_shm), "%s.stage.deadbe-shm", db_path);
+    ASSERT_EQ(th_write_file(stale_stage, ""), 0);
+    ASSERT_EQ(th_write_file(stale_shm, "stale-shm"), 0);
+
+    /* A body-only change: no added names, so the planner may repair. */
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 2\n");
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *incr = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(incr);
+    int incr_rc = cbm_pipeline_run(incr);
+    cbm_incremental_route_t route = cbm_pipeline_incremental_test_last_route();
+    cbm_pipeline_free(incr);
+
+    bool stale_stage_gone = !path_exists(stale_stage);
+    bool stale_shm_gone = !path_exists(stale_shm);
+    int stage_count = count_generation_stage_artifacts(tmp, "generation.db");
+    int stable_count = -1;
+    int absent_count = -1;
+    observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", &stable_count,
+                             &absent_count);
+    cbm_pipeline_incremental_test_reset_faults();
+    th_rmtree(tmp);
+
+    ASSERT_EQ(incr_rc, 0);
+    ASSERT_TRUE(route != CBM_INCREMENTAL_ROUTE_FORCED_FULL);
+    ASSERT_TRUE(route != CBM_INCREMENTAL_ROUTE_NONE);
+    ASSERT_TRUE(stale_stage_gone);
+    ASSERT_TRUE(stale_shm_gone);
+    ASSERT_EQ(stage_count, 0);
+    ASSERT_EQ(stable_count, 1);
+    ASSERT_EQ(absent_count, 0);
+    PASS();
+}
+
+/* #1839: the sweep removes exactly the stages nobody owns. A dead writer's
+ * stage (main, -wal, and the unlocked .lock sidecar its death left) goes; a
+ * stage whose writer is LIVE -- here the test, holding its lock -- is kept
+ * byte for byte, and goes only once that lock is dropped. No timing: liveness
+ * is the kernel lock, nothing else. */
+TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_sweep_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 1\n");
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(baseline);
+    ASSERT_EQ(cbm_pipeline_run(baseline), 0);
+    char project[256];
+    snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(baseline));
+    cbm_pipeline_free(baseline);
+
+    char dead_stage[600];
+    char dead_wal[600];
+    char dead_lock[600];
+    char live_stage[600];
+    char live_lock[600];
+    snprintf(dead_stage, sizeof(dead_stage), "%s.stage.aaaaaa", db_path);
+    snprintf(dead_wal, sizeof(dead_wal), "%s.stage.aaaaaa-wal", db_path);
+    snprintf(dead_lock, sizeof(dead_lock), "%s.stage.aaaaaa.lock", db_path);
+    snprintf(live_stage, sizeof(live_stage), "%s.stage.bbbbbb", db_path);
+    snprintf(live_lock, sizeof(live_lock), "%s.stage.bbbbbb.lock", db_path);
+    static const char live_bytes[] = "live-stage-bytes";
+    ASSERT_EQ(th_write_file(dead_stage, "dead-stage-bytes"), 0);
+    ASSERT_EQ(th_write_file(dead_wal, "dead-wal"), 0);
+    ASSERT_EQ(th_write_file(dead_lock, ""), 0);
+    ASSERT_EQ(th_write_file(live_stage, live_bytes), 0);
+    int live_fd = cbm_pipeline_stage_lock_hold(live_stage);
+    ASSERT_TRUE(live_fd >= 0);
+
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 2\n");
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *first = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(first);
+    int first_rc = cbm_pipeline_run(first);
+    cbm_pipeline_free(first);
+
+    bool dead_stage_gone = !path_exists(dead_stage);
+    bool dead_wal_gone = !path_exists(dead_wal);
+    bool dead_lock_gone = !path_exists(dead_lock);
+    bool live_kept = file_has_content(live_stage, live_bytes);
+    bool live_lock_kept = path_exists(live_lock);
+    int stable_after_first = -1;
+    int absent_after_first = -1;
+    observe_named_generation(db_path, project, "StableGeneration", "NeverDefined",
+                             &stable_after_first, &absent_after_first);
+
+    cbm_pipeline_stage_lock_drop(live_stage, live_fd);
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *second = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(second);
+    int second_rc = cbm_pipeline_run(second);
+    cbm_pipeline_free(second);
+
+    bool live_gone = !path_exists(live_stage);
+    bool live_lock_gone = !path_exists(live_lock);
+    int stage_count = count_generation_stage_artifacts(tmp, "generation.db");
+    int stable_after_second = -1;
+    int absent_after_second = -1;
+    observe_named_generation(db_path, project, "StableGeneration", "NeverDefined",
+                             &stable_after_second, &absent_after_second);
+    cbm_pipeline_incremental_test_reset_faults();
+    th_rmtree(tmp);
+
+    ASSERT_EQ(first_rc, 0);
+    ASSERT_TRUE(dead_stage_gone);
+    ASSERT_TRUE(dead_wal_gone);
+    ASSERT_TRUE(dead_lock_gone);
+    ASSERT_TRUE(live_kept);
+    ASSERT_TRUE(live_lock_kept);
+    ASSERT_EQ(stable_after_first, 1);
+    ASSERT_EQ(absent_after_first, 0);
+    ASSERT_EQ(second_rc, 0);
+    ASSERT_TRUE(live_gone);
+    ASSERT_TRUE(live_lock_gone);
+    ASSERT_EQ(stage_count, 0);
+    ASSERT_EQ(stable_after_second, 1);
+    ASSERT_EQ(absent_after_second, 0);
+    PASS();
+}
+
+/* #2111 create->register TOCTOU guard. The bug this pins: create_staging_path()
+ * used to make the stage's main file visible (via mkstemp) BEFORE taking its
+ * sidecar lock, and sweep_orphan_stages() treats an absent lock sidecar
+ * (ENOENT) as a confirmed-dead writer (kernel released the lock on death). So a
+ * second, concurrent cbm_pipeline_run() against the SAME final_path, landing
+ * its own sweep in that narrow unlocked window, removed the first run's
+ * in-flight stage out from under it: every extraction pass still completed
+ * (none touch the stage file on disk), but the publish that followed found its
+ * own stage gone. Two writers racing the same project is a real scenario this
+ * PR's own sweep exists to clean up after (auto_index; the recently-fixed
+ * stale-rendezvous-recovery retry path) -- not hypothetical, and exactly the
+ * shape of #2111's windows-guards red (every pass logged success, nothing was
+ * ever committed, "Pipeline failed" surfaced generic; on Windows the sidecar
+ * collision surfaced as EACCES/errno=13).
+ *
+ * The fix takes the sidecar lock BEFORE the main file becomes visible, so this
+ * hook -- fired the instant the main file exists -- finds the stage already
+ * lock-protected and the racing sweep keeps it. The hook fires at the same
+ * point under the old ordering, where the lock was NOT yet held, so this test
+ * goes RED if that ordering ever regresses.
+ *
+ * The racing run is cancelled immediately so it never reaches ITS OWN
+ * publish -- isolating the sweep's effect on the first run's stage from the
+ * separate question of two full runs both completing for the same project. */
+typedef struct {
+    const char *tmp_dir;
+    const char *db_path;
+    bool stage_survived;
+} racing_sweep_arg_t;
+
+static bool find_sole_stage_path(const char *dir, const char *db_basename, char *out,
+                                 size_t out_sz) {
+    cbm_dir_t *d = cbm_opendir(dir);
+    if (!d) {
+        return false;
+    }
+    char prefix[256];
+    snprintf(prefix, sizeof(prefix), "%s.stage.", db_basename);
+    size_t prefix_len = strlen(prefix);
+    bool found = false;
+    cbm_dirent_t *entry;
+    while ((entry = cbm_readdir(d)) != NULL) {
+        size_t name_len = strlen(entry->name);
+        enum { STAGE_SUFFIX_RANDOM_CHARS = 6 }; /* mirrors CBM_STAGE_SUFFIX_RANDOM_CHARS */
+        if (strncmp(entry->name, prefix, prefix_len) == 0 &&
+            name_len == prefix_len + STAGE_SUFFIX_RANDOM_CHARS) {
+            snprintf(out, out_sz, "%s/%s", dir, entry->name);
+            found = true;
+            break;
+        }
+    }
+    cbm_closedir(d);
+    return found;
+}
+
+static void *racing_sweep_thread(void *arg) {
+    racing_sweep_arg_t *a = (racing_sweep_arg_t *)arg;
+    cbm_pipeline_t *p = cbm_pipeline_new(a->tmp_dir, a->db_path, CBM_MODE_FULL);
+    if (p) {
+        /* Its own sweep_orphan_stages() runs at the very start, before this
+         * run mints its own stage -- exactly like the first run's. Cancel
+         * immediately: this run must reach the sweep and nothing past it. */
+        cbm_pipeline_cancel(p);
+        (void)cbm_pipeline_run(p);
+        cbm_pipeline_free(p);
+    }
+    return NULL;
+}
+
+/* Fired from inside the FIRST run's create_staging_path(), the instant its
+ * stage main file exists (under the fix, with its sidecar lock already held;
+ * under the old create-then-lock ordering, before the lock was taken): run a
+ * second, cancelled cbm_pipeline_run() for the same project synchronously on
+ * another thread, so its sweep has every chance to reach the stage before
+ * control returns to the first run. */
+static void racing_sweep_hook(void *userdata) {
+    racing_sweep_arg_t *a = (racing_sweep_arg_t *)userdata;
+    char stage_path[600] = {0};
+    bool had_stage =
+        find_sole_stage_path(a->tmp_dir, "generation.db", stage_path, sizeof(stage_path));
+    cbm_thread_t tid;
+    if (cbm_thread_create(&tid, 0, racing_sweep_thread, a) == 0) {
+        cbm_thread_join(&tid);
+    }
+    a->stage_survived = had_stage && path_exists(stage_path);
+}
+
+TEST(pipeline_concurrent_sweep_must_not_remove_inflight_stage) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_race_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 1\n");
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+
+    cbm_pipeline_incremental_test_reset_faults();
+    racing_sweep_arg_t race_arg = {.tmp_dir = tmp, .db_path = db_path, .stage_survived = false};
+    cbm_pipeline_incremental_test_after_stage_created_once(racing_sweep_hook, &race_arg);
+
+    cbm_pipeline_t *p = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(p);
+    char project[256];
+    snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(p));
+    int rc = cbm_pipeline_run(p);
+    cbm_pipeline_free(p);
+
+    bool db_exists = path_exists(db_path);
+    int defined_count = -1;
+    int absent_count = -1;
+    observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", &defined_count,
+                             &absent_count);
+    int stage_count = count_generation_stage_artifacts(tmp, "generation.db");
+    cbm_pipeline_incremental_test_reset_faults();
+    th_rmtree(tmp);
+
+    ASSERT_TRUE(race_arg.stage_survived);
+    ASSERT_EQ(rc, 0);
+    ASSERT_TRUE(db_exists);
+    ASSERT_EQ(defined_count, 1);
+    ASSERT_EQ(absent_count, 0);
+    ASSERT_EQ(stage_count, 0);
+    PASS();
+}
+
+static char g_route_log_capture[8192];
+static atomic_flag g_route_log_spin = ATOMIC_FLAG_INIT;
+
+/* Keeps only the route decisions; worker threads log too, and only the
+ * appends need serialising. */
+static void capture_route_log_sink(const char *line) {
+    if (!line || !strstr(line, "pipeline.route")) {
+        return;
+    }
+    while (atomic_flag_test_and_set_explicit(&g_route_log_spin, memory_order_acquire)) {}
+    size_t used = strlen(g_route_log_capture);
+    size_t avail = sizeof(g_route_log_capture) - used;
+    if (avail > 1) {
+        int n = snprintf(g_route_log_capture + used, avail, "%s\n", line);
+        if (n < 0 || (size_t)n >= avail) {
+            g_route_log_capture[sizeof(g_route_log_capture) - 1] = '\0';
+        }
+    }
+    atomic_flag_clear_explicit(&g_route_log_spin, memory_order_release);
+}
+
+/* #1864: a first index has no previous generation. The route probe used to
+ * open the run's own EMPTY stage placeholder, fail its integrity check, and
+ * warn "reason=invalid_existing_db" on every first index of every project,
+ * which the report read as a corrupted database and a crash loop. A fresh
+ * index says what it is. */
+TEST(pipeline_fresh_index_never_reports_invalid_existing_db) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_fresh_route_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    write_temp_file(tmp, "generation.py", "def StableGeneration():\n    return 1\n");
+    char db_path[512];
+    snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp);
+
+    g_route_log_capture[0] = '\0';
+    CBMLogLevel previous_level = cbm_log_get_level();
+    cbm_log_set_level(CBM_LOG_DEBUG);
+    cbm_log_set_sink(capture_route_log_sink);
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *fresh = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
+    int fresh_rc = fresh ? cbm_pipeline_run(fresh) : -1;
+    cbm_pipeline_free(fresh);
+    cbm_log_set_sink(NULL);
+    cbm_log_set_level(previous_level);
+
+    bool invalid_reported = strstr(g_route_log_capture, "invalid_existing_db") != NULL;
+    bool fresh_reported = strstr(g_route_log_capture, "reason=no_existing_db") != NULL;
+    bool db_present = path_exists(db_path);
+    int stage_count = count_generation_stage_artifacts(tmp, "generation.db");
+    cbm_pipeline_incremental_test_reset_faults();
+    th_rmtree(tmp);
+
+    ASSERT_EQ(fresh_rc, 0);
+    ASSERT_TRUE(!invalid_reported);
+    ASSERT_TRUE(fresh_reported);
+    ASSERT_TRUE(db_present);
+    ASSERT_EQ(stage_count, 0);
+    PASS();
+}
+
 /* Discovery and extraction must describe the same immutable generation. A
  * source file created after extraction is not present in the original file
  * list, so merely re-hashing that list cannot detect the race. Publication
@@ -7137,19 +7612,18 @@ TEST(pipeline_python_cross_module_call) {
  * unique_name (candidates==1) is #1572 and is not this claim. */
 TEST(pipeline_cross_language_same_name_does_not_share_calls_issue725) {
     const char *files[] = {"store.py", "app.py", "web/src/pages/Editor.js"};
-    const char *contents[] = {
-        "class Store:\n"
-        "    def commit(self):\n"
-        "        return True\n",
+    const char *contents[] = {"class Store:\n"
+                              "    def commit(self):\n"
+                              "        return True\n",
 
-        "from store import Store\n"
-        "\n"
-        "def save():\n"
-        "    return Store().commit()\n",
+                              "from store import Store\n"
+                              "\n"
+                              "def save():\n"
+                              "    return Store().commit()\n",
 
-        "export function commit() {\n"
-        "  return 1;\n"
-        "}\n"};
+                              "export function commit() {\n"
+                              "  return 1;\n"
+                              "}\n"};
 
     if (setup_lang_repo(files, contents, 3) != 0)
         FAIL("tmpdir");
@@ -13723,7 +14197,6 @@ TEST(pipeline_delta_patch_indexes_docstring_into_fts_body) {
     PASS();
 }
 
-
 /* End-to-end for #518/#519: source → docstring → properties JSON → nodes_fts
  * `body` → findable. Each layer has its own test; this one proves they connect.
  * It is also the guard on the size budget: build_def_props drops an oversized
@@ -14280,6 +14753,12 @@ SUITE(pipeline_semantic_manifest_repro) {
     RUN_TEST(pipeline_git_context_change_forces_full_and_refreshes_branch);
     RUN_TEST(pipeline_global_extension_config_change_forces_full);
     RUN_TEST(pipeline_publication_never_uses_a_predictable_staging_path);
+    RUN_TEST(pipeline_stage_names_never_nest);
+    RUN_TEST(pipeline_minted_stage_is_owned_until_released);
+    RUN_TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept);
+    RUN_TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage);
+    RUN_TEST(pipeline_concurrent_sweep_must_not_remove_inflight_stage);
+    RUN_TEST(pipeline_fresh_index_never_reports_invalid_existing_db);
     RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation);
     RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation);
     RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation);