refactor(jsonl): address parser-split review nits

Author: clean6378-max-it

docs/architecture.md

@@ -15,9 +15,9 @@
          ▼                         ▼                         ▼
 ┌─────────────────┐    ┌─────────────────────┐    ┌──────────────────┐
 │ session_path    │    │ jsonl_parser        │    │ exclusion_rules  │
-│ list_projects   │    │ parse_session       │    │ load + match     │
-│ list_sessions   │    │ quick_session_info  │    └────────┬─────────┘
-│ safe_join       │    │ _parse_tool_result  │             │
+│ list_projects   │    │ session_peek        │    │ load + match     │
+│ list_sessions   │    │ tool_dispatch       │    └────────┬─────────┘
+│ safe_join       │    │ jsonl_helpers       │             │
 └────────┬────────┘    └──────────┬──────────┘             │
          │                        │                        │
          └────────────┬───────────┴────────────────────────┘
@@ -48,7 +48,7 @@
 | Layer | Responsibility | Key modules |
 |-------|----------------|-------------|
 | **Data discovery** | Resolve `~/.claude/projects/`, list projects and sessions, prevent path traversal | `utils/session_path.py` |
-| **Parsing** | JSONL → session dict (messages, metadata, tool rendering) | `utils/jsonl_parser.py` |
+| **Parsing** | JSONL → session dict (messages, metadata, tool rendering) | `utils/jsonl_parser.py`, `utils/tool_dispatch.py`, `utils/session_peek.py`, `utils/jsonl_helpers.py` |
 | **Filtering** | Exclude sensitive sessions via rules file | `utils/exclusion_rules.py` |
 | **Statistics** | Aggregates for API and exporters | `utils/session_stats.py` |
 | **Export — Markdown** | Session → YAML-frontmatter Markdown | `utils/md_exporter.py` |

utils/jsonl_helpers.py

@@ -51,6 +51,7 @@ def extract_images(content_parts: Any) -> list[dict[str, Any]]:
                     "data": source["data"],
                 })
         elif part.get("type") == "tool_result":
+            # Nested content is usually a block list; string content is not normalized here.
             nested = part.get("content", [])
             if isinstance(nested, list):
                 for sub in nested:
@@ -64,12 +65,16 @@ def extract_images(content_parts: Any) -> list[dict[str, Any]]:
     return images
 
 
+def first_title_line(text: str, max_chars: int = 100) -> str:
+    """First non-empty line after system-tag strip, truncated for session titles."""
+    return strip_system_tags(text).strip().split("\n")[0][:max_chars]
+
+
 def infer_title(messages: list[MessageDict]) -> str:
     """Use the first line of the first real user message as the session title."""
     for msg in messages:
         if msg["role"] == "user" and msg.get("text"):
-            text = strip_system_tags(msg["text"]).strip()
-            first_line = text.split("\n")[0][:100]
+            first_line = first_title_line(msg["text"])
             if first_line:
                 return first_line
     return "Untitled Session"

utils/jsonl_parser.py

@@ -24,6 +24,7 @@
     "quick_session_info",
     "_parse_tool_result",
     "_TOOL_RESULT_DISPATCH",
+    "_entry_message",
     "_process_user",
     "_process_assistant",
     "_process_system",

utils/session_peek.py

@@ -4,7 +4,7 @@
 import os
 
 from models.session import QuickSessionInfoDict
-from utils.jsonl_helpers import entry_message, extract_text, strip_system_tags
+from utils.jsonl_helpers import entry_message, extract_text, first_title_line
 
 _TAIL_READ_MIN_BYTES = 10 * 1024
 _MAX_HEAD_LINES = 80
@@ -48,8 +48,7 @@ def quick_session_info(filepath: str) -> QuickSessionInfoDict:
                 msg = entry_message(entry)
                 text = extract_text(msg.get("content", []))
                 if text:
-                    clean = strip_system_tags(text).strip()
-                    first_line = clean.split("\n")[0][:100]
+                    first_line = first_title_line(text)
                     if first_line:
                         title = first_line
 
@@ -60,6 +59,7 @@ def quick_session_info(filepath: str) -> QuickSessionInfoDict:
         with open(filepath, "rb") as f:
             f.seek(file_size - chunk_size)
             tail = f.read().decode("utf-8", errors="replace")
+        # First line in tail is often a partial record after seek; json.loads skips it.
         # Parse lines in reverse to find latest timestamp
         for line in reversed(tail.splitlines()):
             line = line.strip()

utils/tool_dispatch.py

@@ -37,13 +37,25 @@ def _tool_result_pred_file_edit(tr: dict[str, Any]) -> bool:
 
 
 def _tool_result_build_file_edit(tr: dict[str, Any], base: dict[str, Any]) -> dict[str, Any]:
+    # Summary fields only; full blob (e.g. structuredPatch) stays on message tool_result.
     result = dict(base)
     result["result_type"] = "file_edit"
     result["file_path"] = tr.get("filePath", "")
     result["replace_all"] = tr.get("replaceAll", False)
     return result
 
 
+def _tool_result_pred_plan(tr: dict[str, Any]) -> bool:
+    return "plan" in tr and "filePath" in tr
+
+
+def _tool_result_build_plan(tr: dict[str, Any], base: dict[str, Any]) -> dict[str, Any]:
+    result = dict(base)
+    result["result_type"] = "plan"
+    result["file_path"] = tr.get("filePath", "")
+    return result
+
+
 def _tool_result_pred_file_write(tr: dict[str, Any]) -> bool:
     return "filePath" in tr and "content" in tr
 
@@ -82,7 +94,7 @@ def _tool_result_build_grep(tr: dict[str, Any], base: dict[str, Any]) -> dict[st
     result["num_lines"] = tr.get("numLines", 0)
     result["duration_ms"] = tr.get("durationMs")
     content = tr.get("content", "")
-    if content and isinstance(content, str):
+    if isinstance(content, str):
         result["content"] = content
     return result
 
@@ -98,7 +110,7 @@ def _tool_result_build_file_read(tr: dict[str, Any], base: dict[str, Any]) -> di
     result["file_path"] = file_obj.get("filePath", "")
     result["num_lines"] = file_obj.get("numLines")
     content = file_obj.get("content", "")
-    if content and isinstance(content, str):
+    if isinstance(content, str):
         result["content"] = content
     return result
 
@@ -217,17 +229,6 @@ def _tool_result_build_user_input(tr: dict[str, Any], base: dict[str, Any]) -> d
     return result
 
 
-def _tool_result_pred_plan(tr: dict[str, Any]) -> bool:
-    return "plan" in tr and "filePath" in tr
-
-
-def _tool_result_build_plan(tr: dict[str, Any], base: dict[str, Any]) -> dict[str, Any]:
-    result = dict(base)
-    result["result_type"] = "plan"
-    result["file_path"] = tr.get("filePath", "")
-    return result
-
-
 # Dispatch registry: **first matching predicate wins** (same as legacy if/elif).
 # Order is load-bearing — do not sort alphabetically or “more specific first”
 # without replaying tests and real session fixtures.
@@ -241,6 +242,8 @@ def _tool_result_build_plan(tr: dict[str, Any], base: dict[str, Any]) -> dict[st
 _TOOL_RESULT_DISPATCH = (
     (_tool_result_pred_bash, _tool_result_build_bash),
     (_tool_result_pred_file_edit, _tool_result_build_file_edit),
+    # plan before file_write: plan blobs may also carry filePath + content
+    (_tool_result_pred_plan, _tool_result_build_plan),
     (_tool_result_pred_file_write, _tool_result_build_file_write),
     (_tool_result_pred_glob, _tool_result_build_glob),
     (_tool_result_pred_grep, _tool_result_build_grep),
@@ -253,7 +256,6 @@ def _tool_result_build_plan(tr: dict[str, Any], base: dict[str, Any]) -> dict[st
     (_tool_result_pred_task_async, _tool_result_build_task_async),
     (_tool_result_pred_todo_write, _tool_result_build_todo_write),
     (_tool_result_pred_user_input, _tool_result_build_user_input),
-    (_tool_result_pred_plan, _tool_result_build_plan),
 )