diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index c494d7a824..e3e78832e9 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -23,6 +23,7 @@ from __future__ import annotations import asyncio +import concurrent.futures import hashlib import importlib import inspect @@ -1918,7 +1919,24 @@ def _make_sync_event_publisher( event_bus: object, handler_name: str, ) -> Callable[[str, bytes], None]: - """Adapt async runtime event-bus publish to legacy sync handler publishers.""" + """Adapt async runtime event-bus publish to legacy sync handler publishers. + + The publisher is constructed during ``wire_from_manifest`` while the runtime + kernel's event loop is running, so that loop is captured here as the owning + loop. Legacy sync handlers (e.g. ``HandlerContextRoiRunner``) execute on a + ``ThreadPoolExecutor`` worker thread — the dispatch engine offloads blocking + sync handlers via ``run_in_executor`` — so a publish issued from inside such + a handler runs on a thread that does not own the kernel loop. + + The publish awaitable returned by the event bus binds its internal Futures to + the kernel loop. Running that awaitable on a *foreign* loop (the previous + behavior: ``asyncio.run`` spun a throwaway loop in the worker thread once + ``get_running_loop`` raised ``RuntimeError``) produced the ``got Future + attached to a different loop`` warning and 2-3 minute terminal-emission retry + delays (OMN-13658). Scheduling the coroutine back onto the owning kernel loop + via ``asyncio.run_coroutine_threadsafe`` keeps every Future on its loop, so + the publish completes immediately from any thread. + """ publish = getattr(event_bus, "publish", None) if not callable(publish): raise ModelOnexError( @@ -1927,21 +1945,25 @@ def _make_sync_event_publisher( f"{type(event_bus).__name__!r} has no callable publish()." ) + try: + kernel_loop = asyncio.get_running_loop() + except RuntimeError as exc: + raise ModelOnexError( + "handler_wiring: the sync event_publisher for handler " + f"{handler_name!r} must be constructed on the runtime kernel event " + "loop (wire_from_manifest runs on it), but no running loop was found." + ) from exc + def _publish(topic: str, payload: bytes) -> None: result = publish(topic, None, payload) if not inspect.isawaitable(result): return publish_awaitable = cast("Awaitable[object]", result) - try: - loop = asyncio.get_running_loop() - except RuntimeError: - asyncio.run(_await_event_bus_publish(publish_awaitable)) - return - task = loop.create_task(_await_event_bus_publish(publish_awaitable)) - - def _log_publish_failure(done: asyncio.Task[None]) -> None: + def _log_publish_failure( + done: asyncio.Future[None] | concurrent.futures.Future[None], + ) -> None: try: done.result() except Exception as exc: # noqa: BLE001 — publish boundary logging @@ -1954,7 +1976,27 @@ def _log_publish_failure(done: asyncio.Task[None]) -> None: exc, ) - task.add_done_callback(_log_publish_failure) + try: + running_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + + if running_loop is kernel_loop: + # Publishing from the kernel loop's own thread (async handler path): + # schedule the coroutine directly on it. + task = kernel_loop.create_task(_await_event_bus_publish(publish_awaitable)) + task.add_done_callback(_log_publish_failure) + return + + # Publishing from a ThreadPoolExecutor worker thread (or any thread that + # does not own the kernel loop): hand the coroutine to the kernel loop in + # a thread-safe way so the publish awaitable's Futures stay on their + # owning loop. This avoids the "got Future attached to a different loop" + # warning and the 2-3 minute retry delay (OMN-13658). + future = asyncio.run_coroutine_threadsafe( + _await_event_bus_publish(publish_awaitable), kernel_loop + ) + future.add_done_callback(_log_publish_failure) return _publish diff --git a/tests/unit/runtime/auto_wiring/test_sync_event_publisher_loop_affinity.py b/tests/unit/runtime/auto_wiring/test_sync_event_publisher_loop_affinity.py new file mode 100644 index 0000000000..caa9ad2ddb --- /dev/null +++ b/tests/unit/runtime/auto_wiring/test_sync_event_publisher_loop_affinity.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Loop-affinity regression for the auto-wired sync event_publisher (OMN-13658). + +Legacy sync handlers (e.g. ``HandlerContextRoiRunner``) run on a +``ThreadPoolExecutor`` worker thread because the dispatch engine offloads +blocking sync handlers via ``run_in_executor``. A publish issued from such a +thread must be scheduled back onto the runtime kernel's owning event loop — +where the publish awaitable's internal Futures are bound — NOT executed on a +fresh throwaway ``asyncio.run`` loop in the worker thread. The latter produced +the ``got Future attached to a different loop`` warning and the 2-3 minute +terminal-emission retry delay this ticket removes. +""" + +from __future__ import annotations + +import asyncio +import threading +import time +from collections.abc import Callable +from unittest.mock import patch + +import pytest + +from omnibase_infra.runtime.auto_wiring.handler_wiring import ( + _make_sync_event_publisher, +) + +_TERMINAL_TOPIC = "onex.evt.platform.context-roi.v1" + + +class _RecordingEventBus: + """Async event bus that records the loop its publish coroutine ran on.""" + + def __init__(self) -> None: + self.published: list[tuple[str, bytes | None, bytes]] = [] + self.publish_loop: asyncio.AbstractEventLoop | None = None + + async def publish(self, topic: str, key: bytes | None, value: bytes) -> None: + # Capturing the running loop proves which loop actually executed the + # coroutine: a fresh worker-thread loop would differ from the kernel loop. + self.publish_loop = asyncio.get_running_loop() + self.published.append((topic, key, value)) + + +async def _build_publisher_on_loop( + event_bus: object, +) -> Callable[[str, bytes], None]: + # Constructed on the kernel loop, exactly as wire_from_manifest does. + return _make_sync_event_publisher( + event_bus=event_bus, + handler_name="HandlerContextRoiRunner", + ) + + +@pytest.mark.unit +def test_sync_publisher_from_worker_thread_runs_on_kernel_loop() -> None: + """A publish from a non-event-loop thread runs on the kernel loop. + + Asserts the DoD: completes without error, does NOT spawn a new + ``asyncio.run`` loop, and the publish coroutine executes on the kernel loop + rather than a throwaway worker-thread loop. + """ + kernel_loop = asyncio.new_event_loop() + loop_thread = threading.Thread(target=kernel_loop.run_forever, daemon=True) + loop_thread.start() + try: + event_bus = _RecordingEventBus() + # Build the publisher ON the kernel loop's own thread. + publisher = asyncio.run_coroutine_threadsafe( + _build_publisher_on_loop(event_bus), kernel_loop + ).result(timeout=5) + + worker_errors: list[BaseException] = [] + + # DoD guard: the sync publisher must NOT spin a new asyncio.run loop. + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring.asyncio.run" + ) as mock_asyncio_run: + + def _worker() -> None: + try: + publisher(_TERMINAL_TOPIC, b'{"ok":true}') + except BaseException as exc: # noqa: BLE001 — surface to assertion + worker_errors.append(exc) + + # A genuine worker thread with no running event loop, mirroring the + # ThreadPoolExecutor worker the dispatch engine uses. + worker = threading.Thread(target=_worker) + worker.start() + worker.join(timeout=5) + + assert not worker.is_alive(), "sync publisher call hung on a worker thread" + mock_asyncio_run.assert_not_called() + + assert worker_errors == [] + + # Wait for the coroutine scheduled onto the kernel loop to complete. + deadline = time.monotonic() + 5.0 + while not event_bus.published and time.monotonic() < deadline: + time.sleep(0.01) + + assert event_bus.published == [(_TERMINAL_TOPIC, None, b'{"ok":true}')] + # The publish coroutine ran on the kernel loop — not a worker-thread loop. + assert event_bus.publish_loop is kernel_loop + finally: + kernel_loop.call_soon_threadsafe(kernel_loop.stop) + loop_thread.join(timeout=5) + kernel_loop.close() + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_sync_publisher_from_kernel_loop_thread_schedules_task() -> None: + """A publish from the kernel loop thread schedules a task on that loop. + + The async handler path (publishing while already on the kernel loop) keeps + using ``create_task`` and must still deliver the event. + """ + event_bus = _RecordingEventBus() + publisher = _make_sync_event_publisher( + event_bus=event_bus, + handler_name="HandlerContextRoiRunner", + ) + + publisher(_TERMINAL_TOPIC, b'{"ok":true}') + # Let the scheduled task run to completion on this loop. + await asyncio.sleep(0) + + assert event_bus.published == [(_TERMINAL_TOPIC, None, b'{"ok":true}')] + assert event_bus.publish_loop is asyncio.get_running_loop()