[Feature]: `ag_ui_strands`: bridge native Strands interrupts to AG-UI

[ciolo]

Pre-flight Checklist

• I have searched existing issues and this hasn't been requested yet.

Problem or Motivation

The wrapper (StrandsAgent in agent.py) does not translate native Strands interrupts into the AG-UI interrupt round-trip, even though both ends of the bridge already exist:

• Strands supports human-in-the-loop pauses via tool_context.interrupt(...) (and BeforeToolCallEvent.interrupt(...)). A paused run ends with AgentResult.stop_reason == "interrupt" and a populated AgentResult.interrupts. Resume is done by calling the agent with a prompt of shape [{"interruptResponse": {"interruptId": <id>, "response": <value>}}].
• AG-UI (ag_ui.core, installed as ag_ui_protocol 0.1.19) already ships every type needed: RunFinishedInterruptOutcome, Interrupt, ResumeEntry, ResumeStatus, and RunAgentInput.resume. RunFinishedEvent.outcome already accepts a RunFinishedInterruptOutcome.

The wrapper never connects the two. In practice a native Strands interrupt is either surfaced as RUN_ERROR or silently finishes as if the run completed, and a client's resume payload is dropped. This blocks any agent that relies on tool_context.interrupt() for HITL over AG-UI.

Environment

• ag_ui_strands: 0.2.2
• ag_ui_protocol: 0.1.19
• Strands agent framework (native interrupt API: strands/interrupt.py, agent/agent.py)

Where it breaks (verified against installed source)

1. The stream loop never inspects the terminal AgentResult

Strands signals a pause only on the final event ({"result": AgentResult} with stop_reason == "interrupt" and AgentResult.interrupts populated). In agent.py, grep finds zero references to interrupts, stop_reason, or AgentResult. Every result reference is a tool result (pending_tool_result_ids, toolResult, result_data) — never the run's terminal result. A paused run is therefore indistinguishable from a completed one.

2. RunFinishedEvent is emitted with no outcome

# Always finish the run - frontend handles keeping action executing
yield RunFinishedEvent(
    type=EventType.RUN_FINISHED,
    thread_id=input_data.thread_id,
    run_id=input_data.run_id,
)

There is no outcome= kwarg, so a pause can never be serialized as RunFinishedInterruptOutcome. The comment shows the design target was frontend-tool round-trips, not agent-initiated pauses.

3. input_data.resume is never read

The prompt is built purely from input_data.messages, and the stream is driven by either:

• strands_agent.stream_async(None) (continuation path), or
• strands_agent.stream_async(user_message) (legacy path).

There is no code path that builds the Strands resume prompt [{"interruptResponse": {"interruptId": ..., "response": ...}}] from RunAgentInput.resume, so a client's resume answer is silently discarded.

4. The per-thread in-memory agent cache is the wrong persistence model for HITL

agent.py:327 declares self._agents_by_thread: Dict[str, StrandsAgentCore] = {}. This keeps one live agent per thread_id in process. Strands does persist
_interrupt_state across processes, but only through the injected SessionManager sync path — which the in-memory cache bypasses.

For stateless / one-container-per-session HTTP deployments, the pause run and the resume run are two separate invocations. The paused interrupt state then has no reliable place to live between them, because the cache starts empty on the resume container.

Impact

• Agents using tool_context.interrupt() cannot do HITL over AG-UI through this wrapper.
• Clients that correctly send RunAgentInput.resume (per AG-UI spec) get their payload dropped.
• The only available substitute is the frontend-tool pattern (declare a body-less tool in RunAgentInput.tools, let the run halt, return a tool-result message next turn). That works, but does not cover agent-initiated, guaranteed pauses, and forces every HITL flow to be modeled as a frontend
  tool.

Proposed Solution

The AG-UI protocol layer already has all required types, so this is entirely a wrapper-side change. Minimum work:

1. Consume input_data.resume. Before selecting the prompt (near the stream_async calls): if input_data.resume is set, build

   resume_prompt = [
       {"interruptResponse": {"interruptId": r.interrupt_id, "response": r.payload}}
       for r in input_data.resume
   ]

   and call stream_async(resume_prompt). Map status == "cancelled" to a documented sentinel payload so the tool can treat it as a denial.

2. Detect the interrupt on the terminal event. In the stream loop, handle the {"result": AgentResult} event: when result.stop_reason == "interrupt" and result.interrupts, capture the interrupts and set an interrupted flag.

3. Emit the interrupt outcome. Replace the bare RunFinishedEvent with one carrying

   outcome=RunFinishedInterruptOutcome(
       type="interrupt",
       interrupts=[Interrupt(id=..., reason=..., message=..., tool_call_id=...,
                             response_schema=...), ...],
   )

   when interrupted. Mapping nuance: Strands Interrupt.reason is a JSON object, whereas AG-UI Interrupt.reason is a categorical string ("tool_call" / "input_required" / "confirmation"). Map the interrupt name → the AG-UI reason string, and carry the Strands reason object in AG-UI message (or metadata).

4. Fix persistence for stateless deployments. Either stop relying on _agents_by_thread for interrupt state and always drive it through the injected SessionManager (so _interrupt_state.to_dict/from_dict round-trips across processes), or document that native interrupts require a durable SessionManager and that the in-memory cache is dev-only.

5. Preserve backward compatibility. Keep emitting the bare RunFinishedEvent (no outcome) when there is no interrupt, so existing non-interrupt clients are unaffected. RunAgentInput uses extra="ignore"-style tolerance, so adding resume handling does not break older clients.

Acceptance criteria

• A tool calling tool_context.interrupt(name, reason={...}) produces a RUN_FINISHED with outcome.type == "interrupt" and one AG-UI Interrupt per Strands interrupt.
• Sending a subsequent RunAgentInput on the same thread_id with resume [ResumeEntry(interrupt_id=..., status="resolved", payload=...)] resumes the paused tool, and the tool's interrupt() call returns exactly payload.
• status="cancelled" resumes with a documented denial sentinel.
• With a durable SessionManager, pause and resume survive across separate processes / containers (stateless HTTP deployment).
• No behavior change for runs that never interrupt.

Alternatives Considered

No response

Additional Context

Minimal repro sketch

from strands import tool
from strands.types.tools import ToolContext

@tool(context=True)
def confirm(tool_context: ToolContext, summary: str) -> str:
    answer = tool_context.interrupt("confirm", reason={"summary": summary})
    return "approved" if str(answer).lower() in ("y", "yes", "true") else "denied"

1. Run 1 — RunAgentInput{messages:[...]} that triggers confirm.
   Expected: RUN_FINISHED{outcome: interrupt, interrupts:[{id, reason, message}]}.
   Actual on 0.2.2: bare RUN_FINISHED (no outcome) or RUN_ERROR; pause not surfaced.
2. Run 2 — RunAgentInput{thread_id: SAME, run_id: NEW, resume:[{interrupt_id, status:"resolved", payload}]}.
   Expected: tool resumes, interrupt() returns payload, run finishes with success.
   Actual on 0.2.2: resume ignored; tool never resumes.

[ciolo]

Opened a PR implementing this: #2286.

It's a wrapper-only change in integrations/aws-strands/python/ — consumes RunAgentInput.resume, detects the native pause via the terminal AgentResult (stop_reason == "interrupt"), and emits RunFinishedInterruptOutcome. The required AG-UI types already ship in ag-ui-protocol 0.1.19, so it bumps the floor accordingly. Includes unit tests (full suite green) and README docs on the durable SessionManager requirement for stateless deployments.

Happy to adjust scope or approach — and to add a dojo human_in_the_loop example if you'd like it in the same PR.