From ce376877d94bb802a532ab688605a1a64458ef3b Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 3 Oct 2026 14:56:34 +0530 Subject: [PATCH] fix(gateway): track deferred-platform load ownership per flow, not thread A plugin's register() runs on a plugin-load deadline worker. When it re-walks the platform registry (i18n/YAML layering -> plugin_entries()), _resolve sees its own parent's in-flight loader, but ownership was keyed on threading.get_ident() -- the parent's thread, not the worker -- so the nested walk event.wait()ed on itself until the deadline abandoned the load (#130575). Record in-flight load keys in a per-registry ContextVar instead. Deadline workers start from copy_context(), so a nested walk on the parent's worker takes the existing "recursively requested while loading" return, while unrelated concurrent loaders (fresh contexts) still wait on the event. --- gateway/platform_registry.py | 13 +++++++++---- tests/gateway/test_platform_registry.py | 23 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/gateway/platform_registry.py b/gateway/platform_registry.py index 62fd6b6407..63fcb2bba3 100644 --- a/gateway/platform_registry.py +++ b/gateway/platform_registry.py @@ -7,6 +7,7 @@ registry first, then the legacy built-in path. Plugin side: ``platform_registry .register(PlatformEntry(...))``; gateway side: ``create_adapter("irc", platform_config)``. """ +import contextvars import logging import sys import threading @@ -126,7 +127,11 @@ class PlatformRegistry: self._scoped_deferred: dict[str, dict[str, _Loader]] = {} self._inflight: dict[_LoadKey, threading.Event] = {} self._inflight_loaders: dict[_LoadKey, _Loader] = {} - self._inflight_owners: dict[_LoadKey, int] = {} + # Load keys whose loader is running in the current flow; copied into plugin deadline + # workers, so a nested walk from register() sees its own parent's load as recursive. + self._loading: contextvars.ContextVar[frozenset[_LoadKey]] = contextvars.ContextVar( + f"platform_registry_loading_{id(self)}", default=frozenset() + ) self._cancelled_inflight: set[_LoadKey] = set() # A failed loader is no longer discoverable, but its identity remains # until ownership teardown can CAS-restore the displaced predecessor. @@ -224,11 +229,10 @@ class PlatformRegistry: event = threading.Event() self._inflight[load_key] = event self._inflight_loaders[load_key] = loader - self._inflight_owners[load_key] = threading.get_ident() is_loader = True if event is None: return - if not is_loader and self._inflight_owners.get(load_key) == threading.get_ident(): + if not is_loader and load_key in self._loading.get(): logger.warning("Deferred platform '%s' recursively requested while loading", name) return if not is_loader: @@ -237,11 +241,13 @@ class PlatformRegistry: # we waited for; resolve that predecessor instead of a one-shot false negative. self._resolve(name, active_scope) return + token = self._loading.set(self._loading.get() | {load_key}) try: loader() except Exception as e: logger.warning("Deferred load of platform '%s' failed: %s", name, e, exc_info=True) finally: + self._loading.reset(token) with self._lock: was_cancelled = load_key in self._cancelled_inflight entries, deferred = self._scope_maps(load_key[0]) @@ -249,7 +255,6 @@ class PlatformRegistry: self._consumed_loaders[load_key] = loader self._inflight.pop(load_key, None) self._inflight_loaders.pop(load_key, None) - self._inflight_owners.pop(load_key, None) self._cancelled_inflight.discard(load_key) event.set() if was_cancelled: diff --git a/tests/gateway/test_platform_registry.py b/tests/gateway/test_platform_registry.py index 68550aafde..e1ef7f6792 100644 --- a/tests/gateway/test_platform_registry.py +++ b/tests/gateway/test_platform_registry.py @@ -677,3 +677,26 @@ class TestMigratedPlatformWiring: "probe and the active installer — status displays would " "pip-install as a side effect" ) + + +def test_registry_walk_from_plugin_load_worker_does_not_wait_on_parent(monkeypatch): + """register() re-walking the registry on its deadline worker must not block on its own load.""" + from hermes_cli import plugins_loader + + monkeypatch.setattr(plugins_loader, "_resolve_plugin_load_timeout", lambda: 0.5) + reg = PlatformRegistry() + abandoned = [] + + class Ctx: + def _abandon_load(self): + abandoned.append(True) + + def register(): + reg.plugin_entries() + reg.register(PlatformEntry(name="reentrant", label="Re-entrant", adapter_factory=lambda cfg: None, + check_fn=lambda: True, source="plugin")) + + reg.register_deferred("reentrant", lambda: plugins_loader.run_with_load_deadline("reentrant", Ctx(), register)) + + assert [e.name for e in reg.plugin_entries()] == ["reentrant"] + assert abandoned == []