task(acp): upgrade SDK and use new HTTP/WS crate (#10082)
Some checks are pending
Canary / bundle-desktop-windows-cuda (push) Blocked by required conditions
Canary / Release (push) Blocked by required conditions
Canary / Prepare Version (push) Waiting to run
Canary / build-cli (push) Blocked by required conditions
Canary / Upload Install Script (push) Blocked by required conditions
Canary / bundle-desktop (push) Blocked by required conditions
Canary / bundle-desktop-intel (push) Blocked by required conditions
Canary / bundle-desktop-linux (push) Blocked by required conditions
Canary / bundle-desktop-windows (push) Blocked by required conditions
Cargo Deny / deny (push) Waiting to run
Unused Dependencies / machete (push) Waiting to run
CI / Check Rust Code Format (push) Blocked by required conditions
CI / changes (push) Waiting to run
CI / Build and Test Rust Project (push) Blocked by required conditions
CI / Build Rust Project on Windows (push) Waiting to run
CI / Check MSRV (push) Blocked by required conditions
CI / Lint Rust Code (push) Blocked by required conditions
CI / Check Generated Schemas are Up-to-Date (push) Blocked by required conditions
CI / Test and Lint Electron Desktop App (push) Blocked by required conditions
Create Minor Release PR / check-version-bump-pr (push) Waiting to run
Create Minor Release PR / release (push) Blocked by required conditions
Live Provider Tests / Smoke Tests (push) Blocked by required conditions
Live Provider Tests / check-fork (push) Waiting to run
Live Provider Tests / changes (push) Blocked by required conditions
Live Provider Tests / Build Binary (push) Blocked by required conditions
Live Provider Tests / Smoke Tests (Code Execution) (push) Blocked by required conditions
Live Provider Tests / Compaction Tests (push) Blocked by required conditions
Live Provider Tests / goose server HTTP integration tests (push) Blocked by required conditions
Publish Docker Image / docker (push) Waiting to run
Scorecard supply-chain security / Scorecard analysis (push) Waiting to run

This commit is contained in:
Alex Hancock 2026-06-29 01:27:11 -04:00 committed by GitHub
parent a1711e75e2
commit 2cc1140dc1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
39 changed files with 748 additions and 1522 deletions

186
Cargo.lock generated
View file

@ -40,43 +40,58 @@ dependencies = [
[[package]]
name = "agent-client-protocol"
version = "0.11.1"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2af62fb84df2af0f933d8f5fd78b843fa5eb0ec5a48fa1b528c41951d0bbe36c"
checksum = "e981a7d0207a0e9eb67d0faafc04e8c6601dea572d67a3163798897c0024521d"
dependencies = [
"agent-client-protocol-derive",
"agent-client-protocol-schema",
"anyhow",
"async-process",
"blocking",
"futures",
"futures-concurrency",
"jsonrpcmsg",
"rmcp",
"rustc-hash 2.1.2",
"schemars 1.2.1",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"shell-words",
"tracing",
"uuid",
"windows-sys 0.61.2",
]
[[package]]
name = "agent-client-protocol-derive"
version = "0.11.1"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cabdc9d845d08ec7ed2d0c9de1ae4a1b198301407d55855261572761be90ec9f"
checksum = "0d589c28cbe2978c76f8714cc304bc08355ee2f05d120806717725d9ffa12143"
dependencies = [
"quote",
"syn 2.0.117",
]
[[package]]
name = "agent-client-protocol-schema"
version = "0.12.0"
name = "agent-client-protocol-http"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49bae57dad1c28a362fbdcf7bab0583316a02b45a70792109fced55780a3b63c"
checksum = "171d55fdd4a4cab6dd1ca33c2deb7ae4990f91050da15a7b045e89a6db387826"
dependencies = [
"agent-client-protocol",
"async-stream",
"axum",
"futures",
"serde_json",
"tokio",
"tower-http 0.7.0",
"tracing",
"uuid",
]
[[package]]
name = "agent-client-protocol-schema"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ac542aba230234b1591ace7286a47c0514fe3efc3037d43296bde31ba7ee5728"
dependencies = [
"anyhow",
"derive_more",
@ -379,6 +394,18 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "async-channel"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2"
dependencies = [
"concurrent-queue",
"event-listener-strategy",
"futures-core",
"pin-project-lite",
]
[[package]]
name = "async-compression"
version = "0.4.42"
@ -391,12 +418,77 @@ dependencies = [
"tokio",
]
[[package]]
name = "async-io"
version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc"
dependencies = [
"autocfg",
"cfg-if 1.0.4",
"concurrent-queue",
"futures-io",
"futures-lite",
"parking",
"polling",
"rustix 1.1.4",
"slab",
"windows-sys 0.61.2",
]
[[package]]
name = "async-lock"
version = "3.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
dependencies = [
"event-listener",
"event-listener-strategy",
"pin-project-lite",
]
[[package]]
name = "async-once-cell"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4288f83726785267c6f2ef073a3d83dc3f9b81464e9f99898240cced85fce35a"
[[package]]
name = "async-process"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75"
dependencies = [
"async-channel",
"async-io",
"async-lock",
"async-signal",
"async-task",
"blocking",
"cfg-if 1.0.4",
"event-listener",
"futures-lite",
"rustix 1.1.4",
]
[[package]]
name = "async-signal"
version = "0.2.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485"
dependencies = [
"async-io",
"async-lock",
"atomic-waker",
"cfg-if 1.0.4",
"futures-core",
"futures-io",
"rustix 1.1.4",
"signal-hook-registry",
"slab",
"windows-sys 0.61.2",
]
[[package]]
name = "async-stream"
version = "0.3.6"
@ -419,6 +511,12 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "async-task"
version = "4.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
[[package]]
name = "async-trait"
version = "0.1.89"
@ -956,6 +1054,7 @@ dependencies = [
"tower",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
@ -974,6 +1073,7 @@ dependencies = [
"sync_wrapper",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
@ -1597,6 +1697,19 @@ dependencies = [
"objc2",
]
[[package]]
name = "blocking"
version = "1.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21"
dependencies = [
"async-channel",
"async-task",
"futures-io",
"futures-lite",
"piper",
]
[[package]]
name = "bon"
version = "3.9.3"
@ -3915,6 +4028,16 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "event-listener-strategy"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
dependencies = [
"event-listener",
"pin-project-lite",
]
[[package]]
name = "exr"
version = "1.74.0"
@ -4755,6 +4878,7 @@ name = "goose"
version = "1.39.0"
dependencies = [
"agent-client-protocol",
"agent-client-protocol-http",
"agent-client-protocol-schema",
"anyhow",
"arboard",
@ -4792,7 +4916,6 @@ dependencies = [
"goose-test-support",
"hf-hub 1.0.0-rc.1",
"http 1.4.2",
"http-body-util",
"icu_calendar",
"icu_locale",
"ignore",
@ -6155,16 +6278,6 @@ dependencies = [
"serde_json",
]
[[package]]
name = "jsonrpcmsg"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d833a15225c779251e13929203518c2ff26e2fe0f322d584b213f4f4dad37bd"
dependencies = [
"serde",
"serde_json",
]
[[package]]
name = "jsonschema"
version = "0.30.0"
@ -8095,6 +8208,17 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
[[package]]
name = "piper"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1"
dependencies = [
"atomic-waker",
"fastrand",
"futures-io",
]
[[package]]
name = "pkcs1"
version = "0.7.5"
@ -8177,6 +8301,20 @@ dependencies = [
"miniz_oxide",
]
[[package]]
name = "polling"
version = "3.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218"
dependencies = [
"cfg-if 1.0.4",
"concurrent-queue",
"hermit-abi",
"pin-project-lite",
"rustix 1.1.4",
"windows-sys 0.61.2",
]
[[package]]
name = "poly1305"
version = "0.8.0"

View file

@ -21,8 +21,9 @@ string_slice = "warn"
[workspace.dependencies]
rmcp = { version = "1.4", default-features = false, features = ["schemars", "auth"] }
agent-client-protocol-schema = { version = "0.12", default-features = false, features = ["unstable"] }
agent-client-protocol = { version = "0.11", default-features = false }
agent-client-protocol-schema = { version = "1.1", default-features = false, features = ["unstable"] }
agent-client-protocol = { version = "1.0", default-features = false }
agent-client-protocol-http = { version = "1.0", default-features = false, features = ["server", "unstable_cancel_request"] }
arboard = { version = "3", default-features = false }
anyhow = { version = "1.0.102", default-features = false, features = ["std"] }
async-stream = { version = "0.3.6", default-features = false }

View file

@ -1,4 +1,4 @@
use agent_client_protocol::schema::{AvailableCommand, ContentBlock, McpServer, SessionInfo};
use agent_client_protocol::schema::v1::{AvailableCommand, ContentBlock, McpServer, SessionInfo};
use agent_client_protocol::{JsonRpcRequest, JsonRpcResponse};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

View file

@ -1,4 +1,4 @@
use agent_client_protocol::schema::SessionInfo;
use agent_client_protocol::schema::v1::SessionInfo;
use agent_client_protocol::{JsonRpcRequest, JsonRpcResponse};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

View file

@ -30,7 +30,7 @@ uniffi = { version = "0.31", features = ["cli"], optional = true }
thiserror = { version = "2", optional = true }
[dev-dependencies]
tokio = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "process", "io-std", "io-util"] }
tokio-util = { workspace = true, features = ["compat", "rt"] }
[package.metadata.cargo-machete]

View file

@ -19,11 +19,11 @@
//! cargo run -p goose-sdk --example acp_client -- --goose-bin ./target/debug/goose "Explain Rust's ownership model in one sentence"
//! ```
use agent_client_protocol::schema::{
ContentBlock, InitializeRequest, ProtocolVersion, RequestPermissionOutcome,
RequestPermissionRequest, RequestPermissionResponse, SelectedPermissionOutcome,
SessionNotification, SessionUpdate,
use agent_client_protocol::schema::v1::{
ContentBlock, InitializeRequest, RequestPermissionOutcome, RequestPermissionRequest,
RequestPermissionResponse, SelectedPermissionOutcome, SessionNotification, SessionUpdate,
};
use agent_client_protocol::schema::ProtocolVersion;
use agent_client_protocol::{Client, ConnectionTo};
use goose_sdk::custom_requests::GetExtensionsRequest;
use std::path::PathBuf;

View file

@ -87,7 +87,7 @@ thiserror = { workspace = true }
futures = { workspace = true }
dirs = { workspace = true }
reqwest = { workspace = true, features = ["json", "cookies", "gzip", "brotli", "deflate", "zstd", "charset", "http2", "stream", "blocking", "multipart", "system-proxy"] }
tokio = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync", "time", "fs", "process", "net", "signal", "io-std", "io-util"] }
serde = { workspace = true }
serde_json = { workspace = true }
serde_path_to_error = { version = "0.1.8", default-features = false }
@ -132,6 +132,7 @@ tokio-cron-scheduler = { version = "0.15", default-features = false }
urlencoding = { workspace = true }
v_htmlescape = { version = "0.17", default-features = false, features = ["std"] }
sqlx = { version = "0.8.5", default-features = false, features = [
"runtime-tokio",
"sqlite",
"chrono",
"json",
@ -158,6 +159,7 @@ tempfile = { workspace = true }
tokio-util = { workspace = true, features = ["compat"] }
agent-client-protocol-schema = { workspace = true }
agent-client-protocol = { workspace = true, features = ["unstable"] }
agent-client-protocol-http = { workspace = true }
unicode-normalization = { version = "0.1.22", default-features = false, features = ["std"] }
# For local Whisper transcription (optional, behind "local-inference" feature)
@ -196,7 +198,6 @@ pastey = { version = "0.2", default-features = false }
shell-words = { workspace = true }
goose-acp-macros = { path = "../goose-acp-macros", default-features = false }
tower-http = { workspace = true, features = ["cors"] }
http-body-util = { version = "0.1.2", default-features = false }
tracing-appender = { workspace = true }
process-wrap = { version = "9", default-features = false, features = ["std"] }
nostr = { version = "0.44", default-features = false, features = ["nip44", "std"], optional = true }

View file

@ -176,6 +176,20 @@
],
"description": "SSE transport configuration\n\nOnly available when the Agent capabilities indicate `mcp_capabilities.sse` is `true`."
},
{
"$ref": "#/$defs/McpServerAcp",
"type": "object",
"properties": {
"type": {
"type": "string",
"const": "acp"
}
},
"required": [
"type"
],
"description": "**UNSTABLE**\n\nThis capability is not part of the spec yet, and may be removed or changed at any point.\n\nACP transport configuration\n\nOnly available when the Agent capabilities indicate `mcp_capabilities.acp` is `true`.\nThe MCP server is provided by an ACP component and communicates over the ACP channel."
},
{
"$ref": "#/$defs/McpServerStdio",
"description": "Stdio transport configuration\n\nAll Agents MUST support this transport."
@ -287,6 +301,36 @@
],
"description": "SSE transport configuration for MCP."
},
"McpServerAcpId": {
"type": "string",
"description": "**UNSTABLE**\n\nThis capability is not part of the spec yet, and may be removed or changed at any point.\n\nUnique identifier for an MCP server using the ACP transport.\n\nThe value is opaque and generated by the ACP component providing the MCP server. It is\nused by `mcp/connect` to route connection requests back to the component that declared the\nserver."
},
"McpServerAcp": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Human-readable name identifying this MCP server."
},
"id": {
"$ref": "#/$defs/McpServerAcpId",
"description": "Unique identifier for this MCP server, generated by the component providing it.\n\nProviders MUST NOT reuse an ID for multiple ACP-transport MCP servers that are visible\non the same ACP connection."
},
"_meta": {
"type": [
"object",
"null"
],
"additionalProperties": {},
"description": "The _meta property is reserved by ACP to allow clients and agents to attach additional\nmetadata to their interactions. Implementations MUST NOT make assumptions about values at\nthese keys.\n\nSee protocol docs: [Extensibility](https://agentclientprotocol.com/protocol/extensibility)"
}
},
"required": [
"name",
"id"
],
"description": "**UNSTABLE**\n\nThis capability is not part of the spec yet, and may be removed or changed at any point.\n\nACP transport configuration for MCP.\n\nThe MCP server is provided by an ACP component and communicates over the ACP channel\nusing `mcp/connect`, `mcp/message`, and `mcp/disconnect`."
},
"McpServerStdio": {
"type": "object",
"properties": {
@ -830,20 +874,25 @@
],
"items": {
"$ref": "#/$defs/Role"
}
},
"description": "Intended recipients for this content, such as the user or assistant.",
"x-deserialize-default-on-error": {},
"x-deserialize-skip-invalid-items": {}
},
"lastModified": {
"type": [
"string",
"null"
]
],
"description": "Timestamp indicating when the underlying resource was last modified."
},
"priority": {
"type": [
"number",
"null"
],
"format": "double"
"format": "double",
"description": "Relative importance of this content when clients choose what to surface."
},
"_meta": {
"type": [
@ -857,10 +906,17 @@
"description": "Optional annotations for the client. The client can use annotations to inform how objects are used or displayed"
},
"Role": {
"type": "string",
"enum": [
"assistant",
"user"
"oneOf": [
{
"type": "string",
"const": "assistant",
"description": "The assistant side of a conversation."
},
{
"type": "string",
"const": "user",
"description": "The user side of a conversation."
}
],
"description": "The sender or recipient of messages and data in a conversation."
},
@ -875,10 +931,13 @@
{
"type": "null"
}
]
],
"description": "Optional annotations that help clients decide how to display or route this content.",
"x-deserialize-default-on-error": {}
},
"text": {
"type": "string"
"type": "string",
"description": "Text payload carried by this content block."
},
"_meta": {
"type": [
@ -905,19 +964,24 @@
{
"type": "null"
}
]
],
"description": "Optional annotations that help clients decide how to display or route this content.",
"x-deserialize-default-on-error": {}
},
"data": {
"type": "string"
"type": "string",
"description": "Base64-encoded media payload."
},
"mimeType": {
"type": "string"
"type": "string",
"description": "MIME type describing the encoded media payload."
},
"uri": {
"type": [
"string",
"null"
]
],
"description": "URI associated with this resource or media payload."
},
"_meta": {
"type": [
@ -945,13 +1009,17 @@
{
"type": "null"
}
]
],
"description": "Optional annotations that help clients decide how to display or route this content.",
"x-deserialize-default-on-error": {}
},
"data": {
"type": "string"
"type": "string",
"description": "Base64-encoded media payload."
},
"mimeType": {
"type": "string"
"type": "string",
"description": "MIME type describing the encoded media payload."
},
"_meta": {
"type": [
@ -979,37 +1047,45 @@
{
"type": "null"
}
]
],
"description": "Optional annotations that help clients decide how to display or route this content.",
"x-deserialize-default-on-error": {}
},
"description": {
"type": [
"string",
"null"
]
],
"description": "Optional human-readable details shown with this protocol object."
},
"mimeType": {
"type": [
"string",
"null"
]
],
"description": "MIME type describing the encoded media payload."
},
"name": {
"type": "string"
"type": "string",
"description": "Human-readable name shown for this protocol object."
},
"size": {
"type": [
"integer",
"null"
]
],
"description": "Optional size of the linked resource in bytes, if known."
},
"title": {
"type": [
"string",
"null"
]
],
"description": "Optional display title for end-user UI."
},
"uri": {
"type": "string"
"type": "string",
"description": "URI associated with this resource or media payload."
},
"_meta": {
"type": [
@ -1029,10 +1105,12 @@
"EmbeddedResourceResource": {
"anyOf": [
{
"$ref": "#/$defs/TextResourceContents"
"$ref": "#/$defs/TextResourceContents",
"description": "Text resource contents embedded directly in the message."
},
{
"$ref": "#/$defs/BlobResourceContents"
"$ref": "#/$defs/BlobResourceContents",
"description": "Binary resource contents embedded directly in the message."
}
],
"description": "Resource content that can be embedded in a message."
@ -1044,13 +1122,16 @@
"type": [
"string",
"null"
]
],
"description": "MIME type describing the encoded media payload."
},
"text": {
"type": "string"
"type": "string",
"description": "Text payload carried by this content block."
},
"uri": {
"type": "string"
"type": "string",
"description": "URI associated with this resource or media payload."
},
"_meta": {
"type": [
@ -1071,16 +1152,19 @@
"type": "object",
"properties": {
"blob": {
"type": "string"
"type": "string",
"description": "Base64-encoded bytes for a binary resource payload."
},
"mimeType": {
"type": [
"string",
"null"
]
],
"description": "MIME type describing the encoded media payload."
},
"uri": {
"type": "string"
"type": "string",
"description": "URI associated with this resource or media payload."
},
"_meta": {
"type": [
@ -1108,10 +1192,13 @@
{
"type": "null"
}
]
],
"description": "Optional annotations that help clients decide how to display or route this content.",
"x-deserialize-default-on-error": {}
},
"resource": {
"$ref": "#/$defs/EmbeddedResourceResource"
"$ref": "#/$defs/EmbeddedResourceResource",
"description": "Embedded resource payload, either text or binary data."
},
"_meta": {
"type": [
@ -4299,21 +4386,23 @@
"items": {
"type": "string"
},
"description": "**UNSTABLE**\n\nThis capability is not part of the spec yet, and may be removed or changed at any point.\n\nAuthoritative ordered additional workspace roots for this session. Each path must be absolute.\n\nWhen omitted or empty, there are no additional roots for the session."
"description": "Additional workspace roots reported for this session. Each path must be absolute.\n\nWhen present, this is the complete ordered additional-root list reported\nby the Agent. Omitted and empty values are equivalent: the response\nreports no additional roots."
},
"title": {
"type": [
"string",
"null"
],
"description": "Human-readable title for the session"
"description": "Human-readable title for the session",
"x-deserialize-default-on-error": {}
},
"updatedAt": {
"type": [
"string",
"null"
],
"description": "ISO 8601 timestamp of last activity"
"description": "ISO 8601 timestamp of last activity",
"x-deserialize-default-on-error": {}
},
"_meta": {
"type": [
@ -4984,7 +5073,8 @@
"type": "null"
}
],
"description": "Input for the command if required"
"description": "Input for the command if required",
"x-deserialize-default-on-error": {}
},
"_meta": {
"type": [

View file

@ -1,112 +0,0 @@
//! Shared adapter classes for converting mpsc channels to AsyncRead/AsyncWrite streams
//! Used by both HTTP and WebSocket transports
use std::{
pin::Pin,
task::{Context, Poll},
};
use tokio::sync::mpsc;
/// Converts an mpsc::Receiver<String> to AsyncRead
/// Each message is terminated with a newline for JSON-RPC framing
pub(crate) struct ReceiverToAsyncRead {
rx: mpsc::Receiver<String>,
buffer: Vec<u8>,
pos: usize,
}
impl ReceiverToAsyncRead {
pub(crate) fn new(rx: mpsc::Receiver<String>) -> Self {
Self {
rx,
buffer: Vec::new(),
pos: 0,
}
}
}
impl tokio::io::AsyncRead for ReceiverToAsyncRead {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
if self.pos < self.buffer.len() {
let remaining = &self.buffer[self.pos..];
let to_copy = remaining.len().min(buf.remaining());
buf.put_slice(&remaining[..to_copy]);
self.pos += to_copy;
if self.pos >= self.buffer.len() {
self.buffer.clear();
self.pos = 0;
}
return Poll::Ready(Ok(()));
}
match Pin::new(&mut self.rx).poll_recv(cx) {
Poll::Ready(Some(msg)) => {
let bytes = format!("{}\n", msg).into_bytes();
let to_copy = bytes.len().min(buf.remaining());
buf.put_slice(&bytes[..to_copy]);
if to_copy < bytes.len() {
self.buffer = bytes[to_copy..].to_vec();
self.pos = 0;
}
Poll::Ready(Ok(()))
}
Poll::Ready(None) => Poll::Ready(Ok(())),
Poll::Pending => Poll::Pending,
}
}
}
/// Converts an unbounded mpsc::Sender<String> to AsyncWrite.
/// Splits incoming data on newlines for JSON-RPC framing.
///
/// Uses an unbounded sender so that bursts of outgoing messages (e.g. replaying
/// a long session history) are never silently dropped due to backpressure.
pub(crate) struct SenderToAsyncWrite {
tx: mpsc::UnboundedSender<String>,
buffer: Vec<u8>,
}
impl SenderToAsyncWrite {
pub(crate) fn new(tx: mpsc::UnboundedSender<String>) -> Self {
Self {
tx,
buffer: Vec::new(),
}
}
}
impl tokio::io::AsyncWrite for SenderToAsyncWrite {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
self.buffer.extend_from_slice(buf);
while let Some(pos) = self.buffer.iter().position(|&b| b == b'\n') {
let line = String::from_utf8_lossy(&self.buffer[..pos]).to_string();
self.buffer.drain(..=pos);
if !line.is_empty() && self.tx.send(line).is_err() {
return Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"Channel closed",
)));
}
}
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
}

View file

@ -1,7 +1,7 @@
use std::str::FromStr;
use crate::permission::Permission;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
RequestPermissionResponse, SelectedPermissionOutcome,
};
@ -110,7 +110,7 @@ fn find_option(options: &[PermissionOption], kind: PermissionOptionKind) -> Opti
#[cfg(test)]
mod tests {
use super::*;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
PermissionOptionId, ToolCallId, ToolCallUpdate, ToolCallUpdateFields,
};
use test_case::test_case;

View file

@ -5,14 +5,14 @@ use crate::agents::platform_extensions::developer::edit::{
};
use crate::agents::platform_extensions::developer::shell::{ShellParams, OUTPUT_LIMIT_BYTES};
use crate::agents::platform_extensions::developer::DeveloperClient;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
CreateTerminalRequest, Diff, KillTerminalRequest, ReadTextFileRequest, ReleaseTerminalRequest,
SessionId, SessionNotification, SessionUpdate, Terminal, TerminalOutputRequest,
ToolCallContent, ToolCallId, ToolCallLocation, ToolCallUpdate, ToolCallUpdateFields, ToolKind,
WaitForTerminalExitRequest, WriteTextFileRequest,
};
use agent_client_protocol::{Client, ConnectionTo};
use agent_client_protocol_schema::TerminalId;
use agent_client_protocol_schema::v1::TerminalId;
use async_trait::async_trait;
use fs_err as fs;
use rmcp::model::{CallToolResult, Content as RmcpContent, Tool, ToolAnnotations};
@ -311,7 +311,7 @@ impl AcpTools {
&self,
terminal_id: &TerminalId,
timeout_secs: Option<u64>,
) -> Result<agent_client_protocol::schema::TerminalOutputResponse, McpError> {
) -> Result<agent_client_protocol::schema::v1::TerminalOutputResponse, McpError> {
let wait_fut = self
.cx
.send_request(WaitForTerminalExitRequest::new(

View file

@ -1,4 +1,3 @@
mod adapters;
mod common;
pub(crate) mod fs;
mod mcp_app_proxy;

View file

@ -1,16 +1,17 @@
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
ClientCapabilities, CloseSessionRequest, ContentBlock, ContentChunk, EnvVariable, HttpHeader,
ImageContent, InitializeRequest, InitializeResponse, McpCapabilities, McpServer, McpServerHttp,
McpServerStdio, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,
ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
SessionConfigKind, SessionConfigOption, SessionConfigOptionCategory,
SessionConfigSelectOptions, SessionId, SessionNotification, SessionUpdate,
SetSessionConfigOptionRequest, SetSessionModeRequest, SetSessionModeResponse, StopReason,
TextContent, ToolCallContent, ToolCallStatus, ToolKind,
};
use agent_client_protocol::schema::ProtocolVersion;
use agent_client_protocol::{Agent, Client, ConnectionTo};
use agent_client_protocol_schema::Usage as AcpUsage;
use agent_client_protocol_schema::AGENT_METHOD_NAMES;
use agent_client_protocol_schema::v1::Usage as AcpUsage;
use agent_client_protocol_schema::v1::AGENT_METHOD_NAMES;
use anyhow::{Context, Result};
use async_stream::try_stream;
use futures::future::BoxFuture;
@ -1118,7 +1119,7 @@ async fn handle_requests(
value,
response_tx,
} => {
let value_id = agent_client_protocol::schema::SessionConfigValueId::new(value);
let value_id = agent_client_protocol::schema::v1::SessionConfigValueId::new(value);
let req = SetSessionConfigOptionRequest::new(session_id, config_id, value_id);
let result: Result<()> = cx
.send_request(req)
@ -1184,7 +1185,7 @@ async fn apply_session_config_options(
session_id: SessionId,
) -> Result<()> {
for (config_id, value) in &config.session_config_options {
let value_id = agent_client_protocol::schema::SessionConfigValueId::new(value.clone());
let value_id = agent_client_protocol::schema::v1::SessionConfigValueId::new(value.clone());
cx.send_request(SetSessionConfigOptionRequest::new(
session_id.clone(),
config_id.clone(),
@ -1531,18 +1532,9 @@ fn resolve_model_info(
}
}
let models = response.models.as_ref().ok_or_else(|| {
ProviderError::RequestFailed(format!(
"{provider_name}: agent returned neither config_options nor models"
))
})?;
let current = models.current_model_id.0.to_string();
let available = models
.available_models
.iter()
.map(|am| am.model_id.0.to_string())
.collect();
Ok((current, available))
Err(ProviderError::RequestFailed(format!(
"{provider_name}: agent returned no model config_options"
)))
}
fn reverse_mode_mapping(
@ -1584,7 +1576,7 @@ fn permission_decision_from_mode(goose_mode: GooseMode) -> Option<PermissionDeci
mod tests {
use super::*;
use crate::agents::extension::Envs;
use agent_client_protocol::schema::SessionConfigSelectOption;
use agent_client_protocol::schema::v1::SessionConfigSelectOption;
use test_case::test_case;
fn prompt_text(block: &ContentBlock) -> &str {
@ -2009,14 +2001,6 @@ mod tests {
#[test_case(
NewSessionResponse::new("s1")
.models(agent_client_protocol::schema::SessionModelState::new(
"default",
vec![
agent_client_protocol::schema::ModelInfo::new("default", "Default (recommended)"),
agent_client_protocol::schema::ModelInfo::new("sonnet", "Sonnet"),
agent_client_protocol::schema::ModelInfo::new("haiku", "Haiku"),
],
))
.config_options(vec![
SessionConfigOption::select("model", "Model", "default", vec![
SessionConfigSelectOption::new("default", "Default (recommended)"),
@ -2026,27 +2010,27 @@ mod tests {
.category(SessionConfigOptionCategory::Model),
])
=> Ok(("default".to_string(), vec!["default".to_string(), "sonnet".to_string(), "haiku".to_string()]))
; "claude-agent-acp config_options supersedes models"
; "model is resolved from config_options"
)]
#[test_case(
NewSessionResponse::new("s1")
.models(agent_client_protocol::schema::SessionModelState::new(
"auto-gemini-3",
vec![
agent_client_protocol::schema::ModelInfo::new("auto-gemini-3", "Auto (Gemini 3)"),
agent_client_protocol::schema::ModelInfo::new("auto-gemini-2.5", "Auto (Gemini 2.5)"),
agent_client_protocol::schema::ModelInfo::new("gemini-2.5-pro", "gemini-2.5-pro"),
],
))
.config_options(vec![
SessionConfigOption::select("model", "Model", "auto-gemini-3", vec![
SessionConfigSelectOption::new("auto-gemini-3", "Auto (Gemini 3)"),
SessionConfigSelectOption::new("auto-gemini-2.5", "Auto (Gemini 2.5)"),
SessionConfigSelectOption::new("gemini-2.5-pro", "gemini-2.5-pro"),
])
.category(SessionConfigOptionCategory::Model),
])
=> Ok(("auto-gemini-3".to_string(), vec!["auto-gemini-3".to_string(), "auto-gemini-2.5".to_string(), "gemini-2.5-pro".to_string()]))
; "gemini falls back to models"
; "model with multiple options"
)]
#[test_case(
NewSessionResponse::new("s1")
=> Err(ProviderError::RequestFailed(
"test: agent returned neither config_options nor models".to_string()
"test: agent returned no model config_options".to_string()
))
; "neither config_options nor models is an error"
; "missing model config_options is an error"
)]
fn test_resolve_model_info(
response: NewSessionResponse,
@ -2098,18 +2082,19 @@ mod tests {
#[test]
fn acp_tool_call_content_handles_text_diff_terminal_and_image() {
use agent_client_protocol::schema::{Diff, Terminal, TerminalId, TextContent};
use agent_client_protocol::schema::v1::{Diff, Terminal, TerminalId, TextContent};
let diff_block = ToolCallContent::Diff(
Diff::new(std::path::PathBuf::from("/tmp/file.txt"), "new\n").old_text("old\n"),
);
let terminal_block = ToolCallContent::Terminal(Terminal::new(TerminalId::new("term-7")));
let text_block = ToolCallContent::Content(agent_client_protocol::schema::Content::new(
let text_block = ToolCallContent::Content(agent_client_protocol::schema::v1::Content::new(
ContentBlock::Text(TextContent::new("hello")),
));
let image_block = ToolCallContent::Content(agent_client_protocol::schema::Content::new(
ContentBlock::Image(ImageContent::new("base64data", "image/png")),
));
let image_block =
ToolCallContent::Content(agent_client_protocol::schema::v1::Content::new(
ContentBlock::Image(ImageContent::new("base64data", "image/png")),
));
let out = acp_tool_call_content_to_rmcp(
Some(vec![text_block, diff_block, terminal_block, image_block]),

View file

@ -3,11 +3,10 @@ use crate::config::{Config, GooseMode};
use crate::providers::inventory::{ProviderInventoryEntry, ProviderInventoryService};
use crate::session::Session;
use crate::slash_commands::types::{SlashCommandEntry, SlashCommandSource};
use agent_client_protocol::schema::{
AvailableCommand, AvailableCommandInput, AvailableCommandsUpdate, ModelId, ModelInfo,
SessionConfigOption, SessionConfigOptionCategory, SessionConfigSelectOption, SessionId,
SessionInfo, SessionMode, SessionModeId, SessionModeState, SessionModelState,
SessionNotification, SessionUpdate, UnstructuredCommandInput,
use agent_client_protocol::schema::v1::{
AvailableCommand, AvailableCommandInput, AvailableCommandsUpdate, SessionConfigOption,
SessionConfigOptionCategory, SessionConfigSelectOption, SessionId, SessionInfo, SessionMode,
SessionModeId, SessionModeState, SessionNotification, SessionUpdate, UnstructuredCommandInput,
};
use agent_client_protocol::{Client, ConnectionTo};
use goose_providers::model::ModelConfig;
@ -110,25 +109,51 @@ pub(super) fn build_session_info(session: Session) -> SessionInfo {
info
}
/// A model and its label, used to build the "model" session config option.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct ModelOption {
pub id: String,
pub name: String,
}
/// The currently selected model and the set of available models for a session.
///
/// Replaces the removed `SessionModelState` ACP schema type; goose now surfaces
/// model selection through the generic session config option API.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct ModelSelection {
pub current_model_id: String,
pub available_models: Vec<ModelOption>,
}
pub(super) fn build_model_state(
current_model: &str,
inventory: &ProviderInventoryEntry,
) -> SessionModelState {
) -> ModelSelection {
let mut available_models = inventory
.models
.iter()
.map(|model| ModelInfo::new(ModelId::new(model.id.as_str()), model.name.as_str()))
.map(|model| ModelOption {
id: model.id.clone(),
name: model.name.clone(),
})
.collect::<Vec<_>>();
if !available_models
.iter()
.any(|model| model.model_id.0.as_ref() == current_model)
.any(|model| model.id == current_model)
{
available_models.insert(
0,
ModelInfo::new(ModelId::new(current_model), current_model),
ModelOption {
id: current_model.to_string(),
name: current_model.to_string(),
},
);
}
SessionModelState::new(ModelId::new(current_model), available_models)
ModelSelection {
current_model_id: current_model.to_string(),
available_models,
}
}
struct ProviderOptionEntry {
@ -209,27 +234,20 @@ pub(super) fn build_mode_state(
pub(super) async fn build_session_setup_config(
provider_inventory: &ProviderInventoryService,
session: &Session,
) -> Result<
(
SessionModeState,
Option<SessionModelState>,
Option<Vec<SessionConfigOption>>,
),
agent_client_protocol::Error,
> {
) -> Result<(SessionModeState, Option<Vec<SessionConfigOption>>), agent_client_protocol::Error> {
let mode_state = build_mode_state(session.goose_mode)?;
let (Some(provider_name), Some(model_config)) = (
session.provider_name.as_deref(),
session.model_config.as_ref(),
) else {
return Ok((mode_state, None, None));
return Ok((mode_state, None));
};
let Some(inventory) = provider_inventory
.find_entry_for_provider(provider_name)
.await
else {
return Ok((mode_state, None, None));
return Ok((mode_state, None));
};
let model_state = build_model_state(model_config.model_name.as_str(), &inventory);
let provider_selection = session_provider_selection(session);
@ -241,12 +259,12 @@ pub(super) async fn build_session_setup_config(
provider_selection,
provider_options,
);
Ok((mode_state, Some(model_state), Some(config_options)))
Ok((mode_state, Some(config_options)))
}
pub(super) fn build_config_options(
mode_state: &SessionModeState,
model_state: &SessionModelState,
model_state: &ModelSelection,
model_config: &ModelConfig,
provider_selection: &str,
provider_options: Vec<SessionConfigSelectOption>,
@ -262,7 +280,7 @@ pub(super) fn build_config_options(
let model_options: Vec<SessionConfigSelectOption> = model_state
.available_models
.iter()
.map(|m| SessionConfigSelectOption::new(m.model_id.0.clone(), m.name.clone()))
.map(|m| SessionConfigSelectOption::new(m.id.clone(), m.name.clone()))
.collect();
let thinking_effort_options = thinking_effort_values(model_config)
.iter()
@ -289,7 +307,7 @@ pub(super) fn build_config_options(
SessionConfigOption::select(
"model",
"Model",
model_state.current_model_id.0.clone(),
model_state.current_model_id.clone(),
model_options,
)
.category(SessionConfigOptionCategory::Model),
@ -404,28 +422,33 @@ pub(super) fn send_session_setup_notifications(
#[cfg(test)]
mod tests {
use super::*;
use agent_client_protocol::schema::SessionConfigKind;
use agent_client_protocol::schema::v1::SessionConfigKind;
use test_case::test_case;
fn model_selection(current: &str, models: &[&str]) -> ModelSelection {
ModelSelection {
current_model_id: current.to_string(),
available_models: models
.iter()
.map(|m| ModelOption {
id: m.to_string(),
name: m.to_string(),
})
.collect(),
}
}
#[test_case(
vec!["model-a".into(), "model-b".into()]
=> SessionModelState::new(
ModelId::new("unused"),
vec![ModelInfo::new(ModelId::new("unused"), "unused"),
ModelInfo::new(ModelId::new("model-a"), "model-a"),
ModelInfo::new(ModelId::new("model-b"), "model-b")],
)
=> model_selection("unused", &["unused", "model-a", "model-b"])
; "returns current and available models"
)]
#[test_case(
vec![]
=> SessionModelState::new(
ModelId::new("unused"),
vec![ModelInfo::new(ModelId::new("unused"), "unused")],
)
=> model_selection("unused", &["unused"])
; "empty model list"
)]
fn test_build_model_state(models: Vec<String>) -> SessionModelState {
fn test_build_model_state(models: Vec<String>) -> ModelSelection {
let inventory = ProviderInventoryEntry {
provider_id: "mock".to_string(),
provider_name: "Mock".to_string(),
@ -547,10 +570,7 @@ mod tests {
SessionConfigSelectOption::new("anthropic", "anthropic"),
SessionConfigSelectOption::new("openai", "openai"),
],
SessionModelState::new(
ModelId::new("gpt-4"),
vec![ModelInfo::new(ModelId::new("gpt-4"), "gpt-4"), ModelInfo::new(ModelId::new("gpt-3.5"), "gpt-3.5")],
)
model_selection("gpt-4", &["gpt-4", "gpt-3.5"])
=> vec![
SessionConfigOption::select(
"provider", "Provider", "openai",
@ -588,7 +608,7 @@ mod tests {
build_mode_state(GooseMode::Approve).unwrap(),
"openai",
vec![SessionConfigSelectOption::new("openai", "openai")],
SessionModelState::new(ModelId::new("only-model"), vec![ModelInfo::new(ModelId::new("only-model"), "only-model")])
model_selection("only-model", &["only-model"])
=> vec![
SessionConfigOption::select(
"provider", "Provider", "openai",
@ -620,9 +640,9 @@ mod tests {
mode_state: SessionModeState,
provider_name: &'static str,
provider_options: Vec<SessionConfigSelectOption>,
model_state: SessionModelState,
model_state: ModelSelection,
) -> Vec<SessionConfigOption> {
let model_config = ModelConfig::new(model_state.current_model_id.0.as_ref())
let model_config = ModelConfig::new(model_state.current_model_id.as_str())
.with_merged_request_params(std::collections::HashMap::from([(
"thinking_effort".to_string(),
serde_json::json!("off"),
@ -639,13 +659,7 @@ mod tests {
#[test]
fn test_build_config_options_uses_current_thinking_effort() {
let mode_state = build_mode_state(GooseMode::Auto).unwrap();
let model_state = SessionModelState::new(
ModelId::new("claude-sonnet-4"),
vec![ModelInfo::new(
ModelId::new("claude-sonnet-4"),
"claude-sonnet-4",
)],
);
let model_state = model_selection("claude-sonnet-4", &["claude-sonnet-4"]);
let model_config = ModelConfig::new("claude-sonnet-4").with_merged_request_params(
std::collections::HashMap::from([(
"thinking_effort".to_string(),
@ -675,10 +689,7 @@ mod tests {
#[test]
fn test_build_config_options_masks_non_reasoning_thinking_effort() {
let mode_state = build_mode_state(GooseMode::Auto).unwrap();
let model_state = SessionModelState::new(
ModelId::new("gpt-4"),
vec![ModelInfo::new(ModelId::new("gpt-4"), "gpt-4")],
);
let model_state = model_selection("gpt-4", &["gpt-4"]);
let mut model_config =
ModelConfig::new("gpt-4").with_merged_request_params(std::collections::HashMap::from(
[("thinking_effort".to_string(), serde_json::json!("high"))],
@ -704,7 +715,7 @@ mod tests {
assert_eq!(select.current_value.0.as_ref(), "off");
assert_eq!(
select.options,
agent_client_protocol::schema::SessionConfigSelectOptions::Ungrouped(vec![
agent_client_protocol::schema::v1::SessionConfigSelectOptions::Ungrouped(vec![
SessionConfigSelectOption::new("off", "off")
])
);

View file

@ -39,7 +39,7 @@ use crate::session::{
};
use crate::source_roots::SourceRoot;
use crate::utils::sanitize_unicode_tags;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
AgentCapabilities, Annotations, AuthMethod, AuthMethodAgent, AuthenticateRequest,
AuthenticateResponse, BlobResourceContents, CancelNotification, CloseSessionRequest,
CloseSessionResponse, ConfigOptionUpdate, Content, ContentBlock, ContentChunk, Cost,
@ -51,10 +51,9 @@ use agent_client_protocol::schema::{
RequestPermissionOutcome, RequestPermissionRequest, ResourceLink, SessionCapabilities,
SessionCloseCapabilities, SessionConfigOption, SessionId, SessionInfoUpdate,
SessionListCapabilities, SessionNotification, SessionUpdate, SetSessionConfigOptionRequest,
SetSessionConfigOptionResponse, SetSessionModeRequest, SetSessionModeResponse,
SetSessionModelRequest, SetSessionModelResponse, StopReason, TextContent, TextResourceContents,
ToolCall, ToolCallContent, ToolCallId, ToolCallLocation, ToolCallStatus, ToolCallUpdate,
ToolCallUpdateFields, ToolKind, Usage, UsageUpdate,
SetSessionConfigOptionResponse, SetSessionModeRequest, SetSessionModeResponse, StopReason,
TextContent, TextResourceContents, ToolCall, ToolCallContent, ToolCallId, ToolCallLocation,
ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields, ToolKind, Usage, UsageUpdate,
};
use agent_client_protocol::util::MatchDispatchFrom;
use agent_client_protocol::{
@ -1259,10 +1258,10 @@ impl GooseAcpAgent {
roles
.iter()
.filter_map(|r| match r {
agent_client_protocol::schema::Role::Assistant => {
agent_client_protocol::schema::v1::Role::Assistant => {
Some(Role::Assistant)
}
agent_client_protocol::schema::Role::User => {
agent_client_protocol::schema::v1::Role::User => {
Some(Role::User)
}
_ => None,
@ -2759,7 +2758,7 @@ impl GooseAcpAgent {
&self,
session_id: &str,
model_id: &str,
) -> Result<SetSessionModelResponse, agent_client_protocol::Error> {
) -> Result<(), agent_client_protocol::Error> {
let agent = self.get_session_agent(session_id).await?;
let current_provider = agent
.provider()
@ -2784,7 +2783,7 @@ impl GooseAcpAgent {
.await
.internal_err_ctx("Failed to recreate provider")?;
// model_config is already updated on the session by the agent's update_provider call.
Ok(SetSessionModelResponse::new())
Ok(())
}
async fn build_config_update(
@ -3001,6 +3000,38 @@ where
})
}
/// A lazily-initialized agent connection used by the HTTP/WebSocket transport.
///
/// The `agent-client-protocol-http` server takes a synchronous factory that
/// yields a [`ConnectTo<Client>`] per connection, but creating a goose agent is
/// async. Agent creation is therefore deferred into [`ConnectTo::connect_to`],
/// which runs as the connection's serving future.
pub struct GooseAgentConnection {
server: Arc<crate::acp::server_factory::AcpServer>,
}
impl GooseAgentConnection {
pub fn new(server: Arc<crate::acp::server_factory::AcpServer>) -> Self {
Self { server }
}
}
impl agent_client_protocol::ConnectTo<Client> for GooseAgentConnection {
async fn connect_to(
self,
client: impl agent_client_protocol::ConnectTo<SacpAgent>,
) -> std::result::Result<(), agent_client_protocol::Error> {
let agent = self.server.create_agent().await.internal_err()?;
let handler = GooseAcpHandler { agent };
SacpAgent
.builder()
.name("goose-acp")
.with_handler(handler)
.connect_to(client)
.await
}
}
pub async fn run(builtins: Vec<String>) -> Result<()> {
info!("listening on stdio");
@ -3026,7 +3057,7 @@ mod tests {
use super::*;
use crate::conversation::message::{ToolRequest, ToolResponse};
use crate::session::session_manager::SessionType;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
EnvVariable, HttpHeader, McpServer, McpServerHttp, McpServerSse, McpServerStdio,
PermissionOptionId, ResourceLink, SelectedPermissionOutcome,
};
@ -3872,7 +3903,7 @@ print(\"hello, world\")
let request =
InitializeRequest::new(agent_client_protocol::schema::ProtocolVersion::LATEST)
.client_capabilities(
agent_client_protocol::schema::ClientCapabilities::new().meta(meta),
agent_client_protocol::schema::v1::ClientCapabilities::new().meta(meta),
);
let goose_client_capabilities =
extract_client_capabilities_meta(&request).and_then(|meta| meta.goose);

View file

@ -95,7 +95,7 @@ impl HandleDispatchFrom<Client> for GooseAcpHandler {
Ok(())
})
.await
// set_config_option (SACP 11) and legacy set_mode/set_model; custom _goose/* in otherwise.
// set_config_option (SACP 11) and set_mode; custom _goose/* in otherwise.
.if_request({
let agent = agent.clone();
let cx = cx.clone();
@ -320,28 +320,6 @@ impl HandleDispatchFrom<Client> for GooseAcpHandler {
}
})
.await
.if_request({
let agent = agent.clone();
let cx = cx.clone();
|req: SetSessionModelRequest, responder: Responder<SetSessionModelResponse>| async move {
let cx_spawn = cx.clone();
cx.spawn(async move {
let cx = cx_spawn;
let session_id = req.session_id.clone();
match agent.on_set_model(&session_id.0, &req.model_id.0).await {
Ok(resp) => {
let (notification, _) = agent.build_config_update(&session_id).await?;
cx.send_notification(notification)?;
responder.respond(resp)?;
}
Err(e) => responder.respond_with_error(e)?,
}
Ok(())
})?;
Ok(())
}
})
.await
.if_request({
let agent = agent.clone();
let cx = cx.clone();

View file

@ -1,6 +1,6 @@
use std::sync::Arc;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
CreateElicitationRequest, CreateElicitationResponse, ElicitationAction as AcpElicitationAction,
ElicitationFormMode, ElicitationSchema, ElicitationSessionScope, Meta, SessionId,
CLIENT_METHOD_NAMES,
@ -204,7 +204,7 @@ impl JsonRpcResponse for CreateElicitationResponseMessage {
}
pub(super) fn client_supports_form_elicitation(
args: &agent_client_protocol::schema::InitializeRequest,
args: &agent_client_protocol::schema::v1::InitializeRequest,
) -> bool {
args.client_capabilities
.elicitation

View file

@ -1,7 +1,7 @@
use super::*;
use crate::agents::extension::Envs;
use crate::config::extensions::ExtensionEntry;
use agent_client_protocol::schema::{HttpHeader, McpServer, McpServerHttp, McpServerStdio};
use agent_client_protocol::schema::v1::{HttpHeader, McpServer, McpServerHttp, McpServerStdio};
impl GooseAcpAgent {
pub(super) async fn on_add_session_extension(
@ -362,7 +362,7 @@ fn empty_string_to_none(value: &str) -> Option<String> {
mod tests {
use super::*;
use crate::agents::extension::Envs;
use agent_client_protocol::schema::{McpServer, McpServerSse};
use agent_client_protocol::schema::v1::{McpServer, McpServerSse};
use std::collections::HashMap;
#[test]
@ -641,8 +641,11 @@ mod tests {
fn goose_mcp_stdio_extension_extracts_literal_envs_for_config_add() {
let extension = GooseExtension::Mcp {
server: McpServer::Stdio(McpServerStdio::new("test-stdio", "test-command").env(vec![
agent_client_protocol::schema::EnvVariable::new("SECRET_TOKEN", "literal-secret"),
agent_client_protocol::schema::EnvVariable::new("OTHER_TOKEN", "other-secret"),
agent_client_protocol::schema::v1::EnvVariable::new(
"SECRET_TOKEN",
"literal-secret",
),
agent_client_protocol::schema::v1::EnvVariable::new("OTHER_TOKEN", "other-secret"),
])),
env_keys: vec!["SECRET_TOKEN".to_string()],
description: Some("Test stdio".to_string()),

View file

@ -62,16 +62,13 @@ impl GooseAcpAgent {
meta.insert("extensionResults".to_string(), v);
}
let (mode_state, model_state, config_options) =
let (mode_state, config_options) =
build_session_setup_config(&self.provider_inventory, &goose_session).await?;
let mut response = ForkSessionResponse::new(acp_session_id.clone())
.modes(mode_state)
.meta(meta);
if let Some(ms) = model_state {
response = response.models(ms);
}
if let Some(co) = config_options {
response = response.config_options(co);
}

View file

@ -2,7 +2,9 @@ use super::{build_session_info, meta_string, GooseAcpAgent, ResultExt};
use crate::session::session_manager::{
SessionListCursor, SessionListFilters, SessionListPageQuery, SessionType,
};
use agent_client_protocol::schema::{ListSessionsRequest, ListSessionsResponse, Meta, SessionInfo};
use agent_client_protocol::schema::v1::{
ListSessionsRequest, ListSessionsResponse, Meta, SessionInfo,
};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

View file

@ -5,8 +5,8 @@ fn replay_audience_annotations(audience: &[Role]) -> Annotations {
audience
.iter()
.map(|role| match role {
Role::Assistant => agent_client_protocol::schema::Role::Assistant,
Role::User => agent_client_protocol::schema::Role::User,
Role::Assistant => agent_client_protocol::schema::v1::Role::Assistant,
Role::User => agent_client_protocol::schema::v1::Role::User,
})
.collect::<Vec<_>>(),
)
@ -206,15 +206,12 @@ impl GooseAcpAgent {
.update_working_dir(&session.working_dir)
.await;
let (mode_state, model_state, config_options) =
let (mode_state, config_options) =
build_session_setup_config(&self.provider_inventory, &session).await?;
send_session_setup_notifications(cx, &session, self.supports_goose_custom_notifications())?;
let mut response = LoadSessionResponse::new().modes(mode_state);
if let Some(ms) = model_state {
response = response.models(ms);
}
if let Some(co) = config_options {
response = response.config_options(co);
}

View file

@ -6,7 +6,7 @@ use crate::recipe::{Recipe, Settings};
use crate::session::{ExtensionData, Session, SessionType};
use super::GooseAcpAgent;
use agent_client_protocol::schema::{Meta, NewSessionRequest, NewSessionResponse, SessionId};
use agent_client_protocol::schema::v1::{Meta, NewSessionRequest, NewSessionResponse, SessionId};
use agent_client_protocol::{Client, ConnectionTo};
use goose_providers::model::ModelConfig;
use std::collections::HashMap;
@ -219,14 +219,11 @@ impl GooseAcpAgent {
session: &Session,
extension_results: &[ExtensionLoadResult],
) -> Result<NewSessionResponse, agent_client_protocol::Error> {
let (mode_state, model_state, config_options) =
let (mode_state, config_options) =
super::build_session_setup_config(&self.provider_inventory, session).await?;
let mut response =
NewSessionResponse::new(SessionId::new(session.id.clone())).modes(mode_state);
if let Some(ms) = model_state {
response = response.models(ms);
}
if let Some(co) = config_options {
response = response.config_options(co);
}

View file

@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use agent_client_protocol::schema::Meta;
use agent_client_protocol::schema::v1::Meta;
use agent_client_protocol::{
Client, ConnectionTo, JsonRpcMessage, JsonRpcRequest, JsonRpcResponse, UntypedMessage,
};

View file

@ -1,4 +1,4 @@
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
Meta, SessionUpdate, ToolCallId, ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields,
};
use rmcp::model::{LoggingMessageNotificationParam, ProgressNotificationParam, ServerNotification};

View file

@ -1,425 +0,0 @@
//! Per-connection state. Server→client messages fan out to a connection-scoped
//! stream, a per-session stream for each active `sessionId`, and an
//! all-outbound stream consumed by WebSocket.
use std::{
collections::{HashMap, VecDeque},
sync::Arc,
};
use anyhow::Result;
use serde_json::Value;
use tokio::sync::{broadcast, mpsc, Mutex, RwLock};
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use tracing::{error, info, trace, warn};
use crate::acp::adapters::{ReceiverToAsyncRead, SenderToAsyncWrite};
use crate::acp::server_factory::AcpServer;
const OUTBOUND_BROADCAST_CAPACITY: usize = 1024;
/// Buffers messages emitted before a subscriber attaches (e.g. session
/// notifications that land before the client opens the session GET stream).
const PRE_SUBSCRIBE_BUFFER_CAPACITY: usize = 1024;
#[derive(Clone, Debug)]
pub(crate) enum ResponseRoute {
Connection,
Session(String),
}
struct OutboundStream {
tx: broadcast::Sender<String>,
pre_subscribe_buffer: Mutex<Option<VecDeque<String>>>,
}
impl OutboundStream {
fn new() -> Self {
let (tx, _) = broadcast::channel(OUTBOUND_BROADCAST_CAPACITY);
Self {
tx,
pre_subscribe_buffer: Mutex::new(Some(VecDeque::new())),
}
}
async fn push(&self, msg: String) {
let mut guard = self.pre_subscribe_buffer.lock().await;
match guard.as_mut() {
Some(buf) => {
if buf.len() >= PRE_SUBSCRIBE_BUFFER_CAPACITY {
warn!(
"Pre-subscribe buffer full ({} messages); dropping oldest",
PRE_SUBSCRIBE_BUFFER_CAPACITY
);
buf.pop_front();
}
buf.push_back(msg);
}
None => {
drop(guard);
let _ = self.tx.send(msg);
}
}
}
async fn subscribe_with_replay(&self) -> (Vec<String>, broadcast::Receiver<String>) {
let mut guard = self.pre_subscribe_buffer.lock().await;
let receiver = self.tx.subscribe();
let replay = guard.take().map(Vec::from).unwrap_or_default();
(replay, receiver)
}
}
pub(crate) struct Connection {
pub to_agent_tx: mpsc::Sender<String>,
/// Consumed once by `handle_initialize` to read the synchronous initialize
/// response before the router task takes over.
pub init_receiver: Mutex<Option<mpsc::UnboundedReceiver<String>>>,
pub init_complete: Mutex<bool>,
pub agent_handle: tokio::task::JoinHandle<()>,
pub router_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
connection_stream: Arc<OutboundStream>,
session_streams: Arc<RwLock<HashMap<String, Arc<OutboundStream>>>>,
all_outbound: Arc<OutboundStream>,
pending_routes: Arc<Mutex<HashMap<Value, ResponseRoute>>>,
}
pub(crate) struct ConnectionRegistry {
pub server: Arc<AcpServer>,
connections: RwLock<HashMap<String, Arc<Connection>>>,
}
impl ConnectionRegistry {
pub fn new(server: Arc<AcpServer>) -> Self {
Self {
server,
connections: RwLock::new(HashMap::new()),
}
}
pub async fn create_connection(&self) -> Result<(String, Arc<Connection>)> {
let (to_agent_tx, to_agent_rx) = mpsc::channel::<String>(256);
let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
let agent = self.server.create_agent().await?;
let connection_id = uuid::Uuid::new_v4().to_string();
let read_stream = ReceiverToAsyncRead::new(to_agent_rx);
let write_stream = SenderToAsyncWrite::new(from_agent_tx);
let fut =
crate::acp::server::serve(agent, read_stream.compat(), write_stream.compat_write());
let conn_id_for_task = connection_id.clone();
let agent_handle = tokio::spawn(async move {
if let Err(e) = fut.await {
error!(connection_id = %conn_id_for_task, "ACP agent task error: {}", e);
}
});
let connection = Arc::new(Connection {
to_agent_tx,
init_receiver: Mutex::new(Some(from_agent_rx)),
init_complete: Mutex::new(false),
agent_handle,
router_handle: Mutex::new(None),
connection_stream: Arc::new(OutboundStream::new()),
session_streams: Arc::new(RwLock::new(HashMap::new())),
all_outbound: Arc::new(OutboundStream::new()),
pending_routes: Arc::new(Mutex::new(HashMap::new())),
});
self.connections
.write()
.await
.insert(connection_id.clone(), connection.clone());
info!(connection_id = %connection_id, "Connection created");
Ok((connection_id, connection))
}
pub async fn get(&self, connection_id: &str) -> Option<Arc<Connection>> {
self.connections.read().await.get(connection_id).cloned()
}
pub async fn remove(&self, connection_id: &str) -> Option<Arc<Connection>> {
self.connections.write().await.remove(connection_id)
}
}
impl Connection {
pub async fn start_router(self: &Arc<Self>) {
let mut complete = self.init_complete.lock().await;
if *complete {
return;
}
let Some(mut rx) = self.init_receiver.lock().await.take() else {
return;
};
let me = self.clone();
let handle = tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
me.route_outbound(msg).await;
}
});
*self.router_handle.lock().await = Some(handle);
*complete = true;
}
async fn route_outbound(self: &Arc<Self>, msg: String) {
self.all_outbound.push(msg.clone()).await;
let parsed: Option<Value> = serde_json::from_str(&msg).ok();
let target = match parsed.as_ref() {
Some(v) => self.classify(v).await,
None => Target::Connection,
};
match target {
Target::Connection => {
trace!(target = "connection", "→ connection-scoped stream");
self.connection_stream.push(msg).await;
}
Target::Session(sid) => {
trace!(target = %sid, "→ session-scoped stream");
let stream = self.get_or_create_session_stream(&sid).await;
stream.push(msg).await;
}
}
}
async fn classify(self: &Arc<Self>, v: &Value) -> Target {
let has_method = v.get("method").is_some();
let has_id = v.get("id").is_some();
let has_result_or_error = v.get("result").is_some() || v.get("error").is_some();
if has_method {
if let Some(sid) = extract_session_id_from_params(v) {
return Target::Session(sid);
}
return Target::Connection;
}
if has_id && has_result_or_error {
let id = v.get("id").cloned().unwrap_or(Value::Null);
let route = self.pending_routes.lock().await.remove(&id);
return match route {
Some(ResponseRoute::Session(sid)) => Target::Session(sid),
Some(ResponseRoute::Connection) | None => Target::Connection,
};
}
Target::Connection
}
pub async fn record_pending_route(&self, id: Value, route: ResponseRoute) {
if id.is_null() {
return;
}
self.pending_routes.lock().await.insert(id, route);
}
pub async fn subscribe_connection_stream(&self) -> (Vec<String>, broadcast::Receiver<String>) {
self.connection_stream.subscribe_with_replay().await
}
pub async fn subscribe_session_stream(
&self,
session_id: &str,
) -> Option<(Vec<String>, broadcast::Receiver<String>)> {
let stream = self.session_streams.read().await.get(session_id).cloned()?;
Some(stream.subscribe_with_replay().await)
}
pub async fn ensure_session(&self, session_id: &str) {
self.get_or_create_session_stream(session_id).await;
}
async fn get_or_create_session_stream(&self, session_id: &str) -> Arc<OutboundStream> {
if let Some(s) = self.session_streams.read().await.get(session_id) {
return s.clone();
}
let mut w = self.session_streams.write().await;
w.entry(session_id.to_string())
.or_insert_with(|| Arc::new(OutboundStream::new()))
.clone()
}
pub async fn subscribe_all_outbound(&self) -> (Vec<String>, broadcast::Receiver<String>) {
self.all_outbound.subscribe_with_replay().await
}
pub async fn shutdown(&self) {
self.agent_handle.abort();
if let Some(h) = self.router_handle.lock().await.take() {
h.abort();
}
}
}
#[derive(Debug)]
enum Target {
Connection,
Session(String),
}
fn extract_session_id_from_params(v: &Value) -> Option<String> {
v.get("params")
.and_then(|p| p.get("sessionId"))
.and_then(|s| s.as_str())
.map(|s| s.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
use tokio::time::timeout;
fn fake_connection() -> (Arc<Connection>, mpsc::UnboundedSender<String>) {
let (to_agent_tx, _to_agent_rx) = mpsc::channel::<String>(256);
let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
let agent_handle = tokio::spawn(async {
std::future::pending::<()>().await;
});
let connection = Arc::new(Connection {
to_agent_tx,
init_receiver: Mutex::new(Some(from_agent_rx)),
init_complete: Mutex::new(false),
agent_handle,
router_handle: Mutex::new(None),
connection_stream: Arc::new(OutboundStream::new()),
session_streams: Arc::new(RwLock::new(HashMap::new())),
all_outbound: Arc::new(OutboundStream::new()),
pending_routes: Arc::new(Mutex::new(HashMap::new())),
});
(connection, from_agent_tx)
}
#[tokio::test]
async fn buffers_connection_scoped_messages_before_first_subscribe() {
let (conn, agent_tx) = fake_connection();
conn.start_router().await;
agent_tx
.send(r#"{"id":1,"result":{"capabilities":{}}}"#.to_string())
.unwrap();
tokio::time::sleep(Duration::from_millis(20)).await;
let (replay, _rx) = conn.subscribe_connection_stream().await;
assert_eq!(replay.len(), 1);
assert!(replay[0].contains("\"capabilities\""));
conn.shutdown().await;
}
#[tokio::test]
async fn routes_session_scoped_notification_to_session_stream() {
let (conn, agent_tx) = fake_connection();
conn.start_router().await;
conn.ensure_session("sess_abc").await;
let (_, mut rx) = conn.subscribe_session_stream("sess_abc").await.unwrap();
agent_tx
.send(
r#"{"method":"session/update","params":{"sessionId":"sess_abc","update":{}}}"#
.to_string(),
)
.unwrap();
let got = timeout(Duration::from_secs(1), rx.recv())
.await
.unwrap()
.unwrap();
assert!(got.contains("session/update"));
let (replay, _) = conn.subscribe_connection_stream().await;
assert!(
replay.is_empty(),
"connection stream should not have session-scoped messages"
);
conn.shutdown().await;
}
#[tokio::test]
async fn routes_response_using_pending_route_table() {
let (conn, agent_tx) = fake_connection();
conn.start_router().await;
conn.ensure_session("sess_xyz").await;
conn.record_pending_route(
Value::from(42),
ResponseRoute::Session("sess_xyz".to_string()),
)
.await;
let (_, mut rx) = conn.subscribe_session_stream("sess_xyz").await.unwrap();
agent_tx
.send(r#"{"id":42,"result":{"stopReason":"end_turn"}}"#.to_string())
.unwrap();
let got = timeout(Duration::from_secs(1), rx.recv())
.await
.unwrap()
.unwrap();
assert!(got.contains("\"stopReason\""));
conn.shutdown().await;
}
#[tokio::test]
async fn websocket_all_outbound_sees_everything() {
let (conn, agent_tx) = fake_connection();
conn.start_router().await;
agent_tx
.send(r#"{"id":1,"result":{}}"#.to_string())
.unwrap();
agent_tx
.send(r#"{"method":"session/update","params":{"sessionId":"s1"}}"#.to_string())
.unwrap();
tokio::time::sleep(Duration::from_millis(20)).await;
let (replay, _all_rx) = conn.subscribe_all_outbound().await;
assert_eq!(replay.len(), 2);
assert!(replay[0].contains("\"id\":1"));
assert!(replay[1].contains("session/update"));
conn.shutdown().await;
}
#[tokio::test]
async fn unknown_session_subscribe_returns_none() {
let (conn, _agent_tx) = fake_connection();
conn.start_router().await;
assert!(conn.subscribe_session_stream("nope").await.is_none());
conn.shutdown().await;
}
#[tokio::test]
async fn pre_subscribe_buffer_is_bounded() {
let (conn, agent_tx) = fake_connection();
conn.start_router().await;
for i in 0..(PRE_SUBSCRIBE_BUFFER_CAPACITY + 50) {
agent_tx
.send(format!(r#"{{"id":{},"result":{{}}}}"#, i))
.unwrap();
}
tokio::time::sleep(Duration::from_millis(50)).await;
let (replay, _rx) = conn.subscribe_connection_stream().await;
assert_eq!(replay.len(), PRE_SUBSCRIBE_BUFFER_CAPACITY);
}
}

View file

@ -1,282 +0,0 @@
use std::{convert::Infallible, sync::Arc, time::Duration};
use axum::{
body::Body,
extract::State,
http::{HeaderValue, Request, StatusCode},
response::{IntoResponse, Response, Sse},
};
use http_body_util::BodyExt;
use serde_json::Value;
use tokio::sync::broadcast;
use tracing::{debug, error, info, trace};
use super::connection::{Connection, ConnectionRegistry, ResponseRoute};
use super::*;
pub(crate) async fn handle_post(
State(registry): State<Arc<ConnectionRegistry>>,
request: Request<Body>,
) -> Response {
if !content_type_is_json(&request) {
return (
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"Unsupported Media Type: Content-Type must be application/json",
)
.into_response();
}
let connection_id = header_value(&request, HEADER_CONNECTION_ID);
let session_id = header_value(&request, HEADER_SESSION_ID);
let body_bytes = match request.into_body().collect().await {
Ok(collected) => collected.to_bytes(),
Err(e) => {
error!("Failed to read request body: {}", e);
return (StatusCode::BAD_REQUEST, "Failed to read request body").into_response();
}
};
let json_message: Value = match serde_json::from_slice(&body_bytes) {
Ok(v) => v,
Err(e) => {
return (StatusCode::BAD_REQUEST, format!("Invalid JSON: {}", e)).into_response();
}
};
if json_message.is_array() {
return (
StatusCode::NOT_IMPLEMENTED,
"Batch requests are not supported",
)
.into_response();
}
if is_initialize_request(&json_message) {
return handle_initialize(registry, json_message).await;
}
let Some(connection_id) = connection_id else {
return (
StatusCode::BAD_REQUEST,
"Bad Request: Acp-Connection-Id header required",
)
.into_response();
};
let Some(connection) = registry.get(&connection_id).await else {
return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response();
};
if let Some(method) = json_message.get("method").and_then(|m| m.as_str()) {
if method_requires_session_header(method) && session_id.is_none() {
return (
StatusCode::BAD_REQUEST,
"Bad Request: Acp-Session-Id header required for session-scoped methods",
)
.into_response();
}
}
if !is_jsonrpc_request_with_id(&json_message)
&& !is_jsonrpc_notification(&json_message)
&& !is_jsonrpc_response(&json_message)
{
return (StatusCode::BAD_REQUEST, "Invalid JSON-RPC message").into_response();
}
if let Some(sid) = session_id.as_deref() {
connection.ensure_session(sid).await;
}
if is_jsonrpc_request_with_id(&json_message) {
if let Some(id) = json_message.get("id") {
let route = match session_id.as_deref() {
Some(sid) => ResponseRoute::Session(sid.to_string()),
None => ResponseRoute::Connection,
};
connection.record_pending_route(id.clone(), route).await;
}
}
let message_str = serde_json::to_string(&json_message).unwrap();
trace!(connection_id = %connection_id, payload = %message_str, "POST → agent");
if connection.to_agent_tx.send(message_str).await.is_err() {
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to forward message to agent",
)
.into_response();
}
StatusCode::ACCEPTED.into_response()
}
async fn handle_initialize(registry: Arc<ConnectionRegistry>, json_message: Value) -> Response {
let (connection_id, connection) = match registry.create_connection().await {
Ok(pair) => pair,
Err(e) => {
error!("Failed to create connection: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to create connection",
)
.into_response();
}
};
let message_str = serde_json::to_string(&json_message).unwrap();
trace!(connection_id = %connection_id, payload = %message_str, "initialize → agent");
if connection.to_agent_tx.send(message_str).await.is_err() {
registry.remove(&connection_id).await;
connection.shutdown().await;
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to forward initialize to agent",
)
.into_response();
}
let init_response = {
let mut guard = connection.init_receiver.lock().await;
let Some(rx) = guard.as_mut() else {
registry.remove(&connection_id).await;
connection.shutdown().await;
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Initialize receiver already consumed",
)
.into_response();
};
rx.recv().await
};
let init_response = match init_response {
Some(msg) => msg,
None => {
registry.remove(&connection_id).await;
connection.shutdown().await;
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Agent closed before initialize response",
)
.into_response();
}
};
connection.start_router().await;
let mut response = (
StatusCode::OK,
[(axum::http::header::CONTENT_TYPE, JSON_MIME_TYPE)],
init_response,
)
.into_response();
if let Ok(v) = HeaderValue::from_str(&connection_id) {
response.headers_mut().insert(HEADER_CONNECTION_ID, v);
}
info!(connection_id = %connection_id, "Initialize complete");
response
}
pub(crate) async fn handle_get(
registry: Arc<ConnectionRegistry>,
request: Request<Body>,
) -> Response {
if !accepts_mime_type(&request, EVENT_STREAM_MIME_TYPE) {
return (
StatusCode::NOT_ACCEPTABLE,
"Not Acceptable: Client must accept text/event-stream",
)
.into_response();
}
let Some(connection_id) = header_value(&request, HEADER_CONNECTION_ID) else {
return (
StatusCode::BAD_REQUEST,
"Bad Request: Acp-Connection-Id header required",
)
.into_response();
};
let Some(connection) = registry.get(&connection_id).await else {
return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response();
};
let session_id = header_value(&request, HEADER_SESSION_ID);
let (replay, receiver) = match session_id.as_deref() {
Some(sid) => {
connection.ensure_session(sid).await;
connection
.subscribe_session_stream(sid)
.await
.expect("session stream exists after ensure_session")
}
None => connection.subscribe_connection_stream().await,
};
let sse = build_sse_stream(connection.clone(), replay, receiver);
let mut response = sse.into_response();
if let Ok(v) = HeaderValue::from_str(&connection_id) {
response.headers_mut().insert(HEADER_CONNECTION_ID, v);
}
if let Some(sid) = session_id {
if let Ok(v) = HeaderValue::from_str(&sid) {
response.headers_mut().insert(HEADER_SESSION_ID, v);
}
}
response
}
fn build_sse_stream(
_connection: Arc<Connection>,
replay: Vec<String>,
mut receiver: broadcast::Receiver<String>,
) -> Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, Infallible>>> {
let stream = async_stream::stream! {
for msg in replay {
trace!(payload = %msg, "SSE → client (replay)");
yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg));
}
loop {
match receiver.recv().await {
Ok(msg) => {
trace!(payload = %msg, "SSE → client");
yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg));
}
Err(broadcast::error::RecvError::Lagged(n)) => {
debug!("SSE subscriber lagged {} messages", n);
continue;
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
};
Sse::new(stream).keep_alive(
axum::response::sse::KeepAlive::new()
.interval(Duration::from_secs(15))
.text(""),
)
}
pub(crate) async fn handle_delete(
State(registry): State<Arc<ConnectionRegistry>>,
request: Request<Body>,
) -> Response {
let Some(connection_id) = header_value(&request, HEADER_CONNECTION_ID) else {
return (
StatusCode::BAD_REQUEST,
"Bad Request: Acp-Connection-Id header required",
)
.into_response();
};
let Some(connection) = registry.remove(&connection_id).await else {
return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response();
};
connection.shutdown().await;
info!(connection_id = %connection_id, "Connection terminated via DELETE");
StatusCode::ACCEPTED.into_response()
}

View file

@ -1,101 +1,30 @@
pub mod auth;
pub mod connection;
pub mod http;
pub mod websocket;
use std::sync::Arc;
use agent_client_protocol_http::{AcpHttpServer, CorsOptions, ServerOptions};
use axum::{
body::Body,
extract::{
ws::{rejection::WebSocketUpgradeRejection, WebSocketUpgrade},
State,
},
http::{header, HeaderName, Method, Request},
response::Response,
routing::{delete, get, post},
http::{header, HeaderName, Method},
routing::get,
Router,
};
use serde_json::Value;
use tower_http::cors::{Any, CorsLayer};
use crate::acp::server::GooseAgentConnection;
use crate::acp::server_factory::AcpServer;
pub(crate) const HEADER_CONNECTION_ID: &str = "Acp-Connection-Id";
pub(crate) const HEADER_SESSION_ID: &str = "Acp-Session-Id";
pub(crate) const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream";
pub(crate) const JSON_MIME_TYPE: &str = "application/json";
pub(crate) fn accepts_mime_type(request: &Request<Body>, mime_type: &str) -> bool {
request
.headers()
.get(axum::http::header::ACCEPT)
.and_then(|v| v.to_str().ok())
.is_some_and(|accept| accept.contains(mime_type))
}
pub(crate) fn content_type_is_json(request: &Request<Body>) -> bool {
request
.headers()
.get(axum::http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.is_some_and(|ct| ct.starts_with(JSON_MIME_TYPE))
}
pub(crate) fn header_value(request: &Request<Body>, name: &str) -> Option<String> {
request
.headers()
.get(name)
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string())
}
pub(crate) fn is_jsonrpc_request_with_id(value: &Value) -> bool {
value.get("method").is_some() && value.get("id").is_some()
}
pub(crate) fn is_jsonrpc_notification(value: &Value) -> bool {
value.get("method").is_some() && value.get("id").is_none()
}
pub(crate) fn is_jsonrpc_response(value: &Value) -> bool {
value.get("id").is_some()
&& value.get("method").is_none()
&& (value.get("result").is_some() || value.get("error").is_some())
}
pub(crate) fn is_initialize_request(value: &Value) -> bool {
value.get("method").is_some_and(|m| m == "initialize") && value.get("id").is_some()
}
/// Methods that are scoped to a session and require an Acp-Session-Id header.
pub(crate) fn method_requires_session_header(method: &str) -> bool {
matches!(
method,
"session/prompt"
| "session/cancel"
| "session/load"
| "session/set_mode"
| "session/set_model"
)
}
async fn handle_get(
ws_upgrade: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
State(state): State<Arc<connection::ConnectionRegistry>>,
request: Request<Body>,
) -> Response {
match ws_upgrade {
Ok(ws) => websocket::handle_ws_upgrade(state, ws).await,
Err(_) => http::handle_get(state, request).await,
fn acp_http_options() -> ServerOptions {
ServerOptions {
path: "/acp".to_string(),
cors: CorsOptions::allow_any_origin(),
health_endpoint: false,
}
}
async fn health() -> &'static str {
"ok"
}
fn acp_cors_layer() -> CorsLayer {
/// CORS for the auxiliary routes (`/health`, `/status`, MCP app proxy) served by
/// `goose serve`. The ACP routes get their CORS from `AcpHttpServer`; this also
/// allows the `x-secret-key` auth header the proxy routes rely on.
fn aux_cors_layer() -> CorsLayer {
CorsLayer::new()
.allow_origin(Any)
.allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
@ -103,43 +32,37 @@ fn acp_cors_layer() -> CorsLayer {
header::CONTENT_TYPE,
header::ACCEPT,
HeaderName::from_static("x-secret-key"),
HeaderName::from_static("acp-connection-id"),
HeaderName::from_static("acp-session-id"),
header::SEC_WEBSOCKET_VERSION,
header::SEC_WEBSOCKET_KEY,
header::CONNECTION,
header::UPGRADE,
])
.expose_headers([
HeaderName::from_static("acp-connection-id"),
HeaderName::from_static("acp-session-id"),
])
}
fn create_acp_routes(server: Arc<AcpServer>) -> Router {
let registry = Arc::new(connection::ConnectionRegistry::new(server));
Router::new()
.route("/acp", post(http::handle_post).with_state(registry.clone()))
.route("/acp", get(handle_get).with_state(registry.clone()))
.route("/acp", delete(http::handle_delete).with_state(registry))
}
/// The bare ACP HTTP/WebSocket router (POST/GET/DELETE on `/acp`), without auth
/// or goose-specific auxiliary routes.
pub fn create_acp_router(server: Arc<AcpServer>) -> Router {
create_acp_routes(server).layer(acp_cors_layer())
AcpHttpServer::new(move || GooseAgentConnection::new(server.clone()))
.with_options(acp_http_options())
.into_router()
}
async fn health() -> &'static str {
"ok"
}
/// The full standalone ACP server router used by `goose serve`: ACP transport,
/// optional token auth, health/status endpoints, and the MCP app proxy.
pub fn create_router(server: Arc<AcpServer>, secret_key: String, require_token: bool) -> Router {
let mut acp_routes = create_acp_routes(server);
let mut acp_routes = create_acp_router(server);
if require_token {
acp_routes = acp_routes.layer(axum::middleware::from_fn_with_state(
secret_key.clone(),
auth::check_acp_token,
));
}
acp_routes
let aux_routes = Router::new()
.route("/health", get(health))
.route("/status", get(health))
.merge(super::mcp_app_proxy::routes(secret_key))
.layer(acp_cors_layer())
.layer(aux_cors_layer());
acp_routes.merge(aux_routes)
}

View file

@ -1,125 +0,0 @@
use std::sync::Arc;
use axum::{
extract::ws::{Message, WebSocket, WebSocketUpgrade},
http::{HeaderValue, StatusCode},
response::{IntoResponse, Response},
};
use futures::{SinkExt, StreamExt};
use tracing::{debug, error, info, trace, warn};
use super::connection::ConnectionRegistry;
use super::HEADER_CONNECTION_ID;
pub(crate) async fn handle_ws_upgrade(
registry: Arc<ConnectionRegistry>,
ws: WebSocketUpgrade,
) -> Response {
let (connection_id, connection) = match registry.create_connection().await {
Ok(pair) => pair,
Err(e) => {
error!("Failed to create WebSocket connection: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to create WebSocket connection",
)
.into_response();
}
};
connection.start_router().await;
let conn_id_for_handler = connection_id.clone();
let registry_for_handler = registry.clone();
let mut response = ws.on_upgrade(move |socket| async move {
run_ws(
socket,
registry_for_handler,
conn_id_for_handler,
connection,
)
.await
});
if let Ok(v) = HeaderValue::from_str(&connection_id) {
response.headers_mut().insert(HEADER_CONNECTION_ID, v);
}
info!(connection_id = %connection_id, "WebSocket connection created");
response
}
async fn run_ws(
socket: WebSocket,
registry: Arc<ConnectionRegistry>,
connection_id: String,
connection: Arc<super::connection::Connection>,
) {
let (mut ws_tx, mut ws_rx) = socket.split();
let (replay, mut outbound_rx) = connection.subscribe_all_outbound().await;
debug!(connection_id = %connection_id, "Starting WebSocket message loop");
for text in replay {
trace!(connection_id = %connection_id, payload = %text, "Agent → Client (replay): {} bytes", text.len());
if ws_tx.send(Message::Text(text.into())).await.is_err() {
error!(connection_id = %connection_id, "WebSocket send failed during replay");
if let Some(conn) = registry.remove(&connection_id).await {
conn.shutdown().await;
}
return;
}
}
loop {
tokio::select! {
msg_result = ws_rx.next() => {
match msg_result {
Some(Ok(Message::Text(text))) => {
let text_str = text.to_string();
trace!(connection_id = %connection_id, payload = %text_str, "Client → Agent: {} bytes", text_str.len());
if connection.to_agent_tx.send(text_str).await.is_err() {
error!(connection_id = %connection_id, "Agent channel closed");
break;
}
}
Some(Ok(Message::Close(frame))) => {
debug!(connection_id = %connection_id, "Client closed connection: {:?}", frame);
break;
}
Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => continue,
Some(Ok(Message::Binary(_))) => {
warn!(connection_id = %connection_id, "Ignoring binary message (ACP uses text)");
continue;
}
Some(Err(e)) => {
error!(connection_id = %connection_id, "WebSocket error: {}", e);
break;
}
None => break,
}
}
recv = outbound_rx.recv() => {
match recv {
Ok(text) => {
trace!(connection_id = %connection_id, payload = %text, "Agent → Client: {} bytes", text.len());
if ws_tx.send(Message::Text(text.into())).await.is_err() {
error!(connection_id = %connection_id, "WebSocket send failed");
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
warn!(connection_id = %connection_id, "WebSocket lagged {} messages", n);
continue;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
}
}
debug!(connection_id = %connection_id, "Cleaning up WebSocket connection");
if let Some(conn) = registry.remove(&connection_id).await {
conn.shutdown().await;
}
}

View file

@ -75,7 +75,7 @@ impl ProviderDef for CodexAcpProvider {
// servers are configured so codex-acp can connect to them.
let has_http_mcp = mcp_servers
.iter()
.any(|s| matches!(s, agent_client_protocol::schema::McpServer::Http(_)));
.any(|s| matches!(s, agent_client_protocol::schema::v1::McpServer::Http(_)));
if has_http_mcp {
args.extend([
"-c".to_string(),

View file

@ -4,9 +4,9 @@
#[path = "../acp_fixtures/mod.rs"]
pub mod fixtures;
use agent_client_protocol::schema::{
ContentBlock, ListSessionsResponse, McpServer, McpServerHttp, ModelId, SessionInfo,
SessionModeId, SessionUpdate, ToolCallStatus, ToolKind,
use agent_client_protocol::schema::v1::{
ContentBlock, ListSessionsResponse, McpServer, McpServerHttp, SessionInfo, SessionModeId,
SessionUpdate, ToolCallStatus, ToolKind,
};
use fixtures::{
assert_notifications, Connection, FsFixture, Notification, OpenAiFixture, PermissionDecision,
@ -540,7 +540,7 @@ pub async fn run_load_model<C: Connection>() {
assert_eq!(output.text, "2");
let SessionData { models, .. } = conn.load_session(&session_id, vec![]).await.unwrap();
assert_eq!(&*models.unwrap().current_model_id.0, "gpt-4.1");
assert_eq!(models.unwrap().current_model_id, "gpt-4.1");
}
pub async fn run_load_session_mcp<C: Connection>() {
@ -762,7 +762,6 @@ async fn run_mode_set_impl<C: Connection>(via: SetModeVia) {
let config = TestConnectionConfig {
data_root: temp_dir.path().to_path_buf(),
strip_config_options: matches!(via, SetModeVia::Dedicated),
..Default::default()
};
let mut conn = C::new(config, openai).await;
@ -885,7 +884,7 @@ pub async fn run_model_list<C: Connection>() {
let models = models.unwrap();
assert!(!models.available_models.is_empty());
assert_eq!(models.current_model_id, ModelId::new(TEST_MODEL));
assert_eq!(models.current_model_id, TEST_MODEL);
}
#[allow(dead_code)]
@ -940,19 +939,14 @@ pub async fn run_new_session_uses_current_config_mode<C: Connection>() {
}
pub async fn run_config_option_model_set<C: Connection>() {
run_model_set_impl::<C>(SetModelVia::ConfigOption).await;
run_model_set_impl::<C>().await;
}
pub async fn run_model_set<C: Connection>() {
run_model_set_impl::<C>(SetModelVia::Dedicated).await;
run_model_set_impl::<C>().await;
}
enum SetModelVia {
Dedicated,
ConfigOption,
}
async fn run_model_set_impl<C: Connection>(via: SetModelVia) {
async fn run_model_set_impl<C: Connection>() {
// Use a Chat Completions model so the canned SSE fixtures parse correctly.
// TODO: add a Responses API mock to OpenAiFixture for responses-routed models.
let expected_session_id = C::expected_session_id();
@ -973,10 +967,7 @@ async fn run_model_set_impl<C: Connection>(via: SetModelVia) {
)
.await;
let config = TestConnectionConfig {
strip_config_options: matches!(via, SetModelVia::Dedicated),
..Default::default()
};
let config = TestConnectionConfig::default();
let mut conn = C::new(config, openai).await;
// Session A: default model
@ -991,13 +982,9 @@ async fn run_model_set_impl<C: Connection>(via: SetModelVia) {
..
} = conn.new_session().await.unwrap();
let session_id = &session_b.session_id().0;
match via {
SetModelVia::Dedicated => conn.set_model(session_id, "gpt-4.1").await.unwrap(),
SetModelVia::ConfigOption => conn
.set_config_option(session_id, "model", "gpt-4.1")
.await
.unwrap(),
}
conn.set_config_option(session_id, "model", "gpt-4.1")
.await
.unwrap();
let set_model_notifs = session_b.notifications();
@ -1009,9 +996,8 @@ async fn run_model_set_impl<C: Connection>(via: SetModelVia) {
.unwrap();
assert_eq!(output.text, "2");
// Some connections emit a ConfigOption update immediately on model change,
// while the stripped legacy provider path only updates local state before
// the next prompt.
// Connections may emit a ConfigOption update immediately on model change,
// or only update local state before the next prompt.
let prompt_notifs = session_b.notifications();
let mut all = set_model_notifs;
all.extend(prompt_notifs);

View file

@ -2,7 +2,7 @@
#[path = "acp_common_tests/mod.rs"]
mod common_tests;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
ContentBlock, PromptRequest, SessionUpdate, StopReason, TextContent,
};
use common_tests::fixtures::server::AcpServerConnection;
@ -1097,9 +1097,11 @@ fn test_developer_fs_requests_use_acp_session_id() {
current_model: "gpt-4.1".to_string(),
read_text_file: Some(Arc::new(move |req| {
*seen_session_id_clone.lock().unwrap() = Some(req.session_id.0.to_string());
Ok(agent_client_protocol::schema::ReadTextFileResponse::new(
"test-read-content-12345",
))
Ok(
agent_client_protocol::schema::v1::ReadTextFileResponse::new(
"test-read-content-12345",
),
)
})),
..Default::default()
};

View file

@ -1,11 +1,11 @@
#![recursion_limit = "256"]
#![allow(unused_attributes)]
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
CreateTerminalResponse, KillTerminalResponse, ListSessionsResponse, McpServer,
ReadTextFileRequest, ReadTextFileResponse, ReleaseTerminalResponse, SessionModeState,
SessionModelState, SessionUpdate, TerminalExitStatus, TerminalId, TerminalOutputResponse,
ToolCallContent, ToolCallStatus, ToolKind, WaitForTerminalExitResponse, WriteTextFileRequest,
SessionUpdate, TerminalExitStatus, TerminalId, TerminalOutputResponse, ToolCallContent,
ToolCallStatus, ToolKind, WaitForTerminalExitResponse, WriteTextFileRequest,
WriteTextFileResponse,
};
use async_trait::async_trait;
@ -665,10 +665,16 @@ impl TerminalFixture {
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelStateFixture {
pub current_model_id: String,
pub available_models: Vec<String>,
}
#[derive(Debug)]
pub struct SessionData<S> {
pub session: S,
pub models: Option<SessionModelState>,
pub models: Option<ModelStateFixture>,
pub modes: Option<SessionModeState>,
}
@ -682,9 +688,6 @@ pub struct TestConnectionConfig {
pub read_text_file: Option<ReadTextFileHandler>,
pub write_text_file: Option<WriteTextFileHandler>,
pub terminal: Option<Arc<TerminalFixture>>,
// When true, strips config_options from responses to test the legacy set_mode/set_model path.
#[allow(dead_code)]
pub strip_config_options: bool,
// The model the server-side provider starts with. Defaults to TEST_MODEL.
pub current_model: String,
pub disable_session_naming: bool,
@ -702,7 +705,6 @@ impl Default for TestConnectionConfig {
read_text_file: None,
write_text_file: None,
terminal: None,
strip_config_options: false,
current_model: TEST_MODEL.to_string(),
disable_session_naming: true,
}
@ -739,7 +741,7 @@ pub trait Connection: Sized {
#[async_trait]
pub trait Session: std::fmt::Debug {
fn session_id(&self) -> &agent_client_protocol::schema::SessionId;
fn session_id(&self) -> &agent_client_protocol::schema::v1::SessionId;
fn work_dir(&self) -> std::path::PathBuf;
/// Drains and returns raw session updates collected by the fixture.
fn session_updates(&self) -> Vec<SessionUpdate>;

View file

@ -1,12 +1,11 @@
use super::{
spawn_acp_server_in_process, Connection, DuplexTransport, OpenAiFixture, PermissionDecision,
spawn_acp_server_in_process, Connection, ModelStateFixture, OpenAiFixture, PermissionDecision,
Session, SessionData, TestConnectionConfig, TestOutput,
};
use agent_client_protocol::schema::{
ListSessionsResponse, McpServer, ModelId, ModelInfo, SessionModelState, SessionUpdate,
ToolCallStatus,
use agent_client_protocol::schema::v1::{
ListSessionsResponse, McpServer, SessionUpdate, ToolCallStatus,
};
use agent_client_protocol::{Channel, Client, ConnectTo, DynConnectTo};
use agent_client_protocol::{Client, DynConnectTo};
use async_trait::async_trait;
use futures::StreamExt;
use goose::acp::{AcpProvider, AcpProviderConfig};
@ -17,7 +16,7 @@ use goose::permission::{Permission, PermissionConfirmation};
use goose::providers::base::Provider;
use goose_providers::model::ModelConfig;
use goose_test_support::{ExpectedSessionId, IgnoreSessionId, TEST_MODEL};
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use strum::VariantNames;
@ -34,7 +33,6 @@ pub struct AcpProviderConnection {
session_counter: usize,
notification_sink: NotificationSink,
session_models: SessionModels,
strip_config_options: bool,
work_dir: std::path::PathBuf,
data_root: std::path::PathBuf,
_openai: OpenAiFixture,
@ -45,7 +43,7 @@ pub struct AcpProviderConnection {
#[allow(dead_code)]
pub struct AcpProviderSession {
provider: Arc<Mutex<Option<AcpProvider>>>,
session_id: agent_client_protocol::schema::SessionId,
session_id: agent_client_protocol::schema::v1::SessionId,
notification_sink: NotificationSink,
session_models: SessionModels,
work_dir: std::path::PathBuf,
@ -197,12 +195,7 @@ impl Connection for AcpProviderConnection {
})),
};
// Server always advertises both configOptions and legacy; only the client fallback needs testing.
let transport: DynConnectTo<Client> = if config.strip_config_options {
DynConnectTo::new(strip_config_options(transport))
} else {
DynConnectTo::new(transport)
};
let transport: DynConnectTo<Client> = DynConnectTo::new(transport);
let provider = AcpProvider::connect_with_transport(
"acp-test".to_string(),
goose_mode,
@ -218,7 +211,6 @@ impl Connection for AcpProviderConnection {
session_counter: 0,
notification_sink,
session_models,
strip_config_options: config.strip_config_options,
work_dir: cwd_path,
data_root,
_openai: openai,
@ -231,24 +223,19 @@ impl Connection for AcpProviderConnection {
self.session_counter += 1;
let goose_id = format!("test-session-{}", self.session_counter);
let models = if self.strip_config_options {
None
} else {
let models = {
let provider = self.provider.lock().await;
let provider = provider.as_ref().unwrap();
let available_models = provider.fetch_supported_models().await?;
Some(SessionModelState::new(
ModelId::new(TEST_MODEL.to_string()),
available_models
.into_iter()
.map(|model_id| ModelInfo::new(ModelId::new(model_id.clone()), model_id))
.collect(),
))
Some(ModelStateFixture {
current_model_id: TEST_MODEL.to_string(),
available_models,
})
};
let session = AcpProviderSession {
provider: Arc::clone(&self.provider),
session_id: agent_client_protocol::schema::SessionId::new(goose_id),
session_id: agent_client_protocol::schema::v1::SessionId::new(goose_id),
notification_sink: self.notification_sink.clone(),
session_models: self.session_models.clone(),
work_dir: self.work_dir.clone(),
@ -318,7 +305,7 @@ impl Connection for AcpProviderConnection {
#[async_trait]
impl Session for AcpProviderSession {
fn session_id(&self) -> &agent_client_protocol::schema::SessionId {
fn session_id(&self) -> &agent_client_protocol::schema::v1::SessionId {
&self.session_id
}
@ -356,111 +343,3 @@ impl Session for AcpProviderSession {
self.send_message(message, decision).await
}
}
// Strips config_options from responses so goose falls back to legacy set_mode/set_model.
#[allow(dead_code)]
fn strip_config_options(transport: DuplexTransport) -> Channel {
let (server, server_future) = ConnectTo::<Client>::into_channel_and_future(transport);
let (client_channel, filter) = Channel::duplex();
tokio::spawn(async move {
if let Err(e) = server_future.await {
tracing::error!("config_options filter transport error: {e}");
}
});
tokio::spawn(async move {
let mut stripped_initial_config = HashSet::new();
let goose_to_server = async {
let mut from_goose = filter.rx;
while let Some(msg) = from_goose.next().await {
if server.tx.unbounded_send(msg).is_err() {
break;
}
}
};
let server_to_goose = async {
let mut from_server = server.rx;
while let Some(msg) = from_server.next().await {
let msg = match msg {
Ok(m) => match m {
agent_client_protocol::jsonrpcmsg::Message::Response(mut resp) => {
if let Some(ref mut result) = resp.result {
if let Some(obj) = result.as_object_mut() {
obj.remove("configOptions");
}
}
Ok(Some(agent_client_protocol::jsonrpcmsg::Message::Response(
resp,
)))
}
agent_client_protocol::jsonrpcmsg::Message::Request(req)
if req.id.is_none()
&& req.method == "session/update"
&& req
.params
.as_ref()
.and_then(|params| match params {
agent_client_protocol::jsonrpcmsg::Params::Object(obj) => {
Some(obj)
}
_ => None,
})
.and_then(|obj| obj.get("update"))
.and_then(|update| update.get("sessionUpdate"))
.and_then(|session_update| session_update.as_str())
== Some("config_option_update") =>
{
let session_id = req
.params
.as_ref()
.and_then(|params| match params {
agent_client_protocol::jsonrpcmsg::Params::Object(obj) => {
Some(obj)
}
_ => None,
})
.and_then(|obj| obj.get("sessionId"))
.and_then(|session_id| session_id.as_str())
.map(str::to_owned);
if let Some(session_id) = session_id {
if stripped_initial_config.insert(session_id) {
Ok(None)
} else {
Ok(Some(agent_client_protocol::jsonrpcmsg::Message::Request(
req,
)))
}
} else {
Ok(Some(agent_client_protocol::jsonrpcmsg::Message::Request(
req,
)))
}
}
other => Ok(Some(other)),
},
Err(err) => Err(err),
};
match msg {
Ok(Some(msg)) => {
if filter.tx.unbounded_send(Ok(msg)).is_err() {
break;
}
}
Ok(None) => continue,
Err(err) => {
if filter.tx.unbounded_send(Err(err)).is_err() {
break;
}
}
}
}
};
futures::join!(goose_to_server, server_to_goose);
});
client_channel
}

View file

@ -1,18 +1,18 @@
use super::{
map_permission_response, spawn_acp_server_in_process, Connection, PermissionDecision, Session,
SessionData, TestConnectionConfig, TestOutput,
map_permission_response, spawn_acp_server_in_process, Connection, ModelStateFixture,
PermissionDecision, Session, SessionData, TestConnectionConfig, TestOutput,
};
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
ClientCapabilities, CloseSessionRequest, ContentBlock, CreateTerminalRequest,
FileSystemCapabilities, ImageContent, InitializeRequest, KillTerminalRequest,
ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, McpServer, ModelId, ModelInfo,
NewSessionRequest, PromptRequest, ProtocolVersion, ReadTextFileRequest, ReleaseTerminalRequest,
RequestPermissionRequest, SessionConfigKind, SessionConfigOptionCategory,
SessionConfigOptionValue, SessionId, SessionModeId, SessionModelState, SessionNotification,
SessionUpdate, SetSessionConfigOptionRequest, SetSessionModeRequest, SetSessionModelRequest,
StopReason, TerminalOutputRequest, TextContent, ToolCallStatus, WaitForTerminalExitRequest,
WriteTextFileRequest,
ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, McpServer, NewSessionRequest,
PromptRequest, ReadTextFileRequest, ReleaseTerminalRequest, RequestPermissionRequest,
SessionConfigKind, SessionConfigOptionCategory, SessionConfigOptionValue, SessionId,
SessionModeId, SessionNotification, SessionUpdate, SetSessionConfigOptionRequest,
SetSessionModeRequest, StopReason, TerminalOutputRequest, TextContent, ToolCallStatus,
WaitForTerminalExitRequest, WriteTextFileRequest,
};
use agent_client_protocol::schema::ProtocolVersion;
use agent_client_protocol::{Agent, Client, ConnectionTo};
use async_trait::async_trait;
use goose::config::PermissionManager;
@ -37,7 +37,7 @@ pub struct AcpServerConnection {
pub struct AcpServerSession {
cx: ConnectionTo<Agent>,
session_id: agent_client_protocol::schema::SessionId,
session_id: agent_client_protocol::schema::v1::SessionId,
updates: Arc<Mutex<Vec<SessionNotification>>>,
permission: Arc<Mutex<PermissionDecision>>,
notify: Arc<Notify>,
@ -342,9 +342,7 @@ impl Connection for AcpServerConnection {
notify: self.notify.clone(),
_work_dir: work_dir,
};
let models = response.models.or_else(|| {
extract_model_state_from_config_options(response.config_options.as_deref())
});
let models = extract_model_state_from_config_options(response.config_options.as_deref());
self.updates.lock().unwrap().clear();
Ok(SessionData {
session,
@ -360,7 +358,7 @@ impl Connection for AcpServerConnection {
) -> anyhow::Result<SessionData<AcpServerSession>> {
self.updates.lock().unwrap().clear();
let work_dir = tempfile::tempdir().unwrap();
let session_id = agent_client_protocol::schema::SessionId::new(session_id.to_string());
let session_id = agent_client_protocol::schema::v1::SessionId::new(session_id.to_string());
let response = self
.cx
.send_request(
@ -377,9 +375,10 @@ impl Connection for AcpServerConnection {
notify: self.notify.clone(),
_work_dir: work_dir,
};
let models = extract_model_state_from_config_options(response.config_options.as_deref());
Ok(SessionData {
session,
models: response.models,
models,
modes: response.modes,
})
}
@ -425,15 +424,7 @@ impl Connection for AcpServerConnection {
}
async fn set_model(&self, session_id: &str, model_id: &str) -> anyhow::Result<()> {
self.cx
.send_request(SetSessionModelRequest::new(
SessionId::new(session_id),
model_id.to_string(),
))
.block_task()
.await
.map(|_| ())
.map_err(|e| e.into())
self.set_config_option(session_id, "model", model_id).await
}
async fn set_config_option(
@ -470,7 +461,7 @@ impl Connection for AcpServerConnection {
#[async_trait]
impl Session for AcpServerSession {
fn session_id(&self) -> &agent_client_protocol::schema::SessionId {
fn session_id(&self) -> &agent_client_protocol::schema::v1::SessionId {
&self.session_id
}
@ -514,8 +505,8 @@ impl Session for AcpServerSession {
}
fn extract_model_state_from_config_options(
config_options: Option<&[agent_client_protocol::schema::SessionConfigOption]>,
) -> Option<SessionModelState> {
config_options: Option<&[agent_client_protocol::schema::v1::SessionConfigOption]>,
) -> Option<ModelStateFixture> {
let option = config_options?
.iter()
.find(|option| option.category.as_ref() == Some(&SessionConfigOptionCategory::Model))?;
@ -524,33 +515,28 @@ fn extract_model_state_from_config_options(
};
let available_models = match &select.options {
agent_client_protocol::schema::SessionConfigSelectOptions::Ungrouped(options) => options
.iter()
.map(|option| {
ModelInfo::new(
ModelId::new(option.value.0.to_string()),
option.name.clone(),
)
})
.collect(),
agent_client_protocol::schema::SessionConfigSelectOptions::Grouped(groups) => groups
agent_client_protocol::schema::v1::SessionConfigSelectOptions::Ungrouped(options) => {
options
.iter()
.map(|option| option.value.0.to_string())
.collect()
}
agent_client_protocol::schema::v1::SessionConfigSelectOptions::Grouped(groups) => groups
.iter()
.flat_map(|group| {
group.options.iter().map(|option| {
ModelInfo::new(
ModelId::new(option.value.0.to_string()),
option.name.clone(),
)
})
group
.options
.iter()
.map(|option| option.value.0.to_string())
})
.collect(),
_ => Vec::new(),
};
Some(SessionModelState::new(
ModelId::new(select.current_value.0.to_string()),
Some(ModelStateFixture {
current_model_id: select.current_value.0.to_string(),
available_models,
))
})
}
fn collect_agent_text(updates: &Arc<Mutex<Vec<SessionNotification>>>) -> String {

View file

@ -2,7 +2,7 @@
#[path = "acp_common_tests/mod.rs"]
mod common_tests;
use agent_client_protocol::schema::{ForkSessionRequest, ForkSessionResponse, SessionId};
use agent_client_protocol::schema::v1::{ForkSessionRequest, ForkSessionResponse, SessionId};
use common_tests::fixtures::server::AcpServerConnection;
use common_tests::fixtures::{run_test, Connection, OpenAiFixture, TestConnectionConfig};
use goose::config::GooseMode;

View file

@ -1,7 +1,7 @@
#[allow(dead_code)]
#[path = "acp_common_tests/mod.rs"]
mod common_tests;
use agent_client_protocol::schema::{
use agent_client_protocol::schema::v1::{
ListSessionsRequest, ListSessionsResponse, NewSessionRequest, SessionConfigKind,
SessionConfigOptionCategory, SessionConfigOptionValue, SessionInfo,
SetSessionConfigOptionRequest,
@ -445,7 +445,7 @@ fn test_get_session_info() {
assert_eq!(
response.session.session_id,
agent_client_protocol::schema::SessionId::new(session.id)
agent_client_protocol::schema::v1::SessionId::new(session.id)
);
assert_eq!(response.session.cwd, cwd.to_path_buf());
assert_eq!(response.session.title.as_deref(), Some("Session info"));

File diff suppressed because one or more lines are too long

View file

@ -40,7 +40,7 @@ export type GooseExtension = {
*
* See protocol docs: [MCP Servers](https://agentclientprotocol.com/protocol/session-setup#mcp-servers)
*/
export type McpServer = McpServerHttp | McpServerSse | McpServerStdio;
export type McpServer = McpServerHttp | McpServerSse | McpServerAcp | McpServerStdio;
/**
* An HTTP header to set when making requests to the MCP server.
@ -124,6 +124,53 @@ export type McpServerSse = {
type: 'sse';
};
/**
* **UNSTABLE**
*
* This capability is not part of the spec yet, and may be removed or changed at any point.
*
* Unique identifier for an MCP server using the ACP transport.
*
* The value is opaque and generated by the ACP component providing the MCP server. It is
* used by `mcp/connect` to route connection requests back to the component that declared the
* server.
*/
export type McpServerAcpId = string;
/**
* **UNSTABLE**
*
* This capability is not part of the spec yet, and may be removed or changed at any point.
*
* ACP transport configuration for MCP.
*
* The MCP server is provided by an ACP component and communicates over the ACP channel
* using `mcp/connect`, `mcp/message`, and `mcp/disconnect`.
*/
export type McpServerAcp = {
/**
* Human-readable name identifying this MCP server.
*/
name: string;
/**
* Unique identifier for this MCP server, generated by the component providing it.
*
* Providers MUST NOT reuse an ID for multiple ACP-transport MCP servers that are visible
* on the same ACP connection.
*/
id: McpServerAcpId;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
* metadata to their interactions. Implementations MUST NOT make assumptions about values at
* these keys.
*
* See protocol docs: [Extensibility](https://agentclientprotocol.com/protocol/extensibility)
*/
_meta?: {
[key: string]: unknown;
} | null;
};
/**
* Stdio transport configuration for MCP.
*/
@ -380,8 +427,17 @@ export type ContentBlock = ({
* Optional annotations for the client. The client can use annotations to inform how objects are used or displayed
*/
export type Annotations = {
/**
* Intended recipients for this content, such as the user or assistant.
*/
audience?: Array<Role> | null;
/**
* Timestamp indicating when the underlying resource was last modified.
*/
lastModified?: string | null;
/**
* Relative importance of this content when clients choose what to surface.
*/
priority?: number | null;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -404,7 +460,13 @@ export type Role = 'assistant' | 'user';
* Text provided to or from an LLM.
*/
export type TextContent = {
/**
* Optional annotations that help clients decide how to display or route this content.
*/
annotations?: Annotations | null;
/**
* Text payload carried by this content block.
*/
text: string;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -422,9 +484,21 @@ export type TextContent = {
* An image provided to or from an LLM.
*/
export type ImageContent = {
/**
* Optional annotations that help clients decide how to display or route this content.
*/
annotations?: Annotations | null;
/**
* Base64-encoded media payload.
*/
data: string;
/**
* MIME type describing the encoded media payload.
*/
mimeType: string;
/**
* URI associated with this resource or media payload.
*/
uri?: string | null;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -442,8 +516,17 @@ export type ImageContent = {
* Audio provided to or from an LLM.
*/
export type AudioContent = {
/**
* Optional annotations that help clients decide how to display or route this content.
*/
annotations?: Annotations | null;
/**
* Base64-encoded media payload.
*/
data: string;
/**
* MIME type describing the encoded media payload.
*/
mimeType: string;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -461,12 +544,33 @@ export type AudioContent = {
* A resource that the server is capable of reading, included in a prompt or tool call result.
*/
export type ResourceLink = {
/**
* Optional annotations that help clients decide how to display or route this content.
*/
annotations?: Annotations | null;
/**
* Optional human-readable details shown with this protocol object.
*/
description?: string | null;
/**
* MIME type describing the encoded media payload.
*/
mimeType?: string | null;
/**
* Human-readable name shown for this protocol object.
*/
name: string;
/**
* Optional size of the linked resource in bytes, if known.
*/
size?: number | null;
/**
* Optional display title for end-user UI.
*/
title?: string | null;
/**
* URI associated with this resource or media payload.
*/
uri: string;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -489,8 +593,17 @@ export type EmbeddedResourceResource = TextResourceContents | BlobResourceConten
* Text-based resource contents.
*/
export type TextResourceContents = {
/**
* MIME type describing the encoded media payload.
*/
mimeType?: string | null;
/**
* Text payload carried by this content block.
*/
text: string;
/**
* URI associated with this resource or media payload.
*/
uri: string;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -508,8 +621,17 @@ export type TextResourceContents = {
* Binary resource contents.
*/
export type BlobResourceContents = {
/**
* Base64-encoded bytes for a binary resource payload.
*/
blob: string;
/**
* MIME type describing the encoded media payload.
*/
mimeType?: string | null;
/**
* URI associated with this resource or media payload.
*/
uri: string;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -527,7 +649,13 @@ export type BlobResourceContents = {
* The contents of a resource, embedded into a prompt or tool call result.
*/
export type EmbeddedResource = {
/**
* Optional annotations that help clients decide how to display or route this content.
*/
annotations?: Annotations | null;
/**
* Embedded resource payload, either text or binary data.
*/
resource: EmbeddedResourceResource;
/**
* The _meta property is reserved by ACP to allow clients and agents to attach additional
@ -1607,13 +1735,11 @@ export type SessionInfo = {
*/
cwd: string;
/**
* **UNSTABLE**
* Additional workspace roots reported for this session. Each path must be absolute.
*
* This capability is not part of the spec yet, and may be removed or changed at any point.
*
* Authoritative ordered additional workspace roots for this session. Each path must be absolute.
*
* When omitted or empty, there are no additional roots for the session.
* When present, this is the complete ordered additional-root list reported
* by the Agent. Omitted and empty values are equivalent: the response
* reports no additional roots.
*/
additionalDirectories?: Array<string>;
/**

View file

@ -42,6 +42,38 @@ export const zMcpServerSse = z.object({
type: z.literal('sse')
});
/**
* **UNSTABLE**
*
* This capability is not part of the spec yet, and may be removed or changed at any point.
*
* Unique identifier for an MCP server using the ACP transport.
*
* The value is opaque and generated by the ACP component providing the MCP server. It is
* used by `mcp/connect` to route connection requests back to the component that declared the
* server.
*/
export const zMcpServerAcpId = z.string();
/**
* **UNSTABLE**
*
* This capability is not part of the spec yet, and may be removed or changed at any point.
*
* ACP transport configuration for MCP.
*
* The MCP server is provided by an ACP component and communicates over the ACP channel
* using `mcp/connect`, `mcp/message`, and `mcp/disconnect`.
*/
export const zMcpServerAcp = z.object({
name: z.string(),
id: zMcpServerAcpId,
_meta: z.union([
z.record(z.unknown()),
z.null()
]).optional()
});
/**
* An environment variable to set when launching an MCP server.
*/
@ -79,6 +111,7 @@ export const zMcpServerStdio = z.object({
export const zMcpServer = z.union([
zMcpServerHttp,
zMcpServerSse,
zMcpServerAcp,
zMcpServerStdio
]);
@ -321,7 +354,10 @@ export const zSetSessionSystemPromptRequest_unstable = z.object({
/**
* The sender or recipient of messages and data in a conversation.
*/
export const zRole = z.enum(['assistant', 'user']);
export const zRole = z.union([
z.literal('assistant'),
z.literal('user')
]);
/**
* Optional annotations for the client. The client can use annotations to inform how objects are used or displayed