Miscellaneous bug fixes (#10)

* Fix code review findings for project name handling and type safety

- Expand bad project name detection to include _tmp and _var prefixes
- Add marker directories (code, projects, etc.) to fallback skip list
- Harden escape_html to handle non-string values gracefully
- Add type validation and exception handling in extract_project_from_cwd
- Add comprehensive unit tests for all fixes

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Fix review findings: accept PathLike, allow project names like src/code

- Accept Path and os.PathLike in extract_project_from_cwd using os.fspath
- Only skip true system directories (users, home, var, tmp, private) in
  fallback; names like code, src, dev are now valid project names
- Fix test_non_string_with_html_chars to use custom object with HTML in __str__
- Add test for Path objects in extract_project_from_cwd

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: wesm

agent_session_viewer/main.py

@@ -301,7 +301,14 @@ async def export_session(session_id: str):
 
 
 def escape_html(text: str) -> str:
-    """Escape HTML special characters."""
+    """Escape HTML special characters.
+
+    Handles non-string values gracefully by coercing to string.
+    """
+    if text is None:
+        return ""
+    if not isinstance(text, str):
+        text = str(text)
     if not text:
         return ""
     return (

agent_session_viewer/parser.py

@@ -1,9 +1,10 @@
 """Parse Claude Code JSONL session files."""
 
 import json
+import os
 from pathlib import Path
 from datetime import datetime
-from typing import Optional, Generator
+from typing import Optional, Generator, Union
 from dataclasses import dataclass
 
 
@@ -246,18 +247,27 @@ def normalize_project_name(name: str) -> str:
     return name.replace("-", "_") if name else ""
 
 
-def extract_project_from_cwd(cwd: str) -> str:
+def extract_project_from_cwd(cwd: Union[str, os.PathLike]) -> str:
     """Extract project name from a cwd path.
 
     Uses the last component of the path as the project name.
     Works for both Claude Code and Codex sessions.
     Guards against unsafe path components like . and ..
     Normalizes the name (replaces - with _) for consistency.
+    Accepts str, Path, or any os.PathLike object.
     """
     if not cwd:
         return ""
-    path = Path(cwd)
-    name = path.name or ""
+    try:
+        # Normalize PathLike objects to string, then to Path
+        if isinstance(cwd, os.PathLike):
+            cwd = os.fspath(cwd)
+        if not isinstance(cwd, str):
+            return ""
+        path = Path(cwd)
+        name = path.name or ""
+    except (ValueError, TypeError):
+        return ""
 
     # Guard against unsafe path components
     if name in (".", "..", "") or "/" in name or "\\" in name:

agent_session_viewer/sync.py

@@ -72,8 +72,10 @@ def get_project_name(dir_path: Path) -> str:
 
     # No marker found - take the last non-empty part as the project name
     # This handles cases like -Users-wesm -> wesm
+    # Only skip true system directories; names like "code", "src", "dev" are valid project names
+    system_dirs = {"users", "home", "var", "tmp", "private"}
     for part in reversed(parts):
-        if part and part.lower() not in ("users", "home", "var", "tmp", "private"):
+        if part and part.lower() not in system_dirs:
             return part.replace("-", "_")
 
     # Ultimate fallback
@@ -209,11 +211,16 @@ def sync_session_file(
                 stored_project = stored_session.get("project", "") if stored_session else ""
 
                 # Detect bad project names that look like encoded paths
+                # Covers: _Users*, _home*, _private*, _tmp*, _var*
+                # and paths containing system directory segments
                 needs_reparse = (
                     stored_project.startswith("_Users") or
                     stored_project.startswith("_home") or
                     stored_project.startswith("_private") or
-                    "_var_folders_" in stored_project
+                    stored_project.startswith("_tmp") or
+                    stored_project.startswith("_var") or
+                    "_var_folders_" in stored_project or
+                    "_var_tmp_" in stored_project
                 )
 
                 if not needs_reparse:

tests/test_main.py

@@ -98,6 +98,35 @@ def test_normal_text(self):
         """Normal text should pass through."""
         assert escape_html("Hello world") == "Hello world"
 
+    def test_non_string_integer(self):
+        """Integer should be coerced to string."""
+        assert escape_html(42) == "42"
+        assert escape_html(0) == "0"
+
+    def test_non_string_list(self):
+        """List should be coerced to string."""
+        result = escape_html(["a", "b"])
+        assert isinstance(result, str)
+        assert result  # Not empty
+
+    def test_non_string_dict(self):
+        """Dict should be coerced to string."""
+        result = escape_html({"key": "value"})
+        assert isinstance(result, str)
+        assert result  # Not empty
+
+    def test_non_string_with_html_chars(self):
+        """Non-string values that produce HTML chars should be escaped."""
+        # Use a custom object whose __str__ returns HTML characters
+        class HtmlObject:
+            def __str__(self):
+                return "<tag>content</tag>"
+
+        result = escape_html(HtmlObject())
+        assert "&lt;tag&gt;" in result
+        assert "&lt;/tag&gt;" in result
+        assert "<tag>" not in result
+
 
 class TestGenerateExportHtml:
     """Tests for HTML export generation."""
@@ -143,3 +172,41 @@ def test_url_encoding_in_filename(self):
         filename = sanitize_filename("my project#1.html")
         assert '"' not in filename
         assert '\n' not in filename
+
+    def test_non_string_role_does_not_crash(self):
+        """Non-string role values should not crash export."""
+        session = {"project": "test", "agent": "claude", "message_count": 1}
+        messages = [{"role": 123, "content": "test", "timestamp": ""}]
+
+        # Should not raise, should produce valid HTML
+        html = generate_export_html(session, messages)
+        assert "123" in html
+        assert isinstance(html, str)
+
+    def test_non_string_agent_does_not_crash(self):
+        """Non-string agent values should not crash export."""
+        session = {"project": "test", "agent": 456, "message_count": 1}
+        messages = []
+
+        # Should not raise, should produce valid HTML
+        html = generate_export_html(session, messages)
+        assert "456" in html
+        assert isinstance(html, str)
+
+    def test_none_role_does_not_crash(self):
+        """None role values should not crash export."""
+        session = {"project": "test", "agent": "claude", "message_count": 1}
+        messages = [{"role": None, "content": "test", "timestamp": ""}]
+
+        # Should not raise
+        html = generate_export_html(session, messages)
+        assert isinstance(html, str)
+
+    def test_missing_role_does_not_crash(self):
+        """Missing role key should not crash export."""
+        session = {"project": "test", "agent": "claude", "message_count": 1}
+        messages = [{"content": "test", "timestamp": ""}]
+
+        # Should not raise
+        html = generate_export_html(session, messages)
+        assert isinstance(html, str)

tests/test_sync.py

@@ -395,3 +395,181 @@ def test_extract_project_from_cwd_rejects_dotdot(self):
     def test_extract_project_from_cwd_no_hyphens_passthrough(self):
         """Project names without hyphens should pass through."""
         assert extract_project_from_cwd("/Users/user/myproject") == "myproject"
+
+    def test_extract_project_from_cwd_non_string_types(self):
+        """Should return empty string for invalid types like int, list, dict."""
+        assert extract_project_from_cwd(123) == ""
+        assert extract_project_from_cwd(["list"]) == ""
+        assert extract_project_from_cwd({"dict": "value"}) == ""
+
+    def test_extract_project_from_cwd_accepts_path_objects(self):
+        """Should accept Path and os.PathLike objects."""
+        assert extract_project_from_cwd(Path("/Users/user/my-app")) == "my_app"
+        assert extract_project_from_cwd(Path("/Users/user/Projects/project")) == "project"
+
+    def test_extract_project_from_cwd_embedded_null(self):
+        """Should handle strings with embedded null safely."""
+        # Path() may raise or behave unexpectedly with embedded NUL
+        result = extract_project_from_cwd("/Users/user\x00/project")
+        # Just verify it doesn't crash - result may vary by platform
+        assert isinstance(result, str)
+
+
+class TestGetProjectNameFallback:
+    """Tests for get_project_name fallback behavior."""
+
+    def test_accepts_marker_directories_as_project_names(self):
+        """Common directory names (code, src, etc.) are valid project names."""
+        # These are legitimate project names when they're the working directory
+        assert get_project_name(Path("-Users-wesm-code")) == "code"
+        assert get_project_name(Path("-Users-wesm-projects")) == "projects"
+        assert get_project_name(Path("-Users-wesm-repos")) == "repos"
+        assert get_project_name(Path("-Users-wesm-src")) == "src"
+        assert get_project_name(Path("-Users-wesm-work")) == "work"
+        assert get_project_name(Path("-Users-wesm-dev")) == "dev"
+
+    def test_skips_system_directories_in_fallback(self):
+        """System directories (users, home, var, tmp, private) should be skipped."""
+        assert get_project_name(Path("-Users-wesm")) == "wesm"
+        assert get_project_name(Path("-home-ubuntu")) == "ubuntu"
+        # "home" is a system directory and should be skipped even if it appears last
+        # (unlike "code", "src" which are valid project names)
+        assert get_project_name(Path("-Users-wesm-home")) == "wesm"
+        assert get_project_name(Path("-Users-wesm-tmp")) == "wesm"
+        assert get_project_name(Path("-Users-wesm-var")) == "wesm"
+
+    def test_marker_extraction_takes_precedence(self):
+        """When a marker has content after it, that content is the project name."""
+        # code/my-app -> my_app (marker extraction works)
+        assert get_project_name(Path("-Users-wesm-code-my-app")) == "my_app"
+        assert get_project_name(Path("-Users-wesm-src-project")) == "project"
+
+
+class TestReparseBehavior:
+    """Tests for reparse detection of bad project names."""
+
+    def test_sync_reparses_users_prefix(self, tmp_path):
+        """Sessions with _Users* project names should be reparsed."""
+        from agent_session_viewer import db
+        from agent_session_viewer import sync as sync_module
+
+        # Create a session file with cwd
+        project_dir = tmp_path / "-Users-alice-code-my-app"
+        project_dir.mkdir()
+        session_file = project_dir / "test-session.jsonl"
+        session_file.write_text(
+            '{"type": "user", "cwd": "/Users/alice/code/my-app", "message": {"content": "hello"}, "timestamp": "2025-01-01T00:00:00Z"}\n'
+        )
+
+        # Mock db functions and CLAUDE_PROJECTS_DIR
+        with patch.object(sync_module, "CLAUDE_PROJECTS_DIR", tmp_path), \
+             patch.object(sync_module, "SESSIONS_DIR", tmp_path / "sessions"), \
+             patch.object(db, "get_session_file_info") as mock_file_info, \
+             patch.object(db, "get_session") as mock_get_session, \
+             patch.object(db, "upsert_session") as mock_upsert, \
+             patch.object(db, "delete_session_messages"), \
+             patch.object(db, "insert_messages_batch"):
+
+            # Simulate: file hash matches but stored project is bad
+            file_hash = sync_module.compute_file_hash(session_file)
+            mock_file_info.return_value = (session_file.stat().st_size, file_hash)
+            mock_get_session.return_value = {"project": "_Users_alice_code_my_app"}
+
+            result = sync_module.sync_session_file(
+                session_file, "_Users_alice_code_my_app", "local"
+            )
+
+            # Should have reparsed (not skipped) and updated with correct project name
+            assert result is not None
+            assert result.get("skipped") is False
+            assert result.get("project") == "my_app"
+
+            # Verify upsert was called with the correct project name
+            mock_upsert.assert_called_once()
+            call_kwargs = mock_upsert.call_args.kwargs
+            assert call_kwargs["project"] == "my_app"
+
+    def test_sync_reparses_tmp_prefix(self, tmp_path):
+        """Sessions with _tmp* project names should be reparsed."""
+        from agent_session_viewer import db
+        from agent_session_viewer import sync as sync_module
+
+        project_dir = tmp_path / "-tmp-my-app"
+        project_dir.mkdir()
+        session_file = project_dir / "test-session.jsonl"
+        session_file.write_text(
+            '{"type": "user", "cwd": "/tmp/my-app", "message": {"content": "hello"}, "timestamp": "2025-01-01T00:00:00Z"}\n'
+        )
+
+        with patch.object(sync_module, "CLAUDE_PROJECTS_DIR", tmp_path), \
+             patch.object(sync_module, "SESSIONS_DIR", tmp_path / "sessions"), \
+             patch.object(db, "get_session_file_info") as mock_file_info, \
+             patch.object(db, "get_session") as mock_get_session, \
+             patch.object(db, "upsert_session") as mock_upsert, \
+             patch.object(db, "delete_session_messages"), \
+             patch.object(db, "insert_messages_batch"):
+
+            file_hash = sync_module.compute_file_hash(session_file)
+            mock_file_info.return_value = (session_file.stat().st_size, file_hash)
+            mock_get_session.return_value = {"project": "_tmp_my_app"}
+
+            result = sync_module.sync_session_file(
+                session_file, "_tmp_my_app", "local"
+            )
+
+            assert result is not None
+            assert result.get("skipped") is False
+
+    def test_sync_reparses_var_prefix(self, tmp_path):
+        """Sessions with _var* project names should be reparsed."""
+        from agent_session_viewer import db
+        from agent_session_viewer import sync as sync_module
+
+        project_dir = tmp_path / "-var-tmp-my-app"
+        project_dir.mkdir()
+        session_file = project_dir / "test-session.jsonl"
+        session_file.write_text(
+            '{"type": "user", "cwd": "/var/tmp/my-app", "message": {"content": "hello"}, "timestamp": "2025-01-01T00:00:00Z"}\n'
+        )
+
+        with patch.object(sync_module, "CLAUDE_PROJECTS_DIR", tmp_path), \
+             patch.object(sync_module, "SESSIONS_DIR", tmp_path / "sessions"), \
+             patch.object(db, "get_session_file_info") as mock_file_info, \
+             patch.object(db, "get_session") as mock_get_session, \
+             patch.object(db, "upsert_session") as mock_upsert, \
+             patch.object(db, "delete_session_messages"), \
+             patch.object(db, "insert_messages_batch"):
+
+            file_hash = sync_module.compute_file_hash(session_file)
+            mock_file_info.return_value = (session_file.stat().st_size, file_hash)
+            mock_get_session.return_value = {"project": "_var_tmp_my_app"}
+
+            result = sync_module.sync_session_file(
+                session_file, "_var_tmp_my_app", "local"
+            )
+
+            assert result is not None
+            assert result.get("skipped") is False
+
+    def test_sync_skips_good_project_names(self, tmp_path):
+        """Sessions with good project names should be skipped if hash matches."""
+        from agent_session_viewer import db
+        from agent_session_viewer import sync as sync_module
+
+        project_dir = tmp_path / "my_app"
+        project_dir.mkdir()
+        session_file = project_dir / "test-session.jsonl"
+        session_file.write_text('{"type": "user", "message": {"content": "hello"}}\n')
+
+        with patch.object(sync_module, "CLAUDE_PROJECTS_DIR", tmp_path), \
+             patch.object(db, "get_session_file_info") as mock_file_info, \
+             patch.object(db, "get_session") as mock_get_session:
+
+            file_hash = sync_module.compute_file_hash(session_file)
+            mock_file_info.return_value = (session_file.stat().st_size, file_hash)
+            mock_get_session.return_value = {"project": "my_app"}
+
+            result = sync_module.sync_session_file(session_file, "my_app", "local")
+
+            assert result is not None
+            assert result.get("skipped") is True