"""GeminiVisionEvalNode — adapter wrapping the Phase 2 gemini_vision provider behind the EvalNode Protocol (CP-9 Phase 4). Design: Phase 3 (`pipeline/core/eval.py`) defined the EvalNode Protocol but deferred the only "real" production EvalNode to Phase 4 because it depends on the Gemini Vision adapter shipped in Phase 2. This module is the bridge. Placement decision (per Phase 4 hand-off note): the node is shared across three eval modality runners (image / video / audio) — each runner instantiates its own `GeminiVisionEvalNode(artifact_modality=...)` so the node knows which modality literal to pass to `gemini_vision.score_artifact`. Living in a dedicated `_gemini_vision_eval_node.py` module avoids duplicating ~80 lines across the three runner files and avoids picking one runner as the canonical home (which would create a stealth dependency arrow). Underscore prefix signals "shared internal" — the public surface is the EvalNode it produces, not the class itself, though it is exported via `pipeline.core.runners`. The class implements the runtime-checkable :class:`EvalNode` Protocol: judge_id: str model_used: str evaluate(context: EvalContext) -> EvalResult Internally `evaluate(...)` calls :func:`execution.providers.gemini_vision.score_artifact` and translates the returned :class:`EvalProviderResult` into an :class:`EvalResult`. Adapter-level exceptions (`EvalProviderError` and subclasses) propagate to the caller — the wrapping eval runner is responsible for translating them into failure-RunResults (so the runner contract "always returns RunResult" remains intact). """ from __future__ import annotations import logging from typing import Any, Callable, Optional from recoil.execution.providers import gemini_vision as _gv from recoil.pipeline.core.eval import EvalContext, EvalResult logger = logging.getLogger(__name__) class GeminiVisionEvalNode: """EvalNode Protocol implementation — wraps gemini_vision.score_artifact. Constructor takes the artifact modality literal so the node can be registered three times (one per modality) under three judge_ids without branching in `evaluate()`. Stateless after construction (no per-call side state). Args: artifact_modality: One of ``"image"`` / ``"video"`` / ``"audio"``. Validated against :data:`gemini_vision.SUPPORTED_MODALITIES`. judge_id: Stable judge id surfaced on the EvalResult and threaded into the adapter's raw_metadata for panel-level correlation. Defaults to ``f"eval_{artifact_modality}_v1"`` (mirrors the registry's MODALITY_EVAL_*_V1 string convention). model_id: Gemini model id. Defaults to :data:`gemini_vision.DEFAULT_MODEL_ID` (``"gemini-3.1-pro-preview"``). api_key_env_var: env var to read the API key from. Defaults to ``GEMINI_API_KEY`` with ``GOOGLE_API_KEY`` as fallback (per adapter precedence). timeout_s: per-call timeout, threaded into score_artifact. """ def __init__( self, *, artifact_modality: str, judge_id: Optional[str] = None, model_id: str = _gv.DEFAULT_MODEL_ID, api_key_env_var: str = _gv.DEFAULT_AUTH_ENV_VAR, timeout_s: float = _gv.DEFAULT_TIMEOUT_S, ) -> None: if artifact_modality not in _gv.SUPPORTED_MODALITIES: raise ValueError( f"GeminiVisionEvalNode: unsupported artifact_modality " f"{artifact_modality!r}; expected one of " f"{_gv.SUPPORTED_MODALITIES}" ) self.artifact_modality: str = artifact_modality self.judge_id: str = judge_id or f"eval_{artifact_modality}_v1" self.model_used: str = model_id self._model_id: str = model_id self._api_key_env_var: str = api_key_env_var self._timeout_s: float = timeout_s def evaluate(self, context: EvalContext) -> EvalResult: """Score one artifact via Gemini Vision; translate provider result. The EvalNode Protocol's contract for ``evaluate()`` is "return an EvalResult or raise". This implementation calls :func:`gemini_vision.score_artifact` and: - On success: builds an :class:`EvalResult` with score / reasoning / cost_usd / model_used and the adapter's raw_metadata threaded through. - On any raise from the adapter (incl. ``EvalProviderError`` subclasses): re-raises so the wrapping runner's exception handler translates to a failure-RunResult. Per Phase 4 § 12c, runners mirror :func:`audio_runner._failure_metadata`'s 6-key shape and stamp ``error_class`` for diagnostics. Test injection: ``context.metadata["_transport"]`` (when present) is forwarded into the adapter so unit tests can mock the HTTP call without env-var fiddling. """ transport: Optional[Callable] = None if isinstance(context.metadata, dict): transport = context.metadata.get("_transport") provider_result = _gv.score_artifact( artifact_path=context.target_artifact_path, artifact_modality=self.artifact_modality, prompt=context.rubric, judge_id=self.judge_id, model_id=self._model_id, api_key_env_var=self._api_key_env_var, timeout_s=self._timeout_s, transport=transport, ) # Build the EvalResult metadata from adapter raw_metadata + select # adapter-level fields the panel may want to surface (request_id, # artifact_modality echo). Don't drop adapter warnings. meta: dict[str, Any] = {} raw = getattr(provider_result, "raw_metadata", None) or {} if isinstance(raw, dict): meta.update(raw) # request_id is on the dataclass top level — copy it in for panel # correlation regardless of whether raw_metadata had it. if provider_result.request_id is not None: meta.setdefault("request_id", provider_result.request_id) meta.setdefault("artifact_modality", self.artifact_modality) return EvalResult( score=float(provider_result.score), reasoning=str(provider_result.reasoning), judge_id=self.judge_id, model_used=str(provider_result.model_used or self._model_id), cost_usd=float(provider_result.cost_usd or 0.0), metadata=meta, ) __all__ = ["GeminiVisionEvalNode"]