normalizer.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228
  1. """Normalizes backend-specific session logs to a common tool call schema."""
  2. from __future__ import annotations
  3. import json
  4. from collections.abc import Callable
  5. from pathlib import Path
  6. from typing import Any
  7. NATIVE_TOOLS: set[str] = {
  8. "EnterWorktree",
  9. "ExitWorktree",
  10. "EnterPlanMode",
  11. "ExitPlanMode",
  12. "TaskCreate",
  13. "TaskUpdate",
  14. "TaskList",
  15. "TaskGet",
  16. "Skill",
  17. "Agent",
  18. "Read",
  19. "Write",
  20. "Edit",
  21. "Glob",
  22. "Grep",
  23. }
  24. LOG_EXTENSIONS: tuple[str, ...] = ("*.jsonl", "*.json")
  25. def snapshot_log_dir(log_dir: Path) -> set[str]:
  26. """Snapshot all session log files in a log directory (recursive)."""
  27. if not log_dir.exists():
  28. return set()
  29. files: set[str] = set()
  30. for ext in LOG_EXTENSIONS:
  31. files.update(str(f.relative_to(log_dir)) for f in log_dir.rglob(ext))
  32. return files
  33. def collect_new_logs(log_dir: Path, snapshot: set[str]) -> list[Path]:
  34. """Find session log files created after the snapshot (recursive)."""
  35. if not log_dir.exists():
  36. return []
  37. current: dict[str, Path] = {}
  38. for ext in LOG_EXTENSIONS:
  39. current.update({str(f.relative_to(log_dir)): f for f in log_dir.rglob(ext)})
  40. new_keys: set[str] = set(current.keys()) - snapshot
  41. return [current[k] for k in sorted(new_keys)]
  42. def filter_codex_logs_by_cwd(paths: list[Path], target_cwd: str) -> list[Path]:
  43. """Drop codex rollouts whose session_meta.cwd doesn't match target_cwd.
  44. Codex stores all sessions under a shared ~/.codex/sessions/ tree, so when
  45. multiple drill scenarios run in parallel each one's snapshot diff sees every
  46. other run's rollouts. Each rollout's first line is a `session_meta` event
  47. that records the cwd the codex CLI was launched in — use it to attribute
  48. rollouts to the run that produced them.
  49. """
  50. matched: list[Path] = []
  51. for path in paths:
  52. try:
  53. with path.open() as f:
  54. first_line = f.readline()
  55. entry = json.loads(first_line)
  56. except (OSError, json.JSONDecodeError):
  57. continue
  58. if entry.get("type") != "session_meta":
  59. continue
  60. cwd = entry.get("payload", {}).get("cwd", "")
  61. if cwd == target_cwd:
  62. matched.append(path)
  63. return matched
  64. def normalize_claude_logs(raw_content: str) -> list[dict[str, Any]]:
  65. """Normalize Claude Code session logs.
  66. CC logs are JSONL where assistant messages have:
  67. {"type": "assistant", "message": {"content": [{"type": "tool_use", "name": "...",
  68. "input": {...}}]}}
  69. """
  70. results: list[dict[str, Any]] = []
  71. for line in raw_content.strip().split("\n"):
  72. if not line.strip():
  73. continue
  74. try:
  75. entry = json.loads(line)
  76. except json.JSONDecodeError:
  77. continue
  78. # Handle nested CC format: assistant messages contain tool_use in content array
  79. if entry.get("type") == "assistant":
  80. message = entry.get("message", {})
  81. for block in message.get("content", []):
  82. if block.get("type") == "tool_use":
  83. tool_name = block.get("name", "")
  84. source = "native" if tool_name in NATIVE_TOOLS else "shell"
  85. results.append(
  86. {"tool": tool_name, "args": block.get("input", {}), "source": source}
  87. )
  88. # Also handle flat format (for test compatibility)
  89. elif entry.get("type") == "tool_use":
  90. tool_name = entry.get("name", "")
  91. source = "native" if tool_name in NATIVE_TOOLS else "shell"
  92. results.append({"tool": tool_name, "args": entry.get("input", {}), "source": source})
  93. return results
  94. def normalize_codex_logs(raw_content: str) -> list[dict[str, Any]]:
  95. """Normalize Codex rollout logs.
  96. Codex logs use: {"type": "response_item", "payload": {"type": "function_call", ...}}
  97. Tool calls are "function_call" with name "exec_command" (shell) or other names.
  98. """
  99. results: list[dict[str, Any]] = []
  100. for line in raw_content.strip().split("\n"):
  101. if not line.strip():
  102. continue
  103. try:
  104. entry = json.loads(line)
  105. except json.JSONDecodeError:
  106. continue
  107. if entry.get("type") != "response_item":
  108. continue
  109. # Codex uses "payload" not "item"
  110. payload = entry.get("payload", entry.get("item", {}))
  111. payload_type = payload.get("type", "")
  112. if payload_type == "function_call":
  113. name = payload.get("name", "")
  114. raw_args = payload.get("arguments", "{}")
  115. # Arguments are JSON-encoded strings in codex
  116. if isinstance(raw_args, str):
  117. try:
  118. args = json.loads(raw_args)
  119. except json.JSONDecodeError:
  120. args = {"raw": raw_args}
  121. else:
  122. args = raw_args
  123. # exec_command is codex's shell tool
  124. if name == "exec_command":
  125. results.append(
  126. {"tool": "Bash", "args": {"command": args.get("cmd", "")}, "source": "shell"}
  127. )
  128. elif name == "apply_patch":
  129. results.append({"tool": "Edit", "args": args, "source": "native"})
  130. else:
  131. source = "native" if name in NATIVE_TOOLS else "shell"
  132. results.append({"tool": name, "args": args, "source": source})
  133. elif payload_type == "local_shell_call":
  134. action = payload.get("action", {})
  135. cmd = action.get("command", [])
  136. cmd_str = " ".join(cmd) if isinstance(cmd, list) else str(cmd)
  137. results.append({"tool": "Bash", "args": {"command": cmd_str}, "source": "shell"})
  138. return results
  139. # Reverse mapping: Gemini tool names → Claude Code canonical names
  140. GEMINI_TOOL_MAP: dict[str, str] = {
  141. "run_shell_command": "Bash",
  142. "read_file": "Read",
  143. "write_file": "Write",
  144. "replace": "Edit",
  145. "grep_search": "Grep",
  146. "glob": "Glob",
  147. "activate_skill": "Skill",
  148. "google_web_search": "WebSearch",
  149. "web_fetch": "WebFetch",
  150. "write_todos": "TodoWrite",
  151. "list_directory": "Glob",
  152. "enter_plan_mode": "EnterPlanMode",
  153. "exit_plan_mode": "ExitPlanMode",
  154. }
  155. def normalize_gemini_logs(raw_content: str) -> list[dict[str, Any]]:
  156. """Normalize Gemini CLI session logs.
  157. Gemini logs may be a single JSON file with a messages array, or JSONL
  158. session files in newer CLI versions. Each "gemini" message may have a
  159. toolCalls array:
  160. {"name": "run_shell_command", "args": {"command": "..."}, "status": "success"}
  161. """
  162. results: list[dict[str, Any]] = []
  163. messages: list[dict[str, Any]] = []
  164. try:
  165. data = json.loads(raw_content)
  166. except json.JSONDecodeError:
  167. for line in raw_content.strip().split("\n"):
  168. if not line.strip():
  169. continue
  170. try:
  171. entry = json.loads(line)
  172. except json.JSONDecodeError:
  173. continue
  174. if isinstance(entry, dict):
  175. messages.append(entry)
  176. else:
  177. if isinstance(data, dict) and "messages" in data:
  178. messages = [m for m in data.get("messages", []) if isinstance(m, dict)]
  179. elif isinstance(data, dict):
  180. messages = [data]
  181. elif isinstance(data, list):
  182. messages = [m for m in data if isinstance(m, dict)]
  183. seen_tool_calls: set[str] = set()
  184. for message in messages:
  185. if message.get("type") != "gemini":
  186. continue
  187. for tc in message.get("toolCalls", []):
  188. tool_call_id = tc.get("id")
  189. if tool_call_id and tool_call_id in seen_tool_calls:
  190. continue
  191. if tool_call_id:
  192. seen_tool_calls.add(tool_call_id)
  193. gemini_name = tc.get("name", "")
  194. canonical = GEMINI_TOOL_MAP.get(gemini_name, gemini_name)
  195. args = tc.get("args", {})
  196. source = "native" if canonical in NATIVE_TOOLS else "shell"
  197. results.append({"tool": canonical, "args": args, "source": source})
  198. return results
  199. NORMALIZERS: dict[str, Callable[[str], list[dict[str, Any]]]] = {
  200. "claude": normalize_claude_logs,
  201. "codex": normalize_codex_logs,
  202. "gemini": normalize_gemini_logs,
  203. }