hack-house/hh/scripts/bench/olympics/transcript.py

"""Replayable, OTel-aligned event log for one (team, event).

Every room frame, model call, tool call, phase change, ACL grant and guard
verdict becomes one ``Event`` whose ``kind`` follows the OpenTelemetry GenAI
semantic conventions (``message`` / ``invoke_agent`` / ``execute_tool`` /
``phase`` / ``acl`` / ``guard``). The transcript is a pure record: re-rendering
it (``replay``) or re-judging it under a new rubric is a function of this file
alone, which is the SPEC's reproducibility contract (§11).

A ``Transcript`` also carries a ``manifest`` — the config hash, seed, model
versions, challenge id and budget — so a result is self-describing.
"""

from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any

# kinds, aligned to OTel GenAI agent spans
KIND_MESSAGE = "message"          # a chat utterance posted into the room
KIND_AGENT = "invoke_agent"       # a model inference call
KIND_TOOL = "execute_tool"        # a VM command / code execution
KIND_PHASE = "phase"              # a phase transition of the loop
KIND_ACL = "acl"                  # a drive-grant / revoke
KIND_GUARD = "guard"              # a destructive-guard / safety verdict


@dataclass
class Event:
    ts: float
    kind: str
    phase: str
    actor: str
    payload: dict[str, Any] = field(default_factory=dict)
    role: str = ""
    model: str = ""
    tokens: dict[str, int] = field(default_factory=dict)


class Transcript:
    """An append-only event log for one team's run at one event."""

    def __init__(self, team: str, challenge: str, manifest: dict | None = None):
        self.team = team
        self.challenge = challenge
        self.manifest = manifest or {}
        self.events: list[Event] = []
        self._t0 = time.time()

    def add(self, kind: str, phase: str, actor: str, payload: dict | None = None,
            *, role: str = "", model: str = "", tokens: dict | None = None) -> Event:
        ev = Event(ts=round(time.time() - self._t0, 3), kind=kind, phase=phase,
                   actor=actor, payload=payload or {}, role=role, model=model,
                   tokens=tokens or {})
        self.events.append(ev)
        return ev

    # convenience emitters --------------------------------------------------
    def message(self, phase, actor, text, *, role="", model="", tokens=None):
        return self.add(KIND_MESSAGE, phase, actor, {"text": text},
                        role=role, model=model, tokens=tokens)

    def phase_change(self, phase, note=""):
        return self.add(KIND_PHASE, phase, "referee", {"note": note})

    def tool(self, phase, actor, payload, *, role=""):
        return self.add(KIND_TOOL, phase, actor, payload, role=role)

    def acl(self, phase, payload):
        return self.add(KIND_ACL, phase, "referee", payload)

    def guard(self, phase, actor, payload):
        return self.add(KIND_GUARD, phase, actor, payload)

    # tokens accounting -----------------------------------------------------
    def total_tokens(self) -> dict[str, int]:
        agg = {"in": 0, "out": 0}
        for ev in self.events:
            agg["in"] += ev.tokens.get("in", 0)
            agg["out"] += ev.tokens.get("out", 0)
        return agg

    # persistence -----------------------------------------------------------
    def to_dict(self) -> dict:
        return {"team": self.team, "challenge": self.challenge,
                "manifest": self.manifest,
                "tokens_total": self.total_tokens(),
                "events": [asdict(e) for e in self.events]}

    def save(self, path: str | Path) -> Path:
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(self.to_dict(), indent=2))
        return p

    @classmethod
    def load(cls, path: str | Path) -> "Transcript":
        data = json.loads(Path(path).read_text())
        t = cls(data["team"], data["challenge"], data.get("manifest", {}))
        t.events = [Event(**e) for e in data["events"]]
        return t


def replay(path: str | Path, *, show_tools: bool = True) -> None:
    """Re-render a saved transcript as a readable room log for audit."""
    t = Transcript.load(path)
    print("=" * 76)
    print(f"replay · team={t.team} · challenge={t.challenge}")
    m = t.manifest
    if m:
        print(f"  models={m.get('models')} seed={m.get('seed')} "
              f"budget={m.get('budget')}")
    print("-" * 76)
    for ev in t.events:
        stamp = f"[{ev.ts:7.2f}s {ev.phase:<10}]"
        if ev.kind == KIND_PHASE:
            print(f"{stamp} ── phase: {ev.phase} {ev.payload.get('note', '')}")
        elif ev.kind == KIND_MESSAGE:
            print(f"{stamp} {ev.actor}({ev.role}): {ev.payload.get('text', '')}")
        elif ev.kind == KIND_AGENT:
            print(f"{stamp} ~ {ev.actor} infer ({ev.model}) "
                  f"tok={ev.tokens.get('out', 0)}")
        elif ev.kind == KIND_ACL:
            print(f"{stamp} ⚿ acl {ev.payload}")
        elif ev.kind == KIND_GUARD:
            print(f"{stamp} ⛨ guard {ev.payload}")
        elif ev.kind == KIND_TOOL and show_tools:
            p = ev.payload
            print(f"{stamp} ⛧ {ev.actor} tool rc={p.get('rc')} "
                  f"{p.get('label', '')}")
            out = (p.get("out") or "").strip()
            if out:
                for line in out.splitlines()[:8]:
                    print(f"{'':>22}| {line}")
    print("-" * 76)
    print(f"{len(t.events)} events · tokens={t.total_tokens()}")
    print("=" * 76)