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 <project root>/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
This commit is contained in:
0xDevNinja
2026-06-29 13:24:48 +05:30
parent 7ee36dd6b6
commit add955696d
2 changed files with 329 additions and 0 deletions

View File

@@ -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"

View File

@@ -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 '<project root>/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),
)