feat(graph): report edge counts, and narrow the NREM read in SQL

features/dashboard.py

@@ -182,6 +182,7 @@ async def get_stats(self, user_id: str = "default") -> dict[str, Any]:
 
         um = self.mm.user_memory(user_id)
         am = self.mm.agent_memory(user_id)
+        ug_edges = await ug.count_edges(user_id)
 
         return {
             "l1_buffer": um.l1.size(),
@@ -190,6 +191,9 @@ async def get_stats(self, user_id: str = "default") -> dict[str, Any]:
             "l4_facts": await um.l4.count(user_id),
             "wiki_pages": await uw.count(user_id),
             "graph_nodes": await ug.count_nodes(user_id),
+            "graph_edges": ug_edges["total"],
+            "graph_edges_live": ug_edges["live"],
+            "graph_edges_dead": ug_edges["dead"],
             "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

@@ -317,6 +317,41 @@ async def count_nodes(self, user_id: str | None = None) -> int:
         row = await cur.fetchone()
         return row[0] if row else 0
 
+    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 — the same convention `memory_graph_edges` already
+        uses. An edge whose source node is gone is counted by none of the three
+        numbers rather than guessed into one.
+
+        "Dead" is `weight <= 0`, not `weight < NREM_FLOOR`: measured on all
+        three house databases, only 0.1% of edges fall strictly between zero and
+        the 0.05 floor, so the coarse split is the accurate one. A dead edge is
+        not garbage — it is the resting state from which the next miner pass
+        revives it (see `_insert_edge`), which is why nothing prunes zero
+        eagerly.
+        """
+        conn = await self._cm.get(DB_NAME)
+        where = "s.layer = ?"
+        params: tuple[Any, ...] = (self.layer,)
+        if user_id:
+            where += " AND s.user_id = ?"
+            params = (self.layer, user_id)
+        cur = await conn.execute(
+            "SELECT COUNT(*),"
+            " COALESCE(SUM(CASE WHEN e.weight > 0 THEN 1 ELSE 0 END), 0),"
+            " COALESCE(SUM(CASE WHEN e.weight <= 0 THEN 1 ELSE 0 END), 0)"
+            f" FROM epi_edges e JOIN epi_nodes s ON e.source_id = s.node_id WHERE {where}",
+            params,
+        )
+        row = await cur.fetchone()
+        if not row:
+            return {"total": 0, "live": 0, "dead": 0}
+        return {"total": int(row[0]), "live": int(row[1]), "dead": int(row[2])}
+
     async def delete_nodes_older_than(self, user_id: str, cutoff: float) -> int:
         """Delete this layer's nodes created after cutoff; cleans tags/edges."""
         sql_ids = "SELECT node_id FROM epi_nodes WHERE layer=? AND user_id=? AND created_at > ?"

lifecycle/graph_enrich.py

@@ -76,9 +76,29 @@ async def _dream_nrem(conn: Any, now: float) -> dict[str, int]:
 
     Freshly co-fired heuristic edges +0.05, stale inactive ones −0.01;
     weight < floor — prune.
+
+    Only the two actionable age bands are read, and the bounds are decided in
+    SQL. An edge between a day and a month old is most of the graph and nothing
+    happens to it, so pulling those rows into Python only to compute an age and
+    fall through was the bulk of the work: on the busiest house database this
+    read 155 486 heuristic edges to touch a few thousand. The comparison is the
+    same one the loop made — `created_at < now − 30 d` is stale, and
+    `created_at >= now − 1 d` is fresh, inclusive, matching `age_days <= 1`.
+
+    A row with a NULL `created_at` is now skipped rather than raising
+    `float(None)`; no house database has one, and "unknown age" is not a reason
+    to decay, boost, or prune an edge.
     """
+    stale_before = now - NREM_STALE_DAYS * 86400.0
+    fresh_from = now - NREM_FRESH_DAYS * 86400.0
     decayed = pruned = boosted = 0
-    for r in await _rows(conn, "SELECT source_id, target_id, relation, weight, created_at FROM epi_edges WHERE tags LIKE '%heuristic:%'"):
+    rows = await _rows(
+        conn,
+        "SELECT source_id, target_id, relation, weight, created_at FROM epi_edges"
+        " WHERE tags LIKE '%heuristic:%' AND (created_at < ? OR created_at >= ?)",
+        (stale_before, fresh_from),
+    )
+    for r in rows:
         w = float(r["weight"])
         age_days = (now - float(r["created_at"])) / 86400.0
         key = (r["source_id"], r["target_id"], r["relation"])

mcp_server/models.py

@@ -59,6 +59,9 @@ class StatsResult(BaseModel):
     l4_facts: int = Field(description="L4 fact count")
     wiki_pages: int = Field(description="Wiki page count")
     graph_nodes: int = Field(description="Graph node count")
+    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)")
     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

@@ -58,6 +58,7 @@ async def memory_stats(
     wiki = _get_wiki(app, layer)
     graph = _get_graph(app, layer)
     l3_count = await mem.l3.count(user_id)
+    graph_edges = await graph.count_edges(user_id)
     from features.recall_telemetry import count_recalls
 
     return StatsResult(
@@ -67,6 +68,9 @@ async def memory_stats(
         l4_facts=await mem.l4.count(user_id),
         wiki_pages=await wiki.count(),
         graph_nodes=await graph.count_nodes(user_id),
+        graph_edges=graph_edges["total"],
+        graph_edges_live=graph_edges["live"],
+        graph_edges_dead=graph_edges["dead"],
         avg_session_quality=await mem.l2.avg_quality(user_id),
         recall_count=await count_recalls(app.mm._cm, user_id),
     ).dict()

tests/test_graph/test_edge_tags.py

@@ -63,3 +63,75 @@ async def test_add_edge_backcompat_no_tags(hermetic_graph):
     dst = await g.add_node("u1", "b", "fact")
     await g.add_edge(src, dst, "knows")  # no tags — old call sites keep working
     assert True
+
+
+# ── count_edges: the graph's edge numbers, split live/dead ──
+
+
+async def test_count_edges_splits_live_and_dead(hermetic_graph):
+    """A zeroed weight is counted as dead, not dropped and not mixed into live.
+
+    Zero is what `lateral_inhibition` writes and what `_insert_edge` later
+    overwrites, so it must be visible as its own number: reporting only a total
+    would hide that most of a graph can be sitting at zero.
+    """
+    g, tmp = hermetic_graph
+    a = await g.add_node("u1", "узел a", "fact")
+    b = await g.add_node("u1", "узел b", "fact")
+    c = await g.add_node("u1", "узел c", "fact")
+    await g.add_edge(a, b, "knows", weight=0.8)
+    await g.add_edge(b, c, "knows", weight=0.8)
+
+    conn = sqlite3.connect(tmp / "memory.db")
+    conn.execute("UPDATE epi_edges SET weight=0 WHERE source_id=? AND target_id=?", (b, c))
+    conn.commit()
+    conn.close()
+
+    assert await g.count_edges("u1") == {"total": 2, "live": 1, "dead": 1}
+
+
+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}
+
+
+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")
+
+    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")
+    conn = sqlite3.connect(tmp / "memory.db")
+    conn.execute("UPDATE epi_nodes SET layer='agent' WHERE node_id IN (?,?)", (agent_src, agent_dst))
+    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}
+
+
+async def test_count_edges_ignores_edges_whose_source_node_is_gone(hermetic_graph):
+    """An orphan edge is in no bucket rather than guessed into one."""
+    g, tmp = hermetic_graph
+    a = await g.add_node("u1", "a", "fact")
+    b = await g.add_node("u1", "b", "fact")
+    orphan = await g.add_node("u1", "orphan", "fact")
+    await g.add_edge(a, b, "knows")
+    await g.add_edge(orphan, b, "knows")
+
+    conn = sqlite3.connect(tmp / "memory.db")
+    conn.execute("DELETE FROM epi_nodes WHERE node_id=?", (orphan,))
+    conn.commit()
+    rows = conn.execute("SELECT COUNT(*) FROM epi_edges").fetchone()[0]
+    conn.close()
+
+    assert rows == 2, "ребро осталось в таблице"
+    assert await g.count_edges("u1") == {"total": 1, "live": 1, "dead": 0}

tests/test_integration.py

@@ -170,6 +170,8 @@ async def test_dashboard():
     assert isinstance(stats, dict)
     assert "l1_buffer" in stats
     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"]
 
 
 @pytest.mark.asyncio

tests/test_lifecycle/test_graph_enrich.py

@@ -195,6 +195,35 @@ async def test_dream_nrem_decays_and_prunes_weak_edges(graph, no_miners):
     assert dream["nrem_decayed"] >= 2 and dream["nrem_pruned"] >= 1, f"NREM-статистика, dream={dream}"
 
 
+@pytest.mark.asyncio
+async def test_dream_nrem_leaves_midlife_edges_alone(graph, no_miners):
+    """An edge between a day and a month old is neither boosted nor decayed.
+
+    This is the band the pass now filters out in SQL instead of reading every
+    heuristic edge and falling through in Python. If the bounds were wrong the
+    edge would come back changed, so the untouched weight is the assertion that
+    narrowing the read kept the same behaviour.
+    """
+    ids = await _dream_graph(graph)
+    conn = await connection_manager.get("memory.db")
+    await _age_edge(conn, ids["a"], ids["b"], days=10)
+    await _age_edge(conn, ids["b"], ids["c"], days=10, weight=0.42)
+    await conn.commit()
+
+    from lifecycle.graph_enrich import graph_enrich
+
+    result = await graph_enrich(layer="user")
+    dream = result["dream"]
+
+    rows = await (await conn.execute("SELECT source_id, target_id, weight FROM epi_edges")).fetchall()
+    weights = {(r["source_id"], r["target_id"]): float(r["weight"]) for r in rows}
+    assert weights[(ids["a"], ids["b"])] == pytest.approx(0.6), "ребро в 10 суток не усилено и не ослаблено"
+    assert weights[(ids["b"], ids["c"])] == pytest.approx(0.42), "ребро в 10 суток не усилено и не ослаблено"
+    # `nrem_decayed` folds in boosts (see the dream result builder), so a zero here
+    # means neither band matched anything.
+    assert dream["nrem_decayed"] == 0 and dream["nrem_pruned"] == 0, f"средний возраст вне компетенции NREM, dream={dream}"
+
+
 @pytest.mark.asyncio
 async def test_dream_rem_bridges_isolated_duplicates(graph, no_miners):
     ids = await _dream_graph(graph)

tests/test_mcp/test_tools_unit.py

@@ -158,9 +158,14 @@ async def test_stats(monkeypatch):
     wiki = app.user_wiki
     wiki.count = AsyncMock(return_value=0)
     graph = app.user_graph
-    graph.count_nodes = AsyncMock(return_value=0)
+    graph.count_nodes = AsyncMock(return_value=7)
+    graph.count_edges = AsyncMock(return_value={"total": 12, "live": 9, "dead": 3})
     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
 
 
 # ── memory_graph_edges direction (B1.1 backlinks) ──