From 20dfc1886ced5af5d9d0aef67f40c5ec01821ed9 Mon Sep 17 00:00:00 2001 From: bozbas Date: Mon, 24 Aug 2026 06:57:28 +0000 Subject: [PATCH] fix(sensor): support current Codex tool records Summary: Intent: - Preserve Codex sessions and tool activity as event schemas evolve. - Derive tool outcomes from explicit structured result envelopes. Changes: - Accept alternate call identifiers, argument shapes, output containers, and malformed timestamps. - Normalize status and error data while retaining typed results and MCP namespaces. --- Sensor/adr_sensor/parsers/codex_parser.py | 418 ++++++++++++++++---- Sensor/tests/test_parsers.py | 462 ++++++++++++++++++++++ 2 files changed, 806 insertions(+), 74 deletions(-) diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 4987c95..03e1e86 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -4,16 +4,43 @@ """ import json +import re import traceback +from collections.abc import Mapping from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from ..schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage from ..utils.string_utils import truncate_middle from ..utils.timestamp_utils import normalize_timestamp from .base_parser import BaseParser +MAX_ARGUMENT_JSON_LENGTH = 100_000 +MAX_NORMALIZATION_DEPTH = 8 +MAX_COLLECTION_ITEMS = 100 +MAX_TEXT_LENGTH = 1000 +_DEPTH_LIMIT_MARKER = "[truncated: maximum depth reached]" +_MCP_NAME_PART = re.compile(r"^[A-Za-z0-9_-]+$") +_CALL_ID_KEYS = ( + "call_id", + "callId", + "callID", + "tool_call_id", + "toolCallId", + "toolCallID", + "id", +) +_SUCCESS_STATES = frozenset({"success", "succeeded", "completed", "complete", "done", "ok"}) +_ERROR_STATES = frozenset( + {"error", "failed", "failure", "cancelled", "canceled", "rejected", "timed_out", "timeout"} +) +_PENDING_STATES = frozenset({"pending", "in_progress", "running", "queued", "started"}) +_NON_TEXT_CONTENT_TYPES = frozenset( + {"computer_screenshot", "image", "image_url", "input_image", "output_image"} +) +_TEXT_CONTENT_TYPES = frozenset({"input_text", "output_text", "text"}) + class CodexParser(BaseParser): """Parser for OpenAI Codex CLI JSONL log files.""" @@ -61,8 +88,17 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: continue try: event = json.loads(line) - self._process_event(event, session_data) - except json.JSONDecodeError: + if not isinstance(event, Mapping): + continue + + payload = event.get("payload") + if not isinstance(payload, Mapping): + continue + + self._process_event(event, payload, session_data) + except Exception: + # Rollout records evolve independently; keep a malformed + # record from invalidating the rest of the session. continue if not session_data["id"]: @@ -74,6 +110,7 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: ToolUsage( tool_name=tool_dict["tool_name"], tool_type=tool_dict["tool_type"], + server_name=tool_dict.get("server_name"), arguments=tool_dict["arguments"], result=tool_dict.get("result"), status=tool_dict.get("status"), @@ -108,76 +145,260 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: traceback.print_exc() return None - def _truncate_large_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]: - """Truncate large string values in tool arguments.""" - if not isinstance(arguments, dict): - return arguments + def _bound_value(self, value: Any, depth: int = 0) -> Any: + """Bound nested values before retaining them in an event.""" + if isinstance(value, str): + return truncate_middle(value, max_length=MAX_TEXT_LENGTH, edge_chars=400) - truncated = {} - for key, value in arguments.items(): - if isinstance(value, str) and len(value) > 1000: - truncated[key] = truncate_middle(value, max_length=1000, edge_chars=400) - else: - truncated[key] = value + if value is None or isinstance(value, (bool, int, float)): + return value - return truncated + if depth >= MAX_NORMALIZATION_DEPTH: + return _DEPTH_LIMIT_MARKER - def _parse_tool_arguments(self, raw_arguments: Any) -> Dict[str, Any]: - """Coerce a tool's raw argument payload into a dict. + if isinstance(value, Mapping): + bounded: Dict[str, Any] = {} + for index, (key, item) in enumerate(value.items()): + if index >= MAX_COLLECTION_ITEMS: + break + bounded_key = truncate_middle(str(key), max_length=MAX_TEXT_LENGTH, edge_chars=400) + bounded[bounded_key] = self._bound_value(item, depth + 1) + return bounded + + if isinstance(value, (list, tuple)): + return [self._bound_value(item, depth + 1) for item in value[:MAX_COLLECTION_ITEMS]] + + return truncate_middle(str(value), max_length=MAX_TEXT_LENGTH, edge_chars=400) - Falls back to {"raw": ...} for anything that is not a JSON object, so a - non-JSON command string is preserved rather than silently discarded. - """ - if isinstance(raw_arguments, dict): - return raw_arguments + def _parse_tool_arguments(self, raw_arguments: Any) -> Dict[str, Any]: + """Coerce a tool's raw arguments into a recursively bounded dict.""" + if raw_arguments is None or raw_arguments == "": + return {} if isinstance(raw_arguments, str): + if len(raw_arguments) <= MAX_ARGUMENT_JSON_LENGTH: + try: + raw_arguments = json.loads(raw_arguments) + except (json.JSONDecodeError, RecursionError, ValueError): + pass + + if isinstance(raw_arguments, Mapping): + return self._bound_value(raw_arguments) + + return {"raw": self._bound_value(raw_arguments)} + + @staticmethod + def _decode_output_container(output: Any) -> Any: + """Decode bounded JSON object/array strings so structural signals remain visible.""" + if not isinstance(output, str) or len(output) > MAX_ARGUMENT_JSON_LENGTH: + return output + + try: + decoded = json.loads(output) + except (json.JSONDecodeError, RecursionError, ValueError): + return output + return decoded if isinstance(decoded, (Mapping, list)) else output + + def _flatten_result_parts(self, value: Any, depth: int = 0, in_collection: bool = False) -> List[str]: + """Flatten common content containers without interpreting their prose.""" + if value is None: + return [] + + if isinstance(value, str): + return [truncate_middle(value, max_length=MAX_TEXT_LENGTH, edge_chars=400)] + + if depth >= MAX_NORMALIZATION_DEPTH: + return [_DEPTH_LIMIT_MARKER] + + if isinstance(value, list): + parts: List[str] = [] + for item in value[:MAX_COLLECTION_ITEMS]: + if isinstance(item, (str, list, Mapping)): + parts.extend(self._flatten_result_parts(item, depth + 1, in_collection=True)) + return parts + + if isinstance(value, Mapping): + item_type = value.get("type") + if isinstance(item_type, str): + normalized_type = item_type.lower() + if normalized_type in _NON_TEXT_CONTENT_TYPES: + return [] + if normalized_type not in _TEXT_CONTENT_TYPES: + try: + return [json.dumps(self._bound_value(value), ensure_ascii=False, separators=(",", ":"))] + except (TypeError, ValueError): + return [truncate_middle(str(value), max_length=MAX_TEXT_LENGTH, edge_chars=400)] + + for wrapper_key in ("Err", "Ok"): + if wrapper_key in value: + parts = self._flatten_result_parts(value[wrapper_key], depth + 1) + if parts: + return parts + + for content_key in ("content", "text", "output", "result", "message", "error", "value"): + if content_key in value: + parts = self._flatten_result_parts(value[content_key], depth + 1) + if parts: + return parts + + stream_parts: List[str] = [] + for stream_key in ("stdout", "stderr"): + if stream_key in value: + stream_parts.extend(self._flatten_result_parts(value[stream_key], depth + 1)) + if stream_parts: + return stream_parts + try: - parsed = json.loads(raw_arguments) - except (json.JSONDecodeError, ValueError): - return {"raw": raw_arguments} - return parsed if isinstance(parsed, dict) else {"raw": raw_arguments} + return [json.dumps(self._bound_value(value), ensure_ascii=False, separators=(",", ":"))] + except (TypeError, ValueError): + return [truncate_middle(str(value), max_length=MAX_TEXT_LENGTH, edge_chars=400)] - return {"raw": raw_arguments} if raw_arguments else {} + if in_collection: + return [] + return [truncate_middle(str(value), max_length=MAX_TEXT_LENGTH, edge_chars=400)] def _normalize_tool_output(self, output: Any) -> Optional[str]: - """Normalize a tool result to a truncated string. - - function_call_output carries a plain string; custom_tool_call_output - carries a list of content items ({"type": "input_text", "text": ...}). - """ + """Normalize string, list, or mapping output into bounded text.""" if output is None: return None - if isinstance(output, str): - text = output - elif isinstance(output, list): - parts = [] - for item in output: - if isinstance(item, str): - parts.append(item) - elif isinstance(item, dict): - value = item.get("text") - if isinstance(value, str): - parts.append(value) - text = "\n".join(parts) - else: - text = str(output) - - if not text: - return text - - return truncate_middle(text, max_length=1000, edge_chars=400) - - def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): + output = self._decode_output_container(output) + text = "\n".join(part for part in self._flatten_result_parts(output) if part) + return truncate_middle(text, max_length=MAX_TEXT_LENGTH, edge_chars=400) if text else text + + @staticmethod + def _classify_tool( + tool_name: Any, fallback_type: str, namespace: Any = None + ) -> Tuple[str, Optional[str]]: + """Classify well-formed encoded names or explicit MCP namespaces.""" + if not isinstance(tool_name, str): + return fallback_type, None + + parts = tool_name.split("__") + if ( + len(parts) == 3 + and parts[0] == "mcp" + and _MCP_NAME_PART.fullmatch(parts[1]) + and _MCP_NAME_PART.fullmatch(parts[2]) + ): + return "mcp_tool", parts[1] + if isinstance(namespace, str): + namespace_parts = namespace.split("__", 1) + if ( + len(namespace_parts) == 2 + and namespace_parts[0] == "mcp" + and _MCP_NAME_PART.fullmatch(namespace_parts[1]) + and _MCP_NAME_PART.fullmatch(tool_name) + ): + return "mcp_tool", namespace_parts[1] + return fallback_type, None + + @staticmethod + def _call_ids(payload: Mapping[str, Any]) -> List[str]: + """Return distinct call identifiers from common wire aliases.""" + call_ids: List[str] = [] + for key in _CALL_ID_KEYS: + value = payload.get(key) + if isinstance(value, bool) or not isinstance(value, (str, int)): + continue + call_id = str(value).strip() + if call_id and call_id not in call_ids: + call_ids.append(call_id) + return call_ids + + @staticmethod + def _canonical_status(value: Any) -> Optional[str]: + if not isinstance(value, str): + return None + normalized = value.strip().lower().replace("-", "_").replace(" ", "_") + if normalized in _SUCCESS_STATES: + return "success" + if normalized in _ERROR_STATES: + return "error" + if normalized in _PENDING_STATES: + return "pending" + return None + + def _scan_outcome( + self, + value: Any, + signals: set, + error_details: List[Any], + ) -> None: + """Collect outcome signals only from an explicit result envelope.""" + if not isinstance(value, Mapping): + return + + if "Err" in value: + signals.add("error") + error_details.append(value.get("Err")) + if "Ok" in value: + signals.add("success") + if isinstance(value.get("Ok"), Mapping): + self._scan_outcome(value["Ok"], signals, error_details) + + for key in ("isError", "is_error"): + is_error = value.get(key) + if isinstance(is_error, bool): + signals.add("error" if is_error else "success") + + for key in ("status", "state"): + status = self._canonical_status(value.get(key)) + if status: + signals.add(status) + + for key in ("exit_code", "exitCode", "exit_status", "exitStatus"): + exit_code = value.get(key) + if isinstance(exit_code, (int, float)) and not isinstance(exit_code, bool): + if exit_code == 0: + signals.add("success") + else: + signals.add("error") + error_details.append(f"Exit code: {exit_code}") + + if "error" in value and value.get("error") not in (None, False, ""): + signals.add("error") + error_details.append(value.get("error")) + + def _infer_tool_outcome(self, sources: List[Any], default: str) -> Tuple[str, Any]: + """Infer a canonical status, preferring error evidence on conflicts.""" + signals = set() + error_details: List[Any] = [] + for source in sources: + self._scan_outcome(source, signals, error_details) + + if "error" in signals: + return "error", next((detail for detail in error_details if detail is not None), None) + if "success" in signals: + return "success", None + if "pending" in signals: + return "pending", None + return default, None + + def _process_event( + self, event: Mapping[str, Any], payload: Mapping[str, Any], session_data: Dict[str, Any] + ): """Process a single event.""" evt_type = event.get("type") - payload = event.get("payload", {}) if evt_type == "session_meta": - session_data["id"] = payload.get("id") - if payload.get("timestamp"): - session_data["timestamp"] = normalize_timestamp(payload.get("timestamp")) + if session_data["id"] is not None: + return + + session_id = payload.get("id") + if not isinstance(session_id, str) or not session_id: + return + + timestamp = payload.get("timestamp") + normalized_timestamp = None + if timestamp: + try: + normalized_timestamp = normalize_timestamp(timestamp) + except (TypeError, ValueError, OverflowError, OSError): + pass + + session_data["id"] = session_id + session_data["timestamp"] = normalized_timestamp session_data["cwd"] = payload.get("cwd") elif evt_type == "turn_context": @@ -189,17 +410,33 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): if item_type == "message": role = payload.get("role") - content_list = payload.get("content", []) - text_content = "" - for content_item in content_list: - if content_item.get("type") in ["input_text", "output_text"]: - text_content += content_item.get("text", "") + content = payload.get("content", []) + if isinstance(content, str): + text_content = content + else: + if isinstance(content, Mapping): + content_items = [content] + elif isinstance(content, list): + content_items = content + else: + content_items = [] + + text_parts = [] + for content_item in content_items: + if isinstance(content_item, str): + text_parts.append(content_item) + elif ( + isinstance(content_item, Mapping) + and content_item.get("type") in ("input_text", "output_text") + and isinstance(content_item.get("text"), str) + ): + text_parts.append(content_item["text"]) + text_content = "".join(text_parts) if text_content: session_data["messages"].append({"role": role, "content": text_content, "tools": []}) elif item_type in ("function_call", "custom_tool_call"): - call_id = payload.get("call_id") tool_name = payload.get("name") # The two record shapes differ in where the arguments live and how @@ -214,37 +451,70 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): raw_arguments = payload.get("arguments", "{}") arguments = self._parse_tool_arguments(raw_arguments) - arguments = self._truncate_large_arguments(arguments) + tool_type, server_name = self._classify_tool( + tool_name, item_type, payload.get("namespace")) + status, error_detail = self._infer_tool_outcome([payload], default="pending") + error = self._normalize_tool_output(error_detail) if error_detail is not None else None tool_dict = { "tool_name": tool_name, - "tool_type": item_type, + "tool_type": tool_type, + "server_name": server_name, "arguments": arguments, - "status": payload.get("status") or "pending", + "status": status, "result": None, + "error": error, } if not session_data["messages"] or session_data["messages"][-1]["role"] != "assistant": session_data["messages"].append({"role": "assistant", "content": "", "tools": []}) session_data["messages"][-1]["tools"].append(tool_dict) - session_data["pending_tool_calls"][call_id] = tool_dict + for call_id in self._call_ids(payload): + session_data["pending_tool_calls"][call_id] = tool_dict elif item_type in ("function_call_output", "custom_tool_call_output"): - call_id = payload.get("call_id") - output = self._normalize_tool_output(payload.get("output")) + tool_dict = next( + ( + session_data["pending_tool_calls"][call_id] + for call_id in self._call_ids(payload) + if call_id in session_data["pending_tool_calls"] + ), + None, + ) + if tool_dict is not None: + raw_output = next( + (payload[key] for key in ("output", "result", "content") if key in payload), + None, + ) + structured_output = self._decode_output_container(raw_output) + output = self._normalize_tool_output(structured_output) + default_status = ( + tool_dict["status"] if tool_dict["status"] in ("success", "error") else "success" + ) + status, error_detail = self._infer_tool_outcome( + [payload, structured_output], default=default_status + ) + error = self._normalize_tool_output(error_detail) if error_detail is not None else None + if status == "error": + error = error or tool_dict.get("error") or output + else: + error = None - if call_id in session_data["pending_tool_calls"]: - tool_dict = session_data["pending_tool_calls"][call_id] tool_dict["result"] = output - tool_dict["status"] = "success" + tool_dict["status"] = status + tool_dict["error"] = error elif item_type == "reasoning": summary_list = payload.get("summary", []) reasoning_text = "" - for summary_item in summary_list: - if summary_item.get("type") == "summary_text": - reasoning_text += summary_item.get("text", "") + "\n" + if isinstance(summary_list, list): + for summary_item in summary_list: + if not isinstance(summary_item, Mapping): + continue + summary_text = summary_item.get("text") + if summary_item.get("type") == "summary_text" and isinstance(summary_text, str): + reasoning_text += summary_text + "\n" if reasoning_text: if ( diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 4112a14..78d03be 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -316,6 +316,316 @@ def test_function_call_still_works(self, tmp_path): assert tool.arguments == {"path": "main.py"} assert tool.result == "def main(): pass" + def test_mcp_function_call_is_strictly_classified(self, tmp_path): + jsonl_file = tmp_path / "rollout-mcp.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "mcp-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "mcp-call", + "name": "mcp__sample-server__lookup_item", + "arguments": {"item": "example"}, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.tool_name == "mcp__sample-server__lookup_item" + assert tool.tool_type == "mcp_tool" + assert tool.server_name == "sample-server" + + @pytest.mark.parametrize( + "name", + [ + "mcp__sample-server", + "mcp____lookup_item", + "mcp__sample-server__", + "mcp__sample-server__lookup_item__extra", + "mcp__sample server__lookup_item", + "prefix__sample-server__lookup_item", + ], + ) + def test_malformed_mcp_names_keep_their_classic_type(self, name): + assert CodexParser._classify_tool(name, "custom_tool_call") == ("custom_tool_call", None) + + def test_tool_arguments_are_normalized_and_recursively_bounded(self): + parser = CodexParser() + nested = {"leaf": "kept"} + for _ in range(12): + nested = {"next": nested} + + arguments = parser._parse_tool_arguments( + { + "items": list(range(150)), + "mapping": {f"key-{index}": index for index in range(150)}, + "long": "x" * 5000, + "nested": nested, + } + ) + + assert len(arguments["items"]) == 100 + assert len(arguments["mapping"]) == 100 + assert len(arguments["long"]) < 1200 + assert "[truncated" in arguments["long"] + assert "maximum depth" in json.dumps(arguments["nested"]) + assert parser._parse_tool_arguments('[1, {"key": "value"}]') == { + "raw": [1, {"key": "value"}] + } + assert parser._parse_tool_arguments("") == {} + assert parser._parse_tool_arguments(None) == {} + assert parser._parse_tool_arguments("false") == {"raw": False} + assert parser._parse_tool_arguments(7) == {"raw": 7} + assert parser._parse_tool_arguments('"text"') == {"raw": "text"} + + def test_oversized_json_arguments_are_not_decoded(self): + raw_arguments = json.dumps({"value": "x" * 100_100}) + + arguments = CodexParser()._parse_tool_arguments(raw_arguments) + + assert list(arguments) == ["raw"] + assert isinstance(arguments["raw"], str) + assert len(arguments["raw"]) < 1200 + assert "[truncated" in arguments["raw"] + + def test_output_correlates_across_id_aliases_and_flattens_content(self, tmp_path): + jsonl_file = tmp_path / "rollout-alias.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "alias-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "id": "shared-id", + "name": "lookup", + "arguments": [], + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call_output", + "toolCallId": "shared-id", + "output": { + "content": [ + {"type": "text", "text": "first"}, + {"content": "second"}, + ] + }, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.arguments == {"raw": []} + assert tool.result == "first\nsecond" + assert tool.status == "success" + assert tool.error is None + + @pytest.mark.parametrize( + ("output", "expected_status", "expected_result", "expected_error"), + [ + ({"Ok": {"content": [{"text": "complete"}]}}, "success", "complete", None), + ( + {"Ok": {"isError": True, "content": [{"text": "not completed"}]}}, + "error", + "not completed", + "not completed", + ), + (json.dumps({"Err": {"message": "rejected"}}), "error", "rejected", "rejected"), + ( + {"isError": True, "content": [{"type": "text", "text": "not completed"}]}, + "error", + "not completed", + "not completed", + ), + ({"is_error": False, "content": "complete"}, "success", "complete", None), + ({"status": "failed", "message": "not completed"}, "error", "not completed", "not completed"), + ({"state": "in_progress", "message": "waiting"}, "pending", "waiting", None), + ({"state": "completed", "content": "finished"}, "success", "finished", None), + ({"exit_code": 3, "stderr": "process detail"}, "error", "process detail", "Exit code: 3"), + ({"exitCode": 0, "stdout": "complete"}, "success", "complete", None), + ("ERROR: ordinary returned text", "success", "ERROR: ordinary returned text", None), + ], + ) + def test_tool_output_uses_structural_outcome_signals( + self, tmp_path, output, expected_status, expected_result, expected_error + ): + jsonl_file = tmp_path / "rollout-outcome.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "outcome-session"}}, + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "callId": "outcome-call", + "name": "run_task", + "input": {"value": 1}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "custom_tool_call_output", + "tool_call_id": "outcome-call", + "output": output, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.status == expected_status + assert tool.result == expected_result + assert tool.error == expected_error + + def test_nested_domain_status_does_not_mark_tool_as_failed(self, tmp_path): + jsonl_file = tmp_path / "rollout-domain-status.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "domain-status-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "domain-status-call", + "name": "lookup_record", + "arguments": {}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call_output", + "call_id": "domain-status-call", + "output": { + "content": [ + {"type": "record", "id": "sample-record", "status": "failed"} + ] + }, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.status == "success" + assert tool.result == '{"type":"record","id":"sample-record","status":"failed"}' + assert tool.error is None + + def test_unknown_typed_result_item_is_retained(self): + result = CodexParser()._normalize_tool_output( + [{"type": "artifact", "path": "sample.txt", "state": "ready", "message": "created"}] + ) + + assert result == '{"type":"artifact","path":"sample.txt","state":"ready","message":"created"}' + + @pytest.mark.parametrize( + "output", + [ + {"type": "artifact", "path": "sample.txt", "message": "created"}, + {"content": {"type": "artifact", "path": "sample.txt", "message": "created"}}, + ], + ) + def test_unknown_typed_result_mapping_is_retained_outside_lists(self, output): + result = CodexParser()._normalize_tool_output(output) + + assert result == '{"type":"artifact","path":"sample.txt","message":"created"}' + + def test_explicit_mcp_namespace_classifies_plain_tool_name(self, tmp_path): + jsonl_file = tmp_path / "rollout-mcp-namespace.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "mcp-namespace-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "mcp-call", + "namespace": "mcp__queryrunner_mcp", + "name": "run_query", + "arguments": {"query": "SELECT 1"}, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.tool_type == "mcp_tool" + assert tool.server_name == "queryrunner_mcp" + assert tool.tool_name == "run_query" + + def test_nested_outcome_like_fields_remain_domain_data(self, tmp_path): + jsonl_file = tmp_path / "rollout-nested-domain-outcome.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "nested-domain-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "nested-domain-call", + "name": "lookup_record", + "arguments": {}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call_output", + "call_id": "nested-domain-call", + "output": { + "isError": False, + "content": [ + { + "type": "record", + "result": {"status": "failed", "error": "domain value"}, + "exit_code": 7, + } + ], + }, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.status == "success" + assert '"status":"failed"' in tool.result + assert tool.error is None + + @pytest.mark.parametrize( + ("status", "expected"), + [("completed", "success"), ("failed", "error"), ("running", "pending")], + ) + def test_call_status_is_canonicalized_without_output(self, tmp_path, status, expected): + jsonl_file = tmp_path / f"rollout-{status}.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": f"{status}-session"}}, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": f"{status}-call", + "name": "sample_tool", + "arguments": {}, + "status": status, + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + tool = CodexParser().parse_jsonl_file(jsonl_file).chat_history[0].tools[0] + + assert tool.status == expected + assert tool.result is None + def test_mixed_tool_types_in_one_session(self, tmp_path): """Both record shapes can appear in the same session and must both survive.""" jsonl_file = tmp_path / "rollout-mixed.jsonl" @@ -427,6 +737,158 @@ def test_event_msg_records_are_ignored(self, tmp_path): assert len(entry.chat_history) == 2 # user message + assistant tool turn assert len([t for m in entry.chat_history for t in m.tools]) == 1 + def test_skips_malformed_decoded_records(self, tmp_path): + jsonl_file = tmp_path / "rollout-malformed.jsonl" + records = [ + None, + ["unsupported-record"], + {"type": "response_item"}, + {"type": "response_item", "payload": "unsupported-payload"}, + { + "type": "session_meta", + "payload": {"id": {"unexpected": "value"}, "timestamp": {"unexpected": "value"}}, + }, + { + "type": "session_meta", + "payload": {"id": "valid-session", "timestamp": "2025-01-02T03:04:05Z"}, + }, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "kept message"}, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(record) for record in records)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert entry is not None + assert entry.session_id == "codex_valid-session" + assert [message.content for message in entry.chat_history] == ["kept message"] + + def test_invalid_session_timestamp_keeps_valid_session_identity(self, tmp_path): + jsonl_file = tmp_path / "rollout-invalid-timestamp.jsonl" + events = [ + { + "type": "session_meta", + "payload": {"id": "valid-session", "timestamp": {"unexpected": "value"}}, + }, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "kept message"}, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert entry is not None + assert entry.session_id == "codex_valid-session" + assert [message.content for message in entry.chat_history] == ["kept message"] + + def test_supports_message_content_shapes(self, tmp_path): + jsonl_file = tmp_path / "rollout-content.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "content-session"}}, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "string content"}, + }, + { + "type": "response_item", + "payload": { + "type": "message", + "role": "assistant", + "content": {"type": "output_text", "text": "mapping content"}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "message", + "role": "user", + "content": [ + "mixed ", + {"type": "input_text", "text": "content"}, + None, + 7, + {"type": "input_text", "text": {"unexpected": "value"}}, + {"type": "image", "text": "ignored"}, + {"type": "output_text", "text": " list"}, + ], + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert [(message.role, message.content) for message in entry.chat_history] == [ + ("user", "string content"), + ("assistant", "mapping content"), + ("user", "mixed content list"), + ] + + def test_tolerates_malformed_reasoning_summary_items(self, tmp_path): + jsonl_file = tmp_path / "rollout-reasoning.jsonl" + events = [ + {"type": "session_meta", "payload": {"id": "reasoning-session"}}, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "review this"}, + }, + { + "type": "response_item", + "payload": { + "type": "reasoning", + "summary": [ + None, + "unsupported-item", + {"type": "summary_text", "text": {"unexpected": "value"}}, + {"type": "other", "text": "ignored"}, + {"type": "summary_text", "text": "valid summary"}, + ], + }, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert [message.content for message in entry.chat_history] == [ + "review this", + "[Reasoning]\nvalid summary\n", + ] + + def test_first_valid_session_meta_defines_physical_identity(self, tmp_path): + jsonl_file = tmp_path / "rollout-identity.jsonl" + first_cwd = str(tmp_path / "first-project") + later_cwd = str(tmp_path / "later-project") + events = [ + { + "type": "session_meta", + "payload": {"timestamp": "2024-01-01T00:00:00Z", "cwd": str(tmp_path / "missing-id")}, + }, + { + "type": "session_meta", + "payload": {"id": "physical-session", "timestamp": "2025-01-02T03:04:05Z", "cwd": first_cwd}, + }, + { + "type": "session_meta", + "payload": {"id": "later-session", "timestamp": "2026-02-03T04:05:06Z", "cwd": later_cwd}, + }, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "identity check"}, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert entry.session_id == "codex_physical-session" + assert entry.project_path == first_cwd + assert entry.timestamp == datetime(2025, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + def test_parse_no_directory(self): """Test parse_all when directory doesn't exist.""" parser = CodexParser()