diff --git a/api/streaming.py b/api/streaming.py
index 7ef3dd99..540b44bb 100644
--- a/api/streaming.py
+++ b/api/streaming.py
@@ -10,7 +10,9 @@ import logging
 import mimetypes
 import os
 import queue
+import random
 import re
+import sqlite3
 import shlex
 import sys
 import subprocess
@@ -6333,7 +6335,25 @@ def _build_session_db_for_stream(state_db_path):
     """
     try:
         from hermes_state import SessionDB
-        return SessionDB(db_path=state_db_path)
+        _attempts = 3
+        _last_error = None
+        for _attempt in range(_attempts):
+            try:
+                return SessionDB(db_path=state_db_path)
+            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(
+                        f"[webui] WARNING: SessionDB init attempt {_attempt + 1}/{_attempts} failed, retrying: {_db_err}",
+                        flush=True,
+                    )
+                    time.sleep(0.05 * (2 ** _attempt) + random.uniform(0, 0.05))
+        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
diff --git a/tests/test_issue3548_sessiondb_self_heal.py b/tests/test_issue3548_sessiondb_self_heal.py
index 9241eb08..b79470d0 100644
--- a/tests/test_issue3548_sessiondb_self_heal.py
+++ b/tests/test_issue3548_sessiondb_self_heal.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import sqlite3
 import sys
 import types
 from unittest import mock
@@ -22,28 +23,118 @@ def test_session_db_helper_uses_request_state_db_path():
     fake_state = types.ModuleType("hermes_state")
     fake_state.SessionDB = FakeSessionDB
 
-    with mock.patch.dict(sys.modules, {"hermes_state": fake_state}):
+    with (
+        mock.patch.dict(sys.modules, {"hermes_state": fake_state}),
+        mock.patch.object(streaming.time, "sleep") as sleep,
+    ):
         state_db_path = Path("/tmp/profile") / "state.db"
         db = streaming._build_session_db_for_stream(state_db_path)
 
     assert db is not None
     assert calls["db_path"] == state_db_path
     assert isinstance(db, FakeSessionDB)
+    sleep.assert_not_called()
 
 
-def test_session_db_helper_returns_none_when_constructor_fails():
+def test_session_db_helper_retries_transient_constructor_failure():
     import api.streaming as streaming
 
-    def failing_session_db(db_path=None):
-        raise RuntimeError("SessionDB unavailable")
-
+    state_db_path = Path("/tmp/profile/state.db")
+    created = mock.Mock(name="session_db")
     fake_state = types.ModuleType("hermes_state")
-    fake_state.SessionDB = mock.Mock(side_effect=failing_session_db)
+    fake_random = mock.Mock()
+    fake_random.uniform.return_value = 0.0
+    fake_state.SessionDB = mock.Mock(
+        side_effect=[
+            sqlite3.OperationalError("database is locked"),
+            sqlite3.OperationalError("database is busy"),
+            created,
+        ]
+    )
 
-    with mock.patch.dict(sys.modules, {"hermes_state": fake_state}):
-        db = streaming._build_session_db_for_stream(Path("/tmp/profile/state.db"))
+    with (
+        mock.patch.dict(sys.modules, {"hermes_state": fake_state}),
+        mock.patch.object(streaming, "random", fake_random, create=True),
+        mock.patch.object(streaming.time, "sleep") as sleep,
+    ):
+        db = streaming._build_session_db_for_stream(state_db_path)
+
+    assert db is created
+    assert fake_state.SessionDB.call_args_list == [
+        mock.call(db_path=state_db_path),
+        mock.call(db_path=state_db_path),
+        mock.call(db_path=state_db_path),
+    ]
+    assert sleep.call_args_list == [mock.call(0.05), mock.call(0.1)]
+
+
+def test_session_db_helper_returns_none_after_exhausted_retries():
+    import api.streaming as streaming
+
+    state_db_path = Path("/tmp/profile/state.db")
+    fake_state = types.ModuleType("hermes_state")
+    fake_random = mock.Mock()
+    fake_random.uniform.return_value = 0.0
+    fake_state.SessionDB = mock.Mock(
+        side_effect=sqlite3.OperationalError("database is locked")
+    )
+
+    with (
+        mock.patch.dict(sys.modules, {"hermes_state": fake_state}),
+        mock.patch.object(streaming, "random", fake_random, create=True),
+        mock.patch.object(streaming.time, "sleep") as sleep,
+    ):
+        db = streaming._build_session_db_for_stream(state_db_path)
 
     assert db is None
+    assert fake_state.SessionDB.call_args_list == [
+        mock.call(db_path=state_db_path),
+        mock.call(db_path=state_db_path),
+        mock.call(db_path=state_db_path),
+    ]
+    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():
diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py
index 17a28ddc..a0930603 100644
--- a/tests/test_sprint42.py
+++ b/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(