use super::{Pump, RpcError};
use serde_json::{Value, json};
pub(super) enum Change {
List,
New,
Open(String),
}
pub(super) struct Opening {
pub call: u64,
pub deadline: tokio::time::Instant,
id: Value,
change: Change,
replay: Vec<Value>,
initial: bool,
tentative: Option<String>,
failed: bool,
}
impl Opening {
pub const fn changes_session(&self) -> bool {
!self.failed && !matches!(self.change, Change::List)
}
}
impl Pump {
pub fn opening_deadline(&self) -> Option<tokio::time::Instant> {
if !self.pending.is_empty() || !self.answers.is_empty() {
return None;
}
self.opening
.as_ref()
.filter(|opening| !opening.failed)
.map(|opening| opening.deadline)
}
pub(super) fn session_deadline(&self) -> anyhow::Result<tokio::time::Instant> {
tokio::time::Instant::now()
.checked_add(std::time::Duration::from_millis(
self.seat.harness.session_timeout_ms,
))
.ok_or_else(|| anyhow::anyhow!("session_timeout_overflow"))
}
pub(super) fn reset_opening_deadline(&mut self) -> anyhow::Result<()> {
let deadline = self.session_deadline()?;
if let Some(opening) = self.opening.as_mut() {
opening.deadline = deadline;
}
Ok(())
}
pub(super) fn bootstrap(&mut self) -> anyhow::Result<()> {
let prior = self.seat.harness.session.clone();
let (method, params) = super::super::handshake::session_request(
&self.seat,
self.session.version,
prior.as_deref(),
&self.agent,
)?;
self.output.request(2, method, params)?;
self.opening = Some(Opening {
call: 2,
deadline: self.session_deadline()?,
id: Value::Null,
change: prior.map_or(Change::New, Change::Open),
replay: Vec::new(),
initial: true,
tentative: None,
failed: false,
});
Ok(())
}
pub(super) fn permission_session(&mut self, params: &Value) -> anyhow::Result<String> {
let id = params["sessionId"]
.as_str()
.filter(|id| !id.is_empty() && id.len() <= self.seat.harness.content_bytes)
.ok_or_else(|| anyhow::anyhow!("permission_session_invalid"))?;
if let Some(opening) = self.opening.as_mut() {
anyhow::ensure!(!opening.failed, "permission_session_failed");
match &opening.change {
Change::New if id != self.session.id => {
anyhow::ensure!(
opening.tentative.as_deref().is_none_or(|prior| prior == id),
"permission_session_differs"
);
opening.tentative = Some(id.to_owned());
return Ok(id.to_owned());
}
Change::Open(target) if target == id => return Ok(id.to_owned()),
_ => {}
}
}
anyhow::ensure!(id == self.session.id, "permission_session_differs");
Ok(id.to_owned())
}
pub(super) async fn session_control(
&mut self,
id: &Value,
change: &Value,
) -> anyhow::Result<()> {
if self.opening.is_some() {
return self.refuse(id, "session_busy").await;
}
if self.configuring.is_some() {
return self.refuse(id, "config_busy").await;
}
if !self.pending.is_empty() || !self.answers.is_empty() {
return self.refuse(id, "session_busy").await;
}
let Some(object) = change.as_object().filter(|object| object.len() == 1) else {
return self.refuse(id, "invalid_session").await;
};
let change = if object.get("list") == Some(&json!(true)) {
Change::List
} else if object.get("new") == Some(&json!(true)) {
Change::New
} else if let Some(target) = object
.get("open")
.and_then(Value::as_str)
.filter(|id| !id.is_empty() && id.len() <= self.seat.harness.content_bytes)
{
Change::Open(target.to_owned())
} else {
return self.refuse(id, "invalid_session").await;
};
let refused = if !matches!(change, Change::List) && self.turn.is_some() {
Some("session_busy")
} else if matches!(change, Change::Open(_)) && !self.session.resumable {
Some("session_resume_not_offered")
} else {
None
};
if let Some(code) = refused {
self.say(json!({"kind":"session_failed","request_id":id,"code":code}))
.await?;
return self.refuse(id, code).await;
}
let (method, params) = match &change {
Change::List => ("session/list", json!({"cwd":self.seat.cwd()?})),
Change::New => super::super::handshake::session_request(
&self.seat,
self.session.version,
None,
&self.agent,
)?,
Change::Open(target) => super::super::handshake::session_request(
&self.seat,
self.session.version,
Some(target),
&self.agent,
)?,
};
let call = self.call_id()?;
self.output.request(call, method, params)?;
if !matches!(change, Change::List) {
self.say(json!({"kind":"session_opening","request_id":id,"target":match &change {Change::Open(target)=>json!(target),_=>Value::Null}})).await?;
}
self.opening = Some(Opening {
call,
deadline: self.session_deadline()?,
id: id.clone(),
change,
replay: Vec::new(),
initial: false,
tentative: None,
failed: false,
});
self.ack(id).await
}
pub(super) async fn notification(&mut self, params: Value) -> anyhow::Result<()> {
if let Some(opening) = self.opening.as_mut()
&& match &opening.change {
Change::Open(target) => params["sessionId"] == *target,
Change::New => params["sessionId"] != self.session.id,
Change::List => false,
}
{
if let Change::Open(target) = &opening.change {
anyhow::ensure!(params["sessionId"] == *target, "replay_session_differs");
}
if opening.failed {
return Ok(());
}
if opening.replay.len() >= self.seat.harness.stream_queue {
return self.fail_session("session_replay_limit").await;
}
opening.replay.push(params);
return Ok(());
}
self.update(params, false).await
}
pub(super) async fn opened(&mut self, result: Result<Value, RpcError>) -> anyhow::Result<()> {
let opening = self
.opening
.take()
.ok_or_else(|| anyhow::anyhow!("session_request_absent"))?;
if opening.failed {
return Ok(());
}
if opening.initial
&& let Err(error) = &result
{
anyhow::bail!(
"ACP session creation refused: {} (code {}): {}",
if error.code == -32000 {
"authentication required"
} else {
"harness refusal"
},
error.code,
error.message
);
}
let answer=match result {Ok(answer)=>answer,Err(error)=>return self.say(json!({"kind":"session_failed","request_id":opening.id,"code":error.code,"message":error.message})).await};
if matches!(opening.change, Change::List) {
return self.say(json!({"kind":"sessions","request_id":opening.id,"current":self.session.id,"sessions":listed(&answer,self.seat.harness.session_list,self.seat.harness.title_chars)?,"next":answer["nextCursor"]})).await;
}
let resumed = matches!(opening.change, Change::Open(_));
let prior = match &opening.change {
Change::Open(id) => Some(id.as_str()),
_ => None,
};
let (id, config) =
super::super::handshake::session_answer(self.session.version, prior, answer)?;
anyhow::ensure!(
id.len() <= self.seat.harness.content_bytes,
"session_id_exceeds_content_bytes"
);
anyhow::ensure!(
opening
.tentative
.as_ref()
.is_none_or(|target| target == &id),
"session_new_id_differs_from_permission"
);
anyhow::ensure!(
opening
.replay
.iter()
.all(|params| params["sessionId"] == id),
"session_replay_id_differs"
);
self.session.id = id;
self.session_open = true;
self.commands.clear();
self.session.config = config;
self.say(json!({"kind":"session_opened","request_id":opening.id,"session":self.session.id,"resumed":resumed,"resume_capability":self.session.capability})).await?;
for params in opening.replay {
self.update(params, resumed).await?;
}
self.say(json!({"kind":"config_options","options":self.session.config}))
.await
}
}
fn listed(answer: &Value, bound: usize, title_chars: usize) -> anyhow::Result<Vec<Value>> {
let all = answer["sessions"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("session_list_invalid"))?;
let mut sessions = Vec::new();
for session in all {
let id = session["sessionId"]
.as_str()
.filter(|id| !id.is_empty())
.ok_or_else(|| anyhow::anyhow!("session_list_id_invalid"))?;
let time = session["updatedAt"]
.as_str()
.map(chrono::DateTime::parse_from_rfc3339)
.transpose()?;
let title: String = session["title"]
.as_str()
.unwrap_or_default()
.chars()
.take(title_chars)
.collect();
sessions.push((
time,
json!({"id":id,"title":title,"updated_at":session["updatedAt"]}),
));
}
sessions.sort_by_key(|left| std::cmp::Reverse(left.0));
Ok(sessions
.into_iter()
.take(bound)
.map(|(_, session)| session)
.collect())
}
#[path = "session_failure.rs"]
mod failure;