1use std::io::Write;
2use std::sync::Arc;
3
4use anyhow::{Context as _, Result};
5use assistant_tool::{ActionLog, ToolWorkingSet};
6use chrono::{DateTime, Utc};
7use collections::{BTreeMap, HashMap, HashSet};
8use futures::future::Shared;
9use futures::{FutureExt, StreamExt as _};
10use git;
11use gpui::{App, AppContext, Context, Entity, EventEmitter, SharedString, Task};
12use language_model::{
13 LanguageModel, LanguageModelCompletionEvent, LanguageModelRegistry, LanguageModelRequest,
14 LanguageModelRequestMessage, LanguageModelRequestTool, LanguageModelToolResult,
15 LanguageModelToolUseId, MaxMonthlySpendReachedError, MessageContent, PaymentRequiredError,
16 Role, StopReason, TokenUsage,
17};
18use project::Project;
19use prompt_store::{AssistantSystemPromptWorktree, PromptBuilder};
20use scripting_tool::{ScriptingSession, ScriptingTool};
21use serde::{Deserialize, Serialize};
22use util::{post_inc, ResultExt, TryFutureExt as _};
23use uuid::Uuid;
24
25use crate::context::{attach_context_to_message, ContextId, ContextSnapshot};
26use crate::thread_store::{
27 SerializedMessage, SerializedThread, SerializedToolResult, SerializedToolUse,
28};
29use crate::tool_use::{PendingToolUse, ToolUse, ToolUseState};
30
31#[derive(Debug, Clone, Copy)]
32pub enum RequestKind {
33 Chat,
34 /// Used when summarizing a thread.
35 Summarize,
36}
37
38#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
39pub struct ThreadId(Arc<str>);
40
41impl ThreadId {
42 pub fn new() -> Self {
43 Self(Uuid::new_v4().to_string().into())
44 }
45}
46
47impl std::fmt::Display for ThreadId {
48 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49 write!(f, "{}", self.0)
50 }
51}
52
53#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy, Serialize, Deserialize)]
54pub struct MessageId(pub(crate) usize);
55
56impl MessageId {
57 fn post_inc(&mut self) -> Self {
58 Self(post_inc(&mut self.0))
59 }
60}
61
62/// A message in a [`Thread`].
63#[derive(Debug, Clone)]
64pub struct Message {
65 pub id: MessageId,
66 pub role: Role,
67 pub text: String,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct ProjectSnapshot {
72 pub worktree_snapshots: Vec<WorktreeSnapshot>,
73 pub unsaved_buffer_paths: Vec<String>,
74 pub timestamp: DateTime<Utc>,
75}
76
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct WorktreeSnapshot {
79 pub worktree_path: String,
80 pub git_state: Option<GitState>,
81}
82
83#[derive(Debug, Clone, Serialize, Deserialize)]
84pub struct GitState {
85 pub remote_url: Option<String>,
86 pub head_sha: Option<String>,
87 pub current_branch: Option<String>,
88 pub diff: Option<String>,
89}
90
91/// A thread of conversation with the LLM.
92pub struct Thread {
93 id: ThreadId,
94 updated_at: DateTime<Utc>,
95 summary: Option<SharedString>,
96 pending_summary: Task<Option<()>>,
97 messages: Vec<Message>,
98 next_message_id: MessageId,
99 context: BTreeMap<ContextId, ContextSnapshot>,
100 context_by_message: HashMap<MessageId, Vec<ContextId>>,
101 completion_count: usize,
102 pending_completions: Vec<PendingCompletion>,
103 project: Entity<Project>,
104 prompt_builder: Arc<PromptBuilder>,
105 tools: Arc<ToolWorkingSet>,
106 tool_use: ToolUseState,
107 action_log: Entity<ActionLog>,
108 scripting_session: Entity<ScriptingSession>,
109 scripting_tool_use: ToolUseState,
110 initial_project_snapshot: Shared<Task<Option<Arc<ProjectSnapshot>>>>,
111 cumulative_token_usage: TokenUsage,
112}
113
114impl Thread {
115 pub fn new(
116 project: Entity<Project>,
117 tools: Arc<ToolWorkingSet>,
118 prompt_builder: Arc<PromptBuilder>,
119 cx: &mut Context<Self>,
120 ) -> Self {
121 Self {
122 id: ThreadId::new(),
123 updated_at: Utc::now(),
124 summary: None,
125 pending_summary: Task::ready(None),
126 messages: Vec::new(),
127 next_message_id: MessageId(0),
128 context: BTreeMap::default(),
129 context_by_message: HashMap::default(),
130 completion_count: 0,
131 pending_completions: Vec::new(),
132 project: project.clone(),
133 prompt_builder,
134 tools,
135 tool_use: ToolUseState::new(),
136 scripting_session: cx.new(|cx| ScriptingSession::new(project.clone(), cx)),
137 scripting_tool_use: ToolUseState::new(),
138 action_log: cx.new(|_| ActionLog::new()),
139 initial_project_snapshot: {
140 let project_snapshot = Self::project_snapshot(project, cx);
141 cx.foreground_executor()
142 .spawn(async move { Some(project_snapshot.await) })
143 .shared()
144 },
145 cumulative_token_usage: TokenUsage::default(),
146 }
147 }
148
149 pub fn deserialize(
150 id: ThreadId,
151 serialized: SerializedThread,
152 project: Entity<Project>,
153 tools: Arc<ToolWorkingSet>,
154 prompt_builder: Arc<PromptBuilder>,
155 cx: &mut Context<Self>,
156 ) -> Self {
157 let next_message_id = MessageId(
158 serialized
159 .messages
160 .last()
161 .map(|message| message.id.0 + 1)
162 .unwrap_or(0),
163 );
164 let tool_use = ToolUseState::from_serialized_messages(&serialized.messages, |name| {
165 name != ScriptingTool::NAME
166 });
167 let scripting_tool_use =
168 ToolUseState::from_serialized_messages(&serialized.messages, |name| {
169 name == ScriptingTool::NAME
170 });
171 let scripting_session = cx.new(|cx| ScriptingSession::new(project.clone(), cx));
172
173 Self {
174 id,
175 updated_at: serialized.updated_at,
176 summary: Some(serialized.summary),
177 pending_summary: Task::ready(None),
178 messages: serialized
179 .messages
180 .into_iter()
181 .map(|message| Message {
182 id: message.id,
183 role: message.role,
184 text: message.text,
185 })
186 .collect(),
187 next_message_id,
188 context: BTreeMap::default(),
189 context_by_message: HashMap::default(),
190 completion_count: 0,
191 pending_completions: Vec::new(),
192 project,
193 prompt_builder,
194 tools,
195 tool_use,
196 action_log: cx.new(|_| ActionLog::new()),
197 scripting_session,
198 scripting_tool_use,
199 initial_project_snapshot: Task::ready(serialized.initial_project_snapshot).shared(),
200 // TODO: persist token usage?
201 cumulative_token_usage: TokenUsage::default(),
202 }
203 }
204
205 pub fn id(&self) -> &ThreadId {
206 &self.id
207 }
208
209 pub fn is_empty(&self) -> bool {
210 self.messages.is_empty()
211 }
212
213 pub fn updated_at(&self) -> DateTime<Utc> {
214 self.updated_at
215 }
216
217 pub fn touch_updated_at(&mut self) {
218 self.updated_at = Utc::now();
219 }
220
221 pub fn summary(&self) -> Option<SharedString> {
222 self.summary.clone()
223 }
224
225 pub fn summary_or_default(&self) -> SharedString {
226 const DEFAULT: SharedString = SharedString::new_static("New Thread");
227 self.summary.clone().unwrap_or(DEFAULT)
228 }
229
230 pub fn set_summary(&mut self, summary: impl Into<SharedString>, cx: &mut Context<Self>) {
231 self.summary = Some(summary.into());
232 cx.emit(ThreadEvent::SummaryChanged);
233 }
234
235 pub fn message(&self, id: MessageId) -> Option<&Message> {
236 self.messages.iter().find(|message| message.id == id)
237 }
238
239 pub fn messages(&self) -> impl Iterator<Item = &Message> {
240 self.messages.iter()
241 }
242
243 pub fn is_generating(&self) -> bool {
244 !self.pending_completions.is_empty() || !self.all_tools_finished()
245 }
246
247 pub fn tools(&self) -> &Arc<ToolWorkingSet> {
248 &self.tools
249 }
250
251 pub fn context_for_message(&self, id: MessageId) -> Option<Vec<ContextSnapshot>> {
252 let context = self.context_by_message.get(&id)?;
253 Some(
254 context
255 .into_iter()
256 .filter_map(|context_id| self.context.get(&context_id))
257 .cloned()
258 .collect::<Vec<_>>(),
259 )
260 }
261
262 /// Returns whether all of the tool uses have finished running.
263 pub fn all_tools_finished(&self) -> bool {
264 let mut all_pending_tool_uses = self
265 .tool_use
266 .pending_tool_uses()
267 .into_iter()
268 .chain(self.scripting_tool_use.pending_tool_uses());
269
270 // If the only pending tool uses left are the ones with errors, then
271 // that means that we've finished running all of the pending tools.
272 all_pending_tool_uses.all(|tool_use| tool_use.status.is_error())
273 }
274
275 pub fn tool_uses_for_message(&self, id: MessageId) -> Vec<ToolUse> {
276 self.tool_use.tool_uses_for_message(id)
277 }
278
279 pub fn scripting_tool_uses_for_message(&self, id: MessageId) -> Vec<ToolUse> {
280 self.scripting_tool_use.tool_uses_for_message(id)
281 }
282
283 pub fn tool_results_for_message(&self, id: MessageId) -> Vec<&LanguageModelToolResult> {
284 self.tool_use.tool_results_for_message(id)
285 }
286
287 pub fn tool_result(&self, id: &LanguageModelToolUseId) -> Option<&LanguageModelToolResult> {
288 self.tool_use.tool_result(id)
289 }
290
291 pub fn scripting_tool_results_for_message(
292 &self,
293 id: MessageId,
294 ) -> Vec<&LanguageModelToolResult> {
295 self.scripting_tool_use.tool_results_for_message(id)
296 }
297
298 pub fn scripting_changed_buffers<'a>(
299 &self,
300 cx: &'a App,
301 ) -> impl ExactSizeIterator<Item = &'a Entity<language::Buffer>> {
302 self.scripting_session.read(cx).changed_buffers()
303 }
304
305 pub fn message_has_tool_results(&self, message_id: MessageId) -> bool {
306 self.tool_use.message_has_tool_results(message_id)
307 }
308
309 pub fn message_has_scripting_tool_results(&self, message_id: MessageId) -> bool {
310 self.scripting_tool_use.message_has_tool_results(message_id)
311 }
312
313 pub fn insert_user_message(
314 &mut self,
315 text: impl Into<String>,
316 context: Vec<ContextSnapshot>,
317 cx: &mut Context<Self>,
318 ) -> MessageId {
319 let message_id = self.insert_message(Role::User, text, cx);
320 let context_ids = context.iter().map(|context| context.id).collect::<Vec<_>>();
321 self.context
322 .extend(context.into_iter().map(|context| (context.id, context)));
323 self.context_by_message.insert(message_id, context_ids);
324 message_id
325 }
326
327 pub fn insert_message(
328 &mut self,
329 role: Role,
330 text: impl Into<String>,
331 cx: &mut Context<Self>,
332 ) -> MessageId {
333 let id = self.next_message_id.post_inc();
334 self.messages.push(Message {
335 id,
336 role,
337 text: text.into(),
338 });
339 self.touch_updated_at();
340 cx.emit(ThreadEvent::MessageAdded(id));
341 id
342 }
343
344 pub fn edit_message(
345 &mut self,
346 id: MessageId,
347 new_role: Role,
348 new_text: String,
349 cx: &mut Context<Self>,
350 ) -> bool {
351 let Some(message) = self.messages.iter_mut().find(|message| message.id == id) else {
352 return false;
353 };
354 message.role = new_role;
355 message.text = new_text;
356 self.touch_updated_at();
357 cx.emit(ThreadEvent::MessageEdited(id));
358 true
359 }
360
361 pub fn delete_message(&mut self, id: MessageId, cx: &mut Context<Self>) -> bool {
362 let Some(index) = self.messages.iter().position(|message| message.id == id) else {
363 return false;
364 };
365 self.messages.remove(index);
366 self.context_by_message.remove(&id);
367 self.touch_updated_at();
368 cx.emit(ThreadEvent::MessageDeleted(id));
369 true
370 }
371
372 /// Returns the representation of this [`Thread`] in a textual form.
373 ///
374 /// This is the representation we use when attaching a thread as context to another thread.
375 pub fn text(&self) -> String {
376 let mut text = String::new();
377
378 for message in &self.messages {
379 text.push_str(match message.role {
380 language_model::Role::User => "User:",
381 language_model::Role::Assistant => "Assistant:",
382 language_model::Role::System => "System:",
383 });
384 text.push('\n');
385
386 text.push_str(&message.text);
387 text.push('\n');
388 }
389
390 text
391 }
392
393 /// Serializes this thread into a format for storage or telemetry.
394 pub fn serialize(&self, cx: &mut Context<Self>) -> Task<Result<SerializedThread>> {
395 let initial_project_snapshot = self.initial_project_snapshot.clone();
396 cx.spawn(|this, cx| async move {
397 let initial_project_snapshot = initial_project_snapshot.await;
398 this.read_with(&cx, |this, _| SerializedThread {
399 summary: this.summary_or_default(),
400 updated_at: this.updated_at(),
401 messages: this
402 .messages()
403 .map(|message| SerializedMessage {
404 id: message.id,
405 role: message.role,
406 text: message.text.clone(),
407 tool_uses: this
408 .tool_uses_for_message(message.id)
409 .into_iter()
410 .chain(this.scripting_tool_uses_for_message(message.id))
411 .map(|tool_use| SerializedToolUse {
412 id: tool_use.id,
413 name: tool_use.name,
414 input: tool_use.input,
415 })
416 .collect(),
417 tool_results: this
418 .tool_results_for_message(message.id)
419 .into_iter()
420 .chain(this.scripting_tool_results_for_message(message.id))
421 .map(|tool_result| SerializedToolResult {
422 tool_use_id: tool_result.tool_use_id.clone(),
423 is_error: tool_result.is_error,
424 content: tool_result.content.clone(),
425 })
426 .collect(),
427 })
428 .collect(),
429 initial_project_snapshot,
430 })
431 })
432 }
433
434 pub fn send_to_model(
435 &mut self,
436 model: Arc<dyn LanguageModel>,
437 request_kind: RequestKind,
438 cx: &mut Context<Self>,
439 ) {
440 let mut request = self.to_completion_request(request_kind, cx);
441 request.tools = {
442 let mut tools = Vec::new();
443
444 if self.tools.is_scripting_tool_enabled() {
445 tools.push(LanguageModelRequestTool {
446 name: ScriptingTool::NAME.into(),
447 description: ScriptingTool::DESCRIPTION.into(),
448 input_schema: ScriptingTool::input_schema(),
449 });
450 }
451
452 tools.extend(self.tools().enabled_tools(cx).into_iter().map(|tool| {
453 LanguageModelRequestTool {
454 name: tool.name(),
455 description: tool.description(),
456 input_schema: tool.input_schema(),
457 }
458 }));
459
460 tools
461 };
462
463 self.stream_completion(request, model, cx);
464 }
465
466 pub fn to_completion_request(
467 &self,
468 request_kind: RequestKind,
469 cx: &App,
470 ) -> LanguageModelRequest {
471 let worktree_root_names = self
472 .project
473 .read(cx)
474 .visible_worktrees(cx)
475 .map(|worktree| {
476 let worktree = worktree.read(cx);
477 AssistantSystemPromptWorktree {
478 root_name: worktree.root_name().into(),
479 abs_path: worktree.abs_path(),
480 }
481 })
482 .collect::<Vec<_>>();
483 let system_prompt = self
484 .prompt_builder
485 .generate_assistant_system_prompt(worktree_root_names)
486 .context("failed to generate assistant system prompt")
487 .log_err()
488 .unwrap_or_default();
489
490 let mut request = LanguageModelRequest {
491 messages: vec![LanguageModelRequestMessage {
492 role: Role::System,
493 content: vec![MessageContent::Text(system_prompt)],
494 cache: true,
495 }],
496 tools: Vec::new(),
497 stop: Vec::new(),
498 temperature: None,
499 };
500
501 let mut referenced_context_ids = HashSet::default();
502
503 for message in &self.messages {
504 if let Some(context_ids) = self.context_by_message.get(&message.id) {
505 referenced_context_ids.extend(context_ids);
506 }
507
508 let mut request_message = LanguageModelRequestMessage {
509 role: message.role,
510 content: Vec::new(),
511 cache: false,
512 };
513
514 match request_kind {
515 RequestKind::Chat => {
516 self.tool_use
517 .attach_tool_results(message.id, &mut request_message);
518 self.scripting_tool_use
519 .attach_tool_results(message.id, &mut request_message);
520 }
521 RequestKind::Summarize => {
522 // We don't care about tool use during summarization.
523 }
524 }
525
526 if !message.text.is_empty() {
527 request_message
528 .content
529 .push(MessageContent::Text(message.text.clone()));
530 }
531
532 match request_kind {
533 RequestKind::Chat => {
534 self.tool_use
535 .attach_tool_uses(message.id, &mut request_message);
536 self.scripting_tool_use
537 .attach_tool_uses(message.id, &mut request_message);
538 }
539 RequestKind::Summarize => {
540 // We don't care about tool use during summarization.
541 }
542 };
543
544 request.messages.push(request_message);
545 }
546
547 if !referenced_context_ids.is_empty() {
548 let mut context_message = LanguageModelRequestMessage {
549 role: Role::User,
550 content: Vec::new(),
551 cache: false,
552 };
553
554 let referenced_context = referenced_context_ids
555 .into_iter()
556 .filter_map(|context_id| self.context.get(context_id))
557 .cloned();
558 attach_context_to_message(&mut context_message, referenced_context);
559
560 request.messages.push(context_message);
561 }
562
563 request
564 }
565
566 pub fn stream_completion(
567 &mut self,
568 request: LanguageModelRequest,
569 model: Arc<dyn LanguageModel>,
570 cx: &mut Context<Self>,
571 ) {
572 let pending_completion_id = post_inc(&mut self.completion_count);
573
574 let task = cx.spawn(|thread, mut cx| async move {
575 let stream = model.stream_completion(request, &cx);
576 let stream_completion = async {
577 let mut events = stream.await?;
578 let mut stop_reason = StopReason::EndTurn;
579 let mut current_token_usage = TokenUsage::default();
580
581 while let Some(event) = events.next().await {
582 let event = event?;
583
584 thread.update(&mut cx, |thread, cx| {
585 match event {
586 LanguageModelCompletionEvent::StartMessage { .. } => {
587 thread.insert_message(Role::Assistant, String::new(), cx);
588 }
589 LanguageModelCompletionEvent::Stop(reason) => {
590 stop_reason = reason;
591 }
592 LanguageModelCompletionEvent::UsageUpdate(token_usage) => {
593 thread.cumulative_token_usage =
594 thread.cumulative_token_usage.clone() + token_usage.clone()
595 - current_token_usage.clone();
596 current_token_usage = token_usage;
597 }
598 LanguageModelCompletionEvent::Text(chunk) => {
599 if let Some(last_message) = thread.messages.last_mut() {
600 if last_message.role == Role::Assistant {
601 last_message.text.push_str(&chunk);
602 cx.emit(ThreadEvent::StreamedAssistantText(
603 last_message.id,
604 chunk,
605 ));
606 } else {
607 // If we won't have an Assistant message yet, assume this chunk marks the beginning
608 // of a new Assistant response.
609 //
610 // Importantly: We do *not* want to emit a `StreamedAssistantText` event here, as it
611 // will result in duplicating the text of the chunk in the rendered Markdown.
612 thread.insert_message(Role::Assistant, chunk, cx);
613 };
614 }
615 }
616 LanguageModelCompletionEvent::ToolUse(tool_use) => {
617 if let Some(last_assistant_message) = thread
618 .messages
619 .iter()
620 .rfind(|message| message.role == Role::Assistant)
621 {
622 if tool_use.name.as_ref() == ScriptingTool::NAME {
623 thread
624 .scripting_tool_use
625 .request_tool_use(last_assistant_message.id, tool_use);
626 } else {
627 thread
628 .tool_use
629 .request_tool_use(last_assistant_message.id, tool_use);
630 }
631 }
632 }
633 }
634
635 thread.touch_updated_at();
636 cx.emit(ThreadEvent::StreamedCompletion);
637 cx.notify();
638 })?;
639
640 smol::future::yield_now().await;
641 }
642
643 thread.update(&mut cx, |thread, cx| {
644 thread
645 .pending_completions
646 .retain(|completion| completion.id != pending_completion_id);
647
648 if thread.summary.is_none() && thread.messages.len() >= 2 {
649 thread.summarize(cx);
650 }
651 })?;
652
653 anyhow::Ok(stop_reason)
654 };
655
656 let result = stream_completion.await;
657
658 thread
659 .update(&mut cx, |thread, cx| {
660 match result.as_ref() {
661 Ok(stop_reason) => match stop_reason {
662 StopReason::ToolUse => {
663 cx.emit(ThreadEvent::UsePendingTools);
664 }
665 StopReason::EndTurn => {}
666 StopReason::MaxTokens => {}
667 },
668 Err(error) => {
669 if error.is::<PaymentRequiredError>() {
670 cx.emit(ThreadEvent::ShowError(ThreadError::PaymentRequired));
671 } else if error.is::<MaxMonthlySpendReachedError>() {
672 cx.emit(ThreadEvent::ShowError(
673 ThreadError::MaxMonthlySpendReached,
674 ));
675 } else {
676 let error_message = error
677 .chain()
678 .map(|err| err.to_string())
679 .collect::<Vec<_>>()
680 .join("\n");
681 cx.emit(ThreadEvent::ShowError(ThreadError::Message(
682 SharedString::from(error_message.clone()),
683 )));
684 }
685
686 thread.cancel_last_completion(cx);
687 }
688 }
689 cx.emit(ThreadEvent::DoneStreaming);
690 })
691 .ok();
692 });
693
694 self.pending_completions.push(PendingCompletion {
695 id: pending_completion_id,
696 _task: task,
697 });
698 }
699
700 pub fn summarize(&mut self, cx: &mut Context<Self>) {
701 let Some(provider) = LanguageModelRegistry::read_global(cx).active_provider() else {
702 return;
703 };
704 let Some(model) = LanguageModelRegistry::read_global(cx).active_model() else {
705 return;
706 };
707
708 if !provider.is_authenticated(cx) {
709 return;
710 }
711
712 let mut request = self.to_completion_request(RequestKind::Summarize, cx);
713 request.messages.push(LanguageModelRequestMessage {
714 role: Role::User,
715 content: vec![
716 "Generate a concise 3-7 word title for this conversation, omitting punctuation. Go straight to the title, without any preamble and prefix like `Here's a concise suggestion:...` or `Title:`"
717 .into(),
718 ],
719 cache: false,
720 });
721
722 self.pending_summary = cx.spawn(|this, mut cx| {
723 async move {
724 let stream = model.stream_completion_text(request, &cx);
725 let mut messages = stream.await?;
726
727 let mut new_summary = String::new();
728 while let Some(message) = messages.stream.next().await {
729 let text = message?;
730 let mut lines = text.lines();
731 new_summary.extend(lines.next());
732
733 // Stop if the LLM generated multiple lines.
734 if lines.next().is_some() {
735 break;
736 }
737 }
738
739 this.update(&mut cx, |this, cx| {
740 if !new_summary.is_empty() {
741 this.summary = Some(new_summary.into());
742 }
743
744 cx.emit(ThreadEvent::SummaryChanged);
745 })?;
746
747 anyhow::Ok(())
748 }
749 .log_err()
750 });
751 }
752
753 pub fn use_pending_tools(&mut self, cx: &mut Context<Self>) {
754 let request = self.to_completion_request(RequestKind::Chat, cx);
755 let pending_tool_uses = self
756 .tool_use
757 .pending_tool_uses()
758 .into_iter()
759 .filter(|tool_use| tool_use.status.is_idle())
760 .cloned()
761 .collect::<Vec<_>>();
762
763 for tool_use in pending_tool_uses {
764 if let Some(tool) = self.tools.tool(&tool_use.name, cx) {
765 let task = tool.run(
766 tool_use.input,
767 &request.messages,
768 self.project.clone(),
769 self.action_log.clone(),
770 cx,
771 );
772
773 self.insert_tool_output(tool_use.id.clone(), task, cx);
774 }
775 }
776
777 let pending_scripting_tool_uses = self
778 .scripting_tool_use
779 .pending_tool_uses()
780 .into_iter()
781 .filter(|tool_use| tool_use.status.is_idle())
782 .cloned()
783 .collect::<Vec<_>>();
784
785 for scripting_tool_use in pending_scripting_tool_uses {
786 let task = match ScriptingTool::deserialize_input(scripting_tool_use.input) {
787 Err(err) => Task::ready(Err(err.into())),
788 Ok(input) => {
789 let (script_id, script_task) =
790 self.scripting_session.update(cx, move |session, cx| {
791 session.run_script(input.lua_script, cx)
792 });
793
794 let session = self.scripting_session.clone();
795 cx.spawn(|_, cx| async move {
796 script_task.await;
797
798 let message = session.read_with(&cx, |session, _cx| {
799 // Using a id to get the script output seems impractical.
800 // Why not just include it in the Task result?
801 // This is because we'll later report the script state as it runs,
802 session
803 .get(script_id)
804 .output_message_for_llm()
805 .expect("Script shouldn't still be running")
806 })?;
807
808 Ok(message)
809 })
810 }
811 };
812
813 self.insert_scripting_tool_output(scripting_tool_use.id.clone(), task, cx);
814 }
815 }
816
817 pub fn insert_tool_output(
818 &mut self,
819 tool_use_id: LanguageModelToolUseId,
820 output: Task<Result<String>>,
821 cx: &mut Context<Self>,
822 ) {
823 let insert_output_task = cx.spawn(|thread, mut cx| {
824 let tool_use_id = tool_use_id.clone();
825 async move {
826 let output = output.await;
827 thread
828 .update(&mut cx, |thread, cx| {
829 let pending_tool_use = thread
830 .tool_use
831 .insert_tool_output(tool_use_id.clone(), output);
832
833 cx.emit(ThreadEvent::ToolFinished {
834 tool_use_id,
835 pending_tool_use,
836 canceled: false,
837 });
838 })
839 .ok();
840 }
841 });
842
843 self.tool_use
844 .run_pending_tool(tool_use_id, insert_output_task);
845 }
846
847 pub fn insert_scripting_tool_output(
848 &mut self,
849 tool_use_id: LanguageModelToolUseId,
850 output: Task<Result<String>>,
851 cx: &mut Context<Self>,
852 ) {
853 let insert_output_task = cx.spawn(|thread, mut cx| {
854 let tool_use_id = tool_use_id.clone();
855 async move {
856 let output = output.await;
857 thread
858 .update(&mut cx, |thread, cx| {
859 let pending_tool_use = thread
860 .scripting_tool_use
861 .insert_tool_output(tool_use_id.clone(), output);
862
863 cx.emit(ThreadEvent::ToolFinished {
864 tool_use_id,
865 pending_tool_use,
866 canceled: false,
867 });
868 })
869 .ok();
870 }
871 });
872
873 self.scripting_tool_use
874 .run_pending_tool(tool_use_id, insert_output_task);
875 }
876
877 pub fn attach_tool_results(
878 &mut self,
879 updated_context: Vec<ContextSnapshot>,
880 cx: &mut Context<Self>,
881 ) {
882 self.context.extend(
883 updated_context
884 .into_iter()
885 .map(|context| (context.id, context)),
886 );
887
888 // Insert a user message to contain the tool results.
889 self.insert_user_message(
890 // TODO: Sending up a user message without any content results in the model sending back
891 // responses that also don't have any content. We currently don't handle this case well,
892 // so for now we provide some text to keep the model on track.
893 "Here are the tool results.",
894 Vec::new(),
895 cx,
896 );
897 }
898
899 /// Cancels the last pending completion, if there are any pending.
900 ///
901 /// Returns whether a completion was canceled.
902 pub fn cancel_last_completion(&mut self, cx: &mut Context<Self>) -> bool {
903 if self.pending_completions.pop().is_some() {
904 true
905 } else {
906 let mut canceled = false;
907 for pending_tool_use in self.tool_use.cancel_pending() {
908 canceled = true;
909 cx.emit(ThreadEvent::ToolFinished {
910 tool_use_id: pending_tool_use.id.clone(),
911 pending_tool_use: Some(pending_tool_use),
912 canceled: true,
913 });
914 }
915 canceled
916 }
917 }
918
919 /// Reports feedback about the thread and stores it in our telemetry backend.
920 pub fn report_feedback(&self, is_positive: bool, cx: &mut Context<Self>) -> Task<Result<()>> {
921 let final_project_snapshot = Self::project_snapshot(self.project.clone(), cx);
922 let serialized_thread = self.serialize(cx);
923 let thread_id = self.id().clone();
924 let client = self.project.read(cx).client();
925
926 cx.background_spawn(async move {
927 let final_project_snapshot = final_project_snapshot.await;
928 let serialized_thread = serialized_thread.await?;
929 let thread_data =
930 serde_json::to_value(serialized_thread).unwrap_or_else(|_| serde_json::Value::Null);
931
932 let rating = if is_positive { "positive" } else { "negative" };
933 telemetry::event!(
934 "Assistant Thread Rated",
935 rating,
936 thread_id,
937 thread_data,
938 final_project_snapshot
939 );
940 client.telemetry().flush_events();
941
942 Ok(())
943 })
944 }
945
946 /// Create a snapshot of the current project state including git information and unsaved buffers.
947 fn project_snapshot(
948 project: Entity<Project>,
949 cx: &mut Context<Self>,
950 ) -> Task<Arc<ProjectSnapshot>> {
951 let worktree_snapshots: Vec<_> = project
952 .read(cx)
953 .visible_worktrees(cx)
954 .map(|worktree| Self::worktree_snapshot(worktree, cx))
955 .collect();
956
957 cx.spawn(move |_, cx| async move {
958 let worktree_snapshots = futures::future::join_all(worktree_snapshots).await;
959
960 let mut unsaved_buffers = Vec::new();
961 cx.update(|app_cx| {
962 let buffer_store = project.read(app_cx).buffer_store();
963 for buffer_handle in buffer_store.read(app_cx).buffers() {
964 let buffer = buffer_handle.read(app_cx);
965 if buffer.is_dirty() {
966 if let Some(file) = buffer.file() {
967 let path = file.path().to_string_lossy().to_string();
968 unsaved_buffers.push(path);
969 }
970 }
971 }
972 })
973 .ok();
974
975 Arc::new(ProjectSnapshot {
976 worktree_snapshots,
977 unsaved_buffer_paths: unsaved_buffers,
978 timestamp: Utc::now(),
979 })
980 })
981 }
982
983 fn worktree_snapshot(worktree: Entity<project::Worktree>, cx: &App) -> Task<WorktreeSnapshot> {
984 cx.spawn(move |cx| async move {
985 // Get worktree path and snapshot
986 let worktree_info = cx.update(|app_cx| {
987 let worktree = worktree.read(app_cx);
988 let path = worktree.abs_path().to_string_lossy().to_string();
989 let snapshot = worktree.snapshot();
990 (path, snapshot)
991 });
992
993 let Ok((worktree_path, snapshot)) = worktree_info else {
994 return WorktreeSnapshot {
995 worktree_path: String::new(),
996 git_state: None,
997 };
998 };
999
1000 // Extract git information
1001 let git_state = match snapshot.repositories().first() {
1002 None => None,
1003 Some(repo_entry) => {
1004 // Get branch information
1005 let current_branch = repo_entry.branch().map(|branch| branch.name.to_string());
1006
1007 // Get repository info
1008 let repo_result = worktree.read_with(&cx, |worktree, _cx| {
1009 if let project::Worktree::Local(local_worktree) = &worktree {
1010 local_worktree.get_local_repo(repo_entry).map(|local_repo| {
1011 let repo = local_repo.repo();
1012 (repo.remote_url("origin"), repo.head_sha(), repo.clone())
1013 })
1014 } else {
1015 None
1016 }
1017 });
1018
1019 match repo_result {
1020 Ok(Some((remote_url, head_sha, repository))) => {
1021 // Get diff asynchronously
1022 let diff = repository
1023 .diff(git::repository::DiffType::HeadToWorktree, cx)
1024 .await
1025 .ok();
1026
1027 Some(GitState {
1028 remote_url,
1029 head_sha,
1030 current_branch,
1031 diff,
1032 })
1033 }
1034 Err(_) | Ok(None) => None,
1035 }
1036 }
1037 };
1038
1039 WorktreeSnapshot {
1040 worktree_path,
1041 git_state,
1042 }
1043 })
1044 }
1045
1046 pub fn to_markdown(&self) -> Result<String> {
1047 let mut markdown = Vec::new();
1048
1049 if let Some(summary) = self.summary() {
1050 writeln!(markdown, "# {summary}\n")?;
1051 };
1052
1053 for message in self.messages() {
1054 writeln!(
1055 markdown,
1056 "## {role}\n",
1057 role = match message.role {
1058 Role::User => "User",
1059 Role::Assistant => "Assistant",
1060 Role::System => "System",
1061 }
1062 )?;
1063 writeln!(markdown, "{}\n", message.text)?;
1064
1065 for tool_use in self.tool_uses_for_message(message.id) {
1066 writeln!(
1067 markdown,
1068 "**Use Tool: {} ({})**",
1069 tool_use.name, tool_use.id
1070 )?;
1071 writeln!(markdown, "```json")?;
1072 writeln!(
1073 markdown,
1074 "{}",
1075 serde_json::to_string_pretty(&tool_use.input)?
1076 )?;
1077 writeln!(markdown, "```")?;
1078 }
1079
1080 for tool_result in self.tool_results_for_message(message.id) {
1081 write!(markdown, "**Tool Results: {}", tool_result.tool_use_id)?;
1082 if tool_result.is_error {
1083 write!(markdown, " (Error)")?;
1084 }
1085
1086 writeln!(markdown, "**\n")?;
1087 writeln!(markdown, "{}", tool_result.content)?;
1088 }
1089 }
1090
1091 Ok(String::from_utf8_lossy(&markdown).to_string())
1092 }
1093
1094 pub fn action_log(&self) -> &Entity<ActionLog> {
1095 &self.action_log
1096 }
1097
1098 pub fn cumulative_token_usage(&self) -> TokenUsage {
1099 self.cumulative_token_usage.clone()
1100 }
1101}
1102
1103#[derive(Debug, Clone)]
1104pub enum ThreadError {
1105 PaymentRequired,
1106 MaxMonthlySpendReached,
1107 Message(SharedString),
1108}
1109
1110#[derive(Debug, Clone)]
1111pub enum ThreadEvent {
1112 ShowError(ThreadError),
1113 StreamedCompletion,
1114 StreamedAssistantText(MessageId, String),
1115 DoneStreaming,
1116 MessageAdded(MessageId),
1117 MessageEdited(MessageId),
1118 MessageDeleted(MessageId),
1119 SummaryChanged,
1120 UsePendingTools,
1121 ToolFinished {
1122 #[allow(unused)]
1123 tool_use_id: LanguageModelToolUseId,
1124 /// The pending tool use that corresponds to this tool.
1125 pending_tool_use: Option<PendingToolUse>,
1126 /// Whether the tool was canceled by the user.
1127 canceled: bool,
1128 },
1129}
1130
1131impl EventEmitter<ThreadEvent> for Thread {}
1132
1133struct PendingCompletion {
1134 id: usize,
1135 _task: Task<()>,
1136}