feat: ProcessWebService, service log viewing, and keep services alive

- Add ProcessWebService for subprocess-based services with atexit cleanup
- Add log viewing to global services tab (l key binding)
- Show non-STOPPED services (not just RUNNING) in global services
- Fix OR logic for log file existence checks in both service widgets
- Add stdout/stderr/log_directory delegation to SSHLocalService and MockService
- Stop SSH syncs on service failure in SSHLocalService
- Set local log directory for SSHMockService services
- Don't stop services on experiment exit (cleaned up on process exit)
- Update demo submodule with ProcessWebService TensorBoard

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bpiwowar

docs/demo

@@ -1015,11 +1015,6 @@ def __exit__(self, exc_type, exc_value, traceback):
             if self._register_signals:
                 SIGNAL_HANDLER.remove(self)
 
-            # Stop services
-            for service in self.services.values():
-                logger.info("Closing service %s", service.description())
-                service.stop()
-
             # Set end time for BaseExperiment interface
             self._ended_at = datetime.now()
 

src/experimaestro/scheduler/experiment.py

@@ -126,6 +126,18 @@ def state(self):
     def url(self) -> Optional[str]:
         return getattr(self._inner, "url", None)
 
+    @property
+    def log_directory(self) -> Optional[Path]:
+        return self._inner.log_directory
+
+    @property
+    def stdout(self) -> Optional[Path]:
+        return self._inner.stdout
+
+    @property
+    def stderr(self) -> Optional[Path]:
+        return self._inner.stderr
+
     def description(self) -> str:
         return self._inner.description()
 
@@ -163,8 +175,20 @@ def get_url(self) -> str:
         # Notify initial status
         self._notify_status_change()
 
-        # Start the inner service
-        return self._inner.get_url()
+        # Start the inner service; stop syncs if it fails
+        try:
+            url = self._inner.get_url()
+        except Exception:
+            self._stop_syncs()
+            raise
+
+        # Check if service ended up in error state (e.g. process exited)
+        from experimaestro.scheduler.services import ServiceState
+
+        if self._inner.state == ServiceState.ERROR:
+            self._stop_syncs()
+
+        return url
 
     @property
     def sync_status(self) -> Optional[str]:
@@ -189,16 +213,19 @@ def error(self) -> Optional[str]:
         """Return error message if service failed to start."""
         return self._inner.error
 
+    def _stop_syncs(self) -> None:
+        """Stop all adaptive synchronizers."""
+        for sync in self._synchronizers:
+            sync.stop()
+        self._synchronizers.clear()
+
     def stop(self) -> None:
         """Stop the inner service and stop all syncs."""
         # Stop the inner service first
         if hasattr(self._inner, "stop"):
             self._inner.stop()
 
-        # Stop all synchronizers
-        for sync in self._synchronizers:
-            sync.stop()
-        self._synchronizers.clear()
+        self._stop_syncs()
 
 
 class SSHMockService(MockService):
@@ -306,6 +333,11 @@ def path_translator(remote_path: str) -> Path:
             )
             inner_service.id = self.id
 
+            # Set local log directory for service output
+            local_cache = self._state_provider.local_cache_dir
+            if local_cache:
+                inner_service.log_directory = local_cache / "services" / self.id
+
             # Extract remote paths for sync management
             remote_paths = self._extract_paths()
 

src/experimaestro/scheduler/remote/client.py

@@ -1,4 +1,5 @@
 import abc
+import atexit
 from enum import Enum
 import logging
 import subprocess
@@ -57,6 +58,7 @@ def __init__(self):
         self._listeners: Set[ServiceListener] = set()
         self._listeners_lock = threading.Lock()
         self._error = None
+        self.log_directory: Optional[Path] = None
 
     def set_experiment(self, xp: "Experiment") -> None:
         """Called when the service is added to an experiment.
@@ -68,10 +70,9 @@ def set_experiment(self, xp: "Experiment") -> None:
             xp: The experiment this service is being added to.
         """
         self._experiment = xp
-
-        # Create log directory for this service
-        if self.log_directory:
-            self.log_directory.mkdir(parents=True, exist_ok=True)
+        self.log_directory = (
+            self._experiment.workspace.scheduler_services_path / self.id
+        )
 
     @property
     def experiment_id(self) -> str:
@@ -87,13 +88,6 @@ def run_id(self) -> str:
             return self._experiment.run_id or ""
         return ""
 
-    @property
-    def log_directory(self) -> Optional[Path]:
-        """Return the directory for service logs (None if not attached to experiment)"""
-        if self._experiment is None:
-            return None
-        return self._experiment.workspace.scheduler_services_path / self.id
-
     @property
     def stdout(self) -> Optional[Path]:
         """Return path to stdout log file"""
@@ -261,6 +255,10 @@ def setup_logging(
         if not self.stdout or not self.stderr:
             return None, None
 
+        # Ensure log directory exists
+        if self.log_directory:
+            self.log_directory.mkdir(parents=True, exist_ok=True)
+
         # Get logger for this service
         service_logger = logging.getLogger(f"xpm.service.{self.id}")
         service_logger.setLevel(logging.INFO)
@@ -622,18 +620,27 @@ def _start_process(self):
         """Start the service as a subprocess"""
         # Build command to run service
         cmd = self._build_command()
+        logger.info(
+            "Starting service %s (log_directory=%s): %s",
+            self.id,
+            self.log_directory,
+            " ".join(cmd),
+        )
 
-        # Redirect stdout/stderr to log files
+        # Ensure log directory exists and redirect stdout/stderr to log files
+        if self.log_directory:
+            self.log_directory.mkdir(parents=True, exist_ok=True)
         stdout_file = open(self.stdout, "w") if self.stdout else subprocess.DEVNULL
         stderr_file = open(self.stderr, "w") if self.stderr else subprocess.DEVNULL
 
-        # Start process
+        # Start process (ensure it gets killed when the parent exits)
         self.process = subprocess.Popen(
             cmd,
             stdout=stdout_file,
             stderr=stderr_file,
             cwd=str(self.log_directory) if self.log_directory else None,
         )
+        atexit.register(self._kill_process)
 
         # Monitor process in background thread
         monitor_thread = threading.Thread(
@@ -656,8 +663,9 @@ def _monitor_process(self):
             self.process.wait()
 
         except Exception as e:
-            logger.exception(f"Service {self.id} monitoring failed: {e}")
-            self.state = ServiceState.ERROR
+            logger.error("Service %s failed: %s", self.id, e)
+            self.set_error(str(e))
+
         finally:
             if running_event and not running_event.is_set():
                 running_event.set()
@@ -701,6 +709,16 @@ def stop(self, timeout: float = 2.0):
             running_event.wait()
 
         # Terminate process
+        self._kill_process(timeout=timeout)
+        atexit.unregister(self._kill_process)
+
+        with self._start_lock:
+            self.url = None
+            self._running_event = None
+            self.state = ServiceState.STOPPED
+
+    def _kill_process(self, timeout: float = 2.0):
+        """Kill the subprocess if it is still running."""
         if self.process and self.process.poll() is None:
             logger.info(f"Terminating service {self.id} (PID {self.process.pid})")
             self.process.terminate()
@@ -710,8 +728,3 @@ def stop(self, timeout: float = 2.0):
                 logger.warning(f"Service {self.id} did not terminate, killing")
                 self.process.kill()
                 self.process.wait()
-
-        with self._start_lock:
-            self.url = None
-            self._running_event = None
-            self.state = ServiceState.STOPPED

src/experimaestro/scheduler/services.py

@@ -1776,6 +1776,24 @@ def description(self) -> str:
         """Return service description"""
         return self._description
 
+    @property
+    def log_directory(self) -> Optional[Path]:
+        if self._live_service is not None:
+            return self._live_service.log_directory
+        return None
+
+    @property
+    def stdout(self) -> Optional[Path]:
+        if self._live_service is not None:
+            return self._live_service.stdout
+        return None
+
+    @property
+    def stderr(self) -> Optional[Path]:
+        if self._live_service is not None:
+            return self._live_service.stderr
+        return None
+
     @property
     def sync_status(self) -> Optional[str]:
         """Return sync status for display.

src/experimaestro/scheduler/state_provider.py

@@ -23,6 +23,7 @@ class GlobalServiceSyncs(Vertical):
 
     BINDINGS = [
         Binding("ctrl+k", "stop_service", "Stop Service"),
+        Binding("l", "view_logs", "View Logs", priority=True),
     ]
 
     def __init__(self, state_provider: StateProvider) -> None:
@@ -70,11 +71,11 @@ def _load_services(self) -> None:
             running_services = [
                 s
                 for s in all_services
-                if hasattr(s, "state") and s.state == ServiceState.RUNNING
+                if hasattr(s, "state") and s.state != ServiceState.STOPPED
             ]
             self.log.info(
                 f"GlobalServiceSyncs._load_services: got {len(running_services)} "
-                f"running services (out of {len(all_services)} total)"
+                f"active services (out of {len(all_services)} total)"
             )
         except Exception as e:
             logger.warning(f"Failed to load services: {e}")
@@ -228,3 +229,31 @@ def action_stop_service(self) -> None:
             self.refresh_services()
         except Exception as e:
             self.notify(f"Failed to stop service: {e}", severity="error")
+
+    def action_view_logs(self) -> None:
+        """View service logs"""
+        service = self._get_selected_service()
+        if not service:
+            return
+
+        # Convert to live service to get log paths
+        live_service = service.to_service()
+
+        if not live_service.stdout and not live_service.stderr:
+            self.notify("Service logs not available", severity="warning")
+            return
+
+        stdout_exists = live_service.stdout and live_service.stdout.exists()
+        stderr_exists = live_service.stderr and live_service.stderr.exists()
+        if not stdout_exists and not stderr_exists:
+            self.notify("No log files found", severity="warning")
+            return
+
+        from experimaestro.tui.log_viewer import create_service_log_viewer
+
+        sync_func = None
+        if self.state_provider.is_remote:
+            sync_func = self.state_provider.sync_path
+
+        viewer = create_service_log_viewer(live_service, sync_func)
+        self.app.push_screen(viewer)

src/experimaestro/tui/widgets/global_services.py

@@ -284,11 +284,13 @@ def action_view_logs(self) -> None:
         # Convert to live service to get log paths
         live_service = service.to_service()
 
-        if not live_service.stdout or not live_service.stderr:
+        if not live_service.stdout and not live_service.stderr:
             self.notify("Service logs not available", severity="warning")
             return
 
-        if not live_service.stdout.exists() and not live_service.stderr.exists():
+        stdout_exists = live_service.stdout and live_service.stdout.exists()
+        stderr_exists = live_service.stderr and live_service.stderr.exists()
+        if not stdout_exists and not stderr_exists:
             self.notify("No log files found", severity="warning")
             return