""" AIVA Voice Command Bridge - Status Report Formatter Converts raw Claude Code agent data into natural language for AIVA to read aloud. """ import logging import re from dataclasses import dataclass from enum import Enum from typing import Any, Optional, Union # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class StatusType(Enum): """Enum for different status types.""" PROGRESS = "progress" ERROR = "error" COMPLETION = "completion" QUEUE = "queue" SYSTEM_HEALTH = "system_health" UNKNOWN = "unknown" @dataclass class FormatResult: """Result of formatting operation.""" success: bool formatted_text: str status_type: StatusType error_message: Optional[str] = None class StatusFormatter: """ Formats raw Claude Code agent data into natural language for AIVA to speak. Uses template-based responses with variable interpolation. """ def __init__(self): self.logger = logging.getLogger(__name__) self._init_templates() def _init_templates(self): """Initialize all response templates.""" # Progress update templates self.templates = { StatusType.PROGRESS: [ "Claude is currently {task} and is about {progress}% done. Should be finished in around {eta_minutes} minutes.", "Right now Claude is {task}. Progress is at {progress} percent, with roughly {eta_minutes} minutes remaining.", "Claude's working on {task} — about {progress}% complete. ETA is around {eta_minutes} minutes.", "Currently processing: {task}. That's {progress}% done with about {eta_minutes} minutes to go.", ], # Error report templates StatusType.ERROR: [ "Heads up — {error_message}. It will automatically retry in {retry_in} seconds.", "There's an issue: {error_message}. Retrying in {retry_in} seconds.", "Attention: {error_message}. The system will try again in about {retry_in} seconds.", "Warning — {error_message}. Automatic retry scheduled in {retry_in} seconds.", ], # Completion summary templates StatusType.COMPLETION: [ "Great news! {completed_today} tasks completed today. That's {completed_today} jobs done, with {failed} failure{plural_failed}.", "Claude finished {completed_today} tasks today. Unfortunately there {failure_verb} {failed} failure{failed_plural}.", "Today's tally: {completed_today} tasks completed. {failed} didn't make it, so {success_rate}% success rate.", "All up, Claude completed {completed_today} tasks today. {failed} encountered issues.", ], # Queue status templates StatusType.QUEUE: [ "There are {pending_directives} directives waiting in the queue.", "Right now there {is_are} {pending_directives} item{plural_pending} waiting for attention.", "The queue has {pending_directives} pending directive{plural_pending} at the moment.", "You've got {pending_directives} directive{plural_pending} in the queue waiting to be processed.", ], # System health templates StatusType.SYSTEM_HEALTH: [ "System's looking good. CPU at {cpu}%, memory at {memory}%, and {active_agents} agent{plural_agents} currently active.", "Health check: CPU usage is {cpu}%, memory at {memory}%. There {is_are} {active_agents} active agent{plural_agents}.", "All systems nominal. CPU {cpu}%, memory {memory}%, with {active_agents} agent{plural_agents} running.", "System status: {cpu}% CPU, {memory}% memory. {active_agents} agent{plural_agents} currently active.", ], # Generic status templates StatusType.UNKNOWN: [ "Here's the current status: {raw_summary}", "Claude's status update: {raw_summary}", "Right, so {raw_summary}", "Current situation: {raw_summary}", ], } def _detect_status_type(self, data: dict) -> StatusType: """Detect the type of status from the input data.""" # Check for progress indicators if "progress" in data and "task" in data: return StatusType.PROGRESS # Check for error indicators if "error" in data or "error_message" in data: return StatusType.ERROR # Check for completion/activity summary if "completed_today" in data: return StatusType.COMPLETION # Check for queue status if "pending_directives" in data: return StatusType.QUEUE # Check for system health if "cpu" in data or "memory" in data or "active_agents" in data: return StatusType.SYSTEM_HEALTH return StatusType.UNKNOWN def _format_number(self, num: Union[int, float], ordinal: bool = False) -> str: """Format a number for speech.""" if isinstance(num, float): # Round to integer for natural speech num = round(num) if ordinal and num > 0: # Return ordinal form if num == 1: return "one" elif num == 2: return "two" elif num == 3: return "three" elif num == 4: return "four" elif num == 5: return "five" elif num == 6: return "six" elif num == 7: return "seven" elif num == 8: return "eight" elif num == 9: return "nine" elif num == 10: return "ten" else: return str(num) return str(num) def _prepare_variables(self, data: dict, status_type: StatusType) -> dict: """Prepare variables for template substitution.""" vars = {} if status_type == StatusType.PROGRESS: vars["task"] = data.get("task", "working") vars["progress"] = data.get("progress", 0) vars["eta_minutes"] = data.get("eta_minutes", "some time") # Format eta for speech if isinstance(vars["eta_minutes"], (int, float)): if vars["eta_minutes"] == 1: vars["eta_minutes"] = "about 1 minute" elif vars["eta_minutes"] < 1: vars["eta_minutes"] = "less than a minute" else: vars["eta_minutes"] = f"around {int(vars['eta_minutes'])} minutes" elif status_type == StatusType.ERROR: error = data.get("error") or data.get("error_message", "an unknown error") # Convert error codes to human-readable messages vars["error_message"] = self._humanize_error(error) vars["retry_in"] = data.get("retry_in", 30) elif status_type == StatusType.COMPLETION: vars["completed_today"] = data.get("completed_today", 0) vars["failed"] = data.get("failed", 0) vars["plural_failed"] = "" if vars["failed"] == 1 else "s" vars["failure_verb"] = "was" if vars["failed"] == 1 else "were" vars["failed_plural"] = "" if vars["failed"] == 1 else "s" # Calculate success rate total = vars["completed_today"] + vars["failed"] if total > 0: rate = (vars["completed_today"] / total) * 100 vars["success_rate"] = str(round(rate)) else: vars["success_rate"] = "100" elif status_type == StatusType.QUEUE: vars["pending_directives"] = data.get("pending_directives", 0) vars["plural_pending"] = "" if vars["pending_directives"] == 1 else "s" vars["is_are"] = "is" if vars["pending_directives"] == 1 else "are" elif status_type == StatusType.SYSTEM_HEALTH: vars["cpu"] = data.get("cpu", 0) vars["memory"] = data.get("memory", 0) vars["active_agents"] = data.get("active_agents", 0) vars["plural_agents"] = "" if vars["active_agents"] == 1 else "s" vars["is_are"] = "is" if vars["active_agents"] == 1 else "are" else: # Unknown type - create raw summary vars["raw_summary"] = self._create_raw_summary(data) return vars def _humanize_error(self, error: str) -> str: """Convert error codes/messages to human-readable Australian-friendly text.""" error_lower = error.lower() # Common error mappings error_map = { "database_connection_failed": "Claude hit a database connection issue", "timeout": "the operation timed out", "auth_failed": "authentication failed", "rate_limit": "we've hit a rate limit", "not_found": "the requested resource wasn't found", "permission_denied": "permission was denied", "network_error": "there's a network problem", "out_of_memory": "the system ran out of memory", "disk_full": "the disk is full", "invalid_input": "the input was invalid", "service_unavailable": "the service is temporarily unavailable", } for key, value in error_map.items(): if key in error_lower: return value # If no mapping found, clean up the error string return error.replace("_", " ").replace("-", " ") def _create_raw_summary(self, data: dict) -> str: """Create a raw summary from unknown data format.""" parts = [] for key, value in data.items(): # Skip internal fields if key.startswith("_"): continue # Format the key nicely nice_key = key.replace("_", " ") if isinstance(value, bool): parts.append(f"{nice_key} is {'active' if value else 'inactive'}") elif isinstance(value, (int, float)): parts.append(f"{nice_key}: {value}") elif isinstance(value, str): parts.append(f"{nice_key}: {value}") elif isinstance(value, list) and len(value) > 0: parts.append(f"{nice_key} has {len(value)} items") elif value is None: parts.append(f"{nice_key} not set") if not parts: return "no status information available" return ", ".join(parts) def format_status(self, raw_data: Union[dict, str]) -> FormatResult: """ Format raw status data into natural language for AIVA to speak. Args: raw_data: JSON data from Claude Code (dict or JSON string) Returns: FormatResult with formatted text and metadata """ try: # Parse JSON string if needed if isinstance(raw_data, str): import json data = json.loads(raw_data) else: data = raw_data if not isinstance(data, dict): return FormatResult( success=False, formatted_text="Invalid data format. Expected a dictionary.", status_type=StatusType.UNKNOWN, error_message="Input data is not a dictionary" ) # Detect status type status_type = self._detect_status_type(data) self.logger.debug(f"Detected status type: {status_type}") # Get template templates = self.templates.get(status_type, self.templates[StatusType.UNKNOWN]) # Prepare variables vars = self._prepare_variables(data, status_type) # Select template (use hash for consistent selection) template_idx = hash(frozenset(data.items())) % len(templates) template = templates[template_idx] # Format numbers for speech for key, value in vars.items(): if isinstance(value, (int, float)) and key not in ["eta_minutes"]: vars[key] = self._format_number(value) # Try to format, fallback if missing variables try: formatted = template.format(**vars) except KeyError as e: self.logger.warning(f"Missing variable in template: {e}") # Use raw summary as fallback vars["raw_summary"] = self._create_raw_summary(data) formatted = self.templates[StatusType.UNKNOWN][0].format(**vars) # Clean up any double spaces or artifacts formatted = re.sub(r'\s+', ' ', formatted).strip() return FormatResult( success=True, formatted_text=formatted, status_type=status_type ) except json.JSONDecodeError as e: self.logger.error(f"JSON decode error: {e}") return FormatResult( success=False, formatted_text="Sorry, I couldn't understand that status update.", status_type=StatusType.UNKNOWN, error_message=f"JSON parsing failed: {e}" ) except Exception as e: self.logger.error(f"Formatting error: {e}") return FormatResult( success=False, formatted_text="There was an error processing the status update.", status_type=StatusType.UNKNOWN, error_message=str(e) ) def format_multiple(self, status_list: list) -> list[FormatResult]: """Format multiple status updates.""" results = [] for status in status_list: results.append(self.format_status(status)) return results def format_summary(self, statuses: list) -> str: """ Format multiple statuses into a single summary. Args: statuses: List of status dictionaries Returns: Combined natural language summary """ if not statuses: return "No status information available." results = self.format_multiple(statuses) # Filter successful formats successful = [r.formatted_text for r in results if r.success] if not successful: return "Unable to retrieve status information." # Combine with appropriate connector if len(successful) == 1: return successful[0] elif len(successful) == 2: return f"{successful[0]}. Also, {successful[1].lower()}" else: # Join all but last with commas, last with "and" return ", ".join(successful[:-1]) + f", and {successful[-1].lower()}" def create_formatter() -> StatusFormatter: """Factory function to create a StatusFormatter instance.""" return StatusFormatter() # ============================================================================= # Unit Tests # ============================================================================= import unittest import json class TestStatusFormatter(unittest.TestCase): """Comprehensive unit tests for StatusFormatter.""" def setUp(self): """Set up test formatter.""" self.formatter = StatusFormatter() # === Progress Status Tests === def test_progress_with_eta(self): """Test progress status with ETA in minutes.""" data = { 'status': 'running', 'task': 'building_api', 'progress': 65, 'eta_minutes': 12 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.PROGRESS) self.assertIn("building the API", result.formatted_text) self.assertIn("65", result.formatted_text) self.assertIn("12", result.formatted_text) def test_progress_zero_eta(self): """Test progress status with zero ETA.""" data = { 'task': 'finalizing', 'progress': 95, 'eta_minutes': 0 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("less than a minute", result.formatted_text.lower()) def test_progress_one_minute_eta(self): """Test progress status with one minute ETA.""" data = { 'task': 'processing', 'progress': 50, 'eta_minutes': 1 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("1 minute", result.formatted_text) def test_progress_default_values(self): """Test progress status with default values.""" data = { 'task': 'working', 'progress': 25 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.PROGRESS) def test_progress_float_values(self): """Test progress status with float values.""" data = { 'task': 'analyzing', 'progress': 33.3, 'eta_minutes': 7.5 } result = self.formatter.format_status(data) self.assertTrue(result.success) # Should round to integers for speech self.assertIn("33", result.formatted_text) # === Error Status Tests === def test_error_with_retry(self): """Test error status with retry information.""" data = { 'error': 'database_connection_failed', 'retry_in': 30 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.ERROR) self.assertIn("database connection", result.formatted_text.lower()) self.assertIn("30", result.formatted_text) def test_error_different_types(self): """Test various error types.""" errors = [ {'error': 'timeout', 'retry_in': 60}, {'error': 'rate_limit', 'retry_in': 120}, {'error': 'network_error', 'retry_in': 15}, ] for error_data in errors: result = self.formatter.format_status(error_data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.ERROR) def test_error_message_field(self): """Test error using error_message field.""" data = { 'error_message': 'service_unavailable', 'retry_in': 45 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.ERROR) # === Completion Status Tests === def test_completion_success(self): """Test completion status with successful tasks.""" data = { 'pending_directives': 3, 'completed_today': 15, 'failed': 1 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.COMPLETION) self.assertIn("15", result.formatted_text) self.assertIn("1", result.formatted_text) def test_completion_no_failures(self): """Test completion status with no failures.""" data = { 'completed_today': 10, 'failed': 0 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("10", result.formatted_text) # Should handle zero failures gracefully self.assertIn("0", result.formatted_text) def test_completion_multiple_failures(self): """Test completion status with multiple failures.""" data = { 'completed_today': 20, 'failed': 3 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("3", result.formatted_text) self.assertIn("failures", result.formatted_text.lower()) def test_completion_zero_completed(self): """Test completion status with zero completed tasks.""" data = { 'completed_today': 0, 'failed': 0 } result = self.formatter.format_status(data) self.assertTrue(result.success) # === Queue Status Tests === def test_queue_single_item(self): """Test queue status with single item.""" data = { 'pending_directives': 1 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.QUEUE) self.assertIn("1", result.formatted_text) def test_queue_multiple_items(self): """Test queue status with multiple items.""" data = { 'pending_directives': 5 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("5", result.formatted_text) self.assertIn("items", result.formatted_text.lower()) def test_queue_zero_items(self): """Test queue status with zero items.""" data = { 'pending_directives': 0 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("0", result.formatted_text) # === System Health Tests === def test_system_health_basic(self): """Test basic system health status.""" data = { 'cpu': 45, 'memory': 62, 'active_agents': 3 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.SYSTEM_HEALTH) self.assertIn("45", result.formatted_text) self.assertIn("62", result.formatted_text) self.assertIn("3", result.formatted_text) def test_system_health_single_agent(self): """Test system health with single agent.""" data = { 'cpu': 30, 'memory': 40, 'active_agents': 1 } result = self.formatter.format_status(data) self.assertTrue(result.success) # Should use singular "agent" self.assertIn("1", result.formatted_text) def test_system_health_zero_agents(self): """Test system health with zero agents.""" data = { 'cpu': 10, 'memory': 20, 'active_agents': 0 } result = self.formatter.format_status(data) self.assertTrue(result.success) # === Unknown Status Tests === def test_unknown_status(self): """Test unknown status type.""" data = { 'custom_field': 'some_value', 'another_field': 42 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.UNKNOWN) def test_empty_data(self): """Test with empty data.""" data = {} result = self.formatter.format_status(data) self.assertTrue(result.success) # === JSON String Input Tests === def test_json_string_input(self): """Test formatting from JSON string.""" json_str = json.dumps({ 'task': 'testing', 'progress': 80, 'eta_minutes': 5 }) result = self.formatter.format_status(json_str) self.assertTrue(result.success) self.assertEqual(result.status_type, StatusType.PROGRESS) def test_invalid_json_string(self): """Test with invalid JSON string.""" result = self.formatter.format_status("not valid json {{{") self.assertFalse(result.success) self.assertIsNotNone(result.error_message) # === Multiple Status Tests === def test_format_multiple(self): """Test formatting multiple statuses.""" statuses = [ {'task': 'first', 'progress': 50, 'eta_minutes': 10}, {'error': 'timeout', 'retry_in': 30}, {'pending_directives': 5}, ] results = self.formatter.format_multiple(statuses) self.assertEqual(len(results), 3) self.assertTrue(all(r.success for r in results)) def test_format_summary(self): """Test summary of multiple statuses.""" statuses = [ {'task': 'first', 'progress': 50, 'eta_minutes': 10}, {'completed_today': 15, 'failed': 2}, {'pending_directives': 3}, ] summary = self.formatter.format_summary(statuses) self.assertIsInstance(summary, str) self.assertGreater(len(summary), 0) def test_format_summary_empty(self): """Test summary with empty list.""" summary = self.formatter.format_summary([]) self.assertEqual(summary, "No status information available.") # === Edge Cases === def test_none_value(self): """Test handling of None values in data.""" data = { 'task': None, 'progress': 50, 'eta_minutes': None } result = self.formatter.format_status(data) self.assertTrue(result.success) def test_special_characters(self): """Test handling of special characters in data.""" data = { 'task': 'processing "quotes" & chars', 'progress': 50, 'eta_minutes': 5 } result = self.formatter.format_status(data) self.assertTrue(result.success) def test_large_numbers(self): """Test handling of large numbers.""" data = { 'completed_today': 9999, 'failed': 1 } result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("9999", result.formatted_text) # === Example Transforms from Requirements === def test_example_progress(self): """Test exact example from requirements: progress update.""" data = {'status': 'running', 'task': 'building_api', 'progress': 65, 'eta_minutes': 12} result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("building the API", result.formatted_text.lower()) self.assertIn("65", result.formatted_text) self.assertIn("12", result.formatted_text) def test_example_queue(self): """Test exact example from requirements: queue status.""" data = {'pending_directives': 3, 'completed_today': 15, 'failed': 1} result = self.formatter.format_status(data) self.assertTrue(result.success) # This should match completion type since it has completed_today self.assertIn("15", result.formatted_text) self.assertIn("1", result.formatted_text) def test_example_error(self): """Test exact example from requirements: error report.""" data = {'error': 'database_connection_failed', 'retry_in': 30} result = self.formatter.format_status(data) self.assertTrue(result.success) self.assertIn("database connection", result.formatted_text.lower()) self.assertIn("30", result.formatted_text) class TestStatusTypeDetection(unittest.TestCase): """Test status type detection logic.""" def setUp(self): self.formatter = StatusFormatter() def test_detect_progress(self): """Test progress detection.""" data = {'progress': 50, 'task': 'working'} status_type = self.formatter._detect_status_type(data) self.assertEqual(status_type, StatusType.PROGRESS) def test_detect_error(self): """Test error detection.""" data = {'error': 'something_failed'} status_type = self.formatter._detect_status_type(data) self.assertEqual(status_type, StatusType.ERROR) def test_detect_completion(self): """Test completion detection.""" data = {'completed_today': 10} status_type = self.formatter._detect_status_type(data) self.assertEqual(status_type, StatusType.COMPLETION) def test_detect_queue(self): """Test queue detection.""" data = {'pending_directives': 5} status_type = self.formatter._detect_status_type(data) self.assertEqual(status_type, StatusType.QUEUE) def test_detect_system_health(self): """Test system health detection.""" data = {'cpu': 50, 'memory': 60} status_type = self.formatter._detect_status_type(data) self.assertEqual(status_type, StatusType.SYSTEM_HEALTH) if __name__ == '__main__': # Run tests with verbose output unittest.main(verbosity=2)