feat: refactor hindsight-embed architecture (#66)

* feat: refactor hindsight-embed architecture

* feat: refactor hindsight-embed architecture

* refactor deamin

* refactor deamin

* refactor deamin

* refactor deamin

Author: nicoloboschi

hindsight-api/hindsight_api/config.py

@@ -33,6 +33,10 @@
 ENV_MCP_LOCAL_BANK_ID = "HINDSIGHT_API_MCP_LOCAL_BANK_ID"
 ENV_MCP_INSTRUCTIONS = "HINDSIGHT_API_MCP_INSTRUCTIONS"
 
+# Optimization flags
+ENV_SKIP_LLM_VERIFICATION = "HINDSIGHT_API_SKIP_LLM_VERIFICATION"
+ENV_LAZY_RERANKER = "HINDSIGHT_API_LAZY_RERANKER"
+
 # Default values
 DEFAULT_DATABASE_URL = "pg0"
 DEFAULT_LLM_PROVIDER = "openai"
@@ -107,6 +111,10 @@ class HindsightConfig:
     # Recall
     graph_retriever: str
 
+    # Optimization flags
+    skip_llm_verification: bool
+    lazy_reranker: bool
+
     @classmethod
     def from_env(cls) -> "HindsightConfig":
         """Create configuration from environment variables."""
@@ -133,6 +141,9 @@ def from_env(cls) -> "HindsightConfig":
             mcp_enabled=os.getenv(ENV_MCP_ENABLED, str(DEFAULT_MCP_ENABLED)).lower() == "true",
             # Recall
             graph_retriever=os.getenv(ENV_GRAPH_RETRIEVER, DEFAULT_GRAPH_RETRIEVER),
+            # Optimization flags
+            skip_llm_verification=os.getenv(ENV_SKIP_LLM_VERIFICATION, "false").lower() == "true",
+            lazy_reranker=os.getenv(ENV_LAZY_RERANKER, "false").lower() == "true",
         )
 
     def get_llm_base_url(self) -> str:

hindsight-api/hindsight_api/daemon.py

@@ -0,0 +1,204 @@
+"""
+Daemon mode support for Hindsight API.
+
+Provides idle timeout and lockfile management for running as a background daemon.
+"""
+
+import asyncio
+import fcntl
+import logging
+import os
+import sys
+import time
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# Default daemon configuration
+DEFAULT_DAEMON_PORT = 8889
+DEFAULT_IDLE_TIMEOUT = 0  # 0 = no auto-exit (hindsight-embed passes its own timeout)
+LOCKFILE_PATH = Path.home() / ".hindsight" / "daemon.lock"
+DAEMON_LOG_PATH = Path.home() / ".hindsight" / "daemon.log"
+
+
+class IdleTimeoutMiddleware:
+    """ASGI middleware that tracks activity and exits after idle timeout."""
+
+    def __init__(self, app, idle_timeout: int = DEFAULT_IDLE_TIMEOUT):
+        self.app = app
+        self.idle_timeout = idle_timeout
+        self.last_activity = time.time()
+        self._checker_task = None
+
+    async def __call__(self, scope, receive, send):
+        # Update activity timestamp on each request
+        self.last_activity = time.time()
+        await self.app(scope, receive, send)
+
+    def start_idle_checker(self):
+        """Start the background task that checks for idle timeout."""
+        self._checker_task = asyncio.create_task(self._check_idle())
+
+    async def _check_idle(self):
+        """Background task that exits the process after idle timeout."""
+        # If idle_timeout is 0, don't auto-exit
+        if self.idle_timeout <= 0:
+            return
+
+        while True:
+            await asyncio.sleep(30)  # Check every 30 seconds
+            idle_time = time.time() - self.last_activity
+            if idle_time > self.idle_timeout:
+                logger.info(f"Idle timeout reached ({self.idle_timeout}s), shutting down daemon")
+                # Give a moment for any in-flight requests
+                await asyncio.sleep(1)
+                os._exit(0)
+
+
+class DaemonLock:
+    """
+    File-based lock to prevent multiple daemon instances.
+
+    Uses fcntl.flock for atomic locking on Unix systems.
+    """
+
+    def __init__(self, lockfile: Path = LOCKFILE_PATH):
+        self.lockfile = lockfile
+        self._fd = None
+
+    def acquire(self) -> bool:
+        """
+        Try to acquire the daemon lock.
+
+        Returns True if lock acquired, False if another daemon is running.
+        """
+        self.lockfile.parent.mkdir(parents=True, exist_ok=True)
+
+        try:
+            self._fd = open(self.lockfile, "w")
+            fcntl.flock(self._fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            # Write PID for debugging
+            self._fd.write(str(os.getpid()))
+            self._fd.flush()
+            return True
+        except (IOError, OSError):
+            # Lock is held by another process
+            if self._fd:
+                self._fd.close()
+                self._fd = None
+            return False
+
+    def release(self):
+        """Release the daemon lock."""
+        if self._fd:
+            try:
+                fcntl.flock(self._fd.fileno(), fcntl.LOCK_UN)
+                self._fd.close()
+            except Exception:
+                pass
+            finally:
+                self._fd = None
+                # Remove lockfile
+                try:
+                    self.lockfile.unlink()
+                except Exception:
+                    pass
+
+    def is_locked(self) -> bool:
+        """Check if the lock is held by another process."""
+        if not self.lockfile.exists():
+            return False
+
+        try:
+            fd = open(self.lockfile, "r")
+            fcntl.flock(fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            # We got the lock, so no one else has it
+            fcntl.flock(fd.fileno(), fcntl.LOCK_UN)
+            fd.close()
+            return False
+        except (IOError, OSError):
+            return True
+
+    def get_pid(self) -> int | None:
+        """Get the PID of the daemon holding the lock."""
+        if not self.lockfile.exists():
+            return None
+        try:
+            with open(self.lockfile, "r") as f:
+                return int(f.read().strip())
+        except (ValueError, IOError):
+            return None
+
+
+def daemonize():
+    """
+    Fork the current process into a background daemon.
+
+    Uses double-fork technique to properly detach from terminal.
+    """
+    # First fork
+    pid = os.fork()
+    if pid > 0:
+        # Parent exits
+        sys.exit(0)
+
+    # Create new session
+    os.setsid()
+
+    # Second fork to prevent zombie processes
+    pid = os.fork()
+    if pid > 0:
+        sys.exit(0)
+
+    # Redirect standard file descriptors to log file
+    DAEMON_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+    sys.stdout.flush()
+    sys.stderr.flush()
+
+    # Redirect stdin to /dev/null
+    with open("/dev/null", "r") as devnull:
+        os.dup2(devnull.fileno(), sys.stdin.fileno())
+
+    # Redirect stdout/stderr to log file
+    log_fd = open(DAEMON_LOG_PATH, "a")
+    os.dup2(log_fd.fileno(), sys.stdout.fileno())
+    os.dup2(log_fd.fileno(), sys.stderr.fileno())
+
+
+def check_daemon_running(port: int = DEFAULT_DAEMON_PORT) -> bool:
+    """Check if a daemon is running and responsive on the given port."""
+    import socket
+
+    try:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.settimeout(1)
+        result = sock.connect_ex(("127.0.0.1", port))
+        sock.close()
+        return result == 0
+    except Exception:
+        return False
+
+
+def stop_daemon(port: int = DEFAULT_DAEMON_PORT) -> bool:
+    """Stop a running daemon by sending SIGTERM to the process."""
+    lock = DaemonLock()
+    pid = lock.get_pid()
+
+    if pid is None:
+        return False
+
+    try:
+        import signal
+
+        os.kill(pid, signal.SIGTERM)
+        # Wait for process to exit
+        for _ in range(50):  # Wait up to 5 seconds
+            time.sleep(0.1)
+            try:
+                os.kill(pid, 0)  # Check if process exists
+            except OSError:
+                return True  # Process exited
+        return False
+    except OSError:
+        return False

hindsight-api/hindsight_api/engine/memory_engine.py

@@ -202,6 +202,8 @@ def __init__(
         run_migrations: bool = True,
         operation_validator: "OperationValidatorExtension | None" = None,
         tenant_extension: "TenantExtension | None" = None,
+        skip_llm_verification: bool | None = None,
+        lazy_reranker: bool | None = None,
     ):
         """
         Initialize the temporal + semantic memory system.
@@ -227,12 +229,23 @@ def __init__(
                                 If provided, retain/recall/reflect operations will be validated.
             tenant_extension: Optional extension for multi-tenancy and API key authentication.
                              If provided, operations require a RequestContext for authentication.
+            skip_llm_verification: Skip LLM connection verification during initialization.
+                                  Defaults to HINDSIGHT_API_SKIP_LLM_VERIFICATION env var or False.
+            lazy_reranker: Delay reranker initialization until first use. Useful for retain-only
+                          operations that don't need the cross-encoder. Defaults to
+                          HINDSIGHT_API_LAZY_RERANKER env var or False.
         """
         # Load config from environment for any missing parameters
         from ..config import get_config
 
         config = get_config()
 
+        # Apply optimization flags from config if not explicitly provided
+        self._skip_llm_verification = (
+            skip_llm_verification if skip_llm_verification is not None else config.skip_llm_verification
+        )
+        self._lazy_reranker = lazy_reranker if lazy_reranker is not None else config.lazy_reranker
+
         # Apply defaults from config
         db_url = db_url or config.database_url
         memory_llm_provider = memory_llm_provider or config.llm_provider
@@ -592,6 +605,8 @@ async def init_cross_encoder():
                 await loop.run_in_executor(None, lambda: asyncio.run(cross_encoder.initialize()))
             else:
                 await cross_encoder.initialize()
+            # Mark reranker as initialized
+            self._cross_encoder_reranker._initialized = True
 
         async def init_query_analyzer():
             """Initialize query analyzer model."""
@@ -600,16 +615,26 @@ async def init_query_analyzer():
 
         async def verify_llm():
             """Verify LLM connection is working."""
-            await self._llm_config.verify_connection()
+            if not self._skip_llm_verification:
+                await self._llm_config.verify_connection()
 
-        # Run pg0 and all model initializations in parallel
-        await asyncio.gather(
+        # Build list of initialization tasks
+        init_tasks = [
             start_pg0(),
             init_embeddings(),
-            init_cross_encoder(),
             init_query_analyzer(),
-            verify_llm(),
-        )
+        ]
+
+        # Only init cross-encoder eagerly if not using lazy initialization
+        if not self._lazy_reranker:
+            init_tasks.append(init_cross_encoder())
+
+        # Only verify LLM if not skipping
+        if not self._skip_llm_verification:
+            init_tasks.append(verify_llm())
+
+        # Run pg0 and selected model initializations in parallel
+        await asyncio.gather(*init_tasks)
 
         # Run database migrations if enabled
         if self._run_migrations:
@@ -1639,6 +1664,9 @@ def to_tuple_format(results):
             step_start = time.time()
             reranker_instance = self._cross_encoder_reranker
 
+            # Ensure reranker is initialized (for lazy initialization mode)
+            await reranker_instance.ensure_initialized()
+
             # Rerank using cross-encoder
             scored_results = reranker_instance.rerank(query, merged_candidates)
 

hindsight-api/hindsight_api/engine/search/reranking.py

@@ -26,6 +26,23 @@ def __init__(self, cross_encoder=None):
 
             cross_encoder = create_cross_encoder_from_env()
         self.cross_encoder = cross_encoder
+        self._initialized = False
+
+    async def ensure_initialized(self):
+        """Ensure the cross-encoder model is initialized (for lazy initialization)."""
+        if self._initialized:
+            return
+
+        import asyncio
+
+        cross_encoder = self.cross_encoder
+        # For local providers, run in thread pool to avoid blocking event loop
+        if cross_encoder.provider_name == "local":
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, lambda: asyncio.run(cross_encoder.initialize()))
+        else:
+            await cross_encoder.initialize()
+        self._initialized = True
 
     def rerank(self, query: str, candidates: list[MergedCandidate]) -> list[ScoredResult]:
         """