orchestrata_ultimate.refactor.codemod.report

Reference › orchestrata_ultimate › refactor › codemod › report

The report data structures shared by the passes, the review engine, the transaction layer and the daemon. A ChangeEntry records one file's path, original and mutated LosslessModule; a ChangeSet is a tuple of entries with file_count, paths, iteration and a merge that combines two sets and keeps the earliest original tree per path, so a multi-pass run that edits one file still commits. A Finding is one review result, and a CommitReport carries the outcome of a commit: passes run, files written, findings, legacy findings and duration. CommitReport converts to and from JSON (to_json, from_json, from_dict), which is how the CLI and the daemon client exchange it. It is imported by passes, txn, workspace, server, the review modules and the persistence repository.

At a glance

Class diagram

Dependency diagram

API

Classes

ChangeEntry (dataclass)

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

One mutated lossless module with metadata.

ChangeSet (dataclass)

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

Immutable collection of ChangeEntry objects.

Finding (dataclass)

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

One review finding (error or warning).

CommitReport (dataclass)

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

Machine-readable commit report (JSON-serializable).

Source

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

"""SPEC-110 codemod engine — report data structures.

Provides ChangeEntry, ChangeSet, Finding, CommitReport, WorkspaceStats,
and CodemodEvent for the codemod engine's data flow.
"""

from __future__ import annotations

import json
from collections.abc import Iterator
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any, Literal

from orchestrata_ultimate.refactor.codemod._types import WorkspaceStats
from orchestrata_ultimate.refactor.codemod.observability.events import (
    CodemodEvent,
    CodemodEventType,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.refactor.ctok import LosslessModule

__all__ = [
    "ChangeEntry",
    "ChangeSet",
    "Finding",
    "CommitReport",
    "WorkspaceStats",
    "CodemodEvent",
    "CodemodEventType",
]


@dataclass(frozen=True, slots=True)
class ChangeEntry:
    """One mutated lossless module with metadata."""

    path: str
    original_tree: LosslessModule
    mutated_tree: LosslessModule
    pass_name: str
    description: str = ""

    @property
    def original_bytes(self) -> bytes:
        """Pre-mutation on-disk bytes (for rollback).

        ``LosslessModule.code`` is already ``bytes`` (byte-identical to the
        on-disk source), so this is a direct hand-back -- no re-encode.
        """
        return self.original_tree.code


@dataclass(frozen=True, slots=True)
class ChangeSet:
    """Immutable collection of ChangeEntry objects."""

    entries: tuple[ChangeEntry, ...] = ()

    @property
    def file_count(self) -> int:
        """Number of changed files."""
        return len(self.entries)

    @property
    def paths(self) -> frozenset[str]:
        """Set of changed file paths."""
        return frozenset(e.path for e in self.entries)

    def merge(self, other: ChangeSet) -> ChangeSet:
        """Combine two ChangeSets for a multi-pass run.

        For a path touched by both, the LATER entry's ``mutated_tree`` and
        metadata win, but the EARLIEST entry's ``original_tree`` is kept: it
        is the load-time tree whose bytes are on disk, which is what the
        transactional write-back compares against (REQ-110-013/014).
        """
        merged: dict[str, ChangeEntry] = {e.path: e for e in self.entries}
        for entry in other.entries:
            earlier = merged.get(entry.path)
            if earlier is not None:
                entry = replace(entry, original_tree=earlier.original_tree)
            merged[entry.path] = entry
        return ChangeSet(entries=tuple(merged.values()))

    def __len__(self) -> int:
        return len(self.entries)

    def __iter__(self) -> Iterator[ChangeEntry]:
        return iter(self.entries)


@dataclass(frozen=True, slots=True)
class Finding:
    """One review finding (error or warning)."""

    severity: Literal["error", "warning"]
    rule: str
    path: str
    line: int | None
    message: str
    proof_tree: dict[str, Any] | None = None


@dataclass(frozen=True, slots=True)
class CommitReport:
    """Machine-readable commit report (JSON-serializable).

    ``findings`` is the veto set; ``legacy_findings`` holds the findings the
    original trees already carried (default engine, ADR-0073), reported but
    never vetoing.
    """

    success: bool
    files_written: tuple[str, ...]
    findings: tuple[Finding, ...]
    duration_seconds: float
    dry_run: bool
    passes_run: tuple[str, ...] = ()
    diffstat: str = ""
    correlation_id: str = ""
    legacy_findings: tuple[Finding, ...] = ()

    def to_json(self) -> str:
        """Serialize to JSON string."""

        def _finding(f: Finding) -> dict[str, Any]:
            return {
                "severity": f.severity,
                "rule": f.rule,
                "path": f.path,
                "line": f.line,
                "message": f.message,
            }

        data = {
            "success": self.success,
            "files_written": list(self.files_written),
            "findings": [_finding(f) for f in self.findings],
            "legacy_findings": [_finding(f) for f in self.legacy_findings],
            "duration_seconds": self.duration_seconds,
            "dry_run": self.dry_run,
            "passes_run": list(self.passes_run),
            "diffstat": self.diffstat,
            "correlation_id": self.correlation_id,
        }
        return json.dumps(data, indent=2)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> CommitReport:
        """Rebuild a report from the dict ``to_json`` serializes.

        The inverse of ``to_json`` for every field it writes; a finding's
        ``proof_tree`` is not serialized, so it comes back ``None``.
        """

        def _finding(item: dict[str, Any]) -> Finding:
            return Finding(
                severity=item["severity"],
                rule=item["rule"],
                path=item["path"],
                line=item["line"],
                message=item["message"],
            )

        return cls(
            success=data["success"],
            files_written=tuple(data["files_written"]),
            findings=tuple(_finding(f) for f in data["findings"]),
            duration_seconds=data["duration_seconds"],
            dry_run=data["dry_run"],
            passes_run=tuple(data.get("passes_run", ())),
            diffstat=data.get("diffstat", ""),
            correlation_id=data.get("correlation_id", ""),
            legacy_findings=tuple(_finding(f) for f in data.get("legacy_findings", ())),
        )

    @classmethod
    def from_json(cls, text: str) -> CommitReport:
        """Rebuild a report from the string ``to_json`` returns."""
        return cls.from_dict(json.loads(text))