diff --git a/hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py b/hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py index e92024c082..4cad3c4f7a 100644 --- a/hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py +++ b/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( diff --git a/hindsight-api-slim/tests/test_chunking.py b/hindsight-api-slim/tests/test_chunking.py index c181c8eac3..ea69a46567 100644 --- a/hindsight-api-slim/tests/test_chunking.py +++ b/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]}" + diff --git a/hindsight-api-slim/tests/test_retain_append_mode.py b/hindsight-api-slim/tests/test_retain_append_mode.py index 5bb9567162..d02b65b353 100644 --- a/hindsight-api-slim/tests/test_retain_append_mode.py +++ b/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) +