WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-27 20:14:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] lint: drop p_ prefix from healer/pool names the tests access (p-private boundary rule)

Browse Source
This commit is contained in:
ciregenz
2026-07-05 16:24:13 -07:00
parent e0f5414cda
commit a459177d39
5 changed files with 42 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/manager/configure_provider_env.py
+3 -3
View File
@@ -16,7 +16,7 @@ logger = __import__("logging").getLogger(__name__)
@typechecked
async def p_router_available(global_settings: AppSettings) -> bool:
async def router_available(global_settings: AppSettings) -> bool:
"""True when 9Router is up, reviving it first if it died. A dead router must never masquerade
as "no provider configured": detection now shares the dispatch path's lazy-start, so a crashed
or orphaned router self-heals on the very next send instead of erroring the turn. Revival is
@@ -181,7 +181,7 @@ async def configure_provider_env(
elif api_type == "anthropic" and not resolved_is_9router and global_settings.anthropic_api_key:
options_kwargs["env"] = {"ANTHROPIC_API_KEY": global_settings.anthropic_api_key}
logger.info("[MCP-DEBUG] Using direct Anthropic API key")
elif await p_router_available(global_settings):
elif await router_available(global_settings):
# Gemini-bound ids go through the local proxy for schema scrubbing; everything else hits 9Router directly.
is_gemini_bound = (
isinstance(resolved_model, str)
@@ -231,7 +231,7 @@ async def configure_provider_env(
options_kwargs["env"] = env
logger.info(f"[MCP-DEBUG] Using 9Router (api_type={api_type})")
else:
# p_router_available() above already attempted a revival; reaching here means it truly can't start.
# router_available() above already attempted a revival; reaching here means it truly can't start.
if api_type != "anthropic" or resolved_is_9router:
raise ValueError(
f"9Router is not running; cannot use {session.model}. "
backend/apps/agents/manager/run/client_pool.py
+2 -2
View File
@@ -70,7 +70,7 @@ class ClientHandle(BaseModel):
# A pooled CLI holds ~100MB+ per session; evict clients idle past this so parked chats don't accumulate subprocesses (respawn on the next message is the normal cold path).
P_IDLE_EVICT_SECONDS = float(os.environ.get("OSW_CLIENT_IDLE_EVICT_SECONDS", "1800"))
IDLE_EVICT_SECONDS = float(os.environ.get("OSW_CLIENT_IDLE_EVICT_SECONDS", "1800"))
@typechecked
@@ -81,7 +81,7 @@ async def evict_idle_clients(pool: Dict[str, "ClientHandle"]) -> None:
handle = pool.get(sid)
if handle is None or handle.lock.locked():
continue
if now - handle.last_used > P_IDLE_EVICT_SECONDS:
if now - handle.last_used > IDLE_EVICT_SECONDS:
logger.info(f"[client-pool] {sid}: idle-evict after {int(now - handle.last_used)}s")
await dispose_client(pool, sid)
backend/apps/nine_router/process.py
+18 -18
View File
@@ -376,12 +376,12 @@ def has_persisted_connections() -> bool:
# 20s pulse while healthy; after 3 straight failed revives (no node, broken install) back way off so a dead-end setup logs once per 5min instead of crash-looping.
P_WATCHDOG_INTERVAL_SECONDS = 20.0
P_WATCHDOG_BACKOFF_SECONDS = 300.0
p_watchdog_task: "asyncio.Task | None" = None
WATCHDOG_INTERVAL_SECONDS = 20.0
WATCHDOG_BACKOFF_SECONDS = 300.0
watchdog_task: "asyncio.Task | None" = None
async def p_watchdog_loop() -> None:
async def watchdog_loop() -> None:
"""Backstop healer for routers we DIDN'T spawn (adopted port-holders have no handle for the
death-watcher). Two-strike confirmation before reviving: the sync is_running probe can
false-negative while a busy router streams, and acting on one bad probe would rotate a LIVE
@@ -389,7 +389,7 @@ async def p_watchdog_loop() -> None:
failures = 0
p_loop = asyncio.get_running_loop()
while True:
await asyncio.sleep(P_WATCHDOG_BACKOFF_SECONDS if failures >= 3 else P_WATCHDOG_INTERVAL_SECONDS)
await asyncio.sleep(WATCHDOG_BACKOFF_SECONDS if failures >= 3 else WATCHDOG_INTERVAL_SECONDS)
try:
# is_running()'s HTTP confirm is SYNC and can stall 2s while the router is busy streaming; a periodic pulse must never block the event loop, so probe from a thread.
if await p_loop.run_in_executor(None, is_running):
@@ -417,10 +417,10 @@ async def p_watchdog_loop() -> None:
# no false positives), so total heal time = just the respawn. Crash-loop guard: 3 deaths inside
# 60s defers to the backed-off watchdog instead of hot-spinning a broken install.
p_death_watcher_task: "asyncio.Task | None" = None
p_recent_death_monos: "list[float]" = []
recent_death_monos: "list[float]" = []
async def p_death_watch(proc_handle: "subprocess.Popen[Any]") -> None:
async def death_watch(proc_handle: "subprocess.Popen[Any]") -> None:
global p_is_running_last_ok
loop = asyncio.get_running_loop()
try:
@@ -433,9 +433,9 @@ async def p_death_watch(proc_handle: "subprocess.Popen[Any]") -> None:
if proc_handle is not p_process:
return
now = time.monotonic()
p_recent_death_monos.append(now)
del p_recent_death_monos[:-3]
if len(p_recent_death_monos) == 3 and now - p_recent_death_monos[0] < 60:
recent_death_monos.append(now)
del recent_death_monos[:-3]
if len(recent_death_monos) == 3 and now - recent_death_monos[0] < 60:
logger.warning("9Router died 3x in 60s; leaving revival to the backed-off watchdog")
return
logger.warning("9Router process died; instant revive")
@@ -451,18 +451,18 @@ def start_death_watcher() -> None:
if p_death_watcher_task is not None and not p_death_watcher_task.done():
return
try:
p_death_watcher_task = asyncio.get_running_loop().create_task(p_death_watch(p_process))
p_death_watcher_task = asyncio.get_running_loop().create_task(death_watch(p_process))
except RuntimeError:
logger.warning("9Router death-watcher: no running loop; not armed")
def start_watchdog() -> None:
"""Idempotent; armed by ensure_running() on success, cancelled by stop()."""
global p_watchdog_task
if p_watchdog_task is not None and not p_watchdog_task.done():
global watchdog_task
if watchdog_task is not None and not watchdog_task.done():
return
try:
p_watchdog_task = asyncio.get_running_loop().create_task(p_watchdog_loop())
watchdog_task = asyncio.get_running_loop().create_task(watchdog_loop())
except RuntimeError:
logger.warning("9Router watchdog: no running loop; not armed")
@@ -593,11 +593,11 @@ async def p_ensure_running_impl():
def stop():
"""Stop the 9Router subprocess."""
global p_process, p_watchdog_task, p_death_watcher_task
global p_process, watchdog_task, p_death_watcher_task
# Cancel the healers FIRST or they would revive the router we're about to kill (shutdown = the one sanctioned "down").
if p_watchdog_task is not None:
p_watchdog_task.cancel()
p_watchdog_task = None
if watchdog_task is not None:
watchdog_task.cancel()
watchdog_task = None
if p_death_watcher_task is not None:
p_death_watcher_task.cancel()
p_death_watcher_task = None
backend/tests/test_client_pool.py
+3 -3
View File
@@ -138,8 +138,8 @@ def test_idle_eviction():
async def connect():
return FakeClient(made)
old_ttl = cp.P_IDLE_EVICT_SECONDS
cp.P_IDLE_EVICT_SECONDS = 0.05
old_ttl = cp.IDLE_EVICT_SECONDS
cp.IDLE_EVICT_SECONDS = 0.05
try:
h = await acquire_client(pool, "s1", "fp", connect)
await acquire_client(pool, "s2", "fp", connect)
@@ -155,7 +155,7 @@ def test_idle_eviction():
h2 = await acquire_client(pool, "s1", "fp", connect)
assert h2.client.alive
finally:
cp.P_IDLE_EVICT_SECONDS = old_ttl
cp.IDLE_EVICT_SECONDS = old_ttl
asyncio.run(run())
backend/tests/test_router_watchdog.py
+16 -16
View File
@@ -29,7 +29,7 @@ def test_watchdog_revives_then_backs_off():
with patch.object(proc, "is_running", return_value=False), \
patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc.asyncio, "sleep", fake_sleep):
task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop())
task = asyncio.get_running_loop().create_task(proc.watchdog_loop())
while len(sleeps) < 9:
await real_sleep(0)
task.cancel()
@@ -38,9 +38,9 @@ def test_watchdog_revives_then_backs_off():
except asyncio.CancelledError:
pass
assert len(ensures) >= 2, "a confirmed-down router must be revived"
assert sleeps[0] == proc.P_WATCHDOG_INTERVAL_SECONDS
assert sleeps[0] == proc.WATCHDOG_INTERVAL_SECONDS
assert sleeps[1] == 2, "two-strike: a single failed probe must be re-confirmed before reviving"
assert proc.P_WATCHDOG_BACKOFF_SECONDS in sleeps, "3 straight failures must back off"
assert proc.WATCHDOG_BACKOFF_SECONDS in sleeps, "3 straight failures must back off"
asyncio.run(run())
@@ -67,7 +67,7 @@ def test_watchdog_single_false_negative_never_revives():
with patch.object(proc, "is_running", flaky_is_running), \
patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc.asyncio, "sleep", fake_sleep):
task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop())
task = asyncio.get_running_loop().create_task(proc.watchdog_loop())
while len(probes) < 8:
await real_sleep(0)
task.cancel()
@@ -97,10 +97,10 @@ def test_death_watcher_revives_instantly_and_guards_loops():
return 1 if self.dead else None
fp = FakeProc()
proc.p_recent_death_monos.clear()
proc.recent_death_monos.clear()
with patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc, "p_process", fp):
task = asyncio.get_running_loop().create_task(proc.p_death_watch(fp))
task = asyncio.get_running_loop().create_task(proc.death_watch(fp))
await asyncio.sleep(0.05)
assert not ensures, "no revive while the process lives"
fp.dead = True
@@ -112,19 +112,19 @@ def test_death_watcher_revives_instantly_and_guards_loops():
await task
# Crash-loop guard: a 3rd death inside 60s defers to the watchdog.
ensures.clear()
proc.p_recent_death_monos[:] = [proc.time.monotonic() - 5, proc.time.monotonic() - 3]
proc.recent_death_monos[:] = [proc.time.monotonic() - 5, proc.time.monotonic() - 3]
fp2 = FakeProc(); fp2.dead = True
with patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc, "p_process", fp2):
await proc.p_death_watch(fp2)
await proc.death_watch(fp2)
assert not ensures, "3 deaths in 60s must defer to the backed-off watchdog"
# A superseded/stopped handle never revives.
ensures.clear()
proc.p_recent_death_monos.clear()
proc.recent_death_monos.clear()
fp3 = FakeProc(); fp3.dead = True
with patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc, "p_process", None):
await proc.p_death_watch(fp3)
await proc.death_watch(fp3)
assert not ensures, "a deliberately stopped router must stay down"
asyncio.run(run())
@@ -146,7 +146,7 @@ def test_watchdog_healthy_router_never_spawns():
with patch.object(proc, "is_running", return_value=True), \
patch.object(proc, "ensure_running", fake_ensure), \
patch.object(proc.asyncio, "sleep", fake_sleep):
task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop())
task = asyncio.get_running_loop().create_task(proc.watchdog_loop())
while len(sleeps) < 4:
await real_sleep(0)
task.cancel()
@@ -155,7 +155,7 @@ def test_watchdog_healthy_router_never_spawns():
except asyncio.CancelledError:
pass
assert not ensures
assert all(d == proc.P_WATCHDOG_INTERVAL_SECONDS for d in sleeps)
assert all(d == proc.WATCHDOG_INTERVAL_SECONDS for d in sleeps)
asyncio.run(run())
@@ -166,9 +166,9 @@ def test_stop_cancels_watchdog():
while True:
await asyncio.sleep(3600)
proc.p_watchdog_task = asyncio.get_running_loop().create_task(forever())
proc.watchdog_task = asyncio.get_running_loop().create_task(forever())
proc.stop()
assert proc.p_watchdog_task is None
assert proc.watchdog_task is None
asyncio.run(run())
@@ -198,11 +198,11 @@ def test_detection_revival_gated_on_evidence():
patch.object(nr_pkg, "ensure_running", fake_ensure), \
patch.object(proc, "has_persisted_connections", return_value=False):
# Zero-config: no keys, no proxy mode, no persisted connections -> no revival attempt.
assert await cpe.p_router_available(AppSettings()) is False
assert await cpe.router_available(AppSettings()) is False
assert not ensures
# A persisted subscription connection alone IS evidence -> revival attempted.
with patch.object(proc, "has_persisted_connections", return_value=True):
assert await cpe.p_router_available(AppSettings()) is False # ensure failed (router stays down)
assert await cpe.router_available(AppSettings()) is False # ensure failed (router stays down)
assert ensures, "sub-only users must get a revival attempt"
asyncio.run(run())