feat(mcp): add read_live_metrics tool + bump joserfc (Dependabot #25)

CLAUDE.md

@@ -71,9 +71,9 @@ Artist personal information must never appear in commits or public-facing docume
 | `phantom` | `src/phantom/cli/__init__.py` | CLI entry point (click group) |
 | `phantom-mcp` | `src/phantom/server.py` | MCP server entry point |
 
-### MCP Tools (17)
+### MCP Tools (18)
 
-`analyze_spectrum`, `analyze_loudness`, `analyze_dynamics`, `analyze_stereo`, `analyze_phase`, `compare_phase`, `detect_problems`, `analyze_masking`, `analyze_masking_matrix`, `multi_stem_masking`, `compare_to_profile`, `compare_to_reference`, `match_to_reference`, `separate_stems`, `full_diagnostic`, `batch_diagnostic`, `setup_reaper`
+`analyze_spectrum`, `analyze_loudness`, `analyze_dynamics`, `analyze_stereo`, `analyze_phase`, `compare_phase`, `detect_problems`, `analyze_masking`, `analyze_masking_matrix`, `multi_stem_masking`, `compare_to_profile`, `compare_to_reference`, `match_to_reference`, `separate_stems`, `full_diagnostic`, `batch_diagnostic`, `read_live_metrics`, `setup_reaper`
 
 ### CLI Commands
 

src/phantom/live_metrics.py

@@ -0,0 +1,140 @@
+"""Phantom Link live metrics reader.
+
+The Phantom Studio plugin publishes an analysis snapshot as JSON at 2 Hz to
+``<metrics dir>/<instance-uuid>.json`` and deletes the file when the plugin
+instance closes. This module reads those snapshots so Claude can ground
+mixing advice in the numbers currently on the user's meters.
+
+The metrics directory comes from ``PHANTOM_METRICS_DIR`` when set, otherwise
+the platform default matching the plugin's publisher (``~/Library/
+PhantomStudio/metrics`` on macOS). Reads bypass ``load_audio``'s sandbox, so
+this module does its own path containment: the directory is realpath-resolved
+and only regular ``*.json`` files that resolve inside it are eligible.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+import time
+
+from pydantic import BaseModel
+
+from phantom.exceptions import AnalysisError, PathSecurityError
+
+#: Snapshots older than this are flagged stale. The publisher writes at 2 Hz
+#: and removes its file on shutdown, so a stale file usually means the plugin
+#: host crashed or the DAW froze.
+STALE_AFTER_SECONDS = 10.0
+
+_INSTANCE_ID_REGEX = re.compile(r"^[A-Za-z0-9_-]+$")
+
+_NO_METRICS_MSG = (
+    "No live metrics found — is the Phantom Studio plugin running? "
+    "Load it on a track and start playback, then try again."
+)
+_UNREADABLE_MSG = "Live metrics file is unreadable right now — try again in a second."
+
+
+class LiveMetricsResult(BaseModel):
+    """Typed response for the read_live_metrics server tool."""
+
+    instance_id: str
+    age_seconds: float
+    stale: bool
+    instance_count: int
+    snapshot: dict
+
+
+def get_metrics_dir() -> str:
+    """Resolve the metrics directory (PHANTOM_METRICS_DIR or platform default)."""
+    env = os.environ.get("PHANTOM_METRICS_DIR")
+    if env:
+        return env
+    if sys.platform == "darwin":
+        return os.path.expanduser("~/Library/PhantomStudio/metrics")
+    if sys.platform.startswith("win"):
+        base = os.environ.get("APPDATA") or os.path.expanduser("~")
+        return os.path.join(base, "PhantomStudio", "metrics")
+    return os.path.expanduser("~/.config/PhantomStudio/metrics")
+
+
+def _contained_json_files(real_base: str) -> list[str]:
+    """List regular ``*.json`` files whose realpath stays inside the dir.
+
+    The metrics dir only ever holds files the plugin wrote itself, so a
+    symlink resolving elsewhere is not ours to read and is skipped.
+    """
+    files: list[str] = []
+    for name in os.listdir(real_base):
+        if not name.endswith(".json"):
+            continue
+        real = os.path.realpath(os.path.join(real_base, name))
+        if not real.startswith(real_base + os.sep):
+            continue
+        if os.path.isfile(real):
+            files.append(real)
+    return files
+
+
+def read_live_metrics(instance_id: str | None = None) -> LiveMetricsResult:
+    """Read the newest (or a specific instance's) live metrics snapshot.
+
+    Args:
+        instance_id: Optional plugin instance UUID (the metrics file stem).
+            When omitted, the most recently written snapshot is returned.
+
+    Raises:
+        AnalysisError: No snapshot is available or it cannot be parsed.
+        PathSecurityError: ``instance_id`` is malformed or escapes the dir.
+    """
+    real_base = os.path.realpath(get_metrics_dir())
+    if not os.path.isdir(real_base):
+        raise AnalysisError(_NO_METRICS_MSG)
+
+    candidates = _contained_json_files(real_base)
+
+    if instance_id is not None:
+        if not _INSTANCE_ID_REGEX.match(instance_id):
+            raise PathSecurityError(
+                "Invalid instance id: only letters, digits, '-' and '_' are allowed."
+            )
+        wanted = os.path.realpath(os.path.join(real_base, instance_id + ".json"))
+        if not wanted.startswith(real_base + os.sep):
+            raise PathSecurityError(
+                "Access denied: instance id resolves outside the metrics directory."
+            )
+        if wanted not in candidates:
+            raise AnalysisError(
+                f"No live metrics for instance '{instance_id}' — "
+                "is that plugin instance still open?"
+            )
+        chosen = wanted
+    else:
+        if not candidates:
+            raise AnalysisError(_NO_METRICS_MSG)
+        chosen = max(candidates, key=os.path.getmtime)
+
+    try:
+        mtime = os.path.getmtime(chosen)
+        with open(chosen, encoding="utf-8") as fh:
+            snapshot = json.load(fh)
+    except OSError as e:
+        # File vanished between listing and reading (plugin closed).
+        raise AnalysisError(_NO_METRICS_MSG) from e
+    except json.JSONDecodeError as e:
+        raise AnalysisError(_UNREADABLE_MSG) from e
+
+    if not isinstance(snapshot, dict):
+        raise AnalysisError(_UNREADABLE_MSG)
+
+    age = max(0.0, time.time() - mtime)
+    return LiveMetricsResult(
+        instance_id=os.path.splitext(os.path.basename(chosen))[0],
+        age_seconds=round(age, 3),
+        stale=age > STALE_AFTER_SECONDS,
+        instance_count=len(candidates),
+        snapshot=snapshot,
+    )

src/phantom/server.py

@@ -35,6 +35,7 @@
     analyze_masking_matrix as _analyze_masking_matrix,
     MaskingPair,
 )
+from phantom.live_metrics import read_live_metrics as _read_live_metrics
 from phantom._profiles import load_profile as _load_profile
 from phantom._profiles import list_profiles as _list_profiles
 from phantom.comparison import compare_to_profile as _compare_to_profile
@@ -542,6 +543,18 @@ def multi_stem_masking(file_paths: list[str]) -> dict:
     return result.model_dump()
 
 
+# ---------------------------------------------------------------------------
+# Live metrics (Phantom Link)
+# ---------------------------------------------------------------------------
+
+
+@mcp.tool
+@_phantom_tool
+def read_live_metrics(instance_id: str | None = None) -> dict:
+    """Read the Phantom Studio plugin's live meter snapshot (loudness, true peak, stereo, band energy, verdicts). Omit instance_id for the most recent instance."""
+    return _read_live_metrics(instance_id).model_dump()
+
+
 # ---------------------------------------------------------------------------
 # Entry point
 # ---------------------------------------------------------------------------

tests/test_error_schema.py

@@ -91,6 +91,11 @@ def _get_bad_args(tool_name: str) -> dict:
         "batch_diagnostic": {
             "file_paths": [],  # empty list triggers validation error
         },
+        "read_live_metrics": {
+            # Malformed id fails validation before any filesystem access,
+            # so this stays hermetic even if a real metrics dir exists.
+            "instance_id": "../nonexistent",
+        },
     }
 
     if tool_name in mapping:

tests/test_live_metrics.py

@@ -0,0 +1,204 @@
+"""Tests for the read_live_metrics tool (Phantom Link feed).
+
+Covers the live_metrics module directly and the MCP tool surface via
+FastMCP's in-memory Client. All metrics files are synthetic JSON written
+to per-test tmp_path dirs; PHANTOM_METRICS_DIR points at them.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+
+from phantom.exceptions import AnalysisError, PathSecurityError
+from phantom.live_metrics import (
+    STALE_AFTER_SECONDS,
+    get_metrics_dir,
+    read_live_metrics,
+)
+from phantom.server import mcp
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def client():
+    """In-memory MCP client connected to phantom server."""
+    async with Client(mcp) as c:
+        yield c
+
+
+@pytest.fixture
+def metrics_dir(tmp_path, monkeypatch):
+    """Per-test metrics dir wired up via PHANTOM_METRICS_DIR."""
+    d = tmp_path / "metrics"
+    d.mkdir()
+    monkeypatch.setenv("PHANTOM_METRICS_DIR", str(d))
+    return d
+
+
+def _write_snapshot(d, name="aaaa-1111", **overrides):
+    """Write a minimal Phantom Studio snapshot, return its path."""
+    payload = {
+        "schema_version": 1,
+        "plugin": "Phantom Studio",
+        "track": "Mix Bus",
+        "sample_rate": 48000,
+        "metrics": {
+            "lufs_integrated": -14.2,
+            "true_peak_dbtp": -1.1,
+            "correlation": 0.82,
+            "signal_seconds": 12.5,
+        },
+    }
+    payload.update(overrides)
+    path = d / f"{name}.json"
+    path.write_text(json.dumps(payload))
+    return path
+
+
+# ---------------------------------------------------------------------------
+# Module behaviour
+# ---------------------------------------------------------------------------
+
+
+def test_reads_latest_snapshot(metrics_dir):
+    """A fresh snapshot is returned with its instance id and not stale."""
+    _write_snapshot(metrics_dir)
+    result = read_live_metrics()
+    assert result.instance_id == "aaaa-1111"
+    assert result.snapshot["metrics"]["lufs_integrated"] == -14.2
+    assert result.stale is False
+    assert result.instance_count == 1
+    assert result.age_seconds >= 0.0
+
+
+def test_newest_instance_wins(metrics_dir):
+    """With multiple instances, the most recently written file is chosen."""
+    old = _write_snapshot(metrics_dir, name="old-instance", track="Old")
+    _write_snapshot(metrics_dir, name="new-instance", track="New")
+    past = os.path.getmtime(old) - 60
+    os.utime(old, (past, past))
+    result = read_live_metrics()
+    assert result.instance_id == "new-instance"
+    assert result.instance_count == 2
+
+
+def test_specific_instance(metrics_dir):
+    """instance_id selects that instance's file even if another is newer."""
+    _write_snapshot(metrics_dir, name="alpha", track="Alpha")
+    _write_snapshot(metrics_dir, name="beta", track="Beta")
+    result = read_live_metrics("alpha")
+    assert result.instance_id == "alpha"
+    assert result.snapshot["track"] == "Alpha"
+
+
+def test_stale_flag(metrics_dir):
+    """A snapshot older than the threshold is flagged stale."""
+    path = _write_snapshot(metrics_dir)
+    past = os.path.getmtime(path) - (STALE_AFTER_SECONDS + 30)
+    os.utime(path, (past, past))
+    result = read_live_metrics()
+    assert result.stale is True
+    assert result.age_seconds > STALE_AFTER_SECONDS
+
+
+def test_missing_dir(tmp_path, monkeypatch):
+    """A nonexistent metrics dir yields the musician-friendly error."""
+    monkeypatch.setenv("PHANTOM_METRICS_DIR", str(tmp_path / "nope"))
+    with pytest.raises(AnalysisError, match="is the Phantom Studio plugin running"):
+        read_live_metrics()
+
+
+def test_empty_dir(metrics_dir):
+    """An empty metrics dir yields the musician-friendly error."""
+    with pytest.raises(AnalysisError, match="is the Phantom Studio plugin running"):
+        read_live_metrics()
+
+
+def test_unknown_instance(metrics_dir):
+    """Asking for an instance that has no file is a friendly error."""
+    _write_snapshot(metrics_dir)
+    with pytest.raises(AnalysisError, match="is that plugin instance still open"):
+        read_live_metrics("zzzz-9999")
+
+
+def test_traversal_rejected(metrics_dir):
+    """Path separators and dots in instance_id are rejected outright."""
+    _write_snapshot(metrics_dir)
+    for bad in ("../evil", "a/b", "..", ".hidden"):
+        with pytest.raises(PathSecurityError):
+            read_live_metrics(bad)
+
+
+def test_symlink_escape_skipped(metrics_dir, tmp_path):
+    """A symlink pointing outside the metrics dir is never followed."""
+    outside = tmp_path / "outside.json"
+    outside.write_text(json.dumps({"secret": True}))
+    (metrics_dir / "sneaky.json").symlink_to(outside)
+    # Not eligible for newest-file selection...
+    with pytest.raises(AnalysisError, match="is the Phantom Studio plugin running"):
+        read_live_metrics()
+    # ...and not readable by explicit instance id either.
+    with pytest.raises((AnalysisError, PathSecurityError)):
+        read_live_metrics("sneaky")
+
+
+def test_corrupt_json(metrics_dir):
+    """Unparseable JSON yields a retry-flavoured error, not a traceback."""
+    (metrics_dir / "broken.json").write_text("{not json")
+    with pytest.raises(AnalysisError, match="try again"):
+        read_live_metrics()
+
+
+def test_non_object_json(metrics_dir):
+    """A JSON array/scalar payload is rejected the same way."""
+    (metrics_dir / "weird.json").write_text("[1, 2, 3]")
+    with pytest.raises(AnalysisError, match="try again"):
+        read_live_metrics()
+
+
+def test_env_override_beats_default(metrics_dir, monkeypatch):
+    """PHANTOM_METRICS_DIR takes precedence over the platform default."""
+    assert get_metrics_dir() == str(metrics_dir)
+    monkeypatch.delenv("PHANTOM_METRICS_DIR")
+    assert get_metrics_dir() != str(metrics_dir)
+
+
+# ---------------------------------------------------------------------------
+# MCP tool surface
+# ---------------------------------------------------------------------------
+
+
+async def test_tool_via_client(client, metrics_dir):
+    """read_live_metrics is callable over MCP and returns the snapshot."""
+    _write_snapshot(metrics_dir)
+    result = await client.call_tool("read_live_metrics", {})
+    data = result.data
+    assert data["instance_id"] == "aaaa-1111"
+    assert data["snapshot"]["metrics"]["true_peak_dbtp"] == -1.1
+    assert data["stale"] is False
+
+
+async def test_tool_error_via_client(client, tmp_path, monkeypatch):
+    """The no-metrics case surfaces as a structured ToolError."""
+    monkeypatch.setenv("PHANTOM_METRICS_DIR", str(tmp_path / "empty-nope"))
+    with pytest.raises(ToolError) as exc_info:
+        await client.call_tool("read_live_metrics", {})
+    error = json.loads(str(exc_info.value))
+    assert error["error_type"] == "AnalysisError"
+    assert "Phantom Studio plugin" in error["message"]
+
+
+async def test_tool_listed(client):
+    """read_live_metrics appears in the tool listing."""
+    tools = await client.list_tools()
+    assert "read_live_metrics" in {t.name for t in tools}