MesopGUIMessageVisitor

def error(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    short: Optional[str] = None,
    long: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        Error(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            short=short,
            long=long,
        )
    )

def function_call_execution(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    retval: Any = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        FunctionCallExecution(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            function_name=function_name,
            call_id=call_id,
            retval=retval,
        )
    )

def keep_alive(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        KeepAlive(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
        )
    )

def multiple_choice(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    choices: Optional[list[str]] = None,
    default: Optional[str] = None,
    single: bool = True,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    choices = choices or []
    return self.process_message(
        MultipleChoice(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            prompt=prompt,
            choices=choices,
            default=default,
            single=single,
        )
    )

def process_message(self, message: IOMessage) -> Optional[str]:
    try:
        return self.visit(message)
    except Exception as e:
        logger.warning(f"Failed to render message: {e}")
        self.render_error_message(e, message)
        return None

def render_error_message(
    self,
    e: Exception,
    message: IOMessage,
    *,
    content: Optional[str] = None,
    style: Optional[MesopMessageStyles] = None,
) -> None:
    style = self._styles.message.error or self._styles.message.default
    title = "[Error] " + message.type.replace("_", " ").capitalize()

    with me.box(style=style.box or self._styles.message.default.box):
        self._header(
            message,
            title=title,
            box_style=style.header_box or self._styles.message.default.header_box,
            md_style=style.header_md or self._styles.message.default.header_md,
        )

        content = (
            "Failed to render message:"
            + json.dumps(message.model_dump(), indent=2, cls=UUIDEncoder)
            + f"<br>Error: {e}"
        )

        logger.warning(f"render_error_message: {content=}")
        logger.warning(e, exc_info=True)
        # me.markdown(content, style=style.md or self._styles.message.default.md)
        self._render_content(content, style.md or self._styles.message.default.md)

def suggested_function_call(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    arguments: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    arguments = arguments or {}
    return self.process_message(
        SuggestedFunctionCall(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            function_name=function_name,
            call_id=call_id,
            arguments=arguments,
        )
    )

def system_message(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    message: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    message = message or {}
    return self.process_message(
        SystemMessage(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            message=message,
        )
    )

def text_input(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    suggestions: Optional[list[str]] = None,
    password: bool = False,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    suggestions = suggestions or []
    return self.process_message(
        TextInput(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            prompt=prompt,
            suggestions=suggestions,
            password=password,
        )
    )

def text_message(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    body: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        TextMessage(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            body=body,
        )
    )

def visit(self, message: IOMessage) -> Optional[str]:
    method_name = f"visit_{message.type}"
    method = getattr(self, method_name, self.visit_default)
    return method(message)

def visit_default(
    self,
    message: IOMessage,
    *,
    content: Optional[str] = None,
    style: Optional[MesopMessageStyles] = None,
    error: Optional[bool] = False,
    inner_callback: Optional[Callable[..., None]] = None,
    scrollable: Optional[bool] = False,
) -> None:
    # logger.info(f"visit_default: {message=}")
    style = style or self._styles.message.default
    title = message.type.replace("_", " ").capitalize()
    title = "[Error] " + title if error else title
    with me.box(style=style.box or self._styles.message.default.box):
        self._header(
            message,
            title=title,
            box_style=style.header_box,
            md_style=style.header_md,
        )

        if isinstance(message, IOMessage):
            content = content or json.dumps(
                message.model_dump()["content"], cls=UUIDEncoder
            )
        else:
            content = (
                message.content.content
                if hasattr(message, "content")
                and hasattr(message.content, "content")
                else ""
            )
        content = self._body_to_str(content)

        self._render_content(
            content,
            msg_md_style=style.scrollable_md
            or self._styles.message.default.scrollable_md
            if scrollable
            else style.md or self._styles.message.default.md,
        )

        if inner_callback:
            inner_callback()

def visit_error(self, message: Error) -> None:
    self.visit_default(
        message,
        content=f"### {message.short}\n{message.long}",
        style=self._styles.message.error,
        scrollable=True,
    )

    def visit_function_call_execution(self, message: FunctionCallExecution) -> None:
        content = f"""**function_name**: `{message.function_name}`<br>
**call_id**: `{message.call_id}`<br>
**retval**: {message.retval}"""
        return self.visit_default(
            message,
            content=content,
            style=self._styles.message.function_call_execution,
            scrollable=True,
        )

def visit_keep_alive(self, message: KeepAlive) -> None:
    def on_wakeup(e: mel.WebEvent) -> Iterator[None]:
        logger.debug("waking up, after the keep alive")
        self._conversation_message.feedback_completed = True
        yield from consume_responses(get_more_messages())

    with me.box():
        if not (self._readonly or self._conversation_message.feedback_completed):
            wakeup_component(on_wakeup=on_wakeup)

def visit_multiple_choice(self, message: MultipleChoice) -> str:
    if message.single:
        return self._visit_single_choice(message)
    else:
        return self._visit_many_choices(message)

def visit_run_completion(self, message: "RunCompletionEvent") -> None:
    # We can ignore the RunCompletionEvent as we handle RunResponse already
    pass

    def visit_suggested_function_call(self, message: SuggestedFunctionCall) -> None:
        content = f"""**function_name**: `{message.function_name}`<br>
**call_id**: `{message.call_id}`<br>
**arguments**: {json.dumps(message.arguments)}"""
        self.visit_default(
            message,
            content=content,
            style=self._styles.message.suggested_function_call,
            scrollable=True,
        )

    def visit_system_message(self, message: SystemMessage) -> None:
        content = (
            f"""#### **{message.message["heading"]}**

{message.message["body"]}
"""
            if "heading" in message.message and "body" in message.message
            else json.dumps(message.message)
        )
        self.visit_default(
            message,
            content=content,
            style=self._styles.message.system,
            scrollable=True,
        )

def visit_text(self, message: "TextEvent") -> None:
    content = message.content.content
    self.visit_default(
        message,
        content=content,
        style=self._styles.message.text,
    )

def visit_text_input(self, message: TextInput) -> str:
    def on_input(feedback: str) -> Iterator[None]:
        self._conversation_message.feedback = [feedback]
        self._conversation_message.feedback_completed = True
        yield from self._provide_feedback(feedback)

    def value_if_completed() -> Optional[str]:
        message = self._conversation_message
        return message.feedback[0] if message.feedback_completed else None

    # base_color = "#dff"
    prompt = message.prompt if message.prompt else "Please enter a value"
    if message.suggestions:
        suggestions = ",".join(suggestion for suggestion in message.suggestions)
        prompt += "\n Suggestions: " + suggestions

    self.visit_default(
        message,
        content=prompt,
        style=self._styles.message.text_input,
        inner_callback=lambda: input_text(
            on_input,
            key="prompt",
            disabled=self._readonly or self._has_feedback(),
            value=value_if_completed(),
            style=self._styles.message.text_input_inner,
        ),
    )
    return ""

def visit_text_message(self, message: TextMessage) -> None:
    content = message.body if message.body else ""
    content = content if content.strip() != "" else "*(empty message)*"
    self.visit_default(
        message,
        content=content,
        style=self._styles.message.text,
    )

def visit_using_auto_repy(self, message: "UsingAutoReplyEvent") -> None:
    content = None
    self.visit_default(
        message,
        content=content,
        style=self._styles.message.system,
    )

def visit_workflow_completed(self, message: WorkflowCompleted) -> Optional[str]:
    return self.visit_default(message)

def visit_workflow_started(self, message: WorkflowStarted) -> Optional[str]:
    return self.visit_default(message)

def workflow_completed(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    result: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        WorkflowCompleted(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            result=result,
        )
    )

def workflow_started(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    params: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    params = params or {}
    return self.process_message(
        WorkflowStarted(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            name=name,
            description=description,
            params=params,
        )
    )