@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import ast
 import json
 from collections.abc import Mapping, Sequence
 from dataclasses import asdict, is_dataclass
@@ -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]:
 
@@ -928,6 +928,49 @@ def handler(request: httpx.Request) -> httpx.Response:
             "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
 
@@ -164,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"),
     [
@@ -902,6 +956,38 @@ def test_retriever_dictionaries_become_canonical_documents() -> None:
     assert documents[0].metadata == {"source": "kb"}
 
 
+def test_retriever_input_mapping_extracts_query_and_round_trips_documents() -> None:
+    record = record_from_step(
+        Step(
+            type="retriever",
+            name="policy_search",
+            input={"query": "wire policy", "partition": "banking"},
+            output=[{"content": "Policy text", "metadata": {"source": "kb"}}],
+        )
+    )
+
+    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"})
+    )
+
+    assert isinstance(record, RetrieverSpan)
+    assert record.input == '{"partition": "banking"}'
+    assert GalileoRecordNormalizer.retriever_input({"query": {"phrase": "wire"}}) == (
+        '{"phrase": "wire"}'
+    )
+
+
 def test_trace_and_session_use_children_and_selectors_only_change_the_root() -> None:
     trace = record_from_step(
         Step(