interrupts - swarm - from agent

Author: pgrayy

src/strands/event_loop/event_loop.py

@@ -483,7 +483,8 @@ async def _handle_tool_execution(
 
     if interrupts:
         # Session state stored on AfterInvocationEvent.
-        agent._interrupt_state.activate(context={"tool_use_message": message, "tool_results": tool_results})
+        agent._interrupt_state.context = {"tool_use_message": message, "tool_results": tool_results}
+        agent._interrupt_state.activate()
 
         agent.event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
         yield EventLoopStopEvent(

src/strands/interrupt.py

@@ -53,13 +53,8 @@ class _InterruptState:
     context: dict[str, Any] = field(default_factory=dict)
     activated: bool = False
 
-    def activate(self, context: dict[str, Any] | None = None) -> None:
-        """Activate the interrupt state.
-
-        Args:
-            context: Context associated with the interrupt event.
-        """
-        self.context = context or {}
+    def activate(self) -> None:
+        """Activate the interrupt state."""
         self.activated = True
 
     def deactivate(self) -> None:

src/strands/multiagent/base.py

@@ -8,10 +8,11 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, AsyncIterator, Union
+from typing import Any, AsyncIterator, Sequence, Union
 
 from .._async import run_async
 from ..agent import AgentResult
+from ..interrupt import Interrupt
 from ..types.content import ContentBlock
 from ..types.event_loop import Metrics, Usage
 
@@ -25,16 +26,12 @@ class Status(Enum):
     EXECUTING = "executing"
     COMPLETED = "completed"
     FAILED = "failed"
+    INTERRUPTED = "interrupted"
 
 
 @dataclass
 class NodeResult:
-    """Unified result from node execution - handles both Agent and nested MultiAgentBase results.
-
-    The status field represents the semantic outcome of the node's work:
-    - COMPLETED: The node's task was successfully accomplished
-    - FAILED: The node's task failed or produced an error
-    """
+    """Unified result from node execution - handles both Agent and nested MultiAgentBase results."""
 
     # Core result data - single AgentResult, nested MultiAgentResult, or Exception
     result: Union[AgentResult, "MultiAgentResult", Exception]
@@ -47,6 +44,7 @@ class NodeResult:
     accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
     accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
     execution_count: int = 0
+    interrupts: Sequence[Interrupt] | None = None
 
     def get_agent_results(self) -> list[AgentResult]:
         """Get all AgentResult objects from this node, flattened if nested."""
@@ -78,6 +76,7 @@ def to_dict(self) -> dict[str, Any]:
             "accumulated_usage": self.accumulated_usage,
             "accumulated_metrics": self.accumulated_metrics,
             "execution_count": self.execution_count,
+            "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
         }
 
     @classmethod
@@ -99,6 +98,11 @@ def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
 
         usage = _parse_usage(data.get("accumulated_usage", {}))
         metrics = _parse_metrics(data.get("accumulated_metrics", {}))
+        
+        # Parse interrupts
+        interrupts = []
+        for interrupt_data in data.get("interrupts", []):
+            interrupts.append(Interrupt.from_dict(interrupt_data))
 
         return cls(
             result=result,
@@ -107,6 +111,7 @@ def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
             accumulated_usage=usage,
             accumulated_metrics=metrics,
             execution_count=int(data.get("execution_count", 0)),
+            interrupts=interrupts,
         )
 
 
@@ -125,6 +130,7 @@ class MultiAgentResult:
     accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
     execution_count: int = 0
     execution_time: int = 0
+    interrupts: list[Interrupt] = field(default_factory=list)
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":

src/strands/multiagent/swarm.py

@@ -33,6 +33,7 @@
     MultiAgentInitializedEvent,
 )
 from ..hooks import HookProvider, HookRegistry
+from ..interrupt import Interrupt, _InterruptState
 from ..session import SessionManager
 from ..telemetry import get_tracer
 from ..tools.decorator import tool
@@ -44,6 +45,7 @@
     MultiAgentResultEvent,
 )
 from ..types.content import ContentBlock, Messages
+from ..types.interrupt import InterruptResponseContent
 from ..types.event_loop import Metrics, Usage
 from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
 
@@ -145,7 +147,7 @@ class SwarmState:
     """Current state of swarm execution."""
 
     current_node: SwarmNode | None  # The agent currently executing
-    task: str | list[ContentBlock]  # The original task from the user that is being executed
+    task: str | list[ContentBlock] | list[InterruptResponseContent]  # The original task from the user that is being executed
     completion_status: Status = Status.PENDING  # Current swarm execution status
     shared_context: SharedContext = field(default_factory=SharedContext)  # Context shared between agents
     node_history: list[SwarmNode] = field(default_factory=list)  # Complete history of agents that have executed
@@ -255,11 +257,14 @@ def __init__(
 
         self.shared_context = SharedContext()
         self.nodes: dict[str, SwarmNode] = {}
+
         self.state = SwarmState(
             current_node=None,  # Placeholder, will be set properly
             task="",
             completion_status=Status.PENDING,
         )
+        self._interrupt_state = _InterruptState()
+
         self.tracer = get_tracer()
 
         self.session_manager = session_manager
@@ -277,7 +282,9 @@ def __init__(
         run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
     def __call__(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self,
+        task: str | list[ContentBlock] | list[InterruptResponseContent],
+        invocation_state: dict[str, Any] | None = None, **kwargs: Any,
     ) -> SwarmResult:
         """Invoke the swarm synchronously.
 
@@ -292,7 +299,9 @@ def __call__(
         return run_async(lambda: self.invoke_async(task, invocation_state))
 
     async def invoke_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self,
+        task: str | list[ContentBlock] | list[InterruptResponseContent],
+        invocation_state: dict[str, Any] | None = None, **kwargs: Any,
     ) -> SwarmResult:
         """Invoke the swarm asynchronously.
 
@@ -316,7 +325,9 @@ async def invoke_async(
         return cast(SwarmResult, final_event["result"])
 
     async def stream_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self,
+        task: str | list[ContentBlock] | list[InterruptResponseContent],
+        invocation_state: dict[str, Any] | None = None, **kwargs: Any,
     ) -> AsyncIterator[dict[str, Any]]:
         """Stream events during swarm execution.
 
@@ -334,6 +345,8 @@ async def stream_async(
             - multi_agent_node_stop: When a node stops execution
             - result: Final swarm result
         """
+        self._interrupt_state.resume(task)
+        
         if invocation_state is None:
             invocation_state = {}
 
@@ -644,6 +657,36 @@ def _build_node_input(self, target_node: SwarmNode) -> str:
 
         return context_text
 
+    def _activate_interrupt(self, node: SwarmNode, interrupts: list[Interrupt]) -> Any:
+        """Activate the interrupt state.
+
+        Args:
+            node: The interrupted node.
+            interrupts: The interrupts raised by the user.
+
+        Returns:
+            MultiAgentNodeInterruptEvent
+        """
+
+        logger.debug("node=<%s> | node interrupted", node.node_id)
+        self.state.completion_status = Status.INTERRUPTED
+
+        self._interrupt_state.context.update(
+            {
+                node.node_id: {
+                    "interrupt_state": node.executor._interrupt_state.to_dict(),
+                    "state": node.executor.state.get(),
+                    "messages": node.executor.messages
+                }
+            }
+        )
+        self._interrupt_state.activate()
+
+        # return MultiAgentNodeInterruptEvent(
+        #     node_id=node.node_id,
+        #     interrupts=interrupts,
+        # )
+
     async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
         """Execute swarm and yield TypedEvent objects."""
         try:
@@ -680,9 +723,13 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
 
                 # TODO: Implement cancellation token to stop _execute_node from continuing
                 try:
-                    await self.hooks.invoke_callbacks_async(
-                        BeforeNodeCallEvent(self, current_node.node_id, invocation_state)
+                    _, interrupts = await self.hooks.invoke_callbacks_async(
+                         BeforeNodeCallEvent(self, current_node.node_id, invocation_state)
                     )
+                    if interrupts:
+                        yield self._activate_interrupt(current_node, interrupts)
+                        break
+
                     node_stream = self._stream_with_timeout(
                         self._execute_node(current_node, self.state.task, invocation_state),
                         self.node_timeout,
@@ -691,6 +738,14 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
                     async for event in node_stream:
                         yield event
 
+                    stop_event = cast(MultiAgentNodeStopEvent, event)
+                    node_result = stop_event["node_result"]
+                    if node_result.status == Status.INTERRUPTED:
+                        self._interrupt_state.interrupts.update({interrupt.id: interrupt for interrupt in node_result.interrupts})
+                        # yield self._activate_interrupt(current_node, node_result.interrupts)
+                        self._activate_interrupt(current_node, node_result.interrupts)
+                        break
+
                     self.state.node_history.append(current_node)
                     await self.hooks.invoke_callbacks_async(
                         AfterNodeCallEvent(self, current_node.node_id, invocation_state)
@@ -741,7 +796,10 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
             )
 
     async def _execute_node(
-        self, node: SwarmNode, task: str | list[ContentBlock], invocation_state: dict[str, Any]
+        self,
+        node: SwarmNode,
+        task: str | list[ContentBlock] | list[InterruptResponseContent],
+        invocation_state: dict[str, Any],
     ) -> AsyncIterator[Any]:
         """Execute swarm node and yield TypedEvent objects."""
         start_time = time.time()
@@ -763,8 +821,16 @@ async def _execute_node(
                 # Include additional ContentBlocks in node input
                 node_input = node_input + task
 
+            if self._interrupt_state.activated:
+                node_input = task
+
             # Execute node with streaming
             node.reset_executor_state()
+            if self._interrupt_state.activated:
+                context = self._interrupt_state.context[node.node_id]
+                node.executor.messages = context["messages"]
+                node.executor.state = AgentState(context["state"])
+                node.executor._interrupt_state = _InterruptState.from_dict(context["interrupt_state"])
 
             # Stream agent events with node context and capture final result
             result = None
@@ -779,13 +845,8 @@ async def _execute_node(
             if result is None:
                 raise ValueError(f"Node '{node_name}' did not produce a result event")
 
-            if result.stop_reason == "interrupt":
-                node.executor.messages.pop()  # remove interrupted tool use message
-                node.executor._interrupt_state.deactivate()
-
-                raise RuntimeError("user raised interrupt from agent | interrupts are not yet supported in swarms")
-
             execution_time = round((time.time() - start_time) * 1000)
+            status = Status.INTERRUPTED if result.stop_reason == "interrupt" else Status.COMPLETED
 
             # Create NodeResult with extracted metrics
             result_metrics = getattr(result, "metrics", None)
@@ -795,10 +856,11 @@ async def _execute_node(
             node_result = NodeResult(
                 result=result,
                 execution_time=execution_time,
-                status=Status.COMPLETED,
+                status=status,
                 accumulated_usage=usage,
                 accumulated_metrics=metrics,
                 execution_count=1,
+                interrupts=result.interrupts,
             )
 
             # Store result in state
@@ -849,6 +911,15 @@ def _accumulate_metrics(self, node_result: NodeResult) -> None:
 
     def _build_result(self) -> SwarmResult:
         """Build swarm result from current state."""
+        # Get interrupts from current node (latest iteration only)
+        interrupts = []
+        if (self.state.completion_status == Status.INTERRUPTED and 
+            self.state.current_node and 
+            self.state.current_node.node_id in self.state.results):
+            
+            node_result = self.state.results[self.state.current_node.node_id]
+            interrupts = node_result.interrupts
+        
         return SwarmResult(
             status=self.state.completion_status,
             results=self.state.results,
@@ -857,6 +928,7 @@ def _build_result(self) -> SwarmResult:
             execution_count=len(self.state.node_history),
             execution_time=self.state.execution_time,
             node_history=self.state.node_history,
+            interrupts=interrupts,
         )
 
     def serialize_state(self) -> dict[str, Any]:
@@ -881,6 +953,9 @@ def serialize_state(self) -> dict[str, Any]:
                 "shared_context": getattr(self.state.shared_context, "context", {}) or {},
                 "handoff_message": self.state.handoff_message,
             },
+            "_internal_state": {
+                "interrupt_state": self._interrupt_state.to_dict(),
+            },
         }
 
     def deserialize_state(self, payload: dict[str, Any]) -> None:
@@ -896,6 +971,9 @@ def deserialize_state(self, payload: dict[str, Any]) -> None:
             payload: Dictionary containing persisted state data including status,
                     completed nodes, results, and next nodes to execute.
         """
+        if "_internal_state" in payload:
+            self._interrupt_state = _InterruptState.from_dict(payload["_internal_state"]["interrupt_state"])
+
         if not payload.get("next_nodes_to_execute"):
             for node in self.nodes.values():
                 node.reset_executor_state()

src/strands/session/session_manager.py

@@ -6,7 +6,7 @@
 
 from ..experimental.hooks.multiagent.events import (
     AfterMultiAgentInvocationEvent,
-    AfterNodeCallEvent,
+    BeforeNodeCallEvent,
     MultiAgentInitializedEvent,
 )
 from ..hooks.events import AfterInvocationEvent, AgentInitializedEvent, MessageAddedEvent
@@ -44,7 +44,7 @@ def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
         registry.add_callback(AfterInvocationEvent, lambda event: self.sync_agent(event.agent))
 
         registry.add_callback(MultiAgentInitializedEvent, lambda event: self.initialize_multi_agent(event.source))
-        registry.add_callback(AfterNodeCallEvent, lambda event: self.sync_multi_agent(event.source))
+        registry.add_callback(BeforeNodeCallEvent, lambda event: self.sync_multi_agent(event.source))
         registry.add_callback(AfterMultiAgentInvocationEvent, lambda event: self.sync_multi_agent(event.source))
 
     @abstractmethod