diff --git a/docs/evidence/dep-health-sweep/proof-of-life-001/findings.json b/docs/evidence/dep-health-sweep/proof-of-life-001/findings.json index 4a08ed43c..ca5e165ca 100644 --- a/docs/evidence/dep-health-sweep/proof-of-life-001/findings.json +++ b/docs/evidence/dep-health-sweep/proof-of-life-001/findings.json @@ -3,7 +3,7 @@ "finding_type": "UNTESTED_HANDLER", "severity": "MAJOR", "repo": "test_b_finding_fixture0", - "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1808/test_b_finding_fixture0/src/omnimarket/nodes/node_orphan_handler/handlers/handler_orphan_handler.py", + "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1842/test_b_finding_fixture0/src/omnimarket/nodes/node_orphan_handler/handlers/handler_orphan_handler.py", "symbol": "handler_orphan_handler", "detail": "Contract-referenced handler 'handler_orphan_handler' has no test file and no golden-chain fixture coverage.", "rule_id": "UNTESTED_HANDLER", @@ -13,7 +13,7 @@ "finding_type": "MISSING_TOPIC_EDGE", "severity": "CRITICAL", "repo": "test_b_finding_fixture0", - "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1808/test_b_finding_fixture0/src/omnimarket/nodes/node_orphan_handler/contract.yaml", + "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1842/test_b_finding_fixture0/src/omnimarket/nodes/node_orphan_handler/contract.yaml", "symbol": "onex.cmd.test.orphan-cmd.v1", "detail": "Topic 'onex.cmd.test.orphan-cmd.v1' is published but has no subscriber and is not declared as externally consumed.", "rule_id": "MISSING_TOPIC_EDGE", diff --git a/docs/evidence/dep-health-sweep/proof-of-life-001/projection_rows.json b/docs/evidence/dep-health-sweep/proof-of-life-001/projection_rows.json index a0a0d6dc5..d40b552bd 100644 --- a/docs/evidence/dep-health-sweep/proof-of-life-001/projection_rows.json +++ b/docs/evidence/dep-health-sweep/proof-of-life-001/projection_rows.json @@ -4,23 +4,23 @@ "finding_type": "UNTESTED_HANDLER", "severity": "MAJOR", "repo": "test_d_projection_from_event0", - "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1808/test_d_projection_from_event0/src/omnimarket/nodes/node_orphan_handler/handlers/handler_orphan_handler.py", + "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1842/test_d_projection_from_event0/src/omnimarket/nodes/node_orphan_handler/handlers/handler_orphan_handler.py", "symbol": "handler_orphan_handler", "detail": "Contract-referenced handler 'handler_orphan_handler' has no test file and no golden-chain fixture coverage.", "rule_id": "UNTESTED_HANDLER", "rule_version": "v1", - "captured_at": "2026-05-25T17:57:46.828940+00:00" + "captured_at": "2026-05-25T18:55:25.981111+00:00" }, { "run_id": "proof-of-life-001-projection", "finding_type": "MISSING_TOPIC_EDGE", "severity": "CRITICAL", "repo": "test_d_projection_from_event0", - "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1808/test_d_projection_from_event0/src/omnimarket/nodes/node_orphan_handler/contract.yaml", + "file_path": "/private/var/folders/fs/2g88nd297b7787lch6_stdl40000gn/T/pytest-of-jonah/pytest-1842/test_d_projection_from_event0/src/omnimarket/nodes/node_orphan_handler/contract.yaml", "symbol": "onex.cmd.test.orphan-cmd.v1", "detail": "Topic 'onex.cmd.test.orphan-cmd.v1' is published but has no subscriber and is not declared as externally consumed.", "rule_id": "MISSING_TOPIC_EDGE", "rule_version": "v1", - "captured_at": "2026-05-25T17:57:46.828940+00:00" + "captured_at": "2026-05-25T18:55:25.981111+00:00" } ] diff --git a/docs/evidence/dep-health-sweep/proof-of-life-001/proof_summary.md b/docs/evidence/dep-health-sweep/proof-of-life-001/proof_summary.md index 87ec44bd4..a9bf1290c 100644 --- a/docs/evidence/dep-health-sweep/proof-of-life-001/proof_summary.md +++ b/docs/evidence/dep-health-sweep/proof-of-life-001/proof_summary.md @@ -1,7 +1,7 @@ # Dep-Health Sweep Proof-of-Life Evidence run_id: proof-of-life-001 -timestamp: 2026-05-25T17:57:46.831036+00:00 +timestamp: 2026-05-25T18:55:25.983277+00:00 ## Subtest Outcomes diff --git a/docs/evidence/dep-health-sweep/proof-of-life-001/run_manifest.json b/docs/evidence/dep-health-sweep/proof-of-life-001/run_manifest.json index 4ef61c896..086afc771 100644 --- a/docs/evidence/dep-health-sweep/proof-of-life-001/run_manifest.json +++ b/docs/evidence/dep-health-sweep/proof-of-life-001/run_manifest.json @@ -1,6 +1,6 @@ { "run_id": "proof-of-life-001", - "timestamp": "2026-05-25T17:57:46.829676+00:00", + "timestamp": "2026-05-25T18:55:25.981868+00:00", "repo_roots": [ "" ], diff --git a/pyproject.toml b/pyproject.toml index 53859c9b7..a885eccc0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -212,7 +212,7 @@ node_intelligence_orchestrator = "omnimarket.nodes.node_intelligence_orchestrato node_intelligence_reducer = "omnimarket.nodes.node_intelligence_reducer" node_swarm_supervisor_orchestrator = "omnimarket.nodes.node_swarm_supervisor_orchestrator" node_swarm_dispatch_orchestrator = "omnimarket.nodes.node_swarm_dispatch_orchestrator" -node_swarm_fanout_effect = "omnimarket.nodes.node_swarm_fanout_effect" +node_swarm_fanout_orchestrator = "omnimarket.nodes.node_swarm_fanout_orchestrator" node_swarm_worker_agent = "omnimarket.nodes.node_swarm_worker_agent" node_swarm_aggregator_compute = "omnimarket.nodes.node_swarm_aggregator_compute" node_handoff_effect = "omnimarket.nodes.node_handoff_effect" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/contract.yaml b/src/omnimarket/nodes/node_swarm_fanout_effect/contract.yaml deleted file mode 100644 index 43487de72..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/contract.yaml +++ /dev/null @@ -1,98 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT ---- -name: node_swarm_fanout_effect -contract_name: node_swarm_fanout_effect -node_name: node_swarm_fanout_effect -contract_version: {major: 1, minor: 1, patch: 0} -node_version: {major: 1, minor: 1, patch: 0} -node_type: effect -description: > - Effect node for swarm parallel fan-out dispatch. Sends subtask prompts - to model endpoints via HTTP (direct mode) or via a work queue (queue mode) - for multi-site deployments where workers connect out to the runtime. - -input_model: - name: ModelSwarmFanoutRequest - class: ModelSwarmFanoutRequest - module: omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request - -output_model: - name: ModelSwarmFanoutResult - class: ModelSwarmFanoutResult - module: omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_result - -handler: - name: HandlerSwarmFanout - module: omnimarket.nodes.node_swarm_fanout_effect.handlers.handler_swarm_fanout - class: HandlerSwarmFanout - -descriptor: - node_archetype: effect - purity: effectful - idempotent: false - timeout_ms: 600000 - runtime_profiles: [effects] - -event_bus: - version: {major: 1, minor: 1, patch: 0} - plugin_managed: false - subscribe_topics: - - onex.cmd.omnimarket.swarm-fanout.v1 - - onex.evt.omnimarket.swarm-subtask-completed.v1 - publish_topics: - - onex.evt.omnimarket.swarm-fanout-completed.v1 - - onex.cmd.omnimarket.swarm-subtask-assigned.v1 - -externally_consumed_topics: - - onex.evt.omnimarket.swarm-fanout-completed.v1 - - onex.evt.omnimarket.swarm-subtask-completed.v1 - -terminal_event: onex.evt.omnimarket.swarm-fanout-completed.v1 - -published_events: - - event_type: SwarmFanoutCompleted - topic: onex.evt.omnimarket.swarm-fanout-completed.v1 - description: Swarm fan-out completed with all dispatch results. - - event_type: SwarmSubtaskAssigned - topic: onex.cmd.omnimarket.swarm-subtask-assigned.v1 - description: Subtask assignment published to work queue for worker-pull dispatch. - -io_operations: - - http_call - - kafka_publish - - kafka_subscribe - -error_handling: - retry_policy: - max_retries: 1 - initial_delay_ms: 0 - max_delay_ms: 0 - exponential_base: 1 - retry_on: - - HttpError - - TimeoutError - error_types: - - name: HttpError - description: HTTP dispatch to model endpoint failed - recoverable: true - - name: TimeoutError - description: Per-endpoint or total run timeout exceeded - recoverable: false - -timeout_behavior: - default_ms: 600000 - max_ms: 600000 - terminal_response: - status: timeout - error_message: "Swarm fan-out timed out" - -metadata: - transport_type: kafka - author: OmniNode Team - license: MIT - created: "2026-05-23" - tags: [effect, swarm, fanout, dispatch, parallel, worker-pull, OMN-11812, OMN-12000] - related_tickets: - - "OMN-11812" - - "OMN-12000" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/handlers/handler_swarm_fanout.py b/src/omnimarket/nodes/node_swarm_fanout_effect/handlers/handler_swarm_fanout.py deleted file mode 100644 index 7455b2494..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/handlers/handler_swarm_fanout.py +++ /dev/null @@ -1,713 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -"""Wave-scheduled parallel fan-out dispatcher for swarm execution.""" - -from __future__ import annotations - -import concurrent.futures -import logging -import threading -import time -from pathlib import Path -from typing import Any, Protocol - -import yaml - -from omnimarket.nodes.contract_topics import ( - contract_publish_topics, - contract_subscribe_topics, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import ( - EnumDispatchMode, - EnumExecutionStatus, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_subtask import ModelSubtask -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_config import ( - ModelSwarmConfig, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_dispatch import ( - ModelSwarmDispatch, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_endpoint import ( - ModelSwarmEndpoint, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request import ( - ModelSwarmFanoutRequest, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_result import ( - ModelSwarmFanoutResult, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_subtask_assignment import ( - ModelSwarmSubtaskAssignment, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_subtask_result import ( - ModelSwarmSubtaskResult, -) - -logger = logging.getLogger(__name__) - -_CONTRACT_PATH = Path(__file__).parent.parent / "contract.yaml" - -# Shared endpoint registry used for ID-only requests from the orchestrator. -_DEFAULT_REGISTRY_PATH = ( - Path(__file__).parent.parent.parent - / "node_swarm_registry_compute" - / "contracts" - / "endpoint_registry.yaml" -) - - -def _load_endpoint_registry( - registry_path: Path | None = None, -) -> dict[str, ModelSwarmEndpoint]: - """Load the endpoint registry and return a map of endpoint_id -> ModelSwarmEndpoint.""" - path = registry_path or _DEFAULT_REGISTRY_PATH - try: - raw: dict[str, Any] = yaml.safe_load(path.read_text()) - endpoints: dict[str, ModelSwarmEndpoint] = {} - for ep_data in raw.get("endpoints", []): - ep_id = str(ep_data.get("id", "")) - if not ep_id: - continue - endpoints[ep_id] = ModelSwarmEndpoint( - endpoint_id=ep_id, - base_url=str(ep_data.get("base_url", "")), - model_id=str(ep_data.get("model_id", "")), - status="reachable", - ) - return endpoints - except Exception as exc: - logger.warning("Failed to load endpoint registry from %s: %s", path, exc) - return {} - - -class _HttpResponse(Protocol): - def json(self) -> dict[str, object]: ... - - @property - def status_code(self) -> int: ... - - -class ProtocolHttpClient(Protocol): - def post( - self, - url: str, - *, - json: dict[str, object], - timeout: float, - ) -> _HttpResponse: ... - - -class ProtocolQueuePublisher(Protocol): - def publish(self, topic: str, payload: dict[str, object]) -> None: ... - - -class ProtocolQueueSubscriber(Protocol): - def poll( - self, topic: str, run_id: str, timeout_seconds: float - ) -> list[dict[str, object]]: ... - - -def _compute_waves(subtasks: tuple[ModelSubtask, ...]) -> list[list[ModelSubtask]]: - id_to_subtask = {s.subtask_id: s for s in subtasks} - assigned: dict[str, int] = {} - - def _wave_of(subtask_id: str) -> int: - if subtask_id in assigned: - return assigned[subtask_id] - subtask = id_to_subtask[subtask_id] - if not subtask.depends_on: - assigned[subtask_id] = 0 - return 0 - wave = max(_wave_of(dep) for dep in subtask.depends_on) + 1 - assigned[subtask_id] = wave - return wave - - for s in subtasks: - _wave_of(s.subtask_id) - - if not assigned: - return [] - max_wave = max(assigned.values()) - waves: list[list[ModelSubtask]] = [[] for _ in range(max_wave + 1)] - for s in subtasks: - waves[assigned[s.subtask_id]].append(s) - for wave in waves: - wave.sort(key=lambda s: s.subtask_id) - return waves - - -class HandlerSwarmFanout: - def __init__( - self, - http_client: ProtocolHttpClient | None = None, - queue_publisher: ProtocolQueuePublisher | None = None, - queue_subscriber: ProtocolQueueSubscriber | None = None, - registry_path: Path | None = None, - ) -> None: - self._http_client = http_client - self._queue_publisher = queue_publisher - self._queue_subscriber = queue_subscriber - self._registry_path = registry_path - - def _resolve_endpoint_map( - self, request: ModelSwarmFanoutRequest - ) -> dict[str, ModelSwarmEndpoint]: - """Return endpoint_id -> ModelSwarmEndpoint from request or registry.""" - if request.endpoints: - return {ep.endpoint_id: ep for ep in request.endpoints} - # Orchestrator sent endpoint_health dict (keyed by ID) — resolve from registry. - if request.endpoint_health: - registry = _load_endpoint_registry(self._registry_path) - resolved: dict[str, ModelSwarmEndpoint] = {} - for ep_id in request.endpoint_health: - ep = registry.get(ep_id) - if ep is not None: - resolved[ep_id] = ep - else: - logger.warning( - "endpoint_id %r not found in registry — skipping", ep_id - ) - return resolved - return {} - - def handle(self, request: ModelSwarmFanoutRequest) -> ModelSwarmFanoutResult: - if request.dispatch_mode == EnumDispatchMode.QUEUE: - return self._handle_queue(request) - return self._handle_direct(request) - - def _handle_direct( - self, request: ModelSwarmFanoutRequest - ) -> ModelSwarmFanoutResult: - import httpx - - client: ProtocolHttpClient = self._http_client or httpx.Client() - config = request.config - endpoint_map = self._resolve_endpoint_map(request) - - endpoint_semaphores: dict[str, threading.Semaphore] = { - ep_id: threading.Semaphore(config.max_subtasks_per_endpoint) - for ep_id in endpoint_map - } - - waves = _compute_waves(request.subtasks) - total_deadline = time.monotonic() + config.total_run_timeout_seconds - all_dispatches: list[ModelSwarmDispatch] = [] - failed_ids: set[str] = set() - - wall_start = time.monotonic() - - for wave_num, wave_subtasks in enumerate(waves): - runnable: list[ModelSubtask] = [] - for subtask in wave_subtasks: - if any(dep in failed_ids for dep in subtask.depends_on): - all_dispatches.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED, - failure_reason="dependency_failed", - wave=wave_num, - ) - ) - failed_ids.add(subtask.subtask_id) - else: - runnable.append(subtask) - - if not runnable: - continue - - wave_results = self._dispatch_wave( - wave_num=wave_num, - subtasks=runnable, - assignments=request.assignments, - endpoint_map=endpoint_map, - endpoint_semaphores=endpoint_semaphores, - config=config, - total_deadline=total_deadline, - client=client, - fallback_policy_enabled=config.fallback_policy_enabled, - ) - for d in wave_results: - all_dispatches.append(d) - if d.execution_status != EnumExecutionStatus.SUCCEEDED: - failed_ids.add(d.subtask_id) - - wall_latency_ms = int((time.monotonic() - wall_start) * 1000) - sum_latency = sum(d.latency_ms for d in all_dispatches) - - return ModelSwarmFanoutResult( - dispatches=tuple(all_dispatches), - wall_latency_ms=wall_latency_ms, - sum_subtask_latency_ms=sum_latency, - run_id=request.run_id, - ) - - def _handle_queue(self, request: ModelSwarmFanoutRequest) -> ModelSwarmFanoutResult: - if self._queue_publisher is None or self._queue_subscriber is None: - raise ValueError( - "queue_publisher and queue_subscriber are required for dispatch_mode=queue" - ) - - publish_topics = contract_publish_topics(_CONTRACT_PATH) - assignment_topic = next( - t for t in publish_topics if "swarm-subtask-assigned" in t - ) - subscribe_topics = contract_subscribe_topics(_CONTRACT_PATH) - completion_topic = next( - t for t in subscribe_topics if "swarm-subtask-completed" in t - ) - - config = request.config - waves = _compute_waves(request.subtasks) - total_deadline = time.monotonic() + config.total_run_timeout_seconds - all_dispatches: list[ModelSwarmDispatch] = [] - failed_ids: set[str] = set() - - wall_start = time.monotonic() - - for wave_num, wave_subtasks in enumerate(waves): - runnable: list[ModelSubtask] = [] - for subtask in wave_subtasks: - if any(dep in failed_ids for dep in subtask.depends_on): - all_dispatches.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED, - failure_reason="dependency_failed", - wave=wave_num, - ) - ) - failed_ids.add(subtask.subtask_id) - else: - runnable.append(subtask) - - if not runnable: - continue - - wave_dispatches = self._dispatch_wave_queue( - wave_num=wave_num, - subtasks=runnable, - request=request, - assignment_topic=assignment_topic, - completion_topic=completion_topic, - total_deadline=total_deadline, - ) - for d in wave_dispatches: - all_dispatches.append(d) - if d.execution_status != EnumExecutionStatus.SUCCEEDED: - failed_ids.add(d.subtask_id) - - wall_latency_ms = int((time.monotonic() - wall_start) * 1000) - sum_latency = sum(d.latency_ms for d in all_dispatches) - - return ModelSwarmFanoutResult( - dispatches=tuple(all_dispatches), - wall_latency_ms=wall_latency_ms, - sum_subtask_latency_ms=sum_latency, - run_id=request.run_id, - ) - - def _dispatch_wave_queue( - self, - wave_num: int, - subtasks: list[ModelSubtask], - request: ModelSwarmFanoutRequest, - assignment_topic: str, - completion_topic: str, - total_deadline: float, - ) -> list[ModelSwarmDispatch]: - assert self._queue_publisher is not None - assert self._queue_subscriber is not None - - config = request.config - endpoint_map = self._resolve_endpoint_map(request) - published: dict[str, ModelSwarmSubtaskAssignment] = {} - - for subtask in subtasks: - time_left = total_deadline - time.monotonic() - if time_left <= 0: - continue - - endpoint_id = request.assignments.get(subtask.subtask_id, "") - endpoint = endpoint_map.get(endpoint_id) - worker_id = request.worker_assignments.get(subtask.subtask_id, "") - - assignment = ModelSwarmSubtaskAssignment( - run_id=request.run_id, - subtask_id=subtask.subtask_id, - worker_id=worker_id, - description=subtask.description, - model_id=endpoint.model_id if endpoint else "", - endpoint_url=endpoint.base_url if endpoint else "", - timeout_seconds=min( - config.per_endpoint_timeout_seconds, int(time_left) - ), - correlation_id=request.correlation_id, - ) - self._queue_publisher.publish(assignment_topic, assignment.model_dump()) - published[subtask.subtask_id] = assignment - - if not published: - return [ - _immediate_timeout(s, wave_num, "budget_exhausted") for s in subtasks - ] - - time_left = max(total_deadline - time.monotonic(), 1.0) - raw_results = self._queue_subscriber.poll( - completion_topic, - request.run_id, - timeout_seconds=time_left, - ) - - result_by_subtask: dict[str, ModelSwarmSubtaskResult] = {} - for raw in raw_results: - try: - r = ModelSwarmSubtaskResult.model_validate(raw) - result_by_subtask[r.subtask_id] = r - except Exception: - pass - - dispatches: list[ModelSwarmDispatch] = [] - for subtask in subtasks: - published_assignment: ModelSwarmSubtaskAssignment | None = published.get( - subtask.subtask_id - ) - if published_assignment is None: - dispatches.append( - _immediate_timeout(subtask, wave_num, "budget_exhausted") - ) - continue - - result = result_by_subtask.get(subtask.subtask_id) - endpoint_id = request.assignments.get(subtask.subtask_id, "") - endpoint = endpoint_map.get(endpoint_id) - - if result is None: - dispatches.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint_id, - model_id=endpoint.model_id if endpoint else "", - base_url=endpoint.base_url if endpoint else "", - execution_status=EnumExecutionStatus.TIMEOUT, - failure_reason="no_result_from_worker", - wave=wave_num, - ) - ) - else: - dispatches.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint_id, - model_id=result.model_id - or (endpoint.model_id if endpoint else ""), - base_url=endpoint.base_url if endpoint else "", - execution_status=result.execution_status, - response_text=result.response_text, - failure_reason=result.failure_reason, - latency_ms=result.latency_ms, - wave=wave_num, - ) - ) - return dispatches - - def _dispatch_wave( - self, - wave_num: int, - subtasks: list[ModelSubtask], - assignments: dict[str, str], - endpoint_map: dict[str, ModelSwarmEndpoint], - endpoint_semaphores: dict[str, threading.Semaphore], - config: ModelSwarmConfig, - total_deadline: float, - client: ProtocolHttpClient, - fallback_policy_enabled: bool, - ) -> list[ModelSwarmDispatch]: - worker_count = min(config.max_parallel_subtasks, len(subtasks)) - futures: dict[concurrent.futures.Future[ModelSwarmDispatch], ModelSubtask] = {} - - with concurrent.futures.ThreadPoolExecutor( - max_workers=worker_count - ) as executor: - for subtask in subtasks: - time_left = total_deadline - time.monotonic() - if time_left <= 0: - futures[ - executor.submit( - _immediate_timeout, subtask, wave_num, "budget_exhausted" - ) - ] = subtask - continue - - endpoint_id = assignments.get(subtask.subtask_id, "") - endpoint = endpoint_map.get(endpoint_id) - if endpoint is None: - futures[ - executor.submit(_no_endpoint_dispatch, subtask, wave_num) - ] = subtask - continue - - semaphore = endpoint_semaphores.get( - endpoint_id, threading.Semaphore(config.max_subtasks_per_endpoint) - ) - per_timeout = min(float(config.per_endpoint_timeout_seconds), time_left) - futures[ - executor.submit( - self._run_subtask, - subtask=subtask, - endpoint=endpoint, - semaphore=semaphore, - per_timeout=per_timeout, - wave_num=wave_num, - client=client, - config=config, - endpoint_map=endpoint_map, - endpoint_semaphores=endpoint_semaphores, - assignments=assignments, - fallback_policy_enabled=fallback_policy_enabled, - ) - ] = subtask - - results: list[ModelSwarmDispatch] = [] - remaining = max(total_deadline - time.monotonic(), 1.0) - try: - for future in concurrent.futures.as_completed( - futures, timeout=remaining - ): - try: - results.append(future.result()) - except Exception as exc: - subtask = futures[future] - results.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.FAILED, - failure_class=type(exc).__name__, - failure_reason=str(exc), - wave=wave_num, - ) - ) - except concurrent.futures.TimeoutError: - for future, subtask in futures.items(): - if not future.done(): - future.cancel() - results.append( - ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.TIMEOUT, - failure_reason="total_run_timeout", - wave=wave_num, - ) - ) - - return results - - def _run_subtask( - self, - subtask: ModelSubtask, - endpoint: ModelSwarmEndpoint, - semaphore: threading.Semaphore, - per_timeout: float, - wave_num: int, - client: ProtocolHttpClient, - config: ModelSwarmConfig, - endpoint_map: dict[str, ModelSwarmEndpoint], - endpoint_semaphores: dict[str, threading.Semaphore], - assignments: dict[str, str], - fallback_policy_enabled: bool, - ) -> ModelSwarmDispatch: - started_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - t0 = time.monotonic() - - with semaphore: - try: - response = client.post( - f"{endpoint.base_url}/chat/completions", - json={ - "model": endpoint.model_id, - "messages": [{"role": "user", "content": subtask.description}], - }, - timeout=per_timeout, - ) - data: dict[str, object] = response.json() - choices = data.get("choices", []) - text = "" - if isinstance(choices, list) and choices: - first = choices[0] - if isinstance(first, dict): - msg = first.get("message", {}) - if isinstance(msg, dict): - text = str(msg.get("content", "")) - latency_ms = int((time.monotonic() - t0) * 1000) - completed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint.endpoint_id, - model_id=endpoint.model_id, - base_url=endpoint.base_url, - endpoint_status=endpoint.status, - execution_status=EnumExecutionStatus.SUCCEEDED, - response_text=text, - started_at=started_at, - completed_at=completed_at, - latency_ms=latency_ms, - wave=wave_num, - ) - except Exception as primary_exc: - latency_ms = int((time.monotonic() - t0) * 1000) - completed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - failure_class = type(primary_exc).__name__ - failure_reason = str(primary_exc) - - # Retry: try each available endpoint up to max_retries times - if config.retry_policy_max_retries > 0: - for attempt in range(config.retry_policy_max_retries): - try: - response = client.post( - f"{endpoint.base_url}/chat/completions", - json={ - "model": endpoint.model_id, - "messages": [ - {"role": "user", "content": subtask.description} - ], - }, - timeout=per_timeout, - ) - data = response.json() - choices = data.get("choices", []) - text = "" - if isinstance(choices, list) and choices: - first = choices[0] - if isinstance(first, dict): - msg = first.get("message", {}) - if isinstance(msg, dict): - text = str(msg.get("content", "")) - latency_ms = int((time.monotonic() - t0) * 1000) - completed_at = time.strftime( - "%Y-%m-%dT%H:%M:%SZ", time.gmtime() - ) - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint.endpoint_id, - model_id=endpoint.model_id, - base_url=endpoint.base_url, - endpoint_status=endpoint.status, - execution_status=EnumExecutionStatus.SUCCEEDED, - response_text=text, - retry_count=attempt + 1, - started_at=started_at, - completed_at=completed_at, - latency_ms=latency_ms, - wave=wave_num, - ) - except Exception: - pass - - # Fallback: try a different endpoint - if fallback_policy_enabled: - primary_id = endpoint.endpoint_id - for fb_ep in endpoint_map.values(): - if fb_ep.endpoint_id == primary_id: - continue - fb_sem = endpoint_semaphores.get( - fb_ep.endpoint_id, threading.Semaphore(1) - ) - try: - with fb_sem: - response = client.post( - f"{fb_ep.base_url}/chat/completions", - json={ - "model": fb_ep.model_id, - "messages": [ - { - "role": "user", - "content": subtask.description, - } - ], - }, - timeout=per_timeout, - ) - data = response.json() - choices = data.get("choices", []) - text = "" - if isinstance(choices, list) and choices: - first = choices[0] - if isinstance(first, dict): - msg = first.get("message", {}) - if isinstance(msg, dict): - text = str(msg.get("content", "")) - latency_ms = int((time.monotonic() - t0) * 1000) - completed_at = time.strftime( - "%Y-%m-%dT%H:%M:%SZ", time.gmtime() - ) - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint.endpoint_id, - model_id=endpoint.model_id, - base_url=endpoint.base_url, - endpoint_status=endpoint.status, - execution_status=EnumExecutionStatus.SUCCEEDED, - response_text=text, - retry_count=1, - fallback_endpoint_id=fb_ep.endpoint_id, - started_at=started_at, - completed_at=completed_at, - latency_ms=latency_ms, - wave=wave_num, - ) - except Exception: - continue - - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id=endpoint.endpoint_id, - model_id=endpoint.model_id, - base_url=endpoint.base_url, - endpoint_status=endpoint.status, - execution_status=EnumExecutionStatus.FAILED, - failure_class=failure_class, - failure_reason=failure_reason, - retry_count=config.retry_policy_max_retries, - started_at=started_at, - completed_at=completed_at, - latency_ms=latency_ms, - wave=wave_num, - ) - - -def _no_endpoint_dispatch(subtask: ModelSubtask, wave_num: int) -> ModelSwarmDispatch: - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.FAILED, - failure_reason="no_endpoint_assigned", - wave=wave_num, - ) - - -def _immediate_timeout( - subtask: ModelSubtask, wave_num: int, reason: str -) -> ModelSwarmDispatch: - return ModelSwarmDispatch( - subtask_id=subtask.subtask_id, - endpoint_id="", - model_id="", - base_url="", - execution_status=EnumExecutionStatus.TIMEOUT, - failure_reason=reason, - wave=wave_num, - ) diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/metadata.yaml b/src/omnimarket/nodes/node_swarm_fanout_effect/metadata.yaml deleted file mode 100644 index 1a5fe579a..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/metadata.yaml +++ /dev/null @@ -1,30 +0,0 @@ -name: node_swarm_fanout_effect -version: "1.0.0" -description: "Effect node for swarm parallel fan-out dispatch to model endpoints" -omnibase_core_compat: ">=0.39.0,<1.0.0" -entry_points: - onex.nodes: - node_swarm_fanout_effect: "omnimarket.nodes.node_swarm_fanout_effect" -capabilities: - standalone: false - full_runtime: true - requires_network: true - requires_repo: false - requires_secrets: false - requires_docker: false - side_effect_class: effectful -dependencies: - - "omnibase_core>=0.39.0" - - "httpx>=0.27.0" -authors: ["OmniNode Team"] -license: "MIT" -tags: ["swarm", "fanout", "effect", "dispatch", "parallel", "OMN-11812"] -pack: "swarm" -display_name: "Swarm Fan-out Effect" -node_role: "effect" -consumed_topics: - - onex.cmd.omnimarket.swarm-fanout.v1 -produced_topics: - - onex.evt.omnimarket.swarm-fanout-completed.v1 -externally_consumed_topics: - - onex.evt.omnimarket.swarm-fanout-completed.v1 diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/enums.py b/src/omnimarket/nodes/node_swarm_fanout_effect/models/enums.py deleted file mode 100644 index e6546fbab..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/models/enums.py +++ /dev/null @@ -1,14 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -from __future__ import annotations - -from enum import StrEnum - -from omnimarket.models.swarm.enum_execution_status import EnumExecutionStatus - -__all__ = ["EnumDispatchMode", "EnumExecutionStatus"] - - -class EnumDispatchMode(StrEnum): - DIRECT = "direct" - QUEUE = "queue" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_request.py b/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_request.py deleted file mode 100644 index afc7d8e4e..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_request.py +++ /dev/null @@ -1,42 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -from __future__ import annotations - -from typing import Any - -from pydantic import BaseModel, ConfigDict - -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import EnumDispatchMode -from omnimarket.nodes.node_swarm_fanout_effect.models.model_subtask import ModelSubtask -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_config import ( - ModelSwarmConfig, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_endpoint import ( - ModelSwarmEndpoint, -) - - -class ModelSwarmFanoutRequest(BaseModel): - """Command/request accepted by the fanout effect node. - - The orchestrator sends ``endpoint_health`` (health-check results keyed - by endpoint id). Direct callers / tests may supply ``endpoints`` - (full endpoint objects) instead. The handler resolves endpoints from - the registry when only ``endpoint_health`` is provided. - """ - - model_config = ConfigDict(frozen=True, extra="forbid") - - subtasks: tuple[ModelSubtask, ...] - assignments: dict[str, str] # subtask_id → endpoint_id - # Full endpoint objects — populated by direct callers / tests. - endpoints: tuple[ModelSwarmEndpoint, ...] = () - # Health dict forwarded by the orchestrator; handler resolves full - # endpoint objects from the registry when ``endpoints`` is empty. - endpoint_health: dict[str, Any] = {} - config: ModelSwarmConfig - correlation_id: str - run_id: str - dispatch_mode: EnumDispatchMode = EnumDispatchMode.DIRECT - # maps subtask_id → worker_id for queue dispatch mode - worker_assignments: dict[str, str] = {} diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout.py b/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout.py deleted file mode 100644 index 69872d52a..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout.py +++ /dev/null @@ -1,270 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -"""Tests for HandlerSwarmFanout — wave scheduling, bounded concurrency, retry, fallback.""" - -from __future__ import annotations - -from typing import Any -from unittest.mock import MagicMock - -import pytest - -from omnimarket.nodes.node_swarm_fanout_effect.handlers.handler_swarm_fanout import ( - HandlerSwarmFanout, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import EnumExecutionStatus -from omnimarket.nodes.node_swarm_fanout_effect.models.model_subtask import ModelSubtask -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_config import ( - ModelSwarmConfig, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_endpoint import ( - ModelSwarmEndpoint, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request import ( - ModelSwarmFanoutRequest, -) - - -def _make_response(text: str = "ok") -> MagicMock: - resp = MagicMock() - resp.status_code = 200 - resp.json.return_value = {"choices": [{"message": {"content": text}}]} - return resp - - -def _make_client(text: str = "ok") -> MagicMock: - client = MagicMock() - client.post.return_value = _make_response(text) - return client - - -def _make_request( - subtasks: list[ModelSubtask], - assignments: dict[str, str], - endpoints: list[ModelSwarmEndpoint], - config: ModelSwarmConfig | None = None, -) -> ModelSwarmFanoutRequest: - return ModelSwarmFanoutRequest( - subtasks=tuple(subtasks), - assignments=assignments, - endpoints=tuple(endpoints), - config=config or ModelSwarmConfig(), - correlation_id="corr-123", - run_id="run-abc", - ) - - -@pytest.mark.unit -def test_single_subtask_success() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - subtask = ModelSubtask(subtask_id="s1", description="do thing") - client = _make_client("result text") - handler = HandlerSwarmFanout(http_client=client) - - result = handler.handle(_make_request([subtask], {"s1": "ep1"}, [ep])) - - assert len(result.dispatches) == 1 - d = result.dispatches[0] - assert d.subtask_id == "s1" - assert d.execution_status == EnumExecutionStatus.SUCCEEDED - assert d.response_text == "result text" - assert d.endpoint_id == "ep1" - assert d.latency_ms >= 0 - assert result.wall_latency_ms >= 0 - - -@pytest.mark.unit -def test_wave_ordering_dependency_runs_after() -> None: - """Subtask with depends_on must run in a later wave.""" - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - wave_order: list[str] = [] - - def side_effect(url: str, *, json: Any, timeout: float) -> MagicMock: - task_content = json["messages"][0]["content"] - wave_order.append(task_content) - return _make_response("ok") - - client = MagicMock() - client.post.side_effect = side_effect - - s1 = ModelSubtask(subtask_id="s1", description="first", depends_on=()) - s2 = ModelSubtask(subtask_id="s2", description="second", depends_on=("s1",)) - - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) - - assert len(result.dispatches) == 2 - assert all( - d.execution_status == EnumExecutionStatus.SUCCEEDED for d in result.dispatches - ) - # "first" must appear before "second" in dispatch order - assert wave_order.index("first") < wave_order.index("second") - # Waves must differ - wave_by_id = {d.subtask_id: d.wave for d in result.dispatches} - assert wave_by_id["s1"] < wave_by_id["s2"] - - -@pytest.mark.unit -def test_dependency_failed_skips_dependent() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - client = MagicMock() - client.post.side_effect = RuntimeError("endpoint down") - - s1 = ModelSubtask(subtask_id="s1", description="first", depends_on=()) - s2 = ModelSubtask(subtask_id="s2", description="second", depends_on=("s1",)) - - config = ModelSwarmConfig(retry_policy_max_retries=0, fallback_policy_enabled=False) - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle( - _make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep], config) - ) - - assert len(result.dispatches) == 2 - by_id = {d.subtask_id: d for d in result.dispatches} - assert by_id["s1"].execution_status == EnumExecutionStatus.FAILED - assert by_id["s2"].execution_status == EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED - - -@pytest.mark.unit -def test_bounded_concurrency_max_parallel() -> None: - """max_parallel_subtasks=1 means subtasks run serially.""" - import threading - - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - concurrent_count = 0 - max_concurrent = 0 - lock = threading.Lock() - - def side_effect(url: str, *, json: Any, timeout: float) -> MagicMock: - nonlocal concurrent_count, max_concurrent - with lock: - concurrent_count += 1 - if concurrent_count > max_concurrent: - max_concurrent = concurrent_count - import time - - time.sleep(0.01) - with lock: - concurrent_count -= 1 - return _make_response("ok") - - client = MagicMock() - client.post.side_effect = side_effect - - subtasks = [ - ModelSubtask(subtask_id=f"s{i}", description=f"task{i}") for i in range(4) - ] - assignments = {s.subtask_id: "ep1" for s in subtasks} - config = ModelSwarmConfig(max_parallel_subtasks=1, max_subtasks_per_endpoint=4) - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle(_make_request(subtasks, assignments, [ep], config)) - - assert all( - d.execution_status == EnumExecutionStatus.SUCCEEDED for d in result.dispatches - ) - assert max_concurrent == 1 - - -@pytest.mark.unit -def test_retry_on_failure_succeeds_second_attempt() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - call_count = 0 - - def side_effect(url: str, *, json: Any, timeout: float) -> MagicMock: - nonlocal call_count - call_count += 1 - if call_count == 1: - raise RuntimeError("transient error") - return _make_response("retry ok") - - client = MagicMock() - client.post.side_effect = side_effect - - config = ModelSwarmConfig(retry_policy_max_retries=1, fallback_policy_enabled=False) - subtask = ModelSubtask(subtask_id="s1", description="flaky task") - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle(_make_request([subtask], {"s1": "ep1"}, [ep], config)) - - assert result.dispatches[0].execution_status == EnumExecutionStatus.SUCCEEDED - assert result.dispatches[0].retry_count == 1 - assert call_count == 2 - - -@pytest.mark.unit -def test_fallback_to_alternate_endpoint() -> None: - ep1 = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - ep2 = ModelSwarmEndpoint(endpoint_id="ep2", base_url="http://ep2", model_id="m2") - - def side_effect(url: str, *, json: Any, timeout: float) -> MagicMock: - if "ep1" in url: - raise RuntimeError("ep1 down") - return _make_response("fallback ok") - - client = MagicMock() - client.post.side_effect = side_effect - - config = ModelSwarmConfig(retry_policy_max_retries=0, fallback_policy_enabled=True) - subtask = ModelSubtask(subtask_id="s1", description="task") - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle(_make_request([subtask], {"s1": "ep1"}, [ep1, ep2], config)) - - d = result.dispatches[0] - assert d.execution_status == EnumExecutionStatus.SUCCEEDED - assert d.fallback_endpoint_id == "ep2" - assert d.response_text == "fallback ok" - - -@pytest.mark.unit -def test_all_subtasks_fail_returns_result_with_failures() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - client = MagicMock() - client.post.side_effect = RuntimeError("always fails") - - config = ModelSwarmConfig(retry_policy_max_retries=0, fallback_policy_enabled=False) - subtasks = [ - ModelSubtask(subtask_id=f"s{i}", description=f"task{i}") for i in range(3) - ] - assignments = {s.subtask_id: "ep1" for s in subtasks} - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle(_make_request(subtasks, assignments, [ep], config)) - - assert len(result.dispatches) == 3 - assert all( - d.execution_status == EnumExecutionStatus.FAILED for d in result.dispatches - ) - assert all(d.failure_reason == "always fails" for d in result.dispatches) - assert result.wall_latency_ms >= 0 - - -@pytest.mark.unit -def test_no_assignment_for_subtask_fails_gracefully() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - subtask = ModelSubtask(subtask_id="s1", description="orphan task") - handler = HandlerSwarmFanout(http_client=MagicMock()) - - result = handler.handle(_make_request([subtask], {}, [ep])) - - d = result.dispatches[0] - assert d.execution_status == EnumExecutionStatus.FAILED - assert "no_endpoint_assigned" in d.failure_reason - - -@pytest.mark.unit -def test_multi_wave_three_levels() -> None: - """A → B → C must produce three distinct waves.""" - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - client = _make_client("ok") - - s_a = ModelSubtask(subtask_id="a", description="a", depends_on=()) - s_b = ModelSubtask(subtask_id="b", description="b", depends_on=("a",)) - s_c = ModelSubtask(subtask_id="c", description="c", depends_on=("b",)) - - handler = HandlerSwarmFanout(http_client=client) - result = handler.handle( - _make_request([s_a, s_b, s_c], {"a": "ep1", "b": "ep1", "c": "ep1"}, [ep]) - ) - - wave_by_id = {d.subtask_id: d.wave for d in result.dispatches} - assert wave_by_id["a"] == 0 - assert wave_by_id["b"] == 1 - assert wave_by_id["c"] == 2 diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout_queue.py b/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout_queue.py deleted file mode 100644 index 26fcbf04e..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_handler_swarm_fanout_queue.py +++ /dev/null @@ -1,249 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -"""Tests for HandlerSwarmFanout queue dispatch mode.""" - -from __future__ import annotations - -from unittest.mock import MagicMock - -import pytest - -from omnimarket.nodes.node_swarm_fanout_effect.handlers.handler_swarm_fanout import ( - HandlerSwarmFanout, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import ( - EnumDispatchMode, - EnumExecutionStatus, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_subtask import ModelSubtask -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_config import ( - ModelSwarmConfig, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_endpoint import ( - ModelSwarmEndpoint, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request import ( - ModelSwarmFanoutRequest, -) - - -def _make_publisher() -> MagicMock: - pub = MagicMock() - pub.publish = MagicMock() - return pub - - -def _make_subscriber(results: list[dict[str, object]]) -> MagicMock: - sub = MagicMock() - sub.poll = MagicMock(return_value=results) - return sub - - -def _make_request( - subtasks: list[ModelSubtask], - assignments: dict[str, str], - endpoints: list[ModelSwarmEndpoint], - worker_assignments: dict[str, str] | None = None, - config: ModelSwarmConfig | None = None, -) -> ModelSwarmFanoutRequest: - return ModelSwarmFanoutRequest( - subtasks=tuple(subtasks), - assignments=assignments, - endpoints=tuple(endpoints), - config=config or ModelSwarmConfig(), - correlation_id="corr-queue-test", - run_id="run-queue-001", - dispatch_mode=EnumDispatchMode.QUEUE, - worker_assignments=worker_assignments or {}, - ) - - -@pytest.mark.unit -def test_direct_mode_still_works_regression() -> None: - """dispatch_mode=direct (default) uses HTTP path unchanged.""" - from unittest.mock import MagicMock - - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - subtask = ModelSubtask(subtask_id="s1", description="do thing") - client = MagicMock() - resp = MagicMock() - resp.status_code = 200 - resp.json.return_value = {"choices": [{"message": {"content": "direct ok"}}]} - client.post.return_value = resp - - handler = HandlerSwarmFanout(http_client=client) - request = ModelSwarmFanoutRequest( - subtasks=(subtask,), - assignments={"s1": "ep1"}, - endpoints=(ep,), - config=ModelSwarmConfig(), - correlation_id="corr-direct", - run_id="run-direct", - dispatch_mode=EnumDispatchMode.DIRECT, - ) - result = handler.handle(request) - - assert len(result.dispatches) == 1 - assert result.dispatches[0].execution_status == EnumExecutionStatus.SUCCEEDED - assert result.dispatches[0].response_text == "direct ok" - client.post.assert_called_once() - - -@pytest.mark.unit -def test_queue_mode_publishes_assignments() -> None: - """Queue mode publishes one assignment per subtask to the assignment topic.""" - ep = ModelSwarmEndpoint( - endpoint_id="ep1", base_url="http://localhost:8000/v1", model_id="qwen" - ) - s1 = ModelSubtask(subtask_id="s1", description="task one") - s2 = ModelSubtask(subtask_id="s2", description="task two") - - pub = _make_publisher() - sub = _make_subscriber( - [ - { - "run_id": "run-queue-001", - "subtask_id": "s1", - "worker_id": "worker-a", - "execution_status": "succeeded", - "response_text": "result one", - "latency_ms": 42, - "failure_reason": "", - "model_id": "qwen", - }, - { - "run_id": "run-queue-001", - "subtask_id": "s2", - "worker_id": "worker-a", - "execution_status": "succeeded", - "response_text": "result two", - "latency_ms": 38, - "failure_reason": "", - "model_id": "qwen", - }, - ] - ) - - handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) - result = handler.handle( - _make_request( - [s1, s2], - {"s1": "ep1", "s2": "ep1"}, - [ep], - worker_assignments={"s1": "worker-a", "s2": "worker-a"}, - ) - ) - - assert pub.publish.call_count == 2 - published_topics = {call.args[0] for call in pub.publish.call_args_list} - assert all("swarm-subtask-assigned" in t for t in published_topics) - - assert len(result.dispatches) == 2 - by_id = {d.subtask_id: d for d in result.dispatches} - assert by_id["s1"].execution_status == EnumExecutionStatus.SUCCEEDED - assert by_id["s1"].response_text == "result one" - assert by_id["s2"].execution_status == EnumExecutionStatus.SUCCEEDED - assert by_id["s2"].response_text == "result two" - - -@pytest.mark.unit -def test_queue_mode_timeout_when_no_worker_result() -> None: - """If worker never returns a result, dispatch is marked TIMEOUT.""" - ep = ModelSwarmEndpoint( - endpoint_id="ep1", base_url="http://localhost:8000/v1", model_id="m1" - ) - s1 = ModelSubtask(subtask_id="s1", description="unanswered task") - - pub = _make_publisher() - sub = _make_subscriber([]) # no results returned - - handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) - result = handler.handle(_make_request([s1], {"s1": "ep1"}, [ep])) - - assert len(result.dispatches) == 1 - assert result.dispatches[0].execution_status == EnumExecutionStatus.TIMEOUT - assert result.dispatches[0].failure_reason == "no_result_from_worker" - - -@pytest.mark.unit -def test_queue_mode_worker_failure_propagates() -> None: - """A FAILED result from a worker maps to FAILED dispatch.""" - ep = ModelSwarmEndpoint( - endpoint_id="ep1", base_url="http://localhost:8000/v1", model_id="m1" - ) - s1 = ModelSubtask(subtask_id="s1", description="failing task") - - pub = _make_publisher() - sub = _make_subscriber( - [ - { - "run_id": "run-queue-001", - "subtask_id": "s1", - "worker_id": "worker-a", - "execution_status": "failed", - "response_text": "", - "latency_ms": 10, - "failure_reason": "connection_refused", - "model_id": "m1", - }, - ] - ) - - handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) - result = handler.handle(_make_request([s1], {"s1": "ep1"}, [ep])) - - assert result.dispatches[0].execution_status == EnumExecutionStatus.FAILED - assert result.dispatches[0].failure_reason == "connection_refused" - - -@pytest.mark.unit -def test_queue_mode_requires_publisher_and_subscriber() -> None: - """queue mode raises ValueError if publisher or subscriber are missing.""" - ep = ModelSwarmEndpoint( - endpoint_id="ep1", base_url="http://localhost:8000/v1", model_id="m1" - ) - s1 = ModelSubtask(subtask_id="s1", description="task") - request = _make_request([s1], {"s1": "ep1"}, [ep]) - - with pytest.raises(ValueError, match="queue_publisher"): - HandlerSwarmFanout().handle(request) - - -@pytest.mark.unit -def test_queue_mode_dependency_chain_respected() -> None: - """Dependency-failed subtasks are skipped in queue mode too.""" - ep = ModelSwarmEndpoint( - endpoint_id="ep1", base_url="http://localhost:8000/v1", model_id="m1" - ) - s1 = ModelSubtask(subtask_id="s1", description="root") - s2 = ModelSubtask(subtask_id="s2", description="child", depends_on=("s1",)) - - pub = _make_publisher() - # s1 comes back as failed; s2 should be skipped without being published - sub = _make_subscriber( - [ - { - "run_id": "run-queue-001", - "subtask_id": "s1", - "worker_id": "w", - "execution_status": "failed", - "response_text": "", - "latency_ms": 5, - "failure_reason": "model_error", - "model_id": "m1", - }, - ] - ) - - handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) - result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) - - assert len(result.dispatches) == 2 - by_id = {d.subtask_id: d for d in result.dispatches} - assert by_id["s1"].execution_status == EnumExecutionStatus.FAILED - assert by_id["s2"].execution_status == EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED - # s2 must never be published to the assignment topic - published_subtask_ids = { - call.args[1]["subtask_id"] for call in pub.publish.call_args_list - } - assert "s2" not in published_subtask_ids diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_models.py b/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_models.py deleted file mode 100644 index 51d5bf1b2..000000000 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/test_models.py +++ /dev/null @@ -1,129 +0,0 @@ -# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -"""Model validation tests for node_swarm_fanout_effect.""" - -from __future__ import annotations - -import pytest -from pydantic import ValidationError - -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import EnumExecutionStatus -from omnimarket.nodes.node_swarm_fanout_effect.models.model_subtask import ModelSubtask -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_config import ( - ModelSwarmConfig, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_dispatch import ( - ModelSwarmDispatch, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_endpoint import ( - ModelSwarmEndpoint, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request import ( - ModelSwarmFanoutRequest, -) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_result import ( - ModelSwarmFanoutResult, -) - - -@pytest.mark.unit -def test_model_subtask_defaults() -> None: - s = ModelSubtask(subtask_id="x", description="desc") - assert s.depends_on == () - assert s.estimated_tokens == 0 - assert s.category == "general" - - -@pytest.mark.unit -def test_model_subtask_frozen() -> None: - s = ModelSubtask(subtask_id="x", description="desc") - with pytest.raises((ValidationError, TypeError)): - s.subtask_id = "y" # type: ignore[misc] - - -@pytest.mark.unit -def test_model_swarm_config_defaults() -> None: - cfg = ModelSwarmConfig() - assert cfg.max_parallel_subtasks == 4 - assert cfg.fallback_policy_enabled is True - - -@pytest.mark.unit -def test_model_swarm_dispatch_required_fields() -> None: - d = ModelSwarmDispatch( - subtask_id="s1", - endpoint_id="ep1", - model_id="m1", - base_url="http://x", - execution_status=EnumExecutionStatus.SUCCEEDED, - ) - assert d.response_text == "" - assert d.retry_count == 0 - assert d.wave == 0 - - -@pytest.mark.unit -def test_model_swarm_dispatch_extra_field_rejected() -> None: - with pytest.raises(ValidationError): - ModelSwarmDispatch( # type: ignore[call-arg] - subtask_id="s1", - endpoint_id="ep1", - model_id="m1", - base_url="http://x", - execution_status=EnumExecutionStatus.SUCCEEDED, - unknown_field="bad", - ) - - -@pytest.mark.unit -def test_model_swarm_fanout_request_roundtrip() -> None: - ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://ep1", model_id="m1") - s = ModelSubtask(subtask_id="s1", description="task") - req = ModelSwarmFanoutRequest( - subtasks=(s,), - assignments={"s1": "ep1"}, - endpoints=(ep,), - config=ModelSwarmConfig(), - correlation_id="c1", - run_id="r1", - ) - assert req.subtasks[0].subtask_id == "s1" - assert req.assignments["s1"] == "ep1" - - -@pytest.mark.unit -def test_model_swarm_fanout_result_sum() -> None: - d1 = ModelSwarmDispatch( - subtask_id="s1", - endpoint_id="ep1", - model_id="m1", - base_url="http://x", - execution_status=EnumExecutionStatus.SUCCEEDED, - latency_ms=100, - ) - d2 = ModelSwarmDispatch( - subtask_id="s2", - endpoint_id="ep1", - model_id="m1", - base_url="http://x", - execution_status=EnumExecutionStatus.FAILED, - latency_ms=50, - ) - result = ModelSwarmFanoutResult( - dispatches=(d1, d2), - wall_latency_ms=120, - sum_subtask_latency_ms=150, - ) - assert result.sum_subtask_latency_ms == 150 - assert len(result.dispatches) == 2 - - -@pytest.mark.unit -def test_enum_execution_status_values() -> None: - assert EnumExecutionStatus.SUCCEEDED.value == "succeeded" - assert EnumExecutionStatus.FAILED.value == "failed" - assert EnumExecutionStatus.TIMEOUT.value == "timeout" - assert ( - EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED.value - == "skipped_dependency_failed" - ) diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/__init__.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/__init__.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/__init__.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/contract.yaml b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/contract.yaml new file mode 100644 index 000000000..5fed3d316 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/contract.yaml @@ -0,0 +1,112 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +--- +name: node_swarm_fanout_orchestrator +contract_name: node_swarm_fanout_orchestrator +node_name: node_swarm_fanout_orchestrator +contract_version: {major: 2, minor: 0, patch: 0} +node_version: {major: 2, minor: 0, patch: 0} +node_type: orchestrator +description: > + Orchestrator node for swarm parallel fan-out dispatch. Publishes N + delegation-execute commands (one per subtask) to + onex.cmd.omnimarket.delegation-execute.v1, then collects completion + events by correlation_id. All inference I/O is handled by + node_llm_delegation_call_effect. This node owns only coordination. + +input_model: + name: ModelSwarmFanoutRequest + class: ModelSwarmFanoutRequest + module: omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request + +output_model: + name: ModelSwarmFanoutResult + class: ModelSwarmFanoutResult + module: omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_result + +handler: + name: HandlerSwarmFanout + module: omnimarket.nodes.node_swarm_fanout_orchestrator.handlers.handler_swarm_fanout + class: HandlerSwarmFanout + +descriptor: + node_archetype: orchestrator + purity: orchestrating + idempotent: false + timeout_ms: 600000 + runtime_profiles: [main] + +fsm: + states: [PLANNING, DISPATCHING, COLLECTING, COMPLETED, FAILED] + initial: PLANNING + transitions: + - from: PLANNING + to: DISPATCHING + on: commands_built + - from: DISPATCHING + to: COLLECTING + on: all_commands_published + - from: COLLECTING + to: COMPLETED + on: all_subtasks_terminal + - from: COLLECTING + to: FAILED + on: fatal_error + +event_bus: + version: {major: 2, minor: 0, patch: 0} + plugin_managed: false + subscribe_topics: + - onex.cmd.omnimarket.swarm-fanout.v1 + - onex.evt.omnimarket.delegation-call-completed.v1 + - onex.evt.omnimarket.delegation-escalation-triggered.v1 + - onex.evt.omnimarket.delegation-all-tiers-failed.v1 + publish_topics: + - onex.evt.omnimarket.swarm-fanout-completed.v1 + - onex.cmd.omnimarket.delegation-execute.v1 + +externally_consumed_topics: + - onex.evt.omnimarket.swarm-fanout-completed.v1 + +terminal_event: onex.evt.omnimarket.swarm-fanout-completed.v1 + +published_events: + - event_type: SwarmFanoutCompleted + topic: onex.evt.omnimarket.swarm-fanout-completed.v1 + description: Swarm fan-out completed; all subtasks terminal. + - event_type: DelegationExecute + topic: onex.cmd.omnimarket.delegation-execute.v1 + description: Per-subtask delegation command forwarded to node_llm_delegation_call_effect. + +io_operations: + - kafka_publish + - kafka_subscribe + +error_handling: + retry_policy: + max_retries: 0 + initial_delay_ms: 0 + max_delay_ms: 0 + exponential_base: 1 + retry_on: [] + error_types: + - name: TimeoutError + description: Per-subtask or total run timeout exceeded + recoverable: false + +timeout_behavior: + default_ms: 600000 + max_ms: 600000 + terminal_response: + status: timeout + error_message: "Swarm fan-out timed out" + +metadata: + transport_type: kafka + author: OmniNode Team + license: MIT + created: "2026-05-25" + tags: [orchestrator, swarm, fanout, dispatch, parallel, OMN-12118] + related_tickets: + - "OMN-12118" + - "OMN-11812" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/handlers/__init__.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/handlers/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/handlers/__init__.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/handlers/__init__.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/handlers/handler_swarm_fanout.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/handlers/handler_swarm_fanout.py new file mode 100644 index 000000000..aee15b317 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/handlers/handler_swarm_fanout.py @@ -0,0 +1,387 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Wave-scheduled swarm fan-out orchestrator. + +Publishes delegation-execute commands per subtask and collects +delegation completion events by correlation_id. No HTTP. No threads. +""" + +from __future__ import annotations + +import hashlib +import logging +import time +from pathlib import Path +from typing import Any, Protocol + +import yaml + +from omnimarket.nodes.contract_topics import ( + contract_publish_topics, + contract_subscribe_topics, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.enums import ( + EnumExecutionStatus, + EnumFanoutFsmState, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_subtask import ( + ModelSubtask, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_dispatch import ( + ModelSwarmDispatch, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_endpoint import ( + ModelSwarmEndpoint, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request import ( + ModelSwarmFanoutRequest, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_result import ( + ModelSwarmFanoutResult, +) + +logger = logging.getLogger(__name__) + +_CONTRACT_PATH = Path(__file__).parent.parent / "contract.yaml" + +_DEFAULT_REGISTRY_PATH = ( + Path(__file__).parent.parent.parent + / "node_swarm_registry_compute" + / "contracts" + / "endpoint_registry.yaml" +) + + +class ProtocolQueuePublisher(Protocol): + def publish(self, topic: str, payload: dict[str, object]) -> None: ... + + +class ProtocolQueueSubscriber(Protocol): + def poll( + self, topics: list[str], run_id: str, timeout_seconds: float + ) -> list[dict[str, object]]: ... + + +def _sha256(text: str) -> str: + return hashlib.sha256(text.encode()).hexdigest() + + +def _compute_waves(subtasks: tuple[ModelSubtask, ...]) -> list[list[ModelSubtask]]: + id_to_subtask = {s.subtask_id: s for s in subtasks} + assigned: dict[str, int] = {} + + def _wave_of(subtask_id: str) -> int: + if subtask_id in assigned: + return assigned[subtask_id] + subtask = id_to_subtask[subtask_id] + if not subtask.depends_on: + assigned[subtask_id] = 0 + return 0 + wave = max(_wave_of(dep) for dep in subtask.depends_on) + 1 + assigned[subtask_id] = wave + return wave + + for s in subtasks: + _wave_of(s.subtask_id) + + if not assigned: + return [] + max_wave = max(assigned.values()) + waves: list[list[ModelSubtask]] = [[] for _ in range(max_wave + 1)] + for s in subtasks: + waves[assigned[s.subtask_id]].append(s) + for wave in waves: + wave.sort(key=lambda s: s.subtask_id) + return waves + + +def _load_endpoint_registry( + registry_path: Path | None = None, +) -> dict[str, ModelSwarmEndpoint]: + path = registry_path or _DEFAULT_REGISTRY_PATH + try: + raw: dict[str, Any] = yaml.safe_load(path.read_text()) + endpoints: dict[str, ModelSwarmEndpoint] = {} + for ep_data in raw.get("endpoints", []): + ep_id = str(ep_data.get("id", "")) + if not ep_id: + continue + endpoints[ep_id] = ModelSwarmEndpoint( + endpoint_id=ep_id, + base_url=str(ep_data.get("base_url", "")), + model_id=str(ep_data.get("model_id", "")), + endpoint_ref=str(ep_data.get("endpoint_ref", "")), + capabilities=tuple(str(c) for c in ep_data.get("capabilities", [])), + status="reachable", + ) + return endpoints + except Exception as exc: + logger.warning("Failed to load endpoint registry from %s: %s", path, exc) + return {} + + +def _correlation_id(run_id: str, subtask_id: str) -> str: + return f"{run_id}-{subtask_id}" + + +def _owns_correlation_id(run_id: str, correlation_id: str) -> bool: + """True if this correlation_id was issued by this fanout run.""" + return correlation_id.startswith(f"{run_id}-") + + +class HandlerSwarmFanout: + """FSM orchestrator: PLANNING → DISPATCHING → COLLECTING → COMPLETED/FAILED. + + No HTTP. No threads. Publishes delegation-execute commands and collects + delegation completion events. The delegation call effect handles all I/O. + """ + + def __init__( + self, + queue_publisher: ProtocolQueuePublisher | None = None, + queue_subscriber: ProtocolQueueSubscriber | None = None, + registry_path: Path | None = None, + ) -> None: + self._queue_publisher = queue_publisher + self._queue_subscriber = queue_subscriber + self._registry_path = registry_path + + def _resolve_endpoint_map( + self, request: ModelSwarmFanoutRequest + ) -> dict[str, ModelSwarmEndpoint]: + if request.endpoints: + return {ep.endpoint_id: ep for ep in request.endpoints} + if request.endpoint_health: + registry = _load_endpoint_registry(self._registry_path) + resolved: dict[str, ModelSwarmEndpoint] = {} + for ep_id in request.endpoint_health: + ep = registry.get(ep_id) + if ep is not None: + resolved[ep_id] = ep + else: + logger.warning("endpoint_id %r not found in registry", ep_id) + return resolved + return {} + + def handle(self, request: ModelSwarmFanoutRequest) -> ModelSwarmFanoutResult: + if self._queue_publisher is None or self._queue_subscriber is None: + raise ValueError( + "queue_publisher and queue_subscriber are required for the fanout orchestrator" + ) + + publish_topics = contract_publish_topics(_CONTRACT_PATH) + delegation_execute_topic = next( + t for t in publish_topics if "delegation-execute" in t + ) + subscribe_topics_all = contract_subscribe_topics(_CONTRACT_PATH) + completion_topics = [ + t + for t in subscribe_topics_all + if any( + k in t + for k in [ + "delegation-call-completed", + "delegation-escalation-triggered", + "delegation-all-tiers-failed", + ] + ) + ] + + # PLANNING: validate inputs, build endpoint map and wave schedule + fsm_state = EnumFanoutFsmState.PLANNING + endpoint_map = self._resolve_endpoint_map(request) + waves = _compute_waves(request.subtasks) + config = request.config + total_deadline = time.monotonic() + config.total_run_timeout_seconds + all_dispatches: list[ModelSwarmDispatch] = [] + failed_ids: set[str] = set() + wall_start = time.monotonic() + + # DISPATCHING: publish commands wave-by-wave, collect after each wave + fsm_state = EnumFanoutFsmState.DISPATCHING + + for wave_num, wave_subtasks in enumerate(waves): + runnable: list[ModelSubtask] = [] + for subtask in wave_subtasks: + if any(dep in failed_ids for dep in subtask.depends_on): + all_dispatches.append( + ModelSwarmDispatch( + subtask_id=subtask.subtask_id, + endpoint_id="", + model_id="", + base_url="", + execution_status=EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED, + failure_reason="dependency_failed", + wave=wave_num, + ) + ) + failed_ids.add(subtask.subtask_id) + else: + runnable.append(subtask) + + if not runnable: + continue + + wave_dispatches = self._dispatch_wave( + wave_num=wave_num, + subtasks=runnable, + request=request, + endpoint_map=endpoint_map, + delegation_execute_topic=delegation_execute_topic, + completion_topics=completion_topics, + total_deadline=total_deadline, + ) + for d in wave_dispatches: + all_dispatches.append(d) + if d.execution_status != EnumExecutionStatus.SUCCEEDED: + failed_ids.add(d.subtask_id) + + fsm_state = EnumFanoutFsmState.COMPLETED + wall_latency_ms = int((time.monotonic() - wall_start) * 1000) + sum_latency = sum(d.latency_ms for d in all_dispatches) + completed_count = sum( + 1 + for d in all_dispatches + if d.execution_status == EnumExecutionStatus.SUCCEEDED + ) + failed_count = len(all_dispatches) - completed_count + _ = fsm_state + + return ModelSwarmFanoutResult( + dispatches=tuple(all_dispatches), + wall_latency_ms=wall_latency_ms, + sum_subtask_latency_ms=sum_latency, + run_id=request.run_id, + completed_count=completed_count, + failed_count=failed_count, + degraded=failed_count > 0, + aggregation_mode="collect_all", + endpoint_registry_hash=request.endpoint_registry_hash, + routing_policy_hash=request.routing_policy_hash, + ) + + def _dispatch_wave( + self, + wave_num: int, + subtasks: list[ModelSubtask], + request: ModelSwarmFanoutRequest, + endpoint_map: dict[str, ModelSwarmEndpoint], + delegation_execute_topic: str, + completion_topics: list[str], + total_deadline: float, + ) -> list[ModelSwarmDispatch]: + assert self._queue_publisher is not None + assert self._queue_subscriber is not None + + config = request.config + pending: dict[str, tuple[ModelSubtask, ModelSwarmEndpoint | None]] = {} + + for subtask in subtasks: + time_left = total_deadline - time.monotonic() + if time_left <= 0: + continue + + endpoint_id = request.assignments.get(subtask.subtask_id, "") + endpoint = endpoint_map.get(endpoint_id) + + corr_id = _correlation_id(request.run_id, subtask.subtask_id) + + if endpoint is None or not endpoint.endpoint_ref: + pending[corr_id] = (subtask, None) + continue + + command: dict[str, object] = { + "request_id": f"{request.run_id}-{subtask.subtask_id}", + "correlation_id": corr_id, + "causation_id": request.run_id, + "model_id": endpoint.model_id, + "endpoint_ref": endpoint.endpoint_ref, + "prompt": subtask.description, + "prompt_hash": _sha256(subtask.description), + "max_tokens": subtask.estimated_tokens or 2048, + "task_type": "swarm_subtask", + "task_id": subtask.subtask_id, + "routing_policy_hash": request.routing_policy_hash, + "registry_hash": request.endpoint_registry_hash, + } + self._queue_publisher.publish(delegation_execute_topic, command) + pending[corr_id] = (subtask, endpoint) + + if not pending: + return [] + + # COLLECTING: wait for terminal events for this wave + time_left = max(total_deadline - time.monotonic(), 1.0) + raw_events = self._queue_subscriber.poll( + completion_topics, + request.run_id, + timeout_seconds=min(float(config.per_endpoint_timeout_seconds), time_left), + ) + + results: dict[str, dict[str, object]] = {} + for raw_ev in raw_events: + corr_id_val = str(raw_ev.get("correlation_id", "")) + if not _owns_correlation_id(request.run_id, corr_id_val): + continue + results[corr_id_val] = raw_ev + + dispatches: list[ModelSwarmDispatch] = [] + for corr_id, (subtask, endpoint) in pending.items(): + endpoint_id = endpoint.endpoint_id if endpoint else "" + model_id = endpoint.model_id if endpoint else "" + base_url = endpoint.base_url if endpoint else "" + + if endpoint is None or not endpoint.endpoint_ref: + dispatches.append( + ModelSwarmDispatch( + subtask_id=subtask.subtask_id, + endpoint_id=endpoint_id, + model_id=model_id, + base_url=base_url, + execution_status=EnumExecutionStatus.FAILED, + failure_reason="no_endpoint_assigned", + wave=wave_num, + ) + ) + continue + + maybe_event = results.get(corr_id) + if maybe_event is None: + dispatches.append( + ModelSwarmDispatch( + subtask_id=subtask.subtask_id, + endpoint_id=endpoint_id, + model_id=model_id, + base_url=base_url, + execution_status=EnumExecutionStatus.TIMEOUT, + failure_reason="no_terminal_event_received", + wave=wave_num, + ) + ) + continue + + event: dict[str, object] = maybe_event + is_failure = "all-tiers-failed" in str(event.get("_topic", "")) + success = bool(event.get("success", True)) and not is_failure + latency_ms = int(str(event.get("latency_ms", 0))) + + dispatches.append( + ModelSwarmDispatch( + subtask_id=subtask.subtask_id, + endpoint_id=endpoint_id, + model_id=str(event.get("model_id", model_id)), + base_url=base_url, + execution_status=( + EnumExecutionStatus.SUCCEEDED + if success + else EnumExecutionStatus.FAILED + ), + failure_reason="" + if success + else str(event.get("failure_class", "")), + latency_ms=latency_ms, + wave=wave_num, + ) + ) + + return dispatches diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/metadata.yaml b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/metadata.yaml new file mode 100644 index 000000000..1672f175a --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/metadata.yaml @@ -0,0 +1,33 @@ +name: node_swarm_fanout_orchestrator +version: "2.0.0" +description: "Orchestrator node for swarm parallel fan-out dispatch via delegation-execute commands" +omnibase_core_compat: ">=0.39.0,<1.0.0" +entry_points: + onex.nodes: + node_swarm_fanout_orchestrator: "omnimarket.nodes.node_swarm_fanout_orchestrator" +capabilities: + standalone: false + full_runtime: true + requires_network: false + requires_repo: false + requires_secrets: false + requires_docker: false + side_effect_class: orchestrating +dependencies: + - "omnibase_core>=0.39.0" +authors: ["OmniNode Team"] +license: "MIT" +tags: ["swarm", "fanout", "orchestrator", "dispatch", "parallel", "OMN-12118"] +pack: "swarm" +display_name: "Swarm Fan-out Orchestrator" +node_role: "orchestrator" +consumed_topics: + - onex.cmd.omnimarket.swarm-fanout.v1 + - onex.evt.omnimarket.delegation-call-completed.v1 + - onex.evt.omnimarket.delegation-escalation-triggered.v1 + - onex.evt.omnimarket.delegation-all-tiers-failed.v1 +produced_topics: + - onex.evt.omnimarket.swarm-fanout-completed.v1 + - onex.cmd.omnimarket.delegation-execute.v1 +externally_consumed_topics: + - onex.evt.omnimarket.swarm-fanout-completed.v1 diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/__init__.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/__init__.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/__init__.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/enums.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/enums.py new file mode 100644 index 000000000..b7d5d838d --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/enums.py @@ -0,0 +1,17 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +from __future__ import annotations + +from enum import StrEnum + +from omnimarket.models.swarm.enum_execution_status import EnumExecutionStatus + +__all__ = ["EnumExecutionStatus", "EnumFanoutFsmState"] + + +class EnumFanoutFsmState(StrEnum): + PLANNING = "PLANNING" + DISPATCHING = "DISPATCHING" + COLLECTING = "COLLECTING" + COMPLETED = "COMPLETED" + FAILED = "FAILED" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_subtask.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_subtask.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_subtask.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_subtask.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_config.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_config.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_config.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_config.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_dispatch.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_dispatch.py similarity index 86% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_dispatch.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_dispatch.py index 6eeef2c73..258f09845 100644 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_dispatch.py +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_dispatch.py @@ -4,7 +4,9 @@ from pydantic import BaseModel, ConfigDict -from omnimarket.nodes.node_swarm_fanout_effect.models.enums import EnumExecutionStatus +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.enums import ( + EnumExecutionStatus, +) class ModelSwarmDispatch(BaseModel): diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_endpoint.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_endpoint.py similarity index 78% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_endpoint.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_endpoint.py index b152c81f1..860edb1dd 100644 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_endpoint.py +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_endpoint.py @@ -11,5 +11,7 @@ class ModelSwarmEndpoint(BaseModel): endpoint_id: str base_url: str model_id: str + endpoint_ref: str = "" + """Name of env var holding the base URL (e.g. 'LLM_LOCAL_PRIMARY_URL').""" status: str = "reachable" capabilities: tuple[str, ...] = () diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_request.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_request.py new file mode 100644 index 000000000..7833084a3 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_request.py @@ -0,0 +1,45 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, ConfigDict + +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_subtask import ( + ModelSubtask, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_config import ( + ModelSwarmConfig, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_endpoint import ( + ModelSwarmEndpoint, +) + + +class ModelSwarmFanoutRequest(BaseModel): + """Command accepted by the fanout orchestrator. + + The dispatch orchestrator sends ``endpoint_health`` (health-check results keyed + by endpoint_id). Direct callers / tests may supply ``endpoints`` (full endpoint + objects) instead. The handler resolves endpoints from the registry when only + ``endpoint_health`` is provided. + + dispatch_mode and worker_assignments are intentionally absent — this node + always publishes delegation-execute commands; it never does direct HTTP. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + subtasks: tuple[ModelSubtask, ...] + assignments: dict[str, str] # subtask_id → endpoint_id + # Full endpoint objects — populated by direct callers / tests. + endpoints: tuple[ModelSwarmEndpoint, ...] = () + # Health dict forwarded by the dispatch orchestrator; handler resolves full + # endpoint objects from the registry when ``endpoints`` is empty. + endpoint_health: dict[str, Any] = {} + config: ModelSwarmConfig + correlation_id: str + run_id: str + endpoint_registry_hash: str = "" + routing_policy_hash: str = "" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_result.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_result.py similarity index 52% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_result.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_result.py index fbbe86601..49d355631 100644 --- a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_fanout_result.py +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_fanout_result.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_dispatch import ( +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_dispatch import ( ModelSwarmDispatch, ) @@ -16,3 +16,12 @@ class ModelSwarmFanoutResult(BaseModel): wall_latency_ms: int sum_subtask_latency_ms: int run_id: str = "" + + # Terminal event payload fields (swarm-fanout-completed.v1) + completed_count: int = 0 + failed_count: int = 0 + degraded: bool = False + aggregation_mode: str = "collect_all" + endpoint_registry_hash: str = "" + routing_policy_hash: str = "" + projection_ref: str = "" diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_subtask_assignment.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_subtask_assignment.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_subtask_assignment.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_subtask_assignment.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_subtask_result.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_subtask_result.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/models/model_swarm_subtask_result.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/models/model_swarm_subtask_result.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_effect/tests/__init__.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_swarm_fanout_effect/tests/__init__.py rename to src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/__init__.py diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout.py new file mode 100644 index 000000000..e3b7be3a1 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout.py @@ -0,0 +1,386 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Tests for HandlerSwarmFanout orchestrator — FSM transitions, command building, result collection.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from omnimarket.nodes.node_swarm_fanout_orchestrator.handlers.handler_swarm_fanout import ( + HandlerSwarmFanout, + _compute_waves, + _correlation_id, + _owns_correlation_id, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.enums import ( + EnumExecutionStatus, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_subtask import ( + ModelSubtask, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_config import ( + ModelSwarmConfig, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_endpoint import ( + ModelSwarmEndpoint, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request import ( + ModelSwarmFanoutRequest, +) + + +def _make_publisher() -> MagicMock: + pub = MagicMock() + pub.publish = MagicMock() + return pub + + +def _make_subscriber(events: list[dict[str, object]]) -> MagicMock: + sub = MagicMock() + sub.poll = MagicMock(return_value=events) + return sub + + +def _make_endpoint( + ep_id: str = "ep1", ref: str = "LLM_LOCAL_PRIMARY_URL" +) -> ModelSwarmEndpoint: + return ModelSwarmEndpoint( + endpoint_id=ep_id, + base_url="http://192.168.86.201:8000/v1", # onex-allow-internal-ip + model_id="Qwen3.6-35B-A3B", + endpoint_ref=ref, + ) + + +def _make_request( + subtasks: list[ModelSubtask], + assignments: dict[str, str], + endpoints: list[ModelSwarmEndpoint], + config: ModelSwarmConfig | None = None, + run_id: str = "run-abc", +) -> ModelSwarmFanoutRequest: + return ModelSwarmFanoutRequest( + subtasks=tuple(subtasks), + assignments=assignments, + endpoints=tuple(endpoints), + config=config or ModelSwarmConfig(), + correlation_id="corr-123", + run_id=run_id, + ) + + +def _completion_event( + run_id: str, subtask_id: str, success: bool = True +) -> dict[str, object]: + return { + "correlation_id": _correlation_id(run_id, subtask_id), + "success": success, + "model_id": "Qwen3.6-35B-A3B", + "latency_ms": 50, + "_topic": "onex.evt.omnimarket.delegation-call-completed.v1", + } + + +def _failure_event(run_id: str, subtask_id: str) -> dict[str, object]: + return { + "correlation_id": _correlation_id(run_id, subtask_id), + "success": False, + "model_id": "", + "latency_ms": 10, + "failure_class": "all_tiers_exhausted", + "_topic": "onex.evt.omnimarket.delegation-all-tiers-failed.v1", + } + + +@pytest.mark.unit +def test_correlation_id_format() -> None: + assert _correlation_id("run-1", "s1") == "run-1-s1" + + +@pytest.mark.unit +def test_owns_correlation_id_true() -> None: + assert _owns_correlation_id("run-1", "run-1-s1") is True + + +@pytest.mark.unit +def test_owns_correlation_id_false_unrelated() -> None: + assert _owns_correlation_id("run-1", "run-2-s1") is False + + +@pytest.mark.unit +def test_missing_publisher_raises() -> None: + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="task") + req = _make_request([s], {"s1": "ep1"}, [ep]) + with pytest.raises(ValueError, match="queue_publisher"): + HandlerSwarmFanout().handle(req) + + +@pytest.mark.unit +def test_single_subtask_success() -> None: + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="do the thing") + pub = _make_publisher() + sub = _make_subscriber([_completion_event("run-abc", "s1")]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + assert len(result.dispatches) == 1 + d = result.dispatches[0] + assert d.subtask_id == "s1" + assert d.execution_status == EnumExecutionStatus.SUCCEEDED + assert result.completed_count == 1 + assert result.failed_count == 0 + assert result.degraded is False + + +@pytest.mark.unit +def test_publishes_delegation_execute_command() -> None: + """Handler publishes one delegation-execute command per subtask.""" + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="some prompt") + pub = _make_publisher() + sub = _make_subscriber([_completion_event("run-abc", "s1")]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + pub.publish.assert_called_once() + topic, payload = pub.publish.call_args.args + assert "delegation-execute" in topic + assert payload["endpoint_ref"] == "LLM_LOCAL_PRIMARY_URL" + assert payload["model_id"] == "Qwen3.6-35B-A3B" + assert payload["correlation_id"] == "run-abc-s1" + assert payload["causation_id"] == "run-abc" + assert payload["prompt"] == "some prompt" + assert "prompt_hash" in payload + + +@pytest.mark.unit +def test_no_http_import_in_handler() -> None: + """Verify the orchestrator handler contains no httpx import.""" + import ast + from pathlib import Path + + handler_src = ( + Path(__file__).parent.parent / "handlers" / "handler_swarm_fanout.py" + ).read_text() + tree = ast.parse(handler_src) + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + if isinstance(node, ast.Import): + for alias in node.names: + assert "httpx" not in alias.name, ( + "httpx import found in orchestrator" + ) + elif isinstance(node, ast.ImportFrom) and node.module: + assert "httpx" not in node.module, "httpx import found in orchestrator" + + +@pytest.mark.unit +def test_wave_ordering_wave0_dispatched_before_wave1() -> None: + """Wave 0 subtasks are published before wave 1 subtasks.""" + ep = _make_endpoint() + s1 = ModelSubtask(subtask_id="s1", description="first", depends_on=()) + s2 = ModelSubtask(subtask_id="s2", description="second", depends_on=("s1",)) + + published_order: list[str] = [] + pub = MagicMock() + + def capture_publish(topic: str, payload: dict[str, object]) -> None: + published_order.append(str(payload.get("task_id", ""))) + + pub.publish = MagicMock(side_effect=capture_publish) + + sub = _make_subscriber( + [ + _completion_event("run-abc", "s1"), + _completion_event("run-abc", "s2"), + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) + + assert len(result.dispatches) == 2 + assert all( + d.execution_status == EnumExecutionStatus.SUCCEEDED for d in result.dispatches + ) + wave_by_id = {d.subtask_id: d.wave for d in result.dispatches} + assert wave_by_id["s1"] < wave_by_id["s2"] + + +@pytest.mark.unit +def test_dependency_failed_skips_dependent() -> None: + """If s1 fails, s2 (which depends on s1) is skipped without being published.""" + ep = _make_endpoint() + s1 = ModelSubtask(subtask_id="s1", description="root", depends_on=()) + s2 = ModelSubtask(subtask_id="s2", description="child", depends_on=("s1",)) + + pub = _make_publisher() + sub = _make_subscriber([_failure_event("run-abc", "s1")]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) + + assert len(result.dispatches) == 2 + by_id = {d.subtask_id: d for d in result.dispatches} + assert by_id["s1"].execution_status == EnumExecutionStatus.FAILED + assert by_id["s2"].execution_status == EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED + + published_task_ids = { + str(call.args[1].get("task_id", "")) for call in pub.publish.call_args_list + } + assert "s2" not in published_task_ids + + +@pytest.mark.unit +def test_timeout_when_no_completion_event() -> None: + """If no terminal event arrives for a subtask, it is marked TIMEOUT.""" + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="slow task") + pub = _make_publisher() + sub = _make_subscriber([]) + config = ModelSwarmConfig(per_endpoint_timeout_seconds=1) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep], config)) + + d = result.dispatches[0] + assert d.execution_status == EnumExecutionStatus.TIMEOUT + assert "no_terminal_event" in d.failure_reason + + +@pytest.mark.unit +def test_no_endpoint_assigned_fails_immediately() -> None: + """If no endpoint_id in assignments, dispatch fails without publishing.""" + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="orphan task") + pub = _make_publisher() + sub = _make_subscriber([]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + result = handler.handle(_make_request([s], {}, [ep])) + + d = result.dispatches[0] + assert d.execution_status == EnumExecutionStatus.FAILED + assert "no_endpoint" in d.failure_reason + pub.publish.assert_not_called() + + +@pytest.mark.unit +def test_endpoint_missing_endpoint_ref_fails_immediately() -> None: + """Endpoint without endpoint_ref cannot be dispatched.""" + ep = ModelSwarmEndpoint( + endpoint_id="ep1", + base_url="http://192.168.86.201:8000/v1", # onex-allow-internal-ip + model_id="m1", + endpoint_ref="", + ) + s = ModelSubtask(subtask_id="s1", description="task") + pub = _make_publisher() + sub = _make_subscriber([]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + d = result.dispatches[0] + assert d.execution_status == EnumExecutionStatus.FAILED + pub.publish.assert_not_called() + + +@pytest.mark.unit +def test_unrelated_correlation_id_filtered_out() -> None: + """Events from a different run_id are not attributed to our subtasks.""" + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="task") + pub = _make_publisher() + sub = _make_subscriber( + [ + { + "correlation_id": "other-run-s1", + "success": True, + "model_id": "m1", + "latency_ms": 10, + "_topic": "onex.evt.omnimarket.delegation-call-completed.v1", + } + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + assert result.dispatches[0].execution_status == EnumExecutionStatus.TIMEOUT + + +@pytest.mark.unit +def test_three_wave_chain() -> None: + """A → B → C produces three distinct waves.""" + ep = _make_endpoint() + s_a = ModelSubtask(subtask_id="a", description="a", depends_on=()) + s_b = ModelSubtask(subtask_id="b", description="b", depends_on=("a",)) + s_c = ModelSubtask(subtask_id="c", description="c", depends_on=("b",)) + + pub = _make_publisher() + sub = _make_subscriber( + [ + _completion_event("run-abc", "a"), + _completion_event("run-abc", "b"), + _completion_event("run-abc", "c"), + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle( + _make_request([s_a, s_b, s_c], {"a": "ep1", "b": "ep1", "c": "ep1"}, [ep]) + ) + + wave_by_id = {d.subtask_id: d.wave for d in result.dispatches} + assert wave_by_id["a"] == 0 + assert wave_by_id["b"] == 1 + assert wave_by_id["c"] == 2 + + +@pytest.mark.unit +def test_degraded_flag_set_when_any_failure() -> None: + ep = _make_endpoint() + s1 = ModelSubtask(subtask_id="s1", description="ok") + s2 = ModelSubtask(subtask_id="s2", description="fail") + + pub = _make_publisher() + sub = _make_subscriber( + [ + _completion_event("run-abc", "s1", success=True), + _failure_event("run-abc", "s2"), + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) + + assert result.degraded is True + assert result.completed_count == 1 + assert result.failed_count == 1 + + +@pytest.mark.unit +def test_compute_waves_empty() -> None: + assert _compute_waves(()) == [] + + +@pytest.mark.unit +def test_compute_waves_no_deps() -> None: + subtasks = tuple( + ModelSubtask(subtask_id=f"s{i}", description="x") for i in range(3) + ) + waves = _compute_waves(subtasks) + assert len(waves) == 1 + assert len(waves[0]) == 3 + + +@pytest.mark.unit +def test_compute_waves_stable_sort_within_wave() -> None: + """Subtasks within a wave are sorted by subtask_id for determinism.""" + s_b = ModelSubtask(subtask_id="b", description="b") + s_a = ModelSubtask(subtask_id="a", description="a") + waves = _compute_waves((s_b, s_a)) + assert [s.subtask_id for s in waves[0]] == ["a", "b"] diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout_queue.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout_queue.py new file mode 100644 index 000000000..e2c06a828 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_handler_swarm_fanout_queue.py @@ -0,0 +1,236 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Additional orchestrator tests — multi-subtask collection, endpoint_ref passthrough.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from omnimarket.nodes.node_swarm_fanout_orchestrator.handlers.handler_swarm_fanout import ( + HandlerSwarmFanout, + _correlation_id, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.enums import ( + EnumExecutionStatus, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_subtask import ( + ModelSubtask, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_config import ( + ModelSwarmConfig, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_endpoint import ( + ModelSwarmEndpoint, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request import ( + ModelSwarmFanoutRequest, +) + + +def _make_publisher() -> MagicMock: + pub = MagicMock() + pub.publish = MagicMock() + return pub + + +def _make_subscriber(events: list[dict[str, object]]) -> MagicMock: + sub = MagicMock() + sub.poll = MagicMock(return_value=events) + return sub + + +def _make_endpoint( + ep_id: str = "ep1", + ref: str = "LLM_LOCAL_PRIMARY_URL", + model_id: str = "qwen", +) -> ModelSwarmEndpoint: + return ModelSwarmEndpoint( + endpoint_id=ep_id, + base_url="http://localhost:8000/v1", + model_id=model_id, + endpoint_ref=ref, + ) + + +def _make_request( + subtasks: list[ModelSubtask], + assignments: dict[str, str], + endpoints: list[ModelSwarmEndpoint], + config: ModelSwarmConfig | None = None, +) -> ModelSwarmFanoutRequest: + return ModelSwarmFanoutRequest( + subtasks=tuple(subtasks), + assignments=assignments, + endpoints=tuple(endpoints), + config=config or ModelSwarmConfig(), + correlation_id="corr-queue-test", + run_id="run-queue-001", + ) + + +def _ok_event(subtask_id: str, model_id: str = "qwen") -> dict[str, object]: + return { + "correlation_id": _correlation_id("run-queue-001", subtask_id), + "success": True, + "model_id": model_id, + "latency_ms": 40, + "_topic": "onex.evt.omnimarket.delegation-call-completed.v1", + } + + +@pytest.mark.unit +def test_two_subtasks_both_succeed() -> None: + ep = _make_endpoint() + s1 = ModelSubtask(subtask_id="s1", description="task one") + s2 = ModelSubtask(subtask_id="s2", description="task two") + + pub = _make_publisher() + sub = _make_subscriber([_ok_event("s1"), _ok_event("s2")]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) + + assert pub.publish.call_count == 2 + published_topics = {call.args[0] for call in pub.publish.call_args_list} + assert all("delegation-execute" in t for t in published_topics) + + assert len(result.dispatches) == 2 + by_id = {d.subtask_id: d for d in result.dispatches} + assert by_id["s1"].execution_status == EnumExecutionStatus.SUCCEEDED + assert by_id["s2"].execution_status == EnumExecutionStatus.SUCCEEDED + assert result.completed_count == 2 + assert result.failed_count == 0 + + +@pytest.mark.unit +def test_endpoint_ref_passed_through_in_command() -> None: + """endpoint_ref from registry is forwarded in the delegation-execute payload.""" + ep = _make_endpoint(ref="LLM_CODER_FAST_URL") + s = ModelSubtask(subtask_id="s1", description="task") + pub = _make_publisher() + sub = _make_subscriber([_ok_event("s1")]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + + handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + _, payload = pub.publish.call_args.args + assert payload["endpoint_ref"] == "LLM_CODER_FAST_URL" + + +@pytest.mark.unit +def test_timeout_when_no_terminal_event() -> None: + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="unanswered task") + pub = _make_publisher() + sub = _make_subscriber([]) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + assert len(result.dispatches) == 1 + assert result.dispatches[0].execution_status == EnumExecutionStatus.TIMEOUT + assert "no_terminal_event" in result.dispatches[0].failure_reason + + +@pytest.mark.unit +def test_all_tiers_failed_event_marks_dispatch_failed() -> None: + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="failing task") + pub = _make_publisher() + sub = _make_subscriber( + [ + { + "correlation_id": _correlation_id("run-queue-001", "s1"), + "success": False, + "model_id": "", + "latency_ms": 10, + "failure_class": "connection_refused", + "_topic": "onex.evt.omnimarket.delegation-all-tiers-failed.v1", + } + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s], {"s1": "ep1"}, [ep])) + + assert result.dispatches[0].execution_status == EnumExecutionStatus.FAILED + + +@pytest.mark.unit +def test_requires_publisher_and_subscriber() -> None: + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="task") + req = _make_request([s], {"s1": "ep1"}, [ep]) + + with pytest.raises(ValueError, match="queue_publisher"): + HandlerSwarmFanout().handle(req) + + +@pytest.mark.unit +def test_dependency_chain_s2_not_published_when_s1_fails() -> None: + ep = _make_endpoint() + s1 = ModelSubtask(subtask_id="s1", description="root") + s2 = ModelSubtask(subtask_id="s2", description="child", depends_on=("s1",)) + + pub = _make_publisher() + sub = _make_subscriber( + [ + { + "correlation_id": _correlation_id("run-queue-001", "s1"), + "success": False, + "model_id": "", + "latency_ms": 5, + "failure_class": "model_error", + "_topic": "onex.evt.omnimarket.delegation-all-tiers-failed.v1", + } + ] + ) + handler = HandlerSwarmFanout(queue_publisher=pub, queue_subscriber=sub) + result = handler.handle(_make_request([s1, s2], {"s1": "ep1", "s2": "ep1"}, [ep])) + + assert len(result.dispatches) == 2 + by_id = {d.subtask_id: d for d in result.dispatches} + assert by_id["s1"].execution_status == EnumExecutionStatus.FAILED + assert by_id["s2"].execution_status == EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED + + published_task_ids = { + str(call.args[1].get("task_id", "")) for call in pub.publish.call_args_list + } + assert "s2" not in published_task_ids + + +@pytest.mark.unit +def test_dispatch_mode_field_rejected() -> None: + """ModelSwarmFanoutRequest has extra='forbid' — dispatch_mode must fail validation.""" + from pydantic import ValidationError + + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="task") + with pytest.raises(ValidationError): + ModelSwarmFanoutRequest( + subtasks=(s,), + assignments={"s1": "ep1"}, + endpoints=(ep,), + config=ModelSwarmConfig(), + correlation_id="c", + run_id="r", + dispatch_mode="direct", # type: ignore[call-arg] + ) + + +@pytest.mark.unit +def test_worker_assignments_field_rejected() -> None: + """worker_assignments is removed — extra='forbid' must raise.""" + from pydantic import ValidationError + + ep = _make_endpoint() + s = ModelSubtask(subtask_id="s1", description="task") + with pytest.raises(ValidationError): + ModelSwarmFanoutRequest( + subtasks=(s,), + assignments={"s1": "ep1"}, + endpoints=(ep,), + config=ModelSwarmConfig(), + correlation_id="c", + run_id="r", + worker_assignments={"s1": "w1"}, # type: ignore[call-arg] + ) diff --git a/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_models.py b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_models.py new file mode 100644 index 000000000..6bd805717 --- /dev/null +++ b/src/omnimarket/nodes/node_swarm_fanout_orchestrator/tests/test_models.py @@ -0,0 +1,207 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Model validation tests for node_swarm_fanout_orchestrator.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.enums import ( + EnumExecutionStatus, + EnumFanoutFsmState, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_subtask import ( + ModelSubtask, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_config import ( + ModelSwarmConfig, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_dispatch import ( + ModelSwarmDispatch, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_endpoint import ( + ModelSwarmEndpoint, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request import ( + ModelSwarmFanoutRequest, +) +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_result import ( + ModelSwarmFanoutResult, +) + + +@pytest.mark.unit +def test_model_subtask_defaults() -> None: + s = ModelSubtask(subtask_id="x", description="desc") + assert s.depends_on == () + assert s.estimated_tokens == 0 + assert s.category == "general" + + +@pytest.mark.unit +def test_model_subtask_frozen() -> None: + s = ModelSubtask(subtask_id="x", description="desc") + with pytest.raises((ValidationError, TypeError)): + s.subtask_id = "y" # type: ignore[misc] + + +@pytest.mark.unit +def test_model_swarm_config_defaults() -> None: + cfg = ModelSwarmConfig() + assert cfg.max_parallel_subtasks == 4 + assert cfg.fallback_policy_enabled is True + + +@pytest.mark.unit +def test_model_swarm_dispatch_required_fields() -> None: + d = ModelSwarmDispatch( + subtask_id="s1", + endpoint_id="ep1", + model_id="m1", + base_url="http://x", + execution_status=EnumExecutionStatus.SUCCEEDED, + ) + assert d.response_text == "" + assert d.retry_count == 0 + assert d.wave == 0 + + +@pytest.mark.unit +def test_model_swarm_dispatch_extra_field_rejected() -> None: + with pytest.raises(ValidationError): + ModelSwarmDispatch( # type: ignore[call-arg] + subtask_id="s1", + endpoint_id="ep1", + model_id="m1", + base_url="http://x", + execution_status=EnumExecutionStatus.SUCCEEDED, + unknown_field="bad", + ) + + +@pytest.mark.unit +def test_model_swarm_endpoint_endpoint_ref_default_empty() -> None: + ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://x", model_id="m1") + assert ep.endpoint_ref == "" + + +@pytest.mark.unit +def test_model_swarm_endpoint_with_ref() -> None: + ep = ModelSwarmEndpoint( + endpoint_id="ep1", + base_url="http://x", + model_id="m1", + endpoint_ref="LLM_LOCAL_PRIMARY_URL", + ) + assert ep.endpoint_ref == "LLM_LOCAL_PRIMARY_URL" + + +@pytest.mark.unit +def test_model_swarm_fanout_request_roundtrip() -> None: + ep = ModelSwarmEndpoint( + endpoint_id="ep1", + base_url="http://ep1", + model_id="m1", + endpoint_ref="LLM_LOCAL_PRIMARY_URL", + ) + s = ModelSubtask(subtask_id="s1", description="task") + req = ModelSwarmFanoutRequest( + subtasks=(s,), + assignments={"s1": "ep1"}, + endpoints=(ep,), + config=ModelSwarmConfig(), + correlation_id="c1", + run_id="r1", + ) + assert req.subtasks[0].subtask_id == "s1" + assert req.assignments["s1"] == "ep1" + + +@pytest.mark.unit +def test_model_swarm_fanout_request_no_dispatch_mode() -> None: + """ModelSwarmFanoutRequest must reject dispatch_mode (removed field).""" + ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://x", model_id="m1") + s = ModelSubtask(subtask_id="s1", description="task") + with pytest.raises(ValidationError): + ModelSwarmFanoutRequest( # type: ignore[call-arg] + subtasks=(s,), + assignments={"s1": "ep1"}, + endpoints=(ep,), + config=ModelSwarmConfig(), + correlation_id="c1", + run_id="r1", + dispatch_mode="direct", + ) + + +@pytest.mark.unit +def test_model_swarm_fanout_request_no_worker_assignments() -> None: + """ModelSwarmFanoutRequest must reject worker_assignments (removed field).""" + ep = ModelSwarmEndpoint(endpoint_id="ep1", base_url="http://x", model_id="m1") + s = ModelSubtask(subtask_id="s1", description="task") + with pytest.raises(ValidationError): + ModelSwarmFanoutRequest( # type: ignore[call-arg] + subtasks=(s,), + assignments={"s1": "ep1"}, + endpoints=(ep,), + config=ModelSwarmConfig(), + correlation_id="c1", + run_id="r1", + worker_assignments={"s1": "w1"}, + ) + + +@pytest.mark.unit +def test_model_swarm_fanout_result_terminal_fields() -> None: + d1 = ModelSwarmDispatch( + subtask_id="s1", + endpoint_id="ep1", + model_id="m1", + base_url="http://x", + execution_status=EnumExecutionStatus.SUCCEEDED, + latency_ms=100, + ) + d2 = ModelSwarmDispatch( + subtask_id="s2", + endpoint_id="ep1", + model_id="m1", + base_url="http://x", + execution_status=EnumExecutionStatus.FAILED, + latency_ms=50, + ) + result = ModelSwarmFanoutResult( + dispatches=(d1, d2), + wall_latency_ms=120, + sum_subtask_latency_ms=150, + completed_count=1, + failed_count=1, + degraded=True, + aggregation_mode="collect_all", + ) + assert result.sum_subtask_latency_ms == 150 + assert len(result.dispatches) == 2 + assert result.completed_count == 1 + assert result.failed_count == 1 + assert result.degraded is True + assert result.aggregation_mode == "collect_all" + + +@pytest.mark.unit +def test_enum_execution_status_values() -> None: + assert EnumExecutionStatus.SUCCEEDED.value == "succeeded" + assert EnumExecutionStatus.FAILED.value == "failed" + assert EnumExecutionStatus.TIMEOUT.value == "timeout" + assert ( + EnumExecutionStatus.SKIPPED_DEPENDENCY_FAILED.value + == "skipped_dependency_failed" + ) + + +@pytest.mark.unit +def test_enum_fanout_fsm_state_values() -> None: + assert EnumFanoutFsmState.PLANNING.value == "PLANNING" + assert EnumFanoutFsmState.DISPATCHING.value == "DISPATCHING" + assert EnumFanoutFsmState.COLLECTING.value == "COLLECTING" + assert EnumFanoutFsmState.COMPLETED.value == "COMPLETED" + assert EnumFanoutFsmState.FAILED.value == "FAILED" diff --git a/src/omnimarket/nodes/node_swarm_registry_compute/contracts/endpoint_registry.yaml b/src/omnimarket/nodes/node_swarm_registry_compute/contracts/endpoint_registry.yaml index c85a6b78d..3f47413b2 100644 --- a/src/omnimarket/nodes/node_swarm_registry_compute/contracts/endpoint_registry.yaml +++ b/src/omnimarket/nodes/node_swarm_registry_compute/contracts/endpoint_registry.yaml @@ -3,6 +3,7 @@ endpoints: - id: "qwen3.6-35b-a3b" base_url: "http://192.168.86.201:8000/v1" # onex-allow-internal-ip OMN-11811 reason="lab LAN endpoint declared in operator registry contract" model_id: "Qwen3.6-35B-A3B" # onex-allow-model-id OMN-11959 reason="operator registry must pin exact vLLM-served model ID" + endpoint_ref: "LLM_LOCAL_PRIMARY_URL" provider: "vllm" capabilities: [code_generation, structured_output, refactoring, analysis, general] context_window: 131072 @@ -13,6 +14,7 @@ endpoints: - id: "qwen3.6-27b-mtp-iq4xs" base_url: "http://192.168.86.201:8001/v1" # onex-allow-internal-ip OMN-11811 reason="lab LAN endpoint declared in operator registry contract" model_id: "Qwen3.6-27B-MTP-IQ4_XS.gguf" # onex-allow-model-id OMN-11959 reason="operator registry must pin exact vLLM-served model ID" + endpoint_ref: "LLM_CODER_FAST_URL" provider: "vllm" capabilities: [reasoning, analysis, math, general] context_window: 98304 @@ -23,6 +25,7 @@ endpoints: - id: "text-embedding-gte" base_url: "http://192.168.86.201:8002/v1" # onex-allow-internal-ip OMN-11811 reason="lab LAN endpoint declared in operator registry contract" model_id: "text-embedding-gte" # onex-allow-model-id OMN-11959 reason="operator registry must pin exact vLLM-served model ID" + endpoint_ref: "LLM_LOCAL_EMBEDDINGS_URL" provider: "vllm" capabilities: [embeddings] context_window: 8192 @@ -33,6 +36,7 @@ endpoints: - id: "deepseek-v4-flash" base_url: "http://192.168.86.200:8101/v1" # onex-allow-internal-ip OMN-11811 reason="lab LAN endpoint declared in operator registry contract" model_id: "deepseek-v4-flash" + endpoint_ref: "LLM_LOCAL_FAST_URL" provider: "ds4" capabilities: [reasoning, analysis, math, planning, general] context_window: 131072 @@ -43,6 +47,7 @@ endpoints: - id: "deepseek-v4-pro" base_url: "http://192.168.86.200:8101/v1" # onex-allow-internal-ip OMN-11811 reason="lab LAN endpoint declared in operator registry contract" model_id: "deepseek-v4-pro" + endpoint_ref: "LLM_REMOTE_FRONTIER_URL" provider: "ds4" capabilities: [reasoning, analysis, math, planning, general] context_window: 131072 diff --git a/tests/test_swarm_schema_alignment.py b/tests/test_swarm_schema_alignment.py index 0fb15c68c..748312b92 100644 --- a/tests/test_swarm_schema_alignment.py +++ b/tests/test_swarm_schema_alignment.py @@ -46,10 +46,10 @@ from omnimarket.nodes.node_swarm_endpoint_health_effect.models.model_swarm_health_check_result import ( ModelSwarmHealthCheckResult, ) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_request import ( +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_request import ( ModelSwarmFanoutRequest, ) -from omnimarket.nodes.node_swarm_fanout_effect.models.model_swarm_fanout_result import ( +from omnimarket.nodes.node_swarm_fanout_orchestrator.models.model_swarm_fanout_result import ( ModelSwarmFanoutResult, ) from omnimarket.nodes.node_swarm_registry_compute.models.model_swarm_endpoint_selection_request import (