Build your own glass box.

"""
glassbox.py -- Build your own glass box (VCN Glass Box).

Your agent is a black box: a prompt goes in, an answer comes out, and when it is
slow or wrong you are guessing. OpenTelemetry makes it a GLASS box. You wrap each
step (the LLM call, the tool call) in a SPAN; the spans nest into a TRACE; and a
waterfall of that trace shows you exactly where the latency, the cost, and the
errors live.

This file is a faithful, MINIMAL, stdlib-only OpenTelemetry-shaped tracer so the
lab boots on any machine with zero install and no API key. It mirrors the real
OTel data model (trace_id / span_id / parent_span_id / name / kind / start+end /
attributes / status / events) and the GenAI semantic conventions (gen_ai.*). It
prints a real span-tree waterfall to the console -- the same shape a real OTel
ConsoleSpanExporter or a backend like Jaeger / Grafana Tempo would show.

Durations are driven by a deterministic VIRTUAL CLOCK so the trace is identical
on every machine (great for a workshop). Swap in the real SDK for wall-clock
timing -- see SECTION 7 for the 3-line change.

Run it:
    python glassbox.py            # instrument the demo agent, print the trace
    python glassbox.py --slow     # make the tool call slow, watch it light up

ASCII-only on purpose: Windows stdout is cp1252 and crashes on a unicode
arrow / em-dash / curly-quote. Stdlib only. No network. No API key.
"""

import sys

# ---------------------------------------------------------------------------
# GROUNDING -- this mirrors the real OpenTelemetry model, not a toy.
# Sources (research/citations.yaml, verified via WebFetch 2026-06-25):
#   [OTEL-TRACES] OpenTelemetry Traces concepts -- trace = path of a request;
#       span = one unit of work with span context (trace id / span id),
#       parent span id, name, start/end, attributes, span events, status.
#       https://opentelemetry.io/docs/concepts/signals/traces/
#   [OTEL-GENAI]  OpenTelemetry GenAI semantic conventions (gen_ai.* registry).
#       STABILITY = "Development"/experimental -- keys can change.
#       https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/
# Note: this lab uses the CURRENT key gen_ai.provider.name (the older
# gen_ai.system is the superseded legacy alias).
# ---------------------------------------------------------------------------

try:
    sys.stdout.reconfigure(encoding="utf-8")
except Exception:
    pass


# ---------------------------------------------------------------------------
# 1. THE CLOCK  (deterministic virtual time -> reproducible traces)
# ---------------------------------------------------------------------------
# Real OTel reads the wall clock. We use a virtual clock advanced by explicit
# "work" so the printed waterfall is byte-identical on every machine. The span
# MODEL is real; only the time source is swapped for reproducibility.

class Clock:
    def __init__(self, base_ns=1_700_000_000_000_000_000):
        self.t = base_ns

    def now(self):
        return self.t

    def advance_ms(self, ms):
        self.t += int(ms * 1_000_000)


CLOCK = Clock()


# ---------------------------------------------------------------------------
# 2. THE SPAN  (one unit of work; the OTel data model)
# ---------------------------------------------------------------------------

# Deterministic but realistic-looking ids (real OTel uses random 128/64-bit ids;
# we seed from fixed hex + a step so the trace is identical on every run yet
# still reads like a real trace_id, not 0000...0001).
_TRACE_BASE = int("4bf92f3577b34da6a3ce929d0e0e4736", 16)
_SPAN_BASE = int("00f067aa0ba902b7", 16)
_next_id = {"trace": 0, "span": 0}


def _mk_id(kind, width):
    _next_id[kind] += 1
    base = _TRACE_BASE if kind == "trace" else _SPAN_BASE
    return ("%0*x" % (width, base + (_next_id[kind] - 1) * 0x1011))


STATUS_OK = "OK"
STATUS_ERROR = "ERROR"


class Span:
    def __init__(self, name, trace_id, span_id, parent_span_id, kind):
        self.name = name
        self.trace_id = trace_id
        self.span_id = span_id
        self.parent_span_id = parent_span_id
        self.kind = kind                 # INTERNAL / CLIENT / etc.
        self.start_ns = CLOCK.now()
        self.end_ns = None
        self.attributes = {}             # gen_ai.* and friends
        self.events = []                 # (name, attrs)
        self.status = STATUS_OK
        self.status_msg = ""

    def set(self, key, value):
        self.attributes[key] = value
        return self

    def add_event(self, name, **attrs):
        self.events.append((name, attrs))
        return self

    def error(self, msg):
        self.status = STATUS_ERROR
        self.status_msg = msg
        return self

    def duration_ms(self):
        return (self.end_ns - self.start_ns) / 1_000_000.0


# ---------------------------------------------------------------------------
# 3. THE TRACER  (starts spans, tracks parent/child, collects finished spans)
# ---------------------------------------------------------------------------

class Tracer:
    def __init__(self):
        self.trace_id = _mk_id("trace", 32)
        self._stack = []
        self.finished = []

    def span(self, name, kind="INTERNAL", **attributes):
        parent = self._stack[-1].span_id if self._stack else None
        sp = Span(name, self.trace_id, _mk_id("span", 16), parent, kind)
        for k, v in attributes.items():
            sp.attributes[k.replace("__", ".")] = v
        return _SpanCtx(self, sp)


class _SpanCtx:
    def __init__(self, tracer, span):
        self.tracer = tracer
        self.span = span

    def __enter__(self):
        self.tracer._stack.append(self.span)
        return self.span

    def __exit__(self, exc_type, exc, tb):
        self.span.end_ns = CLOCK.now()
        self.tracer._stack.pop()
        self.tracer.finished.append(self.span)
        if exc_type is not None:
            self.span.error("%s: %s" % (exc_type.__name__, exc))
        return False


# ---------------------------------------------------------------------------
# 4. THE EXPORTER  (console waterfall -- what a backend would render)
# ---------------------------------------------------------------------------

def _bar(start_off_ms, dur_ms, total_ms, width=34):
    if total_ms <= 0:
        return " " * width
    lead = int(round(start_off_ms / total_ms * width))
    span = max(1, int(round(dur_ms / total_ms * width)))
    lead = min(lead, width - 1)
    span = min(span, width - lead)
    return (" " * lead) + ("#" * span) + (" " * (width - lead - span))


def export_console(tracer, slow_ms=400.0):
    spans = sorted(tracer.finished, key=lambda s: (s.start_ns, s.span_id))
    if not spans:
        print("(no spans)")
        return
    t0 = min(s.start_ns for s in spans)
    t1 = max(s.end_ns for s in spans)
    total_ms = (t1 - t0) / 1_000_000.0

    # depth by walking parent links
    by_id = {s.span_id: s for s in spans}
    def depth(s):
        d, p = 0, s.parent_span_id
        while p is not None and p in by_id:
            d += 1
            p = by_id[p].parent_span_id
        return d

    print("TRACE %s   total %.0f ms   %d spans" % (tracer.trace_id[:16], total_ms, len(spans)))
    print("-" * 78)
    print("%-26s %9s  %-34s" % ("span", "dur", "waterfall"))
    print("-" * 78)
    for s in spans:
        off = (s.start_ns - t0) / 1_000_000.0
        d = s.duration_ms()
        flag = ""
        if s.status == STATUS_ERROR:
            flag = "  ERR"
        elif d >= slow_ms and s.parent_span_id is not None:
            # only flag leaf/child spans; the root is trivially the sum
            flag = "  SLOW"
        label = ("  " * depth(s)) + s.name
        print("%-26s %7.0fms  |%s|%s" % (label[:26], d, _bar(off, d, total_ms), flag))
    print("-" * 78)


def dump_spans(tracer):
    """Flat OTLP-ish dump: the raw fields a collector would receive."""
    print("")
    print("SPANS (OTLP-shaped) -- what gets exported to the collector")
    print("=" * 78)
    for s in sorted(tracer.finished, key=lambda s: s.start_ns):
        print("- name=%s kind=%s status=%s dur=%.0fms" % (s.name, s.kind, s.status, s.duration_ms()))
        print("  trace_id=%s span_id=%s parent=%s" % (s.trace_id[:16], s.span_id, s.parent_span_id))
        for k in sorted(s.attributes):
            print("  attr %s = %s" % (k, s.attributes[k]))
        for name, attrs in s.events:
            print("  event %s %s" % (name, attrs if attrs else ""))
        if s.status == STATUS_ERROR:
            print("  status_msg = %s" % s.status_msg)


# ---------------------------------------------------------------------------
# 5. THE AGENT  (a tiny LLM + tool loop -- the thing we instrument)
# ---------------------------------------------------------------------------
# This is a SIMULATED agent: deterministic, no API key. The point is the
# instrumentation, not the model. Each step advances the virtual clock to model
# its latency and is wrapped in a span carrying GenAI-semantic-convention attrs.

def fake_llm(tracer, prompt_tokens, completion, out_tokens, finish, latency_ms):
    with tracer.span("chat gpt-4o", kind="CLIENT",
                     gen_ai__provider__name="openai",
                     gen_ai__operation__name="chat",
                     gen_ai__request__model="gpt-4o",
                     gen_ai__request__temperature=0.7) as sp:
        CLOCK.advance_ms(latency_ms)
        sp.set("gen_ai.usage.input_tokens", prompt_tokens)
        sp.set("gen_ai.usage.output_tokens", out_tokens)
        sp.set("gen_ai.response.finish_reasons", [finish])
        sp.add_event("gen_ai.choice", finish_reason=finish)
        return completion


def fake_tool(tracer, name, args, result, latency_ms, fail=False):
    with tracer.span("execute_tool %s" % name, kind="INTERNAL",
                     tool__name=name,
                     tool__call__arguments=args) as sp:
        CLOCK.advance_ms(latency_ms)
        if fail:
            sp.error("ToolError: upstream 503")
            sp.add_event("exception", **{"exception.type": "ToolError"})
            return None
        sp.set("tool.call.result", result)
        return result


def run_agent(slow=False):
    tracer = Tracer()
    tool_latency = 1850.0 if slow else 240.0
    with tracer.span("invoke_agent", kind="INTERNAL",
                     gen_ai__operation__name="invoke_agent",
                     agent__name="weather-bot") as root:
        root.set("input.value", "What should I wear in San Francisco today?")
        # 1) LLM decides to call a tool
        fake_llm(tracer, prompt_tokens=812, completion="call get_weather",
                 out_tokens=19, finish="tool_calls", latency_ms=430.0)
        # 2) the tool runs (this is usually where the latency hides)
        weather = fake_tool(tracer, "get_weather", '{"city":"San Francisco"}',
                            '{"tempF":58,"cond":"foggy"}', latency_ms=tool_latency)
        # 3) LLM turns the tool result into a final answer
        fake_llm(tracer, prompt_tokens=case_tokens(weather), completion="Wear a light jacket; it is 58F and foggy.",
                 out_tokens=14, finish="stop", latency_ms=300.0)
        root.set("output.value", "Wear a light jacket; it is 58F and foggy.")
    return tracer


def case_tokens(weather):
    # the second LLM call carries the tool result back in -> bigger prompt
    return 851 if weather else 833


# ---------------------------------------------------------------------------
# 6. READ THE TRACE  (the payoff: latency, cost, errors at a glance)
# ---------------------------------------------------------------------------

def summarize(tracer):
    spans = tracer.finished
    total_ms = max(s.end_ns for s in spans) - min(s.start_ns for s in spans)
    total_ms /= 1_000_000.0
    # the useful insight is the slowest OPERATION (leaf span), not the root sum
    leaves = [s for s in spans if s.parent_span_id is not None] or spans
    slowest = max(leaves, key=lambda s: s.duration_ms())
    in_tok = sum(s.attributes.get("gen_ai.usage.input_tokens", 0) for s in spans)
    out_tok = sum(s.attributes.get("gen_ai.usage.output_tokens", 0) for s in spans)
    errors = [s for s in spans if s.status == STATUS_ERROR]
    print("")
    print("WHAT THE TRACE TELLS YOU")
    print("=" * 78)
    print("  end-to-end latency : %.0f ms" % total_ms)
    print("  slowest span       : %s (%.0f ms, %.0f%% of the trace)"
          % (slowest.name, slowest.duration_ms(), 100.0 * slowest.duration_ms() / total_ms))
    print("  token cost         : %d in + %d out = %d tokens" % (in_tok, out_tok, in_tok + out_tok))
    print("  errored spans      : %d%s" % (len(errors), (" (" + ", ".join(e.name for e in errors) + ")") if errors else ""))
    print("  -> the black box is now a glass box: you can SEE where the time and tokens go.")


# ---------------------------------------------------------------------------
# 7. SWAP IN THE REAL OTEL SDK  (when you want wall-clock + a real backend)
# ---------------------------------------------------------------------------
# pip install opentelemetry-sdk opentelemetry-exporter-otlp
#
#   from opentelemetry import trace
#   from opentelemetry.sdk.trace import TracerProvider
#   from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
#   trace.set_tracer_provider(TracerProvider())
#   trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
#   tracer = trace.get_tracer("glassbox")
#   with tracer.start_as_current_span("chat gpt-4o") as span:
#       span.set_attribute("gen_ai.system", "openai")   # same gen_ai.* keys as above
#
# Point the OTLP exporter at a collector (Jaeger / Grafana Tempo / Honeycomb /
# Langfuse) and the SAME spans render as a live waterfall in your browser.


def main(argv):
    slow = "--slow" in argv[1:]
    print("=" * 78)
    print("GLASS BOX  agent trace  (VCN: trace agents with OpenTelemetry)")
    if slow:
        print("mode: --slow  (the tool call is made slow on purpose)")
    print("=" * 78)
    print("")
    tracer = run_agent(slow=slow)
    export_console(tracer)
    summarize(tracer)
    dump_spans(tracer)
    print("")
    print("NOTE: durations come from a deterministic virtual clock so this trace")
    print("is identical on every machine. Swap in the real OTel SDK (section 7)")
    print("for wall-clock timing and a live backend.")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

python glassbox.py

python glassbox.py --slow