fix(#6108): bound SessionDB retries to transient sqlite contention

api/streaming.py

@@ -12,6 +12,7 @@ import os
 import queue
 import random
 import re
+import sqlite3
 import shlex
 import sys
 import subprocess
@@ -6339,7 +6340,12 @@ def _build_session_db_for_stream(state_db_path):
         for _attempt in range(_attempts):
             try:
                 return SessionDB(db_path=state_db_path)
-            except Exception as _db_err:
+            except sqlite3.OperationalError as _db_err:
+                _db_err_text = str(_db_err).lower()
+                if not (
+                    "locked" in _db_err_text or "busy" in _db_err_text
+                ):
+                    raise
                 _last_error = _db_err
                 if _attempt < _attempts - 1:
                     print(
@@ -6347,7 +6353,7 @@ def _build_session_db_for_stream(state_db_path):
                         flush=True,
                     )
                     time.sleep(0.05 * (2 ** _attempt) + random.uniform(0, 0.05))
-        raise _last_error
+        raise _last_error or RuntimeError("SessionDB construction exhausted all attempts")
     except Exception as _db_err:
         print(f"[webui] WARNING: SessionDB init failed - session_search will be unavailable: {_db_err}", flush=True)
         return None

tests/test_issue3548_sessiondb_self_heal.py

@@ -47,7 +47,7 @@ def test_session_db_helper_retries_transient_constructor_failure():
     fake_state.SessionDB = mock.Mock(
         side_effect=[
             sqlite3.OperationalError("database is locked"),
-            sqlite3.OperationalError("database is locked"),
+            sqlite3.OperationalError("database is busy"),
             created,
         ]
     )
@@ -95,6 +95,48 @@ def test_session_db_helper_returns_none_after_exhausted_retries():
     assert sleep.call_args_list == [mock.call(0.05), mock.call(0.1)]
 
 
+def test_session_db_helper_does_not_retry_permanent_constructor_failure():
+    import api.streaming as streaming
+
+    state_db_path = Path("/tmp/profile/state.db")
+    permanent_error = TypeError("unsupported SessionDB argument")
+    fake_state = types.ModuleType("hermes_state")
+    fake_state.SessionDB = mock.Mock(side_effect=permanent_error)
+
+    with (
+        mock.patch.dict(sys.modules, {"hermes_state": fake_state}),
+        mock.patch.object(streaming, "random", mock.Mock(), create=True) as random,
+        mock.patch.object(streaming.time, "sleep") as sleep,
+    ):
+        db = streaming._build_session_db_for_stream(state_db_path)
+
+    assert db is None
+    fake_state.SessionDB.assert_called_once_with(db_path=state_db_path)
+    random.uniform.assert_not_called()
+    sleep.assert_not_called()
+
+
+def test_session_db_helper_does_not_retry_noncontention_operational_error():
+    import api.streaming as streaming
+
+    state_db_path = Path("/tmp/profile/state.db")
+    permanent_error = sqlite3.OperationalError("locking protocol")
+    fake_state = types.ModuleType("hermes_state")
+    fake_state.SessionDB = mock.Mock(side_effect=permanent_error)
+
+    with (
+        mock.patch.dict(sys.modules, {"hermes_state": fake_state}),
+        mock.patch.object(streaming, "random", mock.Mock(), create=True) as random,
+        mock.patch.object(streaming.time, "sleep") as sleep,
+    ):
+        db = streaming._build_session_db_for_stream(state_db_path)
+
+    assert db is None
+    fake_state.SessionDB.assert_called_once_with(db_path=state_db_path)
+    random.uniform.assert_not_called()
+    sleep.assert_not_called()
+
+
 def test_self_heal_session_db_handle_is_replaced_safely():
     import api.streaming as streaming
 

tests/test_sprint42.py

@@ -71,6 +71,15 @@ class TestSessionDBInjection(unittest.TestCase):
             "SessionDB init helper must use try/except for non-fatal error handling",
         )
 
+    def test_sessiondb_retry_only_targets_transient_sqlite_errors(self):
+        """Permanent constructor errors must leave the retry loop immediately."""
+        helper_start = STREAMING_PY.find("def _build_session_db_for_stream")
+        helper_end = STREAMING_PY.find("\n\ndef _attempt_credential_self_heal", helper_start)
+        helper_src = STREAMING_PY[helper_start:helper_end]
+        self.assertIn("except sqlite3.OperationalError as _db_err", helper_src)
+        self.assertIn('"locked" in _db_err_text or "busy" in _db_err_text', helper_src)
+        self.assertIn("raise _last_error or RuntimeError", helper_src)
+
     def test_sessiondb_failure_logs_warning(self):
         """A failure initializing SessionDB must print a WARNING (not silently drop the error)."""
         self.assertIn(