use super::wire::{Frame, Pipe};
use agent_client_protocol_schema::{ProtocolVersion, v1, v2};
use serde::Serialize;
use serde_json::{Value, json};
use std::path::Path;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Version {
One,
Two,
}
pub(super) fn negotiated(version: u16) -> anyhow::Result<Version> {
match version {
1 => Ok(Version::One),
2 => Ok(Version::Two),
_ => anyhow::bail!("ACP protocol version {version} refused: offered 2, supported 2 and 1"),
}
}
impl Version {
pub const fn number(self) -> u16 {
match self {
Self::One => 1,
Self::Two => 2,
}
}
}
pub(super) struct Session {
pub id: String,
pub version: Version,
pub resumable: bool,
pub updates: Vec<Value>,
pub config: Value,
pub cwd: std::path::PathBuf,
pub seat: std::path::PathBuf,
}
pub(super) fn encoded<T: Serialize>(value: T) -> anyhow::Result<Value> {
serde_json::to_value(value).map_err(|_| anyhow::anyhow!("ACP schema encoding failed"))
}
pub(super) async fn open(
pipe: &mut Pipe,
cwd: &Path,
seat: &Path,
prior: Option<&str>,
config: &super::config::Harness,
) -> anyhow::Result<Session> {
let info = v2::Implementation::new("liminal", env!("CARGO_PKG_VERSION"));
let mut initialize = encoded(v2::InitializeRequest::new(ProtocolVersion::V2, info))?;
initialize["clientInfo"] = initialize["info"].clone();
initialize["clientCapabilities"] =
json!({"fs":{"readTextFile":false,"writeTextFile":false}, "terminal":false});
let mut deferred = Vec::new();
let answer = call(
pipe,
1,
v2::AGENT_METHOD_NAMES.initialize,
initialize,
&mut deferred,
config.stream_queue,
)
.await?;
let (version, resumable) = capabilities(answer)?;
anyhow::ensure!(
prior.is_none() || resumable,
"session resume refused: agent did not advertise loadSession"
);
let (method, params) = session_request(version, cwd, seat, prior)?;
let replay_start = deferred.len();
let answer = call(pipe, 2, method, params, &mut deferred, config.stream_queue).await?;
let advertised = answer.get("configOptions").cloned().unwrap_or(Value::Null);
let id = match (version, prior) {
(Version::One, None) => serde_json::from_value::<v1::NewSessionResponse>(answer)
.map_err(|_| anyhow::anyhow!("session/new is not an ACP 1 response"))?
.session_id
.to_string(),
(Version::Two, None) => serde_json::from_value::<v2::NewSessionResponse>(answer)
.map_err(|_| anyhow::anyhow!("session/new is not an ACP 2 response"))?
.session_id
.to_string(),
(Version::One, Some(id)) => {
serde_json::from_value::<v1::LoadSessionResponse>(answer)
.map_err(|_| anyhow::anyhow!("session/load is not an ACP 1 response"))?;
id.to_owned()
}
(Version::Two, Some(id)) => {
serde_json::from_value::<v2::ResumeSessionResponse>(answer)
.map_err(|_| anyhow::anyhow!("session/resume is not an ACP 2 response"))?;
id.to_owned()
}
};
anyhow::ensure!(
!id.is_empty() && id.len() <= config.content_bytes,
"session id is empty or exceeds content_bytes"
);
let mut updates = Vec::new();
for (index, params) in deferred.into_iter().enumerate() {
if let Some(item) = super::stream::project(version, params, &id, config.content_bytes)? {
let item = if prior.is_some() && index >= replay_start {
super::stream::replayed(item, config.content_bytes)?
} else {
item
};
updates.push(item);
}
}
if advertised.is_array() {
updates.push(json!({"type":"config_options","options":advertised}));
}
Ok(Session {
id,
version,
resumable,
updates,
config: advertised,
cwd: cwd.to_path_buf(),
seat: seat.to_path_buf(),
})
}
fn capabilities(answer: Value) -> anyhow::Result<(Version, bool)> {
let number: ProtocolVersion = serde_json::from_value(answer["protocolVersion"].clone())
.map_err(|_| anyhow::anyhow!("initialize has no numeric protocolVersion"))?;
let version = negotiated(number.as_u16())?;
let resumable = match version {
Version::One => {
serde_json::from_value::<v1::InitializeResponse>(answer)
.map_err(|_| anyhow::anyhow!("initialize is not an ACP 1 response"))?
.agent_capabilities
.load_session
}
Version::Two => {
let response: v2::InitializeResponse = serde_json::from_value(answer)
.map_err(|_| anyhow::anyhow!("initialize is not an ACP 2 response"))?;
anyhow::ensure!(
response.capabilities.session.is_some(),
"ACP 2 agent did not advertise the session surface"
);
true
}
};
Ok((version, resumable))
}
pub(super) fn session_request(
version: Version,
cwd: &Path,
seat: &Path,
prior: Option<&str>,
) -> anyhow::Result<(&'static str, Value)> {
let executable = std::env::current_exe()?;
let seat = seat
.to_str()
.ok_or_else(|| anyhow::anyhow!("seat path is not Unicode"))?;
let args = vec![
"agent".to_owned(),
"tools".to_owned(),
"--config".to_owned(),
seat.to_owned(),
];
let (method, params) = match (version, prior) {
(Version::One, None) => (
v1::AGENT_METHOD_NAMES.session_new,
encoded(
v1::NewSessionRequest::new(cwd.to_path_buf()).mcp_servers(vec![
v1::McpServer::Stdio(v1::McpServerStdio::new("liminal", executable).args(args)),
]),
)?,
),
(Version::One, Some(id)) => (
v1::AGENT_METHOD_NAMES.session_load,
encoded(
v1::LoadSessionRequest::new(id.to_owned(), cwd.to_path_buf()).mcp_servers(vec![
v1::McpServer::Stdio(v1::McpServerStdio::new("liminal", executable).args(args)),
]),
)?,
),
(Version::Two, None) => (
v2::AGENT_METHOD_NAMES.session_new,
encoded(
v2::NewSessionRequest::new(cwd.to_path_buf()).mcp_servers(vec![
v2::McpServer::Stdio(v2::McpServerStdio::new("liminal", executable).args(args)),
]),
)?,
),
(Version::Two, Some(id)) => (
v2::AGENT_METHOD_NAMES.session_resume,
encoded(
v2::ResumeSessionRequest::new(id.to_owned(), cwd.to_path_buf()).mcp_servers(vec![
v2::McpServer::Stdio(v2::McpServerStdio::new("liminal", executable).args(args)),
]),
)?,
),
};
Ok((method, params))
}
async fn call(
pipe: &mut Pipe,
id: u64,
method: &str,
params: Value,
deferred: &mut Vec<Value>,
bound: usize,
) -> anyhow::Result<Value> {
pipe.output.request(id, method, params)?;
loop {
match pipe
.input
.recv()
.await
.ok_or_else(|| anyhow::anyhow!("ACP reader ended during handshake"))??
.ok_or_else(|| anyhow::anyhow!("ACP stdout closed during handshake"))?
{
Frame::Response {
id: received,
result,
} => {
anyhow::ensure!(
received == json!(id),
"ACP handshake response id was never requested"
);
return result.map_err(|code| {
if code == -32000 {
anyhow::anyhow!("ACP authentication required before {method} (code {code})")
} else {
anyhow::anyhow!("ACP {method} refused (code {code})")
}
});
}
Frame::Request { id, .. } => {
pipe.output
.refuse(id, -32601, "method is not advertised during handshake")?;
}
Frame::Notification { method, .. } if method.starts_with('_') => {}
Frame::Notification { method, params } => {
anyhow::ensure!(
method == v1::CLIENT_METHOD_NAMES.session_update,
"ACP notification is not advertised during handshake"
);
anyhow::ensure!(
deferred.len() < bound,
"handshake updates exceed stream_queue {bound}"
);
deferred.push(params);
}
}
}
}
pub(super) fn prompt(session: &Session, text: String) -> anyhow::Result<Value> {
match session.version {
Version::One => encoded(v1::PromptRequest::new(
session.id.clone(),
vec![v1::ContentBlock::Text(v1::TextContent::new(text))],
)),
Version::Two => encoded(v2::PromptRequest::new(
session.id.clone(),
vec![v2::ContentBlock::Text(v2::TextContent::new(text))],
)),
}
}