Reads exact token usage for the just-finished turn from the profile's hermes state.db (sessions table). Best-effort: returns None when the session row cannot be found.
(profile_dir, started_after)
| 412 | |
| 413 | |
| 414 | def read_hermes_usage(profile_dir, started_after): |
| 415 | """Reads exact token usage for the just-finished turn from the profile's |
| 416 | hermes state.db (sessions table). Best-effort: returns None when the |
| 417 | session row cannot be found.""" |
| 418 | state_db = profile_dir / "state.db" |
| 419 | if not state_db.exists(): |
| 420 | return None |
| 421 | try: |
| 422 | db = sqlite3.connect(f"file:{state_db}?mode=ro", uri=True) |
| 423 | row = db.execute( |
| 424 | "SELECT id, model, billing_provider, tool_call_count, input_tokens, " |
| 425 | "output_tokens, cache_read_tokens, reasoning_tokens " |
| 426 | "FROM sessions WHERE started_at >= ? ORDER BY started_at DESC LIMIT 1", |
| 427 | (started_after - 5,), |
| 428 | ).fetchone() |
| 429 | db.close() |
| 430 | except sqlite3.Error: |
| 431 | return None |
| 432 | if row is None: |
| 433 | return None |
| 434 | return { |
| 435 | "session_id": row[0], |
| 436 | "model": row[1], |
| 437 | "billing_provider": row[2], |
| 438 | "tool_calls": row[3], |
| 439 | "input_tokens": row[4], |
| 440 | "output_tokens": row[5], |
| 441 | "cache_read_tokens": row[6], |
| 442 | "reasoning_tokens": row[7], |
| 443 | } |
| 444 | |
| 445 | |
| 446 | def drive_cursor_agent(args, scenario, fixture, log_dir, eval_env): |
no test coverage detected