feat(ui): add duplicate session ID prevention (#148)

Implements a real-time check on the Session ID input field to prevent accidental data overwrites.

- Adds a validation function that checks if a session ID already exists on the filesystem.
- Displays a clear, non-blocking warning to the user if a duplicate ID is detected.
- Implements an in-memory, time-based cache to ensure the validation check is performant and does not cause UI lag, addressing a key performance concern.
- Includes comprehensive unit tests for both the duplicate detection and the caching mechanism.

This change directly addresses task UX-04.

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: google-labs-jules[bot]

src/ui/process_session_helpers.py

@@ -47,7 +47,7 @@
     - `src.status_tracker`: Status tracking infrastructure
 """
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 import re
 from typing import Any, Callable, Dict, List, Optional, Tuple
@@ -56,6 +56,7 @@
 from src.party_config import PartyConfigManager
 from src.file_tracker import FileProcessingTracker
 from src.status_tracker import StatusTracker
+from src.session_manager import SessionManager
 from src.ui.constants import StatusIndicators as SI
 from src.ui.helpers import StatusMessages
 
@@ -64,6 +65,17 @@
 ALLOWED_AUDIO_EXTENSIONS: Tuple[str, ...] = (".m4a", ".mp3", ".wav", ".flac")
 SESSION_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
 
+# Cache for session IDs to avoid frequent I/O in real-time validation
+# Why: The session ID validation runs on every keystroke. Hitting the filesystem
+# to check for duplicates each time would cause significant UI lag, especially
+# with many sessions. This cache provides a near-instant check, refreshing
+# only periodically.
+_session_id_cache = {
+    "timestamp": None,
+    "ids": set(),
+}
+CACHE_TTL_SECONDS = 5  # Cache session IDs for 5 seconds
+
 
 def validate_session_id_realtime(session_id: str) -> str:
     """
@@ -82,9 +94,7 @@ def validate_session_id_realtime(session_id: str) -> str:
     session_id = session_id.strip()
 
     # Check against pattern
-    if SESSION_ID_PATTERN.match(session_id):
-        return f"### {SI.SUCCESS} Valid\n\n[v] Session ID `{session_id}` is valid."
-    else:
+    if not SESSION_ID_PATTERN.match(session_id):
         # Find invalid characters, ensuring they are ASCII-compatible to match the regex
         invalid_chars = {
             char for char in session_id
@@ -105,6 +115,32 @@ def validate_session_id_realtime(session_id: str) -> str:
                 "[x] Session ID format is invalid. Use only letters, numbers, underscores (_), and hyphens (-)."
             )
 
+    # UX-04: Add duplicate Session ID prevention with caching.
+    # Why: To prevent UI lag from frequent filesystem access, we check against a
+    # cached set of session IDs.
+    now = datetime.now()
+    cache_is_stale = (
+        _session_id_cache["timestamp"] is None or
+        (now - _session_id_cache["timestamp"]).total_seconds() > CACHE_TTL_SECONDS
+    )
+
+    if cache_is_stale:
+        # Why: This block refreshes the cache if it's empty or has expired.
+        # This is the only time the filesystem is accessed during validation.
+        session_manager = SessionManager()
+        _session_id_cache["ids"] = {s.session_id for s in session_manager.discover_sessions()}
+        _session_id_cache["timestamp"] = now
+
+    if session_id in _session_id_cache["ids"]:
+        return (
+            f"### {SI.WARNING} Duplicate Session ID\n\n"
+            f"[!] Session ID `{session_id}` already exists.\n\n"
+            "Reusing an ID will overwrite existing data. It is recommended to use a unique ID."
+        )
+
+    # If all checks pass, the ID is valid
+    return f"### {SI.SUCCESS} Valid\n\n[v] Session ID `{session_id}` is valid."
+
 
 def validate_processing_readiness(
     audio_file,

tests/ui/test_process_session_helpers.py

@@ -820,6 +820,55 @@ def test_whitespace_session_id(self):
         result = validate_session_id_realtime("   ")
         assert result == ""
 
+    @patch('src.ui.process_session_helpers.SessionManager')
+    def test_duplicate_session_id(self, mock_session_manager):
+        """A session ID that already exists should return a warning."""
+        # This test ensures the core functionality of UX-04, preventing overwrites.
+        mock_instance = mock_session_manager.return_value
+        mock_session = MagicMock()
+        mock_session.session_id = "existing_session"
+        mock_instance.discover_sessions.return_value = [mock_session]
+
+        # Invalidate the cache by resetting its timestamp
+        from src.ui import process_session_helpers
+        process_session_helpers._session_id_cache["timestamp"] = None
+
+        result = validate_session_id_realtime("existing_session")
+
+        assert "Duplicate Session ID" in result
+        assert "`existing_session` already exists" in result
+
+    @patch('src.ui.process_session_helpers.SessionManager')
+    def test_session_id_caching_behavior(self, mock_session_manager):
+        """Test that the session ID cache is used to prevent excessive I/O."""
+        # Why this test is important:
+        # This test directly addresses the performance feedback from the code
+        # review. It ensures that the caching mechanism works, preventing the
+        # expensive `discover_sessions` call from running on every keystroke,
+        # which would cause UI lag.
+
+        # Arrange: Mock the SessionManager.
+        mock_instance = mock_session_manager.return_value
+        mock_session = MagicMock()
+        mock_session.session_id = "cached_session"
+        mock_instance.discover_sessions.return_value = [mock_session]
+
+        # Invalidate the cache to ensure the first call hits the mock.
+        from src.ui import process_session_helpers
+        process_session_helpers._session_id_cache["timestamp"] = None
+
+        # Act 1: First call should populate the cache.
+        validate_session_id_realtime("cached_session")
+
+        # Assert 1: The mock should have been called once.
+        mock_instance.discover_sessions.assert_called_once()
+
+        # Act 2: Second call should use the cache, not the mock.
+        validate_session_id_realtime("cached_session")
+
+        # Assert 2: The mock call count should still be one.
+        mock_instance.discover_sessions.assert_called_once()
+
     def test_update_with_nonexistent_party(self):
         """Test updating display with nonexistent party."""
         with patch('src.ui.process_session_helpers.PartyConfigManager') as mock_manager_class: