diff --git a/src/claude_agent_sdk/_internal/session_import.py b/src/claude_agent_sdk/_internal/session_import.py index 30d7823bb..4798f1afd 100644 --- a/src/claude_agent_sdk/_internal/session_import.py +++ b/src/claude_agent_sdk/_internal/session_import.py @@ -128,7 +128,7 @@ async def _append_jsonl_file_in_batches( """Stream-read a JSONL file line-by-line, parsing each line and flushing to ``store.append()`` in batches of ``batch_size`` entries (or ``MAX_PENDING_BYTES`` of line text, whichever comes first). Skips blank - lines.""" + lines and corrupt/truncated lines.""" batch: list[SessionStoreEntry] = [] nbytes = 0 with file_path.open(encoding="utf-8") as f: @@ -136,7 +136,14 @@ async def _append_jsonl_file_in_batches( line = line.rstrip("\n") if not line: continue - batch.append(json.loads(line)) + try: + entry = json.loads(line) + except (json.JSONDecodeError, ValueError): + # Match the read path (_parse_transcript_entries): a + # partially-written final line — the normal outcome when the + # CLI is killed mid-append — is skipped, not raised. + continue + batch.append(entry) nbytes += len(line) if len(batch) >= batch_size or nbytes >= MAX_PENDING_BYTES: await store.append(key, batch) diff --git a/tests/test_session_import.py b/tests/test_session_import.py index dae428c93..6a714cb08 100644 --- a/tests/test_session_import.py +++ b/tests/test_session_import.py @@ -109,6 +109,30 @@ async def test_skips_blank_lines( key: SessionKey = {"project_key": project_key, "session_id": SESSION_ID} assert store.get_entries(key) == [_entry(0), _entry(1)] + @pytest.mark.anyio + async def test_skips_corrupt_truncated_line( + self, claude_dir: Path, cwd: Path, project_key: str + ) -> None: + """A partially-written final line (the normal outcome when the CLI is + killed mid-append) is skipped, not raised — matching the read path.""" + path = claude_dir / f"{SESSION_ID}.jsonl" + path.write_text( + json.dumps(_entry(0)) + + "\n" + + json.dumps(_entry(1)) + + "\n" + + '{"type": "user", "uuid": "u2", "messa', # truncated, unterminated + encoding="utf-8", + ) + + store = InMemorySessionStore() + await import_session_to_store( + SESSION_ID, store, directory=str(cwd), batch_size=1 + ) + + key: SessionKey = {"project_key": project_key, "session_id": SESSION_ID} + assert store.get_entries(key) == [_entry(0), _entry(1)] + @pytest.mark.anyio async def test_nonpositive_batch_size_uses_default( self, claude_dir: Path, cwd: Path, project_key: str