From d1e6ec4192e5d917147c89e725886b61b4ccb45a Mon Sep 17 00:00:00 2001 From: "blu@blu.cx" <38222767+thebrandonlucas@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:20:04 -0600 Subject: [PATCH 1/2] perf(gpui): parse each agent activity record once when it arrives Agent activity records were kept as JSON text and parsed again on every read. Selecting the turns a message started (request_activity) asked raw_records for the channel's entries, which parsed each record and pretty-printed each batch child, then parsed that output again for the turn id, seq, kind, timestamp and triggers. The turn-start index built on each publish also re-parsed every live record. With layer 1 the bubble answers are remembered per snapshot, but each new snapshot still paid string -> parse -> pretty-print -> parse for every record. A record now keeps the few fields selection needs, read from the JSON it was already parsed into on arrival: for the envelope, or each batch child, its channel, kind, usable turn id, bounded seq, timestamp and (for turn starts only) its string trigger ids. Events naming no channel, or with no usable turn id that are not turn starts, are not kept, as no selection can read them. request_activity and the turn-start index read these fields with no JSON parsing or serialization. The plaintext and the Raw view's output are unchanged. Selections are unchanged: a test compares request_activity against the previous parse-based reading over batches, other channels, invalid or non-object JSON, missing and oversized turn ids, seq and timestamp bounds, and duplicate starts. An empty channel, which no caller asks for, now selects nothing instead of reading unscoped records. Refs: DEV-3013 Signed-off-by: blu@blu.cx <38222767+thebrandonlucas@users.noreply.github.com> --- .../crates/app/src/agent_activity/current.rs | 1 + .../crates/app/src/agent_activity/mod.rs | 91 +++--- .../crates/app/src/agent_activity/request.rs | 143 ++++++--- .../app/src/agent_activity/request/tests.rs | 294 ++++++++++++++++++ .../src/agent_activity/transcript_model.rs | 1 + .../crates/app/src/mission_control/tests.rs | 1 + 6 files changed, 436 insertions(+), 95 deletions(-) diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/current.rs b/apps/buzz-gpui/crates/app/src/agent_activity/current.rs index bf2997c2bea..34ab9a14df1 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/current.rs +++ b/apps/buzz-gpui/crates/app/src/agent_activity/current.rs @@ -721,6 +721,7 @@ mod tests { historical: false, kind: plaintext["kind"].as_str().unwrap_or("unknown").into(), channel_ids: vec!["c1".into()], + events: super::super::request::events(&plaintext), }) } diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs b/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs index 07a473da441..fa4e92fb365 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs +++ b/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs @@ -84,6 +84,39 @@ pub struct Record { pub kind: String, /// The channels the envelope and its batch children name. pub channel_ids: Vec, + /// What request selection reads, parsed from `plaintext` on arrival + /// so views never parse it again. + pub events: Vec, +} + +impl Record { + /// A record of `frame`, whose plaintext parsed as `raw`. Everything + /// views select by is read from `raw` here, once. + fn new(frame: Frame, raw: &Value, received_at: u64, historical: bool) -> Self { + let mut channel_ids: Vec = Vec::new(); + let mut values = vec![raw]; + if raw.get("kind").and_then(Value::as_str) == Some("batch") { + values.extend(children(raw)); + } + for value in values { + if let Some(channel) = text(value.get("channelId")) + && !channel_ids.iter().any(|c| c == channel) + { + channel_ids.push(channel.to_string()); + } + } + Self { + id: frame.id, + agent: frame.agent, + created_at: frame.created_at, + plaintext: frame.plaintext.into(), + received_at, + historical, + kind: text(raw.get("kind")).unwrap_or("unknown").to_string(), + channel_ids, + events: request::events(raw), + } + } } /// Whether the live feed is receiving (`Snapshot.status`). @@ -210,9 +243,12 @@ pub struct HistoryRead { /// A short non-empty string (`text` in activity.ts). pub(crate) fn text(value: Option<&Value>) -> Option<&str> { - value - .and_then(Value::as_str) - .filter(|s| !s.is_empty() && s.chars().count() <= 256) + value.and_then(Value::as_str).filter(|s| short(s)) +} + +/// Non-empty and at most 256 characters. +fn short(value: &str) -> bool { + !value.is_empty() && value.chars().count() <= 256 } /// The envelope's children: a batch's events, or the envelope itself. @@ -438,30 +474,8 @@ impl Activity { // A live copy replaces the saved one. self.records.remove(index); } - let mut channel_ids: Vec = Vec::new(); - let mut values = vec![raw]; - if raw.get("kind").and_then(Value::as_str) == Some("batch") { - values.extend(children(raw)); - } - for value in values { - if let Some(channel) = text(value.get("channelId")) - && !channel_ids.iter().any(|c| c == channel) - { - channel_ids.push(channel.to_string()); - } - } - let kind = text(raw.get("kind")).unwrap_or("unknown").to_string(); let size = frame.plaintext.len(); - let record = Arc::new(Record { - id: frame.id, - agent: frame.agent, - created_at: frame.created_at, - plaintext: frame.plaintext.into(), - received_at, - historical, - kind, - channel_ids, - }); + let record = Arc::new(Record::new(frame, raw, received_at, historical)); let at = self .records .partition_point(|r| r.received_at <= record.received_at); @@ -714,29 +728,12 @@ impl Activity { fn requested(records: &[Arc]) -> Vec<(String, String)> { let mut pairs: Vec<(String, String)> = Vec::new(); for record in records.iter().filter(|r| !r.historical) { - if record.kind != "turn_started" && record.kind != "batch" { - continue; - } - let Ok(raw) = serde_json::from_str::(&record.plaintext) else { - continue; - }; - let mut values = vec![&raw]; - if record.kind == "batch" { - values.extend(children(&raw)); - } - for value in values { - if value.get("kind").and_then(Value::as_str) != Some("turn_started") { + for event in &record.events { + if event.kind != "turn_started" || !short(&event.channel) { continue; } - let Some(channel) = text(value.get("channelId")) else { - continue; - }; - let ids = value - .get("payload") - .and_then(|p| p.get("triggeringEventIds")) - .and_then(Value::as_array); - for id in ids.into_iter().flatten().filter_map(Value::as_str) { - let pair = (channel.to_string(), id.to_string()); + for id in &event.triggers { + let pair = (event.channel.clone(), id.clone()); if hex64(id) && !pairs.contains(&pair) { pairs.push(pair); } diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/request.rs b/apps/buzz-gpui/crates/app/src/agent_activity/request.rs index a5a43c502cd..5f9522ee872 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/request.rs +++ b/apps/buzz-gpui/crates/app/src/agent_activity/request.rs @@ -7,7 +7,7 @@ use super::local::{LocalWork, LocalWorkState, WorkStep, breathing, local_activity}; use super::transcript_model::{ItemKind, TranscriptTurn}; use super::turns::{ActivityTurn, TurnState, parse_time}; -use super::{FeedStatus, Record, Snapshot, raw_records}; +use super::{FeedStatus, Record, Snapshot, children}; use buzz_relay_client::EventId; use serde_json::Value; use std::collections::{BTreeSet, HashMap, HashSet}; @@ -108,9 +108,77 @@ pub struct Selected { pub uncertain: bool, } +/// What request selection reads from one envelope, or one batch child, +/// parsed once when its record arrives (`requestActivity` reads the same +/// fields from the JSON each time). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Event { + /// `channelId`, exactly as sent. + pub channel: String, + /// `kind`, or "" when absent. + pub kind: String, + /// `turnId`, when non-empty and at most 256 bytes. + pub turn: Option, + /// `seq`, when within `0..2^53`. + pub seq: Option, + /// `timestamp`, in milliseconds. + pub at: Option, + /// A turn start's string `payload.triggeringEventIds`; empty for other + /// kinds, which no selection asks. + pub triggers: Vec, +} + +/// The events of a record's parsed plaintext, in order: a batch's +/// children, else the envelope itself. Events naming no channel, and +/// events with no usable turn id that are not turn starts, are left out: +/// no selection can read them. +pub(crate) fn events(raw: &Value) -> Vec { + children(raw) + .into_iter() + .filter_map(|value| { + let channel = value.get("channelId")?.as_str()?; + let kind = value.get("kind").and_then(Value::as_str).unwrap_or(""); + let turn = value + .get("turnId") + .and_then(Value::as_str) + .filter(|turn| !turn.is_empty() && turn.len() <= 256); + let start = kind == "turn_started"; + if turn.is_none() && !start { + return None; + } + let triggers = value + .get("payload") + .and_then(|p| p.get("triggeringEventIds")) + .and_then(Value::as_array) + .filter(|_| start) + .map(|ids| { + ids.iter() + .filter_map(|id| id.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default(); + Some(Event { + channel: channel.to_string(), + kind: kind.to_string(), + turn: turn.map(str::to_string), + seq: value + .get("seq") + .and_then(Value::as_i64) + .filter(|seq| *seq >= 0 && *seq < (1 << 53)), + at: value + .get("timestamp") + .and_then(Value::as_str) + .and_then(parse_time), + triggers, + }) + }) + .collect() +} + /// Turns whose start names one of `ids`, from captured trigger ids only: /// no inference from time or names, and no wider fallback when the start -/// is missing. +/// is missing. Reads the events parsed on arrival, never the JSON; an +/// empty `channel` names no channel and selects nothing. pub fn request_activity( records: &[Arc], agent: &str, @@ -119,52 +187,34 @@ pub fn request_activity( ) -> Selected { let mut selected = Selected::default(); let wanted: HashSet<&str> = ids.iter().copied().filter(|id| super::hex64(id)).collect(); - if wanted.is_empty() { + if wanted.is_empty() || channel.is_empty() { return selected; } - struct Parsed { - kind: String, - turn: String, + struct Parsed<'a> { + kind: &'a str, + turn: &'a str, seq: Option, admitted: bool, - triggers: Vec, + triggers: &'a [String], } - let parsed: Vec = raw_records(records, agent, Some(channel)) - .into_iter() - .filter_map(|record| { - let value: Value = serde_json::from_str(&record.plaintext).ok()?; - let turn = value.get("turnId")?.as_str()?; - if turn.is_empty() || turn.len() > 256 { - return None; - } - let admitted = value - .get("timestamp") - .and_then(Value::as_str) - .and_then(parse_time) - .is_some_and(|at| at <= record.received_at + 5000); - Some(Parsed { - kind: value - .get("kind") - .and_then(Value::as_str) - .unwrap_or("") - .into(), - turn: turn.into(), - seq: value - .get("seq") - .and_then(Value::as_i64) - .filter(|seq| *seq >= 0 && *seq < (1 << 53)), - admitted, - triggers: value - .get("payload") - .and_then(|p| p.get("triggeringEventIds")) - .and_then(Value::as_array) - .map(|ids| { - ids.iter() - .filter_map(|id| id.as_str().map(str::to_string)) - .collect() + // In receipt order, then batch order. + let parsed: Vec = records + .iter() + .filter(|record| record.agent == agent) + .flat_map(|record| { + record + .events + .iter() + .filter(|event| event.channel == channel) + .filter_map(|event| { + Some(Parsed { + kind: &event.kind, + turn: event.turn.as_deref()?, + seq: event.seq, + admitted: event.at.is_some_and(|at| at <= record.received_at + 5000), + triggers: &event.triggers, }) - .unwrap_or_default(), - }) + }) }) .collect(); for start in parsed.iter().filter(|p| p.kind == "turn_started") { @@ -191,7 +241,7 @@ pub fn request_activity( selected.uncertain = true; continue; } - selected.turn_ids.insert(start.turn.clone()); + selected.turn_ids.insert(start.turn.to_string()); if !start.admitted { selected.uncertain = true; continue; @@ -199,17 +249,14 @@ pub fn request_activity( // The first admitted terminal frame in receipt order, not the // largest seq: a later error must not rewrite an earlier end. let end = turn.iter().find(|p| { - matches!( - p.kind.as_str(), - "turn_completed" | "turn_error" | "agent_panic" - ) && p.admitted + matches!(p.kind, "turn_completed" | "turn_error" | "agent_panic") && p.admitted }); if let Some(end) = end && end.seq.is_some_and(|seq| seq >= from) { selected .terminal - .insert(start.turn.clone(), end.kind == "turn_completed"); + .insert(start.turn.to_string(), end.kind == "turn_completed"); } } selected diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs b/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs index 6b8c751d07d..94f369377b9 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs +++ b/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs @@ -300,3 +300,297 @@ fn a_memo_is_bounded_and_never_copied_or_compared() { assert!(copy.memo.work.lock().unwrap().is_empty()); assert_eq!(©, s); } + +/// A record as `admit` builds it; plaintext that is not JSON keeps no +/// events, as no read of it could parse it. +fn record(n: u64, agent: &str, plaintext: &str) -> Arc { + let raw = serde_json::from_str(plaintext).unwrap_or(Value::Null); + let frame = Frame { + id: format!("{n:064x}"), + agent: agent.into(), + created_at: NOW / 1000, + plaintext: plaintext.into(), + }; + Arc::new(Record::new(frame, &raw, NOW + n, false)) +} + +/// Kind, turn, seq, admitted and triggers, as read from the JSON. +type Reread = (String, String, Option, bool, Vec); + +/// `request_activity` as it read the JSON before records kept their +/// events: the Raw view's entries, each parsed again. +fn reread(records: &[Arc], agent: &str, channel: &str, ids: &[&str]) -> Selected { + let mut selected = Selected::default(); + let wanted: HashSet<&str> = ids + .iter() + .copied() + .filter(|id| super::super::hex64(id)) + .collect(); + if wanted.is_empty() { + return selected; + } + let parsed: Vec = super::super::raw_records(records, agent, Some(channel)) + .into_iter() + .filter_map(|record| { + let value: Value = serde_json::from_str(&record.plaintext).ok()?; + let turn = value.get("turnId")?.as_str()?; + if turn.is_empty() || turn.len() > 256 { + return None; + } + let admitted = value + .get("timestamp") + .and_then(Value::as_str) + .and_then(parse_time) + .is_some_and(|at| at <= record.received_at + 5000); + let kind = value.get("kind").and_then(Value::as_str).unwrap_or(""); + let seq = value + .get("seq") + .and_then(Value::as_i64) + .filter(|seq| *seq >= 0 && *seq < (1 << 53)); + let triggers = value + .get("payload") + .and_then(|p| p.get("triggeringEventIds")) + .and_then(Value::as_array) + .map(|ids| { + ids.iter() + .filter_map(|id| id.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default(); + Some((kind.into(), turn.into(), seq, admitted, triggers)) + }) + .collect(); + for (kind, start, from, admitted, triggers) in &parsed { + if kind != "turn_started" || !triggers.iter().any(|id| wanted.contains(id.as_str())) { + continue; + } + let turn: Vec<_> = parsed.iter().filter(|p| p.1 == *start).collect(); + if turn.iter().filter(|p| p.0 == "turn_started").count() != 1 + || turn.iter().any(|p| p.2.is_none()) + { + selected.uncertain = true; + continue; + } + let Some(from) = *from else { continue }; + let after: Vec = turn + .iter() + .filter_map(|p| p.2) + .filter(|s| *s >= from) + .collect(); + if after.iter().collect::>().len() != after.len() { + selected.uncertain = true; + continue; + } + selected.turn_ids.insert(start.clone()); + if !admitted { + selected.uncertain = true; + continue; + } + let end = turn.iter().find(|p| { + matches!( + p.0.as_str(), + "turn_completed" | "turn_error" | "agent_panic" + ) && p.3 + }); + if let Some(end) = end + && end.2.is_some_and(|seq| seq >= from) + { + selected + .terminal + .insert(start.clone(), end.0 == "turn_completed"); + } + } + selected +} + +fn frame(kind: &str, turn: Value, seq: Value, channel: &str, at: &str, triggers: Value) -> Value { + json!({ + "kind": kind, "turnId": turn, "seq": seq, "channelId": channel, + "timestamp": at, "payload": {"triggeringEventIds": triggers}, + }) +} + +#[test] +fn selection_reads_batch_children_in_their_own_channel() { + let (request, other) = (key('c'), key('d')); + let mut f = Fixture::new(); + let batch = json!({"kind": "batch", "payload": {"events": [ + frame("turn_started", json!("one"), json!(1), "channel", T0, json!([request])), + // The same turn elsewhere is another channel's business. + frame("turn_started", json!("one"), json!(1), "elsewhere", T0, json!([request])), + frame("turn_started", json!("two"), json!(1), "elsewhere", T0, json!([other])), + // Only a start's triggers are kept. + frame("turn_completed", json!("one"), json!(2), "channel", T0, json!([request])), + ]}}); + f.activity.receive( + Frame { + id: key('9'), + agent: key('a'), + created_at: NOW / 1000, + plaintext: batch.to_string(), + }, + NOW, + &|_| true, + ); + let records = &f.snapshot().records; + let selected = request_activity(records, &key('a'), "channel", &[&request]); + assert_eq!(selected.turn_ids, ["one".to_string()].into()); + assert_eq!(selected.terminal, [("one".to_string(), true)].into()); + assert!(!selected.uncertain); + assert!( + request_activity(records, &key('a'), "channel", &[&other]) + .turn_ids + .is_empty() + ); + let elsewhere = request_activity(records, &key('a'), "elsewhere", &[&other]); + assert_eq!(elsewhere.turn_ids, ["two".to_string()].into()); + assert!(elsewhere.terminal.is_empty()); + assert_eq!( + &*f.snapshot().requested, + [ + ("channel".to_string(), request.clone()), + ("elsewhere".to_string(), request.clone()), + ("elsewhere".to_string(), other.clone()), + ] + ); + // Only events selection can read are kept. + assert_eq!(records[0].events.len(), 4); + assert!(records[0].events[3].triggers.is_empty()); +} + +#[test] +fn selection_matches_reading_the_json_on_every_edge() { + let (a, b) = (key('a'), key('b')); + let (c, d, e) = (key('c'), key('d'), key('e')); + let (f, g, h) = (key('f'), key('7'), key('8')); + // The batch below arrives at NOW + 7: 12:00:10.007. + let late = "2026-10-10T12:00:16Z"; + let timed = |turn: &str, at: &str, trigger: &str| { + [ + frame( + "turn_started", + json!(turn), + json!(1), + "channel", + at, + json!([trigger]), + ), + frame( + "turn_completed", + json!(turn), + json!(2), + "channel", + T0, + json!([]), + ), + ] + }; + let one = |kind: &str, turn: &str, seq: Value, triggers: Value| { + frame(kind, json!(turn), seq, "channel", T0, triggers) + }; + let mut plaintexts = vec![ + one("turn_started", "plain", json!(1), json!([c])).to_string(), + one("turn_completed", "plain", json!(2), json!([])).to_string(), + "not json".to_string(), + "[1, 2]".to_string(), + json!({"kind": "batch"}).to_string(), + json!({"kind": "batch", "payload": {"events": "no"}}).to_string(), + // A turn id on a batch envelope is not an event. + json!({"kind": "batch", "turnId": "plain", "channelId": "channel", "seq": 9}).to_string(), + ]; + let mut children = vec![ + one("turn_started", "batched", json!(1), json!([1, d, null])), + one("turn_error", "batched", json!(3), json!([])), + one("turn_completed", "batched", json!(4), json!([])), + json!({"kind": "turn_started", "turnId": "nochannel", "seq": 1, "payload": {"triggeringEventIds": [e]}}), + json!({"turnId": "nokind", "seq": 1, "channelId": "channel", "timestamp": T0}), + one("turn_started", "", json!(1), json!([e])), + one("turn_started", &"x".repeat(257), json!(1), json!([e])), + one("turn_started", &"y".repeat(256), json!(1), json!([e])), + one("turn_started", "negative", json!(-1), json!([e])), + one("turn_started", "huge", json!(1u64 << 53), json!([e])), + one("turn_started", "edge", json!((1u64 << 53) - 1), json!([e])), + one("turn_started", "fraction", json!(1.5), json!([e])), + one("turn_started", "repeat", json!(1), json!([e])), + one("turn_liveness", "repeat", json!(1), json!([])), + frame( + "turn_started", + json!("late"), + json!(1), + "channel", + late, + json!([g]), + ), + frame( + "turn_started", + json!("untimed"), + json!(1), + "channel", + "soon", + json!([e]), + ), + frame( + "turn_started", + json!(7), + json!(1), + "channel", + T0, + json!([e]), + ), + json!("not an object"), + ]; + children.extend(timed("ontime", "2026-10-10T12:00:15.007Z", &f)); + children.extend(timed("overdue", "2026-10-10T12:00:15.008Z", &h)); + plaintexts.push(json!({"kind": "batch", "payload": {"events": children}}).to_string()); + plaintexts.push(one("turn_started", "plain", json!(5), json!(c)).to_string()); + let mut records: Vec> = plaintexts + .iter() + .enumerate() + .map(|(n, text)| record(n as u64, &a, text)) + .collect(); + records.push(record( + 99, + &b, + &one("turn_started", "theirs", json!(1), json!([c])).to_string(), + )); + let id_sets: [&[&str]; 9] = [ + &[&c], + &[&d], + &[&e], + &[&f], + &[&g], + &[&h], + &[&c, &d, &e, &f, &g, &h], + &["short"], + &[], + ]; + let mut seen = Selected::default(); + for agent in [&a, &b] { + for channel in ["channel", "elsewhere"] { + for ids in id_sets { + let selected = request_activity(&records, agent, channel, ids); + assert_eq!( + selected, + reread(&records, agent, channel, ids), + "{channel} {ids:?}" + ); + seen.turn_ids.extend(selected.turn_ids); + seen.terminal.extend(selected.terminal); + seen.uncertain |= selected.uncertain; + } + } + } + // The corpus reaches selections, ends and doubt. + assert!(seen.turn_ids.contains("batched") && seen.turn_ids.contains(&"y".repeat(256))); + assert!(seen.turn_ids.contains("edge") && seen.turn_ids.contains("late")); + assert_eq!(seen.terminal.get("batched"), Some(&false)); + assert_eq!(seen.terminal.get("ontime"), Some(&true)); + assert!(!seen.terminal.contains_key("overdue")); + assert!(seen.uncertain); + assert!(!seen.turn_ids.contains("huge") && !seen.turn_ids.contains("nochannel")); + // An empty channel names none. + assert_eq!( + request_activity(&records, &a, "", &[&c]), + Selected::default() + ); +} diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/transcript_model.rs b/apps/buzz-gpui/crates/app/src/agent_activity/transcript_model.rs index e4c5be54cb7..63510a96196 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/transcript_model.rs +++ b/apps/buzz-gpui/crates/app/src/agent_activity/transcript_model.rs @@ -974,6 +974,7 @@ mod tests { historical: false, kind: "batch".into(), channel_ids: Vec::new(), + events: super::super::request::events(&plaintext), }) } diff --git a/apps/buzz-gpui/crates/app/src/mission_control/tests.rs b/apps/buzz-gpui/crates/app/src/mission_control/tests.rs index c3b487a301a..c48ca08c314 100644 --- a/apps/buzz-gpui/crates/app/src/mission_control/tests.rs +++ b/apps/buzz-gpui/crates/app/src/mission_control/tests.rs @@ -41,6 +41,7 @@ fn snapshot(events: Vec, state: TurnState, evidence: u64) -> Snapshot { historical: false, kind: "batch".into(), channel_ids: vec!["c1".into()], + events: crate::agent_activity::request::events(&plaintext), })] .into(), turns: vec![ActivityTurn { From 3110c2f92fad73952e4a077c559fa3085fecf492 Mon Sep 17 00:00:00 2001 From: Brandon Lucas <38222767+thebrandonlucas@users.noreply.github.com> Date: Thu, 5 Nov 2026 23:31:08 +0000 Subject: [PATCH 2/2] fix(gpui): unregister idle codex servers before stopping agents Co-Authored-By: Claude Sonnet 5.5 --- .../crates/app/src/agent_plugins/codex/runtime.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/codex/runtime.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/codex/runtime.rs index 986e2dd7c84..2727fc49558 100644 --- a/apps/buzz-gpui/crates/app/src/agent_plugins/codex/runtime.rs +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/codex/runtime.rs @@ -712,6 +712,7 @@ impl CodexRuntime { entries.retain(|pubkey, _| pubkeys.contains(pubkey)); gone }; + let mut stops = Vec::new(); for entry in gone { entry.abort.abort(ABORTED); locked(&self.inner.views).remove(&entry.pubkey); @@ -719,7 +720,7 @@ impl CodexRuntime { .get(&entry.spawn.host()) .cloned(); if let Some(server) = server { - self.stop(server, entry); + stops.push((server, entry)); } } let mut hosts: Vec = locked(&self.inner.entries) @@ -741,6 +742,11 @@ impl CodexRuntime { servers.retain(|host, _| hosts.contains(host)); idle }; + // Idle servers are unregistered above, so a stop that sees its + // server missing can tell the server closed. + for (server, entry) in stops { + self.stop(server, entry); + } for server in idle { drop(server.rpc.close()); }