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/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/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..cfc98d71e 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,20 @@ 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, + ) + # 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)") def bind_network(self, network) -> bool: @@ -196,6 +213,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", @@ -291,6 +319,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.""" @@ -333,20 +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, - ) + # 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}" @@ -389,7 +565,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 +577,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 +592,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"), }, ) @@ -709,6 +893,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", @@ -1287,20 +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, - ) + # 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}" @@ -1348,6 +1550,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/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/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/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/connectors/grpc_connector.py b/sdk/src/openagents/sdk/connectors/grpc_connector.py index 69748cfc9..106d03329 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 @@ -528,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/event_gateway.py b/sdk/src/openagents/sdk/event_gateway.py index e235db1f0..d3c22f887 100644 --- a/sdk/src/openagents/sdk/event_gateway.py +++ b/sdk/src/openagents/sdk/event_gateway.py @@ -8,6 +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 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 @@ -105,6 +111,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 +189,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."): @@ -263,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 new file mode 100644 index 000000000..0a77d7038 --- /dev/null +++ b/sdk/src/openagents/sdk/react_context.py @@ -0,0 +1,114 @@ +"""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 asyncio +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 +) + +# 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" + + +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 + and current_react_task.get() is _current_task() + ) + + +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().""" + agent_token = current_react_agent.set(agent_id) + task_token = current_react_task.set(_current_task()) + try: + yield + finally: + current_react_task.reset(task_token) + current_react_agent.reset(agent_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 + """ + if not in_react(): + return + agent_id = current_react_agent.get() + + 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/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 new file mode 100644 index 000000000..47bfa091a --- /dev/null +++ b/sdk/src/openagents/sdk/wait_graph.py @@ -0,0 +1,248 @@ +"""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. +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. +""" + +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" + +# 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. +DEFAULT_EDGE_TTL_SECONDS = 60.0 + +# Upper bound on any edge lifetime, whatever deadline the client claims. +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. + + 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") + + 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 blocking-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: Edge lifetime for requests without a deadline + """ + 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 + + # 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_flag(metadata.get("blocking_wait")): + return + + waiter = canonical_agent_id(event.source_id) + target = canonical_agent_id(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 + + ttl = _metadata_number(metadata.get("wait_timeout")) + if ttl is None: + 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() + ttl, + ) + + 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: 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:")): + 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[_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) + + edge_path: List[_WaitEdge] = [] + visited = set() + + 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(edge.target) + edge_path.append(edge) + found = dfs(edge.target) + if found: + return found + edge_path.pop() + return None + + return dfs(start) + + 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 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)}" + ) + 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 a1e0926b2..183c9cae9 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,65 @@ 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) + # 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. 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 + ) + + 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: + 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 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})" @@ -231,28 +316,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 +328,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 +614,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 +630,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 +669,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 +681,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 +705,7 @@ def post_condition(msg): ) if response: - return response.payload.get("message", {}) + return response.payload return None except Exception as e: @@ -636,6 +724,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 +736,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 +760,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 +777,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 +1130,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/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/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 18f453234..cab1f5b9f 100644 --- a/tests/mods/test_task_delegation.py +++ b/tests/mods/test_task_delegation.py @@ -848,5 +848,310 @@ 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_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.""" + 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..d4884fa73 --- /dev/null +++ b/tests/network/test_wait_graph.py @@ -0,0 +1,284 @@ +"""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, EventSubscription +from openagents.sdk.wait_graph import DEADLOCK_EVENT_NAME, WaitGraphMonitor + + +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={}, + ) + + +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(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 = 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(blocking_request("bob", "alice")) + assert delivered == [] + + async def test_mutual_wait_detected_and_both_notified(self, monitor, delivered): + request_ab = blocking_request("alice", "bob") + request_ba = blocking_request("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_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(blocking_request("alice", "bob")) + await monitor.observe(blocking_request("bob", "charlie")) + assert delivered == [] + + 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_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")) + assert delivered == [] + + async def test_notification_relays_do_not_create_edges(self, monitor, delivered): + 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(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) + + monitor = WaitGraphMonitor( + network_id="test-network", deliver=deliver, edge_ttl_seconds=0.0 + ) + # 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. + 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 new file mode 100644 index 000000000..e2d94d8ab --- /dev/null +++ b/tests/workspace/test_sync_wait_governance.py @@ -0,0 +1,475 @@ +"""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_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_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( + 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 + + 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.""" + + 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