fix(hindsight): guard client lifecycle with a leaf lock

plugins/memory/hindsight/__init__.py

@@ -329,6 +329,15 @@ def backup_paths(self) -> List[str]:
 
     def __init__(self):
         self._config = self._api_key = self._client = None
+        # LEAF lock guarding _client construction/retirement (#11923): two threads
+        # racing a cold or just-nulled client would both build one, and the loser
+        # owns an aiohttp session nothing ever closes. Never held across
+        # _run_sync/operation(client) or while taking _prefetch_lock /
+        # _pending_retain_ops_lock.
+        self._client_lock = threading.Lock()
+        # Client observed broken by the stale-daemon retry; only its observer may
+        # retire it (identity-checked against _client under _client_lock).
+        self._broken_client = None
         self._api_url, self._llm_base_url, self._mode = _DEFAULT_API_URL, "", "cloud"
         self._timeout, self._idle_timeout = _DEFAULT_TIMEOUT, _DEFAULT_IDLE_TIMEOUT
         self._bank_id, self._budget, self._bank_id_template = "hermes", "mid", ""
@@ -367,6 +376,10 @@ def __init__(self):
         self._prefetch_result, self._prefetch_count = "", 0
         self._prefetch_lock = threading.Lock()
         self._prefetch_thread = None
+        # Prefetch-slot ownership (guarded by _prefetch_lock): bumped per spawned
+        # worker and on session switch/shutdown, so a late or superseded worker
+        # whose captured generation is stale can never publish into the slot.
+        self._prefetch_generation = 0
         self._last_recall_returned, self._last_recall_count = False, 0
         self._apply_recall_settings({})
 
@@ -497,11 +510,25 @@ def _new_cloud_client(self):
                      self._api_url, bool(self._api_key), kwargs["timeout"])
         return Hindsight(**kwargs)
 
-    def _get_client(self):
-        """Return the cached Hindsight client (created once, reused)."""
-        if self._client is None:
-            self._client = self._new_embedded_client() if self._mode == "local_embedded" else self._new_cloud_client()
-        return self._client
+    def _get_client(self, *, recreate: bool = False):
+        """Return the cached Hindsight client (created once, reused).
+
+        *recreate* is used by the stale-daemon retry path: it drops the cached
+        client (if it is still the broken one we observed) and rebuilds it
+        exactly once, under the lock.
+        """
+        if not recreate and (client := self._client) is not None:
+            return client
+        with self._client_lock:
+            if recreate:
+                # Only retire the client we actually observed as broken; a sibling
+                # thread may have already rebuilt it after our failure.
+                if self._client is not None and self._client is not getattr(self, "_broken_client", None):
+                    return self._client
+                self._client = None
+            if self._client is None:
+                self._client = self._new_embedded_client() if self._mode == "local_embedded" else self._new_cloud_client()
+            return self._client
 
     def _run_sync(self, coro):
         """Schedule *coro* on the shared loop using the configured timeout."""
@@ -517,8 +544,9 @@ def _run_hindsight_operation(self, operation):
             if self._mode != "local_embedded" or not any(m in text for m in _RETRIABLE_CONNECTION_MARKERS):
                 raise
             logger.info("Hindsight embedded daemon appears unreachable; recreating client and retrying once: %s", exc)
-            self._client = None
-            self._client = client = self._get_client()
+            self._broken_client = self._client
+            self._client = client = self._get_client(recreate=True)
+            self._broken_client = None
             return self._run_sync(operation(client))
 
     # -- retain writer thread + server-side visibility -------------------------
@@ -963,15 +991,36 @@ def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
         if self._recall_sync or self._recall_disabled():
             return
 
+        # One worker at a time (mirrors _ensure_writer): rapid turns must not
+        # stack concurrent recalls against one embedded daemon — the LAST
+        # finisher would win the slot, not the newest warm.
+        if self._prefetch_thread is not None and self._prefetch_thread.is_alive():
+            logger.debug("Prefetch: prior worker still running; skipping this warm")
+            return
+
+        with self._prefetch_lock:
+            self._prefetch_generation += 1
+            generation = self._prefetch_generation
+
         def _run():
             # Wait (bounded, off the reply path) for the just-completed turn's
             # retain to be recall-visible so the warmed context includes it.
             if self._prefetch_waits_for_retain:
                 self._wait_for_retains_drained(self._prefetch_retain_drain_timeout)
+            # A superseded/shut-down worker must not even spend a recall
+            # against the daemon on a result nobody will use.
+            with self._prefetch_lock:
+                if generation != self._prefetch_generation or self._shutting_down.is_set():
+                    return
             text, count = self._do_recall(query)
             if text:
+                # Fenced publish: the slot can outlive the switching session's
+                # 3s join (drain up to 10s + recall up to 120s), so only the
+                # generation that owns the slot may write it — a stale worker
+                # must not inject the old session's memories into the new one.
                 with self._prefetch_lock:
-                    self._prefetch_result, self._prefetch_count = text, count
+                    if generation == self._prefetch_generation and not self._shutting_down.is_set():
+                        self._prefetch_result, self._prefetch_count = text, count
 
         self._prefetch_thread = spawn_context_thread(_run, name="hindsight-prefetch")
         self._prefetch_thread.start()
@@ -1193,6 +1242,9 @@ def _flush():
         # 2. Drain the old session's in-flight prefetch and drop its result.
         self._join_prefetch(3.0)
         with self._prefetch_lock:
+            # Fence out any worker still running past the join: the slot now
+            # belongs to the new session.
+            self._prefetch_generation += 1
             self._prefetch_result = ""
 
         # 3. Rotate to the new session.
@@ -1204,25 +1256,29 @@ def _flush():
         logger.debug("Hindsight on_session_switch: new_session=%s parent=%s reset=%s doc=%s",
                      self._session_id, self._parent_session_id, reset, self._document_id)
 
-    def _close_client(self) -> None:
+    def _close_client_of(self, client) -> None:
         if self._mode != "local_embedded":
-            self._run_sync(self._client.aclose())
+            self._run_sync(client.aclose())
             return
         # HindsightEmbedded.close() closes its sync client from this thread ("attached
         # to a different loop" before aiohttp releases the session): aclose the inner
         # client on the shared loop first, then let the wrapper clean up bookkeeping.
-        inner_client = getattr(self._client, "_client", None)
+        inner_client = getattr(client, "_client", None)
         if inner_client is not None and hasattr(inner_client, "aclose"):
             _run_sync(inner_client.aclose())
             with contextlib.suppress(Exception):
-                self._client._client = None
+                client._client = None
         with contextlib.suppress(RuntimeError):
-            self._client.close()
+            client.close()
 
     def shutdown(self) -> None:
         logger.debug("Hindsight shutdown: stopping writer + waiting for background threads")
         # Stop accepting retain jobs first so late sync_turn() calls are dropped.
         self._shutting_down.set()
+        # Fence any in-flight prefetch worker: a recall finishing during or
+        # after the joins below must never publish into a closing provider.
+        with self._prefetch_lock:
+            self._prefetch_generation += 1
         # The writer finishes in-flight work then exits on the sentinel; the
         # bounded join keeps shutdown predictable even if the daemon is wedged.
         if (writer := self._writer_thread) is not None and writer.is_alive():
@@ -1232,10 +1288,14 @@ def shutdown(self) -> None:
                 logger.warning("Hindsight writer did not stop within 10s; abandoning %d pending retain(s)",
                                self._retain_queue.qsize())
         self._join_prefetch(5.0)
-        if self._client is not None:
-            with contextlib.suppress(Exception):
-                self._close_client()
+        # Retire the client under the lock BEFORE closing it, so a concurrent
+        # _get_client() sees None and rebuilds instead of racing the close.
+        with self._client_lock:
+            client = self._client
             self._client = None
+        if client is not None:
+            with contextlib.suppress(Exception):
+                self._close_client_of(client)
         # The module-global loop is intentionally NOT stopped: it's shared by every
         # provider in the process (one per gateway chat session); stopping it would
         # strand siblings' aiohttp sessions ("Unclosed client session"). Daemon