fix: flatten sync_turn content to enable turn-aware chunking for hindsight plugin

contributors/emails/qxxaa@users.noreply.github.com

@@ -0,0 +1 @@
+qxxaa

plugins/memory/hindsight/__init__.py

@@ -1010,7 +1010,8 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         if session_id:
             self._session_id = str(session_id).strip()
 
-        self._session_turns.append(json.dumps(self._build_turn_messages(user_content, assistant_content), ensure_ascii=False))
+        turn = ",".join(json.dumps(m, ensure_ascii=False) for m in self._build_turn_messages(user_content, assistant_content))
+        self._session_turns.append(turn)
         self._turn_counter = self._turn_index = self._turn_counter + 1
         if remainder := self._turn_counter % self._retain_every_n_turns:
             logger.debug("sync_turn: buffered turn %d (will retain at turn %d)",

tests/plugins/memory/test_hindsight_provider.py

@@ -927,11 +927,11 @@ def test_sync_turn_retains_metadata_rich_turn(self, provider_with_config, monkey
         assert item["context"] == "conversation between Hermes Agent and the User"
         assert item["tags"] == ["conv", "session1", "session:session-1"]
         content = json.loads(item["content"])
-        assert len(content) == 1
-        assert content[0][0]["role"] == "user"
-        assert content[0][0]["content"] == "User (fakeusername): hello"
-        assert content[0][1]["role"] == "assistant"
-        assert content[0][1]["content"] == "Assistant (fakeassistantname): hi there"
+        assert len(content) == 2
+        assert content[0]["role"] == "user"
+        assert content[0]["content"] == "User (fakeusername): hello"
+        assert content[1]["role"] == "assistant"
+        assert content[1]["content"] == "Assistant (fakeassistantname): hi there"
         assert item["metadata"]["source"] == "hermes"
         assert item["metadata"]["session_id"] == "session-1"
         assert item["metadata"]["platform"] == "discord"
@@ -944,8 +944,8 @@ def test_sync_turn_retains_metadata_rich_turn(self, provider_with_config, monkey
         assert item["metadata"]["agent_identity"] == "fakeassistantname"
         assert item["metadata"]["turn_index"] == "1"
         assert item["metadata"]["message_count"] == "2"
-        assert content[0][0]["timestamp"] == event_time.isoformat(timespec="seconds")
-        assert content[0][1]["timestamp"] == event_time.isoformat(timespec="seconds")
+        assert content[0]["timestamp"] == event_time.isoformat(timespec="seconds")
+        assert content[1]["timestamp"] == event_time.isoformat(timespec="seconds")
         assert re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", item["metadata"]["retained_at"])
         assert item["timestamp"] == event_time.isoformat(timespec="seconds")
 
@@ -981,6 +981,174 @@ async def test_retain_timestamp_is_serialized_by_pinned_client(self, provider):
         finally:
             await client.aclose()
 
+    def test_sync_turn_skipped_when_auto_retain_off(self, provider_with_config):
+        p = provider_with_config(auto_retain=False)
+        p.sync_turn("hello", "hi")
+        assert p._sync_thread is None
+        p._client.aretain_batch.assert_not_called()
+
+    def test_sync_turn_with_tags(self, provider_with_config):
+        p = provider_with_config(retain_tags=["conv", "session1"])
+        p.sync_turn("hello", "hi")
+        p._retain_queue.join()
+        item = p._client.aretain_batch.call_args.kwargs["items"][0]
+        assert "conv" in item["tags"]
+        assert "session1" in item["tags"]
+        assert "session:test-session" in item["tags"]
+
+    def test_sync_turn_uses_aretain_batch(self, provider):
+        """sync_turn should use aretain_batch with retain_async."""
+        provider.sync_turn("hello", "hi")
+        provider._retain_queue.join()
+        provider._client.aretain_batch.assert_called_once()
+        call_kwargs = provider._client.aretain_batch.call_args.kwargs
+        assert call_kwargs["document_id"].startswith("test-session-")
+        assert call_kwargs["retain_async"] is True
+        assert len(call_kwargs["items"]) == 1
+        assert call_kwargs["items"][0]["context"] == "conversation between Hermes Agent and the User"
+
+    def test_sync_turn_custom_context(self, provider_with_config):
+        p = provider_with_config(retain_context="my-agent")
+        p.sync_turn("hello", "hi")
+        p._retain_queue.join()
+        item = p._client.aretain_batch.call_args.kwargs["items"][0]
+        assert item["context"] == "my-agent"
+
+    def test_sync_turn_every_n_turns(self, provider_with_config):
+        p = provider_with_config(retain_every_n_turns=3, retain_async=False)
+        p.sync_turn("turn1-user", "turn1-asst")
+        assert p._sync_thread is None
+        p.sync_turn("turn2-user", "turn2-asst")
+        assert p._sync_thread is None
+        p.sync_turn("turn3-user", "turn3-asst")
+        p._retain_queue.join()
+        p._client.aretain_batch.assert_called_once()
+        call_kwargs = p._client.aretain_batch.call_args.kwargs
+        assert call_kwargs["document_id"].startswith("test-session-")
+        assert call_kwargs["retain_async"] is False
+        item = call_kwargs["items"][0]
+        content = json.loads(item["content"])
+        assert len(content) == 6  # 3 turns x 2 messages each
+        assert content[-2]["role"] == "user"
+        assert content[-2]["content"] == "User: turn3-user"
+        assert content[-1]["role"] == "assistant"
+        assert content[-1]["content"] == "Assistant: turn3-asst"
+        assert item["metadata"]["turn_index"] == "3"
+        assert item["metadata"]["message_count"] == "6"
+
+    def test_sync_turn_produces_flat_conversation_array(self, provider_with_config):
+        """Content must be a flat JSON array of message dicts for turn-aware chunking.
+
+        Hindsight's chunk_text() routes to _chunk_conversation() only when:
+            parsed = json.loads(content)
+            isinstance(parsed, list) and all(isinstance(turn, dict) for turn in parsed)
+
+        A nested array (list of pairs) fails this check and falls through to
+        dumb text splitting which can orphan messages mid-turn.
+        """
+        p = provider_with_config(retain_every_n_turns=2)
+        p.sync_turn("msg1-user", "msg1-asst")
+        p.sync_turn("msg2-user", "msg2-asst")
+        p._retain_queue.join()
+
+        item = p._client.aretain_batch.call_args.kwargs["items"][0]
+        content = json.loads(item["content"])
+
+        # Must be a flat list of dicts (not nested list of lists)
+        assert isinstance(content, list)
+        assert all(isinstance(entry, dict) for entry in content), (
+            "Content must be a flat array of message dicts for Hindsight's "
+            "turn-aware chunking path to activate"
+        )
+        # 2 turns x 2 messages = 4 entries
+        assert len(content) == 4
+        # Every entry must have role, content, timestamp
+        for entry in content:
+            assert "role" in entry
+            assert "content" in entry
+            assert "timestamp" in entry
+        # Verify ordering
+        assert content[0]["role"] == "user"
+        assert content[1]["role"] == "assistant"
+        assert content[2]["role"] == "user"
+        assert content[3]["role"] == "assistant"
+
+    def test_sync_turn_accumulates_full_session_without_append_support(self, provider_with_config):
+        """Legacy/overwrite APIs (no update_mode=append) resend the ENTIRE session each retain."""
+        p = provider_with_config(retain_every_n_turns=2)
+
+        p.sync_turn("turn1-user", "turn1-asst")
+        p.sync_turn("turn2-user", "turn2-asst")
+        p._retain_queue.join()
+
+        p._client.aretain_batch.reset_mock()
+
+        p.sync_turn("turn3-user", "turn3-asst")
+        p.sync_turn("turn4-user", "turn4-asst")
+        p._retain_queue.join()
+
+        content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
+        # Without append support the document is overwritten, so it must
+        # contain ALL turns from the session.
+        assert "turn1-user" in content
+        assert "turn2-user" in content
+        assert "turn3-user" in content
+        assert "turn4-user" in content
+
+    def test_sync_turn_appends_only_delta_when_append_supported(self, provider_with_config, monkeypatch):
+        """On append-capable APIs each retain ships only the new turns, not the whole session."""
+        monkeypatch.setattr(
+            "plugins.memory.hindsight._fetch_hindsight_api_version",
+            lambda *a, **kw: "0.5.6",
+        )
+        from plugins.memory.hindsight import _append_capability_cache, _append_capability_lock
+        # Clear before AND after: the capability cache is module-global and keyed
+        # per api_url, so a stale entry would leak into other tests.
+        with _append_capability_lock:
+            _append_capability_cache.clear()
+        try:
+            p = provider_with_config(retain_every_n_turns=2)
+
+            p.sync_turn("turn1-user", "turn1-asst")
+            p.sync_turn("turn2-user", "turn2-asst")
+            p._retain_queue.join()
+
+            first = p._client.aretain_batch.call_args.kwargs
+            first_item = first["items"][0]
+            assert first["document_id"] == "test-session"
+            assert first_item["update_mode"] == "append"
+            assert "turn1-user" in first_item["content"]
+            assert "turn2-user" in first_item["content"]
+
+            p._client.aretain_batch.reset_mock()
+
+            p.sync_turn("turn3-user", "turn3-asst")
+            p.sync_turn("turn4-user", "turn4-asst")
+            p._retain_queue.join()
+
+            second = p._client.aretain_batch.call_args.kwargs
+            second_item = second["items"][0]
+            assert second["document_id"] == "test-session"
+            assert second_item["update_mode"] == "append"
+            # Only the delta - the already-retained turns must NOT be resent.
+            assert "turn1-user" not in second_item["content"]
+            assert "turn2-user" not in second_item["content"]
+            assert "turn3-user" in second_item["content"]
+            assert "turn4-user" in second_item["content"]
+            # message_count reflects only the delta (2 turns -> 4 messages).
+            assert second_item["metadata"]["message_count"] == "4"
+        finally:
+            with _append_capability_lock:
+                _append_capability_cache.clear()
+
+    def test_sync_turn_passes_document_id(self, provider):
+        """sync_turn should pass document_id (session_id + per-startup ts)."""
+        provider.sync_turn("hello", "hi")
+        provider._retain_queue.join()
+        call_kwargs = provider._client.aretain_batch.call_args.kwargs
+        # Format: {session_id}-{YYYYMMDD_HHMMSS_microseconds}
+        assert call_kwargs["document_id"].startswith("test-session-")
+        assert call_kwargs["document_id"] == provider._document_id
 
     def test_resume_creates_new_document(self, tmp_path, monkeypatch):
         """Resuming a session (re-initializing) gets a new document_id
@@ -1128,11 +1296,21 @@ def test_buffered_turns_flushed_before_clear(self, provider_with_config):
         kw = p._client.aretain_batch.call_args.kwargs
         assert kw["document_id"] == old_doc
         item = kw["items"][0]
-        # Both buffered turns must be present in the flushed payload.
+        # Both buffered turns must be present in the flushed payload
+        # AND must satisfy Hindsight's flat-dict contract for turn-aware chunking.
         content = json.loads(item["content"])
-        flat = json.dumps(content)
-        assert "turn1-user" in flat
-        assert "turn2-user" in flat
+        assert isinstance(content, list)
+        assert all(isinstance(entry, dict) for entry in content), (
+            "Session-switch flush must produce a flat array of message dicts"
+        )
+        # 2 turns x 2 messages = 4 entries in user/assistant order
+        assert len(content) == 4
+        assert content[0]["role"] == "user"
+        assert content[1]["role"] == "assistant"
+        assert content[2]["role"] == "user"
+        assert content[3]["role"] == "assistant"
+        assert "turn1-user" in content[0]["content"]
+        assert "turn2-user" in content[2]["content"]
         # Old session id must appear in lineage tags / metadata.
         assert "session:test-session" in item["tags"]
         assert item["metadata"]["session_id"] == "test-session"