Generic Event Transport

agentic_ci.telemetry transports producer-owned event mappings through the existing OTLP trace path used by the MLflow exporter. It does not define lifecycle stages, outcomes, schemas, issue trackers, or workflow orchestration.

Event boundary

An event must provide a non-empty event_type field. Producers can pass a JSON-compatible mapping or an immutable object implementing to_dict(). Additional safe fields are preserved unchanged and serialized in the OTLP span event body.

The transport rejects content rather than silently redacting and changing a producer schema. Sensitive field names, email addresses, bearer credentials, private-key material, URL userinfo, and secret-bearing query parameters are not accepted in event or correlation values. Prompt, description, comment, diff, source-code, request/response body, username, and credential fields must be reduced to approved identifiers or aggregate metrics by the producer. Events are also bounded by nesting, collection, string, and serialized-size limits. Token-count fields such as input_tokens remain valid.

Correlation identifiers are supplied separately as an arbitrary mapping. The transport emits them as telemetry.correlation.<name> attributes. A valid 32-character hexadecimal trace_id joins the event to an existing trace; otherwise the transport creates a trace ID from the event identity. Existing root spans in a JSONL log are used as the event span's parent.

Emit or export

from agentic_ci.telemetry import emit_event

emit_event(
    {
        "event_type": "workflow.completed",
        "event_id": "run-42-complete",
        "result": "accepted",
    },
    log_root="_run",
    correlation={
        "work_item": "item-42",
        "pipeline": "pipeline-7",
        "job": "job-12",
        "trace_id": "0123456789abcdef0123456789abcdef",
    },
)

File emission always appends to claude-otel.jsonl beneath the runner-owned log_root. Symbolic-link roots and symbolic-link log files are rejected, and the file is opened without following links. Producers cannot select an arbitrary output path.

Use a trusted endpoint="http://collector:4318" to POST the same OTLP trace payload to an HTTP collector, or provide both destinations. Endpoint URLs and authorization headers are runner configuration, not producer event input. emit_event() and export_event() raise transport errors. The caller decides whether an export failure is fatal, so event transport cannot silently change a workflow result.

telemetry

Generic event transport for the existing OTLP trace pipeline.

This module owns transport and serialization only. Producers own event schemas, workflow timing, outcome classification, and correlation-field meaning.

TelemetryEvent

Bases: Protocol

Protocol accepted by :func:emit_event.

Producers can pass any immutable event object that returns a JSON-like mapping from to_dict(). Plain mappings are accepted as well.

to_dict()

Return the event's JSON-compatible fields.

Source code in src/agentic_ci/telemetry.py

def to_dict(self) -> Mapping[str, JSONValue]:
    """Return the event's JSON-compatible fields."""

TelemetryEventError

Bases: ValueError

Raised when a producer event cannot be transported safely.

TelemetryExportError

Bases: RuntimeError

Raised when an OTLP event export request fails.

validate_event(event)

Validate and copy a generic event before transport.

Transport requires a non-empty event_type field. Other fields remain producer-owned and are preserved unchanged after privacy and size checks.

Source code in src/agentic_ci/telemetry.py

def validate_event(event: Mapping[str, JSONValue] | TelemetryEvent) -> dict[str, JSONValue]:
    """Validate and copy a generic event before transport.

    Transport requires a non-empty ``event_type`` field. Other fields remain
    producer-owned and are preserved unchanged after privacy and size checks.
    """
    if isinstance(event, Mapping):
        payload = dict(event)
    else:
        to_dict = getattr(event, "to_dict", None)
        if not callable(to_dict):
            raise TelemetryEventError("event must be a mapping or provide to_dict()")
        raw_payload = to_dict()
        if not isinstance(raw_payload, Mapping):
            raise TelemetryEventError("to_dict() must return a mapping")
        payload = dict(raw_payload)
    _validate_json(payload)
    event_type = payload.get("event_type")
    if not isinstance(event_type, str) or not event_type.strip():
        raise TelemetryEventError("event_type must be a non-empty string")
    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if len(encoded) > MAX_EVENT_BYTES:
        raise TelemetryEventError("event exceeds the maximum serialized size")
    return payload

build_event_record(event, *, correlation=None, parent_span_id=None)

Build an OTLP trace record for one generic event.

The record uses a zero-duration span so existing JSONL and MLflow trace exporters consume events without a separate storage or UI integration. Correlation values are emitted as namespaced attributes and never interpreted by this module.

Source code in src/agentic_ci/telemetry.py

def build_event_record(
    event: Mapping[str, JSONValue] | TelemetryEvent,
    *,
    correlation: Correlation | None = None,
    parent_span_id: str | None = None,
) -> dict[str, Any]:
    """Build an OTLP trace record for one generic event.

    The record uses a zero-duration span so existing JSONL and MLflow trace
    exporters consume events without a separate storage or UI integration.
    Correlation values are emitted as namespaced attributes and never
    interpreted by this module.
    """
    payload = validate_event(event)
    correlation_values = _validate_correlation(correlation)
    trace_id = _trace_id(payload, correlation_values)
    span_id = uuid.uuid4().hex[:16]
    timestamp_ns = time.time_ns()
    attributes = [
        _attribute("event.name", str(payload["event_type"])),
    ]
    event_id = payload.get("event_id")
    if isinstance(event_id, str) and event_id:
        attributes.append(_attribute("event.id", event_id))
    for key, value in correlation_values.items():
        if value is not None:
            attributes.append(_attribute(f"telemetry.correlation.{key}", value))

    span: dict[str, Any] = {
        "traceId": trace_id,
        "spanId": span_id,
        "name": str(payload["event_type"]),
        "kind": 1,
        "startTimeUnixNano": str(timestamp_ns),
        "endTimeUnixNano": str(timestamp_ns),
        "status": {"code": 0},
        "attributes": attributes,
        "events": [
            {
                "name": str(payload["event_type"]),
                "timeUnixNano": str(timestamp_ns),
                "attributes": [_attribute("telemetry.event", json.dumps(payload, sort_keys=True))],
            }
        ],
    }
    if _valid_id(parent_span_id, 16):
        span["parentSpanId"] = parent_span_id

    return {
        "ts": datetime.now(timezone.utc).isoformat(),
        "path": "/v1/traces",
        "payload": {
            "resourceSpans": [
                {
                    "resource": {"attributes": [_attribute("service.name", "agentic-ci")]},
                    "scopeSpans": [
                        {
                            "scope": {"name": "agentic-ci.telemetry"},
                            "spans": [span],
                        }
                    ],
                }
            ]
        },
    }

append_event(log_root, event, *, correlation=None, parent_span_id=None)

Append an event to the fixed OTLP JSONL file beneath trusted log_root.

Source code in src/agentic_ci/telemetry.py

def append_event(
    log_root: str | Path,
    event: Mapping[str, JSONValue] | TelemetryEvent,
    *,
    correlation: Correlation | None = None,
    parent_span_id: str | None = None,
) -> dict[str, Any]:
    """Append an event to the fixed OTLP JSONL file beneath trusted ``log_root``."""
    root = _resolve_log_root(log_root)
    record = build_event_record(event, correlation=correlation, parent_span_id=parent_span_id)
    span = record["payload"]["resourceSpans"][0]["scopeSpans"][0]["spans"][0]
    if "parentSpanId" not in span:
        parent = _find_parent_span_id(root, span["traceId"])
        if parent:
            span["parentSpanId"] = parent
    with _open_log(root, os.O_WRONLY | os.O_CREAT | os.O_APPEND, "a") as stream:
        stream.write(json.dumps(record) + "\n")
    return record

export_event(endpoint, event, *, correlation=None, parent_span_id=None, headers=None, timeout=30)

Export one generic event to an OTLP HTTP trace endpoint.

Source code in src/agentic_ci/telemetry.py

def export_event(
    endpoint: str,
    event: Mapping[str, JSONValue] | TelemetryEvent,
    *,
    correlation: Correlation | None = None,
    parent_span_id: str | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: float = 30,
) -> dict[str, Any]:
    """Export one generic event to an OTLP HTTP trace endpoint."""
    record = build_event_record(event, correlation=correlation, parent_span_id=parent_span_id)
    _post_record(endpoint, record, headers=headers, timeout=timeout)
    return record

emit_event(event, *, log_root=None, endpoint=None, correlation=None, parent_span_id=None, headers=None, timeout=30)

Emit one event to JSONL, OTLP HTTP, or both.

At least one destination is required. This function does not catch write or export errors, allowing callers to preserve their workflow's result while deciding how to report transport failure.

Source code in src/agentic_ci/telemetry.py

def emit_event(
    event: Mapping[str, JSONValue] | TelemetryEvent,
    *,
    log_root: str | Path | None = None,
    endpoint: str | None = None,
    correlation: Correlation | None = None,
    parent_span_id: str | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: float = 30,
) -> dict[str, Any]:
    """Emit one event to JSONL, OTLP HTTP, or both.

    At least one destination is required. This function does not catch write
    or export errors, allowing callers to preserve their workflow's result
    while deciding how to report transport failure.
    """
    if log_root is None and endpoint is None:
        raise TelemetryEventError("log_root or endpoint is required")
    if endpoint is not None and not endpoint.strip():
        raise TelemetryEventError("endpoint must be a non-empty string")
    if log_root is not None:
        record = append_event(
            log_root,
            event,
            correlation=correlation,
            parent_span_id=parent_span_id,
        )
        if endpoint is not None:
            _post_record(
                endpoint,
                record,
                headers=headers,
                timeout=timeout,
            )
        return record
    return export_event(
        endpoint or "",
        event,
        correlation=correlation,
        parent_span_id=parent_span_id,
        headers=headers,
        timeout=timeout,
    )