diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/__init__.py index e3bea664..0183ad97 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/__init__.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/__init__.py @@ -32,11 +32,11 @@ ) from agent_control_evaluator_galileo.records import ( GalileoRecord, + GalileoRecordNormalizer, RecordFactoryError, UnsupportedStepTypeError, build_galileo_record, build_record, - record_from_scorer_invoke_record, record_from_step, ) @@ -51,10 +51,10 @@ "LunaOperator", "LUNA_AVAILABLE", "GalileoRecord", + "GalileoRecordNormalizer", "RecordFactoryError", "UnsupportedStepTypeError", "build_galileo_record", "build_record", - "record_from_scorer_invoke_record", "record_from_step", ] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/client.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/client.py index 45d65333..d6cc72c1 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/client.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/client.py @@ -11,13 +11,14 @@ from hmac import new as hmac_new from json import dumps from time import time -from typing import Literal, cast, get_args +from typing import Any, Literal, get_args from urllib.parse import urlsplit import httpx from agent_control_models import JSONObject, JSONValue, Step -from pydantic import BaseModel, Field, PrivateAttr, model_validator +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator +from ..records import UnsupportedStepTypeError, record_from_step from .config import ScorerInvokeConfig logger = logging.getLogger(__name__) @@ -57,6 +58,7 @@ "session", ] SUPPORTED_SCORER_INVOKE_RECORD_TYPES = frozenset(get_args(ScorerInvokeRecordType)) +_MISSING_SELECTED_DATA = object() def _b64url(data: bytes) -> str: @@ -219,6 +221,8 @@ class ScorerInvokeRecord(BaseModel): represented here. Orbit hydrates those fields from trusted server context. """ + model_config = ConfigDict(extra="allow") + type: ScorerInvokeRecordType name: str | None = None input: JSONValue = None @@ -272,11 +276,13 @@ def to_dict(self) -> JSONObject: return request -def _orbit_record_from_step( +def _scorer_invoke_record_from_step( step: Step | None, *, selected_input: JSONValue, selected_output: JSONValue, + selected_data: Any = _MISSING_SELECTED_DATA, + payload_field: str = "input", ) -> ScorerInvokeRecord | None: """Translate a generic Agent Control step into Orbit's record contract. @@ -297,21 +303,30 @@ def _orbit_record_from_step( Returns: An Orbit-compatible record, or ``None`` for absent/unsupported steps. """ - if step is None or step.type not in SUPPORTED_SCORER_INVOKE_RECORD_TYPES: + if step is None: return None - - # The membership check above narrows the runtime value to Orbit's known - # discriminator set, but static type checkers cannot infer that relationship. - record_type = cast(ScorerInvokeRecordType, step.type) - return ScorerInvokeRecord( - type=record_type, - name=step.name, - input=selected_input if selected_input is not None else step.input, - output=selected_output if selected_output is not None else step.output, - context=step.context, - tools=step.tools, - dataset_output=step.ground_truth, - ) + try: + if selected_data is _MISSING_SELECTED_DATA: + galileo_core_record = record_from_step( + step, + selected_input=selected_input, + selected_output=selected_output, + ) + else: + galileo_core_record = record_from_step( + step, + selected_data=selected_data, + payload_field=payload_field, + ) + except UnsupportedStepTypeError: + return None + # record_from_step() returns Galileo Core models (LlmSpan, Trace, etc.). Dumping + # and revalidating coerces the model into the Luna request shape while preserving + # extra canonical fields that Orbit accepts but ScorerInvokeRecord does not declare. + canonical_record = galileo_core_record.model_dump(mode="json", exclude_none=True) + if canonical_record.get("type") not in SUPPORTED_SCORER_INVOKE_RECORD_TYPES: + return None + return ScorerInvokeRecord.model_validate(canonical_record) class ScorerInvokeResponse(BaseModel): @@ -487,6 +502,8 @@ async def invoke( input: JSONValue = None, output: JSONValue = None, step: Step | None = None, + selected_data: Any = _MISSING_SELECTED_DATA, + selected_data_payload_field: str = "input", execution_context: GalileoExecutionContext | None = None, config: ScorerInvokeConfig | JSONObject | None = None, timeout: float = DEFAULT_TIMEOUT_SECS, @@ -502,6 +519,9 @@ async def invoke( input: Optional user/system prompt text. output: Optional model response text. step: Optional complete runtime step used for structured dual-write. + selected_data: Raw selector-selected value used to construct the record, + preserving its structure independently from serialized legacy inputs. + selected_data_payload_field: Record input/output side for scalar selector values. execution_context: Optional authenticated Galileo scorer context. config: Optional Orbit-supported scorer invocation configuration. timeout: Request timeout in seconds. @@ -531,6 +551,17 @@ async def invoke( invoke_config, http_timeout_seconds=timeout, ) + record = ( + _scorer_invoke_record_from_step( + step, + selected_input=input, + selected_output=output, + selected_data=selected_data, + payload_field=selected_data_payload_field, + ) + if step is not None and scorer_version_id is not None + else None + ) request_body = ScorerInvokeRequest( scorer_id=scorer_id, scorer_version_id=scorer_version_id, @@ -541,11 +572,7 @@ async def invoke( ground_truth=step.ground_truth if step is not None else None, tools=step.tools if step is not None else None, ), - record=_orbit_record_from_step( - step, - selected_input=input, - selected_output=output, - ), + record=record, execution_context=execution_context, config=invoke_config, ).to_dict() diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/config.py index 0bc0fa37..52c315ec 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/config.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/config.py @@ -56,8 +56,8 @@ class LunaEvaluatorConfig(EvaluatorConfig): Attributes: scorer_id: Required scorer identifier for Luna scorer invocation. - scorer_version_id: Deprecated optional compatibility identifier. Orbit - currently invokes the scorer's current default version. + scorer_version_id: Optional for legacy requests; required when the + contextual evaluator sends a structured record. scorer_label: Optional display/metadata label. threshold: Local threshold used by the evaluator for comparison. operator: Local comparison operator. Numeric operators use threshold as a number. @@ -75,8 +75,8 @@ class LunaEvaluatorConfig(EvaluatorConfig): default=None, min_length=1, description=( - "Deprecated optional compatibility identifier. Orbit currently invokes " - "the scorer's current default version." + "Optional for legacy requests and required when contextual evaluation " + "sends a structured record." ), ) scorer_label: str | None = Field( diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/evaluator.py index 14b58d88..a09b4ccf 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/evaluator.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/luna/evaluator.py @@ -281,6 +281,8 @@ async def _evaluate( scorer_kwargs = self._scorer_kwargs() if step is not None: scorer_kwargs["step"] = step + scorer_kwargs["selected_data"] = data + scorer_kwargs["selected_data_payload_field"] = self.config.payload_field if execution_context is not None: scorer_kwargs["execution_context"] = execution_context response = await self._get_client().invoke( diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/__init__.py index 7265e8a1..4260202c 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/__init__.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/__init__.py @@ -6,7 +6,6 @@ UnsupportedStepTypeError, build_galileo_record, build_record, - record_from_scorer_invoke_record, record_from_step, ) from .normalization import GalileoRecordNormalizer @@ -18,6 +17,5 @@ "UnsupportedStepTypeError", "build_galileo_record", "build_record", - "record_from_scorer_invoke_record", "record_from_step", ] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/factory.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/factory.py index d34047f5..75a3e5b6 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/factory.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/factory.py @@ -2,20 +2,26 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence -from typing import Any, cast +from collections.abc import Mapping +from typing import Any from agent_control_models import Step from galileo_core.schemas.logging.session import Session -from galileo_core.schemas.logging.span import LlmSpan, RetrieverSpan, ToolSpan +from galileo_core.schemas.logging.span import ( + AgentSpan, + ControlSpan, + LlmSpan, + RetrieverSpan, + ToolSpan, + WorkflowSpan, +) from galileo_core.schemas.logging.trace import Trace -from pydantic import BaseModel from .normalization import GalileoRecordNormalizer type GalileoRecord = LlmSpan | ToolSpan | RetrieverSpan | Trace | Session type GalileoSpan = LlmSpan | ToolSpan | RetrieverSpan - +type GalileoCoreSpan = AgentSpan | WorkflowSpan | LlmSpan | RetrieverSpan | ToolSpan | ControlSpan _MISSING = object() _SPAN_TYPES = {"llm", "tool", "retriever"} _RECORD_TYPES = _SPAN_TYPES | {"trace", "session"} @@ -30,6 +36,7 @@ class UnsupportedStepTypeError(RecordFactoryError): def _mapping(value: Any) -> Mapping[str, Any] | None: + """Return mapping values unchanged, or ``None`` for non-mapping inputs.""" return value if isinstance(value, Mapping) else None @@ -47,185 +54,13 @@ def _selected_values( return selected_data, None -def _child_payloads(context: Mapping[str, Any], *, key: str) -> list[Any]: - value: Any = context.get(key) - if value is None and key == "spans": - value = context.get("children") - if value is None: - return [] - if not isinstance(value, Sequence) or isinstance(value, str | bytes | bytearray): - raise RecordFactoryError(f"Galileo {key} context must be a list.") - return list(value) - - -def _span_payloads(raw: Mapping[str, Any], context: Mapping[str, Any]) -> list[Any]: - """Read child spans from either a public model dump or step context.""" - if "spans" in raw: - value = raw["spans"] - if not isinstance(value, Sequence) or isinstance(value, str | bytes | bytearray): - raise RecordFactoryError("Galileo span context must include a 'spans' list.") - return list(value) - return _child_payloads(context, key="spans") - - -def _nested_record(value: Any) -> GalileoRecord: - if isinstance(value, (LlmSpan, ToolSpan, RetrieverSpan, Trace, Session)): - return value - raw = _mapping(value) - if raw is None: - raise RecordFactoryError("Nested Galileo records must be objects with a supported type.") - raw_type_value = raw.get("type") - raw_type = ( - raw_type_value - if isinstance(raw_type_value, str) - else getattr(raw_type_value, "value", None) - ) - if not isinstance(raw_type, str): - raise RecordFactoryError("Nested Galileo records require a string 'type'.") - if raw_type not in _RECORD_TYPES: - raise UnsupportedStepTypeError(f"Unsupported Galileo nested step type: {raw_type!r}") - return _record_from_mapping(raw, raw_type) - - -def _nested_span(value: Any) -> GalileoSpan: - record = _nested_record(value) - if not isinstance(record, (LlmSpan, ToolSpan, RetrieverSpan)): - raise RecordFactoryError("Nested span context must contain a span record.") - return record - - -def _record_from_mapping(raw: Mapping[str, Any], record_type: str) -> GalileoRecord: - context = _mapping(raw.get("context")) or {} - common = { - "name": "" if raw.get("name") is None else str(raw["name"]), - "user_metadata": GalileoRecordNormalizer.metadata( - raw.get("user_metadata", raw.get("metadata")) - ), - } - for field in ( - "tags", - "status_code", - "external_id", - "step_number", - "created_at", - "metrics", - "dataset_input", - "dataset_output", - "dataset_metadata", - ): - if raw.get(field) is not None: - common[field] = ( - GalileoRecordNormalizer.json_text(raw[field]) - if field in {"dataset_input", "dataset_output"} - else GalileoRecordNormalizer.metadata(raw[field]) - if field == "dataset_metadata" - else raw[field] - ) - - input_value = raw.get("input") - output_value = raw.get("output") - if record_type == "llm": - kwargs = { - **common, - "input": GalileoRecordNormalizer.llm_input(input_value), - "output": GalileoRecordNormalizer.llm_output(output_value), - } - if raw.get("tools") is not None: - kwargs["tools"] = GalileoRecordNormalizer.llm_tools(raw["tools"]) - for field in ("events", "model", "temperature", "finish_reason"): - if raw.get(field) is not None: - kwargs[field] = raw[field] - if raw.get("redacted_input") is not None: - kwargs["redacted_input"] = GalileoRecordNormalizer.llm_redacted_input( - raw["redacted_input"] - ) - if raw.get("redacted_output") is not None: - kwargs["redacted_output"] = GalileoRecordNormalizer.llm_redacted_output( - raw["redacted_output"] - ) - return LlmSpan(**kwargs) - if record_type == "tool": - kwargs = { - **common, - "input": GalileoRecordNormalizer.tool_input(input_value), - "output": GalileoRecordNormalizer.tool_output(output_value), - } - if raw.get("tool_call_id") is not None: - kwargs["tool_call_id"] = str(raw["tool_call_id"]) - if raw.get("redacted_input") is not None: - kwargs["redacted_input"] = GalileoRecordNormalizer.tool_redacted_input( - raw["redacted_input"] - ) - if raw.get("redacted_output") is not None: - kwargs["redacted_output"] = GalileoRecordNormalizer.tool_redacted_output( - raw["redacted_output"] - ) - kwargs["spans"] = [_nested_span(item) for item in _span_payloads(raw, context)] - return ToolSpan(**kwargs) - if record_type == "retriever": - kwargs = { - **common, - "input": GalileoRecordNormalizer.retriever_input(input_value), - "output": GalileoRecordNormalizer.retriever_output(output_value), - } - if raw.get("redacted_input") is not None: - kwargs["redacted_input"] = GalileoRecordNormalizer.retriever_redacted_input( - raw["redacted_input"] - ) - if raw.get("redacted_output") is not None: - kwargs["redacted_output"] = GalileoRecordNormalizer.retriever_redacted_output( - raw["redacted_output"] - ) - kwargs["spans"] = [_nested_span(item) for item in _span_payloads(raw, context)] - return RetrieverSpan(**kwargs) - if record_type == "trace": - spans = raw.get("spans", context.get("spans", context.get("children"))) - if not isinstance(spans, Sequence) or isinstance(spans, str | bytes | bytearray): - raise RecordFactoryError("Galileo trace context must include a 'spans' list.") - trace_kwargs: dict[str, Any] = { - **common, - "input": GalileoRecordNormalizer.trace_input(input_value), - "output": GalileoRecordNormalizer.trace_output(output_value), - "spans": [_nested_span(item) for item in spans], - } - if raw.get("redacted_input") is not None: - trace_kwargs["redacted_input"] = GalileoRecordNormalizer.trace_redacted_input( - raw["redacted_input"] - ) - if raw.get("redacted_output") is not None: - trace_kwargs["redacted_output"] = GalileoRecordNormalizer.trace_redacted_output( - raw["redacted_output"] - ) - return Trace( - **trace_kwargs, - ) - traces = raw.get("traces", context.get("traces")) - if not isinstance(traces, Sequence) or isinstance(traces, str | bytes | bytearray): - raise RecordFactoryError("Galileo session context must include a 'traces' list.") - nested_records = GalileoRecordNormalizer.session_traces( - traces, - normalize_record=_nested_record, - ) - nested_traces = [item for item in nested_records if isinstance(item, Trace)] - if len(nested_traces) != len(nested_records): - raise RecordFactoryError("Galileo session 'traces' must contain trace records.") - session_kwargs: dict[str, Any] = { - **common, - "input": GalileoRecordNormalizer.session_input(input_value), - "output": GalileoRecordNormalizer.session_output(output_value), - "traces": cast(list[Trace], nested_traces), - } - if raw.get("redacted_input") is not None: - session_kwargs["redacted_input"] = GalileoRecordNormalizer.session_redacted_input( - raw["redacted_input"] - ) - if raw.get("redacted_output") is not None: - session_kwargs["redacted_output"] = GalileoRecordNormalizer.session_redacted_output( - raw["redacted_output"] - ) - return Session( - **session_kwargs, - ) +def _span_children(children: list[GalileoRecord], *, parent_type: str) -> list[GalileoCoreSpan]: + spans: list[GalileoCoreSpan] = [] + for child in children: + if not isinstance(child, (LlmSpan, ToolSpan, RetrieverSpan)): + raise RecordFactoryError(f"Galileo {parent_type} children must be span steps.") + spans.append(child) + return spans def record_from_step( @@ -241,7 +76,7 @@ def record_from_step( The complete step determines the record discriminator. Selector-selected values override only the corresponding input/output fields, preserving the selected evaluator payload while the unselected side comes from ``step``. - Trace and session records require structured child context; scalar selected + Trace and session records require child steps; scalar selected values are never promoted into those record types. The result is a concrete Galileo Core step model (a ``BaseStep`` subclass) that can be JSON-serialized and revalidated using that subtype's schema. The factory doesn't assign @@ -259,26 +94,37 @@ def record_from_step( payload_field=payload_field, ) context = dict(_mapping(step.context) or {}) - step_envelope = _mapping(step.input) - if record_type == "trace" and not context and step_envelope is not None: - if "spans" in step_envelope or "children" in step_envelope: - context = dict(step_envelope) - if record_type == "session" and not context and step_envelope is not None: - if "traces" in step_envelope: - context = dict(step_envelope) + hierarchy_fields = {"children", "spans", "traces"}.intersection(context) + if hierarchy_fields: + fields = ", ".join(sorted(hierarchy_fields)) + raise RecordFactoryError( + f"Step.context cannot define Galileo hierarchy fields ({fields}); " + "use Step.children instead." + ) + children = step.children or [] + allowed_child_types = { + "trace": _SPAN_TYPES, + "session": {"trace"}, + "tool": _SPAN_TYPES, + "retriever": _SPAN_TYPES, + "llm": set(), + }[record_type] + child_records: list[GalileoRecord] = [] + for child in children: + if not isinstance(child, Step): + raise RecordFactoryError("Step.children must contain Agent Control Step objects.") + child_type = child.type.strip().lower() + if child_type not in allowed_child_types: + raise RecordFactoryError( + f"Galileo {record_type} step cannot contain {child_type!r} child steps." + ) + child_records.append(record_from_step(child)) input_value = ( step.input if selected_input is _MISSING or selected_input is None else selected_input ) output_value = ( step.output if selected_output is _MISSING or selected_output is None else selected_output ) - if selected_input is _MISSING or selected_input is None: - if step_envelope is not None and record_type in {"trace", "session"}: - input_value = step_envelope.get("input", step.input) - if selected_output is _MISSING or selected_output is None: - if step_envelope is not None and record_type in {"trace", "session"}: - output_value = step_envelope.get("output", step.output) - if record_type in {"trace", "session"}: selected_value = selected_data if selected_data is _MISSING: @@ -287,52 +133,54 @@ def record_from_step( selected_value, Mapping ): raise RecordFactoryError( - f"Galileo {record_type} records require structured context; " + f"Galileo {record_type} records require structured selector values; " "an untyped scalar selector value is not sufficient." ) - if record_type == "trace" and not ( - "spans" in context or "children" in context or isinstance(context.get("trace"), Mapping) - ): - raise RecordFactoryError("Galileo trace context is missing; provide a 'spans' list.") - if record_type == "session" and not ( - "traces" in context or isinstance(context.get("session"), Mapping) - ): - raise RecordFactoryError("Galileo session context is missing; provide a 'traces' list.") - - source: dict[str, Any] = { - "type": record_type, + common: dict[str, Any] = { "name": step.name, - "input": input_value, - "output": output_value, - "context": context, - "tools": step.tools, + "user_metadata": GalileoRecordNormalizer.metadata(context.get("metadata")), } - if record_type == "trace" and isinstance(context.get("trace"), Mapping): - source.update(context["trace"]) - if record_type == "session" and isinstance(context.get("session"), Mapping): - source.update(context["session"]) - if selected_input is not _MISSING and selected_input is not None: - source["input"] = input_value - if selected_output is not _MISSING and selected_output is not None: - source["output"] = output_value if step.ground_truth is not None: - source["dataset_output"] = step.ground_truth - source["metadata"] = context.get("metadata") - return _record_from_mapping(source, record_type) - - -def record_from_scorer_invoke_record(record: BaseModel) -> GalileoRecord: - """Convert the legacy Pydantic Luna request record to a canonical record. - - The Luna wire contract intentionally remains owned by - :class:`ScorerInvokeRecord`. This adapter accepts that existing request - model at the boundary and uses the published SDK models internally. - """ - raw = record.model_dump(mode="python", exclude_none=True) - record_type = raw.get("type") - if not isinstance(record_type, str) or record_type not in _RECORD_TYPES: - raise UnsupportedStepTypeError(f"Unsupported Galileo record type: {record_type!r}") - return _record_from_mapping(raw, record_type) + common["dataset_output"] = GalileoRecordNormalizer.json_text(step.ground_truth) + if record_type == "llm": + kwargs = { + **common, + "input": GalileoRecordNormalizer.llm_input(input_value), + "output": GalileoRecordNormalizer.llm_output(output_value), + } + if step.tools is not None: + kwargs["tools"] = GalileoRecordNormalizer.llm_tools(step.tools) + return LlmSpan(**kwargs) + if record_type == "tool": + return ToolSpan( + **common, + input=GalileoRecordNormalizer.tool_input(input_value), + output=GalileoRecordNormalizer.tool_output(output_value), + spans=_span_children(child_records, parent_type=record_type), + ) + if record_type == "retriever": + return RetrieverSpan( + **common, + input=GalileoRecordNormalizer.retriever_input(input_value), + output=GalileoRecordNormalizer.retriever_output(output_value), + spans=_span_children(child_records, parent_type=record_type), + ) + if record_type == "trace": + return Trace( + **common, + input=GalileoRecordNormalizer.trace_input(input_value), + output=GalileoRecordNormalizer.trace_output(output_value), + spans=_span_children(child_records, parent_type=record_type), + ) + traces = [child for child in child_records if isinstance(child, Trace)] + if len(traces) != len(child_records): + raise RecordFactoryError("Galileo session children must be trace steps.") + return Session( + **common, + input=GalileoRecordNormalizer.session_input(input_value), + output=GalileoRecordNormalizer.session_output(output_value), + traces=traces, + ) def build_record(selected_data: Any, step: Step, *, payload_field: str = "input") -> GalileoRecord: diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/normalization.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/normalization.py index 5144cbbb..53f118bf 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/normalization.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/records/normalization.py @@ -2,8 +2,9 @@ from __future__ import annotations +import ast import json -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Mapping, Sequence from dataclasses import asdict, is_dataclass from datetime import UTC, date, datetime from enum import Enum @@ -180,6 +181,36 @@ def _message_sequence(value: Any) -> bool: ) +def _llm_message_sequence(value: Any) -> Sequence[Any] | None: + """Extract a role-bearing LLM message list from structured or string input.""" + if isinstance(value, str): + try: + parsed = json.loads(value) + except (RecursionError, ValueError): + try: + parsed = ast.literal_eval(value) + except (SyntaxError, ValueError, TypeError, RecursionError): + return None + if isinstance(parsed, Mapping): + candidates = parsed.get("messages") + if not isinstance(candidates, list): + return None + elif isinstance(parsed, list): + candidates = parsed + else: + return None + else: + candidates = value.get("messages") if isinstance(value, Mapping) else value + if not isinstance(candidates, Sequence) or isinstance(candidates, str | bytes | bytearray): + return None + if not all( + isinstance(item, Message) or (isinstance(item, Mapping) and "role" in item) + for item in candidates + ): + return None + return candidates + + def _flatten_message_sequence(value: Sequence[Any]) -> list[dict[str, Any]]: blocks: list[dict[str, Any]] = [] for message in value: @@ -246,11 +277,18 @@ def llm_input(cls, value: Any) -> Any: if value is None: return "" if isinstance(value, str | Message): - return value + messages = _llm_message_sequence(value) + if messages is None: + return value + return [_json_compatible(item) for item in messages] if isinstance(value, Mapping): + messages = _llm_message_sequence(value) + if messages is not None: + return [_json_compatible(item) for item in messages] return _json_compatible(value) if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray): - if all(isinstance(item, Message | Mapping) for item in value): + messages = _llm_message_sequence(value) + if messages is not None: return [_json_compatible(item) for item in value] return cls.json_text(value) @@ -300,7 +338,11 @@ def tool_redacted_output(cls, value: Any) -> str | None: @classmethod def retriever_input(cls, value: Any) -> str: - return "" if value is None else cls.json_text(value) + if value is None: + return "" + if isinstance(value, Mapping) and "query" in value: + return cls.json_text(value["query"]) + return cls.json_text(value) @staticmethod def retriever_output(value: Any) -> list[Document]: @@ -459,12 +501,3 @@ def session_redacted_input(cls, value: Any) -> Any: @classmethod def session_redacted_output(cls, value: Any) -> Any: return None if value is None else cls.session_output(value) - - @staticmethod - def session_traces( - values: Sequence[Any], - *, - normalize_record: Callable[[Any], Any], - ) -> list[Any]: - """Normalize every nested session trace through the shared record path.""" - return [normalize_record(value) for value in values] diff --git a/evaluators/contrib/galileo/tests/test_luna_evaluator.py b/evaluators/contrib/galileo/tests/test_luna_evaluator.py index cf3d5423..3d986216 100644 --- a/evaluators/contrib/galileo/tests/test_luna_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_luna_evaluator.py @@ -212,6 +212,8 @@ def test_scorer_invoke_record_preserves_legacy_pydantic_contract(self) -> None: context={"session_id": "session-1"}, tools=[{"name": "search"}], dataset_output={"expected": "world"}, + metrics={"duration_ns": 12}, + custom_field={"value": "preserved"}, ) assert record.model_dump() == { @@ -222,6 +224,8 @@ def test_scorer_invoke_record_preserves_legacy_pydantic_contract(self) -> None: "context": {"session_id": "session-1"}, "tools": [{"name": "search"}], "dataset_output": {"expected": "world"}, + "metrics": {"duration_ns": 12}, + "custom_field": {"value": "preserved"}, } with pytest.raises(ValidationError, match="type"): ScorerInvokeRecord(type="unsupported", name="answer") @@ -777,6 +781,7 @@ def handler(_request: httpx.Request) -> httpx.Response: @pytest.mark.asyncio async def test_client_dual_writes_legacy_inputs_and_structured_record(self) -> None: from agent_control_evaluator_galileo.luna import GalileoLunaClient + from agent_control_evaluator_galileo.records import record_from_step captured: dict[str, object] = {} @@ -800,7 +805,7 @@ def handler(request: httpx.Request) -> httpx.Response: name="answer", input={"messages": [{"role": "user", "content": "question"}]}, output={"text": "answer"}, - context={"session": {"id": "s-1", "attributes": {"region": "west"}}}, + context={"metadata": {"source": "runtime"}, "request_id": "r-1"}, tools=[{"name": "search", "description": "Search", "input_schema": {}}], ground_truth={"text": "expected"}, ) @@ -821,7 +826,7 @@ def handler(request: httpx.Request) -> httpx.Response: finally: await client.close() - # Then: old inputs and an equivalent Orbit record are sent together + # Then: legacy inputs and the shared factory's canonical record are sent together assert response.score == 0.9 assert response.scorer_label == "toxicity" assert response.status == "success" @@ -837,30 +842,36 @@ def handler(request: httpx.Request) -> httpx.Response: "ground_truth": {"text": "expected"}, "tools": [{"name": "search", "description": "Search", "input_schema": {}}], }, - "record": { - "type": "llm", - "name": "answer", - "input": "selected question", - "output": "selected answer", - "context": {"session": {"id": "s-1", "attributes": {"region": "west"}}}, - "tools": [{"name": "search", "description": "Search", "input_schema": {}}], - "dataset_output": {"text": "expected"}, - }, + "record": record_from_step( + step, + selected_input="selected question", + selected_output="selected answer", + ).model_dump(mode="json", exclude_none=True), "config": {"request_timeout_seconds": 8.0}, } - assert captured["body"] == expected_body + actual_body = captured["body"] + assert isinstance(actual_body, dict) + actual_record = actual_body["record"] + expected_record = expected_body["record"] + assert isinstance(actual_record, dict) + assert isinstance(expected_record, dict) + # Each factory call assigns a fresh created_at timestamp. + actual_record.pop("created_at", None) + expected_record.pop("created_at", None) + assert actual_body == expected_body # The emitted body also satisfies the exact additive Orbit #1720 shape. orbit_request = _Orbit1720Request.model_validate(expected_body) assert orbit_request.record is not None - assert orbit_request.inputs.query == orbit_request.record.input - assert orbit_request.inputs.response == orbit_request.record.output + assert orbit_request.inputs.query == "selected question" + assert orbit_request.inputs.response == "selected answer" + assert "selected question" in json.dumps(orbit_request.record.input) + assert "selected answer" in json.dumps(orbit_request.record.output) assert orbit_request.inputs.tools == orbit_request.record.tools - assert orbit_request.record.dataset_output == {"text": "expected"} + assert json.loads(orbit_request.record.dataset_output) == {"text": "expected"} assert orbit_request.inputs.ground_truth == {"text": "expected"} - assert orbit_request.record.context == { - "session": {"id": "s-1", "attributes": {"region": "west"}} - } + assert orbit_request.record.user_metadata == {"source": "runtime"} + assert "context" not in expected_body["record"] assert orbit_request.config is not None assert orbit_request.config.request_timeout_seconds == 8.0 @@ -874,6 +885,92 @@ def handler(request: httpx.Request) -> httpx.Response: "config": {"request_timeout_seconds": 8.0}, } + @pytest.mark.asyncio + async def test_client_omits_record_when_step_has_no_pinned_version(self) -> None: + from agent_control_evaluator_galileo.luna import GalileoLunaClient + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.5, "status": "success"}) + + step = Step( + type="llm", + name="answer", + input="complete prompt", + output="complete response", + tools=[{"name": "search", "input_schema": {}}], + ground_truth="expected answer", + ) + with patch.dict(os.environ, LUNA_ENV, clear=True): + client = GalileoLunaClient() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + await client.invoke( + scorer_id=SCORER_ID, + input="selected prompt", + output="selected response", + step=step, + ) + finally: + await client.close() + + assert captured["body"] == { + "scorer_id": SCORER_ID, + "inputs": { + "query": "selected prompt", + "response": "selected response", + "ground_truth": "expected answer", + "tools": [{"name": "search", "input_schema": {}}], + }, + "config": {"request_timeout_seconds": 8.0}, + } + + @patch.dict(os.environ, LUNA_ENV) + @pytest.mark.asyncio + async def test_evaluator_normalizes_record_without_changing_legacy_inputs(self) -> None: + from agent_control_evaluator_galileo.luna import LunaEvaluator + + captured: dict[str, object] = {} + raw_messages = "{'messages': [{'role': 'user', 'content': 'wire policy'}]}" + raw_response = "{'answer': 'approved'}" + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.9, "status": "success"}) + + step = Step( + type="llm", + name="answer", + input=raw_messages, + output=raw_response, + ) + evaluator = LunaEvaluator.from_dict( + {"scorer_id": SCORER_ID, "scorer_version_id": SCORER_VERSION_ID} + ) + evaluator._client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + result = await evaluator.evaluate_with_context( + {"input": raw_messages, "output": raw_response}, + step, + ) + finally: + await evaluator._client.close() + + assert result.error is None + body = captured["body"] + assert isinstance(body, dict) + assert body["inputs"]["query"] == raw_messages + assert body["inputs"]["response"] == raw_response + assert body["scorer_version_id"] == SCORER_VERSION_ID + record = body["record"] + assert isinstance(record, dict) + assert record["input"] == [{"role": "user", "content": "wire policy"}] + assert record["output"] == {"content": raw_response, "role": "assistant"} + @pytest.mark.asyncio async def test_client_omits_record_for_step_type_orbit_does_not_support(self) -> None: from agent_control_evaluator_galileo.luna import GalileoLunaClient @@ -903,6 +1000,390 @@ def handler(request: httpx.Request) -> httpx.Response: "config": {"request_timeout_seconds": 8.0}, } + @pytest.mark.parametrize( + ("step", "record_type"), + [ + (Step(type="llm", name="answer", input="question", output="answer"), "llm"), + (Step(type="tool", name="search", input={"query": "q"}, output="result"), "tool"), + ( + Step(type="retriever", name="retrieve", input="question", output=["document"]), + "retriever", + ), + ], + ) + @pytest.mark.asyncio + async def test_client_uses_canonical_factory_for_flat_steps( + self, + step: Step, + record_type: str, + ) -> None: + from agent_control_evaluator_galileo.luna import GalileoLunaClient + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.5, "status": "success"}) + + with patch.dict(os.environ, LUNA_ENV, clear=True): + client = GalileoLunaClient() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + await client.invoke( + scorer_id="scorer-123", + scorer_version_id=SCORER_VERSION_ID, + input="selected input", + output="selected output", + step=step, + ) + finally: + await client.close() + + body = captured["body"] + assert isinstance(body, dict) + record = body["record"] + assert isinstance(record, dict) + assert record["type"] == record_type + assert "context" not in record + if record_type in {"tool", "retriever"}: + assert record["spans"] == [] + + @pytest.mark.asyncio + async def test_client_serializes_nested_session_trace_and_tool_spans(self) -> None: + from agent_control_evaluator_galileo.luna import GalileoLunaClient + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.5, "status": "success"}) + + step = Step( + type="session", + name="conversation", + input="session input", + children=[ + Step( + type="trace", + name="request", + input="trace input", + children=[ + Step( + type="tool", + name="search", + input={"query": "q"}, + output="result", + children=[Step(type="tool", name="nested", input={"id": 1})], + ) + ], + ) + ], + ) + with patch.dict(os.environ, LUNA_ENV, clear=True): + client = GalileoLunaClient() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + await client.invoke( + scorer_id="scorer-123", + scorer_version_id=SCORER_VERSION_ID, + input={"query": "selected"}, + output={"answer": "selected"}, + step=step, + ) + finally: + await client.close() + + body = captured["body"] + assert isinstance(body, dict) + record = body["record"] + assert isinstance(record, dict) + assert record["type"] == "session" + traces = record["traces"] + assert isinstance(traces, list) + trace = traces[0] + assert isinstance(trace, dict) + assert trace["type"] == "trace" + spans = trace["spans"] + assert isinstance(spans, list) + tool = spans[0] + assert isinstance(tool, dict) + assert tool["type"] == "tool" + nested_spans = tool["spans"] + assert isinstance(nested_spans, list) + assert nested_spans[0]["name"] == "nested" + + @patch.dict(os.environ, LUNA_ENV) + @pytest.mark.asyncio + async def test_evaluator_posts_trace_with_raw_structured_selector_data(self) -> None: + from agent_control_evaluator_galileo.luna import LunaEvaluator + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.8, "status": "success"}) + + step = Step( + type="trace", + name="account_lookup_trace", + input={"messages": [{"role": "user", "content": "Find my account information."}]}, + output={"content": "Your account belongs to Jane Doe."}, + children=[ + Step( + type="llm", + name="account_agent", + input=[{"role": "user", "content": "Find my account information."}], + output={"content": "Looking up the account."}, + ), + Step( + type="tool", + name="lookup_account", + input={"account_id": "123"}, + output={"name": "Jane Doe"}, + children=[ + Step( + type="tool", + name="account_service", + input={"account_id": "123"}, + output={"name": "Jane Doe"}, + ) + ], + ), + Step( + type="retriever", + name="account_docs", + input={"query": "account 123"}, + output=[{"content": "Account owner: Jane Doe", "metadata": {"id": "d1"}}], + ), + ], + tools=[{"name": "lookup_account", "description": "Lookup", "input_schema": {}}], + ground_truth={"owner": "Jane Doe"}, + ) + selected = step.model_dump(mode="json") + evaluator = LunaEvaluator.from_dict( + { + "scorer_id": SCORER_ID, + "scorer_version_id": SCORER_VERSION_ID, + "scorer_label": "account-check", + "threshold": 0.5, + "config": {"request_timeout_seconds": 6}, + "timeout_ms": 9000, + } + ) + evaluator._client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + result = await evaluator.evaluate_with_extensions( + selected, + step, + { + "namespace_key": "org-1", + "caller_id": "user-2", + "target_type": "log_stream", + "target_id": "run-3", + "metadata": {"project_id": "project-4"}, + }, + ) + finally: + await evaluator._client.close() + + assert result.error is None + body = captured["body"] + assert isinstance(body, dict) + assert body["scorer_id"] == SCORER_ID + assert body["scorer_version_id"] == SCORER_VERSION_ID + assert body["scorer_label"] == "account-check" + assert body["execution_context"] == { + "organization_id": "org-1", + "user_id": "user-2", + "project_id": "project-4", + "run_id": "run-3", + } + assert body["inputs"] == { + "query": json.dumps(step.input, ensure_ascii=False, sort_keys=True), + "response": json.dumps(step.output, ensure_ascii=False, sort_keys=True), + "ground_truth": {"owner": "Jane Doe"}, + "tools": step.tools, + } + assert body["config"] == {"request_timeout_seconds": 6.0} + record = body["record"] + assert isinstance(record, dict) + assert record["type"] == "trace" + assert json.loads(record["input"]) == step.input + assert json.loads(record["output"]) == step.output + spans = record["spans"] + assert isinstance(spans, list) + assert [child["type"] for child in spans] == ["llm", "tool", "retriever"] + assert spans[0]["input"][0]["content"] == "Find my account information." + assert json.loads(spans[1]["input"]) == {"account_id": "123"} + assert spans[1]["spans"][0]["name"] == "account_service" + assert spans[2]["output"][0]["content"] == "Account owner: Jane Doe" + + @patch.dict(os.environ, LUNA_ENV) + @pytest.mark.asyncio + async def test_evaluator_posts_session_with_raw_structured_selector_data(self) -> None: + from agent_control_evaluator_galileo.luna import LunaEvaluator + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.8, "status": "success"}) + + step = Step( + type="session", + name="support_session", + input={"customer": "Jane Doe"}, + output={"resolution": "account found"}, + children=[ + Step( + type="trace", + name="account_lookup", + input={"query": "account info"}, + output={"owner": "Jane Doe"}, + children=[ + Step( + type="tool", + name="lookup_account", + input={"account_id": "123"}, + output={"name": "Jane Doe"}, + ) + ], + ) + ], + ) + evaluator = LunaEvaluator.from_dict( + {"scorer_id": SCORER_ID, "scorer_version_id": SCORER_VERSION_ID} + ) + evaluator._client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + result = await evaluator.evaluate_with_context(step.model_dump(mode="json"), step) + finally: + await evaluator._client.close() + + assert result.error is None + body = captured["body"] + assert isinstance(body, dict) + assert body["inputs"]["query"] == json.dumps( + step.input, ensure_ascii=False, sort_keys=True + ) + assert body["inputs"]["response"] == json.dumps( + step.output, ensure_ascii=False, sort_keys=True + ) + record = body["record"] + assert isinstance(record, dict) + assert record["type"] == "session" + assert json.loads(record["input"]) == step.input + assert json.loads(record["output"]) == step.output + traces = record["traces"] + assert isinstance(traces, list) + assert traces[0]["type"] == "trace" + assert traces[0]["spans"][0]["type"] == "tool" + + @patch.dict(os.environ, LUNA_ENV) + @pytest.mark.asyncio + async def test_evaluator_does_not_post_trace_for_scalar_selector_or_invalid_hierarchy( + self, + ) -> None: + from agent_control_evaluator_galileo.luna import LunaEvaluator + + request_count = 0 + + def handler(_request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response(200, json={"score": 0.8, "status": "success"}) + + evaluator = LunaEvaluator.from_dict( + {"scorer_id": SCORER_ID, "scorer_version_id": SCORER_VERSION_ID} + ) + evaluator._client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + trace = Step( + type="trace", + name="trace", + input={"question": "q"}, + children=[Step(type="llm", name="answer", input="q", output="a")], + ) + invalid_session = Step( + type="session", + name="session", + input="q", + children=[Step(type="tool", name="invalid-child", input={"q": "q"})], + ) + + try: + scalar_result = await evaluator.evaluate_with_context("response text", trace) + hierarchy_result = await evaluator.evaluate_with_context( + invalid_session.model_dump(mode="json"), invalid_session + ) + finally: + await evaluator._client.close() + + assert scalar_result.error is not None + assert "structured selector values" in scalar_result.error + assert hierarchy_result.error is not None + assert "cannot contain 'tool' child steps" in hierarchy_result.error + assert request_count == 0 + + @pytest.mark.parametrize("field", ["children", "spans", "traces"]) + @pytest.mark.asyncio + async def test_client_rejects_hierarchy_fields_from_step_context(self, field: str) -> None: + from agent_control_evaluator_galileo.luna import GalileoLunaClient + from agent_control_evaluator_galileo.records import RecordFactoryError + + step = Step( + type="tool", + name="search", + input={"query": "q"}, + context={field: []}, + ) + with patch.dict(os.environ, LUNA_ENV, clear=True): + client = GalileoLunaClient() + client._client = httpx.AsyncClient( + transport=httpx.MockTransport( + lambda request: httpx.Response(200, json={"score": 0.5, "status": "success"}) + ) + ) + + try: + with pytest.raises(RecordFactoryError, match="Step.context cannot define"): + await client.invoke( + scorer_id="scorer-123", + scorer_version_id=SCORER_VERSION_ID, + input="selected", + step=step, + ) + finally: + await client.close() + + @pytest.mark.asyncio + async def test_client_uses_legacy_inputs_when_step_is_missing(self) -> None: + from agent_control_evaluator_galileo.luna import GalileoLunaClient + + captured: dict[str, object] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["body"] = json.loads(request.content.decode()) + return httpx.Response(200, json={"score": 0.5, "status": "success"}) + + with patch.dict(os.environ, LUNA_ENV, clear=True): + client = GalileoLunaClient() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + try: + await client.invoke(scorer_id="scorer-123", input="legacy query") + finally: + await client.close() + + assert captured["body"] == { + "scorer_id": "scorer-123", + "inputs": {"query": "legacy query", "response": ""}, + "config": {"request_timeout_seconds": 8.0}, + } + @pytest.mark.asyncio async def test_client_forwards_scorer_version_id_when_configured(self) -> None: from agent_control_evaluator_galileo.luna import GalileoLunaClient @@ -1073,6 +1554,8 @@ async def test_evaluator_contextual_hook_forwards_complete_step(self) -> None: mock_invoke.assert_awaited_once_with( scorer_id="scorer-123", step=step, + selected_data="selected input", + selected_data_payload_field="input", input="selected input", output=None, config=None, @@ -1110,6 +1593,8 @@ async def test_evaluator_consumes_verified_opaque_extensions(self, target_type: mock_invoke.assert_awaited_once_with( scorer_id="scorer-123", step=Step(type="llm", name="answer", input="prompt"), + selected_data="selected input", + selected_data_payload_field="input", execution_context=GalileoExecutionContext( organization_id="org-1", user_id="verified-user-2", @@ -1149,6 +1634,8 @@ async def test_evaluator_allows_missing_caller_id(self) -> None: mock_invoke.assert_awaited_once_with( scorer_id="scorer-123", step=Step(type="llm", name="answer", input="prompt"), + selected_data="selected input", + selected_data_payload_field="input", execution_context=GalileoExecutionContext( organization_id="org-1", user_id=None, @@ -1189,6 +1676,8 @@ async def test_evaluator_omits_run_id_for_unsupported_target_type(self) -> None: mock_invoke.assert_awaited_once_with( scorer_id="scorer-123", step=Step(type="llm", name="answer", input="prompt"), + selected_data="selected input", + selected_data_payload_field="input", execution_context=GalileoExecutionContext( organization_id="org-1", user_id="verified-user-2", diff --git a/evaluators/contrib/galileo/tests/test_records_factory.py b/evaluators/contrib/galileo/tests/test_records_factory.py index f47f4c91..61f0be74 100644 --- a/evaluators/contrib/galileo/tests/test_records_factory.py +++ b/evaluators/contrib/galileo/tests/test_records_factory.py @@ -16,7 +16,6 @@ UnsupportedStepTypeError, build_galileo_record, build_record, - record_from_scorer_invoke_record, record_from_step, ) from agent_control_models import Step @@ -28,22 +27,7 @@ from galileo_core.schemas.shared.content_parts import FileContentPart, TextContentPart from galileo_core.schemas.shared.document import Document from galileo_core.schemas.shared.records import BaseRecord, RecordTypeAdapter -from pydantic import BaseModel, ConfigDict, Field - - -class _RecordPayload(BaseModel): - """Flexible Pydantic boundary model used to exercise the adapter contract.""" - - model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") - - type: str - name: str | None = None - input: object | None = None - output: object | None = None - context: object | None = None - spans: object | None = None - traces: object | None = None - tool_call_id: object | None = None +from pydantic import BaseModel, Field def test_required_galileo_core_public_exports_are_importable() -> None: @@ -64,7 +48,7 @@ def test_factory_records_are_serializable_concrete_base_steps() -> None: name="request", input="question", output="answer", - context={"spans": [{"type": "llm", "name": "answer", "input": "question"}]}, + children=[Step(type="llm", name="answer", input="question")], ) ), record_from_step( @@ -72,16 +56,14 @@ def test_factory_records_are_serializable_concrete_base_steps() -> None: type="session", name="conversation", input="question", - context={ - "traces": [ - { - "type": "trace", - "name": "request", - "input": "question", - "spans": [{"type": "llm", "name": "answer", "input": "question"}], - } - ] - }, + children=[ + Step( + type="trace", + name="request", + input="question", + children=[Step(type="llm", name="answer", input="question")], + ) + ], ) ), ] @@ -117,10 +99,10 @@ def test_factory_steps_convert_to_core_records_after_execution_ids_are_added() - Step(type="retriever", name="retrieve", input="question", output=["document"]) ), record_from_step( - Step(type="trace", name="request", input="question", context={"spans": []}) + Step(type="trace", name="request", input="question", children=[]) ), record_from_step( - Step(type="session", name="conversation", input="question", context={"traces": []}) + Step(type="session", name="conversation", input="question", children=[]) ), ] project_id = uuid4() @@ -182,6 +164,60 @@ def test_llm_messages_use_public_canonical_validation() -> None: assert record.dataset_output == json.dumps({"expected": "answer"}) +@pytest.mark.parametrize( + ("source", "expected_messages"), + [ + ("A plain prompt", None), + ( + '[{"role": "user", "content": "question"}]', + [{"role": "user", "content": "question"}], + ), + ( + "{'messages': [{'role': 'user', 'content': 'question'}]}", + [{"role": "user", "content": "question"}], + ), + ("{'messages': [", None), + ('{"content": "document without a role"}', None), + ('[{"content": "document without a role"}]', None), + ], +) +def test_llm_string_input_only_parses_role_bearing_message_shapes( + source: str, + expected_messages: list[dict[str, str]] | None, +) -> None: + record = record_from_step(Step(type="llm", name="answer", input=source)) + + assert isinstance(record, LlmSpan) + if expected_messages is None: + assert len(record.input) == 1 + assert record.input[0].content == source + else: + assert [ + {"role": message.role.value, "content": message.content} + for message in record.input + ] == expected_messages + + restored = LlmSpan.model_validate(record.model_dump(mode="json", exclude_none=True)) + assert [message.content for message in restored.input] == [ + message.content for message in record.input + ] + + +def test_llm_mapping_with_messages_key_becomes_public_message_models() -> None: + record = record_from_step( + Step( + type="llm", + name="answer", + input={"messages": [{"role": "user", "content": "question"}]}, + ) + ) + + assert isinstance(record, LlmSpan) + assert [message.role.value for message in record.input] == ["user"] + restored = LlmSpan.model_validate(record.model_dump(mode="json", exclude_none=True)) + assert [message.content for message in restored.input] == ["question"] + + @pytest.mark.parametrize( ("output", "expected"), [ @@ -314,17 +350,10 @@ def test_retriever_results_become_documents_with_scalar_metadata_only() -> None: def test_retriever_accepts_the_public_galileo_core_document_model() -> None: - from pydantic import BaseModel - document = Document(content="context", metadata={"source": "kb"}) - - class RecordInput(BaseModel): - type: str = "retriever" - name: str = "retrieve" - input: str = "question" - output: object - - record = record_from_scorer_invoke_record(RecordInput(output=document)) + record = RetrieverSpan.model_validate( + {"name": "retrieve", "input": "question", "output": [document]} + ) assert isinstance(record, RetrieverSpan) assert record.output[0].content == "context" @@ -337,12 +366,11 @@ def test_public_document_metadata_variants_are_supported() -> None: {"content": "model metadata", "metadata": {"source": "kb"}} ) - record = record_from_scorer_invoke_record( - _RecordPayload( - type="retriever", - input="question", - output=[document_without_metadata, document_with_model_metadata], - ) + record = RetrieverSpan.model_validate( + { + "input": "question", + "output": [document_without_metadata, document_with_model_metadata], + } ) assert isinstance(record, RetrieverSpan) @@ -358,17 +386,15 @@ def test_nested_trace_and_session_records_are_preserved() -> None: type="trace", name="request", input="question", - context={ - "spans": [ - { - "type": "tool", - "name": "search", - "input": {"q": "x"}, - "spans": [{"type": "tool", "name": "nested", "input": {"q": "y"}}], - }, - {"type": "retriever", "name": "retrieve", "input": "x", "output": []}, - ] - }, + children=[ + Step( + type="tool", + name="search", + input={"q": "x"}, + children=[Step(type="tool", name="nested", input={"q": "y"})], + ), + Step(type="retriever", name="retrieve", input="x", output=[]), + ], ) ) session = record_from_step( @@ -376,7 +402,12 @@ def test_nested_trace_and_session_records_are_preserved() -> None: type="session", name="conversation", input="question", - context={"traces": [trace.model_dump(mode="json")]}, + children=[Step(type="trace", name="request", input="question", children=[ + Step(type="tool", name="search", input={"q": "x"}, children=[ + Step(type="tool", name="nested", input={"q": "y"}) + ]), + Step(type="retriever", name="retrieve", input="x", output=[]), + ])], ) ) @@ -438,20 +469,14 @@ def test_public_model_dump_preserves_canonical_fields() -> None: traces=[], ) - rebuilt_llm = record_from_scorer_invoke_record( - _RecordPayload(**llm_source.model_dump(mode="json", exclude_none=True)) - ) - rebuilt_tool = record_from_scorer_invoke_record( - _RecordPayload(**tool_source.model_dump(mode="json", exclude_none=True)) + rebuilt_llm = LlmSpan.model_validate(llm_source.model_dump(mode="json", exclude_none=True)) + rebuilt_tool = ToolSpan.model_validate(tool_source.model_dump(mode="json", exclude_none=True)) + rebuilt_retriever = RetrieverSpan.model_validate( + retriever_source.model_dump(mode="json", exclude_none=True) ) - rebuilt_retriever = record_from_scorer_invoke_record( - _RecordPayload(**retriever_source.model_dump(mode="json", exclude_none=True)) - ) - rebuilt_trace = record_from_scorer_invoke_record( - _RecordPayload(**trace_source.model_dump(mode="json", exclude_none=True)) - ) - rebuilt_session = record_from_scorer_invoke_record( - _RecordPayload(**session_source.model_dump(mode="json", exclude_none=True)) + rebuilt_trace = Trace.model_validate(trace_source.model_dump(mode="json", exclude_none=True)) + rebuilt_session = Session.model_validate( + session_source.model_dump(mode="json", exclude_none=True) ) assert isinstance(rebuilt_llm, LlmSpan) @@ -491,9 +516,7 @@ def test_trace_public_content_parts_are_not_reserialized() -> None: spans=[], ) - rebuilt = record_from_scorer_invoke_record( - _RecordPayload(**source.model_dump(mode="json", exclude_none=True)) - ) + rebuilt = Trace.model_validate(source.model_dump(mode="json", exclude_none=True)) assert isinstance(rebuilt, Trace) assert rebuilt.model_dump(mode="json", exclude_none=True) == source.model_dump( @@ -505,9 +528,7 @@ def test_trace_public_content_parts_are_not_reserialized() -> None: output=[{"type": "file", "file_id": str(uuid4())}], spans=[], ) - file_rebuilt = record_from_scorer_invoke_record( - _RecordPayload(**file_source.model_dump(mode="json", exclude_none=True)) - ) + file_rebuilt = Trace.model_validate(file_source.model_dump(mode="json", exclude_none=True)) assert isinstance(file_rebuilt, Trace) assert file_rebuilt.model_dump(mode="json", exclude_none=True) == file_source.model_dump( mode="json", exclude_none=True @@ -515,81 +536,41 @@ def test_trace_public_content_parts_are_not_reserialized() -> None: def test_malformed_trace_content_parts_follow_sdk_serialization() -> None: - scalar_item = record_from_scorer_invoke_record( - _RecordPayload(type="trace", input="input", output=[1], spans=[]) - ) - invalid_file = record_from_scorer_invoke_record( - _RecordPayload( - type="trace", - input="input", - output=[{"type": "file", "file_id": "not-a-uuid"}], - spans=[], + from pydantic import ValidationError + + with pytest.raises(ValidationError): + Trace.model_validate({"input": "input", "output": [1], "spans": []}) + with pytest.raises(ValidationError): + Trace.model_validate( + { + "input": "input", + "output": [{"type": "file", "file_id": "not-a-uuid"}], + "spans": [], + } ) - ) - assert isinstance(scalar_item, Trace) - assert isinstance(invalid_file, Trace) - assert scalar_item.output == json.dumps([1]) - assert invalid_file.output == json.dumps([{"type": "file", "file_id": "not-a-uuid"}]) - -def test_nested_record_errors_and_existing_models_are_explicit() -> None: - from agent_control_evaluator_galileo.records.factory import _record_from_mapping - - existing_child = ToolSpan(name="existing", input="input") - with pytest.raises(RecordFactoryError, match="objects"): - record_from_scorer_invoke_record( - _RecordPayload(type="tool", input="input", spans=[1]) - ) - with pytest.raises(RecordFactoryError, match="string 'type'"): - record_from_scorer_invoke_record( - _RecordPayload(type="tool", input="input", spans=[{}]) - ) - with pytest.raises(UnsupportedStepTypeError, match="custom"): - record_from_scorer_invoke_record( - _RecordPayload(type="tool", input="input", spans=[{"type": "custom"}]) - ) - with pytest.raises(RecordFactoryError, match="span record"): - record_from_scorer_invoke_record( - _RecordPayload( - type="tool", - input="input", - spans=[{"type": "trace", "input": "input", "spans": []}], +def test_context_hierarchy_fields_are_rejected_and_metadata_remains_supported() -> None: + for hierarchy_field in ("children", "spans", "traces"): + with pytest.raises(RecordFactoryError, match="Step.context cannot define.*Step.children"): + record_from_step( + Step( + type="tool", + name="tool", + input={"q": "input"}, + context={hierarchy_field: []}, + ) ) + metadata_record = record_from_step( + Step( + type="tool", + name="tool", + input={"q": "input"}, + context={"metadata": {"provider": "test"}, "request_id": "abc"}, ) - - record = _record_from_mapping( - {"type": "tool", "input": "input", "spans": [existing_child]}, "tool" - ) - assert isinstance(record, ToolSpan) - assert record.spans[0].name == existing_child.name - - -def test_invalid_nested_context_shapes_are_rejected() -> None: - valid_context_record = record_from_step( - Step(type="tool", name="tool", input={"q": "input"}, context={"spans": []}) - ) - children_context_record = record_from_step( - Step(type="tool", name="tool", input={"q": "input"}, context={"children": []}) ) - assert isinstance(valid_context_record, ToolSpan) - assert isinstance(children_context_record, ToolSpan) - with pytest.raises(RecordFactoryError, match="context must be a list"): - record_from_step( - Step(type="tool", name="tool", input={"q": "input"}, context={"spans": "invalid"}) - ) - with pytest.raises(RecordFactoryError, match="span context"): - record_from_scorer_invoke_record( - _RecordPayload(type="tool", input="input", spans="invalid") - ) - with pytest.raises(RecordFactoryError, match="trace context"): - record_from_scorer_invoke_record(_RecordPayload(type="trace", input="input")) - with pytest.raises(RecordFactoryError, match="session context"): - record_from_scorer_invoke_record(_RecordPayload(type="session", input="input")) - with pytest.raises(RecordFactoryError, match="trace records"): - record_from_scorer_invoke_record( - _RecordPayload(type="session", input="input", traces=[{"type": "tool"}]) - ) + assert isinstance(metadata_record, ToolSpan) + assert metadata_record.user_metadata == {"provider": "test"} def test_trace_structured_values_are_normalized_without_losing_messages() -> None: @@ -601,7 +582,7 @@ def test_trace_structured_values_are_normalized_without_losing_messages() -> Non name="request", input=trace_input, output=trace_output, - context={"spans": []}, + children=[], ) ) @@ -622,7 +603,7 @@ def test_trace_content_blocks_preserve_text_files_and_unrepresentable_data() -> name="multimodal", input=blocks, output=blocks, - context={"spans": []}, + children=[], ) ) serialized_data_block = { @@ -636,7 +617,7 @@ def test_trace_content_blocks_preserve_text_files_and_unrepresentable_data() -> name="inline-image", input=[serialized_data_block], output=[serialized_data_block], - context={"spans": []}, + children=[], ) ) @@ -665,7 +646,7 @@ def test_trace_message_sequences_flatten_every_text_and_content_part() -> None: ] trace = record_from_step( - Step(type="trace", name="messages", input="question", output=output, context={"spans": []}) + Step(type="trace", name="messages", input="question", output=output, children=[]) ) assert isinstance(trace, Trace) @@ -953,13 +934,10 @@ class DocumentModel(BaseModel): def test_llm_tool_definitions_normalize_nested_pydantic_uuid_and_datetime_values() -> None: identifier = uuid4() timestamp = datetime(2025, 1, 2, tzinfo=UTC) - source = _RecordPayload( - type="llm", - input="question", - tools=[{"id": identifier, "created_at": timestamp}], + normalized_tools = GalileoRecordNormalizer.llm_tools( + [{"id": identifier, "created_at": timestamp}] ) - - record = record_from_scorer_invoke_record(source) + record = LlmSpan.model_validate({"input": "question", "tools": normalized_tools}) assert isinstance(record, LlmSpan) assert record.tools == [ @@ -978,89 +956,149 @@ def test_retriever_dictionaries_become_canonical_documents() -> None: assert documents[0].metadata == {"source": "kb"} -def test_trace_and_session_envelopes_are_supported() -> None: - trace = record_from_step( - Step( - type="trace", - name="outer", - input={"input": "question", "spans": []}, - ) - ) - session = record_from_step( +def test_retriever_input_mapping_extracts_query_and_round_trips_documents() -> None: + record = record_from_step( Step( - type="session", - name="outer", - input={"input": "question", "traces": []}, + type="retriever", + name="policy_search", + input={"query": "wire policy", "partition": "banking"}, + output=[{"content": "Policy text", "metadata": {"source": "kb"}}], ) ) - wrapped_trace = record_from_step( - Step( - type="trace", - name="outer", - input="ignored", - context={"trace": {"name": "inner", "input": "question", "spans": []}}, - ) + + assert isinstance(record, RetrieverSpan) + assert record.input == "wire policy" + assert len(record.output) == 1 + assert isinstance(record.output[0], Document) + restored = RetrieverSpan.model_validate(record.model_dump(mode="json", exclude_none=True)) + assert restored.input == "wire policy" + assert isinstance(restored.output[0], Document) + assert restored.output[0].content == "Policy text" + + +def test_retriever_input_mapping_without_query_uses_json_fallback() -> None: + record = record_from_step( + Step(type="retriever", name="search", input={"partition": "banking"}) ) - wrapped_session = record_from_step( - Step( - type="session", - name="outer", - input="ignored", - context={"session": {"name": "inner", "input": "question", "traces": []}}, - ) + + assert isinstance(record, RetrieverSpan) + assert record.input == '{"partition": "banking"}' + assert GalileoRecordNormalizer.retriever_input({"query": {"phrase": "wire"}}) == ( + '{"phrase": "wire"}' ) - selected_trace = record_from_step( + + +def test_trace_and_session_use_children_and_selectors_only_change_the_root() -> None: + trace = record_from_step( Step( type="trace", name="outer", - input="ignored", - context={ - "trace": { - "name": "inner", - "input": "wrapped input", - "output": "wrapped output", - "spans": [], - } - }, + input="base question", + children=[ + Step( + type="tool", + name="child", + input={"query": "child query"}, + output="child result", + context={"metadata": {"owner": "child"}}, + ), + Step( + type="llm", + name="child llm", + input="child prompt", + output="child response", + tools=[{"name": "lookup"}], + ) + ], ), - selected_data={"input": "selected input", "output": "selected output"}, + selected_data={"input": "selected question", "output": "selected answer"}, ) - selected_session = record_from_step( + session = record_from_step( Step( type="session", - name="outer", - input="ignored", - context={ - "session": { - "name": "inner", - "input": "wrapped input", - "output": "wrapped output", - "traces": [], - } - }, - ), - selected_data={"input": "selected input", "output": "selected output"}, + name="conversation", + input="base session", + children=[ + Step( + type="trace", + name="child trace", + input="child question", + children=[Step(type="llm", name="child llm", input="child input")], + ) + ], + ) ) assert isinstance(trace, Trace) + assert trace.input == "selected question" + assert trace.output == "selected answer" + assert isinstance(trace.spans[0], ToolSpan) + assert json.loads(trace.spans[0].input) == {"query": "child query"} + assert trace.spans[0].output == "child result" + assert trace.spans[0].user_metadata == {"owner": "child"} + assert isinstance(trace.spans[1], LlmSpan) + assert trace.spans[1].tools == [{"name": "lookup"}] + assert trace.spans[1].output.content == "child response" + assert isinstance(session, Session) + assert session.traces[0].name == "child trace" + assert session.traces[0].input == "child question" + assert isinstance(session.traces[0].spans[0], LlmSpan) + + +def test_children_must_be_valid_for_parent() -> None: + with pytest.raises(RecordFactoryError, match="cannot contain 'session'"): + record_from_step( + Step(type="trace", name="trace", input="q", children=[ + Step(type="session", name="session", input="q", children=[]) + ]) + ) + with pytest.raises(RecordFactoryError, match="cannot contain 'tool'"): + record_from_step( + Step(type="session", name="session", input="q", children=[ + Step(type="tool", name="tool", input={}) + ]) + ) + + +def test_trace_and_session_without_children_have_empty_child_collections() -> None: + trace = record_from_step(Step(type="trace", name="request", input="question")) + session = record_from_step(Step(type="session", name="conversation", input="question")) + + assert isinstance(trace, Trace) + assert trace.spans == [] assert isinstance(session, Session) - assert isinstance(wrapped_trace, Trace) - assert isinstance(wrapped_session, Session) - assert isinstance(selected_trace, Trace) - assert isinstance(selected_session, Session) - assert trace.input == "question" - assert session.input == "question" - assert wrapped_trace.name == "inner" - assert wrapped_session.name == "inner" - assert selected_trace.input == "selected input" - assert selected_trace.output == "selected output" - assert selected_session.input == "selected input" - assert selected_session.output == "selected output" + assert session.traces == [] + + +def test_nested_tool_and_retriever_children_are_recursively_converted() -> None: + record = record_from_step( + Step( + type="tool", + name="outer", + input={}, + children=[ + Step( + type="retriever", + name="inner retriever", + input="query", + output=["document"], + children=[Step(type="tool", name="leaf tool", input={"x": 1})], + ) + ], + ) + ) + + assert isinstance(record, ToolSpan) + retriever = record.spans[0] + assert isinstance(retriever, RetrieverSpan) + assert retriever.output[0].content == "document" + assert isinstance(retriever.spans[0], ToolSpan) + assert json.loads(retriever.spans[0].input) == {"x": 1} def test_trace_missing_output_stays_none() -> None: record = record_from_step( - Step(type="trace", name="request", input="question", context={"spans": []}) + Step(type="trace", name="request", input="question", children=[]) ) assert isinstance(record, Trace) @@ -1076,9 +1114,7 @@ def test_optional_fields_and_aliases_are_translated() -> None: tools=[{"name": "search"}], ) ) - tool = record_from_scorer_invoke_record( - _RecordPayload(type="tool", input="input", tool_call_id=123) - ) + tool = ToolSpan.model_validate({"tool_call_id": "123"}) alias = build_galileo_record("question", Step(type="llm", name="answer", input="base")) assert isinstance(llm, LlmSpan) @@ -1089,7 +1125,6 @@ def test_optional_fields_and_aliases_are_translated() -> None: def test_session_message_and_document_sequences_use_public_validators() -> None: - traces = [{"type": "trace", "name": "request", "input": "question", "spans": []}] messages = [{"role": "user", "content": "question"}] documents = [{"content": "answer", "metadata": {"source": "kb"}}] @@ -1099,7 +1134,7 @@ def test_session_message_and_document_sequences_use_public_validators() -> None: name="conversation", input=messages, output=documents, - context={"traces": traces}, + children=[Step(type="trace", name="request", input="question", children=[])], ) ) @@ -1123,15 +1158,11 @@ def test_session_message_and_document_sequences_use_public_validators() -> None: ] -def test_trace_and_session_require_structured_context() -> None: - with pytest.raises(RecordFactoryError, match="trace context is missing"): - record_from_step(Step(type="trace", name="request", input="question")) - with pytest.raises(RecordFactoryError, match="session context is missing"): - record_from_step(Step(type="session", name="conversation", input="question")) +def test_trace_and_session_selector_values_must_be_structured() -> None: with pytest.raises(RecordFactoryError, match="untyped scalar"): build_record( "question", - Step(type="trace", name="request", input="question", context={"spans": []}), + Step(type="trace", name="request", input="question", children=[]), ) @@ -1149,15 +1180,275 @@ def test_record_serialization_is_json_safe() -> None: assert payload["input"] == json.dumps({"q": "x"}) -def test_factory_accepts_the_existing_pydantic_luna_record() -> None: +def test_public_core_model_validates_luna_request_record_payload() -> None: from agent_control_evaluator_galileo.luna import ScorerInvokeRecord - record = record_from_scorer_invoke_record( - ScorerInvokeRecord(type="tool", name="search", input={"query": "q"}) + source = ToolSpan(name="search", input=json.dumps({"query": "q"})) + request_record = ScorerInvokeRecord.model_validate( + source.model_dump(mode="json", exclude_none=True) ) + record = ToolSpan.model_validate(request_record.model_dump(mode="json", exclude_none=True)) - assert isinstance(record, ToolSpan) - assert json.loads(record.input) == {"query": "q"} + assert record == source + + +def test_canonical_records_round_trip_through_luna_wire_model() -> None: + from agent_control_evaluator_galileo.luna import ScorerInvokeRecord + + llm_step = Step( + type="llm", + name="answer", + input=[{"role": "user", "content": "question"}], + output="answer", + tools=[{"name": "lookup"}], + context={"metadata": {"source": "llm"}}, + ) + nested_tool = Step( + type="tool", + name="search", + input={"query": "q"}, + output="result", + context={"metadata": {"source": "tool"}}, + children=[llm_step], + ) + records = [ + ( + llm_step, + LlmSpan, + ), + ( + Step(type="tool", name="flat tool", input={"id": 1}, output="done"), + ToolSpan, + ), + ( + Step( + type="retriever", + name="retrieve", + input="query", + output=[{"content": "document", "metadata": {"source": "kb"}}], + ), + RetrieverSpan, + ), + ( + Step( + type="trace", + name="trace", + input="trace input", + output="trace output", + children=[nested_tool], + ), + Trace, + ), + ( + Step( + type="session", + name="session", + input="session input", + output="session output", + children=[ + Step( + type="trace", + name="session trace", + input="trace input", + children=[ + Step( + type="tool", + name="session tool", + input={"query": "session query"}, + output="session result", + ) + ], + ) + ], + ), + Session, + ), + ] + + for step, core_model in records: + canonical = record_from_step(step) + wire_json = canonical.model_dump(mode="json", exclude_none=True) + request_record = ScorerInvokeRecord.model_validate(wire_json) + reconstructed = core_model.model_validate( + request_record.model_dump(mode="json", exclude_none=True) + ) + + assert type(reconstructed) is type(canonical) + assert reconstructed.model_dump(mode="json", exclude_none=True) == wire_json + + trace = record_from_step(records[3][0]) + assert isinstance(trace, Trace) + assert isinstance(trace.spans[0], ToolSpan) + assert isinstance(trace.spans[0].spans[0], LlmSpan) + assert trace.spans[0].user_metadata == {"source": "tool"} + assert trace.spans[0].spans[0].tools == [{"name": "lookup"}] + + session = record_from_step(records[4][0]) + assert isinstance(session, Session) + assert isinstance(session.traces[0], Trace) + assert isinstance(session.traces[0].spans[0], ToolSpan) + + +def test_luna_wire_hierarchy_uses_top_level_fields_not_context() -> None: + from agent_control_evaluator_galileo.luna import ScorerInvokeRecord + from pydantic import ValidationError + + canonical_trace = record_from_step( + Step( + type="trace", + name="canonical trace", + input="trace input", + children=[Step(type="tool", name="canonical tool", input={})], + ) + ) + canonical_session = record_from_step( + Step( + type="session", + name="canonical session", + input="session input", + children=[ + Step(type="trace", name="canonical child trace", input="trace input", children=[]) + ], + ) + ) + trace_payload = canonical_trace.model_dump(mode="json", exclude_none=True) + session_payload = canonical_session.model_dump(mode="json", exclude_none=True) + assert "spans" in trace_payload + assert "traces" in session_payload + assert "context" not in trace_payload + assert "context" not in session_payload + + wire_trace = ScorerInvokeRecord.model_validate( + { + "type": "trace", + "name": "trace", + "input": "trace input", + "spans": [{"type": "tool", "name": "wire tool", "input": "{}"}], + "context": { + "children": [{"type": "llm", "name": "ignored child"}], + "spans": [{"type": "llm", "name": "ignored span"}], + "traces": [{"type": "trace", "name": "ignored trace"}], + }, + } + ) + wire_trace_json = wire_trace.model_dump(mode="json", exclude_none=True) + try: + rebuilt_trace = Trace.model_validate(wire_trace_json) + except ValidationError as exc: + assert any("context" in error["loc"] for error in exc.errors()) + else: + assert len(rebuilt_trace.spans) == 1 + assert isinstance(rebuilt_trace.spans[0], ToolSpan) + assert rebuilt_trace.spans[0].name == "wire tool" + assert all(span.name != "ignored span" for span in rebuilt_trace.spans) + + # Galileo Core supplies empty hierarchy defaults when top-level fields are absent. + # Context-only hierarchy must not be treated as canonical spans or traces. + context_only_trace = ScorerInvokeRecord.model_validate( + { + "type": "trace", + "name": "context only trace", + "context": { + "children": [{"type": "tool", "name": "ignored child"}], + "spans": [{"type": "tool", "name": "ignored span"}], + }, + } + ) + rebuilt_context_only_trace = Trace.model_validate( + context_only_trace.model_dump(mode="json", exclude_none=True) + ) + assert rebuilt_context_only_trace.spans == [] + + context_only_session = ScorerInvokeRecord.model_validate( + { + "type": "session", + "name": "session", + "context": {"traces": [{"type": "trace", "name": "ignored trace"}]}, + } + ) + context_only_json = context_only_session.model_dump(mode="json", exclude_none=True) + rebuilt_context_only_session = Session.model_validate(context_only_json) + assert rebuilt_context_only_session.traces == [] + + +def test_luna_wire_round_trip_preserves_all_trace_span_types_and_nested_tool_span() -> None: + from agent_control_evaluator_galileo.luna import ScorerInvokeRecord + + trace_step = Step( + type="trace", + name="trace", + input="question", + children=[ + Step(type="llm", name="answer", input="question", output="answer"), + Step(type="tool", name="search", input={"query": "q"}, output="result"), + Step( + type="retriever", + name="retrieve", + input="query", + output=[{"content": "document"}], + ), + ], + ) + tool_step = Step( + type="tool", + name="outer tool", + input={"query": "q"}, + children=[Step(type="llm", name="inner llm", input="tool prompt")], + ) + + for step, core_model in ((trace_step, Trace), (tool_step, ToolSpan)): + canonical = record_from_step(step) + wire_json = canonical.model_dump(mode="json", exclude_none=True) + request_record = ScorerInvokeRecord.model_validate(wire_json) + reconstructed = core_model.model_validate( + request_record.model_dump(mode="json", exclude_none=True) + ) + + assert type(reconstructed) is type(canonical) + assert reconstructed.model_dump(mode="json", exclude_none=True) == wire_json + + trace = record_from_step(trace_step) + assert isinstance(trace, Trace) + assert [type(span) for span in trace.spans] == [LlmSpan, ToolSpan, RetrieverSpan] + + tool = record_from_step(tool_step) + assert isinstance(tool, ToolSpan) + assert isinstance(tool.spans[0], LlmSpan) + assert tool.spans[0].name == "inner llm" + + +@pytest.mark.parametrize( + ("payload", "core_model"), + [ + ( + { + "type": "session", + "name": "session", + "traces": [{"type": "tool", "name": "invalid child"}], + }, + Session, + ), + ( + { + "type": "trace", + "name": "trace", + "spans": [{"type": "session", "name": "invalid child", "traces": []}], + }, + Trace, + ), + ], +) +def test_luna_wire_records_reject_invalid_hierarchy( + payload: dict[str, object], + core_model: type[BaseModel], +) -> None: + from agent_control_evaluator_galileo.luna import ScorerInvokeRecord + from pydantic import ValidationError + + request_record = ScorerInvokeRecord.model_validate(payload) + + with pytest.raises(ValidationError): + core_model.model_validate(request_record.model_dump(mode="json", exclude_none=True)) def test_factory_rejects_invalid_boundaries_and_normalizes_missing_text() -> None: @@ -1165,8 +1456,6 @@ def test_factory_rejects_invalid_boundaries_and_normalizes_missing_text() -> Non with pytest.raises(RecordFactoryError, match="complete Agent Control Step"): record_from_step(object()) # type: ignore[arg-type] - with pytest.raises(UnsupportedStepTypeError, match="unsupported"): - record_from_scorer_invoke_record(_RecordPayload(type="unsupported")) assert GalileoRecordNormalizer.retriever_input(None) == "" assert GalileoRecordNormalizer.session_input(Document(content="plain")) == json.dumps( {"content": "plain"} diff --git a/sdks/python/README.md b/sdks/python/README.md index f9a608b0..9522cf2e 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -25,6 +25,65 @@ async def search_database(query: str) -> str: Controls are defined centrally and enforced automatically at runtime. See [Python SDK Documentation](https://docs.agentcontrol.dev/sdk/python-sdk) for complete reference. +## Supplying available tools to an LLM control + +The `@control` decorator is a convenience path for function-based steps. An LLM +control can include static tool definitions, or a provider can build them from +the function arguments for each pre- and post-execution check: + +```python +AVAILABLE_TOOLS = [ + { + "type": "function", + "function": { + "name": "search", + "description": "Search internal documents", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + } +] + +@agent_control.control(step_type="llm", tools=AVAILABLE_TOOLS) +async def answer(message: str) -> str: + ... +``` + +For per-call tools, pass a provider such as +`tools=lambda args, kwargs: tools_for(kwargs.get("model"))`. The provider +receives the raw positional call arguments in `args` and keyword arguments in +`kwargs` for each pre- and post-execution check. The provider may return `None` +when the step has no available tools. + +For framework adapters and advanced callers, construct the shared `Step` +directly and send it through the SDK's `check_evaluation` helper. This keeps +framework-specific message and tool formats in the adapter, without requiring +the decorator or a framework dependency: + +```python +step = agent_control.Step( + type="llm", + name="openai_agent", + input=messages, + output=response, + tools=available_tools, +) + +async with agent_control.AgentControlClient() as client: + result = await agent_control.evaluation.check_evaluation( + client, + agent_name="my-agent", + step=step, + stage="post", + ) +``` + +The existing `evaluate_controls` helper remains available for callers that +prefer its field-based convenience arguments. + ## Sharing an OpenTelemetry provider with Google ADK When Google ADK and Agent Control should export through the same OpenTelemetry diff --git a/sdks/python/src/agent_control/__init__.py b/sdks/python/src/agent_control/__init__.py index 6c6e117f..04d040f6 100644 --- a/sdks/python/src/agent_control/__init__.py +++ b/sdks/python/src/agent_control/__init__.py @@ -1650,6 +1650,7 @@ async def main(): # Models (re-exported when available) "Agent", "Step", + "JSONObject", "StepSchema", "EvaluationRequest", "EvaluationResult", diff --git a/sdks/python/src/agent_control/control_decorators.py b/sdks/python/src/agent_control/control_decorators.py index fee78333..8c12ae4c 100644 --- a/sdks/python/src/agent_control/control_decorators.py +++ b/sdks/python/src/agent_control/control_decorators.py @@ -31,11 +31,12 @@ async def chat(message: str) -> str: import functools import inspect import time -from collections.abc import Callable +from collections.abc import Callable, Mapping, Sequence +from copy import deepcopy from dataclasses import dataclass, field from typing import Any, TypeVar -from agent_control_models import Step, normalize_action +from agent_control_models import JSONObject, Step, normalize_action from agent_control_telemetry import get_trace_context_from_provider from agent_control import AgentControlClient @@ -58,6 +59,10 @@ async def chat(message: str) -> str: logger = get_logger(__name__) F = TypeVar("F", bound=Callable[..., Any]) +type ToolsProvider = Callable[ + [tuple[Any, ...], Mapping[str, Any]], Sequence[JSONObject] | None +] +type ToolsConfiguration = Sequence[JSONObject] | ToolsProvider | None def _resolve_control_trace_context() -> tuple[str, str]: @@ -98,6 +103,7 @@ class ControlContext: start_time: float step_name: str | None = None step_type: str | None = None + tools: ToolsConfiguration = None # Stats (mutually exclusive: errors vs matches vs non_matches) total_executions: int = 0 @@ -134,6 +140,7 @@ def pre_payload(self) -> dict[str, Any]: output=None, step_name=self.step_name, explicit_step_type=self.step_type, + tools=self.tools, ) def post_payload(self, output: Any) -> dict[str, Any]: @@ -145,6 +152,7 @@ def post_payload(self, output: Any) -> dict[str, Any]: output=output, step_name=self.step_name, explicit_step_type=self.step_type, + tools=self.tools, ) def process_result(self, result: dict[str, Any], check_stage: str) -> None: @@ -247,6 +255,10 @@ def __init__( ) +class _InvalidToolsConfigurationError(ValueError): + """Raised when decorator tools are configured for a non-LLM step.""" + + def _get_current_agent() -> Any | None: """Get the current agent from agent_control module.""" try: @@ -451,6 +463,8 @@ async def _run_control_check( ctx.process_result(result, stage) except (ControlViolationError, ControlSteerError): raise + except _InvalidToolsConfigurationError: + raise except Exception as e: stage_name = "Pre" if stage == "pre" else "Post" logger.error("%s-execution control check failed: %s", stage_name, e, exc_info=True) @@ -492,6 +506,7 @@ def _create_evaluation_payload( output: Any = None, step_name: str | None = None, explicit_step_type: str | None = None, + tools: ToolsConfiguration = None, ) -> dict[str, Any]: """ Create evaluation payload for server, detecting if it's a tool step or LLM step. @@ -505,6 +520,7 @@ def _create_evaluation_payload( output: Function output (None for pre-execution) step_name: Optional explicit step name to override auto-detection explicit_step_type: Optional explicit step type to override inference + tools: Optional static tool definitions or a provider called with args/kwargs """ sig = inspect.signature(func) bound = sig.bind(*args, **kwargs) @@ -536,33 +552,45 @@ def _create_evaluation_payload( else inferred_type ) + if tools is not None and step_type != "llm": + raise _InvalidToolsConfigurationError( + f"@control(tools=...) requires an LLM step; received step_type={step_type!r}." + ) + + resolved_tools = tools(args, kwargs) if callable(tools) else tools + + def validate_payload(payload: dict[str, Any]) -> dict[str, Any]: + """Validate decorator payloads against the shared runtime Step model.""" + Step.model_validate(payload) + return payload + if step_type not in ("llm", "tool"): # Custom step types use the complete bound argument mapping. Unlike an # LLM step, there is no generic convention for selecting one argument # as the prompt/input for a retriever, trace, session, etc. - return { + return validate_payload({ "type": step_type, "name": determined_name, "input": dict(bound.arguments), "output": output if isinstance(output, (str, int, float, bool, dict, list)) else ( None if output is None else str(output) ), - } + }) if step_type == "tool": # This is a tool step - return { + return validate_payload({ "type": step_type, "name": determined_name, "input": dict(bound.arguments), "output": output if isinstance(output, (str, int, float, bool, dict, list)) else ( None if output is None else str(output) ), - } + }) # This is an LLM step input_data = _extract_input_from_args(func, args, kwargs) - return { + payload = { "type": step_type, "name": determined_name, "input": input_data, @@ -570,6 +598,9 @@ def _create_evaluation_payload( None if output is None else str(output) ), } + if resolved_tools is not None: + payload["tools"] = deepcopy(list(resolved_tools)) + return validate_payload(payload) def _handle_evaluation_result(result: dict[str, Any]) -> None: @@ -733,6 +764,7 @@ async def _execute_with_control( is_async: bool, step_name: str | None = None, step_type: str | None = None, + tools: ToolsConfiguration = None, ) -> Any: """ Core control execution logic for both async and sync functions. @@ -787,12 +819,20 @@ async def _execute_with_control( start_time=time.perf_counter(), step_name=step_name, step_type=step_type, + tools=tools, ) ctx.log_start() try: # PRE-EXECUTION: Check controls with check_stage="pre" - await _run_control_check(ctx, "pre", ctx.pre_payload(), controls) + try: + pre_payload = ctx.pre_payload() + except _InvalidToolsConfigurationError: + raise + except Exception as e: + logger.error("Pre-execution control payload creation failed: %s", e, exc_info=True) + raise RuntimeError(_unexpected_control_failure_message("pre", e)) from e + await _run_control_check(ctx, "pre", pre_payload, controls) # Execute the function if is_async: @@ -801,7 +841,14 @@ async def _execute_with_control( output = func(*args, **kwargs) # POST-EXECUTION: Check controls with check_stage="post" - await _run_control_check(ctx, "post", ctx.post_payload(output), controls) + try: + post_payload = ctx.post_payload(output) + except _InvalidToolsConfigurationError: + raise + except Exception as e: + logger.error("Post-execution control payload creation failed: %s", e, exc_info=True) + raise RuntimeError(_unexpected_control_failure_message("post", e)) from e + await _run_control_check(ctx, "post", post_payload, controls) return output finally: @@ -812,6 +859,10 @@ def control( policy: str | None = None, step_name: str | None = None, step_type: str | None = None, + *, + tools: Sequence[JSONObject] | Callable[ + [tuple[Any, ...], Mapping[str, Any]], Sequence[JSONObject] | None + ] | None = None, ) -> Callable[[F], F]: """ Decorator to apply server-defined controls at this code location. @@ -827,6 +878,10 @@ def control( step_type: Optional custom type for this step. If not provided, tool-like functions are classified as ``tool`` and other functions as ``llm``. + tools: Optional tool definitions for LLM steps. Static definitions are + copied when the decorator is created. A provider receives the + raw call-site positional arguments as ``args`` and keyword + arguments as ``kwargs`` at each pre- and post-execution check. Returns: Decorated function @@ -885,6 +940,8 @@ async def handle_user_input(user_message: str) -> str: # controls associated with the agent via policy and direct links. _ = policy + configured_tools = deepcopy(list(tools)) if tools is not None and not callable(tools) else tools + def decorator(func: F) -> F: # Register this function's step schema for auto-discovery by init() from agent_control._control_registry import register @@ -900,6 +957,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: is_async=True, step_name=step_name, step_type=step_type, + tools=configured_tools, ) # Copy over ALL attributes from the original function (important for LangChain tools) @@ -920,6 +978,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: is_async=False, step_name=step_name, step_type=step_type, + tools=configured_tools, ) ) diff --git a/sdks/python/tests/test_control_decorators.py b/sdks/python/tests/test_control_decorators.py index 808c30c4..c883b03f 100644 --- a/sdks/python/tests/test_control_decorators.py +++ b/sdks/python/tests/test_control_decorators.py @@ -1,12 +1,13 @@ """Tests for @control decorator.""" +from copy import deepcopy from unittest.mock import MagicMock, patch import httpx import pytest -from agent_control_telemetry import clear_trace_context_provider, set_trace_context_provider - from agent_control.control_decorators import ControlSteerError, ControlViolationError, control +from agent_control_telemetry import clear_trace_context_provider, set_trace_context_provider +from pydantic import ValidationError # ============================================================================= # FIXTURES @@ -218,6 +219,212 @@ async def retrieve(query: str, top_k: int, filters: dict[str, str]) -> str: "filters": {"language": "en"}, } + @pytest.mark.asyncio + async def test_omitted_tools_preserves_legacy_payload(self, mock_agent, mock_safe_response): + payloads = [] + + async def capture_payload(*args, **kwargs): + payloads.append(args[1]) + return mock_safe_response + + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate", side_effect=capture_payload), + ): + + @control() + async def answer(message: str) -> str: + return "answer" + + await answer("question") + + assert payloads == [ + {"type": "llm", "name": "answer", "input": "question", "output": None}, + {"type": "llm", "name": "answer", "input": "question", "output": "answer"}, + ] + + @pytest.mark.asyncio + async def test_static_tools_are_copied_into_pre_and_post_payloads( + self, mock_agent, mock_safe_response + ): + original_tools = [ + { + "type": "function", + "function": { + "name": "search", + "description": "Search documents", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + }, + }, + } + ] + expected_tools = [ + { + "type": "function", + "function": { + "name": "search", + "description": "Search documents", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + }, + }, + } + ] + payloads = [] + + async def capture_payload(*args, **kwargs): + payload = args[1] + payloads.append( + { + "type": payload["type"], + "name": payload["name"], + "input": payload["input"], + "output": payload["output"], + "tools": deepcopy(payload["tools"]), + } + ) + payload["tools"][0]["function"]["parameters"]["properties"]["query"][ + "type" + ] = "integer" + return mock_safe_response + + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate", side_effect=capture_payload), + ): + @control(tools=original_tools) + async def answer(message: str) -> str: + return "answer" + + original_tools[0]["function"]["description"] = "caller mutation" + await answer("question") + + assert [payload["tools"] for payload in payloads] == [expected_tools, expected_tools] + assert original_tools[0]["function"]["description"] == "caller mutation" + assert original_tools[0]["function"]["parameters"]["properties"]["query"][ + "type" + ] == "string" + + @pytest.mark.asyncio + async def test_dynamic_tools_receive_call_arguments(self, mock_agent, mock_safe_response): + calls = [] + payloads = [] + available_tools = [{"name": "lookup", "parameters": {"type": "object"}}] + + def provide_tools(args, kwargs): + calls.append((args, dict(kwargs))) + return available_tools + + async def capture_payload(*args, **kwargs): + payloads.append(args[1]) + return mock_safe_response + + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate", side_effect=capture_payload), + ): + + @control(tools=provide_tools) + async def answer(message: str, *, language: str) -> str: + return "answer" + + await answer("question", language="en") + + assert calls == [ + (("question",), {"language": "en"}), + (("question",), {"language": "en"}), + ] + assert payloads[0]["tools"] == available_tools + assert payloads[1]["tools"] == available_tools + + @pytest.mark.asyncio + async def test_dynamic_tool_provider_can_return_none(self, mock_agent, mock_safe_response): + payloads = [] + + async def capture_payload(*args, **kwargs): + payloads.append(args[1]) + return mock_safe_response + + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate", side_effect=capture_payload), + ): + + @control(tools=lambda args, kwargs: None) + async def answer(message: str) -> str: + return "answer" + + await answer("question") + + assert all("tools" not in payload for payload in payloads) + + @pytest.mark.asyncio + async def test_invalid_tool_definitions_use_step_validation( + self, mock_agent + ): + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate") as mock_evaluate, + ): + + @control(tools=[{"name": "invalid", "schema": object()}]) + async def answer(message: str) -> str: + return "should not run" + + with pytest.raises(RuntimeError) as exc_info: + await answer("question") + + assert isinstance(exc_info.value.__cause__, ValidationError) + assert "tools" in str(exc_info.value.__cause__) + mock_evaluate.assert_not_called() + + @pytest.mark.parametrize("step_type", ["tool", "retriever"]) + @pytest.mark.asyncio + async def test_tools_are_rejected_for_non_llm_steps( + self, mock_agent, step_type + ): + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate") as mock_evaluate, + ): + + @control(step_type=step_type, tools=[]) + async def answer(message: str) -> str: + return "should not run" + + with pytest.raises(ValueError, match="requires an LLM step"): + await answer("question") + + mock_evaluate.assert_not_called() + + @pytest.mark.asyncio + async def test_dynamic_tool_provider_failure_blocks_execution(self, mock_agent): + executed = False + + def provider(args, kwargs): + raise ValueError("provider failed") + + with ( + patch("agent_control.control_decorators._get_current_agent", return_value=mock_agent), + patch("agent_control.control_decorators._evaluate") as mock_evaluate, + ): + + @control(tools=provider) + async def answer(message: str) -> str: + nonlocal executed + executed = True + return "answer" + + with pytest.raises(RuntimeError, match="Execution blocked for safety") as exc_info: + await answer("question") + + assert str(exc_info.value.__cause__) == "provider failed" + assert executed is False + mock_evaluate.assert_not_called() + def test_explicit_step_type_overrides_tool_inference( self, mock_agent, mock_safe_response ): diff --git a/sdks/python/tests/test_evaluation.py b/sdks/python/tests/test_evaluation.py index f4d5915d..94f596fc 100644 --- a/sdks/python/tests/test_evaluation.py +++ b/sdks/python/tests/test_evaluation.py @@ -108,6 +108,54 @@ def json(self) -> dict[str, object]: } +@pytest.mark.asyncio +async def test_check_evaluation_accepts_step_with_tool_definitions(): + """Framework callers can send a fully structured Step directly.""" + + class DummyResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"is_safe": True, "confidence": 1.0} + + client = MagicMock() + client.http_client = MagicMock() + client.http_client.post = AsyncMock(return_value=DummyResponse()) + tools = [ + { + "type": "function", + "function": { + "name": "search", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + }, + }, + } + ] + step = Step( + type="llm", + name="openai_agent", + input=[{"role": "user", "content": "question"}], + output={"role": "assistant", "content": "answer"}, + tools=tools, + ) + + await evaluation.check_evaluation( + client=client, + agent_name="Agent-Example_01", + step=step, + stage="post", + ) + + sent_step = client.http_client.post.await_args.kwargs["json"]["step"] + assert sent_step["tools"] == tools + assert sent_step["tools"][0]["function"]["parameters"]["properties"]["query"] == { + "type": "string" + } + + @pytest.mark.asyncio async def test_evaluate_controls_requires_server_url(): """evaluate_controls should require server_url to be configured."""