feat: narrow MessageDict.role to RoleLiteral (#89)

* feat: narrow MessageDict.role to RoleLiteral

* fix: skip metadata-only JSONL entry types in role fallback path

* fix: address PR #89 review feedback on role coercion and fallback messages

* fix: skip summary metadata entries in JSONL role fallback path

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,19 +686,6 @@ def test_empty_file_returns_skeleton(self):
         finally:
             os.unlink(path)
 
-    def test_unknown_entry_type_silently_ignored(self):
-        path = _write_jsonl(
-            [
-                {"type": "custom", "timestamp": "2026-01-01T00:00:00Z"},
-            ]
-        )
-        try:
-            s = parse_session(path)
-            assert s["messages"] == []
-            assert s["metadata"]["entry_counts"].get("custom") == 1
-        finally:
-            os.unlink(path)
-
     def test_is_sidechain_increments_counter(self):
         path = _write_jsonl(
             [
@@ -729,6 +716,26 @@ def test_file_history_snapshot_timestamp(self):
             s = parse_session(path)
             assert s["metadata"]["first_timestamp"] == "2026-01-02T12:00:00Z"
             assert s["metadata"]["last_timestamp"] == "2026-01-02T12:00:00Z"
+            assert len(s["messages"]) == 0
+        finally:
+            os.unlink(path)
+
+    def test_summary_entry_type_produces_no_message(self, caplog):
+        path = _write_jsonl(
+            [
+                {
+                    "type": "summary",
+                    "timestamp": "2026-01-03T08:00:00Z",
+                    "summary": "Session recap metadata",
+                },
+            ]
+        )
+        try:
+            with caplog.at_level("WARNING", logger="utils.jsonl_parser"):
+                s = parse_session(path)
+            assert len(s["messages"]) == 0
+            assert s["metadata"]["entry_counts"].get("summary") == 1
+            assert "Unknown message role" not in caplog.text
         finally:
             os.unlink(path)
 
@@ -981,3 +988,82 @@ 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", logger="utils.jsonl_parser"):
+                s = parse_session(path)
+            assert len(s["messages"]) == 1
+            assert s["messages"][0]["role"] == "system"
+            assert s["messages"][0]["text"] == "forward-compat payload"
+            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_fallback_message_extracts_text_from_structured_content(self):
+        path = _write_jsonl(
+            [
+                {
+                    "type": "result",
+                    "timestamp": "2026-01-01T00:00:00Z",
+                    "content": [{"type": "text", "text": "block body"}],
+                }
+            ]
+        )
+        try:
+            s = parse_session(path)
+            assert s["messages"][0]["text"] == "block body"
+            assert s["messages"][0]["content"] == "block body"
+        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)
+
+
+class TestCoerceRole:
+    def test_known_roles_returned_unchanged(self):
+        from utils.jsonl_parser import _coerce_role
+
+        for role in ("user", "assistant", "system", "result", "progress"):
+            assert _coerce_role(role) == role
+
+    def test_unknown_role_maps_to_system_with_warning(self, caplog):
+        from utils.jsonl_parser import _coerce_role
+
+        with caplog.at_level("WARNING", logger="utils.jsonl_parser"):
+            result = _coerce_role("totally_unknown")
+        assert result == "system"
+        assert "totally_unknown" in caplog.text

tests/test_jsonl_validation.py

@@ -70,6 +70,12 @@ def test_valid_payload_returns_session_dict(self):
         assert result["session_id"] == "abc123"
         assert result["messages"][0]["role"] == "user"
 
+    def test_invalid_role_in_message(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(messages=[{"role": "custom", "text": "x"}]))
+        assert exc_info.value.path == "messages[0].role"
+        assert "custom" in exc_info.value.detail
+
 
 class TestParseSessionValidationRegression:
     def test_session_minimal_fixture_unchanged(self):

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,38 @@
 
 __all__ = ["parse_session", "quick_session_info"]
 
+# Metadata-only JSONL entry types: contribute timestamps/counts but not messages.
+_SKIP_ENTRY_TYPES = frozenset({"file-history-snapshot", "summary"})
+_VALID_ROLES = frozenset(get_args(RoleLiteral))
+_log = logging.getLogger(__name__)
+
+
+def _coerce_role(raw: str) -> RoleLiteral:
+    if raw in _VALID_ROLES:
+        return cast(RoleLiteral, raw)
+    _log.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."""
+    raw_content = entry.get("content", "")
+    if isinstance(raw_content, str):
+        text = raw_content
+    elif raw_content is not None:
+        text = _extract_text(_normalize_content(raw_content))
+    else:
+        text = ""
+    return {
+        "role": role,
+        "uuid": entry.get("uuid"),
+        "parent_uuid": entry.get("parentUuid"),
+        "timestamp": entry.get("timestamp"),
+        "text": text,
+        "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 +160,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 _SKIP_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"])

utils/validation.py

@@ -1,9 +1,11 @@
 """Runtime validation for TypedDict shapes at untrusted-data boundaries."""
 
-from typing import Any, cast
+from typing import Any, cast, get_args
 
 from models.errors import SessionValidationError
-from models.session import SessionDict
+from models.session import RoleLiteral, SessionDict
+
+_VALID_ROLES = frozenset(get_args(RoleLiteral))
 
 _ROOT_PATH = "(root)"
 
@@ -49,6 +51,11 @@ def validate_session_dict(data: dict[str, Any]) -> SessionDict:
     for index, message in enumerate(messages):
         path = f"messages[{index}]"
         msg_dict = _require_value(path, message, dict, "dict")
-        _require_field(msg_dict, "role", str, "str", path=f"{path}.role")
+        role = _require_field(msg_dict, "role", str, "str", path=f"{path}.role")
+        if role not in _VALID_ROLES:
+            raise SessionValidationError(
+                f"{path}.role",
+                f"expected one of {sorted(_VALID_ROLES)!r}, got {role!r}",
+            )
 
     return cast(SessionDict, data)