orchestrata_ultimate.refactor.codemod.observability.debug_panel

Reference › orchestrata_ultimate › refactor › codemod › observability › debug_panel

DebugPanelObserver writes one trace file per codemod operation for post-mortem analysis. Each CodemodEvent becomes one JSON line (timestamp, event type, payload) appended to <output_dir>/<correlation_id>.ndjson. The directory defaults to pipeline_output/traces and is created on the first write, and the file handles are kept open per correlation id until close is called. The workspace is its user, as one of its default observers.

At a glance

Class diagram

Dependency diagram

API

Classes

DebugPanelObserver

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

Writes NDJSON trace files to {output_dir}/{correlation_id}.ndjson.

Source

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

"""DebugPanelObserver — writes NDJSON trace files for post-mortem analysis.

SPEC-110 §15.5: Each codemod run produces one NDJSON file at
``{output_dir}/{correlation_id}.ndjson``, with one JSON object per line
carrying timestamp, event_type, and payload.
"""

from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import TextIO

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.observability.events import CodemodEvent

logger = logging.getLogger(__name__)


@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class DebugPanelObserver:
    """Writes NDJSON trace files to ``{output_dir}/{correlation_id}.ndjson``.

    Each :class:`CodemodEvent` is serialized as one JSON line. The output
    directory is created lazily on first write. Files are opened in append
    mode so events accumulate across the lifetime of a single correlation
    run.
    """

    def __init__(self, output_dir: Path | None = None) -> None:
        self._output_dir = output_dir or Path("pipeline_output/traces")
        # TextIO, not object: these are open() handles and the class calls
        # .write/.flush/.close on them. `object` type-checked the dict and
        # then lost all three methods.
        self._handles: dict[str, TextIO] = {}

    def on_event(self, event: CodemodEvent) -> None:
        """Append *event* as a JSON line to the correlation's trace file."""
        corr = event.correlation_id
        record = {
            "timestamp": event.timestamp,
            "event_type": event.type.value,
            "payload": event.data,
        }
        line = json.dumps(record, separators=(",", ":"))

        handle = self._handles.get(corr)
        if handle is None:
            self._output_dir.mkdir(parents=True, exist_ok=True)
            path = self._output_dir / f"{corr}.ndjson"
            handle = open(path, "a", encoding="utf-8")  # noqa: SIM115
            self._handles[corr] = handle

        handle.write(line + "\n")
        handle.flush()

    def close(self) -> None:
        """Close all open file handles."""
        for handle in self._handles.values():
            try:
                handle.close()
            except OSError:
                logger.exception("Failed to close trace file handle")
        self._handles.clear()