ag_ui/client/
apply.rs

1//! Turning a stream of events into materialised state.
2//!
3//! Consuming AG-UI is not "read a response". The agent sends deltas — a message
4//! opens, text arrives a fragment at a time, tool arguments accumulate as
5//! unparseable JSON fragments, state moves by RFC 6902 patch. [`Applier`] is the
6//! state machine that folds all of that back into something a UI can draw:
7//! a message list, a JSON state document, reasoning text, activities.
8//!
9//! It is a plain synchronous struct. No async, no runtime, no I/O — feed it
10//! events from anywhere.
11//!
12//! ```
13//! use ag_ui::client::apply::Applier;
14//! use ag_ui::{Event, TextMessageRole};
15//!
16//! let mut applier = Applier::new();
17//! for event in [
18//!     Event::run_started("thread-1", "run-1"),
19//!     Event::text_message_start("msg-1", TextMessageRole::Assistant),
20//!     Event::text_message_content("msg-1", "Hello, "),
21//!     Event::text_message_content("msg-1", "world"),
22//!     Event::text_message_end("msg-1"),
23//!     Event::run_finished_success("thread-1", "run-1"),
24//! ] {
25//!     applier.apply(&event)?;
26//! }
27//!
28//! assert_eq!(applier.messages().len(), 1);
29//! assert_eq!(applier.text_of("msg-1"), Some("Hello, world"));
30//! # Ok::<(), ag_ui::client::Error>(())
31//! ```
32//!
33//! # What it does not do
34//!
35//! The applier is *tolerant*: an orphan `TEXT_MESSAGE_CONTENT` opens a message
36//! rather than failing, because a half-drawn conversation beats a blank screen.
37//! Catching the producer's mistake is [`crate::client::verify`]'s job, and
38//! [`crate::client::Session`] runs both. The one thing the applier refuses to do
39//! quietly is corrupt state: a patch that does not apply is an error.
40
41// The THINKING_* events are deprecated but still arrive on real streams, so
42// this module has to name them. Downstream users still get the warnings.
43#![allow(deprecated)]
44
45use std::collections::{HashMap, HashSet};
46
47use crate::{
48    ActivityDeltaEvent, ActivityMessage, ActivitySnapshotEvent, AssistantMessage, DeveloperMessage,
49    Event, InputContent, Interrupt, JsonObject, Message, MessageId, PatchOperation,
50    ReasoningEncryptedValueEvent, ReasoningEncryptedValueSubtype, ReasoningMessage,
51    ReasoningMessageChunkEvent, RunId, RunOutcome, SubagentErrorEvent, SubagentFinishedEvent,
52    SubagentOutcome, SubagentRunId, SubagentStartedEvent, SystemMessage, TextInputContent,
53    TextMessageChunkEvent, TextMessageRole, ThreadId, ToolCall, ToolCallChunkEvent, ToolCallId,
54    ToolMessage, UserContent, UserMessage,
55};
56use serde::Deserialize;
57use serde_json::Value;
58
59use crate::client::error::{Error, Result};
60use crate::metadata::merge_metadata_into;
61
62/// What one event changed.
63///
64/// Returned by [`Applier::apply`] so a UI can redraw one row instead of the
65/// whole conversation. Every variant that names a message carries its index
66/// into [`Applier::messages`].
67#[derive(Clone, Debug, PartialEq)]
68#[non_exhaustive]
69pub enum Changed {
70    /// Nothing a view would redraw: `STEP_*`, `RAW`, `CUSTOM`, or an event
71    /// whose target does not exist.
72    Nothing,
73    /// One message was created, appended to, or completed.
74    Message(MessageChange),
75    /// `MESSAGES_SNAPSHOT` replaced the whole list. Messages may have been
76    /// removed, so a view must redraw all of it.
77    MessagesReplaced,
78    /// The application state was replaced or patched.
79    State,
80    /// Reasoning content changed. Reasoning is kept out of [`Applier::messages`];
81    /// see [`Applier::reasoning`].
82    Reasoning(ReasoningChange),
83    /// A subagent was announced, resumed, finished, suspended or failed. What
84    /// it *produces* arrives as ordinary message and reasoning changes,
85    /// carrying its id; this is the lifecycle. See [`Applier::subagents`].
86    Subagent(SubagentChange),
87    /// `RUN_STARTED`.
88    RunStarted {
89        /// The conversation the run belongs to.
90        thread_id: ThreadId,
91        /// The run that started.
92        run_id: RunId,
93    },
94    /// `RUN_FINISHED`. An absent outcome is reported as
95    /// [`RunOutcome::Success`](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/outcome/enum.RunOutcome.html#variant.Success), which is what the protocol says it means.
96    RunFinished {
97        /// How the run ended.
98        outcome: RunOutcome,
99        /// The agent's return value, if it sent one.
100        result: Option<Value>,
101    },
102    /// `RUN_ERROR`.
103    RunError {
104        /// What went wrong, for a human.
105        message: String,
106        /// The machine-readable code, when the agent sent one.
107        code: Option<String>,
108    },
109}
110
111/// Which message changed, and how.
112#[derive(Clone, Debug, PartialEq)]
113pub struct MessageChange {
114    /// Index into [`Applier::messages`].
115    pub index: usize,
116    /// The message's id.
117    pub id: MessageId,
118    /// What happened to it.
119    pub kind: MessageChangeKind,
120}
121
122/// What happened to a message.
123#[derive(Clone, Debug, PartialEq)]
124#[non_exhaustive]
125pub enum MessageChangeKind {
126    /// The message was created and is now open for content.
127    Started,
128    /// Text was appended.
129    Content {
130        /// The text this event appended.
131        delta: String,
132    },
133    /// The message was closed; no more content will arrive for it.
134    Ended,
135    /// A tool call was attached to the message. Several can be open at once —
136    /// a model asking for two things at once opens two.
137    ToolCallStarted {
138        /// The call's id.
139        tool_call_id: ToolCallId,
140        /// The tool being called.
141        name: String,
142    },
143    /// Argument JSON was appended to a tool call.
144    ///
145    /// Parallel calls interleave their events, so consecutive `ToolCallArgs`
146    /// need not belong to the same call: `tool_call_id` is what separates them,
147    /// and arrival order is the only nesting there is — see the [session module
148    /// docs](crate::client::session).
149    ToolCallArgs {
150        /// The call being appended to.
151        tool_call_id: ToolCallId,
152        /// The fragment this event appended. Rarely valid JSON on its own.
153        delta: String,
154    },
155    /// A tool call's arguments are complete.
156    ToolCallEnded {
157        /// The call that closed.
158        tool_call_id: ToolCallId,
159    },
160    /// A tool result arrived and was appended as a new message.
161    ToolResult {
162        /// The call this result answers.
163        tool_call_id: ToolCallId,
164    },
165    /// An activity was published or patched.
166    Activity,
167    /// A provider's opaque reasoning blob was attached to the message or to one
168    /// of its tool calls.
169    EncryptedValue,
170}
171
172/// Which reasoning message changed, and how.
173#[derive(Clone, Debug, PartialEq)]
174pub struct ReasoningChange {
175    /// The reasoning message's id.
176    pub id: MessageId,
177    /// What happened to it.
178    pub kind: ReasoningChangeKind,
179}
180
181/// What happened to a reasoning message.
182///
183/// [`Started`](Self::Started) and [`Ended`](Self::Ended) arrive **once per id**.
184/// The protocol brackets a thought twice — `REASONING_START` opens the block and
185/// `REASONING_MESSAGE_START` the message inside it, both under the same id, and
186/// the two terminators close it the same way — but that is protocol framing, not
187/// two things happening. A consumer that prints a finished thought prints it
188/// once.
189#[derive(Clone, Debug, PartialEq)]
190#[non_exhaustive]
191pub enum ReasoningChangeKind {
192    /// The reasoning message was created.
193    Started,
194    /// Reasoning text was appended.
195    Content {
196        /// The text this event appended.
197        delta: String,
198    },
199    /// The reasoning message was closed; no more content will arrive for it.
200    Ended,
201    /// A provider's opaque reasoning blob was attached.
202    EncryptedValue,
203}
204
205/// One subagent invocation the run has announced, as the applier tracks it.
206///
207/// Keyed by [`run_id`](Self::run_id), which names *this invocation* — running
208/// the same subagent twice yields two entries — while [`name`](Self::name) is
209/// the reusable half, for display. Key transient UI by the id; persist
210/// nothing by it.
211#[derive(Clone, Debug, PartialEq)]
212pub struct Subagent {
213    /// The invocation's id, as every event it produced carries it.
214    pub run_id: SubagentRunId,
215    /// The subagent's declared type or name.
216    pub name: String,
217    /// Human-readable description, when announced.
218    pub description: Option<String>,
219    /// The enclosing subagent, when subagents nest.
220    pub parent_subagent_run_id: Option<SubagentRunId>,
221    /// The tool call that spawned it, for the agents-as-tools pattern.
222    pub parent_tool_call_id: Option<ToolCallId>,
223    /// The message that held that tool call.
224    pub parent_message_id: Option<MessageId>,
225    /// Where the invocation stands.
226    pub status: SubagentStatus,
227}
228
229/// Where a subagent invocation stands.
230#[derive(Clone, Debug, PartialEq)]
231#[non_exhaustive]
232pub enum SubagentStatus {
233    /// Announced and not yet closed.
234    Running,
235    /// Closed with a success outcome — or with none, the legacy reading.
236    Finished {
237        /// The completion payload, if it sent one.
238        result: Option<Value>,
239    },
240    /// Closed because the run paused on interrupts the subagent raised. The
241    /// same id may be announced again by the run that resumes it, and that
242    /// arrives as [`SubagentChangeKind::Resumed`] rather than as a second
243    /// subagent.
244    Suspended {
245        /// The completion payload so far, if it sent one.
246        result: Option<Value>,
247        /// The run-level interrupts it directly owns — empty for an ancestor
248        /// suspended because a descendant interrupted. Each such
249        /// [`Interrupt`] carries the id back as `subagent_run_id`.
250        interrupt_ids: Vec<String>,
251    },
252    /// Closed by `SUBAGENT_ERROR`.
253    Failed {
254        /// What went wrong, for a human.
255        message: String,
256        /// The machine-readable code, when the agent sent one.
257        code: Option<String>,
258    },
259}
260
261/// Which subagent changed, and how.
262#[derive(Clone, Debug, PartialEq)]
263pub struct SubagentChange {
264    /// Index into [`Applier::subagents`].
265    pub index: usize,
266    /// The invocation's id.
267    pub run_id: SubagentRunId,
268    /// What happened to it.
269    pub kind: SubagentChangeKind,
270}
271
272/// What happened to a subagent invocation.
273#[derive(Clone, Debug, PartialEq)]
274#[non_exhaustive]
275pub enum SubagentChangeKind {
276    /// A new invocation was announced.
277    Started,
278    /// A suspended invocation was announced again — a continuation, not a
279    /// second subagent. Move its group from waiting back to running.
280    Resumed,
281    /// The invocation completed.
282    Finished,
283    /// The invocation paused on interrupts; see [`SubagentStatus::Suspended`].
284    Suspended,
285    /// The invocation failed.
286    Failed,
287}
288
289/// The materialised view of a run.
290///
291/// See the [module docs](self) for the shape of the problem this solves.
292#[derive(Clone, Debug)]
293pub struct Applier {
294    messages: Vec<Message>,
295    by_id: HashMap<MessageId, usize>,
296    /// Tool call id to the index of the assistant message that owns it.
297    tool_calls: HashMap<ToolCallId, usize>,
298    /// Text messages open for content, with the subagent each belongs to —
299    /// several at once under concurrent subagents. What a chunk that names
300    /// no message is resolved against.
301    open_text: Vec<(Option<SubagentRunId>, MessageId)>,
302    state: Value,
303    reasoning: Vec<ReasoningMessage>,
304    reasoning_by_id: HashMap<MessageId, usize>,
305    /// Reasoning ids started and not yet ended. What makes a thought report one
306    /// [`ReasoningChangeKind::Started`] and one [`ReasoningChangeKind::Ended`]
307    /// however many events the protocol brackets it with.
308    open_reasoning: HashSet<MessageId>,
309    /// The reasoning message the events that carry no id belong to — the
310    /// `THINKING_*` family and a `REASONING_MESSAGE_CHUNK` after the first.
311    current_reasoning: Option<MessageId>,
312    /// `THINKING_*` carries no message id, so ids are minted for it.
313    thinking_counter: u64,
314    /// Reasoning messages open for content, with the subagent each belongs
315    /// to — the reasoning counterpart of `open_text`, for the chunk shorthand.
316    reasoning_streams: Vec<(Option<SubagentRunId>, MessageId)>,
317    thread_id: Option<ThreadId>,
318    run_id: Option<RunId>,
319    interrupts: Vec<Interrupt>,
320    /// The subagent invocations announced so far, in announcement order.
321    subagents: Vec<Subagent>,
322    subagent_by_id: HashMap<SubagentRunId, usize>,
323}
324
325impl Default for Applier {
326    fn default() -> Self {
327        Self::new()
328    }
329}
330
331impl Applier {
332    /// An empty applier: no messages, and `{}` for state.
333    ///
334    /// The state starts as an empty object rather than `null` so that the first
335    /// `STATE_DELTA` of a run has something to patch.
336    pub fn new() -> Self {
337        Self {
338            messages: Vec::new(),
339            by_id: HashMap::new(),
340            tool_calls: HashMap::new(),
341            open_text: Vec::new(),
342            state: Value::Object(JsonObject::new()),
343            reasoning: Vec::new(),
344            reasoning_by_id: HashMap::new(),
345            open_reasoning: HashSet::new(),
346            current_reasoning: None,
347            thinking_counter: 0,
348            reasoning_streams: Vec::new(),
349            thread_id: None,
350            run_id: None,
351            interrupts: Vec::new(),
352            subagents: Vec::new(),
353            subagent_by_id: HashMap::new(),
354        }
355    }
356
357    /// Seeds the applier with an existing conversation.
358    #[must_use]
359    pub fn with_messages(mut self, messages: impl Into<Vec<Message>>) -> Self {
360        self.replace_messages(messages.into());
361        self
362    }
363
364    /// Seeds the applier with an existing state document.
365    #[must_use]
366    pub fn with_state(mut self, state: impl Into<Value>) -> Self {
367        self.state = state.into();
368        self
369    }
370
371    /// The assembled conversation, oldest first.
372    pub fn messages(&self) -> &[Message] {
373        &self.messages
374    }
375
376    /// The message with this id, if the applier has seen one.
377    pub fn message(&self, id: &MessageId) -> Option<&Message> {
378        self.by_id
379            .get(id)
380            .and_then(|index| self.messages.get(*index))
381    }
382
383    /// The text of the message with this id, for the roles that carry plain
384    /// text. Multimodal user messages return only their text parts' first
385    /// fragment; use [`Applier::message`] for the whole payload.
386    pub fn text_of(&self, id: impl Into<MessageId>) -> Option<&str> {
387        match self.message(&id.into())? {
388            Message::Assistant(m) => m.content.as_deref(),
389            Message::System(m) => Some(&m.content),
390            Message::Developer(m) => Some(&m.content),
391            Message::Tool(m) => Some(&m.content),
392            Message::Reasoning(m) => Some(&m.content),
393            Message::User(m) => match &m.content {
394                UserContent::Text(text) => Some(text),
395                UserContent::Parts(parts) => parts.iter().find_map(|part| match part {
396                    InputContent::Text(text) => Some(text.text.as_str()),
397                    _ => None,
398                }),
399            },
400            Message::Activity(_) => None,
401        }
402    }
403
404    /// The application state, as the JSON document the protocol carries.
405    ///
406    /// This is the authority: snapshots replace it and deltas patch it. A typed
407    /// view is a projection of this — see [`Applier::state_as`].
408    pub fn state(&self) -> &Value {
409        &self.state
410    }
411
412    /// Deserializes the application state into a caller-defined type.
413    ///
414    /// ```
415    /// # use ag_ui::client::apply::Applier;
416    /// # use ag_ui::Event;
417    /// # use serde::Deserialize;
418    /// #[derive(Deserialize)]
419    /// struct Ui {
420    ///     step: u32,
421    /// }
422    ///
423    /// let mut applier = Applier::new();
424    /// applier.apply(&Event::state_snapshot(serde_json::json!({ "step": 2 })))?;
425    /// assert_eq!(applier.state_as::<Ui>()?.step, 2);
426    /// # Ok::<(), ag_ui::client::Error>(())
427    /// ```
428    pub fn state_as<T: for<'de> Deserialize<'de>>(&self) -> Result<T> {
429        T::deserialize(&self.state).map_err(Error::State)
430    }
431
432    /// Replaces the application state without going through an event.
433    pub fn set_state(&mut self, state: impl Into<Value>) {
434        self.state = state.into();
435    }
436
437    /// The reasoning messages, oldest first.
438    ///
439    /// Reasoning is deliberately not in [`Applier::messages`]: a UI shows it in
440    /// a separate pane, or not at all, and folding it into the transcript would
441    /// make "the assistant's reply" ambiguous.
442    pub fn reasoning(&self) -> &[ReasoningMessage] {
443        &self.reasoning
444    }
445
446    /// The accumulated reasoning text for one reasoning message.
447    pub fn reasoning_text(&self, id: &MessageId) -> Option<&str> {
448        self.reasoning_by_id
449            .get(id)
450            .and_then(|index| self.reasoning.get(*index))
451            .map(|message| message.content.as_str())
452    }
453
454    /// The thread of the run last seen on `RUN_STARTED`.
455    pub fn thread_id(&self) -> Option<&ThreadId> {
456        self.thread_id.as_ref()
457    }
458
459    /// The run id last seen on `RUN_STARTED`.
460    pub fn run_id(&self) -> Option<&RunId> {
461        self.run_id.as_ref()
462    }
463
464    /// The interrupts the last `RUN_FINISHED` paused on. Empty unless the run
465    /// is waiting for human input; cleared when the next run starts.
466    pub fn interrupts(&self) -> &[Interrupt] {
467        &self.interrupts
468    }
469
470    /// The subagent invocations the run has announced, in announcement order.
471    ///
472    /// Kept across runs on purpose: a subagent suspended in one run is
473    /// announced again by the run that resumes it, and the entry it left is
474    /// what makes that read as a continuation rather than a duplicate.
475    pub fn subagents(&self) -> &[Subagent] {
476        &self.subagents
477    }
478
479    /// One subagent invocation, by id.
480    pub fn subagent(&self, run_id: &SubagentRunId) -> Option<&Subagent> {
481        self.subagent_by_id
482            .get(run_id)
483            .and_then(|index| self.subagents.get(*index))
484    }
485
486    /// Appends a message the client itself produced — typically the user's
487    /// turn, before starting a run.
488    ///
489    /// Returns its index in [`Applier::messages`].
490    pub fn push_message(&mut self, message: Message) -> usize {
491        let index = self.messages.len();
492        self.by_id.insert(message.id().clone(), index);
493        if let Message::Assistant(assistant) = &message {
494            for call in assistant.tool_calls.iter().flatten() {
495                self.tool_calls.insert(call.id.clone(), index);
496            }
497        }
498        self.messages.push(message);
499        index
500    }
501
502    /// Applies one event and reports what it changed.
503    ///
504    /// Chunk events are accepted here as well as their expanded form, so an
505    /// applier driven directly from a raw stream still assembles correctly —
506    /// but [`crate::client::chunks`] is the place that turns them into the explicit
507    /// triples the rest of the protocol is written in.
508    pub fn apply(&mut self, event: &Event) -> Result<Changed> {
509        let changed = self.apply_inner(event)?;
510        // Metadata folds into whatever the event built, once the event has
511        // built it — see `crate::metadata` for which events build anything.
512        if let Some(metadata) = event.metadata() {
513            self.merge_event_metadata(event, &changed, metadata);
514        }
515        Ok(changed)
516    }
517
518    fn apply_inner(&mut self, event: &Event) -> Result<Changed> {
519        match event {
520            Event::TextMessageStart(e) => Ok(self.text_start(
521                e.message_id.clone(),
522                e.role,
523                e.name.clone(),
524                e.subagent_run_id.clone(),
525            )),
526            Event::TextMessageContent(e) => {
527                self.text_content(&e.message_id, &e.delta, &e.subagent_run_id)
528            }
529            Event::TextMessageEnd(e) => Ok(self.text_end(&e.message_id)),
530            Event::TextMessageChunk(e) => self.text_chunk(e),
531
532            Event::ToolCallStart(e) => Ok(self.tool_call_start(
533                e.tool_call_id.clone(),
534                e.tool_call_name.clone(),
535                e.parent_message_id.clone(),
536                e.subagent_run_id.clone(),
537            )),
538            Event::ToolCallArgs(e) => self.tool_call_args(&e.tool_call_id, &e.delta),
539            Event::ToolCallEnd(e) => Ok(self.tool_call_end(&e.tool_call_id)),
540            Event::ToolCallChunk(e) => self.tool_call_chunk(e),
541            Event::ToolCallResult(e) => Ok(self.tool_call_result(
542                e.message_id.clone(),
543                e.tool_call_id.clone(),
544                e.content.clone(),
545                e.subagent_run_id.clone(),
546            )),
547
548            Event::StateSnapshot(e) => {
549                self.state = e.snapshot.clone();
550                Ok(Changed::State)
551            }
552            Event::StateDelta(e) => {
553                apply_patch(&mut self.state, &e.delta, "state")?;
554                Ok(Changed::State)
555            }
556            Event::MessagesSnapshot(e) => {
557                self.merge_snapshot(e.messages.clone());
558                Ok(Changed::MessagesReplaced)
559            }
560
561            Event::ActivitySnapshot(e) => Ok(self.activity_snapshot(e)),
562            Event::ActivityDelta(e) => self.activity_delta(e),
563
564            Event::ReasoningStart(e) => {
565                Ok(self.reasoning_start(e.message_id.clone(), e.subagent_run_id.clone()))
566            }
567            Event::ReasoningMessageStart(e) => {
568                Ok(self.reasoning_start(e.message_id.clone(), e.subagent_run_id.clone()))
569            }
570            Event::ReasoningMessageContent(e) => {
571                Ok(self.reasoning_content(&e.message_id, &e.delta, &e.subagent_run_id))
572            }
573            Event::ReasoningMessageEnd(e) => Ok(self.reasoning_end(&e.message_id)),
574            Event::ReasoningEnd(e) => Ok(self.reasoning_end(&e.message_id)),
575            Event::ReasoningMessageChunk(e) => self.reasoning_chunk(e),
576            Event::ReasoningEncryptedValue(e) => Ok(self.encrypted_value(e)),
577
578            // The THINKING_* family predates subagents and carries no
579            // attribution, so a thought it opens is the parent's.
580            Event::ThinkingStart(_) => {
581                let id = self.mint_thinking_id();
582                Ok(self.reasoning_start(id, None))
583            }
584            Event::ThinkingTextMessageStart(_) => {
585                let id = self.thinking_id();
586                Ok(self.reasoning_start(id, None))
587            }
588            Event::ThinkingTextMessageContent(e) => {
589                let id = self.thinking_id();
590                Ok(self.reasoning_content(&id, &e.delta, &None))
591            }
592            Event::ThinkingTextMessageEnd(_) | Event::ThinkingEnd(_) => {
593                Ok(match self.current_reasoning.clone() {
594                    Some(id) => self.reasoning_end(&id),
595                    None => Changed::Nothing,
596                })
597            }
598
599            Event::RunStarted(e) => {
600                self.thread_id = Some(e.thread_id.clone());
601                self.run_id = Some(e.run_id.clone());
602                self.interrupts.clear();
603                Ok(Changed::RunStarted {
604                    thread_id: e.thread_id.clone(),
605                    run_id: e.run_id.clone(),
606                })
607            }
608            Event::RunFinished(e) => {
609                let outcome = e.outcome.clone().unwrap_or(RunOutcome::Success);
610                outcome.validate()?;
611                self.interrupts = outcome.interrupts().to_vec();
612                Ok(Changed::RunFinished {
613                    outcome,
614                    result: e.result.clone(),
615                })
616            }
617            Event::RunError(e) => Ok(Changed::RunError {
618                message: e.message.clone(),
619                code: e.code.clone(),
620            }),
621
622            Event::StepStarted(_) | Event::StepFinished(_) | Event::Raw(_) | Event::Custom(_) => {
623                Ok(Changed::Nothing)
624            }
625
626            Event::SubagentStarted(e) => Ok(self.subagent_started(e)),
627            Event::SubagentFinished(e) => Ok(self.subagent_finished(e)),
628            Event::SubagentError(e) => Ok(self.subagent_error(e)),
629        }
630    }
631
632    // ---- messages -------------------------------------------------------
633
634    fn replace_messages(&mut self, messages: Vec<Message>) {
635        self.by_id.clear();
636        self.tool_calls.clear();
637        for (index, message) in messages.iter().enumerate() {
638            self.by_id.insert(message.id().clone(), index);
639            if let Message::Assistant(assistant) = message {
640                for call in assistant.tool_calls.iter().flatten() {
641                    self.tool_calls.insert(call.id.clone(), index);
642                }
643            }
644        }
645        self.messages = messages;
646        // A snapshot can drop a message that was being streamed into.
647        let by_id = &self.by_id;
648        self.open_text.retain(|(_, open)| by_id.contains_key(open));
649    }
650
651    /// Folds a `MESSAGES_SNAPSHOT` into the conversation.
652    ///
653    /// A snapshot is an *edit*, not a replacement — upstream
654    /// (`client/src/apply/default.ts`, `case EventType.MESSAGES_SNAPSHOT`)
655    /// rebuilds the list by filtering the local one and appending whatever the
656    /// snapshot adds. Three consequences, all of them load-bearing:
657    ///
658    /// - the order the client already had wins for every id the snapshot also
659    ///   carries, so a backend that reorders its own history does not reshuffle
660    ///   the transcript under the user;
661    /// - messages the snapshot leaves out are dropped — that is how a
662    ///   summarizing backend deletes turns;
663    /// - except `activity`, which survives a snapshot that carries none.
664    ///   Activity never travels back to the backend, so one that does not track
665    ///   it *cannot* list it, and dropping the local copies would clear a pane
666    ///   of the UI on every snapshot. A snapshot that does carry activity is
667    ///   declaring the whole set, so an activity missing from it has been
668    ///   deleted; without that half, a client-side activity would be
669    ///   undeletable.
670    ///
671    /// Upstream's rule has a fourth clause, for `reasoning`. It needs no
672    /// equivalent here: reasoning lives in [`Applier::reasoning`], not in
673    /// [`Applier::messages`], so a snapshot of the conversation cannot drop it.
674    fn merge_snapshot(&mut self, snapshot: Vec<Message>) {
675        let snapshot_owns_activity = snapshot.iter().any(|m| matches!(m, Message::Activity(_)));
676
677        // `Option` slots so a message can be moved out when it is placed, and
678        // whatever is left over is exactly what the snapshot added.
679        let mut incoming: Vec<Option<Message>> = snapshot.into_iter().map(Some).collect();
680        let mut position: HashMap<MessageId, usize> = HashMap::with_capacity(incoming.len());
681        for (index, message) in incoming.iter().enumerate() {
682            let id = message.as_ref().expect("every slot starts full").id();
683            position.insert(id.clone(), index);
684        }
685
686        let previous = std::mem::take(&mut self.messages);
687        let mut merged = Vec::with_capacity(previous.len().max(incoming.len()));
688        for message in previous {
689            let keep_local = !snapshot_owns_activity && matches!(message, Message::Activity(_));
690            if let Some(index) = position.get(message.id()).copied() {
691                if keep_local {
692                    // Claim the slot so the snapshot's copy is not appended on
693                    // top of the local one further down.
694                    incoming[index] = None;
695                } else if let Some(replacement) = incoming[index].take() {
696                    merged.push(replacement);
697                    continue;
698                }
699            }
700            if keep_local {
701                merged.push(message);
702            }
703        }
704        merged.extend(incoming.into_iter().flatten());
705
706        self.replace_messages(merged);
707    }
708
709    fn message_change(&self, id: &MessageId, kind: MessageChangeKind) -> Changed {
710        match self.by_id.get(id) {
711            Some(index) => Changed::Message(MessageChange {
712                index: *index,
713                id: id.clone(),
714                kind,
715            }),
716            None => Changed::Nothing,
717        }
718    }
719
720    fn text_start(
721        &mut self,
722        id: MessageId,
723        role: TextMessageRole,
724        name: Option<String>,
725        owner: Option<SubagentRunId>,
726    ) -> Changed {
727        self.open_text.retain(|(_, open)| open != &id);
728        self.open_text.push((owner.clone(), id.clone()));
729        if let Some(index) = self.by_id.get(&id) {
730            // Re-opening a known id keeps the message and appends to it, which
731            // is what a producer that restarts a stream means — and keeps its
732            // attribution, which the opener transferred once.
733            return Changed::Message(MessageChange {
734                index: *index,
735                id,
736                kind: MessageChangeKind::Started,
737            });
738        }
739        // Attribution transfers to the message the event creates, so a
740        // renderer can group a conversation by subagent without replaying
741        // the stream.
742        let mut message = empty_message(id.clone(), role, name);
743        message.set_subagent_run_id(owner);
744        let index = self.push_message(message);
745        Changed::Message(MessageChange {
746            index,
747            id,
748            kind: MessageChangeKind::Started,
749        })
750    }
751
752    fn text_content(
753        &mut self,
754        id: &MessageId,
755        delta: &str,
756        owner: &Option<SubagentRunId>,
757    ) -> Result<Changed> {
758        if !self.by_id.contains_key(id) {
759            // Tolerant: an orphan content event opens the message it names.
760            self.text_start(id.clone(), TextMessageRole::Assistant, None, owner.clone());
761        }
762        let Some(index) = self.by_id.get(id).copied() else {
763            return Ok(Changed::Nothing);
764        };
765        let Some(message) = self.messages.get_mut(index) else {
766            return Ok(Changed::Nothing);
767        };
768        append_text(message, delta)?;
769        Ok(Changed::Message(MessageChange {
770            index,
771            id: id.clone(),
772            kind: MessageChangeKind::Content {
773                delta: delta.to_owned(),
774            },
775        }))
776    }
777
778    fn text_end(&mut self, id: &MessageId) -> Changed {
779        self.open_text.retain(|(_, open)| open != id);
780        self.message_change(id, MessageChangeKind::Ended)
781    }
782
783    /// A `TEXT_MESSAGE_CHUNK` applied directly.
784    ///
785    /// Reachable only for a caller driving the applier itself: a
786    /// [`Session`](crate::client::Session) puts a
787    /// [`ChunkNormalizer`](crate::client::ChunkNormalizer) in front, which expands
788    /// chunks before they get here. The difference is that the normalizer also
789    /// synthesizes the *end* of a chunk stream; this does not, because an
790    /// applier never invents an event nobody sent.
791    fn text_chunk(&mut self, event: &TextMessageChunkEvent) -> Result<Changed> {
792        let id = match &event.message_id {
793            Some(id) => id.clone(),
794            None => resolve_open(
795                &self.open_text,
796                &event.subagent_run_id,
797                "TEXT_MESSAGE_CHUNK",
798                "message",
799            )?,
800        };
801        if !self.open_text.iter().any(|(_, open)| open == &id) {
802            self.text_start(
803                id.clone(),
804                event.role.unwrap_or_default(),
805                event.name.clone(),
806                event.subagent_run_id.clone(),
807            );
808        }
809        match &event.delta {
810            Some(delta) => self.text_content(&id, delta, &event.subagent_run_id),
811            None => Ok(self.message_change(&id, MessageChangeKind::Started)),
812        }
813    }
814
815    // ---- tool calls -----------------------------------------------------
816
817    fn tool_call_start(
818        &mut self,
819        tool_call_id: ToolCallId,
820        name: String,
821        parent_message_id: Option<MessageId>,
822        owner: Option<SubagentRunId>,
823    ) -> Changed {
824        // A call with no parent belongs to a message of its own; the call id is
825        // the only id available to name it. A message created here takes the
826        // call's attribution; one that already exists keeps its own.
827        let parent =
828            parent_message_id.unwrap_or_else(|| MessageId::new(format!("{tool_call_id}-message")));
829        let index = match self.by_id.get(&parent) {
830            Some(index) => *index,
831            None => self.push_message(Message::Assistant(AssistantMessage {
832                id: parent.clone(),
833                subagent_run_id: owner,
834                ..Default::default()
835            })),
836        };
837        // The parent may exist and not be an assistant message, in which case
838        // there is nowhere to hang the call and nothing changes.
839        let Some(Message::Assistant(assistant)) = self.messages.get_mut(index) else {
840            return Changed::Nothing;
841        };
842        let calls = assistant.tool_calls.get_or_insert_with(Vec::new);
843        if !calls.iter().any(|call| call.id == tool_call_id) {
844            calls.push(ToolCall::new(tool_call_id.clone(), name.clone(), ""));
845        }
846        self.tool_calls.insert(tool_call_id.clone(), index);
847        Changed::Message(MessageChange {
848            index,
849            id: parent,
850            kind: MessageChangeKind::ToolCallStarted { tool_call_id, name },
851        })
852    }
853
854    fn tool_call_args(&mut self, tool_call_id: &ToolCallId, delta: &str) -> Result<Changed> {
855        let Some(index) = self.tool_calls.get(tool_call_id).copied() else {
856            return Err(Error::protocol(format!(
857                "TOOL_CALL_ARGS for unknown tool call {tool_call_id:?}"
858            )));
859        };
860        let Some(Message::Assistant(assistant)) = self.messages.get_mut(index) else {
861            return Ok(Changed::Nothing);
862        };
863        let Some(call) = assistant
864            .tool_calls
865            .as_mut()
866            .and_then(|calls| calls.iter_mut().find(|call| &call.id == tool_call_id))
867        else {
868            return Ok(Changed::Nothing);
869        };
870        call.function.arguments.push_str(delta);
871        Ok(Changed::Message(MessageChange {
872            index,
873            id: assistant.id.clone(),
874            kind: MessageChangeKind::ToolCallArgs {
875                tool_call_id: tool_call_id.clone(),
876                delta: delta.to_owned(),
877            },
878        }))
879    }
880
881    fn tool_call_end(&mut self, tool_call_id: &ToolCallId) -> Changed {
882        match self.tool_calls.get(tool_call_id).copied() {
883            Some(index) => match self.messages.get(index) {
884                Some(message) => Changed::Message(MessageChange {
885                    index,
886                    id: message.id().clone(),
887                    kind: MessageChangeKind::ToolCallEnded {
888                        tool_call_id: tool_call_id.clone(),
889                    },
890                }),
891                None => Changed::Nothing,
892            },
893            None => Changed::Nothing,
894        }
895    }
896
897    fn tool_call_chunk(&mut self, event: &ToolCallChunkEvent) -> Result<Changed> {
898        let known = event
899            .tool_call_id
900            .as_ref()
901            .is_some_and(|id| self.tool_calls.contains_key(id));
902        match (&event.tool_call_id, known) {
903            (Some(id), false) => {
904                let Some(name) = event.tool_call_name.clone() else {
905                    return Err(Error::protocol(format!(
906                        "TOOL_CALL_CHUNK opens tool call {id:?} without a toolCallName"
907                    )));
908                };
909                let started = self.tool_call_start(
910                    id.clone(),
911                    name,
912                    event.parent_message_id.clone(),
913                    event.subagent_run_id.clone(),
914                );
915                match &event.delta {
916                    Some(delta) => self.tool_call_args(id, delta),
917                    None => Ok(started),
918                }
919            }
920            (Some(id), true) => match &event.delta {
921                Some(delta) => self.tool_call_args(id, delta),
922                None => Ok(self.tool_call_end(id)),
923            },
924            (None, _) => Err(Error::protocol(
925                "TOOL_CALL_CHUNK carries no toolCallId and no call is open",
926            )),
927        }
928    }
929
930    fn tool_call_result(
931        &mut self,
932        message_id: MessageId,
933        tool_call_id: ToolCallId,
934        content: String,
935        owner: Option<SubagentRunId>,
936    ) -> Changed {
937        let index = match self.by_id.get(&message_id).copied() {
938            Some(index) => {
939                if let Some(Message::Tool(tool)) = self.messages.get_mut(index) {
940                    tool.content = content;
941                    // The newest mint wins, as both verifiers record it.
942                    tool.subagent_run_id = owner;
943                }
944                index
945            }
946            // The result's own attribution, not the call's: whoever executed
947            // the tool owns the message that reports it.
948            None => self.push_message(Message::Tool(ToolMessage {
949                id: message_id.clone(),
950                content,
951                tool_call_id: tool_call_id.clone(),
952                subagent_run_id: owner,
953                ..Default::default()
954            })),
955        };
956        Changed::Message(MessageChange {
957            index,
958            id: message_id,
959            kind: MessageChangeKind::ToolResult { tool_call_id },
960        })
961    }
962
963    // ---- activities -----------------------------------------------------
964
965    fn activity_snapshot(&mut self, event: &ActivitySnapshotEvent) -> Changed {
966        let is_new = !self.by_id.contains_key(&event.message_id);
967        let index = self.activity_index(&event.message_id, &event.activity_type);
968        if let Some(Message::Activity(activity)) = self.messages.get_mut(index) {
969            activity.activity_type = event.activity_type.clone();
970            // A replacing snapshot re-mints the activity, attribution
971            // included: one that carries none takes the activity back for
972            // the parent. A merge leaves the message where it was, and the
973            // verifier holds a following delta to that owner.
974            if is_new || event.replace {
975                activity.subagent_run_id = event.subagent_run_id.clone();
976            }
977            if event.replace {
978                activity.content = event.content.clone();
979            } else {
980                // `replace: false` is a merge, and RFC 7396 is the merge the
981                // protocol's sibling patch format defines.
982                let mut merged = Value::Object(std::mem::take(&mut activity.content));
983                json_patch::merge(&mut merged, &Value::Object(event.content.clone()));
984                if let Value::Object(object) = merged {
985                    activity.content = object;
986                }
987            }
988        }
989        Changed::Message(MessageChange {
990            index,
991            id: event.message_id.clone(),
992            kind: MessageChangeKind::Activity,
993        })
994    }
995
996    fn activity_delta(&mut self, event: &ActivityDeltaEvent) -> Result<Changed> {
997        let index = self.activity_index(&event.message_id, &event.activity_type);
998        if let Some(Message::Activity(activity)) = self.messages.get_mut(index) {
999            let what = format!("activity {}", event.message_id);
1000            // Patched on a copy, and committed only once the result is still an
1001            // object. An activity's content is an object by definition, so a
1002            // whole-document operation — `{"op":"replace","path":"","value":7}`
1003            // — has nowhere to land; taking the content out first would leave
1004            // the activity holding nothing at all.
1005            let mut content = Value::Object(activity.content.clone());
1006            apply_patch(&mut content, &event.patch, &what)?;
1007            let Value::Object(object) = content else {
1008                return Err(Error::Patch {
1009                    target: what,
1010                    message: format!(
1011                        "patch replaced the whole activity with {}, which is not an object",
1012                        kind_of(&content)
1013                    ),
1014                });
1015            };
1016            activity.content = object;
1017        }
1018        Ok(Changed::Message(MessageChange {
1019            index,
1020            id: event.message_id.clone(),
1021            kind: MessageChangeKind::Activity,
1022        }))
1023    }
1024
1025    /// The index of the activity message with this id, creating it if the
1026    /// producer patched an activity it never published.
1027    fn activity_index(&mut self, id: &MessageId, activity_type: &str) -> usize {
1028        match self.by_id.get(id).copied() {
1029            Some(index) => index,
1030            None => self.push_message(Message::Activity(ActivityMessage {
1031                id: id.clone(),
1032                activity_type: activity_type.to_owned(),
1033                content: JsonObject::new(),
1034                ..Default::default()
1035            })),
1036        }
1037    }
1038
1039    // ---- reasoning ------------------------------------------------------
1040
1041    /// Opens a reasoning message, or reports nothing when it is already open.
1042    ///
1043    /// `REASONING_START` and `REASONING_MESSAGE_START` both land here under the
1044    /// same id — the block and the message inside it — and so does the
1045    /// `THINKING_*` pair. Only the first of them starts anything.
1046    fn reasoning_start(&mut self, id: MessageId, owner: Option<SubagentRunId>) -> Changed {
1047        self.current_reasoning = Some(id.clone());
1048        self.reasoning_streams.retain(|(_, open)| open != &id);
1049        self.reasoning_streams.push((owner.clone(), id.clone()));
1050        if !self.reasoning_by_id.contains_key(&id) {
1051            self.reasoning_by_id
1052                .insert(id.clone(), self.reasoning.len());
1053            self.reasoning.push(ReasoningMessage {
1054                id: id.clone(),
1055                subagent_run_id: owner,
1056                ..Default::default()
1057            });
1058        }
1059        if !self.open_reasoning.insert(id.clone()) {
1060            return Changed::Nothing;
1061        }
1062        Changed::Reasoning(ReasoningChange {
1063            id,
1064            kind: ReasoningChangeKind::Started,
1065        })
1066    }
1067
1068    /// A fresh id for a `THINKING_*` block.
1069    ///
1070    /// Those events carry no `messageId` at all, so the applier has to invent
1071    /// one — and it has to be the same one for the whole block, which is what
1072    /// [`Applier::thinking_id`] is for.
1073    fn mint_thinking_id(&mut self) -> MessageId {
1074        self.thinking_counter += 1;
1075        MessageId::new(format!("thinking-{}", self.thinking_counter))
1076    }
1077
1078    /// The id a `THINKING_*` event belongs to, minting one when the producer
1079    /// sent content before its `THINKING_START`.
1080    ///
1081    /// Only resolves the id; opening it is the caller's, so that the caller can
1082    /// report whether the open was the one that started the thought.
1083    fn thinking_id(&mut self) -> MessageId {
1084        match self.current_reasoning.clone() {
1085            Some(id) => id,
1086            None => self.mint_thinking_id(),
1087        }
1088    }
1089
1090    fn reasoning_content(
1091        &mut self,
1092        id: &MessageId,
1093        delta: &str,
1094        owner: &Option<SubagentRunId>,
1095    ) -> Changed {
1096        if !self.reasoning_by_id.contains_key(id) {
1097            self.reasoning_start(id.clone(), owner.clone());
1098        }
1099        if let Some(message) = self
1100            .reasoning_by_id
1101            .get(id)
1102            .and_then(|index| self.reasoning.get_mut(*index))
1103        {
1104            message.content.push_str(delta);
1105        }
1106        Changed::Reasoning(ReasoningChange {
1107            id: id.clone(),
1108            kind: ReasoningChangeKind::Content {
1109                delta: delta.to_owned(),
1110            },
1111        })
1112    }
1113
1114    /// A `REASONING_MESSAGE_CHUNK` applied directly. See
1115    /// [`Applier::text_chunk`] for when that happens.
1116    fn reasoning_chunk(&mut self, event: &ReasoningMessageChunkEvent) -> Result<Changed> {
1117        let id = match &event.message_id {
1118            Some(id) => id.clone(),
1119            None => resolve_open(
1120                &self.reasoning_streams,
1121                &event.subagent_run_id,
1122                "REASONING_MESSAGE_CHUNK",
1123                "reasoning message",
1124            )?,
1125        };
1126        let started = if self.open_reasoning.contains(&id) {
1127            Changed::Nothing
1128        } else {
1129            self.reasoning_start(id.clone(), event.subagent_run_id.clone())
1130        };
1131        Ok(match &event.delta {
1132            Some(delta) => self.reasoning_content(&id, delta, &event.subagent_run_id),
1133            // A chunk with no delta only opens, so it reports whatever the open
1134            // did — nothing at all when the message was already open.
1135            None => started,
1136        })
1137    }
1138
1139    /// Closes a reasoning message, or reports nothing when it is already closed.
1140    ///
1141    /// The mirror of [`Applier::reasoning_start`]: `REASONING_MESSAGE_END` and
1142    /// `REASONING_END` are two events closing one thought, and a consumer that
1143    /// prints a finished thought has to see it finish once.
1144    fn reasoning_end(&mut self, id: &MessageId) -> Changed {
1145        if self.current_reasoning.as_ref() == Some(id) {
1146            self.current_reasoning = None;
1147        }
1148        self.reasoning_streams.retain(|(_, open)| open != id);
1149        if !self.open_reasoning.remove(id) {
1150            return Changed::Nothing;
1151        }
1152        Changed::Reasoning(ReasoningChange {
1153            id: id.clone(),
1154            kind: ReasoningChangeKind::Ended,
1155        })
1156    }
1157
1158    fn encrypted_value(&mut self, event: &ReasoningEncryptedValueEvent) -> Changed {
1159        let blob = event.encrypted_value.clone();
1160        match event.subtype {
1161            ReasoningEncryptedValueSubtype::ToolCall => {
1162                let tool_call_id = ToolCallId::new(event.entity_id.clone());
1163                let Some(index) = self.tool_calls.get(&tool_call_id).copied() else {
1164                    return Changed::Nothing;
1165                };
1166                let Some(Message::Assistant(assistant)) = self.messages.get_mut(index) else {
1167                    return Changed::Nothing;
1168                };
1169                if let Some(call) = assistant
1170                    .tool_calls
1171                    .as_mut()
1172                    .and_then(|calls| calls.iter_mut().find(|call| call.id == tool_call_id))
1173                {
1174                    call.encrypted_value = Some(blob);
1175                }
1176                Changed::Message(MessageChange {
1177                    index,
1178                    id: assistant.id.clone(),
1179                    kind: MessageChangeKind::EncryptedValue,
1180                })
1181            }
1182            ReasoningEncryptedValueSubtype::Message => {
1183                let id = MessageId::new(event.entity_id.clone());
1184                if let Some(message) = self
1185                    .reasoning_by_id
1186                    .get(&id)
1187                    .and_then(|index| self.reasoning.get_mut(*index))
1188                {
1189                    message.encrypted_value = Some(blob);
1190                    return Changed::Reasoning(ReasoningChange {
1191                        id,
1192                        kind: ReasoningChangeKind::EncryptedValue,
1193                    });
1194                }
1195                let Some(index) = self.by_id.get(&id).copied() else {
1196                    return Changed::Nothing;
1197                };
1198                if let Some(message) = self.messages.get_mut(index) {
1199                    set_encrypted_value(message, blob);
1200                }
1201                Changed::Message(MessageChange {
1202                    index,
1203                    id,
1204                    kind: MessageChangeKind::EncryptedValue,
1205                })
1206            }
1207        }
1208    }
1209
1210    // ---- subagents ------------------------------------------------------
1211
1212    fn subagent_started(&mut self, event: &SubagentStartedEvent) -> Changed {
1213        let run_id = event.subagent_run_id.clone();
1214        if let Some(index) = self.subagent_by_id.get(&run_id).copied() {
1215            // The one legal re-announcement: a suspended invocation continuing
1216            // on the resuming run. Anything else is a producer bug the
1217            // verifier already reported, and the entry is reset rather than
1218            // duplicated so a view keeps one row per id.
1219            let subagent = &mut self.subagents[index];
1220            let kind = if matches!(subagent.status, SubagentStatus::Suspended { .. }) {
1221                SubagentChangeKind::Resumed
1222            } else {
1223                SubagentChangeKind::Started
1224            };
1225            subagent.name.clone_from(&event.name);
1226            if event.description.is_some() {
1227                subagent.description.clone_from(&event.description);
1228            }
1229            if event.parent_subagent_run_id.is_some() {
1230                subagent
1231                    .parent_subagent_run_id
1232                    .clone_from(&event.parent_subagent_run_id);
1233            }
1234            if event.parent_tool_call_id.is_some() {
1235                subagent
1236                    .parent_tool_call_id
1237                    .clone_from(&event.parent_tool_call_id);
1238                subagent
1239                    .parent_message_id
1240                    .clone_from(&event.parent_message_id);
1241            }
1242            subagent.status = SubagentStatus::Running;
1243            return Changed::Subagent(SubagentChange {
1244                index,
1245                run_id,
1246                kind,
1247            });
1248        }
1249        let index = self.push_subagent(Subagent {
1250            run_id: run_id.clone(),
1251            name: event.name.clone(),
1252            description: event.description.clone(),
1253            parent_subagent_run_id: event.parent_subagent_run_id.clone(),
1254            parent_tool_call_id: event.parent_tool_call_id.clone(),
1255            parent_message_id: event.parent_message_id.clone(),
1256            status: SubagentStatus::Running,
1257        });
1258        Changed::Subagent(SubagentChange {
1259            index,
1260            run_id,
1261            kind: SubagentChangeKind::Started,
1262        })
1263    }
1264
1265    fn subagent_finished(&mut self, event: &SubagentFinishedEvent) -> Changed {
1266        let run_id = event.subagent_run_id.clone();
1267        let index = self.subagent_index(&run_id);
1268        let (status, kind) = match &event.outcome {
1269            Some(SubagentOutcome::Suspended { interrupt_ids }) => (
1270                SubagentStatus::Suspended {
1271                    result: event.result.clone(),
1272                    interrupt_ids: interrupt_ids.clone().unwrap_or_default(),
1273                },
1274                SubagentChangeKind::Suspended,
1275            ),
1276            // Absent means success: the legacy reading.
1277            Some(SubagentOutcome::Success) | None => (
1278                SubagentStatus::Finished {
1279                    result: event.result.clone(),
1280                },
1281                SubagentChangeKind::Finished,
1282            ),
1283        };
1284        self.subagents[index].status = status;
1285        Changed::Subagent(SubagentChange {
1286            index,
1287            run_id,
1288            kind,
1289        })
1290    }
1291
1292    fn subagent_error(&mut self, event: &SubagentErrorEvent) -> Changed {
1293        let run_id = event.subagent_run_id.clone();
1294        let index = self.subagent_index(&run_id);
1295        self.subagents[index].status = SubagentStatus::Failed {
1296            message: event.message.clone(),
1297            code: event.code.clone(),
1298        };
1299        Changed::Subagent(SubagentChange {
1300            index,
1301            run_id,
1302            kind: SubagentChangeKind::Failed,
1303        })
1304    }
1305
1306    /// The index of the subagent with this id, creating a bare entry when a
1307    /// producer closed one it never announced. Tolerant, like the rest of the
1308    /// applier: the verifier is where that is a complaint, and a view still
1309    /// wants a row to hang the outcome on.
1310    fn subagent_index(&mut self, run_id: &SubagentRunId) -> usize {
1311        match self.subagent_by_id.get(run_id).copied() {
1312            Some(index) => index,
1313            None => self.push_subagent(Subagent {
1314                run_id: run_id.clone(),
1315                name: run_id.to_string(),
1316                description: None,
1317                parent_subagent_run_id: None,
1318                parent_tool_call_id: None,
1319                parent_message_id: None,
1320                status: SubagentStatus::Running,
1321            }),
1322        }
1323    }
1324
1325    fn push_subagent(&mut self, subagent: Subagent) -> usize {
1326        let index = self.subagents.len();
1327        self.subagent_by_id.insert(subagent.run_id.clone(), index);
1328        self.subagents.push(subagent);
1329        index
1330    }
1331
1332    // ---- metadata -------------------------------------------------------
1333
1334    /// Folds an event's metadata into what the event built: the message for
1335    /// the text, result and activity families, the tool call for `TOOL_CALL_*`,
1336    /// the reasoning message for `REASONING_MESSAGE_*`. Everything else keeps
1337    /// its metadata to itself — see [`crate::metadata`].
1338    fn merge_event_metadata(&mut self, event: &Event, changed: &Changed, metadata: &JsonObject) {
1339        match event {
1340            Event::TextMessageStart(_)
1341            | Event::TextMessageContent(_)
1342            | Event::TextMessageEnd(_)
1343            | Event::TextMessageChunk(_)
1344            | Event::ToolCallResult(_)
1345            | Event::ActivitySnapshot(_)
1346            | Event::ActivityDelta(_) => {
1347                let Changed::Message(change) = changed else {
1348                    return;
1349                };
1350                if let Some(message) = self.messages.get_mut(change.index) {
1351                    merge_metadata_into(message.metadata_mut(), Some(metadata));
1352                }
1353            }
1354            Event::ToolCallStart(_)
1355            | Event::ToolCallArgs(_)
1356            | Event::ToolCallEnd(_)
1357            | Event::ToolCallChunk(_) => {
1358                let Changed::Message(change) = changed else {
1359                    return;
1360                };
1361                let tool_call_id = match &change.kind {
1362                    MessageChangeKind::ToolCallStarted { tool_call_id, .. }
1363                    | MessageChangeKind::ToolCallArgs { tool_call_id, .. }
1364                    | MessageChangeKind::ToolCallEnded { tool_call_id } => tool_call_id.clone(),
1365                    _ => return,
1366                };
1367                let Some(Message::Assistant(assistant)) = self.messages.get_mut(change.index)
1368                else {
1369                    return;
1370                };
1371                if let Some(call) = assistant
1372                    .tool_calls
1373                    .as_mut()
1374                    .and_then(|calls| calls.iter_mut().find(|call| call.id == tool_call_id))
1375                {
1376                    merge_metadata_into(&mut call.metadata, Some(metadata));
1377                }
1378            }
1379            Event::ReasoningMessageStart(_)
1380            | Event::ReasoningMessageContent(_)
1381            | Event::ReasoningMessageEnd(_)
1382            | Event::ReasoningMessageChunk(_) => {
1383                let id = match (event, changed) {
1384                    (Event::ReasoningMessageStart(e), _) => e.message_id.clone(),
1385                    (Event::ReasoningMessageContent(e), _) => e.message_id.clone(),
1386                    (Event::ReasoningMessageEnd(e), _) => e.message_id.clone(),
1387                    (_, Changed::Reasoning(change)) => change.id.clone(),
1388                    _ => return,
1389                };
1390                if let Some(message) = self
1391                    .reasoning_by_id
1392                    .get(&id)
1393                    .and_then(|index| self.reasoning.get_mut(*index))
1394                {
1395                    merge_metadata_into(&mut message.metadata, Some(metadata));
1396                }
1397            }
1398            _ => {}
1399        }
1400    }
1401}
1402
1403/// The stream a chunk without an id continues: the one its subagent has
1404/// open; for an untagged chunk the parent's, else the sole open one — and when
1405/// several subagents could claim it, an error rather than a guess.
1406fn resolve_open(
1407    streams: &[(Option<SubagentRunId>, MessageId)],
1408    tag: &Option<SubagentRunId>,
1409    kind: &str,
1410    what: &str,
1411) -> Result<MessageId> {
1412    if tag.is_some() {
1413        return streams
1414            .iter()
1415            .rev()
1416            .find(|(owner, _)| owner == tag)
1417            .map(|(_, id)| id.clone())
1418            .ok_or_else(|| {
1419                Error::protocol(format!(
1420                    "{kind} carries no messageId and subagent {:?} has no {what} open",
1421                    tag.as_deref().unwrap_or_default()
1422                ))
1423            });
1424    }
1425    if let Some((_, id)) = streams.iter().rev().find(|(owner, _)| owner.is_none()) {
1426        return Ok(id.clone());
1427    }
1428    match streams {
1429        [] => Err(Error::protocol(format!(
1430            "{kind} carries no messageId and no {what} is open"
1431        ))),
1432        [(_, id)] => Ok(id.clone()),
1433        _ => Err(Error::protocol(format!(
1434            "{kind} carries no messageId and several subagents have a {what} open; \
1435             attribute the chunk"
1436        ))),
1437    }
1438}
1439
1440/// Applies an RFC 6902 patch, leaving `target` untouched when it fails.
1441fn apply_patch(target: &mut Value, operations: &[PatchOperation], what: &str) -> Result<()> {
1442    // The protocol's operation type and the patch engine's are both the RFC
1443    // wire format, so JSON is the conversion. Deserializing is also where a
1444    // malformed JSON Pointer is caught, before anything is mutated.
1445    let document = serde_json::to_value(operations)?;
1446    let patch: json_patch::Patch =
1447        serde_json::from_value(document).map_err(|error| Error::Patch {
1448            target: what.to_owned(),
1449            message: format!("invalid patch document: {error}"),
1450        })?;
1451    json_patch::patch(target, &patch).map_err(|error| Error::Patch {
1452        target: what.to_owned(),
1453        message: error.to_string(),
1454    })
1455}
1456
1457/// Names a JSON value's type, for an error message.
1458fn kind_of(value: &Value) -> &'static str {
1459    match value {
1460        Value::Null => "null",
1461        Value::Bool(_) => "a boolean",
1462        Value::Number(_) => "a number",
1463        Value::String(_) => "a string",
1464        Value::Array(_) => "an array",
1465        Value::Object(_) => "an object",
1466    }
1467}
1468
1469/// Builds the empty message a `TEXT_MESSAGE_START` opens.
1470fn empty_message(id: MessageId, role: TextMessageRole, name: Option<String>) -> Message {
1471    match role {
1472        TextMessageRole::Assistant => Message::Assistant(AssistantMessage {
1473            id,
1474            content: Some(String::new()),
1475            name,
1476            ..Default::default()
1477        }),
1478        TextMessageRole::User => Message::User(UserMessage {
1479            id,
1480            content: UserContent::Text(String::new()),
1481            name,
1482            ..Default::default()
1483        }),
1484        TextMessageRole::System => Message::System(SystemMessage {
1485            id,
1486            content: String::new(),
1487            name,
1488            ..Default::default()
1489        }),
1490        TextMessageRole::Developer => Message::Developer(DeveloperMessage {
1491            id,
1492            content: String::new(),
1493            name,
1494            ..Default::default()
1495        }),
1496    }
1497}
1498
1499/// Appends streamed text to whichever field of the message carries it.
1500fn append_text(message: &mut Message, delta: &str) -> Result<()> {
1501    match message {
1502        Message::Assistant(m) => m.content.get_or_insert_with(String::new).push_str(delta),
1503        Message::System(m) => m.content.push_str(delta),
1504        Message::Developer(m) => m.content.push_str(delta),
1505        Message::Reasoning(m) => m.content.push_str(delta),
1506        Message::Tool(m) => m.content.push_str(delta),
1507        Message::User(m) => match &mut m.content {
1508            UserContent::Text(text) => text.push_str(delta),
1509            UserContent::Parts(parts) => match parts.last_mut() {
1510                Some(InputContent::Text(text)) => text.text.push_str(delta),
1511                _ => parts.push(InputContent::Text(TextInputContent {
1512                    text: delta.to_owned(),
1513                })),
1514            },
1515        },
1516        Message::Activity(m) => {
1517            return Err(Error::protocol(format!(
1518                "text streamed into activity message {:?}, which has no text",
1519                m.id
1520            )));
1521        }
1522    }
1523    Ok(())
1524}
1525
1526/// Attaches a provider's opaque reasoning blob to a message.
1527fn set_encrypted_value(message: &mut Message, blob: String) {
1528    match message {
1529        Message::Assistant(m) => m.encrypted_value = Some(blob),
1530        Message::System(m) => m.encrypted_value = Some(blob),
1531        Message::Developer(m) => m.encrypted_value = Some(blob),
1532        Message::User(m) => m.encrypted_value = Some(blob),
1533        Message::Tool(m) => m.encrypted_value = Some(blob),
1534        Message::Reasoning(m) => m.encrypted_value = Some(blob),
1535        // An activity has no field for it.
1536        Message::Activity(_) => {}
1537    }
1538}
ag_ui/client/apply.rs

ag_ui/client/
apply.rs
