agent_base/base.py

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, Iterable, Optional, Sequence


def _normalize_function_list(function_list: Optional[Iterable[str]]) -> Optional[list[str]]:
    if function_list is None:
        return None
    normalized: list[str] = []
    for raw_name in function_list:
        name = str(raw_name).strip()
        if name:
            normalized.append(name)
    return normalized


def agent_role(
    *,
    name: str,
    role_prompt: str = "",
    function_list: Optional[Iterable[str]] = None,
):
    """
    Class decorator used by upper-layer frameworks to declare agent defaults.

    This keeps the lower-layer execution loop generic while allowing subclasses
    to provide role-specific prompt addenda and tool restrictions declaratively.
    """

    def decorator(cls):
        cls.role_name = str(name).strip() or cls.__name__
        cls.default_role_prompt = str(role_prompt).strip()
        cls.default_function_list = _normalize_function_list(function_list)
        return cls

    return decorator


class BaseAgent(ABC):
    """Abstract base class for agents built on top of ResearchHarness."""

    role_name: str = "agent"
    default_role_prompt: str = ""
    default_function_list: Optional[list[str]] = None

    @classmethod
    def resolve_function_list(cls, function_list: Optional[Sequence[str]]) -> Optional[list[str]]:
        if function_list is not None:
            return _normalize_function_list(function_list) or []
        default_tools = getattr(cls, "default_function_list", None)
        if default_tools is None:
            return None
        return list(default_tools)

    @classmethod
    def resolve_role_prompt(cls, role_prompt: Optional[str]) -> str:
        if role_prompt is None:
            role_prompt = getattr(cls, "default_role_prompt", "")
        return str(role_prompt or "").strip()

    def should_accept_plaintext_result(
        self,
        *,
        result_text: str,
        workspace_root: Optional[str],
        messages: Sequence[dict[str, Any]],
    ) -> bool:
        """
        Decide whether a plain assistant text reply with no tool calls is terminal.

        The default behavior preserves the original ResearchHarness semantics:
        any meaningful assistant text without tool calls is accepted as the final
        result. Upper layers may override this hook to require extra completion
        artifacts before termination.
        """

        return True

    def rejected_plaintext_result_message(
        self,
        *,
        result_text: str,
        workspace_root: Optional[str],
        messages: Sequence[dict[str, Any]],
    ) -> str:
        """
        Explain why a plain assistant text reply was not accepted as terminal.

        Returning an empty string falls back to the generic runtime message.
        """

        return ""

    def should_accept_terminal_error(
        self,
        *,
        error_text: str,
        workspace_root: Optional[str],
        messages: Sequence[dict[str, Any]],
    ) -> bool:
        """
        Decide whether a terminal LLM/runtime error can still be accepted.

        The default behavior is conservative: terminal errors are not accepted.
        Upper layers may override this hook when benchmark-specific completion
        artifacts are already present and the remaining assistant text is not
        semantically important.
        """

        return False

    def accepted_terminal_error_result_text(
        self,
        *,
        error_text: str,
        workspace_root: Optional[str],
        messages: Sequence[dict[str, Any]],
    ) -> str:
        """
        Provide a synthetic terminal result when a terminal error is accepted.

        Returning an empty string falls back to a generic runtime completion
        message.
        """

        return ""

    @abstractmethod
    def run(self, prompt: str, workspace_root: Optional[str] = None):
        raise NotImplementedError