Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions plugins/agent/model/src/adapters/body.rs
Original file line number Diff line number Diff line change
Expand Up @@ -280,6 +280,7 @@ mod tests {
provider_namespace: "openai".into(),
metadata_namespace: "openai".into(),
multipart_tool_content: false,
reasoning_field: Default::default(),
}
}

Expand Down
23 changes: 11 additions & 12 deletions plugins/agent/model/src/adapters/request.rs
Original file line number Diff line number Diff line change
Expand Up @@ -172,11 +172,15 @@ pub(super) async fn prepare(request: Request) -> Result<Prepared, ModelError> {
"messages",
)
} else {
let namespace = match request.provider.kind {
ProviderKind::OpenaiCompatible { name } => name,
ProviderKind::OpenaiChatCompletions => "openai".into(),
_ => return Err(ModelError::Adapter("unsupported provider protocol".into())),
};
let options = super::wire::chat_options(&request.provider.kind, &request.provider_options)?;
let include_usage = request
.provider_options
.get("maka")
.and_then(|options| options.get("includeUsage"))
.map(|value| serde_json::from_value::<bool>(value.clone()))
.transpose()
.map_err(|error| ModelError::Adapter(error.to_string()))?
.unwrap_or(true);
let tools = request
.tools
.into_iter()
Expand Down Expand Up @@ -214,17 +218,12 @@ pub(super) async fn prepare(request: Request) -> Result<Prepared, ModelError> {
streaming: true,
},
&chat_completions::request::Config {
provider_namespace: namespace.clone(),
include_usage: true,
provider_namespace: options.provider_namespace.clone(),
include_usage,
..Default::default()
},
)
.map_err(failure)?;
let options = chat_completions::request::PromptOptions {
provider_namespace: namespace.clone(),
metadata_namespace: prepared.metadata_namespace.clone(),
multipart_tool_content: false,
};
if let Some(wire) = request
.prompt
.wire(crate::prompt::WireFormat::ChatCompletions)
Expand Down
21 changes: 16 additions & 5 deletions plugins/agent/model/src/adapters/wire.rs
Original file line number Diff line number Diff line change
Expand Up @@ -62,17 +62,28 @@ pub(super) fn chat_options(
ProviderKind::OpenaiCompatible { name } => name,
_ => return Err(ModelError::Adapter("not a Chat Completions codec".into())),
};
Ok(chat::prompt_options(
let mut prompt_options = chat::prompt_options(
namespace,
&prompt::extensions(Some(options.clone()))?,
false,
))
);
prompt_options.reasoning_field = options
.get("maka")
.and_then(|options| options.get("openAiChatReasoningField"))
.map(|value| serde_json::from_value(value.clone()))
.transpose()
.map_err(|error| ModelError::Adapter(error.to_string()))?
.unwrap_or_default();
Ok(prompt_options)
}
pub(super) fn chat_codec(language: Language, options: &chat::PromptOptions) -> String {
serde_json::json!({"codec":"maka.chat-items.v1","language":language,
let mut codec = serde_json::json!({"codec":"maka.chat-items.v1","language":language,
"namespace":options.provider_namespace,"metadataNamespace":options.metadata_namespace,
"multipartToolContent":options.multipart_tool_content})
.to_string()
"multipartToolContent":options.multipart_tool_content});
if options.reasoning_field != maka_provider_protocol::output::ReasoningField::Content {
codec["reasoningField"] = serde_json::json!(options.reasoning_field);
}
codec.to_string()
}

/// Immutable message encoding policy. Tool schemas belong to the request header;
Expand Down
2 changes: 1 addition & 1 deletion plugins/agent/model/src/adapters/wire/chat_source.rs
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,7 @@ async fn encode(
}
if nonempty_reasoning {
fields.push((
"reasoning_content",
options.reasoning_field.as_str(),
assistant::plan(parts, assistant::Mode::Reasoning)?,
));
}
Expand Down
76 changes: 76 additions & 0 deletions plugins/agent/model/tests/integration/provider_compatible.rs
Original file line number Diff line number Diff line change
Expand Up @@ -460,3 +460,79 @@ async fn full_request_byte_count_matches_upload_and_stops_oversized_headers() {
"count covers actual model/options/messages/overlay envelope"
);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ollama_cloud_policy_sends_reasoning_replay_and_receives_a_reply() {
use maka_agent_facts::configuration::ModelInfo;
use maka_agent_ports::provider::{Connection, Resolve};
use maka_model::prompt::{AssistantPart, Message};
use maka_providers::{api::ApiProviders, catalog::Snapshot};

tokio::time::timeout(Duration::from_secs(20), async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1", listener.local_addr().unwrap());
let binding = ApiProviders::definition("ollama-cloud", Arc::new(Snapshot::bundled().unwrap())).unwrap();
let model = binding.prepare(Resolve {
connection: Connection { id: "cloud".into(), revision: 1, configuration: binding.configure(json!({"baseUrl":url})).unwrap() },
model: ModelInfo::new("deepseek-v4-flash"),
overrides: None,
thinking_level: None,
}).await.unwrap();
let mut input = request(&url);
input.provider.kind = model.protocol;
input.provider.model = model.info.id;
input.provider.base_url = model.base_url;
input.provider_options = model.provider_options;
input.prompt = vec![Message::Assistant {
content: vec![AssistantPart::Reasoning { text: "previous thought".into(), provider_options: None }, AssistantPart::Text { text: "previous reply".into(), provider_options: None }],
provider_options: None,
}, Message::user("continue")].into();
let server = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let wire = body(&mut socket).await;
assert_eq!(wire["messages"][0], json!({"role":"assistant","content":"previous reply","reasoning":"previous thought"}));
assert_eq!(wire["stream_options"], json!({"include_usage":true}));
assert!(wire.get("maka").is_none());
response(&mut socket, 200, &completion(&[json!({"reasoning":"new thought"}), json!({"content":"reply"})])).await;
});
let executor = ModelExecutor::new(Duration::from_secs(10)).unwrap();
let mut stream = executor.stream(input, CancellationToken::new()).await.unwrap();
let mut builder = StepBuilder::default();
while let Some(event) = stream.next().await {
builder.push(event.unwrap().event).unwrap();
}
stream.cancel_and_wait().await.unwrap();
let step = builder.finish().unwrap();
assert_eq!(step.usage.output_tokens, Some(6));
assert!(step.parts.iter().any(|part| matches!(part, ModelPart::Text { text_kind: TextKind::Text, text, .. } if text.chunks().collect::<String>() == "reply")));
server.await.unwrap();
}).await.unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn explicit_usage_opt_out_is_respected_on_the_first_request() {
tokio::time::timeout(Duration::from_secs(20), async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let wire = body(&mut socket).await;
assert!(wire.get("stream_options").is_none());
response(&mut socket, 200, &completion(&[json!({"content":"ok"})])).await;
});
let mut input = request(&url);
input.provider_options = json!({"maka":{"includeUsage":false}});
let executor = ModelExecutor::new(Duration::from_secs(10)).unwrap();
let mut stream = executor
.stream(input, CancellationToken::new())
.await
.unwrap();
while let Some(event) = stream.next().await {
event.unwrap();
}
stream.cancel_and_wait().await.unwrap();
server.await.unwrap();
})
.await
.unwrap();
}
98 changes: 98 additions & 0 deletions plugins/agent/model/tests/integration/wire_encoding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -262,3 +262,101 @@ async fn notifications_keep_all_output_formats_without_a_second_provider_receipt
}
}
}

#[test]
fn default_chat_codec_preserves_the_legacy_identity() {
let input = request(
ProviderKind::OpenaiChatCompletions,
"http://localhost/v1".into(),
);
let encoding = WireEncoding::new(
input.language,
&input.provider,
&input.provider_options,
&input.tools,
)
.unwrap();
assert_eq!(
encoding.codec(),
r#"{"codec":"maka.chat-items.v1","language":"en","namespace":"openai","metadataNamespace":"openai","multipartToolContent":false}"#
);
let reasoning = WireEncoding::new(
input.language,
&input.provider,
&json!({"maka":{"openAiChatReasoningField":"reasoning"}}),
&input.tools,
)
.unwrap();
assert_ne!(encoding.codec(), reasoning.codec());
}

#[tokio::test]
async fn generated_chat_reasoning_uses_the_selected_field() {
use maka_provider_protocol::{
chat_completions::request as chat, input as wire, output::ReasoningField,
};
for (field, expected) in [
(
ReasoningField::Content,
json!({"role":"assistant","content":"reply","reasoning_content":"思考"}),
),
(
ReasoningField::Reasoning,
json!({"role":"assistant","content":"reply","reasoning":"思考"}),
),
] {
let mut input = request(
ProviderKind::OpenaiCompatible {
name: "ollama-cloud".into(),
},
"http://localhost/v1".into(),
);
input.provider_options = json!({"maka":{"openAiChatReasoningField":field}});
let encoding = WireEncoding::new(
input.language,
&input.provider,
&input.provider_options,
&input.tools,
)
.unwrap();
let messages = vec![Message::Assistant {
content: vec![
AssistantPart::Reasoning {
text: "思考".into(),
provider_options: None,
},
AssistantPart::Text {
text: "reply".into(),
provider_options: None,
},
],
provider_options: None,
}];
drop(input);
let frozen = encoding.prepare_source(Arc::new(maka_agent_facts::model::prompt_source::source::generated::GeneratedMessages::new(&messages)), resources()).await.unwrap();
assert_eq!(items(&frozen).await, vec![expected.clone()]);
let direct = chat::messages(
&[wire::Message::Assistant {
parts: vec![
wire::AssistantPart::Reasoning {
text: "思考".into(),
extensions: None,
},
wire::AssistantPart::Text {
text: "reply".into(),
extensions: None,
},
],
extensions: None,
}],
&chat::PromptOptions {
provider_namespace: "ollama-cloud".into(),
metadata_namespace: "ollamaCloud".into(),
multipart_tool_content: false,
reasoning_field: field,
},
)
.unwrap();
assert_eq!(direct, vec![expected]);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,7 @@ pub fn prompt_options(
provider_namespace: namespace.to_owned(),
metadata_namespace,
multipart_tool_content,
reasoning_field: Default::default(),
}
}

Expand Down Expand Up @@ -388,6 +389,7 @@ fn prepare_inner(
&PromptOptions {
provider_namespace: namespace.clone(),
metadata_namespace: metadata_namespace.clone(),
reasoning_field: Default::default(),
multipart_tool_content: config.multipart_tool_content
}
)?);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ pub struct PromptOptions {
pub provider_namespace: String,
pub metadata_namespace: String,
pub multipart_tool_content: bool,
pub reasoning_field: crate::output::ReasoningField,
}
fn metadata<'a>(extensions: &'a Extensions, namespace: &str) -> Option<&'a Value> {
extensions.as_ref()?.get(namespace)
Expand Down Expand Up @@ -370,10 +371,16 @@ pub fn messages(messages: &[Message], options: &PromptOptions) -> Result<Vec<Val
continue;
}
};
result.push(overlay(
serde_json::to_value(wire).expect("message"),
extensions,
));
let mut wire = serde_json::to_value(wire).expect("message");
if options.reasoning_field == crate::output::ReasoningField::Reasoning
&& let Some(reasoning) = wire
.as_object_mut()
.expect("message")
.remove("reasoning_content")
{
wire["reasoning"] = reasoning;
}
result.push(overlay(wire, extensions));
}
Ok(result)
}
5 changes: 4 additions & 1 deletion plugins/agent/provider-protocol/src/output.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,11 @@ pub enum TextKind {
Text,
Reasoning,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReasoningField {
#[default]
#[serde(rename = "reasoning_content")]
Content,
Reasoning,
}
Expand Down
28 changes: 28 additions & 0 deletions plugins/agent/providers/src/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -318,3 +318,31 @@ impl Provider for ApiProvider {
Box::pin(async move { effects::verify(self, request, context).await })
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn catalog_usage_opt_out_survives_chat_preparation() {
let facts = facts::provider_facts("github-copilot").unwrap();
let provider = ApiProvider {
id: "github-copilot",
facts,
snapshot: Arc::new(crate::catalog::Snapshot::bundled().unwrap()),
};
let model = provider
.prepare(Resolve {
connection: Connection {
id: "copilot".into(),
revision: 1,
configuration: json!({"baseUrl":facts.base_url}),
},
model: ModelInfo::new("kimi-k3"),
overrides: None,
thinking_level: None,
})
.unwrap();
assert_eq!(model.provider_options["maka"]["includeUsage"], false);
}
}
18 changes: 17 additions & 1 deletion plugins/agent/providers/src/api/options.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,23 @@ pub(super) fn resolve(
))
}
ProviderKind::OpenaiCompatible { name } => {
Ok(compatible::chat_completions(name, thinking_level, parallel))
let mut options = compatible::chat_completions(name, thinking_level, parallel);
if let crate::facts::adapter::AdapterKind::OpenaiCompatible {
include_usage,
replay_assistant_reasoning_as,
..
} = &route.adapter.kind
{
if let Some(include) = include_usage {
options["maka"]["includeUsage"] = json!(include);
}
if let Some(crate::facts::adapter::ReasoningField::Reasoning) =
replay_assistant_reasoning_as
{
options["maka"]["openAiChatReasoningField"] = json!("reasoning");
}
}
Ok(options)
}
ProviderKind::OpenResponses(contract) => {
if parallel == Some(false) {
Expand Down
Loading
Loading