fix(updates): restart gateway after in-app agent update (#5156)
This commit is contained in:
Rod Boev
nesquena-hermes
parent
8729dbbff7
commit
6d80dd41d6
7 changed files with 543 additions and 326 deletions
154
api/gateway_restart.py
Normal file
154
api/gateway_restart.py
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
"""Helpers for restarting the active-profile Hermes gateway."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from api.profiles import get_active_hermes_home
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_GATEWAY_RESTART_LOCK = threading.Lock()
|
||||
|
||||
|
||||
def _resolve_hermes_command() -> str:
|
||||
"""Resolve the CLI path used for active-profile gateway restarts."""
|
||||
hermes_cmd = shutil.which("hermes")
|
||||
if hermes_cmd:
|
||||
return hermes_cmd
|
||||
|
||||
sibling = Path(sys.executable).parent / "hermes"
|
||||
if sibling.exists():
|
||||
return str(sibling)
|
||||
return "hermes"
|
||||
|
||||
|
||||
def _consume_stream(stream) -> None:
|
||||
"""Drain a subprocess stream to prevent stdout/stderr pipe deadlocks."""
|
||||
try:
|
||||
while stream and stream.read(4096):
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _release_lock() -> None:
|
||||
try:
|
||||
_GATEWAY_RESTART_LOCK.release()
|
||||
except RuntimeError:
|
||||
# The lock may already have been released by another path.
|
||||
pass
|
||||
|
||||
|
||||
def restart_active_profile_gateway(
|
||||
*,
|
||||
quick_timeout_seconds: float = 2.0,
|
||||
background_wait_seconds: float = 240.0,
|
||||
) -> dict:
|
||||
"""Run a non-blocking ``hermes gateway restart`` for the active profile.
|
||||
|
||||
Returns a short status dict with these values:
|
||||
- completed: command finished quickly and succeeded.
|
||||
- in_progress: command did not finish within ``quick_timeout_seconds``.
|
||||
- failed: command finished quickly with non-zero exit status.
|
||||
- busy: restart already in progress from another caller.
|
||||
"""
|
||||
if not _GATEWAY_RESTART_LOCK.acquire(blocking=False):
|
||||
return {
|
||||
"status": "busy",
|
||||
"message": "Restart already in progress. Please wait a moment and try again.",
|
||||
}
|
||||
|
||||
try:
|
||||
active_home = get_active_hermes_home()
|
||||
env = os.environ.copy()
|
||||
env["HERMES_HOME"] = str(active_home)
|
||||
hermes_cmd = _resolve_hermes_command()
|
||||
|
||||
logger.info(
|
||||
"Restarting gateway service via CLI command: %s gateway restart (HERMES_HOME=%s)",
|
||||
hermes_cmd,
|
||||
active_home,
|
||||
)
|
||||
proc = subprocess.Popen(
|
||||
[hermes_cmd, "gateway", "restart"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
try:
|
||||
stdout, stderr = proc.communicate(timeout=quick_timeout_seconds)
|
||||
_release_lock()
|
||||
stdout = (stdout or "").strip()
|
||||
stderr = (stderr or "").strip()
|
||||
if proc.returncode == 0:
|
||||
logger.info("Gateway service restarted successfully: %s", stdout)
|
||||
return {
|
||||
"status": "completed",
|
||||
"message": "Gateway service restarted successfully",
|
||||
"detail": stdout or stderr,
|
||||
}
|
||||
|
||||
logger.error("Gateway service restart failed with code %s: %s", proc.returncode, stderr)
|
||||
return {
|
||||
"status": "failed",
|
||||
"message": f"Restart failed: {stderr or stdout or f'exit code {proc.returncode}'}",
|
||||
"detail": stdout or stderr,
|
||||
"returncode": proc.returncode,
|
||||
}
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.info(
|
||||
"Gateway restart is taking longer than %.1fs (likely draining in-flight runs);"
|
||||
" continuing in background",
|
||||
quick_timeout_seconds,
|
||||
)
|
||||
|
||||
threading.Thread(target=_consume_stream, args=(proc.stdout,), daemon=True).start()
|
||||
threading.Thread(target=_consume_stream, args=(proc.stderr,), daemon=True).start()
|
||||
|
||||
def _wait_and_release() -> None:
|
||||
try:
|
||||
proc.wait(timeout=background_wait_seconds)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error(
|
||||
"Gateway restart process timed out after %.1fs. Terminating process.",
|
||||
background_wait_seconds,
|
||||
)
|
||||
try:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
try:
|
||||
proc.wait(timeout=5.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error(
|
||||
"Gateway restart process refused to die even after SIGKILL.",
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("Failed to terminate timed out gateway restart process.")
|
||||
finally:
|
||||
_release_lock()
|
||||
|
||||
threading.Thread(target=_wait_and_release, daemon=True).start()
|
||||
return {
|
||||
"status": "in_progress",
|
||||
"message": "Gateway service restart initiated (in progress)",
|
||||
}
|
||||
except Exception as exc:
|
||||
_release_lock()
|
||||
logger.exception("Failed to run gateway restart command")
|
||||
return {
|
||||
"status": "failed",
|
||||
"message": f"Internal error running restart: {type(exc).__name__}: {exc}",
|
||||
}
|
||||
125
api/routes.py
125
api/routes.py
|
|
@ -43,6 +43,7 @@ from api.session_events import (
|
|||
subscribe_session_events,
|
||||
unsubscribe_session_events,
|
||||
)
|
||||
from api.gateway_restart import restart_active_profile_gateway
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -9653,120 +9654,28 @@ def _handle_shutdown(handler) -> bool:
|
|||
return True
|
||||
|
||||
|
||||
_RESTART_LOCK = threading.Lock()
|
||||
|
||||
|
||||
def _handle_health_restart(handler) -> bool:
|
||||
"""Restart the Hermes messaging gateway service."""
|
||||
# Acquire the lock to prevent concurrent restart invocations
|
||||
if not _RESTART_LOCK.acquire(blocking=False):
|
||||
logger.warning("Gateway restart already in progress, rejecting concurrent request.")
|
||||
outcome = restart_active_profile_gateway()
|
||||
|
||||
if outcome.get("status") == "completed":
|
||||
return j(handler, {"ok": True, "message": "Gateway service restarted successfully"})
|
||||
|
||||
if outcome.get("status") == "in_progress":
|
||||
return j(handler, {"ok": True, "message": "Gateway service restart initiated (in progress)"})
|
||||
|
||||
if outcome.get("status") == "busy":
|
||||
return j(
|
||||
handler,
|
||||
{"ok": False, "error": "Restart already in progress. Please wait a moment and try again."},
|
||||
status=429
|
||||
{"ok": False, "error": outcome.get("message", "Restart already in progress. Please wait a moment and try again.")},
|
||||
status=429,
|
||||
)
|
||||
|
||||
lock_released = False
|
||||
try:
|
||||
# 1. Resolve HERMES_HOME for the active profile
|
||||
active_home = get_active_hermes_home()
|
||||
|
||||
# 2. Build the environment dictionary with the correct HERMES_HOME
|
||||
env = os.environ.copy()
|
||||
env["HERMES_HOME"] = str(active_home)
|
||||
|
||||
# 3. Resolve the path to the hermes CLI binary
|
||||
hermes_cmd = shutil.which("hermes")
|
||||
if not hermes_cmd:
|
||||
sys_exe = Path(sys.executable)
|
||||
sibling = sys_exe.parent / "hermes"
|
||||
if sibling.exists():
|
||||
hermes_cmd = str(sibling)
|
||||
else:
|
||||
hermes_cmd = "hermes"
|
||||
|
||||
# 4. Run the restart command
|
||||
logger.info("Restarting gateway service via CLI command: %s gateway restart (HERMES_HOME=%s)", hermes_cmd, active_home)
|
||||
|
||||
proc = subprocess.Popen(
|
||||
[hermes_cmd, "gateway", "restart"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
env=env
|
||||
)
|
||||
try:
|
||||
# Wait up to 2.0 seconds for the restart command to complete quickly
|
||||
stdout, stderr = proc.communicate(timeout=2.0)
|
||||
# Since it finished quickly, we can release the lock now
|
||||
_RESTART_LOCK.release()
|
||||
lock_released = True
|
||||
|
||||
if proc.returncode == 0:
|
||||
logger.info("Gateway service restarted successfully: %s", stdout)
|
||||
return j(handler, {"ok": True, "message": "Gateway service restarted successfully"})
|
||||
else:
|
||||
logger.error("Gateway service restart failed with code %d: %s", proc.returncode, stderr)
|
||||
return j(
|
||||
handler,
|
||||
{"ok": False, "error": f"Restart failed: {stderr.strip() or stdout.strip()}"},
|
||||
status=500
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
# If the process doesn't finish within 2 seconds, it is likely draining
|
||||
# in-flight agent runs. We let it continue running in the background.
|
||||
logger.info("Gateway restart is taking longer than 2.0s (likely draining in-flight runs). Letting it run in the background.")
|
||||
|
||||
# Start background threads to consume the stdout/stderr streams to prevent
|
||||
# the child process from blocking on pipe buffer limits when printing logs.
|
||||
import threading
|
||||
|
||||
def consume_stream(stream):
|
||||
try:
|
||||
while stream.read(4096):
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def wait_and_release():
|
||||
try:
|
||||
proc.wait(timeout=240.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error("Gateway restart process timed out after 240.0s. Terminating process.")
|
||||
try:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
try:
|
||||
proc.wait(timeout=5.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error("Gateway restart process refused to die even after SIGKILL.")
|
||||
except Exception as e:
|
||||
logger.exception("Failed to terminate timed out gateway restart process: %s", e)
|
||||
finally:
|
||||
try:
|
||||
_RESTART_LOCK.release()
|
||||
except RuntimeError:
|
||||
# Already released or similar
|
||||
pass
|
||||
|
||||
threading.Thread(target=consume_stream, args=(proc.stdout,), daemon=True).start()
|
||||
threading.Thread(target=consume_stream, args=(proc.stderr,), daemon=True).start()
|
||||
# This thread will release the lock when the child process exits
|
||||
threading.Thread(target=wait_and_release, daemon=True).start()
|
||||
|
||||
# Lock will be released by wait_and_release thread
|
||||
lock_released = True
|
||||
|
||||
return j(handler, {"ok": True, "message": "Gateway service restart initiated (in progress)"})
|
||||
except Exception as exc:
|
||||
if not lock_released:
|
||||
_RESTART_LOCK.release()
|
||||
logger.exception("Failed to run gateway restart command")
|
||||
return j(handler, {"ok": False, "error": f"Internal error running restart: {type(exc).__name__}: {exc}"}, status=500)
|
||||
return j(
|
||||
handler,
|
||||
{"ok": False, "error": outcome.get("message", "Internal error running restart")},
|
||||
status=500,
|
||||
)
|
||||
|
||||
|
||||
def _serve_manifest(handler) -> bool:
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ from collections import OrderedDict
|
|||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from api.gateway_restart import restart_active_profile_gateway
|
||||
from api.config import REPO_ROOT, STREAMS, STREAMS_LOCK
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -1245,6 +1246,38 @@ def _schedule_restart(delay: float = 2.0) -> None:
|
|||
threading.Thread(target=_do, daemon=True).start()
|
||||
|
||||
|
||||
def _ensure_gateway_restart_for_agent_update() -> tuple[bool, dict]:
|
||||
"""Run the active-profile gateway restart when agent checkout changed.
|
||||
|
||||
Returns:
|
||||
(ok, restart_payload) where:
|
||||
- ok is False when restart did not complete and callers must abort success.
|
||||
- restart_payload contains helper status fields for response shaping.
|
||||
"""
|
||||
restart_result = restart_active_profile_gateway()
|
||||
status = str(restart_result.get("status") or "")
|
||||
if status in {"completed", "in_progress"}:
|
||||
return True, restart_result
|
||||
return False, restart_result
|
||||
|
||||
|
||||
def _agent_gateway_restart_failure_message(target: str, restart_result: dict) -> str:
|
||||
if not isinstance(restart_result, dict):
|
||||
return (
|
||||
f'{target} updated, but gateway restart did not complete. '
|
||||
'Run `hermes gateway restart` manually.'
|
||||
)
|
||||
if restart_result.get("message"):
|
||||
return (
|
||||
f'{target} updated, but gateway restart did not complete: '
|
||||
f'{restart_result["message"]}. Run `hermes gateway restart` manually.'
|
||||
)
|
||||
return (
|
||||
f'{target} updated, but gateway restart did not complete. '
|
||||
'Run `hermes gateway restart` manually.'
|
||||
)
|
||||
|
||||
|
||||
def apply_force_update(target: str) -> dict:
|
||||
"""Force-reset the target repo to the latest remote HEAD.
|
||||
|
||||
|
|
@ -1316,14 +1349,27 @@ def apply_force_update(target: str) -> dict:
|
|||
with _cache_lock:
|
||||
_update_cache['checked_at'] = 0
|
||||
|
||||
if target == 'agent':
|
||||
gateway_ok, gateway_result = _ensure_gateway_restart_for_agent_update()
|
||||
if not gateway_ok:
|
||||
return {
|
||||
'ok': False,
|
||||
'message': _agent_gateway_restart_failure_message(target, gateway_result),
|
||||
'target': target,
|
||||
'gateway_restart': gateway_result.get('status'),
|
||||
}
|
||||
|
||||
_schedule_restart()
|
||||
|
||||
return {
|
||||
response = {
|
||||
'ok': True,
|
||||
'message': f'{target} force-updated to {compare_ref}',
|
||||
'target': target,
|
||||
'restart_scheduled': True,
|
||||
}
|
||||
if target == 'agent':
|
||||
response['gateway_restart'] = gateway_result.get('status')
|
||||
return response
|
||||
finally:
|
||||
_apply_lock.release()
|
||||
|
||||
|
|
@ -1532,8 +1578,18 @@ def _apply_update_inner(target):
|
|||
}
|
||||
with _cache_lock:
|
||||
_update_cache['checked_at'] = 0
|
||||
|
||||
if target == 'agent':
|
||||
gateway_ok, gateway_result = _ensure_gateway_restart_for_agent_update()
|
||||
if not gateway_ok:
|
||||
return {
|
||||
'ok': False,
|
||||
'message': _agent_gateway_restart_failure_message(target, gateway_result),
|
||||
'target': target,
|
||||
'gateway_restart': gateway_result.get('status'),
|
||||
}
|
||||
_schedule_restart()
|
||||
return {
|
||||
response = {
|
||||
'ok': True,
|
||||
'message': (
|
||||
f'{target} updated to the latest version. Your local '
|
||||
|
|
@ -1547,11 +1603,24 @@ def _apply_update_inner(target):
|
|||
'restart_scheduled': True,
|
||||
'stash_conflict': True,
|
||||
}
|
||||
if target == 'agent':
|
||||
response['gateway_restart'] = gateway_result.get('status')
|
||||
return response
|
||||
|
||||
# Invalidate cache
|
||||
with _cache_lock:
|
||||
_update_cache['checked_at'] = 0
|
||||
|
||||
if target == 'agent':
|
||||
gateway_ok, gateway_result = _ensure_gateway_restart_for_agent_update()
|
||||
if not gateway_ok:
|
||||
return {
|
||||
'ok': False,
|
||||
'message': _agent_gateway_restart_failure_message(target, gateway_result),
|
||||
'target': target,
|
||||
'gateway_restart': gateway_result.get('status'),
|
||||
}
|
||||
|
||||
# Schedule a self-restart so the updated code is loaded fresh. A plain
|
||||
# git pull leaves stale Python modules in sys.modules — agent imports that
|
||||
# reference new symbols (functions, classes) added in the update will fail
|
||||
|
|
@ -1571,9 +1640,12 @@ def _apply_update_inner(target):
|
|||
'entry may still be present because git stash drop failed.'
|
||||
)
|
||||
|
||||
return {
|
||||
response = {
|
||||
'ok': True,
|
||||
'message': message,
|
||||
'target': target,
|
||||
'restart_scheduled': True,
|
||||
}
|
||||
if target == 'agent':
|
||||
response['gateway_restart'] = gateway_result.get('status')
|
||||
return response
|
||||
|
|
|
|||
|
|
@ -7959,8 +7959,8 @@ async function applyUpdates(){
|
|||
const forceBtnReset=$('btnForceUpdate');
|
||||
if(forceBtnReset){forceBtnReset.style.display='none';forceBtnReset.dataset.target='';}
|
||||
const targets=[];
|
||||
if(window._updateData?.webui?.behind>0) targets.push('webui');
|
||||
if(window._updateData?.agent?.behind>0) targets.push('agent');
|
||||
if(window._updateData?.webui?.behind>0) targets.push('webui');
|
||||
if(!targets.length){
|
||||
const msg='No update target selected. Refresh update status and retry.';
|
||||
if(errEl){errEl.textContent=msg;errEl.style.display='block';}
|
||||
|
|
|
|||
|
|
@ -1,232 +1,62 @@
|
|||
import io
|
||||
import subprocess
|
||||
"""Health route restart contract checks for /api/health/restart."""
|
||||
|
||||
import types
|
||||
from api import routes
|
||||
|
||||
|
||||
class MockPopen:
|
||||
def __init__(self, args, stdout=None, stderr=None, text=True, env=None):
|
||||
self.args = args
|
||||
self.env = env
|
||||
self.returncode = 0
|
||||
self.stdout = io.StringIO("✓ Service restarted")
|
||||
self.stderr = io.StringIO("")
|
||||
self._should_timeout = False
|
||||
self._should_raise = False
|
||||
self._wait_should_timeout = False
|
||||
self.terminated = False
|
||||
self.killed = False
|
||||
import api.routes as routes
|
||||
|
||||
def communicate(self, timeout=None):
|
||||
if self._should_raise:
|
||||
raise OSError("Subprocess execution failed")
|
||||
if self._should_timeout:
|
||||
raise subprocess.TimeoutExpired(self.args, timeout)
|
||||
return self.stdout.getvalue(), self.stderr.getvalue()
|
||||
|
||||
def wait(self, timeout=None):
|
||||
if self._wait_should_timeout:
|
||||
raise subprocess.TimeoutExpired(self.args, timeout)
|
||||
return self.returncode
|
||||
|
||||
def terminate(self):
|
||||
self.terminated = True
|
||||
|
||||
def kill(self):
|
||||
self.killed = True
|
||||
def _call_health_restart(monkeypatch, helper_result):
|
||||
handler = types.SimpleNamespace()
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
monkeypatch.setattr(routes, "restart_active_profile_gateway", lambda: dict(helper_result))
|
||||
return routes._handle_health_restart(handler), responses
|
||||
|
||||
|
||||
def test_handle_health_restart_success(monkeypatch):
|
||||
import threading
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
# Mock profiles home path
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
|
||||
# Mock shutil.which to find hermes CLI
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
called_args = []
|
||||
called_env = {}
|
||||
|
||||
def mock_popen(args, stdout=None, stderr=None, text=True, env=None):
|
||||
called_args.append(args)
|
||||
called_env.update(env or {})
|
||||
return MockPopen(args, stdout, stderr, text, env)
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
# Mock response helper j
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler = types.SimpleNamespace()
|
||||
|
||||
# Call _handle_health_restart
|
||||
result = routes._handle_health_restart(handler)
|
||||
|
||||
result, responses = _call_health_restart(
|
||||
monkeypatch,
|
||||
{"status": "completed", "message": "Gateway service restarted successfully"},
|
||||
)
|
||||
assert result is True
|
||||
assert called_args == [["/mock/bin/hermes", "gateway", "restart"]]
|
||||
assert called_env.get("HERMES_HOME") == "/mock/hermes/home"
|
||||
assert responses == [({"ok": True, "message": "Gateway service restarted successfully"}, 200)]
|
||||
|
||||
|
||||
def test_handle_health_restart_failure(monkeypatch):
|
||||
import threading
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
# Mock profiles home path
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
# Mock subprocess.Popen failure
|
||||
def mock_popen(args, stdout=None, stderr=None, text=True, env=None):
|
||||
mp = MockPopen(args, stdout, stderr, text, env)
|
||||
mp.returncode = 1
|
||||
mp.stdout = io.StringIO("")
|
||||
mp.stderr = io.StringIO("Error: something went wrong")
|
||||
return mp
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler = types.SimpleNamespace()
|
||||
result = routes._handle_health_restart(handler)
|
||||
|
||||
assert result is True
|
||||
assert responses == [({"ok": False, "error": "Restart failed: Error: something went wrong"}, 500)]
|
||||
|
||||
|
||||
def test_handle_health_restart_exception(monkeypatch):
|
||||
import threading
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
# Mock profiles home path
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
# Mock Popen raising OSError immediately on start
|
||||
def mock_popen(args, **kwargs):
|
||||
raise OSError("Subprocess execution failed")
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler = types.SimpleNamespace()
|
||||
result = routes._handle_health_restart(handler)
|
||||
|
||||
assert result is True
|
||||
assert responses[0][0]["ok"] is False
|
||||
assert "Internal error running restart" in responses[0][0]["error"]
|
||||
assert responses[0][1] == 500
|
||||
|
||||
|
||||
def test_handle_health_restart_timeout(monkeypatch):
|
||||
import threading
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
# Mock profiles home path
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
# Mock Popen raising TimeoutExpired on communicate
|
||||
def mock_popen(args, stdout=None, stderr=None, text=True, env=None):
|
||||
mp = MockPopen(args, stdout, stderr, text, env)
|
||||
mp._should_timeout = True
|
||||
return mp
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler = types.SimpleNamespace()
|
||||
result = routes._handle_health_restart(handler)
|
||||
|
||||
result, responses = _call_health_restart(
|
||||
monkeypatch,
|
||||
{"status": "in_progress", "message": "Gateway service restart initiated (in progress)"},
|
||||
)
|
||||
assert result is True
|
||||
assert responses == [({"ok": True, "message": "Gateway service restart initiated (in progress)"}, 200)]
|
||||
|
||||
|
||||
def test_handle_health_restart_failure(monkeypatch):
|
||||
result, responses = _call_health_restart(
|
||||
monkeypatch,
|
||||
{"status": "failed", "message": "Restart failed: bad thing"},
|
||||
)
|
||||
assert result is True
|
||||
assert responses == [({"ok": False, "error": "Restart failed: bad thing"}, 500)]
|
||||
|
||||
|
||||
def test_handle_health_restart_internal_error(monkeypatch):
|
||||
_, responses = _call_health_restart(
|
||||
monkeypatch,
|
||||
{"status": "failed", "message": "Internal error running restart: OSError: bad spawn"},
|
||||
)
|
||||
assert responses == [({"ok": False, "error": "Internal error running restart: OSError: bad spawn"}, 500)]
|
||||
|
||||
|
||||
def test_handle_health_restart_concurrency(monkeypatch):
|
||||
import threading
|
||||
import time
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
|
||||
# Mock profiles home path
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
# Mock Popen that blocks on communicate
|
||||
barrier = threading.Barrier(2)
|
||||
|
||||
class BlockingMockPopen(MockPopen):
|
||||
def communicate(self, timeout=None):
|
||||
barrier.wait() # synchronize with test thread
|
||||
time.sleep(0.5) # hold the lock briefly
|
||||
return self.stdout.getvalue(), self.stderr.getvalue()
|
||||
|
||||
def mock_popen(args, stdout=None, stderr=None, text=True, env=None):
|
||||
return BlockingMockPopen(args, stdout, stderr, text, env)
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler1 = types.SimpleNamespace()
|
||||
handler2 = types.SimpleNamespace()
|
||||
|
||||
# Run first restart in a separate thread (it will block on communicate)
|
||||
t1 = threading.Thread(target=routes._handle_health_restart, args=(handler1,))
|
||||
t1.start()
|
||||
|
||||
# Wait until the first thread starts Popen and blocks on communicate
|
||||
barrier.wait()
|
||||
|
||||
# Attempt a second restart concurrently; it should fail immediately with 429
|
||||
result2 = routes._handle_health_restart(handler2)
|
||||
|
||||
t1.join()
|
||||
|
||||
assert result2 is True
|
||||
# The second response must be a 429 Conflict / Too Many Requests
|
||||
assert responses[0][0]["ok"] is False
|
||||
assert "Restart already in progress" in responses[0][0]["error"]
|
||||
assert responses[0][1] == 429
|
||||
|
||||
|
||||
def test_handle_health_restart_wait_timeout(monkeypatch):
|
||||
import threading
|
||||
import time
|
||||
routes._RESTART_LOCK = threading.Lock()
|
||||
monkeypatch.setattr("api.routes.get_active_hermes_home", lambda: "/mock/hermes/home")
|
||||
monkeypatch.setattr("shutil.which", lambda cmd: "/mock/bin/hermes" if cmd == "hermes" else None)
|
||||
|
||||
mock_instances = []
|
||||
def mock_popen(args, stdout=None, stderr=None, text=True, env=None):
|
||||
mp = MockPopen(args, stdout, stderr, text, env)
|
||||
mp._should_timeout = True
|
||||
mp._wait_should_timeout = True
|
||||
mock_instances.append(mp)
|
||||
return mp
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", mock_popen)
|
||||
|
||||
responses = []
|
||||
monkeypatch.setattr(routes, "j", lambda handler, payload, **kw: responses.append((payload, kw.get("status", 200))) or True)
|
||||
|
||||
handler = types.SimpleNamespace()
|
||||
assert not routes._RESTART_LOCK.locked()
|
||||
|
||||
result = routes._handle_health_restart(handler)
|
||||
assert result is True
|
||||
assert responses == [({"ok": True, "message": "Gateway service restart initiated (in progress)"}, 200)]
|
||||
|
||||
# Wait for background thread to run and finish wait_and_release
|
||||
time.sleep(0.5)
|
||||
|
||||
# The lock should be released even if wait() timed out
|
||||
assert not routes._RESTART_LOCK.locked()
|
||||
assert len(mock_instances) == 1
|
||||
assert mock_instances[0].terminated is True
|
||||
assert mock_instances[0].killed is True
|
||||
_, responses = _call_health_restart(
|
||||
monkeypatch,
|
||||
{"status": "busy", "message": "Restart already in progress. Please wait a moment and try again."},
|
||||
)
|
||||
assert responses == [
|
||||
(
|
||||
{"ok": False, "error": "Restart already in progress. Please wait a moment and try again."},
|
||||
429,
|
||||
)
|
||||
]
|
||||
|
|
|
|||
|
|
@ -101,3 +101,27 @@ def test_update_apply_rejects_zero_target_success_path():
|
|||
assert zero_target_guard >= 0, (
|
||||
"applyUpdates must return before the success/restart flow when targets is empty"
|
||||
)
|
||||
|
||||
|
||||
def test_apply_updates_queues_agent_before_webui():
|
||||
src = _ui_js()
|
||||
apply_start = src.index("async function applyUpdates()")
|
||||
next_fn = src.index("function _showUpdateError", apply_start)
|
||||
body = src[apply_start:next_fn]
|
||||
|
||||
agent_idx = body.index("if(window._updateData?.agent?.behind>0) targets.push('agent');")
|
||||
webui_idx = body.index("if(window._updateData?.webui?.behind>0) targets.push('webui');")
|
||||
assert agent_idx < webui_idx, "agent target should be queued before webui"
|
||||
|
||||
|
||||
def test_apply_updates_wait_for_all_targets_before_reload():
|
||||
src = _ui_js()
|
||||
apply_start = src.index("async function applyUpdates()")
|
||||
next_fn = src.index("function _showUpdateError", apply_start)
|
||||
body = src[apply_start:next_fn]
|
||||
|
||||
loop_end = body.index("for(const target of targets)")
|
||||
restart_wait = body.index("_waitForServerThenReload({baselineServerIdentity});")
|
||||
assert loop_end < restart_wait, (
|
||||
"_waitForServerThenReload should happen only after the update loop"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -696,6 +696,10 @@ class TestSuccessfulUpdateReturnsRestartScheduled:
|
|||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
|
||||
monkeypatch.setattr(
|
||||
'api.updates.restart_active_profile_gateway',
|
||||
lambda: {'status': 'completed', 'message': 'Gateway service restarted successfully'},
|
||||
)
|
||||
|
||||
result = upd.apply_update('agent')
|
||||
assert result['ok'] is True
|
||||
|
|
@ -791,6 +795,230 @@ class TestApplyForceUpdate:
|
|||
assert result['ok'] is False
|
||||
|
||||
|
||||
class TestAgentUpdateRequiresGatewayRestart:
|
||||
"""Agent updates must prove gateway restart before returning ok=True."""
|
||||
|
||||
def test_apply_update_agent_requires_gateway_restart(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
ran = []
|
||||
gateway_restarts = []
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
ran.append(args)
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[0] == 'tag':
|
||||
return '', True
|
||||
if args[:2] == ['status', '--porcelain']:
|
||||
return '', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[0] == 'pull':
|
||||
return 'Already up to date.', True
|
||||
return '', True
|
||||
|
||||
def fake_gateway_restart():
|
||||
gateway_restarts.append('called')
|
||||
return {'status': 'completed', 'message': 'Gateway service restarted successfully'}
|
||||
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
|
||||
monkeypatch.setattr('api.updates.restart_active_profile_gateway', fake_gateway_restart)
|
||||
|
||||
result = upd.apply_update('agent')
|
||||
assert result['ok'] is True
|
||||
assert result['target'] == 'agent'
|
||||
assert result['restart_scheduled'] is True
|
||||
assert result['gateway_restart'] == 'completed'
|
||||
assert gateway_restarts == ['called']
|
||||
|
||||
def test_apply_update_agent_stash_conflict_success_invokes_gateway_restart(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
gateway_restarts = []
|
||||
ran = []
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
ran.append(args)
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[0] == 'tag':
|
||||
return '', True
|
||||
if args[:2] == ['status', '--porcelain']:
|
||||
return 'M file', True
|
||||
if args[:2] == ['status', '--porcelain', '--untracked-files=no']:
|
||||
return 'M file', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[:2] == ['rev-parse', '--short']:
|
||||
return 'abc1234', True
|
||||
if args[:2] == ['stash', 'push']:
|
||||
return '', True
|
||||
if args[:2] == ['stash', 'apply']:
|
||||
return '', False
|
||||
if args[0] == 'stash':
|
||||
return '', True
|
||||
if args[:2] == ['reset', '--hard', 'HEAD']:
|
||||
return '', True
|
||||
if args[0] == 'pull':
|
||||
return 'Updating', True
|
||||
return '', True
|
||||
|
||||
def fake_gateway_restart():
|
||||
gateway_restarts.append('called')
|
||||
return {'status': 'in_progress', 'message': 'Gateway service restart initiated (in progress)'}
|
||||
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
|
||||
monkeypatch.setattr('api.updates.restart_active_profile_gateway', fake_gateway_restart)
|
||||
|
||||
result = upd.apply_update('agent')
|
||||
assert result['ok'] is True
|
||||
assert result['stash_conflict'] is True
|
||||
assert result['target'] == 'agent'
|
||||
assert result['restart_scheduled'] is True
|
||||
assert result['gateway_restart'] == 'in_progress'
|
||||
assert gateway_restarts == ['called']
|
||||
|
||||
def test_apply_update_agent_without_gateway_restart_result_fails(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[0] == 'tag':
|
||||
return '', True
|
||||
if args[:2] == ['status', '--porcelain']:
|
||||
return '', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[:2] == ['rev-parse', '--short', 'origin/master']:
|
||||
return 'abc1234', True
|
||||
if args[0] == 'pull':
|
||||
return 'Already up to date.', True
|
||||
return '', True
|
||||
|
||||
restart_calls = []
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: (_ for _ in ()).throw(AssertionError('must not restart')))
|
||||
monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda: (
|
||||
restart_calls.append('called'),
|
||||
{'status': 'busy', 'message': 'Restart already in progress. Please wait a moment and try again.'},
|
||||
)[1])
|
||||
|
||||
result = upd.apply_update('agent')
|
||||
assert result['ok'] is False
|
||||
assert 'restart_scheduled' not in result
|
||||
assert result['target'] == 'agent'
|
||||
assert result['gateway_restart'] == 'busy'
|
||||
assert 'hermes gateway restart' in result['message']
|
||||
assert restart_calls == ['called']
|
||||
|
||||
def test_apply_force_update_agent_uses_gateway_restart_status(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
ran = []
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
ran.append(args)
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[0] == 'checkout':
|
||||
return '', True
|
||||
if args[0] == 'reset':
|
||||
return '', True
|
||||
return '', True
|
||||
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
|
||||
monkeypatch.setattr('api.updates.restart_active_profile_gateway', lambda: {'status': 'completed', 'message': 'Gateway service restarted successfully'})
|
||||
|
||||
result = upd.apply_force_update('agent')
|
||||
assert result['ok'] is True
|
||||
assert result['target'] == 'agent'
|
||||
assert result['restart_scheduled'] is True
|
||||
assert result['gateway_restart'] == 'completed'
|
||||
|
||||
def test_apply_force_update_agent_fails_when_gateway_restart_busy(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[0] == 'checkout':
|
||||
return '', True
|
||||
if args[0] == 'reset':
|
||||
return '', True
|
||||
return '', True
|
||||
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: (_ for _ in ()).throw(AssertionError('must not restart')))
|
||||
monkeypatch.setattr(
|
||||
'api.updates.restart_active_profile_gateway',
|
||||
lambda: {'status': 'busy', 'message': 'Restart already in progress. Please wait a moment and try again.'},
|
||||
)
|
||||
|
||||
result = upd.apply_force_update('agent')
|
||||
assert result['ok'] is False
|
||||
assert result['target'] == 'agent'
|
||||
assert result['gateway_restart'] == 'busy'
|
||||
assert 'hermes gateway restart' in result['message']
|
||||
|
||||
def test_apply_update_webui_does_not_call_gateway_restart(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
(tmp_path / '.git').mkdir()
|
||||
monkeypatch.setattr(
|
||||
'api.updates.restart_active_profile_gateway',
|
||||
lambda: (_ for _ in ()).throw(AssertionError('helper must not run for webui updates')),
|
||||
)
|
||||
|
||||
def fake_run(args, cwd, timeout=10):
|
||||
if args[0] == 'fetch':
|
||||
return '', True
|
||||
if args[0] == 'tag':
|
||||
return '', True
|
||||
if args[:2] == ['status', '--porcelain']:
|
||||
return '', True
|
||||
if args[:2] == ['rev-parse', '--abbrev-ref']:
|
||||
return 'origin/master', True
|
||||
if args[0] == 'pull':
|
||||
return 'Already up to date.', True
|
||||
return '', True
|
||||
|
||||
monkeypatch.setattr(upd, '_run_git', fake_run)
|
||||
monkeypatch.setattr(upd, 'REPO_ROOT', tmp_path)
|
||||
monkeypatch.setattr(upd, '_AGENT_DIR', tmp_path)
|
||||
monkeypatch.setattr(upd, '_schedule_restart', lambda delay=2.0: None)
|
||||
|
||||
result = upd.apply_update('webui')
|
||||
assert result['ok'] is True
|
||||
assert result['target'] == 'webui'
|
||||
assert result['restart_scheduled'] is True
|
||||
|
||||
|
||||
# ── api/routes.py ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestForceUpdateRoute:
|
||||
|
|
|
|||
Loading…
Reference in a new issue