diff --git a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py b/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py index 654bb40741..b7bd215172 100644 --- a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py +++ b/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py @@ -18,6 +18,7 @@ class EnumOmnibaseInfraTopic(str, Enum): All values are raw topic strings as declared in contract.yaml. Members are sorted by (kind, event_name, version). """ + CMD_BASELINE_COMPARISON_REQUEST_V1 = "onex.cmd.omnibase-infra.baseline-comparison-request.v1" # onex.cmd.omnibase-infra.baseline-comparison-request.v1 CMD_BUILD_LOOP_BUILD_V1 = "onex.cmd.omnibase-infra.build-loop-build.v1" # onex.cmd.omnibase-infra.build-loop-build.v1 CMD_BUILD_LOOP_CLASSIFY_V1 = "onex.cmd.omnibase-infra.build-loop-classify.v1" # onex.cmd.omnibase-infra.build-loop-classify.v1 CMD_BUILD_LOOP_CLOSEOUT_V1 = "onex.cmd.omnibase-infra.build-loop-closeout.v1" # onex.cmd.omnibase-infra.build-loop-closeout.v1 @@ -26,7 +27,9 @@ class EnumOmnibaseInfraTopic(str, Enum): CMD_BUILD_LOOP_VERIFY_V1 = "onex.cmd.omnibase-infra.build-loop-verify.v1" # onex.cmd.omnibase-infra.build-loop-verify.v1 CMD_CHAIN_LEARN_V1 = "onex.cmd.omnibase-infra.chain-learn.v1" # onex.cmd.omnibase-infra.chain-learn.v1 CMD_CONSUMER_RESTART_V1 = "onex.cmd.omnibase-infra.consumer-restart.v1" # onex.cmd.omnibase-infra.consumer-restart.v1 + CMD_DELEGATION_QUALITY_GATE_REQUEST_V1 = "onex.cmd.omnibase-infra.delegation-quality-gate-request.v1" # onex.cmd.omnibase-infra.delegation-quality-gate-request.v1 CMD_DELEGATION_REQUEST_V1 = "onex.cmd.omnibase-infra.delegation-request.v1" # onex.cmd.omnibase-infra.delegation-request.v1 + CMD_DELEGATION_ROUTING_REQUEST_V1 = "onex.cmd.omnibase-infra.delegation-routing-request.v1" # onex.cmd.omnibase-infra.delegation-routing-request.v1 CMD_LLM_COMPLETION_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-completion-request.v1" # onex.cmd.omnibase-infra.llm-completion-request.v1 CMD_LLM_EMBEDDING_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-embedding-request.v1" # onex.cmd.omnibase-infra.llm-embedding-request.v1 CMD_LLM_INFERENCE_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-inference-request.v1" # onex.cmd.omnibase-infra.llm-inference-request.v1 @@ -53,6 +56,7 @@ class EnumOmnibaseInfraTopic(str, Enum): EVT_EVAL_COMPLETED_V1 = "onex.evt.omnibase-infra.eval-completed.v1" # onex.evt.omnibase-infra.eval-completed.v1 EVT_EVENT_FORWARDED_V1 = "onex.evt.omnibase-infra.event-forwarded.v1" # onex.evt.omnibase-infra.event-forwarded.v1 EVT_GMAIL_INTENT_RECEIVED_V1 = "onex.evt.omnibase-infra.gmail-intent-received.v1" # onex.evt.omnibase-infra.gmail-intent-received.v1 + EVT_INFERENCE_RESPONSE_V1 = "onex.evt.omnibase-infra.inference-response.v1" # onex.evt.omnibase-infra.inference-response.v1 EVT_LLM_CALL_COMPLETED_V1 = "onex.evt.omnibase-infra.llm-call-completed.v1" # onex.evt.omnibase-infra.llm-call-completed.v1 EVT_LLM_COMPLETION_COMPLETED_V1 = "onex.evt.omnibase-infra.llm-completion-completed.v1" # onex.evt.omnibase-infra.llm-completion-completed.v1 EVT_QUALITY_GATE_RESULT_V1 = "onex.evt.omnibase-infra.quality-gate-result.v1" # onex.evt.omnibase-infra.quality-gate-result.v1 diff --git a/src/omnibase_infra/enums/generated/enum_omniclaude_topic.py b/src/omnibase_infra/enums/generated/enum_omniclaude_topic.py index 3dc7c07356..5feca4f221 100644 --- a/src/omnibase_infra/enums/generated/enum_omniclaude_topic.py +++ b/src/omnibase_infra/enums/generated/enum_omniclaude_topic.py @@ -21,3 +21,4 @@ class EnumOmniclaudeTopic(str, Enum): EVT_CONTEXT_AUDIT_DLQ_V1 = "onex.evt.omniclaude.context-audit-dlq.v1" # onex.evt.omniclaude.context-audit-dlq.v1 EVT_SESSION_COORDINATION_SIGNAL_V1 = "onex.evt.omniclaude.session-coordination-signal.v1" # onex.evt.omniclaude.session-coordination-signal.v1 EVT_SESSION_STATUS_CHANGED_V1 = "onex.evt.omniclaude.session-status-changed.v1" # onex.evt.omniclaude.session-status-changed.v1 + EVT_TASK_DELEGATED_V1 = "onex.evt.omniclaude.task-delegated.v1" # onex.evt.omniclaude.task-delegated.v1 diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 4c0edc0e7e..c46c41cdc1 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -479,10 +479,34 @@ def derive_dlq_topic_for_event_type( ) """Event topic for quality gate evaluation results.""" +TOPIC_DELEGATION_ROUTING_REQUEST: Final[str] = ( + "onex.cmd.omnibase-infra.delegation-routing-request.v1" +) +"""Command topic for routing reducer invocation from the delegation orchestrator.""" + +TOPIC_DELEGATION_QUALITY_GATE_REQUEST: Final[str] = ( + "onex.cmd.omnibase-infra.delegation-quality-gate-request.v1" +) +"""Command topic for quality gate reducer invocation from the delegation orchestrator.""" + +TOPIC_DELEGATION_INFERENCE_RESPONSE: Final[str] = ( + "onex.evt.omnibase-infra.inference-response.v1" +) +"""Event topic for LLM inference responses in the delegation pipeline.""" + +TOPIC_DELEGATION_TASK_DELEGATED: Final[str] = "onex.evt.omniclaude.task-delegated.v1" +"""Backward-compatible event topic for omnidash delegation projection.""" + +TOPIC_DELEGATION_BASELINE_COMPARISON: Final[str] = ( + "onex.cmd.omnibase-infra.baseline-comparison-request.v1" +) +"""Command topic for baseline comparison compute from the delegation orchestrator.""" __all__ = [ "TOPIC_DELEGATION_COMPLETED", "TOPIC_DELEGATION_FAILED", + "TOPIC_DELEGATION_INFERENCE_RESPONSE", + "TOPIC_DELEGATION_QUALITY_GATE_REQUEST", "TOPIC_DELEGATION_QUALITY_GATE_RESULT", "TOPIC_DELEGATION_REQUEST", "TOPIC_DELEGATION_ROUTING_DECISION", diff --git a/src/omnibase_infra/nodes/node_delegation_orchestrator/delegation_intent_bridge.py b/src/omnibase_infra/nodes/node_delegation_orchestrator/delegation_intent_bridge.py new file mode 100644 index 0000000000..cea384dd8e --- /dev/null +++ b/src/omnibase_infra/nodes/node_delegation_orchestrator/delegation_intent_bridge.py @@ -0,0 +1,227 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Delegation intent bridge — executes intents emitted by the orchestrator. + +The delegation orchestrator's dispatchers return intents as output_events: + - ModelRoutingIntent -> needs routing reducer delta() + - ModelInferenceIntent -> needs LLM call + - ModelQualityGateIntent -> needs quality gate reducer delta() + +Without this bridge, those intents are published as raw events with no +consumer on the other side. The bridge subscribes to the event bus and +executes each intent by calling the appropriate reducer or effect, then +publishes the result back to the topic that the orchestrator subscribes to. + +This bridges the gap between the orchestrator's intent-based output and +the event-driven wiring, completing the delegation chain end-to-end. + +Related: + - OMN-7040: Node-based delegation pipeline + - OMN-7381: Wire handler_build_dispatch to delegation orchestrator +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Protocol +from uuid import UUID + +from pydantic import BaseModel + +if TYPE_CHECKING: + from omnibase_infra.protocols import ProtocolEventBusLike + +from omnibase_infra.event_bus.topic_constants import ( + TOPIC_DELEGATION_INFERENCE_RESPONSE, + TOPIC_DELEGATION_QUALITY_GATE_RESULT, + TOPIC_DELEGATION_ROUTING_DECISION, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_inference_intent import ( + ModelInferenceIntent, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_inference_response_data import ( + ModelInferenceResponseData, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_quality_gate_intent import ( + ModelQualityGateIntent, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_routing_intent import ( + ModelRoutingIntent, +) +from omnibase_infra.nodes.node_delegation_quality_gate_reducer.handlers.handler_quality_gate import ( + delta as quality_gate_delta, +) +from omnibase_infra.nodes.node_delegation_quality_gate_reducer.models.model_quality_gate_result import ( + ModelQualityGateResult, +) +from omnibase_infra.nodes.node_delegation_routing_reducer.handlers.handler_delegation_routing import ( + delta as routing_delta, +) +from omnibase_infra.nodes.node_delegation_routing_reducer.models.model_routing_decision import ( + ModelRoutingDecision, +) + +logger = logging.getLogger(__name__) + + +class ProtocolLlmCaller(Protocol): + """Protocol for calling LLM inference.""" + + async def call( + self, intent: ModelInferenceIntent + ) -> ModelInferenceResponseData: ... + + +class DelegationIntentBridge: + """Bridges delegation intents to reducers/effects and publishes results. + + This class completes the delegation chain by: + 1. Receiving intents emitted by the orchestrator dispatchers + 2. Executing the appropriate reducer (routing, quality gate) or effect (LLM) + 3. Publishing results back to topics the orchestrator subscribes to + + The bridge can operate in two modes: + - **Direct mode**: Called directly with intent objects (for testing) + - **Event bus mode**: Wired as a subscriber on intent topics + + Args: + event_bus: Event bus for publishing results back to the orchestrator. + llm_caller: Callable for LLM inference. If None, inference intents + will raise an error. + """ + + def __init__( + self, + event_bus: ProtocolEventBusLike, + llm_caller: ProtocolLlmCaller | None = None, + ) -> None: + self._event_bus: ProtocolEventBusLike = event_bus + self._llm_caller = llm_caller + + async def handle_routing_intent( + self, intent: ModelRoutingIntent + ) -> ModelRoutingDecision: + """Execute routing reducer and publish decision. + + Calls the routing reducer delta() with the delegation request payload, + then publishes the resulting ModelRoutingDecision to the routing-decision + topic. + """ + decision = routing_delta(intent.payload) + logger.info( + "Routing intent resolved: model=%s, endpoint=%s, correlation_id=%s", + decision.selected_model, + decision.endpoint_url, + decision.correlation_id, + ) + await self._publish(decision, TOPIC_DELEGATION_ROUTING_DECISION) + return decision + + async def handle_inference_intent( + self, intent: ModelInferenceIntent + ) -> ModelInferenceResponseData: + """Execute LLM inference and publish response. + + Calls the LLM caller with the inference intent, then publishes + the response data to the inference-response topic. + """ + if self._llm_caller is None: + msg = "No LLM caller configured — cannot execute inference intent" + raise RuntimeError(msg) + response = await self._llm_caller.call(intent) + logger.info( + "Inference intent resolved: model=%s, tokens=%d, correlation_id=%s", + response.model_used, + response.total_tokens, + response.correlation_id, + ) + await self._publish(response, TOPIC_DELEGATION_INFERENCE_RESPONSE) + return response + + async def handle_quality_gate_intent( + self, intent: ModelQualityGateIntent + ) -> ModelQualityGateResult: + """Execute quality gate reducer and publish result. + + Calls the quality gate reducer delta() with the gate input payload, + then publishes the resulting ModelQualityGateResult to the + quality-gate-result topic. + """ + result = quality_gate_delta(intent.payload) + logger.info( + "Quality gate intent resolved: passed=%s, score=%.3f, correlation_id=%s", + result.passed, + result.quality_score, + result.correlation_id, + ) + await self._publish(result, TOPIC_DELEGATION_QUALITY_GATE_RESULT) + return result + + async def handle_output_event(self, event: BaseModel) -> BaseModel | None: + """Route an output event from a dispatcher to the appropriate handler. + + This is the main entry point when wired into the DispatchResultApplier + or called from tests. It inspects the event type and dispatches to the + correct handler method. + + Returns the result model, or None if the event type is not handled. + """ + if isinstance(event, ModelRoutingIntent): + return await self.handle_routing_intent(event) + if isinstance(event, ModelInferenceIntent): + return await self.handle_inference_intent(event) + if isinstance(event, ModelQualityGateIntent): + return await self.handle_quality_gate_intent(event) + return None + + async def _publish(self, model: BaseModel, topic: str) -> None: + """Publish a Pydantic model as an event envelope to the bus.""" + from datetime import UTC, datetime + + from omnibase_core.models.events.model_event_envelope import ( + ModelEventEnvelope, + ) + + correlation_id: UUID | None = getattr(model, "correlation_id", None) + envelope: ModelEventEnvelope[BaseModel] = ModelEventEnvelope( + payload=model, + correlation_id=correlation_id, + envelope_timestamp=datetime.now(UTC), + ) + await self._event_bus.publish_envelope(envelope=envelope, topic=topic) + + +class MockLlmCaller: + """Mock LLM caller for testing — returns synthetic responses.""" + + def __init__(self, response_content: str = "") -> None: + self._response_content = response_content or ( + "def test_example():\n" + ' """Test example function.\n' + " Args:\n" + " None\n" + " Returns:\n" + " None\n" + ' """\n' + " @pytest.mark.unit\n" + " result = example()\n" + " assert result is not None\n" + ) + + async def call(self, intent: ModelInferenceIntent) -> ModelInferenceResponseData: + return ModelInferenceResponseData( + correlation_id=intent.correlation_id, + content=self._response_content, + model_used=intent.model, + latency_ms=42, + prompt_tokens=100, + completion_tokens=200, + total_tokens=300, + ) + + +__all__: list[str] = [ + "DelegationIntentBridge", + "MockLlmCaller", + "ProtocolLlmCaller", +] diff --git a/tests/unit/contracts/test_protocol_ownership.py b/tests/unit/contracts/test_protocol_ownership.py index 7e9ea676d6..413eadf801 100644 --- a/tests/unit/contracts/test_protocol_ownership.py +++ b/tests/unit/contracts/test_protocol_ownership.py @@ -73,6 +73,7 @@ "ProtocolPostgresAdapter": "nodes/node_registry_effect/protocols/protocol_postgres_adapter.py", "ProtocolToolExecutor": "handlers/mcp/protocols.py", "ProtocolLlmHandler": "nodes/node_llm_inference_effect/services/protocol_llm_handler.py", + "ProtocolLlmCaller": "nodes/node_delegation_orchestrator/delegation_intent_bridge.py", "ProtocolContractPublisherSource": "services/contract_publisher/sources/protocol.py", "ProtocolInjectionEffectivenessReader": "services/observability/injection_effectiveness/protocol_reader.py", "ProtocolTopicCatalogService": "services/protocol_topic_catalog_service.py", diff --git a/tests/unit/delegation/test_delegation_chain_e2e.py b/tests/unit/delegation/test_delegation_chain_e2e.py new file mode 100644 index 0000000000..6191669ccb --- /dev/null +++ b/tests/unit/delegation/test_delegation_chain_e2e.py @@ -0,0 +1,316 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""End-to-end delegation chain test using EventBusInmemory. + +Exercises the full delegation pipeline: + 1. Build dispatch creates a delegation request + 2. DispatcherDelegationRequest -> HandlerDelegationWorkflow -> emits ModelRoutingIntent + 3. DelegationIntentBridge -> routing delta() -> publishes ModelRoutingDecision + 4. DispatcherRoutingDecision -> HandlerDelegationWorkflow -> emits ModelInferenceIntent + 5. DelegationIntentBridge -> mock LLM -> publishes ModelInferenceResponseData + 6. (Inference response triggers) -> HandlerDelegationWorkflow -> emits ModelQualityGateIntent + 7. DelegationIntentBridge -> quality gate delta() -> publishes ModelQualityGateResult + 8. DispatcherQualityGateResult -> HandlerDelegationWorkflow -> emits delegation-completed + +This test proves the chain completes end-to-end with the intent bridge wired. + +Related: + - OMN-7040: Node-based delegation pipeline +""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest + +from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory +from omnibase_infra.event_bus.topic_constants import ( + TOPIC_DELEGATION_COMPLETED, + TOPIC_DELEGATION_FAILED, + TOPIC_DELEGATION_INFERENCE_RESPONSE, + TOPIC_DELEGATION_QUALITY_GATE_RESULT, + TOPIC_DELEGATION_REQUEST, + TOPIC_DELEGATION_ROUTING_DECISION, +) +from omnibase_infra.nodes.node_delegation_orchestrator.delegation_intent_bridge import ( + DelegationIntentBridge, + MockLlmCaller, +) +from omnibase_infra.nodes.node_delegation_orchestrator.enums import ( + EnumDelegationState, +) +from omnibase_infra.nodes.node_delegation_orchestrator.handlers.handler_delegation_workflow import ( + HandlerDelegationWorkflow, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_delegation_request import ( + ModelDelegationRequest, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_inference_intent import ( + ModelInferenceIntent, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_inference_response_data import ( + ModelInferenceResponseData, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_quality_gate_intent import ( + ModelQualityGateIntent, +) +from omnibase_infra.nodes.node_delegation_orchestrator.models.model_routing_intent import ( + ModelRoutingIntent, +) +from omnibase_infra.nodes.node_delegation_quality_gate_reducer.models.model_quality_gate_result import ( + ModelQualityGateResult, +) +from omnibase_infra.nodes.node_delegation_routing_reducer.models.model_routing_decision import ( + ModelRoutingDecision, +) + + +@pytest.mark.unit +class TestDelegationChainE2E: + """End-to-end delegation chain tests using in-memory event bus.""" + + @pytest.fixture(autouse=True) + def _setup_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Set required LLM endpoint env vars for routing reducer.""" + monkeypatch.setenv("LLM_CODER_URL", "http://test-coder:8000") + monkeypatch.setenv("LLM_CODER_FAST_URL", "http://test-fast:8001") + monkeypatch.setenv("LLM_DEEPSEEK_R1_URL", "http://test-deepseek:8101") + + @pytest.fixture + def handler(self) -> HandlerDelegationWorkflow: + return HandlerDelegationWorkflow() + + @pytest.fixture + def correlation_id(self) -> object: + return uuid4() + + @pytest.fixture + def delegation_request(self, correlation_id: object) -> ModelDelegationRequest: + return ModelDelegationRequest( + prompt="Run ticket-pipeline for OMN-9999", + task_type="research", + correlation_id=correlation_id, + max_tokens=4096, + emitted_at=datetime.now(UTC), + ) + + @pytest.mark.asyncio + async def test_full_chain_completes_with_passing_gate( + self, + handler: HandlerDelegationWorkflow, + delegation_request: ModelDelegationRequest, + ) -> None: + """Full chain: request -> route -> infer -> gate pass -> COMPLETED.""" + bus = EventBusInmemory(environment="test", group="delegation-test") + await bus.start() + + # Good response content that passes quality gate for "research" task + good_content = ( + "Based on my analysis of the codebase, the ticket OMN-9999 " + "requires changes to the delegation pipeline. The routing " + "reducer needs to handle the new task type. Here is a detailed " + "breakdown of the changes needed across the affected modules." + ) + bridge = DelegationIntentBridge( + event_bus=bus, + llm_caller=MockLlmCaller(response_content=good_content), + ) + + cid = delegation_request.correlation_id + + # Step 1: Handle delegation request -> get routing intents + routing_intents = handler.handle_delegation_request(delegation_request) + assert len(routing_intents) == 1 + assert isinstance(routing_intents[0], ModelRoutingIntent) + + # Step 2: Bridge executes routing intent -> routing decision + decision = await bridge.handle_routing_intent(routing_intents[0]) + assert isinstance(decision, ModelRoutingDecision) + assert decision.correlation_id == cid + assert decision.task_type == "research" + + # Step 3: Handle routing decision -> get inference intents + inference_intents = handler.handle_routing_decision(decision) + assert len(inference_intents) == 1 + assert isinstance(inference_intents[0], ModelInferenceIntent) + assert inference_intents[0].correlation_id == cid + + # Step 4: Bridge executes inference intent -> inference response + response = await bridge.handle_inference_intent(inference_intents[0]) + assert isinstance(response, ModelInferenceResponseData) + assert response.correlation_id == cid + assert response.content == good_content + + # Step 5: Handle inference response -> get quality gate intents + gate_intents = handler.handle_inference_response(response) + assert len(gate_intents) == 1 + assert isinstance(gate_intents[0], ModelQualityGateIntent) + + # Step 6: Bridge executes quality gate intent -> gate result + gate_result = await bridge.handle_quality_gate_intent(gate_intents[0]) + assert isinstance(gate_result, ModelQualityGateResult) + assert gate_result.correlation_id == cid + assert gate_result.passed is True, ( + f"Quality gate should pass for good research content, " + f"but got score={gate_result.quality_score}, " + f"reasons={gate_result.failure_reasons}" + ) + + # Step 7: Handle gate result -> get completion events + events = handler.handle_gate_result(gate_result) + assert len(events) >= 1 # delegation-completed + compat event + baseline + + # Verify terminal state + workflow = handler.workflows[cid] + assert workflow.state == EnumDelegationState.COMPLETED + + # Verify events published to bus + history = await bus.get_event_history(limit=100) + topics_published = [msg.topic for msg in history] + assert TOPIC_DELEGATION_ROUTING_DECISION in topics_published + assert TOPIC_DELEGATION_INFERENCE_RESPONSE in topics_published + assert TOPIC_DELEGATION_QUALITY_GATE_RESULT in topics_published + + await bus.close() + + @pytest.mark.asyncio + async def test_full_chain_fails_with_refusal( + self, + handler: HandlerDelegationWorkflow, + delegation_request: ModelDelegationRequest, + ) -> None: + """Full chain: request -> route -> infer (refusal) -> gate fail -> FAILED.""" + bus = EventBusInmemory(environment="test", group="delegation-test") + await bus.start() + + # Bad response that triggers refusal detection + refusal_content = "I cannot help with that request. As an AI, I'm sorry." + bridge = DelegationIntentBridge( + event_bus=bus, + llm_caller=MockLlmCaller(response_content=refusal_content), + ) + + cid = delegation_request.correlation_id + + # Run through the chain + routing_intents = handler.handle_delegation_request(delegation_request) + decision = await bridge.handle_routing_intent(routing_intents[0]) + inference_intents = handler.handle_routing_decision(decision) + response = await bridge.handle_inference_intent(inference_intents[0]) + gate_intents = handler.handle_inference_response(response) + gate_result = await bridge.handle_quality_gate_intent(gate_intents[0]) + + assert gate_result.passed is False + assert any("REFUSAL" in r for r in gate_result.failure_reasons) + + events = handler.handle_gate_result(gate_result) + assert len(events) >= 1 + + workflow = handler.workflows[cid] + assert workflow.state == EnumDelegationState.FAILED + + await bus.close() + + @pytest.mark.asyncio + async def test_handle_output_event_routes_correctly( + self, + handler: HandlerDelegationWorkflow, + delegation_request: ModelDelegationRequest, + ) -> None: + """Verify handle_output_event dispatches to correct handler.""" + bus = EventBusInmemory(environment="test", group="delegation-test") + await bus.start() + + bridge = DelegationIntentBridge( + event_bus=bus, + llm_caller=MockLlmCaller(), + ) + + # Get a routing intent from the handler + routing_intents = handler.handle_delegation_request(delegation_request) + intent = routing_intents[0] + + # Use handle_output_event (the generic dispatcher) + result = await bridge.handle_output_event(intent) + assert isinstance(result, ModelRoutingDecision) + + await bus.close() + + @pytest.mark.asyncio + async def test_bridge_publishes_to_correct_topics( + self, + handler: HandlerDelegationWorkflow, + delegation_request: ModelDelegationRequest, + ) -> None: + """Verify bridge publishes results to the correct event bus topics.""" + bus = EventBusInmemory(environment="test", group="delegation-test") + await bus.start() + + bridge = DelegationIntentBridge( + event_bus=bus, + llm_caller=MockLlmCaller(), + ) + + cid = delegation_request.correlation_id + + # Run routing intent + routing_intents = handler.handle_delegation_request(delegation_request) + await bridge.handle_routing_intent(routing_intents[0]) + + # Check routing decision was published + routing_history = await bus.get_event_history( + topic=TOPIC_DELEGATION_ROUTING_DECISION + ) + assert len(routing_history) == 1 + + # Continue chain to get inference intent + decision = ModelRoutingDecision.model_validate( + _extract_payload(routing_history[0]) + ) + inference_intents = handler.handle_routing_decision(decision) + await bridge.handle_inference_intent(inference_intents[0]) + + # Check inference response was published + inference_history = await bus.get_event_history( + topic=TOPIC_DELEGATION_INFERENCE_RESPONSE + ) + assert len(inference_history) == 1 + + await bus.close() + + @pytest.mark.asyncio + async def test_chain_idempotent_duplicate_request( + self, + handler: HandlerDelegationWorkflow, + delegation_request: ModelDelegationRequest, + ) -> None: + """Duplicate delegation request returns empty (idempotent).""" + bus = EventBusInmemory(environment="test", group="delegation-test") + await bus.start() + + bridge = DelegationIntentBridge( + event_bus=bus, + llm_caller=MockLlmCaller(), + ) + + # First request creates workflow + intents1 = handler.handle_delegation_request(delegation_request) + assert len(intents1) == 1 + + # Duplicate request returns empty + intents2 = handler.handle_delegation_request(delegation_request) + assert len(intents2) == 0 + + await bus.close() + + +def _extract_payload(message: object) -> dict[str, object]: + """Extract payload dict from an EventBusInmemory message.""" + import json + + value = getattr(message, "value", b"{}") + data = json.loads(value.decode("utf-8") if isinstance(value, bytes) else value) + return data.get("payload", data)