feat(pipeline): add Socket.IO and EventEmitter channel detection

Detect emit/listen channel patterns in JS/TS/Python source files during
indexing. Extracts socket.emit/on, io.emit/on, emitter.emit/on patterns
with a regex scanner that identifies receiver names against a whitelist
of known channel communicators (socket, io, emitter, eventBus, etc.).

Filters out generic Node.js stream events (error, close, data, etc.)
and classifies transport as 'socketio' or 'eventemitter' based on
receiver name.

New schema: 'channels' table (project, channel_name, direction, transport,
node_id, file_path, function_name) with indexes on channel_name and project.

New store API: cbm_store_detect_channels() scans source from disk for all
indexed Function/Method/Module nodes in JS/TS/Python files.
cbm_store_find_channels() queries by project and/or channel name with
partial matching. Automatic cross-repo matching at query time (no link step).

New MCP tool: get_channels returns matched channels with emitter/listener
info, filterable by channel name and project.

Tested: TS monorepo detects 210 channel references including Socket.IO
subscribe/unsubscribe flows between UI and server.

Author: Your Name

src/mcp/mcp.c

@@ -288,6 +288,14 @@ static const tool_def_t TOOLS[] = {
      "community detection. Returns up to 300 processes ordered by step count.",
      "{\"type\":\"object\",\"properties\":{\"project\":{\"type\":\"string\"}},\"required\":[\"project\"]}"},
 
+    {"get_channels",
+     "Find message channels (Socket.IO events, EventEmitter signals) across projects. "
+     "Shows which functions emit and listen on each channel, enabling cross-service "
+     "message flow tracing. Auto-detects patterns during indexing. "
+     "Query by channel name (partial match) and/or project.",
+     "{\"type\":\"object\",\"properties\":{\"project\":{\"type\":\"string\"},"
+     "\"channel\":{\"type\":\"string\",\"description\":\"Channel name filter (partial match)\"}}}"},
+
     {"search_code",
      "Graph-augmented code search. Finds text patterns via grep, then enriches results with "
      "the knowledge graph: deduplicates matches into containing functions, ranks by structural "
@@ -1165,6 +1173,53 @@ static char *handle_delete_project(cbm_mcp_server_t *srv, const char *args) {
     return result;
 }
 
+static char *handle_get_channels(cbm_mcp_server_t *srv, const char *args) {
+    char *project = cbm_mcp_get_string_arg(args, "project");
+    char *channel = cbm_mcp_get_string_arg(args, "channel");
+    cbm_store_t *store = resolve_store(srv, project);
+    REQUIRE_STORE(store, project);
+
+    cbm_channel_info_t *channels = NULL;
+    int count = 0;
+    cbm_store_find_channels(store, project, channel, &channels, &count);
+
+    yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
+    yyjson_mut_val *root = yyjson_mut_obj(doc);
+    yyjson_mut_doc_set_root(doc, root);
+
+    yyjson_mut_obj_add_int(doc, root, "total", count);
+
+    /* Group by channel name for readable output */
+    yyjson_mut_val *arr = yyjson_mut_arr(doc);
+    for (int i = 0; i < count; i++) {
+        yyjson_mut_val *item = yyjson_mut_obj(doc);
+        yyjson_mut_obj_add_strcpy(doc, item, "channel",
+                                  channels[i].channel_name ? channels[i].channel_name : "");
+        yyjson_mut_obj_add_strcpy(doc, item, "direction",
+                                  channels[i].direction ? channels[i].direction : "");
+        yyjson_mut_obj_add_strcpy(doc, item, "transport",
+                                  channels[i].transport ? channels[i].transport : "");
+        yyjson_mut_obj_add_strcpy(doc, item, "project",
+                                  channels[i].project ? channels[i].project : "");
+        yyjson_mut_obj_add_strcpy(doc, item, "file",
+                                  channels[i].file_path ? channels[i].file_path : "");
+        yyjson_mut_obj_add_strcpy(doc, item, "function",
+                                  channels[i].function_name ? channels[i].function_name : "");
+        yyjson_mut_arr_add_val(arr, item);
+    }
+    yyjson_mut_obj_add_val(doc, root, "channels", arr);
+
+    char *json = yy_doc_to_str(doc);
+    yyjson_mut_doc_free(doc);
+    cbm_store_free_channels(channels, count);
+    free(project);
+    free(channel);
+
+    char *result = cbm_mcp_text_result(json, false);
+    free(json);
+    return result;
+}
+
 static char *handle_list_processes(cbm_mcp_server_t *srv, const char *args) {
     char *project = cbm_mcp_get_string_arg(args, "project");
     cbm_store_t *store = resolve_store(srv, project);
@@ -2968,6 +3023,9 @@ char *cbm_mcp_handle_tool(cbm_mcp_server_t *srv, const char *tool_name, const ch
     if (strcmp(tool_name, "list_processes") == 0) {
         return handle_list_processes(srv, args_json);
     }
+    if (strcmp(tool_name, "get_channels") == 0) {
+        return handle_get_channels(srv, args_json);
+    }
 
     /* Pipeline-dependent tools */
     if (strcmp(tool_name, "index_repository") == 0) {

src/pipeline/httplink.c

@@ -362,7 +362,6 @@ static int count_segments(const char *path) {
     return count;
 }
 
-/* Jaccard similarity of path segments (intersection/union) */
 static double segment_jaccard(const char *norm_call, const char *norm_route) {
     /* Split into segments */
     char a[1024];
@@ -1907,3 +1906,92 @@ int cbm_httplink_all_exclude_paths(const cbm_httplink_config_t *cfg, const char
 
     return count;
 }
+
+/* ── Channel extraction: Socket.IO / EventEmitter ────────────────── */
+
+typedef struct cbm_channel_match {
+    char channel[256];
+    char direction[8]; /* "emit" or "listen" */
+    char transport[32]; /* "socketio", "eventemitter" */
+} cbm_channel_match_t;
+
+int cbm_extract_channels(const char *source, cbm_channel_match_t *out, int max_out) {
+    if (!source || !*source) return 0;
+
+    cbm_regex_t re;
+    if (cbm_regcomp(&re,
+            "([a-zA-Z_][a-zA-Z0-9_]*)\\.("
+            "emit|on|once|addListener|removeListener"
+            ")\\([[:space:]]*['\"`]([^'\"`]{1,128})['\"`]",
+            CBM_REG_EXTENDED) != 0) {
+        return 0;
+    }
+
+    static const char *channel_receivers[] = {
+        "socket", "io", "client", "server", "connection",
+        "emitter", "eventEmitter", "eventBus", "bus", "pubsub",
+        "producer", "consumer", "channel", "broker",
+        "nsp", "namespace", "this", NULL
+    };
+
+    int count = 0;
+    const char *p = source;
+    cbm_regmatch_t match[4];
+
+    while (count < max_out && cbm_regexec(&re, p, 4, match, 0) == 0) {
+        int rlen = match[1].rm_eo - match[1].rm_so;
+        char receiver[64];
+        if (rlen >= (int)sizeof(receiver)) rlen = (int)sizeof(receiver) - 1;
+        memcpy(receiver, p + match[1].rm_so, (size_t)rlen);
+        receiver[rlen] = '\0';
+
+        bool is_channel = false;
+        for (int i = 0; channel_receivers[i]; i++) {
+            if (strcasecmp(receiver, channel_receivers[i]) == 0) {
+                is_channel = true;
+                break;
+            }
+        }
+
+        if (is_channel) {
+            int mlen = match[2].rm_eo - match[2].rm_so;
+            char method[32];
+            if (mlen >= (int)sizeof(method)) mlen = (int)sizeof(method) - 1;
+            memcpy(method, p + match[2].rm_so, (size_t)mlen);
+            method[mlen] = '\0';
+
+            int clen = match[3].rm_eo - match[3].rm_so;
+            if (clen >= (int)sizeof(out[count].channel))
+                clen = (int)sizeof(out[count].channel) - 1;
+            memcpy(out[count].channel, p + match[3].rm_so, (size_t)clen);
+            out[count].channel[clen] = '\0';
+
+            const char *ch = out[count].channel;
+            if (strcmp(ch, "error") != 0 && strcmp(ch, "close") != 0 &&
+                strcmp(ch, "end") != 0 && strcmp(ch, "data") != 0 &&
+                strcmp(ch, "connect") != 0 && strcmp(ch, "disconnect") != 0 &&
+                strcmp(ch, "connection") != 0 && strcmp(ch, "message") != 0 &&
+                strcmp(ch, "open") != 0 && strcmp(ch, "drain") != 0 &&
+                strcmp(ch, "finish") != 0 && strcmp(ch, "pipe") != 0 &&
+                strcmp(ch, "unpipe") != 0 && strcmp(ch, "readable") != 0 &&
+                strcmp(ch, "resume") != 0 && strcmp(ch, "pause") != 0) {
+                if (strcmp(method, "emit") == 0) {
+                    strncpy(out[count].direction, "emit", sizeof(out[count].direction) - 1);
+                } else {
+                    strncpy(out[count].direction, "listen", sizeof(out[count].direction) - 1);
+                }
+                if (strcasecmp(receiver, "socket") == 0 || strcasecmp(receiver, "io") == 0 ||
+                    strcasecmp(receiver, "nsp") == 0 || strcasecmp(receiver, "namespace") == 0) {
+                    strncpy(out[count].transport, "socketio", sizeof(out[count].transport) - 1);
+                } else {
+                    strncpy(out[count].transport, "eventemitter", sizeof(out[count].transport) - 1);
+                }
+                count++;
+            }
+        }
+        p += match[0].rm_eo;
+    }
+
+    cbm_regfree(&re);
+    return count;
+}

src/pipeline/pipeline.c

@@ -839,6 +839,17 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
                     cbm_store_close(proc_store);
                 }
             }
+
+            /* ── Channel detection: scan source for emit/on patterns ── */
+            {
+                cbm_store_t *ch_store = cbm_store_open_path(db_path);
+                if (ch_store) {
+                    int nch = cbm_store_detect_channels(ch_store, p->project_name, p->repo_path);
+                    cbm_log_info("pass.done", "pass", "channels",
+                                 "detected", itoa_buf(nch));
+                    cbm_store_close(ch_store);
+                }
+            }
         }
     }
 

src/store/store.c

@@ -206,6 +206,18 @@ static int init_schema(cbm_store_t *s) {
                        "  step INTEGER NOT NULL,"
                        "  PRIMARY KEY (process_id, step)"
                        ");"
+                       "CREATE TABLE IF NOT EXISTS channels ("
+                       "  id INTEGER PRIMARY KEY AUTOINCREMENT,"
+                       "  project TEXT NOT NULL REFERENCES projects(name) ON DELETE CASCADE,"
+                       "  channel_name TEXT NOT NULL,"
+                       "  direction TEXT NOT NULL,"  /* 'emit' or 'listen' */
+                       "  transport TEXT NOT NULL DEFAULT 'socketio',"
+                       "  node_id INTEGER NOT NULL,"
+                       "  file_path TEXT DEFAULT '',"
+                       "  function_name TEXT DEFAULT ''"
+                       ");"
+                       "CREATE INDEX IF NOT EXISTS idx_channels_name ON channels(channel_name);"
+                       "CREATE INDEX IF NOT EXISTS idx_channels_project ON channels(project);"
                        "CREATE TABLE IF NOT EXISTS project_summaries ("
                       "  project TEXT PRIMARY KEY,"
                       "  summary TEXT NOT NULL,"
@@ -4778,6 +4790,179 @@ void cbm_store_free_process_steps(cbm_process_step_t *arr, int count) {
     free(arr);
 }
 
+/* ── Channels (cross-service message tracing) ────────────────────── */
+
+/* Forward declaration of channel extractor from httplink.c */
+typedef struct {
+    char channel[256];
+    char direction[8];
+    char transport[32];
+} cbm_channel_match_t;
+int cbm_extract_channels(const char *source, cbm_channel_match_t *out, int max_out);
+
+int cbm_store_detect_channels(cbm_store_t *s, const char *project, const char *repo_path) {
+    if (!s || !s->db || !project || !repo_path) return 0;
+
+    /* Clear existing channels for this project */
+    char del[256];
+    snprintf(del, sizeof(del), "DELETE FROM channels WHERE project = '%s'", project);
+    exec_sql(s, del);
+
+    /* Find all JS/TS Function/Method nodes with source file references */
+    const char *sql = "SELECT id, name, file_path, start_line, end_line FROM nodes "
+                      "WHERE project = ?1 AND label IN ('Function','Method','Module') "
+                      "AND (file_path LIKE '%.ts' OR file_path LIKE '%.js' "
+                      "OR file_path LIKE '%.tsx' OR file_path LIKE '%.py')";
+    sqlite3_stmt *stmt = NULL;
+    if (sqlite3_prepare_v2(s->db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
+    bind_text(stmt, 1, project);
+
+    sqlite3_stmt *ins = NULL;
+    sqlite3_prepare_v2(s->db,
+        "INSERT INTO channels(project,channel_name,direction,transport,node_id,file_path,function_name) "
+        "VALUES(?1,?2,?3,?4,?5,?6,?7)", -1, &ins, NULL);
+
+    exec_sql(s, "BEGIN TRANSACTION");
+    int total = 0;
+
+    while (sqlite3_step(stmt) == SQLITE_ROW) {
+        int64_t node_id = sqlite3_column_int64(stmt, 0);
+        const char *name = (const char *)sqlite3_column_text(stmt, 1);
+        const char *fpath = (const char *)sqlite3_column_text(stmt, 2);
+        int start = sqlite3_column_int(stmt, 3);
+        int end = sqlite3_column_int(stmt, 4);
+
+        if (!fpath || !fpath[0] || start <= 0 || end <= 0) continue;
+
+        /* Read source lines from disk */
+        char full_path[2048];
+        snprintf(full_path, sizeof(full_path), "%s/%s", repo_path, fpath);
+
+        FILE *f = fopen(full_path, "r");
+        if (!f) continue;
+
+        /* Read relevant lines */
+        char *source = NULL;
+        size_t src_len = 0;
+        size_t src_cap = 0;
+        int line_num = 0;
+        char line[4096];
+
+        while (fgets(line, sizeof(line), f)) {
+            line_num++;
+            if (line_num < start) continue;
+            if (line_num > end) break;
+            size_t ll = strlen(line);
+            if (src_len + ll >= src_cap) {
+                src_cap = (src_cap == 0) ? 4096 : src_cap * 2;
+                source = safe_realloc(source, src_cap);
+            }
+            memcpy(source + src_len, line, ll);
+            src_len += ll;
+        }
+        fclose(f);
+
+        if (source) {
+            source[src_len] = '\0';
+            cbm_channel_match_t matches[64];
+            int mc = cbm_extract_channels(source, matches, 64);
+            for (int i = 0; i < mc && ins; i++) {
+                sqlite3_reset(ins);
+                bind_text(ins, 1, project);
+                bind_text(ins, 2, matches[i].channel);
+                bind_text(ins, 3, matches[i].direction);
+                bind_text(ins, 4, matches[i].transport);
+                sqlite3_bind_int64(ins, 5, node_id);
+                bind_text(ins, 6, fpath);
+                bind_text(ins, 7, name ? name : "");
+                sqlite3_step(ins);
+                total++;
+            }
+            free(source);
+        }
+    }
+
+    exec_sql(s, "COMMIT");
+    sqlite3_finalize(stmt);
+    if (ins) sqlite3_finalize(ins);
+    return total;
+}
+
+int cbm_store_find_channels(cbm_store_t *s, const char *project, const char *channel,
+                            cbm_channel_info_t **out, int *count) {
+    *out = NULL;
+    *count = 0;
+
+    /* Build query — if project is NULL, search all; if channel is NULL, return all */
+    char sql[1024];
+    if (project && channel) {
+        snprintf(sql, sizeof(sql),
+                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "FROM channels WHERE project = ?1 AND channel_name LIKE ?2 "
+                 "ORDER BY channel_name LIMIT 500");
+    } else if (project) {
+        snprintf(sql, sizeof(sql),
+                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "FROM channels WHERE project = ?1 ORDER BY channel_name LIMIT 500");
+    } else if (channel) {
+        snprintf(sql, sizeof(sql),
+                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "FROM channels WHERE channel_name LIKE ?1 ORDER BY channel_name LIMIT 500");
+    } else {
+        snprintf(sql, sizeof(sql),
+                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "FROM channels ORDER BY channel_name LIMIT 500");
+    }
+
+    sqlite3_stmt *stmt = NULL;
+    if (sqlite3_prepare_v2(s->db, sql, -1, &stmt, NULL) != SQLITE_OK) return CBM_STORE_OK;
+
+    int bi = 0;
+    if (project && channel) {
+        bind_text(stmt, 1, project);
+        char pat[256];
+        snprintf(pat, sizeof(pat), "%%%s%%", channel);
+        bind_text(stmt, 2, pat);
+    } else if (project) {
+        bind_text(stmt, 1, project);
+    } else if (channel) {
+        char pat[256];
+        snprintf(pat, sizeof(pat), "%%%s%%", channel);
+        bind_text(stmt, 1, pat);
+    }
+    (void)bi;
+
+    int cap = 64;
+    int n = 0;
+    cbm_channel_info_t *arr = calloc((size_t)cap, sizeof(cbm_channel_info_t));
+    while (sqlite3_step(stmt) == SQLITE_ROW) {
+        if (n >= cap) { cap *= 2; arr = safe_realloc(arr, (size_t)cap * sizeof(cbm_channel_info_t)); }
+        arr[n].channel_name = heap_strdup((const char *)sqlite3_column_text(stmt, 0));
+        arr[n].direction = heap_strdup((const char *)sqlite3_column_text(stmt, 1));
+        arr[n].transport = heap_strdup((const char *)sqlite3_column_text(stmt, 2));
+        arr[n].project = heap_strdup((const char *)sqlite3_column_text(stmt, 3));
+        arr[n].file_path = heap_strdup((const char *)sqlite3_column_text(stmt, 4));
+        arr[n].function_name = heap_strdup((const char *)sqlite3_column_text(stmt, 5));
+        n++;
+    }
+    sqlite3_finalize(stmt);
+    *out = arr;
+    *count = n;
+    return CBM_STORE_OK;
+}
+
+void cbm_store_free_channels(cbm_channel_info_t *arr, int count) {
+    for (int i = 0; i < count; i++) {
+        free((void *)arr[i].channel_name);
+        free((void *)arr[i].direction);
+        free((void *)arr[i].transport);
+        free((void *)arr[i].project);
+        free((void *)arr[i].file_path);
+        free((void *)arr[i].function_name);
+    }
+    free(arr);
+}
+
 /* ── ADR (Architecture Decision Record) ────────────────────────── */
 
 static const char *canonical_sections[] = {"PURPOSE",  "STACK",     "ARCHITECTURE",