From d8ecd258e9315a09c24853a12c9fefd2a54d8093 Mon Sep 17 00:00:00 2001
From: 7000pctAUTO <lukems3823@gmail.com>
Date: Wed, 4 Feb 2026 12:49:08 +0000
Subject: [PATCH] fix: resolve CI linting and type errors

---
 app/src/promptforge/registry/local.py | 153 ++++++++++++++++++++++++++
 1 file changed, 153 insertions(+)
 create mode 100644 app/src/promptforge/registry/local.py

diff --git a/app/src/promptforge/registry/local.py b/app/src/promptforge/registry/local.py
new file mode 100644
index 0000000..da5f6c5
--- /dev/null
+++ b/app/src/promptforge/registry/local.py
@@ -0,0 +1,153 @@
+import json
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from .models import RegistryEntry, RegistrySearchResult
+from ..core.exceptions import RegistryError
+
+
+class LocalRegistry:
+    """Local prompt registry stored as JSON files."""
+
+    def __init__(self, registry_path: Optional[str] = None):
+        """Initialize local registry.
+
+        Args:
+            registry_path: Path to registry directory. Defaults to ~/.promptforge/registry
+        """
+        self.registry_path = Path(registry_path or self._default_path())
+        self.registry_path.mkdir(parents=True, exist_ok=True)
+        self._index_file = self.registry_path / "index.json"
+
+    def _default_path(self) -> str:
+        import os
+        return os.path.expanduser("~/.promptforge/registry")
+
+    def _load_index(self) -> Dict[str, RegistryEntry]:
+        """Load registry index."""
+        if not self._index_file.exists():
+            return {}
+        try:
+            with open(self._index_file, 'r') as f:
+                data = json.load(f)
+            return {
+                k: RegistryEntry(**v) for k, v in data.items()
+            }
+        except Exception as e:
+            raise RegistryError(f"Failed to load registry index: {e}")
+
+    def _save_index(self, index: Dict[str, RegistryEntry]) -> None:
+        """Save registry index."""
+        try:
+            data = {k: v.model_dump() for k, v in index.items()}
+            with open(self._index_file, 'w') as f:
+                json.dump(data, f, indent=2)
+        except Exception as e:
+            raise RegistryError(f"Failed to save registry index: {e}")
+
+    def add(self, entry: RegistryEntry) -> None:
+        """Add an entry to the registry.
+
+        Args:
+            entry: Registry entry to add.
+        """
+        index = self._load_index()
+        entry.id = str(entry.id or uuid.uuid4())
+        entry.added_at = entry.added_at or datetime.utcnow()
+        entry.updated_at = datetime.utcnow()
+        index[entry.id] = entry
+        self._save_index(index)
+
+    def remove(self, entry_id: str) -> bool:
+        """Remove an entry from the registry.
+
+        Args:
+            entry_id: ID of entry to remove.
+
+        Returns:
+            True if entry was removed, False if not found.
+        """
+        index = self._load_index()
+        if entry_id not in index:
+            return False
+        del index[entry_id]
+        self._save_index(index)
+        return True
+
+    def get(self, entry_id: str) -> Optional[RegistryEntry]:
+        """Get an entry by ID.
+
+        Args:
+            entry_id: Entry ID.
+
+        Returns:
+            Registry entry or None if not found.
+        """
+        index = self._load_index()
+        return index.get(entry_id)
+
+    def list(
+        self,
+        tag: Optional[str] = None,
+        author: Optional[str] = None,
+        limit: int = 50,
+    ) -> List[RegistryEntry]:
+        """List entries in the registry.
+
+        Args:
+            tag: Filter by tag.
+            author: Filter by author.
+            limit: Maximum results to return.
+
+        Returns:
+            List of matching entries.
+        """
+        index = self._load_index()
+        results = list(index.values())
+
+        if tag:
+            results = [e for e in results if tag in e.tags]
+
+        if author:
+            results = [e for e in results if e.author == author]
+
+        results.sort(key=lambda e: e.added_at or datetime.min, reverse=True)
+        return results[:limit]
+
+    def search(self, query: str) -> List[RegistrySearchResult]:
+        """Search registry entries.
+
+        Args:
+            query: Search query.
+
+        Returns:
+            List of matching entries with relevance scores.
+        """
+        entries = self.list(limit=100)
+        results = []
+
+        query_lower = query.lower()
+        for entry in entries:
+            score = 0
+
+            if query_lower in entry.name.lower():
+                score += 10
+            if entry.description and query_lower in entry.description.lower():
+                score += 5
+            if any(query_lower in tag for tag in entry.tags):
+                score += 3
+
+            if score > 0:
+                results.append(RegistrySearchResult(
+                    entry=entry,
+                    relevance_score=score,
+                ))
+
+        return sorted(results, key=lambda r: r.relevance_score, reverse=True)
+
+    def count(self) -> int:
+        """Get total number of entries."""
+        index = self._load_index()
+        return len(index)