CurrentMessage

def create_message(self) -> IOMessage:
    kwargs = {k: v for k, v in asdict(self).items() if v is not None}
    # logger.info(f"CurrentMessage.create_message(): {kwargs=}")
    return IOMessage.create(**kwargs)

def process_chunk(self, chunk: str) -> bool:  # noqa: C901
    # logger.info(f"CurrentMessage.process_chunk({chunk=}):")
    if _match("end_of_message", chunk):
        return True

    if _match("auto_reply", chunk):
        # logger.info("CurrentMessage.process_chunk(): auto_reply detected")
        self.auto_reply = True
    elif _match("sender_recipient", chunk):
        # logger.info("CurrentMessage.process_chunk(): sender_recipient detected")
        self.sender, self.recipient = _findall("sender_recipient", chunk)
    elif _match("suggested_function_call", chunk):
        # logger.info("CurrentMessage.process_chunk(): suggested_function_call detected")
        self.call_id, self.function_name = _findall(
            "suggested_function_call", chunk
        )
        self.type = "suggested_function_call"
    elif _match("stars", chunk):
        # logger.info("CurrentMessage.process_chunk(): stars detected")
        pass
    elif _match("function_call_execution", chunk):
        # logger.info("CurrentMessage.process_chunk(): function_call_execution detected")
        self.function_name = _findall("function_call_execution", chunk)  # type: ignore[assignment]
        self.type = "function_call_execution"
    elif _match("response_from_calling_tool", chunk):
        # logger.info("CurrentMessage.process_chunk(): response_from_calling_tool detected")
        self.type = "function_call_execution"
        self.call_id = _findall("response_from_calling_tool", chunk)  # type: ignore[assignment]
    elif _match("no_human_input_received", chunk):
        # logger.info("CurrentMessage.process_chunk(): no_human_input_received detected")
        pass
    elif _match("user_interrupted", chunk):
        # logger.info("CurrentMessage.process_chunk(): user_interrupted detected")
        pass
    else:
        if self.type == "suggested_function_call":
            if _match("arguments", chunk):
                # logger.info("CurrentMessage.process_chunk(): parsing arguments")
                arguments_json: str = _findall("arguments", chunk)  # type: ignore[assignment]
                self.arguments = json.loads(arguments_json)
            else:
                logger.warning(
                    f"CurrentMessage.process_chunk(): unexpected chunk: {chunk=}, {self=}"
                )
        elif self.type == "function_call_execution":
            # logger.info("CurrentMessage.process_chunk(): parsing retval")
            self.retval = chunk
        else:
            # logger.info("CurrentMessage.process_chunk(): parsing body")
            self.body = chunk if self.body is None else self.body + chunk

    return False

def process_input(
    self, prompt: str, password: bool, messages: list[IOMessage]
) -> AskingMessage:
    last_message = messages[-1]
    sender, recipient = None, None
    message: AskingMessage

    if _match("auto_reply_input", prompt):
        # logger.info("IOStreamAdapter.input(): auto_reply_input detected")
        sender, recipient = _findall("auto_reply_input", prompt)  # type: ignore[assignment]

    if last_message.type == "suggested_function_call":
        # logger.info("IOStreamAdapter.input(): suggested_function_call detected")
        message = MultipleChoice(
            sender=sender,
            recipient=recipient,
            prompt="Please approve the suggested function call.",
            choices=["Approve", "Reject", "Exit"],
            default="Approve",
        )
    else:
        # logger.info("IOStreamAdapter.input(): text_message detected")
        message = TextInput(
            sender=None, recipient=None, prompt=prompt, password=password
        )

    return message