Skip to repository content1432 lines · 51.4 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T01:49:51.035Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
thread_import.rs
1use std::time::Duration;
2
3use acp_thread::AgentSessionListRequest;
4use agent::ThreadStore;
5use agent_client_protocol::schema::v1 as acp;
6use chrono::Utc;
7use collections::{HashMap, HashSet};
8use db::kvp::Dismissable;
9use db::sqlez;
10use fs::Fs;
11use futures::FutureExt as _;
12use gpui::{
13 Animation, AnimationExt as _, App, Context, DismissEvent, Entity, EventEmitter, FocusHandle,
14 Focusable, MouseDownEvent, Render, SharedString, Task, TaskExt, WeakEntity, Window,
15 pulsating_between,
16};
17use itertools::Itertools as _;
18use notifications::status_toast::StatusToast;
19use project::{AgentId, AgentRegistryStore, AgentServerStore};
20use release_channel::ReleaseChannel;
21use remote::RemoteConnectionOptions;
22use ui::{
23 Checkbox, CommonAnimationExt, KeyBinding, ListItem, ListItemSpacing, Modal, ModalFooter,
24 ModalHeader, Section, Tooltip, prelude::*,
25};
26use util::ResultExt;
27use workspace::{ModalView, MultiWorkspace, Workspace};
28
29use crate::{
30 Agent, AgentPanel,
31 agent_connection_store::AgentConnectionStore,
32 thread_metadata_store::{ThreadId, ThreadMetadata, ThreadMetadataStore, WorktreePaths},
33};
34
35pub struct AcpThreadImportOnboarding;
36pub struct CrossChannelImportOnboarding;
37
38impl AcpThreadImportOnboarding {
39 pub fn dismissed(cx: &App) -> bool {
40 <Self as Dismissable>::dismissed(cx)
41 }
42
43 pub fn dismiss(cx: &mut App) {
44 <Self as Dismissable>::set_dismissed(true, cx);
45 }
46}
47
48impl Dismissable for AcpThreadImportOnboarding {
49 const KEY: &'static str = "dismissed-acp-thread-import";
50}
51
52impl CrossChannelImportOnboarding {
53 pub fn dismissed(cx: &App) -> bool {
54 <Self as Dismissable>::dismissed(cx)
55 }
56
57 pub fn dismiss(cx: &mut App) {
58 <Self as Dismissable>::set_dismissed(true, cx);
59 }
60}
61
62impl Dismissable for CrossChannelImportOnboarding {
63 const KEY: &'static str = "dismissed-cross-channel-thread-import";
64}
65
66pub fn channels_with_threads(cx: &App) -> Task<Vec<SharedString>> {
67 let Some(current_channel) = ReleaseChannel::try_global(cx) else {
68 return Task::ready(Vec::new());
69 };
70 let database_dir = paths::database_dir();
71
72 let channel_has_threads = |database_dir: &std::path::Path, channel: ReleaseChannel| {
73 let db_path = db::db_path(database_dir, channel);
74 if !db_path.exists() {
75 return false;
76 }
77 let connection = sqlez::connection::Connection::open_file(&db_path.to_string_lossy());
78 connection
79 .select_row::<bool>("SELECT 1 FROM sidebar_threads LIMIT 1")
80 .ok()
81 .and_then(|mut query| query().ok().flatten())
82 .unwrap_or(false)
83 };
84
85 cx.background_spawn(async move {
86 ReleaseChannel::ALL
87 .iter()
88 .copied()
89 .filter(|channel| {
90 *channel != current_channel
91 && *channel != ReleaseChannel::Dev
92 && channel_has_threads(database_dir, *channel)
93 })
94 .map(|channel| SharedString::new_static(channel.display_name()))
95 .collect()
96 })
97}
98
99#[derive(Clone)]
100struct AgentEntry {
101 agent_id: AgentId,
102 display_name: SharedString,
103 icon_path: Option<SharedString>,
104}
105
106#[derive(Clone)]
107enum AgentImportStatus {
108 Loading,
109 Ready { importable_count: usize },
110 Unsupported,
111 Error(SharedString),
112}
113
114impl AgentImportStatus {
115 fn is_selectable(&self) -> bool {
116 matches!(self, Self::Ready { importable_count } if *importable_count > 0)
117 }
118
119 fn tooltip_text(&self) -> Option<SharedString> {
120 match self {
121 Self::Loading => Some("Fetching Sessions…".into()),
122 Self::Ready { .. } => None,
123 Self::Unsupported => Some("Importing threads from this agent is not possible as it doesn't support ACP's session/list capability.".into()),
124 Self::Error(error) => Some(format!("Failed to fetch sessions: {error}").into()),
125 }
126 }
127}
128
129pub struct ThreadImportModal {
130 focus_handle: FocusHandle,
131 workspace: WeakEntity<Workspace>,
132 multi_workspace: WeakEntity<MultiWorkspace>,
133 agent_entries: Vec<AgentEntry>,
134 unchecked_agents: HashSet<AgentId>,
135 agent_import_statuses: HashMap<AgentId, AgentImportStatus>,
136 sessions_by_agent: Vec<SessionByAgent>,
137 selected_index: Option<usize>,
138 is_fetching_sessions: bool,
139 is_importing: bool,
140 last_error: Option<SharedString>,
141}
142
143impl ThreadImportModal {
144 pub fn new(
145 agent_server_store: Entity<AgentServerStore>,
146 agent_registry_store: Entity<AgentRegistryStore>,
147 workspace: WeakEntity<Workspace>,
148 multi_workspace: WeakEntity<MultiWorkspace>,
149 _window: &mut Window,
150 cx: &mut Context<Self>,
151 ) -> Self {
152 AcpThreadImportOnboarding::dismiss(cx);
153
154 let agent_entries = agent_server_store
155 .read(cx)
156 .external_agents()
157 .map(|agent_id| {
158 let display_name = agent_server_store
159 .read(cx)
160 .agent_display_name(agent_id)
161 .or_else(|| {
162 agent_registry_store
163 .read(cx)
164 .agent(agent_id)
165 .map(|agent| agent.name().clone())
166 })
167 .unwrap_or_else(|| agent_id.0.clone());
168 let icon_path = agent_server_store
169 .read(cx)
170 .agent_icon(agent_id)
171 .or_else(|| {
172 agent_registry_store
173 .read(cx)
174 .agent(agent_id)
175 .and_then(|agent| agent.icon_path().cloned())
176 });
177
178 AgentEntry {
179 agent_id: agent_id.clone(),
180 display_name,
181 icon_path,
182 }
183 })
184 .sorted_unstable_by_key(|entry| entry.display_name.to_lowercase())
185 .collect::<Vec<_>>();
186
187 let this = Self {
188 focus_handle: cx.focus_handle(),
189 workspace,
190 multi_workspace,
191 agent_entries,
192 unchecked_agents: HashSet::default(),
193 agent_import_statuses: HashMap::default(),
194 sessions_by_agent: Vec::new(),
195 selected_index: None,
196 is_fetching_sessions: false,
197 is_importing: false,
198 last_error: None,
199 };
200 cx.spawn(async move |this, cx| this.update(cx, |this, cx| this.fetch_sessions(cx)))
201 .detach_and_log_err(cx);
202 this
203 }
204
205 fn agent_ids(&self) -> Vec<AgentId> {
206 self.agent_entries
207 .iter()
208 .map(|entry| entry.agent_id.clone())
209 .collect()
210 }
211
212 fn fetch_sessions(&mut self, cx: &mut Context<Self>) {
213 if self.agent_entries.is_empty() {
214 return;
215 }
216
217 let Some(multi_workspace) = self.multi_workspace.upgrade() else {
218 self.mark_all_agents_failed("Could not find workspace to import from.");
219 return;
220 };
221
222 let stores = resolve_agent_connection_stores(&multi_workspace, cx);
223 if stores.is_empty() {
224 log::error!("Did not find any workspaces to import from");
225 self.mark_all_agents_failed("Did not find any workspaces to import from.");
226 return;
227 }
228
229 self.is_fetching_sessions = true;
230 self.last_error = None;
231 self.sessions_by_agent.clear();
232 self.agent_import_statuses = self
233 .agent_ids()
234 .into_iter()
235 .map(|agent_id| (agent_id, AgentImportStatus::Loading))
236 .collect();
237
238 let existing_sessions: HashSet<acp::SessionId> = ThreadMetadataStore::global(cx)
239 .read(cx)
240 .entries()
241 .filter_map(|metadata| metadata.session_id.clone())
242 .collect();
243
244 for agent_id in self.agent_ids() {
245 let task =
246 fetch_sessions_for_agent(agent_id, existing_sessions.clone(), stores.clone(), cx);
247 cx.spawn(async move |this, cx| {
248 let result = task.await;
249 this.update(cx, |this, cx| {
250 let AgentSessionFetchResult {
251 agent_id,
252 sessions_by_agent,
253 status,
254 } = result;
255 this.sessions_by_agent
256 .retain(|sessions| sessions.agent_id != agent_id);
257 this.sessions_by_agent.extend(sessions_by_agent);
258 this.agent_import_statuses.insert(agent_id, status);
259 this.is_fetching_sessions = this.has_loading_agents();
260 cx.notify();
261 })
262 })
263 .detach_and_log_err(cx);
264 }
265 }
266
267 fn mark_all_agents_failed(&mut self, message: impl Into<SharedString>) {
268 let message = message.into();
269 self.is_fetching_sessions = false;
270 self.sessions_by_agent.clear();
271 self.last_error = Some(message.clone());
272 self.agent_import_statuses = self
273 .agent_ids()
274 .into_iter()
275 .map(|agent_id| (agent_id, AgentImportStatus::Error(message.clone())))
276 .collect();
277 }
278
279 fn agent_is_selectable(&self, agent_id: &AgentId) -> bool {
280 self.agent_import_statuses
281 .get(agent_id)
282 .map_or(false, AgentImportStatus::is_selectable)
283 }
284
285 fn has_checked_selectable_agent(&self) -> bool {
286 self.agent_entries.iter().any(|entry| {
287 self.agent_is_selectable(&entry.agent_id)
288 && !self.unchecked_agents.contains(&entry.agent_id)
289 })
290 }
291
292 fn has_loading_agents(&self) -> bool {
293 self.agent_import_statuses
294 .values()
295 .any(|status| matches!(status, AgentImportStatus::Loading))
296 }
297
298 fn toggle_agent_checked(&mut self, agent_id: AgentId, cx: &mut Context<Self>) {
299 if self.is_importing || !self.agent_is_selectable(&agent_id) {
300 return;
301 }
302
303 if self.unchecked_agents.contains(&agent_id) {
304 self.unchecked_agents.remove(&agent_id);
305 } else {
306 self.unchecked_agents.insert(agent_id);
307 }
308 cx.notify();
309 }
310
311 fn select_next(&mut self, _: &menu::SelectNext, _window: &mut Window, cx: &mut Context<Self>) {
312 if self.agent_entries.is_empty() {
313 return;
314 }
315 self.selected_index = Some(match self.selected_index {
316 Some(ix) if ix + 1 >= self.agent_entries.len() => 0,
317 Some(ix) => ix + 1,
318 None => 0,
319 });
320 cx.notify();
321 }
322
323 fn select_previous(
324 &mut self,
325 _: &menu::SelectPrevious,
326 _window: &mut Window,
327 cx: &mut Context<Self>,
328 ) {
329 if self.agent_entries.is_empty() {
330 return;
331 }
332 self.selected_index = Some(match self.selected_index {
333 Some(0) => self.agent_entries.len() - 1,
334 Some(ix) => ix - 1,
335 None => self.agent_entries.len() - 1,
336 });
337 cx.notify();
338 }
339
340 fn confirm(&mut self, _: &menu::Confirm, _window: &mut Window, cx: &mut Context<Self>) {
341 if let Some(ix) = self.selected_index {
342 if let Some(entry) = self.agent_entries.get(ix) {
343 self.toggle_agent_checked(entry.agent_id.clone(), cx);
344 }
345 }
346 }
347
348 fn cancel(&mut self, _: &menu::Cancel, _: &mut Window, cx: &mut Context<Self>) {
349 cx.emit(DismissEvent);
350 }
351
352 fn import_threads(
353 &mut self,
354 _: &menu::SecondaryConfirm,
355 _: &mut Window,
356 cx: &mut Context<Self>,
357 ) {
358 if self.is_importing || !self.has_checked_selectable_agent() {
359 return;
360 }
361
362 self.is_importing = true;
363 self.last_error = None;
364
365 let existing_sessions: HashSet<acp::SessionId> = ThreadMetadataStore::global(cx)
366 .read(cx)
367 .entries()
368 .filter_map(|metadata| metadata.session_id.clone())
369 .collect();
370
371 let selected_sessions_by_agent = self
372 .sessions_by_agent
373 .iter()
374 .filter(|sessions| {
375 self.agent_is_selectable(&sessions.agent_id)
376 && !self.unchecked_agents.contains(&sessions.agent_id)
377 })
378 .cloned()
379 .collect::<Vec<_>>();
380 let threads = collect_importable_threads(selected_sessions_by_agent, existing_sessions);
381 let imported_count = threads.len();
382
383 ThreadMetadataStore::global(cx).update(cx, |store, cx| store.save_all(threads, cx));
384
385 self.is_importing = false;
386 self.show_imported_threads_toast(imported_count, cx);
387 cx.emit(DismissEvent);
388 }
389
390 fn show_imported_threads_toast(&self, imported_count: usize, cx: &mut App) {
391 let status_toast = if imported_count == 0 {
392 StatusToast::new("No threads found to import.", cx, |this, _cx| {
393 this.icon(
394 Icon::new(IconName::Info)
395 .size(IconSize::Small)
396 .color(Color::Muted),
397 )
398 .dismiss_button(true)
399 })
400 } else {
401 let message = if imported_count == 1 {
402 "Imported 1 thread.".to_string()
403 } else {
404 format!("Imported {imported_count} threads.")
405 };
406 StatusToast::new(message, cx, |this, _cx| {
407 this.icon(
408 Icon::new(IconName::Check)
409 .size(IconSize::Small)
410 .color(Color::Success),
411 )
412 .dismiss_button(true)
413 })
414 };
415
416 self.workspace
417 .update(cx, |workspace, cx| {
418 workspace.toggle_status_toast(status_toast, cx);
419 })
420 .log_err();
421 }
422}
423
424impl EventEmitter<DismissEvent> for ThreadImportModal {}
425
426impl Focusable for ThreadImportModal {
427 fn focus_handle(&self, _cx: &App) -> FocusHandle {
428 self.focus_handle.clone()
429 }
430}
431
432impl ModalView for ThreadImportModal {}
433
434impl Render for ThreadImportModal {
435 fn render(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
436 let has_agents = !self.agent_entries.is_empty();
437 let disabled_import_thread =
438 self.is_importing || !has_agents || !self.has_checked_selectable_agent();
439
440 let agent_rows = self
441 .agent_entries
442 .iter()
443 .enumerate()
444 .map(|(ix, entry)| {
445 let status = self
446 .agent_import_statuses
447 .get(&entry.agent_id)
448 .cloned()
449 .unwrap_or(AgentImportStatus::Loading);
450 let is_selectable = status.is_selectable();
451 let is_checked = is_selectable && !self.unchecked_agents.contains(&entry.agent_id);
452 let is_focused = self.selected_index == Some(ix);
453 let row_disabled = self.is_importing || !is_selectable;
454 let checkbox_state = if is_checked {
455 ToggleState::Selected
456 } else {
457 ToggleState::Unselected
458 };
459 let end_slot = match &status {
460 AgentImportStatus::Loading
461 | AgentImportStatus::Unsupported
462 | AgentImportStatus::Error(_) => {
463 Checkbox::new(("thread-import-agent-checkbox", ix), checkbox_state)
464 .disabled(true)
465 .into_any_element()
466 }
467 AgentImportStatus::Ready { .. } => {
468 Checkbox::new(("thread-import-agent-checkbox", ix), checkbox_state)
469 .disabled(row_disabled)
470 .into_any_element()
471 }
472 };
473
474 let is_loading = matches!(status, AgentImportStatus::Loading);
475
476 let icon_color = if is_checked {
477 Color::Muted
478 } else {
479 Color::Disabled
480 };
481
482 let item = h_flex()
483 .w_full()
484 .gap_2()
485 .child(if let Some(icon_path) = entry.icon_path.clone() {
486 Icon::from_external_svg(icon_path)
487 .color(icon_color)
488 .size(IconSize::Small)
489 } else {
490 Icon::new(IconName::Sparkle)
491 .color(icon_color)
492 .size(IconSize::Small)
493 })
494 .child(
495 Label::new(entry.display_name.clone())
496 .when(!is_checked, |s| s.color(Color::Disabled)),
497 )
498 .map(|this| match status {
499 AgentImportStatus::Loading => this,
500 AgentImportStatus::Ready {
501 importable_count: count,
502 } => {
503 let label: SharedString = if count == 0 {
504 "No threads".into()
505 } else {
506 format!("{} threads", count).into()
507 };
508 this.child(Label::new(label).size(LabelSize::Small).color(Color::Muted))
509 }
510 AgentImportStatus::Unsupported => this.child(
511 Icon::new(IconName::Warning)
512 .color(Color::Warning)
513 .size(IconSize::Small),
514 ),
515 AgentImportStatus::Error(_) => this.child(
516 Icon::new(IconName::XCircle)
517 .color(Color::Error)
518 .size(IconSize::Small),
519 ),
520 });
521
522 let item = if is_loading {
523 item.with_animation(
524 "pulsating-icon",
525 Animation::new(Duration::from_secs(1))
526 .repeat()
527 .with_easing(pulsating_between(0.2, 0.6)),
528 |icon, delta| icon.opacity(delta),
529 )
530 .into_any_element()
531 } else {
532 item.into_any_element()
533 };
534
535 ListItem::new(("thread-import-agent", ix))
536 .rounded()
537 .spacing(ListItemSpacing::Sparse)
538 .focused(is_focused)
539 .disabled(row_disabled)
540 .child(item)
541 .end_slot(end_slot)
542 .when_some(status.tooltip_text(), |this, tooltip| {
543 this.tooltip(Tooltip::text(tooltip))
544 })
545 .on_click({
546 let agent_id = entry.agent_id.clone();
547 cx.listener(move |this, _event, _window, cx| {
548 this.toggle_agent_checked(agent_id.clone(), cx);
549 })
550 })
551 })
552 .collect::<Vec<_>>();
553
554 v_flex()
555 .id("thread-import-modal")
556 .key_context("ThreadImportModal")
557 .w(rems(34.))
558 .elevation_3(cx)
559 .overflow_hidden()
560 .track_focus(&self.focus_handle)
561 .on_action(cx.listener(Self::cancel))
562 .on_action(cx.listener(Self::confirm))
563 .on_action(cx.listener(Self::select_next))
564 .on_action(cx.listener(Self::select_previous))
565 .on_action(cx.listener(Self::import_threads))
566 .on_any_mouse_down(cx.listener(|this, _: &MouseDownEvent, window, cx| {
567 this.focus_handle.focus(window, cx);
568 }))
569 .child(
570 Modal::new("import-threads", None)
571 .header(
572 ModalHeader::new()
573 .headline("Import External Agent Threads")
574 .description(
575 "Import threads from agents like Claude Agent, Codex, and more, whether started in Omega or another client. \
576 Choose which agents to include, and their threads will appear in your thread history."
577 )
578 .show_dismiss_button(true),
579
580 )
581 .section(
582 Section::new().child(
583 v_flex()
584 .id("thread-import-agent-list")
585 .max_h(rems_from_px(320.))
586 .pb_1()
587 .overflow_y_scroll()
588 .when(has_agents, |this| this.children(agent_rows))
589 .when(!has_agents, |this| {
590 this.child(
591 Label::new("No external agents available.")
592 .color(Color::Muted)
593 .size(LabelSize::Small),
594 )
595 }),
596 ),
597 )
598 .footer(
599 ModalFooter::new()
600 .when(self.is_fetching_sessions, |this| {
601 this.start_slot(
602 h_flex()
603 .gap_1()
604 .child(
605 Icon::new(IconName::LoadCircle)
606 .size(IconSize::Small)
607 .color(Color::Muted)
608 .with_rotate_animation(3),
609 )
610 .child(Label::new("Fetching Agent Threads…")
611 .size(LabelSize::Small)
612 .color(Color::Muted))
613
614 )
615 })
616 .when_some(self.last_error.clone(), |this, error| {
617 this.start_slot(
618 Label::new(error)
619 .size(LabelSize::Small)
620 .color(Color::Error)
621 .truncate(),
622 )
623 })
624 .end_slot(
625 Button::new("import-threads", "Import Threads")
626 .loading(self.is_importing)
627 .disabled(disabled_import_thread)
628 .key_binding(
629 KeyBinding::for_action(&menu::SecondaryConfirm, cx)
630 .map(|kb| kb.size(rems_from_px(12.))),
631 )
632 .on_click(cx.listener(|this, _, window, cx| {
633 this.import_threads(&menu::SecondaryConfirm, window, cx);
634 })),
635 ),
636 ),
637 )
638 }
639}
640
641fn resolve_agent_connection_stores(
642 multi_workspace: &Entity<MultiWorkspace>,
643 cx: &App,
644) -> Vec<Entity<AgentConnectionStore>> {
645 let mut stores = Vec::new();
646 let mut included_local_store = false;
647
648 for workspace in multi_workspace.read(cx).workspaces() {
649 let workspace = workspace.read(cx);
650 let project = workspace.project().read(cx);
651
652 // We only want to include scores from one local workspace, since we
653 // know that they live on the same machine
654 let include_store = if project.is_remote() {
655 true
656 } else if project.is_local() && !included_local_store {
657 included_local_store = true;
658 true
659 } else {
660 false
661 };
662
663 if !include_store {
664 continue;
665 }
666
667 if let Some(panel) = workspace.panel::<AgentPanel>(cx) {
668 stores.push(panel.read(cx).connection_store().clone());
669 }
670 }
671
672 stores
673}
674
675struct AgentSessionFetchResult {
676 agent_id: AgentId,
677 sessions_by_agent: Vec<SessionByAgent>,
678 status: AgentImportStatus,
679}
680
681#[derive(Default)]
682struct AgentSessionFetchStats {
683 supported_attempt_count: usize,
684 successful_attempt_count: usize,
685 unsupported_attempt_count: usize,
686 errors: Vec<SharedString>,
687}
688
689fn fetch_sessions_for_agent(
690 agent_id: AgentId,
691 existing_sessions: HashSet<acp::SessionId>,
692 stores: Vec<Entity<AgentConnectionStore>>,
693 cx: &mut App,
694) -> Task<AgentSessionFetchResult> {
695 let mut wait_for_connection_tasks = Vec::new();
696
697 for store in stores {
698 let remote_connection = store
699 .read(cx)
700 .project()
701 .read(cx)
702 .remote_connection_options(cx);
703 let agent = Agent::from(agent_id.clone());
704 let server = agent.server(<dyn Fs>::global(cx), ThreadStore::global(cx));
705 let entry = store.update(cx, |store, cx| store.request_connection(agent, server, cx));
706
707 wait_for_connection_tasks.push(entry.read(cx).wait_for_connection().map({
708 let agent_id = agent_id.clone();
709 let remote_connection = remote_connection.clone();
710 move |result| (agent_id, remote_connection, result)
711 }));
712 }
713
714 cx.spawn(async move |cx| {
715 let mut stats = AgentSessionFetchStats::default();
716 let results = futures::future::join_all(wait_for_connection_tasks).await;
717
718 let mut page_tasks = Vec::new();
719 for (agent_id, remote_connection, result) in results {
720 let state = match result {
721 Ok(state) => state,
722 Err(error) => {
723 log::warn!("Failed to connect to {agent_id} to list sessions: {error}");
724 stats.errors.push(error.to_string().into());
725 continue;
726 }
727 };
728
729 let Some(list) = cx.update(|cx| state.connection.session_list(cx)) else {
730 stats.unsupported_attempt_count += 1;
731 continue;
732 };
733
734 stats.supported_attempt_count += 1;
735 page_tasks.push(cx.spawn({
736 let list = list.clone();
737 let agent_id_for_error = agent_id.clone();
738 async move |cx| {
739 (
740 agent_id_for_error,
741 collect_all_sessions(agent_id, remote_connection, list, cx).await,
742 )
743 }
744 }));
745 }
746
747 let mut sessions_by_agent = Vec::new();
748 for (agent_id, result) in futures::future::join_all(page_tasks).await {
749 match result {
750 Ok(sessions) => {
751 stats.successful_attempt_count += 1;
752 sessions_by_agent.push(sessions);
753 }
754 Err(error) => {
755 log::warn!("Failed to list sessions for {agent_id}: {error}");
756 stats.errors.push(error.to_string().into());
757 }
758 }
759 }
760
761 let importable_counts_by_agent =
762 count_importable_threads_by_agent(&sessions_by_agent, &existing_sessions);
763 let status = if stats.successful_attempt_count > 0 {
764 AgentImportStatus::Ready {
765 importable_count: importable_counts_by_agent
766 .get(&agent_id)
767 .copied()
768 .unwrap_or(0),
769 }
770 } else if stats.supported_attempt_count > 0 {
771 AgentImportStatus::Error(
772 stats
773 .errors
774 .first()
775 .cloned()
776 .unwrap_or_else(|| "Failed to list sessions.".into()),
777 )
778 } else if stats.unsupported_attempt_count > 0 {
779 AgentImportStatus::Unsupported
780 } else if let Some(error) = stats.errors.first().cloned() {
781 AgentImportStatus::Error(error)
782 } else {
783 AgentImportStatus::Unsupported
784 };
785
786 AgentSessionFetchResult {
787 agent_id,
788 sessions_by_agent,
789 status,
790 }
791 })
792}
793
794async fn collect_all_sessions(
795 agent_id: AgentId,
796 remote_connection: Option<RemoteConnectionOptions>,
797 list: std::rc::Rc<dyn acp_thread::AgentSessionList>,
798 cx: &mut gpui::AsyncApp,
799) -> anyhow::Result<SessionByAgent> {
800 let mut sessions = Vec::new();
801 let mut cursor: Option<String> = None;
802 loop {
803 let request = AgentSessionListRequest {
804 cursor: cursor.clone(),
805 ..Default::default()
806 };
807 let task = cx.update(|cx| list.list_sessions(request, cx));
808 let response = task.await?;
809 sessions.extend(response.sessions);
810 match response.next_cursor {
811 Some(next) if Some(&next) != cursor.as_ref() => cursor = Some(next),
812 _ => break,
813 }
814 }
815 Ok(SessionByAgent {
816 agent_id,
817 remote_connection,
818 sessions,
819 })
820}
821
822#[derive(Clone)]
823struct SessionByAgent {
824 agent_id: AgentId,
825 remote_connection: Option<RemoteConnectionOptions>,
826 sessions: Vec<acp_thread::AgentSessionInfo>,
827}
828
829fn count_importable_threads_by_agent(
830 sessions_by_agent: &[SessionByAgent],
831 existing_sessions: &HashSet<acp::SessionId>,
832) -> HashMap<AgentId, usize> {
833 let mut counts_by_agent = HashMap::default();
834 let mut seen_sessions_by_agent = HashMap::<AgentId, HashSet<acp::SessionId>>::default();
835
836 for sessions_for_agent in sessions_by_agent {
837 let seen_sessions = seen_sessions_by_agent
838 .entry(sessions_for_agent.agent_id.clone())
839 .or_insert_with(|| existing_sessions.clone());
840 for session in &sessions_for_agent.sessions {
841 if !seen_sessions.insert(session.session_id.clone()) {
842 continue;
843 }
844 if session.work_dirs.is_some() {
845 *counts_by_agent
846 .entry(sessions_for_agent.agent_id.clone())
847 .or_insert(0) += 1;
848 }
849 }
850 }
851
852 counts_by_agent
853}
854
855fn collect_importable_threads(
856 sessions_by_agent: Vec<SessionByAgent>,
857 mut existing_sessions: HashSet<acp::SessionId>,
858) -> Vec<ThreadMetadata> {
859 let mut to_insert = Vec::new();
860 for SessionByAgent {
861 agent_id,
862 remote_connection,
863 sessions,
864 } in sessions_by_agent
865 {
866 for session in sessions {
867 if !existing_sessions.insert(session.session_id.clone()) {
868 continue;
869 }
870 let Some(folder_paths) = session.work_dirs else {
871 continue;
872 };
873 to_insert.push(ThreadMetadata {
874 thread_id: ThreadId::new(),
875 session_id: Some(session.session_id),
876 agent_id: agent_id.clone(),
877 title: session.title,
878 title_override: None,
879 updated_at: session.updated_at.unwrap_or_else(|| Utc::now()),
880 created_at: session.created_at,
881 interacted_at: None,
882 worktree_paths: WorktreePaths::from_folder_paths(&folder_paths),
883 remote_connection: remote_connection.clone(),
884 archived: true,
885 });
886 }
887 }
888 to_insert
889}
890
891pub fn import_threads_from_other_channels(_workspace: &mut Workspace, cx: &mut Context<Workspace>) {
892 let database_dir = paths::database_dir().clone();
893 import_threads_from_other_channels_in(database_dir, cx);
894}
895
896fn import_threads_from_other_channels_in(
897 database_dir: std::path::PathBuf,
898 cx: &mut Context<Workspace>,
899) {
900 let current_channel = ReleaseChannel::global(cx);
901
902 let existing_thread_ids: HashSet<ThreadId> = ThreadMetadataStore::global(cx)
903 .read(cx)
904 .entries()
905 .map(|metadata| metadata.thread_id)
906 .collect();
907
908 let workspace_handle = cx.weak_entity();
909 cx.spawn(async move |_this, cx| {
910 let mut imported_threads = Vec::new();
911
912 for channel in &ReleaseChannel::ALL {
913 if *channel == current_channel || *channel == ReleaseChannel::Dev {
914 continue;
915 }
916
917 match read_threads_from_channel(&database_dir, *channel) {
918 Ok(threads) => {
919 let new_threads = threads
920 .into_iter()
921 .filter(|thread| !existing_thread_ids.contains(&thread.thread_id));
922 imported_threads.extend(new_threads);
923 }
924 Err(error) => {
925 log::warn!(
926 "Failed to read threads from {} channel database: {}",
927 channel.dev_name(),
928 error
929 );
930 }
931 }
932 }
933
934 let imported_count = imported_threads.len();
935
936 cx.update(|cx| {
937 ThreadMetadataStore::global(cx)
938 .update(cx, |store, cx| store.save_all(imported_threads, cx));
939
940 show_cross_channel_import_toast(&workspace_handle, imported_count, cx);
941 })
942 })
943 .detach();
944}
945
946fn read_threads_from_channel(
947 database_dir: &std::path::Path,
948 channel: ReleaseChannel,
949) -> anyhow::Result<Vec<ThreadMetadata>> {
950 let db_path = db::db_path(database_dir, channel);
951 if !db_path.exists() {
952 return Ok(Vec::new());
953 }
954 let connection = sqlez::connection::Connection::open_file(&db_path.to_string_lossy());
955 crate::thread_metadata_store::list_thread_metadata_from_connection(&connection)
956}
957
958fn show_cross_channel_import_toast(
959 workspace: &WeakEntity<Workspace>,
960 imported_count: usize,
961 cx: &mut App,
962) {
963 let status_toast = if imported_count == 0 {
964 StatusToast::new("No new threads found to import.", cx, |this, _cx| {
965 this.icon(Icon::new(IconName::Info).color(Color::Muted))
966 .dismiss_button(true)
967 })
968 } else {
969 let message = if imported_count == 1 {
970 "Imported 1 thread from other channels.".to_string()
971 } else {
972 format!("Imported {imported_count} threads from other channels.")
973 };
974 StatusToast::new(message, cx, |this, _cx| {
975 this.icon(Icon::new(IconName::Check).color(Color::Success))
976 .dismiss_button(true)
977 })
978 };
979
980 workspace
981 .update(cx, |workspace, cx| {
982 workspace.toggle_status_toast(status_toast, cx);
983 })
984 .log_err();
985}
986
987#[cfg(test)]
988mod tests {
989 use super::*;
990 use acp_thread::AgentSessionInfo;
991 use chrono::Utc;
992 use gpui::TestAppContext;
993 use std::path::Path;
994 use workspace::PathList;
995
996 fn make_session(
997 session_id: &str,
998 title: Option<&str>,
999 work_dirs: Option<PathList>,
1000 updated_at: Option<chrono::DateTime<Utc>>,
1001 created_at: Option<chrono::DateTime<Utc>>,
1002 ) -> AgentSessionInfo {
1003 AgentSessionInfo {
1004 session_id: acp::SessionId::new(session_id),
1005 title: title.map(|t| SharedString::from(t.to_string())),
1006 work_dirs,
1007 updated_at,
1008 created_at,
1009 meta: None,
1010 }
1011 }
1012
1013 #[test]
1014 fn test_collect_skips_sessions_already_in_existing_set() {
1015 let existing = HashSet::from_iter(vec![acp::SessionId::new("existing-1")]);
1016 let paths = PathList::new(&[Path::new("/project")]);
1017
1018 let sessions_by_agent = vec![SessionByAgent {
1019 agent_id: AgentId::new("agent-a"),
1020 remote_connection: None,
1021 sessions: vec![
1022 make_session(
1023 "existing-1",
1024 Some("Already There"),
1025 Some(paths.clone()),
1026 None,
1027 None,
1028 ),
1029 make_session("new-1", Some("Brand New"), Some(paths), None, None),
1030 ],
1031 }];
1032
1033 let result = collect_importable_threads(sessions_by_agent, existing);
1034
1035 assert_eq!(result.len(), 1);
1036 assert_eq!(result[0].session_id.as_ref().unwrap().0.as_ref(), "new-1");
1037 assert_eq!(result[0].display_title(), "Brand New");
1038 }
1039
1040 #[test]
1041 fn test_collect_skips_sessions_without_work_dirs() {
1042 let existing = HashSet::default();
1043 let paths = PathList::new(&[Path::new("/project")]);
1044
1045 let sessions_by_agent = vec![SessionByAgent {
1046 agent_id: AgentId::new("agent-a"),
1047 remote_connection: None,
1048 sessions: vec![
1049 make_session("has-dirs", Some("With Dirs"), Some(paths), None, None),
1050 make_session("no-dirs", Some("No Dirs"), None, None, None),
1051 ],
1052 }];
1053
1054 let result = collect_importable_threads(sessions_by_agent, existing);
1055
1056 assert_eq!(result.len(), 1);
1057 assert_eq!(
1058 result[0].session_id.as_ref().unwrap().0.as_ref(),
1059 "has-dirs"
1060 );
1061 }
1062
1063 #[test]
1064 fn test_collect_marks_all_imported_threads_as_archived() {
1065 let existing = HashSet::default();
1066 let paths = PathList::new(&[Path::new("/project")]);
1067
1068 let sessions_by_agent = vec![SessionByAgent {
1069 agent_id: AgentId::new("agent-a"),
1070 remote_connection: None,
1071 sessions: vec![
1072 make_session("s1", Some("Thread 1"), Some(paths.clone()), None, None),
1073 make_session("s2", Some("Thread 2"), Some(paths), None, None),
1074 ],
1075 }];
1076
1077 let result = collect_importable_threads(sessions_by_agent, existing);
1078
1079 assert_eq!(result.len(), 2);
1080 assert!(result.iter().all(|t| t.archived));
1081 }
1082
1083 #[test]
1084 fn test_collect_assigns_correct_agent_id_per_session() {
1085 let existing = HashSet::default();
1086 let paths = PathList::new(&[Path::new("/project")]);
1087
1088 let sessions_by_agent = vec![
1089 SessionByAgent {
1090 agent_id: AgentId::new("agent-a"),
1091 remote_connection: None,
1092 sessions: vec![make_session(
1093 "s1",
1094 Some("From A"),
1095 Some(paths.clone()),
1096 None,
1097 None,
1098 )],
1099 },
1100 SessionByAgent {
1101 agent_id: AgentId::new("agent-b"),
1102 remote_connection: None,
1103 sessions: vec![make_session("s2", Some("From B"), Some(paths), None, None)],
1104 },
1105 ];
1106
1107 let result = collect_importable_threads(sessions_by_agent, existing);
1108
1109 assert_eq!(result.len(), 2);
1110 let s1 = result
1111 .iter()
1112 .find(|t| t.session_id.as_ref().map(|s| s.0.as_ref()) == Some("s1"))
1113 .unwrap();
1114 let s2 = result
1115 .iter()
1116 .find(|t| t.session_id.as_ref().map(|s| s.0.as_ref()) == Some("s2"))
1117 .unwrap();
1118 assert_eq!(s1.agent_id.as_ref(), "agent-a");
1119 assert_eq!(s2.agent_id.as_ref(), "agent-b");
1120 }
1121
1122 #[test]
1123 fn test_collect_deduplicates_across_agents() {
1124 let existing = HashSet::default();
1125 let paths = PathList::new(&[Path::new("/project")]);
1126
1127 let sessions_by_agent = vec![
1128 SessionByAgent {
1129 agent_id: AgentId::new("agent-a"),
1130 remote_connection: None,
1131 sessions: vec![make_session(
1132 "shared-session",
1133 Some("From A"),
1134 Some(paths.clone()),
1135 None,
1136 None,
1137 )],
1138 },
1139 SessionByAgent {
1140 agent_id: AgentId::new("agent-b"),
1141 remote_connection: None,
1142 sessions: vec![make_session(
1143 "shared-session",
1144 Some("From B"),
1145 Some(paths),
1146 None,
1147 None,
1148 )],
1149 },
1150 ];
1151
1152 let result = collect_importable_threads(sessions_by_agent, existing);
1153
1154 assert_eq!(result.len(), 1);
1155 assert_eq!(
1156 result[0].session_id.as_ref().unwrap().0.as_ref(),
1157 "shared-session"
1158 );
1159 assert_eq!(
1160 result[0].agent_id.as_ref(),
1161 "agent-a",
1162 "first agent encountered should win"
1163 );
1164 }
1165
1166 #[test]
1167 fn test_collect_all_existing_returns_empty() {
1168 let paths = PathList::new(&[Path::new("/project")]);
1169 let existing =
1170 HashSet::from_iter(vec![acp::SessionId::new("s1"), acp::SessionId::new("s2")]);
1171
1172 let sessions_by_agent = vec![SessionByAgent {
1173 agent_id: AgentId::new("agent-a"),
1174 remote_connection: None,
1175 sessions: vec![
1176 make_session("s1", Some("T1"), Some(paths.clone()), None, None),
1177 make_session("s2", Some("T2"), Some(paths), None, None),
1178 ],
1179 }];
1180
1181 let result = collect_importable_threads(sessions_by_agent, existing);
1182 assert!(result.is_empty());
1183 }
1184
1185 #[test]
1186 fn test_count_importable_threads_by_agent_counts_each_agent_independently() {
1187 let existing = HashSet::from_iter(vec![acp::SessionId::new("existing")]);
1188 let paths = PathList::new(&[Path::new("/project")]);
1189 let sessions_by_agent = vec![
1190 SessionByAgent {
1191 agent_id: AgentId::new("agent-a"),
1192 remote_connection: None,
1193 sessions: vec![
1194 make_session(
1195 "existing",
1196 Some("Existing"),
1197 Some(paths.clone()),
1198 None,
1199 None,
1200 ),
1201 make_session("shared", Some("Shared A"), Some(paths.clone()), None, None),
1202 make_session("no-dirs", Some("No Dirs"), None, None, None),
1203 ],
1204 },
1205 SessionByAgent {
1206 agent_id: AgentId::new("agent-b"),
1207 remote_connection: None,
1208 sessions: vec![make_session(
1209 "shared",
1210 Some("Shared B"),
1211 Some(paths),
1212 None,
1213 None,
1214 )],
1215 },
1216 ];
1217
1218 let counts = count_importable_threads_by_agent(&sessions_by_agent, &existing);
1219
1220 assert_eq!(counts.get(&AgentId::new("agent-a")), Some(&1));
1221 assert_eq!(counts.get(&AgentId::new("agent-b")), Some(&1));
1222 }
1223
1224 fn create_channel_db(
1225 db_dir: &std::path::Path,
1226 channel: ReleaseChannel,
1227 ) -> db::sqlez::connection::Connection {
1228 let db_path = db::db_path(db_dir, channel);
1229 std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
1230 let connection = db::sqlez::connection::Connection::open_file(&db_path.to_string_lossy());
1231 crate::thread_metadata_store::run_thread_metadata_migrations(&connection);
1232 connection
1233 }
1234
1235 fn insert_thread(
1236 connection: &db::sqlez::connection::Connection,
1237 title: &str,
1238 updated_at: &str,
1239 archived: bool,
1240 ) {
1241 let thread_id = uuid::Uuid::new_v4();
1242 let session_id = uuid::Uuid::new_v4().to_string();
1243 connection
1244 .exec_bound::<(uuid::Uuid, &str, &str, &str, bool)>(
1245 "INSERT INTO sidebar_threads \
1246 (thread_id, session_id, title, updated_at, archived) \
1247 VALUES (?1, ?2, ?3, ?4, ?5)",
1248 )
1249 .unwrap()((thread_id, session_id.as_str(), title, updated_at, archived))
1250 .unwrap();
1251 }
1252
1253 #[test]
1254 fn test_returns_empty_when_channel_db_missing() {
1255 let dir = tempfile::tempdir().unwrap();
1256 let threads = read_threads_from_channel(dir.path(), ReleaseChannel::Nightly).unwrap();
1257 assert!(threads.is_empty());
1258 }
1259
1260 #[test]
1261 fn test_preserves_archived_state() {
1262 let dir = tempfile::tempdir().unwrap();
1263 let connection = create_channel_db(dir.path(), ReleaseChannel::Nightly);
1264
1265 insert_thread(&connection, "Active Thread", "2025-01-15T10:00:00Z", false);
1266 insert_thread(&connection, "Archived Thread", "2025-01-15T09:00:00Z", true);
1267 drop(connection);
1268
1269 let threads = read_threads_from_channel(dir.path(), ReleaseChannel::Nightly).unwrap();
1270 assert_eq!(threads.len(), 2);
1271
1272 let active = threads
1273 .iter()
1274 .find(|t| t.display_title().as_ref() == "Active Thread")
1275 .unwrap();
1276 assert!(!active.archived);
1277
1278 let archived = threads
1279 .iter()
1280 .find(|t| t.display_title().as_ref() == "Archived Thread")
1281 .unwrap();
1282 assert!(archived.archived);
1283 }
1284
1285 fn init_test(cx: &mut TestAppContext) {
1286 let fs = fs::FakeFs::new(cx.executor());
1287 cx.update(|cx| {
1288 let settings_store = settings::SettingsStore::test(cx);
1289 cx.set_global(settings_store);
1290 theme_settings::init(theme::LoadThemes::JustBase, cx);
1291 release_channel::init("0.0.0".parse().unwrap(), cx);
1292 <dyn fs::Fs>::set_global(fs, cx);
1293 ThreadMetadataStore::init_global(cx);
1294 });
1295 cx.run_until_parked();
1296 }
1297
1298 /// Returns two release channels that are not the current one and not Dev.
1299 /// This ensures tests work regardless of which release channel branch
1300 /// they run on.
1301 fn foreign_channels(cx: &TestAppContext) -> (ReleaseChannel, ReleaseChannel) {
1302 let current = cx.update(|cx| ReleaseChannel::global(cx));
1303 let mut channels = ReleaseChannel::ALL
1304 .iter()
1305 .copied()
1306 .filter(|ch| *ch != current && *ch != ReleaseChannel::Dev);
1307 (channels.next().unwrap(), channels.next().unwrap())
1308 }
1309
1310 #[gpui::test]
1311 async fn test_import_threads_from_other_channels(cx: &mut TestAppContext) {
1312 init_test(cx);
1313
1314 let dir = tempfile::tempdir().unwrap();
1315 let database_dir = dir.path().to_path_buf();
1316
1317 let (channel_a, channel_b) = foreign_channels(cx);
1318
1319 // Set up databases for two foreign channels.
1320 let db_a = create_channel_db(dir.path(), channel_a);
1321 insert_thread(&db_a, "Thread A1", "2025-01-15T10:00:00Z", false);
1322 insert_thread(&db_a, "Thread A2", "2025-01-15T11:00:00Z", true);
1323 drop(db_a);
1324
1325 let db_b = create_channel_db(dir.path(), channel_b);
1326 insert_thread(&db_b, "Thread B1", "2025-01-15T12:00:00Z", false);
1327 drop(db_b);
1328
1329 // Create a workspace and run the import.
1330 let fs = fs::FakeFs::new(cx.executor());
1331 let project = project::Project::test(fs, [], cx).await;
1332 let multi_workspace =
1333 cx.add_window(|window, cx| MultiWorkspace::test_new(project, window, cx));
1334 let workspace_entity = multi_workspace
1335 .read_with(cx, |mw, _cx| mw.workspace().clone())
1336 .unwrap();
1337 let mut vcx = gpui::VisualTestContext::from_window(multi_workspace.into(), cx);
1338
1339 workspace_entity.update_in(&mut vcx, |_workspace, _window, cx| {
1340 import_threads_from_other_channels_in(database_dir, cx);
1341 });
1342 cx.run_until_parked();
1343
1344 // Verify all three threads were imported into the store.
1345 cx.update(|cx| {
1346 let store = ThreadMetadataStore::global(cx);
1347 let store = store.read(cx);
1348 let titles: collections::HashSet<String> = store
1349 .entries()
1350 .map(|m| m.display_title().to_string())
1351 .collect();
1352
1353 assert_eq!(titles.len(), 3);
1354 assert!(titles.contains("Thread A1"));
1355 assert!(titles.contains("Thread A2"));
1356 assert!(titles.contains("Thread B1"));
1357
1358 // Verify archived state is preserved.
1359 let thread_a2 = store
1360 .entries()
1361 .find(|m| m.display_title().as_ref() == "Thread A2")
1362 .unwrap();
1363 assert!(thread_a2.archived);
1364
1365 let thread_b1 = store
1366 .entries()
1367 .find(|m| m.display_title().as_ref() == "Thread B1")
1368 .unwrap();
1369 assert!(!thread_b1.archived);
1370 });
1371 }
1372
1373 #[gpui::test]
1374 async fn test_import_skips_already_existing_threads(cx: &mut TestAppContext) {
1375 init_test(cx);
1376
1377 let dir = tempfile::tempdir().unwrap();
1378 let database_dir = dir.path().to_path_buf();
1379
1380 let (channel_a, _) = foreign_channels(cx);
1381
1382 // Set up a database for a foreign channel.
1383 let db_a = create_channel_db(dir.path(), channel_a);
1384 insert_thread(&db_a, "Thread A", "2025-01-15T10:00:00Z", false);
1385 insert_thread(&db_a, "Thread B", "2025-01-15T11:00:00Z", false);
1386 drop(db_a);
1387
1388 // Read the threads so we can pre-populate one into the store.
1389 let foreign_threads = read_threads_from_channel(dir.path(), channel_a).unwrap();
1390 let thread_a = foreign_threads
1391 .iter()
1392 .find(|t| t.display_title().as_ref() == "Thread A")
1393 .unwrap()
1394 .clone();
1395
1396 // Pre-populate Thread A into the store.
1397 cx.update(|cx| {
1398 ThreadMetadataStore::global(cx).update(cx, |store, cx| store.save(thread_a, cx));
1399 });
1400 cx.run_until_parked();
1401
1402 // Run the import.
1403 let fs = fs::FakeFs::new(cx.executor());
1404 let project = project::Project::test(fs, [], cx).await;
1405 let multi_workspace =
1406 cx.add_window(|window, cx| MultiWorkspace::test_new(project, window, cx));
1407 let workspace_entity = multi_workspace
1408 .read_with(cx, |mw, _cx| mw.workspace().clone())
1409 .unwrap();
1410 let mut vcx = gpui::VisualTestContext::from_window(multi_workspace.into(), cx);
1411
1412 workspace_entity.update_in(&mut vcx, |_workspace, _window, cx| {
1413 import_threads_from_other_channels_in(database_dir, cx);
1414 });
1415 cx.run_until_parked();
1416
1417 // Verify only Thread B was added (Thread A already existed).
1418 cx.update(|cx| {
1419 let store = ThreadMetadataStore::global(cx);
1420 let store = store.read(cx);
1421 assert_eq!(store.entries().count(), 2);
1422
1423 let titles: collections::HashSet<String> = store
1424 .entries()
1425 .map(|m| m.display_title().to_string())
1426 .collect();
1427 assert!(titles.contains("Thread A"));
1428 assert!(titles.contains("Thread B"));
1429 });
1430 }
1431}
1432