From 5be35744cc9bf5254511a517877812d66140b4e5 Mon Sep 17 00:00:00 2001 From: bozbas Date: Mon, 24 Aug 2026 06:58:27 +0000 Subject: [PATCH] feat(sensor): capture rich Codex activity and provenance Summary: Intent: - Preserve completed command, file, MCP, and multi-agent activity from current Codex events. - Retain bounded session lineage and agent context. Changes: - Parse rich activity records and reconcile them with classic tool records only within safe boundaries. - Capture agent lifecycle events, session provenance, lineage, and direct role metadata. --- Sensor/README.md | 27 + Sensor/adr_sensor/parsers/codex_parser.py | 991 ++++++++++++- Sensor/tests/test_parsers.py | 1574 ++++++++++++++++++++- 3 files changed, 2548 insertions(+), 44 deletions(-) diff --git a/Sensor/README.md b/Sensor/README.md index bd81ecf..80b6c60 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -233,6 +233,33 @@ agent mode populates the richest version: } ``` +Codex emits bounded provenance as optional, flat keys in the same object: + +```json +{ + "session_context": { + "originator": "client-name", + "cli_version": "1.2.3", + "model_provider": "provider-name", + "git_branch": "feature/example", + "parent_thread_id": "parent-id", + "forked_from_id": "fork-id", + "agent_path": "root/worker", + "agent_nickname": "worker-name", + "subagent_history_start_ordinal": 4, + "thread_source": "subagent", + "agent_depth": 1, + "agent_role": "reviewer", + "root_session_id": "root-id", + "inherited_session_ids": ["inherited-id"] + } +} +``` + +Only scalar Codex metadata is retained, and long strings are middle-truncated. +Later `session_meta` records contribute unique inherited IDs without changing the +physical session's `source`, `session_id`, `project_path`, or `user_id`. + ## Adding a New Parser ADR Sensor is designed to be extensible. To add support for a new AI agent: diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 03e1e86..2332ffe 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -3,6 +3,7 @@ Reads JSONL files from ~/.codex/sessions/ """ +import hashlib import json import re import traceback @@ -20,6 +21,7 @@ MAX_NORMALIZATION_DEPTH = 8 MAX_COLLECTION_ITEMS = 100 MAX_TEXT_LENGTH = 1000 +MAX_AGENT_PARTY_LENGTH = 100 _DEPTH_LIMIT_MARKER = "[truncated: maximum depth reached]" _MCP_NAME_PART = re.compile(r"^[A-Za-z0-9_-]+$") _CALL_ID_KEYS = ( @@ -33,13 +35,35 @@ ) _SUCCESS_STATES = frozenset({"success", "succeeded", "completed", "complete", "done", "ok"}) _ERROR_STATES = frozenset( - {"error", "failed", "failure", "cancelled", "canceled", "rejected", "timed_out", "timeout"} + { + "error", + "failed", + "failure", + "cancelled", + "canceled", + "declined", + "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"}) +_SESSION_CONTEXT_FIELDS = ( + "originator", + "cli_version", + "model_provider", + "parent_thread_id", + "forked_from_id", + "agent_path", + "agent_nickname", + "agent_role", + "subagent_history_start_ordinal", + "thread_source", +) class CodexParser(BaseParser): @@ -79,6 +103,16 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: "model": None, "messages": [], "pending_tool_calls": {}, + "_event_index": 0, + "_action_boundaries": [], + "_custom_tool_wrappers": [], + "_rich_actions": [], + "_rich_mcp_calls": [], + "_classic_mcp_calls": [], + "_tool_records": {}, + "_current_turn_id": None, + "_previous_trigger_turn": False, + "session_context": {}, } with open(file_path, encoding="utf-8") as file: @@ -89,18 +123,24 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: try: event = json.loads(line) if not isinstance(event, Mapping): + session_data["_previous_trigger_turn"] = False continue payload = event.get("payload") if not isinstance(payload, Mapping): + session_data["_previous_trigger_turn"] = False 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. + session_data["_previous_trigger_turn"] = False continue + self._reconcile_rich_actions(session_data) + self._reconcile_rich_mcp_calls(session_data) + if not session_data["id"]: return None @@ -138,6 +178,7 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: model=session_data["model"], chat_history=chat_history, raw_log_path=str(file_path), + session_context=session_data["session_context"] or None, ) except Exception as e: @@ -310,7 +351,8 @@ def _call_ids(payload: Mapping[str, Any]) -> List[str]: def _canonical_status(value: Any) -> Optional[str]: if not isinstance(value, str): return None - normalized = value.strip().lower().replace("-", "_").replace(" ", "_") + normalized = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", value.strip()) + normalized = normalized.lower().replace("-", "_").replace(" ", "_") if normalized in _SUCCESS_STATES: return "success" if normalized in _ERROR_STATES: @@ -375,14 +417,760 @@ def _infer_tool_outcome(self, sources: List[Any], default: str) -> Tuple[str, An return "pending", None return default, None + @staticmethod + def _wire_token(value: Any) -> str: + """Normalize wire enum aliases without accepting unrelated item kinds.""" + return re.sub(r"[^a-z0-9]", "", value.lower()) if isinstance(value, str) else "" + + @staticmethod + def _agent_message_text(content: Any) -> str: + """Extract only locally readable agent-message text.""" + if not isinstance(content, list): + return "" + return "".join( + item["text"] + for item in content + if isinstance(item, Mapping) + and item.get("type") == "input_text" + and isinstance(item.get("text"), str) + ) + + @staticmethod + def _format_agent_party(value: Any) -> str: + """Render a bounded scalar as a JSON value for the metadata prefix.""" + if not isinstance(value, (str, bool, int, float)): + return "null" + bounded = truncate_middle(str(value), max_length=MAX_AGENT_PARTY_LENGTH, edge_chars=30) + return json.dumps(bounded, ensure_ascii=False) + + @staticmethod + def _first_value(value: Mapping[str, Any], keys: Tuple[str, ...]) -> Any: + for key in keys: + if key in value: + return value[key] + return None + + def _event_turn_id(self, *sources: Any) -> Optional[str]: + """Read an optional turn ID from direct fields or response metadata.""" + for source in sources: + if not isinstance(source, Mapping): + continue + + turn_id = self._first_value(source, ("turn_id", "turnId", "turnID")) + if isinstance(turn_id, (str, int)) and not isinstance(turn_id, bool): + normalized = str(turn_id).strip() + if normalized: + return truncate_middle(normalized, max_length=MAX_TEXT_LENGTH, edge_chars=400) + + metadata = self._first_value( + source, + ( + "internal_chat_message_metadata_passthrough", + "internalChatMessageMetadataPassthrough", + ), + ) + if isinstance(metadata, Mapping): + turn_id = self._first_value(metadata, ("turn_id", "turnId", "turnID")) + if isinstance(turn_id, (str, int)) and not isinstance(turn_id, bool): + normalized = str(turn_id).strip() + if normalized: + return truncate_middle(normalized, max_length=MAX_TEXT_LENGTH, edge_chars=400) + return None + + def _normalize_command(self, value: Any) -> Any: + """Retain command strings or argv lists in their source shape, with bounds.""" + if isinstance(value, str): + return truncate_middle(value, max_length=MAX_TEXT_LENGTH, edge_chars=400) + if isinstance(value, (list, tuple)): + return [ + truncate_middle(part, max_length=MAX_TEXT_LENGTH, edge_chars=400) + for part in value[:MAX_COLLECTION_ITEMS] + if isinstance(part, str) + ] + return None + + @staticmethod + def _is_plain_command_text(value: str) -> bool: + """Keep signatures conservative; opaque scripts are intentionally not parsed.""" + stripped = value.strip() + if not stripped or len(stripped) > MAX_TEXT_LENGTH or "\n" in stripped or "\r" in stripped: + return False + return not re.match(r"^(?:async\s+)?(?:const|function|let|var)\b", stripped) + + def _command_signatures(self, value: Any, allow_plain_text: bool = False) -> Tuple[str, ...]: + """Build bounded command identities used only for wrapper reconciliation.""" + if isinstance(value, str) and len(value) <= MAX_ARGUMENT_JSON_LENGTH: + try: + value = json.loads(value) + except (json.JSONDecodeError, RecursionError, ValueError): + pass + + if isinstance(value, Mapping): + value = self._first_value(value, ("command", "cmd", "argv")) + elif isinstance(value, str) and not allow_plain_text: + return () + + if isinstance(value, str): + if not self._is_plain_command_text(value): + return () + command = truncate_middle(value.strip(), max_length=MAX_TEXT_LENGTH, edge_chars=400) + return (command,) + + command = self._normalize_command(value) + if not isinstance(command, list) or not command: + return () + + signatures = [ + truncate_middle( + "\0".join(command), + max_length=MAX_TEXT_LENGTH, + edge_chars=400, + ) + ] + executable = command[0].replace("\\", "/").rsplit("/", 1)[-1].lower() + if len(command) >= 3 and executable in {"bash", "cmd", "dash", "powershell", "pwsh", "sh", "zsh"}: + option = command[-2].lower() + if option == "/c" or option in {"-c", "-command"} or ( + option.startswith("-") and "c" in option[1:] + ): + nested = command[-1].strip() + if self._is_plain_command_text(nested): + signatures.append( + truncate_middle(nested, max_length=MAX_TEXT_LENGTH, edge_chars=400) + ) + return tuple(dict.fromkeys(signatures)) + + @staticmethod + def _wrapper_action_kinds(tool_name: Any) -> Tuple[str, ...]: + token = CodexParser._wire_token(tool_name) + if token in {"exec", "execute"}: + return ("command", "file") + if token in {"commandexecution", "execcommand", "runcommand", "shell", "shellcommand"}: + return ("command",) + if token in {"applypatch", "filechange", "patch"}: + return ("file",) + return () + + def _rich_output(self, item: Mapping[str, Any]) -> Any: + """Select the richest completed output without interpreting its prose.""" + for keys in ( + ("aggregated_output", "aggregatedOutput"), + ("formatted_output", "formattedOutput"), + ("output", "result", "content"), + ): + output = self._first_value(item, keys) + if output not in (None, ""): + return output + + streams = {} + for canonical, aliases in ( + ("stdout", ("stdout", "standard_output", "standardOutput")), + ("stderr", ("stderr", "standard_error", "standardError")), + ): + stream = self._first_value(item, aliases) + if stream not in (None, ""): + streams[canonical] = stream + return streams or None + + def _rich_outcome(self, item: Mapping[str, Any], output: Any) -> Tuple[str, Optional[str], Optional[str]]: + """Apply the structural outcome rules shared with classic tool results.""" + outcome_source = dict(item) + for canonical, aliases in ( + ("status", ("status", "state")), + ("exit_code", ("exit_code", "exitCode", "exit_status", "exitStatus")), + ("error", ("error", "error_message", "errorMessage")), + ): + if canonical not in outcome_source: + alias_value = self._first_value(item, aliases) + if alias_value is not None: + outcome_source[canonical] = alias_value + + structured_output = self._decode_output_container(output) + result = self._normalize_tool_output(structured_output) + status, error_detail = self._infer_tool_outcome( + [outcome_source, structured_output], default="success" + ) + error = self._normalize_tool_output(error_detail) if error_detail is not None else None + if status == "error": + error = error or result + else: + error = None + return status, result, error + + def _normalize_rich_command( + self, item: Mapping[str, Any] + ) -> Optional[Tuple[Dict[str, Any], Tuple[str, ...]]]: + raw_command = self._first_value(item, ("command", "cmd", "argv")) + command = self._normalize_command(raw_command) + if command is None or not command or (isinstance(command, str) and not command.strip()): + return None + + arguments: Dict[str, Any] = {"command": command} + cwd = self._first_value(item, ("cwd", "workdir", "working_directory", "workingDirectory")) + if isinstance(cwd, str): + arguments["cwd"] = truncate_middle(cwd, max_length=MAX_TEXT_LENGTH, edge_chars=400) + + duration = self._first_value(item, ("duration", "duration_ms", "durationMs")) + if duration is not None: + arguments["_codex"] = {"duration": self._bound_value(duration)} + + status, result, error = self._rich_outcome(item, self._rich_output(item)) + return ( + { + "tool_name": "exec_command", + "tool_type": "function_call", + "server_name": None, + "arguments": arguments, + "status": status, + "result": result, + "error": error, + }, + self._command_signatures(raw_command, allow_plain_text=True), + ) + + @staticmethod + def _change_type(value: Any) -> Optional[str]: + if not isinstance(value, str) or not value.strip(): + return None + token = CodexParser._wire_token(value) + aliases = { + "add": "add", + "added": "add", + "create": "add", + "created": "add", + "delete": "delete", + "deleted": "delete", + "remove": "delete", + "removed": "delete", + "modify": "update", + "modified": "update", + "update": "update", + "updated": "update", + } + return aliases.get(token, truncate_middle(value.strip().lower(), max_length=MAX_TEXT_LENGTH, edge_chars=400)) + + @staticmethod + def _private_text_summary(value: Any) -> Optional[Dict[str, Any]]: + if not isinstance(value, str): + return None + encoded = value.encode("utf-8") + return {"utf8_bytes": len(encoded), "sha256": hashlib.sha256(encoded).hexdigest()} + + def _summarize_file_changes(self, raw_changes: Any) -> Optional[Tuple[List[Dict[str, Any]], int]]: + """Build a bounded allowlisted view of file changes, hashing private bodies.""" + if isinstance(raw_changes, Mapping): + singular_path = self._first_value(raw_changes, ("path", "file_path", "filePath")) + if isinstance(singular_path, str) and any( + key in raw_changes for key in ("type", "change_type", "changeType", "kind") + ): + entries = [ + ( + singular_path, + raw_changes, + ) + ] + total_changes = 1 + else: + total_changes = len(raw_changes) + entries = [] + for index, entry in enumerate(raw_changes.items()): + if index >= MAX_COLLECTION_ITEMS: + break + entries.append(entry) + elif isinstance(raw_changes, (list, tuple)): + total_changes = len(raw_changes) + entries = [] + for change in raw_changes[:MAX_COLLECTION_ITEMS]: + if isinstance(change, Mapping): + entries.append( + ( + self._first_value(change, ("path", "file_path", "filePath")), + change, + ) + ) + else: + return None + + summaries = [] + for path, change in entries: + if not isinstance(path, str) or not isinstance(change, Mapping): + continue + change_type = self._change_type( + self._first_value(change, ("type", "change_type", "changeType", "kind")) + ) + if change_type is None: + continue + + summary: Dict[str, Any] = { + "path": truncate_middle(path, max_length=MAX_TEXT_LENGTH, edge_chars=400), + "type": change_type, + } + move_path = self._first_value( + change, + ("move_path", "movePath", "new_path", "newPath", "destination"), + ) + if isinstance(move_path, str): + summary["move_path"] = truncate_middle( + move_path, max_length=MAX_TEXT_LENGTH, edge_chars=400 + ) + + for canonical, aliases in ( + ("content", ("content", "file_content", "fileContent")), + ("unified_diff", ("unified_diff", "unifiedDiff", "diff")), + ): + private_summary = self._private_text_summary(self._first_value(change, aliases)) + if private_summary is not None: + summary[canonical] = private_summary + summaries.append(summary) + + return summaries, max(0, total_changes - MAX_COLLECTION_ITEMS) + + def _normalize_rich_file_change(self, item: Mapping[str, Any]) -> Optional[Dict[str, Any]]: + summarized = self._summarize_file_changes( + self._first_value(item, ("changes", "file_changes", "fileChanges")) + ) + if summarized is None: + return None + + changes, truncated_changes = summarized + if not changes: + return None + arguments: Dict[str, Any] = {"changes": changes} + if truncated_changes: + arguments["_codex"] = {"truncated_changes": truncated_changes} + + outcome_item = dict(item) + for key in ("changes", "file_changes", "fileChanges"): + outcome_item.pop(key, None) + status, result, error = self._rich_outcome(outcome_item, self._rich_output(item)) + return { + "tool_name": "file_change", + "tool_type": "function_call", + "server_name": None, + "arguments": arguments, + "status": status, + "result": result, + "error": error, + } + + @staticmethod + def _first_present( + sources: Tuple[Mapping[str, Any], ...], keys: Tuple[str, ...] + ) -> Tuple[bool, Any]: + for source in sources: + for key in keys: + if key in source: + return True, source[key] + return False, None + + def _first_nonempty_string( + self, sources: Tuple[Mapping[str, Any], ...], keys: Tuple[str, ...] + ) -> Optional[str]: + for source in sources: + for key in keys: + value = source.get(key) + if isinstance(value, str) and value.strip(): + return truncate_middle( + value.strip(), max_length=MAX_TEXT_LENGTH, edge_chars=400 + ) + return None + + @staticmethod + def _codex_metadata(arguments: Dict[str, Any]) -> Dict[str, Any]: + metadata = arguments.get("_codex") + if not isinstance(metadata, dict): + metadata = {} + arguments["_codex"] = metadata + return metadata + + def _normalize_rich_mcp_call( + self, record: Mapping[str, Any] + ) -> Optional[Tuple[Dict[str, Any], Tuple[str, ...]]]: + """Normalize a completed rich MCP record without relying on server identity.""" + invocation = None + server_name = None + tool_name = None + for key in ("invocation", "tool_call", "toolCall", "call"): + candidate = record.get(key) + if not isinstance(candidate, Mapping): + continue + candidate_server = self._first_nonempty_string( + (candidate,), + ("server", "server_name", "serverName", "mcp_server", "mcpServer"), + ) + candidate_tool = self._first_nonempty_string( + (candidate,), ("tool", "tool_name", "toolName", "name") + ) + if candidate_server and candidate_tool: + invocation = candidate + server_name = candidate_server + tool_name = candidate_tool + break + + if invocation is None: + invocation = record + server_name = self._first_nonempty_string( + (record,), + ("server", "server_name", "serverName", "mcp_server", "mcpServer"), + ) + tool_name = self._first_nonempty_string( + (record,), ("tool", "tool_name", "toolName", "name") + ) + if not server_name or not tool_name: + return None + + _, raw_arguments = self._first_present( + (invocation,), + ("arguments", "args", "input", "params", "parameters"), + ) + arguments = self._parse_tool_arguments(raw_arguments) + + if self._wire_token(tool_name) == "invoketool": + effective_server = self._first_nonempty_string( + (arguments,), + ( + "server", + "server_name", + "serverName", + "mcp_server", + "mcpServer", + "target_server", + "targetServer", + ), + ) + effective_tool = self._first_nonempty_string( + (arguments,), + ("tool", "tool_name", "toolName", "name", "target_tool", "targetTool"), + ) + has_nested_arguments, nested_arguments = self._first_present( + (arguments,), + ("arguments", "args", "input", "params", "parameters"), + ) + if effective_server and effective_tool and has_nested_arguments: + outer_server, outer_tool = server_name, tool_name + server_name, tool_name = effective_server, effective_tool + arguments = self._parse_tool_arguments(nested_arguments) + arguments["_codex_mcp_wrapper"] = { + "server_name": outer_server, + "tool_name": outer_tool, + } + + metadata: Dict[str, Any] = {} + sources = (record, invocation) if invocation is not record else (record,) + duration = next( + ( + source[key] + for source in sources + for key in ("duration", "duration_ms", "durationMs", "elapsed_ms", "elapsedMs") + if key in source and source[key] is not None + ), + None, + ) + if duration is not None: + metadata["duration"] = self._bound_value(duration) + + read_only_hint = next( + ( + source[key] + for source in sources + for key in ("read_only_hint", "readOnlyHint") + if isinstance(source.get(key), bool) + ), + None, + ) + if isinstance(read_only_hint, bool): + metadata["read_only_hint"] = read_only_hint + + connector = {} + for canonical, aliases in ( + ("connector_id", ("connector_id", "connectorId")), + ("link_id", ("link_id", "linkId")), + ("app_name", ("app_name", "appName")), + ("action_name", ("action_name", "actionName")), + ): + value = next( + ( + source[key] + for source in sources + for key in aliases + if isinstance(source.get(key), (str, bool, int, float)) + ), + None, + ) + if value is not None: + connector[canonical] = self._bound_value(value) + if connector: + metadata["connector"] = connector + if metadata: + self._codex_metadata(arguments).update(metadata) + + output = self._rich_output(record) + if output is None: + _, output = self._first_present((record,), ("response",)) + if output is None and invocation is not record: + output = self._rich_output(invocation) + if output is None: + _, output = self._first_present((invocation,), ("response",)) + outcome_record = dict(invocation) + outcome_record.update(record) + status, result, error = self._rich_outcome(outcome_record, output) + + call_ids = [] + for source in (record, invocation): + for call_id in self._call_ids(source): + if call_id not in call_ids: + call_ids.append(call_id) + return ( + { + "tool_name": tool_name, + "tool_type": "mcp_tool", + "server_name": server_name, + "arguments": arguments, + "status": status, + "result": result, + "error": error, + }, + tuple(call_ids), + ) + + def _normalize_rich_action( + self, item: Mapping[str, Any] + ) -> Optional[Tuple[str, Dict[str, Any], Tuple[str, ...]]]: + item_type = self._wire_token(item.get("type")) + if item_type == "commandexecution": + command = self._normalize_rich_command(item) + if command is None: + return None + tool, signatures = command + return "command", tool, signatures + if item_type == "filechange": + tool = self._normalize_rich_file_change(item) + return ("file", tool, ()) if tool is not None else None + return None + + def _process_subagent_activity( + self, item: Mapping[str, Any], session_data: Dict[str, Any] + ) -> None: + activity = {} + for key in ("kind", "agent_thread_id", "agent_path"): + value = item.get(key) + if isinstance(value, (str, bool, int, float)): + activity[key] = self._bound_value(value) + + tool_dict = None + call_ids = self._call_ids(item) + for key in ("event_id", "eventId", "eventID"): + value = item.get(key) + if isinstance(value, (str, int)) and not isinstance(value, bool): + call_id = str(value).strip() + if call_id and call_id not in call_ids: + call_ids.append(call_id) + + for call_id in call_ids: + candidate = session_data["pending_tool_calls"].get(call_id) + tool_record = session_data["_tool_records"].get(id(candidate)) + if tool_record is not None and tool_record.get("item_type") == "function_call": + tool_dict = candidate + break + + if tool_dict is not None: + self._codex_metadata(tool_dict["arguments"])["subagent_activity"] = activity + if tool_dict.get("status") == "pending": + tool_dict["status"] = "success" + return + + self._append_tool( + session_data, + { + "tool_name": "subagent_activity", + "tool_type": "function_call", + "server_name": None, + "arguments": {"_codex": {"subagent_activity": activity}}, + "status": "success", + "result": None, + "error": None, + }, + ) + + @staticmethod + def _append_tool(session_data: Dict[str, Any], tool: Dict[str, Any]) -> Dict[str, Any]: + if not session_data["messages"] or session_data["messages"][-1]["role"] != "assistant": + session_data["messages"].append({"role": "assistant", "content": "", "tools": []}) + message = session_data["messages"][-1] + message["tools"].append(tool) + return message + + @staticmethod + def _turns_are_compatible(left: Optional[str], right: Optional[str]) -> bool: + return not left or not right or left == right + + def _reconcile_rich_actions(self, session_data: Dict[str, Any]) -> None: + """Replace confidently matched custom wrappers with completed rich actions.""" + wrappers = session_data.get("_custom_tool_wrappers", []) + rich_actions = session_data.get("_rich_actions", []) + if not wrappers or not rich_actions: + return + + boundaries = sorted(set(session_data.get("_action_boundaries", []))) + for wrapper in wrappers: + wrapper["window_end"] = next( + (index for index in boundaries if index > wrapper["index"]), + float("inf"), + ) + wrapper["matched"] = [] + + for rich_action in sorted(rich_actions, key=lambda action: action["index"]): + candidates = [] + for wrapper in wrappers: + if not (wrapper["index"] < rich_action["index"] < wrapper["window_end"]): + continue + if rich_action["kind"] not in wrapper["kinds"]: + continue + if not self._turns_are_compatible(wrapper["turn_id"], rich_action["turn_id"]): + continue + if ( + rich_action["kind"] == "command" + and wrapper["signatures"] + and rich_action["signatures"] + and not set(wrapper["signatures"]).intersection(rich_action["signatures"]) + ): + continue + candidates.append(wrapper) + + if candidates: + max(candidates, key=lambda wrapper: wrapper["index"])["matched"].append(rich_action) + + matched_actions = [ + action for wrapper in wrappers for action in wrapper["matched"] + ] + if not matched_actions: + return + + matched_tool_ids = {id(action["tool"]) for action in matched_actions} + for message in session_data["messages"]: + message["tools"] = [ + tool for tool in message["tools"] if id(tool) not in matched_tool_ids + ] + + for wrapper in wrappers: + if not wrapper["matched"]: + continue + tools = wrapper["message"]["tools"] + wrapper_index = next( + (index for index, tool in enumerate(tools) if tool is wrapper["tool"]), + None, + ) + if wrapper_index is None: + continue + replacements = [ + action["tool"] for action in sorted(wrapper["matched"], key=lambda action: action["index"]) + ] + tools[wrapper_index : wrapper_index + 1] = replacements + + def _mcp_signature(self, tool: Mapping[str, Any]) -> Tuple[str, str, str]: + """Build a stable signature for classic and rich forms of one MCP call.""" + server_name = tool.get("server_name") + tool_name = tool.get("tool_name") + if isinstance(tool_name, str): + parts = tool_name.split("__") + if len(parts) == 3 and parts[0] == "mcp": + tool_name = parts[2] + + arguments = tool.get("arguments") + if isinstance(arguments, Mapping): + arguments = { + key: value + for key, value in arguments.items() + if not str(key).startswith("_codex") + } + try: + encoded_arguments = json.dumps(arguments, sort_keys=True, separators=(",", ":")) + except (TypeError, ValueError): + encoded_arguments = str(arguments) + + return ( + server_name.strip() if isinstance(server_name, str) else "", + tool_name.strip() if isinstance(tool_name, str) else "", + encoded_arguments, + ) + + def _reconcile_rich_mcp_calls(self, session_data: Dict[str, Any]) -> None: + """Replace classic MCP calls matched by ID or an exact call signature.""" + classic_calls = session_data.get("_classic_mcp_calls", []) + rich_calls = session_data.get("_rich_mcp_calls", []) + if not classic_calls or not rich_calls: + return + + boundaries = sorted(set(session_data.get("_action_boundaries", []))) + for classic_call in classic_calls: + classic_call["window_end"] = next( + (index for index in boundaries if index > classic_call["index"]), + float("inf"), + ) + + matched_rich_tool_ids = set() + matched_classic_tool_ids = set() + for rich_call in sorted(rich_calls, key=lambda call: call["index"]): + rich_ids = set(rich_call["call_ids"]) + candidates = [ + classic_call + for classic_call in classic_calls + if id(classic_call["tool"]) not in matched_classic_tool_ids + and rich_ids.intersection(classic_call["call_ids"]) + ] + if not candidates: + rich_signature = self._mcp_signature(rich_call["tool"]) + candidates = [ + classic_call + for classic_call in classic_calls + if id(classic_call["tool"]) not in matched_classic_tool_ids + and (not rich_ids or not classic_call["call_ids"]) + and classic_call["index"] <= rich_call["index"] + and rich_call["index"] < classic_call["window_end"] + and self._turns_are_compatible( + classic_call.get("turn_id"), rich_call.get("turn_id") + ) + and self._mcp_signature(classic_call["tool"]) == rich_signature + ] + if not candidates: + continue + + preceding = [ + candidate + for candidate in candidates + if candidate["index"] <= rich_call["index"] + ] + if preceding: + classic_call = max(preceding, key=lambda call: call["index"]) + else: + classic_call = min(candidates, key=lambda call: call["index"]) + classic_call["tool"].clear() + classic_call["tool"].update(rich_call["tool"]) + matched_classic_tool_ids.add(id(classic_call["tool"])) + matched_rich_tool_ids.add(id(rich_call["tool"])) + + if matched_rich_tool_ids: + for message in session_data["messages"]: + message["tools"] = [ + tool for tool in message["tools"] if id(tool) not in matched_rich_tool_ids + ] + 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") + previous_trigger_turn = session_data.get("_previous_trigger_turn", False) + session_data["_previous_trigger_turn"] = bool( + evt_type == "inter_agent_communication_metadata" + and payload.get("trigger_turn") is True + ) + event_index = session_data.setdefault("_event_index", 0) + session_data["_event_index"] = event_index + 1 if evt_type == "session_meta": if session_data["id"] is not None: + self._add_inherited_session_id(payload.get("id"), session_data) return session_id = payload.get("id") @@ -400,11 +1188,20 @@ def _process_event( session_data["id"] = session_id session_data["timestamp"] = normalized_timestamp session_data["cwd"] = payload.get("cwd") + session_data["session_context"] = self._extract_session_context(payload, session_id) elif evt_type == "turn_context": + session_data["_action_boundaries"].append(event_index) + turn_id = self._event_turn_id(event, payload) + session_data["_current_turn_id"] = turn_id if payload.get("model"): session_data["model"] = payload.get("model") + elif self._wire_token(evt_type) == "subagentactivity": + activity = dict(event) + activity.update(payload) + self._process_subagent_activity(activity, session_data) + elif evt_type == "response_item": item_type = payload.get("type") @@ -436,7 +1233,22 @@ def _process_event( if text_content: session_data["messages"].append({"role": role, "content": text_content, "tools": []}) + elif item_type == "agent_message": + text_content = self._agent_message_text(payload.get("content")) + if text_content: + author = self._format_agent_party(payload.get("author")) + recipient = self._format_agent_party(payload.get("recipient")) + trigger_turn = str(previous_trigger_turn).lower() + content = ( + f"[agent_message author={author} recipient={recipient} " + f"trigger_turn={trigger_turn}]\n{text_content}" + ) + session_data["messages"].append( + {"role": "user", "content": content, "tools": []} + ) + elif item_type in ("function_call", "custom_tool_call"): + session_data["_action_boundaries"].append(event_index) tool_name = payload.get("name") # The two record shapes differ in where the arguments live and how @@ -456,6 +1268,7 @@ def _process_event( 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 + call_ids = self._call_ids(payload) tool_dict = { "tool_name": tool_name, "tool_type": tool_type, @@ -466,11 +1279,32 @@ def _process_event( "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) - for call_id in self._call_ids(payload): + message = self._append_tool(session_data, tool_dict) + turn_id = self._event_turn_id(event, payload) or session_data.get("_current_turn_id") + tool_record = { + "tool": tool_dict, + "message": message, + "index": event_index, + "turn_id": turn_id, + "call_ids": tuple(call_ids), + "item_type": item_type, + } + session_data["_tool_records"][id(tool_dict)] = tool_record + if item_type == "function_call" and tool_type == "mcp_tool": + session_data["_classic_mcp_calls"].append(tool_record) + if item_type == "custom_tool_call": + signatures = self._command_signatures(raw_arguments, allow_plain_text=True) + kinds = self._wrapper_action_kinds(tool_name) + if signatures and self._wire_token(tool_name) in {"exec", "execute"}: + kinds = ("command",) + tool_record.update( + { + "kinds": kinds, + "signatures": signatures, + } + ) + session_data["_custom_tool_wrappers"].append(tool_record) + for call_id in call_ids: session_data["pending_tool_calls"][call_id] = tool_dict elif item_type in ("function_call_output", "custom_tool_call_output"): @@ -483,6 +1317,13 @@ def _process_event( None, ) if tool_dict is not None: + tool_record = session_data["_tool_records"].get(id(tool_dict)) + if tool_record is not None: + tool_record["turn_id"] = ( + tool_record["turn_id"] + or self._event_turn_id(event, payload) + or session_data.get("_current_turn_id") + ) raw_output = next( (payload[key] for key in ("output", "result", "content") if key in payload), None, @@ -531,3 +1372,139 @@ def _process_event( ) else: session_data["messages"][-1]["content"] = "[Reasoning]\n" + reasoning_text + + elif self._wire_token(evt_type) == "eventmsg": + payload_type = self._wire_token(payload.get("type")) + if payload_type == "mcptoolcallend": + normalized_mcp = self._normalize_rich_mcp_call(payload) + if normalized_mcp is None: + return + tool_dict, call_ids = normalized_mcp + self._append_tool(session_data, tool_dict) + session_data["_rich_mcp_calls"].append( + { + "tool": tool_dict, + "index": event_index, + "call_ids": call_ids, + "turn_id": self._event_turn_id(event, payload) + or session_data.get("_current_turn_id"), + } + ) + return + + if payload_type != "itemcompleted": + return + item = payload.get("item") + if not isinstance(item, Mapping): + return + + item_type = self._wire_token(item.get("type")) + if item_type == "subagentactivity": + self._process_subagent_activity(item, session_data) + return + + if item_type in ("mcptoolcall", "mcptoolcallend"): + normalized_mcp = self._normalize_rich_mcp_call(item) + if normalized_mcp is None: + return + tool_dict, call_ids = normalized_mcp + self._append_tool(session_data, tool_dict) + session_data["_rich_mcp_calls"].append( + { + "tool": tool_dict, + "index": event_index, + "call_ids": call_ids, + "turn_id": self._event_turn_id(event, payload, item) + or session_data.get("_current_turn_id"), + } + ) + return + + normalized = self._normalize_rich_action(item) + if normalized is None: + return + + kind, tool_dict, signatures = normalized + self._append_tool(session_data, tool_dict) + session_data["_rich_actions"].append( + { + "tool": tool_dict, + "index": event_index, + "turn_id": self._event_turn_id(event, payload, item) + or session_data.get("_current_turn_id"), + "kind": kind, + "signatures": signatures, + } + ) + + @staticmethod + def _normalize_context_scalar(value: Any) -> Optional[Any]: + """Return a bounded JSON scalar, ignoring structured metadata values.""" + if not isinstance(value, (str, int, float, bool)): + return None + if isinstance(value, str): + return truncate_middle(value, max_length=1000, edge_chars=400) + return value + + def _extract_session_context( + self, payload: Mapping[str, Any], physical_session_id: str + ) -> Dict[str, Any]: + """Extract bounded provenance from the physical session metadata.""" + context: Dict[str, Any] = {} + + source = payload.get("source") + thread_spawn = None + if isinstance(source, Mapping): + subagent = source.get("subagent") + if isinstance(subagent, Mapping): + candidate = subagent.get("thread_spawn") + if isinstance(candidate, Mapping): + thread_spawn = candidate + + for key in _SESSION_CONTEXT_FIELDS: + value = self._normalize_context_scalar(payload.get(key)) + if value is None and thread_spawn is not None and key in ( + "parent_thread_id", + "agent_path", + "agent_nickname", + ): + value = self._normalize_context_scalar(thread_spawn.get(key)) + if value is not None: + context[key] = value + + git_info = payload.get("git") + if isinstance(git_info, Mapping): + git_branch = self._normalize_context_scalar(git_info.get("branch")) + if git_branch is not None: + context["git_branch"] = git_branch + + if thread_spawn is not None: + for source_key, context_key in (("depth", "agent_depth"), ("agent_role", "agent_role")): + value = self._normalize_context_scalar(thread_spawn.get(source_key)) + if value is not None and context_key not in context: + context[context_key] = value + + root_session_id = payload.get("session_id") + normalized_root_id = ( + self._normalize_context_scalar(str(root_session_id)) + if isinstance(root_session_id, (str, int, float, bool)) + else None + ) + if normalized_root_id is not None and str(root_session_id) != physical_session_id: + context["root_session_id"] = normalized_root_id + + return context + + def _add_inherited_session_id(self, inherited_id: Any, session_data: Dict[str, Any]) -> None: + """Record later session metadata as lineage without replacing identity.""" + normalized_id = ( + self._normalize_context_scalar(str(inherited_id)) + if isinstance(inherited_id, (str, int, float, bool)) + else None + ) + if normalized_id is None or normalized_id == "" or str(inherited_id) == session_data["id"]: + return + + inherited_ids = session_data["session_context"].setdefault("inherited_session_ids", []) + if normalized_id not in inherited_ids: + inherited_ids.append(normalized_id) diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 78d03be..a89a0fe 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -1,5 +1,6 @@ """Tests for ADR Sensor parsers.""" +import hashlib import json import os import sqlite3 @@ -157,6 +158,19 @@ def test_parse_no_directory(self): class TestCodexParser: + @staticmethod + def _parse_events(tmp_path, events): + jsonl_file = tmp_path / "rollout-rich.jsonl" + records = [{"type": "session_meta", "payload": {"id": "rich-session"}}, *events] + jsonl_file.write_text("\n".join(json.dumps(record) for record in records), encoding="utf-8") + entry = CodexParser().parse_jsonl_file(jsonl_file) + assert entry is not None + return entry + + @staticmethod + def _tools(entry): + return [tool for message in entry.chat_history for tool in message.tools] + def test_parse_jsonl_file(self, tmp_path): """Test parsing a Codex CLI JSONL file.""" jsonl_file = tmp_path / "rollout-001.jsonl" @@ -170,46 +184,1399 @@ def test_parse_jsonl_file(self, tmp_path): "role": "user", "content": [{"type": "input_text", "text": "List all Python files"}], }, - }, - { - "type": "response_item", - "payload": { - "type": "function_call", - "call_id": "call1", - "name": "shell", - "arguments": '{"command": "find . -name \\"*.py\\""}', + }, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "call1", + "name": "shell", + "arguments": '{"command": "find . -name \\"*.py\\""}', + }, + }, + { + "type": "response_item", + "payload": {"type": "function_call_output", "call_id": "call1", "output": "main.py\ntest.py"}, + }, + { + "type": "response_item", + "payload": { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Found 2 Python files."}], + }, + }, + ] + with open(jsonl_file, "w") as f: + for event in events: + f.write(json.dumps(event) + "\n") + + parser = CodexParser() + entry = parser.parse_jsonl_file(jsonl_file) + + assert entry is not None + assert entry.source == "codex" + assert entry.session_id == "codex_sess1" + assert entry.model == "o3-mini" + assert len(entry.chat_history) >= 2 + + # Check that tool was parsed + assistant_msgs = [m for m in entry.chat_history if m.role == "assistant"] + has_tools = any(len(m.tools) > 0 for m in assistant_msgs) + assert has_tools + + def test_standalone_rich_command_is_normalized(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "turn_id": "turn-a", + "item": { + "type": "CommandExecution", + "command": ["sh", "-lc", "printf ready"], + "cwd": "/workspace/sample", + "status": "completed", + "aggregated_output": "ready", + "formatted_output": "secondary output", + "stdout": "fallback output", + "exit_code": 0, + "duration": {"secs": 1, "nanos": 250_000_000}, + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert tool.tool_name == "exec_command" + assert tool.tool_type == "function_call" + assert tool.arguments == { + "command": ["sh", "-lc", "printf ready"], + "cwd": "/workspace/sample", + "_codex": {"duration": {"secs": 1, "nanos": 250_000_000}}, + } + assert tool.result == "ready" + assert tool.status == "success" + assert tool.error is None + + @pytest.mark.parametrize( + ("item", "expected_result", "expected_status", "expected_error", "expected_duration"), + [ + ( + { + "type": "command_execution", + "cmd": "printf waiting", + "workdir": "/workspace/alias", + "state": "inProgress", + "aggregatedOutput": "waiting", + "durationMs": 25, + }, + "waiting", + "pending", + None, + 25, + ), + ( + { + "type": "commandExecution", + "argv": ["sample-command", "--check"], + "workingDirectory": "/workspace/alias", + "standardOutput": "partial output", + "standardError": "failure detail", + "exitCode": 2, + "duration": "short", + }, + "partial output\nfailure detail", + "error", + "Exit code: 2", + "short", + ), + ( + { + "type": "command-execution", + "command": "sample-command --approve", + "status": "declined", + "errorMessage": "approval declined", + }, + None, + "error", + "approval declined", + None, + ), + ], + ) + def test_rich_command_aliases_outputs_status_and_duration( + self, + tmp_path, + item, + expected_result, + expected_status, + expected_error, + expected_duration, + ): + entry = self._parse_events( + tmp_path, + [{"type": "eventMsg", "payload": {"type": "itemCompleted", "item": item}}], + ) + + tool = self._tools(entry)[0] + assert tool.result == expected_result + assert tool.status == expected_status + assert tool.error == expected_error + if expected_duration is None: + assert "_codex" not in tool.arguments + else: + assert tool.arguments["_codex"]["duration"] == expected_duration + + def test_standalone_file_change_hashes_private_bodies(self, tmp_path): + content = "private body\nwith unicode: café" + unified_diff = "@@ -1 +1 @@\n-old\n+new\n" + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "FileChange", + "changes": { + "src/added.txt": {"type": "add", "content": content}, + "src/current.txt": { + "type": "update", + "move_path": "src/moved.txt", + "unified_diff": unified_diff, + }, + }, + "status": "completed", + "stdout": "changes applied", + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert tool.tool_name == "file_change" + assert tool.tool_type == "function_call" + assert tool.arguments == { + "changes": [ + { + "path": "src/added.txt", + "type": "add", + "content": { + "utf8_bytes": len(content.encode("utf-8")), + "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), + }, + }, + { + "path": "src/current.txt", + "type": "update", + "move_path": "src/moved.txt", + "unified_diff": { + "utf8_bytes": len(unified_diff.encode("utf-8")), + "sha256": hashlib.sha256(unified_diff.encode("utf-8")).hexdigest(), + }, + }, + ] + } + serialized_arguments = json.dumps(tool.arguments) + assert content not in serialized_arguments + assert unified_diff not in serialized_arguments + assert tool.result == "changes applied" + assert tool.status == "success" + + def test_file_change_aliases_are_canonicalized(self, tmp_path): + unified_diff = "@@ sample @@" + entry = self._parse_events( + tmp_path, + [ + { + "type": "event-msg", + "payload": { + "type": "item-completed", + "item": { + "type": "file_change", + "fileChanges": [ + { + "filePath": "src/original.txt", + "changeType": "Modified", + "newPath": "src/renamed.txt", + "unifiedDiff": unified_diff, + } + ], + "status": "failed", + "standardError": "change rejected", + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert tool.arguments["changes"] == [ + { + "path": "src/original.txt", + "type": "update", + "move_path": "src/renamed.txt", + "unified_diff": { + "utf8_bytes": len(unified_diff.encode("utf-8")), + "sha256": hashlib.sha256(unified_diff.encode("utf-8")).hexdigest(), + }, + } + ] + assert tool.result == "change rejected" + assert tool.status == "error" + assert tool.error == "change rejected" + + def test_rich_action_collections_and_text_are_bounded(self, tmp_path): + long_path = "src/" + "p" * 5000 + ".txt" + changes = { + long_path: { + "type": "update", + "move_path": "dst/" + "m" * 5000 + ".txt", + "unified_diff": "private diff", + } + } + changes.update( + { + f"src/file-{index}.txt": {"type": "add", "content": f"body-{index}"} + for index in range(104) + } + ) + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": ["x" * 5000, *[f"arg-{index}" for index in range(104)]], + "cwd": "/workspace/" + "c" * 5000, + "status": "completed", + }, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "FileChange", + "changes": changes, + "status": "completed", + }, + }, + }, + ], + ) + + command, file_change = self._tools(entry) + assert len(command.arguments["command"]) == 100 + assert "[truncated" in command.arguments["command"][0] + assert "[truncated" in command.arguments["cwd"] + assert len(file_change.arguments["changes"]) == 100 + assert file_change.arguments["_codex"]["truncated_changes"] == 5 + assert "[truncated" in file_change.arguments["changes"][0]["path"] + assert "[truncated" in file_change.arguments["changes"][0]["move_path"] + + def test_malformed_and_unsupported_rich_records_are_skipped(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + {"type": "event_msg", "payload": {"type": "item_completed", "item": None}}, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": {"type": "CommandExecution", "command": {"unexpected": "value"}}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": {"type": "FileChange", "changes": [None, "unsupported"]}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": {"type": "McpToolCall", "tool": "lookup"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": {"type": "CollabAgentToolCall", "tool": "delegate"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": "printf valid", + "status": "completed", + }, + }, + }, + ], + ) + + tools = self._tools(entry) + assert len(tools) == 1 + assert tools[0].arguments["command"] == "printf valid" + + def test_agent_messages_capture_adjacent_trigger_and_plaintext_only(self, tmp_path): + def agent_message(text, encrypted_text=None): + content = [] + if text is not None: + content.extend( + [ + {"type": "input_text", "text": text[0]}, + {"type": "output_text", "text": "ignored output"}, + {"type": "input_text", "text": text[1]}, + ] + ) + if encrypted_text is not None: + content.append( + {"type": "encrypted_content", "encrypted_content": encrypted_text} + ) + return { + "type": "response_item", + "payload": { + "type": "agent_message", + "author": "worker-a", + "recipient": "worker-b", + "content": content, + }, + } + + entry = self._parse_events( + tmp_path, + [ + {"type": "inter_agent_communication_metadata", "payload": {"trigger_turn": True}}, + agent_message(("triggered ", "message"), "encrypted-triggered"), + {"type": "inter_agent_communication_metadata", "payload": {"trigger_turn": False}}, + agent_message(("untriggered ", "message"), "encrypted-untriggered"), + {"type": "inter_agent_communication_metadata", "payload": {"trigger_turn": True}}, + {"type": "event_msg", "payload": {"type": "token_count"}}, + agent_message(("nonadjacent ", "message")), + {"type": "inter_agent_communication_metadata", "payload": {"trigger_turn": True}}, + "malformed record", + agent_message(("after malformed ", "message")), + {"type": "inter_agent_communication_metadata", "payload": {"trigger_turn": True}}, + agent_message(None, "opaque-cipher-only"), + agent_message(("after encrypted-only ", "message")), + ], + ) + + assert [(message.role, message.content) for message in entry.chat_history] == [ + ( + "user", + '[agent_message author="worker-a" recipient="worker-b" trigger_turn=true]\n' + "triggered message", + ), + ( + "user", + '[agent_message author="worker-a" recipient="worker-b" trigger_turn=false]\n' + "untriggered message", + ), + ( + "user", + '[agent_message author="worker-a" recipient="worker-b" trigger_turn=false]\n' + "nonadjacent message", + ), + ( + "user", + '[agent_message author="worker-a" recipient="worker-b" trigger_turn=false]\n' + "after malformed message", + ), + ( + "user", + '[agent_message author="worker-a" recipient="worker-b" trigger_turn=false]\n' + "after encrypted-only message", + ), + ] + serialized = entry.to_json() + assert "encrypted-triggered" not in serialized + assert "encrypted-untriggered" not in serialized + assert "opaque-cipher-only" not in serialized + assert "ignored output" not in serialized + + def test_agent_message_parties_are_bounded_and_malformed_values_use_null(self, tmp_path): + long_author = "a" * 5000 + long_recipient = "r" * 5000 + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "agent_message", + "author": long_author, + "recipient": long_recipient, + "content": [{"type": "input_text", "text": "bounded parties"}], + }, + }, + { + "type": "response_item", + "payload": { + "type": "agent_message", + "author": {"unexpected": "mapping"}, + "recipient": ["unexpected", "list"], + "content": [{"type": "input_text", "text": "malformed parties"}], + }, + }, + ], + ) + + prefix = entry.chat_history[0].content.split("\n", 1)[0] + author_json = prefix[len("[agent_message author=") : prefix.index(" recipient=")] + recipient_start = prefix.index(" recipient=") + len(" recipient=") + recipient_json = prefix[recipient_start : prefix.index(" trigger_turn=")] + assert len(json.loads(author_json)) <= 100 + assert len(json.loads(recipient_json)) <= 100 + assert "[truncated" in json.loads(author_json) + assert "[truncated" in json.loads(recipient_json) + assert entry.chat_history[1].content == ( + "[agent_message author=null recipient=null trigger_turn=false]\nmalformed parties" + ) + + def test_subagent_activity_enriches_pending_function_call_without_duplicate(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "toolCallID": "activity-call", + "name": "delegate_work", + "arguments": {"task": "inspect sample"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "SubAgentActivity", + "call_id": "activity-call", + "kind": "started", + "agent_thread_id": "thread-child", + "agent_path": "/worker/child", + "encrypted_content": "not-retained", + }, + }, + }, + ], + ) + + tools = self._tools(entry) + assert len(tools) == 1 + assert tools[0].tool_name == "delegate_work" + assert tools[0].status == "success" + assert tools[0].arguments == { + "task": "inspect sample", + "_codex": { + "subagent_activity": { + "kind": "started", + "agent_thread_id": "thread-child", + "agent_path": "/worker/child", + } + }, + } + assert "not-retained" not in entry.to_json() + + def test_legacy_top_level_subagent_activity_uses_event_id(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "legacy-activity", + "name": "delegate_work", + "arguments": {"task": "inspect sample"}, + }, + }, + { + "type": "sub_agent_activity", + "event_id": "legacy-activity", + "payload": { + "kind": "started", + "agent_thread_id": "thread-child", + "agent_path": "/worker/child", + }, + }, + ], + ) + + tools = self._tools(entry) + + assert len(tools) == 1 + assert tools[0].tool_name == "delegate_work" + assert tools[0].status == "success" + assert tools[0].arguments["_codex"]["subagent_activity"] == { + "kind": "started", + "agent_thread_id": "thread-child", + "agent_path": "/worker/child", + } + + def test_unmatched_subagent_activity_emits_standalone_tool(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "sub_agent_activity", + "itemId": "standalone-activity", + "kind": "interacted", + "agent_thread_id": "thread-standalone", + "agent_path": "/worker/standalone", + }, + }, + } + ], + ) + + assert entry.source == "codex" + assert entry.session_id == "codex_rich-session" + tool = self._tools(entry)[0] + assert tool.tool_name == "subagent_activity" + assert tool.tool_type == "function_call" + assert tool.status == "success" + assert tool.result is None + assert tool.error is None + assert tool.arguments == { + "_codex": { + "subagent_activity": { + "kind": "interacted", + "agent_thread_id": "thread-standalone", + "agent_path": "/worker/standalone", + } + } + } + + def test_top_level_rich_mcp_success_preserves_metadata(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "call_id": "direct-call", + "invocation": { + "server": "catalog-server", + "tool": "lookup_record", + "arguments": {"record_id": "record-1"}, + }, + "connector_id": "connector-1", + "link_id": "link-1", + "app_name": "Catalog", + "action_name": "Lookup", + "read_only_hint": True, + "duration": {"secs": 1, "nanos": 5}, + "result": { + "Ok": { + "content": [{"type": "text", "text": '{"found":true}'}], + "isError": False, + } + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert (tool.server_name, tool.tool_name, tool.tool_type) == ( + "catalog-server", + "lookup_record", + "mcp_tool", + ) + assert tool.arguments == { + "record_id": "record-1", + "_codex": { + "duration": {"secs": 1, "nanos": 5}, + "read_only_hint": True, + "connector": { + "connector_id": "connector-1", + "link_id": "link-1", + "app_name": "Catalog", + "action_name": "Lookup", + }, + }, + } + assert tool.result == '{"found":true}' + assert tool.status == "success" + assert tool.error is None + + def test_item_completed_rich_mcp_structured_error(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "eventMsg", + "payload": { + "type": "itemCompleted", + "item": { + "type": "McpToolCallEnd", + "id": "failed-call", + "toolCall": { + "serverName": "records-server", + "toolName": "update_record", + "args": '{"record_id":"record-2"}', + }, + "status": "failed", + "result": { + "content": [{"type": "text", "text": "request rejected"}], + "isError": True, + }, + "error": {"message": "request rejected", "code": "denied"}, + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert (tool.server_name, tool.tool_name) == ("records-server", "update_record") + assert tool.arguments == {"record_id": "record-2"} + assert tool.result == "request rejected" + assert tool.status == "error" + assert tool.error == "request rejected" + + @pytest.mark.parametrize( + ("argument_key", "raw_arguments", "expected"), + [ + ("arguments", {"record_id": "record-3"}, {"record_id": "record-3"}), + ("args", '{"record_id":"record-4"}', {"record_id": "record-4"}), + ("input", ["record-5", 5], {"raw": ["record-5", 5]}), + ("params", "opaque input", {"raw": "opaque input"}), + ("parameters", None, {}), + ], + ) + def test_rich_mcp_argument_shapes_are_normalized( + self, tmp_path, argument_key, raw_arguments, expected + ): + invocation = { + "server_name": "shape-server", + "tool_name": "inspect_record", + argument_key: raw_arguments, + } + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "call": invocation, + "result": {"Ok": {"content": []}}, + }, + } + ], + ) + + assert self._tools(entry)[0].arguments == expected + + def test_rich_mcp_metadata_is_scalar_only_and_bounded(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "McpToolCall", + "id": "bounded-call", + "server": "metadata-server", + "tool": "inspect_record", + "arguments": {}, + "connectorId": "c" * 5000, + "linkId": {"unsupported": "mapping"}, + "appName": 7, + "actionName": False, + "readOnlyHint": False, + "durationMs": { + "samples": list(range(150)), + "detail": "d" * 5000, + }, + "status": "completed", + "result": { + "content": [{"type": "text", "text": "inspection complete"}], + "isError": False, + }, + }, + }, + } + ], + ) + + tool = self._tools(entry)[0] + metadata = tool.arguments["_codex"] + assert len(metadata["duration"]["samples"]) == 100 + assert "[truncated" in metadata["duration"]["detail"] + assert metadata["read_only_hint"] is False + assert "[truncated" in metadata["connector"]["connector_id"] + assert metadata["connector"]["app_name"] == 7 + assert metadata["connector"]["action_name"] is False + assert "link_id" not in metadata["connector"] + assert tool.result == "inspection complete" + assert tool.status == "success" + assert tool.error is None + + @pytest.mark.parametrize("wrapper_server", ["bridge-alpha", "bridge-beta"]) + def test_rich_mcp_structural_wrapper_reports_effective_target( + self, tmp_path, wrapper_server + ): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "call_id": "wrapped-call", + "invocation": { + "server": wrapper_server, + "tool": "invokeTool", + "arguments": { + "targetServer": "effective-server", + "targetTool": "inspect_record", + "params": '{"record_id":"record-6"}', + }, + }, + "result": {"Err": {"message": "inspection denied"}}, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert (tool.server_name, tool.tool_name) == ("effective-server", "inspect_record") + assert tool.arguments == { + "record_id": "record-6", + "_codex_mcp_wrapper": { + "server_name": wrapper_server, + "tool_name": "invokeTool", + }, + } + assert tool.status == "error" + assert tool.error == "inspection denied" + + def test_invoke_tool_without_nested_arguments_is_not_unwrapped(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": { + "server": "bridge-server", + "tool": "invoke_tool", + "arguments": { + "target_server": "effective-server", + "target_tool": "inspect_record", + }, + }, + "result": {"Ok": {"content": []}}, + }, + } + ], + ) + + tool = self._tools(entry)[0] + assert (tool.server_name, tool.tool_name) == ("bridge-server", "invoke_tool") + assert "_codex_mcp_wrapper" not in tool.arguments + + def test_malformed_rich_mcp_records_are_skipped(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": {"server": " ", "tool": "inspect_record"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "server": "records-server", + "tool": "", + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "McpToolCall", + "server": "records-server", + "tool": ["unsupported"], + }, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CollabAgentToolCall", + "server": "ignored-server", + "tool": "ignored_tool", + }, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "serverName": "valid-server", + "toolName": "valid_tool", + "arguments": {}, + "result": {"Ok": {"content": []}}, + }, + }, + ], + ) + + tools = self._tools(entry) + assert len(tools) == 1 + assert (tools[0].server_name, tools[0].tool_name) == ("valid-server", "valid_tool") + + def test_shared_call_id_reconciles_classic_mcp_in_place(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "before-call", + "name": "before_tool", + "arguments": {}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call", + "toolCallID": "shared-call", + "name": "mcp__legacy-server__lookup_record", + "arguments": {"record_id": "classic"}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "after-call", + "name": "after_tool", + "arguments": {}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "id": "shared-call", + "invocation": { + "server": "effective-server", + "tool": "lookup_record", + "arguments": {"record_id": "rich"}, + }, + "result": {"Ok": {"content": [{"type": "text", "text": "found"}]}}, + }, + }, + ], + ) + + tools = self._tools(entry) + assert [tool.tool_name for tool in tools] == [ + "before_tool", + "lookup_record", + "after_tool", + ] + assert tools[1].server_name == "effective-server" + assert tools[1].arguments == {"record_id": "rich"} + assert tools[1].result == "found" + assert tools[1].status == "success" + + def test_exact_signature_reconciles_rich_mcp_without_call_id(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "classic-only-id", + "name": "mcp__records-server__lookup_record", + "arguments": {"record_id": "sample"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": { + "server": "records-server", + "tool": "lookup_record", + "arguments": {"record_id": "sample"}, + }, + "result": {"Ok": {"content": [{"type": "text", "text": "found"}]}}, + }, }, - }, - { - "type": "response_item", - "payload": {"type": "function_call_output", "call_id": "call1", "output": "main.py\ntest.py"}, - }, - { - "type": "response_item", - "payload": { - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": "Found 2 Python files."}], + ], + ) + + tools = self._tools(entry) + + assert len(tools) == 1 + assert tools[0].tool_name == "lookup_record" + assert tools[0].server_name == "records-server" + assert tools[0].result == "found" + + def test_namespace_classic_mcp_reconciles_with_rich_record(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "classic-only-id", + "namespace": "mcp__queryrunner_mcp", + "name": "run_query", + "arguments": {"query": "SELECT 1"}, + }, }, - }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": { + "server": "queryrunner_mcp", + "tool": "run_query", + "arguments": {"query": "SELECT 1"}, + }, + "result": {"Ok": {"content": [{"type": "text", "text": "one"}]}}, + }, + }, + ], + ) + + tools = self._tools(entry) + + assert len(tools) == 1 + assert (tools[0].server_name, tools[0].tool_name) == ("queryrunner_mcp", "run_query") + assert tools[0].result == "one" + + def test_signature_fallback_does_not_cross_a_later_call_boundary(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "classic-only-id", + "name": "mcp__records-server__lookup_record", + "arguments": {"record_id": "sample"}, + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "later-call", + "name": "other_tool", + "arguments": {}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": { + "server": "records-server", + "tool": "lookup_record", + "arguments": {"record_id": "sample"}, + }, + "result": {"Ok": {"content": []}}, + }, + }, + ], + ) + + assert [tool.tool_name for tool in self._tools(entry)] == [ + "mcp__records-server__lookup_record", + "other_tool", + "lookup_record", ] - with open(jsonl_file, "w") as f: - for event in events: - f.write(json.dumps(event) + "\n") - parser = CodexParser() - entry = parser.parse_jsonl_file(jsonl_file) + def test_unrelated_classic_and_rich_mcp_calls_remain_separate(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "classic-call", + "name": "mcp__records-server__lookup_record", + "arguments": {"record_id": "classic"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "call_id": "rich-call", + "invocation": { + "server": "records-server", + "tool": "lookup_record", + "arguments": {"record_id": "rich"}, + }, + "result": {"Ok": {"content": []}}, + }, + }, + ], + ) - assert entry is not None - assert entry.source == "codex" - assert entry.session_id == "codex_sess1" - assert entry.model == "o3-mini" - assert len(entry.chat_history) >= 2 + tools = self._tools(entry) + assert [tool.tool_name for tool in tools] == [ + "mcp__records-server__lookup_record", + "lookup_record", + ] + assert [tool.arguments["record_id"] for tool in tools] == ["classic", "rich"] - # Check that tool was parsed - assistant_msgs = [m for m in entry.chat_history if m.role == "assistant"] - has_tools = any(len(m.tools) > 0 for m in assistant_msgs) - assert has_tools + def test_signature_fallback_preserves_mcp_name_punctuation(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "classic-only-id", + "name": "mcp__a-b__lookup_record", + "arguments": {"record_id": "sample"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "mcp_tool_call_end", + "invocation": { + "server": "ab", + "tool": "lookup_record", + "arguments": {"record_id": "sample"}, + }, + "result": {"Ok": {"content": []}}, + }, + }, + ], + ) + + tools = self._tools(entry) + + assert len(tools) == 2 + assert [tool.server_name for tool in tools] == ["a-b", "ab"] + + def test_rich_command_replaces_matching_custom_wrapper(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "wrapper-call", + "name": "exec", + "input": "printf ready", + "internal_chat_message_metadata_passthrough": {"turn_id": "turn-a"}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "turn_id": "turn-a", + "item": { + "type": "CommandExecution", + "command": ["sh", "-lc", "printf ready"], + "status": "completed", + "aggregated_output": "rich output", + }, + }, + }, + { + "type": "response_item", + "payload": { + "type": "custom_tool_call_output", + "call_id": "wrapper-call", + "output": "wrapper output", + }, + }, + ], + ) + + tools = self._tools(entry) + assert len(tools) == 1 + assert tools[0].tool_name == "exec_command" + assert tools[0].result == "rich output" + + @pytest.mark.parametrize( + ("wrapper_name", "wrapper_input", "wrapper_turn", "rich_turn", "rich_command"), + [ + ("exec", "printf first", "turn-a", "turn-a", "printf second"), + ("fetch", "printf same", "turn-a", "turn-a", "printf same"), + ("exec", "printf same", "turn-a", "turn-b", "printf same"), + ("apply_patch", "printf same", "turn-a", "turn-a", "printf same"), + ], + ) + def test_unrelated_wrappers_are_preserved( + self, + tmp_path, + wrapper_name, + wrapper_input, + wrapper_turn, + rich_turn, + rich_command, + ): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "wrapper-call", + "name": wrapper_name, + "input": wrapper_input, + "internal_chat_message_metadata_passthrough": {"turn_id": wrapper_turn}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "turn_id": rich_turn, + "item": { + "type": "CommandExecution", + "command": rich_command, + "status": "completed", + }, + }, + }, + ], + ) + + assert [tool.tool_name for tool in self._tools(entry)] == [wrapper_name, "exec_command"] + + def test_command_wrapper_is_not_replaced_by_unrelated_file_change(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "command-call", + "name": "exec", + "input": "printf unchanged", + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "FileChange", + "changes": {"src/sample.txt": {"type": "add", "content": "private"}}, + "status": "completed", + }, + }, + }, + ], + ) + + assert [tool.tool_name for tool in self._tools(entry)] == ["exec", "file_change"] + + def test_late_rich_command_replaces_completed_wrapper(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "late-call", + "name": "exec_command", + "input": '{"command":"printf late"}', + }, + }, + { + "type": "response_item", + "payload": { + "type": "custom_tool_call_output", + "call_id": "late-call", + "output": "wrapper output", + }, + }, + {"type": "event_msg", "payload": {"type": "token_count", "info": {}}}, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": "printf late", + "status": "completed", + "aggregated_output": "late output", + }, + }, + }, + ], + ) + + tools = self._tools(entry) + assert len(tools) == 1 + assert tools[0].tool_name == "exec_command" + assert tools[0].result == "late output" + + def test_identical_rich_command_does_not_cross_a_later_call_boundary(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "first-call", + "name": "exec", + "input": "printf same", + }, + }, + { + "type": "response_item", + "payload": { + "type": "function_call", + "call_id": "later-call", + "name": "other_tool", + "arguments": {}, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": "printf same", + "status": "completed", + }, + }, + }, + ], + ) + + assert [tool.tool_name for tool in self._tools(entry)] == [ + "exec", + "other_tool", + "exec_command", + ] + + def test_one_wrapper_can_expand_to_multiple_rich_actions(self, tmp_path): + entry = self._parse_events( + tmp_path, + [ + { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "call_id": "multi-call", + "name": "exec", + "input": "const opaque = true;", + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "FileChange", + "changes": {"src/sample.txt": {"type": "add", "content": "private"}}, + "status": "completed", + }, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": "printf first", + "status": "completed", + "aggregated_output": "first", + }, + }, + }, + { + "type": "event_msg", + "payload": { + "type": "item_completed", + "item": { + "type": "CommandExecution", + "command": "printf second", + "status": "failed", + "stderr": "second failed", + }, + }, + }, + { + "type": "response_item", + "payload": { + "type": "custom_tool_call_output", + "call_id": "multi-call", + "output": "wrapper output", + }, + }, + ], + ) + + assert len(entry.chat_history) == 1 + tools = self._tools(entry) + assert [tool.tool_name for tool in tools] == [ + "file_change", + "exec_command", + "exec_command", + ] + assert [tool.status for tool in tools] == ["success", "success", "error"] + assert [tool.result for tool in tools] == [None, "first", "second failed"] def test_parses_custom_tool_call(self, tmp_path): """custom_tool_call records must yield tools with their arguments intact. @@ -866,16 +2233,39 @@ def test_first_valid_session_meta_defines_physical_identity(self, tmp_path): events = [ { "type": "session_meta", - "payload": {"timestamp": "2024-01-01T00:00:00Z", "cwd": str(tmp_path / "missing-id")}, + "payload": { + "timestamp": "2024-01-01T00:00:00Z", + "cwd": str(tmp_path / "missing-id"), + "originator": "ignored-originator", + }, + }, + { + "type": "session_meta", + "payload": { + "id": "physical-session", + "timestamp": "2025-01-02T03:04:05Z", + "cwd": first_cwd, + "originator": "first-originator", + }, + }, + { + "type": "session_meta", + "payload": { + "id": "later-session", + "timestamp": "2026-02-03T04:05:06Z", + "cwd": later_cwd, + "originator": "later-originator", + }, }, { "type": "session_meta", - "payload": {"id": "physical-session", "timestamp": "2025-01-02T03:04:05Z", "cwd": first_cwd}, + "payload": {"id": "later-session", "cwd": "/duplicate"}, }, { "type": "session_meta", - "payload": {"id": "later-session", "timestamp": "2026-02-03T04:05:06Z", "cwd": later_cwd}, + "payload": {"id": "physical-session", "cwd": "/same-session"}, }, + {"type": "session_meta", "payload": {"id": 42}}, { "type": "response_item", "payload": {"type": "message", "role": "user", "content": "identity check"}, @@ -888,6 +2278,116 @@ def test_first_valid_session_meta_defines_physical_identity(self, tmp_path): 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) + assert entry.session_context == { + "originator": "first-originator", + "inherited_session_ids": ["later-session", "42"], + } + + def test_captures_codex_session_provenance_without_changing_identity(self, tmp_path): + jsonl_file = tmp_path / "rollout-context.jsonl" + long_originator = "origin-start-" + ("x" * 1200) + "-origin-end" + events = [ + { + "type": "session_meta", + "payload": { + "id": "physical-session", + "session_id": "root-session", + "timestamp": "2025-01-02T03:04:05Z", + "cwd": "/workspace/example", + "originator": long_originator, + "cli_version": "1.2.3", + "model_provider": "provider-name", + "git": {"branch": "feature/session-context"}, + "parent_thread_id": "parent-thread", + "forked_from_id": "forked-thread", + "agent_path": "root/worker", + "agent_nickname": "worker-name", + "agent_role": "reviewer", + "subagent_history_start_ordinal": 4, + "thread_source": "subagent", + }, + }, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "provenance check"}, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert entry.source == "codex" + assert entry.session_id == "codex_physical-session" + assert entry.project_path == "/workspace/example" + assert entry.user_id is None + assert entry.session_context == { + "originator": entry.session_context["originator"], + "cli_version": "1.2.3", + "model_provider": "provider-name", + "parent_thread_id": "parent-thread", + "forked_from_id": "forked-thread", + "agent_path": "root/worker", + "agent_nickname": "worker-name", + "agent_role": "reviewer", + "subagent_history_start_ordinal": 4, + "thread_source": "subagent", + "git_branch": "feature/session-context", + "root_session_id": "root-session", + } + assert entry.session_context["originator"].startswith("origin-start-") + assert entry.session_context["originator"].endswith("-origin-end") + assert "[truncated" in entry.session_context["originator"] + assert len(entry.session_context["originator"]) < 1000 + + def test_captures_nested_subagent_provenance_and_ignores_structured_values(self, tmp_path): + jsonl_file = tmp_path / "rollout-subagent-context.jsonl" + events = [ + { + "type": "session_meta", + "payload": { + "id": "child-session", + "session_id": {"not": "scalar"}, + "originator": ["not", "scalar"], + "cli_version": {"not": "scalar"}, + "model_provider": ["not", "scalar"], + "git": {"branch": {"not": "scalar"}}, + "parent_thread_id": {"not": "scalar"}, + "agent_path": ["not", "scalar"], + "agent_nickname": {"not": "scalar"}, + "agent_role": "planner", + "forked_from_id": ["not", "scalar"], + "subagent_history_start_ordinal": {"not": "scalar"}, + "thread_source": ["not", "scalar"], + "source": { + "subagent": { + "thread_spawn": { + "parent_thread_id": "parent-session", + "agent_path": "root/child", + "agent_nickname": "child-name", + "depth": 2, + "agent_role": "reviewer", + } + } + }, + }, + }, + {"type": "session_meta", "payload": {"id": ["not", "scalar"]}}, + { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": "nested provenance"}, + }, + ] + jsonl_file.write_text("\n".join(json.dumps(event) for event in events)) + + entry = CodexParser().parse_jsonl_file(jsonl_file) + + assert entry.session_context == { + "parent_thread_id": "parent-session", + "agent_path": "root/child", + "agent_nickname": "child-name", + "agent_depth": 2, + "agent_role": "planner", + } def test_parse_no_directory(self): """Test parse_all when directory doesn't exist."""