fix(ensemble): segment-anchored class matching; cbm_fopen; add ROUTES_TO pipeline test

- class_name_matches() replaces strstr for method_belongs_to_production:
  exact or dot-prefixed suffix match prevents 'Ens' matching 'Ens.BusinessService'
- find_entry_point: strcmp instead of strstr for QN matching
- read_file: cbm_fopen for Windows UTF-8 path compatibility
- pipeline_ensemble_routing_routes_to_edges: full pipeline test with two
  .cls fixtures asserting >= 2 EnsembleItem nodes and >= 1 ROUTES_TO edge

Signed-off-by: Thomas Dyar <tdyar@intersystems.com>

Author: isc-tdyar

src/pipeline/pass_ensemble_routing.c

@@ -176,7 +176,7 @@ static ens_prod_def_t *parse_production_xml(const char *xml, const char *class_q
 }
 
 static char *read_file(const char *full_path) {
-    FILE *f = fopen(full_path, "rb");
+    FILE *f = cbm_fopen(full_path, "rb");
     if (!f)
         return NULL;
     fseek(f, 0, SEEK_END);
@@ -234,7 +234,7 @@ static int64_t find_entry_point(cbm_pipeline_ctx_t *ctx, const char *class_name)
         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 && strstr(nodes[ni]->qualified_name, suffix))
+            if (nodes[ni]->qualified_name && strcmp(nodes[ni]->qualified_name, suffix) == 0)
                 return nodes[ni]->id;
         }
     }
@@ -422,14 +422,30 @@ static void collect_prod_defs(cbm_pipeline_ctx_t *ctx, ens_prod_def_t ***defs_ou
     *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 (strstr(parent_class, def->items[i].class_name))
+        if (class_name_matches(parent_class, def->items[i].class_name))
             return true;
     }
     return false;

tests/test_pipeline.c

@@ -6881,6 +6881,81 @@ TEST(pipeline_seq_ts_cross_uses_shared_registry) {
     PASS();
 }
 
+TEST(pipeline_ensemble_routing_routes_to_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);
+
+    /* EnsembleItem nodes emitted for both production items */
+    cbm_node_t *items = NULL;
+    int item_count = 0;
+    cbm_store_find_nodes_by_label(s, project, "EnsembleItem", &items, &item_count);
+    ASSERT_GTE(item_count, 2);
+    cbm_store_free_nodes(items, item_count);
+
+    /* At least one ROUTES_TO edge from SendRequestSync literal match */
+    int routes = cbm_store_count_edges_by_type(s, project, "ROUTES_TO");
+    ASSERT_GTE(routes, 1);
+
+    cbm_store_close(s);
+    cbm_pipeline_free(p);
+    th_rmtree(tmpdir);
+    PASS();
+}
+
 SUITE(pipeline) {
     /* Index lock */
     RUN_TEST(pipeline_lock_try_acquire);
@@ -7157,4 +7232,6 @@ 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_routes_to_edges);
 }