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 <ATHundt@gmail.com>

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 <stdint.h>
 #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",

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";

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;
         }
 

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

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 */

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);