From 37cf318d74ec13fa1ed45eec2bdce8bf45804baf Mon Sep 17 00:00:00 2001 From: chen Date: Tue, 26 May 2026 01:31:06 +0800 Subject: [PATCH 1/4] 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. --- models/errors.py | 10 ++++- tests/test_jsonl_validation.py | 72 +++++++++++++++++++++++++++++++ utils/jsonl_parser.py | 8 ++-- utils/validation.py | 78 ++++++++++++++++++++++++++++++++++ 4 files changed, 163 insertions(+), 5 deletions(-) create mode 100644 tests/test_jsonl_validation.py create mode 100644 utils/validation.py diff --git a/models/errors.py b/models/errors.py index 6af19dd..85d8b1a 100644 --- a/models/errors.py +++ b/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}") diff --git a/tests/test_jsonl_validation.py b/tests/test_jsonl_validation.py new file mode 100644 index 0000000..69d0228 --- /dev/null +++ b/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" diff --git a/utils/jsonl_parser.py b/utils/jsonl_parser.py index 51ce8d9..14b71c3 100644 --- a/utils/jsonl_parser.py +++ b/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, - }, + } ) diff --git a/utils/validation.py b/utils/validation.py new file mode 100644 index 0000000..20e2f58 --- /dev/null +++ b/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, + }, + ) From 49393b29b739a7c4e3db26af360419ca852ede06 Mon Sep 17 00:00:00 2001 From: chen Date: Tue, 26 May 2026 01:48:58 +0800 Subject: [PATCH 2/4] fix: expose SessionValidationError.detail and fix test mypy types --- models/errors.py | 1 + tests/test_jsonl_validation.py | 5 +++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/models/errors.py b/models/errors.py index 85d8b1a..40c3eb3 100644 --- a/models/errors.py +++ b/models/errors.py @@ -12,4 +12,5 @@ class SessionValidationError(ValueError): def __init__(self, path: str, detail: str) -> None: self.path = path + self.detail = detail super().__init__(f"Session validation failed at {path}: {detail}") diff --git a/tests/test_jsonl_validation.py b/tests/test_jsonl_validation.py index 69d0228..a6f0ab4 100644 --- a/tests/test_jsonl_validation.py +++ b/tests/test_jsonl_validation.py @@ -2,6 +2,7 @@ import os import sys +from typing import Any import pytest @@ -14,8 +15,8 @@ FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") -def _valid_payload(**overrides: object) -> dict: - base = { +def _valid_payload(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { "session_id": "abc123", "title": "Test Session", "messages": [{"role": "user", "text": "hello"}], From 131194f081b72692eb269512da696e212765edfd Mon Sep 17 00:00:00 2001 From: chen Date: Tue, 26 May 2026 03:32:34 +0800 Subject: [PATCH 3/4] fix: address PR #49 review (preserve dict keys, role test, parser type) Return validated input dict instead of reconstructing four keys. Add test_missing_role_in_message and annotate build_cli_parser(). --- app.py | 5 ++--- tests/test_jsonl_validation.py | 7 +++++++ utils/validation.py | 14 ++++---------- 3 files changed, 13 insertions(+), 13 deletions(-) diff --git a/app.py b/app.py index fcb25e5..634442b 100644 --- a/app.py +++ b/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") diff --git a/tests/test_jsonl_validation.py b/tests/test_jsonl_validation.py index a6f0ab4..3f979d1 100644 --- a/tests/test_jsonl_validation.py +++ b/tests/test_jsonl_validation.py @@ -46,6 +46,13 @@ def test_null_role_in_message(self): ) 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")) diff --git a/utils/validation.py b/utils/validation.py index 20e2f58..4a95901 100644 --- a/utils/validation.py +++ b/utils/validation.py @@ -3,13 +3,14 @@ from typing import Any, cast from models.errors import SessionValidationError -from models.session import MessageDict, SessionDict +from models.session import 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.""" + # Runtime guard for dynamic callers; mypy already types the parameter as dict. if not isinstance(data, dict): raise SessionValidationError("$", "expected dict") @@ -18,6 +19,7 @@ def validate_session_dict(data: dict[str, Any]) -> SessionDict: raise SessionValidationError(key, "missing required field") session_id = data["session_id"] + # Explicit null check before isinstance so errors say "must not be null". if session_id is None: raise SessionValidationError("session_id", "must not be null") if not isinstance(session_id, str): @@ -67,12 +69,4 @@ def validate_session_dict(data: dict[str, Any]) -> SessionDict: "metadata", f"expected dict, got {type(metadata).__name__}" ) - return cast( - SessionDict, - { - "session_id": session_id, - "title": title, - "messages": cast(list[MessageDict], messages), - "metadata": metadata, - }, - ) + return cast(SessionDict, data) From c7122dce37ff0327dedf041ffea00d49f7ee8c1c Mon Sep 17 00:00:00 2001 From: chen Date: Tue, 26 May 2026 05:26:53 +0800 Subject: [PATCH 4/4] refactor(validation): DRY field checks, clearer root path, null test --- tests/test_jsonl_validation.py | 6 +++ utils/validation.py | 90 ++++++++++++++-------------------- 2 files changed, 43 insertions(+), 53 deletions(-) diff --git a/tests/test_jsonl_validation.py b/tests/test_jsonl_validation.py index 3f979d1..c9899a2 100644 --- a/tests/test_jsonl_validation.py +++ b/tests/test_jsonl_validation.py @@ -39,6 +39,12 @@ def test_wrong_type_session_id(self): 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( diff --git a/utils/validation.py b/utils/validation.py index 4a95901..3d7dcf7 100644 --- a/utils/validation.py +++ b/utils/validation.py @@ -5,68 +5,52 @@ from models.errors import SessionValidationError from models.session import SessionDict -_REQUIRED_SESSION_KEYS = ("session_id", "title", "messages", "metadata") +_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("$", "expected dict") - - for key in _REQUIRED_SESSION_KEYS: - if key not in data: - raise SessionValidationError(key, "missing required field") - - session_id = data["session_id"] - # Explicit null check before isinstance so errors say "must not be null". - 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__}" - ) + raise SessionValidationError(_ROOT_PATH, "expected dict") - 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__}" - ) + _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}]" - 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__}" - ) + msg_dict = _require_value(path, message, dict, "dict") + _require_field(msg_dict, "role", str, "str", path=f"{path}.role") return cast(SessionDict, data)