Function parse_cli_event

messaging/event_parser.py:12–181

Parse a CLI event and return a structured result. Args: event: Raw event dictionary from CLI log_raw_cli: When True, log full error text from the CLI. Default is metadata-only (lengths / exit codes) to avoid leaking user content. Returns: List of pa

(event: Any, *, log_raw_cli: bool = False)

def parse_cli_event(event: Any, *, log_raw_cli: bool = False) -> list[dict]:
    """
    Parse a CLI event and return a structured result.

    Args:
        event: Raw event dictionary from CLI
        log_raw_cli: When True, log full error text from the CLI. Default is
            metadata-only (lengths / exit codes) to avoid leaking user content.

    Returns:
        List of parsed event dicts. Empty list if not recognized.
    """
    if not isinstance(event, dict):
        return []

    etype = event.get("type")
    results: list[dict[str, Any]] = []

    # Some CLI/proxy layers emit "system" events that are not user-visible and
    # carry no transcript content. Ignore them explicitly to avoid noisy logs.
    if etype == "system":
        return []

    # 1. Handle full messages (assistant/user or result)
    msg_obj = None
    if etype == "assistant" or etype == "user":
        msg_obj = event.get("message")
    elif etype == "result":
        res = event.get("result")
        if isinstance(res, dict):
            msg_obj = res.get("message")
            # Some variants put content directly on the result.
            if not msg_obj and isinstance(res.get("content"), list):
                msg_obj = {"content": res.get("content")}
        if not msg_obj:
            msg_obj = event.get("message")
        # Some variants put content directly on the event.
        if not msg_obj and isinstance(event.get("content"), list):
            msg_obj = {"content": event.get("content")}

    if msg_obj and isinstance(msg_obj, dict):
        content = msg_obj.get("content", [])
        if isinstance(content, list):
            # Preserve order exactly as content blocks appear.
            for c in content:
                if not isinstance(c, dict):
                    continue
                ctype = c.get("type")
                if ctype == "text":
                    results.append({"type": "text_chunk", "text": c.get("text", "")})
                elif ctype == "thinking":
                    results.append(
                        {"type": "thinking_chunk", "text": c.get("thinking", "")}
                    )
                elif ctype == "tool_use":
                    results.append(
                        {
                            "type": "tool_use",