//! 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::orchestrator;
use crate::remote::{ClientMessage, RemoteTask, ThreadAction, ThreadDetail, ThreadSummary, Upload, client::RemoteClient};
use crate::settings::{self, BranchNaming, Streaming};
use crate::worktree::{self, PullRequest, SetupStage, StageStatus, 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<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>,
/// What a subagent (Task or Agent tool) is up to; `None` for other tools.
pub subagent: Option<SubagentProgress>,
}
impl ToolCall {
/// The subagents this call started, as the subagents panel lists them:
/// one for Claude's Task or Agent call, one per task for Pi's `subagent`
/// extension, and the agents a jcode `swarm` spawn started.
pub fn subagents(&self) -> Vec<SubagentEntry> {
let Some(progress) = &self.subagent else { return Vec::new() };
let call_running = self.state == ToolState::Running;
if !progress.members.is_empty() {
return progress
.members
.iter()
.map(|member| {
let mut status = TaskStatus::from_cli(&member.status);
// A Pi subagent runs inside its call; a swarm agent or
// a delegated task outlives the call that started it.
if status == TaskStatus::Running && !call_running && !self.outlives_call() {
status = TaskStatus::Stopped;
}
let running = status == TaskStatus::Running;
SubagentEntry {
label: member.label.clone(),
kind: member.kind.clone(),
status,
background: false,
tool_uses: member.tool_uses,
tokens: member.tokens,
elapsed: elapsed_between(member.started, member.finished, running),
tool: member.tool.clone().filter(|_| running),
summary: member.summary.clone(),
thread: self.is_delegation().then(|| member.id.clone()),
stop: running.then(|| {
if self.is_delegation() {
SubagentStop::Thread(member.id.clone())
} else if self.name == SWARM_TOOL {
SubagentStop::SwarmAgent(member.id.clone())
} else {
SubagentStop::Turn
}
}),
}
})
.collect();
}
// Only a swarm call that spawns an agent starts one.
if self.name == SWARM_TOOL && tool_field(&self.input, &["action"]).as_deref() != Some("spawn") {
return Vec::new();
}
let status = match (self.background, self.state) {
(Some(status), ToolState::Running | ToolState::Done) => status,
(_, ToolState::Error) => TaskStatus::Failed,
(None, ToolState::Running) => TaskStatus::Running,
(None, ToolState::Done) => TaskStatus::Completed,
};
let running = status == TaskStatus::Running;
let label = tool_field(&self.input, &["description", "label", "title", "task"])
.filter(|label| !label.trim().is_empty())
.or_else(|| {
let tasks = self.input.get("tasks").or_else(|| self.input.get("chain"))?.as_array()?.len();
Some(format!("{tasks} subagents"))
})
.unwrap_or_else(|| "Subagent".into());
vec![SubagentEntry {
label,
kind: tool_field(&self.input, &["subagent_type", "agent"]),
status,
background: self.background.is_some(),
tool_uses: self.subagent_tool_count,
tokens: progress.tokens,
elapsed: elapsed_between(progress.started, progress.finished, running),
tool: progress.tool.clone().filter(|_| running),
summary: progress.summary.clone(),
thread: None,
// Claude's subagents are tasks; a delegation that isn't tied to
// its thread yet has nothing to stop.
stop: self.task_id.clone().filter(|_| running).map(SubagentStop::Task),
}]
}
/// A call to the app's `delegate_task`, whose task runs in a thread of its own.
pub fn is_delegation(&self) -> bool {
orchestrator::own_tool(&self.name) == Some(orchestrator::DELEGATE)
}
/// Whether the agents it starts carry on after the call returns.
fn outlives_call(&self) -> bool {
self.name == SWARM_TOOL || self.is_delegation()
}
}
/// A subagent as the subagents panel shows it.
pub struct SubagentEntry {
/// What it was asked to do.
pub label: String,
/// The kind of agent, e.g. `Explore`.
pub kind: Option<String>,
pub status: TaskStatus,
pub background: bool,
pub tool_uses: usize,
pub tokens: Option<u64>,
pub elapsed: Option<Duration>,
/// Its latest tool call, while it works.
pub tool: Option<(String, Value)>,
pub summary: Option<String>,
/// The thread it runs in, for a task delegated to the app.
pub thread: Option<String>,
/// How to stop it, while it runs.
pub stop: Option<SubagentStop>,
}
/// How the subagents panel stops a subagent.
#[derive(Clone, Debug, PartialEq)]
pub enum SubagentStop {
/// Claude's background-task control, by task id.
Task(String),
/// A task delegated to the app: its thread, by id.
Thread(String),
/// A jcode swarm agent, by session id.
SwarmAgent(String),
/// Nothing smaller than the whole turn can stop it (Pi's `subagent`
/// extension runs its tasks inside one tool call).
Turn,
}
/// jcode's tool for running a swarm of agents, each in a session of its own.
const SWARM_TOOL: &str = "swarm";
/// How often the app checks on a thread's working swarm agents.
const SWARM_POLL: Duration = Duration::from_secs(3);
/// The jcode requests that check on swarm agents, by what they're for.
const LIST_AGENTS: &str = "list the swarm's agents";
const PEEK_AGENT: &str = "read a swarm agent's progress";
/// The jcode request that registers the app's tools.
const CONFIGURE_TOOLS: &str = "set up the app's tools";
/// One agent among several a call started: a task of Pi's `subagent` call,
/// or an agent a jcode `swarm` call spawned.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct SubagentMember {
/// The task's place in a Pi call, or the jcode agent's session id.
pub id: String,
pub label: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub kind: Option<String>,
/// `running`, `completed`, `failed` or `stopped`.
pub status: String,
pub started: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tokens: Option<u64>,
#[serde(default)]
pub tool_uses: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool: Option<(String, Value)>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
}
impl SubagentMember {
/// Moves it to `status`, noting when it stopped running (or started
/// again: a swarm agent can be handed more work).
fn set_status(&mut self, status: TaskStatus) {
self.status = status.key().to_string();
if status == TaskStatus::Running {
self.finished = None;
} else {
self.finished.get_or_insert_with(|| to_millis(SystemTime::now()));
}
}
}
/// How long something ran from `started` to `finished` (or now, while it
/// runs), in milliseconds since the epoch; `None` when that isn't known.
fn elapsed_between(started: u64, finished: Option<u64>, running: bool) -> Option<Duration> {
let end = match finished {
Some(finished) => finished,
None if running => to_millis(SystemTime::now()),
None => return None,
};
(started > 0).then(|| Duration::from_millis(end.saturating_sub(started)))
}
/// A subagent's progress, for the subagents panel. It's saved with the
/// thread, so it reaches remote clients too.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct SubagentProgress {
/// When it started and finished, in milliseconds since the epoch.
pub started: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished: Option<u64>,
/// The tokens it has used, as the CLI reports them.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tokens: Option<u64>,
/// Its latest tool call: the tool and the input's short text fields.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool: Option<(String, Value)>,
/// The CLI's running summary of what it's doing.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
/// The agents it started, when it starts several or they live apart
/// from it (Pi's `subagent` tasks, jcode's swarm agents).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub members: Vec<SubagentMember>,
}
impl SubagentProgress {
fn finish(&mut self) {
self.finished.get_or_insert_with(|| to_millis(SystemTime::now()));
}
}
/// Whether a tool starts subagents: Claude's Task and Agent, Pi's
/// `subagent` extension (and jcode's old tool of that name), and jcode's
/// swarm, whose spawns do.
pub fn is_subagent_tool(name: &str) -> bool {
matches!(name, "Task" | "Agent" | "subagent" | SWARM_TOOL) || orchestrator::own_tool(name) == Some(orchestrator::DELEGATE)
}
/// The tool call a model's message ends on, from Pi's or jcode's message
/// content: its name and its input's short text fields.
fn last_tool_call(content: &Value) -> Option<(String, Value)> {
content.as_array()?.iter().rev().find_map(|block| {
(block.get("type").and_then(Value::as_str) == Some("toolCall")).then(|| {
let name = block.get("name").and_then(Value::as_str).unwrap_or("tool").to_string();
(name, brief_input(&block["arguments"]))
})
})
}
/// Folds the results Pi's `subagent` extension reports (one per task, in
/// `details.results`) into the call's members.
fn pi_subagent_results(tool: &mut ToolCall, details: Option<&Value>) {
let (Some(progress), Some(results)) =
(tool.subagent.as_mut(), details.and_then(|details| details.get("results")).and_then(Value::as_array))
else {
return;
};
let now = to_millis(SystemTime::now());
for (ix, result) in results.iter().enumerate() {
let id = ix.to_string();
if !progress.members.iter().any(|member| member.id == id) {
progress.members.push(SubagentMember { id: id.clone(), started: now, ..Default::default() });
}
let Some(member) = progress.members.iter_mut().find(|member| member.id == id) else { continue };
let text = |key: &str| result.get(key).and_then(Value::as_str).filter(|text| !text.trim().is_empty());
member.label = text("task")
.and_then(|task| task.lines().map(str::trim).find(|line| !line.is_empty()))
.map(|line| line.chars().take(200).collect())
.unwrap_or_else(|| "Subagent".into());
member.kind = text("agent").map(str::to_string);
let messages: Vec<&Value> = result.get("messages").and_then(Value::as_array).into_iter().flatten().collect();
let content = |message: &&Value| message.get("content").cloned().unwrap_or(Value::Null);
member.tool_uses = messages
.iter()
.filter_map(|message| message.get("content")?.as_array())
.flatten()
.filter(|block| block.get("type").and_then(Value::as_str) == Some("toolCall"))
.count();
member.tool = messages.iter().rev().find_map(|message| last_tool_call(&content(message)));
member.summary = messages
.iter()
.rev()
.filter(|message| message.get("role").and_then(Value::as_str) == Some("assistant"))
.find_map(|message| last_line(&tool_result_text(&content(message))));
let usage = &result["usage"];
let tokens: u64 = ["input", "output", "cacheRead", "cacheWrite"]
.iter()
.filter_map(|key| usage.get(*key).and_then(Value::as_u64))
.sum();
member.tokens = (tokens > 0).then_some(tokens);
let exit = result.get("exitCode").and_then(Value::as_i64).unwrap_or(-1);
let status = match text("stopReason") {
_ if exit < 0 => TaskStatus::Running,
Some("aborted") => TaskStatus::Stopped,
Some("error") => TaskStatus::Failed,
_ if exit != 0 => TaskStatus::Failed,
_ => TaskStatus::Completed,
};
member.set_status(status);
}
}
/// A swarm agent's state from jcode's session list: its swarm lifecycle
/// status, or failing that whether its session is mid-turn. `ready` is an
/// agent done with its work and waiting for more.
fn jcode_agent_status(swarm: Option<&str>, session: Option<&str>) -> TaskStatus {
match swarm {
Some("completed" | "done" | "ready") => TaskStatus::Completed,
Some("failed" | "crashed") => TaskStatus::Failed,
Some("stopped" | "closed" | "disconnected") => TaskStatus::Stopped,
Some(_) => TaskStatus::Running,
None if session == Some("running") => TaskStatus::Running,
None => TaskStatus::Completed,
}
}
/// The last line of what an agent said, for its summary.
fn last_line(text: &str) -> Option<String> {
let line = text.lines().rev().map(str::trim).find(|line| !line.is_empty())?;
Some(line.chars().take(200).collect())
}
/// A tool input cut down to its short text fields: enough to say what the
/// call does, without a file's contents.
pub(crate) fn brief_input(input: &Value) -> Value {
let Some(fields) = input.as_object() else { return Value::Null };
fields
.iter()
.filter_map(|(key, value)| {
let text = value.as_str()?;
let first = text.lines().next().unwrap_or_default();
Some((key.clone(), Value::String(first.chars().take(200).collect())))
})
.collect::<serde_json::Map<_, _>>()
.into()
}
/// Where a background task stands.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TaskStatus {
Running,
Completed,
Failed,
Stopped,
}
impl TaskStatus {
pub 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,
}
}
}
/// How a worktree's setup was called off.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SetupCancel {
/// Don't send the message.
Cancel,
/// Send it in the project folder instead.
Local,
}
/// A thread's worktree being made for its first message, stage by stage,
/// for T3 Code's setup card.
pub struct WorktreeSetup {
pub stages: Vec<(SetupStage, StageStatus, Option<Instant>, Option<Instant>)>,
pub started: Instant,
pub base: String,
pub branch: Option<String>,
pub path: Option<PathBuf>,
pub error: Option<String>,
pub cancel: Option<SetupCancel>,
pub cancelled: bool,
}
impl WorktreeSetup {
fn new(plan: &WorktreePlan) -> Self {
Self {
stages: SetupStage::ALL.iter().map(|&stage| (stage, StageStatus::Pending, None, None)).collect(),
started: Instant::now(),
base: plan.base.clone().unwrap_or_else(|| "HEAD".into()),
branch: None,
path: None,
error: None,
cancel: None,
cancelled: false,
}
}
fn mark(&mut self, stage: SetupStage, status: StageStatus) {
let now = Instant::now();
if let Some(entry) = self.stages.iter_mut().find(|entry| entry.0 == stage) {
entry.1 = status;
match status {
StageStatus::Running => entry.2 = Some(now),
StageStatus::Done | StageStatus::Failed => {
entry.2.get_or_insert(now);
entry.3 = Some(now);
}
StageStatus::Pending | StageStatus::Skipped => {}
}
}
}
}
/// 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,
/// The options selected on each question: at most one on a
/// single-select question.
pub picked: Vec<Vec<usize>>,
/// What the user typed as their own answer to each question. A
/// non-empty one wins over the selected options.
pub custom: Vec<String>,
/// The answers sent, once the form is submitted.
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],
custom: vec![String::new(); count],
answers: vec![None; count],
})
}
/// The answer question `q` has so far: the user's own words if they
/// typed any, otherwise the selected options. `None` while it has neither.
pub fn resolved(&self, q: usize) -> Option<String> {
let custom = self.custom.get(q).map(|custom| custom.trim()).unwrap_or_default();
if !custom.is_empty() {
return Some(custom.to_string());
}
let question = self.questions.get(q)?;
let mut picked = self.picked.get(q)?.clone();
picked.sort_unstable();
let labels: Vec<&str> =
picked.iter().filter_map(|&option| question.options.get(option)).map(|option| option.label.as_ref()).collect();
(!labels.is_empty()).then(|| labels.join(", "))
}
/// Whether every question has an answer.
pub fn is_complete(&self) -> bool {
(0..self.questions.len()).all(|q| self.resolved(q).is_some())
}
pub fn is_last(&self) -> bool {
self.current + 1 >= self.questions.len()
}
}
#[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>,
}
/// 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<TextViewState>,
/// How much of `text` the view shows.
shown: usize,
}
pub enum ChatItem {
User {
text: SharedString,
/// File names of what was attached.
attachments: Vec<SharedString>,
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<f64>, 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<String>,
pub client: Rc<RemoteClient>,
/// 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<u64>,
}
/// 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),
/// jcode called one of the app's tools, to be answered with
/// [`Session::answer_app_tool`].
AppTool { call_id: String, tool: String, input: Value },
}
/// Where a subagent's thread starts: its parent thread's id, folder and branch.
pub struct SubagentParent {
pub id: String,
pub cwd: PathBuf,
pub branch: Option<String>,
}
/// What makes a thread a subagent: the thread that delegated its task. The
/// task goes by the subagent thread's id.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Delegation {
pub parent: String,
/// Its result reached the parent, in a reply or a message.
#[serde(default)]
pub delivered: bool,
/// Stopped by `task_cancel` or by the user.
#[serde(default)]
pub cancelled: bool,
}
/// A message waiting for the running turn to end.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct QueuedMessage {
pub text: String,
pub attachments: Vec<PathBuf>,
}
/// 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>,
/// 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<u64, usize>,
}
/// 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<String>,
/// 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<u64, &'static str>,
/// The chat item each streamed text message renders into.
texts: HashMap<String, usize>,
/// The serving provider, e.g. `Anthropic`, which decides how token
/// counts add up.
provider: Option<String>,
/// The browser panel's instructions, sent along with the next message.
browser_note: Option<String>,
/// Checks on the swarm agents the thread spawned, while any work.
agent_poll: Option<Task<()>>,
/// The app's tools are to be registered again once the turn ends:
/// jcode turns down tool changes mid-turn.
tools_pending: bool,
/// The app's instructions went out with a message.
instructions_sent: bool,
}
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<Worktree>,
/// The worktree to create when the first message is sent.
pub pending_worktree: Option<WorktreePlan>,
pub pull_request: Option<PullRequest>,
/// The branch the thread last ran on, when it runs in the project's
/// checkout (not a worktree of its own).
pub branch: Option<String>,
/// Set for a thread mirrored from another device.
pub remote: Option<RemoteLink>,
/// Set for a subagent's thread: a task another thread delegated.
pub delegation: Option<Delegation>,
pub items: Vec<ChatItem>,
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<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,
/// The access mode plan mode hands back to when it's switched off.
pub plan_return: 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>,
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<SystemTime>,
/// When the thread was settled, for the Settled shelf's "settled 2h ago".
pub settled_at: Option<SystemTime>,
/// Something happened the user hasn't seen: a turn finished while the
/// thread wasn't open, or they marked it unread.
pub unread: bool,
/// Woke from a snooze and the user hasn't looked yet.
pub woke: bool,
/// Whether the app may settle the thread on its own (Auto-settle behavior).
pub auto_settle: bool,
/// Taken out of the sidebar; kept under Settings → Archive.
pub archived: bool,
/// A title is being regenerated.
pub regenerating_title: bool,
/// When a usage limit that stopped the thread resets.
pub limit_resets: Option<SystemTime>,
/// When to continue the thread on its own, after a usage limit.
pub resume_at: Option<SystemTime>,
/// Messages held until the running turn ends.
pub queue: Vec<QueuedMessage>,
/// 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,
/// The worktree being made for the first message, while it's made (or
/// when its setup failed or was cancelled).
pub worktree_setup: Option<WorktreeSetup>,
/// Tokens in the context window as of the last response.
pub context_tokens: Option<u64>,
pub context_window: Option<u64>,
/// How fast the last response came out, in output tokens a second.
pub tokens_per_sec: Option<f64>,
/// When the response being streamed sent its first token, and the
/// output tokens it has reported so far.
stream_clock: Option<(Instant, 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,
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<String>,
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<WorktreePlan>,
}
impl EventEmitter<SessionEvent> for Session {}
impl Session {
pub fn new(new: NewSession, cx: &mut Context<Self>) -> 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
}
/// A subagent's thread: it runs a task `parent` delegated, in the
/// parent's working folder, and starts on the task right away.
pub fn new_subagent(new: NewSession, parent: &SubagentParent, title: Option<String>, task: String, cx: &mut Context<Self>) -> 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 mut session = Self::empty(id, new, now);
session.cwd = parent.cwd.clone();
session.branch = parent.branch.clone();
session.delegation = Some(Delegation { parent: parent.id.clone(), delivered: false, cancelled: false });
if let Some(title) = title.filter(|title| !title.trim().is_empty()) {
session.title = title.trim().to_string().into();
session.title_custom = true;
}
session.send_message(task, Vec::new(), cx);
session
}
/// A subagent's thread, which the sidebar leaves out.
pub fn is_subagent(&self) -> bool {
self.delegation.is_some()
}
/// What the thread's agent last said: the end of its latest answer.
pub fn last_answer(&self) -> Option<String> {
self.items.iter().rev().find_map(|item| match item {
ChatItem::Assistant(answer) if !answer.text.trim().is_empty() => Some(answer.text.clone()),
_ => None,
})
}
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,
branch: None,
remote: None,
delegation: None,
items: Vec::new(),
status: SessionStatus::Starting,
thinking: false,
resume_id: None,
model,
active_model: None,
permission_mode,
plan_return: if permission_mode == PermissionMode::Plan { PermissionMode::Default } else { 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,
settled_at: None,
unread: false,
woke: false,
auto_settle: true,
archived: false,
regenerating_title: false,
limit_resets: None,
resume_at: None,
queue: Vec::new(),
interrupted: false,
preparing: false,
worktree_setup: None,
context_tokens: None,
context_window: None,
tokens_per_sec: None,
stream_clock: 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<String>, cx: &mut Context<Self>) {
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: settings::get_for(&self.project, |settings| settings.agent_browser)
.then(|| browser::session_name(&self.id)),
thread: self.id.clone(),
tools: self.tool_caps(),
};
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()
.filter(|_| config.browser_session.is_some())
.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<PathBuf>, cx: &mut Context<Self>) {
if self.remote.is_some() {
self.send_remote(text, attachments, false, cx);
return;
}
if text.trim().is_empty() && attachments.is_empty() {
return;
}
if !self.preparing {
self.worktree_setup = None;
}
let names: Vec<SharedString> = 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.settled_at = None;
self.snoozed_until = None;
self.woke = false;
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(worktree) = self.worktree.clone().filter(|worktree| !worktree.root.exists()) {
// Settings → Storage's cleanup removed it: bring it back from its
// branch, then carry on there.
self.status = SessionStatus::Working;
self.preparing = true;
self.turn_started = Some(Instant::now());
self.turn_cost = 0.;
let task = cx.background_executor().spawn(async move { worktree::restore(&worktree) });
cx.spawn(async move |this, cx| {
let restored = task.await;
let _ = this.update(cx, |this, cx| {
this.preparing = false;
if let Err(err) = restored {
this.push_notice(NoticeKind::Error, format!("Couldn't bring the worktree back: {err}"));
}
// The agent ran in the removed folder; it resumes in the new one.
this.process.take();
this._events_task = Task::ready(());
this.pending_controls.clear();
this.snapshot_base(cx);
this.deliver(text, attachments, cx);
cx.notify();
});
})
.detach();
cx.notify();
return;
}
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.;
self.worktree_setup = Some(WorktreeSetup::new(&plan));
let (progress_tx, progress_rx) = smol::channel::unbounded();
let task = cx.background_executor().spawn(async move {
worktree::create(&plan, &|stage, status| {
let _ = progress_tx.try_send((stage, status));
})
});
cx.spawn(async move |this, cx| {
while let Ok((stage, status)) = progress_rx.recv().await {
let alive = this.update(cx, |this, cx| {
if let Some(setup) = this.worktree_setup.as_mut() {
setup.mark(stage, status);
cx.notify();
}
});
if alive.is_err() {
return;
}
}
})
.detach();
let first_message = text.clone();
cx.spawn(async move |this, cx| {
let created = task.await;
let _ = this.update(cx, |this, cx| this.worktree_created(created, first_message, text, attachments, cx));
})
.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();
}
/// Tells the agent something the user doesn't need to see, like a
/// subagent's result: it reaches the agent like a message (mid-turn, as
/// soon as it can take it in; otherwise it starts a turn), but the
/// transcript only shows the agent's reply.
pub fn send_hidden(&mut self, text: String, cx: &mut Context<Self>) {
if self.remote.is_some() || text.trim().is_empty() {
return;
}
self.updated = SystemTime::now();
self.resume_at = None;
self.interrupted = false;
if !self.is_working() {
self.turn_started = Some(Instant::now());
self.turn_cost = 0.;
self.snapshot_base(cx);
}
self.deliver(text, Vec::new(), cx);
cx.notify();
}
/// Stops one subagent of the thread's, by the panel's Stop button.
pub fn stop_subagent(&mut self, stop: &SubagentStop, cx: &mut Context<Self>) {
match stop {
SubagentStop::Task(task_id) => self.stop_task(task_id, cx),
SubagentStop::Turn => self.interrupt(cx),
SubagentStop::SwarmAgent(agent) => {
self.stop_swarm_agent(agent);
if let Some(member) = self.swarm_member_mut(agent) {
member.set_status(TaskStatus::Stopped);
}
cx.notify();
}
// The workspace stops a delegated task's thread.
SubagentStop::Thread(_) => {}
}
}
/// Cancels a swarm agent's turn. jcode's API cancels only the session a
/// connection is attached to, so this attaches a short connection of its
/// own to the agent's session.
fn stop_swarm_agent(&self, agent: &str) {
let instance = settings::instance(&self.instance, self.provider);
let agent = agent.to_string();
let _ = std::thread::Builder::new().name("swarm-stop".into()).spawn(move || {
use std::io::{BufRead as _, Write as _};
let Ok((_, mut command)) = crate::claude::base_command(&instance) else { return };
command.args(crate::claude::split_args(&instance.args));
command
.args(["--no-update", "api", "--stdio"])
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null());
let Ok(mut child) = command.spawn() else { return };
let pid = child.id() as i32;
// Don't hang around if jcode doesn't answer.
let _ = std::thread::spawn(move || {
std::thread::sleep(Duration::from_secs(15));
unsafe { libc::kill(pid, libc::SIGTERM) };
});
if let Some(mut stdin) = child.stdin.take() {
let requests = [
json!({"v": 1, "id": 1, "req": "hello", "min_version": 1, "max_version": 1, "client": JCODE_CLIENT}),
json!({"v": 1, "id": 2, "req": "attach_session", "session_id": agent}),
json!({"v": 1, "id": 3, "req": "cancel", "session_id": agent}),
];
for request in requests {
let _ = writeln!(stdin, "{request}");
}
let _ = stdin.flush();
// Kept open until the cancel is through.
if let Some(stdout) = child.stdout.take() {
for line in std::io::BufReader::new(stdout).lines().map_while(Result::ok) {
let reply = serde_json::from_str::<Value>(&line).unwrap_or_default();
if reply.get("reply_to").and_then(Value::as_u64) == Some(3) {
break;
}
}
}
}
let _ = child.kill();
let _ = child.wait();
});
}
/// Holds a message until the running turn ends; sends it now if none runs.
pub fn queue_message(&mut self, text: String, attachments: Vec<PathBuf>, cx: &mut Context<Self>) {
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<Self>) {
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<Self>) {
if self.forward(ThreadAction::RemoveQueued { ix }) {
return;
}
if ix < self.queue.len() {
self.queue.remove(ix);
cx.notify();
}
}
/// Replaces a queued message with an edited one, in place.
pub fn update_queued(&mut self, ix: usize, text: String, attachments: Vec<PathBuf>, cx: &mut Context<Self>) {
if self.remote.is_some() {
return;
}
if let Some(message) = self.queue.get_mut(ix) {
*message = QueuedMessage { text, attachments };
cx.notify();
}
}
/// Moves a queued message up or down the queue.
pub fn move_queued(&mut self, from: usize, to: usize, cx: &mut Context<Self>) {
if self.remote.is_some() || from >= self.queue.len() || to >= self.queue.len() || from == to {
return;
}
let message = self.queue.remove(from);
self.queue.insert(to, message);
cx.notify();
}
/// Sends the next queued message once a turn has ended.
fn send_next_queued(&mut self, cx: &mut Context<Self>) {
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<Self>) {
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<PathBuf>, cx: &mut Context<Self>) {
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<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"] = "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});
// jcode's system prompt can't be added to, so the app's
// instructions go along with the first message.
let instructions = (!self.jcode.instructions_sent)
.then(|| orchestrator::instructions(self.tool_caps()))
.filter(|instructions| !instructions.is_empty());
self.jcode.instructions_sent = true;
let reminder: Vec<String> = instructions.into_iter().chain(self.jcode.browser_note.take()).collect();
if !reminder.is_empty() {
request["system_reminder"] = reminder.join("\n\n").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>) {
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<Self>) {
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<Self>) {
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<Self>) {
if settings::get_for(&self.project, |settings| settings.text_model.is_empty()) {
return;
}
let project = self.project.clone();
let task = cx.background_executor().spawn(async move { crate::textgen::title(&project, &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();
}
/// Takes the worktree the first message asked for, or where its setup
/// was cancelled or failed, carries on without it; then hands the
/// message to the agent (unless cancelled).
fn worktree_created(
&mut self,
created: Result<(Worktree, PathBuf), String>,
first_message: String,
text: String,
attachments: Vec<PathBuf>,
cx: &mut Context<Self>,
) {
self.preparing = false;
let cancel = self.worktree_setup.as_ref().and_then(|setup| setup.cancel);
match (created, cancel) {
(Ok((worktree, _)), Some(_)) => {
// Cancelled while git was busy: undo it.
cx.background_executor()
.spawn(async move {
let _ = worktree::remove(&worktree);
let _ = git::git(&worktree.repo, &["branch", "-D", &worktree.branch]);
})
.detach();
}
(Ok((worktree, cwd)), None) => {
self.cwd = cwd;
if let Some(setup) = self.worktree_setup.as_mut() {
setup.branch = Some(worktree.branch.clone());
setup.path = Some(worktree.root.clone());
}
self.name_branch(&worktree, first_message, cx);
self.worktree = Some(worktree);
}
(Err(err), _) => {
if let Some(setup) = self.worktree_setup.as_mut() {
setup.error = Some(err.clone());
}
self.push_notice(
NoticeKind::Error,
format!("Couldn't create a worktree: {err}\n\nThe thread runs in the project folder instead."),
);
}
}
if cancel == Some(SetupCancel::Cancel) {
if let Some(setup) = self.worktree_setup.as_mut() {
setup.cancelled = true;
}
self.status = SessionStatus::Idle;
self.turn_started = None;
self.start(None, cx);
cx.notify();
return;
}
if let Some(setup) = self.worktree_setup.as_mut() {
setup.mark(SetupStage::Agent, StageStatus::Running);
}
self.start(None, cx);
self.snapshot_base(cx);
self.deliver(text, attachments, cx);
// A clean setup leaves the conversation once the agent has it.
match self.worktree_setup.as_mut() {
Some(setup) if setup.error.is_none() && setup.cancel.is_none() => self.worktree_setup = None,
Some(setup) => setup.mark(SetupStage::Agent, StageStatus::Done),
None => {}
}
cx.notify();
}
/// Stops waiting for the worktree: `Cancel` keeps the message unsent,
/// `Local` sends it in the project folder instead.
pub fn cancel_worktree_setup(&mut self, mode: SetupCancel, cx: &mut Context<Self>) {
if let Some(setup) = self.worktree_setup.as_mut().filter(|_| self.preparing) {
setup.cancel = Some(mode);
cx.notify();
}
}
/// Renames a new worktree's temporary branch after the thread's first
/// message, as Settings → Source Control → Worktree branch naming says.
/// The folder keeps its name.
fn name_branch(&mut self, worktree: &Worktree, message: String, cx: &mut Context<Self>) {
if settings::get_for(&self.project, |settings| settings.text_model.is_empty()) || message.trim().is_empty() {
return;
}
let project = self.project.clone();
let (naming, prefix, instructions) = settings::get_for(&self.project, |settings| {
(settings.branch_naming, settings.branch_prefix.clone(), settings.branch_instructions.clone())
});
let worktree = worktree.clone();
let task = cx.background_executor().spawn(async move {
let suggestion = crate::textgen::branch_name(&project, &message, naming, &instructions)?;
let name = worktree::branch_from_suggestion(&suggestion, naming, &prefix);
worktree::rename_branch(&worktree, &name, naming != BranchNaming::Custom).ok()
});
cx.spawn(async move |this, cx| {
let Some(branch) = task.await else { return };
let _ = this.update(cx, |this, cx| {
if let Some(worktree) = this.worktree.as_mut() {
worktree.branch = branch;
cx.notify();
}
});
})
.detach();
}
/// Writes a new title from the first message with the text generation
/// model, replacing a custom one.
pub fn regenerate_title(&mut self, cx: &mut Context<Self>) {
let Some(message) = self.first_user_text() else { return };
if self.regenerating_title {
return;
}
self.regenerating_title = true;
cx.notify();
let project = self.project.clone();
let task = cx.background_executor().spawn(async move { crate::textgen::title(&project, &message) });
cx.spawn(async move |this, cx| {
let title = task.await;
let _ = this.update(cx, |this, cx| {
this.regenerating_title = false;
if let Some(title) = title {
this.title = title.into();
this.title_custom = false;
}
cx.notify();
});
})
.detach();
}
/// The branch a new worktree starts from.
pub fn set_worktree_base(&mut self, base: String, cx: &mut Context<Self>) {
if let Some(plan) = self.pending_worktree.as_mut() {
plan.base = Some(base);
cx.notify();
}
}
/// The text of the first message the user sent.
pub fn first_user_text(&self) -> Option<String> {
self.items.iter().find_map(|item| match item {
ChatItem::User { text, attachments, .. } if !text.trim().is_empty() || !attachments.is_empty() => {
Some(if text.trim().is_empty() { attachments.join(", ") } else { text.to_string() })
}
_ => None,
})
}
pub fn set_unread(&mut self, unread: bool, cx: &mut Context<Self>) {
if self.unread != unread {
self.unread = unread;
cx.notify();
}
}
/// Whether a new worktree starts from origin's copy of its branch.
pub fn set_worktree_from_origin(&mut self, on: bool, cx: &mut Context<Self>) {
if let Some(plan) = self.pending_worktree.as_mut() {
plan.from_origin = on;
cx.notify();
}
}
/// Acknowledges a thread the user opened: it's read, and no longer woke.
pub fn mark_seen(&mut self, cx: &mut Context<Self>) {
if self.unread || self.woke {
self.unread = false;
self.woke = false;
cx.notify();
}
}
/// Points the (still empty) thread at a worktree that already exists,
/// such as the one a previous thread worked in, or back at the project
/// folder with `None`.
pub fn use_worktree(&mut self, worktree: Option<Worktree>, cx: &mut Context<Self>) {
if self.remote.is_some() || !self.items.is_empty() {
return;
}
let inside = |repo: &Path| self.project.strip_prefix(repo).map(Path::to_path_buf).unwrap_or_default();
let cwd = match &worktree {
Some(worktree) => worktree.root.join(inside(&worktree.repo)),
None => self.project.clone(),
};
self.pending_worktree = None;
self.worktree = worktree;
if cwd != self.cwd {
self.cwd = cwd;
// Restart the agent where the thread now runs.
self.process.take();
self._events_task = Task::ready(());
self.pending_controls.clear();
self.status = SessionStatus::Idle;
self.start(None, cx);
}
cx.notify();
}
pub fn dismiss_woke(&mut self, cx: &mut Context<Self>) {
if self.woke {
self.woke = false;
cx.notify();
}
}
pub fn set_auto_settle(&mut self, on: bool, cx: &mut Context<Self>) {
self.auto_settle = on;
cx.notify();
}
/// Takes the thread out of the sidebar, or brings it back.
pub fn set_archived(&mut self, archived: bool, cx: &mut Context<Self>) {
self.archived = archived;
if archived {
self.pinned = false;
}
cx.notify();
}
/// Ends a snooze because its time came, so the sidebar says it woke.
pub fn wake(&mut self, cx: &mut Context<Self>) {
self.snooze(None, cx);
self.woke = true;
cx.notify();
}
pub fn rename(&mut self, title: String, cx: &mut Context<Self>) {
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<WorktreePlan>, cx: &mut Context<Self>) {
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();
}
/// Stops the agent process; the next message starts a fresh one that
/// resumes the conversation. Does nothing mid-turn.
pub fn restart_agent(&mut self, cx: &mut Context<Self>) -> bool {
if self.remote.is_some() || self.is_working() || self.process.is_none() {
return false;
}
self.process.take();
self._events_task = Task::ready(());
self.pending_controls.clear();
self.drop_background_tasks();
if self.status != SessionStatus::Exited {
self.status = SessionStatus::Idle;
}
cx.notify();
true
}
/// Moves a fresh thread into another thread's worktree, as "New thread on
/// {branch}" does for a worktree thread, restarting its agent there.
pub fn adopt_worktree(&mut self, worktree: Worktree, cwd: PathBuf, cx: &mut Context<Self>) {
if !self.items.is_empty() || self.remote.is_some() {
return;
}
self.process.take();
self._events_task = Task::ready(());
self.pending_controls.clear();
self.pending_worktree = None;
self.cwd = cwd;
self.worktree = Some(worktree);
self.start(None, cx);
cx.notify();
}
pub fn set_pinned(&mut self, pinned: bool, cx: &mut Context<Self>) {
self.forward(ThreadAction::Pin { pinned });
self.pinned = pinned;
cx.notify();
}
pub fn snooze(&mut self, until: Option<SystemTime>, cx: &mut Context<Self>) {
self.forward(ThreadAction::Snooze { until: until.map(to_millis) });
self.snoozed_until = until;
if until.is_none() {
self.updated = SystemTime::now();
} else {
self.woke = false;
}
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<SystemTime> {
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>) {
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<Self>) {
// Settling clears the pin: done beats keep on top.
self.pinned = false;
self.settled_at = Some(SystemTime::now());
self.woke = false;
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>) {
self.forward(ThreadAction::Unsettle);
self.settled = false;
self.settled_at = None;
self.updated = SystemTime::now();
cx.notify();
}
pub fn interrupt(&mut self, cx: &mut Context<Self>) {
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<String>, cx: &mut Context<Self>) {
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<Self>) {
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();
}
/// The access mode the composer shows: the thread's own, or while it
/// plans, the one plan mode hands back to.
pub fn access_mode(&self) -> PermissionMode {
if !self.provider.asks_permission() {
PermissionMode::BypassPermissions
} else if self.permission_mode == PermissionMode::Plan {
self.plan_return
} else {
self.permission_mode
}
}
/// Picks the access mode; while the thread plans, it applies once plan
/// mode is switched off.
pub fn set_access_mode(&mut self, mode: PermissionMode, cx: &mut Context<Self>) {
if self.permission_mode == PermissionMode::Plan {
self.plan_return = mode;
cx.notify();
} else {
self.set_permission_mode(mode, cx);
}
}
pub fn is_planning(&self) -> bool {
self.permission_mode == PermissionMode::Plan
}
/// Switches between plan mode and building with the access mode.
pub fn toggle_plan(&mut self, cx: &mut Context<Self>) {
if self.is_planning() {
let mode = self.plan_return;
self.set_permission_mode(mode, cx);
} else {
self.plan_return = self.permission_mode;
self.set_permission_mode(PermissionMode::Plan, cx);
}
}
pub fn set_reasoning(&mut self, reasoning: Reasoning, cx: &mut Context<Self>) {
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<Self>) {
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<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, 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<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 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,
}
}
/// Selects an option on the current question: the only one on a
/// single-select question, one more (or one fewer) on a multi-select one.
/// Either way it replaces what the user typed as their own answer.
/// Returns whether the question is single-select, which moves on by itself.
pub fn ask_toggle(&mut self, ix: usize, option: usize, cx: &mut Context<Self>) -> bool {
let Some(ask) = self.pending_ask_mut(ix) else { return false };
let current = ask.current;
let question = &ask.questions[current];
if option >= question.options.len() {
return false;
}
let multi = question.multi_select;
let picked = &mut ask.picked[current];
if multi {
match picked.iter().position(|&picked| picked == option) {
Some(position) => {
picked.remove(position);
}
None => picked.push(option),
}
} else {
*picked = vec![option];
}
ask.custom[current].clear();
cx.notify();
!multi
}
/// Takes what the composer holds as the current question's own answer.
/// Typing one deselects the options; clearing it brings them back.
pub fn ask_set_custom(&mut self, ix: usize, text: String, cx: &mut Context<Self>) {
let Some(ask) = self.pending_ask_mut(ix) else { return };
let current = ask.current;
if ask.custom[current] != text {
if !text.trim().is_empty() && ask.questions[current].multi_select {
ask.picked[current].clear();
}
ask.custom[current] = text;
cx.notify();
}
}
/// Moves on from the current question once it has an answer: to the next
/// question, or, on the last one, submits the form when it's complete.
pub fn ask_advance(&mut self, ix: usize, cx: &mut Context<Self>) {
let Some(ask) = self.pending_ask_mut(ix) else { return };
if !ask.is_last() {
if ask.resolved(ask.current).is_some() {
ask.current += 1;
cx.notify();
}
return;
}
if !ask.is_complete() {
return;
}
let answers: Vec<String> = (0..ask.questions.len()).filter_map(|q| ask.resolved(q)).collect();
self.ask_submit(ix, answers, cx);
}
/// Goes back to the previous question.
pub fn ask_previous(&mut self, ix: usize, cx: &mut Context<Self>) {
if let Some(ask) = self.pending_ask_mut(ix)
&& ask.current > 0
{
ask.current -= 1;
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<String>, cx: &mut Context<Self>) {
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<Self>) {
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<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>) {
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<Self>) {
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<i32>, cx: &mut Context<Self>) {
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<f64>, cx: &mut Context<Self>) {
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<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("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<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);
}
}
}
// A subagent's running totals, as it works.
Some("task_progress") => {
let Some(tool) = self.find_tool_mut(text("tool_use_id").unwrap_or_default()) else { return };
let uses = message.pointer("/usage/tool_uses").and_then(Value::as_u64);
let tokens = message.pointer("/usage/total_tokens").and_then(Value::as_u64);
if let Some(uses) = uses {
tool.subagent_tool_count = tool.subagent_tool_count.max(uses as usize);
}
if let Some(progress) = tool.subagent.as_mut() {
progress.tokens = tokens.or(progress.tokens);
// A background subagent's own messages don't come through.
if let Some(name) = text("last_tool_name")
&& progress.tool.as_ref().is_none_or(|(tool, _)| tool != name)
{
progress.tool = Some((name.to_string(), Value::Null));
}
if let Some(summary) = text("summary").filter(|summary| !summary.trim().is_empty()) {
progress.summary = Some(summary.trim().to_string());
}
}
}
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);
}
if let Some(progress) = tool.subagent.as_mut() {
if let Some(tokens) = message.pointer("/usage/total_tokens").and_then(Value::as_u64) {
progress.tokens = Some(tokens);
}
if finished && tool.background.is_some() {
progress.finish();
}
}
}
// 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<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() });
self.stream_clock = None;
}
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") => {
self.stream_clock.get_or_insert((Instant::now(), 0));
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::<Value>(&tool.partial_input)
{
tool.input = input;
}
}
Some("message_delta") => {
if let (Some(output), Some(clock)) =
(event.pointer("/usage/output_tokens").and_then(Value::as_u64), self.stream_clock.as_mut())
{
clock.1 = output;
}
}
Some("message_stop") => {
self.thinking = false;
self.streaming = None;
self.finish_speed(None);
}
_ => {}
}
}
/// Works out how fast the response that just finished came out, from
/// its first token to its last.
fn finish_speed(&mut self, output: Option<u64>) {
let Some((started, reported)) = self.stream_clock.take() else { return };
let output = output.unwrap_or(reported);
let secs = started.elapsed().as_secs_f64();
// Too short a response says little about the model's speed.
if output >= 16 && secs >= 0.25 {
self.tokens_per_sec = Some(output as f64 / secs);
}
}
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(),
);
}
}
}
_ => {}
}
}
}
/// A subagent's own message: counts its tool calls and notes the latest.
fn count_subagent_tools(&mut self, parent: &str, message: &Value) {
let calls: Vec<&Value> = message
.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use"))
.collect();
if let Some(tool) = self.find_tool_mut(parent) {
tool.subagent_tool_count += calls.len();
if let (Some(progress), Some(call)) = (tool.subagent.as_mut(), calls.last()) {
let name = call.get("name").and_then(Value::as_str).unwrap_or("Tool").to_string();
progress.tool = Some((name, brief_input(&call["input"])));
}
}
}
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 };
// A background subagent's tool returns as it starts.
if (is_error || tool.background.is_none())
&& let Some(progress) = tool.subagent.as_mut()
{
progress.finish();
}
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>) {
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<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_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<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() });
self.stream_clock = None;
}
Some("message_update") => {
self.stream_clock.get_or_insert((Instant::now(), 0));
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;
self.finish_speed(message.pointer("/usage/output").and_then(Value::as_u64));
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),
);
}
}
// Pi's `subagent` extension reports its tasks as they go.
Some("tool_execution_update") => {
let id = message.get("toolCallId").and_then(Value::as_str).unwrap_or_default();
if let Some(tool) = self.find_tool_mut(id) {
pi_subagent_results(tool, message.pointer("/partialResult/details"));
}
}
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) {
pi_subagent_results(tool, message.pointer("/result/details"));
tool.result = Some(text);
tool.state = if is_error { ToolState::Error } else { ToolState::Done };
if let Some(progress) = tool.subagent.as_mut() {
progress.finish();
}
}
}
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<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
}
};
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<Self>) {
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);
}
self.configure_jcode_tools();
// Swarm agents carry on in jcode while the app is away.
self.watch_swarm(cx);
}
"sessions" if request == Some(LIST_AGENTS) => self.update_swarm(&message["sessions"]),
"history" if request == Some(PEEK_AGENT) => {
let agent = text("session_id");
let said = message
.get("messages")
.and_then(Value::as_array)
.into_iter()
.flatten()
.rev()
.filter(|message| message.get("role").and_then(Value::as_str) == Some("assistant"))
.find_map(|message| last_line(message.get("content").and_then(Value::as_str).unwrap_or_default()));
if let Some(member) = self.swarm_member_mut(agent) {
member.summary = said.or(member.summary.take());
}
}
"error" => {
let error = text("message");
match request {
// Checking on swarm agents is best effort, and repeats.
Some(LIST_AGENTS | PEEK_AGENT) => {}
// Mid-turn, it waits for the turn to end. Daemons too old
// for session tools go without.
Some(CONFIGURE_TOOLS) => self.jcode.tools_pending = error.contains("busy"),
// 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),
"tool_call" => cx.emit(SessionEvent::AppTool {
call_id: text("call_id").to_string(),
tool: text("name").to_string(),
input: message.get("input").cloned().unwrap_or(Value::Null),
}),
"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 };
// A swarm spawn names the agent's session: watch it from here.
let spawned = (tool.name == SWARM_TOOL && error.is_none())
.then(|| output.trim().strip_prefix("Spawned new agent:").map(|id| id.trim().to_string()))
.flatten()
.filter(|id| !id.is_empty());
if let (Some(agent), Some(progress)) = (spawned, tool.subagent.as_mut()) {
let label = tool_field(&tool.input, &["label", "prompt"])
.and_then(|label| label.lines().next().map(|line| line.chars().take(200).collect()))
.unwrap_or_else(|| "Agent".into());
progress.members.push(SubagentMember {
id: agent,
label,
kind: tool_field(&tool.input, &["role"]).filter(|role| role != "agent"),
status: TaskStatus::Running.key().to_string(),
started: progress.started,
..Default::default()
});
}
tool.result = Some(match error {
Some(error) if output.trim().is_empty() => error,
Some(error) => format!("{output}\n\n{error}"),
None => output,
});
self.watch_swarm(cx);
}
"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);
if std::mem::take(&mut self.jcode.tools_pending) {
self.configure_jcode_tools();
}
self.send_next_queued(cx);
}
_ => {}
}
}
/// Registers the app's tools with the thread's jcode session (a
/// subagent's thread gets none) and switches off jcode's own ways of
/// spawning agents. jcode keeps the setup in memory, so it's sent on
/// every attach.
fn configure_jcode_tools(&mut self) {
let tools = orchestrator::jcode_tools(self.tool_caps());
self.jcode_request(CONFIGURE_TOOLS, json!({"req": "configure_tools", "tools": tools}));
}
/// Which of the app's tools this thread's agent gets.
pub fn tool_caps(&self) -> orchestrator::Caps {
orchestrator::Caps { delegate: self.delegation.is_none(), devices: crate::device::agent_access_for(&self.project) }
}
/// Answers a call jcode made to one of the app's tools.
pub fn answer_app_tool(&mut self, call_id: &str, outcome: Result<Value, String>) {
let (output, error) = match outcome {
Ok(value) => (serde_json::to_string_pretty(&value).unwrap_or_default(), None),
Err(error) => (String::new(), Some(error)),
};
let request = json!({"req": "tool_result", "call_id": call_id, "output": output, "error": error});
self.jcode_request("answer a tool call", request);
}
/// Ties a task the app started for this thread to the `delegate_task`
/// call that asked for it: the latest one with that task and no
/// subagent yet.
pub fn link_delegation(&mut self, args: &Value, member: SubagentMember, cx: &mut Context<Self>) {
let task = args.get("task").and_then(Value::as_str);
let unlinked = |tool: &ToolCall| {
tool.is_delegation() && tool.subagent.as_ref().is_some_and(|progress| progress.members.is_empty())
};
let ix = self
.items
.iter()
.rposition(|item| {
matches!(item, ChatItem::Tool(tool) if unlinked(tool) && tool.input.get("task").and_then(Value::as_str) == task)
})
.or_else(|| self.items.iter().rposition(|item| matches!(item, ChatItem::Tool(tool) if unlinked(tool))));
if let Some(ChatItem::Tool(tool)) = ix.and_then(|ix| self.items.get_mut(ix))
&& let Some(progress) = tool.subagent.as_mut()
{
progress.members.push(member);
cx.notify();
}
}
/// Updates the subagent row of a task this thread delegated.
pub fn update_delegation(&mut self, task: &str, update: impl FnOnce(&mut SubagentMember), cx: &mut Context<Self>) {
let member = self.items.iter_mut().find_map(|item| match item {
ChatItem::Tool(tool) if tool.is_delegation() => {
tool.subagent.as_mut()?.members.iter_mut().find(|member| member.id == task)
}
_ => None,
});
if let Some(member) = member {
let before = member.clone();
update(member);
if *member != before {
cx.notify();
}
}
}
/// 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<Self>) -> 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,
})
}
/// The swarm agents this thread spawned, across its swarm calls.
fn swarm_members(&self) -> impl Iterator<Item = &SubagentMember> {
self.items.iter().filter_map(|item| match item {
ChatItem::Tool(tool) if tool.name == SWARM_TOOL => tool.subagent.as_ref(),
_ => None,
})
.flat_map(|progress| &progress.members)
}
fn swarm_member_mut(&mut self, agent: &str) -> Option<&mut SubagentMember> {
self.items
.iter_mut()
.filter_map(|item| match item {
ChatItem::Tool(tool) if tool.name == SWARM_TOOL => tool.subagent.as_mut(),
_ => None,
})
.flat_map(|progress| progress.members.iter_mut())
.find(|member| member.id == agent)
}
/// Whether the thread has swarm agents to keep an eye on: ones that work,
/// or a spawn that hasn't named its agent yet.
fn swarm_busy(&self) -> bool {
self.swarm_members().any(|member| member.status == TaskStatus::Running.key())
|| self.items.iter().any(|item| {
matches!(item, ChatItem::Tool(tool) if tool.name == SWARM_TOOL && tool.state == ToolState::Running)
})
}
/// Starts checking on the thread's swarm agents every few seconds, if
/// any work and it isn't already. jcode's API doesn't stream another
/// session's progress, so the app asks: the session list says where each
/// agent stands, and a peek at its transcript what it last said.
fn watch_swarm(&mut self, cx: &mut Context<Self>) {
if self.jcode.agent_poll.is_some() || !self.swarm_busy() {
return;
}
self.jcode.agent_poll = Some(cx.spawn(async move |this, cx| {
loop {
cx.background_executor().timer(SWARM_POLL).await;
let busy = this.update(cx, |this, _| {
let busy = this.process.is_some() && this.swarm_busy();
if busy {
this.poll_swarm();
} else if let Some(poll) = this.jcode.agent_poll.take() {
// This very loop: let it end on its own.
poll.detach();
}
busy
});
if !matches!(busy, Ok(true)) {
break;
}
}
}));
}
fn poll_swarm(&mut self) {
self.jcode_send(LIST_AGENTS, json!({"req": "list_sessions", "limit": 200}));
let working: Vec<String> = self
.swarm_members()
.filter(|member| member.status == TaskStatus::Running.key())
.map(|member| member.id.clone())
.collect();
for agent in working {
self.jcode_send(PEEK_AGENT, json!({"req": "peek_session", "session_id": agent, "limit": 4}));
}
}
/// Takes in where the thread's swarm agents stand, from jcode's session
/// list. Agents another agent of the thread spawned list too, under the
/// latest swarm call.
fn update_swarm(&mut self, sessions: &Value) {
let Some(own) = self.jcode.session.clone() else { return };
let Some(sessions) = sessions.as_array() else { return };
let ours: HashSet<String> = self.swarm_members().map(|member| member.id.clone()).collect();
let mut family: HashSet<String> = ours.iter().cloned().chain([own.clone()]).collect();
// Spawned by this session, or by one of its agents.
loop {
let before = family.len();
for session in sessions {
let parent = session.get("parent_session_id").and_then(Value::as_str).unwrap_or_default();
if let Some(id) = session.get("session_id").and_then(Value::as_str)
&& family.contains(parent)
{
family.insert(id.to_string());
}
}
if family.len() == before {
break;
}
}
for session in sessions {
let Some(id) = session.get("session_id").and_then(Value::as_str) else { continue };
if id == own || !family.contains(id) {
continue;
}
let text = |key: &str| session.get(key).and_then(Value::as_str).filter(|text| !text.trim().is_empty());
let status = jcode_agent_status(text("swarm_status"), text("status"));
if self.swarm_member_mut(id).is_none() {
// An agent the app didn't see spawned: file it under the latest swarm call.
let Some(progress) = self.items.iter_mut().rev().find_map(|item| match item {
ChatItem::Tool(tool) if tool.name == SWARM_TOOL => tool.subagent.as_mut(),
_ => None,
}) else {
return;
};
progress.members.push(SubagentMember {
id: id.to_string(),
started: to_millis(SystemTime::now()),
..Default::default()
});
}
let Some(member) = self.swarm_member_mut(id) else { continue };
if let Some(label) = text("agent_label").or(text("title")) {
member.label = label.to_string();
}
if member.label.is_empty() {
member.label = "Agent".into();
}
member.set_status(status);
}
}
// ---- helpers -----------------------------------------------------
fn push_assistant_text(&mut self, text: &str, cx: &mut Context<Self>) -> 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<Self>) {
let streaming = settings::get(|settings| settings.streaming);
let Some(ChatItem::Assistant(answer)) = self.items.get_mut(ix) else { return };
answer.text.push_str(delta);
let target = match streaming {
Streaming::Tokens => answer.text.len(),
Streaming::Paragraphs => finished_prefix(&answer.text),
Streaming::Turn => return,
};
answer.show(target, cx);
}
/// Shows the rest of an answer's text once its message is complete,
/// unless answers wait for the end of the turn.
fn flush_item(&mut self, ix: usize, cx: &mut Context<Self>) {
if settings::get(|settings| settings.streaming) != Streaming::Turn {
self.show_all(ix, cx);
}
}
fn show_all(&mut self, ix: usize, cx: &mut Context<Self>) {
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<Self>) {
for ix in 0..self.items.len() {
self.show_all(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,
subagent: is_subagent_tool(name)
.then(|| SubagentProgress { started: to_millis(SystemTime::now()), ..Default::default() }),
}));
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);
if let Some(progress) = tool.subagent.as_mut() {
progress.finish();
}
}
}
}
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());
if let Some(progress) = tool.subagent.as_mut() {
progress.finish();
}
}
}
}
}
// ---- 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<Worktree>,
#[serde(default)]
pub pending_worktree: Option<WorktreePlan>,
#[serde(default)]
pub pull_request: Option<PullRequest>,
/// The branch the thread last ran on, in the project's checkout.
#[serde(default)]
pub branch: Option<String>,
pub resume_id: Option<String>,
pub model: Option<String>,
pub active_model: Option<String>,
pub permission_mode: String,
pub reasoning: String,
pub fast_mode: bool,
pub total_cost: f64,
pub context_tokens: Option<u64>,
pub context_window: Option<u64>,
/// 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<u64>,
#[serde(default)]
pub settled_at: Option<u64>,
#[serde(default)]
pub unread: bool,
#[serde(default)]
pub woke: bool,
#[serde(default = "default_true")]
pub auto_settle: bool,
#[serde(default)]
pub archived: bool,
pub limit_resets: Option<u64>,
pub resume_at: Option<u64>,
pub queue: Vec<QueuedMessage>,
/// The agent was mid-turn when the thread was saved.
pub working: bool,
pub items: Vec<SavedItem>,
/// What each finished turn changed, by its `TurnEnd` index.
pub turn_changes: Vec<(usize, TurnChanges)>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub delegation: Option<Delegation>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum SavedItem {
User { text: String, attachments: Vec<String>, at: u64 },
Assistant { text: String },
Tool {
id: String,
name: String,
input: Value,
result: Option<String>,
/// `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<String>,
/// The rules a pending prompt offers to remember.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
suggestions: Vec<Value>,
subagent_tool_count: usize,
/// An AskUserQuestion form's answers so far, one per question.
#[serde(default, skip_serializing_if = "Option::is_none")]
answers: Option<Vec<Option<String>>>,
/// `running`, `completed`, `failed` or `stopped`, if it ran in the
/// background.
#[serde(default, skip_serializing_if = "Option::is_none")]
background: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
subagent: Option<SubagentProgress>,
},
Notice { error: bool, text: String },
TaskEvent { summary: String, status: String },
Question {
title: String,
message: Option<String>,
options: Vec<String>,
#[serde(default)]
confirm: bool,
answer: Option<String>,
},
TurnEnd { duration_ms: u64, cost: Option<f64>, at: u64 },
}
fn default_true() -> bool {
true
}
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<SavedItem> {
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()),
subagent: tool.subagent.clone(),
},
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<SavedItem>) -> 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(),
branch: self.branch.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),
settled_at: self.settled_at.map(to_millis),
unread: self.unread,
woke: self.woke,
auto_settle: self.auto_settle,
archived: self.archived,
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(),
delegation: self.delegation.clone(),
}
}
/// 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>) -> 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.branch = saved.branch;
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.settled_at = saved.settled_at.map(from_millis);
session.unread = saved.unread;
session.woke = saved.woke;
session.auto_settle = saved.auto_settle;
session.archived = saved.archived;
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();
session.delegation = saved.delegation;
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<Self>) -> 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,
subagent,
} => {
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)
});
// Threads saved before subagents were tracked still list them.
let subagent = subagent.or_else(|| is_subagent_tool(&name).then(SubagentProgress::default));
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,
// Threads saved before subagents were tracked still list them.
subagent,
})
}
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<Upload, String> {
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<Upload>) -> Vec<PathBuf> {
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>) -> 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<PathBuf>, queue: bool, cx: &mut Context<Self>) {
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<Self>) {
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;
}
// The host hands the task's result over; the mirror only stays out
// of the sidebar.
self.delegation =
summary.parent.clone().map(|parent| Delegation { parent, delivered: true, cancelled: false });
}
/// 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<SavedItem>,
cx: &mut Context<Self>,
) {
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(),
parent: self.delegation.as_ref().map(|delegation| delegation.parent.clone()),
}
}
/// 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>) {
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<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))
}
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::<Value>(&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::<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
}
#[cfg(test)]
mod tests {
use super::{TaskStatus, ToolCall, ToolState, finished_prefix, jcode_agent_status, pi_subagent_results};
use serde_json::{Value, json};
fn tool(name: &str, input: Value) -> ToolCall {
ToolCall {
id: "call".into(),
name: name.into(),
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,
subagent: super::is_subagent_tool(name).then(super::SubagentProgress::default),
}
}
#[::core::prelude::v1::test]
fn pi_parallel_subagents() {
let mut call = tool("subagent", json!({"tasks": [{"agent": "scout", "task": "Find it"}, {"agent": "fixer", "task": "Fix it"}]}));
let details = json!({"mode": "parallel", "results": [
{"agent": "scout", "task": "Find it\nthoroughly", "exitCode": -1,
"usage": {"input": 1000, "output": 200},
"messages": [{"role": "assistant", "content": [
{"type": "text", "text": "Looking around"},
{"type": "toolCall", "id": "a", "name": "read", "arguments": {"path": "src/main.rs", "content": 7}}]}]},
{"agent": "fixer", "task": "Fix it", "exitCode": 1, "stopReason": "error", "messages": []},
]});
pi_subagent_results(&mut call, Some(&details));
let entries = call.subagents();
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].label, "Find it");
assert_eq!(entries[0].kind.as_deref(), Some("scout"));
assert_eq!(entries[0].status, TaskStatus::Running);
assert_eq!(entries[0].tokens, Some(1200));
assert_eq!(entries[0].tool_uses, 1);
assert_eq!(entries[0].tool, Some(("read".into(), json!({"path": "src/main.rs"}))));
assert_eq!(entries[0].summary.as_deref(), Some("Looking around"));
assert_eq!(entries[1].status, TaskStatus::Failed);
// Once the call is over, a task still marked running was cut off.
call.state = ToolState::Done;
assert_eq!(call.subagents()[0].status, TaskStatus::Stopped);
}
#[::core::prelude::v1::test]
fn swarm_calls() {
// Only spawns start agents; talking to the swarm doesn't.
assert!(tool("swarm", json!({"action": "message", "message": "hi"})).subagents().is_empty());
let spawn = tool("swarm", json!({"action": "spawn", "label": "Refactor"}));
assert_eq!(spawn.subagents()[0].label, "Refactor");
assert_eq!(jcode_agent_status(Some("running"), None), TaskStatus::Running);
assert_eq!(jcode_agent_status(Some("ready"), Some("idle")), TaskStatus::Completed);
assert_eq!(jcode_agent_status(Some("crashed"), None), TaskStatus::Failed);
assert_eq!(jcode_agent_status(None, Some("running")), TaskStatus::Running);
}
#[::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());
}
}