diff --git a/CLAUDE.md b/CLAUDE.md index 5155175..e9a3c10 100644 --- a/CLAUDE.md +++ b/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 diff --git a/src/phantom/live_metrics.py b/src/phantom/live_metrics.py new file mode 100644 index 0000000..b0ad65f --- /dev/null +++ b/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 +``/.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, + ) diff --git a/src/phantom/server.py b/src/phantom/server.py index f2c3cb9..c99d9d7 100644 --- a/src/phantom/server.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/test_error_schema.py b/tests/test_error_schema.py index fcbc990..b190aec 100644 --- a/tests/test_error_schema.py +++ b/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: diff --git a/tests/test_live_metrics.py b/tests/test_live_metrics.py new file mode 100644 index 0000000..653f3b3 --- /dev/null +++ b/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} diff --git a/uv.lock b/uv.lock index 14a4e19..d9abf69 100644 --- a/uv.lock +++ b/uv.lock @@ -908,14 +908,14 @@ wheels = [ [[package]] name = "joserfc" -version = "1.6.4" +version = "1.7.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "cryptography" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/de/c6/de8fdbdfa75c8ca04fead38a82d573df8a82906e984c349d58665f459558/joserfc-1.6.4.tar.gz", hash = "sha256:34ce5f499bfcc5e9ad4cc75077f9278ab3227b71da9aaf28f9ab705f8a560d3c", size = 231866, upload-time = "2026-04-13T13:15:40.632Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f1/26/abe1ad855eb334b5ebc9c6495d4798e12bee70e5e8e815d54570710b8312/joserfc-1.7.2.tar.gz", hash = "sha256:537ffb8888b2df039cb5b6d017d7cff6f09d521ce65d89cc9b8ab752b1cff947", size = 233183, upload-time = "2026-06-29T09:03:10.868Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b6/f7/210b27752e972edb36d239315b08d3eb6b14824cc4a590da2337d195260b/joserfc-1.6.4-py3-none-any.whl", hash = "sha256:3e4a22b509b41908989237a045e25c8308d5fd47ab96bdae2dd8057c6451003a", size = 70464, upload-time = "2026-04-13T13:15:39.259Z" }, + { url = "https://files.pythonhosted.org/packages/13/80/d1b30336582cced4dce0dae776508a6011723e32f907bc7a702c0b25890a/joserfc-1.7.2-py3-none-any.whl", hash = "sha256:ddd818c0ca9b4f17bbc2d72cb3966e6ded7502be089316c62c3cc64ae86132b5", size = 70426, upload-time = "2026-06-29T09:03:09.393Z" }, ] [[package]]