diff --git a/plugins/agent/engine/src/history/prepared.rs b/plugins/agent/engine/src/history/prepared.rs index f22e3f24858..afbca6e915c 100644 --- a/plugins/agent/engine/src/history/prepared.rs +++ b/plugins/agent/engine/src/history/prepared.rs @@ -92,14 +92,9 @@ enum Plan { } pub(super) enum MessageBuilder { User(Arc>), - Assistant { - parts: Range, - options: Option, - }, + Assistant { parts: Range }, Tool(Box), - DeferredUser { - parts: Range, - }, + DeferredUser { parts: Range }, } struct Producer { id: [u8; 32], @@ -108,7 +103,6 @@ struct Producer { struct AssistantGroup { key: u64, start: usize, - options: Option, } #[allow(clippy::too_many_arguments)] @@ -336,21 +330,12 @@ async fn capture_owned( parts.push(part); builders.push(MessageBuilder::Assistant { parts: start..parts.len(), - options: None, }); } ProjectedRow::Assistant(part) => { ensure_group(row.order.source_sequence, &mut builders, &parts, &mut group); parts.push(part); } - ProjectedRow::AssistantMetadata { options } => { - ensure_group(row.order.source_sequence, &mut builders, &parts, &mut group); - if let Some(group) = &mut group - && group.options.is_none() - { - group.options = Some(options); - } - } ProjectedRow::Skip => {} } } @@ -395,12 +380,9 @@ async fn capture_owned( content, provider_options: None, }, - MessageBuilder::Assistant { - parts: range, - options, - } => MessageRecipe::Assistant { + MessageBuilder::Assistant { parts: range } => MessageRecipe::Assistant { content: Arc::new(Items::range(parts.clone(), range, owner.clone())), - provider_options: options, + provider_options: None, }, MessageBuilder::DeferredUser { parts: range } => MessageRecipe::User { content: Arc::new(Items::range( @@ -500,11 +482,10 @@ fn flush( group: &mut Option, ) { if let Some(group) = group.take() - && (group.start != parts.len() || group.options.is_some()) + && group.start != parts.len() { builders.push(MessageBuilder::Assistant { parts: group.start..parts.len(), - options: group.options, }); } } @@ -521,7 +502,6 @@ fn ensure_group( *group = Some(AssistantGroup { key, start: parts.len(), - options: None, }); } fn add(total: &mut usize, bytes: usize) -> io::Result<()> { diff --git a/plugins/agent/engine/src/history/prepared/backed.rs b/plugins/agent/engine/src/history/prepared/backed.rs index 050cf18a983..24fb05f93c1 100644 --- a/plugins/agent/engine/src/history/prepared/backed.rs +++ b/plugins/agent/engine/src/history/prepared/backed.rs @@ -29,7 +29,6 @@ use maka_agent_facts::model::prompt_source::source::generated; struct Group { key: u64, start: usize, - options: Option, } struct Output { list: List, @@ -52,13 +51,13 @@ async fn flush_group( cancel: &CancellationToken, ) -> io::Result<()> { if let Some(group) = group.take() - && (group.start != part_count || group.options.is_some()) + && group.start != part_count { let message = store .message_list( MessageRole::Assistant, parts.range(group.start..part_count)?, - group.options, + None, budget, cancel, ) @@ -223,38 +222,11 @@ pub(super) async fn capture( group.get_or_insert(Group { key: row.order.source_sequence, start: part_count, - options: None, }); store.put(parts, part_count, part).await.map_err(error)?; part_count += 1; } } - ArenaProjectedRow::AssistantMetadata(options) => { - if group - .as_ref() - .is_some_and(|g| g.key != row.order.source_sequence) - { - flush_group( - &mut group, - parts, - part_count, - &mut out, - &store, - budget.clone(), - cancellation, - ) - .await - .map_err(error)?; - } - let g = group.get_or_insert(Group { - key: row.order.source_sequence, - start: part_count, - options: None, - }); - if g.options.is_none() { - g.options = Some(options); - } - } ArenaProjectedRow::Skip => {} } } diff --git a/plugins/agent/engine/src/history/source_projector.rs b/plugins/agent/engine/src/history/source_projector.rs index 0bf55b44791..f1f63f1c193 100644 --- a/plugins/agent/engine/src/history/source_projector.rs +++ b/plugins/agent/engine/src/history/source_projector.rs @@ -103,7 +103,6 @@ impl ProjectContext<'_> { pub(super) enum ProjectedRow { User(Arc>), Assistant(AssistantRecipe), - AssistantMetadata { options: JsonSource }, Tool(ToolResultRecipe), Skip, } @@ -494,42 +493,6 @@ async fn project_text( thinking: bool, context: &ProjectContext<'_>, ) -> io::Result { - if thinking - && matches!(context.provider, ProviderKind::OpenaiCompatible { .. }) - && let Some(options) = provider_options.as_ref() - && let Some(field) = json::field( - options, - &["maka", "openAiChatReasoningField"], - context.budget.clone(), - ) - .await? - && field.kind == JsonKind::String - && (json::text_eq(&field.text()?, "reasoning", context.budget.clone()).await? - || json::text_eq(&field.text()?, "reasoning_content", context.budget.clone()).await?) - { - if text.encoding() != TextEncoding::JsonString { - return Err(io::Error::new( - io::ErrorKind::Unsupported, - "compatible reasoning requires captured quoted text source", - )); - } - let prefix = if json::text_eq(&field.text()?, "reasoning", context.budget.clone()).await? { - "{\"openaiCompatible\":{\"reasoning\":" - } else { - "{\"openaiCompatible\":{\"reasoning_content\":" - }; - let bytes = super::joined_source::join( - vec![ - literal(prefix).into_bytes(), - text.into_bytes(), - literal("}}").into_bytes(), - ], - context.directory_owner.clone(), - )?; - return Ok(ProjectedRow::AssistantMetadata { - options: JsonSource::new(bytes), - }); - } if thinking && let ProviderKind::OpenResponses(contract) = context.provider { match contract.reasoning_replay { maka_agent_facts::model::PlaintextReasoningReplay::PlaintextContent => { @@ -1181,22 +1144,6 @@ async fn text_footprint( context: &ProjectContext<'_>, ) -> io::Result { let mut bytes = raw_footprint(text.bytes())?; - if thinking - && matches!(context.provider, ProviderKind::OpenaiCompatible { .. }) - && let Some(options) = options - && let Some(field) = json::field( - options, - &["maka", "openAiChatReasoningField"], - context.budget.clone(), - ) - .await? - && field.kind == JsonKind::String - && (json::text_eq(&field.text()?, "reasoning", context.budget.clone()).await? - || json::text_eq(&field.text()?, "reasoning_content", context.budget.clone()).await?) - { - footprint_add(&mut bytes, super::joined_source::metadata_bytes(3)?)?; - return Ok(bytes); - } if thinking && let ProviderKind::OpenResponses(contract) = context.provider && contract.reasoning_replay diff --git a/plugins/agent/engine/src/history/source_projector/arena.rs b/plugins/agent/engine/src/history/source_projector/arena.rs index ab00e6ee504..a913ac58308 100644 --- a/plugins/agent/engine/src/history/source_projector/arena.rs +++ b/plugins/agent/engine/src/history/source_projector/arena.rs @@ -25,7 +25,6 @@ pub(in crate::history) enum ArenaProjectedRow { User(List), Assistant(Node), Tool(Node), - AssistantMetadata(JsonSource), Skip, } @@ -105,9 +104,6 @@ pub(in crate::history) async fn project_row( .result(result, context.budget.clone(), context.cancellation) .await?, )), - ProjectedRow::AssistantMetadata { options } => { - Ok(ArenaProjectedRow::AssistantMetadata(options)) - } ProjectedRow::Skip => Ok(ArenaProjectedRow::Skip), } } diff --git a/plugins/agent/engine/tests/integration/agent_loop.rs b/plugins/agent/engine/tests/integration/agent_loop.rs index d8621311402..7c7028e518a 100644 --- a/plugins/agent/engine/tests/integration/agent_loop.rs +++ b/plugins/agent/engine/tests/integration/agent_loop.rs @@ -262,3 +262,179 @@ async fn loop_commits_before_effects_replays_history_after_reopen_and_rejects_du .await .expect("agent loop must make bounded progress"); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn compatible_reasoning_replay_honors_override_observed_empty_and_multipart_steps() { + use tokio::io::AsyncWriteExt; + + async fn stream_sse(socket: &mut tokio::net::TcpStream, deltas: &[Value], tool: bool) { + let mut body = String::new(); + for delta in deltas { + let chunk = json!({"id":"reply","object":"chat.completion.chunk","created":1,"model":"test","choices":[{"index":0,"delta":delta,"finish_reason":null}]}); + body.push_str(&format!("data: {chunk}\n\n")); + } + let finish = json!({"id":"reply","object":"chat.completion.chunk","created":1,"model":"test","choices":[{"index":0,"delta":{},"finish_reason":if tool {"tool_calls"} else {"stop"}}],"usage":{"prompt_tokens":4,"completion_tokens":3,"total_tokens":7}}); + body.push_str(&format!("data: {finish}\n\ndata: [DONE]\n\n")); + socket + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + } + + tokio::time::timeout(Duration::from_secs(30), async { + let directory = tempfile::tempdir().unwrap(); + let engine = Engine::new( + ModelExecutor::new(Duration::from_secs(10)).unwrap(), + CodeExecutor::new(CellLimits::default()).unwrap(), + ) + .unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}/v1", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse( + &mut socket, + &[ + json!({"reasoning_content":"part-1 "}), + json!({"content":"checking"}), + json!({"reasoning_content":"part-2"}), + json!({"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"echo","arguments":"{\"value\":42}"}}]}), + ], + true, + ) + .await; + + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse( + &mut socket, + &[json!({"reasoning_content":""}), json!({"content":"done"})], + false, + ) + .await; + + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse(&mut socket, &[json!({"content":"final"})], false).await; + requests + }); + + let log = Arc::new( + SessionFixture::open(&directory.path().join("override.sqlite")) + .await + .unwrap(), + ); + let effect = Arc::new(Effect { + log: log.clone(), + count: Arc::new(AtomicUsize::new(0)), + }); + for suffix in ["first", "next"] { + let mut run = input(&base, suffix, effect.clone()); + run.provider_id = "ollama-cloud".into(); + run.provider.kind = maka_model::ProviderKind::OpenaiCompatible { + name: "ollama-cloud".into(), + }; + run.provider_options = + json!({"maka":{"includeUsage":true,"openAiChatReasoningField":"reasoning"}}); + engine + .run(log.store(), run, CancellationToken::new()) + .await + .unwrap(); + } + let requests = server.await.unwrap(); + assert_eq!(requests[0]["stream_options"], json!({"include_usage":true})); + let assistants = |request: &Value| { + request["messages"] + .as_array() + .unwrap() + .iter() + .filter(|message| message["role"] == "assistant") + .cloned() + .collect::>() + }; + let step_two = assistants(&requests[1]); + assert_eq!(step_two.len(), 1); + assert_eq!(step_two[0]["reasoning"], "part-1 part-2"); + assert!(step_two[0].get("reasoning_content").is_none()); + + let turn_two = assistants(&requests[2]); + assert_eq!(turn_two.len(), 2); + assert_eq!(turn_two[0]["reasoning"], "part-1 part-2"); + assert!(turn_two[0].get("reasoning_content").is_none()); + assert_eq!(turn_two[1]["reasoning"], ""); + assert!(turn_two[1].get("reasoning_content").is_none()); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}/v1", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse( + &mut socket, + &[ + json!({"reasoning_content":""}), + json!({"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"echo","arguments":"{\"value\":42}"}}]}), + ], + true, + ) + .await; + + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse( + &mut socket, + &[json!({"reasoning":"native-why"}), json!({"content":"done"})], + false, + ) + .await; + + let (mut socket, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut socket).await); + stream_sse(&mut socket, &[json!({"content":"final"})], false).await; + requests + }); + + let log = Arc::new( + SessionFixture::open(&directory.path().join("observed.sqlite")) + .await + .unwrap(), + ); + let effect = Arc::new(Effect { + log: log.clone(), + count: Arc::new(AtomicUsize::new(0)), + }); + for suffix in ["first", "next"] { + let mut run = input(&base, suffix, effect.clone()); + run.provider.kind = maka_model::ProviderKind::OpenaiCompatible { + name: "fixture".into(), + }; + run.provider_options = json!({"maka":{"includeUsage":false}}); + engine + .run(log.store(), run, CancellationToken::new()) + .await + .unwrap(); + } + let requests = server.await.unwrap(); + assert!(requests[0].get("stream_options").is_none()); + let turn_two = assistants(&requests[2]); + assert_eq!(turn_two.len(), 2); + assert_eq!(turn_two[0]["reasoning_content"], ""); + assert!(turn_two[0].get("reasoning").is_none()); + assert_eq!(turn_two[1]["reasoning"], "native-why"); + assert!(turn_two[1].get("reasoning_content").is_none()); + + engine.drain().await; + }) + .await + .expect("compatible reasoning replay must make bounded progress"); +} diff --git a/plugins/agent/model/src/adapters/body.rs b/plugins/agent/model/src/adapters/body.rs index 78d3ee815ed..a0ee5251203 100644 --- a/plugins/agent/model/src/adapters/body.rs +++ b/plugins/agent/model/src/adapters/body.rs @@ -280,6 +280,7 @@ mod tests { provider_namespace: "openai".into(), metadata_namespace: "openai".into(), multipart_tool_content: false, + reasoning_field: Default::default(), } } diff --git a/plugins/agent/model/src/adapters/request.rs b/plugins/agent/model/src/adapters/request.rs index cba7642fb0f..ca248284274 100644 --- a/plugins/agent/model/src/adapters/request.rs +++ b/plugins/agent/model/src/adapters/request.rs @@ -172,11 +172,15 @@ pub(super) async fn prepare(request: Request) -> Result { "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::(value.clone())) + .transpose() + .map_err(|error| ModelError::Adapter(error.to_string()))? + .unwrap_or(true); let tools = request .tools .into_iter() @@ -214,17 +218,12 @@ pub(super) async fn prepare(request: Request) -> Result { 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) diff --git a/plugins/agent/model/src/adapters/wire.rs b/plugins/agent/model/src/adapters/wire.rs index f628a941785..70c7c50ca24 100644 --- a/plugins/agent/model/src/adapters/wire.rs +++ b/plugins/agent/model/src/adapters/wire.rs @@ -62,16 +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()))?; + Ok(prompt_options) } pub(super) fn chat_codec(language: Language, options: &chat::PromptOptions) -> String { - serde_json::json!({"codec":"maka.chat-items.v1","language":language, - "namespace":options.provider_namespace,"metadataNamespace":options.metadata_namespace, - "multipartToolContent":options.multipart_tool_content}) + serde_json::json!({ + "codec": "maka.chat-items.v1", + "language": language, + "namespace": options.provider_namespace, + "metadataNamespace": options.metadata_namespace, + "multipartToolContent": options.multipart_tool_content, + "reasoningField": options.reasoning_field, + }) .to_string() } diff --git a/plugins/agent/model/src/adapters/wire/chat_source.rs b/plugins/agent/model/src/adapters/wire/chat_source.rs index 2214a5b36c7..fde3436a135 100644 --- a/plugins/agent/model/src/adapters/wire/chat_source.rs +++ b/plugins/agent/model/src/adapters/wire/chat_source.rs @@ -62,6 +62,42 @@ async fn validate_options( } Ok(()) } +async fn observed_reasoning_field( + options: &JsonSource, + budget: &Arc, +) -> Result, ModelError> { + let Some(field) = recipe::json::field( + options, + &["maka", "openAiChatReasoningField"], + budget.clone(), + ) + .await + .map_err(error)? + else { + return Ok(None); + }; + if field.kind != recipe::json::JsonKind::String { + return Ok(None); + } + let text = field.text().map_err(error)?; + if recipe::json::text_eq(&text, "reasoning", budget.clone()) + .await + .map_err(error)? + { + Ok(Some( + maka_provider_protocol::output::ReasoningField::Reasoning, + )) + } else if recipe::json::text_eq(&text, "reasoning_content", budget.clone()) + .await + .map_err(error)? + { + Ok(Some( + maka_provider_protocol::output::ReasoningField::Content, + )) + } else { + Ok(None) + } +} async fn overlay( options: Option, fields: Vec<(&'static str, Vec)>, @@ -374,6 +410,7 @@ async fn encode( let mut cursor = parts.open(0); let mut nonempty_text = false; let mut nonempty_reasoning = false; + let mut observed_reasoning = None; let mut has_calls = false; while let Some(part) = cursor.next().await.map_err(error)? { tokio::task::yield_now().await; @@ -401,10 +438,22 @@ async fn encode( .await .map_err(error)? } - AssistantRecipe::Reasoning { text, .. } if !nonempty_reasoning => { - nonempty_reasoning = !source::escaped_equals(&text, b"", budget.clone()) - .await - .map_err(error)? + AssistantRecipe::Reasoning { + text, + provider_options, + } => { + if !nonempty_reasoning { + nonempty_reasoning = + !source::escaped_equals(&text, b"", budget.clone()) + .await + .map_err(error)?; + } + if observed_reasoning.is_none() + && let Some(options) = provider_options + { + observed_reasoning = + observed_reasoning_field(&options, &budget).await?; + } } AssistantRecipe::ToolCall { .. } => has_calls = true, _ => {} @@ -425,9 +474,14 @@ async fn encode( )?, )); } - if nonempty_reasoning { + if let Some(field) = (nonempty_reasoning || observed_reasoning.is_some()).then(|| { + options + .reasoning_field + .or(observed_reasoning) + .unwrap_or(maka_provider_protocol::output::ReasoningField::Content) + }) { fields.push(( - "reasoning_content", + field.as_str(), assistant::plan(parts, assistant::Mode::Reasoning)?, )); } diff --git a/plugins/agent/model/tests/integration/provider_compatible.rs b/plugins/agent/model/tests/integration/provider_compatible.rs index 01051814213..ca8dabc72b1 100644 --- a/plugins/agent/model/tests/integration/provider_compatible.rs +++ b/plugins/agent/model/tests/integration/provider_compatible.rs @@ -460,3 +460,86 @@ 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: Some(json!({"maka":{"openAiChatReasoningField":"reasoning_content"}})), + }, + 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!(wire["messages"][0].get("reasoning_content").is_none()); + assert_eq!(wire["stream_options"], json!({"include_usage":true})); + assert!(wire.get("maka").is_none()); + response(&mut socket, 200, &completion(&[json!({"reasoning_content":"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::() == "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(); +} diff --git a/plugins/agent/model/tests/integration/wire_encoding.rs b/plugins/agent/model/tests/integration/wire_encoding.rs index 8b453f284ac..371a942451a 100644 --- a/plugins/agent/model/tests/integration/wire_encoding.rs +++ b/plugins/agent/model/tests/integration/wire_encoding.rs @@ -262,3 +262,136 @@ async fn notifications_keep_all_output_formats_without_a_second_provider_receipt } } } + +#[tokio::test] +async fn generated_chat_reasoning_resolves_override_and_observed_fields() { + use maka_provider_protocol::{ + chat_completions::request as chat, input as wire, output::ReasoningField, + }; + let observed_content = Some(json!({"maka":{"openAiChatReasoningField":"reasoning_content"}})); + let observed_reasoning = Some(json!({"maka":{"openAiChatReasoningField":"reasoning"}})); + for (override_field, part_options, text, expected) in [ + ( + None, + None, + "思考", + json!({"role":"assistant","content":"reply","reasoning_content":"思考"}), + ), + ( + None, + observed_reasoning.clone(), + "思考", + json!({"role":"assistant","content":"reply","reasoning":"思考"}), + ), + ( + None, + observed_content.clone(), + "", + json!({"role":"assistant","content":"reply","reasoning_content":""}), + ), + ( + Some(ReasoningField::Reasoning), + observed_content.clone(), + "思考", + json!({"role":"assistant","content":"reply","reasoning":"思考"}), + ), + ( + Some(ReasoningField::Reasoning), + observed_content, + "", + json!({"role":"assistant","content":"reply","reasoning":""}), + ), + ( + Some(ReasoningField::Content), + observed_reasoning, + "思考", + json!({"role":"assistant","content":"reply","reasoning_content":"思考"}), + ), + ] { + let mut input = request( + ProviderKind::OpenaiCompatible { + name: "ollama-cloud".into(), + }, + "http://localhost/v1".into(), + ); + if let Some(field) = override_field { + 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: text.into(), + provider_options: part_options.clone(), + }, + AssistantPart::Text { + text: "reply".into(), + provider_options: None, + }, + ], + provider_options: None, + }]; + 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: text.into(), + extensions: part_options + .map(|value| serde_json::from_value(value).unwrap()), + }, + 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: override_field, + }, + ) + .unwrap(); + assert_eq!(direct, vec![expected]); + } + let codecs = [ + json!({}), + json!({"maka":{"openAiChatReasoningField":"reasoning"}}), + json!({"maka":{"openAiChatReasoningField":"reasoning_content"}}), + ] + .map(|options| { + let input = request( + ProviderKind::OpenaiCompatible { + name: "ollama-cloud".into(), + }, + "http://localhost/v1".into(), + ); + WireEncoding::new(input.language, &input.provider, &options, &input.tools) + .unwrap() + .codec() + .to_owned() + }); + assert_ne!(codecs[0], codecs[1]); + assert_ne!(codecs[0], codecs[2]); + assert_ne!(codecs[1], codecs[2]); +} diff --git a/plugins/agent/provider-protocol/src/chat_completions/request.rs b/plugins/agent/provider-protocol/src/chat_completions/request.rs index eb318261bc3..a19aa77918d 100644 --- a/plugins/agent/provider-protocol/src/chat_completions/request.rs +++ b/plugins/agent/provider-protocol/src/chat_completions/request.rs @@ -223,6 +223,7 @@ pub fn prompt_options( provider_namespace: namespace.to_owned(), metadata_namespace, multipart_tool_content, + reasoning_field: Default::default(), } } @@ -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 } )?); diff --git a/plugins/agent/provider-protocol/src/chat_completions/request/prompt.rs b/plugins/agent/provider-protocol/src/chat_completions/request/prompt.rs index ab610ecd0ed..478f23db474 100644 --- a/plugins/agent/provider-protocol/src/chat_completions/request/prompt.rs +++ b/plugins/agent/provider-protocol/src/chat_completions/request/prompt.rs @@ -26,6 +26,7 @@ pub struct PromptOptions { pub provider_namespace: String, pub metadata_namespace: String, pub multipart_tool_content: bool, + pub reasoning_field: Option, } fn metadata<'a>(extensions: &'a Extensions, namespace: &str) -> Option<&'a Value> { extensions.as_ref()?.get(namespace) @@ -209,8 +210,6 @@ enum WireMessage { content: Option, #[serde(skip_serializing_if = "Vec::is_empty")] tool_calls: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - reasoning_content: Option, }, Tool { tool_call_id: String, @@ -292,11 +291,22 @@ pub fn messages(messages: &[Message], options: &PromptOptions) -> Result { let mut text = String::new(); let mut reasoning = String::new(); + let mut observed_reasoning = None; let mut calls = vec![]; for part in parts { match part { AssistantPart::Text { text: delta, .. } => text.push_str(delta), - AssistantPart::Reasoning { text: delta, .. } => reasoning.push_str(delta), + AssistantPart::Reasoning { + text: delta, + extensions, + } => { + reasoning.push_str(delta); + if observed_reasoning.is_none() { + observed_reasoning = metadata(extensions, "maka") + .and_then(|value| value.get("openAiChatReasoningField")) + .and_then(|value| serde_json::from_value(value.clone()).ok()); + } + } AssistantPart::ToolCall(call) => { let mut wire = overlay( serde_json::to_value(Call { @@ -327,18 +337,27 @@ pub fn messages(messages: &[Message], options: &PromptOptions) -> Result {} } } - ( - WireMessage::Assistant { - content: if !calls.is_empty() && text.is_empty() { - None - } else { - Some(text) - }, - tool_calls: calls, - reasoning_content: (!reasoning.is_empty()).then_some(reasoning), + let reasoning_field = + (!reasoning.is_empty() || observed_reasoning.is_some()).then(|| { + options + .reasoning_field + .or(observed_reasoning) + .unwrap_or(crate::output::ReasoningField::Content) + }); + let mut wire = serde_json::to_value(WireMessage::Assistant { + content: if !calls.is_empty() && text.is_empty() { + None + } else { + Some(text) }, - extensions, - ) + tool_calls: calls, + }) + .expect("message"); + if let Some(field) = reasoning_field { + wire[field.as_str()] = json!(reasoning); + } + result.push(overlay(wire, extensions)); + continue; } Message::Tool { results, .. } => { for tool in results { diff --git a/plugins/agent/provider-protocol/src/output.rs b/plugins/agent/provider-protocol/src/output.rs index b79820ae7bd..eef9a840c8f 100644 --- a/plugins/agent/provider-protocol/src/output.rs +++ b/plugins/agent/provider-protocol/src/output.rs @@ -27,8 +27,10 @@ pub enum TextKind { Text, Reasoning, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum ReasoningField { + #[serde(rename = "reasoning_content")] Content, Reasoning, } diff --git a/plugins/agent/providers/src/api.rs b/plugins/agent/providers/src/api.rs index b269d33d9b0..d9cd0106960 100644 --- a/plugins/agent/providers/src/api.rs +++ b/plugins/agent/providers/src/api.rs @@ -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); + } +} diff --git a/plugins/agent/providers/src/api/options.rs b/plugins/agent/providers/src/api/options.rs index 71641c54366..fd97c230862 100644 --- a/plugins/agent/providers/src/api/options.rs +++ b/plugins/agent/providers/src/api/options.rs @@ -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) { diff --git a/plugins/agent/providers/src/api/route.rs b/plugins/agent/providers/src/api/route.rs index 02b34d7ae41..d622f104bfe 100644 --- a/plugins/agent/providers/src/api/route.rs +++ b/plugins/agent/providers/src/api/route.rs @@ -65,8 +65,6 @@ fn check_execution(adapter: &RuntimeAdapter, wire: Wire) -> Result<(), Error> { responses: Some(_), .. } if wire == Wire::OpenaiResponses => true, AdapterKind::OpenaiCompatible { - include_usage: None, - replay_assistant_reasoning_as: None, replay_assistant_reasoning_details: None, normalize_usage: None, .. @@ -83,7 +81,7 @@ pub(crate) struct Route<'a> { pub wire: Wire, pub kind: ProviderKind, pub base_url: String, - adapter: &'a RuntimeAdapter, + pub(super) adapter: &'a RuntimeAdapter, } impl Route<'_> { diff --git a/plugins/agent/providers/tests/integration/api/policy.rs b/plugins/agent/providers/tests/integration/api/policy.rs index c2cfd079fdf..440cbfe71a6 100644 --- a/plugins/agent/providers/tests/integration/api/policy.rs +++ b/plugins/agent/providers/tests/integration/api/policy.rs @@ -180,3 +180,31 @@ async fn resolved_openai_routes_preserve_explicit_effort_and_parallel_policy() { assert_eq!(model.main_output_limit, Some(4000)); } } + +#[tokio::test] +async fn ollama_cloud_catalog_models_are_executable() { + let snapshot = snapshot(); + let binding = binding("ollama-cloud"); + let models: Vec<_> = snapshot + .models(None) + .filter(|(provider, _)| *provider == "ollama-cloud") + .map(|(_, model)| model.clone()) + .collect(); + assert!(!models.is_empty()); + for model in models { + let mut input = request(&binding, &model.id, ModelOverride::default()); + input.model = model; + input.connection.configuration = binding + .configure(json!({"baseUrl":"https://ollama.com/v1"})) + .unwrap(); + let prepared = binding.prepare(input).await.unwrap(); + assert!( + matches!(prepared.protocol, ProviderKind::OpenaiCompatible { ref name } if name == "ollama-cloud") + ); + assert_eq!(prepared.base_url, "https://ollama.com/v1"); + assert_eq!( + prepared.provider_options["maka"], + json!({"includeUsage":true,"openAiChatReasoningField":"reasoning"}) + ); + } +} diff --git a/plugins/settings/native/src/service/tests.rs b/plugins/settings/native/src/service/tests.rs index 433247701f1..29e8ee4cb72 100644 --- a/plugins/settings/native/src/service/tests.rs +++ b/plugins/settings/native/src/service/tests.rs @@ -333,3 +333,26 @@ async fn model_search_returns_executable_selections_and_effective_thinking_metad assert_eq!(empty["choices"], json!([])); fixture.finish().await; } + +#[tokio::test] +async fn ollama_cloud_enabled_models_appear_in_settings_choices() { + let fixture = Fixture::new().await; + let revision = fixture.configuration.catalog().await.unwrap().revision; + let committed = fixture.configuration.create_connection(serde_json::from_value(json!({ + "expectedCatalogRevision":revision, + "connection":{ + "slug":"ollama-cloud","name":"Ollama Cloud", + "provider":{"packageId":maka_providers::api::ID,"entryId":maka_providers::api::ID,"scope":"profile","name":"ollama-cloud"}, + "configuration":{"baseUrl":"https://ollama.com/v1"}, + "enabled":true,"enabledModelIds":["deepseek-v4-flash"] + } + })).unwrap()).await.unwrap(); + assert!(matches!(committed, CatalogMutationResult::Committed { .. })); + let page = fixture.models(json!({"kind":"start"})).await.unwrap(); + assert_eq!(page["choices"].as_array().unwrap().len(), 1); + assert_eq!( + page["choices"][0]["selection"]["modelId"], + "deepseek-v4-flash" + ); + fixture.finish().await; +}