1  """
  2  Provides debugging utilities for the declarative test framework,
  3  including a pretty-printer for A2A event history.
  4  """
  5  
  6  import json
  7  from datetime import datetime
  8  from typing import List, Dict, Any
  9  
 10  
 11  def _truncate(s: str, max_len: int) -> str:
 12      """Truncates a string if it exceeds max_len, appending '...'."""
 13      if not isinstance(s, str):
 14          s = str(s)
 15      if max_len <= 0 or len(s) <= max_len:
 16          return s
 17      if max_len <= 3:
 18          return s[:max_len]
 19      return s[: max_len - 3] + "..."
 20  
 21  
 22  def _format_a2a_parts(parts: List[Dict], max_string_length: int) -> str:
 23      """Helper to format A2A message parts into a readable string."""
 24      if not parts:
 25          return "[No Parts]"
 26  
 27      formatted_parts = []
 28      for part in parts:
 29          part_type = part.get("type", "unknown")
 30          if part_type == "text":
 31              text = part.get("text", "")
 32              formatted_parts.append(
 33                  f"  - [Text]: '{_truncate(text, max_string_length)}'"
 34              )
 35          elif part_type == "data":
 36              data_content = json.dumps(part.get("data", {}))
 37              formatted_parts.append(
 38                  f"  - [Data]: {_truncate(data_content, max_string_length)}"
 39              )
 40          elif part_type == "file":
 41              file_info = part.get("file", {})
 42              name = file_info.get("name", "N/A")
 43              mime = file_info.get("mimeType", "N/A")
 44              formatted_parts.append(
 45                  f"  - [File]: {_truncate(name, max_string_length)} ({mime})"
 46              )
 47          else:
 48              part_str = json.dumps(part)
 49              formatted_parts.append(
 50                  f"  - [Unknown Part]: {_truncate(part_str, max_string_length)}"
 51              )
 52      return "\n".join(formatted_parts)
 53  
 54  
 55  def _truncate_dict_strings(data: Any, max_len: int) -> Any:
 56      """Recursively traverses a dict/list and truncates all string values."""
 57      if max_len <= 0:
 58          return data
 59      if isinstance(data, dict):
 60          return {k: _truncate_dict_strings(v, max_len) for k, v in data.items()}
 61      elif isinstance(data, list):
 62          return [_truncate_dict_strings(item, max_len) for item in data]
 63      elif isinstance(data, str):
 64          return _truncate(data, max_len)
 65      else:
 66          return data
 67  
 68  
 69  def pretty_print_event_history(
 70      event_history: List[Dict[str, Any]], max_string_length: int = 200
 71  ):
 72      """
 73      Formats and prints a list of A2A event payloads for debugging.
 74      """
 75      if not event_history:
 76          print("\n" + "=" * 25 + " NO EVENTS RECORDED " + "=" * 25)
 77          print("--- The test failed before any events were received from the agent. ---")
 78          print("=" * 70 + "\n")
 79          return
 80  
 81      print("\n" + "=" * 25 + " TASK EVENT HISTORY " + "=" * 25)
 82      for i, event_payload in enumerate(event_history):
 83          print(f"\n--- Event {i+1} ---")
 84  
 85          event_type = "Unknown Event"
 86          details = ""
 87  
 88          result = event_payload.get("result", {})
 89          error = event_payload.get("error")
 90  
 91          if error:
 92              event_type = "Error Response"
 93              details += f"  Error Code: {error.get('code')}\n"
 94              details += f"  Error Message: {_truncate(error.get('message'), max_string_length)}\n"
 95  
 96          elif result.get("status") and result.get("final") is not None:
 97              event_type = "Task Status Update"
 98              status = result.get("status", {})
 99              state = status.get("state", "UNKNOWN")
100              message = status.get("message", {})
101              parts = message.get("parts", [])
102              details += f"  State: {state}\n"
103              details += f"  Parts:\n{_format_a2a_parts(parts, max_string_length)}\n"
104  
105          elif result.get("status") and result.get("sessionId"):
106              event_type = "Final Task Response"
107              status = result.get("status", {})
108              state = status.get("state", "UNKNOWN")
109              message = status.get("message", {})
110              parts = message.get("parts", [])
111              details += f"  Final State: {state}\n"
112              details += (
113                  f"  Final Parts:\n{_format_a2a_parts(parts, max_string_length)}\n"
114              )
115              if result.get("artifacts"):
116                  artifacts_str = json.dumps(result.get("artifacts"))
117                  details += (
118                      f"  Artifacts: {_truncate(artifacts_str, max_string_length)}\n"
119                  )
120  
121          elif result.get("artifact"):
122              event_type = "Task Artifact Update"
123              artifact = result.get("artifact", {})
124              details += f"  Artifact Name: {_truncate(artifact.get('name'), max_string_length)}\n"
125              details += f"  Artifact Parts:\n{_format_a2a_parts(artifact.get('parts', []), max_string_length)}\n"
126  
127          print(f"Type: {event_type}")
128          if details:
129              print(details, end="")
130  
131          print("Raw Payload:")
132          truncated_payload = _truncate_dict_strings(event_payload, max_string_length)
133          print(json.dumps(truncated_payload, indent=2))
134  
135      print("=" * 70 + "\n")