Extract token usage from OpenCode events. OpenCode emits per-step usage as ``tokens.input`` / ``tokens.output`` plus cache and reasoning buckets. Older/test fixtures may use SDK-style usage names, which are treated as possibly cumulative and deduplicated by max.
(events: list[dict[str, Any]])
| 321 | |
| 322 | |
| 323 | def _usage_from_events(events: list[dict[str, Any]]) -> tuple[int, int]: |
| 324 | """Extract token usage from OpenCode events. |
| 325 | |
| 326 | OpenCode emits per-step usage as ``tokens.input`` / ``tokens.output`` plus |
| 327 | cache and reasoning buckets. Older/test fixtures may use SDK-style usage |
| 328 | names, which are treated as possibly cumulative and deduplicated by max. |
| 329 | """ |
| 330 | input_tokens = 0 |
| 331 | output_tokens = 0 |
| 332 | sdk_input_tokens = 0 |
| 333 | sdk_output_tokens = 0 |
| 334 | for event in events: |
| 335 | for item in _walk_dicts(event): |
| 336 | tokens = item.get("tokens") |
| 337 | if isinstance(tokens, dict): |
| 338 | in_value, out_value = _opencode_usage_tokens(tokens) |
| 339 | input_tokens += in_value |
| 340 | output_tokens += out_value |
| 341 | continue |
| 342 | |
| 343 | in_value = ( |
| 344 | item.get("input_tokens") |
| 345 | or item.get("inputTokens") |
| 346 | or item.get("prompt_tokens") |
| 347 | or item.get("promptTokens") |
| 348 | ) |
| 349 | out_value = ( |
| 350 | item.get("output_tokens") |
| 351 | or item.get("outputTokens") |
| 352 | or item.get("completion_tokens") |
| 353 | or item.get("completionTokens") |
| 354 | ) |
| 355 | sdk_input_tokens = max(sdk_input_tokens, _token_count(in_value)) |
| 356 | sdk_output_tokens = max(sdk_output_tokens, _token_count(out_value)) |
| 357 | return input_tokens or sdk_input_tokens, output_tokens or sdk_output_tokens |
| 358 | |
| 359 | |
| 360 | _REASONING_HINTS = ("reasoning", "thinking", "thought") |
no test coverage detected