SessionMetadataDict: enforce required fields for mypy shape checking

Promote session_id, models_used, and first_timestamp to required TypedDict keys; keep extended accounting fields as NotRequired. Add _finalize_session_metadata() as the single typed construction site and document the required/optional contract.

Author: clean6378-max-it

models/session.py

@@ -52,36 +52,49 @@ 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]]
 
 
 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_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

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,7 @@
 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 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,6 +70,41 @@ def _safe_int(val: Any) -> int:
     return 0
 
 
+def _finalize_session_metadata(raw: dict[str, Any]) -> SessionMetadataDict:
+    """Convert the mutable parse-time metadata builder into a SessionMetadataDict."""
+    return {
+        "session_id": raw["session_id"],
+        "models_used": sorted(raw["models_used"]),
+        "first_timestamp": raw["first_timestamp"],
+        "last_timestamp": raw["last_timestamp"],
+        "total_input_tokens": raw["total_input_tokens"],
+        "total_output_tokens": raw["total_output_tokens"],
+        "total_cache_read_tokens": raw["total_cache_read_tokens"],
+        "total_cache_creation_tokens": raw["total_cache_creation_tokens"],
+        "total_tool_calls": raw["total_tool_calls"],
+        "tool_call_counts": raw["tool_call_counts"],
+        "version": raw["version"],
+        "cwd": raw["cwd"],
+        "git_branch": raw["git_branch"],
+        "permission_mode": raw["permission_mode"],
+        "compactions": raw["compactions"],
+        "total_ephemeral_5m_tokens": raw["total_ephemeral_5m_tokens"],
+        "total_ephemeral_1h_tokens": raw["total_ephemeral_1h_tokens"],
+        "service_tiers": sorted(raw["service_tiers"]),
+        "session_wall_time_seconds": raw["session_wall_time_seconds"],
+        "compact_boundaries": raw["compact_boundaries"],
+        "api_errors": raw["api_errors"],
+        "files_read": sorted(raw["files_read"]),
+        "files_written": sorted(raw["files_written"]),
+        "files_created": sorted(raw["files_created"]),
+        "bash_commands": raw["bash_commands"],
+        "web_fetches": raw["web_fetches"],
+        "sidechain_messages": raw["sidechain_messages"],
+        "stop_reasons": raw["stop_reasons"],
+        "entry_counts": raw["entry_counts"],
+    }
+
+
 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),
@@ -165,12 +200,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 +218,7 @@ def parse_session(filepath: str) -> SessionDict:
             "session_id": session_id,
             "title": title,
             "messages": messages,
-            "metadata": metadata,
+            "metadata": _finalize_session_metadata(metadata),
         }
     )