From 11a37079ff99e47423dcabc18b9cc9d0b09469fa Mon Sep 17 00:00:00 2001 From: Andrew Hundt Date: Sat, 4 Jul 2026 23:20:34 -0400 Subject: [PATCH] fix(pipeline): align sequential edge parity Match sequential CALLS property capacity to the parallel pass for edge JSON emitted through calls_emit_edge so changed-file overlay/exact paths preserve later call arguments. Require THROWS/RAISES source nodes to be callable scopes through a shared pipeline helper, preventing sequential file/module fallback edges from diverging from parallel full-index behavior. Add regression coverage for the eighth CALLS arg and top-level raise sequential/parallel parity. Validated with focused canaries, full pipeline and parallel suites, source-safety, product build, guarded FastAPI matrix, and an isolated guard-removal experiment that still failed active equality. Signed-off-by: Andrew Hundt --- src/pipeline/pass_calls.c | 19 ++++++--- src/pipeline/pass_parallel.c | 2 +- src/pipeline/pass_usages.c | 6 ++- src/pipeline/pipeline_internal.h | 5 +++ tests/test_parallel.c | 54 ++++++++++++++++++++++++++ tests/test_pipeline.c | 66 ++++++++++++++++++++++++++++++++ 6 files changed, 144 insertions(+), 8 deletions(-) diff --git a/src/pipeline/pass_calls.c b/src/pipeline/pass_calls.c index 537bc7ed6..b59ae3c98 100644 --- a/src/pipeline/pass_calls.c +++ b/src/pipeline/pass_calls.c @@ -11,7 +11,16 @@ */ #include "foundation/constants.h" -enum { PC_RING = 4, PC_RING_MASK = 3, PC_SIG_SCAN = 15, PC_REGEX_GRP = 2 }; +enum { + PC_RING = 4, + PC_RING_MASK = 3, + PC_SIG_SCAN = 15, + PC_REGEX_GRP = 2, + /* Keep sequential CALLS/HTTP/CONFIG edge property capacity aligned with + * pass_parallel.c so incremental overlays serialize the same call args as + * a fresh full index. */ + PC_CALL_PROPS_CAP = CBM_SZ_2K, +}; #include "pipeline/pipeline.h" #include #include "pipeline/pipeline_internal.h" @@ -160,7 +169,7 @@ static void emit_http_async_edge(cbm_pipeline_ctx_t *ctx, const CBMCall *call, if (!is_url && !is_topic) { char esc_callee[CBM_SZ_256]; cbm_json_escape(esc_callee, sizeof(esc_callee), call->callee_name); - char props[CBM_SZ_512]; + char props[PC_CALL_PROPS_CAP]; snprintf(props, sizeof(props), "{\"callee\":\"%s\",\"confidence\":%.2f,\"strategy\":\"%s\",\"candidates\":%d}", esc_callee, res->confidence, res->strategy ? res->strategy : "unknown", @@ -182,7 +191,7 @@ static void emit_http_async_edge(cbm_pipeline_ctx_t *ctx, const CBMCall *call, char esc_url[CBM_SZ_256]; cbm_json_escape(esc_callee, sizeof(esc_callee), call->callee_name); cbm_json_escape(esc_url, sizeof(esc_url), url_or_topic); - char props[CBM_SZ_512]; + char props[PC_CALL_PROPS_CAP]; snprintf(props, sizeof(props), "{\"callee\":\"%s\",\"url_path\":\"%s\"%s%s%s%s%s}", esc_callee, esc_url, method ? ",\"method\":\"" : "", method ? method : "", method ? "\"" : "", broker ? ",\"broker\":\"" : "", broker ? broker : ""); @@ -219,7 +228,7 @@ static bool emit_classified_edge(cbm_pipeline_ctx_t *ctx, const CBMCall *call, char esc_k[CBM_SZ_256]; cbm_json_escape(esc_c, sizeof(esc_c), call->callee_name); cbm_json_escape(esc_k, sizeof(esc_k), call->first_string_arg ? call->first_string_arg : ""); - char props[CBM_SZ_512]; + char props[PC_CALL_PROPS_CAP]; snprintf(props, sizeof(props), "{\"callee\":\"%s\",\"key\":\"%s\",\"confidence\":%.2f}", esc_c, esc_k, res->confidence); calls_emit_edge(ctx->gbuf, source->id, target->id, "CONFIGURES", props, sizeof(props), @@ -228,7 +237,7 @@ static bool emit_classified_edge(cbm_pipeline_ctx_t *ctx, const CBMCall *call, } char esc_c2[CBM_SZ_256]; cbm_json_escape(esc_c2, sizeof(esc_c2), call->callee_name); - char props[CBM_SZ_512]; + char props[PC_CALL_PROPS_CAP]; snprintf(props, sizeof(props), "{\"callee\":\"%s\",\"confidence\":%.2f,\"strategy\":\"%s\",\"candidates\":%d}", esc_c2, res->confidence, res->strategy ? res->strategy : "unknown", diff --git a/src/pipeline/pass_parallel.c b/src/pipeline/pass_parallel.c index 7e39ea70b..0c444bb0c 100644 --- a/src/pipeline/pass_parallel.c +++ b/src/pipeline/pass_parallel.c @@ -1562,7 +1562,7 @@ static void resolve_file_throws(resolve_ctx_t *rc, resolve_worker_state_t *ws, continue; } const cbm_gbuf_node_t *src = cbm_gbuf_find_by_qn(rc->main_gbuf, thr->enclosing_func_qn); - if (!src) { + if (!cbm_pipeline_node_is_callable_scope(src)) { continue; } const char *edge_type = is_checked_exception(thr->exception_name) ? "THROWS" : "RAISES"; diff --git a/src/pipeline/pass_usages.c b/src/pipeline/pass_usages.c index 805999fcd..64bc7eb78 100644 --- a/src/pipeline/pass_usages.c +++ b/src/pipeline/pass_usages.c @@ -136,6 +136,7 @@ static int resolve_usage_edges(cbm_pipeline_ctx_t *ctx, const CBMFileResult *res static int resolve_throw_edges(cbm_pipeline_ctx_t *ctx, const CBMFileResult *result, const char *rel, const char *module_qn, const char **imp_keys, const char **imp_vals, int imp_count) { + (void)rel; /* Throws intentionally match the parallel pass: no file-node fallback. */ int resolved = 0; for (int t = 0; t < result->throws.count; t++) { CBMThrow *thr = &result->throws.items[t]; @@ -143,8 +144,9 @@ static int resolve_throw_edges(cbm_pipeline_ctx_t *ctx, const CBMFileResult *res continue; } - const cbm_gbuf_node_t *src = find_enclosing_node(ctx, thr->enclosing_func_qn, rel); - if (!src) { + const cbm_gbuf_node_t *src = + thr->enclosing_func_qn ? cbm_gbuf_find_by_qn(ctx->gbuf, thr->enclosing_func_qn) : NULL; + if (!cbm_pipeline_node_is_callable_scope(src)) { continue; } diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index e5343eb0c..01ee42f96 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -139,6 +139,11 @@ static inline bool cbm_pipeline_label_is_import_target(const char *label) { strcmp(label, "File") == 0); } +static inline bool cbm_pipeline_node_is_callable_scope(const cbm_gbuf_node_t *node) { + return node && node->label && + (strcmp(node->label, "Function") == 0 || strcmp(node->label, "Method") == 0); +} + /* Time unit conversions */ #define CBM_NS_PER_SEC 1000000000LL #define CBM_US_PER_SEC 1000000LL diff --git a/tests/test_parallel.c b/tests/test_parallel.c index 31b40e3a9..b4b57b594 100644 --- a/tests/test_parallel.c +++ b/tests/test_parallel.c @@ -500,6 +500,59 @@ static int count_route_registration_for_path(cbm_gbuf_t *gbuf, const char *url_p return c.route_registration_calls; } +static void count_exception_edges(const cbm_gbuf_edge_t *edge, void *ud) { + int *count = (int *)ud; + if (!edge || !edge->type || !count) { + return; + } + if (strcmp(edge->type, "THROWS") == 0 || strcmp(edge->type, "RAISES") == 0) { + (*count)++; + } +} + +static int exception_edge_count(cbm_gbuf_t *gbuf) { + int count = 0; + cbm_gbuf_foreach_edge(gbuf, count_exception_edges, &count); + return count; +} + +TEST(parallel_top_level_raise_matches_sequential_no_file_fallback) { + char dir[CBM_PATH_MAX]; + int n = snprintf(dir, sizeof(dir), "%s/cbm_top_raise_XXXXXX", cbm_tmpdir()); + ASSERT_GT(n, 0); + ASSERT_LT((size_t)n, sizeof(dir)); + ASSERT_TRUE(cbm_mkdtemp(dir) != NULL); + + const char *source = + "class HTTPException(Exception):\n" + " pass\n\n" + "raise HTTPException()\n"; + + char path[CBM_PATH_MAX]; + n = snprintf(path, sizeof(path), "%s/app.py", dir); + ASSERT_GT(n, 0); + ASSERT_LT((size_t)n, sizeof(path)); + ASSERT_EQ(th_write_file(path, source), 0); + + cbm_file_info_t files[1] = {0}; + files[0].path = path; + files[0].rel_path = (char *)"app.py"; + files[0].language = CBM_LANG_PYTHON; + + cbm_gbuf_t *seq = run_sequential("cbm_top_raise", dir, files, 1); + cbm_gbuf_t *par = run_parallel("cbm_top_raise", dir, files, 1, 1); + ASSERT_NOT_NULL(seq); + ASSERT_NOT_NULL(par); + + ASSERT_EQ(exception_edge_count(seq), 0); + ASSERT_EQ(exception_edge_count(par), 0); + + cbm_gbuf_free(seq); + cbm_gbuf_free(par); + th_rmtree(dir); + PASS(); +} + TEST(parallel_fastapi_websocket_route_registration_matches_sequential) { char dir[256]; snprintf(dir, sizeof(dir), "/tmp/cbm_ws_routes_XXXXXX"); @@ -1256,6 +1309,7 @@ SUITE(parallel) { RUN_TEST(parallel_empty_files); RUN_TEST(parallel_args_json_no_overflow); RUN_TEST(parallel_unresolved_route_suffix_does_not_emit_self_call); + RUN_TEST(parallel_top_level_raise_matches_sequential_no_file_fallback); RUN_TEST(parallel_fastapi_websocket_route_registration_matches_sequential); /* Cleanup shared state */ diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 73639eb12..7fd3f7cb9 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -541,6 +541,71 @@ TEST(pipeline_call_edge_props_include_args_and_line) { PASS(); } +TEST(pipeline_sequential_call_edges_preserve_eighth_arg) { + if (setup_test_repo() != 0) { + FAIL("failed to create temp dir"); + } + + const char *py_path = TH_PATH(g_tmpdir, "wide_args.py"); + ASSERT_EQ(th_write_file(py_path, + "def wide_target(**kwargs):\n" + " return kwargs\n" + "\n" + "def wide_caller():\n" + " first_expression_value = 1\n" + " second_expression_value = 2\n" + " third_expression_value = 3\n" + " fourth_expression_value = 4\n" + " fifth_expression_value = 5\n" + " sixth_expression_value = 6\n" + " seventh_expression_value = 7\n" + " eighth_expression_value = 8\n" + " return wide_target(\n" + " first_keyword_argument=first_expression_value,\n" + " second_keyword_argument=second_expression_value,\n" + " third_keyword_argument=third_expression_value,\n" + " fourth_keyword_argument=fourth_expression_value,\n" + " fifth_keyword_argument=fifth_expression_value,\n" + " sixth_keyword_argument=sixth_expression_value,\n" + " seventh_keyword_argument=seventh_expression_value,\n" + " eighth_keyword_argument=eighth_expression_value,\n" + " )\n"), + 0); + + char db_path[CBM_SZ_512]; + int n = snprintf(db_path, sizeof(db_path), "%s/wide_args.db", g_tmpdir); + ASSERT_GT(n, 0); + ASSERT_LT((size_t)n, sizeof(db_path)); + + cbm_pipeline_t *p = cbm_pipeline_new(g_tmpdir, db_path, CBM_MODE_FAST); + ASSERT_NOT_NULL(p); + ASSERT_EQ(cbm_pipeline_run(p), 0); + const char *project = cbm_pipeline_project_name(p); + ASSERT_NOT_NULL(project); + + sqlite3 *db = NULL; + ASSERT_EQ(sqlite3_open(db_path, &db), SQLITE_OK); + sqlite3_stmt *stmt = NULL; + const char sql[] = + "SELECT e.properties FROM edges e " + "JOIN nodes t ON t.id = e.target_id " + "WHERE e.project = ?1 AND e.type = 'CALLS' AND t.name = 'wide_target' " + "LIMIT 1"; + ASSERT_EQ(sqlite3_prepare_v2(db, sql, -1, &stmt, NULL), SQLITE_OK); + sqlite3_bind_text(stmt, 1, project, -1, SQLITE_TRANSIENT); + ASSERT_EQ(sqlite3_step(stmt), SQLITE_ROW); + const char *edge_props = (const char *)sqlite3_column_text(stmt, 0); + ASSERT_NOT_NULL(edge_props); + ASSERT(strstr(edge_props, "\"k\":\"eighth_keyword_argument\"") != NULL); + ASSERT(strstr(edge_props, "\"e\":\"eighth_expression_value\"") != NULL); + + sqlite3_finalize(stmt); + sqlite3_close(db); + cbm_pipeline_free(p); + teardown_test_repo(); + PASS(); +} + TEST(pipeline_fast_mode) { if (setup_test_repo() != 0) { FAIL("failed to create temp dir"); @@ -14436,6 +14501,7 @@ SUITE(pipeline) { RUN_TEST(pipeline_project_name_derived); RUN_TEST(pipeline_mode_global_semantic_edges_policy); RUN_TEST(pipeline_call_edge_props_include_args_and_line); + RUN_TEST(pipeline_sequential_call_edges_preserve_eighth_arg); RUN_TEST(pipeline_fast_mode); /* Definitions pass */ RUN_TEST(pipeline_definitions_function_nodes);