From 131f4be7b3f1d8633fce7a42008230587b875a90 Mon Sep 17 00:00:00 2001 From: Patrick Nilan Date: Sun, 8 Mar 2026 20:27:31 -0700 Subject: [PATCH 1/4] =?UTF-8?q?Phase=203:=20Patch=20library=20=E2=80=94=20?= =?UTF-8?q?SQLite-backed=20patch=20storage,=20tools,=20and=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add PatchLibrary with SQLite storage for saving, loading, recalling, listing, and deleting named synth patches. Recall sends all saved CC values back to hardware to restore a sound. New files: - patchwork/patch_library.py: PatchLibrary class, Patch dataclass - patchwork/tools/patches.py: save/load/recall/list/delete patch tools - tests/test_patch_library.py: 13 tests for database operations - tests/test_patch_tools.py: 18 tests for tool functions - data/.gitkeep: track data directory for runtime DB Modified: - patchwork/deps.py: add patches field to PatchworkDeps - patchwork/agent.py: register patch tools, update system prompt - patchwork/cli.py: initialize and close PatchLibrary - tests/test_midi_control_tools.py: update helper for new deps field - .gitignore: ignore data/*.db instead of entire data/ --- .gitignore | 2 +- data/.gitkeep | 0 patchwork/agent.py | 27 +++- patchwork/cli.py | 6 +- patchwork/deps.py | 2 + patchwork/patch_library.py | 134 ++++++++++++++++ patchwork/tools/patches.py | 161 +++++++++++++++++++ tests/test_midi_control_tools.py | 5 +- tests/test_patch_library.py | 145 +++++++++++++++++ tests/test_patch_tools.py | 260 +++++++++++++++++++++++++++++++ 10 files changed, 738 insertions(+), 4 deletions(-) create mode 100644 data/.gitkeep create mode 100644 patchwork/patch_library.py create mode 100644 patchwork/tools/patches.py create mode 100644 tests/test_patch_library.py create mode 100644 tests/test_patch_tools.py diff --git a/.gitignore b/.gitignore index 8ec5ac3..b2ac8ef 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,4 @@ wheels/ .env # Data -data/ +data/*.db diff --git a/data/.gitkeep b/data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/patchwork/agent.py b/patchwork/agent.py index f1c9e21..9701974 100644 --- a/patchwork/agent.py +++ b/patchwork/agent.py @@ -8,6 +8,13 @@ send_cc, send_patch, ) +from patchwork.tools.patches import ( + delete_patch, + list_patches, + load_patch, + recall_patch, + save_patch, +) SYSTEM_PROMPT = """You are Patchwork, an expert synthesizer sound design assistant. @@ -32,6 +39,16 @@ - send_cc: Send a single MIDI CC value to a synth parameter - send_patch: Send multiple MIDI CC values at once to set a full patch +You also have tools to manage a patch library: +- save_patch: Save the current CC values as a named patch +- load_patch: View a saved patch's settings (does not send to hardware) +- recall_patch: Load a patch AND send all its CC values to the synth +- list_patches: List saved patches, optionally filtered by synth +- delete_patch: Remove a saved patch + +Workflow: after sending CC values to a synth and the user likes the sound, +save it as a named patch. Later, recall it to restore the exact same settings. + IMPORTANT: When the user asks you to do something that a tool can handle, ALWAYS call the tool immediately. Never respond with "let me check" or "sure" without actually calling the tool. Specifically: @@ -40,6 +57,11 @@ - "connect" → call connect_midi - "set [param] to [value]" → call send_cc - Any request to dial in a patch → call send_patch +- "save this/that patch" → call save_patch with the CC values that were just sent +- "load/show patch X" → call load_patch +- "recall patch X" → call recall_patch +- "list patches" → call list_patches +- "delete patch X" → call delete_patch After a tool call, report the results to the user. Tone: conversational but concise. Use musical and technical terminology naturally. @@ -51,7 +73,10 @@ system_prompt=SYSTEM_PROMPT, deps_type=PatchworkDeps, defer_model_check=True, - tools=[list_midi_ports, connect_midi, list_synths, send_cc, send_patch], + tools=[ + list_midi_ports, connect_midi, list_synths, send_cc, send_patch, + save_patch, load_patch, recall_patch, list_patches, delete_patch, + ], ) diff --git a/patchwork/cli.py b/patchwork/cli.py index 5bf560d..4111dc1 100644 --- a/patchwork/cli.py +++ b/patchwork/cli.py @@ -6,6 +6,7 @@ from patchwork.agent import agent from patchwork.deps import PatchworkDeps from patchwork.midi import MidiConnection +from patchwork.patch_library import PatchLibrary from patchwork.synth_definitions import load_synth_definitions console = Console() @@ -13,8 +14,10 @@ async def main(): midi = MidiConnection() + patches = PatchLibrary() + patches.open() synths = load_synth_definitions() - deps = PatchworkDeps(midi=midi, synths=synths) + deps = PatchworkDeps(midi=midi, synths=synths, patches=patches) console.print("[bold]patchwork[/bold] — synth research agent\n") if synths: @@ -52,6 +55,7 @@ async def main(): console.print(f"\n[bold red]error:[/bold red] {e}") finally: midi.close() + patches.close() def main_cli(): diff --git a/patchwork/deps.py b/patchwork/deps.py index bb5ccc8..20c9cd8 100644 --- a/patchwork/deps.py +++ b/patchwork/deps.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from patchwork.midi import MidiConnection +from patchwork.patch_library import PatchLibrary from patchwork.synth_definitions import SynthDefinition @@ -8,3 +9,4 @@ class PatchworkDeps: midi: MidiConnection synths: dict[str, SynthDefinition] + patches: PatchLibrary diff --git a/patchwork/patch_library.py b/patchwork/patch_library.py new file mode 100644 index 0000000..d239e7e --- /dev/null +++ b/patchwork/patch_library.py @@ -0,0 +1,134 @@ +import json +import sqlite3 +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +_DEFAULT_DB_PATH = Path(__file__).resolve().parent.parent / "data" / "patches.db" + + +@dataclass +class Patch: + """A saved patch retrieved from the database.""" + + id: int + name: str + synth: str + description: str | None + settings: dict[str, int] + created_at: datetime + updated_at: datetime + + +class PatchLibrary: + """SQLite-backed storage for synth patches.""" + + def __init__(self, db_path: Path = _DEFAULT_DB_PATH): + self._db_path = db_path + self._conn: sqlite3.Connection | None = None + + def __enter__(self): + self.open() + return self + + def __exit__(self, *exc): + self.close() + + def open(self): + """Open the database connection and ensure the schema exists.""" + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(str(self._db_path)) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA journal_mode=WAL") + self._create_schema() + + def close(self): + """Close the database connection.""" + if self._conn is not None: + self._conn.close() + self._conn = None + + @property + def _db(self) -> sqlite3.Connection: + if self._conn is None: + raise RuntimeError("Database not open — call open() first") + return self._conn + + def _create_schema(self): + self._db.execute(""" + CREATE TABLE IF NOT EXISTS patches ( + id INTEGER PRIMARY KEY, + name TEXT UNIQUE NOT NULL, + synth TEXT NOT NULL, + description TEXT, + settings TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + self._db.commit() + + def save( + self, + name: str, + synth: str, + settings: dict[str, int], + description: str | None = None, + ) -> Patch: + """Save a patch. If a patch with this name exists, update it.""" + settings_json = json.dumps(settings) + now = datetime.now().isoformat() + try: + self._db.execute( + """INSERT INTO patches (name, synth, description, settings, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (name, synth, description, settings_json, now, now), + ) + except sqlite3.IntegrityError: + self._db.execute( + """UPDATE patches SET synth=?, description=?, settings=?, updated_at=? + WHERE name=?""", + (synth, description, settings_json, now, name), + ) + self._db.commit() + return self.get(name) # type: ignore[return-value] + + def get(self, name: str) -> Patch | None: + """Get a patch by name. Returns None if not found.""" + row = self._db.execute( + "SELECT * FROM patches WHERE name = ?", (name,) + ).fetchone() + if row is None: + return None + return self._row_to_patch(row) + + def list(self, synth: str | None = None) -> list[Patch]: + """List all patches, optionally filtered by synth name.""" + if synth: + rows = self._db.execute( + "SELECT * FROM patches WHERE LOWER(synth) = ? ORDER BY updated_at DESC", + (synth.lower(),), + ).fetchall() + else: + rows = self._db.execute( + "SELECT * FROM patches ORDER BY updated_at DESC" + ).fetchall() + return [self._row_to_patch(row) for row in rows] + + def delete(self, name: str) -> bool: + """Delete a patch by name. Returns True if a patch was deleted.""" + cursor = self._db.execute("DELETE FROM patches WHERE name = ?", (name,)) + self._db.commit() + return cursor.rowcount > 0 + + @staticmethod + def _row_to_patch(row: sqlite3.Row) -> Patch: + return Patch( + id=row["id"], + name=row["name"], + synth=row["synth"], + description=row["description"], + settings=json.loads(row["settings"]), + created_at=datetime.fromisoformat(row["created_at"]), + updated_at=datetime.fromisoformat(row["updated_at"]), + ) diff --git a/patchwork/tools/patches.py b/patchwork/tools/patches.py new file mode 100644 index 0000000..fe9e415 --- /dev/null +++ b/patchwork/tools/patches.py @@ -0,0 +1,161 @@ +from pydantic_ai import RunContext + +from patchwork.deps import PatchworkDeps + + +async def save_patch( + ctx: RunContext[PatchworkDeps], + name: str, + synth: str, + settings: dict[str, int], + description: str = "", +) -> str: + """Save a patch to the library. If a patch with this name already exists, it will be updated. + + Args: + name: Name for the patch (e.g. "dark-techno-bass") + synth: Synth name this patch is for (e.g. "minitaur", "tb-03") + settings: Dict of parameter names to CC values (e.g. {"filter_cutoff": 45, "resonance": 80}) + description: Optional description of the sound + """ + synth_key = synth.lower() + synth_def = ctx.deps.synths.get(synth_key) + if synth_def is None: + available = ", ".join(ctx.deps.synths.keys()) + return f"Unknown synth '{synth}'. Available: {available}" + + # Validate parameter names and normalize keys to canonical form (lowercase, underscored) + normalized_settings: dict[str, int] = {} + invalid_params = [] + for param_name, value in settings.items(): + param_key = param_name.lower().replace(" ", "_") + if param_key not in synth_def.cc_map: + invalid_params.append(param_name) + else: + normalized_settings[param_key] = value + if invalid_params: + available = ", ".join(synth_def.cc_map.keys()) + return ( + f"Unknown parameter(s) for {synth_def.name}: {', '.join(invalid_params)}. " + f"Available: {available}" + ) + + patch = ctx.deps.patches.save( + name=name, + synth=synth_key, + settings=normalized_settings, + description=description or None, + ) + param_lines = [f" {k} = {v}" for k, v in patch.settings.items()] + return ( + f"Saved patch '{patch.name}' for {synth_def.name}:\n" + + "\n".join(param_lines) + ) + + +async def load_patch( + ctx: RunContext[PatchworkDeps], + name: str, +) -> str: + """Load a patch from the library and display its settings (does NOT send to hardware). + + Args: + name: Name of the patch to load + """ + patch = ctx.deps.patches.get(name) + if patch is None: + return f"No patch found with name '{name}'." + + synth_def = ctx.deps.synths.get(patch.synth) + synth_display = synth_def.name if synth_def else patch.synth + + lines = [f"Patch '{patch.name}' for {synth_display}:"] + if patch.description: + lines.append(f" Description: {patch.description}") + for param, value in patch.settings.items(): + lines.append(f" {param} = {value}") + lines.append(f" Saved: {patch.created_at:%Y-%m-%d %H:%M}") + return "\n".join(lines) + + +async def recall_patch( + ctx: RunContext[PatchworkDeps], + name: str, +) -> str: + """Load a patch from the library AND send all its CC values to the hardware synth. + + Args: + name: Name of the patch to recall + """ + patch = ctx.deps.patches.get(name) + if patch is None: + return f"No patch found with name '{name}'." + + synth_def = ctx.deps.synths.get(patch.synth) + if synth_def is None: + return ( + f"Patch '{name}' is for synth '{patch.synth}', " + f"but no definition is loaded for that synth." + ) + + if not ctx.deps.midi.is_connected: + return ( + "MIDI not connected. Use list_midi_ports to see available ports, " + "then ask me to connect." + ) + + results = [] + for param_name, value in patch.settings.items(): + param = synth_def.cc_map.get(param_name) + if param is None: + results.append(f" Skipped '{param_name}' (not in current CC map)") + continue + ctx.deps.midi.send_cc(synth_def.midi_channel, param.cc, value) + results.append(f" {param_name} = {value} (CC {param.cc})") + + return ( + f"Recalled patch '{patch.name}' to {synth_def.name} " + f"ch.{synth_def.midi_channel}:\n" + "\n".join(results) + ) + + +async def list_patches( + ctx: RunContext[PatchworkDeps], + synth: str | None = None, +) -> str: + """List saved patches, optionally filtered by synth. + + Args: + synth: Optional synth name to filter by (e.g. "minitaur"). If omitted, lists all patches. + """ + patches = ctx.deps.patches.list(synth=synth) + if not patches: + if synth: + return f"No patches saved for '{synth}'." + return "No patches saved yet." + + lines = [] + for patch in patches: + synth_def = ctx.deps.synths.get(patch.synth) + synth_display = synth_def.name if synth_def else patch.synth + desc = f" — {patch.description}" if patch.description else "" + param_count = len(patch.settings) + lines.append(f" {patch.name} ({synth_display}, {param_count} params){desc}") + + header = f"Saved patches ({len(patches)}):" + return header + "\n" + "\n".join(lines) + + +async def delete_patch( + ctx: RunContext[PatchworkDeps], + name: str, +) -> str: + """Delete a patch from the library. + + Args: + name: Name of the patch to delete + """ + deleted = ctx.deps.patches.delete(name) + if deleted: + return f"Deleted patch '{name}'." + return f"No patch found with name '{name}'." diff --git a/tests/test_midi_control_tools.py b/tests/test_midi_control_tools.py index 63a3635..db37a48 100644 --- a/tests/test_midi_control_tools.py +++ b/tests/test_midi_control_tools.py @@ -32,13 +32,16 @@ def _make_synth() -> SynthDefinition: def _make_ctx( midi: MidiConnection | None = None, synths: dict[str, SynthDefinition] | None = None, + patches=None, ) -> RunContext[PatchworkDeps]: if midi is None: midi = MidiConnection() if synths is None: synth = _make_synth() synths = {synth.name.lower(): synth} - deps = PatchworkDeps(midi=midi, synths=synths) + if patches is None: + patches = MagicMock() + deps = PatchworkDeps(midi=midi, synths=synths, patches=patches) return RunContext(deps=deps, model=MagicMock(), usage=RunUsage()) diff --git a/tests/test_patch_library.py b/tests/test_patch_library.py new file mode 100644 index 0000000..3543fb3 --- /dev/null +++ b/tests/test_patch_library.py @@ -0,0 +1,145 @@ +import time + +import pytest + +from patchwork.patch_library import PatchLibrary + + +@pytest.fixture +def patch_lib(tmp_path): + with PatchLibrary(db_path=tmp_path / "test.db") as lib: + yield lib + + +def test_save_and_get(patch_lib): + patch = patch_lib.save( + name="dark-bass", + synth="minitaur", + settings={"cutoff": 45, "resonance": 80}, + description="A dark bass patch", + ) + assert patch.name == "dark-bass" + assert patch.synth == "minitaur" + assert patch.settings == {"cutoff": 45, "resonance": 80} + assert patch.description == "A dark bass patch" + + loaded = patch_lib.get("dark-bass") + assert loaded is not None + assert loaded.name == patch.name + assert loaded.synth == patch.synth + assert loaded.settings == patch.settings + assert loaded.description == patch.description + + +def test_save_overwrites_existing(patch_lib): + patch1 = patch_lib.save( + name="bass", + synth="minitaur", + settings={"cutoff": 45}, + ) + time.sleep(0.01) + patch2 = patch_lib.save( + name="bass", + synth="minitaur", + settings={"cutoff": 90, "resonance": 50}, + description="updated", + ) + assert patch2.name == "bass" + assert patch2.settings == {"cutoff": 90, "resonance": 50} + assert patch2.description == "updated" + assert patch2.updated_at > patch1.updated_at + + +def test_get_nonexistent(patch_lib): + assert patch_lib.get("nonexistent") is None + + +def test_list_all(patch_lib): + patch_lib.save(name="a", synth="minitaur", settings={"cutoff": 10}) + time.sleep(0.01) + patch_lib.save(name="b", synth="tb03", settings={"cutoff": 20}) + time.sleep(0.01) + patch_lib.save(name="c", synth="minitaur", settings={"cutoff": 30}) + + patches = patch_lib.list() + assert len(patches) == 3 + # Most recently updated first + assert patches[0].name == "c" + assert patches[1].name == "b" + assert patches[2].name == "a" + + +def test_list_filtered_by_synth(patch_lib): + patch_lib.save(name="a", synth="minitaur", settings={"cutoff": 10}) + patch_lib.save(name="b", synth="tb03", settings={"cutoff": 20}) + patch_lib.save(name="c", synth="minitaur", settings={"cutoff": 30}) + + patches = patch_lib.list(synth="minitaur") + assert len(patches) == 2 + names = {p.name for p in patches} + assert names == {"a", "c"} + + +def test_list_filtered_case_insensitive(patch_lib): + patch_lib.save(name="a", synth="minitaur", settings={"cutoff": 10}) + patch_lib.save(name="b", synth="tb03", settings={"cutoff": 20}) + + patches = patch_lib.list(synth="MINITAUR") + assert len(patches) == 1 + assert patches[0].name == "a" + + +def test_list_empty(patch_lib): + assert patch_lib.list() == [] + + +def test_delete_existing(patch_lib): + patch_lib.save(name="bass", synth="minitaur", settings={"cutoff": 45}) + assert patch_lib.delete("bass") is True + assert patch_lib.get("bass") is None + + +def test_delete_nonexistent(patch_lib): + assert patch_lib.delete("nonexistent") is False + + +def test_settings_roundtrip(patch_lib): + settings = { + "cutoff": 45, + "resonance": 80, + "osc_mix": 100, + "lfo_rate": 0, + "volume": 127, + } + patch_lib.save(name="complex", synth="minitaur", settings=settings) + loaded = patch_lib.get("complex") + assert loaded is not None + assert loaded.settings == settings + + +def test_open_creates_directory(tmp_path): + db_path = tmp_path / "subdir" / "nested" / "test.db" + lib = PatchLibrary(db_path=db_path) + lib.open() + try: + assert db_path.parent.exists() + finally: + lib.close() + + +def test_description_none(patch_lib): + patch_lib.save(name="nodesc", synth="minitaur", settings={"cutoff": 10}, description=None) + loaded = patch_lib.get("nodesc") + assert loaded is not None + assert loaded.description is None + + +def test_context_manager(tmp_path): + db_path = tmp_path / "ctx.db" + with PatchLibrary(db_path=db_path) as lib: + lib.save(name="test", synth="minitaur", settings={"cutoff": 50}) + loaded = lib.get("test") + assert loaded is not None + assert loaded.settings == {"cutoff": 50} + # After exiting, connection should be closed + assert lib._conn is None diff --git a/tests/test_patch_tools.py b/tests/test_patch_tools.py new file mode 100644 index 0000000..ac56f95 --- /dev/null +++ b/tests/test_patch_tools.py @@ -0,0 +1,260 @@ +from unittest.mock import MagicMock + +import pytest +from pydantic_ai import RunContext +from pydantic_ai.usage import RunUsage + +from patchwork.deps import PatchworkDeps +from patchwork.midi import MidiConnection +from patchwork.patch_library import PatchLibrary +from patchwork.synth_definitions import CCParameter, SynthDefinition +from patchwork.tools.patches import ( + delete_patch, + list_patches, + load_patch, + recall_patch, + save_patch, +) + + +@pytest.fixture +def patch_lib(tmp_path): + with PatchLibrary(db_path=tmp_path / "test.db") as lib: + yield lib + + +def _make_synth() -> SynthDefinition: + return SynthDefinition( + name="TestSynth", + manufacturer="TestCo", + midi_channel=1, + cc_map={ + "cutoff": CCParameter(cc=74), + "resonance": CCParameter(cc=71), + }, + ) + + +def _make_ctx( + midi: MidiConnection | None = None, + synths: dict[str, SynthDefinition] | None = None, + patches: PatchLibrary | None = None, +) -> RunContext[PatchworkDeps]: + if midi is None: + midi = MidiConnection() + if synths is None: + synth = _make_synth() + synths = {synth.name.lower(): synth} + if patches is None: + patches = MagicMock() + deps = PatchworkDeps(midi=midi, synths=synths, patches=patches) + return RunContext(deps=deps, model=MagicMock(), usage=RunUsage()) + + +@pytest.mark.asyncio +async def test_save_patch_valid(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}) + assert "Saved patch 'bass'" in result + assert "cutoff = 45" in result + assert "resonance = 80" in result + + +@pytest.mark.asyncio +async def test_save_patch_unknown_synth(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await save_patch(ctx, name="bass", synth="unknown", settings={"cutoff": 45}) + assert "Unknown synth 'unknown'" in result + assert "testsynth" in result + + +@pytest.mark.asyncio +async def test_save_patch_invalid_parameter(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await save_patch(ctx, name="bass", synth="testsynth", settings={"nonexistent": 45}) + assert "Unknown parameter(s)" in result + assert "nonexistent" in result + assert "cutoff" in result + + +@pytest.mark.asyncio +async def test_save_patch_normalizes_keys(patch_lib): + ctx = _make_ctx(patches=patch_lib) + # Use mixed case and spaces — should normalize to lowercase underscored + synth = SynthDefinition( + name="TestSynth", + manufacturer="TestCo", + midi_channel=1, + cc_map={ + "filter_cutoff": CCParameter(cc=74), + }, + ) + ctx = _make_ctx(synths={"testsynth": synth}, patches=patch_lib) + result = await save_patch(ctx, name="bass", synth="testsynth", settings={"Filter Cutoff": 45}) + assert "Saved patch" in result + stored = patch_lib.get("bass") + assert stored is not None + assert "filter_cutoff" in stored.settings + assert stored.settings["filter_cutoff"] == 45 + + +@pytest.mark.asyncio +async def test_save_patch_empty_description_stored_as_none(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}, description="") + stored = patch_lib.get("bass") + assert stored is not None + assert stored.description is None + + +@pytest.mark.asyncio +async def test_save_patch_with_description(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}, description="A deep bass") + stored = patch_lib.get("bass") + assert stored is not None + assert stored.description == "A deep bass" + + +@pytest.mark.asyncio +async def test_load_patch_exists(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}, description="Deep bass") + result = await load_patch(ctx, name="bass") + assert "Patch 'bass'" in result + assert "cutoff = 45" in result + assert "resonance = 80" in result + assert "Deep bass" in result + + +@pytest.mark.asyncio +async def test_load_patch_not_found(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await load_patch(ctx, name="nonexistent") + assert "No patch found" in result + + +@pytest.mark.asyncio +async def test_recall_patch_sends_cc(patch_lib): + midi = MidiConnection() + mock_out = MagicMock() + midi._out = mock_out + midi._port_name = "Test Port" + + ctx = _make_ctx(midi=midi, patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}) + + result = await recall_patch(ctx, name="bass") + assert "Recalled patch 'bass'" in result + assert "cutoff = 45" in result + assert "resonance = 80" in result + # Verify CC messages were sent (channel 1 = status 0xB0) + assert mock_out.send_message.call_count == 2 + calls = [c.args[0] for c in mock_out.send_message.call_args_list] + assert [0xB0, 74, 45] in calls # cutoff CC 74 + assert [0xB0, 71, 80] in calls # resonance CC 71 + + +@pytest.mark.asyncio +async def test_recall_patch_not_found(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await recall_patch(ctx, name="nonexistent") + assert "No patch found" in result + + +@pytest.mark.asyncio +async def test_recall_patch_synth_not_loaded(patch_lib): + ctx = _make_ctx(patches=patch_lib) + # Save a patch for "testsynth" + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + # Now create a context without the synth definition + ctx_no_synth = _make_ctx(synths={}, patches=patch_lib) + result = await recall_patch(ctx_no_synth, name="bass") + assert "no definition is loaded" in result + + +@pytest.mark.asyncio +async def test_recall_patch_midi_not_connected(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + result = await recall_patch(ctx, name="bass") + assert "MIDI not connected" in result + + +@pytest.mark.asyncio +async def test_recall_patch_skips_unknown_params(patch_lib): + # Save a patch with cutoff and resonance + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}) + + # Now create a synth def that only has cutoff (resonance removed) + limited_synth = SynthDefinition( + name="TestSynth", + manufacturer="TestCo", + midi_channel=1, + cc_map={"cutoff": CCParameter(cc=74)}, + ) + midi = MidiConnection() + mock_out = MagicMock() + midi._out = mock_out + midi._port_name = "Test Port" + + ctx2 = _make_ctx(midi=midi, synths={"testsynth": limited_synth}, patches=patch_lib) + result = await recall_patch(ctx2, name="bass") + assert "cutoff = 45" in result + assert "Skipped 'resonance'" in result + assert mock_out.send_message.call_count == 1 + + +@pytest.mark.asyncio +async def test_list_patches_all(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + await save_patch(ctx, name="lead", synth="testsynth", settings={"cutoff": 100, "resonance": 90}) + result = await list_patches(ctx) + assert "bass" in result + assert "lead" in result + assert "Saved patches (2)" in result + + +@pytest.mark.asyncio +async def test_list_patches_filtered(patch_lib): + synth2 = SynthDefinition( + name="OtherSynth", + manufacturer="OtherCo", + midi_channel=2, + cc_map={"volume": CCParameter(cc=7)}, + ) + synths = { + "testsynth": _make_synth(), + "othersynth": synth2, + } + ctx = _make_ctx(synths=synths, patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + await save_patch(ctx, name="pad", synth="othersynth", settings={"volume": 100}) + + result = await list_patches(ctx, synth="testsynth") + assert "bass" in result + assert "pad" not in result + + +@pytest.mark.asyncio +async def test_list_patches_empty(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await list_patches(ctx) + assert "No patches saved yet" in result + + +@pytest.mark.asyncio +async def test_delete_patch_exists(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + result = await delete_patch(ctx, name="bass") + assert "Deleted patch 'bass'" in result + + +@pytest.mark.asyncio +async def test_delete_patch_not_found(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await delete_patch(ctx, name="nonexistent") + assert "No patch found" in result From 3f885f66695a7c8ee7974108f8152f39f456a24a Mon Sep 17 00:00:00 2001 From: Patrick Nilan Date: Sun, 8 Mar 2026 20:48:23 -0700 Subject: [PATCH 2/4] Address PR review: validation parity, cross-synth guard, and nits - save_patch: validate CC values against value_range before saving - save_patch: reject cross-synth overwrites (same name, different synth) - recall_patch: skip out-of-range values with warning (matches send_patch) - recall_patch: fix MIDI-not-connected message to match midi_control.py - load_patch: show updated_at instead of created_at - cli.py: use PatchLibrary context manager, fix except syntax - Add tests: value-range validation, cross-synth rejection, empty settings, same-synth overwrite, recall out-of-range skip, operations after close --- patchwork/cli.py | 81 +++++++++++++++++---------------- patchwork/tools/patches.py | 32 ++++++++++--- tests/test_patch_library.py | 8 ++++ tests/test_patch_tools.py | 89 +++++++++++++++++++++++++++++++++++++ 4 files changed, 164 insertions(+), 46 deletions(-) diff --git a/patchwork/cli.py b/patchwork/cli.py index 4111dc1..f47339d 100644 --- a/patchwork/cli.py +++ b/patchwork/cli.py @@ -14,48 +14,47 @@ async def main(): midi = MidiConnection() - patches = PatchLibrary() - patches.open() synths = load_synth_definitions() - deps = PatchworkDeps(midi=midi, synths=synths, patches=patches) - - console.print("[bold]patchwork[/bold] — synth research agent\n") - if synths: - synth_names = ", ".join(s.name for s in synths.values()) - console.print(f"[dim]loaded {len(synths)} synth(s): {synth_names}[/dim]\n") - else: - console.print("[dim]no synth definitions found in synths/[/dim]\n") - message_history = [] - - try: - while True: - try: - user_input = console.input("[bold cyan]patch>[/bold cyan] ") - except KeyboardInterrupt, EOFError: - console.print("\n[dim]goodbye[/dim]") - break - - if not user_input.strip(): - continue - - if user_input.strip().lower() in ("quit", "exit"): - console.print("[dim]goodbye[/dim]") - break - - try: - async with agent.run_stream( - user_input, message_history=message_history, deps=deps - ) as result: - async for chunk in result.stream_text(delta=True): - console.print(chunk, end="", markup=False, highlight=False) - console.print() # newline after stream - - message_history = result.all_messages() - except Exception as e: - console.print(f"\n[bold red]error:[/bold red] {e}") - finally: - midi.close() - patches.close() + + with PatchLibrary() as patches: + deps = PatchworkDeps(midi=midi, synths=synths, patches=patches) + + console.print("[bold]patchwork[/bold] — synth research agent\n") + if synths: + synth_names = ", ".join(s.name for s in synths.values()) + console.print(f"[dim]loaded {len(synths)} synth(s): {synth_names}[/dim]\n") + else: + console.print("[dim]no synth definitions found in synths/[/dim]\n") + message_history = [] + + try: + while True: + try: + user_input = console.input("[bold cyan]patch>[/bold cyan] ") + except (KeyboardInterrupt, EOFError): + console.print("\n[dim]goodbye[/dim]") + break + + if not user_input.strip(): + continue + + if user_input.strip().lower() in ("quit", "exit"): + console.print("[dim]goodbye[/dim]") + break + + try: + async with agent.run_stream( + user_input, message_history=message_history, deps=deps + ) as result: + async for chunk in result.stream_text(delta=True): + console.print(chunk, end="", markup=False, highlight=False) + console.print() # newline after stream + + message_history = result.all_messages() + except Exception as e: + console.print(f"\n[bold red]error:[/bold red] {e}") + finally: + midi.close() def main_cli(): diff --git a/patchwork/tools/patches.py b/patchwork/tools/patches.py index fe9e415..db3c3b4 100644 --- a/patchwork/tools/patches.py +++ b/patchwork/tools/patches.py @@ -24,21 +24,37 @@ async def save_patch( available = ", ".join(ctx.deps.synths.keys()) return f"Unknown synth '{synth}'. Available: {available}" + # Reject cross-synth overwrites + existing = ctx.deps.patches.get(name) + if existing and existing.synth != synth_key: + return ( + f"A patch named '{name}' already exists for {existing.synth}. " + f"Delete it first or choose a different name." + ) + # Validate parameter names and normalize keys to canonical form (lowercase, underscored) normalized_settings: dict[str, int] = {} invalid_params = [] + out_of_range = [] for param_name, value in settings.items(): param_key = param_name.lower().replace(" ", "_") - if param_key not in synth_def.cc_map: + param = synth_def.cc_map.get(param_key) + if param is None: invalid_params.append(param_name) else: - normalized_settings[param_key] = value + low, high = param.value_range + if not (low <= value <= high): + out_of_range.append(f"{param_key}: {value} (valid: {low}-{high})") + else: + normalized_settings[param_key] = value if invalid_params: available = ", ".join(synth_def.cc_map.keys()) return ( f"Unknown parameter(s) for {synth_def.name}: {', '.join(invalid_params)}. " f"Available: {available}" ) + if out_of_range: + return f"Value(s) out of range for {synth_def.name}: {'; '.join(out_of_range)}" patch = ctx.deps.patches.save( name=name, @@ -74,7 +90,7 @@ async def load_patch( lines.append(f" Description: {patch.description}") for param, value in patch.settings.items(): lines.append(f" {param} = {value}") - lines.append(f" Saved: {patch.created_at:%Y-%m-%d %H:%M}") + lines.append(f" Updated: {patch.updated_at:%Y-%m-%d %H:%M}") return "\n".join(lines) @@ -100,8 +116,8 @@ async def recall_patch( if not ctx.deps.midi.is_connected: return ( - "MIDI not connected. Use list_midi_ports to see available ports, " - "then ask me to connect." + "MIDI not connected. Use list_midi_ports to see available ports," + " then ask me to connect." ) results = [] @@ -110,6 +126,12 @@ async def recall_patch( if param is None: results.append(f" Skipped '{param_name}' (not in current CC map)") continue + low, high = param.value_range + if not (low <= value <= high): + results.append( + f" Skipped '{param_name}': value {value} out of range ({low}-{high})" + ) + continue ctx.deps.midi.send_cc(synth_def.midi_channel, param.cc, value) results.append(f" {param_name} = {value} (CC {param.cc})") diff --git a/tests/test_patch_library.py b/tests/test_patch_library.py index 3543fb3..f202718 100644 --- a/tests/test_patch_library.py +++ b/tests/test_patch_library.py @@ -134,6 +134,14 @@ def test_description_none(patch_lib): assert loaded.description is None +def test_operations_after_close(tmp_path): + lib = PatchLibrary(db_path=tmp_path / "test.db") + lib.open() + lib.close() + with pytest.raises(RuntimeError, match="Database not open"): + lib.save(name="test", synth="minitaur", settings={"cutoff": 50}) + + def test_context_manager(tmp_path): db_path = tmp_path / "ctx.db" with PatchLibrary(db_path=db_path) as lib: diff --git a/tests/test_patch_tools.py b/tests/test_patch_tools.py index ac56f95..066a7d3 100644 --- a/tests/test_patch_tools.py +++ b/tests/test_patch_tools.py @@ -107,6 +107,70 @@ async def test_save_patch_empty_description_stored_as_none(patch_lib): assert stored.description is None +@pytest.mark.asyncio +async def test_save_patch_value_out_of_range(patch_lib): + synth = SynthDefinition( + name="TestSynth", + manufacturer="TestCo", + midi_channel=1, + cc_map={ + "special": CCParameter(cc=50, value_range=(0, 64)), + }, + ) + ctx = _make_ctx(synths={"testsynth": synth}, patches=patch_lib) + result = await save_patch(ctx, name="bass", synth="testsynth", settings={"special": 100}) + assert "out of range" in result + assert "100" in result + # Verify nothing was saved + assert patch_lib.get("bass") is None + + +@pytest.mark.asyncio +async def test_save_patch_cross_synth_rejected(patch_lib): + synth2 = SynthDefinition( + name="OtherSynth", + manufacturer="OtherCo", + midi_channel=2, + cc_map={"volume": CCParameter(cc=7)}, + ) + synths = { + "testsynth": _make_synth(), + "othersynth": synth2, + } + ctx = _make_ctx(synths=synths, patches=patch_lib) + # Save "bass" for testsynth + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + # Try to save "bass" for a different synth + result = await save_patch(ctx, name="bass", synth="othersynth", settings={"volume": 100}) + assert "already exists" in result + assert "testsynth" in result + # Verify original patch is unchanged + stored = patch_lib.get("bass") + assert stored is not None + assert stored.synth == "testsynth" + + +@pytest.mark.asyncio +async def test_save_patch_same_synth_overwrites(patch_lib): + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + result = await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 90}) + assert "Saved patch 'bass'" in result + stored = patch_lib.get("bass") + assert stored is not None + assert stored.settings["cutoff"] == 90 + + +@pytest.mark.asyncio +async def test_save_patch_empty_settings(patch_lib): + ctx = _make_ctx(patches=patch_lib) + result = await save_patch(ctx, name="empty", synth="testsynth", settings={}) + assert "Saved patch 'empty'" in result + stored = patch_lib.get("empty") + assert stored is not None + assert stored.settings == {} + + @pytest.mark.asyncio async def test_save_patch_with_description(patch_lib): ctx = _make_ctx(patches=patch_lib) @@ -181,6 +245,31 @@ async def test_recall_patch_midi_not_connected(patch_lib): assert "MIDI not connected" in result +@pytest.mark.asyncio +async def test_recall_patch_skips_out_of_range(patch_lib): + # Save a patch with a value that's valid for default range (0-127) + ctx = _make_ctx(patches=patch_lib) + await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}) + + # Now create a synth def where cutoff has a narrower range + narrow_synth = SynthDefinition( + name="TestSynth", + manufacturer="TestCo", + midi_channel=1, + cc_map={"cutoff": CCParameter(cc=74, value_range=(0, 30))}, + ) + midi = MidiConnection() + mock_out = MagicMock() + midi._out = mock_out + midi._port_name = "Test Port" + + ctx2 = _make_ctx(midi=midi, synths={"testsynth": narrow_synth}, patches=patch_lib) + result = await recall_patch(ctx2, name="bass") + assert "Skipped 'cutoff'" in result + assert "out of range" in result + assert mock_out.send_message.call_count == 0 + + @pytest.mark.asyncio async def test_recall_patch_skips_unknown_params(patch_lib): # Save a patch with cutoff and resonance From e903f5bfcc632ada4980bb60e5ba85a6ce6bf93a Mon Sep 17 00:00:00 2001 From: Patrick Nilan Date: Sun, 8 Mar 2026 20:50:19 -0700 Subject: [PATCH 3/4] Fix line length lint errors in test_patch_tools.py --- tests/test_patch_tools.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/test_patch_tools.py b/tests/test_patch_tools.py index 066a7d3..1711790 100644 --- a/tests/test_patch_tools.py +++ b/tests/test_patch_tools.py @@ -54,7 +54,9 @@ def _make_ctx( @pytest.mark.asyncio async def test_save_patch_valid(patch_lib): ctx = _make_ctx(patches=patch_lib) - result = await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}) + result = await save_patch( + ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80} + ) assert "Saved patch 'bass'" in result assert "cutoff = 45" in result assert "resonance = 80" in result @@ -174,7 +176,10 @@ async def test_save_patch_empty_settings(patch_lib): @pytest.mark.asyncio async def test_save_patch_with_description(patch_lib): ctx = _make_ctx(patches=patch_lib) - await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45}, description="A deep bass") + await save_patch( + ctx, name="bass", synth="testsynth", + settings={"cutoff": 45}, description="A deep bass", + ) stored = patch_lib.get("bass") assert stored is not None assert stored.description == "A deep bass" @@ -183,7 +188,10 @@ async def test_save_patch_with_description(patch_lib): @pytest.mark.asyncio async def test_load_patch_exists(patch_lib): ctx = _make_ctx(patches=patch_lib) - await save_patch(ctx, name="bass", synth="testsynth", settings={"cutoff": 45, "resonance": 80}, description="Deep bass") + await save_patch( + ctx, name="bass", synth="testsynth", + settings={"cutoff": 45, "resonance": 80}, description="Deep bass", + ) result = await load_patch(ctx, name="bass") assert "Patch 'bass'" in result assert "cutoff = 45" in result From 534210bbf1463a5a7ea4fff848ce2a0658ecea02 Mon Sep 17 00:00:00 2001 From: Patrick Nilan Date: Sun, 8 Mar 2026 20:51:29 -0700 Subject: [PATCH 4/4] chore: format --- patchwork/agent.py | 12 ++++++++++-- patchwork/cli.py | 2 +- patchwork/patch_library.py | 8 ++------ patchwork/tools/patches.py | 9 ++------- tests/test_patch_tools.py | 14 ++++++++++---- 5 files changed, 25 insertions(+), 20 deletions(-) diff --git a/patchwork/agent.py b/patchwork/agent.py index 9701974..c929754 100644 --- a/patchwork/agent.py +++ b/patchwork/agent.py @@ -74,8 +74,16 @@ deps_type=PatchworkDeps, defer_model_check=True, tools=[ - list_midi_ports, connect_midi, list_synths, send_cc, send_patch, - save_patch, load_patch, recall_patch, list_patches, delete_patch, + list_midi_ports, + connect_midi, + list_synths, + send_cc, + send_patch, + save_patch, + load_patch, + recall_patch, + list_patches, + delete_patch, ], ) diff --git a/patchwork/cli.py b/patchwork/cli.py index f47339d..4955c76 100644 --- a/patchwork/cli.py +++ b/patchwork/cli.py @@ -31,7 +31,7 @@ async def main(): while True: try: user_input = console.input("[bold cyan]patch>[/bold cyan] ") - except (KeyboardInterrupt, EOFError): + except KeyboardInterrupt, EOFError: console.print("\n[dim]goodbye[/dim]") break diff --git a/patchwork/patch_library.py b/patchwork/patch_library.py index d239e7e..f87cad3 100644 --- a/patchwork/patch_library.py +++ b/patchwork/patch_library.py @@ -95,9 +95,7 @@ def save( def get(self, name: str) -> Patch | None: """Get a patch by name. Returns None if not found.""" - row = self._db.execute( - "SELECT * FROM patches WHERE name = ?", (name,) - ).fetchone() + row = self._db.execute("SELECT * FROM patches WHERE name = ?", (name,)).fetchone() if row is None: return None return self._row_to_patch(row) @@ -110,9 +108,7 @@ def list(self, synth: str | None = None) -> list[Patch]: (synth.lower(),), ).fetchall() else: - rows = self._db.execute( - "SELECT * FROM patches ORDER BY updated_at DESC" - ).fetchall() + rows = self._db.execute("SELECT * FROM patches ORDER BY updated_at DESC").fetchall() return [self._row_to_patch(row) for row in rows] def delete(self, name: str) -> bool: diff --git a/patchwork/tools/patches.py b/patchwork/tools/patches.py index db3c3b4..92af5f8 100644 --- a/patchwork/tools/patches.py +++ b/patchwork/tools/patches.py @@ -63,10 +63,7 @@ async def save_patch( description=description or None, ) param_lines = [f" {k} = {v}" for k, v in patch.settings.items()] - return ( - f"Saved patch '{patch.name}' for {synth_def.name}:\n" - + "\n".join(param_lines) - ) + return f"Saved patch '{patch.name}' for {synth_def.name}:\n" + "\n".join(param_lines) async def load_patch( @@ -128,9 +125,7 @@ async def recall_patch( continue low, high = param.value_range if not (low <= value <= high): - results.append( - f" Skipped '{param_name}': value {value} out of range ({low}-{high})" - ) + results.append(f" Skipped '{param_name}': value {value} out of range ({low}-{high})") continue ctx.deps.midi.send_cc(synth_def.midi_channel, param.cc, value) results.append(f" {param_name} = {value} (CC {param.cc})") diff --git a/tests/test_patch_tools.py b/tests/test_patch_tools.py index 1711790..0c0389f 100644 --- a/tests/test_patch_tools.py +++ b/tests/test_patch_tools.py @@ -177,8 +177,11 @@ async def test_save_patch_empty_settings(patch_lib): async def test_save_patch_with_description(patch_lib): ctx = _make_ctx(patches=patch_lib) await save_patch( - ctx, name="bass", synth="testsynth", - settings={"cutoff": 45}, description="A deep bass", + ctx, + name="bass", + synth="testsynth", + settings={"cutoff": 45}, + description="A deep bass", ) stored = patch_lib.get("bass") assert stored is not None @@ -189,8 +192,11 @@ async def test_save_patch_with_description(patch_lib): async def test_load_patch_exists(patch_lib): ctx = _make_ctx(patches=patch_lib) await save_patch( - ctx, name="bass", synth="testsynth", - settings={"cutoff": 45, "resonance": 80}, description="Deep bass", + ctx, + name="bass", + synth="testsynth", + settings={"cutoff": 45, "resonance": 80}, + description="Deep bass", ) result = await load_patch(ctx, name="bass") assert "Patch 'bass'" in result