From 47385fc85901f29604cbe031ccc5505880067114 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Fri, 31 Jul 2026 10:32:32 +0000 Subject: [PATCH 1/5] Fix sync-wait deadlock risks in agent-to-agent messaging Two agents that block on each other inside react() stall until their timeouts, and replies could be lost or misclassified. This commit governs the blocking wait primitives - register reply waiters before sending in send_and_wait, post_and_wait and Workspace.channels, closing the lost-wakeup race where a fast reply arrived before the waiter existed - correlate replies precisely via the existing Event.response_to field. The messaging mod now propagates response_to and requires_response across the direct-message relay and exposes message_id so receivers can reply with correlation. A counter-request from the peer is no longer mistaken for a reply - add AgentClient.expect_event and EventWaiter as the register-then-send primitive, and implement the previously missing AgentClient.wait_mod_message that channel wait helpers were calling (they always raised AttributeError and silently returned None) - match channel wait conditions against the notification events the messaging mod actually emits (thread.reply.notification, thread.channel_message.notification, thread.reaction.notification) - mark react() execution via a ContextVar in the runner and warn with DeprecationWarning when a blocking wait primitive is called inside react(). Setting OPENAGENTS_STRICT_NO_BLOCKING_WAIT=1 upgrades the warning to BlockingWaitInReactError. Agent-to-mod RPCs are unaffected - add WorkerAgent.reply_direct for correlated replies to direct messages Covered by tests/workspace/test_sync_wait_governance.py --- sdk/src/openagents/agents/runner.py | 4 +- sdk/src/openagents/agents/worker_agent.py | 29 ++ .../mods/workspace/messaging/mod.py | 8 + sdk/src/openagents/sdk/client.py | 95 ++++- sdk/src/openagents/sdk/react_context.py | 92 +++++ sdk/src/openagents/sdk/workspace.py | 338 ++++++++++------ tests/workspace/test_sync_wait_governance.py | 376 ++++++++++++++++++ 7 files changed, 794 insertions(+), 148 deletions(-) create mode 100644 sdk/src/openagents/sdk/react_context.py create mode 100644 tests/workspace/test_sync_wait_governance.py diff --git a/sdk/src/openagents/agents/runner.py b/sdk/src/openagents/agents/runner.py index 850415218..7ff6306e4 100644 --- a/sdk/src/openagents/agents/runner.py +++ b/sdk/src/openagents/agents/runner.py @@ -21,6 +21,7 @@ from openagents.models.event_context import EventContext from openagents.models.tool import AgentTool from openagents.core.client import AgentClient +from openagents.sdk.react_context import react_scope from openagents.utils.mod_loaders import load_mod_adapters from openagents.utils.verbose import verbose_print from openagents.models.event_response import EventResponse @@ -549,7 +550,8 @@ async def _async_loop(self): import time start_time = time.time() - await self.react(context) + with react_scope(self.agent_id): + await self.react(context) elapsed = time.time() - start_time print(f"✅ AGENT RESPONSE COMPLETED: {unprocessed_message.message_id[:8]}") diff --git a/sdk/src/openagents/agents/worker_agent.py b/sdk/src/openagents/agents/worker_agent.py index dfa4f0175..57b71d968 100644 --- a/sdk/src/openagents/agents/worker_agent.py +++ b/sdk/src/openagents/agents/worker_agent.py @@ -546,6 +546,35 @@ async def send_direct( agent_connection = self.workspace().agent(to) return await agent_connection.send(message_content, **kwargs) + async def reply_direct( + self, + context: EventContext, + text: str = None, + content: Dict[str, Any] = None, + **kwargs, + ) -> EventResponse: + """Reply to a received direct message with request/response correlation. + + Sets ``response_to`` to the incoming message's id so the sender's + ``send_and_wait`` (or any waiter matching on ``response_to``) can + recognize this message as the reply to its request. + + Args: + context: The EventContext of the direct message being replied to + text: Text content to send + content: Dict content to send (alternative to text) + **kwargs: Additional parameters + + Returns: + EventResponse: Response from the event system + """ + incoming = context.incoming_event + sender = (incoming.payload or {}).get("sender_id") or incoming.source_id + message_id = (incoming.payload or {}).get("message_id") or incoming.event_id + return await self.send_direct( + to=sender, text=text, content=content, response_to=message_id, **kwargs + ) + async def post_to_channel( self, channel: str, text: str = None, content: Dict[str, Any] = None, **kwargs ) -> EventResponse: diff --git a/sdk/src/openagents/mods/workspace/messaging/mod.py b/sdk/src/openagents/mods/workspace/messaging/mod.py index b086cff82..b9c1758f8 100644 --- a/sdk/src/openagents/mods/workspace/messaging/mod.py +++ b/sdk/src/openagents/mods/workspace/messaging/mod.py @@ -1339,12 +1339,20 @@ async def _process_direct_message(self, message: Event) -> None: original_payload = message.payload or {} notification_payload = original_payload.copy() notification_payload["sender_id"] = message.source_id + # Expose the message id so the receiver can correlate its reply + # (send(..., response_to=)). + notification_payload["message_id"] = message.event_id notification = EventModel( event_name="thread.direct_message.notification", source_id=message.source_id, # Keep original sender payload=notification_payload, destination_id=target_agent_id, + # Preserve request/response correlation across the relay so + # waiters can match replies precisely and tell counter-requests + # apart from replies. + response_to=message.response_to, + requires_response=message.requires_response, ) await self.network.process_event(notification) diff --git a/sdk/src/openagents/sdk/client.py b/sdk/src/openagents/sdk/client.py index 1f282306f..b83aca8e8 100644 --- a/sdk/src/openagents/sdk/client.py +++ b/sdk/src/openagents/sdk/client.py @@ -65,6 +65,52 @@ def __init__( self.result = result if result is not None else {} +class EventWaiter: + """A registered wait for a matching inbound event. + + The waiter is registered with the client the moment it is created, so it + can be set up *before* the request that provokes the reply is sent. This + closes the lost-wakeup race where a fast reply arrives between sending the + request and starting to wait, and is the building block for + request/response helpers such as ``send_and_wait``. + + Use as an async context manager to guarantee deregistration:: + + async with client.expect_event(condition) as waiter: + await client.send_event(request) + reply = await waiter.wait(timeout=30.0) + """ + + def __init__(self, client: "AgentClient", condition: Optional[Callable[[Event], bool]]): + self._client = client + self._entry = EventWaitingEntry( + event=asyncio.Event(), condition=condition, result={"event": None} + ) + client._event_waiters.append(self._entry) + + async def wait(self, timeout: float = 30.0) -> Optional[Event]: + """Wait for a matching event. Returns None on timeout.""" + try: + await asyncio.wait_for(self._entry.event.wait(), timeout=timeout) + return self._entry.result["event"] + except asyncio.TimeoutError: + logger.debug(f"Timeout waiting for event (timeout: {timeout}s)") + return None + finally: + self.cancel() + + def cancel(self) -> None: + """Deregister the waiter. Safe to call more than once.""" + if self._entry in self._client._event_waiters: + self._client._event_waiters.remove(self._entry) + + async def __aenter__(self) -> "EventWaiter": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + self.cancel() + + class AgentClient: """Core client implementation for OpenAgents. @@ -834,6 +880,20 @@ async def wait_event( # Wait for any event that matches the condition return await self._wait_for_message(condition, timeout) + def expect_event( + self, condition: Optional[Callable[[Event], bool]] = None + ) -> EventWaiter: + """Register a waiter for a matching inbound event without blocking. + + Register the waiter *before* sending the request that provokes the + reply, then await ``waiter.wait(timeout)`` — this way a reply that + arrives while the request is still being sent is not lost. + + Returns: + EventWaiter: The registered waiter (also an async context manager) + """ + return EventWaiter(self, condition) + async def _wait_for_message( self, condition: Optional[Callable] = None, timeout: float = 30.0 ) -> Optional[Event]: @@ -850,28 +910,25 @@ async def _wait_for_message( logger.warning(f"Agent {self.agent_id} is not connected to a network") return None - # Create event and waiter entry - event_waiter = asyncio.Event() - result_event = {"event": None} + return await EventWaiter(self, condition).wait(timeout) - waiter_entry = EventWaitingEntry( - event=event_waiter, condition=condition, result=result_event - ) + async def wait_mod_message( + self, + condition: Optional[Callable[[Event], bool]] = None, + timeout: float = 30.0, + ) -> Optional[Event]: + """Wait for a mod-generated event that matches the given condition. - # Add to event waiters list - self._event_waiters.append(waiter_entry) + Alias of :meth:`wait_event`, kept for the workspace channel helpers. - try: - # Wait for any event with timeout - await asyncio.wait_for(event_waiter.wait(), timeout=timeout) - return result_event["event"] - except asyncio.TimeoutError: - logger.debug(f"Timeout waiting for event (timeout: {timeout}s)") - return None - finally: - # Clean up - remove waiter from list - if waiter_entry in self._event_waiters: - self._event_waiters.remove(waiter_entry) + Args: + condition: Optional function to filter events + timeout: Maximum time to wait in seconds + + Returns: + Event if found within timeout, None otherwise + """ + return await self._wait_for_message(condition, timeout) async def _notify_event_waiters(self, event: Event) -> None: """Notify all waiters that match the given event. diff --git a/sdk/src/openagents/sdk/react_context.py b/sdk/src/openagents/sdk/react_context.py new file mode 100644 index 000000000..56e646fe7 --- /dev/null +++ b/sdk/src/openagents/sdk/react_context.py @@ -0,0 +1,92 @@ +"""Tracking of the agent react() execution context. + +The agent runner processes inbound events strictly one at a time: the next +event is not dequeued until the current react() call returns. Any primitive +that blocks inside react() waiting for another agent's reply therefore stalls +every other inbound event for that agent — and if the peer agent blocks the +same way waiting on us, both agents stall until their timeouts expire +(a bounded deadlock). The safe pattern is to send without waiting and handle +the reply as a new event in the next react() round. + +This module lets blocking agent-to-agent wait primitives detect that they are +being called from inside react() and warn about it. Setting the environment +variable ``OPENAGENTS_STRICT_NO_BLOCKING_WAIT=1`` upgrades the warning to a +``BlockingWaitInReactError``. + +Agent-to-mod request/response calls (e.g. project management RPCs) are not +affected: the mod runs on the network server and never blocks waiting on the +calling agent, so no wait cycle can form. +""" + +import logging +import os +import warnings +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Iterator, Optional + +logger = logging.getLogger(__name__) + +# Agent id of the agent currently executing react(), if any. Set by the agent +# runner around each react() call; propagates into everything react() awaits. +current_react_agent: ContextVar[Optional[str]] = ContextVar( + "current_react_agent", default=None +) + +STRICT_ENV_VAR = "OPENAGENTS_STRICT_NO_BLOCKING_WAIT" + + +class BlockingWaitInReactError(RuntimeError): + """Raised in strict mode when a blocking wait primitive runs inside react().""" + + +def in_react() -> bool: + """Return True if the current coroutine is executing inside a react() call.""" + return current_react_agent.get() is not None + + +def strict_mode_enabled() -> bool: + return os.environ.get(STRICT_ENV_VAR, "").strip().lower() in ( + "1", + "true", + "yes", + "on", + ) + + +@contextmanager +def react_scope(agent_id: str) -> Iterator[None]: + """Mark the enclosed block as running inside the given agent's react().""" + token = current_react_agent.set(agent_id) + try: + yield + finally: + current_react_agent.reset(token) + + +def check_blocking_wait(api_name: str, target: Optional[str] = None) -> None: + """Warn (or raise, in strict mode) when a blocking agent-to-agent wait + primitive is invoked from inside react(). + + Args: + api_name: Name of the blocking API, for the diagnostic message + target: Optional peer agent / channel the caller is waiting on + """ + agent_id = current_react_agent.get() + if agent_id is None: + return + + target_desc = f" on {target!r}" if target else "" + message = ( + f"{api_name}{target_desc} was called inside react() of agent " + f"{agent_id!r}. The runner processes one event at a time, so blocking " + f"here stalls all other inbound events for this agent; if the peer " + f"blocks the same way waiting on you, both agents deadlock until " + f"their timeouts. Send without waiting and handle the reply as a new " + f"event in the next react() instead. Set {STRICT_ENV_VAR}=1 to turn " + f"this warning into an error." + ) + if strict_mode_enabled(): + raise BlockingWaitInReactError(message) + warnings.warn(message, DeprecationWarning, stacklevel=3) + logger.warning(message) diff --git a/sdk/src/openagents/sdk/workspace.py b/sdk/src/openagents/sdk/workspace.py index a1e0926b2..42c9df77b 100644 --- a/sdk/src/openagents/sdk/workspace.py +++ b/sdk/src/openagents/sdk/workspace.py @@ -13,6 +13,7 @@ from datetime import datetime from openagents.sdk.client import AgentClient +from openagents.sdk.react_context import check_blocking_wait from openagents.models.event import Event from openagents.models.event_response import EventResponse from openagents.models.messages import EventNames @@ -39,14 +40,41 @@ def __init__(self, agent_id: str, workspace: "Workspace"): self.workspace = workspace self._client = workspace.client + def _build_direct_message( + self, content: Union[str, Dict[str, Any]], **kwargs + ) -> Event: + """Build the thread.direct_message.send event for this agent.""" + if isinstance(content, str): + message_content = {"text": content} + else: + message_content = content.copy() + + return Event( + event_name="thread.direct_message.send", + source_id=self._client.agent_id, + destination_id=self.agent_id, + payload={ + "target_agent_id": self.agent_id, + "message_type": "direct_message", + "content": message_content, # Nested structure like channel messages + }, + relevant_mod=WORKSPACE_MESSAGING_MOD_NAME, + **kwargs, + ) + async def send( self, content: Union[str, Dict[str, Any]], **kwargs ) -> EventResponse: """Send a direct message to this agent. + To reply to a message received from this agent, pass + ``response_to=`` so the peer can + correlate the reply with its request (e.g. in ``send_and_wait``). + Args: content: Message content (string or dict) - **kwargs: Additional message parameters + **kwargs: Additional message parameters (passed to the Event, + e.g. ``response_to``) Returns: EventResponse: Response from the event system @@ -59,25 +87,7 @@ async def send( ) try: - # Prepare message content - if isinstance(content, str): - message_content = {"text": content} - else: - message_content = content.copy() - - # Create direct message event for thread messaging with nested structure - direct_message = Event( - event_name="thread.direct_message.send", - source_id=self._client.agent_id, - destination_id=self.agent_id, - payload={ - "target_agent_id": self.agent_id, - "message_type": "direct_message", - "content": message_content, # Nested structure like channel messages - }, - relevant_mod=WORKSPACE_MESSAGING_MOD_NAME, - **kwargs, - ) + direct_message = self._build_direct_message(content, **kwargs) # Send through client and get immediate response response = await self.workspace.send_event(direct_message) @@ -117,12 +127,17 @@ async def get_agent_info(self) -> Optional[Dict[str, Any]]: async def wait_for_message(self, timeout: float = 30.0) -> Optional[Dict[str, Any]]: """Wait for a direct message from this agent. + Blocking: do not call inside react() — handle the message in the next + react() round instead. + Args: timeout: Timeout in seconds Returns: Dict containing the message content, or None if timeout """ + check_blocking_wait("AgentConnection.wait_for_message", self.agent_id) + # Ensure we're connected to the client if not await self.workspace._ensure_connected(): logger.error("Could not establish client connection") @@ -133,7 +148,10 @@ async def wait_for_message(self, timeout: float = 30.0) -> Optional[Dict[str, An def message_condition(msg): """Check if this is a direct message from our target agent.""" try: - return msg.source_id == self.agent_id + return ( + msg.event_name == "thread.direct_message.notification" + and msg.source_id == self.agent_id + ) except (AttributeError, KeyError): return False @@ -153,6 +171,10 @@ def message_condition(msg): async def wait_for_reply(self, timeout: float = 30.0) -> Optional[Dict[str, Any]]: """Wait for a reply from this agent (alias for wait_for_message). + Note: this matches any direct message from the agent. For precise + request/reply correlation use ``send_and_wait``, which matches on the + reply's ``response_to`` field. + Args: timeout: Timeout in seconds @@ -164,7 +186,20 @@ async def wait_for_reply(self, timeout: float = 30.0) -> Optional[Dict[str, Any] async def send_and_wait( self, content: Union[str, Dict[str, Any]], timeout: float = 30.0, **kwargs ) -> Optional[Dict[str, Any]]: - """Send a direct message and wait for a reply. + """Send a direct message and wait for the reply to it. + + The reply waiter is registered before the message is sent, so a fast + reply cannot be lost. A reply is recognized by ``response_to`` matching + the id of the sent message (the receiving agent replies with + ``send(content, response_to=)``; the message_id is + delivered in the notification payload). For peers that do not set + ``response_to``, a plain direct message from the target agent is + accepted as a fallback — unless that message itself demands a response + (``requires_response``), which indicates a counter-request rather than + a reply. + + Blocking: do not call inside react() — send() and handle the reply in + the next react() round instead. Args: content: Message content to send @@ -174,15 +209,44 @@ async def send_and_wait( Returns: Dict containing the reply message content, or None if timeout or send failed """ - # Send the message first - success = await self.send(content, **kwargs) - if not success: - logger.error(f"Failed to send message to agent {self.agent_id}") + check_blocking_wait("AgentConnection.send_and_wait", self.agent_id) + + # Ensure we're connected to the client + if not await self.workspace._ensure_connected(): + logger.error("Could not establish client connection") return None - # Wait for reply - logger.info(f"Sent message to {self.agent_id}, waiting for reply...") - return await self.wait_for_reply(timeout=timeout) + request = self._build_direct_message( + content, requires_response=True, **kwargs + ) + + def reply_condition(msg): + try: + if msg.event_name != "thread.direct_message.notification": + return False + if msg.source_id != self.agent_id: + return False + if msg.response_to: + return msg.response_to == request.event_id + # Legacy peer that does not correlate replies: accept any + # plain message, but never one that is itself a request. + return not msg.requires_response + except (AttributeError, KeyError): + return False + + # Register the waiter before sending to avoid the lost-wakeup race. + async with self._client.expect_event(reply_condition) as waiter: + response = await self.workspace.send_event(request) + if not response or not response.success: + logger.error(f"Failed to send message to agent {self.agent_id}") + return None + + logger.info(f"Sent message to {self.agent_id}, waiting for reply...") + reply = await waiter.wait(timeout=timeout) + + if reply: + return reply.payload + return None def __str__(self) -> str: return f"AgentConnection({self.agent_id})" @@ -231,28 +295,7 @@ async def post( ) try: - # Prepare message content - if isinstance(content, str): - message_content = {"text": content} - else: - message_content = content.copy() - - # Create mod message for thread messaging - mod_message = Event( - event_name="thread.channel_message.post", - source_id=self._client.agent_id, - relevant_mod=WORKSPACE_MESSAGING_MOD_NAME, - destination_id=f"channel:{self.name.lstrip('#')}", # Proper channel destination - payload={ - "action": "channel_message", - "message_type": "channel_message", - "channel": ( - self.name.lstrip("#") if self.name else self.name - ), # Normalize channel name - "content": message_content, - **kwargs, - }, - ) + mod_message = self._build_post(content, **kwargs) # Send through workspace and get immediate response response = await self.workspace.send_event(mod_message) @@ -264,6 +307,29 @@ async def post( success=False, message=f"Failed to send message: {str(e)}" ) + def _build_post(self, content: Union[str, Dict[str, Any]], **kwargs) -> Event: + """Build the thread.channel_message.post event for this channel.""" + if isinstance(content, str): + message_content = {"text": content} + else: + message_content = content.copy() + + return Event( + event_name="thread.channel_message.post", + source_id=self._client.agent_id, + relevant_mod=WORKSPACE_MESSAGING_MOD_NAME, + destination_id=f"channel:{self.name.lstrip('#')}", # Proper channel destination + payload={ + "action": "channel_message", + "message_type": "channel_message", + "channel": ( + self.name.lstrip("#") if self.name else self.name + ), # Normalize channel name + "content": message_content, + **kwargs, + }, + ) + async def post_with_mention( self, content: Union[str, Dict[str, Any]], mention_agent_id: str, **kwargs ) -> bool: @@ -527,33 +593,15 @@ async def wait_for_reply( Returns: Dict containing the reply message, or None if timeout """ + check_blocking_wait("ChannelConnection.wait_for_reply", self.name) + # Ensure we're connected to the client if not await self.workspace._ensure_connected(): logger.error("Could not establish client connection") return None try: - - def reply_condition(msg): - """Check if this is a reply message in our channel.""" - try: - content = msg.payload - # Look for reply messages from thread messaging mod - if content.get("action") == "channel_message_notification": - msg_data = content.get("message", {}) - # Check if it's in our channel - if msg_data.get("channel") != self.name.lstrip("#"): - return False - # Check if it's a reply (has reply_to_id) - if not msg_data.get("reply_to_id"): - return False - # If specific message_id provided, check if it matches - if message_id and msg_data.get("reply_to_id") != message_id: - return False - return True - return False - except (AttributeError, KeyError): - return False + reply_condition = self._reply_condition(message_id) # Wait for the reply response = await self._client.wait_mod_message( @@ -561,13 +609,33 @@ def reply_condition(msg): ) if response: - return response.payload.get("message", {}) + return response.payload return None except Exception as e: logger.error(f"Error waiting for reply in channel {self.name}: {e}") return None + def _reply_condition(self, message_id: Optional[str]): + """Build a condition matching reply notifications in this channel.""" + + def reply_condition(msg): + try: + # Reply notifications are delivered as thread.reply.notification + # with the reply payload flattened at the top level. + if msg.event_name != "thread.reply.notification": + return False + content = msg.payload + if content.get("channel") != self.name.lstrip("#"): + return False + if message_id and content.get("reply_to_id") != message_id: + return False + return True + except (AttributeError, KeyError): + return False + + return reply_condition + async def wait_for_post( self, from_agent: Optional[str] = None, timeout: float = 30.0 ) -> Optional[Dict[str, Any]]: @@ -580,6 +648,8 @@ async def wait_for_post( Returns: Dict containing the post message, or None if timeout """ + check_blocking_wait("ChannelConnection.wait_for_post", self.name) + # Ensure we're connected to the client if not await self.workspace._ensure_connected(): logger.error("Could not establish client connection") @@ -590,24 +660,21 @@ async def wait_for_post( def post_condition(msg): """Check if this is a new post (not reply) in our channel.""" try: + # Channel posts are delivered as + # thread.channel_message.notification with the post payload + # flattened at the top level. + if msg.event_name != "thread.channel_message.notification": + return False content = msg.payload - # Look for channel messages from thread messaging mod - if content.get("action") == "channel_message_notification": - msg_data = content.get("message", {}) - # Check if it's in our channel - if msg_data.get("channel") != self.name.lstrip("#"): - return False - # Check if it's NOT a reply (no reply_to_id) - if msg_data.get("reply_to_id"): - return False - # If specific agent provided, check sender - if from_agent and msg_data.get("sender_id") != from_agent: - return False - # Don't wait for our own messages - if msg_data.get("sender_id") == self._client.agent_id: - return False - return True - return False + if content.get("channel") != self.name.lstrip("#"): + return False + # If specific agent provided, check sender + if from_agent and msg.source_id != from_agent: + return False + # Don't wait for our own messages + if msg.source_id == self._client.agent_id: + return False + return True except (AttributeError, KeyError): return False @@ -617,7 +684,7 @@ def post_condition(msg): ) if response: - return response.payload.get("message", {}) + return response.payload return None except Exception as e: @@ -636,6 +703,8 @@ async def wait_for_reaction( Returns: Dict containing the reaction info, or None if timeout """ + check_blocking_wait("ChannelConnection.wait_for_reaction", self.name) + # Ensure we're connected to the client if not await self.workspace._ensure_connected(): logger.error("Could not establish client connection") @@ -646,11 +715,11 @@ async def wait_for_reaction( def reaction_condition(msg): """Check if this is a reaction to our message.""" try: - content = msg.payload - # Look for reaction notifications from thread messaging mod - if content.get("action") == "reaction_notification": - return content.get("target_message_id") == message_id - return False + # Reactions are delivered as thread.reaction.notification + return ( + msg.event_name == "thread.reaction.notification" + and msg.payload.get("target_message_id") == message_id + ) except (AttributeError, KeyError): return False @@ -670,7 +739,14 @@ def reaction_condition(msg): async def post_and_wait( self, content: Union[str, Dict[str, Any]], timeout: float = 30.0, **kwargs ) -> Optional[Dict[str, Any]]: - """Post a message and wait for any reply to it. + """Post a message and wait for a reply to that specific message. + + The reply waiter is registered before the message is posted, so a fast + reply cannot be lost, and only replies whose ``reply_to_id`` references + the posted message are accepted. + + Blocking: do not call inside react() — post() and handle the reply in + the next react() round instead. Args: content: Message content to post @@ -680,16 +756,32 @@ async def post_and_wait( Returns: Dict containing the reply message, or None if timeout or post failed """ - # Post the message first - success = await self.post(content, **kwargs) - if not success: - logger.error(f"Failed to post message to {self.name}") + check_blocking_wait("ChannelConnection.post_and_wait", self.name) + + # Ensure we're connected to the client + if not await self.workspace._ensure_connected(): + logger.error("Could not establish client connection") return None - # For demo purposes, we'll wait for any reply since we don't have the actual message ID - # In a real implementation, the post method would return the message ID - logger.info(f"Posted message to {self.name}, waiting for replies...") - return await self.wait_for_reply(timeout=timeout) + post_event = self._build_post(content, **kwargs) + + # The post's event_id is the message_id repliers reference via + # reply_to_id. Register the waiter before posting to avoid the + # lost-wakeup race. + async with self._client.expect_event( + self._reply_condition(post_event.event_id) + ) as waiter: + response = await self.workspace.send_event(post_event) + if not response or not response.success: + logger.error(f"Failed to post message to {self.name}") + return None + + logger.info(f"Posted message to {self.name}, waiting for replies...") + reply = await waiter.wait(timeout=timeout) + + if reply: + return reply.payload + return None def __str__(self) -> str: return f"ChannelConnection({self.name})" @@ -1017,30 +1109,20 @@ def response_condition(msg): except (AttributeError, KeyError): return False - # Start waiting for response before sending request - wait_task = asyncio.create_task( - self._client.wait_mod_message( - condition=response_condition, timeout=timeout - ) - ) - - # Give the wait task a moment to start - await asyncio.sleep(0.01) - - # Send request - success = await self.send_event(mod_message) - if not success: - wait_task.cancel() - logger.error("Failed to send list_channels request") - # Return cached or default channels as fallback - return ( - list(self._channels_cache.keys()) - if self._channels_cache - else DEFAULT_CHANNELS - ) + # Register the waiter before sending so the response cannot be lost + async with self._client.expect_event(response_condition) as waiter: + success = await self.send_event(mod_message) + if not success: + logger.error("Failed to send list_channels request") + # Return cached or default channels as fallback + return ( + list(self._channels_cache.keys()) + if self._channels_cache + else DEFAULT_CHANNELS + ) - # Wait for response - response = await wait_task + # Wait for response + response = await waiter.wait(timeout=timeout) if response is None: logger.warning( diff --git a/tests/workspace/test_sync_wait_governance.py b/tests/workspace/test_sync_wait_governance.py new file mode 100644 index 000000000..e77d87077 --- /dev/null +++ b/tests/workspace/test_sync_wait_governance.py @@ -0,0 +1,376 @@ +"""Tests for synchronous-wait governance in agent-to-agent messaging. + +Covers the failure modes behind "two agents waiting on each other": + +1. Lost wakeup: a reply arriving before the waiter was registered was lost + (send_and_wait used to send first and register the waiter afterwards). +2. Misclassification: a counter-request from the peer used to be mistaken for + the reply because waiters matched on source_id only. +3. Concurrent waiters: two in-flight requests to the same peer used to race + for whichever message came back first. +4. Blocking waits inside react(): the runner processes one event at a time, so + blocking in react() stalls the agent; primitives now warn (or raise in + strict mode) when called from a react() scope. +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from openagents.models.event import Event +from openagents.models.event_response import EventResponse +from openagents.sdk.client import AgentClient +from openagents.sdk.react_context import ( + BlockingWaitInReactError, + STRICT_ENV_VAR, + current_react_agent, + react_scope, +) +from openagents.sdk.workspace import Workspace + + +AGENT_A = "agent-alice" +AGENT_B = "agent-bob" + + +def make_workspace(agent_id: str = AGENT_A) -> Workspace: + """Build a workspace whose client is 'connected' but sends nothing.""" + client = AgentClient(agent_id=agent_id) + client.connector = MagicMock() # non-None so wait primitives don't bail out + workspace = Workspace(client) + workspace._ensure_connected = AsyncMock(return_value=True) + return workspace + + +def dm_notification( + sender: str, + text: str, + response_to: str = None, + requires_response: bool = False, +) -> Event: + """Build a thread.direct_message.notification as the messaging mod delivers it.""" + return Event( + event_name="thread.direct_message.notification", + source_id=sender, + destination_id=AGENT_A, + payload={ + "sender_id": sender, + "content": {"text": text}, + }, + response_to=response_to, + requires_response=requires_response, + ) + + +class TestSendAndWaitCorrelation: + """AgentConnection.send_and_wait request/reply correlation.""" + + async def test_reply_arriving_during_send_is_not_lost(self): + """The waiter must be registered before the request is sent.""" + workspace = make_workspace() + client = workspace.client + + async def send_event_with_instant_reply(event): + # Reply lands while send_event is still in flight — before the + # old implementation would have registered its waiter. + await client._handle_event( + dm_notification(AGENT_B, "instant reply", response_to=event.event_id) + ) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=send_event_with_instant_reply) + + reply = await workspace.agent(AGENT_B).send_and_wait("ping", timeout=1.0) + + assert reply is not None + assert reply["content"]["text"] == "instant reply" + + async def test_counter_request_is_not_mistaken_for_reply(self): + """A request from the peer (requires_response) must not satisfy the wait.""" + workspace = make_workspace() + client = workspace.client + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + async def peer_sends_counter_request(): + await asyncio.sleep(0.05) + await client._handle_event( + dm_notification(AGENT_B, "counter request", requires_response=True) + ) + + peer_task = asyncio.create_task(peer_sends_counter_request()) + reply = await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.3) + await peer_task + + assert reply is None + + async def test_correlated_reply_wins_over_unrelated_traffic(self): + """Only the reply carrying our request id is accepted.""" + workspace = make_workspace() + client = workspace.client + + sent_events = [] + + async def capture_send(event): + sent_events.append(event) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=capture_send) + + async def peer_traffic(): + await asyncio.sleep(0.05) + # A counter-request first, then the real correlated reply. + await client._handle_event( + dm_notification(AGENT_B, "counter request", requires_response=True) + ) + await client._handle_event( + dm_notification( + AGENT_B, "real reply", response_to=sent_events[0].event_id + ) + ) + + peer_task = asyncio.create_task(peer_traffic()) + reply = await workspace.agent(AGENT_B).send_and_wait("ping", timeout=1.0) + await peer_task + + assert reply is not None + assert reply["content"]["text"] == "real reply" + + async def test_concurrent_waiters_each_get_their_own_reply(self): + """Two in-flight requests to the same agent must not swap replies.""" + workspace = make_workspace() + client = workspace.client + + sent_events = [] + + async def capture_send(event): + sent_events.append(event) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=capture_send) + + async def peer_replies_in_reverse_order(): + while len(sent_events) < 2: + await asyncio.sleep(0.01) + for event in reversed(sent_events): + text = f"reply to {event.payload['content']['text']}" + await client._handle_event( + dm_notification(AGENT_B, text, response_to=event.event_id) + ) + + peer_task = asyncio.create_task(peer_replies_in_reverse_order()) + connection = workspace.agent(AGENT_B) + reply_one, reply_two = await asyncio.gather( + connection.send_and_wait("first", timeout=1.0), + connection.send_and_wait("second", timeout=1.0), + ) + await peer_task + + assert reply_one["content"]["text"] == "reply to first" + assert reply_two["content"]["text"] == "reply to second" + + async def test_legacy_reply_without_response_to_still_accepted(self): + """Peers that don't correlate replies keep working (fallback match).""" + workspace = make_workspace() + client = workspace.client + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + async def peer_plain_reply(): + await asyncio.sleep(0.05) + await client._handle_event(dm_notification(AGENT_B, "plain reply")) + + peer_task = asyncio.create_task(peer_plain_reply()) + reply = await workspace.agent(AGENT_B).send_and_wait("ping", timeout=1.0) + await peer_task + + assert reply is not None + assert reply["content"]["text"] == "plain reply" + + async def test_send_failure_returns_none_and_deregisters_waiter(self): + workspace = make_workspace() + workspace.send_event = AsyncMock( + return_value=EventResponse(success=False, message="boom") + ) + + reply = await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.2) + + assert reply is None + assert workspace.client._event_waiters == [] + + +class TestEventWaiterSemantics: + async def test_all_matching_waiters_are_woken_by_one_event(self): + """Documented semantics: one event satisfies every matching waiter.""" + client = AgentClient(agent_id=AGENT_A) + client.connector = MagicMock() + + waiter_one = client.expect_event(lambda e: e.source_id == AGENT_B) + waiter_two = client.expect_event(lambda e: e.source_id == AGENT_B) + + await client._handle_event(dm_notification(AGENT_B, "broadcast-ish")) + + event_one = await waiter_one.wait(timeout=0.2) + event_two = await waiter_two.wait(timeout=0.2) + assert event_one is not None + assert event_two is not None + + async def test_waiter_cancel_removes_registration(self): + client = AgentClient(agent_id=AGENT_A) + client.connector = MagicMock() + + waiter = client.expect_event(lambda e: True) + assert len(client._event_waiters) == 1 + waiter.cancel() + assert client._event_waiters == [] + waiter.cancel() # idempotent + + +class TestChannelReplyCorrelation: + def reply_notification(self, channel: str, reply_to_id: str, text: str) -> Event: + """Build a thread.reply.notification as the messaging mod delivers it.""" + return Event( + event_name="thread.reply.notification", + source_id=AGENT_B, + destination_id=AGENT_A, + payload={ + "channel": channel, + "reply_to_id": reply_to_id, + "content": {"text": text}, + }, + ) + + async def test_post_and_wait_matches_reply_to_posted_message(self): + workspace = make_workspace() + client = workspace.client + + sent_events = [] + + async def capture_send(event): + sent_events.append(event) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=capture_send) + + async def peer_replies(): + while not sent_events: + await asyncio.sleep(0.01) + post_id = sent_events[0].event_id + # A reply to some other message must be ignored... + await client._handle_event( + self.reply_notification("general", "unrelated-id", "wrong thread") + ) + # ...and the reply to our post accepted. + await client._handle_event( + self.reply_notification("general", post_id, "the answer") + ) + + peer_task = asyncio.create_task(peer_replies()) + reply = await workspace.channel("general").post_and_wait( + "question", timeout=1.0 + ) + await peer_task + + assert reply is not None + assert reply["content"]["text"] == "the answer" + + +class TestReactScopeGuard: + async def test_blocking_wait_inside_react_warns(self): + workspace = make_workspace() + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + with react_scope(AGENT_A): + with pytest.warns(DeprecationWarning, match="send_and_wait"): + await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.05) + + async def test_blocking_wait_inside_react_strict_raises(self, monkeypatch): + monkeypatch.setenv(STRICT_ENV_VAR, "1") + workspace = make_workspace() + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + with react_scope(AGENT_A): + with pytest.raises(BlockingWaitInReactError): + await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.05) + + # Nothing was sent and no waiter leaked. + workspace.send_event.assert_not_awaited() + assert workspace.client._event_waiters == [] + + async def test_no_warning_outside_react(self, recwarn): + workspace = make_workspace() + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.05) + + deprecations = [ + w for w in recwarn.list if issubclass(w.category, DeprecationWarning) + ] + assert deprecations == [] + + def test_react_scope_sets_and_resets_context(self): + assert current_react_agent.get() is None + with react_scope(AGENT_A): + assert current_react_agent.get() == AGENT_A + assert current_react_agent.get() is None + + +class TestMessagingModCorrelationRelay: + """The mod must preserve correlation fields when relaying direct messages.""" + + async def test_direct_message_notification_carries_correlation_fields(self): + from openagents.mods.workspace.messaging.mod import ThreadMessagingNetworkMod + + mod = ThreadMessagingNetworkMod() + mod._network = MagicMock() + mod._network.process_event = AsyncMock() + + request = Event( + event_name="thread.direct_message.send", + source_id=AGENT_A, + destination_id=AGENT_B, + payload={ + "target_agent_id": AGENT_B, + "message_type": "direct_message", + "content": {"text": "ping"}, + }, + requires_response=True, + ) + + await mod._process_direct_message(request) + + mod._network.process_event.assert_awaited_once() + notification = mod._network.process_event.await_args.args[0] + assert notification.event_name == "thread.direct_message.notification" + assert notification.destination_id == AGENT_B + assert notification.requires_response is True + assert notification.payload["message_id"] == request.event_id + + # And the reply relays its response_to back to the requester. + mod._network.process_event.reset_mock() + reply = Event( + event_name="thread.direct_message.send", + source_id=AGENT_B, + destination_id=AGENT_A, + payload={ + "target_agent_id": AGENT_A, + "message_type": "direct_message", + "content": {"text": "pong"}, + }, + response_to=request.event_id, + ) + + await mod._process_direct_message(reply) + + notification = mod._network.process_event.await_args.args[0] + assert notification.response_to == request.event_id + assert notification.requires_response is False From 87697aea1dd3fa68d4c3a54cdc240e3afb4ea1b5 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Fri, 31 Jul 2026 10:42:47 +0000 Subject: [PATCH 2/5] Add delegation lineage checks and wait-graph deadlock detection Task delegation had no defense against circular delegation. A->B->A (or any longer cycle) created tasks forever, each burning its own timeout while the loop kept spawning new work. Delegation lineage (task_delegation mod) - clients may pass parent_task_id when delegating as part of handling a task. The server derives root_task_id, delegation_chain and delegation_depth from the parent task in the task store, so a client cannot forge or truncate the chain - rejects self-delegation (agent ids normalized so agent prefix aliases cannot bypass the check), delegation cycles, chains beyond max_delegation_depth, and delegators over the max_active_delegations_per_agent fuse, since cycle checks alone cannot stop non-cyclic exponential fan-out - both creation paths are covered, direct task.delegate and capability-routed tasks - assignment notifications carry the lineage so assignees can extend the chain with parent_task_id on their own delegations Wait-graph deadlock detection (event gateway) - every event flows through the central gateway, so a WaitGraphMonitor builds a wait-for graph there from requires_response requests and clears edges on matching response_to replies, with a TTL for abandoned waits - when an edge closes a cycle, every agent on it receives an agent.wait.deadlock_detected event naming the cycle. send_and_wait recognizes reports about its own request and fails fast with a clear log instead of blocking until timeout - this covers what lineage checks cannot see, mutual send_and_wait, independent tasks pointing at each other, and mixed-mechanism cycles Covered by tests/mods/test_task_delegation.py (TestDelegationLineage), tests/network/test_wait_graph.py and tests/workspace/test_sync_wait_governance.py --- .../task_delegation/a2a_delegation.py | 49 ++++ .../coordination/task_delegation/adapter.py | 27 ++ .../mods/coordination/task_delegation/mod.py | 184 +++++++++++++- sdk/src/openagents/sdk/event_gateway.py | 14 + sdk/src/openagents/sdk/wait_graph.py | 182 +++++++++++++ sdk/src/openagents/sdk/workspace.py | 19 +- tests/mods/test_task_delegation.py | 240 ++++++++++++++++++ tests/network/test_wait_graph.py | 112 ++++++++ tests/workspace/test_sync_wait_governance.py | 38 +++ 9 files changed, 861 insertions(+), 4 deletions(-) create mode 100644 sdk/src/openagents/sdk/wait_graph.py create mode 100644 tests/network/test_wait_graph.py diff --git a/sdk/src/openagents/mods/coordination/task_delegation/a2a_delegation.py b/sdk/src/openagents/mods/coordination/task_delegation/a2a_delegation.py index f81ad5e1b..c71dabe77 100644 --- a/sdk/src/openagents/mods/coordination/task_delegation/a2a_delegation.py +++ b/sdk/src/openagents/mods/coordination/task_delegation/a2a_delegation.py @@ -38,6 +38,26 @@ # Default timeout in seconds DEFAULT_TIMEOUT_SECONDS = 300 +# Default maximum delegation chain depth (root delegation = depth 1) +DEFAULT_MAX_DELEGATION_DEPTH = 10 + +# Default maximum concurrent non-terminal delegations per delegator +DEFAULT_MAX_ACTIVE_DELEGATIONS_PER_AGENT = 50 + + +def normalize_agent_id(agent_id: Optional[str]) -> str: + """Normalize an agent id for identity comparisons in delegation checks. + + Strips whitespace and a leading ``agent:`` prefix so that ``agent:alice`` + and ``alice`` cannot bypass self-delegation or cycle checks. + """ + if not agent_id: + return "" + normalized = str(agent_id).strip() + if normalized.startswith("agent:"): + normalized = normalized[len("agent:"):] + return normalized + # ============================================================================= # Delegation Metadata Helpers @@ -54,6 +74,10 @@ def create_delegation_metadata( is_external_assignee: bool = False, delegator_url: Optional[str] = None, assignee_url: Optional[str] = None, + parent_task_id: Optional[str] = None, + root_task_id: Optional[str] = None, + delegation_chain: Optional[List[str]] = None, + delegation_depth: int = 1, ) -> Dict[str, Any]: """Create delegation metadata for an A2A Task. @@ -67,6 +91,10 @@ def create_delegation_metadata( is_external_assignee: True if assignee is an external A2A agent delegator_url: A2A URL if external delegator assignee_url: A2A URL if external assignee + parent_task_id: Task this delegation was spawned from, if any + root_task_id: Root task of the delegation tree (None for root tasks) + delegation_chain: Server-derived chain of agent ids, delegator first + delegation_depth: Number of delegation hops from the root (root = 1) Returns: Metadata dictionary for the A2A Task @@ -83,6 +111,11 @@ def create_delegation_metadata( "is_external_assignee": is_external_assignee, "delegator_url": delegator_url, "assignee_url": assignee_url, + "parent_task_id": parent_task_id, + "root_task_id": root_task_id, + "delegation_chain": delegation_chain + or [normalize_agent_id(delegator_id), normalize_agent_id(assignee_id)], + "delegation_depth": delegation_depth, }, "payload": payload or {}, "progress_summary": { @@ -120,6 +153,10 @@ def extract_delegation_metadata(task: Task) -> Dict[str, Any]: "is_external_assignee": delegation.get("is_external_assignee", False), "delegator_url": delegation.get("delegator_url"), "assignee_url": delegation.get("assignee_url"), + "parent_task_id": delegation.get("parent_task_id"), + "root_task_id": delegation.get("root_task_id"), + "delegation_chain": delegation.get("delegation_chain"), + "delegation_depth": delegation.get("delegation_depth", 1), "payload": task.metadata.get("payload", {}), "error": task.metadata.get("error_details", {}).get("error"), "is_timeout": task.metadata.get("error_details", {}).get("is_timeout", False), @@ -197,6 +234,10 @@ def create_delegation_task( is_external_assignee: bool = False, delegator_url: Optional[str] = None, assignee_url: Optional[str] = None, + parent_task_id: Optional[str] = None, + root_task_id: Optional[str] = None, + delegation_chain: Optional[List[str]] = None, + delegation_depth: int = 1, ) -> Task: """Create an A2A Task for a delegation request. @@ -211,6 +252,10 @@ def create_delegation_task( is_external_assignee: True if assignee is an external A2A agent delegator_url: A2A URL if external delegator assignee_url: A2A URL if external assignee + parent_task_id: Task this delegation was spawned from, if any + root_task_id: Root task of the delegation tree (None for root tasks) + delegation_chain: Server-derived chain of agent ids, delegator first + delegation_depth: Number of delegation hops from the root (root = 1) Returns: New A2A Task in SUBMITTED state @@ -225,6 +270,10 @@ def create_delegation_task( is_external_assignee=is_external_assignee, delegator_url=delegator_url, assignee_url=assignee_url, + parent_task_id=parent_task_id, + root_task_id=root_task_id, + delegation_chain=delegation_chain, + delegation_depth=delegation_depth, ) # Create initial message describing the task diff --git a/sdk/src/openagents/mods/coordination/task_delegation/adapter.py b/sdk/src/openagents/mods/coordination/task_delegation/adapter.py index a480dd266..9173a16b5 100644 --- a/sdk/src/openagents/mods/coordination/task_delegation/adapter.py +++ b/sdk/src/openagents/mods/coordination/task_delegation/adapter.py @@ -84,6 +84,15 @@ def get_tools(self) -> List[AgentTool]: "description": "Optional timeout in seconds (default 300)", "default": 300, }, + "parent_task_id": { + "type": "string", + "description": ( + "If you are delegating as part of handling a task " + "assigned to you, pass that task's id so the " + "network can track the delegation chain and " + "reject cycles" + ), + }, }, "required": ["assignee_id", "description"], }, @@ -371,6 +380,15 @@ def get_tools(self) -> List[AgentTool]: "type": "string", "description": "Agent ID to use if no capability match found", }, + "parent_task_id": { + "type": "string", + "description": ( + "If you are delegating as part of handling a task " + "assigned to you, pass that task's id so the " + "network can track the delegation chain and " + "reject cycles" + ), + }, }, "required": ["description"], }, @@ -386,6 +404,7 @@ async def delegate_task( description: str, payload: Optional[Dict[str, Any]] = None, timeout_seconds: int = 300, + parent_task_id: Optional[str] = None, ) -> Dict[str, Any]: """Delegate a task to another agent. @@ -394,6 +413,9 @@ async def delegate_task( description: Description of the task payload: Optional task data/parameters timeout_seconds: Timeout in seconds (default 300) + parent_task_id: If this delegation is part of handling another + task, that task's id. The network derives the delegation + chain from it and rejects delegation cycles. Returns: Response containing task_id and status on success, or error on failure @@ -413,6 +435,7 @@ async def delegate_task( "description": description, "payload": payload or {}, "timeout_seconds": timeout_seconds, + "parent_task_id": parent_task_id, }, relevant_mod="openagents.mods.coordination.task_delegation", ) @@ -837,6 +860,7 @@ async def route_task( timeout_seconds: int = 300, selection_strategy: str = "first", fallback_assignee_id: Optional[str] = None, + parent_task_id: Optional[str] = None, ) -> Dict[str, Any]: """Route a task to an agent based on required capabilities. @@ -849,6 +873,8 @@ async def route_task( timeout_seconds: Timeout in seconds (default 300) selection_strategy: How to select from matches ("first" or "random") fallback_assignee_id: Agent to use if no capability match found + parent_task_id: If this routing is part of handling another task, + that task's id, so the network can track the delegation chain Returns: Response containing task_id and assigned agent on success @@ -871,6 +897,7 @@ async def route_task( "payload": payload or {}, "timeout_seconds": timeout_seconds, "selection_strategy": selection_strategy, + "parent_task_id": parent_task_id, } if required_capabilities: diff --git a/sdk/src/openagents/mods/coordination/task_delegation/mod.py b/sdk/src/openagents/mods/coordination/task_delegation/mod.py index 8c6341c93..620f64baa 100644 --- a/sdk/src/openagents/mods/coordination/task_delegation/mod.py +++ b/sdk/src/openagents/mods/coordination/task_delegation/mod.py @@ -35,6 +35,8 @@ ) from .a2a_delegation import ( + DEFAULT_MAX_ACTIVE_DELEGATIONS_PER_AGENT, + DEFAULT_MAX_DELEGATION_DEPTH, DEFAULT_TIMEOUT_SECONDS, TERMINAL_STATES, create_delegation_metadata, @@ -44,6 +46,7 @@ extract_delegation_metadata, increment_progress_count, is_task_expired, + normalize_agent_id, update_delegation_metadata, ) from .capability_matcher import ( @@ -98,6 +101,16 @@ def __init__(self, mod_name: str = "openagents.mods.coordination.task_delegation "timeout_check_interval", self.DEFAULT_TIMEOUT_CHECK_INTERVAL ) + # Delegation lineage limits (guard against delegation cycles and + # runaway fan-out); configurable via mod config + self._max_delegation_depth = self.config.get( + "max_delegation_depth", DEFAULT_MAX_DELEGATION_DEPTH + ) + self._max_active_delegations_per_agent = self.config.get( + "max_active_delegations_per_agent", + DEFAULT_MAX_ACTIVE_DELEGATIONS_PER_AGENT, + ) + logger.info("Initializing Task Delegation network mod (A2A-compatible)") def bind_network(self, network) -> bool: @@ -291,6 +304,141 @@ def _create_response( """Create a standardized event response.""" return EventResponse(success=success, message=message, data=data or {}) + async def _validate_delegation_lineage( + self, + delegator_id: str, + assignee_id: str, + parent_task_id: Optional[str], + ) -> Tuple[Optional[EventResponse], Dict[str, Any]]: + """Validate a delegation against its lineage and derive chain metadata. + + The delegation chain is derived server-side from the parent task in + the task store — clients only submit ``parent_task_id`` — so a client + cannot forge, truncate or omit the chain to sneak past cycle checks. + + Returns: + Tuple of (error response or None, lineage metadata for the task) + """ + delegator = normalize_agent_id(delegator_id) + assignee = normalize_agent_id(assignee_id) + + def _error(code: str, message: str, **extra: Any) -> EventResponse: + return self._create_response( + success=False, message=message, data={"error": code, **extra} + ) + + if not delegator: + return _error("invalid_delegator", "delegator id is required"), {} + + if delegator == assignee: + return ( + _error( + "self_delegation_rejected", + f"Agent {delegator} cannot delegate a task to itself", + ), + {}, + ) + + chain: List[str] = [delegator] + root_task_id: Optional[str] = None + + if parent_task_id: + parent = await self.task_store.get_task(parent_task_id) + if parent is None: + return ( + _error( + "parent_task_not_found", + f"Parent task {parent_task_id} not found", + ), + {}, + ) + if parent.status.state in TERMINAL_STATES: + return ( + _error( + "parent_task_terminal", + f"Parent task {parent_task_id} is already " + f"{parent.status.state.value}", + ), + {}, + ) + + parent_delegation = (parent.metadata or {}).get("delegation", {}) + parent_assignee = normalize_agent_id( + parent_delegation.get("assignee_id") + ) + if parent_assignee != delegator: + return ( + _error( + "delegator_not_parent_assignee", + f"Agent {delegator} is not the assignee of parent " + f"task {parent_task_id}", + ), + {}, + ) + + parent_chain = parent_delegation.get("delegation_chain") or [ + normalize_agent_id(parent_delegation.get("delegator_id")), + parent_assignee, + ] + chain = [normalize_agent_id(entry) for entry in parent_chain] + root_task_id = parent_delegation.get("root_task_id") or parent_task_id + + if assignee in chain: + return ( + _error( + "delegation_cycle_detected", + f"Delegating to {assignee} would close a delegation " + f"cycle: {' -> '.join(chain + [assignee])}", + delegation_chain=chain, + ), + {}, + ) + + new_chain = chain + [assignee] + depth = len(new_chain) - 1 + if depth > self._max_delegation_depth: + return ( + _error( + "delegation_depth_exceeded", + f"Delegation depth {depth} exceeds the maximum of " + f"{self._max_delegation_depth}", + delegation_chain=new_chain, + ), + {}, + ) + + # Resource fuse: cap concurrent non-terminal delegations per delegator. + # Cycle detection cannot stop non-cyclic exponential fan-out. + active_count = 0 + all_tasks = await self.task_store.list_tasks(limit=100000) + for existing in all_tasks: + if existing.status.state in TERMINAL_STATES: + continue + existing_delegator = normalize_agent_id( + (existing.metadata or {}) + .get("delegation", {}) + .get("delegator_id") + ) + if existing_delegator == delegator: + active_count += 1 + if active_count >= self._max_active_delegations_per_agent: + return ( + _error( + "delegation_limit_exceeded", + f"Agent {delegator} already has {active_count} active " + f"delegations (limit " + f"{self._max_active_delegations_per_agent})", + ), + {}, + ) + + return None, { + "parent_task_id": parent_task_id, + "root_task_id": root_task_id, + "delegation_chain": new_chain, + "delegation_depth": depth, + } + @mod_event_handler("task.delegate") async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: """Handle task delegation requests.""" @@ -324,6 +472,16 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: data={"error": "timeout_seconds must be a positive number"}, ) + # Validate lineage (self-delegation, cycles, depth, fan-out limits) + # and derive the server-side delegation chain. + lineage_error, lineage = await self._validate_delegation_lineage( + delegator_id=delegator_id, + assignee_id=assignee_id, + parent_task_id=payload.get("parent_task_id"), + ) + if lineage_error: + return lineage_error + # Check if assignee is an external A2A agent is_external = ( self._external_delegator @@ -342,6 +500,7 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: timeout_seconds=int(timeout_seconds), is_external_assignee=is_external, assignee_url=external_url, + **lineage, ) # Store the task @@ -389,7 +548,9 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: data={"error": str(e)}, ) else: - # Local assignee - send notification + # Local assignee - send notification. Lineage fields let the + # assignee delegate follow-up work with parent_task_id=task_id so + # the server can track the chain. await self._send_notification( "task.notification.assigned", assignee_id, @@ -399,6 +560,10 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: "description": description, "payload": payload.get("payload", {}), "timeout_seconds": timeout_seconds, + "parent_task_id": lineage.get("parent_task_id"), + "root_task_id": lineage.get("root_task_id"), + "delegation_chain": lineage.get("delegation_chain"), + "delegation_depth": lineage.get("delegation_depth"), }, ) @@ -410,6 +575,8 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: "task_id": task.id, "status": task.status.state.value, "created_at": delegation.get("created_at"), + "delegation_chain": delegation.get("delegation_chain"), + "delegation_depth": delegation.get("delegation_depth"), }, ) @@ -1278,6 +1445,16 @@ async def _delegate_routed_task( """Create and delegate a task after routing.""" timeout_seconds = payload.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS) + # Routed tasks go through the same lineage validation as direct + # delegations — capability routing must not bypass cycle checks. + lineage_error, lineage = await self._validate_delegation_lineage( + delegator_id=delegator_id, + assignee_id=assignee_id, + parent_task_id=payload.get("parent_task_id"), + ) + if lineage_error: + return lineage_error + # Check if assignee is external is_external = ( self._external_delegator @@ -1296,6 +1473,7 @@ async def _delegate_routed_task( timeout_seconds=int(timeout_seconds), is_external_assignee=is_external, assignee_url=external_url, + **lineage, ) # Store the task @@ -1348,6 +1526,10 @@ async def _delegate_routed_task( "payload": payload.get("payload", {}), "timeout_seconds": timeout_seconds, "routed_by_capability": True, + "parent_task_id": lineage.get("parent_task_id"), + "root_task_id": lineage.get("root_task_id"), + "delegation_chain": lineage.get("delegation_chain"), + "delegation_depth": lineage.get("delegation_depth"), }, ) diff --git a/sdk/src/openagents/sdk/event_gateway.py b/sdk/src/openagents/sdk/event_gateway.py index e235db1f0..9018e25dd 100644 --- a/sdk/src/openagents/sdk/event_gateway.py +++ b/sdk/src/openagents/sdk/event_gateway.py @@ -8,6 +8,7 @@ from openagents.models.event import Event, EventSubscription from openagents.models.event_response import EventResponse from openagents.models.network_role import NetworkRole +from openagents.sdk.wait_graph import WaitGraphMonitor if TYPE_CHECKING: from openagents.sdk.network import AgentNetwork @@ -105,6 +106,12 @@ def __init__(self, network: "AgentNetwork"): self.mod_event_processor = ModEventProcessor(network.mods) # Activity counters for reputation reporting self._activity_counters: Dict[str, Dict[str, int]] = {} # agent_id -> {event_type: count} + # Wait-for graph over requires_response/response_to traffic; detects + # agents blocked waiting on each other and notifies them. + self.wait_graph = WaitGraphMonitor( + network_id=getattr(network, "network_id", "network"), + deliver=self.deliver_event, + ) async def process_system_command(self, event: Event) -> Optional[EventResponse]: """ @@ -177,6 +184,13 @@ async def process_event( self._activity_counters[src] = {} self._activity_counters[src]["messages_sent"] = self._activity_counters[src].get("messages_sent", 0) + 1 + # Track request/response waits for deadlock detection; never let the + # monitor break event processing. + try: + await self.wait_graph.observe(event) + except Exception as e: + logger.error(f"Wait-graph observation failed: {e}") + # Process the event through the pipeline response = None if event.event_name.startswith("system."): diff --git a/sdk/src/openagents/sdk/wait_graph.py b/sdk/src/openagents/sdk/wait_graph.py new file mode 100644 index 000000000..055c392ab --- /dev/null +++ b/sdk/src/openagents/sdk/wait_graph.py @@ -0,0 +1,182 @@ +"""Server-side wait-for graph deadlock detection. + +Every event in a network flows through the central event gateway, which makes +the server the natural place to watch for agents waiting on each other. An +event sent with ``requires_response=True`` from agent A to agent B registers a +directed wait edge A -> B keyed by the request's event id; an event whose +``response_to`` references that request clears the edge. If adding an edge +closes a cycle (A waits on B while B — directly or transitively — waits on A), +every agent on the cycle is notified with an ``agent.wait.deadlock_detected`` +event so its waiter can fail fast with the cycle path instead of blocking +until its timeout with no explanation. + +This catches the wait cycles that delegation-chain validation cannot see: +mutual ``send_and_wait`` calls, independent tasks pointing at each other, and +mixed-mechanism cycles. + +Edges expire after a TTL so that abandoned waits (client crash, timeout +without a response event) cannot poison the graph forever. +""" + +import logging +import time +from typing import Awaitable, Callable, Dict, List, Optional + +from openagents.models.event import Event + +logger = logging.getLogger(__name__) + +DEADLOCK_EVENT_NAME = "agent.wait.deadlock_detected" + +# Client-side blocking waits default to 30s; keep edges a bit longer so slow +# responses still clear them, but abandoned waits eventually vanish. +DEFAULT_EDGE_TTL_SECONDS = 60.0 + + +class _WaitEdge: + __slots__ = ("request_id", "waiter", "target", "expires_at") + + def __init__(self, request_id: str, waiter: str, target: str, expires_at: float): + self.request_id = request_id + self.waiter = waiter + self.target = target + self.expires_at = expires_at + + +class WaitGraphMonitor: + """Observes gateway traffic and detects request/response wait cycles.""" + + def __init__( + self, + network_id: str, + deliver: Callable[[Event], Awaitable[None]], + edge_ttl_seconds: float = DEFAULT_EDGE_TTL_SECONDS, + ): + """Initialize the monitor. + + Args: + network_id: Used as source_id of deadlock notifications + deliver: Coroutine used to deliver a notification event to its + destination agent (the gateway's deliver_event) + edge_ttl_seconds: How long an unanswered wait edge is kept + """ + self._network_id = network_id + self._deliver = deliver + self._edge_ttl = edge_ttl_seconds + # request_id -> edge; the wait graph adjacency is derived on demand + self._edges: Dict[str, _WaitEdge] = {} + + async def observe(self, event: Event) -> None: + """Update the wait graph from one event and report any new cycle.""" + self._purge_expired() + + # A response clears the wait it answers, whatever event carries it + # (the original reply or a mod relay of it). + if event.response_to: + self._edges.pop(event.response_to, None) + return + + if not event.requires_response: + return + + waiter = event.source_id + target = event.destination_id + if not self._is_agent_to_agent(waiter, target): + return + + # Mod relays (…​.notification) duplicate the original request with a + # fresh event id; the reply's response_to references the original, so + # a relay edge would never be cleared. Track the original only. + if event.event_name.endswith(".notification"): + return + + self._edges[event.event_id] = _WaitEdge( + request_id=event.event_id, + waiter=waiter, + target=target, + expires_at=time.time() + self._edge_ttl, + ) + + cycle = self._find_cycle(waiter) + if cycle: + await self._notify_cycle(cycle) + + def waiting_targets(self, waiter: str) -> List[str]: + """Agents the given agent currently waits on (for introspection).""" + return [e.target for e in self._edges.values() if e.waiter == waiter] + + def _is_agent_to_agent(self, waiter: Optional[str], target: Optional[str]) -> bool: + if not waiter or not target or waiter == target: + return False + for endpoint in (waiter, target): + if endpoint.startswith(("channel:", "group:", "network:", "mod:")): + return False + if "broadcast" in endpoint: + return False + return True + + def _purge_expired(self) -> None: + now = time.time() + expired = [rid for rid, e in self._edges.items() if e.expires_at <= now] + for rid in expired: + del self._edges[rid] + + def _find_cycle(self, start: str) -> Optional[List[str]]: + """DFS from `start` over wait edges; return the cycle path if one + leads back to `start`.""" + adjacency: Dict[str, List[str]] = {} + for edge in self._edges.values(): + adjacency.setdefault(edge.waiter, []).append(edge.target) + + path: List[str] = [start] + visited = set() + + def dfs(node: str) -> Optional[List[str]]: + for neighbor in adjacency.get(node, []): + if neighbor == start: + return path + [start] + if neighbor in visited: + continue + visited.add(neighbor) + path.append(neighbor) + found = dfs(neighbor) + if found: + return found + path.pop() + return None + + return dfs(start) + + async def _notify_cycle(self, cycle: List[str]) -> None: + """Deliver a deadlock notification to every agent on the cycle.""" + agents = list(dict.fromkeys(cycle)) # de-duplicate, keep order + request_ids = [ + e.request_id + for e in self._edges.values() + if e.waiter in agents and e.target in agents + ] + logger.warning( + f"Wait-for cycle detected between agents: {' -> '.join(cycle)}" + ) + for agent_id in agents: + notification = Event( + event_name=DEADLOCK_EVENT_NAME, + source_id=self._network_id, + destination_id=agent_id, + payload={ + "cycle": cycle, + "request_ids": request_ids, + "message": ( + "Deadlock detected. These agents are all blocked " + "waiting for a response from the next agent in the " + "cycle, so no response can ever be produced. Stop " + "waiting and handle the peer's request first." + ), + }, + ) + try: + await self._deliver(notification) + except Exception as e: + logger.error( + f"Failed to deliver deadlock notification to {agent_id}: {e}" + ) diff --git a/sdk/src/openagents/sdk/workspace.py b/sdk/src/openagents/sdk/workspace.py index 42c9df77b..cf8daf5a5 100644 --- a/sdk/src/openagents/sdk/workspace.py +++ b/sdk/src/openagents/sdk/workspace.py @@ -222,6 +222,12 @@ async def send_and_wait( def reply_condition(msg): try: + # The network reports wait-for cycles; fail fast instead of + # blocking until timeout when our request is part of one. + if msg.event_name == "agent.wait.deadlock_detected": + return request.event_id in (msg.payload or {}).get( + "request_ids", [] + ) if msg.event_name != "thread.direct_message.notification": return False if msg.source_id != self.agent_id: @@ -244,9 +250,16 @@ def reply_condition(msg): logger.info(f"Sent message to {self.agent_id}, waiting for reply...") reply = await waiter.wait(timeout=timeout) - if reply: - return reply.payload - return None + if reply is None: + return None + if reply.event_name == "agent.wait.deadlock_detected": + cycle = (reply.payload or {}).get("cycle") + logger.error( + f"send_and_wait to {self.agent_id} aborted, the network " + f"detected a wait-for deadlock: {cycle}" + ) + return None + return reply.payload def __str__(self) -> str: return f"AgentConnection({self.agent_id})" diff --git a/tests/mods/test_task_delegation.py b/tests/mods/test_task_delegation.py index 18f453234..d1e337aba 100644 --- a/tests/mods/test_task_delegation.py +++ b/tests/mods/test_task_delegation.py @@ -848,5 +848,245 @@ def test_task_from_dict(self): assert task.metadata == {"outcome": "success"} +class TestDelegationLineage: + """Server-derived delegation chains, cycle detection and fuses.""" + + def delegate_event( + self, + delegator: str, + assignee: str, + description: str = "some task", + parent_task_id: str = None, + ) -> Event: + return Event( + event_name="task.delegate", + source_id=delegator, + payload={ + "assignee_id": assignee, + "description": description, + "parent_task_id": parent_task_id, + "timeout_seconds": 300, + }, + ) + + async def delegate(self, mod, *args, **kwargs): + return await mod._handle_task_delegate(self.delegate_event(*args, **kwargs)) + + @pytest.mark.asyncio + async def test_self_delegation_rejected(self, task_delegation_mod): + response = await self.delegate(task_delegation_mod, "agent_alice", "agent_alice") + assert response.success is False + assert response.data["error"] == "self_delegation_rejected" + + @pytest.mark.asyncio + async def test_self_delegation_rejected_despite_prefix_alias( + self, task_delegation_mod + ): + """agent:alice and alice are the same agent for lineage checks.""" + response = await self.delegate(task_delegation_mod, "agent:alice", "alice") + assert response.success is False + assert response.data["error"] == "self_delegation_rejected" + + @pytest.mark.asyncio + async def test_root_delegation_records_chain(self, task_delegation_mod): + response = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + assert response.success is True + assert response.data["delegation_chain"] == ["agent_alice", "agent_bob"] + assert response.data["delegation_depth"] == 1 + + @pytest.mark.asyncio + async def test_parent_child_cycle_rejected(self, task_delegation_mod): + """A delegates to B, B delegating back to A must be rejected.""" + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + assert first.success is True + parent_id = first.data["task_id"] + + second = await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_alice", + parent_task_id=parent_id, + ) + assert second.success is False + assert second.data["error"] == "delegation_cycle_detected" + assert second.data["delegation_chain"] == ["agent_alice", "agent_bob"] + + @pytest.mark.asyncio + async def test_three_hop_cycle_rejected(self, task_delegation_mod): + """A -> B -> C -> A closes a cycle through the derived chain.""" + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + second = await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_charlie", + parent_task_id=first.data["task_id"], + ) + assert second.success is True + assert second.data["delegation_chain"] == [ + "agent_alice", + "agent_bob", + "agent_charlie", + ] + + third = await self.delegate( + task_delegation_mod, + "agent_charlie", + "agent_alice", + parent_task_id=second.data["task_id"], + ) + assert third.success is False + assert third.data["error"] == "delegation_cycle_detected" + + @pytest.mark.asyncio + async def test_normal_chain_delegation_succeeds(self, task_delegation_mod): + """A -> B -> C is a legitimate chain and carries lineage metadata.""" + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + second = await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_charlie", + parent_task_id=first.data["task_id"], + ) + assert second.success is True + assert second.data["delegation_depth"] == 2 + + task = await task_delegation_mod.task_store.get_task(second.data["task_id"]) + delegation = task.metadata["delegation"] + assert delegation["parent_task_id"] == first.data["task_id"] + assert delegation["root_task_id"] == first.data["task_id"] + assert delegation["delegation_chain"] == [ + "agent_alice", + "agent_bob", + "agent_charlie", + ] + + @pytest.mark.asyncio + async def test_parent_task_not_found(self, task_delegation_mod): + response = await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_charlie", + parent_task_id="no-such-task", + ) + assert response.success is False + assert response.data["error"] == "parent_task_not_found" + + @pytest.mark.asyncio + async def test_delegator_must_be_parent_assignee(self, task_delegation_mod): + """Only the assignee of the parent task may extend its chain.""" + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + + response = await self.delegate( + task_delegation_mod, + "agent_mallory", + "agent_charlie", + parent_task_id=first.data["task_id"], + ) + assert response.success is False + assert response.data["error"] == "delegator_not_parent_assignee" + + @pytest.mark.asyncio + async def test_terminal_parent_rejected(self, task_delegation_mod): + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + task_id = first.data["task_id"] + + complete_event = Event( + event_name="task.complete", + source_id="agent_bob", + payload={"task_id": task_id, "result": {"done": True}}, + ) + completion = await task_delegation_mod._handle_task_complete(complete_event) + assert completion.success is True + + response = await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_charlie", + parent_task_id=task_id, + ) + assert response.success is False + assert response.data["error"] == "parent_task_terminal" + + @pytest.mark.asyncio + async def test_depth_limit_enforced(self, task_delegation_mod): + task_delegation_mod._max_delegation_depth = 2 + + first = await self.delegate(task_delegation_mod, "agent_a", "agent_b") + second = await self.delegate( + task_delegation_mod, + "agent_b", + "agent_c", + parent_task_id=first.data["task_id"], + ) + assert second.success is True + + third = await self.delegate( + task_delegation_mod, + "agent_c", + "agent_d", + parent_task_id=second.data["task_id"], + ) + assert third.success is False + assert third.data["error"] == "delegation_depth_exceeded" + + @pytest.mark.asyncio + async def test_active_delegation_fuse(self, task_delegation_mod): + """Cycle checks can't stop non-cyclic fan-out; the fuse caps it.""" + task_delegation_mod._max_active_delegations_per_agent = 2 + + assert (await self.delegate(task_delegation_mod, "agent_a", "agent_b")).success + assert (await self.delegate(task_delegation_mod, "agent_a", "agent_c")).success + + third = await self.delegate(task_delegation_mod, "agent_a", "agent_d") + assert third.success is False + assert third.data["error"] == "delegation_limit_exceeded" + + # Other delegators are unaffected. + assert (await self.delegate(task_delegation_mod, "agent_b", "agent_c")).success + + @pytest.mark.asyncio + async def test_routed_delegation_checks_lineage(self, task_delegation_mod): + """Capability routing must not bypass cycle checks.""" + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + + response = await task_delegation_mod._delegate_routed_task( + delegator_id="agent_bob", + assignee_id="agent_alice", + description="routed back", + payload={"parent_task_id": first.data["task_id"]}, + matched_count=1, + ) + assert response.success is False + assert response.data["error"] == "delegation_cycle_detected" + + @pytest.mark.asyncio + async def test_assignment_notification_carries_lineage( + self, task_delegation_mod, mock_network + ): + first = await self.delegate(task_delegation_mod, "agent_alice", "agent_bob") + await self.delegate( + task_delegation_mod, + "agent_bob", + "agent_charlie", + parent_task_id=first.data["task_id"], + ) + + notifications = [ + call.args[0] + for call in mock_network.process_event.await_args_list + if call.args[0].event_name == "task.notification.assigned" + ] + assert len(notifications) == 2 + child_payload = notifications[1].payload + assert child_payload["parent_task_id"] == first.data["task_id"] + assert child_payload["root_task_id"] == first.data["task_id"] + assert child_payload["delegation_chain"] == [ + "agent_alice", + "agent_bob", + "agent_charlie", + ] + assert child_payload["delegation_depth"] == 2 + + if __name__ == "__main__": pytest.main([__file__, "-v", "-s"]) diff --git a/tests/network/test_wait_graph.py b/tests/network/test_wait_graph.py new file mode 100644 index 000000000..89f865cd0 --- /dev/null +++ b/tests/network/test_wait_graph.py @@ -0,0 +1,112 @@ +"""Tests for the server-side wait-for graph deadlock detection.""" + +import pytest + +from openagents.models.event import Event +from openagents.sdk.wait_graph import DEADLOCK_EVENT_NAME, WaitGraphMonitor + + +def request_event(source: str, target: str, name: str = "thread.direct_message.send") -> Event: + return Event( + event_name=name, + source_id=source, + destination_id=target, + requires_response=True, + payload={}, + ) + + +def response_event(source: str, target: str, response_to: str) -> Event: + return Event( + event_name="thread.direct_message.send", + source_id=source, + destination_id=target, + response_to=response_to, + payload={}, + ) + + +@pytest.fixture +def delivered(): + return [] + + +@pytest.fixture +def monitor(delivered): + async def deliver(event): + delivered.append(event) + + return WaitGraphMonitor(network_id="test-network", deliver=deliver) + + +class TestWaitGraphMonitor: + async def test_single_wait_no_deadlock(self, monitor, delivered): + await monitor.observe(request_event("alice", "bob")) + assert delivered == [] + assert monitor.waiting_targets("alice") == ["bob"] + + async def test_response_clears_the_edge(self, monitor, delivered): + request = request_event("alice", "bob") + await monitor.observe(request) + await monitor.observe(response_event("bob", "alice", request.event_id)) + assert monitor.waiting_targets("alice") == [] + + # bob can now wait on alice without a (stale) cycle being reported + await monitor.observe(request_event("bob", "alice")) + assert delivered == [] + + async def test_mutual_wait_detected_and_both_notified(self, monitor, delivered): + request_ab = request_event("alice", "bob") + request_ba = request_event("bob", "alice") + await monitor.observe(request_ab) + await monitor.observe(request_ba) + + assert len(delivered) == 2 + notified = {event.destination_id for event in delivered} + assert notified == {"alice", "bob"} + for event in delivered: + assert event.event_name == DEADLOCK_EVENT_NAME + assert set(event.payload["request_ids"]) == { + request_ab.event_id, + request_ba.event_id, + } + assert event.payload["cycle"][0] == event.payload["cycle"][-1] + + async def test_three_agent_cycle_detected(self, monitor, delivered): + await monitor.observe(request_event("alice", "bob")) + await monitor.observe(request_event("bob", "charlie")) + assert delivered == [] + + await monitor.observe(request_event("charlie", "alice")) + + notified = {event.destination_id for event in delivered} + assert notified == {"alice", "bob", "charlie"} + + async def test_no_false_positive_on_shared_target(self, monitor, delivered): + await monitor.observe(request_event("alice", "bob")) + await monitor.observe(request_event("charlie", "bob")) + assert delivered == [] + + async def test_notification_relays_do_not_create_edges(self, monitor, delivered): + relay = request_event( + "alice", "bob", name="thread.direct_message.notification" + ) + await monitor.observe(relay) + assert monitor.waiting_targets("alice") == [] + + async def test_channel_and_broadcast_traffic_ignored(self, monitor, delivered): + await monitor.observe(request_event("alice", "channel:general")) + await monitor.observe(request_event("alice", "agent:broadcast")) + assert monitor.waiting_targets("alice") == [] + + async def test_expired_edges_are_purged(self, delivered): + async def deliver(event): + delivered.append(event) + + monitor = WaitGraphMonitor( + network_id="test-network", deliver=deliver, edge_ttl_seconds=0.0 + ) + await monitor.observe(request_event("alice", "bob")) + # The edge expired immediately; the reverse wait closes no cycle. + await monitor.observe(request_event("bob", "alice")) + assert delivered == [] diff --git a/tests/workspace/test_sync_wait_governance.py b/tests/workspace/test_sync_wait_governance.py index e77d87077..ec5322579 100644 --- a/tests/workspace/test_sync_wait_governance.py +++ b/tests/workspace/test_sync_wait_governance.py @@ -190,6 +190,44 @@ async def peer_plain_reply(): assert reply is not None assert reply["content"]["text"] == "plain reply" + async def test_deadlock_notification_fails_fast(self): + """A network deadlock report aborts the wait before the timeout.""" + workspace = make_workspace() + client = workspace.client + + sent_events = [] + + async def capture_send(event): + sent_events.append(event) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=capture_send) + + async def network_reports_deadlock(): + while not sent_events: + await asyncio.sleep(0.01) + await client._handle_event( + Event( + event_name="agent.wait.deadlock_detected", + source_id="test-network", + destination_id=AGENT_A, + payload={ + "cycle": [AGENT_A, AGENT_B, AGENT_A], + "request_ids": [sent_events[0].event_id], + }, + ) + ) + + network_task = asyncio.create_task(network_reports_deadlock()) + # Much shorter than the 5s timeout — the deadlock report unblocks us. + reply = await asyncio.wait_for( + workspace.agent(AGENT_B).send_and_wait("ping", timeout=5.0), + timeout=1.0, + ) + await network_task + + assert reply is None + async def test_send_failure_returns_none_and_deregisters_waiter(self): workspace = make_workspace() workspace.send_event = AsyncMock( From 14e523c178744e2b1a14ed8c760d37777035d4d6 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Fri, 31 Jul 2026 10:48:34 +0000 Subject: [PATCH 3/5] Clean up delegation hygiene issues and add gRPC send deadline - task.complete responses reported completed_at as None because the delegation snapshot was taken before the completion timestamp was written; re-extract after the update - a locally timed-out external task kept its background polling loop alive until process shutdown; stop polling when the timeout marks the task failed - gRPC SendEvent had no deadline, so a stalled server blocked the caller forever; use a 30s deadline in line with the client wait defaults - document the no-blocking-wait-in-react rule and the continuation pattern with reply_direct in the workspace interface guide --- docs/python-interface/workspace-interface.mdx | 48 +++++++++++++++++++ .../mods/coordination/task_delegation/mod.py | 15 ++++++ .../sdk/connectors/grpc_connector.py | 5 +- 3 files changed, 66 insertions(+), 2 deletions(-) diff --git a/docs/python-interface/workspace-interface.mdx b/docs/python-interface/workspace-interface.mdx index b0f417cf0..8a170a3d3 100644 --- a/docs/python-interface/workspace-interface.mdx +++ b/docs/python-interface/workspace-interface.mdx @@ -305,6 +305,54 @@ async def send_and_wait_pattern(ws: Workspace): print(f"❌ Status check failed: {e}") ``` + +**Do not call blocking waits inside an agent's `react()` / `on_direct()` handlers.** + +Worker agents process one incoming event at a time: the next event is not +handled until the current handler returns. If your handler blocks in +`send_and_wait()` (or any `wait_for_*()`), every other incoming message stalls +behind it — and if the peer agent blocks the same way waiting on you, both +agents deadlock until their timeouts expire. Calling these APIs inside a +handler emits a `DeprecationWarning`; set the environment variable +`OPENAGENTS_STRICT_NO_BLOCKING_WAIT=1` to turn it into an error. + +`send_and_wait()` is fine in scripts, tools and orchestration code that runs +*outside* the agent's event loop (like the examples above). + + +### Request/Reply Inside Agent Handlers (Continuation Pattern) + +Inside a `WorkerAgent`, send without waiting and treat the reply as the next +incoming event. Use `reply_direct()` on the responding side so the network can +correlate the reply with the request: + +```python +class CoordinatorAgent(WorkerAgent): + async def on_startup(self): + # Fire the request and return; do NOT block waiting for the answer + await self.send_direct( + to="assistant-agent", + text="What's the current status of the data processing task?", + ) + + async def on_direct(self, context): + # The reply arrives as a new event in the next react() round + text = context.incoming_event.payload.get("content", {}).get("text", "") + print(f"📋 Status report: {text}") + + +class AssistantAgent(WorkerAgent): + async def on_direct(self, context): + # reply_direct() correlates the reply with the incoming request + # (sets response_to), so waiters on the other side match it precisely + await self.reply_direct(context, text="All shards processed, 0 errors.") +``` + +If two agents do end up blocked waiting on each other, the network detects the +wait cycle and sends each of them an `agent.wait.deadlock_detected` event; +`send_and_wait()` recognizes it and returns `None` immediately instead of +hanging until the timeout. + ## File Management ### File Sharing Operations diff --git a/sdk/src/openagents/mods/coordination/task_delegation/mod.py b/sdk/src/openagents/mods/coordination/task_delegation/mod.py index 620f64baa..8e281efcf 100644 --- a/sdk/src/openagents/mods/coordination/task_delegation/mod.py +++ b/sdk/src/openagents/mods/coordination/task_delegation/mod.py @@ -209,6 +209,17 @@ async def _timeout_task_handler(self, task: Task): ), ) + # The task is dead locally; stop polling any external counterpart so + # the polling coroutine does not outlive it. + if self._external_delegator and delegation.get("is_external_assignee"): + try: + await self._external_delegator.stop_background_polling(task.id) + except Exception as e: + logger.warning( + f"Failed to stop external polling for timed-out task " + f"{task.id}: {e}" + ) + # Notify delegator await self._send_notification( "task.notification.timeout", @@ -876,6 +887,10 @@ async def _handle_task_complete(self, event: Event) -> Optional[EventResponse]: logger.info(f"Task {task_id} completed by {completer_id}") + # Re-extract after the update; the earlier snapshot predates + # completed_at being written. + delegation = extract_delegation_metadata(task) + # Notify delegator await self._send_notification( "task.notification.completed", diff --git a/sdk/src/openagents/sdk/connectors/grpc_connector.py b/sdk/src/openagents/sdk/connectors/grpc_connector.py index 69748cfc9..8ee81d126 100644 --- a/sdk/src/openagents/sdk/connectors/grpc_connector.py +++ b/sdk/src/openagents/sdk/connectors/grpc_connector.py @@ -339,8 +339,9 @@ async def send_event(self, message: Event) -> EventResponse: # Send event via unified gRPC SendEvent grpc_event = self._to_grpc_event(message) - # Send the event to the server using unified SendEvent - response = await self.stub.SendEvent(grpc_event) + # Send the event to the server using unified SendEvent. Without a + # deadline a stalled server would block the caller forever. + response = await self.stub.SendEvent(grpc_event, timeout=30.0) # Convert gRPC response to EventResponse response_data = None From da5e23f91dfdd68298d209cfb7096037a16b8400 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Fri, 31 Jul 2026 11:14:07 +0000 Subject: [PATCH 4/5] Address review blockers in wait-graph deadlock detection Four merge blockers from review plus two follow-up suggestions - only explicit blocking waits become wait-graph edges. send_and_wait now marks its request with blocking_wait and wait_timeout metadata; requires_response alone no longer creates an edge, so continuation and fire-and-forget requests cannot form false cycles that abort real waiters. Edges expire at the waiter's own deadline instead of a fixed TTL - deadlock notifications bypass event subscription filtering in the gateway; an agent subscribed only to thread.* still receives agent.wait.deadlock_detected instead of silently degrading to a blind timeout - wait-graph endpoints are canonicalized (agent prefix stripped) so alias forms join one graph, and a self-wait is reported as an immediate single-node cycle instead of being ignored - delegation lineage validation and task creation now run under one lock, so concurrent delegations cannot both pass the active-delegation fuse before either task is stored - DFS returns the exact edges forming the cycle and only those request ids are reported, leaving unrelated concurrent waits between the same agents untouched - react scope is bound to its owning asyncio task; background tasks spawned inside react() inherit the ContextVar but are no longer treated as inside react, so post-handler continuations cannot trigger false warnings or strict-mode errors Adds regression tests for all of the above and for the earlier hygiene fixes (completed_at now populated, external polling stopped on local timeout, gRPC SendEvent deadline). Rebased onto latest develop. --- .../mods/coordination/task_delegation/mod.py | 105 ++++++----- sdk/src/openagents/sdk/event_gateway.py | 16 +- sdk/src/openagents/sdk/react_context.py | 32 +++- sdk/src/openagents/sdk/wait_graph.py | 129 ++++++++----- sdk/src/openagents/sdk/workspace.py | 8 +- tests/grpc/test_grpc_send_event_deadline.py | 42 +++++ tests/mods/test_task_delegation.py | 65 +++++++ tests/network/test_wait_graph.py | 173 ++++++++++++++++-- tests/workspace/test_sync_wait_governance.py | 61 ++++++ 9 files changed, 514 insertions(+), 117 deletions(-) create mode 100644 tests/grpc/test_grpc_send_event_deadline.py diff --git a/sdk/src/openagents/mods/coordination/task_delegation/mod.py b/sdk/src/openagents/mods/coordination/task_delegation/mod.py index 8e281efcf..cfc98d71e 100644 --- a/sdk/src/openagents/mods/coordination/task_delegation/mod.py +++ b/sdk/src/openagents/mods/coordination/task_delegation/mod.py @@ -110,6 +110,10 @@ def __init__(self, mod_name: str = "openagents.mods.coordination.task_delegation "max_active_delegations_per_agent", DEFAULT_MAX_ACTIVE_DELEGATIONS_PER_AGENT, ) + # Serializes lineage validation with task creation so concurrent + # delegations cannot both pass the active-delegation fuse before + # either task is stored. + self._delegation_lock = asyncio.Lock() logger.info("Initializing Task Delegation network mod (A2A-compatible)") @@ -483,16 +487,6 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: data={"error": "timeout_seconds must be a positive number"}, ) - # Validate lineage (self-delegation, cycles, depth, fan-out limits) - # and derive the server-side delegation chain. - lineage_error, lineage = await self._validate_delegation_lineage( - delegator_id=delegator_id, - assignee_id=assignee_id, - parent_task_id=payload.get("parent_task_id"), - ) - if lineage_error: - return lineage_error - # Check if assignee is an external A2A agent is_external = ( self._external_delegator @@ -502,21 +496,33 @@ async def _handle_task_delegate(self, event: Event) -> Optional[EventResponse]: if is_external: external_url = self._external_delegator.resolve_external_url(assignee_id) - # Create the A2A task - task = create_delegation_task( - delegator_id=delegator_id, - assignee_id=assignee_id, - description=description, - payload=payload.get("payload", {}), - timeout_seconds=int(timeout_seconds), - is_external_assignee=is_external, - assignee_url=external_url, - **lineage, - ) + # Validate lineage (self-delegation, cycles, depth, fan-out limits) + # and create the task under one lock, so concurrent delegations + # cannot both pass the fuse before either task is stored. + async with self._delegation_lock: + lineage_error, lineage = await self._validate_delegation_lineage( + delegator_id=delegator_id, + assignee_id=assignee_id, + parent_task_id=payload.get("parent_task_id"), + ) + if lineage_error: + return lineage_error + + # Create the A2A task + task = create_delegation_task( + delegator_id=delegator_id, + assignee_id=assignee_id, + description=description, + payload=payload.get("payload", {}), + timeout_seconds=int(timeout_seconds), + is_external_assignee=is_external, + assignee_url=external_url, + **lineage, + ) - # Store the task - await self.task_store.create_task(task) - await self._save_task(task) + # Store the task + await self.task_store.create_task(task) + await self._save_task(task) logger.info( f"Task {task.id} delegated from {delegator_id} to {assignee_id}: {description}" @@ -1460,16 +1466,6 @@ async def _delegate_routed_task( """Create and delegate a task after routing.""" timeout_seconds = payload.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS) - # Routed tasks go through the same lineage validation as direct - # delegations — capability routing must not bypass cycle checks. - lineage_error, lineage = await self._validate_delegation_lineage( - delegator_id=delegator_id, - assignee_id=assignee_id, - parent_task_id=payload.get("parent_task_id"), - ) - if lineage_error: - return lineage_error - # Check if assignee is external is_external = ( self._external_delegator @@ -1479,21 +1475,34 @@ async def _delegate_routed_task( if is_external: external_url = self._external_delegator.resolve_external_url(assignee_id) - # Create the task - task = create_delegation_task( - delegator_id=delegator_id, - assignee_id=assignee_id, - description=description, - payload=payload.get("payload", {}), - timeout_seconds=int(timeout_seconds), - is_external_assignee=is_external, - assignee_url=external_url, - **lineage, - ) + # Routed tasks go through the same lineage validation as direct + # delegations — capability routing must not bypass cycle checks. + # Validation and creation share one lock so concurrent delegations + # cannot both pass the fuse before either task is stored. + async with self._delegation_lock: + lineage_error, lineage = await self._validate_delegation_lineage( + delegator_id=delegator_id, + assignee_id=assignee_id, + parent_task_id=payload.get("parent_task_id"), + ) + if lineage_error: + return lineage_error + + # Create the task + task = create_delegation_task( + delegator_id=delegator_id, + assignee_id=assignee_id, + description=description, + payload=payload.get("payload", {}), + timeout_seconds=int(timeout_seconds), + is_external_assignee=is_external, + assignee_url=external_url, + **lineage, + ) - # Store the task - await self.task_store.create_task(task) - await self._save_task(task) + # Store the task + await self.task_store.create_task(task) + await self._save_task(task) logger.info( f"Task {task.id} routed from {delegator_id} to {assignee_id}: {description}" diff --git a/sdk/src/openagents/sdk/event_gateway.py b/sdk/src/openagents/sdk/event_gateway.py index 9018e25dd..d3c22f887 100644 --- a/sdk/src/openagents/sdk/event_gateway.py +++ b/sdk/src/openagents/sdk/event_gateway.py @@ -8,7 +8,12 @@ from openagents.models.event import Event, EventSubscription from openagents.models.event_response import EventResponse from openagents.models.network_role import NetworkRole -from openagents.sdk.wait_graph import WaitGraphMonitor +from openagents.sdk.wait_graph import DEADLOCK_EVENT_NAME, WaitGraphMonitor + +# Network control events that must reach the agent regardless of its event +# subscriptions — filtering them out would silently disable the mechanism +# (e.g. a deadlock report degrading back into a blind timeout). +SUBSCRIPTION_EXEMPT_EVENT_NAMES = {DEADLOCK_EVENT_NAME} if TYPE_CHECKING: from openagents.sdk.network import AgentNetwork @@ -277,6 +282,15 @@ async def deliver_to_agent(self, event: Event, agent_id: str): logger.debug(f"Agent {agent_id} has no event queue, skipping delivery") return + # Network control events bypass subscription filtering — an agent + # subscribed only to e.g. thread.* must still receive them. + if event.event_name in SUBSCRIPTION_EXEMPT_EVENT_NAMES: + await self.agent_event_queues[agent_id].put(event) + logger.debug( + f"Delivered control event {event.event_name} to agent {agent_id}" + ) + return + # Check if agent has any subscriptions if agent_id in self.agent_subscriptions and self.agent_subscriptions[agent_id]: # Agent has subscriptions - check if event matches any of them diff --git a/sdk/src/openagents/sdk/react_context.py b/sdk/src/openagents/sdk/react_context.py index 56e646fe7..0a77d7038 100644 --- a/sdk/src/openagents/sdk/react_context.py +++ b/sdk/src/openagents/sdk/react_context.py @@ -18,6 +18,7 @@ calling agent, so no wait cycle can form. """ +import asyncio import logging import os import warnings @@ -33,6 +34,15 @@ "current_react_agent", default=None ) +# The asyncio task that owns the react() call. ContextVars are inherited by +# asyncio.create_task(), so a background task spawned inside react() would +# otherwise still look like it is "inside react()" after the handler returned; +# requiring the current task to match the owner confines the scope to the +# react() call itself. +current_react_task: ContextVar[Optional[object]] = ContextVar( + "current_react_task", default=None +) + STRICT_ENV_VAR = "OPENAGENTS_STRICT_NO_BLOCKING_WAIT" @@ -40,9 +50,19 @@ class BlockingWaitInReactError(RuntimeError): """Raised in strict mode when a blocking wait primitive runs inside react().""" +def _current_task() -> Optional[object]: + try: + return asyncio.current_task() + except RuntimeError: + return None + + def in_react() -> bool: """Return True if the current coroutine is executing inside a react() call.""" - return current_react_agent.get() is not None + return ( + current_react_agent.get() is not None + and current_react_task.get() is _current_task() + ) def strict_mode_enabled() -> bool: @@ -57,11 +77,13 @@ def strict_mode_enabled() -> bool: @contextmanager def react_scope(agent_id: str) -> Iterator[None]: """Mark the enclosed block as running inside the given agent's react().""" - token = current_react_agent.set(agent_id) + agent_token = current_react_agent.set(agent_id) + task_token = current_react_task.set(_current_task()) try: yield finally: - current_react_agent.reset(token) + current_react_task.reset(task_token) + current_react_agent.reset(agent_token) def check_blocking_wait(api_name: str, target: Optional[str] = None) -> None: @@ -72,9 +94,9 @@ def check_blocking_wait(api_name: str, target: Optional[str] = None) -> None: api_name: Name of the blocking API, for the diagnostic message target: Optional peer agent / channel the caller is waiting on """ - agent_id = current_react_agent.get() - if agent_id is None: + if not in_react(): return + agent_id = current_react_agent.get() target_desc = f" on {target!r}" if target else "" message = ( diff --git a/sdk/src/openagents/sdk/wait_graph.py b/sdk/src/openagents/sdk/wait_graph.py index 055c392ab..da38d6c39 100644 --- a/sdk/src/openagents/sdk/wait_graph.py +++ b/sdk/src/openagents/sdk/wait_graph.py @@ -1,21 +1,29 @@ """Server-side wait-for graph deadlock detection. Every event in a network flows through the central event gateway, which makes -the server the natural place to watch for agents waiting on each other. An -event sent with ``requires_response=True`` from agent A to agent B registers a -directed wait edge A -> B keyed by the request's event id; an event whose -``response_to`` references that request clears the edge. If adding an edge -closes a cycle (A waits on B while B — directly or transitively — waits on A), -every agent on the cycle is notified with an ``agent.wait.deadlock_detected`` -event so its waiter can fail fast with the cycle path instead of blocking -until its timeout with no explanation. +the server the natural place to watch for agents waiting on each other. +Blocking wait primitives (``send_and_wait``) mark their request event with +``metadata={"blocking_wait": True, "wait_timeout": }``; such an event +from agent A to agent B registers a directed wait edge A -> B keyed by the +request's event id. An event whose ``response_to`` references that request +clears the edge, and an edge expires at the waiter's own deadline, so an +abandoned wait (client crash, timeout without a response event) cannot poison +the graph. + +``requires_response`` alone deliberately does NOT create an edge: it only +means a reply is expected some time, not that the sender is blocked — a +continuation-style request must never be reported as part of a deadlock. + +If adding an edge closes a cycle (A waits on B while B — directly or +transitively — waits on A), every agent on the cycle is notified with an +``agent.wait.deadlock_detected`` event naming the cycle and the exact request +ids forming it, so the affected waiters can fail fast instead of blocking +until timeout with no explanation. Other concurrent requests between the same +agents are not named and keep waiting normally. This catches the wait cycles that delegation-chain validation cannot see: mutual ``send_and_wait`` calls, independent tasks pointing at each other, and mixed-mechanism cycles. - -Edges expire after a TTL so that abandoned waits (client crash, timeout -without a response event) cannot poison the graph forever. """ import logging @@ -28,10 +36,28 @@ DEADLOCK_EVENT_NAME = "agent.wait.deadlock_detected" +# Fallback edge lifetime when the request does not carry its own deadline. # Client-side blocking waits default to 30s; keep edges a bit longer so slow -# responses still clear them, but abandoned waits eventually vanish. +# responses still clear them. DEFAULT_EDGE_TTL_SECONDS = 60.0 +# Upper bound on any edge lifetime, whatever deadline the client claims. +MAX_EDGE_TTL_SECONDS = 3600.0 + + +def canonical_agent_id(agent_id: Optional[str]) -> str: + """Normalize an agent id for wait-graph identity comparisons. + + Strips whitespace and a leading ``agent:`` prefix so that + ``alice -> agent:bob`` and ``bob -> agent:alice`` join into one graph. + """ + if not agent_id: + return "" + normalized = str(agent_id).strip() + if normalized.startswith("agent:"): + normalized = normalized[len("agent:"):] + return normalized + class _WaitEdge: __slots__ = ("request_id", "waiter", "target", "expires_at") @@ -44,7 +70,7 @@ def __init__(self, request_id: str, waiter: str, target: str, expires_at: float) class WaitGraphMonitor: - """Observes gateway traffic and detects request/response wait cycles.""" + """Observes gateway traffic and detects blocking-wait cycles.""" def __init__( self, @@ -58,7 +84,7 @@ def __init__( network_id: Used as source_id of deadlock notifications deliver: Coroutine used to deliver a notification event to its destination agent (the gateway's deliver_event) - edge_ttl_seconds: How long an unanswered wait edge is kept + edge_ttl_seconds: Edge lifetime for requests without a deadline """ self._network_id = network_id self._deliver = deliver @@ -76,11 +102,14 @@ async def observe(self, event: Event) -> None: self._edges.pop(event.response_to, None) return - if not event.requires_response: + # Only explicitly-marked blocking waits become edges; a plain + # requires_response request does not prove the sender is blocked. + metadata = event.metadata or {} + if not metadata.get("blocking_wait"): return - waiter = event.source_id - target = event.destination_id + waiter = canonical_agent_id(event.source_id) + target = canonical_agent_id(event.destination_id) if not self._is_agent_to_agent(waiter, target): return @@ -90,23 +119,31 @@ async def observe(self, event: Event) -> None: if event.event_name.endswith(".notification"): return + ttl = metadata.get("wait_timeout") + if not isinstance(ttl, (int, float)) or ttl <= 0: + ttl = self._edge_ttl + ttl = min(ttl, MAX_EDGE_TTL_SECONDS) + self._edges[event.event_id] = _WaitEdge( request_id=event.event_id, waiter=waiter, target=target, - expires_at=time.time() + self._edge_ttl, + expires_at=time.time() + ttl, ) - cycle = self._find_cycle(waiter) - if cycle: - await self._notify_cycle(cycle) + cycle_edges = self._find_cycle(waiter) + if cycle_edges: + await self._notify_cycle(cycle_edges) def waiting_targets(self, waiter: str) -> List[str]: """Agents the given agent currently waits on (for introspection).""" + waiter = canonical_agent_id(waiter) return [e.target for e in self._edges.values() if e.waiter == waiter] - def _is_agent_to_agent(self, waiter: Optional[str], target: Optional[str]) -> bool: - if not waiter or not target or waiter == target: + def _is_agent_to_agent(self, waiter: str, target: str) -> bool: + # A self-wait (waiter == target) is a valid single-node cycle and is + # intentionally let through. + if not waiter or not target: return False for endpoint in (waiter, target): if endpoint.startswith(("channel:", "group:", "network:", "mod:")): @@ -121,40 +158,42 @@ def _purge_expired(self) -> None: for rid in expired: del self._edges[rid] - def _find_cycle(self, start: str) -> Optional[List[str]]: - """DFS from `start` over wait edges; return the cycle path if one - leads back to `start`.""" - adjacency: Dict[str, List[str]] = {} + def _find_cycle(self, start: str) -> Optional[List[_WaitEdge]]: + """DFS from `start` over wait edges; return the exact edges forming a + cycle back to `start`, or None.""" + adjacency: Dict[str, List[_WaitEdge]] = {} for edge in self._edges.values(): - adjacency.setdefault(edge.waiter, []).append(edge.target) + adjacency.setdefault(edge.waiter, []).append(edge) - path: List[str] = [start] + edge_path: List[_WaitEdge] = [] visited = set() - def dfs(node: str) -> Optional[List[str]]: - for neighbor in adjacency.get(node, []): - if neighbor == start: - return path + [start] - if neighbor in visited: + def dfs(node: str) -> Optional[List[_WaitEdge]]: + for edge in adjacency.get(node, []): + if edge.target == start: + return edge_path + [edge] + if edge.target in visited: continue - visited.add(neighbor) - path.append(neighbor) - found = dfs(neighbor) + visited.add(edge.target) + edge_path.append(edge) + found = dfs(edge.target) if found: return found - path.pop() + edge_path.pop() return None return dfs(start) - async def _notify_cycle(self, cycle: List[str]) -> None: - """Deliver a deadlock notification to every agent on the cycle.""" + async def _notify_cycle(self, cycle_edges: List[_WaitEdge]) -> None: + """Deliver a deadlock notification to every agent on the cycle. + + Only the request ids of the edges actually forming the cycle are + reported, so unrelated concurrent waits between the same agents are + not aborted. + """ + cycle = [cycle_edges[0].waiter] + [e.target for e in cycle_edges] agents = list(dict.fromkeys(cycle)) # de-duplicate, keep order - request_ids = [ - e.request_id - for e in self._edges.values() - if e.waiter in agents and e.target in agents - ] + request_ids = [e.request_id for e in cycle_edges] logger.warning( f"Wait-for cycle detected between agents: {' -> '.join(cycle)}" ) diff --git a/sdk/src/openagents/sdk/workspace.py b/sdk/src/openagents/sdk/workspace.py index cf8daf5a5..cf0be01ef 100644 --- a/sdk/src/openagents/sdk/workspace.py +++ b/sdk/src/openagents/sdk/workspace.py @@ -216,8 +216,14 @@ async def send_and_wait( logger.error("Could not establish client connection") return None + # Mark the request as a blocking wait (with its deadline) so the + # network's wait-graph can detect deadlocks. requires_response alone + # would not prove we are blocked — continuation-style requests also + # set it. + wait_metadata = {"blocking_wait": True, "wait_timeout": timeout} + wait_metadata.update(kwargs.pop("metadata", None) or {}) request = self._build_direct_message( - content, requires_response=True, **kwargs + content, requires_response=True, metadata=wait_metadata, **kwargs ) def reply_condition(msg): diff --git a/tests/grpc/test_grpc_send_event_deadline.py b/tests/grpc/test_grpc_send_event_deadline.py new file mode 100644 index 000000000..7227e442c --- /dev/null +++ b/tests/grpc/test_grpc_send_event_deadline.py @@ -0,0 +1,42 @@ +"""Regression test: gRPC SendEvent must carry a deadline. + +Without a timeout on the SendEvent RPC, a stalled server blocks the calling +agent forever — send_and_wait timeouts never fire because the send itself +never returns. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from openagents.models.event import Event +from openagents.sdk.connectors.grpc_connector import GRPCNetworkConnector + + +@pytest.mark.asyncio +async def test_send_event_uses_deadline(): + connector = GRPCNetworkConnector(host="localhost", port=50051, agent_id="alice") + connector.is_connected = True + connector.stub = MagicMock() + # Proto stubs are only loaded on connect(); the wire encoding is not what + # this test is about. + connector._to_grpc_event = MagicMock(return_value=MagicMock()) + + grpc_response = MagicMock() + grpc_response.success = True + grpc_response.message = "ok" + grpc_response.data = None + connector.stub.SendEvent = AsyncMock(return_value=grpc_response) + + event = Event( + event_name="thread.direct_message.send", + source_id="alice", + destination_id="bob", + payload={"content": {"text": "ping"}}, + ) + + await connector.send_event(event) + + connector.stub.SendEvent.assert_awaited_once() + call = connector.stub.SendEvent.await_args + assert call.kwargs.get("timeout") == 30.0 diff --git a/tests/mods/test_task_delegation.py b/tests/mods/test_task_delegation.py index d1e337aba..cab1f5b9f 100644 --- a/tests/mods/test_task_delegation.py +++ b/tests/mods/test_task_delegation.py @@ -1044,6 +1044,71 @@ async def test_active_delegation_fuse(self, task_delegation_mod): # Other delegators are unaffected. assert (await self.delegate(task_delegation_mod, "agent_b", "agent_c")).success + @pytest.mark.asyncio + async def test_active_delegation_fuse_is_atomic(self, task_delegation_mod): + """Concurrent delegations must not both slip past the fuse before + either task is stored.""" + import asyncio + + task_delegation_mod._max_active_delegations_per_agent = 1 + + results = await asyncio.gather( + self.delegate(task_delegation_mod, "agent_a", "agent_b"), + self.delegate(task_delegation_mod, "agent_a", "agent_c"), + ) + + successes = [r for r in results if r.success] + rejections = [r for r in results if not r.success] + assert len(successes) == 1 + assert len(rejections) == 1 + assert rejections[0].data["error"] == "delegation_limit_exceeded" + + @pytest.mark.asyncio + async def test_completed_task_reports_completed_at(self, task_delegation_mod): + """Regression: the completion response used to report completed_at as + None because it was read from a pre-update snapshot.""" + delegated = await self.delegate( + task_delegation_mod, "agent_alice", "agent_bob" + ) + complete_event = Event( + event_name="task.complete", + source_id="agent_bob", + payload={"task_id": delegated.data["task_id"], "result": {"ok": True}}, + ) + + response = await task_delegation_mod._handle_task_complete(complete_event) + + assert response.success is True + assert response.data["completed_at"] is not None + + @pytest.mark.asyncio + async def test_timeout_stops_external_polling(self, task_delegation_mod): + """Regression: a locally timed-out external task kept its polling + loop alive until process shutdown.""" + from unittest.mock import AsyncMock, MagicMock + + from openagents.mods.coordination.task_delegation.a2a_delegation import ( + create_delegation_task, + ) + + task = create_delegation_task( + delegator_id="agent_alice", + assignee_id="a2a:remote-agent", + description="external work", + timeout_seconds=1, + is_external_assignee=True, + assignee_url="https://remote.example.com", + ) + await task_delegation_mod.task_store.create_task(task) + + external = MagicMock() + external.stop_background_polling = AsyncMock() + task_delegation_mod._external_delegator = external + + await task_delegation_mod._timeout_task_handler(task) + + external.stop_background_polling.assert_awaited_once_with(task.id) + @pytest.mark.asyncio async def test_routed_delegation_checks_lineage(self, task_delegation_mod): """Capability routing must not bypass cycle checks.""" diff --git a/tests/network/test_wait_graph.py b/tests/network/test_wait_graph.py index 89f865cd0..23b13ce98 100644 --- a/tests/network/test_wait_graph.py +++ b/tests/network/test_wait_graph.py @@ -1,17 +1,38 @@ """Tests for the server-side wait-for graph deadlock detection.""" +import asyncio +from unittest.mock import MagicMock + import pytest -from openagents.models.event import Event +from openagents.models.event import Event, EventSubscription from openagents.sdk.wait_graph import DEADLOCK_EVENT_NAME, WaitGraphMonitor -def request_event(source: str, target: str, name: str = "thread.direct_message.send") -> Event: +def blocking_request( + source: str, + target: str, + name: str = "thread.direct_message.send", + wait_timeout: float = 30.0, +) -> Event: + """A request sent by a blocking wait primitive (send_and_wait).""" return Event( event_name=name, source_id=source, destination_id=target, requires_response=True, + metadata={"blocking_wait": True, "wait_timeout": wait_timeout}, + payload={}, + ) + + +def continuation_request(source: str, target: str) -> Event: + """A fire-and-forget request that expects a reply but does not block.""" + return Event( + event_name="thread.direct_message.send", + source_id=source, + destination_id=target, + requires_response=True, payload={}, ) @@ -41,23 +62,44 @@ async def deliver(event): class TestWaitGraphMonitor: async def test_single_wait_no_deadlock(self, monitor, delivered): - await monitor.observe(request_event("alice", "bob")) + await monitor.observe(blocking_request("alice", "bob")) assert delivered == [] assert monitor.waiting_targets("alice") == ["bob"] + async def test_fire_and_forget_requests_create_no_edges( + self, monitor, delivered + ): + """requires_response alone does not prove the sender is blocked; a + continuation request must never join a deadlock report.""" + await monitor.observe(continuation_request("alice", "bob")) + assert monitor.waiting_targets("alice") == [] + + # Even opposing continuation requests report nothing. + await monitor.observe(continuation_request("bob", "alice")) + assert delivered == [] + + async def test_continuation_does_not_falsify_real_wait( + self, monitor, delivered + ): + """A blocking wait one way plus a continuation the other way is not + a deadlock — the continuation sender is free to answer.""" + await monitor.observe(blocking_request("alice", "bob")) + await monitor.observe(continuation_request("bob", "alice")) + assert delivered == [] + async def test_response_clears_the_edge(self, monitor, delivered): - request = request_event("alice", "bob") + request = blocking_request("alice", "bob") await monitor.observe(request) await monitor.observe(response_event("bob", "alice", request.event_id)) assert monitor.waiting_targets("alice") == [] # bob can now wait on alice without a (stale) cycle being reported - await monitor.observe(request_event("bob", "alice")) + await monitor.observe(blocking_request("bob", "alice")) assert delivered == [] async def test_mutual_wait_detected_and_both_notified(self, monitor, delivered): - request_ab = request_event("alice", "bob") - request_ba = request_event("bob", "alice") + request_ab = blocking_request("alice", "bob") + request_ba = blocking_request("bob", "alice") await monitor.observe(request_ab) await monitor.observe(request_ba) @@ -72,33 +114,74 @@ async def test_mutual_wait_detected_and_both_notified(self, monitor, delivered): } assert event.payload["cycle"][0] == event.payload["cycle"][-1] + async def test_agent_prefix_aliases_join_one_graph(self, monitor, delivered): + """alice -> agent:bob and bob -> agent:alice must close a cycle.""" + await monitor.observe(blocking_request("alice", "agent:bob")) + await monitor.observe(blocking_request("bob", "agent:alice")) + + notified = {event.destination_id for event in delivered} + assert notified == {"alice", "bob"} + + async def test_self_wait_reported_as_cycle(self, monitor, delivered): + """An agent blocking on itself can never be answered.""" + await monitor.observe(blocking_request("alice", "agent:alice")) + + assert len(delivered) == 1 + assert delivered[0].destination_id == "alice" + assert delivered[0].payload["cycle"] == ["alice", "alice"] + async def test_three_agent_cycle_detected(self, monitor, delivered): - await monitor.observe(request_event("alice", "bob")) - await monitor.observe(request_event("bob", "charlie")) + await monitor.observe(blocking_request("alice", "bob")) + await monitor.observe(blocking_request("bob", "charlie")) assert delivered == [] - await monitor.observe(request_event("charlie", "alice")) + await monitor.observe(blocking_request("charlie", "alice")) notified = {event.destination_id for event in delivered} assert notified == {"alice", "bob", "charlie"} + async def test_only_cycle_edges_are_reported(self, monitor, delivered): + """An unrelated concurrent request between the same agents must not + be named in the report (its waiter would abort for no reason).""" + in_cycle = blocking_request("alice", "bob") + unrelated = blocking_request("alice", "bob") + closes_cycle = blocking_request("bob", "alice") + + await monitor.observe(in_cycle) + await monitor.observe(unrelated) + await monitor.observe(closes_cycle) + + assert delivered + request_ids = set(delivered[0].payload["request_ids"]) + assert closes_cycle.event_id in request_ids + assert len(request_ids) == 2 + assert unrelated.event_id not in request_ids + async def test_no_false_positive_on_shared_target(self, monitor, delivered): - await monitor.observe(request_event("alice", "bob")) - await monitor.observe(request_event("charlie", "bob")) + await monitor.observe(blocking_request("alice", "bob")) + await monitor.observe(blocking_request("charlie", "bob")) assert delivered == [] async def test_notification_relays_do_not_create_edges(self, monitor, delivered): - relay = request_event( + relay = blocking_request( "alice", "bob", name="thread.direct_message.notification" ) await monitor.observe(relay) assert monitor.waiting_targets("alice") == [] async def test_channel_and_broadcast_traffic_ignored(self, monitor, delivered): - await monitor.observe(request_event("alice", "channel:general")) - await monitor.observe(request_event("alice", "agent:broadcast")) + await monitor.observe(blocking_request("alice", "channel:general")) + await monitor.observe(blocking_request("alice", "agent:broadcast")) assert monitor.waiting_targets("alice") == [] + async def test_edges_expire_at_the_waiter_deadline(self, monitor, delivered): + """The edge lives only as long as the wait it represents — after the + waiter's own timeout it must not report cycles anymore.""" + await monitor.observe(blocking_request("alice", "bob", wait_timeout=0.01)) + await asyncio.sleep(0.05) + await monitor.observe(blocking_request("bob", "alice")) + assert delivered == [] + async def test_expired_edges_are_purged(self, delivered): async def deliver(event): delivered.append(event) @@ -106,7 +189,63 @@ async def deliver(event): monitor = WaitGraphMonitor( network_id="test-network", deliver=deliver, edge_ttl_seconds=0.0 ) - await monitor.observe(request_event("alice", "bob")) + # No per-request deadline: falls back to the (zero) default TTL. + request = blocking_request("alice", "bob") + request.metadata = {"blocking_wait": True} + await monitor.observe(request) # The edge expired immediately; the reverse wait closes no cycle. - await monitor.observe(request_event("bob", "alice")) + reverse = blocking_request("bob", "alice") + reverse.metadata = {"blocking_wait": True} + await monitor.observe(reverse) assert delivered == [] + + +class TestDeadlockNotificationDelivery: + """Deadlock reports must reach the agent whatever it subscribed to.""" + + def make_gateway(self): + from openagents.sdk.event_gateway import EventGateway + + network = MagicMock() + network.network_id = "test-network" + network.mods = {} + return EventGateway(network) + + async def test_deadlock_event_bypasses_subscription_filter(self): + gateway = self.make_gateway() + gateway.register_agent("alice") + # alice only subscribed to thread.* events + gateway.agent_subscriptions["alice"] = [ + EventSubscription(agent_id="alice", event_patterns=["thread.*"]) + ] + + deadlock = Event( + event_name=DEADLOCK_EVENT_NAME, + source_id="test-network", + destination_id="alice", + payload={"cycle": ["alice", "bob", "alice"], "request_ids": []}, + ) + await gateway.deliver_to_agent(deadlock, "alice") + + events = [ + gateway.agent_event_queues["alice"].get_nowait() + for _ in range(gateway.agent_event_queues["alice"].qsize()) + ] + assert [e.event_name for e in events] == [DEADLOCK_EVENT_NAME] + + async def test_ordinary_events_still_filtered(self): + gateway = self.make_gateway() + gateway.register_agent("alice") + gateway.agent_subscriptions["alice"] = [ + EventSubscription(agent_id="alice", event_patterns=["thread.*"]) + ] + + unrelated = Event( + event_name="feed.post.created", + source_id="bob", + destination_id="alice", + payload={}, + ) + await gateway.deliver_to_agent(unrelated, "alice") + + assert gateway.agent_event_queues["alice"].qsize() == 0 diff --git a/tests/workspace/test_sync_wait_governance.py b/tests/workspace/test_sync_wait_governance.py index ec5322579..e2d94d8ab 100644 --- a/tests/workspace/test_sync_wait_governance.py +++ b/tests/workspace/test_sync_wait_governance.py @@ -228,6 +228,29 @@ async def network_reports_deadlock(): assert reply is None + async def test_request_is_marked_as_blocking_wait(self): + """send_and_wait must tag its request so the network wait-graph can + tell real blocking waits apart from continuation requests.""" + workspace = make_workspace() + + sent_events = [] + + async def capture_send(event): + sent_events.append(event) + return EventResponse(success=True, message="ok") + + workspace.send_event = AsyncMock(side_effect=capture_send) + + await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.05) + + assert len(sent_events) == 1 + metadata = sent_events[0].metadata + assert metadata["blocking_wait"] is True + assert metadata["wait_timeout"] == 0.05 + # A plain send() carries no such marker. + await workspace.agent(AGENT_B).send("fire and forget") + assert not (sent_events[1].metadata or {}).get("blocking_wait") + async def test_send_failure_returns_none_and_deregisters_waiter(self): workspace = make_workspace() workspace.send_event = AsyncMock( @@ -361,6 +384,44 @@ def test_react_scope_sets_and_resets_context(self): assert current_react_agent.get() == AGENT_A assert current_react_agent.get() is None + async def test_background_task_spawned_in_react_is_not_in_react(self): + """ContextVars are inherited by asyncio.create_task(); the react scope + must not leak into background continuations spawned by a handler.""" + from openagents.sdk.react_context import in_react + + results = {} + + async def background_work(): + results["background_in_react"] = in_react() + + with react_scope(AGENT_A): + results["handler_in_react"] = in_react() + task = asyncio.create_task(background_work()) + await task + + assert results["handler_in_react"] is True + assert results["background_in_react"] is False + + async def test_blocking_wait_in_background_task_does_not_warn(self, recwarn): + """A wait inside a background task spawned from react() is legal — + it does not stall the runner loop.""" + workspace = make_workspace() + workspace.send_event = AsyncMock( + return_value=EventResponse(success=True, message="ok") + ) + + async def background_wait(): + await workspace.agent(AGENT_B).send_and_wait("ping", timeout=0.05) + + with react_scope(AGENT_A): + task = asyncio.create_task(background_wait()) + await task + + deprecations = [ + w for w in recwarn.list if issubclass(w.category, DeprecationWarning) + ] + assert deprecations == [] + class TestMessagingModCorrelationRelay: """The mod must preserve correlation fields when relaying direct messages.""" From 64b3c38cae1767370a44add8e093367bb6858dbd Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Fri, 31 Jul 2026 11:24:45 +0000 Subject: [PATCH 5/5] Preserve correlation over gRPC and clear resolved wait-graph edges Second-round review blockers - the gRPC Event proto had no requires_response / response_to fields, so both were dropped on ingress. Precise reply matching fell back to the legacy source-only heuristic and wait-graph edges were never cleared by responses from gRPC agents. Added the fields to the proto (backward-compatible tag numbers 11 and 12), regenerated the pb2 modules, set them in the client-side conversion and read them defensively in a new internal_event_from_grpc helper on the server - gRPC metadata is transported as map, which stringified wait markers. The wait-graph now parses metadata leniently, the string False no longer counts as a blocking wait and numeric strings keep their value as the edge deadline - edges forming a detected cycle are removed before the notification is delivered. The notified waiters return immediately, so keeping their edges alive until the original timeout made later unrelated waits between the same agents look like fresh deadlocks - send_and_wait now applies its internal blocking_wait / wait_timeout metadata after caller-supplied metadata, so a caller cannot disable detection or forge the deadline Adds an Event to gRPC to Event roundtrip test for the correlation fields, plus wait-graph tests for stringified metadata and for post-notification edge cleanup. --- sdk/src/openagents/proto/agent_service.proto | 2 + sdk/src/openagents/proto/agent_service_pb2.py | 76 +++++++++---------- .../proto/agent_service_pb2_grpc.py | 2 +- .../sdk/connectors/grpc_connector.py | 2 + sdk/src/openagents/sdk/transports/grpc.py | 35 ++++++--- sdk/src/openagents/sdk/wait_graph.py | 33 +++++++- sdk/src/openagents/sdk/workspace.py | 8 +- tests/grpc/test_grpc_event_roundtrip.py | 69 +++++++++++++++++ tests/network/test_wait_graph.py | 33 ++++++++ 9 files changed, 204 insertions(+), 56 deletions(-) create mode 100644 tests/grpc/test_grpc_event_roundtrip.py diff --git a/sdk/src/openagents/proto/agent_service.proto b/sdk/src/openagents/proto/agent_service.proto index 1b83941cf..5606e2bde 100644 --- a/sdk/src/openagents/proto/agent_service.proto +++ b/sdk/src/openagents/proto/agent_service.proto @@ -37,6 +37,8 @@ message Event { string visibility = 8; // "local", "network", "external" string relevant_mod = 9; // Specific mod to process this event (for MOD_ONLY visibility) string secret = 10; // Authentication secret for the source agent + bool requires_response = 11; // Whether this event expects a response + string response_to = 12; // If this is a response, the original event_id } // Unified event response diff --git a/sdk/src/openagents/proto/agent_service_pb2.py b/sdk/src/openagents/proto/agent_service_pb2.py index 3179f7bc8..fd6b9d59f 100644 --- a/sdk/src/openagents/proto/agent_service_pb2.py +++ b/sdk/src/openagents/proto/agent_service_pb2.py @@ -25,7 +25,7 @@ from google.protobuf import any_pb2 as google_dot_protobuf_dot_any__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x13\x61gent_service.proto\x12\nopenagents\x1a\x19google/protobuf/any.proto\"\xb1\x02\n\x05\x45vent\x12\x10\n\x08\x65vent_id\x18\x01 \x01(\t\x12\x12\n\nevent_name\x18\x02 \x01(\t\x12\x11\n\tsource_id\x18\x03 \x01(\t\x12\x17\n\x0ftarget_agent_id\x18\x04 \x01(\t\x12%\n\x07payload\x18\x05 \x01(\x0b\x32\x14.google.protobuf.Any\x12\x11\n\ttimestamp\x18\x06 \x01(\x03\x12\x31\n\x08metadata\x18\x07 \x03(\x0b\x32\x1f.openagents.Event.MetadataEntry\x12\x12\n\nvisibility\x18\x08 \x01(\t\x12\x14\n\x0crelevant_mod\x18\t \x01(\t\x12\x0e\n\x06secret\x18\n \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"i\n\rEventResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\x12\"\n\x04\x64\x61ta\x18\x03 \x01(\x0b\x32\x14.google.protobuf.Any\x12\x12\n\nevent_name\x18\x04 \x01(\t\"\xf6\x01\n\x14RegisterAgentRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12@\n\x08metadata\x18\x02 \x03(\x0b\x32..openagents.RegisterAgentRequest.MetadataEntry\x12\x14\n\x0c\x63\x61pabilities\x18\x03 \x03(\t\x12\x17\n\x0f\x66orce_reconnect\x18\x04 \x01(\x08\x12\x13\n\x0b\x63\x65rtificate\x18\x05 \x01(\t\x12\x15\n\rpassword_hash\x18\x06 \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"O\n\x15RegisterAgentResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\x12\x0e\n\x06secret\x18\x03 \x01(\t\":\n\x16UnregisterAgentRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x0e\n\x06secret\x18\x02 \x01(\t\"A\n\x17UnregisterAgentResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\"6\n\x15\x44iscoverAgentsRequest\x12\x1d\n\x15required_capabilities\x18\x01 \x03(\t\"?\n\x16\x44iscoverAgentsResponse\x12%\n\x06\x61gents\x18\x01 \x03(\x0b\x32\x15.openagents.AgentInfo\"\'\n\x13GetAgentInfoRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\"K\n\x14GetAgentInfoResponse\x12$\n\x05\x61gent\x18\x01 \x01(\x0b\x32\x15.openagents.AgentInfo\x12\r\n\x05\x66ound\x18\x02 \x01(\x08\"\xd7\x01\n\tAgentInfo\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x35\n\x08metadata\x18\x02 \x03(\x0b\x32#.openagents.AgentInfo.MetadataEntry\x12\x14\n\x0c\x63\x61pabilities\x18\x03 \x03(\t\x12\x11\n\tlast_seen\x18\x04 \x01(\x03\x12\x16\n\x0etransport_type\x18\x05 \x01(\t\x12\x0f\n\x07\x61\x64\x64ress\x18\x06 \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"7\n\x10HeartbeatRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\"7\n\x11HeartbeatResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\"\x14\n\x12NetworkInfoRequest\"\xf2\x01\n\x13NetworkInfoResponse\x12\x12\n\nnetwork_id\x18\x01 \x01(\t\x12\x14\n\x0cnetwork_name\x18\x02 \x01(\t\x12\x12\n\nis_running\x18\x03 \x01(\x08\x12\x16\n\x0euptime_seconds\x18\x04 \x01(\x03\x12\x13\n\x0b\x61gent_count\x18\x05 \x01(\x05\x12%\n\x06\x61gents\x18\x06 \x03(\x0b\x32\x15.openagents.AgentInfo\x12\x15\n\rtopology_mode\x18\x07 \x01(\t\x12\x16\n\x0etransport_type\x18\x08 \x01(\t\x12\x0c\n\x04host\x18\t \x01(\t\x12\x0c\n\x04port\x18\n \x01(\x05\x32\xfe\x04\n\x0c\x41gentService\x12\x39\n\tSendEvent\x12\x11.openagents.Event\x1a\x19.openagents.EventResponse\x12\x38\n\x0cStreamEvents\x12\x11.openagents.Event\x1a\x11.openagents.Event(\x01\x30\x01\x12T\n\rRegisterAgent\x12 .openagents.RegisterAgentRequest\x1a!.openagents.RegisterAgentResponse\x12Z\n\x0fUnregisterAgent\x12\".openagents.UnregisterAgentRequest\x1a#.openagents.UnregisterAgentResponse\x12W\n\x0e\x44iscoverAgents\x12!.openagents.DiscoverAgentsRequest\x1a\".openagents.DiscoverAgentsResponse\x12Q\n\x0cGetAgentInfo\x12\x1f.openagents.GetAgentInfoRequest\x1a .openagents.GetAgentInfoResponse\x12H\n\tHeartbeat\x12\x1c.openagents.HeartbeatRequest\x1a\x1d.openagents.HeartbeatResponse\x12Q\n\x0eGetNetworkInfo\x12\x1e.openagents.NetworkInfoRequest\x1a\x1f.openagents.NetworkInfoResponseb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x13\x61gent_service.proto\x12\nopenagents\x1a\x19google/protobuf/any.proto\"\xe1\x02\n\x05\x45vent\x12\x10\n\x08\x65vent_id\x18\x01 \x01(\t\x12\x12\n\nevent_name\x18\x02 \x01(\t\x12\x11\n\tsource_id\x18\x03 \x01(\t\x12\x17\n\x0ftarget_agent_id\x18\x04 \x01(\t\x12%\n\x07payload\x18\x05 \x01(\x0b\x32\x14.google.protobuf.Any\x12\x11\n\ttimestamp\x18\x06 \x01(\x03\x12\x31\n\x08metadata\x18\x07 \x03(\x0b\x32\x1f.openagents.Event.MetadataEntry\x12\x12\n\nvisibility\x18\x08 \x01(\t\x12\x14\n\x0crelevant_mod\x18\t \x01(\t\x12\x0e\n\x06secret\x18\n \x01(\t\x12\x19\n\x11requires_response\x18\x0b \x01(\x08\x12\x13\n\x0bresponse_to\x18\x0c \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"i\n\rEventResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\x12\"\n\x04\x64\x61ta\x18\x03 \x01(\x0b\x32\x14.google.protobuf.Any\x12\x12\n\nevent_name\x18\x04 \x01(\t\"\xf6\x01\n\x14RegisterAgentRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12@\n\x08metadata\x18\x02 \x03(\x0b\x32..openagents.RegisterAgentRequest.MetadataEntry\x12\x14\n\x0c\x63\x61pabilities\x18\x03 \x03(\t\x12\x17\n\x0f\x66orce_reconnect\x18\x04 \x01(\x08\x12\x13\n\x0b\x63\x65rtificate\x18\x05 \x01(\t\x12\x15\n\rpassword_hash\x18\x06 \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"O\n\x15RegisterAgentResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\x12\x0e\n\x06secret\x18\x03 \x01(\t\":\n\x16UnregisterAgentRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x0e\n\x06secret\x18\x02 \x01(\t\"A\n\x17UnregisterAgentResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x15\n\rerror_message\x18\x02 \x01(\t\"6\n\x15\x44iscoverAgentsRequest\x12\x1d\n\x15required_capabilities\x18\x01 \x03(\t\"?\n\x16\x44iscoverAgentsResponse\x12%\n\x06\x61gents\x18\x01 \x03(\x0b\x32\x15.openagents.AgentInfo\"\'\n\x13GetAgentInfoRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\"K\n\x14GetAgentInfoResponse\x12$\n\x05\x61gent\x18\x01 \x01(\x0b\x32\x15.openagents.AgentInfo\x12\r\n\x05\x66ound\x18\x02 \x01(\x08\"\xd7\x01\n\tAgentInfo\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x35\n\x08metadata\x18\x02 \x03(\x0b\x32#.openagents.AgentInfo.MetadataEntry\x12\x14\n\x0c\x63\x61pabilities\x18\x03 \x03(\t\x12\x11\n\tlast_seen\x18\x04 \x01(\x03\x12\x16\n\x0etransport_type\x18\x05 \x01(\t\x12\x0f\n\x07\x61\x64\x64ress\x18\x06 \x01(\t\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"7\n\x10HeartbeatRequest\x12\x10\n\x08\x61gent_id\x18\x01 \x01(\t\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\"7\n\x11HeartbeatResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\"\x14\n\x12NetworkInfoRequest\"\xf2\x01\n\x13NetworkInfoResponse\x12\x12\n\nnetwork_id\x18\x01 \x01(\t\x12\x14\n\x0cnetwork_name\x18\x02 \x01(\t\x12\x12\n\nis_running\x18\x03 \x01(\x08\x12\x16\n\x0euptime_seconds\x18\x04 \x01(\x03\x12\x13\n\x0b\x61gent_count\x18\x05 \x01(\x05\x12%\n\x06\x61gents\x18\x06 \x03(\x0b\x32\x15.openagents.AgentInfo\x12\x15\n\rtopology_mode\x18\x07 \x01(\t\x12\x16\n\x0etransport_type\x18\x08 \x01(\t\x12\x0c\n\x04host\x18\t \x01(\t\x12\x0c\n\x04port\x18\n \x01(\x05\x32\xfe\x04\n\x0c\x41gentService\x12\x39\n\tSendEvent\x12\x11.openagents.Event\x1a\x19.openagents.EventResponse\x12\x38\n\x0cStreamEvents\x12\x11.openagents.Event\x1a\x11.openagents.Event(\x01\x30\x01\x12T\n\rRegisterAgent\x12 .openagents.RegisterAgentRequest\x1a!.openagents.RegisterAgentResponse\x12Z\n\x0fUnregisterAgent\x12\".openagents.UnregisterAgentRequest\x1a#.openagents.UnregisterAgentResponse\x12W\n\x0e\x44iscoverAgents\x12!.openagents.DiscoverAgentsRequest\x1a\".openagents.DiscoverAgentsResponse\x12Q\n\x0cGetAgentInfo\x12\x1f.openagents.GetAgentInfoRequest\x1a .openagents.GetAgentInfoResponse\x12H\n\tHeartbeat\x12\x1c.openagents.HeartbeatRequest\x1a\x1d.openagents.HeartbeatResponse\x12Q\n\x0eGetNetworkInfo\x12\x1e.openagents.NetworkInfoRequest\x1a\x1f.openagents.NetworkInfoResponseb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -39,41 +39,41 @@ _globals['_AGENTINFO_METADATAENTRY']._loaded_options = None _globals['_AGENTINFO_METADATAENTRY']._serialized_options = b'8\001' _globals['_EVENT']._serialized_start=63 - _globals['_EVENT']._serialized_end=368 - _globals['_EVENT_METADATAENTRY']._serialized_start=321 - _globals['_EVENT_METADATAENTRY']._serialized_end=368 - _globals['_EVENTRESPONSE']._serialized_start=370 - _globals['_EVENTRESPONSE']._serialized_end=475 - _globals['_REGISTERAGENTREQUEST']._serialized_start=478 - _globals['_REGISTERAGENTREQUEST']._serialized_end=724 - _globals['_REGISTERAGENTREQUEST_METADATAENTRY']._serialized_start=321 - _globals['_REGISTERAGENTREQUEST_METADATAENTRY']._serialized_end=368 - _globals['_REGISTERAGENTRESPONSE']._serialized_start=726 - _globals['_REGISTERAGENTRESPONSE']._serialized_end=805 - _globals['_UNREGISTERAGENTREQUEST']._serialized_start=807 - _globals['_UNREGISTERAGENTREQUEST']._serialized_end=865 - _globals['_UNREGISTERAGENTRESPONSE']._serialized_start=867 - _globals['_UNREGISTERAGENTRESPONSE']._serialized_end=932 - _globals['_DISCOVERAGENTSREQUEST']._serialized_start=934 - _globals['_DISCOVERAGENTSREQUEST']._serialized_end=988 - _globals['_DISCOVERAGENTSRESPONSE']._serialized_start=990 - _globals['_DISCOVERAGENTSRESPONSE']._serialized_end=1053 - _globals['_GETAGENTINFOREQUEST']._serialized_start=1055 - _globals['_GETAGENTINFOREQUEST']._serialized_end=1094 - _globals['_GETAGENTINFORESPONSE']._serialized_start=1096 - _globals['_GETAGENTINFORESPONSE']._serialized_end=1171 - _globals['_AGENTINFO']._serialized_start=1174 - _globals['_AGENTINFO']._serialized_end=1389 - _globals['_AGENTINFO_METADATAENTRY']._serialized_start=321 - _globals['_AGENTINFO_METADATAENTRY']._serialized_end=368 - _globals['_HEARTBEATREQUEST']._serialized_start=1391 - _globals['_HEARTBEATREQUEST']._serialized_end=1446 - _globals['_HEARTBEATRESPONSE']._serialized_start=1448 - _globals['_HEARTBEATRESPONSE']._serialized_end=1503 - _globals['_NETWORKINFOREQUEST']._serialized_start=1505 - _globals['_NETWORKINFOREQUEST']._serialized_end=1525 - _globals['_NETWORKINFORESPONSE']._serialized_start=1528 - _globals['_NETWORKINFORESPONSE']._serialized_end=1770 - _globals['_AGENTSERVICE']._serialized_start=1773 - _globals['_AGENTSERVICE']._serialized_end=2411 + _globals['_EVENT']._serialized_end=416 + _globals['_EVENT_METADATAENTRY']._serialized_start=369 + _globals['_EVENT_METADATAENTRY']._serialized_end=416 + _globals['_EVENTRESPONSE']._serialized_start=418 + _globals['_EVENTRESPONSE']._serialized_end=523 + _globals['_REGISTERAGENTREQUEST']._serialized_start=526 + _globals['_REGISTERAGENTREQUEST']._serialized_end=772 + _globals['_REGISTERAGENTREQUEST_METADATAENTRY']._serialized_start=369 + _globals['_REGISTERAGENTREQUEST_METADATAENTRY']._serialized_end=416 + _globals['_REGISTERAGENTRESPONSE']._serialized_start=774 + _globals['_REGISTERAGENTRESPONSE']._serialized_end=853 + _globals['_UNREGISTERAGENTREQUEST']._serialized_start=855 + _globals['_UNREGISTERAGENTREQUEST']._serialized_end=913 + _globals['_UNREGISTERAGENTRESPONSE']._serialized_start=915 + _globals['_UNREGISTERAGENTRESPONSE']._serialized_end=980 + _globals['_DISCOVERAGENTSREQUEST']._serialized_start=982 + _globals['_DISCOVERAGENTSREQUEST']._serialized_end=1036 + _globals['_DISCOVERAGENTSRESPONSE']._serialized_start=1038 + _globals['_DISCOVERAGENTSRESPONSE']._serialized_end=1101 + _globals['_GETAGENTINFOREQUEST']._serialized_start=1103 + _globals['_GETAGENTINFOREQUEST']._serialized_end=1142 + _globals['_GETAGENTINFORESPONSE']._serialized_start=1144 + _globals['_GETAGENTINFORESPONSE']._serialized_end=1219 + _globals['_AGENTINFO']._serialized_start=1222 + _globals['_AGENTINFO']._serialized_end=1437 + _globals['_AGENTINFO_METADATAENTRY']._serialized_start=369 + _globals['_AGENTINFO_METADATAENTRY']._serialized_end=416 + _globals['_HEARTBEATREQUEST']._serialized_start=1439 + _globals['_HEARTBEATREQUEST']._serialized_end=1494 + _globals['_HEARTBEATRESPONSE']._serialized_start=1496 + _globals['_HEARTBEATRESPONSE']._serialized_end=1551 + _globals['_NETWORKINFOREQUEST']._serialized_start=1553 + _globals['_NETWORKINFOREQUEST']._serialized_end=1573 + _globals['_NETWORKINFORESPONSE']._serialized_start=1576 + _globals['_NETWORKINFORESPONSE']._serialized_end=1818 + _globals['_AGENTSERVICE']._serialized_start=1821 + _globals['_AGENTSERVICE']._serialized_end=2459 # @@protoc_insertion_point(module_scope) diff --git a/sdk/src/openagents/proto/agent_service_pb2_grpc.py b/sdk/src/openagents/proto/agent_service_pb2_grpc.py index 8713c8b5d..d611bdff8 100644 --- a/sdk/src/openagents/proto/agent_service_pb2_grpc.py +++ b/sdk/src/openagents/proto/agent_service_pb2_grpc.py @@ -5,7 +5,7 @@ from . import agent_service_pb2 as agent__service__pb2 -GRPC_GENERATED_VERSION = '1.74.0' +GRPC_GENERATED_VERSION = '1.75.1' GRPC_VERSION = grpc.__version__ _version_not_supported = False diff --git a/sdk/src/openagents/sdk/connectors/grpc_connector.py b/sdk/src/openagents/sdk/connectors/grpc_connector.py index 8ee81d126..106d03329 100644 --- a/sdk/src/openagents/sdk/connectors/grpc_connector.py +++ b/sdk/src/openagents/sdk/connectors/grpc_connector.py @@ -529,6 +529,8 @@ def _to_grpc_event(self, event: Event): visibility=event.visibility if hasattr(event, "visibility") else "network", relevant_mod=event.relevant_mod or "", secret=getattr(event, 'secret', '') or '', + requires_response=bool(event.requires_response), + response_to=event.response_to or "", ) # Add metadata diff --git a/sdk/src/openagents/sdk/transports/grpc.py b/sdk/src/openagents/sdk/transports/grpc.py index 58da5c414..e5f144dfc 100644 --- a/sdk/src/openagents/sdk/transports/grpc.py +++ b/sdk/src/openagents/sdk/transports/grpc.py @@ -33,6 +33,29 @@ logger = logging.getLogger(__name__) +def internal_event_from_grpc(request, payload: Dict[str, Any]) -> Event: + """Build an internal Event from a gRPC Event message. + + Correlation fields (requires_response / response_to) must survive the + transport — reply matching and wait-graph edge clearing depend on them. + getattr guards keep compatibility with clients built from an older proto + that lacks the fields. + """ + return Event( + event_name=request.event_name, + source_id=request.source_id, + destination_id=request.target_agent_id or None, + payload=payload, + event_id=request.event_id, + timestamp=request.timestamp if request.timestamp else int(time.time()), + metadata=dict(request.metadata) if request.metadata else {}, + visibility=request.visibility if request.visibility else "network", + secret=request.secret if hasattr(request, 'secret') else None, + requires_response=bool(getattr(request, "requires_response", False)), + response_to=getattr(request, "response_to", "") or None, + ) + + class OpenAgentsGRPCServicer(agent_service_pb2_grpc.AgentServiceServicer): """gRPC servicer for the OpenAgents transport.""" @@ -50,17 +73,7 @@ async def SendEvent(self, request, context): payload = self._extract_payload_from_protobuf(request.payload) # Create internal Event from gRPC Event - event = Event( - event_name=request.event_name, - source_id=request.source_id, - destination_id=request.target_agent_id or None, - payload=payload, - event_id=request.event_id, - timestamp=request.timestamp if request.timestamp else int(time.time()), - metadata=dict(request.metadata) if request.metadata else {}, - visibility=request.visibility if request.visibility else "network", - secret=request.secret if hasattr(request, 'secret') else None, - ) + event = internal_event_from_grpc(request, payload) # Route through unified handler event_response = await self._handle_sent_event(event) diff --git a/sdk/src/openagents/sdk/wait_graph.py b/sdk/src/openagents/sdk/wait_graph.py index da38d6c39..47bfa091a 100644 --- a/sdk/src/openagents/sdk/wait_graph.py +++ b/sdk/src/openagents/sdk/wait_graph.py @@ -45,6 +45,28 @@ MAX_EDGE_TTL_SECONDS = 3600.0 +def _metadata_flag(value) -> bool: + """Interpret a metadata value as a boolean. + + Transports may stringify metadata (gRPC uses map), so + "False"/"0" must not count as truthy. + """ + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on") + return bool(value) + + +def _metadata_number(value) -> Optional[float]: + """Interpret a metadata value as a finite positive number, else None.""" + try: + number = float(value) + except (TypeError, ValueError): + return None + if number <= 0 or number != number or number == float("inf"): + return None + return number + + def canonical_agent_id(agent_id: Optional[str]) -> str: """Normalize an agent id for wait-graph identity comparisons. @@ -105,7 +127,7 @@ async def observe(self, event: Event) -> None: # Only explicitly-marked blocking waits become edges; a plain # requires_response request does not prove the sender is blocked. metadata = event.metadata or {} - if not metadata.get("blocking_wait"): + if not _metadata_flag(metadata.get("blocking_wait")): return waiter = canonical_agent_id(event.source_id) @@ -119,8 +141,8 @@ async def observe(self, event: Event) -> None: if event.event_name.endswith(".notification"): return - ttl = metadata.get("wait_timeout") - if not isinstance(ttl, (int, float)) or ttl <= 0: + ttl = _metadata_number(metadata.get("wait_timeout")) + if ttl is None: ttl = self._edge_ttl ttl = min(ttl, MAX_EDGE_TTL_SECONDS) @@ -194,6 +216,11 @@ async def _notify_cycle(self, cycle_edges: List[_WaitEdge]) -> None: cycle = [cycle_edges[0].waiter] + [e.target for e in cycle_edges] agents = list(dict.fromkeys(cycle)) # de-duplicate, keep order request_ids = [e.request_id for e in cycle_edges] + # The notified waiters return immediately, so their waits no longer + # exist — drop the edges now or they would keep "deadlocking" new + # requests until their original timeouts. + for edge in cycle_edges: + self._edges.pop(edge.request_id, None) logger.warning( f"Wait-for cycle detected between agents: {' -> '.join(cycle)}" ) diff --git a/sdk/src/openagents/sdk/workspace.py b/sdk/src/openagents/sdk/workspace.py index cf0be01ef..183c9cae9 100644 --- a/sdk/src/openagents/sdk/workspace.py +++ b/sdk/src/openagents/sdk/workspace.py @@ -219,9 +219,11 @@ async def send_and_wait( # Mark the request as a blocking wait (with its deadline) so the # network's wait-graph can detect deadlocks. requires_response alone # would not prove we are blocked — continuation-style requests also - # set it. - wait_metadata = {"blocking_wait": True, "wait_timeout": timeout} - wait_metadata.update(kwargs.pop("metadata", None) or {}) + # set it. The internal fields are applied last so caller-supplied + # metadata cannot disable detection or forge the deadline. + wait_metadata = dict(kwargs.pop("metadata", None) or {}) + wait_metadata["blocking_wait"] = True + wait_metadata["wait_timeout"] = timeout request = self._build_direct_message( content, requires_response=True, metadata=wait_metadata, **kwargs ) diff --git a/tests/grpc/test_grpc_event_roundtrip.py b/tests/grpc/test_grpc_event_roundtrip.py new file mode 100644 index 000000000..fc132fbbd --- /dev/null +++ b/tests/grpc/test_grpc_event_roundtrip.py @@ -0,0 +1,69 @@ +"""Regression test: request correlation must survive the gRPC transport. + +response_to drives precise reply matching and wait-graph edge clearing; +requires_response tells counter-requests apart from replies. Both used to be +dropped on gRPC ingress because the proto had no fields for them. +""" + +import pytest + +from openagents.models.event import Event +from openagents.proto import agent_service_pb2 +from openagents.sdk.connectors.grpc_connector import GRPCNetworkConnector +from openagents.sdk.transports.grpc import internal_event_from_grpc + + +def make_connector() -> GRPCNetworkConnector: + connector = GRPCNetworkConnector(host="localhost", port=50051, agent_id="alice") + # Normally loaded on connect() + connector.agent_service_pb2 = agent_service_pb2 + return connector + + +@pytest.mark.asyncio +async def test_event_correlation_survives_grpc_roundtrip(): + connector = make_connector() + + original = Event( + event_name="thread.direct_message.send", + source_id="alice", + destination_id="bob", + payload={"content": {"text": "pong"}}, + requires_response=True, + response_to="request-123", + metadata={"blocking_wait": True, "wait_timeout": 5.0}, + ) + + grpc_event = connector._to_grpc_event(original) + assert grpc_event.requires_response is True + assert grpc_event.response_to == "request-123" + + # Server-side conversion (payload extraction is tested elsewhere) + restored = internal_event_from_grpc(grpc_event, dict(original.payload)) + + assert restored.event_id == original.event_id + assert restored.requires_response is True + assert restored.response_to == "request-123" + assert restored.destination_id == "bob" + # gRPC metadata is stringified; the wait-graph parses these leniently + assert restored.metadata["blocking_wait"] == "True" + assert restored.metadata["wait_timeout"] == "5.0" + + +@pytest.mark.asyncio +async def test_plain_event_roundtrip_defaults(): + connector = make_connector() + + original = Event( + event_name="thread.direct_message.send", + source_id="alice", + destination_id="bob", + payload={}, + ) + + restored = internal_event_from_grpc( + connector._to_grpc_event(original), {} + ) + + assert restored.requires_response is False + assert restored.response_to is None diff --git a/tests/network/test_wait_graph.py b/tests/network/test_wait_graph.py index 23b13ce98..d4884fa73 100644 --- a/tests/network/test_wait_graph.py +++ b/tests/network/test_wait_graph.py @@ -157,6 +157,39 @@ async def test_only_cycle_edges_are_reported(self, monitor, delivered): assert len(request_ids) == 2 assert unrelated.event_id not in request_ids + async def test_cycle_edges_removed_after_notification(self, monitor, delivered): + """The notified waiters return immediately, so a later one-way wait + between the same agents must not be reported again.""" + await monitor.observe(blocking_request("alice", "bob")) + await monitor.observe(blocking_request("bob", "alice")) + assert len(delivered) == 2 + delivered.clear() + + # Both waits are gone; a fresh one-way wait is not a deadlock. + await monitor.observe(blocking_request("bob", "alice")) + assert delivered == [] + assert monitor.waiting_targets("alice") == [] + assert monitor.waiting_targets("bob") == ["alice"] + + async def test_stringified_metadata_from_grpc_is_parsed( + self, monitor, delivered + ): + """gRPC transports metadata as map; 'False' must not + count as a blocking wait and numeric strings must keep their value.""" + stringly_false = blocking_request("alice", "bob") + stringly_false.metadata = {"blocking_wait": "False"} + await monitor.observe(stringly_false) + assert monitor.waiting_targets("alice") == [] + + stringly_true = blocking_request("alice", "bob") + stringly_true.metadata = {"blocking_wait": "True", "wait_timeout": "0.01"} + await monitor.observe(stringly_true) + assert monitor.waiting_targets("alice") == ["bob"] + + await asyncio.sleep(0.05) + await monitor.observe(blocking_request("bob", "alice")) + assert delivered == [] # the 0.01s edge expired; no cycle + async def test_no_false_positive_on_shared_target(self, monitor, delivered): await monitor.observe(blocking_request("alice", "bob")) await monitor.observe(blocking_request("charlie", "bob"))