#!/usr/bin/env python3 """ Comprehensive test to verify AI agent prompt synchronization. """ import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'src')) from config.prompt_management import PromptController def test_agent_synchronization(): """Test that all AI agents are properly synchronized with shared components.""" print("Testing AI agent prompt synchronization...") controller = PromptController() # Test agents agents = ['spiritual_monitor', 'triage_question', 'triage_evaluator'] print(f"\n1. Testing {len(agents)} agents for synchronization...") # Get configurations for all agents configs = {} for agent in agents: try: configs[agent] = controller.get_prompt(agent) print(f" ✓ {agent}: loaded successfully") except Exception as e: print(f" ✗ {agent}: failed to load - {e}") return False # Test 2: Verify shared indicators consistency print("\n2. Testing shared indicators consistency...") indicator_sets = {} for agent, config in configs.items(): indicator_sets[agent] = {ind.name: ind.to_dict() for ind in config.shared_indicators} # Compare all agents base_agent = agents[0] base_indicators = indicator_sets[base_agent] for other_agent in agents[1:]: other_indicators = indicator_sets[other_agent] # Check indicator names if set(base_indicators.keys()) != set(other_indicators.keys()): print(f" ✗ Indicator names differ between {base_agent} and {other_agent}") return False # Check indicator definitions for ind_name in base_indicators: if base_indicators[ind_name] != other_indicators[ind_name]: print(f" ✗ Indicator '{ind_name}' differs between {base_agent} and {other_agent}") return False print(f" ✓ All agents have identical {len(base_indicators)} indicators") # Test 3: Verify shared rules consistency print("\n3. Testing shared rules consistency...") rule_sets = {} for agent, config in configs.items(): rule_sets[agent] = {rule.rule_id: rule.to_dict() for rule in config.shared_rules} base_rules = rule_sets[base_agent] for other_agent in agents[1:]: other_rules = rule_sets[other_agent] # Check rule IDs if set(base_rules.keys()) != set(other_rules.keys()): print(f" ✗ Rule IDs differ between {base_agent} and {other_agent}") return False # Check rule definitions for rule_id in base_rules: if base_rules[rule_id] != other_rules[rule_id]: print(f" ✗ Rule '{rule_id}' differs between {base_agent} and {other_agent}") return False print(f" ✓ All agents have identical {len(base_rules)} rules") # Test 4: Verify rule priority consistency print("\n4. Testing rule priority consistency...") for agent, config in configs.items(): sorted_rules = sorted(config.shared_rules, key=lambda r: r.priority) rule_order = [rule.rule_id for rule in sorted_rules] if agent == base_agent: base_rule_order = rule_order else: if rule_order != base_rule_order: print(f" ✗ Rule priority order differs for {agent}") return False print(" ✓ All agents have identical rule priority order") # Test 5: Verify template consistency print("\n5. Testing template consistency...") template_sets = {} for agent, config in configs.items(): template_sets[agent] = {tmpl.template_id: tmpl.to_dict() for tmpl in config.templates} base_templates = template_sets[base_agent] for other_agent in agents[1:]: other_templates = template_sets[other_agent] # Check template IDs if set(base_templates.keys()) != set(other_templates.keys()): print(f" ✗ Template IDs differ between {base_agent} and {other_agent}") return False # Check template definitions for tmpl_id in base_templates: if base_templates[tmpl_id] != other_templates[tmpl_id]: print(f" ✗ Template '{tmpl_id}' differs between {base_agent} and {other_agent}") return False print(f" ✓ All agents have identical {len(base_templates)} templates") # Test 6: Verify category definitions consistency print("\n6. Testing category definitions consistency...") categories = controller.category_definitions.get_all_categories() required_categories = ['GREEN', 'YELLOW', 'RED'] for category in required_categories: if category not in categories: print(f" ✗ Missing required category: {category}") return False print(f" ✓ All required categories present: {required_categories}") # Test 7: Verify validation consistency print("\n7. Testing validation consistency...") validation_result = controller.validate_consistency() if validation_result.errors: print(" ⚠ Validation errors found:") for error in validation_result.errors: print(f" - {error}") else: print(" ✓ No validation errors") if validation_result.warnings: print(" ⚠ Validation warnings:") for warning in validation_result.warnings: print(f" - {warning}") else: print(" ✓ No validation warnings") # Test 8: Test session isolation print("\n8. Testing session isolation...") session_id = "sync_test_session" test_override = "Test session override for synchronization testing" # Set override for one agent success = controller.set_session_override('spiritual_monitor', test_override, session_id) if not success: print(" ✗ Failed to set session override") return False # Verify other agents unaffected for agent in agents: config = controller.get_prompt(agent, session_id=session_id) if agent == 'spiritual_monitor': if config.session_override != test_override: print(f" ✗ Session override not applied to {agent}") return False else: if config.session_override is not None: print(f" ✗ Session override incorrectly applied to {agent}") return False # Clean up controller.clear_session_overrides(session_id) print(" ✓ Session isolation working correctly") # Test 9: Performance metrics print("\n9. Testing performance metrics...") # Log some test metrics for agent in agents: controller.log_performance_metric(agent, 0.5, 0.85, False) # Verify metrics are recorded for agent in agents: metrics = controller.get_performance_metrics(agent) if metrics['total_executions'] == 0: print(f" ✗ No metrics recorded for {agent}") return False print(" ✓ Performance metrics working correctly") # Summary print("\n" + "="*60) print("SYNCHRONIZATION TEST SUMMARY") print("="*60) print(f"✓ Agents tested: {len(agents)}") print(f"✓ Shared indicators: {len(base_indicators)}") print(f"✓ Shared rules: {len(base_rules)}") print(f"✓ Shared templates: {len(base_templates)}") print(f"✓ Category definitions: {len(categories)}") print("✓ All agents are properly synchronized!") return True if __name__ == "__main__": success = test_agent_synchronization() sys.exit(0 if success else 1)