"""Test multi-LLM conversation mesh, skill building pools, and auto category adder.""" import sys import os import tempfile sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from splitbit_llm.agents.conversation_mesh import ( ConversationMesh, AutoCategoryManager, SkillPool, ConversationMessage, ) def test_auto_category_manager(): """Test auto skill category adder — discovers new categories dynamically.""" mgr = AutoCategoryManager() # Should start with seed categories cats = mgr.get_categories() assert "conversation" in cats assert "code" in cats assert "speed" in cats print(f" Seed categories: {len(cats)} — {cats[:5]}...") # Discover categories from a topic discovered = mgr.discover_from_topic("How to improve database query optimization") assert len(discovered) > 0 print(f" Discovered from 'database query optimization': {discovered}") # Should have auto-added new categories auto_cats = mgr.get_auto_categories() assert len(auto_cats) > 0, f"Expected auto categories, got: {auto_cats}" print(f" Auto-added categories: {auto_cats}") # Discover from another topic discovered2 = mgr.discover_from_topic("Building better network security protocols") print(f" Discovered from 'network security protocols': {discovered2}") # Stats stats = mgr.get_stats() assert stats["categories_total"] > len(mgr.SEED_CATEGORIES) assert stats["categories_auto_added"] > 0 print(f" Stats: {stats['categories_total']} total, {stats['categories_auto_added']} auto-added") def test_category_matching(): """Test that similar keywords map to existing categories.""" mgr = AutoCategoryManager() # First discovery creates the category mgr.discover_from_topic("optimization techniques") assert "optimization" in mgr.get_categories() # Similar keyword should map to existing category mgr.discover_from_topic("optimize performance") # "optimize" should map to "optimization" via prefix matching cats = mgr.get_categories() print(f" Categories after 'optimize': {cats}") stats = mgr.get_stats() print(f" Keywords indexed: {stats['keywords_indexed']}") def test_conversation_mesh(): """Test multi-LLM conversation mesh — agents converse to build skills.""" mesh = ConversationMesh(harness=None) # Run a round-robin conversation result = mesh.run_conversation(mode="round_robin", topic="How to improve code quality") assert result["mode"] == "round_robin" assert result["topic"] == "How to improve code quality" assert result["messages"] > 0 assert len(result["categories"]) > 0 print(f" Round-robin: {result['messages']} messages, categories: {result['categories']}") # Run a brainstorm result2 = mesh.run_conversation(mode="brainstorm", topic="Efficient algorithms for pattern matching") assert result2["mode"] == "brainstorm" assert result2["messages"] > 0 print(f" Brainstorm: {result2['messages']} messages, categories: {result2['categories']}") # Run a debate result3 = mesh.run_conversation(mode="debate", topic="Best approaches to data compression") assert result3["mode"] == "debate" print(f" Debate: {result3['messages']} messages") # Run a teaching session result4 = mesh.run_conversation(mode="teaching", topic="Methods for adaptive learning") assert result4["mode"] == "teaching" print(f" Teaching: {result4['messages']} messages") # Run a pairwise discussion result5 = mesh.run_conversation(mode="pairwise", topic="Strategies for error handling") assert result5["mode"] == "pairwise" print(f" Pairwise: {result5['messages']} messages") def test_skill_building_pools(): """Test skill building pools — collaborative skills from multiple agents.""" mesh = ConversationMesh(harness=None) # Run multiple conversations to build pools for i in range(5): mesh.run_conversation(topic=f"Optimizing memory usage in system {i}") pools = mesh.get_skill_pools() assert len(pools) > 0, f"Expected skill pools, got {len(pools)}" print(f" Skill pools created: {len(pools)}") # Check pool structure pool = pools[0] assert "id" in pool assert "name" in pool assert "category" in pool assert "contributors" in pool assert "confidence" in pool print(f" Pool: {pool['name']} (category: {pool['category']}, contributors: {pool['contributors']})") # Stats stats = mesh.get_stats() assert stats["skills_pooled"] > 0 assert stats["cross_agent_skills"] > 0, "Expected cross-agent skills" print(f" Stats: {stats['skills_pooled']} pooled, {stats['cross_agent_skills']} cross-agent") def test_mesh_auto_categories(): """Test that mesh conversations auto-discover new categories.""" mesh = ConversationMesh(harness=None) # Run conversations on diverse topics topics = [ "Improving database query performance", "Building neural network architectures", "Optimizing cache invalidation strategies", "Enhancing cryptographic security measures", "Implementing blockchain consensus algorithms", ] all_cats = set() for topic in topics: result = mesh.run_conversation(topic=topic) all_cats.update(result["categories"]) cats = mesh.get_categories() auto_cats = mesh.get_auto_categories() assert len(auto_cats) > 0, f"Expected auto categories: {auto_cats}" print(f" Total categories: {len(cats)}") print(f" Auto-discovered: {auto_cats}") # Verify some expected categories were discovered # (at least some of: database, neural, cache, cryptographic, blockchain) discovered_lower = [c.lower() for c in auto_cats] print(f" Discovered categories: {discovered_lower}") def test_mesh_stats(): """Test mesh stats tracking.""" mesh = ConversationMesh(harness=None) # Run a few conversations mesh.run_conversation(topic="Testing mesh statistics tracking") mesh.run_conversation(topic="Another topic for skill building") stats = mesh.get_stats() assert stats["conversations_total"] >= 2 assert stats["messages_exchanged"] > 0 cat_stats = stats["categories"] assert cat_stats["categories_total"] > 0 print(f" Conversations: {stats['conversations_total']}") print(f" Messages: {stats['messages_exchanged']}") print(f" Skills pooled: {stats['skills_pooled']}") print(f" Categories: {cat_stats['categories_total']}") def test_random_mode_selection(): """Test that random mode selection works.""" mesh = ConversationMesh(harness=None) modes_used = set() for _ in range(10): result = mesh.run_conversation() # no mode specified → random modes_used.add(result["mode"]) # Should have used at least 2 different modes in 10 random runs assert len(modes_used) >= 2, f"Expected variety of modes, got: {modes_used}" print(f" Modes used: {modes_used}") def test_skill_cascade(): """Test that building a skill pool cascades into building related skill pools. When a skill pool is created on topic X, the system should automatically generate related skill pools on similar topics (advanced techniques, best practices, pitfalls, testing, integration, etc.) """ mesh = ConversationMesh(harness=None) # Run a single conversation — should cascade into multiple related ones result = mesh.run_conversation(topic="Optimizing database query performance") # The original conversation should have cascaded cascaded = result.get("cascaded", []) assert len(cascaded) > 0, f"Expected cascade results, got: {cascaded}" print(f" Original topic: 'Optimizing database query performance'") print(f" Cascaded into {len(cascaded)} related conversations:") for c in cascaded: print(f" → {c['topic'][:60]} (mode: {c['mode']}, pool: {c['pool_id'] is not None})") # Should have created multiple skill pools (original + cascaded) pools = mesh.get_skill_pools() assert len(pools) > 1, f"Expected multiple pools from cascade, got {len(pools)}" print(f" Total skill pools: {len(pools)}") # Stats should show cascade activity stats = mesh.get_stats() assert stats["cascade_pools_created"] > 0, "Expected cascade pools created" assert stats["cascade_conversations"] > 0, "Expected cascade conversations" print(f" Cascade stats: {stats['cascade_pools_created']} pools, " f"{stats['cascade_conversations']} conversations, depth {stats['cascade_depth']}") def test_cascade_related_topics(): """Test that related topics are generated correctly from categories.""" mesh = ConversationMesh(harness=None) # Generate related topics from a topic and categories related = mesh._generate_related_topics( "Optimizing database performance", ["database", "performance", "optimization"] ) assert len(related) > 0, "Expected related topics" assert len(related) <= 5, f"Should limit to 5, got {len(related)}" print(f" Related topics for 'database, performance, optimization':") for t in related: print(f" → {t}") # Should include category-based variations (first category always included) assert any("database" in t.lower() for t in related) def test_cascade_depth_limit(): """Test that cascade depth is limited to prevent infinite recursion.""" mesh = ConversationMesh(harness=None) # Run a conversation — cascade should be limited to depth 3 result = mesh.run_conversation(topic="Building neural network architectures") stats = mesh.get_stats() assert stats["cascade_depth"] <= 2, f"Cascade depth should be <= 2, got {stats['cascade_depth']}" print(f" Cascade depth: {stats['cascade_depth']} (max 2)") print(f" Total pools: {stats['skill_pools']}") print(f" Total conversations: {stats['conversations_total']}") if __name__ == "__main__": print("Running conversation mesh, skill pools, and auto category tests...") test_auto_category_manager() print(" ✓ test_auto_category_manager") test_category_matching() print(" ✓ test_category_matching") test_conversation_mesh() print(" ✓ test_conversation_mesh") test_skill_building_pools() print(" ✓ test_skill_building_pools") test_mesh_auto_categories() print(" ✓ test_mesh_auto_categories") test_mesh_stats() print(" ✓ test_mesh_stats") test_random_mode_selection() print(" ✓ test_random_mode_selection") test_skill_cascade() print(" ✓ test_skill_cascade") test_cascade_related_topics() print(" ✓ test_cascade_related_topics") test_cascade_depth_limit() print(" ✓ test_cascade_depth_limit") print("\nAll conversation mesh tests passed!")