fix(retain): merge JSON arrays in append mode to preserve conversation-aware chunking (#2412)

* fix(retain): merge JSON arrays in append mode to preserve conversation-aware chunking

When update_mode=append prepends existing document text as a second
content item, combined_content is built with "\n".join(...). For
conversation-format content (flat JSON arrays of message dicts), this
produces "[...]\n[...]" which is not valid JSON.

On subsequent append cycles, chunk_text() fails to parse the corrupted
original_text. _chunk_jsonl() also rejects it (lines are arrays, not
dicts). The text falls through to RecursiveCharacterTextSplitter, which
splits on sentence boundaries with no awareness of conversation turn
structure. This produces chunks that begin mid-sentence without speaker
attribution, causing the extraction LLM to misattribute statements.

Fix: after the append-mode block assembles contents_dicts with the
existing and new content items, detect when all items are JSON arrays
of dicts and merge them into a single flat array. Non-conversation
content (plain text, JSONL) is unaffected.

close #2409

* Enhance chunking tests for JSON array formats

Add tests for chunking newline-joined and merged JSON arrays.

* Add test for valid JSON in append mode

This test ensures that appending conversation arrays maintains the original_text as a valid flat JSON array after multiple append cycles, preventing degradation of the data structure.

* add missing json import to test_retain_append_mode

Author: qxxaa

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -834,6 +834,28 @@ async def retain_batch(
             if first.get("tags"):
                 existing_content["tags"] = first["tags"]
             contents_dicts = [existing_content, *contents_dicts]
+            # Merge JSON arrays to keep original_text valid (#2409).
+            # Without this, combined_content joins items with "\n", producing
+            # "[...]\n[...]" which is not valid JSON. On the next append cycle
+            # chunk_text() fails to parse it and falls through to sentence-
+            # boundary text splitting, breaking speaker attribution.
+            try:
+                _merged = []
+                for _item in contents_dicts:
+                    _parsed = json.loads(_item.get("content", ""))
+                    if isinstance(_parsed, list) and all(isinstance(_e, dict) for _e in _parsed):
+                        _merged.extend(_parsed)
+                    else:
+                        _merged = None
+                        break
+                if _merged is not None:
+                    contents_dicts = [{"content": json.dumps(_merged, ensure_ascii=False)}]
+                    if first.get("context"):
+                        contents_dicts[0]["context"] = first["context"]
+                    if first.get("tags"):
+                        contents_dicts[0]["tags"] = first["tags"]
+            except (json.JSONDecodeError, ValueError, TypeError):
+                pass
             # Rebuild contents list to match
             contents = _build_contents(contents_dicts, document_tags)
             log_buffer.append(

hindsight-api-slim/tests/test_chunking.py

@@ -395,3 +395,64 @@ def test_rechunk_preserves_one_chunk_id_per_pre_chunk():
             chunk_ids.append(f"bank_doc_{global_idx}")
 
     assert len(chunk_ids) == len(set(chunk_ids)), f"duplicate chunk_ids in one batch: {chunk_ids}"
+
+
+# ---------------------------------------------------------------------------
+# Append-mode JSON array merge simulation (issue #2409)
+# ---------------------------------------------------------------------------
+
+
+def test_newline_joined_json_arrays_bypass_conversation_chunking():
+    """Newline-joined JSON arrays (the pre-fix append-mode storage format)
+    fail both the conversation and JSONL detection paths and fall through
+    to sentence-boundary text splitting.
+
+    This test documents the broken state that issue #2409 fixes at the
+    orchestrator level. chunk_text() itself is not changed; the fix
+    merges the arrays before they reach chunk_text().
+    """
+    turn1 = json.dumps([{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there"}])
+    turn2 = json.dumps([{"role": "user", "content": "How are you"}, {"role": "assistant", "content": "Fine"}])
+    corrupted = turn1 + "\n" + turn2
+
+    chunks = chunk_text(corrupted, max_chars=80)
+
+    # The corrupted format does NOT route through _chunk_conversation.
+    # At least one chunk will not be a valid JSON array of dicts.
+    has_non_json_chunk = False
+    for chunk in chunks:
+        try:
+            parsed = json.loads(chunk)
+            if not (isinstance(parsed, list) and all(isinstance(e, dict) for e in parsed)):
+                has_non_json_chunk = True
+        except json.JSONDecodeError:
+            has_non_json_chunk = True
+    assert has_non_json_chunk, (
+        "Newline-joined JSON arrays should NOT produce valid conversation chunks. "
+        "If this fails, chunk_text() learned to handle the format and the "
+        "orchestrator-level merge in #2409 may be redundant."
+    )
+
+
+def test_merged_json_array_routes_to_conversation_chunking():
+    """A properly merged flat JSON array (the post-fix format) routes
+    through _chunk_conversation and produces chunks that are each valid
+    JSON arrays of complete message dicts.
+    """
+    messages = [
+        {"role": "user", "content": "Hello"},
+        {"role": "assistant", "content": "Hi there"},
+        {"role": "user", "content": "How are you"},
+        {"role": "assistant", "content": "Fine, thanks for asking"},
+    ]
+    text = json.dumps(messages)
+
+    chunks = chunk_text(text, max_chars=120)
+
+    assert len(chunks) > 1, "Should produce multiple chunks at this budget"
+    for chunk in chunks:
+        parsed = json.loads(chunk)
+        assert isinstance(parsed, list), f"Chunk must be a JSON array: {chunk[:60]}"
+        assert all(isinstance(e, dict) for e in parsed), f"Every element must be a dict: {chunk[:60]}"
+        assert all("role" in e for e in parsed), f"Every element must have a role key: {chunk[:60]}"
+        

hindsight-api-slim/tests/test_retain_append_mode.py

@@ -2,6 +2,7 @@
 Tests for retain update_mode='append' — appends new content to existing documents.
 """
 
+import json
 import logging
 from datetime import datetime, timezone
 
@@ -238,3 +239,94 @@ async def test_replace_mode_is_default(memory, request_context):
 
     finally:
         await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_append_mode_conversation_arrays_produce_valid_json(memory, request_context):
+    """When conversation-format JSON arrays are appended, original_text
+    must remain a valid flat JSON array after multiple append cycles.
+
+    Regression test for #2409: without the merge fix, original_text
+    becomes newline-joined arrays which breaks conversation-aware chunking.
+    """
+    bank_id = f"test_append_conv_{_ts()}"
+    document_id = "conversation-json-append"
+
+    try:
+        # First retain - JSON conversation array
+        turn1 = json.dumps([
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi there"},
+        ])
+        await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {
+                    "content": turn1,
+                    "context": "conversation",
+                    "document_id": document_id,
+                }
+            ],
+            request_context=request_context,
+        )
+
+        # Second retain - append more turns
+        turn2 = json.dumps([
+            {"role": "user", "content": "How are you"},
+            {"role": "assistant", "content": "Doing well"},
+        ])
+        await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {
+                    "content": turn2,
+                    "context": "conversation",
+                    "document_id": document_id,
+                    "update_mode": "append",
+                }
+            ],
+            request_context=request_context,
+        )
+
+        # Verify original_text is valid JSON (not newline-joined arrays)
+        doc = await memory.get_document(document_id, bank_id, request_context=request_context)
+        text = doc["original_text"]
+
+        parsed = json.loads(text)
+        assert isinstance(parsed, list), "original_text must be a JSON array"
+        assert all(isinstance(e, dict) for e in parsed), (
+            "original_text must be a flat array of dicts, not nested arrays"
+        )
+        assert len(parsed) == 4, "Should contain all 4 messages from both retains"
+
+        # Third retain - append again, verify no degradation
+        turn3 = json.dumps([
+            {"role": "user", "content": "What is new"},
+            {"role": "assistant", "content": "Not much"},
+        ])
+        await memory.retain_batch_async(
+            bank_id=bank_id,
+            contents=[
+                {
+                    "content": turn3,
+                    "context": "conversation",
+                    "document_id": document_id,
+                    "update_mode": "append",
+                }
+            ],
+            request_context=request_context,
+        )
+
+        doc = await memory.get_document(document_id, bank_id, request_context=request_context)
+        text = doc["original_text"]
+
+        parsed = json.loads(text)
+        assert isinstance(parsed, list), "original_text must remain a JSON array after 3rd append"
+        assert all(isinstance(e, dict) for e in parsed), (
+            "original_text must remain a flat array of dicts after 3rd append"
+        )
+        assert len(parsed) == 6, "Should contain all 6 messages from three retains"
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+