diff --git a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/dynamic_workflow_agent.py b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/dynamic_workflow_agent.py index 1a4b541e7..ce3a174f6 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/dynamic_workflow_agent.py +++ b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/dynamic_workflow_agent.py @@ -9,6 +9,7 @@ import logging from typing import AsyncGenerator, ClassVar +from urllib.parse import urlsplit, urlunsplit from uuid import uuid4 from a2a.types import ( AgentCard, @@ -21,6 +22,7 @@ from google.adk.events.event import Event from google.genai import types +from agents.supervisors.recruiter.card import RECRUITER_SUPERVISOR_CARD from agents.supervisors.recruiter.models import ( STATE_KEY_RECRUITED_AGENTS, STATE_KEY_SELECTED_AGENT, @@ -28,14 +30,58 @@ AgentProtocol, AgentRecord, ) -from agents.supervisors.recruiter.recruiter_client import ( - _event_consumer, - _event_interceptor, -) from agents.supervisors.recruiter.shared import a2a_client_factory +from common.a2a_event_middleware import ( + EventEmittingInterceptor, + make_event_emitting_consumer, +) +from common.stable_agent_id import stable_agent_id_for_name +from config.config import DISCOVERED_AGENT_HOST logger = logging.getLogger("lungo.recruiter.supervisor.dynamic_workflow") +# Matches the seeded recruiter node and discovery anchor in recruiter_client. +_RECRUITER_ANCHOR_STABLE_AGENT_ID = stable_agent_id_for_name("Agentic Recruiter agent") + +# Delegation emits anchor nodes + edges only (no transport / duplicate caller). +_event_interceptor = EventEmittingInterceptor( + caller_card=RECRUITER_SUPERVISOR_CARD, + delegation_mode="edge_only", + delegation_anchor_stable_agent_id=_RECRUITER_ANCHOR_STABLE_AGENT_ID, +) +_event_consumer = make_event_emitting_consumer( + caller_card=RECRUITER_SUPERVISOR_CARD, + delegation_mode="edge_only", +) + +# Hosts an agent advertises for its own bind address; these are not routable +# from the supervisor and get rewritten to DISCOVERED_AGENT_HOST. +_UNROUTABLE_ADVERTISED_HOSTS = frozenset({"0.0.0.0", "127.0.0.1", "localhost"}) + + +def _reachable_url(url: str) -> str: + """Rewrite an unroutable advertised host to a reachable one.""" + if not url: + return url + parts = urlsplit(url) + if parts.scheme not in ("http", "https"): + return url + if parts.hostname not in _UNROUTABLE_ADVERTISED_HOSTS: + return url + netloc = f"{DISCOVERED_AGENT_HOST}:{parts.port}" if parts.port else DISCOVERED_AGENT_HOST + return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) + + +def _reachable_card(card: AgentCard) -> AgentCard: + """Return a card copy with HTTP interface URLs rewritten for delegation.""" + reachable = card.model_copy(deep=True) + reachable.url = _reachable_url(reachable.url) + if reachable.additional_interfaces: + for interface in reachable.additional_interfaces: + interface.url = _reachable_url(interface.url) + return reachable + + class DynamicWorkflowAgent(BaseAgent): """Executes tasks by sending messages to selected recruited agents via A2A HTTP. @@ -213,11 +259,12 @@ async def _run_async_impl( ) return - agent_card = record.card + agent_card = _reachable_card(record.card) logger.info( - "[agent:dynamic_workflow] Sending to agent: %s", + "[agent:dynamic_workflow] Sending to agent: %s at %s", record.name, + agent_card.url, ) # Send A2A message diff --git a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/recruiter_client.py b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/recruiter_client.py index 73a3b4408..574b84e0b 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/recruiter_client.py +++ b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/recruiter_client.py @@ -7,9 +7,10 @@ import asyncio import json import logging -from uuid import uuid4 +from uuid import NAMESPACE_DNS, UUID, uuid4, uuid5 import httpx +from pydantic import ValidationError from a2a.client import ClientConfig, ClientFactory from a2a.types import ( AgentCard, @@ -24,7 +25,6 @@ ) from google.adk.tools.tool_context import ToolContext -from agents.supervisors.recruiter.card import RECRUITER_SUPERVISOR_CARD from agents.supervisors.recruiter.models import ( STATE_KEY_EVALUATION_RESULTS, STATE_KEY_RECRUITED_AGENTS, @@ -33,17 +33,45 @@ from agents.supervisors.recruiter.recruiter_service_card import ( RECRUITER_AGENT_CARD, ) -from common.a2a_event_middleware import ( - EventEmittingInterceptor, - make_event_emitting_consumer, +from common.stable_agent_id import stable_agent_id_for_name +from common.workflow_context_prop import read_workflow_context +from common.workflow_utils.builders import build_event, make_node +from common.workflow_utils.event_sink import WorkflowAPIEventSink +from common.workflow_utils.inflight import read_trace_context +from common.workflow_utils.workflow_catalog import lookup_workflow +from config.config import EMIT_WORKFLOW_EVENTS +from schema.types import ( + EdgeId, + InstanceId, + NodeId, + Operation, + PartialEdge, + PartialTopology, ) logger = logging.getLogger("lungo.recruiter.supervisor.recruiter_client") - -# A2A middleware singletons that capture outbound and inbound calls. -_event_interceptor = EventEmittingInterceptor(caller_card=RECRUITER_SUPERVISOR_CARD) -_event_consumer = make_event_emitting_consumer(caller_card=RECRUITER_SUPERVISOR_CARD) +# Discovered agents are merged into the live workflow instance topology so the +# frontend renders and disposes of them through the normal topology channel. +# The anchor node carries the seeded recruiter's stable_agent_id so the backend +# merge layer (reconcile_event_node_identities) recognizes it as the same node, +# drops the duplicate, and re-points the discovered-agent edges onto the real +# "Agentic Recruiter" node. The frontend makes no anchoring decision. +# _RECRUITER_ANCHOR_RECORD_NAME must match the seeded recruiter record "name" +# (oasf/agents/recruiter.json) so the derived stable_agent_id matches the seed. +_RECRUITER_ANCHOR_LABEL = "Agentic Recruiter" +_RECRUITER_ANCHOR_RECORD_NAME = "Agentic Recruiter agent" +_DISCOVERY_NS = uuid5(NAMESPACE_DNS, "recruiter.discovery.lungo") +_discovery_event_sink = WorkflowAPIEventSink() if EMIT_WORKFLOW_EVENTS else None + + +def _is_valid_instance_id(value: str) -> bool: + """True when value is a canonical ``instance://`` id (schema-validated).""" + try: + InstanceId(root=value) + except ValidationError: + return False + return True # --------------------------------------------------------------------------- # Side-channel queue for streaming A2A events to main.py @@ -108,6 +136,158 @@ def _extract_parts(parts: list[Part]) -> RecruitmentResponse: ) +def _discovery_node_id(cid: str) -> str: + """Deterministic runtime node id for a discovered agent CID.""" + return f"node://{uuid5(_DISCOVERY_NS, f'agent::{cid}')}" + + +def _discovery_edge_id(source_nid: str, target_nid: str) -> str: + """Deterministic edge id for an anchor -> discovered-agent connection.""" + return f"edge://{uuid5(_DISCOVERY_NS, f'{source_nid}->{target_nid}')}" + + +async def _emit_discovery_topology(agent_records: dict[str, dict]) -> None: + """Merge discovered agents into the live workflow instance topology. + + Emits a STATE_PROGRESS_UPDATE that creates one node per discovered agent + (carrying its inline agent record / flat AgentCard dict on ``oasf_record``) + plus an anchor node carrying the + seeded recruiter's stable_agent_id. The backend merge layer reconciles the + anchor onto the existing "Agentic Recruiter" node and re-points the new + edges there, so nodes arrive, render, and dispose through the normal + topology channel without any frontend anchoring decision. + """ + if _discovery_event_sink is None or not agent_records: + return + + wf_ctx = read_workflow_context() + workflow_name = wf_ctx.workflow_name + instance_id = wf_ctx.instance_id + if not workflow_name or lookup_workflow(workflow_name) is None: + logger.debug( + "recruit discovery: no/unknown workflow_name=%r; skipping topology emit", + workflow_name, + ) + return + if not instance_id or not _is_valid_instance_id(instance_id): + logger.debug( + "recruit discovery: missing/invalid instance_id=%r; skipping topology emit", + instance_id, + ) + return + + trace_ctx = read_trace_context() + anchor_id = f"node://{uuid5(_DISCOVERY_NS, 'recruiter-anchor')}" + nodes: list = [ + make_node( + anchor_id, + operation=Operation.CREATE, + node_type="customNode", + label=_RECRUITER_ANCHOR_LABEL, + layer_index=0, + stable_agent_id=stable_agent_id_for_name(_RECRUITER_ANCHOR_RECORD_NAME), + ) + ] + edges: list = [] + for cid, record in agent_records.items(): + name = str(record.get("name") or "").strip() or cid + node_id = _discovery_node_id(cid) + nodes.append( + make_node( + node_id, + operation=Operation.CREATE, + node_type="customNode", + label=name, + layer_index=1, + stable_agent_id=stable_agent_id_for_name(name), + extras={"oasf_record": record, "agent_cid": cid}, + ) + ) + edges.append( + PartialEdge( + id=EdgeId(_discovery_edge_id(anchor_id, node_id)), + operation=Operation.CREATE, + type="custom", + source=NodeId(anchor_id), + target=NodeId(node_id), + bidirectional=False, + weight=1.0, + ) + ) + + topology = PartialTopology(nodes=nodes, edges=edges) + correlation_id = ( + f"correlation://{UUID(int=trace_ctx.trace_id)}" + if trace_ctx.trace_id is not None + else f"correlation://{uuid4()}" + ) + try: + event = build_event( + source="recruiter_supervisor", + workflow_name=workflow_name, + instance_id=instance_id, + topology=topology, + correlation_id=correlation_id, + correlation_message=f"discovered {len(agent_records)} agent(s)", + trace_id=trace_ctx.trace_id, + span_id=trace_ctx.span_id, + ) + except RuntimeError: + logger.warning( + "recruit discovery: build_event failed for workflow=%r", workflow_name + ) + return + + await _discovery_event_sink.emit(event) + logger.info( + "recruit discovery: emitted %d discovered node(s) to instance %s", + len(agent_records), + instance_id, + ) + + +async def _forward_status_update( + update: TaskStatusUpdateEvent, tool_name: str +) -> None: + """Forward one intermediate A2A status update onto the side-channel queue. + + Shared by recruit_agents and evaluate_agent: each non-final text status part + is republished as a ``status_update`` event for main.py's stream_generator + to drain in parallel. + """ + if update.final or not update.status or not update.status.message: + return + message = update.status.message + for part in message.parts or []: + root = part.root + if not isinstance(root, TextPart) or not root.text: + continue + status_text = root.text + author = None + event_type = "a2a_status" + if message.metadata: + author = message.metadata.get("author") or message.metadata.get( + "from_agent" + ) + event_type = message.metadata.get("event_type", "a2a_status") + logger.info( + "[tool:%s] A2A status: %s (state=%s, final=%s)", + tool_name, + status_text[:120], + update.status.state if update.status else "?", + update.final, + ) + await _a2a_event_queue.put( + { + "event_type": "status_update", + "message": status_text, + "state": "working", + "author": author or "recruiter_service", + "a2a_event_type": event_type, + } + ) + + async def recruit_agents(query: str, tool_context: ToolContext) -> str: """Search the AGNTCY directory for agents matching a task description. @@ -127,11 +307,11 @@ async def recruit_agents(query: str, tool_context: ToolContext) -> str: async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as httpx_client: config = ClientConfig(httpx_client=httpx_client, streaming=True) factory = ClientFactory(config) - client = factory.create( - RECRUITER_AGENT_CARD, - interceptors=[_event_interceptor], - consumers=[_event_consumer], - ) + # No A2A topology middleware here: the internal hop to the recruiter + # service is an implementation detail. Discovery results are surfaced + # via _emit_discovery_topology so only the seeded recruiter node and + # the discovered agents appear on the graph. + client = factory.create(RECRUITER_AGENT_CARD) message = Message( role=Role.user, @@ -147,37 +327,8 @@ async def recruit_agents(query: str, tool_context: ToolContext) -> str: task, update = event # Forward intermediate status updates via the side-channel queue - if isinstance(update, TaskStatusUpdateEvent) and not update.final: - if update.status and update.status.message: - parts = update.status.message.parts - if parts: - for p in parts: - if isinstance(p.root, TextPart) and p.root.text: - status_text = p.root.text - # Determine author from metadata - author = None - if update.status.message.metadata: - author = update.status.message.metadata.get("author") - if not author: - author = update.status.message.metadata.get("from_agent") - event_type = "a2a_status" - if update.status.message.metadata: - event_type = update.status.message.metadata.get("event_type", "a2a_status") - - logger.info( - "[tool:recruit_agents] A2A status: %s (state=%s, final=%s)", - status_text[:120], - update.status.state if update.status else "?", - update.final, - ) - # Push to side-channel for main.py to pick up - await _a2a_event_queue.put({ - "event_type": "status_update", - "message": status_text, - "state": "working", - "author": author or "recruiter_service", - "a2a_event_type": event_type, - }) + if isinstance(update, TaskStatusUpdateEvent): + await _forward_status_update(update, "recruit_agents") # Extract final result from completed task if ( @@ -202,6 +353,10 @@ async def recruit_agents(query: str, tool_context: ToolContext) -> str: existing_evals.update(response_data.evaluation_results) tool_context.state[STATE_KEY_EVALUATION_RESULTS] = existing_evals + # Merge discovered agents into the live instance topology so the frontend + # renders/disposes them like any other instance node. + await _emit_discovery_topology(response_data.agent_records) + logger.info( "[tool:recruit_agents] Found %d agent(s), %d evaluation(s)", len(response_data.agent_records), @@ -321,47 +476,16 @@ async def evaluate_agent( async with httpx.AsyncClient(timeout=httpx.Timeout(300.0)) as httpx_client: config = ClientConfig(httpx_client=httpx_client, streaming=True) factory = ClientFactory(config) - client = factory.create( - RECRUITER_AGENT_CARD, - interceptors=[_event_interceptor], - consumers=[_event_consumer], - ) + # See recruit_agents: no A2A topology middleware on the internal hop. + client = factory.create(RECRUITER_AGENT_CARD) async for event in client.send_message(message): if isinstance(event, tuple) and len(event) == 2: task, update = event # Forward intermediate status updates via the side-channel queue - if isinstance(update, TaskStatusUpdateEvent) and not update.final: - if update.status and update.status.message: - for p in update.status.message.parts or []: - if isinstance(p.root, TextPart) and p.root.text: - status_text = p.root.text - author = None - if update.status.message.metadata: - author = ( - update.status.message.metadata.get("author") - or update.status.message.metadata.get("from_agent") - ) - event_type = "a2a_status" - if update.status.message.metadata: - event_type = update.status.message.metadata.get( - "event_type", "a2a_status" - ) - - logger.info( - "[tool:evaluate_agent] A2A status: %s (state=%s, final=%s)", - status_text[:120], - update.status.state if update.status else "?", - update.final, - ) - await _a2a_event_queue.put({ - "event_type": "status_update", - "message": status_text, - "state": "working", - "author": author or "recruiter_service", - "a2a_event_type": event_type, - }) + if isinstance(update, TaskStatusUpdateEvent): + await _forward_status_update(update, "evaluate_agent") # Extract final result from completed task if ( diff --git a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/shared.py b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/shared.py index 358e5970a..24382fb75 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/shared.py +++ b/coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/shared.py @@ -2,8 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 from typing import Optional +import httpx from agntcy_app_sdk.factory import AgntcyFactory -from config.config import OTEL_SDK_DISABLED +from config.config import OTEL_SDK_DISABLED, A2A_CLIENT_TIMEOUT_SECONDS from agntcy_app_sdk.semantic.a2a.client.factory import A2AClientFactory from common.a2a_transport_config import build_a2a_client_config @@ -25,6 +26,11 @@ def get_factory() -> AgntcyFactory: include_nats=True, ) +# Discovered agents are reached over JSONRPC; httpx defaults to a 5s read timeout. +config.httpx_client = httpx.AsyncClient( + timeout=httpx.Timeout(A2A_CLIENT_TIMEOUT_SECONDS, connect=10.0) +) + # -- A2A client factory -- # Holds all transports; callers set preferred_transport on the card # before calling create(). Factory negotiates based on card interfaces. diff --git a/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py b/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py index c5ec72935..3366ea03e 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py +++ b/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py @@ -10,14 +10,14 @@ from __future__ import annotations import logging -import re as _re from time import monotonic -from typing import Any, Mapping +from typing import Any, Literal, Mapping from a2a.client import ClientEvent from a2a.client.middleware import ClientCallContext, ClientCallInterceptor from a2a.types import AgentCard, Message, Task, TaskState from opentelemetry import trace as _otel_trace +from pydantic import ValidationError from common.stable_agent_id import stable_agent_id_for_name as _stable_agent_id from common.workflow_utils.builders import build_event, make_edge, make_node @@ -33,17 +33,23 @@ from common.workflow_utils.workflow_catalog import lookup_workflow from config.config import EMIT_WORKFLOW_EVENTS from schema.types import ( + InstanceId, Operation, PartialTopology, TopologyEdgeItem, TopologyNodeItem, ) -from schema.types.event import _INSTANCE_ID_REGEX as INSTANCE_ID_REGEX - logger = logging.getLogger("lungo.common.event_middleware") -_INSTANCE_ID_PATTERN = _re.compile(INSTANCE_ID_REGEX) + +def _is_valid_instance_id(value: str) -> bool: + """True when value is a canonical ``instance://`` id (schema-validated).""" + try: + InstanceId(root=value) + except ValidationError: + return False + return True def _slugify_source(card: AgentCard) -> str: @@ -175,6 +181,54 @@ async def _outbound_topology( return PartialTopology(nodes=nodes, edges=edges) +async def _delegation_edge_topology( + anchor_stable_agent_id: str, + anchor_label: str, + remote_agent_ids: list[str], + layer_index: int, + *, + allocator: RuntimeIdAllocator, +) -> PartialTopology: + """Build recruiter-style delegation: anchor nodes + edges only (no transport). + + Anchor nodes carry ``stable_agent_id`` values so the merge layer reconciles + them onto existing seeded/discovered nodes instead of creating duplicates. + """ + anchor_id = await allocator.node_id("delegation-recruiter-anchor") + nodes: list[TopologyNodeItem] = [ + make_node( + anchor_id, + operation=Operation.CREATE, + node_type="customNode", + label=anchor_label, + layer_index=layer_index, + stable_agent_id=anchor_stable_agent_id, + ), + ] + edges: list[TopologyEdgeItem] = [] + for agent_id in remote_agent_ids: + remote_anchor_id = await allocator.node_id(f"delegation-remote-{agent_id}") + nodes.append( + make_node( + remote_anchor_id, + operation=Operation.CREATE, + node_type="customNode", + label=agent_id, + layer_index=layer_index + 1, + stable_agent_id=_stable_agent_id(agent_id), + ), + ) + edges.append( + await make_edge( + anchor_id, + remote_anchor_id, + operation=Operation.CREATE, + allocator=allocator, + ), + ) + return PartialTopology(nodes=nodes, edges=edges) + + async def _inbound_topology( caller_agent_id: str, remote_agent_id: str, @@ -242,10 +296,20 @@ def __init__( *, caller_card: AgentCard, agent_call_graph_layer: int = 0, + delegation_mode: Literal["full", "edge_only"] = "full", + delegation_anchor_stable_agent_id: str | None = None, + delegation_anchor_label: str = "Agentic Recruiter", ) -> None: self._caller_agent_id = caller_card.name self._source = _slugify_source(caller_card) self._agent_call_graph_layer = agent_call_graph_layer + self._delegation_mode = delegation_mode + self._delegation_anchor_stable_agent_id = delegation_anchor_stable_agent_id + self._delegation_anchor_label = delegation_anchor_label + + if delegation_mode == "edge_only" and not delegation_anchor_stable_agent_id: + msg = "delegation_anchor_stable_agent_id is required when delegation_mode='edge_only'" + raise ValueError(msg) if not EMIT_WORKFLOW_EVENTS: logger.warning( @@ -312,7 +376,7 @@ async def intercept( ) return request_payload, http_kwargs - if not propagated_instance_id or not _INSTANCE_ID_PATTERN.match(propagated_instance_id): + if not propagated_instance_id or not _is_valid_instance_id(propagated_instance_id): logger.warning( "EventEmittingInterceptor [%s]: missing or malformed propagated " "workflow_instance_id=%r on method '%s'; skipping event emission.", @@ -354,13 +418,25 @@ async def intercept( ) if remote_agent_ids: - topology = await _outbound_topology( - self._caller_agent_id, - remote_agent_ids, - transport_label=active_transport, - layer_index=self._agent_call_graph_layer, - allocator=allocator, - ) + if ( + self._delegation_mode == "edge_only" + and self._delegation_anchor_stable_agent_id is not None + ): + topology = await _delegation_edge_topology( + self._delegation_anchor_stable_agent_id, + self._delegation_anchor_label, + remote_agent_ids, + layer_index=self._agent_call_graph_layer, + allocator=allocator, + ) + else: + topology = await _outbound_topology( + self._caller_agent_id, + remote_agent_ids, + transport_label=active_transport, + layer_index=self._agent_call_graph_layer, + allocator=allocator, + ) correlation_msg = f"outbound {method_name} to [{', '.join(remote_agent_ids)}]" else: caller_node = await allocator.node_id(f"agent-{self._caller_agent_id}") @@ -456,6 +532,7 @@ def make_event_emitting_consumer( *, caller_card: AgentCard, agent_call_graph_layer: int = 0, + delegation_mode: Literal["full", "edge_only"] = "full", ): """Create consumer callback that emits inbound topology updates.""" if not EMIT_WORKFLOW_EVENTS: @@ -476,6 +553,9 @@ async def event_emitting_consumer( ) -> None: """Emit a StateProgressUpdate for an inbound response.""" + if delegation_mode == "edge_only": + return + t_receive_start = monotonic() agent_id = agent_id_from_card(agent_card) or "unknown" diff --git a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/__init__.py b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/__init__.py index 00f02338a..a562333a2 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/__init__.py +++ b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/__init__.py @@ -9,7 +9,12 @@ WorkflowInstanceDataStore, WorkflowInstanceEventFanout, ) -from common.workflow_instance_store.merge import merge_event_data, merge_topology_delta +from common.workflow_instance_store.discovery_layout import enrich_discovery_node_layout +from common.workflow_instance_store.merge import ( + merge_event_data, + merge_topology_delta, + reconcile_event_node_identities, +) from common.workflow_instance_store.notifier import NoOpNotifier, NotifierProtocol from common.workflow_instance_store.errors import WorkflowInstanceStoreClosedError from common.workflow_instance_store.store import WorkflowInstanceStateStore @@ -22,6 +27,8 @@ "WorkflowInstanceEventFanout", "WorkflowInstanceStateStore", "WorkflowInstanceStoreClosedError", + "enrich_discovery_node_layout", "merge_event_data", "merge_topology_delta", + "reconcile_event_node_identities", ] diff --git a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/discovery_layout.py b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/discovery_layout.py new file mode 100644 index 000000000..b6f01a441 --- /dev/null +++ b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/discovery_layout.py @@ -0,0 +1,250 @@ +# Copyright AGNTCY Contributors (https://github.com/agntcy) +# SPDX-License-Identifier: Apache-2.0 + +"""Assign wire ``position`` to runtime-discovered topology nodes near their anchor. + +Runs after :func:`reconcile_event_node_identities` and before +:func:`merge_event_data`. When the anchor node has a defined ``position``, +discovered agents (inline ``oasf_record`` agent dict on create) receive a +non-overlapping slot +using backend-only offset constants. When the anchor lacks ``position``, nodes +are left unchanged so the frontend auto-layout applies. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from schema.types import Data, Event, Operation, Workflow + +_DISCOVERY_LAYOUT_X_OFFSET = 285 +_DISCOVERY_LAYOUT_Y_OFFSET = 185 +_DISCOVERY_LAYOUT_MAX_RING = 20 + + +def _node_id(node: Any) -> str | None: + nid = getattr(node, "id", None) + if nid is None: + return None + root = getattr(nid, "root", None) + if isinstance(root, str): + return root + return nid if isinstance(nid, str) else None + + +def _node_stable_agent_id(node: Any) -> str | None: + sid = getattr(node, "stable_agent_id", None) + if sid is None: + return None + root = getattr(sid, "root", None) + if isinstance(root, str): + return root + return sid if isinstance(sid, str) else None + + +def _node_field(node: Any, key: str) -> Any: + if hasattr(node, "model_dump"): + return node.model_dump(mode="python").get(key) + if isinstance(node, dict): + return node.get(key) + return getattr(node, key, None) + + +def _node_position(node: Any) -> tuple[float, float] | None: + pos = _node_field(node, "position") + if not isinstance(pos, dict): + return None + x = pos.get("x") + y = pos.get("y") + if isinstance(x, (int, float)) and isinstance(y, (int, float)): + return (float(x), float(y)) + return None + + +def _topology_nodes_for_instance( + state_wf: Workflow | None, + instance_id: str, +) -> list: + if state_wf is None: + return [] + inst = state_wf.instances.get(instance_id) + if inst is not None and inst.topology is not None and inst.topology.nodes: + return inst.topology.nodes + if ( + state_wf.starting_topology is not None + and state_wf.starting_topology.nodes + ): + return state_wf.starting_topology.nodes + return [] + + +def _is_discovered_create(node: Any) -> bool: + op = getattr(node, "operation", None) + if op != Operation.CREATE: + return False + if _node_position(node) is not None: + return False + record = _node_field(node, "oasf_record") + return isinstance(record, dict) + + +def _anchor_id_for_target(edges: list | None, target_id: str) -> str | None: + if not edges: + return None + for edge in edges: + target = getattr(edge, "target", None) + source = getattr(edge, "source", None) + if target is None or source is None: + continue + target_root = getattr(target, "root", target) + source_root = getattr(source, "root", source) + if target_root == target_id and isinstance(source_root, str): + return source_root + return None + + +def _slot_is_free( + cx: float, + cy: float, + occupied: list[tuple[float, float]], + x_off: float, + y_off: float, +) -> bool: + for nx, ny in occupied: + if abs(cx - nx) < x_off and abs(cy - ny) < y_off: + return False + return True + + +def _candidate_slots( + anchor_x: float, + anchor_y: float, +) -> Iterator[tuple[float, float]]: + for ring in range(1, _DISCOVERY_LAYOUT_MAX_RING + 1): + x_step = _DISCOVERY_LAYOUT_X_OFFSET * ring + y_step = _DISCOVERY_LAYOUT_Y_OFFSET * ring + yield (anchor_x, anchor_y + y_step) + yield (anchor_x + x_step, anchor_y) + yield (anchor_x - x_step, anchor_y) + yield (anchor_x, anchor_y - y_step) + yield (anchor_x + x_step, anchor_y + y_step) + yield (anchor_x - x_step, anchor_y + y_step) + yield (anchor_x + x_step, anchor_y - y_step) + yield (anchor_x - x_step, anchor_y - y_step) + + +def _find_free_slot( + anchor_x: float, + anchor_y: float, + occupied: list[tuple[float, float]], +) -> tuple[float, float] | None: + for cx, cy in _candidate_slots(anchor_x, anchor_y): + if _slot_is_free( + cx, + cy, + occupied, + _DISCOVERY_LAYOUT_X_OFFSET, + _DISCOVERY_LAYOUT_Y_OFFSET, + ): + return (cx, cy) + return None + + +def _occupied_positions( + batch_occupied: list[tuple[str, float, float]], + *, + exclude_ids: set[str], +) -> list[tuple[float, float]]: + return [ + (x, y) + for node_id, x, y in batch_occupied + if node_id not in exclude_ids + ] + + +def enrich_discovery_node_layout(state: Data, event: Event) -> Event: + """Assign ``position`` to discovered create nodes when the anchor has one. + + Must run after :func:`reconcile_event_node_identities`. Returns a new + :class:`Event`; the input is never mutated. + """ + result = event.model_copy(deep=True) + for wf_name, wf in result.data.workflows.items(): + state_wf = state.workflows.get(wf_name) + for instance_id, inst in wf.instances.items(): + topology = inst.topology + if topology is None or not topology.nodes: + continue + + state_nodes = _topology_nodes_for_instance(state_wf, instance_id) + node_by_id = { + nid: node for node in state_nodes if (nid := _node_id(node)) + } + + batch_occupied: list[tuple[str, float, float]] = [] + for node in state_nodes: + nid = _node_id(node) + pos = _node_position(node) + if nid is not None and pos is not None: + batch_occupied.append((nid, pos[0], pos[1])) + + target_to_source: dict[str, str] = {} + for edge in topology.edges or []: + target = getattr(edge, "target", None) + source = getattr(edge, "source", None) + if target is None or source is None: + continue + target_root = getattr(target, "root", target) + source_root = getattr(source, "root", source) + if isinstance(target_root, str) and isinstance(source_root, str): + target_to_source[target_root] = source_root + + discovered = [ + node + for node in topology.nodes + if _is_discovered_create(node) + ] + discovered.sort( + key=lambda n: ( + _node_stable_agent_id(n) or "", + _node_id(n) or "", + ) + ) + + updates: dict[str, Any] = {} + for node in discovered: + node_id = _node_id(node) + if node_id is None: + continue + anchor_id = _anchor_id_for_target(topology.edges, node_id) + if anchor_id is None: + continue + anchor_node = node_by_id.get(anchor_id) + if anchor_node is None: + continue + anchor_pos = _node_position(anchor_node) + if anchor_pos is None: + continue + + exclude_ids = {node_id, anchor_id} + occupied = _occupied_positions(batch_occupied, exclude_ids=exclude_ids) + slot = _find_free_slot(anchor_pos[0], anchor_pos[1], occupied) + if slot is None: + continue + + cx, cy = slot + updates[node_id] = node.model_copy( + update={"position": {"x": cx, "y": cy}} + ) + batch_occupied.append((node_id, cx, cy)) + + if not updates: + continue + + topology.nodes = [ + updates[nid] if (nid := _node_id(node)) in updates else node + for node in topology.nodes + ] + + return result diff --git a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/merge.py b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/merge.py index 8c3f88f34..024f26ab2 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/merge.py +++ b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/merge.py @@ -41,7 +41,7 @@ import copy from typing import Any -from schema.types import Data, Event, Operation +from schema.types import Data, Event, NodeId, Operation, Workflow def _topology_lists_insertion_order(by_id: dict[str, dict]) -> list[dict]: @@ -229,3 +229,89 @@ def merge_event_data(existing: Data | None, event: Event) -> Data: out[key] = copy.deepcopy(val) return Data.model_validate(out) + + +def _node_stable_agent_id(node: Any) -> str | None: + """Return a node's ``stable_agent_id`` string, or ``None`` when absent.""" + sid = getattr(node, "stable_agent_id", None) + if sid is None: + return None + root = getattr(sid, "root", None) + if isinstance(root, str): + return root + return sid if isinstance(sid, str) else None + + +def _existing_stable_agent_id_index( + state_wf: Workflow | None, + instance_id: str, +) -> dict[str, str]: + """Map ``stable_agent_id`` -> existing node id for one instance. + + Uses the instance topology when it already has nodes, else falls back to + the workflow ``starting_topology`` (mirrors the instance-seeding policy in + :func:`_merge_workflow`). + """ + if state_wf is None: + return {} + nodes: list = [] + inst = state_wf.instances.get(instance_id) + if inst is not None and inst.topology is not None and inst.topology.nodes: + nodes = inst.topology.nodes + elif ( + state_wf.starting_topology is not None and state_wf.starting_topology.nodes + ): + nodes = state_wf.starting_topology.nodes + index: dict[str, str] = {} + for node in nodes: + sid = _node_stable_agent_id(node) + if sid is not None: + index.setdefault(sid, node.id.root) + return index + + +def reconcile_event_node_identities(state: Data, event: Event) -> Event: + """Resolve node identity server-side before merging. + + When an incoming ``create`` node carries a ``stable_agent_id`` that already + exists in the target instance (a cross-service "anchor" reference, e.g. the + recruiter pointing discovered agents at the seeded recruiter node), it is + treated as the **same** node: the duplicate ``create`` is dropped and every + edge endpoint referencing its id is repointed to the existing node id. This + makes anchoring an authoritative backend decision rather than a frontend + one. + + Returns a new :class:`Event`; the input is never mutated. + """ + result = event.model_copy(deep=True) + for wf_name, wf in result.data.workflows.items(): + state_wf = state.workflows.get(wf_name) + for instance_id, inst in wf.instances.items(): + topology = inst.topology + if topology is None: + continue + existing = _existing_stable_agent_id_index(state_wf, instance_id) + if not existing: + continue + id_remap: dict[str, str] = {} + kept_nodes: list = [] + for node in topology.nodes or []: + sid = _node_stable_agent_id(node) + existing_id = existing.get(sid) if sid is not None else None + if ( + node.operation == Operation.CREATE + and existing_id is not None + and existing_id != node.id.root + ): + id_remap[node.id.root] = existing_id + continue + kept_nodes.append(node) + if not id_remap: + continue + topology.nodes = kept_nodes + for edge in topology.edges or []: + if edge.source is not None and edge.source.root in id_remap: + edge.source = NodeId(root=id_remap[edge.source.root]) + if edge.target is not None and edge.target.root in id_remap: + edge.target = NodeId(root=id_remap[edge.target.root]) + return result diff --git a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/store.py b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/store.py index 0a59959c5..7f4c3d3c1 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/store.py +++ b/coffeeAGNTCY/coffee_agents/lungo/common/workflow_instance_store/store.py @@ -41,7 +41,13 @@ from schema.validation import validate_data_against_schema from common.workflow_instance_store.errors import WorkflowInstanceStoreClosedError -from common.workflow_instance_store.merge import merge_event_data +from common.workflow_instance_store.discovery_layout import ( + enrich_discovery_node_layout, +) +from common.workflow_instance_store.merge import ( + merge_event_data, + reconcile_event_node_identities, +) from common.workflow_instance_store.notifier import NoOpNotifier, NotifierProtocol EVENT_SCHEMA = "event_v1" @@ -138,9 +144,11 @@ def _merge_loop(self) -> None: try: touched = _touched_instance_ids(item) with self._state_lock: - self._state = merge_event_data(self._state, item) + normalized = reconcile_event_node_identities(self._state, item) + normalized = enrich_discovery_node_layout(self._state, normalized) + self._state = merge_event_data(self._state, normalized) if touched: - self._on_merged(item, touched) + self._on_merged(normalized, touched) except Exception: logger.exception("WorkflowInstanceStateStore merge worker failed") finally: diff --git a/coffeeAGNTCY/coffee_agents/lungo/config/config.py b/coffeeAGNTCY/coffee_agents/lungo/config/config.py index a98bb210a..ee8ba30da 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/config/config.py +++ b/coffeeAGNTCY/coffee_agents/lungo/config/config.py @@ -23,6 +23,14 @@ DEFAULT_MESSAGE_TRANSPORT = os.getenv("DEFAULT_MESSAGE_TRANSPORT", "SLIM") +# Host used to reach discovered agents that advertise an unroutable bind +# address (e.g. http://0.0.0.0:9999). Defaults to localhost for host runs; +# set to host.docker.internal inside containers. +DISCOVERED_AGENT_HOST = os.getenv("DISCOVERED_AGENT_HOST", "localhost") + +# Read timeout (seconds) for the recruiter's JSONRPC delegation client. +A2A_CLIENT_TIMEOUT_SECONDS = float(os.getenv("A2A_CLIENT_TIMEOUT_SECONDS", "30")) + if os.getenv("SLIM_SHARED_SECRET") is None: # set a default value for development/testing os.environ["SLIM_SHARED_SECRET"] = "slim-shared-secret-REPLACE_WITH_RANDOM_32PLUS_CHARS" diff --git a/coffeeAGNTCY/coffee_agents/lungo/docker-compose.yaml b/coffeeAGNTCY/coffee_agents/lungo/docker-compose.yaml index c6b0f1f8f..f94151fa5 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/docker-compose.yaml +++ b/coffeeAGNTCY/coffee_agents/lungo/docker-compose.yaml @@ -596,6 +596,9 @@ services: - CORS_ALLOWED_ORIGINS=${CORS_ALLOWED_ORIGINS:-} - WORKFLOW_API_URL=http://agentic-workflows-api:9105 - WORKFLOW_API_KEY=${WORKFLOW_API_KEY:?WORKFLOW_API_KEY must be set in .env} + - DISCOVERED_AGENT_HOST=${DISCOVERED_AGENT_HOST:-host.docker.internal} + extra_hosts: + - "host.docker.internal:host-gateway" ports: - "8882:8882" depends_on: diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/api/agenticWorkflowsTypes.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/api/agenticWorkflowsTypes.ts index 7281741fe..653c06ee8 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/api/agenticWorkflowsTypes.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/api/agenticWorkflowsTypes.ts @@ -26,6 +26,10 @@ export interface TopologyNodeWire { position?: TopologyPosition agent_record_uri?: string stable_agent_id?: string | { root: string } + /** Inline agent record (flat AgentCard dict) for runtime-discovered agents. */ + oasf_record?: Record + /** Directory content id (CID) for a discovered agent. */ + agent_cid?: string [key: string]: unknown } diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.test.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.test.ts index d6b14851b..6e39cb10b 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.test.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.test.ts @@ -5,6 +5,7 @@ import type { Edge, Node } from "@xyflow/react" import { describe, expect, it } from "vitest" +import type { GraphConfig } from "@/utils/graphConfigs" import { A2A_HTTP_CONFIG, GROUP_MESSAGING_CONFIG, @@ -16,6 +17,18 @@ import { resolveStreamAuthorToNodeId, } from "./chatStreamGraphHighlight" +function liveNode( + id: string, + type: string, + data: Record, +): Node { + return { id, type, position: { x: 0, y: 0 }, data } +} + +function liveEdge(id: string, source: string, target: string): Edge { + return { id, source, target } +} + describe("resolveStreamAuthorToNodeId", () => { it.each([ { @@ -36,6 +49,40 @@ describe("resolveStreamAuthorToNodeId", () => { config: GROUP_MESSAGING_CONFIG, expected: NODE_IDS.AUCTION_AGENT, }, + { + caseName: "discovered agent author via inline record name", + author: "Brazil Farm Agent", + config: { + title: "Discovery", + animationSequence: [], + nodes: [ + liveNode("agent://brazil", "customNode", { + label1: "Brazil", + oasfRecord: { name: "Brazil Farm Agent" }, + agentCid: "cid-brazil", + }), + ], + edges: [], + } satisfies GraphConfig, + expected: "agent://brazil", + }, + { + caseName: "discovered agent author via cid", + author: "cid-brazil", + config: { + title: "Discovery", + animationSequence: [], + nodes: [ + liveNode("agent://brazil", "customNode", { + label1: "Brazil", + oasfRecord: { name: "Brazil Farm Agent" }, + agentCid: "cid-brazil", + }), + ], + edges: [], + } satisfies GraphConfig, + expected: "agent://brazil", + }, ])("$caseName", ({ author, config, expected }) => { expect(resolveStreamAuthorToNodeId(author, config)).toBe(expected) }) @@ -57,18 +104,6 @@ describe("animationSequenceStepIds", () => { }) }) -function liveNode( - id: string, - type: string, - data: Record, -): Node { - return { id, type, position: { x: 0, y: 0 }, data } -} - -function liveEdge(id: string, source: string, target: string): Edge { - return { id, source, target } -} - describe("deriveAnimationSequenceFromGraph", () => { const nodes: Node[] = [ liveNode("agent://auction", "customNode", {}), diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.ts index d5f54fc84..9cdb729d9 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.ts @@ -5,9 +5,9 @@ * Map chat stream authors / sequence steps to React Flow graph element ids. * * The backend owns the graph, so there are no authored NODE_IDS/EDGE_IDS to map - * from. Stream authors resolve through a slug bridge (recruiter/directory slug - * aliases, then live labels), and the streaming animation sequence is derived - * straight from the live graph topology rather than translated from a static one. + * from. Stream authors resolve through recruiter/directory slug aliases, inline + * discovered-agent record keys (CID / record name / label), then live labels. + * The streaming animation sequence is derived from the live graph topology. */ import type { Edge, Node } from "@xyflow/react" @@ -24,6 +24,41 @@ function normalizeAuthor(value: string): string { return value.trim().toLowerCase() } +function discoveredAgentLookupKeys( + data: CustomNodeData | undefined, +): readonly string[] { + if (!data) return [] + const keys: string[] = [] + if (typeof data.agentCid === "string" && data.agentCid.trim()) { + keys.push(normalizeAuthor(data.agentCid)) + } + const record = data.oasfRecord + if (record && typeof record === "object") { + const name = record.name + if (typeof name === "string" && name.trim()) { + keys.push(normalizeAuthor(name)) + } + } + if (typeof data.label1 === "string" && data.label1.trim()) { + keys.push(normalizeAuthor(data.label1)) + } + return keys +} + +function discoveredAgentNodeIdMap( + graphConfig: GraphConfig | undefined, +): Map { + const map = new Map() + for (const node of graphConfig?.nodes ?? []) { + const data = node.data as unknown as CustomNodeData | undefined + if (!data?.oasfRecord && !data?.agentCid) continue + for (const key of discoveredAgentLookupKeys(data)) { + if (!map.has(key)) map.set(key, node.id) + } + } + return map +} + /** Known recruiter / directory stream authors mapped to a canonical node slug. */ const AUTHOR_SLUG_ALIASES: Readonly> = { recruiter_service: "recruiter", @@ -35,13 +70,12 @@ const AUTHOR_SLUG_ALIASES: Readonly> = { } /** - * Stable identity key shared by static-config and API-driven nodes. Transport and - * group nodes resolve by type; directory has no OASF slug, so it is detected from - * its labels; everything else defers to `getOasfSlugFromNodeData`. + * Stable identity key shared by static-config and API-driven nodes. Transport + * resolves by type; directory has no OASF slug, so it is detected from its + * labels; everything else defers to `getOasfSlugFromNodeData`. */ function nodeSlugKey(node: Node): string | null { if (node.type === NODE_TYPES.TRANSPORT) return "transport" - if (node.type === NODE_TYPES.GROUP) return "logistics-group" const data = node.data as unknown as CustomNodeData | undefined const label1 = data?.label1?.toLowerCase() ?? "" @@ -86,6 +120,9 @@ export function resolveStreamAuthorToNodeId( if (bySlug) return bySlug } + const byDiscovered = discoveredAgentNodeIdMap(graphConfig).get(normalized) + if (byDiscovered) return byDiscovered + const labelMap = buildSenderToNodeMap(graphConfig) return labelMap[author] ?? labelMap[normalized] ?? null } diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/Graph/Elements/types.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/Graph/Elements/types.ts index 93f889fae..113c21287 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/Graph/Elements/types.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/Graph/Elements/types.ts @@ -14,6 +14,8 @@ export interface CustomNodeData { active?: boolean selected?: boolean agentCid?: string + /** Inline agent record for runtime-discovered agents (skips directory fetch). */ + oasfRecord?: Record handles?: "all" | "target" | "source" extraHandles?: ExtraHandle[] verificationStatus?: "verified" | "failed" | "pending" diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainArea.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainArea.ts index b9bed5360..66e9804e8 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainArea.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainArea.ts @@ -11,9 +11,7 @@ import { useViewportAwareFitView } from "@/hooks/useViewportAwareFitView" import { useModalManager } from "@/hooks/useModalManager" import { NODE_IDS } from "@/utils/const.ts" import { applyDynamicTransportLabels } from "@/utils/dynamicTransportLabels" -import type { DiscoveryResponseEvent } from "@/types/agent" import type { CustomNodeData } from "./Graph/Elements/types" -import { useMainAreaDiscoveryGraph } from "./useMainAreaDiscoveryGraph" import { useMainAreaGraphEffects } from "./useMainAreaGraphEffects" import { useWorkflowGraphFromAgenticApi } from "@/hooks/useWorkflowGraphFromAgenticApi" import { useNodeTransportInterfaces } from "./useNodeTransportInterfaces" @@ -29,12 +27,10 @@ export interface MainAreaProps { buttonClicked: boolean setButtonClicked: (clicked: boolean) => void aiReplied: boolean - setAiReplied: (replied: boolean) => void chatHeight?: number isExpanded?: boolean groupCommResponseReceived?: boolean onNodeHighlight?: (highlightFunction: (nodeId: string) => void) => void - discoveryResponseEvent?: DiscoveryResponseEvent | null selectedAgentCid?: string | null /** Latest graph snapshot for chat/feeds (GroupCommunication sender map). */ onLiveGraphConfig?: (config: GraphConfig) => void @@ -49,12 +45,10 @@ export function useMainArea({ buttonClicked, setButtonClicked, aiReplied, - setAiReplied, chatHeight = 0, isExpanded = false, groupCommResponseReceived = false, onNodeHighlight, - discoveryResponseEvent, selectedAgentCid, onLiveGraphConfig, }: MainAreaProps) { @@ -115,15 +109,6 @@ export function useMainArea({ logger.error("agentic-workflows/graph-session", { detail: agenticError }) }, [agenticError]) - useMainAreaDiscoveryGraph({ - pattern, - discoveryResponseEvent, - setNodes, - setEdges, - handleOpenIdentityModal, - handleOpenOasfModal, - }) - const nodeAgentCidKey = useMemo( () => nodes @@ -230,7 +215,6 @@ export function useMainArea({ buttonClicked, setButtonClicked, aiReplied, - setAiReplied, pattern, updateStyle, config.animationSequence, diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainAreaDiscoveryGraph.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainAreaDiscoveryGraph.ts deleted file mode 100644 index 621f93dab..000000000 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/MainArea/useMainAreaDiscoveryGraph.ts +++ /dev/null @@ -1,292 +0,0 @@ -/** - * Copyright AGNTCY Contributors (https://github.com/agntcy) - * SPDX-License-Identifier: Apache-2.0 - **/ - -import React, { useEffect, useRef, useCallback } from "react" -import type { Node, Edge } from "@xyflow/react" -import { - PUBLISH_SUBSCRIBE_CONFIG, - GROUP_MESSAGING_CONFIG, -} from "@/utils/graphConfigs" -import { EDGE_LABELS, HANDLE_TYPES } from "@/utils/const.ts" -import { logger } from "@/utils/logger" -import { AgentRecord, DiscoveryResponseEvent } from "@/types/agent" -import type { CustomNodeData } from "./Graph/Elements/types" -import { CUSTOM_NODE_WIDTH } from "@/utils/graphNodeDimensions" - -const DISCOVERY_KEYWORDS = [ - "brazil", - "vietnam", - "colombia", - "shipper", - "tatooine", - "accountant", -] as const -type DiscoveryKeyword = (typeof DISCOVERY_KEYWORDS)[number] - -const RECRUITER_NODE_ID = "recruiter-agent" - -function findRecruiterNodeId(nodes: Node[]): string { - const byLabel = nodes.find((n) => { - const l1 = String((n.data as { label1?: string } | undefined)?.label1 ?? "") - .toLowerCase() - .trim() - return l1.includes("recruiter") - }) - if (byLabel) return byLabel.id - const legacy = nodes.find((n) => n.id === RECRUITER_NODE_ID) - return legacy?.id ?? RECRUITER_NODE_ID -} -const RECRUITER_POS = { x: 400, y: 300 } -const NODE_WIDTH = CUSTOM_NODE_WIDTH -const START_Y_OFFSET = 450 -const HORIZONTAL_GAP = 40 - -const hasKeyword = (name: string, kw: string) => - new RegExp(`\\b${kw}\\b`, "i").test(name) -const safeIdPart = (s: string) => s.replace(/[^a-zA-Z0-9_-]/g, "_") - -const TEMPLATES_BY_KEYWORD: Record = { - brazil: PUBLISH_SUBSCRIBE_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("brazil"), - ), - vietnam: PUBLISH_SUBSCRIBE_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("vietnam"), - ), - colombia: PUBLISH_SUBSCRIBE_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("colombia"), - ), - shipper: GROUP_MESSAGING_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("shipper"), - ), - tatooine: GROUP_MESSAGING_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("tatooine"), - ), - accountant: GROUP_MESSAGING_CONFIG.nodes.find((n) => - String(n.data?.label1 ?? "") - .toLowerCase() - .includes("accountant"), - ), -} - -export interface UseMainAreaDiscoveryGraphParams { - pattern: string - discoveryResponseEvent: DiscoveryResponseEvent | null | undefined - setNodes: React.Dispatch> - setEdges: React.Dispatch> - handleOpenIdentityModal: (nodeId: string, nodeData: CustomNodeData) => void - handleOpenOasfModal: (nodeData: CustomNodeData) => void -} - -/** Syncs discovery response events to the graph: adds/removes discovery nodes and edges. */ -export function useMainAreaDiscoveryGraph({ - pattern, - discoveryResponseEvent, - setNodes, - setEdges, - handleOpenIdentityModal, - handleOpenOasfModal, -}: UseMainAreaDiscoveryGraphParams) { - const seqRef = useRef(0) - const lastTsRef = useRef(null) - - const addDiscoveryResponseGraph = useCallback( - (_evt: DiscoveryResponseEvent) => { - logger.debug("Adding discovery response graph for event", _evt) - const raw = (_evt as { agent_records?: Record }) - ?.agent_records - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return - - const entries = Object.entries(raw) - .map(([id, rec]) => ({ - id: String(id), - name: String(rec?.name ?? "").trim(), - record: rec, - })) - .filter((e) => e.name.length > 0) - - if (entries.length === 0) { - setNodes((prevNodes) => - prevNodes.filter((n) => !n.id.startsWith("discovery-")), - ) - setEdges((prevEdges) => - prevEdges.filter( - (e) => - !e.id.startsWith("edge-recruiter-agent-") && - !e.id.startsWith("edge-node://"), - ), - ) - return - } - - const baseId = ++seqRef.current - const templateNodeId = (kind: DiscoveryKeyword) => - `discovery-${kind}-${baseId}` - const generatedNodeId = (agentId: string) => - `discovery-agent-${baseId}-${safeIdPart(agentId)}` - const edgeId = (targetId: string, sourceId: string) => - `edge-${sourceId}-${baseId}-${targetId}` - - setNodes((prevNodes) => { - const existingIds = new Set(prevNodes.map((n) => n.id)) - const existingNames = new Set( - prevNodes - .map((n) => - String((n.data as Record)?.label1 ?? "") - .trim() - .toLowerCase(), - ) - .filter(Boolean), - ) - const existingCids = new Set( - prevNodes - .map((n) => (n.data as Record)?.agentCid) - .filter(Boolean), - ) - const recruiterSourceId = findRecruiterNodeId(prevNodes) - const recruiterNode = prevNodes.find((n) => n.id === recruiterSourceId) - const recruiterIcons = recruiterNode?.data - ? { - icon: recruiterNode.data.icon, - iconUrl: recruiterNode.data.iconUrl, - image: recruiterNode.data.image, - } - : {} - const templateKeywordsToAdd: DiscoveryKeyword[] = - DISCOVERY_KEYWORDS.filter((kw) => - entries.some((e) => hasKeyword(e.name, kw)), - ).filter((kw) => { - const alreadyExists = prevNodes.some( - (n) => - n.id.startsWith("discovery-") && - hasKeyword( - String((n.data as Record)?.label1 ?? ""), - kw, - ), - ) - return !alreadyExists - }) - const generatedEntriesToAdd = (() => { - const seen = new Set() - return entries.filter((e) => { - if (DISCOVERY_KEYWORDS.some((kw) => hasKeyword(e.name, kw))) - return false - const key = e.name.toLowerCase() - if (existingNames.has(key)) return false - if (seen.has(key)) return false - if (existingCids.has(e.id)) return false - seen.add(key) - return true - }) - })() - const total = - templateKeywordsToAdd.length + generatedEntriesToAdd.length - const startCol = -Math.floor(total / 2) - let col = 0 - const newNodes: Node[] = [] - const newEdges: Edge[] = [] - - for (const kind of templateKeywordsToAdd) { - const template = TEMPLATES_BY_KEYWORD[kind] - if (!template) continue - const id = templateNodeId(kind) - if (existingIds.has(id)) continue - const matchedEntry = entries.find((e) => hasKeyword(e.name, kind)) - if (matchedEntry?.id && existingCids.has(matchedEntry.id)) continue - const x = - RECRUITER_POS.x + (startCol + col) * (NODE_WIDTH + HORIZONTAL_GAP) - const y = RECRUITER_POS.y + START_Y_OFFSET - col++ - const { parentId, extent, ...rest } = template - newNodes.push({ - ...rest, - id, - position: { x, y }, - ...(parentId && existingIds.has(parentId) - ? { parentId, extent } - : {}), - data: { - ...(template.data ?? {}), - active: false, - selected: false, - agentCid: matchedEntry?.id, - isModalOpen: false, - onOpenIdentityModal: handleOpenIdentityModal, - onOpenOasfModal: handleOpenOasfModal, - }, - }) - newEdges.push({ - id: edgeId(id, recruiterSourceId), - source: recruiterSourceId, - target: id, - type: "custom", - data: { label: EDGE_LABELS.A2A_OVER_HTTP }, - }) - } - - for (const entry of generatedEntriesToAdd) { - const id = generatedNodeId(entry.id) - if (existingIds.has(id)) continue - const x = - RECRUITER_POS.x + (startCol + col) * (NODE_WIDTH + HORIZONTAL_GAP) - const y = RECRUITER_POS.y + START_Y_OFFSET - col++ - newNodes.push({ - id, - type: "customNode", - position: { x, y }, - data: { - label1: entry.name, - ...recruiterIcons, - active: false, - selected: false, - agentCid: entry.id, - isModalOpen: false, - handles: HANDLE_TYPES.TARGET, - agentDirectoryLink: "place-holder", - oasfRecord: entry.record, - onOpenIdentityModal: handleOpenIdentityModal, - onOpenOasfModal: handleOpenOasfModal, - }, - }) - newEdges.push({ - id: edgeId(id, recruiterSourceId), - source: recruiterSourceId, - target: id, - type: "custom", - data: { label: EDGE_LABELS.A2A_OVER_HTTP }, - }) - } - - if (newNodes.length === 0) return prevNodes - setEdges((prevEdges) => { - const existingEdgeIds = new Set(prevEdges.map((e) => e.id)) - const filtered = newEdges.filter((e) => !existingEdgeIds.has(e.id)) - return filtered.length ? [...prevEdges, ...filtered] : prevEdges - }) - return [...prevNodes, ...newNodes] - }) - }, - [setNodes, setEdges, handleOpenIdentityModal, handleOpenOasfModal], - ) - - useEffect(() => { - if (pattern !== "a2a_http") return - if (!discoveryResponseEvent) return - if (lastTsRef.current === discoveryResponseEvent.ts) return - lastTsRef.current = discoveryResponseEvent.ts - if (!discoveryResponseEvent.response?.trim()) return - addDiscoveryResponseGraph(discoveryResponseEvent) - }, [pattern, discoveryResponseEvent, addDiscoveryResponseGraph]) -} diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphFromAgenticApi.types.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphFromAgenticApi.types.ts index be5644374..6685bf472 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphFromAgenticApi.types.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphFromAgenticApi.types.ts @@ -15,22 +15,6 @@ export const SSE_RECONNECT_BACKOFF_MS = 250 /** Auto-clear messaging highlights if no newer event refreshes them. */ export const MESSAGING_HIGHLIGHT_TTL_MS = 2_500 -export function mergeDiscoveryNodes(base: Node[], prev: Node[]): Node[] { - const overlay = prev.filter((n) => n.id.startsWith("discovery-")) - if (overlay.length === 0) return base - return [...base, ...overlay] -} - -export function mergeDiscoveryEdges(base: Edge[], prev: Edge[]): Edge[] { - const overlay = prev.filter( - (e) => - e.source.startsWith("discovery-") || e.target.startsWith("discovery-"), - ) - if (overlay.length === 0) return base - const ids = new Set(base.map((e) => e.id)) - return [...base, ...overlay.filter((e) => !ids.has(e.id))] -} - export interface UseWorkflowGraphFromAgenticApiParams { pattern: PatternType selectedWorkflowSummary: WorkflowSummary | null diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphTopologySync.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphTopologySync.ts index dfa52b773..14d3f345f 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphTopologySync.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/hooks/useWorkflowGraphTopologySync.ts @@ -15,8 +15,6 @@ import { identityUiVariantForPattern } from "@/utils/agenticTopologyIdentityUiMa import { topologyWireToReactFlow } from "@/utils/topologyToReactFlow" import type { PatternType } from "@/utils/patternUtils" import { - mergeDiscoveryEdges, - mergeDiscoveryNodes, REFETCH_DEBOUNCE_MS, type WorkflowGraphAgenticSession, } from "./useWorkflowGraphFromAgenticApi.types" @@ -49,12 +47,11 @@ export function useWorkflowGraphTopologySync({ identityUiVariant: identityUiVariantForPattern(patternRef.current), }) const withHandlers = mappedNodes.map(attachHandlers) - setNodes((prev) => { - const merged = mergeDiscoveryNodes(withHandlers, prev) - lastAppliedGraphNodeIdsRef.current = new Set(merged.map((n) => n.id)) - return merged - }) - setEdges((prev) => mergeDiscoveryEdges(mappedEdges, prev)) + lastAppliedGraphNodeIdsRef.current = new Set( + withHandlers.map((n) => n.id), + ) + setNodes(withHandlers) + setEdges(mappedEdges) onAppliedRef.current?.() queueMicrotask(() => { restoreEdgeAnimation() diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/pages/RootPage.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/pages/RootPage.tsx index b6473a5b8..fe445c3d7 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/pages/RootPage.tsx +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/pages/RootPage.tsx @@ -55,7 +55,6 @@ const RootPage: React.FC = () => { showRecruiterStreaming, showFinalResponse, groupCommResponseReceived, - discoveryResponseEvent, handleUserInput, handleApiResponse, handleSendPrompt, @@ -188,12 +187,10 @@ const RootPage: React.FC = () => { buttonClicked={buttonClicked} setButtonClicked={setButtonClicked} aiReplied={aiReplied} - setAiReplied={setAiReplied} chatHeight={chatHeightValue} isExpanded={isExpanded} groupCommResponseReceived={groupCommResponseReceived} onNodeHighlight={handleNodeHighlightSetup} - discoveryResponseEvent={discoveryResponseEvent} selectedAgentCid={ typeof recruiterSelectedAgent?.cid === "string" ? recruiterSelectedAgent.cid diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/types/agent.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/types/agent.ts index 74de6604e..1ddbb062f 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/types/agent.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/types/agent.ts @@ -3,17 +3,9 @@ * SPDX-License-Identifier: Apache-2.0 **/ -/** Shared agent/directory record shape (discovery, recruiter streaming, etc.). */ +/** Shared agent/directory record shape (recruiter streaming state, etc.). */ export type AgentRecord = { id?: string name?: string [key: string]: unknown } - -/** Event payload when discovery completes; used to sync graph with agent_records. */ -export type DiscoveryResponseEvent = { - response: string - ts: number - sessionId?: string - agent_records?: Record -} diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts index 2dfb65c7e..e0dabe4cc 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts @@ -18,7 +18,6 @@ import { useAppChatState } from "@/hooks/useAppChatState" import { useAgentAPI } from "@/hooks/useAgentAPI" import { getGraphConfig, type GraphConfig } from "@/utils/graphConfigs" import { PATTERNS, PatternType } from "@/utils/patternUtils" -import { DiscoveryResponseEvent } from "@/types/agent" import { AGENTIC_WORKFLOWS_CATALOG_LOG_PATH, fetchWorkflowSummariesWithRetry, @@ -77,18 +76,11 @@ export function useApp() { const chat = useAppChatState({ selectedPattern, canvasMode }) const streamCompleteRef = useRef(false) - const [discoveryResponseEvent, setDiscoveryResponseEvent] = - useState(null) - const lastDiscoveryKeyRef = useRef(null) const [highlightNodeFunction, setHighlightNodeFunction] = useState< ((nodeId: string) => void) | null >(null) - const handleDiscoveryResponse = useCallback((evt: DiscoveryResponseEvent) => { - setDiscoveryResponseEvent(evt) - }, []) - const selectWorkflowFromCatalog = useCallback( (summary: WorkflowSummary) => { const slug = mapWorkflowNameToSlug(summary.name) @@ -111,7 +103,6 @@ export function useApp() { setSelectedWorkflowSummary(summary) setLiveGraphConfig(null) setSelectedReferencePattern(null) - lastDiscoveryKeyRef.current = null }, // eslint-disable-next-line react-hooks/exhaustive-deps -- streaming/chat refs stable enough; full deps cause unnecessary runs [streaming.reset, streaming.resetRecruiter, streaming.resetGroup, chat], @@ -271,21 +262,6 @@ export function useApp() { false, ) } - - const agentKeys = streaming.recruiterAgentRecords - ? Object.keys(streaming.recruiterAgentRecords).sort().join(",") - : "" - const discoveryKey = `${streaming.recruiterSessionId ?? ""}:${agentKeys}` - - if (lastDiscoveryKeyRef.current !== discoveryKey) { - lastDiscoveryKeyRef.current = discoveryKey - handleDiscoveryResponse({ - response: streaming.recruiterFinalMessage ?? "", - ts: Date.now(), - sessionId: streaming.recruiterSessionId ?? undefined, - agent_records: streaming.recruiterAgentRecords ?? undefined, - }) - } } else if ( streaming.recruiterStatus === "error" && streaming.recruiterError @@ -309,7 +285,6 @@ export function useApp() { chat.handleApiResponse, chat.setIsAgentLoading, chat.setShowFinalResponse, - handleDiscoveryResponse, ]) const handleSendPrompt = useCallback( @@ -401,7 +376,6 @@ export function useApp() { chat.resetChatState() streaming.resetGroup() streaming.resetRecruiter() - lastDiscoveryKeyRef.current = null // In pattern_doc mode the server keys conversation state by session_id; // rotate it so a cleared chat starts a genuinely fresh backend session. if (selectedReferencePattern !== null) { @@ -487,7 +461,6 @@ export function useApp() { showRecruiterStreaming: chat.showRecruiterStreaming, showFinalResponse: chat.showFinalResponse, groupCommResponseReceived: chat.groupCommResponseReceived, - discoveryResponseEvent, handleUserInput: chat.handleUserInput, handleApiResponse: chat.handleApiResponse, handleSendPrompt, @@ -495,7 +468,6 @@ export function useApp() { handleClearConversation, handleNodeHighlightSetup, handleSenderHighlight, - handleDiscoveryResponse, graphConfig, events: streaming.events, status: streaming.status, diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.test.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.test.ts index 407a42156..6866502b7 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.test.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.test.ts @@ -8,6 +8,7 @@ import type { TopologyNodeWire } from "@/api/agenticWorkflowsTypes" import { LUNGO_FRONTEND_URLS } from "@/urls" import type { CustomNodeData } from "@/components/MainArea/Graph/Elements/types" import { + applyDiscoveredAgentInlineUi, getOasfSlugFromNodeData, IDENTITY_UI_BY_STABLE_AGENT_UUID, isDirectoryLabel, @@ -166,6 +167,48 @@ describe("agenticTopologyIdentityUiMap", () => { }) }) + describe("applyDiscoveredAgentInlineUi", () => { + const baseData: CustomNodeData = { + icon: null, + label1: "Brazil", + label2: "", + handles: "all", + } + + it("returns data unchanged when the wire has no inline OASF record", () => { + const out = applyDiscoveredAgentInlineUi(baseData, { + id: "n1", + } as TopologyNodeWire) + expect(out).toBe(baseData) + }) + + it("threads inline OASF record, cid, target handle and directory link", () => { + const record = { name: "Brazil", url: "http://brazil:9000" } + const out = applyDiscoveredAgentInlineUi(baseData, { + id: "n1", + oasf_record: record, + agent_cid: "cidB", + } as unknown as TopologyNodeWire) + expect(out.oasfRecord).toEqual(record) + expect(out.agentCid).toBe("cidB") + expect(out.handles).toBe("target") + expect(out.agentDirectoryLink).toBe( + LUNGO_FRONTEND_URLS.agentDirectory.baseUrl, + ) + }) + + it("keeps an existing directory link when already present", () => { + const out = applyDiscoveredAgentInlineUi( + { ...baseData, agentDirectoryLink: "https://example.com/x" }, + { + id: "n1", + oasf_record: { name: "Brazil" }, + } as unknown as TopologyNodeWire, + ) + expect(out.agentDirectoryLink).toBe("https://example.com/x") + }) + }) + describe("label predicates", () => { it.each([ { caseName: "weather mcp server", label: "Weather MCP Server", v: true }, diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.ts index f4890e792..4cc5bb22e 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/agenticTopologyIdentityUiMap.ts @@ -14,7 +14,7 @@ import { v5 as uuidv5 } from "uuid" import type { CustomNodeData } from "@/components/MainArea/Graph/Elements/types" import type { TopologyNodeWire } from "@/api/agenticWorkflowsTypes" -import { VERIFICATION_STATUS } from "@/utils/const" +import { HANDLE_TYPES, VERIFICATION_STATUS } from "@/utils/const" import { PATTERNS, type PatternType } from "@/utils/patternUtils" import { LUNGO_FRONTEND_URLS } from "@/urls" import { SecurityClass } from "@/utils/SecurityClass" @@ -355,6 +355,30 @@ export function enrichAgenticTopologyWellKnownUi( return data } +/** + * Apply inline UI for runtime-discovered agents. The recruiter emits the flat + * agent record (and CID) on the wire ``oasf_record`` field, so the directory + * modal reads it directly instead of resolving a static directory slug. + * Discovered agents are leaf targets of the recruiter edge, so they expose a + * target handle only. + */ +export function applyDiscoveredAgentInlineUi( + data: CustomNodeData, + wire: TopologyNodeWire, +): CustomNodeData { + const record = wire.oasf_record + if (!record || typeof record !== "object") return data + const cid = typeof wire.agent_cid === "string" ? wire.agent_cid : undefined + return { + ...data, + oasfRecord: record as Record, + agentCid: cid, + handles: HANDLE_TYPES.TARGET, + agentDirectoryLink: + data.agentDirectoryLink ?? LUNGO_FRONTEND_URLS.agentDirectory.baseUrl, + } +} + /** * Merge map + resolved GitHub into agentic `CustomNodeData`. Does not set legacy `slug` * (use `identityAppsSlug` / `directoryAgentSlug` on node data). diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/graphConfigsData.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/graphConfigsData.tsx index 8bfdd0529..b7210b844 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/graphConfigsData.tsx +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/graphConfigsData.tsx @@ -239,15 +239,11 @@ export const GROUP_MESSAGING_CONFIG: GraphConfig = { { id: NODE_IDS.LOGISTICS_GROUP, type: NODE_TYPES.GROUP, - data: { label: "Logistics Group" }, + data: {}, position: { x: 50, y: 50 }, - style: { - width: 900, - height: 650, - backgroundColor: "var(--group-background)", - border: "none", - borderRadius: "8px", - }, + hidden: true, + width: 900, + height: 650, }, { id: NODE_IDS.AUCTION_AGENT, diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyGroupLayout.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyGroupLayout.ts deleted file mode 100644 index 60e69f29a..000000000 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyGroupLayout.ts +++ /dev/null @@ -1,301 +0,0 @@ -/** - * Copyright AGNTCY Contributors (https://github.com/agntcy) - * SPDX-License-Identifier: Apache-2.0 - **/ - -import type { Node } from "@xyflow/react" -import { NODE_TYPES } from "@/utils/const" -import { - bucketNodeLayerY, - CUSTOM_NODE_HEIGHT, - CUSTOM_NODE_WIDTH, - CUSTOM_NODE_X_GAP, - TRANSPORT_NODE_HEIGHT_COMPACT, - TRANSPORT_NODE_WIDTH_BAR, - TRANSPORT_NODE_WIDTH_COMPACT, -} from "./topologyLayout" - -const GROUP_CONTENT_PADDING = 48 -const TRANSPORT_NODE_HEIGHT_BAR = 52 - -function parseGroupStyleWidth( - style: { width?: unknown } | undefined, -): number | null { - const width = style?.width - if (typeof width === "number" && Number.isFinite(width)) return width - if (typeof width === "string") { - const parsed = Number.parseFloat(width) - if (Number.isFinite(parsed)) return parsed - } - return null -} - -function nodeLayoutWidth(node: Node): number { - if (node.type === NODE_TYPES.TRANSPORT) { - const compact = Boolean( - (node.data as { compact?: boolean } | undefined)?.compact, - ) - return compact ? TRANSPORT_NODE_WIDTH_COMPACT : TRANSPORT_NODE_WIDTH_BAR - } - if (node.type === NODE_TYPES.CUSTOM) { - return CUSTOM_NODE_WIDTH - } - return CUSTOM_NODE_WIDTH -} - -function nodeLayoutHeight(node: Node): number { - if (node.type === NODE_TYPES.TRANSPORT) { - const compact = Boolean( - (node.data as { compact?: boolean } | undefined)?.compact, - ) - return compact ? TRANSPORT_NODE_HEIGHT_COMPACT : TRANSPORT_NODE_HEIGHT_BAR - } - return CUSTOM_NODE_HEIGHT -} - -/** - * Shrinks the group container to its children and pads evenly so fitView centers - * the graph instead of leaving empty margin inside a wide group box. - * Uses card/body sizes only — side icons are excluded from bounds. - */ -export function fitGroupContainerToContent(nodes: Node[]): Node[] { - const group = nodes.find((node) => node.type === NODE_TYPES.GROUP) - if (!group) return nodes - - const children = nodes.filter((node) => node.parentId === group.id) - if (children.length === 0) return nodes - - let minX = Infinity - let minY = Infinity - let maxX = -Infinity - let maxY = -Infinity - - for (const child of children) { - const width = nodeLayoutWidth(child) - const height = nodeLayoutHeight(child) - minX = Math.min(minX, child.position.x) - minY = Math.min(minY, child.position.y) - maxX = Math.max(maxX, child.position.x + width) - maxY = Math.max(maxY, child.position.y + height) - } - - const contentWidth = maxX - minX - const contentHeight = maxY - minY - const shiftX = GROUP_CONTENT_PADDING - minX - const shiftY = GROUP_CONTENT_PADDING - minY - const newGroupWidth = contentWidth + GROUP_CONTENT_PADDING * 2 - const newGroupHeight = contentHeight + GROUP_CONTENT_PADDING * 2 - - return nodes.map((node) => { - if (node.id === group.id) { - return { - ...node, - style: { - ...node.style, - width: newGroupWidth, - height: newGroupHeight, - }, - } - } - if (node.parentId === group.id) { - return { - ...node, - position: { - x: node.position.x + shiftX, - y: node.position.y + shiftY, - }, - } - } - return node - }) -} - -/** - * Centers a single child (e.g. compact transport) horizontally in its parent group. - * Other children are shifted by the same delta so relative positions are preserved. - */ -export function centerGroupChildrenOnChildX( - nodes: Node[], - childId: string, - childWidth: number, -): Node[] { - const child = nodes.find((node) => node.id === childId) - if (!child?.parentId) return nodes - - const group = nodes.find( - (node) => node.id === child.parentId && node.type === NODE_TYPES.GROUP, - ) - const groupWidth = parseGroupStyleWidth(group?.style) - if (groupWidth == null) return nodes - - const childCenterX = child.position.x + childWidth / 2 - const dx = groupWidth / 2 - childCenterX - if (Math.abs(dx) < 0.01) return nodes - - return nodes.map((node) => - node.parentId === child.parentId - ? { - ...node, - position: { ...node.position, x: node.position.x + dx }, - } - : node, - ) -} - -/** - * Lays out custom nodes per row inside the group with equal left/right margins - * and even spacing between cards. Transport position is not changed. - */ -export function layoutCustomNodesInGroupSymmetric( - nodes: Node[], - transportId: string, -): Node[] { - const group = nodes.find((node) => node.type === NODE_TYPES.GROUP) - if (!group) return nodes - - const groupWidth = parseGroupStyleWidth(group.style) - if (groupWidth == null) return nodes - - const customChildren = nodes.filter( - (node) => - node.parentId === group.id && - node.type === NODE_TYPES.CUSTOM && - node.id !== transportId, - ) - if (customChildren.length === 0) return nodes - - const positions = new Map( - nodes.map((node) => [node.id, { ...node.position }]), - ) - - const byLayer = new Map() - for (const node of customChildren) { - const layer = bucketNodeLayerY(node.position.y) - if (!byLayer.has(layer)) byLayer.set(layer, []) - byLayer.get(layer)!.push(node.id) - } - - for (const ids of byLayer.values()) { - const sorted = [...ids].sort( - (a, b) => positions.get(a)!.x - positions.get(b)!.x, - ) - const count = sorted.length - const contentSpan = (count - 1) * CUSTOM_NODE_X_GAP + CUSTOM_NODE_WIDTH - const leftMargin = (groupWidth - contentSpan) / 2 - sorted.forEach((id, index) => { - const point = positions.get(id)! - positions.set(id, { - x: leftMargin + index * CUSTOM_NODE_X_GAP, - y: point.y, - }) - }) - } - - return nodes.map((node) => { - if (node.id === transportId) return node - const next = positions.get(node.id) - return next ? { ...node, position: next } : node - }) -} - -/** - * Single-pass compact group layout: sizes the group from symmetric row spans, - * centers each custom row and the transport with equal padding, and normalizes Y. - */ -export function layoutCompactGroupSymmetric( - nodes: Node[], - transportId: string, -): Node[] { - const group = nodes.find((node) => node.type === NODE_TYPES.GROUP) - if (!group) return nodes - - const transport = nodes.find((node) => node.id === transportId) - if (!transport?.parentId || transport.parentId !== group.id) return nodes - - const customChildren = nodes.filter( - (node) => - node.parentId === group.id && - node.type === NODE_TYPES.CUSTOM && - node.id !== transportId, - ) - if (customChildren.length === 0) return nodes - - const positions = new Map( - nodes.map((node) => [node.id, { ...node.position }]), - ) - - const byLayer = new Map() - for (const node of customChildren) { - const layer = bucketNodeLayerY(node.position.y) - if (!byLayer.has(layer)) byLayer.set(layer, []) - byLayer.get(layer)!.push(node.id) - } - - let contentWidth = TRANSPORT_NODE_WIDTH_COMPACT - for (const ids of byLayer.values()) { - const span = (ids.length - 1) * CUSTOM_NODE_X_GAP + CUSTOM_NODE_WIDTH - contentWidth = Math.max(contentWidth, span) - } - const groupWidth = contentWidth + GROUP_CONTENT_PADDING * 2 - - for (const ids of byLayer.values()) { - const sorted = [...ids].sort( - (a, b) => positions.get(a)!.x - positions.get(b)!.x, - ) - const layerSpan = - (sorted.length - 1) * CUSTOM_NODE_X_GAP + CUSTOM_NODE_WIDTH - const layerLeft = GROUP_CONTENT_PADDING + (contentWidth - layerSpan) / 2 - sorted.forEach((id, index) => { - const point = positions.get(id)! - positions.set(id, { - x: layerLeft + index * CUSTOM_NODE_X_GAP, - y: point.y, - }) - }) - } - - const transportPos = positions.get(transportId)! - positions.set(transportId, { - x: - GROUP_CONTENT_PADDING + (contentWidth - TRANSPORT_NODE_WIDTH_COMPACT) / 2, - y: transportPos.y, - }) - - const groupChildIds = new Set( - [...customChildren, transport].map((node) => node.id), - ) - let minY = Infinity - let maxY = -Infinity - for (const id of groupChildIds) { - const pos = positions.get(id)! - const height = - id === transportId ? TRANSPORT_NODE_HEIGHT_COMPACT : CUSTOM_NODE_HEIGHT - minY = Math.min(minY, pos.y) - maxY = Math.max(maxY, pos.y + height) - } - const shiftY = GROUP_CONTENT_PADDING - minY - const groupHeight = maxY - minY + GROUP_CONTENT_PADDING * 2 - - for (const id of groupChildIds) { - const pos = positions.get(id)! - positions.set(id, { x: pos.x, y: pos.y + shiftY }) - } - - return nodes.map((node) => { - if (node.id === group.id) { - return { - ...node, - style: { - ...node.style, - width: groupWidth, - height: groupHeight, - }, - } - } - if (node.parentId === group.id) { - const next = positions.get(node.id) - return next ? { ...node, position: next } : node - } - return node - }) -} diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.test.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.test.ts index f6e29999d..523ce0cf0 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.test.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.test.ts @@ -216,7 +216,7 @@ describe("layoutSlimTransportGraph (compact group)", () => { it("centers transport and custom rows with equal group margins", () => { const config = getGraphConfig("group_messaging") const group = config.nodes.find((node) => node.type === NODE_TYPES.GROUP)! - const groupWidth = group.style?.width as number + const groupWidth = group.width as number const transport = config.nodes.find( (node) => node.id === NODE_IDS.TRANSPORT, )! diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.ts index 0c351fb2c..36d6db307 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyLayout.ts @@ -163,10 +163,9 @@ function layoutCompactGroup(nodes: Node[], transport: Node): Node[] { return nodes.map((node) => { if (node.id === group.id) { - return { - ...node, - style: { ...node.style, width: groupWidth, height: groupHeight }, - } + // The group is hidden, so extent clamping resolves its box from top-level + // width/height (style is never measured). Write the fitted size there. + return { ...node, width: groupWidth, height: groupHeight } } if (node.parentId === group.id) { const next = positions.get(node.id) diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.discovery.test.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.discovery.test.tsx new file mode 100644 index 000000000..f3759b702 --- /dev/null +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.discovery.test.tsx @@ -0,0 +1,85 @@ +/** + * Copyright AGNTCY Contributors (https://github.com/agntcy) + * SPDX-License-Identifier: Apache-2.0 + **/ + +import { describe, expect, it } from "vitest" +import { NODE_TYPES } from "@/utils/const" +import { topologyWireToReactFlow } from "@/utils/topologyToReactFlow" +import type { CustomNodeData } from "@/components/MainArea/Graph/Elements/types" + +const SEED_RECRUITER = "node://4a000001-0001-4000-a001-000000000001" +const ANCHOR = "node://6f1d2c3a-0000-4000-8000-000000000001" +const DISCOVERED = "node://6f1d2c3a-0000-4000-8000-000000000002" +const DISCOVERED_SID = "agent://9a6cc736-d6fb-5a3e-82d6-3552d09b5ae9" + +/** + * The recruiter emits an anchor node sharing the seeded "Agentic Recruiter" + * label (no stable id) so it dedups onto the seed and the discovered-agent + * edge re-points there, plus the discovered node carrying its OASF record. + */ +describe("topologyWireToReactFlow discovery merge", () => { + const record = { name: "Brazil", url: "http://brazil:9000" } + const { nodes, edges } = topologyWireToReactFlow( + { + nodes: [ + { + id: SEED_RECRUITER, + type: "customNode", + label: "Agentic Recruiter", + layer_index: 0, + agent_record_uri: + "../../agents/supervisors/recruiter/oasf/agents/recruiter.json", + }, + { + id: ANCHOR, + type: "customNode", + label: "Agentic Recruiter", + layer_index: 0, + }, + { + id: DISCOVERED, + type: "customNode", + label: "Brazil", + layer_index: 1, + stable_agent_id: DISCOVERED_SID, + oasf_record: record, + agent_cid: "cidB", + }, + ], + edges: [ + { + id: "edge://discovery-1", + type: "custom", + source: ANCHOR, + target: DISCOVERED, + }, + ], + }, + { validateUrls: false }, + ) + + it("collapses the anchor onto the single seeded recruiter node", () => { + const recruiters = nodes.filter( + (n) => (n.data as unknown as CustomNodeData)?.label1 === "Agentic", + ) + expect(recruiters).toHaveLength(1) + expect(recruiters[0]?.id).toBe(SEED_RECRUITER) + }) + + it("renders the discovered node with inline OASF and a target handle", () => { + const discovered = nodes.find((n) => n.id === DISCOVERED_SID) + const data = discovered?.data as unknown as CustomNodeData | undefined + expect(discovered?.type).toBe(NODE_TYPES.CUSTOM) + expect(data?.oasfRecord).toEqual(record) + expect(data?.agentCid).toBe("cidB") + expect(data?.handles).toBe("target") + expect(data?.agentDirectoryLink).toBeDefined() + }) + + it("re-points the discovered edge from the seeded recruiter to the agent", () => { + expect(edges).toHaveLength(1) + expect(edges[0]?.source).toBe(SEED_RECRUITER) + expect(edges[0]?.target).toBe(DISCOVERED_SID) + }) +}) diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.test.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.test.tsx index b29b19f0a..08ac0520e 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.test.tsx +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.test.tsx @@ -7,7 +7,6 @@ import { describe, expect, it } from "vitest" import { NODE_TYPES, EDGE_TYPES, EDGE_LABELS } from "@/utils/const" import { topologyWireToReactFlow } from "@/utils/topologyToReactFlow" import { stableAgentUuidForRecordName } from "@/utils/agenticTopologyIdentityUiMap" - function wireNode(id: string, type: string, label: string, layerIndex: number) { return { id, type, label, layer_index: layerIndex } } @@ -180,7 +179,7 @@ describe("topologyWireToReactFlow", () => { expect(data?.label2).toBe("Farm Agent") }) - it("renders a group container with children parented and compact transport", () => { + it("keeps a hidden group container with children parented and compact transport", () => { const groupId = "node://aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" const { nodes } = topologyWireToReactFlow( { @@ -209,7 +208,8 @@ describe("topologyWireToReactFlow", () => { (node) => node.type === "customNode" && node.parentId === groupId, ) expect(group?.type).toBe("group") - expect(group?.style).toBeDefined() + expect(group?.hidden).toBe(true) + expect((group?.width ?? 0) > 0 && (group?.height ?? 0) > 0).toBe(true) expect(transport?.parentId).toBe(groupId) expect((transport?.data as { compact?: boolean })?.compact).toBe(true) expect(child).toBeDefined() diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.tsx index fe4e7d2dd..8a610895d 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.tsx +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.tsx @@ -26,6 +26,7 @@ import { layoutSlimTransportGraph, } from "@/utils/topologyLayout" import { + applyDiscoveredAgentInlineUi, enrichAgenticTopologyWellKnownUi, isDirectoryLabel, isMcpServerLabel, @@ -224,18 +225,17 @@ export function topologyWireToReactFlow( ) if (n.type === NODE_TYPES.GROUP) { + // The group is never drawn; it only exists so children can anchor to it + // via parentId/extent. width/height must live on the node (not style) so + // extent clamping still resolves the box while the node stays hidden. return { id: rfId, type: NODE_TYPES.GROUP, position, - data: { label: labelStr }, - style: { - width: GROUP_DEFAULT_WIDTH, - height: GROUP_DEFAULT_HEIGHT, - backgroundColor: "var(--group-background)", - border: "none", - borderRadius: "8px", - }, + hidden: true, + width: GROUP_DEFAULT_WIDTH, + height: GROUP_DEFAULT_HEIGHT, + data: {}, } } @@ -272,6 +272,7 @@ export function topologyWireToReactFlow( identityUiVariant: options.identityUiVariant, }) data = enrichAgenticTopologyWellKnownUi(data, n, { validateUrls }) + data = applyDiscoveredAgentInlineUi(data, n) if (data.directoryAgentSlug) { data = { ...data, diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/test_a2a_event_middleware.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/test_a2a_event_middleware.py index 36b164ad7..dac689e23 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/test_a2a_event_middleware.py +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/test_a2a_event_middleware.py @@ -382,6 +382,40 @@ async def test_missing_agent_card_is_noop( assert out_kwargs is http_kwargs assert captured_events == [] + async def test_edge_only_delegation_emits_anchors_without_transport( + self, agent_card_factory, otel_span, patch_emit_events, captured_events, + ): + """Recruiter delegation should anchor onto existing nodes, not add transport.""" + from common.stable_agent_id import stable_agent_id_for_name + from common.a2a_event_middleware.middleware import EventEmittingInterceptor + + patch_emit_events(True) + caller_card = MagicMock() + caller_card.name = "Recruiter Supervisor" + recruiter_sid = stable_agent_id_for_name("Agentic Recruiter agent") + interceptor = EventEmittingInterceptor( + caller_card=caller_card, + delegation_mode="edge_only", + delegation_anchor_stable_agent_id=recruiter_sid, + ) + remote = agent_card_factory("Brazil") + ctx = ClientCallContext(state={"tool": "delegate_task"}) + + with otel_span(trace_id=0x1, span_id=0x2, parent_span_id=0x3): + await interceptor.intercept( + "send_message", {}, {}, agent_card=remote, context=ctx, + ) + + [event] = captured_events + instance = _first_instance(event) + node_types = {n.type for n in instance.topology.nodes} + labels = {n.label for n in instance.topology.nodes} + assert "transportNode" not in node_types + assert "Recruiter Supervisor" not in labels + assert "Agentic Recruiter" in labels + assert "Brazil" in labels + assert len(instance.topology.edges) == 1 + # --------------------------------------------------------------------------- # make_event_emitting_consumer diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_discovery_layout.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_discovery_layout.py new file mode 100644 index 000000000..2a9e6ef20 --- /dev/null +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_discovery_layout.py @@ -0,0 +1,339 @@ +# Copyright AGNTCY Contributors (https://github.com/agntcy) +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for enrich_discovery_node_layout.""" + +from __future__ import annotations + +from typing import Callable, NamedTuple + +import pytest + +from schema.types import Data, Event + +from common.workflow_instance_store.discovery_layout import enrich_discovery_node_layout +from common.workflow_instance_store.merge import ( + merge_event_data, + reconcile_event_node_identities, +) + +INST = "instance://550e8400-e29b-41d4-a716-446655440003" +RECRUITER_RUNTIME_ID = "node://550e8400-e29b-41d4-a716-446655440100" +ANCHOR_ID = "node://550e8400-e29b-41d4-a716-446655440101" +DISCOVERED_ID = "node://550e8400-e29b-41d4-a716-446655440102" +DISCOVERED_ID_2 = "node://550e8400-e29b-41d4-a716-446655440103" +BLOCKER_ID = "node://550e8400-e29b-41d4-a716-446655440104" +RECRUITER_SID = "agent://550e8400-e29b-41d4-a716-446655440200" +DISCOVERED_SID = "agent://550e8400-e29b-41d4-a716-446655440202" +DISCOVERED_SID_2 = "agent://550e8400-e29b-41d4-a716-446655440203" +DISCOVERY_EDGE_ID = "edge://550e8400-e29b-41d4-a716-446655440300" +DISCOVERY_EDGE_ID_2 = "edge://550e8400-e29b-41d4-a716-446655440301" + +RECRUITER_POSITION = {"x": 400, "y": 300} +DIRECTORY_POSITION = {"x": 800, "y": 100} +BELOW_SLOT = {"x": 400, "y": 485} +RIGHT_SLOT = {"x": 685, "y": 300} + +_METADATA = { + "timestamp": "2026-01-01T00:00:00Z", + "schema_version": "1.0.0", + "correlation": {"id": "correlation://550e8400-e29b-41d4-a716-446655440001"}, + "id": "event://550e8400-e29b-41d4-a716-446655440002", + "type": "StateProgressUpdate", + "source": "test", +} + + +def _evt(data: dict) -> Event: + return Event.model_validate({"metadata": _METADATA, "data": data}) + + +def _dump(m: Data) -> dict: + return m.model_dump(mode="python") + + +def _recruiter_seed_node(*, with_position: bool = True) -> dict: + node = { + "id": RECRUITER_RUNTIME_ID, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": RECRUITER_SID, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440200", + } + if with_position: + node["position"] = dict(RECRUITER_POSITION) + return node + + +def _directory_seed_node() -> dict: + return { + "id": "node://550e8400-e29b-41d4-a716-446655440099", + "operation": "create", + "type": "customNode", + "label": "AGNTCY Agent Directory", + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "position": dict(DIRECTORY_POSITION), + } + + +def _discovered_node( + node_id: str, + stable_agent_id: str, + *, + label: str = "Brazil", + cid: str = "cidB", +) -> dict: + return { + "id": node_id, + "operation": "create", + "type": "customNode", + "label": label, + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "stable_agent_id": stable_agent_id, + "agent_record_uri": f"agent-card://{stable_agent_id.removeprefix('agent://')}", + "oasf_record": {"name": label}, + "agent_cid": cid, + } + + +def _discovery_topology_event( + discovered_nodes: list[dict], + edges: list[dict], +) -> Event: + anchor = { + "id": ANCHOR_ID, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": RECRUITER_SID, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440200", + } + return _evt( + { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + INST: { + "id": INST, + "topology": { + "nodes": [anchor, *discovered_nodes], + "edges": edges, + }, + } + }, + } + } + } + ) + + +def _discovery_event() -> Event: + discovered = _discovered_node(DISCOVERED_ID, DISCOVERED_SID) + edge = { + "id": DISCOVERY_EDGE_ID, + "operation": "create", + "type": "custom", + "source": ANCHOR_ID, + "target": DISCOVERED_ID, + "bidirectional": False, + "weight": 1.0, + } + return _discovery_topology_event([discovered], [edge]) + + +def _wf_with_nodes(nodes: list[dict]) -> dict: + return { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + INST: { + "id": INST, + "topology": {"nodes": nodes, "edges": []}, + } + }, + } + } + } + + +def _state_recruiter_in_instance(*, with_position: bool = True) -> Data: + return merge_event_data( + None, _evt(_wf_with_nodes([_recruiter_seed_node(with_position=with_position)])) + ) + + +def _state_recruiter_in_starting_only(*, with_position: bool = True) -> Data: + return Data.model_validate( + { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": { + "nodes": [_recruiter_seed_node(with_position=with_position)], + "edges": [], + }, + "instances": { + INST: { + "id": INST, + "topology": {"nodes": [], "edges": []}, + } + }, + } + } + } + ) + + +def _discovered_position(event: Event) -> dict | None: + topo = event.data.workflows["w"].instances[INST].topology + for node in topo.nodes or []: + if node.id.root == DISCOVERED_ID: + dumped = node.model_dump(mode="python") + pos = dumped.get("position") + return pos if isinstance(pos, dict) else None + return None + + +def _normalize_discovery(state: Data, event: Event) -> Event: + reconciled = reconcile_event_node_identities(state, event) + return enrich_discovery_node_layout(state, reconciled) + + +class LayoutStateCase(NamedTuple): + case_id: str + state_factory: Callable[..., Data] + + +_LAYOUT_STATE_CASES = ( + LayoutStateCase("recruiter_in_instance", _state_recruiter_in_instance), + LayoutStateCase( + "recruiter_in_starting_topology", _state_recruiter_in_starting_only + ), +) + + +@pytest.mark.parametrize("case", _LAYOUT_STATE_CASES, ids=lambda c: c.case_id) +def test_enrich_assigns_below_anchor_when_anchor_has_position(case: LayoutStateCase): + state = case.state_factory(with_position=True) + enriched = _normalize_discovery(state, _discovery_event()) + assert _discovered_position(enriched) == BELOW_SLOT + + +def test_enrich_skips_position_when_anchor_has_no_position(): + state = _state_recruiter_in_instance(with_position=False) + enriched = _normalize_discovery(state, _discovery_event()) + assert _discovered_position(enriched) is None + + +def test_enrich_uses_next_slot_when_below_is_blocked(): + blocker = { + "id": BLOCKER_ID, + "operation": "create", + "type": "customNode", + "label": "Blocker", + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "position": dict(BELOW_SLOT), + } + state = merge_event_data( + None, _evt(_wf_with_nodes([_recruiter_seed_node(), blocker])) + ) + enriched = _normalize_discovery(state, _discovery_event()) + assert _discovered_position(enriched) == RIGHT_SLOT + + +def test_enrich_assigns_distinct_positions_for_two_discoveries(): + state = _state_recruiter_in_instance(with_position=True) + discovered_a = _discovered_node(DISCOVERED_ID, DISCOVERED_SID, label="Brazil") + discovered_b = _discovered_node( + DISCOVERED_ID_2, + DISCOVERED_SID_2, + label="Colombia", + cid="cidC", + ) + edges = [ + { + "id": DISCOVERY_EDGE_ID, + "operation": "create", + "type": "custom", + "source": ANCHOR_ID, + "target": DISCOVERED_ID, + "bidirectional": False, + "weight": 1.0, + }, + { + "id": DISCOVERY_EDGE_ID_2, + "operation": "create", + "type": "custom", + "source": ANCHOR_ID, + "target": DISCOVERED_ID_2, + "bidirectional": False, + "weight": 1.0, + }, + ] + event = _discovery_topology_event([discovered_a, discovered_b], edges) + enriched = _normalize_discovery(state, event) + topo = enriched.data.workflows["w"].instances[INST].topology + positions = { + node.id.root: node.model_dump(mode="python").get("position") + for node in topo.nodes or [] + if node.id.root in {DISCOVERED_ID, DISCOVERED_ID_2} + } + assert positions[DISCOVERED_ID] == BELOW_SLOT + assert positions[DISCOVERED_ID_2] == RIGHT_SLOT + assert positions[DISCOVERED_ID] != positions[DISCOVERED_ID_2] + + +@pytest.mark.parametrize("case", _LAYOUT_STATE_CASES, ids=lambda c: c.case_id) +def test_reconcile_enrich_merge_retains_position(case: LayoutStateCase): + state = case.state_factory(with_position=True) + normalized = _normalize_discovery(state, _discovery_event()) + out = merge_event_data(state, normalized) + topo = _dump(out)["workflows"]["w"]["instances"][INST]["topology"] + discovered = next(n for n in topo["nodes"] if n["id"] == DISCOVERED_ID) + assert discovered["position"] == BELOW_SLOT + + +def test_enrich_idempotent_on_reemit(): + state = _state_recruiter_in_instance(with_position=True) + ev = _discovery_event() + first = _normalize_discovery(state, ev) + merged = merge_event_data(state, first) + second = _normalize_discovery(merged, ev) + assert _discovered_position(first) == _discovered_position(second) + + +def test_enrich_does_not_mutate_input_event(): + state = _state_recruiter_in_instance(with_position=True) + ev = _discovery_event() + enrich_discovery_node_layout(state, reconcile_event_node_identities(state, ev)) + assert _discovered_position(ev) is None + + +def test_enrich_with_directory_in_state_still_places_below_recruiter(): + state = merge_event_data( + None, + _evt(_wf_with_nodes([_recruiter_seed_node(), _directory_seed_node()])), + ) + enriched = _normalize_discovery(state, _discovery_event()) + assert _discovered_position(enriched) == BELOW_SLOT diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_merge.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_merge.py index 336267d5b..7b719f66d 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_merge.py +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_merge.py @@ -7,14 +7,21 @@ import copy +import pytest + from schema.types import Data, Event -from common.workflow_instance_store.merge import merge_event_data, merge_topology_delta +from common.workflow_instance_store.merge import ( + merge_event_data, + merge_topology_delta, + reconcile_event_node_identities, +) NODE_A = "node://550e8400-e29b-41d4-a716-446655440010" NODE_B = "node://550e8400-e29b-41d4-a716-446655440011" NODE_Z = "node://550e8400-e29b-41d4-a716-446655440099" INST = "instance://550e8400-e29b-41d4-a716-446655440003" +STABLE_AGENT = "agent://550e8400-e29b-41d4-a716-446655440020" _METADATA = { "timestamp": "2026-01-01T00:00:00Z", @@ -615,6 +622,64 @@ def test_merge_topology_delta_same_existing_twice_no_mutation_of_nested(): assert node1["meta"]["outer"]["inner"] == 2 +def _discovered_agent_node(**overrides) -> dict: + """Agent node carrying its OASF record + CID inline as extras (extra="allow").""" + node = { + "id": NODE_A, + "operation": "create", + "type": "customNode", + "label": "Brazil", + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "stable_agent_id": STABLE_AGENT, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440020", + "oasf_record": {"name": "Brazil", "url": "http://brazil:9000", "skills": ["x"]}, + "agent_cid": "cidB", + } + node.update(overrides) + return node + + +def _wf_with_nodes(nodes: list[dict]) -> dict: + return { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + INST: { + "id": INST, + "topology": {"nodes": nodes, "edges": []}, + } + }, + } + } + } + + +def test_merge_preserves_inline_oasf_record_on_create(): + """Discovered-agent extras (oasf_record, agent_cid) survive create round-trip.""" + node_in = _discovered_agent_node() + out = merge_event_data(None, _evt(_wf_with_nodes([node_in]))) + node = _dump(out)["workflows"]["w"]["instances"][INST]["topology"]["nodes"][0] + assert node["oasf_record"] == node_in["oasf_record"] + assert node["agent_cid"] == "cidB" + + +def test_update_preserves_inline_oasf_record_extras(): + """A later partial update that omits the extras must not drop them.""" + base = merge_event_data(None, _evt(_wf_with_nodes([_discovered_agent_node()]))) + update = {"id": NODE_A, "operation": "update", "label": "Brazil Coffee Farm"} + out = merge_event_data(base, _evt(_wf_with_nodes([update]))) + node = _dump(out)["workflows"]["w"]["instances"][INST]["topology"]["nodes"][0] + assert node["label"] == "Brazil Coffee Farm" + assert node["oasf_record"]["name"] == "Brazil" + assert node["agent_cid"] == "cidB" + + def test_merge_only_empty_workflows_and_extra_data(): ev = _evt({"workflows": {}, "app_state": {"counter": 1}}) out = merge_event_data(None, ev) @@ -623,6 +688,186 @@ def test_merge_only_empty_workflows_and_extra_data(): assert d["app_state"] == {"counter": 1} +# --------------------------------------------------------------------------- +# reconcile_event_node_identities (backend-authoritative anchoring) +# --------------------------------------------------------------------------- + +RECRUITER_RUNTIME_ID = "node://550e8400-e29b-41d4-a716-446655440100" +ANCHOR_ID = "node://550e8400-e29b-41d4-a716-446655440101" +DISCOVERED_ID = "node://550e8400-e29b-41d4-a716-446655440102" +RECRUITER_SID = "agent://550e8400-e29b-41d4-a716-446655440200" +DISCOVERED_SID = "agent://550e8400-e29b-41d4-a716-446655440202" +DISCOVERY_EDGE_ID = "edge://550e8400-e29b-41d4-a716-446655440300" + + +def _recruiter_seed_node() -> dict: + """Full agent node for the seeded recruiter (carries runtime id + sid).""" + return { + "id": RECRUITER_RUNTIME_ID, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": RECRUITER_SID, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440200", + } + + +def _discovery_event() -> Event: + """Anchor (recruiter sid) + discovered agent + anchor->discovered edge.""" + anchor = { + "id": ANCHOR_ID, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": RECRUITER_SID, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440200", + } + discovered = { + "id": DISCOVERED_ID, + "operation": "create", + "type": "customNode", + "label": "Brazil", + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "stable_agent_id": DISCOVERED_SID, + "agent_record_uri": "agent-card://550e8400-e29b-41d4-a716-446655440202", + "oasf_record": {"name": "Brazil"}, + "agent_cid": "cidB", + } + edge = { + "id": DISCOVERY_EDGE_ID, + "operation": "create", + "type": "custom", + "source": ANCHOR_ID, + "target": DISCOVERED_ID, + "bidirectional": False, + "weight": 1.0, + } + return _evt( + { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + INST: { + "id": INST, + "topology": { + "nodes": [anchor, discovered], + "edges": [edge], + }, + } + }, + } + } + } + ) + + +def _state_recruiter_in_instance() -> Data: + """Recruiter node lives in the merged instance topology.""" + return merge_event_data(None, _evt(_wf_with_nodes([_recruiter_seed_node()]))) + + +def _state_recruiter_in_starting_only() -> Data: + """Recruiter node only in starting_topology; instance topology still empty.""" + return Data.model_validate( + { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": { + "nodes": [_recruiter_seed_node()], + "edges": [], + }, + "instances": { + INST: { + "id": INST, + "topology": {"nodes": [], "edges": []}, + } + }, + } + } + } + ) + + +@pytest.mark.parametrize( + "description,state_factory", + [ + ("recruiter in instance topology", _state_recruiter_in_instance), + ("recruiter only in starting_topology", _state_recruiter_in_starting_only), + ], +) +def test_reconcile_drops_anchor_and_repoints_edge(description, state_factory): + normalized = reconcile_event_node_identities(state_factory(), _discovery_event()) + topo = normalized.data.workflows["w"].instances[INST].topology + node_ids = [n.id.root for n in topo.nodes] + assert ANCHOR_ID not in node_ids + assert node_ids == [DISCOVERED_ID] + assert len(topo.edges) == 1 + assert topo.edges[0].source.root == RECRUITER_RUNTIME_ID + assert topo.edges[0].target.root == DISCOVERED_ID + + +def test_reconcile_keeps_node_when_sid_absent_from_state(): + """A genuinely new node (no matching sid) is preserved untouched.""" + ev = _evt(_wf_with_nodes([_discovered_agent_node()])) + normalized = reconcile_event_node_identities(_state_recruiter_in_instance(), ev) + nodes = normalized.data.workflows["w"].instances[INST].topology.nodes + assert [n.id.root for n in nodes] == [NODE_A] + + +def test_reconcile_does_not_mutate_input_event(): + ev = _discovery_event() + reconcile_event_node_identities(_state_recruiter_in_instance(), ev) + original_ids = { + n.id.root for n in ev.data.workflows["w"].instances[INST].topology.nodes + } + assert original_ids == {ANCHOR_ID, DISCOVERED_ID} + assert ev.data.workflows["w"].instances[INST].topology.edges[0].source.root == ( + ANCHOR_ID + ) + + +def test_reconcile_then_merge_single_recruiter_no_duplicate_anchor(): + state = _state_recruiter_in_instance() + out = merge_event_data( + state, reconcile_event_node_identities(state, _discovery_event()) + ) + topo = _dump(out)["workflows"]["w"]["instances"][INST]["topology"] + node_ids = [n["id"] for n in topo["nodes"]] + assert node_ids.count(RECRUITER_RUNTIME_ID) == 1 + assert ANCHOR_ID not in node_ids + assert DISCOVERED_ID in node_ids + assert len(topo["edges"]) == 1 + assert topo["edges"][0]["source"] == RECRUITER_RUNTIME_ID + assert topo["edges"][0]["target"] == DISCOVERED_ID + + +def test_reconcile_then_merge_idempotent_on_reemit(): + state = _state_recruiter_in_instance() + ev = _discovery_event() + out1 = merge_event_data(state, reconcile_event_node_identities(state, ev)) + out2 = merge_event_data(out1, reconcile_event_node_identities(out1, ev)) + topo = _dump(out2)["workflows"]["w"]["instances"][INST]["topology"] + node_ids = [n["id"] for n in topo["nodes"]] + assert node_ids.count(DISCOVERED_ID) == 1 + assert ANCHOR_ID not in node_ids + assert len(topo["edges"]) == 1 + assert topo["edges"][0]["source"] == RECRUITER_RUNTIME_ID + + def test_merge_empty_workflows_preserves_existing_workflows_and_merges_extra(): base = merge_event_data( None, diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_store.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_store.py index cff3f00ea..7e170ea19 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_store.py +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/common/workflow_instance_store/test_store.py @@ -219,6 +219,159 @@ def listener(_ev: Event) -> None: store.close() +def test_dispatched_event_is_normalized_for_discovery_anchor(): + """SSE-bound event has the anchor dropped and edges repointed to the seed.""" + recruiter_runtime = "node://550e8400-e29b-41d4-a716-446655440100" + anchor = "node://550e8400-e29b-41d4-a716-446655440101" + discovered = "node://550e8400-e29b-41d4-a716-446655440102" + recruiter_sid = "agent://550e8400-e29b-41d4-a716-446655440200" + discovered_sid = "agent://550e8400-e29b-41d4-a716-446655440202" + edge_id = "edge://550e8400-e29b-41d4-a716-446655440300" + + store = WorkflowInstanceStateStore() + try: + seed = { + "metadata": { + "timestamp": "2026-01-01T00:00:00Z", + "schema_version": "1.0.0", + "correlation": { + "id": "correlation://550e8400-e29b-41d4-a716-446655440001" + }, + "id": "event://550e8400-e29b-41d4-a716-4466554400d0", + "type": "StateProgressUpdate", + "source": "test", + }, + "data": { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + _INSTANCE_KEY: { + "id": _INSTANCE_KEY, + "topology": { + "nodes": [ + { + "id": recruiter_runtime, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": recruiter_sid, + "agent_record_uri": ( + "agent-card://" + "550e8400-e29b-41d4-a716-446655440200" + ), + } + ], + "edges": [], + }, + } + }, + } + } + }, + } + store.submit_event_sync(seed) + store.wait_merge_idle() + store.wait_dispatch_idle() + + captured: list[Event] = [] + + def listener(ev: Event) -> None: + captured.append(ev) + + store.subscribe(_INSTANCE_KEY, listener) + + discovery = { + "metadata": { + "timestamp": "2026-01-01T00:00:01Z", + "schema_version": "1.0.0", + "correlation": { + "id": "correlation://550e8400-e29b-41d4-a716-446655440002" + }, + "id": "event://550e8400-e29b-41d4-a716-4466554400d1", + "type": "StateProgressUpdate", + "source": "recruiter_supervisor", + }, + "data": { + "workflows": { + "w": { + "name": "n", + "pattern": "p", + "use_case": "u", + "scenario": "s", + "starting_topology": {"nodes": [], "edges": []}, + "instances": { + _INSTANCE_KEY: { + "id": _INSTANCE_KEY, + "topology": { + "nodes": [ + { + "id": anchor, + "operation": "create", + "type": "customNode", + "label": "Agentic Recruiter", + "size": {"width": 1, "height": 1}, + "layer_index": 0, + "stable_agent_id": recruiter_sid, + "agent_record_uri": ( + "agent-card://" + "550e8400-e29b-41d4-a716-446655440200" + ), + }, + { + "id": discovered, + "operation": "create", + "type": "customNode", + "label": "Brazil", + "size": {"width": 1, "height": 1}, + "layer_index": 1, + "stable_agent_id": discovered_sid, + "agent_record_uri": ( + "agent-card://" + "550e8400-e29b-41d4-a716-446655440202" + ), + }, + ], + "edges": [ + { + "id": edge_id, + "operation": "create", + "type": "custom", + "source": anchor, + "target": discovered, + "bidirectional": False, + "weight": 1.0, + } + ], + }, + } + }, + } + } + }, + } + store.submit_event_sync(discovery) + store.wait_merge_idle() + store.wait_dispatch_idle() + + assert len(captured) == 1 + topo = captured[0].data.workflows["w"].instances[_INSTANCE_KEY].topology + node_ids = [n.id.root for n in topo.nodes] + assert anchor not in node_ids + assert discovered in node_ids + assert len(topo.edges) == 1 + assert topo.edges[0].source.root == recruiter_runtime + assert topo.edges[0].target.root == discovered + finally: + store.close() + + @pytest.mark.asyncio async def test_concurrent_submit_serializes_merges(): store = WorkflowInstanceStateStore() diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_dynamic_workflow_agent.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_dynamic_workflow_agent.py index a31212907..5dfb78e12 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_dynamic_workflow_agent.py +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_dynamic_workflow_agent.py @@ -7,8 +7,11 @@ import pytest +from agents.supervisors.recruiter import dynamic_workflow_agent as dwa from agents.supervisors.recruiter.dynamic_workflow_agent import ( DynamicWorkflowAgent, + _reachable_card, + _reachable_url, ) from agents.supervisors.recruiter.models import ( STATE_KEY_RECRUITED_AGENTS, @@ -26,6 +29,87 @@ def _make_ctx(state: dict | None = None): return ctx +class TestReachableUrl: + @pytest.mark.parametrize( + "case, url, host, expected", + [ + ( + "rewrites 0.0.0.0 keeping port", + "http://0.0.0.0:9999", + "host.docker.internal", + "http://host.docker.internal:9999", + ), + ( + "rewrites 0.0.0.0 to localhost for host run", + "http://0.0.0.0:9998", + "localhost", + "http://localhost:9998", + ), + ( + "rewrites 127.0.0.1", + "http://127.0.0.1:9997/a2a", + "host.docker.internal", + "http://host.docker.internal:9997/a2a", + ), + ( + "rewrites localhost", + "http://localhost:9999", + "host.docker.internal", + "http://host.docker.internal:9999", + ), + ( + "leaves routable host untouched", + "http://brazil-farm-server:9999", + "host.docker.internal", + "http://brazil-farm-server:9999", + ), + ( + "leaves non-http transport untouched", + "slim://0.0.0.0:46357", + "host.docker.internal", + "slim://0.0.0.0:46357", + ), + ( + "handles missing port", + "http://0.0.0.0", + "host.docker.internal", + "http://host.docker.internal", + ), + ( + "returns empty unchanged", + "", + "host.docker.internal", + "", + ), + ], + ) + def test_reachable_url(self, case, url, host, expected): + with patch.object(dwa, "DISCOVERED_AGENT_HOST", host): + assert _reachable_url(url) == expected, case + + +class TestReachableCard: + def test_reachable_card_does_not_mutate_original(self): + from a2a.types import AgentCard + + original = AgentCard( + name="Brazil Farm", + description="Brazil coffee farm agent", + url="http://0.0.0.0:9999", + version="1.0.0", + capabilities={}, + default_input_modes=["text"], + default_output_modes=["text"], + skills=[], + ) + with patch.object(dwa, "DISCOVERED_AGENT_HOST", "host.docker.internal"): + reachable = _reachable_card(original) + + assert original.url == "http://0.0.0.0:9999" + assert reachable.url == "http://host.docker.internal:9999" + assert reachable is not original + + class TestDynamicWorkflowAgentConstruction: def test_can_instantiate(self): agent = DynamicWorkflowAgent( diff --git a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_recruiter_client.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_recruiter_client.py index 1116be30a..3ba71be76 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_recruiter_client.py +++ b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_recruiter_client.py @@ -20,6 +20,9 @@ STATE_KEY_EVALUATION_RESULTS, STATE_KEY_RECRUITED_AGENTS, ) +from common.stable_agent_id import stable_agent_id_for_name +from common.workflow_context_prop import WorkflowContext +from common.workflow_utils.inflight import TraceContext # --------------------------------------------------------------------------- # _parse_dict_values @@ -345,3 +348,180 @@ async def fake_send_message(msg, context=None): assert "cid_task" in tool_context.state[STATE_KEY_RECRUITED_AGENTS] assert "Task completed" in result + + +# --------------------------------------------------------------------------- +# _emit_discovery_topology +# --------------------------------------------------------------------------- + + +_VALID_INSTANCE = "instance://11111111-1111-4111-8111-111111111111" +_NO_TRACE = TraceContext(trace_id=None, span_id=None, owner_span_id=None) + + +def _patch_discovery_context( + recruiter_client_mod, + *, + instance_id: str | None, + workflow_name: str | None, + workflow_known: bool = True, +): + """Build the patch stack shared by discovery-emit tests.""" + return [ + patch.object( + recruiter_client_mod, + "read_workflow_context", + return_value=WorkflowContext( + instance_id=instance_id, workflow_name=workflow_name + ), + ), + patch.object( + recruiter_client_mod, "read_trace_context", return_value=_NO_TRACE + ), + patch.object( + recruiter_client_mod, + "lookup_workflow", + return_value=object() if workflow_known else None, + ), + ] + + +class TestEmitDiscoveryTopology: + @pytest.mark.asyncio + async def test_emits_anchor_node_and_discovered_agents(self, recruiter_client_mod): + """A valid context emits an anchor node plus one node/edge per agent.""" + records = { + "cidB": {"name": "Brazil", "url": "http://brazil:9000"}, + "cidC": {"name": "Colombia", "url": "http://colombia:9000"}, + } + sink = AsyncMock() + build_event_mock = MagicMock(return_value=MagicMock()) + + ctx_patches = _patch_discovery_context( + recruiter_client_mod, + instance_id=_VALID_INSTANCE, + workflow_name="recruiter_pattern", + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client_mod, "build_event", build_event_mock + ), patch.object(recruiter_client_mod, "_discovery_event_sink", sink): + await recruiter_client_mod._emit_discovery_topology(records) + + sink.emit.assert_awaited_once() + topology = build_event_mock.call_args.kwargs["topology"] + labels = [node.label for node in topology.nodes] + assert labels[0] == "Agentic Recruiter" + assert "Brazil" in labels and "Colombia" in labels + # The anchor carries the seeded recruiter's stable_agent_id so the + # backend merge layer reconciles it onto the real recruiter node. + assert topology.nodes[0].stable_agent_id == stable_agent_id_for_name( + "Agentic Recruiter agent" + ) + # One edge per discovered agent, all sourced from the anchor node. + anchor_id = topology.nodes[0].id + assert len(topology.edges) == 2 + assert all(edge.source == anchor_id for edge in topology.edges) + # The inline agent record travels on the discovered node (``oasf_record``). + brazil = next(node for node in topology.nodes if node.label == "Brazil") + assert brazil.oasf_record == records["cidB"] + assert brazil.agent_cid == "cidB" + + @pytest.mark.asyncio + async def test_discovery_node_ids_are_deterministic(self, recruiter_client_mod): + """Re-discovering the same CID yields the same node id (idempotent merge).""" + records = {"cidB": {"name": "Brazil"}} + first = MagicMock(return_value=MagicMock()) + second = MagicMock(return_value=MagicMock()) + + for build_mock in (first, second): + ctx_patches = _patch_discovery_context( + recruiter_client_mod, + instance_id=_VALID_INSTANCE, + workflow_name="recruiter_pattern", + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client_mod, "build_event", build_mock + ), patch.object( + recruiter_client_mod, "_discovery_event_sink", AsyncMock() + ): + await recruiter_client_mod._emit_discovery_topology(records) + + first_node = first.call_args.kwargs["topology"].nodes[1].id + second_node = second.call_args.kwargs["topology"].nodes[1].id + assert first_node == second_node + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "description,records,instance_id,workflow_name,workflow_known", + [ + ("empty records", {}, _VALID_INSTANCE, "recruiter_pattern", True), + ( + "missing instance id", + {"cidB": {"name": "Brazil"}}, + None, + "recruiter_pattern", + True, + ), + ( + "invalid instance id", + {"cidB": {"name": "Brazil"}}, + "not-an-instance", + "recruiter_pattern", + True, + ), + ( + "unknown workflow", + {"cidB": {"name": "Brazil"}}, + _VALID_INSTANCE, + "ghost_workflow", + False, + ), + ( + "missing workflow name", + {"cidB": {"name": "Brazil"}}, + _VALID_INSTANCE, + None, + True, + ), + ], + ) + async def test_no_emit_when_context_incomplete( + self, + recruiter_client_mod, + description, + records, + instance_id, + workflow_name, + workflow_known, + ): + sink = AsyncMock() + build_event_mock = MagicMock() + ctx_patches = _patch_discovery_context( + recruiter_client_mod, + instance_id=instance_id, + workflow_name=workflow_name, + workflow_known=workflow_known, + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client_mod, "build_event", build_event_mock + ), patch.object(recruiter_client_mod, "_discovery_event_sink", sink): + await recruiter_client_mod._emit_discovery_topology(records) + + sink.emit.assert_not_awaited() + build_event_mock.assert_not_called() + + @pytest.mark.asyncio + async def test_no_emit_when_sink_disabled(self, recruiter_client_mod): + """With events disabled (_discovery_event_sink is None) nothing is emitted.""" + ctx_patches = _patch_discovery_context( + recruiter_client_mod, + instance_id=_VALID_INSTANCE, + workflow_name="recruiter_pattern", + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client_mod, "_discovery_event_sink", None + ): + await recruiter_client_mod._emit_discovery_topology( + {"cidB": {"name": "Brazil"}} + ) + # No exception means the guard short-circuited cleanly.