iteration_stat.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/iteration_stat.py
 
"""
One-iteration analytics record from the ReAct loop.
 
Purpose
-------
:class:`IterationStat` is a frozen dataclass capturing every metric
produced during a single ReAct iteration: timestamp, model, effort,
tools called, token counts, cache hits/misses, and wall-clock
duration.  Instances are created by the ReAct loop and emitted to
the presenter for both human-readable display and JSON-lines
export to a ``--stats-file`` sink.
 
Responsibilities
----------------
- :class:`IterationStat` — frozen, slot-based, serialisable record.
- :func:`iteration_stat_to_json` — lossless serialisation for the
  JSON-lines sink.
- :func:`iteration_stat_human_line` — compact one-line format for
  the terminal (used by ConsolePresenter and RichPresenter).
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- All fields are primitive types (str, int, tuple) — no nested
  references or lambdas, so the dataclass is always serialisable.
- ``ts`` is ISO-8601 with millisecond precision, UTC.
- ``tools`` is a tuple (hashable);  empty tuple means no tools were
  called this iteration (final answer).
- ``dur_ms`` is wall-clock time from start to finish of the iteration,
  including API streaming and tool dispatch.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
 
from __future__ import annotations
 
import json
from dataclasses import dataclass
from datetime import datetime, timezone
 
from oc_diagnostics import _dbg
 
 
@dataclass(slots=True, frozen=True)
class IterationStat:
    """One ReAct-iteration analytics record."""
 
    ts: str
    """Timestamp string."""
 
    iter: int
    """Zero-based iteration index within the current turn."""
 
    model: str
    """Model name used for this iteration (after routing)."""
 
    effort: str
    """Reasoning effort (``"low"`` / ``"medium"`` / ``"high"`` / ``""``)."""
 
    tools: tuple[str, ...]
    """Tool names called this iteration.  Empty tuple = final answer."""
 
    prompt_tokens: int
    """Prompt tokens consumed this iteration."""
 
    completion_tokens: int
    """Completion tokens produced this iteration."""
 
    cache_hit_tokens: int
    """DeepSeek prefix-cache hit tokens (0 if unavailable)."""
 
    cache_miss_tokens: int
    """DeepSeek prefix-cache miss tokens (0 if unavailable)."""
 
    dur_ms: int
    """Wall-clock duration of this iteration in milliseconds."""
 
 
_dbg("OC_CHAT.CLIENT", "module", "loaded", 4)
 
 
# ── Serialisation ──────────────────────────────────────────────────────────
 
 
def iteration_stat_to_json(stat: IterationStat) -> str:
    """Lossless JSON-serialisation of an ``IterationStat``.
 
    Converts the ``tools`` tuple to a list for JSON compatibility.
    Returns a single line (no indentation).
    """
    d = {
        "ts": stat.ts,
        "iter": stat.iter,
        "model": stat.model,
        "effort": stat.effort,
        "tools": list(stat.tools),
        "prompt_tokens": stat.prompt_tokens,
        "completion_tokens": stat.completion_tokens,
        "cache_hit_tokens": stat.cache_hit_tokens,
        "cache_miss_tokens": stat.cache_miss_tokens,
        "dur_ms": stat.dur_ms,
    }
    return json.dumps(d, ensure_ascii=False, sort_keys=False)
 
 
def iteration_stat_human_line(stat: IterationStat) -> str:
    """Compact one-line human-readable representation.
 
    Example::
 
        [14:32:01.123] iter=2 model=deepseek-v4-pro tools=[read_file,text_search]
                        p=12410 c=342 cache=11800/610 dur=1.84s
 
    The bracketed time is local wall-clock, matching the offset in
    ``stat.ts`` (e.g. the full JSONL emission is
    ``2026-05-17T14:32:01.123+02:00``). The compact display drops both
    the date and the offset to fit a per-iteration scrollback line.
    """
    # ``stat.ts`` is ``YYYY-MM-DDTHH:MM:SS.mmm±HH:MM``; index 11:23 is
    # the ``HH:MM:SS.mmm`` slice regardless of the trailing zone offset.
    ts_part = stat.ts[11:23]
    tools_part = ",".join(stat.tools) if stat.tools else "-"
    cache_part = (
        f"cache={stat.cache_hit_tokens}/{stat.cache_miss_tokens}"
        if stat.cache_hit_tokens or stat.cache_miss_tokens
        else ""
    )
    dur_s = f"{stat.dur_ms / 1000:.2f}s" if stat.dur_ms >= 1000 else f"{stat.dur_ms}ms"
 
    parts = [
        f"[{ts_part}]",
        f"iter={stat.iter}",
        f"model={stat.model}",
        f"tools=[{tools_part}]",
        f"p={stat.prompt_tokens}",
        f"c={stat.completion_tokens}",
    ]
    if cache_part:
        parts.append(cache_part)
    parts.append(f"dur={dur_s}")
 
    return "  ".join(parts)