fix(multiagent): preserve shared context and cumulative accounting across serialize/deserialize

strands-py/src/strands/multiagent/graph.py

@@ -56,7 +56,7 @@
 from ..types.multiagent import MultiAgentInput
 from ..types.session import decode_bytes_values, encode_bytes_values
 from ..types.traces import AttributeValue
-from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
+from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status, _parse_metrics, _parse_usage
 
 logger = logging.getLogger(__name__)
 
@@ -1275,6 +1275,10 @@ def serialize_state(self) -> dict[str, Any]:
             "next_nodes_to_execute": next_nodes,
             "current_task": encode_bytes_values(self.state.task),
             "execution_order": [n.node_id for n in self.state.execution_order],
+            "accumulated_usage": self.state.accumulated_usage,
+            "accumulated_metrics": self.state.accumulated_metrics,
+            "execution_count": self.state.execution_count,
+            "execution_time": self.state.execution_time,
             "_internal_state": {
                 "interrupt_state": self._interrupt_state.to_dict(),
             },
@@ -1415,6 +1419,13 @@ def _from_dict(self, payload: dict[str, Any]) -> None:
         # Task
         self.state.task = decode_bytes_values(payload.get("current_task", self.state.task))
 
+        # Cumulative accounting: restore so the timeout budget (should_continue) and the reported
+        # totals stay correct across a resume, rather than resetting to zero.
+        self.state.accumulated_usage = _parse_usage(payload.get("accumulated_usage") or {})
+        self.state.accumulated_metrics = _parse_metrics(payload.get("accumulated_metrics") or {})
+        self.state.execution_count = int(payload.get("execution_count") or 0)
+        self.state.execution_time = int(payload.get("execution_time") or 0)
+
         # next nodes to execute
         next_nodes = [self.nodes[nid] for nid in (payload.get("next_nodes_to_execute") or []) if nid in self.nodes]
         self._resume_next_nodes = next_nodes

strands-py/src/strands/multiagent/swarm.py

@@ -56,7 +56,7 @@
 from ..types.multiagent import MultiAgentInput
 from ..types.session import decode_bytes_values, encode_bytes_values
 from ..types.traces import AttributeValue
-from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
+from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status, _parse_metrics, _parse_usage
 
 logger = logging.getLogger(__name__)
 
@@ -993,6 +993,9 @@ def serialize_state(self) -> dict[str, Any]:
             "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
             "next_nodes_to_execute": next_nodes,
             "current_task": encode_bytes_values(self.state.task),
+            "accumulated_usage": self.state.accumulated_usage,
+            "accumulated_metrics": self.state.accumulated_metrics,
+            "execution_time": self.state.execution_time,
             "context": {
                 "shared_context": getattr(self.state.shared_context, "context", {}) or {},
                 "handoff_node": self.state.handoff_node.node_id if self.state.handoff_node else None,
@@ -1036,12 +1039,20 @@ def deserialize_state(self, payload: dict[str, Any]) -> None:
 
     def _from_dict(self, payload: dict[str, Any]) -> None:
         self.state.completion_status = Status(payload["status"])
+        # Point the state's shared context at the swarm-owned object, matching the identity a fresh run
+        # establishes in stream_async, so the serialize path (reads self.state.shared_context) and the
+        # node-input builder (reads self.shared_context) stay in sync after resume.
+        self.state.shared_context = self.shared_context
         # Hydrate completed nodes & results
         context = payload["context"] or {}
         self.shared_context.context = context.get("shared_context") or {}
         self.state.handoff_message = context.get("handoff_message")
         self.state.handoff_node = self.nodes[context["handoff_node"]] if context.get("handoff_node") else None
 
+        self.state.accumulated_usage = _parse_usage(payload.get("accumulated_usage") or {})
+        self.state.accumulated_metrics = _parse_metrics(payload.get("accumulated_metrics") or {})
+        self.state.execution_time = int(payload.get("execution_time") or 0)
+
         self.state.node_history = [self.nodes[nid] for nid in (payload.get("node_history") or []) if nid in self.nodes]
 
         raw_results = payload.get("node_results") or {}

strands-py/tests/strands/multiagent/test_graph.py

@@ -2046,6 +2046,55 @@ async def test_graph_persisted(mock_strands_tracer, mock_use_span):
     assert "test_node" in final_state["node_results"]
 
 
+def test_graph_serialize_deserialize_serialize_preserves_cumulative_state():
+    """Resuming then re-serializing must not lose cumulative accounting.
+
+    Guards the serialize -> deserialize -> serialize round-trip on the resume path: the second
+    serialize must equal the first for accumulated_usage / accumulated_metrics / execution_count /
+    execution_time. Before the fix these were never persisted, so a resumed graph reset them to zero,
+    corrupting the timeout budget (should_continue) and the totals reported in GraphResult.
+    """
+    builder = GraphBuilder()
+    builder.add_node(create_mock_agent("test_agent"), "test_node")
+    builder.set_entry_point("test_node")
+    graph = builder.build()
+
+    payload = {
+        "type": "graph",
+        "id": "default_graph",
+        "status": "executing",
+        "completed_nodes": [],
+        "failed_nodes": [],
+        "interrupted_nodes": [],
+        "node_results": {},
+        "next_nodes_to_execute": ["test_node"],
+        "current_task": "resume me",
+        "execution_order": [],
+        "accumulated_usage": {"inputTokens": 11, "outputTokens": 22, "totalTokens": 33},
+        "accumulated_metrics": {"latencyMs": 44},
+        "execution_count": 3,
+        "execution_time": 555,
+        "_internal_state": {"interrupt_state": {"activated": False, "context": {}, "interrupts": {}}},
+    }
+
+    graph.deserialize_state(payload)
+
+    # Cumulative accounting is restored, not reset to zero.
+    assert graph.state.accumulated_usage == {"inputTokens": 11, "outputTokens": 22, "totalTokens": 33}
+    assert graph.state.accumulated_metrics == {"latencyMs": 44}
+    assert graph.state.execution_count == 3
+    assert graph.state.execution_time == 555
+
+    serialize1 = graph.serialize_state()
+    graph.deserialize_state(serialize1)
+    serialize2 = graph.serialize_state()
+
+    assert serialize2["accumulated_usage"] == serialize1["accumulated_usage"]
+    assert serialize2["accumulated_metrics"] == serialize1["accumulated_metrics"]
+    assert serialize2["execution_count"] == serialize1["execution_count"]
+    assert serialize2["execution_time"] == serialize1["execution_time"]
+
+
 @pytest.mark.parametrize(
     ("cancel_node", "cancel_message"),
     [(True, "node cancelled by user"), ("custom cancel message", "custom cancel message")],

strands-py/tests/strands/multiagent/test_swarm.py

@@ -1212,6 +1212,60 @@ async def test_swarm_persistence(mock_strands_tracer, mock_use_span):
     assert "test_agent" in final_state["node_results"]
 
 
+def test_swarm_serialize_deserialize_serialize_preserves_state():
+    """Resuming then re-serializing must not lose shared context or cumulative accounting.
+
+    Guards the serialize -> deserialize -> serialize round-trip on the resume path: the second
+    serialize must equal the first for shared_context and the cumulative fields. Before the fix,
+    deserialize restored shared_context onto self.shared_context while serialize read
+    self.state.shared_context, so the next serialize emitted {}; and execution_time /
+    accumulated_usage / accumulated_metrics were never persisted, resetting to zero on resume.
+    """
+    agent = create_mock_agent("first")
+    swarm = Swarm([agent])
+
+    payload = {
+        "type": "swarm",
+        "id": "default_swarm",
+        "status": "executing",
+        "node_history": [],
+        "node_results": {},
+        "next_nodes_to_execute": ["first"],
+        "current_task": "resume me",
+        "accumulated_usage": {"inputTokens": 11, "outputTokens": 22, "totalTokens": 33},
+        "accumulated_metrics": {"latencyMs": 44},
+        "execution_time": 555,
+        "context": {
+            "shared_context": {"first": {"fact": "persist-me"}},
+            "handoff_node": None,
+            "handoff_message": None,
+        },
+        "_internal_state": {"interrupt_state": {"activated": False, "context": {}, "interrupts": {}}},
+    }
+
+    swarm.deserialize_state(payload)
+
+    # The swarm-owned and state-owned shared contexts must be the same restored object.
+    assert swarm.shared_context.context == {"first": {"fact": "persist-me"}}
+    assert swarm.state.shared_context.context == {"first": {"fact": "persist-me"}}
+    assert swarm.state.shared_context is swarm.shared_context
+
+    # Cumulative accounting is restored, not reset to zero.
+    assert swarm.state.accumulated_usage == {"inputTokens": 11, "outputTokens": 22, "totalTokens": 33}
+    assert swarm.state.accumulated_metrics == {"latencyMs": 44}
+    assert swarm.state.execution_time == 555
+
+    serialize1 = swarm.serialize_state()
+    swarm.deserialize_state(serialize1)
+    serialize2 = swarm.serialize_state()
+
+    assert serialize2["context"]["shared_context"] == serialize1["context"]["shared_context"]
+    assert serialize2["context"]["shared_context"] == {"first": {"fact": "persist-me"}}
+    assert serialize2["accumulated_usage"] == serialize1["accumulated_usage"]
+    assert serialize2["accumulated_metrics"] == serialize1["accumulated_metrics"]
+    assert serialize2["execution_time"] == serialize1["execution_time"]
+
+
 @pytest.mark.asyncio
 async def test_swarm_handle_handoff():
     first_agent = create_mock_agent("first")