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

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)
+