refactor(logging): consolidate job status display into remote backend

- Removed the log.py file and integrated the job status display functionality directly into the remote backend.
- Enhanced the status display with support for terminal and notebook environments, including color-coded updates and spinners.
- Updated response handling to utilize the new status display system for improved user feedback during job execution.
- Streamlined the logging process by eliminating the previous remote logger implementation.

Author: JadenFiotto-Kaufman

src/nnsight/intervention/backends/remote.py

@@ -3,6 +3,7 @@
 import inspect
 import io
 import os
+import sys
 import time
 from sys import version as python_version
 from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple
@@ -22,6 +23,244 @@
 from .base import Backend
 
 
+def _supports_color():
+    """Check if the terminal supports color output."""
+    if os.environ.get("NO_COLOR"):
+        return False
+    if os.environ.get("FORCE_COLOR"):
+        return True
+    # IPython/Jupyter notebooks support ANSI colors
+    if __IPYTHON__:
+        return True
+    if not hasattr(sys.stdout, "isatty") or not sys.stdout.isatty():
+        return False
+    return True
+
+
+_SUPPORTS_COLOR = _supports_color()
+
+
+class JobStatusDisplay:
+    """Manages single-line status display for remote job execution."""
+
+    # ANSI color codes
+    class Colors:
+        RESET = "\033[0m" if _SUPPORTS_COLOR else ""
+        BOLD = "\033[1m" if _SUPPORTS_COLOR else ""
+        DIM = "\033[2m" if _SUPPORTS_COLOR else ""
+        CYAN = "\033[36m" if _SUPPORTS_COLOR else ""
+        YELLOW = "\033[33m" if _SUPPORTS_COLOR else ""
+        GREEN = "\033[32m" if _SUPPORTS_COLOR else ""
+        RED = "\033[31m" if _SUPPORTS_COLOR else ""
+        MAGENTA = "\033[35m" if _SUPPORTS_COLOR else ""
+        BLUE = "\033[34m" if _SUPPORTS_COLOR else ""
+        WHITE = "\033[37m" if _SUPPORTS_COLOR else ""
+
+    # Status icons (Unicode)
+    class Icons:
+        RECEIVED = "◉"
+        QUEUED = "◎"
+        DISPATCHED = "◈"
+        RUNNING = "●"
+        COMPLETED = "✓"
+        ERROR = "✗"
+        LOG = "ℹ"
+        STREAM = "⇄"
+        SPINNER = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
+
+    def __init__(self, enabled: bool = True, verbose: bool = False):
+        self.enabled = enabled
+        self.verbose = verbose
+        self.status_start_time: Optional[float] = None
+        self.job_id: Optional[str] = None
+        self.spinner_idx = 0
+        self.last_status = None
+        self._line_written = False
+        self._notebook_display_id: Optional[str] = None
+
+    def _format_elapsed(self) -> str:
+        """Format elapsed time in current status."""
+        if self.status_start_time is None:
+            return "0.0s"
+        elapsed = time.time() - self.status_start_time
+        if elapsed < 60:
+            return f"{elapsed:.1f}s"
+        elif elapsed < 3600:
+            mins = int(elapsed // 60)
+            secs = elapsed % 60
+            return f"{mins}m {secs:.0f}s"
+        else:
+            hours = int(elapsed // 3600)
+            mins = int((elapsed % 3600) // 60)
+            return f"{hours}h {mins}m"
+
+    def _get_status_style(self, status_name: str) -> tuple:
+        """Get icon and color for a status."""
+        status_map = {
+            "RECEIVED": (self.Icons.RECEIVED, self.Colors.CYAN),
+            "QUEUED": (self.Icons.QUEUED, self.Colors.YELLOW),
+            "DISPATCHED": (self.Icons.DISPATCHED, self.Colors.MAGENTA),
+            "RUNNING": (self.Icons.RUNNING, self.Colors.BLUE),
+            "COMPLETED": (self.Icons.COMPLETED, self.Colors.GREEN),
+            "ERROR": (self.Icons.ERROR, self.Colors.RED),
+            "NNSIGHT_ERROR": (self.Icons.ERROR, self.Colors.RED),
+            "LOG": (self.Icons.LOG, self.Colors.DIM),
+            "STREAM": (self.Icons.STREAM, self.Colors.CYAN),
+        }
+        return status_map.get(status_name, ("•", self.Colors.WHITE))
+
+    def _get_spinner(self) -> str:
+        """Get next spinner frame."""
+        spinner = self.Icons.SPINNER[self.spinner_idx % len(self.Icons.SPINNER)]
+        self.spinner_idx += 1
+        return spinner
+
+    def update(self, job_id: str, status_name: str, description: str = ""):
+        """Update the status display on a single line."""
+        if not self.enabled:
+            return
+
+        status_changed = status_name != self.last_status
+
+        # Reset timer when status changes
+        if status_changed:
+            self.status_start_time = time.time()
+            self.job_id = job_id
+
+        icon, color = self._get_status_style(status_name)
+        elapsed = self._format_elapsed()
+
+        # Build the status line
+        # Format: ● STATUS (elapsed) [job_id] description
+
+        is_terminal = status_name in ("COMPLETED", "ERROR", "NNSIGHT_ERROR")
+        is_active = status_name in ("QUEUED", "RUNNING", "DISPATCHED")
+
+        # For active states, show spinner
+        if is_active:
+            prefix = f"{self.Colors.DIM}{self._get_spinner()}{self.Colors.RESET}"
+        else:
+            prefix = f"{color}{icon}{self.Colors.RESET}"
+
+        # Build status text - full job ID shown so users can reference it
+        status_text = (
+            f"{prefix} "
+            f"{self.Colors.DIM}[{job_id}]{self.Colors.RESET} "
+            f"{color}{self.Colors.BOLD}{status_name.ljust(10)}{self.Colors.RESET} "
+            f"{self.Colors.DIM}({elapsed}){self.Colors.RESET}"
+        )
+
+        if description:
+            status_text += f" {self.Colors.DIM}{description}{self.Colors.RESET}"
+
+        # Display the status
+        self._display(status_text, status_changed, is_terminal)
+
+        self._line_written = True
+        self.last_status = status_name
+
+    def _display(self, text: str, status_changed: bool, is_terminal: bool):
+        """Display text, handling terminal vs notebook environments."""
+        if __IPYTHON__:
+            self._display_notebook(text, status_changed, is_terminal)
+        else:
+            self._display_terminal(text, status_changed, is_terminal)
+
+    def _display_terminal(self, text: str, status_changed: bool, is_terminal: bool):
+        """Display in terminal with in-place updates."""
+        # In verbose mode, print new line when status changes
+        if self.verbose and status_changed and self._line_written:
+            sys.stdout.write("\n")
+        else:
+            # Clear current line for in-place update
+            sys.stdout.write("\r\033[K")
+
+        sys.stdout.write(text)
+
+        if is_terminal:
+            sys.stdout.write("\n")
+
+        sys.stdout.flush()
+
+    def _ansi_to_html(self, text: str) -> str:
+        """Convert ANSI color codes to HTML spans."""
+        import re
+
+        # Map ANSI codes to CSS styles
+        ansi_to_css = {
+            "0": "",  # Reset
+            "1": "font-weight:bold",  # Bold
+            "2": "opacity:0.7",  # Dim
+            "31": "color:#e74c3c",  # Red
+            "32": "color:#2ecc71",  # Green
+            "33": "color:#f39c12",  # Yellow
+            "34": "color:#3498db",  # Blue
+            "35": "color:#9b59b6",  # Magenta
+            "36": "color:#00bcd4",  # Cyan
+            "37": "color:#ecf0f1",  # White
+        }
+
+        result = []
+        open_spans = 0
+        i = 0
+
+        while i < len(text):
+            # Match ANSI escape sequence
+            match = re.match(r"\x1b\[([0-9;]+)m", text[i:])
+            if match:
+                codes = match.group(1).split(";")
+                for code in codes:
+                    if code == "0":
+                        # Close all open spans
+                        result.append("</span>" * open_spans)
+                        open_spans = 0
+                    elif code in ansi_to_css and ansi_to_css[code]:
+                        result.append(f'<span style="{ansi_to_css[code]}">')
+                        open_spans += 1
+                i += len(match.group(0))
+            else:
+                # Escape HTML special chars
+                char = text[i]
+                if char == "<":
+                    result.append("&lt;")
+                elif char == ">":
+                    result.append("&gt;")
+                elif char == "&":
+                    result.append("&amp;")
+                else:
+                    result.append(char)
+                i += 1
+
+        # Close any remaining spans
+        result.append("</span>" * open_spans)
+        return "".join(result)
+
+    def _display_notebook(self, text: str, status_changed: bool, is_terminal: bool):
+        """Display in notebook using display_id for flicker-free updates."""
+        from IPython.display import display, update_display, HTML
+
+        html_text = self._ansi_to_html(text)
+        html_content = HTML(
+            f'<pre style="margin:0;font-family:monospace;background:transparent;">{html_text}</pre>'
+        )
+
+        if self.verbose and status_changed and self._line_written:
+            # Verbose mode: create new display for new status, keep old one visible
+            self._notebook_display_id = f"nnsight_status_{id(self)}_{time.time()}"
+            display(html_content, display_id=self._notebook_display_id)
+        elif self._notebook_display_id is None:
+            # First display
+            self._notebook_display_id = f"nnsight_status_{id(self)}"
+            display(html_content, display_id=self._notebook_display_id)
+        else:
+            # Update existing display in place (no flicker)
+            update_display(html_content, display_id=self._notebook_display_id)
+
+        if is_terminal:
+            # Reset for next job
+            self._notebook_display_id = None
+
+
 class RemoteException(Exception):
     pass
 
@@ -44,6 +283,7 @@ def __init__(
         job_id: str = None,
         api_key: str = "",
         callback: str = "",
+        verbose: bool = False,
     ) -> None:
 
         self.model_key = model_key
@@ -65,6 +305,10 @@ def __init__(
             self.ws_address = "ws://" + self.address[7:]
 
         self.job_status = None
+        self.status_display = JobStatusDisplay(
+            enabled=CONFIG.APP.REMOTE_LOGGING,
+            verbose=verbose,
+        )
 
     def request(self, tracer: Tracer) -> Tuple[bytes, Dict[str, str]]:
 
@@ -124,11 +368,16 @@ def handle_response(
         self.job_status = response.status
 
         if response.status == ResponseModel.JobStatus.ERROR:
+            self.status_display.update(
+                response.id, response.status.name, response.description or ""
+            )
             raise RemoteException(f"{response.description}\nRemote exception.")
 
-        # Log response for user
-        response.log()
-        self.job_status = response.status
+        # Log response for user (skip STREAM status - it's internal)
+        if response.status != ResponseModel.JobStatus.STREAM:
+            self.status_display.update(
+                response.id, response.status.name, response.description or ""
+            )
 
         # If job is completed:
         if response.status == ResponseModel.JobStatus.COMPLETED:
@@ -310,8 +559,18 @@ def blocking_request(self, tracer: Tracer) -> Optional[RESULT]:
                 # Loop until
                 while True:
 
-                    # Get pickled bytes value from the websocket.
-                    response = sio.receive()[1]
+                    # Use timeout only when remote logging is enabled to update spinner/elapsed time
+                    timeout = 0.1 if CONFIG.APP.REMOTE_LOGGING else None
+                    try:
+                        response = sio.receive(timeout=timeout)[1]
+                    except socketio.exceptions.TimeoutError:
+                        # Refresh the status display to update spinner and elapsed time
+                        if self.job_id and self.job_status:
+                            self.status_display.update(
+                                self.job_id, self.job_status.name
+                            )
+                        continue
+
                     # Convert to pydantic object.
                     response = ResponseModel.unpickle(response)
                     # Handle the response.
@@ -365,8 +624,18 @@ async def async_request(self, tracer: Tracer) -> Optional[RESULT]:
                 # Loop until
                 while True:
 
-                    # Get pickled bytes value from the websocket.
-                    response = (await sio.receive())[1]
+                    # Use timeout only when remote logging is enabled to update spinner/elapsed time
+                    timeout = 0.1 if CONFIG.APP.REMOTE_LOGGING else None
+                    try:
+                        response = (await sio.receive(timeout=timeout))[1]
+                    except socketio.exceptions.TimeoutError:
+                        # Refresh the status display to update spinner and elapsed time
+                        if self.job_id and self.job_status:
+                            self.status_display.update(
+                                self.job_id, self.job_status.name
+                            )
+                        continue
+
                     # Convert to pydantic object.
                     response = ResponseModel.unpickle(response)
                     # Handle the response.