Phase 7A+7C: Subsystem aggregator + Flow tracer (post-review fixes)

flow.js

@@ -0,0 +1,208 @@
+const GraphStore = require('./graph.js');
+const { buildSubsystems, relPath, subsystemOf } = require('./subsystem.js');
+
+/**
+ * Phase 7C: Flow Tracer
+ * Walks the call graph across subsystem boundaries to produce sequenced data flow narratives.
+ */
+
+/**
+ * Build reusable indexes from a graph + subsystem map.
+ * Call once, then pass to traceFlow for each entry point.
+ */
+function buildFlowIndex(graph, subsystemMap, opts = {}) {
+  const godThreshold = opts.godThreshold || 50;
+
+  // File → subsystem lookup
+  const fileSub = new Map();
+  for (const sub of subsystemMap.subsystems) {
+    for (const f of sub.files) fileSub.set(f, sub.name);
+  }
+
+  // Outgoing CALLS index: source → [targets]
+  const callsOut = new Map();
+  for (const e of graph.edges) {
+    if (e.type !== 'CALLS') continue;
+    if (!callsOut.has(e.source)) callsOut.set(e.source, []);
+    callsOut.get(e.source).push(e.target);
+  }
+
+  // Function name → qualified IDs
+  const funcLookup = new Map();
+  for (const [id, node] of graph.nodes) {
+    if (node.type === 'Function' || node.type === 'Class') {
+      if (!funcLookup.has(node.name)) funcLookup.set(node.name, []);
+      funcLookup.get(node.name).push(id);
+    }
+  }
+
+  // In-degree per qualified ID
+  const inDegree = new Map();
+  for (const e of graph.edges) {
+    if (e.type !== 'CALLS') continue;
+    if (e.target.includes('/')) {
+      inDegree.set(e.target, (inDegree.get(e.target) || 0) + 1);
+    } else {
+      // For bare names, increment all candidates
+      const candidates = funcLookup.get(e.target);
+      if (candidates) {
+        for (const c of candidates) {
+          inDegree.set(c, (inDegree.get(c) || 0) + 1);
+        }
+      }
+    }
+  }
+
+  // God objects: qualified IDs with in-degree > threshold
+  const godObjects = new Set();
+  for (const [id, deg] of inDegree) {
+    if (deg > godThreshold) godObjects.add(id);
+  }
+
+  // Subsystem cache
+  const subCache = new Map();
+  function getSubsystem(entityId) {
+    if (!entityId) return null;
+    if (subCache.has(entityId)) return subCache.get(entityId);
+    const file = relPath(entityId.split(':')[0]);
+    const result = fileSub.get(file) || null;
+    subCache.set(entityId, result);
+    return result;
+  }
+
+  // Resolve bare function name → qualified ID, preferring caller's subsystem
+  function resolveTarget(bareName, callerSub) {
+    const candidates = funcLookup.get(bareName);
+    if (!candidates || candidates.length === 0) return null;
+    if (candidates.length === 1) return candidates[0];
+    for (const c of candidates) {
+      if (getSubsystem(c) === callerSub) return c;
+    }
+    return candidates[0];
+  }
+
+  return { fileSub, callsOut, funcLookup, godObjects, getSubsystem, resolveTarget };
+}
+
+/**
+ * Trace a flow from an entry point through the call graph.
+ * @param {string} entryPoint - Entity ID (e.g. "channels/telegram.ts:onTelegramMessage")
+ * @param {object} index - Precomputed index from buildFlowIndex
+ * @param {object} opts - Options: { maxDepth, timeoutMs }
+ */
+function traceFlow(entryPoint, index, opts = {}) {
+  const maxDepth = opts.maxDepth || 8;
+  const timeout = opts.timeoutMs || 5000;
+  const startTime = Date.now();
+
+  const { callsOut, godObjects, getSubsystem, resolveTarget } = index;
+
+  const visited = new Set();
+  const flow = [];
+  const cyclesDetected = [];
+  const excludedNodes = new Set();
+
+  const entrySub = getSubsystem(entryPoint);
+  if (!entrySub) {
+    return { entryPoint, error: `Entry point "${entryPoint}" not found in any subsystem`, flow: [], subsystemSequence: [], cyclesDetected: [], excludedNodes: [] };
+  }
+
+  // BFS with index pointer (no shift)
+  const queue = [[entryPoint, 0]];
+  let head = 0;
+  visited.add(entryPoint);
+  flow.push({ subsystem: entrySub, entity: entryPoint, depth: 0 });
+
+  const subsystemSequence = [entrySub];
+  const seqSet = new Set([entrySub]);
+
+  while (head < queue.length) {
+    if (Date.now() - startTime > timeout) break;
+
+    const [current, depth] = queue[head++];
+    const currentSub = getSubsystem(current);
+
+    const targets = callsOut.get(current) || [];
+    for (const rawTarget of targets) {
+      // Skip test files
+      if (rawTarget.includes('.test.') || rawTarget.includes('.spec.') || rawTarget.includes('__tests__/')) continue;
+
+      let resolvedTarget = rawTarget;
+      let targetSub = null;
+
+      if (rawTarget.includes('/')) {
+        // Qualified target — check god object by qualified ID
+        if (godObjects.has(rawTarget)) {
+          excludedNodes.add(rawTarget);
+          continue;
+        }
+        targetSub = getSubsystem(rawTarget);
+      } else {
+        // Bare name — resolve to qualified ID first, then check god status
+        resolvedTarget = resolveTarget(rawTarget, currentSub);
+        if (!resolvedTarget) continue;
+        if (godObjects.has(resolvedTarget)) {
+          excludedNodes.add(resolvedTarget);
+          continue;
+        }
+        targetSub = getSubsystem(resolvedTarget);
+      }
+
+      if (!targetSub) continue;
+
+      // Cycle detection
+      if (visited.has(resolvedTarget)) {
+        cyclesDetected.push({ at: current, backEdgeTo: resolvedTarget });
+        continue;
+      }
+
+      // Compute new depth
+      const isCrossSubsystem = targetSub !== currentSub;
+      const newDepth = depth + (isCrossSubsystem ? 1 : 0.5);
+
+      if (newDepth > maxDepth) continue;
+
+      visited.add(resolvedTarget);
+      flow.push({
+        subsystem: targetSub,
+        entity: resolvedTarget,
+        depth: newDepth,
+        ...(isCrossSubsystem ? { crossedVia: 'CALLS' } : {})
+      });
+
+      if (isCrossSubsystem && !seqSet.has(targetSub)) {
+        subsystemSequence.push(targetSub);
+        seqSet.add(targetSub);
+      }
+
+      queue.push([resolvedTarget, newDepth]);
+    }
+  }
+
+  return {
+    entryPoint,
+    depth: maxDepth,
+    godThreshold: index.godObjects.size > 0 ? 'applied' : 'none',
+    excludedNodes: Array.from(excludedNodes),
+    cyclesDetected,
+    flow,
+    subsystemSequence
+  };
+}
+
+if (require.main === module) {
+  const snapshotPath = process.argv[2];
+  const entryPoint = process.argv[3];
+  const godThreshold = parseInt(process.argv[4]) || 50;
+  if (!snapshotPath || !entryPoint) {
+    console.error('Usage: node flow.js <snapshot.json> <entryPoint> [godThreshold]');
+    process.exit(1);
+  }
+  const graph = GraphStore.loadSnapshot(snapshotPath);
+  const subsystemMap = buildSubsystems(graph);
+  const index = buildFlowIndex(graph, subsystemMap, { godThreshold });
+  const result = traceFlow(entryPoint, index);
+  console.log(JSON.stringify(result, null, 2));
+}
+
+module.exports = { buildFlowIndex, traceFlow };