From add955696d17cb4c949826cef4bc91bfd2c1d898 Mon Sep 17 00:00:00 2001 From: 0xDevNinja Date: Mon, 29 Jun 2026 13:24:48 +0530 Subject: [PATCH] feat(audio): add music_library tool so user tracks surface at preflight AGENT_GUIDE.md requires the music decision to be made at the proposal stage, but the only check for the user's music_library/ folder lived in the asset-director skills, which run later. A user could approve a creative direction without ever being told a free, intentional music option was sitting on disk (issue #168). music_library/ was already referenced as a source_tool in asset artifacts but had no backing tool. Add a small read-only tool that scans the library folder (default /music_library, override via MUSIC_LIBRARY_DIR or a library_dir input) and lists the audio tracks it finds, with best-effort durations via ffprobe when present. Because it inherits BaseTool, the registry auto-discovers it and it appears in the preflight provider menu alongside music_gen and the stock music sources: - AVAILABLE when the folder holds at least one audio track - UNAVAILABLE otherwise, with install_instructions telling the user how to add tracks So the user sees their music options before approving creative direction, with no orchestration code changes. Read-only: no side effects, no cost. Closes #168 --- tests/tools/test_music_library.py | 108 +++++++++++++++ tools/audio/music_library.py | 221 ++++++++++++++++++++++++++++++ 2 files changed, 329 insertions(+) create mode 100644 tests/tools/test_music_library.py create mode 100644 tools/audio/music_library.py diff --git a/tests/tools/test_music_library.py b/tests/tools/test_music_library.py new file mode 100644 index 00000000..9f57e1f1 --- /dev/null +++ b/tests/tools/test_music_library.py @@ -0,0 +1,108 @@ +"""Tests for the music_library tool. + +Covers the tool contract, registry discovery, status behavior (folder +present/absent, with/without tracks), and the track listing returned by +execute(). +""" + +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from tools.audio.music_library import MusicLibrary +from tools.base_tool import ToolStatus, ToolTier +from tools.tool_registry import ToolRegistry + + +def _make_track(path: Path, data: bytes = b"\x00\x01\x02\x03") -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + + +def test_contract_metadata(): + tool = MusicLibrary() + info = tool.get_info() + assert info["name"] == "music_library" + assert info["capability"] == "music_library" + assert info["provider"] == "local" + assert info["runtime"] == "local" + assert info["tier"] == ToolTier.SOURCE.value + assert info["resource_profile"]["network_required"] is False + # Read-only tool: no side effects, no cost. + assert tool.side_effects == [] + assert tool.estimate_cost({}) == 0.0 + + +def test_status_unavailable_when_dir_missing(tmp_path, monkeypatch): + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path / "does_not_exist")) + assert MusicLibrary().get_status() == ToolStatus.UNAVAILABLE + + +def test_status_unavailable_when_dir_empty(tmp_path, monkeypatch): + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path)) + assert MusicLibrary().get_status() == ToolStatus.UNAVAILABLE + + +def test_status_available_with_tracks(tmp_path, monkeypatch): + _make_track(tmp_path / "calm_dawn.mp3") + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path)) + assert MusicLibrary().get_status() == ToolStatus.AVAILABLE + + +def test_status_ignores_non_audio_files(tmp_path, monkeypatch): + _make_track(tmp_path / "notes.txt") + _make_track(tmp_path / "cover.jpg") + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path)) + assert MusicLibrary().get_status() == ToolStatus.UNAVAILABLE + + +def test_execute_lists_tracks_sorted(tmp_path, monkeypatch): + _make_track(tmp_path / "zebra.wav") + _make_track(tmp_path / "alpha.mp3") + _make_track(tmp_path / "nested" / "bravo.flac") + _make_track(tmp_path / "ignore.txt") + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path)) + + result = MusicLibrary().execute({}) + assert result.success is True + assert result.data["track_count"] == 3 + names = [t["name"] for t in result.data["tracks"]] + assert names == ["alpha.mp3", "bravo.flac", "zebra.wav"] + assert all(t["size_bytes"] > 0 for t in result.data["tracks"]) + assert result.data["exists"] is True + + +def test_execute_input_dir_overrides_env(tmp_path, monkeypatch): + env_dir = tmp_path / "env" + arg_dir = tmp_path / "arg" + _make_track(env_dir / "env_track.mp3") + _make_track(arg_dir / "arg_track.mp3") + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(env_dir)) + + result = MusicLibrary().execute({"library_dir": str(arg_dir)}) + names = [t["name"] for t in result.data["tracks"]] + assert names == ["arg_track.mp3"] + + +def test_execute_empty_library(tmp_path, monkeypatch): + monkeypatch.setenv("MUSIC_LIBRARY_DIR", str(tmp_path / "missing")) + result = MusicLibrary().execute({}) + assert result.success is True + assert result.data["track_count"] == 0 + assert result.data["tracks"] == [] + assert result.data["exists"] is False + assert result.data["total_duration_seconds"] is None + + +def test_registry_discovers_music_library(): + reg = ToolRegistry() + reg.discover() + assert reg.get("music_library") is not None + # Top-level capability family lookup. + assert reg.get_by_capability("music_library")[0].name == "music_library" + # Granular capability declared in capabilities[]. + assert reg.find_by_capability("list_user_music_tracks")[0].name == "music_library" diff --git a/tools/audio/music_library.py b/tools/audio/music_library.py new file mode 100644 index 00000000..031bd576 --- /dev/null +++ b/tools/audio/music_library.py @@ -0,0 +1,221 @@ +"""User music library — local royalty-free track discovery. + +Surfaces the tracks a user has dropped into ``music_library/`` so the agent can +present them at the *proposal* stage, before creative direction is approved. + +AGENT_GUIDE.md requires the music decision to be made at the proposal stage, but +the only check for ``music_library/`` historically lived in the asset-director +skills, which run later. That meant a user could approve a creative direction +without ever being told a free, intentional music option was sitting on disk. +This read-only tool makes the library a first-class, auto-discovered provider so +it shows up in the preflight provider menu alongside ``music_gen`` and the stock +music sources. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import time +from pathlib import Path +from typing import Any, Optional + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + +# Repository root: tools/audio/music_library.py -> parents[2] +_PROJECT_ROOT = Path(__file__).resolve().parents[2] + +# Common royalty-free audio container extensions a user might drop in. +_AUDIO_EXTENSIONS = { + ".mp3", + ".wav", + ".m4a", + ".aac", + ".flac", + ".ogg", + ".opus", + ".aiff", + ".aif", +} + + +class MusicLibrary(BaseTool): + name = "music_library" + version = "0.1.0" + tier = ToolTier.SOURCE + capability = "music_library" + provider = "local" + stability = ToolStability.PRODUCTION + execution_mode = ExecutionMode.SYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.LOCAL + + dependencies = [] # pure filesystem scan; ffprobe is used opportunistically + install_instructions = ( + "Create a 'music_library/' folder in the project root and drop " + "royalty-free audio tracks into it (e.g. .mp3, .wav, .m4a, .flac, .ogg). " + "Sources: your own files, YouTube Audio Library, Jamendo, Freesound, etc. " + "Override the location with the MUSIC_LIBRARY_DIR environment variable." + ) + + agent_skills = ["music"] + + capabilities = ["list_user_music_tracks"] + supports = { + "local_offline": True, + "free": True, + "duration_when_ffprobe_present": True, + } + best_for = [ + "user-provided, intentional background music", + "free music with no API key or generation cost", + "knowing music options at the proposal stage", + ] + not_good_for = [ + "generating new music (use music_gen / suno_music)", + "searching an external catalog (use freesound_music / pixabay_music)", + ] + + input_schema = { + "type": "object", + "properties": { + "library_dir": { + "type": "string", + "description": ( + "Optional override for the library folder. Defaults to the " + "MUSIC_LIBRARY_DIR env var, then '/music_library'." + ), + }, + }, + } + output_schema = { + "type": "object", + "properties": { + "library_dir": {"type": "string"}, + "exists": {"type": "boolean"}, + "track_count": {"type": "integer"}, + "total_duration_seconds": {"type": ["number", "null"]}, + "tracks": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "path": {"type": "string"}, + "size_bytes": {"type": "integer"}, + "duration_seconds": {"type": ["number", "null"]}, + }, + }, + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=64, vram_mb=0, disk_mb=0, network_required=False + ) + side_effects = [] # read-only + user_visible_verification = [ + "Confirm the listed tracks are the ones you intend to choose from", + ] + + # ---- Library resolution ---- + + def _library_dir(self, inputs: Optional[dict[str, Any]] = None) -> Path: + if inputs and inputs.get("library_dir"): + return Path(inputs["library_dir"]).expanduser() + env_dir = os.environ.get("MUSIC_LIBRARY_DIR") + if env_dir: + return Path(env_dir).expanduser() + return _PROJECT_ROOT / "music_library" + + def _list_tracks(self, library_dir: Path) -> list[Path]: + if not library_dir.is_dir(): + return [] + tracks = [ + p + for p in library_dir.rglob("*") + if p.is_file() and p.suffix.lower() in _AUDIO_EXTENSIONS + ] + return sorted(tracks, key=lambda p: p.as_posix().lower()) + + @staticmethod + def _probe_duration(path: Path) -> Optional[float]: + """Best-effort track duration via ffprobe; None if unavailable.""" + if shutil.which("ffprobe") is None: + return None + try: + out = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(path), + ], + capture_output=True, + text=True, + timeout=15, + check=True, + ) + value = out.stdout.strip() + return round(float(value), 2) if value else None + except (subprocess.SubprocessError, ValueError): + return None + + # ---- Status ---- + + def get_status(self) -> ToolStatus: + return ToolStatus.AVAILABLE if self._list_tracks(self._library_dir()) else ToolStatus.UNAVAILABLE + + # ---- Execution ---- + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + return 1.0 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + start = time.time() + library_dir = self._library_dir(inputs) + track_paths = self._list_tracks(library_dir) + + tracks: list[dict[str, Any]] = [] + total_duration = 0.0 + have_any_duration = False + for path in track_paths: + duration = self._probe_duration(path) + if duration is not None: + have_any_duration = True + total_duration += duration + tracks.append( + { + "name": path.name, + "path": str(path), + "size_bytes": path.stat().st_size, + "duration_seconds": duration, + } + ) + + return ToolResult( + success=True, + data={ + "library_dir": str(library_dir), + "exists": library_dir.is_dir(), + "track_count": len(tracks), + "total_duration_seconds": round(total_duration, 2) if have_any_duration else None, + "tracks": tracks, + }, + duration_seconds=round(time.time() - start, 2), + )