feat: runtime SessionDict validation at JSONL parse boundary

app.py

@@ -1,5 +1,6 @@
 """Flask app that serves the web GUI for browsing sessions."""
 
+import argparse
 import os
 import sys
 
@@ -35,10 +36,8 @@ def index():
     return app
 
 
-def build_cli_parser():
+def build_cli_parser() -> argparse.ArgumentParser:
     """CLI argument parser for ``python app.py`` (stdlib only; safe to import in tests)."""
-    import argparse
-
     parser = argparse.ArgumentParser(description="Claude Code Chat Browser")
     parser.add_argument("--port", type=int, default=5000)
     parser.add_argument("--host", default="127.0.0.1")

models/errors.py

@@ -1,7 +1,16 @@
-"""HTTP error response shapes."""
+"""HTTP error response shapes and domain validation errors."""
 
 from typing import TypedDict
 
 
 class ErrorResponse(TypedDict):
     error: str
+
+
+class SessionValidationError(ValueError):
+    """Raised when parsed JSONL output does not match SessionDict contract."""
+
+    def __init__(self, path: str, detail: str) -> None:
+        self.path = path
+        self.detail = detail
+        super().__init__(f"Session validation failed at {path}: {detail}")

tests/test_jsonl_validation.py

@@ -0,0 +1,86 @@
+"""Runtime validation at the JSONL → SessionDict boundary."""
+
+import os
+import sys
+from typing import Any
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from models.errors import SessionValidationError  # noqa: E402
+from utils.jsonl_parser import parse_session  # noqa: E402
+from utils.validation import validate_session_dict  # noqa: E402
+
+FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+def _valid_payload(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "session_id": "abc123",
+        "title": "Test Session",
+        "messages": [{"role": "user", "text": "hello"}],
+        "metadata": {"session_id": "abc123"},
+    }
+    base.update(overrides)
+    return base
+
+
+class TestValidateSessionDict:
+    def test_missing_session_id(self):
+        payload = _valid_payload()
+        del payload["session_id"]
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(payload)
+        assert exc_info.value.path == "session_id"
+
+    def test_wrong_type_session_id(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(session_id=123))
+        assert exc_info.value.path == "session_id"
+
+    def test_null_session_id(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(session_id=None))
+        assert exc_info.value.path == "session_id"
+        assert exc_info.value.detail == "must not be null"
+
+    def test_null_role_in_message(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(messages=[{"role": None, "text": "x"}])
+            )
+        assert exc_info.value.path == "messages[0].role"
+
+    def test_missing_role_in_message(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(messages=[{"text": "no role key"}])
+            )
+        assert exc_info.value.path == "messages[0].role"
+
+    def test_metadata_not_dict(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(metadata="not-a-dict"))
+        assert exc_info.value.path == "metadata"
+
+    def test_message_not_dict(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(messages=["not-a-dict"]))
+        assert exc_info.value.path == "messages[0]"
+
+    def test_valid_payload_returns_session_dict(self):
+        result = validate_session_dict(_valid_payload())
+        assert result["session_id"] == "abc123"
+        assert result["messages"][0]["role"] == "user"
+
+
+class TestParseSessionValidationRegression:
+    def test_session_minimal_fixture_unchanged(self):
+        path = os.path.join(FIXTURES, "session_minimal.jsonl")
+        session = parse_session(path)
+        assert session["session_id"] == "session_minimal"
+        assert session["title"] == "Hello from integration fixture"
+        assert len(session["messages"]) == 2
+        assert session["messages"][0]["role"] == "user"
+        assert session["messages"][1]["role"] == "assistant"

utils/jsonl_parser.py

@@ -4,9 +4,10 @@
 import json
 import os
 from datetime import datetime
-from typing import Any, cast
+from typing import Any
 
 from models.session import MessageDict, QuickSessionInfoDict, SessionDict
+from utils.validation import validate_session_dict
 
 
 def parse_session(filepath: str) -> SessionDict:
@@ -120,14 +121,13 @@ def parse_session(filepath: str) -> SessionDict:
 
     title = _infer_title(messages)
 
-    return cast(
-        SessionDict,
+    return validate_session_dict(
         {
             "session_id": session_id,
             "title": title,
             "messages": messages,
             "metadata": metadata,
-        },
+        }
     )
 
 

utils/validation.py

@@ -0,0 +1,56 @@
+"""Runtime validation for TypedDict shapes at untrusted-data boundaries."""
+
+from typing import Any, cast
+
+from models.errors import SessionValidationError
+from models.session import SessionDict
+
+_ROOT_PATH = "(root)"
+
+
+def _require_field(
+    container: dict[str, Any],
+    key: str,
+    expected_type: type[Any],
+    type_label: str,
+    *,
+    path: str | None = None,
+) -> Any:
+    field_path = path or key
+    if key not in container:
+        raise SessionValidationError(field_path, "missing required field")
+    return _require_value(field_path, container[key], expected_type, type_label)
+
+
+def _require_value(
+    path: str,
+    val: Any,
+    expected_type: type[Any],
+    type_label: str,
+) -> Any:
+    if val is None:
+        raise SessionValidationError(path, "must not be null")
+    if not isinstance(val, expected_type):
+        raise SessionValidationError(
+            path, f"expected {type_label}, got {type(val).__name__}"
+        )
+    return val
+
+
+def validate_session_dict(data: dict[str, Any]) -> SessionDict:
+    """Validate a plain dict matches SessionDict before returning it."""
+    # Runtime guard for dynamic callers; mypy already types the parameter as dict.
+    if not isinstance(data, dict):
+        raise SessionValidationError(_ROOT_PATH, "expected dict")
+
+    _require_field(data, "session_id", str, "str")
+    _require_field(data, "title", str, "str")
+    messages = _require_field(data, "messages", list, "list")
+    _require_field(data, "metadata", dict, "dict")
+
+    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")
+
+    return cast(SessionDict, data)