Full Module Index

class MappingComponentHandler:
    """Handler for one string value in a mapping the caller owns.

    Serializes, applies and restores ``mapping[key]``. The agent argument of
    each method is ignored, because the value does not live on the agent.

    Attributes:
        mapping (MutableMapping[str, str]): The caller's mapping, held by
            reference so writes are visible to its other readers.
        key (str): The mapping key this handler evolves.

    Examples:
        ```python
        prompts = {"greeting": "hello"}
        handler = MappingComponentHandler(prompts, key="greeting")
        original = handler.apply(None, "hi")  # prompts["greeting"] == "hi"
        handler.restore(None, original)  # prompts["greeting"] == "hello"
        ```

    Notes:
        The handler keeps a reference to the mapping, not a copy. Replacing
        the mapping object elsewhere detaches the handler from it.
    """

    def __init__(self, mapping: MutableMapping[str, str], key: str) -> None:
        """Bind the handler to one key of the caller's mapping.

        Args:
            mapping: The caller-owned mapping holding the prompt text.
            key: The key whose value this handler evolves.

        Raises:
            ValueError: If ``key`` is not in ``mapping``.
            TypeError: If ``mapping[key]`` is not a ``str``.

        Examples:
            ```python
            handler = MappingComponentHandler({"k": "v"}, key="k")
            ```
        """
        if key not in mapping:
            raise ValueError(f"Mapping has no key {key!r} to evolve")
        value: object = mapping[key]
        if not isinstance(value, str):
            raise TypeError(
                f"Mapping value for key {key!r} must be str, got {type(value).__name__}"
            )
        self.mapping = mapping
        self.key = key

    def serialize(self, agent: Any) -> str:
        """Return the current text of the mapping key.

        Args:
            agent: Ignored; may be ``None``.

        Returns:
            The value of ``mapping[key]``, or an empty string when the caller
            has removed the key or replaced its value with non-text since
            registration.

        Examples:
            ```python
            text = handler.serialize(None)
            ```

        Notes:
            Never raises, per the ``ComponentHandler`` contract. A missing or
            non-text value is logged as ``mapping_handler.key_missing``.
        """
        value = self.mapping.get(self.key)
        if not isinstance(value, str):
            logger.warning(
                "mapping_handler.key_missing",
                key=self.key,
                value_type=type(value).__name__,
            )
            return ""
        return value

    def apply(self, agent: Any, value: str) -> str:
        """Write new text to the mapping key and return the previous text.

        Args:
            agent: Ignored; may be ``None``.
            value: The new text for the key.

        Returns:
            The text the key held before this call.

        Examples:
            ```python
            original = handler.apply(None, "new text")
            ```
        """
        original = self.mapping[self.key]
        self.mapping[self.key] = value
        logger.debug(
            "mapping_handler.apply",
            key=self.key,
            original_preview=original[:50],
            new_preview=value[:50],
        )
        return original

    def restore(self, agent: Any, original: str) -> None:
        """Write the original text back to the mapping key.

        Args:
            agent: Ignored; may be ``None``.
            original: The text returned by ``apply()``.

        Examples:
            ```python
            handler.restore(None, original)
            ```
        """
        self.mapping[self.key] = original
        logger.debug("mapping_handler.restore", key=self.key, preview=original[:50])

class LabelAgreementScorer:
    """Scorer that measures exact agreement between output and label.

    With ``field`` set, the output is parsed as a JSON object and
    ``str(parsed[field]).strip()`` is compared with ``expected.strip()``.
    Without ``field``, the output and ``expected`` agree when both parse as equal JSON
    documents, or otherwise when their stripped texts are equal.

    Attributes:
        field (str | None): Name of the output JSON field holding the
            label, or ``None`` to compare the whole output.

    Examples:
        Evolve a schema agent against labelled rows:

        ```python
        from gepa_adk import LabelAgreementScorer, evolve, run_sync

        trainset = [{"input": "Buy now!!!", "expected": "spam"}]
        result = run_sync(
            evolve(agent, trainset, scorer=LabelAgreementScorer(field="label"))
        )
        ```

    Notes:
        Adheres to the Scorer protocol. The score is 0.0 with
        ``metadata["reason"]`` set to ``"missing_expected"`` when the row has
        no label, ``"output_not_json"`` when ``field`` is set and the output
        does not parse, and ``"field_missing"`` when the parsed output is not
        an object or lacks ``field``. Metadata always carries ``field``,
        ``actual``, ``expected`` and ``agreement``.
    """

    def __init__(self, field: str | None = None) -> None:
        """Store the output field to compare.

        Args:
            field: Name of the JSON field holding the label, or ``None`` to
                compare the whole output.
        """
        self.field = field

    def _result(
        self,
        *,
        actual: str | None,
        expected: str | None,
        agreement: bool,
        reason: str | None = None,
    ) -> tuple[float, dict[str, Any]]:
        """Build the score and metadata pair.

        Args:
            actual: Extracted value that was compared, if any.
            expected: Label from the trainset row.
            agreement: Whether the value agreed with the label.
            reason: Why no comparison was possible, if applicable.

        Returns:
            The score and its metadata dictionary.
        """
        metadata: dict[str, Any] = {
            "field": self.field,
            "actual": actual,
            "expected": expected,
            "agreement": agreement,
        }
        if reason is not None:
            metadata["reason"] = reason
        return (1.0 if agreement else 0.0), metadata

    def score(
        self,
        input_text: str,
        output: str,
        expected: str | None = None,
    ) -> tuple[float, dict[str, Any]]:
        """Score the output by agreement with the expected label.

        Args:
            input_text: Input given to the agent. Not used.
            output: Agent output text.
            expected: Label from the trainset row.

        Returns:
            ``(1.0, metadata)`` on agreement, else ``(0.0, metadata)``.
        """
        if expected is None:
            return self._result(
                actual=None,
                expected=None,
                agreement=False,
                reason="missing_expected",
            )
        if self.field is None:
            return self._score_whole(output, expected)
        parsed = _parse_json(output)
        if parsed is _MISSING:
            return self._result(
                actual=None,
                expected=expected,
                agreement=False,
                reason="output_not_json",
            )
        if not isinstance(parsed, dict) or self.field not in parsed:
            return self._result(
                actual=None,
                expected=expected,
                agreement=False,
                reason="field_missing",
            )
        actual = str(parsed[self.field]).strip()
        return self._result(
            actual=actual,
            expected=expected,
            agreement=actual == expected.strip(),
        )

    def _score_whole(self, output: str, expected: str) -> tuple[float, dict[str, Any]]:
        """Compare the whole output with the label.

        Args:
            output: Agent output text.
            expected: Label from the trainset row.

        Returns:
            The score and its metadata dictionary.
        """
        actual = output.strip()
        parsed_output = _parse_json(output)
        parsed_expected = _parse_json(expected)
        if parsed_output is not _MISSING and parsed_expected is not _MISSING:
            agreement = parsed_output == parsed_expected
        else:
            agreement = actual == expected.strip()
        return self._result(actual=actual, expected=expected, agreement=agreement)

    async def async_score(
        self,
        input_text: str,
        output: str,
        expected: str | None = None,
    ) -> tuple[float, dict[str, Any]]:
        """Score asynchronously by delegating to ``score``.

        Args:
            input_text: Input given to the agent. Not used.
            output: Agent output text.
            expected: Label from the trainset row.

        Returns:
            The same result as ``score``.
        """
        return self.score(input_text, output, expected)

class RequireToolScorer:
    """Scorer that gates an inner scorer on a required tool call.

    Attributes:
        inner (Scorer): Wrapped scorer consulted when the tool ran.
        tool (str): Name of the tool that must appear in the trajectory.

    Examples:
        ```python
        from gepa_adk import LabelAgreementScorer, RequireToolScorer, evolve

        scorer = RequireToolScorer(LabelAgreementScorer(field="label"), tool="ask_x")
        result = await evolve(agent, trainset, scorer=scorer)
        ```

    Notes:
        Adheres to the Scorer protocol. Metadata always carries
        ``required_tool`` and ``tool_called``. The score is 0.0 with
        ``metadata["reason"]`` set to ``"trajectory_unavailable"`` when no
        trajectory was given and ``"tool_not_called"`` when the tool is
        absent. Otherwise the inner scorer's score and metadata are returned.
    """

    def __init__(self, inner: Scorer, tool: str) -> None:
        """Store the wrapped scorer and the required tool name.

        Args:
            inner: Scorer to delegate to when the tool ran.
            tool: Name of the tool that must appear in the trajectory.

        Raises:
            ValueError: If ``tool`` is not a non-blank string.
            TypeError: If ``inner`` lacks ``score`` or ``async_score``.

        Notes:
            Records separately whether ``inner.score`` and
            ``inner.async_score`` accept ``trajectory`` by keyword.
        """
        if not isinstance(tool, str) or not tool.strip():
            raise ValueError(f"tool must be a non-blank string, got {tool!r}")
        if not hasattr(inner, "score") or not hasattr(inner, "async_score"):
            raise TypeError(
                f"inner must implement the Scorer protocol, got {type(inner)}"
            )
        self.inner = inner
        self.tool = tool
        # Each method is inspected separately: an inner scorer may declare
        # ``trajectory`` on one and not the other.
        self._inner_async_accepts_trajectory = scorer_accepts_trajectory(inner)
        self._inner_sync_accepts_trajectory = _method_accepts_trajectory(inner.score)

    def _gate(
        self, trajectory: ADKTrajectory | None
    ) -> tuple[float, dict[str, Any]] | None:
        """Return the zero result when the tool check fails.

        Args:
            trajectory: Row trajectory, or ``None`` when unavailable.

        Returns:
            ``(0.0, metadata)`` when the trajectory is missing or lacks the
            tool, else ``None`` to signal delegation.
        """
        if trajectory is None:
            reason = "trajectory_unavailable"
        elif any(call.name == self.tool for call in trajectory.tool_calls):
            return None
        else:
            reason = "tool_not_called"
        return 0.0, {
            "reason": reason,
            "required_tool": self.tool,
            "tool_called": False,
        }

    def _merge(self, result: Any) -> tuple[float, dict[str, Any]]:
        """Attach the tool metadata to the inner scorer's result.

        Args:
            result: Inner result, either ``(score, metadata)`` or a bare float.

        Returns:
            The inner score with its metadata plus ``required_tool`` and
            ``tool_called``.
        """
        if isinstance(result, tuple):
            score, metadata = result[0], dict(result[1] or {})
        else:
            score, metadata = float(result), {}
        metadata["required_tool"] = self.tool
        metadata["tool_called"] = True
        return score, metadata

    def score(
        self,
        input_text: str,
        output: str,
        expected: str | None = None,
        *,
        trajectory: ADKTrajectory | None = None,
    ) -> tuple[float, dict[str, Any]]:
        """Score synchronously, requiring the tool before delegating.

        Args:
            input_text: Input given to the agent.
            output: Agent output text.
            expected: Label from the trainset row, if any.
            trajectory: Row trajectory from the adapter, if available.

        Returns:
            ``(0.0, metadata)`` when the tool did not run, else the inner
            scorer's result with the tool metadata added.

        Notes:
            Forwards ``trajectory`` only when the corresponding inner method
            declares it.
        """
        gated = self._gate(trajectory)
        if gated is not None:
            return gated
        if self._inner_sync_accepts_trajectory:
            # The Scorer protocol omits ``trajectory``; the inner scorer
            # declared it, so call through an untyped reference.
            inner: Any = self.inner
            result = inner.score(input_text, output, expected, trajectory=trajectory)
        else:
            result = self.inner.score(input_text, output, expected)
        return self._merge(result)

    async def async_score(
        self,
        input_text: str,
        output: str,
        expected: str | None = None,
        *,
        trajectory: ADKTrajectory | None = None,
    ) -> tuple[float, dict[str, Any]]:
        """Score asynchronously, requiring the tool before delegating.

        Args:
            input_text: Input given to the agent.
            output: Agent output text.
            expected: Label from the trainset row, if any.
            trajectory: Row trajectory from the adapter, if available.

        Returns:
            ``(0.0, metadata)`` when the tool did not run, else the inner
            scorer's result with the tool metadata added.

        Notes:
            Forwards ``trajectory`` only when the corresponding inner method
            declares it.
        """
        gated = self._gate(trajectory)
        if gated is not None:
            return gated
        if self._inner_async_accepts_trajectory:
            # The Scorer protocol omits ``trajectory``; the inner scorer
            # declared it, so call through an untyped reference.
            inner: Any = self.inner
            result = await inner.async_score(
                input_text, output, expected, trajectory=trajectory
            )
        else:
            result = await self.inner.async_score(input_text, output, expected)
        return self._merge(result)