Feature/issue 7

integrations/aws-strands/python/src/ag_ui_strands/agent.py

@@ -83,7 +83,7 @@ def _wrap_resume_response(status: str, payload: Any) -> dict:
     destructures it (e.g. via ``.get("cancelled")`` / ``.get("response")``).
     """
     if status == "cancelled":
-        return INTERRUPT_CANCELLED
+        return dict(INTERRUPT_CANCELLED)
     return {"response": payload}
 
 
@@ -151,7 +151,14 @@ def _extract_interrupts(agent: Any, terminal_result: Any) -> list:
                 return list(interrupts)
     interrupt_state = getattr(agent, "_interrupt_state", None)
     if interrupt_state is not None and getattr(interrupt_state, "activated", False):
-        return list(getattr(interrupt_state, "interrupts", {}).values())
+        # Mirrors Strands' own gate (strands/types/interrupt.py: ``if interrupt_.response:``)
+        # — an interrupt with a truthy response was already answered by a prior partial
+        # resume and must not be re-reported as still pending.
+        return [
+            interrupt
+            for interrupt in getattr(interrupt_state, "interrupts", {}).values()
+            if not getattr(interrupt, "response", None)
+        ]
     return []
 
 
@@ -1250,6 +1257,13 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
             agent_stream = strands_agent.stream_async(resume_prompt)
             try:
                 async for event in agent_stream:
+                    # Capture the terminal ``AgentResult`` (always emitted last
+                    # by ``stream_async``) so a native interrupt pause can be
+                    # detected after the loop. Recorded first so it is never
+                    # dropped, even on the halt-event-stream break below.
+                    if "result" in event and event["result"] is not None:
+                        terminal_result = event["result"]
+
                     # Frontend-tool halt: STOP the loop rather than muting the
                     # wire and draining it. The proxy tool returns a SUCCESSFUL
                     # "Forwarded to client" placeholder, so Strands has every
@@ -1273,13 +1287,6 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
 
                     logger.debug(f"Received event: {event}")
 
-                    # Capture the terminal ``AgentResult`` (always emitted last
-                    # by ``stream_async``) so a native interrupt pause can be
-                    # detected after the loop. Recorded before the
-                    # ``complete``/``force_stop`` break so it is never dropped.
-                    if "result" in event and event["result"] is not None:
-                        terminal_result = event["result"]
-
                     # Skip lifecycle events
                     if event.get("init_event_loop") or event.get("start_event_loop"):
                         continue

integrations/aws-strands/python/tests/test_interrupt.py

@@ -146,6 +146,80 @@ async def test_pause_emits_interrupt_outcome(self):
             "strands_reason": {"summary": "delete all"},
         }
 
+    @pytest.mark.asyncio
+    async def test_terminal_result_captured_despite_halt_in_same_cycle(self):
+        """The terminal ``AgentResult`` capture must run before the
+        ``halt_event_stream`` break check — otherwise a native interrupt whose
+        terminal event arrives on/after the same cycle that triggers a
+        frontend-tool halt is silently dropped (the run finishes bare instead
+        of surfacing the interrupt).
+        """
+        open_interrupt = StrandsInterrupt(
+            id="v1:tool_call:tu-native:00000000-0000-0000-0000-000000000000",
+            name="confirm",
+        )
+        events = [
+            {
+                "current_tool_use": {
+                    "toolUseId": "tu-fe",
+                    "name": "get_cell",
+                    "input": '{"cell": "B4"}',
+                }
+            },
+            {"event": {"contentBlockStop": {}}},
+            # Empty content models the interrupted turn skipping
+            # ToolResultMessageEvent; pending_halt still
+            # latches halt_event_stream here regardless.
+            {"message": {"role": "user", "content": []}},
+            {"result": _agent_result_with_interrupt([open_interrupt])},
+        ]
+        core = _MockStrandsCore(terminal_events=events)
+        agent = _make_base_agent()
+        frontend_tool = Tool(name="get_cell", description="Read a cell", parameters={})
+
+        with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core):
+            events_out = await _collect_events(agent, _make_run_input(tools=[frontend_tool]))
+
+        finished = next(e for e in events_out if e.type == EventType.RUN_FINISHED)
+        assert (
+            finished.outcome is not None
+        ), "terminal interrupt result was dropped on the halt path (round1.md #7a)"
+        assert finished.outcome.type == "interrupt"
+        assert finished.outcome.interrupts[0].id == open_interrupt.id
+
+    @pytest.mark.asyncio
+    async def test_fallback_excludes_already_answered_interrupts(self):
+        """When the terminal ``AgentResult`` is unavailable and ``_extract_interrupts``
+        falls back to the live ``_interrupt_state``, an interrupt that was already
+        answered by a prior partial resume (truthy ``.response``) must not be
+        re-reported as still pending alongside the genuinely open one.
+        """
+        answered = StrandsInterrupt(
+            id="v1:tool_call:tu-answered:00000000-0000-0000-0000-000000000000",
+            name="answered",
+            response={"response": "yes"},
+        )
+        open_interrupt = StrandsInterrupt(
+            id="v1:tool_call:tu-open:00000000-0000-0000-0000-000000000000",
+            name="open",
+        )
+        # No terminal ``{"result": ...}`` event — mirrors the halt-event-stream
+        # path where the stream breaks before a terminal AgentResult is captured.
+        core = _MockStrandsCore(
+            terminal_events=[],
+            interrupts=[answered, open_interrupt],
+        )
+        agent = _make_base_agent()
+
+        with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core):
+            events = await _collect_events(agent, _make_run_input())
+
+        finished = next(e for e in events if e.type == EventType.RUN_FINISHED)
+        assert finished.outcome is not None
+        assert finished.outcome.type == "interrupt"
+        reported_ids = {i.id for i in finished.outcome.interrupts}
+        assert reported_ids == {open_interrupt.id}
+
     @pytest.mark.asyncio
     async def test_no_interrupt_finishes_bare(self):
         """A normal run finishes with no outcome (back-compat, no behavior change)."""
@@ -350,9 +424,7 @@ async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors(
 
     ``False`` keeps coverage that a latched-but-unfired halt does not corrupt
     the interrupt path (moving the latch earlier would break this param and
-    not the other); ``True`` models immediate hand-off and is the only config
-    the TypeScript adapter — which latches per-tool on ``afterToolCallEvent``
-    — emits the interrupt outcome for at all.
+    not the other); ``True`` models immediate hand-off.
     """
     tool_behaviors = {
         "confirm_action": ToolBehavior(