mirror of
https://github.com/codewhale-hq/Codewhale.git
synced 2026-11-11 01:51:34 -10:00
feat(engine,app-server): ACP options route, typed control refusals and agent-reported usage (CW-7879 EA-04)
GET /v1/agents/external/{id}/options?cwd= answers with the agent's own
session/new offer (models, modes, config options, capabilities) read on a
throwaway process that is stopped on every path, plus which Codewhale
actions it carries. A saved model choice is validated against the complete
offer at setup and written through the method it was offered on; a choice
the agent does not offer or no longer offers, and per-turn controls an
agent cannot honor, are stable typed refusals instead of strings.
Agent-reported usage is recorded as turn.agent_usage events: context
window, a per-turn token total, cost unknown, and unknown (not zero) when
the agent reports nothing. A started agent process is now owned by a guard
that stops it on error or panic and logs a failed stop.
Checks (Boat farm, CARGO_INCREMENTAL=0): engine --lib external_agent
fleet::acp 20 passed 0 failed; app-server --lib external_agent 3 passed
0 failed; clippy -D warnings engine+app-server --all-targets clean.
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
This commit is contained in:
CodeWhale BotandClaude Sonnet 5.5 committed
1 parent
3da52ab015
commit
10ca1deee0
8 files changed
+1807
-88
No files matched your search
@@ -117,6 +117,7 @@ mod avatars;
|
||||
mod computer_display;
|
||||
mod context;
|
||||
mod diagnostics;
|
||||
mod external_agent_options;
|
||||
mod external_agents;
|
||||
mod git;
|
||||
mod jobs;
|
||||
@@ -3462,6 +3463,10 @@ pub fn build_router(state: RuntimeApiState) -> Router {
|
||||
"/v1/agents/external",
|
||||
get(external_agents::list_external_agents),
|
||||
)
|
||||
.route(
|
||||
"/v1/agents/external/{id}/options",
|
||||
get(external_agent_options::get_external_agent_options),
|
||||
)
|
||||
.route("/v1/providers", get(list_providers))
|
||||
.route("/v1/providers/{id}/models", get(list_provider_models))
|
||||
.route(
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
//! `GET /v1/agents/external/{id}/options?cwd=` — what one external agent
|
||||
//! offers for one workspace.
|
||||
//!
|
||||
//! The answer is the agent's own `session/new` offer, read on a throwaway
|
||||
//! process the Engine stops before it returns. Nothing here comes from
|
||||
//! Codewhale's provider catalog.
|
||||
//!
|
||||
//! Known limitations: probing starts a real agent session in `cwd`, which an
|
||||
//! agent may keep in its own history. The route is gated like the rest of the
|
||||
//! Runtime API; it does not tie `cwd` to a thread owner.
|
||||
|
||||
use std::path::PathBuf;
|
||||
|
||||
use axum::Json;
|
||||
use axum::extract::{Path, Query, State};
|
||||
use serde::Deserialize;
|
||||
|
||||
use codewhale_engine::fleet::acp::AgentOptions;
|
||||
use codewhale_engine::fleet::agents::ExecutorUnavailable;
|
||||
|
||||
use super::{ApiError, RuntimeApiState, external_agent_refusal};
|
||||
|
||||
const MAX_AGENT_ID_BYTES: usize = 64;
|
||||
const MAX_CWD_BYTES: usize = 4096;
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(super) struct OptionsQuery {
|
||||
cwd: Option<PathBuf>,
|
||||
}
|
||||
|
||||
fn valid_agent_id(id: &str) -> bool {
|
||||
!id.is_empty()
|
||||
&& id.len() <= MAX_AGENT_ID_BYTES
|
||||
&& id
|
||||
.chars()
|
||||
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_'))
|
||||
}
|
||||
|
||||
pub(super) async fn get_external_agent_options(
|
||||
State(state): State<RuntimeApiState>,
|
||||
Path(id): Path<String>,
|
||||
Query(query): Query<OptionsQuery>,
|
||||
) -> Result<Json<AgentOptions>, ApiError> {
|
||||
if !valid_agent_id(&id) {
|
||||
return Err(ApiError::bad_request(
|
||||
"the agent id must name an agent from GET /v1/agents/external",
|
||||
));
|
||||
}
|
||||
let cwd = query
|
||||
.cwd
|
||||
.ok_or_else(|| ApiError::bad_request("cwd is required"))?;
|
||||
let text = cwd.to_string_lossy();
|
||||
if text.len() > MAX_CWD_BYTES || text.chars().any(char::is_control) || !cwd.is_absolute() {
|
||||
return Err(ApiError::bad_request(
|
||||
"cwd must be an absolute directory path",
|
||||
));
|
||||
}
|
||||
let probe = cwd.clone();
|
||||
let is_dir = tokio::task::spawn_blocking(move || probe.is_dir())
|
||||
.await
|
||||
.map_err(|_| ApiError::internal("cwd check failed"))?;
|
||||
if !is_dir {
|
||||
return Err(ApiError::not_found(
|
||||
"cwd is not a directory on this computer",
|
||||
));
|
||||
}
|
||||
match state
|
||||
.runtime_threads
|
||||
.external_agent_options(&id, &cwd)
|
||||
.await
|
||||
{
|
||||
Ok(options) => Ok(Json(options)),
|
||||
Err(error) => {
|
||||
let mut refusal = external_agent_refusal(&error);
|
||||
if let Some(refusal) = refusal.as_mut()
|
||||
&& refusal.code == Some(ExecutorUnavailable::UNKNOWN_AGENT)
|
||||
{
|
||||
refusal.status = axum::http::StatusCode::NOT_FOUND;
|
||||
}
|
||||
Err(refusal.unwrap_or_else(|| ApiError::internal(error.to_string())))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,8 @@ use uuid::Uuid;
|
||||
mod command_catalog;
|
||||
#[cfg(unix)]
|
||||
mod control_socket_selection;
|
||||
#[cfg(unix)]
|
||||
mod external_agent_options;
|
||||
mod headless_catalog;
|
||||
#[cfg(unix)]
|
||||
mod owner_bootstrap;
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
use super::*;
|
||||
use reqwest::StatusCode as HttpStatus;
|
||||
|
||||
fn options_fixture() -> PathBuf {
|
||||
Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../engine/tests/fixtures/acp-options-agent.mjs")
|
||||
.canonicalize()
|
||||
.expect("the options fixture agent")
|
||||
}
|
||||
|
||||
fn with_query(base: &str, pairs: &[(&str, &str)]) -> reqwest::Url {
|
||||
let mut url = reqwest::Url::parse(base).expect("a url");
|
||||
for (key, value) in pairs {
|
||||
url.query_pairs_mut().append_pair(key, value);
|
||||
}
|
||||
url
|
||||
}
|
||||
|
||||
fn received(workspace: &Path) -> Vec<Value> {
|
||||
fs::read_to_string(workspace.join("options-received.jsonl"))
|
||||
.unwrap_or_default()
|
||||
.lines()
|
||||
.filter_map(|line| serde_json::from_str(line).ok())
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn wait_for_usage_total(
|
||||
manager: &SharedRuntimeThreadManager,
|
||||
thread_id: &str,
|
||||
) -> Result<RuntimeEventRecord> {
|
||||
for _ in 0..600 {
|
||||
let events = manager.events_since_async(thread_id, None).await?;
|
||||
if let Some(event) = events.into_iter().find(|event| {
|
||||
event.event == "turn.agent_usage" && event.payload["kind"] == "turn_total"
|
||||
}) {
|
||||
return Ok(event);
|
||||
}
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
bail!("the agent's usage was never recorded")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_options_route_offers_the_agents_choices_and_a_chosen_one_reaches_it() -> Result<()> {
|
||||
let _env = lock_test_env();
|
||||
let _fixture = EnvVarGuard::set("CODEWHALE_ACP_FIXTURE_AGENT", options_fixture());
|
||||
let Some((addr, manager, handle)) = spawn_test_server().await? else {
|
||||
return Ok(());
|
||||
};
|
||||
let client = codewhale_release::tls::reqwest_client();
|
||||
let workspace = tempfile::tempdir()?;
|
||||
let cwd = workspace.path().canonicalize()?;
|
||||
let url = format!("http://{addr}/v1/agents/external/fixture/options");
|
||||
|
||||
let offered: Value = client
|
||||
.get(with_query(&url, &[("cwd", cwd.to_str().unwrap())]))
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json()
|
||||
.await?;
|
||||
assert_eq!(offered["agent"]["id"], "fixture");
|
||||
let controls = offered["controls"].as_array().unwrap();
|
||||
let model = controls.iter().find(|c| c["id"] == "model").unwrap();
|
||||
let values: Vec<&str> = model["choices"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|c| c["value"].as_str().unwrap())
|
||||
.collect();
|
||||
assert!(values.len() >= 2, "{values:?}");
|
||||
assert!(controls.iter().any(|c| c["id"] == "mode"));
|
||||
assert!(
|
||||
controls.iter().any(|c| c["id"] == "config:verbose"
|
||||
&& c["unavailable"]["code"] == "control_type_unsupported")
|
||||
);
|
||||
assert!(
|
||||
offered["actions"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.any(|a| a["action"] == "steering" && a["supported"] == false)
|
||||
);
|
||||
let leaked = fs::read_to_string(cwd.join("options-env.json"))?;
|
||||
assert!(!leaked.contains("CODEWHALE_ACP_FIXTURE_AGENT"));
|
||||
|
||||
for (query, status, code) in [
|
||||
(vec![], HttpStatus::BAD_REQUEST, None),
|
||||
(vec![("cwd", "relative/dir")], HttpStatus::BAD_REQUEST, None),
|
||||
(
|
||||
vec![("cwd", "/definitely/not/here")],
|
||||
HttpStatus::NOT_FOUND,
|
||||
None,
|
||||
),
|
||||
] {
|
||||
let response = client.get(with_query(&url, &query)).send().await?;
|
||||
assert_eq!(response.status(), status, "{query:?}");
|
||||
let body: Value = response.json().await?;
|
||||
assert_eq!(body["error"]["code"].as_str(), code, "{body}");
|
||||
}
|
||||
let unknown = client
|
||||
.get(with_query(
|
||||
&format!("http://{addr}/v1/agents/external/nope/options"),
|
||||
&[("cwd", cwd.to_str().unwrap())],
|
||||
))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(unknown.status(), HttpStatus::NOT_FOUND);
|
||||
let body: Value = unknown.json().await?;
|
||||
assert_eq!(body["error"]["code"], "unknown_agent");
|
||||
let bad_id = client
|
||||
.get(with_query(
|
||||
&format!("http://{addr}/v1/agents/external/..%2Fsh/options"),
|
||||
&[("cwd", cwd.to_str().unwrap())],
|
||||
))
|
||||
.send()
|
||||
.await?;
|
||||
assert!(bad_id.status().is_client_error());
|
||||
|
||||
let chosen = values[1];
|
||||
let created: Value = client
|
||||
.post(format!("http://{addr}/v1/threads"))
|
||||
.json(&json!({
|
||||
"workspace": cwd,
|
||||
"permission_posture": "ask",
|
||||
"executor": {"kind": "acp", "agent_id": "fixture", "config": {"model": chosen}}
|
||||
}))
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json()
|
||||
.await?;
|
||||
let thread_id = created["id"].as_str().unwrap().to_string();
|
||||
|
||||
let refused = client
|
||||
.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
|
||||
.json(&json!({"prompt": "x", "model": "other"}))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(refused.status(), HttpStatus::FAILED_DEPENDENCY);
|
||||
let body: Value = refused.json().await?;
|
||||
assert_eq!(body["error"]["code"], "unsupported_control");
|
||||
|
||||
client
|
||||
.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
|
||||
.json(&json!({"prompt": "hello"}))
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?;
|
||||
let total = wait_for_usage_total(&manager, &thread_id).await?;
|
||||
assert_eq!(total.payload["source"], "agent reported");
|
||||
assert_eq!(total.payload["cost"], "unknown");
|
||||
assert_eq!(total.payload["tokens"]["total_tokens"], 150);
|
||||
let log = received(&cwd);
|
||||
let prompt = log
|
||||
.iter()
|
||||
.find(|e| e["method"] == "session/prompt")
|
||||
.unwrap();
|
||||
assert_eq!(prompt["state"]["model"], chosen);
|
||||
|
||||
let stale = client
|
||||
.post(format!("http://{addr}/v1/threads"))
|
||||
.json(&json!({
|
||||
"workspace": cwd,
|
||||
"executor": {"kind": "acp", "agent_id": "fixture", "config": {"model": "model-z"}}
|
||||
}))
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json::<Value>()
|
||||
.await?;
|
||||
let stale_id = stale["id"].as_str().unwrap();
|
||||
let response = client
|
||||
.post(format!("http://{addr}/v1/threads/{stale_id}/turns"))
|
||||
.json(&json!({"prompt": "x"}))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(response.status(), HttpStatus::FAILED_DEPENDENCY);
|
||||
let body: Value = response.json().await?;
|
||||
assert_eq!(body["error"]["code"], "choice_not_offered");
|
||||
|
||||
handle.abort();
|
||||
Ok(())
|
||||
}
|
||||
+668
-11
@@ -22,6 +22,7 @@ use std::sync::mpsc::{Receiver, RecvTimeoutError, sync_channel};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde::Serialize;
|
||||
use serde_json::{Value, json};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
@@ -32,6 +33,21 @@ const METHOD_NOT_FOUND: i64 = -32601;
|
||||
pub const MAX_MESSAGE_BYTES: usize = 8 * 1024 * 1024;
|
||||
pub const AUTH_REQUIRED: i64 = -32000;
|
||||
const ABORT_POLL: Duration = Duration::from_millis(25);
|
||||
pub const MAX_CONTROLS: usize = 32;
|
||||
pub const MAX_CHOICES: usize = 256;
|
||||
pub const MAX_VALUE_BYTES: usize = 128;
|
||||
const MAX_NAME_CHARS: usize = 200;
|
||||
const MAX_DESCRIPTION_CHARS: usize = 400;
|
||||
|
||||
pub mod refusal {
|
||||
pub const CONTROL_NOT_OFFERED: &str = "control_not_offered";
|
||||
pub const CONTROL_UNAVAILABLE: &str = "control_unavailable";
|
||||
pub const CHOICE_NOT_OFFERED: &str = "choice_not_offered";
|
||||
pub const CHOICE_REJECTED: &str = "choice_rejected";
|
||||
pub const UNSUPPORTED_CONTROL: &str = "unsupported_control";
|
||||
pub const CONTROL_TYPE_UNSUPPORTED: &str = "control_type_unsupported";
|
||||
pub const CONTROL_HAS_NO_CHOICES: &str = "control_has_no_choices";
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum PermissionPolicy {
|
||||
@@ -117,6 +133,482 @@ pub struct AgentIdentity {
|
||||
pub protocol_version: u64,
|
||||
pub load_session: bool,
|
||||
pub resume_session: bool,
|
||||
pub capabilities: AgentCapabilities,
|
||||
}
|
||||
|
||||
/// What the agent said it can do at `initialize`; absent means not reported.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
|
||||
pub struct AgentCapabilities {
|
||||
pub load_session: bool,
|
||||
pub resume_session: bool,
|
||||
pub fork_session: bool,
|
||||
pub list_sessions: bool,
|
||||
pub prompt_image: bool,
|
||||
pub prompt_audio: bool,
|
||||
pub prompt_embedded_context: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct Choice {
|
||||
pub value: String,
|
||||
pub name: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub description: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum ControlKind {
|
||||
Model,
|
||||
Mode,
|
||||
Config,
|
||||
}
|
||||
|
||||
/// The one ACP method that writes a control, so a choice is never sent through
|
||||
/// a different door than the agent offered it on.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
#[serde(tag = "method")]
|
||||
pub enum ControlWrite {
|
||||
#[serde(rename = "session/set_config_option")]
|
||||
SetConfigOption { config_id: String },
|
||||
#[serde(rename = "session/set_model")]
|
||||
SetModel,
|
||||
#[serde(rename = "session/set_mode")]
|
||||
SetMode,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct ControlRefusal {
|
||||
pub code: &'static str,
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct Control {
|
||||
pub id: String,
|
||||
pub kind: ControlKind,
|
||||
pub name: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub description: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub category: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub current: Option<String>,
|
||||
pub choices: Vec<Choice>,
|
||||
pub truncated: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub write: Option<ControlWrite>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub unavailable: Option<ControlRefusal>,
|
||||
}
|
||||
|
||||
impl Control {
|
||||
pub fn offers(&self, value: &str) -> bool {
|
||||
self.unavailable.is_none() && self.choices.iter().any(|choice| choice.value == value)
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything a session handshake offered, bounded. Choices are the only values
|
||||
/// a caller may write; names and descriptions are display text.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
|
||||
pub struct SessionOffer {
|
||||
pub controls: Vec<Control>,
|
||||
pub malformed_controls: usize,
|
||||
}
|
||||
|
||||
impl SessionOffer {
|
||||
pub fn control(&self, id: &str) -> Option<&Control> {
|
||||
self.controls.iter().find(|control| control.id == id)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct ActionSupport {
|
||||
pub action: &'static str,
|
||||
pub supported: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub code: Option<&'static str>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reason: Option<&'static str>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub agent_reports: Option<bool>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct OptionsAgent {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub version: String,
|
||||
pub protocol_version: u64,
|
||||
}
|
||||
|
||||
/// What `GET /v1/agents/external/{id}/options` answers: the agent's own offer
|
||||
/// for one workspace, and which Codewhale actions it will and will not carry.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct AgentOptions {
|
||||
pub agent: OptionsAgent,
|
||||
pub cwd: String,
|
||||
pub capabilities: AgentCapabilities,
|
||||
pub controls: Vec<Control>,
|
||||
pub malformed_controls: usize,
|
||||
pub actions: Vec<ActionSupport>,
|
||||
}
|
||||
|
||||
impl AgentOptions {
|
||||
pub fn new(identity: &AgentIdentity, id: &str, cwd: &Path, offer: SessionOffer) -> Self {
|
||||
let actions = action_support(&identity.capabilities, &offer);
|
||||
Self {
|
||||
agent: OptionsAgent {
|
||||
id: id.to_string(),
|
||||
name: identity.name.clone(),
|
||||
version: identity.version.clone(),
|
||||
protocol_version: identity.protocol_version,
|
||||
},
|
||||
cwd: cwd.display().to_string(),
|
||||
capabilities: identity.capabilities.clone(),
|
||||
controls: offer.controls,
|
||||
malformed_controls: offer.malformed_controls,
|
||||
actions,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn action_support(capabilities: &AgentCapabilities, offer: &SessionOffer) -> Vec<ActionSupport> {
|
||||
let offered = |kind: ControlKind| {
|
||||
offer
|
||||
.controls
|
||||
.iter()
|
||||
.any(|control| control.kind == kind && control.unavailable.is_none())
|
||||
};
|
||||
let choice = |action, kind| {
|
||||
let supported = offered(kind);
|
||||
ActionSupport {
|
||||
action,
|
||||
supported,
|
||||
code: (!supported).then_some(refusal::CONTROL_NOT_OFFERED),
|
||||
reason: (!supported).then_some("The agent did not offer this control."),
|
||||
agent_reports: None,
|
||||
}
|
||||
};
|
||||
let refused = |action, reason, agent_reports| ActionSupport {
|
||||
action,
|
||||
supported: false,
|
||||
code: Some(refusal::UNSUPPORTED_CONTROL),
|
||||
reason: Some(reason),
|
||||
agent_reports,
|
||||
};
|
||||
vec![
|
||||
choice("model_choice", ControlKind::Model),
|
||||
choice("mode_choice", ControlKind::Mode),
|
||||
choice("config_choice", ControlKind::Config),
|
||||
refused(
|
||||
"steering",
|
||||
"Agent Client Protocol has no way to steer a running turn; interrupt it and send a new message.",
|
||||
None,
|
||||
),
|
||||
refused(
|
||||
"compaction",
|
||||
"Codewhale cannot compact another agent's context.",
|
||||
None,
|
||||
),
|
||||
refused(
|
||||
"fork",
|
||||
"Codewhale does not fork another agent's session.",
|
||||
Some(capabilities.fork_session),
|
||||
),
|
||||
refused(
|
||||
"images",
|
||||
"Codewhale sends text prompts to another agent only.",
|
||||
Some(capabilities.prompt_image),
|
||||
),
|
||||
refused(
|
||||
"dynamic_tools",
|
||||
"The agent uses its own tools; Codewhale tools cannot be added.",
|
||||
None,
|
||||
),
|
||||
refused(
|
||||
"output_limit",
|
||||
"Codewhale cannot cap another agent's output tokens.",
|
||||
None,
|
||||
),
|
||||
refused(
|
||||
"provider_override",
|
||||
"Codewhale providers do not apply; choose the agent's own model.",
|
||||
None,
|
||||
),
|
||||
refused(
|
||||
"remote_placement",
|
||||
"The agent runs on this computer only.",
|
||||
None,
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
pub struct NewSession {
|
||||
pub session_id: String,
|
||||
pub offer: SessionOffer,
|
||||
}
|
||||
|
||||
/// Token usage an agent put on a prompt answer. Absent fields are unknown.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
|
||||
pub struct AgentTokenUsage {
|
||||
pub input_tokens: Option<u64>,
|
||||
pub output_tokens: Option<u64>,
|
||||
pub total_tokens: Option<u64>,
|
||||
pub thought_tokens: Option<u64>,
|
||||
pub cached_read_tokens: Option<u64>,
|
||||
pub cached_write_tokens: Option<u64>,
|
||||
}
|
||||
|
||||
impl AgentTokenUsage {
|
||||
pub fn is_empty(&self) -> bool {
|
||||
*self == Self::default()
|
||||
}
|
||||
|
||||
fn from_result(result: &Value) -> Option<Self> {
|
||||
let usage = result.get("usage")?.as_object()?;
|
||||
let field = |name: &str| usage.get(name).and_then(Value::as_u64);
|
||||
let parsed = Self {
|
||||
input_tokens: field("inputTokens"),
|
||||
output_tokens: field("outputTokens"),
|
||||
total_tokens: field("totalTokens"),
|
||||
thought_tokens: field("thoughtTokens"),
|
||||
cached_read_tokens: field("cachedReadTokens"),
|
||||
cached_write_tokens: field("cachedWriteTokens"),
|
||||
};
|
||||
(!parsed.is_empty()).then_some(parsed)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct PromptReply {
|
||||
pub stop_reason: String,
|
||||
pub usage: Option<AgentTokenUsage>,
|
||||
}
|
||||
|
||||
fn bounded_text(value: Option<&Value>, max_chars: usize) -> Option<String> {
|
||||
let text: String = value?
|
||||
.as_str()?
|
||||
.chars()
|
||||
.filter(|ch| !ch.is_control())
|
||||
.take(max_chars)
|
||||
.collect();
|
||||
let text = text.trim();
|
||||
(!text.is_empty()).then(|| text.to_string())
|
||||
}
|
||||
|
||||
fn offered_value(value: Option<&Value>) -> Option<String> {
|
||||
let value = value?.as_str()?;
|
||||
(!value.is_empty() && value.len() <= MAX_VALUE_BYTES && !value.chars().any(char::is_control))
|
||||
.then(|| value.to_string())
|
||||
}
|
||||
|
||||
fn push_choice(
|
||||
choices: &mut Vec<Choice>,
|
||||
truncated: &mut bool,
|
||||
value: Option<String>,
|
||||
name: Option<&Value>,
|
||||
description: Option<&Value>,
|
||||
) {
|
||||
let Some(value) = value else {
|
||||
*truncated = true;
|
||||
return;
|
||||
};
|
||||
if choices.iter().any(|choice| choice.value == value) {
|
||||
return;
|
||||
}
|
||||
if choices.len() >= MAX_CHOICES {
|
||||
*truncated = true;
|
||||
return;
|
||||
}
|
||||
choices.push(Choice {
|
||||
name: bounded_text(name, MAX_NAME_CHARS).unwrap_or_else(|| value.clone()),
|
||||
description: bounded_text(description, MAX_DESCRIPTION_CHARS),
|
||||
value,
|
||||
});
|
||||
}
|
||||
|
||||
fn config_choices(options: &[Value]) -> (Vec<Choice>, bool) {
|
||||
let mut choices = Vec::new();
|
||||
let mut truncated = false;
|
||||
for option in options {
|
||||
if let Some(group) = option.get("options").and_then(Value::as_array) {
|
||||
for grouped in group {
|
||||
push_choice(
|
||||
&mut choices,
|
||||
&mut truncated,
|
||||
offered_value(grouped.get("value")),
|
||||
grouped.get("name"),
|
||||
grouped.get("description"),
|
||||
);
|
||||
}
|
||||
} else {
|
||||
push_choice(
|
||||
&mut choices,
|
||||
&mut truncated,
|
||||
offered_value(option.get("value")),
|
||||
option.get("name"),
|
||||
option.get("description"),
|
||||
);
|
||||
}
|
||||
}
|
||||
(choices, truncated)
|
||||
}
|
||||
|
||||
impl SessionOffer {
|
||||
/// Parse the `modes`, `models` and `configOptions` a `session/new`,
|
||||
/// `session/load` or `session/resume` answer carries. A config option that
|
||||
/// names a model or mode owns that control; the older `models` and `modes`
|
||||
/// blocks fill only what no config option claimed.
|
||||
pub fn from_session_result(result: &Value) -> Self {
|
||||
let mut offer = Self::default();
|
||||
let mut model_claimed = false;
|
||||
let mut mode_claimed = false;
|
||||
let options = result
|
||||
.get("configOptions")
|
||||
.and_then(Value::as_array)
|
||||
.map(Vec::as_slice)
|
||||
.unwrap_or_default();
|
||||
for option in options {
|
||||
if offer.controls.len() >= MAX_CONTROLS {
|
||||
offer.malformed_controls += 1;
|
||||
continue;
|
||||
}
|
||||
let Some(config_id) = offered_value(option.get("id")) else {
|
||||
offer.malformed_controls += 1;
|
||||
continue;
|
||||
};
|
||||
let category = bounded_text(option.get("category"), 64);
|
||||
let kind = if !model_claimed
|
||||
&& (config_id == "model" || category.as_deref() == Some("model"))
|
||||
{
|
||||
model_claimed = true;
|
||||
ControlKind::Model
|
||||
} else if !mode_claimed && (config_id == "mode" || category.as_deref() == Some("mode"))
|
||||
{
|
||||
mode_claimed = true;
|
||||
ControlKind::Mode
|
||||
} else {
|
||||
ControlKind::Config
|
||||
};
|
||||
let id = match kind {
|
||||
ControlKind::Model => "model".to_string(),
|
||||
ControlKind::Mode => "mode".to_string(),
|
||||
ControlKind::Config => format!("config:{config_id}"),
|
||||
};
|
||||
if offer.control(&id).is_some() {
|
||||
offer.malformed_controls += 1;
|
||||
continue;
|
||||
}
|
||||
let is_select = option.get("type").and_then(Value::as_str) == Some("select");
|
||||
let (choices, truncated) = if is_select {
|
||||
config_choices(
|
||||
option
|
||||
.get("options")
|
||||
.and_then(Value::as_array)
|
||||
.map(Vec::as_slice)
|
||||
.unwrap_or_default(),
|
||||
)
|
||||
} else {
|
||||
(Vec::new(), false)
|
||||
};
|
||||
let unavailable = if !is_select {
|
||||
Some(ControlRefusal {
|
||||
code: refusal::CONTROL_TYPE_UNSUPPORTED,
|
||||
reason: "Codewhale can only write select-type agent options.".to_string(),
|
||||
})
|
||||
} else if choices.is_empty() {
|
||||
Some(ControlRefusal {
|
||||
code: refusal::CONTROL_HAS_NO_CHOICES,
|
||||
reason: "The agent offered this option with no choices.".to_string(),
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
offer.controls.push(Control {
|
||||
name: bounded_text(option.get("name"), MAX_NAME_CHARS)
|
||||
.unwrap_or_else(|| config_id.clone()),
|
||||
description: bounded_text(option.get("description"), MAX_DESCRIPTION_CHARS),
|
||||
category,
|
||||
current: offered_value(option.get("currentValue")),
|
||||
write: unavailable
|
||||
.is_none()
|
||||
.then_some(ControlWrite::SetConfigOption { config_id }),
|
||||
id,
|
||||
kind,
|
||||
choices,
|
||||
truncated,
|
||||
unavailable,
|
||||
});
|
||||
}
|
||||
if !model_claimed
|
||||
&& let Some(models) = result.get("models").filter(|value| value.is_object())
|
||||
{
|
||||
offer.push_legacy(
|
||||
ControlKind::Model,
|
||||
models.get("currentModelId"),
|
||||
models.get("availableModels"),
|
||||
"modelId",
|
||||
ControlWrite::SetModel,
|
||||
);
|
||||
}
|
||||
if !mode_claimed && let Some(modes) = result.get("modes").filter(|value| value.is_object())
|
||||
{
|
||||
offer.push_legacy(
|
||||
ControlKind::Mode,
|
||||
modes.get("currentModeId"),
|
||||
modes.get("availableModes"),
|
||||
"id",
|
||||
ControlWrite::SetMode,
|
||||
);
|
||||
}
|
||||
offer
|
||||
}
|
||||
|
||||
fn push_legacy(
|
||||
&mut self,
|
||||
kind: ControlKind,
|
||||
current: Option<&Value>,
|
||||
available: Option<&Value>,
|
||||
value_key: &str,
|
||||
write: ControlWrite,
|
||||
) {
|
||||
let mut choices = Vec::new();
|
||||
let mut truncated = false;
|
||||
for entry in available
|
||||
.and_then(Value::as_array)
|
||||
.map(Vec::as_slice)
|
||||
.unwrap_or_default()
|
||||
{
|
||||
push_choice(
|
||||
&mut choices,
|
||||
&mut truncated,
|
||||
offered_value(entry.get(value_key)),
|
||||
entry.get("name"),
|
||||
entry.get("description"),
|
||||
);
|
||||
}
|
||||
if choices.is_empty() {
|
||||
self.malformed_controls += 1;
|
||||
return;
|
||||
}
|
||||
let (id, name) = match kind {
|
||||
ControlKind::Model => ("model", "Model"),
|
||||
_ => ("mode", "Mode"),
|
||||
};
|
||||
self.controls.push(Control {
|
||||
id: id.to_string(),
|
||||
kind,
|
||||
name: name.to_string(),
|
||||
description: None,
|
||||
category: None,
|
||||
current: offered_value(current),
|
||||
choices,
|
||||
truncated,
|
||||
write: Some(write),
|
||||
unavailable: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
@@ -456,7 +948,25 @@ impl AcpClient {
|
||||
.and_then(Value::as_u64)
|
||||
.filter(|version| *version == 1)
|
||||
.ok_or_else(|| AcpError::new("the agent did not negotiate ACP protocol version 1"))?;
|
||||
let flag = |pointer: &str| {
|
||||
capabilities
|
||||
.pointer(pointer)
|
||||
.is_some_and(|value| !value.is_null() && value != &Value::Bool(false))
|
||||
};
|
||||
let capability_report = AgentCapabilities {
|
||||
load_session: capabilities
|
||||
.get("loadSession")
|
||||
.and_then(Value::as_bool)
|
||||
.unwrap_or(false),
|
||||
resume_session: flag("/sessionCapabilities/resume"),
|
||||
fork_session: flag("/sessionCapabilities/fork"),
|
||||
list_sessions: flag("/sessionCapabilities/list"),
|
||||
prompt_image: flag("/promptCapabilities/image"),
|
||||
prompt_audio: flag("/promptCapabilities/audio"),
|
||||
prompt_embedded_context: flag("/promptCapabilities/embeddedContext"),
|
||||
};
|
||||
Ok(AgentIdentity {
|
||||
capabilities: capability_report,
|
||||
name: info
|
||||
.get("name")
|
||||
.and_then(Value::as_str)
|
||||
@@ -478,7 +988,7 @@ impl AcpClient {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn new_session(&mut self, cwd: &Path, timeout: Duration) -> Result<String, AcpError> {
|
||||
pub fn new_session(&mut self, cwd: &Path, timeout: Duration) -> Result<NewSession, AcpError> {
|
||||
let deadline = Instant::now() + timeout;
|
||||
let result = self.request(
|
||||
"session/new",
|
||||
@@ -488,11 +998,15 @@ impl AcpClient {
|
||||
&mut DiscardHandler,
|
||||
None,
|
||||
)?;
|
||||
result
|
||||
let session_id = result
|
||||
.get("sessionId")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| AcpError::new("session/new did not name a session"))
|
||||
.ok_or_else(|| AcpError::new("session/new did not name a session"))?;
|
||||
Ok(NewSession {
|
||||
session_id,
|
||||
offer: SessionOffer::from_session_result(&result),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn reopen_session(
|
||||
@@ -501,7 +1015,7 @@ impl AcpClient {
|
||||
session: &str,
|
||||
cwd: &Path,
|
||||
timeout: Duration,
|
||||
) -> Result<(), AcpError> {
|
||||
) -> Result<SessionOffer, AcpError> {
|
||||
let method = if identity.resume_session {
|
||||
"session/resume"
|
||||
} else if identity.load_session {
|
||||
@@ -520,6 +1034,45 @@ impl AcpClient {
|
||||
&mut DiscardHandler,
|
||||
None,
|
||||
)
|
||||
.map(|result| SessionOffer::from_session_result(&result))
|
||||
}
|
||||
|
||||
/// Write one offered choice through the method its control was offered on.
|
||||
pub fn apply_control(
|
||||
&mut self,
|
||||
session: &str,
|
||||
control: &Control,
|
||||
value: &str,
|
||||
timeout: Duration,
|
||||
) -> Result<(), AcpError> {
|
||||
let deadline = Instant::now() + timeout;
|
||||
let (method, params) = match control.write.as_ref() {
|
||||
Some(ControlWrite::SetConfigOption { config_id }) => (
|
||||
"session/set_config_option",
|
||||
json!({"sessionId": session, "configId": config_id, "value": value}),
|
||||
),
|
||||
Some(ControlWrite::SetModel) => (
|
||||
"session/set_model",
|
||||
json!({"sessionId": session, "modelId": value}),
|
||||
),
|
||||
Some(ControlWrite::SetMode) => (
|
||||
"session/set_mode",
|
||||
json!({"sessionId": session, "modeId": value}),
|
||||
),
|
||||
None => {
|
||||
return Err(AcpError::new(
|
||||
"the agent did not offer this control for writing",
|
||||
));
|
||||
}
|
||||
};
|
||||
self.request(
|
||||
method,
|
||||
params,
|
||||
Some(session),
|
||||
deadline,
|
||||
&mut DiscardHandler,
|
||||
None,
|
||||
)
|
||||
.map(|_| ())
|
||||
}
|
||||
|
||||
@@ -549,7 +1102,7 @@ impl AcpClient {
|
||||
timeout: Duration,
|
||||
handler: &mut dyn AcpHandler,
|
||||
cancel: Option<&CancellationToken>,
|
||||
) -> Result<String, AcpError> {
|
||||
) -> Result<PromptReply, AcpError> {
|
||||
let deadline = Instant::now() + timeout;
|
||||
let result = self.request(
|
||||
"session/prompt",
|
||||
@@ -559,12 +1112,16 @@ impl AcpClient {
|
||||
handler,
|
||||
cancel,
|
||||
)?;
|
||||
result
|
||||
let stop_reason = result
|
||||
.get("stopReason")
|
||||
.and_then(Value::as_str)
|
||||
.filter(|reason| !reason.is_empty())
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| AcpError::new("session/prompt did not report a stop reason"))
|
||||
.ok_or_else(|| AcpError::new("session/prompt did not report a stop reason"))?;
|
||||
Ok(PromptReply {
|
||||
stop_reason,
|
||||
usage: AgentTokenUsage::from_result(&result),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn prompt(
|
||||
@@ -575,7 +1132,7 @@ impl AcpClient {
|
||||
timeout: Duration,
|
||||
) -> Result<PromptOutcome, AcpError> {
|
||||
let mut outcome = PromptOutcome::default();
|
||||
let stop_reason = self.prompt_with(
|
||||
let reply = self.prompt_with(
|
||||
session,
|
||||
text,
|
||||
timeout,
|
||||
@@ -585,7 +1142,7 @@ impl AcpClient {
|
||||
},
|
||||
None,
|
||||
)?;
|
||||
outcome.stop_reason = stop_reason;
|
||||
outcome.stop_reason = reply.stop_reason;
|
||||
Ok(outcome)
|
||||
}
|
||||
|
||||
@@ -644,6 +1201,104 @@ mod tests {
|
||||
request
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_session_offer_is_parsed_bounded_and_never_widened() {
|
||||
let many: Vec<Value> = (0..(MAX_CHOICES + 40))
|
||||
.map(|index| json!({"value": format!("v{index}"), "name": "x".repeat(900)}))
|
||||
.collect();
|
||||
let offer = SessionOffer::from_session_result(&json!({
|
||||
"configOptions": [
|
||||
{"id": "model", "name": "Model", "category": "model", "type": "select",
|
||||
"currentValue": "a",
|
||||
"options": [{"group": "g", "name": "G", "options": [
|
||||
{"value": "a", "name": "A"}, {"value": "a", "name": "dup"}, {"value": "b", "name": "B", "description": "d"}]},
|
||||
{"value": "c", "name": "C"}]},
|
||||
{"id": "flag", "name": "Flag", "type": "boolean", "currentValue": true},
|
||||
{"id": "none", "name": "None", "type": "select", "options": []},
|
||||
{"id": "", "name": "No id", "type": "select", "options": []},
|
||||
{"id": "big", "name": "Big", "type": "select", "currentValue": "v0", "options": many},
|
||||
{"id": "bad\nid", "name": "Control char", "type": "select", "options": [{"value": "x"}]}
|
||||
],
|
||||
"modes": {"currentModeId": "m1", "availableModes": [{"id": "m1", "name": "M1"}, {"id": "", "name": "bad"}]},
|
||||
"models": {"currentModelId": "legacy", "availableModels": [{"modelId": "legacy", "name": "Legacy"}]}
|
||||
}));
|
||||
let ids: Vec<&str> = offer.controls.iter().map(|c| c.id.as_str()).collect();
|
||||
assert_eq!(
|
||||
ids,
|
||||
["model", "config:flag", "config:none", "config:big", "mode"]
|
||||
);
|
||||
assert_eq!(offer.malformed_controls, 2);
|
||||
let model = offer.control("model").unwrap();
|
||||
let values: Vec<&str> = model.choices.iter().map(|c| c.value.as_str()).collect();
|
||||
assert_eq!(
|
||||
values,
|
||||
["a", "b", "c"],
|
||||
"groups flatten and duplicates collapse"
|
||||
);
|
||||
assert_eq!(model.choices[1].description.as_deref(), Some("d"));
|
||||
assert!(model.offers("c") && !model.offers("legacy") && !model.offers(""));
|
||||
assert_eq!(
|
||||
offer
|
||||
.control("config:flag")
|
||||
.unwrap()
|
||||
.unavailable
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.code,
|
||||
refusal::CONTROL_TYPE_UNSUPPORTED
|
||||
);
|
||||
assert!(!offer.control("config:flag").unwrap().offers("true"));
|
||||
assert_eq!(
|
||||
offer
|
||||
.control("config:none")
|
||||
.unwrap()
|
||||
.unavailable
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.code,
|
||||
refusal::CONTROL_HAS_NO_CHOICES
|
||||
);
|
||||
let big = offer.control("config:big").unwrap();
|
||||
assert_eq!(big.choices.len(), MAX_CHOICES);
|
||||
assert!(big.truncated);
|
||||
assert!(big.choices[0].name.chars().count() <= MAX_NAME_CHARS);
|
||||
assert!(!big.offers(&format!("v{}", MAX_CHOICES + 1)));
|
||||
let mode = offer.control("mode").unwrap();
|
||||
assert_eq!(mode.write, Some(ControlWrite::SetMode));
|
||||
assert_eq!(mode.choices.len(), 1);
|
||||
assert!(
|
||||
mode.truncated,
|
||||
"a dropped malformed choice marks the list incomplete"
|
||||
);
|
||||
|
||||
let legacy = SessionOffer::from_session_result(&json!({
|
||||
"models": {"currentModelId": "legacy", "availableModels": [{"modelId": "legacy", "name": "Legacy"}]}
|
||||
}));
|
||||
assert_eq!(
|
||||
legacy.control("model").unwrap().write,
|
||||
Some(ControlWrite::SetModel)
|
||||
);
|
||||
assert!(
|
||||
SessionOffer::from_session_result(&json!({}))
|
||||
.controls
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_usage_keeps_only_what_the_agent_reported() {
|
||||
let usage = AgentTokenUsage::from_result(&json!({
|
||||
"stopReason": "end_turn",
|
||||
"usage": {"inputTokens": 5, "outputTokens": "many", "totalTokens": 9}
|
||||
}))
|
||||
.unwrap();
|
||||
assert_eq!(usage.input_tokens, Some(5));
|
||||
assert_eq!(usage.output_tokens, None);
|
||||
assert_eq!(usage.total_tokens, Some(9));
|
||||
assert!(AgentTokenUsage::from_result(&json!({"stopReason": "end_turn"})).is_none());
|
||||
assert!(AgentTokenUsage::from_result(&json!({"usage": {}})).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_json_rpc_worker_speaks_acp_end_to_end() {
|
||||
let Some(_node) = node() else {
|
||||
@@ -677,7 +1332,8 @@ mod tests {
|
||||
|
||||
let session = client
|
||||
.new_session(workspace.path(), Duration::from_secs(20))
|
||||
.unwrap();
|
||||
.unwrap()
|
||||
.session_id;
|
||||
let allowed = client
|
||||
.prompt(
|
||||
&session,
|
||||
@@ -821,7 +1477,8 @@ mod tests {
|
||||
client.initialize(Duration::from_secs(20)).unwrap();
|
||||
let session = client
|
||||
.new_session(workspace.path(), Duration::from_secs(20))
|
||||
.unwrap();
|
||||
.unwrap()
|
||||
.session_id;
|
||||
let canceller = client.canceller(&session);
|
||||
let prompt_session = session.clone();
|
||||
let blocked = std::thread::spawn(move || {
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
use std::sync::mpsc as std_mpsc;
|
||||
|
||||
use super::*;
|
||||
use crate::fleet::acp::{AcpCanceller, AcpClient, AcpHandler, AgentIdentity, PermissionAnswer};
|
||||
use crate::fleet::acp::{
|
||||
AcpCanceller, AcpClient, AcpError, AcpHandler, AgentIdentity, AgentOptions, AgentTokenUsage,
|
||||
PermissionAnswer, PromptReply, SessionOffer, refusal,
|
||||
};
|
||||
use crate::fleet::agents::ExternalAgentLaunch;
|
||||
use crate::fleet::host::{
|
||||
FleetHostAdapter, FleetWorkerCommand, FleetWorkerStartRequest, FleetWorkerTransport,
|
||||
@@ -19,6 +22,11 @@ const MAX_ID_BYTES: usize = 128;
|
||||
const MAX_TOOL_ITEMS_PER_TURN: usize = 512;
|
||||
const MAX_LOCATIONS: usize = 32;
|
||||
const MAX_OPTIONS: usize = 16;
|
||||
const MAX_CONCURRENT_OPTION_PROBES: usize = 4;
|
||||
static OPTIONS_PROBES: std::sync::LazyLock<Arc<tokio::sync::Semaphore>> =
|
||||
std::sync::LazyLock::new(|| {
|
||||
Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_OPTION_PROBES))
|
||||
});
|
||||
const KILLED_REASON: &str = "The agent did not stop; Codewhale ended its process.";
|
||||
|
||||
pub(crate) struct ExternalThreadState {
|
||||
@@ -47,7 +55,7 @@ enum AgentSignal {
|
||||
params: Value,
|
||||
reply: std_mpsc::Sender<PermissionAnswer>,
|
||||
},
|
||||
Finished(Result<String, String>),
|
||||
Finished(Result<PromptReply, String>),
|
||||
}
|
||||
|
||||
pub(crate) struct ExternalSession {
|
||||
@@ -145,6 +153,149 @@ struct SessionStart {
|
||||
session_id: String,
|
||||
}
|
||||
|
||||
struct AgentProcess {
|
||||
host: Option<LocalProcessFleetHostAdapter>,
|
||||
worker_id: String,
|
||||
}
|
||||
|
||||
impl AgentProcess {
|
||||
fn spawn(
|
||||
log_root: &Path,
|
||||
worker_id: String,
|
||||
workspace: &Path,
|
||||
launch: &ExternalAgentLaunch,
|
||||
) -> Result<(Self, AcpClient)> {
|
||||
let mut host = LocalProcessFleetHostAdapter::new(log_root);
|
||||
let mut request = FleetWorkerStartRequest::new(
|
||||
worker_id.clone(),
|
||||
FleetWorkerCommand::new(launch.program.display().to_string(), launch.args.clone()),
|
||||
);
|
||||
request.transport = FleetWorkerTransport::JsonRpc;
|
||||
request.cwd = Some(workspace.to_path_buf());
|
||||
request.env_allowlist = launch.env.keys().cloned().collect();
|
||||
request.env = launch.env.clone();
|
||||
host.start_worker(request)
|
||||
.map_err(|error| anyhow!("{} could not start: {error}", launch.agent_name))?;
|
||||
let io = host.take_worker_io(&worker_id);
|
||||
let process = Self {
|
||||
host: Some(host),
|
||||
worker_id,
|
||||
};
|
||||
let Some(io) = io else {
|
||||
bail!("{} started without a protocol channel", launch.agent_name);
|
||||
};
|
||||
Ok((process, AcpClient::new(io)))
|
||||
}
|
||||
|
||||
fn release(mut self) -> LocalProcessFleetHostAdapter {
|
||||
self.host.take().expect("an agent process is released once")
|
||||
}
|
||||
|
||||
fn stop(&mut self) {
|
||||
if let Some(mut host) = self.host.take()
|
||||
&& let Err(error) = host.stop_worker(&self.worker_id)
|
||||
{
|
||||
tracing::warn!(worker = %self.worker_id, %error, "stopping an external agent failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AgentProcess {
|
||||
fn drop(&mut self) {
|
||||
self.stop();
|
||||
}
|
||||
}
|
||||
|
||||
fn sign_in_or(
|
||||
error: anyhow::Error,
|
||||
agent_name: &str,
|
||||
login: Option<&'static str>,
|
||||
) -> anyhow::Error {
|
||||
if error
|
||||
.downcast_ref::<AcpError>()
|
||||
.is_some_and(AcpError::is_auth_required)
|
||||
{
|
||||
let hint = login
|
||||
.map(|command| format!(" Run `{command}` in a terminal, then try again."))
|
||||
.unwrap_or_default();
|
||||
return crate::fleet::agents::ExecutorUnavailable::new(
|
||||
crate::fleet::agents::ExecutorUnavailable::NEEDS_SIGN_IN,
|
||||
format!("{agent_name} needs you to sign in to it first.{hint}"),
|
||||
)
|
||||
.into();
|
||||
}
|
||||
if error
|
||||
.downcast_ref::<crate::fleet::agents::ExecutorUnavailable>()
|
||||
.is_some()
|
||||
{
|
||||
return error;
|
||||
}
|
||||
anyhow!("{agent_name}: {error}")
|
||||
}
|
||||
|
||||
fn executor_choices(executor: &ExecutorRecord) -> Vec<(&'static str, String)> {
|
||||
executor
|
||||
.model
|
||||
.iter()
|
||||
.map(|model| ("model", model.clone()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn refuse(code: &'static str, message: impl Into<String>) -> anyhow::Error {
|
||||
crate::fleet::agents::ExecutorUnavailable::new(code, message).into()
|
||||
}
|
||||
|
||||
fn apply_choices(
|
||||
client: &mut AcpClient,
|
||||
session_id: &str,
|
||||
offer: &SessionOffer,
|
||||
choices: &[(&'static str, String)],
|
||||
) -> Result<()> {
|
||||
for (id, value) in choices {
|
||||
let shown = label(Some(&Value::String(value.clone())), "");
|
||||
if offer.controls.is_empty() && *id == "model" {
|
||||
client
|
||||
.set_config_option(session_id, "model", value, SETUP_TIMEOUT)
|
||||
.map_err(|error| rejected_choice(id, &shown, error))?;
|
||||
continue;
|
||||
}
|
||||
let Some(control) = offer.control(id) else {
|
||||
return Err(refuse(
|
||||
refusal::CONTROL_NOT_OFFERED,
|
||||
format!("The agent does not offer a {id} choice, so `{shown}` was not applied."),
|
||||
));
|
||||
};
|
||||
if let Some(why) = control.unavailable.as_ref() {
|
||||
return Err(refuse(refusal::CONTROL_UNAVAILABLE, why.reason.clone()));
|
||||
}
|
||||
if !control.offers(value) {
|
||||
return Err(refuse(
|
||||
refusal::CHOICE_NOT_OFFERED,
|
||||
format!(
|
||||
"The agent no longer offers `{shown}` for {id}; choose one of its current options."
|
||||
),
|
||||
));
|
||||
}
|
||||
if control.current.as_deref() == Some(value.as_str()) {
|
||||
continue;
|
||||
}
|
||||
client
|
||||
.apply_control(session_id, control, value, SETUP_TIMEOUT)
|
||||
.map_err(|error| rejected_choice(id, &shown, error))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn rejected_choice(id: &str, shown: &str, error: AcpError) -> anyhow::Error {
|
||||
if error.is_auth_required() {
|
||||
return error.into();
|
||||
}
|
||||
refuse(
|
||||
refusal::CHOICE_REJECTED,
|
||||
format!("The agent refused {id} `{shown}`: {error}"),
|
||||
)
|
||||
}
|
||||
|
||||
fn start_session_blocking(
|
||||
log_root: PathBuf,
|
||||
thread_id: String,
|
||||
@@ -155,77 +306,44 @@ fn start_session_blocking(
|
||||
abandoned: CancellationToken,
|
||||
) -> Result<SessionStart> {
|
||||
let worker_id = format!("acp-{thread_id}");
|
||||
let mut host = LocalProcessFleetHostAdapter::new(&log_root);
|
||||
let mut request = FleetWorkerStartRequest::new(
|
||||
worker_id.clone(),
|
||||
FleetWorkerCommand::new(launch.program.display().to_string(), launch.args.clone()),
|
||||
);
|
||||
request.transport = FleetWorkerTransport::JsonRpc;
|
||||
request.cwd = Some(workspace.clone());
|
||||
request.env_allowlist = launch.env.keys().cloned().collect();
|
||||
request.env = launch.env.clone();
|
||||
host.start_worker(request)
|
||||
.map_err(|error| anyhow!("{} could not start: {error}", launch.agent_name))?;
|
||||
let Some(io) = host.take_worker_io(&worker_id) else {
|
||||
let _ = host.stop_worker(&worker_id);
|
||||
bail!("{} started without a protocol channel", launch.agent_name);
|
||||
};
|
||||
let mut client = AcpClient::new(io);
|
||||
let (process, mut client) =
|
||||
AgentProcess::spawn(&log_root, worker_id.clone(), &workspace, &launch)?;
|
||||
client.abort_requests_on(Some(abandoned));
|
||||
let setup = (|| {
|
||||
let choices = executor_choices(&executor);
|
||||
let setup = (|| -> Result<(AgentIdentity, String)> {
|
||||
let identity = client.initialize(SETUP_TIMEOUT)?;
|
||||
let session_id = match executor.foreign_session_id.as_deref() {
|
||||
let (session_id, offer) = match executor.foreign_session_id.as_deref() {
|
||||
Some(session_id) => {
|
||||
client.reopen_session(&identity, session_id, &workspace, SETUP_TIMEOUT)?;
|
||||
session_id.to_string()
|
||||
let offer =
|
||||
client.reopen_session(&identity, session_id, &workspace, SETUP_TIMEOUT)?;
|
||||
(session_id.to_string(), offer)
|
||||
}
|
||||
None => {
|
||||
let started = client.new_session(&workspace, SETUP_TIMEOUT)?;
|
||||
(started.session_id, started.offer)
|
||||
}
|
||||
None => client.new_session(&workspace, SETUP_TIMEOUT)?,
|
||||
};
|
||||
if let Some(model) = executor.model.as_deref() {
|
||||
client.set_config_option(&session_id, "model", model, SETUP_TIMEOUT)?;
|
||||
}
|
||||
Ok::<_, crate::fleet::acp::AcpError>((identity, session_id))
|
||||
apply_choices(&mut client, &session_id, &offer, &choices)?;
|
||||
Ok((identity, session_id))
|
||||
})();
|
||||
let (identity, session_id) = match setup {
|
||||
Ok(setup) => setup,
|
||||
Err(error) => {
|
||||
let _ = host.stop_worker(&worker_id);
|
||||
if error.is_auth_required() {
|
||||
let hint = login
|
||||
.map(|command| format!(" Run `{command}` in a terminal, then try again."))
|
||||
.unwrap_or_default();
|
||||
return Err(crate::fleet::agents::ExecutorUnavailable::new(
|
||||
crate::fleet::agents::ExecutorUnavailable::NEEDS_SIGN_IN,
|
||||
format!(
|
||||
"{} needs you to sign in to it first.{hint}",
|
||||
launch.agent_name
|
||||
),
|
||||
)
|
||||
.into());
|
||||
}
|
||||
bail!("{}: {error}", launch.agent_name);
|
||||
}
|
||||
};
|
||||
let (identity, session_id) =
|
||||
setup.map_err(|error| sign_in_or(error, &launch.agent_name, login))?;
|
||||
if !valid_id(&session_id) {
|
||||
let _ = host.stop_worker(&worker_id);
|
||||
bail!("{} named an invalid session id", launch.agent_name);
|
||||
}
|
||||
client.abort_requests_on(None);
|
||||
let canceller = client.canceller(&session_id);
|
||||
let (commands, receiver) = std_mpsc::channel();
|
||||
let session_for_thread = session_id.clone();
|
||||
if let Err(error) = std::thread::Builder::new()
|
||||
std::thread::Builder::new()
|
||||
.name(format!("acp-{}", launch.agent_id))
|
||||
.spawn(move || run_session(client, session_for_thread, receiver))
|
||||
{
|
||||
let _ = host.stop_worker(&worker_id);
|
||||
bail!("starting the agent session thread failed: {error}");
|
||||
}
|
||||
.map_err(|error| anyhow!("starting the agent session thread failed: {error}"))?;
|
||||
Ok(SessionStart {
|
||||
session: ExternalSession {
|
||||
commands: parking_lot::Mutex::new(Some(commands)),
|
||||
canceller,
|
||||
host: parking_lot::Mutex::new(host),
|
||||
host: parking_lot::Mutex::new(process.release()),
|
||||
worker_id,
|
||||
},
|
||||
identity,
|
||||
@@ -233,6 +351,31 @@ fn start_session_blocking(
|
||||
})
|
||||
}
|
||||
|
||||
fn probe_options_blocking(
|
||||
log_root: PathBuf,
|
||||
workspace: PathBuf,
|
||||
launch: ExternalAgentLaunch,
|
||||
login: Option<&'static str>,
|
||||
abandoned: CancellationToken,
|
||||
) -> Result<AgentOptions> {
|
||||
let worker_id = format!("acp-options-{}", runtime_record_id("probe"));
|
||||
let (mut process, mut client) = AgentProcess::spawn(&log_root, worker_id, &workspace, &launch)?;
|
||||
client.abort_requests_on(Some(abandoned));
|
||||
let probed = (|| -> Result<AgentOptions> {
|
||||
let identity = client.initialize(SETUP_TIMEOUT)?;
|
||||
let started = client.new_session(&workspace, SETUP_TIMEOUT)?;
|
||||
Ok(AgentOptions::new(
|
||||
&identity,
|
||||
&launch.agent_id,
|
||||
&workspace,
|
||||
started.offer,
|
||||
))
|
||||
})();
|
||||
drop(client);
|
||||
process.stop();
|
||||
probed.map_err(|error| sign_in_or(error, &launch.agent_name, login))
|
||||
}
|
||||
|
||||
fn valid_id(value: &str) -> bool {
|
||||
!value.is_empty()
|
||||
&& value.len() <= MAX_ID_BYTES
|
||||
@@ -394,6 +537,8 @@ struct ExternalProjection {
|
||||
plan: Option<TurnItemRecord>,
|
||||
title: Option<String>,
|
||||
failure: Option<String>,
|
||||
context: Option<(Option<u64>, Option<u64>)>,
|
||||
tokens: Option<AgentTokenUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@@ -637,6 +782,50 @@ impl RuntimeThreadManager {
|
||||
Ok(session)
|
||||
}
|
||||
|
||||
/// Ask an agent what it offers for one workspace, on a throwaway process
|
||||
/// that is stopped before this returns, on every path.
|
||||
pub async fn external_agent_options(&self, agent_id: &str, cwd: &Path) -> Result<AgentOptions> {
|
||||
self.ensure_accepting_execution()?;
|
||||
let agent = crate::fleet::agents::find(agent_id).ok_or_else(|| {
|
||||
crate::fleet::agents::ExecutorUnavailable::new(
|
||||
crate::fleet::agents::ExecutorUnavailable::UNKNOWN_AGENT,
|
||||
format!("No external agent is named `{agent_id}`."),
|
||||
)
|
||||
})?;
|
||||
let workspace = tokio::fs::canonicalize(cwd)
|
||||
.await
|
||||
.map_err(|error| anyhow!("cwd {} is unavailable: {error}", cwd.display()))?;
|
||||
anyhow::ensure!(
|
||||
tokio::fs::metadata(&workspace)
|
||||
.await
|
||||
.is_ok_and(|metadata| metadata.is_dir()),
|
||||
"cwd {} is not a directory",
|
||||
workspace.display()
|
||||
);
|
||||
let launch = crate::fleet::agents::launch(agent.id).await?;
|
||||
let log_root = self
|
||||
.session_store_binding()
|
||||
.data_dir
|
||||
.join("external-agents");
|
||||
tokio::fs::create_dir_all(&log_root)
|
||||
.await
|
||||
.context("creating the external agent log directory failed")?;
|
||||
let abandoned = self.cancel_token.child_token();
|
||||
let _abandon = abandoned.clone().drop_guard();
|
||||
let permit = tokio::select! {
|
||||
permit = Arc::clone(&OPTIONS_PROBES).acquire_owned() => permit?,
|
||||
() = abandoned.cancelled() => bail!("Runtime is shutting down; execution admission is closed"),
|
||||
};
|
||||
self.ensure_accepting_execution()?;
|
||||
let login = agent.login;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let _permit = permit;
|
||||
probe_options_blocking(log_root, workspace, launch, login, abandoned)
|
||||
})
|
||||
.await
|
||||
.context("external agent options worker failed")?
|
||||
}
|
||||
|
||||
pub(super) async fn external_thread_loaded(&self, thread_id: &str) -> bool {
|
||||
self.active.lock().await.external.contains_key(thread_id)
|
||||
}
|
||||
@@ -724,27 +913,36 @@ impl RuntimeThreadManager {
|
||||
matches!(input_source, RuntimeTurnInputSource::ExternalUser),
|
||||
"this kind of turn is not available in a session with another agent"
|
||||
);
|
||||
anyhow::ensure!(
|
||||
req.images.is_empty(),
|
||||
"image inputs are not available in a session with another agent"
|
||||
);
|
||||
anyhow::ensure!(
|
||||
req.dynamic_tools.is_empty(),
|
||||
"dynamic tools are not available in a session with another agent"
|
||||
);
|
||||
anyhow::ensure!(
|
||||
req.model.is_none()
|
||||
&& req.reasoning_effort.is_none()
|
||||
&& req.max_output_tokens.is_none()
|
||||
&& req.model_provider.is_none()
|
||||
&& req.model_provider_id.is_none()
|
||||
&& req.account_model_owner.is_none(),
|
||||
"a session with another agent uses its saved agent model; per-turn model controls are unavailable"
|
||||
);
|
||||
anyhow::ensure!(
|
||||
req.allowed_tools.is_none(),
|
||||
"a session with another agent uses that agent's own tools; tool allowlists are unavailable"
|
||||
);
|
||||
if !req.images.is_empty() {
|
||||
return Err(refuse(
|
||||
refusal::UNSUPPORTED_CONTROL,
|
||||
"Image inputs are not available in a session with another agent.",
|
||||
));
|
||||
}
|
||||
if !req.dynamic_tools.is_empty() {
|
||||
return Err(refuse(
|
||||
refusal::UNSUPPORTED_CONTROL,
|
||||
"Dynamic tools are not available in a session with another agent.",
|
||||
));
|
||||
}
|
||||
if req.model.is_some()
|
||||
|| req.reasoning_effort.is_some()
|
||||
|| req.max_output_tokens.is_some()
|
||||
|| req.model_provider.is_some()
|
||||
|| req.model_provider_id.is_some()
|
||||
|| req.account_model_owner.is_some()
|
||||
{
|
||||
return Err(refuse(
|
||||
refusal::UNSUPPORTED_CONTROL,
|
||||
"A session with another agent keeps its saved agent model; per-turn model, reasoning, output and provider controls are unavailable.",
|
||||
));
|
||||
}
|
||||
if req.allowed_tools.is_some() {
|
||||
return Err(refuse(
|
||||
refusal::UNSUPPORTED_CONTROL,
|
||||
"A session with another agent uses that agent's own tools; tool allowlists are unavailable.",
|
||||
));
|
||||
}
|
||||
if req.environment_id.is_some() {
|
||||
return Err(crate::fleet::agents::ExecutorUnavailable::new(
|
||||
crate::fleet::agents::ExecutorUnavailable::ENVIRONMENT_NOT_SUPPORTED,
|
||||
@@ -1041,10 +1239,21 @@ impl RuntimeThreadManager {
|
||||
};
|
||||
let _ = reply.send(answer);
|
||||
}
|
||||
Some(AgentSignal::Finished(result)) => break result,
|
||||
Some(AgentSignal::Finished(result)) => {
|
||||
break result.map(|reply| {
|
||||
projection.tokens = reply.usage;
|
||||
reply.stop_reason
|
||||
});
|
||||
}
|
||||
None => break Err("the agent session ended".to_string()),
|
||||
}
|
||||
};
|
||||
if let Err(error) = self
|
||||
.emit_agent_usage(&thread.id, &turn_id, &projection)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(thread_id = %thread.id, turn_id = %turn_id, %error, "recording an external agent's usage failed");
|
||||
}
|
||||
if let Err(error) = self
|
||||
.settle_external_turn(&thread, &turn_id, projection, outcome)
|
||||
.await
|
||||
@@ -1053,6 +1262,32 @@ impl RuntimeThreadManager {
|
||||
}
|
||||
}
|
||||
|
||||
async fn emit_agent_usage(
|
||||
&self,
|
||||
thread_id: &str,
|
||||
turn_id: &str,
|
||||
projection: &ExternalProjection,
|
||||
) -> Result<()> {
|
||||
let tokens = projection
|
||||
.tokens
|
||||
.as_ref()
|
||||
.filter(|tokens| !tokens.is_empty());
|
||||
let reported = tokens.is_some() || projection.context.is_some();
|
||||
let payload = json!({
|
||||
"kind": "turn_total",
|
||||
"source": if reported { "agent reported" } else { "none" },
|
||||
"tokens": tokens.map_or_else(|| json!("unknown"), |tokens| json!(tokens)),
|
||||
"context": projection.context.map_or_else(
|
||||
|| json!("unknown"),
|
||||
|(used, size)| json!({ "used": used, "size": size }),
|
||||
),
|
||||
"cost": "unknown",
|
||||
});
|
||||
self.emit_event(thread_id, Some(turn_id), None, "turn.agent_usage", payload)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn open_external_item(
|
||||
&self,
|
||||
thread_id: &str,
|
||||
@@ -1265,12 +1500,13 @@ impl RuntimeThreadManager {
|
||||
let used = update.get("used").and_then(Value::as_u64);
|
||||
let size = update.get("size").and_then(Value::as_u64);
|
||||
if used.is_some() || size.is_some() {
|
||||
projection.context = Some((used, size));
|
||||
self.emit_event(
|
||||
thread_id,
|
||||
Some(turn_id),
|
||||
None,
|
||||
"turn.agent_usage",
|
||||
json!({ "used": used, "size": size, "source": "agent reported", "cost": "unknown" }),
|
||||
json!({ "kind": "context_window", "used": used, "size": size, "source": "agent reported", "cost": "unknown" }),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
@@ -715,3 +715,390 @@ async fn an_abandoned_external_startup_stops_its_agent_inside_the_shutdown_barri
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn options_fixture() -> PathBuf {
|
||||
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/acp-options-agent.mjs")
|
||||
}
|
||||
|
||||
fn received(workspace: &Path) -> Vec<Value> {
|
||||
std::fs::read_to_string(workspace.join("options-received.jsonl"))
|
||||
.unwrap_or_default()
|
||||
.lines()
|
||||
.filter_map(|line| serde_json::from_str(line).ok())
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn pid_alive(pid: u64) -> bool {
|
||||
std::process::Command::new("kill")
|
||||
.args(["-0", &pid.to_string()])
|
||||
.stderr(std::process::Stdio::null())
|
||||
.status()
|
||||
.is_ok_and(|status| status.success())
|
||||
}
|
||||
|
||||
fn refusal_code(error: &anyhow::Error) -> Option<&'static str> {
|
||||
error
|
||||
.downcast_ref::<crate::fleet::agents::ExecutorUnavailable>()
|
||||
.map(|refusal| refusal.code)
|
||||
}
|
||||
|
||||
fn options_thread_request(workspace: &Path, model: &str) -> CreateThreadRequest {
|
||||
let mut request = fixture_thread_request(workspace, "ask");
|
||||
request.executor.as_mut().unwrap().config.model = Some(model.to_string());
|
||||
request
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn an_options_probe_reports_the_agents_own_offer_and_leaves_no_process() -> Result<()> {
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let _secret = crate::test_support::EnvVarGuard::set("CODEWHALE_TEST_SECRET", "must-not-travel");
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
|
||||
let options = manager
|
||||
.external_agent_options(FIXTURE_AGENT_ID, &workspace)
|
||||
.await?;
|
||||
let json = serde_json::to_value(&options)?;
|
||||
let controls: Vec<&str> = options.controls.iter().map(|c| c.id.as_str()).collect();
|
||||
assert_eq!(
|
||||
controls,
|
||||
["model", "config:effort", "config:verbose", "mode"]
|
||||
);
|
||||
let model = &options.controls[0];
|
||||
assert_eq!(model.current.as_deref(), Some("model-a"));
|
||||
let values: Vec<&str> = model.choices.iter().map(|c| c.value.as_str()).collect();
|
||||
assert_eq!(values, ["model-a", "model-b", "model-c"]);
|
||||
assert_eq!(
|
||||
json["controls"][0]["write"]["method"],
|
||||
"session/set_config_option"
|
||||
);
|
||||
assert_eq!(json["controls"][3]["write"]["method"], "session/set_mode");
|
||||
let verbose = &options.controls[2];
|
||||
assert_eq!(
|
||||
verbose.unavailable.as_ref().map(|u| u.code),
|
||||
Some("control_type_unsupported")
|
||||
);
|
||||
assert!(verbose.write.is_none());
|
||||
assert_eq!(json["capabilities"]["prompt_image"], true);
|
||||
assert_eq!(json["capabilities"]["fork_session"], true);
|
||||
let action = |name: &str| {
|
||||
json["actions"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.find(|entry| entry["action"] == name)
|
||||
.cloned()
|
||||
.unwrap()
|
||||
};
|
||||
assert_eq!(action("model_choice")["supported"], true);
|
||||
assert_eq!(action("steering")["code"], "unsupported_control");
|
||||
assert_eq!(action("images")["supported"], false);
|
||||
assert_eq!(action("images")["agent_reports"], true);
|
||||
|
||||
let log = received(&workspace);
|
||||
let pid = log[0]["pid"].as_u64().expect("the probe agent's pid");
|
||||
assert!(!pid_alive(pid), "the probe's agent process was stopped");
|
||||
assert!(log.iter().all(|entry| entry["method"] != "session/prompt"));
|
||||
let dumped = std::fs::read_to_string(workspace.join("options-env.json"))?;
|
||||
assert!(!dumped.contains("CODEWHALE_TEST_SECRET"));
|
||||
|
||||
let wrong = manager
|
||||
.external_agent_options(FIXTURE_AGENT_ID, &workspace.join("missing"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(wrong.to_string().contains("unavailable"), "{wrong}");
|
||||
let unknown = manager
|
||||
.external_agent_options("nope", &workspace)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(refusal_code(&unknown), Some("unknown_agent"));
|
||||
|
||||
std::fs::write(
|
||||
workspace.join("options-fixture.json"),
|
||||
r#"{"auth_required":true}"#,
|
||||
)?;
|
||||
let unauthenticated = manager
|
||||
.external_agent_options(FIXTURE_AGENT_ID, &workspace)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(refusal_code(&unauthenticated), Some("needs_sign_in"));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn an_options_probe_ends_when_the_runtime_shuts_down() -> Result<()> {
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
std::fs::write(workspace.join("options-fixture.json"), r#"{"hang":true}"#)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
let probing = manager.clone();
|
||||
let probe_workspace = workspace.clone();
|
||||
let probe = tokio::spawn(async move {
|
||||
probing
|
||||
.external_agent_options(FIXTURE_AGENT_ID, &probe_workspace)
|
||||
.await
|
||||
});
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
let pid = loop {
|
||||
if let Some(pid) = received(&workspace)
|
||||
.iter()
|
||||
.find(|entry| entry["method"] == "session/new")
|
||||
.and_then(|entry| entry["pid"].as_u64())
|
||||
{
|
||||
break pid;
|
||||
}
|
||||
anyhow::ensure!(
|
||||
Instant::now() < deadline,
|
||||
"the probe never reached session/new"
|
||||
);
|
||||
sleep(Duration::from_millis(20)).await;
|
||||
};
|
||||
assert!(pid_alive(pid));
|
||||
manager.close_execution_admission().await;
|
||||
let ended = tokio::time::timeout(Duration::from_secs(20), probe).await??;
|
||||
assert!(ended.is_err());
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
while pid_alive(pid) {
|
||||
anyhow::ensure!(
|
||||
Instant::now() < deadline,
|
||||
"the probe's agent outlived shutdown"
|
||||
);
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn a_chosen_model_reaches_the_agent_survives_reopen_and_usage_is_agent_reported() -> Result<()>
|
||||
{
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
let offered = manager
|
||||
.external_agent_options(FIXTURE_AGENT_ID, &workspace)
|
||||
.await?;
|
||||
let chosen = &offered.controls[0].choices[1].value;
|
||||
assert_eq!(chosen, "model-b");
|
||||
|
||||
let thread = manager
|
||||
.create_thread(options_thread_request(&workspace, chosen))
|
||||
.await?;
|
||||
let turn = manager.start_turn(&thread.id, prompt("first")).await?;
|
||||
let settled = wait_for_turn(&manager, &turn.id).await?;
|
||||
assert_eq!(settled.status, RuntimeTurnStatus::Completed, "{settled:?}");
|
||||
let detail = manager.get_thread_detail(&thread.id).await?;
|
||||
let message = detail
|
||||
.items
|
||||
.iter()
|
||||
.find(|item| item.kind == TurnItemKind::AgentMessage)
|
||||
.and_then(|item| item.detail.clone())
|
||||
.unwrap_or_default();
|
||||
assert!(message.contains("model=model-b"), "{message}");
|
||||
|
||||
let usage: Vec<_> = events(&manager, &thread.id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.filter(|event| event.event == "turn.agent_usage")
|
||||
.collect();
|
||||
let window = usage
|
||||
.iter()
|
||||
.find(|event| event.payload["kind"] == "context_window")
|
||||
.expect("a context window update");
|
||||
assert_eq!(window.payload["used"], 1200);
|
||||
assert_eq!(window.payload["size"], 200000);
|
||||
let total = usage
|
||||
.iter()
|
||||
.find(|event| event.payload["kind"] == "turn_total")
|
||||
.expect("a turn total");
|
||||
assert_eq!(total.payload["source"], "agent reported");
|
||||
assert_eq!(total.payload["cost"], "unknown");
|
||||
assert_eq!(total.payload["tokens"]["total_tokens"], 150);
|
||||
assert_eq!(total.payload["tokens"]["input_tokens"], 100);
|
||||
assert_eq!(total.payload["context"]["used"], 1200);
|
||||
|
||||
assert!(manager.stop_external_agent(&thread.id).await?);
|
||||
let reopened = manager.start_turn(&thread.id, prompt("second")).await?;
|
||||
let settled = wait_for_turn(&manager, &reopened.id).await?;
|
||||
assert_eq!(settled.status, RuntimeTurnStatus::Completed, "{settled:?}");
|
||||
let log = received(&workspace);
|
||||
let sets: Vec<_> = log
|
||||
.iter()
|
||||
.filter(|entry| entry["method"] == "session/set_config_option")
|
||||
.collect();
|
||||
assert_eq!(sets.len(), 2, "{log:?}");
|
||||
assert!(sets.iter().all(|entry| entry["value"] == "model-b"));
|
||||
assert_ne!(
|
||||
sets[0]["pid"], sets[1]["pid"],
|
||||
"a new agent process took the saved choice"
|
||||
);
|
||||
let prompts: Vec<_> = log
|
||||
.iter()
|
||||
.filter(|entry| entry["method"] == "session/prompt")
|
||||
.collect();
|
||||
assert_eq!(prompts.len(), 2);
|
||||
assert!(log.iter().any(|entry| entry["method"] == "session/load"));
|
||||
assert_eq!(prompts[1]["state"]["model"], "model-b");
|
||||
assert_eq!(
|
||||
manager
|
||||
.store
|
||||
.load_thread(&thread.id)?
|
||||
.executor
|
||||
.and_then(|record| record.model)
|
||||
.as_deref(),
|
||||
Some("model-b")
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn an_agent_that_reports_no_usage_says_unknown() -> Result<()> {
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
std::fs::write(
|
||||
workspace.join("options-fixture.json"),
|
||||
r#"{"no_usage":true}"#,
|
||||
)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
let thread = manager
|
||||
.create_thread(fixture_thread_request(&workspace, "ask"))
|
||||
.await?;
|
||||
let turn = manager.start_turn(&thread.id, prompt("quiet")).await?;
|
||||
wait_for_turn(&manager, &turn.id).await?;
|
||||
let events = events(&manager, &thread.id).await?;
|
||||
let usage: Vec<_> = events
|
||||
.iter()
|
||||
.filter(|event| event.event == "turn.agent_usage")
|
||||
.collect();
|
||||
assert_eq!(usage.len(), 1, "{usage:?}");
|
||||
assert_eq!(usage[0].payload["tokens"], "unknown");
|
||||
assert_eq!(usage[0].payload["context"], "unknown");
|
||||
assert_eq!(usage[0].payload["source"], "none");
|
||||
assert_eq!(usage[0].payload["cost"], "unknown");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn a_stale_or_withdrawn_choice_is_refused_before_any_prompt() -> Result<()> {
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
|
||||
let invented = manager
|
||||
.create_thread(options_thread_request(&workspace, "model-z"))
|
||||
.await?;
|
||||
let error = manager
|
||||
.start_turn(&invented.id, prompt("never"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(
|
||||
refusal_code(&error),
|
||||
Some("choice_not_offered"),
|
||||
"{error:#}"
|
||||
);
|
||||
|
||||
std::fs::write(
|
||||
workspace.join("options-fixture.json"),
|
||||
r#"{"withdraw":["model-b"]}"#,
|
||||
)?;
|
||||
let withdrawn = manager
|
||||
.create_thread(options_thread_request(&workspace, "model-b"))
|
||||
.await?;
|
||||
let error = manager
|
||||
.start_turn(&withdrawn.id, prompt("never"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(
|
||||
refusal_code(&error),
|
||||
Some("choice_not_offered"),
|
||||
"{error:#}"
|
||||
);
|
||||
|
||||
let log = received(&workspace);
|
||||
assert!(log.iter().all(|entry| entry["method"] != "session/prompt"));
|
||||
assert!(
|
||||
log.iter()
|
||||
.all(|entry| entry["method"] != "session/set_config_option")
|
||||
);
|
||||
for pid in log.iter().filter_map(|entry| entry["pid"].as_u64()) {
|
||||
assert!(!pid_alive(pid), "a refused setup left agent {pid} running");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn per_turn_controls_an_agent_cannot_honor_are_typed_refusals() -> Result<()> {
|
||||
if !node_available().await {
|
||||
return Ok(());
|
||||
}
|
||||
let _env = crate::test_support::lock_test_env();
|
||||
let _fixture = crate::test_support::EnvVarGuard::set(FIXTURE_AGENT_ENV, options_fixture());
|
||||
let root = tempfile::tempdir()?;
|
||||
let workspace = root.path().join("work");
|
||||
std::fs::create_dir_all(&workspace)?;
|
||||
let manager = test_manager(root.path().join("runtime"))?;
|
||||
let thread = manager
|
||||
.create_thread(fixture_thread_request(&workspace, "ask"))
|
||||
.await?;
|
||||
for request in [
|
||||
StartTurnRequest {
|
||||
allowed_tools: Some(Vec::new()),
|
||||
..prompt("x")
|
||||
},
|
||||
StartTurnRequest {
|
||||
model: Some("model-b".to_string()),
|
||||
..prompt("x")
|
||||
},
|
||||
StartTurnRequest {
|
||||
reasoning_effort: Some("high".to_string()),
|
||||
..prompt("x")
|
||||
},
|
||||
] {
|
||||
let error = manager.start_turn(&thread.id, request).await.unwrap_err();
|
||||
assert_eq!(
|
||||
refusal_code(&error),
|
||||
Some("unsupported_control"),
|
||||
"{error:#}"
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
received(&workspace).is_empty(),
|
||||
"no agent was started for a refused turn"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
+165
@@ -0,0 +1,165 @@
|
||||
// A deterministic Agent Client Protocol agent that advertises its own models,
|
||||
// a mode, a config option and one control Codewhale cannot write, and reports
|
||||
// token usage. Behavior is steered by `options-fixture.json` in the session
|
||||
// cwd; everything it receives is appended to `options-received.jsonl` there.
|
||||
import { appendFileSync, existsSync, readFileSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
|
||||
const plan = () => {
|
||||
const path = join(process.cwd(), 'options-fixture.json');
|
||||
return existsSync(path) ? JSON.parse(readFileSync(path, 'utf8')) : {};
|
||||
};
|
||||
const record = (entry) =>
|
||||
appendFileSync(join(process.cwd(), 'options-received.jsonl'), `${JSON.stringify({ pid: process.pid, ...entry })}\n`);
|
||||
writeFileSync(join(process.cwd(), 'options-env.json'), JSON.stringify(process.env));
|
||||
|
||||
const MODELS = [
|
||||
{ value: 'model-a', name: 'Model A', description: 'Fast' },
|
||||
{ value: 'model-b', name: 'Model B', description: 'Careful' },
|
||||
{ value: 'model-c', name: 'Model C' },
|
||||
];
|
||||
const EFFORTS = [
|
||||
{ value: 'low', name: 'Low' },
|
||||
{ value: 'high', name: 'High' },
|
||||
];
|
||||
const MODES = [
|
||||
{ id: 'default', name: 'Default' },
|
||||
{ id: 'review', name: 'Review only', description: 'Reads, never edits' },
|
||||
];
|
||||
|
||||
let buffer = '';
|
||||
let sessionId = null;
|
||||
const state = { model: 'model-a', effort: 'low', mode: 'default' };
|
||||
|
||||
const send = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
|
||||
const result = (id, value) => send({ jsonrpc: '2.0', id, result: value });
|
||||
const failure = (id, code, message) => send({ jsonrpc: '2.0', id, error: { code, message } });
|
||||
const notify = (method, params) => send({ jsonrpc: '2.0', method, params });
|
||||
|
||||
function offer() {
|
||||
const withdrawn = new Set(plan().withdraw ?? []);
|
||||
const models = MODELS.filter((model) => !withdrawn.has(model.value));
|
||||
const configOptions = [
|
||||
{ id: 'model', name: 'Model', category: 'model', type: 'select', currentValue: state.model, options: models },
|
||||
{ id: 'effort', name: 'Reasoning effort', category: 'thought_level', type: 'select', currentValue: state.effort, options: EFFORTS },
|
||||
{ id: 'verbose', name: 'Verbose logging', type: 'boolean', currentValue: false },
|
||||
];
|
||||
const reply = {
|
||||
configOptions,
|
||||
modes: { currentModeId: state.mode, availableModes: MODES },
|
||||
};
|
||||
if (plan().malformed) {
|
||||
reply.configOptions.push({ id: '', name: 'No id', type: 'select', currentValue: 'x', options: [] });
|
||||
reply.configOptions.push({ id: 'empty', name: 'Empty', type: 'select', currentValue: 'x', options: [] });
|
||||
}
|
||||
if (plan().huge) {
|
||||
reply.configOptions.push({
|
||||
id: 'many',
|
||||
name: 'Many',
|
||||
type: 'select',
|
||||
currentValue: 'v0',
|
||||
options: Array.from({ length: 600 }, (_, index) => ({ value: `v${index}`, name: `Value ${index}` })),
|
||||
});
|
||||
}
|
||||
return reply;
|
||||
}
|
||||
|
||||
function offered(config, value) {
|
||||
const found = offer().configOptions.find((option) => option.id === config);
|
||||
return Boolean(found && found.type === 'select' && found.options.some((option) => option.value === value));
|
||||
}
|
||||
|
||||
function prompt(request) {
|
||||
const text = (request.params?.prompt ?? []).map((block) => block.text ?? '').join('');
|
||||
record({ method: 'session/prompt', text, state: { ...state } });
|
||||
notify('session/update', {
|
||||
sessionId,
|
||||
update: {
|
||||
sessionUpdate: 'agent_message_chunk',
|
||||
content: { type: 'text', text: `model=${state.model} effort=${state.effort} mode=${state.mode}` },
|
||||
},
|
||||
});
|
||||
if (!plan().no_usage) {
|
||||
notify('session/update', { sessionId, update: { sessionUpdate: 'usage_update', used: 1200, size: 200000 } });
|
||||
}
|
||||
result(request.id, {
|
||||
stopReason: 'end_turn',
|
||||
...(plan().no_usage ? {} : { usage: { inputTokens: 100, outputTokens: 50, totalTokens: 150, thoughtTokens: 7 } }),
|
||||
});
|
||||
}
|
||||
|
||||
function handle(message) {
|
||||
if (message.id !== undefined && message.method === undefined) return;
|
||||
const params = message.params ?? {};
|
||||
switch (message.method) {
|
||||
case 'initialize':
|
||||
result(message.id, {
|
||||
protocolVersion: 1,
|
||||
agentInfo: { name: 'options-fixture', version: '1.0.0' },
|
||||
authMethods: [],
|
||||
agentCapabilities: {
|
||||
loadSession: true,
|
||||
promptCapabilities: { image: true, embeddedContext: true },
|
||||
sessionCapabilities: { fork: {} },
|
||||
},
|
||||
});
|
||||
break;
|
||||
case 'session/new':
|
||||
if (plan().auth_required) {
|
||||
failure(message.id, -32000, 'Authentication required');
|
||||
break;
|
||||
}
|
||||
sessionId = `options-session-${process.pid}`;
|
||||
record({ method: 'session/new', cwd: params.cwd });
|
||||
if (plan().hang) break;
|
||||
result(message.id, { sessionId, ...offer() });
|
||||
break;
|
||||
case 'session/load':
|
||||
case 'session/resume':
|
||||
sessionId = params.sessionId ?? null;
|
||||
record({ method: message.method, sessionId });
|
||||
result(message.id, offer());
|
||||
break;
|
||||
case 'session/set_config_option':
|
||||
record({ method: message.method, configId: params.configId, value: params.value });
|
||||
if (!offered(params.configId, params.value)) {
|
||||
failure(message.id, -32602, 'value not offered');
|
||||
break;
|
||||
}
|
||||
state[params.configId] = params.value;
|
||||
result(message.id, { configOptions: offer().configOptions });
|
||||
break;
|
||||
case 'session/set_mode':
|
||||
record({ method: message.method, modeId: params.modeId });
|
||||
if (!MODES.some((mode) => mode.id === params.modeId)) {
|
||||
failure(message.id, -32602, 'mode not offered');
|
||||
break;
|
||||
}
|
||||
state.mode = params.modeId;
|
||||
result(message.id, {});
|
||||
break;
|
||||
case 'session/prompt':
|
||||
prompt(message);
|
||||
break;
|
||||
case 'session/cancel':
|
||||
break;
|
||||
default:
|
||||
if (message.id !== undefined) failure(message.id, -32601, `no method ${message.method}`);
|
||||
}
|
||||
}
|
||||
|
||||
process.stdin.on('data', (chunk) => {
|
||||
buffer += chunk;
|
||||
let index;
|
||||
while ((index = buffer.indexOf('\n')) >= 0) {
|
||||
const line = buffer.slice(0, index);
|
||||
buffer = buffer.slice(index + 1);
|
||||
if (!line.trim()) continue;
|
||||
try {
|
||||
handle(JSON.parse(line));
|
||||
} catch (error) {
|
||||
process.stderr.write(`options fixture: bad line: ${String(error)}\n`);
|
||||
}
|
||||
}
|
||||
});
|
||||
process.stdin.on('end', () => process.exit(0));
|
||||
Reference in new issue
Block a user