Follow up on recall and topology latency

Marsu6996 · codex · Marsu6996 · commit 625cf7cae640 · 2026-07-06T08:03:40.000+02:00
Remove the macOS launchd Background process policy that throttled native cue embedding in the daemon, and reuse the cue embedding already computed by the ANN dispatch path instead of embedding the same cue twice.

Also reuse runtime-graph structural results for topology so the synchronous topology surface does not discard cached assignment/rich-club data and fall back toward exact in-parent centrality work.

Designed by Marsu — Refined by Codex.

Co-Authored-By: Codex &lt;codex@openai.com&gt;
diff --git a/scripts/com.iai-mcp.daemon.plist.template b/scripts/com.iai-mcp.daemon.plist.template
@@ -33,9 +33,6 @@
   <key>ThrottleInterval</key>
   <integer>5</integer>
 
-  <key>ProcessType</key>
-  <string>Background</string>
-
   <!-- FD floor for a socket-serving daemon. -->
   <key>SoftResourceLimits</key>
   <dict>
diff --git a/src/iai_mcp/_deploy/launchd/com.iai-mcp.daemon.plist b/src/iai_mcp/_deploy/launchd/com.iai-mcp.daemon.plist
@@ -34,9 +34,6 @@
     <key>ThrottleInterval</key>
     <integer>5</integer>
 
-    <key>ProcessType</key>
-    <string>Background</string>
-
     <!--
       Belt-and-suspenders FD floor for a socket-serving daemon.
       The daemon also calls setrlimit(RLIMIT_NOFILE) at boot, but setting
diff --git a/src/iai_mcp/cli/_analytics.py b/src/iai_mcp/cli/_analytics.py
@@ -300,8 +300,10 @@ def _render(d: dict) -> None:
 
     try:
         store = MemoryStore()
-        graph, _assignment, _rich_club = build_runtime_graph(store)
-        snap = compute_topology_snapshot(graph)
+        graph, assignment, rich_club = build_runtime_graph(store)
+        snap = compute_topology_snapshot(
+            graph, assignment=assignment, rich_club=rich_club
+        )
     except HippoLockHeldError:
         _render({})
         return 0
diff --git a/src/iai_mcp/core/__init__.py b/src/iai_mcp/core/__init__.py
@@ -369,6 +369,7 @@ def _mark_recall_stage(name: str) -> None:
                         knobs_applied=knobs_applied,
                         arousal_state=_arousal_diag,
                         tv_maps=(_tv_outgoing_l1, _tv_ts_l1) if _tv_ts_l1 else None,
+                        cue_embedding=_cue_vec,
                     )
                     _mark_recall_stage("recall_for_response")
                     resp.ann_path_used = True
@@ -842,8 +843,19 @@ def _mark_recall_stage(name: str) -> None:
             }
         try:
             graph_bundle = retrieve.build_runtime_graph(store)
-            graph = graph_bundle[0] if isinstance(graph_bundle, tuple) else graph_bundle
-            return sigma_mod.compute_topology_snapshot(graph)
+            assignment = None
+            rich_club = None
+            if isinstance(graph_bundle, tuple):
+                graph = graph_bundle[0]
+                if len(graph_bundle) > 1:
+                    assignment = graph_bundle[1]
+                if len(graph_bundle) > 2:
+                    rich_club = graph_bundle[2]
+            else:
+                graph = graph_bundle
+            return sigma_mod.compute_topology_snapshot(
+                graph, assignment=assignment, rich_club=rich_club
+            )
         except Exception as exc:
             write_event(
                 store,
diff --git a/src/iai_mcp/pipeline.py b/src/iai_mcp/pipeline.py
@@ -450,6 +450,7 @@ def _recall_core(
     spread_hops: int = 2,
     cue_intent: str | None = None,
     contradicts_outgoing: dict[str, list[str]] | None = None,
+    cue_embedding: list[float] | None = None,
 ) -> _RecallCoreResult:
     profile_state = profile_state or {}
 
@@ -517,20 +518,23 @@ def _recall_core(
                 budget_used=budget_used_l0,
             )
 
-    try:
-        cue_emb = embedder.embed(cue)
-    except Exception as exc:
-        write_event(
-            store,
-            TELEMETRY_EMBED_NATIVE_FAILURE,
-            {
-                "op_type": "recall_cue",
-                "backend": "rust",
-                "error_type": type(exc).__name__,
-                "error": str(exc),
-            },
-        )
-        raise NativeError(f"recall cue encode failed: {exc}") from exc
+    if cue_embedding is None:
+        try:
+            cue_emb = embedder.embed(cue)
+        except Exception as exc:
+            write_event(
+                store,
+                TELEMETRY_EMBED_NATIVE_FAILURE,
+                {
+                    "op_type": "recall_cue",
+                    "backend": "rust",
+                    "error_type": type(exc).__name__,
+                    "error": str(exc),
+                },
+            )
+            raise NativeError(f"recall cue encode failed: {exc}") from exc
+    else:
+        cue_emb = cue_embedding
 
     records_cache: dict[UUID, "object"] = {}
     try:
@@ -1155,6 +1159,7 @@ def recall_for_response(
     knobs_applied: dict | None = None,
     arousal_state: dict | None = None,
     tv_maps: "tuple[dict, dict] | None" = None,
+    cue_embedding: list[float] | None = None,
 ) -> RecallResponse:
     import time as _time
     global _last_recall_latency_ms
@@ -1193,6 +1198,7 @@ def recall_for_response(
         spread_hops=_s_hops,
         cue_intent=_cue_intent,
         contradicts_outgoing=_tv_outgoing,
+        cue_embedding=cue_embedding,
     )
 
     derive_temporal_validity(
diff --git a/src/iai_mcp/sigma.py b/src/iai_mcp/sigma.py
@@ -250,15 +250,13 @@ def classify_regime(N: int, sigma: Optional[float]) -> str:
     return "healthy"
 
 
-def compute_topology_snapshot(graph, *, assignment=None) -> dict:
+def compute_topology_snapshot(graph, *, assignment=None, rich_club=None) -> dict:
     """Topology metrics for `graph`.
 
-    Computes community detection in-process by default. The request-synchronous
-    callers (the `topology` health surface, used by `iai status` and the
-    topology tool, plus operator analytics) rely on this near-instant path and
-    pass no `assignment`. Background callers that have already computed the
-    community assignment off the interactive path may pass it in to skip the
-    in-process recompute.
+    Computes community detection in-process by default only for callers that
+    have not already built the runtime graph. Request-synchronous callers should
+    pass the `assignment` and `rich_club` returned by `build_runtime_graph()` so
+    this health surface does not re-run the expensive structural passes.
     """
     from iai_mcp.graph import MemoryGraph
 
@@ -324,7 +322,16 @@ def compute_topology_snapshot(graph, *, assignment=None) -> dict:
         except (RuntimeError, ValueError, TypeError):
             community_count = 0
         try:
-            rc = rich_club_nodes(graph, percent=0.10)
+            if rich_club is None:
+                centrality = {
+                    node_id: float(graph.get_centrality(node_id))
+                    for node_id in graph.iter_nodes()
+                }
+                rc = rich_club_nodes(
+                    graph, percent=0.10, centrality=centrality
+                )
+            else:
+                rc = rich_club
             rich_club_ratio = (len(rc) / N) if N > 0 else 0.0
         except (RuntimeError, ValueError, TypeError):
             rich_club_ratio = 0.0
diff --git a/tests/test_daemon.py b/tests/test_daemon.py
@@ -201,6 +201,7 @@ def test_launchd_plist_valid_xml_with_required_keys():
     assert "SuccessfulExit" not in keepalive
 
     assert data["ThrottleInterval"] == 5
+    assert "ProcessType" not in data
     assert "StandardOutPath" in data
     assert "StandardErrorPath" in data
     assert "WorkingDirectory" in data
diff --git a/tests/test_native_fail_loud.py b/tests/test_native_fail_loud.py
@@ -88,6 +88,52 @@ def test_topology_empty_graph_returns_stub_without_raising(tmp_path):
         f"legitimate path; got {rows}"
     )
 
+def test_topology_passes_cached_structural_results(tmp_path, monkeypatch):
+    from iai_mcp import core, retrieve, sigma as sigma_mod
+
+    store = _make_store(tmp_path)
+    _seed_one_record(store)
+
+    graph = object()
+    assignment = object()
+    rich_club = [object()]
+    captured: dict = {}
+
+    monkeypatch.setattr(
+        retrieve,
+        "build_runtime_graph",
+        lambda _store: (graph, assignment, rich_club),
+    )
+
+    def _fake_snapshot(graph_arg, *, assignment=None, rich_club=None):
+        captured.update(
+            {
+                "graph": graph_arg,
+                "assignment": assignment,
+                "rich_club": rich_club,
+            }
+        )
+        return {
+            "N": 1,
+            "C": 0.0,
+            "L": 0.0,
+            "sigma": None,
+            "community_count": 1,
+            "rich_club_ratio": 1.0,
+            "regime": "insufficient_data",
+        }
+
+    monkeypatch.setattr(sigma_mod, "compute_topology_snapshot", _fake_snapshot)
+
+    result = core.dispatch(store, "topology", {})
+
+    assert result["N"] == 1
+    assert captured == {
+        "graph": graph,
+        "assignment": assignment,
+        "rich_club": rich_club,
+    }
+
 def test_recall_cue_encode_failure_emits_store_event_and_raises(
     tmp_path, monkeypatch
 ):
diff --git a/tests/test_plist_template_lint.py b/tests/test_plist_template_lint.py
@@ -43,13 +43,17 @@ def test_template_has_required_keys() -> None:
         "<true/>",
         "<key>KeepAlive</key>",
         "<key>Crashed</key>",
-        "<key>ProcessType</key>",
         "<key>SoftResourceLimits</key>",
         "IAI_MCP_LAUNCHD_MANAGED",
     ]
     missing = [m for m in required_markers if m not in text]
     assert not missing, f"template missing required markers: {missing}"
 
+def test_template_does_not_run_under_background_policy() -> None:
+    text = TEMPLATE.read_text()
+    assert "<key>ProcessType</key>" not in text
+    assert "<string>Background</string>" not in text
+
 def test_template_has_RunAtLoad_true() -> None:
     text = TEMPLATE.read_text()
     match = re.search(r"<key>RunAtLoad</key>\s*<true/>", text)
diff --git a/tests/test_recall_cue_router.py b/tests/test_recall_cue_router.py
@@ -424,6 +424,39 @@ def fake_recall_for_response(**kwargs):
     assert response["cue_mode"] == "verbatim"
 
 
+def test_dispatch_passes_cue_embedding_to_recall_for_response(tmp_path, monkeypatch):
+    from iai_mcp import core
+    from iai_mcp import embed as _embed_mod
+    from iai_mcp import pipeline as _pipeline_mod
+    from iai_mcp.types import RecallResponse
+
+    store, embedder, _cue, _rec = _seed_populated_store(tmp_path)
+    monkeypatch.setattr(_embed_mod, "embedder_for_store", lambda _store: embedder)
+
+    captured: dict = {}
+
+    def fake_recall_for_response(**kwargs):
+        captured.update(kwargs)
+        return RecallResponse(
+            hits=[], anti_hits=[], activation_trace=[], budget_used=0,
+            cue_mode=kwargs.get("mode", "concept"),
+            patterns_observed=[],
+        )
+
+    monkeypatch.setattr(_pipeline_mod, "recall_for_response", fake_recall_for_response)
+
+    cue = "verbatim recall this exact quote"
+    expected_vec = embedder.embed(cue)
+    embedder.set_fixed(cue, expected_vec)
+
+    core.dispatch(
+        store, "memory_recall",
+        {"cue": cue, "session_id": "cue_embedding_capture"},
+    )
+
+    assert captured["cue_embedding"] == expected_vec
+
+
 def test_dispatch_passes_mode_kwarg_to_retrieve_recall(tmp_path, monkeypatch):
     from iai_mcp import core
     from iai_mcp import retrieve as _retrieve_mod
diff --git a/tests/test_recall_for_response.py b/tests/test_recall_for_response.py
@@ -25,6 +25,12 @@ def embed_batch(self, texts: list[str]) -> list[list[float]]:
         return [list(self._vec) for _ in texts]
 
 
+class _ExplodingEmbedder(_FakeEmbedder):
+
+    def embed(self, text: str) -> list[float]:
+        raise AssertionError("cue embedding should be reused, not recomputed")
+
+
 def _make(
     vec: list[float], text: str = "rec", aaak: str = "", tier: str = "episodic",
 ) -> MemoryRecord:
@@ -130,6 +136,22 @@ def test_recall_for_response_returns_recall_response_type(tmp_path) -> None:
     assert isinstance(resp.patterns_observed, list)
 
 
+def test_recall_for_response_reuses_supplied_cue_embedding(tmp_path) -> None:
+    from iai_mcp.pipeline import recall_for_response
+
+    store, graph, recs = _build_store_and_graph(tmp_path, n=5)
+    assignment = _flat_assignment(recs)
+
+    resp = recall_for_response(
+        store=store, graph=graph, assignment=assignment,
+        rich_club=[], embedder=_ExplodingEmbedder(),
+        cue="test", session_id="s2b",
+        cue_embedding=[1.0] + [0.0] * (EMBED_DIM - 1),
+    )
+
+    assert isinstance(resp, RecallResponse)
+
+
 def test_recall_for_response_packs_under_budget(tmp_path) -> None:
     from iai_mcp.pipeline import recall_for_response
 
diff --git a/tests/test_sigma.py b/tests/test_sigma.py
@@ -31,6 +31,56 @@ def test_compute_sigma_returns_none_below_floor():
     assert compute_sigma(mg) is None
 
 
+def test_topology_snapshot_uses_graph_centrality_payload(monkeypatch):
+    from types import SimpleNamespace
+
+    from iai_mcp.sigma import compute_topology_snapshot
+
+    g = nx.Graph()
+    g.add_edge(0, 1)
+    g.add_edge(1, 2)
+    mg = _nx_graph_to_memory_graph(g)
+    nodes = list(mg.iter_nodes())
+    for i, node_id in enumerate(nodes):
+        mg.set_node_centrality(node_id, float(i))
+
+    def _forbid_exact_centrality():
+        raise AssertionError("topology must not recompute exact centrality")
+
+    monkeypatch.setattr(mg, "centrality", _forbid_exact_centrality)
+    assignment = SimpleNamespace(community_centroids={"c1": [0.0]})
+
+    snap = compute_topology_snapshot(mg, assignment=assignment)
+
+    assert snap["N"] == 3
+    assert snap["community_count"] == 1
+    assert snap["rich_club_ratio"] > 0.0
+
+
+def test_topology_snapshot_reuses_supplied_rich_club(monkeypatch):
+    from types import SimpleNamespace
+
+    from iai_mcp.sigma import compute_topology_snapshot
+
+    g = nx.path_graph(4)
+    mg = _nx_graph_to_memory_graph(g)
+    nodes = list(mg.iter_nodes())
+
+    def _forbid_rich_club(*args, **kwargs):
+        raise AssertionError("topology must reuse supplied rich_club")
+
+    monkeypatch.setattr("iai_mcp.richclub.rich_club_nodes", _forbid_rich_club)
+    assignment = SimpleNamespace(community_centroids={"c1": [0.0]})
+
+    snap = compute_topology_snapshot(
+        mg, assignment=assignment, rich_club=nodes[:2]
+    )
+
+    assert snap["N"] == 4
+    assert snap["community_count"] == 1
+    assert snap["rich_club_ratio"] == 0.5
+
+
 def test_fast_sigma_small_world_above_one_at_n_250():
     from iai_mcp.sigma import fast_sigma
 
diff --git a/tests/test_topology_socket_first.py b/tests/test_topology_socket_first.py
