orchestrata_ultimate.refactor.codemod.workspace

Reference › orchestrata_ultimate › refactor › codemod › workspace

CodemodWorkspace is the main entry point of the codemod engine (SPEC-110). It loads a repository's Python files into memory, builds the code intelligence graph over them, runs transform passes over the in-memory trees, and commits the result to disk after review. The package facade, the daemon server, _daemon_entry and scripts.codemod use it. It imports the lossless layer from ctok, the graph, the transaction layer, the review engine and the observability modules. See the refactor package narrative for the packet history.

• load(root, include, parallel, cache_dir, observers) folds each include root onto the repository (a root outside the repo is refused), collects the *.py files in sorted order so node ids are deterministic, parses them to LosslessModule objects (optionally on a thread pool, and through a content-addressed ParseCache when cache_dir is set), builds the graph, and emits INDEX_BUILT and WORKSPACE_LOADED. Files that fail to parse are listed in unparsable.
• run(passes, observers) applies passes in order, so pass N sees the mutations of pass N-1, emitting PASS_STARTED, FILE_MUTATED and PASS_COMPLETED and merging the results into one ChangeSet. After each pass the import index is updated for the touched modules.
• commit(changes, reviews, dry_run, observers, review_engine, run_review_engine) runs the review engine on the mutated forest first. By default only findings the change introduced abort the commit (ADR-0073); legacy findings are reported in the CommitReport. A dry run produces the report without writing. A real commit writes through txn.commit_changes, and a failure emits rollback and abort events and re-raises.
• restore(originals) puts earlier trees back and updates the index, which the daemon uses after a dry run or a refused commit.
• Queries: dependencies, dependents, symbol_table, stats, forest, cig, modules.

Every operation stamps a single correlation id on its events; observers passed to a call apply to that call only, while subscribe attaches a workspace-wide observer. default_observers() returns the five standard observers, which the CLI attaches with --observe.

At a glance

Class diagram

Dependency diagram

API

Classes

_Operation

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

Emitter for ONE top-level operation (one run()/commit()/load).

_ForestSources

Bases: Mapping[str, str] · Line 159

Source text of a forest's modules, decoded on access.

CodemodWorkspace

Decorators: traces(spec='110', story='US-110-001', reqs=['REQ-110-001', 'REQ-110-002', 'REQ-110-003', 'REQ-110-004']) · Line 248

In-memory CST forest with Code Intelligence Graph.

Functions

Source

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

"""CodemodWorkspace: Main entry point for loading and transforming code.

Loads an entire Python codebase as a lossless module forest, builds the
Code Intelligence Graph, and provides methods for running transform
passes and committing changes atomically.
"""

from __future__ import annotations

import ast
import functools
import logging
import os
import time
import uuid
from collections.abc import Callable, Iterator, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any

import psutil

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.patterns.observer import SyncObservable
from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
from orchestrata_ultimate.refactor.codemod._cig_builder_ast import build_cig_ast
from orchestrata_ultimate.refactor.codemod._parse_cache import ParseCache, cache_key
from orchestrata_ultimate.refactor.codemod._types import WorkspaceStats
from orchestrata_ultimate.refactor.codemod.observability.debug_panel import (
    DebugPanelObserver,
)
from orchestrata_ultimate.refactor.codemod.observability.error_detector import (
    ErrorDetector,
)
from orchestrata_ultimate.refactor.codemod.observability.events import (
    CodemodEvent,
    CodemodEventType,
    Observer,
)
from orchestrata_ultimate.refactor.codemod.observability.graph_writer import (
    GraphWriterObserver,
)
from orchestrata_ultimate.refactor.codemod.observability.metrics import (
    MetricsObserver,
)
from orchestrata_ultimate.refactor.codemod.observability.tracing import (
    TracingObserver,
)
from orchestrata_ultimate.refactor.codemod.report import (
    ChangeSet,
    CommitReport,
    Finding,
)
from orchestrata_ultimate.refactor.codemod.review import PrologSemanticReview
from orchestrata_ultimate.refactor.codemod.review_engine import InMemoryReviewEngine
from orchestrata_ultimate.refactor.codemod.txn import commit_changes
from orchestrata_ultimate.refactor.ctok import LosslessModule

logger = logging.getLogger(__name__)


@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
def default_observers() -> tuple[Observer[CodemodEvent], ...]:
    """The 5 built-in observers (REQ-110-026), freshly constructed.

    Each observer is built with its own safe, no-live-infra-required
    default (``TracingObserver`` logs via the stdlib ``logging`` module,
    ``MetricsObserver`` skips Prometheus registry registration when none
    is given, ``ErrorDetector``/``GraphWriterObserver`` no-op without a
    configured event bus/repository). Constructed fresh per call so
    per-correlation state (span timers, error windows, file handles)
    never leaks across unrelated ``run()``/``commit()`` invocations.
    """
    return (
        TracingObserver(),
        MetricsObserver(),
        DebugPanelObserver(),
        ErrorDetector(),
        GraphWriterObserver(),
    )


@traces(spec="110", story="US-110-005", reqs=["REQ-110-025", "REQ-110-027"])
def _compose_observable(
    observers: Sequence[Observer[CodemodEvent]] | None,
) -> SyncObservable[CodemodEvent]:
    """Compose a ``SyncObservable[CodemodEvent]`` holding ``observers``.

    REQ-110-025: the workspace composes a ``SyncObservable[CodemodEvent]``
    and emits events for its operations. REQ-110-027: the default observer
    set is EMPTY (zero cost when nobody watches; no implicit Prometheus or
    graph writes); ``observers`` subscribes the given set. Observer dispatch
    failures are logged and swallowed by ``SyncObservable.emit`` itself
    (ERR-110-011).
    """
    observable: SyncObservable[CodemodEvent] = SyncObservable()
    for observer in observers or ():
        observable.subscribe(observer)
    return observable


@traces(spec="110", story="US-110-005", reqs=["REQ-110-025", "REQ-110-027"])
class _Operation:
    """Emitter for ONE top-level operation (one ``run()``/``commit()``/load).

    Owns the operation's ``correlation_id`` and fans every event out to the
    workspace-owned bus plus a private bus holding the call's ``observers``
    override. The private bus keeps a per-call override from receiving the
    events of a concurrent operation on the same workspace; dropping it at
    the end of the call is the unsubscribe.
    """

    def __init__(
        self,
        bus: SyncObservable[CodemodEvent] | None,
        observers: Sequence[Observer[CodemodEvent]] | None,
    ) -> None:
        self.correlation_id = uuid.uuid4().hex
        buses: list[SyncObservable[CodemodEvent]] = []
        if bus is not None:
            buses.append(bus)
        if observers:
            # An observer already subscribed on the workspace bus receives
            # each event once, through that bus (REQ-110-027).
            workspace_observers = bus.observers if bus is not None else ()
            private = [
                o for o in observers if not any(o is w for w in workspace_observers)
            ]
            if private:
                buses.append(_compose_observable(private))
        self._buses = tuple(buses)

    def emit(self, kind: CodemodEventType, data: dict[str, Any]) -> None:
        """Deliver one event; a no-op when nobody is subscribed."""
        if not any(bus.observers for bus in self._buses):
            return
        event = CodemodEvent(
            type=kind,
            timestamp=time.time(),
            data=data,
            correlation_id=self.correlation_id,
        )
        for bus in self._buses:
            bus.emit(event)

    def error(self, phase: str, exc: Exception, **extra: Any) -> None:
        """Emit ``ERROR_DETECTED`` for an exception the caller re-raises."""
        self.emit(
            CodemodEventType.ERROR_DETECTED,
            {
                "phase": phase,
                "error_type": type(exc).__name__,
                "message": str(exc),
                **extra,
            },
        )


class _ForestSources(Mapping[str, str]):
    """Source text of a forest's modules, decoded on access.

    ``build_cig_ast`` reads text only for a module it holds no tree for; the
    forest holds every tree, so nothing is decoded on a normal load. An eager
    ``{path: decoded text}`` dict was a second full copy of the tree's source
    (non-ASCII files widen to 2-4 bytes per character) live while the graph
    is built (REQ-110-004).
    """

    def __init__(self, forest: Mapping[str, LosslessModule]) -> None:
        self._forest = forest

    def __getitem__(self, path: str) -> str:
        return self._forest[path].stream.source.decode("utf-8")

    def __iter__(self) -> Iterator[str]:
        return iter(self._forest)

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


@traces(spec="110", story="US-110-001", reqs=["REQ-110-001", "REQ-110-002"])
def _parse_file(path: Path) -> tuple[Path, LosslessModule | Exception, bytes | None]:
    """Parse a single Python file.

    Args:
        path: Absolute path to the Python file.

    Returns:
        Tuple of (path, LosslessModule or Exception, source bytes or None).
    """
    try:
        data = path.read_bytes()
        return (path, LosslessModule.parse(data), data)
    except Exception as e:
        logger.debug("Failed to parse %s: %s", path, e)
        return (path, e, None)


def _parse_file_cached(
    path: Path, cache: ParseCache
) -> tuple[Path, LosslessModule | Exception, bytes | None]:
    """Parse a single Python file, going through the disk parse cache."""
    try:
        data = path.read_bytes()
        key = cache_key(data)
        module = cache.get(key)
        if module is not None:
            return (path, module, data)
        module = LosslessModule.parse(data)
        cache.put(key, module)
        return (path, module, data)
    except Exception as e:
        logger.debug("Failed to parse %s: %s", path, e)
        return (path, e, None)


def _include_root(root: Path, subdir: str) -> str | None:
    """``subdir`` as a repo-relative POSIX root without a trailing slash.

    Absolute, ``.`` and ``..`` forms are folded onto ``root`` first, so the
    stored root matches the repo-relative paths of the forest. ``None`` means
    the include is the repo root itself (no restriction). An include that
    resolves outside ``root`` raises ``ValueError``, by its lexical form or,
    through a symlink, by the path it actually reaches (``Path.resolve``).
    """
    rel = os.path.relpath(
        os.path.normpath(os.path.join(os.path.abspath(root), subdir)),
        os.path.abspath(root),
    ).replace(os.sep, "/")
    outside = ValueError(
        f"include {subdir!r} resolves outside the repo root {str(root)!r}"
    )
    if rel == ".." or rel.startswith("../"):
        raise outside
    try:
        (Path(root) / subdir).resolve().relative_to(Path(root).resolve())
    except ValueError:
        raise outside from None
    return None if rel == "." else rel


@traces(
    spec="110",
    story="US-110-001",
    reqs=["REQ-110-001", "REQ-110-002", "REQ-110-003", "REQ-110-004"],
)
class CodemodWorkspace:
    """In-memory CST forest with Code Intelligence Graph."""

    def __init__(
        self,
        root: Path,
        forest: dict[str, LosslessModule],
        unparsable: tuple[tuple[str, str], ...],
        cig: CodeIntelligenceGraph,
        stats: WorkspaceStats,
        include_roots: tuple[str, ...] = (),
        cache_dir: str | Path | None = None,
    ) -> None:
        """Wire CodemodWorkspace's composed collaborators.

        Args:
            root: Repository root directory.
            forest: Mapping of repo-relative paths to parsed LosslessModules.
            unparsable: Tuple of (path, error_message) for failed parses.
            cig: Code Intelligence Graph built from the forest.
            stats: Workspace load statistics.
            include_roots: Repo-relative roots the forest was loaded from
                (empty when the workspace was not built by ``load``).
            cache_dir: Parse-cache directory the load used, if any (a reload
                reuses it).
        """
        self._root = root
        self._include_roots = include_roots
        self._cache_dir = cache_dir
        self._forest = forest
        self._unparsable = unparsable
        self._cig = cig
        self._stats = stats
        self._module_map: dict[str, str] = {}
        self._reverse_module_map: dict[str, str] = {}
        self._observable: SyncObservable[CodemodEvent] = SyncObservable()
        self._build_module_map()

    @property
    @traces(spec="110", story="US-110-005", reqs=["REQ-110-025"])
    def observable(self) -> SyncObservable[CodemodEvent]:
        """The workspace-owned event bus (REQ-110-025); no subscribers by default."""
        return self._observable

    @traces(spec="110", story="US-110-005", reqs=["REQ-110-027"])
    def subscribe(self, observer: Observer[CodemodEvent]) -> Callable[[], None]:
        """Subscribe ``observer`` to every operation on this workspace.

        Returns:
            Zero-arg callable that unsubscribes ``observer``.
        """
        return self._observable.subscribe(observer)

    @traces(
        spec="110",
        story="US-110-001",
        reqs=["REQ-110-001", "REQ-110-002", "REQ-110-003", "REQ-110-004"],
    )
    @classmethod
    def load(
        cls,
        root: Path,
        include: tuple[str, ...] = ("src", "scripts", "tests"),
        parallel: bool = False,
        cache_dir: str | Path | None = None,
        observers: Sequence[Observer[CodemodEvent]] | None = None,
    ) -> CodemodWorkspace:
        """Parse all .py files under include roots into a lossless forest.

        Emits ``INDEX_BUILT`` and ``WORKSPACE_LOADED`` (REQ-110-025) to
        ``observers`` (None means nobody; the workspace does not exist yet,
        so only this call's override can watch its own load).

        Args:
            root: Repository root directory.
            include: Tuple of directory names to include.
            parallel: If True, use ThreadPoolExecutor for parallel parse
                and parallel CIG extraction (serial replay preserves
                identical output).
            cache_dir: Optional directory for the content-addressed parse
                cache. When set, unchanged files (by xxhash of bytes) are
                unpickled instead of re-parsed (~7x faster per file).
                Leave unset for cold-load benchmarks (REQ-110-003).

        Returns:
            CodemodWorkspace with populated forest and CIG.
        """
        op = _Operation(None, observers)
        load_started = time.monotonic()
        process = psutil.Process()
        rss_before = process.memory_info().rss

        # Folded onto the root before any file is read: an include outside the
        # repo is refused, and one entry naming the root itself lifts the
        # restriction (an empty tuple means "no roots").
        folded = [_include_root(root, subdir) for subdir in include]
        include_roots = () if None in folded else tuple(r for r in folded if r)

        files: list[Path] = []
        for rel_root in folded:
            subdir_path = root / rel_root if rel_root else root
            if subdir_path.exists():
                # Sorted: rglob order is filesystem-dependent, and node ids
                # follow file order (deterministic across machines).
                files.extend(sorted(subdir_path.rglob("*.py")))

        start_time = time.monotonic()

        parse_fn: Callable[
            [Path], tuple[Path, LosslessModule | Exception, bytes | None]
        ]
        if cache_dir is not None:
            cache = ParseCache(cache_dir)
            parse_fn = functools.partial(_parse_file_cached, cache=cache)
        else:
            parse_fn = _parse_file

        if parallel:
            max_workers = os.cpu_count() or 4
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                results = list(executor.map(parse_fn, files))
        else:
            results = [parse_fn(f) for f in files]

        parse_time = time.monotonic() - start_time

        forest: dict[str, LosslessModule] = {}
        unparsable_list: list[tuple[str, str]] = []

        for path, result, _data in results:
            rel_path = str(path.relative_to(root))
            if isinstance(result, LosslessModule):
                forest[rel_path] = result
            else:
                unparsable_list.append((rel_path, str(result)))
                op.error("parse", result, path=rel_path)

        # build_cig_ast re-parses any module it is only given text for, but
        # the forest already holds every module's ast.Module (parsed by
        # LosslessModule.parse) — pass the trees through so each file is
        # parsed exactly once. sources stays for the path set / fallback.
        sources = _ForestSources(forest)
        index_started = time.monotonic()
        try:
            cig = build_cig_ast(
                sources,
                parallel=parallel,
                trees={rel_path: module.tree for rel_path, module in forest.items()},
            )
        except Exception as exc:
            op.error("index", exc)
            raise
        op.emit(
            CodemodEventType.INDEX_BUILT,
            {
                "module_count": len(forest),
                "node_count": cig.node_count,
                "elapsed_seconds": time.monotonic() - index_started,
            },
        )

        rss_after = process.memory_info().rss
        rss_delta = rss_after - rss_before

        # stream.source is the raw bytes already in memory; m.code would
        # RECONSTRUCT every module (a full re-serialization) just to count
        # bytes — measured 2026-08-11 at ~5.4s of a full-tree serial load.
        total_bytes = sum(len(m.stream.source) for m in forest.values())

        stats = WorkspaceStats(
            file_count=len(forest),
            total_bytes=total_bytes,
            rss_delta_bytes=rss_delta,
            parse_time_seconds=parse_time,
            unparsable_count=len(unparsable_list),
        )

        op.emit(
            CodemodEventType.WORKSPACE_LOADED,
            {
                "module_count": len(forest),
                "unparsable_count": len(unparsable_list),
                "elapsed_seconds": time.monotonic() - load_started,
                "cache_enabled": cache_dir is not None,
            },
        )

        return cls(
            root=root,
            forest=forest,
            unparsable=tuple(unparsable_list),
            cig=cig,
            stats=stats,
            include_roots=include_roots,
            cache_dir=cache_dir,
        )

    def _build_module_map(self) -> None:
        """Build bidirectional mapping between dotted names and paths."""
        for path in self._forest:
            dotted = _path_to_dotted(path)
            self._module_map[dotted] = path
            self._module_map[path] = dotted

    @property
    def include_roots(self) -> tuple[str, ...]:
        """Repo-relative directories the forest was loaded from."""
        return self._include_roots

    @property
    def cache_dir(self) -> str | Path | None:
        """Parse-cache directory the forest was loaded with (None: uncached)."""
        return self._cache_dir

    @property
    def forest(self) -> Mapping[str, LosslessModule]:
        """Repo-relative path to LosslessModule mapping."""
        return self._forest

    @property
    def unparsable(self) -> tuple[tuple[str, str], ...]:
        """(path, error_message) for files that failed to parse."""
        return self._unparsable

    @property
    def cig(self) -> CodeIntelligenceGraph:
        """Code Intelligence Graph."""
        return self._cig

    @property
    def root(self) -> Path:
        """Repository root directory."""
        return self._root

    def modules(self) -> Mapping[str, str]:
        """Bidirectional mapping: dotted name and repo-relative path."""
        return self._module_map

    def dependencies(self, module: str) -> frozenset[str]:
        """Modules this module imports (direct only).

        Args:
            module: Dotted module name or repo-relative path.

        Returns:
            Set of module names/paths that this module imports.
        """
        path = self._resolve_module(module)
        if path is None:
            return frozenset()

        key = f"module:{path}"
        if key not in self._cig.key_to_id:
            return frozenset()

        node_id = self._cig.resolve(key)
        dep_ids = self._cig.dependencies(node_id)
        result: set[str] = set()
        for dep_id in dep_ids:
            dep_node = self._cig.nodes.get(dep_id)
            if dep_node and dep_node.node_type == "Module":
                result.add(dep_node.path)
        return frozenset(result)

    def dependents(self, module: str) -> frozenset[str]:
        """Modules that import this module (direct only).

        Args:
            module: Dotted module name or repo-relative path.

        Returns:
            Set of module names/paths that import this module.
        """
        path = self._resolve_module(module)
        if path is None:
            return frozenset()

        key = f"module:{path}"
        if key not in self._cig.key_to_id:
            return frozenset()

        node_id = self._cig.resolve(key)
        dep_ids = self._cig.dependents(node_id)
        result: set[str] = set()
        for dep_id in dep_ids:
            dep_node = self._cig.nodes.get(dep_id)
            if dep_node and dep_node.node_type == "Module":
                result.add(dep_node.path)
        return frozenset(result)

    def symbol_table(self) -> Mapping[str, frozenset[str]]:
        """Module path to top-level defined names.

        Returns:
            Mapping of module paths to sets of defined names
            (classes, functions, module-level variables).
        """
        table: dict[str, set[str]] = {}
        for path, module in self._forest.items():
            names: set[str] = set()
            for stmt in module.tree.body:
                if isinstance(
                    stmt, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
                ):
                    names.add(stmt.name)
                elif isinstance(stmt, ast.Assign):
                    for target in stmt.targets:
                        if isinstance(target, ast.Name):
                            names.add(target.id)
            table[path] = names
        return {k: frozenset(v) for k, v in table.items()}

    @traces(spec="110", story="US-110-001", reqs=["REQ-110-004"])
    def stats(self) -> WorkspaceStats:
        """Workspace load statistics."""
        return self._stats

    @traces(
        spec="110",
        story="US-110-003",
        reqs=["REQ-110-008", "REQ-110-009", "REQ-110-025", "REQ-110-027"],
    )
    def run(
        self,
        passes: Sequence,
        observers: Sequence[Observer[CodemodEvent]] | None = None,
    ) -> ChangeSet:
        """Apply transform passes sequentially over the forest.

        Pass N sees pass N-1's mutations. Returns composed ChangeSet.
        Emits ``PASS_STARTED``, one ``FILE_MUTATED`` per mutated file and
        ``PASS_COMPLETED`` around each pass (REQ-110-025), all sharing one
        ``correlation_id`` for this call.

        Args:
            passes: Sequence of TransformPass instances.
            observers: Observers subscribed for the duration of this call
                (REQ-110-027), in addition to those subscribed on the
                workspace. ``None`` adds none; the default set is empty.

        Returns:
            ChangeSet with all mutations from all passes.
        """
        op = _Operation(self._observable, observers)

        composed = ChangeSet(entries=())
        for pass_ in passes:
            pass_name = getattr(pass_, "name", type(pass_).__name__)
            pass_type = getattr(pass_, "pass_type", "transform")
            op.emit(
                CodemodEventType.PASS_STARTED,
                {"pass_name": pass_name, "pass_type": pass_type},
            )
            pass_start = time.monotonic()
            try:
                result = pass_.apply(self)
            except Exception as exc:
                op.error("pass", exc, pass_name=pass_name)
                raise
            pass_duration = time.monotonic() - pass_start
            composed = composed.merge(result)
            for entry in result.entries:
                self._forest[entry.path] = entry.mutated_tree
                op.emit(
                    CodemodEventType.FILE_MUTATED,
                    {"path": entry.path, "pass_name": entry.pass_name},
                )
            op.emit(
                CodemodEventType.PASS_COMPLETED,
                {
                    "pass_name": pass_name,
                    "pass_type": pass_type,
                    "file_count": result.file_count,
                    "duration_seconds": pass_duration,
                },
            )
        return composed

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-137"])
    def restore(self, originals: Mapping[str, LosslessModule]) -> None:
        """Put the load-time trees back into the forest.

        ``run`` writes every mutated tree into the forest; a dry run (or a
        commit that wrote nothing) leaves that forest ahead of the disk. A
        daemon serving the forest calls this with the ``original_tree`` of
        each path the run touched so the next client sees the disk state.

        Args:
            originals: Repo-relative path to the tree the path held before
                the run mutated it.
        """
        for path, tree in originals.items():
            self._forest[path] = tree

    @traces(
        spec="110",
        story="US-110-005",
        reqs=[
            "REQ-110-013",
            "REQ-110-014",
            "REQ-110-015",
            "REQ-110-016",
            "REQ-110-017",
            "REQ-110-025",
            "REQ-110-027",
            "REQ-110-033",
            "REQ-110-035",
        ],
    )
    def commit(
        self,
        changes: ChangeSet,
        reviews: Sequence | None = None,
        dry_run: bool = True,
        observers: Sequence[Observer[CodemodEvent]] | None = None,
        review_engine: InMemoryReviewEngine | None = None,
        run_review_engine: bool = True,
    ) -> CommitReport:
        """Review mutated forest, then write changes atomically.

        Emits ``REVIEW_STARTED``/``REVIEW_COMPLETED`` with one
        ``FINDING_EMITTED`` per finding, then ``COMMIT_STARTED`` followed by
        ``COMMIT_COMPLETED``; a review-gate refusal emits ``COMMIT_ABORTED``,
        and a failed write emits ``ROLLBACK_TRIGGERED`` (from the
        transaction), ``ERROR_DETECTED`` and ``COMMIT_ABORTED`` before the
        exception propagates (REQ-110-025). All events of one call share one
        ``correlation_id``.

        Args:
            changes: ChangeSet from workspace.run().
            reviews: Review passes to run before write.
            dry_run: If True, produce diff and report without writing.
            observers: Observers subscribed for the duration of this call
                (REQ-110-027), in addition to those subscribed on the
                workspace. ``None`` adds none; the default set is empty.
            review_engine: The in-memory 40-check engine to run before the
                ``reviews`` (REQ-110-033/035). ``None`` with
                ``reviews=None`` builds ``InMemoryReviewEngine()``; an
                explicit ``reviews`` list runs as given. By default an
                error-severity finding aborts the commit only when the change
                introduced it (ADR-0073); pass
                ``review_engine=InMemoryReviewEngine(baseline='absolute')``
                for the unconditional abort.
            run_review_engine: ``False`` skips the engine entirely.

        Returns:
            CommitReport with passes run, files written, findings, duration.
            ``legacy_findings`` holds what the default engine found already
            present in the original trees (reported, never vetoing); each is
            also emitted as ``FINDING_EMITTED`` with ``legacy: True``.
        """
        op = _Operation(self._observable, observers)
        correlation_id = op.correlation_id
        start_time = time.monotonic()
        all_findings: list[Finding] = []
        legacy_findings: list[Finding] = []
        active = _resolve_reviews(reviews, review_engine, run_review_engine)

        if active:
            op.emit(CodemodEventType.REVIEW_STARTED, {"review_count": len(active)})
            for review in active:
                try:
                    if isinstance(review, InMemoryReviewEngine):
                        outcome = review.review_detailed(self, changes)
                        findings = list(outcome.introduced)
                        legacy = list(outcome.legacy)
                    else:
                        findings = review.review(self, changes)
                        legacy = []
                except Exception as exc:
                    op.error("review", exc, review_name=getattr(review, "name", ""))
                    op.emit(CodemodEventType.COMMIT_ABORTED, {"reason": "review_error"})
                    raise
                all_findings.extend(findings)
                legacy_findings.extend(legacy)
                for finding in findings:
                    op.emit(CodemodEventType.FINDING_EMITTED, _finding_payload(finding))
                for finding in legacy:
                    op.emit(
                        CodemodEventType.FINDING_EMITTED,
                        {**_finding_payload(finding), "legacy": True},
                    )
            error_count = sum(1 for f in all_findings if f.severity == "error")
            warning_count = sum(1 for f in all_findings if f.severity == "warning")
            op.emit(
                CodemodEventType.REVIEW_COMPLETED,
                {
                    "finding_count": len(all_findings),
                    "error_count": error_count,
                    "warning_count": warning_count,
                    "legacy_count": len(legacy_findings),
                },
            )

        error_findings = [f for f in all_findings if f.severity == "error"]
        if error_findings:
            op.emit(
                CodemodEventType.COMMIT_ABORTED,
                {"reason": "review_errors", "error_count": len(error_findings)},
            )
            duration = time.monotonic() - start_time
            return CommitReport(
                success=False,
                files_written=(),
                findings=tuple(all_findings),
                duration_seconds=duration,
                dry_run=dry_run,
                passes_run=tuple(r.name for r in active),
                correlation_id=correlation_id,
                legacy_findings=tuple(legacy_findings),
            )

        op.emit(
            CodemodEventType.COMMIT_STARTED,
            {"file_count": changes.file_count, "dry_run": dry_run},
        )

        try:
            files_written, _rolled_back = commit_changes(
                self._root, changes, dry_run=dry_run, on_event=op.emit
            )
        except Exception as exc:
            op.error("commit", exc)
            op.emit(CodemodEventType.COMMIT_ABORTED, {"reason": "write_failed"})
            raise

        duration = time.monotonic() - start_time
        op.emit(
            CodemodEventType.COMMIT_COMPLETED,
            {
                "files_written": len(files_written),
                "file_count": len(files_written),
                "duration_seconds": duration,
                "dry_run": dry_run,
                "pass_count": len(active),
                "finding_count": len(all_findings),
                "status": "success",
            },
        )

        return CommitReport(
            success=True,
            files_written=tuple(files_written),
            findings=tuple(all_findings),
            duration_seconds=duration,
            dry_run=dry_run,
            passes_run=tuple(r.name for r in active),
            correlation_id=correlation_id,
            legacy_findings=tuple(legacy_findings),
        )

    def _resolve_module(self, module: str) -> str | None:
        """Resolve a module name to a repo-relative path."""
        if module in self._forest:
            return module
        if module in self._module_map:
            mapped = self._module_map[module]
            if mapped in self._forest:
                return mapped
        dotted = _path_to_dotted(module)
        if dotted in self._module_map:
            return self._module_map[dotted]
        return None


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
def _finding_payload(finding: Finding) -> dict[str, Any]:
    """The ``FINDING_EMITTED`` event payload of ``finding``."""
    return {
        "severity": finding.severity,
        "rule": finding.rule,
        "path": finding.path,
        "line": finding.line,
        "message": finding.message,
    }


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-035"])
def _resolve_reviews(
    reviews: Sequence | None,
    review_engine: InMemoryReviewEngine | None,
    run_review_engine: bool,
) -> list:
    """The review passes ``commit`` runs, the in-memory engine first.

    An explicit ``review_engine`` always runs; with ``reviews=None`` and no
    engine, the default ``InMemoryReviewEngine()`` is built. A caller list is
    run as given and does not gain an engine. When the list already holds a
    ``PrologSemanticReview`` the engine skips its own phase 3 so the Prolog
    pass does not run twice.
    """
    passes = list(reviews) if reviews else []
    if not run_review_engine:
        return passes
    engine = review_engine
    if engine is None and reviews is None:
        engine = InMemoryReviewEngine()
    if engine is None:
        return passes
    if any(isinstance(r, PrologSemanticReview) for r in passes):
        engine = engine.without_phase3()
    return [engine, *passes]


def _path_to_dotted(path: str) -> str:
    """Convert a repo-relative path to a dotted module name.

    Strips src/ prefix and .py suffix, converts / to dots.
    """
    p = path
    for prefix in ("src/", "scripts/", "tests/"):
        if p.startswith(prefix):
            p = p[len(prefix) :]
            break
    if p.endswith(".py"):
        p = p[:-3]
    if p.endswith("/__init__"):
        p = p[:-9]
    return p.replace("/", ".")