chemenu/tools/chemenu/telemetry/schema.py

"""The trace event contract: one JSON object per line, one line per event.

Why the schema lives in code rather than under `types/`: `types/` is the page
type system - every spec there describes a `kb/` page, carries a `base_dir`, and
is discovered by `wikitool types list`. A trace event is not a page, so putting
it there would put a non-page into the page catalog.

Ordering. Events are appended by several processes at once (the CLI in one
process, a hook handler in another, one per tool call), so `seq` is a
*per-process* counter and cannot be compared across processes. Sort a trace by
`(ts, pid, seq)`.

Degradation. No consumer may require an event class that some harness cannot
produce - Mistral Vibe has three hooks where Claude Code has thirty. `session.start`
carries a `completeness` list naming the classes its harness can emit, so a scorer
can say "not measurable here" instead of silently scoring zero. The common core
every surface provides is `tool.pre`, `tool.post`, `wikitool.call`, `gate.refused`.
"""
from __future__ import annotations

import os
from datetime import datetime, timezone

SCHEMA_VERSION = 1

# Where an event came from. `runner` is the eval runner itself, which owns the
# session boundaries because not every harness reports them.
SOURCES = frozenset(
    {
        "wikitool",
        "runner",
        "claude-code",
        "copilot-cli",
        "vscode-chat",
        "mistral-vibe",
    }
)

# The common core, available on every surface. Scorers may depend on these.
CORE_EVENTS = frozenset({"tool.pre", "tool.post", "wikitool.call", "gate.refused"})

# Everything else refines a scorer but may never be a precondition for one.
OPTIONAL_EVENTS = frozenset(
    {
        "session.start",
        "session.end",
        "session.error",
        "prompt.submitted",
        "assistant.message",
        "turn.end",
        "tool.error",
        "instructions.loaded",
        "subagent.start",
        "subagent.stop",
        "compaction",
        "page.written",
        "publish.commit",
        "budget.state",
        "gate.cleared",
    }
)

EVENTS = CORE_EVENTS | OPTIONAL_EVENTS

# Which event classes each harness can actually produce, from its documented
# hook surface. Written into `session.start` as `completeness` so a trace is
# self-describing: a missing class means "this harness cannot report it", not
# "the agent never did it".
HARNESS_CAPABILITIES: dict[str, tuple[str, ...]] = {
    "claude-code": (
        "session.start",
        "session.end",
        "prompt.submitted",
        "tool.pre",
        "tool.post",
        "tool.error",
        "turn.end",
        "instructions.loaded",
        "subagent.start",
        "subagent.stop",
        "compaction",
    ),
    "copilot-cli": (
        "session.start",
        "session.end",
        "session.error",
        "prompt.submitted",
        "tool.pre",
        "tool.post",
        "tool.error",
        "turn.end",
        "subagent.start",
        "subagent.stop",
        "compaction",
    ),
    # Three hooks only: pre_tool, post_tool, post_agent. No session, prompt,
    # compaction, permission or subagent lifecycle event exists to hook.
    "mistral-vibe": ("tool.pre", "tool.post", "turn.end"),
    # Post-hoc import from the chronicle store: turns and touched files, no
    # tool-level lifecycle.
    "vscode-chat": ("session.start", "prompt.submitted", "assistant.message", "tool.post"),
    "wikitool": (
        "wikitool.call",
        "gate.refused",
        "page.written",
        "publish.commit",
        "budget.state",
        "gate.cleared",
    ),
    "runner": ("session.start", "session.end"),
}


def utc_now_iso() -> str:
    """Timestamp with microseconds, so events inside one millisecond still order."""
    return datetime.now(timezone.utc).isoformat(timespec="microseconds")


def make_event(
    source: str,
    event: str,
    attrs: dict | None = None,
    *,
    session_id: str,
    seq: int,
    run_key: str | None = None,
    ts: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
) -> dict:
    record = {
        "v": SCHEMA_VERSION,
        "ts": ts or utc_now_iso(),
        "session_id": session_id,
        "pid": os.getpid(),
        "seq": seq,
        "source": source,
        "event": event,
        "attrs": attrs or {},
    }
    if run_key:
        record["run_key"] = run_key
    if trace_id:
        record["trace_id"] = trace_id
    if span_id:
        record["span_id"] = span_id
    return record


def validation_errors(record: dict) -> list[str]:
    """Return a list of contract violations; empty means valid.

    Deliberately hand-written rather than jsonschema: this runs on the hot path
    of every hook invocation, and the telemetry package must stay stdlib-only so
    a hook can run it without the venv.
    """
    errors: list[str] = []
    for field in ("v", "ts", "session_id", "pid", "seq", "source", "event", "attrs"):
        if field not in record:
            errors.append(f"missing required field '{field}'")
    if record.get("v") != SCHEMA_VERSION:
        errors.append(f"unknown schema version {record.get('v')!r}")
    if record.get("source") not in SOURCES:
        errors.append(f"unknown source {record.get('source')!r}")
    if record.get("event") not in EVENTS:
        errors.append(f"unknown event {record.get('event')!r}")
    if not isinstance(record.get("attrs", {}), dict):
        errors.append("'attrs' must be an object")
    if not isinstance(record.get("seq"), int):
        errors.append("'seq' must be an integer")
    if not isinstance(record.get("session_id"), str) or not record.get("session_id"):
        errors.append("'session_id' must be a non-empty string")
    return errors