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/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()); } 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 {