report.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/timing/report.py
 
"""
Purpose
-------
Report stage and session timings against host baselines.
 
Responsibilities
----------------
- Keep local timing data bounded, private and advisory.
 
Diagnostics
-----------
Domain: GENERAL
Levels:
    L1 -- instrumentation failures
    L2 -- lifecycle
    L3 -- semantic details
    L4 -- deep tracing
 
Contracts
---------
- Timing instrumentation never changes the wrapped operation's outcome.
- Only the fixed timing schema is persisted; command arguments are never stored.
 
Dependencies
------------
- oc_diagnostics (structured timing diagnostics)
- oct.core.octrc (load_octrc_safe for the project's tier budget overrides)
- oct.timing.ledger (read_records for the recorded stage timings)
- oct.timing.tiers (each stage's tier and effective tier budget)
- oct.tools.atomic_write (atomic_write_text for saving the host timing baseline)
"""
from __future__ import annotations
 
from collections import defaultdict
import json
import math
import os
from pathlib import Path
import socket
from statistics import median
from datetime import datetime
from oc_diagnostics import _dbg
from oct.core.octrc import load_octrc_safe
from oct.timing.ledger import read_records
from oct.timing.tiers import budget_for, budgets_ms, tier_for
from oct.tools.atomic_write import atomic_write_text
 
def statistics(values: list[float]) -> dict:
    """Return median and nearest-rank p95 (including small samples)."""
    ordered = sorted(values)
    return {"median_ms": median(ordered), "p95_ms": ordered[math.ceil(0.95 * len(ordered)) - 1], "count": len(ordered)}
 
def _load_baseline(root: Path) -> dict:
    """Read baseline strictly so a malformed file cannot be overwritten."""
    path = root / ".option_c" / "timing_baseline.json"
    if not path.exists():
        return {"hosts": {}}
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict) or not isinstance(data.get("hosts"), dict):
        raise ValueError("invalid timing baseline")
    return data
 
def timing_report(root: Path, last: int = 20) -> dict:
    """Aggregate the current host and CPU count, excluding deliberate waits."""
    func = "timing_report"
    _dbg("GENERAL", func, "aggregate timing records", 3)
    host, cpus = socket.gethostname(), os.cpu_count() or 1
    groups = defaultdict(list)
    for row in read_records(root):
        value = row.get("wall_ms")
        if row.get("host") == host and row.get("cpus") == cpus and row.get("kind") == "stage" and isinstance(row.get("stage"), str) and type(value) in (int, float) and math.isfinite(value) and value >= 0:
            groups[row["stage"]].append(value)
    baseline = _load_baseline(root)
    entry = baseline["hosts"].get(host, {}).get(str(cpus))
    timing = load_octrc_safe(root).get("timing")
    tier_budgets = budgets_ms(timing.get("budgets_ms") if isinstance(timing, dict) else None)
    stages = {}
    for stage, values in sorted(groups.items()):
        numbers = statistics(values[-last:])
        notes = []
        reference = (entry or {}).get("stages", {}).get(stage)
        if reference:
            tolerance = reference.get("tolerance", (entry or {}).get("tolerance", 25))
            if numbers["median_ms"] > reference["median_ms"] * (1 + tolerance / 100):
                notes.append("median exceeds baseline tolerance")
        budget = (entry or {}).get("budgets", baseline.get("budgets", {})).get(stage)
        if budget is not None and numbers["median_ms"] > budget:
            notes.append("median exceeds budget")
        tier_budget = budget_for(stage, tier_budgets)
        stages[stage] = {**numbers, "baseline_median_ms": reference["median_ms"] if reference else None, "advisories": notes,
                         "tier": tier_for(stage), "budget_ms": tier_budget,
                         "note": "over budget" if tier_budget is not None and numbers["median_ms"] > tier_budget else ""}
    return {"host": host, "cpus": cpus, "last": last, "baseline_status": "matched" if entry is not None else "no baseline for this host", "stages": stages}
 
def write_baseline(root: Path, report: dict, *, force_host: bool = False) -> None:
    """Preserve all hosts; require acknowledgment before adding a foreign host."""
    func = "write_baseline"
    baseline = _load_baseline(root)
    hosts = baseline["hosts"]
    host, cpus = report["host"], str(report["cpus"])
    if hosts and cpus not in hosts.get(host, {}) and not force_host:
        raise ValueError("baseline belongs to another host or CPU count; use --force-host")
    entry = hosts.setdefault(host, {}).setdefault(cpus, {})
    entry.setdefault("tolerance", 25)
    entry["stages"] = {stage: {"median_ms": values["median_ms"], "count": values["count"]} for stage, values in report["stages"].items()}
    path = root / ".option_c" / "timing_baseline.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_text(path, json.dumps(baseline, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    _dbg("GENERAL", func, "wrote host timing baseline", 3)
 
def session_report(path: Path) -> dict:
    """Pair calls by session and call ID and sum token samples per turn."""
    func = "session_report"
    starts = {}
    tools = defaultdict(list)
    turns = {}
    seen = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        try:
            event = json.loads(line)
            if not isinstance(event, dict):
                continue
            identity = event.get("event_id")
            if identity and identity in seen:
                continue
            if identity:
                seen.add(identity)
            payload = event.get("payload", {})
            kind = event.get("kind")
            key = (event.get("session_id"), event.get("tool_call_id") or payload.get("tool_call_id"))
            if kind == "tool_call_started":
                starts[key] = (datetime.fromisoformat(event["ts"]), payload["tool_name"])
            elif kind == "tool_call_completed" and key in starts:
                started, name = starts.pop(key)
                elapsed = (datetime.fromisoformat(event["ts"]) - started).total_seconds() * 1000
                if elapsed >= 0:
                    tools[name].append(elapsed)
            elif kind == "provider_budget_consumed":
                turn = event.get("turn_id") or "unknown"
                numbers = turns.setdefault(turn, dict(input_tokens=0, output_tokens=0, total_tokens=0))
                for field in ("input_tokens", "output_tokens"):
                    value = payload.get(field, 0)
                    if type(value) is int and value >= 0:
                        numbers[field] += value
                        numbers["total_tokens"] += value
        except (ValueError, KeyError, TypeError, AttributeError):
            # OCT-LINT: disable=no_hardcoded_secrets reason="Fixed diagnostic message; no credential material"
            _dbg("GENERAL", func, "VALIDATION_ERROR: invalid session timing row", 3)
    return {"tools": {name: statistics(values) for name, values in sorted(tools.items())}, "turns": turns}
 
def render_report(report: dict) -> str:
    """Render precisely the same statistics exposed in JSON."""
    lines = ["Timing", report["baseline_status"], "stage: median ms | p95 ms | count | baseline median ms | tier | budget ms | note"]
    for stage, values in report["stages"].items():
        lines.append(f"{stage}: {values['median_ms']:g} | {values['p95_ms']:g} | {values['count']} | {values['baseline_median_ms']} | {values['tier']} | {values['budget_ms']} | {values['note']}")
        lines.extend(f"  advisory: {note}" for note in values["advisories"])
    session = report.get("session", {})
    if session:
        lines.append("Session tools: median ms | p95 ms | count")
        for name, values in session["tools"].items():
            lines.append(f"{name}: {values['median_ms']:g} | {values['p95_ms']:g} | {values['count']}")
        lines.append("Tokens per turn: input | output | total")
        for turn, values in session["turns"].items():
            lines.append(f"{turn}: {values['input_tokens']} | {values['output_tokens']} | {values['total_tokens']}")
    return "\n".join(lines)