fix: skip corrupt/truncated lines in import_session_to_store

src/claude_agent_sdk/_internal/session_import.py

@@ -128,15 +128,22 @@ 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:
         for line in f:
             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)

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