//! One chat thread: owns the background agent process (Claude Code or Pi) and
//! folds its event stream into a list of renderable chat items.
use std::{
collections::{HashMap, HashSet, VecDeque},
path::{Path, PathBuf},
sync::atomic::{AtomicU64, Ordering},
time::Instant,
};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use gpui_kit::component::text::TextViewState;
use gpui_kit::*;
use serde_json::{Value, json};
use crate::claude::{AgentProcess, Effort, LaunchConfig, PermissionMode, ProcessEvent};
use crate::git::{self, TurnChanges};
use crate::models::{Catalog, Provider};
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;
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<Value>,
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<String>,
pub state: ToolState,
pub permission: Option<PermissionPrompt>,
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<UserQuestions>,
/// 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<String>,
/// Set once the tool runs as a background task.
pub background: Option<TaskStatus>,
}
/// Where a background task stands.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TaskStatus {
Running,
Completed,
Failed,
Stopped,
}
impl TaskStatus {
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<SharedString>,
}
pub struct AskQuestion {
pub question: SharedString,
/// A short chip label, e.g. "Auth method".
pub header: SharedString,
pub options: Vec<AskOption>,
pub multi_select: bool,
}
/// The questions an AskUserQuestion call puts to the user, and their answers
/// so far.
pub struct UserQuestions {
pub questions: Vec<AskQuestion>,
/// The question on screen.
pub current: usize,
/// Options ticked on each multi-select question, not yet committed.
pub picked: Vec<Vec<usize>>,
pub answers: Vec<Option<SharedString>>,
}
impl UserQuestions {
fn parse(input: &Value) -> Option<Self> {
let text = |value: &Value, key: &str| {
value.get(key).and_then(Value::as_str).map(|text| SharedString::from(text.to_string()))
};
let questions: Vec<AskQuestion> = 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<SharedString>,
pub options: Vec<SharedString>,
/// A yes/no confirmation rather than a pick from a list.
pub confirm: bool,
pub answer: Option<SharedString>,
}
pub enum ChatItem {
User {
text: SharedString,
/// File names of what was attached.
attachments: Vec<SharedString>,
},
Assistant(Entity<TextViewState>),
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<f64> },
}
/// 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<Self> {
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<Effort> {
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<usize>,
/// The turn's `TurnEnd` item.
pub end: usize,
pub duration_ms: u64,
pub cost: Option<f64>,
/// 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<u64, usize>,
}
pub struct Session {
pub provider: Provider,
pub title: SharedString,
pub cwd: PathBuf,
pub items: Vec<ChatItem>,
pub status: SessionStatus,
pub thinking: bool,
/// What a restarted process resumes: Claude's session id or Pi's session file.
resume_id: Option<String>,
/// The model picked for the thread; `None` means the agent's default.
pub model: Option<String>,
/// The concrete model the agent reported, e.g. `claude-opus-5-5`.
pub active_model: Option<String>,
pub permission_mode: PermissionMode,
pub reasoning: Reasoning,
pub fast_mode: bool,
/// The thinking levels Pi offers for the current model.
pub pi_efforts: Vec<Effort>,
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<Instant>,
/// When the thread last saw activity, for ordering the thread list.
pub updated: Instant,
/// Filed away by the user: its process is stopped until the next message.
pub settled: bool,
/// Tokens in the context window as of the last response.
pub context_tokens: Option<u64>,
pub context_window: Option<u64>,
/// Tasks running in the background, oldest first.
pub background_tasks: Vec<BackgroundTask>,
/// When each task the CLI reported started, and the tool that started it.
task_origins: HashMap<String, (Instant, Option<String>)>,
/// Finished turns the user unfolded, keyed by their `TurnEnd` index.
expanded_turns: HashSet<usize>,
/// What each finished turn changed on disk, keyed by its `TurnEnd` index.
turn_changes: HashMap<usize, TurnChanges>,
/// Folders the user opened in a turn's change overview.
expanded_dirs: HashMap<usize, HashSet<String>>,
/// The private git index snapshots are staged into.
snapshot_index: PathBuf,
/// The working tree as the running turn found it.
base_snapshot: Option<Task<Option<String>>>,
/// Set when the user stops a turn, so its error result isn't reported.
interrupt_requested: bool,
stderr_tail: VecDeque<String>,
process: Option<AgentProcess>,
streaming: Option<StreamingMessage>,
streamed_message_ids: HashSet<String>,
pending_controls: HashMap<String, PendingControl>,
next_request_id: u64,
_events_task: Task<()>,
}
impl Session {
pub fn new(
provider: Provider,
cwd: PathBuf,
model: Option<String>,
permission_mode: PermissionMode,
reasoning: Reasoning,
fast_mode: bool,
cx: &mut Context<Self>,
) -> Self {
let snapshot_index = std::env::temp_dir().join(format!(
"vectrum-code-{}-{}.index",
std::process::id(),
NEXT_SESSION.fetch_add(1, Ordering::Relaxed)
));
let mut session = Self {
provider,
title: DEFAULT_TITLE.into(),
cwd,
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,
updated: Instant::now(),
settled: 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,
_events_task: Task::ready(()),
};
session.start(None, cx);
session
}
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 {
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<String>, cx: &mut Context<Self>) {
let (tx, rx) = smol::channel::unbounded();
let config = LaunchConfig {
provider: self.provider,
cwd: self.cwd.clone(),
model: self.model.clone(),
permission_mode: self.permission_mode,
effort: self.reasoning.effort(),
resume,
};
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"}));
}
}
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);
}
}
// ---- user actions -------------------------------------------------
/// Sends a message. Images are attached inline; other files are listed by
/// path for the agent to read.
pub fn send_message(&mut self, text: String, attachments: Vec<PathBuf>, cx: &mut Context<Self>) {
let mut images = Vec::new();
let mut names = Vec::new();
let mut file_paths = Vec::new();
for path in &attachments {
let name = path
.file_name()
.map(|name| name.to_string_lossy().to_string())
.unwrap_or_else(|| path.display().to_string());
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 {name}: {err}"));
continue;
}
},
None => file_paths.push(path.display().to_string()),
}
names.push(SharedString::from(name));
}
let mut prompt = text.clone();
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() {
cx.notify();
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 {
cx.notify();
return;
};
match self.provider {
Provider::Claude => {
let mut content: Vec<Value> = 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::<Vec<_>>(),
});
// Pi rejects a prompt mid-run unless told when to deliver it.
if self.is_working() {
command["streamingBehavior"] = "followUp".into();
}
self.pi(command);
}
}
if self.title.as_ref() == DEFAULT_TITLE {
let source = if text.trim().is_empty() { names.join(", ") } else { text.clone() };
self.title = title_from(&source).into();
}
self.items.push(ChatItem::User { text: text.into(), attachments: names });
self.begin_turn(cx);
self.settled = false;
cx.notify();
}
/// 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>) {
self.updated = Instant::now();
if self.is_working() {
return;
}
self.turn_started = Some(Instant::now());
self.turn_cost = 0.;
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) }));
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<Self>) {
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<Self>) {
self.control(PendingControl::StopTask, json!({"subtype": "stop_task", "task_id": task_id}));
cx.notify();
}
/// 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<Self>) {
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;
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>) {
self.settled = false;
self.updated = Instant::now();
cx.notify();
}
pub fn interrupt(&mut self, cx: &mut Context<Self>) {
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"})),
}
cx.notify();
}
pub fn set_model(&mut self, model: Option<String>, cx: &mut Context<Self>) {
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"}));
}
}
}
cx.notify();
}
pub fn set_permission_mode(&mut self, mode: PermissionMode, cx: &mut Context<Self>) {
// Pi has no permission prompts to configure.
if self.provider != Provider::Claude {
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<Self>) {
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()}));
}
}
}
cx.notify();
}
pub fn set_fast_mode(&mut self, on: bool, cx: &mut Context<Self>) {
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<Self>) {
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<Self>) {
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<String>, cx: &mut Context<Self>) {
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<FinishedTurn> {
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 } = 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, has_work })
}
pub fn toggle_tool(&mut self, ix: usize, cx: &mut Context<Self>) {
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<Self>,
) {
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<Self>) {
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<Self>) {
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<Self>) {
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<Self>) {
let Some(ask) = self.pending_ask_mut(ix) else { return };
if ask.answer(text.into()) {
self.submit_ask(ix);
}
cx.notify();
}
/// Dismisses the form without answering.
pub fn ask_skip(&mut self, ix: usize, cx: &mut Context<Self>) {
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) {
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<String, Value> = 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<Self>) {
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<Self>) {
match event {
ProcessEvent::Message(message) => match self.provider {
Provider::Claude => self.handle_message(message, cx),
Provider::Pi => self.handle_pi_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),
}
}
fn handle_exit(&mut self, code: Option<i32>) {
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 = match self.provider {
Provider::Claude => "Claude Code",
Provider::Pi => "Pi",
};
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<f64>, cx: &mut Context<Self>) {
self.turn_started = None;
self.updated = Instant::now();
self.items.push(ChatItem::TurnEnd { duration_ms, cost });
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.notify();
});
}
})
.detach();
}
// ---- Claude Code (stream-json) -------------------------------------
fn handle_message(&mut self, message: Value, cx: &mut Context<Self>) {
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("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<Self>) {
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 = Instant::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<Self>) {
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(markdown))) => {
let text = delta.get("text").and_then(Value::as_str).unwrap_or_default();
markdown.update(cx, |state, cx| state.push_str(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(ChatItem::Tool(tool)) = self.streamed_item(index).and_then(|ix| self.items.get_mut(ix))
&& let Ok(input) = serde_json::from_str::<Value>(&tool.partial_input)
{
tool.input = input;
}
}
Some("message_stop") => {
self.thinking = false;
self.streaming = None;
}
_ => {}
}
}
fn streamed_item(&self, index: u64) -> Option<usize> {
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<Self>) {
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.");
}
}
_ => {}
}
}
}
fn handle_result(&mut self, message: &Value, cx: &mut Context<Self>) {
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);
}
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 = Instant::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<Self>) {
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_models(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<Self>) {
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);
}
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<Self>) {
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
}
};
if let Some(ChatItem::Assistant(markdown)) = self.items.get(ix) {
markdown.update(cx, |state, cx| state.push_str(delta, 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 = Instant::now();
}
// ---- helpers -----------------------------------------------------
fn push_assistant_text(&mut self, text: &str, cx: &mut Context<Self>) -> usize {
let markdown = cx.new(|cx| TextViewState::markdown(text, cx).selectable(true));
self.items.push(ChatItem::Assistant(markdown));
self.items.len() - 1
}
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<SharedString>) {
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());
}
}
}
}
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<String, String> {
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))
}
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
}
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::<Vec<_>>()
.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<String> {
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
}