entityseven/openidea
2
0 Fork 0
mirror of https://github.com/l0ng-ai/openidea.git synced 2026-11-08 17:02:17 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
93f87f49c176dfee983ab4dce2e2e5ffb35ed4c9
openidea/crates/ide-agent/src/lib.rs
T
l0ng-ai 93f87f49c1 feat(agents): Current branch and time spent in the agent session header 
A local-checkout task shows the repository's current branch in its header, and
Changes compares against it; both follow branch switches. The header also shows
the time the agent has spent on the task's turns.

A usage update without a cost (Claude sends one for its rate limits) no longer
clears the session cost.
2026-10-28 11:57:18 +08:00

1144 lines
45 KiB
Rust
Raw Blame History

//! Coding agents (Claude Code, Codex, Gemini CLI…) driven over the Agent Client Protocol.
//!
//! A task is one agent conversation, working in its own git worktree or in the project
//! checkout. Each task runs its agent's ACP server as a child process (newline-delimited
//! JSON-RPC over stdio), folds `session/update` notifications into a timeline, and streams
//! timeline and status changes to the subscribed window. Tasks are saved under the
//! project's store directory and survive restarts; the agent session is reloaded on the
//! next prompt when the agent supports `session/load`.
mod config;
mod model;
mod provider;
mod skills;
mod timeline;
mod worktree;
pub use model::*;
use ide_core::{IdeError, IdeResult, IpcPath};
use ide_jsonrpc::{Framing, Handler, Peer, RpcError, codes};
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::{BTreeSet, HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::{Arc, Weak};
use std::time::Duration;
use timeline::Timeline;
/// Receives the events of one project's tasks.
pub type AgentSink = Arc<dyn Fn(AgentEvent) + Send + Sync>;
/// Entry changes are batched for this long before they are sent.
const FLUSH_DELAY: Duration = Duration::from_millis(40);
/// Updates that arrive this late after `session/load` returns are still replayed history.
const REPLAY_GRACE: Duration = Duration::from_millis(300);
/// Notifications are delivered on their own thread: the last `session/update`s of a turn
/// can trail its `session/prompt` response by this much.
const TURN_GRACE: Duration = Duration::from_millis(100);
const STDERR_LINES: usize = 20;
/// Referenced files up to this size are embedded in the prompt (when the agent takes it).
const MAX_EMBED: u64 = 256 * 1024;
/// The agent tasks of one project.
pub struct AgentManager {
shared: Arc<Shared>,
tasks: RwLock<HashMap<String, Arc<Task>>>,
}
pub struct AgentPaths {
pub project_root: PathBuf,
/// One `<task id>.json` per task.
pub store_dir: PathBuf,
/// Parent of the tasks' worktrees.
pub worktrees_dir: PathBuf,
/// Settings each agent offered last time (shared by all projects), and its commands per project.
pub models_file: PathBuf,
/// User-defined agents, one `<id>.toml` each (`name`, `command`, `args`, `env`).
pub custom_agents_dir: Option<PathBuf>,
/// The user's home: its `.claude` holds their own skills and commands.
pub home_dir: Option<PathBuf>,
}
/// Starts the ACP server of a provider in a directory, speaking to `client`.
type Connector = Arc<
dyn Fn(&provider::ProviderSpec, &Path, Arc<dyn Handler>, Box<dyn Fn(String) + Send + Sync>) -> std::io::Result<Peer>
+ Send
+ Sync,
>;
struct Shared {
paths: AgentPaths,
runtime: tokio::runtime::Handle,
connector: Connector,
sink: RwLock<Option<AgentSink>>,
known: Mutex<HashMap<String, Known>>,
providers: RwLock<Vec<provider::ProviderSpec>>,
}
/// What an agent offered in its last session, for the New Task page.
#[derive(Clone, Default, Serialize, Deserialize)]
struct Known {
#[serde(default)]
config: Vec<AgentConfigOption>,
/// By project root: skills and custom commands differ per project.
#[serde(default, rename = "projectCommands")]
commands: HashMap<String, Vec<AgentCommand>>,
}
impl Shared {
fn spec(&self, id: &str) -> Option<provider::ProviderSpec> {
self.providers.read().iter().find(|p| p.id == id).cloned()
}
fn emit(&self, event: AgentEvent) {
let sink = self.sink.read().clone();
if let Some(sink) = sink {
sink(event);
}
}
fn known_config(&self, provider: &str) -> Vec<AgentConfigOption> {
self.known.lock().get(provider).map(|k| k.config.clone()).unwrap_or_default()
}
/// The commands the agent offered last time in this project.
fn known_commands(&self, provider: &str) -> Vec<AgentCommand> {
let cache = self.known.lock();
cache.get(provider).and_then(|k| k.commands.get(&self.root_key())).cloned().unwrap_or_default()
}
fn root_key(&self) -> String {
self.paths.project_root.to_string_lossy().into_owned()
}
fn remember(&self, provider: &str, config: Option<&[AgentConfigOption]>, commands: Option<&[AgentCommand]>) {
let snapshot = {
let mut cache = self.known.lock();
let known = cache.entry(provider.to_owned()).or_default();
let mut changed = false;
if let Some(c) = config.filter(|c| !c.is_empty() && known.config != *c) {
known.config = c.to_vec();
changed = true;
}
let root = self.root_key();
if let Some(c) =
commands.filter(|c| !c.is_empty() && known.commands.get(&root).map(Vec::as_slice) != Some(c))
{
known.commands.insert(root, c.to_vec());
changed = true;
}
if !changed {
return;
}
serde_json::to_vec_pretty(&*cache)
};
if let Ok(bytes) = snapshot
&& let Err(e) = write_file(&self.paths.models_file, &bytes)
{
tracing::warn!("saving agent settings: {e}");
}
}
}
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct TaskFile {
task: AgentTask,
session_id: Option<String>,
entries: Vec<AgentEntry>,
/// Settings the user chose, re-applied to every new or reloaded session.
#[serde(default)]
wanted: Vec<AgentConfigChoice>,
}
struct TaskState {
task: AgentTask,
session_id: Option<String>,
timeline: Timeline,
wanted: Vec<AgentConfigChoice>,
/// A turn is running or about to (prompts arriving now are queued).
busy: bool,
}
/// A running agent and its session.
#[derive(Clone)]
struct Live {
peer: Peer,
session: String,
/// `promptCapabilities.image`.
images: bool,
/// `promptCapabilities.embeddedContext`: referenced files are sent with their text.
embedded: bool,
}
struct Task {
id: String,
shared: Arc<Shared>,
state: Mutex<TaskState>,
live: Mutex<Option<Live>>,
/// Serializes starting the agent.
connecting: tokio::sync::Mutex<()>,
/// Unanswered permission requests: where to send the chosen option id.
permissions: Mutex<HashMap<u32, std::sync::mpsc::Sender<Option<String>>>>,
next_permission: AtomicU32,
/// Set while `session/load` replays history we already have.
replaying: AtomicBool,
/// The current turn is being cancelled to send the first queued message right away.
send_now: AtomicBool,
dirty: Mutex<BTreeSet<usize>>,
flush_scheduled: AtomicBool,
stderr: Mutex<VecDeque<String>>,
/// Set by `delete`; held while saving.
deleted: Mutex<bool>,
}
impl AgentManager {
/// Loads the project's saved tasks. Spawned work runs on `runtime`.
pub fn new(paths: AgentPaths, runtime: tokio::runtime::Handle) -> Self {
let connector: Connector = Arc::new(|spec, cwd, client, on_stderr| {
Peer::spawn_framed(&spec.command, &spec.args, cwd, &spec.env, client, on_stderr, Framing::Lines)
});
Self::with_connector(paths, runtime, connector)
}
fn with_connector(paths: AgentPaths, runtime: tokio::runtime::Handle, connector: Connector) -> Self {
let known =
std::fs::read(&paths.models_file).ok().and_then(|b| serde_json::from_slice(&b).ok()).unwrap_or_default();
let providers = RwLock::new(provider::load(paths.custom_agents_dir.as_deref()));
let shared = Arc::new(Shared {
paths,
runtime,
connector,
sink: RwLock::new(None),
known: Mutex::new(known),
providers,
});
let mut tasks = HashMap::new();
for file in std::fs::read_dir(&shared.paths.store_dir).into_iter().flatten().flatten() {
let path = file.path();
if path.extension().is_none_or(|e| e != "json") {
continue;
}
match std::fs::read(&path)
.map_err(|e| e.to_string())
.and_then(|b| serde_json::from_slice::<Value>(&b).map_err(|e| e.to_string()))
.and_then(|v| serde_json::from_value::<TaskFile>(upgrade(v)).map_err(|e| e.to_string()))
{
Ok(mut f) => {
// The agent process did not survive the restart: whatever was in flight is over.
if matches!(f.task.status, AgentStatus::Running | AgentStatus::NeedsInput | AgentStatus::Starting) {
f.task.status = AgentStatus::Ready;
let mut timeline = Timeline::from_entries(std::mem::take(&mut f.entries));
timeline.end_turn(None);
f.entries = timeline.entries;
}
let task = Task::restore(f, shared.clone());
tasks.insert(task.id.clone(), task);
}
Err(e) => tracing::warn!("skipping agent task {}: {e}", path.display()),
}
}
Self { shared, tasks: RwLock::new(tasks) }
}
/// Routes events to `sink` (replacing the previous subscriber); returns every task.
pub fn subscribe(&self, sink: AgentSink) -> Vec<AgentTask> {
*self.shared.sink.write() = Some(sink);
let mut tasks: Vec<AgentTask> = self.tasks.read().values().map(|t| t.state.lock().task.clone()).collect();
tasks.sort_by(|a, b| b.updated_ms.total_cmp(&a.updated_ms));
tasks
}
/// The agents (custom ones re-read), whether each is installed, and what they offered last time.
pub fn providers(&self) -> Vec<AgentProvider> {
let specs = provider::load(self.shared.paths.custom_agents_dir.as_deref());
let list = specs
.iter()
.map(|p| AgentProvider {
id: p.id.clone(),
name: p.name.clone(),
command: std::iter::once(p.command.as_str())
.chain(p.args.iter().map(String::as_str))
.collect::<Vec<_>>()
.join(" "),
installed: provider::installed(&p.command),
config: self.shared.known_config(&p.id),
commands: self.shared.known_commands(&p.id),
})
.collect();
*self.shared.providers.write() = specs;
list
}
/// The slash commands to complete for `provider`, in `task`'s directory or (without one)
/// in the project: what the agent offered last time there, plus, for Claude Code, the
/// skills and custom commands on disk it has not listed.
pub async fn commands(&self, provider: &str, task: Option<&str>) -> IdeResult<Vec<AgentCommand>> {
let (provider, offered, cwd) = match task {
Some(id) => {
let task = self.task(id)?;
let s = task.state.lock();
(s.task.provider.clone(), s.task.commands.clone(), s.task.cwd.0.clone())
}
None => (provider.to_owned(), self.shared.known_commands(provider), self.shared.paths.project_root.clone()),
};
if provider != "claude" {
return Ok(offered);
}
let home = self.shared.paths.home_dir.clone();
let found = blocking(move || Ok(skills::discover(&cwd, home.as_deref()))).await?;
Ok(skills::merge(offered, found))
}
fn task(&self, id: &str) -> IdeResult<Arc<Task>> {
self.tasks.read().get(id).cloned().ok_or_else(|| IdeError::not_found(format!("no agent task {id}")))
}
pub fn entries(&self, id: &str) -> IdeResult<Vec<AgentEntry>> {
Ok(self.task(id)?.state.lock().timeline.entries.clone())
}
/// Creates the task (and its worktree) and sends `req.prompt` as its first message.
pub async fn create(&self, req: NewAgentTask) -> IdeResult<AgentTask> {
let spec = self
.shared
.spec(&req.provider)
.ok_or_else(|| IdeError::invalid(format!("unknown agent {}", req.provider)))?;
let prompt = clean(req.prompt).ok_or_else(|| IdeError::invalid("describe the task first"))?;
let title = title_of(&prompt);
let root = self.shared.paths.project_root.clone();
let (cwd, branch, base, base_commit) = match req.isolation {
Isolation::Worktree => {
let dir = self.shared.paths.worktrees_dir.clone();
let (t, b) = (title.clone(), req.base.clone());
let c = blocking(move || worktree::create(&root, &dir, &t, b.as_deref())).await?;
(c.path, Some(c.branch), Some(c.base), Some(c.base_commit))
}
Isolation::Local => {
let r = root.clone();
let head = blocking(move || worktree::head_of(&r)).await.ok();
let (base, commit) = head.unzip();
(root, None, base, commit)
}
};
// Until the session reports its own, show what the agent offered last time with the choices applied.
let mut config = self.shared.known_config(&spec.id);
for choice in &req.config {
if let Some(o) = config.iter_mut().find(|o| o.id == choice.id) {
o.current = Some(choice.value.clone());
}
}
let now = now_ms();
let task = AgentTask {
id: uuid::Uuid::new_v4().simple().to_string(),
title,
provider: spec.id.clone(),
model: model_of(&config),
status: AgentStatus::Starting,
isolation: req.isolation,
cwd: IpcPath(cwd),
branch,
base,
base_commit,
error: None,
created_ms: now,
updated_ms: now,
config,
commands: self.shared.known_commands(&spec.id),
usage: None,
queue: Vec::new(),
};
let file = TaskFile { task: task.clone(), session_id: None, entries: Vec::new(), wanted: req.config };
let handle = Task::restore(file, self.shared.clone());
handle.state.lock().busy = true;
self.tasks.write().insert(task.id.clone(), handle.clone());
handle.emit_task();
handle.persist();
self.shared.runtime.spawn(async move { handle.run(prompt).await });
Ok(task)
}
/// Sends a message. While the agent works it is queued, or with `now` the current turn
/// is cancelled and the message goes next.
pub async fn prompt(&self, id: &str, prompt: AgentPrompt, now: bool) -> IdeResult<()> {
let task = self.task(id)?;
let prompt = clean(prompt).ok_or_else(|| IdeError::invalid("empty message"))?;
let start = {
let mut s = task.state.lock();
if s.task.status == AgentStatus::Archived {
return Err(IdeError::invalid("the task is archived"));
}
if s.busy {
let item = AgentQueued { id: uuid::Uuid::new_v4().simple().to_string(), prompt };
if now {
s.task.queue.insert(0, item);
} else {
s.task.queue.push(item);
}
None
} else {
s.busy = true;
Some(prompt)
}
};
match start {
Some(prompt) => {
self.shared.runtime.spawn(async move { task.run(prompt).await });
}
None => {
task.persist();
task.emit_task();
if now {
task.send_now.store(true, Ordering::Release);
task.cancel();
}
}
}
Ok(())
}
/// Drops a queued message.
pub fn unqueue(&self, id: &str, queued: &str) -> IdeResult<AgentQueued> {
let task = self.task(id)?;
let item = {
let mut s = task.state.lock();
let i =
s.task.queue.iter().position(|q| q.id == queued).ok_or_else(|| IdeError::not_found("already sent"))?;
s.task.queue.remove(i)
};
task.persist();
task.emit_task();
Ok(item)
}
/// Sends queued message `queued` next, cancelling the current turn if there is one.
pub fn send_queued_now(&self, id: &str, queued: &str) -> IdeResult<()> {
let task = self.task(id)?;
let idle = {
let mut s = task.state.lock();
let i =
s.task.queue.iter().position(|q| q.id == queued).ok_or_else(|| IdeError::not_found("already sent"))?;
let item = s.task.queue.remove(i);
if s.busy {
s.task.queue.insert(0, item);
None
} else {
s.busy = true;
Some(item)
}
};
task.persist();
task.emit_task();
match idle {
Some(item) => {
self.shared.runtime.spawn(async move { task.run(item.prompt).await });
}
None => {
task.send_now.store(true, Ordering::Release);
task.cancel();
}
}
Ok(())
}
/// Changes a session setting (model, mode…): right away when the agent runs, else when it next starts.
pub async fn set_config(&self, id: &str, option: &str, value: String) -> IdeResult<()> {
let task = self.task(id)?;
let target = {
let s = task.state.lock();
let o =
s.task.config.iter().find(|o| o.id == option).ok_or_else(|| IdeError::not_found("no such setting"))?;
if !o.values.iter().any(|v| v.value == value) {
return Err(IdeError::invalid(format!("{} has no value {value}", o.name)));
}
o.clone()
};
let live = task.live.lock().clone().filter(|l| !l.peer.is_closed());
if let Some(live) = live {
let (method, mut params) = config::set_request(&target, &value);
params["sessionId"] = json!(live.session);
let reply: Value =
live.peer.request(method, params, None).await.map_err(|e| IdeError::internal(task.describe(e)))?;
if let Some(options) = reply.get("configOptions") {
config::merge_update(&mut task.state.lock().task.config, options);
}
}
{
let mut s = task.state.lock();
s.wanted.retain(|c| c.id != option);
s.wanted.push(AgentConfigChoice { id: option.to_owned(), value: value.clone() });
if let Some(o) = s.task.config.iter_mut().find(|o| o.id == option) {
o.current = Some(value);
}
s.task.model = model_of(&s.task.config).or(s.task.model.take());
}
task.persist();
task.emit_task();
Ok(())
}
/// Stops the current turn (`session/cancel`); pending permission requests are cancelled.
/// Queued messages stay queued.
pub fn cancel(&self, id: &str) -> IdeResult<()> {
self.task(id)?.cancel();
Ok(())
}
/// Answers permission request `permission` with option `option` (`None` cancels it).
pub fn respond(&self, id: &str, permission: u32, option: Option<String>) -> IdeResult<()> {
let task = self.task(id)?;
let pending = task.permissions.lock().remove(&permission);
match pending {
Some(reply) => {
let _ = reply.send(option);
Ok(())
}
None => Err(IdeError::not_found("the request was already answered")),
}
}
/// Stops the agent and moves the task to Archived; its worktree stays.
pub fn archive(&self, id: &str) -> IdeResult<()> {
let task = self.task(id)?;
task.kill();
task.set_status(AgentStatus::Archived, None);
Ok(())
}
/// Forgets the task; with `remove_worktree` also deletes its worktree (the branch stays).
pub async fn delete(&self, id: &str, remove_worktree: bool) -> IdeResult<()> {
let task = self.task(id)?;
task.kill();
let (isolation, cwd) = {
let s = task.state.lock();
(s.task.isolation, s.task.cwd.0.clone())
};
if remove_worktree && isolation == Isolation::Worktree && cwd.exists() {
let root = self.shared.paths.project_root.clone();
blocking(move || worktree::remove(&root, &cwd)).await?;
}
self.tasks.write().remove(id);
*task.deleted.lock() = true;
let _ = std::fs::remove_file(self.shared.paths.store_dir.join(format!("{id}.json")));
self.shared.emit(AgentEvent::Removed { id: id.to_owned() });
Ok(())
}
/// Files the task changed since it started.
pub async fn changes(&self, id: &str) -> IdeResult<Vec<AgentChange>> {
let (cwd, base) = self.task(id)?.cwd_and_base();
blocking(move || worktree::changes(&cwd, &base)).await
}
/// Content of `path` (in the task's directory) when the task started.
pub async fn base_text(&self, id: &str, path: PathBuf) -> IdeResult<Option<String>> {
let (cwd, base) = self.task(id)?.cwd_and_base();
blocking(move || worktree::file_at(&cwd, &path, &base)).await
}
/// Stops every agent (project closed).
pub fn shutdown(&self) {
for task in self.tasks.read().values() {
task.kill();
}
*self.shared.sink.write() = None;
}
}
impl Task {
fn restore(file: TaskFile, shared: Arc<Shared>) -> Arc<Self> {
let task = file.task;
let timeline = Timeline::from_entries(file.entries);
Arc::new(Self {
id: task.id.clone(),
shared,
state: Mutex::new(TaskState {
task,
session_id: file.session_id,
timeline,
wanted: file.wanted,
busy: false,
}),
live: Mutex::new(None),
connecting: tokio::sync::Mutex::new(()),
permissions: Mutex::new(HashMap::new()),
next_permission: AtomicU32::new(1),
replaying: AtomicBool::new(false),
send_now: AtomicBool::new(false),
dirty: Mutex::new(BTreeSet::new()),
flush_scheduled: AtomicBool::new(false),
stderr: Mutex::new(VecDeque::new()),
deleted: Mutex::new(false),
})
}
fn cwd_and_base(&self) -> (PathBuf, String) {
let s = self.state.lock();
// In the project's own checkout the user commits too: compare against the current HEAD,
// as the Commit tool window does, so commits made since the task started drop out.
let base = match s.task.isolation {
Isolation::Local => None,
Isolation::Worktree => s.task.base_commit.clone(),
};
(s.task.cwd.0.clone(), base.unwrap_or_else(|| "HEAD".into()))
}
fn emit_task(&self) {
let task = self.state.lock().task.clone();
self.shared.emit(AgentEvent::Task { task });
}
fn set_status(&self, status: AgentStatus, error: Option<String>) {
{
let mut s = self.state.lock();
s.task.status = status;
s.task.error = error;
s.task.updated_ms = now_ms();
}
// Saved first: whoever reacts to the status change finds it on disk.
self.persist();
self.emit_task();
}
fn persist(&self) {
// Held from snapshot to rename, so saves land in order and none follows a delete.
let deleted = self.deleted.lock();
if *deleted {
return;
}
let bytes = {
let s = self.state.lock();
serde_json::to_vec(&TaskFile {
task: s.task.clone(),
session_id: s.session_id.clone(),
entries: s.timeline.entries.clone(),
wanted: s.wanted.clone(),
})
};
let path = self.shared.paths.store_dir.join(format!("{}.json", self.id));
if let Err(e) = bytes.map_err(|e| e.to_string()).and_then(|b| write_file(&path, &b)) {
tracing::warn!("saving agent task {}: {e}", self.id);
}
}
fn mark_dirty(self: &Arc<Self>, indices: impl IntoIterator<Item = usize>) {
self.dirty.lock().extend(indices);
if !self.flush_scheduled.swap(true, Ordering::AcqRel) {
let task = self.clone();
self.shared.runtime.spawn(async move {
tokio::time::sleep(FLUSH_DELAY).await;
task.flush_scheduled.store(false, Ordering::Release);
task.flush();
});
}
}
fn flush(&self) {
let indices = std::mem::take(&mut *self.dirty.lock());
let events: Vec<AgentEvent> = {
let s = self.state.lock();
indices
.into_iter()
.filter_map(|i| {
let entry = s.timeline.entries.get(i)?.clone();
Some(AgentEvent::Entry { task: self.id.clone(), index: i as u32, entry })
})
.collect()
};
for e in events {
self.shared.emit(e);
}
}
fn push_entry(self: &Arc<Self>, entry: AgentEntry) -> usize {
let i = self.state.lock().timeline.push(entry);
self.mark_dirty([i]);
i
}
/// Runs `prompt`, then the queued messages, until the queue is empty, a turn fails, or
/// the user stops one. The caller set `busy`.
async fn run(self: Arc<Self>, mut prompt: AgentPrompt) {
loop {
let result = self.turn(&prompt).await;
tokio::time::sleep(TURN_GRACE).await;
let changed = self.state.lock().timeline.end_turn(Some(now_ms()));
self.mark_dirty(changed);
let send_now = self.send_now.swap(false, Ordering::AcqRel);
let next = {
let mut s = self.state.lock();
let go_on = s.task.status != AgentStatus::Archived
&& match &result {
Ok(stop) => stop != "cancelled" || send_now,
Err(_) => false,
};
let next = (go_on && !s.task.queue.is_empty()).then(|| s.task.queue.remove(0));
if next.is_none() {
s.busy = false;
}
next
};
if self.state.lock().task.status == AgentStatus::Archived {
self.persist();
return;
}
if let Ok(stop) = &result
&& stop == "refusal"
{
self.push_entry(AgentEntry::Error { message: "The agent declined to continue.".into() });
}
match (next, result) {
(Some(queued), _) => prompt = queued.prompt,
(None, Ok(_)) => return self.set_status(AgentStatus::Ready, None),
(None, Err(message)) => {
self.push_entry(AgentEntry::Error { message: message.clone() });
return self.set_status(AgentStatus::Failed, Some(message));
}
}
}
}
/// Sends one prompt, starting the agent first if needed; returns the stop reason.
async fn turn(self: &Arc<Self>, prompt: &AgentPrompt) -> Result<String, String> {
let attachments = prompt
.files
.iter()
.map(|f| file_name(&f.0))
.chain(prompt.images.iter().map(|i| i.name.clone().unwrap_or_else(|| "image".into())))
.collect();
self.push_entry(AgentEntry::User {
text: prompt.text.clone(),
attachments,
sent_ms: Some(now_ms()),
done_ms: None,
});
// Launching the agent can take a while (`npx` fetches its adapter on first use).
let starting = self.live.lock().as_ref().is_none_or(|l| l.peer.is_closed());
self.set_status(if starting { AgentStatus::Starting } else { AgentStatus::Running }, None);
let live = self.ensure_session().await?;
self.set_status(AgentStatus::Running, None);
if !prompt.images.is_empty() && !live.images {
return Err("This agent can't read images.".into());
}
let contents = if live.embedded && !prompt.files.is_empty() {
let files: Vec<PathBuf> = prompt.files.iter().map(|f| f.0.clone()).collect();
tokio::task::spawn_blocking(move || files.iter().map(|f| embeddable(f)).collect()).await.unwrap_or_default()
} else {
Vec::new()
};
let result: Result<Value, RpcError> = live
.peer
.request("session/prompt", json!({ "sessionId": live.session, "prompt": blocks(prompt, &contents) }), None)
.await;
match result {
Ok(v) => Ok(v.get("stopReason").and_then(Value::as_str).unwrap_or("end_turn").to_owned()),
// Signed out, or the account was refused (Claude Code: "API Error: 403").
Err(RpcError::Server { message, .. }) if message.to_lowercase().contains("auth") => {
let provider = self.state.lock().task.provider.clone();
Err(match self.shared.spec(&provider) {
Some(spec) => format!("{} needs you to sign in. {}", spec.name, spec.login_hint),
None => message,
})
}
Err(e) => Err(self.describe(e)),
}
}
async fn ensure_session(self: &Arc<Self>) -> Result<Live, String> {
let _guard = self.connecting.lock().await;
if let Some(live) = self.live.lock().clone()
&& !live.peer.is_closed()
{
return Ok(live);
}
let (provider_id, cwd, saved) = {
let s = self.state.lock();
(s.task.provider.clone(), s.task.cwd.0.clone(), s.session_id.clone())
};
let spec = self.shared.spec(&provider_id).ok_or_else(|| format!("unknown agent {provider_id}"))?;
self.stderr.lock().clear();
let weak = Arc::downgrade(self);
let on_stderr: Box<dyn Fn(String) + Send + Sync> = Box::new(move |line| {
if let Some(t) = weak.upgrade() {
let mut buf = t.stderr.lock();
if buf.len() == STDERR_LINES {
buf.pop_front();
}
buf.push_back(line);
}
});
let client = Arc::new(Client { task: Arc::downgrade(self) });
let peer = (self.shared.connector)(&spec, &cwd, client, on_stderr)
.map_err(|e| format!("Couldn't start {} (`{}`): {e}", spec.name, spec.command))?;
let init: Value = peer
.request(
"initialize",
json!({
"protocolVersion": 1,
"clientCapabilities": { "fs": { "readTextFile": false, "writeTextFile": false }, "terminal": false },
"clientInfo": { "name": "OpenIDEA", "version": env!("CARGO_PKG_VERSION") },
}),
None,
)
.await
.map_err(|e| self.describe(e))?;
let can_load = init.pointer("/agentCapabilities/loadSession").and_then(Value::as_bool) == Some(true);
let capable = |c: &str| {
init.pointer(&format!("/agentCapabilities/promptCapabilities/{c}")).and_then(Value::as_bool) == Some(true)
};
let (images, embedded) = (capable("image"), capable("embeddedContext"));
let cwd_s = cwd.to_string_lossy().into_owned();
let mut session = None;
if let (Some(id), true) = (saved, can_load) {
self.replaying.store(true, Ordering::Release);
let loaded: Result<Value, RpcError> =
peer.request("session/load", json!({ "sessionId": id, "cwd": cwd_s, "mcpServers": [] }), None).await;
tokio::time::sleep(REPLAY_GRACE).await;
self.replaying.store(false, Ordering::Release);
match loaded {
Ok(v) => session = Some((id, v)),
Err(e) => tracing::info!("agent task {}: session/load failed, starting a new session: {e}", self.id),
}
}
let (session_id, result) = match session {
Some(s) => s,
None => {
let v: Value = peer
.request("session/new", json!({ "cwd": cwd_s, "mcpServers": [] }), None)
.await
.map_err(|e| match e {
RpcError::Server { ref message, .. } if message.to_lowercase().contains("auth") => {
format!("{} needs you to sign in. {}", spec.name, spec.login_hint)
}
e => self.describe(e),
})?;
let id = v.get("sessionId").and_then(Value::as_str).ok_or("the agent returned no session")?.to_owned();
(id, v)
}
};
// The session's settings, then the user's choices on top.
// A session that reports none keeps the last known ones, but what is set is unknown.
let reported = config::options_of(&result);
let known = !reported.is_empty();
let (mut options, wanted) = {
let s = self.state.lock();
(if known { reported } else { s.task.config.clone() }, s.wanted.clone())
};
self.shared.remember(&provider_id, Some(&options), None);
for choice in &wanted {
let Some(o) = options.iter().find(|o| o.id == choice.id).cloned() else { continue };
let already = known && o.current.as_deref() == Some(choice.value.as_str());
if already || !o.values.iter().any(|v| v.value == choice.value) {
continue;
}
let (method, mut params) = config::set_request(&o, &choice.value);
params["sessionId"] = json!(session_id);
match peer.request::<_, Value>(method, params, None).await {
Ok(reply) => {
if let Some(fresh) = reply.get("configOptions") {
config::merge_update(&mut options, fresh);
}
if let Some(o) = options.iter_mut().find(|o| o.id == choice.id) {
o.current = Some(choice.value.clone());
}
}
Err(e) => tracing::warn!("agent task {}: setting {} to {} failed: {e}", self.id, o.id, choice.value),
}
}
{
let mut s = self.state.lock();
s.task.model = model_of(&options).or(s.task.model.take());
s.task.config = options;
s.session_id = Some(session_id.clone());
}
self.persist();
self.emit_task();
let live = Live { peer, session: session_id, images, embedded };
*self.live.lock() = Some(live.clone());
Ok(live)
}
/// A user-facing message for a failed request, with the agent's last stderr lines when it died.
fn describe(&self, e: RpcError) -> String {
match e {
RpcError::Closed => {
let tail: Vec<String> = self.stderr.lock().iter().cloned().collect();
if tail.is_empty() {
"The agent process exited.".into()
} else {
format!("The agent process exited:\n{}", tail.join("\n"))
}
}
RpcError::Server { message, .. } => message,
e => e.to_string(),
}
}
fn on_update(self: &Arc<Self>, update: &Value) {
let kind = update.get("sessionUpdate").and_then(Value::as_str).unwrap_or_default();
if kind == "session_info_update" {
// The agent's own title for the session replaces the first line of the prompt.
let title = update.get("title").and_then(Value::as_str).map(str::trim).filter(|t| !t.is_empty());
if let Some(title) = title {
self.state.lock().task.title = title.to_owned();
self.persist();
self.emit_task();
}
return;
}
// Session state (also while `session/load` replays history).
let provider = self.state.lock().task.provider.clone();
let task_changed = {
let mut s = self.state.lock();
match kind {
"config_option_update" => {
if let Some(options) = update.get("configOptions") {
config::merge_update(&mut s.task.config, options);
s.task.model = model_of(&s.task.config).or(s.task.model.take());
}
true
}
"current_mode_update" => {
if let Some(mode) = update.get("currentModeId").and_then(Value::as_str) {
config::set_mode(&mut s.task.config, mode);
}
true
}
"available_commands_update" => {
s.task.commands = config::commands_of(update);
true
}
"usage_update" => {
s.task.usage = config::usage_of(update, s.task.usage.take());
true
}
_ => false,
}
};
if task_changed {
let (config, commands) = {
let s = self.state.lock();
(s.task.config.clone(), s.task.commands.clone())
};
self.shared.remember(&provider, Some(&config), Some(&commands));
if kind != "usage_update" {
self.persist();
}
self.emit_task();
return;
}
if self.replaying.load(Ordering::Acquire) {
return;
}
let applied = self.state.lock().timeline.apply(update);
self.mark_dirty(applied.changed);
}
/// Blocks the calling (blocking-pool) thread until the user answers.
fn ask_permission(self: &Arc<Self>, params: &Value) -> Value {
let id = self.next_permission.fetch_add(1, Ordering::Relaxed);
let title = params
.pointer("/toolCall/title")
.and_then(Value::as_str)
.filter(|t| !t.is_empty())
.unwrap_or("The agent asks for permission")
.to_owned();
let options: Vec<AgentPermissionOption> = params
.get("options")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|o| {
Some(AgentPermissionOption {
id: o.get("optionId")?.as_str()?.to_owned(),
name: o.get("name").and_then(Value::as_str).unwrap_or_default().to_owned(),
kind: o.get("kind").and_then(Value::as_str).unwrap_or_default().to_owned(),
})
})
.collect()
})
.unwrap_or_default();
let (tx, rx) = std::sync::mpsc::channel();
let entry = self.push_entry(AgentEntry::Permission { id, title, options: options.clone(), chosen: None });
self.permissions.lock().insert(id, tx);
self.set_status(AgentStatus::NeedsInput, None);
let answer = rx.recv().ok().flatten();
let name = answer
.as_ref()
.and_then(|a| options.iter().find(|o| &o.id == a))
.map_or_else(|| "Cancelled".to_owned(), |o| o.name.clone());
{
let mut s = self.state.lock();
if let Some(AgentEntry::Permission { chosen, .. }) = s.timeline.entries.get_mut(entry) {
*chosen = Some(name);
}
}
self.mark_dirty([entry]);
let still_asking = !self.permissions.lock().is_empty();
if !still_asking && self.state.lock().task.status == AgentStatus::NeedsInput {
self.set_status(AgentStatus::Running, None);
}
match answer {
Some(option) => json!({ "outcome": { "outcome": "selected", "optionId": option } }),
None => json!({ "outcome": { "outcome": "cancelled" } }),
}
}
fn cancel_permissions(&self) {
for (_, reply) in self.permissions.lock().drain() {
let _ = reply.send(None);
}
}
fn cancel(&self) {
self.cancel_permissions();
if let Some(live) = self.live.lock().clone() {
let _ = live.peer.notify("session/cancel", json!({ "sessionId": live.session }));
}
}
fn kill(&self) {
self.cancel_permissions();
if let Some(live) = self.live.lock().take() {
live.peer.kill();
}
}
}
/// Answers the agent's requests on behalf of one task.
struct Client {
task: Weak<Task>,
}
impl Handler for Client {
fn on_notification(&self, method: &str, params: Value) {
if method == "session/update"
&& let (Some(task), Some(update)) = (self.task.upgrade(), params.get("update"))
{
task.on_update(update);
}
}
fn on_request(&self, method: &str, params: Value) -> Result<Value, (i64, String)> {
let task = self.task.upgrade().ok_or((codes::INTERNAL_ERROR, "task closed".to_owned()))?;
match method {
"session/request_permission" => Ok(task.ask_permission(&params)),
_ => Err((codes::METHOD_NOT_FOUND, format!("unsupported method {method}"))),
}
}
}
/// The prompt trimmed, or `None` when there is nothing to send.
fn clean(mut prompt: AgentPrompt) -> Option<AgentPrompt> {
prompt.text = prompt.text.trim().to_owned();
(!prompt.text.is_empty() || !prompt.images.is_empty() || !prompt.files.is_empty()).then_some(prompt)
}
/// ACP content blocks: the text, each file (embedded when its text is in `contents`, else
/// a `resource_link`), then the images.
fn blocks(prompt: &AgentPrompt, contents: &[Option<String>]) -> Vec<Value> {
let mut out = Vec::new();
if !prompt.text.is_empty() {
out.push(json!({ "type": "text", "text": prompt.text }));
}
for (i, f) in prompt.files.iter().enumerate() {
let uri = file_uri(&f.0);
out.push(match contents.get(i).cloned().flatten() {
Some(text) => json!({ "type": "resource", "resource": { "uri": uri, "text": text } }),
None => json!({ "type": "resource_link", "uri": uri, "name": file_name(&f.0) }),
});
}
for i in &prompt.images {
out.push(json!({ "type": "image", "mimeType": i.mime_type, "data": i.data }));
}
out
}
/// The text of a referenced file, when it is a small UTF-8 file.
fn embeddable(path: &Path) -> Option<String> {
let meta = std::fs::metadata(path).ok()?;
if !meta.is_file() || meta.len() > MAX_EMBED {
return None;
}
String::from_utf8(std::fs::read(path).ok()?).ok()
}
/// A task file from before a field existed, with that field's empty value.
fn upgrade(mut file: Value) -> Value {
if let Some(task) = file.get_mut("task").and_then(Value::as_object_mut) {
for (key, empty) in
[("config", json!([])), ("commands", json!([])), ("usage", Value::Null), ("queue", json!([]))]
{
task.entry(key).or_insert(empty);
}
}
for entry in file.get_mut("entries").and_then(Value::as_array_mut).into_iter().flatten() {
if entry.get("type").and_then(Value::as_str) == Some("user")
&& let Some(e) = entry.as_object_mut()
{
e.entry("attachments").or_insert(json!([]));
}
}
file
}
fn model_of(config: &[AgentConfigOption]) -> Option<String> {
config.iter().find(|o| o.category.as_deref() == Some("model")).and_then(|o| o.current.clone())
}
fn file_name(path: &Path) -> String {
path.file_name().map_or_else(|| path.to_string_lossy().into_owned(), |n| n.to_string_lossy().into_owned())
}
/// `file://` URI of an absolute path, percent-encoded.
fn file_uri(path: &Path) -> String {
let s = path.to_string_lossy().replace('\\', "/");
let mut out = String::from(if s.starts_with('/') { "file://" } else { "file:///" });
for b in s.bytes() {
if b.is_ascii_alphanumeric() || b"/-._~:".contains(&b) {
out.push(b as char);
} else {
out.push_str(&format!("%{b:02X}"));
}
}
out
}
fn title_of(prompt: &AgentPrompt) -> String {
let line = prompt.text.lines().map(str::trim).find(|l| !l.is_empty());
let line = match line {
Some(l) => l.to_owned(),
None => prompt.files.first().map(|f| file_name(&f.0)).unwrap_or_else(|| "Image".into()),
};
match line.char_indices().nth(60) {
Some((cut, _)) => format!("{}…", &line[..cut]),
None => line,
}
}
fn now_ms() -> f64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0.0, |d| d.as_millis() as f64)
}
fn write_file(path: &Path, bytes: &[u8]) -> Result<(), String> {
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir).map_err(|e| e.to_string())?;
}
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, bytes).and_then(|()| std::fs::rename(&tmp, path)).map_err(|e| e.to_string())
}
async fn blocking<T: Send + 'static>(f: impl FnOnce() -> IdeResult<T> + Send + 'static) -> IdeResult<T> {
tokio::task::spawn_blocking(f).await.map_err(|e| IdeError::internal(e.to_string()))?
}
#[cfg(test)]
mod tests;
Reference in new issue View Git Blame Copy Permalink