perf(mcp): wire persistent HTTP MCP server + in-memory adata cache + step timing

Eliminates the per-tool-call subprocess spawn that dominated specialist
latency (~44s/step). The HTTP path existed in-tree but had zero call sites.

- GradioAgentUI.__init__ starts the resident `server.py --transport http`
once per container and registers tools via add_mcp_http(url) in _prewarm
and the per-session rebuild; stdio kept as a health-checked fallback.
- Add agent.add_mcp_http / tool_manager.add_mcp_http_server delegating to
mcp_manager.add_mcp_http; fix its discovery to normalize MCP Tool objects
to dicts (was crashing _process_remote_server with 'Tool' has no 'get').
- Add process-lifetime in-memory AnnData cache (cache.read_h5ad_cached),
keyed by path+mtime+size, returns .copy() so it's safe under the shared
resident server / concurrent sessions; wire into hot DE-path reads.
- Add src/core/perf.py [perf] instrumentation on generate/execute and each
MCP tool call so the win is measurable in logs.

Verified locally: 52 tools register over HTTP, tool calls 26-43ms (vs tens
of seconds stdio), resident process count stays 1, zero stdio spawns. Full
suite 947 passed/55 skipped + new tests/test_inmemory_adata_cache.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/core/perf.py

@@ -0,0 +1,49 @@
+"""Lightweight per-step performance instrumentation.
+
+Cheap timing hooks so the latency win from the persistent HTTP MCP server
+(no per-tool-call subprocess spawn) is measurable rather than anecdotal.
+
+Two things are timed:
+  - agent generate vs execute phases (src/agent.py)
+  - each MCP tool call (src/managers/tools/mcp_manager.py)
+
+Each event is printed with a ``[perf]`` prefix (visible in HF Space logs) and
+appended to a bounded module-level ring buffer so a test or a trace dump can
+read the timings back without scraping stdout.
+"""
+from __future__ import annotations
+
+import time
+from collections import deque
+from contextlib import contextmanager
+from typing import Deque, Dict
+
+# Bounded so a long-running container can't grow this unboundedly.
+_EVENTS: Deque[Dict] = deque(maxlen=2000)
+
+
+def record(kind: str, name: str, seconds: float) -> None:
+    """Record a single timed event and echo it to stdout."""
+    seconds = round(seconds, 3)
+    _EVENTS.append({"kind": kind, "name": name, "seconds": seconds, "t": time.time()})
+    print(f"[perf] {kind} {name}: {seconds}s", flush=True)
+
+
+@contextmanager
+def timed(kind: str, name: str):
+    """Context manager that records monotonic wall time of the block."""
+    t0 = time.monotonic()
+    try:
+        yield
+    finally:
+        record(kind, name, time.monotonic() - t0)
+
+
+def get_events() -> list:
+    """Return a snapshot of recorded events (oldest first)."""
+    return list(_EVENTS)
+
+
+def reset() -> None:
+    """Clear recorded events (used by tests)."""
+    _EVENTS.clear()

tests/test_inmemory_adata_cache.py

@@ -0,0 +1,73 @@
+"""Tests for the process-lifetime in-memory AnnData cache (src/cache.py).
+
+These guard the two correctness properties the persistent HTTP server relies on:
+  1. Each read returns an independent copy, so a caller mutating (or writing
+     back) the result cannot corrupt the shared, read-mostly cache — this is
+     what makes one resident server safe to share across concurrent sessions.
+  2. A rewritten file (newer mtime/size) is transparently re-read instead of
+     serving a stale parse.
+"""
+import time
+
+import numpy as np
+import anndata as ad
+import pytest
+
+from src import cache
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache():
+    cache._MEM_CACHE.clear()
+    yield
+    cache._MEM_CACHE.clear()
+
+
+def _write(path, fill):
+    a = ad.AnnData(X=np.full((10, 5), float(fill)))
+    a.write_h5ad(path)
+    return a
+
+
+def test_second_read_hits_cache_single_entry(tmp_path):
+    p = tmp_path / "x.h5ad"
+    _write(p, 1)
+    cache.read_h5ad_cached(str(p))
+    cache.read_h5ad_cached(str(p))
+    assert len(cache._MEM_CACHE) == 1  # one entry, reused
+
+
+def test_returns_independent_copies(tmp_path):
+    p = tmp_path / "x.h5ad"
+    _write(p, 1)
+    a = cache.read_h5ad_cached(str(p))
+    b = cache.read_h5ad_cached(str(p))
+    assert a is not b
+    # Mutating one copy must not affect the cache or other copies.
+    a.X[:] = 999
+    c = cache.read_h5ad_cached(str(p))
+    assert float(c.X.mean()) == 1.0
+
+
+def test_rewrite_invalidates_stale_entry(tmp_path):
+    p = tmp_path / "x.h5ad"
+    _write(p, 1)
+    assert float(cache.read_h5ad_cached(str(p)).X.mean()) == 1.0
+    time.sleep(1.1)  # ensure mtime advances at 1s resolution
+    _write(p, 7)
+    assert float(cache.read_h5ad_cached(str(p)).X.mean()) == 7.0
+    # Stale entry for the same path is evicted, not accumulated.
+    assert len(cache._MEM_CACHE) == 1
+
+
+def test_missing_path_falls_through(tmp_path):
+    with pytest.raises(Exception):
+        cache.read_h5ad_cached(str(tmp_path / "nope.h5ad"))
+
+
+def test_eviction_bounds_entries(tmp_path):
+    for i in range(cache._MEM_MAX_ENTRIES + 3):
+        p = tmp_path / f"d{i}.h5ad"
+        _write(p, i)
+        cache.read_h5ad_cached(str(p))
+    assert len(cache._MEM_CACHE) <= cache._MEM_MAX_ENTRIES