feat: narrow MessageDict.role to RoleLiteral

Author: clean6378-max-it

models/__init__.py

@@ -8,6 +8,7 @@
 from models.session import (
     MessageDict,
     QuickSessionInfoDict,
+    RoleLiteral,
     SessionDict,
     SessionMetadataDict,
     ToolUseDict,
@@ -23,6 +24,7 @@
     "ProjectDict",
     "ProjectSessionRowDict",
     "QuickSessionInfoDict",
+    "RoleLiteral",
     "SearchHitDict",
     "SessionDict",
     "SessionListItemDict",

models/search.py

@@ -2,11 +2,13 @@
 
 from typing import TypedDict
 
+from models.session import RoleLiteral
+
 
 class SearchHitDict(TypedDict):
     project: str
     session_id: str
     title: str
-    role: str
+    role: RoleLiteral
     timestamp: str | None
     snippet: str

models/session.py

@@ -5,6 +5,8 @@
 from models.record_data import RecordDataUnion
 from models.tool_results import ToolNameLiteral, ToolResultUnion
 
+RoleLiteral = Literal["user", "assistant", "system", "result", "progress"]
+
 
 class ToolUseDict(TypedDict, total=False):
     id: str
@@ -25,7 +27,7 @@ class MessageUsageDict(TypedDict, total=False):
 
 
 class MessageDict(TypedDict):
-    role: str
+    role: RoleLiteral
     uuid: NotRequired[str | None]
     parent_uuid: NotRequired[str | None]
     timestamp: NotRequired[str | None]

tests/test_jsonl_parser.py

@@ -686,16 +686,19 @@ def test_empty_file_returns_skeleton(self):
         finally:
             os.unlink(path)
 
-    def test_unknown_entry_type_silently_ignored(self):
+    def test_unknown_entry_type_maps_to_system(self, caplog):
         path = _write_jsonl(
             [
                 {"type": "custom", "timestamp": "2026-01-01T00:00:00Z"},
             ]
         )
         try:
-            s = parse_session(path)
-            assert s["messages"] == []
+            with caplog.at_level("WARNING"):
+                s = parse_session(path)
+            assert len(s["messages"]) == 1
+            assert s["messages"][0]["role"] == "system"
             assert s["metadata"]["entry_counts"].get("custom") == 1
+            assert "Unknown message role" in caplog.text
         finally:
             os.unlink(path)
 
@@ -981,3 +984,48 @@ def test_tool_use_result_string_returns_none(self):
             ]
         )
         assert s["messages"][0]["tool_result_parsed"] is None
+
+
+# ---------------------------------------------------------------------------
+# Unknown role coercion
+# ---------------------------------------------------------------------------
+
+
+class TestUnknownRoleCoercion:
+    def test_unknown_entry_type_maps_to_system_with_warning(self, caplog):
+        path = _write_jsonl(
+            [
+                {
+                    "type": "mystery_future_type",
+                    "timestamp": "2026-01-01T00:00:00Z",
+                    "content": "forward-compat payload",
+                }
+            ]
+        )
+        try:
+            with caplog.at_level("WARNING"):
+                s = parse_session(path)
+            assert len(s["messages"]) == 1
+            assert s["messages"][0]["role"] == "system"
+            assert s["messages"][0]["content"] == "forward-compat payload"
+            assert "Unknown message role" in caplog.text
+            assert "mystery_future_type" in caplog.text
+        finally:
+            os.unlink(path)
+
+    def test_valid_unhandled_result_type_emits_result_role(self):
+        path = _write_jsonl(
+            [
+                {
+                    "type": "result",
+                    "timestamp": "2026-01-01T00:00:00Z",
+                    "content": "task outcome",
+                }
+            ]
+        )
+        try:
+            s = parse_session(path)
+            assert len(s["messages"]) == 1
+            assert s["messages"][0]["role"] == "result"
+        finally:
+            os.unlink(path)

tests/test_parser_fuzz.py

@@ -240,8 +240,10 @@ def test_unknown_record_type_is_graceful(tmp_path: Path) -> None:
     path = _write_jsonl(tmp_path / "unknown.jsonl", lines)
     session = parse_session(path)
     assert session["metadata"]["entry_counts"].get("totally-new-claude-record") == 1
-    # Unknown type produces no message; only the valid user line does.
-    assert len(session["messages"]) == 1
+    # Unknown type is coerced to a system message; the valid user line follows.
+    assert len(session["messages"]) == 2
+    assert session["messages"][0]["role"] == "system"
+    assert session["messages"][1]["role"] == "user"
 
 
 def test_non_numeric_usage_tokens_do_not_crash(tmp_path: Path) -> None:

utils/jsonl_parser.py

@@ -2,13 +2,14 @@
 actually work with -- messages, tool calls, token counts, file activity, etc."""
 
 import json
+import logging
 import math
 import os
 from datetime import datetime
-from typing import Any
+from typing import Any, cast, get_args
 
 from models.record_data import RecordDataUnion
-from models.session import MessageDict, SessionDict, ToolUseDict
+from models.session import MessageDict, RoleLiteral, SessionDict, ToolUseDict
 from models.tool_results import ToolResultUnion, is_tool_result_dict
 from utils.jsonl_helpers import (
     entry_message as _entry_message,
@@ -23,6 +24,30 @@
 
 __all__ = ["parse_session", "quick_session_info"]
 
+_HANDLED_ENTRY_TYPES = frozenset({"user", "assistant", "system", "progress"})
+_VALID_ROLES = frozenset(get_args(RoleLiteral))
+
+
+def _coerce_role(raw: str) -> RoleLiteral:
+    if raw in _VALID_ROLES:
+        return cast(RoleLiteral, raw)
+    logging.warning("Unknown message role %r; mapping to 'system'", raw)
+    return "system"
+
+
+def _fallback_message(entry: dict[str, Any], role: RoleLiteral) -> MessageDict:
+    """Minimal message for JSONL entry types without a dedicated processor."""
+    content = entry.get("content", "")
+    text = content if isinstance(content, str) else (str(content) if content is not None else "")
+    return {
+        "role": role,
+        "uuid": entry.get("uuid"),
+        "parent_uuid": entry.get("parentUuid"),
+        "timestamp": entry.get("timestamp"),
+        "content": text,
+        "is_sidechain": entry.get("isSidechain", False),
+    }
+
 
 def _safe_int(val: Any) -> int:
     """Coerce a value to a non-negative int for token accounting; non-numeric,
@@ -127,6 +152,10 @@ def parse_session(filepath: str) -> SessionDict:
                 _process_system(entry, messages, metadata)
             elif entry_type == "progress":
                 _process_progress(entry, messages)
+            elif entry_type:
+                type_str = entry_type if isinstance(entry_type, str) else str(entry_type)
+                if type_str not in _HANDLED_ENTRY_TYPES:
+                    messages.append(_fallback_message(entry, _coerce_role(type_str)))
 
     metadata["models_used"] = sorted(metadata["models_used"])
     metadata["service_tiers"] = sorted(metadata["service_tiers"])