fix(OMN-12489): read delegation projection from analytics DB

src/omnimarket/nodes/node_llm_delegation_call_effect/contract.yaml

@@ -94,11 +94,11 @@ event_bus:
     - "onex.cmd.omnibase-infra.delegation-inference-request.v1"
 
   publish_topics:
+    - "onex.evt.omnibase-infra.inference-response.v1"
     - "onex.evt.omnimarket.delegation-call-completed.v1"
     - "onex.evt.omnimarket.delegation-escalation-triggered.v1"
     - "onex.evt.omnimarket.delegation-all-tiers-failed.v1"
     - "onex.evt.omnimarket.delegation-model-degraded.v1"
-    - "onex.evt.omnibase-infra.inference-response.v1"
 
   subscribe_topic_metadata:
     "onex.cmd.omnimarket.delegation-execute.v1":
@@ -133,7 +133,7 @@ externally_consumed_topics:
 
 # === HANDLER ROUTING ===
 handler_routing:
-  routing_strategy: "operation_match"
+  routing_strategy: "payload_type_match"
   handlers:
     - operation: "execute_delegation_call"
       event_model:

src/omnimarket/nodes/node_llm_delegation_call_effect/handlers/handler_llm_delegation_call.py

@@ -59,11 +59,44 @@
 _subscribe = contract_subscribe_topics(_CONTRACT)
 _publish = contract_publish_topics(_CONTRACT)
 
-TOPIC_DELEGATION_EXECUTE = _subscribe[0]
-TOPIC_DELEGATION_CALL_COMPLETED = _publish[0]
-TOPIC_DELEGATION_ESCALATION_TRIGGERED = _publish[1]
-TOPIC_DELEGATION_ALL_TIERS_FAILED = _publish[2]
-TOPIC_DELEGATION_MODEL_DEGRADED = _publish[3]
+_DELEGATION_EXECUTE_SUFFIX = (
+    "delegation-execute.v1"  # onex-topic-allow: suffix used for contract lookup
+)
+_DELEGATION_CALL_COMPLETED_SUFFIX = (
+    "delegation-call-completed.v1"  # onex-topic-allow: suffix used for contract lookup
+)
+_DELEGATION_ESCALATION_TRIGGERED_SUFFIX = "delegation-escalation-triggered.v1"  # onex-topic-allow: suffix used for contract lookup
+_DELEGATION_ALL_TIERS_FAILED_SUFFIX = "delegation-all-tiers-failed.v1"  # onex-topic-allow: suffix used for contract lookup
+_DELEGATION_MODEL_DEGRADED_SUFFIX = (
+    "delegation-model-degraded.v1"  # onex-topic-allow: suffix used for contract lookup
+)
+
+
+def _single_contract_topic(topics: tuple[str, ...], suffix: str, section: str) -> str:
+    matches = tuple(topic for topic in topics if topic.endswith(suffix))
+    if len(matches) != 1:
+        raise RuntimeError(
+            f"Contract {_CONTRACT} must declare exactly one event_bus.{section} "
+            f"topic ending with {suffix!r}; found {matches!r}."
+        )
+    return matches[0]
+
+
+TOPIC_DELEGATION_EXECUTE = _single_contract_topic(
+    _subscribe, _DELEGATION_EXECUTE_SUFFIX, "subscribe_topics"
+)
+TOPIC_DELEGATION_CALL_COMPLETED = _single_contract_topic(
+    _publish, _DELEGATION_CALL_COMPLETED_SUFFIX, "publish_topics"
+)
+TOPIC_DELEGATION_ESCALATION_TRIGGERED = _single_contract_topic(
+    _publish, _DELEGATION_ESCALATION_TRIGGERED_SUFFIX, "publish_topics"
+)
+TOPIC_DELEGATION_ALL_TIERS_FAILED = _single_contract_topic(
+    _publish, _DELEGATION_ALL_TIERS_FAILED_SUFFIX, "publish_topics"
+)
+TOPIC_DELEGATION_MODEL_DEGRADED = _single_contract_topic(
+    _publish, _DELEGATION_MODEL_DEGRADED_SUFFIX, "publish_topics"
+)
 
 __all__ = ["HandlerLlmDelegationCall"]
 

src/omnimarket/projection/api_server.py

@@ -91,9 +91,13 @@ def _json_value(value: Any, *, decode_json: bool = False) -> Any:
 
 
 def _dsn() -> str:
-    dsn = os.environ.get("OMNIBASE_INFRA_DB_URL")
+    dsn = os.environ.get("OMNIDASH_ANALYTICS_DB_URL") or os.environ.get(
+        "OMNIBASE_INFRA_DB_URL"
+    )
     if not dsn:
-        raise RuntimeError("OMNIBASE_INFRA_DB_URL is required for projection API")
+        raise RuntimeError(
+            "OMNIDASH_ANALYTICS_DB_URL or OMNIBASE_INFRA_DB_URL is required for projection API"
+        )
     return dsn
 
 

tests/test_projection_api_server.py

@@ -170,14 +170,27 @@ def _with_pool(
 
 
 class TestPostgresDsn:
-    def test_dsn_requires_contract_db_url(self, monkeypatch) -> None:
+    def test_dsn_requires_projection_or_contract_db_url(self, monkeypatch) -> None:
+        monkeypatch.delenv("OMNIDASH_ANALYTICS_DB_URL", raising=False)
         monkeypatch.delenv("OMNIBASE_INFRA_DB_URL", raising=False)
 
-        with pytest.raises(RuntimeError, match="OMNIBASE_INFRA_DB_URL is required"):
+        with pytest.raises(
+            RuntimeError,
+            match="OMNIDASH_ANALYTICS_DB_URL or OMNIBASE_INFRA_DB_URL is required",
+        ):
             _dsn()
 
-    def test_dsn_uses_contract_db_url(self, monkeypatch) -> None:
+    def test_dsn_prefers_analytics_db_url(self, monkeypatch) -> None:
+        expected = "postgresql://projection:pw@db.internal:15436/omnidash_analytics"
+        fallback = "postgresql://projection:pw@db.internal:15436/omnibase_infra"
+        monkeypatch.setenv("OMNIDASH_ANALYTICS_DB_URL", expected)
+        monkeypatch.setenv("OMNIBASE_INFRA_DB_URL", fallback)
+
+        assert _dsn() == expected
+
+    def test_dsn_falls_back_to_contract_db_url(self, monkeypatch) -> None:
         expected = "postgresql://projection:pw@db.internal:15436/omnibase_infra"
+        monkeypatch.delenv("OMNIDASH_ANALYTICS_DB_URL", raising=False)
         monkeypatch.setenv("OMNIBASE_INFRA_DB_URL", expected)
 
         assert _dsn() == expected

tests/unit/nodes/node_llm_delegation_call_effect/test_handler_llm_delegation_call.py

@@ -21,6 +21,7 @@
     TOPIC_DELEGATION_ALL_TIERS_FAILED,
     TOPIC_DELEGATION_CALL_COMPLETED,
     TOPIC_DELEGATION_ESCALATION_TRIGGERED,
+    TOPIC_DELEGATION_MODEL_DEGRADED,
     HandlerLlmDelegationCall,
     _health_cache,
     _is_endpoint_healthy,
@@ -109,6 +110,22 @@ class TestHandlerLlmDelegationCall:
     def setup_method(self) -> None:
         _health_cache.clear()
 
+    @pytest.mark.unit
+    def test_delegation_publish_topics_are_resolved_by_suffix(self) -> None:
+        contract_path = (
+            Path(__file__).resolve().parents[4]
+            / "src/omnimarket/nodes/node_llm_delegation_call_effect/contract.yaml"
+        )
+        with contract_path.open(encoding="utf-8") as fh:
+            contract = yaml.safe_load(fh)
+
+        publish_topics = contract["event_bus"]["publish_topics"]
+        assert publish_topics[0].endswith("inference-response.v1")
+        assert TOPIC_DELEGATION_CALL_COMPLETED in publish_topics
+        assert TOPIC_DELEGATION_ESCALATION_TRIGGERED in publish_topics
+        assert TOPIC_DELEGATION_ALL_TIERS_FAILED in publish_topics
+        assert TOPIC_DELEGATION_MODEL_DEGRADED in publish_topics
+
     @pytest.mark.unit
     def test_missing_endpoint_env_var_returns_failure(
         self, monkeypatch: pytest.MonkeyPatch