File size: 2,161 Bytes
63c3c5c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """JSONL telemetry for HAKO's five mandated streams.
Streams (one append-only JSONL file each):
learning -- losses, effective learning rate, Lyapunov value
routing -- router action probs/entropy, MoE gates, load balance
orchestration -- cyclic-thinking potential P, cycles, GHSOM growth, N
diffusion -- game level, difficulty, reward, denoiser loss
thinking -- request decomposition/feedback/completeness trace
Proposition P-TEL (lossless auditability). Each record is a self-contained
JSON object with monotone field 't' inside its stream; because the writer is
append-only and atomic per line, any prefix of the file is a valid partial
trace of the run (crash-consistent audit).
"""
from __future__ import annotations
import json
import os
import time
from pathlib import Path
from typing import Any, Dict
_STREAMS = ("learning", "routing", "orchestration", "diffusion", "thinking")
class Telemetry:
def __init__(self, cfg) -> None:
self.dir = Path(cfg.telemetry_dir)
self.dir.mkdir(parents=True, exist_ok=True)
self.paths = {s: self.dir / f"{s}.jsonl" for s in _STREAMS}
self.counters = {s: 0 for s in _STREAMS}
self.enabled = True
def log(self, stream: str, record: Dict[str, Any]) -> None:
if not self.enabled or stream not in self.paths:
return
rec = {"t": self.counters[stream], "ts": time.time(), **record}
self.counters[stream] += 1
line = json.dumps(rec, ensure_ascii=True, default=float)
with open(self.paths[stream], "a", encoding="utf-8") as fh:
fh.write(line + "\n")
fh.flush()
os.fsync(fh.fileno())
# convenience wrappers -------------------------------------------------
def learning(self, **kw): self.log("learning", kw)
def routing(self, **kw): self.log("routing", kw)
def orchestration(self, **kw): self.log("orchestration", kw)
def diffusion(self, **kw): self.log("diffusion", kw)
def thinking(self, **kw): self.log("thinking", kw)
def summary(self) -> Dict[str, int]:
return {s: self.counters[s] for s in _STREAMS}
|