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 412cb6ade..01ba5ff7c 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 @@ -21,6 +21,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 +29,20 @@ 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, +) logger = logging.getLogger("lungo.recruiter.supervisor.dynamic_workflow") +# A2A middleware singletons that capture the delegation call to the selected +# agent (drawing its execution edges). The recruiter-service discovery hop in +# recruiter_client intentionally omits this so only the discovered agents show. +_event_interceptor = EventEmittingInterceptor(caller_card=RECRUITER_SUPERVISOR_CARD) +_event_consumer = make_event_emitting_consumer(caller_card=RECRUITER_SUPERVISOR_CARD) + class DynamicWorkflowAgent(BaseAgent): """Executes tasks by sending messages to selected recruited agents via A2A HTTP. 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..5676bc659 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 recruiter OASF 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,157 @@ 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 full OASF record inline) 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 +306,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 +326,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 +352,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 +475,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/common/a2a_event_middleware/middleware.py b/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py index c5ec72935..8979dffc0 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py +++ b/coffeeAGNTCY/coffee_agents/lungo/common/a2a_event_middleware/middleware.py @@ -10,7 +10,6 @@ from __future__ import annotations import logging -import re as _re from time import monotonic from typing import Any, Mapping @@ -18,6 +17,7 @@ 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: @@ -312,7 +318,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.", 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..c7f49dcf3 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,11 @@ WorkflowInstanceDataStore, WorkflowInstanceEventFanout, ) -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, +) 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 @@ -24,4 +28,5 @@ "WorkflowInstanceStoreClosedError", "merge_event_data", "merge_topology_delta", + "reconcile_event_node_identities", ] 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..cedfb612a 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,10 @@ 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.merge import ( + merge_event_data, + reconcile_event_node_identities, +) from common.workflow_instance_store.notifier import NoOpNotifier, NotifierProtocol EVENT_SCHEMA = "event_v1" @@ -138,9 +141,10 @@ 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) + 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/frontend/src/api/agenticWorkflowsTypes.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/api/agenticWorkflowsTypes.ts index 7281741fe..b41d6bb13 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 } + /** Full OASF record carried inline for agents discovered at runtime. */ + 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/ChatArea.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/ChatArea.tsx index 4ae31f402..7102248db 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/ChatArea.tsx +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/ChatArea.tsx @@ -3,7 +3,7 @@ * SPDX-License-Identifier: Apache-2.0 **/ -import React, { useCallback, useRef, useState } from "react" +import React, { useRef, useState } from "react" import { useScrollPanelOnContentResize } from "@/utils/chatScroll" @@ -25,11 +25,10 @@ import { LUNGO_FRONTEND_URLS, } from "@/urls" import type { GraphConfig } from "@/utils/graphConfigs" -import { DiscoveryResponseEvent } from "@/types/agent" import type { AuctionStreamingState } from "@/stores/auctionStreaming.types" import type { RecruiterStreamingState } from "@/stores/recruiterStreaming.types" import type { ApiResponse } from "@/types/api" -import { PATTERNS, usesStreamingChatSend } from "@/utils/patternUtils" +import { usesStreamingChatSend } from "@/utils/patternUtils" /** Panel expanded/collapsed by the chat header minimize control. */ export const CHAT_MESSAGE_PANEL_ID = "chat-message-panel" @@ -63,7 +62,6 @@ interface ChatAreaProps { auctionState?: AuctionStreamingState recruiterState?: RecruiterStreamingState grafanaUrl?: string - onDiscoveryResponse?: (evt: DiscoveryResponseEvent) => void } const ChatArea: React.FC = ({ @@ -95,7 +93,6 @@ const ChatArea: React.FC = ({ auctionState, recruiterState, grafanaUrl = getGrafanaUrl(), - onDiscoveryResponse, }) => { const [content, setContent] = useState("") const [loading, setLoading] = useState(false) @@ -110,22 +107,6 @@ const ChatArea: React.FC = ({ Boolean(currentUserMessage) && !isMinimized, ) - const onApiSuccess = useCallback( - (apiResponse: ApiResponse) => { - if (pattern !== PATTERNS.A2A_HTTP) { - return - } - - onDiscoveryResponse?.({ - response: apiResponse.response, - ts: Date.now(), - sessionId: apiResponse.session_id, - agent_records: apiResponse.agent_records, - }) - }, - [pattern, onDiscoveryResponse], - ) - const handleMinimize = () => { setIsMinimized(true) } @@ -159,8 +140,6 @@ const ChatArea: React.FC = ({ logger.debug("[ChatArea] API call successful, response:", response) - onApiSuccess(response) - if (onApiResponse) { onApiResponse(response.response ?? "", false) } 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 c640c07d6..29ba75da4 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 @@ -12,6 +12,8 @@ export interface CustomNodeData { active?: boolean selected?: boolean agentCid?: string + /** Inline OASF 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 a441968c1..406b29440 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 type { WorkflowSummary } from "@/utils/agenticWorkflowsApi" @@ -33,7 +31,6 @@ export interface MainAreaProps { 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 @@ -53,7 +50,6 @@ export function useMainArea({ isExpanded = false, groupCommResponseReceived = false, onNodeHighlight, - discoveryResponseEvent, selectedAgentCid, onLiveGraphConfig, }: MainAreaProps) { @@ -114,15 +110,6 @@ export function useMainArea({ logger.error("agentic-workflows/graph-session", { detail: agenticError }) }, [agenticError]) - useMainAreaDiscoveryGraph({ - pattern, - discoveryResponseEvent, - setNodes, - setEdges, - handleOpenIdentityModal, - handleOpenOasfModal, - }) - const nodeAgentCidKey = useMemo( () => nodes 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 7004ec158..86e14cf8c 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, @@ -63,7 +62,6 @@ const RootPage: React.FC = () => { handleClearConversation, handleNodeHighlightSetup, handleSenderHighlight, - handleDiscoveryResponse, graphConfig, events, status, @@ -184,7 +182,6 @@ const RootPage: React.FC = () => { isExpanded={isExpanded} groupCommResponseReceived={groupCommResponseReceived} onNodeHighlight={handleNodeHighlightSetup} - discoveryResponseEvent={discoveryResponseEvent} selectedAgentCid={ typeof recruiterSelectedAgent?.cid === "string" ? recruiterSelectedAgent.cid @@ -256,7 +253,6 @@ const RootPage: React.FC = () => { evaluationResults: recruiterEvaluationResults, selectedAgent: recruiterSelectedAgent, }} - onDiscoveryResponse={handleDiscoveryResponse} /> diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/stores/recruiterStreamingStore.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/stores/recruiterStreamingStore.ts index 43826ed72..409a6b4d6 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/stores/recruiterStreamingStore.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/stores/recruiterStreamingStore.ts @@ -34,7 +34,7 @@ interface RecruiterStreamingStoreState { agentRecords: Record | null evaluationResults: Record | null selectedAgent: Record | null - connect: (prompt: string) => Promise + connect: (prompt: string, workflowInstanceId?: string | null) => Promise disconnect: () => void reset: () => void } @@ -56,7 +56,7 @@ export const useRecruiterStreamingStore = create( (set) => ({ ...initialState, - connect: async (prompt: string) => { + connect: async (prompt: string, workflowInstanceId?: string | null) => { const abortController = new AbortController() set({ status: "connecting", @@ -78,7 +78,12 @@ export const useRecruiterStreamingStore = create( method: "POST", credentials: isLocalDev ? "omit" : "include", headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ prompt }), + body: JSON.stringify({ + prompt, + ...(workflowInstanceId + ? { workflow_instance_id: workflowInstanceId } + : {}), + }), signal: abortController.signal, }) diff --git a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts index d4bc7a587..92db2c7be 100644 --- a/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts +++ b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/useApp.ts @@ -12,7 +12,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, @@ -51,18 +50,11 @@ export function useApp() { const chat = useAppChatState({ selectedPattern }) 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) @@ -84,7 +76,6 @@ export function useApp() { setSelectedPattern(slug) setSelectedWorkflowSummary(summary) setLiveGraphConfig(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], @@ -189,21 +180,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 @@ -226,7 +202,6 @@ export function useApp() { chat.handleApiResponse, chat.setIsAgentLoading, chat.setShowFinalResponse, - handleDiscoveryResponse, ]) const handleSendPrompt = useCallback( @@ -267,7 +242,7 @@ export function useApp() { chat.setAgentResponse(undefined) streaming.resetRecruiter() try { - await streaming.connectRecruiter(query) + await streaming.connectRecruiter(query, activeWorkflowInstanceId) } catch (err) { logger.apiError(LUNGO_FRONTEND_URLS.apiPaths.agentPromptStream, err) chat.setShowFinalResponse(true) @@ -312,7 +287,6 @@ export function useApp() { chat.resetChatState() streaming.resetGroup() streaming.resetRecruiter() - lastDiscoveryKeyRef.current = null // eslint-disable-next-line react-hooks/exhaustive-deps -- stable reset fns only }, [chat.resetChatState, streaming.resetGroup, streaming.resetRecruiter]) @@ -388,7 +362,6 @@ export function useApp() { showRecruiterStreaming: chat.showRecruiterStreaming, showFinalResponse: chat.showFinalResponse, groupCommResponseReceived: chat.groupCommResponseReceived, - discoveryResponseEvent, handleUserInput: chat.handleUserInput, handleApiResponse: chat.handleApiResponse, handleSendPrompt, @@ -396,7 +369,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..e94453ee7 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 = { + icon: null, + label1: "Brazil", + label2: "", + handles: "all", + } as unknown as CustomNodeData + + 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..193bef8a9 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,29 @@ export function enrichAgenticTopologyWellKnownUi( return data } +/** + * Apply inline UI for runtime-discovered agents. The recruiter emits the full + * OASF record (and CID) on the wire node, so the OASF 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/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.tsx b/coffeeAGNTCY/coffee_agents/lungo/frontend/src/utils/topologyToReactFlow.tsx index 017834e6e..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, @@ -271,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/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_recruiter_client.py b/coffeeAGNTCY/coffee_agents/lungo/tests/unit/recruiter/test_recruiter_client.py index ee4013831..fa31f2f12 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 @@ -23,11 +23,16 @@ STATE_KEY_RECRUITED_AGENTS, RecruitmentResponse, ) +from agents.supervisors.recruiter import recruiter_client from agents.supervisors.recruiter.recruiter_client import ( + _emit_discovery_topology, _extract_parts, _parse_dict_values, recruit_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 # --------------------------------------------------------------------------- @@ -327,3 +332,160 @@ 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( + *, + 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, + "read_workflow_context", + return_value=WorkflowContext( + instance_id=instance_id, workflow_name=workflow_name + ), + ), + patch.object(recruiter_client, "read_trace_context", return_value=_NO_TRACE), + patch.object( + recruiter_client, + "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): + """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( + instance_id=_VALID_INSTANCE, workflow_name="recruiter_pattern" + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client, "build_event", build_event_mock + ), patch.object(recruiter_client, "_discovery_event_sink", sink): + await _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 full OASF record travels inline on the discovered node. + 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): + """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( + instance_id=_VALID_INSTANCE, workflow_name="recruiter_pattern" + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client, "build_event", build_mock + ), patch.object(recruiter_client, "_discovery_event_sink", AsyncMock()): + await _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, description, records, instance_id, workflow_name, workflow_known + ): + sink = AsyncMock() + build_event_mock = MagicMock() + ctx_patches = _patch_discovery_context( + 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, "build_event", build_event_mock + ), patch.object(recruiter_client, "_discovery_event_sink", sink): + await _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): + """With events disabled (_discovery_event_sink is None) nothing is emitted.""" + ctx_patches = _patch_discovery_context( + instance_id=_VALID_INSTANCE, workflow_name="recruiter_pattern" + ) + with ctx_patches[0], ctx_patches[1], ctx_patches[2], patch.object( + recruiter_client, "_discovery_event_sink", None + ): + await _emit_discovery_topology({"cidB": {"name": "Brazil"}}) + # No exception means the guard short-circuited cleanly.