orchestrata_ultimate.refactor.codemod.client

Reference › orchestrata_ultimate › refactor › codemod › client

The JSON-RPC 2.0 client for the codemod daemon. scripts.codemod is its only importer; it talks to the server over the Unix socket that the daemon manages, and decodes commit results into the CommitReport type from report.

CodemodClient.connect(socket_path) opens the socket and raises ConnectionError if no daemon answers; the class is a context manager that closes the connection. Each public method wraps one RPC call through _call, which serializes requests under a lock and reads the matching response; an error reply raises CodemodRpcError with the code and message. The graph queries are query_dependencies, query_dependents, query_blast_radius, query_cycles, query_topological_order, query_centrality, query_impact_score, query_coupling_score and query_abstraction_level, and resolve_nodes turns node ids into names. The write path is run_pass(pass_name, params, run_id, observe), which opens or extends a pending run, then commit(run_id, dry_run, baseline, observe), which returns a CommitReport. status and reload report on and refresh the daemon.

At a glance

Class diagram

Dependency diagram

API

Classes

CodemodRpcError

Bases: RuntimeError · Line 25

The daemon answered a request with a JSON-RPC error.

CodemodClient

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

JSON-RPC 2.0 client for the codemod daemon.

Source

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

"""SPEC-110 codemod engine — JSON-RPC 2.0 client for the codemod daemon.

Connects to the CodemodServer via Unix socket and exposes JSON-RPC methods
as Python functions. Supports context manager protocol.
"""

from __future__ import annotations

import json
import logging
import socket
import threading
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.report import CommitReport

logger = logging.getLogger(__name__)

JSONRPC_VERSION = "2.0"
RECV_BUFFER = 65536


class CodemodRpcError(RuntimeError):
    """The daemon answered a request with a JSON-RPC error."""

    def __init__(self, code: int, message: str) -> None:
        super().__init__(f"JSON-RPC error {code}: {message}")
        self.code = code
        self.rpc_message = message


@traces(spec="110", story="US-110-013", reqs=["REQ-110-139", "REQ-110-140"])
class CodemodClient:
    """JSON-RPC 2.0 client for the codemod daemon.

    Connects to the Unix socket created by CodemodServer and exposes
    workspace operations as Python methods. Supports context manager
    protocol for automatic connection cleanup.
    """

    def __init__(self) -> None:
        """Initialize CodemodClient with no active connection."""
        self._socket: socket.socket | None = None
        self._request_id = 0
        self._lock = threading.Lock()
        self._buffer = b""
        self._socket_path: Path | None = None

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-139"])
    def connect(self, socket_path: Path) -> CodemodClient:
        """Connect to the codemod daemon Unix socket.

        Args:
            socket_path: Path to the Unix domain socket.

        Returns:
            Self for chaining.

        Raises:
            ConnectionError: If the daemon is not running.
        """
        self._socket_path = socket_path
        self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            self._socket.connect(str(socket_path))
        except (ConnectionRefusedError, FileNotFoundError, OSError) as e:
            self._socket = None
            raise ConnectionError(
                f"Cannot connect to codemod daemon at {socket_path}: {e}"
            ) from e
        self._buffer = b""
        logger.info("Connected to codemod daemon at %s", socket_path)
        return self

    def close(self) -> None:
        """Close the connection to the daemon."""
        if self._socket is not None:
            try:
                self._socket.close()
            except OSError as e:
                logger.warning("Error closing client socket: %s", e)
            self._socket = None

    def __enter__(self) -> CodemodClient:
        """Context manager entry."""
        return self

    def __exit__(self, *args: object) -> None:
        """Context manager exit — close connection."""
        self.close()

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_dependencies(self, node_id: str) -> list[int]:
        """Query direct dependencies of a node.

        Args:
            node_id: Qualified name (CIG key) of the node.

        Returns:
            List of dependency node IDs. These are CIG integer ids, not keys:
            the server returns ``cig.dependencies(cig.resolve(node_id))``, and
            ``CodeIntelligenceGraph.dependencies`` is ``(int) -> list[int]``.
            The old ``list[str]`` annotation was wrong in both directions --
            the argument is a key, the result is ids.
        """
        result: list[int] = self._call("query_dependencies", {"node_id": node_id})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_dependents(self, node_id: str) -> list[int]:
        """Query direct dependents of a node.

        Args:
            node_id: Qualified name (CIG key) of the node.

        Returns:
            List of dependent node IDs -- CIG integer ids, for the same reason
            as :meth:`query_dependencies`.
        """
        result: list[int] = self._call("query_dependents", {"node_id": node_id})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_blast_radius(self, node_id: str, hops: int = 3) -> dict[str, Any]:
        """Query blast radius of a node (transitive dependents).

        Args:
            node_id: Qualified name or key of the node.
            hops: Maximum traversal depth.

        Returns:
            BlastRadius dict with affected_nodes, affected_files, impact_score.
        """
        # dict, not list[str]: the server hands back
        # {"affected_nodes": [...], "affected_files": [...], "impact_score": float}
        # (server.py's query_blast_radius branch). The declared list[str] was
        # wrong -- a caller iterating the result would have walked dict KEYS.
        # Unlike the sibling query_dependencies/query_dependents, which really
        # do return lists, this one is a record. Nothing called it, and its only
        # test asserted the raises-on-missing-parameter path, so the mismatch
        # was never exercised.
        result: dict[str, Any] = self._call(
            "query_blast_radius", {"node_id": node_id, "hops": hops}
        )
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_cycles(self) -> list[list[int]]:
        """Import cycles: each a list of CIG node ids."""
        result: list[list[int]] = self._call("query_cycles", {})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_topological_order(self) -> list[int]:
        """Node ids in dependency order (dependencies first)."""
        result: list[int] = self._call("query_topological_order", {})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_centrality(
        self, top_k: int = 20, exact: bool = False
    ) -> list[tuple[int, float]]:
        """Betweenness centrality: the ``top_k`` ``(node id, score)`` pairs."""
        raw = self._call("query_centrality", {"top_k": top_k, "exact": exact})
        return [(int(node_id), float(score)) for node_id, score in raw]

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_impact_score(self, node_id: str) -> float:
        """PageRank-based impact score of a node."""
        result: float = self._call("query_impact_score", {"node_id": node_id})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_coupling_score(self, node_id: str) -> dict[str, Any]:
        """Coupling of a node: ``efferent``, ``afferent``, ``instability``."""
        result: dict[str, Any] = self._call(
            "query_coupling_score", {"node_id": node_id}
        )
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def query_abstraction_level(self, node_id: str) -> dict[str, Any]:
        """Abstraction of a node: ``abstract_count``, ``concrete_count``, ``ratio``."""
        result: dict[str, Any] = self._call(
            "query_abstraction_level", {"node_id": node_id}
        )
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def resolve_nodes(self, node_ids: list[int]) -> list[dict[str, Any]]:
        """Names of CIG node ids: ``{"id", "qualified_name", "path"}`` records.

        Ids the graph does not hold are omitted from the reply.
        """
        result: list[dict[str, Any]] = self._call(
            "resolve_nodes", {"node_ids": list(node_ids)}
        )
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def run_pass(
        self,
        pass_name: str,
        params: dict,
        run_id: str | None = None,
        observe: bool = False,
    ) -> dict[str, Any]:
        """Apply a transform pass inside a pending run held by the daemon.

        The daemon keeps the mutated trees; ``commit`` names the run, so no
        ChangeSet crosses the socket. The first call opens the run (and the
        daemon's write lock); pass the returned ``run_id`` to stack more
        passes onto it.

        Args:
            pass_name: Name of the TransformPass class.
            params: Constructor parameters for the pass.
            run_id: The run to extend; ``None`` opens a new one.
            observe: Attach the REQ-110-026 observers to this pass.

        Returns:
            ``{"run_id", "file_count", "paths", "pass_name"}``; the counts
            cover the whole run so far.
        """
        request: dict[str, Any] = {
            "pass_name": pass_name,
            "params": params,
            "observe": observe,
        }
        if run_id is not None:
            request["run_id"] = run_id
        result: dict[str, Any] = self._call("run_pass", request)
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-138"])
    def commit(
        self,
        run_id: str,
        *,
        dry_run: bool = True,
        baseline: str = "delta",
        observe: bool = False,
    ) -> CommitReport:
        """Review and commit a pending run; the run ends either way.

        Args:
            run_id: The run ``run_pass`` returned.
            dry_run: If True, produce the report without writing; the daemon
                restores the trees the run mutated.
            baseline: The review engine's baseline (``delta`` or ``absolute``).
            observe: Attach the REQ-110-026 observers to this commit.

        Returns:
            CommitReport with success, files_written, findings, duration.
        """
        result: dict[str, Any] = self._call(
            "commit",
            {
                "changes": {"run_id": run_id},
                "dry_run": dry_run,
                "baseline": baseline,
                "observe": observe,
            },
        )
        return CommitReport.from_dict(result["report"])

    def status(self) -> dict:
        """Query daemon status.

        Returns:
            Dict with memory_rss, client_count, file_count.
        """
        result: dict = self._call("status", {})
        return result

    def reload(self) -> dict:
        """Force workspace reload (bypass inotify).

        Returns:
            Dict with reload status.
        """
        result: dict = self._call("reload", {})
        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-139"])
    def _call(self, method: str, params: dict) -> Any:
        """Send a JSON-RPC 2.0 request and return the result.

        Args:
            method: JSON-RPC method name.
            params: Method parameters.

        Returns:
            The 'result' field from the JSON-RPC response.

        Raises:
            ConnectionError: If not connected.
            RuntimeError: If the server returns an error.
        """
        if self._socket is None:
            raise ConnectionError("Not connected to codemod daemon")

        with self._lock:
            self._request_id += 1
            request_id = self._request_id

            request = {
                "jsonrpc": JSONRPC_VERSION,
                "method": method,
                "params": params,
                "id": request_id,
            }

            payload = json.dumps(request) + "\n"
            try:
                self._socket.sendall(payload.encode("utf-8"))
            except OSError as e:
                self._socket = None
                raise ConnectionError(f"Failed to send request: {e}") from e

            response = self._recv_response(request_id)

        if "error" in response:
            err = response["error"]
            raise CodemodRpcError(err.get("code", -1), err.get("message", "Unknown"))

        return response.get("result")

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-139"])
    def _recv_response(self, expected_id: int) -> dict:
        """Receive and parse a JSON-RPC response from the socket."""
        while True:
            if b"\n" in self._buffer:
                line, self._buffer = self._buffer.split(b"\n", 1)
                if line.strip():
                    try:
                        response: dict = json.loads(line.decode("utf-8"))
                        if response.get("id") == expected_id:
                            return response
                    except (json.JSONDecodeError, UnicodeDecodeError) as e:
                        logger.warning("Malformed response: %s", e)
                        continue

            if self._socket is None:
                raise ConnectionError("Socket closed while waiting for response")

            try:
                data = self._socket.recv(RECV_BUFFER)
            except OSError as e:
                self._socket = None
                raise ConnectionError(f"Failed to receive response: {e}") from e

            if not data:
                self._socket = None
                raise ConnectionError("Daemon closed connection")

            self._buffer += data