From 8345029d8fb19ff0112b474331f3e6deb86ce6be Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Wed, 27 May 2026 17:11:34 -0400 Subject: [PATCH 1/2] fix(OMN-12245): run projection handlers off event loop --- .../runtime/auto_wiring/handler_wiring.py | 49 +++++- ...jection_handler_wiring_runtime_dispatch.py | 100 ++++++++++++ .../test_handler_wiring_db_injection.py | 145 ++++++++++++++++++ 3 files changed, 291 insertions(+), 3 deletions(-) create mode 100644 tests/integration/test_projection_handler_wiring_runtime_dispatch.py diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index f4bf3421e4..6196034045 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -850,7 +850,15 @@ def upsert(self, table: str, conflict_key: str, row: dict[str, object]) -> bool: ) conflict_columns = ", ".join(f'"{key}"' for key in conflict_keys) adapted_row = { - key: psycopg2.extras.Json(value) if isinstance(value, dict) else value + key: ( + psycopg2.extras.Json(value) + if isinstance(value, dict) + or ( + isinstance(value, list) + and str(key).endswith(("_json", "_jsonb")) + ) + else value + ) for key, value in row.items() } # table/conflict_key/cols validated by _TABLE_NAME_RE — not raw user input @@ -889,6 +897,29 @@ def query( return SyncPsycopg2Adapter(db_url) +def _connect_projection_runner_db_if_needed(handler_instance: object) -> None: + """Connect BaseProjectionRunner-style DB adapters before direct dispatch.""" + db = getattr(handler_instance, "db", None) + if db is None or getattr(db, "_pool", None) is not None: + return + connect = getattr(db, "connect", None) + if not callable(connect): + return + result = connect() + if asyncio.iscoroutine(result): + asyncio.run(result) + + +def _is_projection_runner_handler(handler_instance: object) -> bool: + """Detect standalone Kafka projection runners exposed in handler_routing.""" + return ( + type(handler_instance).__name__.endswith("ProjectionRunner") + and hasattr(handler_instance, "project_event") + and hasattr(handler_instance, "topics") + and hasattr(handler_instance, "db") + ) + + def _make_projection_dispatch_callback( handler_instance: object, db_tables: list[dict[str, str]], @@ -920,6 +951,13 @@ def _make_projection_dispatch_callback( async def _callback( envelope: ModelEventEnvelope[object], ) -> ModelDispatchResult | None: + if _is_projection_runner_handler(handler_instance): + logger.debug( + "Projection runner skipped by DB-injection auto-wiring: handler=%s topic=%s", + type(handler_instance).__name__, + _extract_projection_topic(envelope) or "unknown", + ) + return None db_url = os.environ.get(db_url_env, "") if not db_url: log_level = ( @@ -952,8 +990,13 @@ async def _callback( input_data = payload.model_dump(mode="json") # type: ignore[union-attr] input_data["_db"] = adapter input_data["_event_type"] = event_type - # Why: Control flow narrows this union at runtime before the attribute access. - result = handler_instance.handle(input_data) # type: ignore[union-attr, attr-defined] + + def _invoke_projection_handler() -> object: + _connect_projection_runner_db_if_needed(handler_instance) + # Why: Control flow narrows this union at runtime before the attribute access. + return handler_instance.handle(input_data) # type: ignore[union-attr, attr-defined] + + result = await asyncio.to_thread(_invoke_projection_handler) if asyncio.iscoroutine(result): result = await cast("Awaitable[object]", result) projected = True diff --git a/tests/integration/test_projection_handler_wiring_runtime_dispatch.py b/tests/integration/test_projection_handler_wiring_runtime_dispatch.py new file mode 100644 index 0000000000..eb1f48852f --- /dev/null +++ b/tests/integration/test_projection_handler_wiring_runtime_dispatch.py @@ -0,0 +1,100 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Integration coverage for runtime projection DB-injection dispatch [OMN-12245].""" + +from __future__ import annotations + +import asyncio +import threading +from unittest.mock import MagicMock, patch + +import pytest + +from omnibase_infra.runtime.auto_wiring.handler_wiring import ( + _make_projection_dispatch_callback, +) + +_PATCH_BUILD_ADAPTER = ( + "omnibase_infra.runtime.auto_wiring.handler_wiring._build_sync_db_adapter" +) +_PATCH_ENVIRON_GET = "omnibase_infra.runtime.auto_wiring.handler_wiring.os.environ.get" + + +@pytest.mark.integration +def test_runtime_projection_dispatch_skips_standalone_runner_classes() -> None: + """Kafka projection runners must not be invoked by DB-injection auto-wiring.""" + + class DelegationProjectionRunner: + topics = ["onex.evt.omniclaude.task-delegated.v1"] + + def __init__(self) -> None: + self.db = object() + self.called = False + + async def project_event(self) -> None: + self.called = True + + def handle(self, input_data: dict[str, object]) -> dict[str, bool]: + self.called = True + return {"projected": True} + + handler = DelegationProjectionRunner() + callback = _make_projection_dispatch_callback( + handler, + [{"name": "delegation_events", "database": "omnidash_analytics"}], + ("onex.evt.omniclaude.task-delegated.v1",), + ) + + envelope = MagicMock() + envelope.topic = "onex.evt.omniclaude.task-delegated.v1" + envelope.payload = {"correlation_id": "corr-1", "task_type": "release-proof"} + + result = asyncio.run(callback(envelope)) + + assert result is None + assert handler.called is False + + +@pytest.mark.integration +def test_runtime_projection_dispatch_runs_sync_handler_off_event_loop() -> None: + """Regular sync projection handlers run in a worker thread under runtime dispatch.""" + + loop_thread_id = threading.get_ident() + handler_thread_ids: list[int] = [] + received: list[dict[str, object]] = [] + + class HandlerProjectionDelegation: + def handle(self, input_data: dict[str, object]) -> dict[str, int]: + handler_thread_ids.append(threading.get_ident()) + received.append(dict(input_data)) + return {"rows_upserted": 1} + + callback = _make_projection_dispatch_callback( + HandlerProjectionDelegation(), + [{"name": "delegation_events", "database": "omnidash_analytics"}], + ("onex.evt.omniclaude.task-delegated.v1",), + ) + + envelope = MagicMock() + envelope.topic = "onex.evt.omniclaude.task-delegated.v1" + envelope.payload = { + "correlation_id": "corr-2", + "task_type": "release-proof", + "quality_gates_checked": 1, + } + fake_adapter = MagicMock() + + with patch( + _PATCH_ENVIRON_GET, + return_value="postgresql://user:pass@host:5432/omnidash_analytics", + ): + with patch(_PATCH_BUILD_ADAPTER, return_value=fake_adapter): + result = asyncio.run(callback(envelope)) + + assert result is None + assert len(received) == 1 + assert received[0]["_db"] is fake_adapter + assert received[0]["_event_type"] == "task-delegated" + assert received[0]["task_type"] == "release-proof" + assert handler_thread_ids == [handler_thread_ids[0]] + assert handler_thread_ids[0] != loop_thread_id diff --git a/tests/unit/runtime/auto_wiring/test_handler_wiring_db_injection.py b/tests/unit/runtime/auto_wiring/test_handler_wiring_db_injection.py index 096ca680b9..3eb1a76a27 100644 --- a/tests/unit/runtime/auto_wiring/test_handler_wiring_db_injection.py +++ b/tests/unit/runtime/auto_wiring/test_handler_wiring_db_injection.py @@ -117,6 +117,122 @@ def handle(self, input_data: dict) -> dict: assert received[0]["_db"] is fake_adapter +@pytest.mark.unit +def test_projection_callback_runs_sync_handler_outside_active_loop() -> None: + """Sync projection handlers may legitimately own their own event loop.""" + + class LoopOwningHandler: + def handle(self, input_data: dict) -> dict: + async def _project() -> dict: + return {"event_type": input_data["_event_type"]} + + return asyncio.run(_project()) + + db_tables = [{"name": "delegation_events", "database": "omnidash_analytics"}] + callback = _make_projection_dispatch_callback( + LoopOwningHandler(), + db_tables, + ("onex.evt.omniclaude.task-delegated.v1",), + ) + + envelope = MagicMock() + envelope.topic = "onex.evt.omniclaude.task-delegated.v1" + envelope.payload = {"correlation_id": "corr-1", "task_type": "release-proof"} + + with patch( + _PATCH_ENVIRON_GET, + return_value="postgresql://user:pass@host:5432/omnidash_analytics", + ): + with patch(_PATCH_BUILD_ADAPTER, return_value=MagicMock()): + result = asyncio.run(callback(envelope)) + + assert result is None + + +@pytest.mark.unit +def test_projection_callback_connects_runner_db_before_handle() -> None: + """BaseProjectionRunner-style handlers need their async DB pool initialized.""" + + class FakeRunnerDb: + def __init__(self) -> None: + self._pool = None + self.connected = False + + async def connect(self) -> None: + await asyncio.sleep(0) + self._pool = object() + self.connected = True + + class DelegationProjectionRunner: + def __init__(self) -> None: + self.db = FakeRunnerDb() + self.handled = False + + def handle(self, input_data: dict) -> dict: + assert self.db.connected is True + self.handled = True + return {"projected": True} + + handler = DelegationProjectionRunner() + db_tables = [{"name": "delegation_events", "database": "omnidash_analytics"}] + callback = _make_projection_dispatch_callback( + handler, + db_tables, + ("onex.evt.omniclaude.task-delegated.v1",), + ) + + envelope = MagicMock() + envelope.topic = "onex.evt.omniclaude.task-delegated.v1" + envelope.payload = {"correlation_id": "corr-1", "task_type": "release-proof"} + + with patch( + _PATCH_ENVIRON_GET, + return_value="postgresql://user:pass@host:5432/omnidash_analytics", + ): + with patch(_PATCH_BUILD_ADAPTER, return_value=MagicMock()): + result = asyncio.run(callback(envelope)) + + assert result is None + assert handler.handled is True + assert handler.db.connected is True + + +@pytest.mark.unit +def test_projection_callback_skips_standalone_projection_runner() -> None: + """Standalone Kafka runners are not safe as direct DB-injection callbacks.""" + + class DelegationProjectionRunner: + topics = ["onex.evt.omniclaude.task-delegated.v1"] + + def __init__(self) -> None: + self.db = object() + self.called = False + + async def project_event(self) -> None: + self.called = True + + def handle(self, input_data: dict) -> dict: + self.called = True + return {"projected": True} + + handler = DelegationProjectionRunner() + db_tables = [{"name": "delegation_events", "database": "omnidash_analytics"}] + callback = _make_projection_dispatch_callback( + handler, + db_tables, + ("onex.evt.omniclaude.task-delegated.v1",), + ) + + envelope = MagicMock() + envelope.topic = "onex.evt.omniclaude.task-delegated.v1" + envelope.payload = {"correlation_id": "corr-1", "task_type": "release-proof"} + + result = asyncio.run(callback(envelope)) + + assert result is None + assert handler.called is False + + @pytest.mark.unit def test_projection_callback_maps_introspection_event_type() -> None: """node-introspection topic maps to _event_type='introspection'.""" @@ -283,6 +399,35 @@ def test_sync_db_adapter_accepts_multi_column_conflict_key() -> None: assert '"savings_usd" = EXCLUDED."savings_usd"' in sql +@pytest.mark.unit +def test_sync_db_adapter_json_adapts_list_values() -> None: + """JSONB list fields must not be sent to Postgres as text arrays.""" + import psycopg2.extras + + cursor = MagicMock() + cursor_context = MagicMock() + cursor_context.__enter__.return_value = cursor + conn = MagicMock() + conn.closed = False + conn.cursor.return_value = cursor_context + + with patch("psycopg2.connect", return_value=conn): + adapter = _build_sync_db_adapter("postgresql://user:pass@host/db") + result = adapter.upsert( + "delegation_events", + "correlation_id", + { + "correlation_id": "corr-1", + "quality_gates_checked": 1, + "quality_gates_checked_jsonb": ["delegate-skill-terminal"], + }, + ) + + assert result is True + params = cursor.execute.call_args.args[1] + assert isinstance(params["quality_gates_checked_jsonb"], psycopg2.extras.Json) + + # --------------------------------------------------------------------------- # Tests: terminal event emission (OMN-11187) # --------------------------------------------------------------------------- From e999e76ea8a21ee05e415efcad57372646ceb9d4 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Wed, 27 May 2026 17:39:53 -0400 Subject: [PATCH 2/2] chore(OMN-12245): refresh dev backmerge evidence gate