feat: improve local service reliability and test isolation

developer.md

@@ -174,7 +174,11 @@ uv run pytest -k "test_name"           # by name
 
 ### Self-bootstrapping test harnesses
 
-Today the E2E suite runs `Reflexio` in-process with `StorageConfigSQLite` configured against a `tmp_path` fixture (`tests/e2e_tests/conftest.py`), so tests neither bind production ports nor write to `~/.reflexio`.
+The test bootstrap redirects `REFLEXIO_LOG_DIR` and `LOCAL_STORAGE_PATH` to a
+temporary directory before test collection uses storage. In-process E2E
+fixtures also configure `StorageConfigSQLite` against a `tmp_path` fixture
+(`tests/e2e_tests/conftest.py`). Together these guards keep tests from binding
+production ports or writing to `~/.reflexio`.
 
 If you add a future harness that boots services from a clean checkout, it must:
 

reflexio/cli/utils.py

@@ -13,8 +13,9 @@
 import tempfile
 import threading
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from pathlib import Path
+from typing import Protocol
 
 from reflexio.cli.log_format import format_service_line
 
@@ -182,6 +183,11 @@ def remove_pidfile(pidfile: Path) -> None:
         pidfile.unlink(missing_ok=True)
 
 
+def get_stop_request_path(service_name: str, port: int) -> Path:
+    """Get the marker path used to communicate an intentional child stop."""
+    return Path(tempfile.gettempdir()) / f"reflexio_service_{service_name}_{port}.stop"
+
+
 # Patterns that indicate a service is ready to accept requests
 _READY_PATTERNS: dict[str, list[str]] = {
     "backend": ["Application startup complete"],
@@ -198,6 +204,18 @@ def remove_pidfile(pidfile: Path) -> None:
     "docs": {"NODE_NO_WARNINGS": "1"},
 }
 
+# Keep supervisor tuning internal until operational evidence justifies exposing it.
+_SERVICE_READY_TIMEOUT_SECS = 60.0
+_SERVICE_READY_POLL_INTERVAL_SECS = 0.1
+_SERVICE_MONITOR_POLL_INTERVAL_SECS = 0.5
+_SERVICE_HEALTHY_WINDOW_SECS = 30.0
+_SERVICE_MAX_RAPID_FAILURES = 5
+_SERVICE_RESPAWN_DELAY_SECS = 2.0
+
+
+class _PollableProcess(Protocol):
+    def poll(self) -> int | None: ...
+
 
 def _stream_output(
     proc: subprocess.Popen[bytes],
@@ -228,6 +246,162 @@ def _stream_output(
             ready_event.set()
 
 
+def _wait_for_all_ready(
+    ready_events: dict[str, threading.Event],
+    processes: Mapping[str, _PollableProcess],
+) -> bool:
+    """Wait for readiness, returning early when a child exits or times out."""
+    deadline = time.monotonic() + _SERVICE_READY_TIMEOUT_SECS
+    while time.monotonic() < deadline:
+        if any(proc.poll() is not None for proc in processes.values()):
+            return False
+        if all(event.is_set() for event in ready_events.values()):
+            return True
+        remaining = deadline - time.monotonic()
+        time.sleep(min(_SERVICE_READY_POLL_INTERVAL_SECS, remaining))
+    return all(event.is_set() for event in ready_events.values())
+
+
+@dataclasses.dataclass
+class _ServiceSupervisor:
+    service_configs: dict[str, ServiceConfig]
+    pidfile: Path
+    stop_request_paths: dict[str, Path]
+    processes: dict[str, subprocess.Popen[bytes]] = dataclasses.field(
+        default_factory=dict
+    )
+    threads: dict[str, threading.Thread] = dataclasses.field(default_factory=dict)
+    ready_events: dict[str, threading.Event] = dataclasses.field(default_factory=dict)
+    recent_failures: dict[str, list[float]] = dataclasses.field(default_factory=dict)
+    pending_respawns: dict[str, float] = dataclasses.field(default_factory=dict)
+    output_lock: threading.Lock = dataclasses.field(default_factory=threading.Lock)
+    shutting_down: bool = False
+
+    def write_output(self, message: str) -> None:
+        with self.output_lock:
+            sys.stdout.write(message + "\n")
+            sys.stdout.flush()
+
+    def write_current_pidfile(self) -> None:
+        write_pidfile(
+            self.pidfile, {name: proc.pid for name, proc in self.processes.items()}
+        )
+
+    def start_service(self, svc: ServiceConfig, *, respawn: bool = False) -> None:
+        noise_env = _NOISE_SUPPRESSION_ENV.get(svc.name, {})
+        merged_env = {**os.environ, **(svc.env or {}), **noise_env}
+        action = "Respawning" if respawn else "Starting"
+        self.write_output(f"{action} {svc.name}...")
+        proc = subprocess.Popen(
+            svc.command,
+            cwd=svc.cwd,
+            env=merged_env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
+        try:
+            self.processes[svc.name] = proc
+            ready_event = threading.Event()
+            self.ready_events[svc.name] = ready_event
+            thread = threading.Thread(
+                target=_stream_output,
+                args=(proc, svc.name, self.output_lock, ready_event),
+                daemon=True,
+            )
+            thread.start()
+            self.threads[svc.name] = thread
+        except (OSError, RuntimeError):
+            self.processes.pop(svc.name, None)
+            self.ready_events.pop(svc.name, None)
+            with contextlib.suppress(OSError):
+                proc.terminate()
+            raise
+        self.write_output(f"  {svc.name} started (PID {proc.pid})")
+
+    def stop_started_services(self) -> None:
+        for proc in self.processes.values():
+            with contextlib.suppress(OSError):
+                proc.terminate()
+        deadline = time.time() + 3.0
+        for proc in self.processes.values():
+            remaining = max(0, deadline - time.time())
+            try:
+                proc.wait(timeout=remaining)
+            except subprocess.TimeoutExpired:
+                with contextlib.suppress(OSError):
+                    proc.kill()
+        for thread in self.threads.values():
+            thread.join(timeout=1.0)
+
+    def shutdown(self, _signum: int | None = None, _frame: object = None) -> None:
+        self.shutting_down = True
+        self.write_output("\nShutting down services...")
+        self.stop_started_services()
+        remove_pidfile(self.pidfile)
+        for path in self.stop_request_paths.values():
+            remove_pidfile(path)
+        sys.exit(0)
+
+    def handle_service_exit(self, name: str, ret: int) -> None:
+        self.write_output(format_service_line(name, f"exited with code {ret}"))
+        del self.processes[name]
+        self.threads.pop(name, None)
+        self.ready_events.pop(name, None)
+        self.write_current_pidfile()
+
+        stop_requested = self.stop_request_paths[name].exists()
+        self.pending_respawns.pop(name, None)
+        if ret == 0 or self.shutting_down or stop_requested:
+            self.recent_failures.pop(name, None)
+            remove_pidfile(self.stop_request_paths[name])
+            return
+
+        now = time.monotonic()
+        failures = [
+            timestamp
+            for timestamp in self.recent_failures.get(name, [])
+            if now - timestamp <= _SERVICE_HEALTHY_WINDOW_SECS
+        ]
+        failures.append(now)
+        self.recent_failures[name] = failures
+        if len(failures) >= _SERVICE_MAX_RAPID_FAILURES:
+            self.write_output(
+                format_service_line(
+                    name,
+                    f"marked degraded after {len(failures)} "
+                    "rapid failures; will not respawn",
+                )
+            )
+            return
+
+        self.pending_respawns[name] = now + _SERVICE_RESPAWN_DELAY_SECS
+
+    def respawn_due_services(self) -> None:
+        if self.shutting_down:
+            self.pending_respawns.clear()
+            return
+        now = time.monotonic()
+        for name, respawn_at in list(self.pending_respawns.items()):
+            if respawn_at > now:
+                continue
+            self.pending_respawns.pop(name)
+            if self.stop_request_paths[name].exists():
+                remove_pidfile(self.stop_request_paths[name])
+                continue
+            try:
+                self.start_service(self.service_configs[name], respawn=True)
+                self.write_current_pidfile()
+            except (OSError, RuntimeError) as exc:
+                self.ready_events.pop(name, None)
+                proc = self.processes.pop(name, None)
+                if proc is not None:
+                    with contextlib.suppress(OSError):
+                        proc.terminate()
+                self.write_output(
+                    format_service_line(name, f"respawn failed; marked degraded: {exc}")
+                )
+
+
 def run_services(
     services: list[ServiceConfig],
     ports: dict[str, int],
@@ -237,104 +411,66 @@ def run_services(
     """Launch services, pipe output with prefixes, and manage lifecycle.
 
     Each service's stdout/stderr is captured and prefixed with a colored
-    service tag (e.g., [backend]). A callback is invoked when all services
-    report ready (or after a timeout).
+    service tag (e.g., [backend]). Unexpected exits are respawned after a
+    bounded delay; failure history expires outside the healthy window, and a
+    service is marked degraded after reaching the rapid-failure limit. Clean
+    exits, shutdowns, and explicitly requested stops are not respawned. A
+    callback is invoked when all services report ready (or after a timeout).
 
     Args:
         services: List of service configurations to launch.
         ports: Port mapping for pidfile identification.
         on_all_ready: Optional callback invoked with ports when all services are ready.
     """
-    processes: dict[str, subprocess.Popen[bytes]] = {}
-    threads: list[threading.Thread] = []
-    ready_events: dict[str, threading.Event] = {}
-    output_lock = threading.RLock()
     pidfile = get_pidfile_path(ports)
+    supervisor = _ServiceSupervisor(
+        service_configs={svc.name: svc for svc in services},
+        pidfile=pidfile,
+        stop_request_paths={
+            name: get_stop_request_path(name, ports[name]) for name in ports
+        },
+    )
 
-    def shutdown(_signum: int | None = None, _frame: object = None) -> None:
-        with output_lock:
-            sys.stdout.write("\nShutting down services...\n")
-            sys.stdout.flush()
-        for proc in processes.values():
-            with contextlib.suppress(OSError):
-                proc.terminate()
-        deadline = time.time() + 3.0
-        for proc in processes.values():
-            remaining = max(0, deadline - time.time())
-            try:
-                proc.wait(timeout=remaining)
-            except subprocess.TimeoutExpired:
-                with contextlib.suppress(OSError):
-                    proc.kill()
-        for t in threads:
-            t.join(timeout=1.0)
-        remove_pidfile(pidfile)
-        sys.exit(0)
-
-    signal.signal(signal.SIGINT, shutdown)
-    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, supervisor.shutdown)
+    signal.signal(signal.SIGTERM, supervisor.shutdown)
 
     ensure_requested_ports_available(ports)
+    for path in supervisor.stop_request_paths.values():
+        remove_pidfile(path)
 
-    for svc in services:
-        noise_env = _NOISE_SUPPRESSION_ENV.get(svc.name, {})
-        merged_env = {**os.environ, **(svc.env or {}), **noise_env}
-        with output_lock:
-            sys.stdout.write(f"Starting {svc.name}...\n")
-            sys.stdout.flush()
-        proc = subprocess.Popen(
-            svc.command,
-            cwd=svc.cwd,
-            env=merged_env,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-        )
-        processes[svc.name] = proc
-        ready_event = threading.Event()
-        ready_events[svc.name] = ready_event
-        t = threading.Thread(
-            target=_stream_output,
-            args=(proc, svc.name, output_lock, ready_event),
-            daemon=True,
-        )
-        t.start()
-        threads.append(t)
-        with output_lock:
-            sys.stdout.write(f"  {svc.name} started (PID {proc.pid})\n")
-            sys.stdout.flush()
-
-    # Write pidfile
-    pids = {name: proc.pid for name, proc in processes.items()}
-    write_pidfile(pidfile, pids)
+    try:
+        for svc in services:
+            supervisor.start_service(svc)
+        supervisor.write_current_pidfile()
+    except (OSError, RuntimeError):
+        supervisor.stop_started_services()
+        remove_pidfile(pidfile)
+        for path in supervisor.stop_request_paths.values():
+            remove_pidfile(path)
+        raise
 
     # Wait for all services to be ready (or timeout with shared deadline)
-    if on_all_ready:
-        deadline = time.monotonic() + 60
-        all_ready = all(
-            evt.wait(timeout=max(0, deadline - time.monotonic()))
-            for evt in ready_events.values()
-        )
-        if all_ready:
-            on_all_ready(ports)
+    if on_all_ready and _wait_for_all_ready(
+        supervisor.ready_events, supervisor.processes
+    ):
+        on_all_ready(ports)
 
     # Wait for any child to exit
     try:
-        while processes:
-            for name, proc in list(processes.items()):
+        while supervisor.processes or supervisor.pending_respawns:
+            supervisor.respawn_due_services()
+            for name, proc in list(supervisor.processes.items()):
                 ret = proc.poll()
                 if ret is not None:
-                    with output_lock:
-                        sys.stdout.write(
-                            format_service_line(name, f"exited with code {ret}") + "\n"
-                        )
-                        sys.stdout.flush()
-                    del processes[name]
-            if processes:
-                time.sleep(0.5)
+                    supervisor.handle_service_exit(name, ret)
+            if supervisor.processes or supervisor.pending_respawns:
+                time.sleep(_SERVICE_MONITOR_POLL_INTERVAL_SECS)
     except KeyboardInterrupt:
-        shutdown()
+        supervisor.shutdown()
 
     remove_pidfile(pidfile)
+    for path in supervisor.stop_request_paths.values():
+        remove_pidfile(path)
 
 
 def stop_services(
@@ -353,6 +489,8 @@ def stop_services(
     saved_pids = read_pidfile(pidfile)
 
     for name, port in port_map.items():
+        stop_request_path = get_stop_request_path(name, port)
+        stop_request_path.touch()
         pids_from_port = find_pids_on_port(port)
         pids_from_pattern = (
             find_pids_by_pattern(process_patterns[name])