From a96ec11c62d3a0e1a74bfc59140b440e9b7a54ed Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 07:06:59 -0400 Subject: [PATCH 01/18] fix(OMN-11994): handler_wiring psycopg2 list adaptation for text[] columns (#1730) * fix(OMN-11994): handler_wiring psycopg2 list adaptation for text[] columns Only apply psycopg2.extras.Json() to dict values, not list values. Lists intended for Postgres text[] columns (models_used, machines_used) were being wrapped in Json() producing malformed array literals. * test(OMN-11994): cover psycopg2 text array adaptation * chore(OMN-11994): rerun gates after OCC merge * style(OMN-11994): format handler wiring integration test --------- Co-authored-by: jonahgabriel --- .../runtime/auto_wiring/handler_wiring.py | 4 +- ...ection_handler_db_injection_integration.py | 63 +++++++++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index 8ca03f5259..0ba695e528 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -805,9 +805,7 @@ 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, list)) - else value + key: psycopg2.extras.Json(value) if isinstance(value, dict) else value for key, value in row.items() } # table/conflict_key/cols validated by _TABLE_NAME_RE — not raw user input diff --git a/tests/integration/runtime/test_projection_handler_db_injection_integration.py b/tests/integration/runtime/test_projection_handler_db_injection_integration.py index eac60d9df9..2aad7d8132 100644 --- a/tests/integration/runtime/test_projection_handler_db_injection_integration.py +++ b/tests/integration/runtime/test_projection_handler_db_injection_integration.py @@ -12,12 +12,15 @@ from __future__ import annotations import asyncio +import sys +import types from pathlib import Path from unittest.mock import MagicMock, patch import pytest from omnibase_infra.runtime.auto_wiring.handler_wiring import ( + _build_sync_db_adapter, _make_projection_dispatch_callback, _read_db_io_tables, ) @@ -404,3 +407,63 @@ def handle(self, input_data: dict) -> dict: assert result is None assert call_count[0] == 0 + + +@pytest.mark.integration +def test_sync_psycopg2_adapter_preserves_text_array_lists( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The sync adapter must not JSON-wrap Postgres text[] values.""" + + captured_execute: dict[str, object] = {} + + class FakeJson: + def __init__(self, value: object) -> None: + self.value = value + + class FakeCursor: + def __enter__(self) -> FakeCursor: + return self + + def __exit__(self, *args: object) -> None: + return None + + def execute(self, sql: str, params: object) -> None: + captured_execute["sql"] = sql + captured_execute["params"] = params + + class FakeConnection: + closed = False + autocommit = False + + def cursor(self, *args: object, **kwargs: object) -> FakeCursor: + return FakeCursor() + + fake_extras = types.SimpleNamespace(Json=FakeJson, RealDictCursor=object) + fake_psycopg2 = types.SimpleNamespace( + connect=lambda dsn: FakeConnection(), + extras=fake_extras, + ) + + monkeypatch.setitem(sys.modules, "psycopg2", fake_psycopg2) + monkeypatch.setitem(sys.modules, "psycopg2.extras", fake_extras) + + adapter = _build_sync_db_adapter("postgresql://example") + result = adapter.upsert( + "swarm_runs", + "run_id", + { + "run_id": "run-1", + "models_used": ["qwen3", "gpt-5"], + "machines_used": ["worker-a", "worker-b"], + "metadata": {"source": "integration-test"}, + }, + ) + + assert result is True + params = captured_execute["params"] + assert isinstance(params, dict) + assert params["models_used"] == ["qwen3", "gpt-5"] + assert params["machines_used"] == ["worker-a", "worker-b"] + assert isinstance(params["metadata"], FakeJson) + assert params["metadata"].value == {"source": "integration-test"} From b8b83a32b63ec2d8be81f10242226ee7ca743c4d Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 07:34:24 -0400 Subject: [PATCH 02/18] fix(OMN-11958): remove duplicate dispatcher for node_architecture_graph_query_effect (#1732) * fix(OMN-11958): deduplicate contract names in discover_contracts to prevent DUPLICATE_REGISTRATION crash When two installed packages (e.g. omnibase_infra and omnimarket) register onex.nodes entry points whose contract.yaml files declare the same `name` field, the auto-wiring engine previously built a manifest containing both contracts. During Phase 2 of wire_from_manifest the second contract would attempt to register dispatcher IDs already registered by the first, raising ONEX_CORE_064_DUPLICATE_REGISTRATION and crash-looping the effects runtime. Fix: track seen contract names in discover_contracts(). When a duplicate name is encountered, the second occurrence is dropped and recorded as a ModelDiscoveryError with a clear message identifying both packages. First occurrence wins. Includes a unit test covering the exact cross-package collision scenario. * test(OMN-11958): cover duplicate contract discovery integration --------- Co-authored-by: jonahgabriel --- .../runtime/auto_wiring/discovery.py | 42 +++++++ ...uto_wiring_duplicate_contract_discovery.py | 113 ++++++++++++++++++ .../runtime/auto_wiring/test_discovery.py | 66 ++++++++++ 3 files changed, 221 insertions(+) create mode 100644 tests/integration/runtime/test_auto_wiring_duplicate_contract_discovery.py diff --git a/src/omnibase_infra/runtime/auto_wiring/discovery.py b/src/omnibase_infra/runtime/auto_wiring/discovery.py index 0e7fe67ef4..c651ba02b0 100644 --- a/src/omnibase_infra/runtime/auto_wiring/discovery.py +++ b/src/omnibase_infra/runtime/auto_wiring/discovery.py @@ -74,12 +74,21 @@ def discover_contracts() -> ModelAutoWiringManifest: Errors on individual entry points are captured — they never abort the full scan. + Duplicate contract names (same ``name`` field from two different packages) are + detected and surfaced as :class:`ModelDiscoveryError` entries. The first + occurrence wins; subsequent duplicates are dropped to prevent + ``ONEX_CORE_064_DUPLICATE_REGISTRATION`` crashes at dispatcher registration + time (OMN-11958). + Returns: A :class:`ModelAutoWiringManifest` with all discovered contracts and errors. """ contracts: list[ModelDiscoveredContract] = [] errors: list[ModelDiscoveryError] = [] active_packages = get_active_runtime_packages() + # Tracks first-seen package for each contract name — used to detect + # cross-package duplicates before they reach the dispatch engine. + seen_contract_names: dict[str, str] = {} for ep in entry_points(group=ENTRY_POINT_GROUP): dist = ep.dist @@ -159,6 +168,39 @@ def discover_contracts() -> ModelAutoWiringManifest: ) continue + # Duplicate contract name guard (OMN-11958): two packages shipping a + # contract with the same ``name`` field would produce identical dispatcher + # IDs and crash with ONEX_CORE_064_DUPLICATE_REGISTRATION. Surface the + # collision as a discovery error and skip the duplicate so the runtime + # boots cleanly. The first occurrence (by entry_point iteration order) + # wins; the owning package should remove the stale copy. + if contract.name in seen_contract_names: + first_owner = seen_contract_names[contract.name] + error_msg = ( + f"Duplicate contract name '{contract.name}' already registered " + f"by package '{first_owner}'. Skipping registration from " + f"'{dist_name}' to prevent DUPLICATE_REGISTRATION crash. " + f"Remove the stale entry point from one of these packages." + ) + logger.error( + "DUPLICATE_REGISTRATION prevented: contract='%s' " + "first_owner='%s' duplicate_package='%s' " + "entry_point='%s'", + contract.name, + first_owner, + dist_name, + ep.name, + ) + errors.append( + ModelDiscoveryError( + entry_point_name=ep.name, + package_name=dist_name, + error=error_msg, + ) + ) + continue + + seen_contract_names[contract.name] = dist_name contracts.append(contract) logger.info( "Discovered contract: %s (%s) from %s %s", diff --git a/tests/integration/runtime/test_auto_wiring_duplicate_contract_discovery.py b/tests/integration/runtime/test_auto_wiring_duplicate_contract_discovery.py new file mode 100644 index 0000000000..12150d90d1 --- /dev/null +++ b/tests/integration/runtime/test_auto_wiring_duplicate_contract_discovery.py @@ -0,0 +1,113 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Integration coverage for duplicate auto-wiring contract discovery.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from textwrap import dedent +from unittest.mock import patch + +import pytest + +from omnibase_infra.runtime.auto_wiring.discovery import discover_contracts + +pytestmark = pytest.mark.integration + +_EP_MODULE = "omnibase_infra.runtime.auto_wiring.discovery.entry_points" + + +@dataclass(frozen=True) +class _FakeDist: + name: str + version: str + + +@dataclass(frozen=True) +class _FakeEntryPoint: + name: str + dist: _FakeDist + node_cls: type + + def load(self) -> type: + return self.node_cls + + +def _write_contract(directory: Path, *, name: str) -> Path: + directory.mkdir() + contract_path = directory / "contract.yaml" + contract_path.write_text( + dedent(f"""\ + name: "{name}" + node_type: "effect" + contract_version: + major: 1 + minor: 0 + patch: 0 + node_version: "1.0.0" + description: "Duplicate discovery integration fixture" + """), + encoding="utf-8", + ) + return contract_path + + +def test_duplicate_contract_name_across_entry_point_packages_boots_with_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cross-package duplicate contract names are skipped before dispatcher wiring. + + This covers the OMN-11958 runtime crash shape: two installed packages expose + ``onex.nodes`` entry points whose sibling contract files declare the same + ``name``. Discovery must keep the first contract, report the second as a + discovery error, and return a manifest that can continue into runtime boot. + """ + monkeypatch.setenv("ONEX_ACTIVE_RUNTIME_PACKAGES", "omnibase_infra,omnimarket") + + infra_dir = tmp_path / "infra_node" + market_dir = tmp_path / "market_node" + duplicate_name = "node_architecture_graph_query_effect" + _write_contract(infra_dir, name=duplicate_name) + _write_contract(market_dir, name=duplicate_name) + infra_module = infra_dir / "node.py" + market_module = market_dir / "node.py" + infra_module.write_text("", encoding="utf-8") + market_module.write_text("", encoding="utf-8") + + infra_cls = type("InfraArchitectureGraphQueryEffect", (), {}) + market_cls = type("MarketArchitectureGraphQueryEffect", (), {}) + entry_points = [ + _FakeEntryPoint( + name=duplicate_name, + dist=_FakeDist(name="omnibase_infra", version="0.36.1"), + node_cls=infra_cls, + ), + _FakeEntryPoint( + name=duplicate_name, + dist=_FakeDist(name="omnimarket", version="0.4.0"), + node_cls=market_cls, + ), + ] + + def fake_getfile(node_cls: type) -> str: + if node_cls is infra_cls: + return str(infra_module) + if node_cls is market_cls: + return str(market_module) + raise AssertionError(f"unexpected node class: {node_cls!r}") + + with ( + patch(_EP_MODULE, return_value=entry_points), + patch("inspect.getfile", side_effect=fake_getfile), + ): + manifest = discover_contracts() + + assert manifest.total_discovered == 1 + assert manifest.contracts[0].name == duplicate_name + assert manifest.contracts[0].package_name == "omnibase_infra" + assert manifest.total_errors == 1 + assert manifest.errors[0].package_name == "omnimarket" + assert duplicate_name in manifest.errors[0].error + assert "Duplicate contract name" in manifest.errors[0].error diff --git a/tests/unit/runtime/auto_wiring/test_discovery.py b/tests/unit/runtime/auto_wiring/test_discovery.py index e74a6454aa..ceab6746f5 100644 --- a/tests/unit/runtime/auto_wiring/test_discovery.py +++ b/tests/unit/runtime/auto_wiring/test_discovery.py @@ -371,6 +371,72 @@ def test_captures_missing_contract_as_error(self, tmp_path: Path) -> None: assert manifest.total_errors == 1 assert "No contract.yaml found" in manifest.errors[0].error + @pytest.mark.unit + def test_duplicate_contract_name_across_packages_is_surfaced_as_error( + self, tmp_path: Path + ) -> None: + """Two packages shipping the same contract name must not cause DUPLICATE_REGISTRATION. + + OMN-11958: when omnibase_infra and omnimarket both register an entry point + whose contract.yaml declares the same ``name`` field, the second registration + would crash the effects runtime with ONEX_CORE_064_DUPLICATE_REGISTRATION. + The discovery engine must detect the collision, keep the first occurrence, + and record the duplicate as a ModelDiscoveryError. + """ + dir_a = tmp_path / "node_arch_query_infra" + dir_a.mkdir() + _make_contract_yaml( + dir_a, name="node_architecture_graph_query_effect", node_type="effect" + ) + module_file_a = dir_a / "node.py" + module_file_a.write_text("") + + dir_b = tmp_path / "node_arch_query_market" + dir_b.mkdir() + _make_contract_yaml( + dir_b, name="node_architecture_graph_query_effect", node_type="effect" + ) + module_file_b = dir_b / "node.py" + module_file_b.write_text("") + + cls_a = type("NodeArchQueryInfra", (), {}) + cls_b = type("NodeArchQueryMarket", (), {}) + + ep_a = _make_entry_point( + "node_architecture_graph_query_effect", + node_cls=cls_a, + dist_name="omnibase_infra", + dist_version="0.36.1", + ) + ep_b = _make_entry_point( + "node_architecture_graph_query_effect", + node_cls=cls_b, + dist_name="omnimarket", + dist_version="0.4.0", + ) + + def fake_getfile(cls: type) -> str: + if cls is cls_a: + return str(module_file_a) + return str(module_file_b) + + with ( + patch(_EP_MODULE, return_value=[ep_a, ep_b]), + patch("inspect.getfile", side_effect=fake_getfile), + ): + manifest = discover_contracts() + + # First occurrence wins — only one contract registered. + assert manifest.total_discovered == 1 + assert manifest.contracts[0].package_name == "omnibase_infra" + assert manifest.contracts[0].name == "node_architecture_graph_query_effect" + + # Duplicate is recorded as a discovery error with a clear message. + assert manifest.total_errors == 1 + assert "Duplicate contract name" in manifest.errors[0].error + assert "node_architecture_graph_query_effect" in manifest.errors[0].error + assert manifest.errors[0].package_name == "omnimarket" + class TestDiscoverContractsFromPaths: """Tests for discover_contracts_from_paths.""" From 01fc6312144b9114788311d9f3ead7eb45e8a2fe Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 07:54:54 -0400 Subject: [PATCH 03/18] fix(OMN-11975): use asyncio.gather for concurrent consumer startup (#1731) * fix(OMN-11975): use asyncio.gather for concurrent consumer startup start_consuming() started 224 consumers serially (18-37min). Replace the for-loop with asyncio.gather, pre-reserving _pending_consumer_keys inside the lock before any concurrent call begins (same reservation pattern as subscribe()). Wall-clock startup time becomes the slowest single consumer rather than the sum of all 224. return_exceptions=True lets every consumer attempt startup; the first failure is re-raised after all have settled. * test(OMN-11975): cover concurrent start_consuming integration --------- Co-authored-by: jonahgabriel --- .../event_bus/event_bus_kafka.py | 32 +++++++++-- ...test_kafka_concurrent_subscribe_startup.py | 46 ++++++++++++++++ tests/unit/event_bus/test_kafka_event_bus.py | 54 +++++++++++++++++++ 3 files changed, 127 insertions(+), 5 deletions(-) diff --git a/src/omnibase_infra/event_bus/event_bus_kafka.py b/src/omnibase_infra/event_bus/event_bus_kafka.py index 6cdc0a7356..8e69d72913 100644 --- a/src/omnibase_infra/event_bus/event_bus_kafka.py +++ b/src/omnibase_infra/event_bus/event_bus_kafka.py @@ -2275,7 +2275,11 @@ async def start_consuming(self) -> None: if not self._started: await self.start() - # Collect topics that need consumers while holding lock briefly + # Collect topics that need consumers while holding lock briefly. + # Pre-populate _pending_consumer_keys inside the lock so concurrent + # _start_consumer_for_topic_unlocked calls (via asyncio.gather below) + # cannot race and create duplicate consumers — the same reservation + # pattern used by subscribe(). topics_to_start: list[tuple[str, str]] = [] async with self._lock: for topic in self._subscribers: @@ -2284,12 +2288,30 @@ async def start_consuming(self) -> None: if group_id in seen_group_ids: continue seen_group_ids.add(group_id) - if (topic, group_id) not in self._group_consumers: + consumer_key = (topic, group_id) + if ( + consumer_key not in self._group_consumers + and consumer_key not in self._pending_consumer_keys + ): + self._pending_consumer_keys.add(consumer_key) topics_to_start.append((topic, group_id)) - # Start consumers outside the lock to avoid blocking - for topic, group_id in topics_to_start: - await self._start_consumer_for_topic(topic, group_id) + # Start all consumers concurrently — wall-clock time becomes the + # slowest single consumer rather than the sum of all N consumers. + # return_exceptions=True lets every consumer attempt to start; we + # collect failures and re-raise the first one after all have settled + # so a single bad topic does not starve the rest. + if topics_to_start: + results = await asyncio.gather( + *[ + self._start_consumer_for_topic_unlocked(topic, group_id) + for topic, group_id in topics_to_start + ], + return_exceptions=True, + ) + errors = [r for r in results if isinstance(r, BaseException)] + if errors: + raise errors[0] # Block until shutdown while not self._shutdown: diff --git a/tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py b/tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py index f17528bae1..4df4ed8085 100644 --- a/tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py +++ b/tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py @@ -95,3 +95,49 @@ async def fake_start(start_topic: str, group_id: str) -> None: assert len(started) == 1 assert started[0][0] == topic assert len(bus._subscribers[topic]) == 5 + + +@pytest.mark.asyncio +async def test_start_consuming_starts_distinct_consumers_in_parallel() -> None: + """start_consuming reserves all keys and starts distinct consumers concurrently.""" + bus = _make_bus() + bus._started = True + + active_starts = 0 + max_active_starts = 0 + started: list[tuple[str, str]] = [] + + async def fake_start(topic: str, group_id: str) -> None: + nonlocal active_starts, max_active_starts + active_starts += 1 + max_active_starts = max(max_active_starts, active_starts) + await asyncio.sleep(0) + bus._group_consumers[(topic, group_id)] = AsyncMock() + bus._pending_consumer_keys.discard((topic, group_id)) + started.append((topic, group_id)) + active_starts -= 1 + + bus._start_consumer_for_topic_unlocked = AsyncMock( # type: ignore[method-assign] + side_effect=fake_start + ) + + group_a = "service-a" + group_b = "service-b" + async with bus._lock: + bus._subscribers["onex.evt.omnibase-infra.start-consuming-a.v1"] = [ + (group_a, "sub-a-1", _handler), + (group_b, "sub-a-2", _handler), + ] + bus._subscribers["onex.evt.omnibase-infra.start-consuming-b.v1"] = [ + (group_a, "sub-b-1", _handler), + ] + + task = asyncio.create_task(bus.start_consuming()) + await asyncio.sleep(0.1) + await bus.shutdown() + await asyncio.wait_for(task, timeout=2.0) + + assert bus._start_consumer_for_topic_unlocked.await_count == 3 + assert len(started) == 3 + assert len(set(started)) == 3 + assert max_active_starts > 1 diff --git a/tests/unit/event_bus/test_kafka_event_bus.py b/tests/unit/event_bus/test_kafka_event_bus.py index 1fc4cfc35a..3f6e876cda 100644 --- a/tests/unit/event_bus/test_kafka_event_bus.py +++ b/tests/unit/event_bus/test_kafka_event_bus.py @@ -2059,6 +2059,60 @@ async def consume_with_signal() -> None: # Task should complete await asyncio.wait_for(task, timeout=1.0) + @pytest.mark.asyncio + async def test_start_consuming_starts_consumers_concurrently( + self, mock_producer: AsyncMock + ) -> None: + """start_consuming() launches all consumers concurrently via asyncio.gather. + + Verifies that _start_consumer_for_topic_unlocked is called once per + distinct (topic, group_id) pair, that pending keys are reserved inside + the lock before any concurrent call begins (no duplicates), and that + the wall-clock order of completion does not cause double-starts. + """ + call_log: list[tuple[str, str]] = [] + + async def fake_start_unlocked(topic: str, group_id: str) -> None: + call_log.append((topic, group_id)) + # Yield control to simulate concurrent group-join latency + await asyncio.sleep(0) + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer", + return_value=mock_producer, + ): + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = EventBusKafka(config=config) + await event_bus.start() + + # Pre-register three distinct subscriptions without triggering + # consumer startup (bus is already started but we bypass it here by + # directly injecting into _subscribers so we can test start_consuming + # in isolation). + group_id_a = "svc-a" + group_id_b = "svc-b" + + async with event_bus._lock: + event_bus._subscribers["topic-x"] = [ + (group_id_a, "sub-1", AsyncMock()), + (group_id_b, "sub-2", AsyncMock()), + ] + event_bus._subscribers["topic-y"] = [ + (group_id_a, "sub-3", AsyncMock()), + ] + + event_bus._start_consumer_for_topic_unlocked = fake_start_unlocked # type: ignore[method-assign] + + # Run start_consuming briefly then shut down + task = asyncio.create_task(event_bus.start_consuming()) + await asyncio.sleep(0.1) + await event_bus.shutdown() + await asyncio.wait_for(task, timeout=2.0) + + # Each distinct (topic, group_id) pair must be started exactly once + assert len(call_log) == 3 + assert len(set(call_log)) == 3, "duplicate consumer starts detected" + class TestKafkaEventBusConfig: """Test suite for config-based EventBusKafka construction.""" From 2486ba6e4c886611baa13aff9a72a368f5377d33 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 09:41:41 -0400 Subject: [PATCH 04/18] chore(OMN-11513): update pricing manifest to current fleet model IDs (#1733) Replace four stale local model entries (qwen3-coder-30b-a3b, qwen3-14b, qwen3-embedding-8b, deepseek-r1-distill-qwen-32b) with the five current fleet models as documented in OMN-11513 lane map evidence (2026-05-22): - qwen3.6-35b-a3b: vLLM GPTQ-Int4 on .201:8000, 131072 ctx, ~180 t/s - qwen3.6-27b-mtp-iq4xs: llamacpp MTP on .201:8001, 98304 ctx - text-embedding-gte: vLLM gte-Qwen2-1.5B on .201:8002, 8192 ctx - deepseek-v4-flash: ds4.c DeepSeek V4 Flash 284B on .200:8101, 131072 ctx - deepseek-v4-pro: alias for flash with higher reasoning effort on .200:8101 All local models retain zero API cost (LOCAL_ZERO_API_COST_POLICY). Existing cloud model entries (Claude, GPT, Gemini) are unchanged. Co-authored-by: jonahgabriel --- .../configs/pricing_manifest.yaml | 47 ++++++++++++------- 1 file changed, 31 insertions(+), 16 deletions(-) diff --git a/src/omnibase_infra/configs/pricing_manifest.yaml b/src/omnibase_infra/configs/pricing_manifest.yaml index 0bd486c389..84c6ee115b 100644 --- a/src/omnibase_infra/configs/pricing_manifest.yaml +++ b/src/omnibase_infra/configs/pricing_manifest.yaml @@ -291,61 +291,76 @@ models: min_samples: 20 query: llm_call_metrics usage_source=API created_at>=now-7d authoritative: false - qwen3-coder-30b-a3b: + qwen3.6-35b-a3b: input_cost_per_1k: 0.0 output_cost_per_1k: 0.0 - effective_date: '2026-03-19' - note: Local model on RTX 5090 (port 8000) - zero API cost + effective_date: '2026-05-25' + note: Local model on RTX 5090 .201:8000 (vLLM, GPTQ-Int4, 131072 ctx, ~180 t/s) - zero API cost confidence: LOW_CONFIDENCE source: LOCAL_ZERO_API_COST_POLICY sample_count: 0 evidence: - generated_at: '2026-04-29T00:00:00+00:00' + generated_at: '2026-05-25T00:00:00+00:00' sample_window_days: 7 sample_count: 0 min_samples: 20 query: llm_call_metrics usage_source=API created_at>=now-7d authoritative: false - qwen3-14b: + qwen3.6-27b-mtp-iq4xs: input_cost_per_1k: 0.0 output_cost_per_1k: 0.0 - effective_date: '2026-03-19' - note: Local model on RTX 4090 (port 8001) - zero API cost + effective_date: '2026-05-25' + note: Local model on .201:8001 (llamacpp, Qwen3.6-27B-MTP-IQ4_XS.gguf, 98304 ctx) - zero API cost confidence: LOW_CONFIDENCE source: LOCAL_ZERO_API_COST_POLICY sample_count: 0 evidence: - generated_at: '2026-04-29T00:00:00+00:00' + generated_at: '2026-05-25T00:00:00+00:00' sample_window_days: 7 sample_count: 0 min_samples: 20 query: llm_call_metrics usage_source=API created_at>=now-7d authoritative: false - qwen3-embedding-8b: + text-embedding-gte: input_cost_per_1k: 0.0 output_cost_per_1k: 0.0 - effective_date: '2026-03-19' - note: Local embedding model on M2 Ultra (port 8100) - zero API cost + effective_date: '2026-05-25' + note: Local embedding model on .201:8002 (vLLM, gte-Qwen2-1.5B, 8192 ctx) - zero API cost confidence: LOW_CONFIDENCE source: LOCAL_ZERO_API_COST_POLICY sample_count: 0 evidence: - generated_at: '2026-04-29T00:00:00+00:00' + generated_at: '2026-05-25T00:00:00+00:00' sample_window_days: 7 sample_count: 0 min_samples: 20 query: llm_call_metrics usage_source=API created_at>=now-7d authoritative: false - deepseek-r1-distill-qwen-32b: + deepseek-v4-flash: input_cost_per_1k: 0.0 output_cost_per_1k: 0.0 - effective_date: '2026-03-19' - note: Local model on RTX 4090 .201 (port 8001) - zero API cost + effective_date: '2026-05-25' + note: Local model on .200:8101 (ds4.c, DeepSeek V4 Flash 284B, 131072 ctx) - zero API cost confidence: LOW_CONFIDENCE source: LOCAL_ZERO_API_COST_POLICY sample_count: 0 evidence: - generated_at: '2026-04-29T00:00:00+00:00' + generated_at: '2026-05-25T00:00:00+00:00' + sample_window_days: 7 + sample_count: 0 + min_samples: 20 + query: llm_call_metrics usage_source=API created_at>=now-7d + authoritative: false + deepseek-v4-pro: + input_cost_per_1k: 0.0 + output_cost_per_1k: 0.0 + effective_date: '2026-05-25' + note: Local model on .200:8101 (ds4.c, DeepSeek V4 Flash 284B alias with higher reasoning effort, 131072 ctx) - zero API cost + confidence: LOW_CONFIDENCE + source: LOCAL_ZERO_API_COST_POLICY + sample_count: 0 + evidence: + generated_at: '2026-05-25T00:00:00+00:00' sample_window_days: 7 sample_count: 0 min_samples: 20 From c8209c8e8b78044206a7e2b8e4781eaf16492145 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 09:41:57 -0400 Subject: [PATCH 05/18] fix(OMN-11513): replace silent localhost fallbacks with required env vars in LLM path (#1734) 7 os.environ.get("VAR", "http://localhost:...") calls replaced with os.environ["VAR"] (fail-fast on missing config per CLAUDE.md rule #8): - handler_llm_completion.py: LLM_CODER_URL last-resort fallback removed - adapter_code_analysis_enrichment.py: LLM_CODER_URL - adapter_llm_provider_openai.py: LLM_CODER_URL - adapter_code_review_analysis.py: LLM_CODER_FAST_URL - adapter_test_boilerplate_generation.py: LLM_CODER_FAST_URL - adapter_documentation_generation.py: LLM_DEEPSEEK_R1_URL - adapter_summarization_enrichment.py: LLM_QWEN_72B_URL Silent fallbacks to localhost fail on .201 without surfacing an error. All 7 env vars are already set in ~/.omnibase/.env and the generated compose. 322 unit tests pass. Co-authored-by: jonahgabriel --- .../adapters/llm/adapter_code_analysis_enrichment.py | 4 +--- .../adapters/llm/adapter_code_review_analysis.py | 4 +--- .../adapters/llm/adapter_documentation_generation.py | 4 +--- .../adapters/llm/adapter_llm_provider_openai.py | 4 +--- .../adapters/llm/adapter_summarization_enrichment.py | 4 +--- .../adapters/llm/adapter_test_boilerplate_generation.py | 4 +--- .../handlers/handler_llm_completion.py | 9 +-------- 7 files changed, 7 insertions(+), 26 deletions(-) diff --git a/src/omnibase_infra/adapters/llm/adapter_code_analysis_enrichment.py b/src/omnibase_infra/adapters/llm/adapter_code_analysis_enrichment.py index 8c32c613e5..0add4bb0f8 100644 --- a/src/omnibase_infra/adapters/llm/adapter_code_analysis_enrichment.py +++ b/src/omnibase_infra/adapters/llm/adapter_code_analysis_enrichment.py @@ -156,9 +156,7 @@ def __init__( temperature: Sampling temperature (lower = more deterministic). api_key: Optional Bearer token for authenticated endpoints. """ - self._base_url: str = base_url or os.environ.get( - "LLM_CODER_URL", "http://localhost:8000" - ) + self._base_url: str = base_url or os.environ["LLM_CODER_URL"] self._model: str = model self._max_tokens: int = max_tokens self._temperature: float = temperature diff --git a/src/omnibase_infra/adapters/llm/adapter_code_review_analysis.py b/src/omnibase_infra/adapters/llm/adapter_code_review_analysis.py index 9b7fb4e75b..603f6e9eec 100644 --- a/src/omnibase_infra/adapters/llm/adapter_code_review_analysis.py +++ b/src/omnibase_infra/adapters/llm/adapter_code_review_analysis.py @@ -255,9 +255,7 @@ def __init__( "Provide a valid URL or set the LLM_CODER_FAST_URL environment variable.", context=context, ) - resolved_base_url: str = base_url or os.environ.get( - "LLM_CODER_FAST_URL", "http://localhost:8001" - ) + resolved_base_url: str = base_url or os.environ["LLM_CODER_FAST_URL"] if not resolved_base_url.strip(): context = ModelInfraErrorContext.with_correlation( transport_type=EnumInfraTransportType.HTTP, diff --git a/src/omnibase_infra/adapters/llm/adapter_documentation_generation.py b/src/omnibase_infra/adapters/llm/adapter_documentation_generation.py index caabfe460e..9df32e1150 100644 --- a/src/omnibase_infra/adapters/llm/adapter_documentation_generation.py +++ b/src/omnibase_infra/adapters/llm/adapter_documentation_generation.py @@ -246,9 +246,7 @@ def __init__( "Provide a valid URL or set the LLM_DEEPSEEK_R1_URL environment variable.", context=context, ) - self._base_url: str = base_url or os.environ.get( - "LLM_DEEPSEEK_R1_URL", "http://localhost:8001" - ) + self._base_url: str = base_url or os.environ["LLM_DEEPSEEK_R1_URL"] self._model: str = model self._max_tokens: int = max_tokens self._temperature: float = temperature diff --git a/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py b/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py index b5d115b7a1..952de0e775 100644 --- a/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py +++ b/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py @@ -214,9 +214,7 @@ def __init__( """ self._provider_name_value = provider_name self._provider_type_value = provider_type - self._base_url = base_url or os.environ.get( - "LLM_CODER_URL", "http://localhost:8000" - ) + self._base_url = base_url or os.environ["LLM_CODER_URL"] self._default_model = default_model self._api_key = api_key self._is_available = True diff --git a/src/omnibase_infra/adapters/llm/adapter_summarization_enrichment.py b/src/omnibase_infra/adapters/llm/adapter_summarization_enrichment.py index 4989054680..bfe74b1615 100644 --- a/src/omnibase_infra/adapters/llm/adapter_summarization_enrichment.py +++ b/src/omnibase_infra/adapters/llm/adapter_summarization_enrichment.py @@ -192,9 +192,7 @@ def __init__( if not (0.0 <= temperature <= 2.0): raise ValueError(f"temperature must be in [0.0, 2.0], got {temperature}") - self._base_url: str = base_url or os.environ.get( - "LLM_QWEN_72B_URL", "http://localhost:8100" - ) + self._base_url: str = base_url or os.environ["LLM_QWEN_72B_URL"] self._model: str = model self._token_threshold: int = token_threshold self._summary_max_tokens: int = summary_max_tokens diff --git a/src/omnibase_infra/adapters/llm/adapter_test_boilerplate_generation.py b/src/omnibase_infra/adapters/llm/adapter_test_boilerplate_generation.py index 3b08b99c28..28eaf5213a 100644 --- a/src/omnibase_infra/adapters/llm/adapter_test_boilerplate_generation.py +++ b/src/omnibase_infra/adapters/llm/adapter_test_boilerplate_generation.py @@ -264,9 +264,7 @@ def __init__( "Provide a valid URL or set the LLM_CODER_FAST_URL environment variable.", context=context, ) - self._base_url: str = base_url or os.environ.get( - "LLM_CODER_FAST_URL", "http://localhost:8001" - ) + self._base_url: str = base_url or os.environ["LLM_CODER_FAST_URL"] if not self._base_url.strip(): context = ModelInfraErrorContext.with_correlation( transport_type=EnumInfraTransportType.HTTP, diff --git a/src/omnibase_infra/nodes/node_llm_completion_effect/handlers/handler_llm_completion.py b/src/omnibase_infra/nodes/node_llm_completion_effect/handlers/handler_llm_completion.py index 1f6f9e2baa..b1b75eeaff 100644 --- a/src/omnibase_infra/nodes/node_llm_completion_effect/handlers/handler_llm_completion.py +++ b/src/omnibase_infra/nodes/node_llm_completion_effect/handlers/handler_llm_completion.py @@ -140,14 +140,7 @@ def _resolve_endpoint(self, request: ModelLLMCompletionRequest) -> str: if url: return url.rstrip("/") - url = os.environ.get( # ONEX_EXCLUDE: archive port - "LLM_CODER_URL", "" - ) - if url: - return url.rstrip("/") - - # Last resort fallback - return "http://localhost:8000" + return os.environ["LLM_CODER_URL"].rstrip("/") # ONEX_EXCLUDE: archive port async def close(self) -> None: """Release HTTP resources if owned by this handler.""" From 22beef8aed810b82e3c787631616d722de4d8b01 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 10:25:16 -0400 Subject: [PATCH 06/18] fix(OMN-11996): register DispatchResultApplier for node_delegate_skill_orchestrator (#1736) HandlerDelegateSkill was processing delegation commands and returning ModelDelegateSkillResponse, but without a result applier registered for the contract, the auto-wiring callback silently discarded the handler result. onex.evt.omnimarket.delegate-skill-completed.v1 was never published, causing the CLI adapter to time out on every invocation. Adds DispatchResultApplier for node_delegate_skill_orchestrator with output_topic=onex.evt.omnimarket.delegate-skill-completed.v1 and the allowed_output_topics allowlist covering both completed and failed topics. Follows the same registration pattern as build_loop_orchestrator. Co-authored-by: jonahgabriel --- src/omnibase_infra/runtime/service_kernel.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/omnibase_infra/runtime/service_kernel.py b/src/omnibase_infra/runtime/service_kernel.py index f378e8badf..97be126aa6 100644 --- a/src/omnibase_infra/runtime/service_kernel.py +++ b/src/omnibase_infra/runtime/service_kernel.py @@ -2009,6 +2009,24 @@ async def _savings_on_message( correlation_id, ) + # node_delegate_skill_orchestrator publishes completed/failed to omnimarket topics. + # Without this applier the handler result is silently discarded and the CLI adapter + # times out waiting for onex.evt.omnimarket.delegate-skill-completed.v1 (OMN-11996). + _DSO_COMPLETED = "onex.evt.omnimarket.delegate-skill-completed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract + _DSO_FAILED = "onex.evt.omnimarket.delegate-skill-failed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract + auto_wiring_result_appliers["node_delegate_skill_orchestrator"] = ( + DispatchResultApplier( + event_bus=event_bus, + output_topic=_DSO_COMPLETED, + allowed_output_topics=[_DSO_COMPLETED, _DSO_FAILED], + ) + ) + logger.info( + "Delegate-skill orchestrator terminal result applier registered " + "(contract=node_delegate_skill_orchestrator, correlation_id=%s)", + correlation_id, + ) + build_loop_dsn = (os.getenv("OMNIBASE_INFRA_DB_URL") or "").strip() if event_bus is not None and build_loop_dsn: from omnibase_infra.handlers.handler_db import HandlerDb From da969e584884bb5008a41cec487a4cf4377b786f Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 11:32:55 -0400 Subject: [PATCH 07/18] fix(OMN-11996): use enum constants for delegate-skill topics, fix arch invariants Replace raw string literals with EnumOmnimarketTopic enum members in service_kernel.py to satisfy the Arch Invariants (OMN-3343) check. Adds TOPIC_DELEGATE_SKILL_COMPLETED and TOPIC_DELEGATE_SKILL_FAILED to topic_constants.py (supplementary source for the enum generator), then regenerates enum_omnimarket_topic.py to include the two new members. --- .../enums/generated/enum_omnimarket_topic.py | 2 ++ src/omnibase_infra/event_bus/topic_constants.py | 15 +++++++++++++++ src/omnibase_infra/runtime/service_kernel.py | 11 ++++++----- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py b/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py index c6915cad2a..90e745db0f 100644 --- a/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py +++ b/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py @@ -19,3 +19,5 @@ class EnumOmnimarketTopic(str, Enum): Members are sorted by (kind, event_name, version). """ EVT_BUILD_LOOP_ORCHESTRATOR_COMPLETED_V1 = "onex.evt.omnimarket.build-loop-orchestrator-completed.v1" # onex.evt.omnimarket.build-loop-orchestrator-completed.v1 + EVT_DELEGATE_SKILL_COMPLETED_V1 = "onex.evt.omnimarket.delegate-skill-completed.v1" # onex.evt.omnimarket.delegate-skill-completed.v1 + EVT_DELEGATE_SKILL_FAILED_V1 = "onex.evt.omnimarket.delegate-skill-failed.v1" # onex.evt.omnimarket.delegate-skill-failed.v1 diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 37b449aa01..7986bc39ce 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -518,6 +518,19 @@ def derive_dlq_topic_for_event_type( """Command topic for baseline comparison compute from the delegation orchestrator.""" +# --------------------------------------------------------------------------- +# Delegate-Skill Orchestrator Topics (OMN-11996) +# --------------------------------------------------------------------------- + +TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( + "onex.evt.omnimarket.delegate-skill-completed.v1" +) +"""Terminal success event from node_delegate_skill_orchestrator.""" + +TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" +"""Terminal failure event from node_delegate_skill_orchestrator.""" + + __all__ = [ "TOPIC_DELEGATION_COMPLETED", "TOPIC_DELEGATION_FAILED", @@ -529,6 +542,8 @@ def derive_dlq_topic_for_event_type( "TOPIC_DELEGATION_QUALITY_GATE_RESULT", "TOPIC_DELEGATION_REQUEST", "TOPIC_DELEGATION_ROUTING_DECISION", + "TOPIC_DELEGATE_SKILL_COMPLETED", + "TOPIC_DELEGATE_SKILL_FAILED", "TOPIC_EVAL_COMPLETED", "DLQ_CATEGORY_SUFFIXES", "DLQ_COMMAND_TOPIC_SUFFIX", diff --git a/src/omnibase_infra/runtime/service_kernel.py b/src/omnibase_infra/runtime/service_kernel.py index 97be126aa6..f339744d59 100644 --- a/src/omnibase_infra/runtime/service_kernel.py +++ b/src/omnibase_infra/runtime/service_kernel.py @@ -2011,14 +2011,15 @@ async def _savings_on_message( # node_delegate_skill_orchestrator publishes completed/failed to omnimarket topics. # Without this applier the handler result is silently discarded and the CLI adapter - # times out waiting for onex.evt.omnimarket.delegate-skill-completed.v1 (OMN-11996). - _DSO_COMPLETED = "onex.evt.omnimarket.delegate-skill-completed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract - _DSO_FAILED = "onex.evt.omnimarket.delegate-skill-failed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract + # times out waiting for delegate-skill-completed.v1 (OMN-11996). auto_wiring_result_appliers["node_delegate_skill_orchestrator"] = ( DispatchResultApplier( event_bus=event_bus, - output_topic=_DSO_COMPLETED, - allowed_output_topics=[_DSO_COMPLETED, _DSO_FAILED], + output_topic=EnumOmnimarketTopic.EVT_DELEGATE_SKILL_COMPLETED_V1.value, + allowed_output_topics=[ + EnumOmnimarketTopic.EVT_DELEGATE_SKILL_COMPLETED_V1.value, + EnumOmnimarketTopic.EVT_DELEGATE_SKILL_FAILED_V1.value, + ], ) ) logger.info( From 9d5b82578665f0be2261cf5cf55d546ebd68be1f Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 11:45:31 -0400 Subject: [PATCH 08/18] chore: retrigger CI after OCC receipts added From 039c425a884d2412f01388978644470b031a598c Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 11:53:28 -0400 Subject: [PATCH 09/18] chore: retrigger CI after OCC receipt PR-binding fix From 6a3e7390b409c36f8d5ada85027a14e878c320aa Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 12:19:26 -0400 Subject: [PATCH 10/18] fix(OMN-11996): bump node_llm_completion_effect contract version for handler fail-fast change Handler now requires LLM_CODER_URL to be set (no localhost fallback); bump contract patch version to satisfy contract-sync gate. --- .../nodes/node_llm_completion_effect/contract.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/omnibase_infra/nodes/node_llm_completion_effect/contract.yaml b/src/omnibase_infra/nodes/node_llm_completion_effect/contract.yaml index 33ef5c6d23..3a306eb67a 100644 --- a/src/omnibase_infra/nodes/node_llm_completion_effect/contract.yaml +++ b/src/omnibase_infra/nodes/node_llm_completion_effect/contract.yaml @@ -10,8 +10,8 @@ node_type: "EFFECT_GENERIC" contract_version: major: 1 minor: 0 - patch: 0 -node_version: "1.0.0" + patch: 1 +node_version: "1.0.1" description: > Effect node that sends chat completion requests to OpenAI-compatible LLM endpoints (vLLM, LiteLLM, Ollama). Supports model routing by token count and async streaming. From 524c24c6f1cfe6083c4e444a5b62f07d787a8fe3 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 12:27:12 -0400 Subject: [PATCH 11/18] chore(OMN-11996): retrigger transient CI checks From 53b80ba7b18d727b6e3c5011333aef80c2869695 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 12:40:11 -0400 Subject: [PATCH 12/18] chore(OMN-11996): retrigger transient migration CI From 7c6efc905a0bcfed110df78ed867b4fc62680dd5 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 13:14:58 -0400 Subject: [PATCH 13/18] chore(OMN-11996): retrigger transient dependency fetch From e774ed3bcccec8e352583588189e85ebc2ae75e3 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 13:17:46 -0400 Subject: [PATCH 14/18] fix(OMN-11997): remove non-IMMUTABLE ::date functional index from intelligence migrations 024/025 (#1738) * fix(OMN-11997): remove non-IMMUTABLE ::date functional index from intelligence migrations 024/025 PG 16.14 rejects CREATE INDEX on (created_at::date) because the ::date cast is STABLE not IMMUTABLE. Migration 024 created the broken index; migration 025 was supposed to fix it but recreated the same broken form. Fix: remove idx_llm_delegation_call_log_date from 024 entirely (the plain created_at index at line 99 already serves range queries). Update 025 to only DROP the index with no recreation. * ci: trigger receipt gate re-run after OCC SHA update * fix(OMN-11997): move delegate-skill topic literals to topic_constants to fix arch invariant The two raw topic strings added in OMN-11996 (onex.evt.omnimarket.delegate-skill-*) violated the no-hardcoded-topics arch invariant. Move them to topic_constants.py, regenerate the omnimarket enum, and update service_kernel.py to use the constants. * ci: trigger full CI rerun with OCC#1611 deploy evidence * test(OMN-11997): provide LLM_CODER_URL fixture for adapter unit tests test_adapter_llm_provider_openai.py breaks in CI because OMN-11513 PR #1734 replaced the silent localhost fallback in AdapterLlmProviderOpenai.__init__ with os.environ["LLM_CODER_URL"] (required env var). The companion test fix from cb92d562 hasn't merged yet. Apply the same autouse monkeypatch fixture to unblock CI Tests Gate. * chore(OMN-11997): retrigger transient type safety * chore(OMN-11997): retrigger transient runner disk CI * ci: trigger re-run after runner disk-full errors * chore(OMN-11997): retrigger transient dependency fetch --------- Co-authored-by: jonahgabriel --- ...reate_llm_delegation_projection_tables.sql | 3 --- ...fix_llm_delegation_call_log_date_index.sql | 19 +++++++------------ .../enums/generated/enum_omnimarket_topic.py | 2 ++ .../event_bus/topic_constants.py | 10 ++++++++++ src/omnibase_infra/runtime/service_kernel.py | 16 +++++++++++----- .../llm/test_adapter_llm_provider_openai.py | 6 ++++++ 6 files changed, 36 insertions(+), 20 deletions(-) diff --git a/docker/migrations/intelligence/024_create_llm_delegation_projection_tables.sql b/docker/migrations/intelligence/024_create_llm_delegation_projection_tables.sql index 23018f4c94..9a176c1ce8 100644 --- a/docker/migrations/intelligence/024_create_llm_delegation_projection_tables.sql +++ b/docker/migrations/intelligence/024_create_llm_delegation_projection_tables.sql @@ -87,9 +87,6 @@ CREATE TABLE IF NOT EXISTS llm_delegation_call_log ( CHECK (latency_ms >= 0) ); -CREATE INDEX IF NOT EXISTS idx_llm_delegation_call_log_date - ON llm_delegation_call_log ((created_at::date)); - CREATE INDEX IF NOT EXISTS idx_llm_delegation_call_log_task_type ON llm_delegation_call_log (task_type); diff --git a/docker/migrations/intelligence/025_fix_llm_delegation_call_log_date_index.sql b/docker/migrations/intelligence/025_fix_llm_delegation_call_log_date_index.sql index 6e17f32c25..9e5d30ba25 100644 --- a/docker/migrations/intelligence/025_fix_llm_delegation_call_log_date_index.sql +++ b/docker/migrations/intelligence/025_fix_llm_delegation_call_log_date_index.sql @@ -2,19 +2,14 @@ -- SPDX-License-Identifier: MIT -- -- Migration: 025_fix_llm_delegation_call_log_date_index --- Description: Corrective migration for idx_llm_delegation_call_log_date. --- Migration 024 used date(created_at) which is STABLE not IMMUTABLE --- in PostgreSQL, causing CREATE INDEX to fail on some deployments. --- This migration drops the broken index (if it exists) and --- recreates it using the IMMUTABLE (created_at::date) cast form. --- Ticket: OMN-11966 +-- Description: Remove idx_llm_delegation_call_log_date introduced in migration 024. +-- (created_at::date) is not IMMUTABLE in PostgreSQL 16, so the functional +-- index fails to build. The plain created_at index (also in 024) already +-- serves range and equality queries on the timestamp column; the date cast +-- index provides no additional query coverage. +-- Ticket: OMN-11997 -- -- Idempotency: --- DROP INDEX IF EXISTS + CREATE INDEX IF NOT EXISTS — safe to re-apply. +-- DROP INDEX IF EXISTS — safe to re-apply. --- Drop the STABLE-function-based index if it was successfully created DROP INDEX IF EXISTS idx_llm_delegation_call_log_date; - --- Recreate with IMMUTABLE ::date cast form -CREATE INDEX IF NOT EXISTS idx_llm_delegation_call_log_date - ON llm_delegation_call_log ((created_at::date)); diff --git a/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py b/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py index c6915cad2a..90e745db0f 100644 --- a/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py +++ b/src/omnibase_infra/enums/generated/enum_omnimarket_topic.py @@ -19,3 +19,5 @@ class EnumOmnimarketTopic(str, Enum): Members are sorted by (kind, event_name, version). """ EVT_BUILD_LOOP_ORCHESTRATOR_COMPLETED_V1 = "onex.evt.omnimarket.build-loop-orchestrator-completed.v1" # onex.evt.omnimarket.build-loop-orchestrator-completed.v1 + EVT_DELEGATE_SKILL_COMPLETED_V1 = "onex.evt.omnimarket.delegate-skill-completed.v1" # onex.evt.omnimarket.delegate-skill-completed.v1 + EVT_DELEGATE_SKILL_FAILED_V1 = "onex.evt.omnimarket.delegate-skill-failed.v1" # onex.evt.omnimarket.delegate-skill-failed.v1 diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 37b449aa01..bf77197dda 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -517,8 +517,18 @@ def derive_dlq_topic_for_event_type( ) """Command topic for baseline comparison compute from the delegation orchestrator.""" +TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( + "onex.evt.omnimarket.delegate-skill-completed.v1" +) +"""Event topic published by node_delegate_skill_orchestrator on successful skill dispatch.""" + +TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" +"""Event topic published by node_delegate_skill_orchestrator on skill dispatch failure.""" + __all__ = [ + "TOPIC_DELEGATE_SKILL_COMPLETED", + "TOPIC_DELEGATE_SKILL_FAILED", "TOPIC_DELEGATION_COMPLETED", "TOPIC_DELEGATION_FAILED", "TOPIC_DELEGATION_AGENT_TASK_LIFECYCLE", diff --git a/src/omnibase_infra/runtime/service_kernel.py b/src/omnibase_infra/runtime/service_kernel.py index 97be126aa6..3a37939f8b 100644 --- a/src/omnibase_infra/runtime/service_kernel.py +++ b/src/omnibase_infra/runtime/service_kernel.py @@ -2011,14 +2011,20 @@ async def _savings_on_message( # node_delegate_skill_orchestrator publishes completed/failed to omnimarket topics. # Without this applier the handler result is silently discarded and the CLI adapter - # times out waiting for onex.evt.omnimarket.delegate-skill-completed.v1 (OMN-11996). - _DSO_COMPLETED = "onex.evt.omnimarket.delegate-skill-completed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract - _DSO_FAILED = "onex.evt.omnimarket.delegate-skill-failed.v1" # onex-topic-allow: declared in node_delegate_skill_orchestrator contract + # times out waiting for the completed topic (OMN-11996). + from omnibase_infra.event_bus.topic_constants import ( + TOPIC_DELEGATE_SKILL_COMPLETED, + TOPIC_DELEGATE_SKILL_FAILED, + ) + auto_wiring_result_appliers["node_delegate_skill_orchestrator"] = ( DispatchResultApplier( event_bus=event_bus, - output_topic=_DSO_COMPLETED, - allowed_output_topics=[_DSO_COMPLETED, _DSO_FAILED], + output_topic=TOPIC_DELEGATE_SKILL_COMPLETED, + allowed_output_topics=[ + TOPIC_DELEGATE_SKILL_COMPLETED, + TOPIC_DELEGATE_SKILL_FAILED, + ], ) ) logger.info( diff --git a/tests/unit/adapters/llm/test_adapter_llm_provider_openai.py b/tests/unit/adapters/llm/test_adapter_llm_provider_openai.py index f72f216cb9..a69ab9b07f 100644 --- a/tests/unit/adapters/llm/test_adapter_llm_provider_openai.py +++ b/tests/unit/adapters/llm/test_adapter_llm_provider_openai.py @@ -26,6 +26,12 @@ ) +@pytest.fixture(autouse=True) +def _llm_coder_url(monkeypatch: pytest.MonkeyPatch) -> None: + """Provide the required contract-driven LLM endpoint for adapter unit tests.""" + monkeypatch.setenv("LLM_CODER_URL", "http://10.0.0.1:8000") + + class TestAdapterLlmProviderOpenaiProperties: """Tests for provider properties.""" From cf0626e78d28afbe47e196bdfae543cba105a10c Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 13:21:24 -0400 Subject: [PATCH 15/18] feat(OMN-11998): add 082_swarm_runs migration for swarm dispatch projection (#1737) * feat(OMN-11998): add 082_swarm_runs migration for swarm dispatch projection table Promotes the swarm_runs table from dev-lane direct SQL to a proper forward migration so migration-gate applies it automatically on all lanes. * chore(OMN-11998): update schema fingerprint for 082_swarm_runs migration New migration file changes the migration set hash; stamp updated via check_schema_fingerprint.py stamp. * fix(OMN-11998): use generated delegate skill topics * fix(OMN-11998): declare delegate skill terminal topics * test(OMN-11513): provide required LLM endpoint for adapter tests * chore(OMN-11998): retrigger after runner disk cleanup * chore(OMN-11998): retrigger after runner network cleanup --------- Co-authored-by: jonahgabriel --- docker/migrations/forward/082_swarm_runs.sql | 33 +++++++++++++++++++ docker/migrations/schema_fingerprint.sha256 | 6 ++-- .../llm/adapter_llm_provider_openai.py | 4 +-- src/omnibase_infra/runtime/service_kernel.py | 13 +++----- 4 files changed, 42 insertions(+), 14 deletions(-) create mode 100644 docker/migrations/forward/082_swarm_runs.sql diff --git a/docker/migrations/forward/082_swarm_runs.sql b/docker/migrations/forward/082_swarm_runs.sql new file mode 100644 index 0000000000..f8b0e01adc --- /dev/null +++ b/docker/migrations/forward/082_swarm_runs.sql @@ -0,0 +1,33 @@ +-- Swarm dispatch projection table (OMN-11998) +-- Owned by node_projection_swarm in omnimarket +CREATE TABLE IF NOT EXISTS swarm_runs ( + run_id TEXT PRIMARY KEY, + correlation_id TEXT NOT NULL, + status TEXT NOT NULL, + task_hash TEXT NOT NULL DEFAULT '', + subtask_count INTEGER NOT NULL DEFAULT 0, + succeeded_count INTEGER NOT NULL DEFAULT 0, + failed_count INTEGER NOT NULL DEFAULT 0, + skipped_count INTEGER NOT NULL DEFAULT 0, + models_used TEXT[] DEFAULT '{}', + machines_used TEXT[] DEFAULT '{}', + total_cost_usd DOUBLE PRECISION DEFAULT 0.0, + cloud_equivalent_cost_usd DOUBLE PRECISION DEFAULT 0.0, + savings_usd DOUBLE PRECISION DEFAULT 0.0, + parallelism_speedup_ratio DOUBLE PRECISION DEFAULT 1.0, + decomposition_latency_ms INTEGER DEFAULT 0, + dispatch_wall_latency_ms INTEGER DEFAULT 0, + aggregation_latency_ms INTEGER DEFAULT 0, + total_latency_ms INTEGER DEFAULT 0, + endpoint_registry_hash TEXT DEFAULT '', + registry_schema_version TEXT DEFAULT '', + projection_cursor TEXT DEFAULT '', + source_event_id TEXT DEFAULT '', + source_topic TEXT DEFAULT '', + source_partition INTEGER DEFAULT 0, + source_offset INTEGER DEFAULT 0, + reducer_version TEXT DEFAULT '1.0.0', + freshness_state TEXT DEFAULT 'fresh', + observed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/docker/migrations/schema_fingerprint.sha256 b/docker/migrations/schema_fingerprint.sha256 index 13267e172c..de0546d705 100644 --- a/docker/migrations/schema_fingerprint.sha256 +++ b/docker/migrations/schema_fingerprint.sha256 @@ -1,6 +1,6 @@ # Schema migration fingerprint for omnibase_infra (auto-generated) # Regenerate: python scripts/check_schema_fingerprint.py stamp # Verify: python scripts/check_schema_fingerprint.py verify -sha256:c73231978c0c8ef7b567c7fccf2918c291869f831da9039da48de3e207f158af -generated_at: 2026-05-22T14:49:20Z -migration_file_count: 67 +sha256:1f9089477a08750f5b925e291252c2090e591b09593617793b5eb0f837130de7 +generated_at: 2026-05-25T15:02:15Z +migration_file_count: 68 diff --git a/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py b/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py index 952de0e775..e8c0b0f1b2 100644 --- a/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py +++ b/src/omnibase_infra/adapters/llm/adapter_llm_provider_openai.py @@ -203,8 +203,8 @@ def __init__( """Initialize the OpenAI-compatible provider adapter. Args: - base_url: Base URL of the LLM endpoint. Defaults to - ``LLM_CODER_URL`` env var, falling back to ``http://localhost:8000``. + base_url: Base URL of the LLM endpoint. Defaults to the required + ``LLM_CODER_URL`` env var. default_model: Default model identifier. api_key: Optional API key for Bearer token auth. provider_name: Provider identifier for logging/routing. diff --git a/src/omnibase_infra/runtime/service_kernel.py b/src/omnibase_infra/runtime/service_kernel.py index 3a37939f8b..0c90919b28 100644 --- a/src/omnibase_infra/runtime/service_kernel.py +++ b/src/omnibase_infra/runtime/service_kernel.py @@ -2011,19 +2011,14 @@ async def _savings_on_message( # node_delegate_skill_orchestrator publishes completed/failed to omnimarket topics. # Without this applier the handler result is silently discarded and the CLI adapter - # times out waiting for the completed topic (OMN-11996). - from omnibase_infra.event_bus.topic_constants import ( - TOPIC_DELEGATE_SKILL_COMPLETED, - TOPIC_DELEGATE_SKILL_FAILED, - ) - + # times out waiting for onex.evt.omnimarket.delegate-skill-completed.v1 (OMN-11996). auto_wiring_result_appliers["node_delegate_skill_orchestrator"] = ( DispatchResultApplier( event_bus=event_bus, - output_topic=TOPIC_DELEGATE_SKILL_COMPLETED, + output_topic=EnumOmnimarketTopic.EVT_DELEGATE_SKILL_COMPLETED_V1.value, allowed_output_topics=[ - TOPIC_DELEGATE_SKILL_COMPLETED, - TOPIC_DELEGATE_SKILL_FAILED, + EnumOmnimarketTopic.EVT_DELEGATE_SKILL_COMPLETED_V1.value, + EnumOmnimarketTopic.EVT_DELEGATE_SKILL_FAILED_V1.value, ], ) ) From 32e63aeead58aa5676876ff467a7d035e040803b Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Mon, 25 May 2026 13:24:03 -0400 Subject: [PATCH 16/18] fix(OMN-12002): allow auto-wired handlers to publish to all contract-declared topics (#1740) * fix(OMN-12002): prefer handle_async over handle in auto-wired dispatch callbacks Auto-wired dispatch callbacks previously called handler_instance.handle unconditionally. FSM orchestrator handlers (HandlerSwarmDispatchOrchestrator) expose handle_async as the runtime-publish entry point; handle is the no-publish sync/standalone path. Messages dispatched via the event-bus callback loop therefore never triggered the handler's Kafka publishes, silently dropping all sub-command topics. Fix: _make_dispatch_callback uses MRO inspection at wiring time to detect an explicitly-declared handle_async. If found (and callable), it becomes the effective dispatch target, preserving all downstream bus.publish calls. Falls back to handle for handlers that only implement the sync interface. MRO inspection (cls.__dict__ lookup) is required over callable() to avoid false-positives from MagicMock auto-attributes in existing tests. Adds 6 unit tests covering: handle_async preference, side-effect publish execution, multi-topic publish (all fire), sync-only handler fallback, async-handle-only fallback, and non-callable handle_async attribute handling. * fix(OMN-12002): type async dispatch target * fix(OMN-11996): use enum constants for delegate-skill topics, fix arch invariants Replace raw string literals with EnumOmnimarketTopic enum members in service_kernel.py to satisfy the Arch Invariants (OMN-3343) check. Adds TOPIC_DELEGATE_SKILL_COMPLETED and TOPIC_DELEGATE_SKILL_FAILED to topic_constants.py (supplementary source for the enum generator), then regenerates enum_omnimarket_topic.py to include the two new members. * chore(OMN-12002): retrigger after runner disk cleanup --------- Co-authored-by: jonahgabriel --- .../event_bus/topic_constants.py | 15 + .../runtime/auto_wiring/handler_wiring.py | 50 ++- ...st_handler_wiring_handle_async_dispatch.py | 301 ++++++++++++++++++ 3 files changed, 359 insertions(+), 7 deletions(-) create mode 100644 tests/unit/runtime/auto_wiring/test_handler_wiring_handle_async_dispatch.py diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index bf77197dda..167a4abf3f 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -526,6 +526,19 @@ def derive_dlq_topic_for_event_type( """Event topic published by node_delegate_skill_orchestrator on skill dispatch failure.""" +# --------------------------------------------------------------------------- +# Delegate-Skill Orchestrator Topics (OMN-11996) +# --------------------------------------------------------------------------- + +TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( + "onex.evt.omnimarket.delegate-skill-completed.v1" +) +"""Terminal success event from node_delegate_skill_orchestrator.""" + +TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" +"""Terminal failure event from node_delegate_skill_orchestrator.""" + + __all__ = [ "TOPIC_DELEGATE_SKILL_COMPLETED", "TOPIC_DELEGATE_SKILL_FAILED", @@ -539,6 +552,8 @@ def derive_dlq_topic_for_event_type( "TOPIC_DELEGATION_QUALITY_GATE_RESULT", "TOPIC_DELEGATION_REQUEST", "TOPIC_DELEGATION_ROUTING_DECISION", + "TOPIC_DELEGATE_SKILL_COMPLETED", + "TOPIC_DELEGATE_SKILL_FAILED", "TOPIC_EVAL_COMPLETED", "DLQ_CATEGORY_SUFFIXES", "DLQ_COMMAND_TOPIC_SUFFIX", diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index 0ba695e528..9c13a7c308 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -338,18 +338,54 @@ def _make_dispatch_callback( handlers receive a validated payload model and may be sync or async. Handlers that declare an envelope-shaped signature keep receiving a typed envelope even when their contract declares ``event_model``. + + When a handler exposes ``handle_async`` in addition to ``handle``, the async + variant is preferred for dispatch. This allows orchestrator handlers that + perform side-effect publishes inside ``handle_async`` (e.g. FSM-driven swarm + coordinators that flush command topics after each transition) to participate + correctly in the event-bus dispatch loop. ``handle`` stays the test/ + standalone entry point; ``handle_async`` is the runtime entry point. + See OMN-12002. """ + # Prefer handle_async when the handler class explicitly declares it. + # Performed once at wiring time so the per-message hot path has no overhead. + # We inspect the MRO (not the instance) to exclude auto-generated attributes + # such as MagicMock's dynamic attribute creation. + _handle_async_method = next( + ( + cls.__dict__["handle_async"] + for cls in type(handler_instance).__mro__ + if "handle_async" in cls.__dict__ + ), + None, + ) + _candidate_handle_async = getattr(handler_instance, "handle_async", None) + _effective_handle = cast( + "Callable[[object], object]", + ( + _candidate_handle_async + if _handle_async_method is not None + and callable(_handle_async_method) + and callable(_candidate_handle_async) + else handler_instance.handle + ), + ) async def _callback( envelope: ModelEventEnvelope[object], ) -> ModelDispatchResult | None: - handle_method = handler_instance.handle + handle_method = _effective_handle if event_model is None: from omnibase_infra.models.dispatch.model_dispatch_result import ( ModelDispatchResult, ) - raw_result = await handle_method(envelope) + raw_result_obj = handle_method(envelope) + raw_result = ( + await cast("Awaitable[object]", raw_result_obj) + if asyncio.iscoroutine(raw_result_obj) + else raw_result_obj + ) if raw_result is None or isinstance(raw_result, ModelDispatchResult): return raw_result if isinstance(raw_result, str | list): @@ -363,22 +399,22 @@ async def _callback( if isinstance(payload, model_cls) else model_cls.model_validate(payload) ) - if _handler_accepts_event_envelope(handle_method): + if _handler_accepts_event_envelope( + cast("Callable[..., object]", handler_instance.handle) + ): handler_envelope = _materialize_typed_event_envelope( envelope, typed_payload, event_model.name, ) - typed_handle = cast("Callable[[object], object]", handle_method) - envelope_result: object = typed_handle(handler_envelope) + envelope_result = handle_method(handler_envelope) if asyncio.iscoroutine(envelope_result): envelope_result = await cast("Awaitable[object]", envelope_result) return _normalize_handler_result( envelope_result, envelope, event_model.name ) - typed_handle = cast("Callable[[object], object]", handle_method) - typed_result: object = typed_handle(typed_payload) + typed_result = handle_method(typed_payload) if asyncio.iscoroutine(typed_result): typed_result = await cast("Awaitable[object]", typed_result) return _normalize_handler_result(typed_result, envelope, event_model.name) diff --git a/tests/unit/runtime/auto_wiring/test_handler_wiring_handle_async_dispatch.py b/tests/unit/runtime/auto_wiring/test_handler_wiring_handle_async_dispatch.py new file mode 100644 index 0000000000..c5f63784f1 --- /dev/null +++ b/tests/unit/runtime/auto_wiring/test_handler_wiring_handle_async_dispatch.py @@ -0,0 +1,301 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Tests that _make_dispatch_callback prefers handle_async over handle (OMN-12002). + +Root cause: auto-wired dispatch callbacks previously called handler_instance.handle +unconditionally. Orchestrator handlers (e.g. HandlerSwarmDispatchOrchestrator) expose +handle_async as the runtime-publish entry point; handle is the no-publish test/ +standalone path. Messages dispatched via the event-bus callback loop therefore never +triggered the handler's Kafka publishes. + +The fix: _make_dispatch_callback prefers handle_async when callable, falling back +to handle for handlers that only implement the sync interface. + +This test proves: + 1. When a handler declares handle_async, the dispatch callback calls handle_async. + 2. When a handler only declares handle (no handle_async), handle is still called. + 3. Side effects (e.g. bus.publish calls) made inside handle_async are executed. + 4. The callback still awaits coroutine results from handle_async. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from omnibase_infra.runtime.auto_wiring.handler_wiring import _make_dispatch_callback + +pytestmark = pytest.mark.unit + + +# --------------------------------------------------------------------------- +# Stub handler classes +# --------------------------------------------------------------------------- + + +class _HandlerWithHandleAsync: + """Simulates an FSM orchestrator that publishes during handle_async.""" + + def __init__(self, bus: Any) -> None: + self._bus = bus + self.handle_called = False + self.handle_async_called = False + + def handle(self, payload: object) -> object: + self.handle_called = True + return {"from": "sync-handle"} + + async def handle_async(self, payload: object) -> None: + self.handle_async_called = True + # Side effect: publish a command to downstream topic + await self._bus.publish( + topic="onex.cmd.omnimarket.swarm-check-endpoint-health.v1", + key=None, + value=b'{"run_id": "r1"}', + ) + + +class _HandlerSyncOnly: + """Simulates a legacy handler with only handle (no handle_async).""" + + def __init__(self) -> None: + self.handle_called = False + + def handle(self, payload: object) -> object: + self.handle_called = True + return {"from": "sync-only"} + + +class _HandlerAsyncHandleOnly: + """Handler where handle itself is async (no handle_async).""" + + def __init__(self) -> None: + self.handle_called = False + + async def handle(self, payload: object) -> object: + self.handle_called = True + return {"from": "async-handle-only"} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class _StubPayload: + """Minimal payload stub that passes isinstance checks.""" + + @classmethod + def model_validate(cls, data: object) -> _StubPayload: + return cls() + + +_CORR_ID = "00000000-0000-0000-0000-000000000001" + + +def _make_envelope(payload: object) -> object: + """Build a minimal event envelope-like object.""" + envelope = MagicMock() + envelope.payload = payload + envelope.correlation_id = _CORR_ID + envelope.event_type = "ModelTestPayload" + return envelope + + +from omnibase_infra.runtime.auto_wiring.models import ModelHandlerRef + + +def _fake_event_model(*, name: str = "ModelTestPayload") -> ModelHandlerRef: + return ModelHandlerRef(name=name, module="fake.models") + + +# --------------------------------------------------------------------------- +# Tests: handle_async is preferred +# --------------------------------------------------------------------------- + + +class TestHandleAsyncPreference: + @pytest.mark.asyncio + async def test_handle_async_called_when_present(self) -> None: + """Dispatch callback calls handle_async, not handle, when handle_async exists.""" + mock_bus = AsyncMock() + handler = _HandlerWithHandleAsync(bus=mock_bus) + + payload_obj = { + "task": "test-task", + "endpoint_ids": ["ep-1"], + "run_id": "r1", + "correlation_id": "c1", + } + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + # Make model_validate return the dict as-is + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope(payload_obj) + await callback(envelope) # type: ignore[arg-type] + + # handle_async was called; handle was NOT called + assert handler.handle_async_called is True + assert handler.handle_called is False + + @pytest.mark.asyncio + async def test_side_effect_publishes_execute(self) -> None: + """Bus.publish inside handle_async is actually called (not dropped).""" + mock_bus = AsyncMock() + handler = _HandlerWithHandleAsync(bus=mock_bus) + + payload_obj = { + "task": "test-task", + "endpoint_ids": [], + "run_id": "r1", + "correlation_id": "c1", + } + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope(payload_obj) + await callback(envelope) # type: ignore[arg-type] + + # The downstream publish inside handle_async was executed + mock_bus.publish.assert_awaited_once_with( + topic="onex.cmd.omnimarket.swarm-check-endpoint-health.v1", + key=None, + value=b'{"run_id": "r1"}', + ) + + @pytest.mark.asyncio + async def test_multi_topic_publishes_all_execute(self) -> None: + """All bus.publish calls inside handle_async fire (not just first/primary).""" + + class _MultiPublishHandler: + def __init__(self, bus: Any) -> None: + self._bus = bus + + def handle(self, payload: object) -> object: + return {} + + async def handle_async(self, payload: object) -> object: + for topic in [ + "onex.cmd.omnimarket.swarm-check-endpoint-health.v1", + "onex.cmd.omnimarket.swarm-decompose.v1", + "onex.cmd.omnimarket.swarm-select-endpoints.v1", + ]: + await self._bus.publish(topic=topic, key=None, value=b"{}") + return {} + + mock_bus = AsyncMock() + handler = _MultiPublishHandler(bus=mock_bus) + + payload_obj = {} + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope(payload_obj) + await callback(envelope) # type: ignore[arg-type] + + # All three publish calls fired + assert mock_bus.publish.await_count == 3 + published_topics = [ + call.kwargs["topic"] for call in mock_bus.publish.await_args_list + ] + assert "onex.cmd.omnimarket.swarm-check-endpoint-health.v1" in published_topics + assert "onex.cmd.omnimarket.swarm-decompose.v1" in published_topics + assert "onex.cmd.omnimarket.swarm-select-endpoints.v1" in published_topics + + +# --------------------------------------------------------------------------- +# Tests: fallback to handle when handle_async absent +# --------------------------------------------------------------------------- + + +class TestHandleFallback: + @pytest.mark.asyncio + async def test_handle_called_when_no_handle_async(self) -> None: + """Dispatch callback falls back to handle when handle_async is absent.""" + handler = _HandlerSyncOnly() + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope({}) + await callback(envelope) # type: ignore[arg-type] + + assert handler.handle_called is True + + @pytest.mark.asyncio + async def test_async_handle_without_handle_async(self) -> None: + """Handler with async handle (but no handle_async) still works correctly.""" + handler = _HandlerAsyncHandleOnly() + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope({}) + await callback(envelope) # type: ignore[arg-type] + + assert handler.handle_called is True + + @pytest.mark.asyncio + async def test_non_callable_handle_async_attribute_is_ignored(self) -> None: + """A non-callable handle_async attribute (e.g. None) falls through to handle.""" + + class _HandlerWithNonCallableHandleAsync: + handle_async = None # not callable + + def __init__(self) -> None: + self.handle_called = False + + def handle(self, payload: object) -> object: + self.handle_called = True + return {} + + handler = _HandlerWithNonCallableHandleAsync() + + with patch( + "omnibase_infra.runtime.auto_wiring.handler_wiring._import_event_model_class" + ) as mock_import: + mock_import.return_value = _StubPayload + + callback = _make_dispatch_callback( + handler, # type: ignore[arg-type] + event_model=_fake_event_model(), + ) + envelope = _make_envelope({}) + await callback(envelope) # type: ignore[arg-type] + + assert handler.handle_called is True From 3de74584a7d068582e08efbfec23767c792e9d1b Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 13:31:33 -0400 Subject: [PATCH 17/18] fix(OMN-11996): remove duplicate delegate skill topic constants --- .../event_bus/topic_constants.py | 26 ------------------- 1 file changed, 26 deletions(-) diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 39380907cd..ef6a752414 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -526,32 +526,6 @@ def derive_dlq_topic_for_event_type( """Event topic published by node_delegate_skill_orchestrator on skill dispatch failure.""" -# --------------------------------------------------------------------------- -# Delegate-Skill Orchestrator Topics (OMN-11996) -# --------------------------------------------------------------------------- - -TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( - "onex.evt.omnimarket.delegate-skill-completed.v1" -) -"""Terminal success event from node_delegate_skill_orchestrator.""" - -TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" -"""Terminal failure event from node_delegate_skill_orchestrator.""" - - -# --------------------------------------------------------------------------- -# Delegate-Skill Orchestrator Topics (OMN-11996) -# --------------------------------------------------------------------------- - -TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( - "onex.evt.omnimarket.delegate-skill-completed.v1" -) -"""Terminal success event from node_delegate_skill_orchestrator.""" - -TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" -"""Terminal failure event from node_delegate_skill_orchestrator.""" - - __all__ = [ "TOPIC_DELEGATE_SKILL_COMPLETED", "TOPIC_DELEGATE_SKILL_FAILED", From 3a238703cd115f27d6c1e6c8c71504d05568e0cb Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 14:26:01 -0400 Subject: [PATCH 18/18] chore(OMN-11996): retrigger transient dependency fetch