diff --git a/crates/keiki-api/src/lib.rs b/crates/keiki-api/src/lib.rs index 1cdbf4007..252ed24dd 100644 --- a/crates/keiki-api/src/lib.rs +++ b/crates/keiki-api/src/lib.rs @@ -2,10 +2,11 @@ use std::time::{Duration, Instant}; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; pub use keiki_model::{ - AgentGroupSummary, AgentInput, AgentSummary, AgentTemplateSummary, AvatarState, AvatarTheme, - BlockConversationResponse, ClearConversationResponse, ConversationDetail, ConversationLocator, - ConversationSearchHit, ConversationSummary, ConversationTakeover, ConversationThreadPeer, - CreateAgentFromTemplate, CreateAgentResponse, EndConversationResponse, McpPreset, + AgentGroupRoster, AgentGroupSummary, AgentInput, AgentSummary, AgentTemplateSummary, + AvatarState, AvatarTheme, BlockConversationResponse, CancelSubagentTaskResponse, + ClearConversationResponse, ConversationDetail, ConversationLocator, ConversationSearchHit, + ConversationSummary, ConversationTakeover, ConversationThreadPeer, CreateAgentFromTemplate, + CreateAgentResponse, EndAgentGroupResponse, EndConversationResponse, McpPreset, OrganizationSummary, SendConversationMessageResponse, SessionResponse, SessionUser, SteerConversationResponse, SwitchOrgResponse, TakeoverResponse, }; @@ -601,6 +602,54 @@ impl Client { Ok(response.groups) } + pub async fn agent_group_roster( + &self, + access_token: &str, + group_id: &str, + ) -> Result { + self.send_json( + self.http + .get(self.endpoint(&format!( + "/api/webapp/agent-groups/{}/roster", + utf8_percent_encode(group_id, NON_ALPHANUMERIC) + ))) + .bearer_auth(access_token), + ) + .await + } + + pub async fn end_agent_group( + &self, + access_token: &str, + group_id: &str, + ) -> Result { + self.send_json( + self.http + .post(self.endpoint(&format!( + "/api/webapp/agent-groups/{}/end", + utf8_percent_encode(group_id, NON_ALPHANUMERIC) + ))) + .bearer_auth(access_token), + ) + .await + } + + pub async fn cancel_subagent_task( + &self, + access_token: &str, + task_id: &str, + ) -> Result { + self.send_json( + self.http + .post(self.endpoint(&format!( + "/api/webapp/subagent-tasks/{}/cancel", + utf8_percent_encode(task_id, NON_ALPHANUMERIC) + ))) + .bearer_auth(access_token), + ) + .await + } + /// Ask the daemon to (re)start a saved preset's authorization. The /// returned preset carries the provider URL to open in the user's browser /// when one is needed — the same response the dashboard's connect button diff --git a/crates/keiki-model/src/lib.rs b/crates/keiki-model/src/lib.rs index b500f7575..7f2028ced 100644 --- a/crates/keiki-model/src/lib.rs +++ b/crates/keiki-model/src/lib.rs @@ -112,6 +112,71 @@ pub struct AgentGroupsResponse { pub groups: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum AgentGroupMemberRole { + Hub, + Spoke, + Peer, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum AgentGroupMemberState { + Running, + Waiting, + Idle, + Ended, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentGroupMemberStatus { + pub agent_id: String, + pub name: String, + pub role: AgentGroupMemberRole, + pub status: AgentGroupMemberState, + pub running_turns: u32, + pub pending_asks: u32, + pub waiting_on_peer: u32, + pub background_tasks: u32, + pub tokens_in: u64, + pub tokens_out: u64, + pub cost_usd: f64, + pub last_activity_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentGroupRosterTotals { + pub members: u32, + pub hubs: u32, + pub spokes: u32, + pub running: u32, + pub waiting: u32, + pub ended: u32, + pub background_tasks: u32, + pub cost_usd: f64, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentGroupRoster { + pub members: Vec, + pub totals: AgentGroupRosterTotals, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct EndAgentGroupResponse { + pub threads_ended: u32, + pub tasks_cancelled: u32, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CancelSubagentTaskResponse { + pub cancelled: bool, +} + /// One saved MCP/service preset on an agent, as the connect and status /// endpoints return it. `authorization_url` is the provider page to open when /// `status` is `needs_auth`. @@ -322,6 +387,36 @@ pub enum MessageDirection { Outbound, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ConversationLiveness { + Working, + Waiting, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SubagentTaskStatus { + Running, + Suspended, + Completed, + Error, + Cancelled, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SubagentTaskSummary { + pub id: String, + pub subagent: String, + pub status: SubagentTaskStatus, + pub request: String, + pub resumptions: u32, + pub trace_id: Option, + pub created_at: String, + pub ended_at: Option, +} + impl MessageDirection { pub fn as_str(self) -> &'static str { match self { @@ -345,6 +440,8 @@ pub struct ConversationSummary { pub message_count: u32, pub is_active: bool, pub has_errors: bool, + #[serde(default)] + pub liveness: Option, /// The asker behind the thread — set only for inter-agent (`agent:`) /// conversations, where another agent is the contact. #[serde(default)] @@ -397,6 +494,7 @@ impl std::fmt::Debug for ConversationSummary { .field("message_count", &self.message_count) .field("is_active", &self.is_active) .field("has_errors", &self.has_errors) + .field("liveness", &self.liveness) .field("peer", &self.peer) .finish() } @@ -447,6 +545,10 @@ pub struct ConversationDetail { /// while the agent is idle. Older backends omit it. #[serde(default)] pub active_turn_started_at: Option, + #[serde(default)] + pub liveness: Option, + #[serde(default)] + pub tasks: Vec, /// The asker behind the thread when this is an inter-agent /// (`agent:`) conversation. #[serde(default)] @@ -855,4 +957,130 @@ mod tests { assert!(!format!("{agentless:?}").contains("agentless-secret")); assert!(!format!("{locator:?}").contains("agentless-secret")); } + + #[test] + fn swarm_observability_models_round_trip() { + let liveness = ConversationLiveness::Waiting; + assert_eq!( + serde_json::from_value::(serde_json::to_value(liveness).unwrap()) + .unwrap(), + liveness + ); + + let task = SubagentTaskSummary { + id: "task-1".into(), + subagent: "worker".into(), + status: SubagentTaskStatus::Suspended, + request: "Inspect the deployment".into(), + resumptions: 2, + trace_id: Some("trace-1".into()), + created_at: "2026-08-28T12:00:00Z".into(), + ended_at: None, + }; + assert_eq!( + serde_json::from_value::(serde_json::to_value(&task).unwrap()) + .unwrap(), + task + ); + + let roster = AgentGroupRoster { + members: vec![AgentGroupMemberStatus { + agent_id: "agent-1".into(), + name: "Worker".into(), + role: AgentGroupMemberRole::Spoke, + status: AgentGroupMemberState::Running, + running_turns: 1, + pending_asks: 0, + waiting_on_peer: 0, + background_tasks: 2, + tokens_in: 10, + tokens_out: 20, + cost_usd: 0.42, + last_activity_at: Some("2026-08-28T12:00:00Z".into()), + }], + totals: AgentGroupRosterTotals { + members: 1, + hubs: 0, + spokes: 1, + running: 1, + waiting: 0, + ended: 0, + background_tasks: 2, + cost_usd: 0.42, + }, + }; + assert_eq!( + serde_json::from_value::(serde_json::to_value(&roster).unwrap()) + .unwrap(), + roster + ); + + for response in [ + EndAgentGroupResponse { + threads_ended: 2, + tasks_cancelled: 1, + }, + EndAgentGroupResponse { + threads_ended: 0, + tasks_cancelled: 0, + }, + ] { + assert_eq!( + serde_json::from_value::( + serde_json::to_value(&response).unwrap() + ) + .unwrap(), + response + ); + } + let response = CancelSubagentTaskResponse { cancelled: true }; + assert_eq!( + serde_json::from_value::( + serde_json::to_value(&response).unwrap() + ) + .unwrap(), + response + ); + } + + #[test] + fn missing_swarm_observability_fields_default() { + let summary: ConversationSummary = serde_json::from_value(serde_json::json!({ + "phone": "tg:123", + "contactName": null, + "agentName": "Orchid", + "agentId": "agent-1", + "apiKey": "secret", + "lastMessage": "Latest", + "lastMessageAt": "2026-08-28T12:00:00Z", + "lastDirection": "inbound", + "messageCount": 1, + "isActive": true, + "hasErrors": false + })) + .unwrap(); + assert_eq!(summary.liveness, None); + + let detail: ConversationDetail = serde_json::from_value(serde_json::json!({ + "phone": "tg:123", + "meta": { + "contactName": null, + "contactEmail": null, + "agentName": "Orchid", + "agentId": "agent-1", + "apiKey": "secret", + "messageCount": 0, + "firstSeen": "2026-08-28T11:00:00Z", + "lastSeen": "2026-08-28T12:00:00Z" + }, + "messages": [], + "spans": {}, + "agent": null, + "blocked": false, + "takeover": null + })) + .unwrap(); + assert_eq!(detail.liveness, None); + assert!(detail.tasks.is_empty()); + } } diff --git a/crates/ui/src/keiki.rs b/crates/ui/src/keiki.rs index 945df56e6..1ce770481 100644 --- a/crates/ui/src/keiki.rs +++ b/crates/ui/src/keiki.rs @@ -60,6 +60,7 @@ pub struct KeikiConversation { /// peer agent or a contact this desktop did not send. Reported by the /// conversation fetch, so it survives a refresh (unlike `pending`). pub remote_turn_started: Option>, + pub tasks: Vec, pub error: Option, pub steer_reply: Option, } @@ -73,6 +74,7 @@ impl KeikiConversation { pending: None, pending_started: None, remote_turn_started: None, + tasks: Vec::new(), error: None, steer_reply: None, } @@ -1690,11 +1692,16 @@ pub(crate) async fn refresh_keiki_snapshot( .iter() .cloned() .collect::>(), + state + .keiki_expanded_groups + .iter() + .cloned() + .collect::>(), )) }) .map_err(|error| request_task_error("Keiki state read", error))? .ok_or_else(|| keiki_api::Error::Local("Keiki credentials are unavailable".into()))?; - let (client, token, credentials, expanded_agents) = context; + let (client, token, credentials, expanded_agents, expanded_groups) = context; let groups = authorized( &entity, client.clone(), @@ -1705,6 +1712,28 @@ pub(crate) async fn refresh_keiki_snapshot( cx, ) .await?; + let mut rosters = Vec::new(); + for group_id in expanded_groups { + let roster = authorized( + &entity, + client.clone(), + token.clone(), + credentials.clone(), + "Keiki agent group roster", + { + let group_id = group_id.clone(); + move |client, access_token| { + let group_id = group_id.clone(); + async move { client.agent_group_roster(&access_token, &group_id).await } + } + }, + cx, + ) + .await; + if let Ok(roster) = roster { + rosters.push((group_id, roster)); + } + } let mut conversations = authorized( &entity, client.clone(), @@ -1741,15 +1770,24 @@ pub(crate) async fn refresh_keiki_snapshot( } let spaces = agents.iter().map(map_agent).collect(); let mut seen = HashSet::new(); - let chats = conversations + let mapped_conversations: Vec<_> = conversations .iter() .filter_map(map_conversation) .filter(|chat| seen.insert(chat.id.clone())) .collect(); + let liveness = conversations + .iter() + .filter_map(|conversation| { + let chat = map_conversation(conversation)?; + Some((chat.id, conversation.liveness)) + }) + .collect(); entity .update(cx, |state, cx| { - state.apply_keiki_snapshot(spaces, chats); + state.apply_keiki_snapshot(spaces, mapped_conversations); state.keiki_agent_groups = groups; + state.keiki_liveness = liveness; + state.keiki_group_rosters = rosters.into_iter().collect(); state.keiki_expanding_agents.clear(); cx.notify(); }) @@ -1818,6 +1856,74 @@ pub(crate) async fn create_agent_from_template( .await } +async fn authorized_request( + entity: &WeakEntity, + operation: &'static str, + make_request: F, + cx: &mut AsyncApp, +) -> Result +where + T: Send + 'static, + F: Fn(Client, String) -> Fut + Clone + Send + 'static, + Fut: Future> + Send + 'static, +{ + let context = entity + .update(cx, |state, _| { + Some(( + state.keiki_client.clone()?, + state.keiki_token.clone()?, + state.keiki_credentials.clone()?, + )) + }) + .map_err(|error| request_task_error(operation, error))? + .ok_or_else(|| keiki_api::Error::Local("Keiki credentials are unavailable".into()))?; + let (client, token, credentials) = context; + authorized( + entity, + client, + token, + credentials, + operation, + make_request, + cx, + ) + .await +} + +pub(crate) async fn cancel_subagent_task( + entity: &WeakEntity, + task_id: String, + cx: &mut AsyncApp, +) -> Result { + authorized_request( + entity, + "Keiki task cancellation", + move |client, access_token| { + let task_id = task_id.clone(); + async move { client.cancel_subagent_task(&access_token, &task_id).await } + }, + cx, + ) + .await +} + +pub(crate) async fn end_agent_group( + entity: &WeakEntity, + group_id: String, + cx: &mut AsyncApp, +) -> Result { + authorized_request( + entity, + "Keiki group end", + move |client, access_token| { + let group_id = group_id.clone(); + async move { client.end_agent_group(&access_token, &group_id).await } + }, + cx, + ) + .await +} + async fn poll(entity: gpui::WeakEntity, cx: &mut gpui::AsyncApp) { loop { let context = match entity.update(cx, |state, _| { @@ -2402,6 +2508,8 @@ mod tests { blocked: false, takeover: None, active_turn_started_at: None, + liveness: None, + tasks: Vec::new(), peer: None, }; @@ -2478,6 +2586,8 @@ mod tests { blocked: false, takeover: None, active_turn_started_at: None, + liveness: None, + tasks: Vec::new(), peer: None, }; diff --git a/crates/ui/src/shell.rs b/crates/ui/src/shell.rs index 0be17edf0..77b68fe35 100644 --- a/crates/ui/src/shell.rs +++ b/crates/ui/src/shell.rs @@ -431,6 +431,8 @@ pub enum RightSurface { Subagent(u64), /// The selected agent's agent-to-agent threads (one per chat). Conversations, + /// The selected Keiki conversation's background tasks. + Tasks, /// The conversation sandbox's live desktop — the handle keys /// [`Shell::desktops`]. Desktop(u64), @@ -896,6 +898,8 @@ pub struct Shell { user_menu: popover::Popup<()>, /// Inline sidebar error strip (mutation failures); click dismisses. sidebar_notice: Option, + /// Confirmation state for ending all live threads in a Keiki group. + end_group_confirm: Option<(String, u32, u32)>, /// Access token last synchronized into the device-local Copilot holder. copilot_synced_token: Option, mutate_task: Option>, @@ -1029,6 +1033,48 @@ impl Shell { self.sidebar_notice = Some(notice.into()); } + fn confirm_end_keiki_group( + &mut self, + group_id: String, + threads: u32, + tasks: u32, + cx: &mut Context, + ) { + self.end_group_confirm = Some((group_id, threads, tasks)); + cx.notify(); + } + + fn submit_end_keiki_group(&mut self, group_id: String, cx: &mut Context) { + self.end_group_confirm = None; + let state = self.state.downgrade(); + cx.spawn(async move |this, cx| { + let result = crate::keiki::end_agent_group(&state, group_id, cx).await; + match result { + Ok(response) => { + let _ = crate::keiki::refresh_keiki_snapshot(state, cx).await; + let _ = this.update(cx, |shell, cx| { + shell.sidebar_notice = Some( + format!( + "Ended {} threads and cancelled {} tasks", + response.threads_ended, response.tasks_cancelled + ) + .into(), + ); + cx.notify(); + }); + } + Err(error) => { + crate::notify::post("Keiki group end failed", &error.to_string()); + let _ = this.update(cx, |shell, cx| { + shell.sidebar_notice = Some(error.to_string().into()); + cx.notify(); + }); + } + } + }) + .detach(); + } + /// "Check for Updates…": ignored while a download or install is already on /// screen — replacing the task would cancel it mid-swap. fn check_for_updates(&mut self, cx: &mut Context) { @@ -1192,6 +1238,7 @@ impl Shell { sound_prev: std::collections::HashMap::new(), user_menu: popover::Popup::default(), sidebar_notice: None, + end_group_confirm: None, copilot_synced_token: None, mutate_task: None, boot, @@ -1770,6 +1817,7 @@ impl Shell { RightSurface::Conversations => { Some((*surface, SharedString::from("Conversations"))) } + RightSurface::Tasks => Some((*surface, SharedString::from("Tasks"))), RightSurface::Desktop(id) => self .desktops .contains_key(id) @@ -1856,6 +1904,7 @@ impl Shell { // activation needs no revalidation. RightSurface::Subagent(_) => {} RightSurface::Conversations + | RightSurface::Tasks | RightSurface::Desktop(_) | RightSurface::Browser | RightSurface::Picker => {} @@ -1895,6 +1944,199 @@ impl Shell { self.set_right_active(RightSurface::Conversations, cx); } + fn tasks_offered(&self, cx: &App) -> bool { + let state = self.state.read(cx); + crate::keiki::is_keiki_chat(&self.active_chat) + && state + .keiki_conversation + .as_ref() + .is_some_and(|conversation| { + conversation.chat_id == self.active_chat + && (!conversation.tasks.is_empty() + || state + .keiki_liveness + .get(&self.active_chat) + .and_then(|liveness| *liveness) + .is_some()) + }) + } + + fn ensure_tasks_surface(&mut self, cx: &mut Context) { + let key = self.panel_key(cx); + let offered = self.tasks_offered(cx); + let tabs = self.right_tabs.entry(key).or_default(); + if offered { + if !tabs.contains(&RightSurface::Tasks) { + tabs.push(RightSurface::Tasks); + } + } else { + tabs.retain(|surface| *surface != RightSurface::Tasks); + } + } + + fn add_tasks_surface(&mut self, cx: &mut Context) { + if self.tasks_offered(cx) { + self.ensure_tasks_surface(cx); + self.set_right_active(RightSurface::Tasks, cx); + } + } + + fn cancel_keiki_task(&mut self, task_id: String, cx: &mut Context) { + let chat_id = self.active_chat.clone(); + let state = self.state.downgrade(); + cx.spawn(async move |this, cx| { + let result = crate::keiki::cancel_subagent_task(&state, task_id, cx).await; + match result { + Ok(_) => { + let _ = state.update(cx, |state, cx| { + state.transcript_task = + Some(crate::keiki::spawn_transcript_watch(cx, chat_id)); + cx.notify(); + }); + } + Err(error) => { + crate::notify::post("Keiki task cancellation failed", &error.to_string()); + let _ = this.update(cx, |shell, cx| { + shell.sidebar_notice = Some(error.to_string().into()); + cx.notify(); + }); + } + } + }) + .detach(); + } + + fn render_tasks_surface(&mut self, cx: &mut Context) -> AnyElement { + let theme = Theme::of(cx).clone(); + let (tasks, can_manage) = { + let state = self.state.read(cx); + ( + state + .keiki_conversation + .as_ref() + .filter(|conversation| conversation.chat_id == self.active_chat) + .map(|conversation| conversation.tasks.clone()) + .unwrap_or_default(), + state + .keiki_session + .as_ref() + .is_some_and(KeikiSessionInfo::can_manage), + ) + }; + let now = Utc::now(); + let time_ago = |value: &str| { + chrono::DateTime::parse_from_rfc3339(value) + .ok() + .map(|value| format_time_ago(value.with_timezone(&Utc), now)) + .unwrap_or_else(|| "—".into()) + }; + div() + .size_full() + .flex() + .flex_col() + .gap(px(8.0)) + .p(px(14.0)) + .child( + div() + .text_size(crate::typography::ui_rems(14.0)) + .font_weight(gpui::FontWeight::SEMIBOLD) + .text_color(theme.text) + .child("Background tasks"), + ) + .children(tasks.into_iter().map(|task| { + let (status, status_color, cancellable) = match task.status { + keiki_model::SubagentTaskStatus::Running => ("Running", theme.busy, true), + keiki_model::SubagentTaskStatus::Suspended => { + ("Suspended", theme.warning, true) + } + keiki_model::SubagentTaskStatus::Completed => { + ("Completed", theme.text_muted, false) + } + keiki_model::SubagentTaskStatus::Error => ("Error", theme.danger, false), + keiki_model::SubagentTaskStatus::Cancelled => { + ("Cancelled", theme.text_faint, false) + } + }; + let task_id = task.id.clone(); + let request = crate::transcript::single_line(&task.request); + let request = if request.chars().count() > 100 { + format!("{}…", request.chars().take(100).collect::()) + } else { + request + }; + div() + .w_full() + .p(px(10.0)) + .rounded(px(8.0)) + .border_1() + .border_color(theme.border) + .flex() + .flex_col() + .gap(px(5.0)) + .child( + div() + .flex() + .items_center() + .gap(px(6.0)) + .child( + div() + .text_size(crate::typography::ui_rems(12.0)) + .font_weight(gpui::FontWeight::MEDIUM) + .text_color(theme.text) + .child(task.subagent), + ) + .child( + div() + .px(px(6.0)) + .py(px(2.0)) + .rounded(px(5.0)) + .bg(status_color.opacity(0.14)) + .text_size(crate::typography::ui_rems(10.0)) + .text_color(status_color) + .child(status), + ) + .when(can_manage && cancellable, |el| { + el.child( + div() + .id(SharedString::from(format!("cancel-task-{}", task.id))) + .ml_auto() + .px(px(7.0)) + .py(px(3.0)) + .rounded(px(5.0)) + .bg(theme.element_hover) + .text_size(crate::typography::ui_rems(10.0)) + .text_color(theme.text_muted) + .cursor_pointer() + .on_click(cx.listener(move |this, _, _, cx| { + this.cancel_keiki_task(task_id.clone(), cx); + })) + .child("Cancel"), + ) + }), + ) + .child( + div() + .text_size(crate::typography::ui_rems(11.0)) + .text_color(theme.text_muted) + .child(request), + ) + .child( + div() + .text_size(crate::typography::ui_rems(10.0)) + .text_color(theme.text_faint) + .child(SharedString::from(format!( + "Created {}{}", + time_ago(&task.created_at), + task.ended_at + .as_deref() + .map(|ended| format!(" · ended {}", time_ago(ended))) + .unwrap_or_default() + ))), + ) + })) + .into_any_element() + } + /// The picker's Git card / the `+` menu's Diff row: every click opens a /// FRESH diff tab with its own scope/base selection (multiple diff /// panels, user request). @@ -2119,6 +2361,7 @@ impl Shell { self.browsers.remove(&self.active_chat); } RightSurface::Conversations | RightSurface::Picker => {} + RightSurface::Tasks => {} } self.panels.update(&key, |p| { if p.right_active == surface { @@ -4854,6 +5097,40 @@ impl Shell { overlays.push(popover::modal("delete-chat-dialog", viewport, card)); } + if let Some((group_id, threads, tasks)) = self.end_group_confirm.clone() { + let card = popover::dialog_card(&theme) + .child(popover::dialog_title(&theme, "End all group work?")) + .child(div().mt(px(6.0)).child(popover::dialog_body( + &theme, + format!("Ends {threads} live threads and cancels {tasks} tasks."), + ))) + .child( + div() + .mt(px(16.0)) + .flex() + .flex_row() + .justify_end() + .gap(px(8.0)) + .child( + popover::btn_ghost(&theme, "Cancel", "end-group-cancel") + .id("end-group-cancel") + .on_click(cx.listener(|this, _, _, cx| { + this.end_group_confirm = None; + cx.notify(); + })), + ) + .child( + popover::btn_danger(&theme, "End all") + .id("end-group-confirm") + .on_click(cx.listener(move |this, _, _, cx| { + this.submit_end_keiki_group(group_id.clone(), cx); + })), + ), + ) + .into_any_element(); + overlays.push(popover::modal("end-group-dialog", viewport, card)); + } + overlays } @@ -5447,6 +5724,7 @@ impl Shell { .into_any_element() } RightSurface::Conversations => self.render_peer_threads_surface(cx), + RightSurface::Tasks => self.render_tasks_surface(cx), RightSurface::Desktop(id) if self.desktops.contains_key(&id) => self .desktops .get(&id) @@ -5510,6 +5788,7 @@ impl Shell { .read(cx) .selected_space_row() .is_some_and(|space| crate::keiki::is_keiki_space(&space.id)); + let tasks_offered = self.tasks_offered(cx); let row = |id: &'static str, icon_path: &'static str, title: &'static str| { div() .id(id) @@ -5588,6 +5867,15 @@ impl Shell { })), ) }) + .when(tasks_offered, |el| { + el.child( + row("surface-card-tasks", icons::CHAT_ROUND_LINE, "Tasks").on_click( + cx.listener(|this, _, _, cx| { + this.add_tasks_surface(cx); + }), + ), + ) + }) // Git only where there IS git — the pane itself no // longer gates on it (terminals work anywhere). .when(self.space_git_detected(cx), |el| { @@ -5618,6 +5906,7 @@ impl Shell { const CHIP_SLOT: f32 = CHIP_W + 4.0; // + the strip's own gap let theme = Theme::of(cx).clone(); + self.ensure_tasks_surface(cx); // Heal drag state if the pointer was released outside the strip. if self.right_tab_drag.is_some() && !cx.has_active_drag() { self.right_tab_drag = None; @@ -5686,6 +5975,7 @@ impl Shell { RightSurface::Diff(_) => icons::GIT_BRANCH, RightSurface::Subagent(_) => icons::BOT, RightSurface::Conversations => icons::CHAT_ROUND_LINE, + RightSurface::Tasks => icons::CHAT_ROUND_LINE, RightSurface::Desktop(_) => icons::LAPTOP, RightSurface::Browser => icons::MONITOR, _ => icons::TERMINAL, diff --git a/crates/ui/src/shell/spaces.rs b/crates/ui/src/shell/spaces.rs index a8edf9f3b..6cfb4d19b 100644 --- a/crates/ui/src/shell/spaces.rs +++ b/crates/ui/src/shell/spaces.rs @@ -645,6 +645,28 @@ impl Shell { .all(|key| self.settings.sidebar_collapsed_groups.contains(key)) } + fn sync_keiki_expanded_groups(&mut self, cx: &mut Context) { + let expanded = if self.settings.sidebar_organization == SidebarOrganization::ByAgent { + let collapsed = &self.settings.sidebar_collapsed_groups; + self.state + .read(cx) + .keiki_agent_groups + .iter() + .map(|group| group.id.clone()) + .filter(|group_id| { + !collapsed.contains(&format!("agent:{}", crate::keiki::group_id(group_id))) + }) + .collect() + } else { + std::collections::HashSet::new() + }; + if self.state.read(cx).keiki_expanded_groups != expanded { + self.state.update(cx, |state, _| { + state.keiki_expanded_groups = expanded; + }); + } + } + /// Fold every visible group, or unfold them all when already folded. Runs /// no disclosure tween: a whole-list snap reads cleaner than a dozen /// staggered accordions. @@ -1360,6 +1382,7 @@ impl Shell { let local_device_id = self.state.read(cx).local_device_id.clone(); promote_local_device_group(&mut groups, local_device_id.as_deref()); } + self.sync_keiki_expanded_groups(cx); let selected = self.state.read(cx).selected_chat.clone(); // Re-checked at render so the chips drop the FRAME a popover opens, @@ -1657,16 +1680,127 @@ impl Shell { } else { 0.0 }; + let roster = key + .strip_prefix(crate::keiki::GROUP_PREFIX) + .and_then(|group_id| { + self.state + .read(cx) + .keiki_group_rosters + .get(group_id) + .cloned() + }); + let roster_summary = roster.as_ref().map(|roster| { + format!( + "{} members · {} running · {} waiting · ${:.2}", + roster.totals.members, + roster.totals.running, + roster.totals.waiting, + roster.totals.cost_usd + ) + }); + let roster_members = roster.as_ref().map(|roster| { + let mut members = roster.members.iter().collect::>(); + members.sort_by(|left, right| { + let status_rank = |status| match status { + keiki_model::AgentGroupMemberState::Running => 0, + keiki_model::AgentGroupMemberState::Waiting => 1, + keiki_model::AgentGroupMemberState::Idle => 2, + keiki_model::AgentGroupMemberState::Ended => 3, + }; + status_rank(left.status) + .cmp(&status_rank(right.status)) + .then_with(|| left.name.cmp(&right.name)) + }); + let shown = members.into_iter().take(25).collect::>(); + let hidden = roster.members.len().saturating_sub(shown.len()); + (shown, hidden) + }); + let roster_row_count = roster_members + .as_ref() + .map(|(members, hidden)| members.len() + if *hidden > 0 { 1 } else { 0 }) + .unwrap_or(0); + let roster_rows = if !collapsed { + roster_members.as_ref().map(|(members, hidden)| { + div() + .w_full() + .px(px(Theme::SPACE_SM)) + .flex() + .flex_col() + .gap(px(3.0)) + .children(members.iter().map(|member| { + let (status, color) = match member.status { + keiki_model::AgentGroupMemberState::Running => { + ("running", theme.busy) + } + keiki_model::AgentGroupMemberState::Waiting => { + ("waiting", theme.warning) + } + keiki_model::AgentGroupMemberState::Idle => { + ("idle", theme.text_faint) + } + keiki_model::AgentGroupMemberState::Ended => { + ("ended", theme.danger) + } + }; + let role = match member.role { + keiki_model::AgentGroupMemberRole::Hub => "hub", + keiki_model::AgentGroupMemberRole::Spoke => "spoke", + keiki_model::AgentGroupMemberRole::Peer => "peer", + }; + div() + .flex() + .items_center() + .gap(px(5.0)) + .text_size(crate::typography::ui_rems(10.0)) + .text_color(theme.text_muted) + .child(div().size(px(5.0)).rounded(px(3.0)).bg(color)) + .child(div().flex_1().min_w_0().truncate().child( + SharedString::from(format!( + "{} · {} · {}", + member.name, role, status + )), + )) + .child(SharedString::from(format!( + "{} in / {} out · ${:.2}", + member.tokens_in, member.tokens_out, member.cost_usd + ))) + })) + .when(*hidden > 0, |element| { + element.child( + div() + .text_size(crate::typography::ui_rems(10.0)) + .text_color(theme.text_faint) + .child(SharedString::from(format!("+{hidden} more"))), + ) + }) + }) + } else { + None + }; + let body_height = body_height + + if !collapsed { + (roster_row_count > 0) + .then(|| { + 4.0 + roster_row_count as f32 * 18.0 + + SIDEBAR_LIST_GAP * roster_row_count.saturating_sub(1) as f32 + }) + .unwrap_or(0.0) + } else { + 0.0 + }; let body = div() .w_full() .flex() .flex_col() .pt(px(SIDEBAR_DISCLOSURE_BODY_INSET)) .gap(px(SIDEBAR_LIST_GAP)) + .children(roster_rows) .children(rendered_rows.into_iter().map(|(_, _, row)| row)) .children(history_toggle); let visible_label: SharedString = if collapsed { format!("{label} ({row_count})").into() + } else if let Some(summary) = roster_summary { + format!("{label} · {summary}").into() } else { label.into() }; @@ -1723,31 +1857,75 @@ impl Shell { .then(|| key.strip_prefix(crate::keiki::AGENT_PREFIX)) .flatten() .map(|agent_id| self.render_new_conversation_button(agent_id, theme, cx)); - let header = sidebar_disclosure_header( - theme, - visible_label, - chevron, - group_avatar, - new_conversation, - ) - .id(SharedString::from(format!("sidebar-group-{collapse_key}"))) - .on_click(cx.listener(move |this, _, _, cx| { - let was_open = !this.settings.sidebar_collapsed_groups.contains(&toggle_key); - this.begin_sidebar_disclosure_motion( - &toggle_motion_key, - if was_open { body_height } else { 0.0 }, - if was_open { 0.0 } else { body_height }, - ); - if was_open { - this.settings - .sidebar_collapsed_groups - .insert(toggle_key.clone()); - } else { - this.settings.sidebar_collapsed_groups.remove(&toggle_key); - } - this.schedule_save(cx); - cx.notify(); - })); + let end_all = roster + .as_ref() + .and_then(|roster| { + let can_manage = self + .state + .read(cx) + .keiki_session + .as_ref() + .is_some_and(KeikiSessionInfo::can_manage); + let live = roster.totals.running + roster.totals.waiting; + (can_manage && live > 0).then(|| { + let group_id = key.strip_prefix(crate::keiki::GROUP_PREFIX)?.to_string(); + let tasks = roster.totals.background_tasks; + Some( + div() + .id(SharedString::from(format!("keiki-end-all-{group_id}"))) + .size(px(20.0)) + .flex() + .items_center() + .justify_center() + .rounded(px(5.0)) + .cursor_pointer() + .hover(|s| s.bg(theme.element_hover)) + .on_click(cx.listener(move |this, _, _, cx| { + cx.stop_propagation(); + this.confirm_end_keiki_group(group_id.clone(), live, tasks, cx); + })) + .child(icon(icons::STOP).size(px(12.0)).text_color(theme.danger)) + .into_any_element(), + ) + }) + }) + .flatten(); + let trailing = match (new_conversation, end_all) { + (Some(new_conversation), Some(end_all)) => Some( + div() + .flex() + .items_center() + .gap(px(2.0)) + .child(end_all) + .child(new_conversation) + .into_any_element(), + ), + (Some(new_conversation), None) => Some(new_conversation), + (None, Some(end_all)) => Some(end_all), + (None, None) => None, + }; + let header = + sidebar_disclosure_header(theme, visible_label, chevron, group_avatar, trailing) + .id(SharedString::from(format!("sidebar-group-{collapse_key}"))) + .on_click(cx.listener(move |this, _, _, cx| { + let was_open = + !this.settings.sidebar_collapsed_groups.contains(&toggle_key); + this.begin_sidebar_disclosure_motion( + &toggle_motion_key, + if was_open { body_height } else { 0.0 }, + if was_open { 0.0 } else { body_height }, + ); + if was_open { + this.settings + .sidebar_collapsed_groups + .insert(toggle_key.clone()); + } else { + this.settings.sidebar_collapsed_groups.remove(&toggle_key); + } + this.sync_keiki_expanded_groups(cx); + this.schedule_save(cx); + cx.notify(); + })); let body = self.render_sidebar_disclosure_body( &motion_key, !collapsed, diff --git a/crates/ui/src/shell/tabs.rs b/crates/ui/src/shell/tabs.rs index e65372dc6..066774da4 100644 --- a/crates/ui/src/shell/tabs.rs +++ b/crates/ui/src/shell/tabs.rs @@ -220,6 +220,14 @@ impl Shell { None => (SharedString::from(""), None, None, true, None), } }; + let liveness = self + .state + .read(cx) + .selected_chat + .as_deref() + .and_then(|chat_id| self.state.read(cx).keiki_liveness.get(chat_id)) + .copied() + .flatten(); let (source_orb, target_orb) = peer .map(|(peer, target_state)| { ( @@ -534,6 +542,27 @@ impl Shell { .child(target), ) }) + .when_some(liveness, |el, liveness| { + let (label, color) = match liveness { + keiki_model::ConversationLiveness::Working => { + ("Working", theme.busy) + } + keiki_model::ConversationLiveness::Waiting => { + ("Waiting on peer", theme.text_muted) + } + }; + el.child( + div() + .flex_none() + .flex() + .items_center() + .gap(px(4.0)) + .text_size(crate::typography::ui_rems(11.0)) + .text_color(color) + .child(div().size(px(5.0)).rounded(px(3.0)).bg(color)) + .child(SharedString::from(label)), + ) + }) .when_some(conversation_status, |el, conversation| { let takeover = conversation.takeover.as_ref().and_then(|takeover| { crate::keiki::parse_timestamp(&takeover.expires_at).map(|expires| { diff --git a/crates/ui/src/state.rs b/crates/ui/src/state.rs index 84c260de4..5b599da2a 100644 --- a/crates/ui/src/state.rs +++ b/crates/ui/src/state.rs @@ -58,6 +58,10 @@ impl KeikiSessionInfo { pub fn switchable_orgs(&self) -> &[keiki_api::OrganizationSummary] { if self.orgs.len() > 1 { &self.orgs } else { &[] } } + + pub fn can_manage(&self) -> bool { + matches!(self.role.as_deref(), Some("owner" | "admin")) + } } // --------------------------------------------------------------------------- @@ -387,6 +391,7 @@ pub struct AppState { pub(crate) keiki_error: Option, pub(crate) keiki_task: Option>, pub(crate) keiki_conversation: Option, + pub(crate) keiki_liveness: HashMap>, /// The in-flight steered turn's task on a keiki conversation. Held so a /// Stop can cancel it: dropping the task drops the response stream, which /// is the signal the platform reads as "stop this turn". Session-only. @@ -397,6 +402,8 @@ pub struct AppState { /// The org's agent groups; an inter-agent thread whose two agents share /// one is listed under the group in the sidebar. Session-only. pub(crate) keiki_agent_groups: Vec, + pub(crate) keiki_expanded_groups: HashSet, + pub(crate) keiki_group_rosters: HashMap, /// Peer threads this desktop ended (blocked) from the Conversations /// surface — only the selected conversation carries a fetched `blocked`. pub(crate) keiki_ended_threads: HashSet, @@ -459,9 +466,12 @@ impl AppState { keiki_error: None, keiki_task: None, keiki_conversation: None, + keiki_liveness: HashMap::new(), keiki_steer_task: None, keiki_expanded_agents: HashSet::new(), keiki_agent_groups: Vec::new(), + keiki_expanded_groups: HashSet::new(), + keiki_group_rosters: HashMap::new(), keiki_ended_threads: HashSet::new(), keiki_expanding_agents: HashSet::new(), keiki_draft_chats: HashSet::new(), @@ -701,6 +711,9 @@ impl AppState { self.chats .retain(|chat| !crate::keiki::is_keiki_chat(&chat.id)); self.keiki_draft_chats.clear(); + self.keiki_liveness.clear(); + self.keiki_expanded_groups.clear(); + self.keiki_group_rosters.clear(); self.keiki_conversation = None; if selected_keiki_chat { self.selected_chat = None; @@ -1239,6 +1252,8 @@ impl AppState { if self.selected_chat.as_deref() != Some(chat_id) { return; } + self.keiki_liveness + .insert(chat_id.to_string(), detail.liveness); let conversation = self .keiki_conversation .get_or_insert_with(|| KeikiConversation::new(chat_id.to_string())); @@ -1247,6 +1262,7 @@ impl AppState { } conversation.blocked = detail.blocked; conversation.takeover = detail.takeover.clone(); + conversation.tasks = detail.tasks.clone(); conversation.remote_turn_started = crate::keiki::remote_turn_started(detail); conversation.pending = None; conversation.pending_started = None; @@ -1269,12 +1285,15 @@ impl AppState { chat_id: &str, detail: &keiki_model::ConversationDetail, ) { + self.keiki_liveness + .insert(chat_id.to_string(), detail.liveness); if let Some(conversation) = self .keiki_conversation .as_mut() .filter(|conversation| conversation.chat_id == chat_id) { conversation.remote_turn_started = crate::keiki::remote_turn_started(detail); + conversation.tasks = detail.tasks.clone(); } }