operator
b7e45010c3
- Add clean line-by-line transcript style toggle ('b' key / /clean / /boxed)
- Implement right-click context menus for chat list, fleet agent panel, and transcript
- Add cooldown mode lock bypass (2-second double-confirm force sync)
- Implement dual copy support: clean text (strip reply metadata) vs full context
- Add clickable [📋 Copy] and [📑+ Context] buttons to message headers
- Add unit test suites for transcript cleaning, context menus, rate limits, and copy actions
123 lines
5.3 KiB
Python
123 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_agent_ranking.py — Unit test suite for Fleet Agent interaction ranking & sorting:
|
|
1. Agent interaction recording on true input / chat [insert].
|
|
2. Fleet agent sorting/ranking by last interaction descending.
|
|
3. Stable fallback ordering for un-interacted agents.
|
|
4. Persistence and seeding of interaction timestamps.
|
|
5. format_recency helper accuracy.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import time
|
|
import tempfile
|
|
import json
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
|
|
|
spec = importlib.util.spec_from_file_location("muse_tui", str(TUI_PATH))
|
|
muse_mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(muse_mod)
|
|
|
|
|
|
class TestAgentRanking(unittest.TestCase):
|
|
"""Test suite covering fleet agent interaction sorting & recency tracking."""
|
|
|
|
def setUp(self):
|
|
self.mock_stdscr = MagicMock()
|
|
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
|
self.tui = muse_mod.MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
|
self.tui.safe_addstr = MagicMock()
|
|
self.tui.data.history_cache = {}
|
|
self.tui.data.read_msg_counts = {}
|
|
|
|
def test_format_recency(self):
|
|
"""Verify format_recency produces compact, accurate recency strings."""
|
|
now = time.time()
|
|
self.assertEqual(muse_mod.format_recency(0.0), "never")
|
|
self.assertEqual(muse_mod.format_recency(None), "never")
|
|
self.assertEqual(muse_mod.format_recency(now - 10), "10s ago")
|
|
self.assertEqual(muse_mod.format_recency(now - 120), "2m ago")
|
|
self.assertEqual(muse_mod.format_recency(now - 7200), "2h ago")
|
|
self.assertEqual(muse_mod.format_recency(now - 90000), "1d ago")
|
|
|
|
def test_sort_nodes_by_interaction(self):
|
|
"""Verify sort_nodes ranks nodes by last interaction timestamp descending."""
|
|
data = self.tui.data
|
|
now = time.time()
|
|
|
|
# Set specific timestamps
|
|
data.agent_interactions["muse"] = now - 500
|
|
data.agent_interactions["pip"] = now - 100
|
|
data.agent_interactions["646"] = now - 10 # Most recent
|
|
data.agent_interactions["opm"] = 0.0 # Never
|
|
data.agent_interactions["def"] = 0.0 # Never
|
|
|
|
data.sort_nodes()
|
|
|
|
# Most recent first: 646, then pip, then muse
|
|
self.assertEqual(data.nodes[0], "646")
|
|
self.assertEqual(data.nodes[1], "pip")
|
|
self.assertEqual(data.nodes[2], "muse")
|
|
# Un-interacted agents should be at the end, retaining baseline order
|
|
self.assertIn("opm", data.nodes[3:])
|
|
self.assertIn("def", data.nodes[3:])
|
|
|
|
def test_record_interaction_promotes_node_to_top(self):
|
|
"""Verify record_interaction immediately updates timestamp and promotes node to rank 1."""
|
|
data = self.tui.data
|
|
# Ensure 'pip' is not at index 0 initially
|
|
data.record_interaction("646")
|
|
self.assertEqual(data.nodes[0], "646")
|
|
|
|
# Now operator directly interacts with 'pip'
|
|
t0 = time.time()
|
|
data.record_interaction("pip")
|
|
self.assertEqual(data.nodes[0], "pip", "'pip' must be promoted to rank 1 (index 0)")
|
|
self.assertEqual(data.nodes[1], "646", "'646' must be rank 2")
|
|
self.assertGreaterEqual(data.agent_interactions["pip"], t0)
|
|
|
|
def test_chat_message_submission_triggers_ranking(self):
|
|
"""Verify submitting chat message in INSERT mode promotes the active node."""
|
|
self.tui.editor_mode = "INSERT"
|
|
self.tui.data.active_node = "def"
|
|
self.tui.data.active_thread_id = "test-session-123"
|
|
self.tui.data.active_thread_title = "Def Task Session"
|
|
|
|
# Mock async send to avoid subprocess
|
|
with patch.object(self.tui, "_async_send_message"):
|
|
self.tui._execute_input_line("Deploy updated service configuration")
|
|
|
|
# 'def' was chatted with directly; it must now be rank 1 at index 0
|
|
self.assertEqual(self.tui.data.nodes[0], "def", "Chat message send must promote target node to top of list")
|
|
|
|
def test_persistence_of_interactions(self):
|
|
"""Verify interaction records persist and load from json correctly."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
fake_int_file = Path(tmpdir) / "agent_interactions.json"
|
|
now = time.time()
|
|
fake_int_file.write_text(json.dumps({"opm": now - 50, "dev": now - 10}))
|
|
|
|
with patch("pathlib.Path.home", return_value=Path(tmpdir)), \
|
|
patch.object(muse_mod.Path, "home", return_value=Path(tmpdir)):
|
|
# Adjust path to match ~/.config/muse-cli/
|
|
cfg_dir = Path(tmpdir) / ".config" / "muse-cli"
|
|
cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
(cfg_dir / "agent_interactions.json").write_text(json.dumps({"opm": now - 50, "dev": now - 10}))
|
|
|
|
mgr = muse_mod.FleetDataManager()
|
|
mgr.stop() # Stop background poller
|
|
|
|
self.assertAlmostEqual(mgr.agent_interactions.get("dev", 0.0), now - 10, delta=1.0)
|
|
self.assertAlmostEqual(mgr.agent_interactions.get("opm", 0.0), now - 50, delta=1.0)
|
|
self.assertEqual(mgr.nodes[0], "dev", "Persisted recent node 'dev' must be rank 1")
|
|
self.assertEqual(mgr.nodes[1], "opm", "'opm' must be rank 2")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|