From ee126155ab97bd21a0e65faa57d71d30b3dd52b5 Mon Sep 17 00:00:00 2001 From: Thomas Dyar Date: Mon, 24 Aug 2026 15:25:24 -0400 Subject: [PATCH] feat(pipeline): add Ensemble production topology routing pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parses XData ProductionDefinition blocks to extract production topology and emits Route/ASYNC_CALLS/HANDLES graph edges for Ensemble interoperability routing. Key design points: - Emit Route nodes (qn: __route__ensemble__.) with props {broker, class, enabled, production} — visible in get_architecture and search_graph at the same priority as HTTP routes - Emit ASYNC_CALLS edges from source method/item to target Route node via SendRequestSync literal scan and TargetConfigName(s) settings - Emit HANDLES edges from each item's entry-point method to its Route node, completing the code-to-topology link - Fix scan_source_for_send_targets: scope search to calling method body using brace-depth tracking to eliminate cross-method contamination - Handle TargetConfigNames (plural, comma-separated) by emitting one ASYNC_CALLS edge per token - Segment-anchored class_name_matches() prevents prefix collisions ("Ens" not matching "Ens.BusinessService") - Language-gate: skip entirely when no ObjectScript nodes present Tests: pipeline_ensemble_routing_edges (Route nodes + ASYNC_CALLS + HANDLES); pipeline_ensemble_routing_method_scoping (two methods, two targets, no cross-contamination). Fixes #467 Signed-off-by: Thomas Dyar --- Makefile.cbm | 1 + src/pipeline/pass_ensemble_routing.c | 626 +++++++++++++++++++++++++++ src/pipeline/pass_ensemble_routing.h | 8 + src/pipeline/pipeline.c | 14 +- tests/test_pipeline.c | 212 +++++++++ 5 files changed, 857 insertions(+), 4 deletions(-) create mode 100644 src/pipeline/pass_ensemble_routing.c create mode 100644 src/pipeline/pass_ensemble_routing.h diff --git a/Makefile.cbm b/Makefile.cbm index 28eb21e27..e0673de04 100644 --- a/Makefile.cbm +++ b/Makefile.cbm @@ -356,6 +356,7 @@ PIPELINE_SRCS = \ src/pipeline/pass_complexity.c \ src/pipeline/pass_cross_repo.c \ src/pipeline/artifact.c \ + src/pipeline/pass_ensemble_routing.c \ src/pipeline/pass_pkgmap.c # SimHash / MinHash module diff --git a/src/pipeline/pass_ensemble_routing.c b/src/pipeline/pass_ensemble_routing.c new file mode 100644 index 000000000..ab23316c5 --- /dev/null +++ b/src/pipeline/pass_ensemble_routing.c @@ -0,0 +1,626 @@ +#include "pipeline/pass_ensemble_routing.h" +#include "pipeline/pipeline_internal.h" +#include "graph_buffer/graph_buffer.h" +#include "foundation/log.h" +#include "foundation/compat.h" +#include "foundation/compat_fs.h" +#include "foundation/constants.h" +#include "foundation/str_util.h" + +#include +#include +#include +#include +#include + +#define CONF_LITERAL 0.95 +#define CONF_PROP 0.85 + +#define MAX_ITEMS 256 +#define MAX_SETTINGS 8 + +/* ── Language gate ───────────────────────────────────────────────── */ + +/* Visitor state for ObjectScript detection. */ +typedef struct { + bool found; +} ens_lang_check_t; + +static void check_objectscript_node(const cbm_gbuf_node_t *node, void *userdata) { + ens_lang_check_t *s = (ens_lang_check_t *)userdata; + if (s->found || !node->file_path) + return; + const char *fp = node->file_path; + size_t len = strlen(fp); + if (len >= 4 && (strcmp(fp + len - 4, ".cls") == 0 || strcmp(fp + len - 4, ".mac") == 0 || + strcmp(fp + len - 4, ".int") == 0)) { + s->found = true; + } +} + +/* Return true when the graph buffer contains at least one node from an + * ObjectScript source file (.cls / .mac / .int). */ +static bool has_objectscript_nodes(cbm_gbuf_t *gbuf) { + ens_lang_check_t state = {false}; + cbm_gbuf_foreach_node(gbuf, check_objectscript_node, &state); + return state.found; +} + +static const char *TOPOLOGY_SETTINGS[] = {"TargetConfigName", "TargetConfigNames", "PatientHost", + "ConformanceOperation", NULL}; + +static const char *ENTRY_POINTS[] = {"OnProcessInput", "OnMessage", "OnRequest", "OnTask", NULL}; + +typedef struct { + char setting_name[CBM_SZ_256]; + char value[CBM_SZ_256]; +} ens_setting_t; + +typedef struct { + char item_name[CBM_SZ_256]; + char class_name[CBM_SZ_256]; + bool enabled; + ens_setting_t settings[MAX_SETTINGS]; + int n_settings; +} ens_item_t; + +typedef struct { + char production_class[CBM_SZ_256]; + char file_path[CBM_SZ_512]; + ens_item_t items[MAX_ITEMS]; + int n_items; +} ens_prod_def_t; + +static void extract_xml_attr(const char *xml, int offset, const char *attr, char *out, int outsz) { + char needle[CBM_SZ_64]; + snprintf(needle, sizeof(needle), "%s=\"", attr); + const char *p = strstr(xml + offset, needle); + out[0] = '\0'; + if (!p) + return; + p += strlen(needle); + const char *e = strchr(p, '"'); + if (!e) + return; + int len = (int)(e - p); + if (len >= outsz) + len = outsz - 1; + memcpy(out, p, (size_t)len); + out[len] = '\0'; +} + +static bool is_topology_setting(const char *name) { + for (int i = 0; TOPOLOGY_SETTINGS[i]; i++) + if (strcmp(name, TOPOLOGY_SETTINGS[i]) == 0) + return true; + return false; +} + +static ens_prod_def_t *parse_production_xml(const char *xml, const char *class_qn, + const char *file_path) { + ens_prod_def_t *def = calloc(1, sizeof(ens_prod_def_t)); + if (!def) + return NULL; + snprintf(def->production_class, CBM_SZ_256, "%s", class_qn); + snprintf(def->file_path, sizeof(def->file_path), "%s", file_path ? file_path : ""); + + const char *p = xml; + while (*p && def->n_items < MAX_ITEMS) { + const char *item_start = strstr(p, "items[def->n_items]; + memset(item, 0, sizeof(*item)); + item->enabled = true; + + int off = (int)(item_start - xml); + extract_xml_attr(xml, off, "Name", item->item_name, CBM_SZ_256); + extract_xml_attr(xml, off, "ClassName", item->class_name, CBM_SZ_256); + char en[16]; + extract_xml_attr(xml, off, "Enabled", en, sizeof(en)); + if (en[0] && strcasecmp(en, "false") == 0) + item->enabled = false; + + if (!item->item_name[0] || !item->class_name[0]) { + p = item_start + 6; + continue; + } + + const char *item_end = strstr(item_start, ""); + if (!item_end) + item_end = item_start + strlen(item_start); + + const char *sp = item_start; + while (sp < item_end && item->n_settings < MAX_SETTINGS) { + const char *set = strstr(sp, "= item_end) + break; + int soff = (int)(set - xml); + char tgt[64], sname[CBM_SZ_256]; + extract_xml_attr(xml, soff, "Target", tgt, sizeof(tgt)); + extract_xml_attr(xml, soff, "Name", sname, CBM_SZ_256); + if (strcmp(tgt, "Host") == 0 && is_topology_setting(sname)) { + const char *vs = strchr(set + 9, '>'); + if (vs) { + vs++; + const char *ve = strstr(vs, ""); + if (ve && ve < item_end) { + int vlen = (int)(ve - vs); + if (vlen > 0 && vlen < CBM_SZ_256) { + ens_setting_t *s = &item->settings[item->n_settings++]; + snprintf(s->setting_name, CBM_SZ_256, "%s", sname); + memcpy(s->value, vs, (size_t)vlen); + s->value[vlen] = '\0'; + } + } + } + } + sp = set + 9; + } + /* Warn when topology settings were truncated at the cap. */ + if (item->n_settings == MAX_SETTINGS && strstr(sp, "item_name, "warn", + "settings truncated at MAX_SETTINGS cap"); + } + def->n_items++; + p = item_end + 7; + } + /* Warn when items were truncated at the cap. */ + if (def->n_items == MAX_ITEMS && strstr(p, "production_class, "warn", + "items truncated at MAX_ITEMS cap"); + } + return def; +} + +static char *read_file(const char *full_path) { + FILE *f = cbm_fopen(full_path, "rb"); + if (!f) + return NULL; + fseek(f, 0, SEEK_END); + long sz = ftell(f); + fseek(f, 0, SEEK_SET); + if (sz <= 0 || sz > 8 * 1024 * 1024) { + fclose(f); + return NULL; + } + char *buf = malloc((size_t)sz + 1); + if (!buf) { + fclose(f); + return NULL; + } + fread(buf, 1, (size_t)sz, f); + buf[sz] = '\0'; + fclose(f); + return buf; +} + +static const char *jstr(const char *json, const char *key, char *buf, int sz) { + if (!json || !key) + return NULL; + char needle[CBM_SZ_64]; + snprintf(needle, sizeof(needle), "\"%s\":\"", key); + const char *s = strstr(json, needle); + if (!s) + return NULL; + s += strlen(needle); + const char *e = strchr(s, '"'); + if (!e) + return NULL; + int len = (int)(e - s); + if (len >= sz) + len = sz - 1; + memcpy(buf, s, (size_t)len); + buf[len] = '\0'; + return buf; +} + +static const ens_item_t *find_item(const ens_prod_def_t *def, const char *name) { + for (int i = 0; i < def->n_items; i++) + if (strcmp(def->items[i].item_name, name) == 0) + return &def->items[i]; + return NULL; +} + +static int64_t find_entry_point(cbm_pipeline_ctx_t *ctx, const char *class_name) { + for (int ei = 0; ENTRY_POINTS[ei]; ei++) { + char suffix[CBM_SZ_512]; + snprintf(suffix, sizeof(suffix), "%s.%s", class_name, ENTRY_POINTS[ei]); + + const cbm_gbuf_node_t **nodes = NULL; + int count = 0; + cbm_gbuf_find_by_name(ctx->gbuf, ENTRY_POINTS[ei], (const cbm_gbuf_node_t ***)&nodes, + &count); + for (int ni = 0; ni < count; ni++) { + if (nodes[ni]->qualified_name && strcmp(nodes[ni]->qualified_name, suffix) == 0) + return nodes[ni]->id; + } + } + return 0; +} + +static void emit_async_call(cbm_pipeline_ctx_t *ctx, int64_t src_id, const ens_item_t *item, + const char *via, double confidence, const char *production_class) { + char item_qn[CBM_SZ_512]; + snprintf(item_qn, sizeof(item_qn), "%s.%s", production_class, item->item_name); + char route_qn[CBM_SZ_512]; + snprintf(route_qn, sizeof(route_qn), "__route__ensemble__%s", item_qn); + const cbm_gbuf_node_t *route = cbm_gbuf_find_by_qn(ctx->gbuf, route_qn); + if (!route) + return; + char conf_str[32]; + snprintf(conf_str, sizeof(conf_str), "%.2f", confidence); + char props[CBM_SZ_512]; + snprintf(props, sizeof(props), + "{\"via\":\"%s\",\"production\":\"%s\",\"item_name\":\"%s\"," + "\"confidence\":%s,\"enabled\":%s}", + via, production_class, item->item_name, conf_str, item->enabled ? "true" : "false"); + cbm_gbuf_insert_edge(ctx->gbuf, src_id, route->id, "ASYNC_CALLS", props); +} + +static void emit_handles(cbm_pipeline_ctx_t *ctx, const ens_item_t *item, + const char *production_class) { + int64_t method_id = find_entry_point(ctx, item->class_name); + if (!method_id) + return; + char item_qn[CBM_SZ_512]; + snprintf(item_qn, sizeof(item_qn), "%s.%s", production_class, item->item_name); + char route_qn[CBM_SZ_512]; + snprintf(route_qn, sizeof(route_qn), "__route__ensemble__%s", item_qn); + const cbm_gbuf_node_t *route = cbm_gbuf_find_by_qn(ctx->gbuf, route_qn); + if (!route) + return; + cbm_gbuf_insert_edge(ctx->gbuf, method_id, route->id, "HANDLES", NULL); +} + +/* Scan a .cls source file for SendRequestSync call targets and + * InitialExpression values for a given method/property name. */ +static void scan_source_for_send_targets(const char *source, const char *method_name, + char *literal_out, int lit_sz, char *prop_name_out, + int prop_sz) { + literal_out[0] = '\0'; + prop_name_out[0] = '\0'; + if (!source || !method_name) + return; + + /* Locate the method body */ + const char *body_start = NULL; + char needle[CBM_SZ_256]; + snprintf(needle, sizeof(needle), "Method %s(", method_name); + body_start = strstr(source, needle); + if (!body_start) { + snprintf(needle, sizeof(needle), "Method %s ", method_name); + body_start = strstr(source, needle); + } + if (!body_start) { + snprintf(needle, sizeof(needle), "ClassMethod %s(", method_name); + body_start = strstr(source, needle); + } + if (!body_start) { + snprintf(needle, sizeof(needle), "ClassMethod %s ", method_name); + body_start = strstr(source, needle); + } + if (!body_start) + return; + + const char *brace = strchr(body_start, '{'); + if (!brace) + return; + body_start = brace; + + int depth = 0; + const char *body_end = body_start; + while (*body_end) { + if (*body_end == '{') + depth++; + else if (*body_end == '}') { + depth--; + if (depth == 0) + break; + } + body_end++; + } + + const char *p = body_start; + while (p < body_end && (p = strstr(p, "SendRequestSync")) != NULL && p < body_end) { + p += 15; + while (*p == ' ' || *p == '\t') + p++; + if (*p != '(') + continue; + p++; + while (*p == ' ' || *p == '\t') + p++; + if (*p == '"') { + const char *ns = p + 1, *ne = strchr(ns, '"'); + if (ne && ne < body_end) { + int len = (int)(ne - ns); + if (len > 0 && len < lit_sz) { + memcpy(literal_out, ns, (size_t)len); + literal_out[len] = '\0'; + return; + } + } + } else if (p[0] == '.' && p[1] == '.') { + const char *ps = p + 2; + int plen = 0; + while (ps[plen] && (isalnum((unsigned char)ps[plen]) || ps[plen] == '_')) + plen++; + if (plen > 0 && plen < prop_sz) { + memcpy(prop_name_out, ps, (size_t)plen); + prop_name_out[plen] = '\0'; + return; + } + } + } +} + +/* Find InitialExpression value for a Property in the source. */ +static void scan_initial_expression(const char *source, const char *prop_name, char *out, + int outsz) { + out[0] = '\0'; + if (!source || !prop_name) + return; + char needle[CBM_SZ_256]; + snprintf(needle, sizeof(needle), "Property %s ", prop_name); + const char *p = strstr(source, needle); + if (!p) { + snprintf(needle, sizeof(needle), "Property %s[", prop_name); + p = strstr(source, needle); + } + if (!p) + return; + const char *ie = strstr(p, "InitialExpression ="); + if (!ie) + return; + ie = strchr(ie, '"'); + if (!ie) + return; + ie++; + const char *ie_end = strchr(ie, '"'); + if (!ie_end) + return; + int len = (int)(ie_end - ie); + if (len >= outsz) + len = outsz - 1; + memcpy(out, ie, (size_t)len); + out[len] = '\0'; +} + +static void collect_prod_defs(cbm_pipeline_ctx_t *ctx, ens_prod_def_t ***defs_out, int *count_out) { + const cbm_gbuf_node_t **xdata_nodes = NULL; + int xdata_count = 0; + cbm_gbuf_find_by_label(ctx->gbuf, "XData", (const cbm_gbuf_node_t ***)&xdata_nodes, + &xdata_count); + + ens_prod_def_t **defs = NULL; + int n = 0; + + for (int xi = 0; xi < xdata_count; xi++) { + const cbm_gbuf_node_t *xd = xdata_nodes[xi]; + if (!xd->name || strcmp(xd->name, "ProductionDefinition") != 0) + continue; + if (!xd->file_path || !ctx->repo_path) + continue; + + char full_path[CBM_SZ_1K]; + snprintf(full_path, sizeof(full_path), "%s/%s", ctx->repo_path, xd->file_path); + + char *source = read_file(full_path); + if (!source) + continue; + + char class_qn[CBM_SZ_256]; + class_qn[0] = '\0'; + if (xd->qualified_name) { + const char *dot = strrchr(xd->qualified_name, '.'); + if (dot) { + int len = (int)(dot - xd->qualified_name); + if (len > 0 && len < CBM_SZ_256) { + memcpy(class_qn, xd->qualified_name, (size_t)len); + class_qn[len] = '\0'; + } + } + } + if (!class_qn[0]) { + free(source); + continue; + } + + const char *xml_start = strstr(source, "file_path); + free(source); + if (!def) + continue; + + char n_items_buf[32]; + snprintf(n_items_buf, sizeof(n_items_buf), "%d", def->n_items); + cbm_log_info("ensemble_routing.parse", "class", class_qn, "items", n_items_buf); + + for (int i = 0; i < def->n_items; i++) { + ens_item_t *item = &def->items[i]; + char item_qn[CBM_SZ_512]; + snprintf(item_qn, sizeof(item_qn), "%s.%s", class_qn, item->item_name); + char route_qn[CBM_SZ_512]; + snprintf(route_qn, sizeof(route_qn), "__route__ensemble__%s", item_qn); + char iprops[CBM_SZ_512]; + snprintf( + iprops, sizeof(iprops), + "{\"broker\":\"ensemble\",\"class\":\"%s\",\"enabled\":%s,\"production\":\"%s\"}", + item->class_name, item->enabled ? "true" : "false", class_qn); + cbm_gbuf_upsert_node(ctx->gbuf, "Route", item->item_name, route_qn, xd->file_path, + xd->start_line, 0, iprops); + } + + ens_prod_def_t **tmp = realloc(defs, (size_t)(n + 1) * sizeof(ens_prod_def_t *)); + if (!tmp) { + free(def); + continue; + } + defs = tmp; + defs[n++] = def; + } + *defs_out = defs; + *count_out = n; +} + +/* True when haystack equals needle exactly, or ends with ".needle" + * (segment-anchored). Prevents "Ens" from matching "Ens.BusinessService". */ +static bool class_name_matches(const char *haystack, const char *needle) { + if (!haystack || !needle) + return false; + size_t hlen = strlen(haystack); + size_t nlen = strlen(needle); + if (nlen == 0) + return false; + if (hlen == nlen) + return strcmp(haystack, needle) == 0; + if (hlen > nlen && haystack[hlen - nlen - 1] == '.') + return strcmp(haystack + hlen - nlen, needle) == 0; + return false; +} + +static bool method_belongs_to_production(const cbm_gbuf_node_t *method, const ens_prod_def_t *def) { + if (!method->properties_json) + return false; + char parent_class[CBM_SZ_512]; + if (!jstr(method->properties_json, "parent_class", parent_class, sizeof(parent_class))) + return false; + for (int i = 0; i < def->n_items; i++) { + if (class_name_matches(parent_class, def->items[i].class_name)) + return true; + } + return false; +} + +static void resolve_method_routes(cbm_pipeline_ctx_t *ctx, const cbm_gbuf_node_t *method, + const char *source, const ens_prod_def_t *def) { + if (!method->properties_json) + return; + if (!method_belongs_to_production(method, def)) + return; + if (!strstr(source, "SendRequestSync")) + return; + + char literal[CBM_SZ_256], prop_name[CBM_SZ_256]; + scan_source_for_send_targets(source, method->name, literal, sizeof(literal), prop_name, + sizeof(prop_name)); + + if (literal[0]) { + const ens_item_t *item = find_item(def, literal); + if (item) + emit_async_call(ctx, method->id, item, "literal", CONF_LITERAL, def->production_class); + } else if (prop_name[0]) { + char init_expr[CBM_SZ_256]; + scan_initial_expression(source, prop_name, init_expr, sizeof(init_expr)); + if (init_expr[0]) { + const ens_item_t *item = find_item(def, init_expr); + if (item) + emit_async_call(ctx, method->id, item, prop_name, CONF_PROP, def->production_class); + } + } +} + +void cbm_pipeline_pass_ensemble_routing(cbm_pipeline_ctx_t *ctx) { + if (!ctx || !ctx->gbuf || !ctx->repo_path) + return; + + /* Early-exit: skip entirely when no ObjectScript source files are present + * in the graph buffer. Avoids a full XData scan on non-IRIS projects. */ + if (!has_objectscript_nodes(ctx->gbuf)) + return; + + ens_prod_def_t **defs = NULL; + int n_defs = 0; + collect_prod_defs(ctx, &defs, &n_defs); + if (n_defs == 0) + return; + + const cbm_gbuf_node_t **method_nodes = NULL; + int method_count = 0; + cbm_gbuf_find_by_label(ctx->gbuf, "Method", (const cbm_gbuf_node_t ***)&method_nodes, + &method_count); + + int before = cbm_gbuf_edge_count_by_type(ctx->gbuf, "ASYNC_CALLS"); + + /* Emit HANDLES edges: each item's entry-point method → its Route node */ + for (int di = 0; di < n_defs; di++) { + ens_prod_def_t *def = defs[di]; + for (int ii = 0; ii < def->n_items; ii++) + emit_handles(ctx, &def->items[ii], def->production_class); + } + + for (int di = 0; di < n_defs; di++) { + ens_prod_def_t *def = defs[di]; + + for (int mi = 0; mi < method_count; mi++) { + const cbm_gbuf_node_t *m = method_nodes[mi]; + if (!m->properties_json || !m->file_path) + continue; + if (!method_belongs_to_production(m, def)) + continue; + + char meth_full_path[CBM_SZ_1K]; + snprintf(meth_full_path, sizeof(meth_full_path), "%s/%s", ctx->repo_path, m->file_path); + char *meth_source = read_file(meth_full_path); + if (!meth_source) + continue; + resolve_method_routes(ctx, m, meth_source, def); + free(meth_source); + } + + for (int ii = 0; ii < def->n_items; ii++) { + const ens_item_t *item = &def->items[ii]; + for (int si = 0; si < item->n_settings; si++) { + const ens_setting_t *setting = &item->settings[si]; + if (!setting->value[0]) + continue; + char item_qn[CBM_SZ_512]; + snprintf(item_qn, sizeof(item_qn), "%s.%s", def->production_class, item->item_name); + const cbm_gbuf_node_t *item_node = cbm_gbuf_find_by_qn(ctx->gbuf, item_qn); + if (!item_node) + continue; + if (strcmp(setting->setting_name, "TargetConfigNames") == 0) { + char names_copy[CBM_SZ_256]; + snprintf(names_copy, sizeof(names_copy), "%s", setting->value); + char *tok = strtok(names_copy, ","); + while (tok) { + while (*tok == ' ') + tok++; + char *end = tok + strlen(tok) - 1; + while (end > tok && *end == ' ') + *end-- = '\0'; + const ens_item_t *target = find_item(def, tok); + if (target) + emit_async_call(ctx, item_node->id, target, setting->setting_name, + CONF_PROP, def->production_class); + tok = strtok(NULL, ","); + } + } else { + const ens_item_t *target = find_item(def, setting->value); + if (target) + emit_async_call(ctx, item_node->id, target, setting->setting_name, + CONF_PROP, def->production_class); + } + } + } + + free(defs[di]); + } + free(defs); + + int routes = cbm_gbuf_edge_count_by_type(ctx->gbuf, "ASYNC_CALLS") - before; + char n_defs_buf[32], n_routes_buf[32]; + snprintf(n_defs_buf, sizeof(n_defs_buf), "%d", n_defs); + snprintf(n_routes_buf, sizeof(n_routes_buf), "%d", routes); + cbm_log_info("ensemble_routing.done", "productions", n_defs_buf, "routes", n_routes_buf); +} diff --git a/src/pipeline/pass_ensemble_routing.h b/src/pipeline/pass_ensemble_routing.h new file mode 100644 index 000000000..3017bf6a8 --- /dev/null +++ b/src/pipeline/pass_ensemble_routing.h @@ -0,0 +1,8 @@ +#ifndef CBM_PASS_ENSEMBLE_ROUTING_H +#define CBM_PASS_ENSEMBLE_ROUTING_H + +#include "pipeline/pipeline_internal.h" + +void cbm_pipeline_pass_ensemble_routing(cbm_pipeline_ctx_t *ctx); + +#endif /* CBM_PASS_ENSEMBLE_ROUTING_H */ diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 65ac75183..7ceb610c3 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -19,6 +19,7 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6 }; #include "pipeline/pipeline_internal.h" #include "pipeline/lsp_surface.h" #include "pipeline/pass_lsp_cross.h" +#include "pipeline/pass_ensemble_routing.h" #include "pipeline/worker_pool.h" #include "graph_buffer/graph_buffer.h" #include "git/git_context.h" @@ -934,17 +935,22 @@ static void predump_cfg(cbm_pipeline_ctx_t *ctx) { static void predump_complexity(cbm_pipeline_ctx_t *ctx) { cbm_pipeline_pass_complexity(ctx); } +static void predump_ensemble(cbm_pipeline_ctx_t *ctx) { + cbm_pipeline_pass_ensemble_routing(ctx); +} + static void run_predump_passes(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx) { static const struct { predump_pass_fn fn; const char *name; bool moderate_only; /* true = skip in fast mode */ } passes[] = { - {predump_deco, "decorator_tags", false}, {predump_cfg, "configlink", false}, - {predump_route, "route_match", false}, {predump_sim, "similarity", true}, - {predump_sem, "semantic_edges", true}, {predump_complexity, "complexity", false}, + {predump_deco, "decorator_tags", false}, {predump_cfg, "configlink", false}, + {predump_route, "route_match", false}, {predump_ensemble, "ensemble_routing", false}, + {predump_sim, "similarity", true}, {predump_sem, "semantic_edges", true}, + {predump_complexity, "complexity", false}, }; - enum { PREDUMP_PASS_COUNT = 6 }; + enum { PREDUMP_PASS_COUNT = 7 }; struct timespec t; for (int i = 0; i < PREDUMP_PASS_COUNT && !check_cancel(p); i++) { /* "moderate_only" passes (similarity/semantic edges) run in FULL, diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index fb33bcc72..56284b024 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -12069,6 +12069,215 @@ TEST(pipeline_lsp_surface_persisted_and_body_edit_invariant) { PASS(); } +TEST(pipeline_ensemble_routing_edges) { + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/cbm_ens_XXXXXX"); + if (!cbm_mkdtemp(tmpdir)) + FAIL("failed to create temp dir"); + + char path[512]; + + /* Production class with two items */ + snprintf(path, sizeof(path), "%s/MyProduction.cls", tmpdir); + FILE *f = fopen(path, "w"); + if (!f) { + th_rmtree(tmpdir); + FAIL("fopen production"); + } + fprintf(f, "Class MyApp.MyProduction Extends Ens.Production\n" + "{\n" + "XData ProductionDefinition\n" + "{\n" + "\n" + " \n" + " \n" + " \n" + " \n" + "\n" + "}\n" + "}\n"); + fclose(f); + + /* Service class with OnProcessInput that routes to MyOperation via literal */ + snprintf(path, sizeof(path), "%s/MyService.cls", tmpdir); + f = fopen(path, "w"); + if (!f) { + th_rmtree(tmpdir); + FAIL("fopen service"); + } + fprintf(f, "Class MyApp.MyService Extends Ens.BusinessService\n" + "{\n" + "Method OnProcessInput(pRequest As %%Library.Object," + " Output pResponse As %%Library.Object) As %%Status\n" + "{\n" + " Do ..SendRequestSync(\"MyOperation\", pRequest, .pResponse)\n" + " Quit $$$OK\n" + "}\n" + "}\n"); + fclose(f); + + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/ens.db", tmpdir); + + cbm_pipeline_t *p = cbm_pipeline_new(tmpdir, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + ASSERT_EQ(cbm_pipeline_run(p), 0); + + cbm_store_t *s = cbm_store_open_path(db_path); + ASSERT_NOT_NULL(s); + const char *project = cbm_pipeline_project_name(p); + + /* Route nodes with __route__ensemble__ qn emitted for both production items */ + cbm_node_t *routes = NULL; + int route_count = 0; + cbm_store_find_nodes_by_label(s, project, "Route", &routes, &route_count); + int ens_route_count = 0; + for (int i = 0; i < route_count; i++) { + if (routes[i].qualified_name && + strstr(routes[i].qualified_name, "__route__ensemble__") != NULL) + ens_route_count++; + } + ASSERT_GTE(ens_route_count, 2); + cbm_store_free_nodes(routes, route_count); + + /* At least one ASYNC_CALLS edge from SendRequestSync literal match */ + int async_calls = cbm_store_count_edges_by_type(s, project, "ASYNC_CALLS"); + ASSERT_GTE(async_calls, 1); + + cbm_store_close(s); + cbm_pipeline_free(p); + th_rmtree(tmpdir); + PASS(); +} + +TEST(pipeline_ensemble_routing_method_scoping) { + /* Verifies scan_source_for_send_targets scopes to the calling method body. + * Two methods in the same class each route to a different operation; each + * must produce exactly one edge to its own target with no cross-contamination. */ + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/cbm_ens2_XXXXXX"); + if (!cbm_mkdtemp(tmpdir)) + FAIL("failed to create temp dir"); + + char path[512]; + + snprintf(path, sizeof(path), "%s/TwoProd.cls", tmpdir); + FILE *f = fopen(path, "w"); + if (!f) { + th_rmtree(tmpdir); + FAIL("fopen production"); + } + fprintf(f, "Class MyApp.TwoProd Extends Ens.Production\n" + "{\n" + "XData ProductionDefinition\n" + "{\n" + "\n" + " \n" + " \n" + " \n" + " \n" + " \n" + " \n" + "\n" + "}\n" + "}\n"); + fclose(f); + + snprintf(path, sizeof(path), "%s/TwoMethodService.cls", tmpdir); + f = fopen(path, "w"); + if (!f) { + th_rmtree(tmpdir); + FAIL("fopen service"); + } + fprintf(f, "Class MyApp.TwoMethodService Extends Ens.BusinessService\n" + "{\n" + "Method MethodOne(pRequest As %%Library.Object," + " Output pResponse As %%Library.Object) As %%Status\n" + "{\n" + " Do ..SendRequestSync(\"OperationA\", pRequest, .pResponse)\n" + " Quit $$$OK\n" + "}\n" + "Method MethodTwo(pRequest As %%Library.Object," + " Output pResponse As %%Library.Object) As %%Status\n" + "{\n" + " Do ..SendRequestSync(\"OperationB\", pRequest, .pResponse)\n" + " Quit $$$OK\n" + "}\n" + "}\n"); + fclose(f); + + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/ens2.db", tmpdir); + + cbm_pipeline_t *p = cbm_pipeline_new(tmpdir, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + ASSERT_EQ(cbm_pipeline_run(p), 0); + + cbm_store_t *s = cbm_store_open_path(db_path); + ASSERT_NOT_NULL(s); + const char *project = cbm_pipeline_project_name(p); + + cbm_node_t *m1_nodes = NULL, *m2_nodes = NULL; + int m1_count = 0, m2_count = 0; + cbm_store_find_nodes_by_name(s, project, "MethodOne", &m1_nodes, &m1_count); + cbm_store_find_nodes_by_name(s, project, "MethodTwo", &m2_nodes, &m2_count); + ASSERT_GTE(m1_count, 1); + ASSERT_GTE(m2_count, 1); + + cbm_node_t *route_nodes = NULL; + int rn_count = 0; + cbm_store_find_nodes_by_label(s, project, "Route", &route_nodes, &rn_count); + int64_t route_a_id = 0, route_b_id = 0; + for (int i = 0; i < rn_count; i++) { + if (route_nodes[i].qualified_name) { + if (strstr(route_nodes[i].qualified_name, "__route__ensemble__") && + strstr(route_nodes[i].qualified_name, "OperationA")) + route_a_id = route_nodes[i].id; + if (strstr(route_nodes[i].qualified_name, "__route__ensemble__") && + strstr(route_nodes[i].qualified_name, "OperationB")) + route_b_id = route_nodes[i].id; + } + } + cbm_store_free_nodes(route_nodes, rn_count); + ASSERT_NEQ(route_a_id, 0); + ASSERT_NEQ(route_b_id, 0); + + cbm_edge_t *m1_edges = NULL; + int m1_ec = 0; + cbm_store_find_edges_by_source_type(s, m1_nodes[0].id, "ASYNC_CALLS", &m1_edges, &m1_ec); + int m1_to_a = 0, m1_to_b = 0; + for (int i = 0; i < m1_ec; i++) { + if (m1_edges[i].target_id == route_a_id) + m1_to_a++; + if (m1_edges[i].target_id == route_b_id) + m1_to_b++; + } + ASSERT_GTE(m1_to_a, 1); + ASSERT_EQ(m1_to_b, 0); + + cbm_edge_t *m2_edges = NULL; + int m2_ec = 0; + cbm_store_find_edges_by_source_type(s, m2_nodes[0].id, "ASYNC_CALLS", &m2_edges, &m2_ec); + int m2_to_a = 0, m2_to_b = 0; + for (int i = 0; i < m2_ec; i++) { + if (m2_edges[i].target_id == route_a_id) + m2_to_a++; + if (m2_edges[i].target_id == route_b_id) + m2_to_b++; + } + ASSERT_GTE(m2_to_b, 1); + ASSERT_EQ(m2_to_a, 0); + + cbm_store_free_edges(m1_edges, m1_ec); + cbm_store_free_edges(m2_edges, m2_ec); + cbm_store_free_nodes(m1_nodes, m1_count); + cbm_store_free_nodes(m2_nodes, m2_count); + cbm_store_close(s); + cbm_pipeline_free(p); + th_rmtree(tmpdir); + PASS(); +} + SUITE(pipeline) { RUN_TEST(pipeline_lsp_surface_persisted_and_body_edit_invariant); /* Index lock */ @@ -12380,6 +12589,9 @@ SUITE(pipeline) { /* Project name edge cases */ RUN_TEST(project_name_special_chars); RUN_TEST(project_name_trailing_slash); + /* Ensemble routing pass */ + RUN_TEST(pipeline_ensemble_routing_edges); + RUN_TEST(pipeline_ensemble_routing_method_scoping); } /* Focused semantic-manifest and publication contracts. Kept separate from the