orchestrata_ultimate.refactor.codemod.server

Reference › orchestrata_ultimate › refactor › codemod › server

The Unix-socket JSON-RPC 2.0 server that holds a loaded workspace in memory and answers queries and transformations from clients. It is started by _daemon_entry and uses the graph, the passes, the review engine and the analysis protocol. Its design is told in the S4 and S5 packets of the refactor package narrative.

CodemodServer.serve(socket_path, workspace) removes a stale socket, binds and listens, and starts one thread per connection (_handle_client), each reading requests and replying through handle_request and _dispatch. Read methods run in parallel; write methods are serialized by a mutex. A write is a two-step transaction: run_pass opens or extends a pending run (_PendingRun, identified by run_id) that records each touched file's original tree once and holds the write lock, and commit looks the run up, runs the review engine with the requested baseline, commits or dry-runs it and returns the report. After a dry run, a refused review, a failed commit, a closed connection or an expired lease (120 seconds idle by default) the server restores the forest, so one client's mutations never leak to the next. A blocked writer waits at most the lease plus five seconds and then fails with a named error.

replace_workspace swaps in a reloaded workspace under the same lock, and refuses the swap if a commit has written to disk since the reload began. The status reply merges in the identity fields given at construction. Errors are returned as JSON-RPC error objects.

At a glance

Class diagram

Dependency diagram

API

Classes

_PendingRun (dataclass)

Decorators: dataclass · Line 92

A run the daemon holds by reference between run_pass and commit.

CodemodServer

Decorators: traces(spec='110', story='US-110-013', reqs=['REQ-110-134', 'REQ-110-137', 'REQ-110-138']) · Line 139

Unix socket JSON-RPC 2.0 server for the codemod workspace.

Functions

Source

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

"""SPEC-110 codemod engine — Unix socket JSON-RPC 2.0 server.

Serves the CodemodWorkspace over a Unix domain socket using JSON-RPC 2.0
protocol. Supports concurrent read operations and serialized writes.
"""

from __future__ import annotations

import itertools
import json
import logging
import socket
import threading
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._types import BlastRadius
from orchestrata_ultimate.refactor.codemod.analysis.protocol import resolve_node_id
from orchestrata_ultimate.refactor.codemod.report import ChangeSet
from orchestrata_ultimate.refactor.codemod.review_engine import InMemoryReviewEngine
from orchestrata_ultimate.refactor.codemod.workspace import (
    CodemodWorkspace,
    default_observers,
)

if TYPE_CHECKING:
    from collections.abc import Sequence

    from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
    from orchestrata_ultimate.refactor.ctok import LosslessModule

logger = logging.getLogger(__name__)

JSONRPC_VERSION = "2.0"
MAX_MESSAGE_SIZE = 10 * 1024 * 1024

READ_METHODS = frozenset(
    {
        "query_dependencies",
        "query_dependents",
        "query_blast_radius",
        "query_breaking_changes",
        "query_cycles",
        "query_topological_order",
        "query_affected_tests",
        "query_coupling_score",
        "query_abstraction_level",
        "query_centrality",
        "query_impact_score",
        "resolve_nodes",
        "status",
    }
)

WRITE_METHODS = frozenset(
    {
        "run_pass",
        "commit",
        "reload",
    }
)

# A blocked writer waits ``lease_seconds + _LOCK_WAIT_MARGIN`` for the write
# lock, polling every ``_LOCK_POLL_SECONDS`` so an idle holder's lease expiry
# frees it. ``_EXPIRED_MEMORY`` bounds the run_ids remembered as expired.
_LOCK_WAIT_MARGIN = 5.0
_LOCK_POLL_SECONDS = 0.1
_EXPIRED_MEMORY = 256


@traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
def _jsonrpc_error(code: int, message: str, request_id: Any = None) -> dict:
    """Build a JSON-RPC 2.0 error response."""
    return {
        "jsonrpc": JSONRPC_VERSION,
        "error": {"code": code, "message": message},
        "id": request_id,
    }


@traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
def _jsonrpc_result(result: Any, request_id: Any = None) -> dict:
    """Build a JSON-RPC 2.0 success response."""
    return {"jsonrpc": JSONRPC_VERSION, "result": result, "id": request_id}


@dataclass
class _PendingRun:
    """A run the daemon holds by reference between ``run_pass`` and ``commit``.

    ``originals`` keeps, per touched path, the tree the path held before the
    run first mutated it, so the forest can be put back after a dry run.
    """

    changes: ChangeSet
    originals: dict[str, LosslessModule]
    conn_id: int | None
    # The lease clock: refreshed by every ``run_pass`` / ``commit`` of the run.
    touched: float = field(default_factory=time.monotonic)
    # True while a request of the owning connection works on the run; a busy
    # run is never expired (its owner is not idle).
    busy: bool = False


def _close_observers(observers: Sequence[Any] | None) -> None:
    """Release the handles the default observers hold (the debug panel's)."""
    for observer in observers or ():
        close = getattr(observer, "close", None)
        if callable(close):
            close()


def _commit_run_id(params: dict) -> str | None:
    """The ``run_id`` a ``commit`` names in its ``changes`` parameter."""
    changes = params.get("changes")
    run_id = changes.get("run_id") if isinstance(changes, dict) else None
    return run_id if isinstance(run_id, str) else None


def _resolve_key(cig: CodeIntelligenceGraph, key: object) -> int:
    """Node id for a module key, by the strategies the CLI's library mode uses.

    Raises:
        KeyError: ``key`` names no node (reported as ``Unknown node``).
    """
    node_id = resolve_node_id(cig, key) if isinstance(key, str) else None
    if node_id is None:
        raise KeyError(key)
    return node_id


@traces(
    spec="110", story="US-110-013", reqs=["REQ-110-134", "REQ-110-137", "REQ-110-138"]
)
class CodemodServer:
    """Unix socket JSON-RPC 2.0 server for the codemod workspace.

    Serves workspace queries and transformations over a Unix domain socket.
    Read operations run in parallel; write operations are serialized via mutex.
    """

    def __init__(
        self, identity: dict | None = None, *, lease_seconds: float = 120.0
    ) -> None:
        """Initialize CodemodServer with no active socket.

        Args:
            identity: Static fields (repo_root, code_root, deployed_commit,
                pid) merged into every ``status`` reply so a deployed daemon
                can be told apart from one running out of a checkout
                (ADR-0067: identity, not health).
            lease_seconds: How long a pending run may sit idle before another
                connection's write expires it (restores the forest, frees the
                write lock). A blocked writer waits at most
                ``lease_seconds + 5`` seconds for the lock.
        """
        self._identity: dict = dict(identity) if identity else {}
        self._lease_seconds = lease_seconds
        # run_ids a lease expired, so the owner's late request says so.
        self._expired: dict[str, None] = {}
        self._workspace: CodemodWorkspace | None = None
        self._socket: socket.socket | None = None
        self._running = False
        self._write_lock = threading.Lock()
        self._client_count = 0
        self._client_lock = threading.Lock()
        self._socket_path: Path | None = None
        # Runs held by reference between run_pass and commit. A pending run
        # owns ``_write_lock`` until its commit or its connection's close, so
        # at most one exists at a time.
        self._pending: dict[str, _PendingRun] = {}
        self._pending_lock = threading.Lock()
        self._conn_ids = itertools.count(1)
        # Counts the commits that wrote the disk (bumped under ``_write_lock``);
        # a reload loaded before such a commit holds a pre-commit tree and is
        # dropped by ``replace_workspace``.
        self.commit_generation = 0

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def serve(self, socket_path: Path, workspace: CodemodWorkspace) -> None:
        """Start listening on Unix socket and serving JSON-RPC requests.

        Args:
            socket_path: Path to the Unix domain socket file.
            workspace: The CodemodWorkspace to serve.

        Raises:
            OSError: If socket creation or binding fails.
        """
        self._workspace = workspace
        self._socket_path = socket_path

        if socket_path.exists():
            socket_path.unlink()

        socket_path.parent.mkdir(parents=True, exist_ok=True)

        self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._socket.bind(str(socket_path))
        self._socket.listen(16)
        self._socket.settimeout(1.0)
        self._running = True

        logger.info("CodemodServer listening on %s", socket_path)

        try:
            while self._running:
                try:
                    conn, _ = self._socket.accept()
                except TimeoutError:
                    continue
                except OSError:
                    if self._running:
                        logger.error("Socket accept error", exc_info=True)
                    break

                with self._client_lock:
                    self._client_count += 1

                t = threading.Thread(
                    target=self._handle_client,
                    args=(conn,),
                    daemon=True,
                )
                t.start()
        finally:
            self._cleanup_socket()

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def handle_request(self, request: dict, *, conn_id: int | None = None) -> dict:
        """Dispatch a JSON-RPC 2.0 request to the appropriate handler.

        Args:
            request: Parsed JSON-RPC 2.0 request dict.
            conn_id: The connection the request arrived on; a pending run is
                owned by it. ``None`` is the in-process caller (tests).

        Returns:
            JSON-RPC 2.0 response dict.
        """
        if self._workspace is None:
            return _jsonrpc_error(
                -32603, "Workspace not initialized", request.get("id")
            )

        if request.get("jsonrpc") != JSONRPC_VERSION:
            return _jsonrpc_error(-32600, "Invalid JSON-RPC version", request.get("id"))

        method = request.get("method")
        if not isinstance(method, str):
            return _jsonrpc_error(-32600, "Invalid method", request.get("id"))

        params = request.get("params", {})
        request_id = request.get("id")

        if method not in READ_METHODS and method not in WRITE_METHODS:
            return _jsonrpc_error(-32601, f"Method not found: {method}", request_id)

        if method in WRITE_METHODS:
            return self._handle_write(method, params, request_id, conn_id)
        return self._dispatch(method, params, request_id)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-137"])
    def _handle_write(
        self, method: str, params: Any, request_id: Any, conn_id: int | None
    ) -> dict:
        """Serialize a write method (REQ-110-137).

        A ``run_pass`` that opens a run takes ``_write_lock`` and keeps it
        until that run's ``commit`` or its connection's close; a ``run_pass``
        or ``commit`` naming a run its connection already holds runs under
        that lock without taking it again.
        """
        if not isinstance(params, dict):
            return _jsonrpc_error(-32602, "Invalid params", request_id)
        if method == "run_pass":
            return self._open_or_extend_run(params, request_id, conn_id)
        if method == "commit":
            return self._dispatch(method, params, request_id, conn_id=conn_id)
        if self._owned_run_id(conn_id) is not None:
            return _jsonrpc_error(
                -32602, "Commit the pending run before reload", request_id
            )
        if not self._acquire_write_lock(conn_id):
            return _jsonrpc_error(-32000, self._lock_timeout_message(), request_id)
        try:
            return self._dispatch(method, params, request_id, conn_id=conn_id)
        finally:
            self._write_lock.release()

    def _lock_wait_seconds(self) -> float:
        """The longest a blocked writer waits for the write lock."""
        return self._lease_seconds + _LOCK_WAIT_MARGIN

    def _lock_timeout_message(self) -> str:
        return f"Write lock held for more than {self._lock_wait_seconds():g}s"

    def _expired_message(self) -> str:
        return f"Run expired after {self._lease_seconds:g}s idle"

    def _acquire_write_lock(self, conn_id: int | None) -> bool:
        """Take the write lock, expiring idle pending runs of other connections.

        Waits at most ``lease_seconds + 5`` seconds; ``False`` on timeout.
        """
        deadline = time.monotonic() + self._lock_wait_seconds()
        while True:
            self._expire_stale_runs(conn_id)
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return False
            if self._write_lock.acquire(timeout=min(_LOCK_POLL_SECONDS, remaining)):
                return True

    def _expire_stale_runs(self, requester: int | None) -> None:
        """End every pending run of another connection idle past its lease."""
        now = time.monotonic()
        with self._pending_lock:
            stale = [
                run_id
                for run_id, run in self._pending.items()
                if run.conn_id != requester
                and not run.busy
                and now - run.touched > self._lease_seconds
            ]
        for run_id in stale:
            if self._end_run(run_id, expire=True):
                logger.warning("Pending run %s expired after its lease", run_id)

    def _open_or_extend_run(
        self, params: dict, request_id: Any, conn_id: int | None
    ) -> dict:
        """Run a pass inside a pending run, opening the run (and taking the
        write lock) when its ``run_id`` is new."""
        run_id = params.get("run_id") or uuid.uuid4().hex
        if not isinstance(run_id, str):
            return _jsonrpc_error(-32602, "Invalid run_id", request_id)
        params = {**params, "run_id": run_id}
        with self._pending_lock:
            existing = self._pending.get(run_id)
            expired = existing is None and run_id in self._expired
            if existing is not None and existing.conn_id == conn_id:
                existing.busy = True
        if expired:
            return _jsonrpc_error(-32602, self._expired_message(), request_id)
        if existing is not None:
            if existing.conn_id != conn_id:
                return _jsonrpc_error(-32602, "Unknown or foreign run_id", request_id)
            try:
                return self._dispatch("run_pass", params, request_id, conn_id=conn_id)
            finally:
                existing.busy = False
                existing.touched = time.monotonic()
        if self._owned_run_id(conn_id) is not None:
            return _jsonrpc_error(
                -32602, "Connection already holds a pending run", request_id
            )
        if not self._acquire_write_lock(conn_id):
            return _jsonrpc_error(-32000, self._lock_timeout_message(), request_id)
        try:
            response = self._dispatch("run_pass", params, request_id, conn_id=conn_id)
        except BaseException:
            self._write_lock.release()
            raise
        with self._pending_lock:
            opened = run_id in self._pending
        if not opened:
            self._write_lock.release()
        return response

    def _owned_run_id(self, conn_id: int | None) -> str | None:
        """The run_id of the pending run ``conn_id`` holds, if any."""
        with self._pending_lock:
            for run_id, run in self._pending.items():
                if run.conn_id == conn_id:
                    return run_id
        return None

    def _drop_run(self, run_id: str) -> None:
        """Forget a pending run and release the write lock it held."""
        with self._pending_lock:
            dropped = self._pending.pop(run_id, None)
        if dropped is not None:
            self._write_lock.release()

    def _end_run(self, run_id: str, *, expire: bool = False) -> bool:
        """Abandon a pending run: restore its forest, free the write lock.

        The pop is atomic, so a lease expiry and a connection close racing
        for the same run restore it exactly once. ``expire`` remembers the
        run_id so its owner's late request is told the lease ran out, and
        re-checks the lease under the lock: the run must still be pending,
        idle and past its lease, because its owner may have gone busy since
        the caller listed it as stale (a commit in flight is never expired).
        """
        with self._pending_lock:
            pending = self._pending.get(run_id)
            if pending is None:
                return False
            if expire and (
                pending.busy
                or time.monotonic() - pending.touched <= self._lease_seconds
            ):
                return False
            del self._pending[run_id]
            if expire:
                self._expired[run_id] = None
                while len(self._expired) > _EXPIRED_MEMORY:
                    del self._expired[next(iter(self._expired))]
        if self._workspace is not None:
            self._workspace.restore(pending.originals)
        self._write_lock.release()
        return True

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-137"])
    def release_connection(self, conn_id: int | None) -> None:
        """Abandon the pending run a closed connection held.

        Puts the load-time trees back into the forest and frees the write
        lock, so a client that disconnects mid-run leaves no mutation and no
        wedged daemon behind.
        """
        run_id = self._owned_run_id(conn_id)
        if run_id is not None:
            self._end_run(run_id)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-137"])
    def replace_workspace(
        self, new_ws: CodemodWorkspace, loaded_at_generation: int | None = None
    ) -> bool:
        """Swap the served workspace under the write lock.

        The inotify reload calls this with a workspace it loaded outside the
        lock (the load is seconds long); the swap waits for a pending run to
        commit or end, so no run ever sees the forest change between its
        passes. A commit that wrote the disk while the reload loaded makes the
        loaded tree older than the disk: when ``loaded_at_generation`` (the
        ``commit_generation`` read before the load) differs from the current
        one the swap is dropped, and the commit's own writes queue the next
        reload.

        Args:
            new_ws: The freshly loaded workspace.
            loaded_at_generation: ``commit_generation`` as read before the
                load; ``None`` swaps unconditionally.

        Returns:
            True when the workspace was swapped, False when it was dropped as
            older than a commit.

        Raises:
            TimeoutError: The write lock stayed held past the lease and its
                margin.
        """
        if not self._acquire_write_lock(None):
            raise TimeoutError(self._lock_timeout_message())
        try:
            if (
                loaded_at_generation is not None
                and loaded_at_generation != self.commit_generation
            ):
                logger.info(
                    "Reload dropped: loaded at commit generation %d, now %d",
                    loaded_at_generation,
                    self.commit_generation,
                )
                return False
            self._workspace = new_ws
            return True
        finally:
            self._write_lock.release()

    def stop(self) -> None:
        """Stop the server and clean up the socket."""
        self._running = False
        if self._socket is not None:
            try:
                self._socket.close()
            except OSError as e:
                logger.warning("Error closing server socket: %s", e)
            self._socket = None
        self._cleanup_socket()
        logger.info("CodemodServer stopped")

    def request_stop(self) -> None:
        """Async-signal-safe stop: flag + close socket, no logging/locks.

        Safe to call from a signal handler. Full cleanup (logging, socket
        file removal) happens in serve()'s finally / a later stop() call.
        """
        self._running = False
        if self._socket is not None:
            try:
                self._socket.close()
            except OSError:
                pass

    @property
    def client_count(self) -> int:
        """Number of currently connected clients."""
        return self._client_count

    def _handle_client(self, conn: socket.socket) -> None:
        """Handle a single client connection, processing messages in a loop."""
        with self._client_lock:
            conn_id = next(self._conn_ids)
        buffer = b""
        try:
            conn.settimeout(300.0)
            while self._running:
                try:
                    data = conn.recv(65536)
                except TimeoutError:
                    continue
                except OSError:
                    break

                if not data:
                    break

                buffer += data

                while b"\n" in buffer:
                    line, buffer = buffer.split(b"\n", 1)
                    if not line.strip():
                        continue
                    try:
                        request = json.loads(line.decode("utf-8"))
                    except (json.JSONDecodeError, UnicodeDecodeError) as e:
                        response = _jsonrpc_error(-32700, f"Parse error: {e}")
                        self._send_response(conn, response)
                        continue

                    try:
                        response = self.handle_request(request, conn_id=conn_id)
                    except Exception:
                        logger.error("Unhandled error in request", exc_info=True)
                        # `request` is any parsed JSON value here; a non-dict
                        # (e.g. a list) has no .get and must not kill the
                        # handler thread while building the error response.
                        request_id = (
                            request.get("id") if isinstance(request, dict) else None
                        )
                        response = _jsonrpc_error(-32603, "Internal error", request_id)

                    self._send_response(conn, response)
        finally:
            # A client that vanishes mid-run abandons the run: restore the
            # forest and free the write lock for the next client.
            self.release_connection(conn_id)
            with self._client_lock:
                self._client_count -= 1
            try:
                conn.close()
            except OSError:
                pass

    def _send_response(self, conn: socket.socket, response: dict) -> None:
        """Send a JSON-RPC response over the connection."""
        try:
            payload = json.dumps(response) + "\n"
            conn.sendall(payload.encode("utf-8"))
        except OSError as e:
            logger.warning("Error sending response: %s", e)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def _dispatch(
        self,
        method: str,
        params: dict,
        request_id: Any,
        *,
        conn_id: int | None = None,
    ) -> dict:
        """Route a method call to the workspace or CIG."""
        ws = self._workspace
        assert ws is not None

        try:
            if method == "query_dependencies":
                node_id = params.get("node_id", "")
                result = ws.cig.dependencies(_resolve_key(ws.cig, node_id))
                return _jsonrpc_result(result, request_id)

            if method == "query_dependents":
                node_id = params.get("node_id", "")
                result = ws.cig.dependents(_resolve_key(ws.cig, node_id))
                return _jsonrpc_result(result, request_id)

            if method == "query_blast_radius":
                node_id = params.get("node_id", "")
                hops = params.get("hops", 3)
                br: BlastRadius = ws.cig.blast_radius(
                    _resolve_key(ws.cig, node_id), hops
                )
                return _jsonrpc_result(
                    {
                        "affected_nodes": list(br.affected_nodes),
                        "affected_files": list(br.affected_files),
                        "impact_score": br.impact_score,
                    },
                    request_id,
                )

            if method == "query_breaking_changes":
                from orchestrata_ultimate.refactor.codemod._types import CodeChange

                change_data = params.get("change", {})
                change = CodeChange(**change_data)
                breaking = ws.cig.breaking_changes(change)
                return _jsonrpc_result(
                    [
                        {
                            "change_path": bc.change_path,
                            "consumer_path": bc.consumer_path,
                            "consumer_line": bc.consumer_line,
                            "incompatibility": bc.incompatibility,
                        }
                        for bc in breaking
                    ],
                    request_id,
                )

            if method == "query_cycles":
                return _jsonrpc_result(ws.cig.cycles(), request_id)

            if method == "query_topological_order":
                return _jsonrpc_result(ws.cig.topological_order(), request_id)

            if method == "query_affected_tests":
                function_ids = frozenset(params.get("function_ids", []))
                affected = ws.cig.affected_tests(function_ids)
                return _jsonrpc_result(list(affected), request_id)

            if method == "query_coupling_score":
                node_id = params.get("node_id", "")
                cs = ws.cig.coupling_score(_resolve_key(ws.cig, node_id))
                return _jsonrpc_result(
                    {
                        "efferent": cs.efferent,
                        "afferent": cs.afferent,
                        "instability": cs.instability,
                    },
                    request_id,
                )

            if method == "query_abstraction_level":
                node_id = params.get("node_id", "")
                al = ws.cig.abstraction_level(_resolve_key(ws.cig, node_id))
                return _jsonrpc_result(
                    {
                        "abstract_count": al.abstract_count,
                        "concrete_count": al.concrete_count,
                        "ratio": al.ratio,
                    },
                    request_id,
                )

            if method == "query_centrality":
                top_k = params.get("top_k", 20)
                exact = params.get("exact", False)
                return _jsonrpc_result(
                    ws.cig.centrality(top_k, exact=exact), request_id
                )

            if method == "query_impact_score":
                # RPC name kept for backward compatibility; the CIG method
                # is pagerank_score (renamed to avoid colliding with
                # BlastRadius.impact_score).
                node_id = params.get("node_id", "")
                return _jsonrpc_result(
                    ws.cig.pagerank_score(_resolve_key(ws.cig, node_id)), request_id
                )

            if method == "run_pass":
                return self._run_pass(ws, params, request_id, conn_id)

            if method == "commit":
                return self._commit(ws, params, request_id, conn_id)

            if method == "resolve_nodes":
                nodes = ws.cig.nodes
                found = [
                    {
                        "id": node_id,
                        "qualified_name": nodes[node_id].qualified_name,
                        "path": nodes[node_id].path,
                    }
                    for node_id in params.get("node_ids", [])
                    if node_id in nodes
                ]
                return _jsonrpc_result(found, request_id)

            if method == "reload":
                # The roots and cache the daemon was loaded with; no stored
                # roots means the whole repo, which load spells ".".
                new_ws = CodemodWorkspace.load(
                    ws.root,
                    include=tuple(ws.include_roots) or (".",),
                    parallel=True,
                    cache_dir=ws.cache_dir,
                )
                self._workspace = new_ws
                return _jsonrpc_result({"status": "reloaded"}, request_id)

            if method == "status":
                import psutil

                proc = psutil.Process()
                return _jsonrpc_result(
                    {
                        "memory_rss": proc.memory_info().rss,
                        "client_count": self._client_count,
                        "pending_runs": len(self._pending),
                        "file_count": ws.stats().file_count,
                        "include": list(ws.include_roots),
                        **self._identity,
                    },
                    request_id,
                )

            return _jsonrpc_error(-32601, f"Method not found: {method}", request_id)

        except KeyError as e:
            # Every KeyError reachable from this dispatcher comes from
            # cig.resolve(): the parameter was PRESENT, its value just names
            # no node in the graph. "Missing parameter" was unconditionally
            # false -- no branch here indexes `params` (they all use .get with
            # a default), so a genuinely absent parameter cannot raise
            # KeyError at all. Three tests asserted that wording and passed
            # for the wrong reason.
            return _jsonrpc_error(-32602, f"Unknown node: {e}", request_id)
        except Exception:
            logger.error("Error dispatching %s", method, exc_info=True)
            return _jsonrpc_error(-32603, "Internal error", request_id)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def _run_pass(
        self, ws: CodemodWorkspace, params: dict, request_id: Any, conn_id: int | None
    ) -> dict:
        """Apply one pass and record it in the run named by ``run_id``.

        The mutations stay in the forest (the next pass of the run sees
        them); the pending run keeps the pre-run tree of every path the run
        touched, and the ChangeSet itself, so ``commit`` needs no payload.
        """
        pass_name = params.get("pass_name", "")
        pass_cls = _resolve_pass(pass_name)
        if pass_cls is None:
            return _jsonrpc_error(-32602, f"Unknown pass: {pass_name}", request_id)
        run_id = params.get("run_id") or uuid.uuid4().hex
        observers = default_observers() if params.get("observe") else None
        try:
            pass_instance = pass_cls(**params.get("params", {}))
            result = ws.run([pass_instance], observers=observers)
        finally:
            _close_observers(observers)
        with self._pending_lock:
            pending = self._pending.setdefault(
                run_id, _PendingRun(ChangeSet(), {}, conn_id)
            )
        for entry in result.entries:
            pending.originals.setdefault(entry.path, entry.original_tree)
        pending.changes = pending.changes.merge(result)
        return _jsonrpc_result(
            {
                "run_id": run_id,
                "file_count": pending.changes.file_count,
                "paths": sorted(pending.changes.paths),
                "pass_name": getattr(
                    pass_instance, "name", type(pass_instance).__name__
                ),
            },
            request_id,
        )

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def _commit(
        self, ws: CodemodWorkspace, params: dict, request_id: Any, conn_id: int | None
    ) -> dict:
        """Review and (unless ``dry_run``) write the run ``changes.run_id`` names.

        Whatever the outcome, the run ends here: the forest is put back
        unless the files were written, and the write lock is released.
        """
        run_id = _commit_run_id(params)
        with self._pending_lock:
            pending = self._pending.get(run_id) if run_id is not None else None
            if pending is not None and pending.conn_id == conn_id:
                pending.busy = True
            expired = run_id in self._expired
        if run_id is None or pending is None or pending.conn_id != conn_id:
            if pending is None and expired:
                return _jsonrpc_error(-32602, self._expired_message(), request_id)
            return _jsonrpc_error(-32602, "Unknown or foreign run_id", request_id)
        dry_run = params.get("dry_run", True)
        baseline = params.get("baseline", "delta")
        observers = default_observers() if params.get("observe") else None
        written = False
        try:
            report = ws.commit(
                pending.changes,
                dry_run=dry_run,
                observers=observers,
                review_engine=InMemoryReviewEngine(baseline=baseline),
            )
            written = report.success and not dry_run
            if written:
                self.commit_generation += 1  # still under the run's write lock
        except Exception as exc:
            # The run ends (the finally restores and releases); the caller
            # gets the cause, not an anonymous internal error.
            logger.error("Commit of run %s failed", run_id, exc_info=True)
            return _jsonrpc_error(-32000, f"{type(exc).__name__}: {exc}", request_id)
        finally:
            _close_observers(observers)
            if not written:
                ws.restore(pending.originals)
            self._drop_run(run_id)
        return _jsonrpc_result(
            {
                "success": report.success,
                "files_written": list(report.files_written),
                "findings": [
                    {
                        "severity": f.severity,
                        "rule": f.rule,
                        "path": f.path,
                        "line": f.line,
                        "message": f.message,
                    }
                    for f in report.findings
                ],
                "duration_seconds": report.duration_seconds,
                "dry_run": report.dry_run,
                "report": json.loads(report.to_json()),
                "baseline": baseline,
            },
            request_id,
        )

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def _cleanup_socket(self) -> None:
        """Remove the Unix socket file if it exists."""
        if self._socket_path is not None and self._socket_path.exists():
            try:
                self._socket_path.unlink()
            except OSError as e:
                logger.warning("Error removing socket file: %s", e)


@traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
def _resolve_pass(pass_name: str) -> type | None:
    """Resolve a pass class by name from the passes module."""
    from orchestrata_ultimate.refactor.codemod import passes

    cls = getattr(passes, pass_name, None)
    if cls is not None and isinstance(cls, type):
        resolved: type = cls
        return resolved
    return None