feat(pipeline): add Ensemble production topology routing pass

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__<prod>.<item>) 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 <tdyar@intersystems.com>
This commit is contained in:
Thomas Dyar committed 2026-08-24 19:55:59 -04:00
1 parent 010569fa6c
commit ee126155ab
5 files changed
+857 -4

No files matched your search

+1
View File
@@ -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
+626
View File
@@ -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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <ctype.h>
#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, "<Item ");
if (!item_start)
break;
ens_item_t *item = &def->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, "</Item>");
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, "<Setting ");
if (!set || set >= 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, "</Setting>");
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, "<Setting ") != NULL &&
strstr(sp, "<Setting ") < item_end) {
cbm_log_warn("ensemble_routing.parse", "item", item->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, "<Item ") != NULL) {
cbm_log_warn("ensemble_routing.parse", "production", def->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, "<Production ");
if (!xml_start)
xml_start = strstr(source, "<Production\n");
if (!xml_start) {
free(source);
continue;
}
ens_prod_def_t *def = parse_production_xml(xml_start, class_qn, xd->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);
}
+8
View File
@@ -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 */
+10 -4
View File
@@ -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,
+212
View File
@@ -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"
"<Production Name=\"MyApp.MyProduction\">\n"
" <Item Name=\"MyService\" ClassName=\"MyApp.MyService\" Enabled=\"true\">\n"
" </Item>\n"
" <Item Name=\"MyOperation\" ClassName=\"MyApp.MyOperation\" Enabled=\"true\">\n"
" </Item>\n"
"</Production>\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"
"<Production Name=\"MyApp.TwoProd\">\n"
" <Item Name=\"OperationA\" ClassName=\"MyApp.OpA\" Enabled=\"true\">\n"
" </Item>\n"
" <Item Name=\"OperationB\" ClassName=\"MyApp.OpB\" Enabled=\"true\">\n"
" </Item>\n"
" <Item Name=\"TwoSvc\" ClassName=\"MyApp.TwoMethodService\" Enabled=\"true\">\n"
" </Item>\n"
"</Production>\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