feat: runtime SessionDict validation at JSONL parse boundary

Replace cast(SessionDict, ...) in parse_session() with validate_session_dict()
so untrusted .jsonl files are checked at ingress. Add SessionValidationError
with path/detail fields, utils/validation.py, and tests for five malformed
shapes plus session_minimal.jsonl regression.

Author: clean6378-max-it

models/errors.py

@@ -1,7 +1,15 @@
-"""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
+        super().__init__(f"Session validation failed at {path}: {detail}")

tests/test_jsonl_validation.py

@@ -0,0 +1,72 @@
+"""Runtime validation at the JSONL → SessionDict boundary."""
+
+import os
+import sys
+
+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: object) -> dict:
+    base = {
+        "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_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_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,78 @@
+"""Runtime validation for TypedDict shapes at untrusted-data boundaries."""
+
+from typing import Any, cast
+
+from models.errors import SessionValidationError
+from models.session import MessageDict, SessionDict
+
+_REQUIRED_SESSION_KEYS = ("session_id", "title", "messages", "metadata")
+
+
+def validate_session_dict(data: dict[str, Any]) -> SessionDict:
+    """Validate a plain dict matches SessionDict before returning it."""
+    if not isinstance(data, dict):
+        raise SessionValidationError("$", "expected dict")
+
+    for key in _REQUIRED_SESSION_KEYS:
+        if key not in data:
+            raise SessionValidationError(key, "missing required field")
+
+    session_id = data["session_id"]
+    if session_id is None:
+        raise SessionValidationError("session_id", "must not be null")
+    if not isinstance(session_id, str):
+        raise SessionValidationError(
+            "session_id", f"expected str, got {type(session_id).__name__}"
+        )
+
+    title = data["title"]
+    if title is None:
+        raise SessionValidationError("title", "must not be null")
+    if not isinstance(title, str):
+        raise SessionValidationError(
+            "title", f"expected str, got {type(title).__name__}"
+        )
+
+    messages = data["messages"]
+    if messages is None:
+        raise SessionValidationError("messages", "must not be null")
+    if not isinstance(messages, list):
+        raise SessionValidationError(
+            "messages", f"expected list, got {type(messages).__name__}"
+        )
+
+    for index, message in enumerate(messages):
+        path = f"messages[{index}]"
+        if message is None:
+            raise SessionValidationError(path, "must not be null")
+        if not isinstance(message, dict):
+            raise SessionValidationError(
+                path, f"expected dict, got {type(message).__name__}"
+            )
+        if "role" not in message:
+            raise SessionValidationError(f"{path}.role", "missing required field")
+        role = message["role"]
+        if role is None:
+            raise SessionValidationError(f"{path}.role", "must not be null")
+        if not isinstance(role, str):
+            raise SessionValidationError(
+                f"{path}.role", f"expected str, got {type(role).__name__}"
+            )
+
+    metadata = data["metadata"]
+    if metadata is None:
+        raise SessionValidationError("metadata", "must not be null")
+    if not isinstance(metadata, dict):
+        raise SessionValidationError(
+            "metadata", f"expected dict, got {type(metadata).__name__}"
+        )
+
+    return cast(
+        SessionDict,
+        {
+            "session_id": session_id,
+            "title": title,
+            "messages": cast(list[MessageDict], messages),
+            "metadata": metadata,
+        },
+    )