From 2443e92bece270d1207c715962707faee23387c3 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:21:46 -0700 Subject: [PATCH 1/7] fix: preserve CPU headroom during background indexing Fixes #1084 Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- src/daemon/application.c | 11 ++-- src/mcp/mcp.c | 6 ++- src/pipeline/pipeline.c | 35 +++++++++++-- src/pipeline/pipeline.h | 12 +++++ src/pipeline/pipeline_incremental.c | 6 +-- tests/test_daemon_application.c | 6 +++ tests/test_mcp.c | 68 ++++++++++++++++++++++++ tests/test_pipeline.c | 80 +++++++++++++++++++++++++++++ 8 files changed, 210 insertions(+), 14 deletions(-) diff --git a/src/daemon/application.c b/src/daemon/application.c index 22c62029b4..91b5937ece 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -2199,10 +2199,10 @@ static char *application_auto_index_args(cbm_daemon_application_t *application, return NULL; } yyjson_mut_doc_set_root(document, root); - char *args = yyjson_mut_obj_add_strcpy(document, root, "repo_path", root_path) && - application_index_args_add_policy(application, document, root) - ? yyjson_mut_write(document, 0, NULL) - : NULL; + bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", root_path) && + application_index_args_add_policy(application, document, root) && + yyjson_mut_obj_add_bool(document, root, "_background", true); + char *args = encoded ? yyjson_mut_write(document, 0, NULL) : NULL; yyjson_mut_doc_free(document); return args; } @@ -3894,7 +3894,8 @@ static int application_background_index(cbm_daemon_application_t *application, } yyjson_mut_doc_set_root(document, root); bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", canonical_root) && - application_index_args_add_policy(application, document, root); + application_index_args_add_policy(application, document, root) && + yyjson_mut_obj_add_bool(document, root, "_background", true); char *default_project = cbm_project_name_from_path(canonical_root); bool custom_project = project_name[0] && (!default_project || strcmp(default_project, project_name) != 0); diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index c877d51102..0d15ac0175 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -11109,7 +11109,8 @@ static char *index_run_supervised_path(cbm_mcp_server_t *srv, const char *root_p yyjson_mut_val *root = yyjson_mut_obj(doc); yyjson_mut_doc_set_root(doc, root); if (!yyjson_mut_obj_add_strcpy(doc, root, "repo_path", root_path) || - !cbm_mcp_index_policy_add_to_args(doc, root, &policy)) { + !cbm_mcp_index_policy_add_to_args(doc, root, &policy) || + !yyjson_mut_obj_add_bool(doc, root, "_background", true)) { yyjson_mut_doc_free(doc); return NULL; } @@ -11498,6 +11499,7 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); char *mode_str = cbm_mcp_get_string_arg(args, "mode"); char *name_override = cbm_mcp_get_string_arg(args, "name"); + bool background = cbm_mcp_get_bool_arg(args, "_background"); cbm_normalize_path_sep(repo_path); if (!repo_path) { @@ -11660,6 +11662,7 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { free(repo_path); return cbm_mcp_text_result("failed to create pipeline", true); } + cbm_pipeline_set_background(p, background); if (name_override && name_override[0] && !cbm_pipeline_set_project_name(p, name_override)) { cbm_pipeline_free(p); mcp_project_mutation_end(srv, mutation_project); @@ -18106,6 +18109,7 @@ static void *autoindex_thread(void *arg) { cbm_log_warn("autoindex.err", "msg", "pipeline_create_failed"); return NULL; } + cbm_pipeline_set_background(p, true); /* Block until any concurrent pipeline finishes */ cbm_pipeline_lock(); diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index ba41b877c8..f2aeb47bed 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -71,6 +71,18 @@ static inline void *intptr_to_ptr(intptr_t v) { * Atomic spinlock: 0 = free, 1 = locked. */ static atomic_int g_pipeline_busy = 0; +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API +static atomic_int g_last_worker_count = 0; + +void cbm_pipeline_worker_count_test_reset(void) { + atomic_store(&g_last_worker_count, 0); +} + +int cbm_pipeline_worker_count_test_last(void) { + return atomic_load(&g_last_worker_count); +} +#endif + #if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API static atomic_bool g_persist_test_fail_after_stage_dump = false; static atomic_bool g_persist_test_cancel_after_predump = false; @@ -192,6 +204,7 @@ struct cbm_pipeline { bool persistence; /* write .codebase-memory/graph.db.zst after indexing */ cbm_index_resource_policy_t resource_policy; cbm_index_resource_violation_t resource_violation; + bool background; /* reserve CPU headroom when no user is waiting */ /* Snapshot of the artifact export failure of THIS run (set only by * export_after_publish failure, zeroed at run start, cleared on success). @@ -400,6 +413,12 @@ const char *cbm_pipeline_export_error(const cbm_pipeline_t *p) { return p ? p->export_error : ""; } +void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background) { + if (p) { + p->background = background; + } +} + bool cbm_pipeline_set_project_name(cbm_pipeline_t *p, const char *name) { if (!p || !name || !name[0]) { return false; @@ -613,12 +632,18 @@ void cbm_pipeline_set_committed_counts(cbm_pipeline_t *p, int nodes, int edges) * crasher; a parallel re-run would race the marker. Honour that override * everywhere the worker count drives the parallel/sequential decision, so the * whole extraction phase collapses to the deterministic sequential path. */ -static int effective_worker_count(bool initial) { +int cbm_pipeline_worker_count(const cbm_pipeline_t *p) { const char *st = getenv("CBM_INDEX_SINGLE_THREAD"); + int workers; if (st && st[0] == '1') { - return 1; + workers = 1; + } else { + workers = cbm_default_worker_count(!p || !p->background); } - return cbm_default_worker_count(initial); +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API + atomic_store(&g_last_worker_count, workers); +#endif + return workers; } /* Resolve the DB path for this pipeline. Caller must free(). */ @@ -2590,7 +2615,7 @@ static int run_githistory(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx) { gh_compute_arg_t gh_arg = {.repo_path = ctx->repo_path, .result = &gh_result}; if (p->mode != CBM_MODE_FAST) { - if (effective_worker_count(true) > SKIP_ONE) { + if (cbm_pipeline_worker_count(p) > SKIP_ONE) { if (cbm_thread_create(&gh_thread, 0, gh_compute_thread_fn, &gh_arg) == 0) { gh_threaded = true; } @@ -2690,7 +2715,7 @@ static int run_extraction_phase(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, return CBM_NOT_FOUND; } - int worker_count = effective_worker_count(true); + int worker_count = cbm_pipeline_worker_count(p); CBM_PROF_START(t_extract_total); int rc = (worker_count > SKIP_ONE && file_count > MIN_FILES_FOR_PARALLEL) ? run_parallel_pipeline(p, ctx, files, file_count, worker_count, &t) diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index c1d571ecdc..e525136737 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -59,6 +59,18 @@ void cbm_pipeline_set_resource_policy(cbm_pipeline_t *p, const cbm_index_resourc /* Copy the exact discovery violation from the most recent run. */ void cbm_pipeline_get_resource_violation(const cbm_pipeline_t *p, cbm_index_resource_violation_t *violation); +/* Mark work that runs without a waiting user. Background pipelines reserve + * CPU headroom; foreground pipelines retain the initial-index all-core policy. */ +void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background); + +/* Resolve the worker policy for this pipeline, including environment and + * crash-recovery overrides. */ +int cbm_pipeline_worker_count(const cbm_pipeline_t *p); + +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API +void cbm_pipeline_worker_count_test_reset(void); +int cbm_pipeline_worker_count_test_last(void); +#endif /* Snapshot of the artifact export failure of the last cbm_pipeline_run, or "" * when the run succeeded / did not reach post-publish export. Used to diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 154518658d..4e7d61dc33 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -1212,7 +1212,7 @@ static int run_extract_resolve(cbm_pipeline_ctx_t *ctx, cbm_file_info_t *changed * a full build takes, which is what makes its output converge. */ #define MIN_FILES_FOR_PARALLEL_INCR 50 - int worker_count = cbm_default_worker_count(true); + int worker_count = cbm_pipeline_worker_count(ctx->pipeline); bool use_parallel = closure != NULL || (worker_count > SKIP_ONE && ci > MIN_FILES_FOR_PARALLEL_INCR); @@ -1590,7 +1590,7 @@ static int closure_probe_surfaces(cbm_pipeline_t *p, const char *project, _Atomic int64_t probe_ids; atomic_init(&probe_ids, cbm_gbuf_next_id(probe_gbuf)); rc = cbm_parallel_extract(&probe_ctx, probe_files, probe_count, cache, &probe_ids, - cbm_default_worker_count(true)); + cbm_pipeline_worker_count(p)); } if (rc == 0) { char **def_modules = (char **)calloc((size_t)probe_count, sizeof(char *)); @@ -2061,7 +2061,7 @@ static int run_closure_delta(cbm_pipeline_t *p, const char *db_path, const char elig[elig_count++] = row; } } - int workers = cbm_default_worker_count(true); + int workers = cbm_pipeline_worker_count(p); if (workers < 1) { workers = 1; } diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index 7f4c9cc288..50abe5a8c1 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -1375,6 +1375,7 @@ typedef struct { * itself instead, which is the edge a reader on another thread needs. */ atomic_bool args_captured[APP_FAKE_MAX_ATTEMPTS]; size_t memory_budgets[APP_FAKE_MAX_ATTEMPTS]; + bool background_requests[APP_FAKE_MAX_ATTEMPTS]; } app_fake_worker_context_t; typedef struct { @@ -1442,6 +1443,8 @@ static int app_fake_worker_start(void *opaque, const char *args_json, size_t mem args_json ? args_json : ""); atomic_store(&context->args_captured[worker->attempt], true); context->memory_budgets[worker->attempt] = memory_budget_bytes; + context->background_requests[worker->attempt] = + cbm_mcp_get_bool_arg(args_json, "_background"); if (marker_file) { (void)snprintf(context->marker_paths[worker->attempt], APP_TEST_PATH_CAP, "%s", marker_file); @@ -2202,6 +2205,7 @@ TEST(daemon_application_initialize_coalesces_auto_index_for_full_sessions) { ASSERT_TRUE(first_initialized); ASSERT_TRUE(first_owned); ASSERT_TRUE(auto_policy_propagated); + ASSERT_TRUE(fake.background_requests[0]); ASSERT_TRUE(second_initialized); ASSERT_TRUE(coalesced); ASSERT_TRUE(restricted_disconnect_kept_job); @@ -3634,6 +3638,7 @@ TEST(daemon_application_request_cancel_preserves_persistent_watch_and_session) { ASSERT_TRUE(thread_started); ASSERT_TRUE(subscribed); ASSERT_TRUE(worker_started); + ASSERT_FALSE(fixture.fake.background_requests[0]); ASSERT_TRUE(request_returned); ASSERT_TRUE(thread_joined); ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED); @@ -3931,6 +3936,7 @@ TEST(daemon_application_final_cancel_drains_admitted_watcher_job) { ASSERT_TRUE(fixture_ready); ASSERT_TRUE(thread_started); ASSERT_TRUE(worker_started); + ASSERT_TRUE(fixture.fake.background_requests[0]); ASSERT_TRUE(job_active); ASSERT_EQ(watches_after_cancel, 0); ASSERT_TRUE(thread_joined); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index 22af972af2..34563e0c9a 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -20006,6 +20006,73 @@ TEST(autoindex_limit_guards_git_root_issue713) { PASS(); } +TEST(mcp_auto_index_in_process_uses_background_worker_policy) { + char cache[256]; + char repo[512]; + (void)snprintf(cache, sizeof(cache), "%s/cbm-autoindex-workers-XXXXXX", cbm_tmpdir()); + bool cache_ready = cbm_mkdtemp(cache) != NULL; + (void)snprintf(repo, sizeof(repo), "%s/repo", cache); + char source[640]; + (void)snprintf(source, sizeof(source), "%s/main.py", repo); + bool repo_ready = cache_ready && th_mkdir_p(repo) == 0 && + th_write_file(source, "def background_index():\n return True\n") == 0; + + mcp_test_env_backup_t environment[] = { + {.name = "CBM_CACHE_DIR"}, + {.name = "CBM_WORKERS"}, + {.name = "CBM_INDEX_SINGLE_THREAD"}, + }; + bool environment_saved = true; + for (size_t i = 0; i < sizeof(environment) / sizeof(environment[0]); i++) { + const char *value = getenv(environment[i].name); + environment[i].present = value != NULL; + environment[i].value = value ? strdup(value) : NULL; + environment_saved = environment_saved && (!value || environment[i].value); + } + bool environment_ready = + environment_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 && + cbm_unsetenv("CBM_WORKERS") == 0 && cbm_unsetenv("CBM_INDEX_SINGLE_THREAD") == 0; + + char old_cwd[CBM_SZ_4K] = {0}; + bool cwd_ready = repo_ready && environment_ready && cbm_getcwd(old_cwd, sizeof(old_cwd)) && + cbm_chdir(repo) == 0; + cbm_config_t *config = cwd_ready ? cbm_config_open(cache) : NULL; + bool config_ready = config && cbm_config_set(config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(config, CBM_CONFIG_AUTO_WATCH, "false") == 0; + cbm_pipeline_worker_count_test_reset(); + cbm_mcp_server_t *server = config_ready ? cbm_mcp_server_new(NULL) : NULL; + char *response = NULL; + if (server) { + cbm_mcp_server_set_config(server, config); + response = cbm_mcp_server_handle( + server, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}"); + cbm_mcp_server_free(server); /* joins the automatic index thread */ + } + int selected_workers = cbm_pipeline_worker_count_test_last(); + bool server_ready = server != NULL; + bool response_ready = response != NULL; + + free(response); + cbm_config_close(config); + if (cwd_ready) { + (void)cbm_chdir(old_cwd); + } + mcp_test_restore_env(environment, sizeof(environment) / sizeof(environment[0])); + bool cleaned = !cache_ready || th_rmtree(cache) == 0; + + ASSERT_TRUE(cache_ready); + ASSERT_TRUE(repo_ready); + ASSERT_TRUE(environment_saved); + ASSERT_TRUE(environment_ready); + ASSERT_TRUE(cwd_ready); + ASSERT_TRUE(config_ready); + ASSERT_TRUE(server_ready); + ASSERT_TRUE(response_ready); + ASSERT_EQ(selected_workers, cbm_default_worker_count(false)); + ASSERT_TRUE(cleaned); + PASS(); +} + /* ══════════════════════════════════════════════════════════════════ * #853 — auto_watch=false must ALSO gate the SUPERVISED fresh-index * watcher registration (keystone × #849 merge interaction) @@ -21361,6 +21428,7 @@ SUITE(mcp) { RUN_TEST(autoindex_limit_guards_non_git_root_issue713); RUN_TEST(autoindex_limit_admits_non_git_root_under_limit_issue713); RUN_TEST(autoindex_limit_guards_git_root_issue713); + RUN_TEST(mcp_auto_index_in_process_uses_background_worker_policy); } /* Kept separate so daemon-coordination regressions can be iterated without diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 56be6b88e3..32f4b1a4c9 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -117,6 +117,84 @@ TEST(pipeline_create_free) { PASS(); } +TEST(pipeline_background_worker_policy_preserves_overrides) { + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + cbm_unsetenv("CBM_WORKERS"); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + + cbm_pipeline_t *foreground = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + cbm_pipeline_t *background = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + ASSERT_NOT_NULL(foreground); + ASSERT_NOT_NULL(background); + cbm_pipeline_set_background(background, true); + + int foreground_default = cbm_pipeline_worker_count(foreground); + int background_default = cbm_pipeline_worker_count(background); + bool defaults_match = foreground_default == cbm_default_worker_count(true) && + background_default == cbm_default_worker_count(false) && + foreground_default >= 1 && background_default >= 1; + + cbm_setenv("CBM_WORKERS", "3", 1); + bool env_override = cbm_pipeline_worker_count(foreground) == 3 && + cbm_pipeline_worker_count(background) == 3; + + cbm_setenv("CBM_INDEX_SINGLE_THREAD", "1", 1); + bool recovery_override = cbm_pipeline_worker_count(foreground) == 1 && + cbm_pipeline_worker_count(background) == 1; + + cbm_pipeline_free(foreground); + cbm_pipeline_free(background); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + + ASSERT_TRUE(defaults_match); + ASSERT_TRUE(env_override); + ASSERT_TRUE(recovery_override); + PASS(); +} + +TEST(pipeline_background_policy_preserves_index_results) { + ASSERT_EQ(setup_test_repo(), 0); + char foreground_db[512]; + char background_db[512]; + (void)snprintf(foreground_db, sizeof(foreground_db), "%s/foreground.db", g_tmpdir); + (void)snprintf(background_db, sizeof(background_db), "%s/background.db", g_tmpdir); + + cbm_pipeline_t *foreground = + cbm_pipeline_new(g_tmpdir, foreground_db, CBM_MODE_FULL); + cbm_pipeline_t *background = + cbm_pipeline_new(g_tmpdir, background_db, CBM_MODE_FULL); + ASSERT_NOT_NULL(foreground); + ASSERT_NOT_NULL(background); + cbm_pipeline_set_background(background, true); + + int foreground_rc = cbm_pipeline_run(foreground); + int background_rc = cbm_pipeline_run(background); + int foreground_nodes = -1; + int foreground_edges = -1; + int background_nodes = -1; + int background_edges = -1; + cbm_pipeline_get_committed_counts(foreground, &foreground_nodes, &foreground_edges); + cbm_pipeline_get_committed_counts(background, &background_nodes, &background_edges); + + cbm_pipeline_free(foreground); + cbm_pipeline_free(background); + teardown_test_repo(); + + ASSERT_EQ(foreground_rc, 0); + ASSERT_EQ(background_rc, 0); + ASSERT_GT(foreground_nodes, 0); + ASSERT_EQ(background_nodes, foreground_nodes); + ASSERT_EQ(background_edges, foreground_edges); + PASS(); +} + TEST(pipeline_null_repo) { cbm_pipeline_t *p = cbm_pipeline_new(NULL, NULL, CBM_MODE_FULL); ASSERT_NULL(p); @@ -16284,6 +16362,8 @@ SUITE(pipeline) { RUN_TEST(pipeline_lock_release_allows_contender); /* Lifecycle */ RUN_TEST(pipeline_create_free); + RUN_TEST(pipeline_background_worker_policy_preserves_overrides); + RUN_TEST(pipeline_background_policy_preserves_index_results); RUN_TEST(pipeline_null_repo); RUN_TEST(pipeline_free_null); RUN_TEST(pipeline_cancel); From 9bfd5cbe609e24f2ee77dd05f48aefed2199bfba Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Mon, 21 Sep 2026 23:56:56 -0700 Subject: [PATCH 2/7] fix: route the remaining all-core sites and fix the review findings Addresses the four blockers from review: - Capture the expected worker count while CBM_WORKERS is still unset in mcp_auto_index_in_process_uses_background_worker_policy, so the assertion no longer reads the lane override restored just before it. - Give pipeline_background_policy_preserves_index_results a fixture past MIN_FILES_FOR_PARALLEL and drop any inherited CBM_WORKERS / CBM_INDEX_SINGLE_THREAD pin, so it really compares an all-core index against a headroom index instead of two sequential runs. - Route the two remaining all-core sites through cbm_pipeline_worker_count: the LSP surface pass and incremental manifest hashing. The pipeline is plumbed into cbm_pipeline_build_semantic_manifest for that. - Rename the hidden key _background to _cbm_background, matching the existing _cbm_index_policy convention, across every producer and consumer. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Du315CaKLufAYPEdEwcocq Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- src/daemon/application.c | 4 +- src/mcp/mcp.c | 4 +- src/pipeline/lsp_surface.c | 3 +- src/pipeline/pipeline.c | 2 +- src/pipeline/pipeline_incremental.c | 22 ++++++---- src/pipeline/pipeline_internal.h | 6 +-- tests/test_daemon_application.c | 2 +- tests/test_mcp.c | 5 ++- tests/test_pipeline.c | 64 ++++++++++++++++++++++------- 9 files changed, 78 insertions(+), 34 deletions(-) diff --git a/src/daemon/application.c b/src/daemon/application.c index 91b5937ece..d1f78841be 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -2201,7 +2201,7 @@ static char *application_auto_index_args(cbm_daemon_application_t *application, yyjson_mut_doc_set_root(document, root); bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", root_path) && application_index_args_add_policy(application, document, root) && - yyjson_mut_obj_add_bool(document, root, "_background", true); + yyjson_mut_obj_add_bool(document, root, "_cbm_background", true); char *args = encoded ? yyjson_mut_write(document, 0, NULL) : NULL; yyjson_mut_doc_free(document); return args; @@ -3895,7 +3895,7 @@ static int application_background_index(cbm_daemon_application_t *application, yyjson_mut_doc_set_root(document, root); bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", canonical_root) && application_index_args_add_policy(application, document, root) && - yyjson_mut_obj_add_bool(document, root, "_background", true); + yyjson_mut_obj_add_bool(document, root, "_cbm_background", true); char *default_project = cbm_project_name_from_path(canonical_root); bool custom_project = project_name[0] && (!default_project || strcmp(default_project, project_name) != 0); diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 0d15ac0175..cd7b5c7112 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -11110,7 +11110,7 @@ static char *index_run_supervised_path(cbm_mcp_server_t *srv, const char *root_p yyjson_mut_doc_set_root(doc, root); if (!yyjson_mut_obj_add_strcpy(doc, root, "repo_path", root_path) || !cbm_mcp_index_policy_add_to_args(doc, root, &policy) || - !yyjson_mut_obj_add_bool(doc, root, "_background", true)) { + !yyjson_mut_obj_add_bool(doc, root, "_cbm_background", true)) { yyjson_mut_doc_free(doc); return NULL; } @@ -11499,7 +11499,7 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); char *mode_str = cbm_mcp_get_string_arg(args, "mode"); char *name_override = cbm_mcp_get_string_arg(args, "name"); - bool background = cbm_mcp_get_bool_arg(args, "_background"); + bool background = cbm_mcp_get_bool_arg(args, "_cbm_background"); cbm_normalize_path_sep(repo_path); if (!repo_path) { diff --git a/src/pipeline/lsp_surface.c b/src/pipeline/lsp_surface.c index 4c3cde3363..95ed7f63b8 100644 --- a/src/pipeline/lsp_surface.c +++ b/src/pipeline/lsp_surface.c @@ -231,8 +231,9 @@ int cbm_lsp_surface_build_rows(const cbm_pipeline_ctx_t *ctx, const char *projec .rows = rows, }; atomic_init(&job.failed, false); + int workers = cbm_pipeline_worker_count(ctx ? ctx->pipeline : NULL); cbm_parallel_for(file_count, surface_row_one, &job, - (cbm_parallel_for_opts_t){.max_workers = 0, .force_pthreads = false}); + (cbm_parallel_for_opts_t){.max_workers = workers, .force_pthreads = false}); if (atomic_load_explicit(&job.failed, memory_order_relaxed)) { cbm_store_free_lsp_surfaces(rows, file_count); /* untouched rows are all NULL */ return -1; diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index f2aeb47bed..6f6655c5f0 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -2810,7 +2810,7 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p) { rc = mode_promoted ? cbm_pipeline_build_fresh_semantic_manifest(p, p->project_name, &baseline_manifest, &baseline_count) - : cbm_pipeline_build_semantic_manifest(p->project_name, p->repo_path, files, + : cbm_pipeline_build_semantic_manifest(p, p->project_name, p->repo_path, files, file_count, p->excluded_dirs, p->excluded_count, &p->git_ctx, p->userconfig, &baseline_manifest, &baseline_count); diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 4e7d61dc33..8c348c3f76 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -413,9 +413,9 @@ static void *manifest_hash_worker(void *arg) { 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, +int cbm_pipeline_build_semantic_manifest(const cbm_pipeline_t *p, const char *project, + const char *repo_path, const cbm_file_info_t *files, + int file_count, char **excluded_dirs, int excluded_count, const cbm_git_context_t *git_ctx, const cbm_userconfig_t *userconfig, cbm_file_hash_t **out, int *out_count) { @@ -456,7 +456,7 @@ int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_p } struct timespec t_hash; cbm_clock_gettime(CLOCK_MONOTONIC, &t_hash); - int hash_workers = cbm_default_worker_count(true); + int hash_workers = cbm_pipeline_worker_count(p); 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)); @@ -634,9 +634,9 @@ int cbm_pipeline_build_fresh_semantic_manifest(cbm_pipeline_t *p, const char *pr &fresh_ignored_total); } if (rc == 0) { - rc = cbm_pipeline_build_semantic_manifest(project, repo_path, fresh_files, fresh_file_count, - fresh_excluded, fresh_excluded_count, - &fresh_git_ctx, fresh_userconfig, out, out_count); + rc = cbm_pipeline_build_semantic_manifest( + p, project, repo_path, fresh_files, fresh_file_count, fresh_excluded, + fresh_excluded_count, &fresh_git_ctx, fresh_userconfig, out, out_count); } cbm_set_user_lang_config(previous_userconfig); cbm_git_context_free(&fresh_git_ctx); @@ -1602,8 +1602,12 @@ static int closure_probe_surfaces(cbm_pipeline_t *p, const char *project, if (def_modules && def_starts) { defs = cbm_pxc_collect_all_defs(NULL, &probe_arena, cache, probe_files, probe_count, project, def_modules, &def_count, def_starts); - rc = cbm_lsp_surface_build_rows(NULL, project, cache, probe_files, probe_count, defs, - def_starts, out_rows, out_count); + /* Extract left pipeline NULL so file errors are not recorded twice. + * Surface rows only borrow it for the background worker policy. */ + probe_ctx.pipeline = p; + rc = cbm_lsp_surface_build_rows(&probe_ctx, project, cache, probe_files, probe_count, + defs, def_starts, out_rows, out_count); + probe_ctx.pipeline = NULL; } else { rc = -1; } diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index 7f266eae98..c2a87e0bf3 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -808,9 +808,9 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil #define CBM_SEMANTIC_INPUT_GLOBAL_CONFIG CBM_SEMANTIC_INPUT_PREFIX "global-extension-config-v1" #define CBM_SEMANTIC_INPUT_PROJECT_CONFIG CBM_SEMANTIC_INPUT_PREFIX "project-extension-config-v1" -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, +int cbm_pipeline_build_semantic_manifest(const cbm_pipeline_t *p, const char *project, + const char *repo_path, const cbm_file_info_t *files, + int file_count, char **excluded_dirs, int excluded_count, const cbm_git_context_t *git_ctx, const cbm_userconfig_t *userconfig, cbm_file_hash_t **out, int *out_count); diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index 50abe5a8c1..0bec333a8a 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -1444,7 +1444,7 @@ static int app_fake_worker_start(void *opaque, const char *args_json, size_t mem atomic_store(&context->args_captured[worker->attempt], true); context->memory_budgets[worker->attempt] = memory_budget_bytes; context->background_requests[worker->attempt] = - cbm_mcp_get_bool_arg(args_json, "_background"); + cbm_mcp_get_bool_arg(args_json, "_cbm_background"); if (marker_file) { (void)snprintf(context->marker_paths[worker->attempt], APP_TEST_PATH_CAP, "%s", marker_file); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index 34563e0c9a..d334c47745 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -20057,6 +20057,9 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { if (cwd_ready) { (void)cbm_chdir(old_cwd); } + /* Capture while CBM_WORKERS is still unset. Restoring first makes the + * expected count follow the lane override instead of the headroom policy. */ + int expected = cbm_default_worker_count(false); mcp_test_restore_env(environment, sizeof(environment) / sizeof(environment[0])); bool cleaned = !cache_ready || th_rmtree(cache) == 0; @@ -20068,7 +20071,7 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { ASSERT_TRUE(config_ready); ASSERT_TRUE(server_ready); ASSERT_TRUE(response_ready); - ASSERT_EQ(selected_workers, cbm_default_worker_count(false)); + ASSERT_EQ(selected_workers, expected); ASSERT_TRUE(cleaned); PASS(); } diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 32f4b1a4c9..c541db5e2a 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -161,32 +161,68 @@ TEST(pipeline_background_worker_policy_preserves_overrides) { TEST(pipeline_background_policy_preserves_index_results) { ASSERT_EQ(setup_test_repo(), 0); + /* 64 fillers plus setup_test_repo's three files exceed MIN_FILES_FOR_PARALLEL + * (50). Below that threshold both policies take the sequential path, so the + * comparison stays green even with the background policy removed. */ + for (int i = 0; i < 64; i++) { + char name[32]; + char body[96]; + (void)snprintf(name, sizeof(name), "pad%02d.go", i); + (void)snprintf(body, sizeof(body), "package main\n\nfunc Pad%02d() int {\n\treturn %d\n}\n", + i, i); + if (th_write_file(TH_PATH(g_tmpdir, name), body) != 0) { + teardown_test_repo(); + FAIL("failed to pad fixture past MIN_FILES_FOR_PARALLEL"); + } + } + + /* An inherited CBM_WORKERS pin makes both runs use the same count. Drop it + * so this compares an all-core index against a headroom index. */ + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + if ((old_workers && !saved_workers) || (old_single && !saved_single)) { + free(saved_workers); + free(saved_single); + teardown_test_repo(); + FAIL("failed to save worker env"); + } + cbm_unsetenv("CBM_WORKERS"); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + char foreground_db[512]; char background_db[512]; (void)snprintf(foreground_db, sizeof(foreground_db), "%s/foreground.db", g_tmpdir); (void)snprintf(background_db, sizeof(background_db), "%s/background.db", g_tmpdir); - cbm_pipeline_t *foreground = - cbm_pipeline_new(g_tmpdir, foreground_db, CBM_MODE_FULL); - cbm_pipeline_t *background = - cbm_pipeline_new(g_tmpdir, background_db, CBM_MODE_FULL); - ASSERT_NOT_NULL(foreground); - ASSERT_NOT_NULL(background); - cbm_pipeline_set_background(background, true); - - int foreground_rc = cbm_pipeline_run(foreground); - int background_rc = cbm_pipeline_run(background); + cbm_pipeline_t *foreground = cbm_pipeline_new(g_tmpdir, foreground_db, CBM_MODE_FULL); + cbm_pipeline_t *background = cbm_pipeline_new(g_tmpdir, background_db, CBM_MODE_FULL); + int foreground_rc = -1; + int background_rc = -1; int foreground_nodes = -1; int foreground_edges = -1; int background_nodes = -1; int background_edges = -1; - cbm_pipeline_get_committed_counts(foreground, &foreground_nodes, &foreground_edges); - cbm_pipeline_get_committed_counts(background, &background_nodes, &background_edges); + bool created = foreground && background; + if (created) { + cbm_pipeline_set_background(background, true); + foreground_rc = cbm_pipeline_run(foreground); + background_rc = cbm_pipeline_run(background); + cbm_pipeline_get_committed_counts(foreground, &foreground_nodes, &foreground_edges); + cbm_pipeline_get_committed_counts(background, &background_nodes, &background_edges); + } cbm_pipeline_free(foreground); cbm_pipeline_free(background); teardown_test_repo(); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + ASSERT_TRUE(created); ASSERT_EQ(foreground_rc, 0); ASSERT_EQ(background_rc, 0); ASSERT_GT(foreground_nodes, 0); @@ -5458,8 +5494,8 @@ TEST(pipeline_semantic_manifest_rejects_non_directory_root) { cbm_file_hash_t *manifest = NULL; int manifest_count = -1; - int rc = cbm_pipeline_build_semantic_manifest("manifest-fail-closed", root_path, NULL, 0, NULL, - 0, NULL, NULL, &manifest, &manifest_count); + int rc = cbm_pipeline_build_semantic_manifest(NULL, "manifest-fail-closed", root_path, NULL, 0, + NULL, 0, NULL, NULL, &manifest, &manifest_count); cbm_pipeline_free_semantic_manifest(manifest, manifest_count > 0 ? manifest_count : 0); th_rmtree(tmp); From 26eece514fef17bef35cafb23b16c647c3358aaf Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Fri, 25 Sep 2026 07:55:29 -0700 Subject: [PATCH 3/7] fix(mcp): keep full width for a project's first index Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- src/mcp/mcp.c | 23 +++++++- src/pipeline/pipeline.c | 2 +- tests/test_mcp.c | 128 +++++++++++++++++++++++++++++++++------- 3 files changed, 129 insertions(+), 24 deletions(-) diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index cd7b5c7112..31efa9420e 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -11370,6 +11370,22 @@ static bool index_root_owner_resolve(const char *repo_path, char **owner_out, ch return false; } +/* Background CPU headroom is for a refresh of a graph that already serves. + * A project's first index keeps every core, even when it was started + * automatically: the caller is waiting on a graph that does not exist yet. + * An empty or unreadable file is not a committed index. Both sites that + * honour automatic indexing — handle_index_repository's _cbm_background + * flag and the in-process autoindex_thread — go through this so they + * cannot drift. */ +static bool project_has_committed_index(const char *project) { + if (!project || !project[0]) { + return false; + } + char db_path[CBM_SZ_1K]; + project_db_path(project, db_path, sizeof(db_path)); + return db_path[0] && project_db_is_servable(project, db_path); +} + /* The three heap strings handle_index_repository owns from * cbm_mcp_get_string_arg / resolved_repo_path_from_project_arg. One release * point keeps the dozen early-return paths in step; free(NULL) is a no-op, so @@ -11662,7 +11678,6 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { free(repo_path); return cbm_mcp_text_result("failed to create pipeline", true); } - cbm_pipeline_set_background(p, background); if (name_override && name_override[0] && !cbm_pipeline_set_project_name(p, name_override)) { cbm_pipeline_free(p); mcp_project_mutation_end(srv, mutation_project); @@ -11672,6 +11687,10 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { return cbm_mcp_text_result("invalid project name", true); } free(name_override); + /* Headroom follows the final project name, decided before artifact + * bootstrap can create the first local copy of this index. */ + cbm_pipeline_set_background(p, background && + project_has_committed_index(cbm_pipeline_project_name(p))); cbm_pipeline_set_persistence(p, persistence); cbm_pipeline_set_resource_policy(p, &resource_policy); @@ -18109,7 +18128,7 @@ static void *autoindex_thread(void *arg) { cbm_log_warn("autoindex.err", "msg", "pipeline_create_failed"); return NULL; } - cbm_pipeline_set_background(p, true); + cbm_pipeline_set_background(p, project_has_committed_index(cbm_pipeline_project_name(p))); /* Block until any concurrent pipeline finishes */ cbm_pipeline_lock(); diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 6f6655c5f0..be881942e0 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -204,7 +204,7 @@ struct cbm_pipeline { bool persistence; /* write .codebase-memory/graph.db.zst after indexing */ cbm_index_resource_policy_t resource_policy; cbm_index_resource_violation_t resource_violation; - bool background; /* reserve CPU headroom when no user is waiting */ + bool background; /* reserve CPU headroom when no user is waiting */ /* Snapshot of the artifact export failure of THIS run (set only by * export_after_publish failure, zeroed at run start, cleared on success). diff --git a/tests/test_mcp.c b/tests/test_mcp.c index d334c47745..f42f0dc6a6 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -20006,10 +20006,43 @@ TEST(autoindex_limit_guards_git_root_issue713) { PASS(); } -TEST(mcp_auto_index_in_process_uses_background_worker_policy) { +/* A servable project row is a committed index. user_version stays 0, so a + * later run cannot take the incremental no-op and skip recording workers. */ +static bool mcp_publish_committed_index(const char *root) { + char *project = root ? cbm_project_name_from_path(root) : NULL; + cbm_store_t *store = project ? cbm_store_open(project) : NULL; + bool published = store && cbm_store_upsert_project(store, project, root) == CBM_STORE_OK; + cbm_store_close(store); + free(project); + return published; +} + +#ifdef CBM_ENABLE_TEST_SEAMS +typedef struct { + const char *root; + bool published; +} mcp_committed_seed_t; + +/* maybe_auto_index returns before the thread when .db already + * exists. Publish the committed index from the count hook, which runs after + * that check and before autoindex_thread chooses a width. */ +static void mcp_autoindex_publish_committed(void *context) { + mcp_committed_seed_t *seed = context; + if (seed) { + seed->published = mcp_publish_committed_index(seed->root); + } +} +#endif + +/* First index, including one started automatically, uses every core. Headroom + * applies only once a servable project database already exists. + * automatic: in-process auto-index (initialize → autoindex_thread). + * Otherwise index_repository with _cbm_background, the daemon/watcher path. */ +static int mcp_assert_inprocess_worker_policy(bool committed, bool automatic) { char cache[256]; char repo[512]; - (void)snprintf(cache, sizeof(cache), "%s/cbm-autoindex-workers-XXXXXX", cbm_tmpdir()); + (void)snprintf(cache, sizeof(cache), "%s/cbm-worker-policy-%d%d-XXXXXX", cbm_tmpdir(), + committed ? 1 : 0, automatic ? 1 : 0); bool cache_ready = cbm_mkdtemp(cache) != NULL; (void)snprintf(repo, sizeof(repo), "%s/repo", cache); char source[640]; @@ -20029,24 +20062,57 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { environment[i].value = value ? strdup(value) : NULL; environment_saved = environment_saved && (!value || environment[i].value); } - bool environment_ready = - environment_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 && - cbm_unsetenv("CBM_WORKERS") == 0 && cbm_unsetenv("CBM_INDEX_SINGLE_THREAD") == 0; + bool environment_ready = environment_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 && + cbm_unsetenv("CBM_WORKERS") == 0 && + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD") == 0; + bool seeded = !committed; char old_cwd[CBM_SZ_4K] = {0}; - bool cwd_ready = repo_ready && environment_ready && cbm_getcwd(old_cwd, sizeof(old_cwd)) && - cbm_chdir(repo) == 0; - cbm_config_t *config = cwd_ready ? cbm_config_open(cache) : NULL; - bool config_ready = config && cbm_config_set(config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && - cbm_config_set(config, CBM_CONFIG_AUTO_WATCH, "false") == 0; - cbm_pipeline_worker_count_test_reset(); - cbm_mcp_server_t *server = config_ready ? cbm_mcp_server_new(NULL) : NULL; + bool cwd_ready = true; + cbm_config_t *config = NULL; + bool config_ready = true; + cbm_mcp_server_t *server = NULL; char *response = NULL; - if (server) { - cbm_mcp_server_set_config(server, config); - response = cbm_mcp_server_handle( - server, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}"); - cbm_mcp_server_free(server); /* joins the automatic index thread */ + if (automatic) { + cwd_ready = repo_ready && environment_ready && cbm_getcwd(old_cwd, sizeof(old_cwd)) && + cbm_chdir(repo) == 0; + config = cwd_ready ? cbm_config_open(cache) : NULL; + config_ready = config && cbm_config_set(config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(config, CBM_CONFIG_AUTO_WATCH, "false") == 0; + cbm_pipeline_worker_count_test_reset(); + server = config_ready ? cbm_mcp_server_new(NULL) : NULL; + if (server) { + cbm_mcp_server_set_config(server, config); +#ifdef CBM_ENABLE_TEST_SEAMS + mcp_committed_seed_t seed = {.root = repo, .published = !committed}; + if (committed) { + cbm_mcp_server_set_auto_index_count_test_hook( + server, mcp_autoindex_publish_committed, &seed); + } +#endif + response = cbm_mcp_server_handle( + server, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}"); +#ifdef CBM_ENABLE_TEST_SEAMS + seeded = seed.published; +#endif + cbm_mcp_server_free(server); /* joins the automatic index thread */ + } + } else { + if (committed && repo_ready && environment_ready) { + seeded = mcp_publish_committed_index(repo); + } + cbm_pipeline_worker_count_test_reset(); + server = repo_ready && environment_ready && seeded ? cbm_mcp_server_new(NULL) : NULL; + if (server) { + char repo_json[512]; + char args[CBM_SZ_1K]; + (void)snprintf(repo_json, sizeof(repo_json), "%s", repo); + cbm_normalize_path_sep(repo_json); + (void)snprintf(args, sizeof(args), "{\"_cbm_background\":true,\"repo_path\":\"%s\"}", + repo_json); + response = cbm_mcp_handle_tool(server, "index_repository", args); + cbm_mcp_server_free(server); + } } int selected_workers = cbm_pipeline_worker_count_test_last(); bool server_ready = server != NULL; @@ -20054,12 +20120,12 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { free(response); cbm_config_close(config); - if (cwd_ready) { + if (automatic && cwd_ready) { (void)cbm_chdir(old_cwd); } /* Capture while CBM_WORKERS is still unset. Restoring first makes the - * expected count follow the lane override instead of the headroom policy. */ - int expected = cbm_default_worker_count(false); + * expected count follow the lane override instead of this policy. */ + int expected = cbm_default_worker_count(!committed); mcp_test_restore_env(environment, sizeof(environment) / sizeof(environment[0])); bool cleaned = !cache_ready || th_rmtree(cache) == 0; @@ -20069,6 +20135,7 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { ASSERT_TRUE(environment_ready); ASSERT_TRUE(cwd_ready); ASSERT_TRUE(config_ready); + ASSERT_TRUE(seeded); ASSERT_TRUE(server_ready); ASSERT_TRUE(response_ready); ASSERT_EQ(selected_workers, expected); @@ -20076,6 +20143,22 @@ TEST(mcp_auto_index_in_process_uses_background_worker_policy) { PASS(); } +TEST(mcp_auto_index_in_process_fresh_project_uses_full_width) { + return mcp_assert_inprocess_worker_policy(false, true); +} + +TEST(mcp_auto_index_in_process_existing_index_keeps_headroom) { + return mcp_assert_inprocess_worker_policy(true, true); +} + +TEST(mcp_background_flag_fresh_project_uses_full_width) { + return mcp_assert_inprocess_worker_policy(false, false); +} + +TEST(mcp_background_flag_existing_index_keeps_headroom) { + return mcp_assert_inprocess_worker_policy(true, false); +} + /* ══════════════════════════════════════════════════════════════════ * #853 — auto_watch=false must ALSO gate the SUPERVISED fresh-index * watcher registration (keystone × #849 merge interaction) @@ -21431,7 +21514,10 @@ SUITE(mcp) { RUN_TEST(autoindex_limit_guards_non_git_root_issue713); RUN_TEST(autoindex_limit_admits_non_git_root_under_limit_issue713); RUN_TEST(autoindex_limit_guards_git_root_issue713); - RUN_TEST(mcp_auto_index_in_process_uses_background_worker_policy); + RUN_TEST(mcp_auto_index_in_process_fresh_project_uses_full_width); + RUN_TEST(mcp_auto_index_in_process_existing_index_keeps_headroom); + RUN_TEST(mcp_background_flag_fresh_project_uses_full_width); + RUN_TEST(mcp_background_flag_existing_index_keeps_headroom); } /* Kept separate so daemon-coordination regressions can be iterated without From a909f8418e9fb0fe52372d82f92c41a53e2f6fe7 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 20 Oct 2026 19:27:00 +0200 Subject: [PATCH 4/7] fix(daemon): let an explicit index join a running automatic job Automatic jobs (session auto-index, watcher re-index) now carry _cbm_background, and the job registry decides whether a new request may join the job already running for its project by comparing the index arguments after normalising defaults. An explicit index_repository for the same root therefore stopped matching the automatic job and was refused with "another index operation for this project is active with different options" -- the refusal the repo_path separator fold (9846c3f1) removed on Windows, now on every platform. _cbm_background chooses the worker count and never what is indexed, so drop it from the comparison the same way default-valued options are dropped. The explicit request joins the running job again. The new daemon_application test holds an auto-index job open with the fake worker, checks that its args carry the marker, and sends an explicit async index_repository for the same root. Without this change it is refused as an options conflict; with it the request joins (one worker start, one active job). The async fixture gains an optional stored config so the test can enable auto_index. Refs #1084 Signed-off-by: Martin Vogel --- src/daemon/application.c | 3 ++ tests/test_daemon_application.c | 86 +++++++++++++++++++++++++++++++-- 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/src/daemon/application.c b/src/daemon/application.c index 92e5cb873c..5c17aa6395 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -1943,6 +1943,9 @@ static bool application_index_args_normalize_defaults(yyjson_mut_val *root) { if (name && (!yyjson_mut_is_str(name) || yyjson_mut_get_len(name) == 0)) { (void)yyjson_mut_obj_remove_key(root, "name"); } + /* Automatic jobs carry _cbm_background, which picks the worker count and + * never what is indexed, so an explicit request joins them. */ + (void)yyjson_mut_obj_remove_key(root, "_cbm_background"); return true; } diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index 811cd31e04..4a86968242 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -6215,8 +6215,9 @@ typedef struct { uint32_t context_length; } app_async_fixture_t; -static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *label, - bool permanent) { +/* stored_config is optional; a test that needs auto_index passes its own. */ +static bool app_async_fixture_init_with_config(app_async_fixture_t *fixture, const char *label, + bool permanent, cbm_config_t *stored_config) { memset(fixture, 0, sizeof(*fixture)); app_fake_worker_context_init(&fixture->fake); fixture->ops = (cbm_daemon_application_worker_ops_t){ @@ -6227,7 +6228,8 @@ static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *lab .log_path = app_fake_worker_log_path, .destroy = app_fake_worker_destroy, }; - cbm_daemon_application_config_t config = {.worker_ops = &fixture->ops}; + cbm_daemon_application_config_t config = {.config = stored_config, + .worker_ops = &fixture->ops}; fixture->application = cbm_daemon_application_new(&config); if (!fixture->application) { return false; @@ -6248,6 +6250,11 @@ static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *lab &fixture->context_length); } +static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *label, + bool permanent) { + return app_async_fixture_init_with_config(fixture, label, permanent, NULL); +} + static cbm_daemon_runtime_application_session_t *app_async_session(app_async_fixture_t *fixture, cbm_daemon_client_id_t id) { cbm_daemon_runtime_application_session_t *session = app_test_open(&fixture->callbacks, id); @@ -6625,6 +6632,78 @@ TEST(daemon_application_cut_short_sync_index_advice_reaches_next_call) { * the daemon's only live session is refused async (the daemon would stop and * cancel the job as soon as the command exits), while status stays allowed and * a long-lived MCP session on the same temporary daemon remains a valid host. */ +/* Automatic indexing (session auto-index, watcher re-index) marks its job + * _cbm_background so the worker leaves CPU headroom. The marker picks the + * worker count, never what is indexed, so an explicit index_repository for the + * same root still JOINS the automatic job already running for it, the contract + * the repo_path fold above restored, instead of being refused as an options + * conflict. The worker is held, so the join is observed while the job runs. */ +TEST(daemon_application_explicit_index_joins_running_auto_index_job) { + app_env_backup_t cache_environment; + bool cache_saved = app_env_backup_capture(&cache_environment, "CBM_CACHE_DIR"); + char cache[APP_TEST_PATH_CAP]; + (void)snprintf(cache, sizeof(cache), "%s/cbm-app-auto-join-cache-XXXXXX", cbm_tmpdir()); + bool cache_ready = + cache_saved && cbm_mkdtemp(cache) != NULL && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0; + cbm_config_t *stored_config = cache_ready ? cbm_config_open(cache) : NULL; + bool config_ready = stored_config && + cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(stored_config, CBM_CONFIG_AUTO_WATCH, "false") == 0; + app_async_fixture_t fixture; + bool setup = config_ready && + app_async_fixture_init_with_config(&fixture, "auto-join", true, stored_config); + + cbm_daemon_runtime_application_session_t *automatic = + setup ? app_test_open(&fixture.callbacks, 7801) : NULL; + bool auto_started = automatic && + app_test_initialize_profile(&fixture.callbacks, automatic, fixture.root, + CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL) && + app_wait_for_atomic_int(&fixture.fake.starts, 1); + /* args_json is published by the args_captured release store; the + * background_requests slot is written after it, so read the args. */ + bool auto_marked = auto_started && + app_wait_for_atomic_bool(&fixture.fake.args_captured[0], true) && + cbm_mcp_get_bool_arg(fixture.fake.args_json[0], "_cbm_background"); + + cbm_daemon_runtime_application_session_t *explicit_session = + auto_marked ? app_async_session(&fixture, 7802) : NULL; + char *joined = + explicit_session ? app_async_call_now(&fixture, explicit_session, ",\"async\":true") : NULL; + int starts_while_running = atomic_load(&fixture.fake.starts); + size_t active_while_running = + setup ? cbm_daemon_application_active_jobs(fixture.application) : 0; + + if (explicit_session) { + fixture.callbacks.session_close(fixture.callbacks.context, explicit_session); + } + if (automatic) { + fixture.callbacks.session_cancel(fixture.callbacks.context, automatic); + fixture.callbacks.session_close(fixture.callbacks.context, automatic); + } + bool stopped = !setup || app_async_fixture_finish(&fixture); + cbm_config_close(stored_config); + bool cleaned = !cache_ready || th_rmtree(cache) == 0; + bool cache_restored = app_env_backup_restore(&cache_environment); + + ASSERT_TRUE(cache_ready); + ASSERT_TRUE(config_ready); + ASSERT_TRUE(setup); + ASSERT_TRUE(auto_started); + ASSERT_TRUE(auto_marked); + ASSERT_NOT_NULL(explicit_session); + ASSERT_NOT_NULL(joined); + ASSERT_FALSE(app_async_contains(joined, "different options")); + ASSERT_FALSE(app_async_contains(joined, "\"isError\":true")); + ASSERT_TRUE(app_async_contains(joined, "\"joined\":true")); + ASSERT_EQ(starts_while_running, 1); + ASSERT_EQ(active_while_running, 1); + ASSERT_TRUE(stopped); + ASSERT_TRUE(cleaned); + ASSERT_TRUE(cache_restored); + free(joined); + PASS(); +} + TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daemon) { app_async_fixture_t fixture; bool setup = app_async_fixture_init(&fixture, "async-refuse", false); @@ -6953,6 +7032,7 @@ SUITE(daemon_application) { RUN_TEST(daemon_application_async_and_status_validation_errors); RUN_TEST(daemon_application_cancelled_sync_index_reply_offers_async); RUN_TEST(daemon_application_cut_short_sync_index_advice_reaches_next_call); + RUN_TEST(daemon_application_explicit_index_joins_running_auto_index_job); RUN_TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daemon); RUN_TEST(daemon_application_unnamed_status_finds_async_job_of_root_owner_issue2134); RUN_TEST(daemon_application_unnamed_status_keeps_path_derived_key_issue2134); From 1d7ce9ba6ac0294d7e1a2ef3a3ae3c9c0259b996 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Wed, 21 Oct 2026 22:09:31 +0200 Subject: [PATCH 5/7] fix(pipeline): cap surface callbacks at the worker budget Signed-off-by: Martin Vogel --- src/pipeline/lsp_surface.c | 50 ++++++++++++++- src/pipeline/lsp_surface.h | 7 ++ src/pipeline/worker_pool.c | 21 ++++++ src/pipeline/worker_pool.h | 7 ++ tests/test_pipeline.c | 127 +++++++++++++++++++++++++++++++++++++ 5 files changed, 210 insertions(+), 2 deletions(-) diff --git a/src/pipeline/lsp_surface.c b/src/pipeline/lsp_surface.c index 5b9e283b61..66f6f3b25f 100644 --- a/src/pipeline/lsp_surface.c +++ b/src/pipeline/lsp_surface.c @@ -483,10 +483,12 @@ typedef struct { const CBMLSPDef *all_defs; const int *def_starts; cbm_lsp_surface_row_t *rows; + int file_count; + _Atomic int next_file; _Atomic bool failed; } surface_job_t; -static void surface_row_one(int i, void *arg) { +static void surface_row_build(int i, void *arg) { surface_job_t *job = (surface_job_t *)arg; if (atomic_load_explicit(&job->failed, memory_order_relaxed)) { return; @@ -526,6 +528,45 @@ static void surface_row_one(int i, void *arg) { } } +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +static cbm_lsp_surface_row_observer_fn g_row_observer; +static void *g_row_observer_ctx; + +void cbm_lsp_surface_test_set_row_observer(cbm_lsp_surface_row_observer_fn observer, void *ctx) { + g_row_observer = observer; + g_row_observer_ctx = ctx; +} +#endif + +static void surface_row_one(int i, void *arg) { +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (g_row_observer) { + g_row_observer(i, true, g_row_observer_ctx); + } +#endif + surface_row_build(i, arg); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (g_row_observer) { + g_row_observer(i, false, g_row_observer_ctx); + } +#endif +} + +/* The pool also runs work on its caller. Dispatch at most one coarse job per + * allowed participant, each stealing files, so the total active row builders + * cannot exceed the pipeline budget even when the caller takes a job. */ +static void surface_rows_worker(int worker, void *arg) { + (void)worker; + surface_job_t *job = arg; + for (;;) { + int i = atomic_fetch_add_explicit(&job->next_file, 1, memory_order_relaxed); + if (i >= job->file_count) { + break; + } + surface_row_one(i, job); + } +} + int cbm_lsp_surface_build_rows(const cbm_pipeline_ctx_t *ctx, const char *project, CBMFileResult **cache, const cbm_file_info_t *files, int file_count, const CBMLSPDef *all_defs, const int *def_starts, @@ -547,10 +588,15 @@ int cbm_lsp_surface_build_rows(const cbm_pipeline_ctx_t *ctx, const char *projec .all_defs = all_defs, .def_starts = def_starts, .rows = rows, + .file_count = file_count, }; + atomic_init(&job.next_file, 0); atomic_init(&job.failed, false); int workers = cbm_pipeline_worker_count(ctx ? ctx->pipeline : NULL); - cbm_parallel_for(file_count, surface_row_one, &job, + if (workers > file_count) { + workers = file_count; + } + cbm_parallel_for(workers, surface_rows_worker, &job, (cbm_parallel_for_opts_t){.max_workers = workers, .force_pthreads = false}); if (atomic_load_explicit(&job.failed, memory_order_relaxed)) { cbm_store_free_lsp_surfaces(rows, file_count); /* untouched rows are all NULL */ diff --git a/src/pipeline/lsp_surface.h b/src/pipeline/lsp_surface.h index dc198bb654..1a73b34274 100644 --- a/src/pipeline/lsp_surface.h +++ b/src/pipeline/lsp_surface.h @@ -68,4 +68,11 @@ int cbm_lsp_surface_python_namespace_from_json(CBMArena *arena, const char *json */ int cbm_lsp_surface_python_change(const char *old_json, const char *new_json); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +/* Bracket each file callback, including skipped/error rows. Install only with + * no surface build running; clear after the observed build returns. */ +typedef void (*cbm_lsp_surface_row_observer_fn)(int index, bool entering, void *ctx); +void cbm_lsp_surface_test_set_row_observer(cbm_lsp_surface_row_observer_fn observer, void *ctx); +#endif + #endif /* CBM_PIPELINE_LSP_SURFACE_H */ diff --git a/src/pipeline/worker_pool.c b/src/pipeline/worker_pool.c index 2943fd746a..96611bf8dd 100644 --- a/src/pipeline/worker_pool.c +++ b/src/pipeline/worker_pool.c @@ -22,9 +22,28 @@ enum { WP_TRUE = 1, WP_MIN = 1, WP_STEP = 1 }; /* CBM_WORKER_STACK_SIZE (8 MB per worker) lives in worker_pool.h. */ +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +static cbm_parallel_spawn_observer_fn g_spawn_observer; +static void *g_spawn_observer_ctx; + +void cbm_parallel_test_set_spawn_observer(cbm_parallel_spawn_observer_fn observer, void *ctx) { + g_spawn_observer = observer; + g_spawn_observer_ctx = ctx; +} + +static void observe_spawn(int count, int helpers) { + if (g_spawn_observer) { + g_spawn_observer(count, helpers, g_spawn_observer_ctx); + } +} +#else +#define observe_spawn(count, helpers) ((void)0) +#endif + /* ── Serial fallback ─────────────────────────────────────────────── */ static void run_serial(int count, cbm_parallel_fn fn, void *ctx) { + observe_spawn(count, 0); for (int i = 0; i < count; i++) { fn(i, ctx); } @@ -103,6 +122,7 @@ static void run_pthreads_counted(int count, cbm_parallel_fn fn, void *ctx, int n } started++; } + observe_spawn(count, started); int main_done = 0; uint64_t main_ops_before = cbm_memev_thread_ops(); while (WP_TRUE) { @@ -164,6 +184,7 @@ static void run_pthreads(int count, cbm_parallel_fn fn, void *ctx, int nworkers) } } + observe_spawn(count, nworkers); /* Main thread also participates */ while (WP_TRUE) { int idx = atomic_fetch_add_explicit(&next_idx, WP_STEP, memory_order_relaxed); diff --git a/src/pipeline/worker_pool.h b/src/pipeline/worker_pool.h index 91697b961d..49b3efd1f1 100644 --- a/src/pipeline/worker_pool.h +++ b/src/pipeline/worker_pool.h @@ -36,4 +36,11 @@ typedef struct { * If count <= 1 or workers <= 1, runs single-threaded. */ void cbm_parallel_for(int count, cbm_parallel_fn fn, void *ctx, cbm_parallel_for_opts_t opts); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +/* Observe the completed spawn attempt before the caller takes work. Install + * only while no dispatch is running; clear after the observed dispatch joins. */ +typedef void (*cbm_parallel_spawn_observer_fn)(int count, int helpers, void *ctx); +void cbm_parallel_test_set_spawn_observer(cbm_parallel_spawn_observer_fn observer, void *ctx); +#endif + #endif /* CBM_WORKER_POOL_H */ diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 1c75446313..9a364e9414 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -13,6 +13,7 @@ #include "lsp/c_lsp.h" #include "lsp/py_lsp.h" #include "pipeline/lsp_surface.h" +#include "pipeline/worker_pool.h" #include "pipeline/pass_lsp_cross.h" #include "pipeline/pipeline_internal.h" #include "pipeline/helm_template.h" @@ -164,6 +165,129 @@ TEST(pipeline_background_worker_policy_preserves_overrides) { PASS(); } +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +enum { SURFACE_BUDGET_FILES = 5 }; + +typedef struct { + atomic_int expected; + atomic_int arrivals; + atomic_int live; + atomic_int peak; + atomic_bool released; + atomic_int visits[SURFACE_BUDGET_FILES]; + int helpers; +} surface_budget_probe_t; + +static void surface_budget_spawned(int count, int helpers, void *arg) { + surface_budget_probe_t *probe = arg; + probe->helpers = helpers; + int participants = helpers + 1; + atomic_store(&probe->expected, count < participants ? count : participants); +} + +static void surface_budget_row(int index, bool entering, void *arg) { + surface_budget_probe_t *probe = arg; + if (!entering) { + atomic_fetch_sub(&probe->live, 1); + return; + } + atomic_fetch_add(&probe->visits[index], 1); + int live = atomic_fetch_add(&probe->live, 1) + 1; + int peak = atomic_load(&probe->peak); + while (peak < live && !atomic_compare_exchange_weak(&probe->peak, &peak, live)) {} + atomic_fetch_add(&probe->arrivals, 1); + /* No timing window: hold the first wave until every possible participant + * has entered. The post-spawn observation accounts for partial thread + * creation and serial fallback, so neither can strand this gate. The test + * runner's watchdog bounds a broken dispatch without making time a verdict. */ + while (!atomic_load(&probe->released)) { + int expected = atomic_load(&probe->expected); + if (expected > 0 && atomic_load(&probe->arrivals) >= expected) { + atomic_store(&probe->released, true); + } + } +} + +TEST(pipeline_surface_respects_total_worker_budget) { + CBMFileResult results[SURFACE_BUDGET_FILES] = {0}; + CBMFileResult *cache[SURFACE_BUDGET_FILES]; + cbm_file_info_t files[SURFACE_BUDGET_FILES] = { + {.rel_path = "a.go", .language = CBM_LANG_GO}, + {.rel_path = "b.go", .language = CBM_LANG_GO}, + {.rel_path = "c.go", .language = CBM_LANG_GO}, + {.rel_path = "d.go", .language = CBM_LANG_GO}, + {.rel_path = "e.go", .language = CBM_LANG_GO}, + }; + surface_budget_probe_t probe = {0}; + for (int i = 0; i < SURFACE_BUDGET_FILES; i++) { + cache[i] = &results[i]; + atomic_init(&probe.visits[i], 0); + } + cache[1] = NULL; /* The skipped row must not change stable file ordering. */ + atomic_init(&probe.expected, 0); + atomic_init(&probe.arrivals, 0); + atomic_init(&probe.live, 0); + atomic_init(&probe.peak, 0); + atomic_init(&probe.released, false); + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + cbm_setenv("CBM_WORKERS", "2", 1); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + cbm_pipeline_t *pipeline = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + cbm_pipeline_set_background(pipeline, true); + cbm_pipeline_ctx_t ctx = {.pipeline = pipeline}; + cbm_lsp_surface_row_t *rows = NULL; + int count = 0; + cbm_parallel_test_set_spawn_observer(surface_budget_spawned, &probe); + cbm_lsp_surface_test_set_row_observer(surface_budget_row, &probe); + int rc = cbm_lsp_surface_build_rows(&ctx, "budget", cache, files, SURFACE_BUDGET_FILES, NULL, + NULL, &rows, &count); + int parallel_peak = atomic_load(&probe.peak); + int parallel_helpers = probe.helpers; + atomic_store(&probe.expected, 0); + atomic_store(&probe.arrivals, 0); + atomic_store(&probe.peak, 0); + atomic_store(&probe.released, false); + cbm_setenv("CBM_INDEX_SINGLE_THREAD", "1", 1); + cbm_lsp_surface_row_t *serial_rows = NULL; + int serial_count = 0; + int serial_rc = cbm_lsp_surface_build_rows(&ctx, "budget", cache, files, SURFACE_BUDGET_FILES, + NULL, NULL, &serial_rows, &serial_count); + cbm_lsp_surface_test_set_row_observer(NULL, NULL); + cbm_parallel_test_set_spawn_observer(NULL, NULL); + bool same = rc == 0 && serial_rc == 0 && rows && serial_rows && + count == SURFACE_BUDGET_FILES - 1 && count == serial_count; + for (int i = 0; same && i < count; i++) { + same = strcmp(rows[i].rel_path, files[i < 1 ? i : i + 1].rel_path) == 0 && + strcmp(rows[i].defs_json, serial_rows[i].defs_json) == 0 && + strcmp(rows[i].surface_sha, serial_rows[i].surface_sha) == 0; + } + cbm_store_free_lsp_surfaces(rows, count); + cbm_store_free_lsp_surfaces(serial_rows, serial_count); + cbm_pipeline_free(pipeline); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + + ASSERT_EQ(rc, 0); + ASSERT_EQ(serial_rc, 0); + ASSERT_TRUE(same); + ASSERT_GTE(parallel_helpers, 1); /* A serial setup is not a concurrency repro. */ + ASSERT_EQ(probe.helpers, 0); + ASSERT_EQ(atomic_load(&probe.live), 0); + for (int i = 0; i < SURFACE_BUDGET_FILES; i++) { + ASSERT_EQ(atomic_load(&probe.visits[i]), 2); + } + ASSERT_EQ(atomic_load(&probe.peak), 1); + ASSERT_EQ(parallel_peak, 2); + PASS(); +} +#endif + TEST(pipeline_background_policy_preserves_index_results) { ASSERT_EQ(setup_test_repo(), 0); /* 64 fillers plus setup_test_repo's three files exceed MIN_FILES_FOR_PARALLEL @@ -27721,6 +27845,9 @@ SUITE(pipeline) { /* Lifecycle */ RUN_TEST(pipeline_create_free); RUN_TEST(pipeline_background_worker_policy_preserves_overrides); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + RUN_TEST(pipeline_surface_respects_total_worker_budget); +#endif RUN_TEST(pipeline_background_policy_preserves_index_results); RUN_TEST(pipeline_null_repo); RUN_TEST(pipeline_free_null); From db74648c22b9f057a16ae25814d5e762c415ab32 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Sun, 25 Oct 2026 21:03:17 +0100 Subject: [PATCH 6/7] fix(pipeline): keep background refreshes on the parallel route A background refresh used performance cores minus one, which is a single worker on a host with two or fewer performance cores. One worker sends the indexer down the sequential route, which still builds a different graph than the parallel route (fewer Route and HANDLES edges, churned CALLS). Floor the background width at two whenever the foreground width is two or more; a host whose foreground width is one keeps one. CBM_WORKERS and CBM_INDEX_SINGLE_THREAD keep their precedence, and larger hosts keep their headroom unchanged. The pipeline and mcp worker-policy tests now expect the floored width, and a new pipeline test pins the width for one-core, two-core and larger host shapes. Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 21 ++++++++++++++++++++- src/pipeline/pipeline.h | 5 +++++ tests/test_mcp.c | 4 +++- tests/test_pipeline.c | 37 ++++++++++++++++++++++++++++++++----- 4 files changed, 60 insertions(+), 7 deletions(-) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 5474e8f5e5..99bf23f6e3 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -1437,6 +1437,22 @@ void cbm_pipeline_set_committed_counts(cbm_pipeline_t *p, int nodes, int edges) } } +/* Two workers is the smallest width that takes the parallel route. */ +enum { PL_PARALLEL_MIN_WORKERS = 2 }; + +/* Background refreshes leave CPU headroom, but never drop below two workers + * while the foreground width is two or more. WHY: one worker sends the indexer + * down the sequential route, which still builds a different graph than the + * parallel route (fewer Route and HANDLES edges, churned CALLS). Until the two + * routes agree, an automatic refresh must not change the graph only because + * the host has two or fewer performance cores. */ +int cbm_pipeline_background_worker_width(int foreground, int headroom) { + if (foreground >= PL_PARALLEL_MIN_WORKERS && headroom < PL_PARALLEL_MIN_WORKERS) { + return PL_PARALLEL_MIN_WORKERS; + } + return headroom; +} + /* Effective worker count. The crash supervisor re-runs its worker single- * threaded (CBM_INDEX_SINGLE_THREAD=1) so a per-file marker can pin the EXACT * crasher; a parallel re-run would race the marker. Honour that override @@ -1447,8 +1463,11 @@ int cbm_pipeline_worker_count(const cbm_pipeline_t *p) { int workers; if (st && st[0] == '1') { workers = 1; + } else if (p && p->background) { + workers = cbm_pipeline_background_worker_width(cbm_default_worker_count(true), + cbm_default_worker_count(false)); } else { - workers = cbm_default_worker_count(!p || !p->background); + workers = cbm_default_worker_count(true); } #if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API atomic_store(&g_last_worker_count, workers); diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index a9ac979357..e7222e4ddb 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -79,6 +79,11 @@ void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background); * crash-recovery overrides. */ int cbm_pipeline_worker_count(const cbm_pipeline_t *p); +/* Background width from the foreground width and the headroom width: the + * headroom, floored at two whenever the foreground width is two or more so an + * automatic refresh keeps the parallel route. */ +int cbm_pipeline_background_worker_width(int foreground, int headroom); + #if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API void cbm_pipeline_worker_count_test_reset(void); int cbm_pipeline_worker_count_test_last(void); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index 2d6ba3ea38..e20de20348 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -23303,7 +23303,9 @@ static int mcp_assert_inprocess_worker_policy(bool committed, bool automatic) { } /* Capture while CBM_WORKERS is still unset. Restoring first makes the * expected count follow the lane override instead of this policy. */ - int expected = cbm_default_worker_count(!committed); + int expected = committed ? cbm_pipeline_background_worker_width(cbm_default_worker_count(true), + cbm_default_worker_count(false)) + : cbm_default_worker_count(true); mcp_test_restore_env(environment, sizeof(environment) / sizeof(environment[0])); bool cleaned = !cache_ready || th_rmtree(cache) == 0; diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 96825bccad..e9c636d83b 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -140,12 +140,19 @@ TEST(pipeline_background_worker_policy_preserves_overrides) { int foreground_default = cbm_pipeline_worker_count(foreground); int background_default = cbm_pipeline_worker_count(background); bool defaults_match = foreground_default == cbm_default_worker_count(true) && - background_default == cbm_default_worker_count(false) && - foreground_default >= 1 && background_default >= 1; - + background_default == cbm_pipeline_background_worker_width( + cbm_default_worker_count(true), + cbm_default_worker_count(false)) && + foreground_default >= 1 && background_default >= 1 && + (foreground_default < 2 || background_default >= 2); + + /* A pinned single worker leaves no parallel route to preserve. */ + cbm_setenv("CBM_WORKERS", "1", 1); + bool env_override = + cbm_pipeline_worker_count(foreground) == 1 && cbm_pipeline_worker_count(background) == 1; cbm_setenv("CBM_WORKERS", "3", 1); - bool env_override = cbm_pipeline_worker_count(foreground) == 3 && - cbm_pipeline_worker_count(background) == 3; + env_override = env_override && cbm_pipeline_worker_count(foreground) == 3 && + cbm_pipeline_worker_count(background) == 3; cbm_setenv("CBM_INDEX_SINGLE_THREAD", "1", 1); bool recovery_override = cbm_pipeline_worker_count(foreground) == 1 && @@ -165,6 +172,25 @@ TEST(pipeline_background_worker_policy_preserves_overrides) { PASS(); } +/* An automatic refresh keeps CPU headroom but never falls to one worker, the + * sequential route, while the foreground width could run in parallel. Each + * host shape is given as the two widths cbm_default_worker_count() derives + * from it (foreground: every core; headroom: performance cores minus one, at + * least one), so the verdict does not depend on the machine running it. */ +TEST(pipeline_background_worker_width_keeps_parallel_route) { + /* One core: foreground 1, headroom 1. Nothing to preserve. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(1, 1), 1); + /* Two performance cores: headroom alone would be 1. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(2, 1), 2); + /* Two performance cores beside efficiency cores: same floor. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(8, 1), 2); + /* Larger hosts keep their headroom unchanged. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(3, 2), 2); + ASSERT_EQ(cbm_pipeline_background_worker_width(4, 3), 3); + ASSERT_EQ(cbm_pipeline_background_worker_width(18, 5), 5); + PASS(); +} + #if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS enum { SURFACE_BUDGET_FILES = 5 }; @@ -28325,6 +28351,7 @@ SUITE(pipeline) { /* Lifecycle */ RUN_TEST(pipeline_create_free); RUN_TEST(pipeline_background_worker_policy_preserves_overrides); + RUN_TEST(pipeline_background_worker_width_keeps_parallel_route); #if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS RUN_TEST(pipeline_surface_respects_total_worker_budget); #endif From 4316e93c2be30d3cbabe02ccf3263ce15bbc0da0 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 27 Oct 2026 23:44:34 +0100 Subject: [PATCH 7/7] test(pipeline): pass the pipeline argument in the skip-locked manifest test Main added the manifest-skip-locked test (9f57edfa) with the old cbm_pipeline_build_semantic_manifest signature. This branch gives the function a leading pipeline parameter, so the tree merged from main failed to compile. Pass NULL like the branch manifest-fail-closed test already does. Signed-off-by: Martin Vogel --- tests/test_pipeline.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 24ba07cd43..9cdfb17e34 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -8750,7 +8750,7 @@ TEST(pipeline_semantic_manifest_skips_unreadable_subdirectory) { cbm_file_hash_t *manifest = NULL; int manifest_count = -1; - int rc = cbm_pipeline_build_semantic_manifest("manifest-skip-locked", tmp, NULL, 0, NULL, 0, + int rc = cbm_pipeline_build_semantic_manifest(NULL, "manifest-skip-locked", tmp, NULL, 0, NULL, 0, NULL, NULL, &manifest, &manifest_count); bool found_sibling_gitignore = false;