Glass box to glass house.

"""
a2a_trace.py -- Trace two agents talking (VCN Glass Box, A2A expansion).

Tracing ONE agent is table stakes. The real building is a SOCIETY of agents: one
agent calls another (agent-to-agent / A2A), and you want the WHOLE conversation
to be a single trace -- a glass HOUSE, not just a glass box.

The mechanic is W3C Trace Context. When Agent A calls Agent B, A's tracer
INJECTS its current context into the request as a `traceparent` header; B
EXTRACTS it and starts its spans as children of A's span. Because both carry the
same trace-id and B's root points at A's span-id, the two agents' work joins into
ONE multi-agent span waterfall.

This is a faithful, MINIMAL, stdlib-only model (no install, no API key, no
network). Deterministic virtual clock so the trace is identical on every machine.
Grounded in research/citations.yaml:
  [W3C-TRACECTX] traceparent = version-traceid-parentid-flags (e.g.
      00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01)
  [A2A]         Agent2Agent protocol v1.0.0 (client -> server, Message/Task)
  [OTEL-TRACES] the span model + context propagation
Note: OTel propagates context via traceparent; an A2A call is an HTTP call that
CAN carry it. (Whether A2A mandates it is not asserted here.)

Run it:
    python a2a_trace.py            # orchestrator calls researcher; print the joined trace
    python a2a_trace.py --broken   # the callee drops the traceparent -> two orphan traces

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

import sys

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


# ---------------------------------------------------------------------------
# virtual clock (deterministic -> reproducible trace)
# ---------------------------------------------------------------------------
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()

# deterministic, realistic-looking ids (real OTel uses random 128/64-bit ids)
_SPAN_BASE = int("00f067aa0ba902b7", 16)
_span_n = [0]
def new_span_id():
    _span_n[0] += 1
    return "%016x" % (_SPAN_BASE + (_span_n[0] - 1) * 0x1011)

# the trace-id is the canonical W3C Trace Context spec example
TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"


# ---------------------------------------------------------------------------
# W3C Trace Context  -- the carrier across the A2A boundary  [W3C-TRACECTX]
# ---------------------------------------------------------------------------
def format_traceparent(trace_id, span_id, sampled=True):
    """version - trace-id(32 hex) - parent-id(16 hex) - flags(2 hex)."""
    return "00-%s-%s-%02x" % (trace_id, span_id, 1 if sampled else 0)

def parse_traceparent(tp):
    version, trace_id, parent_id, flags = tp.split("-")
    return {"version": version, "trace_id": trace_id,
            "parent_id": parent_id, "sampled": int(flags, 16) & 1 == 1}


# ---------------------------------------------------------------------------
# span model + tracer (one tracer per agent/service)
# ---------------------------------------------------------------------------
ALL_SPANS = []   # the "collector": every agent exports here; joined by trace_id

class Span:
    def __init__(self, name, service, span_id, parent_span_id, kind):
        self.name = name
        self.service = service           # which agent emitted it
        self.trace_id = TRACE_ID
        self.span_id = span_id
        self.parent_span_id = parent_span_id
        self.kind = kind
        self.start_ns = CLOCK.now()
        self.end_ns = None
        self.attributes = {}
    def set(self, k, v):
        self.attributes[k] = v
        return self
    def duration_ms(self):
        return (self.end_ns - self.start_ns) / 1_000_000.0


class Tracer:
    """One per agent. remote_parent seeds the root from an extracted traceparent."""
    def __init__(self, service):
        self.service = service
        self._stack = []
    def span(self, name, kind="INTERNAL", remote_parent=None, **attrs):
        if self._stack:
            parent = self._stack[-1].span_id
        else:
            parent = remote_parent      # cross-agent link (or None for a true root)
        sp = Span(name, self.service, new_span_id(), parent, kind)
        for k, v in attrs.items():
            sp.attributes[k.replace("__", ".")] = v
        return _Ctx(self, sp)

class _Ctx:
    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):
        self.span.end_ns = CLOCK.now()
        self.tracer._stack.pop()
        ALL_SPANS.append(self.span)
        return False


# ---------------------------------------------------------------------------
# Agent B: the RESEARCHER (A2A server / remote agent)
# ---------------------------------------------------------------------------
def researcher_handle(incoming_traceparent):
    """An A2A server. EXTRACTS the traceparent so its spans join the caller's trace."""
    tracer = Tracer("researcher")
    ctx = parse_traceparent(incoming_traceparent) if incoming_traceparent else None
    remote_parent = ctx["parent_id"] if ctx else None
    # root span of the remote agent's work, parented at the caller's span id
    with tracer.span("handle_message researcher", kind="SERVER",
                     remote_parent=remote_parent,
                     a2a__role="agent", a2a__task__id="task-7f3a",
                     a2a__method="message/send") as root:
        if ctx:
            root.set("trace_context.extracted_from", "traceparent")
        CLOCK.advance_ms(40)
        with tracer.span("chat gpt-4o", kind="CLIENT",
                         gen_ai__provider__name="openai", gen_ai__operation__name="chat",
                         gen_ai__request__model="gpt-4o") as s:
            CLOCK.advance_ms(380)
            s.set("gen_ai.usage.input_tokens", 640)
            s.set("gen_ai.usage.output_tokens", 22)
            s.set("gen_ai.response.finish_reasons", ["tool_calls"])
        with tracer.span("execute_tool web_search", kind="INTERNAL",
                         tool__name="web_search",
                         tool__call__arguments='{"q":"OTel A2A trace context"}') as s:
            CLOCK.advance_ms(520)
            s.set("tool.call.result", '{"hits":3}')
        with tracer.span("chat gpt-4o", kind="CLIENT",
                         gen_ai__provider__name="openai", gen_ai__operation__name="chat",
                         gen_ai__request__model="gpt-4o") as s:
            CLOCK.advance_ms(300)
            s.set("gen_ai.usage.input_tokens", 712)
            s.set("gen_ai.usage.output_tokens", 18)
            s.set("gen_ai.response.finish_reasons", ["stop"])
        root.set("a2a.task.state", "TASK_STATE_COMPLETED")
    return "researched: context propagates via traceparent"


# ---------------------------------------------------------------------------
# Agent A: the ORCHESTRATOR (A2A client) -- calls the researcher
# ---------------------------------------------------------------------------
def run(broken=False):
    tracer = Tracer("orchestrator")
    with tracer.span("invoke orchestrator", kind="INTERNAL",
                     gen_ai__operation__name="invoke_agent",
                     agent__name="orchestrator") as root:
        root.set("input.value", "Research how A2A traces stay connected, then summarize.")
        with tracer.span("chat gpt-4o", kind="CLIENT",
                         gen_ai__provider__name="openai", gen_ai__operation__name="chat",
                         gen_ai__request__model="gpt-4o") as s:
            CLOCK.advance_ms(410)
            s.set("gen_ai.usage.input_tokens", 588)
            s.set("gen_ai.usage.output_tokens", 16)
            s.set("gen_ai.response.finish_reasons", ["tool_calls"])
        # THE A2A BOUNDARY: orchestrator calls researcher, injecting traceparent
        with tracer.span("a2a.call researcher", kind="CLIENT",
                         a2a__method="message/send", a2a__peer="researcher",
                         rpc__system="jsonrpc", server__address="researcher.agents.local") as call:
            tp = format_traceparent(TRACE_ID, call.span_id, sampled=True)
            call.set("traceparent.injected", tp)
            # the call travels the network; the callee extracts (or drops) it
            CLOCK.advance_ms(30)
            researcher_handle(None if broken else tp)
            CLOCK.advance_ms(20)
        with tracer.span("chat gpt-4o", kind="CLIENT",
                         gen_ai__provider__name="openai", gen_ai__operation__name="chat",
                         gen_ai__request__model="gpt-4o") as s:
            CLOCK.advance_ms(290)
            s.set("gen_ai.usage.input_tokens", 905)
            s.set("gen_ai.usage.output_tokens", 24)
            s.set("gen_ai.response.finish_reasons", ["stop"])
        root.set("output.value", "Summarized the researcher's findings for the user.")
        return tp


# ---------------------------------------------------------------------------
# exporter: the MULTI-AGENT waterfall (with an agent lane)
# ---------------------------------------------------------------------------
def depth_of(sp, by_id):
    d, p = 0, sp.parent_span_id
    while p is not None and p in by_id:
        d += 1
        p = by_id[p].parent_span_id
    return d

def print_waterfall():
    spans = sorted(ALL_SPANS, key=lambda s: (s.start_ns, s.span_id))
    by_id = {s.span_id: s for s in spans}
    # group by distinct trace root chains to detect orphans (broken propagation)
    roots = [s for s in spans if s.parent_span_id is None or s.parent_span_id not in by_id]
    t0 = min(s.start_ns for s in spans)
    t1 = max(s.end_ns for s in spans)
    total = (t1 - t0) / 1_000_000.0
    width = 30

    def bar(off, dur):
        lead = min(width - 1, int(round(off / total * width))) if total else 0
        span = max(1, int(round(dur / total * width))) if total else 1
        span = min(span, width - lead)
        return (" " * lead) + ("#" * span) + (" " * (width - lead - span))

    print("TRACE %s   total %.0f ms   %d spans across %d agent(s)"
          % (TRACE_ID[:16], total, len(spans), len(set(s.service for s in spans))))
    if len(roots) > 1:
        print("!! %d disconnected roots -> trace-context propagation BROKE (orphan spans)" % len(roots))
    print("-" * 92)
    print("%-12s %-26s %8s  %-30s" % ("agent", "span", "dur", "waterfall"))
    print("-" * 92)
    for s in spans:
        off = (s.start_ns - t0) / 1_000_000.0
        label = ("  " * depth_of(s, by_id)) + s.name
        boundary = "  <A2A" if s.name.startswith("a2a.call") else ""
        print("%-12s %-26s %6.0fms  |%s|%s" % (s.service, label[:26], s.duration_ms(), bar(off, s.duration_ms()), boundary))
    print("-" * 92)


def summarize(tp):
    spans = ALL_SPANS
    per_agent = {}
    for s in spans:
        per_agent.setdefault(s.service, 0.0)
    # sum leaf-ish time per agent (use each span's own duration for a rough split)
    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)
    agents = sorted(set(s.service for s in spans))
    print("")
    print("WHAT THE MULTI-AGENT TRACE TELLS YOU")
    print("=" * 92)
    print("  agents in trace    : " + ", ".join(agents))
    print("  the carrier        : traceparent = %s" % tp)
    print("    (version-traceid-parentid-flags; both agents share the trace-id above)")
    print("  token cost (all)   : %d in + %d out = %d tokens" % (in_tok, out_tok, in_tok + out_tok))
    print("  -> one conversation, two agents, ONE trace: the glass house.")


def main(argv):
    broken = "--broken" in argv[1:]
    print("=" * 92)
    print("A2A TRACE  two agents talking  (VCN Glass Box: glass box -> glass house)")
    if broken:
        print("mode: --broken  (callee DROPS the traceparent -> context propagation breaks)")
    print("=" * 92)
    print("")
    tp = run(broken=broken)
    print_waterfall()
    summarize(tp)
    # exit non-zero if propagation broke (orphan roots) in normal mode
    spans = ALL_SPANS
    by_id = {s.span_id: s for s in spans}
    roots = [s for s in spans if s.parent_span_id is None or s.parent_span_id not in by_id]
    if broken:
        return 0  # broken mode is a teaching demo; orphans are expected
    return 0 if len(roots) == 1 else 1


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

python a2a_trace.py

python a2a_trace.py --broken