Skip to content

AI Diagnostics

obslog's defining goal is that its output lets an external AI agent reconstruct execution and localize a root cause deterministically. The SDK never calls an LLM (PRIN-008); it produces the evidence an agent consumes.

The contract: the Record schema

Every record conforms to a published, versioned JSON Schema (schemas/obslog.record.schema.json). Consumers — AI agents and tests — rely on it: stable field names, stable types, and a schema_version that only evolves additively within a major version. See the Record schema reference.

Correlation → causal trees

Records carry execution_id, parent_execution, trace_id, and span_id. These form a causal execution tree that spans sync and async boundaries. operation() scopes link a child execution to its parent automatically.

Stable error identifiers

error_code is a stable, namespaced string suitable as a clustering/join key — the handle an agent groups failures by. The SDK defines no codes itself; register your own:

from obslog.diagnostics import CodeRegistry, CheckCodes

codes = CodeRegistry()
codes.register("DB_TIMEOUT", category="timeout", retryable=True)

# In a pipeline, validate and annotate records against the registry:
from obslog.sinks import Pipeline, Stream

obslog.configure(sinks=[Pipeline([CheckCodes(codes, annotate=True)], Stream())])

Diagnostic hints

A hint is application-supplied structured guidance, carried verbatim:

from obslog.diagnostics import hint

log.error(
    "gateway.failed",
    error=exc,
    error_code="DB_TIMEOUT",
    hint=hint(category="timeout", retryable=True),
)

Diagnostic bundles

A bundle serializes a correlated execution subtree into one deterministic document — a flat, ordered records list plus a causal executions tree — for handoff to an agent (schema: schemas/obslog.bundle.schema.json).

from obslog.diagnostics import BundleSink
from obslog.processors import Redact
from obslog.sinks import Pipeline

bundle = BundleSink()
# Redact before the bundle so secrets never reach the exported evidence:
obslog.configure(sinks=[Pipeline([Redact(["password"])], bundle)])
...
agent_input = bundle.to_json(indent=2)

Determinism & testing

Serialization is deterministic (sorted keys) given identical inputs. For golden tests, obslog.testing.snapshot() normalizes volatile fields (timestamps, ids, durations, host/pid, stacks) while preserving correlation structure — the same id maps to the same <id:N> token everywhere it appears.

Why it works with OpenTelemetry

Severity numbers are OpenTelemetry-aligned from the start, so obslog.integrations.opentelemetry.record_to_otel maps a record to an OTel log record losslessly.