src/agents/db_executor/codeact/utils/pretty_state.py

import json
from rich.console import Console
from rich.syntax import Syntax
from rich.panel import Panel
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

console = Console(width=100, soft_wrap=False)

_last_context_snapshot = None  # used to suppress repeated context
_last_message_ids = set()  # track printed messages



def serialize_message(msg) -> dict:
    """Convert LangChain message objects into serializable dicts."""
    if hasattr(msg, "dict"):
        return msg.dict()
    elif hasattr(msg, "__dict__"):
        return {k: serialize_message(v) for k, v in msg.__dict__.items()}
    elif isinstance(msg, list):
        return [serialize_message(v) for v in msg]
    elif isinstance(msg, dict):
        return {k: serialize_message(v) for k, v in msg.items()}
    else:
        return msg


def pretty_print_state(state: dict, show_context: bool = True) -> None:
    """
    Pretty-print the agent's state in a clean, color-coded way.

    Parameters
    ----------
    state : dict
        The LangGraph agent state chunk (from the stream).
    show_context : bool, optional
        Whether to display the context (default True).
        If True, only shows context when it has changed since last call.
    """
    global _last_context_snapshot

    # --- Display message chunks ---
    for msg in state.get("messages", []):

        msg_id = getattr(msg, "id", id(msg))
        if msg_id in _last_message_ids:
            continue  # skip duplicates
        _last_message_ids.add(msg_id)

        msg_dict = serialize_message(msg)
        msg_json = json.dumps(msg_dict, indent=2)

        if isinstance(msg, HumanMessage):
            color, title = "cyan", "🧑 HumanMessage"
        elif isinstance(msg, ToolMessage):
            color, title = "yellow", f"🧰 ToolMessage ({msg_dict.get('name','?')})"
        elif isinstance(msg, AIMessage):
            color, title = "magenta", "🤖 AIMessage"
        else:
            color, title = "white", "Other"

        syntax = Syntax(msg_json, "json", theme="monokai", line_numbers=False)
        console.print(Panel(syntax, title=title, border_style=color))

    # --- Optional context view ---
    #if show_context:
    #    context = state.get("context", {})
    #    if context and context != _last_context_snapshot:
    #        _last_context_snapshot = context.copy()  # cache for next comparison

    #        context_json = json.dumps(context, indent=2, default=str)
    #        syntax = Syntax(context_json, "json", theme="monokai", line_numbers=False)
     #       console.print(Panel(syntax, title="🧠 Context (updated)", border_style="green"))