theirongolddev/amc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dashboard): add click-outside dismissal for autocomplete dropdown

Browse Source 
Closes bd-3ny. Added mousedown listener that dismisses the dropdown when
clicking outside both the dropdown and textarea. Uses early return to avoid
registering listeners when dropdown is already closed.
This commit is contained in:
teernisse
2026-02-26 16:52:36 -05:00
parent ba16daac2a
commit db3d2a2e31
35 changed files with 5560 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

399
tests/test_state.py
View File

@@ -1,21 +1,30 @@
import json
import subprocess
import tempfile
import time
import unittest
from unittest.mock import patch
from pathlib import Path
from unittest.mock import patch, MagicMock
import amc_server.mixins.state as state_mod
from amc_server.mixins.state import StateMixin
from amc_server.mixins.parsing import SessionParsingMixin
from amc_server.mixins.discovery import SessionDiscoveryMixin
class DummyStateHandler(StateMixin):
class DummyStateHandler(StateMixin, SessionParsingMixin, SessionDiscoveryMixin):
pass
class StateMixinTests(unittest.TestCase):
def test_get_active_zellij_sessions_uses_resolved_binary_and_parses_output(self):
handler = DummyStateHandler()
class TestGetActiveZellijSessions(unittest.TestCase):
"""Tests for _get_active_zellij_sessions edge cases."""
def setUp(self):
state_mod._zellij_cache["sessions"] = None
state_mod._zellij_cache["expires"] = 0
def test_parses_output_with_metadata(self):
handler = DummyStateHandler()
completed = subprocess.CompletedProcess(
args=[],
returncode=0,
@@ -23,15 +32,389 @@ class StateMixinTests(unittest.TestCase):
stderr="",
)
with patch.object(state_mod, "ZELLIJ_BIN", "/opt/homebrew/bin/zellij"), patch(
"amc_server.mixins.state.subprocess.run", return_value=completed
) as run_mock:
with patch.object(state_mod, "ZELLIJ_BIN", "/opt/homebrew/bin/zellij"), \
patch("amc_server.mixins.state.subprocess.run", return_value=completed) as run_mock:
sessions = handler._get_active_zellij_sessions()
self.assertEqual(sessions, {"infra", "work"})
args = run_mock.call_args.args[0]
self.assertEqual(args, ["/opt/homebrew/bin/zellij", "list-sessions", "--no-formatting"])
def test_empty_output_returns_empty_set(self):
handler = DummyStateHandler()
completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
with patch("amc_server.mixins.state.subprocess.run", return_value=completed):
sessions = handler._get_active_zellij_sessions()
self.assertEqual(sessions, set())
def test_whitespace_only_lines_ignored(self):
handler = DummyStateHandler()
completed = subprocess.CompletedProcess(
args=[], returncode=0, stdout="session1\n \n\nsession2\n", stderr=""
)
with patch("amc_server.mixins.state.subprocess.run", return_value=completed):
sessions = handler._get_active_zellij_sessions()
self.assertEqual(sessions, {"session1", "session2"})
def test_nonzero_exit_returns_none(self):
handler = DummyStateHandler()
completed = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="error")
with patch("amc_server.mixins.state.subprocess.run", return_value=completed):
sessions = handler._get_active_zellij_sessions()
self.assertIsNone(sessions)
def test_timeout_returns_none(self):
handler = DummyStateHandler()
with patch("amc_server.mixins.state.subprocess.run",
side_effect=subprocess.TimeoutExpired("cmd", 2)):
sessions = handler._get_active_zellij_sessions()
self.assertIsNone(sessions)
def test_file_not_found_returns_none(self):
handler = DummyStateHandler()
with patch("amc_server.mixins.state.subprocess.run", side_effect=FileNotFoundError()):
sessions = handler._get_active_zellij_sessions()
self.assertIsNone(sessions)
def test_cache_used_when_fresh(self):
handler = DummyStateHandler()
state_mod._zellij_cache["sessions"] = {"cached"}
state_mod._zellij_cache["expires"] = time.time() + 100
with patch("amc_server.mixins.state.subprocess.run") as mock_run:
sessions = handler._get_active_zellij_sessions()
mock_run.assert_not_called()
self.assertEqual(sessions, {"cached"})
class TestIsSessionDead(unittest.TestCase):
"""Tests for _is_session_dead edge cases."""
def setUp(self):
self.handler = DummyStateHandler()
def test_starting_session_not_dead(self):
session = {"status": "starting", "agent": "claude", "zellij_session": "s"}
self.assertFalse(self.handler._is_session_dead(session, {"s"}, set()))
def test_claude_without_zellij_session_is_dead(self):
session = {"status": "active", "agent": "claude", "zellij_session": ""}
self.assertTrue(self.handler._is_session_dead(session, set(), set()))
def test_claude_with_missing_zellij_session_is_dead(self):
session = {"status": "active", "agent": "claude", "zellij_session": "deleted"}
active_zellij = {"other_session"}
self.assertTrue(self.handler._is_session_dead(session, active_zellij, set()))
def test_claude_with_active_zellij_session_not_dead(self):
session = {"status": "active", "agent": "claude", "zellij_session": "existing"}
active_zellij = {"existing", "other"}
self.assertFalse(self.handler._is_session_dead(session, active_zellij, set()))
def test_claude_unknown_zellij_status_assumes_alive(self):
# When we can't query zellij (None), assume alive to avoid false positives
session = {"status": "active", "agent": "claude", "zellij_session": "unknown"}
self.assertFalse(self.handler._is_session_dead(session, None, set()))
def test_codex_without_transcript_path_is_dead(self):
session = {"status": "active", "agent": "codex", "transcript_path": ""}
self.assertTrue(self.handler._is_session_dead(session, None, set()))
def test_codex_with_active_transcript_not_dead(self):
session = {"status": "active", "agent": "codex", "transcript_path": "/path/to/file.jsonl"}
active_files = {"/path/to/file.jsonl"}
self.assertFalse(self.handler._is_session_dead(session, None, active_files))
def test_codex_without_active_transcript_checks_lsof(self):
session = {"status": "active", "agent": "codex", "transcript_path": "/path/to/file.jsonl"}
# Simulate lsof finding the file open
with patch.object(self.handler, "_is_file_open", return_value=True):
result = self.handler._is_session_dead(session, None, set())
self.assertFalse(result)
# Simulate lsof not finding the file
with patch.object(self.handler, "_is_file_open", return_value=False):
result = self.handler._is_session_dead(session, None, set())
self.assertTrue(result)
def test_unknown_agent_assumes_alive(self):
session = {"status": "active", "agent": "unknown_agent"}
self.assertFalse(self.handler._is_session_dead(session, None, set()))
class TestIsFileOpen(unittest.TestCase):
"""Tests for _is_file_open edge cases."""
def setUp(self):
self.handler = DummyStateHandler()
def test_lsof_finds_pid_returns_true(self):
completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="12345\n", stderr="")
with patch("amc_server.mixins.state.subprocess.run", return_value=completed):
result = self.handler._is_file_open("/some/file.jsonl")
self.assertTrue(result)
def test_lsof_no_result_returns_false(self):
completed = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="")
with patch("amc_server.mixins.state.subprocess.run", return_value=completed):
result = self.handler._is_file_open("/some/file.jsonl")
self.assertFalse(result)
def test_lsof_timeout_returns_false(self):
with patch("amc_server.mixins.state.subprocess.run",
side_effect=subprocess.TimeoutExpired("cmd", 2)):
result = self.handler._is_file_open("/some/file.jsonl")
self.assertFalse(result)
def test_lsof_not_found_returns_false(self):
with patch("amc_server.mixins.state.subprocess.run", side_effect=FileNotFoundError()):
result = self.handler._is_file_open("/some/file.jsonl")
self.assertFalse(result)
class TestCleanupStale(unittest.TestCase):
"""Tests for _cleanup_stale edge cases."""
def setUp(self):
self.handler = DummyStateHandler()
def test_removes_orphan_event_logs_older_than_24h(self):
with tempfile.TemporaryDirectory() as tmpdir:
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
# Create an orphan event log (no matching session)
orphan_log = events_dir / "orphan.jsonl"
orphan_log.write_text('{"event": "test"}\n')
# Set mtime to 25 hours ago
old_time = time.time() - (25 * 3600)
import os
os.utime(orphan_log, (old_time, old_time))
with patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
self.handler._cleanup_stale([]) # No active sessions
self.assertFalse(orphan_log.exists())
def test_keeps_orphan_event_logs_younger_than_24h(self):
with tempfile.TemporaryDirectory() as tmpdir:
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
# Create a recent orphan event log
recent_log = events_dir / "recent.jsonl"
recent_log.write_text('{"event": "test"}\n')
with patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
self.handler._cleanup_stale([])
self.assertTrue(recent_log.exists())
def test_keeps_event_logs_with_active_session(self):
with tempfile.TemporaryDirectory() as tmpdir:
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
# Create an old event log that HAS an active session
event_log = events_dir / "active-session.jsonl"
event_log.write_text('{"event": "test"}\n')
old_time = time.time() - (25 * 3600)
import os
os.utime(event_log, (old_time, old_time))
sessions = [{"session_id": "active-session"}]
with patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
self.handler._cleanup_stale(sessions)
self.assertTrue(event_log.exists())
def test_removes_stale_starting_sessions(self):
with tempfile.TemporaryDirectory() as tmpdir:
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
# Create a stale "starting" session
stale_session = sessions_dir / "stale.json"
stale_session.write_text(json.dumps({"status": "starting"}))
# Set mtime to 2 hours ago (> 1 hour threshold)
old_time = time.time() - (2 * 3600)
import os
os.utime(stale_session, (old_time, old_time))
with patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
self.handler._cleanup_stale([])
self.assertFalse(stale_session.exists())
def test_keeps_stale_active_sessions(self):
with tempfile.TemporaryDirectory() as tmpdir:
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
# Create an old "active" session (should NOT be deleted)
active_session = sessions_dir / "active.json"
active_session.write_text(json.dumps({"status": "active"}))
old_time = time.time() - (2 * 3600)
import os
os.utime(active_session, (old_time, old_time))
with patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
self.handler._cleanup_stale([])
self.assertTrue(active_session.exists())
class TestCollectSessions(unittest.TestCase):
"""Tests for _collect_sessions edge cases."""
def setUp(self):
self.handler = DummyStateHandler()
def test_invalid_json_session_file_skipped(self):
with tempfile.TemporaryDirectory() as tmpdir:
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
# Create an invalid JSON file
bad_file = sessions_dir / "bad.json"
bad_file.write_text("not json")
# Create a valid session
good_file = sessions_dir / "good.json"
good_file.write_text(json.dumps({
"session_id": "good",
"agent": "claude",
"status": "active",
"last_event_at": "2024-01-01T00:00:00Z",
}))
with patch.object(state_mod, "SESSIONS_DIR", sessions_dir), \
patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(self.handler, "_discover_active_codex_sessions"), \
patch.object(self.handler, "_get_active_zellij_sessions", return_value=None), \
patch.object(self.handler, "_get_active_transcript_files", return_value=set()), \
patch.object(self.handler, "_get_context_usage_for_session", return_value=None), \
patch.object(self.handler, "_get_conversation_mtime", return_value=None):
sessions = self.handler._collect_sessions()
# Should only get the good session
self.assertEqual(len(sessions), 1)
self.assertEqual(sessions[0]["session_id"], "good")
def test_non_dict_session_file_skipped(self):
with tempfile.TemporaryDirectory() as tmpdir:
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
# Create a file with array instead of dict
array_file = sessions_dir / "array.json"
array_file.write_text("[1, 2, 3]")
with patch.object(state_mod, "SESSIONS_DIR", sessions_dir), \
patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(self.handler, "_discover_active_codex_sessions"), \
patch.object(self.handler, "_get_active_zellij_sessions", return_value=None), \
patch.object(self.handler, "_get_active_transcript_files", return_value=set()), \
patch.object(self.handler, "_get_context_usage_for_session", return_value=None), \
patch.object(self.handler, "_get_conversation_mtime", return_value=None):
sessions = self.handler._collect_sessions()
self.assertEqual(len(sessions), 0)
def test_orphan_starting_session_deleted(self):
with tempfile.TemporaryDirectory() as tmpdir:
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
# Create a starting session with a non-existent zellij session
orphan_file = sessions_dir / "orphan.json"
orphan_file.write_text(json.dumps({
"session_id": "orphan",
"status": "starting",
"zellij_session": "deleted_session",
}))
active_zellij = {"other_session"} # orphan's session not in here
with patch.object(state_mod, "SESSIONS_DIR", sessions_dir), \
patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(self.handler, "_discover_active_codex_sessions"), \
patch.object(self.handler, "_get_active_zellij_sessions", return_value=active_zellij), \
patch.object(self.handler, "_get_active_transcript_files", return_value=set()), \
patch.object(self.handler, "_get_context_usage_for_session", return_value=None), \
patch.object(self.handler, "_get_conversation_mtime", return_value=None):
sessions = self.handler._collect_sessions()
# Orphan should be deleted and not in results
self.assertEqual(len(sessions), 0)
self.assertFalse(orphan_file.exists())
def test_sessions_sorted_by_id(self):
with tempfile.TemporaryDirectory() as tmpdir:
sessions_dir = Path(tmpdir) / "sessions"
sessions_dir.mkdir()
events_dir = Path(tmpdir) / "events"
events_dir.mkdir()
for sid in ["zebra", "alpha", "middle"]:
(sessions_dir / f"{sid}.json").write_text(json.dumps({
"session_id": sid,
"status": "active",
"last_event_at": "2024-01-01T00:00:00Z",
}))
with patch.object(state_mod, "SESSIONS_DIR", sessions_dir), \
patch.object(state_mod, "EVENTS_DIR", events_dir), \
patch.object(self.handler, "_discover_active_codex_sessions"), \
patch.object(self.handler, "_get_active_zellij_sessions", return_value=None), \
patch.object(self.handler, "_get_active_transcript_files", return_value=set()), \
patch.object(self.handler, "_get_context_usage_for_session", return_value=None), \
patch.object(self.handler, "_get_conversation_mtime", return_value=None):
sessions = self.handler._collect_sessions()
ids = [s["session_id"] for s in sessions]
self.assertEqual(ids, ["alpha", "middle", "zebra"])
if __name__ == "__main__":
unittest.main()