(feat-redraw)Update animation handling

coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/dynamic_workflow_agent.py

@@ -60,7 +60,14 @@
 
 
 def _reachable_url(url: str) -> str:
-    """Rewrite an unroutable advertised host to a reachable one."""
+    """Rewrite an unroutable advertised host to a reachable one.
+
+    Discovered records advertise the agent's own bind address (e.g.
+    ``http://0.0.0.0:9999``), which the supervisor cannot dial. Replace such
+    hosts with DISCOVERED_AGENT_HOST while preserving scheme, port, and path.
+    Non-HTTP transports (slim://, nats://) and already-routable hosts are
+    returned unchanged.
+    """
     if not url:
         return url
     parts = urlsplit(url)
@@ -72,16 +79,6 @@ def _reachable_url(url: str) -> str:
     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.
 
@@ -111,14 +108,14 @@ async def _send_a2a_message(
             message[:100],
         )
 
-        # negotiate and create the client based on the card's preferred transport
-        client = await a2a_client_factory.create(
-            card,
-            interceptors=[_event_interceptor],
-            consumers=[_event_consumer],
-        )
-
         try:
+            # negotiate and create the client based on the card's preferred transport
+            client = await a2a_client_factory.create(
+                card,
+                interceptors=[_event_interceptor],
+                consumers=[_event_consumer],
+            )
+
             result_text = None
             async for response in client.send_message(a2a_message):
                 logger.info(
@@ -156,7 +153,7 @@ async def _send_a2a_message(
                 str(e),
                 exc_info=True,
             )
-            return f"Error communicating with {agent_name}: {str(e)}"
+            raise
 
     @staticmethod
     def _protocol_from_record(record: AgentRecord) -> AgentProtocol:
@@ -214,7 +211,7 @@ async def _run_async_impl(
             return
 
         try:
-            record = AgentRecord.from_record_data(selected_cid, record_data)
+            record = AgentRecord.from_record(selected_cid, record_data)
         except Exception:
             logger.warning(
                 f"Failed to parse agent record for CID {selected_cid}.",
@@ -259,20 +256,47 @@ async def _run_async_impl(
             )
             return
 
-        agent_card = _reachable_card(record.card)
+        agent_card = record.to_agent_card()
+        agent_card.url = _reachable_url(agent_card.url)
+        if agent_card.additional_interfaces:
+            for interface in agent_card.additional_interfaces:
+                interface.url = _reachable_url(interface.url)
 
         logger.info(
             "[agent:dynamic_workflow] Sending to agent: %s at %s",
             record.name,
             agent_card.url,
         )
 
-        # Send A2A message
-        response_text = await self._send_a2a_message(
-            card=agent_card,
-            message=task_message,
-            agent_name=record.name,
-        )
+        try:
+            response_text = await self._send_a2a_message(
+                card=agent_card,
+                message=task_message,
+                agent_name=record.name,
+            )
+        except Exception as e:
+            logger.error(
+                "[agent:dynamic_workflow] Delegation to %s failed: %s",
+                record.name,
+                str(e),
+                exc_info=True,
+            )
+            yield Event(
+                author=self.name,
+                invocation_id=ctx.invocation_id,
+                content=types.Content(
+                    role="model",
+                    parts=[
+                        types.Part(
+                            text=(
+                                f"Failed to delegate to agent '{record.name}': {e}"
+                            )
+                        )
+                    ],
+                ),
+            )
+            return
+
         combined = f"**{record.name}**:\n{response_text}"
 
         yield Event(

coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/models.py

@@ -4,10 +4,11 @@
 """Shared state keys and data models for the Recruiter Supervisor."""
 
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
-from a2a.types import AgentCard
-from pydantic import BaseModel
+from a2a.types import AgentCard, AgentCapabilities
+from agntcy_app_sdk.directory.oasf_converter import oasf_to_agent_card
+from pydantic import BaseModel, PrivateAttr
 
 
 class AgentProtocol(str, Enum):
@@ -25,21 +26,45 @@ class AgentProtocol(str, Enum):
 STATE_KEY_TASK_MESSAGE = "task_message"  # str: message to forward to selected agents
 STATE_KEY_SELECTED_AGENT = "selected_agent"  # str: CID of the currently selected agent
 
+
 # ---------------------------------------------------------------------------
 # Data models
 # ---------------------------------------------------------------------------
 
 
+def _card_has_transport_endpoint(card: AgentCard) -> bool:
+    """True when the card advertises at least one non-empty transport URL."""
+    if (card.url or "").strip():
+        return True
+    if card.additional_interfaces:
+        for interface in card.additional_interfaces:
+            if (interface.url or "").strip():
+                return True
+    return False
+
+
+def _require_delegation_card(card: AgentCard) -> None:
+    """Reject cards that cannot be used for A2A delegation."""
+    if not (card.name or "").strip():
+        raise ValueError("agent record is missing a name")
+    if not _card_has_transport_endpoint(card):
+        raise ValueError("agent record has no transport endpoint")
+
+
 class AgentRecord(BaseModel):
     """A recruited agent: its CID plus the full A2A AgentCard.
 
     The card is preserved verbatim so the client factory can negotiate transport.
+    The original record dict is retained for OASF-aware conversion when the card
+    is nested under ``modules[].data.card_data``.
     """
 
     cid: str
     card: AgentCard
     protocol: AgentProtocol = AgentProtocol.A2A
 
+    _raw: dict[str, Any] = PrivateAttr(default_factory=dict)
+
     @classmethod
     def from_record_data(cls, cid: str, data: dict) -> "AgentRecord":
         """Build from a stored record dict (a full A2A AgentCard)."""
@@ -49,7 +74,55 @@ def from_record_data(cls, cid: str, data: dict) -> "AgentRecord":
             if isinstance(raw_protocol, str)
             else AgentProtocol.A2A
         )
-        return cls(cid=cid, card=AgentCard.model_validate(data), protocol=protocol)
+        instance = cls(cid=cid, card=AgentCard.model_validate(data), protocol=protocol)
+        instance._raw = data
+        return instance
+
+    @classmethod
+    def from_record(cls, cid: str, record: dict[str, Any]) -> "AgentRecord":
+        """Parse a recruiter record dict, retaining the raw form for conversion."""
+        card = oasf_to_agent_card(record)
+        if card is None:
+            preferred = record.get("preferredTransport") or record.get(
+                "preferred_transport"
+            )
+            kwargs: dict[str, Any] = {
+                "name": record.get("name", ""),
+                "url": record.get("url", ""),
+                "description": record.get("description", ""),
+                "version": record.get("version", "1.0.0"),
+                "capabilities": AgentCapabilities(streaming=False),
+                "skills": [],
+                "defaultInputModes": ["text"],
+                "defaultOutputModes": ["text"],
+                "supportsAuthenticatedExtendedCard": False,
+            }
+            if preferred:
+                kwargs["preferred_transport"] = str(preferred)
+            card = AgentCard(**kwargs)
+
+        raw_protocol = record.get("protocol")
+        protocol = (
+            AgentProtocol(raw_protocol.lower())
+            if isinstance(raw_protocol, str)
+            else AgentProtocol.A2A
+        )
+        _require_delegation_card(card)
+        instance = cls(cid=cid, card=card, protocol=protocol)
+        instance._raw = record
+        return instance
+
+    def to_agent_card(self) -> AgentCard:
+        """Convert to an A2A AgentCard for delegation.
+
+        Recruited agents are stored as full OASF records with the card nested
+        under ``modules[].data.card_data``; prefer that so url and transport
+        survive. Fall back to the stored card for flat AgentCard records.
+        """
+        card = oasf_to_agent_card(self._raw)
+        if card is not None:
+            return card
+        return self.card.model_copy(deep=True)
 
     @property
     def name(self) -> str:

coffeeAGNTCY/coffee_agents/lungo/agents/supervisors/recruiter/shared.py

@@ -26,7 +26,9 @@ def get_factory() -> AgntcyFactory:
     include_nats=True,
 )
 
-# Discovered agents are reached over JSONRPC; httpx defaults to a 5s read timeout.
+# Discovered agents are reached over JSONRPC (their OASF records advertise it),
+# and httpx defaults to a 5s read timeout — too short for LLM-backed agents.
+# Only the JSONRPC transport consumes httpx_client; SLIM/NATS ignore it.
 config.httpx_client = httpx.AsyncClient(
     timeout=httpx.Timeout(A2A_CLIENT_TIMEOUT_SECONDS, connect=10.0)
 )

coffeeAGNTCY/coffee_agents/lungo/config/config.py

@@ -24,11 +24,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.
+# address (e.g. http://0.0.0.0:9999). Defaults to localhost for host runs
+# (the farm ports are published there); set to host.docker.internal inside
+# containers so the supervisor reaches the host-published ports.
 DISCOVERED_AGENT_HOST = os.getenv("DISCOVERED_AGENT_HOST", "localhost")
 
 # Read timeout (seconds) for the recruiter's JSONRPC delegation client.
+# Discovered agents are reached over JSONRPC, and httpx's 5s default is too
+# short for LLM-backed agents that take several seconds to respond.
 A2A_CLIENT_TIMEOUT_SECONDS = float(os.getenv("A2A_CLIENT_TIMEOUT_SECONDS", "30"))
 
 if os.getenv("SLIM_SHARED_SECRET") is None:

coffeeAGNTCY/coffee_agents/lungo/docker-compose.yaml

@@ -596,8 +596,11 @@ 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 farm records advertise http://0.0.0.0:<port>; reach them via
+      # the host gateway and their host-published ports from inside the network.
       - DISCOVERED_AGENT_HOST=${DISCOVERED_AGENT_HOST:-host.docker.internal}
     extra_hosts:
+      # Make host.docker.internal resolvable on Linux (no-op on Docker Desktop).
       - "host.docker.internal:host-gateway"
     ports:
       - "8882:8882"

coffeeAGNTCY/coffee_agents/lungo/frontend/src/components/Chat/chatStreamGraphHighlight.test.ts

@@ -15,6 +15,7 @@ import {
   animationSequenceStepIds,
   deriveAnimationSequenceFromGraph,
   resolveStreamAuthorToNodeId,
+  selectAnimationSequence,
 } from "./chatStreamGraphHighlight"
 
 function liveNode(
@@ -179,4 +180,56 @@ describe("deriveAnimationSequenceFromGraph", () => {
     )
     expect(seq).toEqual([])
   })
+
+  it("pulses fan-in / back-to-source edges without looping", () => {
+    const seq = deriveAnimationSequenceFromGraph(
+      [
+        liveNode("agent://auction", "customNode", {}),
+        liveNode("transport://slim", "transportNode", {}),
+        liveNode("agent://brazil", "customNode", {}),
+      ],
+      [
+        liveEdge("e-auction-transport", "agent://auction", "transport://slim"),
+        liveEdge("e-transport-brazil", "transport://slim", "agent://brazil"),
+        liveEdge("e-brazil-transport", "agent://brazil", "transport://slim"),
+      ],
+    )
+    expect(seq.map((step) => step.ids)).toEqual([
+      ["agent://auction"],
+      ["e-auction-transport"],
+      ["transport://slim"],
+      ["e-transport-brazil"],
+      ["agent://brazil"],
+      ["e-brazil-transport"],
+    ])
+  })
+})
+
+describe("selectAnimationSequence", () => {
+  const nodes: Node[] = [
+    liveNode("agent://recruiter", "customNode", {}),
+    liveNode("agent://directory", "customNode", {}),
+  ]
+  const edges: Edge[] = [
+    liveEdge("e-recruiter-directory", "agent://recruiter", "agent://directory"),
+  ]
+  const staticSequence = [{ ids: ["static-node"] }]
+
+  it("derives from the live graph in agentic mode", () => {
+    expect(
+      selectAnimationSequence(true, nodes, edges, staticSequence).map(
+        (step) => step.ids,
+      ),
+    ).toEqual([
+      ["agent://recruiter"],
+      ["e-recruiter-directory"],
+      ["agent://directory"],
+    ])
+  })
+
+  it("uses the authored static sequence outside agentic mode", () => {
+    expect(selectAnimationSequence(false, nodes, edges, staticSequence)).toBe(
+      staticSequence,
+    )
+  })
 })