Tighten metadata validation and typed builder for parser review feedback

- Require each models_used entry to be a string at validate_session_dict boundary
- Add _entry_timestamp() so invalid top-level timestamps fall back to snapshot
- Introduce SessionMetadataBuilderDict for parse-time accumulator typing
- Return explicit SessionMetadataDict from finalize (no cast)
- Align tool_dispatch file-activity handlers with builder type

Author: clean6378-max-it

models/session.py

@@ -136,6 +136,40 @@ class SessionMetadataDict(TypedDict):
 )
 
 
+class SessionMetadataBuilderDict(TypedDict):
+    """Mutable metadata accumulator during JSONL parsing; sets are sorted at finalize."""
+
+    session_id: str
+    models_used: set[str]
+    first_timestamp: str | None
+    last_timestamp: str | None
+    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]
+    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: set[str]
+    session_wall_time_seconds: float | None
+    compact_boundaries: list[dict[str, Any]]
+    api_errors: int
+    files_read: set[str]
+    files_written: set[str]
+    files_created: set[str]
+    bash_commands: list[Any]
+    web_fetches: list[Any]
+    sidechain_messages: int
+    stop_reasons: dict[str, int]
+    entry_counts: dict[str, int]
+
+
 class SessionDict(TypedDict):
     session_id: str
     title: str

tests/test_jsonl_parser.py

@@ -738,6 +738,23 @@ def test_file_history_snapshot_timestamp(self):
         finally:
             os.unlink(path)
 
+    def test_file_history_snapshot_timestamp_falls_back_when_top_level_invalid(self):
+        path = _write_jsonl(
+            [
+                {
+                    "type": "file-history-snapshot",
+                    "timestamp": 1,
+                    "snapshot": {"timestamp": "2026-01-02T12:00:00Z"},
+                },
+            ]
+        )
+        try:
+            s = parse_session(path)
+            assert s["metadata"]["first_timestamp"] == "2026-01-02T12:00:00Z"
+            assert s["metadata"]["last_timestamp"] == "2026-01-02T12:00:00Z"
+        finally:
+            os.unlink(path)
+
     def test_summary_entry_type_produces_no_message(self, caplog):
         path = _write_jsonl(
             [

tests/test_jsonl_validation.py

@@ -89,6 +89,19 @@ def test_metadata_first_timestamp_null_allowed(self):
         )
         assert result["metadata"]["first_timestamp"] is None
 
+    def test_metadata_models_used_requires_string_elements(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(
+                _valid_payload(
+                    metadata={
+                        "session_id": "abc123",
+                        "models_used": ["claude-sonnet", 42],
+                        "first_timestamp": None,
+                    }
+                )
+            )
+        assert exc_info.value.path == "metadata.models_used[1]"
+
     def test_message_not_dict(self):
         with pytest.raises(SessionValidationError) as exc_info:
             validate_session_dict(_valid_payload(messages=["not-a-dict"]))

utils/jsonl_parser.py

@@ -10,10 +10,10 @@
 
 from models.record_data import RecordDataUnion
 from models.session import (
-    SESSION_METADATA_FIELD_NAMES,
     MessageDict,
     RoleLiteral,
     SessionDict,
+    SessionMetadataBuilderDict,
     SessionMetadataDict,
     ToolUseDict,
 )
@@ -77,13 +77,7 @@ def _safe_int(val: Any) -> int:
     return 0
 
 
-# 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]:
+def _new_session_metadata_builder(session_id: str) -> SessionMetadataBuilderDict:
     """Fresh mutable metadata accumulator for ``parse_session()``."""
     return {
         "session_id": session_id,
@@ -118,15 +112,53 @@ def _new_session_metadata_builder(session_id: str) -> dict[str, Any]:
     }
 
 
-def _finalize_session_metadata(raw: dict[str, Any]) -> SessionMetadataDict:
+def _finalize_session_metadata(raw: SessionMetadataBuilderDict) -> 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)
+    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 _entry_timestamp(entry: dict[str, Any]) -> str | None:
+    """Return a usable ISO timestamp string from an entry, or None."""
+    ts = entry.get("timestamp")
+    if isinstance(ts, str) and ts:
+        return ts
+    if entry.get("type") == "file-history-snapshot":
+        snap = entry.get("snapshot")
+        if isinstance(snap, dict):
+            snap_ts = snap.get("timestamp")
+            if isinstance(snap_ts, str) and snap_ts:
+                return snap_ts
+    return None
 
 
 def parse_session(filepath: str) -> SessionDict:
@@ -151,14 +183,9 @@ def parse_session(filepath: str) -> SessionDict:
                 continue
 
             entry_type = entry.get("type")
-            ts = entry.get("timestamp")
-            # file-history-snapshot stores timestamp inside snapshot
-            if not ts and entry_type == "file-history-snapshot":
-                snap = entry.get("snapshot")
-                if isinstance(snap, dict):
-                    ts = snap.get("timestamp")
-
-            if isinstance(ts, str) and ts:
+            ts = _entry_timestamp(entry)
+
+            if ts:
                 if metadata["first_timestamp"] is None:
                     metadata["first_timestamp"] = ts
                 metadata["last_timestamp"] = ts
@@ -210,7 +237,7 @@ def parse_session(filepath: str) -> SessionDict:
 
 
 def _process_user(
-    entry: dict[str, Any], messages: list[MessageDict], metadata: dict[str, Any]
+    entry: dict[str, Any], messages: list[MessageDict], metadata: SessionMetadataBuilderDict
 ) -> None:
     """Pull out text, tool results, and session-level metadata (cwd, version, etc.)
     from a user entry."""
@@ -257,7 +284,7 @@ def _process_user(
 
 
 def _process_assistant(
-    entry: dict[str, Any], messages: list[MessageDict], metadata: dict[str, Any]
+    entry: dict[str, Any], messages: list[MessageDict], metadata: SessionMetadataBuilderDict
 ) -> None:
     """Handle assistant responses -- splits content into text, thinking blocks,
     and tool_use calls, and accumulates token/model/tool stats."""
@@ -353,7 +380,7 @@ def _process_assistant(
 
 
 def _process_system(
-    entry: dict[str, Any], messages: list[MessageDict], metadata: dict[str, Any]
+    entry: dict[str, Any], messages: list[MessageDict], metadata: SessionMetadataBuilderDict
 ) -> None:
     """Handle system entries (mostly compact_boundary markers from context
     compaction)."""

utils/tool_dispatch.py

@@ -35,6 +35,7 @@
 from collections.abc import Callable
 from typing import Any, cast
 
+from models.session import SessionMetadataBuilderDict
 from models.tool_results import (
     ToolResultDict,
     ToolResultUnion,
@@ -240,40 +241,42 @@ def _tool_result_build_user_input(tr: ToolResultDict, base: dict[str, object]) -
 # ``_FILE_ACTIVITY_HANDLERS`` is the single registry; ``KNOWN_TOOL_TYPES`` is derived.
 
 
-def _file_activity_read(tool_input: dict[str, Any], metadata: dict[str, Any]) -> None:
+def _file_activity_read(tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict) -> None:
     raw_fp = tool_input.get("file_path", "")
     fp = raw_fp if isinstance(raw_fp, str) else ""
     if fp:
         metadata["files_read"].add(fp)
 
 
-def _file_activity_write(tool_input: dict[str, Any], metadata: dict[str, Any]) -> None:
+def _file_activity_write(tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict) -> None:
     raw_fp = tool_input.get("file_path", "")
     fp = raw_fp if isinstance(raw_fp, str) else ""
     if fp:
         metadata["files_created"].add(fp)
 
 
-def _file_activity_edit(tool_input: dict[str, Any], metadata: dict[str, Any]) -> None:
+def _file_activity_edit(tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict) -> None:
     raw_fp = tool_input.get("file_path", "")
     fp = raw_fp if isinstance(raw_fp, str) else ""
     if fp:
         metadata["files_written"].add(fp)
 
 
-def _file_activity_bash(tool_input: dict[str, Any], metadata: dict[str, Any]) -> None:
+def _file_activity_bash(tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict) -> None:
     cmd = tool_input.get("command", "")
     if isinstance(cmd, str) and cmd:
         metadata["bash_commands"].append(cmd)
 
 
-def _file_activity_web(tool_input: dict[str, Any], metadata: dict[str, Any]) -> None:
+def _file_activity_web(tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict) -> None:
     url_or_query = tool_input.get("url") or tool_input.get("query", "")
     if isinstance(url_or_query, str) and url_or_query:
         metadata["web_fetches"].append(url_or_query)
 
 
-_FILE_ACTIVITY_HANDLERS: dict[str, Callable[[dict[str, Any], dict[str, Any]], None] | None] = {
+_FILE_ACTIVITY_HANDLERS: dict[
+    str, Callable[[dict[str, Any], SessionMetadataBuilderDict], None] | None
+] = {
     "AskUserQuestion": None,
     "Bash": _file_activity_bash,
     "Edit": _file_activity_edit,
@@ -290,7 +293,7 @@ def _file_activity_web(tool_input: dict[str, Any], metadata: dict[str, Any]) ->
 
 
 def track_tool_file_activity(
-    tool_name: str, tool_input: dict[str, Any], metadata: dict[str, Any]
+    tool_name: str, tool_input: dict[str, Any], metadata: SessionMetadataBuilderDict
 ) -> None:
     """Record file/bash/web side effects for tools listed in ``KNOWN_TOOL_TYPES``."""
     if tool_name not in KNOWN_TOOL_TYPES:

utils/validation.py

@@ -45,10 +45,24 @@ def _require_optional_str(path: str, val: Any) -> str | None:
     return val
 
 
+def _require_str_list(path: str, val: Any) -> list[str]:
+    if not isinstance(val, list):
+        raise SessionValidationError(path, f"expected list, got {type(val).__name__}")
+    for index, item in enumerate(val):
+        if not isinstance(item, str):
+            raise SessionValidationError(
+                f"{path}[{index}]",
+                f"expected str, got {type(item).__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 "models_used" not in metadata:
+        raise SessionValidationError("metadata.models_used", "missing required field")
+    _require_str_list("metadata.models_used", 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"])