# SPDX-License-Identifier: MIT
"""Render a deterministic SVG from the sanitized Langfuse evidence."""
from __future__ import annotations
import argparse
from html import escape
import json
from pathlib import Path
from typing import Dict, List
from run_integration import assert_shape
def scenario_map(payload: Dict[str, object], name: str) -> Dict[str, object]:
scenarios = payload["scenarios"]
if not isinstance(scenarios, list):
raise ValueError("evidence has no scenario list")
for scenario in scenarios:
if isinstance(scenario, dict) and scenario.get("scenario") == name:
return scenario
raise ValueError(f"missing scenario: {name}")
def observation_map(scenario: Dict[str, object]) -> Dict[str, Dict[str, object]]:
observations = scenario["observations"]
if not isinstance(observations, list):
raise ValueError("scenario has no observations")
assert_shape(str(scenario["scenario"]), observations)
return {
str(item["metadata"]["logical_observation_id"]): item
for item in observations
}
def card(
x: int,
y: int,
width: int,
title: str,
detail: str,
color: str,
) -> str:
return f"""
{escape(title)}
{escape(detail)}
"""
def render(payload: Dict[str, object]) -> str:
retry = observation_map(scenario_map(payload, "retry"))
terminal = observation_map(scenario_map(payload, "terminal"))
versions = payload["versions"]
unavailable = payload["unavailableExporter"]
if not isinstance(versions, dict) or not isinstance(unavailable, dict):
raise ValueError("evidence manifest is incomplete")
rows = [
("retrieval-1", "retrieve-account-policy", "retrieval"),
("plan-1", "draft-plan", "recorded generation"),
("subscription-1", "get-subscription · attempt 1", "timeout retained"),
("subscription-2", "get-subscription · attempt 2", "success retained"),
("refund-1", "get-refund-window", "tool success"),
("answer-1", "compose-answer", "recorded generation"),
]
palette = {
"ERROR": "#ef6a6a",
"WARNING": "#e2a832",
"DEFAULT": "#5ad1a4",
}
cards: List[str] = []
connectors: List[str] = []
root_y = 190
child_x = 230
child_width = 600
for index, (logical_id, title, note) in enumerate(rows):
item = retry[logical_id]
y = 320 + index * 76
level = str(item["level"])
color = palette[level]
detail = f"{item['type']} · {level} · {note}"
cards.append(card(child_x, y, child_width, title, detail, color))
center_y = y + 30
connectors.append(
f''
)
terminal_refund = terminal["refund-1"]
terminal_root = terminal["task-1"]
sibling_levels = [
terminal[logical_id]["level"]
for logical_id in ("retrieval-1", "plan-1", "answer-1")
]
if any(level == "ERROR" for level in sibling_levels):
raise AssertionError("terminal scenario contains a failed sibling")
return f"""
"""
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--input", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
return parser.parse_args()
if __name__ == "__main__":
arguments = parse_args()
evidence = json.loads(arguments.input.read_text(encoding="utf-8"))
arguments.output.write_text(render(evidence), encoding="utf-8")