From 6cbebff223462811422ff2568646009ee0a83736 Mon Sep 17 00:00:00 2001 From: wenchanghan Date: Tue, 14 Jul 2026 23:50:11 -0700 Subject: [PATCH 1/5] feat: add bounded local service respawns --- .env.example | 11 ++ developer.md | 3 + reflexio/cli/utils.py | 194 +++++++++++++++++++++++++------- tests/cli/test_utils.py | 242 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 407 insertions(+), 43 deletions(-) diff --git a/.env.example b/.env.example index d2df15bb..2dbb10e6 100644 --- a/.env.example +++ b/.env.example @@ -104,3 +104,14 @@ MOCK_LLM_RESPONSE=false # preserved, so e.g. REFLEXIO_LOG_DIR=/tmp/foo writes logs to # /tmp/foo/.reflexio/logs/. (optional; default $HOME -> ~/.reflexio/logs/) REFLEXIO_LOG_DIR= + +# ============================================================================= +# 6. LOCAL SERVICE SUPERVISOR +# ============================================================================= +# Restart local services that exit unexpectedly while the parent is still alive. +# A service that fails repeatedly within the healthy window is marked degraded +# and left stopped so one crash loop cannot take down its siblings. +# (optional; defaults shown) +REFLEXIO_SERVICE_HEALTHY_SECS=30 +REFLEXIO_SERVICE_MAX_FAILS=5 +REFLEXIO_SERVICE_RESPAWN_DELAY=2 diff --git a/developer.md b/developer.md index c2194513..da0c3eb8 100644 --- a/developer.md +++ b/developer.md @@ -102,6 +102,9 @@ Copy `.env.example` to `.env` and fill in values. Key variables: - **Storage**: `LOCAL_STORAGE_PATH` (defaults to `~/.reflexio/data`) — houses the SQLite DB file. (Not used by `supabase`/`postgres`, which use external connections.) - **Storage backend**: `REFLEXIO_STORAGE` — `sqlite` (default), `supabase`, or `postgres`. Selects the data storage backend independently from auth configuration. - **Testing**: `IS_TEST_ENV`, `DEBUG_LOG_TO_CONSOLE`, `MOCK_LLM_RESPONSE` +- **Local service supervisor**: `REFLEXIO_SERVICE_HEALTHY_SECS` (default 30), + `REFLEXIO_SERVICE_MAX_FAILS` (default 5), and + `REFLEXIO_SERVICE_RESPAWN_DELAY` (default 2 seconds) Never change env variable values in `.env` directly for port overrides — use shell exports instead. diff --git a/reflexio/cli/utils.py b/reflexio/cli/utils.py index 1b6d7c15..b034f470 100644 --- a/reflexio/cli/utils.py +++ b/reflexio/cli/utils.py @@ -6,6 +6,7 @@ import dataclasses import hashlib import json +import math import os import signal import subprocess @@ -17,6 +18,7 @@ from pathlib import Path from reflexio.cli.log_format import format_service_line +from reflexio.server.env_utils import env_str @dataclasses.dataclass @@ -182,6 +184,34 @@ def remove_pidfile(pidfile: Path) -> None: pidfile.unlink(missing_ok=True) +def _get_float_env(name: str, default: float, *, minimum: float) -> float: + """Read a bounded floating-point environment setting.""" + raw = env_str(name, str(default)) + try: + value = float(raw) + except ValueError: + print(f"Warning: invalid {name}={raw!r}, using default {default}") + return default + if not math.isfinite(value) or value < minimum: + print(f"Warning: invalid {name}={raw!r}, using default {default}") + return default + return value + + +def _get_int_env(name: str, default: int, *, minimum: int) -> int: + """Read a bounded integer environment setting.""" + raw = env_str(name, str(default)) + try: + value = int(raw) + except ValueError: + print(f"Warning: invalid {name}={raw!r}, using default {default}") + return default + if value < minimum: + print(f"Warning: invalid {name}={raw!r}, using default {default}") + return default + return value + + # Patterns that indicate a service is ready to accept requests _READY_PATTERNS: dict[str, list[str]] = { "backend": ["Application startup complete"], @@ -228,6 +258,23 @@ def _stream_output( ready_event.set() +def _wait_for_all_ready(ready_events: dict[str, threading.Event]) -> bool: + """Wait up to 60 seconds for every service to report readiness.""" + deadline = time.monotonic() + 60 + return all( + event.wait(timeout=max(0, deadline - time.monotonic())) + for event in ready_events.values() + ) + + +def _sleep_while_services_run( + processes: dict[str, subprocess.Popen[bytes]], +) -> None: + """Avoid a busy monitor loop while at least one child remains alive.""" + if processes: + time.sleep(0.5) + + def run_services( services: list[ServiceConfig], ports: dict[str, int], @@ -248,10 +295,56 @@ def run_services( processes: dict[str, subprocess.Popen[bytes]] = {} threads: list[threading.Thread] = [] ready_events: dict[str, threading.Event] = {} + service_configs = {svc.name: svc for svc in services} + recent_failures: dict[str, list[float]] = {} output_lock = threading.RLock() pidfile = get_pidfile_path(ports) + healthy_secs = _get_float_env("REFLEXIO_SERVICE_HEALTHY_SECS", 30.0, minimum=0.001) + max_fails = _get_int_env("REFLEXIO_SERVICE_MAX_FAILS", 5, minimum=1) + respawn_delay = _get_float_env("REFLEXIO_SERVICE_RESPAWN_DELAY", 2.0, minimum=0.0) + shutting_down = False + + def write_current_pidfile() -> None: + write_pidfile(pidfile, {name: proc.pid for name, proc in processes.items()}) + + def start_service(svc: ServiceConfig, *, respawn: bool = False) -> None: + noise_env = _NOISE_SUPPRESSION_ENV.get(svc.name, {}) + merged_env = {**os.environ, **(svc.env or {}), **noise_env} + with output_lock: + action = "Respawning" if respawn else "Starting" + sys.stdout.write(f"{action} {svc.name}...\n") + sys.stdout.flush() + proc = subprocess.Popen( + svc.command, + cwd=svc.cwd, + env=merged_env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + try: + 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) + except (OSError, RuntimeError): + processes.pop(svc.name, None) + ready_events.pop(svc.name, None) + with contextlib.suppress(OSError): + proc.terminate() + raise + with output_lock: + sys.stdout.write(f" {svc.name} started (PID {proc.pid})\n") + sys.stdout.flush() def shutdown(_signum: int | None = None, _frame: object = None) -> None: + nonlocal shutting_down + shutting_down = True with output_lock: sys.stdout.write("\nShutting down services...\n") sys.stdout.flush() @@ -271,51 +364,72 @@ def shutdown(_signum: int | None = None, _frame: object = None) -> None: remove_pidfile(pidfile) sys.exit(0) + def handle_service_exit(name: str, ret: int) -> None: + with output_lock: + sys.stdout.write( + format_service_line(name, f"exited with code {ret}") + "\n" + ) + sys.stdout.flush() + del processes[name] + ready_events.pop(name, None) + write_current_pidfile() + + if ret == 0 or shutting_down: + recent_failures.pop(name, None) + return + + now = time.monotonic() + failures = [ + timestamp + for timestamp in recent_failures.get(name, []) + if now - timestamp <= healthy_secs + ] + failures.append(now) + recent_failures[name] = failures + if len(failures) >= max_fails: + with output_lock: + sys.stdout.write( + format_service_line( + name, + f"marked degraded after {len(failures)} " + "rapid failures; will not respawn", + ) + + "\n" + ) + sys.stdout.flush() + return + + time.sleep(respawn_delay) + if shutting_down: + return + try: + start_service(service_configs[name], respawn=True) + write_current_pidfile() + except (OSError, RuntimeError) as exc: + ready_events.pop(name, None) + with contextlib.suppress(KeyError, OSError): + processes.pop(name).terminate() + with output_lock: + sys.stdout.write( + format_service_line(name, f"respawn failed; marked degraded: {exc}") + + "\n" + ) + sys.stdout.flush() + signal.signal(signal.SIGINT, shutdown) signal.signal(signal.SIGTERM, shutdown) ensure_requested_ports_available(ports) 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() + start_service(svc) # Write pidfile - pids = {name: proc.pid for name, proc in processes.items()} - write_pidfile(pidfile, pids) + write_current_pidfile() # 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(ready_events): + on_all_ready(ports) # Wait for any child to exit try: @@ -323,14 +437,8 @@ def shutdown(_signum: int | None = None, _frame: object = None) -> None: for name, proc in list(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) + handle_service_exit(name, ret) + _sleep_while_services_run(processes) except KeyboardInterrupt: shutdown() diff --git a/tests/cli/test_utils.py b/tests/cli/test_utils.py index 3fea8245..ccef13e3 100644 --- a/tests/cli/test_utils.py +++ b/tests/cli/test_utils.py @@ -2,13 +2,58 @@ from __future__ import annotations +import io +import signal import tempfile from pathlib import Path from subprocess import CompletedProcess +import pytest + from reflexio.cli import utils +class _FakeProcess: + def __init__( + self, *, pid: int, polls_to_exit: int | None, returncode: int = 0 + ) -> None: + self.pid = pid + self.stdout = io.BytesIO() + self._polls_to_exit = polls_to_exit + self._returncode = returncode + self._poll_count = 0 + self.terminated = False + + def poll(self) -> int | None: + self._poll_count += 1 + if self._polls_to_exit is not None and self._poll_count >= self._polls_to_exit: + return self._returncode + return None + + def terminate(self) -> None: + self.terminated = True + + def wait(self, timeout: float | None = None) -> int: + return self._returncode + + def kill(self) -> None: + self.terminated = True + + +def _patch_run_services_environment( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> Path: + pidfile = tmp_path / "services.json" + monkeypatch.setattr(utils, "ensure_requested_ports_available", lambda _ports: None) + monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) + monkeypatch.setattr(utils.signal, "signal", lambda *_args: None) + monkeypatch.setattr(utils.time, "sleep", lambda _seconds: None) + monkeypatch.setenv("REFLEXIO_SERVICE_HEALTHY_SECS", "30") + monkeypatch.setenv("REFLEXIO_SERVICE_MAX_FAILS", "2") + monkeypatch.setenv("REFLEXIO_SERVICE_RESPAWN_DELAY", "0") + return pidfile + + def test_pidfile_path_uses_platform_temp_dir() -> None: path = utils.get_pidfile_path({"backend": 8071}) @@ -61,3 +106,200 @@ def test_requested_port_conflicts_detect_occupied_ports(monkeypatch) -> None: conflicts = utils.get_requested_port_conflicts({"frontend": 8090, "docs": 8092}) assert conflicts == ["frontend port 8090 is already in use by PID(s): 123, 456"] + + +@pytest.mark.unit +def test_run_services_respawns_unexpected_exit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Respawn a service after an unexpected non-zero exit.""" + _patch_run_services_environment(monkeypatch, tmp_path) + started: list[_FakeProcess] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess( + pid=1000 + len(started), + polls_to_exit=1, + returncode=1 if not started else 0, + ) + started.append(proc) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [utils.ServiceConfig(name="backend", command=["fake-backend"])], + {"backend": 8071}, + ) + + assert len(started) == 2 + assert started[0].pid != started[1].pid + + +@pytest.mark.unit +def test_run_services_gives_up_on_crash_loop_and_keeps_sibling_running( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Give up on one crash-looping service while supervising its sibling.""" + _patch_run_services_environment(monkeypatch, tmp_path) + started: list[tuple[str, _FakeProcess]] = [] + + def fake_popen(command, **_kwargs) -> _FakeProcess: + name = command[0] + proc = _FakeProcess( + pid=1000 + len(started), + polls_to_exit=1 if name == "crash" else 10, + returncode=1 if name == "crash" else 0, + ) + started.append((name, proc)) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [ + utils.ServiceConfig(name="crash", command=["crash"]), + utils.ServiceConfig(name="sibling", command=["sibling"]), + ], + {"crash": 8071, "sibling": 8072}, + ) + + assert [name for name, _proc in started] == ["crash", "sibling", "crash"] + sibling = next(proc for name, proc in started if name == "sibling") + assert sibling._poll_count == 10 + assert sibling.terminated is False + assert "marked degraded after 2 rapid failures" in capsys.readouterr().out + + +@pytest.mark.unit +def test_run_services_does_not_respawn_during_shutdown( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Do not restart children after the supervisor receives SIGTERM.""" + _patch_run_services_environment(monkeypatch, tmp_path) + handlers: dict[signal.Signals, object] = {} + started: list[_FakeProcess] = [] + + def capture_signal(signum, handler) -> None: + handlers[signum] = handler + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess(pid=1000 + len(started), polls_to_exit=None) + started.append(proc) + return proc + + def request_shutdown(_seconds: float) -> None: + handler = handlers[signal.SIGTERM] + assert callable(handler) + handler(signal.SIGTERM, None) + + monkeypatch.setattr(utils.signal, "signal", capture_signal) + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + monkeypatch.setattr(utils.time, "sleep", request_shutdown) + + with pytest.raises(SystemExit): + utils.run_services( + [utils.ServiceConfig(name="backend", command=["backend"])], + {"backend": 8071}, + ) + + assert len(started) == 1 + assert started[0].terminated is True + + +@pytest.mark.unit +def test_run_services_does_not_respawn_clean_exit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Treat a clean exit as intentional and do not restart the service.""" + _patch_run_services_environment(monkeypatch, tmp_path) + started: list[_FakeProcess] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess(pid=1000 + len(started), polls_to_exit=1, returncode=0) + started.append(proc) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [utils.ServiceConfig(name="backend", command=["backend"])], + {"backend": 8071}, + ) + + assert len(started) == 1 + + +@pytest.mark.unit +def test_run_services_rejects_nonfinite_supervisor_env_values( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Fall back safely when supervisor float settings are non-finite.""" + _patch_run_services_environment(monkeypatch, tmp_path) + monkeypatch.setenv("REFLEXIO_SERVICE_HEALTHY_SECS", "nan") + monkeypatch.setenv("REFLEXIO_SERVICE_RESPAWN_DELAY", "inf") + started: list[_FakeProcess] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess( + pid=1000 + len(started), + polls_to_exit=1, + returncode=1 if not started else 0, + ) + started.append(proc) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [utils.ServiceConfig(name="backend", command=["backend"])], + {"backend": 8071}, + ) + + output = capsys.readouterr().out + assert len(started) == 2 + assert "invalid REFLEXIO_SERVICE_HEALTHY_SECS='nan'" in output + assert "invalid REFLEXIO_SERVICE_RESPAWN_DELAY='inf'" in output + + +@pytest.mark.unit +def test_run_services_drops_failed_respawn_without_stopping_sibling( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Drop a child when respawn fails while keeping its sibling supervised.""" + _patch_run_services_environment(monkeypatch, tmp_path) + started: list[tuple[str, _FakeProcess]] = [] + popen_calls = 0 + + def fake_popen(command, **_kwargs) -> _FakeProcess: + nonlocal popen_calls + popen_calls += 1 + if popen_calls == 3: + raise OSError("simulated spawn failure") + name = command[0] + proc = _FakeProcess( + pid=1000 + len(started), + polls_to_exit=1 if name == "crash" else 3, + returncode=1 if name == "crash" else 0, + ) + started.append((name, proc)) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [ + utils.ServiceConfig(name="crash", command=["crash"]), + utils.ServiceConfig(name="sibling", command=["sibling"]), + ], + {"crash": 8071, "sibling": 8072}, + ) + + assert popen_calls == 3 + assert [name for name, _proc in started] == ["crash", "sibling"] + sibling = next(proc for name, proc in started if name == "sibling") + assert sibling._poll_count == 3 + assert sibling.terminated is False + assert "respawn failed; marked degraded: simulated spawn failure" in ( + capsys.readouterr().out + ) From e621f64842f813f6471b3e33f6062115f80e7dab Mon Sep 17 00:00:00 2001 From: wenchanghan Date: Wed, 15 Jul 2026 00:07:31 -0700 Subject: [PATCH 2/5] refactor: keep service supervisor policy internal --- .env.example | 11 ---------- developer.md | 3 --- reflexio/cli/utils.py | 44 +++++++-------------------------------- tests/cli/test_utils.py | 46 ++++++++--------------------------------- 4 files changed, 17 insertions(+), 87 deletions(-) diff --git a/.env.example b/.env.example index 2dbb10e6..d2df15bb 100644 --- a/.env.example +++ b/.env.example @@ -104,14 +104,3 @@ MOCK_LLM_RESPONSE=false # preserved, so e.g. REFLEXIO_LOG_DIR=/tmp/foo writes logs to # /tmp/foo/.reflexio/logs/. (optional; default $HOME -> ~/.reflexio/logs/) REFLEXIO_LOG_DIR= - -# ============================================================================= -# 6. LOCAL SERVICE SUPERVISOR -# ============================================================================= -# Restart local services that exit unexpectedly while the parent is still alive. -# A service that fails repeatedly within the healthy window is marked degraded -# and left stopped so one crash loop cannot take down its siblings. -# (optional; defaults shown) -REFLEXIO_SERVICE_HEALTHY_SECS=30 -REFLEXIO_SERVICE_MAX_FAILS=5 -REFLEXIO_SERVICE_RESPAWN_DELAY=2 diff --git a/developer.md b/developer.md index da0c3eb8..c2194513 100644 --- a/developer.md +++ b/developer.md @@ -102,9 +102,6 @@ Copy `.env.example` to `.env` and fill in values. Key variables: - **Storage**: `LOCAL_STORAGE_PATH` (defaults to `~/.reflexio/data`) — houses the SQLite DB file. (Not used by `supabase`/`postgres`, which use external connections.) - **Storage backend**: `REFLEXIO_STORAGE` — `sqlite` (default), `supabase`, or `postgres`. Selects the data storage backend independently from auth configuration. - **Testing**: `IS_TEST_ENV`, `DEBUG_LOG_TO_CONSOLE`, `MOCK_LLM_RESPONSE` -- **Local service supervisor**: `REFLEXIO_SERVICE_HEALTHY_SECS` (default 30), - `REFLEXIO_SERVICE_MAX_FAILS` (default 5), and - `REFLEXIO_SERVICE_RESPAWN_DELAY` (default 2 seconds) Never change env variable values in `.env` directly for port overrides — use shell exports instead. diff --git a/reflexio/cli/utils.py b/reflexio/cli/utils.py index b034f470..9ba67b40 100644 --- a/reflexio/cli/utils.py +++ b/reflexio/cli/utils.py @@ -6,7 +6,6 @@ import dataclasses import hashlib import json -import math import os import signal import subprocess @@ -18,7 +17,6 @@ from pathlib import Path from reflexio.cli.log_format import format_service_line -from reflexio.server.env_utils import env_str @dataclasses.dataclass @@ -184,34 +182,6 @@ def remove_pidfile(pidfile: Path) -> None: pidfile.unlink(missing_ok=True) -def _get_float_env(name: str, default: float, *, minimum: float) -> float: - """Read a bounded floating-point environment setting.""" - raw = env_str(name, str(default)) - try: - value = float(raw) - except ValueError: - print(f"Warning: invalid {name}={raw!r}, using default {default}") - return default - if not math.isfinite(value) or value < minimum: - print(f"Warning: invalid {name}={raw!r}, using default {default}") - return default - return value - - -def _get_int_env(name: str, default: int, *, minimum: int) -> int: - """Read a bounded integer environment setting.""" - raw = env_str(name, str(default)) - try: - value = int(raw) - except ValueError: - print(f"Warning: invalid {name}={raw!r}, using default {default}") - return default - if value < minimum: - print(f"Warning: invalid {name}={raw!r}, using default {default}") - return default - return value - - # Patterns that indicate a service is ready to accept requests _READY_PATTERNS: dict[str, list[str]] = { "backend": ["Application startup complete"], @@ -228,6 +198,11 @@ def _get_int_env(name: str, default: int, *, minimum: int) -> int: "docs": {"NODE_NO_WARNINGS": "1"}, } +# Keep supervisor tuning internal until operational evidence justifies exposing it. +_SERVICE_HEALTHY_WINDOW_SECS = 30.0 +_SERVICE_MAX_RAPID_FAILURES = 5 +_SERVICE_RESPAWN_DELAY_SECS = 2.0 + def _stream_output( proc: subprocess.Popen[bytes], @@ -299,9 +274,6 @@ def run_services( recent_failures: dict[str, list[float]] = {} output_lock = threading.RLock() pidfile = get_pidfile_path(ports) - healthy_secs = _get_float_env("REFLEXIO_SERVICE_HEALTHY_SECS", 30.0, minimum=0.001) - max_fails = _get_int_env("REFLEXIO_SERVICE_MAX_FAILS", 5, minimum=1) - respawn_delay = _get_float_env("REFLEXIO_SERVICE_RESPAWN_DELAY", 2.0, minimum=0.0) shutting_down = False def write_current_pidfile() -> None: @@ -382,11 +354,11 @@ def handle_service_exit(name: str, ret: int) -> None: failures = [ timestamp for timestamp in recent_failures.get(name, []) - if now - timestamp <= healthy_secs + if now - timestamp <= _SERVICE_HEALTHY_WINDOW_SECS ] failures.append(now) recent_failures[name] = failures - if len(failures) >= max_fails: + if len(failures) >= _SERVICE_MAX_RAPID_FAILURES: with output_lock: sys.stdout.write( format_service_line( @@ -399,7 +371,7 @@ def handle_service_exit(name: str, ret: int) -> None: sys.stdout.flush() return - time.sleep(respawn_delay) + time.sleep(_SERVICE_RESPAWN_DELAY_SECS) if shutting_down: return try: diff --git a/tests/cli/test_utils.py b/tests/cli/test_utils.py index ccef13e3..149d2c50 100644 --- a/tests/cli/test_utils.py +++ b/tests/cli/test_utils.py @@ -48,9 +48,6 @@ def _patch_run_services_environment( monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) monkeypatch.setattr(utils.signal, "signal", lambda *_args: None) monkeypatch.setattr(utils.time, "sleep", lambda _seconds: None) - monkeypatch.setenv("REFLEXIO_SERVICE_HEALTHY_SECS", "30") - monkeypatch.setenv("REFLEXIO_SERVICE_MAX_FAILS", "2") - monkeypatch.setenv("REFLEXIO_SERVICE_RESPAWN_DELAY", "0") return pidfile @@ -164,11 +161,18 @@ def fake_popen(command, **_kwargs) -> _FakeProcess: {"crash": 8071, "sibling": 8072}, ) - assert [name for name, _proc in started] == ["crash", "sibling", "crash"] + assert [name for name, _proc in started] == [ + "crash", + "sibling", + "crash", + "crash", + "crash", + "crash", + ] sibling = next(proc for name, proc in started if name == "sibling") assert sibling._poll_count == 10 assert sibling.terminated is False - assert "marked degraded after 2 rapid failures" in capsys.readouterr().out + assert "marked degraded after 5 rapid failures" in capsys.readouterr().out @pytest.mark.unit @@ -230,38 +234,6 @@ def fake_popen(*_args, **_kwargs) -> _FakeProcess: assert len(started) == 1 -@pytest.mark.unit -def test_run_services_rejects_nonfinite_supervisor_env_values( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] -) -> None: - """Fall back safely when supervisor float settings are non-finite.""" - _patch_run_services_environment(monkeypatch, tmp_path) - monkeypatch.setenv("REFLEXIO_SERVICE_HEALTHY_SECS", "nan") - monkeypatch.setenv("REFLEXIO_SERVICE_RESPAWN_DELAY", "inf") - started: list[_FakeProcess] = [] - - def fake_popen(*_args, **_kwargs) -> _FakeProcess: - proc = _FakeProcess( - pid=1000 + len(started), - polls_to_exit=1, - returncode=1 if not started else 0, - ) - started.append(proc) - return proc - - monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) - - utils.run_services( - [utils.ServiceConfig(name="backend", command=["backend"])], - {"backend": 8071}, - ) - - output = capsys.readouterr().out - assert len(started) == 2 - assert "invalid REFLEXIO_SERVICE_HEALTHY_SECS='nan'" in output - assert "invalid REFLEXIO_SERVICE_RESPAWN_DELAY='inf'" in output - - @pytest.mark.unit def test_run_services_drops_failed_respawn_without_stopping_sibling( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] From 34953d5d39c94907ab892100f725abab23742769 Mon Sep 17 00:00:00 2001 From: wenchanghan Date: Wed, 15 Jul 2026 01:01:55 -0700 Subject: [PATCH 3/5] fix: harden local service supervision --- reflexio/cli/utils.py | 296 ++++++++++++++++++++++++---------------- tests/cli/test_utils.py | 145 +++++++++++++++++++- 2 files changed, 320 insertions(+), 121 deletions(-) diff --git a/reflexio/cli/utils.py b/reflexio/cli/utils.py index 9ba67b40..3edba9c9 100644 --- a/reflexio/cli/utils.py +++ b/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"], @@ -199,11 +205,18 @@ def remove_pidfile(pidfile: Path) -> None: } # 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], service_name: str, @@ -233,59 +246,52 @@ def _stream_output( ready_event.set() -def _wait_for_all_ready(ready_events: dict[str, threading.Event]) -> bool: - """Wait up to 60 seconds for every service to report readiness.""" - deadline = time.monotonic() + 60 - return all( - event.wait(timeout=max(0, deadline - time.monotonic())) - for event in ready_events.values() - ) - - -def _sleep_while_services_run( - processes: dict[str, subprocess.Popen[bytes]], -) -> None: - """Avoid a busy monitor loop while at least one child remains alive.""" - if processes: - time.sleep(0.5) - +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()) -def run_services( - services: list[ServiceConfig], - ports: dict[str, int], - *, - on_all_ready: Callable[[dict[str, int]], None] | None = None, -) -> None: - """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). - - 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] = {} - service_configs = {svc.name: svc for svc in services} - recent_failures: dict[str, list[float]] = {} - output_lock = threading.RLock() - pidfile = get_pidfile_path(ports) - shutting_down = False +@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() -> None: - write_pidfile(pidfile, {name: proc.pid for name, proc in processes.items()}) + def write_current_pidfile(self) -> None: + write_pidfile( + self.pidfile, {name: proc.pid for name, proc in self.processes.items()} + ) - def start_service(svc: ServiceConfig, *, respawn: bool = False) -> None: + 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} - with output_lock: - action = "Respawning" if respawn else "Starting" - sys.stdout.write(f"{action} {svc.name}...\n") - sys.stdout.flush() + action = "Respawning" if respawn else "Starting" + self.write_output(f"{action} {svc.name}...") proc = subprocess.Popen( svc.command, cwd=svc.cwd, @@ -294,127 +300,177 @@ def start_service(svc: ServiceConfig, *, respawn: bool = False) -> None: stderr=subprocess.STDOUT, ) try: - processes[svc.name] = proc + self.processes[svc.name] = proc ready_event = threading.Event() - ready_events[svc.name] = ready_event - t = threading.Thread( + self.ready_events[svc.name] = ready_event + thread = threading.Thread( target=_stream_output, - args=(proc, svc.name, output_lock, ready_event), + args=(proc, svc.name, self.output_lock, ready_event), daemon=True, ) - t.start() - threads.append(t) + thread.start() + self.threads[svc.name] = thread except (OSError, RuntimeError): - processes.pop(svc.name, None) - ready_events.pop(svc.name, None) + self.processes.pop(svc.name, None) + self.ready_events.pop(svc.name, None) with contextlib.suppress(OSError): proc.terminate() raise - with output_lock: - sys.stdout.write(f" {svc.name} started (PID {proc.pid})\n") - sys.stdout.flush() + self.write_output(f" {svc.name} started (PID {proc.pid})") - def shutdown(_signum: int | None = None, _frame: object = None) -> None: - nonlocal shutting_down - shutting_down = True - with output_lock: - sys.stdout.write("\nShutting down services...\n") - sys.stdout.flush() - for proc in processes.values(): + 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 processes.values(): + 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 t in threads: - t.join(timeout=1.0) - remove_pidfile(pidfile) + 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(name: str, ret: int) -> None: - with output_lock: - sys.stdout.write( - format_service_line(name, f"exited with code {ret}") + "\n" - ) - sys.stdout.flush() - del processes[name] - ready_events.pop(name, None) - write_current_pidfile() - - if ret == 0 or shutting_down: - recent_failures.pop(name, None) + 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 recent_failures.get(name, []) + for timestamp in self.recent_failures.get(name, []) if now - timestamp <= _SERVICE_HEALTHY_WINDOW_SECS ] failures.append(now) - recent_failures[name] = failures + self.recent_failures[name] = failures if len(failures) >= _SERVICE_MAX_RAPID_FAILURES: - with output_lock: - sys.stdout.write( - format_service_line( - name, - f"marked degraded after {len(failures)} " - "rapid failures; will not respawn", - ) - + "\n" + self.write_output( + format_service_line( + name, + f"marked degraded after {len(failures)} " + "rapid failures; will not respawn", ) - sys.stdout.flush() + ) return - time.sleep(_SERVICE_RESPAWN_DELAY_SECS) - if shutting_down: + self.pending_respawns[name] = now + _SERVICE_RESPAWN_DELAY_SECS + + def respawn_due_services(self) -> None: + if self.shutting_down: + self.pending_respawns.clear() return - try: - start_service(service_configs[name], respawn=True) - write_current_pidfile() - except (OSError, RuntimeError) as exc: - ready_events.pop(name, None) - with contextlib.suppress(KeyError, OSError): - processes.pop(name).terminate() - with output_lock: - sys.stdout.write( + 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}") - + "\n" ) - sys.stdout.flush() - signal.signal(signal.SIGINT, shutdown) - signal.signal(signal.SIGTERM, shutdown) - ensure_requested_ports_available(ports) +def run_services( + services: list[ServiceConfig], + ports: dict[str, int], + *, + on_all_ready: Callable[[dict[str, int]], None] | None = None, +) -> None: + """Launch services, pipe output with prefixes, and manage lifecycle. - for svc in services: - start_service(svc) + Each service's stdout/stderr is captured and prefixed with a colored + 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). - # Write pidfile - write_current_pidfile() + 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. + """ + 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 + }, + ) + + 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) + + 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 and _wait_for_all_ready(ready_events): + 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: - handle_service_exit(name, ret) - _sleep_while_services_run(processes) + 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( @@ -433,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]) diff --git a/tests/cli/test_utils.py b/tests/cli/test_utils.py index 149d2c50..7f33a946 100644 --- a/tests/cli/test_utils.py +++ b/tests/cli/test_utils.py @@ -5,6 +5,7 @@ import io import signal import tempfile +import threading from pathlib import Path from subprocess import CompletedProcess @@ -47,7 +48,13 @@ def _patch_run_services_environment( monkeypatch.setattr(utils, "ensure_requested_ports_available", lambda _ports: None) monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) monkeypatch.setattr(utils.signal, "signal", lambda *_args: None) - monkeypatch.setattr(utils.time, "sleep", lambda _seconds: None) + clock = [0.0] + monkeypatch.setattr(utils.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr( + utils.time, + "sleep", + lambda seconds: clock.__setitem__(0, clock[0] + seconds), + ) return pidfile @@ -110,7 +117,15 @@ def test_run_services_respawns_unexpected_exit( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: """Respawn a service after an unexpected non-zero exit.""" - _patch_run_services_environment(monkeypatch, tmp_path) + pidfile = _patch_run_services_environment(monkeypatch, tmp_path) + pidfile_writes: list[tuple[Path, dict[str, int]]] = [] + original_write_pidfile = utils.write_pidfile + + def record_pidfile(path: Path, pids: dict[str, int]) -> None: + original_write_pidfile(path, pids) + pidfile_writes.append((path, pids.copy())) + + monkeypatch.setattr(utils, "write_pidfile", record_pidfile) started: list[_FakeProcess] = [] def fake_popen(*_args, **_kwargs) -> _FakeProcess: @@ -131,6 +146,132 @@ def fake_popen(*_args, **_kwargs) -> _FakeProcess: assert len(started) == 2 assert started[0].pid != started[1].pid + assert (pidfile, {"backend": started[1].pid}) in pidfile_writes + + +@pytest.mark.unit +def test_run_services_cleans_up_when_initial_service_start_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Terminate earlier children when a later initial launch fails.""" + pidfile = _patch_run_services_environment(monkeypatch, tmp_path) + started: list[_FakeProcess] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + if started: + raise OSError("simulated initial spawn failure") + proc = _FakeProcess(pid=1000, polls_to_exit=None) + started.append(proc) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + with pytest.raises(OSError, match="simulated initial spawn failure"): + utils.run_services( + [ + utils.ServiceConfig(name="backend", command=["backend"]), + utils.ServiceConfig(name="embedding", command=["embedding"]), + ], + {"backend": 8071, "embedding": 8072}, + ) + + assert started[0].terminated is True + assert pidfile.exists() is False + + +@pytest.mark.unit +def test_wait_for_all_ready_returns_when_process_exits() -> None: + """Do not wait for the full readiness timeout after a child exits.""" + ready_event = threading.Event() + process = _FakeProcess(pid=1000, polls_to_exit=1, returncode=1) + + assert ( + utils._wait_for_all_ready({"backend": ready_event}, {"backend": process}) + is False + ) + assert process._poll_count == 1 + + +@pytest.mark.unit +def test_run_services_calls_on_all_ready_once( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Invoke the readiness callback once after all services are ready.""" + _patch_run_services_environment(monkeypatch, tmp_path) + monkeypatch.setattr(utils, "_wait_for_all_ready", lambda _events, _processes: True) + started: list[_FakeProcess] = [] + ready_calls: list[dict[str, int]] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess(pid=1000 + len(started), polls_to_exit=1, returncode=0) + started.append(proc) + return proc + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [utils.ServiceConfig(name="backend", command=["backend"])], + {"backend": 8071}, + on_all_ready=lambda ports: ready_calls.append(ports), + ) + + assert ready_calls == [{"backend": 8071}] + + +@pytest.mark.unit +def test_run_services_does_not_respawn_an_explicitly_stopped_service( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Honor a stop request written by a separate stop-services command.""" + _patch_run_services_environment(monkeypatch, tmp_path) + monkeypatch.setattr(utils, "_wait_for_all_ready", lambda _events, _processes: True) + started: list[_FakeProcess] = [] + + def fake_popen(*_args, **_kwargs) -> _FakeProcess: + proc = _FakeProcess(pid=1000 + len(started), polls_to_exit=1, returncode=1) + started.append(proc) + return proc + + def request_stop(_ports: dict[str, int]) -> None: + utils.get_stop_request_path("backend", 8071).touch() + + monkeypatch.setattr(utils.subprocess, "Popen", fake_popen) + + utils.run_services( + [utils.ServiceConfig(name="backend", command=["backend"])], + {"backend": 8071}, + on_all_ready=request_stop, + ) + + assert len(started) == 1 + + +def test_stop_services_writes_intent_before_killing_saved_pids( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + pidfile = tmp_path / "services.json" + pidfile.write_text('{"backend": 1234}') + stop_requests: list[list[int]] = [] + stop_request_path = utils.get_stop_request_path("backend", 8071) + monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) + monkeypatch.setattr(utils, "find_pids_on_port", lambda _port: []) + monkeypatch.setattr(utils, "find_pids_by_pattern", lambda _pattern: []) + + def record_stop(pids: list[int], **_kwargs: object) -> None: + assert stop_request_path.exists() + stop_requests.append(pids) + + monkeypatch.setattr( + utils, + "kill_processes", + record_stop, + ) + + utils.stop_services({"backend": 8071}, {"backend": "backend"}) + + assert stop_requests == [[1234]] + assert stop_request_path.exists() + utils.remove_pidfile(stop_request_path) @pytest.mark.unit From 7554cecaff142f17d63af118c174328f68376912 Mon Sep 17 00:00:00 2001 From: wenchanghan Date: Wed, 15 Jul 2026 01:31:19 -0700 Subject: [PATCH 4/5] fix: isolate test runtime data Keep default test storage and debug logs in a temporary directory while preserving provider-backed test behavior. --- developer.md | 6 +++++- tests/conftest.py | 18 ++++++++++++++---- .../services/storage/test_storage_defaults.py | 8 +++++++- 3 files changed, 26 insertions(+), 6 deletions(-) diff --git a/developer.md b/developer.md index c2194513..7c002cbd 100644 --- a/developer.md +++ b/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: diff --git a/tests/conftest.py b/tests/conftest.py index 27d1c402..a8a4bf7e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,16 +8,12 @@ import pytest -from reflexio.server.extensions import reset_services - _THIS_DIR = Path(__file__).resolve().parent # tests/ PROJECT_ROOT = _THIS_DIR.parent.parent # repo root if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -from reflexio.test_support.llm_mock import cleanup_llm_mock, configure_llm_mock - # Env vars that change OSS code paths and must not leak in from a developer's # `~/.reflexio/.env` or the enterprise worktree `.env`. CI sets none of these, # so the suite passes there even without the cleanup. Cleared once per session @@ -34,6 +30,13 @@ for _var in _OSS_TEST_POLLUTING_ENV_VARS: os.environ.pop(_var, None) +# Load the developer's provider credentials without importing the server yet. +# The server configures file handlers during import, so the temporary paths +# below must be in place before that import occurs. +from reflexio.cli.env_loader import load_reflexio_env # noqa: E402 + +load_reflexio_env() + # Redirect `~/.reflexio` for the entire test session so tests that call # `reflexio.cli.paths.reflexio_home()` (e.g. via `LocalFileConfigStorage`'s # default `base_dir`) don't pick up the developer's existing @@ -44,6 +47,13 @@ # the leftover was from a prior `--storage supabase` run. _REFLEXIO_TEST_HOME = Path(tempfile.mkdtemp(prefix="reflexio-test-home-")) os.environ["REFLEXIO_LOG_DIR"] = str(_REFLEXIO_TEST_HOME) +os.environ["LOCAL_STORAGE_PATH"] = str(_REFLEXIO_TEST_HOME / ".reflexio" / "data") +import reflexio.server as _test_server # noqa: E402 +from reflexio.server.extensions import reset_services # noqa: E402 + +_test_server.LOCAL_STORAGE_PATH = os.environ["LOCAL_STORAGE_PATH"] + +from reflexio.test_support.llm_mock import cleanup_llm_mock, configure_llm_mock def pytest_configure(config): diff --git a/tests/server/services/storage/test_storage_defaults.py b/tests/server/services/storage/test_storage_defaults.py index 7d8a94fb..4f061aa8 100644 --- a/tests/server/services/storage/test_storage_defaults.py +++ b/tests/server/services/storage/test_storage_defaults.py @@ -20,9 +20,15 @@ def test_local_storage_path_defaults_to_home_reflexio_data() -> None: expected = str(reflexio_home() / "data") - env = {k: v for k, v in os.environ.items() if k != "LOCAL_STORAGE_PATH"} import reflexio.server as server_module + assert ( + Path(server_module.LOCAL_STORAGE_PATH).resolve() + == Path(os.environ["LOCAL_STORAGE_PATH"]).resolve() + ) + + env = {k: v for k, v in os.environ.items() if k != "LOCAL_STORAGE_PATH"} + try: with patch.dict(os.environ, env, clear=True): reloaded = importlib.reload(server_module) From 599fa6317b78f2955940264c820be08658e6c267 Mon Sep 17 00:00:00 2001 From: wenchanghan Date: Wed, 15 Jul 2026 01:40:55 -0700 Subject: [PATCH 5/5] test: isolate service test artifacts Keep stop markers and environment reloads inside the test sandbox. --- tests/cli/test_utils.py | 18 ++++++++++++++++-- .../services/storage/test_storage_defaults.py | 5 ++++- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/tests/cli/test_utils.py b/tests/cli/test_utils.py index 7f33a946..605061f2 100644 --- a/tests/cli/test_utils.py +++ b/tests/cli/test_utils.py @@ -47,13 +47,22 @@ def _patch_run_services_environment( pidfile = tmp_path / "services.json" monkeypatch.setattr(utils, "ensure_requested_ports_available", lambda _ports: None) monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) + monkeypatch.setattr( + utils, + "get_stop_request_path", + lambda name, port: tmp_path / f"{name}_{port}.stop", + ) monkeypatch.setattr(utils.signal, "signal", lambda *_args: None) clock = [0.0] monkeypatch.setattr(utils.time, "monotonic", lambda: clock[0]) + + def advance_time(seconds: float) -> None: + clock[0] += seconds + monkeypatch.setattr( utils.time, "sleep", - lambda seconds: clock.__setitem__(0, clock[0] + seconds), + advance_time, ) return pidfile @@ -252,8 +261,13 @@ def test_stop_services_writes_intent_before_killing_saved_pids( pidfile = tmp_path / "services.json" pidfile.write_text('{"backend": 1234}') stop_requests: list[list[int]] = [] - stop_request_path = utils.get_stop_request_path("backend", 8071) + stop_request_path = tmp_path / "backend_8071.stop" monkeypatch.setattr(utils, "get_pidfile_path", lambda _ports: pidfile) + monkeypatch.setattr( + utils, + "get_stop_request_path", + lambda _name, _port: stop_request_path, + ) monkeypatch.setattr(utils, "find_pids_on_port", lambda _port: []) monkeypatch.setattr(utils, "find_pids_by_pattern", lambda _pattern: []) diff --git a/tests/server/services/storage/test_storage_defaults.py b/tests/server/services/storage/test_storage_defaults.py index 4f061aa8..607a6480 100644 --- a/tests/server/services/storage/test_storage_defaults.py +++ b/tests/server/services/storage/test_storage_defaults.py @@ -30,7 +30,10 @@ def test_local_storage_path_defaults_to_home_reflexio_data() -> None: env = {k: v for k, v in os.environ.items() if k != "LOCAL_STORAGE_PATH"} try: - with patch.dict(os.environ, env, clear=True): + with ( + patch.dict(os.environ, env, clear=True), + patch("reflexio.cli.env_loader.load_reflexio_env"), + ): reloaded = importlib.reload(server_module) assert expected == reloaded.LOCAL_STORAGE_PATH finally: