ConsoleUI

@contextmanager
def create(self, app: Runnable, import_string: str) -> Iterator[None]:
    yield

def create_subconversation(self) -> "ConsoleUI":
    sub_conversation = ConsoleUI(self)
    self.sub_conversations.append(sub_conversation)

    return sub_conversation

def create_workflow_ui(self: UIBase, workflow_uuid: str) -> "UI":
    return UI(uibase=self, workflow_uuid=workflow_uuid)

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]:
    # logger.info(f"process_message(): {message=}")
    return self.visit(message)

def start(
    self,
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    params: dict[str, Any],
    single_run: bool = False,
) -> None:
    workflow_uuid = uuid4().hex
    ui = self.create_workflow_ui(workflow_uuid=workflow_uuid)
    name = name or app.provider.names[0]
    app.provider.run(name=name, ui=ui, **params)

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) -> None:
    if hasattr(message, "content"):
        content = message.content
        console_msg = self.ConsoleMessage(
            sender=getattr(content, "sender", None) or "Workflow",
            recipient=getattr(content, "recipient", None) or "User",
            heading=message.type,
            body=getattr(content, "content", None),
        )
        self._format_and_print(console_msg)
    else:
        content = message.model_dump()["content"]
        console_msg = self.ConsoleMessage(
            sender=getattr(message, "sender", None) or "Workflow",
            recipient=getattr(message, "recipient", None) or "User",
            heading=message.type,
            body=json.dumps(content, indent=2),
        )
        self._format_and_print(console_msg)

def visit_error(self, message: Union[Error, ErrorEvent]) -> None:
    # Handle both fastagency Error IOMessage and ag2 ErrorEvent
    # fastagency Error has: short, long, sender, recipient
    # ag2 ErrorEvent has: content.error, content.uuid
    if isinstance(message, Error):
        # fastagency Error IOMessage
        error_msg = message.long or message.short or "Unknown error"
        sender = message.sender or "Workflow"
        recipient = message.recipient or "User"
    elif isinstance(message, ErrorEvent):
        # ag2 ErrorEvent
        content = message.content
        error_obj = getattr(content, "error", content)
        error_msg = str(error_obj) if error_obj is not None else "Unknown error"
        sender = "Workflow"
        recipient = "User"
    else:
        # Unknown error type - fallback
        error_msg = str(message)
        sender = "Workflow"
        recipient = "User"

    console_msg = self.ConsoleMessage(
        sender=sender,
        recipient=recipient,
        heading="error",
        body=error_msg,
    )
    self._format_and_print(console_msg)

def visit_execute_function(self, message: "ExecuteFunctionEvent") -> None:
    content = message.content

    body = f"\n>>>>>>>> EXECUTING FUNCTION {content.func_name}...\nCall ID: {content.call_id}\nInput arguments: {content.arguments}"
    console_msg = self.ConsoleMessage(
        sender="Workflow",
        recipient=content.recipient,
        heading=message.type,
        body=body,
    )
    self._format_and_print(console_msg)

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

def visit_input_request(self, message: "InputRequestEvent") -> str:
    prompt = message.content.prompt
    if message.content.password:
        result = getpass.getpass(prompt if prompt != "" else "Password: ")
    else:
        result = input(prompt)
    message.content.respond(result)
    return result

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

def visit_multiple_choice(self, message: MultipleChoice) -> str:
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=f"{message.prompt} (choices: {', '.join(message.choices)}, default: {message.default})",
    )

    prompt = self._format_message(console_msg)
    prompt = self._indent(prompt)
    while True:
        # logger.info(f"visit_multiple_choice(): {prompt=}")
        retval = input(prompt)
        if retval in message.choices:
            return retval
        elif retval == "" and message.default:
            return message.default
        else:
            print(f"Invalid choice ('{retval}'). Please try again.")  # noqa: T201 `print` found

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
) -> Optional[str]:
    return self.visit_default(message)

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

def visit_termination(self, message: "TerminationEvent") -> None:
    pass

def visit_text(self, message: "TextEvent") -> None:
    content = message.content
    console_msg = self.ConsoleMessage(
        sender=content.sender,
        recipient=content.recipient,
        heading=message.type,
        body=content.content,
    )
    self._format_and_print(console_msg)

def visit_text_input(self, message: TextInput) -> str:
    suggestions = (
        f" (suggestions: {', '.join(message.suggestions)})"
        if message.suggestions
        else ""
    )
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=f"{message.prompt}{suggestions}:",
    )

    prompt = self._format_message(console_msg)
    prompt = self._indent(prompt)
    if message.password:
        return getpass.getpass(prompt)
    else:
        return input(prompt)

def visit_text_message(self, message: TextMessage) -> None:
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=message.body,
    )
    self._format_and_print(console_msg)

def visit_using_auto_reply(self, message: "UsingAutoReplyEvent") -> None:
    # Do nothing if it is of type UsingAutoReplyEvent
    pass

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,
        )
    )