Phase 3: Patch library — SQLite storage, tools, and tests

.gitignore

@@ -13,4 +13,4 @@ wheels/
 .env
 
 # Data
-data/
+data/*.db

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,18 @@
     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,
+    ],
 )
 
 

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()
@@ -14,44 +15,46 @@
 async def main():
     midi = MidiConnection()
     synths = load_synth_definitions()
-    deps = PatchworkDeps(midi=midi, synths=synths)
-
-    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()
+
+    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():

patchwork/deps.py

@@ -1,10 +1,12 @@
 from dataclasses import dataclass
 
 from patchwork.midi import MidiConnection
+from patchwork.patch_library import PatchLibrary
 from patchwork.synth_definitions import SynthDefinition
 
 
 @dataclass
 class PatchworkDeps:
     midi: MidiConnection
     synths: dict[str, SynthDefinition]
+    patches: PatchLibrary

patchwork/patch_library.py

@@ -0,0 +1,130 @@
+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"]),
+        )