Metadata-Version: 2.4
Name: harbor-atif2otel
Version: 0.1.0
Summary: Convert ATIF agent trajectories to OpenTelemetry protobuf spans with pluggable uploaders.
Project-URL: Repository, https://github.com/harbor-framework/harbor
Project-URL: Issues, https://github.com/harbor-framework/harbor/issues
Author-email: Jeremy Eder <jeder@redhat.com>
License-Expression: Apache-2.0
Requires-Python: >=3.12
Requires-Dist: opentelemetry-proto>=1.42.1
Provides-Extra: plugin
Requires-Dist: harbor; extra == 'plugin'
Provides-Extra: test
Requires-Dist: pytest>=8.4.1; extra == 'test'
Description-Content-Type: text/markdown

# harbor-atif2otel

Convert [ATIF](../../rfcs/0001-trajectory-format.md) agent trajectories to [OpenTelemetry](https://opentelemetry.io/) spans for visualization in any OTel-compatible backend.

## Install

```bash
pip install harbor-atif2otel
```

## Quick Start

```python
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

```bash
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

```bash
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:

```python
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`:

```python
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](../../LICENSE).
