Tracing

Dreadnode tracing is OpenTelemetry-based and implemented via logfire. The SDK exposes TaskSpan context managers for structured traces plus lightweight Span objects when you only need timing and tags.

Quick start

import dreadnode as dn

dn.configure(
    server="https://api.dreadnode.io",
    api_key="dn_...",
    organization="acme",
    project="support-bots",
    trace_backend="remote",
)

with dn.task_span("data-prep", type="task", tags=["etl"]) as span:
    span.log_input("rows", 250)
    span.log_metric("latency_ms", 112, step=1)
    span.log_output("status", "ok")

TaskSpan vs Span

TaskSpan captures inputs, outputs, metrics, params, artifacts, and child tasks. Use Span for lightweight timing and tagging when you do not need structured inputs/outputs.

import dreadnode as dn
from dreadnode.tracing.span import Span, TaskSpan

with dn.span("cache-warmup") as span:
    assert isinstance(span, Span)

with dn.task_span("model-inference", type="task") as task_span:
    assert isinstance(task_span, TaskSpan)

SpanType values

SpanType is a literal type that describes what kind of work a span represents: "task", "evaluation", "agent", "study", "tool", "trial", "sample", "generation", "scorer", and "span".

from dreadnode.tracing.constants import SpanType
import dreadnode as dn

span_type: SpanType = "evaluation"

with dn.task_span("eval:baseline", type=span_type):
    pass

Runs and task decorators

dn.run()

Create a top-level task span with trace infrastructure setup. This is the recommended entry point for standalone scripts.

import dreadnode as dn

with dn.run("experiment-1", project="my-project", tags=["baseline"]) as run:
    run.log_param("model", "gpt-4o")
    run.log_input("prompt", "Hello")
    # ... do work ...
    run.log_output("result", "World")
    run.log_metric("score", 0.95)

dn.task_and_run()

Context manager that creates a task span and auto-initializes trace infrastructure if no parent context exists. Useful when you want both a run and a task in one call.

import dreadnode as dn

with dn.task_and_run("eval-run", task_name="scoring", task_type="evaluation") as task:
    dn.log_input("dataset", "test-v2")
    dn.log_metric("accuracy", 0.92)

@dn.task decorator

Wraps a function so that every call is traced as a task span. Optionally attach scorers that run automatically.

import dreadnode as dn


@dn.task
async def classify(text: str) -> str:
    # function body is automatically traced
    return "positive"


# With options
@dn.task(name="classifier", scorers=[my_scorer], log_inputs=True, log_output=True)
async def classify_v2(text: str) -> str:
    return "positive"

Logging

Parameters

import dreadnode as dn

dn.log_param("model", "gpt-4o")
dn.log_params(model="gpt-4o", temperature=0.7, max_tokens=1000)

Inputs and outputs

import dreadnode as dn

# Single
dn.log_input("prompt", "Write a haiku about security")
dn.log_output("completion", "Firewalls stand guard...")

# Batch
dn.log_inputs(prompt="Write a haiku", context="security domain")
dn.log_outputs(completion="Firewalls stand guard...", tokens_used=42)

Metrics

import dreadnode as dn

# Single metric
dn.log_metric("score", 0.91, step=0)

# With aggregation mode
dn.log_metric("latency_ms", 112, aggregation="avg")

# Batch metrics
dn.log_metrics({"accuracy": 0.95, "f1": 0.88}, step=1)

Aggregation modes: "direct" (default), "min", "max", "avg", "sum", "count".

Samples

Log input/output pairs as ephemeral task spans. Each sample automatically links its output to its input.

import dreadnode as dn

# Single sample
dn.log_sample(
    "sample-1",
    input="input text",
    output="output text",
    metrics={"accuracy": 0.95},
)

# Batch samples
dn.log_samples(
    "eval-set",
    [
        ("input 1", "output 1"),
        ("input 2", "output 2", {"score": 0.8}),  # with metrics
    ],
)

Object linking

Associate two runtime objects together (e.g. link a model output to its input).

import dreadnode as dn

dn.link_objects(output_obj, input_obj, attributes={"relation": "generated_from"})

Flushing

Force-flush pending span data to the backend immediately.

import dreadnode as dn

dn.push_update()

Logging artifacts

Artifacts require storage to be configured (via configure). The SDK uploads files to workspace CAS and stores metadata on the span. Directories are recursively uploaded.

import dreadnode as dn

dn.configure(
    server="https://api.dreadnode.io",
    api_key="dn_...",
    organization="acme",
    project="support-bots",
)

with dn.task_span("training"):
    dn.log_artifact("./checkpoints/model.bin", name="model.bin")
    dn.log_artifact("./results/")  # uploads all files in the directory

Accessing current spans

Retrieve the active span from anywhere in your code.

import dreadnode as dn

run = dn.get_current_run()  # current top-level run span, or None
task = dn.get_current_task()  # current task span, or None

TaskSpan properties

Key properties available on TaskSpan instances:

Property	Type	Description
`task_id`	`str`	Unique task identifier.
`root_id`	`str`	Root span ID for the trace.
`inputs`	`dict`	Logged inputs.
`outputs`	`dict`	Logged outputs.
`metrics`	`dict`	Logged metrics.
`params`	`dict`	Logged parameters.
`duration`	`float \| None`	Span duration in seconds.
`active`	`bool`	Whether the span is currently active.
`failed`	`bool`	Whether the span recorded a failure.
`tasks`	`list`	Direct child task spans.
`all_tasks`	`list`	All descendant task spans.

Distributed tracing

Serialize a task context for cross-process or cross-host continuation using W3C TraceContext propagation.

import dreadnode as dn

# On the originating process
instance = dn.get_default_instance()
context = instance.get_task_context()
# context is a TypedDict: {task_id, task_name, project, trace_context}

# Send context to the remote process (e.g. via message queue)...

# On the remote process
instance = dn.get_default_instance()
with instance.continue_task(context) as task:
    task.log_metric("remote_score", 0.88)

Binding session IDs

Use bind_session_id to route spans to a session-scoped JSONL file.

import dreadnode as dn
from dreadnode.tracing.span import bind_session_id

with bind_session_id("session_123"):
    with dn.task_span("chat:turn-1"):
        dn.log_input("prompt", "Hi")
        dn.log_output("response", "Hello!")

TraceBackend configuration

TraceBackend controls whether the SDK streams spans remotely in addition to local JSONL.

Setting	Local JSONL	Remote OTLP
`"local"`	Always written	Disabled
`"remote"`	Always written	Enabled
`None` (default)	Always written	Auto-enabled if `server` + `api_key` + `organization` are all provided

from dreadnode.tracing.exporters import TraceBackend
import dreadnode as dn

backend: TraceBackend = "remote"

dn.configure(
    server="https://api.dreadnode.io",
    api_key="dn_...",
    organization="acme",
    project="support-bots",
    trace_backend=backend,
)

Remote spans are sent via OTLP to {server}/api/v1/org/{org}/otel/traces with gzip compression and API key authentication.

Local trace files

Local JSONL files are always written regardless of trace_backend setting.

Session-bound spans: ~/.dreadnode/sessions/<session_id>/chat_<session_id>.jsonl
Run-scoped spans: ~/.dreadnode/projects/<project>/<run_id>/spans.jsonl