Skip to content

pyagent-studio API Reference

pyagent_studio

PyAgent Studio — Kubernetes Dashboard for Agent Systems.

BlueprintService

Headless service for loading, validating, and compiling blueprints.

Wraps pyagent-blueprint for use by TUI screens and tests.

Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
 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
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class BlueprintService:
    """Headless service for loading, validating, and compiling blueprints.

    Wraps ``pyagent-blueprint`` for use by TUI screens and tests.
    """

    def __init__(self) -> None:
        self._compiler = BlueprintCompiler()
        self._validator = BlueprintValidator()
        self._spec: BlueprintSpec | None = None
        self._graph: RuntimeGraph | None = None
        self._path: Path | None = None

    def load(self, path: str | Path) -> BlueprintSpec:
        """Load a blueprint from file.

        Args:
            path: Path to YAML/JSON blueprint.

        Returns:
            Validated ``BlueprintSpec``.

        Raises:
            BlueprintLoadError: On load/validation failure.
        """
        self._path = Path(path)
        self._spec = load_blueprint(self._path)
        self._graph = None
        return self._spec

    def validate(self) -> list[ValidationIssue]:
        """Run static validation on the loaded spec.

        Returns:
            List of validation issues.

        Raises:
            RuntimeError: If no spec loaded.
        """
        if self._spec is None:
            raise RuntimeError("No blueprint loaded. Call load() first.")
        return self._validator.validate(self._spec)

    def compile(self) -> RuntimeGraph:
        """Compile the loaded spec into a RuntimeGraph.

        Returns:
            Executable ``RuntimeGraph``.

        Raises:
            RuntimeError: If no spec loaded.
        """
        if self._spec is None:
            raise RuntimeError("No blueprint loaded. Call load() first.")
        self._graph = self._compiler.compile(self._spec)
        return self._graph

    def discover_blueprints(self, directory: str | Path = ".") -> list[Path]:
        """Find all YAML/JSON blueprint files in a directory.

        Args:
            directory: Root directory to search.

        Returns:
            List of file paths.
        """
        root = Path(directory)
        files: list[Path] = []
        for ext in ("*.yaml", "*.yml", "*.json"):
            files.extend(root.glob(f"**/{ext}"))
        return sorted(files)

    @property
    def spec(self) -> BlueprintSpec | None:
        return self._spec

    @property
    def graph(self) -> RuntimeGraph | None:
        return self._graph

    @property
    def path(self) -> Path | None:
        return self._path

    def summary(self) -> dict[str, Any]:
        """Quick summary of loaded blueprint."""
        if self._spec is None:
            return {"loaded": False}
        return {
            "loaded": True,
            "name": self._spec.metadata.name,
            "version": self._spec.metadata.version,
            "agents": len(self._spec.agents),
            "workflows": len(self._spec.workflows),
            "providers": len(self._spec.providers),
            "contracts": len(self._spec.contracts),
        }

load(path)

Load a blueprint from file.

Parameters:

Name Type Description Default
path str | Path

Path to YAML/JSON blueprint.

 required

Returns:

Type Description
BlueprintSpec

Validated BlueprintSpec.

Raises:

Type Description
BlueprintLoadError

On load/validation failure.
Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def load(self, path: str | Path) -> BlueprintSpec:
    """Load a blueprint from file.

    Args:
        path: Path to YAML/JSON blueprint.

    Returns:
        Validated ``BlueprintSpec``.

    Raises:
        BlueprintLoadError: On load/validation failure.
    """
    self._path = Path(path)
    self._spec = load_blueprint(self._path)
    self._graph = None
    return self._spec

validate()

Run static validation on the loaded spec.

Returns:

Type Description
list[ValidationIssue]

List of validation issues.

Raises:

Type Description
RuntimeError

If no spec loaded.
Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
50
51
52
53
54
55
56
57
58
59
60
61
def validate(self) -> list[ValidationIssue]:
    """Run static validation on the loaded spec.

    Returns:
        List of validation issues.

    Raises:
        RuntimeError: If no spec loaded.
    """
    if self._spec is None:
        raise RuntimeError("No blueprint loaded. Call load() first.")
    return self._validator.validate(self._spec)

compile()

Compile the loaded spec into a RuntimeGraph.

Returns:

Type Description
RuntimeGraph

Executable RuntimeGraph.

Raises:

Type Description
RuntimeError

If no spec loaded.
Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def compile(self) -> RuntimeGraph:
    """Compile the loaded spec into a RuntimeGraph.

    Returns:
        Executable ``RuntimeGraph``.

    Raises:
        RuntimeError: If no spec loaded.
    """
    if self._spec is None:
        raise RuntimeError("No blueprint loaded. Call load() first.")
    self._graph = self._compiler.compile(self._spec)
    return self._graph

discover_blueprints(directory='.')

Find all YAML/JSON blueprint files in a directory.

Parameters:

Name Type Description Default
directory str | Path

Root directory to search.

 '.'

Returns:

Type Description
list[Path]

List of file paths.
Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def discover_blueprints(self, directory: str | Path = ".") -> list[Path]:
    """Find all YAML/JSON blueprint files in a directory.

    Args:
        directory: Root directory to search.

    Returns:
        List of file paths.
    """
    root = Path(directory)
    files: list[Path] = []
    for ext in ("*.yaml", "*.yml", "*.json"):
        files.extend(root.glob(f"**/{ext}"))
    return sorted(files)

summary()

Quick summary of loaded blueprint.

Source code in packages/pyagent-studio/src/pyagent_studio/services/blueprint_service.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
def summary(self) -> dict[str, Any]:
    """Quick summary of loaded blueprint."""
    if self._spec is None:
        return {"loaded": False}
    return {
        "loaded": True,
        "name": self._spec.metadata.name,
        "version": self._spec.metadata.version,
        "agents": len(self._spec.agents),
        "workflows": len(self._spec.workflows),
        "providers": len(self._spec.providers),
        "contracts": len(self._spec.contracts),
    }

GovernanceService

Run governance checks on blueprints.

Combines validation, compliance scoring, and diffing.

Source code in packages/pyagent-studio/src/pyagent_studio/services/governance_service.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class GovernanceService:
    """Run governance checks on blueprints.

    Combines validation, compliance scoring, and diffing.
    """

    def __init__(self) -> None:
        self._validator = BlueprintValidator()
        self._differ = BlueprintDiffer()

    def check_compliance(self, spec: BlueprintSpec) -> ComplianceReport:
        """Run all validation checks and compute compliance score.

        Args:
            spec: Blueprint to check.

        Returns:
            ``ComplianceReport`` with score and issues.
        """
        issues = self._validator.validate(spec)
        # Count checks as: 6 standard categories
        total_checks = 6
        error_count = sum(1 for i in issues if i.severity == IssueSeverity.ERROR)
        passed = max(0, total_checks - error_count)
        score = passed / total_checks if total_checks > 0 else 1.0

        return ComplianceReport(
            total_checks=total_checks,
            passed=passed,
            issues=issues,
            score=score,
        )

    def diff(self, old: BlueprintSpec, new: BlueprintSpec) -> list[Change]:
        """Compute semantic diff between two specs.

        Args:
            old: Previous version.
            new: Updated version.

        Returns:
            List of ``Change`` objects.
        """
        return self._differ.diff(old, new)

    def diff_summary(self, old: BlueprintSpec, new: BlueprintSpec) -> str:
        """Human-readable diff summary.

        Args:
            old: Previous version.
            new: Updated version.

        Returns:
            Summary string.
        """
        changes = self.diff(old, new)
        return self._differ.summary(changes)

    def format_report(self, report: ComplianceReport) -> str:
        """Format a compliance report as text."""
        lines: list[str] = []
        lines.append(
            f"Compliance Score: {report.score:.0%} ({report.passed}/{report.total_checks})"
        )

        if report.issues:
            lines.append("\nIssues:")
            for issue in report.issues:
                lines.append(f"  [{issue.severity}] {issue.path}: {issue.message}")
        else:
            lines.append("No issues found.")

        return "\n".join(lines)

check_compliance(spec)

Run all validation checks and compute compliance score.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint to check.

 required

Returns:

Type Description
ComplianceReport

ComplianceReport with score and issues.
Source code in packages/pyagent-studio/src/pyagent_studio/services/governance_service.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def check_compliance(self, spec: BlueprintSpec) -> ComplianceReport:
    """Run all validation checks and compute compliance score.

    Args:
        spec: Blueprint to check.

    Returns:
        ``ComplianceReport`` with score and issues.
    """
    issues = self._validator.validate(spec)
    # Count checks as: 6 standard categories
    total_checks = 6
    error_count = sum(1 for i in issues if i.severity == IssueSeverity.ERROR)
    passed = max(0, total_checks - error_count)
    score = passed / total_checks if total_checks > 0 else 1.0

    return ComplianceReport(
        total_checks=total_checks,
        passed=passed,
        issues=issues,
        score=score,
    )

diff(old, new)

Compute semantic diff between two specs.

Parameters:

Name Type Description Default
old BlueprintSpec

Previous version.

 required
new BlueprintSpec

Updated version.

 required

Returns:

Type Description
list[Change]

List of Change objects.
Source code in packages/pyagent-studio/src/pyagent_studio/services/governance_service.py 
69
70
71
72
73
74
75
76
77
78
79
def diff(self, old: BlueprintSpec, new: BlueprintSpec) -> list[Change]:
    """Compute semantic diff between two specs.

    Args:
        old: Previous version.
        new: Updated version.

    Returns:
        List of ``Change`` objects.
    """
    return self._differ.diff(old, new)

diff_summary(old, new)

Human-readable diff summary.

Parameters:

Name Type Description Default
old BlueprintSpec

Previous version.

 required
new BlueprintSpec

Updated version.

 required

Returns:

Type Description
str

Summary string.
Source code in packages/pyagent-studio/src/pyagent_studio/services/governance_service.py 
81
82
83
84
85
86
87
88
89
90
91
92
def diff_summary(self, old: BlueprintSpec, new: BlueprintSpec) -> str:
    """Human-readable diff summary.

    Args:
        old: Previous version.
        new: Updated version.

    Returns:
        Summary string.
    """
    changes = self.diff(old, new)
    return self._differ.summary(changes)

format_report(report)

Format a compliance report as text.

Source code in packages/pyagent-studio/src/pyagent_studio/services/governance_service.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def format_report(self, report: ComplianceReport) -> str:
    """Format a compliance report as text."""
    lines: list[str] = []
    lines.append(
        f"Compliance Score: {report.score:.0%} ({report.passed}/{report.total_checks})"
    )

    if report.issues:
        lines.append("\nIssues:")
        for issue in report.issues:
            lines.append(f"  [{issue.severity}] {issue.path}: {issue.message}")
    else:
        lines.append("No issues found.")

    return "\n".join(lines)

SimulationService

Run simulations against compiled blueprints.

Uses MockLLM by default so no API keys are needed.

Source code in packages/pyagent-studio/src/pyagent_studio/services/simulation_service.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class SimulationService:
    """Run simulations against compiled blueprints.

    Uses ``MockLLM`` by default so no API keys are needed.
    """

    def __init__(
        self,
        compiler: BlueprintCompiler | None = None,
        event_bus: TraceEventBus | None = None,
        use_live: bool = False,
    ) -> None:
        self._compiler = compiler or BlueprintCompiler()
        self._event_bus = event_bus
        self._use_live = use_live

    async def run(
        self,
        spec: BlueprintSpec,
        workflow: str,
        task: str,
        use_live: bool | None = None,
    ) -> SimulationResult:
        """Run a single simulation.

        Args:
            spec: Blueprint spec.
            workflow: Workflow name to run.
            task: Input task string.

        Returns:
            ``SimulationResult`` with output and timing.
        """
        try:
            graph = self._compiler.compile(spec)
        except Exception as exc:
            return SimulationResult(
                workflow=workflow,
                input_task=task,
                success=False,
                error=f"Compilation failed: {exc}",
            )

        if self._event_bus:
            from pyagent_trace.events import TraceEvent

            self._event_bus.emit(
                TraceEvent(
                    timestamp=time.time(),
                    event_type="pattern_start",
                    pattern_type=workflow,
                    payload={
                        "task": task,
                        "live": use_live if use_live is not None else self._use_live,
                    },
                )
            )

        start = time.perf_counter()
        try:
            result = await graph.run(workflow, task)
            elapsed = (time.perf_counter() - start) * 1000
            return SimulationResult(
                workflow=workflow,
                input_task=task,
                output=result.output,
                elapsed_ms=elapsed,
                success=True,
            )
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            return SimulationResult(
                workflow=workflow,
                input_task=task,
                elapsed_ms=elapsed,
                success=False,
                error=f"{type(exc).__name__}: {exc}",
            )

    async def run_all(
        self,
        spec: BlueprintSpec,
        tasks: dict[str, str],
        use_live: bool | None = None,
    ) -> list[SimulationResult]:
        """Run simulations for multiple workflows.

        Args:
            spec: Blueprint spec.
            tasks: Mapping of workflow name → task string.

        Returns:
            List of simulation results.
        """
        results: list[SimulationResult] = []
        for workflow, task in tasks.items():
            result = await self.run(spec, workflow, task, use_live=use_live)
            results.append(result)
        return results

run(spec, workflow, task, use_live=None) async

Run a single simulation.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint spec.

 required
workflow str

Workflow name to run.

 required
task str

Input task string.

 required

Returns:

Type Description
SimulationResult

SimulationResult with output and timing.
Source code in packages/pyagent-studio/src/pyagent_studio/services/simulation_service.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def run(
    self,
    spec: BlueprintSpec,
    workflow: str,
    task: str,
    use_live: bool | None = None,
) -> SimulationResult:
    """Run a single simulation.

    Args:
        spec: Blueprint spec.
        workflow: Workflow name to run.
        task: Input task string.

    Returns:
        ``SimulationResult`` with output and timing.
    """
    try:
        graph = self._compiler.compile(spec)
    except Exception as exc:
        return SimulationResult(
            workflow=workflow,
            input_task=task,
            success=False,
            error=f"Compilation failed: {exc}",
        )

    if self._event_bus:
        from pyagent_trace.events import TraceEvent

        self._event_bus.emit(
            TraceEvent(
                timestamp=time.time(),
                event_type="pattern_start",
                pattern_type=workflow,
                payload={
                    "task": task,
                    "live": use_live if use_live is not None else self._use_live,
                },
            )
        )

    start = time.perf_counter()
    try:
        result = await graph.run(workflow, task)
        elapsed = (time.perf_counter() - start) * 1000
        return SimulationResult(
            workflow=workflow,
            input_task=task,
            output=result.output,
            elapsed_ms=elapsed,
            success=True,
        )
    except Exception as exc:
        elapsed = (time.perf_counter() - start) * 1000
        return SimulationResult(
            workflow=workflow,
            input_task=task,
            elapsed_ms=elapsed,
            success=False,
            error=f"{type(exc).__name__}: {exc}",
        )

run_all(spec, tasks, use_live=None) async

Run simulations for multiple workflows.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint spec.

 required
tasks dict[str, str]

Mapping of workflow name → task string.

 required

Returns:

Type Description
list[SimulationResult]

List of simulation results.
Source code in packages/pyagent-studio/src/pyagent_studio/services/simulation_service.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def run_all(
    self,
    spec: BlueprintSpec,
    tasks: dict[str, str],
    use_live: bool | None = None,
) -> list[SimulationResult]:
    """Run simulations for multiple workflows.

    Args:
        spec: Blueprint spec.
        tasks: Mapping of workflow name → task string.

    Returns:
        List of simulation results.
    """
    results: list[SimulationResult] = []
    for workflow, task in tasks.items():
        result = await self.run(spec, workflow, task, use_live=use_live)
        results.append(result)
    return results

TraceService

Load and query trace JSONL files from pyagent-trace Recorder.

Each line in a JSONL file is a JSON object representing a trace event.

Source code in packages/pyagent-studio/src/pyagent_studio/services/trace_service.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class TraceService:
    """Load and query trace JSONL files from pyagent-trace Recorder.

    Each line in a JSONL file is a JSON object representing a trace event.
    """

    def __init__(self) -> None:
        self._spans: list[TraceSpan] = []
        self._path: Path | None = None

    def load(self, path: str | Path) -> list[TraceSpan]:
        """Load spans from a JSONL file.

        Args:
            path: Path to ``.jsonl`` trace file.

        Returns:
            List of parsed ``TraceSpan`` objects.

        Raises:
            FileNotFoundError: If file doesn't exist.
        """
        self._path = Path(path)
        if not self._path.exists():
            raise FileNotFoundError(f"Trace file not found: {self._path}")

        self._spans = []
        for line in self._path.read_text().splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                data = json.loads(line)
                span = TraceSpan(
                    event_type=data.get("event_type", data.get("type", "unknown")),
                    agent_name=data.get("agent_name", data.get("agent", "")),
                    timestamp=data.get("timestamp", ""),
                    duration_ms=data.get("duration_ms", 0.0),
                    tokens=data.get("tokens", data.get("token_count", 0)),
                    data=data,
                )
                self._spans.append(span)
            except json.JSONDecodeError:
                continue

        return self._spans

    def query(
        self,
        event_type: str | None = None,
        agent_name: str | None = None,
    ) -> list[TraceSpan]:
        """Filter loaded spans.

        Args:
            event_type: Filter by event type.
            agent_name: Filter by agent name.

        Returns:
            Filtered list of spans.
        """
        results = self._spans
        if event_type:
            results = [s for s in results if s.event_type == event_type]
        if agent_name:
            results = [s for s in results if s.agent_name == agent_name]
        return results

    @property
    def spans(self) -> list[TraceSpan]:
        return list(self._spans)

    def summary(self) -> dict[str, Any]:
        """Summary statistics for loaded traces."""
        if not self._spans:
            return {"loaded": False, "count": 0}
        return {
            "loaded": True,
            "count": len(self._spans),
            "event_types": sorted({s.event_type for s in self._spans}),
            "agents": sorted({s.agent_name for s in self._spans if s.agent_name}),
            "total_tokens": sum(s.tokens for s in self._spans),
            "total_duration_ms": sum(s.duration_ms for s in self._spans),
        }

load(path)

Load spans from a JSONL file.

Parameters:

Name Type Description Default
path str | Path

Path to .jsonl trace file.

 required

Returns:

Type Description
list[TraceSpan]

List of parsed TraceSpan objects.

Raises:

Type Description
FileNotFoundError

If file doesn't exist.
Source code in packages/pyagent-studio/src/pyagent_studio/services/trace_service.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def load(self, path: str | Path) -> list[TraceSpan]:
    """Load spans from a JSONL file.

    Args:
        path: Path to ``.jsonl`` trace file.

    Returns:
        List of parsed ``TraceSpan`` objects.

    Raises:
        FileNotFoundError: If file doesn't exist.
    """
    self._path = Path(path)
    if not self._path.exists():
        raise FileNotFoundError(f"Trace file not found: {self._path}")

    self._spans = []
    for line in self._path.read_text().splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            data = json.loads(line)
            span = TraceSpan(
                event_type=data.get("event_type", data.get("type", "unknown")),
                agent_name=data.get("agent_name", data.get("agent", "")),
                timestamp=data.get("timestamp", ""),
                duration_ms=data.get("duration_ms", 0.0),
                tokens=data.get("tokens", data.get("token_count", 0)),
                data=data,
            )
            self._spans.append(span)
        except json.JSONDecodeError:
            continue

    return self._spans

query(event_type=None, agent_name=None)

Filter loaded spans.

Parameters:

Name Type Description Default
event_type str | None

Filter by event type.

 None
agent_name str | None

Filter by agent name.

 None

Returns:

Type Description
list[TraceSpan]

Filtered list of spans.
Source code in packages/pyagent-studio/src/pyagent_studio/services/trace_service.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def query(
    self,
    event_type: str | None = None,
    agent_name: str | None = None,
) -> list[TraceSpan]:
    """Filter loaded spans.

    Args:
        event_type: Filter by event type.
        agent_name: Filter by agent name.

    Returns:
        Filtered list of spans.
    """
    results = self._spans
    if event_type:
        results = [s for s in results if s.event_type == event_type]
    if agent_name:
        results = [s for s in results if s.agent_name == agent_name]
    return results

summary()

Summary statistics for loaded traces.

Source code in packages/pyagent-studio/src/pyagent_studio/services/trace_service.py 
104
105
106
107
108
109
110
111
112
113
114
115
def summary(self) -> dict[str, Any]:
    """Summary statistics for loaded traces."""
    if not self._spans:
        return {"loaded": False, "count": 0}
    return {
        "loaded": True,
        "count": len(self._spans),
        "event_types": sorted({s.event_type for s in self._spans}),
        "agents": sorted({s.agent_name for s in self._spans if s.agent_name}),
        "total_tokens": sum(s.tokens for s in self._spans),
        "total_duration_ms": sum(s.duration_ms for s in self._spans),
    }