#802: on-demand memory-growth diagnostic (tracemalloc, browser-drivable)

A user reports ~0.7 MiB/s RSS growth; the one theory offered so far
(connection leak) was debunked, so instead of guessing: measure. New
core/diagnostics/memory_tracker wraps tracemalloc behind three GET endpoints
the user can drive from a browser:

  /api/debug/memory/start   begin tracing + baseline snapshot (idempotent)
  /api/debug/memory/report  top allocation sites by GROWTH since the baseline
                            (?top=N), with traced totals + process RSS so we
                            can see how much of the real growth tracing
                            accounts for; 15-frame tracebacks name the caller
  /api/debug/memory/stop    end tracing, free trace bookkeeping

Opt-in by design — tracemalloc shadows every allocation while active, so it
never runs by default. RSS via psutil with a /proc fallback.

Tests: report-without-tracking returns a hint (not an error); a real
start->hog->report->stop roundtrip attributes a genuine 5MB allocation to the
test file (fun fact encoded in the test: 'x'*1000 constant-folds into ONE
shared string and traces as ~40KB — the hog must allocate at runtime); the
stat formatter is duck-typed and unit-tested.

core/diagnostics/memory_tracker.py

@@ -0,0 +1,136 @@
+"""On-demand memory-growth diagnostic (issue #802: ~0.7 MiB/s RSS growth).
+
+Wraps ``tracemalloc`` so a user seeing runaway memory can capture WHERE the
+allocations come from instead of us guessing:
+
+    1. start_tracking()  — begins tracing + stores a baseline snapshot
+    2. ...reproduce the growth for a few minutes...
+    3. report()          — top allocation sites, with the DELTA since baseline
+                           (the delta is the leak; absolute sizes are mostly
+                           startup noise)
+    4. stop_tracking()   — ends tracing, frees trace memory
+
+Opt-in by design: tracemalloc costs CPU and memory while active (it shadows
+every allocation), so it must never run by default. The Flask endpoints that
+expose this live in web_server (GET /api/debug/memory/...) so a user can drive
+the whole flow from a browser.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import tracemalloc
+from typing import Any, Dict, List, Optional
+
+from utils.logging_config import get_logger
+
+logger = get_logger("diagnostics.memory")
+
+_baseline: Optional[tracemalloc.Snapshot] = None
+_started_at: Optional[float] = None
+
+# Allocation-site traces this deep give useful "who called it" context without
+# pathological overhead.
+_TRACE_FRAMES = 15
+
+
+def is_tracking() -> bool:
+    return tracemalloc.is_tracing()
+
+
+def start_tracking() -> Dict[str, Any]:
+    """Begin tracing and store the baseline snapshot. Idempotent."""
+    global _baseline, _started_at
+    if tracemalloc.is_tracing():
+        return {"tracking": True, "already_running": True, "started_at": _started_at}
+    tracemalloc.start(_TRACE_FRAMES)
+    _baseline = tracemalloc.take_snapshot()
+    _started_at = time.time()
+    logger.info("Memory tracking started (tracemalloc, %d frames)", _TRACE_FRAMES)
+    return {"tracking": True, "already_running": False, "started_at": _started_at}
+
+
+def stop_tracking() -> Dict[str, Any]:
+    """End tracing and free the trace bookkeeping."""
+    global _baseline, _started_at
+    was = tracemalloc.is_tracing()
+    if was:
+        tracemalloc.stop()
+        logger.info("Memory tracking stopped")
+    _baseline = None
+    _started_at = None
+    return {"tracking": False, "was_tracking": was}
+
+
+def _rss_mb() -> Optional[float]:
+    """Process RSS in MiB, best-effort (psutil, then /proc fallback)."""
+    try:
+        import psutil
+        return round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 1)
+    except Exception:  # noqa: S110 — RSS is optional context; fall through to /proc
+        pass
+    try:
+        with open("/proc/self/status", encoding="utf-8") as fh:
+            for line in fh:
+                if line.startswith("VmRSS:"):
+                    return round(int(line.split()[1]) / 1024, 1)
+    except Exception:  # noqa: S110 — no /proc on this platform; RSS stays None
+        pass
+    return None
+
+
+def format_stat(stat: Any) -> Dict[str, Any]:
+    """Project one tracemalloc StatisticDiff/Statistic into a plain dict.
+    Duck-typed (reads size/count/size_diff/count_diff/traceback) so it's
+    unit-testable without real snapshots."""
+    tb = getattr(stat, "traceback", None)
+    frames: List[str] = []
+    if tb:
+        # Most-recent-call-last reads naturally top-down in a report.
+        for frame in list(tb)[-3:]:
+            frames.append(f"{frame.filename}:{frame.lineno}")
+    return {
+        "location": frames[-1] if frames else "?",
+        "trace": frames,
+        "size_mb": round(getattr(stat, "size", 0) / (1024 * 1024), 3),
+        "size_diff_mb": round(getattr(stat, "size_diff", 0) / (1024 * 1024), 3),
+        "count": getattr(stat, "count", 0),
+        "count_diff": getattr(stat, "count_diff", 0),
+    }
+
+
+def report(top: int = 25) -> Dict[str, Any]:
+    """Current snapshot vs the start_tracking() baseline: the top allocation
+    sites by GROWTH (size_diff). Includes traced totals + process RSS so the
+    user can see how much of the real growth tracemalloc accounts for."""
+    if not tracemalloc.is_tracing():
+        return {
+            "tracking": False,
+            "rss_mb": _rss_mb(),
+            "hint": "Start with /api/debug/memory/start, reproduce the growth "
+                    "for a few minutes, then call this again.",
+        }
+    snapshot = tracemalloc.take_snapshot()
+    # Filter the tracer's own bookkeeping out of the picture.
+    snapshot = snapshot.filter_traces((
+        tracemalloc.Filter(False, tracemalloc.__file__),
+        tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
+    ))
+    current, peak = tracemalloc.get_traced_memory()
+
+    if _baseline is not None:
+        stats = snapshot.compare_to(_baseline, "traceback")
+        stats.sort(key=lambda s: s.size_diff, reverse=True)
+    else:
+        stats = snapshot.statistics("traceback")
+
+    return {
+        "tracking": True,
+        "started_at": _started_at,
+        "elapsed_seconds": round(time.time() - _started_at, 1) if _started_at else None,
+        "traced_current_mb": round(current / (1024 * 1024), 1),
+        "traced_peak_mb": round(peak / (1024 * 1024), 1),
+        "rss_mb": _rss_mb(),
+        "top_growth": [format_stat(s) for s in stats[:top]],
+    }

tests/test_memory_tracker.py

@@ -0,0 +1,60 @@
+"""Seam tests for the #802 memory-growth diagnostic (core/diagnostics)."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import core.diagnostics.memory_tracker as mt
+
+
+def teardown_function(_fn):
+    # Never leave tracemalloc running across tests — it shadows every
+    # allocation in the process.
+    mt.stop_tracking()
+
+
+def test_report_without_tracking_is_a_hint_not_an_error():
+    mt.stop_tracking()
+    out = mt.report()
+    assert out['tracking'] is False
+    assert 'start' in out['hint']
+
+
+def test_start_report_stop_roundtrip_captures_growth():
+    assert mt.start_tracking()['tracking'] is True
+    # Idempotent start
+    assert mt.start_tracking()['already_running'] is True
+
+    # Allocate something measurable after the baseline.
+    # bytearray(1000) allocates at RUNTIME — a constant expression like
+    # 'x' * 1000 gets folded into ONE shared string and traces as ~40KB.
+    hog = [bytearray(1000) for _ in range(5000)]  # ~5 MB, genuinely allocated
+
+    out = mt.report(top=10)
+    assert out['tracking'] is True
+    assert out['elapsed_seconds'] is not None
+    assert out['traced_current_mb'] > 0
+    assert isinstance(out['top_growth'], list) and out['top_growth']
+    # The hog must show up as growth attributed to THIS file.
+    top_locations = ' '.join(s['location'] for s in out['top_growth'])
+    assert 'test_memory_tracker.py' in top_locations
+    assert any(s['size_diff_mb'] > 1 for s in out['top_growth'])
+    del hog
+
+    stopped = mt.stop_tracking()
+    assert stopped == {'tracking': False, 'was_tracking': True}
+    assert mt.is_tracking() is False
+
+
+def test_format_stat_projects_duck_typed_stat():
+    frame = SimpleNamespace(filename='core/foo.py', lineno=42)
+    stat = SimpleNamespace(
+        size=2 * 1024 * 1024, size_diff=1024 * 1024,
+        count=10, count_diff=4,
+        traceback=[frame, SimpleNamespace(filename='core/bar.py', lineno=7)],
+    )
+    out = mt.format_stat(stat)
+    assert out['location'] == 'core/bar.py:7'
+    assert out['trace'] == ['core/foo.py:42', 'core/bar.py:7']
+    assert out['size_mb'] == 2.0 and out['size_diff_mb'] == 1.0
+    assert out['count'] == 10 and out['count_diff'] == 4

web_server.py

@@ -2785,6 +2785,43 @@ def get_debug_info():
     return _debug_info_get()
 
 
+# ── Memory-growth diagnostic (#802) ──
+# Opt-in tracemalloc capture, drivable entirely from a browser:
+#   /api/debug/memory/start  -> begin tracing (baseline snapshot)
+#   ...reproduce the growth for a few minutes...
+#   /api/debug/memory/report -> top allocation sites by GROWTH since baseline
+#   /api/debug/memory/stop   -> end tracing, free the trace bookkeeping
+# GET on purpose so a user can paste URLs; tracing costs CPU+memory while
+# active, which is why it never runs by default.
+
+@app.route('/api/debug/memory/start')
+def debug_memory_start():
+    try:
+        from core.diagnostics.memory_tracker import start_tracking
+        return jsonify(start_tracking())
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/api/debug/memory/report')
+def debug_memory_report():
+    try:
+        from core.diagnostics.memory_tracker import report
+        top = request.args.get('top', 25, type=int)
+        return jsonify(report(top=max(1, min(top, 100))))
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/api/debug/memory/stop')
+def debug_memory_stop():
+    try:
+        from core.diagnostics.memory_tracker import stop_tracking
+        return jsonify(stop_tracking())
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
 @app.route('/api/activity/feed')
 def get_activity_feed():
     """Get recent activity feed for dashboard"""