Feature/issue 2 additive resume

integrations/aws-strands/python/README.md

@@ -95,9 +95,15 @@ interrupt round-trip:
   `metadata.strands_reason`.
 - To resume, the client sends the next `RunAgentInput` on the **same
   `thread_id`** with `resume=[ResumeEntry(interrupt_id=..., status="resolved",
-  payload=...)]`. The tool's `interrupt()` call returns exactly `payload`.
+  payload=...)]`. Strands' resume gate is truthiness-based (`if
+  interrupt_.response:`), so a falsy `payload` (`None`, `False`, `""`, `0`,
+  `[]`, `{}`) would otherwise re-raise the same interrupt and re-run the tool
+  body forever. To prevent that, `interrupt()` does **not** return `payload`
+  directly — it returns a truthy envelope: `{"response": payload}` on
+  resolve, `{"cancelled": True}` on cancel. Destructure it with
+  `.get("response")` / `.get("cancelled")`.
 - `status="cancelled"` resumes the tool with the sentinel
-  `{"cancelled": True}` (`ag_ui_strands.agent.INTERRUPT_CANCELLED`) so it can
+  `{"cancelled": True}` (`ag_ui_strands.INTERRUPT_CANCELLED`) so it can
   treat the pause as a denial.
 - **Re-execution on resume:** resuming a paused tool re-runs its body from
   the top — any code before the `interrupt()` call executes again. Guard
@@ -108,15 +114,15 @@ interrupt round-trip:
   def charge_card(tool_context: ToolContext, amount: float) -> str:
       # Unsafe: re-runs (and re-charges) on every resume.
       charge(amount)
-      approved = tool_context.interrupt("confirm_charge", reason={"amount": amount})
-      return "charged" if approved else "cancelled"
+      envelope = tool_context.interrupt("confirm_charge", reason={"amount": amount})
+      return "cancelled" if envelope.get("cancelled") or not envelope.get("response") else "charged"
 
 
   @tool
   def charge_card(tool_context: ToolContext, amount: float) -> str:
       # Safe: side effect happens only after the pause resolves.
-      approved = tool_context.interrupt("confirm_charge", reason={"amount": amount})
-      if not approved:
+      envelope = tool_context.interrupt("confirm_charge", reason={"amount": amount})
+      if envelope.get("cancelled") or not envelope.get("response"):
           return "cancelled"
       charge(amount)
       return "charged"

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

@@ -5,7 +5,7 @@
 frontend proxy-tool sync, per-thread session management, and the two-tier
 A2UI surface generation (``get_a2ui_tools`` / ``plan_a2ui_injection``).
 """
-from .agent import StrandsAgent
+from .agent import INTERRUPT_CANCELLED, StrandsAgent
 from .a2ui_tool import (
     A2UI_OPERATIONS_KEY,
     A2UI_STREAM_KEY,
@@ -30,6 +30,7 @@
 
 __all__ = [
     "StrandsAgent",
+    "INTERRUPT_CANCELLED",
     "A2UI_STREAM_KEY",
     "A2UI_OPERATIONS_KEY",
     "A2UIToolParams",

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

@@ -13,6 +13,7 @@
 
 from strands import Agent as StrandsAgentCore
 from strands.session import SessionManager
+from strands.types.interrupt import InterruptResponseContent
 
 # Params handled explicitly by StrandsAgent — excluded from auto-forwarding.
 # "messages" is excluded: per-thread agents start with no history;
@@ -1122,36 +1123,28 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
             replay_history = (
                 self.config.replay_history_into_strands and session_manager is None
             )
+            # An active interrupt means a sibling frontend tool's placeholder may
+            # be stashed in ``_interrupt_state.context["tool_results"]``; reconcile
+            # even when this turn's frontend result is void, since that stash has
+            # no other correction path and is destroyed once the interrupt resumes.
+            has_active_interrupt = bool(
+                getattr(getattr(strands_agent, "_interrupt_state", None), "activated", False)
+            )
             reconcile_session_results = (
                 session_manager is not None
                 and self.config.replay_history_into_strands
-                and has_nonvoid_frontend_result
+                and (has_nonvoid_frontend_result or has_active_interrupt)
             )
 
-            # A client answering to an interrupt sends its responses
-            # in ``RunAgentInput.resume`` (as per the AG-UI interrupt round-trip),
-            # not as a new user message. Translate those into the Strands resume
-            # prompt shape ``[{"interruptResponse": {"interruptId", "response"}}]``
-            # and drive the stream with it — this takes precedence over every
-            # other path, since a resume run carries no fresh prompt.
-            resume_prompt = None
-            resume_entries = getattr(input_data, "resume", None)
-            if isinstance(resume_entries, list) and resume_entries:
-                resume_prompt = [
-                    {
-                        "interruptResponse": {
-                            "interruptId": entry.interrupt_id,
-                            "response": _wrap_resume_response(
-                                entry.status, entry.payload
-                            ),
-                        }
-                    }
-                    for entry in resume_entries
-                ]
-
-            if resume_prompt is not None:
-                agent_stream = strands_agent.stream_async(resume_prompt)
-            elif replay_history:
+            # Default prompt: the legacy path, passing only the latest user
+            # message and trusting Strands (via session_manager) to track
+            # history. Each branch below may narrow this further; a resume run
+            # can carry BOTH a fresh frontend tool result and an interrupt
+            # response in the same batch, so the resume-entries translation
+            # below runs unconditionally after the other branches and layers
+            # on top, rather than short-circuiting them.
+            resume_prompt: str | List[Dict[str, Any]] | list[InterruptResponseContent] | None = user_message
+            if replay_history:
                 native_history = _build_strands_history(input_data.messages)
                 # Apply ``state_context_builder`` to the last user-text
                 # message in the reconciled history rather than to the
@@ -1178,11 +1171,11 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
                                 )
                             break
                 strands_agent.messages = native_history
-                # ``stream_async(None)`` tells Strands to use existing
-                # ``self.messages`` as-is. The LLM sees real tool results
-                # (including ones produced by the frontend) and emits a
-                # proper follow-up turn instead of re-calling the tool.
-                agent_stream = strands_agent.stream_async(None)
+                # ``None`` tells Strands to use existing ``self.messages`` as-is.
+                # The LLM sees real tool results (including ones produced by the
+                # frontend) and emits a proper follow-up turn instead of
+                # re-calling the tool.
+                resume_prompt = None
             elif reconcile_session_results:
                 try:
                     corrected_native_ids = reconcile_frontend_tool_results(
@@ -1218,13 +1211,26 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
                     getattr(strands_agent, "messages", None) or [],
                     only_ids=set(resolved_native_results),
                 )
-                agent_stream = strands_agent.stream_async(
-                    None if reconciled else user_message
-                )
-            else:
-                # Legacy path: pass only the latest user message and trust
-                # Strands (via session_manager) to track history.
-                agent_stream = strands_agent.stream_async(user_message)
+                resume_prompt = None if reconciled else user_message
+
+            # A client answering to an interrupt sends its responses
+            # in ``RunAgentInput.resume`` (as per the AG-UI interrupt round-trip),
+            # not as a new user message. Translate those into the Strands resume
+            # prompt shape ``[{"interruptResponse": {"interruptId", "response"}}]``
+            # and drive the stream with it — this runs after (and takes
+            # precedence over) every branch above, since a resume batch may
+            # still carry a fresh frontend tool result that needed reconciling.
+            resume_entries = getattr(input_data, "resume", None)
+            if isinstance(resume_entries, list) and resume_entries:
+                resume_prompt = [
+                    {
+                        "interruptResponse": {
+                            "interruptId": entry.interrupt_id,
+                            "response": _wrap_resume_response(entry.status, entry.payload),
+                        }
+                    }
+                    for entry in resume_entries
+                ]
 
             # Drop only the entries whose placeholder was actually corrected
             # this turn — they won't recur. Entries that were NOT corrected
@@ -1241,6 +1247,7 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
                 if len(remaining) != len(wire_to_native):
                     strands_agent.state.set(AG_UI_WIRE_MAP_STATE_KEY, remaining)
 
+            agent_stream = strands_agent.stream_async(resume_prompt)
             try:
                 async for event in agent_stream:
                     # Frontend-tool halt: STOP the loop rather than muting the

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

@@ -11,10 +11,13 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any, Iterable, Mapping
 
 from .client_proxy_tool import PROXY_RESULT_PLACEHOLDER
 
+logger = logging.getLogger(__name__)
+
 # Key under which the adapter stores the ``{wire_tool_call_id: native_toolUseId}``
 # map on the Strands agent's session state. Namespaced to avoid clashing with
 # user-managed state keys.
@@ -87,6 +90,27 @@ def reconcile_frontend_tool_results(
     for message in getattr(agent, "messages", None) or []:
         corrected |= _correct_message(message, pending_results)
 
+    # Correct the agent's live interrupt state too. Isolated in its own
+    # try/except: a failure here must not discard the ``corrected`` ids
+    # already fixed above (agent.messages / session), and — unlike those,
+    # which stay retryable via ``wire_to_native`` on a later turn — this
+    # correction has no retry path. Strands clears
+    # ``_interrupt_state.context`` once the resume succeeds (see
+    # ``event_loop.py``), so a placeholder missed here (e.g. due to a
+    # transient error) is unrecoverable after this turn.
+    try:
+        interrupt_state = getattr(agent, "_interrupt_state", None)
+        if interrupt_state is not None and getattr(interrupt_state, "activated", False):
+            tool_results = interrupt_state.context.get("tool_results")
+            if tool_results:
+                corrected |= _correct_tool_results(tool_results, pending_results)
+    except Exception as e:  # noqa: BLE001 — degrade, don't lose already-corrected ids
+        logger.warning(
+            "Interrupt-context tool result reconciliation failed; the "
+            f"stashed placeholder may reach the model unresolved: {e}",
+            exc_info=True,
+        )
+
     return corrected
 
 
@@ -119,6 +143,28 @@ def has_placeholder_results(messages: Iterable[Any], only_ids: Any = None) -> bo
     return False
 
 
+def _correct_tool_result(tool_result, pending_results: Mapping[str, str]) -> str | None:
+    """Rewrite matching placeholder ToolResult dict. Return the corrected tool_use_id or None if no
+    correction was done."""
+    if not isinstance(tool_result, dict):
+        return None
+
+    tool_use_id = tool_result.get("toolUseId")
+    if tool_use_id in pending_results and _is_placeholder(tool_result.get("content")):
+        tool_result["content"] = [{"text": pending_results[tool_use_id]}]
+        return tool_use_id
+
+
+def _correct_tool_results(tool_results, pending_results: Mapping[str, str]) -> set[str]:
+    """Rewrite matching placeholder ToolResult dicts in *tool_results* in place."""
+    changed: set[str] = set()
+    for tool_result in tool_results:
+        tool_use_id = _correct_tool_result(tool_result, pending_results)
+        if tool_use_id:
+            changed.add(tool_use_id)
+    return changed
+
+
 def _correct_message(message: Any, pending_results: Mapping[str, str]) -> set[str]:
     """Rewrite matching placeholder ``toolResult`` blocks in *message* in place.
 
@@ -131,11 +177,8 @@ def _correct_message(message: Any, pending_results: Mapping[str, str]) -> set[st
         if not isinstance(block, dict):
             continue
         tool_result = block.get("toolResult")
-        if not isinstance(tool_result, dict):
-            continue
-        tool_use_id = tool_result.get("toolUseId")
-        if tool_use_id in pending_results and _is_placeholder(tool_result.get("content")):
-            tool_result["content"] = [{"text": pending_results[tool_use_id]}]
+        tool_use_id = _correct_tool_result(tool_result, pending_results)
+        if tool_use_id:
             changed.add(tool_use_id)
     return changed
 

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

@@ -22,6 +22,7 @@
 from strands.interrupt import Interrupt as StrandsInterrupt
 from strands.interrupt import _InterruptState
 from strands.models.model import Model as StrandsModel
+from strands.session import FileSessionManager
 
 from ag_ui_strands.agent import INTERRUPT_CANCELLED, StrandsAgent
 from ag_ui_strands.config import StrandsAgentConfig, ToolBehavior
@@ -182,9 +183,7 @@ async def test_resolved_resume_builds_interrupt_response_prompt(self):
 
     @pytest.mark.parametrize("falsy_payload", [None, False, "", 0, [], {}])
     @pytest.mark.asyncio
-    async def test_resolved_resume_wraps_falsy_payload_in_truthy_envelope(
-        self, falsy_payload
-    ):
+    async def test_resolved_resume_wraps_falsy_payload_in_truthy_envelope(self, falsy_payload):
         """Falsy resume payloads must be wrapped so Strands' ``if response:`` gate passes.
 
         Regression for ``round1.md`` #1: without the envelope, ``None``/``False``/
@@ -309,21 +308,66 @@ async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs):
             yield {"messageStop": {"stopReason": "end_turn"}}
 
 
-@pytest.mark.asyncio
-async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors():
-    """Regression for docs/round1.md items #1, #2, #3 — real Agent, real
-    tool, scripted model, no network. Intentionally failing until those are
-    fixed; see docs/round1.md."""
+def _make_e2e_agent(config: StrandsAgentConfig) -> tuple[StrandsAgent, _InterruptFlowModel]:
     model = _InterruptFlowModel()
     core = StrandsAgentCore(model=model, tools=[confirm_action], system_prompt="test")
+    return StrandsAgent(core, name="e2e-interrupt", config=config), model
+
+
+@pytest.mark.parametrize("recreate_agent", [False, True])
+@pytest.mark.parametrize("fe_continues", [False, True])
+@pytest.mark.asyncio
+async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors(
+    tmp_path, recreate_agent, fe_continues
+):
+    """Regression for mixed FE tools & interrupts.
+
+    Uses a real ``FileSessionManager`` — the no-session-manager path (in-memory
+    ``replay_history_into_strands``) is out of scope for this regression.
+
+    Parametrized on ``recreate_agent``: ``False`` exercises resume through the
+    same in-memory ``StrandsAgent``/``StrandsAgentCore`` (the per-thread cache
+    still holds the paused agent); ``True`` discards them after turn 1 and
+    resumes through freshly constructed ones sharing the same
+    ``FileSessionManager``-backed session — the cross-process resume scenario
+    the README's "Persistence" caveat describes, where nothing survives in
+    memory from turn 1.
+
+    Parametrized on ``fe_continues`` (``continue_after_frontend_call`` for the
+    frontend tool) because in THIS batch shape the flag is near-moot, and both
+    settings must reach the same interrupt outcome:
+
+    * The model commits both ``toolUse`` blocks in ONE assistant message, so
+      the halt cannot pre-empt ``confirm_action`` — it is already dispatched
+      concurrently by Strands' ``ConcurrentToolExecutor``.
+    * ``confirm_action`` interrupts, so Strands returns early at
+      ``event_loop.py:501`` WITHOUT appending the ``role=user`` tool-result
+      message. That message is the only place ``pending_halt`` is promoted to
+      ``halt_event_stream`` (``agent.py:1479-1480``), so the halt latches but
+      never fires; the interrupt stops the loop instead. Measured consequence:
+      the flag only changes where the frontend ``TOOL_CALL_END`` lands on the
+      wire.
+
+    ``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.
+    """
+    tool_behaviors = {
+        "confirm_action": ToolBehavior(
+            state_from_result=lambda ctx: {"confirmed_key": ctx.result_data}
+        )
+    }
+    if fe_continues:
+        tool_behaviors["approveTool"] = ToolBehavior(continue_after_frontend_call=True)
     config = StrandsAgentConfig(
-        tool_behaviors={
-            "confirm_action": ToolBehavior(
-                state_from_result=lambda ctx: {"confirmed_key": ctx.result_data}
-            )
-        }
+        tool_behaviors=tool_behaviors,
+        session_manager_provider=lambda input_data: FileSessionManager(
+            session_id=input_data.thread_id, storage_dir=str(tmp_path)
+        ),
     )
-    agent = StrandsAgent(core, name="e2e-interrupt", config=config)
+    agent, model = _make_e2e_agent(config)
 
     approve_tool = Tool(name="approveTool", description="approve", parameters={})
     inp1 = _make_run_input(
@@ -342,6 +386,15 @@ async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors():
         if e.type == EventType.TOOL_CALL_START and e.tool_call_name == "approveTool"
     )
 
+    if recreate_agent:
+        # Discard the wrapper and underlying core entirely — turn 2 must
+        # restore interrupt state, the wire->native map, and history purely
+        # from the FileSessionManager-backed session, not from memory. Carry
+        # over the turn count so it reads the same as the non-recreated case.
+        prior_turn = model.turn
+        agent, model = _make_e2e_agent(config)
+        model.turn = prior_turn
+
     inp2 = _make_run_input(
         run_id="run-2",
         messages=[
@@ -364,7 +417,7 @@ async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors():
         and finished2.outcome.type == "interrupt"
         and finished2.outcome.interrupts[0].id == interrupt_id
     )
-    assert not still_stuck, "falsy resume payload re-emitted the same interrupt (round1.md #1)"
+    assert not still_stuck, "falsy resume payload re-emitted the same interrupt"
 
     # --- The frontend tool's REAL result must reach the model. ---
     assert model.turn >= 2, "resume never advanced the event loop past the interrupt"
@@ -375,4 +428,4 @@ async def test_mixed_resume_batch_with_falsy_payload_and_tool_behaviors():
     # --- state_from_result must fire for a tool resolved on resume. ---
     assert any(
         e.type == EventType.STATE_SNAPSHOT and e.snapshot.get("confirmed_key") for e in events2
-    ), "state_from_result did not fire for confirm_action on the resume run (round1.md #3)"
+    ), "state_from_result did not fire for confirm_action on the resume run"