1use crate::{
2 ContextServerRegistry, Thread, ThreadEvent, ThreadsDatabase, ToolCallAuthorization,
3 UserMessageContent, templates::Templates,
4};
5use crate::{HistoryStore, TerminalHandle, ThreadEnvironment, TitleUpdated, TokenUsageUpdated};
6use acp_thread::{AcpThread, AgentModelSelector};
7use action_log::ActionLog;
8use agent_client_protocol as acp;
9use anyhow::{Context as _, Result, anyhow};
10use collections::{HashSet, IndexMap};
11use fs::Fs;
12use futures::channel::{mpsc, oneshot};
13use futures::future::Shared;
14use futures::{StreamExt, future};
15use gpui::{
16 App, AppContext, AsyncApp, Context, Entity, SharedString, Subscription, Task, WeakEntity,
17};
18use language_model::{LanguageModel, LanguageModelProvider, LanguageModelRegistry};
19use project::{Project, ProjectItem, ProjectPath, Worktree};
20use prompt_store::{
21 ProjectContext, PromptId, PromptStore, RulesFileContext, UserRulesContext, WorktreeContext,
22};
23use settings::{LanguageModelSelection, update_settings_file};
24use std::any::Any;
25use std::collections::HashMap;
26use std::path::{Path, PathBuf};
27use std::rc::Rc;
28use std::sync::Arc;
29use util::ResultExt;
30
31const RULES_FILE_NAMES: [&str; 9] = [
32 ".rules",
33 ".cursorrules",
34 ".windsurfrules",
35 ".clinerules",
36 ".github/copilot-instructions.md",
37 "CLAUDE.md",
38 "AGENT.md",
39 "AGENTS.md",
40 "GEMINI.md",
41];
42
43pub struct RulesLoadingError {
44 pub message: SharedString,
45}
46
47/// Holds both the internal Thread and the AcpThread for a session
48struct Session {
49 /// The internal thread that processes messages
50 thread: Entity<Thread>,
51 /// The ACP thread that handles protocol communication
52 acp_thread: WeakEntity<acp_thread::AcpThread>,
53 pending_save: Task<()>,
54 _subscriptions: Vec<Subscription>,
55}
56
57pub struct LanguageModels {
58 /// Access language model by ID
59 models: HashMap<acp_thread::AgentModelId, Arc<dyn LanguageModel>>,
60 /// Cached list for returning language model information
61 model_list: acp_thread::AgentModelList,
62 refresh_models_rx: watch::Receiver<()>,
63 refresh_models_tx: watch::Sender<()>,
64 _authenticate_all_providers_task: Task<()>,
65}
66
67impl LanguageModels {
68 fn new(cx: &mut App) -> Self {
69 let (refresh_models_tx, refresh_models_rx) = watch::channel(());
70
71 let mut this = Self {
72 models: HashMap::default(),
73 model_list: acp_thread::AgentModelList::Grouped(IndexMap::default()),
74 refresh_models_rx,
75 refresh_models_tx,
76 _authenticate_all_providers_task: Self::authenticate_all_language_model_providers(cx),
77 };
78 this.refresh_list(cx);
79 this
80 }
81
82 fn refresh_list(&mut self, cx: &App) {
83 let providers = LanguageModelRegistry::global(cx)
84 .read(cx)
85 .providers()
86 .into_iter()
87 .filter(|provider| provider.is_authenticated(cx))
88 .collect::<Vec<_>>();
89
90 let mut language_model_list = IndexMap::default();
91 let mut recommended_models = HashSet::default();
92
93 let mut recommended = Vec::new();
94 for provider in &providers {
95 for model in provider.recommended_models(cx) {
96 recommended_models.insert((model.provider_id(), model.id()));
97 recommended.push(Self::map_language_model_to_info(&model, provider));
98 }
99 }
100 if !recommended.is_empty() {
101 language_model_list.insert(
102 acp_thread::AgentModelGroupName("Recommended".into()),
103 recommended,
104 );
105 }
106
107 let mut models = HashMap::default();
108 for provider in providers {
109 let mut provider_models = Vec::new();
110 for model in provider.provided_models(cx) {
111 let model_info = Self::map_language_model_to_info(&model, &provider);
112 let model_id = model_info.id.clone();
113 if !recommended_models.contains(&(model.provider_id(), model.id())) {
114 provider_models.push(model_info);
115 }
116 models.insert(model_id, model);
117 }
118 if !provider_models.is_empty() {
119 language_model_list.insert(
120 acp_thread::AgentModelGroupName(provider.name().0.clone()),
121 provider_models,
122 );
123 }
124 }
125
126 self.models = models;
127 self.model_list = acp_thread::AgentModelList::Grouped(language_model_list);
128 self.refresh_models_tx.send(()).ok();
129 }
130
131 fn watch(&self) -> watch::Receiver<()> {
132 self.refresh_models_rx.clone()
133 }
134
135 pub fn model_from_id(
136 &self,
137 model_id: &acp_thread::AgentModelId,
138 ) -> Option<Arc<dyn LanguageModel>> {
139 self.models.get(model_id).cloned()
140 }
141
142 fn map_language_model_to_info(
143 model: &Arc<dyn LanguageModel>,
144 provider: &Arc<dyn LanguageModelProvider>,
145 ) -> acp_thread::AgentModelInfo {
146 acp_thread::AgentModelInfo {
147 id: Self::model_id(model),
148 name: model.name().0,
149 icon: Some(provider.icon()),
150 }
151 }
152
153 fn model_id(model: &Arc<dyn LanguageModel>) -> acp_thread::AgentModelId {
154 acp_thread::AgentModelId(format!("{}/{}", model.provider_id().0, model.id().0).into())
155 }
156
157 fn authenticate_all_language_model_providers(cx: &mut App) -> Task<()> {
158 let authenticate_all_providers = LanguageModelRegistry::global(cx)
159 .read(cx)
160 .providers()
161 .iter()
162 .map(|provider| (provider.id(), provider.name(), provider.authenticate(cx)))
163 .collect::<Vec<_>>();
164
165 cx.background_spawn(async move {
166 for (provider_id, provider_name, authenticate_task) in authenticate_all_providers {
167 if let Err(err) = authenticate_task.await {
168 match err {
169 language_model::AuthenticateError::CredentialsNotFound => {
170 // Since we're authenticating these providers in the
171 // background for the purposes of populating the
172 // language selector, we don't care about providers
173 // where the credentials are not found.
174 }
175 language_model::AuthenticateError::ConnectionRefused => {
176 // Not logging connection refused errors as they are mostly from LM Studio's noisy auth failures.
177 // LM Studio only has one auth method (endpoint call) which fails for users who haven't enabled it.
178 // TODO: Better manage LM Studio auth logic to avoid these noisy failures.
179 }
180 _ => {
181 // Some providers have noisy failure states that we
182 // don't want to spam the logs with every time the
183 // language model selector is initialized.
184 //
185 // Ideally these should have more clear failure modes
186 // that we know are safe to ignore here, like what we do
187 // with `CredentialsNotFound` above.
188 match provider_id.0.as_ref() {
189 "lmstudio" | "ollama" => {
190 // LM Studio and Ollama both make fetch requests to the local APIs to determine if they are "authenticated".
191 //
192 // These fail noisily, so we don't log them.
193 }
194 "copilot_chat" => {
195 // Copilot Chat returns an error if Copilot is not enabled, so we don't log those errors.
196 }
197 _ => {
198 log::error!(
199 "Failed to authenticate provider: {}: {err}",
200 provider_name.0
201 );
202 }
203 }
204 }
205 }
206 }
207 }
208 })
209 }
210}
211
212pub struct NativeAgent {
213 /// Session ID -> Session mapping
214 sessions: HashMap<acp::SessionId, Session>,
215 history: Entity<HistoryStore>,
216 /// Shared project context for all threads
217 project_context: Entity<ProjectContext>,
218 project_context_needs_refresh: watch::Sender<()>,
219 _maintain_project_context: Task<Result<()>>,
220 context_server_registry: Entity<ContextServerRegistry>,
221 /// Shared templates for all threads
222 templates: Arc<Templates>,
223 /// Cached model information
224 models: LanguageModels,
225 project: Entity<Project>,
226 prompt_store: Option<Entity<PromptStore>>,
227 fs: Arc<dyn Fs>,
228 _subscriptions: Vec<Subscription>,
229}
230
231impl NativeAgent {
232 pub async fn new(
233 project: Entity<Project>,
234 history: Entity<HistoryStore>,
235 templates: Arc<Templates>,
236 prompt_store: Option<Entity<PromptStore>>,
237 fs: Arc<dyn Fs>,
238 cx: &mut AsyncApp,
239 ) -> Result<Entity<NativeAgent>> {
240 log::debug!("Creating new NativeAgent");
241
242 let project_context = cx
243 .update(|cx| Self::build_project_context(&project, prompt_store.as_ref(), cx))?
244 .await;
245
246 cx.new(|cx| {
247 let mut subscriptions = vec![
248 cx.subscribe(&project, Self::handle_project_event),
249 cx.subscribe(
250 &LanguageModelRegistry::global(cx),
251 Self::handle_models_updated_event,
252 ),
253 ];
254 if let Some(prompt_store) = prompt_store.as_ref() {
255 subscriptions.push(cx.subscribe(prompt_store, Self::handle_prompts_updated_event))
256 }
257
258 let (project_context_needs_refresh_tx, project_context_needs_refresh_rx) =
259 watch::channel(());
260 Self {
261 sessions: HashMap::new(),
262 history,
263 project_context: cx.new(|_| project_context),
264 project_context_needs_refresh: project_context_needs_refresh_tx,
265 _maintain_project_context: cx.spawn(async move |this, cx| {
266 Self::maintain_project_context(this, project_context_needs_refresh_rx, cx).await
267 }),
268 context_server_registry: cx.new(|cx| {
269 ContextServerRegistry::new(project.read(cx).context_server_store(), cx)
270 }),
271 templates,
272 models: LanguageModels::new(cx),
273 project,
274 prompt_store,
275 fs,
276 _subscriptions: subscriptions,
277 }
278 })
279 }
280
281 fn register_session(
282 &mut self,
283 thread_handle: Entity<Thread>,
284 cx: &mut Context<Self>,
285 ) -> Entity<AcpThread> {
286 let connection = Rc::new(NativeAgentConnection(cx.entity()));
287
288 let thread = thread_handle.read(cx);
289 let session_id = thread.id().clone();
290 let title = thread.title();
291 let project = thread.project.clone();
292 let action_log = thread.action_log.clone();
293 let prompt_capabilities_rx = thread.prompt_capabilities_rx.clone();
294 let acp_thread = cx.new(|cx| {
295 acp_thread::AcpThread::new(
296 title,
297 connection,
298 project.clone(),
299 action_log.clone(),
300 session_id.clone(),
301 prompt_capabilities_rx,
302 cx,
303 )
304 });
305
306 let registry = LanguageModelRegistry::read_global(cx);
307 let summarization_model = registry.thread_summary_model().map(|c| c.model);
308
309 thread_handle.update(cx, |thread, cx| {
310 thread.set_summarization_model(summarization_model, cx);
311 thread.add_default_tools(
312 Rc::new(AcpThreadEnvironment {
313 acp_thread: acp_thread.downgrade(),
314 }) as _,
315 cx,
316 )
317 });
318
319 let subscriptions = vec![
320 cx.observe_release(&acp_thread, |this, acp_thread, _cx| {
321 this.sessions.remove(acp_thread.session_id());
322 }),
323 cx.subscribe(&thread_handle, Self::handle_thread_title_updated),
324 cx.subscribe(&thread_handle, Self::handle_thread_token_usage_updated),
325 cx.observe(&thread_handle, move |this, thread, cx| {
326 this.save_thread(thread, cx)
327 }),
328 ];
329
330 self.sessions.insert(
331 session_id,
332 Session {
333 thread: thread_handle,
334 acp_thread: acp_thread.downgrade(),
335 _subscriptions: subscriptions,
336 pending_save: Task::ready(()),
337 },
338 );
339 acp_thread
340 }
341
342 pub fn models(&self) -> &LanguageModels {
343 &self.models
344 }
345
346 async fn maintain_project_context(
347 this: WeakEntity<Self>,
348 mut needs_refresh: watch::Receiver<()>,
349 cx: &mut AsyncApp,
350 ) -> Result<()> {
351 while needs_refresh.changed().await.is_ok() {
352 let project_context = this
353 .update(cx, |this, cx| {
354 Self::build_project_context(&this.project, this.prompt_store.as_ref(), cx)
355 })?
356 .await;
357 this.update(cx, |this, cx| {
358 this.project_context = cx.new(|_| project_context);
359 })?;
360 }
361
362 Ok(())
363 }
364
365 fn build_project_context(
366 project: &Entity<Project>,
367 prompt_store: Option<&Entity<PromptStore>>,
368 cx: &mut App,
369 ) -> Task<ProjectContext> {
370 let worktrees = project.read(cx).visible_worktrees(cx).collect::<Vec<_>>();
371 let worktree_tasks = worktrees
372 .into_iter()
373 .map(|worktree| {
374 Self::load_worktree_info_for_system_prompt(worktree, project.clone(), cx)
375 })
376 .collect::<Vec<_>>();
377 let default_user_rules_task = if let Some(prompt_store) = prompt_store.as_ref() {
378 prompt_store.read_with(cx, |prompt_store, cx| {
379 let prompts = prompt_store.default_prompt_metadata();
380 let load_tasks = prompts.into_iter().map(|prompt_metadata| {
381 let contents = prompt_store.load(prompt_metadata.id, cx);
382 async move { (contents.await, prompt_metadata) }
383 });
384 cx.background_spawn(future::join_all(load_tasks))
385 })
386 } else {
387 Task::ready(vec![])
388 };
389
390 cx.spawn(async move |_cx| {
391 let (worktrees, default_user_rules) =
392 future::join(future::join_all(worktree_tasks), default_user_rules_task).await;
393
394 let worktrees = worktrees
395 .into_iter()
396 .map(|(worktree, _rules_error)| {
397 // TODO: show error message
398 // if let Some(rules_error) = rules_error {
399 // this.update(cx, |_, cx| cx.emit(rules_error)).ok();
400 // }
401 worktree
402 })
403 .collect::<Vec<_>>();
404
405 let default_user_rules = default_user_rules
406 .into_iter()
407 .flat_map(|(contents, prompt_metadata)| match contents {
408 Ok(contents) => Some(UserRulesContext {
409 uuid: match prompt_metadata.id {
410 PromptId::User { uuid } => uuid,
411 PromptId::EditWorkflow => return None,
412 },
413 title: prompt_metadata.title.map(|title| title.to_string()),
414 contents,
415 }),
416 Err(_err) => {
417 // TODO: show error message
418 // this.update(cx, |_, cx| {
419 // cx.emit(RulesLoadingError {
420 // message: format!("{err:?}").into(),
421 // });
422 // })
423 // .ok();
424 None
425 }
426 })
427 .collect::<Vec<_>>();
428
429 ProjectContext::new(worktrees, default_user_rules)
430 })
431 }
432
433 fn load_worktree_info_for_system_prompt(
434 worktree: Entity<Worktree>,
435 project: Entity<Project>,
436 cx: &mut App,
437 ) -> Task<(WorktreeContext, Option<RulesLoadingError>)> {
438 let tree = worktree.read(cx);
439 let root_name = tree.root_name().into();
440 let abs_path = tree.abs_path();
441
442 let mut context = WorktreeContext {
443 root_name,
444 abs_path,
445 rules_file: None,
446 };
447
448 let rules_task = Self::load_worktree_rules_file(worktree, project, cx);
449 let Some(rules_task) = rules_task else {
450 return Task::ready((context, None));
451 };
452
453 cx.spawn(async move |_| {
454 let (rules_file, rules_file_error) = match rules_task.await {
455 Ok(rules_file) => (Some(rules_file), None),
456 Err(err) => (
457 None,
458 Some(RulesLoadingError {
459 message: format!("{err}").into(),
460 }),
461 ),
462 };
463 context.rules_file = rules_file;
464 (context, rules_file_error)
465 })
466 }
467
468 fn load_worktree_rules_file(
469 worktree: Entity<Worktree>,
470 project: Entity<Project>,
471 cx: &mut App,
472 ) -> Option<Task<Result<RulesFileContext>>> {
473 let worktree = worktree.read(cx);
474 let worktree_id = worktree.id();
475 let selected_rules_file = RULES_FILE_NAMES
476 .into_iter()
477 .filter_map(|name| {
478 worktree
479 .entry_for_path(name)
480 .filter(|entry| entry.is_file())
481 .map(|entry| entry.path.clone())
482 })
483 .next();
484
485 // Note that Cline supports `.clinerules` being a directory, but that is not currently
486 // supported. This doesn't seem to occur often in GitHub repositories.
487 selected_rules_file.map(|path_in_worktree| {
488 let project_path = ProjectPath {
489 worktree_id,
490 path: path_in_worktree.clone(),
491 };
492 let buffer_task =
493 project.update(cx, |project, cx| project.open_buffer(project_path, cx));
494 let rope_task = cx.spawn(async move |cx| {
495 buffer_task.await?.read_with(cx, |buffer, cx| {
496 let project_entry_id = buffer.entry_id(cx).context("buffer has no file")?;
497 anyhow::Ok((project_entry_id, buffer.as_rope().clone()))
498 })?
499 });
500 // Build a string from the rope on a background thread.
501 cx.background_spawn(async move {
502 let (project_entry_id, rope) = rope_task.await?;
503 anyhow::Ok(RulesFileContext {
504 path_in_worktree,
505 text: rope.to_string().trim().to_string(),
506 project_entry_id: project_entry_id.to_usize(),
507 })
508 })
509 })
510 }
511
512 fn handle_thread_title_updated(
513 &mut self,
514 thread: Entity<Thread>,
515 _: &TitleUpdated,
516 cx: &mut Context<Self>,
517 ) {
518 let session_id = thread.read(cx).id();
519 let Some(session) = self.sessions.get(session_id) else {
520 return;
521 };
522 let thread = thread.downgrade();
523 let acp_thread = session.acp_thread.clone();
524 cx.spawn(async move |_, cx| {
525 let title = thread.read_with(cx, |thread, _| thread.title())?;
526 let task = acp_thread.update(cx, |acp_thread, cx| acp_thread.set_title(title, cx))?;
527 task.await
528 })
529 .detach_and_log_err(cx);
530 }
531
532 fn handle_thread_token_usage_updated(
533 &mut self,
534 thread: Entity<Thread>,
535 usage: &TokenUsageUpdated,
536 cx: &mut Context<Self>,
537 ) {
538 let Some(session) = self.sessions.get(thread.read(cx).id()) else {
539 return;
540 };
541 session
542 .acp_thread
543 .update(cx, |acp_thread, cx| {
544 acp_thread.update_token_usage(usage.0.clone(), cx);
545 })
546 .ok();
547 }
548
549 fn handle_project_event(
550 &mut self,
551 _project: Entity<Project>,
552 event: &project::Event,
553 _cx: &mut Context<Self>,
554 ) {
555 match event {
556 project::Event::WorktreeAdded(_) | project::Event::WorktreeRemoved(_) => {
557 self.project_context_needs_refresh.send(()).ok();
558 }
559 project::Event::WorktreeUpdatedEntries(_, items) => {
560 if items.iter().any(|(path, _, _)| {
561 RULES_FILE_NAMES
562 .iter()
563 .any(|name| path.as_ref() == Path::new(name))
564 }) {
565 self.project_context_needs_refresh.send(()).ok();
566 }
567 }
568 _ => {}
569 }
570 }
571
572 fn handle_prompts_updated_event(
573 &mut self,
574 _prompt_store: Entity<PromptStore>,
575 _event: &prompt_store::PromptsUpdatedEvent,
576 _cx: &mut Context<Self>,
577 ) {
578 self.project_context_needs_refresh.send(()).ok();
579 }
580
581 fn handle_models_updated_event(
582 &mut self,
583 _registry: Entity<LanguageModelRegistry>,
584 _event: &language_model::Event,
585 cx: &mut Context<Self>,
586 ) {
587 self.models.refresh_list(cx);
588
589 let registry = LanguageModelRegistry::read_global(cx);
590 let default_model = registry.default_model().map(|m| m.model);
591 let summarization_model = registry.thread_summary_model().map(|m| m.model);
592
593 for session in self.sessions.values_mut() {
594 session.thread.update(cx, |thread, cx| {
595 if thread.model().is_none()
596 && let Some(model) = default_model.clone()
597 {
598 thread.set_model(model, cx);
599 cx.notify();
600 }
601 thread.set_summarization_model(summarization_model.clone(), cx);
602 });
603 }
604 }
605
606 pub fn open_thread(
607 &mut self,
608 id: acp::SessionId,
609 cx: &mut Context<Self>,
610 ) -> Task<Result<Entity<AcpThread>>> {
611 let database_future = ThreadsDatabase::connect(cx);
612 cx.spawn(async move |this, cx| {
613 let database = database_future.await.map_err(|err| anyhow!(err))?;
614 let db_thread = database
615 .load_thread(id.clone())
616 .await?
617 .with_context(|| format!("no thread found with ID: {id:?}"))?;
618
619 let thread = this.update(cx, |this, cx| {
620 let action_log = cx.new(|_cx| ActionLog::new(this.project.clone()));
621 cx.new(|cx| {
622 Thread::from_db(
623 id.clone(),
624 db_thread,
625 this.project.clone(),
626 this.project_context.clone(),
627 this.context_server_registry.clone(),
628 action_log.clone(),
629 this.templates.clone(),
630 cx,
631 )
632 })
633 })?;
634 let acp_thread =
635 this.update(cx, |this, cx| this.register_session(thread.clone(), cx))?;
636 let events = thread.update(cx, |thread, cx| thread.replay(cx))?;
637 cx.update(|cx| {
638 NativeAgentConnection::handle_thread_events(events, acp_thread.downgrade(), cx)
639 })?
640 .await?;
641 Ok(acp_thread)
642 })
643 }
644
645 pub fn thread_summary(
646 &mut self,
647 id: acp::SessionId,
648 cx: &mut Context<Self>,
649 ) -> Task<Result<SharedString>> {
650 let thread = self.open_thread(id.clone(), cx);
651 cx.spawn(async move |this, cx| {
652 let acp_thread = thread.await?;
653 let result = this
654 .update(cx, |this, cx| {
655 this.sessions
656 .get(&id)
657 .unwrap()
658 .thread
659 .update(cx, |thread, cx| thread.summary(cx))
660 })?
661 .await?;
662 drop(acp_thread);
663 Ok(result)
664 })
665 }
666
667 fn save_thread(&mut self, thread: Entity<Thread>, cx: &mut Context<Self>) {
668 if thread.read(cx).is_empty() {
669 return;
670 }
671
672 let database_future = ThreadsDatabase::connect(cx);
673 let (id, db_thread) =
674 thread.update(cx, |thread, cx| (thread.id().clone(), thread.to_db(cx)));
675 let Some(session) = self.sessions.get_mut(&id) else {
676 return;
677 };
678 let history = self.history.clone();
679 session.pending_save = cx.spawn(async move |_, cx| {
680 let Some(database) = database_future.await.map_err(|err| anyhow!(err)).log_err() else {
681 return;
682 };
683 let db_thread = db_thread.await;
684 database.save_thread(id, db_thread).await.log_err();
685 history.update(cx, |history, cx| history.reload(cx)).ok();
686 });
687 }
688}
689
690/// Wrapper struct that implements the AgentConnection trait
691#[derive(Clone)]
692pub struct NativeAgentConnection(pub Entity<NativeAgent>);
693
694impl NativeAgentConnection {
695 pub fn thread(&self, session_id: &acp::SessionId, cx: &App) -> Option<Entity<Thread>> {
696 self.0
697 .read(cx)
698 .sessions
699 .get(session_id)
700 .map(|session| session.thread.clone())
701 }
702
703 fn run_turn(
704 &self,
705 session_id: acp::SessionId,
706 cx: &mut App,
707 f: impl 'static
708 + FnOnce(Entity<Thread>, &mut App) -> Result<mpsc::UnboundedReceiver<Result<ThreadEvent>>>,
709 ) -> Task<Result<acp::PromptResponse>> {
710 let Some((thread, acp_thread)) = self.0.update(cx, |agent, _cx| {
711 agent
712 .sessions
713 .get_mut(&session_id)
714 .map(|s| (s.thread.clone(), s.acp_thread.clone()))
715 }) else {
716 return Task::ready(Err(anyhow!("Session not found")));
717 };
718 log::debug!("Found session for: {}", session_id);
719
720 let response_stream = match f(thread, cx) {
721 Ok(stream) => stream,
722 Err(err) => return Task::ready(Err(err)),
723 };
724 Self::handle_thread_events(response_stream, acp_thread, cx)
725 }
726
727 fn handle_thread_events(
728 mut events: mpsc::UnboundedReceiver<Result<ThreadEvent>>,
729 acp_thread: WeakEntity<AcpThread>,
730 cx: &App,
731 ) -> Task<Result<acp::PromptResponse>> {
732 cx.spawn(async move |cx| {
733 // Handle response stream and forward to session.acp_thread
734 while let Some(result) = events.next().await {
735 match result {
736 Ok(event) => {
737 log::trace!("Received completion event: {:?}", event);
738
739 match event {
740 ThreadEvent::UserMessage(message) => {
741 acp_thread.update(cx, |thread, cx| {
742 for content in message.content {
743 thread.push_user_content_block(
744 Some(message.id.clone()),
745 content.into(),
746 cx,
747 );
748 }
749 })?;
750 }
751 ThreadEvent::AgentText(text) => {
752 acp_thread.update(cx, |thread, cx| {
753 thread.push_assistant_content_block(
754 acp::ContentBlock::Text(acp::TextContent {
755 text,
756 annotations: None,
757 meta: None,
758 }),
759 false,
760 cx,
761 )
762 })?;
763 }
764 ThreadEvent::AgentThinking(text) => {
765 acp_thread.update(cx, |thread, cx| {
766 thread.push_assistant_content_block(
767 acp::ContentBlock::Text(acp::TextContent {
768 text,
769 annotations: None,
770 meta: None,
771 }),
772 true,
773 cx,
774 )
775 })?;
776 }
777 ThreadEvent::ToolCallAuthorization(ToolCallAuthorization {
778 tool_call,
779 options,
780 response,
781 }) => {
782 let outcome_task = acp_thread.update(cx, |thread, cx| {
783 thread.request_tool_call_authorization(
784 tool_call, options, true, cx,
785 )
786 })??;
787 cx.background_spawn(async move {
788 if let acp::RequestPermissionOutcome::Selected { option_id } =
789 outcome_task.await
790 {
791 response
792 .send(option_id)
793 .map(|_| anyhow!("authorization receiver was dropped"))
794 .log_err();
795 }
796 })
797 .detach();
798 }
799 ThreadEvent::ToolCall(tool_call) => {
800 acp_thread.update(cx, |thread, cx| {
801 thread.upsert_tool_call(tool_call, cx)
802 })??;
803 }
804 ThreadEvent::ToolCallUpdate(update) => {
805 acp_thread.update(cx, |thread, cx| {
806 thread.update_tool_call(update, cx)
807 })??;
808 }
809 ThreadEvent::Retry(status) => {
810 acp_thread.update(cx, |thread, cx| {
811 thread.update_retry_status(status, cx)
812 })?;
813 }
814 ThreadEvent::Stop(stop_reason) => {
815 log::debug!("Assistant message complete: {:?}", stop_reason);
816 return Ok(acp::PromptResponse {
817 stop_reason,
818 meta: None,
819 });
820 }
821 }
822 }
823 Err(e) => {
824 log::error!("Error in model response stream: {:?}", e);
825 return Err(e);
826 }
827 }
828 }
829
830 log::debug!("Response stream completed");
831 anyhow::Ok(acp::PromptResponse {
832 stop_reason: acp::StopReason::EndTurn,
833 meta: None,
834 })
835 })
836 }
837}
838
839impl AgentModelSelector for NativeAgentConnection {
840 fn list_models(&self, cx: &mut App) -> Task<Result<acp_thread::AgentModelList>> {
841 log::debug!("NativeAgentConnection::list_models called");
842 let list = self.0.read(cx).models.model_list.clone();
843 Task::ready(if list.is_empty() {
844 Err(anyhow::anyhow!("No models available"))
845 } else {
846 Ok(list)
847 })
848 }
849
850 fn select_model(
851 &self,
852 session_id: acp::SessionId,
853 model_id: acp_thread::AgentModelId,
854 cx: &mut App,
855 ) -> Task<Result<()>> {
856 log::debug!("Setting model for session {}: {}", session_id, model_id);
857 let Some(thread) = self
858 .0
859 .read(cx)
860 .sessions
861 .get(&session_id)
862 .map(|session| session.thread.clone())
863 else {
864 return Task::ready(Err(anyhow!("Session not found")));
865 };
866
867 let Some(model) = self.0.read(cx).models.model_from_id(&model_id) else {
868 return Task::ready(Err(anyhow!("Invalid model ID {}", model_id)));
869 };
870
871 thread.update(cx, |thread, cx| {
872 thread.set_model(model.clone(), cx);
873 });
874
875 update_settings_file(self.0.read(cx).fs.clone(), cx, move |settings, _cx| {
876 let provider = model.provider_id().0.to_string();
877 let model = model.id().0.to_string();
878 settings
879 .agent
880 .get_or_insert_default()
881 .set_model(LanguageModelSelection {
882 provider: provider.into(),
883 model,
884 });
885 });
886
887 Task::ready(Ok(()))
888 }
889
890 fn selected_model(
891 &self,
892 session_id: &acp::SessionId,
893 cx: &mut App,
894 ) -> Task<Result<acp_thread::AgentModelInfo>> {
895 let session_id = session_id.clone();
896
897 let Some(thread) = self
898 .0
899 .read(cx)
900 .sessions
901 .get(&session_id)
902 .map(|session| session.thread.clone())
903 else {
904 return Task::ready(Err(anyhow!("Session not found")));
905 };
906 let Some(model) = thread.read(cx).model() else {
907 return Task::ready(Err(anyhow!("Model not found")));
908 };
909 let Some(provider) = LanguageModelRegistry::read_global(cx).provider(&model.provider_id())
910 else {
911 return Task::ready(Err(anyhow!("Provider not found")));
912 };
913 Task::ready(Ok(LanguageModels::map_language_model_to_info(
914 model, &provider,
915 )))
916 }
917
918 fn watch(&self, cx: &mut App) -> watch::Receiver<()> {
919 self.0.read(cx).models.watch()
920 }
921}
922
923impl acp_thread::AgentConnection for NativeAgentConnection {
924 fn new_thread(
925 self: Rc<Self>,
926 project: Entity<Project>,
927 cwd: &Path,
928 cx: &mut App,
929 ) -> Task<Result<Entity<acp_thread::AcpThread>>> {
930 let agent = self.0.clone();
931 log::debug!("Creating new thread for project at: {:?}", cwd);
932
933 cx.spawn(async move |cx| {
934 log::debug!("Starting thread creation in async context");
935
936 // Create Thread
937 let thread = agent.update(
938 cx,
939 |agent, cx: &mut gpui::Context<NativeAgent>| -> Result<_> {
940 // Fetch default model from registry settings
941 let registry = LanguageModelRegistry::read_global(cx);
942 // Log available models for debugging
943 let available_count = registry.available_models(cx).count();
944 log::debug!("Total available models: {}", available_count);
945
946 let default_model = registry.default_model().and_then(|default_model| {
947 agent
948 .models
949 .model_from_id(&LanguageModels::model_id(&default_model.model))
950 });
951 Ok(cx.new(|cx| {
952 Thread::new(
953 project.clone(),
954 agent.project_context.clone(),
955 agent.context_server_registry.clone(),
956 agent.templates.clone(),
957 default_model,
958 cx,
959 )
960 }))
961 },
962 )??;
963 agent.update(cx, |agent, cx| agent.register_session(thread, cx))
964 })
965 }
966
967 fn auth_methods(&self) -> &[acp::AuthMethod] {
968 &[] // No auth for in-process
969 }
970
971 fn authenticate(&self, _method: acp::AuthMethodId, _cx: &mut App) -> Task<Result<()>> {
972 Task::ready(Ok(()))
973 }
974
975 fn model_selector(&self) -> Option<Rc<dyn AgentModelSelector>> {
976 Some(Rc::new(self.clone()) as Rc<dyn AgentModelSelector>)
977 }
978
979 fn prompt(
980 &self,
981 id: Option<acp_thread::UserMessageId>,
982 params: acp::PromptRequest,
983 cx: &mut App,
984 ) -> Task<Result<acp::PromptResponse>> {
985 let id = id.expect("UserMessageId is required");
986 let session_id = params.session_id.clone();
987 log::info!("Received prompt request for session: {}", session_id);
988 log::debug!("Prompt blocks count: {}", params.prompt.len());
989
990 self.run_turn(session_id, cx, |thread, cx| {
991 let content: Vec<UserMessageContent> = params
992 .prompt
993 .into_iter()
994 .map(Into::into)
995 .collect::<Vec<_>>();
996 log::debug!("Converted prompt to message: {} chars", content.len());
997 log::debug!("Message id: {:?}", id);
998 log::debug!("Message content: {:?}", content);
999
1000 thread.update(cx, |thread, cx| thread.send(id, content, cx))
1001 })
1002 }
1003
1004 fn resume(
1005 &self,
1006 session_id: &acp::SessionId,
1007 _cx: &App,
1008 ) -> Option<Rc<dyn acp_thread::AgentSessionResume>> {
1009 Some(Rc::new(NativeAgentSessionResume {
1010 connection: self.clone(),
1011 session_id: session_id.clone(),
1012 }) as _)
1013 }
1014
1015 fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
1016 log::info!("Cancelling on session: {}", session_id);
1017 self.0.update(cx, |agent, cx| {
1018 if let Some(agent) = agent.sessions.get(session_id) {
1019 agent.thread.update(cx, |thread, cx| thread.cancel(cx));
1020 }
1021 });
1022 }
1023
1024 fn truncate(
1025 &self,
1026 session_id: &agent_client_protocol::SessionId,
1027 cx: &App,
1028 ) -> Option<Rc<dyn acp_thread::AgentSessionTruncate>> {
1029 self.0.read_with(cx, |agent, _cx| {
1030 agent.sessions.get(session_id).map(|session| {
1031 Rc::new(NativeAgentSessionTruncate {
1032 thread: session.thread.clone(),
1033 acp_thread: session.acp_thread.clone(),
1034 }) as _
1035 })
1036 })
1037 }
1038
1039 fn set_title(
1040 &self,
1041 session_id: &acp::SessionId,
1042 _cx: &App,
1043 ) -> Option<Rc<dyn acp_thread::AgentSessionSetTitle>> {
1044 Some(Rc::new(NativeAgentSessionSetTitle {
1045 connection: self.clone(),
1046 session_id: session_id.clone(),
1047 }) as _)
1048 }
1049
1050 fn telemetry(&self) -> Option<Rc<dyn acp_thread::AgentTelemetry>> {
1051 Some(Rc::new(self.clone()) as Rc<dyn acp_thread::AgentTelemetry>)
1052 }
1053
1054 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
1055 self
1056 }
1057}
1058
1059impl acp_thread::AgentTelemetry for NativeAgentConnection {
1060 fn agent_name(&self) -> String {
1061 "Zed".into()
1062 }
1063
1064 fn thread_data(
1065 &self,
1066 session_id: &acp::SessionId,
1067 cx: &mut App,
1068 ) -> Task<Result<serde_json::Value>> {
1069 let Some(session) = self.0.read(cx).sessions.get(session_id) else {
1070 return Task::ready(Err(anyhow!("Session not found")));
1071 };
1072
1073 let task = session.thread.read(cx).to_db(cx);
1074 cx.background_spawn(async move {
1075 serde_json::to_value(task.await).context("Failed to serialize thread")
1076 })
1077 }
1078}
1079
1080struct NativeAgentSessionTruncate {
1081 thread: Entity<Thread>,
1082 acp_thread: WeakEntity<AcpThread>,
1083}
1084
1085impl acp_thread::AgentSessionTruncate for NativeAgentSessionTruncate {
1086 fn run(&self, message_id: acp_thread::UserMessageId, cx: &mut App) -> Task<Result<()>> {
1087 match self.thread.update(cx, |thread, cx| {
1088 thread.truncate(message_id.clone(), cx)?;
1089 Ok(thread.latest_token_usage())
1090 }) {
1091 Ok(usage) => {
1092 self.acp_thread
1093 .update(cx, |thread, cx| {
1094 thread.update_token_usage(usage, cx);
1095 })
1096 .ok();
1097 Task::ready(Ok(()))
1098 }
1099 Err(error) => Task::ready(Err(error)),
1100 }
1101 }
1102}
1103
1104struct NativeAgentSessionResume {
1105 connection: NativeAgentConnection,
1106 session_id: acp::SessionId,
1107}
1108
1109impl acp_thread::AgentSessionResume for NativeAgentSessionResume {
1110 fn run(&self, cx: &mut App) -> Task<Result<acp::PromptResponse>> {
1111 self.connection
1112 .run_turn(self.session_id.clone(), cx, |thread, cx| {
1113 thread.update(cx, |thread, cx| thread.resume(cx))
1114 })
1115 }
1116}
1117
1118struct NativeAgentSessionSetTitle {
1119 connection: NativeAgentConnection,
1120 session_id: acp::SessionId,
1121}
1122
1123impl acp_thread::AgentSessionSetTitle for NativeAgentSessionSetTitle {
1124 fn run(&self, title: SharedString, cx: &mut App) -> Task<Result<()>> {
1125 let Some(session) = self.connection.0.read(cx).sessions.get(&self.session_id) else {
1126 return Task::ready(Err(anyhow!("session not found")));
1127 };
1128 let thread = session.thread.clone();
1129 thread.update(cx, |thread, cx| thread.set_title(title, cx));
1130 Task::ready(Ok(()))
1131 }
1132}
1133
1134pub struct AcpThreadEnvironment {
1135 acp_thread: WeakEntity<AcpThread>,
1136}
1137
1138impl ThreadEnvironment for AcpThreadEnvironment {
1139 fn create_terminal(
1140 &self,
1141 command: String,
1142 cwd: Option<PathBuf>,
1143 output_byte_limit: Option<u64>,
1144 cx: &mut AsyncApp,
1145 ) -> Task<Result<Rc<dyn TerminalHandle>>> {
1146 let task = self.acp_thread.update(cx, |thread, cx| {
1147 thread.create_terminal(command, vec![], vec![], cwd, output_byte_limit, cx)
1148 });
1149
1150 let acp_thread = self.acp_thread.clone();
1151 cx.spawn(async move |cx| {
1152 let terminal = task?.await?;
1153
1154 let (drop_tx, drop_rx) = oneshot::channel();
1155 let terminal_id = terminal.read_with(cx, |terminal, _cx| terminal.id().clone())?;
1156
1157 cx.spawn(async move |cx| {
1158 drop_rx.await.ok();
1159 acp_thread.update(cx, |thread, cx| thread.release_terminal(terminal_id, cx))
1160 })
1161 .detach();
1162
1163 let handle = AcpTerminalHandle {
1164 terminal,
1165 _drop_tx: Some(drop_tx),
1166 };
1167
1168 Ok(Rc::new(handle) as _)
1169 })
1170 }
1171}
1172
1173pub struct AcpTerminalHandle {
1174 terminal: Entity<acp_thread::Terminal>,
1175 _drop_tx: Option<oneshot::Sender<()>>,
1176}
1177
1178impl TerminalHandle for AcpTerminalHandle {
1179 fn id(&self, cx: &AsyncApp) -> Result<acp::TerminalId> {
1180 self.terminal.read_with(cx, |term, _cx| term.id().clone())
1181 }
1182
1183 fn wait_for_exit(&self, cx: &AsyncApp) -> Result<Shared<Task<acp::TerminalExitStatus>>> {
1184 self.terminal
1185 .read_with(cx, |term, _cx| term.wait_for_exit())
1186 }
1187
1188 fn current_output(&self, cx: &AsyncApp) -> Result<acp::TerminalOutputResponse> {
1189 self.terminal
1190 .read_with(cx, |term, cx| term.current_output(cx))
1191 }
1192}
1193
1194#[cfg(test)]
1195mod tests {
1196 use crate::HistoryEntryId;
1197
1198 use super::*;
1199 use acp_thread::{
1200 AgentConnection, AgentModelGroupName, AgentModelId, AgentModelInfo, MentionUri,
1201 };
1202 use fs::FakeFs;
1203 use gpui::TestAppContext;
1204 use indoc::indoc;
1205 use language_model::fake_provider::FakeLanguageModel;
1206 use serde_json::json;
1207 use settings::SettingsStore;
1208 use util::path;
1209
1210 #[gpui::test]
1211 async fn test_maintaining_project_context(cx: &mut TestAppContext) {
1212 init_test(cx);
1213 let fs = FakeFs::new(cx.executor());
1214 fs.insert_tree(
1215 "/",
1216 json!({
1217 "a": {}
1218 }),
1219 )
1220 .await;
1221 let project = Project::test(fs.clone(), [], cx).await;
1222 let context_store = cx.new(|cx| assistant_context::ContextStore::fake(project.clone(), cx));
1223 let history_store = cx.new(|cx| HistoryStore::new(context_store, cx));
1224 let agent = NativeAgent::new(
1225 project.clone(),
1226 history_store,
1227 Templates::new(),
1228 None,
1229 fs.clone(),
1230 &mut cx.to_async(),
1231 )
1232 .await
1233 .unwrap();
1234 agent.read_with(cx, |agent, cx| {
1235 assert_eq!(agent.project_context.read(cx).worktrees, vec![])
1236 });
1237
1238 let worktree = project
1239 .update(cx, |project, cx| project.create_worktree("/a", true, cx))
1240 .await
1241 .unwrap();
1242 cx.run_until_parked();
1243 agent.read_with(cx, |agent, cx| {
1244 assert_eq!(
1245 agent.project_context.read(cx).worktrees,
1246 vec![WorktreeContext {
1247 root_name: "a".into(),
1248 abs_path: Path::new("/a").into(),
1249 rules_file: None
1250 }]
1251 )
1252 });
1253
1254 // Creating `/a/.rules` updates the project context.
1255 fs.insert_file("/a/.rules", Vec::new()).await;
1256 cx.run_until_parked();
1257 agent.read_with(cx, |agent, cx| {
1258 let rules_entry = worktree.read(cx).entry_for_path(".rules").unwrap();
1259 assert_eq!(
1260 agent.project_context.read(cx).worktrees,
1261 vec![WorktreeContext {
1262 root_name: "a".into(),
1263 abs_path: Path::new("/a").into(),
1264 rules_file: Some(RulesFileContext {
1265 path_in_worktree: Path::new(".rules").into(),
1266 text: "".into(),
1267 project_entry_id: rules_entry.id.to_usize()
1268 })
1269 }]
1270 )
1271 });
1272 }
1273
1274 #[gpui::test]
1275 async fn test_listing_models(cx: &mut TestAppContext) {
1276 init_test(cx);
1277 let fs = FakeFs::new(cx.executor());
1278 fs.insert_tree("/", json!({ "a": {} })).await;
1279 let project = Project::test(fs.clone(), [], cx).await;
1280 let context_store = cx.new(|cx| assistant_context::ContextStore::fake(project.clone(), cx));
1281 let history_store = cx.new(|cx| HistoryStore::new(context_store, cx));
1282 let connection = NativeAgentConnection(
1283 NativeAgent::new(
1284 project.clone(),
1285 history_store,
1286 Templates::new(),
1287 None,
1288 fs.clone(),
1289 &mut cx.to_async(),
1290 )
1291 .await
1292 .unwrap(),
1293 );
1294
1295 let models = cx.update(|cx| connection.list_models(cx)).await.unwrap();
1296
1297 let acp_thread::AgentModelList::Grouped(models) = models else {
1298 panic!("Unexpected model group");
1299 };
1300 assert_eq!(
1301 models,
1302 IndexMap::from_iter([(
1303 AgentModelGroupName("Fake".into()),
1304 vec![AgentModelInfo {
1305 id: AgentModelId("fake/fake".into()),
1306 name: "Fake".into(),
1307 icon: Some(ui::IconName::ZedAssistant),
1308 }]
1309 )])
1310 );
1311 }
1312
1313 #[gpui::test]
1314 async fn test_model_selection_persists_to_settings(cx: &mut TestAppContext) {
1315 init_test(cx);
1316 let fs = FakeFs::new(cx.executor());
1317 fs.create_dir(paths::settings_file().parent().unwrap())
1318 .await
1319 .unwrap();
1320 fs.insert_file(
1321 paths::settings_file(),
1322 json!({
1323 "agent": {
1324 "default_model": {
1325 "provider": "foo",
1326 "model": "bar"
1327 }
1328 }
1329 })
1330 .to_string()
1331 .into_bytes(),
1332 )
1333 .await;
1334 let project = Project::test(fs.clone(), [], cx).await;
1335
1336 let context_store = cx.new(|cx| assistant_context::ContextStore::fake(project.clone(), cx));
1337 let history_store = cx.new(|cx| HistoryStore::new(context_store, cx));
1338
1339 // Create the agent and connection
1340 let agent = NativeAgent::new(
1341 project.clone(),
1342 history_store,
1343 Templates::new(),
1344 None,
1345 fs.clone(),
1346 &mut cx.to_async(),
1347 )
1348 .await
1349 .unwrap();
1350 let connection = NativeAgentConnection(agent.clone());
1351
1352 // Create a thread/session
1353 let acp_thread = cx
1354 .update(|cx| {
1355 Rc::new(connection.clone()).new_thread(project.clone(), Path::new("/a"), cx)
1356 })
1357 .await
1358 .unwrap();
1359
1360 let session_id = cx.update(|cx| acp_thread.read(cx).session_id().clone());
1361
1362 // Select a model
1363 let model_id = AgentModelId("fake/fake".into());
1364 cx.update(|cx| connection.select_model(session_id.clone(), model_id.clone(), cx))
1365 .await
1366 .unwrap();
1367
1368 // Verify the thread has the selected model
1369 agent.read_with(cx, |agent, _| {
1370 let session = agent.sessions.get(&session_id).unwrap();
1371 session.thread.read_with(cx, |thread, _| {
1372 assert_eq!(thread.model().unwrap().id().0, "fake");
1373 });
1374 });
1375
1376 cx.run_until_parked();
1377
1378 // Verify settings file was updated
1379 let settings_content = fs.load(paths::settings_file()).await.unwrap();
1380 let settings_json: serde_json::Value = serde_json::from_str(&settings_content).unwrap();
1381
1382 // Check that the agent settings contain the selected model
1383 assert_eq!(
1384 settings_json["agent"]["default_model"]["model"],
1385 json!("fake")
1386 );
1387 assert_eq!(
1388 settings_json["agent"]["default_model"]["provider"],
1389 json!("fake")
1390 );
1391 }
1392
1393 #[gpui::test]
1394 #[cfg_attr(target_os = "windows", ignore)] // TODO: Fix this test on Windows
1395 async fn test_save_load_thread(cx: &mut TestAppContext) {
1396 init_test(cx);
1397 let fs = FakeFs::new(cx.executor());
1398 fs.insert_tree(
1399 "/",
1400 json!({
1401 "a": {
1402 "b.md": "Lorem"
1403 }
1404 }),
1405 )
1406 .await;
1407 let project = Project::test(fs.clone(), [path!("/a").as_ref()], cx).await;
1408 let context_store = cx.new(|cx| assistant_context::ContextStore::fake(project.clone(), cx));
1409 let history_store = cx.new(|cx| HistoryStore::new(context_store, cx));
1410 let agent = NativeAgent::new(
1411 project.clone(),
1412 history_store.clone(),
1413 Templates::new(),
1414 None,
1415 fs.clone(),
1416 &mut cx.to_async(),
1417 )
1418 .await
1419 .unwrap();
1420 let connection = Rc::new(NativeAgentConnection(agent.clone()));
1421
1422 let acp_thread = cx
1423 .update(|cx| {
1424 connection
1425 .clone()
1426 .new_thread(project.clone(), Path::new(""), cx)
1427 })
1428 .await
1429 .unwrap();
1430 let session_id = acp_thread.read_with(cx, |thread, _| thread.session_id().clone());
1431 let thread = agent.read_with(cx, |agent, _| {
1432 agent.sessions.get(&session_id).unwrap().thread.clone()
1433 });
1434
1435 // Ensure empty threads are not saved, even if they get mutated.
1436 let model = Arc::new(FakeLanguageModel::default());
1437 let summary_model = Arc::new(FakeLanguageModel::default());
1438 thread.update(cx, |thread, cx| {
1439 thread.set_model(model.clone(), cx);
1440 thread.set_summarization_model(Some(summary_model.clone()), cx);
1441 });
1442 cx.run_until_parked();
1443 assert_eq!(history_entries(&history_store, cx), vec![]);
1444
1445 let send = acp_thread.update(cx, |thread, cx| {
1446 thread.send(
1447 vec![
1448 "What does ".into(),
1449 acp::ContentBlock::ResourceLink(acp::ResourceLink {
1450 name: "b.md".into(),
1451 uri: MentionUri::File {
1452 abs_path: path!("/a/b.md").into(),
1453 }
1454 .to_uri()
1455 .to_string(),
1456 annotations: None,
1457 description: None,
1458 mime_type: None,
1459 size: None,
1460 title: None,
1461 meta: None,
1462 }),
1463 " mean?".into(),
1464 ],
1465 cx,
1466 )
1467 });
1468 let send = cx.foreground_executor().spawn(send);
1469 cx.run_until_parked();
1470
1471 model.send_last_completion_stream_text_chunk("Lorem.");
1472 model.end_last_completion_stream();
1473 cx.run_until_parked();
1474 summary_model.send_last_completion_stream_text_chunk("Explaining /a/b.md");
1475 summary_model.end_last_completion_stream();
1476
1477 send.await.unwrap();
1478 acp_thread.read_with(cx, |thread, cx| {
1479 assert_eq!(
1480 thread.to_markdown(cx),
1481 indoc! {"
1482 ## User
1483
1484 What does [@b.md](file:///a/b.md) mean?
1485
1486 ## Assistant
1487
1488 Lorem.
1489
1490 "}
1491 )
1492 });
1493
1494 cx.run_until_parked();
1495
1496 // Drop the ACP thread, which should cause the session to be dropped as well.
1497 cx.update(|_| {
1498 drop(thread);
1499 drop(acp_thread);
1500 });
1501 agent.read_with(cx, |agent, _| {
1502 assert_eq!(agent.sessions.keys().cloned().collect::<Vec<_>>(), []);
1503 });
1504
1505 // Ensure the thread can be reloaded from disk.
1506 assert_eq!(
1507 history_entries(&history_store, cx),
1508 vec![(
1509 HistoryEntryId::AcpThread(session_id.clone()),
1510 "Explaining /a/b.md".into()
1511 )]
1512 );
1513 let acp_thread = agent
1514 .update(cx, |agent, cx| agent.open_thread(session_id.clone(), cx))
1515 .await
1516 .unwrap();
1517 acp_thread.read_with(cx, |thread, cx| {
1518 assert_eq!(
1519 thread.to_markdown(cx),
1520 indoc! {"
1521 ## User
1522
1523 What does [@b.md](file:///a/b.md) mean?
1524
1525 ## Assistant
1526
1527 Lorem.
1528
1529 "}
1530 )
1531 });
1532 }
1533
1534 fn history_entries(
1535 history: &Entity<HistoryStore>,
1536 cx: &mut TestAppContext,
1537 ) -> Vec<(HistoryEntryId, String)> {
1538 history.read_with(cx, |history, _| {
1539 history
1540 .entries()
1541 .map(|e| (e.id(), e.title().to_string()))
1542 .collect::<Vec<_>>()
1543 })
1544 }
1545
1546 fn init_test(cx: &mut TestAppContext) {
1547 env_logger::try_init().ok();
1548 cx.update(|cx| {
1549 let settings_store = SettingsStore::test(cx);
1550 cx.set_global(settings_store);
1551 Project::init_settings(cx);
1552 agent_settings::init(cx);
1553 language::init(cx);
1554 LanguageModelRegistry::test(cx);
1555 });
1556 }
1557}