harbor-atif2otel

Convert ATIF agent trajectories to OpenTelemetry spans for visualization in any OTel-compatible backend.

Install

pip install harbor-atif2otel

Quick Start

import json
from harbor_atif2otel import convert_trajectory
from harbor_atif2otel.uploaders.mlflow_protobuf import MlflowProtobufUploader

# Load an ATIF trajectory
with open("trajectory.json") as f:
    trajectory = json.load(f)

# Convert to OTel ResourceSpans
resource_spans = convert_trajectory(trajectory)

# Upload to MLflow
uploader = MlflowProtobufUploader(
    endpoint="https://mlflow.example.com",
    experiment_name="my-eval",
    token="my-auth-token",
    workspace="default",
)
uploader.upload(resource_spans)

API

convert_trajectory(trajectory, trace_seed=None, service_name="harbor", max_attribute_bytes=10240)

Convert a single ATIF trajectory dict to an OTel ResourceSpans protobuf.

• trajectory: Parsed ATIF JSON (dict)
• trace_seed: Optional seed for deterministic trace/span IDs. Defaults to session_id or trajectory_id.
• service_name: OTel resource service.name attribute
• max_attribute_bytes: Truncation limit for large string attributes

Returns an opentelemetry.proto.trace.v1.trace_pb2.ResourceSpans.

convert_trajectories(trajectories, **kwargs)

Batch convert. Returns list[ResourceSpans].

validate_trajectory(trajectory)

Validate an ATIF trajectory dict. Returns list[str] of issues (empty = valid).

ATIF → OTel Mapping

ATIF Concept	OTel Span
Trajectory	Root AGENT span
Conversational turn	Nested AGENT span (multi-turn only)
Agent step (source: "agent")	LLM span
Tool call (tool_calls[])	TOOL span (sibling of LLM)
Subagent delegation	Nested AGENT span tree

Span Hierarchy

Single-turn:

AGENT (root)
├── LLM (agent step 1)
├── TOOL (Read)
├── TOOL (Edit)
├── LLM (agent step 2)
└── TOOL (Bash)

Multi-turn:

AGENT (root)
├── AGENT (turn 1)
│   ├── LLM
│   └── TOOL
└── AGENT (turn 2)
    ├── LLM
    └── TOOL

Span Attributes

Spans carry OpenInference semantic attributes for LLM observability:

Attribute	Set On	Source
openinference.span.kind	All	AGENT / LLM / TOOL
session.id	All	trajectory.session_id
llm.model_name	AGENT, LLM	agent.model_name or step.model_name
llm.token_count.prompt	AGENT, LLM	final_metrics or step.metrics
llm.token_count.completion	AGENT, LLM	final_metrics or step.metrics
llm.token_count.prompt_details.cache_read	LLM	step.metrics.cached_tokens
llm.cost.total	AGENT, LLM	final_metrics.total_cost_usd or step.metrics.cost_usd
tool.name	TOOL	tool_call.function_name
input.value	All	Message or arguments
output.value	All	Response or observation result

ATIF v1.7 Feature Support

Feature	Status
Core steps (user/agent/system)	Supported
Tool calls + observation matching	Supported
Multi-turn splitting	Supported
final_metrics / metrics token counts	Supported
reasoning_content	Supported
tool_definitions	Supported
llm_call_count: 0 (deterministic dispatch)	Supported
llm_call_count > 1 (aggregated)	Supported
is_copied_context filtering	Supported
subagent_trajectories embedding	Supported
Multimodal ContentPart (v1.6+)	Supported (text extracted, images as metadata)
context_management system steps	Supported

Harbor Job Plugin

The package includes a Harbor job plugin (OtelPlugin) that automatically exports OTel traces during harbor run. It supports two modes:

Streaming — upload each trial as it completes

harbor run --dataset terminal-bench@2.0 --agent claude-code \
  --plugin atif2otel \
  --plugin-kwarg endpoint=https://mlflow.example.com \
  --plugin-kwarg experiment_name=my-eval

Batch — write flat files after the job ends

harbor run --dataset terminal-bench@2.0 --agent claude-code \
  --plugin atif2otel \
  --plugin-kwarg output_dir=./otel-traces \
  --plugin-kwarg encoding=json

Both modes can be combined (stream to endpoint + write files). Mode is auto-detected from which outputs are configured, or set explicitly with --plugin-kwarg mode=stream|batch.

Plugin kwargs

Kwarg	Env var fallback	Description
endpoint	OTEL_EXPORTER_OTLP_ENDPOINT	OTLP endpoint URL
output_dir	HARBOR_OTEL_OUTPUT_DIR	Directory for flat file output
experiment_name	MLFLOW_EXPERIMENT_NAME	MLflow experiment name (defaults to job name)
token	MLFLOW_TRACKING_TOKEN	Auth token for the endpoint
workspace	—	MLflow workspace (default: "default")
encoding	—	"json" (JSONL) or "pb" (protobuf)
mode	—	"auto", "stream", or "batch"

Shared Export API

The export functions are available for programmatic use:

from harbor_atif2otel.export import export_trial, export_trials

# Single trial
rs = export_trial(Path("jobs/my-job/trial-001"))

# Batch with file output
result = export_trials(trial_dirs, output=Path("out.jsonl"), encoding="json")
print(f"{result.converted} converted, {result.errors} errors")

Writing a Custom Uploader

Implement harbor_atif2otel.uploaders.base.Uploader:

from harbor_atif2otel.uploaders.base import Uploader
from opentelemetry.proto.trace.v1.trace_pb2 import ResourceSpans

class MyUploader(Uploader):
    def upload(self, resource_spans: ResourceSpans) -> None:
        # Serialize and send to your backend
        ...

The upload_batch() method is provided by the base class and calls upload() in a loop with error counting.

License

Apache 2.0 — see LICENSE.