diff --git a/integrations/aws-strands/python/README.md b/integrations/aws-strands/python/README.md index cdba7fe59b..6a59ddbbdf 100644 --- a/integrations/aws-strands/python/README.md +++ b/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" diff --git a/integrations/aws-strands/python/src/ag_ui_strands/__init__.py b/integrations/aws-strands/python/src/ag_ui_strands/__init__.py index 5212688cc2..3910307bd7 100644 --- a/integrations/aws-strands/python/src/ag_ui_strands/__init__.py +++ b/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", diff --git a/integrations/aws-strands/python/src/ag_ui_strands/agent.py b/integrations/aws-strands/python/src/ag_ui_strands/agent.py index 57e58a55f9..e638a0c046 100644 --- a/integrations/aws-strands/python/src/ag_ui_strands/agent.py +++ b/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 diff --git a/integrations/aws-strands/python/src/ag_ui_strands/session_reconcile.py b/integrations/aws-strands/python/src/ag_ui_strands/session_reconcile.py index 55e5eb102e..79c79b4cb3 100644 --- a/integrations/aws-strands/python/src/ag_ui_strands/session_reconcile.py +++ b/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_all_tools(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_single_tool(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_all_tools(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_single_tool(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_single_tool(tool_result, pending_results) + if tool_use_id: changed.add(tool_use_id) return changed diff --git a/integrations/aws-strands/python/tests/test_interrupt.py b/integrations/aws-strands/python/tests/test_interrupt.py index 2186af54e6..bc85363a07 100644 --- a/integrations/aws-strands/python/tests/test_interrupt.py +++ b/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"