Follower Mode

def get_close(self) -> bool:
    """
    Check if the plan is closed.
    :return: True if the plan need closed, False otherwise.
    """

    return self.plan.get("close", False)

def get_host_agent_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    object_name = self.get_operation_object()

    request = (
        f"Open and select the application of {object_name}, and output the FINISH status immediately, without assigning any subtask"
        "You must output the selected application with their control text and label even if it is already open."
    )

    return request

def get_host_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    task = self.get_task()
    object_name = self.get_operation_object()
    if object_name in self.support_apps:
        request = task
    else:
        request = (
            f"Your task is '{task}'. And open the application of {object_name}. "
            "You must output the selected application with their control text and label even if it is already open."
        )
    return request

def get_initial_request(self) -> str:
    """
    Get the initial request in the plan.
    :return: The initial request.
    """

    task = self.get_task()
    object_name = self.get_operation_object()

    request = f"{task} in {object_name}"

    return request

def get_operation_object(self) -> str:
    """
    Get the operation object in the step.
    :return: The operation object.
    """

    return self.plan.get("object", None).lower()

def get_root_path(self) -> str:
    """
    Get the root path of the plan.
    :return: The root path of the plan.
    """

    return os.path.dirname(os.path.abspath(self.plan_file))

def get_steps(self) -> List[str]:
    """
    Get the steps in the plan.
    :return: The steps in the plan.
    """

    return self.plan.get("steps", [])

def get_support_apps(self) -> List[str]:
    """
    Get the support apps in the plan.
    :return: The support apps in the plan.
    """

    return self.support_apps

def get_task(self) -> str:
    """
    Get the task name.
    :return: The task name.
    """

    return self.plan.get("task", "")

def next_step(self) -> Optional[str]:
    """
    Get the next step in the plan.
    :return: The next step.
    """

    if self.remaining_steps:
        step = self.remaining_steps.pop(0)
        return step

    return None

def task_finished(self) -> bool:
    """
    Check if the task is finished.
    :return: True if the task is finished, False otherwise.
    """

    return not self.remaining_steps

def create_new_round(self) -> None:
    """
    Create a new round.
    """
    from ufo.agents.agent.host_agent import HostAgent

    # Get a request for the new round.
    request = self.next_request()

    # Create a new round and return None if the session is finished.
    if self.is_finished():
        return None

    if self.total_rounds == 0:
        console.print("Complete the following request:", style="yellow")
        console.print(self.plan_reader.get_initial_request(), style="cyan")
        agent: HostAgent = self._host_agent
    else:
        host_agent: HostAgent = self._host_agent
        self.context.set(ContextNames.SUBTASK, request)
        agent = host_agent.create_subagent(context=self.context)

        # Clear the memory and set the state to continue the app agent.
        agent.clear_memory()
        agent.blackboard.requests.clear()

        agent.set_state(ContinueAppAgentState())

    round = BaseRound(
        request=request,
        agent=agent,
        context=self.context,
        should_evaluate=ufo_config.system.eva_round,
        id=self.total_rounds,
    )

    self.add_round(round.id, round)

    return round

def next_request(self) -> str:
    """
    Get the request for the new round.
    """

    # If the task is finished, return an empty string.
    if self.plan_reader.task_finished():
        self._finish = True
        return ""

    # Get the request from the plan reader.
    if self.total_rounds == 0:
        return self.plan_reader.get_host_agent_request()
    else:
        return self.plan_reader.next_step()

def request_to_evaluate(self) -> str:
    """
    Get the request to evaluate.
    return: The request(s) to evaluate.
    """

    return self.plan_reader.get_task()