amphibious¶

class Exposure(ABC, Generic[T]):
    """
    Abstract base class for field-level data exposure.

    Manages list-like data (e.g., history records, tool lists) with a unified interface.
    Subclasses determine the exposure strategy:
    - LayeredExposure: supports progressive disclosure (summary + per-item details)
    - EntireExposure: only provides summary (no per-item details)

    Methods
    -------
    add(item)
        Add an element and return its index.
    summary()
        Return a list of summary strings for all elements.
    get_all()
        Return a copy of all elements.
    """

    def __init__(self):
        self._items: List[T] = []
        self._llm: Optional[Any] = None

    def __len__(self) -> int:
        return len(self._items)

    def __getitem__(self, index: int) -> T:
        return self._items[index]

    def __iter__(self) -> Iterator[T]:
        return iter(self._items)

    def add(self, item: T) -> int:
        """
        Add an element to the collection.

        Parameters
        ----------
        item : T
            The element to add.

        Returns
        -------
        int
            Index of the newly added element (0-based).
        """
        self._items.append(item)
        return len(self._items) - 1

    def get_all(self) -> List[T]:
        """
        Return a copy of all elements.

        Returns
        -------
        List[T]
            A copy of all elements.
        """
        return self._items.copy()

    def set_llm(self, llm: Any) -> None:
        """
        Set the LLM for this exposure.

        Stored as ``self._llm`` and available to subclasses that need it
        (e.g. ``CognitiveHistory`` uses it for compression).
        Called automatically by ``Context.set_llm()`` for fields marked with
        ``json_schema_extra={"use_llm": True}``.

        Parameters
        ----------
        llm : Any
            LLM instance to store.
        """
        self._llm = llm

class LayeredExposure(Exposure[T]):
    """
    Exposure with progressive disclosure support.

    Provides two-level information architecture:
    - summary(): overview of all items
    - get_details(index): detailed information for a specific item

    Use this for data where the LLM may need to request details
    about specific items (e.g., execution history, skills).

    Disclosure state is owned internally: once revealed, an item's detail
    is cached in _revealed. Call reset_revealed() to clear all cached reveals
    (e.g., at phase boundaries in a multi-phase agent).
    """

    def __init__(self):
        super().__init__()
        self._revealed: Dict[int, str] = {}  # index → cached detail string

    def reveal(self, index: int) -> Optional[str]:
        """
        Get and cache detailed information for a specific element.

        Returns the cached value if already revealed; otherwise calls
        get_details(index), stores the result, and returns it.

        Parameters
        ----------
        index : int
            Element index (0-based).

        Returns
        -------
        Optional[str]
            Detailed information string, or None if index is invalid.
        """
        if index in self._revealed:
            return self._revealed[index]
        detail = self.get_details(index)
        if detail is not None:
            self._revealed[index] = detail
        return detail

    def reset_revealed(self) -> None:
        """
        Clear all cached reveals.

        Use at phase boundaries to allow the LLM to re-request details
        that were disclosed in a previous phase.
        """
        self._revealed.clear()

    @abstractmethod
    def summary(self) -> List[str]:
        """
        Generate summary strings for all elements.

        Returns
        -------
        List[str]
            One summary string per element.
        """
        ...

    @abstractmethod
    def get_details(self, index: int) -> Optional[str]:
        """
        Get detailed information for a specific element.

        Parameters
        ----------
        index : int
            Element index (0-based).

        Returns
        -------
        Optional[str]
            Detailed information string, or None if index is invalid.
        """
        ...

class EntireExposure(Exposure[T]):
    """
    Exposure without progressive disclosure.

    Only provides summary() - all information is exposed at once.
    Use this for data where per-item details are not needed
    or the full information should always be available (e.g., tools).
    """
    @abstractmethod
    def summary(self) -> List[str]:
        """
        Generate summary strings for all elements.

        Returns
        -------
        List[str]
            One summary string per element.
        """
        ...

class Context(BaseModel):
    """
    Base class for agent context with automatic Exposure field detection.

    Provides unified access to context data through summary and detail retrieval.
    Automatically discovers Exposure-typed fields and distinguishes between
    LayeredExposure (supports details) and EntireExposure (summary only).

    Methods
    -------
    summary()
        Get a dictionary of all field values or Exposure summaries.
    get_details(field, idx)
        Get detailed information for a specific LayeredExposure item.
    get_revealed_items()
        Get all (field, index) pairs that have been revealed so far.
    reset_revealed()
        Clear reveal caches on all LayeredExposure fields.

    Examples
    --------
    >>> class MyContext(Context):
    ...     model_config = ConfigDict(arbitrary_types_allowed=True)
    ...     goal: str
    ...     history: CognitiveHistory = Field(default_factory=CognitiveHistory)  # LayeredExposure
    ...     tools: CognitiveTools = Field(default_factory=CognitiveTools)  # EntireExposure
    ...
    >>> ctx = MyContext(goal="Complete task")
    >>> ctx.history.add(Step(content="Step 1", status=True))
    >>> ctx.summary()  # Returns dict with goal and summaries of history/tools
    >>> ctx.get_details("history", 0)  # Works - history is LayeredExposure
    >>> ctx.get_details("tools", 0)  # Returns None - tools is EntireExposure
    """

    # Class-level cache for detected fields
    _exposure_fields: Optional[Dict[str, Dict[str, Any]]] = None

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._exposure_fields = None

    def model_post_init(self, __context: Any) -> None:
        super().model_post_init(__context)
        if self.__class__._exposure_fields is None:
            self.__class__._exposure_fields = self._detect_exposure_fields()

        # Call __post_init__ if defined in subclass (provides dataclass-like API)
        if hasattr(self, '__post_init__') and callable(getattr(self, '__post_init__')):
            self.__post_init__()

    @classmethod
    def _detect_exposure_fields(cls) -> Dict[str, Dict[str, Any]]:
        """Detect all Exposure fields and classify them."""
        exposure_fields = {}

        def _get_exposure_type(field_type: Any) -> Optional[str]:
            """Return 'layered', 'entire', or None."""
            if inspect.isclass(field_type):
                if issubclass(field_type, LayeredExposure):
                    return 'layered'
                elif issubclass(field_type, EntireExposure):
                    return 'entire'
                elif issubclass(field_type, Exposure):
                    # Base Exposure - treat as entire (no details)
                    return 'entire'
                else:
                    return None

            origin = get_origin(field_type)
            if origin and inspect.isclass(origin):
                if issubclass(origin, LayeredExposure):
                    return 'layered'
                elif issubclass(origin, EntireExposure):
                    return 'entire'
                elif issubclass(origin, Exposure):
                    return 'entire'
                else:
                    return None

            return None

        for field_name, field_info in cls.model_fields.items():
            exposure_type = _get_exposure_type(field_info.annotation)
            if exposure_type:
                exposure_fields[field_name] = {
                    'field_name': field_name,
                    'exposure_type': exposure_type,  # 'layered' or 'entire'
                }

        return exposure_fields

    @classmethod
    def get_hidden_fields(cls) -> List[str]:
        """
        Get list of field names that should be hidden from summary.

        Returns
        -------
        List[str]
            Field names where display is explicitly set to False.

        Examples
        --------
        >>> class MyContext(Context):
        ...     visible: str = Field(default="")
        ...     hidden: str = Field(default="", json_schema_extra={"display": False})
        >>> MyContext.get_hidden_fields()
        ['hidden']
        """
        hidden = []
        for field_name, field_info in cls.model_fields.items():
            extra = field_info.json_schema_extra or {}
            display = extra.get("display", True)
            if not display:
                hidden.append(field_name)
        return hidden

    def set_llm(self, llm: Any) -> None:
        """
        Propagate an LLM to all Exposure fields that opt in via ``use_llm=True``.

        Called by ``AmphibiousAutoma`` at run start. Fields opt in by declaring:

            my_field: MyExposure = Field(
                ..., json_schema_extra={"use_llm": True}
            )

        Only Exposure fields are considered; non-Exposure fields are ignored.
        Exposure fields without ``use_llm=True`` are also ignored.

        Parameters
        ----------
        llm : Any
            The LLM instance to propagate.
        """
        exposure_fields = self.__class__._exposure_fields or {}
        for field_name in exposure_fields:
            field_info = self.__class__.model_fields.get(field_name)
            if field_info is None:
                continue
            extra = field_info.json_schema_extra or {}
            if extra.get("use_llm", False):
                field_value = getattr(self, field_name, None)
                if field_value is not None:
                    field_value.set_llm(llm)

    def summary(self) -> Dict[str, str]:
        """
        Generate a summary dictionary with formatted strings for each field.

        Automatically filters out fields marked with json_schema_extra={"display": False}.
        Subclasses should override this to provide custom formatting for each field.
        The returned dictionary maps field names to their formatted string representations.

        Returns
        -------
        Dict[str, str]
            Field name to formatted summary string mapping.
            Each value should be a complete, formatted string ready for prompt inclusion.
            Hidden fields (display=False) are excluded.
        """
        result = {}
        exposure_fields = self.__class__._exposure_fields or {}
        hidden_fields = set(self.get_hidden_fields())

        # Add non-Exposure fields as simple string (skip hidden fields)
        for field_name in self.__class__.model_fields:
            if field_name not in exposure_fields and field_name not in hidden_fields:
                value = getattr(self, field_name)
                if value is not None:
                    # Get field description if available
                    field_info = self.__class__.model_fields.get(field_name)
                    description = field_info.description if field_info and field_info.description else None

                    if description:
                        result[field_name] = f"{field_name} ({description}):\n {value}"
                    else:
                        result[field_name] = f"{field_name}:\n {value}"

        # Add Exposure field summaries (skip hidden fields)
        for field_name in exposure_fields:
            if field_name not in hidden_fields:
                field_value = getattr(self, field_name)
                if field_value and len(field_value) > 0:
                    # Get field description if available
                    field_info = self.__class__.model_fields.get(field_name)
                    description = field_info.description if field_info and field_info.description else None
                    summaries = field_value.summary()
                    if description:
                        result[field_name] = f"{field_name} ({description}):\n" + "\n".join(f"  {s}" for s in summaries)
                    else:
                        result[field_name] = f"{field_name}:\n" + "\n".join(f"  {s}" for s in summaries)

        return result

    def format_summary(
        self,
        include: Optional[List[str]] = None,
        exclude: Optional[List[str]] = None,
        separator: str = "\n"
    ) -> str:
        """
        Format the summary dictionary into a string with field selection.

        Parameters
        ----------
        include : Optional[List[str]]
            If provided, only include these fields (takes priority over exclude).
        exclude : Optional[List[str]]
            If provided, exclude these fields from the output.
        separator : str
            Separator between field summaries. Default is newline.

        Returns
        -------
        str
            Formatted summary string with selected fields.

        Examples
        --------
        >>> ctx.format_summary()  # All fields
        >>> ctx.format_summary(include=['tools', 'skills'])  # Only capabilities
        >>> ctx.format_summary(exclude=['tools', 'skills'])  # Only task context
        """
        summary_dict = self.summary()

        if include is not None:
            fields = [f for f in include if f in summary_dict]
        elif exclude is not None:
            fields = [f for f in summary_dict if f not in exclude]
        else:
            fields = list(summary_dict.keys())

        return separator.join(summary_dict[f] for f in fields if summary_dict.get(f))

    def get_field(self, field: str) -> Tuple[Optional[List[str]], Any]:
        """
        Get field information with type-aware return.

        Parameters
        ----------
        field : str
            Name of the field.

        Returns
        -------
        Tuple[Optional[List[str]], Any]
            - If field is an Exposure: (summary list, all items via get_all())
            - Otherwise: (None, raw field value)
        """
        exposure_fields = self.__class__._exposure_fields or {}
        field_value = getattr(self, field, None)

        if field in exposure_fields:
            # Exposure field: return (summary, get_all())
            if field_value and hasattr(field_value, 'summary') and hasattr(field_value, 'get_all'):
                return (field_value.summary(), field_value.get_all())
            return ([], [])
        else:
            # Non-Exposure field: return (None, value)
            return (None, field_value)

    def get_details(self, field: str, idx: int) -> Optional[str]:
        """
        Get detailed information for a LayeredExposure field item.

        Only works for LayeredExposure fields. Returns None for EntireExposure.

        Parameters
        ----------
        field : str
            Name of the Exposure field.
        idx : int
            Item index within the field (0-based).

        Returns
        -------
        Optional[str]
            Detailed information string, or None if:
            - Field doesn't exist
            - Field is not a LayeredExposure
            - Index is invalid
        """
        exposure_fields = self.__class__._exposure_fields or {}

        if field not in exposure_fields:
            return None

        # Only LayeredExposure supports get_details
        if exposure_fields[field].get('exposure_type') != 'layered':
            return None

        field_value = getattr(self, field, None)
        if field_value and hasattr(field_value, 'reveal'):
            return field_value.reveal(idx)

        return None

    def get_revealed_items(self) -> List[Tuple[str, int]]:
        """
        Return all (field_name, index) pairs that have been revealed across
        all LayeredExposure fields on this context.

        Returns
        -------
        List[Tuple[str, int]]
            Ordered list of (field_name, item_index) that have cached reveals.
        """
        result: List[Tuple[str, int]] = []
        for fname, fval in self:
            if isinstance(fval, LayeredExposure):
                for idx in fval._revealed:
                    result.append((fname, idx))
        return result

    def reset_revealed(self) -> None:
        """
        Reset reveal state on all LayeredExposure fields.

        Equivalent to calling field.reset_revealed() on every LayeredExposure
        field. Use at phase boundaries so the next phase can request fresh details.
        """
        for _, fval in self:
            if isinstance(fval, LayeredExposure):
                fval.reset_revealed()

    def __iter__(self) -> Iterator[Tuple[str, Any]]:
        """Yield ``(field_name, field_value)`` for every model field on this context."""
        for field_name in type(self).model_fields:
            yield field_name, getattr(self, field_name)

    def __str__(self) -> str:
        """
        Return a formatted string representation of the context.

        Automatically formats all defined fields:
        - Exposure fields: displays their summary
        - Other fields: displays field name and value

        Returns
        -------
        str
            Formatted string representation.
        """
        exposure_fields = self.__class__._exposure_fields or {}
        lines = []
        separator = "-" * 50

        lines.append(f"{'=' * 50}")
        lines.append(f"  {self.__class__.__name__}")
        lines.append(f"{'=' * 50}")

        # Format non-Exposure fields first
        for field_name in self.__class__.model_fields:
            if field_name not in exposure_fields:
                value = getattr(self, field_name)
                if value is not None:
                    lines.append(f"\n[{field_name}]")
                    lines.append(f"  {value}")

        # Format Exposure fields
        for field_name, field_info in exposure_fields.items():
            field_value = getattr(self, field_name, None)
            lines.append(f"\n[{field_name}] ({field_info.get('exposure_type', 'unknown')})")
            if field_value and len(field_value) > 0:
                for i, summary in enumerate(field_value.summary()):
                    lines.append(f"  [{i}] {summary}")
            else:
                lines.append("  (empty)")

        lines.append(f"\n{'=' * 50}")
        return "\n".join(lines)

    def __repr__(self) -> str:
        """Return a concise representation of the context."""
        exposure_fields = self.__class__._exposure_fields or {}
        parts = []

        for field_name in self.__class__.model_fields:
            if field_name not in exposure_fields:
                value = getattr(self, field_name)
                if value is not None:
                    parts.append(f"{field_name}={value!r}")

        for field_name in exposure_fields:
            field_value = getattr(self, field_name, None)
            count = len(field_value) if field_value else 0
            parts.append(f"{field_name}=[{count} items]")

        return f"{self.__class__.__name__}({', '.join(parts)})"