orchestrata_ultimate.core.tracing.tracer

Reference › orchestrata_ultimate › core › tracing › tracer

A small, config-gated runtime tracer for generated Command boilerplate. It is distinct from the process-wide sys.monitoring engine in monitoring.engine: Tracer is a light per-component object that the gen-skeleton generator (see plans/build-factory-architecture.md) injects by constructor into every generated tool skeleton, so generated execute() code can wrap its body in self._tracer.span(...) and never uses a hard-coded print. It also separates the two tracing axes: static requirement tracing with @traces, and this dynamic runtime tracing.

TraceConfig is a frozen dataclass with enabled (default off), log_level (default DEBUG) and logger_name (orchestrata_ultimate.tracing). TraceConfig.from_env reads ORCHESTRATA_TOOL_TRACE (true when one of 1, true, yes, on) and ORCHESTRATA_TOOL_TRACE_LEVEL (a logging level name or number, falling back to DEBUG if invalid). Tracer.default() returns a process-wide singleton bound to that configuration, and reset_default clears it for test isolation. Tracer.span(name) is a context manager that emits a timed span when tracing is enabled and does nothing otherwise. The scanner and several CXP and gateway modules use it.

At a glance

Class diagram

Dependency diagram

API

Classes

TraceConfig (dataclass)

Decorators: dataclass(slots=True, frozen=True) · Line 39

Env/config-driven settings for the injected :class:Tracer.

Tracer

Line 81

Config-gated span/log emitter injected into generated Tool skeletons.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/tracing/tracer.py

"""Injected, config-gated runtime tracer for generated Command boilerplate.

Distinct from :class:`orchestrata_ultimate.core.monitoring.engine.TracerEngine`
(a process-wide ``sys.monitoring`` instrumentation engine that auto-traces
*every* decorated function): ``Tracer`` here is a lightweight, per-component
object that ``gen-skeleton`` (see ``plans/build-factory-architecture.md``)
wires into every generated ``<<Tool>>`` skeleton via constructor DI, so
generated ``execute()`` boilerplate can emit ``self._tracer.span(...)``
around its body — never a hard-coded ``print``.

Two distinct axes (per the build-factory plan):
    - Requirement tracing  — ``@traces(spec, story, reqs)`` (static, for RTM/KG).
    - Runtime tracing/debug — this module (dynamic, env/config-driven).

``TraceConfig`` is env/config-driven (log level, spans on/off) and ``Tracer``
is a thin `Singleton <Tracer.default()>` accessor bound to it, so tests and
deploys can toggle tracing without touching generated code.
"""

from __future__ import annotations

import logging
import os
import time
from collections.abc import Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from typing import ClassVar

#: Environment variable gating whether spans are emitted at all.
_ENV_ENABLED = "ORCHESTRATA_TOOL_TRACE"
#: Environment variable overriding the logging level spans are emitted at.
_ENV_LEVEL = "ORCHESTRATA_TOOL_TRACE_LEVEL"

_TRUTHY = {"1", "true", "yes", "on"}


@dataclass(slots=True, frozen=True)
class TraceConfig:
    """Env/config-driven settings for the injected :class:`Tracer`.

    Attributes:
        enabled: Whether ``Tracer.span()`` emits anything at all. When
            ``False`` (the default), ``span()`` is a zero-cost no-op.
        log_level: The ``logging`` level spans are emitted at.
        logger_name: Name of the ``logging.Logger`` spans are written to.
    """

    enabled: bool = False
    log_level: int = logging.DEBUG
    logger_name: str = "orchestrata_ultimate.tracing"

    @classmethod
    def from_env(cls, environ: Mapping[str, str] | None = None) -> TraceConfig:
        """Build a :class:`TraceConfig` from environment variables.

        Args:
            environ: Optional environment mapping (defaults to ``os.environ``);
                lets tests pass an explicit mapping instead of mutating
                process-global env state.

        Returns:
            A ``TraceConfig`` reflecting ``ORCHESTRATA_TOOL_TRACE`` (bool-ish)
            and ``ORCHESTRATA_TOOL_TRACE_LEVEL`` (a ``logging`` level name or
            number), falling back to disabled/DEBUG when unset or invalid.
        """
        env = os.environ if environ is None else environ
        enabled = env.get(_ENV_ENABLED, "0").strip().lower() in _TRUTHY
        raw_level = env.get(_ENV_LEVEL, "").strip()
        level = logging.DEBUG
        if raw_level:
            if raw_level.isdigit():
                level = int(raw_level)
            else:
                resolved = logging.getLevelName(raw_level.upper())
                if isinstance(resolved, int):
                    level = resolved
        return cls(enabled=enabled, log_level=level)


class Tracer:
    """Config-gated span/log emitter injected into generated Tool skeletons.

    Every ``gen-skeleton``-generated ``<<Tool>>`` composes one of these
    (defaulting to :meth:`default`, a process-wide Singleton) and wraps its
    ``execute()`` body in ``with self._tracer.span("Class.execute"): ...``.
    When ``TraceConfig.enabled`` is ``False`` (the default), ``span()`` is a
    no-op context manager — zero overhead, never a hard-coded print.
    """

    _default: ClassVar[Tracer | None] = None

    def __init__(
        self,
        config: TraceConfig | None = None,
        environ: Mapping[str, str] | None = None,
    ) -> None:
        """Bind this tracer to *config* (or one read from the environment).

        Args:
            config: Explicit ``TraceConfig``; when omitted, reads
                ``TraceConfig.from_env(environ)``.
            environ: Optional environment mapping used only when *config* is
                omitted; lets callers avoid mutating process-global env state.
        """
        self._config = config or TraceConfig.from_env(environ)
        self._logger = logging.getLogger(self._config.logger_name)

    @classmethod
    def default(cls, environ: Mapping[str, str] | None = None) -> Tracer:
        """Process-wide Singleton ``Tracer`` bound to ``TraceConfig.from_env()`` .

        First-call-wins: ``environ`` is only consulted the *first* time
        ``default()`` is called for this process (or since the last
        ``reset_default()``) — subsequent calls return the already-cached
        instance and silently ignore whatever ``environ`` they're given.
        Tests that need to rebind the singleton to a different environment
        must call ``Tracer.reset_default()`` first.
        """
        if cls._default is None:
            cls._default = cls(environ=environ)
        return cls._default

    @classmethod
    def reset_default(cls) -> None:
        """Clear the cached Singleton (test isolation only)."""
        cls._default = None

    @property
    def enabled(self) -> bool:
        """Whether this tracer currently emits spans."""
        return self._config.enabled

    @contextmanager
    def span(self, name: str) -> Iterator[None]:
        """Config-gated span around a block of generated Command boilerplate.

        A zero-cost no-op when ``self.enabled`` is ``False``; otherwise logs
        entry/exit + elapsed milliseconds at ``self._config.log_level``.

        Args:
            name: Span label, conventionally ``"<Class>.<method>"``.
        """
        if not self._config.enabled:
            yield
            return
        start = time.monotonic()
        self._logger.log(self._config.log_level, "span start: %s", name)
        try:
            yield
        finally:
            elapsed_ms = (time.monotonic() - start) * 1000
            self._logger.log(
                self._config.log_level, "span end: %s (%.3fms)", name, elapsed_ms
            )