fix(jsonl): coerce null message before parsing JSONL entries

Author: clean6378-max-it

tests/test_jsonl_parser.py

@@ -729,6 +729,36 @@ def test_missing_usage_dict_no_crash(self):
         finally:
             os.unlink(path)
 
+    def test_null_message_assistant_no_crash(self):
+        path = _write_jsonl([
+            {
+                "type": "assistant",
+                "timestamp": "2026-01-01T00:00:00Z",
+                "message": None,
+            },
+        ])
+        try:
+            s = parse_session(path)
+            assert s["metadata"]["total_input_tokens"] == 0
+            assert len(s["messages"]) == 1
+            assert s["messages"][0]["role"] == "assistant"
+        finally:
+            os.unlink(path)
+
+    def test_non_dict_usage_assistant_no_crash(self):
+        path = _write_jsonl([
+            {
+                "type": "assistant",
+                "timestamp": "2026-01-01T00:00:00Z",
+                "message": {"model": "m", "content": [], "usage": "invalid"},
+            },
+        ])
+        try:
+            s = parse_session(path)
+            assert s["metadata"]["total_input_tokens"] == 0
+        finally:
+            os.unlink(path)
+
 
 # ---------------------------------------------------------------------------
 # quick_session_info

utils/jsonl_parser.py

@@ -137,7 +137,7 @@ def _process_user(entry: dict, messages: list, metadata: dict):
     if metadata["permission_mode"] is None:
         metadata["permission_mode"] = entry.get("permissionMode")
 
-    msg = entry.get("message", {})
+    msg = entry.get("message") or {}
     content = msg.get("content", [])
     text = _extract_text(content)
     images = _extract_images(content)
@@ -170,7 +170,7 @@ def _process_user(entry: dict, messages: list, metadata: dict):
 def _process_assistant(entry: dict, messages: list, metadata: dict):
     """Handle assistant responses -- splits content into text, thinking blocks,
     and tool_use calls, and accumulates token/model/tool stats."""
-    msg = entry.get("message", {})
+    msg = entry.get("message") or {}
     model = msg.get("model", "")
     if model and model != "<synthetic>":
         metadata["models_used"].add(model)
@@ -503,7 +503,7 @@ def quick_session_info(filepath: str) -> dict:
                 last_ts = ts  # keep updating in case file is small
 
             if title is None and entry.get("type") == "user":
-                msg = entry.get("message", {})
+                msg = entry.get("message") or {}
                 text = _extract_text(msg.get("content", []))
                 if text:
                     clean = _strip_system_tags(text).strip()