feat: coordinate concurrent CBM sessions

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

tests/test_subprocess.c

@@ -10,11 +10,18 @@
  */
 #include "test_framework.h"
 #include "../src/foundation/subprocess.h"
+#include "../src/foundation/compat.h"
+#include "../src/foundation/platform.h"
 
+#include <errno.h>
+#include <limits.h>
+#include <stdio.h>
 #include <string.h>
 
 #ifndef _WIN32
+#include <fcntl.h>
 #include <signal.h>
+#include <unistd.h>
 #endif
 
 /* ── Layer 1: pure classifier (all platforms) ─────────────────────────────── */
@@ -37,6 +44,7 @@ TEST(subprocess_classify_windows_crash_codes) {
     ASSERT_EQ(cbm_proc_classify(true, (int)0xC0000005u, 0, false), CBM_PROC_CRASH);
     ASSERT_EQ(cbm_proc_classify(true, (int)0xC00000FDu, 0, false), CBM_PROC_CRASH);
     ASSERT_EQ(cbm_proc_classify(true, (int)0xC000001Du, 0, false), CBM_PROC_CRASH);
+    ASSERT_EQ(cbm_proc_classify(true, (int)0xC000013Au, 0, false), CBM_PROC_KILLED);
     PASS();
 }
 
@@ -168,7 +176,456 @@ TEST(subprocess_run_null_bin_rejected) {
     PASS();
 }
 
-/* ── Layer 3: Windows command-line quoting (pure; every platform) ─────────────
+/* ── Layer 3: nonblocking handle + whole-tree cancellation (POSIX) ──────────
+ *
+ * The daemon cannot block its coordinator thread in cbm_subprocess_run(). It
+ * owns an opaque handle, polls it, and requests cancellation when the final job
+ * subscriber leaves. These probes deliberately use /bin/sh descendants which
+ * ignore SIGTERM: a direct-child-only implementation leaves the grandchild alive,
+ * while a correct process-group implementation escalates and reports quiescence.
+ * All test-side waits have explicit monotonic deadlines. */
+
+#ifndef _WIN32
+
+static void subprocess_test_pause(void) {
+    const struct timespec delay = {0, 10000000L}; /* 10 ms */
+    (void)cbm_nanosleep(&delay, NULL);
+}
+
+static bool poll_until_terminal(cbm_subprocess_t *process, int timeout_ms, cbm_proc_result_t *out) {
+    uint64_t deadline = cbm_now_ms() + (uint64_t)timeout_ms;
+    do {
+        cbm_proc_poll_t state = cbm_subprocess_poll(process, out);
+        if (state == CBM_PROC_POLL_TERMINAL) {
+            return true;
+        }
+        if (state == CBM_PROC_POLL_ERROR) {
+            return false;
+        }
+        subprocess_test_pause();
+    } while (cbm_now_ms() < deadline);
+    return cbm_subprocess_poll(process, out) == CBM_PROC_POLL_TERMINAL;
+}
+
+static bool make_tree_pid_path(char path[64]) {
+    strcpy(path, "/tmp/cbm-subprocess-tree-XXXXXX");
+    int fd = cbm_mkstemp(path);
+    if (fd < 0) {
+        return false;
+    }
+    (void)close(fd);
+    return unlink(path) == 0; /* child creates it only after both traps are installed */
+}
+
+static bool wait_for_tree_pids(const char *path, cbm_subprocess_t *process, pid_t *parent_pid,
+                               pid_t *grandchild_pid, int timeout_ms) {
+    uint64_t deadline = cbm_now_ms() + (uint64_t)timeout_ms;
+    do {
+        FILE *f = fopen(path, "r");
+        if (f) {
+            long parent_value = 0;
+            long grandchild_value = 0;
+            int fields = fscanf(f, "%ld %ld", &parent_value, &grandchild_value);
+            fclose(f);
+            if (fields == 2 && parent_value > 1 && grandchild_value > 1) {
+                *parent_pid = (pid_t)parent_value;
+                *grandchild_pid = (pid_t)grandchild_value;
+                return true;
+            }
+        }
+        cbm_proc_result_t ignored;
+        if (cbm_subprocess_poll(process, &ignored) != CBM_PROC_POLL_RUNNING) {
+            return false;
+        }
+        subprocess_test_pause();
+    } while (cbm_now_ms() < deadline);
+    return false;
+}
+
+static bool wait_pid_gone(pid_t pid, int timeout_ms) {
+    uint64_t deadline = cbm_now_ms() + (uint64_t)timeout_ms;
+    do {
+        errno = 0;
+        if (kill(pid, 0) < 0 && errno == ESRCH) {
+            return true;
+        }
+        subprocess_test_pause();
+    } while (cbm_now_ms() < deadline);
+    errno = 0;
+    return kill(pid, 0) < 0 && errno == ESRCH;
+}
+
+/* Best-effort cleanup for a failing implementation, so a red tree test does not
+ * leave its TERM-ignoring probes behind for later tests. The production API must
+ * still report terminal itself; callers never use this escape hatch. */
+static void force_probe_cleanup(pid_t parent_pid, pid_t grandchild_pid) {
+    if (parent_pid > 1) {
+        (void)kill(-parent_pid, SIGKILL);
+        (void)kill(parent_pid, SIGKILL);
+    }
+    if (grandchild_pid > 1) {
+        (void)kill(grandchild_pid, SIGKILL);
+    }
+}
+
+static int spawn_ignoring_tree(const char *pid_path, int quiet_timeout_ms, int cancel_grace_ms,
+                               cbm_subprocess_t **out) {
+    /* The direct child installs its trap before starting a nested shell. The two
+     * PIDs are written only after the nested process exists, eliminating the
+     * cancellation-before-trap race from the test. */
+    const char *script = "trap '' TERM; "
+                         "/bin/sh -c 'trap \"\" TERM; while :; do sleep 1; done' cbm-grandchild & "
+                         "grandchild=$!; echo \"$$ $grandchild\" > \"$1\"; wait";
+    const char *argv[] = {"/bin/sh", "-c", script, "cbm-parent", pid_path, NULL};
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    opts.quiet_timeout_ms = quiet_timeout_ms;
+    opts.cancel_grace_ms = cancel_grace_ms;
+    return cbm_subprocess_spawn(&opts, out);
+}
+
+static void slow_log_callback(const char *line, void *opaque) {
+    (void)line;
+    int *count = opaque;
+    (*count)++;
+    const struct timespec delay = {0, 2000000L}; /* 2 ms per delivered line */
+    (void)cbm_nanosleep(&delay, NULL);
+}
+
+#endif /* !_WIN32 */
+
+TEST(subprocess_spawn_returns_while_child_is_running) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX /bin/sh nonblocking spawn probe; native Job Object coverage pending");
+#else
+    const char *argv[] = {"/bin/sh", "-c", "sleep 4", NULL};
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    opts.cancel_grace_ms = 100;
+    cbm_subprocess_t *process = NULL;
+
+    uint64_t before_spawn = cbm_now_ms();
+    int spawn_rc = cbm_subprocess_spawn(&opts, &process);
+    uint64_t spawn_elapsed = cbm_now_ms() - before_spawn;
+    ASSERT_EQ(spawn_rc, 0);
+    ASSERT_NOT_NULL(process);
+
+    cbm_proc_result_t result;
+    uint64_t before_poll = cbm_now_ms();
+    cbm_proc_poll_t first_poll = cbm_subprocess_poll(process, &result);
+    uint64_t poll_elapsed = cbm_now_ms() - before_poll;
+    bool cancel_accepted = cbm_subprocess_request_cancel(process);
+    bool terminal = poll_until_terminal(process, 2000, &result);
+    if (terminal) {
+        cbm_subprocess_destroy(process);
+    }
+
+    /* Generous thresholds distinguish a nonblocking call from waiting for the
+     * four-second probe, without making normal scheduler jitter test-significant. */
+    ASSERT_LT(spawn_elapsed, 1500);
+    ASSERT_LT(poll_elapsed, 1000);
+    ASSERT_EQ(first_poll, CBM_PROC_POLL_RUNNING);
+    ASSERT_TRUE(cancel_accepted);
+    ASSERT_TRUE(terminal);
+    ASSERT_TRUE(result.cancellation_requested);
+    ASSERT_TRUE(result.tree_quiesced);
+    PASS();
+#endif
+}
+
+TEST(subprocess_natural_completion_is_cached_across_polls) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX /bin/sh completion-cache probe; native Job Object coverage pending");
+#else
+    const char *argv[] = {"/bin/sh", "-c", "exit 7", NULL};
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    opts.cancel_grace_ms = 100;
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(cbm_subprocess_spawn(&opts, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    cbm_proc_result_t first;
+    ASSERT_TRUE(poll_until_terminal(process, 2000, &first));
+    cbm_proc_result_t second;
+    cbm_proc_result_t third;
+    cbm_proc_poll_t second_poll = cbm_subprocess_poll(process, &second);
+    cbm_proc_poll_t third_poll = cbm_subprocess_poll(process, &third);
+    cbm_subprocess_destroy(process);
+
+    ASSERT_EQ(second_poll, CBM_PROC_POLL_TERMINAL);
+    ASSERT_EQ(third_poll, CBM_PROC_POLL_TERMINAL);
+    ASSERT_EQ(first.outcome, CBM_PROC_EXIT_NONZERO);
+    ASSERT_EQ(first.exit_code, 7);
+    ASSERT_EQ(second.outcome, first.outcome);
+    ASSERT_EQ(second.exit_code, first.exit_code);
+    ASSERT_EQ(second.term_signal, first.term_signal);
+    ASSERT_EQ(second.cancellation_requested, first.cancellation_requested);
+    ASSERT_EQ(second.forced, first.forced);
+    ASSERT_EQ(second.tree_quiesced, first.tree_quiesced);
+    ASSERT_EQ(third.outcome, first.outcome);
+    ASSERT_EQ(third.exit_code, first.exit_code);
+    ASSERT_EQ(third.term_signal, first.term_signal);
+    ASSERT_EQ(third.cancellation_requested, first.cancellation_requested);
+    ASSERT_EQ(third.forced, first.forced);
+    ASSERT_EQ(third.tree_quiesced, first.tree_quiesced);
+    ASSERT_FALSE(first.cancellation_requested);
+    ASSERT_FALSE(first.forced);
+    ASSERT_TRUE(first.tree_quiesced);
+    PASS();
+#endif
+}
+
+TEST(subprocess_cancel_is_idempotent_and_kills_ignoring_tree) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX process-group probe; native Windows Job Object tree probe pending");
+#else
+    char pid_path[64];
+    ASSERT_TRUE(make_tree_pid_path(pid_path));
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(spawn_ignoring_tree(pid_path, 0, 100, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    pid_t parent_pid = -1;
+    pid_t grandchild_pid = -1;
+    bool ready = wait_for_tree_pids(pid_path, process, &parent_pid, &grandchild_pid, 1000);
+    bool first_cancel = ready && cbm_subprocess_request_cancel(process);
+    bool second_cancel = ready && cbm_subprocess_request_cancel(process);
+    cbm_proc_result_t result;
+    bool terminal = ready && poll_until_terminal(process, 2500, &result);
+    bool parent_gone = terminal && wait_pid_gone(parent_pid, 1000);
+    bool grandchild_gone = terminal && wait_pid_gone(grandchild_pid, 1000);
+    if (!terminal) {
+        force_probe_cleanup(parent_pid, grandchild_pid);
+        cbm_proc_result_t cleanup_result;
+        if (poll_until_terminal(process, 1000, &cleanup_result)) {
+            cbm_subprocess_destroy(process);
+        }
+    } else {
+        cbm_subprocess_destroy(process);
+    }
+    (void)unlink(pid_path);
+
+    ASSERT_TRUE(ready);
+    ASSERT_TRUE(first_cancel);
+    ASSERT_TRUE(second_cancel);
+    ASSERT_TRUE(terminal);
+    ASSERT_EQ(result.outcome, CBM_PROC_KILLED);
+    ASSERT_TRUE(result.cancellation_requested);
+    ASSERT_TRUE(result.forced);
+    ASSERT_TRUE(result.tree_quiesced);
+    ASSERT_TRUE(parent_gone);
+    ASSERT_TRUE(grandchild_gone);
+    PASS();
+#endif
+}
+
+TEST(subprocess_quiet_timeout_kills_ignoring_tree) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX process-group probe; native Windows Job Object tree probe pending");
+#else
+    char pid_path[64];
+    ASSERT_TRUE(make_tree_pid_path(pid_path));
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(spawn_ignoring_tree(pid_path, 750, 100, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    pid_t parent_pid = -1;
+    pid_t grandchild_pid = -1;
+    bool ready = wait_for_tree_pids(pid_path, process, &parent_pid, &grandchild_pid, 500);
+    cbm_proc_result_t result;
+    bool terminal = ready && poll_until_terminal(process, 3000, &result);
+    bool parent_gone = terminal && wait_pid_gone(parent_pid, 1000);
+    bool grandchild_gone = terminal && wait_pid_gone(grandchild_pid, 1000);
+    if (!terminal) {
+        force_probe_cleanup(parent_pid, grandchild_pid);
+        cbm_proc_result_t cleanup_result;
+        if (poll_until_terminal(process, 1000, &cleanup_result)) {
+            cbm_subprocess_destroy(process);
+        }
+    } else {
+        cbm_subprocess_destroy(process);
+    }
+    (void)unlink(pid_path);
+
+    ASSERT_TRUE(ready);
+    ASSERT_TRUE(terminal);
+    ASSERT_EQ(result.outcome, CBM_PROC_HANG);
+    ASSERT_FALSE(result.cancellation_requested);
+    ASSERT_TRUE(result.forced);
+    ASSERT_TRUE(result.tree_quiesced);
+    ASSERT_TRUE(parent_gone);
+    ASSERT_TRUE(grandchild_gone);
+    PASS();
+#endif
+}
+
+TEST(subprocess_cancel_grace_is_hard_capped) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX process-group grace cap probe; native Windows coverage pending");
+#else
+    char pid_path[64];
+    ASSERT_TRUE(make_tree_pid_path(pid_path));
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(spawn_ignoring_tree(pid_path, 0, INT_MAX, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    pid_t parent_pid = -1;
+    pid_t grandchild_pid = -1;
+    bool ready = wait_for_tree_pids(pid_path, process, &parent_pid, &grandchild_pid, 1000);
+    bool cancel_accepted = ready && cbm_subprocess_request_cancel(process);
+    cbm_proc_result_t result = {0};
+    bool terminal = cancel_accepted && poll_until_terminal(process, 3500, &result);
+    if (!terminal) {
+        force_probe_cleanup(parent_pid, grandchild_pid);
+        cbm_proc_result_t cleanup_result;
+        if (poll_until_terminal(process, 1000, &cleanup_result)) {
+            cbm_subprocess_destroy(process);
+        }
+    } else {
+        cbm_subprocess_destroy(process);
+    }
+    (void)unlink(pid_path);
+
+    ASSERT_TRUE(ready);
+    ASSERT_TRUE(cancel_accepted);
+    ASSERT_TRUE(terminal);
+    ASSERT_TRUE(result.forced);
+    ASSERT_TRUE(result.tree_quiesced);
+    PASS();
+#endif
+}
+
+TEST(subprocess_poll_has_a_strict_log_delivery_budget) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX shell log budget probe; native Windows coverage pending");
+#else
+    char log_path[] = "/tmp/cbm-subprocess-log-budget-XXXXXX";
+    int log_fd = cbm_mkstemp(log_path);
+    ASSERT_TRUE(log_fd >= 0);
+    (void)close(log_fd);
+
+    const char *script = "i=0; while [ $i -lt 400 ]; do echo line-$i; i=$((i+1)); done; sleep 4";
+    const char *argv[] = {"/bin/sh", "-c", script, NULL};
+    int delivered = 0;
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    opts.log_file = log_path;
+    opts.on_log_line = slow_log_callback;
+    opts.log_ud = &delivered;
+    opts.cancel_grace_ms = 100;
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(cbm_subprocess_spawn(&opts, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    const struct timespec fill_delay = {0, 150000000L};
+    (void)cbm_nanosleep(&fill_delay, NULL);
+    cbm_proc_result_t result = {0};
+    uint64_t before = cbm_now_ms();
+    cbm_proc_poll_t first = cbm_subprocess_poll(process, &result);
+    uint64_t elapsed = cbm_now_ms() - before;
+    bool cancel_accepted = cbm_subprocess_request_cancel(process);
+    bool terminal = poll_until_terminal(process, 3000, &result);
+    if (terminal) {
+        cbm_subprocess_destroy(process);
+    }
+    (void)unlink(log_path);
+
+    ASSERT_EQ(first, CBM_PROC_POLL_RUNNING);
+    ASSERT_LT(elapsed, 500);
+    ASSERT_LT(delivered, 400);
+    ASSERT_TRUE(cancel_accepted);
+    ASSERT_TRUE(terminal);
+    PASS();
+#endif
+}
+
+TEST(subprocess_posix_child_closes_unrelated_descriptors) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX descriptor-inheritance probe; Windows uses a handle allow-list");
+#else
+    char sentinel_path[] = "/tmp/cbm-subprocess-sentinel-XXXXXX";
+    int sentinel = cbm_mkstemp(sentinel_path);
+    ASSERT_TRUE(sentinel > STDERR_FILENO);
+    int flags = fcntl(sentinel, F_GETFD);
+    ASSERT_TRUE(flags >= 0);
+    ASSERT_EQ(fcntl(sentinel, F_SETFD, flags & ~FD_CLOEXEC), 0);
+    char fd_text[32];
+    snprintf(fd_text, sizeof(fd_text), "%d", sentinel);
+    const char *script = "if [ -e /dev/fd/$1 ]; then exit 42; else exit 0; fi";
+    const char *argv[] = {"/bin/sh", "-c", script, "cbm-fd-probe", fd_text, NULL};
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    cbm_proc_result_t result;
+    int run_rc = cbm_subprocess_run(&opts, &result);
+    (void)close(sentinel);
+    (void)unlink(sentinel_path);
+
+    ASSERT_EQ(run_rc, 0);
+    ASSERT_EQ(result.outcome, CBM_PROC_CLEAN);
+    ASSERT_EQ(result.exit_code, 0);
+    PASS();
+#endif
+}
+
+TEST(subprocess_root_exit_drains_surviving_descendant) {
+#ifdef _WIN32
+    SKIP_PLATFORM("POSIX process-group descendant probe; native Windows coverage pending");
+#else
+    char pid_path[64];
+    ASSERT_TRUE(make_tree_pid_path(pid_path));
+    const char *script = "sleep 30 & child=$!; echo $child > \"$1\"; exit 0";
+    const char *argv[] = {"/bin/sh", "-c", script, "cbm-root-exit", pid_path, NULL};
+    cbm_proc_opts_t opts = {0};
+    opts.bin = "/bin/sh";
+    opts.argv = argv;
+    opts.cancel_grace_ms = 100;
+    cbm_subprocess_t *process = NULL;
+    ASSERT_EQ(cbm_subprocess_spawn(&opts, &process), 0);
+    ASSERT_NOT_NULL(process);
+
+    pid_t descendant = -1;
+    uint64_t deadline = cbm_now_ms() + 1000;
+    while (descendant <= 1 && cbm_now_ms() < deadline) {
+        FILE *file = fopen(pid_path, "r");
+        long value = -1;
+        if (file) {
+            if (fscanf(file, "%ld", &value) == 1) {
+                descendant = (pid_t)value;
+            }
+            (void)fclose(file);
+        }
+        subprocess_test_pause();
+    }
+    cbm_proc_result_t result = {0};
+    bool terminal = descendant > 1 && poll_until_terminal(process, 2500, &result);
+    bool descendant_gone = terminal && wait_pid_gone(descendant, 1000);
+    if (!terminal) {
+        force_probe_cleanup(-1, descendant);
+        cbm_proc_result_t cleanup_result;
+        if (poll_until_terminal(process, 1000, &cleanup_result)) {
+            cbm_subprocess_destroy(process);
+        }
+    } else {
+        cbm_subprocess_destroy(process);
+    }
+    (void)unlink(pid_path);
+
+    ASSERT_TRUE(descendant > 1);
+    ASSERT_TRUE(terminal);
+    ASSERT_TRUE(descendant_gone);
+    ASSERT_EQ(result.outcome, CBM_PROC_CLEAN);
+    ASSERT_TRUE(result.tree_quiesced);
+    PASS();
+#endif
+}
+
+/* ── Layer 4: Windows command-line quoting (pure; every platform) ─────────────
  *
  * The Windows index-worker "crash" was a quoting bug: the Windows spawn wrapped each
  * argv element in bare quotes without escaping, so a JSON argument like
@@ -264,10 +721,18 @@ static bool cmdline_roundtrips(const char *const *argv) {
  * Round-trips, AND the emitted line must contain an ESCAPED quote (\") — the bare
  * `"%s"` wrap that caused the bug never would. */
 TEST(win_cmdline_index_worker_json) {
-    const char *const argv[] = {"C:/bin/cbm.exe",           "cli",
-                                "--index-worker",           "index_repository",
-                                "{\"repo_path\":\"C:/r\"}", "--response-out",
-                                "C:/c/w.response",          NULL};
+    const char *const argv[] = {
+        "C:/bin/cbm.exe",
+        "cli",
+        "--index-worker",
+        "--index-worker-build",
+        "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+        "index_repository",
+        "{\"repo_path\":\"C:/r\"}",
+        "--response-out",
+        "C:/c/w.response",
+        NULL,
+    };
     ASSERT(cmdline_roundtrips(argv));
     char cmd[4096];
     ASSERT(cbm_build_win_cmdline(cmd, sizeof(cmd), argv));
@@ -330,6 +795,14 @@ SUITE(subprocess) {
     RUN_TEST(subprocess_run_hang_is_hang);
     RUN_TEST(subprocess_run_spawn_failure);
     RUN_TEST(subprocess_run_null_bin_rejected);
+    RUN_TEST(subprocess_spawn_returns_while_child_is_running);
+    RUN_TEST(subprocess_natural_completion_is_cached_across_polls);
+    RUN_TEST(subprocess_cancel_is_idempotent_and_kills_ignoring_tree);
+    RUN_TEST(subprocess_quiet_timeout_kills_ignoring_tree);
+    RUN_TEST(subprocess_cancel_grace_is_hard_capped);
+    RUN_TEST(subprocess_poll_has_a_strict_log_delivery_budget);
+    RUN_TEST(subprocess_posix_child_closes_unrelated_descriptors);
+    RUN_TEST(subprocess_root_exit_drains_surviving_descendant);
     RUN_TEST(win_cmdline_index_worker_json);
     RUN_TEST(win_cmdline_roundtrip_battery);
     RUN_TEST(win_cmdline_overflow_rejected);