fix(hindsight): implement on_session_end to flush buffered turns at session exit

plugins/memory/hindsight/__init__.py

@@ -1251,6 +1251,79 @@ def _sync():
         self._sync_thread = threading.Thread(target=_sync, daemon=True, name="hindsight-sync")
         self._sync_thread.start()
 
+    def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
+        """Flush any buffered turns to Hindsight at session end.
+
+        Called when the session ends (gateway restart, idle timeout, /reset
+        or /new). sync_turn() batches retention every N turns — turns
+        accumulated since the last boundary would otherwise be lost. This
+        hook forces a final synchronous retention of any remaining buffered
+        turns so that no conversation context is dropped on session exit.
+        """
+        if not self._auto_retain:
+            logger.debug("on_session_end: skipped (auto_retain disabled)")
+            return
+        if not self._client:
+            logger.debug("on_session_end: skipped (no client)")
+            return
+
+        # Wait for any in-flight async sync to finish
+        if self._sync_thread and self._sync_thread.is_alive():
+            logger.debug("on_session_end: waiting for in-flight sync")
+            self._sync_thread.join(timeout=10.0)
+
+        # How many turns have accumulated since the last batch boundary?
+        remaining = self._turn_counter % self._retain_every_n_turns
+        if remaining == 0 and self._turn_counter > 0:
+            # All turns already retained at the last boundary — nothing to flush.
+            logger.debug("on_session_end: all %d turns already retained (batch aligned)",
+                         self._turn_counter)
+            return
+        if remaining == 0 and self._turn_counter == 0:
+            logger.debug("on_session_end: no turns to retain")
+            return
+
+        # Take only the unbuffered remainder (last `remaining` turns)
+        turns_to_flush = self._session_turns[-remaining:] if remaining > 0 else []
+        if not turns_to_flush:
+            logger.debug("on_session_end: no buffered turns to flush")
+            return
+
+        logger.debug("on_session_end: flushing %d buffered turns (session total: %d)",
+                     len(turns_to_flush), self._turn_counter)
+        content = "[" + ",".join(turns_to_flush) + "]"
+
+        lineage_tags: list[str] = []
+        if self._session_id:
+            lineage_tags.append("session:{0}".format(self._session_id))
+        if self._parent_session_id:
+            lineage_tags.append("parent:{0}".format(self._parent_session_id))
+
+        try:
+            item = self._build_retain_kwargs(
+                content,
+                context=self._retain_context,
+                metadata=self._build_metadata(
+                    message_count=len(turns_to_flush) * 2,
+                    turn_index=self._turn_index,
+                ),
+                tags=lineage_tags or None,
+            )
+            item.pop("bank_id", None)
+            item.pop("retain_async", None)
+            self._run_hindsight_operation(
+                lambda client: client.aretain_batch(
+                    bank_id=self._bank_id,
+                    items=[item],
+                    document_id=self._document_id,
+                    retain_async=False,  # synchronous at session end — must complete
+                )
+            )
+            logger.debug("Hindsight on_session_end: flushed %d turns successfully",
+                         len(turns_to_flush))
+        except Exception as e:
+            logger.warning("Hindsight on_session_end flush failed: %s", e, exc_info=True)
+
     def get_tool_schemas(self) -> List[Dict[str, Any]]:
         if self._memory_mode == "context":
             return []

tests/plugins/memory/test_hindsight_provider.py

@@ -1244,3 +1244,127 @@ def test_local_embedded_shutdown_closes_inner_async_client_on_shared_loop(self,
         assert embedded._client is None
         assert provider._client is None
 
+
+class TestOnSessionEnd:
+    def test_flushes_buffered_turns(self, provider_with_config):
+        """Turns accumulated since last batch boundary are flushed at session end."""
+        p = provider_with_config(retain_every_n_turns=5)
+        # Add 3 turns — none retained yet (batch size is 5)
+        p.sync_turn("u1", "a1")
+        p.sync_turn("u2", "a2")
+        p.sync_turn("u3", "a3")
+        if p._sync_thread:
+            p._sync_thread.join(timeout=5.0)
+        p._client.aretain_batch.reset_mock()
+        assert p._turn_counter == 3
+
+        p.on_session_end([])
+
+        # Should have flushed all 3 buffered turns
+        p._client.aretain_batch.assert_called_once()
+        call_kwargs = p._client.aretain_batch.call_args.kwargs
+        assert call_kwargs["retain_async"] is False  # synchronous at session end
+        items = call_kwargs["items"]
+        assert len(items) == 1
+        content = items[0]["content"]
+        assert "u1" in content
+        assert "u2" in content
+        assert "u3" in content
+
+    def test_skipped_when_auto_retain_off(self, provider_with_config):
+        """on_session_end is a no-op when auto_retain is disabled."""
+        p = provider_with_config(auto_retain=False)
+        p.sync_turn("hello", "hi")
+        p._client.aretain_batch.reset_mock()
+
+        p.on_session_end([])
+
+        p._client.aretain_batch.assert_not_called()
+
+    def test_already_aligned_no_flush(self, provider_with_config):
+        """When all turns were retained at the last boundary, nothing to flush."""
+        p = provider_with_config(retain_every_n_turns=3)
+        # Add exactly 3 turns — retain fires on the 3rd
+        p.sync_turn("u1", "a1")
+        p.sync_turn("u2", "a2")
+        p.sync_turn("u3", "a3")
+        if p._sync_thread:
+            p._sync_thread.join(timeout=5.0)
+        p._client.aretain_batch.reset_mock()
+
+        p.on_session_end([])
+
+        # No buffered turns remain — nothing to flush
+        p._client.aretain_batch.assert_not_called()
+
+    def test_no_turns_noop(self, provider):
+        """No turns accumulated — on_session_end is a clean no-op."""
+        provider._client.aretain_batch.reset_mock()
+        provider.on_session_end([])
+        provider._client.aretain_batch.assert_not_called()
+
+    def test_no_client_skips(self, provider):
+        """on_session_end skips when the client hasn't been set."""
+        provider._client = None
+        provider._turn_counter = 5  # simulate accumulated state
+        provider.on_session_end([])  # should not raise
+
+    def test_waits_for_inflight_sync(self, provider_with_config):
+        """If an async sync is in flight, on_session_end waits for it."""
+        import threading
+        p = provider_with_config(retain_every_n_turns=2)
+        # Start a sync and hold it
+        started = threading.Event()
+        held = threading.Event()
+
+        def _slow_sync():
+            started.set()
+            held.wait(timeout=10.0)
+            # Then do the actual retain (mocked)
+            p._run_hindsight_operation(
+                lambda client: client.aretain_batch(
+                    bank_id=p._bank_id,
+                    items=[{"content": "dummy"}],
+                    document_id=p._document_id,
+                    retain_async=True,
+                )
+            )
+
+        p.sync_turn("u1", "a1")
+        p.sync_turn("u2", "a2")
+        # The sync thread should be running
+        assert p._sync_thread is not None
+        p._sync_thread.join(timeout=5.0)
+        p._client.aretain_batch.reset_mock()
+
+        # Now with 0 buffered turns (recently flushed), on_session_end
+        # should still work without deadlocking
+        p.on_session_end([])
+
+    def test_error_does_not_raise(self, provider_with_config):
+        """If the Hindsight API call fails, on_session_end doesn't raise."""
+        p = provider_with_config(retain_every_n_turns=5)
+        p.sync_turn("u1", "a1")
+        p.sync_turn("u2", "a2")
+        if p._sync_thread:
+            p._sync_thread.join(timeout=5.0)
+
+        p._client.aretain_batch.side_effect = RuntimeError("network error")
+        # Should not raise
+        p.on_session_end([])
+
+    def test_session_tags_included(self, provider_with_config):
+        """Flushed turns carry session lineage tags."""
+        p = provider_with_config(retain_every_n_turns=5)
+        p.sync_turn("hello", "hi")
+        p.sync_turn("more", "stuff")
+        if p._sync_thread:
+            p._sync_thread.join(timeout=5.0)
+        p._client.aretain_batch.reset_mock()
+
+        p.on_session_end([])
+
+        p._client.aretain_batch.assert_called_once()
+        item = p._client.aretain_batch.call_args.kwargs["items"][0]
+        assert "session:test-session" in item["tags"]
+