claude-code-chat-browser: SessionMetadataDict — tighten all-NotRequired fields for mypy shape enforcement

models/session.py

@@ -52,36 +52,88 @@ class MessageDict(TypedDict):
     parent_tool_use_id: NotRequired[str | None]
 
 
-class SessionMetadataDict(TypedDict, total=False):
+class SessionMetadataDict(TypedDict):
+    """Metadata accumulated while parsing a Claude Code JSONL session.
+
+    Required keys are always present after ``parse_session()``:
+
+    - ``session_id`` — derived from the ``.jsonl`` filename (stable identity).
+    - ``models_used`` — model names from assistant messages (empty list when none).
+    - ``first_timestamp`` — ISO timestamp of the earliest entry, or ``None`` when
+      the file has no timestamps.
+
+    Remaining fields are optional in partial or stub data (tests, export filters)
+    but are populated with defaults by the parser for full sessions.
+    """
+
     session_id: str
     models_used: list[str]
-    total_input_tokens: int
-    total_output_tokens: int
-    total_cache_read_tokens: int
-    total_cache_creation_tokens: int
-    total_tool_calls: int
-    tool_call_counts: dict[str, int]
     first_timestamp: str | None
-    last_timestamp: str | None
-    version: str | None
-    cwd: str | None
-    git_branch: str | None
-    permission_mode: str | None
-    compactions: int
-    total_ephemeral_5m_tokens: int
-    total_ephemeral_1h_tokens: int
-    service_tiers: list[str]
-    session_wall_time_seconds: float | None
-    compact_boundaries: list[dict[str, Any]]
-    api_errors: int
-    files_read: list[str]
-    files_written: list[str]
-    files_created: list[str]
-    bash_commands: list[Any]
-    web_fetches: list[Any]
-    sidechain_messages: int
-    stop_reasons: dict[str, int]
-    entry_counts: dict[str, int]
+    last_timestamp: NotRequired[str | None]
+    total_input_tokens: NotRequired[int]
+    total_output_tokens: NotRequired[int]
+    total_cache_read_tokens: NotRequired[int]
+    total_cache_creation_tokens: NotRequired[int]
+    total_tool_calls: NotRequired[int]
+    tool_call_counts: NotRequired[dict[str, int]]
+    version: NotRequired[str | None]
+    cwd: NotRequired[str | None]
+    git_branch: NotRequired[str | None]
+    permission_mode: NotRequired[str | None]
+    compactions: NotRequired[int]
+    total_ephemeral_5m_tokens: NotRequired[int]
+    total_ephemeral_1h_tokens: NotRequired[int]
+    service_tiers: NotRequired[list[str]]
+    session_wall_time_seconds: NotRequired[float | None]
+    compact_boundaries: NotRequired[list[dict[str, Any]]]
+    api_errors: NotRequired[int]
+    files_read: NotRequired[list[str]]
+    files_written: NotRequired[list[str]]
+    files_created: NotRequired[list[str]]
+    bash_commands: NotRequired[list[Any]]
+    web_fetches: NotRequired[list[Any]]
+    sidechain_messages: NotRequired[int]
+    stop_reasons: NotRequired[dict[str, int]]
+    entry_counts: NotRequired[dict[str, int]]
+
+
+# Canonical metadata field set for parse_session builder / finalize parity.
+# Keep in sync with SessionMetadataDict above.
+SESSION_METADATA_REQUIRED_KEYS = frozenset({"session_id", "models_used", "first_timestamp"})
+
+SESSION_METADATA_FIELD_NAMES = frozenset(
+    {
+        "session_id",
+        "models_used",
+        "first_timestamp",
+        "last_timestamp",
+        "total_input_tokens",
+        "total_output_tokens",
+        "total_cache_read_tokens",
+        "total_cache_creation_tokens",
+        "total_tool_calls",
+        "tool_call_counts",
+        "version",
+        "cwd",
+        "git_branch",
+        "permission_mode",
+        "compactions",
+        "total_ephemeral_5m_tokens",
+        "total_ephemeral_1h_tokens",
+        "service_tiers",
+        "session_wall_time_seconds",
+        "compact_boundaries",
+        "api_errors",
+        "files_read",
+        "files_written",
+        "files_created",
+        "bash_commands",
+        "web_fetches",
+        "sidechain_messages",
+        "stop_reasons",
+        "entry_counts",
+    }
+)
 
 
 class SessionDict(TypedDict):

tests/test_exclusion_helpers.py

@@ -45,7 +45,11 @@ def _session(
 ) -> dict:
     return {
         "title": title,
-        "metadata": {"models_used": models or []},
+        "metadata": {
+            "session_id": "stub",
+            "models_used": models or [],
+            "first_timestamp": None,
+        },
         "messages": messages or [],
     }
 

tests/test_jsonl_parser.py

@@ -7,14 +7,20 @@
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
+from models.session import SESSION_METADATA_FIELD_NAMES
 from utils.jsonl_helpers import (
     extract_images,
     extract_text,
     infer_title,
     normalize_content,
     strip_system_tags,
 )
-from utils.jsonl_parser import parse_session, quick_session_info
+from utils.jsonl_parser import (
+    _finalize_session_metadata,
+    _new_session_metadata_builder,
+    parse_session,
+    quick_session_info,
+)
 from utils.tool_dispatch import _parse_tool_result
 
 # ---------------------------------------------------------------------------
@@ -38,6 +44,18 @@ def _parse_entries(entries: list) -> dict:
         os.unlink(path)
 
 
+class TestSessionMetadataFinalize:
+    def test_builder_keys_match_field_names_constant(self):
+        raw = _new_session_metadata_builder("parity-test")
+        assert set(raw.keys()) == SESSION_METADATA_FIELD_NAMES
+
+    def test_finalize_preserves_all_builder_keys(self):
+        raw = _new_session_metadata_builder("parity-test")
+        finalized = _finalize_session_metadata(raw)
+        assert set(finalized.keys()) == SESSION_METADATA_FIELD_NAMES
+        assert set(finalized.keys()) == set(raw.keys())
+
+
 # ---------------------------------------------------------------------------
 # _parse_tool_result
 # ---------------------------------------------------------------------------

tests/test_jsonl_validation.py

@@ -20,7 +20,7 @@ def _valid_payload(**overrides: Any) -> dict[str, Any]:
         "session_id": "abc123",
         "title": "Test Session",
         "messages": [{"role": "user", "text": "hello"}],
-        "metadata": {"session_id": "abc123"},
+        "metadata": {"session_id": "abc123", "models_used": [], "first_timestamp": None},
     }
     base.update(overrides)
     return base
@@ -60,6 +60,35 @@ def test_metadata_not_dict(self):
             validate_session_dict(_valid_payload(metadata="not-a-dict"))
         assert exc_info.value.path == "metadata"
 
+    def test_metadata_missing_session_id(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(metadata={"models_used": [], "first_timestamp": None})
+            )
+        assert exc_info.value.path == "metadata.session_id"
+
+    def test_metadata_missing_models_used(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(metadata={"session_id": "abc123", "first_timestamp": None})
+            )
+        assert exc_info.value.path == "metadata.models_used"
+
+    def test_metadata_missing_first_timestamp(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(metadata={"session_id": "abc123", "models_used": []})
+            )
+        assert exc_info.value.path == "metadata.first_timestamp"
+
+    def test_metadata_first_timestamp_null_allowed(self):
+        result = validate_session_dict(
+            _valid_payload(
+                metadata={"session_id": "abc123", "models_used": [], "first_timestamp": None}
+            )
+        )
+        assert result["metadata"]["first_timestamp"] is None
+
     def test_message_not_dict(self):
         with pytest.raises(SessionValidationError) as exc_info:
             validate_session_dict(_valid_payload(messages=["not-a-dict"]))

utils/export_engine.py

@@ -176,7 +176,7 @@ def finalize(self, manifest: list[dict[str, Any]]) -> None:
 
 def _resolve_first_timestamp(meta: SessionMetadataDict, sess_info: SessionListItemDict) -> str:
     """Return first_timestamp from metadata, or synthesise from mtime without mutating *meta*."""
-    ts = (meta.get("first_timestamp") or "").strip()
+    ts = (meta["first_timestamp"] or "").strip()
     if not ts:
         ts = datetime.fromtimestamp(sess_info["modified"], tz=timezone.utc).strftime(
             "%Y-%m-%dT%H:%M:%S"

utils/jsonl_parser.py

@@ -9,7 +9,14 @@
 from typing import Any, cast, get_args
 
 from models.record_data import RecordDataUnion
-from models.session import MessageDict, RoleLiteral, SessionDict, ToolUseDict
+from models.session import (
+    SESSION_METADATA_FIELD_NAMES,
+    MessageDict,
+    RoleLiteral,
+    SessionDict,
+    SessionMetadataDict,
+    ToolUseDict,
+)
 from models.tool_results import ToolResultUnion, is_tool_result_dict
 from utils.jsonl_helpers import (
     entry_message as _entry_message,
@@ -70,13 +77,15 @@ def _safe_int(val: Any) -> int:
     return 0
 
 
-def parse_session(filepath: str) -> SessionDict:
-    """Main entry point. Reads every line from a .jsonl file and builds up
-    a session dict with messages, metadata (tokens, models, tool counts),
-    and file/command activity."""
-    session_id = os.path.basename(filepath).replace(".jsonl", "")
-    messages: list[MessageDict] = []
-    metadata: dict[str, Any] = {
+# Set-backed fields are sorted when finalized; all other builder keys pass through as-is.
+_METADATA_SET_FIELDS = frozenset(
+    {"models_used", "service_tiers", "files_read", "files_written", "files_created"}
+)
+
+
+def _new_session_metadata_builder(session_id: str) -> dict[str, Any]:
+    """Fresh mutable metadata accumulator for ``parse_session()``."""
+    return {
         "session_id": session_id,
         "models_used": set(),
         "total_input_tokens": 0,
@@ -92,30 +101,42 @@ def parse_session(filepath: str) -> SessionDict:
         "git_branch": None,
         "permission_mode": None,
         "compactions": 0,
-        # Extended token accounting
         "total_ephemeral_5m_tokens": 0,
         "total_ephemeral_1h_tokens": 0,
         "service_tiers": set(),
-        # Timing
         "session_wall_time_seconds": None,
-        # Compaction details
         "compact_boundaries": [],
-        # Error tracking
         "api_errors": 0,
-        # File activity (from tool_use inputs)
         "files_read": set(),
         "files_written": set(),
         "files_created": set(),
         "bash_commands": [],
         "web_fetches": [],
-        # Sidechain tracking
         "sidechain_messages": 0,
-        # Stop reasons
         "stop_reasons": {},
-        # Entry type counts
         "entry_counts": {},
     }
 
+
+def _finalize_session_metadata(raw: dict[str, Any]) -> SessionMetadataDict:
+    """Convert the mutable parse-time metadata builder into a SessionMetadataDict."""
+    finalized: dict[str, Any] = {}
+    for key in SESSION_METADATA_FIELD_NAMES:
+        val = raw[key]
+        if key in _METADATA_SET_FIELDS:
+            val = sorted(val)
+        finalized[key] = val
+    return cast(SessionMetadataDict, finalized)
+
+
+def parse_session(filepath: str) -> SessionDict:
+    """Main entry point. Reads every line from a .jsonl file and builds up
+    a session dict with messages, metadata (tokens, models, tool counts),
+    and file/command activity."""
+    session_id = os.path.basename(filepath).replace(".jsonl", "")
+    messages: list[MessageDict] = []
+    metadata = _new_session_metadata_builder(session_id)
+
     with open(filepath, "r", encoding="utf-8", errors="replace") as f:
         for line in f:
             line = line.strip()
@@ -137,7 +158,7 @@ def parse_session(filepath: str) -> SessionDict:
                 if isinstance(snap, dict):
                     ts = snap.get("timestamp")
 
-            if ts:
+            if isinstance(ts, str) and ts:
                 if metadata["first_timestamp"] is None:
                     metadata["first_timestamp"] = ts
                 metadata["last_timestamp"] = ts
@@ -165,12 +186,6 @@ def parse_session(filepath: str) -> SessionDict:
                 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"])
-    metadata["files_read"] = sorted(metadata["files_read"])
-    metadata["files_written"] = sorted(metadata["files_written"])
-    metadata["files_created"] = sorted(metadata["files_created"])
-
     # Compute wall clock time
     first_ts = metadata["first_timestamp"]
     last_ts = metadata["last_timestamp"]
@@ -189,7 +204,7 @@ def parse_session(filepath: str) -> SessionDict:
             "session_id": session_id,
             "title": title,
             "messages": messages,
-            "metadata": metadata,
+            "metadata": _finalize_session_metadata(metadata),
         }
     )
 

utils/validation.py

@@ -37,6 +37,23 @@ def _require_value(
     return val
 
 
+def _require_optional_str(path: str, val: Any) -> str | None:
+    if val is None:
+        return None
+    if not isinstance(val, str):
+        raise SessionValidationError(path, f"expected str or null, got {type(val).__name__}")
+    return val
+
+
+def _validate_session_metadata(metadata: dict[str, Any]) -> None:
+    """Enforce SessionMetadataDict required keys at the runtime boundary."""
+    _require_field(metadata, "session_id", str, "str", path="metadata.session_id")
+    _require_field(metadata, "models_used", list, "list", path="metadata.models_used")
+    if "first_timestamp" not in metadata:
+        raise SessionValidationError("metadata.first_timestamp", "missing required field")
+    _require_optional_str("metadata.first_timestamp", metadata["first_timestamp"])
+
+
 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.
@@ -46,7 +63,8 @@ def validate_session_dict(data: dict[str, Any]) -> SessionDict:
     _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")
+    metadata = _require_field(data, "metadata", dict, "dict")
+    _validate_session_metadata(metadata)
 
     for index, message in enumerate(messages):
         path = f"messages[{index}]"