backmerge(OMN-12245): projection runtime dispatch hotfix

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

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

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)
 # ---------------------------------------------------------------------------