feat(context): named-flow subgraph — call-edges among members (closes #311)

docs/design/0309-named-flow.md

@@ -79,9 +79,11 @@ join with call-edges) and is the piece to delegate.
 
 - **No writing.** Consistent with #305 — deciding a flow's membership is
   authorship, which belongs to the agent.
-- **No call-edges among members (yet).** Showing *how* members connect (the
-  flow subgraph) needs graph resolution; it is the natural phase-2 alongside
-  option (A) and #297 edge-diff ("did the PAYMENT flow's shape change?").
+- ~~No call-edges among members (yet).~~ **Shipped in #311** — `--name X` now
+  includes `edges: [{from, to}]` among members, resolved read-only via
+  `graph_model.query_callees` filtered to the member set (graceful: no graph DB
+  → flat list). The remaining subgraph work is only its pairing with #297
+  edge-diff ("did the PAYMENT flow's shape change between two checkpoints?").
 - **Enclosing-symbol resolution is heuristic**, not a parser. It resolves the
   common docstring and comment-above-def idioms; an unusual layout falls back
   to the file rather than guessing wrong.

scripts/commands/flow.py

@@ -14,11 +14,56 @@
 lives in ``tag_audit_engine``; this only reshapes it flow-first.
 """
 
-from typing import Any, Dict
+import os
+from typing import Any, Dict, List
 
 from tag_audit_engine import audit_tags
 
 
+def _flow_edges(members: List[Dict], workspace, db_path) -> List[Dict[str, str]]:
+    """Call-edges among a flow's members, from the graph if it is populated.
+
+    Graceful by design: a workspace that was never scanned (no graph DB) simply
+    yields no edges and the flat member list stands. Reuses the read-only graph
+    queries — nothing here touches the scan/parser pipeline.
+    """
+    try:
+        from utils import default_db_path
+        import graph_model as gm
+    except Exception:
+        return []
+
+    path = db_path or default_db_path(workspace)
+    if not path or not os.path.exists(path):
+        return []
+
+    # Resolve each member (symbol, file) to a node id. (symbol, file) rather
+    # than symbol alone so two functions sharing a name don't collide.
+    id_to_symbol: Dict[str, str] = {}
+    for m in members:
+        symbol, mfile = m.get("symbol"), m.get("file")
+        if not symbol:
+            continue
+        for node in gm.find_nodes_by_name(symbol, path):
+            if (node.get("file") or "").replace("\\", "/") == mfile:
+                nid = node.get("node_id")
+                if nid:
+                    id_to_symbol[nid] = symbol
+                break
+
+    edges = []
+    seen = set()
+    for nid, from_sym in id_to_symbol.items():
+        for callee in gm.query_callees(nid, path, max_depth=1):
+            cid = callee.get("node_id")
+            if cid in id_to_symbol:  # keep only edges that stay inside the flow
+                pair = (from_sym, id_to_symbol[cid])
+                if pair not in seen:
+                    seen.add(pair)
+                    edges.append({"from": pair[0], "to": pair[1]})
+    return sorted(edges, key=lambda e: (e["from"], e["to"]))
+
+
 def add_args(parser):
     """Register CLI arguments (workspace + name are carried by the umbrella)."""
     parser.add_argument(
@@ -76,4 +121,7 @@
         "found": True,
         "count": match["count"],
         "members": match["members"],
+        # Intra-flow call-edges when the graph is populated; [] otherwise.
+        "edges": _flow_edges(match["members"], workspace,
+                             getattr(args, "db_path", None)),
     }

scripts/formatters/markdown.py

@@ -150,7 +150,7 @@
     return "\n".join(lines)
 
 
 def _md_flow(data: Dict, lines: list) -> None:
     """Markdown for the named-flow view (`context --check flow`, issue #309)."""
     def _member_line(m: Dict) -> str:
         symbol = m.get("symbol") or "(file)"
@@ -169,6 +169,12 @@
         lines.append("")
         for m in data.get("members", []):
             lines.append(_member_line(m))
+        edges = data.get("edges", [])
+        if edges:
+            lines.append("")
+            lines.append("### Call chain")
+            for e in edges:
+                lines.append(f"- `{e.get('from', '')}` → `{e.get('to', '')}`")
         return
 
     # Inventory of all flows.

tests/test_named_flow.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import os
 import sys
 
@@ -133,6 +134,74 @@ def test_bare_flow_lists_every_flow(tree):
     assert out["summary"]["distinct_flows"] == len(out["flows"])
 
 
+# ─── subgraph: call-edges among members (issue #311) ─────
+
+def test_no_graph_db_degrades_to_flat_list(tree):
+    """A workspace that was never scanned yields members but no edges."""
+    out = _run(tree, "PAYMENT")
+    assert out["members"]                 # still collected
+    assert out["edges"] == []             # graceful: no DB, no edges
+
+
+def _populate_graph(root, nodes, edges):
+    cl = os.path.join(root, ".codelens")
+    os.makedirs(cl, exist_ok=True)
+    with open(os.path.join(cl, "backend.json"), "w", encoding="utf-8") as f:
+        json.dump({"nodes": nodes, "edges": edges}, f)
+    import graph_model as gm
+    gm.populate_graph_tables(root)
+
+
+def test_subgraph_edges_among_members(tmp_path):
+    root = str(tmp_path)
+    _write(root, "routes.py",
+           'def checkout(req):\n    """\n    @FLOW: PAY\n    """\n    return validate(req)\n')
+    _write(root, "cart.py",
+           'def validate(c):\n    """\n    @FLOW: PAY\n    """\n    return charge(c)\n')
+    _write(root, "gw.py",
+           'def charge(c):\n    """\n    @FLOW: PAY\n    """\n    return 1\n')
+    _populate_graph(
+        root,
+        nodes=[
+            {"id": "routes.py:1", "fn": "checkout", "file": "routes.py", "line": 1,
+             "ref_count": 0, "status": "active"},
+            {"id": "cart.py:1", "fn": "validate", "file": "cart.py", "line": 1,
+             "ref_count": 1, "status": "active"},
+            {"id": "gw.py:1", "fn": "charge", "file": "gw.py", "line": 1,
+             "ref_count": 1, "status": "active"},
+        ],
+        edges=[
+            {"from": "routes.py:1", "to": "cart.py:1"},
+            {"from": "cart.py:1", "to": "gw.py:1"},
+        ],
+    )
+
+    out = _run(root, "PAY")
+    pairs = {(e["from"], e["to"]) for e in out["edges"]}
+    assert pairs == {("checkout", "validate"), ("validate", "charge")}
+
+
+def test_subgraph_excludes_edges_leaving_the_flow(tmp_path):
+    """An edge to a function outside the flow must not appear."""
+    root = str(tmp_path)
+    _write(root, "a.py",
+           'def a():\n    """\n    @FLOW: F\n    """\n    return helper()\n')
+    _write(root, "h.py", 'def helper():\n    return 1\n')  # not tagged
+    _populate_graph(
+        root,
+        nodes=[
+            {"id": "a.py:1", "fn": "a", "file": "a.py", "line": 1,
+             "ref_count": 0, "status": "active"},
+            {"id": "h.py:1", "fn": "helper", "file": "h.py", "line": 1,
+             "ref_count": 1, "status": "active"},
+        ],
+        edges=[{"from": "a.py:1", "to": "h.py:1"}],
+    )
+
+    out = _run(root, "F")
+    assert out["edges"] == []             # helper is outside the flow
+
+
 if __name__ == "__main__":
     import pytest as _p
     _p.main([__file__, "-v"])