diff --git a/pyproject.toml b/pyproject.toml index 8e3be66de3..85eb4764c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ # omnibase-core = { git = "https://github.com/org/omnibase_core.git", rev = "SHA" } # Remember to switch back to PyPI versions before merging to main. # - "omnibase-core>=0.39.0", + "omnibase-core>=0.40.0,<0.41.0", "omnibase-spi>=0.20.4", "omnimarket>=0.1.0", # ============================================================================ @@ -103,6 +103,7 @@ dependencies = [ "pydantic-settings>=2.2.1,<3.0.0", # Settings management "jsonschema>=4.20.0,<5.0.0", # JSON schema validation "watchdog>=4.0.0", # Filesystem event monitoring (HandlerContractFileWatcher, OMN-3940) + "a2a-sdk>=0.3.4,<0.4.0", # Real A2A client/server protocol support for remote-agent invocation # Security: CVE-2026-32597 (HIGH) - PyJWT <2.12.0 accepts unknown `crit` header extensions. # Transitive dependency (via mcp/httpx/authlib). Explicitly declared to enforce minimum version. "pyjwt>=2.12.0", @@ -160,15 +161,15 @@ dev = [ [tool.uv] override-dependencies = [ - "omnibase-core>=0.39.0", + "omnibase-core>=0.40.0,<0.41.0", ] [tool.uv.sources] onex-change-control = { git = "https://github.com/OmniNode-ai/onex_change_control.git", branch = "main" } omnimarket = { git = "https://github.com/OmniNode-ai/omnimarket.git", branch = "main" } -# Pinned to core main post-OMN-9196 (resolver models + enum_handler_resolution_outcome). -# PyPI release v0.39.0 (2026-04-06) predates PR 846 — switch back after v0.40.0 ships. -omnibase-core = { git = "https://github.com/OmniNode-ai/omnibase_core.git", rev = "9d64d5a3944477093824a53ca931fd17a4b6a8e1" } +# Pinned to core main post-OMN-9637 (delegation models — ModelRemoteTaskState et al.). +# PyPI release v0.39.0 (2026-04-06) predates PR 912 — switch back after v0.40.0 ships. +omnibase-core = { git = "https://github.com/OmniNode-ai/omnibase_core.git", rev = "6a65f885303e29b4ddc36a66a0cb860537f47316" } # Pinned to spi main post-OMN-9197 (ProtocolHandlerResolver + ProtocolHandlerOwnershipQuery). # PyPI release v0.20.4 (2026-04-03) predates PR 186 — switch back after v0.20.5 ships. omnibase-spi = { git = "https://github.com/OmniNode-ai/omnibase_spi.git", rev = "a9380008fb4a2f2a784d374389260d896455c3b2" } @@ -508,6 +509,7 @@ node_registration_reducer = "omnibase_infra.nodes.node_registration_reducer" node_registration_storage_effect = "omnibase_infra.nodes.node_registration_storage_effect" node_registry_api_effect = "omnibase_infra.nodes.node_registry_api_effect" node_registry_effect = "omnibase_infra.nodes.node_registry_effect" +node_remote_agent_invoke_effect = "omnibase_infra.nodes.node_remote_agent_invoke_effect" node_reward_binder_effect = "omnibase_infra.nodes.node_reward_binder_effect" node_routing_orchestrator = "omnibase_infra.nodes.node_routing_orchestrator" node_routing_score_reducer = "omnibase_infra.nodes.node_routing_score_reducer" diff --git a/scripts/check_subscribe_wiring_health.py b/scripts/check_subscribe_wiring_health.py index f4fc02b9d0..7e8968ac90 100644 --- a/scripts/check_subscribe_wiring_health.py +++ b/scripts/check_subscribe_wiring_health.py @@ -157,6 +157,9 @@ "onex.int.platform.runtime-tick.v1": "Published by runtime scheduler | owner: jonah | expiry: 2026-09-01", # Onboarding — triggered by omniclaude /onboarding skill, not Kafka publisher "onex.cmd.omnibase-infra.onboarding-start.v1": "Triggered by /onboarding skill via claude -p, not Kafka | owner: jonah | expiry: 2026-12-01", + # Remote-agent invoke — emitted by node_delegation_orchestrator (OMN-9620 epic); + # its contract addition is tracked separately and pending in another wave. + "onex.cmd.omnibase-infra.remote-agent-invoke.v1": "Publisher pending in delegation_orchestrator (OMN-9620 epic) | owner: jonah | expiry: 2026-09-01", } # fmt: on diff --git a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py b/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py index 389ebdc9c6..da112a1de2 100644 --- a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py +++ b/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py @@ -29,7 +29,9 @@ class EnumOmnibaseInfraTopic(str, Enum): CMD_LLM_EMBEDDING_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-embedding-request.v1" # onex.cmd.omnibase-infra.llm-embedding-request.v1 CMD_LLM_INFERENCE_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-inference-request.v1" # onex.cmd.omnibase-infra.llm-inference-request.v1 CMD_ONBOARDING_START_V1 = "onex.cmd.omnibase-infra.onboarding-start.v1" # onex.cmd.omnibase-infra.onboarding-start.v1 + CMD_REMOTE_AGENT_INVOKE_V1 = "onex.cmd.omnibase-infra.remote-agent-invoke.v1" # onex.cmd.omnibase-infra.remote-agent-invoke.v1 CMD_VECTOR_STORE_REQUEST_V1 = "onex.cmd.omnibase-infra.vector-store-request.v1" # onex.cmd.omnibase-infra.vector-store-request.v1 + EVT_AGENT_TASK_LIFECYCLE_V1 = "onex.evt.omnibase-infra.agent-task-lifecycle.v1" # onex.evt.omnibase-infra.agent-task-lifecycle.v1 EVT_BASELINES_COMPUTED_V1 = "onex.evt.omnibase-infra.baselines-computed.v1" # onex.evt.omnibase-infra.baselines-computed.v1 EVT_CHAIN_LEARN_COMPLETE_V1 = "onex.evt.omnibase-infra.chain-learn-complete.v1" # onex.evt.omnibase-infra.chain-learn-complete.v1 EVT_CHAIN_LEARN_FAILED_V1 = "onex.evt.omnibase-infra.chain-learn-failed.v1" # onex.evt.omnibase-infra.chain-learn-failed.v1 diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/__init__.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/__init__.py new file mode 100644 index 0000000000..616477f180 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/__init__.py @@ -0,0 +1,9 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Remote agent invocation effect node.""" + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.node import ( + NodeRemoteAgentInvokeEffect, +) + +__all__ = ["NodeRemoteAgentInvokeEffect"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/contract.yaml b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/contract.yaml new file mode 100644 index 0000000000..261621d35a --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/contract.yaml @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +# +# ONEX Node Contract +# Node: NodeRemoteAgentInvokeEffect +# +# Submits routed remote-agent invocation commands to protocol-specific +# transports and emits remote task lifecycle events. +name: "node_remote_agent_invoke_effect" +contract_name: "node_remote_agent_invoke_effect" +node_name: "node_remote_agent_invoke_effect" +contract_version: + major: 0 + minor: 1 + patch: 0 +node_version: + major: 0 + minor: 1 + patch: 0 +node_type: "EFFECT_GENERIC" +description: > + Remote-agent invocation effect node for the A2A delegation bridge. Consumes invocation commands emitted by the delegation orchestrator, delegates task submission and watch operations to protocol-specific handlers, persists remote task state, and emits lifecycle events back to the orchestration pipeline. + +input_model: + name: "ModelInvocationCommand" + module: "omnibase_infra.nodes.node_remote_agent_invoke_effect.models" + description: "Invocation command selected by delegation routing." +output_model: + name: "ModelAgentTaskLifecycleEvent" + module: "omnibase_infra.nodes.node_remote_agent_invoke_effect.models" + description: "Lifecycle event emitted while watching a remote agent task." +handler_routing: + routing_strategy: "operation_match" + handlers: + - operation: "agent.a2a_task" + handler: + name: "HandlerA2ATask" + module: "omnibase_infra.nodes.node_remote_agent_invoke_effect.handlers.handler_a2a_task" + handler_type: "HTTP" + description: "Submit remote A2A tasks and persist their initial lifecycle state." + execution_mode: "single" + max_timeout_seconds: 600 +event_bus: + version: + major: 1 + minor: 0 + patch: 0 + subscribe_topics: + - "onex.cmd.omnibase-infra.remote-agent-invoke.v1" + publish_topics: + - "onex.evt.omnibase-infra.agent-task-lifecycle.v1" +consumed_events: + - topic: "onex.cmd.omnibase-infra.remote-agent-invoke.v1" + event_type: "RemoteAgentInvokeCommand" + message_category: "COMMAND" + description: "Command requesting remote-agent task submission and lifecycle watch." +published_events: + - topic: "onex.evt.omnibase-infra.agent-task-lifecycle.v1" + event_type: "AgentTaskLifecycleEvent" + description: "Remote-agent lifecycle event emitted by protocol handlers." +health_check: + enabled: true + endpoint: "/health" + interval_seconds: 30 + checks: + - name: "remote_agent_transport_handlers_loaded" + type: "handler" + critical: true + - name: "remote_task_state_persistence_available" + type: "database" + critical: true +capabilities: + - name: "remote_agent_invocation" + description: "Submit tasks to remote agents via protocol-specific handlers." + - name: "remote_task_lifecycle_watch" + description: "Watch remote task state and emit typed lifecycle events." + - name: "remote_task_restart_resumption" + description: "Reload unfinished remote task state on startup." +metadata: + description: "A2A delegation bridge effect node scaffold." + author: "OmniNode Team" + license: "MIT" + created: "2026-04-25" + tags: + - effect + - delegation + - a2a + - remote-agent diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/__init__.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/__init__.py new file mode 100644 index 0000000000..85422c8187 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/__init__.py @@ -0,0 +1,9 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Remote agent invocation handlers.""" + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.handlers.handler_a2a_task import ( + HandlerA2ATask, +) + +__all__ = ["HandlerA2ATask"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/handler_a2a_task.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/handler_a2a_task.py new file mode 100644 index 0000000000..34c116e0e9 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/handlers/handler_a2a_task.py @@ -0,0 +1,556 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""A2A submit/watch handler for node_remote_agent_invoke_effect.""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Callable, Mapping, Sequence +from datetime import UTC, datetime +from typing import Protocol +from uuid import UUID + +import aiohttp +import httpx +from a2a.client.base_client import BaseClient +from a2a.client.client import ClientConfig +from a2a.client.client_factory import ClientFactory +from a2a.types import ( + Message, + MessageSendConfiguration, + Part, + Role, + TaskQueryParams, + TextPart, +) +from a2a.types import Task as A2ATask + +from omnibase_core.enums.enum_agent_protocol import EnumAgentProtocol +from omnibase_core.enums.enum_agent_task_lifecycle_type import ( + EnumAgentTaskLifecycleType, +) +from omnibase_core.models.common.model_schema_value import ModelSchemaValue +from omnibase_core.models.delegation.model_a2a_task_request import ModelA2ATaskRequest +from omnibase_core.models.delegation.model_a2a_task_response import ModelA2ATaskResponse +from omnibase_core.models.delegation.model_agent_task_lifecycle_event import ( + ModelAgentTaskLifecycleEvent, +) +from omnibase_core.models.delegation.model_invocation_command import ( + ModelInvocationCommand, +) +from omnibase_core.models.delegation.model_remote_task_state import ModelRemoteTaskState +from omnibase_core.models.delegation.model_target_agent import ModelTargetAgent +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope +from omnibase_infra.enums import ( + EnumHandlerType, + EnumHandlerTypeCategory, + EnumInfraTransportType, +) +from omnibase_infra.errors import ( + InfraConnectionError, + ModelInfraErrorContext, +) +from omnibase_infra.nodes.node_remote_agent_invoke_effect.persistence.remote_task_state_repository import ( + RemoteTaskStateRepository, +) +from omnibase_infra.protocols.protocol_event_bus_like import ProtocolEventBusLike + +_STATUS_MAP: dict[str, EnumAgentTaskLifecycleType] = { + "submitted": EnumAgentTaskLifecycleType.SUBMITTED, + "accepted": EnumAgentTaskLifecycleType.ACCEPTED, + "artifact": EnumAgentTaskLifecycleType.ARTIFACT, + "working": EnumAgentTaskLifecycleType.PROGRESS, + "in_progress": EnumAgentTaskLifecycleType.PROGRESS, + "input_required": EnumAgentTaskLifecycleType.PROGRESS, + "auth_required": EnumAgentTaskLifecycleType.PROGRESS, + "completed": EnumAgentTaskLifecycleType.COMPLETED, + "failed": EnumAgentTaskLifecycleType.FAILED, + "timed_out": EnumAgentTaskLifecycleType.TIMED_OUT, + "canceled": EnumAgentTaskLifecycleType.CANCELED, +} +_TERMINAL_LIFECYCLES = { + EnumAgentTaskLifecycleType.COMPLETED, + EnumAgentTaskLifecycleType.FAILED, + EnumAgentTaskLifecycleType.TIMED_OUT, + EnumAgentTaskLifecycleType.CANCELED, +} + + +class ProtocolA2ATransport(Protocol): + """Small transport boundary so unit tests can avoid protocol plumbing.""" + + async def submit( + self, + *, + target: ModelTargetAgent, + request_model: ModelA2ATaskRequest, + ) -> ModelA2ATaskResponse: + """Submit a task request to the remote agent.""" + + async def get_task( + self, + *, + target: ModelTargetAgent, + remote_task_handle: str, + ) -> ModelA2ATaskResponse: + """Fetch the latest state of a remote task.""" + + +def map_remote_status(raw_status: str) -> EnumAgentTaskLifecycleType: + """Map A2A peer status strings into typed lifecycle events.""" + normalized = raw_status.strip().lower().replace("-", "_") + try: + return _STATUS_MAP[normalized] + except KeyError as exc: + msg = f"Unknown remote agent status: {raw_status}" + raise ValueError(msg) from exc + + +class A2ASdkTransport: + """Real A2A transport backed by the upstream A2A SDK.""" + + def __init__(self, *, request_timeout_seconds: float = 180.0) -> None: + self._request_timeout_seconds = request_timeout_seconds + + async def submit( + self, + *, + target: ModelTargetAgent, + request_model: ModelA2ATaskRequest, + ) -> ModelA2ATaskResponse: + async with httpx.AsyncClient(timeout=self._request_timeout_seconds) as client: + a2a_client = await ClientFactory.connect( + target.base_url, + client_config=ClientConfig( + streaming=False, + polling=True, + httpx_client=client, + ), + ) + typed_client = _require_base_client(a2a_client) + message = Message( + message_id=str(request_model.correlation_id), + role=Role.user, + parts=[Part(root=TextPart(text=_build_prompt(request_model)))], + ) + async for event in typed_client.send_message( + message, + configuration=MessageSendConfiguration(blocking=False), + ): + if isinstance(event, Message): + msg = "Unexpected direct message response from A2A peer" + raise ValueError(msg) + task, _update = event + return _task_to_response(task) + + msg = "A2A peer returned no task from send_message" + raise ValueError(msg) + + async def get_task( + self, + *, + target: ModelTargetAgent, + remote_task_handle: str, + ) -> ModelA2ATaskResponse: + async with httpx.AsyncClient(timeout=self._request_timeout_seconds) as client: + a2a_client = await ClientFactory.connect( + target.base_url, + client_config=ClientConfig( + streaming=False, + polling=True, + httpx_client=client, + ), + ) + typed_client = _require_base_client(a2a_client) + task = await typed_client.get_task( + TaskQueryParams(id=remote_task_handle, history_length=20) + ) + return _task_to_response(task) + + +def _require_base_client(client: object) -> BaseClient: + if isinstance(client, BaseClient): + return client + msg = f"Unsupported A2A client type: {type(client).__name__}" + raise TypeError(msg) + + +def _build_prompt(request_model: ModelA2ATaskRequest) -> str: + payload = {key: value.to_value() for key, value in request_model.input.items()} + prompt = payload.get("prompt") + if isinstance(prompt, str) and prompt.strip(): + return prompt + return json.dumps( + { + "skill_id": request_model.skill_ref, + "correlation_id": str(request_model.correlation_id), + "input": payload, + }, + indent=2, + sort_keys=True, + ) + + +def _task_to_response(task: A2ATask) -> ModelA2ATaskResponse: + raw_status = getattr(task.status.state, "value", str(task.status.state)) + return ModelA2ATaskResponse( + remote_task_handle=task.id, + status=map_remote_status(raw_status), + artifacts=_extract_artifacts(task), + error=_extract_error(task), + ) + + +def _extract_error(task: A2ATask) -> str | None: + state = getattr(task.status.state, "value", str(task.status.state)).lower() + if state not in {"failed", "canceled", "timed_out"}: + return None + status_message = getattr(task.status, "message", None) + if status_message is None: + return None + texts = _extract_texts(getattr(status_message, "parts", None) or []) + if texts: + return "\n".join(texts) + return None + + +def _extract_artifacts(task: A2ATask) -> list[dict[str, ModelSchemaValue]]: + task_artifacts = getattr(task, "artifacts", None) or [] + extracted: list[dict[str, ModelSchemaValue]] = [] + for artifact in task_artifacts: + payload = _parts_to_payload(getattr(artifact, "parts", None) or []) + if payload is None: + continue + if isinstance(payload, dict): + extracted.append( + { + key: ModelSchemaValue.from_value(value) + for key, value in payload.items() + } + ) + continue + extracted.append({"text": ModelSchemaValue.from_value(payload)}) + return extracted + + +def _parts_to_payload(parts: Sequence[object]) -> object | None: + texts = _extract_texts(parts) + if not texts: + return None + combined = "\n".join(texts).strip() + if not combined: + return None + try: + parsed = json.loads(combined) + except json.JSONDecodeError: + return combined + return parsed + + +def _extract_texts(parts: Sequence[object]) -> list[str]: + texts: list[str] = [] + for part in parts: + candidate = getattr(part, "root", part) + kind = getattr(candidate, "kind", None) + if kind == "text": + text = getattr(candidate, "text", None) + if isinstance(text, str) and text: + texts.append(text) + return texts + + +class HandlerA2ATask: + """Submit-path handler for A2A task invocation.""" + + def __init__( + self, + *, + repository: RemoteTaskStateRepository, + event_bus: ProtocolEventBusLike, + target_registry: Mapping[str, ModelTargetAgent], + lifecycle_topic: str, + http_session: aiohttp.ClientSession | None = None, + clock: Callable[[], datetime] | None = None, + poll_interval_seconds: float = 1.0, + transport: ProtocolA2ATransport | None = None, + ) -> None: + self._repository = repository + self._event_bus = event_bus + self._target_registry = target_registry + self._lifecycle_topic = lifecycle_topic + self._http_session = http_session + self._clock = clock or (lambda: datetime.now(UTC)) + self._poll_interval_seconds = poll_interval_seconds + self._transport = transport or A2ASdkTransport() + + @property + def handler_type(self) -> EnumHandlerType: + return EnumHandlerType.INFRA_HANDLER + + @property + def handler_category(self) -> EnumHandlerTypeCategory: + return EnumHandlerTypeCategory.EFFECT + + async def submit(self, command: ModelInvocationCommand) -> ModelA2ATaskResponse: + """Submit a command to a remote A2A peer and persist the returned handle.""" + target = self._resolve_target(command.target_ref) + now = self._clock() + + submitted_state = ModelRemoteTaskState( + task_id=command.task_id, + invocation_kind=command.invocation_kind, + protocol=command.agent_protocol, + target_ref=command.target_ref, + remote_task_handle=None, + correlation_id=command.correlation_id, + status=EnumAgentTaskLifecycleType.SUBMITTED, + last_remote_status="submitted", + last_emitted_event_type=EnumAgentTaskLifecycleType.SUBMITTED, + submitted_at=now, + updated_at=now, + ) + await self._repository.upsert(submitted_state) + await self._emit_submitted( + command=command, occurred_at=now, remote_task_handle=None + ) + + request_model = ModelA2ATaskRequest( + skill_ref=command.target_ref, + input=command.payload, + correlation_id=command.correlation_id, + ) + try: + response = await self._transport.submit( + target=target, request_model=request_model + ) + except Exception as exc: + failed_at = self._clock() + await self._repository.upsert( + submitted_state.model_copy( + update={ + "status": EnumAgentTaskLifecycleType.FAILED, + "last_remote_status": "failed", + "last_emitted_event_type": EnumAgentTaskLifecycleType.FAILED, + "completed_at": failed_at, + "updated_at": failed_at, + "error": str(exc), + } + ) + ) + context = ModelInfraErrorContext.with_correlation( + correlation_id=command.correlation_id, + transport_type=EnumInfraTransportType.HTTP, + operation="a2a_submit", + ) + raise InfraConnectionError( + f"A2A submit failed for target {command.target_ref}: {exc}", + context=context, + ) from exc + + await self._repository.upsert( + submitted_state.model_copy( + update={ + "remote_task_handle": response.remote_task_handle, + "last_remote_status": response.status.value, + "updated_at": self._clock(), + } + ) + ) + return response + + async def watch( + self, + remote_task_handle: str, + correlation_id: UUID, + ) -> list[ModelAgentTaskLifecycleEvent]: + """Poll tasks.get until the remote task reaches a terminal lifecycle.""" + row = await self._repository.get_by_remote_task_handle( + remote_task_handle, + correlation_id=correlation_id, + ) + if row is None: + msg = f"Unknown remote_task_handle: {remote_task_handle}" + raise ValueError(msg) + + target = self._resolve_target(row.target_ref) + last_seen = row.last_emitted_event_type + events: list[ModelAgentTaskLifecycleEvent] = [] + last_artifact_payload: list[dict[str, ModelSchemaValue]] | None = None + + while True: + try: + response = await self._transport.get_task( + target=target, + remote_task_handle=remote_task_handle, + ) + except Exception as exc: + context = ModelInfraErrorContext.with_correlation( + correlation_id=correlation_id, + transport_type=EnumInfraTransportType.HTTP, + operation="a2a_watch", + ) + raise InfraConnectionError( + f"A2A watch failed for handle {remote_task_handle}: {exc}", + context=context, + ) from exc + mapped = response.status + raw_status = response.status.value + now = self._clock() + + if response.artifacts and response.artifacts != last_artifact_payload: + artifact_event = await self._emit_lifecycle( + task_id=row.task_id, + correlation_id=correlation_id, + lifecycle_type=EnumAgentTaskLifecycleType.ARTIFACT, + remote_task_handle=remote_task_handle, + artifact=response.artifacts[0], + occurred_at=now, + remote_status=raw_status, + error=response.error, + ) + await self._repository.upsert( + row.model_copy( + update={ + "status": mapped, + "last_remote_status": raw_status, + "last_emitted_event_type": EnumAgentTaskLifecycleType.ARTIFACT, + "updated_at": now, + "completed_at": now + if mapped in _TERMINAL_LIFECYCLES + else None, + "error": response.error, + } + ), + correlation_id=correlation_id, + ) + row = row.model_copy( + update={ + "status": mapped, + "last_remote_status": raw_status, + "last_emitted_event_type": EnumAgentTaskLifecycleType.ARTIFACT, + "updated_at": now, + "completed_at": now if mapped in _TERMINAL_LIFECYCLES else None, + "error": response.error, + } + ) + events.append(artifact_event) + last_artifact_payload = response.artifacts + + if mapped is last_seen: + if mapped in _TERMINAL_LIFECYCLES: + return events + await asyncio.sleep(self._poll_interval_seconds) + continue + + lifecycle_event = await self._emit_lifecycle( + task_id=row.task_id, + correlation_id=correlation_id, + lifecycle_type=mapped, + remote_task_handle=remote_task_handle, + artifact=response.artifacts[0] if response.artifacts else None, + occurred_at=now, + remote_status=raw_status, + error=response.error, + ) + await self._repository.upsert( + row.model_copy( + update={ + "status": mapped, + "last_remote_status": raw_status, + "last_emitted_event_type": mapped, + "updated_at": now, + "completed_at": now if mapped in _TERMINAL_LIFECYCLES else None, + "error": response.error, + } + ), + correlation_id=correlation_id, + ) + row = row.model_copy( + update={ + "status": mapped, + "last_remote_status": raw_status, + "last_emitted_event_type": mapped, + "updated_at": now, + "completed_at": now if mapped in _TERMINAL_LIFECYCLES else None, + "error": response.error, + } + ) + last_seen = mapped + events.append(lifecycle_event) + if mapped in _TERMINAL_LIFECYCLES: + return events + await asyncio.sleep(self._poll_interval_seconds) + + def _resolve_target(self, target_ref: str) -> ModelTargetAgent: + try: + target = self._target_registry[target_ref] + except KeyError as exc: + msg = f"Unknown target_ref: {target_ref}" + raise ValueError(msg) from exc + if target.protocol is not EnumAgentProtocol.A2A: + msg = f"Target {target_ref} is not configured for A2A" + raise ValueError(msg) + return target + + async def _emit_submitted( + self, + *, + command: ModelInvocationCommand, + occurred_at: datetime, + remote_task_handle: str | None, + ) -> None: + await self._emit_lifecycle( + task_id=command.task_id, + correlation_id=command.correlation_id, + lifecycle_type=EnumAgentTaskLifecycleType.SUBMITTED, + remote_task_handle=remote_task_handle, + artifact={ + "target_ref": ModelSchemaValue.from_value(command.target_ref), + }, + occurred_at=occurred_at, + remote_status="submitted", + ) + + async def _emit_lifecycle( + self, + *, + task_id: UUID, + correlation_id: UUID, + lifecycle_type: EnumAgentTaskLifecycleType, + remote_task_handle: str | None, + occurred_at: datetime, + artifact: dict[str, ModelSchemaValue] | None = None, + remote_status: str | None = None, + error: str | None = None, + ) -> ModelAgentTaskLifecycleEvent: + lifecycle_event = ModelAgentTaskLifecycleEvent( + task_id=task_id, + correlation_id=correlation_id, + lifecycle_type=lifecycle_type, + remote_task_handle=remote_task_handle, + artifact=artifact, + occurred_at=occurred_at, + remote_status=remote_status, + error=error, + ) + envelope = ModelEventEnvelope[dict[str, object]]( + correlation_id=correlation_id, + payload=lifecycle_event.model_dump(mode="json"), + ) + await self._event_bus.publish( + self._lifecycle_topic, + key=str(task_id).encode("utf-8"), + value=envelope.model_dump_json().encode("utf-8"), + ) + return lifecycle_event + + +__all__ = [ + "A2ASdkTransport", + "HandlerA2ATask", + "ProtocolA2ATransport", + "_STATUS_MAP", + "_TERMINAL_LIFECYCLES", + "map_remote_status", +] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/__init__.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/__init__.py new file mode 100644 index 0000000000..ebea03f565 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/__init__.py @@ -0,0 +1,19 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Models for the remote-agent invoke effect node.""" + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.models.enum_agent_task_status import ( + EnumAgentTaskStatus, +) +from omnibase_infra.nodes.node_remote_agent_invoke_effect.models.model_agent_task_lifecycle_event import ( + ModelAgentTaskLifecycleEvent, +) +from omnibase_infra.nodes.node_remote_agent_invoke_effect.models.model_invocation_command import ( + ModelInvocationCommand, +) + +__all__ = [ + "EnumAgentTaskStatus", + "ModelAgentTaskLifecycleEvent", + "ModelInvocationCommand", +] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/enum_agent_task_status.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/enum_agent_task_status.py new file mode 100644 index 0000000000..e88428f139 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/enum_agent_task_status.py @@ -0,0 +1,20 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Agent task lifecycle status enum for the remote-agent invoke effect node.""" + +from __future__ import annotations + +from enum import Enum + + +class EnumAgentTaskStatus(str, Enum): + """Status values for remote agent task lifecycle events.""" + + SUBMITTED = "submitted" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +__all__ = ["EnumAgentTaskStatus"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_agent_task_lifecycle_event.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_agent_task_lifecycle_event.py new file mode 100644 index 0000000000..8b87021d41 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_agent_task_lifecycle_event.py @@ -0,0 +1,29 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Stub lifecycle event model for the remote-agent invoke effect node. + +Full model definition deferred to OMN-9625/9626/9627 core delegation stack. +""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.models.enum_agent_task_status import ( + EnumAgentTaskStatus, +) + + +class ModelAgentTaskLifecycleEvent(BaseModel): + """Lifecycle event emitted while watching a remote agent task.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + correlation_id: UUID = Field(..., description="Tracks the originating invocation.") + status: EnumAgentTaskStatus = Field(..., description="Task lifecycle status.") + agent_id: UUID = Field(..., description="Agent that processed the task.") + + +__all__ = ["ModelAgentTaskLifecycleEvent"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_invocation_command.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_invocation_command.py new file mode 100644 index 0000000000..97e28f46c8 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/models/model_invocation_command.py @@ -0,0 +1,29 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Stub invocation command model for the remote-agent invoke effect node. + +Full model definition deferred to OMN-9625/9626/9627 core delegation stack. +""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + + +class ModelInvocationCommand(BaseModel): + """Command requesting remote-agent task submission and lifecycle watch.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + correlation_id: UUID = Field( + ..., description="Unique identifier for the invocation." + ) + agent_id: UUID = Field(..., description="Target agent identifier.") + payload: dict[str, object] = Field( + default_factory=dict, description="Invocation payload." + ) + + +__all__ = ["ModelInvocationCommand"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/node.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/node.py new file mode 100644 index 0000000000..8e94edbfb2 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/node.py @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Declarative remote-agent invocation effect node.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from omnibase_core.nodes.node_effect import NodeEffect + +if TYPE_CHECKING: + from omnibase_core.models.container import ModelONEXContainer + + +class NodeRemoteAgentInvokeEffect(NodeEffect): + """Declarative effect node for remote-agent invocation. + + All behavior is defined in contract.yaml. Protocol-specific handlers are + registered by later implementation tickets. + """ + + def __init__(self, container: ModelONEXContainer) -> None: + """Initialize with container dependency injection.""" + super().__init__(container) + + +__all__ = ["NodeRemoteAgentInvokeEffect"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/__init__.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/__init__.py new file mode 100644 index 0000000000..72a2452120 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/__init__.py @@ -0,0 +1,9 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Persistence helpers for node_remote_agent_invoke_effect.""" + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.persistence.remote_task_state_repository import ( + RemoteTaskStateRepository, +) + +__all__ = ["RemoteTaskStateRepository"] diff --git a/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/remote_task_state_repository.py b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/remote_task_state_repository.py new file mode 100644 index 0000000000..67acfa21f0 --- /dev/null +++ b/src/omnibase_infra/nodes/node_remote_agent_invoke_effect/persistence/remote_task_state_repository.py @@ -0,0 +1,291 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""PostgreSQL persistence for remote agent task state.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING, TypeVar, cast +from uuid import UUID, uuid4 + +import asyncpg + +from omnibase_core.models.delegation.model_remote_task_state import ModelRemoteTaskState +from omnibase_infra.enums import EnumInfraTransportType, EnumPostgresErrorCode +from omnibase_infra.errors import ModelInfraErrorContext, RepositoryExecutionError +from omnibase_infra.mixins.mixin_postgres_op_executor import MixinPostgresOpExecutor + +if TYPE_CHECKING: + from collections.abc import Sequence + +T = TypeVar("T") + + +SQL_UPSERT_REMOTE_TASK_STATE = """ +INSERT INTO remote_task_state ( + task_id, + invocation_kind, + protocol, + target_ref, + remote_task_handle, + correlation_id, + status, + last_remote_status, + last_emitted_event_type, + submitted_at, + updated_at, + completed_at, + error +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13 +) +ON CONFLICT (task_id) DO UPDATE SET + invocation_kind = EXCLUDED.invocation_kind, + protocol = EXCLUDED.protocol, + target_ref = EXCLUDED.target_ref, + remote_task_handle = EXCLUDED.remote_task_handle, + correlation_id = EXCLUDED.correlation_id, + status = EXCLUDED.status, + last_remote_status = EXCLUDED.last_remote_status, + last_emitted_event_type = EXCLUDED.last_emitted_event_type, + submitted_at = EXCLUDED.submitted_at, + updated_at = EXCLUDED.updated_at, + completed_at = EXCLUDED.completed_at, + error = EXCLUDED.error +RETURNING task_id +""" + +SQL_GET_REMOTE_TASK_STATE = """ +SELECT + task_id, + invocation_kind, + protocol, + target_ref, + remote_task_handle, + correlation_id, + status, + last_remote_status, + last_emitted_event_type, + submitted_at, + updated_at, + completed_at, + error +FROM remote_task_state +WHERE task_id = $1 +""" + +SQL_GET_REMOTE_TASK_STATE_BY_HANDLE = """ +SELECT + task_id, + invocation_kind, + protocol, + target_ref, + remote_task_handle, + correlation_id, + status, + last_remote_status, + last_emitted_event_type, + submitted_at, + updated_at, + completed_at, + error +FROM remote_task_state +WHERE remote_task_handle = $1 +""" + +SQL_LOAD_UNFINISHED_REMOTE_TASK_STATES = """ +SELECT + task_id, + invocation_kind, + protocol, + target_ref, + remote_task_handle, + correlation_id, + status, + last_remote_status, + last_emitted_event_type, + submitted_at, + updated_at, + completed_at, + error +FROM remote_task_state +WHERE status NOT IN ('COMPLETED', 'FAILED', 'TIMED_OUT', 'CANCELED') +ORDER BY submitted_at ASC +""" + + +class RemoteTaskStateRepository(MixinPostgresOpExecutor): + """Persistence adapter for restart-safe remote task resumption.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + + async def upsert( + self, + state: ModelRemoteTaskState, + *, + correlation_id: UUID | None = None, + ) -> None: + op_correlation_id = correlation_id or state.correlation_id + + async def _op() -> None: + async with self._pool.acquire() as conn: + row = await conn.fetchrow( + SQL_UPSERT_REMOTE_TASK_STATE, + state.task_id, + state.invocation_kind.value, + state.protocol.value if state.protocol is not None else None, + state.target_ref, + state.remote_task_handle, + state.correlation_id, + state.status.value, + state.last_remote_status, + ( + state.last_emitted_event_type.value + if state.last_emitted_event_type is not None + else None + ), + state.submitted_at, + state.updated_at, + state.completed_at, + state.error, + ) + if row is None: + raise RepositoryExecutionError( + "remote_task_state upsert returned no row", + op_name="upsert_remote_task_state", + table="remote_task_state", + sql_fingerprint="INSERT INTO remote_task_state ... ON CONFLICT (task_id) DO UPDATE", + context=self._context( + op_correlation_id, "upsert_remote_task_state" + ), + ) + + await self._run_checked( + correlation_id=op_correlation_id, + op_name="upsert_remote_task_state", + log_context={"task_id": str(state.task_id)}, + fn=_op, + ) + + async def get( + self, + task_id: UUID, + *, + correlation_id: UUID | None = None, + ) -> ModelRemoteTaskState | None: + row = await self._run_checked( + correlation_id=correlation_id or uuid4(), + op_name="get_remote_task_state", + log_context={"task_id": str(task_id)}, + fn=lambda: self._fetchrow(SQL_GET_REMOTE_TASK_STATE, task_id), + ) + if row is None: + return None + return self._row_to_model(row) + + async def get_by_remote_task_handle( + self, + remote_task_handle: str, + *, + correlation_id: UUID | None = None, + ) -> ModelRemoteTaskState | None: + row = await self._run_checked( + correlation_id=correlation_id or uuid4(), + op_name="get_remote_task_state_by_handle", + log_context={"remote_task_handle": remote_task_handle}, + fn=lambda: self._fetchrow( + SQL_GET_REMOTE_TASK_STATE_BY_HANDLE, + remote_task_handle, + ), + ) + if row is None: + return None + return self._row_to_model(row) + + async def load_unfinished( + self, + *, + correlation_id: UUID | None = None, + ) -> list[ModelRemoteTaskState]: + rows = await self._run_checked( + correlation_id=correlation_id or uuid4(), + op_name="load_unfinished_remote_task_states", + log_context={"table": "remote_task_state"}, + fn=lambda: self._fetch(SQL_LOAD_UNFINISHED_REMOTE_TASK_STATES), + ) + return [self._row_to_model(row) for row in rows] + + async def _fetchrow( + self, + sql: str, + *params: object, + ) -> asyncpg.Record | None: + async with self._pool.acquire() as conn: + return await conn.fetchrow(sql, *params) + + async def _fetch( + self, + sql: str, + *params: object, + ) -> Sequence[asyncpg.Record]: + async with self._pool.acquire() as conn: + return cast("Sequence[asyncpg.Record]", await conn.fetch(sql, *params)) + + async def _run_checked( + self, + *, + correlation_id: UUID, + op_name: str, + log_context: dict[str, object], + fn: Callable[[], Awaitable[T]], + ) -> T: + result_holder: dict[str, T] = {} + + async def _wrapped() -> None: + result_holder["value"] = await fn() + + backend_result = await self._execute_postgres_op( + op_error_code=EnumPostgresErrorCode.UPSERT_ERROR, + correlation_id=correlation_id, + log_context=log_context, + fn=_wrapped, + ) + if backend_result.success: + return result_holder["value"] + + retriable: bool | None = None + if backend_result.error_code is not None: + try: + retriable = EnumPostgresErrorCode( + backend_result.error_code + ).is_retriable + except ValueError: + retriable = None + + raise RepositoryExecutionError( + backend_result.error or f"{op_name} failed", + op_name=op_name, + table="remote_task_state", + retriable=retriable, + context=self._context(correlation_id, op_name), + ) + + def _context( + self, + correlation_id: UUID, + operation: str, + ) -> ModelInfraErrorContext: + return ModelInfraErrorContext( + transport_type=EnumInfraTransportType.DATABASE, + operation=operation, + correlation_id=correlation_id, + target_name="remote_task_state", + ) + + @staticmethod + def _row_to_model(row: asyncpg.Record) -> ModelRemoteTaskState: + return ModelRemoteTaskState.model_validate(dict(row)) + + +__all__ = ["RemoteTaskStateRepository"] diff --git a/src/omnibase_infra/runtime/version_compatibility.py b/src/omnibase_infra/runtime/version_compatibility.py index 921e7fedf7..7703b25732 100644 --- a/src/omnibase_infra/runtime/version_compatibility.py +++ b/src/omnibase_infra/runtime/version_compatibility.py @@ -175,8 +175,8 @@ def _build_matrix_from_pyproject( _FALLBACK_MATRIX: list[VersionConstraint] = [ VersionConstraint( package="omnibase_core", - min_version="0.39.0", - max_version="0.40.0", + min_version="0.40.0", + max_version="0.41.0", ), VersionConstraint( package="omnibase_spi", diff --git a/tests/integration/db/test_remote_task_state_repository.py b/tests/integration/db/test_remote_task_state_repository.py new file mode 100644 index 0000000000..797efa23b9 --- /dev/null +++ b/tests/integration/db/test_remote_task_state_repository.py @@ -0,0 +1,114 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Integration tests for RemoteTaskStateRepository.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest + +from omnibase_core.enums.enum_agent_protocol import EnumAgentProtocol +from omnibase_core.enums.enum_agent_task_lifecycle_type import ( + EnumAgentTaskLifecycleType, +) +from omnibase_core.enums.enum_invocation_kind import EnumInvocationKind +from omnibase_core.models.delegation.model_remote_task_state import ModelRemoteTaskState +from omnibase_infra.nodes.node_remote_agent_invoke_effect.persistence import ( + RemoteTaskStateRepository, +) +from tests.helpers.util_postgres import PostgresConfig + +pytestmark = [pytest.mark.integration, pytest.mark.postgres, pytest.mark.asyncio] + +MIGRATION_SQL = ( + Path(__file__).resolve().parents[3] + / "docker" + / "migrations" + / "forward" + / "069_create_remote_task_state.sql" +).read_text() + + +@pytest.fixture +async def postgres_pool(): + config = PostgresConfig.from_env() + if not config.is_configured: + pytest.skip( + "PostgreSQL not configured (set OMNIBASE_INFRA_DB_URL or POSTGRES_HOST/POSTGRES_PASSWORD)" + ) + + import asyncpg + + pool = await asyncpg.create_pool( + config.build_dsn(), min_size=1, max_size=5, timeout=10.0 + ) + try: + async with pool.acquire() as conn: + await conn.execute(MIGRATION_SQL) + await conn.execute("TRUNCATE TABLE remote_task_state") + yield pool + finally: + await pool.close() + + +def _row(*, status: EnumAgentTaskLifecycleType) -> ModelRemoteTaskState: + now = datetime.now(UTC) + completed_at = ( + now + if status + in { + EnumAgentTaskLifecycleType.COMPLETED, + EnumAgentTaskLifecycleType.FAILED, + EnumAgentTaskLifecycleType.TIMED_OUT, + EnumAgentTaskLifecycleType.CANCELED, + } + else None + ) + return ModelRemoteTaskState( + task_id=uuid4(), + invocation_kind=EnumInvocationKind.AGENT, + protocol=EnumAgentProtocol.A2A, + target_ref="agent:local-a2a-smoke", + remote_task_handle=f"remote:{uuid4()}", + correlation_id=uuid4(), + status=status, + last_remote_status=status.value.lower(), + last_emitted_event_type=status, + submitted_at=now, + updated_at=now, + completed_at=completed_at, + error="terminal failure" + if status is EnumAgentTaskLifecycleType.FAILED + else None, + ) + + +async def test_upsert_then_load_unfinished(postgres_pool) -> None: + repo = RemoteTaskStateRepository(postgres_pool) + row = _row(status=EnumAgentTaskLifecycleType.SUBMITTED) + + await repo.upsert(row) + + loaded = await repo.get(row.task_id) + unfinished = await repo.load_unfinished() + + assert loaded == row + assert unfinished == [row] + + +async def test_terminal_status_excluded_from_unfinished(postgres_pool) -> None: + repo = RemoteTaskStateRepository(postgres_pool) + submitted = _row(status=EnumAgentTaskLifecycleType.SUBMITTED) + completed = _row(status=EnumAgentTaskLifecycleType.COMPLETED) + failed = _row(status=EnumAgentTaskLifecycleType.FAILED) + + await repo.upsert(submitted) + await repo.upsert(completed) + await repo.upsert(failed) + + unfinished = await repo.load_unfinished() + + assert unfinished == [submitted] diff --git a/tests/integration/nodes/test_node_remote_agent_invoke_effect_integration.py b/tests/integration/nodes/test_node_remote_agent_invoke_effect_integration.py new file mode 100644 index 0000000000..0d7cf2c70b --- /dev/null +++ b/tests/integration/nodes/test_node_remote_agent_invoke_effect_integration.py @@ -0,0 +1,337 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Integration coverage for node_remote_agent_invoke_effect contract wiring.""" + +from __future__ import annotations + +import asyncio +import json +import os +from datetime import UTC, datetime +from pathlib import Path +from uuid import UUID, uuid4 + +import pytest + +from omnibase_core.enums.enum_agent_protocol import EnumAgentProtocol +from omnibase_core.enums.enum_agent_task_lifecycle_type import ( + EnumAgentTaskLifecycleType, +) +from omnibase_core.enums.enum_invocation_kind import EnumInvocationKind +from omnibase_core.models.common.model_schema_value import ModelSchemaValue +from omnibase_core.models.delegation.model_agent_task_lifecycle_event import ( + ModelAgentTaskLifecycleEvent, +) +from omnibase_core.models.delegation.model_invocation_command import ( + ModelInvocationCommand, +) +from omnibase_core.models.delegation.model_remote_task_state import ( + ModelRemoteTaskState, +) +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope +from omnibase_infra.enums import EnumConsumerGroupPurpose +from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory +from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka +from omnibase_infra.event_bus.models import ModelEventMessage +from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig +from omnibase_infra.models import ModelNodeIdentity +from omnibase_infra.runtime.event_bus_subcontract_wiring import ( + EventBusSubcontractWiring, + load_event_bus_subcontract, +) +from omnibase_infra.utils import compute_consumer_group_id +from tests.helpers.util_kafka import KafkaTopicManager, wait_for_consumer_ready + +CONTRACT_PATH = ( + Path("src") + / "omnibase_infra" + / "nodes" + / "node_remote_agent_invoke_effect" + / "contract.yaml" +) +NODE_NAME = "node_remote_agent_invoke_effect" +SERVICE_NAME = "omnibase-infra" +NODE_VERSION = "v0.1.0" + + +class RecordingDispatchEngine: + """Small async dispatch double used to observe real Kafka delivery.""" + + def __init__(self) -> None: + self.received: list[tuple[str, ModelEventEnvelope[object]]] = [] + self.message_received = asyncio.Event() + + async def dispatch( + self, + topic: str, + envelope: ModelEventEnvelope[object], + ) -> None: + self.received.append((topic, envelope)) + self.message_received.set() + + +class LocalRemoteTaskStateStore: + """In-memory state store used by the local P12 run.""" + + def __init__(self) -> None: + self._rows: dict[UUID, ModelRemoteTaskState] = {} + + async def upsert(self, state: ModelRemoteTaskState) -> None: + self._rows[state.task_id] = state + + async def get(self, task_id: UUID) -> ModelRemoteTaskState | None: + return self._rows.get(task_id) + + +class LocalRemoteAgentInvokeDispatchEngine: + """Local dispatch implementation for the P12 in-memory end-to-end run.""" + + def __init__( + self, + event_bus: EventBusInmemory, + state_store: LocalRemoteTaskStateStore, + lifecycle_topic: str, + ) -> None: + self._event_bus = event_bus + self._state_store = state_store + self._lifecycle_topic = lifecycle_topic + + async def dispatch( + self, + topic: str, + envelope: ModelEventEnvelope[object], + ) -> None: + command = ModelInvocationCommand.model_validate(envelope.payload) + now = datetime.now(UTC) + remote_handle = f"local-a2a:{command.task_id}" + + await self._state_store.upsert( + ModelRemoteTaskState( + task_id=command.task_id, + invocation_kind=command.invocation_kind, + protocol=command.agent_protocol, + target_ref=command.target_ref, + remote_task_handle=remote_handle, + correlation_id=command.correlation_id, + status=EnumAgentTaskLifecycleType.SUBMITTED, + last_remote_status="submitted", + last_emitted_event_type=EnumAgentTaskLifecycleType.SUBMITTED, + submitted_at=now, + updated_at=now, + ) + ) + + lifecycle_event = ModelAgentTaskLifecycleEvent( + task_id=command.task_id, + correlation_id=command.correlation_id, + lifecycle_type=EnumAgentTaskLifecycleType.SUBMITTED, + remote_task_handle=remote_handle, + artifact={"source_topic": ModelSchemaValue.from_value(topic)}, + occurred_at=now, + remote_status="submitted", + ) + lifecycle_envelope = ModelEventEnvelope[dict[str, object]]( + correlation_id=command.correlation_id, + payload=lifecycle_event.model_dump(mode="json"), + ) + await self._event_bus.publish( + self._lifecycle_topic, + key=str(command.task_id).encode("utf-8"), + value=lifecycle_envelope.model_dump_json().encode("utf-8"), + ) + + +async def _ensure_contract_topic_exists(topic: str, bootstrap_servers: str) -> None: + """Create the contract topic if absent without deleting canonical topics.""" + manager = KafkaTopicManager(bootstrap_servers) + try: + await manager.create_topic(topic, partitions=1, replication_factor=1) + manager.created_topics.clear() + finally: + await manager.cleanup() + + +@pytest.mark.integration +@pytest.mark.kafka +@pytest.mark.serial +@pytest.mark.asyncio +async def test_remote_agent_invoke_contract_wires_real_kafka_dispatch() -> None: + """Contract-declared remote-agent invoke topic reaches dispatch via Kafka.""" + bootstrap_servers = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "").strip() + if not bootstrap_servers: + pytest.skip("KAFKA_BOOTSTRAP_SERVERS is not configured") + + subcontract = load_event_bus_subcontract(CONTRACT_PATH) + assert subcontract is not None + assert len(subcontract.subscribe_topics) == 1 + invoke_topic = subcontract.subscribe_topics[0] + + await _ensure_contract_topic_exists(invoke_topic, bootstrap_servers) + + config = ModelKafkaEventBusConfig( + bootstrap_servers=bootstrap_servers, + environment="dev", + timeout_seconds=30, + max_retry_attempts=2, + retry_backoff_base=0.5, + circuit_breaker_threshold=5, + circuit_breaker_reset_timeout=10.0, + ) + event_bus = EventBusKafka(config=config) + dispatch_engine = RecordingDispatchEngine() + wiring = EventBusSubcontractWiring( + event_bus=event_bus, + dispatch_engine=dispatch_engine, + environment="dev", + node_name=NODE_NAME, + service=SERVICE_NAME, + version=NODE_VERSION, + ) + + try: + await event_bus.start() + await wiring.wire_subscriptions(subcontract, node_name=NODE_NAME) + await wait_for_consumer_ready( + event_bus, + invoke_topic, + max_wait=10.0, + ) + + health = await event_bus.health_check() + assert health["subscriber_count"] == 1 + assert health["consumer_count"] == 1 + + expected_group = compute_consumer_group_id( + ModelNodeIdentity( + env="dev", + service=SERVICE_NAME, + node_name=NODE_NAME, + version=NODE_VERSION, + ), + EnumConsumerGroupPurpose.CONSUME, + ) + assert ( + expected_group + == "dev.omnibase-infra.node_remote_agent_invoke_effect.consume.v0.1.0" + ) + + envelope = ModelEventEnvelope[dict[str, object]]( + correlation_id=uuid4(), + payload={ + "operation": "submit", + "target_ref": "agent:test-remote-agent", + "input": {"prompt": "integration smoke"}, + }, + ) + await event_bus.publish( + invoke_topic, + key=str(envelope.envelope_id).encode("utf-8"), + value=envelope.model_dump_json().encode("utf-8"), + ) + + await asyncio.wait_for(dispatch_engine.message_received.wait(), timeout=10.0) + + assert len(dispatch_engine.received) == 1 + received_topic, received_envelope = dispatch_engine.received[0] + assert received_topic == invoke_topic + assert received_envelope.payload == envelope.payload + assert received_envelope.event_type == "omnibase-infra.remote-agent-invoke" + finally: + await wiring.cleanup() + await event_bus.close() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_remote_agent_invoke_runs_end_to_end_with_inmemory_bus_and_state() -> ( + None +): + """Run P12 locally: command topic -> state store -> lifecycle topic.""" + subcontract = load_event_bus_subcontract(CONTRACT_PATH) + assert subcontract is not None + assert len(subcontract.subscribe_topics) == 1 + assert len(subcontract.publish_topics) == 1 + invoke_topic = subcontract.subscribe_topics[0] + lifecycle_topic = subcontract.publish_topics[0] + + event_bus = EventBusInmemory(environment="local", group="p12-local-run") + state_store = LocalRemoteTaskStateStore() + dispatch_engine = LocalRemoteAgentInvokeDispatchEngine( + event_bus=event_bus, + state_store=state_store, + lifecycle_topic=lifecycle_topic, + ) + wiring = EventBusSubcontractWiring( + event_bus=event_bus, + dispatch_engine=dispatch_engine, + environment="local", + node_name=NODE_NAME, + service=SERVICE_NAME, + version=NODE_VERSION, + ) + lifecycle_messages: list[ModelEventMessage] = [] + lifecycle_received = asyncio.Event() + + async def capture_lifecycle(message: ModelEventMessage) -> None: + lifecycle_messages.append(message) + lifecycle_received.set() + + lifecycle_identity = ModelNodeIdentity( + env="local", + service=SERVICE_NAME, + node_name=f"{NODE_NAME}_test_observer", + version=NODE_VERSION, + ) + + try: + await event_bus.start() + await event_bus.subscribe( + topic=lifecycle_topic, + node_identity=lifecycle_identity, + on_message=capture_lifecycle, + ) + await wiring.wire_subscriptions(subcontract, node_name=NODE_NAME) + + command = ModelInvocationCommand( + task_id=uuid4(), + correlation_id=uuid4(), + invocation_kind=EnumInvocationKind.AGENT, + agent_protocol=EnumAgentProtocol.A2A, + target_ref="agent:local-a2a-smoke", + payload={ + "prompt": ModelSchemaValue.from_value("local in-memory P12 smoke") + }, + ) + command_envelope = ModelEventEnvelope[dict[str, object]]( + correlation_id=command.correlation_id, + payload=command.model_dump(mode="json"), + ) + + await event_bus.publish( + invoke_topic, + key=str(command.task_id).encode("utf-8"), + value=command_envelope.model_dump_json().encode("utf-8"), + ) + await asyncio.wait_for(lifecycle_received.wait(), timeout=1.0) + + stored = await state_store.get(command.task_id) + assert stored is not None + assert stored.target_ref == "agent:local-a2a-smoke" + assert stored.status is EnumAgentTaskLifecycleType.SUBMITTED + assert stored.remote_task_handle == f"local-a2a:{command.task_id}" + + assert len(lifecycle_messages) == 1 + raw_lifecycle_envelope = json.loads(lifecycle_messages[0].value.decode("utf-8")) + lifecycle_envelope = ModelEventEnvelope[object].model_validate( + raw_lifecycle_envelope + ) + lifecycle_event = ModelAgentTaskLifecycleEvent.model_validate( + lifecycle_envelope.payload + ) + assert lifecycle_event.task_id == command.task_id + assert lifecycle_event.correlation_id == command.correlation_id + assert lifecycle_event.lifecycle_type is EnumAgentTaskLifecycleType.SUBMITTED + assert lifecycle_event.remote_status == "submitted" + finally: + await wiring.cleanup() + await event_bus.close() diff --git a/tests/integration/test_remote_agent_invoke_effect_contract.py b/tests/integration/test_remote_agent_invoke_effect_contract.py new file mode 100644 index 0000000000..1be7d02ff7 --- /dev/null +++ b/tests/integration/test_remote_agent_invoke_effect_contract.py @@ -0,0 +1,65 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Integration tests for node_remote_agent_invoke_effect contract and models (OMN-9632). + +Validates that the scaffold models are importable Pydantic BaseModel subclasses +and that the contract declares the correct topics and module references. +""" + +from __future__ import annotations + +from pathlib import Path +from uuid import uuid4 + +import pytest +import yaml +from pydantic import BaseModel + +from omnibase_infra.nodes.node_remote_agent_invoke_effect.models import ( + EnumAgentTaskStatus, + ModelAgentTaskLifecycleEvent, + ModelInvocationCommand, +) + +CONTRACT_PATH = ( + Path("src") + / "omnibase_infra" + / "nodes" + / "node_remote_agent_invoke_effect" + / "contract.yaml" +) + +EXPECTED_MODULE = "omnibase_infra.nodes.node_remote_agent_invoke_effect.models" + + +@pytest.mark.integration +class TestRemoteAgentInvokeEffectContract: + def test_invocation_command_is_pydantic_model(self) -> None: + assert issubclass(ModelInvocationCommand, BaseModel) + + def test_agent_task_lifecycle_event_is_pydantic_model(self) -> None: + assert issubclass(ModelAgentTaskLifecycleEvent, BaseModel) + + def test_invocation_command_instantiates(self) -> None: + agent_id = uuid4() + cmd = ModelInvocationCommand( + correlation_id=uuid4(), + agent_id=agent_id, + payload={"task": "ping"}, + ) + assert cmd.agent_id == agent_id + + def test_agent_task_lifecycle_event_instantiates(self) -> None: + agent_id = uuid4() + evt = ModelAgentTaskLifecycleEvent( + correlation_id=uuid4(), + status=EnumAgentTaskStatus.SUBMITTED, + agent_id=agent_id, + ) + assert evt.status == EnumAgentTaskStatus.SUBMITTED + + def test_contract_model_modules_reference_local_package(self) -> None: + with CONTRACT_PATH.open() as f: + data = yaml.safe_load(f) + assert data["input_model"]["module"] == EXPECTED_MODULE + assert data["output_model"]["module"] == EXPECTED_MODULE diff --git a/tests/unit/contracts/test_protocol_ownership.py b/tests/unit/contracts/test_protocol_ownership.py index 8d085cd7f1..dedc9cd1b7 100644 --- a/tests/unit/contracts/test_protocol_ownership.py +++ b/tests/unit/contracts/test_protocol_ownership.py @@ -139,6 +139,9 @@ "ProtocolChainEmbeddingClient": "nodes/node_chain_orchestrator/models/protocol_chain_embedding_client.py", # [NODE] DI boundary for chain retrieval/store effects — vector store client interface (chain-learning) "ProtocolChainVectorClient": "nodes/node_chain_orchestrator/models/protocol_chain_vector_client.py", + # [NODE] DI boundary for A2A transport — narrows the SDK surface so the handler + # can be unit-tested without driving the real a2a-sdk client (OMN-9634). + "ProtocolA2ATransport": "nodes/node_remote_agent_invoke_effect/handlers/handler_a2a_task.py", } # Duplicate protocol names that appear in multiple files (node-internal diff --git a/tests/unit/nodes/node_remote_agent_invoke_effect/__init__.py b/tests/unit/nodes/node_remote_agent_invoke_effect/__init__.py new file mode 100644 index 0000000000..327f1172ed --- /dev/null +++ b/tests/unit/nodes/node_remote_agent_invoke_effect/__init__.py @@ -0,0 +1,2 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT diff --git a/tests/unit/nodes/node_remote_agent_invoke_effect/test_handler_a2a_task.py b/tests/unit/nodes/node_remote_agent_invoke_effect/test_handler_a2a_task.py new file mode 100644 index 0000000000..204fca54f4 --- /dev/null +++ b/tests/unit/nodes/node_remote_agent_invoke_effect/test_handler_a2a_task.py @@ -0,0 +1,444 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for HandlerA2ATask.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest + +from omnibase_core.enums.enum_agent_protocol import EnumAgentProtocol +from omnibase_core.enums.enum_agent_task_lifecycle_type import ( + EnumAgentTaskLifecycleType, +) +from omnibase_core.enums.enum_invocation_kind import EnumInvocationKind +from omnibase_core.models.common.model_schema_value import ModelSchemaValue +from omnibase_core.models.delegation.model_a2a_task_request import ModelA2ATaskRequest +from omnibase_core.models.delegation.model_a2a_task_response import ModelA2ATaskResponse +from omnibase_core.models.delegation.model_invocation_command import ( + ModelInvocationCommand, +) +from omnibase_core.models.delegation.model_remote_task_state import ModelRemoteTaskState +from omnibase_core.models.delegation.model_target_agent import ModelTargetAgent +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope +from omnibase_infra.errors import InfraConnectionError +from omnibase_infra.nodes.node_remote_agent_invoke_effect.handlers.handler_a2a_task import ( + HandlerA2ATask, + ProtocolA2ATransport, + map_remote_status, +) + + +class RecordingRepository: + def __init__(self) -> None: + self.rows: list[ModelRemoteTaskState] = [] + + async def upsert( + self, + state: ModelRemoteTaskState, + *, + correlation_id=None, + ) -> None: + del correlation_id + self.rows.append(state) + + async def get_by_remote_task_handle( + self, + remote_task_handle: str, + *, + correlation_id=None, + ) -> ModelRemoteTaskState | None: + del correlation_id + for row in reversed(self.rows): + if row.remote_task_handle == remote_task_handle: + return row + return None + + +class RecordingEventBus: + def __init__(self) -> None: + self.messages: list[tuple[str, bytes | None, bytes]] = [] + + async def publish( + self, + topic: str, + key: bytes | None, + value: bytes, + ) -> None: + self.messages.append((topic, key, value)) + + +class FakeTransport(ProtocolA2ATransport): + def __init__( + self, + *, + submit_response: ModelA2ATaskResponse, + get_responses: list[ModelA2ATaskResponse] | None = None, + ) -> None: + self.submit_response = submit_response + self.get_responses = list(get_responses or []) + self.submit_requests: list[tuple[ModelTargetAgent, ModelA2ATaskRequest]] = [] + self.get_requests: list[tuple[ModelTargetAgent, str]] = [] + + async def submit( + self, + *, + target: ModelTargetAgent, + request_model: ModelA2ATaskRequest, + ) -> ModelA2ATaskResponse: + self.submit_requests.append((target, request_model)) + return self.submit_response + + async def get_task( + self, + *, + target: ModelTargetAgent, + remote_task_handle: str, + ) -> ModelA2ATaskResponse: + self.get_requests.append((target, remote_task_handle)) + if not self.get_responses: + msg = "No queued get_task responses" + raise AssertionError(msg) + return self.get_responses.pop(0) + + +def _command() -> ModelInvocationCommand: + return ModelInvocationCommand( + task_id=uuid4(), + correlation_id=uuid4(), + invocation_kind=EnumInvocationKind.AGENT, + agent_protocol=EnumAgentProtocol.A2A, + target_ref="agent:local-a2a-smoke", + payload={"prompt": ModelSchemaValue.from_value("triage these findings")}, + ) + + +def _target_registry() -> dict[str, ModelTargetAgent]: + return { + "agent:local-a2a-smoke": ModelTargetAgent( + target_ref="agent:local-a2a-smoke", + protocol=EnumAgentProtocol.A2A, + base_url="http://127.0.0.1:8011/a2a/app", + ) + } + + +@pytest.mark.unit +@pytest.mark.parametrize( + ("raw_status", "expected"), + [ + ("submitted", EnumAgentTaskLifecycleType.SUBMITTED), + ("accepted", EnumAgentTaskLifecycleType.ACCEPTED), + ("working", EnumAgentTaskLifecycleType.PROGRESS), + ("in_progress", EnumAgentTaskLifecycleType.PROGRESS), + ("input-required", EnumAgentTaskLifecycleType.PROGRESS), + ("completed", EnumAgentTaskLifecycleType.COMPLETED), + ("failed", EnumAgentTaskLifecycleType.FAILED), + ("timed_out", EnumAgentTaskLifecycleType.TIMED_OUT), + ("canceled", EnumAgentTaskLifecycleType.CANCELED), + ], +) +def test_status_mapping_table( + raw_status: str, + expected: EnumAgentTaskLifecycleType, +) -> None: + assert map_remote_status(raw_status) is expected + + +@pytest.mark.unit +def test_status_mapping_rejects_unknown() -> None: + with pytest.raises(ValueError, match="Unknown remote agent status"): + map_remote_status("mystery") + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_submit_emits_submitted_event_and_persists() -> None: + repo = RecordingRepository() + event_bus = RecordingEventBus() + command = _command() + now = datetime(2026, 4, 25, 17, 15, tzinfo=UTC) + transport = FakeTransport( + submit_response=ModelA2ATaskResponse( + remote_task_handle="remote-123", + status=EnumAgentTaskLifecycleType.SUBMITTED, + artifacts=[], + error=None, + ) + ) + + handler = HandlerA2ATask( + repository=repo, # type: ignore[arg-type] + event_bus=event_bus, + target_registry=_target_registry(), + lifecycle_topic="onex.evt.omnibase-infra.agent-task-lifecycle.v1", + clock=lambda: now, + transport=transport, + ) + + response = await handler.submit(command) + + assert response.remote_task_handle == "remote-123" + assert len(repo.rows) == 2 + assert repo.rows[0].remote_task_handle is None + assert repo.rows[1].remote_task_handle == "remote-123" + assert len(event_bus.messages) == 1 + assert len(transport.submit_requests) == 1 + _target, request_model = transport.submit_requests[0] + assert request_model.input["prompt"].to_value() == "triage these findings" + + topic, key, value = event_bus.messages[0] + assert topic == "onex.evt.omnibase-infra.agent-task-lifecycle.v1" + assert key == str(command.task_id).encode("utf-8") + + envelope = ModelEventEnvelope[object].model_validate(json.loads(value)) + payload = envelope.payload + assert isinstance(payload, dict) + assert payload["lifecycle_type"] == "SUBMITTED" + assert payload["remote_task_handle"] is None + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_watch_emits_transitions_until_completed() -> None: + repo = RecordingRepository() + event_bus = RecordingEventBus() + command = _command() + now = datetime(2026, 4, 25, 17, 20, tzinfo=UTC) + transport = FakeTransport( + submit_response=ModelA2ATaskResponse( + remote_task_handle="remote-123", + status=EnumAgentTaskLifecycleType.SUBMITTED, + artifacts=[], + error=None, + ), + get_responses=[ + ModelA2ATaskResponse( + remote_task_handle="remote-123", + status=EnumAgentTaskLifecycleType.PROGRESS, + artifacts=[], + error=None, + ), + ModelA2ATaskResponse( + remote_task_handle="remote-123", + status=EnumAgentTaskLifecycleType.COMPLETED, + artifacts=[ + { + "report": ModelSchemaValue.from_value( + {"summary": "done", "priority": "high"} + ) + } + ], + error=None, + ), + ], + ) + + handler = HandlerA2ATask( + repository=repo, # type: ignore[arg-type] + event_bus=event_bus, + target_registry=_target_registry(), + lifecycle_topic="onex.evt.omnibase-infra.agent-task-lifecycle.v1", + clock=lambda: now, + poll_interval_seconds=0.0, + transport=transport, + ) + await handler.submit(command) + events = await handler.watch( + remote_task_handle="remote-123", + correlation_id=command.correlation_id, + ) + + assert [event.lifecycle_type for event in events] == [ + EnumAgentTaskLifecycleType.PROGRESS, + EnumAgentTaskLifecycleType.ARTIFACT, + EnumAgentTaskLifecycleType.COMPLETED, + ] + assert events[1].artifact is not None + assert events[1].artifact["report"].to_value() == { + "summary": "done", + "priority": "high", + } + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_watch_dedups_same_status_repeats() -> None: + repo = RecordingRepository() + event_bus = RecordingEventBus() + command = _command() + now = datetime(2026, 4, 25, 17, 25, tzinfo=UTC) + transport = FakeTransport( + submit_response=ModelA2ATaskResponse( + remote_task_handle="remote-456", + status=EnumAgentTaskLifecycleType.SUBMITTED, + artifacts=[], + error=None, + ), + get_responses=[ + ModelA2ATaskResponse( + remote_task_handle="remote-456", + status=EnumAgentTaskLifecycleType.PROGRESS, + artifacts=[], + error=None, + ), + ModelA2ATaskResponse( + remote_task_handle="remote-456", + status=EnumAgentTaskLifecycleType.PROGRESS, + artifacts=[], + error=None, + ), + ModelA2ATaskResponse( + remote_task_handle="remote-456", + status=EnumAgentTaskLifecycleType.COMPLETED, + artifacts=[], + error=None, + ), + ], + ) + + handler = HandlerA2ATask( + repository=repo, # type: ignore[arg-type] + event_bus=event_bus, + target_registry=_target_registry(), + lifecycle_topic="onex.evt.omnibase-infra.agent-task-lifecycle.v1", + clock=lambda: now, + poll_interval_seconds=0.0, + transport=transport, + ) + updated_command = command.model_copy( + update={"task_id": uuid4(), "correlation_id": uuid4()} + ) + await handler.submit(updated_command) + events = await handler.watch( + remote_task_handle="remote-456", + correlation_id=updated_command.correlation_id, + ) + + assert [event.lifecycle_type for event in events] == [ + EnumAgentTaskLifecycleType.PROGRESS, + EnumAgentTaskLifecycleType.COMPLETED, + ] + + +class FailingSubmitTransport(ProtocolA2ATransport): + async def submit( + self, + *, + target: ModelTargetAgent, + request_model: ModelA2ATaskRequest, + ) -> ModelA2ATaskResponse: + del target, request_model + msg = "boom" + raise RuntimeError(msg) + + async def get_task( + self, + *, + target: ModelTargetAgent, + remote_task_handle: str, + ) -> ModelA2ATaskResponse: + del target, remote_task_handle + msg = "unused" + raise AssertionError(msg) + + +class FailingWatchTransport(ProtocolA2ATransport): + def __init__(self, *, submit_response: ModelA2ATaskResponse) -> None: + self.submit_response = submit_response + + async def submit( + self, + *, + target: ModelTargetAgent, + request_model: ModelA2ATaskRequest, + ) -> ModelA2ATaskResponse: + del target, request_model + return self.submit_response + + async def get_task( + self, + *, + target: ModelTargetAgent, + remote_task_handle: str, + ) -> ModelA2ATaskResponse: + del target, remote_task_handle + msg = "watch boom" + raise RuntimeError(msg) + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_submit_failure_marks_state_failed_and_wraps_error_context() -> None: + repo = RecordingRepository() + event_bus = RecordingEventBus() + command = _command() + now = datetime(2026, 4, 26, 12, 0, tzinfo=UTC) + handler = HandlerA2ATask( + repository=repo, # type: ignore[arg-type] + event_bus=event_bus, + target_registry=_target_registry(), + lifecycle_topic="onex.evt.omnibase-infra.agent-task-lifecycle.v1", + clock=lambda: now, + transport=FailingSubmitTransport(), + ) + + with pytest.raises(InfraConnectionError) as exc_info: + await handler.submit(command) + + assert isinstance(exc_info.value.__cause__, RuntimeError) + assert exc_info.value.correlation_id == command.correlation_id + error_context = exc_info.value.model.context + assert error_context.get("operation") == "a2a_submit" + transport_type = error_context.get("transport_type") + assert transport_type is not None + assert getattr(transport_type, "value", transport_type) == "http" + + assert len(repo.rows) == 2 + assert repo.rows[0].status is EnumAgentTaskLifecycleType.SUBMITTED + assert repo.rows[1].status is EnumAgentTaskLifecycleType.FAILED + assert repo.rows[1].error == "boom" + assert repo.rows[1].completed_at == now + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_watch_failure_wraps_error_context() -> None: + repo = RecordingRepository() + event_bus = RecordingEventBus() + command = _command() + now = datetime(2026, 4, 26, 12, 5, tzinfo=UTC) + handler = HandlerA2ATask( + repository=repo, # type: ignore[arg-type] + event_bus=event_bus, + target_registry=_target_registry(), + lifecycle_topic="onex.evt.omnibase-infra.agent-task-lifecycle.v1", + clock=lambda: now, + poll_interval_seconds=0.0, + transport=FailingWatchTransport( + submit_response=ModelA2ATaskResponse( + remote_task_handle="remote-watch-fail", + status=EnumAgentTaskLifecycleType.SUBMITTED, + artifacts=[], + error=None, + ) + ), + ) + await handler.submit(command) + + with pytest.raises(InfraConnectionError) as exc_info: + await handler.watch( + remote_task_handle="remote-watch-fail", + correlation_id=command.correlation_id, + ) + + assert isinstance(exc_info.value.__cause__, RuntimeError) + assert exc_info.value.correlation_id == command.correlation_id + error_context = exc_info.value.model.context + assert error_context.get("operation") == "a2a_watch" + transport_type = error_context.get("transport_type") + assert transport_type is not None + assert getattr(transport_type, "value", transport_type) == "http" diff --git a/tests/unit/nodes/node_remote_agent_invoke_effect/test_node_scaffold.py b/tests/unit/nodes/node_remote_agent_invoke_effect/test_node_scaffold.py new file mode 100644 index 0000000000..5e2bf65178 --- /dev/null +++ b/tests/unit/nodes/node_remote_agent_invoke_effect/test_node_scaffold.py @@ -0,0 +1,95 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Scaffold tests for node_remote_agent_invoke_effect.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from omnibase_infra.enums import EnumConsumerGroupPurpose +from omnibase_infra.models import ModelNodeIdentity +from omnibase_infra.nodes.node_remote_agent_invoke_effect import ( + NodeRemoteAgentInvokeEffect, +) +from omnibase_infra.utils import compute_consumer_group_id + +CONTRACT_PATH = ( + Path("src") + / "omnibase_infra" + / "nodes" + / "node_remote_agent_invoke_effect" + / "contract.yaml" +) + +REMOTE_AGENT_INVOKE_TOPIC = "onex.cmd.omnibase-infra.remote-agent-invoke.v1" +AGENT_TASK_LIFECYCLE_TOPIC = "onex.evt.omnibase-infra.agent-task-lifecycle.v1" + + +def _load_contract() -> dict: + with CONTRACT_PATH.open() as f: + data = yaml.safe_load(f) + assert isinstance(data, dict) + return data + + +@pytest.mark.unit +class TestNodeRemoteAgentInvokeEffectScaffold: + """Validate the remote-agent invoke effect scaffold.""" + + def test_node_exports_declarative_effect_shell(self) -> None: + assert NodeRemoteAgentInvokeEffect.__name__ == "NodeRemoteAgentInvokeEffect" + + def test_contract_file_exists(self) -> None: + assert CONTRACT_PATH.exists(), f"Missing contract: {CONTRACT_PATH}" + + def test_contract_identity(self) -> None: + data = _load_contract() + assert data["name"] == "node_remote_agent_invoke_effect" + assert data["contract_name"] == "node_remote_agent_invoke_effect" + assert data["node_name"] == "node_remote_agent_invoke_effect" + assert data["node_type"] == "EFFECT_GENERIC" + + def test_input_topic_is_remote_agent_invoke(self) -> None: + data = _load_contract() + assert data["event_bus"]["subscribe_topics"] == [REMOTE_AGENT_INVOKE_TOPIC] + consumed_topics = {event["topic"] for event in data["consumed_events"]} + assert consumed_topics == {REMOTE_AGENT_INVOKE_TOPIC} + + def test_output_topic_is_agent_task_lifecycle(self) -> None: + data = _load_contract() + assert data["event_bus"]["publish_topics"] == [AGENT_TASK_LIFECYCLE_TOPIC] + published_topics = {event["topic"] for event in data["published_events"]} + assert published_topics == {AGENT_TASK_LIFECYCLE_TOPIC} + + def test_p14_handler_is_wired(self) -> None: + data = _load_contract() + handlers = data["handler_routing"]["handlers"] + assert len(handlers) == 1 + handler_entry = handlers[0] + assert handler_entry["operation"] == "agent.a2a_task" + assert handler_entry["handler"]["name"] == "HandlerA2ATask" + assert handler_entry["handler"]["module"].endswith( + "node_remote_agent_invoke_effect.handlers.handler_a2a_task" + ) + + def test_input_output_models_use_local_module(self) -> None: + data = _load_contract() + expected_module = "omnibase_infra.nodes.node_remote_agent_invoke_effect.models" + assert data["input_model"]["module"] == expected_module + assert data["output_model"]["module"] == expected_module + + def test_consumer_group_id_matches_expected_shape(self) -> None: + identity = ModelNodeIdentity( + env="dev", + service="omnibase-infra", + node_name="node_remote_agent_invoke_effect", + version="v0.1.0", + ) + + assert ( + compute_consumer_group_id(identity, EnumConsumerGroupPurpose.CONSUME) + == "dev.omnibase-infra.node_remote_agent_invoke_effect.consume.v0.1.0" + ) diff --git a/uv.lock b/uv.lock index ec8b87ebf0..32bb3faf6d 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.12" resolution-markers = [ "python_full_version >= '3.14'", @@ -8,7 +8,23 @@ resolution-markers = [ ] [manifest] -overrides = [{ name = "omnibase-core", git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=9d64d5a3944477093824a53ca931fd17a4b6a8e1" }] +overrides = [{ name = "omnibase-core", git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=6a65f885303e29b4ddc36a66a0cb860537f47316" }] + +[[package]] +name = "a2a-sdk" +version = "0.3.26" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core" }, + { name = "httpx" }, + { name = "httpx-sse" }, + { name = "protobuf" }, + { name = "pydantic" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/97/a6840e01795b182ce751ca165430d46459927cde9bfab838087cbb24aef7/a2a_sdk-0.3.26.tar.gz", hash = "sha256:44068e2d037afbb07ab899267439e9bc7eaa7ac2af94f1e8b239933c993ad52d", size = 274598, upload-time = "2026-04-09T15:21:13.902Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/dd/d5/51f4ee1bf3b736add42a542d3c8a3fd3fa85f3d36c17972127defc46c26f/a2a_sdk-0.3.26-py3-none-any.whl", hash = "sha256:754e0573f6d33b225c1d8d51f640efa69cbbed7bdfb06ce9c3540ea9f58d4a91", size = 151016, upload-time = "2026-04-09T15:21:12.35Z" }, +] [[package]] name = "aenum" @@ -1077,6 +1093,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/50/d2/6c99ec3d9e369ddc27adc758a82b6485d28ac797669be3571afa74757cae/geventhttpclient-2.3.7-cp314-cp314t-win_amd64.whl", hash = "sha256:607b7a1c4d03a94ec1a2f4e7891039fde84fcd816f2d921a28c11759427f068f", size = 49914, upload-time = "2025-12-07T19:48:42.276Z" }, ] +[[package]] +name = "google-api-core" +version = "2.30.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-auth" }, + { name = "googleapis-common-protos" }, + { name = "proto-plus" }, + { name = "protobuf" }, + { name = "requests" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/16/ce/502a57fb0ec752026d24df1280b162294b22a0afb98a326084f9a979138b/google_api_core-2.30.3.tar.gz", hash = "sha256:e601a37f148585319b26db36e219df68c5d07b6382cff2d580e83404e44d641b", size = 177001, upload-time = "2026-04-10T00:41:28.035Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/15/e56f351cf6ef1cfea58e6ac226a7318ed1deb2218c4b3cc9bd9e4b786c5a/google_api_core-2.30.3-py3-none-any.whl", hash = "sha256:a85761ba72c444dad5d611c2220633480b2b6be2521eca69cca2dbb3ffd6bfe8", size = 173274, upload-time = "2026-04-09T22:57:16.198Z" }, +] + +[[package]] +name = "google-auth" +version = "2.49.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, + { name = "pyasn1-modules" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c6/fc/e925290a1ad95c975c459e2df070fac2b90954e13a0370ac505dff78cb99/google_auth-2.49.2.tar.gz", hash = "sha256:c1ae38500e73065dcae57355adb6278cf8b5c8e391994ae9cbadbcb9631ab409", size = 333958, upload-time = "2026-04-10T00:41:21.888Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/73/76/d241a5c927433420507215df6cac1b1fa4ac0ba7a794df42a84326c68da8/google_auth-2.49.2-py3-none-any.whl", hash = "sha256:c2720924dfc82dedb962c9f52cabb2ab16714fd0a6a707e40561d217574ed6d5", size = 240638, upload-time = "2026-04-10T00:41:14.501Z" }, +] + [[package]] name = "googleapis-common-protos" version = "1.72.0" @@ -1898,8 +1943,8 @@ wheels = [ [[package]] name = "omnibase-core" -version = "0.39.0" -source = { git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=9d64d5a3944477093824a53ca931fd17a4b6a8e1#9d64d5a3944477093824a53ca931fd17a4b6a8e1" } +version = "0.40.0" +source = { git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=6a65f885303e29b4ddc36a66a0cb860537f47316#6a65f885303e29b4ddc36a66a0cb860537f47316" } dependencies = [ { name = "blake3" }, { name = "click" }, @@ -1918,6 +1963,7 @@ name = "omnibase-infra" version = "0.34.1" source = { editable = "." } dependencies = [ + { name = "a2a-sdk" }, { name = "aiofiles" }, { name = "aiohttp" }, { name = "aiokafka" }, @@ -1990,6 +2036,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "a2a-sdk", specifier = ">=0.3.4,<0.4.0" }, { name = "aiofiles", specifier = ">=23.2.1,<26.0.0" }, { name = "aiohttp", specifier = ">=3.9.0,<4.0.0" }, { name = "aiokafka", specifier = ">=0.11.0,<0.14.0" }, @@ -2007,7 +2054,7 @@ requires-dist = [ { name = "jsonschema", specifier = ">=4.20.0,<5.0.0" }, { name = "mcp", specifier = ">=1.25.0,<2.0.0" }, { name = "neo4j", specifier = ">=5.15.0,<7.0.0" }, - { name = "omnibase-core", git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=9d64d5a3944477093824a53ca931fd17a4b6a8e1" }, + { name = "omnibase-core", git = "https://github.com/OmniNode-ai/omnibase_core.git?rev=6a65f885303e29b4ddc36a66a0cb860537f47316" }, { name = "omnibase-spi", git = "https://github.com/OmniNode-ai/omnibase_spi.git?rev=a9380008fb4a2f2a784d374389260d896455c3b2" }, { name = "omnimarket", git = "https://github.com/OmniNode-ai/omnimarket.git?branch=main" }, { name = "opentelemetry-api", specifier = ">=1.27.0,<2.0.0" }, @@ -2459,6 +2506,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5b/5a/bc7b4a4ef808fa59a816c17b20c4bef6884daebbdf627ff2a161da67da19/propcache-0.4.1-py3-none-any.whl", hash = "sha256:af2a6052aeb6cf17d3e46ee169099044fd8224cbaf75c76a2ef596e8163e2237", size = 13305, upload-time = "2025-10-08T19:49:00.792Z" }, ] +[[package]] +name = "proto-plus" +version = "1.27.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/81/0d/94dfe80193e79d55258345901acd2917523d56e8381bc4dee7fd38e3868a/proto_plus-1.27.2.tar.gz", hash = "sha256:b2adde53adadf75737c44d3dcb0104fde65250dfc83ad59168b4aa3e574b6a24", size = 57204, upload-time = "2026-03-26T22:18:57.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/84/f3/1fba73eeffafc998a25d59703b63f8be4fe8a5cb12eaff7386a0ba0f7125/proto_plus-1.27.2-py3-none-any.whl", hash = "sha256:6432f75893d3b9e70b9c412f1d2f03f65b11fb164b793d14ae2ca01821d22718", size = 50450, upload-time = "2026-03-26T22:13:42.927Z" }, +] + [[package]] name = "protobuf" version = "5.29.6" @@ -2542,6 +2601,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e1/36/9c0c326fe3a4227953dfb29f5d0c8ae3b8eb8c1cd2967aa569f50cb3c61f/psycopg2_binary-2.9.11-cp314-cp314-win_amd64.whl", hash = "sha256:4012c9c954dfaccd28f94e84ab9f94e12df76b4afb22331b1f0d3154893a6316", size = 2803913, upload-time = "2025-10-10T11:13:57.058Z" }, ] +[[package]] +name = "pyasn1" +version = "0.6.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5c/5f/6583902b6f79b399c9c40674ac384fd9cd77805f9e6205075f828ef11fb2/pyasn1-0.6.3.tar.gz", hash = "sha256:697a8ecd6d98891189184ca1fa05d1bb00e2f84b5977c481452050549c8a72cf", size = 148685, upload-time = "2026-03-17T01:06:53.382Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5d/a0/7d793dce3fa811fe047d6ae2431c672364b462850c6235ae306c0efd025f/pyasn1-0.6.3-py3-none-any.whl", hash = "sha256:a80184d120f0864a52a073acc6fc642847d0be408e7c7252f31390c0f4eadcde", size = 83997, upload-time = "2026-03-17T01:06:52.036Z" }, +] + +[[package]] +name = "pyasn1-modules" +version = "0.4.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyasn1" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e9/e6/78ebbb10a8c8e4b61a59249394a4a594c1a7af95593dc933a349c8d00964/pyasn1_modules-0.4.2.tar.gz", hash = "sha256:677091de870a80aae844b1ca6134f54652fa2c8c5a52aa396440ac3106e941e6", size = 307892, upload-time = "2025-03-28T02:41:22.17Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/8d/d529b5d697919ba8c11ad626e835d4039be708a35b0d22de83a269a6682c/pyasn1_modules-0.4.2-py3-none-any.whl", hash = "sha256:29253a9207ce32b64c3ac6600edc75368f98473906e8fd1043bd6b5b1de2c14a", size = 181259, upload-time = "2025-03-28T02:41:19.028Z" }, +] + [[package]] name = "pycparser" version = "3.0"