Observability

Tracing lifecycle, span helpers, exporters, and redaction. See the observability reference for configuration and the span/attribute schema.

apab.observability.tracing

Tracer lifecycle and span helpers with a soft OpenTelemetry dependency.

is_enabled

is_enabled()

Whether tracing is currently active.

Source code in src/apab/observability/tracing.py

def is_enabled() -> bool:
    """Whether tracing is currently active."""
    return _enabled

init_observability

init_observability(spec, run_ctx=None, extra_processors=None)

Set up the APAB tracer provider from spec.

Returns True when tracing is active. Safe to call when the opentelemetry packages are absent: logs a warning and stays disabled. The APAB_OBSERVABILITY=1 env var forces enabled.

extra_processors is a hook for tests to inject an in-memory span processor.

Source code in src/apab/observability/tracing.py

def init_observability(
    spec: ObservabilitySpec,
    run_ctx: RunContext | None = None,
    extra_processors: list[Any] | None = None,
) -> bool:
    """Set up the APAB tracer provider from *spec*.

    Returns True when tracing is active. Safe to call when the
    ``opentelemetry`` packages are absent: logs a warning and stays
    disabled. The ``APAB_OBSERVABILITY=1`` env var forces ``enabled``.

    ``extra_processors`` is a hook for tests to inject an in-memory
    span processor.
    """
    global _provider, _tracer, _enabled

    enabled = spec.enabled or os.environ.get("APAB_OBSERVABILITY") == "1"
    if not enabled:
        return False
    if _enabled:
        # Already initialised (e.g. a second session in one process).
        return True

    try:
        from opentelemetry import trace
        from opentelemetry.sdk.resources import Resource
        from opentelemetry.sdk.trace import TracerProvider
    except ImportError:
        logger.warning(
            "observability is enabled but opentelemetry is not installed; "
            "run: pip install 'apab[observability]'"
        )
        return False

    from apab.observability.export import build_processors

    provider = TracerProvider(
        resource=Resource.create({"service.name": spec.service_name}),
    )
    for processor in build_processors(spec, run_ctx):
        provider.add_span_processor(processor)
    for processor in extra_processors or []:
        provider.add_span_processor(processor)

    if spec.set_global:
        trace.set_tracer_provider(provider)

    _provider = provider
    _tracer = provider.get_tracer("apab")
    _enabled = True
    logger.info("Observability enabled (service=%s)", spec.service_name)
    return True

shutdown_observability

shutdown_observability()

Flush exporters and disable tracing.

Source code in src/apab/observability/tracing.py

def shutdown_observability() -> None:
    """Flush exporters and disable tracing."""
    global _provider, _tracer, _enabled, _remote_ctx
    if _provider is not None:
        try:
            _provider.shutdown()
        except Exception:
            logger.exception("Tracer provider shutdown failed")
    _provider = None
    _tracer = None
    _enabled = False
    _remote_ctx = None

init_remote_parent_from_env

init_remote_parent_from_env()

Adopt a W3C TRACEPARENT from the environment, if present.

A spawned MCP server process has no in-process parent span; a caller (e.g. a Strands client) can hand one across the process boundary via the standard traceparent header value in the TRACEPARENT env var. Root spans opened after this call parent onto it, so both sides of the stdio transport share one trace. Returns True when a remote parent was adopted.

Source code in src/apab/observability/tracing.py

def init_remote_parent_from_env() -> bool:
    """Adopt a W3C ``TRACEPARENT`` from the environment, if present.

    A spawned MCP server process has no in-process parent span; a caller
    (e.g. a Strands client) can hand one across the process boundary via
    the standard ``traceparent`` header value in the ``TRACEPARENT`` env
    var. Root spans opened after this call parent onto it, so both sides
    of the stdio transport share one trace. Returns True when a remote
    parent was adopted.
    """
    global _remote_ctx
    value = os.environ.get("TRACEPARENT")
    if not value:
        return False
    try:
        from opentelemetry.trace.propagation.tracecontext import (
            TraceContextTextMapPropagator,
        )
    except ImportError:
        return False
    _remote_ctx = TraceContextTextMapPropagator().extract({"traceparent": value})
    return True

span

span(name, **attributes)

Open a child span, or yield a no-op span when tracing is off.

Attribute values of None are skipped.

Source code in src/apab/observability/tracing.py

@contextmanager
def span(name: str, **attributes: Any) -> Iterator[Any]:
    """Open a child span, or yield a no-op span when tracing is off.

    Attribute values of ``None`` are skipped.
    """
    if not _enabled or _tracer is None:
        yield _NOOP_SPAN
        return

    # Root spans adopt the remote parent handed over via TRACEPARENT;
    # nested spans keep parenting on the active local span.
    context = None
    if _remote_ctx is not None:
        from opentelemetry import trace

        if not trace.get_current_span().get_span_context().is_valid:
            context = _remote_ctx

    with _tracer.start_as_current_span(name, context=context) as s:
        for key, value in attributes.items():
            if value is not None:
                s.set_attribute(key, value)
        yield s

current_trace_ids

current_trace_ids()

Return (trace_id, span_id) of the current span as hex strings.

Source code in src/apab/observability/tracing.py

def current_trace_ids() -> tuple[str, str] | None:
    """Return (trace_id, span_id) of the current span as hex strings."""
    if not _enabled:
        return None
    from opentelemetry import trace

    ctx = trace.get_current_span().get_span_context()
    if not ctx.is_valid:
        return None
    return format(ctx.trace_id, "032x"), format(ctx.span_id, "016x")

set_span_error

set_span_error(s, exception)

Record exception on span s and mark its status as error.

Source code in src/apab/observability/tracing.py

def set_span_error(s: Any, exception: BaseException) -> None:
    """Record *exception* on span *s* and mark its status as error."""
    try:
        s.record_exception(exception)
        from opentelemetry.trace import Status, StatusCode

        s.set_status(Status(StatusCode.ERROR, str(exception)))
    except ImportError:
        pass

apab.observability.export

Span exporters and processor construction.

Only imported once OpenTelemetry is known to be installed (from :func:apab.observability.tracing.init_observability).

JsonlSpanExporter

Bases: SpanExporter

Write one JSON object per span to a .jsonl file.

Kept dependency-light so trace.jsonl in the run bundle can be read without any OpenTelemetry tooling.

Source code in src/apab/observability/export.py

class JsonlSpanExporter(SpanExporter):
    """Write one JSON object per span to a .jsonl file.

    Kept dependency-light so trace.jsonl in the run bundle can be read
    without any OpenTelemetry tooling.
    """

    def __init__(self, path: Path) -> None:
        self._path = Path(path)

    def export(self, spans: Any) -> SpanExportResult:
        try:
            with self._path.open("a") as fh:
                for s in spans:
                    fh.write(json.dumps(_span_to_dict(s), default=str) + "\n")
        except OSError:
            logger.exception("Failed to write %s", self._path)
            return SpanExportResult.FAILURE
        return SpanExportResult.SUCCESS

    def shutdown(self) -> None:
        pass

build_processors

build_processors(spec, run_ctx)

Build span processors for the configured exporters.

Spans are low-volume here (one per turn/tool call), so simple processors are used throughout: they export synchronously, which keeps trace.jsonl complete even on a crash.

Source code in src/apab/observability/export.py

def build_processors(
    spec: ObservabilitySpec,
    run_ctx: RunContext | None,
) -> list[SimpleSpanProcessor]:
    """Build span processors for the configured exporters.

    Spans are low-volume here (one per turn/tool call), so simple
    processors are used throughout: they export synchronously, which
    keeps trace.jsonl complete even on a crash.
    """
    processors: list[SimpleSpanProcessor] = []

    if spec.trace_jsonl and run_ctx is not None:
        processors.append(SimpleSpanProcessor(
            JsonlSpanExporter(run_ctx.run_dir / "trace.jsonl"),
        ))
    elif spec.trace_jsonl and os.environ.get("APAB_TRACE_JSONL"):
        # A served MCP process has no RunContext; APAB_TRACE_JSONL names a
        # file so its spans still land somewhere a harness can collect.
        processors.append(SimpleSpanProcessor(
            JsonlSpanExporter(Path(os.environ["APAB_TRACE_JSONL"])),
        ))

    if spec.console_exporter:
        # stderr, never stdout: the MCP stdio transport owns stdout, and a
        # span dump there corrupts the JSON-RPC stream.
        processors.append(SimpleSpanProcessor(ConsoleSpanExporter(out=sys.stderr)))

    endpoint = spec.otlp_endpoint or os.environ.get(
        "OTEL_EXPORTER_OTLP_ENDPOINT"
    )
    if endpoint:
        try:
            from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
                OTLPSpanExporter,
            )

            processors.append(SimpleSpanProcessor(
                OTLPSpanExporter(endpoint=f"{endpoint.rstrip('/')}/v1/traces"),
            ))
        except ImportError:
            logger.warning(
                "otlp_endpoint is set but the OTLP exporter is not "
                "installed; run: pip install 'apab[observability]'"
            )

    return processors

apab.observability.redaction

Redaction of captured tool arguments and results for span attributes.

Mirrors the RedactionMode semantics used by the tool-dispatch audit log: none captures values, metadata_only captures shape only, strict captures nothing beyond a content hash.

args_hash

args_hash(arguments)

Deterministic 16-char hash of a tool-call argument dict.

Source code in src/apab/observability/redaction.py

def args_hash(arguments: dict[str, Any]) -> str:
    """Deterministic 16-char hash of a tool-call argument dict."""
    raw = json.dumps(arguments, sort_keys=True, default=str)
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

capture_args

capture_args(arguments, mode)

Return span attributes describing arguments under mode.

Always includes args_hash so identical calls can be correlated across runs without exposing values.

Source code in src/apab/observability/redaction.py

def capture_args(
    arguments: dict[str, Any],
    mode: RedactionMode,
) -> dict[str, Any]:
    """Return span attributes describing *arguments* under *mode*.

    Always includes ``args_hash`` so identical calls can be correlated
    across runs without exposing values.
    """
    attrs: dict[str, Any] = {"args_hash": args_hash(arguments)}
    if mode == RedactionMode.none:
        attrs["args_json"] = json.dumps(arguments, default=str)[:2000]
    elif mode == RedactionMode.metadata_only:
        attrs["arg_keys"] = sorted(arguments.keys())
    # strict: hash only
    return attrs

capture_text

capture_text(text, mode, max_len=200)

Return text truncated per mode, or None if it must be dropped.

Source code in src/apab/observability/redaction.py

def capture_text(
    text: str,
    mode: RedactionMode,
    max_len: int = 200,
) -> str | None:
    """Return *text* truncated per *mode*, or None if it must be dropped."""
    if mode == RedactionMode.strict:
        return None
    if mode == RedactionMode.metadata_only:
        return f"<{len(text)} chars>"
    if len(text) <= max_len:
        return text
    return text[:max_len] + "..."