test(server): add unit tests for conversation mtime tracking

Comprehensive test coverage for _get_conversation_mtime() and its
integration with _collect_sessions().

Test cases:
- Claude sessions: file exists, file missing, OSError on stat,
  missing project_dir, missing session_id
- Codex sessions: transcript_path exists, transcript_path missing,
  discovery fallback, discovery returns None, OSError on stat
- Edge cases: unknown agent type, missing agent key
- Integration: mtime included when file exists, omitted when missing
- Change detection: mtime changes trigger payload hash changes

Uses mock patching to isolate file system interactions and test
error handling paths without requiring actual conversation files.

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

tests/test_conversation_mtime.py

@@ -0,0 +1,357 @@
+"""Tests for conversation_mtime_ns feature in state.py.
+
+This feature enables real-time dashboard updates by tracking the conversation
+file's modification time, which changes on every write (tool call, message, etc.),
+rather than relying solely on hook events which only fire at specific moments.
+"""
+
+import json
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from amc_server.mixins.state import StateMixin
+from amc_server.mixins.parsing import SessionParsingMixin
+from amc_server.mixins.discovery import SessionDiscoveryMixin
+
+
+class CombinedMixin(StateMixin, SessionParsingMixin, SessionDiscoveryMixin):
+    """Combined mixin for testing - mirrors AMCHandler's inheritance."""
+    pass
+
+
+class TestGetConversationMtime(unittest.TestCase):
+    """Tests for _get_conversation_mtime method."""
+
+    def setUp(self):
+        self.handler = CombinedMixin()
+        self.temp_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_claude_session_with_existing_file(self):
+        """When conversation file exists, returns its mtime_ns."""
+        # Create a temp conversation file
+        conv_file = Path(self.temp_dir) / "test-session.jsonl"
+        conv_file.write_text('{"type": "user"}\n')
+        expected_mtime = conv_file.stat().st_mtime_ns
+
+        session_data = {
+            "agent": "claude",
+            "session_id": "test-session",
+            "project_dir": "/some/project",
+        }
+
+        with patch.object(
+            self.handler, "_get_claude_conversation_file", return_value=conv_file
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertEqual(result, expected_mtime)
+
+    def test_claude_session_file_not_found(self):
+        """When _get_claude_conversation_file returns None, returns None."""
+        session_data = {
+            "agent": "claude",
+            "session_id": "nonexistent",
+            "project_dir": "/some/project",
+        }
+
+        with patch.object(
+            self.handler, "_get_claude_conversation_file", return_value=None
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_claude_session_oserror_on_stat(self):
+        """When stat() raises OSError, returns None gracefully."""
+        mock_file = MagicMock()
+        mock_file.stat.side_effect = OSError("Permission denied")
+
+        session_data = {
+            "agent": "claude",
+            "session_id": "test-session",
+            "project_dir": "/some/project",
+        }
+
+        with patch.object(
+            self.handler, "_get_claude_conversation_file", return_value=mock_file
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_claude_session_missing_project_dir(self):
+        """When project_dir is empty, _get_claude_conversation_file returns None."""
+        session_data = {
+            "agent": "claude",
+            "session_id": "test-session",
+            "project_dir": "",
+        }
+
+        # Real method will return None for empty project_dir
+        result = self.handler._get_conversation_mtime(session_data)
+        self.assertIsNone(result)
+
+    def test_claude_session_missing_session_id(self):
+        """When session_id is empty, returns None."""
+        session_data = {
+            "agent": "claude",
+            "session_id": "",
+            "project_dir": "/some/project",
+        }
+
+        # _get_claude_conversation_file needs both session_id and project_dir
+        with patch.object(
+            self.handler, "_get_claude_conversation_file", return_value=None
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_codex_session_with_transcript_path(self):
+        """When transcript_path is provided and exists, returns its mtime_ns."""
+        transcript_file = Path(self.temp_dir) / "codex-transcript.jsonl"
+        transcript_file.write_text('{"type": "response_item"}\n')
+        expected_mtime = transcript_file.stat().st_mtime_ns
+
+        session_data = {
+            "agent": "codex",
+            "session_id": "codex-123",
+            "transcript_path": str(transcript_file),
+        }
+
+        result = self.handler._get_conversation_mtime(session_data)
+        self.assertEqual(result, expected_mtime)
+
+    def test_codex_session_transcript_path_missing_file(self):
+        """When transcript_path points to nonexistent file, falls back to discovery."""
+        session_data = {
+            "agent": "codex",
+            "session_id": "codex-123",
+            "transcript_path": "/nonexistent/path.jsonl",
+        }
+
+        # Mock the discovery fallback
+        with patch.object(
+            self.handler, "_find_codex_transcript_file", return_value=None
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_codex_session_discovery_fallback(self):
+        """When transcript_path not provided, uses _find_codex_transcript_file."""
+        transcript_file = Path(self.temp_dir) / "discovered-transcript.jsonl"
+        transcript_file.write_text('{"type": "response_item"}\n')
+        expected_mtime = transcript_file.stat().st_mtime_ns
+
+        session_data = {
+            "agent": "codex",
+            "session_id": "codex-456",
+            # No transcript_path
+        }
+
+        with patch.object(
+            self.handler, "_find_codex_transcript_file", return_value=transcript_file
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertEqual(result, expected_mtime)
+
+    def test_codex_session_discovery_returns_none(self):
+        """When discovery finds nothing, returns None."""
+        session_data = {
+            "agent": "codex",
+            "session_id": "codex-789",
+        }
+
+        with patch.object(
+            self.handler, "_find_codex_transcript_file", return_value=None
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_codex_session_oserror_on_transcript_stat(self):
+        """When stat() on discovered transcript raises OSError, returns None."""
+        mock_file = MagicMock()
+        mock_file.stat.side_effect = OSError("I/O error")
+
+        session_data = {
+            "agent": "codex",
+            "session_id": "codex-err",
+        }
+
+        with patch.object(
+            self.handler, "_find_codex_transcript_file", return_value=mock_file
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertIsNone(result)
+
+    def test_unknown_agent_returns_none(self):
+        """When agent is neither 'claude' nor 'codex', returns None."""
+        session_data = {
+            "agent": "unknown_agent",
+            "session_id": "test-123",
+            "project_dir": "/some/project",
+        }
+
+        result = self.handler._get_conversation_mtime(session_data)
+        self.assertIsNone(result)
+
+    def test_missing_agent_returns_none(self):
+        """When agent key is missing, returns None."""
+        session_data = {
+            "session_id": "test-123",
+            "project_dir": "/some/project",
+        }
+
+        result = self.handler._get_conversation_mtime(session_data)
+        self.assertIsNone(result)
+
+    def test_mtime_changes_on_file_modification(self):
+        """Verify mtime actually changes when file is modified."""
+        conv_file = Path(self.temp_dir) / "changing-file.jsonl"
+        conv_file.write_text('{"type": "user"}\n')
+        mtime_1 = conv_file.stat().st_mtime_ns
+
+        # Small delay to ensure filesystem mtime granularity is captured
+        time.sleep(0.01)
+        conv_file.write_text('{"type": "user"}\n{"type": "assistant"}\n')
+        mtime_2 = conv_file.stat().st_mtime_ns
+
+        session_data = {
+            "agent": "claude",
+            "session_id": "test-session",
+            "project_dir": "/some/project",
+        }
+
+        with patch.object(
+            self.handler, "_get_claude_conversation_file", return_value=conv_file
+        ):
+            result = self.handler._get_conversation_mtime(session_data)
+
+        self.assertEqual(result, mtime_2)
+        self.assertNotEqual(mtime_1, mtime_2)
+
+
+class TestCollectSessionsIntegration(unittest.TestCase):
+    """Integration tests verifying conversation_mtime_ns is included in session data."""
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.sessions_dir = Path(self.temp_dir) / "sessions"
+        self.sessions_dir.mkdir()
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_collect_sessions_includes_mtime_when_available(self):
+        """_collect_sessions adds conversation_mtime_ns when file exists."""
+        handler = CombinedMixin()
+
+        # Create a session file
+        session_file = self.sessions_dir / "test-session.json"
+        session_data = {
+            "session_id": "test-session",
+            "agent": "claude",
+            "project_dir": "/test/project",
+            "status": "active",
+            "last_event_at": "2024-01-01T00:00:00Z",
+        }
+        session_file.write_text(json.dumps(session_data))
+
+        # Create a conversation file
+        conv_file = Path(self.temp_dir) / "conversation.jsonl"
+        conv_file.write_text('{"type": "user"}\n')
+        expected_mtime = conv_file.stat().st_mtime_ns
+
+        with patch("amc_server.mixins.state.SESSIONS_DIR", self.sessions_dir), \
+             patch("amc_server.mixins.state.EVENTS_DIR", Path(self.temp_dir) / "events"), \
+             patch.object(handler, "_discover_active_codex_sessions"), \
+             patch.object(handler, "_get_active_zellij_sessions", return_value=None), \
+             patch.object(handler, "_get_context_usage_for_session", return_value=None), \
+             patch.object(handler, "_get_claude_conversation_file", return_value=conv_file):
+
+            sessions = handler._collect_sessions()
+
+        self.assertEqual(len(sessions), 1)
+        self.assertEqual(sessions[0]["session_id"], "test-session")
+        self.assertEqual(sessions[0]["conversation_mtime_ns"], expected_mtime)
+
+    def test_collect_sessions_omits_mtime_when_file_missing(self):
+        """_collect_sessions does not add conversation_mtime_ns when file doesn't exist."""
+        handler = CombinedMixin()
+
+        session_file = self.sessions_dir / "no-conv-session.json"
+        session_data = {
+            "session_id": "no-conv-session",
+            "agent": "claude",
+            "project_dir": "/test/project",
+            "status": "active",
+            "last_event_at": "2024-01-01T00:00:00Z",
+        }
+        session_file.write_text(json.dumps(session_data))
+
+        with patch("amc_server.mixins.state.SESSIONS_DIR", self.sessions_dir), \
+             patch("amc_server.mixins.state.EVENTS_DIR", Path(self.temp_dir) / "events"), \
+             patch.object(handler, "_discover_active_codex_sessions"), \
+             patch.object(handler, "_get_active_zellij_sessions", return_value=None), \
+             patch.object(handler, "_get_context_usage_for_session", return_value=None), \
+             patch.object(handler, "_get_claude_conversation_file", return_value=None):
+
+            sessions = handler._collect_sessions()
+
+        self.assertEqual(len(sessions), 1)
+        self.assertNotIn("conversation_mtime_ns", sessions[0])
+
+
+class TestDashboardChangeDetection(unittest.TestCase):
+    """Tests verifying the dashboard uses mtime for change detection."""
+
+    def test_mtime_triggers_state_hash_change(self):
+        """When conversation_mtime_ns changes, payload hash should change."""
+        # This is implicitly tested by the SSE mechanism:
+        # state.py builds payload with conversation_mtime_ns
+        # _serve_stream hashes payload and sends on change
+
+        # Simulate two state payloads with different mtimes
+        payload_1 = {
+            "sessions": [{"session_id": "s1", "conversation_mtime_ns": 1000}],
+            "server_time": "2024-01-01T00:00:00Z",
+        }
+        payload_2 = {
+            "sessions": [{"session_id": "s1", "conversation_mtime_ns": 2000}],
+            "server_time": "2024-01-01T00:00:00Z",
+        }
+
+        import hashlib
+        hash_1 = hashlib.sha1(json.dumps(payload_1).encode()).hexdigest()
+        hash_2 = hashlib.sha1(json.dumps(payload_2).encode()).hexdigest()
+
+        self.assertNotEqual(hash_1, hash_2)
+
+    def test_same_mtime_same_hash(self):
+        """When mtime hasn't changed, hash should be stable."""
+        payload = {
+            "sessions": [{"session_id": "s1", "conversation_mtime_ns": 1000}],
+            "server_time": "2024-01-01T00:00:00Z",
+        }
+
+        import hashlib
+        hash_1 = hashlib.sha1(json.dumps(payload).encode()).hexdigest()
+        hash_2 = hashlib.sha1(json.dumps(payload).encode()).hexdigest()
+
+        self.assertEqual(hash_1, hash_2)
+
+
+if __name__ == "__main__":
+    unittest.main()