Fix agent circular-dependency deadlocks in agent-to-agent messaging and task delegatio

docs/python-interface/workspace-interface.mdx

@@ -305,6 +305,54 @@ async def send_and_wait_pattern(ws: Workspace):
         print(f"❌ Status check failed: {e}")
 ```
 
+<Warning>
+**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).
+</Warning>
+
+### 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

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]}")

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:

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

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: