perf(pipeline): parallelize semantic-manifest hashing

The manifest's per-file sha256 loop was single-threaded -- tens of
thousands of file reads in sequence, the second-largest block of a
kernel-scale incremental run after publication (~20s by residual). The
hash helper is pure per-file work, so files now fan out across
cbm_default_worker_count workers on a stride; ASSEMBLY stays serial and
in discovery order, so the manifest bytes are identical to the serial
build's -- the exactness doctrine is untouched, only the wall clock
moves. Repos under 64 files keep the serial path outright.

A worker that fails to spawn leaves its stride to the calling thread,
so every index is hashed exactly once regardless of thread-creation
failures.

Pinned by a threshold-crossing test: a 72-file repo must route NOOP on
unchanged bytes -- which stands entirely on two parallel builds
producing byte-identical manifests -- and still classify a single edit
into the closure route.

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

src/pipeline/pipeline_incremental.c

@@ -27,6 +27,7 @@ enum { INCR_RING_BUF = 4, INCR_RING_MASK = 3, INCR_TS_BUF = 24 };
 #include "foundation/hash_table.h"
 #include "foundation/compat.h"
 #include "foundation/compat_fs.h"
+#include "foundation/compat_thread.h"
 #include "foundation/platform.h"
 #include "foundation/sha256.h"
 
@@ -386,6 +387,40 @@ static int semantic_manifest_walk_controls(semantic_manifest_builder_t *builder,
     return rc;
 }
 
+/* Parallel pre-hash of the discovery list. Hashing dominates manifest cost
+ * (tens of thousands of file reads; single-threaded it was the second-
+ * largest block of a kernel-scale incremental run after publication), and
+ * semantic_manifest_hash_file is pure per-file work, so the files fan out
+ * across workers while ASSEMBLY stays serial and in discovery order — the
+ * manifest bytes are identical to the serial build's, only sooner. */
+typedef struct {
+    const cbm_file_info_t *files;
+    int file_count;
+    int worker_id;
+    int worker_count;
+    char (*shas)[CBM_SHA256_HEX_LEN + 1];
+    int64_t *mtimes;
+    int64_t *sizes;
+    int8_t *rcs;
+} manifest_hash_worker_t;
+
+static void *manifest_hash_worker(void *arg) {
+    manifest_hash_worker_t *w = (manifest_hash_worker_t *)arg;
+    for (int i = w->worker_id; i < w->file_count; i += w->worker_count) {
+        if (!w->files[i].rel_path || !w->files[i].rel_path[0]) {
+            w->rcs[i] = 1; /* assembly skips exactly as the serial path did */
+            continue;
+        }
+        w->rcs[i] = semantic_manifest_hash_file(w->files[i].path, w->shas[i], &w->mtimes[i],
+                                                &w->sizes[i]) == 0
+                        ? 0
+                        : -1;
+    }
+    return NULL;
+}
+
+enum { MANIFEST_PARALLEL_MIN_FILES = 64 };
+
 int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_path,
                                          const cbm_file_info_t *files, int file_count,
                                          char **excluded_dirs, int excluded_count,
@@ -427,9 +462,83 @@ int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_p
         rc = semantic_manifest_add_digest(&builder, project, CBM_SEMANTIC_INPUT_PROJECT_CONFIG,
                                           project_config_digest, 0, 0);
     }
-    for (int i = 0; rc == 0 && i < file_count; i++) {
-        rc = semantic_manifest_add(&builder, project, files[i].rel_path, files[i].path);
+    struct timespec t_hash;
+    cbm_clock_gettime(CLOCK_MONOTONIC, &t_hash);
+    int hash_workers = cbm_default_worker_count(true);
+    if (rc == 0 && file_count >= MANIFEST_PARALLEL_MIN_FILES && hash_workers > SKIP_ONE) {
+        char (*shas)[CBM_SHA256_HEX_LEN + 1] = malloc((size_t)file_count * sizeof(*shas));
+        int64_t *mtimes = calloc((size_t)file_count, sizeof(int64_t));
+        int64_t *sizes = calloc((size_t)file_count, sizeof(int64_t));
+        int8_t *rcs = calloc((size_t)file_count, sizeof(int8_t));
+        cbm_thread_t *threads = calloc((size_t)hash_workers, sizeof(cbm_thread_t));
+        manifest_hash_worker_t *workers = calloc((size_t)hash_workers, sizeof(*workers));
+        if (shas && mtimes && sizes && rcs && threads && workers) {
+            int spawned = 0;
+            for (int w = 0; w < hash_workers; w++) {
+                workers[w] = (manifest_hash_worker_t){.files = files,
+                                                      .file_count = file_count,
+                                                      .worker_id = w,
+                                                      .worker_count = hash_workers,
+                                                      .shas = shas,
+                                                      .mtimes = mtimes,
+                                                      .sizes = sizes,
+                                                      .rcs = rcs};
+                if (cbm_thread_create(&threads[w], 0, manifest_hash_worker, &workers[w]) != 0) {
+                    break;
+                }
+                spawned++;
+            }
+            /* Workers that failed to spawn leave their stride to this
+             * thread: run the remaining strides inline so every index is
+             * covered exactly once. */
+            for (int w = spawned; w < hash_workers; w++) {
+                workers[w] = (manifest_hash_worker_t){.files = files,
+                                                      .file_count = file_count,
+                                                      .worker_id = w,
+                                                      .worker_count = hash_workers,
+                                                      .shas = shas,
+                                                      .mtimes = mtimes,
+                                                      .sizes = sizes,
+                                                      .rcs = rcs};
+                (void)manifest_hash_worker(&workers[w]);
+            }
+            for (int w = 0; w < spawned; w++) {
+                cbm_thread_join(&threads[w]);
+            }
+            for (int i = 0; rc == 0 && i < file_count; i++) {
+                if (rcs[i] == 1) {
+                    continue;
+                }
+                if (rcs[i] != 0) {
+                    cbm_log_error("semantic_manifest.err", "path", files[i].path);
+                    rc = CBM_NOT_FOUND;
+                    break;
+                }
+                if (cbm_ht_has(builder.seen_paths, files[i].rel_path)) {
+                    continue;
+                }
+                rc = semantic_manifest_add_digest(&builder, project, files[i].rel_path, shas[i],
+                                                  mtimes[i], sizes[i]);
+            }
+        } else {
+            for (int i = 0; rc == 0 && i < file_count; i++) {
+                rc = semantic_manifest_add(&builder, project, files[i].rel_path, files[i].path);
+            }
+        }
+        free(shas);
+        free(mtimes);
+        free(sizes);
+        free(rcs);
+        free(threads);
+        free(workers);
+    } else {
+        for (int i = 0; rc == 0 && i < file_count; i++) {
+            rc = semantic_manifest_add(&builder, project, files[i].rel_path, files[i].path);
+        }
     }
+    cbm_log_info("pass.timing", "pass", "semantic_manifest_hash", "files", itoa_buf(file_count),
+                 "workers", itoa_buf(hash_workers), "elapsed_ms",
+                 itoa_buf((int)elapsed_ms(t_hash)));
     if (rc == 0) {
         rc = semantic_manifest_walk_controls(&builder, project, repo_path, "", 0, excluded_dirs,
                                              excluded_count);

tests/test_pipeline.c

@@ -2367,6 +2367,47 @@ static void closure_fresh_full(const char *tmp, const char *db_path, int *out_no
     cbm_pipeline_free(full);
 }
 
+/* The manifest hash fans out across workers above a file-count threshold;
+ * below it the serial path runs. Byte-identity of the parallel build is
+ * what the exact no-op route stands on, so pin it with a repo big enough
+ * to take the parallel path on BOTH builds: unchanged bytes must route
+ * NOOP, and one changed file must still classify exactly. */
+TEST(pipeline_parallel_manifest_is_byte_stable_above_threshold) {
+    char tmp[256];
+    snprintf(tmp, sizeof(tmp), "/tmp/cbm_par_manifest_XXXXXX");
+    ASSERT_NOT_NULL(cbm_mkdtemp(tmp));
+    for (int i = 0; i < 72; i++) {
+        char name[64];
+        char body[128];
+        snprintf(name, sizeof(name), "mod_%02d.py", i);
+        snprintf(body, sizeof(body), "def par_manifest_%02d():\n    return %d\n", i, i);
+        write_temp_file(tmp, name, body);
+    }
+    char db[512];
+    snprintf(db, sizeof(db), "%s/manifest.db", tmp);
+    cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(baseline);
+    ASSERT_EQ(cbm_pipeline_run(baseline), 0);
+    cbm_pipeline_free(baseline);
+
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *unchanged = cbm_pipeline_new(tmp, db, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(unchanged);
+    ASSERT_EQ(cbm_pipeline_run(unchanged), 0);
+    ASSERT_EQ(cbm_pipeline_incremental_test_last_route(), CBM_INCREMENTAL_ROUTE_NOOP);
+    cbm_pipeline_free(unchanged);
+
+    write_temp_file(tmp, "mod_07.py", "def par_manifest_07():\n    return 700\n");
+    cbm_pipeline_incremental_test_reset_faults();
+    cbm_pipeline_t *edited = cbm_pipeline_new(tmp, db, CBM_MODE_FULL);
+    ASSERT_NOT_NULL(edited);
+    ASSERT_EQ(cbm_pipeline_run(edited), 0);
+    ASSERT_EQ(cbm_pipeline_incremental_test_last_route(), CBM_INCREMENTAL_ROUTE_CLOSURE_REPAIR);
+    cbm_pipeline_free(edited);
+    th_rmtree(tmp);
+    PASS();
+}
+
 TEST(pipeline_closure_repair_body_edit_converges_with_fresh_full) {
     char tmp[256];
     snprintf(tmp, sizeof(tmp), "/tmp/cbm_closure_body_XXXXXX");
@@ -11860,6 +11901,7 @@ SUITE(pipeline) {
  * the default all-suite run still executes it. */
 SUITE(pipeline_semantic_manifest_repro) {
     RUN_TEST(pipeline_incremental_repoints_call_reference_without_stale_edge);
+    RUN_TEST(pipeline_parallel_manifest_is_byte_stable_above_threshold);
     RUN_TEST(pipeline_closure_repair_body_edit_converges_with_fresh_full);
     RUN_TEST(pipeline_closure_repair_removed_def_drops_dependent_edge);
     RUN_TEST(pipeline_closure_repair_added_name_declines_to_full);