AgentOps incorrect message capture causes KeyError in convert_to_openai_messages

[lib4x]

When using AgentOps with Agent Lightning, the library sometimes captures incomplete message objects. These incomplete messages are passed to convert_to_openai_messages, which expects required keys such as "role" and "tool_calls" to always exist.

def convert_to_openai_messages(prompt_completion_list: List[_RawSpanInfo]) -> Generator[OpenAIMessages, None, None]:
"""Convert raw trace payloads into OpenAI-style chat messages.

The function consumes an iterable produced by
[`TraceToMessages.adapt()`][agentlightning.TraceToMessages.adapt] and yields
structures that match the OpenAI fine-tuning JSONL schema, including tool definitions.

Args:
    prompt_completion_list: Raw prompt/completion/tool payloads extracted from a trace.

Returns:
    A generator that yields [`OpenAIMessages`][agentlightning.adapter.messages.OpenAIMessages]
    entries compatible with the OpenAI Functions fine-tuning format.
"""

# Import locally to avoid legacy OpenAI version type import errors
from openai.types.chat import (
    ChatCompletionAssistantMessageParam,
    ChatCompletionFunctionToolParam,
    ChatCompletionMessageFunctionToolCallParam,
    ChatCompletionMessageParam,
)

for pc_entry in prompt_completion_list:
    messages: List[ChatCompletionMessageParam] = []

    # Extract messages
    for msg in pc_entry["prompt"]:
        **role = msg["role"]**

        if role == "assistant" and "tool_calls" in msg:
            # Use the tool_calls directly
            # This branch is usually not used in the wild.
            tool_calls: List[ChatCompletionMessageFunctionToolCallParam] = [
                ChatCompletionMessageFunctionToolCallParam(
                    **id=call["id"],**
                    type="function",
                    function={"name": call["name"], "arguments": call["arguments"]},
                )
                for call in msg["tool_calls"]
            ]
            messages.append(
                ChatCompletionAssistantMessageParam(role="assistant", content=None, tool_calls=tool_calls)
            )
        else:
            # Normal user/system/tool content
            message = cast(
                ChatCompletionMessageParam,
                TypeAdapter(ChatCompletionMessageParam).validate_python(
                    dict(role=role, content=msg.get("content", ""), tool_call_id=msg.get("tool_call_id", None))
                ),
            )
            messages.append(message)

    # Extract completions (assistant outputs after tool responses)
    for comp in pc_entry["completion"]:
        if comp.get("role") == "assistant":
            content = comp.get("content")
            if pc_entry["tools"]:
                tool_calls = [
                    ChatCompletionMessageFunctionToolCallParam(
                        id=tool["call"]["id"],
                        type=tool["call"]["type"],
                        function={"name": tool["name"], "arguments": tool["parameters"]},
                    )
                    for tool in pc_entry["tools"]
                ]
                messages.append(
                    ChatCompletionAssistantMessageParam(role="assistant", content=content, tool_calls=tool_calls)
                )
            else:
                messages.append(ChatCompletionAssistantMessageParam(role="assistant", content=content))

    # Build tools definitions (if available)
    if "functions" in pc_entry["request"]:
        tools = [
            ChatCompletionFunctionToolParam(
                type="function",
                function={
                    "name": fn["name"],
                    "description": fn.get("description", ""),
                    "parameters": (
                        json.loads(fn["parameters"]) if isinstance(fn["parameters"], str) else fn["parameters"]
                    ),
                },
            )
            for fn in pc_entry["request"]["functions"]
        ]
        yield OpenAIMessages(messages=messages, tools=tools)
    else:
        yield OpenAIMessages(messages=messages, tools=None)

[ultmaster]

could you give me a minimal test code snippet to reproduce your issue?

[lib4x]

class MiraRequirement(TypedDict):


class MiraTask(TypedDict):
    user_id: str
    project_id: str
    thread_id: str
    sandbox_id: str
    tools: List[ChatCompletionToolParam]
    thread_manager: ThreadManager
    max_iterations: int
    task_input: MiraRequirement
    init_messages: Optional[List[ChatCompletionMessageParam]]

@rollout
async def mira_agent(task: MiraTask, prompt_template: PromptTemplate) -> float:
    system_prompt = prompt_template.format(**task["task_input"])
    messages: List[ChatCompletionMessageParam] = [
        ChatCompletionSystemMessageParam(content=system_prompt, role="system")
    ]

    if task.get("init_messages"):
        messages.extend(task.get("init_messages"))

    tools = task.get("tools", [])
    thread_manager = task.get("thread_manager")

    client = AsyncOpenAI()

    iteration = 0
    max_iterations = task.get("max_iterations", 20)
    while iteration < max_iterations:
        iteration += 1
        try:
            resp = await client.chat.completions.create(
                model="openai/gpt-5-mini",
                messages=messages,
                tools=tools,
                tool_choice="auto",
                temperature=0.0,
            )

            is_termination = False
            tool_call_params: List[ChatCompletionMessageFunctionToolCallParam] = []
            tool_results: List[ChatCompletionToolMessageParam] = []
            tool_calls = resp.choices[0].message.tool_calls

            if tool_calls:
                for tc in tool_calls:
                    if tc.function.name in TERMINATION_TOOLS:
                        is_termination = True

                    tool_call_params.append(
                        ChatCompletionMessageFunctionToolCallParam(
                            id=tc.id,
                            type="function",
                            function={"name": tc.function.name, "arguments": tc.function.arguments},
                        )
                    )

                    try:
                        tool_output = await execute_tool_call(thread_manager, tc)
                    except Exception as e:
                        tool_output = ToolResult(
                            success=False, output=f"Error executing tool {tc.function.name}: {str(e)}"
                        )

                    tool_results.append(
                        ChatCompletionToolMessageParam(
                            role="tool",
                            tool_call_id=tc.id,
                            content=json.dumps(tool_output, default=str),
                        )
                    )

            if tool_call_params:
                messages.append(
                    ChatCompletionAssistantMessageParam(
                        role="assistant",
                        content=resp.choices[0].message.content,
                        tool_calls=tool_call_params,
                    )
                )
            else:
                messages.append(
                    ChatCompletionAssistantMessageParam(
                        role="assistant",
                        content=resp.choices[0].message.content,
                    )
                )

            if tool_results:
                messages.extend(tool_results)

            if is_termination:
                logger.info(f"✅ Mira agent reached termination condition with tool call")
                break

        except Exception as e:
            logger.error(f"❌ Error in iteration {iteration}: {e}", exc_info=True)
            break

    return mira_agent_grader(messages)

async def execute_tool_call(tm: ThreadManager, tc: ChatCompletionMessageToolCall) -> ToolResult:
    available_functions = tm.tool_registry.get_available_functions()
    tool_func = available_functions.get(tc.function.name)

    if not tool_func:
        error_msg = f"Tool function '{tc.function.name}' not found in registry"
        logger.error(f"  ❌ {error_msg}")
        return ToolResult(success=False, output=error_msg)

    logger.info(f"  Found tool function for '{tc.function.name}', executing...")
    try:
        args = json.loads(tc.function.arguments)
        return await tool_func(**args)
    except Exception as e:
        error_msg = f"Error executing {tc.function.name}: {str(e)}"
        logger.error(f"  ❌ {error_msg}", exc_info=True)
        return ToolResult(success=False, output=error_msg)

[ultmaster]

Does the entry really contains no role or tool calls? In that case, how does the agent run without crashing? I think OpenAI requires the completion call messages to have tole.

[lib4x]

Of course the entry does include tools.
In our design, the MiraTask structure carries both the tool definitions and the initial user message:

class MiraTask(TypedDict):
    user_id: str
    project_id: str
    thread_id: str
    sandbox_id: str
    tools: List[ChatCompletionToolParam]
    thread_manager: ThreadManager
    max_iterations: int
    task_input: MiraRequirement
    init_messages: Optional[List[ChatCompletionMessageParam]]

• The tools field passes all tool definitions required by the agent(100+).
• The init_messages field provides the initial user query, including the proper role (“user”) and any other metadata.

[ultmaster]

I believe this needs further debugging and investigation on the adapter. Probably we need a unit-test to reproduce the issue first.