Extract response content and token usage from codex JSONL output. Looks for ``item.completed`` events whose ``item.type`` is ``"agent_message"`` to capture the assistant text, and ``turn.completed`` events for token usage. Multiple turns are summed.
(jsonl_output: str)
| 92 | |
| 93 | |
| 94 | def _parse_jsonl(jsonl_output: str) -> tuple[str, TokenUsage]: |
| 95 | """Extract response content and token usage from codex JSONL output. |
| 96 | |
| 97 | Looks for ``item.completed`` events whose ``item.type`` is |
| 98 | ``"agent_message"`` to capture the assistant text, and |
| 99 | ``turn.completed`` events for token usage. Multiple turns are |
| 100 | summed. |
| 101 | """ |
| 102 | usage = TokenUsage() |
| 103 | last_message = "" |
| 104 | for line in jsonl_output.splitlines(): |
| 105 | if not line.strip(): |
| 106 | continue |
| 107 | try: |
| 108 | event = json.loads(line) |
| 109 | except json.JSONDecodeError: |
| 110 | continue |
| 111 | etype = event.get("type") |
| 112 | if etype == "item.completed": |
| 113 | item = event.get("item") or {} |
| 114 | if item.get("type") == "agent_message": |
| 115 | last_message = item.get("text", "") |
| 116 | elif etype == "turn.completed": |
| 117 | turn_usage = event.get("usage") or {} |
| 118 | usage.input_tokens += turn_usage.get("input_tokens", 0) |
| 119 | usage.output_tokens += turn_usage.get("output_tokens", 0) |
| 120 | return last_message, usage |