//! One chat thread: owns the background agent process (Claude Code, Pi or //! jcode) and folds its event stream into a list of renderable chat items. use std::{ collections::{HashMap, HashSet, VecDeque}, path::{Path, PathBuf}, rc::Rc, sync::atomic::{AtomicU64, Ordering}, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use gpui_kit::component::text::TextViewState; use gpui_kit::*; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use crate::browser; use crate::claude::{AgentProcess, Effort, LaunchConfig, PermissionMode, ProcessEvent}; use crate::git::{self, TurnChanges}; use crate::models::{Catalog, Provider}; use crate::remote::{ClientMessage, RemoteTask, ThreadAction, ThreadDetail, ThreadSummary, Upload, client::RemoteClient}; use crate::settings::{self, Streaming}; use crate::worktree::{self, PullRequest, Worktree, WorktreePlan}; const DEFAULT_TITLE: &str = "New thread"; const STDERR_TAIL: usize = 20; /// Claude Code's tool for putting multiple-choice questions to the user. pub const ASK_USER_QUESTION: &str = "AskUserQuestion"; /// The API's limit for one base64 image. const MAX_IMAGE_BYTES: u64 = 5 * 1024 * 1024; /// How the app introduces itself to jcode's harness API. pub const JCODE_CLIENT: &str = concat!("vectrum-code/", env!("CARGO_PKG_VERSION")); static NEXT_SESSION: AtomicU64 = AtomicU64::new(1); #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum SessionStatus { Starting, Idle, Working, Exited, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ToolState { Running, Done, Error, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum PermissionState { Pending, Allowed, Denied, Cancelled, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum PermissionDecision { Allow, AllowAlways, Deny, } pub struct PermissionPrompt { pub request_id: String, pub input: Value, pub suggestions: Vec, pub state: PermissionState, } impl PermissionPrompt { /// Label for the "remember this" button, if the CLI offered a rule to save. pub fn always_label(&self) -> Option<&'static str> { self.suggestions.iter().find_map(|suggestion| { match suggestion.get("type").and_then(Value::as_str)? { "setMode" if suggestion.get("mode").and_then(Value::as_str) == Some("acceptEdits") => { Some("Accept all edits") } "addRules" | "replaceRules" => Some("Always allow"), _ => None, } }) } } pub struct ToolCall { pub id: String, pub name: String, pub input: Value, pub partial_input: String, pub result: Option, pub state: ToolState, pub permission: Option, pub expanded: bool, /// Tool calls made by a subagent running inside this (Task) tool. pub subagent_tool_count: usize, /// Created from a permission request for a tool we never saw being called /// (subagent tools), so no result will ever arrive for it. pub detached: bool, /// Claude's AskUserQuestion: the form the user fills in. It's answered /// through the tool's permission prompt. pub ask: Option, /// The CLI's task for a long-running command or subagent, once it has /// one: a foreground task can be sent to the background. pub task_id: Option, /// Set once the tool runs as a background task. pub background: Option, } /// Where a background task stands. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum TaskStatus { Running, Completed, Failed, Stopped, } impl TaskStatus { fn key(self) -> &'static str { match self { TaskStatus::Running => "running", TaskStatus::Completed => "completed", TaskStatus::Failed => "failed", TaskStatus::Stopped => "stopped", } } fn from_cli(status: &str) -> Self { match status { "completed" => TaskStatus::Completed, "failed" => TaskStatus::Failed, "running" | "pending" => TaskStatus::Running, _ => TaskStatus::Stopped, } } } /// A command, monitor or subagent Claude left running in the background. The /// CLI wakes Claude when it finishes. pub struct BackgroundTask { pub task_id: String, pub description: SharedString, pub started: Instant, } pub struct AskOption { pub label: SharedString, pub description: Option, } pub struct AskQuestion { pub question: SharedString, /// A short chip label, e.g. "Auth method". pub header: SharedString, pub options: Vec, pub multi_select: bool, } /// The questions an AskUserQuestion call puts to the user, and their answers /// so far. pub struct UserQuestions { pub questions: Vec, /// The question on screen. pub current: usize, /// Options ticked on each multi-select question, not yet committed. pub picked: Vec>, pub answers: Vec>, } impl UserQuestions { fn parse(input: &Value) -> Option { let text = |value: &Value, key: &str| { value.get(key).and_then(Value::as_str).map(|text| SharedString::from(text.to_string())) }; let questions: Vec = input .get("questions")? .as_array()? .iter() .filter_map(|question| { Some(AskQuestion { question: text(question, "question")?, header: text(question, "header").unwrap_or_default(), options: question .get("options") .and_then(Value::as_array) .into_iter() .flatten() .filter_map(|option| { Some(AskOption { label: text(option, "label")?, description: text(option, "description").filter(|text| !text.is_empty()), }) }) .collect(), multi_select: question.get("multiSelect").and_then(Value::as_bool).unwrap_or(false), }) }) .collect(); if questions.is_empty() { return None; } let count = questions.len(); Some(Self { questions, current: 0, picked: vec![Vec::new(); count], answers: vec![None; count] }) } /// Records an answer and moves to the next unanswered question. Returns /// true once every question has one. fn answer(&mut self, answer: SharedString) -> bool { self.answers[self.current] = Some(answer); let count = self.questions.len(); match (1..count).map(|step| (self.current + step) % count).find(|&q| self.answers[q].is_none()) { Some(next) => { self.current = next; false } None => self.answers.iter().all(Option::is_some), } } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum NoticeKind { Info, Error, } /// A question an agent extension put to the user (Pi's extension UI). pub struct Question { pub request_id: String, pub title: SharedString, pub message: Option, pub options: Vec, /// A yes/no confirmation rather than a pick from a list. pub confirm: bool, pub answer: Option, } /// An answer as the agent wrote it, and the Markdown view showing it. While /// streaming, the view may lag behind the text to show only finished /// paragraphs. pub struct AssistantText { pub text: String, pub view: Entity, /// How much of `text` the view shows. shown: usize, } pub enum ChatItem { User { text: SharedString, /// File names of what was attached. attachments: Vec, at: SystemTime, }, Assistant(AssistantText), Tool(ToolCall), Notice(NoticeKind, SharedString), Question(Question), /// A background task finished; Claude picks the work back up from here. TaskEvent { summary: SharedString, status: TaskStatus }, TurnEnd { duration_ms: u64, cost: Option, at: SystemTime }, } /// A thread that lives on another device's app, mirrored here: actions go to /// the host, and the host sends back how the thread changed. pub struct RemoteLink { /// The connection's id in the settings. pub env: String, /// The thread's id on the host; `None` until its first message creates it. pub thread: Option, pub client: Rc, /// Asked the host for the transcript. pub subscribed: bool, /// Got the transcript at least once. pub loaded: bool, /// What the host last said about pending prompts, until the transcript is in. pub attention: bool, /// The request that creates the thread, while it's on its way. pub creating: Option, } /// What a session tells the views around it. pub enum SessionEvent { /// A finished turn's changes on disk are known, by its `TurnEnd` index. TurnChanges(usize), } /// A message waiting for the running turn to end. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct QueuedMessage { pub text: String, pub attachments: Vec, } /// How hard the model reasons, as picked in the composer. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum Reasoning { /// Whatever the model or agent defaults to. #[default] Default, Level(Effort), /// Claude only: extra-high effort plus multi-agent workflow orchestration. Ultracode, /// Claude only: asks for deeper reasoning on every message. Ultrathink, } impl Reasoning { pub fn label(self) -> &'static str { match self { Reasoning::Default => "Default", Reasoning::Level(effort) => effort.label(), Reasoning::Ultracode => "Ultracode", Reasoning::Ultrathink => "Ultrathink", } } /// How the settings file spells it: `default`, an effort level, /// `ultracode` or `ultrathink`. pub fn key(self) -> &'static str { match self { Reasoning::Default => "default", Reasoning::Level(effort) => effort.as_cli(), Reasoning::Ultracode => "ultracode", Reasoning::Ultrathink => "ultrathink", } } pub fn from_key(key: &str) -> Option { match key { "default" => Some(Reasoning::Default), "ultracode" => Some(Reasoning::Ultracode), "ultrathink" => Some(Reasoning::Ultrathink), key => Effort::from_cli(key).map(Reasoning::Level), } } /// The effort level the agent runs at. fn effort(self) -> Option { match self { Reasoning::Level(effort) => Some(effort), Reasoning::Ultracode => Some(Effort::XHigh), Reasoning::Default | Reasoning::Ultrathink => None, } } } enum PendingControl { Initialize, SetModel, SetPermissionMode, FlagSettings, Interrupt, Background, StopTask, } /// A finished turn, which the transcript folds into one "Worked for …" row /// above its final answer. pub struct FinishedTurn { /// The first item after the user's message; the fold row sits here. pub first: usize, /// The last assistant message, which stays visible. pub answer: Option, /// The turn's `TurnEnd` item. pub end: usize, pub duration_ms: u64, pub cost: Option, /// When it ended. pub at: SystemTime, /// Whether anything besides the answer happened (tools, interim text). pub has_work: bool, } /// Maps the content blocks of the message currently being streamed to the /// chat items they render into. struct StreamingMessage { blocks: HashMap, } /// Where a thread's connection to jcode's harness API stands. Requests are /// numbered; most name the session, so they wait until it's attached. #[derive(Default)] struct JcodeLink { /// The jcode session this process is attached to. session: Option, /// Session requests waiting for the attach. held: Vec<(&'static str, Value)>, /// What each request waiting for its reply was for, to explain errors. requests: HashMap, /// The chat item each streamed text message renders into. texts: HashMap, /// The serving provider, e.g. `Anthropic`, which decides how token /// counts add up. provider: Option, /// The browser panel's instructions, sent along with the next message. browser_note: Option, } pub struct Session { /// Names the thread's file on disk. pub id: String, pub provider: Provider, /// The provider instance running it (`claude`, `claude-2`, `pi`). pub instance: String, pub title: SharedString, /// Set by the user; generated titles don't replace it. pub title_custom: bool, /// The project folder the thread belongs to. pub project: PathBuf, /// Where the agent runs: the project folder, or a place in the thread's worktree. pub cwd: PathBuf, pub worktree: Option, /// The worktree to create when the first message is sent. pub pending_worktree: Option, pub pull_request: Option, /// Set for a thread mirrored from another device. pub remote: Option, pub items: Vec, pub status: SessionStatus, pub thinking: bool, /// What a restarted process resumes: Claude's or jcode's session id, or /// Pi's session file. resume_id: Option, /// The model picked for the thread; `None` means the agent's default. pub model: Option, /// The concrete model the agent reported, e.g. `claude-opus-5-5`. pub active_model: Option, pub permission_mode: PermissionMode, pub reasoning: Reasoning, pub fast_mode: bool, /// The thinking levels Pi offers for the current model. pub pi_efforts: Vec, pub total_cost: f64, /// What the current process has reported spending; restarts reset it. process_cost: f64, /// Pi reports cost per message; this sums the running turn. turn_cost: f64, pub turn_started: Option, pub created: SystemTime, /// When the thread last saw activity, for ordering the thread list. pub updated: SystemTime, /// When the user last sent a message. pub last_message: SystemTime, /// Filed away by the user: its process is stopped until the next message. pub settled: bool, /// Kept at the top of the thread list. pub pinned: bool, /// Hidden from the thread list until then. pub snoozed_until: Option, /// When a usage limit that stopped the thread resets. pub limit_resets: Option, /// When to continue the thread on its own, after a usage limit. pub resume_at: Option, /// Messages held until the running turn ends. pub queue: Vec, /// Restored from disk after the app stopped it mid-turn. pub interrupted: bool, /// Creating the worktree before the first message goes out. pub preparing: bool, /// Tokens in the context window as of the last response. pub context_tokens: Option, pub context_window: Option, /// Tasks running in the background, oldest first. pub background_tasks: Vec, /// When each task the CLI reported started, and the tool that started it. task_origins: HashMap)>, /// Finished turns the user unfolded, keyed by their `TurnEnd` index. expanded_turns: HashSet, /// What each finished turn changed on disk, keyed by its `TurnEnd` index. turn_changes: HashMap, /// Folders the user opened in a turn's change overview. expanded_dirs: HashMap>, /// The private git index snapshots are staged into. snapshot_index: PathBuf, /// The working tree as the running turn found it. base_snapshot: Option>>, /// Set when the user stops a turn, so its error result isn't reported. interrupt_requested: bool, stderr_tail: VecDeque, process: Option, streaming: Option, streamed_message_ids: HashSet, pending_controls: HashMap, next_request_id: u64, jcode: JcodeLink, _events_task: Task<()>, } /// What a new thread starts with. pub struct NewSession { pub instance: String, pub provider: Provider, pub project: PathBuf, pub model: Option, pub permission_mode: PermissionMode, pub reasoning: Reasoning, pub fast_mode: bool, /// Start in a worktree of its own, made when the first message is sent. pub worktree: Option, } impl EventEmitter for Session {} impl Session { pub fn new(new: NewSession, cx: &mut Context) -> Self { let now = SystemTime::now(); let id = format!( "{:x}-{:x}-{}", now.duration_since(UNIX_EPOCH).unwrap_or_default().as_millis(), std::process::id(), NEXT_SESSION.load(Ordering::Relaxed) ); let start = new.worktree.is_none(); let mut session = Self::empty(id, new, now); if start { session.start(None, cx); } else { session.status = SessionStatus::Idle; } session } fn empty(id: String, new: NewSession, now: SystemTime) -> Self { let snapshot_index = std::env::temp_dir().join(format!( "vectrum-code-{}-{}.index", std::process::id(), NEXT_SESSION.fetch_add(1, Ordering::Relaxed) )); let NewSession { instance, provider, project, model, permission_mode, reasoning, fast_mode, worktree } = new; Self { id, provider, instance, title: DEFAULT_TITLE.into(), title_custom: false, cwd: project.clone(), project, worktree: None, pending_worktree: worktree, pull_request: None, remote: None, items: Vec::new(), status: SessionStatus::Starting, thinking: false, resume_id: None, model, active_model: None, permission_mode, reasoning, fast_mode, pi_efforts: Vec::new(), total_cost: 0., process_cost: 0., turn_cost: 0., turn_started: None, created: now, updated: now, last_message: now, settled: false, pinned: false, snoozed_until: None, limit_resets: None, resume_at: None, queue: Vec::new(), interrupted: false, preparing: false, context_tokens: None, context_window: None, background_tasks: Vec::new(), task_origins: HashMap::new(), expanded_turns: HashSet::new(), turn_changes: HashMap::new(), expanded_dirs: HashMap::new(), snapshot_index, base_snapshot: None, interrupt_requested: false, stderr_tail: VecDeque::new(), process: None, streaming: None, streamed_message_ids: HashSet::new(), pending_controls: HashMap::new(), next_request_id: 0, jcode: JcodeLink::default(), _events_task: Task::ready(()), } } pub fn is_working(&self) -> bool { self.status == SessionStatus::Working } /// True while a tool call or question is waiting for the user. pub fn needs_attention(&self) -> bool { if let Some(remote) = self.remote.as_ref().filter(|remote| !remote.loaded) { return remote.attention; } self.items.iter().any(|item| match item { ChatItem::Tool(ToolCall { permission: Some(prompt), .. }) => { prompt.state == PermissionState::Pending } ChatItem::Question(question) => question.answer.is_none(), _ => false, }) } fn start(&mut self, resume: Option, cx: &mut Context) { let (tx, rx) = smol::channel::unbounded(); let config = LaunchConfig { instance: settings::instance(&self.instance, self.provider), cwd: self.cwd.clone(), model: self.model.clone(), permission_mode: self.permission_mode, effort: self.reasoning.effort(), resume: resume.clone(), browser_session: Some(browser::session_name(&self.id)), }; let process = match AgentProcess::spawn(&config, tx) { Ok(process) => process, Err(err) => { self.status = SessionStatus::Exited; self.push_notice(NoticeKind::Error, format!("{err:#}")); return; } }; self.status = SessionStatus::Starting; self.process_cost = 0.; self.stderr_tail.clear(); self.process = Some(process); match self.provider { Provider::Claude => { self.control(PendingControl::Initialize, json!({"subtype": "initialize"})); // Ultracode and fast mode have no launch flag; they're flag settings. let mut settings = serde_json::Map::new(); if self.reasoning == Reasoning::Ultracode { settings.insert("ultracode".into(), true.into()); } if self.fast_mode { settings.insert("fastMode".into(), true.into()); } if !settings.is_empty() { self.control( PendingControl::FlagSettings, json!({"subtype": "apply_flag_settings", "settings": settings}), ); } } Provider::Pi => { self.pi(json!({"type": "get_state"})); self.pi(json!({"type": "get_available_thinking_levels"})); } Provider::Jcode => { self.jcode = JcodeLink { browser_note: browser::find_binary() .map(|binary| browser::jcode_prompt(&browser::session_name(&self.id), &binary)), ..JcodeLink::default() }; let hello = json!({"req": "hello", "min_version": 1, "max_version": 1, "client": JCODE_CLIENT}); self.jcode_send("connect", hello); match resume { Some(session_id) => self.jcode_send("attach", json!({"req": "attach_session", "session_id": session_id})), None => self.jcode_create_session(), } if let Some(model) = self.model.clone() { self.jcode_request("switch models", json!({"req": "set_model", "model": model})); } if let Some(effort) = self.reasoning.effort() { self.jcode_request( "change the reasoning effort", json!({"req": "set_reasoning_effort", "effort": effort.as_cli()}), ); } } } self._events_task = cx.spawn(async move |this, cx| { while let Ok(event) = rx.recv().await { // Fold whatever else is already queued into one update so a // fast token stream doesn't re-render per delta. let mut batch = vec![event]; while let Ok(event) = rx.try_recv() { batch.push(event); } let updated = this.update(cx, |this, cx| { for event in batch { this.handle_event(event, cx); } cx.notify(); }); if updated.is_err() { break; } } }); } fn request_id(&mut self) -> String { self.next_request_id += 1; format!("vectrum-code-{}", self.next_request_id) } /// Sends a Claude control request and remembers what it was for. fn control(&mut self, kind: PendingControl, request: Value) { let request_id = self.request_id(); if let Some(process) = &self.process { self.pending_controls.insert(request_id.clone(), kind); process.send_control(&request_id, request); } } /// Sends a Pi RPC command. fn pi(&mut self, mut command: Value) { let id = self.request_id(); command["id"] = id.into(); if let Some(process) = &self.process { process.send(command); } } /// Sends a jcode API request now. `what` finishes "jcode couldn't …" if /// it fails. fn jcode_send(&mut self, what: &'static str, mut request: Value) { self.next_request_id += 1; let id = self.next_request_id; request["v"] = 1.into(); request["id"] = id.into(); if let Some(process) = &self.process { self.jcode.requests.insert(id, what); process.send(request); } } /// Sends a request about the thread's jcode session, once it's attached. fn jcode_request(&mut self, what: &'static str, mut request: Value) { match self.jcode.session.clone() { Some(session_id) => { request["session_id"] = session_id.into(); self.jcode_send(what, request); } None => self.jcode.held.push((what, request)), } } fn jcode_create_session(&mut self) { let working_dir = self.cwd.display().to_string(); self.jcode_send("start a session", json!({"req": "create_session", "working_dir": working_dir})); } // ---- user actions ------------------------------------------------- /// Sends a message now, even mid-turn, where the agent takes it in as /// soon as it can. Images are attached inline; other files are listed by /// path for the agent to read. The thread's first message also creates /// its worktree, if it starts in one. pub fn send_message(&mut self, text: String, attachments: Vec, cx: &mut Context) { if self.remote.is_some() { self.send_remote(text, attachments, false, cx); return; } if text.trim().is_empty() && attachments.is_empty() { return; } let names: Vec = attachments.iter().map(|path| file_name(path).into()).collect(); let first = !self.items.iter().any(|item| matches!(item, ChatItem::User { .. })); let now = SystemTime::now(); self.items.push(ChatItem::User { text: text.clone().into(), attachments: names.clone(), at: now }); self.last_message = now; self.updated = now; self.settled = false; self.snoozed_until = None; self.resume_at = None; self.interrupted = false; if first { let source = if text.trim().is_empty() { names.join(", ") } else { text.clone() }; if !self.title_custom { self.title = title_from(&source).into(); self.generate_title(source, cx); } } if let Some(plan) = self.pending_worktree.take() { self.status = SessionStatus::Working; self.preparing = true; self.turn_started = Some(Instant::now()); self.turn_cost = 0.; let title = self.title.to_string(); let task = cx.background_executor().spawn(async move { worktree::create(&plan, &title) }); cx.spawn(async move |this, cx| { let created = task.await; let _ = this.update(cx, |this, cx| { this.preparing = false; match created { Ok((worktree, cwd)) => { this.cwd = cwd; this.worktree = Some(worktree); } Err(err) => this.push_notice( NoticeKind::Error, format!("Couldn't create a worktree: {err}\n\nThe thread runs in the project folder instead."), ), } this.start(None, cx); this.snapshot_base(cx); this.deliver(text, attachments, cx); cx.notify(); }); }) .detach(); cx.notify(); return; } if !self.is_working() { self.turn_started = Some(Instant::now()); self.turn_cost = 0.; self.snapshot_base(cx); } self.deliver(text, attachments, cx); self.status = SessionStatus::Working; cx.notify(); } /// Holds a message until the running turn ends; sends it now if none runs. pub fn queue_message(&mut self, text: String, attachments: Vec, cx: &mut Context) { if self.remote.is_some() { let queue = self.is_working() && self.remote.as_ref().is_some_and(|remote| remote.thread.is_some()); self.send_remote(text, attachments, queue, cx); return; } if !self.is_working() { self.send_message(text, attachments, cx); return; } if text.trim().is_empty() && attachments.is_empty() { return; } self.queue.push(QueuedMessage { text, attachments }); cx.notify(); } /// Sends a queued message now instead of after the turn. pub fn steer_queued(&mut self, ix: usize, cx: &mut Context) { if self.forward(ThreadAction::SteerQueued { ix }) { return; } if ix < self.queue.len() { let message = self.queue.remove(ix); self.send_message(message.text, message.attachments, cx); } } pub fn remove_queued(&mut self, ix: usize, cx: &mut Context) { if self.forward(ThreadAction::RemoveQueued { ix }) { return; } if ix < self.queue.len() { self.queue.remove(ix); cx.notify(); } } /// Sends the next queued message once a turn has ended. fn send_next_queued(&mut self, cx: &mut Context) { if !self.is_working() && !self.queue.is_empty() { let message = self.queue.remove(0); self.send_message(message.text, message.attachments, cx); } } /// Records the working tree as the starting turn finds it. fn snapshot_base(&mut self, cx: &mut Context) { let cwd = self.cwd.clone(); let index = self.snapshot_index.clone(); self.base_snapshot = Some(cx.background_executor().spawn(async move { git::snapshot(&cwd, &index) })); } /// Writes a message to the agent, starting (or resuming) its process if /// it isn't running. fn deliver(&mut self, text: String, attachments: Vec, cx: &mut Context) { let mut images = Vec::new(); let mut file_paths = Vec::new(); for path in &attachments { match image_media_type(path) { Some(media_type) => match read_image(path) { Ok(data) => images.push((media_type, data)), Err(err) => { self.push_notice(NoticeKind::Error, format!("Couldn't attach {}: {err}", file_name(path))); continue; } }, None => file_paths.push(path.display().to_string()), } } let mut prompt = text; if !file_paths.is_empty() { if !prompt.is_empty() { prompt.push_str("\n\n"); } prompt.push_str("Attached files:"); for path in &file_paths { prompt.push_str(&format!("\n- {path}")); } } if prompt.is_empty() && images.is_empty() { return; } if self.provider == Provider::Claude && self.reasoning == Reasoning::Ultrathink { // Claude Code's keyword for deeper reasoning on this turn. prompt.push_str("\n\nultrathink"); } if self.process.is_none() { let resume = self.resume_id.clone(); self.start(resume, cx); } let Some(process) = &self.process else { self.status = SessionStatus::Exited; self.turn_started = None; return; }; match self.provider { Provider::Claude => { let mut content: Vec = images .iter() .map(|(media_type, data)| { json!({"type": "image", "source": {"type": "base64", "media_type": media_type, "data": data}}) }) .collect(); // The API rejects empty text blocks; images go first, as it prefers. if !prompt.is_empty() { content.push(json!({"type": "text", "text": prompt})); } process.send_user_content(content); } Provider::Pi => { let mut command = json!({ "type": "prompt", "message": prompt, "images": images .iter() .map(|(media_type, data)| json!({"type": "image", "data": data, "mimeType": media_type})) .collect::>(), }); // Pi rejects a prompt mid-run unless told when to deliver it. if self.is_working() { command["streamingBehavior"] = "steer".into(); } self.pi(command); } Provider::Jcode => { let images: Vec<(&str, &String)> = images.iter().map(|(media_type, data)| (*media_type, data)).collect(); // Mid-turn, a soft interrupt reaches the model at its next // stop without cancelling what it's doing. if self.is_working() { let request = json!({"req": "soft_interrupt", "content": prompt, "images": images}); self.jcode_request("steer the turn", request); } else { let mut request = json!({"req": "send_message", "content": prompt, "images": images}); if let Some(note) = self.jcode.browser_note.take() { request["system_reminder"] = note.into(); } self.jcode_request("send the message", request); } } } self.status = SessionStatus::Working; } /// Marks a turn as running, unless one already is. Turns also start /// without a message, when a background task finishes and the CLI wakes /// Claude up. fn begin_turn(&mut self, cx: &mut Context) { self.updated = SystemTime::now(); if self.is_working() { return; } self.turn_started = Some(Instant::now()); self.turn_cost = 0.; self.snapshot_base(cx); self.status = SessionStatus::Working; } /// Sends a running command or subagent to the background, like Ctrl+B in /// Claude Code's terminal UI. Claude carries on and hears when it's done. pub fn background_tool(&mut self, ix: usize, cx: &mut Context) { if self.forward(ThreadAction::Background { ix }) { return; } let Some(ChatItem::Tool(tool)) = self.items.get(ix) else { return }; if tool.task_id.is_none() || tool.background.is_some() || tool.state != ToolState::Running { return; } let id = tool.id.clone(); self.control(PendingControl::Background, json!({"subtype": "background_tasks", "tool_use_id": id})); cx.notify(); } /// Stops a background task. pub fn stop_task(&mut self, task_id: &str, cx: &mut Context) { if self.forward(ThreadAction::StopTask { task_id: task_id.to_string() }) { return; } self.control(PendingControl::StopTask, json!({"subtype": "stop_task", "task_id": task_id})); cx.notify(); } /// Asks the text generation model for a title, in the background. fn generate_title(&mut self, message: String, cx: &mut Context) { if settings::get(|settings| settings.text_model.is_empty()) { return; } let task = cx.background_executor().spawn(async move { crate::textgen::title(&message) }); cx.spawn(async move |this, cx| { let Some(title) = task.await else { return }; let _ = this.update(cx, |this, cx| { if !this.title_custom { this.title = title.into(); cx.notify(); } }); }) .detach(); } pub fn rename(&mut self, title: String, cx: &mut Context) { self.forward(ThreadAction::Rename { title: title.trim().to_string() }); let title = title.trim(); if title.is_empty() { return; } self.title = title.to_string().into(); self.title_custom = true; cx.notify(); } /// Picks where the (still empty) thread will run: in the project folder, /// or a worktree made with its first message. pub fn set_worktree_plan(&mut self, plan: Option, cx: &mut Context) { if self.remote.is_some() { // The host makes the worktree; this only says whether to. if self.remote.as_ref().is_some_and(|remote| remote.thread.is_none()) { self.pending_worktree = plan; cx.notify(); } return; } if !self.items.is_empty() || self.worktree.is_some() { return; } let had_plan = self.pending_worktree.is_some(); self.pending_worktree = plan; match (had_plan, self.pending_worktree.is_some()) { (false, true) => { // The agent starts in the worktree once there is one. self.process.take(); self._events_task = Task::ready(()); self.pending_controls.clear(); self.status = SessionStatus::Idle; } (true, false) if self.process.is_none() => self.start(None, cx), _ => {} } cx.notify(); } pub fn set_pinned(&mut self, pinned: bool, cx: &mut Context) { self.forward(ThreadAction::Pin { pinned }); self.pinned = pinned; cx.notify(); } pub fn snooze(&mut self, until: Option, cx: &mut Context) { self.forward(ThreadAction::Snooze { until: until.map(to_millis) }); self.snoozed_until = until; if until.is_none() { self.updated = SystemTime::now(); } cx.notify(); } pub fn is_snoozed(&self) -> bool { self.snoozed_until.is_some_and(|until| until > SystemTime::now()) } /// When the usage limit that stopped the thread resets, while it hasn't. pub fn limited_until(&self) -> Option { self.limit_resets.filter(|resets| *resets > SystemTime::now()) } /// Drops the continuation scheduled for when a usage limit resets. pub fn cancel_resume(&mut self, cx: &mut Context) { self.forward(ThreadAction::CancelResume); self.resume_at = None; cx.notify(); } /// Whether the last turn ended in an error. pub fn ended_with_error(&self) -> bool { self.status == SessionStatus::Exited || self .items .iter() .rev() .take_while(|item| !matches!(item, ChatItem::User { .. })) .any(|item| matches!(item, ChatItem::Notice(NoticeKind::Error, _))) } /// Files the thread under Settled and stops its process. The transcript /// stays, and the next message resumes the conversation. pub fn settle(&mut self, cx: &mut Context) { if self.forward(ThreadAction::Settle) { self.settled = true; cx.notify(); return; } self.flush_text(cx); self.process.take(); // Drop the event stream too, so the process exit isn't reported as a crash. self._events_task = Task::ready(()); self.thinking = false; self.streaming = None; self.turn_started = None; self.base_snapshot = None; self.interrupt_requested = false; self.pending_controls.clear(); self.drop_background_tasks(); self.finish_open_tools("The thread was settled before the tool finished."); if self.status != SessionStatus::Exited { self.status = SessionStatus::Idle; } self.settled = true; self.preparing = false; cx.notify(); } /// Brings a settled thread back to the open list. Its process starts again /// with the next message. pub fn unsettle(&mut self, cx: &mut Context) { self.forward(ThreadAction::Unsettle); self.settled = false; self.updated = SystemTime::now(); cx.notify(); } pub fn interrupt(&mut self, cx: &mut Context) { if self.forward(ThreadAction::Interrupt) { return; } if !self.is_working() || self.interrupt_requested || self.process.is_none() { return; } self.interrupt_requested = true; match self.provider { Provider::Claude => self.control(PendingControl::Interrupt, json!({"subtype": "interrupt"})), Provider::Pi => self.pi(json!({"type": "abort"})), Provider::Jcode => self.jcode_request("stop the turn", json!({"req": "cancel"})), } cx.notify(); } pub fn set_model(&mut self, model: Option, cx: &mut Context) { if self.forward(ThreadAction::SetModel { model: model.clone() }) { self.model = model; cx.notify(); return; } self.model = model.clone(); match self.provider { Provider::Claude => self.control( PendingControl::SetModel, json!({"subtype": "set_model", "model": model.as_deref().unwrap_or("default")}), ), Provider::Pi => { if let Some((provider, model_id)) = model.as_deref().and_then(|model| model.split_once('/')) { self.pi(json!({"type": "set_model", "provider": provider, "modelId": model_id})); self.pi(json!({"type": "get_available_thinking_levels"})); } } // Without a process, the next one starts on the new model. Provider::Jcode if self.process.is_some() => { if let Some(model) = model { self.jcode_request("switch models", json!({"req": "set_model", "model": model})); } } Provider::Jcode => {} } cx.notify(); } pub fn set_permission_mode(&mut self, mode: PermissionMode, cx: &mut Context) { if self.forward(ThreadAction::SetPermissionMode { mode: mode.as_cli().into() }) { self.permission_mode = mode; cx.notify(); return; } // Pi and jcode have no permission prompts to configure. if !self.provider.asks_permission() { return; } self.permission_mode = mode; self.control( PendingControl::SetPermissionMode, json!({"subtype": "set_permission_mode", "mode": mode.as_cli()}), ); cx.notify(); } pub fn set_reasoning(&mut self, reasoning: Reasoning, cx: &mut Context) { if self.forward(ThreadAction::SetReasoning { reasoning: reasoning.key().into() }) { self.reasoning = reasoning; cx.notify(); return; } self.reasoning = reasoning; match self.provider { Provider::Claude => self.control( PendingControl::FlagSettings, json!({ "subtype": "apply_flag_settings", "settings": { "effortLevel": reasoning.effort().map(Effort::as_cli), "ultracode": reasoning == Reasoning::Ultracode, }, }), ), Provider::Pi => { if let Some(effort) = reasoning.effort() { self.pi(json!({"type": "set_thinking_level", "level": effort.as_cli()})); } } Provider::Jcode if self.process.is_some() => { if let Some(effort) = reasoning.effort() { self.jcode_request( "change the reasoning effort", json!({"req": "set_reasoning_effort", "effort": effort.as_cli()}), ); } } Provider::Jcode => {} } cx.notify(); } pub fn set_fast_mode(&mut self, on: bool, cx: &mut Context) { if self.forward(ThreadAction::SetFastMode { on }) { self.fast_mode = on; cx.notify(); return; } if self.provider != Provider::Claude { return; } self.fast_mode = on; self.control( PendingControl::FlagSettings, json!({"subtype": "apply_flag_settings", "settings": {"fastMode": on}}), ); cx.notify(); } pub fn turn_expanded(&self, end: usize) -> bool { self.expanded_turns.contains(&end) } pub fn toggle_turn(&mut self, end: usize, cx: &mut Context) { if !self.expanded_turns.remove(&end) { self.expanded_turns.insert(end); } cx.notify(); } pub fn turn_changes(&self, end: usize) -> Option<&TurnChanges> { self.turn_changes.get(&end) } pub fn dir_expanded(&self, end: usize, dir: &str) -> bool { self.expanded_dirs.get(&end).is_some_and(|dirs| dirs.contains(dir)) } pub fn toggle_dir(&mut self, end: usize, dir: String, cx: &mut Context) { let dirs = self.expanded_dirs.entry(end).or_default(); if !dirs.remove(&dir) { dirs.insert(dir); } cx.notify(); } /// Opens every folder of a turn's change overview, or closes them all. pub fn set_dirs(&mut self, end: usize, dirs: HashSet, cx: &mut Context) { self.expanded_dirs.insert(end, dirs); cx.notify(); } /// The finished turn item `ix` belongs to, if its turn has ended. A turn /// runs from after a user message (or the previous turn's end) to the /// next `TurnEnd`. pub fn finished_turn(&self, ix: usize) -> Option { if matches!(self.items.get(ix)?, ChatItem::User { .. }) { return None; } let first = self.items[..ix] .iter() .rposition(|item| matches!(item, ChatItem::User { .. } | ChatItem::TurnEnd { .. })) .map_or(0, |boundary| boundary + 1); let end = ix + self.items[ix..] .iter() .position(|item| matches!(item, ChatItem::User { .. } | ChatItem::TurnEnd { .. }))?; let ChatItem::TurnEnd { duration_ms, cost, at } = self.items[end] else { return None; }; let answer = (first..end) .rev() .find(|&ix| matches!(self.items[ix], ChatItem::Assistant(_))); let has_work = (first..end).any(|ix| { Some(ix) != answer && !matches!(self.items[ix], ChatItem::Notice(NoticeKind::Error, _) | ChatItem::TaskEvent { .. }) }); Some(FinishedTurn { first, answer, end, duration_ms, cost, at, has_work }) } pub fn toggle_tool(&mut self, ix: usize, cx: &mut Context) { if let Some(ChatItem::Tool(tool)) = self.items.get_mut(ix) { tool.expanded = !tool.expanded; cx.notify(); } } pub fn respond_permission( &mut self, ix: usize, decision: PermissionDecision, cx: &mut Context, ) { let decision_key = match decision { PermissionDecision::Allow => "allow", PermissionDecision::AllowAlways => "always", PermissionDecision::Deny => "deny", }; if self.forward(ThreadAction::Permission { ix, decision: decision_key.into() }) { return; } let Some(process) = &self.process else { return }; let Some(ChatItem::Tool(tool)) = self.items.get_mut(ix) else { return }; let Some(prompt) = tool.permission.as_mut() else { return }; if prompt.state != PermissionState::Pending { return; } let response = match decision { PermissionDecision::Allow => json!({"behavior": "allow", "updatedInput": prompt.input}), PermissionDecision::AllowAlways => json!({ "behavior": "allow", "updatedInput": prompt.input, "updatedPermissions": prompt.suggestions, }), PermissionDecision::Deny => { json!({"behavior": "deny", "message": "The user declined this tool use."}) } }; process.respond_control(&prompt.request_id, response); prompt.state = match decision { PermissionDecision::Deny => PermissionState::Denied, _ => PermissionState::Allowed, }; if tool.detached { tool.state = match decision { PermissionDecision::Deny => ToolState::Error, _ => ToolState::Done, }; } if decision == PermissionDecision::AllowAlways { let new_mode = prompt.suggestions.iter().find_map(|suggestion| { (suggestion.get("type").and_then(Value::as_str) == Some("setMode")) .then(|| suggestion.get("mode").and_then(Value::as_str)) .flatten() .and_then(PermissionMode::from_cli) }); if let Some(mode) = new_mode { self.permission_mode = mode; } } cx.notify(); } /// The AskUserQuestion form waiting for answers, if any. pub fn pending_ask(&self) -> Option<(usize, &UserQuestions)> { self.items.iter().enumerate().find_map(|(ix, item)| match item { ChatItem::Tool(ToolCall { ask: Some(ask), permission: Some(prompt), .. }) if prompt.state == PermissionState::Pending => { Some((ix, ask)) } _ => None, }) } fn pending_ask_mut(&mut self, ix: usize) -> Option<&mut UserQuestions> { match self.items.get_mut(ix) { Some(ChatItem::Tool(ToolCall { ask: Some(ask), permission: Some(prompt), .. })) if prompt.state == PermissionState::Pending => { Some(ask) } _ => None, } } /// Shows another of the form's questions. pub fn ask_show(&mut self, ix: usize, question: usize, cx: &mut Context) { if let Some(ask) = self.pending_ask_mut(ix) && question < ask.questions.len() { ask.current = question; cx.notify(); } } /// Picks an option on the current question: answers a single-select /// question, or ticks the option on a multi-select one. pub fn ask_pick(&mut self, ix: usize, option: usize, cx: &mut Context) { let Some(ask) = self.pending_ask_mut(ix) else { return }; let question = &ask.questions[ask.current]; let Some(label) = question.options.get(option).map(|option| option.label.clone()) else { return }; if question.multi_select { let picked = &mut ask.picked[ask.current]; match picked.iter().position(|&picked| picked == option) { Some(position) => { picked.remove(position); } None => picked.push(option), } cx.notify(); return; } if ask.answer(label) { self.submit_ask(ix); } cx.notify(); } /// Commits the options ticked on the current multi-select question. pub fn ask_confirm(&mut self, ix: usize, cx: &mut Context) { let Some(ask) = self.pending_ask_mut(ix) else { return }; let question = &ask.questions[ask.current]; let mut picked = ask.picked[ask.current].clone(); picked.sort_unstable(); let labels: Vec<&str> = picked.iter().filter_map(|&option| question.options.get(option)).map(|option| option.label.as_ref()).collect(); if labels.is_empty() { return; } if ask.answer(labels.join(", ").into()) { self.submit_ask(ix); } cx.notify(); } /// Answers the current question in the user's own words. pub fn ask_custom(&mut self, ix: usize, text: String, cx: &mut Context) { let Some(ask) = self.pending_ask_mut(ix) else { return }; if ask.answer(text.into()) { self.submit_ask(ix); } cx.notify(); } /// Answers the whole form at once, as a client that filled it in asks. pub fn ask_submit(&mut self, ix: usize, answers: Vec, cx: &mut Context) { let Some(ask) = self.pending_ask_mut(ix) else { return }; if answers.len() != ask.questions.len() { return; } ask.answers = answers.into_iter().map(|answer| Some(answer.into())).collect(); self.submit_ask(ix); cx.notify(); } /// Dismisses the form without answering. pub fn ask_skip(&mut self, ix: usize, cx: &mut Context) { if self.forward(ThreadAction::AskSkip { ix }) { return; } let Some(process) = &self.process else { return }; let Some(ChatItem::Tool(tool)) = self.items.get_mut(ix) else { return }; let Some(prompt) = tool.permission.as_mut().filter(|prompt| prompt.state == PermissionState::Pending) else { return; }; process.respond_control( &prompt.request_id, json!({"behavior": "deny", "message": "The user dismissed the questions without answering."}), ); prompt.state = PermissionState::Denied; cx.notify(); } /// Sends the form's answers, keyed by question text as the CLI expects. fn submit_ask(&mut self, ix: usize) { if self.remote.is_some() { if let Some(ChatItem::Tool(ToolCall { ask: Some(ask), .. })) = self.items.get(ix) { let answers = ask.answers.iter().map(|answer| answer.as_ref().map(ToString::to_string).unwrap_or_default()); self.forward(ThreadAction::AskSubmit { ix, answers: answers.collect() }); } return; } let Some(process) = &self.process else { return }; let Some(ChatItem::Tool(ToolCall { ask: Some(ask), permission: Some(prompt), .. })) = self.items.get_mut(ix) else { return; }; let answers: serde_json::Map = ask .questions .iter() .zip(&ask.answers) .filter_map(|(question, answer)| Some((question.question.to_string(), answer.as_ref()?.to_string().into()))) .collect(); let questions = prompt.input.get("questions").cloned().unwrap_or_default(); process.respond_control( &prompt.request_id, json!({"behavior": "allow", "updatedInput": {"questions": questions, "answers": answers}}), ); prompt.state = PermissionState::Allowed; } /// Answers an extension's question with one of its options. pub fn answer_question(&mut self, ix: usize, option: usize, cx: &mut Context) { if self.forward(ThreadAction::Answer { ix, option }) { return; } let Some(process) = &self.process else { return }; let Some(ChatItem::Question(question)) = self.items.get_mut(ix) else { return }; let Some(answer) = question.options.get(option).cloned() else { return }; if question.answer.is_some() { return; } let response = if question.confirm { json!({"type": "extension_ui_response", "id": question.request_id, "confirmed": option == 0}) } else { json!({"type": "extension_ui_response", "id": question.request_id, "value": answer.as_ref()}) }; process.send(response); question.answer = Some(answer); cx.notify(); } pub fn shutdown(&mut self) { self.process.take(); } // ---- protocol handling ------------------------------------------- fn handle_event(&mut self, event: ProcessEvent, cx: &mut Context) { match event { ProcessEvent::Message(message) => match self.provider { Provider::Claude => self.handle_message(message, cx), Provider::Pi => self.handle_pi_message(message, cx), Provider::Jcode => self.handle_jcode_message(message, cx), }, ProcessEvent::Stderr(line) => { if self.stderr_tail.len() == STDERR_TAIL { self.stderr_tail.pop_front(); } self.stderr_tail.push_back(line); } ProcessEvent::Exited(code) => self.handle_exit(code, cx), } } fn handle_exit(&mut self, code: Option, cx: &mut Context) { self.flush_text(cx); self.process = None; self.drop_background_tasks(); self.thinking = false; self.streaming = None; self.status = SessionStatus::Exited; self.turn_started = None; self.base_snapshot = None; self.finish_open_tools("The agent exited before the tool finished."); let agent = self.provider.agent_name(); let mut text = match code { Some(code) => format!("{agent} exited with status {code}."), None => format!("{agent} was terminated."), }; if !self.stderr_tail.is_empty() { text.push_str("\n\n"); text.push_str(&Vec::from(self.stderr_tail.clone()).join("\n")); } text.push_str("\n\nSend a message to restart it."); self.push_notice(NoticeKind::Error, text); } /// Closes the running turn: records its end and, in the background, what /// it changed on disk. fn finish_turn(&mut self, duration_ms: u64, cost: Option, cx: &mut Context) { self.flush_text(cx); self.turn_started = None; let now = SystemTime::now(); self.updated = now; self.items.push(ChatItem::TurnEnd { duration_ms, cost, at: now }); let end = self.items.len() - 1; let Some(base) = self.base_snapshot.take() else { return }; let cwd = self.cwd.clone(); let index = self.snapshot_index.clone(); cx.spawn(async move |this, cx| { let Some(base) = base.await else { return }; let changes = cx .background_executor() .spawn(async move { let head = git::snapshot(&cwd, &index)?; git::changes_between(&cwd, &base, &head) }) .await; if let Some(changes) = changes.filter(|changes| !changes.files.is_empty()) { let _ = this.update(cx, |this, cx| { this.turn_changes.insert(end, changes); cx.emit(SessionEvent::TurnChanges(end)); cx.notify(); }); } }) .detach(); } // ---- Claude Code (stream-json) ------------------------------------- fn handle_message(&mut self, message: Value, cx: &mut Context) { if let Some(session_id) = message.get("session_id").and_then(Value::as_str) && self.resume_id.as_deref() != Some(session_id) { self.resume_id = Some(session_id.to_string()); } let parent_tool_use_id = message.get("parent_tool_use_id").and_then(Value::as_str); match message.get("type").and_then(Value::as_str) { Some("system") => self.handle_system(&message, cx), Some("stream_event") if parent_tool_use_id.is_none() => { self.handle_stream_event(&message["event"], cx) } Some("assistant") => match parent_tool_use_id { Some(parent) => self.count_subagent_tools(parent, &message["message"]), None => { self.record_usage(&message["message"]); self.handle_assistant(&message["message"], cx) } }, Some("user") if parent_tool_use_id.is_none() => self.handle_user(&message["message"]), Some("result") => self.handle_result(&message, cx), Some("rate_limit_event") => self.handle_rate_limit(&message["rate_limit_info"]), Some("control_request") => self.handle_control_request(&message), Some("control_response") => self.handle_control_response(&message["response"], cx), Some("control_cancel_request") => { if let Some(request_id) = message.get("request_id").and_then(Value::as_str) { self.cancel_permission(request_id); } } _ => {} } } /// Tracks how full the context window is from a response's token usage. fn record_usage(&mut self, message: &Value) { let Some(usage) = message.get("usage") else { return }; let tokens: u64 = ["input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens"] .iter() .filter_map(|key| usage.get(*key).and_then(Value::as_u64)) .sum(); if tokens > 0 { self.context_tokens = Some(tokens); } } fn handle_system(&mut self, message: &Value, cx: &mut Context) { if let Some(mode) = message .get("permissionMode") .and_then(Value::as_str) .and_then(PermissionMode::from_cli) { self.permission_mode = mode; } let text = |key: &str| message.get(key).and_then(Value::as_str); match text("subtype") { Some("init") => { if let Some(model) = message.get("model").and_then(Value::as_str) { self.active_model = Some(model.to_string()); } // Each turn opens with `init`; one we didn't start was woken // by a background task. if self.status == SessionStatus::Idle { self.begin_turn(cx); } } Some("task_started") => { let Some(task_id) = text("task_id") else { return }; let tool_use_id = text("tool_use_id").map(str::to_string); self.task_origins.insert(task_id.to_string(), (Instant::now(), tool_use_id.clone())); let backgrounded = message.get("is_backgrounded").and_then(Value::as_bool).unwrap_or(false); if let Some(tool) = self.find_tool_mut(tool_use_id.as_deref().unwrap_or_default()) { tool.task_id = Some(task_id.to_string()); if backgrounded { tool.background = Some(TaskStatus::Running); } } } Some("task_updated") => { let backgrounded = message.pointer("/patch/is_backgrounded").and_then(Value::as_bool) == Some(true); let tool_use_id = text("task_id") .and_then(|task_id| self.task_origins.get(task_id)) .and_then(|(_, tool_use_id)| tool_use_id.clone()); if backgrounded && let Some(tool) = self.find_tool_mut(tool_use_id.as_deref().unwrap_or_default()) { tool.background = Some(TaskStatus::Running); } } // The full list of background tasks, whenever it changes. Some("background_tasks_changed") => { let tasks = message.get("tasks").and_then(Value::as_array).cloned().unwrap_or_default(); self.background_tasks = tasks .iter() .filter_map(|task| { let task_id = task.get("task_id").and_then(Value::as_str)?.to_string(); let started = self.task_origins.get(&task_id).map_or_else(Instant::now, |(started, _)| *started); let description = task.get("description").and_then(Value::as_str).unwrap_or("Background task"); Some(BackgroundTask { task_id, description: description.to_string().into(), started, }) }) .collect(); } Some("task_notification") => { let status = TaskStatus::from_cli(text("status").unwrap_or_default()); // A monitor reports events while it keeps running. let finished = status != TaskStatus::Running; let mut backgrounded = false; if let Some(task_id) = text("task_id") { backgrounded = self.background_tasks.iter().any(|task| task.task_id == task_id); if finished { self.task_origins.remove(task_id); self.background_tasks.retain(|task| task.task_id != task_id); } } if let Some(tool) = self.find_tool_mut(text("tool_use_id").unwrap_or_default()) { backgrounded |= tool.background.is_some(); if tool.background.is_some() { tool.background = Some(status); } } // Long foreground commands report finishing too; only // background tasks wake Claude. if !backgrounded { return; } let summary = text("summary").unwrap_or("A background task finished.").to_string(); self.items.push(ChatItem::TaskEvent { summary: summary.into(), status }); self.updated = SystemTime::now(); } Some("compact_boundary") => { self.push_notice(NoticeKind::Info, "Conversation compacted to free up context.") } _ => {} } } fn handle_stream_event(&mut self, event: &Value, cx: &mut Context) { let index = event.get("index").and_then(Value::as_u64).unwrap_or(0); match event.get("type").and_then(Value::as_str) { Some("message_start") => { if let Some(id) = event.pointer("/message/id").and_then(Value::as_str) { self.streamed_message_ids.insert(id.to_string()); } self.streaming = Some(StreamingMessage { blocks: HashMap::new() }); } Some("content_block_start") => { let block = &event["content_block"]; let item_ix = match block.get("type").and_then(Value::as_str) { Some("text") => { let text = block.get("text").and_then(Value::as_str).unwrap_or_default(); Some(self.push_assistant_text(text, cx)) } Some("tool_use") => Some(self.push_tool( block.get("id").and_then(Value::as_str).unwrap_or_default(), block.get("name").and_then(Value::as_str).unwrap_or("Tool"), block.get("input").cloned().unwrap_or(Value::Null), )), Some("thinking" | "redacted_thinking") => { self.thinking = true; None } _ => None, }; if let (Some(item_ix), Some(streaming)) = (item_ix, self.streaming.as_mut()) { streaming.blocks.insert(index, item_ix); } } Some("content_block_delta") => { let Some(item_ix) = self.streamed_item(index) else { return }; let delta = &event["delta"]; match (delta.get("type").and_then(Value::as_str), self.items.get_mut(item_ix)) { (Some("text_delta"), Some(ChatItem::Assistant(_))) => { let text = delta.get("text").and_then(Value::as_str).unwrap_or_default(); self.append_text(item_ix, text, cx); } (Some("input_json_delta"), Some(ChatItem::Tool(tool))) => { let json = delta.get("partial_json").and_then(Value::as_str); tool.partial_input.push_str(json.unwrap_or_default()); } _ => {} } } Some("content_block_stop") => { self.thinking = false; if let Some(ix) = self.streamed_item(index) { self.flush_item(ix, cx); } if let Some(ChatItem::Tool(tool)) = self.streamed_item(index).and_then(|ix| self.items.get_mut(ix)) && let Ok(input) = serde_json::from_str::(&tool.partial_input) { tool.input = input; } } Some("message_stop") => { self.thinking = false; self.streaming = None; } _ => {} } } fn streamed_item(&self, index: u64) -> Option { self.streaming.as_ref().and_then(|streaming| streaming.blocks.get(&index).copied()) } /// Complete assistant messages. Content we already streamed is skipped; /// they only carry the authoritative tool inputs in that case. fn handle_assistant(&mut self, message: &Value, cx: &mut Context) { let already_streamed = message .get("id") .and_then(Value::as_str) .is_some_and(|id| self.streamed_message_ids.contains(id)); let Some(blocks) = message.get("content").and_then(Value::as_array) else { return }; for block in blocks { match block.get("type").and_then(Value::as_str) { Some("text") if !already_streamed => { let text = block.get("text").and_then(Value::as_str).unwrap_or_default(); if !text.trim().is_empty() { self.push_assistant_text(text, cx); } } Some("tool_use") => { let id = block.get("id").and_then(Value::as_str).unwrap_or_default(); match self.find_tool_mut(id) { Some(tool) => tool.input = block["input"].clone(), None => { self.push_tool( id, block.get("name").and_then(Value::as_str).unwrap_or("Tool"), block["input"].clone(), ); } } } _ => {} } } } fn count_subagent_tools(&mut self, parent: &str, message: &Value) { let count = message .get("content") .and_then(Value::as_array) .map(|blocks| { blocks .iter() .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use")) .count() }) .unwrap_or(0); if let Some(tool) = self.find_tool_mut(parent) { tool.subagent_tool_count += count; } } fn handle_user(&mut self, message: &Value) { let content = match message.get("content") { Some(Value::Array(blocks)) => blocks.clone(), Some(Value::String(text)) => vec![json!({"type": "text", "text": text})], _ => return, }; for block in content { match block.get("type").and_then(Value::as_str) { Some("tool_result") => { let id = block.get("tool_use_id").and_then(Value::as_str).unwrap_or_default(); let is_error = block.get("is_error").and_then(Value::as_bool).unwrap_or(false); let text = tool_result_text(&block["content"]); if let Some(tool) = self.find_tool_mut(id) { tool.result = Some(text); tool.state = if is_error { ToolState::Error } else { ToolState::Done }; if let Some(prompt) = tool.permission.as_mut() && prompt.state == PermissionState::Pending { prompt.state = PermissionState::Cancelled; } } } Some("text") => { let text = block.get("text").and_then(Value::as_str).unwrap_or_default(); if text.starts_with("[Request interrupted") { self.push_notice(NoticeKind::Info, "Interrupted."); } } _ => {} } } } /// Notes when a usage limit that refused this thread resets. fn handle_rate_limit(&mut self, info: &Value) { match info.get("status").and_then(Value::as_str) { Some("rejected") => { self.limit_resets = info.get("resetsAt").and_then(Value::as_f64).map(|resets| { // Seconds since the epoch; milliseconds just in case. let secs = if resets > 1e12 { resets / 1000. } else { resets }; UNIX_EPOCH + Duration::from_secs_f64(secs.max(0.)) }); } Some("allowed") => self.limit_resets = None, _ => {} } } /// After a turn a usage limit stopped: continue or snooze the thread /// once the limit resets, as the settings say. fn schedule_after_limit(&mut self) { let Some(resets) = self.limited_until() else { return }; let (resume, snooze) = settings::get(|settings| (settings.auto_resume_limited, settings.snooze_limited)); if resume { // A minute's grace, so the limit has surely lifted. self.resume_at = Some(resets + Duration::from_secs(60)); } if snooze { self.snoozed_until = Some(resets); } } fn handle_result(&mut self, message: &Value, cx: &mut Context) { self.thinking = false; self.streaming = None; self.status = SessionStatus::Idle; self.finish_open_tools("The turn ended before the tool finished."); self.cancel_pending_permissions(); let is_error = message.get("is_error").and_then(Value::as_bool).unwrap_or(false); let subtype = message.get("subtype").and_then(Value::as_str).unwrap_or_default(); let interrupted = std::mem::take(&mut self.interrupt_requested); if (is_error || subtype != "success") && !interrupted { let detail = message .get("result") .and_then(Value::as_str) .filter(|text| !text.is_empty()) .map(str::to_string) .unwrap_or_else(|| subtype.replace('_', " ")); self.push_notice(NoticeKind::Error, detail); } // `total_cost_usd` is cumulative for the CLI process. let cost = message.get("total_cost_usd").and_then(Value::as_f64).map(|total| { let delta = (total - self.process_cost).max(0.); self.process_cost = total; self.total_cost += delta; delta }); let duration_ms = message .get("duration_ms") .and_then(Value::as_u64) .or_else(|| self.turn_started.map(|start| start.elapsed().as_millis() as u64)) .unwrap_or(0); if let Some(models) = message.get("modelUsage").and_then(Value::as_object) { let window = self .active_model .as_deref() .and_then(|model| models.get(model)) .into_iter() .chain(models.values()) .find_map(|usage| usage.get("contextWindow").and_then(Value::as_u64)); if window.is_some() { self.context_window = window; } } self.finish_turn(duration_ms, cost, cx); self.schedule_after_limit(); self.send_next_queued(cx); } fn handle_control_request(&mut self, message: &Value) { let request_id = message.get("request_id").and_then(Value::as_str).unwrap_or_default(); let request = &message["request"]; if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") { // We don't implement hooks or SDK MCP servers; say so instead of // leaving the CLI waiting forever. if let Some(process) = &self.process { process.send(json!({ "type": "control_response", "response": { "subtype": "error", "request_id": request_id, "error": "Unsupported control request", }, })); } return; } let prompt = PermissionPrompt { request_id: request_id.to_string(), input: request["input"].clone(), suggestions: request .get("permission_suggestions") .and_then(Value::as_array) .cloned() .unwrap_or_default(), state: PermissionState::Pending, }; self.updated = SystemTime::now(); let name = request.get("tool_name").and_then(Value::as_str).unwrap_or("Tool"); let ask = (name == ASK_USER_QUESTION).then(|| UserQuestions::parse(&prompt.input)).flatten(); let tool_use_id = request.get("tool_use_id").and_then(Value::as_str).unwrap_or_default(); if let Some(tool) = self.find_tool_mut(tool_use_id) { tool.input = prompt.input.clone(); tool.permission = Some(prompt); tool.ask = ask; return; } let ix = self.push_tool(tool_use_id, name, prompt.input.clone()); if let Some(ChatItem::Tool(tool)) = self.items.get_mut(ix) { tool.permission = Some(prompt); tool.detached = true; tool.ask = ask; } } fn handle_control_response(&mut self, response: &Value, cx: &mut Context) { let request_id = response.get("request_id").and_then(Value::as_str).unwrap_or_default(); let Some(pending) = self.pending_controls.remove(request_id) else { return }; if response.get("subtype").and_then(Value::as_str) == Some("error") { let error = response.get("error").and_then(Value::as_str).unwrap_or("unknown error"); let action = match pending { PendingControl::Initialize => "start Claude Code", PendingControl::SetModel => "switch model", PendingControl::SetPermissionMode => "change permission mode", PendingControl::FlagSettings => "change reasoning or fast mode", PendingControl::Interrupt => "interrupt", PendingControl::Background => "move the command to the background", PendingControl::StopTask => "stop the background task", }; self.push_notice(NoticeKind::Error, format!("Failed to {action}: {error}")); return; } if let PendingControl::Initialize = pending { let info = &response["response"]; Catalog::set_claude_info(&self.instance, info, cx); if let Some(mode) = info .get("current_permission_mode") .and_then(Value::as_str) .and_then(PermissionMode::from_cli) { self.permission_mode = mode; } if self.status == SessionStatus::Starting { self.status = SessionStatus::Idle; } } } // ---- Pi (JSONL RPC) ----------------------------------------------- fn handle_pi_message(&mut self, message: Value, cx: &mut Context) { match message.get("type").and_then(Value::as_str) { Some("response") => self.handle_pi_response(&message), Some("agent_start") => { self.status = SessionStatus::Working; self.turn_started.get_or_insert_with(Instant::now); } Some("message_start") if message.pointer("/message/role").and_then(Value::as_str) == Some("assistant") => { self.streaming = Some(StreamingMessage { blocks: HashMap::new() }); } Some("message_update") => self.handle_pi_update(&message["assistantMessageEvent"], cx), Some("message_end") if message.pointer("/message/role").and_then(Value::as_str) == Some("assistant") => { let message = &message["message"]; self.thinking = false; self.streaming = None; if let Some(usage) = message.get("usage") { let tokens: u64 = ["input", "cacheRead", "cacheWrite", "output"] .iter() .filter_map(|key| usage.get(*key).and_then(Value::as_u64)) .sum(); if tokens > 0 { self.context_tokens = Some(tokens); } let cost = usage.pointer("/cost/total").and_then(Value::as_f64).unwrap_or(0.); self.turn_cost += cost; self.total_cost += cost; } match message.get("stopReason").and_then(Value::as_str) { Some("error") if !self.interrupt_requested => { let error = message.get("errorMessage").and_then(Value::as_str).unwrap_or("The model returned an error."); self.push_notice(NoticeKind::Error, error.to_string()); } Some("aborted") => self.push_notice(NoticeKind::Info, "Interrupted."), _ => {} } } Some("tool_execution_start") => { let id = message.get("toolCallId").and_then(Value::as_str).unwrap_or_default(); if self.find_tool_mut(id).is_none() { self.push_tool( id, message.get("toolName").and_then(Value::as_str).unwrap_or("tool"), message.get("args").cloned().unwrap_or(Value::Null), ); } } Some("tool_execution_end") => { let id = message.get("toolCallId").and_then(Value::as_str).unwrap_or_default(); let is_error = message.get("isError").and_then(Value::as_bool).unwrap_or(false); let text = tool_result_text(message.pointer("/result/content").unwrap_or(&Value::Null)); if let Some(tool) = self.find_tool_mut(id) { tool.result = Some(text); tool.state = if is_error { ToolState::Error } else { ToolState::Done }; } } Some("agent_settled") => { self.thinking = false; self.streaming = None; self.status = SessionStatus::Idle; self.interrupt_requested = false; self.finish_open_tools("The turn ended before the tool finished."); self.cancel_pending_questions(); let duration_ms = self.turn_started.map_or(0, |start| start.elapsed().as_millis() as u64); let cost = Some(std::mem::take(&mut self.turn_cost)).filter(|cost| *cost > 0.); self.finish_turn(duration_ms, cost, cx); self.send_next_queued(cx); } Some("auto_retry_start") => { let attempt = message.get("attempt").and_then(Value::as_u64).unwrap_or(1); let max = message.get("maxAttempts").and_then(Value::as_u64).unwrap_or(attempt); let error = message.get("errorMessage").and_then(Value::as_str).unwrap_or_default(); self.push_notice(NoticeKind::Info, format!("Retrying ({attempt}/{max}) after: {error}")); } Some("compaction_start") => { self.push_notice(NoticeKind::Info, "Compacting the conversation to free up context.") } Some("thinking_level_changed") => { if let Some(effort) = message.get("level").and_then(Value::as_str).and_then(Effort::from_cli) { self.reasoning = Reasoning::Level(effort); } } Some("extension_ui_request") => self.handle_pi_ui(&message), Some("extension_error") => { let error = message.get("error").and_then(Value::as_str).unwrap_or("unknown error"); self.push_notice(NoticeKind::Error, format!("A Pi extension failed: {error}")); } _ => {} } } fn handle_pi_response(&mut self, message: &Value) { let command = message.get("command").and_then(Value::as_str).unwrap_or_default(); if message.get("success").and_then(Value::as_bool) != Some(true) { let error = message.get("error").and_then(Value::as_str).unwrap_or("unknown error"); if command == "prompt" { // The prompt never started a run, so no `agent_settled` will come. self.status = SessionStatus::Idle; self.turn_started = None; self.base_snapshot = None; } self.push_notice(NoticeKind::Error, format!("Pi couldn't {}: {error}", command.replace('_', " "))); return; } let data = &message["data"]; match command { "get_state" => { self.resume_id = data .get("sessionFile") .or_else(|| data.get("sessionId")) .and_then(Value::as_str) .map(str::to_string); self.apply_pi_model(&data["model"]); // Pi always runs at some level; show which. if let Some(effort) = data.get("thinkingLevel").and_then(Value::as_str).and_then(Effort::from_cli) { self.reasoning = Reasoning::Level(effort); } if self.status == SessionStatus::Starting { self.status = SessionStatus::Idle; } } "set_model" => self.apply_pi_model(data), "get_available_thinking_levels" => { self.pi_efforts = data .get("levels") .and_then(Value::as_array) .map(|levels| levels.iter().filter_map(Value::as_str).filter_map(Effort::from_cli).collect()) .unwrap_or_default(); } "prompt" if data.get("disposition").and_then(Value::as_str) == Some("handled") => { // An extension command consumed the prompt; no run follows. self.status = SessionStatus::Idle; self.turn_started = None; self.base_snapshot = None; } _ => {} } } fn apply_pi_model(&mut self, model: &Value) { if let (Some(provider), Some(id)) = (model.get("provider").and_then(Value::as_str), model.get("id").and_then(Value::as_str)) { self.active_model = Some(format!("{provider}/{id}")); if self.model.is_none() { self.model = self.active_model.clone(); } } if let Some(window) = model.get("contextWindow").and_then(Value::as_u64) { self.context_window = Some(window); } } fn handle_pi_update(&mut self, event: &Value, cx: &mut Context) { let index = event.get("contentIndex").and_then(Value::as_u64).unwrap_or(0); if self.streaming.is_none() { self.streaming = Some(StreamingMessage { blocks: HashMap::new() }); } match event.get("type").and_then(Value::as_str) { Some("text_start") => { let ix = self.push_assistant_text("", cx); self.map_block(index, ix); } Some("text_delta") => { let delta = event.get("delta").and_then(Value::as_str).unwrap_or_default(); let ix = match self.streamed_item(index) { Some(ix) => ix, None => { let ix = self.push_assistant_text("", cx); self.map_block(index, ix); ix } }; self.append_text(ix, delta, cx); } Some("text_end") => { if let Some(ix) = self.streamed_item(index) { self.flush_item(ix, cx); } } Some("thinking_start" | "thinking_delta") => self.thinking = true, Some("thinking_end") => self.thinking = false, Some("toolcall_start") => { let id = event.get("id").and_then(Value::as_str).unwrap_or_default(); let name = event.get("toolName").and_then(Value::as_str).unwrap_or("tool"); let ix = self.push_tool(id, name, Value::Null); self.map_block(index, ix); } Some("toolcall_delta") => { let delta = event.get("delta").and_then(Value::as_str).unwrap_or_default(); if let Some(ChatItem::Tool(tool)) = self.streamed_item(index).and_then(|ix| self.items.get_mut(ix)) { tool.partial_input.push_str(delta); } } Some("toolcall_end") => { let call = &event["toolCall"]; let id = call.get("id").and_then(Value::as_str).unwrap_or_default(); let arguments = call.get("arguments").cloned().unwrap_or(Value::Null); match self.find_tool_mut(id) { Some(tool) => tool.input = arguments, None => { let name = call.get("name").and_then(Value::as_str).unwrap_or("tool"); self.push_tool(id, name, arguments); } } } _ => {} } } fn map_block(&mut self, index: u64, ix: usize) { if let Some(streaming) = self.streaming.as_mut() { streaming.blocks.insert(index, ix); } } /// Dialogs Pi extensions open. Confirms and selects become questions in /// the transcript; free-text input isn't supported yet and is declined. fn handle_pi_ui(&mut self, message: &Value) { let id = message.get("id").and_then(Value::as_str).unwrap_or_default().to_string(); let text = |key: &str| message.get(key).and_then(Value::as_str).map(|text| SharedString::from(text.to_string())); match message.get("method").and_then(Value::as_str) { Some("confirm") => self.items.push(ChatItem::Question(Question { request_id: id, title: text("title").unwrap_or_else(|| "Confirm".into()), message: text("message"), options: vec!["Yes".into(), "No".into()], confirm: true, answer: None, })), Some("select") => self.items.push(ChatItem::Question(Question { request_id: id, title: text("title").unwrap_or_else(|| "Choose".into()), message: None, options: message .get("options") .and_then(Value::as_array) .map(|options| { options.iter().filter_map(Value::as_str).map(|option| option.to_string().into()).collect() }) .unwrap_or_default(), confirm: false, answer: None, })), Some("input" | "editor") => { if let Some(process) = &self.process { process.send(json!({"type": "extension_ui_response", "id": id, "cancelled": true})); } let title = text("title").unwrap_or_default(); self.push_notice( NoticeKind::Info, format!("A Pi extension asked for text ({title}); that isn't supported here yet."), ); } Some("notify") => { let kind = match message.get("notifyType").and_then(Value::as_str) { Some("error") => NoticeKind::Error, _ => NoticeKind::Info, }; if let Some(text) = text("message") { self.push_notice(kind, text); } } _ => {} } self.updated = SystemTime::now(); } // ---- jcode (harness API) ------------------------------------------- fn handle_jcode_message(&mut self, message: Value, cx: &mut Context) { let reply_to = message.get("reply_to").and_then(Value::as_u64); let request = reply_to.and_then(|id| self.jcode.requests.remove(&id)); let text = |key: &str| message.get(key).and_then(Value::as_str).unwrap_or_default(); match text("ev") { "attached" => { let Some(session_id) = message.pointer("/session/session_id").and_then(Value::as_str) else { return }; self.jcode.session = Some(session_id.to_string()); self.resume_id = Some(session_id.to_string()); if self.status == SessionStatus::Starting { self.status = SessionStatus::Idle; } for (what, request) in std::mem::take(&mut self.jcode.held) { self.jcode_request(what, request); } } "error" => { let error = text("message"); match request { // The session is gone from jcode's store; carry on in a new one. Some("attach") => { self.push_notice( NoticeKind::Info, "jcode no longer has this thread's session, so it continues in a new one.", ); self.jcode_create_session(); } Some(what) => { if what == "send the message" { // The message never started a turn, so none will end. self.status = SessionStatus::Idle; self.turn_started = None; self.base_snapshot = None; } self.push_notice(NoticeKind::Error, format!("jcode couldn't {what}: {error}")); } None => self.push_notice(NoticeKind::Error, format!("jcode: {error}")), } } "model_info" => { if let Some(model) = message.get("model").and_then(Value::as_str) { self.active_model = Some(model.to_string()); } self.jcode.provider = message.get("provider").and_then(Value::as_str).map(str::to_string); } "message_accepted" => self.begin_turn(cx), "soft_interrupt_injected" if text("point") == "turn_start" => self.begin_turn(cx), "text_delta" => { self.begin_turn(cx); self.thinking = false; let ix = self.jcode_text_item(text("message_id"), cx); self.append_text(ix, text("text"), cx); } // A rewrite of what was streamed; empty takes the message back. "text_replace" => { let ix = self.jcode_text_item(text("message_id"), cx); if let Some(ChatItem::Assistant(answer)) = self.items.get_mut(ix) { answer.text = text("text").to_string(); answer.shown = answer.text.len(); let replacement = answer.text.clone(); answer.view.update(cx, |view, cx| view.set_text(&replacement, cx)); } } "text_done" => { if let Some(&ix) = self.jcode.texts.get(text("message_id")) { self.flush_item(ix, cx); } } "reasoning_delta" => { self.begin_turn(cx); self.thinking = true; } "reasoning_done" => self.thinking = false, "tool_start" => { self.begin_turn(cx); self.thinking = false; let id = text("call_id"); if self.find_tool_mut(id).is_none() { self.push_tool(id, text("name"), Value::Null); } } "tool_input_delta" => { let delta = text("delta"); if let Some(tool) = self.jcode_tool(text("call_id")) { tool.partial_input.push_str(delta); } } "tool_exec" => { if let Some(tool) = self.jcode_tool(text("call_id")) { take_partial_input(tool); } } "tool_done" => { let output = text("output").to_string(); let error = message.get("error").and_then(Value::as_str).map(str::to_string); let tool = match self.find_tool_mut(text("call_id")) { Some(tool) => tool, None => { let ix = self.push_tool(text("call_id"), text("name"), Value::Null); let Some(ChatItem::Tool(tool)) = self.items.get_mut(ix) else { return }; tool } }; take_partial_input(tool); tool.state = if error.is_some() { ToolState::Error } else { ToolState::Done }; tool.result = Some(match error { Some(error) if output.trim().is_empty() => error, Some(error) => format!("{output}\n\n{error}"), None => output, }); } "token_usage" => { let count = |key: &str| message.get(key).and_then(Value::as_u64).unwrap_or(0); // Anthropic counts cache reads apart from the input; OpenAI // includes them. let anthropic = self .jcode .provider .as_deref() .is_some_and(|provider| provider.to_lowercase().contains("anthropic") || provider.to_lowercase().contains("claude")); let cache_read = if anthropic { count("cache_read_input") } else { 0 }; let tokens = count("input") + count("output") + count("cache_creation_input") + cache_read; if tokens > 0 { self.context_tokens = Some(tokens); } } "connection_phase" if text("phase").starts_with("retrying") => { self.push_notice(NoticeKind::Info, format!("jcode is {}.", text("phase"))); } "compacted" => { let note = text("message"); self.push_notice( NoticeKind::Info, if note.is_empty() { "Compacting the conversation to free up context." } else { note }.to_string(), ); } "turn_stopped" => match text("reason") { "interrupted" => self.push_notice(NoticeKind::Info, "Interrupted."), _ if self.interrupt_requested => {} _ => { let error = text("message"); self.push_notice( NoticeKind::Error, if error.is_empty() { "The turn stopped early." } else { error }.to_string(), ); } }, "turn_done" => { self.thinking = false; self.status = SessionStatus::Idle; self.interrupt_requested = false; self.jcode.texts.clear(); self.finish_open_tools("The turn ended before the tool finished."); let duration_ms = self.turn_started.map_or(0, |start| start.elapsed().as_millis() as u64); self.finish_turn(duration_ms, None, cx); self.send_next_queued(cx); } _ => {} } } /// The chat item a streamed jcode message renders into, made on its first delta. fn jcode_text_item(&mut self, message_id: &str, cx: &mut Context) -> usize { if let Some(&ix) = self.jcode.texts.get(message_id) { return ix; } let ix = self.push_assistant_text("", cx); self.jcode.texts.insert(message_id.to_string(), ix); ix } /// A jcode tool call by id; older daemons leave the id out of input /// deltas, which then belong to the latest call. fn jcode_tool(&mut self, call_id: &str) -> Option<&mut ToolCall> { if !call_id.is_empty() { return self.find_tool_mut(call_id); } self.items.iter_mut().rev().find_map(|item| match item { ChatItem::Tool(tool) if tool.state == ToolState::Running => Some(tool), _ => None, }) } // ---- helpers ----------------------------------------------------- fn push_assistant_text(&mut self, text: &str, cx: &mut Context) -> usize { let view = cx.new(|cx| TextViewState::markdown(text, cx).selectable(true)); self.items.push(ChatItem::Assistant(AssistantText { text: text.to_string(), view, shown: text.len() })); self.items.len() - 1 } /// Adds streamed text to an answer. With "Show finished paragraphs" on, /// the view only takes whole paragraphs and code blocks. fn append_text(&mut self, ix: usize, delta: &str, cx: &mut Context) { let paragraphs = settings::get(|settings| settings.streaming == Streaming::Paragraphs); let Some(ChatItem::Assistant(answer)) = self.items.get_mut(ix) else { return }; answer.text.push_str(delta); let target = if paragraphs { finished_prefix(&answer.text) } else { answer.text.len() }; answer.show(target, cx); } /// Shows the rest of an answer's text. fn flush_item(&mut self, ix: usize, cx: &mut Context) { if let Some(ChatItem::Assistant(answer)) = self.items.get_mut(ix) { let end = answer.text.len(); answer.show(end, cx); } } fn flush_text(&mut self, cx: &mut Context) { for ix in 0..self.items.len() { self.flush_item(ix, cx); } } fn push_tool(&mut self, id: &str, name: &str, input: Value) -> usize { self.items.push(ChatItem::Tool(ToolCall { id: id.to_string(), name: name.to_string(), input, partial_input: String::new(), result: None, state: ToolState::Running, permission: None, expanded: false, subagent_tool_count: 0, detached: false, ask: None, task_id: None, background: None, })); self.items.len() - 1 } fn push_notice(&mut self, kind: NoticeKind, text: impl Into) { self.items.push(ChatItem::Notice(kind, text.into())); } fn find_tool_mut(&mut self, id: &str) -> Option<&mut ToolCall> { if id.is_empty() { return None; } self.items.iter_mut().rev().find_map(|item| match item { ChatItem::Tool(tool) if tool.id == id => Some(tool), _ => None, }) } fn cancel_permission(&mut self, request_id: &str) { for item in &mut self.items { if let ChatItem::Tool(ToolCall { permission: Some(prompt), .. }) = item && prompt.request_id == request_id && prompt.state == PermissionState::Pending { prompt.state = PermissionState::Cancelled; } } } fn cancel_pending_permissions(&mut self) { for item in &mut self.items { if let ChatItem::Tool(ToolCall { permission: Some(prompt), .. }) = item && prompt.state == PermissionState::Pending { prompt.state = PermissionState::Cancelled; } } } fn cancel_pending_questions(&mut self) { for item in &mut self.items { if let ChatItem::Question(question) = item && question.answer.is_none() { question.answer = Some("No answer".into()); } } } /// Background tasks die with the agent process. fn drop_background_tasks(&mut self) { self.background_tasks.clear(); self.task_origins.clear(); for item in &mut self.items { if let ChatItem::Tool(tool) = item && tool.background == Some(TaskStatus::Running) { tool.background = Some(TaskStatus::Stopped); } } } fn finish_open_tools(&mut self, reason: &str) { for item in &mut self.items { if let ChatItem::Tool(tool) = item && tool.state == ToolState::Running { tool.state = ToolState::Error; tool.result.get_or_insert_with(|| reason.to_string()); } } } } // ---- saving and restoring ---------------------------------------------- /// A thread as it's kept on disk between runs of the app. #[derive(Serialize, Deserialize)] pub struct SavedThread { pub id: String, pub instance: String, pub provider: Provider, pub title: String, #[serde(default)] pub title_custom: bool, pub project: PathBuf, pub cwd: PathBuf, #[serde(default)] pub worktree: Option, #[serde(default)] pub pending_worktree: Option, #[serde(default)] pub pull_request: Option, pub resume_id: Option, pub model: Option, pub active_model: Option, pub permission_mode: String, pub reasoning: String, pub fast_mode: bool, pub total_cost: f64, pub context_tokens: Option, pub context_window: Option, /// Milliseconds since the epoch, like every time below. pub created: u64, pub updated: u64, pub last_message: u64, pub settled: bool, pub pinned: bool, pub snoozed_until: Option, pub limit_resets: Option, pub resume_at: Option, pub queue: Vec, /// The agent was mid-turn when the thread was saved. pub working: bool, pub items: Vec, /// What each finished turn changed, by its `TurnEnd` index. pub turn_changes: Vec<(usize, TurnChanges)>, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum SavedItem { User { text: String, attachments: Vec, at: u64 }, Assistant { text: String }, Tool { id: String, name: String, input: Value, result: Option, /// `done`, `running` or `error`; a restored thread counts a running /// tool as failed. state: String, /// `pending`, `allowed`, `denied` or `cancelled`, if it asked for /// permission. A restored thread treats `pending` as cancelled. permission: Option, /// The rules a pending prompt offers to remember. #[serde(default, skip_serializing_if = "Vec::is_empty")] suggestions: Vec, subagent_tool_count: usize, /// An AskUserQuestion form's answers so far, one per question. #[serde(default, skip_serializing_if = "Option::is_none")] answers: Option>>, /// `running`, `completed`, `failed` or `stopped`, if it ran in the /// background. #[serde(default, skip_serializing_if = "Option::is_none")] background: Option, }, Notice { error: bool, text: String }, TaskEvent { summary: String, status: String }, Question { title: String, message: Option, options: Vec, #[serde(default)] confirm: bool, answer: Option, }, TurnEnd { duration_ms: u64, cost: Option, at: u64 }, } pub fn to_millis(time: SystemTime) -> u64 { time.duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as u64 } pub fn from_millis(millis: u64) -> SystemTime { UNIX_EPOCH + Duration::from_millis(millis) } impl Session { pub fn save(&self) -> SavedThread { let items = self.saved_items(0); self.save_with(items) } /// The transcript from item `from` on, as it's saved and sent. pub fn saved_items(&self, from: usize) -> Vec { self.items .iter() .skip(from) .map(|item| match item { ChatItem::User { text, attachments, at } => SavedItem::User { text: text.to_string(), attachments: attachments.iter().map(ToString::to_string).collect(), at: to_millis(*at), }, ChatItem::Assistant(answer) => SavedItem::Assistant { text: answer.text.clone() }, ChatItem::Tool(tool) => SavedItem::Tool { id: tool.id.clone(), name: tool.name.clone(), input: tool.input.clone(), result: tool.result.clone(), state: match tool.state { ToolState::Done => "done", ToolState::Running => "running", ToolState::Error => "error", } .into(), permission: tool.permission.as_ref().map(|prompt| { match prompt.state { PermissionState::Pending => "pending", PermissionState::Allowed => "allowed", PermissionState::Denied => "denied", PermissionState::Cancelled => "cancelled", } .into() }), suggestions: tool .permission .as_ref() .filter(|prompt| prompt.state == PermissionState::Pending) .map(|prompt| prompt.suggestions.clone()) .unwrap_or_default(), subagent_tool_count: tool.subagent_tool_count, answers: tool .ask .as_ref() .map(|ask| ask.answers.iter().map(|answer| answer.as_ref().map(ToString::to_string)).collect()), background: tool.background.map(|status| status.key().to_string()), }, ChatItem::Notice(kind, text) => { SavedItem::Notice { error: *kind == NoticeKind::Error, text: text.to_string() } } ChatItem::TaskEvent { summary, status } => { SavedItem::TaskEvent { summary: summary.to_string(), status: status.key().to_string() } } ChatItem::Question(question) => SavedItem::Question { title: question.title.to_string(), message: question.message.as_ref().map(ToString::to_string), options: question.options.iter().map(ToString::to_string).collect(), confirm: question.confirm, answer: question.answer.as_ref().map(ToString::to_string), }, ChatItem::TurnEnd { duration_ms, cost, at } => { SavedItem::TurnEnd { duration_ms: *duration_ms, cost: *cost, at: to_millis(*at) } } }) .collect() } fn save_with(&self, items: Vec) -> SavedThread { SavedThread { id: self.id.clone(), instance: self.instance.clone(), provider: self.provider, title: self.title.to_string(), title_custom: self.title_custom, project: self.project.clone(), cwd: self.cwd.clone(), worktree: self.worktree.clone(), pending_worktree: self.pending_worktree.clone(), pull_request: self.pull_request.clone(), resume_id: self.resume_id.clone(), model: self.model.clone(), active_model: self.active_model.clone(), permission_mode: self.permission_mode.as_cli().into(), reasoning: self.reasoning.key().into(), fast_mode: self.fast_mode, total_cost: self.total_cost, context_tokens: self.context_tokens, context_window: self.context_window, created: to_millis(self.created), updated: to_millis(self.updated), last_message: to_millis(self.last_message), settled: self.settled, pinned: self.pinned, snoozed_until: self.snoozed_until.map(to_millis), limit_resets: self.limit_resets.map(to_millis), resume_at: self.resume_at.map(to_millis), queue: self.queue.clone(), working: self.is_working(), items, turn_changes: self.turn_changes.iter().map(|(end, changes)| (*end, changes.clone())).collect(), } } /// Brings a saved thread back. Its agent starts again, resuming the /// conversation, with the next message. pub fn restore(saved: SavedThread, cx: &mut Context) -> Self { let new = NewSession { instance: saved.instance, provider: saved.provider, project: saved.project, model: saved.model, permission_mode: PermissionMode::from_cli(&saved.permission_mode).unwrap_or_default(), reasoning: Reasoning::from_key(&saved.reasoning).unwrap_or_default(), fast_mode: saved.fast_mode, worktree: saved.pending_worktree, }; let mut session = Self::empty(saved.id, new, from_millis(saved.created)); session.status = SessionStatus::Idle; session.title = saved.title.into(); session.title_custom = saved.title_custom; session.cwd = saved.cwd; session.worktree = saved.worktree; session.pull_request = saved.pull_request; session.resume_id = saved.resume_id; session.active_model = saved.active_model; session.total_cost = saved.total_cost; session.context_tokens = saved.context_tokens; session.context_window = saved.context_window; session.updated = from_millis(saved.updated); session.last_message = from_millis(saved.last_message); session.settled = saved.settled; session.pinned = saved.pinned; session.snoozed_until = saved.snoozed_until.map(from_millis); session.limit_resets = saved.limit_resets.map(from_millis); session.resume_at = saved.resume_at.map(from_millis); session.queue = saved.queue; session.interrupted = saved.working; session.turn_changes = saved.turn_changes.into_iter().collect(); for item in saved.items { let item = session.item_from_saved(item, false, cx); session.items.push(item); } if session.interrupted { session.push_notice(NoticeKind::Info, "The app stopped while this thread was working."); } session } /// Turns a saved item back into a chat item. `live` keeps pending /// prompts and questions pending (a host's transcript); otherwise they /// count as over (a thread restored from disk). fn item_from_saved(&mut self, item: SavedItem, live: bool, cx: &mut Context) -> ChatItem { match item { SavedItem::User { text, attachments, at } => ChatItem::User { text: text.into(), attachments: attachments.into_iter().map(Into::into).collect(), at: from_millis(at), }, SavedItem::Assistant { text } => { let view = cx.new(|cx| TextViewState::markdown(&text, cx).selectable(true)); let shown = text.len(); ChatItem::Assistant(AssistantText { text, view, shown }) } SavedItem::Tool { id, name, input, result, state, permission, suggestions, subagent_tool_count, answers, background, } => { let pending = live && permission.as_deref() == Some("pending"); let ask = answers.and_then(|answers| { let mut ask = UserQuestions::parse(&input)?; for (slot, answer) in ask.answers.iter_mut().zip(answers) { *slot = answer.map(Into::into); } ask.current = ask.answers.iter().position(Option::is_none).unwrap_or(0); Some(ask) }); let background = background.map(|status| match TaskStatus::from_cli(&status) { // Background tasks die with the agent process. TaskStatus::Running if !live => TaskStatus::Stopped, status => status, }); ChatItem::Tool(ToolCall { id, name, input: input.clone(), partial_input: String::new(), result, state: match state.as_str() { "done" => ToolState::Done, "running" if live => ToolState::Running, _ => ToolState::Error, }, permission: permission.map(|state| PermissionPrompt { request_id: String::new(), input, suggestions, state: match state.as_str() { "allowed" => PermissionState::Allowed, "denied" => PermissionState::Denied, _ if pending => PermissionState::Pending, _ => PermissionState::Cancelled, }, }), expanded: false, subagent_tool_count, detached: false, ask, task_id: None, background, }) } SavedItem::Notice { error, text } => { ChatItem::Notice(if error { NoticeKind::Error } else { NoticeKind::Info }, text.into()) } SavedItem::TaskEvent { summary, status } => { ChatItem::TaskEvent { summary: summary.into(), status: TaskStatus::from_cli(&status) } } SavedItem::Question { title, message, options, confirm, answer } => ChatItem::Question(Question { request_id: String::new(), title: title.into(), message: message.map(Into::into), options: options.into_iter().map(Into::into).collect(), confirm, answer: match answer { Some(answer) => Some(answer.into()), None if live => None, None => Some("No answer".into()), }, }), SavedItem::TurnEnd { duration_ms, cost, at } => ChatItem::TurnEnd { duration_ms, cost, at: from_millis(at) }, } } } // ---- mirroring threads across devices ------------------------------------- fn status_key(status: SessionStatus) -> &'static str { match status { SessionStatus::Starting => "starting", SessionStatus::Idle => "idle", SessionStatus::Working => "working", SessionStatus::Exited => "exited", } } fn status_from_key(key: &str) -> SessionStatus { match key { "starting" => SessionStatus::Starting, "working" => SessionStatus::Working, "exited" => SessionStatus::Exited, _ => SessionStatus::Idle, } } /// Reads a file for sending to a host. fn upload(path: &Path) -> Result { let size = path.metadata().map_err(|err| err.to_string())?.len(); if size > 20 * 1024 * 1024 { return Err("it is over the 20 MB limit for sending to another device".into()); } let bytes = std::fs::read(path).map_err(|err| err.to_string())?; Ok(Upload { name: file_name(path), data: BASE64.encode(bytes) }) } /// Writes files a client sent into a folder of their own, returning their paths. pub fn store_uploads(uploads: Vec) -> Vec { if uploads.is_empty() { return Vec::new(); } let Some(dir) = settings::data_dir().map(|dir| { dir.join("uploads").join(format!("{:x}", to_millis(SystemTime::now()) ^ u64::from(std::process::id()))) }) else { return Vec::new(); }; if std::fs::create_dir_all(&dir).is_err() { return Vec::new(); } uploads .into_iter() .filter_map(|upload| { let name = Path::new(&upload.name).file_name()?.to_owned(); let path = dir.join(name); let bytes = BASE64.decode(upload.data).ok()?; std::fs::write(&path, bytes).ok()?; Some(path) }) .collect() } impl Session { /// A thread on another device, as the host's thread list describes it. pub fn remote(summary: &ThreadSummary, link: RemoteLink, cx: &mut Context) -> Self { let _ = cx; let new = NewSession { instance: summary.instance.clone(), provider: summary.provider, project: summary.project.clone(), model: summary.model.clone(), permission_mode: PermissionMode::default(), reasoning: Reasoning::default(), fast_mode: false, worktree: None, }; let id = summary.id.clone(); let mut session = Self::empty(id, new, from_millis(summary.created)); session.remote = Some(link); session.apply_summary(summary); session } /// A new, still empty thread meant for one of a host's projects. pub fn new_remote(new: NewSession, link: RemoteLink) -> Self { let now = SystemTime::now(); let id = format!("remote-{:x}-{}", to_millis(now), NEXT_SESSION.load(Ordering::Relaxed)); let mut session = Self::empty(id, new, now); session.status = SessionStatus::Idle; session.remote = Some(link); session } /// Asks the host to delete a mirrored thread. pub fn forget_remote(&self) { self.forward(ThreadAction::Delete); } pub fn is_remote(&self) -> bool { self.remote.is_some() } /// Whether nothing was sent in it: not even a remote thread that exists. pub fn is_blank(&self) -> bool { self.items.is_empty() && self.remote.as_ref().is_none_or(|remote| remote.thread.is_none() && remote.creating.is_none()) } /// Hands an action to the host, for a mirrored thread. Returns whether /// the thread is mirrored (then the action is the host's to carry out). fn forward(&self, action: ThreadAction) -> bool { let Some(remote) = &self.remote else { return false }; if let Some(thread) = &remote.thread { remote.client.send(&ClientMessage::Action { thread: thread.clone(), action }); } true } /// Sends (or queues) a message to a mirrored thread, creating the /// thread on the host with its first message. fn send_remote(&mut self, text: String, attachments: Vec, queue: bool, cx: &mut Context) { if text.trim().is_empty() && attachments.is_empty() { return; } let mut uploads = Vec::new(); for path in &attachments { match upload(path) { Ok(file) => uploads.push(file), Err(err) => self.push_notice(NoticeKind::Error, format!("Couldn't attach {}: {err}", file_name(path))), } } let Some(remote) = self.remote.as_mut() else { return }; match remote.thread.clone() { Some(thread) => { let action = if queue { ThreadAction::Queue { text: text.clone(), attachments: uploads } } else { ThreadAction::Send { text: text.clone(), attachments: uploads } }; remote.client.send(&ClientMessage::Action { thread, action }); if queue { self.queue.push(QueuedMessage { text, attachments }); cx.notify(); return; } } None if remote.creating.is_some() => { // The thread is on its way; this waits for it. self.queue.push(QueuedMessage { text, attachments }); cx.notify(); return; } None => { static NEXT_REFERENCE: AtomicU64 = AtomicU64::new(1); let reference = NEXT_REFERENCE.fetch_add(1, Ordering::Relaxed); remote.creating = Some(reference); let plan = self.pending_worktree.clone(); remote.client.send(&ClientMessage::Create { reference, project: self.project.clone(), instance: self.instance.clone(), provider: self.provider, model: self.model.clone(), permission_mode: self.permission_mode.as_cli().into(), reasoning: self.reasoning.key().into(), fast_mode: self.fast_mode, worktree: plan.is_some(), base: plan.and_then(|plan| plan.base), text: text.clone(), attachments: uploads, }); } } // Show the message right away; the host's transcript confirms it. let now = SystemTime::now(); let names = attachments.iter().map(|path| file_name(path).into()).collect(); if self.title.as_ref() == DEFAULT_TITLE { self.title = title_from(&text).into(); } self.items.push(ChatItem::User { text: text.into(), attachments: names, at: now }); self.status = SessionStatus::Working; self.updated = now; self.last_message = now; cx.notify(); } /// Learns the host's id for a thread created from here, and sends what /// was typed meanwhile. pub fn remote_created(&mut self, thread: String, cx: &mut Context) { let Some(remote) = self.remote.as_mut() else { return }; remote.creating = None; remote.thread = Some(thread.clone()); remote.subscribed = true; remote.client.send(&ClientMessage::Subscribe { thread: thread.clone() }); self.id = thread; for message in std::mem::take(&mut self.queue) { self.send_remote(message.text, message.attachments, true, cx); } cx.notify(); } /// Asks the host for the transcript, once. pub fn subscribe_remote(&mut self) { let Some(remote) = self.remote.as_mut() else { return }; if let (Some(thread), false) = (&remote.thread, remote.subscribed) { remote.client.send(&ClientMessage::Subscribe { thread: thread.clone() }); remote.subscribed = true; } } pub fn apply_summary(&mut self, summary: &ThreadSummary) { self.title = summary.title.clone().into(); self.provider = summary.provider; self.instance = summary.instance.clone(); self.model = summary.model.clone(); self.permission_mode = PermissionMode::from_cli(&summary.permission_mode).unwrap_or_default(); self.reasoning = Reasoning::from_key(&summary.reasoning).unwrap_or_default(); self.fast_mode = summary.fast_mode; self.status = status_from_key(&summary.status); self.settled = summary.settled; self.pinned = summary.pinned; self.snoozed_until = summary.snoozed_until.map(from_millis); self.updated = from_millis(summary.updated); self.last_message = from_millis(summary.last_message); self.total_cost = summary.total_cost; self.project = summary.project.clone(); self.worktree = summary.branch.clone().map(|branch| Worktree { root: self.cwd.clone(), branch, repo: summary.project.clone(), }); if let Some(remote) = self.remote.as_mut() { remote.attention = summary.attention; } } /// Takes a host's update of a subscribed thread: its state, and its /// transcript from item `from` on. pub fn apply_remote( &mut self, summary: &ThreadSummary, detail: ThreadDetail, from: usize, len: usize, items: Vec, cx: &mut Context, ) { self.apply_summary(summary); self.cwd = detail.cwd; if let Some(worktree) = self.worktree.as_mut() { worktree.root = self.cwd.clone(); } self.active_model = detail.active_model; self.thinking = detail.thinking; self.preparing = detail.preparing; self.turn_started = detail.turn_elapsed_ms.map(|elapsed| Instant::now() - Duration::from_millis(elapsed)); self.context_tokens = detail.context_tokens; self.context_window = detail.context_window; self.queue = detail.queue; self.limit_resets = detail.limit_resets.map(from_millis); self.resume_at = detail.resume_at.map(from_millis); self.pull_request = detail.pull_request; if !detail.pending_worktree { self.pending_worktree = None; } self.turn_changes = detail.turn_changes.into_iter().collect(); self.background_tasks = detail .background_tasks .into_iter() .map(|task| BackgroundTask { task_id: task.task_id, description: task.description.into(), started: Instant::now() - Duration::from_millis(task.elapsed_ms), }) .collect(); let from = from.min(self.items.len()); for (offset, saved) in items.into_iter().enumerate() { let ix = from + offset; // A streaming answer grows; feed it the new text instead of // rebuilding its view. if let (Some(ChatItem::Assistant(answer)), SavedItem::Assistant { text }) = (self.items.get_mut(ix), &saved) && text.starts_with(answer.text.as_str()) { let new = text[answer.text.len()..].to_string(); answer.text.push_str(&new); let end = answer.text.len(); answer.show(end, cx); continue; } let mut item = self.item_from_saved(saved, true, cx); // Keep tool cards open across updates. if let (Some(ChatItem::Tool(old)), ChatItem::Tool(new)) = (self.items.get(ix), &mut item) && old.id == new.id { new.expanded = old.expanded; } match self.items.get_mut(ix) { Some(slot) => *slot = item, None => self.items.push(item), } } self.items.truncate(len); if let Some(remote) = self.remote.as_mut() { remote.loaded = true; } cx.notify(); } /// The thread as a client's sidebar needs it. pub fn summary(&self) -> ThreadSummary { ThreadSummary { id: self.id.clone(), project: self.project.clone(), title: self.title.to_string(), provider: self.provider, instance: self.instance.clone(), model: self.model.clone(), permission_mode: self.permission_mode.as_cli().into(), reasoning: self.reasoning.key().into(), fast_mode: self.fast_mode, status: status_key(self.status).into(), attention: self.needs_attention(), settled: self.settled, pinned: self.pinned, snoozed_until: self.snoozed_until.filter(|_| self.is_snoozed()).map(to_millis), created: to_millis(self.created), updated: to_millis(self.updated), last_message: to_millis(self.last_message), branch: self.worktree.as_ref().map(|worktree| worktree.branch.clone()), total_cost: self.total_cost, has_items: !self.items.is_empty(), } } /// The rest of the thread's state, for a client that has it open. pub fn detail(&self) -> ThreadDetail { ThreadDetail { cwd: self.cwd.clone(), active_model: self.active_model.clone(), thinking: self.thinking, preparing: self.preparing, turn_elapsed_ms: self.turn_started.map(|start| start.elapsed().as_millis() as u64), context_tokens: self.context_tokens, context_window: self.context_window, queue: self.queue.clone(), limit_resets: self.limit_resets.map(to_millis), resume_at: self.resume_at.map(to_millis), pending_worktree: self.pending_worktree.is_some(), pull_request: self.pull_request.clone(), turn_changes: self.turn_changes.iter().map(|(end, changes)| (*end, changes.clone())).collect(), background_tasks: self .background_tasks .iter() .map(|task| RemoteTask { task_id: task.task_id.clone(), description: task.description.to_string(), elapsed_ms: task.started.elapsed().as_millis() as u64, }) .collect(), } } /// Schedules (or reschedules) continuing after a usage limit. pub fn schedule_resume(&mut self, at: SystemTime, cx: &mut Context) { self.forward(ThreadAction::ScheduleResume { at: to_millis(at) }); self.resume_at = Some(at); cx.notify(); } } impl Drop for Session { fn drop(&mut self) { let _ = std::fs::remove_file(&self.snapshot_index); } } fn image_media_type(path: &Path) -> Option<&'static str> { let extension = path.extension()?.to_str()?.to_ascii_lowercase(); match extension.as_str() { "png" => Some("image/png"), "jpg" | "jpeg" => Some("image/jpeg"), "gif" => Some("image/gif"), "webp" => Some("image/webp"), _ => None, } } fn read_image(path: &Path) -> Result { let size = path.metadata().map_err(|err| err.to_string())?.len(); if size > MAX_IMAGE_BYTES { return Err(format!("it is {:.1} MB, over the 5 MB limit for images", size as f64 / 1_048_576.)); } let bytes = std::fs::read(path).map_err(|err| err.to_string())?; Ok(BASE64.encode(bytes)) } impl AssistantText { /// Moves the view up to `end` bytes of the text. fn show(&mut self, end: usize, cx: &mut App) { if end > self.shown && self.text.is_char_boundary(end) { let new = self.text[self.shown..end].to_string(); self.view.update(cx, |state, cx| state.push_str(&new, cx)); self.shown = end; } } } /// How much of streamed Markdown is finished: up to the last blank line /// outside a code block, or the end of the last closed code block. fn finished_prefix(text: &str) -> usize { let mut fence: Option<&str> = None; let mut finished = 0; let mut offset = 0; for line in text.split_inclusive('\n') { offset += line.len(); if !line.ends_with('\n') { break; } let trimmed = line.trim(); let marker = ["```", "~~~"].into_iter().find(|marker| trimmed.starts_with(marker)); match (fence, marker) { (None, Some(marker)) => fence = Some(marker), (Some(open), Some(marker)) if open == marker && trimmed.trim_start_matches(marker).is_empty() => { fence = None; finished = offset; } (None, None) if trimmed.is_empty() => finished = offset, _ => {} } } finished } fn file_name(path: &Path) -> String { path.file_name().map(|name| name.to_string_lossy().to_string()).unwrap_or_else(|| path.display().to_string()) } fn title_from(text: &str) -> String { let line = text.lines().find(|line| !line.trim().is_empty()).unwrap_or(text).trim(); let mut title: String = line.chars().take(48).collect(); if line.chars().count() > 48 { title.push('…'); } title } /// Parses a tool call's streamed arguments into its input, once they're complete. fn take_partial_input(tool: &mut ToolCall) { if tool.input.is_null() && let Ok(input) = serde_json::from_str::(&tool.partial_input) { tool.input = input; tool.partial_input.clear(); } } fn tool_result_text(content: &Value) -> String { match content { Value::String(text) => text.clone(), Value::Array(blocks) => blocks .iter() .filter_map(|block| match block.get("type").and_then(Value::as_str) { Some("text") => block.get("text").and_then(Value::as_str).map(str::to_string), Some("image") => Some("[image]".to_string()), _ => None, }) .collect::>() .join("\n"), Value::Null => String::new(), other => other.to_string(), } } /// Reads the first of several possible argument names: Claude and Pi name /// the same tool arguments differently (`file_path` vs `path`). pub fn tool_field(input: &Value, keys: &[&str]) -> Option { keys.iter().find_map(|key| input.get(*key).and_then(Value::as_str)).map(str::to_string) } /// A one-line description of what a tool call is doing. pub fn tool_summary(name: &str, input: &Value, cwd: &std::path::Path) -> String { let field = |keys: &[&str]| tool_field(input, keys); let path = |keys: &[&str]| { field(keys).map(|path| { std::path::Path::new(&path) .strip_prefix(cwd) .map(|relative| relative.display().to_string()) .unwrap_or(path) }) }; let summary = match name.to_ascii_lowercase().as_str() { "bash" => field(&["command"]), "read" | "write" | "edit" | "multiedit" => path(&["file_path", "path"]), "notebookedit" => path(&["notebook_path"]), "glob" | "grep" | "find" => field(&["pattern", "query"]), "ls" => path(&["path"]), "webfetch" => field(&["url"]), "websearch" => field(&["query"]), "task" | "agent" => field(&["description"]), "todowrite" => input .get("todos") .and_then(Value::as_array) .map(|todos| format!("{} items", todos.len())), "exitplanmode" => Some("Plan ready for review".to_string()), _ => None, }; let summary = summary.unwrap_or_else(|| match input { Value::Null => String::new(), Value::Object(map) if map.is_empty() => String::new(), other => other.to_string(), }); let first_line = summary.lines().next().unwrap_or_default(); let mut line: String = first_line.chars().take(160).collect(); if first_line.chars().count() > 160 || summary.lines().nth(1).is_some() { line.push('…'); } line } #[cfg(test)] mod tests { use super::finished_prefix; #[::core::prelude::v1::test] fn finished_paragraphs() { assert_eq!(finished_prefix("Half a sent"), 0); assert_eq!(finished_prefix("One.\n\nTwo is still"), "One.\n\n".len()); let open_fence = "Intro.\n\n```rust\nfn a() {}\n\nfn b"; assert_eq!(finished_prefix(open_fence), "Intro.\n\n".len()); let closed = "```rust\nfn a() {}\n```\nAfter"; assert_eq!(finished_prefix(closed), "```rust\nfn a() {}\n```\n".len()); assert_eq!(finished_prefix("Ünïcödé.\n\nMore"), "Ünïcödé.\n\n".len()); } }