fix: resolve DuckDB lock conflicts with guard, read-only connections, and degraded mode

Three-part fix for the recurring duckdb.IOException when multiple processes
try to open trader.duckdb in write mode:

1. Lock guard (db.py): _connect_with_lock_guard() distinguishes stale locks
   (dead PID → retry 10 s) from live conflicts (alive PID → immediate
   RuntimeError with exact `kill <PID>` command).  open_db_readonly() added
   for processes that only need reads — multiple processes can hold read-only
   connections simultaneously without conflicting with the write owner.

2. MCP server degraded mode (mcp/server.py): lifespan() catches lock
   RuntimeError and falls back to :memory: context.  Analysis tools
   (run_analysis, get_regime, get_system_status) keep working; 22 DB-backed
   tools return {"success": false, "degraded_mode": true} instead of crashing
   the Claude session.

3. FastAPI read-only (api/server.py): all GET endpoints now use
   get_portfolio_state_readonly() and get_decision_log_readonly() — the API
   server no longer competes with the MCP server for the write lock.

Also adds get_portfolio_state_readonly() (portfolio_state.py) and
get_decision_log_readonly() (decision_log.py) singletons, 12 new tests
in tests/quant_pod/test_db_lock_guard.py (all pass), and updates
packages/quant_pod/README.md with the multi-process connection model.

Author: kbichave

packages/quant_pod/README.md

@@ -103,6 +103,23 @@ All runtime state lives in a single ACID-safe DuckDB file. No scattered pickle f
 | `strategies` | Strategy registry |
 | `regime_strategy_matrix` | Regime → allocation weights |
 
+### Multi-process access model
+
+DuckDB enforces one write connection per file across OS processes.
+
+| Process | Connection type | Function |
+|---------|----------------|---------|
+| **MCP server** | Write (owner) | `open_db()` — runs migrations, holds lock |
+| **FastAPI server** | Read-only | `open_db_readonly()` — GET endpoints only |
+| **Scripts / scheduler** | Read-only | `open_db_readonly()` |
+
+**Lock conflict handling** (`packages/quant_pod/db.py`):
+- *Stale lock* (owning process died): retries for 10 s, then raises with a `.wal` cleanup hint
+- *Live conflict* (owner is alive): raises `RuntimeError` immediately with `kill <PID>` command
+- *Degraded mode*: if the MCP server encounters a lock on startup, it falls back to an in-memory
+  context. Analysis tools (`run_analysis`, `get_regime`) still work; portfolio/execution tools
+  return `{"success": false, "degraded_mode": true}`
+
 ## Regime Detection
 
 `regime_detector_ic` uses real ADX and ATR indicators — no LLMs, no stubs:

packages/quant_pod/api/server.py

@@ -44,11 +44,15 @@
 from loguru import logger
 from pydantic import BaseModel
 
-from quant_pod.audit.decision_log import extract_indicator_attributions, get_decision_log
+from quant_pod.audit.decision_log import (
+    extract_indicator_attributions,
+    get_decision_log,
+    get_decision_log_readonly,
+)
 from quant_pod.audit.models import AuditQuery
 from quant_pod.execution.broker_factory import get_broker, get_broker_mode
 from quant_pod.execution.kill_switch import get_kill_switch
-from quant_pod.execution.portfolio_state import get_portfolio_state
+from quant_pod.execution.portfolio_state import get_portfolio_state, get_portfolio_state_readonly
 from quant_pod.learning.calibration import get_calibration_tracker
 from quant_pod.monitoring.metrics import (
     get_metrics_content_type,
@@ -110,7 +114,7 @@ class ResetRequest(BaseModel):
 def health() -> dict[str, Any]:
     """Health check — returns service status and kill switch state."""
     kill = get_kill_switch()
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     snapshot = portfolio.get_snapshot()
 
     return {
@@ -127,15 +131,15 @@ def health() -> dict[str, Any]:
 @app.get("/portfolio")
 def get_portfolio() -> dict[str, Any]:
     """Current portfolio snapshot."""
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     snapshot = portfolio.get_snapshot()
     return snapshot.model_dump()
 
 
 @app.get("/portfolio/positions")
 def get_positions() -> list[dict[str, Any]]:
     """Current open positions."""
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     return [p.model_dump() for p in portfolio.get_positions()]
 
 
@@ -219,7 +223,7 @@ def get_audit(
     limit: int = Query(default=50, le=500),
 ) -> list[dict[str, Any]]:
     """Recent audit log entries."""
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     events = log.query(
         AuditQuery(
             symbol=symbol,
@@ -247,7 +251,7 @@ def get_audit(
 @app.get("/audit/{event_id}/trace")
 def get_audit_trace(event_id: str) -> list[dict[str, Any]]:
     """Full decision trace for a specific event."""
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     trace = log.get_decision_trace(event_id)
     if not trace:
         raise HTTPException(status_code=404, detail=f"Event {event_id} not found")
@@ -264,7 +268,7 @@ def get_audit_attribution(event_id: str) -> dict[str, Any]:
       - Derived attributions from the market_data_snapshot (as fallback)
       - IC dissent signals (ICs that disagreed with the consensus)
     """
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     log.query(AuditQuery(limit=1))  # Query by event_id directly
     # query() doesn't filter by event_id, so fetch via trace (single-element trace = the event itself)
     trace = log.get_decision_trace(event_id)
@@ -295,7 +299,7 @@ def get_audit_attribution(event_id: str) -> dict[str, Any]:
 @app.get("/audit/session/{session_id}/summary")
 def get_session_summary(session_id: str) -> dict[str, Any]:
     """High-level summary of a trading session."""
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     return log.get_session_summary(session_id)
 
 
@@ -438,7 +442,7 @@ def get_heartbeat() -> dict[str, Any]:
 
     Status: "ok" | "stale" | "never_seen"
     """
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     now = datetime.now()
     max_silence_hours = 25  # One session = one trading day + buffer
 
@@ -499,7 +503,7 @@ def get_pnl_dashboard() -> dict[str, Any]:
       - total_equity: cash + positions_value
       - recent_trades: last 10 closed trades with P&L
     """
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     snapshot = portfolio.get_snapshot()
     positions = portfolio.get_positions()
 
@@ -549,7 +553,7 @@ def get_anomalies() -> dict[str, Any]:
     """
     anomalies: list[dict[str, Any]] = []
     broker = get_broker()
-    log = get_decision_log()
+    log = get_decision_log_readonly()
     now = datetime.now()
 
     # ---- 1. Order size anomalies ----------------------------------------
@@ -579,7 +583,7 @@ def get_anomalies() -> dict[str, Any]:
                 )
 
     # ---- 2. Win rate degradation (uses ClosedTrade records, not fills) -----
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     recent_closed = portfolio.conn.execute(
         "SELECT realized_pnl FROM closed_trades ORDER BY closed_at DESC LIMIT 20"
     ).fetchall()
@@ -989,7 +993,7 @@ def prometheus_metrics() -> str:
     Also updates NAV and daily P&L gauges on each scrape so they stay current
     without requiring a dedicated background task.
     """
-    portfolio = get_portfolio_state()
+    portfolio = get_portfolio_state_readonly()
     snap = portfolio.get_snapshot()
     record_nav(snap.total_equity)
     record_daily_pnl(snap.daily_pnl)

packages/quant_pod/audit/decision_log.py

@@ -482,8 +482,34 @@ def _attr(
 
 
 def get_decision_log(db_path: str | None = None) -> DecisionLog:
-    """Get the singleton DecisionLog instance."""
+    """Get the singleton DecisionLog instance (write connection)."""
     global _decision_log
     if _decision_log is None:
         _decision_log = DecisionLog(db_path=db_path)
     return _decision_log
+
+
+# Read-only singleton — for processes that must not compete for the write lock.
+_decision_log_ro: DecisionLog | None = None
+
+
+def get_decision_log_readonly() -> DecisionLog:
+    """
+    Get a read-only DecisionLog singleton.
+
+    Use this in processes (FastAPI, scripts) that run alongside the MCP server
+    and only need to QUERY the audit trail.  The returned instance cannot
+    record new events — those calls will raise at runtime.
+
+    DecisionLog.__init__ already skips schema creation when a connection is
+    injected, so no DDL is attempted on the read-only connection.
+
+    Raises:
+        FileNotFoundError: if trader.duckdb doesn't exist yet (MCP server not started).
+    """
+    global _decision_log_ro
+    if _decision_log_ro is None:
+        from quant_pod.db import open_db_readonly
+
+        _decision_log_ro = DecisionLog(conn=open_db_readonly())
+    return _decision_log_ro

packages/quant_pod/db.py

@@ -12,12 +12,38 @@
 Schema ownership lives here.  Services receive an injected connection;
 they do NOT open their own files.
 
+## Connection Model
+
+DuckDB allows only ONE write connection per file across OS processes.
+The MCP server is the canonical write owner — it holds the connection for
+its entire lifetime.  Other processes (FastAPI, scripts) must connect
+read-only via open_db_readonly().
+
+### Lock conflict handling
+
+open_db() wraps duckdb.connect() with a lock guard (_connect_with_lock_guard):
+
+  - Stale lock (owning process is dead): retries for up to
+    _STALE_LOCK_RETRY_SECS (10 s) until the OS releases the lock.
+
+  - Live conflict (owning process is alive): raises RuntimeError immediately
+    with the PID and the exact `kill PID` command to resolve it.
+
+The MCP server's lifespan() catches this RuntimeError and falls back to an
+in-memory context (degraded mode) so the Claude session is never crashed by
+a lock conflict.  Analysis tools keep working; tools that need persistent
+state return {"success": False, "degraded_mode": True}.
+
 Usage:
-    # Production
+    # Production (MCP server — write owner)
     from quant_pod.db import open_db, run_migrations
     conn = open_db("~/.quant_pod/trader.duckdb")
     run_migrations(conn)
 
+    # FastAPI / scripts — read-only, no lock competition
+    from quant_pod.db import open_db_readonly
+    conn = open_db_readonly()   # FileNotFoundError if MCP server not started
+
     # Tests — fully isolated in-memory DB
     conn = open_db(":memory:")
     run_migrations(conn)
@@ -156,6 +182,72 @@ def reset_connection() -> None:
             _conn = None
 
 
+# ---------------------------------------------------------------------------
+# Read-only connection — for processes that must not compete for the write lock
+# ---------------------------------------------------------------------------
+
+_conn_ro: duckdb.DuckDBPyConnection | None = None
+_conn_ro_lock = Lock()
+
+
+def open_db_readonly(path: str = "") -> duckdb.DuckDBPyConnection:
+    """
+    Open (or return a cached) read-only DuckDB connection.
+
+    Multiple processes can hold read-only connections simultaneously without
+    conflicting with the write owner (MCP server).  Does NOT run migrations —
+    the write owner is always responsible for schema.
+
+    For ':memory:' paths (test compat) falls back to the regular write
+    connection so test code that calls this function works identically to
+    production code that calls open_db().
+
+    Raises:
+        FileNotFoundError: if the DB file does not exist.  The write owner
+                           (MCP server) must be started first.
+    """
+    global _conn_ro
+
+    if not path:
+        path = os.getenv("TRADER_DB_PATH", "~/.quant_pod/trader.duckdb")
+
+    if path == ":memory:":
+        # Tests use :memory: — read-only has no meaning there; reuse write conn.
+        return open_db(path)
+
+    resolved = Path(path).expanduser()
+    if not resolved.exists():
+        raise FileNotFoundError(
+            f"DB not found at {resolved}. "
+            "The MCP server must be started (and have run migrations) before "
+            "read-only consumers can connect."
+        )
+
+    path = str(resolved)
+    with _conn_ro_lock:
+        if _conn_ro is None:
+            _conn_ro = duckdb.connect(path, read_only=True)
+            logger.info(f"[DB] Opened read-only connection at {path}")
+        return _conn_ro
+
+
+def reset_connection_readonly() -> None:
+    """
+    Close and clear the cached read-only connection.
+
+    Call this in tests after any test that opens a file-backed read-only
+    connection so the cached singleton doesn't bleed across tests.
+    """
+    global _conn_ro
+    with _conn_ro_lock:
+        if _conn_ro is not None:
+            try:
+                _conn_ro.close()
+            except Exception:
+                pass
+            _conn_ro = None
+
+
 # ---------------------------------------------------------------------------
 # Migrations — idempotent, append-only schema upgrades
 # ---------------------------------------------------------------------------

packages/quant_pod/execution/portfolio_state.py

@@ -110,10 +110,12 @@ def __init__(
         initial_cash: float = 100_000.0,
         # Legacy parameter kept for backward compatibility — ignored when conn is provided
         db_path: str | None = None,
+        read_only: bool = False,
     ):
         # RLock (reentrant) because upsert_position() calls get_position() under the lock
         self._lock = RLock()
         self._initial_cash = initial_cash
+        self._read_only = read_only
 
         if conn is not None:
             # Injected connection (preferred — supports consolidated DB and in-memory tests)
@@ -127,7 +129,10 @@ def __init__(
             self._conn = open_db(db_path)
             run_migrations(self._conn)
 
-        self._seed_cash()
+        # Skip seeding on read-only connections — the write owner seeds on first run,
+        # and DDL/INSERT would raise duckdb.InvalidInputException on a read-only conn.
+        if not self._read_only:
+            self._seed_cash()
         logger.info("PortfolioState initialized")
 
     # -------------------------------------------------------------------------
@@ -579,7 +584,7 @@ def get_portfolio_state(
     # Legacy parameter — ignored when conn is provided
     db_path: str | None = None,
 ) -> PortfolioState:
-    """Get the singleton PortfolioState instance."""
+    """Get the singleton PortfolioState instance (write connection)."""
     global _portfolio_state
     if _portfolio_state is None:
         if conn is None:
@@ -589,3 +594,28 @@ def get_portfolio_state(
             run_migrations(conn)
         _portfolio_state = PortfolioState(conn=conn, initial_cash=initial_cash)
     return _portfolio_state
+
+
+# Read-only singleton — for processes that must not compete for the write lock
+# (e.g. the FastAPI server's GET endpoints).
+_portfolio_state_ro: PortfolioState | None = None
+
+
+def get_portfolio_state_readonly() -> PortfolioState:
+    """
+    Get a read-only PortfolioState singleton.
+
+    Use this in processes (FastAPI, scripts) that run alongside the MCP server
+    and only need to READ portfolio data.  The returned instance cannot execute
+    writes (upsert_position, adjust_cash, etc.) — those calls will raise a
+    duckdb.InvalidInputException at runtime.
+
+    Raises:
+        FileNotFoundError: if trader.duckdb doesn't exist yet (MCP server not started).
+    """
+    global _portfolio_state_ro
+    if _portfolio_state_ro is None:
+        from quant_pod.db import open_db_readonly
+
+        _portfolio_state_ro = PortfolioState(conn=open_db_readonly(), read_only=True)
+    return _portfolio_state_ro