orchestrata_ultimate.refactor.codemod.observability.tracing

Reference › orchestrata_ultimate › refactor › codemod › observability › tracing

TracingObserver writes a structured log entry for each codemod event, with the correlation id, a span name, the event type and a duration, through core.tracing.Tracer (the default tracer when none is passed) and the logger orchestrata_ultimate.codemod.tracing. Start events record a monotonic time and the matching end events (for example pass started and pass completed) compute the elapsed milliseconds from it. The workspace uses it among its default observers.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

TracingObserver

Decorators: traces(spec='110', story='US-110-005', reqs=['REQ-110-026']) · Line 49

Writes structured JSON spans via core.tracing.Tracer.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/observability/tracing.py

"""TracingObserver — emits structured JSON spans via core.tracing.Tracer.

SPEC-110 §15.3: Each CodemodEvent produces a structured log entry with
correlation_id, span_name, duration_ms, and event_type, formatted for
consumption by ``core.logging.StructuredFormatter``.
"""

from __future__ import annotations

import logging
import time

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.refactor.codemod.observability.events import (
    CodemodEvent,
    CodemodEventType,
)

_SPAN_START_TYPES: frozenset[CodemodEventType] = frozenset(
    {
        CodemodEventType.PASS_STARTED,
        CodemodEventType.REVIEW_STARTED,
        CodemodEventType.COMMIT_STARTED,
    }
)

_SPAN_END_TYPES: frozenset[CodemodEventType] = frozenset(
    {
        CodemodEventType.PASS_COMPLETED,
        CodemodEventType.REVIEW_COMPLETED,
        CodemodEventType.COMMIT_COMPLETED,
    }
)

_SPAN_NAMES: dict[CodemodEventType, str] = {
    CodemodEventType.WORKSPACE_LOADED: "CodemodWorkspace.load",
    CodemodEventType.PASS_STARTED: "TransformPass.apply",
    CodemodEventType.PASS_COMPLETED: "TransformPass.apply",
    CodemodEventType.REVIEW_STARTED: "ReviewPass.review",
    CodemodEventType.REVIEW_COMPLETED: "ReviewPass.review",
    CodemodEventType.COMMIT_STARTED: "CodemodWorkspace.commit",
    CodemodEventType.COMMIT_COMPLETED: "CodemodWorkspace.commit",
    CodemodEventType.ERROR: "CodemodEngine.error",
}


@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class TracingObserver:
    """Writes structured JSON spans via ``core.tracing.Tracer``.

    Each :class:`CodemodEvent` produces a structured log entry carrying
    ``correlation_id``, ``span_name``, ``duration_ms``, and ``event_type``.
    Start/end event pairs (PASS_STARTED → PASS_COMPLETED, etc.) are timed
    via ``time.monotonic`` so the emitted ``duration_ms`` reflects wall
    clock elapsed time for that phase.
    """

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer or Tracer.default()
        self._logger = logging.getLogger("orchestrata_ultimate.codemod.tracing")
        self._span_starts: dict[str, float] = {}

    def on_event(self, event: CodemodEvent) -> None:
        """Emit a structured span for *event*."""
        span_name = _SPAN_NAMES.get(event.type, f"codemod.{event.type.value}")
        corr = event.correlation_id

        if event.type in _SPAN_START_TYPES:
            self._span_starts[corr] = time.monotonic()

        duration_ms: float = 0.0
        if event.type in _SPAN_END_TYPES:
            start = self._span_starts.pop(corr, None)
            if start is not None:
                duration_ms = (time.monotonic() - start) * 1000
            else:
                dur_s = event.data.get("duration_seconds", 0)
                duration_ms = dur_s * 1000

        self._logger.info(
            event.type.value,
            extra={
                "correlation_id": corr,
                "event_type": event.type.value,
                "span_name": span_name,
                "duration_ms": round(duration_ms, 3),
                "payload": event.data,
            },
        )