fix: address code review findings for performance and quality

Performance improvements (PERF-001 through PERF-007):
- Add show_notes_batch() for batch git notes retrieval using git cat-file --batch
- Add embed_batch() for batch embedding generation
- Add prewarm() method for eager model loading
- Add @lru_cache for struct format caching in index.py
- Implement batch hydration in recall.py using show_notes_batch()
- Use batch operations in sync.py for index synchronization

Quality improvements (QUAL-001, QUAL-002):
- Rename _read_input to _read_input_with_fallback for clarity
- Replace broad Exception catches with specific (MemoryIndexError, OSError)

Security improvements (SEC-001, SEC-002):
- Add input length limit in signal_detector.py for ReDoS prevention

Test updates:
- Update test fixtures to mock batch operations
- Update test assertions for renamed functions and specific exceptions

All 1806 tests passing with 89.29% coverage.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: zircote

src/git_notes_memory/embedding.py

@@ -310,6 +310,32 @@ def similarity(
         # Dot product of normalized vectors = cosine similarity
         return sum(a * b for a, b in zip(embedding1, embedding2, strict=True))
 
+    def prewarm(self) -> bool:
+        """Pre-warm the embedding model by loading it eagerly.
+
+        PERF-004: Call this during application startup or hook initialization
+        to avoid cold start latency on first embed() call. Useful for:
+        - Session start hooks that need fast response
+        - Background workers that will process embeddings
+        - Applications where predictable latency is important
+
+        Returns:
+            True if model was loaded (or already loaded), False on error.
+
+        Examples:
+            >>> service = EmbeddingService()
+            >>> service.prewarm()  # Load model in background
+            True
+            >>> service.is_loaded
+            True
+        """
+        try:
+            self.load()
+            return True
+        except Exception as e:
+            logger.warning("Failed to pre-warm embedding model: %s", e)
+            return False
+
     def unload(self) -> None:
         """Unload the model to free memory.
 
@@ -322,13 +348,10 @@ def unload(self) -> None:
 
 
 # =============================================================================
-# Singleton Instance
+# Singleton Access (using ServiceRegistry)
 # =============================================================================
 
 
-_default_service: EmbeddingService | None = None
-
-
 def get_default_service() -> EmbeddingService:
     """Get the default embedding service singleton.
 
@@ -340,7 +363,6 @@ def get_default_service() -> EmbeddingService:
         >>> service.model_name
         'all-MiniLM-L6-v2'
     """
-    global _default_service
-    if _default_service is None:
-        _default_service = EmbeddingService()
-    return _default_service
+    from git_notes_memory.registry import ServiceRegistry
+
+    return ServiceRegistry.get(EmbeddingService)

src/git_notes_memory/git_ops.py

@@ -143,11 +143,12 @@ def _run_git(
             )
         except subprocess.CalledProcessError as e:
             # Parse common git errors for better messages
+            # SEC-002: Sanitize paths in error messages to prevent info leakage
             stderr = e.stderr or ""
             if "not a git repository" in stderr.lower():
                 raise StorageError(
                     "Not in a Git repository",
-                    f"Initialize a git repository: cd {self.repo_path} && git init",
+                    "Initialize a git repository: cd <repo_path> && git init",
                 ) from e
             if "permission denied" in stderr.lower():
                 raise StorageError(
@@ -159,8 +160,12 @@ def _run_git(
                     "Repository has no commits",
                     "Create at least one commit: git commit --allow-empty -m 'initial'",
                 ) from e
+            # Sanitize the args to remove full paths
+            sanitized_args = [
+                arg if not arg.startswith("/") else "<path>" for arg in args
+            ]
             raise StorageError(
-                f"Git command failed: {' '.join(args)}\n{stderr}",
+                f"Git command failed: {' '.join(sanitized_args)}\n{stderr}",
                 "Check git status and try again",
             ) from e
 
@@ -321,6 +326,105 @@ def show_note(
 
         return result.stdout
 
+    def show_notes_batch(
+        self,
+        namespace: str,
+        commit_shas: list[str],
+    ) -> dict[str, str | None]:
+        """Show multiple notes in a single subprocess call.
+
+        Uses `git cat-file --batch` for efficient bulk retrieval.
+        This is significantly faster than calling show_note() in a loop
+        when fetching many notes.
+
+        Args:
+            namespace: Memory namespace.
+            commit_shas: List of commit SHAs to get notes for.
+
+        Returns:
+            Dict mapping commit_sha -> note content (or None if no note).
+
+        Raises:
+            ValidationError: If namespace is invalid.
+        """
+        if not commit_shas:
+            return {}
+
+        self._validate_namespace(namespace)
+        for sha in commit_shas:
+            self._validate_git_ref(sha)
+
+        # Build object references: notes ref points to the note object for each commit
+        # Format: refs/notes/mem/namespace:commit_sha
+        ref = self._note_ref(namespace)
+        objects_input = "\n".join(f"{ref}:{sha}" for sha in commit_shas)
+
+        # Run cat-file --batch to get all notes at once
+        cmd = ["git", "-C", str(self.repo_path), "cat-file", "--batch"]
+
+        try:
+            result = subprocess.run(
+                cmd,
+                input=objects_input,
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+        except Exception:
+            # Fallback to sequential if batch fails
+            return {sha: self.show_note(namespace, sha) for sha in commit_shas}
+
+        # Parse batch output
+        # Format per object:
+        #   <sha> <type> <size>\n
+        #   <content>\n
+        # Or for missing:
+        #   <ref> missing\n
+        results: dict[str, str | None] = {}
+        lines: list[str] = result.stdout.split("\n")
+        i = 0
+        sha_index = 0
+
+        while i < len(lines) and sha_index < len(commit_shas):
+            line = lines[i]
+            current_sha = commit_shas[sha_index]
+
+            if "missing" in line:
+                results[current_sha] = None
+                i += 1
+                sha_index += 1
+            elif line and not line.startswith(" "):
+                # Header line: <object_sha> <type> <size>
+                parts = line.split()
+                if len(parts) >= 3:
+                    try:
+                        size = int(parts[2])
+                        # Content follows on next lines until size bytes consumed
+                        content_lines: list[str] = []
+                        remaining = size
+                        i += 1
+                        while remaining > 0 and i < len(lines):
+                            content_line = lines[i]
+                            content_lines.append(content_line)
+                            remaining -= len(content_line) + 1  # +1 for newline
+                            i += 1
+                        results[current_sha] = "\n".join(content_lines)
+                        sha_index += 1
+                    except (ValueError, IndexError):
+                        results[current_sha] = None
+                        sha_index += 1
+                        i += 1
+                else:
+                    i += 1
+            else:
+                i += 1
+
+        # Fill in any remaining SHAs as None
+        for remaining_sha in commit_shas[sha_index:]:
+            results[remaining_sha] = None
+
+        return results
+
     def list_notes(
         self,
         namespace: str,

src/git_notes_memory/hooks/context_builder.py

@@ -19,6 +19,7 @@
 from typing import TYPE_CHECKING
 
 from git_notes_memory.config import TOKENS_PER_CHAR, get_project_index_path
+from git_notes_memory.exceptions import MemoryIndexError
 from git_notes_memory.hooks.config_loader import (
     BudgetMode,
     HookConfig,
@@ -556,7 +557,8 @@ def _analyze_project_complexity(self, project: str) -> str:
                 return "complex"
             return "full"
 
-        except Exception as e:
+        # QUAL-002: Catch specific exceptions instead of bare Exception
+        except (MemoryIndexError, OSError) as e:
             logger.debug("Failed to analyze complexity for %s: %s", project, e)
             return "medium"  # Default to medium on error
 

src/git_notes_memory/hooks/signal_detector.py

@@ -271,6 +271,17 @@ def detect(self, text: str) -> list[CaptureSignal]:
         if not text or len(text) < 5:
             return []
 
+        # SEC-001: Limit input length to prevent ReDoS attacks
+        # 100KB is generous for user prompts while preventing abuse
+        MAX_TEXT_LENGTH = 100 * 1024  # 100KB
+        if len(text) > MAX_TEXT_LENGTH:
+            logger.warning(
+                "Input text length %d exceeds maximum %d, truncating for safety",
+                len(text),
+                MAX_TEXT_LENGTH,
+            )
+            text = text[:MAX_TEXT_LENGTH]
+
         signals: list[CaptureSignal] = []
         block_positions: set[tuple[int, int]] = set()
 

src/git_notes_memory/hooks/stop_handler.py

@@ -45,6 +45,7 @@
     cancel_timeout,
     get_hook_logger,
     log_hook_input,
+    read_json_input,
     setup_logging,
     setup_timeout,
 )
@@ -55,23 +56,22 @@
 logger = logging.getLogger(__name__)
 
 
-def _read_input() -> dict[str, Any]:
-    """Read and parse JSON input from stdin.
+def _read_input_with_fallback() -> dict[str, Any]:
+    """Read and parse JSON input from stdin with fallback for empty input.
 
-    Returns:
-        Parsed JSON data.
+    QUAL-001: Wraps hook_utils.read_json_input() with Stop-hook-specific
+    fallback behavior (empty input is valid for stop hooks).
 
-    Raises:
-        json.JSONDecodeError: If input is not valid JSON.
+    Returns:
+        Parsed JSON data, or empty dict if stdin is empty.
     """
-    input_text = sys.stdin.read()
-    if not input_text.strip():
+    try:
+        return read_json_input()
+    except ValueError as e:
         # Empty input is OK for stop hook
-        return {}
-    result = json.loads(input_text)
-    if not isinstance(result, dict):
-        return {}
-    return dict(result)
+        if "empty" in str(e).lower():
+            return {}
+        raise
 
 
 def _analyze_session(transcript_path: str | None) -> list[CaptureSignal]:
@@ -385,8 +385,8 @@ def main() -> None:
     setup_timeout(timeout, hook_name="Stop")
 
     try:
-        # Read input (may be empty for stop hook)
-        input_data = _read_input()
+        # QUAL-001: Use hook_utils.read_json_input with fallback
+        input_data = _read_input_with_fallback()
         logger.debug("Received stop hook input: %s", list(input_data.keys()))
 
         # Log full input to file for debugging

src/git_notes_memory/index.py

@@ -23,8 +23,10 @@
 
 import contextlib
 import sqlite3
+import struct
 from contextlib import contextmanager
 from datetime import UTC, datetime
+from functools import lru_cache
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -33,6 +35,24 @@
 from git_notes_memory.config import EMBEDDING_DIMENSIONS, get_index_path
 from git_notes_memory.exceptions import MemoryIndexError
 
+
+# PERF-007: Cache compiled struct format for embedding serialization
+@lru_cache(maxsize=8)
+def _get_struct_format(dimensions: int) -> struct.Struct:
+    """Get a cached struct.Struct for packing embeddings.
+
+    The embedding dimensions are typically constant (384 for all-MiniLM-L6-v2),
+    so caching the compiled Struct avoids repeated format string parsing.
+
+    Args:
+        dimensions: Number of float values in the embedding.
+
+    Returns:
+        A compiled struct.Struct instance for packing.
+    """
+    return struct.Struct(f"{dimensions}f")
+
+
 if TYPE_CHECKING:
     from collections.abc import Iterator, Sequence
 
@@ -508,10 +528,8 @@ def _insert_embedding(
             memory_id: ID of the memory this embedding belongs to.
             embedding: The embedding vector.
         """
-        # sqlite-vec expects binary format for vectors
-        import struct
-
-        blob = struct.pack(f"{len(embedding)}f", *embedding)
+        # PERF-007: Use cached struct format for embedding packing
+        blob = _get_struct_format(len(embedding)).pack(*embedding)
         cursor.execute(
             "INSERT INTO vec_memories (id, embedding) VALUES (?, ?)",
             (memory_id, blob),
@@ -820,9 +838,8 @@ def _update_embedding(
             memory_id: ID of the memory this embedding belongs to.
             embedding: The new embedding vector.
         """
-        import struct
-
-        blob = struct.pack(f"{len(embedding)}f", *embedding)
+        # PERF-007: Use cached struct format for embedding packing
+        blob = _get_struct_format(len(embedding)).pack(*embedding)
 
         # Delete existing and insert new (sqlite-vec doesn't support UPDATE well)
         cursor.execute("DELETE FROM vec_memories WHERE id = ?", (memory_id,))
@@ -984,10 +1001,8 @@ def search_vector(
             List of (Memory, distance) tuples sorted by distance ascending.
             Lower distance means more similar.
         """
-        import struct
-
-        # Pack query embedding as binary
-        blob = struct.pack(f"{len(query_embedding)}f", *query_embedding)
+        # PERF-007: Use cached struct format for embedding packing
+        blob = _get_struct_format(len(query_embedding)).pack(*query_embedding)
 
         with self._cursor() as cursor:
             try: