ag_ui/client/
interrupts.rs

1//! The human-in-the-loop round trip.
2//!
3//! A run does not only succeed or fail. It can *pause*: the agent finishes with
4//! an [interrupt outcome](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/outcome/enum.RunOutcome.html#variant.Interrupt), listing what it
5//! needs a human to decide, and the conversation continues when the client
6//! sends the answers back in [`RunAgentInput::resume`](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/input/struct.RunAgentInput.html#structfield.resume).
7//!
8//! That round trip is the whole reason `RUN_FINISHED` carries an outcome, and
9//! this module is the client half of it. With a [`Session`](crate::client::Session) it
10//! is two calls:
11//!
12//! ```
13//! # use ag_ui::client::{Session, Update, transport::ReplayTransport};
14//! # use ag_ui::{Event, Interrupt};
15//! # use futures_util::StreamExt;
16//! # let transport = ReplayTransport::with_runs([
17//! #     vec![
18//! #         Event::run_started("thread-1", "run-1"),
19//! #         Event::run_finished_interrupt("thread-1", "run-1", vec![Interrupt::new("i-1", "tool_approval")]),
20//! #     ],
21//! #     vec![
22//! #         Event::run_started("thread-1", "run-2"),
23//! #         Event::run_finished_success("thread-1", "run-2"),
24//! #     ],
25//! # ]);
26//! # let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
27//! # rt.block_on(async {
28//! let mut session = Session::<_>::new(transport, "thread-1");
29//! let mut pending = Vec::new();
30//!
31//! let mut run = session.send("delete the staging database");
32//! while let Some(update) = run.next().await {
33//!     if let Update::Interrupt(interrupt) = update {
34//!         pending.push(interrupt);
35//!     }
36//! }
37//! drop(run);
38//!
39//! // Ask the human, then answer the agent.
40//! let mut resumed = session.resume(&pending[0], serde_json::json!({ "approved": true }));
41//! while resumed.next().await.is_some() {}
42//! # });
43//! ```
44//!
45//! Without one — a proxy, or anything driving [`crate::client::RemoteAgent`] directly —
46//! [`interrupts_of`] finds the interrupts on the event and [`resume_run`]
47//! builds the next request.
48
49use crate::{Event, Interrupt, ResumeEntry, RunAgentInput, RunId};
50use serde_json::Value;
51
52/// The interrupts a `RUN_FINISHED` paused on, or an empty slice for any other
53/// event.
54///
55/// A `RUN_FINISHED` with no outcome at all is a success: producers that predate
56/// the interrupt protocol omit the field, and reading that as "paused" would
57/// hang every one of them.
58pub fn interrupts_of(event: &Event) -> &[Interrupt] {
59    match event {
60        Event::RunFinished(finished) => match &finished.outcome {
61            Some(outcome) => outcome.interrupts(),
62            None => &[],
63        },
64        _ => &[],
65    }
66}
67
68/// Builds the request that resumes a paused run.
69///
70/// Everything the paused run was given — messages, state, tools, context — is
71/// carried over, because the agent is continuing the same conversation. What
72/// changes is the run id and the `resume` payload.
73///
74/// A new run id, not the paused one: the resumed run emits its own
75/// `RUN_STARTED`, and reusing the finished run's id would make two runs in one
76/// thread indistinguishable in a log. Servers that key resumption on the
77/// original id should be passed that id explicitly.
78pub fn resume_run(
79    previous: &RunAgentInput,
80    run_id: impl Into<RunId>,
81    entries: impl Into<Vec<ResumeEntry>>,
82) -> RunAgentInput {
83    RunAgentInput {
84        run_id: run_id.into(),
85        resume: Some(entries.into()),
86        ..previous.clone()
87    }
88}
89
90/// Answering interrupts, one call per decision.
91///
92/// A run can pause on several interrupts at once — three tool approvals, say —
93/// and they are answered together, in one request. This collects the answers.
94///
95/// ```
96/// use ag_ui::client::interrupts::ResumeBuilder;
97/// use ag_ui::{Interrupt, ResumeStatus};
98///
99/// let approve = Interrupt::new("i-1", "tool_approval");
100/// let deny = Interrupt::new("i-2", "tool_approval");
101///
102/// let entries = ResumeBuilder::new()
103///     .resolve(&approve, serde_json::json!({ "approved": true }))
104///     .cancel(&deny)
105///     .build();
106///
107/// assert_eq!(entries[0].interrupt_id, "i-1");
108/// assert_eq!(entries[1].status, ResumeStatus::Cancelled);
109/// ```
110#[derive(Clone, Debug, Default)]
111pub struct ResumeBuilder {
112    entries: Vec<ResumeEntry>,
113}
114
115impl ResumeBuilder {
116    /// An empty builder.
117    pub fn new() -> Self {
118        Self::default()
119    }
120
121    /// Answers an interrupt.
122    #[must_use]
123    pub fn resolve(mut self, interrupt: &Interrupt, payload: impl Into<Value>) -> Self {
124        self.entries
125            .push(ResumeEntry::resolved(interrupt.id.clone(), payload));
126        self
127    }
128
129    /// Approves a tool call after editing its arguments.
130    ///
131    /// Agents that advertise `approveWithEdits` expect the edited arguments
132    /// under an `editedArgs` key; this writes that shape so callers do not have
133    /// to remember it.
134    #[must_use]
135    pub fn resolve_with_edits(self, interrupt: &Interrupt, edited_args: impl Into<Value>) -> Self {
136        self.resolve(
137            interrupt,
138            serde_json::json!({ "editedArgs": edited_args.into() }),
139        )
140    }
141
142    /// Declines an interrupt — the user said no, or the request expired.
143    #[must_use]
144    pub fn cancel(mut self, interrupt: &Interrupt) -> Self {
145        self.entries
146            .push(ResumeEntry::cancelled(interrupt.id.clone()));
147        self
148    }
149
150    /// Answers an interrupt with a status the caller picked.
151    #[must_use]
152    pub fn entry(mut self, entry: ResumeEntry) -> Self {
153        self.entries.push(entry);
154        self
155    }
156
157    /// Whether any answer has been recorded.
158    pub fn is_empty(&self) -> bool {
159        self.entries.is_empty()
160    }
161
162    /// The answers collected so far.
163    pub fn entries(&self) -> &[ResumeEntry] {
164        &self.entries
165    }
166
167    /// Consumes the builder and returns the answers.
168    pub fn build(self) -> Vec<ResumeEntry> {
169        self.entries
170    }
171}
172
173/// Answering one interrupt, on the interrupt itself.
174///
175/// The protocol type lives in `ag-ui-core`, which has no opinion about
176/// consuming a run; these are the two things a client always does with one.
177pub trait InterruptExt {
178    /// Answers this interrupt with a payload.
179    fn resolve(&self, payload: impl Into<Value>) -> ResumeEntry;
180
181    /// Declines this interrupt.
182    fn cancel(&self) -> ResumeEntry;
183
184    /// Whether the interrupt is asking to approve a specific tool call.
185    fn is_tool_approval(&self) -> bool;
186}
187
188impl InterruptExt for Interrupt {
189    fn resolve(&self, payload: impl Into<Value>) -> ResumeEntry {
190        ResumeEntry::resolved(self.id.clone(), payload)
191    }
192
193    fn cancel(&self) -> ResumeEntry {
194        ResumeEntry::cancelled(self.id.clone())
195    }
196
197    fn is_tool_approval(&self) -> bool {
198        self.tool_call_id.is_some() || self.reason == "tool_approval"
199    }
200}
ag_ui/client/interrupts.rs

ag_ui/client/
interrupts.rs
