fix(traces): support flat trace structure in ha_get_automation_traces (homeassistant-ai#529)

* fix(traces): support flat trace structure in ha_get_automation_traces

* test(e2e): enhance trace test to verify parsing of triggers and actions

* fix(traces): restore support for legacy trace structure keys

Author: julienld

src/ha_mcp/tools/tools_traces.py

@@ -411,11 +411,47 @@ def _format_detailed_trace(
         "state": trace.get("state"),
     }
 
+    raw_trace = trace.get("trace", {})
+    
+    # Initialize lists
+    triggers = []
+    conditions = []
+    actions = []
+    
+    # Home Assistant trace data is stored as a flat dict with path keys
+    # e.g. "trigger/0": [...], "action/0": [...], "action/0/1": [...]
+    for path, steps in raw_trace.items():
+        if not isinstance(steps, list):
+            continue
+            
+        for step in steps:
+            # Create a copy to avoid modifying original
+            step_info = step.copy()
+            step_info["path"] = path
+            
+            if path == "trigger" or path.startswith("trigger/"):
+                triggers.append(step_info)
+            elif path == "condition" or path.startswith("condition/"):
+                conditions.append(step_info)
+            elif path == "action" or path.startswith("action/"):
+                actions.append(step_info)
+    
+    # Sort by timestamp (if available) or path to maintain execution order
+    def sort_key(item):
+        return (item.get("timestamp", ""), item.get("path", ""))
+        
+    triggers.sort(key=sort_key)
+    conditions.sort(key=sort_key)
+    actions.sort(key=sort_key)
+
     # Extract trigger information
-    trigger_trace = trace.get("trace", {}).get("trigger", [])
-    if trigger_trace:
-        trigger_step = trigger_trace[0]
-        trigger_vars = trigger_step.get("variables", {}).get("trigger", {})
+    if triggers:
+        trigger_step = triggers[0]
+        trigger_vars = trigger_step.get("changed_variables", {}).get("trigger", {})
+        # Sometimes variables are in 'variables' key, sometimes 'changed_variables'
+        if not trigger_vars:
+            trigger_vars = trigger_step.get("variables", {}).get("trigger", {})
+            
         result["trigger"] = {
             "platform": trigger_vars.get("platform"),
             "description": trigger_vars.get("description"),
@@ -427,26 +463,29 @@ def _format_detailed_trace(
             result["trigger"]["from_state"] = trigger_vars.get("from_state", {}).get("state")
         if "entity_id" in trigger_vars:
             result["trigger"]["entity_id"] = trigger_vars["entity_id"]
+    
+    # If no trigger info found in traces, try to get it from the top-level trigger field if present
+    # (some HA versions might populate this)
+    if "trigger" not in result and "trigger" in trace:
+        result["trigger"] = {"description": trace["trigger"]}
 
     # Extract condition results
-    condition_trace = trace.get("trace", {}).get("condition", [])
-    if condition_trace:
+    if conditions:
         condition_results = []
-        for cond in condition_trace:
+        for cond in conditions:
             cond_result = {
                 "result": cond.get("result", {}).get("result"),
+                "path": cond.get("path"),
             }
-            # Try to get condition type from the path
-            if "path" in cond:
-                cond_result["path"] = cond["path"]
+            if "timestamp" in cond:
+                cond_result["timestamp"] = cond["timestamp"]
             condition_results.append(cond_result)
         result["condition_results"] = condition_results
 
     # Extract action trace
-    action_trace = trace.get("trace", {}).get("action", [])
-    if action_trace:
+    if actions:
         action_results = []
-        for action in action_trace:
+        for action in actions:
             action_info: dict[str, Any] = {
                 "path": action.get("path"),
             }
@@ -465,12 +504,17 @@ def _format_detailed_trace(
                 action_info["error"] = action["error"]
 
             # Extract variables if they contain useful debugging info
-            variables = action.get("variables", {})
+            # Check both 'variables' and 'changed_variables'
+            variables = action.get("variables") or action.get("changed_variables", {})
             if variables and "trigger" not in variables:  # Skip trigger vars (already shown)
                 # Only include non-empty variable sets
                 useful_vars = {k: v for k, v in variables.items() if v is not None}
                 if useful_vars:
                     action_info["variables"] = useful_vars
+            
+            # Add child execution info (for nested scripts/automations)
+            if "child_id" in action:
+                action_info["child_id"] = action["child_id"]
 
             action_results.append(action_info)
 

tests/src/e2e/workflows/automation/test_traces.py

@@ -170,7 +170,19 @@ async def check_automation_traces():
 
             assert detailed_data.get("success") is True
             assert detailed_data.get("run_id") == run_id
-            logger.info(f"Detailed trace retrieved for run_id: {run_id}")
+            
+            # Verify detailed content structure (Deep verification)
+            # This ensures we correctly parsed the flat structure (trigger/0, action/0)
+            assert "trigger" in detailed_data, "Detailed trace should contain trigger info"
+            assert "action_trace" in detailed_data, "Detailed trace should contain action_trace"
+            assert isinstance(detailed_data["action_trace"], list), "action_trace should be a list"
+            assert len(detailed_data["action_trace"]) > 0, "action_trace should not be empty"
+            
+            # Check for path property to ensure flat structure parsing worked
+            first_action = detailed_data["action_trace"][0]
+            assert "path" in first_action, "Action trace element should contain 'path'"
+            
+            logger.info(f"Detailed trace verified: Found {len(detailed_data['action_trace'])} actions")
 
     async def test_empty_traces_with_diagnostics(
         self, mcp_client, cleanup_tracker, test_data_factory

tests/src/unit/test_tools_traces_detail.py

@@ -0,0 +1,131 @@
+"""Unit tests for tools_traces detailed trace formatting."""
+
+import pytest
+from ha_mcp.tools.tools_traces import _format_detailed_trace
+
+class TestFormatDetailedTrace:
+    """Test _format_detailed_trace function."""
+
+    def test_format_flat_trace_structure(self):
+        """Test parsing of Home Assistant's flat path-based trace structure."""
+        
+        # Data structure as provided by user
+        trace_data = {
+            "timestamp": {"start": "2026-01-29T23:05:00.345824+00:00", "finish": "2026-01-29T23:05:00.356669+00:00"},
+            "state": "stopped",
+            "trigger": "time",
+            "trace": {
+                "trigger/0": [{
+                    "path": "trigger/0",
+                    "timestamp": "2026-01-29T23:05:00.345915+00:00",
+                    "changed_variables": {
+                        "trigger": {
+                            "platform": "time",
+                            "description": "time",
+                            "entity_id": None
+                        }
+                    }
+                }],
+                "action/0": [{
+                    "path": "action/0",
+                    "timestamp": "2026-01-29T23:05:00.346301+00:00",
+                    "result": {"params": {"domain": "light", "service": "turn_on"}}
+                }],
+                "action/0/0": [{
+                    "path": "action/0/0",
+                    "timestamp": "2026-01-29T23:05:00.347072+00:00",
+                    "child_id": {"domain": "script", "item_id": "set_brightness_chambre", "run_id": "04e0241d"},
+                    "result": {"params": {"domain": "script"}}
+                }]
+            },
+            "config": {
+                "alias": "Lumières Chambre 18h05",
+                "mode": "single"
+            }
+        }
+
+        result = _format_detailed_trace("automation.test", "run_123", trace_data)
+
+        assert result["success"] is True
+        assert result["automation_id"] == "automation.test"
+        assert result["run_id"] == "run_123"
+        
+        # Verify Trigger
+        assert "trigger" in result
+        assert result["trigger"]["platform"] == "time"
+        assert result["trigger"]["description"] == "time"
+        
+        # Verify Actions
+        assert "action_trace" in result
+        actions = result["action_trace"]
+        assert len(actions) == 2
+        
+        # Sort order should be preserved (action/0 then action/0/0)
+        assert actions[0]["path"] == "action/0"
+        assert actions[1]["path"] == "action/0/0"
+        
+        # Verify content of actions
+        assert actions[0]["result"]["params"]["service"] == "turn_on"
+        assert actions[1]["child_id"]["item_id"] == "set_brightness_chambre"
+
+    def test_format_legacy_trace_structure(self):
+        """Test fallback parsing of potential legacy trace structure (lists)."""
+        
+        trace_data = {
+            "timestamp": "2026-01-29T23:05:00",
+            "state": "stopped",
+            "trace": {
+                "trigger": [{
+                    "path": "trigger/0",
+                    "variables": {
+                        "trigger": {
+                            "platform": "state",
+                            "description": "state change"
+                        }
+                    }
+                }],
+                "action": [{
+                    "path": "action/0",
+                    "result": {"executed": True}
+                }]
+            }
+        }
+
+        result = _format_detailed_trace("automation.legacy", "run_456", trace_data)
+
+        assert result["success"] is True
+        
+        # Verify Trigger
+        assert result["trigger"]["platform"] == "state"
+        
+        # Verify Actions
+        assert len(result["action_trace"]) == 1
+        assert result["action_trace"][0]["result"]["executed"] is True
+
+    def test_format_mixed_variables_location(self):
+        """Test that variables are found whether in 'variables' or 'changed_variables'."""
+        
+        trace_data = {
+            "trace": {
+                "trigger/0": [{
+                    "variables": {
+                        "trigger": {"platform": "variables_key"}
+                    }
+                }],
+                "trigger/1": [{
+                    "changed_variables": {
+                        "trigger": {"platform": "changed_variables_key"}
+                    }
+                }]
+            }
+        }
+        
+        # Test finding in 'variables' (legacy/standard)
+        result1 = _format_detailed_trace("auto.1", "1", 
+            {"trace": {"trigger/0": trace_data["trace"]["trigger/0"]}})
+        assert result1["trigger"]["platform"] == "variables_key"
+        
+        # Test finding in 'changed_variables' (new flat format)
+        result2 = _format_detailed_trace("auto.2", "2", 
+            {"trace": {"trigger/0": trace_data["trace"]["trigger/1"]}})
+        assert result2["trigger"]["platform"] == "changed_variables_key"