fix(graph): count edges by both endpoints and expose cross-layer bleed

features/dashboard.py

@@ -194,6 +194,7 @@ async def get_stats(self, user_id: str = "default") -> dict[str, Any]:
             "graph_edges": ug_edges["total"],
             "graph_edges_live": ug_edges["live"],
             "graph_edges_dead": ug_edges["dead"],
+            "graph_edges_cross_layer": ug_edges["cross_layer"],
             "agent_l1": am.l1.size(),
             "agent_l2": await am.l2.count_sessions(user_id),
             "agent_l3": await am.l3.count(user_id),

graph/epistemic.py

@@ -334,18 +334,26 @@ async def count_nodes(self, user_id: str | None = None) -> int:
     async def count_edges(self, user_id: str | None = None) -> dict[str, int]:
         """Count this layer's edges, split into live and dead by weight.
 
-        Returns `{"total", "live", "dead"}`.
-
-        `epi_edges` carries no `layer` column, so an edge belongs to the layer
-        of its SOURCE node — that is the convention counted here. It is NOT the
-        convention `memory_graph_edges` lists by: that reader requires both
-        endpoints in the requested layer, because it returns the target node's
-        content and a source-only filter would pull a foreign layer's text into
-        the answer. The two therefore diverge on a cross-layer edge: counted
-        here, not listed there. Today no such edge exists; a node re-layering
-        would create them, and this counter would then exceed what the tool
-        shows. An edge whose source node is gone is counted by none of the three
-        numbers rather than guessed into one.
+        Returns `{"total", "live", "dead", "cross_layer"}`.
+
+        `epi_edges` carries no `layer` column, so an edge's layer has to be
+        derived — and it is derived the same way `memory_graph_edges` reads it:
+        an edge counts as readable for a layer only when BOTH endpoints are in
+        that layer. The reader returns the target node's content, so an edge with
+        a foreign target is not readable there and must not be counted here
+        either; a source-only count would report edges nobody can list, by the
+        thousand after a node re-layering.
+
+        The universe is every edge incident to this layer on either end, which is
+        why `cross_layer` is symmetric: an edge leaving the layer and an edge
+        entering it are the same edge seen from its two sides, and both are
+        bleed. `cross_layer` is never folded into `total`; it is the size of the
+        remaining layer bleed, and the two together are what the layer touches.
+
+        An edge with a missing endpoint is counted by none of the numbers rather
+        than guessed into one, and it is not cross-layer either: nothing says the
+        layers differ. With `user_id`, the filter applies to whichever endpoint
+        sits in this layer.
 
         "Dead" is `weight <= 0`, not `weight < NREM_FLOOR`: measured on all
         three house databases, only 0.1% of edges fall strictly between zero and
@@ -354,24 +362,43 @@ async def count_edges(self, user_id: str | None = None) -> dict[str, int]:
         revives it (see `_insert_edge`), which is why nothing prunes zero
         eagerly.
         """
-        where = "s.layer = ?"
-        params: tuple[Any, ...] = (self.layer,)
+        same = "s.layer = ? AND t.layer = ?"
+        same_params: list[Any] = [self.layer, self.layer]
+        cross_out = "s.layer = ? AND t.layer IS NOT NULL AND t.layer <> ?"
+        cross_in = "t.layer = ? AND s.layer IS NOT NULL AND s.layer <> ?"
+        cross_params: list[Any] = [self.layer, self.layer, self.layer, self.layer]
         if user_id:
-            where += " AND s.user_id = ?"
-            params = (self.layer, user_id)
-        row = await self._dispatch.select_one(
-            DB_NAME,
-            (
-                "SELECT COUNT(*) AS total,"
-                " COALESCE(SUM(CASE WHEN e.weight > 0 THEN 1 ELSE 0 END), 0) AS live,"
-                " COALESCE(SUM(CASE WHEN e.weight <= 0 THEN 1 ELSE 0 END), 0) AS dead"
-                f" FROM epi_edges e JOIN epi_nodes s ON e.source_id = s.node_id WHERE {where}"
-            ),
-            params,
+            # `memory_graph_edges` scopes user_id on both endpoints; a count that
+            # did not would report edges the reader hides.
+            same += " AND s.user_id = ? AND t.user_id = ?"
+            same_params += [user_id, user_id]
+            cross_out = "s.layer = ? AND s.user_id = ? AND t.layer IS NOT NULL AND t.layer <> ?"
+            cross_in = "t.layer = ? AND t.user_id = ? AND s.layer IS NOT NULL AND s.layer <> ?"
+            cross_params = [self.layer, user_id, self.layer, self.layer, user_id, self.layer]
+        sql = (
+            "SELECT"
+            f" COALESCE(SUM(CASE WHEN {same} THEN 1 ELSE 0 END), 0) AS total,"
+            f" COALESCE(SUM(CASE WHEN {same} AND e.weight > 0 THEN 1 ELSE 0 END), 0) AS live,"
+            f" COALESCE(SUM(CASE WHEN {same} AND e.weight <= 0 THEN 1 ELSE 0 END), 0) AS dead,"
+            f" COALESCE(SUM(CASE WHEN ({cross_out}) OR ({cross_in}) THEN 1 ELSE 0 END), 0) AS cross_layer"
+            " FROM epi_edges e"
+            " LEFT JOIN epi_nodes s ON e.source_id = s.node_id"
+            " LEFT JOIN epi_nodes t ON e.target_id = t.node_id"
+            " WHERE (s.layer = ? OR t.layer = ?)"
         )
+        params: list[Any] = same_params * 3 + cross_params + [self.layer, self.layer]
+        if user_id:
+            sql += " AND ((s.layer = ? AND s.user_id = ?) OR (t.layer = ? AND t.user_id = ?))"
+            params += [self.layer, user_id, self.layer, user_id]
+        row = await self._dispatch.select_one(DB_NAME, sql, tuple(params))
         if not row:
-            return {"total": 0, "live": 0, "dead": 0}
-        return {"total": int(row["total"]), "live": int(row["live"]), "dead": int(row["dead"])}
+            return {"total": 0, "live": 0, "dead": 0, "cross_layer": 0}
+        return {
+            "total": int(row["total"]),
+            "live": int(row["live"]),
+            "dead": int(row["dead"]),
+            "cross_layer": int(row["cross_layer"]),
+        }
 
     async def delete_nodes_older_than(self, user_id: str, cutoff: float) -> int:
         """Delete this layer's nodes created after cutoff; cleans tags/edges."""

mcp_server/models.py

@@ -85,6 +85,9 @@ class StatsResult(_ResultBase):
     graph_edges: int = Field(default=0, description="Graph edge count (all rows)")
     graph_edges_live: int = Field(default=0, description="Graph edges with weight > 0")
     graph_edges_dead: int = Field(default=0, description="Graph edges with weight <= 0 (revivable, not yet pruned)")
+    graph_edges_cross_layer: int = Field(
+        default=0, description="Edges touching this layer but with the other endpoint in another layer (not readable here)"
+    )
     avg_session_quality: float | None = Field(default=None, description="Average session quality score (0-100) over scored sessions")
     recall_count: int = Field(default=0, description="Total recall (dream) calls for the user")
 

mcp_server/tools/ops.py

@@ -76,6 +76,7 @@ async def memory_stats(
         graph_edges=graph_edges["total"],
         graph_edges_live=graph_edges["live"],
         graph_edges_dead=graph_edges["dead"],
+        graph_edges_cross_layer=graph_edges["cross_layer"],
         avg_session_quality=await mem.l2.avg_quality(user_id),
         recall_count=await count_recalls(app.mm._cm, user_id),
     ).dict()

shared/archived_memories.py

@@ -20,7 +20,7 @@
 #: were then deleted, that same day, on her word**, so the figures describe a snapshot
 #: and not the live bin: that base now holds 359 rows and **zero** purges, and the
 #: numbers above reproduce only against the snapshot taken before the deletion
-#: (`/home/murat/backups/cowagent-purge-rows-20261011/memory.db`). The refusal stays,
+#: (`<backup-root>/cowagent-purge-rows-20261011/memory.db`). The refusal stays,
 #: because a row can carry such a reason again. If they ever need to come back, that is
 #: an operator script run by hand, not a tool an agent can call — which is why nothing
 #: here, and nothing in the restore path, has a parameter that turns the refusal off.

tests/test_graph/test_edge_tags.py

@@ -87,35 +87,90 @@ async def test_count_edges_splits_live_and_dead(hermetic_graph):
     conn.commit()
     conn.close()
 
-    assert await g.count_edges("u1") == {"total": 2, "live": 1, "dead": 1}
+    assert await g.count_edges("u1") == {"total": 2, "live": 1, "dead": 1, "cross_layer": 0}
 
 
 async def test_count_edges_is_zeroed_graph(hermetic_graph):
     """No edges at all reports zeros rather than raising or returning None."""
     g, _ = hermetic_graph
     await g.add_node("u1", "одинокий узел", "fact")
-    assert await g.count_edges("u1") == {"total": 0, "live": 0, "dead": 0}
+    assert await g.count_edges("u1") == {"total": 0, "live": 0, "dead": 0, "cross_layer": 0}
 
 
-async def test_count_edges_uses_source_node_layer(hermetic_graph):
-    """epi_edges has no layer column, so an edge belongs to its SOURCE node's layer."""
-    g, tmp = hermetic_graph
-    user_src = await g.add_node("u1", "user source", "fact")
-    user_dst = await g.add_node("u1", "user target", "fact")
-    await g.add_edge(user_src, user_dst, "knows")
+async def test_count_edges_requires_both_endpoints_in_layer(hermetic_graph):
+    """An edge is counted for a layer only when BOTH endpoints are in it.
 
-    agent_src = await g.add_node("u1", "agent source", "fact")
-    agent_dst = await g.add_node("u1", "agent target", "fact")
-    await g.add_edge(agent_src, agent_dst, "knows")
+    `memory_graph_edges` lists edges by the same rule, because it returns the
+    target node's content. Counting by the source alone would report edges the
+    reader hides — after a re-layering the two numbers diverge by thousands.
+    """
+    g, tmp = hermetic_graph
+    a = await g.add_node("u1", "user a", "fact")
+    b = await g.add_node("u1", "user b", "fact")
+    moved = await g.add_node("u1", "moved to agent", "fact")
+    await g.add_edge(a, b, "knows")
+    await g.add_edge(a, moved, "mentions")
     conn = sqlite3.connect(tmp / "memory.db")
-    conn.execute("UPDATE epi_nodes SET layer='agent' WHERE node_id IN (?,?)", (agent_src, agent_dst))
+    conn.execute("UPDATE epi_nodes SET layer='agent' WHERE node_id=?", (moved,))
     conn.commit()
     conn.close()
 
     from graph.epistemic import EpistemicGraph
 
-    assert await g.count_edges("u1") == {"total": 1, "live": 1, "dead": 0}
-    assert await EpistemicGraph(layer="agent", cm=connection_manager).count_edges("u1") == {"total": 1, "live": 1, "dead": 0}
+    user = await g.count_edges("u1")
+    assert user["total"] == 1, "the cross-layer edge is not readable content"
+    assert user["live"] == 1
+    assert user["cross_layer"] == 1
+
+    agent = await EpistemicGraph(layer="agent", cm=connection_manager).count_edges("u1")
+    assert agent["total"] == 0, "no edge has both ends in agent"
+    assert agent["cross_layer"] == 1, "the same edge is visible from the far side"
+
+
+async def test_count_edges_scopes_readable_edges_to_one_user(hermetic_graph):
+    """`memory_graph_edges` scopes user_id on both ends; the counter must agree."""
+    g, _ = hermetic_graph
+    a = await g.add_node("u1", "a", "fact")
+    b = await g.add_node("u1", "b", "fact")
+    foreign = await g.add_node("u2", "other user", "fact")
+    await g.add_edge(a, b, "knows")
+    await g.add_edge(a, foreign, "mentions")
+
+    counts = await g.count_edges("u1")
+    assert counts["total"] == 1, "an edge into another user is not readable for u1"
+    assert counts["live"] == 1
+    assert counts["cross_layer"] == 0, "cross_layer is the layer axis, not the user axis"
+
+
+async def test_count_edges_without_user_scopes_only_by_layer(hermetic_graph):
+    """With no user_id the layer is the only boundary, exactly as before."""
+    g, _ = hermetic_graph
+    a = await g.add_node("u1", "a", "fact")
+    b = await g.add_node("u2", "b", "fact")
+    await g.add_edge(a, b, "knows")
+
+    counts = await g.count_edges()
+    assert counts["total"] == 1, "a layer-only scope spans users"
+    assert counts["cross_layer"] == 0
+
+
+async def test_count_edges_keeps_total_equal_live_plus_dead(hermetic_graph):
+    """total stays the sum of the weight split, cross_layer is a separate axis."""
+    g, tmp = hermetic_graph
+    a = await g.add_node("u1", "a", "fact")
+    b = await g.add_node("u1", "b", "fact")
+    moved = await g.add_node("u1", "moved", "fact")
+    await g.add_edge(a, b, "knows")
+    await g.add_edge(b, a, "knows", weight=0.0)
+    await g.add_edge(a, moved, "mentions")
+    conn = sqlite3.connect(tmp / "memory.db")
+    conn.execute("UPDATE epi_nodes SET layer='agent' WHERE node_id=?", (moved,))
+    conn.commit()
+    conn.close()
+
+    counts = await g.count_edges("u1")
+    assert counts["total"] == counts["live"] + counts["dead"] == 2
+    assert counts["cross_layer"] == 1
 
 
 async def test_count_edges_ignores_edges_whose_source_node_is_gone(hermetic_graph):
@@ -134,4 +189,4 @@ async def test_count_edges_ignores_edges_whose_source_node_is_gone(hermetic_grap
     conn.close()
 
     assert rows == 2, "ребро осталось в таблице"
-    assert await g.count_edges("u1") == {"total": 1, "live": 1, "dead": 0}
+    assert await g.count_edges("u1") == {"total": 1, "live": 1, "dead": 0, "cross_layer": 0}

tests/test_integration.py

@@ -172,6 +172,7 @@ async def test_dashboard():
     assert "l4_facts" in stats
     # the graph used to report nodes only; edge numbers are what make the size visible
     assert stats["graph_edges"] == stats["graph_edges_live"] + stats["graph_edges_dead"]
+    assert "graph_edges_cross_layer" in stats
 
 
 @pytest.mark.asyncio

tests/test_mcp/test_tools_unit.py

@@ -159,13 +159,14 @@ async def test_stats(monkeypatch):
     wiki.count = AsyncMock(return_value=0)
     graph = app.user_graph
     graph.count_nodes = AsyncMock(return_value=7)
-    graph.count_edges = AsyncMock(return_value={"total": 12, "live": 9, "dead": 3})
+    graph.count_edges = AsyncMock(return_value={"total": 12, "live": 9, "dead": 3, "cross_layer": 4})
     result = await memory_stats(layer="user", user_id="u1", ctx=ctx)
     assert isinstance(result, dict)
     assert result["graph_nodes"] == 7
     assert result["graph_edges"] == 12
     assert result["graph_edges_live"] == 9
     assert result["graph_edges_dead"] == 3
+    assert result["graph_edges_cross_layer"] == 4
 
 
 # ── memory_graph_edges direction (B1.1 backlinks) ──